Coverage for node / src / stigmem_node / lifecycle / immutability.py: 93%
48 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""ADR-016 L1 append-only journal and projection helpers."""
3from __future__ import annotations
5import json
6import uuid
7from datetime import UTC, datetime
8from typing import Any
11def utc_now_iso() -> str:
12 return datetime.now(UTC).replace(microsecond=0).isoformat()
15def write_fact_journal(
16 conn: Any,
17 *,
18 fact_id: str,
19 event_type: str,
20 tenant_id: str,
21 actor_uri: str | None,
22 source: str | None,
23 scope: str | None,
24 cid: str | None,
25 body: dict[str, Any],
26) -> None:
27 """Append one fact event to the immutable L1 journal."""
28 conn.execute(
29 "INSERT INTO fact_journal "
30 "(id, fact_id, event_type, event_ts, tenant_id, actor_uri, source, scope, cid, body_json) "
31 "VALUES (?,?,?,?,?,?,?,?,?,?)",
32 (
33 str(uuid.uuid4()),
34 fact_id,
35 event_type,
36 utc_now_iso(),
37 tenant_id,
38 actor_uri,
39 source,
40 scope,
41 cid,
42 json.dumps(body, sort_keys=True, separators=(",", ":")),
43 ),
44 )
47def set_embedding_status(
48 conn: Any,
49 *,
50 fact_id: str,
51 embedding_missing: bool,
52 updated_by: str | None = None,
53 last_error: str | None = None,
54) -> None:
55 """Upsert embedding status in the projection table, not on ``facts``."""
56 conn.execute(
57 "INSERT INTO fact_embedding_status "
58 "(fact_id, embedding_missing, updated_at, last_error, updated_by) "
59 "VALUES (?,?,?,?,?) "
60 "ON CONFLICT(fact_id) DO UPDATE SET "
61 "embedding_missing = excluded.embedding_missing, "
62 "updated_at = excluded.updated_at, "
63 "last_error = excluded.last_error, "
64 "updated_by = excluded.updated_by",
65 (
66 fact_id,
67 1 if embedding_missing else 0,
68 utc_now_iso(),
69 last_error,
70 updated_by,
71 ),
72 )
75def set_fact_validity_override(
76 conn: Any,
77 *,
78 fact_id: str,
79 valid_until: str | None = None,
80 confidence: float | None = None,
81 reason: str | None = None,
82 updated_by: str | None = None,
83) -> None:
84 """Upsert derived validity/confidence state outside the base fact row."""
85 conn.execute(
86 "INSERT INTO fact_validity_overrides "
87 "(fact_id, valid_until, confidence, reason, updated_at, updated_by) "
88 "VALUES (?,?,?,?,?,?) "
89 "ON CONFLICT(fact_id) DO UPDATE SET "
90 "valid_until = excluded.valid_until, "
91 "confidence = excluded.confidence, "
92 "reason = excluded.reason, "
93 "updated_at = excluded.updated_at, "
94 "updated_by = excluded.updated_by",
95 (fact_id, valid_until, confidence, reason, utc_now_iso(), updated_by),
96 )
99def set_fact_quarantine_status(
100 conn: Any,
101 *,
102 fact_id: str,
103 quarantine_garden_id: str | None,
104 quarantine_status: str | None,
105 quarantine_reason: str | None = None,
106 quarantine_acted_by: str | None = None,
107 quarantine_acted_at: str | None = None,
108) -> None:
109 """Upsert quarantine workflow state outside the base fact row."""
110 conn.execute(
111 "INSERT INTO fact_quarantine_status "
112 "(fact_id, quarantine_garden_id, quarantine_status, quarantine_reason, "
113 " quarantine_acted_by, quarantine_acted_at, updated_at) "
114 "VALUES (?,?,?,?,?,?,?) "
115 "ON CONFLICT(fact_id) DO UPDATE SET "
116 "quarantine_garden_id = excluded.quarantine_garden_id, "
117 "quarantine_status = excluded.quarantine_status, "
118 "quarantine_reason = excluded.quarantine_reason, "
119 "quarantine_acted_by = excluded.quarantine_acted_by, "
120 "quarantine_acted_at = excluded.quarantine_acted_at, "
121 "updated_at = excluded.updated_at",
122 (
123 fact_id,
124 quarantine_garden_id,
125 quarantine_status,
126 quarantine_reason,
127 quarantine_acted_by,
128 quarantine_acted_at,
129 utc_now_iso(),
130 ),
131 )
134def set_fact_garden_membership(
135 conn: Any,
136 *,
137 fact_id: str,
138 garden_id: str | None,
139 updated_by: str | None = None,
140) -> None:
141 """Upsert derived garden membership outside the base fact row."""
142 conn.execute(
143 "INSERT INTO fact_garden_membership "
144 "(fact_id, garden_id, updated_at, updated_by) "
145 "VALUES (?,?,?,?) "
146 "ON CONFLICT(fact_id) DO UPDATE SET "
147 "garden_id = excluded.garden_id, "
148 "updated_at = excluded.updated_at, "
149 "updated_by = excluded.updated_by",
150 (fact_id, garden_id, utc_now_iso(), updated_by),
151 )
154def set_fact_cid_backfill_status(
155 conn: Any,
156 *,
157 fact_id: str,
158 status: str,
159 error: str | None = None,
160) -> None:
161 """Record CID backfill progress without mutating ``facts.cid``."""
162 conn.execute(
163 "INSERT INTO fact_cid_backfill "
164 "(fact_id, status, attempted_at, error, updated_at) "
165 "VALUES (?,?,?,?,?) "
166 "ON CONFLICT(fact_id) DO UPDATE SET "
167 "status = excluded.status, "
168 "attempted_at = excluded.attempted_at, "
169 "error = excluded.error, "
170 "updated_at = excluded.updated_at",
171 (fact_id, status, utc_now_iso(), error, utc_now_iso()),
172 )
175def rebind_facts_to_cid_v2(conn: Any, *, batch_size: int = 500) -> dict[str, int]:
176 """Re-point every fact's CID alias to its CID v2 (binds ``interpret_as``).
178 CID v2 added ``interpret_as`` to the canonical body. The immutable
179 ``facts.cid`` column is left untouched; the rebuildable ``fact_cid_aliases``
180 projection — which ``projected_cid`` reads and which the read-path CID
181 verification prefers — is repointed to the v2 CID so migrated facts verify
182 under CID v2. The pre-v2 (v1) alias is removed, so a lookup by a fact's old
183 v1 CID stops resolving (the accepted pre-1.0 clean break).
185 Idempotent and collision-safe. Returns ``{"rebound": n, "skipped_collision": n}``.
186 """
187 from ..cid import compute_cid
189 rebound = 0
190 skipped_collision = 0
191 last_id: str | None = None
192 while True:
193 if last_id is None:
194 rows = conn.execute(
195 "SELECT id, entity, relation, value_type, value_v, source, scope, "
196 "confidence, interpret_as, tenant_id FROM facts ORDER BY id LIMIT ?",
197 (batch_size,),
198 ).fetchall()
199 else:
200 rows = conn.execute(
201 "SELECT id, entity, relation, value_type, value_v, source, scope, "
202 "confidence, interpret_as, tenant_id FROM facts WHERE id > ? ORDER BY id LIMIT ?",
203 (last_id, batch_size),
204 ).fetchall()
205 if not rows:
206 break
207 for row in rows:
208 last_id = row["id"]
209 v2 = compute_cid(
210 entity=row["entity"],
211 relation=row["relation"],
212 value_type=row["value_type"],
213 value_v=row["value_v"] or "",
214 source=row["source"],
215 scope=row["scope"],
216 confidence=float(row["confidence"]),
217 interpret_as=(row["interpret_as"] or "content"),
218 )
219 existing = {
220 c["cid"]
221 for c in conn.execute(
222 "SELECT cid FROM fact_cid_aliases WHERE fact_id = ?", (row["id"],)
223 ).fetchall()
224 }
225 if existing == {v2}:
226 set_fact_cid_backfill_status(conn, fact_id=row["id"], status="complete")
227 continue
228 # Tenant-scoped collision check: alias uniqueness is per (cid, tenant_id)
229 # after migration 052 (F-SBOLA4).
230 owner = conn.execute(
231 "SELECT fact_id FROM fact_cid_aliases WHERE cid = ? AND tenant_id = ?",
232 (v2, row["tenant_id"]),
233 ).fetchone()
234 if owner is not None and owner["fact_id"] != row["id"]: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 set_fact_cid_backfill_status(
236 conn, fact_id=row["id"], status="skipped", error="cid_v2_collision"
237 )
238 skipped_collision += 1
239 continue
240 conn.execute("DELETE FROM fact_cid_aliases WHERE fact_id = ?", (row["id"],))
241 conn.execute(
242 "INSERT OR IGNORE INTO fact_cid_aliases (fact_id, cid, tenant_id) VALUES (?, ?, ?)",
243 (row["id"], v2, row["tenant_id"]),
244 )
245 set_fact_cid_backfill_status(conn, fact_id=row["id"], status="complete")
246 rebound += 1
247 conn.commit()
248 return {"rebound": rebound, "skipped_collision": skipped_collision}