Coverage for node / src / stigmem_node / routes / lint.py: 86%
106 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Lint route — spec §14 (v0.7) + async job path (spec §14.5).
3POST /v1/lint
4{ scope, checks?, entity?, relation?, stale_lookahead_s? }
5 → 200 sync result, or 202 { job_id, status, estimated_s } when scope > threshold.
7GET /v1/lint/jobs/:job_id
8 → 200 job status/result, or 404 if not found.
9"""
11from __future__ import annotations
13from datetime import UTC, datetime, timedelta
14from typing import Annotated, Any
16from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
17from fastapi.responses import JSONResponse
19from ..auth import Identity, resolve_identity
20from ..db import db
21from ..jobs import create_job, get_job, mark_done, mark_failed, mark_running
22from ..models.constants import VALID_SCOPES
23from ..models.lint import ALL_CHECKS, LintCheck, LintFinding, LintRequest, LintResult
24from ..settings import settings
26router = APIRouter(tags=["lint"])
28INTENT_ROUTING_RELATIONS = frozenset({"intent:handoff_to", "intent:context_ref"})
30# Constant WHERE-fragment tails for the lint queries. Optional filters are
31# gated via ``(? IS NULL OR …)`` so the SQL strings are module-level constants
32# — no user input ever flows into the query text. Closes the
33# ``py/sql-injection`` taint that the previous conditional-fragment builder
34# triggered (issue #115).
35_COUNT_SQL = (
36 "SELECT COUNT(*) FROM facts f"
37 " WHERE 1=1"
38 " AND f.tenant_id = ?"
39 " AND f.scope = ?"
40 " AND (? IS NULL OR f.entity = ?)"
41 " AND (? IS NULL OR f.relation = ?)"
42)
45def _lint_filter_params(
46 tenant_id: str, scope: str, entity: str | None, relation: str | None
47) -> list[Any]:
48 """Return the bind values for the lint filter tails (tenant-scoped, audit H4).
50 Empty-string entity/relation are normalized to None so the IS-NULL gate
51 preserves the previous ``if entity:`` truthiness behaviour.
52 """
53 entity_p = entity or None
54 relation_p = relation or None
55 return [tenant_id, scope, entity_p, entity_p, relation_p, relation_p]
58_CONFLICT_SQL = (
59 "SELECT c.id AS conflict_id, c.fact_a_id, c.fact_b_id, fa.entity, fa.relation"
60 " FROM conflicts c"
61 " JOIN facts fa ON fa.id = c.fact_a_id"
62 " JOIN facts fb ON fb.id = c.fact_b_id"
63 " WHERE c.status = 'unresolved'"
64 " AND fa.tenant_id = ?"
65 " AND fa.scope = ?"
66 " AND (? IS NULL OR fa.entity = ?)"
67 " AND (? IS NULL OR fa.relation = ?)"
68)
71def _check_contradictions(conn: Any, fa_params: list[Any]) -> list[dict[str, Any]]:
72 """Return contradiction findings for unresolved conflicts in the filtered scope."""
73 findings: list[dict[str, Any]] = []
74 for row in conn.execute(_CONFLICT_SQL, fa_params).fetchall():
75 findings.append(
76 {
77 "check": "contradiction",
78 "severity": "error",
79 "entity": row["entity"],
80 "relation": row["relation"],
81 "fact_ids": [row["fact_a_id"], row["fact_b_id"]],
82 "detail": f"unresolved conflict {row['conflict_id']}",
83 }
84 )
85 return findings
88_STALE_SQL = (
89 "SELECT f.id, f.entity, f.relation, f.valid_until"
90 " FROM facts f"
91 " WHERE f.valid_until IS NOT NULL"
92 " AND f.confidence > 0.0"
93 " AND f.valid_until <= ?"
94 " AND f.tenant_id = ?"
95 " AND f.scope = ?"
96 " AND (? IS NULL OR f.entity = ?)"
97 " AND (? IS NULL OR f.relation = ?)"
98)
101def _check_stale(
102 conn: Any,
103 f_params: list[Any],
104 now: str,
105 lookahead: str,
106 stale_lookahead_s: int,
107) -> list[dict[str, Any]]:
108 """Return stale (already-expired or expiring-soon) findings."""
109 findings: list[dict[str, Any]] = []
110 for row in conn.execute(_STALE_SQL, [lookahead] + f_params).fetchall(): 110 ↛ 111line 110 didn't jump to line 111 because the loop on line 110 never started
111 expired = row["valid_until"] <= now
112 findings.append(
113 {
114 "check": "stale",
115 "severity": "warning" if expired else "info",
116 "entity": row["entity"],
117 "relation": row["relation"],
118 "fact_ids": [row["id"]],
119 "detail": (
120 f"expired at {row['valid_until']}"
121 if expired
122 else f"expires at {row['valid_until']} (within {stale_lookahead_s}s)"
123 ),
124 }
125 )
126 return findings
129_ORPHAN_SQL = (
130 "SELECT entity FROM facts"
131 " WHERE tenant_id = ?"
132 " AND scope = ?"
133 " AND (? IS NULL OR entity = ?)"
134 " GROUP BY entity"
135 " HAVING COUNT(*) > 0"
136 " AND SUM(CASE WHEN confidence > 0.0"
137 " AND (valid_until IS NULL OR valid_until > ?) THEN 1 ELSE 0 END) = 0"
138)
141def _check_orphans(
142 conn: Any, tenant_id: str, scope: str, entity: str | None, now: str
143) -> list[dict[str, Any]]:
144 """Return orphan-entity findings (entities with no live facts in scope)."""
145 findings: list[dict[str, Any]] = []
146 entity_p = entity or None
147 for row in conn.execute(_ORPHAN_SQL, [tenant_id, scope, entity_p, entity_p, now]).fetchall(): 147 ↛ 148line 147 didn't jump to line 148 because the loop on line 147 never started
148 findings.append(
149 {
150 "check": "orphan",
151 "severity": "info",
152 "entity": row["entity"],
153 "relation": None,
154 "fact_ids": [],
155 "detail": f"entity {row['entity']!r} has no live facts in scope={scope}",
156 }
157 )
158 return findings
161_REF_SQL = (
162 "SELECT f.id, f.entity, f.relation, f.value_v"
163 " FROM facts f"
164 " WHERE f.value_type = 'ref'"
165 " AND f.confidence > 0.0"
166 " AND (f.valid_until IS NULL OR f.valid_until > ?)"
167 " AND f.tenant_id = ?"
168 " AND f.scope = ?"
169 " AND (? IS NULL OR f.entity = ?)"
170 " AND (? IS NULL OR f.relation = ?)"
171)
174def _check_broken_refs(
175 conn: Any, tenant_id: str, f_params: list[Any], now: str
176) -> list[dict[str, Any]]:
177 """Return broken-ref findings for value-type=ref facts whose target has no live facts."""
178 findings: list[dict[str, Any]] = []
179 for row in conn.execute(_REF_SQL, [now] + f_params).fetchall(): 179 ↛ 180line 179 didn't jump to line 180 because the loop on line 179 never started
180 target_entity = row["value_v"]
181 live_count = conn.execute(
182 "SELECT COUNT(*) FROM facts"
183 " WHERE entity = ? AND tenant_id = ? AND confidence > 0.0"
184 " AND (valid_until IS NULL OR valid_until > ?)",
185 [target_entity, tenant_id, now],
186 ).fetchone()[0]
187 if live_count == 0:
188 is_intent = row["relation"] in INTENT_ROUTING_RELATIONS
189 findings.append(
190 {
191 "check": "broken_ref",
192 "severity": "error" if is_intent else "warning",
193 "entity": row["entity"],
194 "relation": row["relation"],
195 "fact_ids": [row["id"]],
196 "detail": f"ref target entity {target_entity!r} has no live facts",
197 }
198 )
199 return findings
202_NS_SQL = (
203 "SELECT f.entity, f.relation, GROUP_CONCAT(f.id) AS ids"
204 " FROM facts f"
205 " WHERE f.confidence > 0.0"
206 " AND (f.valid_until IS NULL OR f.valid_until > ?)"
207 " AND instr(f.relation, ':') = 0"
208 " AND f.tenant_id = ?"
209 " AND f.scope = ?"
210 " AND (? IS NULL OR f.entity = ?)"
211 " AND (? IS NULL OR f.relation = ?)"
212 " GROUP BY f.entity, f.relation"
213)
216def _check_namespacing(conn: Any, f_params: list[Any], now: str) -> list[dict[str, Any]]:
217 """Return namespacing findings for live facts whose relation lacks a 'prefix:' namespace."""
218 findings: list[dict[str, Any]] = []
219 for row in conn.execute(_NS_SQL, [now] + f_params).fetchall():
220 findings.append(
221 {
222 "check": "namespacing",
223 "severity": "warning",
224 "entity": row["entity"],
225 "relation": row["relation"],
226 "fact_ids": row["ids"].split(",") if row["ids"] else [],
227 "detail": (
228 f"bare relation {row['relation']!r} has no namespace prefix — "
229 f"rename to 'your-prefix:{row['relation']}' to avoid silent collisions"
230 ),
231 }
232 )
233 return findings
236def _run_lint_sweep(
237 tenant_id: str,
238 scope: str,
239 checks: list[LintCheck],
240 entity: str | None,
241 relation: str | None,
242 stale_lookahead_s: int,
243) -> dict[str, Any]:
244 """Execute the lint sweep and return a dict matching LintResult fields."""
245 now_dt = datetime.now(UTC)
246 now = now_dt.isoformat()
247 lookahead = (now_dt + timedelta(seconds=stale_lookahead_s)).isoformat()
249 f_params = _lint_filter_params(tenant_id, scope, entity, relation)
250 fa_params = _lint_filter_params(tenant_id, scope, entity, relation)
252 findings: list[dict[str, Any]] = []
253 fact_count = 0
255 with db() as conn:
256 fact_count = conn.execute(_COUNT_SQL, f_params).fetchone()[0]
258 if "contradiction" in checks:
259 findings.extend(_check_contradictions(conn, fa_params))
261 if "stale" in checks:
262 findings.extend(_check_stale(conn, f_params, now, lookahead, stale_lookahead_s))
264 if "orphan" in checks:
265 findings.extend(_check_orphans(conn, tenant_id, scope, entity, now))
267 if "broken_ref" in checks:
268 findings.extend(_check_broken_refs(conn, tenant_id, f_params, now))
270 if "namespacing" in checks:
271 findings.extend(_check_namespacing(conn, f_params, now))
273 return {
274 "findings": findings,
275 "checked_at": now,
276 "scope": scope,
277 "checks_run": checks,
278 "fact_count": fact_count,
279 }
282def _lint_job_worker(job_id: str, req: LintRequest, tenant_id: str) -> None:
283 """Background task: run lint sweep and update job status."""
284 mark_running(job_id)
285 try:
286 result = _run_lint_sweep(
287 tenant_id=tenant_id,
288 scope=req.scope,
289 checks=req.checks or ALL_CHECKS,
290 entity=req.entity,
291 relation=req.relation,
292 stale_lookahead_s=req.stale_lookahead_s,
293 )
294 mark_done(job_id, result)
295 except Exception as exc:
296 mark_failed(job_id, str(exc))
299@router.post("/v1/lint")
300def lint_scope(
301 req: LintRequest,
302 background_tasks: BackgroundTasks,
303 identity: Annotated[Identity, Depends(resolve_identity)],
304) -> Any:
305 """Health-check sweep for a scope (Spec-20-Lint-Semantics). Read-only.
307 Returns 200 with results synchronously for scopes ≤ threshold facts.
308 Returns 202 with job_id for larger scopes; poll GET /v1/lint/jobs/:job_id.
309 """
310 if not identity.can_read(): 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 raise HTTPException(status_code=403, detail="read permission required")
312 if req.scope not in VALID_SCOPES:
313 raise HTTPException(status_code=400, detail=f"scope must be one of {VALID_SCOPES}")
315 checks_to_run: list[LintCheck] = req.checks if req.checks else ALL_CHECKS
317 # Count scope facts to choose sync vs. async path (spec §14.5).
318 # Tenant-scoped so the count is not a cross-tenant oracle (audit H4).
319 with db() as conn:
320 scope_count: int = conn.execute(
321 "SELECT COUNT(*) FROM facts WHERE tenant_id = ? AND scope = ?",
322 [identity.tenant_id, req.scope],
323 ).fetchone()[0]
325 if scope_count > settings.async_job_threshold:
326 estimated_s = max(10, scope_count // 5_000)
327 job_id = create_job("lint", req.scope, estimated_s, identity.tenant_id)
328 background_tasks.add_task(_lint_job_worker, job_id, req, identity.tenant_id)
329 return JSONResponse(
330 status_code=202,
331 content={"job_id": job_id, "status": "pending", "estimated_s": estimated_s},
332 )
334 result = _run_lint_sweep(
335 identity.tenant_id,
336 req.scope,
337 checks_to_run,
338 req.entity,
339 req.relation,
340 req.stale_lookahead_s,
341 )
342 return LintResult(
343 findings=[LintFinding(**f) for f in result["findings"]],
344 checked_at=result["checked_at"],
345 scope=result["scope"],
346 checks_run=result["checks_run"],
347 fact_count=result["fact_count"],
348 )
351@router.get("/v1/lint/jobs/{job_id}")
352def get_lint_job(
353 job_id: str,
354 identity: Annotated[Identity, Depends(resolve_identity)],
355) -> Any:
356 """Poll the status of an async lint job (Spec-20-Lint-Semantics)."""
357 if not identity.can_read(): 357 ↛ 358line 357 didn't jump to line 358 because the condition on line 357 was never true
358 raise HTTPException(status_code=403, detail="read permission required")
359 job = get_job(job_id, job_type="lint", tenant_id=identity.tenant_id)
360 if job is None:
361 raise HTTPException(status_code=404, detail="job not found")
362 return job