package sync import gosync "sync" // Phase describes the current sync phase. type Phase string const ( PhaseIdle Phase = "idle" PhaseDiscovering Phase = "discovering" PhasePreparingResync Phase = "preparing_resync" PhaseSyncing Phase = "syncing" PhaseCopyingMetadata Phase = "copying_metadata" PhaseCopyingOrphans Phase = "copying_orphans" PhaseReclassifying Phase = "reclassifying" PhaseRebuildingSearch Phase = "rebuilding_search" PhaseSwappingDatabase Phase = "swapping_database" PhaseDone Phase = "done" ) // Progress reports sync progress to listeners. type Progress struct { Phase Phase `json:"phase"` Detail string `json:"detail,omitempty"` Hint string `json:"hint,omitempty"` Resync bool `json:"resync,omitempty"` CurrentProject string `json:"current_project,omitempty"` ProjectsTotal int `json:"projects_total"` ProjectsDone int `json:"projects_done"` SessionsTotal int `json:"sessions_total"` SessionsDone int `json:"sessions_done"` MessagesIndexed int `json:"messages_indexed"` BytesDone int64 `json:"bytes_done,omitempty"` BytesTotal int64 `json:"bytes_total,omitempty"` } // SyncResult describes the outcome of syncing a single session. type SyncResult struct { SessionID string `json:"session_id"` Project string `json:"project"` Skipped bool `json:"skipped"` Messages int `json:"messages"` } // SyncStats summarizes a full sync run. // // TotalSessions counts discovered files plus DB-backed sessions. // Synced counts sessions (one file can produce multiple via fork // detection; DB-backed agents add sessions directly). Failed counts // files with hard parse/stat errors. filesOK counts files that // produced at least one session — used by ResyncAll to compare // against Failed on the same unit. type SyncStats struct { TotalSessions int `json:"total_sessions"` Synced int `json:"synced"` Skipped int `json:"skipped"` Failed int `json:"failed"` OrphanedCopied int `json:"orphaned_copied,omitempty"` Warnings []string `json:"warnings,omitempty"` Aborted bool `json:"aborted,omitempty"` RebuildPhases []RebuildPhaseStats `json:"rebuild_phases,omitempty"` // Anomalies aggregates per-run parser/sanitizer anomaly signals // surfaced in the CLI sync summary. These are live per-run counters // (reset each sync), not persisted state. A zero value means a clean // run and is omitted from the summary. Anomalies AnomalyStats `json:"anomalies,omitzero"` filesOK int // unexported: file-level success counter filesDiscovered int // file-based total, excludes DB-backed agents // nonContainerDiscovered counts discovered files that are not part of // a self-preserving container store (OpenCode-format storage and its // SQLite virtual paths). The resync empty-discovery guard uses it so a // container store's discovery does not mask the disappearance of plain // file-backed sessions whose directories went empty. nonContainerDiscovered int messagesIndexed int // unexported: progress message counter parserExcludedFiles int // file-level intentional parser exclusions parserExcludedIDs []string // cwdFilteredSessions counts sessions vetoed by the // sync_include_cwd_prefixes allow-list. The resync abort guard uses // it so a run where every discovered session is filtered reads as // an intentional result rather than an unsafe empty rebuild. cwdFilteredSessions int // cwdFilteredFiles counts files whose every parsed session was // vetoed by the allow-list. Together with parserExcludedFiles it // must account for all of filesOK before the resync abort guard // treats a zero-write run as intentional. cwdFilteredFiles int } // AnomalyStats aggregates parser-output anomaly signals observed during a // single sync run. It surfaces numbers that already exist or are already // computed but were previously discarded: per-agent parser malformed-line // counts (parser_malformed_lines) and the per-category counts returned by // the central validateAndSanitize pass. It is a per-run summary only; no // new persisted columns back it. type AnomalyStats struct { // MalformedLinesByAgent maps an agent type to the total number of // parser malformed lines reported by sessions of that agent in this // run. Only non-zero agents are present. MalformedLinesByAgent map[string]int `json:"malformed_lines_by_agent,omitempty"` // MalformedLinesTotal is the grand total across all agents. MalformedLinesTotal int `json:"malformed_lines_total,omitempty"` // UnknownSchemaSessionsByAgent maps an agent type to the number of // sessions written this run that were decoded from an unrecognized // (newer) Antigravity schema -- source_version carries an "agy-schema:" // marker, so the heuristic decode may be incomplete or wrong. Unlike the // malformed-line counter this counts whole SESSIONS, not lines. Only // non-zero agents are present. UnknownSchemaSessionsByAgent map[string]int `json:"unknown_schema_sessions_by_agent,omitempty"` // UnknownSchemaSessionsTotal is the grand total across all agents. UnknownSchemaSessionsTotal int `json:"unknown_schema_sessions_total,omitempty"` // GenMetadataWithoutUsageByAgent maps an agent type to the number of // Antigravity sessions written this run that carried gen_metadata rows // but decoded into zero usage events -- an early warning that a newer agy // build changed the gen_metadata wire format the token-block heuristic // depends on. Like the unknown-schema counter this counts whole SESSIONS. // Only non-zero agents are present. GenMetadataWithoutUsageByAgent map[string]int `json:"gen_metadata_without_usage_by_agent,omitempty"` // GenMetadataWithoutUsageTotal is the grand total across all agents. GenMetadataWithoutUsageTotal int `json:"gen_metadata_without_usage_total,omitempty"` // Sanitize aggregates the central validation/sanitization fix counts // across every session, message, and usage event written this run. Sanitize SanitizeStats `json:"sanitize,omitzero"` } // SanitizeStats mirrors the per-category fix counts produced by the central // validateAndSanitize pass, accumulated across a full sync run. It is the // exported, summary-facing form of the internal validationStats so the CLI // summary can render it. type SanitizeStats struct { ControlCharsStripped int `json:"control_chars_stripped,omitempty"` ModelClamped int `json:"model_clamped,omitempty"` TokensClamped int `json:"tokens_clamped,omitempty"` RoleCoerced int `json:"role_coerced,omitempty"` TimestampsBlanked int `json:"timestamps_blanked,omitempty"` } // Total returns the sum of all sanitize fix counts. func (s SanitizeStats) Total() int { return s.ControlCharsStripped + s.ModelClamped + s.TokensClamped + s.RoleCoerced + s.TimestampsBlanked } // IsZero reports whether no sanitize fixes were recorded. func (s SanitizeStats) IsZero() bool { return s.Total() == 0 } // IsZero reports whether the run observed no anomalies at all, so the CLI // summary can omit the anomaly section entirely on clean runs. func (a AnomalyStats) IsZero() bool { return a.MalformedLinesTotal == 0 && a.UnknownSchemaSessionsTotal == 0 && a.GenMetadataWithoutUsageTotal == 0 && a.Sanitize.IsZero() } // RecordMalformedLines attributes n parser malformed lines to the given // agent and updates the grand total. A zero count is ignored so clean // agents do not appear in the per-agent breakdown. func (a *AnomalyStats) RecordMalformedLines(agent string, n int) { if n <= 0 { return } if a.MalformedLinesByAgent == nil { a.MalformedLinesByAgent = make(map[string]int) } a.MalformedLinesByAgent[agent] += n a.MalformedLinesTotal += n } // RecordUnknownSchemaSessions attributes n unrecognized-schema sessions to the // given agent and updates the grand total. A non-positive count is ignored so // clean agents do not appear in the per-agent breakdown. func (a *AnomalyStats) RecordUnknownSchemaSessions(agent string, n int) { if n <= 0 { return } if a.UnknownSchemaSessionsByAgent == nil { a.UnknownSchemaSessionsByAgent = make(map[string]int) } a.UnknownSchemaSessionsByAgent[agent] += n a.UnknownSchemaSessionsTotal += n } // RecordGenMetadataWithoutUsage attributes n gen_metadata-without-usage // sessions to the given agent and updates the grand total. A non-positive // count is ignored so clean agents do not appear in the per-agent breakdown. func (a *AnomalyStats) RecordGenMetadataWithoutUsage(agent string, n int) { if n <= 0 { return } if a.GenMetadataWithoutUsageByAgent == nil { a.GenMetadataWithoutUsageByAgent = make(map[string]int) } a.GenMetadataWithoutUsageByAgent[agent] += n a.GenMetadataWithoutUsageTotal += n } // addSanitize accumulates the per-category counts from one // validateAndSanitize pass into the aggregate. func (a *AnomalyStats) addSanitize(v validationStats) { a.Sanitize.ControlCharsStripped += v.ControlCharsStripped a.Sanitize.ModelClamped += v.ModelClamped a.Sanitize.TokensClamped += v.TokensClamped a.Sanitize.RoleCoerced += v.RoleCoerced a.Sanitize.TimestampsBlanked += v.TimestampsBlanked } // merge folds another AnomalyStats into the receiver. func (a *AnomalyStats) merge(o AnomalyStats) { for agent, n := range o.MalformedLinesByAgent { a.RecordMalformedLines(agent, n) } for agent, n := range o.UnknownSchemaSessionsByAgent { a.RecordUnknownSchemaSessions(agent, n) } for agent, n := range o.GenMetadataWithoutUsageByAgent { a.RecordGenMetadataWithoutUsage(agent, n) } a.Sanitize.ControlCharsStripped += o.Sanitize.ControlCharsStripped a.Sanitize.ModelClamped += o.Sanitize.ModelClamped a.Sanitize.TokensClamped += o.Sanitize.TokensClamped a.Sanitize.RoleCoerced += o.Sanitize.RoleCoerced a.Sanitize.TimestampsBlanked += o.Sanitize.TimestampsBlanked } // anomalyAccumulator is the engine's per-run, concurrency-safe sink for // anomaly signals recorded at the write seam (prepareSessionWrite, // writeIncremental, and the usage-event conversion path). It mirrors the // phaseStats accumulator pattern: reset at the start of a sync run and // folded into the returned SyncStats before the run completes. Although the // write path is serialized under syncMu today, the mutex keeps recording // safe if a future caller writes from multiple goroutines. type anomalyAccumulator struct { mu gosync.Mutex stats AnomalyStats // malformedFiles tracks source paths whose malformed-line count has // already been recorded this run, so a file that forks into several // sessions counts its malformed lines once. Reset each run. malformedFiles map[string]bool } // reset clears the accumulator at the start of a sync run. func (a *anomalyAccumulator) reset() { a.mu.Lock() a.stats = AnomalyStats{} a.malformedFiles = nil a.mu.Unlock() } // recordMalformedLines accumulates an agent's parser malformed-line count for // one source file. A single source file can fork into several sessions (e.g. // Claude subagents) that each carry the same source-level count, so the count // is recorded once per non-empty source path to avoid multiplying a single // malformed line across the fork. Sessions with no source path (DB-backed // agents) are recorded as-is. func (a *anomalyAccumulator) recordMalformedLines( agent, sourcePath string, n int, ) { if n <= 0 { return } a.mu.Lock() defer a.mu.Unlock() if sourcePath != "" { if a.malformedFiles[sourcePath] { return } if a.malformedFiles == nil { a.malformedFiles = make(map[string]bool) } a.malformedFiles[sourcePath] = true } a.stats.RecordMalformedLines(agent, n) } // recordUnknownSchemaSession attributes one unrecognized-schema session to the // given agent. Unlike malformed lines this counts whole sessions, so it is not // deduplicated per source path: a source that forks into several sessions // records each fork. func (a *anomalyAccumulator) recordUnknownSchemaSession(agent string) { a.mu.Lock() a.stats.RecordUnknownSchemaSessions(agent, 1) a.mu.Unlock() } // recordGenMetadataWithoutUsageSession attributes one // gen_metadata-without-usage session to the given agent. Like the // unknown-schema counter it counts whole sessions, so a source that forks // into several sessions records each fork. func (a *anomalyAccumulator) recordGenMetadataWithoutUsageSession(agent string) { a.mu.Lock() a.stats.RecordGenMetadataWithoutUsage(agent, 1) a.mu.Unlock() } // recordSanitize accumulates one validateAndSanitize pass's fix counts. func (a *anomalyAccumulator) recordSanitize(v validationStats) { if (v == validationStats{}) { return } a.mu.Lock() a.stats.addSanitize(v) a.mu.Unlock() } // applyTo folds the accumulated anomalies into the given SyncStats. func (a *anomalyAccumulator) applyTo(s *SyncStats) { a.mu.Lock() s.Anomalies.merge(a.stats) a.mu.Unlock() } // RecordSkip increments the skipped session counter. func (s *SyncStats) RecordSkip() { s.Skipped++ } // RecordSynced adds n to the synced session counter. func (s *SyncStats) RecordSynced(n int) { s.Synced += n } // RecordFailed increments the hard-failure counter. func (s *SyncStats) RecordFailed() { s.Failed++ } // Percent returns the sync progress as a percentage (0–100). func (p Progress) Percent() float64 { if p.SessionsTotal == 0 { return 0 } return float64(p.SessionsDone) / float64(p.SessionsTotal) * 100 } // ProgressFunc is called with progress updates during sync. type ProgressFunc func(Progress)