package main import ( "context" "fmt" "os" "path/filepath" "runtime" "sort" "strings" "sync" "sync/atomic" "time" "go.uber.org/zap" "github.com/zzet/gortex/internal/config" "github.com/zzet/gortex/internal/contracts" "github.com/zzet/gortex/internal/daemon" "github.com/zzet/gortex/internal/graph" "github.com/zzet/gortex/internal/indexer" gortexmcp "github.com/zzet/gortex/internal/mcp" "github.com/zzet/gortex/internal/pathkey" "github.com/zzet/gortex/internal/platform" "github.com/zzet/gortex/internal/progress" "github.com/zzet/gortex/internal/semantic/lsp" "github.com/zzet/gortex/internal/serverstack" ) // daemonState is the bundle of long-lived objects the daemon owns. One // instance per running daemon; every session the daemon accepts shares // these pointers. type daemonState struct { graph graph.Store indexer *indexer.Indexer multiIndexer *indexer.MultiIndexer configManager *config.ConfigManager mcpServer *gortexmcp.Server // proxyHydrator lazily fills cross-daemon proxy-edge nodes from the // owning remote's /v1/subgraph. nil unless federation.edges is on; // the read path hydrates a proxy target before traversing it. proxyHydrator *daemon.ProxyHydrator // snapshotRepos carries per-repo FileMtimes restored from a daemon // snapshot. Populated by buildDaemonState; consumed by // warmupDaemonState to route each configured repo through // ReconcileRepoCtx (incremental) instead of TrackRepoCtx (full // index). nil or missing entries → fall back to full index. snapshotRepos map[string]*snapshotRepo // snapshotContracts carries the per-repo contract entries restored // from the snapshot. Warmup injects these into each indexer after // ReconcileRepoCtx when IncrementalReindex skipped re-extraction (no // stale files). Without this the per-repo contracts.Registry stays // nil for every quiescent repo, so `contracts` / `contracts check` // return empty results even though the graph holds the nodes. snapshotContracts map[string][]contracts.Contract // snapshotPartial reports that the load shed stale records (dropped // nodes / dropped edges whose target vanished). When true, warmup // forces a full per-repo ResolveAll across every indexer instead of // the incremental "only files whose mtime changed" path. Without // this, edges that the loader dropped never come back — every // restart erodes the graph further until exported methods like // (*Node).Type show zero callers despite having dozens of real // callers in source. The IncrementalReindex path never re-resolves // unchanged files, so the lost edges are invisible to it. snapshotPartial bool // snapshotVector carries the workspace-global semantic-search // vector index restored from the snapshot. When its Index is // non-empty and an embedder is configured, warmupDaemonState // restores it after the per-repo re-index loop (which it runs with // vector building skipped) instead of re-embedding the whole graph. snapshotVector snapshotVector // MultiWatcher is built by warmupDaemonState (after tracked repos // have been re-indexed) and handed to realController via // AttachWatcher — it isn't held on daemonState because no caller // reads it from here. // resolverLSPRegistry composes per-repo ResolverHelpers consulted // by the cross-file resolver's hot path. Populated as repos // are tracked so each language-server instance is scoped to its owning // workspace. nil when the resolve-time LSP path is disabled // (GORTEX_LSP_RESOLVER=0) or when semantic enrichment is off. resolverLSPRegistry *lsp.ResolverHelperRegistry // lspRouter is the daemon-shared LSP server pool. Held here so // the warmup loop can register per-repo helpers via // ResolverHelperRegistry without re-deriving the router from the // semantic manager. lspRouter *lsp.Router // overlays is the editor-overlay manager, retained so the HTTP // handler can share the same instance the MCP server uses. overlays *daemon.OverlayManager // shared is the constructed server stack; its Close() runs the // teardown chain (savings flush, backend close) at daemon shutdown. shared *serverstack.SharedServer } // lspDisabledSet builds the set of LSP spec names that should NOT be // auto-registered by Router.RegisterAvailable. Two inputs are merged: // // 1. Per-spec config overrides — any entry in `semantic.providers` // with `enabled: false` whose name matches a known LSP spec. // Already-disabled-by-config users keep their opt-out without // having to also set the env var. // 2. The GORTEX_LSP_DISABLE env var — comma-separated spec names. // The literal value "all" or "*" disables auto-registration // entirely (the explicit-config loop above still runs). // // The special key "__all__" in the returned map signals // "skip auto-register everywhere" and is checked separately by // callers; per-spec keys carry the spec.Name. func lspDisabledSet(providers []config.SemanticProviderConfig, envVar string) map[string]bool { return serverstack.LspDisabledSet(providers, envVar) } // buildDaemonState builds the daemon's stack through the shared // serverstack constructor, applies the daemon-specific snapshot // warm-start (memory backend only), and returns the long-lived // daemonState the warmup loop and controller share. func buildDaemonState(logger *zap.Logger) (*daemonState, error) { cfg, err := config.Load(cfgFile) if err != nil { return nil, fmt.Errorf("load config: %w", err) } gc, _ := config.LoadGlobal() // Fold --tools / --tools-mode into mcp.tools (flag overrides config; // GORTEX_TOOLS still overrides). Survives --detach because the // re-exec'd child re-parses the same flags. applyToolPresetFlags(cfg, daemonTools, daemonToolsMode) ss, err := serverstack.NewSharedServer(serverstack.SharedServerConfig{ Lifecycle: serverstack.LifecycleDaemon, Backend: daemonBackend, BackendPath: daemonBackendPath, BufferPoolMB: resolveDaemonBufferPoolMB(), Config: cfg, Global: gc, Logger: logger, Version: version, Embedder: serverstack.EmbedderRequest{ FlagChanged: daemonEmbeddingsChanged, FlagEnabled: daemonEmbeddings, FlagURL: daemonEmbeddingsURL, FlagModel: daemonEmbeddingsModel, }, // Workspace-global side-store layout: notes/memories partition // under the "daemon" key in the shared DataDir sidecar; the // notebook lives in a cache dir (no single repo git tree to // anchor to); feedback/combo/frecency stay ephemeral. SideStores: serverstack.SideStores{ NotesDir: platform.DataDir(), NotesRepo: "daemon", NotebookPath: filepath.Join(platform.DataDir(), "notebook-cache"), }, }) if err != nil { return nil, fmt.Errorf("build server stack: %w", err) } // Snapshot warm-start (memory backend only — the sqlite backend reads // from its own on-disk store and needs no gob replay). Replays // nodes/edges into the graph and carries the per-repo FileMtimes / // contracts / vector index warmup needs. When the snapshot already // holds a dimension-matching vector index, skip re-embedding the whole // graph during warmup; warmupDaemonState restores the cached index. var loadResult snapshotLoadResult if mg, ok := ss.Graph.(*graph.Graph); ok { loadResult, err = loadSnapshot(mg, logger) if err != nil { logger.Warn("daemon: snapshot load failed", zap.Error(err)) } if ss.MultiIndexer != nil { if vec := loadResult.Vector; len(vec.Index) > 0 && vec.Dims == ss.EmbedderDims { ss.MultiIndexer.SetSkipVectorBuild(true) logger.Info("daemon: snapshot carries vector index — warmup will restore it instead of re-embedding", zap.Int("vectors", vec.Count), zap.Int("dims", vec.Dims)) } } } return &daemonState{ graph: ss.Graph, indexer: ss.Indexer, multiIndexer: ss.MultiIndexer, configManager: ss.ConfigMgr, mcpServer: ss.MCP, overlays: ss.Overlays, shared: ss, snapshotRepos: loadResult.Repos, snapshotContracts: loadResult.Contracts, snapshotPartial: loadResult.Partial, snapshotVector: loadResult.Vector, resolverLSPRegistry: ss.ResolverLSPRegistry, lspRouter: ss.LSPRouter, }, nil } // warmupTimings collects the per-phase costs of one warmupDaemonState run so // the caller can emit a single summary line instead of reconstructing what a // restart did from five differently-shaped per-phase log lines. Populated by // capturing the same phaseStart/time.Since pairs the existing per-phase logs // already compute — those logs stay untouched; this just also keeps the // values around instead of discarding them. type warmupTimings struct { parse time.Duration resolve time.Duration globalResolve time.Duration endBatch time.Duration // reposChanged is the number of tracked repos whose reindex actually did // work this warmup (cold track, or a reconcile with stale/deleted files). reposChanged int // filesReindexed sums, across every changed repo, either FileCount (a // full retrack) or StaleFileCount+DeletedFileCount (an incremental // reconcile). filesReindexed int // enrichScheduled is the number of repos whose deferred semantic // enrichment was actually dispatched, as returned by RunDeferredPassesAll. enrichScheduled int } // warmupDaemonState performs the per-repo parse loop, resolves references, // runs the background enrichment, and brings up the MultiWatcher. Split out // from buildDaemonState so the daemon can open its socket and accept // connections before this work finishes. markReady is invoked once references // are resolved and the graph is queryable — ahead of the slow enrichment pass // — so the daemon reports ready as soon as find_usages / get_callers return // complete results, not after enrichment finishes. The returned *warmupTimings // is never nil — daemon.go uses it to emit a one-line warmup summary. // healDuplicateRepos drops tracked-repo entries that name the same // directory as an earlier entry under a different path spelling — letter // case or Unicode normalisation on a case-insensitive filesystem — logs // each drop, and persists the cleaned config. Returns the number of // entries removed. It is the startup / load-time repair for configs that // accumulated a duplicate before folding was applied (#270): a duplicate // reaching the warmup loop would flip the daemon into multi-repo mode and // desync the graph. The FIRST (oldest) spelling of each directory is kept. func healDuplicateRepos(gc *config.GlobalConfig, logger *zap.Logger) int { if gc == nil { return 0 } removed := gc.DedupeRepos() if len(removed) == 0 { return 0 } if logger == nil { logger = zap.NewNop() } for _, r := range removed { kept := r.Path for _, k := range gc.Repos { if pathkey.SamePathIdentity(k.Path, r.Path) { kept = k.Path break } } logger.Warn("dropping duplicate tracked-repo entry (same directory, different path spelling)", zap.String("dropped", r.Path), zap.String("kept", kept)) } if err := gc.Save(); err != nil { logger.Warn("persisting deduped repo config failed", zap.Error(err)) } return len(removed) } func warmupDaemonState(state *daemonState, logger *zap.Logger, markReady func()) (*indexer.MultiWatcher, *warmupTimings) { timings := &warmupTimings{} if state.multiIndexer == nil || state.configManager == nil { return nil, timings } ctx := progress.WithReporter(context.Background(), progress.Nop{}) // BeginParallelBatch / EndBatch tells every per-repo Indexer // constructed inside the loop to skip both the graph-wide // derivation passes (InferImplements / InferOverrides / // markTestSymbolsAndEmitEdges) AND the per-repo cross-cutting // passes (ResolveAll / semantic enrich / contract extract+commit). // The latter mutate the shared graph in ways that race when // goroutines run them concurrently across repos, so the parallel // loop below just parses; RunDeferredPassesAll drains the deferred // per-repo passes serially before the global resolve. Without this // batch wrapper, a 100+ repo warmup is O(R · global_size). state.multiIndexer.BeginParallelBatch() // Heal any duplicate tracked-repo entries that name the same directory // under a different path spelling (case, or Unicode normalisation) // BEFORE the warmup loop registers them. Two spellings of one directory // would otherwise both reach the indexer and flip the daemon into // multi-repo mode, desyncing the unprefixed graph (#270). healDuplicateRepos(state.configManager.Global(), logger) repos := state.configManager.Global().Repos // Purge orphaned repo prefixes BEFORE the warmup loop re-registers the // tracked repos, while the store still reflects the prior run's state. A // repo removed from config (or untracked before the sidecar-aware purge // existed) can leave its nodes+edges AND its fifteen repo_prefix-keyed // sidecar tables (file_mtimes, *_enrichment, symbol_fts, content_fts, ...) // behind — a long-lived store then carries thousands of rows for a repo // gone long ago. Key the tracked repos through the same EffectiveRepoPrefix // the warm mtime lookup + LSP-helper registry use (so the comparison // matches what the indexer actually wrote), ask the store for any DISTINCT // prefix outside that set, and PurgeRepo each. '' is never an orphan // (shared global externals / solo data). Capability-probed, so only the // sidecar-bearing on-disk store participates; the in-memory store skips it. if orphaner, ok := state.graph.(interface { OrphanRepoPrefixes(known []string) []string }); ok { if purger, ok := state.graph.(interface{ PurgeRepo(string) error }); ok { known := make([]string, 0, len(repos)) for _, entry := range repos { p := strings.TrimPrefix(indexer.EffectiveRepoPrefix(state.configManager, entry), "/") if p != "" { known = append(known, p) } } for _, prefix := range orphaner.OrphanRepoPrefixes(known) { if err := purger.PurgeRepo(prefix); err != nil { logger.Warn("daemon: purging orphaned repo prefix failed", zap.String("prefix", prefix), zap.Error(err)) continue } logger.Info("daemon: purged orphaned repo prefix (no tracked repo in config keys under it)", zap.String("prefix", prefix)) } } } // One-time store compaction, guarded (see daemon_compact.go). Placed HERE // on purpose: after the orphan purge, so the pages it just freed are // measured and reclaimed in the same pass; before the warmup re-index loop // and the resolver/enrichment passes, whose writes would reuse freelist // pages mid-measurement and whose readers would contend with VACUUM's // exclusive lock. The socket is already listening (it opens before this // goroutine by design), but at this point in boot the daemon's own // workload is quiet and a rare early client query either finishes first or // makes the VACUUM fail cleanly at busy_timeout — a skip, not a fault. // Running before the socket opened instead would hold the daemon // unreachable for the whole VACUUM, turning a compacting boot into an // apparent hang. maybeCompactStore(state.graph, logger) // Register a per-repo resolver-time LSP helper for every // tracked repo BEFORE the parallel warmup loop fires. The // helpers are lazy: language servers are not spawned until the // resolver asks for a supported edge resolution, so there's no // startup cost for repos with no matching code. if state.resolverLSPRegistry != nil && state.lspRouter != nil { poolSize := lsp.ResolverPoolSizeFromEnv(1) registered, skipped, tsRepos, pythonRepos := 0, 0, 0, 0 for _, entry := range repos { absRoot, err := filepath.Abs(entry.Path) if err != nil { continue } helper, specs := serverstack.BuildResolverLSPHelperForRepo(state.lspRouter, absRoot, poolSize, logger) if helper == nil { skipped++ continue } prefix := strings.TrimPrefix(indexer.EffectiveRepoPrefix(state.configManager, entry), "/") state.resolverLSPRegistry.Register(prefix, helper) registered++ for _, spec := range specs { switch spec { case "typescript-language-server": tsRepos++ case "pyright": pythonRepos++ } } } logger.Info("daemon: resolve-time LSP helpers registered", zap.Int("repos", registered), zap.Int("skipped", skipped), zap.Int("ts_repos", tsRepos), zap.Int("python_repos", pythonRepos), zap.Int("pool_size", poolSize)) } // Bounded worker pool — disk I/O dominates parsing for most repos, // but a few CPU-heavy ones overlap with disk waits on others. NumCPU // gives good throughput on local SSDs without thrashing slow // external mounts (which dominate at this scale). Capped so a 32-core // box doesn't over-subscribe a single spinning drive. workers := runtime.NumCPU() if workers < 2 { workers = 2 } if workers > 12 { workers = 12 } if workers > len(repos) { workers = len(repos) } logger.Info("daemon: warmup phase start", zap.String("phase", "parallel_parse"), zap.Int("repos", len(repos)), zap.Int("workers", workers), zap.Bool("snapshot_partial_forces_full_walk", state.snapshotPartial)) publishReadinessPhase(state, "parallel_parse", false, map[string]any{ "tracked_repos": len(repos), "workers": workers, }) phaseStart := time.Now() jobs := make(chan config.RepoEntry, len(repos)) var wg sync.WaitGroup // changedRepos counts repos that actually did indexing work this // warmup: a cold full-track, or a reconcile that re-indexed / evicted // at least one file. When it stays zero, NOTHING on disk changed // since the last shutdown, so the persisted graph already holds every // resolved and derived edge — the global resolution passes below // (RunDeferredPassesAll / RunGlobalResolve / RunGlobalGraphPasses) are // pure recomputation and get skipped, which is what makes a true warm // restart near-instant instead of replaying the full cold-warmup cost. var changedRepos atomic.Int64 // filesReindexed sums the actual file-level work behind changedRepos: a // full retrack counts its whole FileCount, an incremental reconcile // counts only the files it touched (stale + deleted). Read once into // timings.filesReindexed after wg.Wait() below. var filesReindexed atomic.Int64 // changedPrefixes records the repo prefix of every repo that did indexing // work, so the end-of-warmup RunGlobalGraphPasses can scope the per-repo // clone detection + Rebuild to just those repos instead of every tracked // repo. scopeUnknown trips when a changed repo's prefix can't be // determined (e.g. a failed reconcile) — then the scope is dropped and the // whole-workspace clone pass runs, degrading toward correctness. var changedPrefixes sync.Map var scopeUnknown atomic.Bool for i := 0; i < workers; i++ { wg.Add(1) go func() { defer wg.Done() for entry := range jobs { // Per-entry panic guard so one repo's crash during // reindex doesn't kill the worker — the bad repo logs // and skips, the worker proceeds to the next job, and // warmup completes. func(entry config.RepoEntry) { defer func() { if r := recover(); r != nil { logger.Error("daemon: warmup repo panic recovered", zap.String("path", entry.Path), zap.Any("panic", r)) } }() // Route repos whose nodes came from the snapshot through // ReconcileRepoCtx — it calls IncrementalReindex, which // evicts files deleted while the daemon was down and // re-indexes only files whose mtime changed. Repos not in // the snapshot (newly tracked, or first startup after a // schema bump) fall back to TrackRepoCtx, which does a // full walk. Both paths end with the repo registered on // the MultiIndexer; contract reconciliation is deferred // to the single RunGlobalResolve call below. // // snapshotPartial == true forces the full-walk path even // when prior mtimes exist: the partial-load signal means // the persisted resolution state is no longer trustworthy // (stale edges were dropped because their targets vanished), // and the incremental path only re-resolves files whose // mtime changed — so the dropped edges would never come // back. Without this override every restart progressively // erodes the graph until exported methods show zero // callers despite having dozens of real call sites. repoStart := time.Now() // Prefer mtimes stored in the backend's FileMtime // sidecar table — that lifts the persistence off the // gob snapshot for disk-backed backends, which is the // path that actually rebuilds across restarts. Falls // back to the snapshot's per-repo FileMtimes when the // backend doesn't implement the reader (memory) or // hasn't seen this repo yet. priorMtimes := priorMtimesFromStore(state.graph, state.configManager, entry, logger) if len(priorMtimes) == 0 { priorMtimes = priorMtimesForEntry(state.snapshotRepos, entry) } if state.snapshotPartial { priorMtimes = nil } // A backend that crossed a schema-rebuild migration rung // (NeedsRebuild) has on-disk rows in the old shape that an // incremental reconcile cannot fix. Drop prior mtimes so every // file re-indexes into the new schema (the nil branch below // runs a full TrackRepoCtx and marks the repo changed, so the // global resolve/derivation passes re-run too). No-op for // backends without the capability and whenever no rebuild rung // was crossed — the common case. if storeNeedsRebuild(state.graph) { if len(priorMtimes) > 0 { logger.Info("daemon: backend signalled schema rebuild; forcing full re-index", zap.String("path", entry.Path)) } priorMtimes = nil } pathFn := "track" if priorMtimes != nil { pathFn = "reconcile" res, err := state.multiIndexer.ReconcileRepoCtx(ctx, entry, priorMtimes) switch { case err != nil: logger.Warn("daemon: startup reconcile failed", zap.String("path", entry.Path), zap.Error(err)) // Treat a failed reconcile as "changed" so the global // passes still run — degrade toward correctness, not // toward the fast path, when we can't trust the delta. changedRepos.Add(1) scopeUnknown.Store(true) case res != nil && (res.StaleFileCount > 0 || res.DeletedFileCount > 0 || len(res.FailedFiles) > 0 || res.FullRetrack): changedRepos.Add(1) filesReindexed.Add(int64(reconcileFileCount(res))) if res.RepoPrefix != "" { changedPrefixes.Store(res.RepoPrefix, struct{}{}) } else { scopeUnknown.Store(true) } default: // Warm no-op path: the repo re-indexed nothing, so its // graph is served straight from the persisted store. // Vet the freshly-recomputed per-repo counts against // what the snapshot recorded — a material shortfall // means the store came back shape-degraded relative to // the snapshot metadata (a persisted resolution // regression). Mark the repo changed so the // end-of-warmup global re-resolve + derivation passes // run for it instead of silently serving the shrunken // graph, and surface the event so a ratchet can't hide // behind an all-green index_health. if res != nil && bootShapeShortfall(state.snapshotRepos, res.RepoPrefix, res.NodeCount, res.EdgeCount) { indexer.RecordResolutionRegression() logger.Warn("daemon: boot shape-degradation guard — repo graph materially short of snapshot; re-running resolution", zap.String("prefix", res.RepoPrefix), zap.Int("live_nodes", res.NodeCount), zap.Int("live_edges", res.EdgeCount)) changedRepos.Add(1) if res.RepoPrefix != "" { changedPrefixes.Store(res.RepoPrefix, struct{}{}) } else { scopeUnknown.Store(true) } } } } else { // No prior mtimes → full cold (re)index of this repo, // which is "changed" by definition. changedRepos.Add(1) if res, err := state.multiIndexer.TrackRepoCtx(ctx, entry); err != nil { logger.Warn("daemon: startup track failed", zap.String("path", entry.Path), zap.Error(err)) scopeUnknown.Store(true) } else if res != nil && res.RepoPrefix != "" { // A cold TrackRepoCtx is itself a full retrack — its // FileCount is the whole repo's file-level work. filesReindexed.Add(int64(res.FileCount)) changedPrefixes.Store(res.RepoPrefix, struct{}{}) } else { scopeUnknown.Store(true) } } elapsed := time.Since(repoStart) if elapsed > 2*time.Second { logger.Info("daemon: warmup repo elapsed", zap.String("path", entry.Path), zap.String("path_fn", pathFn), zap.Duration("elapsed", elapsed)) } }(entry) } }() } for _, entry := range repos { jobs <- entry } close(jobs) wg.Wait() timings.parse = time.Since(phaseStart) timings.filesReindexed = int(filesReindexed.Load()) logger.Info("daemon: warmup phase done", zap.String("phase", "parallel_parse"), zap.Duration("elapsed", time.Since(phaseStart))) parseStats := progress.Stats(string(progress.PhaseParse), phaseStart, len(repos), len(repos)) publishReadinessPhase(state, "parallel_parse_done", false, map[string]any{ "tracked_repos": len(repos), "elapsed_ms": time.Since(phaseStart).Milliseconds(), "elapsed_human": parseStats.Elapsed, "repos_per_sec": parseStats.ItemsPerSec, }) // Warm-restart fast path. When the reconcile loop above re-indexed // nothing, the persistent backend already carries every resolved and // derived edge from the prior run; the deferred per-repo passes, the // cross-repo resolve, and the graph-wide derivation passes would all // just recompute what's on disk. Skipping them is what turns a warm // restart from a multi-minute replay of the cold-warmup cost into a // near-instant "open store, reconcile zero files, start watching". // The in-memory backend reaches here too, but its snapshot replay // already restored the derived edges, so the skip is equally safe. anyChanged := changedRepos.Load() > 0 timings.reposChanged = int(changedRepos.Load()) logger.Info("daemon: warmup change detection", zap.Int64("changed_repos", changedRepos.Load()), zap.Int("tracked_repos", len(repos)), zap.Bool("global_passes", anyChanged)) // Materialize the changed-repo prefix set once. It feeds two consumers: // the warm-restart resolve scope (RunPreEnrichResolve) and the end-batch // clone-pass scope (ArmBatchScope). Building it once here avoids two // separate walks of the sync.Map. changed := make(map[string]struct{}) changedPrefixes.Range(func(k, _ any) bool { if p, ok := k.(string); ok { changed[p] = struct{}{} } return true }) // Resolve scope: restrict the warm-restart master resolve to the repos // that re-indexed, but only when every whole-graph-safety precondition // holds (see warmupResolveScope). Any uncertainty drops to a nil scope == // whole-graph resolve, exactly the pre-scoping behaviour. The scoped // global passes switch is honoured downstream in runMasterResolve, // matching ArmBatchScope. resolveScope := warmupResolveScope(changed, len(repos), anyChanged, scopeUnknown.Load(), state.snapshotPartial, storeNeedsRebuild(state.graph)) // Resolve references ahead of the slow enrichment pass so find_usages / // get_callers return complete results as soon as the daemon reports ready // — independent of semantic enrichment. RunPreEnrichResolve materialises // go.mod dep nodes, runs the same-repo master resolver, and runs the // cross-repo resolver. On the warm-restart fast path nothing changed, so // the persisted graph already carries resolved edges and we skip straight // to marking ready. if anyChanged { phaseStart = time.Now() publishReadinessPhase(state, "resolve", false, nil) state.multiIndexer.RunPreEnrichResolve(ctx, resolveScope) timings.resolve = time.Since(phaseStart) logger.Info("daemon: warmup phase done", zap.String("phase", "resolve"), zap.Duration("elapsed", time.Since(phaseStart))) publishReadinessPhase(state, "resolve_done", false, map[string]any{ "elapsed_ms": time.Since(phaseStart).Milliseconds(), }) } // References are resolved (or unchanged and already resolved on the warm // path): the graph is queryable. Flip ready before the multi-minute // enrichment so clients can start issuing queries immediately. Everything // below runs in the background after ready and finishes at MarkEnriched. if markReady != nil { markReady() } // Resume enrichment for any repo a prior process left partial / abandoned. // pendingEnrich reflects only this run's re-indexing work, so an unchanged // repo whose completion marker is absent (a cut-short pass writes none) // would never re-run its semantic pass. Seeding re-arms the gate from the // persisted marker so the deferred pass below resumes it — and runs that // block even on a warm restart that changed nothing on disk (anyChanged is // false). Cheap for a fully-enriched workspace: each already-complete repo // pays only a git rev-parse plus one marker lookup. enrichPending := state.multiIndexer.SeedPendingEnrichAll() if enrichPending > 0 && !anyChanged { logger.Info("daemon: warmup resuming incomplete enrichment on an otherwise-unchanged restart", zap.Int("repos_pending_enrich", enrichPending)) } // Drain deferred per-repo passes (semantic enrich / contract // extract+commit) serially across the indexers the parallel loop // populated. These run after ready: enrichment is a precision upgrade on // top of the already-queryable reference graph. RunDeferredPassesAll // re-runs the master resolver at its tail to lift placeholder edges the // enrichment + contract passes add. if anyChanged || enrichPending > 0 { phaseStart = time.Now() publishReadinessPhase(state, "deferred_passes_all", true, nil) timings.enrichScheduled = state.multiIndexer.RunDeferredPassesAll(ctx) logger.Info("daemon: warmup phase done", zap.String("phase", "deferred_passes_all"), zap.Duration("elapsed", time.Since(phaseStart))) publishReadinessPhase(state, "deferred_passes_all_done", true, map[string]any{ "elapsed_ms": time.Since(phaseStart).Milliseconds(), }) } // Rehydrate per-repo contract registries from the snapshot. Only // target indexers whose registry is still nil — a non-nil registry // means IncrementalReindex (or a fresh TrackRepoCtx) re-extracted // contracts from source, and that result is authoritative. Without // this, every steady-state repo's ContractRegistry stays nil and // MergedContractRegistry skips them, so `contracts` returns only // the contracts of repos whose files happened to change since the // last shutdown. { phaseStart = time.Now() injectedRepos, injectedCount := 0, 0 for prefix := range state.multiIndexer.AllMetadata() { idx := state.multiIndexer.GetIndexer(prefix) if idx == nil || idx.ContractRegistry() != nil { continue } // Primary path: rebuild the per-repo registry from // KindContract nodes already in the backend's graph. // The indexer stamps every contract record onto // Node.Meta at commit time, so the graph is the // authoritative source — no gob round-trip needed. reg := contracts.LoadRegistryFromGraph(state.graph, prefix) if reg == nil { // Fallback to the legacy gob-snapshot path for // daemons upgrading across this change. The // snapshot copy is read-only by this point so the // two sources can't drift mid-flight. cs, ok := state.snapshotContracts[prefix] if !ok || len(cs) == 0 { continue } reg = contracts.NewRegistry() for _, c := range cs { reg.Add(c) } } idx.SetContractRegistry(reg) injectedRepos++ injectedCount += len(reg.All()) } if injectedRepos > 0 { logger.Info("daemon: rehydrated contract registries from graph/snapshot", zap.Int("repos", injectedRepos), zap.Int("contracts", injectedCount), zap.Duration("elapsed", time.Since(phaseStart))) } } // Backfill `WorkspaceID` / `ProjectID` onto nodes and contracts // loaded from a legacy snapshot. Old snapshots have these fields // as zero (gob decodes unknown fields silently); without this // stamp the matcher's EffectiveWorkspace falls back to RepoPrefix // and explicit shared-workspace declarations stop working until // every file is touched. Idempotent — re-running on a stamped // graph is a no-op. phaseStart = time.Now() if nodes, conts := state.multiIndexer.BackfillWorkspaceSlugs(); nodes+conts > 0 { logger.Info("daemon: backfilled workspace/project slugs from .gortex.yaml", zap.Int("nodes", nodes), zap.Int("contracts", conts), zap.Duration("elapsed", time.Since(phaseStart))) } // Run a cross-repo resolution pass once warmup has stamped the // workspace slugs. Files touched by IncrementalReindex already // re-resolve via the per-repo Resolver; this catches cross-repo // edges in unchanged files plus stamps cross_workspace_deps // eligibility on stubs. Mirrors what MultiIndexer.IndexAll does // for a fresh-start daemon (where there's no snapshot to reconcile // against). After resolution, contract bridge edges may have // changed too, so ReconcileContractEdges runs again. if anyChanged { phaseStart = time.Now() publishReadinessPhase(state, "global_resolve", true, nil) state.multiIndexer.RunGlobalResolve() timings.globalResolve = time.Since(phaseStart) logger.Info("daemon: warmup phase done", zap.String("phase", "global_resolve"), zap.Duration("elapsed", time.Since(phaseStart))) publishReadinessPhase(state, "global_resolve_done", true, map[string]any{ "elapsed_ms": time.Since(phaseStart).Milliseconds(), }) } // Finish the batch: turn off the per-repo skip flag and run the // graph-wide derivation passes once. RunGlobalResolve above just // lifted the last cross-repo placeholder EdgeCalls, so EdgeTests // derivation here picks up cross-repo test→subject pairs that // were unresolved during the per-repo loop. On the warm-restart fast // path (nothing changed) ResetBatch clears the deferred-batch flags // without re-running those passes — the persisted graph already has // the derived edges. // Scope the per-repo clone detection + clone-index Rebuild in the // end_batch graph passes to the repos that actually re-indexed this // warmup. Dropped when a changed repo's prefix was indeterminate // (scopeUnknown) so a repo whose clones genuinely need recomputing is // never skipped. ArmBatchScope is a no-op when scoped global passes are // disabled or the set is empty (run every repo, the prior behaviour). if anyChanged && !scopeUnknown.Load() { state.multiIndexer.ArmBatchScope(changed) } phaseStart = time.Now() publishReadinessPhase(state, "end_batch", true, nil) if anyChanged { state.multiIndexer.EndBatch() } else { state.multiIndexer.ResetBatch() } timings.endBatch = time.Since(phaseStart) logger.Info("daemon: warmup phase done", zap.String("phase", "end_batch"), zap.Duration("elapsed", time.Since(phaseStart))) publishReadinessPhase(state, "end_batch_done", true, map[string]any{ "elapsed_ms": time.Since(phaseStart).Milliseconds(), }) // Restore the workspace vector index from the snapshot. The warmup // loop above ran with vector building skipped (SetSkipVectorBuild), // so the search backend is text-only at this point; ImportVectorIndex // wraps it into a HybridBackend with the cached vectors. This is the // step that lets a default-on daemon avoid re-embedding the whole // graph on every restart. SetSkipVectorBuild(false) afterwards means // any later file-change re-index rebuilds vectors normally. if vec := state.snapshotVector; len(vec.Index) > 0 { phaseStart = time.Now() if err := state.multiIndexer.ImportVectorIndex(vec.Index, vec.Dims, vec.Count); err != nil { logger.Warn("daemon: vector index restore failed — semantic search will rebuild on next index", zap.Error(err)) } else { logger.Info("daemon: restored vector index from snapshot", zap.Int("vectors", vec.Count), zap.Int("dims", vec.Dims), zap.Duration("elapsed", time.Since(phaseStart))) } state.multiIndexer.SetSkipVectorBuild(false) } watchCfgs := make(map[string]config.WatchConfig) for prefix := range state.multiIndexer.AllMetadata() { watchCfgs[prefix] = state.configManager.GetRepoConfig(prefix).Watch } mw, err := indexer.NewMultiWatcher(state.multiIndexer, watchCfgs, logger) if err != nil { logger.Warn("daemon: multi-watcher init failed", zap.Error(err)) return nil, timings } if err := mw.Start(); err != nil { logger.Warn("daemon: multi-watcher start failed", zap.Error(err)) return nil, timings } logger.Info("daemon: watching", zap.Int("repos", len(watchCfgs))) publishReadinessPhase(state, "watcher_started", true, map[string]any{ "watched_repos": len(watchCfgs), }) return mw, timings } // logWarmupSummary emits the one-line warmup recap: reconstructing what a // restart did otherwise means joining the parallel_parse / resolve / // deferred_passes_all / global_resolve / end_batch per-phase log lines (each // logged separately, above, as the phases run) plus the separate "graph // queryable" / "enrichment complete" lines. queryable is the elapsed time at // markReady (find_usages / get_callers become complete); total is the // elapsed time at MarkEnriched (the full warmup, including enrichment). func logWarmupSummary(logger *zap.Logger, warmup *warmupTimings, queryable, total time.Duration) { if logger == nil || warmup == nil { return } logger.Info("daemon: warmup summary", zap.Float64("parse_s", warmup.parse.Seconds()), zap.Float64("resolve_s", warmup.resolve.Seconds()), zap.Float64("global_resolve_s", warmup.globalResolve.Seconds()), zap.Float64("end_batch_s", warmup.endBatch.Seconds()), zap.Float64("queryable_s", queryable.Seconds()), zap.Int("repos_changed", warmup.reposChanged), zap.Int("files_reindexed", warmup.filesReindexed), zap.Int("enrich_scheduled", warmup.enrichScheduled), zap.Float64("total_s", total.Seconds())) } // publishReadinessPhase forwards a workspace_readiness phase // transition to the MCP server's readiness broadcaster. Safe to // call when the server isn't wired (single-process modes that // bypass the daemon). func publishReadinessPhase(state *daemonState, phase string, ready bool, extra map[string]any) { if state == nil || state.mcpServer == nil { return } state.mcpServer.PublishReadiness(phase, ready, extra) } // priorMtimesFromStore asks the backend for its persisted FileMtime // rows for the repo described by entry. Returns nil when the backend // doesn't implement the reader (in-memory backend) or has no recorded // mtimes for the repo (fresh cold start). When non-nil it short- // circuits the gob-snapshot lookup so the warm path is driven by // data the backend persisted itself. func priorMtimesFromStore(g graph.Store, cm *config.ConfigManager, entry config.RepoEntry, logger *zap.Logger) map[string]int64 { reader, ok := g.(graph.FileMtimeReader) if !ok { if logger != nil { logger.Info("daemon: priorMtimesFromStore: store does not implement FileMtimeReader") } return nil } // Key by the prefix the indexer actually registers the repo under — // a worktree instance persists its mtimes under `@`, // not the bare basename, so a plain ResolvePrefix would load the // canonical checkout's mtimes and force a full re-index every restart. effective := strings.TrimPrefix(indexer.EffectiveRepoPrefix(cm, entry), "/") repoCount := 1 if cm != nil { if g := cm.Global(); g != nil { repoCount = len(g.Repos) } } prefix, ok := warmMtimePrefix(effective, repoCount) if !ok { if logger != nil { logger.Info("daemon: priorMtimesFromStore: empty prefix", zap.String("entry_path", entry.Path), zap.String("entry_name", entry.Name)) } return nil } mtimes := reader.LoadFileMtimes(prefix) if logger != nil { logger.Info("daemon: priorMtimesFromStore loaded", zap.String("prefix", prefix), zap.Bool("single_repo", repoCount < 2), zap.Int("count", len(mtimes))) } return mtimes } // warmMtimePrefix picks the repo_prefix to look up persisted file mtimes // (and, by extension, to decide whether the warm-restart reconcile can run) // for a repo whose EffectiveRepoPrefix is `effective` in a daemon tracking // `repoCount` repos total. // // PURPOSE: single-repo daemons index WITHOUT a prefix — MultiIndexer. // indexSingleRepo / ReconcileRepoCtx only switch on a repo prefix once a // SECOND repo joins (the willBeMultiRepo gate). So a lone repo's nodes and // file_mtimes rows are persisted under "", while EffectiveRepoPrefix returns // the path basename (e.g. "drools"). Looking mtimes up under the basename // finds zero rows and forces a full cold re-index — and, with an API // embedder, a full (paid) re-embed — on every restart. // // RATIONALE: mirror the indexer's own single-vs-multi decision here so the // warm path keys mtimes exactly where they were written. In multi-repo mode // an empty effective prefix is untrustworthy (it would collide across repos), // so report ok=false and let the caller fall back to a cold index. // // KEYWORDS: warm-restart, repo-prefix, single-repo, file_mtimes, re-embed func warmMtimePrefix(effective string, repoCount int) (prefix string, ok bool) { if repoCount < 2 { return "", true } if effective == "" { return "", false } return effective, true } // reconcileFileCount returns the number of files to count as "reindexed" // for one repo's ReconcileRepoCtx result, for the warmup summary's // filesReindexed total. A full retrack has no enumerated changed-file set // (see the fullRetrack comment in ReconcileRepoCtx), so its whole FileCount // stands in for the work done; an incremental reconcile counts only the // files it actually touched (stale + deleted). func reconcileFileCount(res *indexer.IndexResult) int { if res == nil { return 0 } if res.FullRetrack { return res.FileCount } return res.StaleFileCount + res.DeletedFileCount } // storeNeedsRebuild reports whether the backend signalled, via the optional // NeedsRebuild capability, that a schema migration crossed a rung an ALTER // could not satisfy — so its persisted rows are in an old shape and the // warm/incremental reconcile must be bypassed for a full re-index. This is a // generic, opt-in capability probe: a backend implements NeedsRebuild() bool // to participate. The on-disk sqlite store does — it reports true for the one // open in which it dropped an incompatible-schema database and recreated it // empty (see store_sqlite.Store.NeedsRebuild). The in-memory store does not. func storeNeedsRebuild(g any) bool { rb, ok := g.(interface{ NeedsRebuild() bool }) return ok && rb.NeedsRebuild() } // warmupFullResolveForced reports whether the operator pinned the warm-restart // master resolve to the whole graph via GORTEX_WARMUP_FULL_RESOLVE=1 (or // "true"), overriding the changed-repo scoping. An escape hatch for the case // where a scoped resolve is suspected of missing edges. func warmupFullResolveForced() bool { v := os.Getenv("GORTEX_WARMUP_FULL_RESOLVE") return v == "1" || strings.EqualFold(v, "true") } // warmupResolveScope decides the changed-repo scope for the warm-restart // master resolve. It returns the changed set only when scoping is both safe // and beneficial; any whole-graph-safety precondition failure returns nil, // which runMasterResolve treats as a whole-graph resolve (the pre-scoping // behaviour). The guards mirror the reasons the reconcile loop already // distrusts a partial delta: // // - !anyChanged: nothing re-indexed, the resolve is skipped entirely. // - scopeUnknown: a changed repo's prefix was indeterminate — any per-repo // uncertainty forces the whole-graph pass. // - snapshotPartial: the load shed edges only a full pass can restore. // - needsRebuild: the backend crossed a schema-rebuild rung. // - GORTEX_WARMUP_FULL_RESOLVE=1: operator override. // - empty / all-repos-changed: scoping gains nothing over the full pass. func warmupResolveScope(changed map[string]struct{}, totalRepos int, anyChanged, scopeUnknown, snapshotPartial, needsRebuild bool) map[string]struct{} { switch { case !anyChanged, scopeUnknown, snapshotPartial, needsRebuild, warmupFullResolveForced(): return nil case len(changed) == 0, len(changed) >= totalRepos: return nil default: return changed } } // priorMtimesForEntry finds the snapshotted FileMtimes map for a // configured repo entry, matching on absolute RootPath. Falls back to // prefix-based lookup when no path match is found — useful if the // user's config moved but the prefix is stable. Returns nil when no // match exists (first startup, schema bump, or newly-added repo). func priorMtimesForEntry(repos map[string]*snapshotRepo, entry config.RepoEntry) map[string]int64 { if len(repos) == 0 { return nil } absPath, err := filepath.Abs(entry.Path) if err != nil { absPath = entry.Path } for _, r := range repos { if r == nil { continue } if r.RootPath == absPath { return r.FileMtimes } } if prefix := config.ResolvePrefix(entry); prefix != "" && prefix != "." { if r := repos[prefix]; r != nil { return r.FileMtimes } } return nil } // collectSnapshotRepos snapshots the per-repo metadata needed to // reconcile the next startup: RepoPrefix, RootPath, and FileMtimes. // Called from the shutdown and periodic-snapshot paths so restart // warmups can run IncrementalReindex instead of a full walk. func collectSnapshotRepos(mi *indexer.MultiIndexer) []snapshotRepo { if mi == nil { return nil } meta := mi.AllMetadata() if len(meta) == 0 { return nil } out := make([]snapshotRepo, 0, len(meta)) for prefix, m := range meta { if m == nil { continue } // Copy the mtimes map — saveSnapshot encodes asynchronously // on shutdown and we don't want a late watcher event mutating // the live map mid-encode. mtimes := make(map[string]int64, len(m.FileMtimes)) for k, v := range m.FileMtimes { mtimes[k] = v } out = append(out, snapshotRepo{ RepoPrefix: prefix, RootPath: m.RootPath, FileMtimes: mtimes, NodeCount: m.NodeCount, EdgeCount: m.EdgeCount, }) } return out } // collectSnapshotContracts flattens every per-repo contract registry // into a single wire-form slice ordered by repo prefix. The warmup path // will redistribute by RepoPrefix when loading, so cross-repo ordering // is irrelevant here; the stable per-prefix grouping just keeps logs // and diffs readable. Called at the same points as collectSnapshotRepos // so the header counts and the repo/contract records agree. func collectSnapshotContracts(mi *indexer.MultiIndexer) []snapshotContract { if mi == nil { return nil } prefixes := make([]string, 0) for prefix := range mi.AllMetadata() { prefixes = append(prefixes, prefix) } sort.Strings(prefixes) var out []snapshotContract for _, prefix := range prefixes { idx := mi.GetIndexer(prefix) if idx == nil { continue } reg := idx.ContractRegistry() if reg == nil { continue } for _, c := range reg.All() { out = append(out, toSnapshotContract(c)) } } return out } // collectSnapshotVector serializes the workspace-global semantic-search // vector index for the snapshot. The daemon's search backend is shared // across every tracked repo, so there is exactly one vector index; // MultiIndexer.ExportVectorIndex returns an empty blob when embeddings // are disabled or no vectors were built, in which case the snapshot // simply carries no vector data and the next warmup re-embeds. func collectSnapshotVector(mi *indexer.MultiIndexer) snapshotVector { if mi == nil { return snapshotVector{} } data, dims, count := mi.ExportVectorIndex() return snapshotVector{Index: data, Dims: dims, Count: count} }