Spaces:
Running on Zero
Running on Zero
| """Per-character committed-statement ledger (Section 7). | |
| Stores raw utterances + structured claims, persists to | |
| ``runtime/<session>/ledgers/<char>.json``, and provides the deterministic | |
| topic/polarity contradiction check used by the consistency guard. | |
| """ | |
| from __future__ import annotations | |
| from pathlib import Path | |
| from ..models import CharacterLedger, Claim, LedgerEntry | |
| def _opposite(a: str, b: str) -> bool: | |
| return {a, b} == {"affirm", "deny"} | |
| class LedgerStore: | |
| """Loads/saves all character ledgers for a session under one directory.""" | |
| def __init__(self, ledgers_dir: Path) -> None: | |
| self.dir = ledgers_dir | |
| self.dir.mkdir(parents=True, exist_ok=True) | |
| self._cache: dict[str, CharacterLedger] = {} | |
| def _path(self, character: str) -> Path: | |
| safe = character.lower().replace(" ", "_") | |
| return self.dir / f"{safe}.json" | |
| def get(self, character: str) -> CharacterLedger: | |
| if character in self._cache: | |
| return self._cache[character] | |
| path = self._path(character) | |
| if path.exists(): | |
| ledger = CharacterLedger.model_validate_json(path.read_text("utf-8")) | |
| else: | |
| ledger = CharacterLedger(character=character) | |
| self._cache[character] = ledger | |
| return ledger | |
| def append(self, character: str, entry: LedgerEntry) -> None: | |
| ledger = self.get(character) | |
| ledger.entries.append(entry) | |
| self._flush(character) | |
| def _flush(self, character: str) -> None: | |
| path = self._path(character) | |
| path.write_text(self._cache[character].model_dump_json(indent=2), "utf-8") | |
| # -- deterministic contradiction check --------------------------------- | |
| def find_contradictions( | |
| self, character: str, new_claims: list[Claim] | |
| ) -> list[tuple[Claim, Claim]]: | |
| """Same topic + opposite polarity against committed claims (Section 7).""" | |
| prior = self.get(character).claims | |
| hits: list[tuple[Claim, Claim]] = [] | |
| for nc in new_claims: | |
| if nc.polarity == "neutral": | |
| continue | |
| for pc in prior: | |
| if pc.topic == nc.topic and _opposite(pc.polarity, nc.polarity): | |
| hits.append((pc, nc)) | |
| return hits | |
| def claim_by_id(self, character: str, claim_id: str) -> Claim | None: | |
| for c in self.get(character).claims: | |
| if c.claim_id == claim_id: | |
| return c | |
| return None | |