"""Tests for the authorized ``cyberbrain run`` execution path.

These tests pin the fail-closed behaviour of the rewritten CLI:

* every option that grants anything is validated before a bridge exists, so a
  rejected invocation executes nothing at all;
* a code-carrying step ends closed at the scope gate when ``--scopes`` is empty;
* an approval is required before any code runs, and granting it on the shared
  SQLite store from a second "terminal" lets the very same step succeed;
* a step without an explicitly declared action, or one whose dependency did not
  succeed, is never dispatched;
* no command reports success for a run that failed, refused or blocked.

Output conventions under test
-----------------------------
Diagnostics that ``cyberbrain26.cli`` controls go to stderr through
``typer.echo(..., err=True)``, and no command re-raises a handled exception, so
a CLI failure never produces a Python traceback on stdout and never leaves an
unhandled exception behind ``result.exception``. The libraries below the CLI do
log their own diagnostics - including stack traces at ERROR level - on stderr
when a dispatched step is denied; that is deliberate operational logging in
``orchestration.ptah_bridge`` and is left in place, so the assertions below check
stdout for tracebacks and check ``result.exception`` for crashes.
"""

from __future__ import annotations

import json
import shutil
import sqlite3
import threading
import time
from pathlib import Path
from typing import Any

import pytest
from typer.testing import CliRunner

from cyberbrain26.cli import app
from cyberbrain26.product.local_store import open_database, resolve_db_path, workspaces_root
from cyberbrain26.security_plane.approval import ApprovalManager

runner = CliRunner()

requires_bubblewrap = pytest.mark.skipif(
    shutil.which("bwrap") is None, reason="Bubblewrap is unavailable"
)

_SUCCESS_MARKERS = ("completed", "created with", "authorized")


# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------
def _stdout(result: Any) -> str:
    """Return the captured stdout across click versions."""
    try:
        return result.stdout
    except (AttributeError, ValueError):  # pragma: no cover - old click
        return result.output


def _stderr(result: Any) -> str:
    """Return the captured stderr across click versions."""
    try:
        return result.stderr
    except (AttributeError, ValueError):  # pragma: no cover - old click
        return result.output


def _all_output(result: Any) -> str:
    """Return stdout and stderr together."""
    return _stdout(result) + _stderr(result)


def _assert_failed_without_cli_crash(result: Any) -> None:
    """Assert the command failed cleanly: non-zero exit, no crash, no stdout traceback."""
    assert result.exit_code != 0
    assert isinstance(result.exception, SystemExit), (
        f"expected a clean SystemExit, got {result.exception!r}"
    )
    assert "Traceback" not in _stdout(result)


def _assert_no_success_claim(result: Any) -> None:
    """Assert no success wording was printed anywhere for a failed run."""
    output = _all_output(result)
    for marker in _SUCCESS_MARKERS:
        assert marker not in output, f"success marker {marker!r} present in: {output!r}"


def _store(data_dir: Path, **roadmaps: dict[str, Any]) -> None:
    """Write ``roadmaps.json`` into ``data_dir``."""
    (data_dir / "roadmaps.json").write_text(json.dumps(roadmaps, indent=2))


def _code_step(**payload: Any) -> dict[str, Any]:
    """Return a stored coder step whose payload carries executable code."""
    return {
        "id": "s1",
        "role": "coder",
        "description": "execute the code step",
        "payload": {
            "action": "execute_python",
            "code": "print('side-effect')",
            "language": "python",
            **payload,
        },
        "depends_on": [],
        "status": "pending",
    }


def _audit_rows(db_path: Path) -> list[dict[str, Any]]:
    """Return every ``audit_events`` row as a plain dict."""
    connection = sqlite3.connect(str(db_path))
    connection.row_factory = sqlite3.Row
    try:
        return [dict(row) for row in connection.execute("SELECT * FROM audit_events")]
    finally:
        connection.close()


def _task_ids(db_path: Path) -> list[str]:
    """Return every durable task id stored in the ``tasks`` table."""
    connection = sqlite3.connect(str(db_path))
    try:
        return [str(row[0]) for row in connection.execute("SELECT task_id FROM tasks")]
    finally:
        connection.close()


def _table_count(db_path: Path, table: str) -> int:
    """Return the row count of ``table``."""
    connection = sqlite3.connect(str(db_path))
    try:
        return int(connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])
    finally:
        connection.close()


# ----------------------------------------------------------------------
# 1. CLI surface
# ----------------------------------------------------------------------
def test_top_level_help_lists_every_command() -> None:
    result = runner.invoke(app, ["--help"])
    assert result.exit_code == 0
    for command in ("plan", "run", "version", "approvals", "doctor"):
        assert command in _stdout(result)


def test_run_help_exposes_the_authorization_options() -> None:
    result = runner.invoke(app, ["run", "--help"])
    assert result.exit_code == 0
    output = _stdout(result)
    for option in ("--data-dir", "--db", "--approval-timeout", "--scopes"):
        assert option in output


def test_run_help_no_longer_exposes_model_options() -> None:
    result = runner.invoke(app, ["run", "--help"])
    assert result.exit_code == 0
    output = _stdout(result)
    assert "--model" not in output
    assert "--lang" not in output


# ----------------------------------------------------------------------
# 2. Unknown roadmap id
# ----------------------------------------------------------------------
def test_run_with_unknown_roadmap_id_fails_closed(tmp_path: Path) -> None:
    result = runner.invoke(app, ["run", "--id", "no-such-roadmap", "--data-dir", str(tmp_path)])

    _assert_failed_without_cli_crash(result)
    assert "no-such-roadmap" in _stderr(result)
    _assert_no_success_claim(result)


def test_run_with_unknown_roadmap_id_dispatches_no_step(tmp_path: Path) -> None:
    result = runner.invoke(
        app, ["run", "--id", "no-such-roadmap", "--data-dir", str(tmp_path), "--scopes", "python"]
    )

    assert result.exit_code != 0
    db_path = resolve_db_path(data_dir=tmp_path)
    assert _table_count(db_path, "tasks") == 0
    assert _table_count(db_path, "approval_requests") == 0


# ----------------------------------------------------------------------
# 3. Scope gate closes the action when --scopes is empty
# ----------------------------------------------------------------------
def test_code_step_without_scopes_is_denied_and_audited(tmp_path: Path) -> None:
    _store(
        tmp_path,
        gated={"id": "gated", "name": "gated", "status": "pending", "steps": [_code_step()]},
    )

    result = runner.invoke(app, ["run", "--id", "gated", "--data-dir", str(tmp_path)])

    _assert_failed_without_cli_crash(result)
    assert "coder scopes: (none)" in _stdout(result)
    assert "s1" in _stderr(result)
    _assert_no_success_claim(result)

    rows = _audit_rows(resolve_db_path(data_dir=tmp_path))
    denials = [row for row in rows if row["result"] == "deny" and row["action"] == "execute_python"]
    assert denials, f"no scope denial recorded in audit_events: {rows}"
    assert denials[0]["error"].startswith("Agent ptah-coder lacks required scope 'python'")

    scope_checks = [row for row in rows if row["result"] == "scope_check_failed"]
    assert scope_checks
    assert scope_checks[0]["error"] == "missing_scope:python"


def test_code_step_without_scopes_raises_no_approval_request(tmp_path: Path) -> None:
    _store(
        tmp_path,
        gated={"id": "gated", "name": "gated", "status": "pending", "steps": [_code_step()]},
    )

    result = runner.invoke(app, ["run", "--id", "gated", "--data-dir", str(tmp_path)])
    assert result.exit_code != 0
    # The scope gate fires before policy and approval, so nothing was ever
    # offered to a human.
    assert _table_count(resolve_db_path(data_dir=tmp_path), "approval_requests") == 0


# ----------------------------------------------------------------------
# 4. Approval gate, then a granted approval really lets the code run
# ----------------------------------------------------------------------
def test_code_step_without_approval_reports_the_approval_gate(tmp_path: Path) -> None:
    _store(
        tmp_path,
        gated={"id": "gated", "name": "gated", "status": "pending", "steps": [_code_step()]},
    )

    result = runner.invoke(
        app,
        [
            "run",
            "--id",
            "gated",
            "--data-dir",
            str(tmp_path),
            "--scopes",
            "python",
            "--approval-timeout",
            "0.2",
        ],
    )

    _assert_failed_without_cli_crash(result)
    assert "Approval" in _stderr(result)
    assert "execute_python" in _stderr(result)
    _assert_no_success_claim(result)

    db_path = resolve_db_path(data_dir=tmp_path)
    # The request was raised and then decided by the timeout; it did not stay
    # pending, which is what makes the two-terminal flow unusable by accident.
    assert _table_count(db_path, "approval_requests") > 0
    connection = sqlite3.connect(str(db_path))
    try:
        statuses = [row[0] for row in connection.execute("SELECT status FROM approval_requests")]
    finally:
        connection.close()
    assert "pending" not in statuses
    assert _table_count(db_path, "approval_records") == 0


@requires_bubblewrap
def test_code_step_runs_after_an_approval_is_granted_on_the_shared_store(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    # The sandbox binds the workspace read-write, so it has to live inside the
    # sanctioned workspaces root. --data-dir alone does not configure the
    # sandbox, so the data directory is pointed at the same place in the
    # environment the sandbox reads.
    monkeypatch.setenv("CYBERBRAIN_DATA_DIR", str(tmp_path))
    monkeypatch.delenv("CYBERBRAIN_WORKSPACES_ROOT", raising=False)
    workspace = workspaces_root(data_dir=tmp_path) / "workspace"
    workspace.mkdir(parents=True)
    marker = workspace / "ran.txt"
    _store(
        tmp_path,
        gated={
            "id": "gated",
            "name": "gated",
            "status": "pending",
            "steps": [
                _code_step(
                    code="open('/workspace/ran.txt', 'w').write('executed')",
                    workspace=str(workspace),
                )
            ],
        },
    )

    # First attempt: the approval is never granted, so nothing may run.
    blocked = runner.invoke(
        app,
        [
            "run",
            "--id",
            "gated",
            "--data-dir",
            str(tmp_path),
            "--scopes",
            "python",
            "--approval-timeout",
            "0.2",
        ],
    )
    assert blocked.exit_code != 0
    assert not marker.exists()

    # Second attempt in a background thread: this is the two-terminal flow. The
    # operator approves every pending request on the *same* database file while
    # the blocked execution waits, exactly like 'cyberbrain approvals approve'
    # in another terminal would.
    db_path = resolve_db_path(data_dir=tmp_path)
    outcomes: dict[str, Any] = {}

    def _run_in_background() -> None:
        outcomes["result"] = runner.invoke(
            app,
            [
                "run",
                "--id",
                "gated",
                "--data-dir",
                str(tmp_path),
                "--scopes",
                "python",
                "--approval-timeout",
                "30",
            ],
        )

    worker = threading.Thread(target=_run_in_background)
    approved: set[str] = set()
    worker.start()
    try:
        manager = ApprovalManager(database=open_database(db_path=db_path, persist=True))
        assert manager.database is not None
        deadline = time.monotonic() + 60.0
        while worker.is_alive() and time.monotonic() < deadline:
            for request in manager.list_pending():
                if request.request_id in approved:
                    continue
                decision = manager.approve(request.request_id, "test-reviewer", "granted by test")
                if decision is not None:
                    approved.add(request.request_id)
            time.sleep(0.02)
    finally:
        # Never leave the worker running: it would keep swapping the global
        # sys.stdout/sys.stderr that CliRunner installs and break later tests.
        worker.join(timeout=120)
    assert not worker.is_alive(), "the background run never finished"
    assert "result" in outcomes, "the background run never produced a result"

    result = outcomes["result"]
    assert approved, "no approval request was ever raised, so nothing could be approved"
    assert result.exit_code == 0, _all_output(result)
    assert "completed" in _stdout(result)
    assert "Traceback" not in _stdout(result)

    # The approval is only meaningful if the sandboxed code really ran.
    assert marker.exists()
    assert marker.read_text() == "executed"
    assert _table_count(db_path, "approval_records") == len(approved)


# ----------------------------------------------------------------------
# 5. A step without a declared action is refused
# ----------------------------------------------------------------------
def test_step_without_declared_action_is_refused(tmp_path: Path) -> None:
    _store(
        tmp_path,
        silent={
            "id": "silent",
            "name": "silent",
            "status": "pending",
            "steps": [
                {
                    "id": "no-action-step",
                    "role": "planner",
                    "description": "planner output carries no action",
                    "payload": {"goal": "do something"},
                    "depends_on": [],
                    "status": "pending",
                }
            ],
        },
    )

    result = runner.invoke(
        app, ["run", "--id", "silent", "--data-dir", str(tmp_path), "--scopes", "python"]
    )

    _assert_failed_without_cli_crash(result)
    assert "no-action-step" in _stderr(result)
    assert "declares no 'action'" in _stderr(result)
    _assert_no_success_claim(result)

    db_path = resolve_db_path(data_dir=tmp_path)
    assert _table_count(db_path, "tasks") == 0
    assert _table_count(db_path, "approval_requests") == 0


def test_planner_produced_roadmap_is_entirely_refused(tmp_path: Path) -> None:
    result = runner.invoke(
        app, ["plan", "--goal", "ship the thing", "--id", "planned", "--data-dir", str(tmp_path)]
    )
    assert result.exit_code == 0
    stored = json.loads((tmp_path / "roadmaps.json").read_text())
    assert stored["planned"]["steps"], "the planner produced no steps to test with"

    run_result = runner.invoke(
        app, ["run", "--id", "planned", "--data-dir", str(tmp_path), "--scopes", "python"]
    )
    _assert_failed_without_cli_crash(run_result)
    _assert_no_success_claim(run_result)
    # Not one of the stored steps was ever queued; the only durable task in the
    # store is 'plan''s own planner task, created by the plan command above.
    assert [task_id for task_id in _task_ids(resolve_db_path(data_dir=tmp_path)) if "step-1" in task_id] == []


# ----------------------------------------------------------------------
# 6. Unknown scope value
# ----------------------------------------------------------------------
def test_unknown_scope_is_rejected_and_nothing_executes(tmp_path: Path) -> None:
    _store(
        tmp_path,
        gated={"id": "gated", "name": "gated", "status": "pending", "steps": [_code_step()]},
    )

    result = runner.invoke(
        app,
        ["run", "--id", "gated", "--data-dir", str(tmp_path), "--scopes", "python,teleport"],
    )

    assert result.exit_code != 0
    output = _all_output(result)
    assert "teleport" in output
    assert "Valid scopes are:" in output
    assert "python" in output
    # The message is emitted before a bridge, a store or an agent exists.
    assert not resolve_db_path(data_dir=tmp_path).exists()


def test_unknown_scope_is_rejected_even_without_a_roadmap(tmp_path: Path) -> None:
    result = runner.invoke(
        app, ["run", "--id", "anything", "--data-dir", str(tmp_path), "--scopes", "nope"]
    )
    assert result.exit_code != 0
    assert "nope" in _all_output(result)
    assert not resolve_db_path(data_dir=tmp_path).exists()


# ----------------------------------------------------------------------
# 7. Non-positive approval timeout
# ----------------------------------------------------------------------
@pytest.mark.parametrize("timeout", ["0", "-1", "-0.5"])
def test_non_positive_approval_timeout_is_rejected(tmp_path: Path, timeout: str) -> None:
    _store(
        tmp_path,
        gated={"id": "gated", "name": "gated", "status": "pending", "steps": [_code_step()]},
    )

    result = runner.invoke(
        app,
        [
            "run",
            "--id",
            "gated",
            "--data-dir",
            str(tmp_path),
            "--scopes",
            "python",
            "--approval-timeout",
            timeout,
        ],
    )

    assert result.exit_code != 0
    output = _all_output(result)
    assert "--approval-timeout must be a positive number of seconds" in output
    _assert_no_success_claim(result)
    assert not resolve_db_path(data_dir=tmp_path).exists()


# ----------------------------------------------------------------------
# 8. CYBERBRAIN_DATA_DIR keeps roadmaps.json and SQLite together
# ----------------------------------------------------------------------
def test_env_data_dir_keeps_roadmap_index_and_sqlite_in_one_directory(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    data_dir = tmp_path / "env-data"
    monkeypatch.setenv("CYBERBRAIN_DATA_DIR", str(data_dir))
    monkeypatch.delenv("CYBERBRAIN_DB_PATH", raising=False)
    monkeypatch.delenv("CYBERBRAIN_RETENTION_DAYS", raising=False)

    result = runner.invoke(
        app,
        [
            "run",
            "--id",
            "from-env",
            "--data-dir",
            str(data_dir),
            "--scopes",
            "python",
        ],
    )
    assert result.exit_code != 0
    assert (data_dir / "cyberbrain.sqlite3").is_file()
    assert not (tmp_path / "cyberbrain.sqlite3").exists()


def test_env_data_dir_holds_plan_output_and_the_database(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    data_dir = tmp_path / "env-data"
    monkeypatch.setenv("CYBERBRAIN_DATA_DIR", str(data_dir))
    monkeypatch.delenv("CYBERBRAIN_DB_PATH", raising=False)
    monkeypatch.delenv("CYBERBRAIN_RETENTION_DAYS", raising=False)

    planned = runner.invoke(app, ["plan", "--goal", "wire it up", "--id", "env-plan"])
    assert planned.exit_code == 0, _all_output(planned)

    roadmaps_file = data_dir / "roadmaps.json"
    db_file = data_dir / "cyberbrain.sqlite3"
    assert roadmaps_file.is_file()
    assert db_file.is_file()
    assert roadmaps_file.parent == db_file.parent
    assert "env-plan" in json.loads(roadmaps_file.read_text())
    # No --data-dir was passed anywhere, so nothing leaked into the real home.
    assert not (tmp_path / "roadmaps.json").exists()


# ----------------------------------------------------------------------
# 9. Sub-command help still renders
# ----------------------------------------------------------------------
@pytest.mark.parametrize("command", ["approvals", "doctor", "plan"])
def test_subcommand_help_renders(command: str) -> None:
    result = runner.invoke(app, [command, "--help"])
    assert result.exit_code == 0, _all_output(result)
    assert "Usage" in _stdout(result)


# ----------------------------------------------------------------------
# Store robustness: a corrupt index must not become a traceback
# ----------------------------------------------------------------------
def test_corrupt_roadmap_index_fails_closed(tmp_path: Path) -> None:
    (tmp_path / "roadmaps.json").write_text("{not json")

    result = runner.invoke(app, ["run", "--id", "gated", "--data-dir", str(tmp_path)])

    _assert_failed_without_cli_crash(result)
    assert "roadmaps.json" in _stderr(result)
    _assert_no_success_claim(result)


def test_roadmap_with_duplicate_step_ids_fails_closed(tmp_path: Path) -> None:
    step = _code_step()
    _store(
        tmp_path,
        dup={
            "id": "dup",
            "name": "dup",
            "status": "pending",
            "steps": [dict(step), dict(step)],
        },
    )

    result = runner.invoke(app, ["run", "--id", "dup", "--data-dir", str(tmp_path)])

    _assert_failed_without_cli_crash(result)
    assert "malformed" in _stderr(result)
    assert _table_count(resolve_db_path(data_dir=tmp_path), "tasks") == 0
    _assert_no_success_claim(result)


# ----------------------------------------------------------------------
# depends_on is honoured
# ----------------------------------------------------------------------
def test_step_whose_dependency_failed_is_never_dispatched(tmp_path: Path) -> None:
    _store(
        tmp_path,
        chained={
            "id": "chained",
            "name": "chained",
            "status": "pending",
            "steps": [
                {
                    "id": "first",
                    "role": "planner",
                    "description": "declares no action",
                    "payload": {"goal": "g"},
                    "depends_on": [],
                    "status": "pending",
                },
                {
                    "id": "second",
                    "role": "coder",
                    "description": "depends on first",
                    "payload": _code_step()["payload"],
                    "depends_on": ["first"],
                    "status": "pending",
                },
            ],
        },
    )

    result = runner.invoke(
        app,
        ["run", "--id", "chained", "--data-dir", str(tmp_path), "--scopes", "python"],
    )

    _assert_failed_without_cli_crash(result)
    assert "second" in _stderr(result)
    assert "depends on first" in _stderr(result)
    assert "was not dispatched" in _stderr(result)
    _assert_no_success_claim(result)

    # Only the first step ever reached the queue; 'second' was blocked locally.
    db_path = resolve_db_path(data_dir=tmp_path)
    connection = sqlite3.connect(str(db_path))
    try:
        task_ids = [row[0] for row in connection.execute("SELECT task_id FROM tasks")]
    finally:
        connection.close()
    assert "ptah:chained:second" not in task_ids


def test_empty_roadmap_is_not_reported_as_completed(tmp_path: Path) -> None:
    _store(tmp_path, empty={"id": "empty", "name": "empty", "status": "pending", "steps": []})

    result = runner.invoke(app, ["run", "--id", "empty", "--data-dir", str(tmp_path)])

    _assert_failed_without_cli_crash(result)
    assert "contains no steps" in _stderr(result)
    _assert_no_success_claim(result)


# ----------------------------------------------------------------------
# 10. One step, one approval (regression)
# ----------------------------------------------------------------------
@requires_bubblewrap
def test_code_step_completes_after_exactly_one_approval(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A code step must need one human decision, not one per gate.

    Regression test. The operator approves the first request and then approves
    nothing at all - exactly what a person typing one ``approvals approve``
    does. The step must complete. While the run raised a *second* independent
    request, it instead sat blocked until the approval timed out and then failed
    with "Approval pending", so the documented single-approval flow was unusable
    in practice: it only ever finished for a script that approves in a loop.

    The earlier test above cannot catch that, because it approves every pending
    request until the worker exits and is therefore equally happy with one
    approval or with five.
    """
    monkeypatch.setenv("CYBERBRAIN_DATA_DIR", str(tmp_path))
    monkeypatch.delenv("CYBERBRAIN_WORKSPACES_ROOT", raising=False)
    workspace = workspaces_root(data_dir=tmp_path) / "workspace"
    workspace.mkdir(parents=True)
    marker = workspace / "ran.txt"
    _store(
        tmp_path,
        gated={
            "id": "gated",
            "name": "gated",
            "status": "pending",
            "steps": [
                _code_step(
                    code="open('/workspace/ran.txt', 'w').write('executed')",
                    workspace=str(workspace),
                )
            ],
        },
    )

    db_path = resolve_db_path(data_dir=tmp_path)
    # Open the store before the run starts, so the two threads are not racing to
    # create the same tables; the operator's second terminal is what this models.
    manager = ApprovalManager(database=open_database(db_path=db_path, persist=True))
    assert manager.database is not None
    outcomes: dict[str, Any] = {}

    def _run_in_background() -> None:
        outcomes["result"] = runner.invoke(
            app,
            [
                "run",
                "--id",
                "gated",
                "--data-dir",
                str(tmp_path),
                "--scopes",
                "python",
                "--approval-timeout",
                "10",
            ],
        )

    worker = threading.Thread(target=_run_in_background)
    approved: list[str] = []
    worker.start()
    try:
        deadline = time.monotonic() + 60.0
        # Answer the first request, and only the first.
        while not approved and time.monotonic() < deadline:
            for request in manager.list_pending():
                if manager.approve(request.request_id, "test-reviewer", "granted by test"):
                    approved.append(request.request_id)
                    break
            time.sleep(0.02)
        # From here on nothing is approved, so a second request would be fatal.
        while worker.is_alive() and time.monotonic() < deadline:
            time.sleep(0.05)
    finally:
        worker.join(timeout=120)
    assert not worker.is_alive(), "the background run never finished"
    assert "result" in outcomes, "the background run never produced a result"

    result = outcomes["result"]
    assert len(approved) == 1, f"expected exactly one approval to be granted, got {approved}"
    assert result.exit_code == 0, _all_output(result)
    assert "completed" in _stdout(result)
    assert "Traceback" not in _stdout(result)
    # The single approval is only meaningful if the sandboxed code really ran.
    assert marker.read_text() == "executed"

    # One request, one record: the step raised no second approval behind the
    # first, and the run recorded exactly one human decision.
    assert _table_count(db_path, "approval_requests") == 1
    assert _table_count(db_path, "approval_records") == 1

    # The audit trail makes the gating legible: one request was raised, a human
    # granted it, and the inner gate recorded a verified reuse of that grant
    # rather than asking again.
    rows = _audit_rows(db_path)
    step_events = [row for row in rows if row["task_id"] == "s1"]
    assert [row["result"] for row in step_events if row["event_type"] == "approval_required"] == [
        "pending"
    ], step_events
    granted = [
        row["result"] for row in step_events if row["event_type"] == "approval_granted"
    ]
    assert sorted(granted) == ["approved", "reused"], step_events