From 1a9b3f13734f78b17846c6cb73e48eaf2b258239 Mon Sep 17 00:00:00 2001 From: Siddhant Singh Date: Thu, 13 Aug 2026 11:41:28 +0000 Subject: [PATCH] feat: temporal_score and IncludeHistorical on temporal intent Classify intents before SearchOpt so before/when/used-to queries include superseded state. Add temporal_score as a fusion channel (not recency +0.05). Tag memory_type in metadata; set event ends_at on supersede (mig-16 columns). Current-state still prefers Austin. Co-authored-by: Siddhant Singh --- internal/memory/evidence_plane.go | 5 ++++ internal/memory/fusion_v2.go | 21 +++++++++++++ internal/memory/fusion_v2_test.go | 24 +++++++++++++++ internal/memory/recall.go | 13 ++++---- internal/memory/record.go | 6 ++++ internal/memory/service.go | 11 +++++-- internal/memory/service_test.go | 49 +++++++++++++++++++++++++++++++ internal/memory/temporal.go | 47 +++++++++++++++++++++++++++++ internal/memory/temporal_test.go | 19 ++++++++++++ internal/memory/trace.go | 13 +++++++- internal/store/postgres/events.go | 18 ++++++++++++ 11 files changed, 217 insertions(+), 9 deletions(-) diff --git a/internal/memory/evidence_plane.go b/internal/memory/evidence_plane.go index f8cb465d..5dec09ba 100644 --- a/internal/memory/evidence_plane.go +++ b/internal/memory/evidence_plane.go @@ -26,6 +26,11 @@ type EventWriter interface { UpsertMemoryEvent(ctx context.Context, tenantID, subjectID, eventID, eventType, title, description, memoryID, evidenceID string, startsAt *time.Time, confidence float64, participants []string) error } +// EventEnder closes world-valid event intervals on supersede (mig-16 ends_at). +type EventEnder interface { + EndMemoryEventsByMemoryID(ctx context.Context, tenantID, subjectID, memoryID string, endedAt *time.Time) error +} + // CurrentStateStore is the rebuildable projection for stateful predicates. type CurrentStateStore interface { UpsertCurrentState(ctx context.Context, tenantID, subjectID, predicate, memoryID, value, policy string) error diff --git a/internal/memory/fusion_v2.go b/internal/memory/fusion_v2.go index 6dadc18e..b08946ce 100644 --- a/internal/memory/fusion_v2.go +++ b/internal/memory/fusion_v2.go @@ -107,6 +107,27 @@ func ScoreAndRankV2(semantic, bm25, entityBoost, semanticThreshold, semanticOnly return combined, details } +// ScoreAndRankV2Temporal is ScoreAndRankV2 plus a [0,1] temporal_score channel. +// Zero temporal does not change max_possible (not a recency tie-break). +func ScoreAndRankV2Temporal(semantic, bm25, entityBoost, temporal, semanticThreshold, semanticOnlyFloor float64) (combined float64, details map[string]float64) { + combined, details = ScoreAndRankV2(semantic, bm25, entityBoost, semanticThreshold, semanticOnlyFloor) + if temporal <= 0 || combined <= 0 { + details["temporal"] = temporal + return combined, details + } + maxPossible := details["max_possible"] + if maxPossible <= 0 { + maxPossible = 1 + } + raw := combined*maxPossible + temporal + maxPossible += 1.0 + combined = math.Min(raw/maxPossible, 1.0) + details["temporal"] = temporal + details["combined"] = combined + details["max_possible"] = maxPossible + return combined, details +} + // CandidateOverfetch returns Mem0-style internal limit: max(limit*4, 60). func CandidateOverfetch(limit int) int { if limit <= 0 { diff --git a/internal/memory/fusion_v2_test.go b/internal/memory/fusion_v2_test.go index bea4f2a3..09fe7cbe 100644 --- a/internal/memory/fusion_v2_test.go +++ b/internal/memory/fusion_v2_test.go @@ -2,6 +2,21 @@ package memory import "testing" +func TestScoreAndRankV2TemporalAddsChannel(t *testing.T) { + base, _ := ScoreAndRankV2(0.8, 0.6, 0.25, 0.12, 0.42) + withT, parts := ScoreAndRankV2Temporal(0.8, 0.6, 0.25, 1.0, 0.12, 0.42) + if withT <= 0 || withT > 1 { + t.Fatalf("combined=%v", withT) + } + if parts["temporal"] != 1 { + t.Fatalf("temporal=%v", parts["temporal"]) + } + zero, zparts := ScoreAndRankV2Temporal(0.8, 0.6, 0.25, 0, 0.12, 0.42) + if zero != base { + t.Fatalf("zero temporal must not change fusion, got %v want %v %#v", zero, base, zparts) + } +} + func TestScoreAndRankV2Additive(t *testing.T) { combined, parts := ScoreAndRankV2(0.8, 0.6, 0.25, 0.12, 0.42) if combined <= 0 || combined > 1 { @@ -70,6 +85,15 @@ func TestAnalyzeQueryIntents(t *testing.T) { if !found[IntentCurrentState] && !found[IntentEnumeration] && !found[IntentPointFact] { t.Fatalf("expected useful intents, got %v", ints) } + if !WantsHistoricalRetrieval(AnalyzeQueryIntents("where did they live before")) { + t.Fatal("before should request historical retrieval") + } + if WantsHistoricalRetrieval(AnalyzeQueryIntents("where do they currently live")) { + t.Fatal("current-state should not force historical") + } + if !WantsHistoricalRetrieval(AnalyzeQueryIntents("how long have I been doing this")) { + t.Fatal("how long should be temporal") + } } func TestSelectEvidenceSetCoversDistinct(t *testing.T) { diff --git a/internal/memory/recall.go b/internal/memory/recall.go index 4916b6dc..2194e87a 100644 --- a/internal/memory/recall.go +++ b/internal/memory/recall.go @@ -66,15 +66,16 @@ func (s *Service) Recall(ctx context.Context, req RecallRequest) (RecallResponse budget = 4000 } + intents := AnalyzeQueryIntents(req.Query) + hist := req.IncludeHistorical || strings.EqualFold(req.View, "historical") || strings.EqualFold(req.View, "all") || WantsHistoricalRetrieval(intents) search, err := s.SearchOpt(ctx, req.TenantID, req.SubjectID, req.Vertical, "", req.Query, SearchOptions{ - IncludeHistorical: req.IncludeHistorical || strings.EqualFold(req.View, "historical") || strings.EqualFold(req.View, "all"), + IncludeHistorical: hist, Limit: topK, }) if err != nil { return RecallResponse{}, err } - intents := AnalyzeQueryIntents(req.Query) plan := PlanQuery(req.Query, intents) if !modeExplicit { mode = plan.PreferredModeHint @@ -88,9 +89,10 @@ func (s *Service) Recall(ctx context.Context, req RecallRequest) (RecallResponse Intents: intents, Trace: search.Trace, Explain: map[string]any{ - "top_k": topK, - "budget_tokens": budget, - "query_plan": plan, + "top_k": topK, + "budget_tokens": budget, + "query_plan": plan, + "include_historical": hist, }, } if req.View != "" { @@ -114,7 +116,6 @@ func (s *Service) Recall(ctx context.Context, req RecallRequest) (RecallResponse } pkt := BuildEvidencePacket(plan, search.Results, out.Explain) bindPacketToTargets(&pkt, search.Results, req.Query, plan.CoverageTargets) - hist := req.IncludeHistorical || strings.EqualFold(req.View, "historical") || strings.EqualFold(req.View, "all") // Typed hop executor: bind packet from hop joins when hops exist. if plan.NeedsMultiHop && len(plan.Hops) > 0 { hopResults, byKey := s.executeTypedHops(ctx, req.TenantID, req.SubjectID, req.Vertical, hist, plan, topK) diff --git a/internal/memory/record.go b/internal/memory/record.go index 1f47a7d6..245cad96 100644 --- a/internal/memory/record.go +++ b/internal/memory/record.go @@ -161,6 +161,12 @@ func BuildMemoryRecord(memoryID string, now time.Time, req IngestRequest, extrac } record.Content = sanitizeUTF8(record.Content) record.SourceText = sanitizeUTF8(record.SourceText) + if record.Metadata == nil { + record.Metadata = map[string]any{} + } + if v, ok := record.Metadata["memory_type"].(string); !ok || strings.TrimSpace(v) == "" { + record.Metadata["memory_type"] = memoryTypeOf(record) + } return record, nil } diff --git a/internal/memory/service.go b/internal/memory/service.go index a4e30861..0bfc03a1 100644 --- a/internal/memory/service.go +++ b/internal/memory/service.go @@ -314,6 +314,11 @@ func (s *Service) SearchOpt(ctx context.Context, tenantID, subjectID, vertical, queryTokens := tokenize(query) contentQueryTokens := contentBearingTokens(queryTokens) + intents := AnalyzeQueryIntents(query) + if !opts.IncludeHistorical && WantsHistoricalRetrieval(intents) { + opts.IncludeHistorical = true + } + patterns := make([]string, 0, len(contentQueryTokens)+len(queryTokens)) for _, t := range contentQueryTokens { patterns = append(patterns, "%"+t+"%") @@ -327,7 +332,6 @@ func (s *Service) SearchOpt(ctx context.Context, tenantID, subjectID, vertical, includeSuperseded := opts.IncludeHistorical fusionV2 := FusionV2Enabled() - intents := AnalyzeQueryIntents(query) overfetch := CandidateOverfetch(opts.Limit) trace := &SearchTrace{ CandidateOverfetch: overfetch, @@ -634,7 +638,7 @@ func (s *Service) SearchOpt(ctx context.Context, tenantID, subjectID, vertical, // Short entity-probe queries: block template false-friends. semOnlyFloor = 0.78 } - combined, parts := ScoreAndRankV2(embedScore, bm25, hub, 0.12, semOnlyFloor) + combined, parts := ScoreAndRankV2Temporal(embedScore, bm25, hub, TemporalScore(record, intents, includeSuperseded), 0.12, semOnlyFloor) if combined <= 0 && score <= 0 { continue } @@ -1445,6 +1449,9 @@ func AutoSupersedePriorState(ctx context.Context, store Store, record MemoryReco if retirer, ok := indexer.(AtomRetirer); ok { _ = retirer.RetireMemoryAtom(ctx, record.TenantID, record.SubjectID, pred, pval, id, record.ObservedAt) } + if ender, ok := store.(EventEnder); ok { + _ = ender.EndMemoryEventsByMemoryID(ctx, record.TenantID, record.SubjectID, id, record.ObservedAt) + } } return nil } diff --git a/internal/memory/service_test.go b/internal/memory/service_test.go index 64cf73a9..688bd217 100644 --- a/internal/memory/service_test.go +++ b/internal/memory/service_test.go @@ -830,6 +830,55 @@ func TestSupersedeHidesPriorFromDefaultSearch(t *testing.T) { } } +func TestHistoricalIntentRetrievesPriorResidence(t *testing.T) { + store := newMemoryStoreStub() + service := NewService(store) + now := service.now() + store.records["ny"] = MemoryRecord{ + MemoryID: "mem_ny", TenantID: "t1", SubjectID: "u1", + Kind: KindFact, Content: "Alex lives in New York", + DedupeKey: "ny", Status: StatusActive, LifecycleState: LifecycleSuperseded, + Metadata: map[string]any{"predicate": PredicateResidence, "value_norm": "new york", "memory_type": "state"}, + CreatedAt: now, UpdatedAt: now, + } + store.records["au"] = MemoryRecord{ + MemoryID: "mem_au", TenantID: "t1", SubjectID: "u1", + Kind: KindFact, Content: "Alex lives in Austin", + DedupeKey: "au", Status: StatusActive, LifecycleState: LifecycleActive, + Metadata: map[string]any{"predicate": PredicateResidence, "value_norm": "austin", "memory_type": "state"}, + CreatedAt: now, UpdatedAt: now, + } + + cur, err := service.SearchOpt(context.Background(), "t1", "u1", "", "", "where does Alex currently live", SearchOptions{Limit: 10}) + if err != nil { + t.Fatal(err) + } + joined := "" + for _, r := range cur.Results { + joined += " " + r.Content + if strings.Contains(r.Content, "New York") { + t.Fatalf("current-state search leaked superseded NY: %q", r.Content) + } + } + if !strings.Contains(joined, "Austin") { + t.Fatalf("current-state should prefer Austin, got %q", joined) + } + + hist, err := service.SearchOpt(context.Background(), "t1", "u1", "", "", "where did Alex live before", SearchOptions{Limit: 10}) + if err != nil { + t.Fatal(err) + } + foundNY := false + for _, r := range hist.Results { + if strings.Contains(r.Content, "New York") { + foundNY = true + } + } + if !foundNY { + t.Fatalf("historical intent should retrieve NY, results=%+v", hist.Results) + } +} + func TestListQueryDiversifiesThemes(t *testing.T) { store := newMemoryStoreStub() service := NewService(store) diff --git a/internal/memory/temporal.go b/internal/memory/temporal.go index c785ed52..55eff3b5 100644 --- a/internal/memory/temporal.go +++ b/internal/memory/temporal.go @@ -86,6 +86,53 @@ func worldValidTime(record MemoryRecord) *time.Time { return nil } +// TemporalScore is a [0,1] fusion channel for world-valid history vs current state. +// It is not a recency tie-break. Superseded rows score high only on historical intent. +func TemporalScore(record MemoryRecord, intents []string, includeHistorical bool) float64 { + hist := includeHistorical || WantsHistoricalRetrieval(intents) + current := false + for _, intent := range intents { + if intent == IntentCurrentState { + current = true + } + } + superseded := record.LifecycleState == LifecycleSuperseded + memType := memoryTypeOf(record) + switch { + case hist && superseded: + return 1.0 + case current && superseded: + return 0 + case current && !superseded && memType == "state": + return 0.7 + case hist && memType == "state": + return 0.55 + case hist && record.ObservedAt != nil: + return 0.45 + default: + return 0 + } +} + +func memoryTypeOf(record MemoryRecord) string { + if record.Metadata != nil { + if v, ok := record.Metadata["memory_type"].(string); ok && strings.TrimSpace(v) != "" { + return strings.TrimSpace(v) + } + } + if record.Primitive == PrimitiveEpisode { + return "episode" + } + pred := "" + if record.Metadata != nil { + pred, _ = record.Metadata["predicate"].(string) + } + if IsStatefulPredicate(pred) { + return "state" + } + return "event" +} + // ParseAsOf parses RFC3339 / date-only as_of strings. func ParseAsOf(raw string) (time.Time, bool) { raw = strings.TrimSpace(raw) diff --git a/internal/memory/temporal_test.go b/internal/memory/temporal_test.go index a2cb8409..97e98375 100644 --- a/internal/memory/temporal_test.go +++ b/internal/memory/temporal_test.go @@ -160,3 +160,22 @@ func TestPredicateHintsFromQuery(t *testing.T) { t.Fatalf("expected residence hint, got %#v", h) } } + +func TestTemporalScorePrefersSupersededOnHistorical(t *testing.T) { + ny := MemoryRecord{ + LifecycleState: LifecycleSuperseded, + Metadata: map[string]any{"memory_type": "state", "predicate": PredicateResidence}, + } + austin := MemoryRecord{ + LifecycleState: LifecycleActive, + Metadata: map[string]any{"memory_type": "state", "predicate": PredicateResidence}, + } + hist := []string{IntentHistoricalState} + if TemporalScore(ny, hist, true) <= TemporalScore(austin, hist, true) { + t.Fatal("superseded prior should outrank current on historical intent") + } + cur := []string{IntentCurrentState} + if TemporalScore(austin, cur, false) <= TemporalScore(ny, cur, false) { + t.Fatal("current-state should prefer active over superseded") + } +} diff --git a/internal/memory/trace.go b/internal/memory/trace.go index e9dd5b59..93cd677d 100644 --- a/internal/memory/trace.go +++ b/internal/memory/trace.go @@ -80,7 +80,8 @@ func AnalyzeQueryIntents(query string) []string { } if strings.Contains(lower, "when") || strings.Contains(lower, "before") || strings.Contains(lower, "after") || strings.Contains(lower, "ago") || - strings.Contains(lower, "last year") || strings.Contains(lower, "yesterday") { + strings.Contains(lower, "last year") || strings.Contains(lower, "yesterday") || + strings.Contains(lower, "how long") || strings.Contains(lower, "how old") { add(IntentTemporalSequence) } if strings.Contains(lower, "prefer") || strings.Contains(lower, "favorite") || @@ -104,3 +105,13 @@ func AnalyzeQueryIntents(query string) []string { } return out } + +// WantsHistoricalRetrieval is true for before/after/when/used-to style intents. +func WantsHistoricalRetrieval(intents []string) bool { + for _, intent := range intents { + if intent == IntentHistoricalState || intent == IntentTemporalSequence { + return true + } + } + return false +} diff --git a/internal/store/postgres/events.go b/internal/store/postgres/events.go index eb3bb36e..a7e01896 100644 --- a/internal/store/postgres/events.go +++ b/internal/store/postgres/events.go @@ -166,6 +166,24 @@ ON CONFLICT DO NOTHING return nil } +// EndMemoryEventsByMemoryID sets ends_at on events tied to a superseded memory. +func (s *Store) EndMemoryEventsByMemoryID(ctx context.Context, tenantID, subjectID, memoryID string, endedAt *time.Time) error { + if tenantID == "" || memoryID == "" { + return nil + } + end := time.Now().UTC() + if endedAt != nil { + end = endedAt.UTC() + } + _, err := s.pool.Exec(ctx, ` +UPDATE memory_events +SET ends_at = COALESCE(ends_at, $4) +WHERE tenant_id = $1 AND subject_id = $2 AND memory_id = $3 + AND ends_at IS NULL +`, tenantID, subjectID, memoryID, end) + return err +} + // UpsertCurrentState writes a rebuildable current-state projection row. // Callers must ensure the winning assertion is temporally valid before calling. func (s *Store) UpsertCurrentState(ctx context.Context, tenantID, subjectID, predicate, memoryID, value, policy string) error {