diff --git a/docs/internal/session-format-sources.md b/docs/internal/session-format-sources.md index 5384f3da9..5977d7cb3 100644 --- a/docs/internal/session-format-sources.md +++ b/docs/internal/session-format-sources.md @@ -256,6 +256,75 @@ Grok section and remove the explicit registry exception in the coverage test. counts as a failure, matching the existing `exit status N` heuristic, and a timed-out command records `timeout: true` with no `exit` key, so it is not detected here. See #1256. +- **Change detection (SQLite layout):** every session in a root shares one + physical `opencode.db`, so the container's own size and mtime move whenever + any single session is written and cannot discriminate between sessions. + Agentsview instead builds a per-session composite from + `session.time_updated`, `project.time_updated`, `MAX(message.time_updated)`, + and `MAX(part.time_updated)` (`openCodeCompositeMtimeExpr`), and omits the + container size from the per-session fingerprint. Verified 2026-07-27 against + an isolated clone of a production container (13.5 GB, 5,981 sessions, 104k + messages, 508k parts): 432,779 of 508,400 parts (86%) carry + `time_updated != time_created`, so in-place child edits do move the signal; + 437 sessions have `MAX(part.time_updated) > session.time_updated`, so the + session row alone is insufficient; and no project's `time_updated` falls + within 5s of its newest session, so folding `project` in tracks genuine + worktree/metadata changes rather than ordinary session activity. The child + scans cost ~0.6s warm on that container because `part.data` lives in SQLite + overflow pages, so scanning `(session_id, time_updated)` does not read + transcript bytes. A MAX over timestamps cannot see a deletion: on that + container 5,758 of 5,981 sessions (96%) carry a session or project timestamp + at or above every child, so removing a message or part leaves the max + untouched. The fingerprint hash therefore carries a per-session digest of the + watermark plus the child row counts, and freshness compares it + (`FingerprintHashRequiredForFreshness`). An earlier revision of this entry + claimed a revert stays detectable because it lowers the max; that is wrong for + the 96% above, and the row counts are what actually cover deletions. Known + gap: a write that leaves the watermark, the message count and the part count + all unchanged is not attributed to any session, which requires an in-place + edit that does not stamp `time_updated`. Containers whose schema lacks the + child `time_updated` columns (older OpenCode, Kilo, MiMoCode, ICodeMate) fall + back to the session-only mtime plus the container size and emit an empty + digest, preserving prior behavior. Watcher events do not pay the child scan + at all: changed-path classification lists sessions through a bounded + session-row watermark (`MAX(session.time_updated, project.time_updated)`, + `ForEachOpenCodeSessionWatermarkMeta`), compares it per session and + like-for-like against the stored session/project metadata watermark + recovered from the persisted child digest + (`OpenCodeChildDigestMetadataWatermarkNS`; rows without a parseable digest + fall back to the stored composite), loaded in one indexed range query + (`filterFreshWatermarkOnlySources`), and drops covered sessions before + they are materialized into discovered files — only the changed batch flows + into the pipeline and resolves the full composite and digest through the + indexed per-session lookup. The comparison must be like-for-like: the + stored composite can be dominated by a newer child timestamp, and + comparing the session-row watermark against it would hide a metadata + update (title, directory, worktree rename) whose stamp lands below that + child maximum. A session or project row that advances past its own stored + metadata watermark is always a candidate, wherever other sessions' + watermarks or its own child timestamps sit. Periodic full passes and + streamed reconciliation passes over a container whose captured state still + matches the last fully verified pass also list the watermark form + (`SQLiteContainerUnchangedSinceTrust`): every member gate-skips before + fingerprinting, so the child identity scan would be archive-sized work + nothing reads; any write breaks that trust and the next pass carries the + complete digest again. Watermark-only skips additionally require the + pass's container capture to still be valid + (`sqliteContainerPassCaptureValid`) — a container that changes between + listing and the recapture check resolves full per-session digests instead, + so a concurrent child-only write cannot hide beneath an unchanged metadata + watermark. The trade is explicit: any + child-only write that leaves the session and project rows untouched — + wherever its timestamps land relative to the stored composite — is + invisible to a watcher pass and is reconciled by the next full-discovery + pass over the now-untrusted container, whose digest still catches it; on + the production container above, 96% of sessions carry a session/project + timestamp at or above every child, and actively watched sessions bypass + this entirely via the per-session composite poll. Per-event work is + bounded by the changed batch plus one O(session-count) scan of small + fixed-width rows (the session table and the stored-member range query); + that floor is irreducible without a watermark index, which OpenCode's + schema does not have and which is not agentsview's to add. - **Agentsview:** `internal/parser/opencode.go`, `internal/parser/opencode_provider.go`, and `internal/parser/opencode_storage_state.go`; legacy and database layouts are diff --git a/internal/db/sessions.go b/internal/db/sessions.go index 535bf180c..6e3dd1266 100644 --- a/internal/db/sessions.go +++ b/internal/db/sessions.go @@ -2174,6 +2174,86 @@ func (db *DB) GetFileInfoByPath( return s.Int64, m.Int64, true } +// VirtualContainerMemberFreshness is one stored virtual member's freshness +// signal: the newest stored file_mtime for its path, the minimum stored +// data version, and the newest row's fingerprint hash, mirroring +// GetFileInfoByPath, GetDataVersionByPath, and GetFileHashByPath. +type VirtualContainerMemberFreshness struct { + MTimeNS int64 + DataVersion int + Hash string +} + +// ListVirtualContainerMemberFreshness returns the freshness signal for every +// stored session whose file_path is a virtual member of the shared container +// at containerPath ("#"), excluding source-missing +// tombstones, keyed by file_path. Changed-path classification compares a +// watermark-only listing against it in one indexed range query, so a +// one-session write flows one candidate into the sync pipeline instead of +// every session in the container. Returning one small row per stored member +// is O(member count) by design and is the point: per-session freshness needs +// per-session stored state, and this single batched index-range read +// replaces the two point queries per session the pipeline would otherwise +// pay. The range predicate rides idx_sessions_file_path; '$' is the ASCII +// successor of '#', so the half-open range covers exactly the +// "#" prefix. +func (db *DB) ListVirtualContainerMemberFreshness( + ctx context.Context, containerPath string, +) (map[string]VirtualContainerMemberFreshness, error) { + if containerPath == "" { + return nil, nil + } + // Folded in Go rather than GROUP BY: the map needs MAX(file_mtime), + // MIN(data_version), and the hash of the newest-mtime row, and SQLite's + // bare-column-from-the-extreme-row guarantee only holds with exactly one + // min/max aggregate in the query. + rows, err := db.getReader().QueryContext(ctx, + "SELECT file_path, file_mtime, data_version, file_hash FROM sessions"+ + " WHERE file_path >= ? || '#' AND file_path < ? || '$'"+ + " AND (deletion_cause IS NULL"+ + " OR deletion_cause <> '"+deletionCauseSourceMissing+"')", + containerPath, containerPath, + ) + if err != nil { + return nil, fmt.Errorf( + "listing container member freshness %s: %w", containerPath, err, + ) + } + defer rows.Close() + + members := make(map[string]VirtualContainerMemberFreshness) + for rows.Next() { + var path string + var mtime, version sql.NullInt64 + var hash sql.NullString + if err := rows.Scan(&path, &mtime, &version, &hash); err != nil { + return nil, fmt.Errorf( + "scanning container member freshness %s: %w", + containerPath, err, + ) + } + row := VirtualContainerMemberFreshness{ + MTimeNS: mtime.Int64, + DataVersion: int(version.Int64), + Hash: hash.String, + } + member, seen := members[path] + if !seen { + members[path] = row + continue + } + if row.MTimeNS > member.MTimeNS { + member.MTimeNS = row.MTimeNS + member.Hash = row.Hash + } + if row.DataVersion < member.DataVersion { + member.DataVersion = row.DataVersion + } + members[path] = member + } + return members, rows.Err() +} + // GetProjectByPath returns the stored project for the newest // non-deleted session matching file_path. func (db *DB) GetProjectByPath(path string) (project string, ok bool) { diff --git a/internal/parser/capabilities_sync_test.go b/internal/parser/capabilities_sync_test.go index 2fd6e17a8..1902e39c5 100644 --- a/internal/parser/capabilities_sync_test.go +++ b/internal/parser/capabilities_sync_test.go @@ -55,17 +55,26 @@ func TestProviderSyncSemanticsDeclarations(t *testing.T) { AgentShelley: { UnchangedResults: UnchangedResultMTimeAndHash, }, + // The OpenCode family shares one physical container per root, so + // freshness consults the per-session child digest: it is the only + // signal that sees a deleted message or part, which a MAX over + // timestamps cannot. Containers without composite support emit an + // empty hash, which the gate treats as no constraint. AgentOpenCode: { - UnchangedResults: UnchangedResultMTimeAndHash, + UnchangedResults: UnchangedResultMTimeAndHash, + FingerprintHashRequiredForFreshness: true, }, AgentKilo: { - UnchangedResults: UnchangedResultMTimeAndHash, + UnchangedResults: UnchangedResultMTimeAndHash, + FingerprintHashRequiredForFreshness: true, }, AgentMiMoCode: { - UnchangedResults: UnchangedResultMTimeAndHash, + UnchangedResults: UnchangedResultMTimeAndHash, + FingerprintHashRequiredForFreshness: true, }, AgentIcodemate: { - UnchangedResults: UnchangedResultMTimeAndHash, + UnchangedResults: UnchangedResultMTimeAndHash, + FingerprintHashRequiredForFreshness: true, }, AgentOmnigent: { FingerprintHashInCacheKey: true, diff --git a/internal/parser/opencode.go b/internal/parser/opencode.go index 4a416e467..192a2e24f 100644 --- a/internal/parser/opencode.go +++ b/internal/parser/opencode.go @@ -5,13 +5,16 @@ import ( "crypto/sha256" "database/sql" "encoding/json" + "errors" "fmt" "os" "path/filepath" "regexp" "sort" + "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/tidwall/gjson" @@ -25,6 +28,24 @@ type OpenCodeSessionMeta struct { SessionID string VirtualPath string FileMtime int64 + // CompositeMtime reports that FileMtime is the per-session composite + // (see openCodeCompositeMtimeExpr) rather than the session row's own + // time_updated. When true the fingerprint omits the shared container's + // size, because the composite already discriminates per session. + CompositeMtime bool + // ChildDigest is the per-session freshness identity carried into the + // fingerprint's Hash. It folds the composite watermark together with the + // child row counts, so a deletion that leaves the watermark untouched + // still changes it. Empty when the container has no composite support. + ChildDigest string + // WatermarkOnly reports that FileMtime is only the session-row watermark + // (session and project time_updated, no child tables) and ChildDigest is + // deliberately unresolved. Changed-path listings use this bounded form so + // a watcher event does not scan every child row in the container; the + // engine skips such a source when its stored composite watermark already + // covers the carried value, and resolves the full composite and digest + // through the indexed per-session lookup otherwise. + WatermarkOnly bool } // OpenCodeSQLiteSessionExists reports whether a session row with @@ -73,6 +94,124 @@ func ListOpenCodeSessionMeta( return metas, err } +// The two container freshness query shapes, counted so tests can pin that +// watcher-driven passes stay bounded by the changed batch: the grouped +// whole-container child scan must not run on a changed-path pass, and +// per-session child lookups must scale with the number of changed sessions, +// not the archive. +var ( + openCodeContainerChildScans atomic.Int64 + openCodeSessionChildLookups atomic.Int64 +) + +// OpenCodeContainerChildScans returns how many whole-container child-table +// identity scans (grouped message/part aggregation) have run. +func OpenCodeContainerChildScans() int64 { + return openCodeContainerChildScans.Load() +} + +// OpenCodeSessionChildLookups returns how many single-session indexed child +// digest lookups have run. +func OpenCodeSessionChildLookups() int64 { + return openCodeSessionChildLookups.Load() +} + +// ListOpenCodeSessionWatermarkMeta is the materialized form of +// ForEachOpenCodeSessionWatermarkMeta. +func ListOpenCodeSessionWatermarkMeta( + dbPath string, +) ([]OpenCodeSessionMeta, error) { + var metas []OpenCodeSessionMeta + err := ForEachOpenCodeSessionWatermarkMeta( + context.Background(), dbPath, + func(meta OpenCodeSessionMeta) error { + metas = append(metas, meta) + return nil + }, + ) + return metas, err +} + +// ForEachOpenCodeSessionWatermarkMeta streams session rows carrying only the +// session-row watermark: MAX(session.time_updated, project.time_updated), +// touching no child tables. A watcher event on a shared container must not +// pay a whole-archive child scan to find the one session that changed, so +// changed-path classification lists sessions through this bounded form and +// leaves the full composite and child digest to the indexed per-session +// lookup, which the engine runs only for sessions it cannot skip against +// their stored watermark. What this signal cannot see — a child-only write +// that leaves the session and project rows untouched, wherever its +// timestamps land relative to the stored composite — is a known, deliberate +// deferral reconciled by the next full-discovery pass, which still carries +// the complete child digest (see the change-detection entry in +// docs/internal/session-format-sources.md). Legacy containers without +// composite support keep the full listing; their conservative +// container-size fingerprint must not be bypassed by a watermark-only skip. +// +// This listing scans the session table once and that scan is O(session +// count) by design: OpenCode's schema indexes neither time_updated column, +// and the schema is not ours to alter, so any sound candidate selection +// must read every session row. The rows are few and fixed-width (one per +// session, no transcript bytes), which is what makes this the bounded form +// — the quantities that previously scaled with the archive were the child +// tables (two orders of magnitude more rows) and the per-event +// materialization downstream, and both are now bounded by the changed +// batch. +func ForEachOpenCodeSessionWatermarkMeta( + ctx context.Context, + dbPath string, + yield func(OpenCodeSessionMeta) error, +) error { + if _, err := os.Stat(dbPath); os.IsNotExist(err) { + return nil + } + + db, err := openOpenCodeDB(dbPath) + if err != nil { + return err + } + defer db.Close() + + composite, err := openCodeCompositeMtimeSupportedCached(db, dbPath) + if err != nil { + return err + } + query := "SELECT s.id, s.time_updated FROM session s" + if composite { + query = "SELECT s.id, " + openCodeSessionRowWatermarkExpr + + " FROM session s" + openCodeSessionCompositeMtimeJoins + } + + rows, err := db.QueryContext(ctx, query) + if err != nil { + return fmt.Errorf( + "listing opencode session watermarks: %w", err, + ) + } + defer rows.Close() + + for rows.Next() { + var id string + var watermark int64 + if err := rows.Scan(&id, &watermark); err != nil { + return fmt.Errorf( + "scanning opencode session watermark: %w", err, + ) + } + observeStreamingDiscoveryBuffer(ctx, 1) + if err := yield(OpenCodeSessionMeta{ + SessionID: id, + VirtualPath: dbPath + "#" + id, + FileMtime: watermark * 1_000_000, + CompositeMtime: composite, + WatermarkOnly: composite, + }); err != nil { + return err + } + } + return rows.Err() +} + // ForEachOpenCodeSessionMeta streams lightweight session rows directly from // SQLite. The callback runs while the read-only query is open and receives one // row at a time; callers must not retain database-owned values. @@ -91,9 +230,20 @@ func ForEachOpenCodeSessionMeta( } defer db.Close() - rows, err := db.QueryContext(ctx, - "SELECT id, time_updated FROM session", - ) + composite, err := openCodeCompositeMtimeSupportedCached(db, dbPath) + if err != nil { + return err + } + query := "SELECT s.id, s.time_updated, s.time_updated, 0, 0, 0, '', '' " + + "FROM session s" + if composite { + openCodeContainerChildScans.Add(1) + query = "SELECT s.id, " + openCodeCompositeMtimeExpr + ", " + + openCodeCompositeCountsExpr + + " FROM session s" + openCodeCompositeMtimeJoins + } + + rows, err := db.QueryContext(ctx, query) if err != nil { return fmt.Errorf( "listing opencode sessions: %w", err, @@ -103,9 +253,11 @@ func ForEachOpenCodeSessionMeta( for rows.Next() { var id string - var timeUpdated int64 + var agg openCodeChildAggregate if err := rows.Scan( - &id, &timeUpdated, + &id, &agg.watermark, &agg.sessionTime, &agg.projectTime, + &agg.messages, &agg.parts, + &agg.messageIdent, &agg.partIdent, ); err != nil { return fmt.Errorf( "scanning opencode session meta: %w", err, @@ -113,9 +265,11 @@ func ForEachOpenCodeSessionMeta( } observeStreamingDiscoveryBuffer(ctx, 1) if err := yield(OpenCodeSessionMeta{ - SessionID: id, - VirtualPath: dbPath + "#" + id, - FileMtime: timeUpdated * 1_000_000, + SessionID: id, + VirtualPath: dbPath + "#" + id, + FileMtime: agg.watermark * 1_000_000, + CompositeMtime: composite, + ChildDigest: agg.digest(composite), }); err != nil { return err } @@ -123,6 +277,153 @@ func ForEachOpenCodeSessionMeta( return rows.Err() } +// openCodeSessionCompositeMtime returns one session's composite change signal +// in milliseconds, and whether the container schema supports it. Discovery, +// single-session source lookup, and the parse path all resolve mtime through +// this so a session's stored file_mtime always equals the value the freshness +// gate compares it against. +func openCodeSessionCompositeMtime( + db *sql.DB, dbPath, sessionID string, +) (int64, string, bool, error) { + composite, err := openCodeCompositeMtimeSupportedCached(db, dbPath) + if err != nil { + return 0, "", false, err + } + query := "SELECT s.time_updated, s.time_updated, 0, 0, 0, '', '' " + + "FROM session s WHERE s.id = ?" + if composite { + openCodeSessionChildLookups.Add(1) + query = "SELECT " + openCodeSessionCompositeMtimeExpr + ", " + + openCodeSessionCompositeCountsExpr + + " FROM session s" + openCodeSessionCompositeMtimeJoins + + " WHERE s.id = ?" + } + var agg openCodeChildAggregate + if err := db.QueryRow(query, sessionID).Scan( + &agg.watermark, &agg.sessionTime, &agg.projectTime, + &agg.messages, &agg.parts, + &agg.messageIdent, &agg.partIdent, + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return 0, "", composite, nil + } + return 0, "", composite, fmt.Errorf( + "loading opencode session mtime %s#%s: %w", + dbPath, sessionID, err, + ) + } + return agg.watermark, agg.digest(composite), composite, nil +} + +// openCodeSessionWatermark resolves only the composite watermark, skipping the +// eight child COUNT/SUM/MIN/MAX subqueries that make up the digest. Callers +// that stamp or compare an mtime do not need the digest, and one of them +// (OpenCodeSourceMtime) backs the session watcher's 1.5s poll, so computing a +// digest there would burn eight child-range scans per tick per watched session +// for a value the caller discards. +func openCodeSessionWatermark( + db *sql.DB, dbPath, sessionID string, +) (int64, bool, error) { + composite, err := openCodeCompositeMtimeSupportedCached(db, dbPath) + if err != nil { + return 0, false, err + } + query := "SELECT s.time_updated FROM session s WHERE s.id = ?" + if composite { + query = "SELECT " + openCodeSessionCompositeMtimeExpr + + " FROM session s" + openCodeSessionCompositeMtimeJoins + + " WHERE s.id = ?" + } + var watermark int64 + if err := db.QueryRow(query, sessionID).Scan(&watermark); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return 0, composite, nil + } + return 0, composite, fmt.Errorf( + "loading opencode session mtime %s#%s: %w", + dbPath, sessionID, err, + ) + } + return watermark, composite, nil +} + +// openCodeChildAggregate is the cheap per-session identity read alongside the +// watermark. Each component covers a change the others cannot: +// +// - watermark: an edit or insert that advances a timestamp +// - session/project times: a metadata update, including a worktree rename, +// that lands below an already-higher child watermark +// - counts: a deletion, which never moves a MAX +// - child identity: every ordered (id, time_updated) pair, hashed. Reduced +// aggregates are not collision-resistant — swapping a non-boundary row for +// one carrying the same timestamp preserves counts, sums and min/max ids +// alike, so only the complete set separates them. +// +// All of it lives in the child tables' main b-tree pages, so computing it does +// not read the transcript text held in overflow pages. +type openCodeChildAggregate struct { + watermark int64 + sessionTime int64 + projectTime int64 + messages int64 + parts int64 + messageIdent string + partIdent string +} + +// The field layout is load-bearing beyond equality comparison: +// OpenCodeChildDigestMetadataWatermarkNS recovers the session/project times +// from a stored digest by position. Any layout change must bump the prefix +// version so stale digests fail that parse (and the equality gate) instead +// of yielding wrong fields. +func (a openCodeChildAggregate) digest(composite bool) string { + if !composite { + return "" + } + sum := sha256.Sum256([]byte(a.messageIdent + "\x00" + a.partIdent)) + return fmt.Sprintf( + "%s%d:%d:%d:%d:%d:%x", + openCodeChildDigestPrefix, + a.watermark, a.sessionTime, a.projectTime, + a.messages, a.parts, + sum[:16], + ) +} + +const openCodeChildDigestPrefix = "opencode-child:v1:" + +// OpenCodeChildDigestMetadataWatermarkNS recovers the session/project +// metadata watermark (nanoseconds) embedded in a stored child digest. The +// watermark filter compares the live session-row watermark like-for-like +// against this value: the stored composite MTimeNS may be dominated by a +// newer child timestamp, and comparing the session-row watermark against +// that composite would hide a metadata update (title, directory, worktree) +// whose stamp lands below it. Returns false for any other hash shape — +// legacy fingerprints, storage fingerprints, future digest versions — which +// callers treat as "compare against the composite instead", the +// conservative pre-digest behavior. +func OpenCodeChildDigestMetadataWatermarkNS(hash string) (int64, bool) { + rest, ok := strings.CutPrefix(hash, openCodeChildDigestPrefix) + if !ok { + return 0, false + } + fields := strings.Split(rest, ":") + if len(fields) != 6 { + return 0, false + } + // Validate every numeric field, not just the two consumed: a digest with + // any malformed component is not a digest this version wrote, and the + // caller's composite fallback is the safe answer for it. + for _, field := range fields[:5] { + if _, err := strconv.ParseInt(field, 10, 64); err != nil { + return 0, false + } + } + sessionTime, _ := strconv.ParseInt(fields[1], 10, 64) + projectTime, _ := strconv.ParseInt(fields[2], 10, 64) + return max(sessionTime, projectTime) * 1_000_000, true +} + // parseOpenCodeDBSession parses a single session by ID from the // OpenCode SQLite database. The OpenCode-format provider owns this // path; Kilo and MiMoCode reuse it and relabel the result. @@ -378,9 +679,15 @@ type openCodeSessionRow struct { timeUpdated int64 } +// openCodeSessionSchemaCacheEntry memoizes both schema probes for one +// container. Each probe has its own "resolved" flag so populating one never +// makes the other report a false negative from its zero value. type openCodeSessionSchemaCacheEntry struct { - state SQLiteContainerState - hasDirectory bool + state SQLiteContainerState + hasDirectory bool + directoryOnce bool + hasComposite bool + compositeOnce bool } // openCodeSessionSchemaCache memoizes whether session.directory exists for @@ -402,7 +709,7 @@ func openCodeSessionHasDirectoryCached( openCodeSessionSchemaCacheMu.Lock() entry, hit := openCodeSessionSchemaCache[dbPath] openCodeSessionSchemaCacheMu.Unlock() - if hit && entry.state == state { + if hit && entry.state == state && entry.directoryOnce { return entry.hasDirectory, nil } hasDirectory, err := openCodeSessionTableHasDirectory(db) @@ -410,14 +717,208 @@ func openCodeSessionHasDirectoryCached( return false, err } openCodeSessionSchemaCacheMu.Lock() - openCodeSessionSchemaCache[dbPath] = openCodeSessionSchemaCacheEntry{ - state: state, - hasDirectory: hasDirectory, - } + prev := openCodeSessionSchemaCache[dbPath] + if prev.state != state { + prev = openCodeSessionSchemaCacheEntry{} + } + prev.state = state + prev.hasDirectory = hasDirectory + prev.directoryOnce = true + openCodeSessionSchemaCache[dbPath] = prev openCodeSessionSchemaCacheMu.Unlock() return hasDirectory, nil } +// openCodeCompositeMtimeExpr is the per-session change signal for a +// SQLite-backed OpenCode container. Every session in a root shares one +// physical opencode.db, so the container file's own size and mtime move +// whenever any single session is written and cannot discriminate between +// sessions. These four columns can: +// +// - session.time_updated — the session row itself +// - project.time_updated — the owning project (worktree renames re-resolve +// every session in that project, which is the correct scope; verified on a +// production container that this does not track ordinary session activity) +// - max(message.time_updated) / max(part.time_updated) — child content, +// including in-place edits that leave time_created untouched +// +// The child scans read only small columns; OpenCode keeps each part's `data` +// in SQLite overflow pages, so this does not read transcript bytes. +// The streaming form groups the child tables once for the whole container, so +// listing every session costs a single pass over each child table. +const openCodeCompositeMtimeExpr = `MAX(s.time_updated, + COALESCE(pr.time_updated, 0), + COALESCE(m.mx, 0), + COALESCE(p.mx, 0))` + +const openCodeCompositeMtimeJoins = ` + LEFT JOIN project pr ON pr.id = s.project_id + LEFT JOIN ( + SELECT session_id, MAX(time_updated) mx, COUNT(*) n, + group_concat(id || ':' || time_updated) ident + FROM (SELECT session_id, id, time_updated FROM message + ORDER BY session_id, id) + GROUP BY session_id + ) m ON m.session_id = s.id + LEFT JOIN ( + SELECT session_id, MAX(time_updated) mx, COUNT(*) n, + group_concat(id || ':' || time_updated) ident + FROM (SELECT session_id, id, time_updated FROM part + ORDER BY session_id, id) + GROUP BY session_id + ) p ON p.session_id = s.id` + +// openCodeCompositeCountsExpr yields the child row counts that make the +// signal deletion-sensitive. A MAX over timestamps cannot see a delete: on a +// real container the session or project row usually already holds the higher +// value, so removing a message or part leaves the max untouched and the +// session would look fresh with the deleted content still archived. +const openCodeCompositeCountsExpr = `s.time_updated, + COALESCE(pr.time_updated, 0), + COALESCE(m.n, 0), COALESCE(p.n, 0), + COALESCE(m.ident, ''), COALESCE(p.ident, '')` + +const openCodeSessionCompositeCountsExpr = ` + s.time_updated, + COALESCE(pr.time_updated, 0), + (SELECT COUNT(*) FROM message WHERE session_id = s.id), + (SELECT COUNT(*) FROM part WHERE session_id = s.id), + (SELECT COALESCE(group_concat(id || ':' || time_updated), '') + FROM (SELECT id, time_updated FROM message + WHERE session_id = s.id ORDER BY id)), + (SELECT COALESCE(group_concat(id || ':' || time_updated), '') + FROM (SELECT id, time_updated FROM part + WHERE session_id = s.id ORDER BY id))` + +// The single-session form must NOT reuse the grouped subqueries above: a +// GROUP BY subquery is materialized over the whole container before the outer +// WHERE narrows to one session, so every per-session lookup would scan every +// message and part in the container. Correlated aggregates filtered by +// session_id ride the message/part session_id indexes instead, which is the +// difference between an index seek and an archive-wide scan on every call. +const openCodeSessionCompositeMtimeExpr = `MAX(s.time_updated, + COALESCE(pr.time_updated, 0), + COALESCE(( + SELECT MAX(time_updated) FROM message WHERE session_id = s.id + ), 0), + COALESCE(( + SELECT MAX(time_updated) FROM part WHERE session_id = s.id + ), 0))` + +const openCodeSessionCompositeMtimeJoins = ` + LEFT JOIN project pr ON pr.id = s.project_id` + +// openCodeSessionRowWatermarkExpr is the bounded change signal used by +// watermark-only changed-path listings: the session and project rows alone, +// no child aggregation. The session table holds one small row per session, so +// listing every session through this costs a scan of the session and project +// tables only — bounded by session count, never by message/part volume. +const openCodeSessionRowWatermarkExpr = `MAX(s.time_updated, + COALESCE(pr.time_updated, 0))` + +// openCodeCompositeMtimeSupportedCached reports whether this container's schema +// carries every column openCodeCompositeMtimeExpr needs. Older OpenCode-family +// containers (Kilo, MiMoCode, ICodeMate, legacy OpenCode) omit the child +// time_updated columns; those keep the previous session-only mtime and the +// container-stat fallback in Fingerprint. +func openCodeCompositeMtimeSupportedCached( + db *sql.DB, dbPath string, +) (bool, error) { + state, ok := StatSQLiteContainerState(dbPath) + if !ok { + return openCodeSupportsCompositeMtime(db) + } + openCodeSessionSchemaCacheMu.Lock() + entry, hit := openCodeSessionSchemaCache[dbPath] + openCodeSessionSchemaCacheMu.Unlock() + if hit && entry.state == state && entry.compositeOnce { + return entry.hasComposite, nil + } + supported, err := openCodeSupportsCompositeMtime(db) + if err != nil { + return false, err + } + openCodeSessionSchemaCacheMu.Lock() + prev := openCodeSessionSchemaCache[dbPath] + if prev.state != state { + prev = openCodeSessionSchemaCacheEntry{state: state} + } + prev.state = state + prev.hasComposite = supported + prev.compositeOnce = true + openCodeSessionSchemaCache[dbPath] = prev + openCodeSessionSchemaCacheMu.Unlock() + return supported, nil +} + +func openCodeSupportsCompositeMtime(db *sql.DB) (bool, error) { + for _, probe := range []struct{ table, column string }{ + {"message", "time_updated"}, + {"part", "time_updated"}, + {"project", "time_updated"}, + } { + has, err := openCodeTableHasColumn(db, probe.table, probe.column) + if err != nil || !has { + return false, err + } + } + // The per-session lookups are correlated aggregates keyed on session_id. + // SQLite does not index a foreign key automatically, so without a + // session_id index each one degrades to a full child-table scan — and one + // of these backs the session watcher's 1.5s poll. Fall back to the + // session-only mtime rather than put an archive scan on that path. + for _, table := range []string{"message", "part"} { + indexed, err := openCodeTableIndexesColumn(db, table, "session_id") + if err != nil || !indexed { + return false, err + } + } + return true, nil +} + +// openCodeTableIndexesColumn reports whether table has an index whose leftmost +// column is column, which is what makes a WHERE column = ? lookup a seek. +func openCodeTableIndexesColumn( + db *sql.DB, table, column string, +) (bool, error) { + rows, err := db.Query( + `SELECT 1 FROM pragma_index_list(?) il + JOIN pragma_index_info(il.name) ii + WHERE ii.seqno = 0 AND ii.name = ?`, + table, column, + ) + if err != nil { + return false, fmt.Errorf( + "listing opencode %s indexes: %w", table, err, + ) + } + defer rows.Close() + if rows.Next() { + return true, rows.Err() + } + return false, rows.Err() +} + +// openCodeTableHasColumn reports whether table carries column. An unknown +// table yields no PRAGMA rows and reports false rather than erroring, so a +// container missing an optional table degrades to the legacy signal. +func openCodeTableHasColumn( + db *sql.DB, table, column string, +) (bool, error) { + rows, err := db.Query(`SELECT 1 FROM pragma_table_info(?) WHERE name = ?`, + table, column) + if err != nil { + return false, fmt.Errorf( + "listing opencode %s table info: %w", table, err, + ) + } + defer rows.Close() + if rows.Next() { + return true, rows.Err() + } + return false, rows.Err() +} + func openCodeSessionTableHasDirectory(db *sql.DB) (bool, error) { rows, err := db.Query(`PRAGMA table_info(session)`) if err != nil { @@ -622,6 +1123,27 @@ func buildOpenCodeSession( s openCodeSessionRow, cwd, projectWorktree, dbPath, machine string, ) (*ParsedSession, []ParsedMessage, error) { + // Capture the watermark BEFORE reading children. Messages and parts are + // read through separate autocommit queries, so a concurrent write landing + // between them would otherwise be stamped with a watermark newer than the + // content actually read, and every later sync would skip the session as + // fresh — permanently archiving a torn transcript. Reading the watermark + // first inverts the race: the stamp is never newer than the content, so a + // concurrent change leaves the stored value behind the source and the next + // pass re-syncs it. + // + // Stamp the same composite the fingerprint reports, so the stored + // file_mtime is directly comparable to it. Falling back to the session + // row's own time_updated keeps legacy containers on their prior value. + fileMtime := s.timeUpdated + if composite, _, err := openCodeSessionWatermark( + db, dbPath, s.id, + ); err != nil { + return nil, nil, err + } else if composite != 0 { + fileMtime = composite + } + msgs, err := loadOpenCodeMessages(db, s.id) if err != nil { return nil, nil, fmt.Errorf( @@ -641,7 +1163,7 @@ func buildOpenCodeSession( cwd, projectWorktree, dbPath+"#"+s.id, - s.timeUpdated*1_000_000, + fileMtime*1_000_000, machine, msgs, parts, @@ -1403,38 +1925,64 @@ func openCodeStorageFingerprintHash(raw string) string { return fmt.Sprintf("%x", sum) } -func openCodeSQLiteSessionMtime( +// openCodeSQLiteSessionMtimeComposite is openCodeSQLiteSessionMtime with the +// schema-support flag the fingerprint needs to decide whether the shared +// container's size still has to act as a fallback change signal. +func openCodeSQLiteSessionMtimeComposite( dbPath, sessionID string, -) (int64, error) { +) (int64, string, bool, error) { if _, err := os.Stat(dbPath); err != nil { if os.IsNotExist(err) { - return 0, nil + return 0, "", false, nil } - return 0, fmt.Errorf( + return 0, "", false, fmt.Errorf( "stat opencode db %s: %w", dbPath, err, ) } db, err := openOpenCodeDB(dbPath) if err != nil { - return 0, err + return 0, "", false, err } defer db.Close() - row := db.QueryRow( - "SELECT time_updated FROM session WHERE id = ?", - sessionID, + timeUpdated, digest, composite, err := openCodeSessionCompositeMtime( + db, dbPath, sessionID, ) - var timeUpdated int64 - if err := row.Scan(&timeUpdated); err != nil { - if err == sql.ErrNoRows { + if err != nil { + return 0, "", false, err + } + if timeUpdated == 0 { + return 0, digest, composite, nil + } + return timeUpdated * 1_000_000, digest, composite, nil +} + +func openCodeSQLiteSessionMtime( + dbPath, sessionID string, +) (int64, error) { + if _, err := os.Stat(dbPath); err != nil { + if os.IsNotExist(err) { return 0, nil } return 0, fmt.Errorf( - "loading opencode session mtime %s#%s: %w", - dbPath, sessionID, err, + "stat opencode db %s: %w", dbPath, err, ) } + + db, err := openOpenCodeDB(dbPath) + if err != nil { + return 0, err + } + defer db.Close() + + timeUpdated, _, err := openCodeSessionWatermark(db, dbPath, sessionID) + if err != nil { + return 0, err + } + if timeUpdated == 0 { + return 0, nil + } return timeUpdated * 1_000_000, nil } diff --git a/internal/parser/opencode_provider.go b/internal/parser/opencode_provider.go index 7b0fdbd1e..56cc83891 100644 --- a/internal/parser/opencode_provider.go +++ b/internal/parser/opencode_provider.go @@ -70,7 +70,9 @@ func (f openCodeFormatProviderFactory) NewProvider(cfg ProviderConfig) Provider Caps: openCodeFormatProviderCapabilities(), Config: cfg, }, - sources: newOpenCodeFormatSourceSet(cfg.Roots, f.spec), + sources: newOpenCodeFormatSourceSet( + cfg.Roots, f.spec, cfg.SQLiteContainerUnchangedSinceTrust, + ), } } @@ -175,55 +177,71 @@ func (p *openCodeFormatProvider) Parse( // the OpenCode storage and SQLite readers, then relabel the result onto // their own agent and ID prefix. type openCodeProviderSpec struct { - agent AgentType - format openCodeFormat - dbName string - listSQLite func(string) ([]OpenCodeSessionMeta, error) - streamSQLite func(context.Context, string, func(OpenCodeSessionMeta) error) error - sourceMtime func(string) (int64, error) - relabel func(*ParsedSession) + agent AgentType + format openCodeFormat + dbName string + listSQLite func(string) ([]OpenCodeSessionMeta, error) + // listSQLiteWatermark is the bounded changed-path form of listSQLite: it + // carries only the session-row watermark and no child digest, so a + // watcher event on the shared container never scans the child tables. + listSQLiteWatermark func(string) ([]OpenCodeSessionMeta, error) + streamSQLite func(context.Context, string, func(OpenCodeSessionMeta) error) error + // streamSQLiteWatermark is the bounded trusted-container form of + // streamSQLite, used by streamed reconciliation discovery for containers + // the engine's container gate will skip wholesale. + streamSQLiteWatermark func(context.Context, string, func(OpenCodeSessionMeta) error) error + sourceMtime func(string) (int64, error) + relabel func(*ParsedSession) } func openCodeProviderSpecForAgent(agent AgentType) openCodeProviderSpec { switch agent { case AgentOpenCode: return openCodeProviderSpec{ - agent: AgentOpenCode, - format: openCodeFmt, - dbName: openCodeFmt.dbName, - listSQLite: ListOpenCodeSessionMeta, - streamSQLite: ForEachOpenCodeSessionMeta, - sourceMtime: OpenCodeSourceMtime, + agent: AgentOpenCode, + format: openCodeFmt, + dbName: openCodeFmt.dbName, + listSQLite: ListOpenCodeSessionMeta, + listSQLiteWatermark: ListOpenCodeSessionWatermarkMeta, + streamSQLite: ForEachOpenCodeSessionMeta, + streamSQLiteWatermark: ForEachOpenCodeSessionWatermarkMeta, + sourceMtime: OpenCodeSourceMtime, } case AgentKilo: return openCodeProviderSpec{ - agent: AgentKilo, - format: kiloFmt, - dbName: kiloFmt.dbName, - listSQLite: ListKiloSessionMeta, - streamSQLite: streamOpenCodeSessionMetaAs(KiloSQLiteVirtualPath), - sourceMtime: KiloSourceMtime, - relabel: relabelOpenCodeSessionAsKilo, + agent: AgentKilo, + format: kiloFmt, + dbName: kiloFmt.dbName, + listSQLite: ListKiloSessionMeta, + listSQLiteWatermark: listOpenCodeSessionWatermarkMetaAs(KiloSQLiteVirtualPath), + streamSQLite: streamOpenCodeSessionMetaAs(KiloSQLiteVirtualPath), + streamSQLiteWatermark: streamOpenCodeSessionWatermarkMetaAs(KiloSQLiteVirtualPath), + sourceMtime: KiloSourceMtime, + relabel: relabelOpenCodeSessionAsKilo, } case AgentMiMoCode: return openCodeProviderSpec{ - agent: AgentMiMoCode, - format: mimoFmt, - dbName: mimoFmt.dbName, - listSQLite: ListMiMoCodeSessionMeta, - streamSQLite: streamOpenCodeSessionMetaAs(MiMoCodeSQLiteVirtualPath), - sourceMtime: MiMoCodeSourceMtime, - relabel: relabelOpenCodeSessionAsMiMoCode, + agent: AgentMiMoCode, + format: mimoFmt, + dbName: mimoFmt.dbName, + listSQLite: ListMiMoCodeSessionMeta, + listSQLiteWatermark: listOpenCodeSessionWatermarkMetaAs(MiMoCodeSQLiteVirtualPath), + streamSQLite: streamOpenCodeSessionMetaAs(MiMoCodeSQLiteVirtualPath), + streamSQLiteWatermark: streamOpenCodeSessionWatermarkMetaAs(MiMoCodeSQLiteVirtualPath), + sourceMtime: MiMoCodeSourceMtime, + relabel: relabelOpenCodeSessionAsMiMoCode, } case AgentIcodemate: return openCodeProviderSpec{ - agent: AgentIcodemate, - format: icodemateFmt, - dbName: icodemateFmt.dbName, - listSQLite: ListIcodemateSessionMeta, - streamSQLite: streamOpenCodeSessionMetaAs(IcodemateSQLiteVirtualPath), - sourceMtime: IcodemateSourceMtime, - relabel: relabelOpenCodeSessionAsIcodemate, + agent: AgentIcodemate, + format: icodemateFmt, + dbName: icodemateFmt.dbName, + listSQLite: ListIcodemateSessionMeta, + listSQLiteWatermark: listOpenCodeSessionWatermarkMetaAs(IcodemateSQLiteVirtualPath), + streamSQLite: streamOpenCodeSessionMetaAs(IcodemateSQLiteVirtualPath), + streamSQLiteWatermark: streamOpenCodeSessionWatermarkMetaAs(IcodemateSQLiteVirtualPath), + sourceMtime: IcodemateSourceMtime, + relabel: relabelOpenCodeSessionAsIcodemate, } default: return openCodeProviderSpec{} @@ -243,6 +261,37 @@ func streamOpenCodeSessionMetaAs( } } +func streamOpenCodeSessionWatermarkMetaAs( + virtualPath func(string, string) string, +) func(context.Context, string, func(OpenCodeSessionMeta) error) error { + return func( + ctx context.Context, dbPath string, yield func(OpenCodeSessionMeta) error, + ) error { + return ForEachOpenCodeSessionWatermarkMeta( + ctx, dbPath, + func(meta OpenCodeSessionMeta) error { + meta.VirtualPath = virtualPath(dbPath, meta.SessionID) + return yield(meta) + }, + ) + } +} + +func listOpenCodeSessionWatermarkMetaAs( + virtualPath func(string, string) string, +) func(string) ([]OpenCodeSessionMeta, error) { + return func(dbPath string) ([]OpenCodeSessionMeta, error) { + metas, err := ListOpenCodeSessionWatermarkMeta(dbPath) + if err != nil { + return nil, err + } + for i := range metas { + metas[i].VirtualPath = virtualPath(dbPath, metas[i].SessionID) + } + return metas, nil + } +} + // resolve detects the OpenCode storage backend for a root. func (spec openCodeProviderSpec) resolve(root string) OpenCodeSource { return resolveOpenCodeFormatSource(spec.format, root) @@ -312,25 +361,47 @@ func (spec openCodeProviderSpec) parseSQLite( type openCodeFormatSource struct { Root string Path string - // MTimeNS carries the session's time_updated (already listed during + // MTimeNS carries the session's change signal (already listed during // SQLite discovery, scaled to nanoseconds) so Fingerprint does not // reopen the shared DB once per session. Zero means unknown and makes // Fingerprint fall back to querying the DB. MTimeNS int64 + // CompositeMTime reports that MTimeNS is the per-session composite + // (session, project, and child message/part time_updated) rather than + // the session row's own time_updated. It gates dropping the shared + // container's size from the fingerprint. + CompositeMTime bool + // ChildDigest carries the deletion-sensitive per-session identity into + // Fingerprint.Hash. + ChildDigest string + // WatermarkOnly marks MTimeNS as only the session-row watermark from a + // bounded changed-path listing (see OpenCodeSessionMeta.WatermarkOnly). + // The engine may skip such a source against its stored composite + // watermark without resolving the child digest. + WatermarkOnly bool } type openCodeFormatSourceSet struct { roots []string spec openCodeProviderSpec + // containerTrusted, when non-nil, reports that a shared container is + // byte-identical to the last fully verified pass (see + // ProviderConfig.SQLiteContainerUnchangedSinceTrust). Discover answers + // with the bounded watermark-only listing for such containers: the + // engine's container gate skips every member before fingerprinting, so + // the full child digest would be archive-sized work nothing reads. + containerTrusted func(dbPath string) bool } func newOpenCodeFormatSourceSet( roots []string, spec openCodeProviderSpec, + containerTrusted func(dbPath string) bool, ) openCodeFormatSourceSet { return openCodeFormatSourceSet{ - roots: cleanJSONLRoots(roots), - spec: spec, + roots: cleanJSONLRoots(roots), + spec: spec, + containerTrusted: containerTrusted, } } @@ -357,7 +428,10 @@ func (s openCodeFormatSourceSet) Discover(ctx context.Context) ([]SourceRef, err if src.DBPath == "" || !IsRegularFile(src.DBPath) { continue } - dbSources, err := s.sqliteSources(ctx, root, src.DBPath, storageIDs) + trusted := s.containerTrusted != nil && s.containerTrusted(src.DBPath) + dbSources, err := s.sqliteSources( + ctx, root, src.DBPath, storageIDs, trusted, + ) if err != nil { if ctx.Err() != nil { return nil, err @@ -458,7 +532,16 @@ func (s openCodeFormatSourceSet) discoverRootEach( } var callbackErr error var membershipErr error - err := s.spec.streamSQLite(ctx, src.DBPath, func(meta OpenCodeSessionMeta) error { + // A container the engine's gate will skip wholesale streams the bounded + // watermark listing: computing every session's child digest for a pass + // that verifies nothing would be archive-sized work nothing reads (see + // ProviderConfig.SQLiteContainerUnchangedSinceTrust). + stream := s.spec.streamSQLite + if s.containerTrusted != nil && s.containerTrusted(src.DBPath) && + s.spec.streamSQLiteWatermark != nil { + stream = s.spec.streamSQLiteWatermark + } + err := stream(ctx, src.DBPath, func(meta OpenCodeSessionMeta) error { if storageIDs != nil { _, exists, err := storageIDs.get(ctx, meta.SessionID) if err != nil { @@ -603,7 +686,7 @@ func (s openCodeFormatSourceSet) SourcesForChangedPath( } for _, root := range s.roots { sources, ok, err := s.sourcesForChangedPathInRoot( - ctx, root, req.Path, pathExists, + ctx, root, req.Path, pathExists, req.AllowWatermarkOnlySources, ) if err != nil || ok { return sources, err @@ -697,6 +780,19 @@ func (s openCodeFormatSourceSet) FindSource( return SourceRef{}, false, nil } +// sourceMtimeWithComposite resolves a source's change signal when discovery did +// not carry one (FindSource lookups, storage sessions), reporting whether the +// value is the per-session composite. +func (s openCodeFormatSourceSet) sourceMtimeWithComposite( + path string, +) (int64, string, bool, error) { + if dbPath, sessionID, ok := s.spec.parseVirtual(path); ok { + return openCodeSQLiteSessionMtimeComposite(dbPath, sessionID) + } + mtime, err := s.spec.sourceMtime(path) + return mtime, "", false, err +} + func (s openCodeFormatSourceSet) Fingerprint( ctx context.Context, source SourceRef, @@ -709,12 +805,37 @@ func (s openCodeFormatSourceSet) Fingerprint( return SourceFingerprint{}, fmt.Errorf("%s source path unavailable", s.spec.agent) } mtime := sourceCarriedMTimeNS(source) - if mtime == 0 { - var err error - mtime, err = s.spec.sourceMtime(path) + composite := sourceCarriedCompositeMTime(source) + digest := sourceCarriedChildDigest(source) + // Only re-open the container when a digest is actually expected. A legacy + // container reports composite=false and carries an empty digest by design, + // so treating "empty" alone as "missing" would reopen and re-query the + // shared database once per session on every cold or changed-container pass. + if mtime == 0 || (composite && digest == "") { + // Sources rebuilt by FindSource or reconciliation carry no discovery + // metadata, and watermark-only changed-path sources carry a + // deliberately unresolved digest. Without this the hash would be + // empty, and an empty hash is treated as no constraint by the + // freshness gate — so a deletion-only change would pass unnoticed on + // every non-discovery path. + lookupMtime, lookupDigest, lookupComposite, err := + s.sourceMtimeWithComposite(path) if err != nil { return SourceFingerprint{}, err } + // Adopt the looked-up watermark alongside the digest: a + // watermark-only source carries the session-row watermark, which can + // sit below the composite the digest folds in. The stored MTimeNS + // must always be the composite, or the next full-discovery pass + // would see a mismatched watermark and re-parse an unchanged session. + if lookupMtime != 0 { + mtime, composite = lookupMtime, lookupComposite + } else if mtime == 0 { + composite = lookupComposite + } + if digest == "" { + digest = lookupDigest + } } fingerprint := SourceFingerprint{ Key: firstNonEmptyJSONLString(source.FingerprintKey, source.Key, path), @@ -725,7 +846,25 @@ func (s openCodeFormatSourceSet) Fingerprint( if err != nil { return SourceFingerprint{}, fmt.Errorf("stat %s: %w", dbPath, err) } - fingerprint.Size = info.Size() + // The watermark alone cannot see a deleted child, because the session + // or project row usually already holds the higher timestamp. The + // digest folds in the child row counts so a delete changes the + // fingerprint; FingerprintHashRequiredForFreshness makes the gate + // compare it against the stored value. + fingerprint.Hash = digest + // Every session in this root shares one physical container, so the + // container's size moves whenever any single session is written. + // Stamping it onto a per-session fingerprint made one session's + // append change the fingerprint of every other session in the + // container, dropping their freshness skip and re-parsing the whole + // root for one changed session. When MTimeNS is the per-session + // composite it already discriminates per session (including in-place + // child edits and project worktree renames), so the container stat + // is existence-only. Legacy containers whose schema cannot produce + // the composite keep the size as their conservative fallback. + if !composite { + fingerprint.Size = info.Size() + } return fingerprint, nil } info, err := os.Stat(path) @@ -749,6 +888,49 @@ func (s openCodeFormatSourceSet) Fingerprint( // sourceCarriedMTimeNS returns the discovery-listed session mtime carried on // a SQLite-backed source, or zero when the source was built without one // (storage sessions, FindSource lookups). +func sourceCarriedChildDigest(source SourceRef) string { + switch src := source.Opaque.(type) { + case openCodeFormatSource: + return src.ChildDigest + case *openCodeFormatSource: + if src != nil { + return src.ChildDigest + } + } + return "" +} + +// SourceWatermarkOnlyMTimeNS returns the carried session-row watermark for a +// shared-container source listed by a watermark-only changed-path scan, and +// whether the source is such a listing. Full-discovery sources carry the +// composite watermark and child digest instead and report false, as do +// legacy containers without composite support. +func SourceWatermarkOnlyMTimeNS(source SourceRef) (int64, bool) { + switch src := source.Opaque.(type) { + case openCodeFormatSource: + if src.WatermarkOnly { + return src.MTimeNS, true + } + case *openCodeFormatSource: + if src != nil && src.WatermarkOnly { + return src.MTimeNS, true + } + } + return 0, false +} + +func sourceCarriedCompositeMTime(source SourceRef) bool { + switch src := source.Opaque.(type) { + case openCodeFormatSource: + return src.CompositeMTime + case *openCodeFormatSource: + if src != nil { + return src.CompositeMTime + } + } + return false +} + func sourceCarriedMTimeNS(source SourceRef) int64 { switch src := source.Opaque.(type) { case openCodeFormatSource: @@ -790,11 +972,16 @@ func (s openCodeFormatSourceSet) sqliteSources( root string, dbPath string, storageIDs map[string]struct{}, + watermarkOnly bool, ) ([]SourceRef, error) { if err := ctx.Err(); err != nil { return nil, err } - metas, err := s.spec.listSQLite(dbPath) + lister := s.spec.listSQLite + if watermarkOnly && s.spec.listSQLiteWatermark != nil { + lister = s.spec.listSQLiteWatermark + } + metas, err := lister(dbPath) if err != nil { return nil, err } @@ -839,6 +1026,9 @@ func (s openCodeFormatSourceSet) sqliteSourceRefFromMeta( ref := s.newSourceRef(root, path, "") if src, ok := ref.Opaque.(openCodeFormatSource); ok { src.MTimeNS = meta.FileMtime + src.CompositeMTime = meta.CompositeMtime + src.ChildDigest = meta.ChildDigest + src.WatermarkOnly = meta.WatermarkOnly ref.Opaque = src } return ref, true @@ -849,6 +1039,7 @@ func (s openCodeFormatSourceSet) sourcesForChangedPathInRoot( root string, path string, pathExists bool, + watermarkOnly bool, ) ([]SourceRef, bool, error) { rel, ok := relUnder(root, path) if !ok { @@ -883,7 +1074,9 @@ func (s openCodeFormatSourceSet) sourcesForChangedPathInRoot( if s.spec.resolve(root).Mode == OpenCodeSourceStorage { storageIDs = s.spec.storageIDs(root) } - sources, err := s.sqliteSources(ctx, root, dbPath, storageIDs) + sources, err := s.sqliteSources( + ctx, root, dbPath, storageIDs, watermarkOnly, + ) return sources, true, err } @@ -1128,6 +1321,11 @@ func openCodeFormatProviderCapabilities() Capabilities { }, Sync: ProviderSyncSemantics{ UnchangedResults: UnchangedResultMTimeAndHash, + // The per-session digest is the only signal that sees a deleted + // child, so freshness must consult it. Containers without + // composite support produce an empty hash, which the gate treats + // as no constraint, preserving their previous behavior. + FingerprintHashRequiredForFreshness: true, }, } } diff --git a/internal/parser/opencode_provider_test.go b/internal/parser/opencode_provider_test.go index 1adef6b05..5a8f9688c 100644 --- a/internal/parser/opencode_provider_test.go +++ b/internal/parser/opencode_provider_test.go @@ -148,7 +148,7 @@ func TestOpenCodeStreamingPartialSQLiteFailureContinuesLaterRoots(t *testing.T) } } sources := newOpenCodeFormatSourceSet( - []string{partialRoot, healthyRoot}, spec, + []string{partialRoot, healthyRoot}, spec, nil, ) var paths []string @@ -736,9 +736,12 @@ func TestOpenCodeProviderSQLiteFingerprintUsesDiscoveryMeta(t *testing.T) { "fingerprint must not reopen the SQLite DB for a discovered source") assert.Equal(t, OpenCodeSQLiteVirtualPath(dbPath, "ses_meta"), fp.Key) assert.Equal(t, int64(1700000010000000000), fp.MTimeNS, - "fingerprint mtime must be the discovered time_updated in ns") - assert.Equal(t, int64(len(garbage)), fp.Size, - "fingerprint size stays the shared container file size") + "fingerprint mtime must be the discovered composite in ns") + assert.Zero(t, fp.Size, + "a per-session fingerprint must not carry the shared container's "+ + "size: every session in the root shares one opencode.db, so any "+ + "one session's write would change every other session's "+ + "fingerprint and drop its freshness skip") } func TestOpenCodeProviderHybridDiscoveryFiltersSQLiteDuplicate(t *testing.T) { @@ -1100,3 +1103,99 @@ func newTestDBAt( require.NoError(t, err, "open test db") return dbPath, &OpenCodeSeeder{db: db, t: t}, db } + +// TestOpenCodeSingleSessionMtimeDoesNotScanContainer pins the query shape of +// the single-session composite lookup. Reusing the streaming form's grouped +// subqueries here materializes an aggregate over every message and part in the +// container before the outer WHERE narrows to one session, so each per-session +// lookup would scan the whole archive. Assert the plan touches the child tables +// through their session_id indexes rather than a full scan. +func TestOpenCodeSingleSessionMtimeDoesNotScanContainer(t *testing.T) { + root := t.TempDir() + _, seeder, db := newTestDBAt(t, filepath.Join(root, "opencode.db")) + seeder.AddProject("prj_1", "/home/user/code/app") + seeder.AddSession( + "ses_a", "prj_1", "", "A", 1700000000000, 1700000010000, + ) + t.Cleanup(func() { _ = db.Close() }) + + query := "SELECT " + openCodeSessionCompositeMtimeExpr + + " FROM session s" + openCodeSessionCompositeMtimeJoins + + " WHERE s.id = ?" + rows, err := db.Query("EXPLAIN QUERY PLAN "+query, "ses_a") + require.NoError(t, err) + defer rows.Close() + + var plan strings.Builder + for rows.Next() { + var id, parent, notUsed int + var detail string + require.NoError(t, rows.Scan(&id, &parent, ¬Used, &detail)) + plan.WriteString(detail) + plan.WriteString("\n") + } + require.NoError(t, rows.Err()) + + got := plan.String() + for _, table := range []string{"message", "part"} { + assert.NotContains(t, got, "SCAN "+table, + "single-session composite mtime must not full-scan %s; plan:\n%s", + table, got) + // SEARCH alone is not proof of a seek: SQLite reports SEARCH for some + // aggregate plans without an index, so require the index explicitly. + assert.Regexp(t, + `(?s)(SEARCH|SCAN) `+table+`[^\n]*USING (COVERING )?INDEX`, + got, + "single-session composite mtime must reach %s through an index; "+ + "plan:\n%s", table, got) + } +} + +// TestOpenCodeWatermarkOnlyQuerySkipsDigestScans pins that the mtime-only path +// does not compute the digest aggregates. OpenCodeSourceMtime backs the session +// watcher's 1.5s poll, so pulling the eight child COUNT/SUM/MIN/MAX subqueries +// in there would burn child-range scans per tick for a discarded value. +func TestOpenCodeWatermarkOnlyQuerySkipsDigestScans(t *testing.T) { + watermarkOnly := "SELECT " + openCodeSessionCompositeMtimeExpr + + " FROM session s" + openCodeSessionCompositeMtimeJoins + + " WHERE s.id = ?" + full := "SELECT " + openCodeSessionCompositeMtimeExpr + ", " + + openCodeSessionCompositeCountsExpr + + " FROM session s" + openCodeSessionCompositeMtimeJoins + + " WHERE s.id = ?" + + assert.NotContains(t, watermarkOnly, "COUNT(", + "the mtime-only query must not compute child counts") + assert.NotContains(t, watermarkOnly, "group_concat(", + "the mtime-only query must not build child identities") + assert.Contains(t, full, "COUNT(", + "the fingerprint query must still compute the digest aggregates") + + // Both must be executable, not merely string-shaped: assert against a real + // container so a query that only looks right still fails here. + root := t.TempDir() + _, seeder, db := newTestDBAt(t, filepath.Join(root, "opencode.db")) + seeder.AddProject("prj_1", "/home/user/code/app") + seeder.AddSession( + "ses_a", "prj_1", "", "A", 1700000000000, 1700000010000, + ) + t.Cleanup(func() { _ = db.Close() }) + + var watermark int64 + require.NoError(t, + db.QueryRow(watermarkOnly, "ses_a").Scan(&watermark), + "the mtime-only query must execute") + assert.Equal(t, int64(1700000010000), watermark) + + var ( + w, st, pt, mn, pn int64 + mIdent, pIdent string + ) + require.NoError(t, + db.QueryRow(full, "ses_a").Scan( + &w, &st, &pt, &mn, &pn, &mIdent, &pIdent, + ), + "the fingerprint query must execute") + assert.Equal(t, watermark, w, + "both queries must agree on the watermark") +} diff --git a/internal/parser/opencode_test.go b/internal/parser/opencode_test.go index 157cebe7f..75930346a 100644 --- a/internal/parser/opencode_test.go +++ b/internal/parser/opencode_test.go @@ -76,6 +76,14 @@ CREATE TABLE part ( data TEXT NOT NULL, FOREIGN KEY (message_id) REFERENCES message(id) ); + +-- SQLite does not index a foreign key automatically. Production OpenCode +-- declares these, and the per-session freshness lookups depend on them, so the +-- fixture must carry them or plan assertions prove nothing. +CREATE INDEX message_session_time_created_id_idx + ON message (session_id, time_created, id); +CREATE INDEX part_session_idx ON part (session_id); +CREATE INDEX part_message_id_id_idx ON part (message_id, id); ` func assertEq[T comparable](t *testing.T, name string, got, want T) { @@ -1501,6 +1509,108 @@ func TestListOpenCodeSessionMeta_NonexistentDB(t *testing.T) { assertEq(t, "metas len", len(metas), 0) } +// TestListOpenCodeSessionWatermarkMeta pins the bounded changed-path listing: +// on a composite-capable container it carries only the session-row watermark +// (session and project time_updated, never child times) with no digest, so +// listing every session touches no message or part rows. +func TestListOpenCodeSessionWatermarkMeta(t *testing.T) { + dbPath, seeder, db := newTestDB(t) + defer db.Close() + + seeder.AddProject("prj_1", "/home/user/code/app") + seeder.AddSession( + "ses_wm", "prj_1", "", "Watermark", 1700000000000, 1700000060000, + ) + seeder.AddMessage( + "msg_1", "ses_wm", 1700000000000, 1700099999000, `{"role":"user"}`, + ) + seeder.AddPart( + "prt_1", "msg_1", "ses_wm", 1700000000000, 1700099999000, + `{"type":"text","text":"hi"}`, + ) + // Project row above the session row: the watermark is MAX(session, + // project). Child rows sit above both and must NOT be reflected. + _, err := db.Exec( + "UPDATE project SET time_updated = ? WHERE id = ?", + 1700000070000, "prj_1", + ) + require.NoError(t, err, "raise project time") + + metas, err := ListOpenCodeSessionWatermarkMeta(dbPath) + require.NoError(t, err, "ListOpenCodeSessionWatermarkMeta") + require.Len(t, metas, 1) + + m := metas[0] + assert.Equal(t, "ses_wm", m.SessionID) + assert.Equal(t, dbPath+"#ses_wm", m.VirtualPath) + assert.True(t, m.WatermarkOnly, "composite container must list watermark-only") + assert.True(t, m.CompositeMtime) + assert.Empty(t, m.ChildDigest, "watermark listing must not resolve the child digest") + assert.Equal(t, int64(1700000070000)*1_000_000, m.FileMtime, + "watermark must be MAX(session, project) and exclude child times") +} + +// TestOpenCodeChildDigestMetadataWatermarkNS pins the digest round-trip the +// watcher's like-for-like comparison depends on: the session/project times a +// digest embeds must come back out as the metadata watermark, and every +// other hash shape must be rejected so callers fall back to the composite. +func TestOpenCodeChildDigestMetadataWatermarkNS(t *testing.T) { + agg := openCodeChildAggregate{ + watermark: 1700000099000, + sessionTime: 1700000060000, + projectTime: 1700000070000, + messages: 2, + parts: 5, + messageIdent: "m1:1", + partIdent: "p1:1", + } + got, ok := OpenCodeChildDigestMetadataWatermarkNS(agg.digest(true)) + require.True(t, ok, "digest must round-trip its metadata watermark") + assert.Equal(t, int64(1700000070000)*1_000_000, got, + "metadata watermark must be MAX(session, project), not the composite") + + for _, hash := range []string{ + "", + agg.digest(false), + openCodeStorageFingerprintPrefix + "abcdef", + "opencode-child:v2:1:2:3:4:5:aabb", + "opencode-child:v1:1:2:3", + "opencode-child:v1:x:2:3:4:5:aabb", + "opencode-child:v1:1:x:3:4:5:aabb", + "opencode-child:v1:1:2:x:4:5:aabb", + } { + _, ok := OpenCodeChildDigestMetadataWatermarkNS(hash) + assert.False(t, ok, "hash %q must be rejected", hash) + } +} + +// TestListOpenCodeSessionWatermarkMeta_LegacySchema pins that containers +// without composite support keep the full listing's shape: session-only +// mtime, no composite, and no watermark-only marker, so the engine never +// watermark-skips a session whose only change signal is the container size. +func TestListOpenCodeSessionWatermarkMeta_LegacySchema(t *testing.T) { + dbPath, seeder, db := newLegacyOpenCodeTestDB(t) + defer db.Close() + + seeder.AddProject("prj_legacy", "/home/user/code/legacy-app") + seeder.AddSession( + "ses_legacy", "prj_legacy", "", "Legacy", 1700000000000, 1700000060000, + ) + + metas, err := ListOpenCodeSessionWatermarkMeta(dbPath) + require.NoError(t, err, "ListOpenCodeSessionWatermarkMeta legacy") + require.Len(t, metas, 1) + + full, err := ListOpenCodeSessionMeta(dbPath) + require.NoError(t, err, "ListOpenCodeSessionMeta legacy") + require.Len(t, full, 1) + + assert.False(t, metas[0].WatermarkOnly, + "legacy containers must not be marked watermark-only") + assert.Equal(t, full[0], metas[0], + "legacy watermark listing must match the full listing") +} + // TestParseOpenCodeDB_TokenUsage verifies that an assistant // message with modelID and tokens populates ParsedMessage.Model // and TokenUsage in the agentsview-native key shape, and that diff --git a/internal/parser/provider.go b/internal/parser/provider.go index 0202197ec..91d03f961 100644 --- a/internal/parser/provider.go +++ b/internal/parser/provider.go @@ -54,6 +54,17 @@ type ProviderConfig struct { // path (Aider) use it to seed those IDs from the canonical remote path // rather than the changing temp path. Most providers ignore it. PathRewriter func(string) string + // SQLiteContainerUnchangedSinceTrust reports that the shared SQLite + // container at dbPath is byte-identical to the last pass that verified + // every one of its sessions, as captured before this discovery began. + // Providers that fan such a container out to per-session sources may + // answer discovery for it with the bounded watermark-only listing: the + // caller's container gate will skip every member before fingerprinting, + // so computing the full child digest would be archive-sized work for + // values nothing reads. Nil (the default, and every non-discovery + // construction) means no container is trusted and listings stay + // full-fidelity. + SQLiteContainerUnchangedSinceTrust func(dbPath string) bool } // Clone returns an independent config snapshot. @@ -366,6 +377,14 @@ type ChangedPathRequest struct { // still validate ownership against the changed path/watch root before // emitting them. StoredSourcePaths []string + // AllowWatermarkOnlySources tells providers that fan a shared container + // out to per-session virtual sources that the caller's downstream + // freshness gate can cheaply skip watermark-only sources, so the provider + // may answer with a bounded session-row listing instead of computing + // every session's full child digest. Callers that consume the returned + // sources directly (reconciliation, tombstoning) must leave this unset to + // keep full-fidelity fingerprint metadata. + AllowWatermarkOnlySources bool } // FindSourceRequest contains lookup inputs and persisted source hints for diff --git a/internal/sync/engine.go b/internal/sync/engine.go index 91f40d787..53546b21f 100644 --- a/internal/sync/engine.go +++ b/internal/sync/engine.go @@ -1086,6 +1086,12 @@ func (e *Engine) classifyProviderChangedPath( Path: path, EventKind: eventKind, WatchRoot: watchRoot, + // The changed-path pipeline drops watermark-only + // shared-container sources whose stored composite watermark + // already covers them (filterFreshWatermarkOnlySources), so + // providers may answer with the bounded session-row listing + // instead of a whole-container child digest scan. + AllowWatermarkOnlySources: true, } if provider.Capabilities().Source.StoredSourceHints == parser.CapabilitySupported { if resolver, ok := provider.(parser.StoredSourceHintScopeProvider); ok { @@ -1126,6 +1132,9 @@ func (e *Engine) classifyProviderChangedPath( } continue } + sources = e.filterFreshWatermarkOnlySources( + ctx, agentType, roots, path, sources, + ) if def.Type == parser.AgentOmnigent { sources, err = e.expandOmnigentInheritedMetadataSources( ctx, provider, sources, @@ -3286,7 +3295,7 @@ func (e *Engine) reconcileWatchRootsStreamed( preContainerStates := e.captureSQLiteContainerStates(nil) providers, completedScopes, failedRoots, failures, discoveryErr, err := e.streamReconciliationCandidates( - ctx, scope, spool, + ctx, scope, spool, preContainerStates, ) stats.providerFailures = failures if err != nil { @@ -3555,6 +3564,7 @@ func (e *Engine) streamReconciliationCandidates( ctx context.Context, scope *rootSyncScope, spool reconciliationSpoolStore, + preContainerStates map[string]parser.SQLiteContainerState, ) ( map[parser.AgentType]parser.Provider, []reconciliationProviderScope, @@ -3568,6 +3578,13 @@ func (e *Engine) streamReconciliationCandidates( var failedRoots []string var failures int var discoveryErr error + // Trusted containers stream the bounded watermark listing here for the + // same reason full discovery lists them that way: every candidate they + // spool will gate-skip, so the child digest would be archive-sized work + // nothing reads. The predicate is keyed to this pass's pre-discovery + // captures; a container that changes mid-stream fails its recapture + // check and its candidates resolve full fingerprints instead. + containerTrusted := e.sqliteContainerTrustedForDiscovery(preContainerStates) agents := make([]parser.AgentType, 0, len(e.providerFactories)) for agent := range e.providerFactories { agents = append(agents, agent) @@ -3594,6 +3611,7 @@ func (e *Engine) streamReconciliationCandidates( } provider := factory.NewProvider(parser.ProviderConfig{ Roots: roots, Machine: e.machine, PathRewriter: e.pathRewriter, + SQLiteContainerUnchangedSinceTrust: containerTrusted, }) providers[agent] = provider if provider.Capabilities().Source.StreamingDiscovery != parser.CapabilitySupported { @@ -4655,7 +4673,9 @@ func (e *Engine) syncAllLocked( var all []parser.DiscoveredFile counts := make(map[parser.AgentType]int) - providerFound, providerFailures := e.discoverProviderSources(ctx, scope) + providerFound, providerFailures := e.discoverProviderSources( + ctx, scope, preContainerStates, + ) for _, file := range providerFound { counts[file.Agent]++ } @@ -4916,9 +4936,11 @@ const slowProviderDiscoveryThreshold = 100 * time.Millisecond func (e *Engine) discoverProviderSources( ctx context.Context, scope *rootSyncScope, + preContainerStates map[string]parser.SQLiteContainerState, ) ([]parser.DiscoveredFile, int) { var files []parser.DiscoveredFile var failures int + containerTrusted := e.sqliteContainerTrustedForDiscovery(preContainerStates) agents := make([]parser.AgentType, 0, len(e.providerFactories)) for agent := range e.providerFactories { @@ -4954,8 +4976,9 @@ func (e *Engine) discoverProviderSources( continue } provider := factory.NewProvider(parser.ProviderConfig{ - Roots: filteredRoots, - Machine: e.machine, + Roots: filteredRoots, + Machine: e.machine, + SQLiteContainerUnchangedSinceTrust: containerTrusted, }) // Shared-database providers are streamed source-by-source by their // dedicated sync phase. Calling Discover here would build an archive-sized @@ -5440,6 +5463,18 @@ func (e *Engine) discoveredFileEffectiveMtime( } return mtime, nil } + // Watermark-only shared-container sources carry their session-row + // watermark from discovery. Consulting the provider Fingerprint instead + // would resolve the full composite with one indexed child lookup per + // session, scaling cutoff filtering with the container instead of the + // changed batch — and these sources are only listed for containers that + // provably have not changed since their last verified pass, where the + // carried watermark and the composite are equally stale. + if file.ProviderSource != nil { + if wm, ok := parser.SourceWatermarkOnlyMTimeNS(*file.ProviderSource); ok { + return wm, nil + } + } // Provider-authoritative sources resolve freshness through the provider // Fingerprint so composite provider-owned source state participates in // incremental-sync cutoff checks. @@ -7170,6 +7205,19 @@ func (e *Engine) processProviderFile( }, true } + // Watermark-only shared-container sources (changed-path classification) + // carry just the session-row watermark. When it does not advance past + // the stored composite watermark, the session and project rows provably + // did not change, so skip before Fingerprint pays the per-session child + // lookup; a child-only edit this cannot see is reconciled by the next + // full-discovery pass, whose digest comparison still catches it. + if freshMtime, fresh := e.watermarkOnlySQLiteSourceFresh(source, file); fresh { + return processResult{ + skip: true, + mtime: freshMtime, + }, true + } + fingerprint, err := provider.Fingerprint(ctx, source) if err != nil { if file.ForceParse && @@ -8621,7 +8669,7 @@ func (e *Engine) providerIncrementalContentChanged( if !ok || storedHash == "" { return false, false } - curHash, err := ComputeFileHashPrefix(hashPath, info.Size()) + curHash, err := computeFileHashPrefix(hashPath, info.Size()) if err != nil { return false, false } diff --git a/internal/sync/engine_integration_test.go b/internal/sync/engine_integration_test.go index 21af8036e..a888000d7 100644 --- a/internal/sync/engine_integration_test.go +++ b/internal/sync/engine_integration_test.go @@ -694,11 +694,16 @@ func TestSyncEngineOpenCodeSQLiteSameMtimeContentChangeUsesFingerprint( "local_modified_at before rewrite") time.Sleep(20 * time.Millisecond) + // The session row's own time_updated deliberately stays at + // 1779012030000. Production OpenCode stamps time_updated on every child + // row it writes, so the replacement children carry a newer one; that is + // the per-session signal the composite mtime reads, and it must catch a + // content change the session row alone cannot show. oc.replaceTextContent( t, "same-mtime-sqlite", "changed prompt with same session mtime", "changed answer with same session mtime", - 1779012000000, + 1779012600000, ) stats = env.engine.SyncAll(context.Background(), nil) @@ -715,8 +720,11 @@ func TestSyncEngineOpenCodeSQLiteSameMtimeContentChangeUsesFingerprint( require.NotNil(t, after.FileHash, "file_hash after rewrite") require.NotNil(t, after.LocalModifiedAt, "local_modified_at after rewrite") - assert.Equal(t, *before.FileMtime, *after.FileMtime, - "same-mtime rewrite keeps the OpenCode SQLite session mtime") + assert.Greater(t, *after.FileMtime, *before.FileMtime, + "child content newer than the session row must advance the stored "+ + "composite mtime: that per-session signal is what detects the "+ + "change without the shared container's stat invalidating every "+ + "other session in the same opencode.db") assert.NotEqual(t, *before.FileHash, *after.FileHash, "changed SQLite child content must change the storage fingerprint") assert.Greater(t, *after.LocalModifiedAt, *before.LocalModifiedAt, @@ -799,7 +807,7 @@ func TestSyncEngineOpenCodeSQLiteStatIdenticalContentChangeStillReemits( oc.replaceTextContent( t, "stat-twin", "replaced prompt", "replaced answer", - 1779012000000, + 1779012600000, ) after, err := os.Stat(dbPath) require.NoError(t, err, "stat opencode.db after rewrite") @@ -908,8 +916,15 @@ func TestSyncEngineOpenCodeSQLiteCwdFilteredContainerStaysUntrusted( stats = engine.SyncAll(context.Background(), nil) require.False(t, stats.Aborted, "second sync aborted: %+v", stats) - assert.Equal(t, 0, stats.Skipped, - "a container with cwd-vetoed sessions must not be gate-skipped") + // Exactly one skip: the persisted allowed session rides its own + // per-session freshness check. The vetoed session was never written, so + // it has no stored row to be fresh against and must be processed again. + // A trusted-container gate skip would cover both sessions and make this + // 2, which is the promotion violation this test exists to catch. + assert.Equal(t, 1, stats.Skipped, + "a container with cwd-vetoed sessions must not be gate-skipped: "+ + "only the persisted session may skip, and only on its own "+ + "per-session freshness") kept, err := database.GetSessionFull( context.Background(), "opencode:keep-session", @@ -1188,9 +1203,8 @@ func TestSyncEngineOpenCodeSQLiteSameMtimeMetadataChangeUsesFingerprint( assert.Equal(t, "original_app", before.Project) time.Sleep(20 * time.Millisecond) - oc.mustExec(t, "update project worktree", - "UPDATE project SET worktree = ? WHERE id = ?", - "/home/user/code/renamed-app", "proj", + oc.updateProjectWorktree( + t, "proj", "/home/user/code/renamed-app", 1779015630000, ) stats = env.engine.SyncAll(context.Background(), nil) @@ -1205,8 +1219,11 @@ func TestSyncEngineOpenCodeSQLiteSameMtimeMetadataChangeUsesFingerprint( require.NotNil(t, after.FileHash, "file_hash after rewrite") require.NotNil(t, after.LocalModifiedAt, "local_modified_at after rewrite") - assert.Equal(t, *before.FileMtime, *after.FileMtime, - "metadata-only rewrite keeps the OpenCode SQLite session mtime") + assert.Greater(t, *after.FileMtime, *before.FileMtime, + "a project worktree rename must advance the session's composite "+ + "mtime: project.time_updated is part of the per-session change "+ + "signal, which is what re-resolves cwd without the shared "+ + "container's stat invalidating every unrelated session") assert.NotEqual(t, *before.FileHash, *after.FileHash, "changed SQLite metadata must change the storage fingerprint") assert.Greater(t, *after.LocalModifiedAt, *before.LocalModifiedAt, diff --git a/internal/sync/hash.go b/internal/sync/hash.go index 425f7854b..d90ff14ad 100644 --- a/internal/sync/hash.go +++ b/internal/sync/hash.go @@ -30,6 +30,18 @@ func ComputeFileHash(path string) (string, error) { return hash, nil } +// computeFileHashPrefix indirects ComputeFileHashPrefix so cardinality-scaling +// tests can count the source-content reads a sync pass performs *to decide +// freshness*. Only providerIncrementalContentChanged goes through it today. +// +// The incremental-append path also hashes (to refresh a stored fingerprint +// after consuming new bytes) and deliberately calls ComputeFileHashPrefix +// directly: that read is write-side work proportional to data that genuinely +// changed, not a per-session freshness probe, so counting it would blur the +// invariant TestWarmFullSyncDoesNotRehashClaudeArchive pins. Route a new +// freshness gate through this var; leave write-side hashing on the direct call. +var computeFileHashPrefix = ComputeFileHashPrefix + // ComputeFileHashPrefix returns the SHA-256 hex digest of the first size bytes // of the file at path. It returns an error if the file is shorter than size. func ComputeFileHashPrefix(path string, size int64) (string, error) { diff --git a/internal/sync/opencode_container_gate.go b/internal/sync/opencode_container_gate.go index 6b44f8400..9ce15c37e 100644 --- a/internal/sync/opencode_container_gate.go +++ b/internal/sync/opencode_container_gate.go @@ -3,6 +3,7 @@ package sync import ( + "context" "maps" "path/filepath" "strings" @@ -138,6 +139,144 @@ func (e *Engine) captureSQLiteContainerStates( return states } +// openCodeContainerPathForChangedPathEvent maps a changed-path event to the +// shared SQLite container it names for one OpenCode-family agent, or "" +// when the agent has no container or the event is not a container write. +func openCodeContainerPathForChangedPathEvent( + agent parser.AgentType, + roots []string, + path string, +) string { + if openCodeFormatDBName(agent) == "" { + return "" + } + for _, dir := range roots { + if dir == "" || strings.HasPrefix(dir, "s3://") { + continue + } + if container := openCodeContainerPathForEvent(agent, dir, path); container != "" { + return container + } + } + return "" +} + +// filterFreshWatermarkOnlySources drops watermark-only shared-container +// sources whose stored session/project metadata watermark (recovered from +// the stored child digest, see storedSessionRowWatermarkNS) already covers +// the carried session-row watermark, before they are materialized into +// discovered files. The stored values come from one indexed range query over +// the container's virtual members, so a one-session write flows one +// candidate into the sync pipeline instead of every session in the +// container. The comparison is per-session and like-for-like — a session or +// project row that advances past its own stored metadata watermark is always +// kept, wherever other sessions' watermarks or its own child timestamps sit +// — and sessions with no stored row or a stale data version are kept +// unconditionally. +// +// Known, deliberate deferral (not a detection gap to "fix" here): a +// child-only write that leaves the session and project rows untouched is +// invisible to the session-row watermark wherever its timestamps land — +// above or below the stored composite alike. Detecting it per event would +// require reading child rows, which is exactly the archive-sized work this +// path exists to avoid. Such writes reconcile on the next full-discovery +// pass, whose digest still catches them (the write itself broke container +// trust, so that pass carries the full digest); actively watched sessions +// bypass this path entirely via the per-session composite poll. The +// contract is documented in docs/internal/session-format-sources.md and +// pinned by TestOpenCodeWatcherPassDefersChildOnlyEditToFullDiscovery. +// Fails open: on a query error every source is kept and the per-file gate +// decides instead. +func (e *Engine) filterFreshWatermarkOnlySources( + ctx context.Context, + agent parser.AgentType, + roots []string, + path string, + sources []parser.SourceRef, +) []parser.SourceRef { + if len(sources) == 0 || e.forceParse || e.pathRewriter != nil { + return sources + } + container := openCodeContainerPathForChangedPathEvent(agent, roots, path) + if container == "" { + return sources + } + watermarkOnly := false + for i := range sources { + if _, ok := parser.SourceWatermarkOnlyMTimeNS(sources[i]); ok { + watermarkOnly = true + break + } + } + if !watermarkOnly { + return sources + } + stored, err := e.db.ListVirtualContainerMemberFreshness(ctx, container) + if err != nil || len(stored) == 0 { + return sources + } + current := db.CurrentDataVersion() + kept := make([]parser.SourceRef, 0, len(sources)) + for _, source := range sources { + if watermark, ok := parser.SourceWatermarkOnlyMTimeNS(source); ok { + member, found := stored[providerDiscoveredPath(source)] + if found && watermark <= storedSessionRowWatermarkNS(member) && + member.DataVersion >= current { + continue + } + } + kept = append(kept, source) + } + return kept +} + +// storedSessionRowWatermarkNS resolves the stored value a carried session-row +// watermark is compared against, like-for-like: the session/project metadata +// watermark recovered from the stored child digest. Comparing against the +// stored composite MTimeNS instead would over-skip — a composite dominated by +// a newer child timestamp would hide a metadata update (title, directory, +// worktree rename) whose stamp lands below it. Rows without a parseable +// digest (pre-digest fingerprints, future digest versions) fall back to the +// composite, the conservative pre-digest behavior that self-heals on the +// row's next reparse. +func storedSessionRowWatermarkNS( + member db.VirtualContainerMemberFreshness, +) int64 { + if metadata, ok := parser.OpenCodeChildDigestMetadataWatermarkNS( + member.Hash, + ); ok { + return metadata + } + return member.MTimeNS +} + +// sqliteContainerTrustedForDiscovery returns discovery's trust probe: it +// reports containers whose pre-discovery capture matches the last fully +// verified state, meaning every member will gate-skip before fingerprinting +// and the full child digest would be computed for nothing. The probe is +// keyed to the pass's own pre-discovery captures so a container that +// changes between capture and listing can never look trusted with a newer +// session set (the gate separately fails such containers for the pass). +// Nil when nothing was captured or every parse is forced. +func (e *Engine) sqliteContainerTrustedForDiscovery( + preStates map[string]parser.SQLiteContainerState, +) func(string) bool { + if len(preStates) == 0 || e.forceParse { + return nil + } + return func(dbPath string) bool { + dbPath = filepath.Clean(dbPath) + state, ok := preStates[dbPath] + if !ok { + return false + } + e.containerMu.Lock() + trusted, ok := e.trustedSQLiteContainers[dbPath] + e.containerMu.Unlock() + return ok && trusted.state == state + } +} + func openCodeContainerPathForEvent( agent parser.AgentType, root string, @@ -277,6 +416,85 @@ func (e *Engine) sqliteContainerSourceFresh(file parser.DiscoveredFile) bool { e.db.GetSessionFilePath(fullID) == e.effectiveSourcePath(file.Path) } +// watermarkOnlySQLiteSourceFresh reports whether a shared-container session +// whose source carries only the session-row watermark is already covered by +// its stored session/project metadata watermark, compared like-for-like: +// the stored value is recovered from the persisted child digest, falling +// back to the stored composite MTimeNS for rows without a parseable digest. +// A session-row watermark at or below the stored metadata watermark proves +// the session and project rows did not advance, so the parse is skipped +// without resolving the child digest. What the watermark cannot see — any +// child-only write that leaves the session and project rows untouched — is +// deliberately deferred to the next full-discovery pass, whose carried +// digest still catches it (see filterFreshWatermarkOnlySources for the full +// contract). That keeps per-event work bounded by the changed batch instead +// of the archive. +func (e *Engine) watermarkOnlySQLiteSourceFresh( + source parser.SourceRef, + file parser.DiscoveredFile, +) (int64, bool) { + if e.forceParse || file.ForceParse { + return 0, false + } + watermark, ok := parser.SourceWatermarkOnlyMTimeNS(source) + if !ok { + return 0, false + } + // The skip is only sound while the pass's container capture is valid. A + // trusted full discovery lists watermark-only sources; if the container + // changes between that listing and the pass's recapture check, the + // capture is invalidated and a concurrent child-only write may hide + // beneath an unchanged metadata watermark — those sources must fall + // through to Fingerprint and resolve the full digest instead. + if dbPath, _, ok := sqliteContainerSourceForFile(file); !ok || + !e.sqliteContainerPassCaptureValid(dbPath) { + return 0, false + } + lookupPath := providerDiscoveredPath(source) + if lookupPath == "" { + return 0, false + } + if e.pathRewriter != nil { + lookupPath = e.pathRewriter(lookupPath) + } + _, storedMtime, found := e.db.GetFileInfoByPath(lookupPath) + if !found { + return 0, false + } + limit := storedMtime + if hash, ok := e.db.GetFileHashByPath(lookupPath); ok { + if metadata, parsed := parser.OpenCodeChildDigestMetadataWatermarkNS( + hash, + ); parsed { + limit = metadata + } + } + if limit < watermark { + return 0, false + } + if e.db.GetDataVersionByPath(lookupPath) < db.CurrentDataVersion() { + return 0, false + } + return storedMtime, true +} + +// sqliteContainerPassCaptureValid reports whether the current pass still +// holds a live capture for the container: one was taken before discovery, +// the post-discovery recapture matched it, and no processing failure has +// poisoned the container since. Watermark-only skips require this — an +// invalidated capture means the container changed while the pass was +// listing it, and the watermark cannot see what that change touched. +func (e *Engine) sqliteContainerPassCaptureValid(dbPath string) bool { + e.containerMu.Lock() + defer e.containerMu.Unlock() + pass := e.containerPass + if pass == nil || pass.failed[dbPath] { + return false + } + _, ok := pass.captured[dbPath] + return ok +} + // noteSQLiteContainerResult records a processed file's outcome for // promotion bookkeeping. Skips count as completions: a skipped session was // either gate-skipped against an already-trusted state or individually diff --git a/internal/sync/opencode_container_perf_test.go b/internal/sync/opencode_container_perf_test.go new file mode 100644 index 000000000..261dfd67f --- /dev/null +++ b/internal/sync/opencode_container_perf_test.go @@ -0,0 +1,571 @@ +package sync_test + +import ( + "context" + "fmt" + "testing" + + "go.kenn.io/agentsview/internal/parser" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestOpenCodeSharedContainerChangeIsPerSessionBounded pins the "background +// sync work is bounded by the changed batch, not total archive size" rule for +// shared SQLite containers. +// +// Every session in an OpenCode root lives in one physical opencode.db. Stamping +// that container's size onto each session's fingerprint made any single +// session's write change every other session's fingerprint, so one changed +// session re-parsed the whole root — on a production container that is +// thousands of sessions re-read out of a multi-GB database every time the +// watcher fires. The per-session composite mtime (session, project, and child +// message/part time_updated) replaces it, so a one-session change must leave +// every other session skipped regardless of how many there are. +func TestOpenCodeSharedContainerChangeIsPerSessionBounded(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test") + } + + rewritten := make(map[int]int) + for _, n := range []int{20, 200} { + t.Run(fmt.Sprintf("sessions_%d", n), func(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + for i := range n { + seedOpenCodeSQLiteTextSession( + t, oc, "proj", fmt.Sprintf("ses%05d", i), + 1779012000000, 1779012030000, + "prompt", "answer", + ) + } + require.Equal(t, n, + env.engine.SyncAll(context.Background(), nil).Synced) + + // Change exactly one session. This also grows the shared + // container file, which is precisely the signal that used to + // invalidate every other session in it. + oc.updateSessionTime(t, "ses00000", 1779015630000) + oc.replaceTextContent( + t, "ses00000", "changed prompt", "changed answer", + 1779015600000, + ) + + stats := env.engine.SyncAll(context.Background(), nil) + require.False(t, stats.Aborted, "sync aborted: %+v", stats) + assert.Equal(t, 1, stats.Synced, + "only the changed session may be rewritten") + assert.Equal(t, n-1, stats.Skipped, + "every unchanged session in the shared container must skip") + rewritten[n] = stats.Synced + }) + } + + assert.Equal(t, rewritten[20], rewritten[200], + "sessions rewritten for one changed session must not grow with "+ + "container size") +} + +// TestOpenCodeWatcherEventIsWatermarkBounded pins the same rule for the +// watcher's changed-path pass, one level deeper: a one-session write must +// not read the container's child tables at all, and must not even +// materialize the unchanged sessions. Changed-path classification lists +// candidates through the bounded session-row watermark (no message/part +// aggregation) filtered by the container's newest stored watermark, so the +// sources processed and the child rows examined per event both scale with +// the changed batch and not with the archive. +func TestOpenCodeWatcherEventIsWatermarkBounded(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test") + } + + lookups := make(map[int]int64) + processed := make(map[int]int) + for _, n := range []int{20, 200} { + t.Run(fmt.Sprintf("sessions_%d", n), func(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + for i := range n { + seedOpenCodeSQLiteTextSession( + t, oc, "proj", fmt.Sprintf("ses%05d", i), + 1779012000000, 1779012030000, + "prompt", "answer", + ) + } + require.Equal(t, n, + env.engine.SyncAll(context.Background(), nil).Synced) + + oc.updateSessionTime(t, "ses00000", 1779015630000) + oc.replaceTextContent( + t, "ses00000", "changed prompt", "changed answer", + 1779015600000, + ) + + scansBefore := parser.OpenCodeContainerChildScans() + lookupsBefore := parser.OpenCodeSessionChildLookups() + require.NoError(t, env.engine.SyncPathsContext( + context.Background(), []string{oc.path}, + )) + stats := env.engine.LastSyncStats() + assert.Equal(t, 1, stats.Synced, + "only the changed session may be rewritten") + assert.Zero(t, stats.Skipped, + "unchanged sessions must not even be materialized as sources") + assert.Zero(t, + parser.OpenCodeContainerChildScans()-scansBefore, + "a watcher event must not aggregate the whole container's "+ + "child tables") + lookups[n] = parser.OpenCodeSessionChildLookups() - lookupsBefore + processed[n] = stats.Synced + stats.Skipped + stats.Failed + assertMessageContent( + t, env.db, "opencode:ses00000", + "changed prompt", "changed answer", + ) + }) + } + + assert.Equal(t, lookups[20], lookups[200], + "per-session child lookups for one changed session must not grow "+ + "with container size") + assert.Equal(t, processed[20], processed[200], + "sources processed for one changed session must not grow with "+ + "container size") +} + +// TestOpenCodeWatcherPassDefersChildOnlyEditToFullDiscovery documents the +// staleness contract the watermark-only watcher pass trades on: a child-only +// write that leaves the session and project rows untouched is invisible to +// the session-row watermark — wherever its timestamps land relative to the +// stored composite — and stays archived as-is until the next full-discovery +// pass, whose child digest still reconciles it. Both variants are pinned +// here: a replacement below the stored composite and an append above it. +// Actively watched sessions do not rely on this path; the per-session +// watcher poll resolves the composite directly. +func TestOpenCodeWatcherPassDefersChildOnlyEditToFullDiscovery(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + // Session row far ahead of every child, so the replacement below stays + // under the stored composite. + seedOpenCodeSQLiteTextSession( + t, oc, "proj", "below-mark", + 1779012000000, 1779099999000, + "original prompt", "original answer", + ) + require.Equal(t, 1, env.engine.SyncAll(context.Background(), nil).Synced) + + // Child-only replacement: same counts, new rows and content, timestamps + // below the session row's watermark, session and project rows untouched. + oc.replaceTextContent( + t, "below-mark", "swapped prompt", "swapped answer", 1779012500000, + ) + + scansBefore := parser.OpenCodeContainerChildScans() + lookupsBefore := parser.OpenCodeSessionChildLookups() + require.NoError(t, env.engine.SyncPathsContext( + context.Background(), []string{oc.path}, + )) + assert.Zero(t, parser.OpenCodeContainerChildScans()-scansBefore, + "the watcher pass must not scan child tables for a child-only edit") + assert.Zero(t, parser.OpenCodeSessionChildLookups()-lookupsBefore, + "a child-only edit below the watermark yields no candidates") + assertMessageContent( + t, env.db, "opencode:below-mark", + "original prompt", "original answer", + ) + + fullStats := env.engine.SyncAll(context.Background(), nil) + assert.Equal(t, 1, fullStats.Synced, + "full discovery must reconcile the deferred child-only edit") + assertMessageContent( + t, env.db, "opencode:below-mark", + "swapped prompt", "swapped answer", + ) + + // Same deferral when the child write lands ABOVE the stored composite: + // a new message appended with a fresh timestamp while the session row + // stays untouched still cannot move the session-row watermark. + oc.addMessage( + t, "below-mark-msg-late", "below-mark", "assistant", 1779200000000, + ) + oc.addTextPart( + t, "below-mark-part-late", "below-mark", "below-mark-msg-late", + "late answer", 1779200000000, + ) + + scansBefore = parser.OpenCodeContainerChildScans() + lookupsBefore = parser.OpenCodeSessionChildLookups() + require.NoError(t, env.engine.SyncPathsContext( + context.Background(), []string{oc.path}, + )) + assert.Zero(t, parser.OpenCodeContainerChildScans()-scansBefore, + "the watcher pass must not scan child tables for an above-composite "+ + "child append") + assert.Zero(t, parser.OpenCodeSessionChildLookups()-lookupsBefore, + "an above-composite child-only append yields no candidates") + assertMessageContent( + t, env.db, "opencode:below-mark", + "swapped prompt", "swapped answer", + ) + + fullStats = env.engine.SyncAll(context.Background(), nil) + assert.Equal(t, 1, fullStats.Synced, + "full discovery must reconcile the deferred above-composite append") + assertMessageContent( + t, env.db, "opencode:below-mark", + "swapped prompt", "swapped answer", "late answer", + ) +} + +// TestOpenCodeFullPassSkipsAfterWatcherPassParse pins that a session parsed +// through the watermark-only watcher pass stores the full composite +// watermark and digest, not the cheap session-row watermark it was +// discovered with. The children deliberately end above the session row so +// the two values differ; if the cheap watermark leaked into the stored +// fingerprint, the next full pass would see a mismatch and re-parse an +// unchanged session with a fresh child lookup. +func TestOpenCodeFullPassSkipsAfterWatcherPassParse(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + for i := range 3 { + seedOpenCodeSQLiteTextSession( + t, oc, "proj", fmt.Sprintf("ses%05d", i), + 1779012000000, 1779012030000, + "prompt", "answer", + ) + } + require.Equal(t, 3, env.engine.SyncAll(context.Background(), nil).Synced) + + oc.updateSessionTime(t, "ses00000", 1779015630000) + oc.replaceTextContent( + t, "ses00000", "changed prompt", "changed answer", 1779015600000, + ) + oc.mustExec(t, "raise children above the session row", + "UPDATE part SET time_updated = ? WHERE session_id = ?", + 1779099999000, "ses00000") + + require.NoError(t, env.engine.SyncPathsContext( + context.Background(), []string{oc.path}, + )) + require.Equal(t, 1, env.engine.LastSyncStats().Synced) + + lookupsBefore := parser.OpenCodeSessionChildLookups() + stats := env.engine.SyncAll(context.Background(), nil) + assert.Zero(t, stats.Synced, + "the full pass must not rewrite sessions the watcher pass stored") + assert.Equal(t, 3, stats.Skipped) + assert.Zero(t, parser.OpenCodeSessionChildLookups()-lookupsBefore, + "full-pass skips must not pay per-session child lookups") +} + +// TestOpenCodeWatcherCatchesMetadataUpdateUnderChildDominatedComposite pins +// the like-for-like watermark comparison. The stored composite is a MAX over +// session, project, and child times, so when a child timestamp dominates it, +// a later metadata update (title, session/project time) can advance the +// session row while staying below the composite. Comparing the session-row +// watermark against the composite would wrongly skip that session on the +// watcher pass; comparing against the stored session/project metadata +// watermark recovered from the persisted digest catches it — still without +// touching the container's child tables. +func TestOpenCodeWatcherCatchesMetadataUpdateUnderChildDominatedComposite( + t *testing.T, +) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + seedOpenCodeSQLiteTextSession( + t, oc, "proj", "meta-mark", + 1779012000000, 1779012030000, + "prompt", "answer", + ) + // Children exceed both the previous and the soon-to-advance metadata + // timestamps, so the stored composite is child-dominated. + oc.mustExec(t, "raise children above all metadata times", + "UPDATE part SET time_updated = ? WHERE session_id = ?", + 1779099999000, "meta-mark") + require.Equal(t, 1, env.engine.SyncAll(context.Background(), nil).Synced) + + // Metadata advances past its own stored value but stays below the + // child-dominated composite. + oc.mustExec(t, "retitle session below the composite", + "UPDATE session SET title = ?, time_updated = ? WHERE id = ?", + "renamed by watcher", 1779012040000, "meta-mark") + + scansBefore := parser.OpenCodeContainerChildScans() + require.NoError(t, env.engine.SyncPathsContext( + context.Background(), []string{oc.path}, + )) + stats := env.engine.LastSyncStats() + assert.Equal(t, 1, stats.Synced, + "a metadata update below the child-dominated composite must "+ + "re-parse on the watcher pass") + assert.Zero(t, parser.OpenCodeContainerChildScans()-scansBefore, + "the watcher pass must still not scan the container's child tables") + + // OpenCode's LLM-generated title lands in first_message. + var firstMessage string + require.NoError(t, env.db.Reader().QueryRow( + "SELECT first_message FROM sessions WHERE id = ?", + "opencode:meta-mark", + ).Scan(&firstMessage)) + assert.Equal(t, "renamed by watcher", firstMessage, + "the watcher pass must archive the metadata update") +} + +// TestOpenCodeIdleReconcilePassSkipsContainerChildScan pins the same +// trusted-container bound on the streamed reconciliation path: an idle +// ReconcileWatchRoots pass over a trusted, untouched container must not +// aggregate the child tables (its candidates all gate-skip), while any +// write breaks trust and the next reconcile carries the full digest again — +// including for a child-only edit below every watermark. +func TestOpenCodeIdleReconcilePassSkipsContainerChildScan(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + for i := range 5 { + seedOpenCodeSQLiteTextSession( + t, oc, "proj", fmt.Sprintf("ses%05d", i), + 1779012000000, 1779099999000, + "prompt", "answer", + ) + } + require.Equal(t, 5, env.engine.SyncAll(context.Background(), nil).Synced) + + scansBefore := parser.OpenCodeContainerChildScans() + lookupsBefore := parser.OpenCodeSessionChildLookups() + require.NoError(t, env.engine.ReconcileWatchRoots( + context.Background(), []string{env.opencodeDir}, false, + )) + assert.Zero(t, parser.OpenCodeContainerChildScans()-scansBefore, + "an idle reconcile pass must not aggregate the container's child "+ + "tables") + assert.Zero(t, parser.OpenCodeSessionChildLookups()-lookupsBefore, + "an idle reconcile pass must not pay per-session child lookups") + assertMessageContent( + t, env.db, "opencode:ses00000", "prompt", "answer", + ) + + // A child-only replacement below every watermark breaks trust via the + // container state, and the next reconcile carries the digest again. + oc.replaceTextContent( + t, "ses00000", "swapped prompt", "swapped answer", 1779012500000, + ) + require.NoError(t, env.engine.ReconcileWatchRoots( + context.Background(), []string{env.opencodeDir}, false, + )) + assertMessageContent( + t, env.db, "opencode:ses00000", + "swapped prompt", "swapped answer", + ) +} + +// TestOpenCodeIdleFullPassSkipsContainerChildScan pins that a periodic full +// pass over a trusted, untouched container does not aggregate the child +// tables at all: the container gate will skip every member before +// fingerprinting, so discovery lists the bounded watermark form instead of +// computing archive-sized child identities nothing reads. Any write breaks +// container trust, and the next full pass carries the complete digest again +// — including for child-only edits below every watermark. +func TestOpenCodeIdleFullPassSkipsContainerChildScan(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + for i := range 5 { + // Session rows far ahead of every child, so the later child-only + // replacement stays below the stored composite. + seedOpenCodeSQLiteTextSession( + t, oc, "proj", fmt.Sprintf("ses%05d", i), + 1779012000000, 1779099999000, + "prompt", "answer", + ) + } + require.Equal(t, 5, env.engine.SyncAll(context.Background(), nil).Synced) + + scansBefore := parser.OpenCodeContainerChildScans() + lookupsBefore := parser.OpenCodeSessionChildLookups() + stats := env.engine.SyncAll(context.Background(), nil) + assert.Zero(t, stats.Synced) + assert.Equal(t, 5, stats.Skipped, + "every session of a trusted container must gate-skip") + assert.Zero(t, parser.OpenCodeContainerChildScans()-scansBefore, + "an idle full pass must not aggregate the container's child tables") + assert.Zero(t, parser.OpenCodeSessionChildLookups()-lookupsBefore, + "an idle full pass must not pay per-session child lookups") + + // A child-only replacement below every watermark breaks trust via the + // container state, and the next full pass carries the digest again. + oc.replaceTextContent( + t, "ses00000", "swapped prompt", "swapped answer", 1779012500000, + ) + stats = env.engine.SyncAll(context.Background(), nil) + assert.Equal(t, 1, stats.Synced, + "a write breaks container trust and full discovery reconciles it") + assertMessageContent( + t, env.db, "opencode:ses00000", + "swapped prompt", "swapped answer", + ) +} + +// TestOpenCodeDeletedChildIsDetected pins deletion sensitivity. The composite +// mtime is a MAX over session/project/child timestamps, so when the session or +// project row already holds the higher value — the common case on a real +// container — deleting a message or part does not move the max at all. Without +// a deletion-sensitive component the session looks fresh and the removed +// content stays archived indefinitely. +func TestOpenCodeDeletedChildIsDetected(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + // Session row timestamp is deliberately far ahead of every child, so a + // deleted child cannot lower the composite. + seedOpenCodeSQLiteTextSession( + t, oc, "proj", "del-session", + 1779012000000, 1779099999000, + "keep prompt", "drop answer", + ) + + stats := env.engine.SyncAll(context.Background(), nil) + require.False(t, stats.Aborted) + require.Equal(t, 1, stats.Synced) + assertMessageContent( + t, env.db, "opencode:del-session", "keep prompt", "drop answer", + ) + + // Remove the assistant message and its parts, leaving session and project + // timestamps untouched. + oc.mustExec(t, "delete assistant parts", + "DELETE FROM part WHERE session_id = ? AND message_id LIKE ?", + "del-session", "%assistant%") + oc.mustExec(t, "delete assistant message", + "DELETE FROM message WHERE session_id = ? AND id LIKE ?", + "del-session", "%assistant%") + + stats = env.engine.SyncAll(context.Background(), nil) + require.False(t, stats.Aborted) + assert.Equal(t, 1, stats.Synced, + "a deleted child must not be hidden behind an unchanged composite max") +} + +// TestOpenCodeDeletedChildDetectedViaReconciliation covers the same deletion +// hole on the reconciliation path. Sources rebuilt by FindSource rather than +// carried from discovery metadata have no child digest, so the fingerprint hash +// is empty and the freshness gate treats it as no constraint. +func TestOpenCodeDeletedChildDetectedViaReconciliation(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + seedOpenCodeSQLiteTextSession( + t, oc, "proj", "recon-del", + 1779012000000, 1779099999000, + "keep prompt", "drop answer", + ) + require.Equal(t, 1, env.engine.SyncAll(context.Background(), nil).Synced) + + oc.mustExec(t, "delete assistant parts", + "DELETE FROM part WHERE session_id = ? AND message_id LIKE ?", + "recon-del", "%assistant%") + oc.mustExec(t, "delete assistant message", + "DELETE FROM message WHERE session_id = ? AND id LIKE ?", + "recon-del", "%assistant%") + + require.NoError(t, env.engine.ReconcileWatchRoots( + context.Background(), []string{env.opencodeDir}, false, + )) + env.engine.SyncAll(context.Background(), nil) + + // Assert the observable outcome rather than which pass did the write: + // the removed assistant turn must no longer be archived. + for _, m := range fetchMessages(t, env.db, "opencode:recon-del") { + assert.NotContains(t, m.Content, "drop answer", + "deleted child content must not remain archived") + } +} + +// TestOpenCodeSameCountChildReplacementIsDetected covers a replacement that +// preserves both child counts and leaves every new timestamp below the session +// row's already-higher watermark, so neither the watermark nor the counts move. +func TestOpenCodeSameCountChildReplacementIsDetected(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + seedOpenCodeSQLiteTextSession( + t, oc, "proj", "swap-session", + 1779012000000, 1779099999000, + "original prompt", "original answer", + ) + require.Equal(t, 1, env.engine.SyncAll(context.Background(), nil).Synced) + + // Same number of messages and parts, timestamps still below the session + // row's watermark, but different rows and different content. + oc.replaceTextContent( + t, "swap-session", "swapped prompt", "swapped answer", 1779012500000, + ) + + stats := env.engine.SyncAll(context.Background(), nil) + assert.Equal(t, 1, stats.Synced, + "a same-count child replacement below the session watermark must "+ + "still change the fingerprint") +} + +// TestOpenCodeMetadataUpdateBelowWatermarkIsDetected covers a project worktree +// rename whose timestamp lands below an already-higher child watermark. The +// composite MAX cannot move in that case, so the digest has to carry the +// session and project timestamps in their own right. +func TestOpenCodeMetadataUpdateBelowWatermarkIsDetected(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/original-app") + // Children hold the highest timestamp, so a later project rename below + // that value leaves MAX(...) unchanged. + seedOpenCodeSQLiteTextSession( + t, oc, "proj", "below-watermark", + 1779012000000, 1779012030000, + "stable prompt", "stable answer", + ) + oc.mustExec(t, "raise child watermark", + "UPDATE part SET time_updated = ? WHERE session_id = ?", + 1779099999000, "below-watermark") + require.Equal(t, 1, env.engine.SyncAll(context.Background(), nil).Synced) + + // Rename below the child watermark. + oc.updateProjectWorktree( + t, "proj", "/home/user/code/renamed-app", 1779013000000, + ) + + stats := env.engine.SyncAll(context.Background(), nil) + assert.Equal(t, 1, stats.Synced, + "a metadata update below the child watermark must still be detected") +} + +// TestOpenCodeMiddleRowReplacementIsDetected covers a replacement that keeps +// every aggregate the digest currently reduces to: same counts, same timestamp +// sums, and the same min/max ids because the swapped row sorts strictly between +// the extrema. Only a complete child identity can tell these apart. +func TestOpenCodeMiddleRowReplacementIsDetected(t *testing.T) { + env := setupSingleAgentTestEnv(t, parser.AgentOpenCode) + oc := createOpenCodeDB(t, env.opencodeDir) + oc.addProject(t, "proj", "/home/user/code/app") + oc.addSession(t, "mid", "proj", 1779012000000, 1779099999000) + oc.addMessage(t, "mid-msg-a", "mid", "user", 1779012000000) + // Three parts: a, m, z. The middle one gets swapped for a different id + // carrying an identical timestamp, so count, sum and extrema all hold. + oc.addTextPart(t, "mid-part-a", "mid", "mid-msg-a", "alpha", 1779012000000) + oc.addTextPart(t, "mid-part-m", "mid", "mid-msg-a", "middle", 1779012000001) + oc.addTextPart(t, "mid-part-z", "mid", "mid-msg-a", "zulu", 1779012000002) + require.Equal(t, 1, env.engine.SyncAll(context.Background(), nil).Synced) + + oc.mustExec(t, "delete middle part", + "DELETE FROM part WHERE id = ?", "mid-part-m") + oc.addTextPart( + t, "mid-part-n", "mid", "mid-msg-a", "replaced", 1779012000001, + ) + + stats := env.engine.SyncAll(context.Background(), nil) + assert.Equal(t, 1, stats.Synced, + "a middle-row replacement preserving counts, sums and extrema must "+ + "still change the fingerprint") +} diff --git a/internal/sync/perf_invariant_test.go b/internal/sync/perf_invariant_test.go index 1e8760a44..a4de5d11a 100644 --- a/internal/sync/perf_invariant_test.go +++ b/internal/sync/perf_invariant_test.go @@ -385,3 +385,83 @@ func TestRebuildLocalAndRemoteContributorsBulkWriteDiscoveredCount(t *testing.T) assert.Equal(t, large.localWrites, large.remoteWrites, "large equivalent contributors must retain equal work counts") } + +// TestWarmFullSyncDoesNotRehashClaudeArchive pins the "background sync work is +// bounded by the changed batch, not total archive size" rule against the Claude +// DB-freshness gate. +// +// providerSingleSessionFresh runs at the top of every provider process pass. Its +// last guard, providerIncrementalContentChanged, defends against a same-size, +// same-mtime, same-inode in-place rewrite by content-hashing the source, and +// that guard reads the whole stored prefix. Today the verified-source gate +// (verifiedProviderSourceState) absorbs it: a signature is content-verified once +// to earn trust, and every later pass over an unchanged source rides the trusted +// skip without reading bytes. +// +// Nothing else pins that. Losing the trusted skip — by dropping Claude's +// VerifiedLocalStat capability, widening the signature so it never repeats, or +// pruning trust between passes — would silently turn every watcher-triggered +// pass into a full re-read of the entire Claude archive, which is invisible in +// correctness tests and only shows up as daemon CPU on a large archive. Assert +// the steady-state pass reads no content and does not scale with archive size. +func TestWarmFullSyncDoesNotRehashClaudeArchive(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test") + } + + observed := make(map[int]int) + for _, claudeCount := range []int{5, 50} { + t.Run(fmt.Sprintf("claude_%d", claudeCount), func(t *testing.T) { + claudeDir := filepath.Join(t.TempDir(), "claude") + require.NoError(t, os.MkdirAll(claudeDir, 0o755)) + writeClaudeCorpus(t, claudeDir, claudeCount) + + database := openTestDB(t) + engine := NewEngine(database, EngineConfig{ + AgentDirs: map[parser.AgentType][]string{ + parser.AgentClaude: {claudeDir}, + }, + Machine: "local", + }) + t.Cleanup(engine.Close) + + // Cold pass: populates the archive and earns verified-source trust. + require.Equal(t, claudeCount, + engine.SyncAll(context.Background(), nil).Synced) + + var mu gosync.Mutex + reads := 0 + restore := computeFileHashPrefix + computeFileHashPrefix = func(path string, size int64) (string, error) { + mu.Lock() + reads++ + mu.Unlock() + return restore(path, size) + } + t.Cleanup(func() { computeFileHashPrefix = restore }) + + // Each pass leaves the archive untouched on disk. The first warm + // pass may content-verify once to earn verified-source trust; every + // pass after that must ride the trusted-signature skip. + pass := func() int { + mu.Lock() + reads = 0 + mu.Unlock() + engine.SyncAll(context.Background(), nil) + mu.Lock() + defer mu.Unlock() + return reads + } + trustEarning := pass() + steady := pass() + t.Logf("source content reads: trust-earning pass=%d steady pass=%d", + trustEarning, steady) + observed[claudeCount] = steady + }) + } + + assert.Equal(t, observed[5], observed[50], + "warm-pass source reads must not grow with archive size") + assert.Zero(t, observed[50], + "a warm pass over an unchanged archive must not re-read source content") +} diff --git a/internal/sync/test_helpers_test.go b/internal/sync/test_helpers_test.go index dd687a2f4..fa9067b0f 100644 --- a/internal/sync/test_helpers_test.go +++ b/internal/sync/test_helpers_test.go @@ -202,10 +202,16 @@ var ( kiroSQLiteFixtureCache stdsync.Map ) +// openCodeLikeSchema mirrors the production OpenCode container schema, +// including the project/message/part time_updated columns. Those columns are +// the per-session change signal (openCodeCompositeMtimeExpr); a fixture that +// omitted them could not model shared-container freshness at all, which is how +// the whole-container re-parse regression went unnoticed. const openCodeLikeSchema = ` CREATE TABLE project ( id TEXT PRIMARY KEY, - worktree TEXT NOT NULL + worktree TEXT NOT NULL, + time_updated INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE session ( id TEXT PRIMARY KEY, @@ -219,15 +225,21 @@ const openCodeLikeSchema = ` id TEXT PRIMARY KEY, session_id TEXT NOT NULL, data TEXT NOT NULL, - time_created INTEGER NOT NULL + time_created INTEGER NOT NULL, + time_updated INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE part ( id TEXT PRIMARY KEY, session_id TEXT NOT NULL, message_id TEXT NOT NULL, data TEXT NOT NULL, - time_created INTEGER NOT NULL + time_created INTEGER NOT NULL, + time_updated INTEGER NOT NULL DEFAULT 0 ); + CREATE INDEX message_session_time_created_id_idx + ON message (session_id, time_created, id); + CREATE INDEX part_session_idx ON part (session_id); + CREATE INDEX part_message_id_id_idx ON part (message_id, id); ` const kiroSQLiteSchema = ` @@ -430,8 +442,22 @@ func (oc *openCodeTestDB) addProject( ) { t.Helper() oc.mustExec(t, "insert project", - "INSERT INTO project (id, worktree) VALUES (?, ?)", - id, worktree, + "INSERT INTO project (id, worktree, time_updated) VALUES (?, ?, ?)", + id, worktree, 0, + ) +} + +// updateProjectWorktree renames a project's worktree and bumps its +// time_updated, matching production OpenCode. The bump is what lets every +// session in that project re-resolve its cwd/project without the container +// stat acting as a blunt whole-archive invalidator. +func (oc *openCodeTestDB) updateProjectWorktree( + t *testing.T, id, worktree string, timeUpdated int64, +) { + t.Helper() + oc.mustExec(t, "update project worktree", + "UPDATE project SET worktree = ?, time_updated = ? WHERE id = ?", + worktree, timeUpdated, id, ) } @@ -471,9 +497,9 @@ func (oc *openCodeTestDB) addMessage( require.NoError(t, err, "marshal message") oc.mustExec(t, "insert message", `INSERT INTO message - (id, session_id, data, time_created) - VALUES (?, ?, ?, ?)`, - id, sessionID, string(data), timeCreated, + (id, session_id, data, time_created, time_updated) + VALUES (?, ?, ?, ?, ?)`, + id, sessionID, string(data), timeCreated, timeCreated, ) } @@ -483,8 +509,11 @@ func (oc *openCodeTestDB) updateMessageData( t.Helper() raw, err := json.Marshal(data) require.NoError(t, err, "marshal message update") + // Production OpenCode bumps time_updated on an in-place row edit; the + // per-session composite freshness signal depends on it. oc.mustExec(t, "update message data", - "UPDATE message SET data = ? WHERE id = ?", + `UPDATE message SET data = ?, time_updated = time_updated + 1 + WHERE id = ?`, string(raw), id, ) } @@ -502,9 +531,9 @@ func (oc *openCodeTestDB) addTextPart( require.NoError(t, err, "marshal text part") oc.mustExec(t, "insert part", `INSERT INTO part - (id, session_id, message_id, data, time_created) - VALUES (?, ?, ?, ?, ?)`, - id, sessionID, messageID, string(data), timeCreated, + (id, session_id, message_id, data, time_created, time_updated) + VALUES (?, ?, ?, ?, ?, ?)`, + id, sessionID, messageID, string(data), timeCreated, timeCreated, ) } @@ -523,9 +552,9 @@ func (oc *openCodeTestDB) addToolPart( require.NoError(t, err, "marshal tool part") oc.mustExec(t, "insert tool part", `INSERT INTO part - (id, session_id, message_id, data, time_created) - VALUES (?, ?, ?, ?, ?)`, - id, sessionID, messageID, string(data), timeCreated, + (id, session_id, message_id, data, time_created, time_updated) + VALUES (?, ?, ?, ?, ?, ?)`, + id, sessionID, messageID, string(data), timeCreated, timeCreated, ) }