Coverage for node / src / stigmem_node / routes / recall / as_of.py: 88%
62 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Time-travel recall implementation."""
3from __future__ import annotations
5from typing import Any
7from ...auth import Identity
8from ...models.facts import row_to_record
9from ...models.recall import RecallWeights, ScoreBreakdown, ScoredFact
10from ...models.tombstones import TombstoneNotice
11from ...plugins import get_registry
12from ...recall.recall_pipeline import apply_recall_pipeline
13from ..cid_integrity import enforce_read_path_cid
14from .common import _estimate_tokens
15from .ranking import _filter_visible_gardens, _greedy_pack
18def _recall_as_of_impl(
19 conn: Any,
20 *,
21 query: str,
22 scope: str,
23 as_of: str,
24 is_admin_caller: bool,
25 tenant_id: str,
26 max_chunks: int,
27 include_graph: bool,
28 identity: Identity,
29 weights: RecallWeights,
30 depth: int,
31) -> tuple[list[ScoredFact], list[TombstoneNotice], bool]:
32 """Return (scored_facts, tombstone_notices, tombstone_filtered) for a time-travel
33 recall at as_of (§24.4).
35 Fetches facts visible at as_of (existing, not expired, not retracted at T),
36 scores them, applies tombstone filter, and returns packed chunks.
37 """
38 from ..facts import _get_tombstone_filter
40 # Candidate fetch: facts visible at as_of
41 rows = conn.execute(
42 """
43 SELECT f.*,
44 COALESCE(fvo.valid_until, f.valid_until) AS projected_valid_until,
45 COALESCE(fvo.confidence, f.confidence) AS projected_confidence,
46 COALESCE(fgm.garden_id, f.garden_id) AS projected_garden_id,
47 COALESCE(fqs.quarantine_status, f.quarantine_status)
48 AS projected_quarantine_status,
49 COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id)
50 AS projected_quarantine_garden_id,
51 COALESCE(f.cid, (
52 SELECT fca.cid
53 FROM fact_cid_aliases fca
54 WHERE fca.fact_id = f.id
55 ORDER BY fca.cid
56 LIMIT 1
57 )) AS projected_cid
58 FROM facts f
59 LEFT JOIN fact_validity_overrides fvo ON fvo.fact_id = f.id
60 LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id
61 LEFT JOIN fact_quarantine_status fqs ON fqs.fact_id = f.id
62 WHERE f.scope = ?
63 AND f.tenant_id = ?
64 AND f.timestamp <= ?
65 AND (
66 COALESCE(fvo.valid_until, f.valid_until) IS NULL
67 OR COALESCE(fvo.valid_until, f.valid_until) > ?
68 )
69 AND NOT EXISTS (
70 SELECT 1 FROM fact_retractions fr
71 WHERE fr.fact_id = f.id AND fr.retracted_at <= ?
72 )
73 ORDER BY f.timestamp DESC
74 LIMIT ?
75 """,
76 (scope, tenant_id, as_of, as_of, as_of, max_chunks * 5),
77 ).fetchall()
79 if not rows:
80 return [], [], False
82 for row in rows:
83 enforce_read_path_cid(row)
84 all_facts_raw = {row["id"]: row_to_record(row) for row in rows}
86 # Apply recall pipeline (trust, sanitizer)
87 pipeline_out = apply_recall_pipeline(list(all_facts_raw.values()), identity)
88 all_facts = {r.id: r for r in pipeline_out}
90 # Garden ACL (audit M3): the time-travel path has its own scoring loop and
91 # never went through the ranker's per-record garden check, so without this
92 # it returned restricted-garden facts to non-members. Filter the candidate
93 # set here, identically to the live path.
94 all_facts = _filter_visible_gardens(all_facts, identity)
96 # Score using lexical signal against query (recency computed relative to as_of)
97 def _recency_as_of(ts_str: str, as_of_ts: str) -> float:
98 try:
99 from datetime import UTC, datetime
101 ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
102 aof = datetime.fromisoformat(as_of_ts.replace("Z", "+00:00"))
103 if ts.tzinfo is None: 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true
104 ts = ts.replace(tzinfo=UTC)
105 if aof.tzinfo is None: 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 aof = aof.replace(tzinfo=UTC)
107 days_old = (aof - ts).days
108 return max(0.0, 1.0 - days_old / 365.0)
109 except Exception:
110 return 0.5
112 scored: list[ScoredFact] = []
113 q_lower = query.lower()
114 total_weight = weights.lexical + weights.recency
115 if total_weight <= 0: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true
116 total_weight = 1.0
118 for record in all_facts.values():
119 if record.quarantine_status == "pending": 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 continue
121 text = f"{record.entity} {record.relation} {record.value.v or ''}".lower()
122 lex = sum(1.0 for w in q_lower.split() if w and w in text) / max(1, len(q_lower.split()))
123 rec_s = _recency_as_of(record.timestamp, as_of)
124 trust_s = 0.0
125 raw = (weights.lexical * lex + weights.recency * rec_s) / total_weight
126 final_score = raw * max(0.0, record.confidence)
127 scored.append(
128 ScoredFact(
129 fact=record,
130 score=round(final_score, 6),
131 score_breakdown=ScoreBreakdown(
132 lexical=round(lex, 4),
133 recency=round(rec_s, 4),
134 source_trust=round(trust_s, 4),
135 weighted_total=round(final_score, 6),
136 ),
137 hop_distance=0,
138 token_estimate=_estimate_tokens(record),
139 )
140 )
142 source_deltas = get_registry().fire_score_delta(
143 "recall_rank",
144 scored,
145 identity=identity,
146 weights=weights,
147 as_of=as_of,
148 )
149 if source_deltas:
150 scored = [
151 sf.model_copy(
152 update={
153 "score": round(max(0.0, sf.score + source_deltas.get(sf.fact.id, 0.0)), 6),
154 "score_breakdown": sf.score_breakdown.model_copy(
155 update={
156 "weighted_total": round(
157 max(0.0, sf.score + source_deltas.get(sf.fact.id, 0.0)),
158 6,
159 )
160 }
161 ),
162 }
163 )
164 if source_deltas.get(sf.fact.id, 0.0) != 0.0
165 else sf
166 for sf in scored
167 ]
169 scored.sort(key=lambda c: c.score, reverse=True)
171 # Tombstone filter (§24.3)
172 entity_uris = list({sf.fact.entity for sf in scored})
173 excluded, notices = _get_tombstone_filter(conn, entity_uris, scope, is_admin_caller, tenant_id)
174 tombstone_filtered = False
175 if excluded:
176 scored = [sf for sf in scored if sf.fact.entity not in excluded]
177 tombstone_filtered = True
179 packed, _, _ = _greedy_pack(scored[:max_chunks], max_chunks * 20)
180 return packed[:max_chunks], notices, tombstone_filtered