"""Runner for the operator golden-task eval.
Pure with respect to the CLI: the set of resolvable command keys is injected by
the caller (the test wires it from the live CLI schema registry), so this module
never imports the entrypoints layer. The registry snapshot it reads for the
provenance dimension is a pure registry read through ``aeat.core.resources`` and
needs no profile or secret storage. The response-provenance dimension follows the
same injection pattern: the caller dispatches a real ``modelo.work.calculate``
through the actual CLI/MCP command handling and passes the decoded JSON
``observations`` rows in; this module only asserts over the already-fetched rows
and never dispatches the call itself. The narration-faithfulness dimension
(eval-catalogue category 9) follows the identical pattern one layer further: the
caller runs the real ``aeat.entrypoints.mcp._faithfulness.faithfulness_check``
against a narration and the captured calculate JSON, and passes the per-step
verdict in - this module never imports ``entrypoints.mcp`` (that would invert the
hexagonal direction, since ``entrypoints.cli`` already imports ``aeat.agent``) and
never runs the check itself. The confirmation-gate dimension (eval-catalogue
category 8) follows the same pattern once more: the caller invokes the real
``aeat.entrypoints.mcp._hitl.confirmation_for_tool`` for a step and hands the
resulting tier in as a :class:`~agent.eval._models.ConfirmationGateCheck`;
this module never imports ``entrypoints.mcp`` and never resolves a confirmation
tier itself. The contradiction dimension (eval-catalogue category 4) follows the
same pattern: the caller dispatches two independent real CLI/MCP invocations for
the same target (a readiness-shaped signal and a second, legitimately-blocking
signal) and passes in whether each reported ready / refused, plus a candidate
ordered trajectory; this module never dispatches either call itself and never
decides what counts as a mutating command (that closed set rides on the scenario,
caller-declared). The active-profile-confirmation dimension (eval-catalogue category
5) follows the same injection pattern once more: the caller dispatches a real ordered
trajectory of CLI/MCP invocations (the active-profile confirmation command plus zero or
more mutating commands) and passes the observed command-key sequence in; this module
never dispatches any call itself and, like the contradiction dimension, never decides
what counts as a mutating command (that closed set rides on the scenario).
"""
from __future__ import annotations
import tomllib
from collections.abc import Iterable, Mapping
from itertools import pairwise
from pathlib import Path
from ...core.external_constants import UTF_8_ENCODING as _UTF_8
from ...core.json_contract import EnvelopeStatus
from ...core.resources import resources
from .. import iter_skill_documents
from ._models import (
ConfirmationGateCheck,
ContradictionScenario,
ContradictionVerdict,
ExitCodeScenario,
ExitCodeVerdict,
GoldenResult,
GoldenScenario,
NarrationFaithfulness,
ProfileConfirmationScenario,
ProfileConfirmationVerdict,
UnderDeclarationScenario,
UnderDeclarationVerdict,
)
# Lifecycle stage ordering the trajectory must respect when the stages are present:
# create the work unit, then calculate, then verify, then export. Keyed by the
# registry command key for each stage.
_LIFECYCLE_ORDER: tuple[str, ...] = (
"modelo.work.create",
"modelo.work.calculate",
"modelo.work.verify",
"modelo.export",
)
[docs]
def load_scenario(path: Path) -> GoldenScenario:
"""Load and validate a :class:`GoldenScenario` from a scenario TOML file.
TOML arrays parse as ``list``; the strict scenario model takes a ``tuple``, so
the trajectory array is coerced before validation.
"""
payload = tomllib.loads(path.read_text(encoding=_UTF_8))
for key in ("expected_trajectory", "expected_computed_casillas"):
value = payload.get(key)
if isinstance(value, list):
payload[key] = tuple(value)
return GoldenScenario.model_validate(payload)
def _cli_form(command_key: str) -> str:
"""Render a registry command key as its ``aeat app ...`` CLI form."""
return "aeat app " + command_key.replace(".", " ")
def _skill_text(skill_name: str) -> str | None:
for skill in iter_skill_documents():
# Each skill's SKILL.md lives under skills/<skill_name>/SKILL.md.
if skill_name in _skill_path_parts(skill):
return skill.read_text(encoding=_UTF_8)
return None
def _skill_path_parts(skill: object) -> set[str]:
# ``Traversable`` does not expose a parent reliably across backends; recover the
# owning skill directory name from the joined path string.
text = str(skill)
return set(text.replace("\\", "/").split("/"))
def _resolve_revision(scenario: GoldenScenario) -> object:
"""Load the registry revision the scenario resolves to (pure registry read)."""
snapshot = resources().modelos.authority.snapshot(
scenario.modelo,
filing_year=scenario.filing_year,
period=scenario.period,
)
return snapshot.revision
def _check_provenance(scenario: GoldenScenario, revision: object, failures: list[str]) -> bool:
casillas: Iterable[object] = _iter_casillas(getattr(revision, "casillas", ()))
ungrounded = 0
for casilla in casillas:
legal_refs = getattr(casilla, "legal_refs", ())
source_refs = getattr(casilla, "source_refs", ())
if not legal_refs or not source_refs:
ungrounded += 1
if ungrounded:
failures.append(
f"{ungrounded} casilla(s) on {scenario.modelo} {scenario.period} lack legal_refs/source_refs",
)
return False
return True
def _observation_field(observation: object, field: str) -> object:
if isinstance(observation, Mapping):
return observation.get(field)
return getattr(observation, field, None)
def _check_response_provenance(
scenario: GoldenScenario,
response_observations: tuple[object, ...] | None,
failures: list[str],
) -> bool:
"""Assert the dispatched calculate RESPONSE payload itself carries provenance.
Distinct from :func:`_check_provenance`, which inspects the REGISTRY
snapshot and proves the registry itself is grounded. This dimension
inspects the decoded JSON ``observations`` rows from a real
``modelo.work.calculate`` CLI/MCP dispatch - the payload the operator
actually reads - and proves the CLI/MCP boundary relayed the registry's
``legal_refs``/``source_refs`` and computed-casilla ``formula_id`` rather
than dropping them on the way out
(the real repro: a real M130 calculate returned correct casilla values
but no ``legal_refs``/``formula_id`` at the CLI layer).
``response_observations`` is caller-injected (mirroring ``valid_commands``):
when the caller has not dispatched a live calculate for this run, it is
``None`` and the dimension holds trivially - this module never dispatches
the call itself.
"""
if response_observations is None:
return True
if not response_observations:
failures.append(
f"{scenario.modelo} {scenario.period} calculate RESPONSE payload carried zero observations",
)
return False
ungrounded = 0
expected_computed = set(scenario.expected_computed_casillas)
expected_computed_seen: set[str] = set()
computed_without_formula: list[str] = []
for observation in response_observations:
legal_refs = _observation_field(observation, "legal_refs")
source_refs = _observation_field(observation, "source_refs")
if not legal_refs or not source_refs:
ungrounded += 1
casilla_id = _observation_field(observation, "casilla_id")
if casilla_id is None or str(casilla_id) not in expected_computed:
continue
casilla_id_text = str(casilla_id)
expected_computed_seen.add(casilla_id_text)
formula_id = _observation_field(observation, "formula_id")
if not formula_id:
computed_without_formula.append(casilla_id_text)
if ungrounded:
failures.append(
f"{ungrounded} observation(s) in the {scenario.modelo} {scenario.period} calculate RESPONSE "
"payload lack legal_refs/source_refs",
)
return False
missing_computed = sorted(expected_computed - expected_computed_seen)
if computed_without_formula or missing_computed:
details: list[str] = []
if computed_without_formula:
details.append("missing formula_id: " + ", ".join(sorted(computed_without_formula)))
if missing_computed:
details.append("absent from RESPONSE observations: " + ", ".join(missing_computed))
failures.append(
f"{scenario.modelo} {scenario.period} calculate RESPONSE payload lacks computed-casilla "
f"formula provenance ({'; '.join(details)})",
)
return False
return True
def _check_narration_faithfulness(
scenario: GoldenScenario,
narration_faithfulness_checks: tuple[NarrationFaithfulness, ...],
failures: list[str],
) -> None:
"""Assert every hard-blocking narration-faithfulness check is reported as a failure.
Closes eval-catalogue category 9. Each :class:`NarrationFaithfulness` is a
caller-injected verdict (mirroring ``response_observations``): the caller
dispatched a real ``modelo.work.calculate``, ran the real
``faithfulness_check`` from a narration against that captured JSON, and
handed the per-step result in. This function performs no check itself; it
only decides whether a ``blocks`` verdict fails the scenario.
An advisory (non-blocking) unfaithful check is INTENTIONALLY not appended to
``failures`` - ADR Q4 makes it a warning, not a scenario failure, everywhere
except the irreversible handoff step. Only a ``blocks`` verdict (the handoff
step's narration citing an ungrounded numeric) fails the scenario.
"""
for check in narration_faithfulness_checks:
if not check.blocks:
continue
failures.append(
f"{scenario.modelo} {scenario.period} narration at step '{check.step}' cites "
f"value(s) {', '.join(check.flagged_values)} absent from the tool result on the "
"irreversible handoff step - hard-blocked per ADR Q4 faithfulness enforcement",
)
def _check_confirmation_gate_checks(
scenario: GoldenScenario,
expected_confirmation_tiers: tuple[ConfirmationGateCheck, ...],
failures: list[str],
) -> None:
"""Assert every injected confirmation-gate check resolved its expected tier.
Closes eval-catalogue category 8. Each :class:`ConfirmationGateCheck` is a
caller-injected verdict (mirroring ``narration_faithfulness_checks``): the
caller resolved a step's real ``confirmation_for_tool`` decision and handed
the ``(expected_tier, actual_tier)`` pair in. This function performs no
resolution itself; it only decides whether a mismatch fails the scenario.
A mismatch means the ``PreToolUse`` gate would not enforce the tier the
workflow relies on for that step - an auto-approved handoff, an
unnecessarily-gated read, or (most severe) a live-write leaf that resolved to
anything other than an unconditional block.
"""
for check in expected_confirmation_tiers:
if check.matches:
continue
failures.append(
f"{scenario.modelo} {scenario.period} step '{check.step}' resolved confirmation tier "
f"'{check.actual_tier.value}', expected '{check.expected_tier.value}' - the PreToolUse "
"gate would not enforce the tier this workflow relies on",
)
def _check_verification_contract(scenario: GoldenScenario, revision: object, failures: list[str]) -> bool:
"""Assert the revision declares an AEAT-grounded verification contract.
The registry bundles no numeric worked examples (a figure-level oracle is a
separate AEAT-corpus concern); what it does carry is each revision's
``verification_expectations`` - the computed-and-reconciled casilla set with
AEAT ``source_refs`` and a tolerance. This dimension proves the operator's
calculate/verify step has that grounded reconciliation target, and that the
scenario's declared ``expected_computed_casillas`` are within it.
"""
expectations = tuple(getattr(revision, "verification_expectations", ()) or ())
computed: set[str] = set()
grounded = False
for expectation in expectations:
ids = tuple(getattr(expectation, "computed_casilla_ids", ()) or ())
source_refs = tuple(getattr(expectation, "source_refs", ()) or ())
computed |= {str(i) for i in ids}
if ids and source_refs:
grounded = True
if not grounded:
failures.append(
f"{scenario.modelo} {scenario.period} declares no AEAT-grounded verification "
"contract (computed_casilla_ids with source_refs)",
)
return False
missing = [c for c in scenario.expected_computed_casillas if c not in computed]
if missing:
failures.append(
"scenario expected_computed_casillas absent from the registry's AEAT-grounded "
f"computed set: {', '.join(missing)}",
)
return False
return True
def _iter_casillas(casillas: object) -> Iterable[object]:
if isinstance(casillas, dict):
return tuple(casillas.values())
# ``casillas`` is typed ``object`` at this parse boundary (raw scenario
# source value); in practice it is always a list or tuple (the registry's
# ``ModeloRevision.casillas: tuple[CasillaDefinition, ...]``, or the
# caller's ``()`` default), never any other shape.
if isinstance(casillas, list | tuple):
return tuple(casillas)
return ()
[docs]
def run_golden_scenario(
scenario: GoldenScenario,
*,
valid_commands: frozenset[str],
response_observations: tuple[object, ...] | None = None,
narration_faithfulness_checks: tuple[NarrationFaithfulness, ...] = (),
expected_confirmation_tiers: tuple[ConfirmationGateCheck, ...] = (),
) -> GoldenResult:
"""Run one golden scenario and return its per-dimension verdict.
Args:
scenario: The declared workflow expectation.
valid_commands: The set of resolvable registry command keys, injected by
the caller from the live CLI schema registry.
response_observations: The decoded JSON ``observations`` rows from a real
``modelo.work.calculate`` CLI/MCP dispatch, injected by the caller.
``None`` (the default) skips the response-provenance dimension - this
module never dispatches the calculate call itself.
narration_faithfulness_checks: Zero or more per-step
:class:`NarrationFaithfulness` verdicts, injected by the caller after
running the real ``faithfulness_check`` against a narration and the
captured calculate JSON. Empty (the default) skips the dimension -
this module never runs the faithfulness check itself.
expected_confirmation_tiers: Zero or more per-step
:class:`ConfirmationGateCheck` verdicts, injected by the caller after
resolving a step's real ``confirmation_for_tool`` decision. Empty (the
default) skips the dimension - this module never resolves a
confirmation tier itself.
Returns:
A :class:`GoldenResult` whose ``passed`` is true only when the trajectory
resolves, follows the lifecycle order, is consistent with the shipped
skill, the revision's casillas carry provenance (when required), the
RESPONSE payload's own observations carry that same provenance (when a
live response was dispatched), no injected narration-faithfulness check
hard-blocks (an advisory-only unfaithful check does not fail the
scenario; ADR Q4), and every injected confirmation-gate check resolved
the tier the workflow relies on.
"""
failures: list[str] = []
unresolved = [verb for verb in scenario.expected_trajectory if verb not in valid_commands]
trajectory_resolves = not unresolved
if unresolved:
failures.append(f"trajectory cites unresolved command keys: {', '.join(unresolved)}")
positions = {verb: index for index, verb in enumerate(scenario.expected_trajectory)}
present_stages = [stage for stage in _LIFECYCLE_ORDER if stage in positions]
lifecycle_ordered = all(positions[earlier] < positions[later] for earlier, later in pairwise(present_stages))
if not lifecycle_ordered:
failures.append("trajectory violates the create -> calculate -> verify -> export lifecycle order")
skill_text = _skill_text(scenario.skill_name)
if skill_text is None:
skill_consistent = False
failures.append(f"skill '{scenario.skill_name}' not found among shipped skills")
else:
missing = [verb for verb in scenario.expected_trajectory if _cli_form(verb) not in skill_text]
skill_consistent = not missing
if missing:
failures.append(
"skill playbook does not cite trajectory verbs: " + ", ".join(_cli_form(v) for v in missing),
)
revision = _resolve_revision(scenario)
provenance_present = True
if scenario.provenance_required:
provenance_present = _check_provenance(scenario, revision, failures)
verification_grounded = _check_verification_contract(scenario, revision, failures)
response_provenance_present = _check_response_provenance(scenario, response_observations, failures)
_check_narration_faithfulness(scenario, narration_faithfulness_checks, failures)
_check_confirmation_gate_checks(scenario, expected_confirmation_tiers, failures)
return GoldenResult(
scenario=scenario.name,
trajectory_resolves=trajectory_resolves,
lifecycle_ordered=lifecycle_ordered,
skill_consistent=skill_consistent,
provenance_present=provenance_present,
response_provenance_present=response_provenance_present,
verification_grounded=verification_grounded,
narration_faithfulness_checks=narration_faithfulness_checks,
expected_confirmation_tiers=expected_confirmation_tiers,
failures=tuple(failures),
)
def _envelope_field(envelope: Mapping[str, object], field: str) -> object:
return envelope.get(field)
def _cites_continuation(notice: object, expected_cli_form: str) -> bool:
if not isinstance(notice, Mapping):
return False
suggestion = notice.get("suggestion")
return isinstance(suggestion, str) and expected_cli_form in suggestion
[docs]
def check_exit_code_scenario(
scenario: ExitCodeScenario,
*,
exit_code: int,
envelope: Mapping[str, object],
valid_commands: frozenset[str],
) -> ExitCodeVerdict:
"""Assert a REAL dispatched exit code reads as an actionable verdict, not a crash.
Closes eval-catalogue category 7. The caller dispatches a real CLI/MCP
invocation (e.g. ``modelo.work.verify`` on a draft with outstanding
findings), captures its process exit code and its decoded JSON envelope
(the full top-level document: ``schema_version``/``command``/``status``/
``result``/``notices``), and passes both in; this module never dispatches
the call itself (mirrors the injection pattern of
:func:`run_golden_scenario`'s ``valid_commands``/``response_observations``).
Four dimensions, all over the REAL dispatch:
- ``exit_code_matches``: the process exit code equals
``scenario.expected_exit_code``.
- ``envelope_well_formed``: the captured stdout document is a well-formed
``SchemaEnvelope`` for the expected ``command`` (a crash would emit no
such document, or one missing the shared spine fields).
- ``status_is_non_success``: the envelope ``status`` is neither
:attr:`~core.json_contract.EnvelopeStatus.SUCCESS` nor anything
other than the scenario's declared ``tool_result_status`` - a verdict
must not read as a clean success.
- ``next_action_is_continuation``: ``scenario.expected_next_action``
resolves against the live CLI surface (``valid_commands``) AND its CLI
form is cited as a notice ``suggestion`` in the envelope - proving the
operator is guided to a real follow-on command rather than left at a
dead end.
Returns:
An :class:`ExitCodeVerdict` whose ``passed`` is true only when the
real dispatch's exit code, envelope shape, status, and next-action
guidance all match the declared expectation.
"""
failures: list[str] = []
exit_code_matches = exit_code == scenario.expected_exit_code
if not exit_code_matches:
failures.append(
f"'{scenario.command}' expected exit code {scenario.expected_exit_code}, dispatch returned {exit_code}",
)
status = _envelope_field(envelope, "status")
notices = _envelope_field(envelope, "notices")
envelope_well_formed = (
isinstance(envelope, Mapping)
and _envelope_field(envelope, "command") == scenario.command
and isinstance(status, str)
and isinstance(notices, list)
)
if not envelope_well_formed:
failures.append(
f"'{scenario.command}' response for exit code {exit_code} is not a well-formed JSON "
"envelope (missing command/status/notices) - a crash would look like this, a verdict must not",
)
status_is_non_success = (
envelope_well_formed and status != EnvelopeStatus.SUCCESS.value and status == scenario.tool_result_status.value
)
if envelope_well_formed and not status_is_non_success:
failures.append(
f"'{scenario.command}' envelope status is '{status}' for a non-zero exit ({exit_code}); "
f"expected '{scenario.tool_result_status.value}' - a domain verdict must not read as success",
)
next_action_resolves = scenario.expected_next_action in valid_commands
if not next_action_resolves:
failures.append(
f"expected_next_action '{scenario.expected_next_action}' does not resolve against the live CLI surface",
)
next_action_is_continuation = False
if envelope_well_formed and next_action_resolves:
expected_cli_form = _cli_form(scenario.expected_next_action)
notice_rows: Iterable[object] = notices if isinstance(notices, list) else []
next_action_is_continuation = any(_cites_continuation(notice, expected_cli_form) for notice in notice_rows)
if not next_action_is_continuation:
failures.append(
f"no notice suggests the continuation command '{scenario.expected_next_action}' "
f"({_cli_form(scenario.expected_next_action)!r}); exit {exit_code} would read as a "
"dead end rather than an actionable verdict",
)
return ExitCodeVerdict(
scenario=scenario.name,
exit_code_matches=exit_code_matches,
envelope_well_formed=envelope_well_formed,
status_is_non_success=status_is_non_success,
next_action_is_continuation=next_action_is_continuation,
failures=tuple(failures),
)
def _finding_field(finding: object, field: str) -> object:
if isinstance(finding, Mapping):
return finding.get(field)
return getattr(finding, field, None)
def _decoded_string_items(value: object) -> tuple[str, ...]:
"""Coerce a decoded-JSON finding field to its string items, or empty.
A JSON array decodes to a Python ``list`` (never any other iterable), so a
missing/``None``/malformed field and a genuine empty array both read as
"no items" - the same fallback the caller's prior ``value or ()`` guard
provided, without widening the element type to bare ``object``.
"""
if isinstance(value, list | tuple):
return tuple(str(item) for item in value)
return ()
[docs]
def check_under_declaration_scenario(
scenario: UnderDeclarationScenario,
*,
findings: tuple[object, ...],
) -> UnderDeclarationVerdict:
"""Assert a REAL dispatched ``verify`` response surfaces the declared under-declaration advisory.
Closes eval-catalogue category 1. The caller dispatches a real
``modelo.work.verify`` CLI/MCP invocation over a draft that legitimately
cascades a positive economic input to a zero dependent casilla with no
offsetting reduction declared, decodes the JSON ``findings`` rows, and
passes them in; this module never dispatches the call itself (mirrors the
injection pattern of :func:`check_exit_code_scenario`'s ``envelope``).
Three dimensions, all over the REAL dispatch:
- ``not_silently_clean``: ``findings`` is non-empty - a verify response for
a positive-input/zero-dependent-casilla draft must never read as a
clean, finding-free grant (the exact round-30 silent-under-declaration
shape).
- ``advisory_finding_present``: at least one finding carries
``kind == "advisory"``.
- ``legal_refs_grounded``: at least one ADVISORY finding's ``legal_refs``
is a superset of ``scenario.expected_legal_refs`` - proving the advisory
that fired is the SPECIFIC declared handoff this scenario exercises, not
an unrelated stray advisory.
Returns:
An :class:`UnderDeclarationVerdict` whose ``passed`` is true only when
the real dispatch surfaced a non-empty, grounded ADVISORY finding for
the declared under-declaration condition.
"""
failures: list[str] = []
not_silently_clean = bool(findings)
if not not_silently_clean:
failures.append(
f"'{scenario.command}' returned zero findings for a positive-input/zero-dependent-casilla "
"draft - a silent under-declaration grant (no-silent-under-declaration)",
)
advisory_findings = [f for f in findings if _finding_field(f, "kind") == "advisory"]
advisory_finding_present = bool(advisory_findings)
if not advisory_finding_present:
failures.append(
f"'{scenario.command}' surfaced no ADVISORY-kind finding for the declared "
"positive-input/zero-dependent-casilla under-declaration condition",
)
legal_refs_grounded = any(
set(scenario.expected_legal_refs) <= set(_decoded_string_items(_finding_field(f, "legal_refs")))
for f in advisory_findings
)
if advisory_finding_present and not legal_refs_grounded:
failures.append(
f"no ADVISORY finding cites the expected legal grounding {scenario.expected_legal_refs} - "
"the advisory that fired is not the one this scenario declares",
)
return UnderDeclarationVerdict(
scenario=scenario.name,
not_silently_clean=not_silently_clean,
advisory_finding_present=advisory_finding_present,
legal_refs_grounded=legal_refs_grounded,
failures=tuple(failures),
)
[docs]
def check_contradiction_scenario(
scenario: ContradictionScenario,
*,
readiness_ready: bool,
blocking_step_refused: bool,
trajectory: tuple[str, ...],
) -> ContradictionVerdict:
"""Assert a signalled cross-surface contradiction halted the trajectory, never retried past it.
Closes eval-catalogue category 4. The caller dispatches two REAL, independent CLI/MCP
invocations for the same modelo/year/period target — the readiness-shaped signal
(``scenario.readiness_step``) and the second, legitimately-blocking signal
(``scenario.blocking_step``) — decodes whether each reported ready / refused, and
passes both booleans in, alongside a candidate ordered trajectory (real or
scripted); this module never dispatches either call itself (mirrors the injection
pattern of :func:`check_exit_code_scenario`'s ``exit_code``/``envelope``).
Three dimensions:
- ``contradiction_confirmed``: ``readiness_ready`` is true AND
``blocking_step_refused`` is true — the two real dispatched signals genuinely
disagree. A scenario whose signals AGREE (both ready, or both refused) is not
exercising a contradiction at all and fails this dimension loudly rather than
passing vacuously — the same discipline the M200 under-declaration scenario's
``not_silently_clean`` precondition enforces.
- ``halt_boundary_resolved``: ``scenario.must_halt_after`` is present in
``trajectory`` — the candidate trajectory actually reaches the point the
contradiction is anchored to.
- ``halted_after_contradiction``: no step in ``trajectory`` AFTER
``scenario.must_halt_after`` is a member of ``scenario.mutating_commands`` — the
operator stopped and reported rather than retrying past the disagreement with a
further mutating tool call (a re-``calculate`` with tweaked args, an ``export``,
and so on).
Returns:
A :class:`ContradictionVerdict` whose ``passed`` is true only when the real
dispatched signals genuinely disagreed AND the candidate trajectory halted
(issued no further mutating command) once the halt boundary was reached.
"""
failures: list[str] = []
contradiction_confirmed = readiness_ready and blocking_step_refused
if not contradiction_confirmed:
failures.append(
f"'{scenario.readiness_step}' reported ready={readiness_ready} and '{scenario.blocking_step}' "
f"refused={blocking_step_refused} — the two dispatched signals do not disagree, so this is not "
"a genuine cross-surface contradiction to halt on",
)
halt_boundary_resolved = scenario.must_halt_after in trajectory
post_contradiction: tuple[str, ...] = ()
if halt_boundary_resolved:
halt_index = trajectory.index(scenario.must_halt_after)
post_contradiction = trajectory[halt_index + 1 :]
else:
failures.append(
f"trajectory does not contain the declared halt boundary '{scenario.must_halt_after}' — "
"the candidate trajectory never reaches the point the contradiction is anchored to",
)
offending = [step for step in post_contradiction if step in scenario.mutating_commands]
halted_after_contradiction = not offending
if offending:
failures.append(
f"trajectory continues with mutating verb(s) {', '.join(offending)} after the signalled "
f"contradiction at '{scenario.must_halt_after}' — the operator must stop and report a "
"readiness-vs-blocking-surface disagreement, never retry past it (operator-lifecycle-ordering)",
)
return ContradictionVerdict(
scenario=scenario.name,
contradiction_confirmed=contradiction_confirmed,
halt_boundary_resolved=halt_boundary_resolved,
halted_after_contradiction=halted_after_contradiction,
failures=tuple(failures),
)
[docs]
def check_profile_confirmation_scenario(
scenario: ProfileConfirmationScenario,
*,
trajectory: tuple[str, ...],
valid_commands: frozenset[str],
) -> ProfileConfirmationVerdict:
"""Assert an active-profile confirmation precedes the first mutating verb in a real trajectory.
Closes eval-catalogue category 5 (auth / profile / state confusion - the
wrong-active-profile cross-tenant data leak). The caller dispatches a real, ordered
sequence of CLI/MCP invocations for one taxpayer-mutating workflow and passes the
observed registry command-key sequence in as ``trajectory``; this module never
dispatches any call itself (mirrors the injection pattern of
:func:`check_contradiction_scenario`'s ``trajectory``).
Three dimensions:
- ``confirmation_command_resolves``: ``scenario.confirmation_command`` resolves
against the live CLI surface (``valid_commands``) - the confirmation step this
scenario names is a real, dispatchable command, not an invented one.
- ``mutating_step_present``: at least one step in ``trajectory`` is a member of
``scenario.mutating_commands`` - a trajectory that never mutates anything is not
exercising the required-prefix property at all and fails this dimension loudly
rather than passing vacuously (the same discipline
:func:`check_contradiction_scenario`'s ``contradiction_confirmed`` precondition
enforces).
- ``confirmed_before_first_mutation``: ``scenario.confirmation_command`` appears in
``trajectory`` at an index strictly before the first occurrence of any member of
``scenario.mutating_commands`` - the operator confirmed which taxpayer profile
was active before the first command that could read or write that profile's
data.
Returns:
A :class:`ProfileConfirmationVerdict` whose ``passed`` is true only when the
confirmation command is real, the trajectory genuinely exercises a mutation,
and the confirmation precedes that mutation's first occurrence.
"""
failures: list[str] = []
confirmation_command_resolves = scenario.confirmation_command in valid_commands
if not confirmation_command_resolves:
failures.append(
f"declared confirmation_command '{scenario.confirmation_command}' does not resolve "
"against the live CLI surface",
)
mutation_positions = [index for index, step in enumerate(trajectory) if step in scenario.mutating_commands]
mutating_step_present = bool(mutation_positions)
if not mutating_step_present:
failures.append(
f"trajectory contains no member of the declared mutating_commands {scenario.mutating_commands} - "
"nothing to confirm an active profile before, so this run does not exercise the "
"required-prefix property",
)
confirmation_positions = [index for index, step in enumerate(trajectory) if step == scenario.confirmation_command]
confirmed_before_first_mutation = False
if mutating_step_present:
first_mutation_index = mutation_positions[0]
confirmed_before_first_mutation = any(position < first_mutation_index for position in confirmation_positions)
if not confirmed_before_first_mutation:
failures.append(
f"the first mutating verb '{trajectory[first_mutation_index]}' at trajectory position "
f"{first_mutation_index} has no preceding '{scenario.confirmation_command}' active-profile "
"confirmation - an operator could mutate the wrong taxpayer's data without ever confirming "
"which profile is active (the cross-tenant wrong-active-profile leak)",
)
return ProfileConfirmationVerdict(
scenario=scenario.name,
confirmation_command_resolves=confirmation_command_resolves,
mutating_step_present=mutating_step_present,
confirmed_before_first_mutation=confirmed_before_first_mutation,
failures=tuple(failures),
)