Coverage for node / src / stigmem_node / routes / facts / query.py: 93%
205 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""GET /v1/facts query route and query helpers."""
3from __future__ import annotations
5import uuid
6from datetime import UTC, datetime, timedelta
7from typing import Annotated, Any
9from fastapi import Depends, Header, HTTPException, Query, Response, status
11from ... import settings as _settings_pkg
12from ...auth import Identity, resolve_identity
13from ...db import db
14from ...entity_normalizer import NormalizationError, normalize_entity_uri
15from ...fact_visibility import ReadScope, caller_read_scope
16from ...garden_acl import get_garden_by_garden_uri, require_garden_read
17from ...memory_garden_acl_gate import garden_acl_enforced
18from ...metrics import FACT_READ
19from ...models.constants import VALID_SCOPES
20from ...models.facts import FactRecord, QueryResponse, row_to_record
21from ...models.tombstones import TombstoneNotice
22from ...plugins import Deny, Failure, Success, TenantContext, get_registry
23from ...recall.recall_pipeline import apply_recall_pipeline
24from ...session_graph import record_read_scopes
25from ..cid_integrity import enforce_read_path_cid
26from ..time_travel_gate import require_time_travel_enabled
27from .common import _get_tombstone_filter, logger, router
30def _validate_as_of(as_of: str) -> datetime:
31 """Parse and validate an as_of timestamp per §24.2.2."""
32 import re
34 # URL query strings decode + as space; restore the + in timezone offsets like "+00:00".
35 normalized = re.sub(r" (\d{2}:\d{2})$", r"+\1", as_of).replace("Z", "+00:00")
36 try:
37 ts = datetime.fromisoformat(normalized)
38 except ValueError as exc:
39 raise HTTPException(
40 status_code=400,
41 detail={"code": "as_of_invalid_timestamp", "message": str(exc)},
42 ) from exc
43 if ts.tzinfo is None: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true
44 ts = ts.replace(tzinfo=UTC)
45 if ts > datetime.now(UTC) + timedelta(seconds=5):
46 raise HTTPException(
47 status_code=400,
48 detail={"code": "as_of_future", "message": "as_of must not be in the future (§24.2.2)"},
49 )
50 floor = _settings_pkg.settings.as_of_retention_floor
51 if floor:
52 try:
53 floor_ts = datetime.fromisoformat(floor.replace("Z", "+00:00"))
54 if floor_ts.tzinfo is None: 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true
55 floor_ts = floor_ts.replace(tzinfo=UTC)
56 if ts < floor_ts:
57 raise HTTPException(
58 status_code=400,
59 detail={
60 "code": "as_of_before_retention_floor",
61 "message": "as_of predates the retention horizon for this deployment (§24.2.2)", # noqa: E501
62 },
63 )
64 except HTTPException:
65 raise
66 except Exception as exc: # nosec B110
67 logger.warning("could not read retention floor while validating as_of: %s", exc)
68 return ts
71def _legal_hold_blocks_query(conn: Any, entity: str) -> bool:
72 """F-14 §24.3.2: True if a legal-hold tombstone covers *entity* (no active revocation)."""
73 legal_hold_row = conn.execute(
74 # Same-issuer binding: only a revocation from the tombstone's OWN issuer
75 # (r.signed_by = t.signed_by) lifts the suppression — a forged/cross-issuer
76 # revocation can never clear a legal-hold tombstone (RTBF integrity).
77 """SELECT 1 FROM tombstones t
78 WHERE t.entity_uri = ?
79 AND t.legal_hold = 1
80 AND NOT EXISTS (
81 SELECT 1 FROM tombstone_revocations r
82 WHERE r.tombstone_id = t.id AND r.signed_by = t.signed_by
83 )
84 LIMIT 1""",
85 (entity,),
86 ).fetchone()
87 return legal_hold_row is not None
90_AS_OF_SELECT_SQL = (
91 "SELECT f.*, COALESCE(fgm.garden_id, f.garden_id) AS projected_garden_id"
92 " FROM facts f"
93 " LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id"
94 " WHERE f.tenant_id = ?"
95 " AND f.timestamp <= ?"
96 " AND (f.valid_until IS NULL OR f.valid_until > ?)"
97 " AND NOT EXISTS ("
98 " SELECT 1 FROM fact_retractions fr"
99 " WHERE fr.fact_id = f.id AND fr.retracted_at <= ?"
100 " )"
101 " AND (? IS NULL"
102 " OR f.entity = ?"
103 " OR f.entity IN ("
104 " SELECT raw_uri FROM entity_aliases WHERE canonical_uri = ? AND tenant_id = ?))"
105 " AND (? IS NULL OR f.relation = ?)"
106 " AND (? IS NULL OR f.scope = ?)"
107 " AND (? IS NULL OR f.id > ?)"
108 " ORDER BY f.timestamp DESC, f.id DESC"
109 " LIMIT ?"
110)
112_GARDEN_VISIBILITY_NONE = 0
113_GARDEN_VISIBILITY_EXACT = 1
114_GARDEN_VISIBILITY_VISIBLE_SET = 2
115_GARDEN_VISIBILITY_NULL_ONLY = 3
117_FACT_QUERY_SQL = (
118 "SELECT f.*, "
119 "COALESCE(fvo.valid_until, f.valid_until) AS projected_valid_until, "
120 "COALESCE(fvo.confidence, f.confidence) AS projected_confidence, "
121 "COALESCE(fgm.garden_id, f.garden_id) AS projected_garden_id, "
122 "COALESCE(fqs.quarantine_status, f.quarantine_status) AS projected_quarantine_status, "
123 "COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id) "
124 "AS projected_quarantine_garden_id, "
125 "COALESCE(f.cid, (SELECT fca.cid FROM fact_cid_aliases fca "
126 "WHERE fca.fact_id = f.id ORDER BY fca.cid LIMIT 1)) AS projected_cid"
127 " FROM facts f"
128 " LEFT JOIN fact_validity_overrides fvo ON fvo.fact_id = f.id"
129 " LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id"
130 " LEFT JOIN fact_quarantine_status fqs ON fqs.fact_id = f.id"
131 " WHERE COALESCE(fvo.confidence, f.confidence) >= ?"
132 " AND f.tenant_id = ?"
133 " AND ("
134 " ? = 0"
135 " OR (? = 1 AND COALESCE(fgm.garden_id, f.garden_id) = ?)"
136 " OR (? = 2 AND (COALESCE(fgm.garden_id, f.garden_id) IS NULL"
137 " OR EXISTS (SELECT 1 FROM _query_visible_gardens qvg"
138 " WHERE qvg.id = COALESCE(fgm.garden_id, f.garden_id))))"
139 " OR (? = 3 AND COALESCE(fgm.garden_id, f.garden_id) IS NULL)"
140 " )"
141 " AND (? IS NULL OR f.attested = ?)"
142 " AND (? IS NULL"
143 " OR f.entity = ?"
144 " OR f.entity IN ("
145 " SELECT raw_uri FROM entity_aliases WHERE canonical_uri = ? AND tenant_id = ?))"
146 " AND (? IS NULL OR f.relation = ?)"
147 " AND (? IS NULL"
148 " OR f.source = ?"
149 " OR f.source IN ("
150 " SELECT raw_uri FROM entity_aliases WHERE canonical_uri = ? AND tenant_id = ?))"
151 " AND (? IS NULL OR f.scope = ?)"
152 " AND (? IS NULL OR f.timestamp > ?)"
153 " AND (? IS NULL OR f.id > ?)"
154 " AND (? = 1"
155 " OR COALESCE(fvo.valid_until, f.valid_until) IS NULL"
156 " OR COALESCE(fvo.valid_until, f.valid_until) > ?)"
157 " ORDER BY f.timestamp DESC, f.id DESC LIMIT ?"
158)
160def _build_as_of_params(
161 *,
162 entity: str | None,
163 scope: str | None,
164 relation: str | None,
165 as_of: str,
166 tenant_id: str,
167 cursor: str | None,
168 limit: int,
169) -> list[Any]:
170 """Return the bind values for ``_AS_OF_SELECT_SQL``.
172 The SQL text is the ``_AS_OF_SELECT_SQL`` module-level constant; this
173 helper only computes bind values. Keeping the SQL string out of any
174 function that takes user input prevents CodeQL from interprocedurally
175 tainting it — see issue #121 for why a function that takes user
176 inputs and returns ``(sql, params)`` still trips ``py/sql-injection``
177 even when the returned SQL value is invariant.
178 """
179 if scope is not None and scope not in VALID_SCOPES: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 raise HTTPException(status_code=400, detail=f"scope must be one of {VALID_SCOPES}")
182 # Normalize empty strings to None so the IS NULL gate matches the
183 # previous ``if entity:`` truthiness behaviour.
184 entity_p = entity or None
185 relation_p = relation or None
186 scope_p = scope or None
187 cursor_p = cursor or None
189 return [
190 tenant_id,
191 as_of,
192 as_of,
193 as_of,
194 entity_p,
195 entity_p,
196 entity_p,
197 tenant_id, # entity-alias subquery tenant scope
198 relation_p,
199 relation_p,
200 scope_p,
201 scope_p,
202 cursor_p,
203 cursor_p,
204 limit + 1,
205 ]
208def _query_facts_as_of_impl(
209 conn: Any,
210 *,
211 entity: str | None,
212 scope: str | None,
213 relation: str | None,
214 as_of: str,
215 is_admin_caller: bool,
216 tenant_id: str,
217 read_scope: ReadScope,
218 limit: int,
219 cursor: str | None,
220) -> QueryResponse:
221 """Return facts visible at as_of per §24.4.
223 Retraction gating uses fact_retractions.retracted_at (append-only log), NOT facts.confidence.
224 Expiry gating uses facts.valid_until.
225 Tombstone filter per §24.3: retroactive RTBF unless legal_hold=true AND is_admin_caller.
226 """
227 # F-14 §24.3.2: pre-check — agent-key callers get empty results if a legal-hold
228 # tombstone covers the queried entity (short-circuit before executing the query)
229 if entity and not is_admin_caller and _legal_hold_blocks_query(conn, entity):
230 return QueryResponse(facts=[], total=None, cursor=None)
232 params = _build_as_of_params(
233 entity=entity,
234 scope=scope,
235 relation=relation,
236 as_of=as_of,
237 tenant_id=tenant_id,
238 cursor=cursor,
239 limit=limit,
240 )
242 raw = conn.execute(_AS_OF_SELECT_SQL, params).fetchall()
243 has_more = len(raw) > limit
244 page = raw[:limit]
245 # Garden ACL (fail-closed): drop facts whose projected garden the caller
246 # cannot see, BEFORE building records/contradiction counts. The id-cursor
247 # below continues from the page boundary (page[-1]), so dropping rows here
248 # never skips or duplicates a visible fact across pages (audit F-AS-OF-FACTS;
249 # the recall as_of path was fixed in M3, this is the /v1/facts?as_of= sibling).
250 rows = [r for r in page if read_scope.garden_allows(r["projected_garden_id"])]
252 seen: dict[tuple[str, str, str], int] = {}
253 for r in rows:
254 key = (r["entity"], r["relation"], r["scope"])
255 seen[key] = seen.get(key, 0) + 1
257 for r in rows:
258 enforce_read_path_cid(r)
259 records = [
260 row_to_record(r, contradicted=seen[(r["entity"], r["relation"], r["scope"])] > 1)
261 for r in rows
262 ]
264 # §24.3: tombstone filter for as_of queries
265 tombstone_notices: list[TombstoneNotice] = []
266 tombstone_filtered = False
267 if records:
268 entity_uris = list({r.entity for r in records})
269 excluded, tombstone_notices = _get_tombstone_filter(
270 conn, entity_uris, scope or "local", is_admin_caller, tenant_id
271 )
272 if excluded:
273 records = [r for r in records if r.entity not in excluded]
274 tombstone_filtered = True
276 # Cursor advances from the page boundary (not the last visible row) so
277 # garden-filtered rows don't stall pagination.
278 next_cursor = page[-1]["id"] if has_more and page else None
279 # §23.3.3 r.3: suppress total when tombstone filtering was applied to prevent oracle leakage
280 total = None if tombstone_filtered else len(records)
281 return QueryResponse(
282 facts=records,
283 total=total,
284 cursor=next_cursor,
285 tombstone_notices=tombstone_notices,
286 )
289@router.get("", response_model=QueryResponse)
290def query_facts(
291 identity: Annotated[Identity, Depends(resolve_identity)],
292 response: Response,
293 session_id: Annotated[str | None, Header(alias="Stigmem-Session")] = None,
294 entity: str | None = Query(None),
295 relation: str | None = Query(None),
296 source: str | None = Query(None),
297 scope: str | None = Query(None),
298 min_confidence: float = Query(0.0, ge=0.0, le=1.0),
299 include_contradicted: bool = Query(False),
300 include_expired: bool = Query(False),
301 after: str | None = Query(
302 None, description="Return facts with timestamp > this ISO 8601 value"
303 ), # noqa: E501
304 cursor: str | None = Query(None, description="Opaque pagination cursor (fact id)"),
305 limit: int = Query(50, ge=1, le=500),
306 garden_id: str | None = Query(
307 None, description="Filter to facts in this garden (Spec-02-Scopes-and-ACL)"
308 ), # noqa: E501
309 attested: bool | None = Query(
310 None, description="Filter by source-attestation status (Spec-X6-Source-Attestation)"
311 ), # noqa: E501
312 include_low_trust: bool = Query(
313 False,
314 description="Include facts with effective_confidence < 0.3 (Spec-05-Federation-Trust)",
315 ), # noqa: E501
316 as_of: str | None = Query(
317 None,
318 description="Time-travel query: return facts visible at this ISO 8601 timestamp (Spec-X3-Time-Travel-Queries)", # noqa: E501
319 ), # noqa: E501
320) -> QueryResponse:
321 """Query facts by pattern (Spec-03-HTTP-API).
323 Omitted fields are wildcards. Entity/source are normalized by Spec-01-Fact-Model.
324 """
325 if not identity.can_read():
326 raise HTTPException(
327 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required"
328 ) # noqa: E501
330 request_id = str(uuid.uuid4())
331 tenant = TenantContext(
332 tenant_id=identity.tenant_id,
333 metadata={"tenant_context_source": "hook"},
334 )
335 registry = get_registry()
336 query_payload: dict[str, Any] = {
337 "entity": entity,
338 "relation": relation,
339 "source": source,
340 "scope": scope,
341 "min_confidence": min_confidence,
342 "include_contradicted": include_contradicted,
343 "include_expired": include_expired,
344 "after": after,
345 "cursor": cursor,
346 "limit": limit,
347 "garden_id": garden_id,
348 "attested": attested,
349 "include_low_trust": include_low_trust,
350 "as_of": as_of,
351 }
352 decision = registry.fire_voting(
353 "pre_recall_authorize",
354 identity=identity,
355 tenant=tenant,
356 request_id=request_id,
357 query=query_payload,
358 )
359 if isinstance(decision, Deny):
360 registry.fire_fire_and_forget(
361 "post_recall_audit",
362 result=None,
363 identity=identity,
364 tenant=tenant,
365 request_id=request_id,
366 outcome=Failure(reason=decision.reason),
367 )
368 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=decision.reason)
370 rewritten_query = registry.fire_filter_chain(
371 "pre_recall_rewrite",
372 query_payload,
373 identity=identity,
374 tenant=tenant,
375 request_id=request_id,
376 )
377 entity = rewritten_query["entity"]
378 relation = rewritten_query["relation"]
379 source = rewritten_query["source"]
380 scope = rewritten_query["scope"]
381 min_confidence = rewritten_query["min_confidence"]
382 include_contradicted = rewritten_query["include_contradicted"]
383 include_expired = rewritten_query["include_expired"]
384 after = rewritten_query["after"]
385 cursor = rewritten_query["cursor"]
386 limit = rewritten_query["limit"]
387 garden_id = rewritten_query["garden_id"]
388 attested = rewritten_query["attested"]
389 include_low_trust = rewritten_query["include_low_trust"]
390 as_of = rewritten_query["as_of"]
392 # §24.4: time-travel query — delegate to as_of implementation
393 if as_of is not None:
394 require_time_travel_enabled(registry, surface="fact_query")
395 _validate_as_of(as_of)
396 with db() as conn:
397 result = _query_facts_as_of_impl(
398 conn,
399 entity=entity,
400 scope=scope,
401 relation=relation,
402 as_of=as_of,
403 is_admin_caller=identity.is_admin(),
404 tenant_id=identity.tenant_id,
405 read_scope=caller_read_scope(identity),
406 limit=limit,
407 cursor=cursor,
408 )
409 if result.total is not None:
410 response.headers["X-Total-Count"] = str(result.total)
411 with db() as conn:
412 record_read_scopes(
413 conn,
414 identity=identity,
415 session_id=session_id,
416 scopes={fact.scope for fact in result.facts},
417 )
418 registry.fire_fire_and_forget(
419 "post_recall_audit",
420 result=result,
421 identity=identity,
422 tenant=tenant,
423 request_id=request_id,
424 outcome=Success(),
425 )
426 return result
428 FACT_READ.labels(principal=identity.entity_uri, tenant=identity.tenant_id).inc()
430 # Garden ACL: resolve and enforce membership before querying (spec §5.20, §17.3)
431 garden = _resolve_garden_or_404(garden_id, identity)
433 with db() as conn:
434 garden_visibility_mode, exact_garden_id, visible_garden_ids = _resolve_garden_visibility(
435 conn, garden, identity
436 )
437 _prepare_garden_visibility_table(conn, visible_garden_ids)
438 params = _build_query_params(
439 identity=identity,
440 garden_visibility_mode=garden_visibility_mode,
441 exact_garden_id=exact_garden_id,
442 entity=entity,
443 relation=relation,
444 source=source,
445 scope=scope,
446 min_confidence=min_confidence,
447 attested=attested,
448 after=after,
449 cursor=cursor,
450 include_expired=include_expired,
451 limit=limit,
452 )
453 rows = conn.execute(_FACT_QUERY_SQL, params).fetchall()
455 has_more = len(rows) > limit
456 rows = rows[:limit]
458 records = _rows_to_records(rows)
459 if not include_contradicted:
460 records = [r for r in records if not r.contradicted]
462 # v1.1: apply recall-time trust multiplier + content sanitizer (§19.4.4, §19.7)
463 records = apply_recall_pipeline(records, identity=identity, include_low_trust=include_low_trust)
465 # §23.3: tombstone filter — must be applied after scope filtering, before packing
466 records, tombstone_filtered = _apply_tombstone_filter(records, scope, identity)
467 records = registry.fire_filter_chain(
468 "recall_filter",
469 records,
470 identity=identity,
471 tenant=tenant,
472 request_id=request_id,
473 )
474 score_deltas = registry.fire_score_delta(
475 "recall_rank",
476 records,
477 identity=identity,
478 tenant=tenant,
479 request_id=request_id,
480 )
481 if score_deltas: 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true
482 records = sorted(records, key=lambda record: score_deltas.get(record.id, 0.0), reverse=True)
484 next_cursor = rows[-1]["id"] if has_more and rows else None
485 # §23.3.3 r.3: suppress total when tombstone filtering was applied to prevent oracle leakage
486 total = None if tombstone_filtered else len(records)
487 result = QueryResponse(facts=records, total=total, cursor=next_cursor)
488 if result.total is not None:
489 response.headers["X-Total-Count"] = str(result.total)
490 with db() as conn:
491 record_read_scopes(
492 conn,
493 identity=identity,
494 session_id=session_id,
495 scopes={fact.scope for fact in records},
496 )
497 registry.fire_fire_and_forget(
498 "post_recall_audit",
499 result=result,
500 identity=identity,
501 tenant=tenant,
502 request_id=request_id,
503 outcome=Success(),
504 )
505 return result
508def _resolve_garden_or_404(garden_id: str | None, identity: Identity) -> Any:
509 """Return the garden row when ``garden_id`` is set; 404 if missing; enforce read ACL."""
510 if garden_id is None:
511 return None
512 garden = get_garden_by_garden_uri(garden_id, tenant_id=identity.tenant_id)
513 if garden is None: 513 ↛ 514line 513 didn't jump to line 514 because the condition on line 513 was never true
514 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="garden not found")
515 require_garden_read(garden, identity)
516 return garden
519def _resolve_garden_visibility(
520 conn: Any,
521 garden: Any,
522 identity: Identity,
523) -> tuple[int, str | None, list[str]]:
524 """Return query visibility mode, exact garden id, and visible garden id set."""
525 if garden is not None:
526 return _GARDEN_VISIBILITY_EXACT, garden["id"], []
527 if not garden_acl_enforced(): 527 ↛ 528line 527 didn't jump to line 528 because the condition on line 527 was never true
528 return _GARDEN_VISIBILITY_NONE, None, []
530 visible_garden_ids = [
531 row["id"]
532 for row in conn.execute(
533 "SELECT g.id FROM gardens g"
534 " WHERE g.tenant_id = ?"
535 " AND EXISTS ("
536 " SELECT 1 FROM garden_members gm"
537 " WHERE gm.garden_id = g.id AND gm.entity_uri = ?"
538 " )",
539 (identity.tenant_id, identity.entity_uri),
540 ).fetchall()
541 ]
542 if visible_garden_ids:
543 return _GARDEN_VISIBILITY_VISIBLE_SET, None, visible_garden_ids
544 return _GARDEN_VISIBILITY_NULL_ONLY, None, []
547def _prepare_garden_visibility_table(conn: Any, visible_garden_ids: list[str]) -> None:
548 """Populate the per-connection visibility table referenced by ``_FACT_QUERY_SQL``."""
549 conn.execute("CREATE TEMP TABLE IF NOT EXISTS _query_visible_gardens (id TEXT PRIMARY KEY)")
550 conn.execute("DELETE FROM _query_visible_gardens")
551 for garden_id in visible_garden_ids:
552 conn.execute("INSERT OR IGNORE INTO _query_visible_gardens (id) VALUES (?)", (garden_id,))
555def _normalise_uri_or_raw(raw_value: str) -> str:
556 """Best-effort URI normalisation; falls back to the raw value on failure."""
557 try: # noqa: SIM105
558 return normalize_entity_uri(raw_value)
559 except NormalizationError:
560 return raw_value # malformed — fall through to exact match
563def _build_query_params( # noqa: PLR0913 — narrow internal helper, keeps query_facts signature flat
564 *,
565 identity: Identity,
566 garden_visibility_mode: int,
567 exact_garden_id: str | None,
568 entity: str | None,
569 relation: str | None,
570 source: str | None,
571 scope: str | None,
572 min_confidence: float,
573 attested: bool | None,
574 after: str | None,
575 cursor: str | None,
576 include_expired: bool,
577 limit: int,
578) -> list[Any]:
579 """Return bind values for the fixed ``_FACT_QUERY_SQL`` template."""
580 if scope and scope not in VALID_SCOPES: 580 ↛ 581line 580 didn't jump to line 581 because the condition on line 580 was never true
581 raise HTTPException(status_code=400, detail=f"scope must be one of {VALID_SCOPES}")
583 normalised_entity = _normalise_uri_or_raw(entity) if entity else None
584 normalised_source = _normalise_uri_or_raw(source) if source else None
585 attested_value = None if attested is None else 1 if attested else 0
586 now = datetime.now(UTC).isoformat()
587 return [
588 min_confidence,
589 identity.tenant_id,
590 garden_visibility_mode,
591 garden_visibility_mode,
592 exact_garden_id,
593 garden_visibility_mode,
594 garden_visibility_mode,
595 attested_value,
596 attested_value,
597 normalised_entity,
598 normalised_entity,
599 normalised_entity,
600 identity.tenant_id, # entity-alias subquery tenant scope
601 relation or None,
602 relation or None,
603 normalised_source,
604 normalised_source,
605 normalised_source,
606 identity.tenant_id, # source-alias subquery tenant scope
607 scope or None,
608 scope or None,
609 after or None,
610 after or None,
611 cursor or None,
612 cursor or None,
613 1 if include_expired else 0,
614 now,
615 limit + 1,
616 ]
619def _rows_to_records(rows: list[Any]) -> list[FactRecord]:
620 """Convert raw rows into FactRecords; mark within-key duplicates contradicted."""
621 seen: dict[tuple[str, str, str], int] = {}
622 for r in rows:
623 enforce_read_path_cid(r)
624 key = (r["entity"], r["relation"], r["scope"])
625 seen[key] = seen.get(key, 0) + 1
626 return [
627 row_to_record(r, contradicted=seen[(r["entity"], r["relation"], r["scope"])] > 1)
628 for r in rows
629 ]
632def _apply_tombstone_filter(
633 records: list[FactRecord],
634 scope: str | None,
635 identity: Identity,
636) -> tuple[list[FactRecord], bool]:
637 """Return (filtered, tombstone_filtered_flag). Empty input → no work."""
638 if not records:
639 return records, False
640 entity_uris_in_result = list({r.entity for r in records})
641 with db() as _tc_conn:
642 excluded, _notices = _get_tombstone_filter(
643 _tc_conn,
644 entity_uris_in_result,
645 scope or "local",
646 identity.is_admin(),
647 identity.tenant_id,
648 )
649 if not excluded:
650 return records, False
651 return [r for r in records if r.entity not in excluded], True