Back up the daemon and restore its data

The backup command asks the running daemon for an archive. A flushed
memory snapshot holds the write locks during the agent copy, and SQLite
VACUUM INTO saves the WhatsApp session without raw database files.
Compression runs after the write locks are released.

Restore validates the archive in a staging directory before it installs
the files. The token lock excludes a running daemon, and a nonempty
destination fails without replacement of its data.

diff --git a/backup.go b/backup.go
new file mode 100644 (file)
index 0000000..56cf516
--- /dev/null
+++ b/backup.go
@@ -0,0 +1,324 @@
+package main
+
+import (
+       "context"
+       "encoding/json"
+       "errors"
+       "fmt"
+       "io"
+       "io/fs"
+       "net"
+       "os"
+       "path/filepath"
+       "strings"
+)
+
+// backup asks the daemon for a private archive and prints its path and size.
+func backup(ctx context.Context, path string, out io.Writer) (err error) {
+       dir, err := dataDir()
+       if err != nil {
+               return fmt.Errorf("find data directory: %w", err)
+       }
+
+       path, err = filepath.Abs(path)
+       if err != nil {
+               return fmt.Errorf("find backup path: %w", err)
+       }
+
+       var dialer net.Dialer
+
+       conn, err := dialer.DialContext(ctx, "unix", filepath.Join(dir, socketFileName))
+       if err != nil {
+               return errors.New("backup needs a running daemon")
+       }
+
+       defer func() {
+               err = errors.Join(err, conn.Close())
+       }()
+
+       stop := context.AfterFunc(ctx, func() {
+               _ = conn.Close()
+       })
+       defer stop()
+
+       err = json.NewEncoder(conn).Encode(SocketRequest{
+               Type: "backup",
+               Text: path,
+       })
+       if err != nil {
+               return fmt.Errorf("send backup request: %w", err)
+       }
+
+       decoder := json.NewDecoder(conn)
+       for {
+               var event Event
+
+               err = decoder.Decode(&event)
+               if err != nil {
+                       return fmt.Errorf("read backup answer: %w", err)
+               }
+
+               switch event.Kind {
+               case "backup":
+                       _, err = fmt.Fprintln(out, event.Text)
+                       if err != nil {
+                               return fmt.Errorf("write backup answer: %w", err)
+                       }
+
+                       return nil
+               case "error":
+                       return fmt.Errorf("backup failed: %s", event.Text)
+               }
+       }
+}
+
+// restore runs the offline restore command and reports its data directory.
+func restore(ctx context.Context, path string, out io.Writer) error {
+       dir, err := dataDir()
+       if err != nil {
+               return fmt.Errorf("find restore directory: %w", err)
+       }
+
+       err = backupRestore(ctx, path, dir)
+       if err != nil {
+               return fmt.Errorf("restore archive: %w", err)
+       }
+
+       _, err = fmt.Fprintln(out, "Restored", dir)
+       if err != nil {
+               return fmt.Errorf("write restore answer: %w", err)
+       }
+
+       return nil
+}
+
+// backupData stages a session snapshot, then copies the data under snapshot's
+// write locks. Compression runs after the locks are released.
+func backupData(ctx context.Context, dir, path string, whatsapp *WhatsApp, snapshot func(context.Context, func() error) error) (size int64, err error) {
+       if !filepath.IsAbs(path) {
+               return 0, errors.New("backup path must be absolute")
+       }
+
+       stage, err := os.MkdirTemp("", "onechat-backup-*")
+       if err != nil {
+               return 0, fmt.Errorf("create backup directory: %w", err)
+       }
+
+       defer func() {
+               err = errors.Join(err, os.RemoveAll(stage))
+       }()
+
+       err = whatsapp.BackupSession(ctx, filepath.Join(stage, whatsappDirName, whatsappSessionFileName))
+       if err != nil {
+               return 0, fmt.Errorf("copy WhatsApp session: %w", err)
+       }
+
+       err = snapshot(ctx, func() error {
+               return backupCopy(ctx, dir, stage, path)
+       })
+       if err != nil {
+               return 0, fmt.Errorf("copy data directory: %w", err)
+       }
+
+       size, err = backupArchive(ctx, stage, path)
+       if err != nil {
+               return 0, fmt.Errorf("archive backup: %w", err)
+       }
+
+       return size, nil
+}
+
+// backupRestore installs a validated archive into an empty data directory while
+// holding the token lock. It keeps the lock file so daemon starts use that same
+// inode, including after a failed restore.
+func backupRestore(ctx context.Context, path, dir string) (err error) {
+       dir, err = filepath.Abs(dir)
+       if err != nil {
+               return fmt.Errorf("find restore destination: %w", err)
+       }
+
+       entries, err := os.ReadDir(dir)
+       if err != nil {
+               return fmt.Errorf("read restore destination: %w", err)
+       }
+
+       if len(entries) != 0 {
+               hasLock := false
+               for _, entry := range entries {
+                       hasLock = hasLock || entry.Name() == tokenLockFileName
+               }
+
+               if !hasLock {
+                       return errors.New("restore needs an empty data directory")
+               }
+       }
+
+       tokens, err := lockTokens(dir)
+       if err != nil {
+               return fmt.Errorf("restore needs a stopped daemon: %w", err)
+       }
+
+       defer func() {
+               err = errors.Join(err, tokens.Close())
+       }()
+
+       if len(entries) != 0 {
+               return errors.New("restore needs an empty data directory")
+       }
+
+       entries, err = os.ReadDir(dir)
+       if err != nil {
+               return fmt.Errorf("check restore destination: %w", err)
+       }
+
+       if len(entries) != 1 || entries[0].Name() != tokenLockFileName {
+               return errors.New("restore needs an empty data directory")
+       }
+
+       stage, err := os.MkdirTemp(filepath.Dir(dir), ".onechat-restore-*")
+       if err != nil {
+               return fmt.Errorf("create restore directory: %w", err)
+       }
+
+       defer func() {
+               err = errors.Join(err, os.RemoveAll(stage))
+       }()
+
+       err = backupExtract(ctx, path, stage)
+       if err != nil {
+               return fmt.Errorf("extract backup: %w", err)
+       }
+
+       entries, err = os.ReadDir(stage)
+       if err != nil {
+               return fmt.Errorf("read restored files: %w", err)
+       }
+
+       for _, entry := range entries {
+               err = os.Rename(filepath.Join(stage, entry.Name()), filepath.Join(dir, entry.Name()))
+               if err != nil {
+                       return fmt.Errorf("install restored entry: %w", err)
+               }
+       }
+
+       return nil
+}
+
+// backupCopy stages regular files and directories, excluding runtime files, the
+// output archive, and raw session database files. Each file keeps its mode.
+func backupCopy(ctx context.Context, source, target, output string) error {
+       var directories []string
+
+       modes := make(map[string]fs.FileMode)
+
+       err := filepath.WalkDir(source, func(path string, entry fs.DirEntry, err error) error {
+               if err != nil {
+                       return fmt.Errorf("read data entry: %w", err)
+               }
+
+               name, err := filepath.Rel(source, path)
+               if err != nil {
+                       return fmt.Errorf("name data entry: %w", err)
+               }
+
+               if name == "." {
+                       return nil
+               }
+
+               absolute, err := filepath.Abs(path)
+               if err != nil {
+                       return fmt.Errorf("find data entry: %w", err)
+               }
+
+               session := filepath.Join(whatsappDirName, whatsappSessionFileName)
+               if backupExcluded(name) || absolute == output || name == session || strings.HasPrefix(name, session+"-") {
+                       if entry.IsDir() {
+                               return filepath.SkipDir
+                       }
+
+                       return nil
+               }
+
+               info, err := entry.Info()
+               if err != nil {
+                       return fmt.Errorf("read data file: %w", err)
+               }
+
+               destination := filepath.Join(target, name)
+               switch {
+               case info.IsDir():
+                       err = os.MkdirAll(destination, 0o700)
+                       directories = append(directories, destination)
+                       modes[destination] = info.Mode().Perm()
+               case info.Mode().IsRegular():
+                       err = backupStageFile(ctx, path, destination)
+               default:
+                       return fmt.Errorf("unsupported data entry %q", name)
+               }
+
+               if err != nil {
+                       return fmt.Errorf("stage data entry %q: %w", name, err)
+               }
+
+               return nil
+       })
+       if err != nil {
+               return fmt.Errorf("stage backup: %w", err)
+       }
+
+       for i := len(directories) - 1; i >= 0; i-- {
+               path := directories[i]
+               err = os.Chmod(path, modes[path])
+               if err != nil {
+                       return fmt.Errorf("stage directory mode: %w", err)
+               }
+       }
+
+       return nil
+}
+
+// backupStageFile opens before it reads size and mode, so an atomic state file
+// replacement cannot join the metadata of one version to the bytes of another.
+func backupStageFile(ctx context.Context, source, target string) (err error) {
+       input, err := os.Open(source)
+       if err != nil {
+               return fmt.Errorf("open snapshot source: %w", err)
+       }
+
+       defer func() {
+               err = errors.Join(err, input.Close())
+       }()
+
+       info, err := input.Stat()
+       if err != nil {
+               return fmt.Errorf("read snapshot source: %w", err)
+       }
+
+       if !info.Mode().IsRegular() {
+               return errors.New("snapshot source must be a regular file")
+       }
+
+       file, err := os.OpenFile(target, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+       if err != nil {
+               return fmt.Errorf("create staged file: %w", err)
+       }
+
+       defer func() {
+               err = errors.Join(err, file.Close())
+       }()
+
+       _, err = io.CopyN(file, BackupReader{
+               ctx:    ctx,
+               reader: input,
+       }, info.Size())
+       if err != nil {
+               return fmt.Errorf("stage file: %w", err)
+       }
+
+       err = file.Chmod(info.Mode().Perm())
+       if err != nil {
+               return fmt.Errorf("stage file mode: %w", err)
+       }
+
+       return nil
+}

diff --git a/backup_archive.go b/backup_archive.go
index bd6e00367c481c34cb4e4bf3e2775e6e64f5fb46..9560cd1cb6a5818224e05b37ab1155b5c4f9708a 100644 (file)
--- a/backup_archive.go
+++ b/backup_archive.go
@@ -31,9 +31,9 @@ func backupArchive(ctx context.Context, root, path string) (size int64, err erro
 
        compressed := gzip.NewWriter(file)
        archive := tar.NewWriter(compressed)
 
        compressed := gzip.NewWriter(file)
        archive := tar.NewWriter(compressed)

-       err = filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error {
-               if walkErr != nil {
-                       return fmt.Errorf("walk backup: %w", walkErr)
+       err = filepath.WalkDir(root, func(path string, entry fs.DirEntry, err error) error {
+               if err != nil {
+                       return fmt.Errorf("walk backup: %w", err)

                }
 
                if ctx.Err() != nil {
                }
 
                if ctx.Err() != nil {


@@ -44,38 +44,43 @@ func backupArchive(ctx context.Context, root, path string) (size int64, err erro
                        return nil
                }
 
                        return nil
                }
 

-               name, nameErr := filepath.Rel(root, path)
-               if nameErr != nil {
-                       return fmt.Errorf("name archive entry: %w", nameErr)
+               name, err := filepath.Rel(root, path)
+               if err != nil {
+                       return fmt.Errorf("name archive entry: %w", err)

                }
 
                }
 

-               info, infoErr := entry.Info()
-               if infoErr != nil {
-                       return fmt.Errorf("read backup entry: %w", infoErr)
+               info, err := entry.Info()
+               if err != nil {
+                       return fmt.Errorf("read backup entry: %w", err)

                }
 
                if !info.IsDir() && !info.Mode().IsRegular() {
                        return fmt.Errorf("unsupported backup entry %q", name)
                }
 
                }
 
                if !info.IsDir() && !info.Mode().IsRegular() {
                        return fmt.Errorf("unsupported backup entry %q", name)
                }
 

-               header, headerErr := tar.FileInfoHeader(info, "")
-               if headerErr != nil {
-                       return fmt.Errorf("make archive header: %w", headerErr)
+               header, err := tar.FileInfoHeader(info, "")
+               if err != nil {
+                       return fmt.Errorf("make archive header: %w", err)

                }
 
                header.Name = filepath.ToSlash(name)
                header.Uid, header.Gid = 0, 0
                header.Uname, header.Gname = "", ""
                }
 
                header.Name = filepath.ToSlash(name)
                header.Uid, header.Gid = 0, 0
                header.Uname, header.Gname = "", ""

-               headerErr = archive.WriteHeader(header)
-               if headerErr != nil {
-                       return fmt.Errorf("write archive header: %w", headerErr)
+               err = archive.WriteHeader(header)
+               if err != nil {
+                       return fmt.Errorf("write archive header: %w", err)

                }
 
                if info.IsDir() {
                        return nil
                }
 
                }
 
                if info.IsDir() {
                        return nil
                }
 

-               return backupCopyFile(ctx, archive, path, info.Size())
+               err = backupCopyFile(ctx, archive, path, info.Size())
+               if err != nil {
+                       return fmt.Errorf("archive file: %w", err)
+               }
+
+               return nil

        })
        err = errors.Join(err, archive.Close(), compressed.Close())
        if err != nil {
        })
        err = errors.Join(err, archive.Close(), compressed.Close())
        if err != nil {


@@ -122,19 +127,24 @@ func backupExtract(ctx context.Context, path, root string) (err error) {
                err = errors.Join(err, compressed.Close())
        }()
 
                err = errors.Join(err, compressed.Close())
        }()
 

-       archive := tar.NewReader(BackupReader{ctx: ctx, reader: compressed})
+       archive := tar.NewReader(BackupReader{
+               ctx:    ctx,
+               reader: compressed,
+       })

        seen := make(map[string]bool)
 
        var directories []*tar.Header
 
        for {
        seen := make(map[string]bool)
 
        var directories []*tar.Header
 
        for {

-               header, readErr := archive.Next()
-               if errors.Is(readErr, io.EOF) {
+               var header *tar.Header
+
+               header, err = archive.Next()
+               if errors.Is(err, io.EOF) {

                        break
                }
 
                        break
                }
 

-               if readErr != nil {
-                       return fmt.Errorf("read archive entry: %w", readErr)
+               if err != nil {
+                       return fmt.Errorf("read archive entry: %w", err)

                }
 
                name := strings.TrimSuffix(header.Name, "/")
                }
 
                name := strings.TrimSuffix(header.Name, "/")


@@ -218,7 +228,10 @@ func backupCopyFile(ctx context.Context, out io.Writer, path string, size int64)
                err = errors.Join(err, file.Close())
        }()
 
                err = errors.Join(err, file.Close())
        }()
 

-       _, err = io.CopyN(out, BackupReader{ctx: ctx, reader: file}, size)
+       _, err = io.CopyN(out, BackupReader{
+               ctx:    ctx,
+               reader: file,
+       }, size)

        if err != nil {
                return fmt.Errorf("copy backup file: %w", err)
        }
        if err != nil {
                return fmt.Errorf("copy backup file: %w", err)
        }

diff --git a/backup_archive_helper_test.go b/backup_archive_helper_test.go
new file mode 100644 (file)
index 0000000..fccc336
--- /dev/null
+++ b/backup_archive_helper_test.go
@@ -0,0 +1,32 @@
+package main
+
+import (
+       "archive/tar"
+       "compress/gzip"
+       "os"
+       "path/filepath"
+       "testing"
+
+       "github.com/stretchr/testify/require"
+)
+
+func backupTestArchive(t *testing.T, headers ...tar.Header) string {
+       t.Helper()
+
+       path := filepath.Join(t.TempDir(), "input.tar.gz")
+       file, err := os.Create(path)
+       require.NoError(t, err)
+
+       compressed := gzip.NewWriter(file)
+
+       archive := tar.NewWriter(compressed)
+       for _, header := range headers {
+               require.NoError(t, archive.WriteHeader(&header))
+       }
+
+       require.NoError(t, archive.Close())
+       require.NoError(t, compressed.Close())
+       require.NoError(t, file.Close())
+
+       return path
+}

diff --git a/backup_archive_test.go b/backup_archive_test.go
index c3cbcbbf8141d107dc13fa08271591c08f3a2909..e42733395834d26d6ac44fe2ce2672e004dfd639 100644 (file)
--- a/backup_archive_test.go
+++ b/backup_archive_test.go
@@ -2,7 +2,6 @@ package main
 
 import (
        "archive/tar"
 
 import (
        "archive/tar"

-       "compress/gzip"

        "os"
        "path/filepath"
        "testing"
        "os"
        "path/filepath"
        "testing"


@@ -138,24 +137,3 @@ func TestBackupFailureKeepsTheDestination(t *testing.T) {
        require.NoError(t, err)
        require.Len(t, entries, 1)
 }
        require.NoError(t, err)
        require.Len(t, entries, 1)
 }

-
-func backupTestArchive(t *testing.T, headers ...tar.Header) string {
-       t.Helper()
-
-       path := filepath.Join(t.TempDir(), "input.tar.gz")
-       file, err := os.Create(path)
-       require.NoError(t, err)
-
-       compressed := gzip.NewWriter(file)
-
-       archive := tar.NewWriter(compressed)
-       for _, header := range headers {
-               require.NoError(t, archive.WriteHeader(&header))
-       }
-
-       require.NoError(t, archive.Close())
-       require.NoError(t, compressed.Close())
-       require.NoError(t, file.Close())
-
-       return path
-}

diff --git a/backup_test.go b/backup_test.go
new file mode 100644 (file)
index 0000000..440dfe2
--- /dev/null
+++ b/backup_test.go
@@ -0,0 +1,323 @@
+package main
+
+import (
+       "archive/tar"
+       "context"
+       "database/sql"
+       "io"
+       "os"
+       "path/filepath"
+       "strings"
+       "sync"
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestRunningDaemonBackupRestoresTheChatAndLinks(t *testing.T) {
+       dir := t.TempDir()
+       whatsappTestSession(t, dir, true)
+
+       files := map[string]string{
+               "auth.json":                  `{"access_token":"fake-secret"}`,
+               "telegram.json":              `{"token":"fake-token","owner":123}`,
+               "config.json":                `{"endpoint":"api"}`,
+               "agent/instructions.md":      "Keep the instructions.\n",
+               "agent/telegram-offset.json": "23\n",
+       }
+       for name, data := range files {
+               path := filepath.Join(dir, name)
+               require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700))
+               require.NoError(t, os.WriteFile(path, []byte(data), 0o600))
+       }
+
+       memory, whatsapp, subagents := backupStartDaemon(t, dir)
+       require.True(t, whatsapp.isLinked())
+
+       _, err := memory.Append("note", "A durable decision.")
+       require.NoError(t, err)
+
+       _, err = memory.Append("note", "The next decision.")
+       require.NoError(t, err)
+       require.NoError(t, memory.Settle(t.Context()))
+
+       child, err := subagents.Create(t.Context())
+       require.NoError(t, err)
+
+       _, err = child.Append("note", "subagent")
+       require.NoError(t, err)
+
+       _, err = child.Append("user", "A saved task.")
+       require.NoError(t, err)
+
+       childText, err := child.Show()
+       require.NoError(t, err)
+
+       for _, excluded := range []string{"probe/skip", "agent-backup-example/skip"} {
+               path := filepath.Join(dir, excluded)
+               require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700))
+               require.NoError(t, os.WriteFile(path, []byte("excluded"), 0o600))
+       }
+
+       archive := filepath.Join(t.TempDir(), "daily.tar.gz")
+       t.Setenv("ONECHAT_DIR", dir)
+
+       var answer strings.Builder
+
+       require.NoError(t, backup(t.Context(), archive, &answer))
+       require.Regexp(t, `^`+archive+` \([0-9]+ bytes\)\n$`, answer.String())
+
+       info, err := os.Stat(archive)
+       require.NoError(t, err)
+       require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
+
+       restored := t.TempDir()
+       require.NoError(t, backupRestore(t.Context(), archive, restored))
+
+       for name, data := range files {
+               bytes, err := os.ReadFile(filepath.Join(restored, name))
+               require.NoError(t, err)
+               require.Equal(t, data, string(bytes))
+       }
+
+       for _, excluded := range []string{"probe", "agent-backup-example", socketFileName, "agent/lock"} {
+               _, err = os.Stat(filepath.Join(restored, excluded))
+               require.ErrorIs(t, err, os.ErrNotExist)
+       }
+
+       restoredMemory, restoredWhatsApp, restoredSubagents := backupStartDaemon(t, restored)
+       require.Equal(t, memory.Len(), restoredMemory.Len())
+       require.Equal(t, memory.View(), restoredMemory.View())
+
+       for id := range memory.Len() {
+               expected, err := memory.Message(id)
+               require.NoError(t, err)
+
+               actual, err := restoredMemory.Message(id)
+               require.NoError(t, err)
+               require.Equal(t, expected, actual)
+       }
+
+       require.True(t, restoredWhatsApp.isLinked())
+       require.Equal(t, whatsapp.link().own, restoredWhatsApp.link().own)
+
+       restoredChild, err := restoredSubagents.Get(t.Context(), child.name)
+       require.NoError(t, err)
+
+       restoredText, err := restoredChild.Show()
+       require.NoError(t, err)
+       require.Equal(t, childText, restoredText)
+}
+
+func TestRestoreRefusesNonemptyDataWithoutChangingIt(t *testing.T) {
+       dir := t.TempDir()
+       path := filepath.Join(dir, "keep")
+       require.NoError(t, os.WriteFile(path, []byte("keep"), 0o600))
+
+       err := backupRestore(t.Context(), "missing.tar.gz", dir)
+       require.ErrorContains(t, err, "empty data directory")
+
+       entries, err := os.ReadDir(dir)
+       require.NoError(t, err)
+       require.Len(t, entries, 1)
+
+       data, err := os.ReadFile(path)
+       require.NoError(t, err)
+       require.Equal(t, "keep", string(data))
+}
+
+func TestRestoreRefusesTheDaemonTokenLock(t *testing.T) {
+       dir := t.TempDir()
+       tokens, err := lockTokens(dir)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, tokens.Close())
+       }()
+
+       err = backupRestore(t.Context(), "missing.tar.gz", dir)
+       require.ErrorIs(t, err, errTokensInUse)
+}
+
+func TestUnsafeRestoreInstallsNoArchiveFiles(t *testing.T) {
+       archive := backupTestArchive(t,
+               tar.Header{
+                       Name:     "safe",
+                       Typeflag: tar.TypeReg,
+                       Mode:     0o600,
+               },
+               tar.Header{
+                       Name:     "../escape",
+                       Typeflag: tar.TypeReg,
+                       Mode:     0o600,
+               },
+       )
+       dir := t.TempDir()
+       err := backupRestore(t.Context(), archive, dir)
+       require.ErrorContains(t, err, "unsafe archive path")
+
+       entries, err := os.ReadDir(dir)
+       require.NoError(t, err)
+       require.Len(t, entries, 1)
+       require.Equal(t, tokenLockFileName, entries[0].Name())
+}
+
+func TestBackupNeedsARunningDaemon(t *testing.T) {
+       t.Setenv("ONECHAT_DIR", t.TempDir())
+
+       err := backup(t.Context(), filepath.Join(t.TempDir(), "backup.tar.gz"), io.Discard)
+       require.EqualError(t, err, "backup needs a running daemon")
+}
+
+func TestWhatsAppSessionBackupIncludesCommittedWALWrites(t *testing.T) {
+       dir := t.TempDir()
+       whatsappTestSession(t, dir, true)
+
+       whatsapp, _, _ := whatsappStartTest(t, dir, newTestClock())
+       address, err := whatsappSessionAddress(filepath.Join(dir, whatsappDirName, whatsappSessionFileName))
+       require.NoError(t, err)
+
+       database, err := sql.Open("sqlite", address)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, database.Close())
+       }()
+
+       _, err = database.ExecContext(t.Context(), "PRAGMA journal_mode=WAL; CREATE TABLE backup_sample (value TEXT); INSERT INTO backup_sample VALUES ('saved');")
+       require.NoError(t, err)
+
+       target := filepath.Join(t.TempDir(), "copy", whatsappSessionFileName)
+       require.NoError(t, whatsapp.BackupSession(t.Context(), target))
+
+       address, err = whatsappSessionAddress(target)
+       require.NoError(t, err)
+
+       restored, err := sql.Open("sqlite", address)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, restored.Close())
+       }()
+
+       var value string
+
+       err = restored.QueryRowContext(t.Context(), "SELECT value FROM backup_sample").Scan(&value)
+       require.NoError(t, err)
+       require.Equal(t, "saved", value)
+
+       info, err := os.Stat(target)
+       require.NoError(t, err)
+       require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
+}
+
+func TestSessionBackupWaitsForRemovalToFinish(t *testing.T) {
+       dir := t.TempDir()
+       whatsappTestSession(t, dir, true)
+
+       whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())
+       disconnect := &BackupDisconnect{
+               WhatsAppClient: fake,
+               entered:        make(chan struct{}),
+               release:        make(chan struct{}),
+       }
+
+       var release sync.Once
+
+       defer release.Do(func() {
+               close(disconnect.release)
+       })
+
+       whatsapp.mutex.Lock()
+
+       whatsapp.client = disconnect
+       whatsapp.mutex.Unlock()
+
+       removed := make(chan error, 1)
+       go func() {
+               removed <- whatsapp.removeSession()
+       }()
+
+       <-disconnect.entered
+
+       target := filepath.Join(t.TempDir(), "copy", whatsappSessionFileName)
+       backedUp := make(chan error, 1)
+       started := make(chan struct{})
+       go func() {
+               close(started)
+
+               backedUp <- whatsapp.BackupSession(t.Context(), target)
+       }()
+
+       <-started
+       require.Never(t, func() bool {
+               return len(backedUp) != 0
+       }, 50*time.Millisecond, time.Millisecond)
+       release.Do(func() {
+               close(disconnect.release)
+       })
+       require.NoError(t, <-removed)
+       require.NoError(t, <-backedUp)
+
+       for _, path := range []string{whatsapp.sessionPath(), target} {
+               _, err := os.Stat(path)
+               require.ErrorIs(t, err, os.ErrNotExist)
+       }
+}
+
+// BackupDisconnect holds session removal in its unlocked client cleanup.
+type BackupDisconnect struct {
+       WhatsAppClient
+       entered chan struct{}
+       release chan struct{}
+}
+
+func (d *BackupDisconnect) Disconnect() {
+       close(d.entered)
+       <-d.release
+}
+
+// backupStartDaemon serves fake model and WhatsApp clients on the socket of
+// dir. It keeps all state files under dir and holds the daemon token lock.
+func backupStartDaemon(t *testing.T, dir string) (*Memory, *WhatsApp, *Subagents) {
+       t.Helper()
+
+       tokens, err := lockTokens(dir)
+       require.NoError(t, err)
+
+       clock := newTestClock()
+       agent, err := openAgent(filepath.Join(dir, daemonAgentDirName))
+       require.NoError(t, err)
+
+       memory, err := openMemory(t.Context(), agent.dir, clock.Now, newFakeSummarizer(), discardLog())
+       require.NoError(t, err)
+
+       whatsapp, _, _ := whatsappStartTest(t, dir, clock)
+       subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), clock.Now, subagentNameRandom)
+       schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), clock.Now, discardLog())
+       require.NoError(t, err)
+
+       server := newModelFake(t)
+       client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
+       listener, err := daemonListenSocket(filepath.Join(dir, socketFileName))
+       require.NoError(t, err)
+
+       ctx, cancel := context.WithCancel(t.Context())
+       done := make(chan error, 1)
+       go func() {
+               done <- daemonServe(ctx, listener, nil, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
+                       return newTurnLoop(agent, memory, subagents, whatsapp, schedules, client, clock.Now, discardLog(), notify)
+               })
+       }()
+
+       t.Cleanup(func() {
+               cancel()
+               require.NoError(t, <-done)
+               require.NoError(t, subagents.Close())
+               require.NoError(t, memory.Close())
+               require.NoError(t, tokens.Close())
+       })
+
+       return memory, whatsapp, subagents
+}

diff --git a/daemon.go b/daemon.go
index 898de68d34ae0355a0eedb36a8dcc21fbf95556a..e534aeac96a7e1e781c3f55ba9e636848b062fb7 100644 (file)
--- a/daemon.go
+++ b/daemon.go
@@ -204,12 +204,24 @@ func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop)
 
                err := json.Unmarshal(scanner.Bytes(), &request)
                if err == nil {
 
                err := json.Unmarshal(scanner.Bytes(), &request)
                if err == nil {

-                       err = daemonRunRequest(l, h.whatsapp, request, func(text string) {
-                               h.Send(client, Event{
-                                       Kind: "notice",
-                                       Text: text,
+                       if request.Type == "backup" {
+                               var size int64
+
+                               size, err = backupData(ctx, filepath.Dir(l.agent.dir), request.Text, h.whatsapp, l.runs.Snapshot)
+                               if err == nil {
+                                       h.Send(client, Event{
+                                               Kind: "backup",
+                                               Text: fmt.Sprintf("%s (%d bytes)", request.Text, size),
+                                       })
+                               }
+                       } else {
+                               err = daemonRunRequest(l, h.whatsapp, request, func(text string) {
+                                       h.Send(client, Event{
+                                               Kind: "notice",
+                                               Text: text,
+                                       })

                                })
                                })

-                       })
+                       }

                }
 
                if err != nil {
                }
 
                if err != nil {

diff --git a/main.go b/main.go
index a73407f7c8457e973f0fb7731312a139acb7c3d2..efecde3b65b7ceca1652fe97511ba87bd04e7c39 100644 (file)
--- a/main.go
+++ b/main.go
@@ -14,6 +14,8 @@ import (
 const usage = `Usage: onechat [COMMAND]
 
 Commands:
 const usage = `Usage: onechat [COMMAND]
 
 Commands:

+  backup FILE              archive the running daemon data
+  restore FILE             restore into an empty data directory

   chat                     talk to the agent in this terminal (default)
   daemon                   run the agent, Telegram, and WhatsApp
   import [-n] ZIP...        import WhatsApp history (stop the daemon first)
   chat                     talk to the agent in this terminal (default)
   daemon                   run the agent, Telegram, and WhatsApp
   import [-n] ZIP...        import WhatsApp history (stop the daemon first)


@@ -65,6 +67,10 @@ func run() error {
                return daemon(ctx, os.Stdout)
        case len(args) == 0 || len(args) == 1 && args[0] == "chat":
                return terminal(ctx, os.Stdin, os.Stdout)
                return daemon(ctx, os.Stdout)
        case len(args) == 0 || len(args) == 1 && args[0] == "chat":
                return terminal(ctx, os.Stdin, os.Stdout)

+       case len(args) == 2 && args[0] == "backup":
+               return backup(ctx, args[1], os.Stdout)
+       case len(args) == 2 && args[0] == "restore":
+               return restore(ctx, args[1], os.Stdout)

        case len(args) > 0 && args[0] == "import":
                return whatsappImport(ctx, args[1:], os.Stdout)
        case len(args) > 0 && args[0] == "probe":
        case len(args) > 0 && args[0] == "import":
                return whatsappImport(ctx, args[1:], os.Stdout)
        case len(args) > 0 && args[0] == "probe":

diff --git a/memory.go b/memory.go
index 20f46d62ba6d50397cf6a0b7f892f8a59c2cf688..c9fe1c567504c11f0a94be48fdb581298cde510c 100644 (file)
--- a/memory.go
+++ b/memory.go
@@ -247,6 +247,26 @@ func (m *Memory) Flush(ctx context.Context) error {
        return m.syncer.Flush(ctx)
 }
 
        return m.syncer.Flush(ctx)
 }
 

+// Snapshot flushes all memory writes and runs snapshot while no append or
+// summary write can change the log, indexes, tree, or views. snapshot must not
+// call Memory.
+func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       err := m.Flush(ctx)
+       if err != nil {
+               return fmt.Errorf("flush memory snapshot: %w", err)
+       }
+
+       err = snapshot()
+       if err != nil {
+               return fmt.Errorf("copy memory snapshot: %w", err)
+       }
+
+       return nil
+}
+

 // ViewLines returns the lines of the chat view, each with its newline, without
 // the leaf lines of the messages in skip.
 func (m *Memory) ViewLines(skip map[int]bool) []string {
 // ViewLines returns the lines of the chat view, each with its newline, without
 // the leaf lines of the messages in skip.
 func (m *Memory) ViewLines(skip map[int]bool) []string {

diff --git a/socket.go b/socket.go
index a306c5c81e00a6ec5f03391a4e9ae93fe7d4af6e..7b2da2b7a2077e11529c051efe4816de80827bf9 100644 (file)
--- a/socket.go
+++ b/socket.go
@@ -31,7 +31,9 @@ func checkSocketPath(path string) error {
 
 // SocketRequest is one line from a client: "send" with the text of a user
 // message, "interrupt", "stop" with the name of a subagent in Text, or
 
 // SocketRequest is one line from a client: "send" with the text of a user
 // message, "interrupt", "stop" with the name of a subagent in Text, or

-// "whatsapp" with the text after /whatsapp of a WhatsApp command.
+// "whatsapp" with the text after /whatsapp of a WhatsApp command, or "backup"
+// with the absolute output path in Text. The backup answer is a "backup" event
+// with the path and size, or an "error" event.

 type SocketRequest struct {
        Type string `json:"type"`
        Text string `json:"text,omitempty"`
 type SocketRequest struct {
        Type string `json:"type"`
        Text string `json:"text,omitempty"`

diff --git a/subagent.go b/subagent.go
index 132ca475fe8d525f91a2d452fba8ff8451b02fe9..07bfa452f6c8aef7bc18bf10c6bbc097668f5793 100644 (file)
--- a/subagent.go
+++ b/subagent.go
@@ -124,6 +124,30 @@ func (s *Subagents) Get(ctx context.Context, name string) (*Subagent, error) {
        return subagent, nil
 }
 
        return subagent, nil
 }
 

+// Snapshot flushes and locks the open logs while snapshot reads the subagent
+// files. snapshot must not call Subagents or an open Subagent.
+func (s *Subagents) Snapshot(ctx context.Context, snapshot func() error) error {
+       s.mutex.Lock()
+       defer s.mutex.Unlock()
+
+       for _, subagent := range s.open {
+               subagent.mutex.Lock()
+               defer subagent.mutex.Unlock()
+
+               err := subagent.Flush(ctx)
+               if err != nil {
+                       return fmt.Errorf("flush subagent snapshot: %w", err)
+               }
+       }
+
+       err := snapshot()
+       if err != nil {
+               return fmt.Errorf("copy subagent snapshot: %w", err)
+       }
+
+       return nil
+}
+

 // Close makes the writes of each open log durable and closes it.
 func (s *Subagents) Close() error {
        s.mutex.Lock()
 // Close makes the writes of each open log durable and closes it.
 func (s *Subagents) Close() error {
        s.mutex.Lock()

diff --git a/subagent_run.go b/subagent_run.go
index 0780393db3f87c96e5d6aef425535106c102b4f3..8084c390a7012d5213019f1f784ec3214ce08bbe 100644 (file)
--- a/subagent_run.go
+++ b/subagent_run.go
@@ -256,6 +256,18 @@ func (r *SubagentRuns) Resume(ctx context.Context) error {
        return r.save()
 }
 
        return r.save()
 }
 

+// Snapshot locks the run state, memory, and subagent logs while snapshot reads
+// the agent files. The lock order agrees with Stop, which logs under the run
+// lock.
+func (r *SubagentRuns) Snapshot(ctx context.Context, snapshot func() error) error {
+       r.mutex.Lock()
+       defer r.mutex.Unlock()
+
+       return r.memory.Snapshot(ctx, func() error {
+               return r.subagents.Snapshot(ctx, snapshot)
+       })
+}
+

 // Close ends every run and waits for them. No subagent starts after Close.
 func (r *SubagentRuns) Close() {
        r.mutex.Lock()
 // Close ends every run and waits for them. No subagent starts after Close.
 func (r *SubagentRuns) Close() {
        r.mutex.Lock()

diff --git a/whatsapp.go b/whatsapp.go
index 95e1d54ea01e80f8216f29e38d63a56a479cc07a..0c68a44ace8a0184b89789e473537d75ec876199 100644 (file)
--- a/whatsapp.go
+++ b/whatsapp.go
@@ -2,6 +2,7 @@ package main
 
 import (
        "context"
 
 import (
        "context"

+       "database/sql"

        "errors"
        "fmt"
        "log/slog"
        "errors"
        "fmt"
        "log/slog"


@@ -163,7 +164,9 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 // of the client that asked. A logout from the phone comes as an event of the
 // library, and Run deletes the session. The mutex protects the session, the
 // client, and the state of the connection, which the event handler of the
 // of the client that asked. A logout from the phone comes as an event of the
 // library, and Run deletes the session. The mutex protects the session, the
 // client, and the state of the connection, which the event handler of the

-// library changes from its own goroutines.
+// library changes from its own goroutines. The sessionMutex serializes session
+// creation, close, removal, and snapshots. It is held before mutex, and stays
+// held when removal releases mutex for client callbacks.

 //
 // The inbox puts the text messages into batches and logs them, and groups
 // holds the names of the groups. Both get a WhatsAppLink from link for each
 //
 // The inbox puts the text messages into batches and logs them, and groups
 // holds the names of the groups. Both get a WhatsAppLink from link for each


@@ -188,6 +191,7 @@ type WhatsApp struct {
        groups      *WhatsAppGroups
        inbox       *WhatsAppInbox
 
        groups      *WhatsAppGroups
        inbox       *WhatsAppInbox
 

+       sessionMutex   sync.Mutex

        mutex          sync.Mutex
        container      *sqlstore.Container
        device         *store.Device
        mutex          sync.Mutex
        container      *sqlstore.Container
        device         *store.Device


@@ -374,8 +378,62 @@ func (w *WhatsApp) Status() string {
        }
 }
 
        }
 }
 

+// BackupSession writes a consistent SQLite snapshot to path while the session
+// cannot be created, closed, or removed. The source opens without creation
+// permission. No session file means no linked state to copy.
+func (w *WhatsApp) BackupSession(ctx context.Context, path string) (err error) {
+       w.sessionMutex.Lock()
+       defer w.sessionMutex.Unlock()
+
+       w.mutex.Lock()
+       defer w.mutex.Unlock()
+
+       _, err = os.Stat(w.sessionPath())
+       if errors.Is(err, os.ErrNotExist) {
+               return nil
+       }
+
+       if err != nil {
+               return fmt.Errorf("read session file: %w", err)
+       }
+
+       err = os.MkdirAll(filepath.Dir(path), 0o700)
+       if err != nil {
+               return fmt.Errorf("create session snapshot directory: %w", err)
+       }
+
+       address, err := whatsappSessionAddress(w.sessionPath())
+       if err != nil {
+               return fmt.Errorf("find session database: %w", err)
+       }
+
+       database, err := sql.Open("sqlite", address+"&mode=rw")
+       if err != nil {
+               return fmt.Errorf("open session snapshot source: %w", err)
+       }
+
+       defer func() {
+               err = errors.Join(err, database.Close())
+       }()
+
+       _, err = database.ExecContext(ctx, "VACUUM INTO ?", path)
+       if err != nil {
+               return fmt.Errorf("snapshot session database: %w", err)
+       }
+
+       err = os.Chmod(path, 0o600)
+       if err != nil {
+               return fmt.Errorf("set session snapshot mode: %w", err)
+       }
+
+       return nil
+}
+

 // Close closes the session database.
 func (w *WhatsApp) Close() error {
 // Close closes the session database.
 func (w *WhatsApp) Close() error {

+       w.sessionMutex.Lock()
+       defer w.sessionMutex.Unlock()
+

        w.mutex.Lock()
        defer w.mutex.Unlock()
 
        w.mutex.Lock()
        defer w.mutex.Unlock()
 


@@ -618,8 +676,12 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str
 }
 
 // removeSession ends the connection, closes the session database, and deletes
 }
 
 // removeSession ends the connection, closes the session database, and deletes

-// its files.
+// its files. The session lock covers the cleanup after mutex is released, so a
+// snapshot or a new session cannot open files that cleanup removes.

 func (w *WhatsApp) removeSession() error {
 func (w *WhatsApp) removeSession() error {

+       w.sessionMutex.Lock()
+       defer w.sessionMutex.Unlock()
+

        w.mutex.Lock()
 
        client := w.client
        w.mutex.Lock()
 
        client := w.client


@@ -773,6 +835,9 @@ func (w *WhatsApp) pair(ctx, runCtx context.Context, l *TurnLoop, phone string,
 // directory and the file of the session first, with their modes, when they do
 // not exist.
 func (w *WhatsApp) newDevice(ctx context.Context) (*store.Device, error) {
 // directory and the file of the session first, with their modes, when they do
 // not exist.
 func (w *WhatsApp) newDevice(ctx context.Context) (*store.Device, error) {

+       w.sessionMutex.Lock()
+       defer w.sessionMutex.Unlock()
+

        w.mutex.Lock()
        defer w.mutex.Unlock()
 
        w.mutex.Lock()
        defer w.mutex.Unlock()