Skip to content

Python API Reference

Top-level exports

import worldoracle

worldoracle

worldoracle - NPC contradiction detector and belief repair for game worlds.

ClosedLoopError

Bases: RuntimeError

Raised when a worldoracle closed-loop gate refuses the world state.

AgentVote(agent_id, choice, round=0, confidence=1.0) dataclass

One agent vote in a debate/consensus round.

ConsensusCertificate(n_agents, n_rounds, agreement, plurality_choice, faction_count, factions, lambda2_hat, deadline_rounds, converged, vote_count) dataclass

Machine-checkable snapshot of collective agreement.

Attributes:

Name Type Description
n_agents int

Distinct agents seen.
n_rounds int

Max round index + 1.
agreement float

Fraction agreeing with plurality choice (latest round).
plurality_choice str

Mode of latest-round votes.
faction_count int

Number of distinct choices in latest round.
factions dict[str, tuple[str, ...]]

choice → agent_ids (latest round).
lambda2_hat float

Proxy for sub-dominant mixing: 1 - agreement (0 = fully mixed/agreed, closer to 1 = slow/stuck factions).
deadline_rounds int | None

Estimated rounds to agreement under geometric contraction (ceil log residual / log lambda2) when lambda2 < 1.
converged bool

agreement >= min_agreement and faction_count <= max_factions.

BeliefRepairer

Generate RepairFrames for contradictions.

Strategy priority order (applied in sequence, first match wins): 1. prefer_newer - higher timestamp wins 2. prefer_higher_confidence - higher confidence wins (when timestamps tied) 3. prefer_observation - source == 'observation' wins over hearsay 4. default - no tiebreaker; pred_a kept, reason logged

repair(pred_a, pred_b)

Apply repair strategies in priority order and return a RepairFrame.

Source code in src/worldoracle/predicate.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def repair(self, pred_a: WorldPredicate, pred_b: WorldPredicate) -> RepairFrame:
    """Apply repair strategies in priority order and return a RepairFrame."""
    now = time.time()
    # 1. prefer_newer
    if pred_a.timestamp != pred_b.timestamp:
        winner = pred_a if pred_a.timestamp > pred_b.timestamp else pred_b
        return RepairFrame(
            predicate_a_id=pred_a.id,
            predicate_b_id=pred_b.id,
            strategy="prefer_newer",
            resolved_value=winner.value,
            reason=f"Keeping newer belief (timestamp {winner.timestamp})",
            timestamp=now,
        )
    # 2. prefer_higher_confidence
    if pred_a.confidence != pred_b.confidence:
        winner = pred_a if pred_a.confidence > pred_b.confidence else pred_b
        return RepairFrame(
            predicate_a_id=pred_a.id,
            predicate_b_id=pred_b.id,
            strategy="prefer_higher_confidence",
            resolved_value=winner.value,
            reason=f"Keeping higher confidence belief ({winner.confidence:.2f})",
            timestamp=now,
        )
    # 3. prefer_observation
    if pred_a.source == "observation" or pred_b.source == "observation":
        winner = pred_a if pred_a.source == "observation" else pred_b
        return RepairFrame(
            predicate_a_id=pred_a.id,
            predicate_b_id=pred_b.id,
            strategy="prefer_observation",
            resolved_value=winner.value,
            reason="Preferring direct observation over hearsay",
            timestamp=now,
        )
    # Default: no distinguishing strategy - keep pred_a, log reason
    return RepairFrame(
        predicate_a_id=pred_a.id,
        predicate_b_id=pred_b.id,
        strategy="default",
        resolved_value=pred_a.value,
        reason=(
            "No distinguishing strategy (tied timestamps and confidence, "
            "no observation source); defaulting to first predicate"
        ),
        timestamp=now,
    )

BeliefState(npc_id, predicates=list()) dataclass

An NPC's belief state: a set of WorldPredicates.

__post_init__()

Compute initial content-addressed ID.

Source code in src/worldoracle/predicate.py 
62
63
64
def __post_init__(self) -> None:
    """Compute initial content-addressed ID."""
    self._recompute_id()

add(pred)

Add a predicate and recompute ID. Skips duplicates by content ID.

Source code in src/worldoracle/predicate.py 
72
73
74
75
76
77
def add(self, pred: WorldPredicate) -> None:
    """Add a predicate and recompute ID. Skips duplicates by content ID."""
    if any(p.id == pred.id for p in self.predicates):
        return
    self.predicates.append(pred)
    self._recompute_id()

get(subject, attribute)

Get predicates matching subject and attribute.

Source code in src/worldoracle/predicate.py 
79
80
81
def get(self, subject: str, attribute: str) -> list[WorldPredicate]:
    """Get predicates matching subject and attribute."""
    return [p for p in self.predicates if p.subject == subject and p.attribute == attribute]

to_dict()

Serialize to dict.

Source code in src/worldoracle/predicate.py 
83
84
85
86
87
88
89
def to_dict(self) -> dict[str, Any]:
    """Serialize to dict."""
    return {
        "id": self.id,
        "npc_id": self.npc_id,
        "predicates": [p.to_dict() for p in self.predicates],
    }

ContradictionDetector() dataclass

Find contradictions in a BeliefState.

detect(state)

Return pairs of contradicting predicates (same subject+attribute, different values).

Source code in src/worldoracle/predicate.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def detect(self, state: BeliefState) -> list[tuple[WorldPredicate, WorldPredicate]]:
    """Return pairs of contradicting predicates (same subject+attribute, different values)."""
    pairs: list[tuple[WorldPredicate, WorldPredicate]] = []
    groups: dict[tuple[str, str], list[WorldPredicate]] = defaultdict(list)
    for p in state.predicates:
        groups[(p.subject, p.attribute)].append(p)
    for _key, preds in groups.items():
        if len(preds) < 2:
            continue
        for i in range(len(preds)):
            for j in range(i + 1, len(preds)):
                a, b = preds[i], preds[j]
                if a.value != b.value and a.value is not None and b.value is not None:
                    pairs.append((a, b))
    return pairs

RepairFrame(predicate_a_id, predicate_b_id, strategy, resolved_value, reason, timestamp=0.0) dataclass

A suggested repair for a contradiction.

__post_init__()

Compute content-addressed ID.

Source code in src/worldoracle/predicate.py 
125
126
127
128
def __post_init__(self) -> None:
    """Compute content-addressed ID."""
    payload = f"{self.predicate_a_id}|{self.predicate_b_id}|{self.strategy}"
    self.id = hashlib.sha256(payload.encode()).hexdigest()[:16]

to_dict()

Serialize to dict.

Source code in src/worldoracle/predicate.py 
130
131
132
133
134
135
136
137
138
139
140
def to_dict(self) -> dict[str, Any]:
    """Serialize to dict."""
    return {
        "id": self.id,
        "predicate_a_id": self.predicate_a_id,
        "predicate_b_id": self.predicate_b_id,
        "strategy": self.strategy,
        "resolved_value": self.resolved_value,
        "reason": self.reason,
        "timestamp": self.timestamp,
    }

WorldPredicate(subject, attribute, value, source='', confidence=1.0, timestamp=0.0) dataclass

A belief predicate: subject has attribute=value.

__post_init__()

Compute content-addressed ID.

Source code in src/worldoracle/predicate.py 
24
25
26
27
def __post_init__(self) -> None:
    """Compute content-addressed ID."""
    payload = f"{self.subject}|{self.attribute}|{self.value!s}"
    self.id = hashlib.sha256(payload.encode()).hexdigest()[:16]

to_dict()

Serialize to dict.

Source code in src/worldoracle/predicate.py 
29
30
31
32
33
34
35
36
37
38
39
def to_dict(self) -> dict[str, Any]:
    """Serialize to dict."""
    return {
        "id": self.id,
        "subject": self.subject,
        "attribute": self.attribute,
        "value": self.value,
        "source": self.source,
        "confidence": self.confidence,
        "timestamp": self.timestamp,
    }

from_dict(d) classmethod

Deserialize from dict.

Source code in src/worldoracle/predicate.py 
41
42
43
44
45
46
47
48
49
50
51
@classmethod
def from_dict(cls, d: dict[str, Any]) -> WorldPredicate:
    """Deserialize from dict."""
    return cls(
        subject=d["subject"],
        attribute=d["attribute"],
        value=d["value"],
        source=d.get("source", ""),
        confidence=d.get("confidence", 1.0),
        timestamp=d.get("timestamp", 0.0),
    )

WorldOracleStore(path=':memory:')

SQLite-backed persistence for WorldPredicates and RepairFrames.

Open (or create) the store at path. Use ':memory:' for in-process use.

Source code in src/worldoracle/store.py 
37
38
39
40
41
42
43
44
45
46
def __init__(self, path: str | Path = ":memory:") -> None:
    """Open (or create) the store at *path*. Use ':memory:' for in-process use."""
    self.path = Path(path) if path != ":memory:" else None
    if self.path:
        self.path.parent.mkdir(parents=True, exist_ok=True)
    db_path = str(self.path) if self.path else ":memory:"
    self._conn = sqlite3.connect(db_path, check_same_thread=False)
    self._conn.row_factory = sqlite3.Row
    self._conn.executescript(self._SCHEMA)
    self._conn.commit()

close()

Close the database connection.

Source code in src/worldoracle/store.py 
48
49
50
def close(self) -> None:
    """Close the database connection."""
    self._conn.close()

save_predicate(npc_id, pred)

Upsert a predicate for the given NPC.

Source code in src/worldoracle/store.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def save_predicate(self, npc_id: str, pred: WorldPredicate) -> None:
    """Upsert a predicate for the given NPC."""
    self._conn.execute(
        "INSERT OR REPLACE INTO predicates VALUES (?,?,?,?,?,?,?,?)",
        (
            pred.id,
            npc_id,
            pred.subject,
            pred.attribute,
            json.dumps(pred.value),
            pred.source,
            pred.confidence,
            pred.timestamp,
        ),
    )
    self._conn.commit()

get_belief_state(npc_id)

Load all predicates for npc_id and return a BeliefState.

Source code in src/worldoracle/store.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_belief_state(self, npc_id: str) -> BeliefState:
    """Load all predicates for *npc_id* and return a BeliefState."""
    rows = self._conn.execute(
        "SELECT * FROM predicates WHERE npc_id=? ORDER BY timestamp",
        (npc_id,),
    ).fetchall()
    state = BeliefState(npc_id=npc_id)
    for row in rows:
        d = dict(row)
        p = WorldPredicate(
            subject=d["subject"],
            attribute=d["attribute"],
            value=json.loads(d["value"]),
            source=d["source"],
            confidence=d["confidence"],
            timestamp=d["timestamp"],
        )
        state.predicates.append(p)
    state._recompute_id()
    return state

list_npc_ids()

Return all distinct NPC IDs in the store.

Source code in src/worldoracle/store.py 
92
93
94
95
def list_npc_ids(self) -> list[str]:
    """Return all distinct NPC IDs in the store."""
    rows = self._conn.execute("SELECT DISTINCT npc_id FROM predicates").fetchall()
    return [r[0] for r in rows]

save_repair(repair)

Upsert a repair frame.

Source code in src/worldoracle/store.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def save_repair(self, repair: RepairFrame) -> None:
    """Upsert a repair frame."""
    self._conn.execute(
        "INSERT OR REPLACE INTO repairs VALUES (?,?,?,?,?,?,?)",
        (
            repair.id,
            repair.predicate_a_id,
            repair.predicate_b_id,
            repair.strategy,
            json.dumps(repair.resolved_value),
            repair.reason,
            repair.timestamp,
        ),
    )
    self._conn.commit()

get_repairs(predicate_a_id='', predicate_b_id='')

Return repair frames, optionally filtered by predicate ID.

Source code in src/worldoracle/store.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_repairs(
    self,
    predicate_a_id: str = "",
    predicate_b_id: str = "",
) -> list[RepairFrame]:
    """Return repair frames, optionally filtered by predicate ID."""
    if predicate_a_id or predicate_b_id:
        rows = self._conn.execute(
            "SELECT * FROM repairs WHERE predicate_a_id=? OR predicate_b_id=?",
            (predicate_a_id, predicate_b_id),
        ).fetchall()
    else:
        rows = self._conn.execute("SELECT * FROM repairs").fetchall()
    result = []
    for row in rows:
        d = dict(row)
        r = RepairFrame(
            predicate_a_id=d["predicate_a_id"],
            predicate_b_id=d["predicate_b_id"],
            strategy=d["strategy"],
            resolved_value=json.loads(d["resolved_value"]),
            reason=d["reason"],
            timestamp=d["timestamp"],
        )
        result.append(r)
    return result

TemporalBeliefStore(store)

Track how beliefs change over time using a separate snapshots table.

Source code in src/worldoracle/temporal.py 
25
26
27
28
def __init__(self, store: WorldOracleStore) -> None:
    self._store = store
    self._conn = store._conn
    self._init_schema()

record_snapshot()

Record current belief state as a timestamped snapshot. Returns snapshot ID.

Raises RuntimeError if the database fails to assign a row ID.

Source code in src/worldoracle/temporal.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def record_snapshot(self) -> int:
    """Record current belief state as a timestamped snapshot. Returns snapshot ID.

    Raises RuntimeError if the database fails to assign a row ID.
    """
    now = time.time()
    cur = self._conn.execute("INSERT INTO snapshot_registry (taken_at) VALUES (?)", (now,))
    snap_id = cur.lastrowid
    if snap_id is None:
        raise RuntimeError("Failed to insert snapshot registry row: no rowid returned")
    # Get all current predicates
    rows = self._conn.execute("SELECT * FROM predicates").fetchall()
    for row in rows:
        self._conn.execute(
            """INSERT INTO belief_snapshots
               (snapshot_id, snapshot_ts, npc_id, subject, attribute, value, confidence, source)
               VALUES (?,?,?,?,?,?,?,?)""",
            (
                snap_id,
                now,
                row["npc_id"],
                row["subject"],
                row["attribute"],
                row["value"],
                row["confidence"],
                row["source"],
            ),
        )
    self._conn.commit()
    return snap_id

get_belief_at(subject, predicate, timestamp)

Get a belief snapshot at or before the given timestamp.

Source code in src/worldoracle/temporal.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get_belief_at(
    self, subject: str, predicate: str, timestamp: float
) -> BeliefSnapshot | None:
    """Get a belief snapshot at or before the given timestamp."""
    row = self._conn.execute(
        """SELECT * FROM belief_snapshots
           WHERE subject=? AND attribute=? AND snapshot_ts <= ?
           ORDER BY snapshot_ts DESC LIMIT 1""",
        (subject, predicate, timestamp),
    ).fetchone()
    if row is None:
        return None
    return BeliefSnapshot(
        timestamp=row["snapshot_ts"],
        subject=row["subject"],
        predicate=row["attribute"],
        value=self._deserialize_value(row["value"]),
        confidence=row["confidence"],
        source=row["source"],
    )

get_belief_history(subject, predicate)

Full history of a belief, newest first.

Source code in src/worldoracle/temporal.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_belief_history(self, subject: str, predicate: str) -> list[BeliefSnapshot]:
    """Full history of a belief, newest first."""
    rows = self._conn.execute(
        """SELECT * FROM belief_snapshots
           WHERE subject=? AND attribute=?
           ORDER BY snapshot_ts DESC""",
        (subject, predicate),
    ).fetchall()
    result = []
    for row in rows:
        result.append(
            BeliefSnapshot(
                timestamp=row["snapshot_ts"],
                subject=row["subject"],
                predicate=row["attribute"],
                value=self._deserialize_value(row["value"]),
                confidence=row["confidence"],
                source=row["source"],
            )
        )
    return result

belief_drift(subject, predicate)

Measure how much a belief's confidence has changed. 0=stable, 1=volatile.

Source code in src/worldoracle/temporal.py 
132
133
134
135
136
137
138
139
140
def belief_drift(self, subject: str, predicate: str) -> float:
    """Measure how much a belief's confidence has changed. 0=stable, 1=volatile."""
    history = self.get_belief_history(subject, predicate)
    if len(history) < 2:
        return 0.0
    confidences = [h.confidence for h in history]
    max_c = max(confidences)
    min_c = min(confidences)
    return min(1.0, max_c - min_c)

gate_beliefs(store, *, mode='human_required', min_predicates=1)

Run consistency check with explicit repair policy.

Modes: - report: detect only, ok if score==1 - auto_repair: allow automatic repair (NOT for legal gates) - human_required: if any contradiction, FAIL and demand human (default - matches farm rule LEGAL GATES NEVER AUTO-FIX)

Phase guard: empty store → FAIL_LOUD (do not run contradiction checks before data exists - Foundry G-CONTRA ordering bug).

Source code in src/worldoracle/closed_loop.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def gate_beliefs(
    store: WorldOracleStore,
    *,
    mode: Mode = "human_required",
    min_predicates: int = 1,
) -> GateOutcome:
    """Run consistency check with explicit repair policy.

    Modes:
    - report: detect only, ok if score==1
    - auto_repair: allow automatic repair (NOT for legal gates)
    - human_required: if any contradiction, FAIL and demand human
      (default - matches farm rule LEGAL GATES NEVER AUTO-FIX)

    Phase guard: empty store → FAIL_LOUD (do not run contradiction
    checks before data exists - Foundry G-CONTRA ordering bug).
    """
    npc_ids = store.list_npc_ids()
    if not npc_ids:
        return _fail_loud(
            "empty belief store - refuse consistency check before data exists "
            "(Foundry: G-CONTRA ran before clips existed)"
        )

    total = 0
    for nid in npc_ids:
        total += len(store.get_belief_state(nid).predicates)
    if total < min_predicates:
        return _fail_loud(
            f"only {total} predicates (<{min_predicates}) - phase too early for contradiction gate"
        )

    if mode == "auto_repair":
        report = full_consistency_check(store, auto_repair=True)
    else:
        report = full_consistency_check(store, auto_repair=False)

    if report.contradictions_found == 0:
        return GateOutcome(
            True,
            "PASS",
            "no contradictions",
            0,
            report.consistency_score,
            0,
            False,
        )

    if mode == "human_required":
        return GateOutcome(
            False,
            "FAIL",
            f"{report.contradictions_found} contradictions - human review required "
            f"(LEGAL-NO-AUTOFIX); contested={report.most_contested[:3]}",
            1,
            report.consistency_score,
            report.contradictions_found,
            True,
        )

    if mode == "report":
        return GateOutcome(
            False,
            "FAIL",
            f"{report.contradictions_found} contradictions (report mode, no repair)",
            1,
            report.consistency_score,
            report.contradictions_found,
            False,
        )

    # auto_repair path already ran
    if report.unresolved > 0:
        return GateOutcome(
            False,
            "FAIL",
            f"auto_repair left {report.unresolved} unresolved",
            1,
            report.consistency_score,
            report.contradictions_found,
            False,
        )
    return GateOutcome(
        True,
        "PASS",
        f"auto_repaired {report.contradictions_repaired}",
        0,
        report.consistency_score,
        report.contradictions_found,
        False,
    )

assert_collective_consensus_ok(votes, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_collective_consensus is ok.

Source code in src/worldoracle/collective.py 
363
364
365
366
367
368
369
370
371
def assert_collective_consensus_ok(
    votes: Sequence[AgentVote | dict[str, Any]] | None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_collective_consensus` is ok."""
    outcome = gate_collective_consensus(votes, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

certify_consensus(votes, *, min_agreement=0.67, max_factions=1)

Build a consensus certificate from multi-agent vote traces.

Source code in src/worldoracle/collective.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def certify_consensus(
    votes: Sequence[AgentVote | dict[str, Any]],
    *,
    min_agreement: float = 0.67,
    max_factions: int = 1,
) -> ConsensusCertificate:
    """Build a consensus certificate from multi-agent vote traces."""
    parsed = [_as_vote(v, i) for i, v in enumerate(votes)]
    if not parsed:
        return ConsensusCertificate(
            n_agents=0,
            n_rounds=0,
            agreement=0.0,
            plurality_choice="",
            faction_count=0,
            factions={},
            lambda2_hat=1.0,
            deadline_rounds=None,
            converged=False,
            vote_count=0,
        )

    latest = _latest_round_votes(parsed)
    n_agents = len(latest)
    n_rounds = max(v.round for v in parsed) + 1
    counts = Counter(v.choice for v in latest)
    plurality_choice, top_n = counts.most_common(1)[0]
    agreement = top_n / n_agents if n_agents else 0.0
    factions: dict[str, tuple[str, ...]] = defaultdict(tuple)
    by_choice: dict[str, list[str]] = defaultdict(list)
    for v in latest:
        by_choice[v.choice].append(v.agent_id)
    factions = {c: tuple(sorted(aids)) for c, aids in by_choice.items()}
    faction_count = len(factions)

    # λ₂ proxy: disagreement mass (paper: sub-dominant mode scales slow mixing)
    lambda2_hat = max(0.0, min(1.0, 1.0 - agreement))
    deadline = estimate_deadline(lambda2_hat)
    converged = agreement >= min_agreement and faction_count <= max_factions

    return ConsensusCertificate(
        n_agents=n_agents,
        n_rounds=n_rounds,
        agreement=agreement,
        plurality_choice=plurality_choice,
        faction_count=faction_count,
        factions=factions,
        lambda2_hat=lambda2_hat,
        deadline_rounds=deadline,
        converged=converged,
        vote_count=len(parsed),
    )

estimate_deadline(lambda2_hat, *, residual=0.05, max_rounds=10000)

Estimate rounds until residual disagreement under geometric contraction.

deadline ≈ ceil( ln(residual) / ln(λ₂) ) when 0 < λ₂ < 1. Returns 0 if already mixed (λ₂≈0), None if λ₂ >= 1 (no contraction).

Source code in src/worldoracle/collective.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def estimate_deadline(lambda2_hat: float, *, residual: float = 0.05, max_rounds: int = 10_000) -> int | None:
    """Estimate rounds until residual disagreement under geometric contraction.

    deadline ≈ ceil( ln(residual) / ln(λ₂) ) when 0 < λ₂ < 1.
    Returns 0 if already mixed (λ₂≈0), None if λ₂ >= 1 (no contraction).
    """
    import math

    if lambda2_hat <= 1e-12:
        return 0
    if lambda2_hat >= 1.0 - 1e-12:
        return None
    if residual <= 0 or residual >= 1:
        residual = 0.05
    # ln(residual) / ln(lambda2) both negative for residual,lambda2 in (0,1)
    val = math.log(residual) / math.log(lambda2_hat)
    if val < 0:
        return None
    return min(max_rounds, max(0, int(math.ceil(val))))

gate_collective_consensus(votes, *, decision='finalize', min_agreement=0.67, max_factions=1, min_agents=2, max_rounds_without_convergence=None, require_votes=True)

Refuse uncertified multi-agent finalize (COLLECTIVE-CERT / arXiv 2608.05956).

Rules:

  • No votes when required → FAIL_LOUD
  • Fewer than min_agents → FAIL_LOUD
  • decision=finalize without convergence → FAIL (black-box stop)
  • decision=finalize past deadline without agreement → FAIL
  • decision=continue when already converged → FAIL (waste rounds)
  • decision=continue while not converged (under round budget) → PASS
  • decision=finalize when converged → PASS + certificate fields
Source code in src/worldoracle/collective.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
def gate_collective_consensus(
    votes: Sequence[AgentVote | dict[str, Any]] | None,
    *,
    decision: Decision = "finalize",
    min_agreement: float = 0.67,
    max_factions: int = 1,
    min_agents: int = 2,
    max_rounds_without_convergence: int | None = None,
    require_votes: bool = True,
) -> GateOutcome:
    """Refuse uncertified multi-agent finalize (COLLECTIVE-CERT / arXiv 2608.05956).

    Rules:

    * No votes when required → **FAIL_LOUD**
    * Fewer than ``min_agents`` → **FAIL_LOUD**
    * ``decision=finalize`` without convergence → **FAIL** (black-box stop)
    * ``decision=finalize`` past deadline without agreement → **FAIL**
    * ``decision=continue`` when already converged → **FAIL** (waste rounds)
    * ``decision=continue`` while not converged (under round budget) → **PASS**
    * ``decision=finalize`` when converged → **PASS** + certificate fields
    """
    dec = (decision or "").strip().lower()
    if dec not in {"continue", "finalize"}:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=f"COLLECTIVE-CERT: unknown decision={decision!r} (use continue|finalize)",
            exit_code=2,
            human_required=True,
        )

    if not votes:
        if require_votes:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "COLLECTIVE-CERT: no agent votes — refuse consensus certificate "
                    "before interaction traces exist (arXiv 2608.05956 black-box class)"
                ),
                exit_code=2,
                human_required=True,
            )
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="COLLECTIVE-CERT: no votes required",
            exit_code=0,
        )

    try:
        cert = certify_consensus(
            votes, min_agreement=min_agreement, max_factions=max_factions
        )
    except (TypeError, ValueError) as exc:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=f"COLLECTIVE-CERT: invalid vote payload: {exc}",
            exit_code=2,
            human_required=True,
        )

    if cert.n_agents < min_agents:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(
                f"COLLECTIVE-CERT: n_agents={cert.n_agents} < min_agents={min_agents} "
                "— collective certificate needs multi-agent interaction"
            ),
            exit_code=2,
            human_required=True,
            contradictions_found=cert.faction_count,
            consistency_score=cert.agreement,
        )

    if max_rounds_without_convergence is not None:
        if (
            not cert.converged
            and cert.n_rounds > max_rounds_without_convergence
            and dec == "continue"
        ):
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(
                    f"COLLECTIVE-CERT: n_rounds={cert.n_rounds} exceeds "
                    f"max_rounds_without_convergence={max_rounds_without_convergence} "
                    f"with agreement={cert.agreement:.3f} λ2_hat={cert.lambda2_hat:.3f} "
                    f"factions={cert.faction_count} — stuck collective; escalate"
                ),
                exit_code=1,
                human_required=True,
                consistency_score=cert.agreement,
                contradictions_found=cert.faction_count,
            )

    if dec == "finalize" and not cert.converged:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"COLLECTIVE-CERT: decision=finalize but not converged "
                f"agreement={cert.agreement:.3f} (min={min_agreement}) "
                f"factions={cert.faction_count} (max={max_factions}) "
                f"λ2_hat={cert.lambda2_hat:.3f} deadline_rounds={cert.deadline_rounds} "
                f"plurality={cert.plurality_choice!r} — refuse black-box stop "
                f"(arXiv 2608.05956)"
            ),
            exit_code=1,
            human_required=True,
            consistency_score=cert.agreement,
            contradictions_found=cert.faction_count,
        )

    if dec == "continue" and cert.converged:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"COLLECTIVE-CERT: decision=continue but already converged "
                f"agreement={cert.agreement:.3f} choice={cert.plurality_choice!r} "
                f"— refuse extra debate rounds after certificate (waste)"
            ),
            exit_code=1,
            human_required=False,
            consistency_score=cert.agreement,
            contradictions_found=0,
        )

    if dec == "finalize":
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=(
                f"COLLECTIVE-CERT ok: finalize choice={cert.plurality_choice!r} "
                f"agreement={cert.agreement:.3f} agents={cert.n_agents} "
                f"rounds={cert.n_rounds} λ2_hat={cert.lambda2_hat:.3f}"
            ),
            exit_code=0,
            human_required=False,
            consistency_score=cert.agreement,
            contradictions_found=0,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"COLLECTIVE-CERT ok: continue agreement={cert.agreement:.3f} "
            f"factions={cert.faction_count} λ2_hat={cert.lambda2_hat:.3f} "
            f"deadline_rounds={cert.deadline_rounds}"
        ),
        exit_code=0,
        human_required=False,
        consistency_score=cert.agreement,
        contradictions_found=cert.faction_count,
    )

full_consistency_check(store, auto_repair=True)

Run a full consistency check across ALL NPCs and optionally auto-repair contradictions.

Source code in src/worldoracle/consistency.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def full_consistency_check(store: WorldOracleStore, auto_repair: bool = True) -> ConsistencyReport:
    """Run a full consistency check across ALL NPCs and optionally auto-repair contradictions."""
    npc_ids = store.list_npc_ids()
    detector = ContradictionDetector()
    repairer = BeliefRepairer()

    total_predicates = 0
    all_contradictions: list[tuple[Any, Any]] = []
    by_subject: dict[str, int] = defaultdict(int)

    for npc_id in npc_ids:
        state = store.get_belief_state(npc_id)
        total_predicates += len(state.predicates)
        pairs = detector.detect(state)
        for pred_a, pred_b in pairs:
            all_contradictions.append((pred_a, pred_b))
            by_subject[pred_a.subject] += 1

    contradictions_found = len(all_contradictions)
    contradictions_repaired = 0
    repair_summary: list[str] = []

    if auto_repair:
        for pred_a, pred_b in all_contradictions:
            frame = repairer.repair(pred_a, pred_b)
            store.save_repair(frame)
            # Remove the "losing" predicate
            # resolved_value is the winner's value; if both or neither match (edge case),
            # fall back to deleting the less-confident predicate.
            if pred_a.value == frame.resolved_value and pred_b.value != frame.resolved_value:
                loser = pred_b
            elif pred_b.value == frame.resolved_value and pred_a.value != frame.resolved_value:
                loser = pred_a
            else:
                # Neither or both match resolved_value - delete the less-confident one
                loser = pred_a if pred_a.confidence <= pred_b.confidence else pred_b
            store._conn.execute("DELETE FROM predicates WHERE id=?", (loser.id,))
            store._conn.commit()
            contradictions_repaired += 1
            repair_summary.append(
                f"Repaired: {pred_a.subject}.{pred_a.attribute} - "
                f"kept '{frame.resolved_value}' (strategy: {frame.strategy})"
            )

    unresolved = contradictions_found - contradictions_repaired

    if total_predicates == 0:
        consistency_score = 1.0
    else:
        consistency_score = 1.0 - (contradictions_found / total_predicates)
    consistency_score = max(0.0, min(1.0, consistency_score))

    # most_contested: top 5 subjects by contradiction count
    most_contested = sorted(by_subject.keys(), key=lambda s: by_subject[s], reverse=True)[:5]

    return ConsistencyReport(
        total_predicates=total_predicates,
        contradictions_found=contradictions_found,
        contradictions_repaired=contradictions_repaired,
        unresolved=unresolved,
        consistency_score=consistency_score,
        by_subject=dict(by_subject),
        most_contested=most_contested,
        repair_summary=repair_summary,
    )

diff_belief_states(store, before_timestamp, after_timestamp, subject=None)

Diff belief state at two points in time using the snapshots table.

Source code in src/worldoracle/diff.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def diff_belief_states(
    store: WorldOracleStore,
    before_timestamp: float,
    after_timestamp: float,
    subject: str | None = None,
) -> BeliefDiff:
    """Diff belief state at two points in time using the snapshots table."""
    conn = store._conn
    # Check if snapshots table exists
    tbl = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name='belief_snapshots'"
    ).fetchone()
    if tbl is None:
        return BeliefDiff(0, 0, [], 0, 0, 0, 0, "No snapshot data available.")

    # Get latest belief for each (subject, attribute) before or at timestamp.
    # We first find the most-recent snapshot taken at or before `ts`, then
    # return only the beliefs that were recorded in that snapshot.
    def get_beliefs_at(ts: float) -> dict[tuple[str, str], tuple[Any, Any]]:
        snap_row = conn.execute(
            "SELECT snapshot_id FROM snapshot_registry WHERE taken_at <= ? "
            "ORDER BY taken_at DESC LIMIT 1",
            (ts,),
        ).fetchone()
        if snap_row is None:
            return {}
        snap_id = snap_row["snapshot_id"]
        if subject:
            sql = """
                SELECT subject, attribute, value, confidence
                FROM belief_snapshots
                WHERE snapshot_id=? AND subject=?
            """
            params: tuple[Any, ...] = (snap_id, subject)
        else:
            sql = """
                SELECT subject, attribute, value, confidence
                FROM belief_snapshots
                WHERE snapshot_id=?
            """
            params = (snap_id,)
        rows = conn.execute(sql, params).fetchall()
        return {(r["subject"], r["attribute"]): (r["value"], r["confidence"]) for r in rows}

    before_beliefs = get_beliefs_at(before_timestamp)
    after_beliefs = get_beliefs_at(after_timestamp)

    all_keys = set(before_beliefs.keys()) | set(after_beliefs.keys())
    changes = []
    stable = 0

    for subj, attr in all_keys:
        in_before = (subj, attr) in before_beliefs
        in_after = (subj, attr) in after_beliefs

        if in_before and not in_after:
            old_v, old_c = before_beliefs[(subj, attr)]
            changes.append(BeliefChange(subj, attr, old_v, None, old_c, None, "removed"))
        elif not in_before and in_after:
            new_v, new_c = after_beliefs[(subj, attr)]
            changes.append(BeliefChange(subj, attr, None, new_v, None, new_c, "added"))
        else:
            old_v, old_c = before_beliefs[(subj, attr)]
            new_v, new_c = after_beliefs[(subj, attr)]
            if old_v != new_v:
                changes.append(
                    BeliefChange(subj, attr, old_v, new_v, old_c, new_c, "value_changed")
                )
            elif abs(old_c - new_c) > 1e-9:
                changes.append(
                    BeliefChange(subj, attr, old_v, new_v, old_c, new_c, "confidence_changed")
                )
            else:
                stable += 1

    added = sum(1 for c in changes if c.change_type == "added")
    removed = sum(1 for c in changes if c.change_type == "removed")
    modified = sum(1 for c in changes if c.change_type in ("value_changed", "confidence_changed"))
    before_count = len(before_beliefs)
    after_count = len(after_beliefs)

    summary = (
        f"Before: {before_count} beliefs, After: {after_count} beliefs. "
        f"Added: {added}, Removed: {removed}, Modified: {modified}, Stable: {stable}."
    )
    return BeliefDiff(before_count, after_count, changes, added, removed, modified, stable, summary)

print_beliefs(state, console=None)

Print a belief state as a rich table.

Source code in src/worldoracle/report.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def print_beliefs(state: BeliefState, console: Console | None = None) -> None:
    """Print a belief state as a rich table."""
    con = console or _console
    table = Table(title=f"Beliefs: {state.npc_id}", show_header=True)
    table.add_column("Subject")
    table.add_column("Attribute")
    table.add_column("Value")
    table.add_column("Source")
    table.add_column("Confidence")
    for p in state.predicates:
        table.add_row(
            p.subject,
            p.attribute,
            str(p.value),
            p.source,
            f"{p.confidence:.2f}",
        )
    con.print(table)

print_repairs(repairs, console=None)

Print a list of repair frames as a rich table, or a 'no repairs' message.

Source code in src/worldoracle/report.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def print_repairs(repairs: list[RepairFrame], console: Console | None = None) -> None:
    """Print a list of repair frames as a rich table, or a 'no repairs' message."""
    con = console or _console
    if not repairs:
        con.print("[green]No repairs needed.[/green]")
        return
    table = Table(title="Repair Frames", show_header=True)
    table.add_column("ID")
    table.add_column("Strategy")
    table.add_column("Resolved Value")
    table.add_column("Reason")
    for r in repairs:
        table.add_row(r.id, r.strategy, str(r.resolved_value), r.reason)
    con.print(table)

to_json(state, repairs=None)

Serialize a BeliefState (and optional repairs) to JSON.

Source code in src/worldoracle/report.py 
51
52
53
54
55
56
def to_json(state: BeliefState, repairs: list[RepairFrame] | None = None) -> str:
    """Serialize a BeliefState (and optional repairs) to JSON."""
    data = state.to_dict()
    if repairs is not None:
        data["repairs"] = [r.to_dict() for r in repairs]
    return json.dumps(data, indent=2)

to_markdown(states)

Render a list of BeliefStates as a Markdown report.

Source code in src/worldoracle/report.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def to_markdown(states: list[BeliefState]) -> str:
    """Render a list of BeliefStates as a Markdown report."""
    lines = ["# worldoracle Belief Report", ""]
    for state in states:
        lines.append(f"## NPC: {state.npc_id}")
        lines.append("")
        lines.append("| Subject | Attribute | Value | Source | Confidence |")
        lines.append("|---------|-----------|-------|--------|------------|")
        for p in state.predicates:
            lines.append(
                f"| {p.subject} | {p.attribute} | {p.value} | {p.source} | {p.confidence:.2f} |"
            )
        lines.append("")
    return "\n".join(lines)