Coverage for node / src / stigmem_node / routes / facts / common.py: 78%
84 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Shared helpers for fact route modules."""
3from __future__ import annotations
5import logging
6import uuid
7from datetime import UTC, datetime
8from typing import Any
10from fastapi import APIRouter
12from ...hlc import node_hlc
13from ...models.tombstones import TombstoneNotice
15logger = logging.getLogger("stigmem.facts")
17FACT_PROJECTION_SELECT = (
18 "f.*, "
19 "COALESCE(fvo.valid_until, f.valid_until) AS projected_valid_until, "
20 "COALESCE(fvo.confidence, f.confidence) AS projected_confidence, "
21 "COALESCE(fgm.garden_id, f.garden_id) AS projected_garden_id, "
22 "COALESCE(fqs.quarantine_status, f.quarantine_status) AS projected_quarantine_status, "
23 "COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id) "
24 "AS projected_quarantine_garden_id, "
25 "COALESCE(f.cid, (SELECT fca.cid FROM fact_cid_aliases fca "
26 "WHERE fca.fact_id = f.id ORDER BY fca.cid LIMIT 1)) AS projected_cid"
27)
29FACT_PROJECTION_JOINS = (
30 " LEFT JOIN fact_validity_overrides fvo ON fvo.fact_id = f.id"
31 " LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id"
32 " LEFT JOIN fact_quarantine_status fqs ON fqs.fact_id = f.id"
33)
35__all__ = [
36 "FACT_PROJECTION_JOINS",
37 "FACT_PROJECTION_SELECT",
38 "router",
39]
42def _get_tombstone_filter(
43 conn: Any,
44 entity_uris: list[str],
45 scope: str,
46 is_admin_caller: bool,
47 tenant_id: str,
48) -> tuple[set[str], list[TombstoneNotice]]:
49 """Return (excluded_entity_uris, tombstone_notices) for entity_uris in scope (§23.3, §24.3).
51 excluded_entity_uris: entities under active (non-legal-hold) tombstones.
52 tombstone_notices: annotations for legal_hold tombstones visible to admin callers.
54 Suppression is scoped to ``tenant_id`` (the caller's tenant): only tombstones in
55 the caller's own tenant partition may suppress the caller's facts. A tombstone in
56 a different tenant MUST NOT hide this caller's content (R-3 / F-SBOLA3). Single-tenant
57 nodes pass ``tenant_id="default"``, matching the ``default`` rows create_tombstone writes.
58 """
59 from ...lifecycle.tombstone_gate import tombstone_filter_enabled
61 if not entity_uris or not tombstone_filter_enabled():
62 return set(), []
64 placeholders = ",".join("?" * len(entity_uris))
65 # BEGIN IMMEDIATE for SQLite consistency (§23.3.3 rule 5).
66 # On postgres this is a syntax error; rollback clears the failed txn state.
67 try:
68 conn.execute("BEGIN IMMEDIATE")
69 except Exception as exc: # nosec B110
70 logger.debug(
71 "BEGIN IMMEDIATE not supported or failed for tombstone filter transaction "
72 "(expected on some backends, e.g. postgres); continuing with default "
73 "transaction behavior: %s",
74 exc,
75 )
76 try: # noqa: SIM105
77 conn.rollback()
78 except Exception as rollback_exc: # nosec B110
79 logger.debug(
80 "Rollback after BEGIN IMMEDIATE failure also failed; continuing because "
81 "explicit BEGIN IMMEDIATE is optional in this path: %s",
82 rollback_exc,
83 )
84 rows = conn.execute(
85 # Same-issuer binding: only a revocation from the tombstone's OWN issuer
86 # (r.signed_by = t.signed_by) lifts the suppression — a forged/cross-issuer
87 # revocation can never un-suppress content another org tombstoned (RTBF integrity).
88 # Tenant predicate (R-3 / F-SBOLA3): only a tombstone in the CALLER's own tenant
89 # may suppress the caller's facts — a different tenant's tombstone must not hide them.
90 f"""SELECT t.id, t.entity_uri, t.scope, t.created_at, t.legal_hold
91 FROM tombstones t
92 WHERE t.entity_uri IN ({placeholders})
93 AND t.tenant_id = ?
94 AND NOT EXISTS (
95 SELECT 1 FROM tombstone_revocations r
96 WHERE r.tombstone_id = t.id AND r.signed_by = t.signed_by
97 )""", # noqa: S608 # nosec B608 - dynamic SQL is generated placeholders only; entity values are bound params.
98 [*entity_uris, tenant_id],
99 ).fetchall()
100 try: # noqa: SIM105
101 conn.execute("COMMIT")
102 except Exception as exc: # nosec B110
103 logger.debug(
104 "Tombstone filter COMMIT skipped or failed (can occur when explicit "
105 "BEGIN IMMEDIATE was unavailable on this backend); continuing: %s",
106 exc,
107 )
109 excluded: set[str] = set()
110 notices: list[TombstoneNotice] = []
112 for row in rows:
113 uri = row["entity_uri"]
114 row_scope = row["scope"]
115 if row_scope != "*" and row_scope != scope: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true
116 continue
118 if row["legal_hold"]:
119 if is_admin_caller:
120 notices.append(
121 TombstoneNotice(
122 entity_uri=uri,
123 tombstone_id=row["id"],
124 legal_hold=True,
125 tombstone_created_at=row["created_at"],
126 )
127 )
128 else:
129 excluded.add(uri)
130 else:
131 excluded.add(uri)
133 return excluded, notices
136router = APIRouter(prefix="/v1/facts", tags=["facts"])
138_SYSTEM_RELATION_PREFIX = "stigmem:"
141def _validate_relation(relation: str) -> list[str]:
142 """Return convention warnings for a relation name (see relation-convention.md)."""
143 if ":" not in relation:
144 return [
145 f"bare relation {relation!r} has no namespace prefix; "
146 f"rename to 'your-prefix:{relation}' to prevent silent collisions "
147 "(see relation-convention.md)"
148 ]
149 if relation.startswith(_SYSTEM_RELATION_PREFIX):
150 return [
151 f"relation {relation!r} uses reserved system prefix 'stigmem:'; "
152 "non-system callers should use a custom namespace prefix (see spec §9.1)"
153 ]
154 return []
157def _is_valid_entity_uri(uri: str) -> bool:
158 """Return whether a ref value is eligible for graph edge derivation."""
159 return "://" in uri or uri.startswith("urn:")
162def _embed_fact_background(
163 fact_id: str,
164 entity: str,
165 relation: str,
166 value_type: str,
167 value_v: str,
168) -> None:
169 """Background thread: embed one fact and persist to vec_facts."""
170 try:
171 from ... import settings as settings_pkg
172 from ...db import db
173 from ...embedding import get_embedding_model
174 from ...recall.vector_search import check_or_register_model, embed_and_store_fact
176 model = get_embedding_model(settings_pkg.settings)
177 with db() as conn:
178 check_or_register_model(conn, model.model_id, model.dimension)
179 embed_and_store_fact(fact_id, entity, relation, value_type, value_v, conn, model)
180 except Exception as exc:
181 logger.warning("Write-time embedding failed for fact %s: %s", fact_id, exc)
184def _record_contradictions(
185 conn: Any,
186 new_fact_id: str,
187 entity: str,
188 relation: str,
189 scope: str,
190 siblings: list[Any],
191 tenant_id: str = "default",
192) -> None:
193 """Write conflict entities and conflicts table rows for new contradictions."""
194 now = datetime.now(UTC).isoformat()
195 for sibling in siblings:
196 sibling_id = sibling["id"]
198 already = conn.execute(
199 """SELECT id FROM conflicts
200 WHERE (fact_a_id=? AND fact_b_id=?) OR (fact_a_id=? AND fact_b_id=?)""",
201 (new_fact_id, sibling_id, sibling_id, new_fact_id),
202 ).fetchone()
203 if already: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 continue
206 conflict_id = f"stigmem:conflict:{uuid.uuid4()}"
207 h_between = node_hlc.tick()
208 conn.execute(
209 """INSERT INTO facts
210 (id, entity, relation, value_type, value_v, source, timestamp,
211 valid_until, confidence, scope, hlc, received_from, tenant_id)
212 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
213 (
214 str(uuid.uuid4()),
215 conflict_id,
216 "stigmem:conflict:between",
217 "text",
218 f"{new_fact_id} {sibling_id}",
219 "system:stigmem",
220 now,
221 None,
222 1.0,
223 scope,
224 h_between,
225 None,
226 tenant_id,
227 ),
228 )
229 h_status = node_hlc.tick()
230 conn.execute(
231 """INSERT INTO facts
232 (id, entity, relation, value_type, value_v, source, timestamp,
233 valid_until, confidence, scope, hlc, received_from, tenant_id)
234 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
235 (
236 str(uuid.uuid4()),
237 conflict_id,
238 "stigmem:conflict:status",
239 "string",
240 "unresolved",
241 "system:stigmem",
242 now,
243 None,
244 1.0,
245 scope,
246 h_status,
247 None,
248 tenant_id,
249 ),
250 )
251 conn.execute(
252 """INSERT OR IGNORE INTO conflicts (id, fact_a_id, fact_b_id, status, detected_at)
253 VALUES (?,?,?,?,?)""",
254 (conflict_id, new_fact_id, sibling_id, "unresolved", now),
255 )
258def _encode_v(vtype: str, v: Any) -> str:
259 if vtype == "null":
260 return "null"
261 if vtype == "boolean":
262 return "true" if v else "false"
263 return str(v)