"""Daily auditor — Pilon 5 of the cache freshness defense. Sweeps non-stable cache rows whose ``last_audited_at`` is older than the audit interval, runs the cache_judge against fresh evidence from a /v1/gather call, and applies KEEP/INVALIDATE decisions. Atoms that survive consecutive audits accumulate ``consecutive_audit_passes`` so the judge becomes harder to flip them — trusted gold/silver gain inertia over time. Pipeline per atom: 1. SELECT candidate atom from brain_analysis_atom 2. POST /v1/gather to brain to get fresh top-3 evidence for the claim 3. Extract cached_truth from result_processed (TRUE / FALSE / MIXED / UV) 4. cache_judge.judge_cache_validity(claim, cached_truth, evidence, ...) 5. analysis_atom.apply_judge_verdict(atom_id, verdict) — INVALIDATE → expires_at = now() — KEEP_CACHE → bump consecutive_audit_passes (rate-limited) — NEEDS_FULL_RECHECK → no DB change, just logged This task imports brain_api directly (same Python code) for DB pool access and helper functions; HTTP is used only for gather (which orchestrates search + rerank + NLI inside brain itself). """ from __future__ import annotations import asyncio import json from datetime import datetime, timezone from typing import Any from brain_api.db import db from brain_api.services.analysis_atom import ( apply_judge_verdict, is_effectively_fresh, ) from brain_api.services.cache_judge import ( EvidenceSnippet, JudgeVerdict, judge_cache_validity, ) from scheduler.brain_client import BrainClient from scheduler.config import settings from shared.llm_client import LlmClient from shared.logging import get_logger log = get_logger(__name__) def _extract_cached_truth(processed: dict | None) -> str: """Pull a cached_truth label from result_processed. DIDI v1 schema for the claims component uses status: TRUE/FALSE/UV/OP/MIXED. For techniques + ai_tampered the result is a structured score object rather than a truth direction — those components don't benefit from the NLI judge anyway (their result depends on the input text, not the world), so we return UNVERIFIED to make the judge degrade to NEEDS_FULL_RECHECK or KEEP_CACHE depending on evidence. """ if not isinstance(processed, dict): return "UNVERIFIED" raw = processed.get("status") or processed.get("verdict") if not isinstance(raw, str): return "UNVERIFIED" s = raw.strip().upper() if s in ("TRUE", "VT", "VERIFIED_TRUE"): return "TRUE" if s in ("FALSE", "VF", "VERIFIED_FALSE"): return "FALSE" if s in ("MIXED",): return "MIXED" return "UNVERIFIED" async def _select_candidates(limit: int) -> list[dict[str, Any]]: """Pick atoms eligible for audit, oldest-first. Eligibility: - cache_tier IN ('gold','silver') - volatility IS NOT NULL AND volatility != 'stable' - last_audited_at IS NULL OR < now() - auditor_interval - created_at < now() - min_age_hours (let new writes settle) - expires_at IS NULL OR > now() (don't audit dead rows) """ if not db.pool: raise RuntimeError("brain_db not connected") sql = """ SELECT atom_id, content_hash, content_preview, component, tier, cache_tier, volatility, topic_codes, llm_confidence, result_processed, created_at, last_audited_at, consecutive_audit_passes FROM brain_analysis_atom WHERE cache_tier IN ('gold', 'silver') AND volatility IS NOT NULL AND volatility <> 'stable' AND (expires_at IS NULL OR expires_at > now()) AND created_at < now() - ($1 || ' hours')::interval AND ( last_audited_at IS NULL OR last_audited_at < now() - ($2 || ' seconds')::interval ) ORDER BY last_audited_at NULLS FIRST, created_at ASC LIMIT $3 """ async with db.pool.acquire() as conn: rows = await conn.fetch( sql, str(settings.auditor_min_age_hours), str(settings.auditor_interval_s), limit, ) out: list[dict[str, Any]] = [] for r in rows: result_processed = r["result_processed"] if isinstance(result_processed, str): result_processed = json.loads(result_processed) out.append({ "atom_id": r["atom_id"], "content_hash": r["content_hash"], "content_preview": r["content_preview"] or "", "component": r["component"], "tier": r["tier"], "cache_tier": r["cache_tier"], "volatility": r["volatility"], "topic_codes": list(r["topic_codes"]) if r["topic_codes"] else [], "llm_confidence": ( float(r["llm_confidence"]) if r["llm_confidence"] is not None else None ), "result_processed": result_processed or {}, "created_at": r["created_at"], "last_audited_at": r["last_audited_at"], "consecutive_audit_passes": r["consecutive_audit_passes"], }) return out def _evidence_from_gather_response(resp: dict | None) -> list[EvidenceSnippet]: """Convert a brain /v1/gather response payload into EvidenceSnippet list.""" if not resp: return [] items = resp.get("evidence") or [] out: list[EvidenceSnippet] = [] for it in items[:3]: # judge uses top 3 anyway url = (it.get("url") or "").strip() text = ( it.get("full_text") or it.get("summary") or it.get("snippet") or "" ) published = it.get("published_at") if url and text: out.append(EvidenceSnippet(url=url, text=text, published_at=published)) return out async def _gather_fresh_evidence( brain: BrainClient, *, claim: str, volatility: str ) -> list[EvidenceSnippet]: """Call /v1/gather over HTTP to get fresh evidence for the claim. Disables NLI on this internal call (we run our own NLI via cache_judge afterwards, so doing it twice would be wasteful). """ body = { "claim": claim[:1500], "max_evidence": 5, "include_full_text": True, "run_nli": False, "volatility_hint": volatility, } try: resp = await brain._http.post("/v1/gather", json=body) if resp.status_code >= 400: log.debug( "auditor_gather_http_error", status=resp.status_code, ) return [] return _evidence_from_gather_response(resp.json()) except Exception as e: # noqa: BLE001 log.debug("auditor_gather_failed", error=f"{type(e).__name__}:{e}") return [] async def _audit_one( *, candidate: dict[str, Any], brain: BrainClient, llm: LlmClient, ) -> str: """Run a single audit. Returns the decision label for telemetry.""" atom_id = candidate["atom_id"] claim = candidate["content_preview"] volatility = candidate["volatility"] or "evolving" if not claim: # Can't judge without a claim text — touch last_audited_at to defer # this row and move on. log.debug("auditor_no_claim_text", atom_id=atom_id) return "SKIPPED" cached_truth = _extract_cached_truth(candidate["result_processed"]) age_hours = ( datetime.now(tz=timezone.utc) - candidate["created_at"] ).total_seconds() / 3600.0 # Cheap path: if effective confidence is still high we don't even need # to gather. Bumps consecutive_audit_passes via apply_judge_verdict. eff_fresh = is_effectively_fresh( base_confidence=candidate["llm_confidence"], volatility=volatility, age_hours=age_hours, consecutive_audit_passes=candidate["consecutive_audit_passes"], ) if not eff_fresh: # Confidence has decayed below floor — auto-invalidate without # spending an LLM call. verdict = JudgeVerdict( decision="INVALIDATE", nli_skipped=True, reasoning=( f"effective confidence below floor " f"(volatility={volatility}, age={age_hours:.0f}h)" ), evaluated_at=datetime.now(tz=timezone.utc).isoformat(), ) await apply_judge_verdict(atom_id, verdict) log.info( "auditor_invalidated_decay", atom_id=atom_id, volatility=volatility, age_hours=round(age_hours, 1), ) return "INVALIDATE_DECAY" # Active path: gather + judge. evidence = await _gather_fresh_evidence( brain, claim=claim, volatility=volatility ) verdict = await judge_cache_validity( llm, claim=claim, cached_truth=cached_truth, current_evidence=evidence, volatility=volatility, age_hours=age_hours, consecutive_audit_passes=candidate["consecutive_audit_passes"], ) await apply_judge_verdict(atom_id, verdict) log.info( "auditor_decision", atom_id=atom_id, decision=verdict.decision, volatility=volatility, cached_truth=cached_truth, ) return verdict.decision async def _run_one_cycle(brain: BrainClient, llm: LlmClient) -> dict[str, int]: """Pull candidates and audit them sequentially. Sequential rather than parallel because each judge call hits the LLM router — running 200 in parallel would saturate it. The local Qwen handles ~2-4 concurrent requests well, so we could batch with a semaphore later if cycle duration becomes an issue. """ candidates = await _select_candidates(settings.auditor_batch_limit) if not candidates: return {"audited": 0, "candidates": 0} counts: dict[str, int] = {} for c in candidates: try: decision = await _audit_one(candidate=c, brain=brain, llm=llm) counts[decision] = counts.get(decision, 0) + 1 except Exception as e: # noqa: BLE001 log.warning( "auditor_one_failed", atom_id=c.get("atom_id"), error=f"{type(e).__name__}:{e}", ) counts["ERROR"] = counts.get("ERROR", 0) + 1 log.info( "auditor_cycle_done", audited=len(candidates), breakdown=counts, ) return {"audited": len(candidates), **counts} async def run_auditor(brain: BrainClient, llm: LlmClient) -> None: """Long-running task — sweeps daily by default.""" if not settings.auditor_enabled: log.info("auditor_disabled") return log.info( "auditor_loop_start", interval_s=settings.auditor_interval_s, batch_limit=settings.auditor_batch_limit, ) while True: try: await _run_one_cycle(brain, llm) except Exception: # noqa: BLE001 log.exception("auditor_cycle_error") await asyncio.sleep(settings.auditor_interval_s)