zzet--gortex/cmd/gortex/daemon_state.go

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 `<base>@<workspace>`,
	// 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}
}