From d19b579e9d438c05b1bc3c5b28aca81b25a4f890 Mon Sep 17 00:00:00 2001 From: Mark Baker Date: Wed, 17 Jun 2026 00:00:00 -0400 Subject: [PATCH] fix(federation): normalize vector clocks and tighten promotion scoring --- internal/consolidation/heuristic.go | 25 +++++--- internal/consolidation/heuristic_test.go | 18 ++++-- internal/cortex/cortex.go | 28 ++++++++- internal/cortex/cortex_test.go | 78 ++++++++++++++++++++++++ internal/federation/vclock.go | 21 +++++++ internal/federation/vclock_test.go | 19 ++++++ 6 files changed, 173 insertions(+), 16 deletions(-) diff --git a/internal/consolidation/heuristic.go b/internal/consolidation/heuristic.go index e145b0a..760adc4 100644 --- a/internal/consolidation/heuristic.go +++ b/internal/consolidation/heuristic.go @@ -130,13 +130,12 @@ const MinLineageSourcesForCredit = 2 // Exported for tests; package-private callers don't need it because // HeuristicPass is the single consumer. // -// search_hit_count is folded into the reads bucket at the same weight -// because both signals describe passive/weak consumption — the read -// either came from a deliberate get_trace or from being one of the -// top-N hits an agent's search returned. Auto-injection providers like -// Hermes only ever generate search hits, and without this fold-in -// short-tier traces in those cortexes would never accumulate enough -// signal to promote. +// search_hit_count is normally folded into the reads bucket at the same +// weight because both signals describe consumption — the read either came +// from a deliberate get_trace or from being one of the top-N hits an agent's +// search returned. Auto-injection providers like Hermes only ever generate +// search hits, and without this fold-in short-tier traces in those cortexes +// would never accumulate enough signal to promote. // // Lineage credit is gated by MinLineageSourcesForCredit: a trace with // only one derived_from gets zero lineage points, regardless of the @@ -145,12 +144,22 @@ const MinLineageSourcesForCredit = 2 // single session trace it was extracted from) from gliding past the // promotion threshold purely on the lineage head start. Multi-source // traces still earn the full weight per source. +// +// One-source traces also do not get passive search-hit credit unless they +// have stronger intent (an edit or an explicit tier vote). This keeps +// auto-retrieved Hermes session summaries from leaking into mid-tier just +// because they appeared in search results, while still allowing deliberate +// reads, edits, votes, and real multi-source distillations to promote. func scoreCandidate(pc cortex.PromotionCandidate, cfg PassConfig) int { lineageCredit := 0 if pc.DerivedFromCount >= MinLineageSourcesForCredit { lineageCredit = pc.DerivedFromCount * cfg.WeightLineage } - return (pc.ReadCount+pc.SearchHitCount)*cfg.WeightReads + + searchHitCredit := pc.SearchHitCount + if pc.DerivedFromCount == 1 && pc.ModifyCount == 0 && pc.TierVotes == 0 { + searchHitCredit = 0 + } + return (pc.ReadCount+searchHitCredit)*cfg.WeightReads + pc.ModifyCount*cfg.WeightModifies + lineageCredit + pc.TierVotes*cfg.WeightVotes diff --git a/internal/consolidation/heuristic_test.go b/internal/consolidation/heuristic_test.go index 1d017a9..fcc200c 100644 --- a/internal/consolidation/heuristic_test.go +++ b/internal/consolidation/heuristic_test.go @@ -51,6 +51,10 @@ func TestScoreCandidate_BlendedFormula(t *testing.T) { // Lineage credit is gated at >= 2 sources. A single derived_from // is provenance, not consolidation — see MinLineageSourcesForCredit. {"1 lineage ref earns no credit", cortex.PromotionCandidate{DerivedFromCount: 1}, 0}, + {"1 lineage ref + search hits earns no passive credit", cortex.PromotionCandidate{SearchHitCount: 5, DerivedFromCount: 1}, 0}, + {"1 lineage ref + deliberate reads still counts", cortex.PromotionCandidate{ReadCount: 5, DerivedFromCount: 1}, 5}, + {"1 lineage ref + modify unlocks search-hit credit", cortex.PromotionCandidate{SearchHitCount: 3, ModifyCount: 1, DerivedFromCount: 1}, 5}, + {"1 lineage ref + vote unlocks search-hit credit", cortex.PromotionCandidate{SearchHitCount: 1, TierVotes: 1, DerivedFromCount: 1}, 6}, {"2 lineage refs earn full credit", cortex.PromotionCandidate{DerivedFromCount: 2}, 6}, {"3 lineage refs scale linearly", cortex.PromotionCandidate{DerivedFromCount: 3}, 9}, {"1 vote", cortex.PromotionCandidate{TierVotes: 1}, 5}, @@ -75,10 +79,10 @@ func TestScoreCandidate_BlendedFormula(t *testing.T) { func TestHeuristicPass_PromotesAboveThreshold(t *testing.T) { cx := &fakeHeuristicCortex{ candidates: []cortex.PromotionCandidate{ - {ID: "hot", Tier: trace.TierShort, ReadCount: 10}, // score 10 - promote - {ID: "meh", Tier: trace.TierShort, ReadCount: 2}, // score 2 - skip - {ID: "voted", Tier: trace.TierShort, TierVotes: 1}, // score 5 - promote (at threshold) - {ID: "cold", Tier: trace.TierShort}, // score 0 - skip + {ID: "hot", Tier: trace.TierShort, ReadCount: 10}, // score 10 - promote + {ID: "meh", Tier: trace.TierShort, ReadCount: 2}, // score 2 - skip + {ID: "voted", Tier: trace.TierShort, TierVotes: 1}, // score 5 - promote (at threshold) + {ID: "cold", Tier: trace.TierShort}, // score 0 - skip {ID: "referenced", Tier: trace.TierShort, DerivedFromCount: 2}, // score 6 - promote (real consolidation) // Regression guard: a 1-source provenance link with low // engagement must not glide past the threshold. This is the @@ -87,6 +91,10 @@ func TestHeuristicPass_PromotesAboveThreshold(t *testing.T) { // silently inflating mid-tier before MinLineageSourcesForCredit // landed. Score: 1*1 (read) + 0 (lineage gated) = 1. {ID: "session-summary-shaped", Tier: trace.TierShort, ReadCount: 1, DerivedFromCount: 1}, + // Regression guard for the 1-source mid leak detector: passive + // search hits alone must not promote a Hermes session-summary + // shaped trace. + {ID: "session-summary-search-hit", Tier: trace.TierShort, SearchHitCount: 10, DerivedFromCount: 1}, }, } pass := HeuristicPass(cx, PassConfig{}, nil) @@ -102,7 +110,7 @@ func TestHeuristicPass_PromotesAboveThreshold(t *testing.T) { t.Errorf("expected %q to be promoted, promoted=%v", want, cx.promoted) } } - for _, shouldSkip := range []string{"meh", "cold", "session-summary-shaped"} { + for _, shouldSkip := range []string{"meh", "cold", "session-summary-shaped", "session-summary-search-hit"} { if got[shouldSkip] { t.Errorf("unexpected promotion of %q", shouldSkip) } diff --git a/internal/cortex/cortex.go b/internal/cortex/cortex.go index d981429..2cc43ff 100644 --- a/internal/cortex/cortex.go +++ b/internal/cortex/cortex.go @@ -1311,10 +1311,26 @@ func Open(name, dir string) (*Cortex, error) { fmt.Fprintf(os.Stderr, "[cortex] trace_usage backfill warning: %v\n", err) } } + if err := cx.pruneLegacyVectorClockBuckets(); err != nil { + fmt.Fprintf(os.Stderr, "[cortex] vector-clock cleanup warning: %v\n", err) + } return cx, nil } +func (c *Cortex) pruneLegacyVectorClockBuckets() error { + state := federation.NewState(c.DB.DB) + vc, err := state.GetClock() + if err != nil { + return err + } + clean := federation.KeepCortexIDKeys(vc) + if len(clean) == len(vc) { + return nil + } + return state.SetClock(clean) +} + // backfillTraceUsage populates trace_usage from the legacy per-trace // counter columns, crediting them to the local cortex ID. Idempotent: // skipped when trace_usage already has rows. Safe on fresh cortexes @@ -2953,6 +2969,7 @@ func (c *Cortex) emitEvent(tx *sql.Tx, action event.Action, traceID, timestamp s if err != nil { vc = make(federation.VClock) } + vc = federation.KeepCortexIDKeys(vc) vc.Increment(c.ID) e := &event.Event{ @@ -3257,10 +3274,12 @@ func (c *Cortex) replayUpdate(e event.Event) error { if len(e.VClock) > 0 { localEvents, _ := event.ForTrace(c.DB.DB, e.TraceID) if lastLocal := lastMutationEvent(localEvents); lastLocal != nil && len(lastLocal.VClock) > 0 { - rel := federation.Compare(lastLocal.VClock, e.VClock) + localClock := federation.KeepCortexIDKeys(lastLocal.VClock) + remoteClock := federation.KeepCortexIDKeys(e.VClock) + rel := federation.Compare(localClock, remoteClock) if rel == 0 { // Concurrent — neither clock dominates. Create a divergence trace. - return c.createDivergence(r, e, data.Body, lastLocal.VClock, e.VClock) + return c.createDivergence(r, e, data.Body, localClock, remoteClock) } } } @@ -3862,7 +3881,10 @@ func (c *Cortex) MergeClock(remote federation.VClock) error { if err != nil { return err } - merged, err := federation.MergeCapped(local, remote) + merged, err := federation.MergeCapped( + federation.KeepCortexIDKeys(local), + federation.KeepCortexIDKeys(remote), + ) if err != nil { return err } diff --git a/internal/cortex/cortex_test.go b/internal/cortex/cortex_test.go index 72dabea..7a5fcc0 100644 --- a/internal/cortex/cortex_test.go +++ b/internal/cortex/cortex_test.go @@ -312,6 +312,47 @@ func TestOpen_AllowsPinnedPeerWithSameDisplayName(t *testing.T) { cx2.Close() } +func TestOpen_PrunesLegacyVectorClockNameBuckets(t *testing.T) { + dir := t.TempDir() + if _, err := cortex.Create("mycortex", dir); err != nil { + t.Fatalf("Create: %v", err) + } + root := filepath.Join(dir, "mycortex") + + cx, err := cortex.Open("mycortex", root) + if err != nil { + t.Fatalf("first Open: %v", err) + } + state := federation.NewState(cx.DB.DB) + if err := state.SetClock(federation.VClock{ + cx.ID: 4, + "mycortex": 19, + "legacy-p2": 8, + }); err != nil { + t.Fatalf("SetClock: %v", err) + } + cx.Close() + + cx2, err := cortex.Open("mycortex", root) + if err != nil { + t.Fatalf("second Open: %v", err) + } + defer cx2.Close() + + vc, err := cx2.GetClock() + if err != nil { + t.Fatalf("GetClock: %v", err) + } + if vc[cx2.ID] != 4 { + t.Errorf("local ULID bucket = %d, want 4 in %v", vc[cx2.ID], vc) + } + for _, legacy := range []string{"mycortex", "legacy-p2"} { + if _, ok := vc[legacy]; ok { + t.Errorf("legacy bucket %q survived Open cleanup: %v", legacy, vc) + } + } +} + // TestOpen_RejectsReIdentifiedCopy keeps the real copy detection: when events // THIS cortex authored (origin == its name) are recorded under a cortex_id that // differs from the one cortex.md now declares, the directory was copied or @@ -1789,6 +1830,43 @@ func TestVectorClockIncrements(t *testing.T) { } } +func TestMergeClock_DropsLegacyNameBuckets(t *testing.T) { + cx := setup(t) + state := federation.NewState(cx.DB.DB) + if err := state.SetClock(federation.VClock{ + cx.ID: 4, + "mycortex": 19, + "legacy-p2": 8, + }); err != nil { + t.Fatalf("SetClock: %v", err) + } + + remoteID := "01REMOTE000000000000000000" + if err := cx.MergeClock(federation.VClock{ + remoteID: 7, + "mycortex": 19, + "legacy-p3": 2, + }); err != nil { + t.Fatalf("MergeClock: %v", err) + } + + vc, err := cx.GetClock() + if err != nil { + t.Fatalf("GetClock: %v", err) + } + if vc[cx.ID] != 4 { + t.Errorf("local ULID bucket = %d, want 4 in %v", vc[cx.ID], vc) + } + if vc[remoteID] != 7 { + t.Errorf("remote ULID bucket = %d, want 7 in %v", vc[remoteID], vc) + } + for _, legacy := range []string{"mycortex", "legacy-p2", "legacy-p3"} { + if _, ok := vc[legacy]; ok { + t.Errorf("legacy bucket %q survived merge: %v", legacy, vc) + } + } +} + // ---- Backfill ---- // TestBackfillCreateEvents_SyncedTraceGetsCreate is the headline scenario: diff --git a/internal/federation/vclock.go b/internal/federation/vclock.go index 656bd80..30a6ee9 100644 --- a/internal/federation/vclock.go +++ b/internal/federation/vclock.go @@ -13,6 +13,27 @@ const MaxVClockEntries = 256 // VClock is a vector clock: one counter per known peer. type VClock map[string]uint64 +// CortexIDKey reports whether key is shaped like a stable cortex identity. +// Cortex IDs are ULIDs, represented as 26-character strings. Older Noema +// releases used display names as vector-clock keys; those legacy buckets must +// not influence current federation causality. +func CortexIDKey(key string) bool { + return len(key) == 26 +} + +// KeepCortexIDKeys returns a copy of vc containing only stable cortex-ID +// buckets. It is used on production federation paths to stop pre-migration +// name-keyed buckets from surviving forever in new event snapshots. +func KeepCortexIDKeys(vc VClock) VClock { + clean := make(VClock, len(vc)) + for k, v := range vc { + if CortexIDKey(k) { + clean[k] = v + } + } + return clean +} + // Increment bumps the counter for the given peer. func (vc VClock) Increment(peer string) { vc[peer]++ diff --git a/internal/federation/vclock_test.go b/internal/federation/vclock_test.go index 9bcd0ac..c6d5aa1 100644 --- a/internal/federation/vclock_test.go +++ b/internal/federation/vclock_test.go @@ -27,6 +27,25 @@ func TestVClock_Clone(t *testing.T) { } } +func TestKeepCortexIDKeys_DropsLegacyNameBuckets(t *testing.T) { + vc := VClock{ + "01TESTCORTEXIDXXXXXXXXXXXX": 12, + "mycortex": 3, + "peer-a": 7, + } + + clean := KeepCortexIDKeys(vc) + if clean["01TESTCORTEXIDXXXXXXXXXXXX"] != 12 { + t.Errorf("ULID bucket missing from cleaned clock: %v", clean) + } + if _, ok := clean["mycortex"]; ok { + t.Errorf("legacy local-name bucket survived: %v", clean) + } + if _, ok := clean["peer-a"]; ok { + t.Errorf("legacy peer-name bucket survived: %v", clean) + } +} + func TestMerge(t *testing.T) { a := VClock{"a": 10, "b": 7, "c": 3} b := VClock{"a": 9, "b": 8, "c": 3}