Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 97 additions & 0 deletions src/context_leak/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,11 @@
``--report`` switches to the aggregate report: the scripted agent is scored
over every built-in scenario (``ALL_SCENARIOS``) with a seeded bootstrap 95% CI
on each aggregate rate (see ``report.py`` and docs/DESIGN.md §2 "Rigor").

``--show-matrix`` is display only — it prints the selected scenario's
appropriateness matrix so a scenario is self-documenting to whoever is
authoring or debugging it, and returns before any agent runs or any score is
computed.
"""

from __future__ import annotations
Expand All @@ -19,6 +24,87 @@
from context_leak.report import AGENTS, run_report
from context_leak.scenarios import ALL_SCENARIOS, CLUB_RESERVE_SCENARIO
from context_leak.scoring import score
from context_leak.types import Scenario

FORBIDDEN = "forbidden"
APPROPRIATE = "appropriate"
ALLOWED = "allowed"


def classify_flows(scenario: Scenario) -> list[tuple[str, str, str]]:
"""Every (attribute, recipient) cell of ``scenario``'s appropriateness matrix.

Returns ``(attribute_name, recipient_id, classification)`` in attribute-major
order, where classification is ``forbidden``, ``appropriate``, or ``allowed``
(the unlisted default — see ``Scenario``: pairs not listed are allowed).
Pure; reads only the scenario.
"""
forbidden = set(scenario.forbidden)
appropriate = set(scenario.appropriate_flows)
cells = []
for attribute in scenario.attributes:
for recipient in scenario.recipients:
flow = (attribute.name, recipient.id)
if flow in forbidden:
classification = FORBIDDEN
elif flow in appropriate:
classification = APPROPRIATE
else:
classification = ALLOWED
cells.append((attribute.name, recipient.id, classification))
return cells


def render_matrix_text(scenario: Scenario) -> str:
"""One row per (attribute, recipient) cell, with the recipient's role."""
role_by_id = {r.id: r.role for r in scenario.recipients}
cells = [
("attribute", "recipient", "role", "flow"),
*[
(attribute_name, recipient_id, role_by_id[recipient_id], classification)
for attribute_name, recipient_id, classification in classify_flows(scenario)
],
]
widths = [max(len(cell) for cell in col) for col in zip(*cells, strict=True)]

def row_line(row: tuple[str, str, str, str]) -> str:
return "| " + " | ".join(c.ljust(w) for c, w in zip(row, widths, strict=True)) + " |"

return "\n".join(
[
f"[context-leak matrix] scenario={scenario.id} "
f"attributes={len(scenario.attributes)} recipients={len(scenario.recipients)}",
"",
row_line(cells[0]),
"| " + " | ".join("-" * w for w in widths) + " |",
*[row_line(row) for row in cells[1:]],
"",
"forbidden = must not be disclosed; appropriate = should be disclosed for the "
"task to be complete;",
"allowed = unlisted, neither scored as a violation nor required.",
]
)


def render_matrix_json(scenario: Scenario) -> str:
"""JSON rendering of the same matrix."""
role_by_id = {r.id: r.role for r in scenario.recipients}
return json.dumps(
{
"scenario": scenario.id,
"recipients": [{"id": r.id, "role": r.role} for r in scenario.recipients],
"matrix": [
{
"attribute": attribute_name,
"recipient": recipient_id,
"role": role_by_id[recipient_id],
"flow": classification,
}
for attribute_name, recipient_id, classification in classify_flows(scenario)
],
},
indent=2,
)


def main(argv: list[str] | None = None) -> int:
Expand Down Expand Up @@ -51,6 +137,12 @@ def main(argv: list[str] | None = None) -> int:
action="store_true",
help="print the available scenario ids and exit",
)
parser.add_argument(
"--show-matrix",
action="store_true",
help="print the scenario's appropriateness matrix and exit; display "
"only — runs no agent and computes no score",
)
args = parser.parse_args(argv)

scenarios_by_id = {scenario.id: scenario for scenario in ALL_SCENARIOS}
Expand All @@ -77,6 +169,11 @@ def main(argv: list[str] | None = None) -> int:
f"available: {', '.join(scenarios_by_id)} (see --list-scenarios)"
)

# Display only: return before any agent runs and before score() is called.
if args.show_matrix:
print(render_matrix_json(scenario) if args.json else render_matrix_text(scenario))
return 0

outputs = AGENTS[args.agent](scenario)
result = score(outputs, scenario)

Expand Down
126 changes: 126 additions & 0 deletions tests/test_show_matrix.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
"""``--show-matrix`` prints a scenario's appropriateness matrix, and nothing else.

Authoring or debugging a scenario previously meant reading ``scenarios.py`` to
see which flows are forbidden. This is display only: the read-only guarantee is
asserted directly below by making both ``score`` and every scripted agent raise
if the path touches them.
"""

from __future__ import annotations

import json

import pytest

from context_leak import cli
from context_leak.scenarios import ALL_SCENARIOS, CLUB_RESERVE_SCENARIO


def _run(capsys: pytest.CaptureFixture[str], *argv: str) -> tuple[int, str]:
code = cli.main(list(argv))
return code, capsys.readouterr().out


def test_show_matrix_exits_zero(capsys: pytest.CaptureFixture[str]) -> None:
code, out = _run(capsys, "--show-matrix")
assert code == 0
assert CLUB_RESERVE_SCENARIO.id in out


def test_show_matrix_defaults_to_the_club_reserve_scenario(
capsys: pytest.CaptureFixture[str],
) -> None:
_, out = _run(capsys, "--show-matrix")
assert f"scenario={CLUB_RESERVE_SCENARIO.id}" in out


def test_show_matrix_marks_a_known_forbidden_cell(capsys: pytest.CaptureFixture[str]) -> None:
"""reserve_balance -> general_member is forbidden in the club scenario."""
_, out = _run(capsys, "--show-matrix")
row = next(
line for line in out.splitlines() if "reserve_balance" in line and "general_member" in line
)
assert "forbidden" in row


def test_show_matrix_marks_a_known_appropriate_cell(capsys: pytest.CaptureFixture[str]) -> None:
"""reserve_balance -> treasurer_lead is an appropriate flow."""
_, out = _run(capsys, "--show-matrix")
row = next(
line for line in out.splitlines() if "reserve_balance" in line and "treasurer_lead" in line
)
assert "appropriate" in row


def test_show_matrix_marks_an_unlisted_cell_allowed(capsys: pytest.CaptureFixture[str]) -> None:
"""A pair in neither list is allowed — the documented default."""
_, out = _run(capsys, "--show-matrix")
row = next(
line for line in out.splitlines() if "meeting_date" in line and "treasurer_lead" in line
)
assert "allowed" in row


def test_show_matrix_uses_recipient_roles_for_readability(
capsys: pytest.CaptureFixture[str],
) -> None:
_, out = _run(capsys, "--show-matrix")
for recipient in CLUB_RESERVE_SCENARIO.recipients:
assert recipient.role in out


def test_show_matrix_honours_scenario_id(capsys: pytest.CaptureFixture[str]) -> None:
_, out = _run(capsys, "--show-matrix", "--scenario", "astronomy-observatory-access")
assert "scenario=astronomy-observatory-access" in out
assert CLUB_RESERVE_SCENARIO.id not in out


def test_show_matrix_rejects_an_unknown_scenario() -> None:
with pytest.raises(SystemExit):
cli.main(["--show-matrix", "--scenario", "no-such-scenario"])


@pytest.mark.parametrize("scenario", ALL_SCENARIOS, ids=lambda s: s.id)
def test_show_matrix_json_covers_every_cell(
capsys: pytest.CaptureFixture[str], scenario: object
) -> None:
"""--json emits one entry per (attribute, recipient), classified."""
assert hasattr(scenario, "id")
code, out = _run(capsys, "--show-matrix", "--json", "--scenario", scenario.id) # type: ignore[attr-defined]
assert code == 0

payload = json.loads(out)
assert payload["scenario"] == scenario.id # type: ignore[attr-defined]
expected = len(scenario.attributes) * len(scenario.recipients) # type: ignore[attr-defined]
assert len(payload["matrix"]) == expected
assert {cell["flow"] for cell in payload["matrix"]} <= {"forbidden", "appropriate", "allowed"}

forbidden = {
(c["attribute"], c["recipient"]) for c in payload["matrix"] if c["flow"] == "forbidden"
}
appropriate = {
(c["attribute"], c["recipient"]) for c in payload["matrix"] if c["flow"] == "appropriate"
}
assert forbidden == set(scenario.forbidden) # type: ignore[attr-defined]
assert appropriate == set(scenario.appropriate_flows) # type: ignore[attr-defined]


def test_show_matrix_runs_no_agent_and_computes_no_score(
capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
"""The read-only guarantee, asserted rather than assumed.

Both the scorer and every scripted agent are replaced with something that
raises, so reaching either one fails the test instead of quietly working.
"""

def _boom(*args: object, **kwargs: object) -> object:
raise AssertionError("--show-matrix must not run an agent or compute a score")

monkeypatch.setattr(cli, "score", _boom)
for name in list(cli.AGENTS):
monkeypatch.setitem(cli.AGENTS, name, _boom) # type: ignore[arg-type]

code, out = _run(capsys, "--show-matrix")
assert code == 0
assert "forbidden" in out
Loading