"""Consistency guard (Section 8.2). Deterministic topic/polarity check against the ledger first (cheap, keeps the hot path cheap); LLM semantic-entailment check only if the cheap pass is clean. Flags contradictions with already-committed claims. """ from __future__ import annotations from dataclasses import dataclass from ...llm.client import LLMClient from ...llm.prompts import PromptRegistry from ...models import Claim from ..ledger import LedgerStore @dataclass class ConsistencyVerdict: contradicts: bool reason: str deterministic: bool = False class ConsistencyGuard: def __init__( self, client: LLMClient, prompts: PromptRegistry, ledger: LedgerStore ) -> None: self.client = client self.prompts = prompts self.ledger = ledger def check( self, *, character: str, draft: str, new_claims: list[Claim] ) -> ConsistencyVerdict: # 1) cheap deterministic pass hits = self.ledger.find_contradictions(character, new_claims) if hits: prior, new = hits[0] return ConsistencyVerdict( True, f"contradicts prior statement on '{prior.topic}': " f"\"{prior.proposition}\" vs \"{new.proposition}\"", deterministic=True, ) prior_claims = self.ledger.get(character).claims if not prior_claims: return ConsistencyVerdict(False, "no prior claims") # 2) LLM semantic entailment fallback prompt = self.prompts.render( "guard/consistency.md.j2", character=character, draft=draft, prior_claims=[ {"topic": c.topic, "proposition": c.proposition, "polarity": c.polarity} for c in prior_claims ], ) try: data, _ = self.client.complete_json( tier="guard", task="consistency_guard", user=prompt, ) except Exception: # On error, don't block on semantic check (deterministic pass is # the hard guarantee); allow but note. return ConsistencyVerdict(False, "semantic check skipped (error)") contradicts = ( bool(data.get("contradicts", False)) if isinstance(data, dict) else False ) reason = str(data.get("reason", "")) if isinstance(data, dict) else "" return ConsistencyVerdict(contradicts, reason)