Spaces:
Running
Running
File size: 3,621 Bytes
414dc55 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """The Director - deterministic adjudication of an interrogation turn.
This is the mechanical authority. The model's ``internal`` block is advisory; whether
a lie is caught, a contradiction fires, or a fact is genuinely revealed is decided
here by comparing presented evidence and claimed reveals against ground truth. A
table lookup, never a second LLM call - so the per-turn path is exactly one model
call, and a jailbroken confession in prose changes nothing here.
"""
from __future__ import annotations
from dataclasses import dataclass
from ..schemas.case import CaseFile
from ..schemas.enums import Relevance
from ..schemas.interrogation import InterrogationTurn
from ..schemas.suspect import Suspect
from .game_state import SuspectState
from .relevance import assess_relevance
_STRESS_ON_BREAK = 0.40
_STRESS_ON_DIRECT = 0.15
_STRESS_ON_TANGENTIAL = 0.05
_STRESS_DECAY = 0.95
_RAPPORT_NEUTRAL = 0.02
_RAPPORT_ON_BREAK = -0.12
@dataclass(frozen=True)
class Adjudication:
relevance: Relevance
is_contradiction: bool
broken_lie_id: str | None
revealed_fact_ids: tuple[str, ...]
stress_after: float
rapport_after: float
def _clamp(value: float, low: float, high: float) -> float:
return max(low, min(high, value))
def _grounded_reveals(
suspect: Suspect, turn: InterrogationTurn, already: frozenset[str]
) -> tuple[str, ...]:
"""Accept a model-claimed reveal only if the suspect actually knows that fact and
it is not one they must conceal. Keeps reveals grounded in the suspect's slice."""
known = set(suspect.knows_facts)
concealed = set(suspect.must_lie_about)
out: list[str] = []
for fid in turn.internal.revealed_fact_ids:
if fid in known and fid not in concealed and fid not in already:
out.append(fid)
return tuple(out)
def adjudicate(
case: CaseFile,
suspect: Suspect,
state: SuspectState,
turn: InterrogationTurn,
presented_clue_id: str | None,
) -> Adjudication:
result = assess_relevance(case, suspect, presented_clue_id)
is_contradiction = False
broken_lie_id: str | None = None
reveals: list[str] = list(_grounded_reveals(suspect, turn, state.revealed_fact_ids))
if result.relevance is Relevance.BREAKING:
lie_id = result.broken_lie_id
# Fire once: dedup by the anchored lie when known, else by the presented clue.
if lie_id is not None:
already_fired = lie_id in state.broken_lie_ids
else:
already_fired = presented_clue_id is not None and presented_clue_id in state.evidence_shown
if not already_fired:
is_contradiction = True
broken_lie_id = lie_id
if lie_id is not None:
lie = next((x for x in suspect.anchored_lies if x.lie_id == lie_id), None)
if lie and lie.truth_ref and lie.truth_ref not in reveals:
reveals.append(lie.truth_ref)
stress = state.stress * _STRESS_DECAY
if result.relevance is Relevance.BREAKING:
stress += _STRESS_ON_BREAK
elif result.relevance is Relevance.DIRECT:
stress += _STRESS_ON_DIRECT
elif result.relevance is Relevance.TANGENTIAL:
stress += _STRESS_ON_TANGENTIAL
rapport = state.rapport + (_RAPPORT_ON_BREAK if is_contradiction else _RAPPORT_NEUTRAL)
return Adjudication(
relevance=result.relevance,
is_contradiction=is_contradiction,
broken_lie_id=broken_lie_id,
revealed_fact_ids=tuple(reveals),
stress_after=_clamp(stress, 0.0, 1.0),
rapport_after=_clamp(rapport, -1.0, 1.0),
)
|