Coverage for node / src / stigmem_node / routes / _facts_assert.py: 92%
162 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Implementation of POST /v1/facts (assert_fact) extracted from routes/facts.py.
3Imported back into ``routes.facts``; the route stub there delegates to this
4function inside its tracing span. Helper symbols are imported lazily inside
5the function to keep the module-level import graph acyclic.
6No behavioural changes — code was moved verbatim from facts.py.
7"""
9from __future__ import annotations
11import logging
12import sys
13import threading
14import uuid
15from datetime import UTC, datetime
16from typing import Any
18from fastapi import HTTPException, status
20from ..auth import Identity
21from ..billing import BillingEvent, get_hook_bus
22from ..cid import compute_cid
23from ..db import db
24from ..entity_normalizer import NormalizationError, is_informal, normalize_entity_uri
25from ..garden_acl import (
26 get_garden_by_garden_uri,
27 require_garden_write,
28)
29from ..hlc import node_hlc
30from ..lifecycle.immutability import set_embedding_status, write_fact_journal
31from ..metrics import CONTRADICTION, FACT_WRITE
32from ..models.facts import AssertRequest, FactRecord, row_to_record
33from ..plugins import TenantContext, get_registry
34from ..recall.fuzzy_resolver import resolve_entity
35from ..session_graph import encode_derived_from, ensure_write_allowed, record_write_scope
36from ..settings import settings as _settings # noqa: F401 — kept for parity
39def _live_settings() -> Any:
40 """Return the live Settings singleton.
42 Uses sys.modules directly because some test fixtures replace
43 `stigmem_node.settings` (the module attribute on the parent package) with
44 a Settings instance. `from .. import settings` and `import x.y` both go
45 through that patched attribute and would return the instance instead of
46 the module — sys.modules['stigmem_node.settings'] is the only path that
47 reliably reaches the original module so we can read its `.settings`
48 singleton (which IS what tests intend to swap).
49 """
50 return sys.modules["stigmem_node.settings"].settings
53logger = logging.getLogger("stigmem.facts")
56def _verify_or_require_attestation(req: AssertRequest, identity: Identity) -> str | None:
57 """C1: verify the attestation token (when supplied) or fail-closed if required."""
58 from .facts import _encode_v
60 if req.attestation is not None:
61 from .agent_keys import verify_attestation
63 value_v_for_sig = _encode_v(req.value.type, req.value.v)
64 canonical = (
65 f"{req.entity}\n{req.relation}\n{req.value.type}\n{value_v_for_sig}\n{req.source}"
66 ).encode()
67 return verify_attestation(
68 key_id=req.attestation.key_id,
69 signature_b64=req.attestation.signature,
70 canonical_message=canonical,
71 caller_entity_uri=identity.entity_uri,
72 )
73 if _live_settings().attestation_required:
74 raise HTTPException(
75 status_code=status.HTTP_400_BAD_REQUEST,
76 detail="attestation required; register an agent key at POST /v1/auth/agent-keys",
77 )
78 return None
81def _normalise_and_alias_uris(req: AssertRequest, tenant_id: str) -> tuple[str, str]:
82 """Layer-1 strict normalisation + Layer-2 alias lookup. Emits deprecation warnings."""
83 try:
84 entity = normalize_entity_uri(req.entity)
85 source = normalize_entity_uri(req.source)
86 except NormalizationError as exc:
87 raise HTTPException(
88 status_code=status.HTTP_400_BAD_REQUEST,
89 detail=f"invalid_entity_uri: {exc}",
90 ) from exc
92 # Deprecation warning for informal URIs (spec §2.5)
93 if is_informal(req.entity):
94 print(
95 f"[stigmem] DEPRECATED: informal entity URI {req.entity!r} — "
96 f"use stigmem://authority/type/id format (spec §2.5)",
97 file=sys.stderr,
98 )
99 if is_informal(req.source):
100 print(
101 f"[stigmem] DEPRECATED: informal source URI {req.source!r} — "
102 f"use stigmem://authority/type/id format (spec §2.5)",
103 file=sys.stderr,
104 )
106 # Layer 2: resolve user-defined semantic aliases (spec §2.6.6) on canonical
107 # forms, scoped to the caller's tenant.
108 with db() as _alias_conn:
109 return (
110 resolve_entity(_alias_conn, entity, tenant_id),
111 resolve_entity(_alias_conn, source, tenant_id),
112 )
115def _resolve_garden_for_assert(req: AssertRequest, identity: Identity) -> Any:
116 """Spec §17.3: resolve garden_id, enforce scope match + write ACL. Returns row or None."""
117 if req.garden_id is None:
118 return None
119 garden = get_garden_by_garden_uri(req.garden_id, tenant_id=identity.tenant_id)
120 if garden is None:
121 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="garden not found")
122 if garden["scope"] != req.scope:
123 raise HTTPException(
124 status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
125 detail=(
126 f"scope mismatch: garden scope is '{garden['scope']}' "
127 f"but fact scope is '{req.scope}'"
128 ),
129 )
130 require_garden_write(garden, identity)
131 return garden
134def _existing_record_for_cid(
135 conn: Any,
136 fact_cid: str,
137 tenant_id: str,
138) -> FactRecord | None:
139 """Return the existing record for ``fact_cid``, or None when no alias exists yet."""
140 existing_alias = conn.execute(
141 "SELECT fact_id FROM fact_cid_aliases WHERE cid = ? AND tenant_id = ?",
142 (fact_cid, tenant_id),
143 ).fetchone()
144 if existing_alias is None:
145 return None
146 existing_row = conn.execute(
147 "SELECT * FROM facts WHERE id = ? AND tenant_id = ?",
148 (existing_alias["fact_id"], tenant_id),
149 ).fetchone()
150 return row_to_record(existing_row, contradicted=False) if existing_row is not None else None
153def _require_interpretation_write(identity: Identity, interpret_as: str) -> None:
154 if interpret_as != "instruction":
155 return
156 if identity.can_write_instruction():
157 return
158 raise HTTPException(
159 status_code=status.HTTP_403_FORBIDDEN,
160 detail={
161 "code": "instruction_write_required",
162 "message": "writing instruction-typed facts requires instruction:write permission",
163 },
164 )
167def _detect_and_record_contradictions(
168 conn: Any,
169 fact_id: str,
170 entity: str,
171 req: AssertRequest,
172 identity: Identity,
173) -> bool:
174 """Spec §9.1: skip system facts; else find siblings sharing (entity, relation, scope)."""
175 from .facts import _SYSTEM_RELATION_PREFIX, _record_contradictions
177 is_system = (
178 entity.startswith(_SYSTEM_RELATION_PREFIX) and not entity.startswith("stigmem://")
179 ) or (
180 req.relation.startswith(_SYSTEM_RELATION_PREFIX)
181 and not req.relation.startswith("stigmem://")
182 )
183 if is_system:
184 return False
185 siblings = conn.execute(
186 """SELECT id FROM facts
187 WHERE entity=? AND relation=? AND scope=? AND id!=? AND confidence>0.0
188 AND tenant_id=?""",
189 (entity, req.relation, req.scope, fact_id, identity.tenant_id),
190 ).fetchall()
191 if not siblings:
192 return False
193 _record_contradictions(
194 conn,
195 fact_id,
196 entity,
197 req.relation,
198 req.scope,
199 siblings,
200 identity.tenant_id,
201 )
202 print(
203 f"[stigmem] WARN: collision — entity={entity!r} relation={req.relation!r} "
204 f"scope={req.scope!r}: fact {fact_id!r} contradicts {len(siblings)} existing "
205 f"fact(s); verify relation namespacing (see relation-convention.md)",
206 file=sys.stderr,
207 )
208 return True
211def _emit_post_write_hooks(
212 *,
213 fact_id: str,
214 entity: str,
215 source: str,
216 req: AssertRequest,
217 identity: Identity,
218 value_v: str | None,
219 garden_uuid: str | None,
220 now: str,
221 contradicted: bool,
222 _span: object,
223) -> None:
224 """Card-stale + background embed + billing + subscription fan-out + metrics + span attrs."""
225 from .facts import _embed_fact_background
227 # Phase 9: mark entity's memory card stale on every write (ACM-214)
228 try:
229 from ..card_materializer import mark_entity_stale as _mark_stale
231 _mark_stale(entity, req.scope, identity.tenant_id)
232 except Exception as _card_exc:
233 logger.warning("card mark_stale failed for %r: %s", entity, _card_exc)
235 # Phase 9 §2: write-time embedding (background thread, graceful fallback)
236 if _live_settings().embed_enabled: 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true
237 threading.Thread(
238 target=_embed_fact_background,
239 args=(fact_id, entity, req.relation, req.value.type, value_v or ""),
240 daemon=True,
241 ).start()
243 get_hook_bus().emit(
244 BillingEvent(
245 event_type="fact_written",
246 tenant_id=identity.tenant_id,
247 entity_uri=identity.entity_uri,
248 fact_id=fact_id,
249 )
250 )
252 # §20: fan out to subscribers (fast DB insert only; delivery happens in sweep loop)
253 try:
254 import json as _json
256 from ..subscription_delivery import fan_out as _subscription_fan_out
258 _subscription_fan_out(
259 fact_id=fact_id,
260 entity=entity,
261 scope=req.scope,
262 garden_id=garden_uuid,
263 tenant_id=identity.tenant_id,
264 fact_payload_json=_json.dumps(
265 {
266 "id": fact_id,
267 "entity": entity,
268 "relation": req.relation,
269 "value_type": req.value.type,
270 "value_v": value_v,
271 "source": source,
272 "timestamp": now,
273 "scope": req.scope,
274 "confidence": req.confidence,
275 "garden_id": garden_uuid,
276 }
277 ),
278 )
279 except Exception as _sub_exc:
280 print(f"[stigmem] WARN: subscription fan_out failed: {_sub_exc}", file=sys.stderr)
282 FACT_WRITE.labels(principal=identity.entity_uri, tenant=identity.tenant_id).inc()
283 if contradicted:
284 CONTRADICTION.labels(tenant=identity.tenant_id).inc()
285 try:
286 _span.set_attribute("stigmem.fact_id", fact_id) # type: ignore[attr-defined]
287 _span.set_attribute("stigmem.contradicted", contradicted) # type: ignore[attr-defined]
288 except AttributeError as _span_exc:
289 logger.debug("span attribute set skipped: %s", _span_exc)
292def assert_fact_impl(
293 req: AssertRequest,
294 identity: Identity,
295 _span: object,
296 *,
297 request_id: str,
298 tenant: TenantContext,
299 session_id: str | None = None,
300) -> FactRecord:
301 # Lazy imports of sibling helpers to avoid circular import with .facts
302 from .facts import (
303 _encode_v,
304 _is_valid_entity_uri,
305 _validate_relation,
306 )
308 if not identity.can_write():
309 raise HTTPException(
310 status_code=status.HTTP_403_FORBIDDEN,
311 detail="write permission required",
312 )
314 attested_key_id = _verify_or_require_attestation(req, identity)
315 entity, source = _normalise_and_alias_uris(req, identity.tenant_id)
317 # P-INJ-1: a fact's source is attested when it matches the writing principal.
318 # Mismatches are flagged (attested=False); enforce mode rejects them.
319 from ..source_attestation import (
320 evaluate_source_attested,
321 source_attestation_enforce_enabled,
322 )
324 attested = evaluate_source_attested(source, identity)
325 if attested is False and source_attestation_enforce_enabled():
326 raise HTTPException(
327 status_code=status.HTTP_403_FORBIDDEN,
328 detail=(
329 "source_attestation_failed: declared source does not match the "
330 "authenticated principal"
331 ),
332 )
333 garden = _resolve_garden_for_assert(req, identity)
335 garden_uuid = garden["id"] if garden is not None else None
336 attested_int = None if attested is None else (1 if attested else 0)
338 # Relation namespacing convention check (see relation-convention.md)
339 relation_warnings = _validate_relation(req.relation)
340 for w in relation_warnings:
341 print(f"[stigmem] WARN: relation naming: {w}", file=sys.stderr)
343 _require_interpretation_write(identity, req.value.interpret_as)
345 fact_id = str(uuid.uuid4())
346 now = datetime.now(UTC).isoformat()
347 hlc = node_hlc.tick()
348 value_v = _encode_v(req.value.type, req.value.v)
350 # F-AVAIL-1: cap a single fact value to bound storage/DoS. 0 disables.
351 _value_cap = _live_settings().max_fact_value_bytes
352 if _value_cap > 0 and value_v is not None and len(value_v.encode("utf-8")) > _value_cap:
353 raise HTTPException(
354 status_code=413,
355 detail=f"fact value exceeds {_value_cap} bytes",
356 )
358 # §25.7.3: compute CID before write; persisted in the same transaction
359 fact_cid = compute_cid(
360 entity=entity,
361 relation=req.relation,
362 value_type=req.value.type,
363 value_v=value_v or "",
364 source=source,
365 scope=req.scope,
366 confidence=req.confidence,
367 interpret_as=req.value.interpret_as,
368 )
370 _embed_enabled = _live_settings().embed_enabled
371 embedding_missing_val = 1 if _embed_enabled else None
373 derived_from_json = encode_derived_from(req.derived_from)
375 # F-10 §25.7.3: idempotent CID pre-check — if CID already exists, return existing record
376 with db() as _precheck_conn:
377 ensure_write_allowed(
378 _precheck_conn,
379 identity=identity,
380 session_id=session_id,
381 target_scope=req.scope,
382 write_mode=req.write_mode,
383 derived_from=req.derived_from,
384 )
385 existing_record = _existing_record_for_cid(_precheck_conn, fact_cid, identity.tenant_id)
386 if existing_record is not None:
387 return existing_record
389 with db() as conn:
390 ensure_write_allowed(
391 conn,
392 identity=identity,
393 session_id=session_id,
394 target_scope=req.scope,
395 write_mode=req.write_mode,
396 derived_from=req.derived_from,
397 )
398 conn.execute(
399 """INSERT INTO facts
400 (id, entity, relation, value_type, value_v, source, timestamp,
401 valid_until, confidence, scope, hlc, received_from, attested_key_id,
402 garden_id, attested, tenant_id, embedding_missing, cid, derived_from,
403 interpret_as)
404 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
405 (
406 fact_id,
407 entity, # normalized (spec §2.6)
408 req.relation,
409 req.value.type,
410 value_v,
411 source, # normalized (spec §2.6)
412 now,
413 req.valid_until,
414 req.confidence,
415 req.scope,
416 hlc,
417 None, # local write; not received from a peer
418 attested_key_id,
419 garden_uuid,
420 attested_int,
421 identity.tenant_id,
422 embedding_missing_val,
423 fact_cid,
424 derived_from_json,
425 req.value.interpret_as,
426 ),
427 )
428 write_fact_journal(
429 conn,
430 fact_id=fact_id,
431 event_type="fact_insert",
432 tenant_id=identity.tenant_id,
433 actor_uri=identity.entity_uri,
434 source=source,
435 scope=req.scope,
436 cid=fact_cid,
437 body={
438 "entity": entity,
439 "relation": req.relation,
440 "value_type": req.value.type,
441 "value_v": value_v,
442 "source": source,
443 "timestamp": now,
444 "valid_until": req.valid_until,
445 "confidence": req.confidence,
446 "scope": req.scope,
447 "interpret_as": req.value.interpret_as,
448 },
449 )
450 if embedding_missing_val is not None: 450 ↛ 451line 450 didn't jump to line 451 because the condition on line 450 was never true
451 set_embedding_status(
452 conn,
453 fact_id=fact_id,
454 embedding_missing=bool(embedding_missing_val),
455 updated_by="fact_assert",
456 )
458 # F-10 §25.7.3: alias table row — idempotent upsert on CID collision
459 alias_result = conn.execute(
460 "INSERT OR IGNORE INTO fact_cid_aliases (fact_id, cid, tenant_id) VALUES (?, ?, ?)",
461 (fact_id, fact_cid, identity.tenant_id),
462 )
463 if alias_result.rowcount == 0: 463 ↛ 465line 463 didn't jump to line 465 because the condition on line 463 was never true
464 # Concurrent same-CID write race: return existing record
465 existing = conn.execute(
466 "SELECT f.* FROM facts f JOIN fact_cid_aliases a ON a.fact_id = f.id"
467 " WHERE a.cid = ? AND f.tenant_id = ?",
468 (fact_cid, identity.tenant_id),
469 ).fetchone()
470 if existing is not None:
471 return row_to_record(existing, contradicted=False)
473 record_write_scope(
474 conn,
475 identity=identity,
476 session_id=session_id,
477 scope=req.scope,
478 )
480 # C3 / §22.3: write-ahead audit entry for fact_write event (same transaction)
481 from ..audit_event import emit as _emit_audit
483 _emit_audit(
484 "fact_write",
485 entity_uri=identity.entity_uri,
486 tenant_id=identity.tenant_id,
487 oidc_sub=identity.oidc_sub,
488 fact_id=fact_id,
489 source=source,
490 attested_key_id=attested_key_id,
491 scope=req.scope,
492 conn=conn,
493 )
495 # Graph adjacency index (§20.1.1): materialize edge for ref-typed facts
496 if req.value.type == "ref" and value_v and _is_valid_entity_uri(value_v):
497 from ..recall.graph_index import upsert_edge as _upsert_edge
499 _upsert_edge(
500 conn,
501 fact_id=fact_id,
502 subject=entity,
503 relation=req.relation,
504 object_uri=value_v,
505 scope=req.scope,
506 confidence=req.confidence,
507 garden_id=garden_uuid,
508 tenant_id=identity.tenant_id,
509 received_from=None,
510 source_trust=None,
511 valid_until=req.valid_until,
512 )
514 row = conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone()
515 from ..fact_chain import append_fact_chain_entry
517 append_fact_chain_entry(conn, row)
518 persisted_record = row_to_record(row, contradicted=False)
519 get_registry().fire_fire_and_forget(
520 "post_assert_persist",
521 fact=persisted_record,
522 identity=identity,
523 tenant=tenant,
524 request_id=request_id,
525 )
526 contradicted = _detect_and_record_contradictions(conn, fact_id, entity, req, identity)
528 _emit_post_write_hooks(
529 fact_id=fact_id,
530 entity=entity,
531 source=source,
532 req=req,
533 identity=identity,
534 value_v=value_v,
535 garden_uuid=garden_uuid,
536 now=now,
537 contradicted=contradicted,
538 _span=_span,
539 )
541 return row_to_record(row, contradicted=contradicted, warnings=relation_warnings)