Coverage for node / src / stigmem_node / routes / federation / audit_conflicts.py: 85%
108 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Federation audit and conflict routes."""
3from __future__ import annotations
5import json
6import uuid
7from datetime import UTC, datetime
8from typing import Annotated, Any
10from fastapi import Depends, HTTPException, Query, status
12from ...auth import Identity, resolve_identity
13from ...db import db
14from ...hlc import node_hlc
15from ...models.facts import row_to_record
16from ...models.federation import ConflictResolveRequest
17from .common import router
20@router.get("/v1/federation/audit")
21def get_audit_log(
22 identity: Annotated[Identity, Depends(resolve_identity)],
23 peer_id: str | None = Query(None),
24 event_type: str | None = Query(None),
25 limit: int = Query(50, ge=1, le=500),
26 cursor: str | None = Query(None),
27) -> dict[str, Any]:
28 if not identity.can_federate(): 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true
29 raise HTTPException(status_code=403, detail="federate permission required")
31 conditions: list[str] = []
32 params: list[Any] = []
33 if peer_id:
34 conditions.append("peer_id = ?")
35 params.append(peer_id)
36 if event_type:
37 conditions.append("event_type = ?")
38 params.append(event_type)
39 if cursor: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true
40 conditions.append("id > ?")
41 params.append(cursor)
43 where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
44 params.append(limit + 1)
46 with db() as conn:
47 rows = conn.execute(
48 f"SELECT * FROM federation_audit {where} ORDER BY ts DESC, id DESC LIMIT ?", # noqa: S608 # nosec B608 — where built from literal fragments; values in params
49 params,
50 ).fetchall()
52 has_more = len(rows) > limit
53 rows = rows[:limit]
54 next_cursor = rows[-1]["id"] if has_more and rows else None
56 return {
57 "entries": [
58 {
59 "id": r["id"],
60 "peer_id": r["peer_id"],
61 "event_type": r["event_type"],
62 "detail": json.loads(r["detail"]) if r["detail"] else None,
63 "ts": r["ts"],
64 }
65 for r in rows
66 ],
67 "cursor": next_cursor,
68 "has_more": has_more,
69 }
72# ---------------------------------------------------------------------------
73# GET /v1/conflicts — list conflicts (§5.9)
74# ---------------------------------------------------------------------------
77@router.get("/v1/conflicts")
78def list_conflicts(
79 identity: Annotated[Identity, Depends(resolve_identity)],
80 conflict_status: str | None = Query(None, alias="status"),
81 cursor: str | None = Query(None),
82 limit: int = Query(50, ge=1, le=500),
83) -> dict[str, Any]:
84 if not identity.can_read(): 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 raise HTTPException(status_code=403, detail="read permission required")
87 # F-FED-CONFLICT-TENANT: the conflicts table carries no tenant_id, but the
88 # facts it references do (ingest stamps tenant on both the facts and the
89 # conflict facts). Scope the operator view to the caller's tenant by requiring
90 # the conflict's fact_a to live in identity.tenant_id — otherwise an admin
91 # scoped to tenant A could enumerate a conflict (and its facts) from tenant B.
92 conditions: list[str] = [
93 "EXISTS (SELECT 1 FROM facts f WHERE f.id = c.fact_a_id AND f.tenant_id = ?)"
94 ]
95 params: list[Any] = [identity.tenant_id]
96 if conflict_status:
97 conditions.append("c.status = ?")
98 params.append(conflict_status)
99 if cursor: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 conditions.append("c.id > ?")
101 params.append(cursor)
103 where = f"WHERE {' AND '.join(conditions)}"
104 params.append(limit + 1)
106 with db() as conn:
107 sql = (
108 "SELECT c.id, c.fact_a_id, c.fact_b_id, c.status, c.resolution_fact_id, " # noqa: S608 # nosec B608
109 f"c.detected_at FROM conflicts c {where} ORDER BY c.detected_at DESC, "
110 "c.id DESC LIMIT ?"
111 )
112 rows = conn.execute(
113 sql,
114 params,
115 ).fetchall()
117 conflicts: list[dict[str, Any]] = []
118 for r in rows[:limit]:
119 fa = conn.execute(
120 "SELECT * FROM facts WHERE id = ? AND tenant_id = ?",
121 (r["fact_a_id"], identity.tenant_id),
122 ).fetchone()
123 fb = conn.execute(
124 "SELECT * FROM facts WHERE id = ? AND tenant_id = ?",
125 (r["fact_b_id"], identity.tenant_id),
126 ).fetchone()
127 conflicts.append(
128 {
129 "conflict_id": r["id"],
130 "fact_a": row_to_record(fa).model_dump() if fa else None,
131 "fact_b": row_to_record(fb).model_dump() if fb else None,
132 "status": r["status"],
133 "resolved_by": r["resolution_fact_id"],
134 "detected_at": r["detected_at"],
135 }
136 )
138 has_more = len(rows) > limit
139 next_cursor = rows[limit - 1]["id"] if has_more and len(rows) >= limit else None
140 return {"conflicts": conflicts, "cursor": next_cursor, "has_more": has_more}
143# ---------------------------------------------------------------------------
144# POST /v1/conflicts/:conflict_id/resolve — resolve a conflict (§5.10)
145# ---------------------------------------------------------------------------
148def _encode_value(vtype: str, v: Any) -> str:
149 if vtype == "null": 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 return "null"
151 if vtype == "boolean": 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true
152 return "true" if v else "false"
153 return str(v)
156@router.post("/v1/conflicts/{conflict_id}/resolve")
157def resolve_conflict(
158 conflict_id: str,
159 req: ConflictResolveRequest,
160 identity: Annotated[Identity, Depends(resolve_identity)],
161) -> dict[str, Any]:
162 """Assert a canonical resolution fact and close the conflict (Spec-15-Fact-Semantics)."""
163 if not identity.can_write(): 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 raise HTTPException(
165 status_code=status.HTTP_403_FORBIDDEN, detail="write permission required"
166 )
168 with db() as conn:
169 conflict = conn.execute("SELECT * FROM conflicts WHERE id = ?", (conflict_id,)).fetchone()
171 # F-FED-CONFLICT-TENANT: a conflict is only visible/resolvable to a caller
172 # whose tenant owns the conflicting facts. Treat a cross-tenant conflict as
173 # not found (don't leak its existence) by requiring fact_a to live in the
174 # caller's tenant.
175 if conflict is not None:
176 owns = conn.execute(
177 "SELECT 1 FROM facts WHERE id = ? AND tenant_id = ?",
178 (conflict["fact_a_id"], identity.tenant_id),
179 ).fetchone()
180 if owns is None: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true
181 conflict = None
183 if conflict is None:
184 raise HTTPException(status_code=404, detail="conflict not found")
185 if conflict["status"] == "resolved":
186 raise HTTPException(status_code=409, detail="conflict already resolved")
188 fact_a = conn.execute(
189 "SELECT * FROM facts WHERE id = ? AND tenant_id = ?",
190 (conflict["fact_a_id"], identity.tenant_id),
191 ).fetchone()
192 fact_b = conn.execute(
193 "SELECT * FROM facts WHERE id = ? AND tenant_id = ?",
194 (conflict["fact_b_id"], identity.tenant_id),
195 ).fetchone()
197 if fact_a is None or fact_b is None: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true
198 raise HTTPException(status_code=500, detail="conflicting facts not found in store")
200 # Determine value for the resolution fact
201 if req.new_value is not None:
202 res_type = req.new_value.type
203 res_v = _encode_value(req.new_value.type, req.new_value.v)
204 elif req.winning_fact_id is not None:
205 if req.winning_fact_id == fact_a["id"]:
206 winner = fact_a
207 elif req.winning_fact_id == fact_b["id"]: 207 ↛ 210line 207 didn't jump to line 210 because the condition on line 207 was always true
208 winner = fact_b
209 else:
210 raise HTTPException(
211 status_code=422,
212 detail="winning_fact_id must be one of the conflicting facts",
213 )
214 res_type = winner["value_type"]
215 res_v = winner["value_v"]
216 else:
217 raise HTTPException(status_code=422, detail="provide winning_fact_id or new_value")
219 resolution_fact_id = str(uuid.uuid4())
220 now = datetime.now(UTC).isoformat()
221 caller = identity.entity_uri
223 # 1. Assert resolution fact under a namespaced entity so it never shares the
224 # (entity, relation, scope) triple with the conflicting facts. Writing under
225 # the original entity+relation would trigger a new contradiction wave when the
226 # fact is federated to peers (spec §resolution-semantics, EG-51).
227 resolution_entity = f"stigmem:resolution:{conflict_id}"
228 hlc_res = node_hlc.tick()
229 conn.execute(
230 """INSERT INTO facts
231 (id, entity, relation, value_type, value_v, source, timestamp,
232 valid_until, confidence, scope, hlc, received_from, tenant_id)
233 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
234 (
235 resolution_fact_id,
236 resolution_entity,
237 fact_a["relation"],
238 res_type,
239 res_v,
240 caller,
241 now,
242 None,
243 1.0,
244 fact_a["scope"],
245 hlc_res,
246 None,
247 identity.tenant_id,
248 ),
249 )
251 # 2. Assert stigmem:resolves meta-fact (spec §5.10)
252 hlc_meta = node_hlc.tick()
253 conn.execute(
254 """INSERT INTO facts
255 (id, entity, relation, value_type, value_v, source, timestamp,
256 valid_until, confidence, scope, hlc, received_from, tenant_id)
257 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
258 (
259 str(uuid.uuid4()),
260 resolution_fact_id,
261 "stigmem:resolves",
262 "ref",
263 conflict_id,
264 "system:stigmem",
265 now,
266 None,
267 1.0,
268 fact_a["scope"],
269 hlc_meta,
270 None,
271 identity.tenant_id,
272 ),
273 )
275 # 3. Record updated conflict:status as a new fact (status changes are immutable appends)
276 hlc_status = node_hlc.tick()
277 conn.execute(
278 """INSERT INTO facts
279 (id, entity, relation, value_type, value_v, source, timestamp,
280 valid_until, confidence, scope, hlc, received_from, tenant_id)
281 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
282 (
283 str(uuid.uuid4()),
284 conflict_id,
285 "stigmem:conflict:status",
286 "string",
287 "resolved",
288 "system:stigmem",
289 now,
290 None,
291 1.0,
292 fact_a["scope"],
293 hlc_status,
294 None,
295 identity.tenant_id,
296 ),
297 )
299 # 4. Update conflicts table
300 conn.execute(
301 "UPDATE conflicts SET status = 'resolved', resolution_fact_id = ? WHERE id = ?",
302 (resolution_fact_id, conflict_id),
303 )
305 return {"resolution_fact_id": resolution_fact_id, "conflict_status": "resolved"}