Files
wehub-resource-sync f99010fae1
Desktop Artifacts / Desktop Build (Linux) (push) Waiting to run
Desktop Artifacts / Desktop Build (Windows) (push) Waiting to run
Desktop Artifacts / Desktop Build (Linux (arm64)) (push) Waiting to run
Desktop Artifacts (macOS) / Desktop Build (macOS (aarch64)) (push) Waiting to run
Desktop Artifacts (macOS) / Desktop Build (macOS (x86_64)) (push) Waiting to run
CI / lint (push) Failing after 1s
CI / frontend (push) Failing after 1s
CI / scripts (push) Failing after 1s
CI / Go Test (ubuntu-latest) (push) Failing after 0s
CI / frontend-node-25 (push) Failing after 1s
CI / docs (push) Failing after 0s
CI / coverage (push) Failing after 0s
CI / e2e (push) Failing after 0s
Docker / build-and-push (push) Failing after 1s
CI / integration (push) Failing after 4m43s
CI / Go Test (windows-latest) (push) Has been cancelled
CI / Desktop Unit Tests (Windows) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:30:36 +08:00

10093 lines
305 KiB
Go

package sync
import (
"context"
"errors"
"fmt"
"log"
"maps"
"os"
"path/filepath"
"runtime"
"slices"
"strconv"
"strings"
gosync "sync"
"time"
"go.kenn.io/agentsview/internal/db"
"go.kenn.io/agentsview/internal/export"
"go.kenn.io/agentsview/internal/parser"
"go.kenn.io/agentsview/internal/secrets"
"go.kenn.io/agentsview/internal/signals"
"go.kenn.io/agentsview/internal/timeutil"
)
const (
batchSize = 100
maxWorkers = 8
projectIdentityCacheTTL = time.Minute
)
type syncWriteMode int
const (
syncWriteDefault syncWriteMode = iota
syncWriteBulk
)
var errSessionPreserved = errors.New("session preserved")
func isIntentionalSessionSkip(err error) bool {
return errors.Is(err, db.ErrSessionExcluded) ||
errors.Is(err, db.ErrSessionTrashed)
}
// Emitter is notified after a sync pass writes data. Implementations
// must be thread-safe; Emit is called from whatever goroutine runs
// the sync pass (e.g., the file watcher, a periodic timer, or a
// handler goroutine triggered by POST /api/v1/sync).
//
// Emit must not block. A slow implementation can delay the sync
// pipeline; see server.Broadcaster for the production implementation,
// which drops events on full per-subscriber buffers.
type Emitter interface {
Emit(scope string)
}
// EngineConfig holds the configuration needed by the sync
// engine, replacing per-agent positional parameters.
type EngineConfig struct {
AgentDirs map[parser.AgentType][]string
Machine string
BlockedResultCategories []string
// IncludeCwdPrefixes, when non-empty, restricts ingestion to
// sessions whose working directory equals one of the prefixes
// or lives underneath one. Sessions without a recorded cwd are
// skipped while the filter is active. Populated from the
// sync_include_cwd_prefixes config option for local sync;
// remote sync leaves it empty because the prefixes describe
// local paths.
IncludeCwdPrefixes []string
// IDPrefix is prepended to all session IDs. Used by
// remote sync to namespace IDs by host (e.g. "host~").
IDPrefix string
// PathRewriter transforms file paths before storage.
// Used by remote sync to replace temp paths with
// "host:/remote/path" references.
PathRewriter func(string) string
// Ephemeral disables sync-state persistence (timestamps
// and skip cache) so remote sync does not interfere with
// local sync watermarks or pollute the skipped_files table
// with temp-dir paths.
Ephemeral bool
// Emitter, when non-nil, is called once after each sync pass
// that wrote data. Safe to leave nil (e.g., in PG serve mode
// where the engine is not run).
Emitter Emitter
// DeferStartupMaintenance keeps startup backfills blocked until the
// foreground sync that launched the daemon has completed. Maintenance
// still takes syncMu after it is released so later syncs and resyncs
// cannot overlap its database access.
DeferStartupMaintenance bool
// ProviderFactories and ProviderMigrationModes select which concrete
// providers own discovery and parsing for their agents. Nil uses the
// parser package registry/manifest.
ProviderFactories []parser.ProviderFactory
ProviderMigrationModes map[parser.AgentType]parser.ProviderMigrationMode
}
// Engine orchestrates session file discovery and sync.
type Engine struct {
db *db.DB
openCodeArchiveStore db.Store
agentDirs map[parser.AgentType][]string
machine string
blockedResultCategories map[string]bool
cwdFilter cwdPrefixFilter
syncMu gosync.Mutex // serializes all sync operations
mu gosync.RWMutex
lastSync time.Time
lastSyncStats SyncStats
currentProgress *Progress
// skipCache tracks paths that should be skipped on
// subsequent syncs, keyed by path with the file mtime
// at time of caching. Covers parse errors and
// non-interactive sessions (nil result). The file is
// retried when its mtime changes. S3 entries also keep an
// in-memory source fingerprint when one is available.
skipMu gosync.RWMutex
skipCache map[string]int64
skipFingerprints map[string]string
// skipHashKeys maps a source base path to its one current
// ?source_hash= cache key. It is built once when the cache loads so a
// watcher mutation never scans unrelated archive entries.
skipHashKeys map[string]string
s3CodexIndexMu gosync.Mutex
s3CodexIndexCache map[string]s3CodexIndexSnapshot
// idPrefix and pathRewriter support remote sync:
// prefix all session IDs to avoid collisions, rewrite
// temp paths to "host:/remote/path" form.
ephemeral bool
idPrefix string
pathRewriter func(string) string
emitter Emitter
providerFactories map[parser.AgentType]parser.ProviderFactory
providerMigrationModes map[parser.AgentType]parser.ProviderMigrationMode
projectIdentityMu gosync.Mutex
projectIdentityCache map[string]projectIdentityCacheEntry
projectIdentityWritten map[string]struct{}
startupMaintenanceOnce gosync.Once
startupMaintenanceReady chan struct{}
// forceParse disables every stored-state skip (skip cache,
// size/mtime/data_version checks, incremental JSONL deltas) so
// parse-diff fully re-parses every discovered file. Normal sync
// never sets it; behavior must be identical when false.
forceParse bool
// phaseStats accumulates per-phase wall-clock time inside the bulk
// write path. Exposed via PhaseStats() so a CLI driver can log the
// totals after a sync pass completes.
phaseStats PhaseStats
// anomalies accumulates per-run parser/sanitizer anomaly signals
// recorded at the write seam (prepareSessionWrite, writeIncremental,
// toDBUsageEvents). Reset at the start of each sync run and folded
// into the returned SyncStats before the run completes.
anomalies anomalyAccumulator
// signalSched debounces the O(session history) signal/secret
// recompute triggered by incremental writes, so streaming
// sessions don't rescan their whole history on every appended
// line. Close flushes and stops it.
signalSched *signalScheduler
// containerMu guards the OpenCode-family shared-SQLite freshness
// gate (see opencode_container_gate.go). trustedSQLiteContainers
// maps a container DB path to its state and verified session-ID
// set at the end of the last pass that verified every one of its
// discovered sessions; containerPass is the bookkeeping for the
// pass currently running (nil outside passes). Both are in-memory
// only: a restart re-verifies once.
containerMu gosync.Mutex
trustedSQLiteContainers map[string]trustedSQLiteContainer
containerPass *sqliteContainerPass
// storageTrustMu guards the per-session freshness gate for
// OpenCode-family file-backed storage sessions (see
// opencode_storage_gate.go). trustedStorageSessions maps a session
// JSON path to the stat signature captured before the last parse
// whose outcome the archive absorbed (results dropped as already
// stored, or confirmed written). storageTrustGens counts each
// session's invalidations and storageTrustEpoch counts full clears,
// so a promotion whose pre-parse snapshot predates an invalidation
// is discarded instead of resurrecting the invalidated trust. All
// in-memory only: a restart re-verifies once.
storageTrustMu gosync.Mutex
trustedStorageSessions map[string]string
storageTrustGens map[string]uint64
storageTrustEpoch uint64
// verifiedSourceMu guards the local source stat/ctime trust gate (see
// verified_source_gate.go). Each path has one compact record containing
// its trusted signature, invalidation generation, and last full pass seen.
// The epoch vetoes promotions captured before a global clear. State is
// memory-only, so process startup always deep-verifies sources once.
verifiedSourceMu gosync.Mutex
verifiedSources map[string]verifiedSourceRecord
verifiedSourceEpoch uint64
verifiedSourcePass uint64
verifiedSourceActivePass uint64
}
// PhaseStats returns the engine's phase counter. The values reflect only
// the most recent sync pass; callers should read after SyncAll/ResyncAll
// returns.
func (e *Engine) PhaseStats() *PhaseStats { return &e.phaseStats }
// refuseWriteInForceParse guards the public sync entrypoints against an
// engine created by NewDiffEngine, whose forceParse mode exists purely
// for report-only re-parsing. Such an engine is also Ephemeral, so a
// write would persist nothing useful, but it would still rewrite or
// re-derive archive rows -- exactly what the report-only contract
// promises not to do. Rather than widen the read-only surface into a
// separate interface (which would change NewDiffEngine's return type and
// break ParseDiff callers), the write entrypoints refuse and log when
// forceParse is set. A real sync engine never sets forceParse, so this
// is a no-op for every production caller.
//
// It returns true when the caller must abort. op names the refused
// entrypoint for the log line.
func (e *Engine) refuseWriteInForceParse(op string) bool {
if !e.forceParse {
return false
}
log.Printf(
"sync: refusing %s on a report-only (parse-diff) engine; "+
"forceParse engines never write", op,
)
return true
}
// codexExecMigrationKey is the pg_sync_state flag that
// records whether the one-time cleanup of legacy codex_exec
// skip cache entries has already run on this database.
const codexExecMigrationKey = "codex_exec_legacy_migration_v1"
// visualStudioCopilotSkipMigrationKey is the pg_sync_state flag
// that records whether the one-time cleanup of Visual Studio
// Copilot skip cache entries has already run on this database.
// Older builds cached trace read/scan errors keyed by an
// unchanged mtime, which would otherwise suppress retries after
// upgrading to the non-cacheable read-error behavior.
const visualStudioCopilotSkipMigrationKey = "visualstudio_copilot_skip_migration_v1"
// NewEngine creates a sync engine. It pre-populates the
// in-memory skip cache from the database so that files
// skipped in a prior run are not re-parsed on startup, and
// migrates legacy codex_exec skip entries on first run under
// the new bulk-sync behavior.
func NewEngine(
database *db.DB, cfg EngineConfig,
) *Engine {
skipCache := make(map[string]int64)
if !cfg.Ephemeral {
if loaded, err := database.LoadSkippedFiles(); err == nil {
skipCache = loaded
} else {
log.Printf("loading skip cache: %v", err)
}
migrateLegacyCodexExecSkips(database, skipCache)
migrateVisualStudioCopilotSkips(database, skipCache)
}
skipHashKeys, _ := normalizeSourceHashSkipCache(skipCache, nil)
dirs := make(map[parser.AgentType][]string, len(cfg.AgentDirs))
for k, v := range cfg.AgentDirs {
dirs[k] = append([]string(nil), v...)
}
providerFactories := parser.ProviderFactories()
if cfg.ProviderFactories != nil {
providerFactories = cfg.ProviderFactories
}
providerModes := parser.ProviderMigrationModes()
if cfg.ProviderMigrationModes != nil {
maps.Copy(providerModes, cfg.ProviderMigrationModes)
}
e := &Engine{
db: database,
agentDirs: dirs,
machine: cfg.Machine,
blockedResultCategories: blockedCategorySet(cfg.BlockedResultCategories),
cwdFilter: newCwdPrefixFilter(cfg.IncludeCwdPrefixes),
skipCache: skipCache,
skipFingerprints: make(map[string]string),
skipHashKeys: skipHashKeys,
s3CodexIndexCache: make(map[string]s3CodexIndexSnapshot),
ephemeral: cfg.Ephemeral,
idPrefix: cfg.IDPrefix,
pathRewriter: cfg.PathRewriter,
emitter: cfg.Emitter,
providerFactories: providerFactoryMap(providerFactories),
providerMigrationModes: providerModes,
projectIdentityCache: make(map[string]projectIdentityCacheEntry),
projectIdentityWritten: make(map[string]struct{}),
startupMaintenanceReady: make(chan struct{}),
}
if !cfg.DeferStartupMaintenance {
e.ReleaseStartupMaintenance()
}
// Errors are logged inside recomputeSignalsFromDB and are
// non-fatal: the next write or flush retries.
recompute := func(sessionID string) {
_, _ = e.recomputeSignalsFromDB(
context.Background(), sessionID,
)
}
e.signalSched = newSignalScheduler(
signalRecomputeInterval, signalRecomputeQuiet,
// Inline runs happen from markDirty inside writeIncremental,
// whose callers already hold syncMu.
recompute,
// Timer and flush passes happen outside any sync operation,
// so they take syncMu around the whole claim-and-recompute
// pass: otherwise a delayed recompute could read an older
// message snapshot and overwrite signals just written by a
// concurrent sync, or claim a session and block while a
// locked pre-push flush finds nothing left to recompute.
func(flush func()) {
e.syncMu.Lock()
defer e.syncMu.Unlock()
flush()
},
)
return e
}
// Close flushes any pending debounced signal recomputes and stops
// the scheduler. Call once when the engine's owner shuts down;
// safe to call repeatedly.
func (e *Engine) Close() {
e.signalSched.stop()
}
// FlushSignals immediately recomputes signals for sessions with a
// pending debounced recompute, leaving the scheduler running. Push
// paths that read SQLite rows outside a sync operation call it
// first so pushed sessions carry current signal fields. Callers
// must not hold syncMu; work running inside SyncThenRun is flushed
// by the engine instead.
func (e *Engine) FlushSignals() {
e.signalSched.flushAll()
}
func providerFactoryMap(
factories []parser.ProviderFactory,
) map[parser.AgentType]parser.ProviderFactory {
out := make(map[parser.AgentType]parser.ProviderFactory, len(factories))
for _, factory := range factories {
def := factory.Definition()
out[def.Type] = factory
}
return out
}
// migrateLegacyCodexExecSkips removes skip cache entries
// created by older agentsview builds that excluded Codex exec
// sessions from bulk sync. The scrub runs once per database:
// a `pg_sync_state` flag is set after the first successful
// pass so subsequent process starts do not re-scan files.
// New skip entries for real parse errors on exec files are
// untouched here and honored normally on later syncs.
//
// The cleanup builds a rebuilt snapshot and writes it through
// the atomic ReplaceSkippedFiles, then only mutates the
// in-memory map and records the done flag after the persist
// succeeds. A partial failure leaves both the DB and the
// in-memory cache in their prior state so the migration is
// retried on the next startup rather than being falsely
// marked complete.
func migrateLegacyCodexExecSkips(
database *db.DB, skipCache map[string]int64,
) {
done, err := database.GetSyncState(codexExecMigrationKey)
if err != nil {
log.Printf("codex exec migration: %v", err)
return
}
if done != "" {
return
}
cleaned := make(map[string]int64, len(skipCache))
var legacy []string
for path, mtime := range skipCache {
if strings.HasSuffix(path, ".jsonl") &&
parser.IsCodexExecSessionFile(path) {
legacy = append(legacy, path)
continue
}
cleaned[path] = mtime
}
if len(legacy) > 0 {
if err := database.ReplaceSkippedFiles(
cleaned,
); err != nil {
log.Printf(
"codex exec migration: persist cleaned skip cache: %v",
err,
)
return
}
for _, p := range legacy {
delete(skipCache, p)
}
log.Printf(
"codex exec legacy migration: cleared %d skip entries",
len(legacy),
)
}
if err := database.SetSyncState(
codexExecMigrationKey, "done",
); err != nil {
log.Printf(
"codex exec migration: set flag: %v", err,
)
}
}
// migrateVisualStudioCopilotSkips removes skip cache entries for
// Visual Studio Copilot trace files. Older builds cached trace
// read/scan errors keyed by mtime, so an unchanged unreadable
// file would be skipped on later syncs instead of retried. The
// scrub clears both physical trace paths and
// <traceFile>#<conversationID> virtual paths; successfully synced
// conversations are re-cached on the next sync, while read errors
// surface again because they are no longer cacheable.
//
// The scrub runs once per database: a pg_sync_state flag is set
// after the first successful pass. It mirrors
// migrateLegacyCodexExecSkips: the cleaned snapshot is persisted
// through the atomic ReplaceSkippedFiles before the in-memory map
// and done flag are updated, so a partial failure is retried on
// the next startup rather than being falsely marked complete.
func migrateVisualStudioCopilotSkips(
database *db.DB, skipCache map[string]int64,
) {
done, err := database.GetSyncState(visualStudioCopilotSkipMigrationKey)
if err != nil {
log.Printf("visual studio copilot skip migration: %v", err)
return
}
if done != "" {
return
}
cleaned := make(map[string]int64, len(skipCache))
var stale []string
for path, mtime := range skipCache {
if IsVisualStudioCopilotSkipPath(path) {
stale = append(stale, path)
continue
}
cleaned[path] = mtime
}
if len(stale) > 0 {
if err := database.ReplaceSkippedFiles(cleaned); err != nil {
log.Printf(
"visual studio copilot skip migration: "+
"persist cleaned skip cache: %v",
err,
)
return
}
for _, p := range stale {
delete(skipCache, p)
}
log.Printf(
"visual studio copilot skip migration: cleared %d skip entries",
len(stale),
)
}
if err := database.SetSyncState(
visualStudioCopilotSkipMigrationKey, "done",
); err != nil {
log.Printf(
"visual studio copilot skip migration: set flag: %v", err,
)
}
}
// IsVisualStudioCopilotSkipPath reports whether a skip cache key
// belongs to a Visual Studio Copilot trace: either a physical
// trace file or a <traceFile>#<conversationID> virtual path. It
// is shared with remote sync so both the local and remote skip
// migrations classify paths identically.
func IsVisualStudioCopilotSkipPath(path string) bool {
if parser.IsVisualStudioCopilotTraceFile(path) {
return true
}
_, _, ok := parser.SplitVisualStudioCopilotVirtualPath(path)
return ok
}
// blockedCategorySet converts a slice of category names into a
// set for O(1) lookup. Returns nil when the slice is empty.
// Entries are trimmed and title-cased to match parser categories.
func blockedCategorySet(cats []string) map[string]bool {
if len(cats) == 0 {
return nil
}
m := make(map[string]bool, len(cats))
for _, c := range cats {
c = strings.TrimSpace(c)
if c == "" {
continue
}
c = strings.ToUpper(c[:1]) + strings.ToLower(c[1:])
m[c] = true
}
return m
}
// LastSync returns the time of the last completed sync.
func (e *Engine) LastSync() time.Time {
e.mu.RLock()
defer e.mu.RUnlock()
return e.lastSync
}
// LastSyncStats returns statistics from the last sync.
func (e *Engine) LastSyncStats() SyncStats {
e.mu.RLock()
defer e.mu.RUnlock()
return e.lastSyncStats
}
// CurrentProgress returns the most recent in-flight sync progress.
func (e *Engine) CurrentProgress() (Progress, bool) {
e.mu.RLock()
defer e.mu.RUnlock()
if e.currentProgress == nil {
return Progress{}, false
}
return *e.currentProgress, true
}
func (e *Engine) reportProgress(
onProgress ProgressFunc, p Progress,
) {
e.mu.Lock()
e.currentProgress = &p
e.mu.Unlock()
if onProgress != nil {
onProgress(p)
}
}
func (e *Engine) clearCurrentProgress() {
e.mu.Lock()
e.currentProgress = nil
e.mu.Unlock()
}
// Machine returns the machine name this engine writes on sessions.
func (e *Engine) Machine() string {
if e == nil {
return ""
}
return e.machine
}
type syncJob struct {
processResult
path string
}
func (j syncJob) skipCacheKey() string {
return j.processResult.skipCacheKey(j.path)
}
func (r processResult) skipCacheKey(path string) string {
if r.cacheKey != "" {
return r.cacheKey
}
return path
}
// SyncPaths syncs only the specified changed file paths
// instead of discovering and hashing all session files.
// Paths that don't match known session file patterns are
// silently ignored.
func (e *Engine) SyncPaths(paths []string) {
e.SyncPathsContext(context.Background(), paths)
}
// SyncPathsContext is SyncPaths with caller-controlled cancellation. The
// file watcher threads the serve shutdown context through here: its stop
// path waits for the in-flight onChange callback, so a watcher-driven sync
// that ignored SIGTERM would hold shutdown until a service manager's kill
// timeout instead of aborting between files like every other sync path.
func (e *Engine) SyncPathsContext(ctx context.Context, paths []string) {
if e.refuseWriteInForceParse("SyncPaths") {
return
}
// Capture container states before classifyPaths lists any session rows,
// matching the capture-before-discovery ordering of full syncs.
preContainerStates := e.captureSQLiteContainerStates()
files := e.classifyPaths(paths)
if len(files) == 0 {
return
}
e.syncMu.Lock()
// Defers run LIFO: the emit closure (declared first) runs AFTER
// syncMu.Unlock, so an Emitter implementation cannot widen the
// critical section or deadlock by re-entering sync code. The
// stats variable is captured by the closure and populated below.
var stats SyncStats
defer func() {
if stats.Synced > 0 {
e.emit("sessions")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
e.resetS3CodexIndexCache()
e.anomalies.reset()
// Begin a container pass so an already-trusted, unchanged container
// still gates its fan-out (a spurious watcher event on the DB file
// costs nothing), but never promote from here: a changed-path pass is
// not guaranteed to cover a container's complete session set (a hybrid
// root can fan a single message path out to one SQLite-backed
// session), and promotion from a subset would be unsound. The next
// full sync re-verifies and re-trusts (see opencode_container_gate.go).
e.beginSQLiteContainerPass(files, preContainerStates)
results := e.startWorkers(ctx, files)
stats = e.collectAndBatch(
ctx, results, len(files), len(files), nil,
syncWriteDefault,
)
e.finishSQLiteContainerPass(true, false)
e.anomalies.applyTo(&stats)
e.persistSkipCache()
e.mu.Lock()
e.lastSync = time.Now()
e.lastSyncStats = stats
e.mu.Unlock()
if stats.Synced > 0 {
log.Printf(
"sync: %d file(s) updated", stats.Synced,
)
}
}
// classifyPaths maps changed file system paths to
// parser.DiscoveredFile structs, filtering out paths that don't
// match known session file patterns.
func (e *Engine) classifyPaths(
paths []string,
) []parser.DiscoveredFile {
seen := make(map[string]int, len(paths))
files := make([]parser.DiscoveredFile, 0, len(paths))
for _, p := range paths {
// Codex resolved-index events map to potentially several session
// sources and must classify even when the event path was deleted, so
// they are handled by classifyCodexIndexPath. All other changed paths,
// including Antigravity's sidecar fan-out (annotations, brain,
// history.jsonl), are owned by each provider-authoritative
// SourcesForChangedPath via classifyProviderChangedPath.
dfs := e.classifyCodexIndexPath(p)
dfs = append(dfs, e.classifyProviderChangedPath(p)...)
for _, df := range dfs {
e.invalidateVerifiedDiscoveredSource(df)
key := string(df.Agent) + "\x00" + df.Path
if idx, ok := seen[key]; ok {
files[idx] = mergeChangedPathDiscoveredFile(files[idx], df)
continue
}
seen[key] = len(files)
files = append(files, df)
}
}
files = e.expandClaudeDuplicateCandidates(files)
files = dedupeDiscoveredFiles(files)
return e.dedupeClaudeDiscoveredFiles(files)
}
func mergeChangedPathDiscoveredFile(
current parser.DiscoveredFile,
next parser.DiscoveredFile,
) parser.DiscoveredFile {
current.ForceParse = current.ForceParse || next.ForceParse
current.ProviderProcess = current.ProviderProcess || next.ProviderProcess
if current.Project == "" {
current.Project = next.Project
}
if current.ProviderSource == nil && next.ProviderSource != nil {
current.ProviderSource = next.ProviderSource
}
return current
}
func (e *Engine) classifyProviderChangedPath(
path string,
) []parser.DiscoveredFile {
ctx := context.Background()
eventKind := providerChangedPathEventKind(path)
var files []parser.DiscoveredFile
seen := map[string]struct{}{}
agents := make([]parser.AgentType, 0, len(e.providerFactories))
for agent := range e.providerFactories {
agents = append(agents, agent)
}
slices.SortFunc(agents, func(a, b parser.AgentType) int {
return strings.Compare(string(a), string(b))
})
for _, agentType := range agents {
mode := e.providerMigrationModes[agentType]
switch mode {
case parser.ProviderMigrationProviderAuthoritative:
default:
continue
}
// Codex index (session_index.jsonl) events are owned by the engine's
// DB-aware classifyCodexIndexPath, which fans out only to sessions whose
// stored title changed and resolves a UUID's live/archived duplicate to
// the path the DB already tracks. The provider's broad index fan-out
// would re-add every sibling and prefer the live-over-archived layout,
// resurrecting a stale duplicate over the stored copy, so suppress it
// here and let the engine method classify the index event.
if agentType == parser.AgentCodex &&
filepath.Base(path) == parser.CodexSessionIndexFilename {
continue
}
roots := e.agentDirs[agentType]
if len(roots) == 0 {
continue
}
factory, ok := e.providerFactories[agentType]
if !ok || factory == nil {
continue
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: roots,
Machine: e.machine,
})
def := provider.Definition()
watchRoots := providerChangedPathWatchRoots(ctx, provider, roots)
// Every SourcesForChangedPath implementation resolves the
// changed path within the provider's configured roots or plan
// watch roots (stored-path hints are already scoped to the
// watch root by the query), so an agent whose roots cannot
// contain the path never claims it. Skip it before the
// per-root stored-hint DB queries, which otherwise run for
// every registered agent on every watcher event.
if !changedPathWithinAnyRoot(path, roots) &&
!changedPathWithinAnyRoot(path, watchRoots) {
continue
}
for _, watchRoot := range watchRoots {
var storedSourcePaths []string
if provider.Capabilities().Source.StoredSourceHints == parser.CapabilitySupported {
var err error
storedSourcePaths, err = e.db.ListStoredSourcePathHints(
string(def.Type),
[]string{watchRoot},
)
if err != nil {
log.Printf(
"%s provider changed-path stored hints: %v",
def.Type, err,
)
}
}
sources, err := provider.SourcesForChangedPath(
ctx,
parser.ChangedPathRequest{
Path: path,
EventKind: eventKind,
WatchRoot: watchRoot,
StoredSourcePaths: storedSourcePaths,
},
)
if err != nil {
if !errors.Is(err, parser.ErrUnsupportedProviderFeature) {
log.Printf(
"%s provider changed-path classification: %v",
def.Type, err,
)
}
continue
}
for _, source := range sources {
sourcePath := providerDiscoveredPath(source)
if sourcePath == "" {
continue
}
agent := source.Provider
if agent == "" {
agent = def.Type
}
key := string(agent) + "\x00" + sourcePath
if _, ok := seen[key]; ok {
continue
}
if eventKind == "remove" &&
filepath.Clean(sourcePath) == filepath.Clean(path) &&
!parser.IsRegularFile(sourcePath) &&
!providerDeletedPhysicalSQLiteSource(agent, sourcePath) {
continue
}
seen[key] = struct{}{}
sourceCopy := source
discovered := parser.DiscoveredFile{
Path: sourcePath,
Project: source.ProjectHint,
Agent: agent,
ForceParse: providerChangedPathForceParse(agent, sourcePath, path, eventKind, mode),
ProviderSource: &sourceCopy,
ProviderProcess: mode == parser.ProviderMigrationProviderAuthoritative,
}
// A watcher event names a concrete change even when the
// session's stat signature cannot see it (a same-size,
// same-mtime child rewrite), so the storage gate must
// re-verify this session by content on the next pass.
if sessionPath := e.openCodeStorageSessionPath(discovered); sessionPath != "" {
e.invalidateOpenCodeStorageSession(sessionPath)
}
files = append(files, discovered)
}
}
}
return files
}
func providerChangedPathWatchRoots(
ctx context.Context,
provider parser.Provider,
roots []string,
) []string {
plan, err := provider.WatchPlan(ctx)
if err == nil && len(plan.Roots) > 0 {
watchRoots := make([]string, 0, len(plan.Roots))
seen := make(map[string]struct{}, len(plan.Roots))
for _, root := range plan.Roots {
path := filepath.Clean(root.Path)
if path == "" || path == "." {
continue
}
if _, ok := seen[path]; ok {
continue
}
seen[path] = struct{}{}
watchRoots = append(watchRoots, path)
}
if len(watchRoots) > 0 {
return watchRoots
}
}
watchRoots := make([]string, 0, len(roots))
for _, root := range roots {
root = filepath.Clean(root)
if root == "" || root == "." {
continue
}
watchRoots = append(watchRoots, root)
}
return watchRoots
}
func providerChangedPathForceParse(
agent parser.AgentType,
sourcePath string,
eventPath string,
eventKind string,
mode parser.ProviderMigrationMode,
) bool {
if processFileUsesProvider(agent) {
return eventKind == "remove" &&
providerDeletedPhysicalSQLiteSource(agent, sourcePath)
}
if mode != parser.ProviderMigrationProviderAuthoritative {
return true
}
if filepath.Clean(sourcePath) != filepath.Clean(eventPath) &&
!providerVirtualSourceBackedByEvent(sourcePath, eventPath) {
// OpenCode-family storage sessions resolve message/part events
// to their session JSON, whose fingerprint and stat signature
// span those same child files, and the classifier invalidates
// the session's storage-gate trust for every event. The normal
// freshness path therefore re-verifies by content, while a
// forced parse would bypass dropUnchangedSharedSQLiteResults
// and rewrite the whole session on every streamed append.
// Remove events keep the force so a deleted child still
// re-emits through the deletion path.
if eventKind != "remove" &&
isOpenCodeFormatStorageAgent(agent) &&
isOpenCodeFormatStoragePath(agent, sourcePath) {
return false
}
return true
}
return eventKind == "remove" &&
providerDeletedPhysicalSQLiteSource(agent, sourcePath)
}
func providerVirtualSourceBackedByEvent(sourcePath, eventPath string) bool {
idx := strings.LastIndex(sourcePath, "#")
if idx < 0 {
return false
}
dbPath := filepath.Clean(sourcePath[:idx])
eventPath = filepath.Clean(eventPath)
return eventPath == dbPath ||
eventPath == dbPath+"-wal" ||
eventPath == dbPath+"-shm"
}
func providerChangedPathEventKind(path string) string {
if path == "" {
return ""
}
// A virtual source path (e.g. a SQLite per-session path
// "<db>#<sessionID>") is never a real file. Resolve it to its physical
// container so an existence check reflects whether the backing store is
// present rather than always reporting the synthetic path as removed.
statPath := path
if container, _, ok := parser.ParseVirtualSourcePath(path); ok {
statPath = container
}
if _, err := os.Lstat(statPath); err != nil && os.IsNotExist(err) {
return "remove"
}
return "write"
}
func providerDiscoveredPath(source parser.SourceRef) string {
for _, path := range []string{
source.DisplayPath,
source.FingerprintKey,
source.Key,
} {
if path != "" {
return path
}
}
return ""
}
func providerDeletedPhysicalSQLiteSource(
agent parser.AgentType, path string,
) bool {
switch agent {
case parser.AgentZed:
return filepath.Base(path) == "threads.db"
case parser.AgentZCode:
return filepath.Base(path) == parser.ZCodeDBName
case parser.AgentShelley:
return filepath.Base(path) == shelleyDBFile
default:
return false
}
}
func dedupeDiscoveredFiles(
files []parser.DiscoveredFile,
) []parser.DiscoveredFile {
return dedupeDiscoveredFilesByPreference(files, preferDiscoveredFile)
}
func dedupeDiscoveredFilesPreferNewestCodex(
files []parser.DiscoveredFile,
) []parser.DiscoveredFile {
return dedupeDiscoveredFilesByPreference(files, preferNewestCodexDiscoveredFile)
}
func dedupeDiscoveredFilesByPreference(
files []parser.DiscoveredFile,
prefer func(candidate, current parser.DiscoveredFile) bool,
) []parser.DiscoveredFile {
if len(files) < 2 {
return files
}
bestByKey := make(map[string]parser.DiscoveredFile, len(files))
for _, file := range files {
key := discoveredFileKey(file)
if current, ok := bestByKey[key]; ok {
if prefer(file, current) {
bestByKey[key] = file
}
continue
}
bestByKey[key] = file
}
out := make([]parser.DiscoveredFile, 0, len(bestByKey))
for _, file := range files {
key := discoveredFileKey(file)
chosen, ok := bestByKey[key]
if !ok || chosen.Path != file.Path || chosen.Agent != file.Agent {
continue
}
out = append(out, file)
delete(bestByKey, key)
}
return out
}
func discoveredFileKey(file parser.DiscoveredFile) string {
if file.Agent == parser.AgentCodex {
if id := parser.CodexSessionUUIDFromFilename(filepath.Base(file.Path)); id != "" {
return string(file.Agent) + "\x00" +
discoveredFileIDPrefix(file) + "\x00" + id
}
}
return string(file.Agent) + "\x00" + file.Path
}
func discoveredFileIDPrefix(file parser.DiscoveredFile) string {
if isS3SourcePath(file.Path) {
return s3SessionIDPrefix(file.Machine)
}
return ""
}
func preferDiscoveredFile(
candidate, current parser.DiscoveredFile,
) bool {
if candidate.Agent == parser.AgentCodex && current.Agent == parser.AgentCodex {
candLayout := codexLayoutForPath(candidate.Path)
currLayout := codexLayoutForPath(current.Path)
if candLayout != currLayout {
return candLayout == parser.CodexLayoutDated
}
}
return false
}
func preferNewestCodexDiscoveredFile(
candidate, current parser.DiscoveredFile,
) bool {
if candidate.Agent == parser.AgentCodex && current.Agent == parser.AgentCodex {
candMTime, candOK := discoveredFileMTime(candidate.Path)
currMTime, currOK := discoveredFileMTime(current.Path)
if candOK && currOK && candMTime != currMTime {
return candMTime > currMTime
}
}
return preferDiscoveredFile(candidate, current)
}
func discoveredFileMTime(path string) (int64, bool) {
info, err := os.Stat(path)
if err != nil {
return 0, false
}
return info.ModTime().UnixNano(), true
}
func (e *Engine) expandClaudeDuplicateCandidates(
files []parser.DiscoveredFile,
) []parser.DiscoveredFile {
sessionIDs := make(map[string]struct{})
seen := make(map[string]struct{}, len(files))
for _, file := range files {
seen[string(file.Agent)+"\x00"+file.Path] = struct{}{}
if file.Agent != parser.AgentClaude {
continue
}
sessionID := claudeSessionIDFromPath(file.Path)
if sessionID == "" {
continue
}
sessionIDs[sessionID] = struct{}{}
}
if len(sessionIDs) == 0 {
return files
}
out := files
for _, claudeDir := range e.agentDirs[parser.AgentClaude] {
for _, candidate := range parser.ClaudeProjectSessionFiles(claudeDir) {
sessionID := claudeSessionIDFromPath(candidate.Path)
if _, ok := sessionIDs[sessionID]; !ok {
continue
}
key := string(candidate.Agent) + "\x00" + candidate.Path
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
out = append(out, candidate)
}
}
return out
}
func codexLayoutForPath(path string) parser.CodexLayout {
path = filepath.Clean(path)
name := filepath.Base(path)
if parser.CodexSessionUUIDFromFilename(name) == "" {
return parser.CodexLayoutUnknown
}
day := filepath.Base(filepath.Dir(path))
month := filepath.Base(filepath.Dir(filepath.Dir(path)))
year := filepath.Base(filepath.Dir(filepath.Dir(filepath.Dir(path))))
if parser.IsDigits(day) && parser.IsDigits(month) && parser.IsDigits(year) {
return parser.CodexLayoutDated
}
return parser.CodexLayoutArchivedFlat
}
// isUnder checks whether path is strictly inside dir after
// cleaning both paths. Returns the relative path on success.
func isUnder(dir, path string) (string, bool) {
dir = filepath.Clean(dir)
path = filepath.Clean(path)
rel, err := filepath.Rel(dir, path)
if err != nil {
return "", false
}
sep := string(filepath.Separator)
if rel == "." || rel == ".." ||
strings.HasPrefix(rel, ".."+sep) {
return "", false
}
return rel, true
}
// shelleyDBFile is the shared Shelley conversation database basename. Zed and
// Shelley are provider-authoritative, so their changed-path classification and
// parse run through the provider facade; this constant remains for the
// provider-neutral physical-DB deletion and skip-cache checks in the engine.
const shelleyDBFile = "shelley.db"
// resyncTempSuffix is appended to the original DB path to
// form the temp database path during resync.
const resyncTempSuffix = "-resync"
// ResyncAll builds a fresh database from scratch, syncs all
// sessions into it, copies insights from the old DB, then
// atomically swaps the files and reopens the original DB
// handle. This avoids the per-row trigger overhead of bulk
// deleting hundreds of thousands of messages in place.
// shouldAbortResyncSwap decides whether a finished resync pass built a
// database that would be worse than the original, so the swap must be
// abandoned:
// - sync was cancelled (partial rebuild)
// - nothing synced at all (empty discovery, or all skipped)
// when old DB had data
// - more files failed than succeeded (permission errors,
// disk issues)
//
// OpenCode-only rebuilds are allowed to finish with 0 freshly synced
// sessions when every storage parse was intentionally preserved
// against the archive; orphan copy restores those rows immediately
// after the sync pass. A few permanent parse failures are tolerated
// since those files were broken in the old DB too.
// OpenCode-format storage is a self-preserving container store that
// flows through file discovery, so it is excluded from the discovery
// check just as it is subtracted from oldFileSessions by the caller.
// Otherwise its discovery would mask the disappearance of plain
// file-backed sessions whose directories went empty.
func shouldAbortResyncSwap(
stats SyncStats, oldFileSessions, trashedCopied int,
) bool {
emptyDiscovery := stats.nonContainerDiscovered == 0 &&
oldFileSessions > 0
preservedOnly := stats.Synced == 0 &&
stats.TotalSessions > 0 &&
stats.Failed == 0 &&
(oldFileSessions == 0 || trashedCopied > 0)
excludedOnly := stats.Synced == 0 &&
stats.TotalSessions > 0 &&
stats.Failed == 0 &&
stats.parserExcludedFiles > 0 &&
stats.filesOK == stats.parserExcludedFiles
// A zero-write run is intentional when the sync_include_cwd_prefixes
// allow-list vetoed sessions AND every OK file is accounted for as
// either fully filtered or parser-excluded: the swap proceeds and
// the orphan copy restores the archived rows, because the filter
// gates ingestion only. Requiring the full accounting keeps the
// guard armed for mixed runs where other files produced nothing for
// an unexplained reason.
cwdFilteredOnly := stats.Synced == 0 &&
stats.TotalSessions > 0 &&
stats.Failed == 0 &&
stats.cwdFilteredSessions > 0 &&
stats.filesOK == stats.cwdFilteredFiles+stats.parserExcludedFiles
return stats.Aborted ||
emptyDiscovery ||
(stats.Synced == 0 &&
stats.TotalSessions > 0 &&
!preservedOnly &&
!excludedOnly &&
!cwdFilteredOnly) ||
(stats.Failed > 0 && stats.Failed > stats.filesOK)
}
func (e *Engine) ResyncAll(
ctx context.Context, onProgress ProgressFunc,
) (stats SyncStats) {
if e.refuseWriteInForceParse("ResyncAll") {
return SyncStats{}
}
e.syncMu.Lock()
// Defers LIFO: Unlock runs before emit.
defer func() {
if stats.Synced > 0 {
e.emit("sync")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
return e.resyncAllLocked(ctx, onProgress)
}
func (e *Engine) resyncAllLocked(
ctx context.Context, onProgress ProgressFunc,
) (stats SyncStats) {
stats, _ = e.resyncAllWithOptionsLocked(
ctx, onProgress, RebuildOptions{}, productionRebuildOperations,
)
// Preserve the legacy result shape; phase diagnostics are part of the
// options entrypoint's observable contract only.
stats.RebuildPhases = nil
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats
}
// ResyncAllWithOptions atomically rebuilds the archive from the local sources
// plus each configured contributor.
func (e *Engine) ResyncAllWithOptions(
ctx context.Context, onProgress ProgressFunc, opts RebuildOptions,
) (stats SyncStats, err error) {
return e.resyncAllWithOptionsAndOperations(
ctx, onProgress, opts, productionRebuildOperations,
)
}
func (e *Engine) resyncAllWithOptionsAndOperations(
ctx context.Context, onProgress ProgressFunc, opts RebuildOptions,
ops rebuildOperations,
) (stats SyncStats, err error) {
if e.refuseWriteInForceParse("ResyncAllWithOptions") {
return SyncStats{}, nil
}
e.syncMu.Lock()
defer func() {
if stats.Synced > 0 && !stats.Aborted {
e.emit("sync")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
opts.includePhaseDiagnostics = true
return e.resyncAllWithOptionsLocked(
ctx, onProgress, opts, ops,
)
}
func (e *Engine) resyncAllWithOptionsLocked(
ctx context.Context, onProgress ProgressFunc, opts RebuildOptions,
ops rebuildOperations,
) (stats SyncStats, retErr error) {
ops = ops.withDefaults()
reportResyncProgress := func(p Progress) {
p.Resync = true
if p.Phase == PhaseSyncing && p.Detail == "" {
p.Detail = "Syncing sessions into rebuilt database"
}
e.reportProgress(onProgress, p)
}
reportResyncPhase := func(phase Phase, detail, hint string) {
reportResyncProgress(Progress{
Phase: phase,
Detail: detail,
Hint: hint,
})
}
// Resync rebuilds the archive from scratch, so every shared-SQLite
// container and storage session must be re-verified against the
// fresh database.
e.clearTrustedSQLiteContainers()
e.clearTrustedOpenCodeStorageSessions()
e.clearVerifiedSources()
origDB := e.db
origPath := origDB.Path()
tempPath := origPath + resyncTempSuffix
reportResyncPhase(
PhasePreparingResync,
"Preparing full resync",
"",
)
// Snapshot old non-OpenCode-format file-backed session count
// to detect empty-discovery. OpenCode-format agents are
// excluded entirely because a root may legitimately fall back
// between storage and SQLite sources across resyncs. Fail closed:
// if we can't query, assume old DB has file-backed data
// worth protecting.
oldFileSessions, err := e.protectedFileSessionCount(origDB, "", "", false)
if err != nil {
log.Printf("resync: get old file count: %v", err)
oldFileSessions = 1
}
localOldFileSessions := oldFileSessions
rebuildOldFileSessions := oldFileSessions
contributorOldFileSessions := make([]int, len(opts.Contributors))
if len(opts.Contributors) > 0 {
localOldFileSessions, err = e.protectedFileSessionCount(
origDB, e.machine, "", e.machine != "",
)
if err != nil {
log.Printf("resync: get old local file count: %v", err)
localOldFileSessions = 1
}
for i, contributor := range opts.Contributors {
count, countErr := e.protectedFileSessionCount(
origDB, contributor.Config.Machine,
contributor.Config.IDPrefix,
contributor.Config.Machine != "",
)
if countErr != nil {
log.Printf(
"resync: get old contributor %q file count: %v",
contributor.Name, countErr,
)
count = 1
}
contributorOldFileSessions[i] = count
if contributor.Config.Machine == e.machine &&
contributor.Config.IDPrefix != "" {
localOldFileSessions -= count
if localOldFileSessions < 0 {
localOldFileSessions = 0
}
}
}
rebuildOldFileSessions = localOldFileSessions
for _, count := range contributorOldFileSessions {
rebuildOldFileSessions += count
}
}
// Clean up stale temp DB from a prior crash.
removeTempDB(tempPath)
// 1. Snapshot and clear in-memory skip cache. The
// snapshot is restored on early failure so behavior
// matches the persisted DB until the next restart.
e.skipMu.Lock()
savedSkipCache := e.skipCache
savedSkipHashKeys := e.skipHashKeys
e.skipCache = make(map[string]int64)
e.skipHashKeys = make(map[string]string)
e.skipMu.Unlock()
restoreSkipCache := func() {
e.skipMu.Lock()
e.skipCache = savedSkipCache
e.skipHashKeys = savedSkipHashKeys
e.skipMu.Unlock()
}
// 2. Open a fresh DB at the temp path.
reportResyncPhase(
PhasePreparingResync,
"Opening temporary database",
"",
)
newDB, err := db.Open(tempPath)
if err != nil {
log.Printf("resync: open temp db: %v", err)
restoreSkipCache()
stats = SyncStats{
Aborted: true,
Warnings: []string{
"resync failed: " + err.Error(),
},
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
if err := newDB.CopyArchiveIdentityFrom(origPath); err != nil {
log.Printf("resync: preserve archive identity: %v", err)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
stats = SyncStats{
Aborted: true,
Warnings: []string{
"resync failed: preserve archive identity: " + err.Error(),
},
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return
}
// 2b. Copy excluded session IDs from the old DB so that
// UpsertSession skips permanently deleted sessions during
// the sync. This must happen before syncAllLocked.
reportResyncPhase(
PhasePreparingResync,
"Copying deletion state into temporary database",
"",
)
if err := newDB.CopyExcludedSessionsFrom(origPath); err != nil {
log.Printf("resync: pre-sync copy excluded sessions: %v", err)
// Non-fatal: worst case, deleted sessions reappear.
}
trashedCopied := 0
if n, err := newDB.CopyTrashedDataFrom(origPath); err != nil {
log.Printf("resync: pre-sync copy trashed sessions: %v", err)
// Non-fatal: worst case, trashed sessions are reparsed
// and then re-marked as trashed by metadata copy.
} else if n > 0 {
trashedCopied = n
log.Printf("resync: pre-sync copied %d trashed sessions", n)
}
// The temp DB is not swapped into production until the end,
// so avoid per-row FTS trigger work during the bulk load and
// rebuild the index once all message rows are final.
ftsDropped := false
if newDB.HasFTS() {
tFTS := time.Now()
reportResyncPhase(
PhasePreparingResync,
"Disabling temporary search index updates",
"",
)
if err := newDB.DropFTS(); err != nil {
log.Printf("resync: drop temp fts: %v", err)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
stats = SyncStats{
Aborted: true,
Warnings: []string{
"resync failed: drop temp fts: " +
err.Error(),
},
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
ftsDropped = true
log.Printf(
"resync: drop temp fts: %s",
time.Since(tFTS).Round(time.Millisecond),
)
}
// 3. Point engine at newDB and sync into it. Report discovery as its
// own phase first: syncAllLocked walks every source before emitting
// its first syncing event, and on a large archive that walk takes
// minutes. Without this marker the progress printer credits that
// silent time to the preceding (instant) "Disabling ..." phase.
e.openCodeArchiveStore = origDB
e.db = newDB
reportResyncPhase(
PhaseDiscovering,
"Discovering sessions",
"",
)
stats = e.syncAllLocked(
ctx, reportResyncProgress, time.Time{}, nil, syncWriteBulk, true, false,
)
e.db = origDB // restore immediately
e.openCodeArchiveStore = nil
e.phaseStats.Log("resync")
if opts.includePhaseDiagnostics {
stats.RebuildPhases = append(stats.RebuildPhases,
phaseSnapshot("local", &e.phaseStats))
}
localStats := stats
if stats.Aborted || ctx.Err() != nil {
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
stats.Aborted = true
stats.Warnings = append(stats.Warnings, fmt.Sprintf(
"resync aborted: %d synced, %d failed",
stats.Synced, stats.Failed,
))
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
if err := ctx.Err(); err != nil {
return stats, err
}
return stats, nil
}
for contributorIndex, contributor := range opts.Contributors {
contributorEngine := NewEngine(newDB, contributor.Config)
contributorEngine.openCodeArchiveStore = origDB
contributorProgress := func(p Progress) {
if contributor.Progress != nil {
p = contributor.Progress(p)
}
p.SessionsDone += stats.TotalSessions
p.SessionsTotal += stats.TotalSessions
p.MessagesIndexed += stats.messagesIndexed
reportResyncProgress(p)
}
contributorStats := contributorEngine.syncAllLocked(
ctx, contributorProgress, time.Time{}, nil,
syncWriteBulk, true, false,
)
contributorEngine.phaseStats.Log("resync contributor " + contributor.Name)
phase := phaseSnapshot(contributor.Name, &contributorEngine.phaseStats)
contributorSafetyAbort := shouldAbortResyncSwap(
contributorStats,
contributorOldFileSessions[contributorIndex],
0,
)
mergeSyncStats(&stats, contributorStats)
if opts.includePhaseDiagnostics {
stats.RebuildPhases = append(stats.RebuildPhases, phase)
}
if contributorStats.Aborted || stats.Aborted || ctx.Err() != nil ||
contributorSafetyAbort {
contributorEngine.Close()
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
stats.Aborted = true
stats.Warnings = append(stats.Warnings, fmt.Sprintf(
"resync aborted: contributor %q did not complete",
contributor.Name,
))
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
if err := ctx.Err(); err != nil {
return stats, err
}
return stats, nil
}
if contributor.AfterSync != nil {
if err := contributor.AfterSync(contributorEngine, newDB); err != nil {
contributorEngine.Close()
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
if reopenErr := origDB.Reopen(); reopenErr != nil {
log.Printf("resync: contributor failure recovery reopen: %v", reopenErr)
}
stats.Aborted = true
stats.Warnings = append(stats.Warnings, fmt.Sprintf(
"resync contributor %q failed: %v",
contributor.Name, err,
))
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, &RebuildContributorError{
Contributor: contributor.Name,
Err: err,
}
}
}
contributorEngine.Close()
}
localSafetyAbort := false
if len(opts.Contributors) > 0 {
// Evaluate local safety after contributors so a contributor's own
// cancellation or failure remains the reported abort reason. Trash
// copied from the old archive cannot make an empty local pass safe.
localSafetyAbort = shouldAbortResyncSwap(
localStats, localOldFileSessions, 0,
)
}
abortSwap := localSafetyAbort ||
shouldAbortResyncSwap(stats, rebuildOldFileSessions, trashedCopied)
if abortSwap {
log.Printf(
"resync: aborting swap, %d synced / %d failed / %d total",
stats.Synced, stats.Failed, stats.TotalSessions,
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
stats.Aborted = true
stats.Warnings = append(stats.Warnings, fmt.Sprintf(
"resync aborted: %d synced, %d failed",
stats.Synced, stats.Failed,
))
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
if err := ctx.Err(); err != nil {
return stats, err
}
return stats, nil
}
// 4. Close origDB connections first to quiesce writes,
// then copy insights into newDB (which is still open).
// This ensures no insight writes land in the old DB
// after the copy.
reportResyncPhase(
PhaseCopyingMetadata,
"Closing current database before final copy",
"",
)
if err := origDB.CloseConnections(); err != nil {
log.Printf("resync: close orig db: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"close before swap failed: "+err.Error(),
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
// Connections may be partially closed; reopen to
// restore service before returning.
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
// Re-copy excluded session IDs now that origDB is quiesced.
// This catches any permanent deletes that occurred during
// the sync window (between the pre-sync copy and now).
// Also purge any sessions that were synced into newDB
// before the exclusion was recorded.
reportResyncPhase(
PhaseCopyingMetadata,
"Copying sync metadata",
"",
)
if err := newDB.CopyExcludedSessionsFrom(origPath); err != nil {
log.Printf("resync: post-sync copy excluded sessions: %v", err)
}
if err := newDB.PurgeExcludedSessions(); err != nil {
log.Printf("resync: purge excluded sessions: %v", err)
}
if err := newDB.CopySyncStateFrom(origPath); err != nil {
log.Printf("resync: copy sync state: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"sync state copy failed, aborting swap: "+err.Error(),
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
// Copy insights into newDB from the quiesced old DB file.
tInsights := time.Now()
reportResyncPhase(
PhaseCopyingMetadata,
"Copying cached insights",
"",
)
if err := newDB.CopyInsightsFrom(origPath); err != nil {
log.Printf("resync: copy insights: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"insights copy failed, aborting swap: "+
err.Error(),
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
log.Printf(
"resync: copy insights: %s",
time.Since(tInsights).Round(time.Millisecond),
)
// Copy model pricing so usage costs survive the swap. The
// startup seed only runs once per daemon lifetime, so a
// resync triggered through the sync API would otherwise
// leave the rebuilt DB with an empty pricing table and
// every usage cost reading $0.00 until the next restart.
// Non-fatal: a failed copy degrades cost display but does
// not justify aborting the resync, and the next daemon
// startup re-seeds pricing.
if err := newDB.CopyModelPricingFrom(origPath); err != nil {
log.Printf("resync: copy model pricing: %v", err)
stats.Warnings = append(stats.Warnings,
"model pricing copy failed; usage costs show as $0.00 "+
"until the next daemon restart re-seeds pricing: "+
err.Error(),
)
}
// Copy orphaned sessions (source files gone) from the
// old DB so archived data is preserved. Failure aborts
// the swap to avoid losing archived sessions.
reportResyncPhase(
PhaseCopyingOrphans,
"Copying archived sessions",
"",
)
orphaned, err := newDB.CopyOrphanedDataFromExcluding(
origPath, stats.parserExcludedIDs,
)
if err != nil {
log.Printf("resync: copy orphaned sessions: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"orphaned session copy failed, aborting swap: "+
err.Error(),
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
stats.OrphanedCopied = orphaned
// Re-link subagent sessions after orphan copy so copied
// tool_calls.subagent_session_id references are resolved.
if orphaned > 0 {
reportResyncPhase(
PhaseCopyingOrphans,
"Relinking archived subagent sessions",
"",
)
if err := newDB.LinkSubagentSessions(); err != nil {
log.Printf("resync: relink subagent sessions: %v", err)
}
}
// Copy recall entries and their evidence from the quiesced old DB.
// The fresh DB is built from source files, which never contain
// recall entries, so without this every accepted entry is lost on
// resync. Runs after the orphan copy so referenced sessions exist.
// Failure aborts the swap to avoid destroying the recall archive.
if err := newDB.CopyRecallEntriesFrom(origPath); err != nil {
log.Printf("resync: copy recall entries: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"recall copy failed, aborting swap: "+err.Error(),
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
// Merge user-managed data and immutable project-identity snapshots from the
// old DB. Snapshot copy happens after parsing because the destination rows
// reference freshly parsed sessions. Failure must abort the swap: a fresh
// database without those snapshots could no longer export stable identity
// after a source working directory disappears.
reportResyncPhase(
PhaseCopyingMetadata,
"Copying user-managed session metadata",
"",
)
if err := newDB.CopySessionMetadataFrom(origPath); err != nil {
log.Printf("resync: copy session metadata: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"session metadata copy failed, aborting swap: "+err.Error(),
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
if _, err := newDB.ApplyWorktreeProjectMappingsFromSync(
context.Background(), e.machine,
); err != nil {
log.Printf("resync: apply worktree mappings: %v", err)
}
// Reclassify is_automated across every row. Orphan-copied
// rows carry is_automated values computed against the OLD
// DB's classifier set; the temp DB's at-Open backfill ran on
// an empty table and stamped the current hash, so without
// this pass those rows would be permanently stuck with stale
// flags. Non-fatal: worst case, some sessions keep their
// pre-resync classification until the next algorithm bump.
reportResyncPhase(
PhaseReclassifying,
"Reclassifying sessions",
"",
)
if err := newDB.ForceBackfillIsAutomated(); err != nil {
log.Printf("resync: reclassify is_automated: %v", err)
}
if ftsDropped {
tFTS := time.Now()
reportResyncPhase(
PhaseRebuildingSearch,
"Rebuilding search index",
"Rebuilding the search index may take a while on large archives.",
)
if err := ops.rebuildFTS(newDB); err != nil {
log.Printf("resync: rebuild fts: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"fts rebuild failed, aborting swap: "+
err.Error(),
)
newDB.Close()
removeTempDB(tempPath)
restoreSkipCache()
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
log.Printf(
"resync: rebuild fts: %s",
time.Since(tFTS).Round(time.Millisecond),
)
}
// 5. Close newDB and swap files, then reopen origDB.
reportResyncPhase(
PhaseSwappingDatabase,
"Swapping rebuilt database into place",
"",
)
newDB.Close()
removeWAL(origPath)
if err := os.Rename(tempPath, origPath); err != nil {
log.Printf("resync: rename temp db: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"resync swap failed: "+err.Error(),
)
removeTempDB(tempPath)
restoreSkipCache()
// Restore service even on rename failure.
if rerr := origDB.Reopen(); rerr != nil {
log.Printf("resync: recovery reopen: %v", rerr)
}
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
}
removeWAL(tempPath)
if err := ops.reopen(origDB); err != nil {
log.Printf("resync: reopen db: %v", err)
stats.Aborted = true
stats.Warnings = append(stats.Warnings,
"resync swap completed but reopening active database failed: "+
err.Error(),
)
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
return stats, err
} else {
origDB.MarkDataCurrent()
if err := origDB.CheckpointWALTruncateWithRetry(ctx); err != nil {
if errors.Is(err, db.ErrWALCheckpointBusy) {
log.Printf("resync: wal checkpoint busy")
} else {
log.Printf("resync: wal checkpoint: %v", err)
}
}
}
// 6. Persist skip cache into the new DB.
e.persistSkipCache()
e.mu.Lock()
e.lastSyncStats = stats
e.mu.Unlock()
// Emission happens via the deferred closure above, after
// syncMu is released.
return
}
// removeTempDB removes a temp database and its WAL/SHM files.
func removeTempDB(path string) {
for _, suffix := range []string{"", "-wal", "-shm"} {
os.Remove(path + suffix)
}
}
// removeWAL removes WAL and SHM files for a database path.
func removeWAL(path string) {
os.Remove(path + "-wal")
os.Remove(path + "-shm")
}
func (e *Engine) countRootOpenCodeFormatSessions(
database *db.DB, agent parser.AgentType, machine, idPrefix string, scoped bool,
) int {
if !isOpenCodeFormatStorageAgent(agent) {
return 0
}
machinePredicate := ""
args := []any{string(agent)}
if scoped {
machinePredicate = " AND machine = ?"
args = append(args, machine)
}
if idPrefix != "" {
machinePredicate += " AND substr(id, 1, length(?)) = ?"
args = append(args, idPrefix, idPrefix)
}
var count int
err := database.Reader().QueryRow(`
SELECT COUNT(*) FROM sessions
WHERE agent = ?
AND message_count > 0
AND relationship_type NOT IN ('subagent', 'fork')
AND deleted_at IS NULL
`+machinePredicate, args...).Scan(&count)
if err != nil {
log.Printf("count root %s sessions: %v", agent, err)
}
return count
}
func (e *Engine) protectedFileSessionCount(
database *db.DB, machine, idPrefix string, scoped bool,
) (int, error) {
var count int
var err error
if scoped {
if idPrefix != "" {
count, err = database.FileBackedSessionCountForSource(
context.Background(), machine, idPrefix,
)
} else {
count, err = database.FileBackedSessionCountForMachine(
context.Background(), machine,
)
}
} else {
count, err = database.FileBackedSessionCount(context.Background())
}
if err != nil {
return 0, err
}
for _, agent := range []parser.AgentType{
parser.AgentOpenCode,
parser.AgentKilo,
parser.AgentMiMoCode,
parser.AgentIcodemate,
} {
count -= e.countRootOpenCodeFormatSessions(
database, agent, machine, idPrefix, scoped,
)
}
if count < 0 {
count = 0
}
return count, nil
}
// Sync state keys persisted in pg_sync_state.
const (
syncStateStartedAt = "last_sync_started_at"
syncStateFinishedAt = "last_sync_finished_at"
)
// LastSyncStartedAt returns the recorded start time of the
// most recent sync. Returns zero time if no sync has run.
// Use this as the mtime cutoff for quick incremental syncs —
// anything modified at or after this time must be re-evaluated.
func (e *Engine) LastSyncStartedAt() time.Time {
raw, err := e.db.GetSyncState(syncStateStartedAt)
if err != nil || raw == "" {
return time.Time{}
}
t, err := time.Parse(time.RFC3339Nano, raw)
if err != nil {
return time.Time{}
}
return t
}
// SyncThenRun runs the local sync/resync decision and invokes work while
// syncMu is still held. Daemon-owned mirror pushes use this to keep local sync,
// row scanning, and watermark writes serialized against watcher and periodic
// sync passes.
func (e *Engine) SyncThenRun(
ctx context.Context,
full bool,
onProgress ProgressFunc,
work func(forceFull bool) error,
) (stats SyncStats, err error) {
if e.refuseWriteInForceParse("SyncThenRun") {
return SyncStats{}, nil
}
e.syncMu.Lock()
// Defers run LIFO: Unlock runs before emit.
defer func() {
if stats.Synced > 0 {
e.emit("sync")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
stats, err = e.syncThenRunLocked(ctx, full, onProgress, work)
if err == nil {
// Release while syncMu is still held. A timed fallback that was
// already waiting on the mutex can then recheck the gate before it
// performs any duplicate startup work.
e.ReleaseStartupMaintenance()
}
return stats, err
}
func (e *Engine) syncThenRunLocked(
ctx context.Context,
full bool,
onProgress ProgressFunc,
work func(forceFull bool) error,
) (stats SyncStats, err error) {
didResync := full || e.db.NeedsResync()
if didResync {
stats = e.resyncAllLocked(ctx, onProgress)
if stats.Aborted && ctx.Err() == nil {
stats = e.syncAllLocked(
ctx, onProgress, time.Time{}, nil,
syncWriteDefault, true, false,
)
}
} else {
stats = e.syncAllLocked(
ctx, onProgress, time.Time{}, nil,
syncWriteDefault, true, false,
)
}
if ctx.Err() != nil {
return stats, ctx.Err()
}
// work typically scans and pushes SQLite rows, so flush any
// deferred signal recomputes first (inline: syncMu is held) or
// pushed sessions could carry stale signal/secret fields.
e.signalSched.flushAllInline()
if err := work(full || didResync); err != nil {
return stats, err
}
return stats, nil
}
// RebuildCleanup owns resources prepared for a multi-source rebuild. Close
// may be retried when it fails so callers can retain mirror locks and temporary
// roots instead of silently losing cleanup ownership.
type RebuildCleanup interface {
Close() error
}
type rebuildCleanupError struct {
owner RebuildCleanup
err error
}
func (e *rebuildCleanupError) Error() string { return e.err.Error() }
func (e *rebuildCleanupError) Unwrap() error { return e.err }
func (e *rebuildCleanupError) RetryCleanup() error {
return e.owner.Close()
}
// SyncThenRunWithRebuild coordinates local sync, optional contributor
// preparation, an atomic multi-source rebuild, and post-rebuild work under the
// engine's exclusive sync lock. Preparation only runs when a rebuild is
// required, and work never runs after a failed or aborted rebuild.
func (e *Engine) SyncThenRunWithRebuild(
ctx context.Context,
full bool,
onProgress ProgressFunc,
prepare func() (RebuildOptions, RebuildCleanup, error),
work func(forceFull, rebuilt bool) error,
) (stats SyncStats, retErr error) {
if e.refuseWriteInForceParse("SyncThenRunWithRebuild") {
return SyncStats{}, nil
}
e.syncMu.Lock()
defer func() {
if stats.Synced > 0 && !stats.Aborted {
e.emit("sync")
}
}()
defer e.syncMu.Unlock()
defer func() {
if retErr == nil && !stats.Aborted {
// Match SyncThenRun: successful foreground coordination unblocks
// startup backfills while syncMu is still held.
e.ReleaseStartupMaintenance()
}
}()
defer e.clearCurrentProgress()
didResync := full || e.db.NeedsResync()
if didResync {
opts, cleanup, err := prepare()
if cleanup != nil {
defer func() {
if err := cleanup.Close(); err != nil {
retErr = errors.Join(retErr, &rebuildCleanupError{
owner: cleanup,
err: err,
})
}
}()
}
if err != nil {
return SyncStats{}, err
}
opts.includePhaseDiagnostics = true
stats, err = e.resyncAllWithOptionsLocked(
ctx, onProgress, opts, productionRebuildOperations,
)
if err != nil {
return stats, err
}
if stats.Aborted {
return stats, nil
}
} else {
stats = e.syncAllLocked(
ctx, onProgress, time.Time{}, nil,
syncWriteDefault, true, false,
)
}
if err := ctx.Err(); err != nil {
return stats, err
}
e.signalSched.flushAllInline()
if err := work(full || didResync, didResync); err != nil {
return stats, err
}
return stats, nil
}
// ReleaseStartupMaintenance allows startup backfills to begin. It is
// idempotent so the foreground sync and its bounded fallback can coordinate
// completion without assigning separate gate ownership.
func (e *Engine) ReleaseStartupMaintenance() {
if e.startupMaintenanceReady == nil {
return
}
e.startupMaintenanceOnce.Do(func() {
close(e.startupMaintenanceReady)
})
}
// RunStartupMaintenance waits for the daemon-launching foreground sync, then
// runs work under the same mutex used by sync and resync database swaps.
func (e *Engine) RunStartupMaintenance(
ctx context.Context, work func() error,
) error {
if e.startupMaintenanceReady != nil {
select {
case <-ctx.Done():
return ctx.Err()
case <-e.startupMaintenanceReady:
}
}
return e.RunExclusive(work)
}
// RunStartupSyncFallback performs the local sync that a daemon-launching
// client was expected to request. The caller invokes it after a bounded grace
// period when that request may have been abandoned.
func (e *Engine) RunStartupSyncFallback(
ctx context.Context, onProgress ProgressFunc,
) (stats SyncStats, ran bool, err error) {
if e.refuseWriteInForceParse("RunStartupSyncFallback") {
return SyncStats{}, false, nil
}
e.syncMu.Lock()
// Defers run LIFO: Unlock runs before emit.
defer func() {
if stats.Synced > 0 {
e.emit("sync")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
if e.startupMaintenanceReady != nil {
select {
case <-e.startupMaintenanceReady:
return SyncStats{}, false, nil
default:
}
}
stats, err = e.syncThenRunLocked(
ctx, false, onProgress, func(bool) error { return nil },
)
// Once the fallback has attempted the skipped startup sync, maintenance
// must be allowed to proceed even if that attempt was interrupted.
e.ReleaseStartupMaintenance()
return stats, true, err
}
// RunExclusive runs DB-writing work while holding the same mutex used by local
// sync and resync operations. Use this for daemon-owned maintenance operations
// that must serialize with sync but should not force a local sync first.
func (e *Engine) RunExclusive(work func() error) error {
if e.refuseWriteInForceParse("RunExclusive") {
return errors.New(
"RunExclusive refused on report-only parse-diff engine",
)
}
e.syncMu.Lock()
defer e.syncMu.Unlock()
return work()
}
// SyncAll discovers and syncs all session files from all agents.
func (e *Engine) SyncAll(
ctx context.Context, onProgress ProgressFunc,
) (stats SyncStats) {
if e.refuseWriteInForceParse("SyncAll") {
return SyncStats{}
}
e.syncMu.Lock()
// Defers run LIFO: Unlock runs before the emit closure so
// Emitter implementations cannot widen the syncMu critical
// section or deadlock by re-entering sync code.
defer func() {
if stats.Synced > 0 {
e.emit("sessions")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
stats = e.syncAllLocked(
ctx, onProgress, time.Time{}, nil, syncWriteDefault, true, false,
)
return
}
// SyncAllAfterWatcherOverflow performs a full discovery pass after the watcher
// coalesced too many distinct paths to retain them individually. A routine
// SyncAll is insufficient here: the discarded paths may be the only signal for
// same-stat rewrites. Clear event-sensitive trust and force every discovered
// file through its authoritative parse path before rebuilding those caches.
func (e *Engine) SyncAllAfterWatcherOverflow(
ctx context.Context, onProgress ProgressFunc,
) (stats SyncStats) {
if e.refuseWriteInForceParse("SyncAllAfterWatcherOverflow") {
return SyncStats{}
}
e.syncMu.Lock()
defer func() {
if stats.Synced > 0 {
e.emit("sessions")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
e.clearWatcherOverflowCaches()
stats = e.syncAllLocked(
ctx, onProgress, time.Time{}, nil, syncWriteDefault, true, true,
)
return
}
// SyncAllSince syncs only files whose mtime is at or after
// the given cutoff time. Use a zero time to sync everything
// (equivalent to SyncAll). The cutoff is applied after
// discovery; directory traversal still walks all session
// directories. Typical callers pass a small safety margin
// behind the last successful sync start to avoid missing
// files that were being written during a prior sync.
func (e *Engine) SyncAllSince(
ctx context.Context, since time.Time, onProgress ProgressFunc,
) (stats SyncStats) {
if e.refuseWriteInForceParse("SyncAllSince") {
return SyncStats{}
}
e.syncMu.Lock()
defer func() {
if stats.Synced > 0 {
e.emit("sessions")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
stats = e.syncAllLocked(
ctx, onProgress, since, nil, syncWriteDefault, true, false,
)
return
}
// SyncRootsSince syncs only configured roots matching the given
// root paths whose mtimes are at or after the given cutoff. Passing
// "all" in roots is equivalent to SyncAllSince.
func (e *Engine) SyncRootsSince(
ctx context.Context, roots []string, since time.Time,
onProgress ProgressFunc,
) (stats SyncStats) {
if e.refuseWriteInForceParse("SyncRootsSince") {
return SyncStats{}
}
e.syncMu.Lock()
defer func() {
if stats.Synced > 0 {
e.emit("sessions")
}
}()
defer e.syncMu.Unlock()
defer e.clearCurrentProgress()
scope := newRootSyncScope(roots)
stats = e.syncAllLocked(
ctx, onProgress, since, scope, syncWriteDefault, scope == nil, false,
)
return
}
type rootSyncScope struct {
roots []string
}
func newRootSyncScope(roots []string) *rootSyncScope {
if len(roots) == 0 {
return nil
}
scope := &rootSyncScope{}
for _, root := range roots {
if root == "" {
continue
}
if root == "all" {
return nil
}
scope.roots = append(scope.roots, cleanRootPath(root))
}
if len(scope.roots) == 0 {
return nil
}
return scope
}
func (s *rootSyncScope) includes(dir string) bool {
if s == nil {
return true
}
if dir == "" {
return false
}
cleaned := cleanRootPath(dir)
return slices.ContainsFunc(s.roots, func(root string) bool {
return samePathOrDescendant(cleaned, root)
})
}
func (s *rootSyncScope) includesAny(dirs []string) bool {
if s == nil {
return true
}
return slices.ContainsFunc(dirs, s.includes)
}
func cleanRootPath(path string) string {
cleaned := filepath.Clean(path)
abs, err := filepath.Abs(cleaned)
if err != nil {
return cleaned
}
return abs
}
func samePathOrDescendant(path, root string) bool {
if path == root {
return true
}
rel, err := filepath.Rel(root, path)
if err != nil {
return false
}
return rel != ".." &&
!strings.HasPrefix(rel, ".."+string(filepath.Separator))
}
func (e *Engine) syncAllLocked(
ctx context.Context, onProgress ProgressFunc, since time.Time,
scope *rootSyncScope, writeMode syncWriteMode, recordSyncState bool,
forceDiscoveredFiles bool,
) (stats SyncStats) {
if ctx.Err() != nil {
return SyncStats{Aborted: true}
}
if recordSyncState {
e.recordSyncStarted()
}
e.phaseStats.Reset()
e.resetS3CodexIndexCache()
e.anomalies.reset()
// Fold the per-run anomaly accumulator into the returned stats on
// every exit path so the CLI sync summary can surface them.
defer func() { e.anomalies.applyTo(&stats) }()
t0 := time.Now()
// Report discovery as its own phase before the walk. syncAllLocked
// visits every source before emitting any syncing progress, and on a
// large archive that walk takes minutes; without this marker a
// daemon-driven `agentsview sync` shows no terminal feedback until the
// walk and DB-backed count both finish. The resync path emits the same
// marker, so its progress printer dedupes on the matching Detail.
e.reportProgress(onProgress, Progress{
Phase: PhaseDiscovering,
Detail: "Discovering sessions",
})
// Container states must be captured BEFORE discovery lists any session
// rows, so a promoted state can never be newer than the discovered
// session set (see captureSQLiteContainerStates).
preContainerStates := e.captureSQLiteContainerStates()
var all []parser.DiscoveredFile
counts := make(map[parser.AgentType]int)
providerFound, providerFailures := e.discoverProviderSources(ctx, scope)
for _, file := range providerFound {
counts[file.Agent]++
}
all = append(all, providerFound...)
verifiedPass := uint64(0)
verifiedPassFinished := false
if scope == nil && e.pathRewriter == nil {
verifiedPass = e.beginVerifiedSourcePass()
e.markVerifiedDiscoveredSources(providerFound)
defer func() {
if !verifiedPassFinished {
e.finishVerifiedSourcePass(verifiedPass, false)
}
}()
}
// Begin gate bookkeeping from the pre-filter discovery set: promotion
// needs a completion for every discovered session, so a cutoff-filtered
// pass must stay unpromotable (see opencode_container_gate.go).
e.beginSQLiteContainerPass(providerFound, preContainerStates)
quickSyncCutoff := !since.IsZero()
if quickSyncCutoff {
all = e.dedupeClaudeDiscoveredFiles(all)
// A Codex UUID can exist as both a live dated transcript and a flat
// archived copy. The provider's discovery deduplicates them to the
// preferred (live) layout, but the mtime cutoff filter runs before the
// engine's own dedup, so a changed archived copy that is newer than the
// cutoff would be lost behind an older live copy that the cutoff drops.
// Re-expand to every on-disk duplicate before filtering so the cutoff
// sees each copy's real mtime; the quick-sync dedupe below then keeps
// the newest surviving duplicate before falling back to normal layout
// preference.
all = e.expandCodexProviderDuplicates(all, scope)
all = e.filterFilesByMtime(ctx, all, since)
}
if quickSyncCutoff {
all = dedupeDiscoveredFilesPreferNewestCodex(all)
} else {
all = dedupeDiscoveredFiles(all)
}
all = e.dedupeClaudeDiscoveredFiles(all)
all = e.filterShadowedLegacyKiroFiles(all)
if forceDiscoveredFiles {
for i := range all {
all[i].ForceParse = true
}
}
verbose := onProgress == nil
// Always log discovery timing: this is the only window into the
// otherwise-silent provider walk, which dominates resync wall-clock
// on large archives. Suppressing it behind verbose hid that cost on
// the daemon resync and interactive sync paths (both pass onProgress).
log.Printf(
"discovered %d files (%d claude, %d codex, %d copilot, %d gemini, %d cursor, %d amp, %d zencoder, %d iflow, %d vscode-copilot, %d visualstudio-copilot, %d pi, %d omp, %d kiro, %d zed, %d vibe) in %s",
len(all),
counts[parser.AgentClaude],
counts[parser.AgentCodex],
counts[parser.AgentCopilot],
counts[parser.AgentGemini],
counts[parser.AgentCursor],
counts[parser.AgentAmp],
counts[parser.AgentZencoder],
counts[parser.AgentIflow],
counts[parser.AgentVSCodeCopilot],
counts[parser.AgentVSCopilot],
counts[parser.AgentPi],
counts[parser.AgentOMP],
counts[parser.AgentKiro],
counts[parser.AgentZed],
counts[parser.AgentVibe],
time.Since(t0).Round(time.Millisecond),
)
progressTotal := len(all)
tDBCount := time.Now()
dbBackedCount := e.countDBBackedSessions(ctx, scope)
progressTotal += dbBackedCount
log.Printf(
"counted %d db-backed sessions in %s",
dbBackedCount, time.Since(tDBCount).Round(time.Millisecond),
)
e.reportProgress(onProgress, Progress{
Phase: PhaseSyncing,
SessionsTotal: progressTotal,
})
nonContainerDiscovered := 0
for _, f := range all {
if !isOpenCodeFormatStorageAgent(f.Agent) {
nonContainerDiscovered++
}
}
tWorkers := time.Now()
results := e.startWorkers(ctx, all)
stats = e.collectAndBatch(
ctx, results, len(all), progressTotal, onProgress, writeMode,
)
for range providerFailures {
stats.RecordFailed()
}
// Discovery failures cannot be attributed to a provider here, so any
// failure conservatively blocks every container promotion this pass.
// Only unscoped passes discovered every root, so only they may drop
// trusted entries for containers that produced no sources.
e.finishSQLiteContainerPass(
stats.Aborted || ctx.Err() != nil || providerFailures > 0,
scope == nil,
)
if verifiedPass != 0 {
e.finishVerifiedSourcePass(
verifiedPass,
!stats.Aborted && ctx.Err() == nil && providerFailures == 0,
)
verifiedPassFinished = true
}
stats.nonContainerDiscovered = nonContainerDiscovered
if verbose {
log.Printf(
"file sync: %d synced, %d skipped in %s",
stats.Synced, stats.Skipped,
time.Since(tWorkers).Round(time.Millisecond),
)
}
// If cancelled (either collectAndBatch set Aborted, or
// context was cancelled after the loop with no file-backed
// sessions), return partial stats without running further
// phases or mutating state. Don't update lastSync or
// lastSyncStats so the UI still reflects the last
// completed sync.
if stats.Aborted || ctx.Err() != nil {
stats.Aborted = true
return stats
}
dbProgress := Progress{
Phase: PhaseSyncing,
SessionsTotal: progressTotal,
SessionsDone: stats.filesDiscovered,
MessagesIndexed: stats.messagesIndexed,
}
advanceDBProgress := func(total int, pending []pendingWrite) {
if total == 0 {
return
}
dbProgress.SessionsDone += total
for _, pw := range pending {
dbProgress.MessagesIndexed += len(pw.msgs)
}
stats.messagesIndexed = dbProgress.MessagesIndexed
e.reportProgress(onProgress, dbProgress)
}
if ctx.Err() != nil {
stats.Aborted = true
return stats
}
// OpenCode-format sessions (OpenCode and its Kilo and MiMoCode
// forks) are provider-authoritative: discovery and parsing flow
// through the provider facade in the file-sync phase above, so no
// dedicated DB-backed sync pass is needed here.
// Sync Warp, Forge, Piebald, and ZCode sessions. These are provider-authoritative
// DB-backed providers: a shared SQLite DB hosts every session, so the
// provider facade enumerates sources and parses only the changed ones.
if scope.includesAny(e.agentDirs[parser.AgentWarp]) {
if e.syncProviderDBBackedAgent(
ctx, parser.AgentWarp, "warp",
writeMode, verbose, scope, &stats, advanceDBProgress,
) {
stats.Aborted = true
return stats
}
}
if scope.includesAny(e.agentDirs[parser.AgentForge]) {
if e.syncProviderDBBackedAgent(
ctx, parser.AgentForge, "forge",
writeMode, verbose, scope, &stats, advanceDBProgress,
) {
stats.Aborted = true
return stats
}
}
if scope.includesAny(e.agentDirs[parser.AgentPiebald]) {
if e.syncProviderDBBackedAgent(
ctx, parser.AgentPiebald, "piebald",
writeMode, verbose, scope, &stats, advanceDBProgress,
) {
stats.Aborted = true
return stats
}
}
if scope.includesAny(e.agentDirs[parser.AgentZCode]) {
if e.syncProviderDBBackedAgent(
ctx, parser.AgentZCode, "zcode",
writeMode, verbose, scope, &stats, advanceDBProgress,
) {
stats.Aborted = true
return stats
}
}
// Link subagent child sessions to their parents after all DB-backed
// agent writes (including provider-authoritative Forge, Piebald, and ZCode).
// LinkSubagentSessions is idempotent — its WHERE filter and partial index
// make it a cheap no-op when nothing new was written — so no guard is
// needed.
if err := e.db.LinkSubagentSessions(); err != nil {
log.Printf("link subagent sessions: %v", err)
}
tPersist := time.Now()
skipCount := e.persistSkipCache()
if verbose {
log.Printf(
"persist skip cache (%d entries): %s",
skipCount,
time.Since(tPersist).Round(time.Millisecond),
)
}
e.reportProgress(onProgress, Progress{
Phase: PhaseDone,
SessionsTotal: progressTotal,
SessionsDone: progressTotal,
MessagesIndexed: stats.messagesIndexed,
})
// Store the anomaly-folded stats so LastSyncStats (UI) matches the
// value returned to the CLI summary. The deferred applyTo only reads
// the accumulator, so folding a separate copy here does not
// double-count.
persisted := stats
e.anomalies.applyTo(&persisted)
e.mu.Lock()
e.lastSync = time.Now()
e.lastSyncStats = persisted
e.mu.Unlock()
if recordSyncState && providerFailures == 0 {
e.recordSyncFinished()
}
// Emission happens in SyncAll / SyncAllSince after syncMu is
// released; syncAllLocked runs under the caller's lock.
return stats
}
// slowProviderDiscoveryThreshold is the per-provider discovery duration above
// which discovery timing is logged. Most providers finish in well under a
// millisecond; a provider over this bound is doing real per-source work worth
// surfacing.
const slowProviderDiscoveryThreshold = 100 * time.Millisecond
// discoverProviderSources runs full-sync discovery through the provider facade
// for every concrete provider that is authoritative. It is the sole on-disk
// discovery path: every file-based agent owns discovery through its provider.
func (e *Engine) discoverProviderSources(
ctx context.Context,
scope *rootSyncScope,
) ([]parser.DiscoveredFile, int) {
var files []parser.DiscoveredFile
var failures int
agents := make([]parser.AgentType, 0, len(e.providerFactories))
for agent := range e.providerFactories {
agents = append(agents, agent)
}
slices.SortFunc(agents, func(a, b parser.AgentType) int {
return strings.Compare(string(a), string(b))
})
for _, agentType := range agents {
mode := e.providerMigrationModes[agentType]
if mode != parser.ProviderMigrationProviderAuthoritative {
continue
}
roots := e.agentDirs[agentType]
if len(roots) == 0 {
continue
}
filteredRoots := make([]string, 0, len(roots))
for _, root := range roots {
if scope.includes(root) {
filteredRoots = append(filteredRoots, root)
}
}
if len(filteredRoots) == 0 {
continue
}
factory, ok := e.providerFactories[agentType]
if !ok || factory == nil {
continue
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: filteredRoots,
Machine: e.machine,
})
tDiscover := time.Now()
sources, err := provider.Discover(ctx)
// Log only providers whose discovery is slow enough to matter, so a
// single pathological provider (e.g. a per-source map rebuild) stands
// out instead of hiding inside the aggregate discovery timing.
if d := time.Since(tDiscover); d >= slowProviderDiscoveryThreshold {
log.Printf(
"discovery: %s returned %d sources in %s",
agentType, len(sources), d.Round(time.Millisecond),
)
}
if err != nil {
log.Printf("%s provider discovery: %v", agentType, err)
failures++
continue
}
currentSources := providerSourcePathSet(sources)
forceParseSources := map[string]struct{}{}
if agentType == parser.AgentVSCopilot {
missingSources, forceSources :=
e.visualStudioCopilotMissingVS2026PollSources(
ctx, provider, filteredRoots, currentSources,
)
sources = append(sources, missingSources...)
maps.Copy(forceParseSources, forceSources)
}
forceParseSource := func(sourcePath string) bool {
_, ok := forceParseSources[filepath.Clean(sourcePath)]
return ok
}
// Forge, Piebald, Warp, and ZCode are DB-backed providers: a shared SQLite
// DB hosts every session. Full-sync change detection and counting
// run through their dedicated provider-driven DB sync phase
// (syncProviderDBBacked), not the per-source discovery list, so a
// full sync re-counts only changed sessions exactly as the legacy
// path did. The watcher path still classifies their changes through
// classifyProviderChangedPath.
if processFileUsesProvider(agentType) {
continue
}
def := provider.Definition()
for _, source := range sources {
sourcePath := providerDiscoveredPath(source)
if sourcePath == "" {
continue
}
agent := source.Provider
if agent == "" {
agent = def.Type
}
sourceCopy := source
discovered := parser.DiscoveredFile{
Path: sourcePath,
Project: source.ProjectHint,
Agent: agent,
ProviderSource: &sourceCopy,
ProviderProcess: true,
}
if forceParseSource(sourcePath) {
discovered.ForceParse = true
}
// S3-aware source sets carry the durable object metadata in the
// Opaque payload. Thread it into the DiscoveredFile so the S3 sync
// path (object fetch, fingerprinting, machine-ID namespacing) and the
// freshness/dedup/mtime-cutoff logic see the same source identity the
// legacy s3:// discovery emitted directly. Providers read local files,
// so clear ProviderProcess for s3:// objects: processProviderFile must
// decline them so they route through processS3Session rather than the
// provider Fingerprint/Parse path, which cannot read a remote object.
if s3, ok := source.Opaque.(parser.S3DiscoveredSource); ok {
discovered.Machine = s3.Machine
discovered.SourceSize = s3.Size
discovered.SourceMtime = s3.MtimeNS
discovered.SourceFingerprint = s3.Fingerprint
discovered.ProviderProcess = false
if discovered.Project == "" {
discovered.Project = s3.Project
}
}
files = append(files, discovered)
}
}
return files, failures
}
func providerSourcePathSet(sources []parser.SourceRef) map[string]struct{} {
seen := make(map[string]struct{}, len(sources))
for _, source := range sources {
path := providerDiscoveredPath(source)
if path == "" {
continue
}
seen[filepath.Clean(path)] = struct{}{}
}
return seen
}
func (e *Engine) visualStudioCopilotMissingVS2026PollSources(
ctx context.Context,
provider parser.Provider,
roots []string,
currentSources map[string]struct{},
) ([]parser.SourceRef, map[string]struct{}) {
watchRoots := providerChangedPathWatchRoots(ctx, provider, roots)
var out []parser.SourceRef
seenHints := make(map[string]struct{})
forceParseSources := make(map[string]struct{})
for _, watchRoot := range watchRoots {
hints, err := e.db.ListStoredSourcePathHints(
string(parser.AgentVSCopilot), []string{watchRoot},
)
if err != nil {
log.Printf(
"%s provider poll stored hints: %v",
parser.AgentVSCopilot, err,
)
continue
}
for _, hint := range hints {
hint = filepath.Clean(hint)
if _, seen := seenHints[hint]; seen {
continue
}
seenHints[hint] = struct{}{}
container, conversationID, ok :=
parser.SplitVisualStudioCopilotVirtualPath(hint)
if !ok ||
!parser.IsVisualStudioCopilotVS2026SessionPath(container) {
continue
}
if _, ok := currentSources[hint]; ok {
continue
}
if current, ok := e.visualStudioCopilotCurrentPollSource(
ctx, provider, conversationID,
); ok {
sourcePath := providerDiscoveredPath(current)
if sourcePath == "" {
continue
}
path := filepath.Clean(sourcePath)
forceParseSources[path] = struct{}{}
if _, exists := currentSources[path]; !exists {
currentSources[path] = struct{}{}
out = append(out, current)
}
continue
}
if !visualStudioCopilotVS2026PollCanTombstone(
roots, container,
) {
continue
}
tombstones, err := provider.SourcesForChangedPath(
ctx,
parser.ChangedPathRequest{
Path: hint,
EventKind: "remove",
WatchRoot: watchRoot,
StoredSourcePaths: []string{hint},
},
)
if err != nil {
log.Printf(
"%s provider poll tombstone: %v",
parser.AgentVSCopilot, err,
)
continue
}
for _, tombstone := range tombstones {
sourcePath := providerDiscoveredPath(tombstone)
if sourcePath == "" {
continue
}
path := filepath.Clean(sourcePath)
if _, exists := currentSources[path]; exists {
continue
}
currentSources[path] = struct{}{}
forceParseSources[path] = struct{}{}
out = append(out, tombstone)
}
}
}
return out, forceParseSources
}
func visualStudioCopilotVS2026PollCanTombstone(
roots []string,
container string,
) bool {
if container == "" {
return false
}
container = filepath.Clean(container)
if parser.IsRegularFile(container) {
return false
}
if !reachableDir(filepath.Dir(container)) {
return false
}
return slices.ContainsFunc(roots, func(root string) bool {
root = filepath.Clean(root)
return samePathOrDescendant(container, root) && reachableDir(root)
})
}
func reachableDir(path string) bool {
info, err := os.Stat(path)
return err == nil && info != nil && info.IsDir()
}
func (e *Engine) visualStudioCopilotCurrentPollSource(
ctx context.Context,
provider parser.Provider,
conversationID string,
) (parser.SourceRef, bool) {
current, ok, err := provider.FindSource(
ctx,
parser.FindSourceRequest{
RawSessionID: conversationID,
RequireFreshSource: true,
},
)
if err != nil {
log.Printf(
"%s provider poll source lookup: %v",
parser.AgentVSCopilot, err,
)
return parser.SourceRef{}, false
}
return current, ok
}
// expandCodexProviderDuplicates re-adds the on-disk duplicate paths of each
// discovered Codex source. The provider deduplicates a UUID's live and archived
// copies to the preferred layout at discovery time; this restores the dropped
// duplicates (scoped to the configured roots) so an mtime cutoff filter can
// judge each copy on its own mtime, matching the legacy discover-then-filter
// order. Non-Codex files and Codex files without a UUID-shaped name pass through
// unchanged. Duplicates are keyed by path so nothing is added twice.
func (e *Engine) expandCodexProviderDuplicates(
files []parser.DiscoveredFile, scope *rootSyncScope,
) []parser.DiscoveredFile {
pather := e.codexUUIDPathLister(scope)
if pather == nil {
return files
}
seen := make(map[string]struct{}, len(files))
for _, f := range files {
seen[string(f.Agent)+"\x00"+filepath.Clean(f.Path)] = struct{}{}
}
out := files
for _, f := range files {
if f.Agent != parser.AgentCodex {
continue
}
uuid := parser.CodexSessionUUIDFromFilename(filepath.Base(f.Path))
if uuid == "" {
continue
}
for _, dup := range pather(uuid) {
key := string(parser.AgentCodex) + "\x00" + filepath.Clean(dup)
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
out = append(out, parser.DiscoveredFile{
Path: dup,
Agent: parser.AgentCodex,
ProviderProcess: true,
ProviderSource: e.codexPinnedProviderSource(dup),
})
}
}
return out
}
// codexUUIDPathLister returns a function that lists every on-disk Codex
// transcript path for a UUID under the in-scope roots, or nil when the Codex
// provider is unavailable. It scopes a single provider to the in-scope roots so
// the returned paths cover both the live dated and flat archived copies of a
// duplicated UUID, including duplicates that share one root.
func (e *Engine) codexUUIDPathLister(
scope *rootSyncScope,
) func(string) []string {
factory, ok := e.providerFactories[parser.AgentCodex]
if !ok || factory == nil {
return nil
}
roots := make([]string, 0, len(e.agentDirs[parser.AgentCodex]))
for _, root := range e.agentDirs[parser.AgentCodex] {
if root == "" || !scope.includes(root) {
continue
}
roots = append(roots, root)
}
if len(roots) == 0 {
return nil
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: roots,
Machine: e.machine,
})
lister, ok := provider.(interface {
AllSourcePathsForUUID(string) []string
})
if !ok {
return nil
}
return lister.AllSourcePathsForUUID
}
// recordSyncStarted persists the start time of a sync run
// into pg_sync_state. Callers use this to compute mtime
// cutoffs for future quick incremental syncs.
func (e *Engine) recordSyncStarted() {
if e.ephemeral {
return
}
ts := time.Now().UTC().Format(time.RFC3339Nano)
if err := e.db.SetSyncState(syncStateStartedAt, ts); err != nil {
log.Printf("persist sync start time: %v", err)
}
}
// recordSyncFinished persists the finish time of a completed
// sync run. Only called on successful completion (not on
// cancellation or abort).
func (e *Engine) recordSyncFinished() {
if e.ephemeral {
return
}
ts := time.Now().UTC().Format(time.RFC3339Nano)
if err := e.db.SetSyncState(syncStateFinishedAt, ts); err != nil {
log.Printf("persist sync finish time: %v", err)
}
}
// filterFilesByMtime returns only files whose mtime is at or
// after the given cutoff. Files that can't be stat'd are kept
// (so errors surface in the worker rather than being silently
// dropped). The cost is one stat per file — acceptable for
// polling use cases where most files will be skipped.
func (e *Engine) filterFilesByMtime(
ctx context.Context,
files []parser.DiscoveredFile,
cutoff time.Time,
) []parser.DiscoveredFile {
cutoffNs := cutoff.UnixNano()
out := files[:0]
codexIndexRefresh := make(map[string][]parser.DiscoveredFile)
for _, f := range files {
if f.ForceParse {
out = append(out, f)
continue
}
mtime, err := e.discoveredFileEffectiveMtime(ctx, f)
if err != nil {
out = append(out, f)
continue
}
if mtime >= cutoffNs {
out = append(out, f)
continue
}
if isS3SourcePath(f.Path) && e.s3SourceMetadataChanged(f) {
out = append(out, f)
continue
}
if f.Agent != parser.AgentCodex {
continue
}
indexNeedsRefresh := false
if isS3SourcePath(f.Path) {
indexNeedsRefresh = e.s3CodexIndexNeedsRefreshSince(
f, cutoffNs,
)
} else {
indexNeedsRefresh = e.codexIndexNeedsRefreshSince(
f.Path, cutoffNs,
)
}
if !indexNeedsRefresh {
continue
}
key := discoveredFileKey(f)
codexIndexRefresh[key] = append(codexIndexRefresh[key], f)
}
if len(codexIndexRefresh) == 0 {
return out
}
included := make(map[string]struct{}, len(out))
for _, f := range out {
included[discoveredFileKey(f)] = struct{}{}
}
for key, candidates := range codexIndexRefresh {
if _, ok := included[key]; ok {
continue
}
out = append(out, pickPreferredCodexDiscoveredFile(e.db, candidates))
}
return out
}
// discoveredFileEffectiveMtime returns the freshness timestamp used to filter a
// discovered file against an incremental-sync cutoff. For provider-sourced
// files it consults the provider's Fingerprint so composite/sibling-file
// freshness (for example a Positron session whose workspace.json changed while
// the chat transcript did not) is honored without a per-agent legacy helper.
// Files without a provider source fall back to the legacy mtime computation.
func (e *Engine) discoveredFileEffectiveMtime(
ctx context.Context, file parser.DiscoveredFile,
) (int64, error) {
// Codex is excluded from the provider-Fingerprint path on purpose. Its
// Fingerprint folds the shared session_index.jsonl mtime into every
// session's freshness (see CodexEffectiveMtime). That shared signal is
// correct for the skip cache but wrong for the incremental-sync cutoff:
// when the index changes, both the live and archived copies of a UUID
// would look fresh, defeating the per-copy mtime discrimination that
// expandCodexProviderDuplicates relies on to preserve a changed archived
// duplicate. Index refreshes are handled separately by the codexIndexRefresh
// pass in filterFilesByMtime, so codex uses its raw per-file mtime here.
if file.Agent == parser.AgentCodex {
return discoveredFileMtime(file)
}
// S3 objects are discovered through the provider facade (so they carry a
// ProviderSource), but providers read local files and cannot Fingerprint an
// s3:// URI. Routing them through providerSourceMtime below would error, and
// filterFilesByMtime keeps any file whose mtime cannot be resolved, defeating
// the incremental cutoff and reprocessing every old S3 object on each sync.
// The threaded object metadata (or a HEAD stat) gives the timestamp directly.
if isS3SourcePath(file.Path) {
return discoveredFileMtime(file)
}
// Provider-authoritative sources resolve freshness through the provider
// Fingerprint so composite provider-owned source state participates in
// incremental-sync cutoff checks.
if file.ProviderSource != nil && file.ProviderProcess {
if mtime, ok, err := e.providerSourceMtime(ctx, file); err != nil {
return 0, err
} else if ok {
return mtime, nil
}
}
return discoveredFileMtime(file)
}
// providerSourceMtime resolves a provider-sourced file's effective mtime through
// the owning provider's Fingerprint. The boolean reports whether the provider
// runtime produced a usable timestamp; a false result tells the caller to fall
// back to the legacy mtime path.
func (e *Engine) providerSourceMtime(
ctx context.Context, file parser.DiscoveredFile,
) (int64, bool, error) {
if file.ProviderSource == nil {
return 0, false, nil
}
factory, ok := e.providerFactories[file.Agent]
if !ok || factory == nil {
return 0, false, nil
}
source := *file.ProviderSource
if source.Provider != "" && source.Provider != file.Agent {
return 0, false, fmt.Errorf(
"provider source mismatch for %s: %s",
file.Agent,
source.Provider,
)
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: e.agentDirs[file.Agent],
Machine: e.machine,
})
fingerprint, err := provider.Fingerprint(ctx, source)
if err != nil {
return 0, false, err
}
if fingerprint.MTimeNS == 0 {
return 0, false, nil
}
return fingerprint.MTimeNS, true, nil
}
func discoveredFileMtime(
file parser.DiscoveredFile,
) (int64, error) {
if strings.HasPrefix(file.Path, "s3://") {
if file.SourceMtime != 0 {
return file.SourceMtime, nil
}
stat := statS3Object
switch file.Agent {
case parser.AgentClaude:
stat = statClaudeS3Session
case parser.AgentCodex:
stat = statCodexS3Session
}
obj, err := stat(file.Path)
if err != nil {
return 0, err
}
return obj.LastModified.UnixNano(), nil
}
if file.Agent == parser.AgentKiro {
if _, _, ok := parseKiroSQLiteVirtualPath(file.Path); ok {
return parser.KiroSQLiteSourceMtime(file.Path)
}
}
if isOpenCodeFormatStorageAgent(file.Agent) {
if isOpenCodeFormatSQLiteVirtualPath(file.Agent, file.Path) ||
isOpenCodeFormatStoragePath(file.Agent, file.Path) {
return openCodeFormatSourceMtime(
file.Agent, file.Path,
)
}
}
if file.Agent == parser.AgentZed {
dbPath := file.Path
if p, _, ok := parser.ParseVirtualSourcePathForBase(file.Path, "threads.db"); ok {
dbPath = p
}
return zedDBCompositeMtime(dbPath)
}
if file.Agent == parser.AgentShelley {
dbPath := file.Path
if p, _, ok := parser.ParseVirtualSourcePathForBase(file.Path, shelleyDBFile); ok {
dbPath = p
}
return shelleyDBCompositeMtime(dbPath)
}
if file.Agent == parser.AgentVSCopilot {
// Sessions are stored under a <traceFile>#<conversationID> virtual
// path; stat the physical trace so the mtime filter can drop
// conversations whose trace file is unchanged.
info, err := os.Stat(parser.ResolveSourceFilePath(file.Path))
if err != nil {
return 0, err
}
return info.ModTime().UnixNano(), nil
}
if file.Agent == parser.AgentAntigravityCLI {
info, err := parser.AntigravityCLIFileInfo(file.Path)
if err != nil {
return 0, err
}
return info.ModTime().UnixNano(), nil
}
if file.Agent == parser.AgentAntigravity {
info, err := parser.AntigravityFileInfo(file.Path)
if err != nil {
return 0, err
}
return info.ModTime().UnixNano(), nil
}
if file.Agent == parser.AgentCowork {
info, err := os.Stat(file.Path)
if err != nil {
return 0, err
}
return parser.CoworkSessionMtime(
file.Path, info.ModTime().UnixNano(),
), nil
}
if file.Agent == parser.AgentCommandCode {
info, err := os.Stat(file.Path)
if err != nil {
return 0, err
}
return commandCodeEffectiveInfo(file.Path, info).ModTime().UnixNano(), nil
}
if file.Agent == parser.AgentVibe {
info, err := os.Stat(file.Path)
if err != nil {
return 0, err
}
return vibeEffectiveInfo(file.Path, info).ModTime().UnixNano(), nil
}
if file.Agent == parser.AgentReasonix {
info, err := os.Stat(file.Path)
if err != nil {
return 0, err
}
return reasonixEffectiveInfo(file.Path, info).ModTime().UnixNano(), nil
}
info, err := os.Stat(file.Path)
if err != nil {
return 0, err
}
if file.Agent == parser.AgentCopilot {
return copilotEffectiveMtime(file.Path, info), nil
}
return info.ModTime().UnixNano(), nil
}
func (e *Engine) dedupeClaudeDiscoveredFiles(
files []parser.DiscoveredFile,
) []parser.DiscoveredFile {
byKey := make(map[string][]parser.DiscoveredFile)
sessionIDByKey := make(map[string]string)
for _, file := range files {
if file.Agent != parser.AgentClaude {
continue
}
sessionID := claudeSessionIDFromPath(file.Path)
if sessionID == "" {
continue
}
key := claudeDiscoveredFileKey(file, sessionID)
byKey[key] = append(byKey[key], file)
sessionIDByKey[key] = sessionID
}
if len(byKey) == 0 {
return files
}
preferred := make(map[string]parser.DiscoveredFile, len(byKey))
for key, candidates := range byKey {
preferred[key] = e.pickPreferredClaudeDiscoveredFile(
sessionIDByKey[key], candidates,
)
}
out := files[:0]
seen := make(map[string]struct{}, len(preferred))
for _, file := range files {
if file.Agent != parser.AgentClaude {
out = append(out, file)
continue
}
sessionID := claudeSessionIDFromPath(file.Path)
if sessionID == "" {
out = append(out, file)
continue
}
key := claudeDiscoveredFileKey(file, sessionID)
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
out = append(out, preferred[key])
}
return out
}
func claudeDiscoveredFileKey(
file parser.DiscoveredFile, sessionID string,
) string {
return discoveredFileIDPrefix(file) + "\x00" + sessionID
}
func claudeSessionIDFromPath(path string) string {
name := filepath.Base(path)
sessionID, ok := strings.CutSuffix(name, ".jsonl")
if !ok {
return ""
}
return sessionID
}
func (e *Engine) pickPreferredClaudeDiscoveredFile(
sessionID string, candidates []parser.DiscoveredFile,
) parser.DiscoveredFile {
if len(candidates) == 1 {
return candidates[0]
}
idPrefix := e.idPrefix
if isS3SourcePath(candidates[0].Path) {
idPrefix = s3SessionIDPrefix(candidates[0].Machine)
}
fullID := applyIDPrefixToID(idPrefix, sessionID)
storedPath := e.db.GetSessionFilePath(fullID)
if storedPath != "" {
for _, candidate := range candidates {
if e.effectiveSourcePath(candidate.Path) != storedPath {
continue
}
if e.claudeSourceMatchesStored(fullID, candidate) {
best := candidate
for _, competing := range candidates {
if e.effectiveSourcePath(competing.Path) == storedPath ||
!claudeCandidateHasAppendProgress(competing, candidate) {
continue
}
if preferClaudeDiscoveredFile(competing, best) {
best = competing
}
}
return best
}
}
}
best := candidates[0]
for _, candidate := range candidates[1:] {
if preferClaudeDiscoveredFile(candidate, best) {
best = candidate
}
}
return best
}
func (e *Engine) claudeSourceMatchesStored(
sessionID string, file parser.DiscoveredFile,
) bool {
size, mtime, ok := claudeDiscoveredFileSourceInfo(file)
if !ok {
return false
}
storedSize, storedMtime, ok := e.db.GetSessionFileInfo(sessionID)
if !ok {
return false
}
if storedSize != size || storedMtime != mtime {
return false
}
if file.SourceFingerprint != "" {
storedHash, ok := e.db.GetSessionFileHash(sessionID)
if !ok || storedHash != file.SourceFingerprint {
return false
}
}
return e.db.GetSessionDataVersion(sessionID) >= db.CurrentDataVersion()
}
func (e *Engine) effectiveSourcePath(path string) string {
if e.pathRewriter != nil {
return e.pathRewriter(path)
}
return path
}
func claudeCandidateHasAppendProgress(
candidate, current parser.DiscoveredFile,
) bool {
candidateSize, _, candidateOK := claudeDiscoveredFileSourceInfo(candidate)
currentSize, _, currentOK := claudeDiscoveredFileSourceInfo(current)
if !candidateOK || !currentOK {
return false
}
return candidateSize > currentSize
}
func preferClaudeDiscoveredFile(
candidate, current parser.DiscoveredFile,
) bool {
candidateSize, candidateMtime, candidateOK := claudeDiscoveredFileSourceInfo(candidate)
currentSize, currentMtime, currentOK := claudeDiscoveredFileSourceInfo(current)
switch {
case candidateOK && !currentOK:
return true
case !candidateOK && currentOK:
return false
case candidateOK && currentOK:
if candidateSize != currentSize {
return candidateSize > currentSize
}
if candidateMtime != currentMtime {
return candidateMtime > currentMtime
}
}
return candidate.Path < current.Path
}
func claudeDiscoveredFileSourceInfo(
file parser.DiscoveredFile,
) (size, mtime int64, ok bool) {
if isS3SourcePath(file.Path) {
if file.SourceMtime != 0 {
return file.SourceSize, file.SourceMtime, true
}
obj, err := statClaudeS3Session(file.Path)
if err != nil {
return 0, 0, false
}
return obj.Size, obj.LastModified.UnixNano(), true
}
info, err := os.Stat(file.Path)
if err != nil {
return 0, 0, false
}
return info.Size(), info.ModTime().UnixNano(), true
}
// zedDBCompositeMtime returns the maximum mtime across the Zed
// threads.db main file and its WAL/SHM siblings. WAL-only updates
// do not touch threads.db itself, so the composite is needed to
// detect all changes.
func zedDBCompositeMtime(dbPath string) (int64, error) {
var maxMtime int64
for _, suffix := range []string{"", "-wal", "-shm"} {
info, err := os.Stat(dbPath + suffix)
if err != nil {
continue
}
if t := info.ModTime().UnixNano(); t > maxMtime {
maxMtime = t
}
}
if maxMtime == 0 {
return 0, &os.PathError{Op: "stat", Path: dbPath, Err: os.ErrNotExist}
}
return maxMtime, nil
}
// shelleyDBCompositeMtime returns the maximum mtime across the Shelley
// shelley.db main file and its WAL/SHM siblings. The DB is WAL-mode and
// churns constantly, so WAL-only updates that do not touch shelley.db
// itself still need to be detected.
func shelleyDBCompositeMtime(dbPath string) (int64, error) {
var maxMtime int64
for _, suffix := range []string{"", "-wal", "-shm"} {
info, err := os.Stat(dbPath + suffix)
if err != nil {
continue
}
if t := info.ModTime().UnixNano(); t > maxMtime {
maxMtime = t
}
}
if maxMtime == 0 {
return 0, &os.PathError{Op: "stat", Path: dbPath, Err: os.ErrNotExist}
}
return maxMtime, nil
}
func (e *Engine) countDBBackedProgressTotal(
agent parser.AgentType, scope *rootSyncScope,
) int {
if processFileUsesProvider(agent) {
return e.countProviderDBBackedSessions(
context.Background(), agent, scope,
)
}
return 0
}
// countProviderDBBackedSessions counts every session a DB-backed provider
// (Forge, Piebald, Warp) currently exposes, via provider discovery. It feeds
// the progress total so the DB-backed family contributes the same fixed count
// the legacy countOne*Sessions helpers did.
func (e *Engine) countProviderDBBackedSessions(
ctx context.Context, agent parser.AgentType, scope *rootSyncScope,
) int {
roots := make([]string, 0, len(e.agentDirs[agent]))
for _, dir := range e.agentDirs[agent] {
if dir == "" || !scope.includes(dir) {
continue
}
roots = append(roots, dir)
}
if len(roots) == 0 {
return 0
}
factory, ok := e.providerFactories[agent]
if !ok || factory == nil {
return 0
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: roots,
Machine: e.machine,
})
sources, err := provider.Discover(ctx)
if err != nil {
log.Printf("%s provider session count: %v", agent, err)
return 0
}
return len(sources)
}
func (e *Engine) countDBBackedSessions(
ctx context.Context, scope *rootSyncScope,
) int {
if ctx.Err() != nil {
return 0
}
total := 0
for _, agent := range []parser.AgentType{
parser.AgentWarp,
parser.AgentForge,
parser.AgentPiebald,
parser.AgentZCode,
} {
total += e.countDBBackedProgressTotal(agent, scope)
}
return total
}
// syncProviderDBBacked enumerates a DB-backed provider's sources, parses only
// the changed ones through the provider facade, and returns their pending
// writes. Change detection compares the provider fingerprint mtime against the
// stored source mtime and requires the stored data version to be current,
// reproducing the legacy *PendingSessionIDs behavior.
func (e *Engine) syncProviderDBBacked(
ctx context.Context, agent parser.AgentType, scope *rootSyncScope,
) []pendingWrite {
roots := make([]string, 0, len(e.agentDirs[agent]))
for _, dir := range e.agentDirs[agent] {
if dir == "" || !scope.includes(dir) {
continue
}
roots = append(roots, dir)
}
if len(roots) == 0 {
return nil
}
factory, ok := e.providerFactories[agent]
if !ok || factory == nil {
return nil
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: roots,
Machine: e.machine,
})
sources, err := provider.Discover(ctx)
if err != nil {
log.Printf("sync %s: %v", agent, err)
return nil
}
var pending []pendingWrite
for _, source := range sources {
if ctx.Err() != nil {
break
}
fingerprint, err := provider.Fingerprint(ctx, source)
if err != nil {
log.Printf("sync %s fingerprint: %v", agent, err)
continue
}
if e.providerDBBackedSourceFresh(source, fingerprint) {
continue
}
outcome, err := provider.Parse(ctx, parser.ParseRequest{
Source: source,
Fingerprint: fingerprint,
Machine: e.machine,
})
if err != nil {
log.Printf("sync %s parse: %v", agent, err)
continue
}
for _, result := range outcome.Results {
pending = append(pending, pendingWrite{
sess: result.Result.Session,
msgs: result.Result.Messages,
usageEvents: result.Result.UsageEvents,
})
}
}
return pending
}
// providerDBBackedSourceFresh reports whether a DB-backed provider source is
// already stored at the current data version with an unchanged source mtime, so
// it can be skipped during a full sync. This is the change-detection half of
// the legacy *PendingSessionIDs helpers.
func (e *Engine) providerDBBackedSourceFresh(
source parser.SourceRef,
fingerprint parser.SourceFingerprint,
) bool {
if e.forceParse {
return false
}
if fingerprint.MTimeNS == 0 {
return false
}
lookupPath := ""
for _, candidate := range []string{
fingerprint.Key,
source.FingerprintKey,
source.DisplayPath,
source.Key,
} {
if candidate != "" {
lookupPath = candidate
break
}
}
if lookupPath == "" {
return false
}
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(lookupPath)
}
_, storedMtime, ok := e.db.GetFileInfoByPath(lookupPath)
if !ok {
return false
}
if storedMtime != fingerprint.MTimeNS {
return false
}
return e.db.GetDataVersionByPath(lookupPath) >= db.CurrentDataVersion()
}
// syncProviderDBBackedAgent runs the full-sync phase for a provider-authoritative
// DB-backed agent (Forge, Piebald, Warp). It mirrors syncOpenCodeFormatAgent:
// only changed sessions are parsed (so the second sync of unchanged data is a
// no-op), and the per-session write semantics match the legacy DB sync.
func (e *Engine) syncProviderDBBackedAgent(
ctx context.Context, agent parser.AgentType, label string,
writeMode syncWriteMode, verbose bool, scope *rootSyncScope,
stats *SyncStats,
advanceDBProgress func(total int, pending []pendingWrite),
) bool {
start := time.Now()
pending := e.syncProviderDBBacked(ctx, agent, scope)
useWorktreeResolver := agent != parser.AgentPiebald
if len(pending) > 0 {
stats.TotalSessions += len(pending)
tWrite := time.Now()
var written int
if writeMode == syncWriteBulk {
var failedWrites, cwdFiltered int
written, _, failedWrites, cwdFiltered = e.writeBatch(
pending, writeMode, true,
)
for range failedWrites {
stats.RecordFailed()
}
stats.cwdFilteredSessions += cwdFiltered
} else {
resolveWorktreeProject := e.loadWorktreeProjectResolver()
for _, pw := range pending {
if ctx.Err() != nil {
break
}
var err error
if useWorktreeResolver {
err = e.writeSessionFullWithResolver(
pw, resolveWorktreeProject,
)
} else {
err = e.writeSessionFull(pw)
}
switch {
case err == nil:
written++
case isIntentionalSessionSkip(err),
errors.Is(err, errSessionPreserved):
// Intentional skip, not a failure.
default:
stats.RecordFailed()
}
}
}
stats.RecordSynced(written)
if verbose {
log.Printf(
"%s write: %d sessions in %s",
label, len(pending),
time.Since(tWrite).Round(time.Millisecond),
)
}
}
if verbose {
log.Printf(
"%s sync: %s",
label, time.Since(start).Round(time.Millisecond),
)
}
advanceDBProgress(e.countDBBackedProgressTotal(agent, scope), pending)
return ctx.Err() != nil
}
// startWorkers fans out file processing across a worker pool
// and returns a channel of results. When ctx is cancelled,
// workers skip remaining jobs with a context error instead
// of parsing files.
func (e *Engine) startWorkers(
ctx context.Context,
files []parser.DiscoveredFile,
) <-chan syncJob {
workers := min(max(runtime.NumCPU(), 2), maxWorkers)
buffer := max(workers*2, 1)
jobs := make(chan parser.DiscoveredFile, buffer)
results := make(chan syncJob, buffer)
var wg gosync.WaitGroup
for range workers {
wg.Go(func() {
for file := range jobs {
if ctx.Err() != nil {
results <- syncJob{
processResult: processResult{
err: ctx.Err(),
},
path: file.Path,
}
continue
}
results <- syncJob{
processResult: e.processFile(ctx, file),
path: file.Path,
}
}
})
}
go func() {
for _, f := range files {
jobs <- f
}
close(jobs)
wg.Wait()
close(results)
}()
return results
}
// collectAndBatch drains the results channel, batches
// successful parses, and writes them to the database.
// When ctx is cancelled, it stops processing new results
// and returns partial stats.
func (e *Engine) collectAndBatch(
ctx context.Context,
results <-chan syncJob, total int, progressTotal int,
onProgress ProgressFunc,
writeMode syncWriteMode,
) SyncStats {
var stats SyncStats
stats.TotalSessions = total
stats.filesDiscovered = total
if progressTotal == 0 {
progressTotal = total
}
progress := Progress{
Phase: PhaseSyncing,
SessionsTotal: progressTotal,
}
var pending []pendingWrite
for i := range total {
var r syncJob
select {
case <-ctx.Done():
stats.Aborted = true
drainResults(results, total-i)
goto flush
case r = <-results:
}
if r.err != nil {
// Workers emit ctx.Err() for files skipped
// after cancellation — treat the same as the
// ctx.Done() branch above.
if ctx.Err() != nil {
stats.Aborted = true
drainResults(results, total-i-1)
goto flush
}
stats.RecordFailed()
e.noteSQLiteContainerResult(r.path, false)
if r.cacheSkip && r.mtime != 0 && !r.noCacheSkip {
e.cacheSkip(r.skipCacheKey(), r.mtime, r.sourceFingerprint)
}
log.Printf("sync error: %v", r.err)
continue
}
if r.skip {
if r.cacheSkip && r.mtime != 0 && !r.noCacheSkip {
e.cacheSkip(r.skipCacheKey(), r.mtime)
}
stats.RecordSkip()
e.noteSQLiteContainerResult(r.path, true)
progress.SessionsDone++
e.reportProgress(onProgress, progress)
continue
}
excludedSessionIDs := e.applyIDPrefixToSessionIDs(
r.excludedSessionIDs,
)
// A source with no session inside the cwd allow-list must not
// delete archived rows: its exclusions and stale-row cleanup
// would erase sessions whose replacement writes the filter
// vetoes, breaking the ingestion-only contract. Dropping the
// IDs here also keeps them out of parserExcludedIDs, so
// resync's orphan copy still restores the archived rows.
if len(excludedSessionIDs) > 0 &&
!e.sourceAllowsParserExclusions(r.processResult) {
excludedSessionIDs = nil
}
if len(excludedSessionIDs) > 0 {
if _, err := e.db.DeleteParserExcludedSessions(
excludedSessionIDs,
); err != nil {
log.Printf("delete parser-excluded sessions: %v", err)
stats.RecordFailed()
e.noteSQLiteContainerResult(r.path, false)
continue
}
stats.parserExcludedIDs = append(
stats.parserExcludedIDs,
excludedSessionIDs...,
)
}
if len(r.results) == 0 && r.incremental == nil {
if len(r.excludedSessionIDs) > 0 {
stats.filesOK++
stats.parserExcludedFiles++
}
if r.cacheSkip && !r.noCacheSkip {
e.cacheSkip(r.skipCacheKey(), r.mtime, r.sourceFingerprint)
}
e.noteSQLiteContainerResult(r.path, true)
progress.SessionsDone++
e.reportProgress(onProgress, progress)
continue
}
if r.cacheSkip {
e.clearSkip(r.skipCacheKey())
}
stats.filesOK++
// Drop sessions outside the cwd allow-list before batching so
// the sync stats can tell an intentionally filtered file apart
// from one whose sessions vanished for an unexplained reason.
// The prepareSessionWrite veto stays as the write-seam backstop.
// Filtered files are deliberately not skip-cached: a later
// allow-list change must be able to pick them up again.
allowed, vetoed := e.splitResultsByCwdFilter(r.results)
stats.cwdFilteredSessions += vetoed
// A cwd-vetoed session parsed fine but was deliberately not
// persisted, and sessions parsed at DataVersionNeedsRetry are
// deferred work — neither is verified state, so their container
// must stay untrusted. The vetoed case matters because the gate
// must never hide a filtered session from a future allow-list
// change; such containers simply keep the pre-gate re-verify
// behavior.
e.noteSQLiteContainerResult(
r.path, vetoed == 0 && len(r.retrySessionIDs) == 0,
)
if vetoed > 0 && len(allowed) == 0 {
stats.cwdFilteredFiles++
progress.SessionsDone++
e.reportProgress(onProgress, progress)
continue
}
if r.incremental != nil {
if err := e.writeIncremental(r.incremental); err != nil {
log.Printf("%v", err)
stats.RecordFailed()
continue
}
stats.RecordSynced(1)
progress.MessagesIndexed += len(
r.incremental.msgs,
)
stats.messagesIndexed = progress.MessagesIndexed
} else {
for _, pr := range allowed {
pending = append(pending, pendingWrite{
sess: pr.Session,
msgs: pr.Messages,
usageEvents: pr.UsageEvents,
needsRetry: r.needsRetryForSession(pr.Session.ID),
forceReplace: r.forceReplace,
storageTrustPath: r.storageTrustPath,
storageTrustState: r.storageTrustState,
storageTrustSnap: r.storageTrustSnap,
})
}
// A Kiro SQLite store is discovered as one container source
// but fans out into one session per row, so `total` counted it
// as a single file. Add the extra sessions it produced to keep
// TotalSessions a session count, matching the per-session tally
// the legacy syncKiroSQLite phase reported. A zero-session
// container short-circuits at the empty-result branch above and
// stays counted as one discovered source, consistent with how
// every other zero-session file is tallied.
if len(r.results) > 1 &&
filepath.Base(r.path) == kiroSQLiteDBName {
stats.TotalSessions += len(r.results) - 1
}
}
if len(pending) >= batchSize {
writtenSessions, writtenMessages, failedWrites, cwdFiltered :=
e.writeBatch(pending, writeMode, false)
stats.RecordSynced(writtenSessions)
for range failedWrites {
stats.RecordFailed()
}
// Batch write failures cannot be attributed to individual
// sessions, so they block every container promotion this pass.
if failedWrites > 0 {
e.poisonSQLiteContainerPass()
}
e.promoteOpenCodeStorageTrustAfterWrite(
pending, writtenSessions, failedWrites, cwdFiltered,
)
stats.cwdFilteredSessions += cwdFiltered
progress.MessagesIndexed += writtenMessages
stats.messagesIndexed = progress.MessagesIndexed
pending = pending[:0]
}
progress.SessionsDone++
e.reportProgress(onProgress, progress)
}
flush:
if len(pending) > 0 {
writtenSessions, writtenMessages, failedWrites, cwdFiltered :=
e.writeBatch(pending, writeMode, false)
stats.RecordSynced(writtenSessions)
for range failedWrites {
stats.RecordFailed()
}
if failedWrites > 0 {
e.poisonSQLiteContainerPass()
}
e.promoteOpenCodeStorageTrustAfterWrite(
pending, writtenSessions, failedWrites, cwdFiltered,
)
stats.cwdFilteredSessions += cwdFiltered
progress.MessagesIndexed += writtenMessages
stats.messagesIndexed = progress.MessagesIndexed
}
// Link subagent child sessions to their parents via
// tool_calls.subagent_session_id references. Run once
// after all batches to avoid repeated full-table scans.
if err := e.db.LinkSubagentSessions(); err != nil {
log.Printf("link subagent sessions: %v", err)
}
// PhaseDone is emitted by syncAllLocked after DB-backed
// agents finish, so this stage stays in PhaseSyncing.
return stats
}
// drainResults consumes remaining items from the results
// channel so that worker goroutines can exit and be collected.
func drainResults(results <-chan syncJob, remaining int) {
for range remaining {
<-results
}
}
// incrementalUpdate holds the delta produced by an
// incremental JSONL parse, used to partially update the
// session row without overwriting unrelated columns.
type incrementalUpdate struct {
sessionID string
project string
machine string
cwd string
msgs []parser.ParsedMessage
links []parser.ClaudeSubagentLink
endedAt time.Time
terminationStatus *string
msgCount int // total (old + new)
userMsgCount int // total (old + new)
fileSize int64
fileMtime int64
fileHash string
nextOrdinal int
lastEntryUUID string
totalOutputTokens int // absolute (old + new)
peakContextTokens int // absolute max(old, new)
hasTotalOutputTokens bool
hasPeakContextTokens bool
}
// sessionParseError is a per-session parse failure inside a shared
// SQLite store (OpenCode, Zed, Kiro), where one file path fans out to
// many sessions and a single bad payload must not fail the whole db.
type sessionParseError struct {
sessionID string // raw parser-side ID, no engine prefix
virtualPath string // dbPath#rawID source path
err error
}
type processResult struct {
results []parser.ParseResult
excludedSessionIDs []string
// sessionErrs carries per-session parse failures from the
// shared-db fan-out loops. Normal sync logs and skips these;
// parse-diff (forceParse) surfaces them as DiffParseError report
// entries so --fail-on-change cannot pass over a session the
// current binary failed to parse.
sessionErrs []sessionParseError
skip bool
mtime int64
err error
incremental *incrementalUpdate
cacheSkip bool
// sourceFingerprint carries S3 object fingerprints into
// skip-cache writes so same-mtime object rewrites do not stay
// hidden behind a cached parse failure or non-interactive result.
sourceFingerprint string
// noCacheSkip suppresses skip-cache recording even when cacheSkip is set
// for the agent. Read/scan failures and incomplete append boundaries are
// transient: a readability fix or completed record may retain the same file
// mtime, so caching either result would silently skip later work instead of
// retrying it.
noCacheSkip bool
needsRetry bool
// forceReplace requests full message replacement on write,
// even when the existing rows would otherwise be left in
// place. Set when a fall-through to full parse is recovering
// from stale stored rows, such as an atomic file replacement
// or cross-sync streaming split. In those cases the parsed
// messages can reuse existing ordinals, so the default
// append-only writeMessages would silently drop the rewrite.
forceReplace bool
cacheKey string
// retrySessionIDs carries provider per-result data-version state.
// Legacy parsers use needsRetry as a source-wide fallback.
retrySessionIDs map[string]bool
// suppressPresenceSweep marks an incomplete source result where
// missing stored sessions are expected rather than parser drift.
suppressPresenceSweep bool
// storageTrustPath/State/Snap carry an OpenCode-family storage
// session's pre-parse stat signature and invalidation snapshot to
// the write path, which promotes it once the session's batch is
// confirmed fully written (see opencode_storage_gate.go). Empty for
// everything else.
storageTrustPath string
storageTrustState string
storageTrustSnap storageTrustSnapshot
}
func (r processResult) needsRetryForSession(sessionID string) bool {
if r.retrySessionIDs != nil {
return r.retrySessionIDs[sessionID]
}
return r.needsRetry
}
func (r processResult) suppressesPresenceSweepForRetry() bool {
return r.retrySessionIDs == nil && r.needsRetry
}
func (e *Engine) processFile(
ctx context.Context,
file parser.DiscoveredFile,
) processResult {
if res, ok := e.processProviderFile(ctx, file); ok {
return res
}
// Every registered agent is provider-authoritative, so processProviderFile
// owns all local-file processing. The only sources that fall through are
// s3:// Claude/Codex objects, which bypass the provider (its source sets
// read local files) and use the legacy S3 sync path. Anything else is an
// unrecognized agent type.
if !strings.HasPrefix(file.Path, "s3://") {
return processResult{
err: fmt.Errorf("unknown agent type: %s", file.Agent),
}
}
if file.SourceMtime == 0 {
obj, err := statS3SourceObject(file)
if err != nil {
return processResult{
err: fmt.Errorf("stat %s: %w", file.Path, err),
}
}
file.SourceSize = obj.Size
file.SourceMtime = obj.LastModified.UnixNano()
file.SourceFingerprint = obj.Fingerprint
}
info, err := s3SourceFileInfo(file)
if err != nil {
return processResult{
err: fmt.Errorf("stat %s: %w", file.Path, err),
}
}
// Capture mtime once from the initial stat so all downstream cache
// operations use a consistent value.
mtime := info.ModTime().UnixNano()
cacheSkip := e.shouldCacheSkip(file)
sourceFingerprint := s3SourceFingerprint(file)
// Skip files cached from a previous sync whose mtime and source
// fingerprint are unchanged.
if cacheSkip && !e.forceParse && !file.ForceParse { // parse-diff: ignore the skip cache
if e.shouldUseCachedSkip(file, mtime, sourceFingerprint) {
if e.pathNeedsCachedSkipBypass(file.Path) {
e.clearSkip(file.Path)
} else {
return processResult{
skip: true,
mtime: mtime,
cacheSkip: true,
}
}
}
}
var res processResult
switch file.Agent {
case parser.AgentClaude, parser.AgentCodex:
res = e.processS3Session(ctx, file, info)
default:
res = processResult{
err: fmt.Errorf("unsupported s3 agent type: %s", file.Agent),
}
}
res.cacheSkip = cacheSkip
res.mtime = mtime
res.sourceFingerprint = sourceFingerprint
return res
}
func (e *Engine) shouldUseCachedSkip(
file parser.DiscoveredFile, mtime int64, sourceFingerprint string,
) bool {
e.skipMu.RLock()
cachedMtime, cached := e.skipCache[file.Path]
cachedFingerprint := ""
if e.skipFingerprints != nil {
cachedFingerprint = e.skipFingerprints[file.Path]
}
e.skipMu.RUnlock()
if !cached || cachedMtime != mtime {
return false
}
if isS3SourcePath(file.Path) && sourceFingerprint != "" {
return cachedFingerprint == sourceFingerprint
}
return true
}
func (e *Engine) pathNeedsProjectReparse(path string) bool {
if e == nil || e.db == nil {
return false
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
project, ok := e.db.GetProjectByPath(lookupPath)
return ok && parser.NeedsProjectReparse(project)
}
func (e *Engine) pathNeedsCachedSkipBypass(path string) bool {
return e.pathNeedsProjectReparse(path) ||
e.pathNeedsDataVersionReparse(path)
}
func (e *Engine) pathNeedsDataVersionReparse(path string) bool {
if e == nil || e.db == nil {
return false
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
if _, _, ok := e.db.GetFileInfoByPath(lookupPath); !ok {
return false
}
return e.db.GetDataVersionByPath(lookupPath) < db.CurrentDataVersion()
}
func (e *Engine) processProviderFile(
ctx context.Context,
file parser.DiscoveredFile,
) (processResult, bool) {
mode := e.providerMigrationModes[file.Agent]
usesProvider := processFileUsesProvider(file.Agent)
if mode != parser.ProviderMigrationProviderAuthoritative && !usesProvider {
return processResult{}, false
}
// S3 sources are not provider-owned: the provider source sets read local
// files, so s3:// paths use the legacy S3 sync path (processS3Session),
// which handles object fetch, fingerprinting, and per-agent skip logic.
if strings.HasPrefix(file.Path, "s3://") {
return processResult{}, false
}
if file.ProviderSource != nil && !file.ProviderProcess && !usesProvider {
return processResult{}, false
}
// OpenCode-family shared-SQLite gate: when the whole container
// provably has not changed since the last fully verified pass, none
// of its sessions can have changed, so skip before paying for the
// per-session fingerprint (a DB open per source) and parse.
if e.sqliteContainerSourceFresh(file) {
return processResult{skip: true}, true
}
// Processing a file-backed storage session makes (or confirms) the
// storage copy as the archive's canonical content for its ID, so a
// same-ID SQLite row in this root's container is no longer backed by
// what its trusted membership verified. Drop it so a storage copy
// that appears and disappears entirely between full passes still
// forces the re-exposed row to re-verify (see
// dropTrustedSQLiteContainerSessionForStorage).
if sessionPath := e.openCodeStorageSessionPath(file); sessionPath != "" {
e.dropTrustedSQLiteContainerSessionForStorage(file.Agent, sessionPath)
}
// OpenCode-family file-backed storage gate: when the session's
// per-file stat signature matches the last verified pass, its parse
// inputs are unchanged, so skip before re-reading the whole message
// and part tree (see opencode_storage_gate.go). The captured state
// also feeds the post-parse promotion below.
storageState, storageSnap, storageStateOK :=
e.openCodeStorageSessionGateState(file)
if storageStateOK &&
e.openCodeStorageSessionFresh(file.Path, storageState) {
return processResult{skip: true}, true
}
factory, ok := e.providerFactories[file.Agent]
if !ok {
return processResult{
err: fmt.Errorf("provider not found for agent type: %s", file.Agent),
}, true
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: e.agentDirs[file.Agent],
Machine: e.machine,
PathRewriter: e.pathRewriter,
})
source, found, err := e.providerSourceForDiscoveredFile(ctx, provider, file)
if err != nil {
return processResult{err: err}, true
}
if !found {
// A forced parse on a deleted shared SQLite database (Zed, ZCode, Shelley)
// resolves to no source because the physical file is gone. Mirror the
// legacy deleted-source handling: complete the source as an empty
// force-replace so the engine retires every session that lived in the
// removed database instead of failing the sync.
if file.ForceParse &&
providerDeletedPhysicalSQLiteSource(file.Agent, file.Path) {
return processResult{forceReplace: true}, true
}
return processResult{
err: fmt.Errorf(
"%s provider source not found for %s",
file.Agent,
file.Path,
),
}, true
}
// SyncSingleSession resolves a single session by ID and carries the
// caller-preferred project (typically the DB-preserved value, so a
// user override is not reverted) on file.Project without an explicit
// ProviderSource. Provider FindSource re-derives ProjectHint from the
// path, so honor the caller's project as the hint in that case. Full
// discovery and changed-path classification always supply
// file.ProviderSource, whose ProjectHint stays authoritative.
if file.ProviderSource == nil && file.Project != "" {
source.ProjectHint = file.Project
}
verifiedCapture, verifiedMtime, verifiedFresh, verifiedStateOK :=
e.verifiedProviderSourceState(provider, source, file)
if verifiedStateOK && verifiedFresh {
if e.verifiedProviderSourceFreshInDB(
source, verifiedCapture.signature.size, verifiedMtime,
) {
return processResult{
skip: true,
mtime: verifiedMtime,
}, true
}
e.invalidateVerifiedSource(verifiedCapture.path)
}
// DB-freshness skip for single-session JSONL providers (Claude):
// when the stored session's size, mtime, and data version already
// match the source and its project does not need reparse, skip the
// parse entirely. This reproduces the legacy process arm's
// shouldSkipFile gate so an unchanged session is not re-parsed on
// every full sync.
sourceForceReplace := false
if mtime, fresh, forceReplace, contentVerified := e.providerSingleSessionFresh(
ctx, provider, source, file,
); fresh {
if !verifiedStateOK || contentVerified {
if verifiedStateOK {
e.promoteVerifiedSource(verifiedCapture)
}
return processResult{
skip: true,
mtime: mtime,
}, true
}
// A gate-eligible local source without a comparable stored hash (or
// whose hash could not be read) must take the fingerprint path once.
// Otherwise it would retain the legacy stat-only skip forever without
// ever earning verified-source trust.
} else if forceReplace {
sourceForceReplace = true
}
if freshMtime, fresh := e.providerSourceFreshBeforeFingerprint(source, file); fresh {
return processResult{
skip: true,
mtime: freshMtime,
}, true
}
fingerprint, err := provider.Fingerprint(ctx, source)
if err != nil {
if file.ForceParse &&
providerDeletedPhysicalSQLiteSource(file.Agent, file.Path) &&
errors.Is(err, os.ErrNotExist) {
return processResult{
excludedSessionIDs: e.providerSourceSessionIDsForForceReplace(
file.Agent,
source,
),
forceReplace: true,
}, true
}
return processResult{err: err}, true
}
cacheKey := providerProcessCacheKey(file, source, fingerprint)
cacheSkip := e.shouldCacheSkip(file)
if cacheSkip && !e.forceParse && !file.ForceParse {
e.skipMu.RLock()
cachedMtime, cached := e.skipCache[cacheKey]
e.skipMu.RUnlock()
if cached && cachedMtime == fingerprint.MTimeNS {
// A cached skip must not hide a session whose stored row needs
// self-healing (e.g. a parser data-version bump or generated
// roborev CI worktree project): clear the entry and fall through
// to a full reparse, mirroring the legacy process arm.
if !e.providerSkipCacheEntryFreshInDB(file, source, fingerprint) {
e.clearSkip(cacheKey)
} else if e.pathNeedsCachedSkipBypass(file.Path) {
e.clearSkip(cacheKey)
} else if file.Agent == parser.AgentCodex &&
e.codexCachedIndexSessionNameChanged(file.Path) {
// The transcript fingerprint can remain byte-for-byte identical
// while session_index.jsonl changes this session's title. Do not
// let a pre-existing transcript skip entry hide that metadata
// refresh; non-Codex providers avoid the index lookup entirely.
e.clearSkip(cacheKey)
} else {
if verifiedStateOK &&
e.shouldSkipProviderSourceByDB(file, fingerprint) {
e.promoteVerifiedSource(verifiedCapture)
}
return processResult{
skip: true,
mtime: fingerprint.MTimeNS,
cacheSkip: true,
cacheKey: cacheKey,
}, true
}
}
}
if cacheSkip && e.shouldSkipProviderSource(file, source, fingerprint) {
return processResult{
skip: true,
mtime: fingerprint.MTimeNS,
cacheSkip: true,
cacheKey: cacheKey,
}, true
}
// Append-only incremental parse for already-synced JSONL files.
// When the incremental path declines but signals forceReplace,
// carry the flag onto the full parse so the write path replaces
// stored messages instead of appending on top of stale rows.
incRes, incOK := e.tryProviderIncrementalAppend(
ctx, provider, source, file, fingerprint,
)
if incOK {
incRes.mtime = fingerprint.MTimeNS
incRes.cacheSkip = cacheSkip
incRes.cacheKey = cacheKey
return incRes, true
}
incForceReplace := sourceForceReplace || incRes.forceReplace
// DB-stored fingerprint skip. The provider has no database handle, so the
// engine reproduces the legacy DB-aware skip that single-session JSONL
// providers relied on: an unchanged source whose stored size and effective
// mtime already match is not reparsed, even when the in-memory skip cache
// was cleared (e.g. by SyncSingleSession) or never populated (a fresh
// engine). For Codex this also folds in the session_index.jsonl sidecar:
// a shared index mtime bump that did not change this session's title must
// not trigger a reparse.
if !incForceReplace && !e.forceParse && !file.ForceParse &&
e.shouldSkipProviderSourceByDB(file, fingerprint) {
if verifiedStateOK {
e.promoteVerifiedSource(verifiedCapture)
}
return processResult{
skip: true,
mtime: fingerprint.MTimeNS,
cacheSkip: cacheSkip,
cacheKey: cacheKey,
noCacheSkip: true,
}, true
}
// DB-stored-file-info skip: a session whose persisted file_size/file_mtime
// already match the source fingerprint (and whose data_version is current)
// is unchanged and need not be reparsed. This reproduces the legacy
// shouldSkipByPath behavior the per-agent process methods provided before the
// migration, so a repeat full/periodic sync of an untouched
// provider-authoritative session (OpenHands, Cursor, Hermes, Vibe, ...)
// skips instead of rewriting. It only skips on an exact size+mtime match, so
// a provider whose fingerprint mtime differs from the stored value simply
// reparses, matching the prior behavior. Claude and Cowork have their own
// earlier freshness checks; this is the generic fallback for the rest.
if !incForceReplace && !e.forceParse && !file.ForceParse &&
e.providerSourceUnchangedInDB(source, fingerprint) {
return processResult{
skip: true,
mtime: fingerprint.MTimeNS,
cacheSkip: cacheSkip,
cacheKey: cacheKey,
}, true
}
outcome, err := provider.Parse(ctx, parser.ParseRequest{
Source: source,
Fingerprint: fingerprint,
Machine: e.machine,
ForceParse: e.forceParse || file.ForceParse,
})
if err != nil {
return processResult{
err: err,
mtime: fingerprint.MTimeNS,
cacheSkip: cacheSkip,
cacheKey: cacheKey,
noCacheSkip: true,
}, true
}
if err := validateProviderOutcome(
provider.Definition(),
source,
fingerprint,
outcome,
); err != nil {
return processResult{
err: err,
mtime: fingerprint.MTimeNS,
cacheSkip: cacheSkip,
cacheKey: cacheKey,
noCacheSkip: true,
}, true
}
applyProviderFingerprintFileInfo(file.Agent, fingerprint, outcome.Results)
cleanCache := providerOutcomeAllowsCleanSkipCache(outcome)
if outcome.SkipReason != parser.SkipNone {
excludedSessionIDs := append([]string(nil), outcome.ExcludedSessionIDs...)
if outcome.ForceReplace && outcome.ResultSetComplete {
excludedSessionIDs = append(
excludedSessionIDs,
e.providerSourceSessionIDsForForceReplace(file.Agent, source)...,
)
}
return processResult{
skip: !outcome.ForceReplace,
excludedSessionIDs: excludedSessionIDs,
mtime: fingerprint.MTimeNS,
cacheSkip: cacheSkip,
cacheKey: cacheKey,
noCacheSkip: !cleanCache,
forceReplace: outcome.ForceReplace,
}, true
}
parsedResults := parseOutcomeResults(outcome.Results)
parsedCount := len(parsedResults)
res := processResult{
results: e.dropUnchangedSharedSQLiteResults(file, parsedResults),
excludedSessionIDs: append([]string(nil), outcome.ExcludedSessionIDs...),
mtime: fingerprint.MTimeNS,
cacheSkip: cacheSkip,
cacheKey: cacheKey,
noCacheSkip: !cleanCache,
forceReplace: outcome.ForceReplace || incForceReplace,
suppressPresenceSweep: !outcome.ResultSetComplete,
}
// Incremental-append providers (Claude and Codex) need the stored file
// identity so a later sync can detect an atomic file replacement
// (new inode/device) and fall back to a full parse instead of
// appending on top of stale state. Match the legacy process arm,
// which stamped inode/device from the source file stat.
e.stampProviderFileIdentity(provider, source, res.results)
for _, result := range outcome.Results {
if result.DataVersion == parser.DataVersionNeedsRetry {
if res.retrySessionIDs == nil {
res.retrySessionIDs = make(map[string]bool)
}
res.retrySessionIDs[result.Result.Session.ID] = true
}
}
if e.forceParse || file.ForceParse {
for _, sourceErr := range outcome.SourceErrors {
res.sessionErrs = append(res.sessionErrs, sessionParseError{
sessionID: sourceErr.SessionID,
virtualPath: sourceErr.SourceKey,
err: sourceErr.Err,
})
}
}
e.applyProviderFilePathPolicies(provider, file.Agent, &res)
if storageStateOK {
e.stageOpenCodeStorageTrust(
&res, file.Path, storageState, storageSnap,
parsedCount, outcome.ResultSetComplete,
)
}
return res, true
}
// dropUnchangedSharedSQLiteResults reproduces the legacy per-session skip the
// folded processZed/processShelley loops and the aiderFileUnchanged check
// performed. Zed and Shelley keep every session in one shared SQLite database,
// and Aider fans every run out of one shared history file, so the provider
// re-parses every session on any change to that shared source. Without a
// per-session filter the engine would rewrite and recount unchanged sessions.
// This drops results whose stored file_mtime (and, for Shelley's
// second-precision timestamps and Aider's whole-file content hash, the
// fingerprint stored in file_hash) and data_version already match, using the
// path rewriter so remote stored paths resolve. Force-parse runs (parse-diff,
// single-session resync) keep every result so they always re-emit.
func (e *Engine) dropUnchangedSharedSQLiteResults(
file parser.DiscoveredFile,
results []parser.ParseResult,
) []parser.ParseResult {
if e.forceParse || file.ForceParse || len(results) == 0 {
return results
}
compareHash := false
switch file.Agent {
case parser.AgentShelley:
compareHash = true
case parser.AgentAider:
// Every aider run in a history file shares the file's content hash, so
// a same-mtime append/truncate is caught by the hash compare.
compareHash = true
case parser.AgentOpenCode, parser.AgentKilo, parser.AgentMiMoCode, parser.AgentIcodemate:
// OpenCode-family providers fan one shared container out to per-session
// results. The per-session mtime is the session's own updated time, and
// the hash compare uses the opencode storage fingerprint to catch
// same-mtime content changes.
compareHash = true
case parser.AgentZed, parser.AgentKiro:
// Zed and Kiro fan one container DB out to a session per row and have no
// per-row content hash, so unchanged rows are detected by mtime plus
// data version, matching their legacy container sync. Without Kiro here
// every Kiro row is reparsed and rewritten on every full sync.
default:
return results
}
kept := results[:0]
for _, r := range results {
path := r.Session.File.Path
if path == "" {
kept = append(kept, r)
continue
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
_, storedMtime, ok := e.db.GetFileInfoByPath(lookupPath)
if !ok || storedMtime != r.Session.File.Mtime {
kept = append(kept, r)
continue
}
if compareHash {
storedHash, _ := e.db.GetFileHashByPath(lookupPath)
if storedHash != r.Session.File.Hash {
kept = append(kept, r)
continue
}
}
if e.db.GetDataVersionByPath(lookupPath) < db.CurrentDataVersion() {
kept = append(kept, r)
continue
}
// Unchanged: drop so the write batch neither rewrites nor recounts it.
}
return kept
}
func (e *Engine) providerSourceSessionIDsForForceReplace(
agent parser.AgentType,
source parser.SourceRef,
) []string {
root := ""
for _, candidate := range []string{source.DisplayPath, source.FingerprintKey, source.Key} {
if candidate != "" {
root = candidate
break
}
}
if root == "" {
return nil
}
if e.pathRewriter != nil {
root = e.pathRewriter(root)
}
sourcePaths, err := e.db.ListStoredSourcePathHints(string(agent), []string{root})
if err != nil {
log.Printf("list provider force-replace source hints: %v", err)
return nil
}
seen := make(map[string]struct{})
var ids []string
for _, sourcePath := range sourcePaths {
pathIDs, err := e.db.ListSessionIDsByFilePath(sourcePath, string(agent))
if err != nil {
log.Printf("list provider force-replace sessions: %v", err)
continue
}
for _, id := range pathIDs {
if _, ok := seen[id]; ok {
continue
}
seen[id] = struct{}{}
ids = append(ids, id)
}
}
return ids
}
// applyProviderFilePathPolicies reproduces the DB-aware, file-path-scoped
// session bookkeeping that a provider cannot do on its own (it has no database
// handle). It runs only for single-session-per-file providers whose canonical
// ID can change while the source path is unchanged (e.g. Vibe, whose ID flips
// between the meta.json session_id and the directory-name fallback as meta.json
// appears or is removed). Multi-session sources are skipped, where several
// distinct sessions legitimately share one path; for stable-ID providers it is
// a no-op because the stored ID always matches the freshly parsed one.
//
// Two policies are applied per result, keyed by the (path-rewritten) file_path:
//
// 1. Resurrection guard: if the user removed the session occupying this path —
// a trashed row at the same path, or an alternate identity for the path
// (the provider's excluded fallback ID, or a stale stored ID) that is now
// trashed or permanently excluded — the freshly parsed row must not be
// written under its new ID. The result is dropped and its ID is excluded.
// 2. Stale-row cleanup: any other live stored ID at the same path that the
// current parse no longer emits is added to the exclusion list so the
// superseded row is deleted.
func (e *Engine) applyProviderFilePathPolicies(
provider parser.Provider,
agent parser.AgentType,
res *processResult,
) {
if provider.Capabilities().Source.MultiSessionSource == parser.CapabilitySupported {
return
}
if len(res.results) == 0 {
return
}
excluded := make(map[string]struct{}, len(res.excludedSessionIDs))
for _, id := range e.applyIDPrefixToSessionIDs(res.excludedSessionIDs) {
excluded[id] = struct{}{}
}
addExclusion := func(id string) {
if id == "" {
return
}
if _, ok := excluded[id]; ok {
return
}
excluded[id] = struct{}{}
res.excludedSessionIDs = append(res.excludedSessionIDs, id)
}
kept := res.results[:0]
for _, result := range res.results {
path := result.Session.File.Path
if path == "" {
kept = append(kept, result)
continue
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
currentID := result.Session.ID
currentPrefixedID := e.idPrefix + result.Session.ID
existingIDs, err := e.db.ListSessionIDsByFilePath(lookupPath, string(agent))
if err != nil {
log.Printf("list session IDs by file path: %v", err)
kept = append(kept, result)
continue
}
// Resurrection guard. The path's identity is removed when a trashed row
// shares it, or when any alternate identity for the path (the
// provider's excluded fallback IDs or a stale stored ID) is trashed or
// permanently excluded. In that case the new row must not be written.
suppress := e.db.HasTrashedSessionByFilePath(lookupPath, string(agent))
if !suppress {
for id := range excluded {
if id == currentID || id == currentPrefixedID {
continue
}
if e.db.IsSessionExcluded(id) || e.db.IsSessionTrashed(id) {
suppress = true
break
}
}
}
if !suppress {
for _, id := range existingIDs {
if id == currentID || id == currentPrefixedID {
continue
}
if e.db.IsSessionExcluded(id) || e.db.IsSessionTrashed(id) {
suppress = true
break
}
}
}
if suppress {
// Keep a trashed current ID trashed rather than converting it to a
// parser deletion; the upsert's trash guard already hides it.
if (currentPrefixedID == "" || !e.db.IsSessionTrashed(currentPrefixedID)) &&
!e.db.IsSessionTrashed(currentID) {
addExclusion(currentID)
}
continue
}
// Stale-row cleanup for live siblings the current parse supersedes.
for _, id := range existingIDs {
if id == currentID || id == currentPrefixedID {
continue
}
addExclusion(id)
}
kept = append(kept, result)
}
res.results = kept
}
func providerOutcomeAllowsCleanSkipCache(outcome parser.ParseOutcome) bool {
if !outcome.ResultSetComplete {
return false
}
if len(outcome.SourceErrors) > 0 {
return false
}
for _, result := range outcome.Results {
if result.DataVersion == parser.DataVersionNeedsRetry {
return false
}
}
return true
}
func (e *Engine) providerSourceForDiscoveredFile(
ctx context.Context,
provider parser.Provider,
file parser.DiscoveredFile,
) (parser.SourceRef, bool, error) {
if file.ProviderSource != nil {
source := *file.ProviderSource
if source.Provider != file.Agent {
return parser.SourceRef{}, false, fmt.Errorf(
"provider source mismatch for %s: %s",
file.Agent,
source.Provider,
)
}
return source, true, nil
}
return provider.FindSource(ctx, parser.FindSourceRequest{
StoredFilePath: file.Path,
FingerprintKey: file.Path,
RequireFreshSource: !e.forceParse && !file.ForceParse,
})
}
func providerProcessCacheKey(
file parser.DiscoveredFile,
source parser.SourceRef,
fingerprint parser.SourceFingerprint,
) string {
agent := file.Agent
if agent == "" {
agent = source.Provider
}
key := ""
if key := plannedSkipKey(source, fingerprint); key != "" {
return providerProcessCacheKeyWithHash(key, agent, fingerprint)
}
key = file.Path
return providerProcessCacheKeyWithHash(key, agent, fingerprint)
}
func providerProcessCacheKeyWithHash(
key string,
agent parser.AgentType,
fingerprint parser.SourceFingerprint,
) string {
if key == "" {
return ""
}
if fingerprint.Hash == "" || !providerFingerprintHashInCacheKey(agent) {
return key
}
return key + "?source_hash=" + fingerprint.Hash
}
func providerFingerprintHashInCacheKey(agent parser.AgentType) bool {
switch agent {
case parser.AgentClaude, parser.AgentCodex, parser.AgentDevin, parser.AgentQoder, parser.AgentWindsurf:
return true
default:
return false
}
}
// providerFingerprintHashRequiredForFreshness also protects stored rows. Hash
// cache keys protect rowless parser exclusions and failures; cacheSkip removes
// older hash siblings so hot append-only files retain only one content version.
func providerFingerprintHashRequiredForFreshness(agent parser.AgentType) bool {
switch agent {
case parser.AgentClaude, parser.AgentCodex, parser.AgentDevin, parser.AgentQoder, parser.AgentWindsurf:
return true
default:
return false
}
}
func (e *Engine) providerSkipCacheEntryFreshInDB(
file parser.DiscoveredFile,
source parser.SourceRef,
fingerprint parser.SourceFingerprint,
) bool {
agent := file.Agent
if agent == "" {
agent = source.Provider
}
if fingerprint.Hash == "" || !providerFingerprintHashRequiredForFreshness(agent) {
return true
}
lookupPath := providerSkipLookupPath(file, source, fingerprint)
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(lookupPath)
}
if agent == parser.AgentClaude || agent == parser.AgentCodex {
storedIDs, err := e.db.ListSessionIDsByFilePath(
lookupPath, string(agent),
)
if err == nil && len(storedIDs) == 0 {
// A cached parse failure or intentionally ignored source has no
// persisted row or hash to compare. Retry suppression is therefore
// mtime/source-signal based until a row exists: a same-mtime rewrite
// cannot be distinguished in this no-row state. Hash validation applies
// once a session has actually been stored.
return true
}
}
return e.providerFingerprintHashMatchesDB(agent, lookupPath, fingerprint)
}
func processFileUsesProvider(agent parser.AgentType) bool {
switch agent {
case parser.AgentForge, parser.AgentPiebald, parser.AgentWarp, parser.AgentZCode:
return true
default:
return false
}
}
func (e *Engine) shouldSkipProviderSource(
file parser.DiscoveredFile,
source parser.SourceRef,
fingerprint parser.SourceFingerprint,
) bool {
agent := file.Agent
if agent == "" {
agent = source.Provider
}
if !providerSourceSupportsPersistedFreshness(agent) {
return false
}
if e.forceParse || file.ForceParse {
return false
}
lookupPath := providerSkipLookupPath(file, source, fingerprint)
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(lookupPath)
}
storedSize, storedMtime, ok := e.db.GetFileInfoByPath(lookupPath)
if !ok {
return false
}
if fingerprint.Size != 0 && storedSize != fingerprint.Size {
return false
}
if storedMtime != fingerprint.MTimeNS {
return false
}
if !e.providerFingerprintHashMatchesDB(agent, lookupPath, fingerprint) {
return false
}
return e.db.GetDataVersionByPath(lookupPath) >= db.CurrentDataVersion()
}
func providerSourceSupportsPersistedFreshness(agent parser.AgentType) bool {
switch agent {
case parser.AgentForge, parser.AgentWarp, parser.AgentZCode:
return true
default:
return false
}
}
func providerSkipLookupPath(
file parser.DiscoveredFile,
source parser.SourceRef,
fingerprint parser.SourceFingerprint,
) string {
for _, path := range []string{
fingerprint.Key,
source.FingerprintKey,
source.DisplayPath,
source.Key,
file.Path,
} {
if path != "" {
return path
}
}
return file.Path
}
func (e *Engine) shouldCacheSkip(
file parser.DiscoveredFile,
) bool {
if file.Agent == parser.AgentKiro {
if filepath.Base(file.Path) == kiroSQLiteDBName {
return false
}
if _, _, ok := parseKiroSQLiteVirtualPath(file.Path); ok {
return false
}
}
if file.Agent == parser.AgentZed {
if filepath.Base(file.Path) == "threads.db" {
return false
}
if _, _, ok := parser.ParseVirtualSourcePathForBase(file.Path, "threads.db"); ok {
return false
}
}
if file.Agent == parser.AgentZCode {
if filepath.Base(file.Path) == parser.ZCodeDBName {
return false
}
if _, _, ok := parser.ParseVirtualSourcePathForBase(file.Path, parser.ZCodeDBName); ok {
return false
}
}
if file.Agent == parser.AgentShelley {
if filepath.Base(file.Path) == shelleyDBFile {
return false
}
if _, _, ok := parser.ParseVirtualSourcePathForBase(file.Path, shelleyDBFile); ok {
return false
}
}
if file.Agent == parser.AgentVSCopilot {
// Visual Studio Copilot conversations are skipped by a composite
// fingerprint spanning every sibling trace file (see
// processVisualStudioCopilot). The generic skip cache keys on the
// representative file's mtime alone, so a cached entry would bypass that
// composite check and miss a sibling-only change or removal.
if parser.IsVisualStudioCopilotTraceFile(file.Path) {
return false
}
if _, _, ok :=
parser.SplitVisualStudioCopilotVirtualPath(file.Path); ok {
return false
}
}
if file.Agent == parser.AgentAider {
// Aider fans one physical history file out into per-run virtual
// sessions. A mtime-only skip can hide same-mtime content changes,
// missing run rows, or stale per-run data versions before the
// provider fingerprint and dropUnchangedSharedSQLiteResults hash
// checks run, so all Aider freshness stays on that provider-aware path.
return false
}
if !isOpenCodeFormatStorageAgent(file.Agent) {
return true
}
if filepath.Base(file.Path) == openCodeFormatDBName(file.Agent) {
return false
}
if isOpenCodeFormatSQLiteVirtualPath(file.Agent, file.Path) {
return false
}
for _, dir := range e.agentDirs[file.Agent] {
if dir == "" {
continue
}
src := resolveOpenCodeFormatSource(file.Agent, dir)
if src.Mode != parser.OpenCodeSourceStorage {
continue
}
if rel, ok := isUnder(dir, file.Path); ok {
rel = filepath.ToSlash(rel)
sessionPrefix := "storage/" +
filepath.Base(src.SessionRoot) + "/"
return !strings.HasPrefix(rel, sessionPrefix)
}
}
return true
}
const sourceHashSkipMarker = "?source_hash="
// normalizeSourceHashSkipCache performs the one archive-sized pass needed to
// repair legacy duplicate source-hash entries and build the watcher-time
// sibling index. A family with multiple hashed keys is ambiguous: same-mtime
// rewrites are why the hash exists, so no stored key can safely be called
// current. Drop that family so the source reparses once and establishes a
// trustworthy key.
func normalizeSourceHashSkipCache(
cache map[string]int64, fingerprints map[string]string,
) (map[string]string, map[string]struct{}) {
index := make(map[string]string)
counts := make(map[string]int)
ambiguous := make(map[string]struct{})
for path := range cache {
base, _, hashed := strings.Cut(path, sourceHashSkipMarker)
if !hashed {
continue
}
counts[base]++
if counts[base] == 1 {
index[base] = path
}
}
for path := range cache {
base, _, hashed := strings.Cut(path, sourceHashSkipMarker)
if hashed && counts[base] > 1 {
delete(cache, path)
ambiguous[base] = struct{}{}
if fingerprints != nil {
delete(fingerprints, path)
}
}
}
for base := range index {
delete(cache, base)
if fingerprints != nil {
delete(fingerprints, base)
}
if counts[base] > 1 {
delete(index, base)
}
}
return index, ambiguous
}
// cacheSkip records a file so it won't be retried until its mtime changes.
// The returned work count measures sibling-index probes and is used by the
// cardinality regression to keep watcher-time work independent of cache size.
func (e *Engine) cacheSkip(
path string, mtime int64, sourceFingerprint ...string,
) int {
e.skipMu.Lock()
work := e.removeSkipHashSiblingsLocked(path)
e.skipCache[path] = mtime
if base, _, hashed := strings.Cut(path, sourceHashSkipMarker); hashed {
e.skipHashKeys[base] = path
}
fingerprint := ""
if len(sourceFingerprint) > 0 {
fingerprint = sourceFingerprint[0]
}
if fingerprint != "" {
if e.skipFingerprints == nil {
e.skipFingerprints = make(map[string]string)
}
e.skipFingerprints[path] = fingerprint
} else if e.skipFingerprints != nil {
delete(e.skipFingerprints, path)
}
e.skipMu.Unlock()
return work
}
// clearSkip removes a skip-cache entry when a file produces a valid session.
// Its work count has the same cardinality-regression role as cacheSkip's.
func (e *Engine) clearSkip(path string) int {
e.skipMu.Lock()
work := e.removeSkipHashSiblingsLocked(path)
delete(e.skipCache, path)
delete(e.skipFingerprints, path)
e.skipMu.Unlock()
_ = e.db.DeleteSkippedFile(path)
return work
}
func (e *Engine) removeSkipHashSiblingsLocked(path string) int {
if e.skipHashKeys == nil {
e.skipHashKeys, _ = normalizeSourceHashSkipCache(
e.skipCache, e.skipFingerprints,
)
}
base, _, hasHash := strings.Cut(path, sourceHashSkipMarker)
if !hasHash {
if sibling, ok := e.skipHashKeys[path]; ok {
delete(e.skipCache, sibling)
delete(e.skipFingerprints, sibling)
delete(e.skipHashKeys, path)
}
return 1
}
delete(e.skipCache, base)
delete(e.skipFingerprints, base)
if sibling, ok := e.skipHashKeys[base]; ok {
delete(e.skipCache, sibling)
delete(e.skipFingerprints, sibling)
delete(e.skipHashKeys, base)
}
return 2
}
// clearWatcherOverflowCaches invalidates every freshness shortcut whose
// correctness can depend on receiving a concrete changed path. The following
// forced discovery pass rebuilds these caches from parsed source state.
func (e *Engine) clearWatcherOverflowCaches() {
e.skipMu.Lock()
e.skipCache = make(map[string]int64)
e.skipFingerprints = make(map[string]string)
e.skipHashKeys = make(map[string]string)
e.skipMu.Unlock()
if !e.ephemeral {
if err := e.db.ReplaceSkippedFiles(map[string]int64{}); err != nil {
log.Printf("clearing skipped files after watcher overflow: %v", err)
}
}
e.clearTrustedOpenCodeStorageSessions()
e.clearTrustedSQLiteContainers()
e.clearVerifiedSources()
parser.EvictAllCodexSessionIndexes()
}
// InjectSkipCache merges entries into the in-memory skip
// cache. Used by remote sync to pre-populate with
// translated paths.
func (e *Engine) InjectSkipCache(entries map[string]int64) {
e.skipMu.Lock()
defer e.skipMu.Unlock()
if e.skipHashKeys == nil {
e.skipHashKeys, _ = normalizeSourceHashSkipCache(
e.skipCache, e.skipFingerprints,
)
}
incoming := make(map[string]int64, len(entries))
maps.Copy(incoming, entries)
_, ambiguous := normalizeSourceHashSkipCache(incoming, nil)
for base := range ambiguous {
e.removeSkipHashSiblingsLocked(base + sourceHashSkipMarker)
}
for path, mtime := range incoming {
e.removeSkipHashSiblingsLocked(path)
e.skipCache[path] = mtime
if base, _, hashed := strings.Cut(path, sourceHashSkipMarker); hashed {
e.skipHashKeys[base] = path
}
}
}
// SnapshotSkipCache returns a copy of the in-memory skip
// cache.
func (e *Engine) SnapshotSkipCache() map[string]int64 {
e.skipMu.RLock()
defer e.skipMu.RUnlock()
out := make(map[string]int64, len(e.skipCache))
maps.Copy(out, e.skipCache)
return out
}
// persistSkipCache writes the in-memory skip cache to the
// database so skipped files survive process restarts.
// Returns the number of entries persisted.
func (e *Engine) persistSkipCache() int {
if e.ephemeral {
return 0
}
e.skipMu.RLock()
snapshot := make(map[string]int64, len(e.skipCache))
maps.Copy(snapshot, e.skipCache)
e.skipMu.RUnlock()
if err := e.db.ReplaceSkippedFiles(snapshot); err != nil {
log.Printf("persisting skip cache: %v", err)
}
return len(snapshot)
}
// shouldSkipFile returns true when the file's size and mtime
// match what is already stored in the database (by session ID).
// This relies on mtime changing on any write, which holds for
// append-only session files under normal filesystem behavior.
// S3 callers pass an object fingerprint to guard same-size,
// same-timestamp rewrites on object stores with coarse mtimes.
func (e *Engine) shouldSkipFile(
sessionID string, info os.FileInfo,
) bool {
return e.shouldSkipFileWithPrefix(e.idPrefix, sessionID, info)
}
// providerSourceUnchangedInDB reports whether a provider source's persisted
// file metadata already matches its current fingerprint, so a reparse would be
// a no-op. It compares the DB-stored file_size/file_mtime for the source's
// path against the fingerprint and requires a current data_version, mirroring
// shouldSkipByPath for the provider-authoritative runtime. A source with no
// stored row, an empty key, or a non-fingerprint identity (no size, e.g. a
// tombstone) never matches and therefore reparses.
func (e *Engine) providerSourceUnchangedInDB(
source parser.SourceRef,
fingerprint parser.SourceFingerprint,
) bool {
if fingerprint.MTimeNS == 0 && fingerprint.Size == 0 {
return false
}
lookupPath := providerDiscoveredPath(source)
if lookupPath == "" {
return false
}
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(lookupPath)
}
storedSize, storedMtime, ok := e.db.GetFileInfoByPath(lookupPath)
if !ok {
return false
}
if storedSize != fingerprint.Size || storedMtime != fingerprint.MTimeNS {
return false
}
if !e.providerFingerprintHashMatchesDB(source.Provider, lookupPath, fingerprint) {
return false
}
// A stale stored project (e.g. a generated roborev CI worktree name)
// must defeat the unchanged-source skip so the corrected project is
// reparsed, mirroring shouldSkipCodexFingerprint and the in-memory
// skip-cache bypass in processProviderFile.
if project, ok := e.db.GetProjectByPath(lookupPath); ok &&
parser.NeedsProjectReparse(project) {
return false
}
return e.db.GetDataVersionByPath(lookupPath) >= db.CurrentDataVersion()
}
func (e *Engine) providerFingerprintHashMatchesDB(
agent parser.AgentType,
lookupPath string,
fingerprint parser.SourceFingerprint,
) bool {
if fingerprint.Hash == "" || !providerFingerprintHashRequiredForFreshness(agent) {
return true
}
storedHash, ok := e.db.GetFileHashByPath(lookupPath)
return ok && storedHash == fingerprint.Hash
}
// shouldSkipByPath checks file size and mtime against what is
// stored in the database by file_path. Used for codex/gemini
// files where the session ID requires parsing.
func (e *Engine) shouldSkipByPath(
path string, info os.FileInfo,
) bool {
if e.forceParse { // parse-diff: always re-parse
return false
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
storedSize, storedMtime, ok := e.db.GetFileInfoByPath(
lookupPath,
)
if !ok {
return false
}
if storedSize != info.Size() ||
storedMtime != info.ModTime().UnixNano() {
return false
}
if e.db.GetDataVersionByPath(lookupPath) <
db.CurrentDataVersion() {
return false
}
return true
}
// fakeSnapshotInfo wraps a pre-computed size and mtime
// (nanoseconds) as os.FileInfo so that shouldSkipByPath can
// be reused for OpenHands snapshot-based skip detection.
type fakeSnapshotInfo struct {
fName string
fSize int64
fMtime int64
}
func (f fakeSnapshotInfo) Name() string { return f.fName }
func (f fakeSnapshotInfo) Size() int64 { return f.fSize }
func (f fakeSnapshotInfo) Mode() os.FileMode { return 0 }
func (f fakeSnapshotInfo) ModTime() time.Time {
return time.Unix(0, f.fMtime)
}
func (f fakeSnapshotInfo) IsDir() bool { return false }
func (f fakeSnapshotInfo) Sys() any { return nil }
// providerSingleSessionFresh reports whether a single-session JSONL
// provider's source (Claude) maps to a stored session that is already
// up to date: the source size and mtime match what is stored, the row
// is at the current parser data version, and its project does not need
// reparse. It reproduces the legacy Claude process arm's shouldSkipFile
// gate so an unchanged session is skipped instead of re-parsed every
// full sync. Providers without incremental append, multi-session
// sources, or sources that are not a single physical file are never
// considered fresh here and always fall through to the full parse.
func (e *Engine) providerSingleSessionFresh(
ctx context.Context,
provider parser.Provider,
source parser.SourceRef,
file parser.DiscoveredFile,
) (mtime int64, fresh bool, forceReplace bool, contentVerified bool) {
// Match the legacy shouldSkipFile gate, which keyed off the
// engine-wide forceParse (parse-diff) flag only. A per-file
// ForceParse (set by SyncSingleSession to bypass the error skip
// cache) must not defeat the DB-freshness skip: an unchanged session
// is still skipped so a single-session resync does not, for example,
// reapply a worktree project mapping to a file that has not changed.
if e.forceParse {
return 0, false, false, false
}
// Claude is the single-physical-file provider that takes the
// append-only incremental path. Its source stem is the session ID,
// so DB freshness can be checked by that ID even though a DAG fork
// can later split the file into several sessions.
if provider.Definition().Type != parser.AgentClaude {
return 0, false, false, false
}
if provider.Capabilities().Source.IncrementalAppend !=
parser.CapabilitySupported {
return 0, false, false, false
}
path := providerDiscoveredPath(source)
if path == "" {
return 0, false, false, false
}
sessionID := claudeSessionIDFromPath(path)
if sessionID == "" {
return 0, false, false, false
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
// statPath is the on-disk file the stat came from: lookupPath when it
// resolves (local sync, where lookupPath == path), otherwise the physical
// source path (remote sync rewrites path to a non-local logical key). The
// content guard below hashes statPath, so it must be the openable file.
statPath := lookupPath
info, err := os.Stat(lookupPath)
if err != nil {
statPath = path
info, err = os.Stat(path)
if err != nil {
return 0, false, false, false
}
}
if !e.shouldSkipFile(sessionID, info) {
return 0, false, false, false
}
if e.providerIncrementalIdentityChanged(lookupPath, info) {
return 0, false, true, false
}
contentChanged, contentVerified := e.providerIncrementalContentChanged(
e.idPrefix+sessionID, statPath, info,
)
if contentChanged {
return 0, false, true, contentVerified
}
sess, _ := e.db.GetSession(ctx, e.idPrefix+sessionID)
return info.ModTime().UnixNano(), sess != nil &&
sess.Project != "" &&
!parser.NeedsProjectReparse(sess.Project), false, contentVerified
}
func (e *Engine) providerIncrementalIdentityChanged(
lookupPath string,
info os.FileInfo,
) bool {
if e.pathRewriter != nil {
// Remote imports rewrite per-run temp paths to stable source paths;
// the temp inode is expected to change between identical downloads.
return false
}
curInode, curDevice := getFileIdentity(info)
return e.db.FileIdentityChanged(lookupPath, curInode, curDevice)
}
// providerIncrementalContentChanged reports whether a single-session JSONL
// source whose size, mtime, and file identity already match the stored row
// nonetheless holds different bytes than were last parsed. It is the last
// guard against a same-size, same-mtime, same-inode in-place rewrite: two
// fast writes landing in one filesystem mtime granule (or a coarse-mtime
// filesystem) leave every stat signal identical, so only the content hash
// distinguishes a genuine rewrite from an unchanged file.
//
// hashPath is the physical file the stat came from -- the local path for
// local sources, the materialized download for remote (path-rewritten)
// sources. The stored file_hash is computed over those same materialized
// bytes on both the full-parse (hashJSONLSourceFile) and incremental
// (ComputeFileHashPrefix) paths, so the on-disk prefix hash is directly
// comparable regardless of the logical key the row is stored under. That is
// also why this is the correct freshness signal for remote sync: every
// re-download gets a fresh inode, so the inode net is disabled to avoid a
// false-positive re-parse, but identical content still hashes equal here while
// a genuine rewrite does not. shouldSkipFile has already confirmed the stored
// file_size equals the current size, so the prefix hash covers the stored byte
// range. Rows without a stored hash (legacy or non-fingerprinted) report an
// unverified match. Gate-eligible local sources then fall through to the
// fingerprint path once, while sources that cannot use the local-stat gate
// preserve the legacy size/mtime/identity freshness behavior.
func (e *Engine) providerIncrementalContentChanged(
fullID, hashPath string,
info os.FileInfo,
) (changed, verified bool) {
storedHash, ok := e.db.GetSessionFileHash(fullID)
if !ok || storedHash == "" {
return false, false
}
curHash, err := ComputeFileHashPrefix(hashPath, info.Size())
if err != nil {
return false, false
}
return curHash != storedHash, true
}
func (e *Engine) providerSourceFreshBeforeFingerprint(
source parser.SourceRef,
file parser.DiscoveredFile,
) (int64, bool) {
if e.forceParse || file.ForceParse {
return 0, false
}
path := providerDiscoveredPath(source)
if path == "" {
return 0, false
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
info, err := os.Stat(lookupPath)
if err != nil {
info, err = os.Stat(path)
if err != nil {
return 0, false
}
}
switch file.Agent {
case parser.AgentCowork:
mtime := parser.CoworkSessionMtime(path, info.ModTime().UnixNano())
effectiveInfo := fakeSnapshotInfo{
fSize: info.Size(),
fMtime: mtime,
}
if e.shouldSkipByPath(path, effectiveInfo) {
return mtime, true
}
// Gemini is deliberately absent here. Its fingerprint is composite (the
// session file plus projects.json and trustedFolders.json), so a
// pre-fingerprint skip keyed only on the session file's size and mtime
// would skip a session whose project metadata changed while the transcript
// did not, leaving a stale project on scheduled syncs. Gemini relies on the
// post-fingerprint skip cache instead, whose mtime folds in the composite.
case parser.AgentCopilot:
mtime := copilotEffectiveMtime(path, info)
effectiveInfo := fakeSnapshotInfo{
fSize: info.Size(),
fMtime: mtime,
}
if e.shouldSkipByPath(path, effectiveInfo) {
return mtime, true
}
}
return 0, false
}
// stampProviderFileIdentity fills a missing source inode/device on parsed
// results for an incremental-append provider. A provider may have captured an
// authoritative identity from the same descriptor it parsed, so a later path
// stat must not overwrite that snapshot after an atomic replacement. The
// legacy Claude process arm relies on this fallback because Claude does not
// supply descriptor identity itself. Providers whose source is not a single
// physical file, or that do not support incremental append, are left untouched.
func (e *Engine) stampProviderFileIdentity(
provider parser.Provider,
source parser.SourceRef,
results []parser.ParseResult,
) {
if provider.Capabilities().Source.IncrementalAppend !=
parser.CapabilitySupported {
return
}
path := providerDiscoveredPath(source)
if path == "" {
return
}
info, err := os.Stat(path)
if err != nil {
return
}
inode, device := getFileIdentity(info)
for i := range results {
if results[i].Session.File.Inode != 0 ||
results[i].Session.File.Device != 0 {
continue
}
results[i].Session.File.Inode = inode
results[i].Session.File.Device = device
}
}
// tryProviderIncrementalAppend reproduces the legacy incremental-append
// sync path for a provider-authoritative agent that supports append-only
// incremental parsing (Claude or Codex). The provider owns the byte-offset parse
// via ParseIncremental, but the engine still owns the DB-aware
// bookkeeping (session lookup, data-version and identity guards, ordinal
// resume, cross-sync split detection, and cumulative counters), so this
// drives the shared tryIncrementalJSONL with an adapter that calls the
// provider. Returns (result, true) when the incremental path produced a
// terminal result, or (result, false) to fall through to the full
// provider parse (carrying any forceReplace signal).
func (e *Engine) tryProviderIncrementalAppend(
ctx context.Context,
provider parser.Provider,
source parser.SourceRef,
file parser.DiscoveredFile,
fingerprint parser.SourceFingerprint,
) (processResult, bool) {
// Match the legacy tryIncrementalJSONL gate, which suppressed append
// deltas only under the engine-wide forceParse (parse-diff) flag. A
// per-file ForceParse keeps Claude on its incremental path; Codex is the
// explicit exception below because a single-session refresh must rebuild
// head-derived metadata.
if e.forceParse {
return processResult{}, false
}
if provider.Capabilities().Source.IncrementalAppend !=
parser.CapabilitySupported {
return processResult{}, false
}
path := providerDiscoveredPath(source)
if path == "" {
return processResult{}, false
}
if provider.Definition().Type == parser.AgentCodex &&
(file.ForceParse ||
e.codexIndexSessionNameChanged(path) ||
e.pathNeedsProjectReparse(path)) {
// Codex incremental parsing intentionally preserves head-derived
// metadata. A manual refresh, title change, or stale project needs the
// authoritative full parse, and forceReplace prevents the later DB skip
// gates from swallowing that refresh.
return processResult{forceReplace: true}, false
}
info, err := os.Stat(path)
if err != nil {
return processResult{}, false
}
parseFn := func(
_ string, sessionID string, offset int64,
startOrdinal int, lastEntryUUID string,
) ([]parser.ParsedMessage, []parser.ClaudeSubagentLink, time.Time, int64, *string, error) {
outcome, status, perr := provider.ParseIncremental(
ctx,
parser.IncrementalRequest{
Source: source,
Fingerprint: fingerprint,
SessionID: sessionID,
Offset: offset,
StartOrdinal: startOrdinal,
Machine: e.machine,
LastEntryUUID: lastEntryUUID,
},
)
if perr != nil {
return nil, nil, time.Time{}, 0, nil, perr
}
switch status {
case parser.IncrementalNeedsFullParse:
if outcome.ForceReplace {
// Signal the shared helper to fall back to a
// full parse that replaces stored messages.
return nil, nil, time.Time{}, 0, nil,
parser.ErrClaudeIncrementalNeedsFullParse
}
// A plain full-parse fallback (e.g. DAG detected):
// return a non-fallback error so the helper runs a
// normal full parse without forceReplace.
return nil, nil, time.Time{}, 0, nil, parser.ErrDAGDetected
case parser.IncrementalNoNewData:
return nil, nil, time.Time{}, 0, nil, nil
default:
var terminationStatus *string
if outcome.TerminationStatus != nil {
status := string(*outcome.TerminationStatus)
terminationStatus = &status
}
return outcome.Messages, outcome.SubagentLinks,
outcome.EndedAt, outcome.ConsumedBytes, terminationStatus, nil
}
}
return e.tryIncrementalJSONL(file, info, file.Agent, parseFn)
}
// incrementalParseFunc reads new JSONL lines from a file
// starting at the given byte offset with the given starting
// ordinal and persisted session ID. Returns parsed messages, the latest
// timestamp (endedAt), bytes consumed (relative to offset), an optional
// authoritative termination status, and any error. The consumed count covers
// only complete, valid JSON lines so it can be used as a safe resume offset.
type incrementalParseFunc func(
path string, sessionID string, offset int64,
startOrdinal int, lastEntryUUID string,
) ([]parser.ParsedMessage, []parser.ClaudeSubagentLink, time.Time, int64, *string, error)
// tryIncrementalJSONL attempts an incremental parse of an
// append-only JSONL file by reading only bytes appended since
// the last sync. Returns (result, true) on success, or
// (zero, false) to fall through to a full parse. Falls back
// to full parse when the file maps to multiple DB sessions
// (e.g. Claude DAG forks).
func (e *Engine) tryIncrementalJSONL(
file parser.DiscoveredFile,
info os.FileInfo,
agent parser.AgentType,
parseFn incrementalParseFunc,
) (processResult, bool) {
if e.forceParse { // parse-diff: never produce append deltas
return processResult{}, false
}
lookupPath := file.Path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(file.Path)
}
inc, ok := e.db.GetSessionForIncremental(lookupPath)
if !ok || inc.FileSize <= 0 {
return processResult{}, false
}
// A session archived before the cwd allow-list was configured
// must not keep growing through the append path, which bypasses
// the prepareSessionWrite veto. Fall back to the full parse path
// so the same veto applies; it also re-derives the cwd from the
// whole file, which covers a stored cwd that predates cwd capture.
if !e.cwdFilter.allows(inc.Cwd) {
return processResult{}, false
}
// Existing rows from an older parser lack new metadata
// columns. Force a full parse so the rewrite picks them
// up rather than appending new rows on top of stale ones.
if e.db.GetSessionDataVersion(inc.ID) <
db.CurrentDataVersion() {
return processResult{}, false
}
// Claude-only: if the stored preview is empty despite the
// session already having user turns, the parser skipped
// every user message so far (e.g. a session that opens with
// /clear or /effort). Fall back to a full parse so any real
// user message appended this sync becomes first_message.
//
// Other agents can legitimately have UserMsgCount > 0 with
// an empty first_message — for example Codex inserts orphan
// subagent notifications as Role=user messages that bypass
// firstMessage — so this fall-through is gated on Claude.
if agent == parser.AgentClaude &&
inc.FirstMessage == "" && inc.UserMsgCount > 0 {
return processResult{}, false
}
currentSize := info.Size()
// A prior sync that stored no message rows has no safe append
// boundary. Rewritten files can grow in place and keep the same
// identity, which makes a full-file replacement look like an
// append from the old file_size offset.
if inc.MsgCount == 0 {
return processResult{}, false
}
// If the file was replaced (different inode/device), fall
// back to a full parse so we don't append on top of stale
// state. Only check when both sides have a known identity
// (non-zero); zeros mean the data is missing or the
// platform doesn't expose inode/device (Windows).
if e.pathRewriter == nil && inc.FileInode != 0 && inc.FileDevice != 0 {
curInode, curDevice := getFileIdentity(info)
if curInode != 0 && curDevice != 0 &&
(curInode != inc.FileInode ||
curDevice != inc.FileDevice) {
log.Printf(
"incremental %s %s: file identity changed "+
"(inode %d→%d, device %d→%d), full parse",
agent, file.Path,
inc.FileInode, curInode,
inc.FileDevice, curDevice,
)
return processResult{forceReplace: true}, false
}
}
if currentSize < inc.FileSize {
log.Printf(
"incremental %s %s: file truncated from %d to %d, full parse",
agent, file.Path, inc.FileSize, currentSize,
)
return processResult{forceReplace: true}, false
}
if currentSize == inc.FileSize {
if agent == parser.AgentCodex {
// Codex's composite mtime can change when session_index.jsonl does,
// even though the transcript has no new bytes. Let the later Codex
// fingerprint/title check decide whether to skip or full-parse.
return processResult{}, false
}
log.Printf(
"incremental %s %s: file size unchanged at %d but changed since last sync, full parse",
agent, file.Path, currentSize,
)
return processResult{forceReplace: true}, false
}
// Persist the same effective file_mtime a full parse would store. For
// Codex that folds in session_index.jsonl (parser.CodexEffectiveMtime),
// exactly as ParseCodexSession sets File.Mtime; a full sync of the same
// file stores that effective value. Keeping the incremental write on the
// same basis means parse-diff's raced guard -- which reads the freshly
// parsed effective File.Mtime -- compares against a matching stored
// file_mtime no matter whether the last write was incremental or full,
// and shouldSkipCodex's storedMtime==effectiveMtime fast path stays
// accurate. Plain JSONL agents (Claude/Gemini) keep the raw stat.
incMtime := info.ModTime().UnixNano()
if agent == parser.AgentCodex {
incMtime = parser.CodexEffectiveMtime(file.Path, incMtime)
}
newMsgs, links, endedAt, consumed, terminationStatus, err := parseFn(
file.Path, inc.ID, inc.FileSize, inc.NextOrdinal, inc.LastEntryUUID,
)
if err != nil {
if parser.IsIncrementalFullParseFallback(err) {
log.Printf(
"incremental %s %s: %v (explicit full parse fallback)",
agent, file.Path, err,
)
// The fallback fires when appended lines update
// already-stored rows (toolUseResult.agentId
// linkage, same-message.id chunk merging). The
// full parse must replace existing messages —
// otherwise the append-only write path skips
// rows whose ordinal ≤ maxOrd and the updates
// are silently dropped.
return processResult{forceReplace: true}, false
}
log.Printf(
"incremental %s %s: %v (full parse)",
agent, file.Path, err,
)
return processResult{}, false
}
// Use the offset through the last valid JSON line, not
// info.Size(), so partial lines at EOF are retried on
// the next sync.
newOffset := inc.FileSize + consumed
var incHash string
// Refresh the stored content fingerprint on the incremental path. Codex
// needs it for parse-diff's raced-skew detection; Claude needs it so
// providerSingleSessionFresh can compare the stored hash against the
// on-disk bytes and catch a same-size, same-mtime, same-inode in-place
// rewrite that the size/mtime/identity skip signals cannot see.
if agent == parser.AgentCodex || agent == parser.AgentClaude {
if hash, err := ComputeFileHashPrefix(file.Path, newOffset); err == nil {
incHash = hash
}
}
if len(newMsgs) == 0 {
// No new messages, but advance the offset past
// non-message lines (progress events, metadata)
// so they aren't re-read on every sync. Carry
// endedAt forward so session bounds stay current
// with non-message timestamps (e.g. progress).
if consumed > 0 {
return processResult{
incremental: &incrementalUpdate{
sessionID: inc.ID,
project: inc.Project,
machine: inc.Machine,
cwd: inc.Cwd,
links: links,
endedAt: endedAt,
terminationStatus: terminationStatus,
msgCount: inc.MsgCount,
userMsgCount: inc.UserMsgCount,
fileSize: newOffset,
fileMtime: incMtime,
fileHash: incHash,
nextOrdinal: inc.NextOrdinal,
lastEntryUUID: inc.LastEntryUUID,
totalOutputTokens: inc.TotalOutputTokens,
peakContextTokens: inc.PeakContextTokens,
hasTotalOutputTokens: inc.HasTotalOutputTokens,
hasPeakContextTokens: inc.HasPeakContextTokens,
},
}, true
}
// A larger source with no complete record consumed is an unfinished
// append, not evidence that this fingerprint is fully processed. Keep
// the persisted cursor unchanged and suppress the mtime skip entry so a
// completed record is retried even when the writer restores the same
// filesystem timestamp.
return processResult{skip: true, noCacheSkip: true}, true
}
// Claude cross-sync split detection: when the first appended
// assistant message shares its provider message id with the
// last already-stored assistant message for this session, the
// previous sync stopped mid-stream. The incremental path would
// store the new chunk as a separate message instead of merging
// it into the existing one — fall back to a full parse so the
// chunk merge sees the whole run. forceReplace tells the
// downstream write path to use ReplaceSessionMessages: the
// merged tail reuses existing ordinals, so the default
// append-only writeMessages would silently drop it.
if agent == parser.AgentClaude {
first := newMsgs[0]
if first.Role == parser.RoleAssistant &&
first.ClaudeMessageID != "" {
if e.db.LastClaudeMessageID(inc.ID) ==
first.ClaudeMessageID {
log.Printf(
"incremental %s %s: appended chunk shares"+
" message.id with stored tail, full parse",
agent, file.Path,
)
return processResult{forceReplace: true}, false
}
}
}
newUserCount := countUserMsgs(newMsgs)
nextOrdinal := nextParsedOrdinal(inc.NextOrdinal, newMsgs)
lastEntryUUID := lastParsedSourceUUID(inc.LastEntryUUID, newMsgs)
log.Printf(
"incremental %s %s: %d new message(s) "+
"from offset %d",
agent, inc.ID, len(newMsgs), inc.FileSize,
)
totalOut := inc.TotalOutputTokens
peakCtx := inc.PeakContextTokens
hasTotalOut := inc.HasTotalOutputTokens
hasPeakCtx := inc.HasPeakContextTokens
for _, m := range newMsgs {
msgHasCtx, msgHasOut := m.TokenPresence()
// Accumulate from per-message values already bounded to the
// per-message clamp the central pass applies to the stored rows, so
// a corrupt new message cannot inflate the session aggregates past
// what the persisted rows justify (parity with the full path, which
// re-derives message-derived totals from the clamped rows).
if msgHasOut {
totalOut += clampedTokens(m.OutputTokens)
hasTotalOut = true
}
if ctx := clampedTokens(m.ContextTokens); msgHasCtx &&
(!hasPeakCtx || ctx > peakCtx) {
peakCtx = ctx
hasPeakCtx = true
}
}
return processResult{
incremental: &incrementalUpdate{
sessionID: inc.ID,
project: inc.Project,
machine: inc.Machine,
cwd: inc.Cwd,
msgs: newMsgs,
links: links,
endedAt: endedAt,
terminationStatus: terminationStatus,
msgCount: inc.MsgCount + len(newMsgs),
userMsgCount: inc.UserMsgCount + newUserCount,
fileSize: newOffset,
fileMtime: incMtime,
fileHash: incHash,
nextOrdinal: nextOrdinal,
lastEntryUUID: lastEntryUUID,
totalOutputTokens: totalOut,
peakContextTokens: peakCtx,
hasTotalOutputTokens: hasTotalOut,
hasPeakContextTokens: hasPeakCtx,
},
}, true
}
// shouldSkipProviderSourceByDB reports whether a provider-dispatched source is
// already stored at the parsed fingerprint and can be skipped without a reparse.
// It is the engine-side replacement for the DB-aware skip the legacy
// single-session JSONL processors performed, since a provider has no database
// handle. It is scoped to Codex: Codex's effective mtime folds in the shared
// session_index.jsonl sidecar, so a size-and-effective-mtime match plus a
// per-session title check preserves the legacy "skip when only the global index
// advanced but this session's name did not" semantics. Other providers keep
// their existing in-memory skip-cache behavior unchanged.
func (e *Engine) shouldSkipProviderSourceByDB(
file parser.DiscoveredFile, fingerprint parser.SourceFingerprint,
) bool {
if file.Agent != parser.AgentCodex {
return false
}
return e.shouldSkipCodexFingerprint(file.Path, fingerprint)
}
// shouldSkipCodexFingerprint reproduces the legacy shouldSkipCodex decision in
// terms of a provider SourceFingerprint. The fingerprint MTimeNS already folds
// in session_index.jsonl via CodexEffectiveMtime, so:
// - a stored size/hash mismatch or stale data version forces a reparse;
// - an exact effective-mtime match skips;
// - an effective mtime ahead of the stored mtime driven only by the index
// (the raw transcript mtime is still at or below the stored mtime) skips
// unless this session's stored title differs from the current index title.
func (e *Engine) shouldSkipCodexFingerprint(
path string, fingerprint parser.SourceFingerprint,
) bool {
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
storedSize, storedMtime, ok := e.db.GetFileInfoByPath(lookupPath)
if !ok || storedSize != fingerprint.Size {
return false
}
if !e.providerFingerprintHashMatchesDB(
parser.AgentCodex, lookupPath, fingerprint,
) {
return false
}
if project, ok := e.db.GetProjectByPath(lookupPath); ok &&
parser.NeedsProjectReparse(project) {
return false
}
if e.db.GetDataVersionByPath(lookupPath) <
db.CurrentDataVersion() {
return false
}
effectiveMtime := fingerprint.MTimeNS
if storedMtime == effectiveMtime {
return true
}
fileMtime := effectiveMtime
if info, err := os.Stat(path); err == nil {
fileMtime = info.ModTime().UnixNano()
}
return effectiveMtime > storedMtime &&
fileMtime <= storedMtime &&
!e.codexIndexSessionNameChanged(path)
}
// codexIndexNeedsRefreshSince reports whether a Codex session whose transcript
// predates the cutoff still needs a refresh because its session_index.jsonl
// title changed at or after the cutoff. It compares the index title to the
// stored session_name directly rather than gating on indexMtime > storedMtime:
// the incremental write folds the index mtime into the stored file_mtime, so a
// title-only rename whose index mtime is <= that stored value would otherwise
// be filtered out and the stale title would never resolve.
func (e *Engine) codexIndexNeedsRefreshSince(
path string, cutoffNs int64,
) bool {
indexMtime := parser.CodexEffectiveMtime(path, 0)
if indexMtime == 0 || indexMtime < cutoffNs {
return false
}
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
if _, _, ok := e.db.GetFileInfoByPath(lookupPath); !ok {
return false
}
return e.codexIndexSessionNameChanged(path)
}
func (e *Engine) codexIndexSessionNameChanged(path string) bool {
uuid := parser.CodexSessionUUIDFromFilename(filepath.Base(path))
if uuid == "" {
return false
}
currentName := parser.LookupCodexThreadName(path, uuid)
storedName, found, err := e.db.GetSessionName(
context.Background(), e.idPrefix+"codex:"+uuid,
)
if err != nil || !found {
return true
}
return codexSessionNameDiffers(storedName, currentName)
}
// codexCachedIndexSessionNameChanged limits title-based cache invalidation to
// sources that already have stored session state. A cached parse failure has no
// title to refresh and must retain its retry-suppression semantics.
func (e *Engine) codexCachedIndexSessionNameChanged(path string) bool {
lookupPath := path
if e.pathRewriter != nil {
lookupPath = e.pathRewriter(path)
}
if _, _, ok := e.db.GetFileInfoByPath(lookupPath); !ok {
return false
}
return e.codexIndexSessionNameChanged(path)
}
// classifyCodexIndexPath maps a Codex session_index.jsonl change to the
// session files whose stored title no longer matches the index. The live
// watcher sees this file only because its parent directory is watched
// shallowly (see ResolveCodexShallowWatchRoots); without this translation a
// title-only rename would not refresh until the next periodic sync, since the
// session transcript itself is untouched.
func (e *Engine) classifyCodexIndexPath(
path string,
) []parser.DiscoveredFile {
if filepath.Base(path) != parser.CodexSessionIndexFilename {
return nil
}
indexDir := filepath.Dir(path)
var sessionRoots []string
for _, agDir := range e.agentDirs[parser.AgentCodex] {
if agDir != "" && filepath.Dir(agDir) == indexDir {
sessionRoots = append(sessionRoots, agDir)
}
}
if len(sessionRoots) == 0 {
return nil
}
parser.EvictCodexSessionIndex(path)
titles := parser.CodexSessionIndexTitles(path)
if len(titles) == 0 {
return nil
}
var out []parser.DiscoveredFile
for uuid, title := range titles {
if !e.codexStoredNameDiffers(uuid, title) {
continue
}
var candidates []parser.DiscoveredFile
for _, root := range sessionRoots {
if src := e.codexSourceFileForUUID(root, uuid); src != "" {
candidates = append(candidates, parser.DiscoveredFile{
Path: src,
Agent: parser.AgentCodex,
})
}
}
if len(candidates) == 0 {
continue
}
// A UUID can exist in both sessions/ and archived_sessions/.
// Prefer the path the DB already tracks so a title rename does
// not reparse a stale duplicate over the stored copy.
chosen := pickPreferredCodexDiscoveredFile(e.db, candidates)
// Pin the provider source to the chosen path and route it through the
// provider so processProviderFile parses exactly this copy instead of
// re-canonicalizing the UUID to the preferred dated layout, which would
// undo the DB-aware selection above.
chosen.ProviderProcess = true
chosen.ProviderSource = e.codexPinnedProviderSource(chosen.Path)
out = append(out, chosen)
}
return out
}
// codexSourceFileForUUID resolves a Codex session UUID to its on-disk
// transcript path under a single sessions root, preferring the live dated
// layout over a flat archived entry. It scopes a Codex provider to that one
// root so the provider's cross-root live-over-archived canonicalization does
// not collapse a per-root duplicate; classifyCodexIndexPath then applies its
// own DB-aware preference across the per-root candidates. Returns "" when the
// provider, source lookup, or path resolution fails.
func (e *Engine) codexSourceFileForUUID(root, uuid string) string {
factory, ok := e.providerFactories[parser.AgentCodex]
if !ok || factory == nil {
return ""
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: []string{root},
Machine: e.machine,
})
source, found, err := provider.FindSource(
context.Background(),
parser.FindSourceRequest{RawSessionID: uuid},
)
if err != nil || !found {
return ""
}
return providerDiscoveredPath(source)
}
// codexPinnedProviderSource builds a Codex provider SourceRef pinned to the
// exact path, bypassing the provider's live-over-archived canonicalization. It
// is used when the engine's DB-aware or mtime-aware logic has already chosen
// which on-disk copy of a duplicated UUID to parse, so processProviderFile
// parses that copy instead of the provider's preferred dated layout. Returns
// nil when the Codex provider or the path's source shape is unavailable.
func (e *Engine) codexPinnedProviderSource(path string) *parser.SourceRef {
factory, ok := e.providerFactories[parser.AgentCodex]
if !ok || factory == nil {
return nil
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: e.agentDirs[parser.AgentCodex],
Machine: e.machine,
})
pinner, ok := provider.(interface {
SourceRefForPath(string) (parser.SourceRef, bool)
})
if !ok {
return nil
}
source, ok := pinner.SourceRefForPath(path)
if !ok {
return nil
}
return &source
}
// codexStoredNameDiffers reports whether the stored session_name for a Codex
// session differs from the given index title. Unknown sessions return false:
// a brand-new session is synced through its own transcript event, not the
// index, so the index path only refreshes renames of already-synced sessions.
func (e *Engine) codexStoredNameDiffers(uuid, indexTitle string) bool {
return e.codexStoredNameDiffersBySessionID(
e.idPrefix+"codex:"+uuid, indexTitle, false,
)
}
func (e *Engine) codexStoredNameDiffersBySessionID(
sessionID, indexTitle string,
missingDiffers bool,
) bool {
storedName, found, err := e.db.GetSessionName(
context.Background(), sessionID,
)
if err != nil || !found {
return missingDiffers
}
return codexSessionNameDiffers(storedName, indexTitle)
}
func codexSessionNameDiffers(storedName, indexTitle string) bool {
return strings.TrimSpace(indexTitle) != strings.TrimSpace(storedName)
}
func pickPreferredCodexDiscoveredFile(
database *db.DB, candidates []parser.DiscoveredFile,
) parser.DiscoveredFile {
if len(candidates) == 0 {
return parser.DiscoveredFile{}
}
if id := parser.CodexSessionUUIDFromFilename(
filepath.Base(candidates[0].Path),
); id != "" {
sessionID := "codex:" + id
for _, candidate := range candidates {
storedPath := database.GetSessionFilePath(applyIDPrefixToID(
discoveredFileIDPrefix(candidate), sessionID,
))
if storedPath == "" {
continue
}
storedPath = filepath.Clean(storedPath)
for _, candidate := range candidates {
if filepath.Clean(candidate.Path) == storedPath {
return candidate
}
}
}
}
chosen := candidates[0]
for _, candidate := range candidates[1:] {
if preferDiscoveredFile(candidate, chosen) {
chosen = candidate
}
}
return chosen
}
// copilotEffectiveMtime returns max(events.jsonl mtime,
// workspace.yaml mtime). For flat .jsonl sessions (no
// workspace.yaml sibling) it returns the events.jsonl mtime.
func copilotEffectiveMtime(eventsPath string, info os.FileInfo) int64 {
m := info.ModTime().UnixNano()
if filepath.Base(eventsPath) != "events.jsonl" {
return m
}
yamlPath := filepath.Join(
filepath.Dir(eventsPath), "workspace.yaml",
)
if yi, err := os.Stat(yamlPath); err == nil {
if ym := yi.ModTime().UnixNano(); ym > m {
m = ym
}
}
return m
}
// classifyReasonixPath handles Reasonix session classification as a dedicated
// helper to stay within nilaway limits.
func (e *Engine) classifyReasonixPath(
path string,
) (parser.DiscoveredFile, bool) {
sep := string(filepath.Separator)
for _, reasonixDir := range e.agentDirs[parser.AgentReasonix] {
if reasonixDir == "" {
continue
}
if rel, ok := isUnder(reasonixDir, path); ok {
// Map .jsonl.meta sidecar events to sibling .jsonl
if strings.HasSuffix(path, ".jsonl.meta") {
jsonlPath := strings.TrimSuffix(path, ".meta")
if _, err := os.Stat(jsonlPath); err != nil {
continue
}
path = jsonlPath
rel = strings.TrimSuffix(rel, ".meta")
}
if !strings.HasSuffix(path, ".jsonl") {
continue
}
parts := strings.Split(rel, sep)
// Project sessions: projects/{project}/sessions/{id}.jsonl
// or projects/{project}/sessions/{id}/{id}.jsonl
if len(parts) == 4 && parts[0] == "projects" &&
parts[2] == "sessions" &&
strings.HasSuffix(parts[3], ".jsonl") {
return parser.DiscoveredFile{
Path: path,
Project: parts[1],
Agent: parser.AgentReasonix,
}, true
}
// Project sessions: projects/{project}/sessions/{id}/{id}.jsonl
if len(parts) == 5 && parts[0] == "projects" &&
parts[2] == "sessions" {
base := strings.TrimSuffix(parts[4], ".jsonl")
if base != "" && parts[3] == base {
return parser.DiscoveredFile{
Path: path,
Project: parts[1],
Agent: parser.AgentReasonix,
}, true
}
}
// Global or archive sessions
if len(parts) == 2 {
if (parts[0] == "sessions" || parts[0] == "archive") &&
strings.HasSuffix(parts[1], ".jsonl") {
return parser.DiscoveredFile{
Path: path,
Agent: parser.AgentReasonix,
}, true
}
}
// Nested global or subagent: sessions/{id}/{id}.jsonl or sessions/subagents/{id}.jsonl
if len(parts) == 3 {
base := strings.TrimSuffix(parts[2], ".jsonl")
if parts[0] == "sessions" &&
(parts[1] == "subagents" ||
parts[1] == base) {
if base != "" {
return parser.DiscoveredFile{
Path: path,
Agent: parser.AgentReasonix,
}, true
}
}
}
}
}
return parser.DiscoveredFile{}, false
}
func reasonixEffectiveInfo(path string, info os.FileInfo) os.FileInfo {
size := info.Size()
mtime := info.ModTime().UnixNano()
metaPath := path + ".meta"
if metaInfo, err := os.Stat(metaPath); err == nil {
size += metaInfo.Size()
if metaMtime := metaInfo.ModTime().UnixNano(); metaMtime > mtime {
mtime = metaMtime
}
}
return fakeSnapshotInfo{fSize: size, fMtime: mtime}
}
// vibeEffectiveInfo returns size/mtime for a Vibe session that account
// for the sibling meta.json file: size is the sum of both files, and
// mtime is the larger of the two. Returns info unchanged when meta.json
// is absent or unreadable.
func vibeEffectiveInfo(path string, info os.FileInfo) os.FileInfo {
size := info.Size()
mtime := info.ModTime().UnixNano()
metaPath := filepath.Join(filepath.Dir(path), "meta.json")
if metaInfo, err := os.Stat(metaPath); err == nil {
size += metaInfo.Size()
if metaMtime := metaInfo.ModTime().UnixNano(); metaMtime > mtime {
mtime = metaMtime
}
}
return fakeSnapshotInfo{fSize: size, fMtime: mtime}
}
func commandCodeEffectiveInfo(path string, info os.FileInfo) os.FileInfo {
size := info.Size()
mtime := info.ModTime().UnixNano()
metaPath := strings.TrimSuffix(path, ".jsonl") + ".meta.json"
if metaInfo, err := os.Stat(metaPath); err == nil {
size += metaInfo.Size()
if metaMtime := metaInfo.ModTime().UnixNano(); metaMtime > mtime {
mtime = metaMtime
}
}
return fakeSnapshotInfo{fSize: size, fMtime: mtime}
}
// computeFinalStreak counts trailing consecutive failures
// from the end of the tool call list.
func computeFinalStreak(calls []signals.ToolCallRow) int {
streak := 0
for _, v := range slices.Backward(calls) {
if signals.IsFailure(v) {
streak++
} else {
break
}
}
return streak
}
// RecomputeSignals recomputes signals for a single session
// from existing DB data. Returns nil on success (including
// when the session no longer exists). Returns an error when
// the recompute could not complete -- BackfillSignals uses
// that signal to keep the one-shot completion marker unset
// so the next startup can retry.
func (e *Engine) RecomputeSignals(
ctx context.Context, sessionID string,
) error {
if e.refuseWriteInForceParse("RecomputeSignals") {
return errors.New(
"RecomputeSignals refused on report-only parse-diff engine",
)
}
_, err := e.recomputeSignalsFromDB(ctx, sessionID)
return err
}
// BackfillSignalComputer returns a signal recompute closure for archive
// backfills that releases transient heap after enough loaded content has
// crossed the threshold.
func (e *Engine) BackfillSignalComputer() func(context.Context, string) error {
var release recomputeHeapReleaser
return func(ctx context.Context, sessionID string) error {
if e.refuseWriteInForceParse("BackfillSignalComputer") {
return errors.New(
"BackfillSignalComputer refused on report-only parse-diff engine",
)
}
heapBytes, err := e.recomputeSignalsFromDB(ctx, sessionID)
if err != nil {
return err
}
release.Account(heapBytes)
return nil
}
}
// BackfillProjectIdentitySnapshots reconstructs immutable export evidence from
// stored session metadata. Candidate selection and progress are durable, while
// filesystem and Git discovery happen here so database startup remains cheap.
func (e *Engine) BackfillProjectIdentitySnapshots(ctx context.Context) error {
if e.refuseWriteInForceParse("BackfillProjectIdentitySnapshots") {
return errors.New(
"BackfillProjectIdentitySnapshots refused on report-only parse-diff engine",
)
}
if err := e.db.EnsureProjectIdentityBackfillQueued(ctx); err != nil {
return err
}
status, err := e.db.ProjectIdentityBackfillStatus(ctx)
if err != nil {
return err
}
if status.State == "not_needed" || status.State == "completed" {
return nil
}
if err := e.db.StartProjectIdentityBackfill(ctx); err != nil {
return err
}
log.Printf("project identity backfill: processing %d sessions", status.TotalItems)
afterID := ""
for {
if err := ctx.Err(); err != nil {
return err
}
candidates, err := e.db.ProjectIdentityBackfillCandidatesAfter(ctx, afterID)
if err != nil {
return e.failProjectIdentityBackfill(ctx, err)
}
if len(candidates) == 0 {
if err := e.db.CompleteProjectIdentityBackfill(ctx); err != nil {
return err
}
log.Printf("project identity backfill: completed %d sessions",
status.TotalItems)
return nil
}
observations := make([]export.ProjectIdentityObservation, 0, len(candidates))
for _, session := range candidates {
if err := ctx.Err(); err != nil {
return err
}
observations = append(observations,
e.projectIdentityObservationForBackfill(session))
}
if err := e.db.ApplyProjectIdentityBackfillBatch(ctx, observations); err != nil {
return e.failProjectIdentityBackfill(ctx, err)
}
afterID = candidates[len(candidates)-1].ID
}
}
func (e *Engine) projectIdentityObservationForBackfill(
session db.Session,
) export.ProjectIdentityObservation {
obs, ok := e.projectIdentityObservation(session)
if !ok {
obs = export.ProjectIdentityObservation{
SessionID: session.ID,
Project: strings.TrimSpace(session.Project),
Machine: strings.TrimSpace(session.Machine),
RootPath: strings.TrimSpace(session.Cwd),
GitBranch: strings.TrimSpace(session.GitBranch),
ObservedAt: time.Now().UTC(),
}
if obs.GitBranch != "" {
obs.CheckoutState = export.CheckoutBranch
}
}
return obs
}
func (e *Engine) failProjectIdentityBackfill(
ctx context.Context, cause error,
) error {
if err := e.db.FailProjectIdentityBackfill(ctx, cause); err != nil {
return errors.Join(cause, err)
}
return cause
}
// recomputeSignalsFromDB loads a session's full message history
// and stored metadata, runs the pure in-memory signal compute
// over them, and persists the result. Used when callers don't
// already have the message slice in memory (legacy backfill,
// incremental writes).
func (e *Engine) recomputeSignalsFromDB(
ctx context.Context, sessionID string,
) (int, error) {
sess, err := e.db.GetSessionFull(ctx, sessionID)
if err != nil {
return 0, fmt.Errorf(
"loading session %s: %w", sessionID, err,
)
}
if sess == nil {
return 0, nil
}
msgs, err := e.db.GetAllMessages(ctx, sessionID)
if err != nil {
log.Printf(
"signals: load messages %s: %v",
sessionID, err,
)
return 0, fmt.Errorf(
"loading messages %s: %w", sessionID, err,
)
}
update, findings := computeSignalsAndSecrets(*sess, msgs)
heapBytes := recomputeHeapBytes(msgs, findings)
// Findings persist before the signals update: UpdateSessionSignals
// advances quality_signal_version, which BackfillSignals treats as
// proof the whole compute persisted. Writing it last keeps a
// session whose findings write failed below the current version,
// so the next backfill retries it.
if err := e.db.ReplaceSessionSecretFindings(
sessionID, findings, update.SecretLeakCount, update.SecretsRulesVersion,
); err != nil {
log.Printf("secrets: persist %s: %v", sessionID, err)
return 0, fmt.Errorf("persisting findings %s: %w", sessionID, err)
}
if err := e.db.UpdateSessionSignals(
sessionID, update,
); err != nil {
log.Printf(
"signals: update %s: %v", sessionID, err,
)
return 0, fmt.Errorf(
"updating signals %s: %w", sessionID, err,
)
}
return heapBytes, nil
}
type pendingWrite struct {
sess parser.ParsedSession
msgs []parser.ParsedMessage
usageEvents []parser.ParsedUsageEvent
needsRetry bool
forceReplace bool
// storageTrustPath/State/Snap promote the session's OpenCode
// storage-gate trust after its batch is confirmed fully written.
// Empty for everything else.
storageTrustPath string
storageTrustState string
storageTrustSnap storageTrustSnapshot
}
func dataVersionForWrite(pw pendingWrite) int {
if !pw.needsRetry {
return db.CurrentDataVersion()
}
// Keep successfully written fallback content visible while
// forcing the next sync to retry the higher-resolution source.
v := db.CurrentDataVersion() - 1
if v < 0 {
return 0
}
return v
}
type worktreeProjectResolver func(
machine, cwd, currentProject string,
) (string, bool)
func (e *Engine) loadWorktreeProjectResolver() worktreeProjectResolver {
cache := map[string][]db.WorktreeProjectMapping{}
failed := map[string]bool{}
return func(machine, cwd, currentProject string) (string, bool) {
if machine == "" {
return currentProject, false
}
mappings, ok := cache[machine]
if !ok {
if failed[machine] {
return currentProject, false
}
var err error
mappings, err = e.db.ListActiveWorktreeProjectMappings(
context.Background(), machine,
)
if err != nil {
log.Printf(
"load worktree project mappings for machine %s: %v",
machine, err,
)
failed[machine] = true
return currentProject, false
}
cache[machine] = mappings
}
if len(mappings) == 0 {
return currentProject, false
}
return db.ResolveWorktreeProjectFromSortedMappings(
mappings, cwd, currentProject,
)
}
}
func (e *Engine) writeBatch(
batch []pendingWrite,
writeMode syncWriteMode,
forceReplace bool,
) (writtenSessions, writtenMessages, failedSessions, cwdFiltered int) {
if writeMode == syncWriteBulk {
return e.writeBatchBulk(batch, forceReplace)
}
resolveWorktreeProject := e.loadWorktreeProjectResolver()
for _, pw := range batch {
s, msgs, verdict := e.prepareSessionWrite(
pw, resolveWorktreeProject,
)
if verdict != sessionWriteOK {
if verdict == sessionWriteCwdFiltered {
cwdFiltered++
}
continue
}
// Detect stale parser version BEFORE UpsertSession
// overwrites it. Existing message rows from an
// older parser lack new metadata columns, and newly
// emitted compact-boundary messages can shift the
// ordinal stream — both demand a full rewrite
// rather than the append-only writeMessages path.
stale := false
if existing := e.db.GetSessionDataVersion(s.ID); existing > 0 &&
existing < db.CurrentDataVersion() {
stale = true
}
// UpsertSession first: the session row must exist
// before messages can be inserted (FK constraint).
// This is safe because writeBatch runs full parses
// that always recompute all columns. For
// incremental updates (writeIncremental), messages
// are written first since the session already
// exists.
if err := e.db.UpsertSession(s); err != nil {
if isIntentionalSessionSkip(err) {
if pw.sess.File.Path != "" {
e.cacheSkip(
pw.sess.File.Path,
pw.sess.File.Mtime,
pw.sess.File.Hash,
)
}
continue
}
log.Printf("upsert session %s: %v", s.ID, err)
failedSessions++
continue
}
if err := e.writeProjectIdentityObservation(
context.Background(), s,
); err != nil {
log.Printf(
"write project identity observation for %s: %v",
s.ID, err,
)
}
replaceMessages := shouldReplaceFullParseMessages(
pw, forceReplace, stale,
)
update, findings := computeSignalsAndSecrets(s, msgs)
var werr error
if replaceMessages {
werr = e.db.ReplaceSessionContent(s.ID, msgs, update, findings)
} else {
werr = e.writeMessages(s.ID, msgs)
}
if werr != nil {
log.Printf(
"write messages for %s: %v",
s.ID, werr,
)
failedSessions++
continue
}
if err := e.db.ReplaceSessionUsageEvents(
s.ID, e.usageEventsForWrite(s.ID, pw.usageEvents),
); err != nil {
log.Printf(
"write usage events for %s: %v",
s.ID, err,
)
failedSessions++
continue
}
// Advance data_version only after the message write
// succeeded. UpsertSession deliberately does not
// touch this column so a transient write failure
// won't leave the session marked at the current
// parser version with stale messages.
if err := e.db.SetSessionDataVersion(
s.ID, dataVersionForWrite(pw),
); err != nil {
log.Printf(
"set data_version for %s: %v", s.ID, err,
)
}
if !replaceMessages {
// Same ordering contract as recomputeSignalsFromDB: the
// version-advancing signals update only runs after findings
// persisted, so a partial failure leaves the session below
// the current version for the startup backfill to retry.
if err := e.db.ReplaceSessionSecretFindings(
s.ID, findings, update.SecretLeakCount,
update.SecretsRulesVersion); err != nil {
log.Printf("secrets: persist %s: %v", s.ID, err)
} else if err := e.db.UpdateSessionSignals(s.ID, update); err != nil {
log.Printf("signals: update %s: %v", s.ID, err)
}
}
writtenSessions++
writtenMessages += len(msgs)
}
return writtenSessions, writtenMessages, failedSessions, cwdFiltered
}
// sessionWriteVerdict says whether prepareSessionWrite produced a
// writable session and, when it did not, why. The cwd-filter veto is
// distinguished from archive-preserve vetoes so sync stats can count
// filtered sessions: a resync where every discovered session is
// filtered must read as intentional, not as an empty rebuild.
type sessionWriteVerdict int
const (
sessionWriteOK sessionWriteVerdict = iota
sessionWritePreserved
sessionWriteCwdFiltered
)
func (e *Engine) prepareSessionWrite(
pw pendingWrite,
resolveWorktreeProject worktreeProjectResolver,
) (db.Session, []db.Message, sessionWriteVerdict) {
msgs := toDBMessages(pw, e.blockedResultCategories)
s := toDBSession(pw)
applySessionMessageDerivedFields(
&s,
msgs,
pw.sess.CountsAuthoritative,
)
e.applyRemoteRewrites(&s, msgs)
if s.Cwd != "" && resolveWorktreeProject != nil {
if mapped, ok := resolveWorktreeProject(
s.Machine, s.Cwd, s.Project,
); ok {
s.Project = mapped
}
}
// Veto sessions outside the configured cwd allow-list before any
// preserve/merge handling so a filtered session is not written by
// any downstream path.
if !e.cwdFilter.allows(s.Cwd) {
return db.Session{}, nil, sessionWriteCwdFiltered
}
if e.shouldPreserveOpenCodeFormatArchive(
pw.sess.Agent, pw.sess.File.Path, s.ID,
pw.sess.File.Mtime, derefString(s.FileHash), msgs,
) {
return db.Session{}, nil, sessionWritePreserved
}
if mergedMsgs, preserve, archived := e.reconcileVisualStudioCopilotArchive(
pw.sess.Agent, s.ID, pw.sess.File.Size, msgs,
); preserve {
return db.Session{}, nil, sessionWritePreserved
} else if mergedMsgs != nil {
parsedMsgs := msgs
msgs = mergedMsgs
applyVisualStudioCopilotArchiveSessionFields(
&s, archived, parsedMsgs, msgs,
)
applySessionMessageDerivedFields(
&s,
msgs,
pw.sess.CountsAuthoritative,
)
applySessionTokenTotalsFromMessages(&s, msgs)
}
// Snapshot, before sanitizing, whether the session's token aggregates
// are derived from the per-message rows or the per-usage-event rows, by
// matching the stored value against each source's raw sum/max. Aggregates
// set directly from a session-level usage summary -- agents like
// Warp/Vibe/Hermes/Zed -- must survive the per-row clamp untouched.
// Source=="session" usage events mirror those same summary totals, so
// exclude them from the event-derived detector and re-clamp path.
msgTotal, msgHasOut, msgPeak, msgHasCtx := messageTokenTotals(msgs)
evtTotal, evtHasOut, evtPeak, evtHasCtx := usageEventTokenTotals(
pw.usageEvents, false,
)
totalFromMsgs := s.HasTotalOutputTokens == msgHasOut &&
s.TotalOutputTokens == msgTotal
totalFromEvts := s.HasTotalOutputTokens == evtHasOut &&
s.TotalOutputTokens == evtTotal
peakFromMsgs := s.HasPeakContextTokens == msgHasCtx &&
s.PeakContextTokens == msgPeak
peakFromEvts := s.HasPeakContextTokens == evtHasCtx &&
s.PeakContextTokens == evtPeak
// Central validation/sanitization pass: every session write flows
// through here so all agents are covered uniformly. The returned fix
// counts and the parser malformed-line count are accumulated per
// agent for the sync summary's anomaly section.
vs := validateAndSanitize(&s, msgs, nil)
e.anomalies.recordSanitize(vs)
e.anomalies.recordMalformedLines(
s.Agent, pw.sess.File.Path, s.ParserMalformedLines,
)
// An Antigravity session decoded from an unrecognized (newer) schema
// carries an "agy-schema:" source_version; count it as an early warning
// that a new agy build may have broken the heuristic decode. Reuse the
// single-source-of-truth rule so the agent gate stays in one place.
if parser.DecodeConfidence(s.Agent, s.SourceVersion) == parser.DecodeConfidenceLow {
e.anomalies.recordUnknownSchemaSession(s.Agent)
}
// An Antigravity session whose gen_metadata table carried rows but decoded
// into zero usage events warns that a newer agy build may have changed the
// gen_metadata wire format the token-block heuristic depends on. The flag
// is set by the parsers from the final usageEvents, so a sidecar-rescued
// session is not counted here.
if pw.sess.GenMetadataWithoutUsage {
e.anomalies.recordGenMetadataWithoutUsageSession(s.Agent)
}
// A per-row token clamp must not leave an inflated value stranded in a
// row-derived session total while the row that produced it was clamped.
// Re-derive a matched aggregate from its now-clamped source (messages
// clamped above; usage events clamped on the fly the same way
// toDBUsageEvents will store them). Summary-derived aggregates match
// neither source and are left as-is. The sum is re-summed from clamped
// rows rather than clamped to the per-row bound, so a legitimately large
// total over many rows is preserved. Re-deriving is a no-op when nothing
// was clamped, keeping the pass idempotent. Messages take precedence when
// both sources match (identical values).
if totalFromMsgs {
t, h, _, _ := messageTokenTotals(msgs)
s.TotalOutputTokens, s.HasTotalOutputTokens = t, h
} else if totalFromEvts {
t, h, _, _ := usageEventTokenTotals(pw.usageEvents, true)
s.TotalOutputTokens, s.HasTotalOutputTokens = t, h
}
if peakFromMsgs {
_, _, p, h := messageTokenTotals(msgs)
s.PeakContextTokens, s.HasPeakContextTokens = p, h
} else if peakFromEvts {
_, _, p, h := usageEventTokenTotals(pw.usageEvents, true)
s.PeakContextTokens, s.HasPeakContextTokens = p, h
}
return s, msgs, sessionWriteOK
}
func applySessionMessageDerivedFields(
s *db.Session,
msgs []db.Message,
countsAuthoritative bool,
) {
if !countsAuthoritative {
s.MessageCount, s.UserMessageCount = postFilterCounts(msgs)
}
s.IsAutomated = db.IsAutomatedTranscript(
s.UserMessageCount, msgs, s.FirstMessage,
)
}
// messageTokenTotals computes the message-derived session token
// aggregates: the sum of per-message output tokens and the peak
// per-message context tokens, each with a presence flag. It is the
// canonical derivation shared by applySessionTokenTotalsFromMessages and
// the post-sanitize reconciliation that re-derives message-derived totals
// from the clamped rows. Absent values return 0 with a false presence.
func messageTokenTotals(
msgs []db.Message,
) (totalOut int, hasOut bool, peakCtx int, hasCtx bool) {
for _, msg := range msgs {
if msg.HasOutputTokens {
hasOut = true
totalOut += msg.OutputTokens
}
if msg.HasContextTokens {
hasCtx = true
if msg.ContextTokens > peakCtx {
peakCtx = msg.ContextTokens
}
}
}
return totalOut, hasOut, peakCtx, hasCtx
}
func applySessionTokenTotalsFromMessages(s *db.Session, msgs []db.Message) {
totalOut, hasOut, peakCtx, hasCtx := messageTokenTotals(msgs)
s.TotalOutputTokens = totalOut
s.HasTotalOutputTokens = hasOut
s.PeakContextTokens = peakCtx
s.HasPeakContextTokens = hasCtx
}
// usageEventTokenTotals computes event-derived session token aggregates through
// parser.UsageEventTokenAggregate -- the same rollup per-turn event parsers use
// to populate stored session totals (positive output summed, peak full context
// = input + cache-creation + cache-read where positive). Session-summary usage
// events mirror parser summary totals rather than per-turn rows, so they are
// excluded from this detector and re-clamp path. When clamp is true each
// included event token field is first bounded to the per-row plausibility cap,
// matching how sanitizeUsageEvent bounds the stored usage_event row.
func usageEventTokenTotals(
events []parser.ParsedUsageEvent, clamp bool,
) (totalOut int, hasOut bool, peakCtx int, hasCtx bool) {
rolled := make([]parser.ParsedUsageEvent, 0, len(events))
for _, ev := range events {
if ev.Source == "session" {
continue
}
rolled = append(rolled, ev)
}
if clamp {
for i, ev := range rolled {
ev.InputTokens = clampedTokens(ev.InputTokens)
ev.OutputTokens = clampedTokens(ev.OutputTokens)
ev.CacheCreationInputTokens = clampedTokens(
ev.CacheCreationInputTokens,
)
ev.CacheReadInputTokens = clampedTokens(ev.CacheReadInputTokens)
rolled[i] = ev
}
}
return parser.UsageEventTokenAggregate(rolled)
}
func applyVisualStudioCopilotArchiveSessionFields(
s *db.Session, archived *db.Session,
parsedMsgs, mergedMsgs []db.Message,
) {
if archived == nil {
return
}
archiveExtendsBounds := sessionTimeBefore(
archived.StartedAt, s.StartedAt,
) || sessionTimeAfter(archived.EndedAt, s.EndedAt)
if !visualStudioCopilotMergedFirstMessageFromParsed(
parsedMsgs, mergedMsgs,
) {
s.FirstMessage = cloneStringPtr(archived.FirstMessage)
}
if archiveExtendsBounds || stringPtrEmpty(s.SessionName) {
s.SessionName = cloneStringPtr(archived.SessionName)
}
s.StartedAt = earlierSessionTime(archived.StartedAt, s.StartedAt)
s.EndedAt = laterSessionTime(archived.EndedAt, s.EndedAt)
}
func visualStudioCopilotMergedFirstMessageFromParsed(
parsed, merged []db.Message,
) bool {
if len(parsed) == 0 || len(merged) == 0 {
return false
}
mergedFirst := merged[0]
for _, parsedMsg := range parsed {
if visualStudioCopilotMessagePresenceKey(parsedMsg) !=
visualStudioCopilotMessagePresenceKey(mergedFirst) {
continue
}
return !visualStudioCopilotMessageLooksIncomplete(
parsedMsg, mergedFirst,
) && !visualStudioCopilotMessageHasArchiveUpdate(
mergedFirst, parsedMsg,
)
}
return false
}
func stringPtrEmpty(v *string) bool {
return v == nil || strings.TrimSpace(*v) == ""
}
func cloneStringPtr(v *string) *string {
if v == nil {
return nil
}
clone := *v
return &clone
}
func sessionTimeBefore(a, b *string) bool {
return sessionTimeCompares(a, b, func(aTime, bTime time.Time) bool {
return aTime.Before(bTime)
})
}
func sessionTimeAfter(a, b *string) bool {
return sessionTimeCompares(a, b, func(aTime, bTime time.Time) bool {
return aTime.After(bTime)
})
}
func sessionTimeCompares(
a, b *string, compare func(time.Time, time.Time) bool,
) bool {
if a == nil || b == nil {
return a != nil && b == nil
}
aTime, aErr := time.Parse(time.RFC3339Nano, *a)
bTime, bErr := time.Parse(time.RFC3339Nano, *b)
if aErr != nil || bErr != nil {
return false
}
return compare(aTime, bTime)
}
func earlierSessionTime(a, b *string) *string {
return chooseSessionTime(a, b, func(aTime, bTime time.Time) bool {
return aTime.Before(bTime)
})
}
func laterSessionTime(a, b *string) *string {
return chooseSessionTime(a, b, func(aTime, bTime time.Time) bool {
return aTime.After(bTime)
})
}
func chooseSessionTime(
a, b *string, chooseA func(time.Time, time.Time) bool,
) *string {
switch {
case a == nil:
return cloneStringPtr(b)
case b == nil:
return cloneStringPtr(a)
}
aTime, aErr := time.Parse(time.RFC3339Nano, *a)
bTime, bErr := time.Parse(time.RFC3339Nano, *b)
switch {
case aErr != nil:
return cloneStringPtr(b)
case bErr != nil:
return cloneStringPtr(a)
case chooseA(aTime, bTime):
return cloneStringPtr(a)
default:
return cloneStringPtr(b)
}
}
// reconcileVisualStudioCopilotArchive returns either a preserved-archive skip
// or a merged transcript for an incomplete Visual Studio Copilot reparse. A
// conversation's transcript is rebuilt from every sibling trace file and
// written with full message replacement, so when a sibling is rotated away or
// deleted the reparse can see fewer spans or weaker span metadata and would
// otherwise drop messages and tool results already stored in SQLite. If a
// remaining trace gained richer data or new messages, merge those updates into
// the archived transcript while retaining archived-only messages.
func (e *Engine) reconcileVisualStudioCopilotArchive(
agent parser.AgentType, sessionID string,
currentSize int64, currentMsgs []db.Message,
) (merged []db.Message, preserve bool, archived *db.Session) {
if agent != parser.AgentVSCopilot {
return nil, false, nil
}
stored, err := e.db.GetSessionFull(context.Background(), sessionID)
if err != nil || stored == nil {
return nil, false, nil
}
storedSize := derefInt64(stored.FileSize)
storedMsgs, err := e.db.GetAllMessages(context.Background(), sessionID)
if err != nil || len(storedMsgs) == 0 {
return nil, false, nil
}
decision := visualStudioCopilotArchiveDecision(
currentMsgs, storedMsgs,
)
if decision.preserve {
log.Printf(
"preserve %s %s: reparse looks incomplete relative to archived "+
"transcript (%d stored messages, %d parsed messages, "+
"composite trace %d->%d bytes)",
agent, sessionID, len(storedMsgs), len(currentMsgs),
storedSize, currentSize,
)
return storedMsgs, false, stored
}
if decision.merged != nil {
log.Printf(
"merge %s %s: reparse updated archived messages while "+
"retaining archived transcript rows (%d stored "+
"messages, %d parsed messages, composite trace "+
"%d->%d bytes)",
agent, sessionID, len(storedMsgs), len(currentMsgs),
storedSize, currentSize,
)
return decision.merged, false, stored
}
return nil, false, nil
}
type visualStudioCopilotArchiveReconcile struct {
preserve bool
merged []db.Message
}
func visualStudioCopilotArchiveDecision(
parsed, stored []db.Message,
) visualStudioCopilotArchiveReconcile {
if len(stored) == 0 {
return visualStudioCopilotArchiveReconcile{}
}
if parsed == nil {
return visualStudioCopilotArchiveReconcile{preserve: true}
}
storedByKey := make(map[string][]int, len(stored))
for i, msg := range stored {
key := visualStudioCopilotMessagePresenceKey(msg)
storedByKey[key] = append(storedByKey[key], i)
}
matchedStored := make([]bool, len(stored))
updates := make(map[int]db.Message)
additions := make([]db.Message, 0)
hasIncomplete := false
for _, parsedMsg := range parsed {
key := visualStudioCopilotMessagePresenceKey(parsedMsg)
candidates := storedByKey[key]
if len(candidates) == 0 {
additions = append(additions, parsedMsg)
continue
}
storedIndex := candidates[0]
storedByKey[key] = candidates[1:]
matchedStored[storedIndex] = true
storedMsg := stored[storedIndex]
incomplete := visualStudioCopilotMessageLooksIncomplete(
parsedMsg, storedMsg,
)
if incomplete {
hasIncomplete = true
}
if !incomplete &&
visualStudioCopilotMessageHasArchiveUpdate(
parsedMsg, storedMsg,
) {
updates[storedIndex] = parsedMsg
}
}
fallbackMatched := false
additions, fallbackMatched = visualStudioCopilotResolveArchiveAdditions(
stored, matchedStored, updates, additions, &hasIncomplete,
)
hasArchiveOnly := false
for _, matched := range matchedStored {
if !matched {
hasArchiveOnly = true
break
}
}
if hasIncomplete || hasArchiveOnly || fallbackMatched {
if len(updates) > 0 || len(additions) > 0 ||
(fallbackMatched && !hasIncomplete) {
return visualStudioCopilotArchiveReconcile{
merged: visualStudioCopilotMergeArchiveMessages(
stored, updates, additions,
),
}
}
return visualStudioCopilotArchiveReconcile{preserve: true}
}
return visualStudioCopilotArchiveReconcile{}
}
func visualStudioCopilotResolveArchiveAdditions(
stored []db.Message,
matchedStored []bool,
updates map[int]db.Message,
additions []db.Message,
hasIncomplete *bool,
) ([]db.Message, bool) {
matched := false
unresolved := additions[:0]
for _, parsedMsg := range additions {
storedIndex, ok := visualStudioCopilotArchiveFallbackMatch(
parsedMsg, stored, matchedStored,
)
if !ok {
unresolved = append(unresolved, parsedMsg)
continue
}
matched = true
matchedStored[storedIndex] = true
storedMsg := stored[storedIndex]
incomplete := visualStudioCopilotMessageLooksIncomplete(
parsedMsg, storedMsg,
)
if incomplete {
*hasIncomplete = true
continue
}
update := visualStudioCopilotArchiveFallbackUpdate(
parsedMsg, storedMsg,
)
if visualStudioCopilotMessageHasArchiveUpdate(update, storedMsg) {
updates[storedIndex] = update
}
}
return unresolved, matched
}
func visualStudioCopilotArchiveFallbackMatch(
parsed db.Message,
stored []db.Message,
matchedStored []bool,
) (int, bool) {
match := -1
for i, storedMsg := range stored {
if matchedStored[i] {
continue
}
if !visualStudioCopilotMessagesFallbackMatch(parsed, storedMsg) {
continue
}
if match != -1 {
return 0, false
}
match = i
}
if match == -1 {
return 0, false
}
return match, true
}
func visualStudioCopilotMessagesFallbackMatch(
parsed, stored db.Message,
) bool {
if parsed.Role != stored.Role {
return false
}
if visualStudioCopilotMessagesShareToolIdentity(parsed, stored) {
return true
}
return visualStudioCopilotMessagesShareContentIdentity(parsed, stored)
}
func visualStudioCopilotMessagesShareToolIdentity(
parsed, stored db.Message,
) bool {
if len(parsed.ToolCalls) == 0 || len(stored.ToolCalls) == 0 {
return false
}
parsedIDs := make(map[string]string, len(parsed.ToolCalls))
for _, call := range parsed.ToolCalls {
id := strings.TrimSpace(call.ToolUseID)
if id == "" {
continue
}
parsedIDs[id] = strings.TrimSpace(call.ToolName)
}
for _, call := range stored.ToolCalls {
id := strings.TrimSpace(call.ToolUseID)
if id == "" {
continue
}
parsedName, ok := parsedIDs[id]
if !ok {
continue
}
storedName := strings.TrimSpace(call.ToolName)
if parsedName != "" && storedName != "" &&
parsedName != storedName {
continue
}
return true
}
return false
}
func visualStudioCopilotMessagesShareContentIdentity(
parsed, stored db.Message,
) bool {
if len(parsed.ToolCalls) > 0 || len(stored.ToolCalls) > 0 {
return false
}
switch parsed.Role {
case string(parser.RoleAssistant), string(parser.RoleUser):
default:
return false
}
return parsed.Content != "" && parsed.Content == stored.Content
}
func visualStudioCopilotArchiveFallbackUpdate(
parsed, stored db.Message,
) db.Message {
update := parsed
// A duplicate span can be flushed later with a different timestamp; keep
// the archived timestamp as the transcript anchor while taking any richer
// parsed payload such as tool results or token usage.
update.Timestamp = stored.Timestamp
return update
}
func visualStudioCopilotMergeArchiveMessages(
stored []db.Message, updates map[int]db.Message,
additions []db.Message,
) []db.Message {
merged := make([]db.Message, 0, len(stored)+len(additions))
merged = append(merged, stored...)
for index, msg := range updates {
merged[index] = msg
}
merged = append(merged, additions...)
if len(additions) > 0 {
slices.SortStableFunc(
merged, compareVisualStudioCopilotMessageOrder,
)
}
for i := range merged {
merged[i].Ordinal = i
}
return merged
}
func compareVisualStudioCopilotMessageOrder(a, b db.Message) int {
aTime, aOK := visualStudioCopilotMessageTime(a)
bTime, bOK := visualStudioCopilotMessageTime(b)
if aOK && bOK {
switch {
case aTime.Before(bTime):
return -1
case aTime.After(bTime):
return 1
default:
return 0
}
}
if aOK {
return -1
}
if bOK {
return 1
}
return 0
}
func visualStudioCopilotMessageTime(msg db.Message) (time.Time, bool) {
if msg.Timestamp == "" {
return time.Time{}, false
}
parsed, err := time.Parse(time.RFC3339Nano, msg.Timestamp)
if err != nil {
return time.Time{}, false
}
return parsed, true
}
func visualStudioCopilotMessagePresenceKey(msg db.Message) string {
if msg.Timestamp != "" {
return msg.Role + "\x00time\x00" + msg.Timestamp
}
if msg.SourceUUID != "" {
return msg.Role + "\x00source\x00" + msg.SourceUUID
}
return fmt.Sprintf("%s\x00ordinal\x00%d", msg.Role, msg.Ordinal)
}
func visualStudioCopilotMessageLooksIncomplete(
parsed, stored db.Message,
) bool {
if parsed.Role != stored.Role {
return false
}
// Stored rows are sanitized and length-adjusted on write; measure the
// parsed side the same way so a reparse that only stripped control bytes
// is not judged shorter and allowed to bypass archive preservation.
p := sanitizedForArchiveCompare(parsed)
if p.ContentLength < stored.ContentLength {
return true
}
if stored.HasThinking && !p.HasThinking {
return true
}
if stored.HasOutputTokens &&
(!p.HasOutputTokens ||
p.OutputTokens < stored.OutputTokens) {
return true
}
if stored.HasContextTokens &&
(!p.HasContextTokens ||
p.ContextTokens < stored.ContextTokens) {
return true
}
if len(p.ToolCalls) < len(stored.ToolCalls) {
return true
}
if countToolResultEvents(p.ToolCalls) <
countToolResultEvents(stored.ToolCalls) {
return true
}
return countToolResultContentLength(p.ToolCalls) <
countToolResultContentLength(stored.ToolCalls)
}
// sanitizedForArchiveCompare returns a copy of m with the same
// validation/sanitization stored rows receive on write (control runes
// stripped, ContentLength delta-adjusted, tokens clamped), so the VS Copilot
// archive reconcile compares freshly parsed messages against archived rows
// like-for-like. The copy is shallow; sanitizeMessage only rewrites value
// fields, leaving the shared ToolCalls slice untouched.
func sanitizedForArchiveCompare(m db.Message) db.Message {
_ = sanitizeMessage(&m)
return m
}
func visualStudioCopilotMessageHasArchiveUpdate(
parsed, stored db.Message,
) bool {
if parsed.Role != stored.Role {
return false
}
// Stored rows are sanitized and length-adjusted on write, but the parsed
// message still carries raw content here. Compare a sanitized copy so a
// reparse that differs only in stripped control bytes is not treated as
// an archive update, preserving idempotency.
p := sanitizedForArchiveCompare(parsed)
if p.ContentLength > stored.ContentLength {
return true
}
if p.ContentLength == stored.ContentLength &&
p.Content != stored.Content {
return true
}
if p.HasThinking && (!stored.HasThinking ||
p.ThinkingText != stored.ThinkingText) {
return true
}
if p.HasOutputTokens &&
(!stored.HasOutputTokens ||
p.OutputTokens > stored.OutputTokens) {
return true
}
if p.HasContextTokens &&
(!stored.HasContextTokens ||
p.ContextTokens > stored.ContextTokens) {
return true
}
if string(p.TokenUsage) != "" &&
string(p.TokenUsage) != string(stored.TokenUsage) {
return true
}
return visualStudioCopilotToolCallsHaveArchiveUpdate(
p.ToolCalls, stored.ToolCalls,
)
}
func visualStudioCopilotToolCallsHaveArchiveUpdate(
parsed, stored []db.ToolCall,
) bool {
if len(parsed) > len(stored) {
return true
}
for i := 0; i < len(parsed) && i < len(stored); i++ {
if visualStudioCopilotToolCallHasArchiveUpdate(
parsed[i], stored[i],
) {
return true
}
}
return false
}
func visualStudioCopilotToolCallHasArchiveUpdate(
parsed, stored db.ToolCall,
) bool {
if parsed.ResultContentLength > stored.ResultContentLength {
return true
}
if parsed.ResultContentLength == stored.ResultContentLength &&
parsed.ResultContent != "" &&
parsed.ResultContent != stored.ResultContent {
return true
}
if len(parsed.ResultEvents) > len(stored.ResultEvents) {
return true
}
for i := 0; i < len(parsed.ResultEvents) &&
i < len(stored.ResultEvents); i++ {
parsedEvent := parsed.ResultEvents[i]
storedEvent := stored.ResultEvents[i]
if parsedEvent.ContentLength > storedEvent.ContentLength {
return true
}
if parsedEvent.ContentLength == storedEvent.ContentLength &&
parsedEvent.Content != "" &&
parsedEvent.Content != storedEvent.Content {
return true
}
if parsedEvent.Status != "" &&
parsedEvent.Status != storedEvent.Status {
return true
}
}
return false
}
func countToolResultContentLength(calls []db.ToolCall) int {
total := 0
for _, call := range calls {
total += call.ResultContentLength
for _, event := range call.ResultEvents {
total += event.ContentLength
}
}
return total
}
type batchSourceFile struct {
path string
mtime int64
fingerprint string
}
type projectIdentityCacheEntry struct {
rootPath string
repositoryPath string
gitDir string
gitRemoteName string
gitRemote string
remoteResolution export.ProjectResolution
remoteCandidates int
worktreeName string
worktreeRootPath string
worktreeKind export.WorktreeRelationship
expiresAt time.Time
}
type localGitIdentity struct {
rootPath string
repositoryPath string
gitDir string
remotes map[string]string
worktreeKind export.WorktreeRelationship
}
func (e *Engine) writeBatchBulk(
batch []pendingWrite, forceReplace bool,
) (writtenSessions, writtenMessages, failedSessions, cwdFiltered int) {
writes := make([]db.SessionBatchWrite, 0, len(batch))
sources := make(map[string]batchSourceFile, len(batch))
resolveWorktreeProject := e.loadWorktreeProjectResolver()
for _, pw := range batch {
tPrep := time.Now()
s, msgs, verdict := e.prepareSessionWrite(
pw, resolveWorktreeProject,
)
e.phaseStats.PrepNanos.Add(int64(time.Since(tPrep)))
if verdict != sessionWriteOK {
if verdict == sessionWriteCwdFiltered {
cwdFiltered++
}
continue
}
replaceMessages := shouldReplaceFullParseMessages(
pw, forceReplace, false,
)
tScan := time.Now()
update, findings := computeSignalsAndSecrets(s, msgs)
e.phaseStats.ScanNanos.Add(int64(time.Since(tScan)))
writes = append(writes, db.SessionBatchWrite{
Session: s,
Messages: msgs,
UsageEvents: e.usageEventsForWrite(s.ID, pw.usageEvents),
IdentityObservation: identityObservationOrZero(
e.projectIdentityObservation(s),
),
Signals: update,
Findings: findings,
DataVersion: dataVersionForWrite(pw),
ReplaceMessages: replaceMessages,
})
if pw.sess.File.Path != "" {
sources[s.ID] = batchSourceFile{
path: pw.sess.File.Path,
mtime: pw.sess.File.Mtime,
fingerprint: pw.sess.File.Hash,
}
}
}
if len(writes) == 0 {
return 0, 0, 0, cwdFiltered
}
tWrite := time.Now()
result, err := e.db.WriteSessionBatch(writes)
e.phaseStats.WriteNanos.Add(int64(time.Since(tWrite)))
e.phaseStats.Batches.Add(1)
e.phaseStats.WriteBatchSize.Add(int64(len(writes)))
e.phaseStats.BatchedWrites.Add(int64(result.WrittenSessions))
if err != nil {
log.Printf("write session batch: %v", err)
return 0, 0, len(writes), cwdFiltered
}
for _, id := range result.ExcludedIDs {
if source, ok := sources[id]; ok && source.path != "" {
e.cacheSkip(
source.path, source.mtime, source.fingerprint,
)
}
}
for _, err := range result.Errors {
log.Printf("write session batch: %v", err)
}
return result.WrittenSessions,
result.WrittenMessages,
result.FailedSessions,
cwdFiltered
}
func identityObservationOrZero(
obs export.ProjectIdentityObservation,
ok bool,
) export.ProjectIdentityObservation {
if !ok {
return export.ProjectIdentityObservation{}
}
return obs
}
func (e *Engine) projectIdentityObservation(
s db.Session,
) (export.ProjectIdentityObservation, bool) {
project := strings.TrimSpace(s.Project)
machine := strings.TrimSpace(s.Machine)
rootPath := strings.TrimSpace(s.Cwd)
if project == "" || machine == "" {
return export.ProjectIdentityObservation{}, false
}
cached := e.cachedProjectIdentity(machine, rootPath)
obs := export.ProjectIdentityObservation{
SessionID: s.ID,
Project: project,
Machine: machine,
RootPath: cached.rootPath,
ObservedAt: time.Now().UTC(),
}
obs.GitRemoteName = cached.gitRemoteName
obs.GitRemote = cached.gitRemote
obs.RemoteResolution = cached.remoteResolution
obs.RemoteCandidateCount = cached.remoteCandidates
obs.WorktreeName = cached.worktreeName
obs.WorktreeRootPath = cached.worktreeRootPath
obs.RepositoryPath = cached.repositoryPath
obs.WorktreeRelationship = cached.worktreeKind
obs.GitBranch = strings.TrimSpace(s.GitBranch)
if obs.GitBranch != "" {
obs.CheckoutState = export.CheckoutBranch
} else {
obs.CheckoutState, obs.GitBranch = readGitCheckout(cached.gitDir)
}
return obs, true
}
func (e *Engine) cachedProjectIdentity(machine, rootPath string) projectIdentityCacheEntry {
e.projectIdentityMu.Lock()
defer e.projectIdentityMu.Unlock()
if e.projectIdentityCache == nil {
e.projectIdentityCache = make(map[string]projectIdentityCacheEntry)
}
cacheKey := machine + "\x00" + rootPath
now := time.Now()
if cached, ok := e.projectIdentityCache[cacheKey]; ok &&
now.Before(cached.expiresAt) {
return cached
}
identity := projectIdentityCacheEntry{rootPath: rootPath}
// Only probe the local filesystem for sessions recorded on this
// machine: another machine's cwd (e.g. /home/... from a synced Linux
// host) is meaningless here, and on macOS merely stat'ing such paths
// wakes the /home automounter — with tens of thousands of remote
// sessions and a one-minute cache TTL that becomes a sustained
// automountd/opendirectoryd CPU storm.
if e.idPrefix == "" && e.pathRewriter == nil && machine == e.machine {
if normalized, ok, err := export.NormalizeRootPath(rootPath); err == nil && ok {
identity.rootPath = normalized
}
if discovered := discoverLocalGitIdentity(rootPath); discovered.rootPath != "" {
identity.rootPath = discovered.rootPath
identity.repositoryPath = discovered.repositoryPath
identity.gitDir = discovered.gitDir
identity.worktreeRootPath = discovered.rootPath
identity.worktreeName = filepath.Base(discovered.rootPath)
identity.worktreeKind = discovered.worktreeKind
selection := export.ResolveRemoteSelection(discovered.remotes)
identity.remoteResolution = selection.Resolution
if selection.Resolution == export.ProjectResolutionUnknown {
identity.remoteResolution = export.ProjectResolutionResolved
}
identity.remoteCandidates = countNormalizedRemoteCandidates(discovered.remotes)
if selection.Resolution == export.ProjectResolutionResolved {
identity.gitRemoteName = selection.Name
identity.gitRemote = selection.Raw
}
}
}
if identity.worktreeRootPath == "" {
identity.worktreeName = filepath.Base(identity.rootPath)
identity.worktreeRootPath = identity.rootPath
identity.worktreeKind = export.WorktreeUnknown
}
identity.expiresAt = now.Add(projectIdentityCacheTTL)
e.projectIdentityCache[cacheKey] = identity
return identity
}
func (e *Engine) writeProjectIdentityObservation(
ctx context.Context, s db.Session,
) error {
obs, ok := e.projectIdentityObservation(s)
if !ok {
return nil
}
fingerprint := projectIdentityObservationFingerprint(obs)
e.projectIdentityMu.Lock()
if e.projectIdentityWritten == nil {
e.projectIdentityWritten = make(map[string]struct{})
}
if _, ok := e.projectIdentityWritten[fingerprint]; ok {
e.projectIdentityMu.Unlock()
return nil
}
e.projectIdentityMu.Unlock()
if err := e.db.UpsertProjectIdentityObservation(ctx, obs); err != nil {
return err
}
e.projectIdentityMu.Lock()
e.projectIdentityWritten[fingerprint] = struct{}{}
e.projectIdentityMu.Unlock()
return nil
}
func projectIdentityObservationFingerprint(
obs export.ProjectIdentityObservation,
) string {
return strings.Join([]string{
obs.Project,
obs.SessionID,
obs.Machine,
obs.RootPath,
obs.GitRemote,
obs.GitRemoteName,
obs.WorktreeName,
obs.WorktreeRootPath,
obs.RepositoryPath,
string(obs.WorktreeRelationship),
string(obs.CheckoutState),
obs.GitBranch,
string(obs.RemoteResolution),
strconv.Itoa(obs.RemoteCandidateCount),
}, "\x00")
}
func countNormalizedRemoteCandidates(remotes map[string]string) int {
unique := make(map[string]struct{}, len(remotes))
for _, raw := range remotes {
if normalized, ok := export.NormalizeGitRemote(raw); ok {
unique[normalized] = struct{}{}
}
}
return len(unique)
}
func discoverLocalGitIdentity(cwd string) localGitIdentity {
if !safeLocalAbsolutePath(cwd) {
return localGitIdentity{}
}
// Skip macOS automounter namespaces: probing them wakes
// automountd/opendirectoryd for paths that virtually never exist
// locally (see export.IsAutomountNamespacePath).
if export.IsAutomountNamespacePath(runtime.GOOS, filepath.Clean(cwd)) {
return localGitIdentity{}
}
resolved, err := filepath.EvalSymlinks(filepath.Clean(cwd))
if err != nil {
return localGitIdentity{}
}
root := findLocalGitRoot(resolved)
if root == "" {
return localGitIdentity{}
}
gitDir, commonDir, relationship := gitDirectoryContext(root)
result := localGitIdentity{
rootPath: root,
repositoryPath: repositoryPathForGitContext(root, commonDir),
gitDir: gitDir,
worktreeKind: relationship,
}
if commonDir != "" {
result.remotes = readGitRemotes(filepath.Join(commonDir, "config"))
}
return result
}
func safeLocalAbsolutePath(p string) bool {
p = strings.TrimSpace(p)
if p == "" || strings.Contains(p, "://") {
return false
}
if looksWindowsDrivePath(p) {
return runtime.GOOS == "windows" && filepath.IsAbs(p)
}
if looksRemotePrefixedPath(p) {
return false
}
return filepath.IsAbs(p)
}
func looksRemotePrefixedPath(p string) bool {
colon := strings.Index(p, ":")
if colon <= 0 {
return false
}
prefix := p[:colon]
return !strings.ContainsAny(prefix, `/\`)
}
func looksWindowsDrivePath(p string) bool {
if len(p) < 3 || p[1] != ':' {
return false
}
drive := p[0]
if (drive < 'A' || drive > 'Z') && (drive < 'a' || drive > 'z') {
return false
}
return p[2] == '\\' || p[2] == '/'
}
func findLocalGitRoot(start string) string {
dir := filepath.Clean(start)
for {
if info, err := os.Stat(filepath.Join(dir, ".git")); err == nil {
if info.IsDir() || info.Mode().IsRegular() {
return dir
}
}
parent := filepath.Dir(dir)
if parent == dir {
return ""
}
dir = parent
}
}
func gitDirectoryContext(
root string,
) (gitDir, commonDir string, relationship export.WorktreeRelationship) {
gitPath := filepath.Join(root, ".git")
if info, err := os.Stat(gitPath); err == nil && info.IsDir() {
return gitPath, gitPath, export.WorktreeMain
}
data, err := os.ReadFile(gitPath)
if err != nil {
return "", "", export.WorktreeUnknown
}
line := strings.TrimSpace(string(data))
line = strings.TrimPrefix(line, "gitdir:")
line = strings.TrimSpace(line)
if line == "" {
return "", "", export.WorktreeUnknown
}
if !filepath.IsAbs(line) {
line = filepath.Join(root, line)
}
commonDir = line
relationship = export.WorktreeMain
if data, err := os.ReadFile(filepath.Join(line, "commondir")); err == nil {
common := strings.TrimSpace(string(data))
if filepath.IsAbs(common) {
commonDir = common
} else {
commonDir = filepath.Clean(filepath.Join(line, common))
}
relationship = export.WorktreeLinked
}
return filepath.Clean(line), filepath.Clean(commonDir), relationship
}
func repositoryPathForGitContext(root, commonDir string) string {
repositoryPath := commonDir
if commonDir == "" {
repositoryPath = root
} else if filepath.Base(commonDir) == ".git" {
repositoryPath = filepath.Dir(commonDir)
}
if resolved, err := filepath.EvalSymlinks(repositoryPath); err == nil {
return resolved
}
return filepath.Clean(repositoryPath)
}
func readGitCheckout(gitDir string) (export.CheckoutState, string) {
if gitDir == "" {
return export.CheckoutUnknown, ""
}
data, err := os.ReadFile(filepath.Join(gitDir, "HEAD"))
if err != nil {
return export.CheckoutUnknown, ""
}
head := strings.TrimSpace(string(data))
const branchPrefix = "ref: refs/heads/"
if after, ok := strings.CutPrefix(head, branchPrefix); ok {
branch := strings.TrimSpace(after)
if branch != "" {
return export.CheckoutBranch, branch
}
}
if head != "" {
return export.CheckoutDetached, ""
}
return export.CheckoutUnknown, ""
}
func readGitRemotes(configPath string) map[string]string {
data, err := os.ReadFile(configPath)
if err != nil {
return nil
}
remotes := map[string]string{}
var current string
for line := range strings.SplitSeq(string(data), "\n") {
trimmed := strings.TrimSpace(line)
if strings.HasPrefix(trimmed, "[") && strings.HasSuffix(trimmed, "]") {
current = remoteNameFromGitConfigSection(trimmed)
continue
}
if current == "" || !strings.HasPrefix(trimmed, "url") {
continue
}
key, value, ok := strings.Cut(trimmed, "=")
if !ok || strings.TrimSpace(key) != "url" {
continue
}
remotes[current] = strings.TrimSpace(value)
}
return remotes
}
func remoteNameFromGitConfigSection(section string) string {
section = strings.Trim(section, "[]")
if !strings.HasPrefix(section, `remote `) {
return ""
}
name := strings.TrimSpace(strings.TrimPrefix(section, `remote `))
return strings.Trim(name, `"`)
}
func shouldReplaceFullParseMessages(
pw pendingWrite, forceReplace, stale bool,
) bool {
return forceReplace || pw.forceReplace || pw.needsRetry || stale ||
pw.sess.Agent == parser.AgentCowork ||
isOpenCodeFormatStorageAgent(pw.sess.Agent) ||
pw.sess.Agent == parser.AgentVSCopilot ||
pw.sess.Agent == parser.AgentAntigravity ||
pw.sess.Agent == parser.AgentAntigravityCLI ||
pw.sess.Agent == parser.AgentQwenPaw ||
pw.sess.Agent == parser.AgentCortex ||
// Vibe pairs later tool-result carrier records back to an
// earlier assistant tool call. An incremental append would
// only add the new ordinals and leave the existing tool call's
// result_content empty, so force a full replace.
pw.sess.Agent == parser.AgentVibe ||
pw.sess.Agent == parser.AgentReasonix
}
// writeIncremental appends new messages and partially updates
// session metadata without overwriting columns that are not
// recomputed during incremental parsing (e.g. parent_session_id,
// relationship_type). Codex refreshes file_hash because parse-diff
// uses it as the transcript fingerprint for raced-skew detection;
// Claude refreshes it so providerSingleSessionFresh can use the
// stored hash as a content fingerprint against same-size in-place
// rewrites. Other agents pass an empty hash, which COALESCE leaves
// untouched.
func (e *Engine) writeIncremental(
inc *incrementalUpdate,
) error {
// The full path vetoes filtered sessions in prepareSessionWrite;
// this is the equivalent veto at the incremental write seam, so
// no producer can append to a session outside the cwd allow-list.
// tryIncrementalJSONL already refuses such sessions — this guard
// keeps the seam safe for any future producer.
if !e.cwdFilter.allows(inc.cwd) {
log.Printf(
"incremental %s: cwd %q outside the configured "+
"allow-list, skipping append",
inc.sessionID, inc.cwd,
)
return nil
}
dbMsgs := toDBMessages(
pendingWrite{
sess: parser.ParsedSession{ID: inc.sessionID},
msgs: inc.msgs,
},
e.blockedResultCategories,
)
// The incremental append path bypasses prepareSessionWrite, so run
// the central validation/sanitization pass on the new message rows
// here to keep coverage uniform across write paths. The fix counts
// feed the sync summary's anomaly section.
//
// Deliberately only sanitize fixes are recorded here, not malformed-line
// counts. A malformed JSONL line appended to an actively-syncing file is
// skipped by the incremental reader, and incrementalParseFunc carries no
// malformed-line count, so surfacing it on this path would require
// threading a new return value through the incremental parser API across
// every append-only agent. That is intentionally out of scope for this
// best-effort, only-when-nonzero diagnostic: the value is still parsed and
// persisted, and the next full sync (the periodic pass, or any
// parser-version bump that forces a full resync) re-derives the
// malformed-line count for the file. The incremental path therefore
// under-reports a brand-new summary signal by at most one full-sync
// interval; it never loses stored data and is not a regression on any
// prior behavior (no malformed-line count was surfaced anywhere before
// this feature). Full malformed-line coverage on the incremental path is a
// deferred follow-up.
e.anomalies.recordSanitize(validateAndSanitize(nil, dbMsgs, nil))
// Adjust counts for blocked-category filtering.
newTotal, newUser := postFilterCounts(dbMsgs)
filtered := len(inc.msgs) - newTotal
msgCount := inc.msgCount - filtered
userFiltered := countUserMsgs(inc.msgs) - newUser
userMsgCount := inc.userMsgCount - userFiltered
var endedAt *string
if !inc.endedAt.IsZero() {
s := inc.endedAt.Format(time.RFC3339Nano)
endedAt = &s
}
// Run the appended ended_at through the same timestamp plausibility
// check the full path applies in sanitizeSession, so an implausible
// appended timestamp is blanked here instead of persisting via the
// incremental path while a full sync of the same file would blank it
// (an incremental-vs-full parity divergence). The session token
// aggregates (totalOutputTokens/peakContextTokens) are accumulated from
// per-message values already clamped to the per-message bound (see the
// clampedTokens calls feeding this update), so a corrupt new message
// cannot inflate them past what the stored rows justify -- parity with
// the full path, which re-derives message-derived totals from the
// clamped rows. The sum itself is not clamped to the per-message bound,
// since a long session legitimately exceeds it.
endedAt, _ = blankImplausibleTimestampPtr(endedAt)
subagentLinks := make([]db.ToolCallSubagentLink, len(inc.links))
for i, link := range inc.links {
toolCall := db.ToolCall{
ResultContent: parser.DecodeContent(link.ResultContentRaw),
ResultContentLength: link.ResultContentLen,
}
e.anomalies.recordSanitize(db.SanitizeToolCall(&toolCall))
subagentLinks[i] = db.ToolCallSubagentLink{
ToolUseID: link.ToolUseID,
SubagentSessionID: applyIDPrefixToID(
e.idPrefix, link.SubagentSessionID,
),
ResultContent: toolCall.ResultContent,
ResultContentLen: toolCall.ResultContentLength,
HasResult: link.HasResult,
}
}
if err := e.db.WriteSessionIncremental(
inc.sessionID,
dbMsgs,
db.IncrementalSessionUpdate{
EndedAt: endedAt,
TerminationStatus: inc.terminationStatus,
MsgCount: msgCount,
UserMsgCount: userMsgCount,
FileSize: inc.fileSize,
FileMtime: inc.fileMtime,
FileHash: strPtr(inc.fileHash),
NextOrdinal: inc.nextOrdinal,
LastEntryUUID: inc.lastEntryUUID,
TotalOutputTokens: inc.totalOutputTokens,
PeakContextTokens: inc.peakContextTokens,
HasTotalOutputTokens: inc.hasTotalOutputTokens,
HasPeakContextTokens: inc.hasPeakContextTokens,
SubagentLinks: subagentLinks,
BlockedResultCategories: e.blockedResultCategories,
},
); err != nil {
return fmt.Errorf(
"incremental write %s: %w",
inc.sessionID, err,
)
}
if err := e.applyWorktreeMappingToSingleSession(
inc.sessionID,
); err != nil {
return err
}
if err := e.writeProjectIdentityObservation(
context.Background(),
db.Session{
ID: inc.sessionID,
Project: inc.project,
Machine: inc.machine,
Cwd: inc.cwd,
},
); err != nil {
log.Printf(
"incremental project identity observation %s: %v",
inc.sessionID, err,
)
}
// Signal/secret recompute costs O(session history), so it is
// debounced per session instead of running on every appended
// line: the first write after a quiet period recomputes
// inline, writes during a streaming burst coalesce into one
// recompute per interval plus a trailing flush. Recompute
// errors are logged inside recomputeSignalsFromDB and are
// non-fatal; a later write or flush retries.
e.signalSched.markDirty(inc.sessionID)
return nil
}
// writeMessages uses an incremental append when possible.
// Session files are append-only, so if the DB already has
// messages for this session and the new set is larger, we
// only insert the new messages (avoiding expensive FTS5
// delete+reinsert of existing content).
func (e *Engine) writeMessages(
sessionID string, msgs []db.Message,
) error {
maxOrd := e.db.MaxOrdinal(sessionID)
// No existing messages — insert all.
if maxOrd < 0 {
if err := e.db.InsertMessages(msgs); err != nil {
return fmt.Errorf(
"insert messages for %s: %w",
sessionID, err,
)
}
return nil
}
// Find new messages (ordinal > maxOrd).
delta := 0
for i, m := range msgs {
if m.Ordinal > maxOrd {
delta = len(msgs) - i
msgs = msgs[i:]
break
}
}
if delta == 0 {
return nil
}
if err := e.db.InsertMessages(msgs); err != nil {
return fmt.Errorf(
"append messages for %s: %w",
sessionID, err,
)
}
return nil
}
// writeSessionFull upserts a session and does a full
// delete+reinsert of its messages. Used by explicit
// single-session re-syncs where existing content may have
// changed (not just appended).
// writeSessionFull returns nil on success, a session skip
// sentinel for intentional skips, or another error for real
// failures.
func (e *Engine) writeSessionFull(pw pendingWrite) error {
resolveWorktreeProject := e.loadWorktreeProjectResolver()
return e.writeSessionFullWithResolver(pw, resolveWorktreeProject)
}
func (e *Engine) writeSessionFullWithResolver(
pw pendingWrite,
resolveWorktreeProject worktreeProjectResolver,
) error {
s, msgs, verdict := e.prepareSessionWrite(
pw, resolveWorktreeProject,
)
if verdict != sessionWriteOK {
return errSessionPreserved
}
if err := e.db.UpsertSession(s); err != nil {
if isIntentionalSessionSkip(err) {
if pw.sess.File.Path != "" {
e.cacheSkip(
pw.sess.File.Path,
pw.sess.File.Mtime,
pw.sess.File.Hash,
)
}
return err
}
log.Printf("upsert session %s: %v", s.ID, err)
return err
}
update, findings := computeSignalsAndSecrets(s, msgs)
if err := e.db.ReplaceSessionContent(s.ID, msgs, update, findings); err != nil {
log.Printf(
"replace messages for %s: %v",
s.ID, err,
)
return err
}
if err := e.db.ReplaceSessionUsageEvents(
s.ID, e.usageEventsForWrite(s.ID, pw.usageEvents),
); err != nil {
log.Printf(
"replace usage events for %s: %v",
s.ID, err,
)
return err
}
// See writeBatch for why data_version is bumped here
// rather than inside UpsertSession.
if err := e.db.SetSessionDataVersion(
s.ID, dataVersionForWrite(pw),
); err != nil {
log.Printf(
"set data_version for %s: %v", s.ID, err,
)
}
return nil
}
func (e *Engine) shouldPreserveOpenCodeFormatArchive(
agent parser.AgentType, path, sessionID string,
currentMtime int64,
currentHash string,
currentMsgs []db.Message,
) bool {
if !isOpenCodeFormatStorageAgent(agent) {
return false
}
store := e.openCodeArchiveStore
if store == nil {
store = e.db
}
stored, err := store.GetSessionFull(
context.Background(), sessionID,
)
if err != nil || stored == nil {
return false
}
storedHash := derefString(stored.FileHash)
storedPath := derefString(stored.FilePath)
storedMtime := derefInt64(stored.FileMtime)
storedHasStorageFingerprint := hasOpenCodeFormatStorageFingerprint(
agent, storedHash,
)
storedIsSQLiteVirtual := isOpenCodeFormatSQLiteVirtualPath(
agent, storedPath,
)
storedIsStorageArchive := isOpenCodeFormatStoragePath(
agent, storedPath,
) || (storedPath == "" && storedHasStorageFingerprint)
if storedIsSQLiteVirtual {
storedIsStorageArchive = false
}
if isOpenCodeFormatSQLiteVirtualPath(agent, path) &&
!storedIsStorageArchive {
return false
}
storedMsgs, err := store.GetAllMessages(
context.Background(), sessionID,
)
if err != nil || len(storedMsgs) == 0 {
return false
}
// A changed storage fingerprint alone is not enough to
// preserve the archive. OpenCode legitimately rewrites
// live child files in place, so we only preserve when the
// newly parsed transcript also looks incomplete relative
// to what is already archived.
if storedHasStorageFingerprint &&
hasOpenCodeFormatStorageFingerprint(agent, currentHash) &&
!parser.OpenCodeStorageFingerprintMissing(
storedHash, currentHash,
) {
return false
}
if storedIsStorageArchive &&
isOpenCodeFormatSQLiteVirtualPath(agent, path) &&
currentMtime != 0 &&
storedMtime != 0 &&
currentMtime <= storedMtime {
log.Printf(
"skip %s session %s: sqlite fallback is not newer than preserved storage archive",
agent, sessionID,
)
return true
}
if openCodeLegacyArchiveLooksIncomplete(
currentMsgs, storedMsgs,
) {
if hasOpenCodeFormatStorageFingerprint(agent, storedHash) {
log.Printf(
"skip %s session %s: storage fingerprint changed but update looks incomplete relative to archive",
agent, sessionID,
)
} else {
log.Printf(
"skip %s session %s: storage update looks incomplete relative to legacy archive",
agent, sessionID,
)
}
return true
}
return false
}
func isOpenCodeFormatStorageAgent(agent parser.AgentType) bool {
return agent == parser.AgentOpenCode ||
agent == parser.AgentKilo ||
agent == parser.AgentIcodemate ||
agent == parser.AgentMiMoCode
}
func openCodeFormatDBName(agent parser.AgentType) string {
switch agent {
case parser.AgentOpenCode:
return "opencode.db"
case parser.AgentKilo:
return "kilo.db"
case parser.AgentMiMoCode:
return "mimocode.db"
case parser.AgentIcodemate:
return "icodemate.db"
default:
return ""
}
}
func resolveOpenCodeFormatSource(
agent parser.AgentType, dir string,
) parser.OpenCodeSource {
switch agent {
case parser.AgentOpenCode:
return parser.ResolveOpenCodeSource(dir)
case parser.AgentKilo:
return parser.ResolveKiloSource(dir)
case parser.AgentMiMoCode:
return parser.ResolveMiMoCodeSource(dir)
case parser.AgentIcodemate:
return parser.ResolveIcodemateSource(dir)
default:
return parser.OpenCodeSource{}
}
}
func openCodeFormatSourceMtime(
agent parser.AgentType, path string,
) (int64, error) {
switch agent {
case parser.AgentOpenCode:
return parser.OpenCodeSourceMtime(path)
case parser.AgentKilo:
return parser.KiloSourceMtime(path)
case parser.AgentMiMoCode:
return parser.MiMoCodeSourceMtime(path)
case parser.AgentIcodemate:
return parser.IcodemateSourceMtime(path)
default:
return 0, fmt.Errorf("unknown OpenCode-format agent: %s", agent)
}
}
// hasOpenCodeFormatStorageFingerprint reports whether hash is an
// OpenCode storage fingerprint. Kilo reuses OpenCode's storage format
// verbatim, so the same check applies to both agents.
func hasOpenCodeFormatStorageFingerprint(
agent parser.AgentType, hash string,
) bool {
return isOpenCodeFormatStorageAgent(agent) &&
parser.HasOpenCodeStorageFingerprint(hash)
}
func isOpenCodeFormatStoragePath(
agent parser.AgentType, path string,
) bool {
return strings.HasSuffix(path, ".json") &&
!isOpenCodeFormatSQLiteVirtualPath(agent, path)
}
func isOpenCodeFormatSQLiteVirtualPath(
agent parser.AgentType, path string,
) bool {
if !isOpenCodeFormatStorageAgent(agent) {
return false
}
_, _, ok := parser.ParseVirtualSourcePathForBase(
path, openCodeFormatDBName(agent),
)
return ok
}
func derefString(s *string) string {
if s == nil {
return ""
}
return *s
}
func derefInt64(v *int64) int64 {
if v == nil {
return 0
}
return *v
}
func openCodeLegacyArchiveLooksIncomplete(
parsed, stored []db.Message,
) bool {
if parsed == nil {
return len(stored) > 0
}
if len(parsed) < len(stored) {
return true
}
for i := range stored {
if openCodeMessageLooksIncomplete(
parsed[i], stored[i],
) {
return true
}
}
return false
}
func openCodeMessageLooksIncomplete(
parsed, stored db.Message,
) bool {
if parsed.Ordinal != stored.Ordinal ||
parsed.Role != stored.Role {
return false
}
if sanitizedMessageContentLength(parsed) <
sanitizedMessageContentLength(stored) {
return true
}
if parsed.HasThinking != stored.HasThinking &&
stored.HasThinking {
return true
}
if stored.HasOutputTokens &&
(!parsed.HasOutputTokens ||
parsed.OutputTokens < stored.OutputTokens) {
return true
}
if stored.HasContextTokens &&
(!parsed.HasContextTokens ||
parsed.ContextTokens < stored.ContextTokens) {
return true
}
if len(parsed.ToolCalls) < len(stored.ToolCalls) {
return true
}
return countToolResultEvents(parsed.ToolCalls) <
countToolResultEvents(stored.ToolCalls)
}
func sanitizedMessageContentLength(msg db.Message) int {
sanitized := db.SanitizeUTF8(msg.Content)
if sanitized != msg.Content {
return len(sanitized)
}
return msg.ContentLength
}
func countToolResultEvents(calls []db.ToolCall) int {
total := 0
for _, call := range calls {
total += len(call.ResultEvents)
}
return total
}
func (e *Engine) applyIDPrefixToSessionIDs(ids []string) []string {
return applyIDPrefixToIDs(e.idPrefix, ids)
}
// applyRemoteRewrites prefixes session IDs and rewrites
// file paths for remote sync. No-op when idPrefix is empty.
func (e *Engine) applyRemoteRewrites(
s *db.Session, msgs []db.Message,
) {
if e.idPrefix == "" {
return
}
s.ID = applyIDPrefixToID(e.idPrefix, s.ID)
if s.ParentSessionID != nil && *s.ParentSessionID != "" {
p := applyIDPrefixToID(e.idPrefix, *s.ParentSessionID)
s.ParentSessionID = &p
}
if e.pathRewriter != nil && s.FilePath != nil {
fp := e.pathRewriter(*s.FilePath)
s.FilePath = &fp
}
for i := range msgs {
msgs[i].SessionID = s.ID
for j := range msgs[i].ToolCalls {
msgs[i].ToolCalls[j].SessionID = s.ID
if msgs[i].ToolCalls[j].SubagentSessionID != "" {
msgs[i].ToolCalls[j].SubagentSessionID =
applyIDPrefixToID(
e.idPrefix,
msgs[i].ToolCalls[j].SubagentSessionID,
)
}
for k := range msgs[i].ToolCalls[j].ResultEvents {
re := &msgs[i].ToolCalls[j].ResultEvents[k]
if re.SubagentSessionID != "" {
re.SubagentSessionID =
applyIDPrefixToID(
e.idPrefix,
re.SubagentSessionID,
)
}
}
}
}
}
// toDBSession converts a pendingWrite to a db.Session.
func toDBSession(pw pendingWrite) db.Session {
hasTotal, hasPeak := pw.sess.TokenCoverage(pw.msgs)
s := db.Session{
ID: pw.sess.ID,
Project: pw.sess.Project,
Machine: pw.sess.Machine,
Agent: string(pw.sess.Agent),
MessageCount: pw.sess.MessageCount,
UserMessageCount: pw.sess.UserMessageCount,
ParentSessionID: strPtr(pw.sess.ParentSessionID),
RelationshipType: string(pw.sess.RelationshipType),
TotalOutputTokens: pw.sess.TotalOutputTokens,
PeakContextTokens: pw.sess.PeakContextTokens,
HasTotalOutputTokens: hasTotal,
HasPeakContextTokens: hasPeak,
Cwd: pw.sess.Cwd,
GitBranch: pw.sess.GitBranch,
SourceSessionID: pw.sess.SourceSessionID,
SourceVersion: pw.sess.SourceVersion,
TranscriptFidelity: pw.sess.TranscriptFidelity,
ParserMalformedLines: pw.sess.MalformedLines,
IsTruncated: pw.sess.IsTruncated,
TerminationStatus: strPtr(string(pw.sess.TerminationStatus)),
// data_version is intentionally left at the
// existing column default (0). UpsertSession does
// not persist this field; the caller bumps it via
// SetSessionDataVersion only after the message
// rewrite succeeds.
FilePath: strPtr(pw.sess.File.Path),
FileSize: int64Ptr(pw.sess.File.Size),
FileMtime: int64Ptr(pw.sess.File.Mtime),
NextOrdinal: nextParsedOrdinal(0, pw.msgs),
LastEntryUUID: strPtr(lastParsedSourceUUID("", pw.msgs)),
FileInode: int64Ptr(pw.sess.File.Inode),
FileDevice: int64Ptr(pw.sess.File.Device),
FileHash: strPtr(pw.sess.File.Hash),
}
if pw.sess.FirstMessage != "" {
s.FirstMessage = &pw.sess.FirstMessage
}
s.SessionName = db.ParsedSessionName(pw.sess)
if !pw.sess.StartedAt.IsZero() {
s.StartedAt = timeutil.Ptr(pw.sess.StartedAt)
}
if !pw.sess.EndedAt.IsZero() {
s.EndedAt = timeutil.Ptr(pw.sess.EndedAt)
}
return s
}
// toDBMessages converts parsed messages to db.Message rows
// with tool-result pairing and filtering applied.
func toDBMessages(pw pendingWrite, blocked map[string]bool) []db.Message {
msgs := make([]db.Message, len(pw.msgs))
for i, m := range pw.msgs {
hasCtx, hasOut := m.TokenPresence()
msgs[i] = db.Message{
SessionID: pw.sess.ID,
Ordinal: m.Ordinal,
Role: string(m.Role),
Content: m.Content,
ThinkingText: m.ThinkingText,
Timestamp: timeutil.Format(m.Timestamp),
HasThinking: m.HasThinking,
HasToolUse: m.HasToolUse,
ContentLength: m.ContentLength,
IsSystem: m.IsSystem,
Model: m.Model,
TokenUsage: m.TokenUsage,
ContextTokens: m.ContextTokens,
OutputTokens: m.OutputTokens,
HasContextTokens: hasCtx,
HasOutputTokens: hasOut,
ClaudeMessageID: m.ClaudeMessageID,
ClaudeRequestID: m.ClaudeRequestID,
SourceType: m.SourceType,
SourceSubtype: m.SourceSubtype,
SourceUUID: m.SourceUUID,
SourceParentUUID: m.SourceParentUUID,
IsSidechain: m.IsSidechain,
IsCompactBoundary: m.IsCompactBoundary,
ToolCalls: convertToolCalls(
pw.sess.ID, m.ToolCalls,
),
ToolResults: convertToolResults(m.ToolResults),
}
}
return pairAndFilter(msgs, blocked)
}
// toDBUsageEvents converts parser usage events for one session.
// sessionID is the final ID after remote rewrites; parser-stamped
// event session IDs predate the idPrefix and are ignored. It returns the
// fix counts from the central validation/sanitization pass so write paths
// can surface them in the sync summary; diagnostic callers may discard.
func toDBUsageEvents(
sessionID string, events []parser.ParsedUsageEvent,
) ([]db.UsageEvent, validationStats) {
out := make([]db.UsageEvent, 0, len(events))
for _, ev := range events {
out = append(out, db.UsageEvent{
SessionID: sessionID,
MessageOrdinal: ev.MessageOrdinal,
Source: ev.Source,
Model: ev.Model,
InputTokens: ev.InputTokens,
OutputTokens: ev.OutputTokens,
CacheCreationInputTokens: ev.CacheCreationInputTokens,
CacheReadInputTokens: ev.CacheReadInputTokens,
ReasoningTokens: ev.ReasoningTokens,
CostUSD: ev.CostUSD,
CostStatus: ev.CostStatus,
CostSource: ev.CostSource,
OccurredAt: ev.OccurredAt,
DedupKey: ev.DedupKey,
})
}
// Route usage events through the central validation/sanitization
// pass so they get the same treatment as messages and sessions at
// every call site.
return out, validateAndSanitize(nil, nil, out)
}
// usageEventsForWrite converts usage events for a session about to be
// written and records the central-validation fix counts in the per-run
// anomaly accumulator for the sync summary.
func (e *Engine) usageEventsForWrite(
sessionID string, events []parser.ParsedUsageEvent,
) []db.UsageEvent {
out, vs := toDBUsageEvents(sessionID, events)
e.anomalies.recordSanitize(vs)
return out
}
// postFilterCounts returns the total and user message counts
// from a filtered message slice. System-injected messages
// (e.g. Zencoder compaction, continuation notices) are excluded
// from the user count.
func postFilterCounts(msgs []db.Message) (total, user int) {
for _, m := range msgs {
if m.Role == "user" && !m.IsSystem {
user++
}
}
return len(msgs), user
}
// countUserMsgs counts user messages in parsed messages.
func countUserMsgs(msgs []parser.ParsedMessage) int {
n := 0
for _, m := range msgs {
if m.Role == parser.RoleUser {
n++
}
}
return n
}
func nextParsedOrdinal(
current int, msgs []parser.ParsedMessage,
) int {
if len(msgs) == 0 {
return current
}
return msgs[len(msgs)-1].Ordinal + 1
}
func lastParsedSourceUUID(
current string, msgs []parser.ParsedMessage,
) string {
for _, v := range slices.Backward(msgs) {
if v.SourceUUID != "" {
return v.SourceUUID
}
}
return current
}
// FindSourceFile locates the original source file for a
// session ID. It first checks the stored file_path from the
// database (handles cases where filename differs from session
// ID, e.g. Zencoder header ID vs filename), then falls back
// to agent-specific path reconstruction.
func (e *Engine) FindSourceFile(sessionID string) string {
host, rawID := parser.StripHostPrefix(sessionID)
if host != "" {
if fp := e.db.GetSessionFilePath(sessionID); isS3SourcePath(fp) {
return fp
}
// Remote sessions have no local source file.
return ""
}
def, ok := parser.AgentByPrefix(sessionID)
if !ok {
return ""
}
rawSessionID := strings.TrimPrefix(rawID, def.IDPrefix)
if !def.FileBased {
// Forge, Piebald, Warp, and ZCode are DB-backed providers that own
// discovery and source lookup through the provider facade. Their
// virtual <db>#<sessionID> path is resolved by findProviderSourceFile
// below. Non-provider, non-file-based agents (e.g. remote imports)
// have no local source file.
if !e.isProviderAuthoritative(def.Type) {
return ""
}
storedPath := e.db.GetSessionFilePath(sessionID)
if f := e.findProviderSourceFile(
context.Background(), def, sessionID, rawSessionID, storedPath,
); f != "" {
return f
}
return ""
}
if def.Type == parser.AgentKiro {
for _, dir := range e.agentDirs[parser.AgentKiro] {
dbPath := kiroSQLiteDBPath(dir)
if dbPath == "" ||
!parser.KiroSQLiteSessionExists(
dbPath, rawSessionID,
) {
continue
}
return parser.KiroSQLiteVirtualPath(
dbPath, rawSessionID,
)
}
}
bareID := strings.TrimPrefix(rawID, def.IDPrefix)
storedPath := e.db.GetSessionFilePath(sessionID)
if f := e.findProviderSourceFile(
context.Background(), def, sessionID, bareID, storedPath,
); f != "" {
return f
}
// Prefer stored file_path — it's authoritative and handles
// cases where the session ID doesn't match the filename.
// Resolve virtual paths (e.g. Visual Studio Copilot's
// <traceFile>#<conversationID>) for the existence check, but
// return the stored path so downstream parsing stays scoped to
// the requested conversation rather than the whole trace file.
if fp := storedPath; fp != "" {
// s3:// sources have no local file to stat; the path is itself
// the authoritative source and processFile fetches it directly.
if strings.HasPrefix(fp, "s3://") {
return fp
}
if historyPath, idx, ok := parser.ParseAiderVirtualPath(fp); ok {
// aider's stored "<historyPath>#<idx>" is positional: an
// inserted or removed earlier run shifts the index onto a
// different session. Only trust the stored path when run idx
// still recomputes to the requested raw ID; otherwise fall
// through. The provider facade, tried first above, owns raw-ID
// re-resolution.
if got, ok := parser.AiderRawIDAt(historyPath, idx); ok && got == bareID {
return fp
}
} else if _, err := os.Stat(parser.ResolveSourceFilePath(fp)); err == nil {
return fp
}
}
return ""
}
// isProviderAuthoritative reports whether the agent's runtime sync is owned by
// the provider facade rather than a legacy engine dispatch path.
func (e *Engine) isProviderAuthoritative(agent parser.AgentType) bool {
return e.providerMigrationModes[agent] ==
parser.ProviderMigrationProviderAuthoritative
}
// findProviderSourceFile resolves a single session's source file through the
// provider facade for authoritative concrete providers. It is the sole
// source-lookup path, keeping sessions locatable for diagnostics, export, and
// parse-diff lookups.
func (e *Engine) findProviderSourceFile(
ctx context.Context,
def parser.AgentDef,
sessionID string,
rawSessionID string,
storedPath string,
) string {
mode := e.providerMigrationModes[def.Type]
if mode != parser.ProviderMigrationProviderAuthoritative {
return ""
}
factory, ok := e.providerFactories[def.Type]
if !ok || factory == nil {
return ""
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: e.agentDirs[def.Type],
Machine: e.machine,
PathRewriter: e.pathRewriter,
})
source, found, err := provider.FindSource(ctx, parser.FindSourceRequest{
RawSessionID: rawSessionID,
FullSessionID: sessionID,
StoredFilePath: storedPath,
FingerprintKey: storedPath,
RequireFreshSource: true,
PreferStoredSource: true,
})
if err != nil {
log.Printf("%s provider source lookup: %v", def.Type, err)
return ""
}
if !found {
return ""
}
// A fork session ID (Piebald piebald:<chat>-<row>) resolves to its base
// chat source. Confirm the requested fork is actually produced before
// treating the chat source as a hit, mirroring the legacy parse-verify.
if providerSessionIsFork(def, sessionID, rawSessionID) {
outcome, err := provider.Parse(ctx, parser.ParseRequest{
Source: source,
Machine: e.machine,
})
if err != nil || !providerOutcomeContainsSession(outcome, sessionID) {
return ""
}
}
return providerDiscoveredPath(source)
}
// providerSessionSourceMtime resolves a session's authoritative source-backed
// mtime through the provider facade. It is used for sessions whose stored
// file_path is provider-owned (for example a virtual <db>#<sessionID> path), so
// SourceMtime stays on the same composite fingerprint basis sync uses for DB
// freshness checks. Piebald fork IDs (piebald:<chat>-<row>) resolve to their
// base chat source, so a fork is confirmed by parsing the chat and checking the
// requested session ID is actually produced before returning the chat mtime.
func (e *Engine) providerSessionSourceMtime(
ctx context.Context,
def parser.AgentDef,
sessionID string,
rawSessionID string,
storedPath string,
) int64 {
factory, ok := e.providerFactories[def.Type]
if !ok || factory == nil {
return 0
}
provider := factory.NewProvider(parser.ProviderConfig{
Roots: e.agentDirs[def.Type],
Machine: e.machine,
})
source, found, err := provider.FindSource(ctx, parser.FindSourceRequest{
RawSessionID: rawSessionID,
FullSessionID: sessionID,
StoredFilePath: storedPath,
FingerprintKey: storedPath,
RequireFreshSource: true,
PreferStoredSource: true,
})
if err != nil {
log.Printf("%s provider source mtime lookup: %v", def.Type, err)
return 0
}
if !found {
return 0
}
fingerprint, err := provider.Fingerprint(ctx, source)
if err != nil {
log.Printf("%s provider source mtime fingerprint: %v", def.Type, err)
return 0
}
if fingerprint.MTimeNS == 0 {
return 0
}
// A fork session ID resolves to its base chat source. Confirm the
// requested fork exists before treating the chat mtime as authoritative.
if providerSessionIsFork(def, sessionID, rawSessionID) {
outcome, err := provider.Parse(ctx, parser.ParseRequest{
Source: source,
Machine: e.machine,
})
if err != nil || !providerOutcomeContainsSession(outcome, sessionID) {
return 0
}
}
return fingerprint.MTimeNS
}
func providerSourcePathNeedsFingerprint(path string) bool {
if path == "" {
return false
}
if _, _, ok := parser.SplitWindsurfVirtualPath(path); ok {
return true
}
return parser.ResolveSourceFilePath(path) != path
}
func providerSourceMtimeNeedsFingerprint(agent parser.AgentType) bool {
return agent == parser.AgentQoder
}
// providerSessionIsFork reports whether the session ID addresses a fork child
// whose base differs from the resolved source session. Only Piebald uses the
// "<chat>-<row>" fork-ID shape among the DB-backed providers.
func providerSessionIsFork(
def parser.AgentDef,
sessionID string,
rawSessionID string,
) bool {
if def.Type != parser.AgentPiebald {
return false
}
chatID, _, _ := strings.Cut(rawSessionID, "-")
return chatID != rawSessionID
}
// providerOutcomeContainsSession reports whether a parse outcome produced the
// given full session ID.
func providerOutcomeContainsSession(
outcome parser.ParseOutcome,
sessionID string,
) bool {
for _, result := range outcome.Results {
if result.Result.Session.ID == sessionID {
return true
}
}
return false
}
// SourceMtime returns the current source-backed mtime for a
// session. Most file-based agents map directly to a single source
// file, but OpenCode storage sessions derive their effective mtime
// from the session JSON plus related message/part files.
func (e *Engine) SourceMtime(sessionID string) int64 {
host, rawID := parser.StripHostPrefix(sessionID)
if host != "" {
if fp := e.db.GetSessionFilePath(sessionID); isS3SourcePath(fp) {
stat := statS3Object
if def, ok := parser.AgentByPrefix(sessionID); ok &&
def.Type == parser.AgentClaude {
stat = statClaudeS3Session
} else if ok && def.Type == parser.AgentCodex {
stat = statCodexS3Session
}
if sess, err := e.db.GetSession(
context.Background(), sessionID,
); err == nil && sess != nil {
switch sess.Agent {
case string(parser.AgentClaude):
stat = statClaudeS3Session
case string(parser.AgentCodex):
stat = statCodexS3Session
}
}
obj, err := stat(fp)
if err != nil {
return 0
}
return obj.LastModified.UnixNano()
}
return 0
}
def, ok := parser.AgentByPrefix(sessionID)
if !ok {
return 0
}
rawSessionID := strings.TrimPrefix(rawID, def.IDPrefix)
if !def.FileBased {
// Forge, Piebald, Warp, and ZCode are DB-backed providers: their
// per-session source mtime comes from the provider fingerprint
// (which mirrors the legacy List*SessionMeta last-modified value).
// Non-provider, non-file-based agents have no local source.
if e.isProviderAuthoritative(def.Type) {
return e.providerSessionSourceMtime(
context.Background(), def, sessionID, rawSessionID, "",
)
}
return 0
}
path := e.FindSourceFile(sessionID)
if path == "" {
return 0
}
if e.isProviderAuthoritative(def.Type) &&
(providerSourcePathNeedsFingerprint(path) ||
providerSourceMtimeNeedsFingerprint(def.Type)) {
if mtime := e.providerSessionSourceMtime(
context.Background(), def, sessionID, rawSessionID, path,
); mtime != 0 {
return mtime
}
}
if isS3SourcePath(path) {
stat := statS3Object
switch def.Type {
case parser.AgentClaude:
stat = statClaudeS3Session
case parser.AgentCodex:
stat = statCodexS3Session
}
obj, err := stat(path)
if err != nil {
return 0
}
return obj.LastModified.UnixNano()
}
if isOpenCodeFormatStorageAgent(def.Type) {
mtime, err := openCodeFormatSourceMtime(def.Type, path)
if err != nil {
return 0
}
return mtime
}
if def.Type == parser.AgentKiro {
if _, _, ok := parseKiroSQLiteVirtualPath(path); ok {
mtime, err := parser.KiroSQLiteSourceMtime(path)
if err != nil {
return 0
}
return mtime
}
}
if def.Type == parser.AgentZed {
if _, _, ok := parser.ParseVirtualSourcePathForBase(path, "threads.db"); ok {
mtime, err := parser.ZedSQLiteSourceMtime(path)
if err != nil {
return 0
}
return mtime
}
}
if def.Type == parser.AgentShelley {
if _, _, ok := parser.ParseVirtualSourcePathForBase(path, shelleyDBFile); ok {
mtime, err := parser.ShelleySourceMtime(path)
if err != nil {
return 0
}
return mtime
}
}
if def.Type == parser.AgentAntigravityCLI {
info, err := parser.AntigravityCLIFileInfo(path)
if err != nil {
return 0
}
return info.ModTime().UnixNano()
}
if def.Type == parser.AgentAntigravity {
info, err := parser.AntigravityFileInfo(path)
if err != nil {
return 0
}
return info.ModTime().UnixNano()
}
if def.Type == parser.AgentCowork {
info, err := os.Stat(path)
if err != nil {
return 0
}
return parser.CoworkSessionMtime(path, info.ModTime().UnixNano())
}
if def.Type == parser.AgentCommandCode {
info, err := os.Stat(path)
if err != nil {
return 0
}
return commandCodeEffectiveInfo(path, info).ModTime().UnixNano()
}
if def.Type == parser.AgentVSCopilot {
// A conversation's transcript is rebuilt from every sibling trace
// file, so the watcher fallback must compare a composite mtime
// spanning all of them, not just the representative trace file.
_, mtime := parser.VisualStudioCopilotTraceFingerprint(
parser.ResolveSourceFilePath(path),
)
return mtime
}
if def.Type == parser.AgentVibe {
info, err := os.Stat(path)
if err != nil {
return 0
}
return vibeEffectiveInfo(path, info).ModTime().UnixNano()
}
if def.Type == parser.AgentReasonix {
info, err := os.Stat(path)
if err != nil {
return 0
}
return reasonixEffectiveInfo(path, info).ModTime().UnixNano()
}
// FindSourceFile may return a virtual path (e.g. Visual Studio
// Copilot's <traceFile>#<conversationID>); resolve it to the
// physical source for the stat.
info, err := os.Stat(parser.ResolveSourceFilePath(path))
if err != nil {
return 0
}
return info.ModTime().UnixNano()
}
func applyProviderFingerprintFileInfo(
agent parser.AgentType,
fingerprint parser.SourceFingerprint,
results []parser.ParseResultOutcome,
) {
if agent != parser.AgentDevin {
return
}
for i := range results {
if fingerprint.Size != 0 {
results[i].Result.Session.File.Size = fingerprint.Size
}
if fingerprint.MTimeNS != 0 {
results[i].Result.Session.File.Mtime = fingerprint.MTimeNS
}
if fingerprint.Hash != "" {
results[i].Result.Session.File.Hash = fingerprint.Hash
}
}
}
// SyncSingleSession re-syncs a single session by its ID and
// uses the existing DB project as fallback where applicable.
func (e *Engine) SyncSingleSession(sessionID string) (err error) {
return e.SyncSingleSessionContext(context.Background(), sessionID)
}
// SyncSingleSessionContext re-syncs a single session by its ID using ctx for
// cancellable git-backed project resolution and database reads on this path.
func (e *Engine) SyncSingleSessionContext(
ctx context.Context, sessionID string,
) (err error) {
if e.refuseWriteInForceParse("SyncSingleSession") {
return fmt.Errorf(
"cannot sync session %s on a report-only (parse-diff) engine",
sessionID,
)
}
e.syncMu.Lock()
preserved := false
// Defers run LIFO: unlock runs first (releasing syncMu), then
// emit. Keep emission outside the critical section so a future
// Emitter implementation can't widen the lock's scope.
defer func() {
if err == nil && !preserved {
e.emit("messages")
}
}()
defer e.syncMu.Unlock()
e.resetS3CodexIndexCache()
host, _ := parser.StripHostPrefix(sessionID)
if host != "" && !isS3SourcePath(e.db.GetSessionFilePath(sessionID)) {
return fmt.Errorf(
"cannot sync remote session %s locally", sessionID,
)
}
def, ok := parser.AgentByPrefix(sessionID)
if !ok {
return fmt.Errorf("unknown agent for session %s", sessionID)
}
if !def.FileBased {
// Forge, Piebald, Warp, and ZCode are DB-backed providers: re-sync routes
// through FindSourceFile (resolving the virtual <db>#<sessionID>
// path) plus the provider-aware processFile path below, mirroring
// the file-based agents. Other non-file-based agents use the
// OpenCode-format storage path.
if !e.isProviderAuthoritative(def.Type) {
return fmt.Errorf(
"cannot resync non-file-based session %s for agent %s",
sessionID, def.Type,
)
}
}
path := e.FindSourceFile(sessionID)
if path == "" {
return fmt.Errorf(
"source file not found for %s", sessionID,
)
}
// OpenCode-format agents (OpenCode, Kilo, MiMoCode) are
// provider-authoritative: their SQLite virtual paths and storage
// sessions resync through the generic processFile path below, which
// routes to the provider facade.
agent := def.Type
// Clear skip cache so explicit re-sync always processes
// the file, even if it was cached as non-interactive
// during a bulk SyncAll.
file := parser.DiscoveredFile{
Path: path,
Agent: agent,
ForceParse: true,
}
e.hydrateS3DiscoveredFile(ctx, sessionID, &file)
if e.shouldCacheSkip(file) {
e.clearSkip(path)
}
// Reuse processFile for stat and DB-skip logic.
switch agent {
case parser.AgentClaude:
// Try to preserve existing project from DB first
if sess, _ := e.db.GetSession(ctx, sessionID); sess != nil &&
sess.Project != "" &&
!parser.NeedsProjectReparse(sess.Project) {
file.Project = sess.Project
} else {
file.Project = filepath.Base(filepath.Dir(path))
}
case parser.AgentVSCopilot:
// processVisualStudioCopilot persists file.Project into every
// parsed session, so an empty project here would overwrite the
// existing "visualstudio" value. Prefer the stored project; fall
// back to the canonical default discovery assigns.
if sess, _ := e.db.GetSession(ctx, sessionID); sess != nil &&
sess.Project != "" &&
!parser.NeedsProjectReparse(sess.Project) {
file.Project = sess.Project
} else {
file.Project = "visualstudio"
}
case parser.AgentCursor:
// Support both flat and nested transcript layouts.
for _, cursorDir := range e.agentDirs[parser.AgentCursor] {
rel, ok := isUnder(cursorDir, path)
if !ok {
continue
}
projDir, ok := parser.ParseCursorTranscriptRelPath(rel)
if !ok {
continue
}
file.Project = parser.DecodeCursorProjectDir(projDir)
break
}
if file.Project == "" {
file.Project = "unknown"
}
case parser.AgentIflow:
// path is <iflowDir>/<project>/session-<uuid>.jsonl
// Extract project dir name from parent directory
if sess, _ := e.db.GetSession(ctx, sessionID); sess != nil &&
sess.Project != "" &&
!parser.NeedsProjectReparse(sess.Project) {
file.Project = sess.Project
} else {
file.Project = filepath.Base(filepath.Dir(path))
}
case parser.AgentQwenPaw:
// path is <qwenpawDir>/<workspace>/sessions/<name>.json or
// <qwenpawDir>/<workspace>/sessions/<subdir>/<name>.json
// Workspace name is the first path segment relative to the
// QwenPaw root.
for _, qwenpawDir := range e.agentDirs[parser.AgentQwenPaw] {
rel, ok := isUnder(qwenpawDir, path)
if !ok {
continue
}
parts := strings.Split(rel, string(filepath.Separator))
if len(parts) > 0 {
file.Project = parts[0]
}
break
}
// Fallback when the stored file_path points outside any
// currently configured QWENPAW_DIR (e.g. the root was
// removed, or the session was synced from a custom path).
// "qwenpaw::<stem>" and orphan the requested
// "qwenpaw:<workspace>:<stem>" row. Prefer the DB-stored
// Project as the authoritative record; parse the workspace
// from the sessionID prefix as a final fallback that works
// even when the DB row is missing or stale.
if file.Project == "" {
if sess, _ := e.db.GetSession(ctx, sessionID); sess != nil &&
sess.Project != "" &&
!parser.NeedsProjectReparse(sess.Project) {
file.Project = sess.Project
}
}
if file.Project == "" {
bareID := strings.TrimPrefix(sessionID, def.IDPrefix)
if workspace, _, ok := strings.Cut(bareID, ":"); ok &&
workspace != "" {
file.Project = workspace
}
}
case parser.AgentQoder:
for _, qoderDir := range e.agentDirs[parser.AgentQoder] {
rel, ok := isUnder(qoderDir, path)
if !ok {
continue
}
parts := strings.Split(rel, string(filepath.Separator))
if len(parts) == 2 || len(parts) == 4 && parts[2] == "subagents" {
file.Project = parser.DecodeQoderProjectDir(parts[0])
break
}
}
case parser.AgentReasonix:
if classified, ok := e.classifyReasonixPath(path); ok {
file.Project = classified.Project
} else {
if sess, _ := e.db.GetSession(ctx, sessionID); sess != nil &&
sess.Project != "" &&
!parser.NeedsProjectReparse(sess.Project) {
file.Project = sess.Project
}
}
}
res := e.processFile(ctx, file)
if res.err != nil {
if res.cacheSkip && res.mtime != 0 && !res.noCacheSkip {
e.cacheSkip(res.skipCacheKey(path), res.mtime, res.sourceFingerprint)
}
return res.err
}
if res.skip {
return nil
}
if res.cacheSkip {
e.clearSkip(res.skipCacheKey(path))
}
// Delete parser-excluded sessions before writing the parsed
// results, mirroring collectAndBatch. Vibe promotes a session
// from its directory-name fallback ID to the canonical
// meta.json ID and returns the stale fallback ID here; without
// this delete a single-session resync would leave both rows in
// the DB and double-count messages and usage. Like
// collectAndBatch, exclusions from a source with no session
// inside the cwd allow-list are frozen so archived rows survive.
if excluded := e.applyIDPrefixToSessionIDs(
res.excludedSessionIDs,
); len(excluded) > 0 && e.sourceAllowsParserExclusions(res) {
if _, err := e.db.DeleteParserExcludedSessions(
excluded,
); err != nil {
return fmt.Errorf(
"delete parser-excluded sessions: %w", err,
)
}
}
// Handle incremental updates from processFile (e.g.
// append-only JSONL that was already synced).
if res.incremental != nil {
if err := e.writeIncremental(res.incremental); err != nil {
return err
}
return nil
}
if len(res.results) == 0 {
return nil
}
for _, pr := range res.results {
if err := e.writeSessionFull(
pendingWrite{
sess: pr.Session,
msgs: pr.Messages,
usageEvents: pr.UsageEvents,
needsRetry: res.needsRetryForSession(pr.Session.ID),
},
); err != nil &&
!isIntentionalSessionSkip(err) &&
!errors.Is(err, errSessionPreserved) {
return fmt.Errorf("write session %s: %w",
pr.Session.ID, err)
} else if errors.Is(err, errSessionPreserved) {
preserved = true
}
}
// Link subagent child sessions to their parents.
// Required for Zencoder sessions that reference subagent
// session IDs in tool_calls.subagent_session_id.
if err := e.db.LinkSubagentSessions(); err != nil {
log.Printf("link subagent sessions: %v", err)
}
return nil
}
func (e *Engine) applyWorktreeMappingToSingleSession(
sessionID string,
) error {
ctx := context.Background()
sess, err := e.db.GetSession(ctx, sessionID)
if err != nil || sess == nil || sess.Cwd == "" {
return err
}
machine := sess.Machine
if machine == "" {
machine = e.machine
}
_, err = e.db.ApplyWorktreeProjectMappingToSessionFromSync(
ctx, machine, sess.ID, sess.Cwd, sess.Project,
)
if err != nil {
return fmt.Errorf(
"apply worktree mapping to session %s: %w",
sessionID, err,
)
}
return nil
}
// filterShadowedLegacyKiroFiles drops discovered legacy Kiro JSONL sources
// whose logical session ID already exists in a current-store SQLite database
// under any configured Kiro root. The Kiro provider performs the same
// shadowing during its own Discover, but only across the roots it is
// configured with; a scoped sync (e.g. SyncRootsSince over a single root)
// configures the provider with that scope only, so the engine reapplies the
// cross-root shadow here using every configured Kiro root. This keeps a legacy
// file from being imported when its session lives in the SQLite store of a
// different, out-of-scope root.
func (e *Engine) filterShadowedLegacyKiroFiles(
files []parser.DiscoveredFile,
) []parser.DiscoveredFile {
if !hasLegacyKiroCandidates(files) {
return files
}
currentIDs := make(map[string]struct{})
for _, dir := range e.agentDirs[parser.AgentKiro] {
for id := range parser.KiroSQLiteSessionIDs(dir) {
currentIDs[id] = struct{}{}
}
}
if len(currentIDs) == 0 {
return files
}
out := files[:0]
for _, file := range files {
if file.Agent != parser.AgentKiro ||
filepath.Base(file.Path) == kiroSQLiteDBName {
out = append(out, file)
continue
}
legacyID := parser.KiroSessionIDFromPath(file.Path)
if _, shadowed := currentIDs[legacyID]; shadowed {
continue
}
out = append(out, file)
}
return out
}
func hasLegacyKiroCandidates(files []parser.DiscoveredFile) bool {
for _, file := range files {
if file.Agent == parser.AgentKiro &&
filepath.Base(file.Path) != kiroSQLiteDBName {
return true
}
}
return false
}
// kiroSQLiteDBName is the filename of the current-store Kiro SQLite DB.
const kiroSQLiteDBName = "data.sqlite3"
// kiroSQLiteDBPath returns the current-store Kiro SQLite DB path when the
// configured root contains one, or "" otherwise.
func kiroSQLiteDBPath(dir string) string {
if dir == "" {
return ""
}
path := filepath.Join(dir, kiroSQLiteDBName)
info, err := os.Stat(path)
if err != nil || info.IsDir() {
return ""
}
return path
}
// parseKiroSQLiteVirtualPath splits a virtual Kiro SQLite source path back
// into its database path and raw session ID using the provider-neutral
// virtual-source-path resolver.
func parseKiroSQLiteVirtualPath(path string) (string, string, bool) {
return parser.ParseVirtualSourcePathForBase(path, kiroSQLiteDBName)
}
func strPtr(s string) *string {
if s == "" {
return nil
}
return &s
}
func int64Ptr(n int64) *int64 {
if n == 0 {
return nil
}
return &n
}
// convertToolCalls maps parsed tool calls to db.ToolCall
// structs. MessageID is resolved later during insert.
func convertToolCalls(
sessionID string, parsed []parser.ParsedToolCall,
) []db.ToolCall {
if len(parsed) == 0 {
return nil
}
calls := make([]db.ToolCall, len(parsed))
for i, tc := range parsed {
filePath := tc.FilePath
if filePath == "" {
filePath = parser.ResolveFilePathFromJSON(tc.InputJSON)
}
calls[i] = db.ToolCall{
SessionID: sessionID,
ToolName: tc.ToolName,
Category: tc.Category,
ToolUseID: tc.ToolUseID,
InputJSON: tc.InputJSON,
FilePath: filePath,
CallIndex: i,
SkillName: tc.SkillName,
SubagentSessionID: tc.SubagentSessionID,
ResultEvents: convertToolResultEvents(tc.ResultEvents),
}
}
return calls
}
func convertToolResultEvents(
parsed []parser.ParsedToolResultEvent,
) []db.ToolResultEvent {
if len(parsed) == 0 {
return nil
}
events := make([]db.ToolResultEvent, len(parsed))
for i, ev := range parsed {
events[i] = db.ToolResultEvent{
ToolUseID: ev.ToolUseID,
AgentID: ev.AgentID,
SubagentSessionID: ev.SubagentSessionID,
Source: ev.Source,
Status: ev.Status,
Content: ev.Content,
ContentLength: len(ev.Content),
Timestamp: timeutil.Format(ev.Timestamp),
EventIndex: i,
}
}
return events
}
// convertToolResults maps parsed tool results to db.ToolResult
// structs for use in pairing before DB insert.
func convertToolResults(
parsed []parser.ParsedToolResult,
) []db.ToolResult {
if len(parsed) == 0 {
return nil
}
results := make([]db.ToolResult, len(parsed))
for i, tr := range parsed {
results[i] = db.ToolResult{
ToolUseID: tr.ToolUseID,
ContentLength: tr.ContentLength,
ContentRaw: tr.ContentRaw,
}
}
return results
}
// pairAndFilter pairs tool results with their corresponding
// tool calls, then removes user messages that carried only
// tool_result blocks (no displayable text).
func pairAndFilter(msgs []db.Message, blocked map[string]bool) []db.Message {
pairToolResults(msgs, blocked)
pairToolResultEventSummaries(msgs, blocked)
filtered := msgs[:0]
for _, m := range msgs {
if m.Role == "user" &&
len(m.ToolResults) > 0 &&
strings.TrimSpace(m.Content) == "" {
continue
}
filtered = append(filtered, m)
}
return filtered
}
// pairToolResults matches tool_result content to their
// corresponding tool_calls across message boundaries using
// tool_use_id. Categories in blocked are stored without content.
func pairToolResults(msgs []db.Message, blocked map[string]bool) {
idx := make(map[string]*db.ToolCall)
for i := range msgs {
for j := range msgs[i].ToolCalls {
tc := &msgs[i].ToolCalls[j]
if tc.ToolUseID != "" {
idx[tc.ToolUseID] = tc
}
}
}
if len(idx) == 0 {
return
}
for _, m := range msgs {
for _, tr := range m.ToolResults {
if tc, ok := idx[tr.ToolUseID]; ok {
tc.ResultContentLength = tr.ContentLength
if !blocked[tc.Category] {
tc.ResultContent = parser.DecodeContent(tr.ContentRaw)
}
}
}
}
}
func pairToolResultEventSummaries(
msgs []db.Message, blocked map[string]bool,
) {
for i := range msgs {
for j := range msgs[i].ToolCalls {
tc := &msgs[i].ToolCalls[j]
if len(tc.ResultEvents) == 0 {
continue
}
summary := summarizeToolResultEvents(tc.ResultEvents)
tc.ResultContentLength = len(summary)
if blocked[tc.Category] {
tc.ResultContent = ""
tc.ResultEvents = nil
continue
}
tc.ResultContent = summary
}
}
}
func summarizeToolResultEvents(
events []db.ToolResultEvent,
) string {
if len(events) == 0 {
return ""
}
type agentSummary struct {
order int
content string
}
latestByAgent := map[string]agentSummary{}
orderedAgents := make([]string, 0, len(events))
lastAnon := ""
allHaveAgentID := true
for _, ev := range events {
if strings.TrimSpace(ev.Content) == "" {
continue
}
agentID := strings.TrimSpace(ev.AgentID)
if agentID == "" {
allHaveAgentID = false
lastAnon = ev.Content
continue
}
if _, ok := latestByAgent[agentID]; !ok {
latestByAgent[agentID] = agentSummary{
order: len(orderedAgents),
content: ev.Content,
}
orderedAgents = append(orderedAgents, agentID)
continue
}
entry := latestByAgent[agentID]
entry.content = ev.Content
latestByAgent[agentID] = entry
}
if len(latestByAgent) <= 1 {
if len(latestByAgent) == 1 {
summary := latestByAgent[orderedAgents[0]].content
if lastAnon != "" {
return summary + "\n\n" + lastAnon
}
return summary
}
return lastAnon
}
parts := make([]string, 0, len(orderedAgents))
for _, agentID := range orderedAgents {
parts = append(parts, agentID+":\n"+latestByAgent[agentID].content)
}
if !allHaveAgentID && lastAnon != "" {
parts = append(parts, lastAnon)
}
return strings.Join(parts, "\n\n")
}
// emit fires a refresh event if an emitter is wired. Safe to call
// with a nil emitter.
func (e *Engine) emit(scope string) {
if e.emitter != nil {
e.emitter.Emit(scope)
}
}
const scanProgressInterval = 50
// SecretScanInput parameterises ScanSecrets.
type SecretScanInput struct {
Backfill bool
Project string
Agent string
DateFrom string
DateTo string
}
// SecretScanProgress is one progress tick.
type SecretScanProgress struct {
Scanned int `json:"scanned"`
Total int `json:"total"`
}
// SecretScanSummary is the final result of a scan.
type SecretScanSummary struct {
Scanned int `json:"scanned"`
// WithSecrets counts sessions with ≥1 definite finding. It does NOT
// include sessions whose findings are all candidate-tier; the
// presence of those is implied by CandidateFindings > 0 when
// DefiniteFindings is 0.
WithSecrets int `json:"with_secrets"`
TotalFindings int `json:"total_findings"`
DefiniteFindings int `json:"definite_findings"`
CandidateFindings int `json:"candidate_findings"`
}
// ScanSecrets scans candidate sessions and persists their findings, invoking
// progress periodically. Resumable: each scanned session records the current
// rules version, so an interrupted backfill resumes by skipping sessions
// already at that version.
func (e *Engine) ScanSecrets(
ctx context.Context, in SecretScanInput,
progress func(SecretScanProgress),
) (SecretScanSummary, error) {
if e.refuseWriteInForceParse("ScanSecrets") {
return SecretScanSummary{}, errors.New(
"ScanSecrets refused on report-only parse-diff engine",
)
}
ver := secrets.RulesVersion()
ids, err := e.db.SecretScanCandidates(ctx, db.SecretScanCandidateFilter{
CurrentVersion: ver, OnlyStale: in.Backfill,
Project: in.Project, Agent: in.Agent,
DateFrom: in.DateFrom, DateTo: in.DateTo,
})
if err != nil {
return SecretScanSummary{}, err
}
var sum SecretScanSummary
total := len(ids)
for i, id := range ids {
if ctx.Err() != nil {
return sum, ctx.Err()
}
nf, leak, ok := e.scanOneSession(ctx, id, ver)
// A cancellation during the scan must end the run with an error,
// not a partial success. This covers both a failed scan and a
// successful final session whose context was canceled mid-scan,
// since scanOneSession does CPU work and a non-context-aware
// persist after its context-aware reads.
if ctx.Err() != nil {
return sum, ctx.Err()
}
if !ok {
continue
}
sum.Scanned++
sum.TotalFindings += nf
sum.DefiniteFindings += leak
sum.CandidateFindings += nf - leak
if leak > 0 {
sum.WithSecrets++
}
if progress != nil && scanShouldReport(i, total) {
progress(SecretScanProgress{Scanned: sum.Scanned, Total: total})
}
}
return sum, nil
}
// scanOneSession scans one session and persists its findings at ver. Returns
// the finding count, the definite-leak count, and ok=false when the session
// could not be loaded or persisted (skipped, not fatal to the whole run).
//
// Holds syncMu so the read/compute/write path is atomic against a concurrent
// sync replacing this session's messages: otherwise a sync could write fresh
// findings for new messages and then have this scan overwrite them with
// results from a stale snapshot while marking the session current. The lock is
// taken per session, not for the whole scan, so a long backfill does not stall
// the file watcher and periodic sync.
func (e *Engine) scanOneSession(
ctx context.Context, id, ver string,
) (int, int, bool) {
e.syncMu.Lock()
defer e.syncMu.Unlock()
sess, err := e.db.GetSessionFull(ctx, id)
if err != nil || sess == nil {
return 0, 0, false
}
msgs, err := e.db.GetAllMessages(ctx, id)
if err != nil {
return 0, 0, false
}
findings, leak := scanSecretsFromMessages(*sess, msgs, secrets.Scan)
if err := e.db.ReplaceSessionSecretFindings(id, findings, leak, ver); err != nil {
log.Printf("secrets scan: persist %s: %v", id, err)
return 0, 0, false
}
return len(findings), leak, true
}
func scanShouldReport(i, total int) bool {
return (i+1)%scanProgressInterval == 0 || i+1 == total
}