diff --git a/internal/lore/appraise_rrf.go b/internal/lore/appraise_rrf.go index 10583cc..37db8c6 100644 --- a/internal/lore/appraise_rrf.go +++ b/internal/lore/appraise_rrf.go @@ -8,11 +8,31 @@ import ( "log/slog" "strconv" "strings" + "sync/atomic" "time" "github.com/mathomhaus/guild/internal/lore/embed" ) +// coverageGateState caches live vector coverage counts keyed by +// meta.vector_epoch. Appraise's RRF gate reads coverage on every query; +// the epoch check is a single PK lookup, and the two COUNT(*) queries +// run only when the corpus has changed since the last appraise. +type coverageGateState struct { + epoch int64 + num int64 + den int64 +} + +// coverageGate is process-local; two servers reading the same epoch +// make the same decision, so cross-process coherence is automatic for +// a given epoch value. +var coverageGate atomic.Pointer[coverageGateState] + +// liveCoverageQueryCount increments on every cache miss that runs the +// live COUNT(*) pair. Tests assert the hot path does not re-query. +var liveCoverageQueryCount atomic.Int64 + // appraiseRRF is the single-project RRF-fused retrieval path. It is // called only from Appraise, only when params.Embed is Enabled and // params.AllProjects is false. Returns handled=false when the coverage @@ -265,12 +285,12 @@ func appraiseCrossProject( return out, true, nil } -// readCoverage returns (embedder_state, coverage_ratio, err). A zero -// denominator yields coverage=1.0 (the "no active entries" state is -// treated as "nothing to embed, so coverage is fully satisfied"). This -// avoids a divide-by-zero that would otherwise mask a freshly -// initialized DB as "below threshold" and hide it behind BM25 for no -// good reason. +// readCoverage returns (embedder_state, coverage_ratio, err). Coverage +// is computed from live COUNT(*) queries (lore_vectors rows vs active +// entries) cached on meta.vector_epoch so stale meta.vector_coverage_* +// counters cannot silently disable the semantic arm. A zero denominator +// yields coverage=1.0 (the "no active entries" state is treated as +// "nothing to embed, so coverage is fully satisfied"). func readCoverage(ctx context.Context, db *sql.DB) (state string, coverage float64, err error) { scanErr := db.QueryRowContext(ctx, `SELECT value FROM meta WHERE key = 'embedder_state'`, @@ -282,11 +302,7 @@ func readCoverage(ctx context.Context, db *sql.DB) (state string, coverage float return "", 0, fmt.Errorf("lore: appraise: read embedder_state: %w", scanErr) } - num, err := readMetaInt(ctx, db, "vector_coverage_num") - if err != nil { - return "", 0, err - } - den, err := readMetaInt(ctx, db, "vector_coverage_den") + num, den, err := liveCoverageCounts(ctx, db) if err != nil { return "", 0, err } @@ -296,6 +312,41 @@ func readCoverage(ctx context.Context, db *sql.DB) (state string, coverage float return state, float64(num) / float64(den), nil } +// liveCoverageCounts returns (num, den) from live SQL, using an +// epoch-keyed process-local cache. Predicates mirror assessCorpus in +// internal/mcp/embed_autobackfill.go and ReconcileDen's ActivePredicate. +// +// Invalidation is keyed on meta.vector_epoch, which bumps on every +// successful vector write and at backfill cycle end. Denominator-only +// changes from inscribe/seal do not bump epoch, so the cache may be +// briefly stale until the next vector write. That window is bounded and +// far smaller than the permanent meta-counter drift this gate fixes. +func liveCoverageCounts(ctx context.Context, db *sql.DB) (num, den int64, err error) { + epoch, err := readMetaInt(ctx, db, "vector_epoch") + if err != nil { + return 0, 0, err + } + if cached := coverageGate.Load(); cached != nil && cached.epoch == epoch { + return cached.num, cached.den, nil + } + + corpus := embed.LoreCorpus{} + if err := db.QueryRowContext(ctx, + fmt.Sprintf(`SELECT COUNT(*) FROM %s`, corpus.VectorTable()), //nolint:gosec // table from compile-time corpus accessor + ).Scan(&num); err != nil { + return 0, 0, fmt.Errorf("lore: appraise: count lore_vectors: %w", err) + } + if err := db.QueryRowContext(ctx, + fmt.Sprintf(`SELECT COUNT(*) FROM %s WHERE %s`, corpus.EntityTable(), corpus.ActivePredicate()), //nolint:gosec // table + predicate from compile-time corpus accessors + ).Scan(&den); err != nil { + return 0, 0, fmt.Errorf("lore: appraise: count active entries: %w", err) + } + + coverageGate.Store(&coverageGateState{epoch: epoch, num: num, den: den}) + liveCoverageQueryCount.Add(1) + return num, den, nil +} + // readMetaInt reads a meta row and parses its decimal value. Missing // rows yield zero (every caller today tolerates that). func readMetaInt(ctx context.Context, db *sql.DB, key string) (int64, error) { diff --git a/internal/lore/appraise_rrf_test.go b/internal/lore/appraise_rrf_test.go new file mode 100644 index 0000000..cf6f7db --- /dev/null +++ b/internal/lore/appraise_rrf_test.go @@ -0,0 +1,163 @@ +package lore + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/mathomhaus/guild/internal/lore/embed" +) + +func resetCoverageGate(t *testing.T) { + t.Helper() + coverageGate.Store(nil) + liveCoverageQueryCount.Store(0) +} + +func upsertMeta(t *testing.T, db *sql.DB, key, value string) { + t.Helper() + _, err := db.ExecContext(context.Background(), + `INSERT INTO meta (key, value) VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + key, value, + ) + if err != nil { + t.Fatalf("upsert meta %s=%s: %v", key, value, err) + } +} + +func seedVectorsForEntries(t *testing.T, db *sql.DB, entryIDs []int64) { + t.Helper() + now := time.Now().Unix() + for _, id := range entryIDs { + _, err := db.ExecContext(context.Background(), + `INSERT OR REPLACE INTO lore_vectors + (entry_id, model_id, dim, vec, encoded_at, content_hash) + VALUES (?, 'test-model', 4, X'00000000', ?, 'testhash')`, + id, now, + ) + if err != nil { + t.Fatalf("insert lore_vectors for entry %d: %v", id, err) + } + } +} + +// TestReadCoverage_LiveSQLIgnoresStaleMeta verifies that appraise's +// coverage gate uses live COUNT(*) rather than drifted meta counters. +func TestReadCoverage_LiveSQLIgnoresStaleMeta(t *testing.T) { + ctx := context.Background() + db := openTestDB(t, "p") + defer func() { _ = db.Close() }() + resetCoverageGate(t) + + ids := seedCorpus(t, ctx, db, []fixtureEntry{ + {"p", "research", "alpha vector topic", "summary alpha", "t"}, + {"p", "research", "beta vector topic", "summary beta", "t"}, + {"p", "research", "gamma vector topic", "summary gamma", "t"}, + {"p", "research", "delta vector topic", "summary delta", "t"}, + {"p", "research", "epsilon vector topic", "summary epsilon", "t"}, + }) + seedVectorsForEntries(t, db, ids) + + // Simulate the drift scenario from issue #76: vectors are healthy + // but meta counters report 61% coverage. + upsertMeta(t, db, "embedder_state", "enabled") + upsertMeta(t, db, "vector_coverage_num", "1") + upsertMeta(t, db, "vector_coverage_den", "10") + upsertMeta(t, db, "vector_epoch", "7") + + state, cov, err := readCoverage(ctx, db) + if err != nil { + t.Fatalf("readCoverage: %v", err) + } + if state != "enabled" { + t.Fatalf("state = %q, want enabled", state) + } + if cov < CoverageThreshold { + t.Fatalf("coverage = %v, want >= %v (live 5/5 should clear gate)", cov, CoverageThreshold) + } +} + +// TestReadCoverage_EpochCacheAvoidsRepeatedCounts verifies that +// consecutive appraise reads at the same epoch do not re-run COUNT(*). +func TestReadCoverage_EpochCacheAvoidsRepeatedCounts(t *testing.T) { + ctx := context.Background() + db := openTestDB(t, "p") + defer func() { _ = db.Close() }() + resetCoverageGate(t) + + ids := seedCorpus(t, ctx, db, []fixtureEntry{ + {"p", "research", "cache test entry one", "summary one", "t"}, + {"p", "research", "cache test entry two", "summary two", "t"}, + }) + seedVectorsForEntries(t, db, ids) + upsertMeta(t, db, "embedder_state", "enabled") + upsertMeta(t, db, "vector_epoch", "3") + + if _, _, err := readCoverage(ctx, db); err != nil { + t.Fatalf("first readCoverage: %v", err) + } + if got := liveCoverageQueryCount.Load(); got != 1 { + t.Fatalf("after first read: liveCoverageQueryCount = %d, want 1", got) + } + + if _, _, err := readCoverage(ctx, db); err != nil { + t.Fatalf("second readCoverage: %v", err) + } + if got := liveCoverageQueryCount.Load(); got != 1 { + t.Fatalf("after second read: liveCoverageQueryCount = %d, want 1 (cache hit)", got) + } + + // Bump epoch: cache miss should run COUNT(*) again. + upsertMeta(t, db, "vector_epoch", "4") + if _, _, err := readCoverage(ctx, db); err != nil { + t.Fatalf("third readCoverage: %v", err) + } + if got := liveCoverageQueryCount.Load(); got != 2 { + t.Fatalf("after epoch bump: liveCoverageQueryCount = %d, want 2", got) + } +} + +// TestAppraiseRRF_EngagesDespiteStaleMeta verifies the RRF path stays +// active when live vector coverage clears the gate even though meta +// counters are stale. +func TestAppraiseRRF_EngagesDespiteStaleMeta(t *testing.T) { + ctx := context.Background() + db := openTestDB(t, "p") + defer func() { _ = db.Close() }() + resetCoverageGate(t) + + ids := seedCorpus(t, ctx, db, []fixtureEntry{ + {"p", "research", "semantic arm gate sentinel alpha", "summary", "t"}, + {"p", "research", "semantic arm gate sentinel beta", "summary", "t"}, + }) + seedVectorsForEntries(t, db, ids) + upsertMeta(t, db, "embedder_state", "enabled") + upsertMeta(t, db, "vector_coverage_num", "1") + upsertMeta(t, db, "vector_coverage_den", "100") + upsertMeta(t, db, "vector_epoch", "1") + + embedDeps := &EmbedDeps{ + Embedder: embed.NewDeterministicEmbedder(), + ModelID: "test-model", + } + + params := &AppraiseParams{ + Query: "semantic arm gate sentinel", + Project: "p", + Now: time.Now().UTC(), + Embed: embedDeps, + } + + out, handled, err := appraiseRRF(ctx, db, params, params.Now, DefaultScoring(), 5) + if err != nil { + t.Fatalf("appraiseRRF: %v", err) + } + if !handled { + t.Fatal("appraiseRRF: handled=false, want true when live coverage clears gate") + } + if out == nil { + t.Fatal("appraiseRRF: nil output") + } +} diff --git a/internal/lore/coverage_reconcile_cmd.go b/internal/lore/coverage_reconcile_cmd.go index 4e79570..d4020e2 100644 --- a/internal/lore/coverage_reconcile_cmd.go +++ b/internal/lore/coverage_reconcile_cmd.go @@ -14,7 +14,7 @@ type CoverageReconcileInput struct { Project string `json:"project,omitempty"` } -// CoverageReconcileOutput reports the before/after state of vector_coverage_den. +// CoverageReconcileOutput reports the before/after state of vector coverage counters. type CoverageReconcileOutput struct { ProjectID string `json:"project_id"` // DenBefore is the meta.vector_coverage_den value before the reconcile. @@ -22,16 +22,19 @@ type CoverageReconcileOutput struct { // DenAfter is the meta.vector_coverage_den value after the reconcile // (equals the live COUNT(*) of active entries). DenAfter int64 `json:"den_after"` + // NumBefore is the meta.vector_coverage_num value before the reconcile. + NumBefore int64 `json:"num_before"` // NumAfter is the meta.vector_coverage_num value after the reconcile - // (unchanged by this operation; reported for convenience). + // (equals the live COUNT(*) of lore_vectors rows). NumAfter int64 `json:"num_after"` // Drift is DenAfter - DenBefore (positive = den was too low, negative = too high). Drift int64 `json:"drift"` } // CoverageReconcileCommand is the registry spec for `guild lore coverage-reconcile`. -// It resets meta.vector_coverage_den to the live COUNT(*) of active entries -// and reports the before/after values so operators can verify the fix. +// It resets meta.vector_coverage_den and meta.vector_coverage_num to their +// live COUNT(*) values and reports the before/after state so operators can +// verify the fix. // // This is the manual escape hatch for QUEST-220 / LORE-373. Backfill also // calls ReconcileDen automatically, so normal usage should never require this @@ -40,11 +43,12 @@ var CoverageReconcileCommand = &command.Command[CoverageReconcileInput, Coverage Name: "lore_coverage_reconcile", CLIPath: []string{"lore", "coverage-reconcile"}, CLIAliases: []string{"fix-coverage"}, - Short: "reset vector_coverage_den to the live active-entry count", - Long: "Reset meta.vector_coverage_den to the live COUNT(*) WHERE status NOT IN " + - "('archived','parked') and report before/after values. " + - "Corrects num > den drift that produces coverage > 100%. " + - "Backfill also runs this automatically, so this command is a manual escape hatch.", + Short: "reset vector coverage counters to live counts", + Long: "Reset meta.vector_coverage_den and meta.vector_coverage_num to live " + + "COUNT(*) values (active entries and lore_vectors rows) and report " + + "before/after values. Corrects num/den drift that can disable the " + + "appraise semantic arm or produce coverage > 100%. Backfill also runs " + + "den reconcile automatically, so this command is a manual escape hatch.", Args: []command.ArgSpec{ {Name: "project", Short: "p", Kind: command.ArgFlag, Type: command.ArgString, Help: "project override"}, }, @@ -65,10 +69,14 @@ var CoverageReconcileCommand = &command.Command[CoverageReconcileInput, Coverage return CoverageReconcileOutput{}, fmt.Errorf("lore: coverage-reconcile: read before state: %w", err) } denBefore := before.CoverageDen + numBefore := before.CoverageNum - // Run the reconcile. + // Run the reconcile for both counters. + if err := embed.ReconcileNum(ctx, db, embed.LoreCorpus{}); err != nil { + return CoverageReconcileOutput{}, fmt.Errorf("lore: coverage-reconcile: reconcile num: %w", err) + } if err := embed.ReconcileDen(ctx, db, embed.LoreCorpus{}); err != nil { - return CoverageReconcileOutput{}, fmt.Errorf("lore: coverage-reconcile: %w", err) + return CoverageReconcileOutput{}, fmt.Errorf("lore: coverage-reconcile: reconcile den: %w", err) } // Read after state. @@ -81,6 +89,7 @@ var CoverageReconcileCommand = &command.Command[CoverageReconcileInput, Coverage ProjectID: pid, DenBefore: denBefore, DenAfter: after.CoverageDen, + NumBefore: numBefore, NumAfter: after.CoverageNum, Drift: after.CoverageDen - denBefore, }, nil @@ -94,17 +103,24 @@ func formatCoverageReconcile(s lineSink, o CoverageReconcileOutput) string { b.WriteString(s.Line("🔮", "[coverage-reconcile]", fmt.Sprintf("project=%s", o.ProjectID))) b.WriteString(fmt.Sprintf(" den_before: %d\n", o.DenBefore)) b.WriteString(fmt.Sprintf(" den_after: %d\n", o.DenAfter)) + b.WriteString(fmt.Sprintf(" num_before: %d\n", o.NumBefore)) b.WriteString(fmt.Sprintf(" num_after: %d\n", o.NumAfter)) drift := o.Drift sign := "+" if drift < 0 { sign = "" } - b.WriteString(fmt.Sprintf(" drift: %s%d\n", sign, drift)) - if drift == 0 { - b.WriteString(" status: den was already correct\n") + b.WriteString(fmt.Sprintf(" drift (den): %s%d\n", sign, drift)) + numDrift := o.NumAfter - o.NumBefore + numSign := "+" + if numDrift < 0 { + numSign = "" + } + b.WriteString(fmt.Sprintf(" drift (num): %s%d\n", numSign, numDrift)) + if drift == 0 && numDrift == 0 { + b.WriteString(" status: counters were already correct\n") } else { - b.WriteString(fmt.Sprintf(" status: den corrected by %s%d\n", sign, drift)) + b.WriteString(" status: counters reconciled to live counts\n") } return strings.TrimRight(b.String(), "\n") } diff --git a/internal/lore/embed/backfill.go b/internal/lore/embed/backfill.go index 9e1eb5c..9f2cc93 100644 --- a/internal/lore/embed/backfill.go +++ b/internal/lore/embed/backfill.go @@ -190,6 +190,50 @@ func ReconcileDen(ctx context.Context, db *sql.DB, corpus VectorCorpus) error { return nil } +// ReconcileNum resets the corpus's vector_coverage_num meta row to the +// live COUNT(*) of vector rows. Runs inside a single BEGIN IMMEDIATE so +// the write is atomic with respect to concurrent writers. +// +// Pair with ReconcileDen via coverage-reconcile so operators can repair +// both sides of the coverage ratio when meta counters drift from live +// state. LORE-373 / issue #76. +func ReconcileNum(ctx context.Context, db *sql.DB, corpus VectorCorpus) error { + if db == nil { + return fmt.Errorf("embed: ReconcileNum: nil db") + } + if corpus == nil { + corpus = LoreCorpus{} + } + conn, rollback, err := beginImmediateLocal(ctx, db, "reconcile-num") + if err != nil { + return err + } + defer conn.Close() + committed := false + defer rollback(&committed) + + countQuery := fmt.Sprintf(`SELECT COUNT(*) FROM %s`, corpus.VectorTable()) + var count int64 + if err := conn.QueryRowContext(ctx, countQuery).Scan(&count); err != nil { //nolint:sqlcheck // table name is a compile-time corpus accessor. + return fmt.Errorf("embed: ReconcileNum: count vectors: %w", err) + } + + numKey := corpus.MetaKey(FieldVectorCoverageNum) + if _, err := conn.ExecContext(ctx, + `INSERT INTO meta (key, value) VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + numKey, strconv.FormatInt(count, 10), + ); err != nil { + return fmt.Errorf("embed: ReconcileNum: upsert %s: %w", numKey, err) + } + + if _, err := conn.ExecContext(ctx, "COMMIT"); err != nil { + return fmt.Errorf("embed: ReconcileNum: commit: %w", err) + } + committed = true + return nil +} + // backfillFailureLogCap is the maximum number of per-iteration WARN // lines Backfill emits before suppressing the rest with a single // "[N more failures suppressed]" summary. 10 is enough to characterize diff --git a/internal/lore/embed/backfill_test.go b/internal/lore/embed/backfill_test.go index fc23f3e..1225668 100644 --- a/internal/lore/embed/backfill_test.go +++ b/internal/lore/embed/backfill_test.go @@ -307,6 +307,37 @@ func TestReconcileDen_FixesDrift(t *testing.T) { } } +// TestReconcileNum_FixesDrift verifies that ReconcileNum corrects a stale +// vector_coverage_num by setting it to the live COUNT(*) of lore_vectors. +func TestReconcileNum_FixesDrift(t *testing.T) { + db := newTestDB(t) + ids := seedEntries(t, db, 3) + ctx := context.Background() + + // Insert vectors for two of the three entries. + for _, id := range ids[:2] { + if _, err := db.ExecContext(ctx, + `INSERT INTO lore_vectors (entry_id, model_id, dim, vec, encoded_at, content_hash) + VALUES (?, 'test', 4, X'00000000', 1, 'h')`, id, + ); err != nil { + t.Fatalf("insert vector: %v", err) + } + } + + setMetaInt(t, db, "vector_coverage_num", 0) + if got := readMetaInt(t, db, "vector_coverage_num"); got != 0 { + t.Fatalf("precondition: num=%d want 0", got) + } + + if err := ReconcileNum(context.Background(), db, LoreCorpus{}); err != nil { + t.Fatalf("ReconcileNum: %v", err) + } + + if got := readMetaInt(t, db, "vector_coverage_num"); got != 2 { + t.Errorf("num after reconcile: got %d want 2", got) + } +} + // TestBackfill_ReconcilesDenBeforeWriting verifies that Backfill fixes a // stale den before writing vectors, so num <= den holds after the run. func TestBackfill_ReconcilesDenBeforeWriting(t *testing.T) {