Coverage for node / src / stigmem_node / routes / recall / orchestration.py: 94%
164 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Recall route orchestration."""
3from __future__ import annotations
5import hashlib
6import uuid
7from typing import Annotated, Any
9from fastapi import Depends, Header, HTTPException, Query, Response, status
10from fastapi.responses import JSONResponse
12from ...auth import Identity, resolve_identity
13from ...card_materializer import CARD_MIN_CONFIDENCE, get_fresh_card
14from ...db import db
15from ...garden_acl import caller_can_see_garden
16from ...lifecycle.tombstone_cache import is_tombstoned as _is_tombstoned
17from ...memory_garden_acl_gate import garden_acl_enforced
18from ...metrics import FACT_READ, RECALL_RANKER_DURATION, observe_duration
19from ...models.constants import VALID_SCOPES
20from ...models.facts import FactRecord, FactValue
21from ...models.recall import (
22 FactChainProof,
23 RecallRequest,
24 RecallResponse,
25 ScoreBreakdown,
26 ScoredFact,
27)
28from ...plugins import get_registry
29from ...recall.graph import MAX_DEPTH
30from ...recall.recall_pipeline import apply_recall_pipeline
31from ...session_graph import record_read_scopes
32from ...tracing import start_span
33from ..time_travel_gate import require_time_travel_enabled
34from .as_of import _recall_as_of_impl
35from .common import (
36 _estimate_tokens,
37 _fetch_facts_by_ids,
38 _now_iso,
39 _write_recall_audit,
40 logger,
41 router,
42)
43from .graph import _MAX_SEED_ENTITIES, _graph_expand
44from .lexical import _lexical_search
45from .ranking import _filter_visible_gardens, _greedy_pack, _score_candidates
46from .vector import _semantic_search
48_MAX_CANDIDATES = 500
49@router.post("", response_model=RecallResponse)
50def recall(
51 req: RecallRequest,
52 identity: Annotated[Identity, Depends(resolve_identity)],
53 response: Response,
54 session_id: Annotated[str | None, Header(alias="Stigmem-Session")] = None,
55 verify_mode: Annotated[str | None, Header(alias="Stigmem-Verify")] = None,
56 legacy_format: Annotated[
57 bool,
58 Query(
59 description=(
60 "Return the temporary legacy recall response shape without "
61 "`content` / `instructions` channel fields."
62 )
63 ),
64 ] = False,
65) -> RecallResponse | JSONResponse:
66 """Hybrid recall — return the most salient facts for a query, within budget.
68 Combines lexical (FTS5/BM25), dense-vector, and graph-traversal signals.
69 Honors Spec-02-Scopes-and-ACL and Spec-05-Federation-Trust at every step.
70 """
71 with start_span(
72 "stigmem.recall",
73 **{
74 "stigmem.tenant": identity.tenant_id,
75 "stigmem.principal": identity.entity_uri,
76 "stigmem.scope": req.scope,
77 },
78 ) as _span:
79 result = _recall_impl(
80 req,
81 identity,
82 _span,
83 session_id=session_id,
84 verify_full=verify_mode == "full",
85 )
86 headers = {}
87 if result.total_scored is not None:
88 headers["X-Total-Count"] = str(result.total_scored)
89 response.headers["X-Total-Count"] = headers["X-Total-Count"]
90 if legacy_format:
91 return JSONResponse(content=_legacy_recall_payload(result), headers=headers)
92 return result
95def _legacy_recall_payload(result: RecallResponse) -> dict[str, Any]:
96 """Return the one-minor-version compatibility shape for pre-channel clients."""
97 return result.model_dump(mode="json", exclude={"content", "instructions"})
100def _split_interpretation_channels(
101 packed: list[ScoredFact],
102) -> tuple[list[ScoredFact], list[ScoredFact]]:
103 content = [scored for scored in packed if scored.fact.value.interpret_as != "instruction"]
104 instructions = [scored for scored in packed if scored.fact.value.interpret_as == "instruction"]
105 return content, instructions
108def _validate_recall_request(req: RecallRequest, identity: Identity) -> None:
109 """Auth + scope + depth validation. Raises HTTPException on failure."""
110 if not identity.can_read(): 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true
111 raise HTTPException(
112 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
113 )
114 FACT_READ.labels(principal=identity.entity_uri, tenant=identity.tenant_id).inc()
115 if req.scope not in VALID_SCOPES:
116 raise HTTPException(
117 status_code=status.HTTP_400_BAD_REQUEST,
118 detail=f"invalid_scope: must be one of {sorted(VALID_SCOPES)}",
119 )
120 if req.depth > MAX_DEPTH: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 raise HTTPException(
122 status_code=status.HTTP_400_BAD_REQUEST,
123 detail={"code": "graph_depth_exceeded", "message": f"depth must be ≤ {MAX_DEPTH}"},
124 )
127def _chain_proof_or_409(conn: Any, tenant_id: str) -> FactChainProof:
128 from ...fact_chain import FactChainIntegrityError, build_fact_chain_proof
130 try:
131 return FactChainProof(**build_fact_chain_proof(conn, tenant_id=tenant_id))
132 except FactChainIntegrityError as exc:
133 result = exc.result
134 raise HTTPException(
135 status_code=status.HTTP_409_CONFLICT,
136 detail={
137 "code": "fact_chain_mismatch",
138 "message": "local fact hash chain verification failed",
139 "mismatch_reason": result.mismatch_reason,
140 "fact_id": result.fact_id,
141 "chain_seq": result.chain_seq,
142 },
143 ) from exc
146def _handle_as_of_recall(
147 req: RecallRequest,
148 identity: Identity,
149 *,
150 verify_full: bool = False,
151) -> RecallResponse:
152 """§24 time-travel path. Validates as_of, runs the as_of impl, returns the response."""
153 from ..facts import _validate_as_of
155 require_time_travel_enabled(get_registry(), surface="recall")
156 _validate_as_of(req.as_of) # type: ignore[arg-type] # caller guarantees as_of is not None
157 recall_id = str(uuid.uuid4())
158 query_hash = hashlib.sha256(req.query.encode()).hexdigest()
159 with db() as conn:
160 packed, notices, tombstone_filtered = _recall_as_of_impl(
161 conn,
162 query=req.query,
163 scope=req.scope,
164 as_of=req.as_of, # type: ignore[arg-type]
165 is_admin_caller=identity.is_admin(),
166 tenant_id=identity.tenant_id,
167 max_chunks=req.limit,
168 include_graph=req.include_neighbors,
169 identity=identity,
170 weights=req.weights,
171 depth=req.depth,
172 )
173 chain_proof = _chain_proof_or_409(conn, identity.tenant_id) if verify_full else None
174 tokens_used = sum(sf.token_estimate for sf in packed)
175 content, instructions = _split_interpretation_channels(packed)
176 return RecallResponse(
177 recall_id=recall_id,
178 query_hash=query_hash,
179 facts=packed,
180 content=content,
181 instructions=instructions,
182 # §23.3.3 r.3: suppress total_scored when tombstone filtering was applied
183 total_scored=None if tombstone_filtered else len(packed),
184 token_budget=req.token_budget,
185 tokens_used=tokens_used,
186 truncated=False,
187 tombstone_notices=notices,
188 chain_proof=chain_proof,
189 )
192def _gather_direct_matches(
193 conn: Any,
194 req: RecallRequest,
195 identity: Identity,
196 now: str,
197) -> tuple[dict[str, float], dict[str, float]]:
198 """Run the lexical + semantic searches that produce direct-match scores."""
199 lex_scores = (
200 _lexical_search(
201 conn,
202 req.query,
203 req.scope,
204 identity.tenant_id,
205 req.limit,
206 req.min_confidence,
207 now,
208 )
209 if req.weights.lexical > 0
210 else {}
211 )
212 sem_scores = (
213 _semantic_search(conn, req.query, req.scope, identity.tenant_id, req.limit)
214 if req.weights.semantic > 0
215 else {}
216 )
217 return lex_scores, sem_scores
220def _expand_graph_neighbours(
221 conn: Any,
222 req: RecallRequest,
223 identity: Identity,
224 direct_ids: set[str],
225 lex_scores: dict[str, float],
226 sem_scores: dict[str, float],
227 now: str,
228) -> dict[str, int]:
229 """Optionally BFS from the top direct-match seeds. Returns {fact_id: hop_distance}."""
230 if not (req.include_neighbors and req.weights.graph > 0 and direct_ids):
231 return {}
232 top_seeds = sorted(
233 direct_ids,
234 key=lambda fid: lex_scores.get(fid, 0) + sem_scores.get(fid, 0),
235 reverse=True,
236 )[:_MAX_SEED_ENTITIES]
237 return _graph_expand(
238 conn,
239 top_seeds,
240 req.depth,
241 req.scope,
242 identity.tenant_id,
243 identity,
244 req.limit,
245 req.min_confidence,
246 now,
247 )
250def _caller_sees_all_card_gardens(
251 entity_uri: str, scope: str, identity: Identity, conn: Any, now: str
252) -> bool:
253 """True if the caller may see every garden contributing to this entity's card.
255 The card summary aggregates an entity's fact values verbatim and bypasses the
256 ranker's per-fact garden ACL, so the card must not be served when any
257 contributing fact lives in a garden the caller cannot see (audit H1).
258 """
259 # Use the PROJECTED garden (the same COALESCE(fgm, f) the card aggregation and
260 # the rest of recall use): a fact promoted into a restricted garden via
261 # fact_garden_membership has raw facts.garden_id NULL, and checking the raw
262 # column alone would miss it and serve the card (audit F-C bypass of H1).
263 rows = conn.execute(
264 "SELECT DISTINCT COALESCE(fgm.garden_id, f.garden_id) AS gid FROM facts f"
265 " LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id"
266 " WHERE f.entity = ? AND f.scope = ? AND f.tenant_id = ?"
267 " AND f.confidence > 0"
268 " AND (f.valid_until IS NULL OR f.valid_until > ?)"
269 " AND (f.quarantine_status IS NULL OR f.quarantine_status != 'pending')"
270 " AND COALESCE(fgm.garden_id, f.garden_id) IS NOT NULL",
271 (entity_uri, scope, identity.tenant_id, now),
272 ).fetchall()
273 return all(caller_can_see_garden(row["gid"], identity) for row in rows)
276def _build_card_for_entity(
277 entity_uri: str,
278 entity_fact_ids: list[str],
279 req: RecallRequest,
280 identity: Identity,
281 conn: Any,
282 now: str,
283) -> tuple[ScoredFact, list[str]] | None:
284 """Try to build a synthetic ScoredFact from a fresh, high-confidence card.
286 Returns (scored_fact, owned_fact_ids) on success, None when no card qualifies.
287 """
288 if _is_tombstoned(entity_uri, identity.tenant_id): 288 ↛ 290line 288 didn't jump to line 290 because the condition on line 288 was never true
289 # §23.3.2 r.3: cards whose about_entity is tombstoned are fully excluded.
290 return None
291 # Garden ACL (audit H1): the card path bypasses the ranker's per-fact ACL,
292 # so a non-member must not receive a card aggregated from a garden's facts.
293 if garden_acl_enforced() and not _caller_sees_all_card_gardens(
294 entity_uri, req.scope, identity, conn, now
295 ):
296 return None
297 card = get_fresh_card(entity_uri, req.scope, identity.tenant_id, conn)
298 if (
299 card is None
300 or card.is_stale
301 or card.has_contradictions
302 or card.avg_confidence < CARD_MIN_CONFIDENCE
303 ):
304 return None
305 card_record = FactRecord(
306 id=f"card:{entity_uri}",
307 entity=entity_uri,
308 relation="stigmem:card:summary",
309 value=FactValue(type="text", v=card.summary),
310 source="system:stigmem:materializer",
311 timestamp=card.refreshed_at or now,
312 confidence=card.avg_confidence,
313 scope=req.scope,
314 )
315 sf = ScoredFact(
316 fact=card_record,
317 score=round(card.avg_confidence, 6),
318 score_breakdown=ScoreBreakdown(
319 source_trust=round(card.avg_confidence, 4),
320 weighted_total=round(card.avg_confidence, 6),
321 ),
322 hop_distance=0,
323 token_estimate=_estimate_tokens(card_record),
324 from_card=True,
325 )
326 return sf, list(entity_fact_ids)
329def _try_card_fast_path(
330 all_facts_raw: dict[str, FactRecord],
331 req: RecallRequest,
332 identity: Identity,
333 conn: Any,
334 now: str,
335) -> tuple[list[ScoredFact], set[str]]:
336 """Build card-derived ScoredFacts for any candidate entity that has a fresh card.
338 Returns (card_facts, card_owned_fact_ids). Logs and returns ([], set()) on any error
339 so the recall path can fall through to raw-fact scoring.
340 """
341 card_facts: list[ScoredFact] = []
342 card_entity_ids: set[str] = set()
343 try:
344 candidate_entities: dict[str, list[str]] = {}
345 for fid, record in all_facts_raw.items():
346 candidate_entities.setdefault(record.entity, []).append(fid)
347 for entity_uri, entity_fact_ids in candidate_entities.items():
348 built = _build_card_for_entity(entity_uri, entity_fact_ids, req, identity, conn, now)
349 if built is not None:
350 sf, owned = built
351 card_facts.append(sf)
352 card_entity_ids.update(owned)
353 except Exception as _card_exc:
354 logger.warning("card fast-path error (falling through to raw facts): %s", _card_exc)
355 return [], set()
356 return card_facts, card_entity_ids
359def _exclude_card_owned(
360 card_entity_ids: set[str],
361 all_facts_raw: dict[str, FactRecord],
362 lex_scores: dict[str, float],
363 sem_scores: dict[str, float],
364 graph_hops: dict[str, int],
365) -> tuple[dict[str, FactRecord], dict[str, float], dict[str, float], dict[str, int]]:
366 """Drop any fact_id owned by a card-served entity from all four scoring inputs.
368 No-op when ``card_entity_ids`` is empty.
369 """
370 if not card_entity_ids:
371 return all_facts_raw, lex_scores, sem_scores, graph_hops
372 return (
373 {k: v for k, v in all_facts_raw.items() if k not in card_entity_ids},
374 {k: v for k, v in lex_scores.items() if k not in card_entity_ids},
375 {k: v for k, v in sem_scores.items() if k not in card_entity_ids},
376 {k: v for k, v in graph_hops.items() if k not in card_entity_ids},
377 )
380def _set_recall_span_attrs(
381 span: object,
382 recall_id: str,
383 total_scored: int,
384 tokens_used: int,
385 truncated: bool,
386) -> None:
387 """Best-effort: attach recall outcome attributes to the OTel span."""
388 try:
389 span.set_attribute("stigmem.recall_id", recall_id) # type: ignore[attr-defined]
390 span.set_attribute("stigmem.total_scored", total_scored) # type: ignore[attr-defined]
391 span.set_attribute("stigmem.tokens_used", tokens_used) # type: ignore[attr-defined]
392 span.set_attribute("stigmem.truncated", truncated) # type: ignore[attr-defined]
393 except Exception as exc: # noqa: BLE001 # nosec B110 — span attrs best-effort
394 logger.debug("recall span attribute set failed: %s", exc)
397def _recall_impl(
398 req: RecallRequest,
399 identity: Identity,
400 _span: object,
401 *,
402 session_id: str | None = None,
403 verify_full: bool = False,
404) -> RecallResponse:
405 _validate_recall_request(req, identity)
407 if req.as_of is not None:
408 result = _handle_as_of_recall(req, identity, verify_full=verify_full)
409 with db() as conn:
410 record_read_scopes(
411 conn,
412 identity=identity,
413 session_id=session_id,
414 scopes={scored.fact.scope for scored in result.facts},
415 )
416 return result
418 recall_id = str(uuid.uuid4())
419 query_hash = hashlib.sha256(req.query.encode()).hexdigest()
420 now = _now_iso()
422 logger.info(
423 "recall id=%s entity=%s query_hash=%s scope=%s budget=%d",
424 recall_id,
425 identity.entity_uri,
426 query_hash[:12],
427 req.scope,
428 req.token_budget,
429 )
431 with db() as conn:
432 lex_scores, sem_scores = _gather_direct_matches(conn, req, identity, now)
433 direct_ids = set(lex_scores) | set(sem_scores)
435 graph_hops = _expand_graph_neighbours(
436 conn,
437 req,
438 identity,
439 direct_ids,
440 lex_scores,
441 sem_scores,
442 now,
443 )
444 # Direct matches have hop_distance=0 (mark in graph_hops)
445 for fid in direct_ids:
446 if fid not in graph_hops: 446 ↛ 445line 446 didn't jump to line 445 because the condition on line 446 was always true
447 graph_hops[fid] = 0
449 # --- Fetch all candidate facts ---
450 all_candidate_ids = list(direct_ids | set(graph_hops.keys()))[:_MAX_CANDIDATES]
451 all_facts_raw = _fetch_facts_by_ids(conn, all_candidate_ids)
453 # §23.3.2 r.3: exclude facts whose entity has an active tombstone (about_entity).
454 # Uses in-process cache (§23.3.3 r.4) — no per-fact DB read required.
455 pre_tombstone_count = len(all_facts_raw)
456 all_facts_raw = {
457 k: v
458 for k, v in all_facts_raw.items()
459 if not _is_tombstoned(v.entity, identity.tenant_id)
460 }
461 tombstone_filtered = len(all_facts_raw) < pre_tombstone_count
463 # Garden ACL on the shared candidate set (audit M3): drop facts whose
464 # garden the caller cannot see BEFORE the card fast-path and the ranker
465 # consume them, so the per-record garden check (§17 ranker) is a
466 # redundant backstop rather than the sole gate.
467 all_facts_raw = _filter_visible_gardens(all_facts_raw, identity)
469 # --- Card fast-path (§20) ---
470 card_facts, card_entity_ids = _try_card_fast_path(
471 all_facts_raw,
472 req,
473 identity,
474 conn,
475 now,
476 )
477 all_facts_raw, lex_scores, sem_scores, graph_hops = _exclude_card_owned(
478 card_entity_ids,
479 all_facts_raw,
480 lex_scores,
481 sem_scores,
482 graph_hops,
483 )
485 # Apply §19 recall pipeline (source-trust multiplier + content sanitiser)
486 all_facts = {r.id: r for r in apply_recall_pipeline(list(all_facts_raw.values()), identity)}
488 # --- Score (timed for ranker histogram) ---
489 with observe_duration(RECALL_RANKER_DURATION, {"tenant": identity.tenant_id}):
490 candidates = _score_candidates(
491 all_facts,
492 lex_scores,
493 sem_scores,
494 graph_hops,
495 req.weights,
496 identity,
497 req.depth,
498 )
499 candidates.extend(card_facts)
500 total_scored = len(candidates)
502 # --- Token-budget packing ---
503 packed, tokens_used, truncated = _greedy_pack(candidates, req.token_budget)
505 # --- Audit ---
506 _write_recall_audit(
507 conn,
508 recall_id,
509 identity,
510 query_hash,
511 req.scope,
512 req.token_budget,
513 len(packed),
514 tokens_used,
515 truncated,
516 )
517 record_read_scopes(
518 conn,
519 identity=identity,
520 session_id=session_id,
521 scopes={scored.fact.scope for scored in packed},
522 )
523 chain_proof = _chain_proof_or_409(conn, identity.tenant_id) if verify_full else None
525 logger.info(
526 "recall id=%s scored=%d packed=%d tokens=%d truncated=%s",
527 recall_id,
528 total_scored,
529 len(packed),
530 tokens_used,
531 truncated,
532 )
534 _set_recall_span_attrs(_span, recall_id, total_scored, tokens_used, truncated)
536 # §23.3.3 r.3: suppress total_scored when tombstone filtering was applied
537 content, instructions = _split_interpretation_channels(packed)
538 return RecallResponse(
539 recall_id=recall_id,
540 query_hash=query_hash,
541 facts=packed,
542 content=content,
543 instructions=instructions,
544 total_scored=None if tombstone_filtered else total_scored,
545 token_budget=req.token_budget,
546 tokens_used=tokens_used,
547 truncated=truncated,
548 chain_proof=chain_proof,
549 )