--- /dev/null
+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
+}
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 {
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)
}
- 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 = "", ""
- 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
}
- 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, 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 {
- 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
}
- 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, "/")
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)
}
--- /dev/null
+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
+}
import (
"archive/tar"
- "compress/gzip"
"os"
"path/filepath"
"testing"
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
-}
--- /dev/null
+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
+}
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 {
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)
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":
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 {
// 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"`
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()
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()
import (
"context"
+ "database/sql"
"errors"
"fmt"
"log/slog"
// 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
groups *WhatsAppGroups
inbox *WhatsAppInbox
+ sessionMutex sync.Mutex
mutex sync.Mutex
container *sqlstore.Container
device *store.Device
}
}
+// 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 {
+ w.sessionMutex.Lock()
+ defer w.sessionMutex.Unlock()
+
w.mutex.Lock()
defer w.mutex.Unlock()
}
// 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 {
+ w.sessionMutex.Lock()
+ defer w.sessionMutex.Unlock()
+
w.mutex.Lock()
client := w.client
// 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()
projects / onechat / commitdiff