Merge remote-tracking branch 'origin/main' into mcp-support

# Conflicts:
#	strix/interface/viewer/frontend/src/components/live/tool-renderers/index.ts
#	strix/interface/viewer/static/assets/index-Bi_X6kI3.js
#	strix/interface/viewer/static/assets/index-DBJ-RJqo.js
#	strix/interface/viewer/static/assets/index-gEZK6bjO.js
#	strix/interface/viewer/static/index.html

tests/test_agent_tool_registration.py

@@ -112,3 +112,19 @@ def test_wait_for_agents_is_available_in_both_modes() -> None:
     for interactive in (True, False):
         agent = factory.build_strix_agent(is_root=True, interactive=interactive)
         assert "wait_for_agents" in [t.name for t in agent.tools]
+
+
+def test_strict_tool_schemas_can_be_disabled_per_route() -> None:
+    """Claude routes cap strict tools; the toolset must be sendable without strict."""
+    agent = factory.build_strix_agent(is_root=True, strict_tool_schemas=False)
+
+    function_tools = [t for t in agent.tools if isinstance(t, FunctionTool)]
+    assert function_tools
+    assert not any(t.strict_json_schema for t in function_tools)
+
+
+def test_disabling_strict_leaves_shared_tools_untouched() -> None:
+    factory.build_strix_agent(is_root=True, strict_tool_schemas=False)
+    agent = factory.build_strix_agent(is_root=True)
+
+    assert any(t.strict_json_schema for t in agent.tools if isinstance(t, FunctionTool))

tests/test_caido_bootstrap.py

@@ -0,0 +1,81 @@
+"""A bootstrap that dies mid-setup must not leave its transport behind.
+
+The bootstrap now runs concurrently with the scan start, so teardown can
+cancel it at any await — including inside ``Client.connect()``, where the
+client exists but no caller will ever see it to close it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import types
+from typing import Any
+
+import pytest
+
+from strix.runtime.caido_bootstrap import bootstrap_caido
+
+
+class _FakeExecResult:
+    stderr = b""
+    exit_code = 0
+
+    def __init__(self, stdout: str) -> None:
+        self.stdout = stdout
+
+    def ok(self) -> bool:
+        return True
+
+
+class _FakeSession:
+    async def exec(self, *_args: Any, **_kwargs: Any) -> _FakeExecResult:
+        return _FakeExecResult('{"data":{"loginAsGuest":{"token":{"accessToken":"t"}}}}')
+
+
+class _FakeClient:
+    def __init__(self, connect_error: BaseException) -> None:
+        self.connect_error = connect_error
+        self.closed = False
+
+    async def connect(self) -> None:
+        raise self.connect_error
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+async def _bootstrap_expecting(
+    monkeypatch: pytest.MonkeyPatch, error: BaseException
+) -> _FakeClient:
+    """Run a bootstrap whose ``connect()`` fails with ``error``."""
+    client = _FakeClient(error)
+    # The SDK is imported inside bootstrap_caido (it is slow to import), so the
+    # fakes are injected as the modules it imports.
+    sdk = types.ModuleType("caido_sdk_client")
+    sdk.Client = lambda *_a, **_k: client  # type: ignore[attr-defined]
+    sdk.TokenAuthOptions = lambda token: token  # type: ignore[attr-defined]
+    sdk_types = types.ModuleType("caido_sdk_client.types")
+    sdk_types.CreateProjectOptions = lambda **_k: None  # type: ignore[attr-defined]
+    monkeypatch.setitem(sys.modules, "caido_sdk_client", sdk)
+    monkeypatch.setitem(sys.modules, "caido_sdk_client.types", sdk_types)
+
+    with pytest.raises(type(error)):
+        await bootstrap_caido(
+            _FakeSession(),  # type: ignore[arg-type]
+            host_url="http://host",
+            container_url="http://container",
+        )
+    return client
+
+
+async def test_cancellation_during_connect_closes_the_client(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    client = await _bootstrap_expecting(monkeypatch, asyncio.CancelledError())
+    assert client.closed
+
+
+async def test_failed_connect_closes_the_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = await _bootstrap_expecting(monkeypatch, RuntimeError("no listener"))
+    assert client.closed

tests/test_caido_handle.py

@@ -0,0 +1,103 @@
+"""Tests for the concurrent Caido bootstrap handle."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from strix.runtime.caido_handle import CaidoBootstrapHandle
+
+
+class _FakeClient:
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+def _handle(coro: Any) -> CaidoBootstrapHandle:
+    return CaidoBootstrapHandle(asyncio.ensure_future(coro))
+
+
+async def test_get_waits_for_the_bootstrap() -> None:
+    client = _FakeClient()
+    started = asyncio.Event()
+
+    async def _bootstrap() -> Any:
+        started.set()
+        await asyncio.sleep(0.01)
+        return client
+
+    handle = _handle(_bootstrap())
+    await started.wait()
+    assert handle.peek() is None
+    assert await handle.get() is client
+    assert handle.peek() is client
+
+
+async def test_get_reraises_bootstrap_failure_to_every_caller() -> None:
+    async def _bootstrap() -> Any:
+        raise RuntimeError("caido never came up")
+
+    handle = _handle(_bootstrap())
+    for _ in range(2):
+        with pytest.raises(RuntimeError, match="caido never came up"):
+            await handle.get()
+    assert handle.peek() is None
+
+
+async def test_caller_cancellation_does_not_cancel_the_shared_bootstrap() -> None:
+    client = _FakeClient()
+
+    async def _bootstrap() -> Any:
+        await asyncio.sleep(0.05)
+        return client
+
+    handle = _handle(_bootstrap())
+
+    with pytest.raises(TimeoutError):
+        await asyncio.wait_for(handle.get(), timeout=0.01)
+
+    assert await handle.get() is client
+
+
+async def test_aclose_closes_a_finished_client() -> None:
+    client = _FakeClient()
+
+    async def _bootstrap() -> Any:
+        return client
+
+    handle = _handle(_bootstrap())
+    await handle.get()
+    await handle.aclose()
+    assert client.closed is True
+
+
+async def test_aclose_cancels_an_in_flight_bootstrap() -> None:
+    cancelled = asyncio.Event()
+
+    async def _bootstrap() -> Any:
+        try:
+            await asyncio.sleep(10)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+        return _FakeClient()
+
+    handle = _handle(_bootstrap())
+    await asyncio.sleep(0)
+    await handle.aclose()
+    assert cancelled.is_set()
+
+
+async def test_aclose_swallows_a_failed_bootstrap() -> None:
+    async def _bootstrap() -> Any:
+        raise RuntimeError("boom")
+
+    handle = _handle(_bootstrap())
+    with pytest.raises(RuntimeError, match="boom"):
+        await handle.get()
+    await handle.aclose()

tests/test_cli_target_list.py

@@ -227,3 +227,18 @@ def test_resume_still_requires_targets_or_a_workspace(
         cli_main.parse_arguments()
 
     assert "has no targets_info" in capsys.readouterr().err
+
+def test_resume_non_object_run_json_exits(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+    monkeypatch.chdir(tmp_path)
+    run_dir = tmp_path / "strix_runs" / "pentest_abcd"
+    run_dir.mkdir(parents=True)
+    (run_dir / "run.json").write_text("[]", encoding="utf-8")
+
+    monkeypatch.setattr(sys, "argv", ["strix", "--resume", "pentest_abcd"])
+    with pytest.raises(SystemExit) as exc_info:
+        cli_main.parse_arguments()
+
+    assert exc_info.value.code == 2
+    captured = capsys.readouterr()
+    assert "run.json unreadable" in captured.err
+    assert "not an object" in captured.err

tests/test_context_budget.py

@@ -31,7 +31,7 @@ def test_context_window_chatgpt_prefix_skips_provider_auth(
         calls.append(model)
         return {"max_input_tokens": 1_050_000, "max_output_tokens": 128_000}
 
-    monkeypatch.setattr("strix.llm.context_budget.litellm.get_model_info", _model_info)
+    monkeypatch.setattr("litellm.get_model_info", _model_info)
     try:
         assert context_budget.context_window("chatgpt/gpt-5.6-luna") == 1_050_000
         assert calls == ["gpt-5.6-luna"]
@@ -45,7 +45,7 @@ def test_context_window_unmapped_uses_fallback(monkeypatch: pytest.MonkeyPatch)
     def _raise(_model: str) -> dict[str, int]:
         raise ValueError("This model isn't mapped yet.")
 
-    monkeypatch.setattr("strix.llm.context_budget.litellm.get_model_info", _raise)
+    monkeypatch.setattr("litellm.get_model_info", _raise)
     expected = load_settings().context.fallback_context_tokens
     assert context_budget.context_window("totally-made-up-model") == expected
     context_budget._model_info.cache_clear()
@@ -55,7 +55,7 @@ def test_count_tokens_fallback_on_error(monkeypatch: pytest.MonkeyPatch) -> None
     def _raise(**_kwargs: object) -> int:
         raise RuntimeError("no tokenizer")
 
-    monkeypatch.setattr("strix.llm.context_budget.litellm.token_counter", _raise)
+    monkeypatch.setattr("litellm.token_counter", _raise)
     # Falls back to UTF-8 byte length (upper bound on tokens).
     assert context_budget.count_tokens("weird-model", "x" * 400) == 400
     assert context_budget.count_tokens("weird-model", "😀" * 10) == 40

tests/test_coverage_tool.py

@@ -0,0 +1,284 @@
+"""Tests for the scan coverage ledger."""
+
+from __future__ import annotations
+
+import json
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+from strix.tools.coverage.tools import (
+    _list_impl,
+    _record_impl,
+    _update_impl,
+    get_coverage_entries,
+    hydrate_coverage_from_disk,
+    outcome_counts,
+)
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+@pytest.fixture(autouse=True)
+def coverage_store(tmp_path: Path) -> Path:
+    hydrate_coverage_from_disk(tmp_path)
+    return tmp_path
+
+
+def _record(**overrides: str) -> dict[str, Any]:
+    kwargs = {
+        "surface": "POST /api/orders/{id}",
+        "risk_area": "object-level authorization",
+        "outcome": "no_issue_found",
+        "evidence": "Tested with two tenants; both received 403.",
+        "agent_id": "agent-1",
+        "agent_name": "authz-tester",
+    }
+    kwargs.update(overrides)
+    return _record_impl(**kwargs)
+
+
+def test_record_persists_entry(coverage_store: Path) -> None:
+    result = _record()
+    assert result["success"] is True
+
+    entries = get_coverage_entries()
+    assert len(entries) == 1
+    assert entries[0]["surface"] == "POST /api/orders/{id}"
+    assert entries[0]["outcome"] == "no_issue_found"
+    assert entries[0]["agent_name"] == "authz-tester"
+    assert (coverage_store / "coverage.json").exists()
+
+
+def test_record_normalizes_outcome() -> None:
+    assert _record(outcome="Needs Follow-Up")["success"] is True
+    assert get_coverage_entries()[0]["outcome"] == "needs_follow_up"
+
+
+def test_record_rejects_unknown_outcome() -> None:
+    result = _record(outcome="looks fine")
+    assert result["success"] is False
+    assert any("Invalid outcome" in e for e in result["errors"])
+    assert not get_coverage_entries()
+
+
+def test_record_requires_surface_and_risk_area() -> None:
+    result = _record(surface="  ", risk_area="")
+    assert result["success"] is False
+    joined = " ".join(result["errors"])
+    assert "surface" in joined
+    assert "risk_area" in joined
+
+
+@pytest.mark.parametrize("outcome", ["ruled_out", "not_applicable", "needs_follow_up"])
+def test_evidence_required_for_asserted_outcomes(outcome: str) -> None:
+    result = _record(outcome=outcome, evidence="   ")
+    assert result["success"] is False
+    assert any("evidence is required" in e for e in result["errors"])
+
+
+def test_evidence_optional_for_reported() -> None:
+    assert _record(outcome="reported", evidence="")["success"] is True
+
+
+def test_outcome_counts_and_filtering() -> None:
+    _record(surface="/login", outcome="reported", evidence="")
+    _record(surface="/search", outcome="no_issue_found")
+    _record(surface="/upload", outcome="needs_follow_up", evidence="No credentials to test.")
+
+    assert outcome_counts() == {"reported": 1, "no_issue_found": 1, "needs_follow_up": 1}
+
+    listed = _list_impl(outcome="needs_follow_up", surface=None, caller_agent_id="agent-1")
+    assert listed["filtered_count"] == 1
+    assert listed["entries"][0]["surface"] == "/upload"
+    assert listed["entries"][0]["by_you"] is True
+
+    by_surface = _list_impl(outcome=None, surface="sea", caller_agent_id=None)
+    assert by_surface["filtered_count"] == 1
+    assert by_surface["entries"][0]["surface"] == "/search"
+
+
+def test_list_rejects_unknown_outcome_filter() -> None:
+    result = _list_impl(outcome="bogus", surface=None, caller_agent_id=None)
+    assert result["success"] is False
+
+
+def test_hydrate_reloads_from_disk(coverage_store: Path) -> None:
+    _record()
+    hydrate_coverage_from_disk(coverage_store)
+    entries = get_coverage_entries()
+    assert len(entries) == 1
+    assert entries[0]["risk_area"] == "object-level authorization"
+
+
+def _update(entry_id: str, **overrides: str) -> dict[str, Any]:
+    kwargs = {
+        "entry_id": entry_id,
+        "outcome": "reported",
+        "evidence": "Got staging credentials and confirmed the IDOR.",
+        "agent_id": "agent-2",
+        "agent_name": "followup-tester",
+    }
+    kwargs.update(overrides)
+    return _update_impl(**kwargs)
+
+
+def test_update_moves_outcome_and_keeps_history() -> None:
+    recorded = _record(outcome="needs_follow_up", evidence="No credentials to test.")
+    entry_id = str(recorded["entry_id"])
+
+    result = _update(entry_id)
+
+    assert result["success"] is True
+    assert result["previous_outcome"] == "needs_follow_up"
+    assert result["outcome"] == "reported"
+
+    entries = get_coverage_entries()
+    assert len(entries) == 1, "update must not create a parallel entry"
+    entry = entries[0]
+    assert entry["outcome"] == "reported"
+    assert entry["agent_name"] == "followup-tester"
+    assert entry["history"] == [
+        {
+            "outcome": "needs_follow_up",
+            "recorded_at": entry["created_at"],
+            "evidence": "No credentials to test.",
+            "agent_name": "authz-tester",
+        }
+    ]
+    assert outcome_counts() == {"reported": 1}
+
+
+def test_update_can_reopen_a_closed_entry() -> None:
+    recorded = _record(outcome="ruled_out", evidence="Guard at auth.py:40 covers the path.")
+    entry_id = str(recorded["entry_id"])
+
+    _update(
+        entry_id,
+        outcome="needs_follow_up",
+        evidence="The guard is skipped on the /v2 alias; reachability unproven.",
+    )
+
+    assert outcome_counts() == {"needs_follow_up": 1}
+    listed = _list_impl(outcome=None, surface=None, caller_agent_id=None)
+    assert listed["entries"][0]["previous_outcomes"] == ["ruled_out"]
+
+
+def test_update_enforces_evidence_for_closing_outcomes() -> None:
+    entry_id = str(_record(outcome="needs_follow_up", evidence="unknown")["entry_id"])
+
+    result = _update(entry_id, outcome="ruled_out", evidence="  ")
+
+    assert result["success"] is False
+    assert get_coverage_entries()[0]["outcome"] == "needs_follow_up"
+
+
+def test_update_rejects_unknown_entry() -> None:
+    result = _update("nope")
+    assert result["success"] is False
+    assert "list_coverage" in str(result["error"])
+
+
+def test_update_persists_to_disk(coverage_store: Path) -> None:
+    entry_id = str(_record(outcome="needs_follow_up", evidence="No creds.")["entry_id"])
+    _update(entry_id)
+
+    hydrate_coverage_from_disk(coverage_store)
+
+    entry = get_coverage_entries()[0]
+    assert entry["outcome"] == "reported"
+    assert len(entry["history"]) == 1
+
+
+def test_recording_a_duplicate_surface_is_refused_with_the_existing_id() -> None:
+    first = _record_impl(
+        surface="/api/invoices",
+        risk_area="IDOR",
+        outcome="needs_follow_up",
+        evidence="No second tenant account to test cross-tenant reads with.",
+        agent_id="a1",
+        agent_name="Recon",
+    )
+
+    duplicate = _record_impl(
+        surface="  /API/Invoices ",
+        risk_area="idor",
+        outcome="reported",
+        evidence="Cross-tenant read confirmed.",
+        agent_id="a2",
+        agent_name="Authz",
+    )
+
+    assert duplicate["success"] is False
+    assert duplicate["existing_entry_id"] == first["entry_id"]
+    assert duplicate["existing_outcome"] == "needs_follow_up"
+    assert "update_coverage" in duplicate["error"]
+    assert len(get_coverage_entries()) == 1
+
+
+def test_a_different_risk_area_on_one_surface_is_still_its_own_entry() -> None:
+    _record_impl(
+        surface="/api/invoices",
+        risk_area="IDOR",
+        outcome="no_issue_found",
+        evidence="Tenant id read from the session.",
+        agent_id="a1",
+        agent_name="Authz",
+    )
+    second = _record_impl(
+        surface="/api/invoices",
+        risk_area="SQL injection",
+        outcome="no_issue_found",
+        evidence="Parameterized throughout.",
+        agent_id="a1",
+        agent_name="Injection",
+    )
+
+    assert second["success"] is True
+    assert len(get_coverage_entries()) == 2
+
+
+def test_concurrent_records_of_one_surface_yield_a_single_row() -> None:
+    """Duplicate detection and insertion must be one critical section.
+
+    Two agents recording the same surface at the same moment would otherwise
+    both pass the "no duplicate" check, and the report would show a stale
+    conclusion beside its replacement — the exact outcome the rejection exists
+    to prevent.
+    """
+    barrier = threading.Barrier(8)
+
+    def attempt(index: int) -> dict[str, Any]:
+        barrier.wait()
+        return _record(agent_id=f"agent-{index}", agent_name=f"tester-{index}")
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        results = list(pool.map(attempt, range(8)))
+
+    assert sum(1 for result in results if result["success"]) == 1
+    assert len(get_coverage_entries()) == 1
+
+
+def test_concurrent_records_all_survive_persistence(coverage_store: Path) -> None:
+    """A writer holding an older snapshot must not win the rename.
+
+    If it did, the mirror would come back short on resume and coverage
+    recorded before a crash would silently disappear from the report.
+    """
+    barrier = threading.Barrier(8)
+
+    def attempt(index: int) -> dict[str, Any]:
+        barrier.wait()
+        return _record(surface=f"GET /api/resource/{index}", agent_id=f"agent-{index}")
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(attempt, range(8)))
+
+    persisted = json.loads((coverage_store / "coverage.json").read_text(encoding="utf-8"))
+    assert len(persisted) == 8
+    hydrate_coverage_from_disk(coverage_store)
+    assert len(get_coverage_entries()) == 8

tests/test_finish_coverage_gate.py

@@ -0,0 +1,65 @@
+"""finish_scan confronts the root agent with the coverage the runtime can see."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from strix.tools.coverage.tools import _record_impl, hydrate_coverage_from_disk
+from strix.tools.finish.tool import _coverage_summary
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+_GRAPH = {
+    "statuses": {"agent-1": "completed"},
+    "names": {"agent-1": "injection-tester"},
+    "metadata": {"agent-1": {"skills": ["sql_injection", "xss"]}},
+}
+
+
+@pytest.fixture(autouse=True)
+def _empty_ledger(tmp_path: Path) -> None:
+    hydrate_coverage_from_disk(tmp_path)
+
+
+def _record(risk_area: str) -> None:
+    _record_impl(
+        surface="POST /api/orders/{id}",
+        risk_area=risk_area,
+        outcome="no_issue_found",
+        evidence="Parameters fuzzed; no anomalies.",
+        agent_id="agent-1",
+        agent_name="injection-tester",
+    )
+
+
+def test_unrecorded_risk_class_is_reported_back_to_the_root_agent() -> None:
+    _record("SQL injection")
+
+    summary = _coverage_summary(_GRAPH)
+
+    assert summary["coverage_recorded"] == 1
+    assert len(summary["coverage_gaps"]) == 1
+    assert "xss" in summary["coverage_gaps"][0]
+    assert "unexamined" in summary["coverage_gap_warning"]
+
+
+def test_fully_accounted_coverage_raises_no_gap_warning() -> None:
+    _record("SQL injection")
+    _record("cross-site scripting")
+
+    summary = _coverage_summary(_GRAPH)
+
+    assert "coverage_gaps" not in summary
+    assert "coverage_gap_warning" not in summary
+
+
+def test_an_empty_ledger_still_warns_first() -> None:
+    summary = _coverage_summary(_GRAPH)
+
+    assert summary["coverage_recorded"] == 0
+    assert "No coverage was recorded" in summary["coverage_warning"]

tests/test_inputs.py

@@ -10,6 +10,7 @@ import pytest
 
 from strix.core.inputs import (
     build_root_task,
+    build_scan_targets,
     build_scope_context,
     child_initial_input,
     make_model_settings,
@@ -363,6 +364,35 @@ def test_make_model_settings_timeout_survives_reasoning_resolve() -> None:
     assert settings.extra_args["timeout"] == 120.0
 
 
+def test_scan_targets_prefer_the_workspace_checkout_over_the_remote_url() -> None:
+    config = {
+        "targets": [
+            {
+                "type": "repository",
+                "details": {
+                    "target_repo": "https://github.com/acme/billing",
+                    "workspace_subdir": "billing",
+                },
+            },
+            {"type": "web_application", "details": {"target_url": "https://app.example.com"}},
+        ]
+    }
+
+    assert build_scan_targets(config) == ["/workspace/billing", "https://app.example.com"]
+
+
+def test_scan_targets_drop_empty_and_duplicate_entries() -> None:
+    config = {
+        "targets": [
+            {"type": "web_application", "details": {"target_url": "https://app.example.com"}},
+            {"type": "web_application", "details": {"target_url": "https://app.example.com"}},
+            {"type": "ip_address", "details": {}},
+        ]
+    }
+
+    assert build_scan_targets(config) == ["https://app.example.com"]
+
+
 def test_openrouter_attribution_rides_on_the_request_headers() -> None:
     # litellm.headers is ignored once a request carries any header of its own,
     # so the attribution must be part of the per-request headers.

tests/test_models.py

@@ -9,6 +9,7 @@ from strix.config.models import (
     RECOMMENDED_MODEL_NAMES,
     is_recommended_or_frontier_model,
     request_timeout_extra_args,
+    supports_strict_tool_schemas,
 )
 
 
@@ -90,3 +91,24 @@ def test_frontier_model_families_are_accepted(model_name: str) -> None:
 )
 def test_non_frontier_models_are_rejected(model_name: str) -> None:
     assert not is_recommended_or_frontier_model(model_name)
+
+
+@pytest.mark.parametrize(
+    "model_name",
+    [
+        "anthropic/claude-sonnet-4-6",
+        "bedrock/anthropic.claude-opus-4-8-v1:0",
+        "vertex_ai/claude-sonnet-5",
+        "Sonnet-5",
+    ],
+)
+def test_claude_routes_reject_strict_tool_schemas(model_name: str) -> None:
+    assert not supports_strict_tool_schemas(model_name)
+
+
+@pytest.mark.parametrize(
+    "model_name",
+    ["openai/gpt-5.4", "gpt-5.4", "gemini/gemini-3.1-pro-preview", "deepseek/deepseek-v4"],
+)
+def test_other_routes_keep_strict_tool_schemas(model_name: str) -> None:
+    assert supports_strict_tool_schemas(model_name)

tests/test_proxy_client.py

@@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any, cast
 
 import pytest
 
+from strix.runtime.caido_handle import CaidoBootstrapHandle
 from strix.tools.proxy import caido_api, tools
 
 
@@ -198,12 +199,31 @@ class _Ctx:
         self.context = context
 
 
-def test_ctx_client_returns_client_when_present() -> None:
+async def test_ctx_client_returns_client_when_present() -> None:
     client = _FakeClient("host")
-    got = tools._ctx_client(cast("Any", _Ctx({"caido_client": client})))
+    got = await tools._ctx_client(cast("Any", _Ctx({"caido_client": client})))
     assert got is client
 
 
-def test_ctx_client_returns_none_without_client() -> None:
-    assert tools._ctx_client(cast("Any", _Ctx({}))) is None
-    assert tools._ctx_client(cast("Any", _Ctx(None))) is None
+async def test_ctx_client_returns_none_without_client() -> None:
+    assert await tools._ctx_client(cast("Any", _Ctx({}))) is None
+    assert await tools._ctx_client(cast("Any", _Ctx(None))) is None
+
+
+async def test_ctx_client_resolves_bootstrap_handle() -> None:
+    client = _FakeClient("host")
+
+    async def _bootstrap() -> Any:
+        return client
+
+    handle = CaidoBootstrapHandle(asyncio.ensure_future(_bootstrap()))
+    got = await tools._ctx_client(cast("Any", _Ctx({"caido_client": handle})))
+    assert got is client
+
+
+async def test_ctx_client_degrades_when_bootstrap_failed() -> None:
+    async def _bootstrap() -> Any:
+        raise RuntimeError("caido never came up")
+
+    handle = CaidoBootstrapHandle(asyncio.ensure_future(_bootstrap()))
+    assert await tools._ctx_client(cast("Any", _Ctx({"caido_client": handle}))) is None

tests/test_report_coverage.py

@@ -0,0 +1,264 @@
+"""Tests for the coverage artifact assembled in strix.report.coverage."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+
+from strix.report.coverage import (
+    _SKILL_PHRASINGS,
+    build_coverage_document,
+    read_agent_graph,
+    write_coverage,
+)
+from strix.skills import get_available_skills
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _entry(**overrides: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "surface": "POST /api/orders/{id}",
+        "risk_area": "object-level authorization",
+        "outcome": "no_issue_found",
+        "evidence": "Two tenants tested; both received 403.",
+        "agent_id": "agent-1",
+        "agent_name": "authz-tester",
+        "created_at": "2026-07-02 10:00:00 UTC",
+    }
+    base.update(overrides)
+    return base
+
+
+def _graph(**overrides: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "statuses": {"agent-1": "completed"},
+        "names": {"agent-1": "authz-tester"},
+        "metadata": {"agent-1": {"skills": ["idor"], "task": "authz review"}},
+    }
+    base.update(overrides)
+    return base
+
+
+def _document(**overrides: Any) -> dict[str, Any]:
+    kwargs: dict[str, Any] = {
+        "run_record": {"run_id": "r1", "run_name": "run-1", "status": "completed"},
+        "entries": [_entry()],
+        "agent_graph": _graph(),
+        "vulnerability_reports": [],
+    }
+    kwargs.update(overrides)
+    return build_coverage_document(**kwargs)
+
+
+def test_document_reports_surfaces_and_outcomes() -> None:
+    doc = _document()
+
+    assert doc["summary"]["surfaces_reviewed"] == 1
+    assert doc["summary"]["outcomes"] == {"no_issue_found": 1}
+    assert doc["entries"][0]["outcome_label"] == "No issue identified"
+    assert doc["entries"][0]["recorded_by"] == "authz-tester"
+
+
+def test_ledger_entries_are_labelled_as_agent_reported() -> None:
+    """A reader has to be able to tell a self-report from an observation."""
+    doc = _document()
+
+    assert doc["entries"][0]["source"] == "agent_reported"
+    assert doc["machine_observed"]["source"] == "runtime"
+    assert doc["machine_observed"]["skills_exercised"] == ["idor"]
+
+
+def test_assigned_risk_skill_without_coverage_becomes_a_gap() -> None:
+    """An agent carrying the sql_injection skill that records nothing about it
+    leaves the class unexamined, not clean."""
+    doc = _document(
+        agent_graph=_graph(
+            metadata={"agent-1": {"skills": ["idor", "sql_injection"], "task": "review"}}
+        )
+    )
+
+    gaps = [gap for gap in doc["gaps"] if gap["kind"] == "unrecorded_risk_class"]
+    assert [gap["risk_area"] for gap in gaps] == ["sql injection"]
+
+
+def test_recorded_risk_class_is_not_reported_as_a_gap() -> None:
+    doc = _document(
+        entries=[_entry(risk_area="SQL injection", surface="GET /search?q=")],
+        agent_graph=_graph(metadata={"agent-1": {"skills": ["sql_injection"]}}),
+    )
+
+    assert not [gap for gap in doc["gaps"] if gap["kind"] == "unrecorded_risk_class"]
+
+
+def test_synonym_phrasing_counts_as_recorded_coverage() -> None:
+    """The ledger says "object-level authorization"; the skill is called idor."""
+    doc = _document(agent_graph=_graph(metadata={"agent-1": {"skills": ["idor"]}}))
+
+    assert not [gap for gap in doc["gaps"] if gap["kind"] == "unrecorded_risk_class"]
+
+
+def test_non_risk_skills_carry_no_coverage_obligation() -> None:
+    """Tooling skills describe how an agent works, not what it hunts."""
+    doc = _document(agent_graph=_graph(metadata={"agent-1": {"skills": ["idor", "caido"]}}))
+
+    assert not [gap for gap in doc["gaps"] if gap.get("risk_area") == "caido"]
+
+
+def test_agent_that_recorded_nothing_is_a_gap() -> None:
+    doc = _document(
+        agent_graph=_graph(
+            statuses={"agent-1": "completed", "agent-2": "completed"},
+            names={"agent-1": "authz-tester", "agent-2": "recon"},
+            metadata={},
+        )
+    )
+
+    silent = [gap for gap in doc["gaps"] if gap["kind"] == "agent_recorded_no_coverage"]
+    assert [gap["agent_name"] for gap in silent] == ["recon"]
+
+
+def test_needs_follow_up_is_carried_as_an_open_gap() -> None:
+    doc = _document(
+        entries=[_entry(outcome="needs_follow_up", evidence="Auth wall blocked testing.")]
+    )
+
+    assert doc["gaps"][0]["kind"] == "needs_follow_up"
+    assert doc["gaps"][0]["detail"] == "Auth wall blocked testing."
+
+
+def test_completed_run_with_finished_agents_is_complete() -> None:
+    doc = _document(exit_reason="finished_by_tool")
+
+    assert doc["completeness"]["complete"] is True
+    assert doc["completeness"]["caveats"] == []
+
+
+def test_budget_exhausted_run_is_not_a_complete_record() -> None:
+    """A truncated scan must not read like a clean one."""
+    doc = _document(exit_reason="budget_exhausted")
+
+    assert doc["completeness"]["complete"] is False
+    assert "budget_exhausted" in doc["completeness"]["caveats"][0]
+
+
+def test_unfinished_agent_makes_the_record_partial() -> None:
+    doc = _document(
+        agent_graph=_graph(statuses={"agent-1": "crashed"}),
+        exit_reason="finished_by_tool",
+    )
+
+    assert doc["completeness"]["complete"] is False
+    assert "authz-tester" in doc["completeness"]["caveats"][0]
+
+
+def test_failed_run_status_makes_the_record_partial() -> None:
+    doc = _document(
+        run_record={"run_id": "r1", "status": "failed"},
+        exit_reason="finished_by_tool",
+    )
+
+    assert doc["completeness"]["complete"] is False
+
+
+def test_write_coverage_emits_a_top_level_artifact(tmp_path: Path) -> None:
+    path = write_coverage(tmp_path, _document())
+
+    assert path == tmp_path / "coverage.json"
+    assert json.loads(path.read_text(encoding="utf-8"))["schema_version"] == 1
+
+
+def test_read_agent_graph_tolerates_a_missing_or_corrupt_snapshot(tmp_path: Path) -> None:
+    assert read_agent_graph(tmp_path) == {}
+
+    (tmp_path / "agents.json").write_text("{not json", encoding="utf-8")
+    assert read_agent_graph(tmp_path) == {}
+
+
+def test_read_agent_graph_loads_a_snapshot(tmp_path: Path) -> None:
+    (tmp_path / "agents.json").write_text(json.dumps(_graph()), encoding="utf-8")
+
+    assert read_agent_graph(tmp_path)["names"] == {"agent-1": "authz-tester"}
+
+
+def test_multi_token_skill_matches_how_a_pentester_writes_it() -> None:
+    """An agent carrying path_traversal_lfi_rfi records "Path Traversal".
+
+    Requiring the skill's filename verbatim published a false gap for a class
+    that had been tested and even had a finding filed against it.
+    """
+    doc = _document(
+        entries=[_entry(risk_area="Path Traversal / Directory Traversal", surface="/download")],
+        agent_graph=_graph(metadata={"agent-1": {"skills": ["path_traversal_lfi_rfi"]}}),
+    )
+
+    assert not [gap for gap in doc["gaps"] if gap["kind"] == "unrecorded_risk_class"]
+
+
+def _vulnerability_skill_names() -> set[str]:
+    return {skill["name"] for skill in get_available_skills()["vulnerabilities"]}
+
+
+def test_every_vulnerability_skill_declares_its_phrasings() -> None:
+    """A new skill without phrasings would be matched by its filename alone,
+    which is how the false gap above got published."""
+    missing = _vulnerability_skill_names() - set(_SKILL_PHRASINGS)
+
+    assert not missing, f"add ledger phrasings for: {sorted(missing)}"
+
+
+def test_declared_phrasings_name_real_skills() -> None:
+    stale = set(_SKILL_PHRASINGS) - _vulnerability_skill_names()
+
+    assert not stale, f"phrasings for skills that no longer exist: {sorted(stale)}"
+
+
+def _delegating_graph(**overrides: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "statuses": {"root": "completed", "agent-1": "completed"},
+        "names": {"root": "Root Agent", "agent-1": "authz-tester"},
+        "parent_of": {"agent-1": "root"},
+        "metadata": {"agent-1": {"skills": ["idor"]}},
+    }
+    base.update(overrides)
+    return base
+
+
+def test_delegating_root_agent_is_not_a_coverage_gap() -> None:
+    """The root delegates and reconciles; it is not a tester that went quiet.
+    Flagging it would put the same false line in every clean report."""
+    doc = _document(agent_graph=_delegating_graph())
+
+    silent = [gap for gap in doc["gaps"] if gap["kind"] == "agent_recorded_no_coverage"]
+    assert silent == []
+
+
+def test_a_subagent_that_records_nothing_is_still_a_gap() -> None:
+    doc = _document(
+        agent_graph=_delegating_graph(
+            statuses={"root": "completed", "agent-1": "completed", "agent-2": "completed"},
+            names={"root": "Root Agent", "agent-1": "authz-tester", "agent-2": "recon"},
+            parent_of={"agent-1": "root", "agent-2": "root"},
+        )
+    )
+
+    silent = [gap for gap in doc["gaps"] if gap["kind"] == "agent_recorded_no_coverage"]
+    assert [gap["agent_name"] for gap in silent] == ["recon"]
+
+
+def test_a_root_that_worked_alone_is_held_to_the_rule() -> None:
+    """With no subagents there is nobody else the testing could have come
+    from, so silence is a real gap."""
+    doc = _document(
+        entries=[],
+        agent_graph={
+            "statuses": {"root": "completed"},
+            "names": {"root": "Root Agent"},
+            "parent_of": {},
+        },
+    )
+
+    silent = [gap for gap in doc["gaps"] if gap["kind"] == "agent_recorded_no_coverage"]
+    assert [gap["agent_name"] for gap in silent] == ["Root Agent"]

tests/test_report_writer.py

@@ -179,3 +179,30 @@ def test_write_executive_report_writes_markdown(tmp_path: Path) -> None:
     content = (tmp_path / "penetration_test_report.md").read_text(encoding="utf-8")
     assert "# Security Penetration Test Report" in content
     assert "Scan complete. No critical issues." in content
+
+
+def test_render_vulnerability_md_surfaces_calibration_metadata() -> None:
+    """Confidence, the case against the finding, and retest status are part of
+    the deliverable — storing them without rendering hides the reasoning."""
+    md = render_vulnerability_md(
+        {
+            "id": "vuln-0009",
+            "title": "SSRF in URL preview",
+            "severity": "high",
+            "timestamp": "2026-07-02 10:00:00 UTC",
+            "description": "Fetches user-supplied URLs.",
+            "confidence": "medium",
+            "counterevidence": "Egress appears filtered at the network layer.",
+            "confidence_rationale": "Reproduced once out of three attempts.",
+            "severity_change_conditions": "Critical if egress filtering is removed.",
+            "remediation_steps": "Allowlist destinations.",
+            "fix_verification": "Not retested.",
+        }
+    )
+
+    assert "**Confidence:** Medium" in md
+    assert "## Counterevidence" in md
+    assert "Egress appears filtered at the network layer." in md
+    assert "## Confidence Rationale" in md
+    assert "## What Would Change This Severity" in md
+    assert "## Fix Verification" in md

tests/test_reporting_fields.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import pytest
 
@@ -75,6 +75,9 @@ async def test_create_report_persists_new_fields(report_state: ReportState) -> N
         remediation_steps="Context-encode output.",
         evidence="Response echoes the payload verbatim.",
         assumptions="Assumes a victim opens a crafted link.",
+        counterevidence="No output encoding or CSP observed on this response.",
+        confidence="HIGH",
+        severity_change_conditions="A strict CSP would lower the severity.",
         fix_effort="LOW",
         cvss_breakdown=_CVSS,
         endpoint="/search",
@@ -91,6 +94,9 @@ async def test_create_report_persists_new_fields(report_state: ReportState) -> N
     assert report["fix_effort"] == "low"
     assert report["fix_pr_body"] == "## Fix\nEncode output."
     assert report["finding_class"] == "dynamic"
+    assert report["counterevidence"] == "No output encoding or CSP observed on this response."
+    assert report["confidence"] == "high"
+    assert report["severity_change_conditions"] == "A strict CSP would lower the severity."
 
 
 async def test_create_report_requires_evidence_and_assumptions(
@@ -107,6 +113,9 @@ async def test_create_report_requires_evidence_and_assumptions(
         remediation_steps="r",
         evidence="   ",
         assumptions="",
+        counterevidence="none found",
+        confidence="high",
+        severity_change_conditions="n/a",
         fix_effort="low",
         cvss_breakdown=_CVSS,
         endpoint=None,
@@ -134,6 +143,9 @@ async def test_create_report_rejects_invalid_fix_effort(report_state: ReportStat
         remediation_steps="r",
         evidence="e",
         assumptions="a",
+        counterevidence="none found",
+        confidence="high",
+        severity_change_conditions="n/a",
         fix_effort="enormous",
         cvss_breakdown=_CVSS,
         endpoint=None,
@@ -147,6 +159,80 @@ async def test_create_report_rejects_invalid_fix_effort(report_state: ReportStat
     assert not report_state.vulnerability_reports
 
 
+async def _create_with(report_state: ReportState, **overrides: object) -> dict[str, Any]:
+    kwargs: dict[str, object] = {
+        "title": "X",
+        "description": "d",
+        "impact": "i",
+        "target": "t",
+        "technical_analysis": "ta",
+        "poc_description": "p",
+        "poc_script_code": "c",
+        "remediation_steps": "r",
+        "evidence": "e",
+        "assumptions": "a",
+        "counterevidence": "No guard found on this path.",
+        "confidence": "high",
+        "severity_change_conditions": "Proof of internet exposure would raise it.",
+        "fix_effort": "low",
+        "cvss_breakdown": _CVSS,
+        "endpoint": None,
+        "method": None,
+        "cve": None,
+        "cwe": None,
+        "code_locations": None,
+    }
+    kwargs.update(overrides)
+    assert report_state is not None
+    return await _do_create(**kwargs)  # type: ignore[arg-type]
+
+
+async def test_create_report_requires_counterevidence(report_state: ReportState) -> None:
+    result = await _create_with(report_state, counterevidence="   ")
+    assert result["success"] is False
+    assert any("Counterevidence" in e for e in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+async def test_create_report_requires_severity_change_conditions(
+    report_state: ReportState,
+) -> None:
+    result = await _create_with(report_state, severity_change_conditions="")
+    assert result["success"] is False
+    assert any("severity_change_conditions" in e for e in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+async def test_create_report_rejects_invalid_confidence(report_state: ReportState) -> None:
+    result = await _create_with(report_state, confidence="pretty sure")
+    assert result["success"] is False
+    assert any("confidence" in e for e in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+async def test_create_report_requires_rationale_when_confidence_not_high(
+    report_state: ReportState,
+) -> None:
+    result = await _create_with(report_state, confidence="medium")
+    assert result["success"] is False
+    assert any("confidence_rationale" in e for e in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+async def test_create_report_accepts_medium_confidence_with_rationale(
+    report_state: ReportState,
+) -> None:
+    result = await _create_with(
+        report_state,
+        confidence="medium",
+        confidence_rationale="Static-only trace; could not stand up the service.",
+    )
+    assert result["success"] is True
+    report = report_state.vulnerability_reports[0]
+    assert report["confidence"] == "medium"
+    assert report["confidence_rationale"] == "Static-only trace; could not stand up the service."
+
+
 async def test_dependency_report_sets_class_and_metadata(report_state: ReportState) -> None:
     result = await _do_create_dependency(
         title="CVE-2021-23337 in lodash 4.17.20",
@@ -937,6 +1023,56 @@ def test_vuln_tool_exposes_new_params() -> None:
     assert "advisory_cvss" in dep_required
 
 
+_FIX_LOCATION = {
+    "file": "app/views.py",
+    "start_line": 10,
+    "end_line": 12,
+    "fix_before": 'query = f"SELECT * FROM t WHERE id={uid}"',
+    "fix_after": 'query = "SELECT * FROM t WHERE id=%s"',
+}
+
+_INFO_LOCATION = {
+    "file": "app/views.py",
+    "start_line": 10,
+    "end_line": 12,
+    "snippet": 'query = f"SELECT * FROM t WHERE id={uid}"',
+}
+
+
+async def test_fix_after_requires_verification(report_state: ReportState) -> None:
+    result = await _create_with(report_state, code_locations=[_FIX_LOCATION])
+    assert result["success"] is False
+    assert any("fix_verification" in e for e in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+async def test_fix_after_with_verification_persists(report_state: ReportState) -> None:
+    verification = (
+        "Re-ran the PoC against the patched handler: the payload is now bound as a "
+        "parameter and returns no extra rows. Checked the two sibling call sites of "
+        "the same helper and the admin export path; both already parameterized. "
+        "Legitimate numeric ids still resolve and the 404 path is unchanged. "
+        "Ran the focused view tests and ruff."
+    )
+    result = await _create_with(
+        report_state,
+        code_locations=[_FIX_LOCATION],
+        fix_verification=verification,
+    )
+    assert result["success"] is True
+    assert report_state.vulnerability_reports[0]["fix_verification"] == verification
+
+
+async def test_informational_location_needs_no_verification(report_state: ReportState) -> None:
+    result = await _create_with(report_state, code_locations=[_INFO_LOCATION])
+    assert result["success"] is True
+    assert "fix_verification" not in report_state.vulnerability_reports[0]
+
+
+def test_vuln_tool_exposes_fix_verification() -> None:
+    assert "fix_verification" in create_vulnerability_report.params_json_schema["properties"]
+
+
 def test_dep_tool_exposes_contextual_cvss_params() -> None:
     dep_props = create_dependency_report.params_json_schema["properties"]
     for field in (

tests/test_sarif.py

@@ -242,3 +242,132 @@ def test_write_sarif_replaces_atomically_no_partial_on_reemit(tmp_path: Path) ->
     assert leftovers == []
     # And it parses as a complete document with both findings.
     assert len(_read(tmp_path)["runs"][0]["results"]) == 2
+
+
+def _coverage(*entries: dict[str, Any], **overrides: Any) -> dict[str, Any]:
+    doc: dict[str, Any] = {
+        "entries": list(entries),
+        "completeness": {"complete": True, "caveats": []},
+    }
+    doc.update(overrides)
+    return doc
+
+
+def _coverage_entry(**overrides: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "surface": "POST /api/orders/{id}",
+        "risk_area": "SQL injection",
+        "outcome": "no_issue_found",
+        "outcome_label": "No issue identified",
+        "evidence": "14 parameters fuzzed; all queries parameterized.",
+        "recorded_by": "injection-tester",
+        "source": "agent_reported",
+    }
+    base.update(overrides)
+    return base
+
+
+def test_cleared_surface_becomes_a_passing_result(tmp_path: Path) -> None:
+    """ "Tested and clean" is a SARIF pass, not an absent result."""
+    write_sarif(tmp_path, [], coverage=_coverage(_coverage_entry()))
+    results = _read(tmp_path)["runs"][0]["results"]
+
+    assert len(results) == 1
+    assert results[0]["kind"] == "pass"
+    # SARIF requires level "none" on any result that is not a failure.
+    assert results[0]["level"] == "none"
+    assert "14 parameters fuzzed" in results[0]["message"]["text"]
+
+
+def test_coverage_outcomes_map_to_their_sarif_kinds(tmp_path: Path) -> None:
+    write_sarif(
+        tmp_path,
+        [],
+        coverage=_coverage(
+            _coverage_entry(outcome="ruled_out", risk_area="XSS"),
+            _coverage_entry(outcome="not_applicable", risk_area="XXE"),
+            _coverage_entry(outcome="needs_follow_up", risk_area="SSRF"),
+        ),
+    )
+    kinds = [result["kind"] for result in _read(tmp_path)["runs"][0]["results"]]
+
+    assert kinds == ["pass", "notApplicable", "open"]
+
+
+def test_reported_coverage_is_not_duplicated_as_a_pass(tmp_path: Path) -> None:
+    """A surface that produced a finding is already in results as a failure."""
+    write_sarif(
+        tmp_path,
+        [_finding()],
+        coverage=_coverage(_coverage_entry(outcome="reported")),
+    )
+    results = _read(tmp_path)["runs"][0]["results"]
+
+    assert len(results) == 1
+    assert results[0].get("kind", "fail") == "fail"
+
+
+def test_coverage_results_declare_their_own_rules(tmp_path: Path) -> None:
+    write_sarif(
+        tmp_path,
+        [_finding()],
+        coverage=_coverage(
+            _coverage_entry(risk_area="SQL injection"),
+            _coverage_entry(risk_area="SQL injection", surface="GET /search"),
+        ),
+    )
+    run = _read(tmp_path)["runs"][0]
+    rules = run["tool"]["driver"]["rules"]
+    coverage_rules = [rule for rule in rules if rule["id"].startswith("strix-coverage/")]
+
+    # Both entries share one rule, and every result's ruleIndex resolves to it.
+    assert len(coverage_rules) == 1
+    assert coverage_rules[0]["defaultConfiguration"]["level"] == "none"
+    for result in run["results"]:
+        assert rules[result["ruleIndex"]]["id"] == result["ruleId"]
+
+
+def test_incomplete_run_is_flagged_on_the_invocation(tmp_path: Path) -> None:
+    """A scan cut short must not be indistinguishable from a clean one."""
+    write_sarif(
+        tmp_path,
+        [],
+        coverage=_coverage(
+            _coverage_entry(),
+            completeness={"complete": False, "caveats": ["Budget exhausted."]},
+        ),
+    )
+    invocation = _read(tmp_path)["runs"][0]["invocations"][0]
+
+    assert invocation["executionSuccessful"] is False
+    assert invocation["toolExecutionNotifications"][0]["message"]["text"] == "Budget exhausted."
+
+
+def test_complete_run_reports_a_successful_invocation(tmp_path: Path) -> None:
+    write_sarif(tmp_path, [], coverage=_coverage(_coverage_entry()))
+    invocation = _read(tmp_path)["runs"][0]["invocations"][0]
+
+    assert invocation["executionSuccessful"] is True
+    assert "toolExecutionNotifications" not in invocation
+
+
+def test_calibration_metadata_survives_into_result_properties(tmp_path: Path) -> None:
+    write_sarif(
+        tmp_path,
+        [
+            _finding(
+                confidence="medium",
+                counterevidence="WAF blocks the naive payload.",
+                confidence_rationale="Reproduced once out of three attempts.",
+                severity_change_conditions="Critical if the WAF rule is removed.",
+                fix_verification="Not retested.",
+            )
+        ],
+    )
+    strix = _read(tmp_path)["runs"][0]["results"][0]["properties"]["strix"]
+
+    assert strix["confidence"] == "medium"
+    assert strix["counterevidence"] == "WAF blocks the naive payload."
+    assert strix["confidence_rationale"] == "Reproduced once out of three attempts."
+    assert strix["severity_change_conditions"] == "Critical if the WAF rule is removed."
+    assert strix["fix_verification"] == "Not retested."

tests/test_skill_dir_extension.py

@@ -4,7 +4,7 @@ from pathlib import Path
 import pytest
 
 import strix.skills as skills_mod
-from strix.agents.prompt import render_system_prompt
+from strix.agents.prompt import _resolve_skills, render_system_prompt
 from strix.skills import (
     get_all_skill_names,
     get_available_skills,
@@ -232,3 +232,42 @@ def test_builtin_skill_still_loads_when_not_overridden(tmp_path: Path) -> None:
 def test_missing_skill_is_skipped(tmp_path: Path) -> None:
     register_skill_dir(tmp_path)
     assert load_skills(["does_not_exist"]) == {}
+
+
+def test_resolve_skills_always_includes_analysis_baseline() -> None:
+    resolved = _resolve_skills(requested=None)
+
+    assert "analysis/counterevidence" in resolved
+    assert "analysis/severity_calibration" in resolved
+
+
+def test_resolve_skills_adds_diff_mode_only_when_diff_scoped() -> None:
+    assert "scan_modes/diff" not in _resolve_skills(requested=None)
+    diff_scoped = _resolve_skills(requested=None, is_diff_scoped=True)
+    assert "scan_modes/diff" in diff_scoped
+    # Diff scope overlays the depth mode rather than replacing it.
+    assert "scan_modes/deep" in diff_scoped
+
+
+def test_resolve_skills_gates_source_aware_skills_on_whitebox() -> None:
+    blackbox = _resolve_skills(requested=None)
+    assert "analysis/fix_verification" not in blackbox
+    assert "analysis/source_aware_discovery" not in blackbox
+
+    whitebox = _resolve_skills(requested=None, is_whitebox=True)
+    assert "analysis/fix_verification" in whitebox
+    assert "analysis/source_aware_discovery" in whitebox
+
+
+def test_new_skill_files_load() -> None:
+    names = [
+        "analysis/counterevidence",
+        "analysis/severity_calibration",
+        "analysis/fix_verification",
+        "analysis/source_aware_discovery",
+        "scan_modes/diff",
+    ]
+    loaded = load_skills(names)
+    for name in names:
+        key = name.split("/")[-1]
+        assert loaded.get(key), f"{name} failed to load"

tests/test_state_coverage_artifact.py

@@ -0,0 +1,66 @@
+"""coverage.json is a deliverable artifact, not runtime state."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+
+from strix.core.paths import runtime_state_dir
+from strix.report.state import ReportState
+from strix.tools.coverage.tools import _record_impl, hydrate_coverage_from_disk
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+@pytest.fixture
+def state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
+    monkeypatch.chdir(tmp_path)
+    report_state = ReportState(run_name="run-1")
+    hydrate_coverage_from_disk(runtime_state_dir(report_state.get_run_dir()))
+    return report_state
+
+
+def _record_a_cleared_surface() -> None:
+    _record_impl(
+        surface="POST /api/orders/{id}",
+        risk_area="SQL injection",
+        outcome="no_issue_found",
+        evidence="14 parameters fuzzed; every query parameterized.",
+        agent_id="agent-1",
+        agent_name="injection-tester",
+    )
+
+
+def test_coverage_is_written_beside_the_other_artifacts(state: ReportState) -> None:
+    _record_a_cleared_surface()
+
+    state._save_artifacts()
+
+    document = json.loads((state.get_run_dir() / "coverage.json").read_text(encoding="utf-8"))
+    assert document["entries"][0]["risk_area"] == "SQL injection"
+    assert document["summary"]["surfaces_reviewed"] == 1
+
+
+def test_cleared_surfaces_reach_sarif(state: ReportState) -> None:
+    _record_a_cleared_surface()
+
+    state._save_artifacts()
+
+    sarif = json.loads((state.get_run_dir() / "findings.sarif").read_text(encoding="utf-8"))
+    results = sarif["runs"][0]["results"]
+    assert [result["kind"] for result in results] == ["pass"]
+
+
+def test_artifacts_still_land_when_coverage_is_empty(state: ReportState) -> None:
+    state.final_scan_result = "Scan complete."
+
+    state._save_artifacts()
+
+    run_dir = state.get_run_dir()
+    assert (run_dir / "penetration_test_report.md").is_file()
+    document = json.loads((run_dir / "coverage.json").read_text(encoding="utf-8"))
+    assert document["entries"] == []

tests/test_threat_model_tool.py

@@ -0,0 +1,343 @@
+"""Tests for the target-scoped threat model cache."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+
+import pytest
+
+from strix.agents.factory import _BASE_TOOLS
+from strix.tools.threat_model import tools as threat_model_tools
+from strix.tools.threat_model.tools import (
+    _amend_impl,
+    _get_impl,
+    _save_impl,
+    amend_threat_model,
+    get_threat_model,
+    save_threat_model,
+)
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+_MODEL = """# Threat Model
+
+## Overview
+A multi-tenant billing API. Product code lives in `api/`; `scripts/` is
+developer-only tooling and is not deployed.
+
+## Trust Boundaries and Assumptions
+Requests arrive from untrusted tenants through `api/router.py`. The tenant id
+is taken from the signed session, never from the request body. Operators
+configure webhooks; developers control migrations.
+
+## Attack Surface and Attacker Stories
+The public REST surface and the webhook receiver are attacker-reachable. A
+realistic story is a tenant reading another tenant's invoices. Local CLI
+tooling is not a realistic surface.
+
+## Severity Calibration
+Critical: cross-tenant write. High: cross-tenant read. Medium: authenticated
+self-scoped information leak. Low: verbose errors.
+"""
+
+
+def _git(repo: Path, *args: str) -> None:
+    subprocess.run(["/usr/bin/env", "git", *args], cwd=repo, check=True)  # noqa: S603
+
+
+def _make_repo(tmp_path: Path, name: str = "repo") -> Path:
+    repo = tmp_path / name
+    repo.mkdir(parents=True)
+    _git(repo, "init", "-q")
+    _git(repo, "config", "user.email", "t@example.com")
+    _git(repo, "config", "user.name", "t")
+    (repo / "README.md").write_text("hi\n", encoding="utf-8")
+    _git(repo, "add", "README.md")
+    _git(repo, "commit", "-qm", "init")
+    return repo
+
+
+@pytest.fixture(autouse=True)
+def _isolated_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(threat_model_tools, "_CACHE_DIR", tmp_path / "cache")
+
+
+def test_missing_model_reports_not_found(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+
+    result = _get_impl(str(repo))
+
+    assert result["success"] is True
+    assert result["found"] is False
+    assert "save_threat_model" in result["message"]
+
+
+def test_saved_model_round_trips(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+
+    assert _save_impl(str(repo), _MODEL, "Strix")["success"] is True
+    result = _get_impl(str(repo))
+
+    assert result["found"] is True
+    assert result["stale"] is False
+    assert "multi-tenant billing API" in result["content"]
+
+
+def test_model_is_stale_after_new_revision(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    _save_impl(str(repo), _MODEL, None)
+
+    (repo / "next.py").write_text("x = 1\n", encoding="utf-8")
+    _git(repo, "add", "next.py")
+    _git(repo, "commit", "-qm", "next")
+
+    result = _get_impl(str(repo))
+
+    assert result["found"] is True
+    assert result["stale"] is True
+    assert result["content"]
+
+
+def test_cache_is_keyed_per_repository(tmp_path: Path) -> None:
+    first = _make_repo(tmp_path, "first")
+    second = _make_repo(tmp_path, "second")
+    _save_impl(str(first), _MODEL, None)
+
+    assert _get_impl(str(second))["found"] is False
+
+
+def test_rejects_model_missing_required_sections(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    thin = _MODEL.replace("## Severity Calibration", "## Notes")
+
+    result = _save_impl(str(repo), thin, None)
+
+    assert result["success"] is False
+    assert "severity calibration" in result["error"]
+
+
+def test_rejects_stub_model(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+
+    result = _save_impl(str(repo), "overview trust boundaries attack surface", None)
+
+    assert result["success"] is False
+    assert "too thin" in result["error"]
+
+
+def test_rejects_empty_target() -> None:
+    result = _get_impl("   ")
+    assert result["success"] is False
+    assert "target cannot be empty" in result["error"]
+
+
+def test_tools_are_registered() -> None:
+    assert get_threat_model in _BASE_TOOLS
+    assert save_threat_model in _BASE_TOOLS
+
+
+_ADDENDUM = (
+    "The base model calls the webhook receiver operator-controlled. It is "
+    "unauthenticated in `api/webhooks.py:31`, so treat its body as attacker-controlled."
+)
+
+
+def test_amendment_is_returned_with_the_model(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    _save_impl(str(repo), _MODEL, "root")
+
+    assert _amend_impl(str(repo), _ADDENDUM, "webhook-agent")["success"] is True
+    result = _get_impl(str(repo))
+
+    assert result["content"] == _MODEL.strip()
+    assert [a["content"] for a in result["amendments"]] == [_ADDENDUM]
+    assert result["amendments"][0]["by"] == "webhook-agent"
+
+
+def test_amendments_accumulate_without_overwriting(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    _save_impl(str(repo), _MODEL, "root")
+
+    _amend_impl(str(repo), _ADDENDUM, "agent-a")
+    second = "The `scripts/` directory ships in the container image; it is not dev-only."
+    _amend_impl(str(repo), second + " See `Dockerfile:14`.", "agent-b")
+
+    amendments = _get_impl(str(repo))["amendments"]
+    assert len(amendments) == 2
+    assert [a["by"] for a in amendments] == ["agent-a", "agent-b"]
+
+
+def test_amend_requires_an_existing_model(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+
+    result = _amend_impl(str(repo), _ADDENDUM, None)
+
+    assert result["success"] is False
+    assert "save_threat_model" in result["error"]
+
+
+def test_amend_rejects_a_stub(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    _save_impl(str(repo), _MODEL, "root")
+
+    assert _amend_impl(str(repo), "looks wrong", None)["success"] is False
+
+
+def test_save_clears_amendments_and_says_so(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    _save_impl(str(repo), _MODEL, "root")
+    _amend_impl(str(repo), _ADDENDUM, "agent-a")
+
+    result = _save_impl(str(repo), _MODEL.replace("billing API", "billing service"), "root")
+
+    assert result["amendments_cleared"] == 1
+    assert "cleared" in result["message"]
+    assert "amendments" not in _get_impl(str(repo))
+
+
+def test_amend_tool_is_registered() -> None:
+    assert amend_threat_model in _BASE_TOOLS
+
+
+_BLACKBOX_MODEL = _MODEL.replace(
+    "Product code lives in `api/`; `scripts/` is\ndeveloper-only tooling and is not deployed.",
+    "Only the deployed surface is visible; no source. Inferred from recon.",
+)
+
+
+def test_blackbox_target_round_trips() -> None:
+    target = "https://app.example.com"
+
+    assert _save_impl(target, _BLACKBOX_MODEL, "recon")["success"] is True
+    result = _get_impl(target)
+
+    assert result["found"] is True
+    assert result["stale"] is False, "a fresh model with no revision is not stale"
+    assert result["revision"] == "unversioned"
+    assert "Inferred from recon" in result["content"]
+
+
+def test_blackbox_target_spellings_share_one_model() -> None:
+    _save_impl("https://App.Example.com:443/", _BLACKBOX_MODEL, "recon")
+
+    for spelling in ("https://app.example.com", "app.example.com", "https://app.example.com/"):
+        assert _get_impl(spelling)["found"] is True, spelling
+
+    assert _get_impl("https://other.example.com")["found"] is False
+
+
+def test_blackbox_model_goes_stale_with_age() -> None:
+    target = "https://app.example.com"
+    _save_impl(target, _BLACKBOX_MODEL, "recon")
+
+    aged = (datetime.now(UTC) - timedelta(days=threat_model_tools._MAX_AGE_DAYS + 1)).isoformat()
+    path = threat_model_tools._cache_path("app.example.com:443")
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    payload["created_at"] = aged
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+    result = _get_impl(target)
+
+    assert result["stale"] is True
+    assert "re-confirm" in result["message"]
+
+
+def test_blackbox_target_can_be_amended() -> None:
+    target = "https://app.example.com"
+    _save_impl(target, _BLACKBOX_MODEL, "recon")
+
+    addendum = (
+        "The model infers /admin is IP-restricted. It is reachable with any "
+        "authenticated session; the restriction is only on /admin/settings."
+    )
+    assert _amend_impl(target, addendum, "authz-agent")["success"] is True
+    assert _get_impl(target)["amendments"][0]["content"] == addendum
+
+
+def test_checkout_and_its_remote_are_the_same_target(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    _git(repo, "remote", "add", "origin", "https://github.com/acme/billing.git")
+    _save_impl(str(repo), _MODEL, "root")
+
+    clone = _make_repo(tmp_path, "clone")
+    _git(clone, "remote", "add", "origin", "https://github.com/acme/billing.git")
+
+    assert _get_impl(str(clone))["found"] is True
+
+
+def test_path_on_a_known_host_resolves_to_the_scan_target() -> None:
+    scan_targets = ["https://app.example.com"]
+    _save_impl("https://app.example.com", _BLACKBOX_MODEL, "root", scan_targets)
+
+    # An agent testing one page names that page, not the scan's target string.
+    assert _get_impl("https://app.example.com/admin/login", scan_targets)["found"] is True
+
+
+def test_two_scan_targets_on_one_host_stay_separate() -> None:
+    scan_targets = ["https://example.com/tenant-a", "https://example.com/tenant-b"]
+    _save_impl("https://example.com/tenant-a", _BLACKBOX_MODEL, "root", scan_targets)
+
+    assert _get_impl("https://example.com/tenant-b", scan_targets)["found"] is False
+
+
+def test_unknown_host_is_not_snapped_onto_the_scan_target() -> None:
+    scan_targets = ["https://app.example.com"]
+    _save_impl("https://app.example.com", _BLACKBOX_MODEL, "root", scan_targets)
+
+    assert _get_impl("https://unrelated.test", scan_targets)["found"] is False
+
+
+def test_empty_target_falls_back_to_a_single_scan_target() -> None:
+    scan_targets = ["https://app.example.com"]
+    _save_impl("", _BLACKBOX_MODEL, "root", scan_targets)
+
+    assert _get_impl("", scan_targets)["found"] is True
+    assert _get_impl("https://app.example.com")["found"] is True
+
+
+def test_repository_subdirectory_shares_the_repository_model(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    (repo / "src").mkdir()
+    _save_impl(str(repo), _MODEL, "root")
+
+    assert _get_impl(str(repo / "src"))["found"] is True
+
+
+def test_checkout_and_its_clone_url_are_one_identity(tmp_path: Path) -> None:
+    """The model an agent saves inside the checkout must be visible to an agent
+    that names the same repository by the URL it was cloned from."""
+    repo = _make_repo(tmp_path)
+    _git(repo, "remote", "add", "origin", "https://github.com/acme/billing.git")
+    _save_impl(str(repo), _MODEL, "root")
+
+    assert _get_impl("https://github.com/acme/billing")["found"] is True
+    assert _get_impl("https://github.com/acme/billing.git")["found"] is True
+
+
+def test_ssh_and_https_remotes_are_one_identity(tmp_path: Path) -> None:
+    """One repository cloned over scp-style SSH and over HTTPS is one target."""
+    over_ssh = _make_repo(tmp_path, "ssh-clone")
+    _git(over_ssh, "remote", "add", "origin", "git@github.com:acme/billing.git")
+    _save_impl(str(over_ssh), _MODEL, "root")
+
+    over_https = _make_repo(tmp_path, "https-clone")
+    _git(over_https, "remote", "add", "origin", "https://github.com/acme/billing.git")
+
+    assert _get_impl(str(over_https))["found"] is True
+
+
+def test_different_repositories_on_one_host_stay_separate(tmp_path: Path) -> None:
+    first = _make_repo(tmp_path, "billing")
+    _git(first, "remote", "add", "origin", "git@github.com:acme/billing.git")
+    _save_impl(str(first), _MODEL, "root")
+
+    second = _make_repo(tmp_path, "payments")
+    _git(second, "remote", "add", "origin", "git@github.com:acme/payments.git")
+
+    assert _get_impl(str(second))["found"] is False

tests/test_tui_backend_server.py

@@ -13,7 +13,7 @@ from agents.tool import ToolOutputImage
 
 from strix.config.settings import DEFAULT_MAX_TURNS
 from strix.interface.tui.backend.controller import TuiController
-from strix.interface.tui.backend.projection import terminal_projection
+from strix.interface.tui.backend.projection import bounded_state_projection, terminal_projection
 from strix.interface.tui.backend.protocol import (
     MAX_COMMAND_BYTES,
     PROTOCOL_CAPABILITIES,
@@ -215,7 +215,11 @@ def test_unicode_heavy_setup_state_stays_within_control_frame_limit() -> None:
         "Any",
         SimpleNamespace(
             caido_url="https://例え.example/" + "道" * 10_000,
-            get_total_llm_usage=lambda: {f"model-{index}": "費" * 10_000 for index in range(20)},
+            get_total_llm_usage=lambda: {
+                "total_tokens": 720_400,
+                "cost": 20.0,
+                **{f"model-{index}": "🔒" * 10_000 for index in range(20)},
+            },
         ),
     )
     server = TuiBackendServer(controller)
@@ -226,6 +230,26 @@ def test_unicode_heavy_setup_state_stays_within_control_frame_limit() -> None:
     assert len(encoded) <= MAX_COMMAND_BYTES
     assert "🔒".encode() in encoded
     assert snapshot["projection_truncated"] is True
+    assert snapshot["usage"] == {"total_tokens": 720_400, "cost": 20.0}
+
+
+def test_defensive_state_projection_preserves_usage_summary() -> None:
+    controller = TuiController(args())
+    controller.report_state = cast(
+        "Any",
+        SimpleNamespace(
+            caido_url=None,
+            get_total_llm_usage=lambda: {"total_tokens": 720_400, "cost": 20.0},
+        ),
+    )
+    state = controller.snapshot()
+    state["provider"] = None
+    state["future_oversized_field"] = "x" * 100_000
+
+    snapshot = bounded_state_projection(state)
+
+    assert snapshot["projection_truncated"] is True
+    assert snapshot["usage"] == {"total_tokens": 720_400, "cost": 20.0}
 
 
 @pytest.mark.asyncio