package indexer import ( "bytes" "context" "fmt" "os" "path/filepath" "runtime" "sort" "strings" "sync" "time" "go.uber.org/zap" "github.com/zzet/gortex/internal/config" "github.com/zzet/gortex/internal/contracts" "github.com/zzet/gortex/internal/embedding" "github.com/zzet/gortex/internal/graph" "github.com/zzet/gortex/internal/parser" "github.com/zzet/gortex/internal/pathkey" "github.com/zzet/gortex/internal/progress" "github.com/zzet/gortex/internal/resolver" "github.com/zzet/gortex/internal/search" "github.com/zzet/gortex/internal/search/trigram" "github.com/zzet/gortex/internal/semantic" ) // RepoMetadata holds per-repo indexing state. type RepoMetadata struct { RepoPrefix string RootPath string Identity *RepoIdentity LastIndexTime time.Time FileCount int NodeCount int EdgeCount int ParseErrors []IndexError FileMtimes map[string]int64 // IsWorktree records whether RootPath was a linked git worktree // (as opposed to a main checkout) at the time the repo was // tracked. Captured here because once the worktree directory is // removed from disk it can no longer be classified — the janitor // needs this remembered flag to know a vanished root was a // worktree and may be garbage-collected. IsWorktree bool // Unprefixed records that this repo was indexed in single-repo // mode: its nodes carry RepoPrefix="" and raw relative file paths. // The empty-prefix resolution fallback (RepoRoot, ResolveFilePath) // honours only a lone repo that actually minted unprefixed nodes — // without the provenance check, stale unprefixed nodes surviving a // track/untrack transition would resolve against whatever repo // happens to be the lone one. Eviction also branches on it: // unprefixed nodes are invisible to the byRepo bucket EvictRepo // walks, so UntrackRepo must evict them file-by-file. Unprefixed bool } // MultiIndexer orchestrates indexing across multiple repositories. type MultiIndexer struct { graph graph.Store registry *parser.Registry search search.Backend embedder embedding.Provider repos map[string]*RepoMetadata // repoPrefix → metadata indexers map[string]*Indexer // repoPrefix → per-repo indexer configMgr *config.ConfigManager logger *zap.Logger mu sync.RWMutex // reconcileMu serialises ReconcileContractEdges end-to-end. The pass // evicts the prior EdgeMatches / topic / bridge generation and mints // a fresh one across many independent graph-store writes — it is NOT // atomic. Several goroutines drive it concurrently (the periodic // janitor's ReconcileAll, the file-watcher's IncrementalReindex, // MCP-triggered track / untrack / index), and mi.mu is only taken in // fine-grained spots inside, not across the whole pass. Without this // lock two overlapping reconciles can interleave evict and mint and // persist a stale generation (a bridge wiped by the other run's // EvictFile after it was minted). A dedicated outer mutex keeps the // pass self-consistent without widening mi.mu's scope. reconcileMu sync.Mutex // stitchProber / proxyBudget wire the cross-daemon proxy-edge feature: // when set by the daemon entry point (flag on), every CrossRepoResolver // this MultiIndexer builds mints proxy edges to remote-owned symbols // on positive evidence. A nil prober keeps the resolvers on read-only // federation (the default). stitchProber resolver.RemoteDeclarationProber proxyBudget int // deferGlobalPasses, when set, propagates SetDeferGlobalPasses(true) // to every per-repo Indexer constructed by this MultiIndexer. Batch // callers (warmup, ReconcileAll) flip it on around their loop and // invoke RunGlobalGraphPasses once at the end so the O(global) walks // (InferImplements / InferOverrides / markTestSymbolsAndEmitEdges) // don't run R times against an R-repo graph. deferGlobalPasses bool // deferResolve, when set, propagates SetDeferResolve(true) to every // per-repo Indexer constructed by this MultiIndexer. Used by the // parallel warmup path: per-repo ResolveAll / contract extract / // semantic enrich mutate the shared graph, so running them in // parallel across repos races. With this flag the parallel loop // just parses; RunDeferredPassesAll runs the per-repo passes // serially after the loop. Independent of deferGlobalPasses — that // flag covers a separate (cheaper) set of O(global) walks. deferResolve bool // batchChangedPrefixes scopes the per-repo clone-detection and // clone-index Rebuild passes in RunGlobalGraphPasses to the repos that // actually re-indexed in the current batch. nil — the default, and what // every one-off EndBatch caller leaves it as — means "run the clone // passes for every tracked repo", the prior whole-workspace behaviour. // The daemon warmup arms it (ArmBatchScope) before EndBatch so an // N-repo warm restart where only one repo changed stops paying ~N // full-graph clone walks for repos whose clone edges are already on // disk. Clone edges are per-repo (no cross-repo pair is ever formed), // so an unchanged repo's persisted edges stay valid; its in-memory // incremental clone index is reseeded lazily on its first later edit. // Consumed and cleared by RunGlobalGraphPasses. Guarded by mi.mu. batchChangedPrefixes map[string]struct{} // resolverLSPHelper is the resolve-time LSP helper propagated // onto every per-repo Indexer and onto the global post-pass // resolver in RunDeferredPassesAll. nil means no LSP hot-path // (heuristic-only resolution, the pre-N5 behaviour). The // daemon installs the helper via SetResolverLSPHelper after // constructing the LSP router; a multi-repo composite helper // dispatches per-file by repo prefix. resolverLSPHelper resolver.LSPHelper // onRepoTracked, when non-nil, is invoked after a fresh // TrackRepoCtx call has resolved the repo's prefix and // absolute path but before indexing starts. The daemon uses // this hook to register a per-repo resolver-time LSP helper // against the LSPHelper registry so dynamically-tracked repos // participate in the N5 hot path without daemon restart. onRepoTracked func(prefix, absPath string) // skipVectorBuild, when set, propagates SetSkipVectorBuild(true) to // every per-repo Indexer this MultiIndexer constructs, so their // buildSearchIndex passes populate only the text index and never // embed. The daemon flips it on for the warmup loop when a snapshot // already carries the workspace vector index — re-embedding 30k+ // symbols only to overwrite them with the cached index is the // dominant restart cost. After warmup it restores the cached index // once via ImportVectorIndex and clears the flag. skipVectorBuild bool // embedChunkOpts is the AST sub-chunking configuration propagated // to every per-repo Indexer this MultiIndexer constructs. The zero // value leaves the chunker on its built-in defaults. embedChunkOpts embedding.ChunkOptions // embedMaxSymbols overrides the vector-index size cap propagated to // every per-repo Indexer. Zero keeps the built-in default. embedMaxSymbols int // embedAPIConcurrency bounds parallel embedding requests against an // API-backed embedder, propagated to every per-repo Indexer. Zero // keeps the built-in default. embedAPIConcurrency int // semanticMgr is the semantic enrichment manager propagated to // every per-repo Indexer. When nil (the default), per-repo // deferred passes skip semantic enrichment — this is the root // cause of "enrich:0" in daemon mode. Set by the daemon via // SetSemanticManager before IndexAll / TrackRepo. semanticMgr *semantic.Manager } // SetEmbedder installs the embedding provider every per-repo indexer // should use. Must be called before IndexAll / TrackRepo for vectors // to land in the graph — without this the fresh Indexer created per // repo has embedder=nil and buildSearchIndex skips the vector pass. // Safe to call zero or one times; subsequent calls silently replace. func (mi *MultiIndexer) SetEmbedder(e embedding.Provider) { mi.mu.Lock() defer mi.mu.Unlock() mi.embedder = e } // npmAliasResolver builds a resolver.NpmAliasResolver covering every // tracked repo's on-disk root. Installed on the global post-pass // resolver and the cross-repo resolver so a JS/TS import declared // through an npm alias resolves to a locally-vendored real package // anywhere in the workspace. Returns nil when no repo has a usable // root — callers treat that as "no alias rewriting". func (mi *MultiIndexer) npmAliasResolver() resolver.NpmAliasResolver { roots := map[string]string{} for prefix, meta := range mi.AllMetadata() { if meta != nil && meta.RootPath != "" { roots[prefix] = meta.RootPath } } idx := newNpmAliasIndex(roots) if idx == nil { return nil } return idx.Resolve } // workspaceMembershipResolver builds a resolver.WorkspaceMembership // covering every tracked repo's on-disk root. Installed on the global // post-pass resolver and the cross-repo resolver so a same-named import // collision is broken in favour of the importer's own package-manager // workspace member. Returns nil when no repo is a workspace root — // callers treat that as "no workspace signal". func (mi *MultiIndexer) workspaceMembershipResolver() resolver.WorkspaceMembership { roots := map[string]string{} for prefix, meta := range mi.AllMetadata() { if meta != nil && meta.RootPath != "" { roots[prefix] = meta.RootPath } } idx := newWorkspaceMembershipIndex(roots) if idx == nil { return nil } return idx.Lookup } // newPerRepoIndexer constructs a per-repo Indexer with the standard // MultiIndexer wiring (shared search backend, embedder if configured, // deferred-global-passes flag propagated). Centralised so the flag // plumbing stays in one place. func (mi *MultiIndexer) newPerRepoIndexer(cfg config.IndexConfig) *Indexer { idx := New(mi.graph, mi.registry, cfg, mi.logger) idx.search = mi.search if mi.embedder != nil { idx.SetEmbedder(mi.embedder) } idx.SetDeferGlobalPasses(mi.deferGlobalPasses) idx.SetDeferResolve(mi.deferResolve) idx.SetSkipVectorBuild(mi.skipVectorBuild) idx.SetEmbeddingChunkOptions(mi.embedChunkOpts) idx.SetEmbeddingMaxSymbols(mi.embedMaxSymbols) idx.SetEmbeddingAPIConcurrency(mi.embedAPIConcurrency) if mi.resolverLSPHelper != nil { idx.SetResolverLSPHelper(mi.resolverLSPHelper) } if mi.semanticMgr != nil { idx.SetSemanticManager(mi.semanticMgr) } return idx } // SetEmbeddingChunkOptions installs the AST sub-chunking configuration // every per-repo Indexer this MultiIndexer constructs should use, and // re-applies it to every per-repo Indexer already built. Call before // IndexAll / TrackRepo so the warmup indexers pick it up. func (mi *MultiIndexer) SetEmbeddingChunkOptions(opts embedding.ChunkOptions) { mi.mu.Lock() mi.embedChunkOpts = opts live := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { live = append(live, idx) } mi.mu.Unlock() for _, idx := range live { idx.SetEmbeddingChunkOptions(opts) } } // SetEmbeddingMaxSymbols installs the vector-index size cap every // per-repo Indexer this MultiIndexer constructs should use, and // re-applies it to every per-repo Indexer already built. Zero keeps // the built-in default. func (mi *MultiIndexer) SetEmbeddingMaxSymbols(n int) { mi.mu.Lock() mi.embedMaxSymbols = n live := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { live = append(live, idx) } mi.mu.Unlock() for _, idx := range live { idx.SetEmbeddingMaxSymbols(n) } } // SetEmbeddingAPIConcurrency installs the parallel-embedding-request // cap every per-repo Indexer this MultiIndexer constructs should use, // and re-applies it to every per-repo Indexer already built. Zero // keeps the built-in default; the cap only affects API-backed // embedders. func (mi *MultiIndexer) SetEmbeddingAPIConcurrency(n int) { mi.mu.Lock() mi.embedAPIConcurrency = n live := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { live = append(live, idx) } mi.mu.Unlock() for _, idx := range live { idx.SetEmbeddingAPIConcurrency(n) } } // SetSemanticManager installs the semantic enrichment manager every // per-repo Indexer this MultiIndexer constructs should use, and // re-applies it to every per-repo Indexer already built. Without // this call, daemon-mode enrichment produces zero results because // the per-repo Indexers never receive the semantic manager. func (mi *MultiIndexer) SetSemanticManager(m *semantic.Manager) { mi.mu.Lock() mi.semanticMgr = m live := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { live = append(live, idx) } mi.mu.Unlock() for _, idx := range live { idx.SetSemanticManager(m) } } // SetSkipVectorBuild controls whether per-repo Indexers constructed // from now on skip the embedding pass in buildSearchIndex (text index // only). The daemon enables it for the warmup loop when a snapshot // already carries the workspace vector index, then disables it and // restores the cached index once warmup finishes. It also re-applies // the flag to every per-repo Indexer already constructed so a flag // flip mid-lifecycle takes effect everywhere. func (mi *MultiIndexer) SetSkipVectorBuild(skip bool) { mi.mu.Lock() mi.skipVectorBuild = skip live := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { live = append(live, idx) } mi.mu.Unlock() for _, idx := range live { idx.SetSkipVectorBuild(skip) } } // SetResolverLSPHelper installs the resolve-time LSP helper used by // every per-repo Indexer this MultiIndexer constructs from now on, // and by the global post-pass resolver in RunDeferredPassesAll. Pass // nil to detach. Safe to call zero or one times; subsequent calls // silently replace and propagate to every existing per-repo indexer. func (mi *MultiIndexer) SetResolverLSPHelper(h resolver.LSPHelper) { mi.mu.Lock() mi.resolverLSPHelper = h live := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { live = append(live, idx) } mi.mu.Unlock() for _, idx := range live { idx.SetResolverLSPHelper(h) } } // SetOnRepoTracked installs a callback fired once per TrackRepoCtx // after the repo's prefix + absolute path have been resolved but // before indexing starts. The daemon registers per-repo resolver- // time LSP helpers from this hook so runtime-added repos // participate in the N5 hot path. Pass nil to detach. func (mi *MultiIndexer) SetOnRepoTracked(fn func(prefix, absPath string)) { mi.mu.Lock() mi.onRepoTracked = fn mi.mu.Unlock() } // BeginBatch enables deferred-global-passes mode for every per-repo // Indexer that this MultiIndexer constructs after the call AND for // every Indexer already in mi.indexers (so ReconcileAll's per-repo // IncrementalReindex calls also skip the O(global) walks). Pair with // EndBatch. func (mi *MultiIndexer) BeginBatch() { mi.mu.Lock() defer mi.mu.Unlock() mi.deferGlobalPasses = true for _, idx := range mi.indexers { idx.SetDeferGlobalPasses(true) } } // BeginParallelBatch is BeginBatch plus parallel-safety: it also // propagates SetDeferResolve(true) to every per-repo Indexer // constructed during the batch. Use this when running the per-repo // indexing loop across goroutines (warmup) — the parallel parsers // must not race each other inside ResolveAll / contract extract / // semantic enrich, which all mutate the shared graph. Pair with // EndBatch; call RunDeferredPassesAll between the parallel parse and // EndBatch to run the deferred per-repo passes serially. func (mi *MultiIndexer) BeginParallelBatch() { mi.mu.Lock() defer mi.mu.Unlock() mi.deferGlobalPasses = true mi.deferResolve = true for _, idx := range mi.indexers { idx.SetDeferGlobalPasses(true) } } // RunDeferredPassesAll drains the deferred per-repo passes (semantic // enrich / contract extract+commit) serially across the indexers the // parallel parse populated. Pairs with BeginParallelBatch: the parallel // loop parses with deferResolve on; this serial loop runs the passes that // would otherwise race on the shared graph. The references-completeness // resolve runs ahead of this in RunPreEnrichResolve (so the daemon can mark // itself queryable before enrichment); the per-repo resolver pass is // suppressed here because resolver.ResolveAll walks the entire shared graph // — paying it R times is O(R · E). One master resolver.New(graph).ResolveAll // runs at the end to lift the placeholder edges enrichment + contracts added. // // Returns the number of repos whose deferred semantic enrichment was // actually dispatched (pendingEnrich set, or forced via // GORTEX_WARMUP_FORCE_ENRICH) rather than skipped as unchanged. Sampled // before runDeferredEnrichParallel runs, since a successful non-partial // pass clears the flag it reads. // SeedPendingEnrichAll re-arms the deferred-enrichment gate for every tracked // repo whose persisted enrichment is known-incomplete at its current clean HEAD // (see Indexer.MaybeSeedPendingEnrich). The daemon warmup calls it after the // parallel parse and before RunDeferredPassesAll so a repo left partial or // abandoned by a prior process resumes even when no file changed this run. // Returns the number of repos that will enrich (already pending plus newly // seeded) — the caller uses a non-zero count to run the deferred passes on a // warm restart that changed nothing on disk. func (mi *MultiIndexer) SeedPendingEnrichAll() int { mi.mu.RLock() live := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { live = append(live, idx) } mi.mu.RUnlock() pending := 0 for _, idx := range live { if idx.MaybeSeedPendingEnrich() { pending++ } } return pending } func (mi *MultiIndexer) RunDeferredPassesAll(ctx context.Context) int { mi.mu.RLock() indexers := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { indexers = append(indexers, idx) } mi.mu.RUnlock() forced := os.Getenv("GORTEX_WARMUP_FORCE_ENRICH") == "1" enrichScheduled := 0 for _, idx := range indexers { if idx.semanticMgr == nil || !idx.semanticMgr.Enabled() || !idx.semanticMgr.HasProviders() { continue } if idx.pendingEnrich.Load() || forced { enrichScheduled++ } } for _, idx := range indexers { idx.SetSkipResolveInDeferred(true) } // Per-repo deferred work in three phases. gomod (materialises dep // contract nodes) and contracts (extract + commit, which walk repo edges) // mutate the shared graph in ways that race across repos, so they stay // serial. Enrichment is the dominant cost — LSP background-indexing and // hover I/O — and is safe to overlap across repos: the manager hands each // repo its own LSP provider instance, go-types stashes per repo, and every // provider serialises its graph mutations on the backend resolve mutex. for _, idx := range indexers { idx.runDeferredGoMod() } mi.runDeferredEnrichParallel(indexers) for _, idx := range indexers { idx.runDeferredContracts() } for _, idx := range indexers { idx.SetSkipResolveInDeferred(false) } // Re-run the master same-repo resolver to lift the placeholder edges the // enrichment + contract passes just added. The references-completeness // resolve already ran ahead of enrichment in RunPreEnrichResolve, so this // is the idempotent catch-up pass for edges minted during enrichment. // Whole-graph (nil scope): enrichment can mint placeholder edges in any // repo, and scoping this catch-up is left to a follow-up. mi.runMasterResolve(nil) return enrichScheduled } // runMasterResolve runs one same-repo resolver over the whole shared graph, // lifting every placeholder edge to its canonical target. Split out so the // pre-enrichment resolve stage (RunPreEnrichResolve) and the post-enrichment // catch-up (RunDeferredPassesAll) share one implementation. // scope, when non-empty, restricts the pass to the edges that could resolve // into one of the named changed repos (see resolver.SetScope). It is honoured // only when scoped global passes are enabled; a nil / empty scope or a // disabled switch runs the whole-graph resolve, exactly the prior behaviour. func (mi *MultiIndexer) runMasterResolve(scope map[string]struct{}) { if mi.graph == nil { return } master := resolver.New(mi.graph) master.SetLogger(mi.logger) // The master resolve is the only pass with whole-graph evidence, so it is // the one allowed to durably flag terminally-unresolved edges (permanently // external / stdlib / definition-less) that a later scoped warm resolve can // skip. A scoped master pass leaves the flag alone (stamping is gated on an // empty scope inside ResolveAll); only a full pass stamps and self-heals. master.SetStampTerminal(true) // Mirror the resolve-time LSP helper onto the master pass so TS/JS-family // edges pick up LSP-precision answers just like the per-repo passes do. if mi.resolverLSPHelper != nil { master.SetLSPHelper(mi.resolverLSPHelper) } master.SetNpmAliasResolver(mi.npmAliasResolver()) master.SetPathAliasResolver(mi.pathAliasResolver()) master.SetWorkspaceMembership(mi.workspaceMembershipResolver()) scoped := len(scope) > 0 && mi.scopedGlobalPassesEnabled() if scoped { master.SetScope(scope) } mt := time.Now() stats := master.ResolveAll() mi.logger.Info("DEFERRED-TIMING master.ResolveAll", zap.Duration("elapsed", time.Since(mt)), zap.Bool("scoped", scoped), zap.Int("scope_repos", len(scope)), zap.Int("pending_before", stats.PendingBefore), zap.Int("pending_after", stats.PendingAfter)) } // RunPreEnrichResolve runs the resolution stage that makes references queryable // ahead of the slow semantic-enrichment pass. It materialises go.mod dep // contract nodes (so the resolver's import bridge can re-target Go imports of // declared modules), runs the same-repo master resolver to lift every // parse-time placeholder edge to its canonical target, then runs the cross-repo // resolver so references that span repo boundaries resolve too. Contract-bridge // reconciliation is intentionally left to RunGlobalResolve, which runs after the // contract pass has committed its nodes. // // The daemon warmup calls this between the parallel parse and the enrichment // phase, then marks itself ready — so find_usages / get_callers return complete // results as soon as the graph is queryable, independent of enrichment. // scope, when non-empty, restricts the same-repo master resolve to the // changed repos (see runMasterResolve / resolver.SetScope). The daemon warmup // passes the set of repos that re-indexed so a warm restart of one repo out of // many skips a whole-graph same-repo resolve; a nil / empty scope keeps the // whole-graph behaviour. The cross-repo resolve below stays whole-graph // regardless — it is the only pass that binds an unchanged repo's inbound // reference into a symbol a changed repo just added, and those inbound edges // must be resolved before the daemon reports ready (see runCrossRepoResolve). func (mi *MultiIndexer) RunPreEnrichResolve(ctx context.Context, scope map[string]struct{}) { mi.mu.RLock() indexers := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { indexers = append(indexers, idx) } mi.mu.RUnlock() for _, idx := range indexers { idx.runDeferredGoMod() } mi.runMasterResolve(scope) // Cross-repo references resolve here too so a multi-repo workspace is fully // queryable at "ready", not just within each repo. Whole-graph so inbound // references from unchanged repos into the changed repos bind before ready. mi.runCrossRepoResolve(false) } // runDeferredEnrichParallel runs each indexer's semantic enrichment in a // bounded worker pool. Concurrency is capped so at most a few LSP servers // background-index at once (the memory-sensitive part). The manager pins each // repo's LSP provider in-use for the duration of its pass, so the router's // LRU evictor never closes a provider another repo is still enriching against. func (mi *MultiIndexer) runDeferredEnrichParallel(indexers []*Indexer) { // Per-repo language sets computed once from a single graph-stats scan, // shared by the spec-grouped ordering and the batch pool-raise sizing // so the Manager's per-repo enrichment scan is not duplicated here. langSets := mi.batchLanguageSets(indexers) // Deterministic, spec-grouped order: repos needing the same language // servers run contiguously so the router's capped provider pool cycles // through far fewer distinct (spec, workspace) keys — a warmed server // stays alive across the runs that need it instead of being evicted and // respawned per repo. indexers = orderIndexersBySpecGroup(indexers, langSets) conc := enrichConcurrency(len(indexers)) // Temporarily raise the router's live-provider cap for the batch so the // concurrent passes don't evict each other's warmed servers, restoring // it (and logging the churn observed) when the batch drains. restore := mi.scopeRouterPoolForBatch(langSets, conc) defer restore() if conc <= 1 { for _, idx := range indexers { idx.runDeferredEnrich() } return } sem := make(chan struct{}, conc) var wg sync.WaitGroup for _, idx := range indexers { wg.Add(1) sem <- struct{}{} go func(idx *Indexer) { defer func() { <-sem; wg.Done() }() idx.runDeferredEnrich() }(idx) } wg.Wait() } // maxBatchProviders caps the temporary live-provider pool raise during // batch enrichment. Even a batch touching many distinct language servers // at high concurrency is held to this ceiling so warmup cannot spawn an // unbounded number of LSP subprocesses at once. const maxBatchProviders = 12 // batchLanguageSets returns each indexer's sorted set of present languages, // computed from a single RepoStats scan (one pass over the shared graph) // rather than a per-repo node scan. The sets drive both the spec-grouped // enrich ordering and the batch pool-raise sizing. func (mi *MultiIndexer) batchLanguageSets(indexers []*Indexer) map[*Indexer][]string { out := make(map[*Indexer][]string, len(indexers)) var stats map[string]graph.GraphStats if mi.graph != nil { stats = mi.graph.RepoStats() } for _, idx := range indexers { var langs []string if s, ok := stats[idx.repoPrefix]; ok { for l := range s.ByLanguage { langs = append(langs, l) } } sort.Strings(langs) out[idx] = langs } return out } // orderIndexersBySpecGroup returns a stable, deterministic ordering of // indexers grouped by their language set (so repos needing the same LSP // servers run contiguously), breaking ties by repo prefix. It does not // mutate the input slice. func orderIndexersBySpecGroup(indexers []*Indexer, langSets map[*Indexer][]string) []*Indexer { out := make([]*Indexer, len(indexers)) copy(out, indexers) sort.SliceStable(out, func(i, j int) bool { ki := strings.Join(langSets[out[i]], ",") kj := strings.Join(langSets[out[j]], ",") if ki != kj { return ki < kj } return out[i].repoPrefix < out[j].repoPrefix }) return out } // distinctBatchSpecs counts the enabled, available LSP specs whose // languages intersect any language present in the batch — i.e. how many // distinct language servers the batch will actually drive. func distinctBatchSpecs(langSets map[*Indexer][]string, router semantic.LSPRouter) int { langs := make(map[string]bool) for _, ls := range langSets { for _, l := range ls { langs[l] = true } } if len(langs) == 0 { return 0 } seen := make(map[string]bool) for _, name := range router.EnabledSpecNames() { if !router.SpecAvailable(name) { continue } for _, l := range router.SpecLanguages(name) { if langs[l] { seen[name] = true break } } } return len(seen) } // scopeRouterPoolForBatch raises the LSP router's live-provider cap to // enrichConcurrency × distinct-provider-specs (ceiling maxBatchProviders) // for the duration of a batch enrichment pass, so the concurrent passes // don't evict each other's warmed servers. It returns a restore closure // that puts the cap back and logs the eviction churn observed during the // batch. Safe no-op when no router is installed. func (mi *MultiIndexer) scopeRouterPoolForBatch(langSets map[*Indexer][]string, conc int) func() { if mi.semanticMgr == nil { return func() {} } router := mi.semanticMgr.LSPRouter() if router == nil { return func() {} } before := router.EvictionCount() old := router.MaxAlive() raised := false if specs := distinctBatchSpecs(langSets, router); specs > 0 { needed := conc * specs if needed > maxBatchProviders { needed = maxBatchProviders } if needed > old { router.SetMaxAlive(needed) raised = true mi.logger.Info("LSP router pool temporarily raised for batch enrichment", zap.Int("from", old), zap.Int("to", needed), zap.Int("distinct_specs", specs), zap.Int("concurrency", conc), ) } } return func() { if raised { router.SetMaxAlive(old) } mi.logger.Info("batch enrichment LSP provider churn", zap.Uint64("evictions", router.EvictionCount()-before), zap.Bool("pool_raised", raised), ) } } // enrichConcurrency caps how many repos enrich at once during batch warmup. // Half the CPUs, clamped to [1,4] and to the repo count — a few concurrent // LSP servers background-index in parallel without a memory blow-up. func enrichConcurrency(repos int) int { c := runtime.NumCPU() / 2 if c > 4 { c = 4 } if c < 1 { c = 1 } if c > repos { c = repos } return c } // EndBatch turns off deferred-global-passes mode and runs the graph- // wide derivation passes (InferImplements, InferOverrides, // markTestSymbolsAndEmitEdges) once against the shared graph. Restores // the per-Indexer flag too so a subsequent one-off TrackRepoCtx call // runs the passes inline as expected. // ArmBatchScope records the prefixes of the repos that re-indexed in the // batch about to be ended, so the next RunGlobalGraphPasses runs the // per-repo clone-detection + clone-index Rebuild passes only for those // repos instead of for every tracked repo. An empty set, or scoped global // passes being disabled, leaves the scope nil (run all). Only the daemon // warmup arms it; every other EndBatch caller keeps the whole-workspace // behaviour. func (mi *MultiIndexer) ArmBatchScope(changedPrefixes map[string]struct{}) { if mi == nil || len(changedPrefixes) == 0 || !mi.scopedGlobalPassesEnabled() { return } mi.mu.Lock() mi.batchChangedPrefixes = changedPrefixes mi.mu.Unlock() } // takeBatchScope returns the armed clone-pass scope and clears it, so the // scope governs exactly one RunGlobalGraphPasses run. A nil result means // "no scope — run the clone passes for every repo". func (mi *MultiIndexer) takeBatchScope() map[string]struct{} { mi.mu.Lock() scope := mi.batchChangedPrefixes mi.batchChangedPrefixes = nil mi.mu.Unlock() return scope } // scopedGlobalPassesEnabled reports whether per-repo global passes may be // scoped to the changed-repo set. GORTEX_INDEX_SCOPED_GLOBAL_PASSES // overrides the config key; default ON, mirroring Indexer.scopedGlobalPassesEnabled. func (mi *MultiIndexer) scopedGlobalPassesEnabled() bool { if v := os.Getenv("GORTEX_INDEX_SCOPED_GLOBAL_PASSES"); v != "" { return v == "1" || strings.EqualFold(v, "true") } mi.mu.RLock() defer mi.mu.RUnlock() for _, idx := range mi.indexers { return idx.config.ScopedGlobalPassesEnabledOrDefault() } return true } func (mi *MultiIndexer) EndBatch() { mi.mu.Lock() mi.deferGlobalPasses = false mi.deferResolve = false for _, idx := range mi.indexers { idx.SetDeferGlobalPasses(false) } mi.mu.Unlock() mi.RunGlobalGraphPasses(context.Background()) } // ResetBatch clears deferred-batch mode WITHOUT running the graph-wide // derivation passes. It is the warm-restart fast-path counterpart to // EndBatch: when the warmup reconcile loop observed zero changed files // across every repo, the persistent backend already holds every resolved // and derived edge from the prior run, so RunGlobalGraphPasses (plus the // RunDeferredPassesAll / RunGlobalResolve the caller also skips) would // only recompute what's already on disk — the work that turns a warm // restart into a 30s–500s stall. The per-Indexer SetDeferGlobalPasses // flag is still restored so a later watch-triggered TrackRepoCtx / // IncrementalReindex runs its passes inline as normal. func (mi *MultiIndexer) ResetBatch() { mi.mu.Lock() defer mi.mu.Unlock() mi.deferGlobalPasses = false mi.deferResolve = false for _, idx := range mi.indexers { idx.SetDeferGlobalPasses(false) } } // RunGlobalGraphPasses runs the graph-wide derivation passes once // against the shared graph: InferImplements (structural interface // satisfaction), InferOverrides (method-level overrides on // extends/implements/composes parents), and markTestSymbolsAndEmitEdges // (test→subject EdgeTests). Idempotent — graph.AddEdge dedupes by // edgeKey and the resolver passes skip already-present parents. func (mi *MultiIndexer) RunGlobalGraphPasses(ctx context.Context) { if mi.graph == nil { return } reporter := progress.FromContext(ctx) r := resolver.New(mi.graph) // Unconditional per-sub-pass timing. These global passes run serially and // were previously logged only when a pass emitted edges, so a slow but // low-yield pass left no breadcrumb — the source of the multi-minute // "silent" span after resolve on a cold index. globalStart := time.Now() // Acquire the changed-repo scope once for the whole run and derive the two // shapes the passes below consume. A nil scope means whole-graph — the // fresh-index / one-off behaviour every pass keeps as its fallback — so an // unscoped run is byte-identical to before. A non-nil scope (armed by the // daemon warmup / hourly janitor via ArmBatchScope) narrows each pass to the // repos that re-indexed this batch: an unchanged repo's derived edges are // already on disk and were never dropped, so re-deriving them is skipped. // - changedPrefixes: the changed-repo prefix set, for the edge-driven passes // (capability, test edges, framework synthesis, external calls), each of // which owns an edge iff its FROM node is in a changed repo. // - scopedTypeIfaceIDs: the changed repos' type/interface node IDs, for the // implements/overrides inference. The scoped inference keeps a pair when // EITHER endpoint is in this set, so a cross-repo override whose child is // in a changed repo and whose parent is in an unchanged one (or vice // versa) is still re-derived; structural implements never crosses repos // (its same-repo gate), so scoping both its sides here is complete. scope := mi.takeBatchScope() var changedPrefixes map[string]bool var scopedTypeIfaceIDs map[string]bool if scope != nil { changedPrefixes = make(map[string]bool, len(scope)) scopedTypeIfaceIDs = map[string]bool{} for prefix := range scope { changedPrefixes[prefix] = true if prefix == "" { continue } for _, n := range repoNodesLightOrFull(mi.graph, prefix) { if n == nil { continue } if n.Kind == graph.KindType || n.Kind == graph.KindInterface { scopedTypeIfaceIDs[n.ID] = true } } } } implStart := time.Now() implAdded := 0 switch { case scope == nil: implAdded = r.InferImplements() case len(scopedTypeIfaceIDs) > 0: // Empty set => no type/interface changed in the batch => no inferred // implements edge could have been dropped, so the pass is skipped. implAdded = r.InferImplementsScoped(scopedTypeIfaceIDs, scopedTypeIfaceIDs) } mi.logger.Info("global pass: infer implements", zap.Int("added", implAdded), zap.Bool("scoped", scope != nil), zap.Duration("elapsed", time.Since(implStart))) overStart := time.Now() overAdded := 0 switch { case scope == nil: overAdded = r.InferOverrides() case len(scopedTypeIfaceIDs) > 0: overAdded = r.InferOverridesScoped(scopedTypeIfaceIDs) } mi.logger.Info("global pass: infer overrides", zap.Int("added", overAdded), zap.Bool("scoped", scope != nil), zap.Duration("elapsed", time.Since(overStart))) testStart := time.Now() marked, emitted := markTestSymbolsAndEmitEdgesScoped(mi.graph, changedPrefixes) mi.logger.Info("global pass: test edges", zap.Int("test_symbols", marked), zap.Int("edges", emitted), zap.Duration("elapsed", time.Since(testStart))) capStart := time.Now() capRe, capEp, capFa := synthesizeCapabilityEdgesScoped(mi.graph, changedPrefixes) mi.logger.Info("global pass: capability edges", zap.Int("reads_env", capRe), zap.Int("executes_process", capEp), zap.Int("accesses_field", capFa), zap.Duration("elapsed", time.Since(capStart))) // Clone detection is PER-REPOSITORY: each tracked repo gets its own // finalise + detect over its own nodes (scoped by RepoPrefix), so no // cross-repo candidate pair is ever formed and each repo's boilerplate // CMS / threshold is computed from that repo's bodies alone. This // matches the per-repo incremental maintainer (cloneIndex.Rebuild / // UpdateFuncs) so the batch and incremental edge sets agree. reporter.Report("clone detection pass (global)", 0, 0) mi.mu.RLock() cloneIdx := make([]*Indexer, 0, len(mi.indexers)) for _, idx := range mi.indexers { cloneIdx = append(cloneIdx, idx) } mi.mu.RUnlock() // Scope to the repos that re-indexed this batch when the warmup armed a // batch scope. An unchanged repo's clone edges are already on disk and // its incremental clone index is reseeded lazily on its first later edit // (indexFile: if !built → Rebuild), so skipping its full-graph detect + // Rebuild here is sound and cuts an N-repo warm restart from N full-graph // clone walks to just the changed repos'. A nil scope runs every repo. The // scope is taken once at the top of RunGlobalGraphPasses and shared by every // pass; the clone passes reuse it here (takeBatchScope must not be called // twice — it clears the armed scope on the first read). inCloneScope := func(prefix string) bool { if scope == nil { return true } _, ok := scope[prefix] return ok } cloneDetectStart := time.Now() clonesDetected := 0 for _, idx := range cloneIdx { if !inCloneScope(idx.repoPrefix) { continue } clonesDetected++ // Per-repo threshold, NOT a max-over-repos value: the batch must use // the same cutoff the per-repo incremental maintainer uses // (UpdateFuncs/Rebuild → idx.cloneThreshold()), or the batch and // incremental edge sets diverge for any repo whose configured // threshold differs from the workspace maximum. if cs := detectClonesAndEmitEdgesCtx(ctx, mi.graph, idx.repoPrefix, idx.cloneThreshold()); cs.Items > 0 { mi.logger.Info("clone edges emitted (global)", zap.String("repo", idx.repoPrefix), zap.Int("items", cs.Items), zap.Int("clone_pairs", cs.Pairs), zap.Int("edges", cs.Edges), zap.Int("skipped_buckets", cs.SkippedBuckets), zap.Int("skipped_bucket_items", cs.SkippedBucketItems), zap.Int("diffused_pairs", cs.DiffusedPairs), zap.Int("diffused_edges", cs.DiffusedEdges), ) } } // Seed each per-repo indexer's incremental clone index from the // freshly-baselined signatures + sidecar (scoped to that repo's // prefix) so steady-state single-file edits after this batch go // incremental instead of re-running the whole-graph pass per file. cloneRebuildStart := time.Now() clonesRebuilt := 0 for _, idx := range cloneIdx { if !inCloneScope(idx.repoPrefix) { continue } if idx.cloneIndex != nil { idx.cloneIndex.Rebuild(mi.graph, idx.repoPrefix) clonesRebuilt++ } } // Aggregate timing for the clone passes — previously the most expensive // and least observable part of a warm restart (the per-repo logs above // only fire when a repo actually has clone pairs, so a long run could // pass with no breadcrumbs). mi.logger.Info("clone passes done (global)", zap.Bool("scoped", scope != nil), zap.Int("repos_total", len(cloneIdx)), zap.Int("detected", clonesDetected), zap.Int("rebuilt", clonesRebuilt), zap.Duration("detect_elapsed", cloneRebuildStart.Sub(cloneDetectStart)), zap.Duration("rebuild_elapsed", time.Since(cloneRebuildStart)), ) // Framework dynamic-dispatch synthesis (gRPC stubs, Temporal // workflow→activity, in-process / native event channels, native // bridges). After InferImplements/InferOverrides (the // interface-satisfaction signals) and before DetectCrossRepoEdges so // a cross-repo synthesized call gets its parallel cross_repo_calls // edge. reporter.Report("framework dispatch synthesis (global)", 0, 0) fwStart := time.Now() fwRep := resolver.RunFrameworkSynthesizersScoped(mi.graph, changedPrefixes) mi.logger.Info("global pass: framework dispatch synthesis", zap.Int("edges", fwRep.Total), zap.Any("per_synthesizer", fwRep.Per), zap.Int64("gate_ms", fwRep.GateMillis), zap.Int64("claim_ms", fwRep.ClaimMillis), zap.Int64("demote_ms", fwRep.DemoteMillis), zap.Duration("elapsed", time.Since(fwStart))) // External-call placeholder synthesis (opt-in). Runs after the // stub passes so only genuinely un-indexed external targets are // left to materialise into call-chain terminals. reporter.Report("external-call synthesis (global)", 0, 0) extStart := time.Now() extEnabled := mi.externalCallSynthesisEnabled() extCalls := 0 if scope != nil { extCalls = resolver.SynthesizeExternalCallsForRepos(mi.graph, extEnabled, changedPrefixes) } else { extCalls = resolver.SynthesizeExternalCalls(mi.graph, extEnabled) } mi.logger.Info("global pass: external-call synthesis", zap.Int("edges", extCalls), zap.Bool("scoped", scope != nil), zap.Duration("elapsed", time.Since(extStart))) // Cross-repo edge layer. Runs after InferImplements / InferOverrides // so the implements / extends edges they materialise across repo // boundaries pick up their parallel cross_repo_* edges. reporter.Report("cross-repo edges (global)", 0, 0) crStart := time.Now() crossRepoEdges := resolver.DetectCrossRepoEdges(mi.graph) mi.logger.Info("global pass: cross-repo edges", zap.Int("edges", crossRepoEdges), zap.Duration("elapsed", time.Since(crStart))) mi.logger.Info("global passes complete", zap.Duration("total", time.Since(globalStart))) } // repoNodesLightOrFull returns a repo's nodes for read-only structural // inspection (id / kind / repo prefix), preferring the meta-less // LightNodeReader fast path when the backend implements it so the enriched meta // blob a scope build never reads stays server-side. The returned nodes MUST NOT // be written back through AddNode/AddBatch — the light projection drops any // non-promoted meta — which holds here because the only caller reads struct // fields to build an ID set and discards the nodes. Falls back to the full // GetRepoNodes when the backend (e.g. in-memory) has no separate blob to skip. func repoNodesLightOrFull(g graph.Store, prefix string) []*graph.Node { if lr, ok := g.(graph.LightNodeReader); ok { return lr.GetRepoNodesLight(prefix) } return g.GetRepoNodes(prefix) } // externalCallSynthesisEnabled resolves whether external-call placeholder // synthesis should run over the shared graph. The pass is graph-wide, so // it is enabled when any tracked repo opted in — a repo that wants the // external hops in its call chains gets them even when a sibling repo // left the option off. func (mi *MultiIndexer) externalCallSynthesisEnabled() bool { mi.mu.RLock() defer mi.mu.RUnlock() for _, idx := range mi.indexers { if idx.externalCallSynthesisEnabled() { return true } } return false } // NewMultiIndexer creates a MultiIndexer. func NewMultiIndexer( g graph.Store, reg *parser.Registry, s search.Backend, cm *config.ConfigManager, logger *zap.Logger, ) *MultiIndexer { return &MultiIndexer{ graph: g, registry: reg, search: s, repos: make(map[string]*RepoMetadata), indexers: make(map[string]*Indexer), configMgr: cm, logger: logger, } } // IndexAll indexes all active repos concurrently. Each repo gets its own // Indexer instance with repo-specific config. Returns per-repo IndexResults. func (mi *MultiIndexer) IndexAll() (map[string]*IndexResult, error) { return mi.IndexScoped("", "") } // IndexScoped is IndexAll restricted to repos whose workspace and // project slugs match. Empty filters disable that axis (so empty/empty // is equivalent to IndexAll). Resolution honours the same precedence // as resolveWorkspaceID/resolveProjectID — RepoEntry override → // `.gortex.yaml::workspace` → repo prefix — so a `gortex server // --workspace foo` invocation matches both repos that declare // `workspace: foo` in their own `.gortex.yaml` and repos pinned to // `foo` from the user's global config. // // Returns an error when the filters exclude every active repo, so a // `--workspace typo` surfaces as a startup failure rather than a // silently empty graph. func (mi *MultiIndexer) IndexScoped(workspaceSlug, projectSlug string) (map[string]*IndexResult, error) { repos := mi.configMgr.ActiveRepos() if len(repos) == 0 { return nil, nil } if workspaceSlug != "" || projectSlug != "" { filtered := mi.filterReposByScope(repos, workspaceSlug, projectSlug) if len(filtered) == 0 { return nil, fmt.Errorf("scope filter matched zero of %d active repos (workspace=%q project=%q)", len(repos), workspaceSlug, projectSlug) } repos = filtered } // Single-repo mode: delegate without prefixing. if len(repos) == 1 { r, err := mi.indexSingleRepo(repos[0]) if err == nil { mi.ReconcileContractEdges() } return r, err } r, err := mi.indexMultiRepo(repos) if err == nil { mi.ReconcileContractEdges() } return r, err } // filterReposByScope returns the subset of repos whose resolved // workspace and project slugs match the supplied filters. Empty // filters disable that axis. Loads each repo's `.gortex.yaml` first so // resolution sees the workspace/project declared there — matching only // against `RepoEntry.Workspace` would miss repos that declare their // slug in their own config file (the typical case for first-party // repos). func (mi *MultiIndexer) filterReposByScope(repos []config.RepoEntry, workspaceSlug, projectSlug string) []config.RepoEntry { if workspaceSlug == "" && projectSlug == "" { return repos } out := make([]config.RepoEntry, 0, len(repos)) for _, e := range repos { absPath, err := filepath.Abs(e.Path) if err != nil { continue } prefix := config.ResolvePrefix(e) mi.configMgr.LoadWorkspaceConfig(prefix, absPath) cfg := mi.configMgr.GetRepoConfig(prefix) entryCopy := e if workspaceSlug != "" && resolveWorkspaceID(&entryCopy, cfg, prefix) != workspaceSlug { continue } if projectSlug != "" && resolveProjectID(&entryCopy, cfg, prefix) != projectSlug { continue } out = append(out, e) } return out } // indexSingleRepo indexes a single repo without prefixing for backward compatibility. func (mi *MultiIndexer) indexSingleRepo(entry config.RepoEntry) (map[string]*IndexResult, error) { absPath, err := filepath.Abs(entry.Path) if err != nil { return nil, fmt.Errorf("resolving path %s: %w", entry.Path, err) } prefix := config.ResolvePrefix(entry) mi.configMgr.LoadWorkspaceConfig(prefix, absPath) cfg := mi.configMgr.GetRepoConfig(prefix) idx := mi.newPerRepoIndexer(cfg.Index) entryCopy := entry idx.SetWorkspaceID(resolveWorkspaceID(&entryCopy, cfg, prefix)) idx.SetProjectID(resolveProjectID(&entryCopy, cfg, prefix)) // No repo prefix in single-repo mode. result, err := idx.Index(absPath) if err != nil { return nil, fmt.Errorf("indexing %s: %w", absPath, err) } result.RepoPrefix = prefix identity, _ := DetectIdentity(absPath) mi.mu.Lock() mi.repos[prefix] = &RepoMetadata{ RepoPrefix: prefix, RootPath: absPath, Identity: identity, LastIndexTime: time.Now(), FileCount: result.FileCount, NodeCount: result.NodeCount, EdgeCount: result.EdgeCount, ParseErrors: result.Errors, FileMtimes: idx.FileMtimes(), IsWorktree: ResolveWorktree(absPath).IsWorktree, Unprefixed: true, } mi.indexers[prefix] = idx mi.mu.Unlock() return map[string]*IndexResult{prefix: result}, nil } // migrateLoneUnprefixedRepoCtx re-mints the formerly-lone repo's nodes // with its real prefix the moment a second repo joins. Without it, the // first repo's unprefixed nodes become unreachable (the empty-prefix // fallback disarms at two repos) until a cold reindex. Ordering is // crash-safe: the prefixed re-index lands first, the stale unprefixed // nodes are evicted after — an interruption leaves both ID forms // resolvable rather than neither. func (mi *MultiIndexer) migrateLoneUnprefixedRepoCtx(ctx context.Context) { mi.mu.RLock() var oldPrefix string var oldMeta *RepoMetadata if len(mi.repos) == 1 { for p, m := range mi.repos { if m != nil && m.Unprefixed && m.RootPath != "" { oldPrefix, oldMeta = p, m } } } mi.mu.RUnlock() if oldMeta == nil { return } cfg := mi.configMgr.GetRepoConfig(oldPrefix) idx := mi.newPerRepoIndexer(cfg.Index) idx.SetRepoPrefix(oldPrefix) entry := config.RepoEntry{Path: oldMeta.RootPath, Name: oldPrefix} if mi.configMgr != nil { for _, e := range mi.configMgr.Global().Repos { if pathkey.EqualPaths(e.Path, oldMeta.RootPath) { entry = e break } } } idx.SetWorkspaceID(resolveWorkspaceID(&entry, cfg, oldPrefix)) idx.SetProjectID(resolveProjectID(&entry, cfg, oldPrefix)) result, err := idx.IndexCtx(ctx, oldMeta.RootPath) if err != nil { mi.logger.Warn("re-prefixing lone repo failed; its unprefixed nodes stay until a reindex", zap.String("prefix", oldPrefix), zap.Error(err)) return } // The prefixed nodes are live; now drop the unprefixed originals. // They are invisible to EvictRepo (no byRepo bucket entry), so evict // per file — unprefixed paths cannot collide with prefixed ones. for path := range oldMeta.FileMtimes { mi.graph.EvictFile(path) } // EvictFile clears only nodes+edges; the solo repo's sidecar rows // (file_mtimes, repo_index_state, enrichment_state, ...) were written // under the empty prefix and would otherwise be orphaned — the very next // warm restart would look for mtimes under the new prefix, find zero, and // full-re-track a repo that never changed. Re-key the '' sidecar residue // onto the new prefix. The re-mint re-index above already wrote fresh // new-prefix rows; RekeyRepoPrefix folds the prefix/path-keyed ones and // drops the node_id-keyed ones (whose ids changed under the re-mint) — // see its per-table rationale. Safe on '': the store is still single-repo // here, so '' holds only this repo's data (the synthetic global externals // a multi-repo graph parks under '' live in NODES, which the rekey — a // sidecar-only operation — never touches). if rk, ok := mi.graph.(interface { RekeyRepoPrefix(oldPrefix, newPrefix string) error }); ok { if err := rk.RekeyRepoPrefix("", oldPrefix); err != nil { mi.logger.Warn("re-keying lone repo sidecar rows failed; orphaned '' rows remain until purge", zap.String("prefix", oldPrefix), zap.Error(err)) } } mi.mu.Lock() mi.repos[oldPrefix] = &RepoMetadata{ RepoPrefix: oldPrefix, RootPath: oldMeta.RootPath, Identity: oldMeta.Identity, LastIndexTime: time.Now(), FileCount: result.FileCount, NodeCount: result.NodeCount, EdgeCount: result.EdgeCount, ParseErrors: result.Errors, FileMtimes: idx.FileMtimes(), IsWorktree: oldMeta.IsWorktree, } mi.indexers[oldPrefix] = idx mi.mu.Unlock() mi.logger.Info("re-minted lone repo with its prefix for multi-repo mode", zap.String("prefix", oldPrefix), zap.Int("nodes", result.NodeCount)) } // readGoModModule reads the module path from a go.mod file. func readGoModModule(repoPath string) string { data, err := os.ReadFile(filepath.Join(repoPath, "go.mod")) if err != nil { return "" } for _, line := range strings.Split(string(data), "\n") { line = strings.TrimSpace(line) if strings.HasPrefix(line, "module ") { return strings.TrimSpace(strings.TrimPrefix(line, "module")) } } return "" } // indexMultiRepo indexes multiple repos concurrently with repo prefixes. func (mi *MultiIndexer) indexMultiRepo(repos []config.RepoEntry) (map[string]*IndexResult, error) { type repoResult struct { prefix string result *IndexResult idx *Indexer meta *RepoMetadata err error } // Resolve each repo's prefix serially first: git-worktree instancing, // the derived-name persistence, and the cross-batch collision guard // must be deterministic, not raced across the parallel index // goroutines below. The same loop builds the tracked-module map used // for cross-repo dependency detection. type resolvedRepo struct { entry config.RepoEntry prefix string cfg *config.Config absPath string identity *RepoIdentity } trackedModules := make(map[string]string) resolved := make([]resolvedRepo, 0, len(repos)) seenPrefix := make(map[string]string, len(repos)) // prefix → absPath var resolveErrors []string for _, entry := range repos { absPath, err := filepath.Abs(entry.Path) if err != nil { resolveErrors = append(resolveErrors, fmt.Sprintf("resolving path %s: %v", entry.Path, err)) continue } identity, _ := DetectIdentity(absPath) e := entry prefix, cfg := mi.resolveTrackPrefix(&e, absPath, identity) // Two distinct checkouts can still land on the same derived prefix // (e.g. two worktrees declaring the same workspace). resolveTrackPrefix // only sees already-tracked repos, so guard against collisions within // this batch too. if prev, ok := seenPrefix[prefix]; ok && prev != absPath { prefix += "-" + shortPathHash(absPath) e.Name = prefix } seenPrefix[prefix] = absPath resolved = append(resolved, resolvedRepo{entry: e, prefix: prefix, cfg: cfg, absPath: absPath, identity: identity}) if mod := readGoModModule(absPath); mod != "" { trackedModules[prefix] = mod mi.logger.Debug("tracked repo module", zap.String("repo", prefix), zap.String("module", mod)) } } resultCh := make(chan repoResult, len(resolved)) var wg sync.WaitGroup for _, rr := range resolved { wg.Add(1) go func(r resolvedRepo) { defer wg.Done() idx := mi.newPerRepoIndexer(r.cfg.Index) idx.SetRepoPrefix(r.prefix) entryCopy := r.entry idx.SetWorkspaceID(resolveWorkspaceID(&entryCopy, r.cfg, r.prefix)) idx.SetProjectID(resolveProjectID(&entryCopy, r.cfg, r.prefix)) idx.SetTrackedRepoModules(trackedModules) // Defer the per-repo cross-cutting passes (ResolveAll, // semantic enrich, contract extract+commit) so they don't // race against each other across goroutines on the shared // graph. They run serially below via RunDeferredPasses after // wg.Wait(). The graph-wide derivation passes run once after // the loop via mi.RunGlobalGraphPasses(). idx.SetDeferResolve(true) result, err := idx.Index(r.absPath) if err != nil { resultCh <- repoResult{prefix: r.prefix, err: fmt.Errorf("indexing %s: %w", r.absPath, err)} return } result.RepoPrefix = r.prefix meta := &RepoMetadata{ RepoPrefix: r.prefix, RootPath: r.absPath, Identity: r.identity, LastIndexTime: time.Now(), FileCount: result.FileCount, NodeCount: result.NodeCount, EdgeCount: result.EdgeCount, ParseErrors: result.Errors, FileMtimes: idx.FileMtimes(), IsWorktree: ResolveWorktree(r.absPath).IsWorktree, } resultCh <- repoResult{prefix: r.prefix, result: result, idx: idx, meta: meta} }(rr) } go func() { wg.Wait() close(resultCh) }() results := make(map[string]*IndexResult) indexErrors := resolveErrors mi.mu.Lock() for rr := range resultCh { if rr.err != nil { mi.logger.Error("failed to index repo", zap.String("prefix", rr.prefix), zap.Error(rr.err)) indexErrors = append(indexErrors, rr.err.Error()) continue } mi.repos[rr.prefix] = rr.meta mi.indexers[rr.prefix] = rr.idx results[rr.prefix] = rr.result } mi.mu.Unlock() // Run the per-repo passes that the goroutines above deferred. Serial // across repos is the simple correctness fix: ResolveAll mutates // Edge.Meta on edges in the shared graph, and the contract pass walks // every edge — running them in parallel across repos races. Inside a // single repo's ResolveAll the resolver still uses its own worker // pool, and parsing (the dominant cost on a fresh index) already ran // in parallel above, so the wall-time hit is small. deferCtx := context.Background() for prefix := range results { if idx, ok := mi.indexers[prefix]; ok { idx.RunDeferredPasses(deferCtx) } } // Run a global cross-repo resolution pass once every repo is // indexed, with the cross-workspace boundary check wired in. // Without this, repos that import each other have unresolved // edges that only resolve when an editor touches a file (the // watcher path is the only other place this resolver runs). The // boundary lookup means cross-workspace candidates only resolve // when an explicit `cross_workspace_deps` declaration covers // them. cr := resolver.NewCrossRepo(mi.graph) cr.SetCrossWorkspaceDepLookup(mi.crossWorkspaceLookup()) cr.SetNpmAliasResolver(mi.npmAliasResolver()) cr.SetPathAliasResolver(mi.pathAliasResolver()) cr.SetWorkspaceMembership(mi.workspaceMembershipResolver()) mi.applyRemoteStitch(cr) cr.ResolveAll() mi.ReconcileContractEdges() // Graph-wide derivation passes run exactly once after every repo // has been parsed, every per-repo and cross-repo resolver has lifted // placeholder edges, and contract bridges are in place. RunDeferredPasses // intentionally skips these so we don't pay an O(global) walk per // repo (was the dominant cost at R≈100+). mi.RunGlobalGraphPasses(context.Background()) if len(indexErrors) > 0 && len(results) == 0 { return nil, fmt.Errorf("all repos failed to index: %s", strings.Join(indexErrors, "; ")) } return results, nil } // IndexRepo re-indexes a single repo by prefix. Evicts existing data first. func (mi *MultiIndexer) IndexRepo(repoPrefix string) (*IndexResult, error) { mi.mu.RLock() meta, ok := mi.repos[repoPrefix] mi.mu.RUnlock() if !ok { return nil, fmt.Errorf("repository not found: %s", repoPrefix) } // Evict existing data for this repo before re-indexing. Always — a lone // repo is now stored prefixed (see SetRepoPrefix below), so the eviction // must clear the prefixed slice regardless of repo count. mi.graph.EvictRepo(repoPrefix) mi.configMgr.LoadWorkspaceConfig(repoPrefix, meta.RootPath) cfg := mi.configMgr.GetRepoConfig(repoPrefix) idx := mi.newPerRepoIndexer(cfg.Index) // Always stamp the repo prefix, even when this is the only tracked repo. // The multi-repo cold path (indexMultiRepo) already prefixes // unconditionally; gating the single-repo re-index on repo count left the // two paths producing different id / path shapes for the same repo, so // adding a second repo turned the first from bare into prefixed — a lossy // whole-graph rewrite. Uniform prefixing removes the single/multi data-shape // split: a repo is prefixed from its first index, and adding another is // purely additive. idx.SetRepoPrefix(repoPrefix) entry := mi.configMgr.Global().FindRepoByPrefix(repoPrefix) idx.SetWorkspaceID(resolveWorkspaceID(entry, cfg, repoPrefix)) idx.SetProjectID(resolveProjectID(entry, cfg, repoPrefix)) result, err := idx.Index(meta.RootPath) if err != nil { return nil, fmt.Errorf("indexing %s: %w", meta.RootPath, err) } mi.mu.Lock() mi.repos[repoPrefix] = &RepoMetadata{ RepoPrefix: repoPrefix, RootPath: meta.RootPath, Identity: meta.Identity, LastIndexTime: time.Now(), FileCount: result.FileCount, NodeCount: result.NodeCount, EdgeCount: result.EdgeCount, ParseErrors: result.Errors, FileMtimes: idx.FileMtimes(), } mi.indexers[repoPrefix] = idx mi.mu.Unlock() // TODO: After re-indexing, run CrossRepoResolver.ResolveForRepo(repoPrefix) // to update cross-repo edges. This will be implemented in Task 7.1. mi.ReconcileContractEdges() return result, nil } // IncrementalReindexRepo incrementally re-indexes a single tracked repo // by prefix: only the files that changed since the last pass are // re-parsed and deleted files are evicted, against the repo's existing // per-repo Indexer (so its mtime snapshot is preserved). Unlike // IndexRepo it does NOT evict the whole repo first. // // When paths is non-empty the pass is scoped to those files / // directories; otherwise the whole repo root is scanned. Returns an // error when the prefix is not a tracked repo. func (mi *MultiIndexer) IncrementalReindexRepo(repoPrefix string, paths []string) (*IndexResult, error) { mi.mu.RLock() meta, ok := mi.repos[repoPrefix] idx := mi.indexers[repoPrefix] mi.mu.RUnlock() if !ok || meta == nil { return nil, fmt.Errorf("repository not found: %s", repoPrefix) } if idx == nil { // Tracked but no live indexer (e.g. restored from snapshot // without one) — fall back to a full re-index, which rebuilds // the per-repo indexer from scratch. return mi.IndexRepo(repoPrefix) } result, err := idx.IncrementalReindexPaths(meta.RootPath, paths) if err != nil { return nil, fmt.Errorf("reindexing %s: %w", meta.RootPath, err) } mi.mu.Lock() mi.repos[repoPrefix] = &RepoMetadata{ RepoPrefix: repoPrefix, RootPath: meta.RootPath, Identity: meta.Identity, LastIndexTime: time.Now(), FileCount: result.FileCount, NodeCount: result.NodeCount, EdgeCount: result.EdgeCount, ParseErrors: result.Errors, FileMtimes: idx.FileMtimes(), // Carried over from the prior metadata: this pass doesn't // change either property, and dropping them here (both zero // value on a fresh struct literal) used to silently flip a // worktree or an Unprefixed solo repo back to their false // defaults on the very first watcher-triggered incremental // update, defeating callers that key behaviour off them (see // the Unprefixed branch in cmd/gortex daemon status). IsWorktree: meta.IsWorktree, Unprefixed: meta.Unprefixed, } mi.mu.Unlock() mi.ReconcileContractEdges() return result, nil } // TrackRepo validates the path, detects identity, indexes, and adds to config. func (mi *MultiIndexer) TrackRepo(entry config.RepoEntry) (*IndexResult, error) { return mi.TrackRepoCtx(context.Background(), entry) } // resolveTrackPrefix determines the repo prefix that absPath should be // registered under and loads the per-repo `.gortex.yaml` config keyed to // that prefix. It is the single place that decides whether a git // worktree of an already-tracked repo becomes an INDEPENDENT instance // (see WorktreeInstanceName); the track and reconcile paths both call it // before their "already tracked?" check so a worktree that joins a // different workspace than the canonical no longer silently coalesces // into it. // // When a separate instance is created the derived `@` prefix // is written back into entry.Name so it is persisted to config and // reproduced deterministically on the next daemon warmup. entry is // mutated in place; callers pass a pointer to the value they will add to // the global config. func (mi *MultiIndexer) resolveTrackPrefix(entry *config.RepoEntry, absPath string, identity *RepoIdentity) (string, *config.Config) { base := config.ResolvePrefix(*entry) if base == "" || base == "." { if identity != nil { base = identity.RepoPrefix } } if mi.configMgr == nil { return base, config.Default() } // Load the repo's `.gortex.yaml` under the base prefix first so we can // read its declared workspace before deciding the final prefix. mi.configMgr.LoadWorkspaceConfig(base, absPath) cfg := mi.configMgr.GetRepoConfig(base) // An explicit Name already pins the prefix — honour it verbatim. This // is also the warm-restart fast path: once a worktree instance has // been persisted with its derived Name, every later load short-circuits // here without re-deriving. if entry.Name != "" { return base, cfg } declaredWS := entry.Workspace if declaredWS == "" && cfg != nil { declaredWS = cfg.Workspace } name, separate := WorktreeInstanceName(absPath, base, declaredWS, entry.AsWorktree) if !separate { return base, cfg } // Guard against two different checkouts colliding on the same derived // name (e.g. two worktrees that both declare workspace `x`): keep the // first, suffix the rest with a path hash. prefix := name mi.mu.RLock() existing, ok := mi.repos[prefix] mi.mu.RUnlock() if ok && existing != nil && !pathkey.SamePathIdentity(existing.RootPath, absPath) { prefix = name + "-" + shortPathHash(absPath) } entry.Name = prefix // persist the decision so warmup reproduces it mi.configMgr.LoadWorkspaceConfig(prefix, absPath) return prefix, mi.configMgr.GetRepoConfig(prefix) } // EffectiveRepoPrefix returns the prefix a repo entry is tracked under, // accounting for git-worktree instancing — the same value // resolveTrackPrefix registers, minus the (rare) collision-guard suffix // it cannot reproduce without the live registry. Warm-restart keying // (the snapshot-store mtime lookup, the resolve-time LSP helper // registry) uses this instead of config.ResolvePrefix so a disk-backed // reconcile finds the worktree instance's own persisted state rather // than the canonical checkout's. For a plain repo it equals // config.ResolvePrefix(entry). cm may be nil (then only an explicit // RepoEntry.Workspace override can trigger instancing). func EffectiveRepoPrefix(cm *config.ConfigManager, entry config.RepoEntry) string { base := config.ResolvePrefix(entry) if base == "" || base == "." || entry.Name != "" { return base } absPath, err := filepath.Abs(entry.Path) if err != nil { return base } declaredWS := entry.Workspace if declaredWS == "" && cm != nil { cm.LoadWorkspaceConfig(base, absPath) if cfg := cm.GetRepoConfig(base); cfg != nil { declaredWS = cfg.Workspace } } name, _ := WorktreeInstanceName(absPath, base, declaredWS, entry.AsWorktree) return name } // foldDistinctRepoCount counts configured repos by folded path identity, // so two entries that name the same directory under different spellings // (case or Unicode normalisation on a case-insensitive filesystem) count // once. It backstops the willBeMultiRepo decision: startup healing already // prunes such duplicates, but a not-yet-pruned config must not be allowed // to flip the graph into prefixed-ID mode for what is really one repo. func foldDistinctRepoCount(repos []config.RepoEntry) int { n := 0 seen := make([]string, 0, len(repos)) for _, e := range repos { abs, err := filepath.Abs(e.Path) if err != nil { abs = e.Path } dup := false for _, s := range seen { if pathkey.EqualPaths(s, abs) { dup = true break } } if dup { continue } seen = append(seen, abs) n++ } return n } // TrackRepoCtx is TrackRepo with a context, allowing callers to pipe progress // reporters (via progress.WithReporter) through to the underlying Index call. func (mi *MultiIndexer) TrackRepoCtx(ctx context.Context, entry config.RepoEntry) (*IndexResult, error) { absPath, err := filepath.Abs(entry.Path) if err != nil { return nil, fmt.Errorf("resolving path %s: %w", entry.Path, err) } // Normalise the volume (upper-case a Windows drive letter) so a newly // tracked path converges with os.Getwd's convention. The volume is // never part of a repo basename, so this cannot rotate a repo prefix; // no-op on POSIX. absPath = pathkey.NormalizeVolume(absPath) // Validate path exists and is a directory. info, err := os.Stat(absPath) if err != nil { return nil, fmt.Errorf("path does not exist: %s", absPath) } if !info.IsDir() { return nil, fmt.Errorf("path is not a directory: %s", absPath) } if reason, blocked := unsafeRootBlocked(absPath, entry.Force); blocked { return nil, fmt.Errorf("%s; pass force to track it anyway", reason) } identity, err := DetectIdentity(absPath) if err != nil { return nil, fmt.Errorf("detecting identity for %s: %w", absPath, err) } // Resolve the prefix (honouring git-worktree instancing) and load the // per-repo `.gortex.yaml` BEFORE the already-tracked check, so a // worktree that joins a different workspace than the canonical gets // its own `@` prefix instead of coalescing into it. cfg is // keyed to the final prefix; entry.Name is set when a separate // instance is created so the decision persists to config. Loading the // `.gortex.yaml` here also gives GetRepoConfig the workspace / project // slugs declared inside the repo (without it every repo would fall // back to `workspace = repoPrefix`, making shared-workspace cross-repo // matching impossible to express). prefix, cfg := mi.resolveTrackPrefix(&entry, absPath, identity) // Check if already tracked. The prefix-keyed check catches the common // case; the path scan additionally catches a case-only or Unicode // spelling variant of an already-tracked directory on a // case-insensitive filesystem (which resolves to a different derived // prefix and would otherwise be minted as a bogus second instance — // the very thing that flips the daemon into multi-repo mode, #270). mi.mu.RLock() if _, exists := mi.repos[prefix]; exists { mi.mu.RUnlock() return nil, nil // already tracked } for _, meta := range mi.repos { if meta != nil && pathkey.SamePathIdentity(meta.RootPath, absPath) { mi.mu.RUnlock() return nil, nil // same directory, different path spelling } } hook := mi.onRepoTracked mi.mu.RUnlock() if hook != nil { hook(prefix, absPath) } // Determine if we need to prefix. We must consider both repos already // indexed in mi.repos AND the total repos configured — at daemon warmup // TrackRepoCtx is called in a loop over all configured repos, so at // iteration 0 mi.repos is empty while the config already has N entries. // Counting only mi.repos used to leave the first-indexed repo without a // prefix while every later repo got one, producing two ID schemes for // the same graph and halving cross-file edge density. totalConfigured := 1 // ourselves if mi.configMgr != nil { totalConfigured = foldDistinctRepoCount(mi.configMgr.Global().Repos) } willBeMultiRepo := len(mi.repos)+1 >= 2 || totalConfigured >= 2 // A second repo joining a live single-repo daemon flips the graph // into prefixed-ID mode, but the first repo's nodes were minted // unprefixed — re-mint them before they become unreachable. if willBeMultiRepo { mi.migrateLoneUnprefixedRepoCtx(ctx) } idx := mi.newPerRepoIndexer(cfg.Index) if willBeMultiRepo { idx.SetRepoPrefix(prefix) } // Workspace / project slugs stamped on every node. Resolution // order (highest priority first): RepoEntry.Workspace from the // global config (lets users pin OSS repos without committing a // `.gortex.yaml`) → `.gortex.yaml::workspace` → repoPrefix // (default). resolveWorkspaceID encodes the precedence; the // WorkspaceID-keyed contract registry and the boundary-enforced // matcher both consume the result. entryCopy := entry idx.SetWorkspaceID(resolveWorkspaceID(&entryCopy, cfg, prefix)) idx.SetProjectID(resolveProjectID(&entryCopy, cfg, prefix)) result, err := idx.IndexCtx(ctx, absPath) if err != nil { return nil, fmt.Errorf("indexing %s: %w", absPath, err) } result.RepoPrefix = prefix mi.mu.Lock() mi.repos[prefix] = &RepoMetadata{ RepoPrefix: prefix, RootPath: absPath, Identity: identity, LastIndexTime: time.Now(), FileCount: result.FileCount, NodeCount: result.NodeCount, EdgeCount: result.EdgeCount, ParseErrors: result.Errors, FileMtimes: idx.FileMtimes(), IsWorktree: ResolveWorktree(absPath).IsWorktree, Unprefixed: !willBeMultiRepo, } mi.indexers[prefix] = idx mi.mu.Unlock() // Add to global config. entry.Path = absPath if err := mi.configMgr.Global().AddRepo(entry); err != nil { mi.logger.Warn("failed to add repo to config", zap.Error(err)) } // Skip the per-repo contract reconcile when batching: it walks every // edge in the shared graph to evict stale EdgeMatches and rebuilds // the matcher across every indexer, so paying it once per repo on a // warmup over 100+ repos is O(R · E). The batch caller runs it once // after the loop (RunGlobalResolve does, and the janitor's ReconcileAll // fires it post-loop too). if !mi.deferGlobalPasses { mi.ReconcileContractEdges() } return result, nil } // ReconcileRepoCtx registers a repo that already has nodes in the graph // (typically restored from a daemon snapshot) and brings it back into // agreement with the filesystem without a full re-index. priorMtimes // carries the mtimes recorded at the time the snapshot was taken; // IncrementalReindex uses them to detect files that changed offline // (re-indexes) and files that were deleted offline (evicts). // // Falls back to TrackRepoCtx when the repo is not yet tracked AND no // prior mtimes are available — in that case there's nothing to // reconcile against and a full index is the correct path. func (mi *MultiIndexer) ReconcileRepoCtx(ctx context.Context, entry config.RepoEntry, priorMtimes map[string]int64) (*IndexResult, error) { start := time.Now() absPath, err := filepath.Abs(entry.Path) if err != nil { return nil, fmt.Errorf("resolving path %s: %w", entry.Path, err) } // Normalise the volume (upper-case a Windows drive letter) to converge // with os.Getwd's convention. Cosmetic; never touches the basename. absPath = pathkey.NormalizeVolume(absPath) info, err := os.Stat(absPath) if err != nil { return nil, fmt.Errorf("path does not exist: %s", absPath) } if !info.IsDir() { return nil, fmt.Errorf("path is not a directory: %s", absPath) } identity, err := DetectIdentity(absPath) if err != nil { return nil, fmt.Errorf("detecting identity for %s: %w", absPath, err) } // Resolve the prefix (honouring git-worktree instancing) and load the // per-repo `.gortex.yaml` before the already-tracked check. Mirrors // TrackRepoCtx so a worktree instance keeps its derived prefix on the // reconcile path too; config can change between sessions and warmup- // time reconcile runs after a daemon restart. prefix, cfg := mi.resolveTrackPrefix(&entry, absPath, identity) // Already tracked — nothing to do. mi.mu.RLock() _, exists := mi.repos[prefix] mi.mu.RUnlock() if exists { return nil, nil } // Fall back to full TrackRepoCtx when we have no prior mtimes: // there's nothing meaningful to reconcile against, and // IncrementalReindex would treat every file as stale, producing // the same duplicate-writes problem we're fixing. entry.Name is // already pinned by resolveTrackPrefix, so the fallback reproduces // the same prefix. if len(priorMtimes) == 0 { return mi.TrackRepoCtx(ctx, entry) } totalConfigured := 1 if mi.configMgr != nil { totalConfigured = foldDistinctRepoCount(mi.configMgr.Global().Repos) } willBeMultiRepo := len(mi.repos)+1 >= 2 || totalConfigured >= 2 // Same transition guard as TrackRepoCtx: an already-reconciled // lone repo with unprefixed nodes must be re-minted before this // second repo flips the graph into prefixed-ID mode. if willBeMultiRepo { mi.migrateLoneUnprefixedRepoCtx(ctx) } idx := mi.newPerRepoIndexer(cfg.Index) if willBeMultiRepo { idx.SetRepoPrefix(prefix) } entryCopy := entry idx.SetWorkspaceID(resolveWorkspaceID(&entryCopy, cfg, prefix)) idx.SetProjectID(resolveProjectID(&entryCopy, cfg, prefix)) idx.SetRootPath(absPath) idx.SetFileMtimes(priorMtimes) // Choose the reconcile strategy from a census of what changed on disk // while the daemon was down. Scoped incremental is the default: re-index // only the changed files and evict only the deleted ones, leaving the // rest of the already-persisted graph untouched. A whole-repo re-track // (IndexCtx — which evicts and re-parses every file, then bulk-drains) // is reserved for the cases where scoping is unsafe or not worth it: the // census could not be taken, the churn is a large fraction of the repo, // or an operator forced it via GORTEX_WARMUP_FULL_RETRACK. A repo with // zero changes keeps the fast IncrementalReindex no-op (walk + 0 stale → // return), which is what makes an unchanged warm restart near-instant. // // The in-memory backend (*graph.Graph) keeps its exact prior behaviour: // IncrementalReindex is the authoritative path there — it evicts // offline-deleted files in place, has no reopened disk store, and so no // per-edge write to route around. Gate on the store type. _, memoryBacked := mi.graph.(*graph.Graph) var ( result *IndexResult changed, deleted []string route = "incremental" ) // fullRetrack is the whole-repo re-track — the ONE place FullRetrack is // stamped. StaleFileCount keeps its honest incremental-work meaning (0 // here) because the changed-file set is not enumerated on this path, so // callers must key "did this repo change" off FullRetrack instead. fullRetrack := func() (*IndexResult, error) { r, e := idx.IndexCtx(ctx, absPath) if e == nil && r != nil { r.FullRetrack = true } return r, e } switch { case memoryBacked: result, err = idx.IncrementalReindex(absPath) default: var censusErr error changed, deleted, censusErr = idx.ChangedSinceMtimes(absPath) churn := len(changed) + len(deleted) priorCount := len(priorMtimes) forceFull := os.Getenv("GORTEX_WARMUP_FULL_RETRACK") == "1" switch { case censusErr != nil || forceFull: route = "full_retrack" result, err = fullRetrack() case churn == 0: route = "incremental" result, err = idx.IncrementalReindex(absPath) case priorCount > 0 && churn*100 > priorCount*40: route = "full_retrack" result, err = fullRetrack() default: route = "scoped" result, err = idx.IncrementalReindexPaths(absPath, append(changed, deleted...)) } } if err != nil { return nil, fmt.Errorf("reconciling %s: %w", absPath, err) } result.RepoPrefix = prefix mi.mu.Lock() mi.repos[prefix] = &RepoMetadata{ RepoPrefix: prefix, RootPath: absPath, Identity: identity, LastIndexTime: time.Now(), FileCount: result.FileCount, NodeCount: result.NodeCount, EdgeCount: result.EdgeCount, ParseErrors: result.Errors, FileMtimes: idx.FileMtimes(), IsWorktree: ResolveWorktree(absPath).IsWorktree, Unprefixed: !willBeMultiRepo, } mi.indexers[prefix] = idx mi.mu.Unlock() entry.Path = absPath if err := mi.configMgr.Global().AddRepo(entry); err != nil { mi.logger.Warn("failed to add repo to config", zap.Error(err)) } // See TrackRepoCtx for why this is skipped under deferGlobalPasses. if !mi.deferGlobalPasses { mi.ReconcileContractEdges() } mi.logger.Info("daemon: reconciled repo from snapshot", zap.String("prefix", prefix), zap.String("route", route), zap.Int("changed", len(changed)), zap.Int("deleted", len(deleted)), zap.Bool("full_retrack", result.FullRetrack), zap.Int("stale_files_reindexed", result.StaleFileCount), zap.Duration("elapsed", time.Since(start))) if len(changed) > 0 || len(deleted) > 0 { mi.logger.Debug("daemon: reconcile changed-file census", zap.String("prefix", prefix), zap.Strings("changed", firstNStrings(changed, 5)), zap.Strings("deleted", firstNStrings(deleted, 5))) } return result, nil } // firstNStrings returns at most the first n elements of s — used to cap the // changed/deleted samples in the reconcile debug log so a large-churn repo // does not dump thousands of paths into the log line. func firstNStrings(s []string, n int) []string { if len(s) <= n { return s } return s[:n] } // ReconcileAll runs IncrementalReindex on every tracked repo. Used by // the daemon's periodic janitor to catch files whose mutations slipped // past fsnotify — inotify watch limits, NFS / SMB mounts, kernel event // queue overflow, or daemon downtime where edits happened while nobody // was listening. Cheap on steady-state repos (one filepath.WalkDir + // per-file os.Stat per repo), and correctness is self-healing: whatever // was missed gets picked up on the next tick. // // Returns a map of prefix → IndexResult for logging / metrics. Errors // per repo are logged and do not abort the rest of the sweep — a broken // permission bit on one repo should not starve reconciliation on the // others. func (mi *MultiIndexer) ReconcileAll() map[string]*IndexResult { mi.mu.RLock() prefixes := make([]string, 0, len(mi.indexers)) for p := range mi.indexers { prefixes = append(prefixes, p) } mi.mu.RUnlock() // Same batch trick as warmup: each per-repo IncrementalReindex // triggers an O(global) InferImplements/InferOverrides walk if we // don't suppress it. With ~100 repos that's ~100× the work for the // hourly janitor. mi.BeginBatch() // Always restore batch flags on exit (incl. panic) WITHOUT running the // graph-wide derivation passes — those are run explicitly below, and // only when a repo actually reindexed. The hourly janitor used to run // EndBatch unconditionally, walking the full graph (InferImplements / // InferOverrides / clone detection over hundreds of thousands of // edges) every cycle even when nothing changed — wasted CPU and, on a // small resident buffer pool, needless memory churn. defer mi.ResetBatch() results := make(map[string]*IndexResult, len(prefixes)) reindexed := 0 changedPrefixes := make(map[string]struct{}) for _, prefix := range prefixes { mi.mu.RLock() idx, ok := mi.indexers[prefix] meta, metaOK := mi.repos[prefix] mi.mu.RUnlock() if !ok || !metaOK || meta == nil || meta.RootPath == "" { continue } result, err := idx.IncrementalReindex(meta.RootPath) if err != nil { mi.logger.Warn("janitor: reconcile failed", zap.String("prefix", prefix), zap.Error(err)) continue } if result != nil && (result.StaleFileCount > 0 || result.DeletedFileCount > 0) { mi.logger.Info("janitor: reconciled repo", zap.String("prefix", prefix), zap.Int("stale_files_reindexed", result.StaleFileCount)) reindexed++ changedPrefixes[prefix] = struct{}{} } results[prefix] = result // Keep RepoMetadata.FileMtimes in sync so the next snapshot // picks up the reconciled mtimes. mi.mu.Lock() if m, ok := mi.repos[prefix]; ok && m != nil { m.FileMtimes = idx.FileMtimes() m.LastIndexTime = time.Now() } mi.mu.Unlock() } if reindexed > 0 { mi.ReconcileContractEdges() // Only now — when at least one repo actually reindexed — is it // worth the full-graph derivation pass. Nothing changed → skip it // (the deferred ResetBatch still clears the batch flags). Scope the // per-repo clone detection + Rebuild to just the repos that changed // this cycle: an unchanged repo's clone edges are already on disk, so // re-detecting all of them holds the resolve mutex for tens of seconds // and stalls every concurrent interactive edit. Warmup arms the same // scope; without this the janitor re-ran clone detection over every // tracked repo on every cycle that touched even one file. mi.ArmBatchScope(changedPrefixes) mi.RunGlobalGraphPasses(context.Background()) } return results } // UntrackRepo evicts a repo from the graph and removes it from config. func (mi *MultiIndexer) UntrackRepo(repoPrefix string) (int, int) { mi.mu.Lock() meta, ok := mi.repos[repoPrefix] if !ok { mi.mu.Unlock() return 0, 0 } delete(mi.repos, repoPrefix) delete(mi.indexers, repoPrefix) mi.mu.Unlock() var nodesRemoved, edgesRemoved int if meta.Unprefixed { // Single-repo-mode nodes carry RepoPrefix="" and never enter the // byRepo bucket EvictRepo walks — evict them file-by-file off the // recorded file set instead, or they linger in the graph and a // later lone repo would mis-resolve them. for path := range meta.FileMtimes { n, e := mi.graph.EvictFile(path) nodesRemoved += n edgesRemoved += e } } else if purger, ok := mi.graph.(interface{ PurgeRepo(string) error }); ok { // Prefer the full sidecar-aware purge. EvictRepo drops only // nodes+edges and leaves fifteen repo_prefix-keyed sidecar tables // (file_mtimes, *_enrichment, symbol_fts, content_fts, ...) behind, // which accumulate across untrack/retrack cycles until they dominate // a long-lived store. PurgeRepo clears them in one transaction. It // returns no counts, so report the repo's last-index metadata as the // removed estimate; fall back to EvictRepo (real counts) on error. if err := purger.PurgeRepo(repoPrefix); err != nil { mi.logger.Warn("purge repo failed; falling back to node/edge eviction", zap.String("prefix", repoPrefix), zap.Error(err)) nodesRemoved, edgesRemoved = mi.graph.EvictRepo(repoPrefix) } else { nodesRemoved, edgesRemoved = meta.NodeCount, meta.EdgeCount } } else { // Backends without the purge capability (the in-memory store has no // sidecars, so EvictRepo is already complete there). nodesRemoved, edgesRemoved = mi.graph.EvictRepo(repoPrefix) } // Remove from global config. if meta.RootPath != "" { if err := mi.configMgr.Global().RemoveRepo(meta.RootPath); err != nil { mi.logger.Warn("failed to remove repo from config", zap.String("prefix", repoPrefix), zap.Error(err)) } } mi.ReconcileContractEdges() return nodesRemoved, edgesRemoved } // WorktreeGC is the per-repo outcome of GCVanishedWorktrees. type WorktreeGC struct { RepoPrefix string RootPath string NodesRemoved int EdgesRemoved int } // GCVanishedWorktrees garbage-collects the index of any tracked linked // git worktree whose root directory has disappeared from disk — the // `git worktree remove` (or manual deletion) case. Each vanished // worktree's branch-keyed snapshot slot and graph nodes would otherwise // leak forever: a removed worktree never fires a per-file fsnotify // delete for its whole tree, and the janitor's IncrementalReindex just // errors out on the missing root without evicting anything. // // Only repos recorded as worktrees (RepoMetadata.IsWorktree) are // eligible — a vanished *main* checkout is left alone, since that is // far more likely a transient mount problem than an intentional // removal, and untracking it would also orphan every linked worktree // that shares its .git. The directory-existence test uses the same // not-exist-only rule as the per-file deletion detector, so a flaky // filesystem cannot trigger a destructive eviction. // // Returns one WorktreeGC record per repo evicted; an empty slice when // every tracked worktree is still present. func (mi *MultiIndexer) GCVanishedWorktrees() []WorktreeGC { // Snapshot the candidate set under the read lock, then evict // outside it — UntrackRepo takes the write lock itself. type candidate struct { prefix string root string } var candidates []candidate mi.mu.RLock() for prefix, meta := range mi.repos { if meta == nil || !meta.IsWorktree || meta.RootPath == "" { continue } if WorktreeRootGone(meta.RootPath) { candidates = append(candidates, candidate{prefix: prefix, root: meta.RootPath}) } } mi.mu.RUnlock() if len(candidates) == 0 { return nil } out := make([]WorktreeGC, 0, len(candidates)) for _, c := range candidates { nodes, edges := mi.UntrackRepo(c.prefix) mi.logger.Info("janitor: garbage-collected vanished worktree", zap.String("prefix", c.prefix), zap.String("root", c.root), zap.Int("nodes_removed", nodes), zap.Int("edges_removed", edges)) out = append(out, WorktreeGC{ RepoPrefix: c.prefix, RootPath: c.root, NodesRemoved: nodes, EdgesRemoved: edges, }) } return out } // GetMetadata returns the metadata for a specific repo, or nil if not found. func (mi *MultiIndexer) GetMetadata(repoPrefix string) *RepoMetadata { mi.mu.RLock() defer mi.mu.RUnlock() return mi.repos[repoPrefix] } // AllMetadata returns a copy of all repo metadata. func (mi *MultiIndexer) AllMetadata() map[string]*RepoMetadata { mi.mu.RLock() defer mi.mu.RUnlock() out := make(map[string]*RepoMetadata, len(mi.repos)) for k, v := range mi.repos { out[k] = v } return out } // IsMultiRepo returns true when more than one repo is tracked. func (mi *MultiIndexer) IsMultiRepo() bool { mi.mu.RLock() defer mi.mu.RUnlock() return len(mi.repos) > 1 } // RepoForFile returns the repo prefix for a given file path by checking // which repo root contains it. Returns empty string if no match. func (mi *MultiIndexer) RepoForFile(filePath string) string { absPath, err := filepath.Abs(filePath) if err != nil { return "" } mi.mu.RLock() defer mi.mu.RUnlock() var bestPrefix string var bestLen int for prefix, meta := range mi.repos { // Fold-aware containment so a case-variant file path still maps // to its repo on a case-insensitive filesystem. Longest-root-wins // breaks ties by nesting depth; nested roots share a prefix, so // the raw RootPath length orders them the same as their folded // forms would. if pathkey.HasPathPrefix(absPath, meta.RootPath) { if len(meta.RootPath) > bestLen { bestLen = len(meta.RootPath) bestPrefix = prefix } } } return bestPrefix } // GetIndexer returns the Indexer for a specific repo prefix, or nil. func (mi *MultiIndexer) GetIndexer(repoPrefix string) *Indexer { mi.mu.RLock() defer mi.mu.RUnlock() return mi.indexers[repoPrefix] } type grepRepoJob struct { prefix string idx *Indexer } func (mi *MultiIndexer) grepRepoJobs(repoAllow map[string]bool) []grepRepoJob { if mi == nil { return nil } mi.mu.RLock() defer mi.mu.RUnlock() capHint := len(mi.indexers) if repoAllow != nil { capHint = len(repoAllow) } jobs := make([]grepRepoJob, 0, capHint) for prefix, idx := range mi.indexers { if idx == nil { continue } if repoAllow != nil && !repoAllow[prefix] { continue } jobs = append(jobs, grepRepoJob{prefix: prefix, idx: idx}) } sort.Slice(jobs, func(i, j int) bool { return jobs[i].prefix < jobs[j].prefix }) return jobs } func singleAllowedRepo(repoAllow map[string]bool) (string, bool) { if repoAllow == nil { return "", false } var only string count := 0 for prefix, allowed := range repoAllow { if !allowed { continue } only = prefix count++ } return only, count == 1 } func stampGrepMatchPaths(prefix string, hits []trigram.Match) []trigram.Match { if prefix == "" { return hits } for i := range hits { hits[i].Path = prefix + "/" + hits[i].Path } return hits } func capGrepMatches(matches []trigram.Match, limit int) []trigram.Match { if limit > 0 && len(matches) > limit { return matches[:limit] } return matches } // GrepText fans out a trigram-accelerated literal search across every // tracked per-repo Indexer and returns the union, capped at limit. // Match paths are re-stamped from repo-root-relative to repo-prefixed // (e.g. "internal/foo.go" → "gortex/internal/foo.go") so callers can // route hits back to the right working tree without consulting the // MultiIndexer afterwards. A non-positive limit returns every match. // Returns nil when no per-repo indexer can serve the query — the // single-Indexer path (Indexer.GrepText) is used by callers without a // MultiIndexer. func (mi *MultiIndexer) GrepText(query string, limit int) []trigram.Match { return capGrepMatches(mi.GrepTextForRepos(query, nil, limit), limit) } // GrepTextForRepos is the scoped variant of GrepText. When repoAllow is // non-nil, only those repo prefixes are searched. perRepoLimit caps each // searched repo independently; the returned union is intentionally not // globally capped so callers can apply path / graph-scope filters first. func (mi *MultiIndexer) GrepTextForRepos(query string, repoAllow map[string]bool, perRepoLimit int) []trigram.Match { if mi == nil || query == "" { return nil } if prefix, ok := singleAllowedRepo(repoAllow); ok { idx := mi.GetIndexer(prefix) if idx == nil { return nil } return stampGrepMatchPaths(prefix, idx.GrepText(query, perRepoLimit)) } // Per-repo cap mirrors the caller's page size when set. The caller // applies the final cap after any path / graph-scope filters, so a // repo outside those filters cannot consume the page first. Zero / // negative means no per-repo cap (let each searcher return everything). jobs := mi.grepRepoJobs(repoAllow) out := make([]trigram.Match, 0, len(jobs)*8) for _, j := range jobs { hits := j.idx.GrepText(query, perRepoLimit) if len(hits) == 0 { continue } // Trigram emits forward-slash repo-relative paths. Stamp the repo // prefix so downstream tools (resolveGraphPath, path-prefix filters) // see the same shape they get from the graph nodes. out = append(out, stampGrepMatchPaths(j.prefix, hits)...) } return out } // GrepRegexp fans out a trigram-accelerated regex search across every // tracked per-repo Indexer and returns the union, capped at limit. // Mirrors GrepText: match paths are re-stamped from repo-root-relative // to repo-prefixed so downstream tools route hits back to the right // working tree. pathPrefix, when non-empty, restricts the scan to // files under that prefix on each indexer. A pattern that does not // compile in any indexer is reported once; per-indexer errors after // the first compile are otherwise treated as no-match. func (mi *MultiIndexer) GrepRegexp(pattern, pathPrefix string, limit int) ([]trigram.Match, error) { hits, err := mi.GrepRegexpForRepos(pattern, pathPrefix, nil, limit) if err != nil { return nil, err } return capGrepMatches(hits, limit), nil } // GrepRegexpForRepos is the scoped variant of GrepRegexp. repoAllow and // perRepoLimit have the same semantics as GrepTextForRepos. func (mi *MultiIndexer) GrepRegexpForRepos(pattern, pathPrefix string, repoAllow map[string]bool, perRepoLimit int) ([]trigram.Match, error) { if mi == nil || pattern == "" { return nil, nil } if prefix, ok := singleAllowedRepo(repoAllow); ok { idx := mi.GetIndexer(prefix) if idx == nil { return nil, nil } hits, err := idx.GrepRegexp(pattern, pathPrefix, perRepoLimit) if err != nil { return nil, err } return stampGrepMatchPaths(prefix, hits), nil } jobs := mi.grepRepoJobs(repoAllow) out := make([]trigram.Match, 0, len(jobs)*8) for _, j := range jobs { hits, err := j.idx.GrepRegexp(pattern, pathPrefix, perRepoLimit) if err != nil { // First compile error short-circuits — the pattern is the // caller's fault and won't compile in any other indexer // either (the trigram searcher uses the same regexp engine). return nil, err } if len(hits) == 0 { continue } out = append(out, stampGrepMatchPaths(j.prefix, hits)...) } return out, nil } // IndexerForFile routes an absolute path to the per-repo Indexer that // owns it. Returns (nil, "") when no tracked repo contains the path. // Used by the MCP overlay middleware to find the right Indexer for a // pushed file when constructing the per-request overlay layer. func (mi *MultiIndexer) IndexerForFile(absPath string) (*Indexer, string) { prefix := mi.RepoForFile(absPath) if prefix == "" { return nil, "" } return mi.GetIndexer(prefix), prefix } // ResolveFilePath takes a repo-prefixed relative path (e.g. "ade/internal/foo.go") // and returns the absolute filesystem path by looking up the repo's root directory. // Returns empty string if the repo prefix is not found. func (mi *MultiIndexer) ResolveFilePath(prefixedPath string) string { mi.mu.RLock() defer mi.mu.RUnlock() // Longest matching prefix wins. With worktree instances two prefixes // can share a leading segment (`oas-orm` vs `oas-orm@task-ws`); map // iteration order is random, so a plain first-match could resolve a // `oas-orm@task-ws/...` path against the shorter `oas-orm` root. The // "/"-boundary check already keeps the two disjoint, but ranking by // length makes that robust regardless of any future prefix shapes. var bestPrefix, bestRoot string for prefix, meta := range mi.repos { if meta == nil { continue } if strings.HasPrefix(prefixedPath, prefix+"/") && len(prefix) > len(bestPrefix) { bestPrefix, bestRoot = prefix, meta.RootPath } } if bestPrefix == "" { // Single-repo mode mints unprefixed graph paths; resolve them // against the lone registered repo instead of failing. if meta := mi.loneRepoLocked(); meta != nil && meta.RootPath != "" { return filepath.Join(meta.RootPath, prefixedPath) } return "" } // Collision guard for the lone unprefixed repo: its graph paths are // raw relative paths, so one whose first segment happens to equal // the repo's own prefix (repo "api" containing api/handlers.go) // would be hijacked by the prefix-strip join. Prefer the raw join // when that file actually exists on disk. if meta := mi.loneRepoLocked(); meta != nil && meta.RootPath != "" { raw := filepath.Join(meta.RootPath, prefixedPath) if _, err := os.Stat(raw); err == nil { return raw } } return filepath.Join(bestRoot, strings.TrimPrefix(prefixedPath, bestPrefix+"/")) } // RepoPrefixes returns the set of registered repo prefixes. The returned // slice is a snapshot — safe to retain and iterate concurrently with // other multi-indexer operations. Order is unspecified; callers that // need stability should sort. func (mi *MultiIndexer) RepoPrefixes() []string { mi.mu.RLock() defer mi.mu.RUnlock() prefixes := make([]string, 0, len(mi.repos)) for prefix := range mi.repos { prefixes = append(prefixes, prefix) } return prefixes } // RepoRoot returns the local filesystem root for the given repo prefix. // ok is true only when the prefix is registered AND meta.RootPath is non-empty. // Caller is responsible for joining repo-relative file paths against the root. // // The empty prefix resolves to the lone registered repo when exactly one is // tracked: single-repo mode indexes nodes without a repo prefix (see // indexSingleRepo) while registering its metadata under the repo's real // prefix, so every node the single-repo indexer mints carries RepoPrefix="" // — refusing the empty prefix would orphan all of them. With two or more // repos the empty prefix is ambiguous and stays a miss. func (mi *MultiIndexer) RepoRoot(repoPrefix string) (string, bool) { mi.mu.RLock() defer mi.mu.RUnlock() if repoPrefix == "" { if meta := mi.loneRepoLocked(); meta != nil && meta.RootPath != "" { return meta.RootPath, true } return "", false } meta, ok := mi.repos[repoPrefix] if !ok || meta == nil || meta.RootPath == "" { return "", false } return meta.RootPath, true } // loneRepoLocked returns the metadata of the only registered repo when // exactly one repo is tracked AND that repo was indexed unprefixed // (single-repo mode). The provenance check matters: after a 1→2→1 // track/untrack sequence the lone survivor can be a prefixed repo, and // stale unprefixed nodes from the departed repo must keep failing // closed instead of resolving against the wrong checkout. Caller must // hold mi.mu. func (mi *MultiIndexer) loneRepoLocked() *RepoMetadata { if len(mi.repos) != 1 { return nil } for _, meta := range mi.repos { if meta != nil && meta.Unprefixed { return meta } } return nil } // LinkedWorktreeRoots returns the on-disk roots of every tracked linked // git worktree that shares its .git common directory with the checkout // at mainRepoPath — i.e. the worktree siblings of that main repo. The // query is keyed on the resolved MainRepoPath so it matches whether the // caller passes a main checkout or one of its worktrees. // // Used by the edit-tool file resolver: because all worktrees of one // repo reuse a single index identity, a repo-relative path can resolve // against a sibling checkout. When the resolved file belongs to a // linked worktree, the resolver re-roots the write there so an edit // lands in the worktree's copy rather than the main checkout's. func (mi *MultiIndexer) LinkedWorktreeRoots(mainRepoPath string) []string { if mainRepoPath == "" { return nil } wantMain := resolvedMainRepo(mainRepoPath) mi.mu.RLock() defer mi.mu.RUnlock() var out []string for _, meta := range mi.repos { if meta == nil || meta.RootPath == "" || !meta.IsWorktree { continue } if resolvedMainRepo(meta.RootPath) == wantMain { out = append(out, meta.RootPath) } } return out } // resolvedMainRepo resolves a checkout path to its repo's main worktree // root with symlinks evaluated. ResolveWorktree derives the main path // two different ways — filepath.Abs for a main checkout vs git's // canonicalized `commondir` for a linked worktree — and on platforms // where the temp / home tree is a symlink (macOS /var -> /private/var) // those two forms differ for the same repo. Evaluating symlinks on the // result gives one stable identity that both inputs agree on. func resolvedMainRepo(path string) string { main := ResolveWorktree(path).MainRepoPath if resolved, err := filepath.EvalSymlinks(main); err == nil { return resolved } return main } // MergedContractRegistry combines contract registries from all per-repo // indexers into a single registry. In multi-repo mode each repo's indexer // runs extractContracts independently; this merges the results. func (mi *MultiIndexer) MergedContractRegistry() *contracts.Registry { mi.mu.RLock() defer mi.mu.RUnlock() merged := contracts.NewRegistry() for repoPrefix, idx := range mi.indexers { cr := idx.ContractRegistry() if cr == nil { continue } // Re-stamp the workspace/project slugs from the indexer // alongside the repo prefix on merge. The contracts already // carry these slugs from // their source registry, but AddAllScoped is idempotent (skips // non-empty existing values) so this stays correct even if a // future code path forgets the stamp on first insert. merged.AddAllScoped(cr.All(), repoPrefix, idx.WorkspaceID(), idx.ProjectID()) } return merged } // attachInlinedShapes folds the field shape of each contract's // response_type / request_type into the contract's Meta so the // dashboard can render the expanded field list. Targets contracts // where the type has been resolved to a graph node ID (contains // "::") AND the type node has a shape stored in its Meta. // // Type-shape extraction normally runs in commitContracts via // snapshotContractShapes + inlineEnvelopeShapes — but those passes // run during the initial extract and miss contracts added later by // InlineWrappers. This is the post-inline equivalent: it doesn't // re-extract shapes (the type nodes already have them from // snapshotContractShapes if they were referenced anywhere), it just // attaches them to the new contract entries. func (mi *MultiIndexer) attachInlinedShapes(cr *contracts.Registry, g graph.Store) { idsToTouch := map[string]bool{} for _, c := range cr.All() { if c.Meta == nil { continue } for _, key := range []string{"response_type", "request_type"} { if v, _ := c.Meta[key].(string); v != "" && strings.Contains(v, "::") { idsToTouch[c.ID] = true break } } if env, ok := c.Meta["response_envelope"].([]map[string]any); ok && len(env) > 0 { // Touch any contract that has an envelope, even when // the rows still carry bare type names — the loop below // upgrades them. Otherwise we skip them and lose the // shape attachment for sibling-file types. idsToTouch[c.ID] = true _ = env } } srcCache := map[string][]byte{} resolveShape := func(typeID string) any { if typeID == "" || !strings.Contains(typeID, "::") { return nil } node := g.GetNode(typeID) if node == nil { return nil } if node.Kind != graph.KindType && node.Kind != graph.KindInterface { return nil } if node.Meta == nil { node.Meta = map[string]any{} } if shape, ok := node.Meta["shape"]; ok && shape != nil { return shape } // Lazy-extract: snapshotContractShapes only walks types // referenced by the initial bulk extract. Types referenced // ONLY by wrapper-inlined contracts need this fallback or // their fields stay unread. src := srcCache[node.FilePath] if src == nil { data, ok := mi.readNodeSource(node) if !ok { srcCache[node.FilePath] = []byte{} return nil } src = data srcCache[node.FilePath] = src } if len(src) == 0 { return nil } extracted := contracts.ExtractShape(node.FilePath, src, node.StartLine, node.EndLine) if extracted == nil { return nil } node.Meta["shape"] = extracted return extracted } for id := range idsToTouch { items := cr.ByID(id) changed := false for i := range items { if items[i].Meta == nil { continue } // Top-level request/response type shapes. for _, pair := range []struct{ typeKey, shapeKey string }{ {"response_type", "response_shape"}, {"request_type", "request_shape"}, } { if _, has := items[i].Meta[pair.shapeKey]; has { continue } typeID, _ := items[i].Meta[pair.typeKey].(string) if shape := resolveShape(typeID); shape != nil { items[i].Meta[pair.shapeKey] = shape changed = true } } // Envelope rows — upgrade bare type names to graph IDs // (so the shape lookup hits) and attach shapes. if env, ok := items[i].Meta["response_envelope"].([]map[string]any); ok && len(env) > 0 { envChanged := false for ri, row := range env { typeID, _ := row["type"].(string) // Upgrade bare type name → graph ID when the // in-file resolveTypeInFile pass left it bare // (the type lives in a sibling file). if typeID != "" && !strings.Contains(typeID, "::") { matches := g.FindNodesByName(typeID) var resolved string for _, n := range matches { if n.Kind != graph.KindType && n.Kind != graph.KindInterface { continue } resolved = n.ID if items[i].RepoPrefix != "" && strings.HasPrefix(n.ID, items[i].RepoPrefix+"/") { break // prefer same-repo } } if resolved != "" { env[ri]["type"] = resolved typeID = resolved envChanged = true } } if _, has := row["shape"]; has { continue } if shape := resolveShape(typeID); shape != nil { env[ri]["shape"] = shape envChanged = true } } if envChanged { items[i].Meta["response_envelope"] = env changed = true } } } if changed { cr.ReplaceByID(id, items) } } } // readNodeSource returns the source bytes of the file the node lives // in, resolving the repo prefix to a real disk path via tracked-repo // metadata. Mirrors wrapperSourceReader's path-resolution dance. func (mi *MultiIndexer) readNodeSource(node *graph.Node) ([]byte, bool) { if node == nil || node.FilePath == "" { return nil, false } rel := node.FilePath if node.RepoPrefix != "" { meta := mi.GetMetadata(node.RepoPrefix) if meta == nil || meta.RootPath == "" { return nil, false } rel = strings.TrimPrefix(rel, node.RepoPrefix+"/") data, err := os.ReadFile(filepath.Join(meta.RootPath, rel)) if err != nil { return nil, false } return data, true } for _, m := range mi.AllMetadata() { data, err := os.ReadFile(filepath.Join(m.RootPath, rel)) if err == nil { return data, true } } return nil, false } // wrapperSourceReader returns a SourceReader closure that maps a graph // node back to its on-disk bytes by joining the node's repo-relative // FilePath with the repo's RootPath from MultiIndexer metadata. In // single-repo mode (no RepoPrefix on nodes) the indexer's sole root is // used. Read results are memoized inside the closure so multi-caller // wrappers don't trigger N disk reads per file. func (mi *MultiIndexer) wrapperSourceReader() contracts.SourceReader { cache := make(map[string][]byte) miss := make(map[string]struct{}) readFile := func(absPath string) ([]byte, bool) { if b, ok := cache[absPath]; ok { return b, true } if _, skipped := miss[absPath]; skipped { return nil, false } b, err := os.ReadFile(absPath) if err != nil { miss[absPath] = struct{}{} return nil, false } cache[absPath] = b return b, true } return func(n *graph.Node) ([]byte, bool) { if n == nil || n.FilePath == "" { return nil, false } // Multi-repo case: strip "/" from the FilePath and // join with the repo's recorded root. if n.RepoPrefix != "" { meta := mi.GetMetadata(n.RepoPrefix) if meta == nil || meta.RootPath == "" { return nil, false } rel := strings.TrimPrefix(n.FilePath, n.RepoPrefix+"/") return readFile(filepath.Join(meta.RootPath, rel)) } // Single-repo fallback: a node without a RepoPrefix carries its // path relative to the sole indexer root. Try each known repo // root since the single-repo path wraps through indexSingleRepo. for _, meta := range mi.AllMetadata() { if b, ok := readFile(filepath.Join(meta.RootPath, n.FilePath)); ok { return b, true } } // Last resort: treat FilePath as already absolute (tests). return readFile(n.FilePath) } } // ReconcileContractEdges walks the merged contract registry, runs the // consumer↔provider matcher, and writes the results into the graph as // EdgeMatches edges pointing from consumer-contract nodes to their matched // provider-contract nodes. Every call evicts the prior set of EdgeMatches // first so the edges stay in sync with the current contract view; that's // correct after track / untrack / re-index / watcher re-scan. Returns the // number of match edges added so callers can log or test the effect. // // This is what makes get_call_chain traverse service boundaries: without a // persisted contract→contract edge, the matcher's result is only visible // via the `contracts check` tool and traversals stop at each service's // boundary. func (mi *MultiIndexer) ReconcileContractEdges() int { // Serialise the whole pass: the evict-then-mint of EdgeMatches, topic // edges, and the bridge subgraph spans many non-atomic store writes, // and several goroutines call this concurrently (see reconcileMu). mi.reconcileMu.Lock() defer mi.reconcileMu.Unlock() g := mi.Graph() if g == nil { return 0 } // Evict any prior EdgeMatches so the graph reflects only the current // registry. Collect first, remove second — we're mutating the same // out-edge lists we'd be iterating otherwise. type edgeKey struct{ from, to string } var stale []edgeKey var staleTopicProduces, staleTopicConsumes []edgeKey // Collect only the three reconciled edge kinds via the edges_by_kind // index, rather than scanning (and meta-decoding) the whole edge set. for e := range g.EdgesByKind(graph.EdgeMatches) { stale = append(stale, edgeKey{e.From, e.To}) } for e := range g.EdgesByKind(graph.EdgeProducesTopic) { staleTopicProduces = append(staleTopicProduces, edgeKey{e.From, e.To}) } for e := range g.EdgesByKind(graph.EdgeConsumesTopic) { staleTopicConsumes = append(staleTopicConsumes, edgeKey{e.From, e.To}) } for _, k := range stale { g.RemoveEdge(k.from, k.to, graph.EdgeMatches) } // Evict prior topic edges so a renamed topic or removed callsite // doesn't keep its edges alive. The KindTopic node itself is // addressable by ID; when no consumer or producer remains for a // given topic node we orphan-collect it after the new edges are // emitted (see end of this function). for _, k := range staleTopicProduces { g.RemoveEdge(k.from, k.to, graph.EdgeProducesTopic) } for _, k := range staleTopicConsumes { g.RemoveEdge(k.from, k.to, graph.EdgeConsumesTopic) } merged := mi.MergedContractRegistry() if merged == nil { return 0 } // Inline HTTP wrapper callers (T2.4). Codebases that route every // endpoint through a helper like `request(path, ...)` produce one // parametric consumer contract per wrapper at extraction time — // useless for matching. InlineWrappers walks incoming call edges // of each wrapper, re-reads the caller's source, and emits a // specific consumer contract per literal path. The returned // contracts are also persisted into their owning repo's per-repo // registry so subsequent `contracts list`/`check` calls see them // (MergedContractRegistry rebuilds fresh each call and would // otherwise lose them). inlined := contracts.InlineWrappers(merged, g, mi.wrapperSourceReader()) if len(inlined) > 0 { mi.mu.RLock() for _, c := range inlined { if c.RepoPrefix == "" { continue } idx, ok := mi.indexers[c.RepoPrefix] if !ok { continue } cr := idx.ContractRegistry() if cr == nil { continue } // Skip if the same contract is already persisted — // ReconcileContractEdges runs on every repo change, and // appending the same inlined contract on every pass would // blow up the registry with duplicates. Compare on the // Registry.All() dedupe key. alreadyPersisted := false for _, existing := range cr.ByID(c.ID) { if existing.SymbolID == c.SymbolID && existing.FilePath == c.FilePath && existing.Role == c.Role { alreadyPersisted = true break } } if !alreadyPersisted { cr.Add(c) } } mi.mu.RUnlock() // Wrapper-inlined contracts arrive AFTER commitContracts ran // its UpgradeBareTypeRefs pass, so their response_type / // request_type still carries bare names like "ToolInfo". // Re-run the upgrade against the merged graph so downstream // snapshotContractShapes finds the type node and the // dashboard sees fields instead of a string. mi.mu.RLock() lookup := func(name, repoHint string) []string { matches := mi.graph.FindNodesByName(name) if len(matches) == 0 { return nil } ids := make([]string, 0, len(matches)) for _, n := range matches { if n.Kind != graph.KindType && n.Kind != graph.KindInterface { continue } ids = append(ids, n.ID) } // Prefer same-repo when multiple match. if len(ids) > 1 && repoHint != "" { var sameRepo []string for _, id := range ids { if strings.HasPrefix(id, repoHint+"/") { sameRepo = append(sameRepo, id) } } if len(sameRepo) > 0 { return sameRepo } } return ids } for _, idx := range mi.indexers { cr := idx.ContractRegistry() if cr != nil { cr.UpgradeBareTypeRefs(lookup) } } // UpgradeBareTypeRefs leaves names with ≥2 candidates alone // (e.g. a TS app declaring `DashboardSnapshot` in both // `lib/schema.ts` and `lib/types.ts`). disambiguateBareTypesViaImports // re-reads the consumer's source, parses its `import` lines, // and picks the candidate whose graph FilePath matches an // imported module. Runs before attachInlinedShapes so the // shape attachment sees fully-qualified IDs. for _, idx := range mi.indexers { cr := idx.ContractRegistry() if cr != nil { mi.disambiguateBareTypesViaImports(cr, mi.graph) } } // Now that response_type / request_type point at real graph // nodes, fold each referenced type's shape (struct fields) // into the contract's Meta so the dashboard renders the // expanded field list instead of just the type name. Mirrors // what snapshotContractShapes + inlineEnvelopeShapes do for // initially-extracted contracts. for _, idx := range mi.indexers { cr := idx.ContractRegistry() if cr == nil { continue } mi.attachInlinedShapes(cr, mi.graph) } mi.mu.RUnlock() } // Bind provider-contract SymbolIDs that came from spec files // (.proto for gRPC, OpenAPI YAML/JSON for HTTP). Without this // step the matcher finds pairs but the bridge-emission check // below skips them because provider SymbolID is empty. Must run // before Match so Match sees the updated records. contracts.BindProviderSymbols(merged, g) result := contracts.Match(merged) added := 0 // Track which topic nodes we've already materialised so the loop // emits one node per (broker, topic) bucket even when a topic has // fan-out across many consumers. The dedupe key is the topic // node's ID — its repo-prefix is already encoded in Contract.ID. topicNodes := make(map[string]struct{}) for _, m := range result.Matched { // Connect the consumer's enclosing symbol directly to the // provider's enclosing symbol. Contract nodes in the graph are // deduped by Contract.ID, so a provider and a consumer that share // the same ID collapse into one node — a contract→contract edge // would be a self-loop. Symbol→symbol bypasses that and gives // get_call_chain the traversal it needs: saveTuck (extension) → // Handler.CreateTuck (core-api). if m.Provider.SymbolID == "" || m.Consumer.SymbolID == "" { continue } if m.Provider.SymbolID == m.Consumer.SymbolID { continue } g.AddEdge(&graph.Edge{ From: m.Consumer.SymbolID, To: m.Provider.SymbolID, Kind: graph.EdgeMatches, FilePath: m.Consumer.FilePath, Line: m.Consumer.Line, Confidence: 1.0, ConfidenceLabel: "EXTRACTED", Origin: graph.OriginASTResolved, CrossRepo: m.CrossRepo, }) added++ // Materialise the broker topic node + producer/consumer // edges. Only fires for ContractTopic matches; HTTP / gRPC / // WS / GraphQL / env contracts fall through with just the // EdgeMatches edge above. if m.Provider.Type == contracts.ContractTopic { emitTopicEdges(g, m, topicNodes) } } // Persist the matched contract groups as the bridge subgraph: one // KindContractBridge node per group plus EdgeBridges fan-out to // the participating contract nodes. The pass evicts the previous // bridge generation internally, so it stays idempotent across // reconciles and drops bridges whose contracts disappeared. MaterializeContractBridges(g, result.Matched) // Topic nodes whose producer and consumer edges all evaporated // since the previous reconcile remain in the graph as leaf // nodes — Graph has no public RemoveNode and the next reconcile // upserts them anyway. A topic that lost every callsite is an // invisible-but-harmless cost (a single KindTopic node with no // neighbors). Worth revisiting if topic churn ever bloats the // graph; in the meantime we lean on EvictRepo / EvictFile to // reclaim memory when a whole repo's topic vocabulary changes. return added } // emitTopicEdges materialises the KindTopic node and the // EdgeProducesTopic / EdgeConsumesTopic edges that pair a matched // producer/consumer pair across the workspace. The topic ID is // reconstructed from the Contract.ID (already `topic:::: // `) so the node ID matches the contract ID 1:1 — agents that // have the contract ID can also look up the topic node directly. // Meta on the node carries the broker family and the raw topic name // for filterless queries. func emitTopicEdges(g graph.Store, m contracts.CrossLink, topicNodes map[string]struct{}) { // Trust the matcher to bucket only same-broker contracts together // because Contract.ID already includes the broker token; if the // broker isn't on the provider Meta, fall through to the contract // ID's middle segment so the node carries something useful. broker, _ := m.Provider.Meta["broker"].(string) topicName, _ := m.Provider.Meta["topic"].(string) if broker == "" || topicName == "" { // Defensive fallback — extract from the Contract.ID shape // `topic::::`. If the ID isn't structured we // skip rather than emit a node with an unidentifiable broker // (such an edge would be misleading in cross-broker queries). broker2, name2, ok := parseTopicContractID(m.Provider.ID) if !ok { return } if broker == "" { broker = broker2 } if topicName == "" { topicName = name2 } } topicID := m.Provider.ID // canonical: topic:::: if _, ok := topicNodes[topicID]; !ok { topicNodes[topicID] = struct{}{} // Preserve any existing node (a prior reconcile may have // created it) but always refresh Meta so a broker rename // isn't sticky across reconciles. AddNode in this codebase // is upsert-style — see graph.Graph.AddNode. g.AddNode(&graph.Node{ ID: topicID, Kind: graph.KindTopic, Name: topicName, FilePath: m.Provider.FilePath, Language: "contract", RepoPrefix: m.Provider.RepoPrefix, WorkspaceID: m.Provider.EffectiveWorkspace(), ProjectID: m.Provider.EffectiveProject(), Meta: map[string]any{ "broker": broker, "name": topicName, }, }) } g.AddEdge(&graph.Edge{ From: m.Provider.SymbolID, To: topicID, Kind: graph.EdgeProducesTopic, FilePath: m.Provider.FilePath, Line: m.Provider.Line, Confidence: 1.0, ConfidenceLabel: "EXTRACTED", Origin: graph.OriginASTResolved, CrossRepo: false, Meta: map[string]any{ "broker": broker, }, }) g.AddEdge(&graph.Edge{ From: m.Consumer.SymbolID, To: topicID, Kind: graph.EdgeConsumesTopic, FilePath: m.Consumer.FilePath, Line: m.Consumer.Line, Confidence: 1.0, ConfidenceLabel: "EXTRACTED", Origin: graph.OriginASTResolved, CrossRepo: m.CrossRepo, Meta: map[string]any{ "broker": broker, }, }) } // parseTopicContractID splits a Contract.ID of the form // `topic::::` (or its repo-prefixed counterpart // `/topic::::`) into broker + name. Returns // ok==false for any other shape so callers can skip ill-formed // topic contracts rather than fabricate a broker label. func parseTopicContractID(id string) (broker, name string, ok bool) { // Strip any leading repo-prefix segment ("repo/topic::..."). The // applyRepoPrefix step prepends `/` to synthetic IDs of // the form `topic::...`, so we look for the inner `topic::` // marker rather than splitting on the leading slash. idx := strings.Index(id, "topic::") if idx < 0 { return "", "", false } rest := id[idx+len("topic::"):] sep := strings.Index(rest, "::") if sep <= 0 || sep == len(rest)-2 { return "", "", false } return rest[:sep], rest[sep+2:], true } // Graph returns the underlying shared graph. func (mi *MultiIndexer) Graph() graph.Store { return mi.graph } // SetRemoteStitch enables cross-daemon proxy-edge minting on every // CrossRepoResolver this MultiIndexer constructs. Called once by the // daemon entry point when federation.edges.enabled; a nil prober leaves // the resolvers on read-only federation. func (mi *MultiIndexer) SetRemoteStitch(prober resolver.RemoteDeclarationProber, budget int) { mi.mu.Lock() defer mi.mu.Unlock() mi.stitchProber = prober mi.proxyBudget = budget } // applyRemoteStitch wires the proxy-edge mint into a freshly built resolver // when a prober is installed. func (mi *MultiIndexer) applyRemoteStitch(cr *resolver.CrossRepoResolver) { mi.mu.RLock() prober, budget := mi.stitchProber, mi.proxyBudget mi.mu.RUnlock() if prober != nil { cr.EnableRemoteStitch(prober, budget) } } // Search returns the shared search backend. func (mi *MultiIndexer) Search() search.Backend { return mi.search } // ExportVectorIndex serializes the workspace-global semantic-search // vector index — there is one shared HNSW index across every tracked // repo, not one per repo. Returns nil, 0, 0 when no vector index is // active (embeddings disabled, or the backend is still text-only). // Used by the daemon snapshot path so a default-on daemon does not // re-embed the whole graph on every restart. func (mi *MultiIndexer) ExportVectorIndex() ([]byte, int, int) { sw, ok := mi.search.(*search.Swappable) if !ok { return nil, 0, 0 } hybrid, ok := sw.Inner().(*search.HybridBackend) if !ok { return nil, 0, 0 } vec := hybrid.VectorIndex() if vec == nil || vec.Count() == 0 { return nil, 0, 0 } var buf bytes.Buffer if err := vec.Save(&buf); err != nil { mi.logger.Warn("failed to export vector index", zap.Error(err)) return nil, 0, 0 } return buf.Bytes(), vec.Dims(), vec.Count() } // ImportVectorIndex restores a previously-exported vector index into // the shared search backend, wrapping the current text backend in a // HybridBackend. It is a no-op when embeddings are disabled (no // configured embedder) or when the cached index's dimensionality does // not match the active embedder — a provider switch (GloVe 50d → ONNX // 384d) makes the cached vectors meaningless, so the indexer re-embeds // instead. Returns an error only on a structurally corrupt index blob. func (mi *MultiIndexer) ImportVectorIndex(data []byte, dims, count int) error { if len(data) == 0 || mi.embedder == nil { return nil } if embedderDims := mi.embedder.Dimensions(); embedderDims > 0 && embedderDims != dims { mi.logger.Info("vector index dims mismatch, will re-embed", zap.Int("cached_dims", dims), zap.Int("embedder_dims", embedderDims)) return nil } sw, ok := mi.search.(*search.Swappable) if !ok { return nil } vec := search.NewVector(dims) if err := vec.LoadFrom(bytes.NewReader(data)); err != nil { return fmt.Errorf("import vector index: %w", err) } vec.SetCount(count) // Unwrap an existing HybridBackend to its text side before // re-wrapping so we never nest Hybrids (each retains a stale // vector index — see buildSearchIndex for the memory rationale). inner := sw.Inner() if hyb, ok := inner.(*search.HybridBackend); ok { inner = hyb.TextBackend() } sw.Swap(search.NewHybrid(inner, vec, mi.embedder)) mi.logger.Info("restored vector index from snapshot", zap.Int("vectors", count), zap.Int("dims", dims)) return nil } // AutoDetectRepos walks immediate subdirectories of parentPath looking for // .git directories. If parentPath itself is a Git repo, it returns a single // entry (the caller should index it as single-repo). If zero Git repos are // found, it returns nil so the caller can fall back to single-repo mode. // This is gated by the workspace.auto_detect config flag. func (mi *MultiIndexer) AutoDetectRepos(parentPath string) []config.RepoEntry { absPath, err := filepath.Abs(parentPath) if err != nil { mi.logger.Warn("auto-detect: failed to resolve path", zap.String("path", parentPath), zap.Error(err)) return nil } // If the path itself is a Git repo, return it as a single repo. if isGitRepo(absPath) { return []config.RepoEntry{{ Path: absPath, Name: filepath.Base(absPath), }} } // Walk immediate subdirectories (not recursive) for .git dirs. entries, err := os.ReadDir(absPath) if err != nil { mi.logger.Warn("auto-detect: failed to read directory", zap.String("path", absPath), zap.Error(err)) return nil } var repos []config.RepoEntry for _, entry := range entries { if !entry.IsDir() { continue } subDir := filepath.Join(absPath, entry.Name()) if isGitRepo(subDir) { repos = append(repos, config.RepoEntry{ Path: subDir, Name: entry.Name(), // Derive RepoPrefix from subdirectory name. }) } } // If zero Git repos found, return nil — caller falls back to single-repo. if len(repos) == 0 { return nil } return repos } // isGitRepo checks whether the given directory contains a .git subdirectory. func isGitRepo(dir string) bool { info, err := os.Stat(filepath.Join(dir, ".git")) if err != nil { return false } return info.IsDir() }