feat(safety): default to guarded review with TUI approvals

tests/test_cli_safety.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+import json
+import sys
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+from strix.config import loader
+from strix.interface import cli_args
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+@pytest.fixture(autouse=True)
+def isolated_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("STRIX_SAFETY_MODE", raising=False)
+    loader.apply_config_override(tmp_path / "config.json")
+
+
+def test_fresh_runs_default_to_guarded(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(sys, "argv", ["strix"])
+
+    args = cli_args.parse_arguments()
+
+    assert args.needs_setup is True
+    assert args.safety_mode == "guarded"
+
+
+def test_dangerous_flag_disables_safety(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(sys, "argv", ["strix", "--dangerously-disable-safety"])
+
+    args = cli_args.parse_arguments()
+
+    assert args.safety_mode == "off"
+
+
+def test_removed_mode_flag_has_actionable_error(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.setattr(sys, "argv", ["strix", "--safety-mode", "guarded"])
+
+    with pytest.raises(SystemExit):
+        cli_args.parse_arguments()
+
+    error = capsys.readouterr().err
+    assert "--safety-mode was removed" in error
+    assert "--dangerously-disable-safety" in error
+
+
+def test_removed_mode_environment_has_actionable_error(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.setenv("STRIX_SAFETY_MODE", "off")
+    monkeypatch.setattr(sys, "argv", ["strix"])
+
+    with pytest.raises(SystemExit):
+        cli_args.parse_arguments()
+
+    assert "STRIX_SAFETY_MODE was removed" in capsys.readouterr().err
+
+
+def _write_resumable_run(tmp_path: Path, safety_mode: str | None) -> None:
+    work = tmp_path / "project"
+    work.mkdir()
+    run_dir = tmp_path / "strix_runs" / "run-1"
+    state_dir = run_dir / ".state"
+    state_dir.mkdir(parents=True)
+    record: dict[str, Any] = {
+        "run_name": "run-1",
+        "targets_info": [],
+        "workspace_mount": str(work),
+        "local_sources": [],
+    }
+    if safety_mode is not None:
+        record["safety_mode"] = safety_mode
+    (run_dir / "run.json").write_text(json.dumps(record), encoding="utf-8")
+    (state_dir / "agents.json").write_text("{}", encoding="utf-8")
+
+
+@pytest.mark.parametrize("safety_mode", ["off", None])
+def test_off_resume_requires_dangerous_flag(
+    safety_mode: str | None,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    _write_resumable_run(tmp_path, safety_mode)
+    monkeypatch.setattr(sys, "argv", ["strix", "--resume", "run-1"])
+
+    with pytest.raises(SystemExit):
+        cli_args.parse_arguments()
+
+    assert "--dangerously-disable-safety again" in capsys.readouterr().err
+
+
+def test_off_resume_accepts_dangerous_flag(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.chdir(tmp_path)
+    _write_resumable_run(tmp_path, "off")
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["strix", "--resume", "run-1", "--dangerously-disable-safety"],
+    )
+
+    assert cli_args.parse_arguments().safety_mode == "off"
+
+
+def test_guarded_resume_rejects_dangerous_flag(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    _write_resumable_run(tmp_path, "guarded")
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["strix", "--resume", "run-1", "--dangerously-disable-safety"],
+    )
+
+    with pytest.raises(SystemExit):
+        cli_args.parse_arguments()
+
+    assert "cannot disable safety for a guarded run" in capsys.readouterr().err
+
+
+def test_observe_resume_is_rejected(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    _write_resumable_run(tmp_path, "observe")
+    monkeypatch.setattr(sys, "argv", ["strix", "--resume", "run-1"])
+
+    with pytest.raises(SystemExit):
+        cli_args.parse_arguments()
+
+    assert "observe mode was removed" in capsys.readouterr().err

tests/test_cli_target_list.py

@@ -113,6 +113,7 @@ def test_resume_restores_a_target_less_workspace_mount(
             "workspace_mount": str(work),
             "instruction": "audit the auth flow",
             "scan_mode": "deep",
+            "safety_mode": "guarded",
         },
     )
     monkeypatch.setattr(sys, "argv", ["strix", "--resume", "pentest_abcd"])

tests/test_config_loader.py

@@ -173,7 +173,6 @@ def test_apply_override_and_load_settings_round_trip(tmp_path: Path) -> None:
                 "env": {
                     "STRIX_LLM": "round-trip-model",
                     "PERPLEXITY_API_KEY": "pk",
-                    "STRIX_SAFETY_MODE": "guarded",
                     "STRIX_SAFETY_MODEL": "openai/safety-model",
                     "STRIX_SAFETY_TIMEOUT": "12",
                 }
@@ -187,13 +186,28 @@ def test_apply_override_and_load_settings_round_trip(tmp_path: Path) -> None:
 
     assert settings.llm.model == "round-trip-model"
     assert settings.integrations.perplexity_api_key == "pk"
-    assert settings.safety.mode == "guarded"
     assert settings.safety.model == "openai/safety-model"
     assert settings.safety.timeout == 12
     # Second call is memoized -> same object.
     assert loader.load_settings() is settings
 
 
+def test_removed_safety_mode_environment_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("STRIX_SAFETY_MODE", "observe")
+
+    with pytest.raises(ValueError, match="--dangerously-disable-safety"):
+        loader.load_settings()
+
+
+def test_removed_safety_mode_config_is_rejected(tmp_path: Path) -> None:
+    path = tmp_path / "cli-config.json"
+    path.write_text(json.dumps({"env": {"STRIX_SAFETY_MODE": "off"}}), encoding="utf-8")
+    loader.apply_config_override(path)
+
+    with pytest.raises(ValueError, match="STRIX_SAFETY_MODE was removed"):
+        loader.load_settings()
+
+
 def test_apply_config_override_invalidates_cache(tmp_path: Path) -> None:
     first = tmp_path / "first.json"
     first.write_text(json.dumps({"env": {"STRIX_LLM": "first-model"}}), encoding="utf-8")

tests/test_go_tui_runtime.py

@@ -18,6 +18,7 @@ import pytest
 from strix.config.settings import DEFAULT_MAX_TURNS
 from strix.interface.tui import runtime as go_tui
 from strix.interface.tui import sidecar
+from strix.interface.tui.backend.protocol import PROTOCOL_CAPABILITIES, PROTOCOL_VERSION
 from strix.interface.tui.runtime import GoTuiRuntime
 
 
@@ -244,16 +245,9 @@ async def test_runtime_does_not_initialize_or_scan_before_ready(
         await _send_message(
             child,
             {
-                "version": 3,
+                "version": PROTOCOL_VERSION,
                 "type": "ready",
-                "payload": {
-                    "capabilities": [
-                        "state-revisions",
-                        "collection-deltas",
-                        "structured-command-errors",
-                        "agents-collection",
-                    ]
-                },
+                "payload": {"capabilities": list(PROTOCOL_CAPABILITIES)},
             },
         )
         await asyncio.wait_for(run_task, timeout=2)
@@ -780,6 +774,7 @@ async def test_scan_passes_max_turns_and_budget(monkeypatch: pytest.MonkeyPatch)
 
     assert captured["max_turns"] == 37
     assert captured["max_budget_usd"] == 4.25
+    assert captured["safety_approval_callback"] == runtime.controller.safety_approval_callback
     assert runtime.controller.scan_state == "stopped"
 
 

tests/test_runner_rate_limit.py

@@ -79,7 +79,7 @@ async def test_persistent_rate_limit_stops_gracefully(
 
     with caplog.at_level(logging.WARNING):
         result = await runner.run_strix_scan(
-            scan_config={"targets": [], "scan_mode": "deep"},
+            scan_config={"targets": [], "scan_mode": "deep", "safety_mode": "off"},
             scan_id="scan-test",
             image="img",
             coordinator=coordinator,

tests/test_runner_root_prompt.py

@@ -16,6 +16,7 @@ from openai import RateLimitError
 
 import strix.tools.notes.tools as notes_tools
 import strix.tools.todo.tools as todo_tools
+from strix.config.settings import SafetySettings
 from strix.core import runner
 from strix.core.agents import AgentCoordinator
 from strix.runtime import session_manager
@@ -52,6 +53,7 @@ def _patch_engine_scaffold(
             extra_headers=None,
         ),
         runtime=types.SimpleNamespace(max_context_images=3),
+        safety=SafetySettings(),
     )
     monkeypatch.setattr(runner, "load_settings", lambda: settings)
     monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
@@ -177,7 +179,34 @@ async def test_root_prompt_options_default_to_none(
 
     kwargs = captured["kwargs"]
     assert kwargs["instructions_override"] is None
-    assert kwargs["system_prompt_context"] == {"scope": "built-in"}
+    assert kwargs["system_prompt_context"] == {
+        "scope": "built-in",
+        "safety_mode": "guarded",
+        "workspace_isolation": True,
+        "human_approval_available": False,
+    }
+
+
+@pytest.mark.asyncio
+async def test_root_prompt_only_advertises_human_approval_when_callback_is_installed(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Any,
+) -> None:
+    captured = _patch_engine_scaffold(monkeypatch, tmp_path, {})
+
+    async def approval(_request: object) -> bool:
+        return False
+
+    await runner.run_strix_scan(
+        scan_config={"targets": [], "scan_mode": "deep"},
+        scan_id="scan-approval",
+        image="img",
+        coordinator=AgentCoordinator(),
+        interactive=True,
+        safety_approval_callback=approval,
+    )
+
+    assert captured["kwargs"]["system_prompt_context"]["human_approval_available"] is True
 
 
 @pytest.mark.asyncio

tests/test_runner_safety.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+
+from strix.core.runner import _safety_mode, _validate_resume_safety_mode
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _record(run_dir: Path, mode: str | None) -> None:
+    run_dir.mkdir(exist_ok=True)
+    data = {} if mode is None else {"safety_mode": mode}
+    (run_dir / "run.json").write_text(json.dumps(data), encoding="utf-8")
+
+
+def test_programmatic_runs_default_to_guarded() -> None:
+    assert _safety_mode({}) == "guarded"
+
+
+@pytest.mark.parametrize("mode", ["guarded", "off"])
+def test_resume_accepts_unchanged_safety_mode(tmp_path: Path, mode: str) -> None:
+    _record(tmp_path, mode)
+
+    _validate_resume_safety_mode(tmp_path, mode)  # type: ignore[arg-type]
+
+
+def test_legacy_resume_defaults_to_off(tmp_path: Path) -> None:
+    _record(tmp_path, None)
+
+    _validate_resume_safety_mode(tmp_path, "off")
+
+
+@pytest.mark.parametrize(
+    ("persisted", "requested"),
+    [("guarded", "off"), ("off", "guarded"), (None, "guarded")],
+)
+def test_resume_rejects_safety_mode_changes(
+    tmp_path: Path,
+    persisted: str | None,
+    requested: str,
+) -> None:
+    _record(tmp_path, persisted)
+
+    with pytest.raises(ValueError, match="Cannot change safety mode"):
+        _validate_resume_safety_mode(tmp_path, requested)  # type: ignore[arg-type]
+
+
+def test_resume_rejects_removed_observe_mode(tmp_path: Path) -> None:
+    _record(tmp_path, "observe")
+
+    with pytest.raises(ValueError, match="observe mode was removed"):
+        _validate_resume_safety_mode(tmp_path, "guarded")
+
+
+@pytest.mark.parametrize("malformed", [None, "", False, 0])
+def test_resume_rejects_present_malformed_safety_mode(
+    tmp_path: Path,
+    malformed: object,
+) -> None:
+    (tmp_path / "run.json").write_text(
+        json.dumps({"safety_mode": malformed}),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="invalid safety mode"):
+        _validate_resume_safety_mode(tmp_path, "off")

tests/test_safety_evidence.py

@@ -856,9 +856,8 @@ def test_redirect_input_files_are_parsed_not_heredocs() -> None:
 
 
 @pytest.mark.asyncio
-async def test_workspace_input_file_is_attached_for_scope_review() -> None:
-    """A host list read via `< file` is evidence the reviewer needs to judge scope, so
-    its contents ride in the packet instead of leaving the reviewer to block blind."""
+async def test_workspace_input_file_is_attached_for_action_review() -> None:
+    """A host list read via `< file` is frozen with the action evidence."""
     bundle = await _compile(
         'while read -r host; do dig +short "$host"; done < hosts.txt > out.txt',
         {"/workspace/hosts.txt": "admin.fiuu.com\napi.fiuu.com\n"},
@@ -869,7 +868,8 @@ async def test_workspace_input_file_is_attached_for_scope_review() -> None:
         assert [a["path"] for a in inputs] == ["/workspace/hosts.txt"]
         assert "admin.fiuu.com" in inputs[0]["source"]
         assert inputs[0]["truncated"] is False
-        # Attaching contents is not itself a block; the reviewer judges scope.
+        assert bundle.workspace_evidence is True
+        # Attaching contents is not itself a block; the reviewer judges effects.
         assert bundle.deterministic_block is None
     finally:
         bundle.cleanup()
@@ -938,7 +938,7 @@ def test_list_flag_files_are_parsed(command: str, expected: list[str]) -> None:
 
 
 @pytest.mark.asyncio
-async def test_wordlist_flag_file_is_attached_for_scope_review() -> None:
+async def test_wordlist_flag_file_is_attached_for_action_review() -> None:
     """Recon tools route their target list through `-w`/`-l`, not a `<` redirect, so the
     same evidence must be collected for the reviewer to check it against scope."""
     bundle = await _compile(
@@ -950,6 +950,7 @@ async def test_wordlist_flag_file_is_attached_for_scope_review() -> None:
         inputs = [a for a in bundle.packet["artifacts"] if a.get("role") == "input"]
         assert [a["path"] for a in inputs] == ["/workspace/paths.txt"]
         assert "admin" in inputs[0]["source"]
+        assert bundle.workspace_evidence is True
     finally:
         bundle.cleanup()
 

tests/test_safety_prompt.py

@@ -26,9 +26,8 @@ def test_safety_guidance_is_absent_without_a_safety_mode(
 
 
 @pytest.mark.parametrize("phrase", _SAFETY_ONLY_PHRASES)
-@pytest.mark.parametrize("mode", ["guarded", "observe"])
-def test_safety_guidance_is_present_in_a_safety_mode(phrase: str, mode: str) -> None:
-    assert phrase in render_system_prompt(system_prompt_context={"safety_mode": mode})
+def test_safety_guidance_is_present_in_guarded_mode(phrase: str) -> None:
+    assert phrase in render_system_prompt(system_prompt_context={"safety_mode": "guarded"})
 
 
 def test_browser_skill_carries_no_safety_prohibitions() -> None:
@@ -40,8 +39,46 @@ def test_browser_skill_carries_no_safety_prohibitions() -> None:
         assert phrase not in prompt
 
 
-def test_observe_mode_states_its_passive_only_contract() -> None:
-    prompt = render_system_prompt(system_prompt_context={"safety_mode": "observe"})
+def test_guarded_interactive_prompt_explains_human_deferral() -> None:
+    prompt = render_system_prompt(
+        interactive=True,
+        system_prompt_context={
+            "safety_mode": "guarded",
+            "human_approval_available": True,
+        },
+    )
 
-    assert "passive target interaction only" in prompt
-    assert "Guarded mode permits" not in prompt
+    assert "user approves or denies that action" in prompt
+    assert "only the guarded action-safety reviewer may pause" in prompt
+
+
+def test_guarded_autonomous_prompt_has_no_human_channel() -> None:
+    prompt = render_system_prompt(system_prompt_context={"safety_mode": "guarded"})
+
+    assert "No human approval channel exists" in prompt
+    assert "NEVER wait for approval or authorization" in prompt
+
+
+def test_interactive_without_approval_callback_still_fails_closed() -> None:
+    prompt = render_system_prompt(
+        interactive=True,
+        system_prompt_context={"safety_mode": "guarded"},
+    )
+
+    assert "No human approval channel exists" in prompt
+    assert "user approves or denies that action" not in prompt
+
+
+def test_scope_allows_passive_external_research_without_expanding_targets() -> None:
+    prompt = render_system_prompt(
+        system_prompt_context={
+            "authorized_targets": [{"type": "web", "value": "https://example.test"}],
+            "scope_source": "scan",
+            "authorization_source": "user",
+        }
+    )
+
+    assert "certificate transparency services such as crt.sh" in prompt
+    assert "does not make that service a testing target" in prompt
+    assert "authorized domain includes its subdomains" in prompt
+    assert "NEVER actively scan, fuzz, authenticate to, exploit, or mutate" in prompt

tests/test_safety_reviewer.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import json
 from pathlib import Path
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 import pytest
 from agents.tool_context import ToolContext
@@ -234,11 +234,79 @@ async def test_low_confidence_allow_is_refused(tmp_path: Path, monkeypatch: Monk
     )
 
     assert decision.allowed is False
+    assert decision.deferred is False
     assert decision.source == "reviewer"
-    assert "below the 0.75 allow threshold" in decision.reason
+    assert "below the 0.75 threshold" in decision.reason
     assert decision.categories == ("target_mutation",)
 
 
+@pytest.mark.parametrize("model_decision", ["allow", "block"])
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_patched_sdk")
+async def test_interactive_low_confidence_verdict_defers(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    model_decision: Literal["allow", "block"],
+) -> None:
+    monkeypatch.setattr(
+        reviewer_module.Runner,
+        "run",
+        _verdict_run(
+            SafetyVerdict(
+                decision=model_decision,
+                risk="medium",
+                categories=["ambiguous_effect"],
+                reason="effect is unclear",
+                confidence=0.5,
+            )
+        ),
+    )
+
+    decision = await SafetyReviewer(inspection_runner=_InspectionRunner()).review(
+        _bundle(tmp_path, f"case-low-{model_decision}"),
+        human_approval_available=True,
+    )
+
+    assert decision.allowed is False
+    assert decision.deferred is True
+    assert decision.risk == "medium"
+    assert "below the 0.75 threshold" in decision.reason
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_patched_sdk")
+async def test_explicit_defer_requires_an_approval_channel(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        reviewer_module.Runner,
+        "run",
+        _verdict_run(
+            SafetyVerdict(
+                decision="defer",
+                risk="high",
+                categories=["ambiguous_effect"],
+                reason="persistence depends on endpoint behavior",
+                confidence=0.9,
+            )
+        ),
+    )
+    reviewer = SafetyReviewer(inspection_runner=_InspectionRunner())
+
+    interactive = await reviewer.review(
+        _bundle(tmp_path, "case-explicit-interactive"),
+        human_approval_available=True,
+    )
+    noninteractive = await reviewer.review(_bundle(tmp_path, "case-explicit-headless"))
+
+    assert interactive.deferred is True
+    assert interactive.risk == "high"
+    assert noninteractive.allowed is False
+    assert noninteractive.deferred is False
+    assert "no human approval channel" in noninteractive.reason
+
+
 @pytest.mark.asyncio
 @pytest.mark.usefixtures("_patched_sdk")
 async def test_confident_allow_passes(tmp_path: Path, monkeypatch: MonkeyPatch) -> None:
@@ -289,6 +357,7 @@ async def test_block_verdict_is_returned_as_a_block(
     )
 
     assert decision.allowed is False
+    assert decision.deferred is False
     assert decision.source == "reviewer"
     assert decision.reason == "deletes a record"
 
@@ -318,9 +387,11 @@ async def test_missing_model_configuration_blocks(
 
 @pytest.mark.asyncio
 @pytest.mark.usefixtures("_patched_sdk")
-async def test_allow_after_a_failed_inspection_is_refused(
+@pytest.mark.parametrize("model_decision", ["allow", "defer"])
+async def test_non_block_after_a_failed_inspection_is_refused(
     tmp_path: Path,
     monkeypatch: MonkeyPatch,
+    model_decision: Literal["allow", "defer"],
 ) -> None:
     """The reviewer decides from the inspection's own output, so an inspection that failed
     must not be able to underwrite an allow."""
@@ -329,7 +400,7 @@ async def test_allow_after_a_failed_inspection_is_refused(
         context.incomplete = True
         return _Result(
             SafetyVerdict(
-                decision="allow",
+                decision=model_decision,
                 risk="low",
                 categories=[],
                 reason="looked fine",
@@ -340,10 +411,12 @@ async def test_allow_after_a_failed_inspection_is_refused(
     monkeypatch.setattr(reviewer_module.Runner, "run", fake_run)
 
     decision = await SafetyReviewer(inspection_runner=_InspectionRunner()).review(
-        _bundle(tmp_path, "case-bad-inspection")
+        _bundle(tmp_path, "case-bad-inspection"),
+        human_approval_available=True,
     )
 
     assert decision.allowed is False
+    assert decision.deferred is False
     assert decision.categories == ("inspection_incomplete",)
 
 
@@ -383,26 +456,28 @@ def test_prompt_judges_security_testing_by_effect_not_technique() -> None:
     """Pins the effect-based guardrails so a future edit cannot silently revert to
     blocking in-scope offensive testing on the technique alone."""
     prompt = reviewer_module._SAFETY_PROMPT
+    normalized = " ".join(prompt.split())
 
     # Authorization framing and the effect-not-technique rule.
     assert "authorized penetration test" in prompt
-    assert (
-        'That an action is a "SQL\ninjection"' in prompt
-        or "not, by itself, a reason to block" in prompt
-    )
+    assert "not, by itself, a reason to block" in normalized
     # Read probes pass; writes and destruction block.
     assert "OR 1=1" in prompt
     for keyword in ("DROP", "DELETE", "INSERT", "TRUNCATE", "OUTFILE", "xp_cmdshell"):
         assert keyword in prompt
-    # Fail-closed on ambiguity is preserved.
-    assert "does not settle whether the effect is destructive" in prompt
+    # Scope enforcement belongs elsewhere, including for passive third-party services.
+    assert "Do not enforce authorization scope" in prompt
+    assert "crt.sh" in prompt
+    assert "solely because their hostname is outside" in prompt
+    # Ambiguity only reaches a human when an approval channel exists.
+    assert "Return defer only when approval is available" in prompt
+    assert "Without human approval, ambiguity must block" in prompt
     # Non-negotiable guardrails survive.
     assert 'Never allow when completeness.status is not "complete"' in prompt
     assert "Deterministic policy blocks cannot be overridden" in prompt
-    assert "analysis.mutating_request is\nnever passive" in prompt
+    assert "analysis.mutating_request" in prompt
 
 
-def test_prompt_scopes_subdomains_and_input_files() -> None:
-    prompt = reviewer_module._SAFETY_PROMPT
-    assert "authorized domain covers its subdomains" in prompt
-    assert 'role "input"' in prompt or 'role "input"' in prompt
+def test_prompt_explains_input_files() -> None:
+    prompt = " ".join(reviewer_module._SAFETY_PROMPT.split())
+    assert 'role "input"' in prompt

tests/test_safety_runtime.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import io
 import json
 from types import SimpleNamespace
@@ -12,7 +13,7 @@ import pytest
 
 from strix.config.settings import SafetySettings
 from strix.safety.runtime import SafetyRuntime
-from strix.safety.types import SafetyDecision
+from strix.safety.types import SafetyApprovalCallback, SafetyApprovalRequest, SafetyDecision
 
 
 if TYPE_CHECKING:
@@ -42,14 +43,36 @@ class _InspectionRunner:
 class _StubReviewer:
     """Stands in for the model review so a decision's source can be asserted."""
 
-    def __init__(self, on_review: Any = None) -> None:
+    def __init__(
+        self,
+        on_review: Any = None,
+        decision: SafetyDecision | None = None,
+    ) -> None:
         self.on_review = on_review
+        self.decision = decision
         self.calls = 0
+        self.human_approval_available: list[bool] = []
 
-    async def review(self, bundle: Any) -> SafetyDecision:
+    async def review(
+        self,
+        bundle: Any,
+        *,
+        human_approval_available: bool = False,
+    ) -> SafetyDecision:
         self.calls += 1
+        self.human_approval_available.append(human_approval_available)
         if self.on_review is not None:
             await self.on_review()
+        if self.decision is not None:
+            return SafetyDecision(
+                allowed=self.decision.allowed,
+                source=self.decision.source,
+                reason=self.decision.reason,
+                categories=self.decision.categories,
+                case_id=bundle.case_id,
+                risk=self.decision.risk,
+                deferred=self.decision.deferred,
+            )
         return SafetyDecision(
             allowed=True,
             source="reviewer",
@@ -58,7 +81,11 @@ class _StubReviewer:
         )
 
 
-def _runtime(tmp_path: Path, mode: str) -> SafetyRuntime:
+def _runtime(
+    tmp_path: Path,
+    mode: str,
+    approval_callback: SafetyApprovalCallback | None = None,
+) -> SafetyRuntime:
     return SafetyRuntime(
         scan_id="scan-1",
         mode=mode,  # type: ignore[arg-type]
@@ -68,12 +95,32 @@ def _runtime(tmp_path: Path, mode: str) -> SafetyRuntime:
         run_dir=tmp_path,
         sandbox_image="image",
         inspection_runner=_InspectionRunner(),
+        approval_callback=approval_callback,
     )
 
 
-def _ctx(*, agent_id: str = "agent-1", turn_input: list[dict[str, Any]] | None = None) -> Any:
+def _deferred() -> SafetyDecision:
+    return SafetyDecision(
+        allowed=False,
+        source="reviewer",
+        reason="the endpoint effect is ambiguous",
+        categories=("ambiguous_effect",),
+        risk="medium",
+        deferred=True,
+    )
+
+
+def _ctx(
+    *,
+    agent_id: str = "agent-1",
+    turn_input: list[dict[str, Any]] | None = None,
+    coordinator: Any = None,
+) -> Any:
+    context = {"agent_id": agent_id, "sandbox_session": object()}
+    if coordinator is not None:
+        context["coordinator"] = coordinator
     return SimpleNamespace(
-        context={"agent_id": agent_id, "sandbox_session": object()},
+        context=context,
         tool_call_id="call-1",
         turn_input=turn_input or [],
     )
@@ -121,20 +168,15 @@ async def test_browser_sessions_are_disjoint_per_agent(tmp_path: Path) -> None:
 
 
 @pytest.mark.asyncio
-async def test_observe_mode_blocks_a_browser_click_against_a_fresh_snapshot(
+async def test_active_browser_action_reaches_the_reviewer(
     tmp_path: Path,
 ) -> None:
-    """Without the snapshot the click blocks as incomplete evidence in every mode, so the
-    observe-mode rule itself would never be exercised."""
-    runtime = _runtime(tmp_path, "observe")
+    runtime = _runtime(tmp_path, "guarded")
     reviewer = _StubReviewer()
     runtime._reviewer = reviewer
-    invoked = False
 
     async def invoke(_ctx: Any, _raw_input: str) -> str:
-        nonlocal invoked
-        invoked = True
-        return "bad"
+        return "clicked"
 
     result = await runtime.invoke_exec(
         ctx=_ctx(turn_input=_SNAPSHOT_HISTORY),
@@ -142,27 +184,26 @@ async def test_observe_mode_blocks_a_browser_click_against_a_fresh_snapshot(
         invoke_tool=invoke,
     )
 
-    payload = json.loads(result)
-    assert payload["status"] == "blocked"
-    assert payload["safety"]["source"] == "deterministic"
-    assert payload["safety"]["categories"] == ["target_mutation"]
-    assert "not passive" in payload["safety"]["reason"]
-    assert reviewer.calls == 0
-    assert invoked is False
+    assert result == "clicked"
+    assert reviewer.calls == 1
 
 
 @pytest.mark.asyncio
-async def test_observe_mode_allows_a_passive_browser_read(tmp_path: Path) -> None:
+async def test_passive_browser_read_keeps_the_fast_path(tmp_path: Path) -> None:
     async def invoke(_ctx: Any, _raw_input: str) -> str:
         return "snapshot"
 
-    result = await _runtime(tmp_path, "observe").invoke_exec(
+    runtime = _runtime(tmp_path, "guarded")
+    reviewer = _StubReviewer()
+    runtime._reviewer = reviewer
+    result = await runtime.invoke_exec(
         ctx=_ctx(),
         arguments={"cmd": "agent-browser snapshot -i"},
         invoke_tool=invoke,
     )
 
     assert result == "snapshot"
+    assert reviewer.calls == 0
 
 
 @pytest.mark.asyncio
@@ -247,12 +288,154 @@ async def test_write_stdin_is_untouched_when_safety_is_off(tmp_path: Path) -> No
 
 
 @pytest.mark.asyncio
-async def test_observe_mode_blocks_a_mutating_request_without_the_reviewer(
+async def test_mutating_request_is_evidence_for_the_reviewer(
     tmp_path: Path,
 ) -> None:
-    runtime = _runtime(tmp_path, "observe")
+    runtime = _runtime(tmp_path, "guarded")
     reviewer = _StubReviewer()
     runtime._reviewer = reviewer
+
+    async def invoke(_ctx: Any, _raw_input: str) -> str:
+        return "reviewed"
+
+    result = await runtime.invoke_exec(
+        ctx=_ctx(),
+        arguments={"cmd": "curl -X DELETE https://example.test/v1/users/1042"},
+        invoke_tool=invoke,
+    )
+
+    assert result == "reviewed"
+    assert reviewer.calls == 1
+
+
+@pytest.mark.asyncio
+async def test_deferred_action_waits_for_human_and_executes_the_original_call(
+    tmp_path: Path,
+) -> None:
+    requests: list[SafetyApprovalRequest] = []
+    approval_started = asyncio.Event()
+    release_approval = asyncio.Event()
+
+    async def approve(request: SafetyApprovalRequest) -> bool:
+        requests.append(request)
+        approval_started.set()
+        await release_approval.wait()
+        return True
+
+    runtime = _runtime(tmp_path, "guarded", approve)
+    reviewer = _StubReviewer(decision=_deferred())
+    runtime._reviewer = reviewer
+    arguments = {"cmd": "nmap -sV example.test", "workdir": "/workspace"}
+    original_arguments = dict(arguments)
+    invoked: list[dict[str, Any]] = []
+
+    async def invoke(_ctx: Any, raw_input: str) -> str:
+        invoked.append(json.loads(raw_input))
+        return "scanned"
+
+    pending = asyncio.create_task(
+        runtime.invoke_exec(ctx=_ctx(), arguments=arguments, invoke_tool=invoke)
+    )
+    await approval_started.wait()
+
+    assert pending.done() is False
+    arguments["cmd"] = "rm -rf /workspace"
+    release_approval.set()
+
+    assert await pending == "scanned"
+    assert invoked == [original_arguments]
+    assert reviewer.human_approval_available == [True]
+    assert len(requests) == 1
+    request = requests[0]
+    assert request.agent_id == "agent-1"
+    assert request.request_id == request.case_id
+    assert request.tool_call_id == "call-1"
+    assert request.tool_name == "exec_command"
+    assert request.action == '{"cmd":"nmap -sV example.test","workdir":"/workspace"}'
+    assert request.digest == hashlib.sha256(request.action.encode()).hexdigest()
+    assert request.reason == "the endpoint effect is ambiguous"
+    assert request.categories == ("ambiguous_effect",)
+    assert request.risk == "medium"
+
+    audit_path = tmp_path / ".state" / "safety-audit.jsonl"
+    entries = [json.loads(line) for line in audit_path.read_text().splitlines()]
+    assert any(
+        entry["decision_source"] == "reviewer"
+        and entry["summary"].get("approval", {}).get("status") == "requested"
+        for entry in entries
+    )
+    assert any(
+        entry["decision_source"] == "human"
+        and entry["summary"].get("approval", {}).get("status") == "approved"
+        for entry in entries
+    )
+
+
+@pytest.mark.asyncio
+async def test_approval_cannot_execute_after_requesting_agent_stops(tmp_path: Path) -> None:
+    class Coordinator:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def graph_snapshot(self) -> tuple[dict[str, str], dict[str, str], dict, dict]:
+            self.calls += 1
+            status = "running" if self.calls == 1 else "stopped"
+            return {}, {"agent-1": status}, {}, {}
+
+    async def approve(_request: SafetyApprovalRequest) -> bool:
+        return True
+
+    runtime = _runtime(tmp_path, "guarded", approve)
+    runtime._reviewer = _StubReviewer(decision=_deferred())
+    invoked = False
+
+    async def invoke(_ctx: Any, _raw_input: str) -> str:
+        nonlocal invoked
+        invoked = True
+        return "bad"
+
+    result = await runtime.invoke_exec(
+        ctx=_ctx(coordinator=Coordinator()),
+        arguments={"cmd": "nmap example.test"},
+        invoke_tool=invoke,
+    )
+
+    payload = json.loads(result)
+    assert payload["status"] == "blocked"
+    assert payload["safety"]["source"] == "system"
+    assert payload["safety"]["categories"][-1] == "agent_inactive"
+    assert invoked is False
+
+
+@pytest.mark.asyncio
+async def test_oversized_action_cannot_be_deferred_to_human(tmp_path: Path) -> None:
+    approval_called = False
+
+    async def approve(_request: SafetyApprovalRequest) -> bool:
+        nonlocal approval_called
+        approval_called = True
+        return True
+
+    runtime = _runtime(tmp_path, "guarded", approve)
+    runtime._reviewer = _StubReviewer(decision=_deferred())
+    result = await runtime.invoke_exec(
+        ctx=_ctx(),
+        arguments={"cmd": "nmap " + "a" * 600},
+        invoke_tool=_noop_invoke,
+    )
+
+    payload = json.loads(result)
+    assert payload["safety"]["categories"] == ["approval_action_too_large"]
+    assert approval_called is False
+
+
+@pytest.mark.asyncio
+async def test_human_denial_returns_a_human_sourced_block(tmp_path: Path) -> None:
+    async def deny(_request: SafetyApprovalRequest) -> bool:
+        return False
+
+    runtime = _runtime(tmp_path, "guarded", deny)
+    runtime._reviewer = _StubReviewer(decision=_deferred())
     invoked = False
 
     async def invoke(_ctx: Any, _raw_input: str) -> str:
@@ -262,18 +445,126 @@ async def test_observe_mode_blocks_a_mutating_request_without_the_reviewer(
 
     result = await runtime.invoke_exec(
         ctx=_ctx(),
-        arguments={"cmd": "curl -X DELETE https://example.test/v1/users/1042"},
+        arguments={"cmd": "nmap -sV example.test"},
         invoke_tool=invoke,
     )
 
     payload = json.loads(result)
     assert payload["status"] == "blocked"
-    assert payload["safety"]["source"] == "deterministic"
-    assert "DELETE" in payload["safety"]["reason"]
-    assert reviewer.calls == 0
+    assert payload["safety"]["source"] == "human"
+    assert payload["safety"]["risk"] == "medium"
+    assert "Human denied" in payload["safety"]["reason"]
     assert invoked is False
 
 
+@pytest.mark.asyncio
+async def test_lifecycle_cancellation_is_not_recorded_as_human_denial(tmp_path: Path) -> None:
+    async def cancel(_request: SafetyApprovalRequest) -> str:
+        return "cancelled"
+
+    runtime = _runtime(tmp_path, "guarded", cancel)  # type: ignore[arg-type]
+    runtime._reviewer = _StubReviewer(decision=_deferred())
+
+    result = await runtime.invoke_exec(
+        ctx=_ctx(),
+        arguments={"cmd": "nmap example.test"},
+        invoke_tool=_noop_invoke,
+    )
+
+    payload = json.loads(result)
+    assert payload["safety"]["source"] == "system"
+    assert payload["safety"]["categories"][-1] == "approval_cancelled"
+    assert "cancelled" in payload["safety"]["reason"]
+
+
+@pytest.mark.asyncio
+async def test_defer_without_an_approval_channel_blocks(tmp_path: Path) -> None:
+    runtime = _runtime(tmp_path, "guarded")
+    runtime._reviewer = _StubReviewer(decision=_deferred())
+
+    result = await runtime.invoke_exec(
+        ctx=_ctx(),
+        arguments={"cmd": "nmap -sV example.test"},
+        invoke_tool=_noop_invoke,
+    )
+
+    payload = json.loads(result)
+    assert payload["status"] == "blocked"
+    assert payload["safety"]["source"] == "reviewer"
+    assert "no human approval channel" in payload["safety"]["reason"]
+
+
+@pytest.mark.parametrize("command", ["rm -rf /workspace", ""])
+@pytest.mark.asyncio
+async def test_deterministic_and_incomplete_blocks_never_request_approval(
+    tmp_path: Path,
+    command: str,
+) -> None:
+    approval_calls = 0
+
+    async def approve(_request: SafetyApprovalRequest) -> bool:
+        nonlocal approval_calls
+        approval_calls += 1
+        return True
+
+    runtime = _runtime(tmp_path, "guarded", approve)
+    reviewer = _StubReviewer(decision=_deferred())
+    runtime._reviewer = reviewer
+
+    result = await runtime.invoke_exec(
+        ctx=_ctx(),
+        arguments={"cmd": command},
+        invoke_tool=_noop_invoke,
+    )
+
+    assert json.loads(result)["status"] == "blocked"
+    assert reviewer.calls == 0
+    assert approval_calls == 0
+
+
+@pytest.mark.parametrize(
+    "decision",
+    [
+        SafetyDecision(
+            allowed=False,
+            source="review_error",
+            reason="provider failed",
+            categories=("review_error",),
+        ),
+        SafetyDecision(
+            allowed=False,
+            source="reviewer",
+            reason="confidently destructive",
+            categories=("destructive_effect",),
+            risk="high",
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_reviewer_errors_and_confident_blocks_never_request_approval(
+    tmp_path: Path,
+    decision: SafetyDecision,
+) -> None:
+    approval_calls = 0
+
+    async def approve(_request: SafetyApprovalRequest) -> bool:
+        nonlocal approval_calls
+        approval_calls += 1
+        return True
+
+    runtime = _runtime(tmp_path, "guarded", approve)
+    runtime._reviewer = _StubReviewer(decision=decision)
+
+    result = await runtime.invoke_exec(
+        ctx=_ctx(),
+        arguments={"cmd": "nmap -sV example.test"},
+        invoke_tool=_noop_invoke,
+    )
+
+    assert json.loads(result)["status"] == "blocked"
+    assert approval_calls == 0
+
+
 @pytest.mark.asyncio
 async def test_review_does_not_hold_the_workspace_lock(tmp_path: Path) -> None:
     runtime = _runtime(tmp_path, "guarded")
@@ -333,6 +624,37 @@ async def test_workspace_change_during_review_invalidates_the_decision(tmp_path:
     assert invoked is False
 
 
+@pytest.mark.asyncio
+async def test_workspace_change_during_human_approval_invalidates_the_decision(
+    tmp_path: Path,
+) -> None:
+    runtime: SafetyRuntime
+
+    async def approve(_request: SafetyApprovalRequest) -> bool:
+        runtime._workspace_epoch += 1
+        return True
+
+    runtime = _runtime(tmp_path, "guarded", approve)
+    runtime._reviewer = _StubReviewer(decision=_deferred())
+    invoked = False
+
+    async def invoke(_ctx: Any, _raw_input: str) -> str:
+        nonlocal invoked
+        invoked = True
+        return "bad"
+
+    result = await runtime.invoke_exec(
+        ctx=_script_ctx(),
+        arguments={"cmd": "python /workspace/app.py"},
+        invoke_tool=invoke,
+    )
+
+    payload = json.loads(result)
+    assert payload["status"] == "blocked"
+    assert payload["safety"]["categories"] == ["stale_evidence"]
+    assert invoked is False
+
+
 @pytest.mark.asyncio
 async def test_unchanged_workspace_executes_after_review(tmp_path: Path) -> None:
     runtime = _runtime(tmp_path, "guarded")
@@ -350,28 +672,6 @@ async def test_unchanged_workspace_executes_after_review(tmp_path: Path) -> None
     assert result == "ran"
 
 
-@pytest.mark.asyncio
-async def test_observe_mode_blocks_a_workspace_patch(tmp_path: Path) -> None:
-    invoked = False
-
-    async def invoke(_ctx: Any, _raw_input: str) -> str:
-        nonlocal invoked
-        invoked = True
-        return "bad"
-
-    result = await _runtime(tmp_path, "observe").invoke_mutating_tool(
-        ctx=_ctx(),
-        tool_name="apply_patch",
-        raw_input="{}",
-        invoke_tool=invoke,
-    )
-
-    payload = json.loads(result)
-    assert payload["status"] == "blocked"
-    assert payload["safety"]["categories"] == ["state_mutation"]
-    assert invoked is False
-
-
 @pytest.mark.asyncio
 async def test_mutating_tool_is_untouched_when_safety_is_off(tmp_path: Path) -> None:
     async def invoke(_ctx: Any, _raw_input: str) -> str:
@@ -488,18 +788,15 @@ async def test_a_read_only_command_leaves_the_epoch_alone(tmp_path: Path) -> Non
         "agent-browser session clear",
     ],
 )
-async def test_observe_mode_blocks_grouped_browser_verbs(tmp_path: Path, command: str) -> None:
+async def test_grouped_browser_verbs_reach_the_reviewer(tmp_path: Path, command: str) -> None:
     """The bare verb sits in the passive set, so these are the commands that would slip
     through if passivity were decided on the verb alone."""
-    runtime = _runtime(tmp_path, "observe")
+    runtime = _runtime(tmp_path, "guarded")
     reviewer = _StubReviewer()
     runtime._reviewer = reviewer
-    invoked = False
 
     async def invoke(_ctx: Any, _raw_input: str) -> str:
-        nonlocal invoked
-        invoked = True
-        return "bad"
+        return "reviewed"
 
     result = await runtime.invoke_exec(
         ctx=_ctx(),
@@ -507,33 +804,13 @@ async def test_observe_mode_blocks_grouped_browser_verbs(tmp_path: Path, command
         invoke_tool=invoke,
     )
 
-    payload = json.loads(result)
-    assert payload["status"] == "blocked"
-    assert payload["safety"]["categories"] == ["target_mutation"]
-    assert reviewer.calls == 0
-    assert invoked is False
-
-
-@pytest.mark.asyncio
-async def test_guarded_grouped_browser_verb_reaches_the_reviewer(tmp_path: Path) -> None:
-    """In guarded mode it loses only the fast path; the reviewer still gets to decide."""
-    runtime = _runtime(tmp_path, "guarded")
-    reviewer = _StubReviewer()
-    runtime._reviewer = reviewer
-
-    result = await runtime.invoke_exec(
-        ctx=_ctx(),
-        arguments={"cmd": "agent-browser tab new https://example.test/admin"},
-        invoke_tool=_noop_invoke,
-    )
-
-    assert result == "patched"
+    assert result == "reviewed"
     assert reviewer.calls == 1
 
 
 @pytest.mark.asyncio
 async def test_bare_tab_listing_keeps_the_fast_path(tmp_path: Path) -> None:
-    runtime = _runtime(tmp_path, "observe")
+    runtime = _runtime(tmp_path, "guarded")
     reviewer = _StubReviewer()
     runtime._reviewer = reviewer
 

tests/test_tui_backend_controller.py

@@ -4,6 +4,7 @@ import argparse
 import asyncio
 import os
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -327,9 +328,9 @@ async def test_stop_rejects_terminal_agents(status: str) -> None:
         def __init__(self) -> None:
             self.calls: list[str] = []
 
-        async def cancel_descendants_graceful(self, agent_id: str) -> bool:
+        async def cancel_descendants_graceful(self, agent_id: str) -> list[str]:
             self.calls.append(agent_id)
-            return True
+            return [agent_id]
 
     coordinator = Coordinator()
     controller = TuiController(args(), coordinator=coordinator)
@@ -349,9 +350,9 @@ async def test_stop_allows_active_agents(status: str) -> None:
         def __init__(self) -> None:
             self.calls: list[str] = []
 
-        async def cancel_descendants_graceful(self, agent_id: str) -> bool:
+        async def cancel_descendants_graceful(self, agent_id: str) -> list[str]:
             self.calls.append(agent_id)
-            return True
+            return [agent_id]
 
     coordinator = Coordinator()
     controller = TuiController(args(), coordinator=coordinator)
@@ -364,11 +365,41 @@ async def test_stop_allows_active_agents(status: str) -> None:
     assert coordinator.calls == ["agent-1"]
 
 
+@pytest.mark.asyncio
+async def test_stopping_agent_denies_pending_approvals_for_its_subtree() -> None:
+    class Coordinator:
+        async def cancel_descendants_graceful(self, agent_id: str) -> list[str]:
+            return ["agent-child", agent_id]
+
+    controller = TuiController(args(), coordinator=Coordinator())
+    controller.set_runtime(scan_loop=asyncio.get_running_loop())
+    controller.live_view.upsert_agent("agent-1", name="Agent", status="running")
+    approvals = [
+        asyncio.create_task(
+            controller.safety_approval_callback(
+                {
+                    "request_id": f"approval-{agent_id}",
+                    "agent_id": agent_id,
+                    "action": "Run action",
+                    "reason": "Ambiguous effect",
+                }
+            )
+        )
+        for agent_id in ("agent-1", "agent-child")
+    ]
+    await asyncio.sleep(0)
+
+    await controller.handle("agent.stop", {"agent_id": "agent-1"})
+
+    assert await asyncio.gather(*approvals) == ["cancelled", "cancelled"]
+    assert controller.snapshot()["pending_approval"] is None
+
+
 @pytest.mark.asyncio
 async def test_stop_handles_coordinator_rejection_after_stale_active_projection() -> None:
     class Coordinator:
-        async def cancel_descendants_graceful(self, _agent_id: str) -> bool:
-            return False
+        async def cancel_descendants_graceful(self, _agent_id: str) -> list[str]:
+            return []
 
     controller = TuiController(args(), coordinator=Coordinator())
     controller.set_runtime(scan_loop=asyncio.get_running_loop())
@@ -385,6 +416,147 @@ async def test_unknown_command_is_rejected() -> None:
         await controller.handle("nope", {})
 
 
+@pytest.mark.asyncio
+async def test_safety_approvals_queue_and_resolve_in_order() -> None:
+    controller = TuiController(args())
+    first = asyncio.create_task(
+        controller.safety_approval_callback(
+            {"request_id": "approval-1", "action": "Run exploit", "reason": "Mutates state"}
+        )
+    )
+    second = asyncio.create_task(
+        controller.safety_approval_callback(
+            SimpleNamespace(
+                request_id="approval-2",
+                action="Write a file",
+                reason="Changes the workspace",
+            )
+        )
+    )
+    await asyncio.sleep(0)
+
+    assert controller.snapshot()["pending_approval"] == {
+        "request_id": "approval-1",
+        "action": "Run exploit",
+        "reason": "Mutates state",
+        "agent_id": "",
+        "tool_name": "",
+        "digest": "",
+        "risk": "",
+    }
+    with pytest.raises(ValueError, match="duplicate safety approval request_id"):
+        await controller.safety_approval_callback(
+            {"request_id": "approval-1", "action": "Duplicate", "reason": "Duplicate"}
+        )
+    with pytest.raises(RuntimeError, match="stale or unknown"):
+        await controller.handle("safety.resolve", {"request_id": "approval-2", "approved": True})
+
+    assert await controller.handle(
+        "safety.resolve", {"request_id": "approval-1", "approved": True}
+    ) == {"request_id": "approval-1", "approved": True}
+    assert await first is True
+    assert controller.snapshot()["pending_approval"]["request_id"] == "approval-2"
+
+    with pytest.raises(RuntimeError, match="stale or unknown"):
+        await controller.handle("safety.resolve", {"request_id": "approval-1", "approved": False})
+    await controller.handle("safety.resolve", {"request_id": "approval-2", "approved": False})
+    assert await second is False
+    assert controller.snapshot()["pending_approval"] is None
+    with pytest.raises(RuntimeError, match="No safety approval is pending"):
+        await controller.handle("safety.resolve", {"request_id": "approval-2", "approved": False})
+
+
+@pytest.mark.asyncio
+async def test_safety_approval_validates_response_and_sanitizes_display() -> None:
+    controller = TuiController(args())
+    pending = asyncio.create_task(
+        controller.safety_approval_callback(
+            {
+                "request_id": "approval-safe",
+                "action": "run\x1b]52;c;Y2xpcA==\x07 command\x85",
+                "reason": "needs\x1b[31m review\x1b[0m\x7f",
+            }
+        )
+    )
+    await asyncio.sleep(0)
+
+    assert controller.snapshot()["pending_approval"] == {
+        "request_id": "approval-safe",
+        "action": "run command",
+        "reason": "needs review",
+        "agent_id": "",
+        "tool_name": "",
+        "digest": "",
+        "risk": "",
+    }
+    with pytest.raises(TypeError, match="approved must be a boolean"):
+        await controller.handle(
+            "safety.resolve", {"request_id": "approval-safe", "approved": "yes"}
+        )
+    with pytest.raises(ValueError, match="request_id must be a non-empty string"):
+        await controller.handle("safety.resolve", {"request_id": "", "approved": False})
+
+    await controller.handle("safety.resolve", {"request_id": "approval-safe", "approved": False})
+    assert await pending is False
+
+    assert (
+        await controller.safety_approval_callback(
+            {"request_id": "approval-long", "action": "x" * 513, "reason": "Too long"}
+        )
+        is False
+    )
+    assert controller.snapshot()["pending_approval"] is None
+
+
+@pytest.mark.asyncio
+async def test_cancelled_safety_request_is_removed_and_reveals_next() -> None:
+    controller = TuiController(args())
+    first = asyncio.create_task(
+        controller.safety_approval_callback(
+            {"request_id": "approval-1", "action": "First", "reason": "First reason"}
+        )
+    )
+    second = asyncio.create_task(
+        controller.safety_approval_callback(
+            {"request_id": "approval-2", "action": "Second", "reason": "Second reason"}
+        )
+    )
+    await asyncio.sleep(0)
+
+    first.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await first
+
+    assert controller.snapshot()["pending_approval"]["request_id"] == "approval-2"
+    await controller.handle("safety.resolve", {"request_id": "approval-2", "approved": False})
+    assert await second is False
+
+
+@pytest.mark.asyncio
+async def test_quit_denies_all_pending_and_future_safety_approvals() -> None:
+    controller = TuiController(args())
+    requests = [
+        asyncio.create_task(
+            controller.safety_approval_callback(
+                {"request_id": f"approval-{index}", "action": "Action", "reason": "Reason"}
+            )
+        )
+        for index in range(2)
+    ]
+    await asyncio.sleep(0)
+
+    await controller.handle("app.quit", {})
+
+    assert await asyncio.gather(*requests) == ["cancelled", "cancelled"]
+    assert controller.snapshot()["pending_approval"] is None
+    assert (
+        await controller.safety_approval_callback(
+            {"request_id": "approval-late", "action": "Late", "reason": "Late reason"}
+        )
+        == "cancelled"
+    )
+
+
 def test_messages_are_sanitized_and_agents_are_collection_only() -> None:
     controller = TuiController(args())
     controller.add_message("replace\x1b]52;c;Y2xpcA==\x07 key\x85")

tests/test_tui_backend_server.py

@@ -124,7 +124,7 @@ async def test_server_requires_ready_before_state_or_commands() -> None:
     try:
         hello = await receive_message(child)
         assert hello == {
-            "version": 3,
+            "version": PROTOCOL_VERSION,
             "type": "hello",
             "payload": {"capabilities": list(PROTOCOL_CAPABILITIES)},
         }
@@ -135,7 +135,7 @@ async def test_server_requires_ready_before_state_or_commands() -> None:
         await send_message(
             child,
             {
-                "version": 3,
+                "version": PROTOCOL_VERSION,
                 "type": "ready",
                 "payload": {"capabilities": list(PROTOCOL_CAPABILITIES)},
             },
@@ -154,7 +154,7 @@ async def test_server_requires_ready_before_state_or_commands() -> None:
     ("version", "capabilities"),
     [
         (2, list(PROTOCOL_CAPABILITIES)),
-        (3, ["state-revisions"]),
+        (PROTOCOL_VERSION, ["state-revisions"]),
     ],
 )
 async def test_server_rejects_handshake_mismatch(version: int, capabilities: list[str]) -> None:
@@ -186,7 +186,7 @@ async def test_server_command_round_trip_over_inherited_socket() -> None:
         await send_message(
             child,
             {
-                "version": 3,
+                "version": PROTOCOL_VERSION,
                 "type": "setup.add_target",
                 "request_id": "test-1",
                 "payload": {"target": "example.com"},
@@ -252,7 +252,7 @@ async def test_persistence_error_does_not_kill_command_reader(
             await send_message(
                 child,
                 {
-                    "version": 3,
+                    "version": PROTOCOL_VERSION,
                     "type": "setup.select_model",
                     "request_id": request_id,
                     "payload": {"provider": "openai", "model": "openai/gpt-5"},
@@ -295,7 +295,7 @@ async def test_invalid_version_error_is_correlated_and_next_command_succeeds() -
         await send_message(
             child,
             {
-                "version": 3,
+                "version": PROTOCOL_VERSION,
                 "type": "setup.add_target",
                 "request_id": "after-error",
                 "payload": {"target": "example.com"},
@@ -408,7 +408,7 @@ async def test_agents_collection_has_no_state_cap_and_sends_delete_and_resync()
         await send_message(
             child,
             {
-                "version": 3,
+                "version": PROTOCOL_VERSION,
                 "type": "collection.resync",
                 "request_id": "resync-agents",
                 "payload": {"collection": "agents"},