diff --git a/docs/COMMANDS-GUIDE.md b/docs/COMMANDS-GUIDE.md index 8663b84b..ee17d757 100644 --- a/docs/COMMANDS-GUIDE.md +++ b/docs/COMMANDS-GUIDE.md @@ -72,7 +72,7 @@ Portable AI snapshots exclude skill roots, auth by default, caches, histories, d ## claude-mem peer sync -`dot ai memory sync [push|pull|sync|sessions|status]` replicates the claude-mem store (`~/.claude-mem/claude-mem.db`) with a peer Mac over ssh; the peer needs only the dot binary, which serves `max`/`export`/`import`/`count` as JSON over stdio via the hidden `--serve` flag — no python or claude-mem tooling on either side. Incremental export uses the receiver's `MAX(created_at)` minus a 7-day overlap (`--full` disables it); dedupe keys are `(created_at, title)` for observations, `(created_at, request)` for summaries, and `(platform_source, content_session_id)`/`memory_session_id` for sessions. Inserts use the column intersection of both sides minus identity/sync columns, so plugin version skew skips a column (logged once) instead of failing; previously copied rows get a one-off attribution backfill (`agent_type`/`agent_id`/`metadata`/`content_hash`/`generated_by_model`). An import that adds rows kicks the local worker's `/api/admin/restart` when it runs. `dot ai memory install --peer ` installs the hourly LaunchAgent `com.dotfiles.claude-mem-sync` (ssh pre-check `ssh true || exit 0`, so an away peer is a skipped run, exit 0); each scheduled pass refreshes claude-mem to the latest marketplace version first (also `dot ai memory update` and a `dot ai update` step), keeping both Macs on the same plugin version. Last run/result and both sides' counts live in `~/.claude-mem/sync-state.json` and the `dot ai memory status` "peer sync" section. +`dot ai memory sync [push|pull|sync|sessions|status]` replicates the claude-mem store (`~/.claude-mem/claude-mem.db`) with a peer Mac over ssh; the peer needs only the dot binary, which serves `max`/`export`/`import`/`count` as JSON over stdio via the hidden `--serve` flag — no python or claude-mem tooling on either side. Incremental export uses the receiver's `MAX(created_at)` minus a 7-day overlap (`--full` disables it); dedupe keys are `(created_at, title)` for observations, `(created_at, request)` for summaries, and `(platform_source, content_session_id)`/`memory_session_id` for sessions. Inserts use the column intersection of both sides minus identity/sync columns, so plugin version skew skips a column (logged once) instead of failing; previously copied rows get a one-off attribution backfill (`agent_type`/`agent_id`/`metadata`/`content_hash`/`generated_by_model`). An import that adds rows kicks the local worker's `/api/admin/restart` when it runs. A run is bounded by `--timeout` (default 10m): past it, or on SIGTERM, the open import rolls back and the run exits non-zero naming the phase it stopped in; each phase logs its start and end with row counts. `dot ai memory install --peer ` installs the hourly LaunchAgent `com.dotfiles.claude-mem-sync` (ssh pre-check `ssh true || exit 0`, so an away peer is a skipped run, exit 0; a watchdog sends SIGTERM after 15 min and SIGKILL 30 s later, since launchd starts no new run while one is alive); each scheduled pass refreshes claude-mem to the latest marketplace version first (also `dot ai memory update` and a `dot ai update` step), keeping both Macs on the same plugin version. Last run/result and both sides' counts live in `~/.claude-mem/sync-state.json` and the `dot ai memory status` "peer sync" section. ## Apps and profile snapshots diff --git a/docs/commands/dot_ai_memory_sync.md b/docs/commands/dot_ai_memory_sync.md index 8ac9073d..07797c74 100644 --- a/docs/commands/dot_ai_memory_sync.md +++ b/docs/commands/dot_ai_memory_sync.md @@ -10,6 +10,10 @@ with a peer Mac. Incremental by default (receiver's MAX(created_at) minus a needs only the dot binary: the transport is 'ssh dot ai memory sync --serve ' with JSON over stdio. +A run is bounded by --timeout: past it, or on SIGTERM, the open import +rolls back and the run exits non-zero naming the phase it stopped in. +Each phase logs its start and end to stderr. + sessions replicates all sessions both ways. status prints both sides' counts plus the last scheduled run. With no action, sync runs both directions. --peer defaults to the configured dot peer target. @@ -25,6 +29,7 @@ dot ai memory sync [push|pull|sync|sessions|status] [flags] -h, --help help for sync --peer string ssh target to sync with (default: the dot peer target) --remote-db string DB path on the peer (default: the peer's ~/.claude-mem/claude-mem.db) + --timeout duration abort the run (and roll back an open import) after this long (default 10m0s) ``` ### Options inherited from parent commands diff --git a/internal/aisettings/claudemem_sync.go b/internal/aisettings/claudemem_sync.go index 55da58a6..bac381f9 100644 --- a/internal/aisettings/claudemem_sync.go +++ b/internal/aisettings/claudemem_sync.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "io" + "log/slog" "net/http" "os" "path/filepath" @@ -142,11 +143,11 @@ func DefaultSyncDBPath(homeDir string) string { // MaxCreatedAt is the serve `max` op: the receiver's newest created_at per // replicating table ("" when the table is empty). -func (d *SyncDB) MaxCreatedAt() (map[string]string, error) { +func (d *SyncDB) MaxCreatedAt(ctx context.Context) (map[string]string, error) { out := map[string]string{} for _, table := range []string{"observations", "session_summaries"} { var max sql.NullString - if err := d.db.QueryRow("SELECT MAX(created_at) FROM " + table).Scan(&max); err != nil { + if err := d.db.QueryRowContext(ctx, "SELECT MAX(created_at) FROM "+table).Scan(&max); err != nil { return nil, fmt.Errorf("max created_at on %s: %w", table, err) } if max.Valid { @@ -157,7 +158,7 @@ func (d *SyncDB) MaxCreatedAt() (map[string]string, error) { } // Counts is the serve `count` op. -func (d *SyncDB) Counts() (*TableCounts, error) { +func (d *SyncDB) Counts(ctx context.Context) (*TableCounts, error) { var c TableCounts queries := []struct { dst *int @@ -171,7 +172,7 @@ func (d *SyncDB) Counts() (*TableCounts, error) { WHERE s.id IS NULL`}, } for _, q := range queries { - if err := d.db.QueryRow(q.stmt).Scan(q.dst); err != nil { + if err := d.db.QueryRowContext(ctx, q.stmt).Scan(q.dst); err != nil { return nil, err } } @@ -208,22 +209,22 @@ func CutoffFor(receiverMax string, now time.Time) (string, error) { // Sessions ride along when they are newer than the oldest cutoff OR back an // exported observation/summary. onlySessions is the sessions-mode pass: // every session, no content rows. -func (d *SyncDB) Export(cutoffs map[string]string, onlySessions bool) (*Bundle, error) { +func (d *SyncDB) Export(ctx context.Context, cutoffs map[string]string, onlySessions bool) (*Bundle, error) { b := &Bundle{Cutoff: oldestCutoff(cutoffs)} var err error if onlySessions { - b.Sessions, err = d.selectRows("SELECT * FROM sdk_sessions") + b.Sessions, err = d.selectRows(ctx, "SELECT * FROM sdk_sessions") if err == nil { stripSkipColumns(b.Sessions) } return b, err } obsWhere, obsArgs := cutoffWhere(cutoffs["observations"]) - if b.Observations, err = d.selectRows("SELECT * FROM observations"+obsWhere, obsArgs...); err != nil { + if b.Observations, err = d.selectRows(ctx, "SELECT * FROM observations"+obsWhere, obsArgs...); err != nil { return nil, err } sumWhere, sumArgs := cutoffWhere(cutoffs["session_summaries"]) - if b.Summaries, err = d.selectRows("SELECT * FROM session_summaries"+sumWhere, sumArgs...); err != nil { + if b.Summaries, err = d.selectRows(ctx, "SELECT * FROM session_summaries"+sumWhere, sumArgs...); err != nil { return nil, err } sessionSQL := `SELECT * FROM sdk_sessions WHERE memory_session_id IN ( @@ -234,7 +235,7 @@ func (d *SyncDB) Export(cutoffs map[string]string, onlySessions bool) (*Bundle, sessionSQL += " OR started_at > ?" sessionArgs = append(sessionArgs, oldest) } - if b.Sessions, err = d.selectRows(sessionSQL, sessionArgs...); err != nil { + if b.Sessions, err = d.selectRows(ctx, sessionSQL, sessionArgs...); err != nil { return nil, err } // The skip set must not cross the wire either: id is receiver-assigned @@ -282,8 +283,8 @@ func oldestCutoff(cutoffs map[string]string) string { // selectRows runs a query and returns every row as a column→value map. // []byte values are normalized to strings so a Bundle is JSON-clean. -func (d *SyncDB) selectRows(query string, args ...any) ([]map[string]any, error) { - rows, err := d.db.Query(query, args...) +func (d *SyncDB) selectRows(ctx context.Context, query string, args ...any) ([]map[string]any, error) { + rows, err := d.db.QueryContext(ctx, query, args...) if err != nil { return nil, err } @@ -316,8 +317,8 @@ func (d *SyncDB) selectRows(query string, args ...any) ([]map[string]any, error) } // tableColumns is the receiver-side column set for the import intersection. -func (d *SyncDB) tableColumns(table string) (map[string]bool, error) { - rows, err := d.db.Query("PRAGMA table_info(" + table + ")") +func (d *SyncDB) tableColumns(ctx context.Context, table string) (map[string]bool, error) { + rows, err := d.db.QueryContext(ctx, "PRAGMA table_info("+table+")") if err != nil { return nil, err } @@ -354,11 +355,13 @@ func intersect(row map[string]any, receiverCols map[string]bool) (keep []string, } // Import is the serve `import` op: sessions first (content rows reference -// them), then observations, then summaries, in one transaction. -func (d *SyncDB) Import(b *Bundle) (*ImportResult, error) { - tx, err := d.db.Begin() +// them), then observations, then summaries, in one transaction. ctx bounds +// the whole run: cancellation interrupts the current statement and the +// deferred rollback leaves no partial import. +func (d *SyncDB) Import(ctx context.Context, b *Bundle) (*ImportResult, error) { + tx, err := d.db.BeginTx(ctx, nil) if err != nil { - return nil, err + return nil, fmt.Errorf("begin import: %w", err) } defer func() { _ = tx.Rollback() }() res := &ImportResult{} @@ -371,7 +374,8 @@ func (d *SyncDB) Import(b *Bundle) (*ImportResult, error) { } } - sessionCols, err := d.tableColumns("sdk_sessions") + slog.Info("memory sync import", "table", "sdk_sessions", "rows", len(b.Sessions)) + sessionCols, err := d.tableColumns(ctx, "sdk_sessions") if err != nil { return nil, err } @@ -384,7 +388,7 @@ func (d *SyncDB) Import(b *Bundle) (*ImportResult, error) { } seenIdentities[incomingID] = identity } - canonicalID, added, err := importSession(tx, sessionCols, row, logSkew) + canonicalID, added, err := importSession(ctx, tx, sessionCols, row, logSkew) if err != nil { return nil, fmt.Errorf("importing sdk_sessions: %w", err) } @@ -399,7 +403,7 @@ func (d *SyncDB) Import(b *Bundle) (*ImportResult, error) { } } for from, to := range sessionIDMap { - result, err := tx.Exec("UPDATE observations SET memory_session_id = ? WHERE memory_session_id = ?", to, from) + result, err := tx.ExecContext(ctx, "UPDATE observations SET memory_session_id = ? WHERE memory_session_id = ?", to, from) if err != nil { return nil, fmt.Errorf("relinking observations for session %q: %w", from, err) } @@ -408,7 +412,7 @@ func (d *SyncDB) Import(b *Bundle) (*ImportResult, error) { } else { res.RelinkedObservations += int(n) } - result, err = tx.Exec("UPDATE session_summaries SET memory_session_id = ? WHERE memory_session_id = ?", to, from) + result, err = tx.ExecContext(ctx, "UPDATE session_summaries SET memory_session_id = ? WHERE memory_session_id = ?", to, from) if err != nil { return nil, fmt.Errorf("relinking summaries for session %q: %w", from, err) } @@ -418,13 +422,19 @@ func (d *SyncDB) Import(b *Bundle) (*ImportResult, error) { res.RelinkedSummaries += int(n) } } - obsCols, err := d.tableColumns("observations") + + slog.Info("memory sync import", "table", "observations", "rows", len(b.Observations)) + obsCols, err := d.tableColumns(ctx, "observations") if err != nil { return nil, err } + obsKeys, err := naturalKeys(ctx, tx, "observations", "title", b.Observations) + if err != nil { + return nil, fmt.Errorf("importing observations: %w", err) + } for _, row := range b.Observations { mapped := remapSessionRow(row, sessionIDMap) - added, repaired, err := importObservation(tx, obsCols, mapped, logSkew) + added, repaired, err := importObservation(ctx, tx, obsCols, obsKeys, mapped, logSkew) if err != nil { return nil, fmt.Errorf("importing observations: %w", err) } @@ -433,13 +443,19 @@ func (d *SyncDB) Import(b *Bundle) (*ImportResult, error) { } res.Repaired += repaired } - sumCols, err := d.tableColumns("session_summaries") + + slog.Info("memory sync import", "table", "session_summaries", "rows", len(b.Summaries)) + sumCols, err := d.tableColumns(ctx, "session_summaries") if err != nil { return nil, err } + sumKeys, err := naturalKeys(ctx, tx, "session_summaries", "request", b.Summaries) + if err != nil { + return nil, fmt.Errorf("importing session_summaries: %w", err) + } for _, row := range b.Summaries { mapped := remapSessionRow(row, sessionIDMap) - added, err := importDeduped(tx, sumCols, "session_summaries", "request", mapped, logSkew) + added, err := importDeduped(ctx, tx, sumCols, sumKeys, "session_summaries", "request", mapped, logSkew) if err != nil { return nil, fmt.Errorf("importing session_summaries: %w", err) } @@ -464,38 +480,75 @@ func rowString(row map[string]any, col string) string { return "" } -// dedupeExists is the natural-key lookup: (created_at, COALESCE(key,”)). -func dedupeExists(tx *sql.Tx, table, keyCol string, row map[string]any) (bool, error) { - var n int - err := tx.QueryRow(`SELECT COUNT(1) FROM `+table+` WHERE created_at = ? AND COALESCE(`+keyCol+`, '') = ?`, - rowString(row, "created_at"), rowString(row, keyCol)).Scan(&n) - return n > 0, err +// keySet maps the natural key (created_at, COALESCE(key,”)) to the +// receiver rowids that carry it. +type keySet map[[2]string][]int64 + +func naturalKey(row map[string]any, keyCol string) [2]string { + return [2]string{rowString(row, "created_at"), rowString(row, keyCol)} } -// insertRow inserts one row; inserted=false means a receiver-side unique -// index the natural key does not cover (e.g. ux_observations_session_hash) -// settled for "already here": the row is by definition not new. -func insertRow(tx *sql.Tx, table string, cols []string, row map[string]any) (bool, error) { +// naturalKeys loads the receiver's natural keys from the bundle's oldest +// created_at on, in one pass. created_at has no index, so a lookup per +// incoming row scanned the whole table each time: a 50k-row overlap on a +// 220k-row store ran for hours under the write lock (#289). +func naturalKeys(ctx context.Context, tx *sql.Tx, table, keyCol string, rows []map[string]any) (keySet, error) { + keys := keySet{} + if len(rows) == 0 { + return keys, nil + } + from := rowString(rows[0], "created_at") + for _, row := range rows { + if c := rowString(row, "created_at"); c < from { + from = c + } + } + q, err := tx.QueryContext(ctx, `SELECT rowid, created_at, COALESCE(`+keyCol+`, '') FROM `+table+` WHERE created_at >= ?`, from) + if err != nil { + return nil, err + } + defer q.Close() + for q.Next() { + var id int64 + var createdAt, key string + if err := q.Scan(&id, &createdAt, &key); err != nil { + return nil, err + } + k := [2]string{createdAt, key} + keys[k] = append(keys[k], id) + } + return keys, q.Err() +} + +// insertRow inserts one row and returns its rowid; inserted=false means a +// receiver-side unique index the natural key does not cover (e.g. +// ux_observations_session_hash) settled for "already here": the row is by +// definition not new. +func insertRow(ctx context.Context, tx *sql.Tx, table string, cols []string, row map[string]any) (int64, bool, error) { marks := strings.TrimSuffix(strings.Repeat("?,", len(cols)), ",") stmt := `INSERT INTO ` + table + ` (` + strings.Join(cols, ", ") + `) VALUES (` + marks + `)` args := make([]any, len(cols)) for i, col := range cols { args[i] = row[col] } - _, err := tx.Exec(stmt, args...) - if err != nil && strings.Contains(err.Error(), "UNIQUE constraint") { - return false, nil + result, err := tx.ExecContext(ctx, stmt, args...) + if err != nil { + if strings.Contains(err.Error(), "UNIQUE constraint") { + return 0, false, nil + } + return 0, false, err } - return err == nil, err + id, err := result.LastInsertId() + return id, err == nil, err } -func importSession(tx *sql.Tx, receiverCols map[string]bool, row map[string]any, logSkew func([]string)) (string, bool, error) { +func importSession(ctx context.Context, tx *sql.Tx, receiverCols map[string]bool, row map[string]any, logSkew func([]string)) (string, bool, error) { incomingID := rowString(row, "memory_session_id") incomingPlatform := rowString(row, "platform_source") incomingContentID := rowString(row, "content_session_id") if incomingID != "" { var platform, contentID sql.NullString - err := tx.QueryRow(`SELECT platform_source, content_session_id FROM sdk_sessions WHERE memory_session_id = ?`, incomingID).Scan(&platform, &contentID) + err := tx.QueryRowContext(ctx, `SELECT platform_source, content_session_id FROM sdk_sessions WHERE memory_session_id = ?`, incomingID).Scan(&platform, &contentID) if err == nil { if incomingPlatform == "" || incomingContentID == "" { return "", false, fmt.Errorf("memory session %q has an incomplete platform/content identity", incomingID) @@ -513,7 +566,7 @@ func importSession(tx *sql.Tx, receiverCols map[string]bool, row map[string]any, // Only a complete identity can establish equivalence. Blank or NULL fields // are deliberately left unmapped because they cannot safely identify a session. if incomingPlatform != "" && incomingContentID != "" { - rows, err := tx.Query(`SELECT memory_session_id FROM sdk_sessions + rows, err := tx.QueryContext(ctx, `SELECT memory_session_id FROM sdk_sessions WHERE platform_source = ? AND content_session_id = ?`, incomingPlatform, incomingContentID) if err != nil { return "", false, err @@ -544,7 +597,7 @@ func importSession(tx *sql.Tx, receiverCols map[string]bool, row map[string]any, keep, skewed := intersect(row, receiverCols) logSkew(skewed) - added, err := insertRowStrict(tx, "sdk_sessions", keep, row) + added, err := insertRowStrict(ctx, tx, "sdk_sessions", keep, row) if err != nil { return "", false, err } @@ -556,14 +609,14 @@ func importSession(tx *sql.Tx, receiverCols map[string]bool, row map[string]any, // insertRowStrict is used for backing sessions: silently treating a unique // collision as a duplicate could leave incoming content pointing at no session. -func insertRowStrict(tx *sql.Tx, table string, cols []string, row map[string]any) (bool, error) { +func insertRowStrict(ctx context.Context, tx *sql.Tx, table string, cols []string, row map[string]any) (bool, error) { marks := strings.TrimSuffix(strings.Repeat("?,", len(cols)), ",") stmt := `INSERT INTO ` + table + ` (` + strings.Join(cols, ", ") + `) VALUES (` + marks + `)` args := make([]any, len(cols)) for i, col := range cols { args[i] = row[col] } - result, err := tx.Exec(stmt, args...) + result, err := tx.ExecContext(ctx, stmt, args...) if err != nil { return false, err } @@ -591,55 +644,57 @@ func remapSessionRow(row map[string]any, sessionIDMap map[string]string) map[str // importObservation dedupes on (created_at, title); a dedupe hit runs the // one-off attribution repair, backfilling repairColumns the local row lacks // when the incoming bundle carries them. -func importObservation(tx *sql.Tx, receiverCols map[string]bool, row map[string]any, logSkew func([]string)) (bool, int, error) { - exists, err := dedupeExists(tx, "observations", "title", row) - if err != nil { - return false, 0, err +func importObservation(ctx context.Context, tx *sql.Tx, receiverCols map[string]bool, keys keySet, row map[string]any, logSkew func([]string)) (bool, int, error) { + k := naturalKey(row, "title") + if ids, exists := keys[k]; exists { + repaired, err := repairAttribution(ctx, tx, receiverCols, ids, row) + return false, repaired, err } - if !exists { - keep, skewed := intersect(row, receiverCols) - logSkew(skewed) - added, err := insertRow(tx, "observations", keep, row) - return added, 0, err + keep, skewed := intersect(row, receiverCols) + logSkew(skewed) + id, added, err := insertRow(ctx, tx, "observations", keep, row) + if added { + keys[k] = []int64{id} } - repaired, err := repairAttribution(tx, receiverCols, row) - return false, repaired, err + return added, 0, err } -func repairAttribution(tx *sql.Tx, receiverCols map[string]bool, row map[string]any) (int, error) { +func repairAttribution(ctx context.Context, tx *sql.Tx, receiverCols map[string]bool, ids []int64, row map[string]any) (int, error) { repaired := 0 for _, col := range repairColumns { incoming := rowString(row, col) if incoming == "" || !receiverCols[col] { continue } - r, err := tx.Exec(`UPDATE observations SET `+col+` = ? - WHERE created_at = ? AND COALESCE(title, '') = ? AND (`+col+` IS NULL OR `+col+` = '')`, - incoming, rowString(row, "created_at"), rowString(row, "title")) - if err != nil { - if strings.Contains(err.Error(), "UNIQUE constraint") { - continue // content_hash would collide with the session-hash index + for _, id := range ids { + r, err := tx.ExecContext(ctx, `UPDATE observations SET `+col+` = ? + WHERE rowid = ? AND (`+col+` IS NULL OR `+col+` = '')`, incoming, id) + if err != nil { + if strings.Contains(err.Error(), "UNIQUE constraint") { + continue // content_hash would collide with the session-hash index + } + return repaired, err + } + if n, _ := r.RowsAffected(); n > 0 { + repaired += int(n) } - return repaired, err - } - if n, _ := r.RowsAffected(); n > 0 { - repaired += int(n) } } return repaired, nil } -func importDeduped(tx *sql.Tx, receiverCols map[string]bool, table, keyCol string, row map[string]any, logSkew func([]string)) (bool, error) { - exists, err := dedupeExists(tx, table, keyCol, row) - if err != nil { - return false, err - } - if exists { +func importDeduped(ctx context.Context, tx *sql.Tx, receiverCols map[string]bool, keys keySet, table, keyCol string, row map[string]any, logSkew func([]string)) (bool, error) { + k := naturalKey(row, keyCol) + if _, exists := keys[k]; exists { return false, nil } keep, skewed := intersect(row, receiverCols) logSkew(skewed) - return insertRow(tx, table, keep, row) + id, added, err := insertRow(ctx, tx, table, keep, row) + if added { + keys[k] = []int64{id} + } + return added, err } // KickWorkerRestart asks the local claude-mem worker to rebuild its search diff --git a/internal/aisettings/claudemem_sync_test.go b/internal/aisettings/claudemem_sync_test.go index 986aba75..6dbff506 100644 --- a/internal/aisettings/claudemem_sync_test.go +++ b/internal/aisettings/claudemem_sync_test.go @@ -154,7 +154,7 @@ func TestSyncRoundTrip(t *testing.T) { 'claude', 'agent-7', '{"k":"v"}', 'hash-1', 'haiku', 12345, 'macbook', 42, 9)`) seedSummary(t, sender, "mem-1", "2026-09-20T10:02:00.000Z", "req-1") - bundle, err := sender.Export(nil, false) + bundle, err := sender.Export(context.Background(), nil, false) if err != nil { t.Fatalf("Export: %v", err) } @@ -168,7 +168,7 @@ func TestSyncRoundTrip(t *testing.T) { t.Errorf("bundle carries skip column %q", col) } } - res, err := receiver.Import(bundle) + res, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatalf("Import: %v", err) } @@ -201,7 +201,7 @@ func TestSyncRoundTrip(t *testing.T) { } // Idempotency: a second import of the same bundle adds nothing. - res2, err := receiver.Import(bundle) + res2, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatalf("second Import: %v", err) } @@ -230,7 +230,7 @@ func TestImportSessionsRelinksExistingReferences(t *testing.T) { t.Fatal(err) } - result, err := receiver.Import(bundle) + result, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatalf("Import sessions: %v", err) } @@ -249,7 +249,7 @@ func TestImportSessionsRelinksExistingReferences(t *testing.T) { } unknown := "unknown-history" seedObs(t, receiver, unknown, "2026-09-20T10:03:00.000Z", "unknown observation") - result, err = receiver.Import(bundle) + result, err = receiver.Import(context.Background(), bundle) if err != nil { t.Fatalf("second Import sessions: %v", err) } @@ -287,7 +287,7 @@ func TestImportMapsIncomingContentToEquivalentSession(t *testing.T) { if err != nil { t.Fatal(err) } - result, err := receiver.Import(bundle) + result, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatalf("Import: %v", err) } @@ -304,7 +304,7 @@ func TestImportMapsIncomingContentToEquivalentSession(t *testing.T) { if err != nil || !bytes.Equal(before, after) { t.Fatalf("Import mutated caller bundle: err=%v", err) } - result, err = receiver.Import(bundle) + result, err = receiver.Import(context.Background(), bundle) if err != nil { t.Fatalf("second Import: %v", err) } @@ -326,7 +326,7 @@ func TestImportDoesNotMapBlankOrCrossPlatformIdentity(t *testing.T) { {"memory_session_id": "blank-identity-id", "project": "proj", "type": "discovery", "title": "blank-identity", "created_at": "2026-09-20T10:02:00.000Z", "created_at_epoch": 0}, }, } - if _, err := receiver.Import(bundle); err != nil { + if _, err := receiver.Import(context.Background(), bundle); err != nil { t.Fatalf("Import: %v", err) } for title, want := range map[string]string{"cross-platform": "cross-platform-id", "blank-identity": "blank-identity-id"} { @@ -344,14 +344,14 @@ func TestImportConflictingMemoryIdentityFails(t *testing.T) { equivalentSessionRow("new-id", "content-2", "claude"), equivalentSessionRow("existing-id", "different-content", "claude"), }} - if _, err := receiver.Import(bundle); err == nil { + if _, err := receiver.Import(context.Background(), bundle); err == nil { t.Fatal("Import accepted an existing memory ID with a different identity") } var count int if err := receiver.db.QueryRow("SELECT COUNT(*) FROM sdk_sessions WHERE memory_session_id = 'new-id'").Scan(&count); err != nil || count != 0 { t.Fatalf("earlier session insert survived rollback: count=%d err=%v", count, err) } - if _, err := receiver.Import(&Bundle{Sessions: []map[string]any{ + if _, err := receiver.Import(context.Background(), &Bundle{Sessions: []map[string]any{ equivalentSessionRow("existing-id", "", ""), }}); err == nil { t.Fatal("Import accepted an incomplete identity for an existing memory ID") @@ -365,7 +365,7 @@ func TestImportAmbiguousSessionIdentityFails(t *testing.T) { bundle := &Bundle{Sessions: []map[string]any{ equivalentSessionRow("incoming-id", "content-1", "claude"), }} - if _, err := receiver.Import(bundle); err == nil { + if _, err := receiver.Import(context.Background(), bundle); err == nil { t.Fatal("Import accepted an ambiguous platform/content identity") } var count int @@ -385,7 +385,7 @@ func TestImportRelinkUniqueConflictRollsBack(t *testing.T) { equivalentSessionRow("new-id", "content-2", "claude"), equivalentSessionRow("incoming-id", "content-1", "claude"), }} - if _, err := receiver.Import(bundle); err == nil { + if _, err := receiver.Import(context.Background(), bundle); err == nil { t.Fatal("Import accepted a relink that violates the observation session/hash index") } var got string @@ -415,11 +415,11 @@ func TestSyncSchemaSkew(t *testing.T) { execIns(t, sender, `INSERT INTO observations (memory_session_id, project, type, title, created_at, created_at_epoch, metadata, agent_type) VALUES ('m1', 'proj', 'discovery', 'T1', '2026-09-20T10:01:00.000Z', 0, '{"k":"v"}', 'claude')`) - bundle, err := sender.Export(nil, false) + bundle, err := sender.Export(context.Background(), nil, false) if err != nil { t.Fatal(err) } - res, err := receiver.Import(bundle) + res, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatalf("skewed Import must succeed: %v", err) } @@ -451,11 +451,11 @@ func TestSyncAttributionRepair(t *testing.T) { VALUES ('m1', 'proj', 'discovery', 'T1', '2026-09-20T10:01:00.000Z', 0, 'kimi', 'agent-9', '{"m":1}', 'hash-9', 'k2')`) - bundle, err := sender.Export(nil, false) + bundle, err := sender.Export(context.Background(), nil, false) if err != nil { t.Fatal(err) } - res, err := receiver.Import(bundle) + res, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatal(err) } @@ -473,7 +473,7 @@ func TestSyncAttributionRepair(t *testing.T) { t.Fatalf("repair incomplete: %q %q %q %q %q", agentType, agentID, metadata, hash, model) } // Repair is one-off: a third pass repairs nothing. - res2, err := receiver.Import(bundle) + res2, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatal(err) } @@ -493,7 +493,7 @@ func TestExportCutoff(t *testing.T) { seedSummary(t, db, "m-old", "2026-09-01T02:00:00.000Z", "old-req") seedSummary(t, db, "m-new", "2026-09-25T02:00:00.000Z", "new-req") - bundle, err := db.Export(map[string]string{"observations": "2026-09-20T00:00:00.000Z", "session_summaries": "2026-09-20T00:00:00.000Z"}, false) + bundle, err := db.Export(context.Background(), map[string]string{"observations": "2026-09-20T00:00:00.000Z", "session_summaries": "2026-09-20T00:00:00.000Z"}, false) if err != nil { t.Fatal(err) } @@ -510,7 +510,7 @@ func TestExportCutoff(t *testing.T) { // An old session IS included when an exported (overlapping) row backs it. seedObs(t, db, "m-old", "2026-09-26T01:00:00.000Z", "overlap") - bundle, err = db.Export(map[string]string{"observations": "2026-09-20T00:00:00.000Z", "session_summaries": "2026-09-20T00:00:00.000Z"}, false) + bundle, err = db.Export(context.Background(), map[string]string{"observations": "2026-09-20T00:00:00.000Z", "session_summaries": "2026-09-20T00:00:00.000Z"}, false) if err != nil { t.Fatal(err) } @@ -532,7 +532,7 @@ func TestExportPerTableCutoffs(t *testing.T) { // Observations are exported against a September cutoff while summaries // use their own January one: the ancient summary must still be offered. - bundle, err := db.Export(map[string]string{ + bundle, err := db.Export(context.Background(), map[string]string{ "observations": "2026-09-20T00:00:00.000Z", "session_summaries": "2025-12-25T00:00:00.000Z", }, false) @@ -553,7 +553,7 @@ func TestExportOnlySessions(t *testing.T) { seedSession(t, db, "c2", "m2", "2026-09-25T00:00:00.000Z") seedObs(t, db, "m1", "2026-09-26T01:00:00.000Z", "x") - bundle, err := db.Export(nil, true) + bundle, err := db.Export(context.Background(), nil, true) if err != nil { t.Fatal(err) } @@ -591,14 +591,14 @@ func TestMaxAndCounts(t *testing.T) { VALUES ('m-ghost', 'proj', 'discovery', 'orphan', '2026-09-21T10:01:00.000Z', 0)`) seedSummary(t, db, "m1", "2026-09-22T10:02:00.000Z", "req") - maxes, err := db.MaxCreatedAt() + maxes, err := db.MaxCreatedAt(context.Background()) if err != nil { t.Fatal(err) } if maxes["observations"] != "2026-09-21T10:01:00.000Z" || maxes["session_summaries"] != "2026-09-22T10:02:00.000Z" { t.Fatalf("maxes = %v", maxes) } - counts, err := db.Counts() + counts, err := db.Counts(context.Background()) if err != nil { t.Fatal(err) } @@ -672,7 +672,7 @@ func TestImportSessionWithUnassignedMemoryID(t *testing.T) { execIns(t, receiver, `INSERT INTO sdk_sessions (content_session_id, memory_session_id, project, platform_source, started_at, started_at_epoch) VALUES ('pending-content', NULL, 'proj', 'claude', '2026-09-20', 0)`) bundle := &Bundle{Sessions: []map[string]any{{"content_session_id": "pending-content", "memory_session_id": incomingID, "platform_source": "claude"}}} for pass := 0; pass < 2; pass++ { - result, err := receiver.Import(bundle) + result, err := receiver.Import(context.Background(), bundle) if err != nil { t.Fatal(err) } @@ -704,7 +704,7 @@ func TestImportRejectsConflictingBundleSessionIdentity(t *testing.T) { if reverse { bundle.Sessions[0], bundle.Sessions[1] = bundle.Sessions[1], bundle.Sessions[0] } - if _, err := receiver.Import(bundle); err == nil { + if _, err := receiver.Import(context.Background(), bundle); err == nil { t.Fatal("conflicting session identities accepted") } var sessionCount int diff --git a/internal/aisettings/claudemem_syncagent.go b/internal/aisettings/claudemem_syncagent.go index a096fd7d..23376f27 100644 --- a/internal/aisettings/claudemem_syncagent.go +++ b/internal/aisettings/claudemem_syncagent.go @@ -33,14 +33,24 @@ func (m *ClaudeMemManager) SyncLogPath() string { return filepath.Join(m.DataDir(), "logs", "claude-mem-sync.log") } +// SyncAgentHardLimit is the wrapper's backstop in seconds: past it the run +// gets SIGTERM, then SIGKILL 30 s later. It sits above the command's own +// 10-minute --timeout, and launchd starts no new run while one is alive. +const SyncAgentHardLimit = 900 + // RenderSyncAgentPlist renders the hourly sync agent. The invoked command // starts with the ssh pre-check per the sync contract: an unreachable peer // exits 0 before dot runs, so launchd never records a failure for a laptop // that is simply away. DOT_SCHEDULED_RUN tells the sync command to apply -// the same policy to its own in-process probe. +// the same policy to its own in-process probe. A watchdog enforces +// SyncAgentHardLimit; launchd reaps its leftover sleep with the job. func RenderSyncAgentPlist(dotPath, peer, homeDir, logPath string) string { - script := fmt.Sprintf("ssh -o BatchMode=yes -o ConnectTimeout=10 %s true || exit 0\nexec %s ai memory sync --peer %s", - shellQuote(peer), shellQuote(dotPath), shellQuote(peer)) + script := fmt.Sprintf(`ssh -o BatchMode=yes -o ConnectTimeout=10 %s true || exit 0 +%s ai memory sync --peer %s & pid=$! +(sleep %d; kill -TERM $pid; sleep 30; kill -KILL $pid) 2>/dev/null & guard=$! +wait $pid; rc=$? +kill $guard 2>/dev/null +exit $rc`, shellQuote(peer), shellQuote(dotPath), shellQuote(peer), SyncAgentHardLimit) return fmt.Sprintf(` diff --git a/internal/aisettings/claudemem_transport.go b/internal/aisettings/claudemem_transport.go index 683b77f5..b9741bcb 100644 --- a/internal/aisettings/claudemem_transport.go +++ b/internal/aisettings/claudemem_transport.go @@ -216,11 +216,11 @@ func ServeOp(ctx context.Context, dbPath, home, op string, stdin io.Reader, stdo resp := serveResponse{} switch op { case "max": - if resp.Max, err = db.MaxCreatedAt(); err != nil { + if resp.Max, err = db.MaxCreatedAt(ctx); err != nil { resp.Error = err.Error() } case "export": - if resp.Bundle, err = db.Export(req.Cutoffs, req.OnlySessions); err != nil { + if resp.Bundle, err = db.Export(ctx, req.Cutoffs, req.OnlySessions); err != nil { resp.Error = err.Error() } case "import": @@ -228,7 +228,7 @@ func ServeOp(ctx context.Context, dbPath, home, op string, stdin io.Reader, stdo resp.Error = "import request carries no bundle" break } - if resp.Result, err = db.Import(req.Bundle); err != nil { + if resp.Result, err = db.Import(ctx, req.Bundle); err != nil { resp.Error = err.Error() break } @@ -239,7 +239,7 @@ func ServeOp(ctx context.Context, dbPath, home, op string, stdin io.Reader, stdo _ = KickWorkerRestart(ctx, home, nil) } case "count": - if resp.Counts, err = db.Counts(); err != nil { + if resp.Counts, err = db.Counts(ctx); err != nil { resp.Error = err.Error() } default: @@ -290,7 +290,8 @@ func receiverCutoffs(maxes map[string]string, full bool, now time.Time) (map[str // RunMemorySync executes one sync: push (local → peer), pull (peer → // local), or both. onlySessions exports every session both ways with no -// content rows and no cutoff. +// content rows and no cutoff. Each phase logs its start and end, and an +// error names the phase it stopped in, so a stall or a lock shows where. func RunMemorySync(ctx context.Context, local *SyncDB, t SyncTransport, peer SyncPeer, opts SyncOptions, now time.Time) (*SyncReport, error) { report := &SyncReport{Direction: opts.Direction} push := opts.Direction == "push" || opts.Direction == "sync" @@ -301,38 +302,57 @@ func RunMemorySync(ctx context.Context, local *SyncDB, t SyncTransport, peer Syn if !push && !pull { return nil, fmt.Errorf("unknown sync direction %q", opts.Direction) } + full := opts.Full || opts.OnlySessions if push { - maxes, err := t.Max(ctx, peer) - if err != nil { + var maxes map[string]string + var bundle *Bundle + if err := syncPhase("push: peer max", func() (n int, err error) { + maxes, err = t.Max(ctx, peer) + return len(maxes), err + }); err != nil { return nil, err } - cutoffs, err := receiverCutoffs(maxes, opts.Full || opts.OnlySessions, now) - if err != nil { - return nil, err - } - bundle, err := local.Export(cutoffs, opts.OnlySessions) - if err != nil { + if err := syncPhase("push: local export", func() (int, error) { + cutoffs, err := receiverCutoffs(maxes, full, now) + if err != nil { + return 0, err + } + bundle, err = local.Export(ctx, cutoffs, opts.OnlySessions) + return bundle.rows(), err + }); err != nil { return nil, err } - if report.Pushed, err = t.Import(ctx, peer, bundle); err != nil { + if err := syncPhase("push: peer import", func() (n int, err error) { + report.Pushed, err = t.Import(ctx, peer, bundle) + return report.Pushed.added(), err + }); err != nil { return nil, err } } if pull { - maxes, err := local.MaxCreatedAt() - if err != nil { + var maxes map[string]string + var bundle *Bundle + if err := syncPhase("pull: local max", func() (n int, err error) { + maxes, err = local.MaxCreatedAt(ctx) + return len(maxes), err + }); err != nil { return nil, err } - cutoffs, err := receiverCutoffs(maxes, opts.Full || opts.OnlySessions, now) - if err != nil { - return nil, err - } - bundle, err := t.Export(ctx, peer, cutoffs, opts.OnlySessions) - if err != nil { + if err := syncPhase("pull: peer export", func() (int, error) { + cutoffs, err := receiverCutoffs(maxes, full, now) + if err != nil { + return 0, err + } + bundle, err = t.Export(ctx, peer, cutoffs, opts.OnlySessions) + return bundle.rows(), err + }); err != nil { return nil, err } - if report.Pulled, err = local.Import(bundle); err != nil { + if err := syncPhase("pull: local import", func() (n int, err error) { + report.Pulled, err = local.Import(ctx, bundle) + return report.Pulled.added(), err + }); err != nil { return nil, err } } @@ -341,7 +361,7 @@ func RunMemorySync(ctx context.Context, local *SyncDB, t SyncTransport, peer Syn // not turn a completed sync into a reported failure (the rows landed; // the counts are the status row's convenience, not the result). var err error - if report.LocalCounts, err = local.Counts(); err != nil { + if report.LocalCounts, err = local.Counts(ctx); err != nil { report.LocalCounts = nil } if report.PeerCounts, err = t.Counts(ctx, peer); err != nil { @@ -350,6 +370,33 @@ func RunMemorySync(ctx context.Context, local *SyncDB, t SyncTransport, peer Syn return report, nil } +// syncPhase runs one phase between start and end log lines (with the +// phase's row count) and prefixes its error with the phase name. +func syncPhase(name string, fn func() (int, error)) error { + start := time.Now() + slog.Info("memory sync phase start", "phase", name) + n, err := fn() + if err != nil { + return fmt.Errorf("%s: %w", name, err) + } + slog.Info("memory sync phase end", "phase", name, "rows", n, "elapsed", time.Since(start).Round(time.Millisecond)) + return nil +} + +func (b *Bundle) rows() int { + if b == nil { + return 0 + } + return len(b.Observations) + len(b.Summaries) + len(b.Sessions) +} + +func (r *ImportResult) added() int { + if r == nil { + return 0 + } + return r.Obs + r.Sums + r.Sessions +} + // SyncStateEntry is one peer's last-run record for the status row. type SyncStateEntry struct { LastRun time.Time `json:"last_run"` diff --git a/internal/aisettings/claudemem_transport_test.go b/internal/aisettings/claudemem_transport_test.go index 0201c2ed..3d1e1f21 100644 --- a/internal/aisettings/claudemem_transport_test.go +++ b/internal/aisettings/claudemem_transport_test.go @@ -3,7 +3,10 @@ package aisettings import ( "bytes" "context" + "database/sql" "encoding/json" + "errors" + "log/slog" "net/http" "net/http/httptest" "os" @@ -19,17 +22,17 @@ type fakeTransport struct { db *SyncDB } -func (f *fakeTransport) Max(context.Context, SyncPeer) (map[string]string, error) { - return f.db.MaxCreatedAt() +func (f *fakeTransport) Max(ctx context.Context, _ SyncPeer) (map[string]string, error) { + return f.db.MaxCreatedAt(ctx) } -func (f *fakeTransport) Export(_ context.Context, _ SyncPeer, cutoffs map[string]string, onlySessions bool) (*Bundle, error) { - return f.db.Export(cutoffs, onlySessions) +func (f *fakeTransport) Export(ctx context.Context, _ SyncPeer, cutoffs map[string]string, onlySessions bool) (*Bundle, error) { + return f.db.Export(ctx, cutoffs, onlySessions) } -func (f *fakeTransport) Import(_ context.Context, _ SyncPeer, b *Bundle) (*ImportResult, error) { - return f.db.Import(b) +func (f *fakeTransport) Import(ctx context.Context, _ SyncPeer, b *Bundle) (*ImportResult, error) { + return f.db.Import(ctx, b) } -func (f *fakeTransport) Counts(context.Context, SyncPeer) (*TableCounts, error) { - return f.db.Counts() +func (f *fakeTransport) Counts(ctx context.Context, _ SyncPeer) (*TableCounts, error) { + return f.db.Counts(ctx) } func seedPair(t *testing.T) (local, remote *SyncDB) { @@ -61,7 +64,7 @@ func TestRunMemorySync_BothDirections(t *testing.T) { } // Both sides now hold each other's rows. for _, db := range []*SyncDB{local, remote} { - counts, err := db.Counts() + counts, err := db.Counts(context.Background()) if err != nil { t.Fatal(err) } @@ -91,7 +94,7 @@ func TestRunMemorySync_PushPullOnly(t *testing.T) { if report.Pushed == nil || report.Pulled != nil { t.Fatalf("push-only report = %#v", report) } - if counts, _ := local.Counts(); counts.Obs != 1 { + if counts, _ := local.Counts(context.Background()); counts.Obs != 1 { t.Fatalf("push must not pull: %#v", counts) } @@ -102,7 +105,7 @@ func TestRunMemorySync_PushPullOnly(t *testing.T) { if report.Pulled == nil || report.Pushed != nil { t.Fatalf("pull-only report = %#v", report) } - if counts, _ := local.Counts(); counts.Obs != 2 { + if counts, _ := local.Counts(context.Background()); counts.Obs != 2 { t.Fatalf("pull must land remote rows: %#v", counts) } @@ -119,10 +122,10 @@ func TestRunMemorySync_SessionsMode(t *testing.T) { t.Fatal(err) } // Sessions cross; content rows do not. - if counts, _ := local.Counts(); counts.Sessions != 2 || counts.Obs != 1 { + if counts, _ := local.Counts(context.Background()); counts.Sessions != 2 || counts.Obs != 1 { t.Fatalf("local after sessions pass = %#v", counts) } - if counts, _ := remote.Counts(); counts.Sessions != 2 || counts.Obs != 1 { + if counts, _ := remote.Counts(context.Background()); counts.Sessions != 2 || counts.Obs != 1 { t.Fatalf("remote after sessions pass = %#v", counts) } _ = report @@ -312,7 +315,7 @@ func TestServeOp_ImportKicksOnlyTheGivenHome(t *testing.T) { src := newTestSyncDB(t) seedSession(t, src, "c1", "m1", "2026-09-20T10:00:00.000Z") seedObs(t, src, "m1", "2026-09-20T10:01:00.000Z", "x") - bundle, err := src.Export(nil, false) + bundle, err := src.Export(context.Background(), nil, false) if err != nil { t.Fatal(err) } @@ -342,7 +345,7 @@ func TestServeOp_ImportKicksOnlyTheGivenHome(t *testing.T) { src2 := newTestSyncDB(t) seedSession(t, src2, "c2", "m2", "2026-09-21T10:00:00.000Z") seedObs(t, src2, "m2", "2026-09-21T10:01:00.000Z", "y") - bundle2, err := src2.Export(nil, false) + bundle2, err := src2.Export(context.Background(), nil, false) if err != nil { t.Fatal(err) } @@ -467,3 +470,120 @@ func TestServeOpSessionRelinkBackfill(t *testing.T) { t.Fatalf("worker got %d backfill kicks, want one relink kick and no idempotent kick", len(kicks)) } } + +// #289: a store whose write lock another connection holds fails the run +// after the busy timeout, naming the blocked phase, instead of spinning. +func TestRunMemorySync_LockedStoreFailsFastNamingPhase(t *testing.T) { + local, remote := seedPair(t) + holder, err := sql.Open("sqlite", "file:"+local.Path) + if err != nil { + t.Fatal(err) + } + defer holder.Close() + conn, err := holder.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err := conn.ExecContext(context.Background(), "BEGIN IMMEDIATE"); err != nil { + t.Fatal(err) + } + defer conn.ExecContext(context.Background(), "ROLLBACK") + + start := time.Now() + _, err = RunMemorySync(context.Background(), local, &fakeTransport{db: remote}, SyncPeer{Target: "fake"}, SyncOptions{Direction: "pull"}, time.Now()) + if err == nil { + t.Fatal("pull into a locked store succeeded") + } + if !strings.Contains(err.Error(), "pull: local import") || !strings.Contains(err.Error(), "SQLITE_BUSY") { + t.Fatalf("error should name the phase and SQLITE_BUSY: %v", err) + } + if elapsed := time.Since(start); elapsed > 15*time.Second { + t.Fatalf("locked run took %s", elapsed) + } +} + +// stallTransport never answers an export until its context ends. +type stallTransport struct{ fakeTransport } + +func (s *stallTransport) Export(ctx context.Context, _ SyncPeer, _ map[string]string, _ bool) (*Bundle, error) { + <-ctx.Done() + return nil, ctx.Err() +} + +// #289: the deadline ends a stalled phase and the error names it. +func TestRunMemorySync_DeadlineEndsStalledPhase(t *testing.T) { + local, remote := seedPair(t) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + start := time.Now() + _, err := RunMemorySync(ctx, local, &stallTransport{fakeTransport{db: remote}}, SyncPeer{Target: "fake"}, SyncOptions{Direction: "pull"}, time.Now()) + if !errors.Is(err, context.DeadlineExceeded) || !strings.Contains(err.Error(), "pull: peer export") { + t.Fatalf("err = %v", err) + } + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Fatalf("stalled run outlived its deadline by %s", elapsed) + } +} + +// cancelOnTable cancels the import when it reaches a table, standing in for +// a SIGTERM that arrives after earlier tables were written in the tx. +type cancelOnTable struct { + slog.Handler + table string + cancel context.CancelFunc +} + +func (h cancelOnTable) Handle(ctx context.Context, r slog.Record) error { + r.Attrs(func(a slog.Attr) bool { + if a.Key == "table" && a.Value.String() == h.table { + h.cancel() + } + return true + }) + return h.Handler.Handle(ctx, r) +} + +// #289: canceling mid-import rolls the whole transaction back. +func TestImport_CancelMidwayRollsBack(t *testing.T) { + sender, receiver := seedPair(t) + bundle, err := sender.Export(context.Background(), nil, false) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + prev := slog.Default() + slog.SetDefault(slog.New(cancelOnTable{Handler: slog.NewTextHandler(os.Stderr, nil), table: "observations", cancel: cancel})) + defer slog.SetDefault(prev) + + if _, err := receiver.Import(ctx, bundle); !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + counts, err := receiver.Counts(context.Background()) + if err != nil { + t.Fatal(err) + } + // seedPair gave the receiver one session and one observation of its own. + if counts.Sessions != 1 || counts.Obs != 1 || counts.Sums != 0 { + t.Fatalf("partial import left behind: %+v", counts) + } +} + +// The in-memory key set must see rows inserted earlier in the same import. +func TestImport_DedupesWithinOneBundle(t *testing.T) { + sender, receiver := seedPair(t) + bundle, err := sender.Export(context.Background(), nil, false) + if err != nil { + t.Fatal(err) + } + bundle.Observations = append(bundle.Observations, bundle.Observations[0]) + bundle.Summaries = append(bundle.Summaries, bundle.Summaries[0]) + res, err := receiver.Import(context.Background(), bundle) + if err != nil { + t.Fatal(err) + } + if res.Obs != 1 || res.Sums != 1 { + t.Fatalf("duplicate rows in one bundle imported twice: %+v", res) + } +} diff --git a/internal/aisettings/claudemem_update_test.go b/internal/aisettings/claudemem_update_test.go index a0ddbc67..82cfb2c7 100644 --- a/internal/aisettings/claudemem_update_test.go +++ b/internal/aisettings/claudemem_update_test.go @@ -174,7 +174,9 @@ func TestRenderSyncAgentPlist_Contract(t *testing.T) { for _, want := range []string{ ClaudeMemSyncLaunchdLabel, "ssh -o BatchMode=yes -o ConnectTimeout=10 'user@mac2' true || exit 0", - "exec '/usr/local/bin/dot' ai memory sync --peer 'user@mac2'", + "'/usr/local/bin/dot' ai memory sync --peer 'user@mac2' & pid=$!", + "(sleep 900; kill -TERM $pid; sleep 30; kill -KILL $pid)", + "exit $rc", "DOT_SCHEDULED_RUN", "3600", } { diff --git a/internal/aisettings/testdata/claude-mem-sync.plist.golden b/internal/aisettings/testdata/claude-mem-sync.plist.golden index b1ec3a29..6958d64e 100644 --- a/internal/aisettings/testdata/claude-mem-sync.plist.golden +++ b/internal/aisettings/testdata/claude-mem-sync.plist.golden @@ -9,7 +9,11 @@ /bin/sh -c ssh -o BatchMode=yes -o ConnectTimeout=10 'user@mac2' true || exit 0 -exec '/Users/test/.local/bin/dot' ai memory sync --peer 'user@mac2' +'/Users/test/.local/bin/dot' ai memory sync --peer 'user@mac2' & pid=$! +(sleep 900; kill -TERM $pid; sleep 30; kill -KILL $pid) 2>/dev/null & guard=$! +wait $pid; rc=$? +kill $guard 2>/dev/null +exit $rc EnvironmentVariables diff --git a/internal/cli/ai_memory_sync_cmd.go b/internal/cli/ai_memory_sync_cmd.go index 8bbcf8ef..f51d1807 100644 --- a/internal/cli/ai_memory_sync_cmd.go +++ b/internal/cli/ai_memory_sync_cmd.go @@ -1,6 +1,8 @@ package cli import ( + "context" + "errors" "fmt" "log/slog" "os" @@ -29,6 +31,10 @@ with a peer Mac. Incremental by default (receiver's MAX(created_at) minus a needs only the dot binary: the transport is 'ssh dot ai memory sync --serve ' with JSON over stdio. +A run is bounded by --timeout: past it, or on SIGTERM, the open import +rolls back and the run exits non-zero naming the phase it stopped in. +Each phase logs its start and end to stderr. + sessions replicates all sessions both ways. status prints both sides' counts plus the last scheduled run. With no action, sync runs both directions. --peer defaults to the configured dot peer target.`, @@ -39,12 +45,37 @@ directions. --peer defaults to the configured dot peer target.`, c.Flags().String("peer", "", "ssh target to sync with (default: the dot peer target)") c.Flags().Bool("full", false, "disable the incremental cutoff (re-offer all rows)") c.Flags().String("remote-db", "", "DB path on the peer (default: the peer's ~/.claude-mem/claude-mem.db)") + c.Flags().Duration("timeout", defaultMemorySyncTimeout, "abort the run (and roll back an open import) after this long") c.Flags().String("serve", "", "run one serve op against the local DB (used by the ssh transport)") _ = c.Flags().MarkHidden("serve") return c } +// defaultMemorySyncTimeout bounds one run. A healthy incremental sync takes +// seconds; the bound exists so a stalled run cannot hold the store's write +// lock, and the hourly launchd slot, for hours (#289). +const defaultMemorySyncTimeout = 10 * time.Minute + func runAIMemorySync(cmd *cobra.Command, args []string) error { + timeout, _ := cmd.Flags().GetDuration("timeout") + if timeout <= 0 { + return fmt.Errorf("--timeout must be positive") + } + ctx, cancel := context.WithTimeout(cmd.Context(), timeout) + defer cancel() + cmd.SetContext(ctx) + err := runAIMemorySyncBounded(cmd, args) + switch { + case err == nil: + case errors.Is(ctx.Err(), context.DeadlineExceeded): + err = fmt.Errorf("sync aborted after the %s deadline: %w", timeout, err) + case errors.Is(ctx.Err(), context.Canceled): + err = fmt.Errorf("sync canceled: %w", err) + } + return err +} + +func runAIMemorySyncBounded(cmd *cobra.Command, args []string) error { if op, _ := cmd.Flags().GetString("serve"); op != "" { dbPath := memorySyncDBPath(cmd) return aisettings.ServeOp(cmd.Context(), dbPath, homeFor(cmd), op, cmd.InOrStdin(), cmd.OutOrStdout()) diff --git a/internal/cli/ai_memory_sync_cmd_test.go b/internal/cli/ai_memory_sync_cmd_test.go index d8f93de9..8dd0f5de 100644 --- a/internal/cli/ai_memory_sync_cmd_test.go +++ b/internal/cli/ai_memory_sync_cmd_test.go @@ -2,6 +2,7 @@ package cli import ( "bytes" + "context" "database/sql" "encoding/json" "fmt" @@ -345,7 +346,7 @@ func seedSyncTestSchema(db *aisettings.SyncDB) error { "created_at": "2026-09-20T10:02:00.000Z", "created_at_epoch": 0, }}, } - _, err = db.Import(bundle) + _, err = db.Import(context.Background(), bundle) return err } diff --git a/internal/cli/ai_memory_sync_status.go b/internal/cli/ai_memory_sync_status.go index 51ef6e72..fdcd67e3 100644 --- a/internal/cli/ai_memory_sync_status.go +++ b/internal/cli/ai_memory_sync_status.go @@ -1,6 +1,7 @@ package cli import ( + "context" "fmt" "os" "runtime" @@ -22,12 +23,12 @@ func printMemorySyncDryRun(cmd *cobra.Command, db *aisettings.SyncDB, peer aiset t := &aisettings.SSHTransport{} p.Header("claude-mem sync (dry run)") p.KV("Peer", peer.Target) - preview := func(label string, maxes map[string]string, export func(map[string]string, bool) (*aisettings.Bundle, error)) error { + preview := func(label string, maxes map[string]string, export func(context.Context, map[string]string, bool) (*aisettings.Bundle, error)) error { cutoffs, err := memorySyncCutoffsForPreview(maxes, opts) if err != nil { return err } - bundle, err := export(cutoffs, opts.OnlySessions) + bundle, err := export(cmd.Context(), cutoffs, opts.OnlySessions) if err != nil { return err } @@ -46,12 +47,12 @@ func printMemorySyncDryRun(cmd *cobra.Command, db *aisettings.SyncDB, peer aiset } } if opts.OnlySessions || opts.Direction == "pull" || opts.Direction == "sync" { - maxes, err := db.MaxCreatedAt() + maxes, err := db.MaxCreatedAt(cmd.Context()) if err != nil { return err } - if err := preview("would pull (peer export)", maxes, func(cutoffs map[string]string, onlySessions bool) (*aisettings.Bundle, error) { - return t.Export(cmd.Context(), peer, cutoffs, onlySessions) + if err := preview("would pull (peer export)", maxes, func(ctx context.Context, cutoffs map[string]string, onlySessions bool) (*aisettings.Bundle, error) { + return t.Export(ctx, peer, cutoffs, onlySessions) }); err != nil { return err } @@ -101,7 +102,7 @@ func printMemorySyncStatus(p *Printer, cmd *cobra.Command, home string, peer ais return err } defer db.Close() - local, err := db.Counts() + local, err := db.Counts(cmd.Context()) if err != nil { return err } @@ -165,7 +166,7 @@ func printMemorySyncStatusSection(p *Printer, cmd *cobra.Command, mgr *aisetting return } defer db.Close() - if local, err := db.Counts(); err == nil { + if local, err := db.Counts(cmd.Context()); err == nil { p.Bullet(hint, syncCountsLine("local", local)) } counts, err := (&aisettings.SSHTransport{}).Counts(cmd.Context(), aisettings.SyncPeer{Target: target, RemoteDot: remoteDot})