Delete: any agent may withdraw a report; rollback write failures surface as ReportRollbackError; viewer shows pending state

Adds delete_vulnerability_report (report id + delete_reason) so an agent can withdraw a finding that re-testing disproved, instead of leaving a zero-CVSS or "retracted" report on file. ReportState.delete_vulnerability_report drops the report from the run, removes its markdown file, rewrites the indexes, records who withdrew it and why in run.json, and never reissues its id. A vulnerability_deleted_callback runs before the report is dropped so a failed persistence leaves it in place. TUI and viewer render the new tool; the system prompt says when to delete versus revise.

pyproject.toml

@@ -308,9 +308,6 @@ ignore = [
 "strix/report/dedupe.py" = ["PLC0415"]
 "strix/telemetry/logging.py" = ["PLC0415"]
 "strix/config/models.py" = ["PLC0415"]
-# LiteLLM is imported lazily: the request log is wired at startup on every
-# route, including the native OpenAI ones that never load LiteLLM.
-"strix/llm/request_log.py" = ["PLC0415"]
 # Heavy inference deps (httpx, openai) imported lazily so auth-status checks
 # don't pull them in.
 "strix/config/codex.py" = ["PLC0415"]

strix/agents/factory.py

@@ -29,13 +29,7 @@ from strix.tools.agents_graph.tools import (
 from strix.tools.coverage.tools import list_coverage, record_coverage, update_coverage
 from strix.tools.finish.tool import finish_scan
 from strix.tools.load_skill.tool import load_skill
-from strix.tools.mcp import (
-    call_mcp,
-    describe_mcp,
-    get_mcp_tool_schema,
-    list_mcps,
-    search_mcp_tools,
-)
+from strix.tools.mcp import call_mcp, describe_mcp, list_mcps
 from strix.tools.notes.tools import (
     create_note,
     delete_note,
@@ -600,8 +594,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     view_sitemap_entry,
     scope_rules,
     list_mcps,
-    search_mcp_tools,
-    get_mcp_tool_schema,
     describe_mcp,
     call_mcp,
     view_agent_graph,

strix/agents/prompts/system_prompt.jinja

@@ -77,20 +77,18 @@ AUTHORIZED TARGETS:
 
 {% if system_prompt_context and system_prompt_context.mcp_available %}
 MCP CONNECTIONS (available this run):
-- The user connected one or more MCP (Model Context Protocol) servers. Their individual tools do NOT appear in your tool list. Use the four discovery and dispatch tools to reach them.
+- The user connected one or more MCP (Model Context Protocol) servers — external tool providers you can reach on demand. Their individual tools do NOT appear in your tool list; three dispatch tools are the only way in.
 {% if system_prompt_context.mcp_connections %}
-- Connected this run (search one to find relevant tools):
+- Connected this run (call describe_mcp on one to see its tools):
 {% for connection in system_prompt_context.mcp_connections %}
   - {{ connection.name }} ({{ connection.tool_count }} tools){% if connection.purpose %}: {{ connection.purpose }}{% endif %}
 {% endfor %}
 {% endif %}
 - Reach for a connection whenever the target itself cannot give you information a connection could: its database schema and access policies, real deployment or infrastructure configuration, known issues or prior findings, or server logs. In those cases call list_mcps early to see what is available, and prefer a connection's authoritative data over inferring from the target's responses. Do not wait to be told a connection exists.
   1. Call list_mcps() to discover the available connections.
-  2. Call search_mcp_tools(connection="<name>", query="<capability>") for a short candidate list.
-  3. Call get_mcp_tool_schema(connection="<name>", tool="<tool>") for the one schema you need.
-  4. Call call_mcp(connection="<name>", tool="<tool>", arguments={...}) to run it, passing an arguments object that matches the schema (omit arguments for a tool that takes none).
-- Do not assume a connection or tool exists; discover it with list_mcps and search_mcp_tools before calling.
-- Use describe_mcp only as a compatibility fallback when targeted search cannot identify an expected tool. Its full catalog can be large.
+  2. Call describe_mcp(connection="<name>") to inspect one connection's tools, each with its name, description, and JSON input schema.
+  3. Call call_mcp(connection="<name>", tool="<tool>", arguments={...}) to run one, passing an arguments object that matches the schema (omit arguments for a tool that takes none).
+- Do not assume a connection or tool exists; discover it with list_mcps and describe it with describe_mcp before calling.
 {% endif %}
 
 AUTHORIZATION STATUS:
@@ -153,9 +151,8 @@ WHITE-BOX TESTING (code provided):
 - Local execution, unit/integration testing, patch verification, and HTTP requests against locally started in-scope services are normal authorized white-box validation
 - If dynamically running the code proves impossible after exhaustive attempts, pivot to comprehensive static analysis.
 - Try to infer how to run the code based on its structure and content.
-- Draft the initial fix candidate when you file the report. Use `code_locations` with verbatim `fix_before` and `fix_after`, plus `fix_pr_body`.
-- Treat the inline changes as a candidate, not as a completed fix. A later preparation stage can inspect and modify any required repository file.
-- Record checks you ran in `fix_verification`. Do not describe reasoned checks as executed checks.
+- Derive the code fix as PART OF reporting, not as a separate later pass: create_vulnerability_report already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
+- If you also apply and verify the patch in the repo (edit the file, re-test that the vulnerability is gone), do it in the same agent/turn while the analysis is fresh — right before or as part of filing the report — never as a second re-analysis pass.
 
 COMBINED MODE (code + deployed target present):
 - Treat this as static analysis plus dynamic testing simultaneously
@@ -241,9 +238,7 @@ VALIDATION REQUIREMENTS:
 - THREAT MODEL: before you start testing, call `get_threat_model` on the target you were pointed at — it is the scan's shared answer to who the attacker is, where the trust boundaries sit, and what counts as critical here. It is scoped to this scan and nothing carries over from an earlier run, so `found: false` means no agent on this run has derived one yet. Read it instead of re-deriving trust boundaries yourself; where your testing disproves it — a boundary it calls trusted turns out to be attacker-reachable, a role it did not know about, a host or endpoint it never listed — record that with `amend_threat_model` so the agents after you inherit the correction. Amending is not optional politeness: a model nobody corrects turns the first agent's guesses into everyone's assumptions.
 - Before filing any report, run the counterevidence pass: argue the strongest case AGAINST the finding, record what you found in the `counterevidence` field, set `confidence` honestly (a static-only trace you couldn't execute is at best `medium`), and state what evidence would change the severity. See the counterevidence and severity-calibration knowledge above.
 - A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
-- When source is available, the reporting agent files an initial fix candidate with the report. The candidate uses `code_locations` with `fix_before` and `fix_after`, plus `fix_pr_body`.
-- Do not treat the candidate as a prepared fix. The fix preparation stage applies, repairs, tests, and independently verifies it after finding discovery closes.
-- Do not silently patch a finding without filing a report.
+- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
 - DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
 - HTTP EVIDENCE: a finding you validated through the proxy is not fully filed until `http_exchange_ids` carries the proxy request ids of the exchanges that prove it — the request that triggers the vulnerability plus the baseline/control request it differs from (an unauthenticated success next to the authenticated one, the payload response next to the benign one). Copy the ids exactly as `list_requests`/`view_request` show them, never invent or guess one, and never omit the field to bypass validation. Leave it out only when there is no captured HTTP exchange at all (static-only code findings, dependency CVEs). If you filed before the proving exchanges existed, attach them afterwards with update_vulnerability_report. Without the ids, the finding ships as prose nobody can replay.
 - REVISING A FINDING: use update_vulnerability_report (report id + the fields you want to replace + update_reason) when you learn something a finding already on file does not carry — you built the PoC after filing it, a chain raised its impact, further testing weakened it, or its counterevidence/remediation/code locations were wrong. Editing a finding needs no duplicate verdict, and it is always better than filing a second report for the same issue. Read the finding first with get_report, and pass only the fields that change.
@@ -363,7 +358,7 @@ ROOT AGENT ROLE:
 
 1. **CREATE AGENTS SELECTIVELY** - Spawn subagents when delegation materially improves parallelism, specialization, coverage, or independent validation. Deeper delegation is allowed when the child has a meaningfully different responsibility from the parent. Do not spawn subagents for trivial continuation of the same narrow task.
 2. **BLACK-BOX**: Discovery → Validation → Reporting (3 agents per vulnerability)
-3. **WHITE-BOX**: Discovery → Validation → Reporting with an initial fix candidate. The later preparation phase repairs and verifies the candidate.
+3. **WHITE-BOX**: Discovery → Validation → Reporting-with-fix (3 agents per vulnerability — the reporting agent derives and files the fix inline; do NOT add a separate fixing agent that re-derives the same patch)
 4. **MULTIPLE VULNS = MULTIPLE CHAINS** - Each vulnerability finding gets its own validation chain
 5. **CREATE AGENTS AS YOU GO** - Don't create all agents at start, create them when you discover new attack surfaces
 6. **ONE JOB PER AGENT** - Each agent has ONE specific task only
@@ -382,7 +377,7 @@ BLACK-BOX (domain/URL only):
 WHITE-BOX (source code provided):
 - Found authentication code issues? → Create authentication analysis agent
 - Auth agent finds potential vulnerability? → Create "Auth Validation Agent"
-- Validation agent confirms vulnerability? → Create "Auth Reporting Agent" that files the report and an initial fix candidate (`code_locations` + `fix_pr_body`)
+- Validation agent confirms vulnerability? → Create "Auth Reporting Agent" that files the report AND its inline fix (`code_locations` + `fix_pr_body`) in one shot — no separate fixing agent
 
 VULNERABILITY WORKFLOW (MANDATORY FOR EVERY FINDING):
 
@@ -404,10 +399,10 @@ Authentication Code Agent finds weak password validation
 Spawns "Auth Validation Agent" (proves it's exploitable)
     ↓
 If valid → Spawns "Auth Reporting Agent" (creates the vulnerability report
-           with the initial fix candidate: code_locations fix_before/fix_after
-           + fix_pr_body)
+           WITH the fix inline: code_locations fix_before/fix_after + fix_pr_body,
+           applying/verifying the patch in the same turn if desired)
     ↓
-STOP - the scan preparation phase handles repository-wide repair and verification
+STOP - no separate fixing agent; the fix was derived once, at report time
 ```
 
 CRITICAL RULES:
@@ -443,7 +438,7 @@ FOCUS PRINCIPLES:
 REALISTIC TESTING OUTCOMES:
 - **No Findings**: Agent completes testing but finds no vulnerabilities
 - **Validation Failed**: Initial finding was false positive, validation agent confirms it's not exploitable
-- **Valid Vulnerability**: Validation succeeds. The reporting agent files the report and an initial fix candidate for white-box findings.
+- **Valid Vulnerability**: Validation succeeds, spawns a reporting agent that files the report with the fix inline (white-box) — no separate fixing agent
 
 PERSISTENCE IS MANDATORY:
 - Real vulnerabilities take TIME - expect to need 2000+ steps minimum

strix/config/codex.py

@@ -356,8 +356,6 @@ def build_openai_client() -> AsyncOpenAI:
     import httpx
     from openai import AsyncOpenAI
 
-    from strix.llm import request_log
-
     get_valid_token()  # fail fast at configure time if the sign-in is dead
 
     async def _auth_hook(request: httpx.Request) -> None:
@@ -369,7 +367,6 @@ def build_openai_client() -> AsyncOpenAI:
         timeout=httpx.Timeout(600.0, connect=30.0),
         event_hooks={"request": [_auth_hook]},
     )
-    request_log.observe_http_client(http_client)
     return AsyncOpenAI(
         api_key="strix-codex-oauth",  # placeholder; the hook overwrites Authorization
         base_url=CODEX_BASE_URL,

strix/config/models.py

@@ -40,7 +40,6 @@ from strix.config import codex
 from strix.config.loader import load_settings
 from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
 from strix.config.tool_call_limits import TurnToolCallLimiter
-from strix.llm import request_log
 
 
 if TYPE_CHECKING:
@@ -391,7 +390,7 @@ def _guard_event(
     event: TResponseStreamEvent, rewriter: TurnCallIdRewriter, limiter: TurnToolCallLimiter
 ) -> TResponseStreamEvent | None:
     if isinstance(event, ResponseOutputItemAddedEvent | ResponseOutputItemDoneEvent):
-        rewritten = rewriter.rewrite_item(event.item, event.output_index)
+        rewritten = rewriter.rewrite_item(event.item)
         if not limiter.allow(rewritten):
             return None
         if rewritten is not event.item:
@@ -526,37 +525,13 @@ class StrixProvider(MultiProvider):
             # The ChatGPT subscription backend is always streamed; it has no
             # non-streaming mode to fall back to, so LLM_DISABLE_STREAMING
             # does not apply here.
-            model: Model = request_log.RequestLoggingModel(
-                _CodexResponsesModel(
-                    slug,
-                    codex.get_subscription_client(),
-                    reasoning_effort=llm.reasoning_effort,
-                ),
-                model_name=slug,
-                provider="openai-codex",
-                base_url=None,
+            model: Model = _CodexResponsesModel(
+                slug,
+                codex.get_subscription_client(),
+                reasoning_effort=llm.reasoning_effort,
             )
         else:
             model = super().get_model(model_name)
-            resolved_name = model_name or llm.model or "unknown"
-            if _routes_via_litellm(model):
-                # LiteLLM's callbacks log every reply; only a cancelled attempt
-                # (stream idle timeout, abandoned turn) escapes them.
-                model = request_log.RequestLoggingModel(
-                    model,
-                    model_name=resolved_name,
-                    provider=_litellm_provider(resolved_name),
-                    base_url=self._override_base_url or llm.api_base,
-                    route="litellm",
-                    abandoned_only=True,
-                )
-            else:
-                model = request_log.RequestLoggingModel(
-                    model,
-                    model_name=resolved_name,
-                    provider="openai",
-                    base_url=self._override_base_url or llm.api_base,
-                )
             if llm.disable_streaming:
                 model = _NonStreamingModel(model)
                 # The wrapper emits its single event only once the whole request
@@ -570,23 +545,6 @@ class StrixProvider(MultiProvider):
         )
 
 
-def _routes_via_litellm(model: Model) -> bool:
-    """LiteLLM-backed models are captured by the LiteLLM callback, not the wrapper."""
-    from agents.extensions.models.litellm_model import LitellmModel
-
-    return isinstance(model, LitellmModel)
-
-
-def _litellm_provider(model_name: str) -> str | None:
-    """The provider LiteLLM will route ``model_name`` to, if it can tell."""
-    try:
-        import litellm
-
-        return str(litellm.get_llm_provider(model_name)[1])
-    except Exception:  # noqa: BLE001 - unknown model ids are the provider's problem, not the log's
-        return None
-
-
 DEFAULT_MODEL_RETRY = ModelRetrySettings(
     max_retries=5,
     backoff=ModelRetryBackoffSettings(
@@ -663,7 +621,6 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
     """Apply Strix config to SDK-native defaults."""
     llm = settings.llm
     set_tracing_disabled(True)
-    request_log.install()
     if codex.subscription_model(llm.model):
         return
     _configure_litellm_compatibility()
@@ -816,16 +773,12 @@ def _merge_litellm_headers(headers: dict[str, str]) -> None:
 
 def _register_openai_client_with_headers(llm: LlmSettings, headers: dict[str, str]) -> None:
     from agents import set_default_openai_client
-    from agents.models.openai_provider import shared_http_client
     from openai import AsyncOpenAI
 
     client = AsyncOpenAI(
         api_key=llm.api_key or "not-needed",
         base_url=llm.api_base,
         default_headers=dict(headers),
-        # The SDK's shared client is the one the request log observes for
-        # reply status, headers and provider request ids.
-        http_client=shared_http_client(),
     )
     set_default_openai_client(client, use_for_tracing=False)
 

strix/config/tool_call_ids.py

@@ -1,17 +1,12 @@
-"""Keep tool-call ids present and unique within a conversation.
+"""Keep tool-call ids unique within a conversation.
 
 Some providers return per-turn tool-call ids (``exec_command:0``,
 ``exec_command:1``, ...) whose counter restarts on every turn. Once the same
 id appears twice in one conversation, the request payload has two assistant
 tool calls sharing an id and strict providers reject the whole turn, which
 permanently kills the agent because the malformed history is replayed on
-every retry.
-
-Others omit the id altogether, or return it as an empty string. That turns
-the paired tool result into a ``tool`` message with an empty
-``tool_call_id``, which strict providers reject the same way and with the
-same permanent outcome. Rewriting both blank and duplicate ids to fresh
-unique ones keeps the history valid for any provider.
+every retry. Rewriting duplicates to fresh unique ids keeps the history
+valid for any provider.
 """
 
 from __future__ import annotations
@@ -20,7 +15,6 @@ from collections import defaultdict, deque
 from typing import Any
 from uuid import uuid4
 
-from agents.models.fake_id import FAKE_RESPONSES_ID
 from openai.types.responses import ResponseFunctionToolCall
 
 
@@ -28,29 +22,23 @@ def new_call_id() -> str:
     return f"call_{uuid4().hex}"
 
 
-def _pairing_key(call_id: Any) -> str:
-    """Bucket an id for call/output pairing; all blank ids share one bucket."""
-    return call_id if isinstance(call_id, str) else ""
-
-
 def collect_call_ids(items: list[Any]) -> set[str]:
     used: set[str] = set()
     for item in items:
         if isinstance(item, dict):
             call_id = item.get("call_id")
-            if isinstance(call_id, str) and call_id:
+            if isinstance(call_id, str):
                 used.add(call_id)
-        elif isinstance(item, ResponseFunctionToolCall) and item.call_id:
+        elif isinstance(item, ResponseFunctionToolCall):
             used.add(item.call_id)
     return used
 
 
 def dedupe_history_call_ids(items: list[Any]) -> tuple[list[Any], bool]:
-    """Rewrite blank and duplicate call ids in a conversation history.
+    """Rewrite duplicate call ids in a conversation history.
 
     Outputs are paired with their call by order, so parallel calls that share
-    an id — or are all missing one — keep answering the right call after the
-    rewrite.
+    an id keep answering the right call after the rewrite.
     """
     used: set[str] = set()
     pending: dict[str, deque[str]] = defaultdict(deque)
@@ -61,24 +49,22 @@ def dedupe_history_call_ids(items: list[Any]) -> tuple[list[Any], bool]:
         if not isinstance(item, dict):
             rebuilt.append(item)
             continue
-
-        kind = item.get("type")
-        if kind not in ("function_call", "function_call_output"):
+        call_id = item.get("call_id")
+        if not isinstance(call_id, str):
             rebuilt.append(item)
             continue
 
-        call_id = item.get("call_id")
-        key = _pairing_key(call_id)
+        kind = item.get("type")
         if kind == "function_call":
-            effective = key
-            if not effective or effective in used:
+            effective = call_id
+            if call_id in used:
                 effective = new_call_id()
                 item = {**item, "call_id": effective}  # noqa: PLW2901
                 changed = True
             used.add(effective)
-            pending[key].append(effective)
-        else:
-            queue = pending.get(key)
+            pending[call_id].append(effective)
+        elif kind == "function_call_output":
+            queue = pending.get(call_id)
             if queue:
                 effective = queue.popleft()
                 if effective != call_id:
@@ -97,7 +83,7 @@ def dedupe_input(model_input: str | list[Any]) -> str | list[Any]:
 
 
 class TurnCallIdRewriter:
-    """Rewrite a single turn's tool-call ids that are blank or collide with the history.
+    """Rewrite a single turn's tool-call ids that collide with the history.
 
     A turn's items surface several times (streamed item events, then the
     completed response), so the same original id must always map to the same
@@ -107,37 +93,12 @@ class TurnCallIdRewriter:
     def __init__(self, model_input: str | list[Any]) -> None:
         self._used = set() if isinstance(model_input, str) else collect_call_ids(model_input)
         self._remap: dict[str, str] = {}
-        self._blank_remap: dict[str, str] = {}
         self._settled: set[str] = set()
 
-    def _rewrite_blank(
-        self, item: ResponseFunctionToolCall, position: int | None
-    ) -> ResponseFunctionToolCall:
-        """Give a call with no id one that stays the same on every sighting.
-
-        A real item id tells parallel calls apart. Chat Completions routes give
-        every item the same placeholder id, so there the call's position in
-        the turn's output tells them apart instead.
-        """
-        if item.id and item.id != FAKE_RESPONSES_ID:
-            key = item.id
-        else:
-            key = f"{FAKE_RESPONSES_ID}#{position}"
-        replacement = self._blank_remap.get(key)
-        if replacement is None:
-            replacement = new_call_id()
-            self._blank_remap[key] = replacement
-            self._used.add(replacement)
-            self._settled.add(replacement)
-        return item.model_copy(update={"call_id": replacement})
-
-    def rewrite_item(self, item: Any, position: int | None = None) -> Any:
-        """Rewrite one item; ``position`` is its index in the turn's output."""
+    def rewrite_item(self, item: Any) -> Any:
         if not isinstance(item, ResponseFunctionToolCall):
             return item
         original = item.call_id
-        if not original:
-            return self._rewrite_blank(item, position)
         if original in self._settled:
             return item
         replacement = self._remap.get(original)
@@ -153,4 +114,4 @@ class TurnCallIdRewriter:
         return item.model_copy(update={"call_id": replacement})
 
     def rewrite_items(self, items: list[Any]) -> list[Any]:
-        return [self.rewrite_item(item, position) for position, item in enumerate(items)]
+        return [self.rewrite_item(item) for item in items]

strix/core/execution.py

@@ -33,7 +33,6 @@ from strix.core.sessions import (
     seed_initial_input,
     strip_all_images_from_session,
 )
-from strix.llm import request_log
 from strix.llm.compaction import is_context_overflow, maybe_compact
 
 
@@ -198,44 +197,6 @@ async def run_agent_loop(
     start_parked: bool = False,
     event_sink: StreamEventSink | None = None,
     hooks: RunHooks[dict[str, Any]] | None = None,
-) -> RunResultBase | None:
-    agent_name = getattr(agent, "name", None)
-    token = request_log.bind_call_context(
-        agent_id, agent_name if isinstance(agent_name, str) else None
-    )
-    try:
-        return await _run_agent_loop(
-            agent=agent,
-            initial_input=initial_input,
-            run_config=run_config,
-            context=context,
-            max_turns=max_turns,
-            coordinator=coordinator,
-            agent_id=agent_id,
-            interactive=interactive,
-            session=session,
-            start_parked=start_parked,
-            event_sink=event_sink,
-            hooks=hooks,
-        )
-    finally:
-        request_log.reset_call_context(token)
-
-
-async def _run_agent_loop(
-    *,
-    agent: Any,
-    initial_input: Any,
-    run_config: RunConfig,
-    context: dict[str, Any],
-    max_turns: int,
-    coordinator: AgentCoordinator,
-    agent_id: str,
-    interactive: bool,
-    session: Session | None,
-    start_parked: bool,
-    event_sink: StreamEventSink | None,
-    hooks: RunHooks[dict[str, Any]] | None,
 ) -> RunResultBase | None:
     await coordinator.attach_runtime(
         agent_id,
@@ -675,7 +636,7 @@ async def _run_cycle_parked(
         raise
     except Exception as exc:
         logger.exception("error escaped the run cycle for %s; parking as failed", agent_id)
-        await coordinator.set_status(agent_id, "failed", error=request_log.failure_text(exc))
+        await coordinator.set_status(agent_id, "failed", error=str(exc) or type(exc).__name__)
         await notify_parent_on_terminal(coordinator, agent_id, "failed")
         return None
 
@@ -697,7 +658,6 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
     image_strips = 0
     compactions = 0
     model_retries = 0
-    request_log.set_retry_attempt(0)
     while True:
         stream: Any = None
         pre_run_items: list[Any] = []
@@ -825,7 +785,6 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                     exc,
                 )
                 await asyncio.sleep(delay)
-                request_log.set_retry_attempt(model_retries)
                 if session is not None:
                     input_data = []
                 continue
@@ -833,9 +792,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                 await _salvage_stream_to_session(session, pre_run_items, stream, agent_id)
             if isinstance(exc, ProviderRefusalError):
                 logger.warning("agent %s refused by the model provider: %s", agent_id, exc)
-                await coordinator.set_status(
-                    agent_id, "failed", error=request_log.failure_text(exc)
-                )
+                await coordinator.set_status(agent_id, "failed", error=str(exc))
                 await notify_parent_on_terminal(coordinator, agent_id, "failed")
                 return None
             if isinstance(exc, MaxTurnsExceeded):
@@ -849,7 +806,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
             # non-interactive agent's task: a child that dies still owes its parent a
             # report, and the parent would otherwise wait out its timeout on a message
             # the dead child can no longer send.
-            await coordinator.set_status(agent_id, status, error=request_log.failure_text(exc))
+            await coordinator.set_status(agent_id, status, error=str(exc) or type(exc).__name__)
             await notify_parent_on_terminal(coordinator, agent_id, status)
             if not interactive:
                 raise

strix/core/runner.py

@@ -59,8 +59,10 @@ if TYPE_CHECKING:
 
     from strix.runtime.status import StatusSink
     from strix.tools.mcp import (
+        ConnectedMcpServer,
         McpConnectionRequest,
         McpRegistry,
+        SupervisedMcpSession,
     )
 
 
@@ -69,8 +71,8 @@ logger = logging.getLogger(__name__)
 StreamEventSink = Callable[[str, Any], None]
 
 # Receives the run's MCP connection roster as a list of non-secret status dicts
-# ({"name", "provider", "tool_count", "dead", "state"}), once when the connections
-# are registered and again on lifecycle transitions. An interface
+# ({"name", "provider", "tool_count", "dead"}), once when the connections are
+# established and again each time a connection transitions to dead. An interface
 # can persist it, render it, or forward it on as connection status. Kept as a
 # snapshot of the whole roster (not a per-
 # connection delta) so every call carries a consistent, current picture.
@@ -78,21 +80,30 @@ McpStatusSink = Callable[[list[dict[str, Any]]], None]
 
 
 def _mcp_roster_payload(registry: McpRegistry) -> list[dict[str, Any]]:
-    """The run's MCP roster as non-secret lifecycle status dicts."""
+    """The run's MCP roster as non-secret status dicts (name/provider/tool_count/dead)."""
     return [
         {
             "name": status.name,
             "provider": status.provider,
             "tool_count": status.tool_count,
             "dead": status.dead,
-            "state": status.state,
         }
         for status in registry.statuses()
     ]
 
 
-def _record_mcp_connections(connection_names: list[str]) -> None:
-    """Record which MCP servers this run configured, for the interfaces.
+def _mcp_startup_summary(connections: list[ConnectedMcpServer]) -> str:
+    """One user-facing line summarizing the MCP servers that connected."""
+    server_count = len(connections)
+    tool_count = sum(c.tool_count for c in connections)
+    servers_word = "server" if server_count == 1 else "servers"
+    tools_word = "tool" if tool_count == 1 else "tools"
+    names = ", ".join(c.name for c in connections)
+    return f"MCP: connected {server_count} {servers_word} ({tool_count} {tools_word}): {names}"
+
+
+def _record_mcp_connections(connections: list[ConnectedMcpServer]) -> None:
+    """Record which MCP servers this run connected, for the interfaces.
 
     A server's tools are offered to the model under a name built from the
     connection name and the tool's own name, which cannot be split back apart, so
@@ -103,7 +114,7 @@ def _record_mcp_connections(connection_names: list[str]) -> None:
     report_state = get_global_report_state()
     if report_state is None:
         return
-    report_state.record_mcp_connections(connection_names)
+    report_state.record_mcp_connections([connection.name for connection in connections])
 
 
 def _note_exit_reason(reason: str) -> None:
@@ -337,7 +348,7 @@ async def run_strix_scan(
     configure_spill_writer(_spill_to_workspace)
 
     sessions_to_close: list[SQLiteSession] = []
-    mcp_registry: McpRegistry | None = None
+    mcp_sessions: list[SupervisedMcpSession] = []
 
     try:
         targets = scan_config.get("targets") or []
@@ -389,6 +400,7 @@ async def run_strix_scan(
         from strix.tools.mcp import (
             McpConnectionRequest,
             McpRegistry,
+            attach_mcp_requests,
             load_user_mcp_configs,
         )
 
@@ -404,37 +416,56 @@ async def run_strix_scan(
             else:
                 mcp_requests = mcp_connection_requests
             if mcp_requests:
-                for request in mcp_requests:
-                    mcp_registry.register(request)
-                _record_mcp_connections(mcp_registry.names())
-                report(
-                    f"MCP: configured {len(mcp_registry)} connection(s); "
-                    "warming them in the background"
-                )
-                scope_context["mcp_available"] = True
-                scope_context["mcp_connections"] = [
-                    {
-                        "name": summary.name,
-                        "purpose": summary.purpose,
-                        "tool_count": summary.tool_count,
-                    }
-                    for summary in mcp_registry.summaries()
-                ]
+                connections = await attach_mcp_requests(mcp_requests, mcp_registry)
+                mcp_sessions = [c.session for c in connections]
+                # Recorded even when nothing connected, so a resumed run does not
+                # keep attributing tool calls to servers it no longer has.
+                _record_mcp_connections(connections)
+                if connections:
+                    report(_mcp_startup_summary(connections))
+                    # Name the connected servers in the prompt so every agent
+                    # (root and children, both deriving from scope_context) sees
+                    # what is available at the start; they can still re-list or
+                    # inspect them at run time via list_mcps / describe_mcp. Set
+                    # only when a connection exists, so a run with no MCP leaves
+                    # the prompt context unchanged.
+                    scope_context["mcp_available"] = bool(mcp_registry)
+                    scope_context["mcp_connections"] = [
+                        {
+                            "name": summary.name,
+                            "purpose": summary.purpose,
+                            "tool_count": summary.tool_count,
+                        }
+                        for summary in mcp_registry.summaries()
+                    ]
 
-                def _emit_mcp_status() -> None:
-                    roster = _mcp_roster_payload(mcp_registry)
-                    _persist_mcp_status(roster)
-                    if mcp_status_sink is not None:
-                        try:
-                            mcp_status_sink(roster)
-                        except Exception:
-                            logger.exception("MCP status sink failed")
+                    # Feed a non-secret connection roster (name / provider /
+                    # tool_count / dead) to two consumers: once now (all
+                    # currently healthy) and again whenever a connection later
+                    # dies. It is always persisted to run.json so the viewer,
+                    # which re-reads the run's files from disk, can render the
+                    # MCP connections panel and health without an in-memory
+                    # sink. When an interface sink is attached (the TUI backend,
+                    # or pro forwarding into the app's event stream) it also
+                    # receives the same snapshot. In-use is derived separately by
+                    # each interface from the connection-tagged tool-call events,
+                    # so it is not carried here.
+                    def _emit_mcp_status() -> None:
+                        roster = _mcp_roster_payload(mcp_registry)
+                        _persist_mcp_status(roster)
+                        if mcp_status_sink is not None:
+                            try:
+                                mcp_status_sink(roster)
+                            except Exception:
+                                logger.exception("MCP status sink failed")
 
-                mcp_registry.set_status_sink(_emit_mcp_status)
-                _emit_mcp_status()
-                mcp_registry.start_warmup(max_concurrency=6)
+                    for connection_name in mcp_registry.names():
+                        entry = mcp_registry.get(connection_name)
+                        if entry is not None:
+                            entry.session.set_on_dead(_emit_mcp_status)
+                    _emit_mcp_status()
         except Exception:
-            logger.exception("Failed to configure user MCP servers; continuing without them")
+            logger.exception("Failed to connect user MCP servers; continuing without them")
 
         root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context)
         root_instructions = _compose_root_instructions_override(
@@ -630,9 +661,9 @@ async def run_strix_scan(
         for s in sessions_to_close:
             with contextlib.suppress(Exception):
                 s.close()
-        if mcp_registry is not None:
+        for mcp_session in mcp_sessions:
             with contextlib.suppress(Exception):
-                await mcp_registry.close()
+                await mcp_session.aclose()
         with contextlib.suppress(Exception):
             await coordinator._maybe_snapshot()
         if cleanup_on_exit:

strix/fix/__init__.py

@@ -1,64 +0,0 @@
-"""Verified fix preparation contracts and runtime."""
-
-from strix.fix.contracts import (
-    BlockerKind,
-    CandidateLocation,
-    CheckResult,
-    CheckStatus,
-    CommandSpec,
-    FileManifestEntry,
-    FixCandidateV1,
-    FixEdit,
-    FixPreparationAttempt,
-    FixPreparationRequestV1,
-    FixPreparationResultV1,
-    PreparationBlocker,
-    PreparationState,
-    RepairOutcome,
-    RepairStatus,
-    ReproductionSpec,
-    SourceIdentity,
-    SourceIdentityKind,
-    VerificationDecision,
-    VerificationTarget,
-    VerifierResult,
-    candidate_from_legacy_report,
-)
-from strix.fix.prepare import (
-    PreparationCancelledError,
-    PreparationContext,
-    PreparationPolicy,
-    build_git_manifest,
-    prepare_fix,
-)
-
-
-__all__ = [
-    "BlockerKind",
-    "CandidateLocation",
-    "CheckResult",
-    "CheckStatus",
-    "CommandSpec",
-    "FileManifestEntry",
-    "FixCandidateV1",
-    "FixEdit",
-    "FixPreparationAttempt",
-    "FixPreparationRequestV1",
-    "FixPreparationResultV1",
-    "PreparationBlocker",
-    "PreparationCancelledError",
-    "PreparationContext",
-    "PreparationPolicy",
-    "PreparationState",
-    "RepairOutcome",
-    "RepairStatus",
-    "ReproductionSpec",
-    "SourceIdentity",
-    "SourceIdentityKind",
-    "VerificationDecision",
-    "VerificationTarget",
-    "VerifierResult",
-    "build_git_manifest",
-    "candidate_from_legacy_report",
-    "prepare_fix",
-]

strix/fix/contracts.py

@@ -1,331 +0,0 @@
-"""Versioned contracts for verified fix preparation."""
-
-from __future__ import annotations
-
-import hashlib
-import json
-from enum import StrEnum
-from pathlib import PurePosixPath
-from typing import TYPE_CHECKING, Literal, cast
-
-from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
-
-
-if TYPE_CHECKING:
-    from collections.abc import Mapping
-
-
-class ContractModel(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-
-class SourceIdentityKind(StrEnum):
-    COMMIT = "commit"
-    ARCHIVE = "archive"
-
-
-class PreparationState(StrEnum):
-    PREPARING = "preparing"
-    READY = "ready"
-    BLOCKED = "blocked"
-    FAILED = "failed"
-    STALE = "stale"
-
-
-class CheckStatus(StrEnum):
-    PASSED = "passed"
-    FAILED = "failed"
-    UNAVAILABLE = "unavailable"
-    CANCELLED = "cancelled"
-
-
-class VerificationTarget(StrEnum):
-    BASE = "base"
-    PATCHED = "patched"
-
-
-class VerificationDecision(StrEnum):
-    VERIFIED = "verified"
-    REJECTED = "rejected"
-    INCONCLUSIVE = "inconclusive"
-
-
-class RepairStatus(StrEnum):
-    COMPLETE = "complete"
-    BLOCKED = "blocked"
-    BUDGET_EXHAUSTED = "budget_exhausted"
-    INCOMPLETE = "incomplete"
-
-
-class BlockerKind(StrEnum):
-    SOURCE = "source"
-    ENVIRONMENT = "environment"
-    REPOSITORY_BASELINE = "repository_baseline"
-    CREDENTIAL = "credential"
-    EXTERNAL_CONFIGURATION = "external_configuration"
-    SECURITY_EVIDENCE = "security_evidence"
-
-
-class PreparationBlocker(ContractModel):
-    kind: BlockerKind
-    summary: str = Field(min_length=1)
-    user_action: str = Field(min_length=1)
-    details: list[str] = []
-
-
-class SourceIdentity(ContractModel):
-    kind: SourceIdentityKind
-    value: str = Field(min_length=1)
-    repository: str | None = None
-
-    @field_validator("value")
-    @classmethod
-    def validate_value(cls, value: str) -> str:
-        normalized = value.strip().lower()
-        if not normalized or any(char not in "0123456789abcdef" for char in normalized):
-            raise ValueError("source identity must be a hexadecimal digest")
-        return normalized
-
-    @model_validator(mode="after")
-    def validate_digest_length(self) -> SourceIdentity:
-        valid_lengths = {40, 64} if self.kind is SourceIdentityKind.COMMIT else {64}
-        if len(self.value) not in valid_lengths:
-            expected = "40 or 64" if self.kind is SourceIdentityKind.COMMIT else "64"
-            raise ValueError(f"{self.kind} identity must contain {expected} hexadecimal characters")
-        return self
-
-
-def _validate_relative_path(value: str) -> str:
-    path = value.strip()
-    parsed = PurePosixPath(path)
-    if not path or parsed.is_absolute() or ".." in parsed.parts or path.startswith("./"):
-        raise ValueError("path must be relative to the repository root")
-    return path
-
-
-class CandidateLocation(ContractModel):
-    file: str
-    start_line: int = Field(ge=1)
-    end_line: int = Field(ge=1)
-    snippet: str | None = None
-    label: str | None = None
-
-    _relative_file = field_validator("file")(_validate_relative_path)
-
-    @model_validator(mode="after")
-    def validate_range(self) -> CandidateLocation:
-        if self.end_line < self.start_line:
-            raise ValueError("end_line must be greater than or equal to start_line")
-        return self
-
-
-class FixEdit(CandidateLocation):
-    before: str = Field(min_length=1)
-    after: str
-    original_sha256: str | None = None
-
-
-class CommandSpec(ContractModel):
-    name: str = Field(min_length=1, max_length=120)
-    argv: list[str] = Field(min_length=1)
-    required: bool = True
-    timeout_seconds: int = Field(default=300, ge=1, le=3600)
-    cwd: str = "."
-
-    _relative_cwd = field_validator("cwd")(_validate_relative_path)
-
-    @field_validator("argv")
-    @classmethod
-    def validate_argv(cls, value: list[str]) -> list[str]:
-        if any(not argument or "\x00" in argument for argument in value):
-            raise ValueError("command arguments must be non-empty and cannot contain NUL")
-        return value
-
-
-class ReproductionSpec(ContractModel):
-    instructions: str = Field(min_length=1)
-    command: CommandSpec | None = None
-
-
-class ReportedCheck(ContractModel):
-    name: str
-    result: str
-    executed: bool = False
-
-
-class FixCandidateV1(ContractModel):
-    version: Literal["1"] = "1"
-    source_identity: SourceIdentity | None = None
-    security_invariant: str = Field(min_length=1)
-    finding_locations: list[CandidateLocation] = []
-    draft_edits: list[FixEdit] = []
-    reproduction: ReproductionSpec | None = None
-    reported_checks: list[ReportedCheck] = []
-    known_gaps: list[str] = []
-
-    def digest(self) -> str:
-        payload = json.dumps(
-            self.model_dump(mode="json"),
-            sort_keys=True,
-            separators=(",", ":"),
-        ).encode()
-        return hashlib.sha256(payload).hexdigest()
-
-
-class FixPreparationRequestV1(ContractModel):
-    version: Literal["1"] = "1"
-    organization_id: str | None = None
-    scan_id: str
-    finding_id: str
-    repository_id: str | None = None
-    candidate: FixCandidateV1
-    checks: list[CommandSpec] = []
-    max_repair_attempts: int = Field(default=2, ge=1, le=2)
-    timeout_seconds: int = Field(default=1800, ge=30, le=14400)
-    network_allowed: bool = False
-    credentials_allowed: list[str] = []
-
-
-class CheckResult(ContractModel):
-    name: str
-    argv: list[str]
-    status: CheckStatus
-    exit_code: int | None = None
-    duration_seconds: float = Field(ge=0)
-    output: str = ""
-    required: bool = True
-    target: VerificationTarget | None = None
-    baseline_status: CheckStatus | None = None
-    baseline_output: str | None = None
-
-
-class VerifierResult(ContractModel):
-    decision: VerificationDecision
-    summary: str
-    security_invariant_closed: bool = False
-    reproduction_executed: bool = False
-    reproduction_summary: str | None = None
-    sibling_paths_reviewed: list[str] = []
-    preserved_behaviors: list[str] = []
-    gaps: list[str] = []
-    security_tests: list[CheckResult] = []
-    repairable: bool = False
-    blocker: PreparationBlocker | None = None
-
-
-class RepairOutcome(ContractModel):
-    status: RepairStatus
-    summary: str = Field(min_length=1)
-    gaps: list[str] = []
-    reproduction_command: CommandSpec | None = None
-    turns_used: int = Field(default=0, ge=0)
-    blocker: PreparationBlocker | None = None
-
-
-class FixPreparationAttempt(ContractModel):
-    attempt: int = Field(ge=1)
-    repair: RepairOutcome
-    checks: list[CheckResult] = []
-    security_reproduction: CheckResult | None = None
-    verifier: VerifierResult | None = None
-    workspace_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
-
-
-class FileManifestEntry(ContractModel):
-    path: str
-    operation: Literal["add", "modify", "delete"]
-    original_sha256: str | None = None
-    resulting_sha256: str | None = None
-    artifact_ref: str | None = None
-
-    _relative_path = field_validator("path")(_validate_relative_path)
-
-
-class FixPreparationResultV1(ContractModel):
-    version: Literal["1"] = "1"
-    state: PreparationState
-    stop_reason: str
-    source_identity: SourceIdentity | None
-    candidate: FixCandidateV1
-    candidate_digest: str
-    final_file_manifest: list[FileManifestEntry] = []
-    artifact_ref: str | None = None
-    changed_files: list[str] = []
-    diff_summary: str = ""
-    checks: list[CheckResult] = []
-    security_reproduction: CheckResult | None = None
-    verifier: VerifierResult | None = None
-    attempt_history: list[FixPreparationAttempt] = []
-    gaps: list[str] = []
-    blocker: PreparationBlocker | None = None
-    attempts: int = Field(default=0, ge=0)
-    elapsed_seconds: float = Field(default=0, ge=0)
-    cost_usd: float | None = Field(default=None, ge=0)
-
-
-def candidate_from_legacy_report(
-    report: Mapping[str, object],
-    *,
-    source_identity: SourceIdentity | None = None,
-) -> FixCandidateV1 | None:
-    raw_locations = report.get("code_locations")
-    if not isinstance(raw_locations, list):
-        return None
-    location_values = cast("list[object]", raw_locations)
-
-    locations: list[CandidateLocation] = []
-    edits: list[FixEdit] = []
-    for raw_value in location_values:
-        if not isinstance(raw_value, dict):
-            continue
-        raw = cast("dict[str, object]", raw_value)
-        file_path = raw.get("file")
-        start_line = raw.get("start_line")
-        end_line = raw.get("end_line")
-        if not isinstance(file_path, str) or type(start_line) is not int:
-            continue
-        if type(end_line) is not int:
-            end_line = start_line
-        common = {
-            "file": file_path,
-            "start_line": start_line,
-            "end_line": end_line,
-            "snippet": raw.get("snippet") if isinstance(raw.get("snippet"), str) else None,
-            "label": raw.get("label") if isinstance(raw.get("label"), str) else None,
-        }
-        try:
-            locations.append(CandidateLocation.model_validate(common))
-            before = raw.get("fix_before")
-            after = raw.get("fix_after")
-            if isinstance(before, str) and isinstance(after, str):
-                edits.append(FixEdit.model_validate({**common, "before": before, "after": after}))
-        except ValueError:
-            continue
-
-    if not locations:
-        return None
-
-    invariant = str(
-        report.get("remediation_steps") or report.get("technical_analysis") or ""
-    ).strip()
-    if not invariant:
-        invariant = "Resolve the reported security finding without changing legitimate behavior."
-
-    reported_checks: list[ReportedCheck] = []
-    verification = str(report.get("fix_verification") or "").strip()
-    if verification:
-        reported_checks.append(
-            ReportedCheck(name="reporting-agent verification", result=verification, executed=False)
-        )
-
-    return FixCandidateV1(
-        source_identity=source_identity,
-        security_invariant=invariant,
-        finding_locations=locations,
-        draft_edits=edits,
-        reproduction=ReproductionSpec(
-            instructions=str(report.get("poc_description") or report.get("evidence") or invariant)
-        ),
-        reported_checks=reported_checks,
-        known_gaps=["The reporting-agent verification is not independent."],
-    )

strix/fix/locations.py

@@ -1,109 +0,0 @@
-"""Deterministic source anchoring for finding locations and draft edits."""
-
-from __future__ import annotations
-
-import hashlib
-from dataclasses import dataclass
-from enum import StrEnum
-from typing import TYPE_CHECKING
-
-from strix.fix.contracts import CandidateLocation, FixCandidateV1, FixEdit
-
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
-
-class AnchorStatus(StrEnum):
-    UNIQUE = "unique"
-    MISSING = "missing"
-    AMBIGUOUS = "ambiguous"
-    STALE = "stale"
-
-
-@dataclass(frozen=True, slots=True)
-class AnchorResult:
-    status: AnchorStatus
-    location: CandidateLocation | FixEdit
-    matches: tuple[int, ...] = ()
-
-
-def sha256_text(value: str) -> str:
-    return hashlib.sha256(value.encode()).hexdigest()
-
-
-def _find_blocks(content: str, block: str) -> tuple[int, ...]:
-    source_lines = content.splitlines()
-    block_lines = block.splitlines()
-    if not block_lines:
-        return ()
-    width = len(block_lines)
-    return tuple(
-        index + 1
-        for index in range(len(source_lines) - width + 1)
-        if source_lines[index : index + width] == block_lines
-    )
-
-
-def _read_anchor_source(root: Path, file: str) -> str | None:
-    """Read a location's source, refusing symlinks that escape ``root`` and
-    files that cannot be decoded as UTF-8."""
-    resolved_root = root.resolve()
-    file_path = (root / file).resolve()
-    if not file_path.is_relative_to(resolved_root) or not file_path.is_file():
-        return None
-    try:
-        return file_path.read_text(encoding="utf-8")
-    except (OSError, UnicodeError):
-        return None
-
-
-def anchor_location(
-    root: Path,
-    location: CandidateLocation | FixEdit,
-) -> AnchorResult:
-    content = _read_anchor_source(root, location.file)
-    if content is None:
-        return AnchorResult(AnchorStatus.MISSING, location)
-    block = location.before if isinstance(location, FixEdit) else location.snippet
-    if not block:
-        return AnchorResult(AnchorStatus.UNIQUE, location, (location.start_line,))
-
-    matches = _find_blocks(content, block)
-    if not matches:
-        if (
-            isinstance(location, FixEdit)
-            and location.original_sha256
-            and sha256_text(content) != location.original_sha256
-        ):
-            return AnchorResult(AnchorStatus.STALE, location)
-        return AnchorResult(AnchorStatus.MISSING, location)
-    if len(matches) > 1:
-        return AnchorResult(AnchorStatus.AMBIGUOUS, location, matches)
-
-    start_line = next(iter(matches))
-    end_line = start_line + len(block.splitlines()) - 1
-    anchored = location.model_copy(update={"start_line": start_line, "end_line": end_line})
-    return AnchorResult(AnchorStatus.UNIQUE, anchored, matches)
-
-
-def anchor_candidate(
-    root: Path,
-    candidate: FixCandidateV1,
-) -> tuple[FixCandidateV1, list[AnchorResult]]:
-    results = [
-        *(anchor_location(root, location) for location in candidate.finding_locations),
-        *(anchor_location(root, edit) for edit in candidate.draft_edits),
-    ]
-    finding_count = len(candidate.finding_locations)
-    anchored_locations = [result.location for result in results[:finding_count]]
-    anchored_edits = [result.location for result in results[finding_count:]]
-    return (
-        candidate.model_copy(
-            update={
-                "finding_locations": anchored_locations,
-                "draft_edits": anchored_edits,
-            }
-        ),
-        results,
-    )

strix/fix/prepare.py

@@ -1,755 +0,0 @@
-"""Repository fix preparation engine."""
-
-from __future__ import annotations
-
-import asyncio
-import functools
-import hashlib
-import json
-import os
-import subprocess
-import time
-from collections.abc import Awaitable, Callable, Iterable
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Literal
-
-from strix.fix.contracts import (
-    BlockerKind,
-    CheckResult,
-    CheckStatus,
-    CommandSpec,
-    FileManifestEntry,
-    FixCandidateV1,
-    FixPreparationAttempt,
-    FixPreparationRequestV1,
-    FixPreparationResultV1,
-    PreparationBlocker,
-    PreparationState,
-    RepairOutcome,
-    RepairStatus,
-    VerificationDecision,
-    VerificationTarget,
-    VerifierResult,
-)
-from strix.fix.locations import AnchorStatus, anchor_location
-
-
-class PreparationCancelledError(RuntimeError):
-    pass
-
-
-CommandRunner = Callable[[Path, CommandSpec], Awaitable[CheckResult]]
-ManifestBuilder = Callable[[Path], Awaitable[tuple[list[FileManifestEntry], str, str | None]]]
-CancellationCheck = Callable[[], bool]
-
-
-@dataclass(slots=True)
-class PreparationPolicy:
-    max_repair_attempts: int = 2
-    timeout_seconds: int = 1800
-    max_output_chars: int = 20000
-
-    def __post_init__(self) -> None:
-        if not 1 <= self.max_repair_attempts <= 2:
-            raise ValueError("max_repair_attempts must be between 1 and 2")
-
-
-@dataclass(slots=True)
-class PreparationContext:
-    request: FixPreparationRequestV1
-    workspace: Path
-    candidate: FixCandidateV1
-    attempt: int = 0
-    feedback: list[FixPreparationAttempt] = field(default_factory=list)
-
-
-RepairAgent = Callable[
-    [PreparationContext, list[CheckResult]],
-    Awaitable[RepairOutcome | None],
-]
-IndependentVerifier = Callable[
-    [PreparationContext, list[CheckResult]],
-    Awaitable[VerifierResult],
-]
-SourceVerifier = Callable[[PreparationContext], Awaitable[bool]]
-
-
-_COMMAND_ENV_ALLOWLIST = frozenset(
-    {
-        "HOME",
-        "LANG",
-        "LC_ALL",
-        "PATH",
-        "PYTHONHOME",
-        "PYTHONPATH",
-        "SYSTEMROOT",
-        "TEMP",
-        "TMP",
-        "TMPDIR",
-        "VIRTUAL_ENV",
-        "SystemRoot",
-    }
-)
-
-
-@functools.lru_cache(maxsize=1)
-def _network_isolation_prefix() -> tuple[str, ...] | None:
-    """Return a working ``unshare`` prefix that creates an empty network
-    namespace, or None when the platform cannot isolate egress."""
-    for prefix in (("unshare", "-Urn"), ("unshare", "-n")):
-        try:
-            probe = subprocess.run(  # noqa: S603
-                [*prefix, "true"],
-                capture_output=True,
-                timeout=15,
-                check=False,
-            )
-        except (OSError, subprocess.SubprocessError):
-            continue
-        if probe.returncode == 0:
-            return prefix
-    return None
-
-
-def _command_environment(credentials_allowed: Iterable[str]) -> dict[str, str]:
-    allowed = _COMMAND_ENV_ALLOWLIST | set(credentials_allowed)
-    env = {key: value for key, value in os.environ.items() if key in allowed}
-    env["PYTHONDONTWRITEBYTECODE"] = "1"
-    return env
-
-
-async def run_command(
-    workspace: Path,
-    command: CommandSpec,
-    *,
-    credentials_allowed: Iterable[str] = (),
-    network_allowed: bool = False,
-) -> CheckResult:
-    started = time.monotonic()
-    cwd = (workspace / command.cwd).resolve()
-    if not cwd.is_relative_to(workspace.resolve()) or not cwd.is_dir():
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.UNAVAILABLE,
-            duration_seconds=time.monotonic() - started,
-            output="The command working directory is unavailable.",
-            required=command.required,
-        )
-    argv = list(command.argv)
-    if not network_allowed:
-        prefix = _network_isolation_prefix()
-        if prefix is None:
-            return CheckResult(
-                name=command.name,
-                argv=command.argv,
-                status=CheckStatus.UNAVAILABLE,
-                duration_seconds=time.monotonic() - started,
-                output="Network isolation is unavailable, so the command was not run.",
-                required=command.required,
-            )
-        argv = [*prefix, *argv]
-    process: asyncio.subprocess.Process | None = None
-    try:
-        process = await asyncio.create_subprocess_exec(
-            *argv,
-            cwd=cwd,
-            env=_command_environment(credentials_allowed),
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.STDOUT,
-        )
-        output, _ = await asyncio.wait_for(process.communicate(), command.timeout_seconds)
-    except (FileNotFoundError, PermissionError) as exc:
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.UNAVAILABLE,
-            duration_seconds=time.monotonic() - started,
-            output=str(exc),
-            required=command.required,
-        )
-    except TimeoutError:
-        if process is not None:
-            process.kill()
-            await process.wait()
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.FAILED,
-            duration_seconds=time.monotonic() - started,
-            output=f"Timed out after {command.timeout_seconds} seconds.",
-            required=command.required,
-        )
-    return CheckResult(
-        name=command.name,
-        argv=command.argv,
-        status=CheckStatus.PASSED if process.returncode == 0 else CheckStatus.FAILED,
-        exit_code=process.returncode,
-        duration_seconds=time.monotonic() - started,
-        output=output.decode(errors="replace")[-20000:],
-        required=command.required,
-    )
-
-
-async def build_git_manifest(
-    workspace: Path,
-) -> tuple[list[FileManifestEntry], str, str | None]:
-    process = await asyncio.create_subprocess_exec(
-        "git",
-        "status",
-        "--porcelain=v1",
-        "--untracked-files=all",
-        "-z",
-        cwd=workspace,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    output, error = await process.communicate()
-    if process.returncode != 0:
-        raise RuntimeError(error.decode(errors="replace"))
-
-    entries: list[FileManifestEntry] = []
-    changed_files: list[str] = []
-    records = [record for record in output.decode(errors="replace").split("\0") if record]
-    workspace_resolved = workspace.resolve()
-    index = 0
-    while index < len(records):
-        record = records[index]
-        status = record[:2]
-        path_text = record[3:]
-        if "R" in status or "C" in status:
-            index += 1
-        path = workspace / path_text
-        changed_files.append(path_text)
-        operation: Literal["add", "modify", "delete"]
-        if status == "??" or "A" in status:
-            operation = "add"
-        elif "D" in status:
-            operation = "delete"
-        else:
-            operation = "modify"
-        resolved = path.resolve()
-        contained = resolved.is_relative_to(workspace_resolved)
-        if operation == "add" and contained and resolved.is_dir():
-            for child in sorted(resolved.rglob("*")):
-                child_resolved = child.resolve()
-                if (
-                    not child_resolved.is_file()
-                    or not child_resolved.is_relative_to(workspace_resolved)
-                    or ".git" in child.relative_to(resolved).parts
-                ):
-                    continue
-                entries.append(
-                    FileManifestEntry(
-                        path=child_resolved.relative_to(workspace_resolved).as_posix(),
-                        operation="add",
-                        resulting_sha256=hashlib.sha256(child_resolved.read_bytes()).hexdigest(),
-                    )
-                )
-            index += 1
-            continue
-        resulting = (
-            hashlib.sha256(resolved.read_bytes()).hexdigest()
-            if contained and resolved.is_file()
-            else None
-        )
-        original: str | None = None
-        if operation != "add":
-            original_process = await asyncio.create_subprocess_exec(
-                "git",
-                "show",
-                f"HEAD:{path_text}",
-                cwd=workspace,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=subprocess.DEVNULL,
-            )
-            original_bytes, _ = await original_process.communicate()
-            if original_process.returncode == 0:
-                original = hashlib.sha256(original_bytes).hexdigest()
-        entries.append(
-            FileManifestEntry(
-                path=path_text,
-                operation=operation,
-                original_sha256=original,
-                resulting_sha256=resulting,
-            )
-        )
-        index += 1
-
-    summary_process = await asyncio.create_subprocess_exec(
-        "git",
-        "diff",
-        "--stat",
-        "--",
-        cwd=workspace,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=subprocess.DEVNULL,
-    )
-    summary, _ = await summary_process.communicate()
-    return entries, summary.decode(errors="replace").strip(), None
-
-
-async def _workspace_digest(workspace: Path) -> str:
-    manifest, _, _ = await build_git_manifest(workspace)
-    payload = json.dumps(
-        [entry.model_dump(mode="json") for entry in manifest],
-        sort_keys=True,
-        separators=(",", ":"),
-    ).encode()
-    return hashlib.sha256(payload).hexdigest()
-
-
-async def _verify_source(context: PreparationContext) -> bool:
-    identity = context.candidate.source_identity
-    if identity is None:
-        return False
-    if identity.kind == "commit":
-        process = await asyncio.create_subprocess_exec(
-            "git",
-            "rev-parse",
-            "HEAD",
-            cwd=context.workspace,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=subprocess.DEVNULL,
-        )
-        output, _ = await process.communicate()
-        if process.returncode != 0 or output.decode().strip().lower() != identity.value:
-            return False
-    status_process = await asyncio.create_subprocess_exec(
-        "git",
-        "status",
-        "--porcelain=v1",
-        "-z",
-        cwd=context.workspace,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=subprocess.DEVNULL,
-    )
-    status_output, _ = await status_process.communicate()
-    return status_process.returncode == 0 and not status_output.strip(b"\x00")
-
-
-def _result(
-    context: PreparationContext,
-    *,
-    state: PreparationState,
-    reason: str,
-    started: float,
-    checks: list[CheckResult] | None = None,
-    reproduction: CheckResult | None = None,
-    verifier: VerifierResult | None = None,
-    gaps: list[str] | None = None,
-    manifest: list[FileManifestEntry] | None = None,
-    diff_summary: str = "",
-    artifact_ref: str | None = None,
-    attempt_history: list[FixPreparationAttempt] | None = None,
-    blocker: PreparationBlocker | None = None,
-) -> FixPreparationResultV1:
-    return FixPreparationResultV1(
-        state=state,
-        stop_reason=reason,
-        source_identity=context.candidate.source_identity,
-        candidate=context.candidate,
-        candidate_digest=context.candidate.digest(),
-        final_file_manifest=manifest or [],
-        artifact_ref=artifact_ref,
-        changed_files=[entry.path for entry in manifest or []],
-        diff_summary=diff_summary,
-        checks=checks or [],
-        security_reproduction=reproduction,
-        verifier=verifier,
-        attempt_history=attempt_history or [],
-        gaps=gaps or [],
-        blocker=blocker,
-        attempts=context.attempt,
-        elapsed_seconds=time.monotonic() - started,
-    )
-
-
-def _repair_outcome(value: RepairOutcome | None) -> RepairOutcome:
-    if value is not None:
-        return value
-    return RepairOutcome(
-        status=RepairStatus.COMPLETE,
-        summary="The repair implementation returned control for independent evaluation.",
-    )
-
-
-def _required_checks_pass(checks: list[CheckResult]) -> bool:
-    required = [result for result in checks if result.required]
-    return bool(required) and all(result.status is CheckStatus.PASSED for result in required)
-
-
-def _verification_passes(
-    checks: list[CheckResult],
-    verifier: VerifierResult,
-) -> bool:
-    return (
-        _required_checks_pass(checks)
-        and verifier.decision is VerificationDecision.VERIFIED
-        and verifier.security_invariant_closed
-        and verifier.reproduction_executed
-        and any(
-            result.target is VerificationTarget.PATCHED and result.status is CheckStatus.PASSED
-            for result in verifier.security_tests
-        )
-    )
-
-
-def _verification_gaps(
-    repair_outcome: RepairOutcome,
-    checks: list[CheckResult],
-    verifier: VerifierResult,
-) -> list[str]:
-    gaps = list(repair_outcome.gaps)
-    required = [result for result in checks if result.required]
-    if not required:
-        gaps.append("No required repository check was configured.")
-    gaps.extend(
-        f"{result.name}: required check {result.status}"
-        for result in required
-        if result.status is not CheckStatus.PASSED
-    )
-    gaps.extend(
-        f"{result.name}: optional check {result.status}"
-        for result in checks
-        if not result.required and result.status is not CheckStatus.PASSED
-    )
-    if not verifier.security_tests:
-        gaps.append("The independent verifier did not execute a security test.")
-    elif not any(
-        result.target is VerificationTarget.PATCHED and result.status is CheckStatus.PASSED
-        for result in verifier.security_tests
-    ):
-        gaps.append("The independent verifier did not pass a security test on the patch.")
-    if not verifier.reproduction_executed:
-        gaps.append("The independent verifier did not execute the security reproduction.")
-    if not verifier.security_invariant_closed:
-        gaps.append("The independent verifier did not prove the security invariant.")
-    gaps.extend(verifier.gaps)
-    return list(dict.fromkeys(gaps))
-
-
-def _matches_baseline_failure(result: CheckResult) -> bool:
-    if result.baseline_status is not CheckStatus.FAILED:
-        return False
-    candidate_output = " ".join(result.output.split())
-    baseline_output = " ".join((result.baseline_output or "").split())
-    return bool(candidate_output and candidate_output == baseline_output)
-
-
-async def prepare_fix(  # noqa: PLR0915
-    request: FixPreparationRequestV1,
-    workspace: Path,
-    *,
-    repair: RepairAgent,
-    verify: IndependentVerifier,
-    command_runner: CommandRunner = run_command,
-    manifest_builder: ManifestBuilder = build_git_manifest,
-    source_verifier: SourceVerifier = _verify_source,
-    cancelled: CancellationCheck = lambda: False,
-    policy: PreparationPolicy | None = None,
-) -> FixPreparationResultV1:
-    started = time.monotonic()
-    resolved_policy = policy or PreparationPolicy(
-        max_repair_attempts=request.max_repair_attempts,
-        timeout_seconds=request.timeout_seconds,
-    )
-    context = PreparationContext(request=request, workspace=workspace, candidate=request.candidate)
-    runner: CommandRunner = command_runner
-    if runner is run_command:
-        runner = functools.partial(
-            run_command,
-            credentials_allowed=request.credentials_allowed,
-            network_allowed=request.network_allowed,
-        )
-
-    async def execute() -> FixPreparationResultV1:  # noqa: PLR0911, PLR0912, PLR0915
-        if cancelled():
-            raise PreparationCancelledError
-        if not await source_verifier(context):
-            blocker = PreparationBlocker(
-                kind=BlockerKind.SOURCE,
-                summary="The repository no longer matches the finding source.",
-                user_action="Refresh the finding against the current repository revision.",
-            )
-            return _result(
-                context,
-                state=PreparationState.STALE,
-                reason=blocker.summary,
-                blocker=blocker,
-                started=started,
-            )
-
-        anchors = [
-            anchor_location(workspace, location) for location in context.candidate.finding_locations
-        ]
-        if any(result.status is not AnchorStatus.UNIQUE for result in anchors):
-            details = [
-                f"{result.location.file}: {result.status}"
-                for result in anchors
-                if result.status is not AnchorStatus.UNIQUE
-            ]
-            blocker = PreparationBlocker(
-                kind=BlockerKind.SOURCE,
-                summary="The reported finding locations could not be resolved uniquely.",
-                user_action="Refresh the finding or identify the affected source location.",
-                details=details,
-            )
-            return _result(
-                context,
-                state=PreparationState.BLOCKED,
-                reason=blocker.summary,
-                gaps=details,
-                blocker=blocker,
-                started=started,
-            )
-        context.candidate = context.candidate.model_copy(
-            update={"finding_locations": [result.location for result in anchors]}
-        )
-
-        checks: list[CheckResult] = []
-        verifier: VerifierResult | None = None
-        attempt_history: list[FixPreparationAttempt] = []
-        for attempt in range(1, resolved_policy.max_repair_attempts + 1):
-            context.attempt = attempt
-            if cancelled():
-                raise PreparationCancelledError
-            repair_outcome = _repair_outcome(await repair(context, checks))
-
-            if repair_outcome.status is RepairStatus.BLOCKED:
-                manifest, summary, artifact_ref = await manifest_builder(workspace)
-                blocker = repair_outcome.blocker or PreparationBlocker(
-                    kind=BlockerKind.EXTERNAL_CONFIGURATION,
-                    summary=repair_outcome.summary,
-                    user_action=(
-                        "Provide the missing repository, credential, or production "
-                        "configuration prerequisite."
-                    ),
-                    details=repair_outcome.gaps,
-                )
-                return _result(
-                    context,
-                    state=PreparationState.BLOCKED,
-                    reason=blocker.summary,
-                    gaps=repair_outcome.gaps,
-                    manifest=manifest,
-                    diff_summary=summary,
-                    artifact_ref=artifact_ref,
-                    attempt_history=attempt_history,
-                    blocker=blocker,
-                    started=started,
-                )
-
-            if repair_outcome.status is not RepairStatus.COMPLETE:
-                manifest, summary, artifact_ref = await manifest_builder(workspace)
-                return _result(
-                    context,
-                    state=PreparationState.FAILED,
-                    reason=repair_outcome.summary,
-                    gaps=repair_outcome.gaps,
-                    manifest=manifest,
-                    diff_summary=summary,
-                    artifact_ref=artifact_ref,
-                    attempt_history=attempt_history,
-                    started=started,
-                )
-
-            checks = [await runner(workspace, check) for check in request.checks]
-            workspace_digest = await _workspace_digest(workspace)
-            attempt_record = FixPreparationAttempt(
-                attempt=attempt,
-                repair=repair_outcome,
-                checks=checks,
-                workspace_digest=workspace_digest,
-            )
-            attempt_history.append(attempt_record)
-            context.feedback = list(attempt_history)
-
-            required = [result for result in checks if result.required]
-            unavailable = [
-                result for result in required if result.status is CheckStatus.UNAVAILABLE
-            ]
-            if not required or unavailable:
-                details = (
-                    ["No required repository quality check was configured."]
-                    if not required
-                    else [f"{result.name}: {result.output}" for result in unavailable]
-                )
-                blocker = PreparationBlocker(
-                    kind=BlockerKind.ENVIRONMENT,
-                    summary="The repository quality gate could not run.",
-                    user_action=(
-                        "Provide the missing tool or repository setup needed to run "
-                        "the required checks."
-                    ),
-                    details=details,
-                )
-                manifest, summary, artifact_ref = await manifest_builder(workspace)
-                return _result(
-                    context,
-                    state=PreparationState.BLOCKED,
-                    reason=blocker.summary,
-                    checks=checks,
-                    gaps=blocker.details,
-                    manifest=manifest,
-                    diff_summary=summary,
-                    artifact_ref=artifact_ref,
-                    attempt_history=attempt_history,
-                    blocker=blocker,
-                    started=started,
-                )
-
-            failed = [result for result in required if result.status is CheckStatus.FAILED]
-            baseline_failures = [result for result in failed if _matches_baseline_failure(result)]
-            if baseline_failures:
-                blocker = PreparationBlocker(
-                    kind=BlockerKind.REPOSITORY_BASELINE,
-                    summary="Required checks also fail on the unchanged repository.",
-                    user_action=(
-                        "Repair the repository baseline or identify authoritative "
-                        "replacement checks."
-                    ),
-                    details=[result.name for result in baseline_failures],
-                )
-                manifest, summary, artifact_ref = await manifest_builder(workspace)
-                return _result(
-                    context,
-                    state=PreparationState.BLOCKED,
-                    reason=blocker.summary,
-                    checks=checks,
-                    gaps=blocker.details,
-                    manifest=manifest,
-                    diff_summary=summary,
-                    artifact_ref=artifact_ref,
-                    attempt_history=attempt_history,
-                    blocker=blocker,
-                    started=started,
-                )
-            if failed:
-                if attempt < resolved_policy.max_repair_attempts:
-                    continue
-                manifest, summary, artifact_ref = await manifest_builder(workspace)
-                return _result(
-                    context,
-                    state=PreparationState.FAILED,
-                    reason="The repair did not pass the repository quality gate.",
-                    checks=checks,
-                    gaps=[f"{result.name}: {result.output}" for result in failed],
-                    manifest=manifest,
-                    diff_summary=summary,
-                    artifact_ref=artifact_ref,
-                    attempt_history=attempt_history,
-                    started=started,
-                )
-
-            verifier = await verify(context, checks)
-            attempt_record.verifier = verifier
-            attempt_record.security_reproduction = next(
-                (
-                    result
-                    for result in verifier.security_tests
-                    if result.target is VerificationTarget.PATCHED
-                ),
-                None,
-            )
-            context.feedback = list(attempt_history)
-            gaps = _verification_gaps(repair_outcome, checks, verifier)
-
-            if verifier.blocker is not None:
-                manifest, summary, artifact_ref = await manifest_builder(workspace)
-                return _result(
-                    context,
-                    state=PreparationState.BLOCKED,
-                    reason=verifier.blocker.summary,
-                    checks=checks,
-                    reproduction=attempt_record.security_reproduction,
-                    verifier=verifier,
-                    gaps=gaps,
-                    manifest=manifest,
-                    diff_summary=summary,
-                    artifact_ref=artifact_ref,
-                    attempt_history=attempt_history,
-                    blocker=verifier.blocker,
-                    started=started,
-                )
-
-            if _verification_passes(checks, verifier):
-                manifest, summary, artifact_ref = await manifest_builder(workspace)
-                if not manifest:
-                    return _result(
-                        context,
-                        state=PreparationState.FAILED,
-                        reason="The repair did not change repository source.",
-                        checks=checks,
-                        verifier=verifier,
-                        gaps=["No prepared source change was produced."],
-                        attempt_history=attempt_history,
-                        started=started,
-                    )
-                return _result(
-                    context,
-                    state=PreparationState.READY,
-                    reason="The fix passed repository checks and security verification.",
-                    checks=checks,
-                    reproduction=attempt_record.security_reproduction,
-                    verifier=verifier,
-                    manifest=manifest,
-                    diff_summary=summary,
-                    artifact_ref=artifact_ref,
-                    attempt_history=attempt_history,
-                    started=started,
-                )
-
-            if verifier.repairable and attempt < resolved_policy.max_repair_attempts:
-                continue
-
-            manifest, summary, artifact_ref = await manifest_builder(workspace)
-            return _result(
-                context,
-                state=PreparationState.FAILED,
-                reason="The independent security verifier did not approve the repair.",
-                checks=checks,
-                reproduction=attempt_record.security_reproduction,
-                verifier=verifier,
-                gaps=gaps,
-                manifest=manifest,
-                diff_summary=summary,
-                artifact_ref=artifact_ref,
-                attempt_history=attempt_history,
-                started=started,
-            )
-
-        manifest, summary, artifact_ref = await manifest_builder(workspace)
-        return _result(
-            context,
-            state=PreparationState.FAILED,
-            reason="The repair did not satisfy the required gates.",
-            checks=checks,
-            verifier=verifier,
-            manifest=manifest,
-            diff_summary=summary,
-            artifact_ref=artifact_ref,
-            attempt_history=attempt_history,
-            started=started,
-        )
-
-    try:
-        async with asyncio.timeout(resolved_policy.timeout_seconds):
-            return await execute()
-    except PreparationCancelledError:
-        return _result(
-            context,
-            state=PreparationState.FAILED,
-            reason="Fix preparation was cancelled.",
-            attempt_history=context.feedback,
-            started=started,
-        )
-    except TimeoutError:
-        return _result(
-            context,
-            state=PreparationState.FAILED,
-            reason="Fix preparation exceeded its time limit.",
-            attempt_history=context.feedback,
-            started=started,
-        )

strix/interface/utils.py

@@ -209,9 +209,7 @@ def format_vulnerability_report(report: dict[str, Any]) -> Text:  # noqa: PLR091
                 text.append("\n  ")
                 text.append(loc["snippet"], style="dim")
             if loc.get("fix_before") or loc.get("fix_after"):
-                preparation = report.get("fix_preparation")
-                prepared = isinstance(preparation, dict) and preparation.get("state") == "ready"
-                text.append("\n  Prepared fix:" if prepared else "\n  Draft fix candidate:")
+                text.append("\n  Fix:")
                 if loc.get("fix_before"):
                     text.append("\n  - ", style="dim")
                     text.append(loc["fix_before"], style="dim")

strix/report/sarif.py

@@ -28,10 +28,9 @@ Design notes:
   * File locations must be repo-relative POSIX paths. Paths that look
     like URIs, absolute paths, or traversal patterns are rejected rather
     than emitted as invalid code-scanning alerts.
-  * Findings whose fix candidate completed verified preparation
-    (``fix_preparation.state == "ready"`` with a ``candidate_digest``
-    matching the candidate the result carries) are emitted as SARIF
-    ``fixes`` so code-scanning can render a one-click suggested change.
+  * Findings with a fix suggestion (``code_locations[].fix_before`` +
+    ``fix_after``) are emitted as SARIF ``fixes`` so code-scanning can
+    render a one-click suggested change.
   * Endpoint / target-only findings (typical of DAST) carry a SARIF
     ``logicalLocations`` entry so the finding keeps a meaningful anchor
     even without a source file + line.
@@ -57,10 +56,6 @@ import re
 from pathlib import Path, PurePosixPath
 from typing import Any, cast
 
-from pydantic import ValidationError
-
-from strix.fix.contracts import FixCandidateV1
-
 
 logger = logging.getLogger(__name__)
 
@@ -588,44 +583,46 @@ def _result_properties(
 
 
 def _build_fixes(report: dict[str, Any]) -> list[dict[str, Any]] | None:
-    """Build SARIF ``fixes`` from a verified prepared finding.
+    """Build SARIF ``fixes`` from a finding's code-location fix pairs.
 
-    SARIF consumers can apply ``fixes`` automatically. Strix emits them only
-    when preparation recorded a ``ready`` result. The result carries the
-    candidate preparation verified — after anchoring corrected its reported
-    lines — and its ``candidate_digest`` must match that candidate, so the
-    emitted replacements are exactly what preparation verified, never a
-    stale or diverged draft.
+    Strix findings carry the suggested change inline on each code
+    location as ``fix_before`` + ``fix_after``. We map every location
+    that has both (and a safe repo-relative URI + start line) into a
+    SARIF ``artifactChange``, replacing the finding's region with the
+    fixed text. Returns None when no location carries a usable fix pair.
     """
-    preparation = report.get("fix_preparation")
-    if not isinstance(preparation, dict) or preparation.get("state") != "ready":
-        return None
-    raw_candidate = preparation.get("candidate") or report.get("fix_candidate")
-    if not isinstance(raw_candidate, dict):
-        return None
-    try:
-        candidate = FixCandidateV1.model_validate(raw_candidate)
-    except ValidationError:
-        return None
-    if preparation.get("candidate_digest") != candidate.digest():
+    raw_locations = report.get("code_locations")
+    if not isinstance(raw_locations, list):
         return None
 
     artifact_changes: list[dict[str, Any]] = []
-    for edit in candidate.draft_edits:
-        uri = _sarif_uri(edit.file)
+    for location in raw_locations:
+        if not isinstance(location, dict):
+            continue
+        file_path = _string_value(location.get("file"))
+        fix_before = _string_value(location.get("fix_before"))
+        fix_after = _string_value(location.get("fix_after"))
+        start_line = location.get("start_line")
+        if not (file_path and fix_before and fix_after):
+            continue
+        if type(start_line) is not int or start_line < 1:
+            continue
+        uri = _sarif_uri(file_path)
         if uri is None:
             continue
-        deleted_region: dict[str, Any] = {
-            "startLine": edit.start_line,
-            "endLine": edit.end_line,
-        }
+
+        deleted_region: dict[str, Any] = {"startLine": start_line}
+        end_line = location.get("end_line")
+        if type(end_line) is int and end_line >= start_line:
+            deleted_region["endLine"] = end_line
+
         artifact_changes.append(
             {
                 "artifactLocation": {"uri": uri},
                 "replacements": [
                     {
                         "deletedRegion": deleted_region,
-                        "insertedContent": {"text": edit.after},
+                        "insertedContent": {"text": fix_after},
                     }
                 ],
             }

strix/report/state.py

@@ -93,8 +93,6 @@ UPDATABLE_REPORT_FIELDS = frozenset(
         "http_exchange_ids",
         "fix_verification",
         "fix_pr_body",
-        "fix_candidate",
-        "fix_preparation",
     }
 )
 
@@ -359,8 +357,6 @@ class ReportState:
         fix_pr_body: str | None = None,
         finding_class: str | None = None,
         dependency_metadata: dict[str, str] | None = None,
-        fix_candidate: dict[str, Any] | None = None,
-        fix_preparation: dict[str, Any] | None = None,
         agent_id: str | None = None,
         agent_name: str | None = None,
     ) -> str:
@@ -424,10 +420,6 @@ class ReportState:
         report["finding_class"] = (finding_class or "dynamic").strip().lower()
         if dependency_metadata:
             report["dependency_metadata"] = dependency_metadata
-        if fix_candidate:
-            report["fix_candidate"] = fix_candidate
-        if fix_preparation:
-            report["fix_preparation"] = fix_preparation
         if agent_id:
             report["agent_id"] = agent_id
         if agent_name:
@@ -921,9 +913,6 @@ class ReportState:
                 context["ref"] = f"refs/heads/{branch}"
         return context
 
-    def get_repository_context(self) -> dict[str, Any] | None:
-        return self._derive_repository_context()
-
     def _sync_llm_usage_record(self) -> None:
         self.run_record["llm_usage"] = self._build_llm_usage_record()
 
@@ -944,7 +933,6 @@ def openrouter_stream_cost(usage: Any) -> float | None:
     """
     if not isinstance(usage, dict):
         return None
-
     total = 0.0
     cost = usage.get("cost")
     if isinstance(cost, int | float) and cost > 0:

strix/report/writer.py

@@ -332,10 +332,7 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
                 lines.extend(f"  {ln}" for ln in snippet.splitlines())
                 lines.append(f"  {fence}")
             if loc.get("fix_before") or loc.get("fix_after"):
-                preparation = report.get("fix_preparation")
-                prepared = isinstance(preparation, dict) and preparation.get("state") == "ready"
-                label = "Prepared Fix" if prepared else "Draft Fix Candidate"
-                lines.append(f"\n  **{label}:**")
+                lines.append("\n  **Suggested Fix:**")
                 lines.append("```diff")
                 if loc.get("fix_before"):
                     lines.extend(f"- {ln}" for ln in str(loc["fix_before"]).splitlines())
@@ -350,19 +347,10 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
         lines.append("")
 
     if report.get("fix_verification"):
-        lines.append("## Reported Candidate Checks\n")
+        lines.append("## Fix Verification\n")
         lines.append(str(report["fix_verification"]))
         lines.append("")
 
-    if isinstance(report.get("fix_preparation"), dict):
-        preparation = report["fix_preparation"]
-        lines.append("## Fix Preparation\n")
-        lines.append(f"State: `{preparation.get('state', 'unknown')}`")
-        if preparation.get("stop_reason"):
-            lines.append("")
-            lines.append(str(preparation["stop_reason"]))
-        lines.append("")
-
     if report.get("assumptions"):
         lines.append("## Assumptions\n")
         lines.append(str(report["assumptions"]))

strix/tools/mcp/__init__.py

@@ -2,13 +2,7 @@
 
 from __future__ import annotations
 
-from strix.tools.mcp.agent_tools import (
-    call_mcp,
-    describe_mcp,
-    get_mcp_tool_schema,
-    list_mcps,
-    search_mcp_tools,
-)
+from strix.tools.mcp.agent_tools import call_mcp, describe_mcp, list_mcps
 from strix.tools.mcp.client import (
     ConnectedMcpServer,
     attach_mcp_requests,
@@ -25,10 +19,8 @@ from strix.tools.mcp.naming import namespaced_tool_name
 from strix.tools.mcp.registry import (
     CALL_MCP_TOOL,
     DESCRIBE_MCP_TOOL,
-    GET_MCP_TOOL_SCHEMA_TOOL,
     MCP_DISPATCH_TOOLS,
     MCP_REGISTRY_CONTEXT_KEY,
-    SEARCH_MCP_TOOLS_TOOL,
     McpCallInfo,
     McpConnectionEntry,
     McpConnectionRequest,
@@ -43,10 +35,8 @@ from strix.tools.mcp.session import McpConnectionUnavailableError, SupervisedMcp
 __all__ = [
     "CALL_MCP_TOOL",
     "DESCRIBE_MCP_TOOL",
-    "GET_MCP_TOOL_SCHEMA_TOOL",
     "MCP_DISPATCH_TOOLS",
     "MCP_REGISTRY_CONTEXT_KEY",
-    "SEARCH_MCP_TOOLS_TOOL",
     "BearerAuth",
     "ConnectedMcpServer",
     "FailureInfo",
@@ -66,10 +56,8 @@ __all__ = [
     "classify",
     "connect_mcp_servers",
     "describe_mcp",
-    "get_mcp_tool_schema",
     "list_mcps",
     "load_user_mcp_configs",
     "namespaced_tool_name",
     "resolve_mcp_call",
-    "search_mcp_tools",
 ]

strix/tools/mcp/agent_tools.py

@@ -1,21 +1,18 @@
-"""The generic MCP discovery and dispatch tools every agent carries.
+"""The three generic MCP dispatch tools every agent carries.
 
 Under the generic-dispatch model an agent does not get one tool per MCP tool.
-It gets four primary tools and discovers connections on demand:
+It gets exactly these three and discovers connections on demand:
 
 - ``list_mcps()`` returns the connections available this run — each connection's
   id, name, description, and tool count, with no tool schemas — so the model can
   discover what it can reach without any inventory in the system prompt.
-- ``search_mcp_tools(connection, query)`` returns a small ranked candidate set.
-- ``get_mcp_tool_schema(connection, tool)`` returns one exact input schema.
+- ``describe_mcp(connection)`` returns, as text, one connection's tools with
+  their names, descriptions, and JSON input schemas — the schemas the model
+  needs, fetched on demand instead of loaded onto every request up front.
 - ``call_mcp(connection, tool, arguments)`` dispatches one call to a
   connection's tool and returns its result.
 
-``describe_mcp`` remains as a compatibility path for older prompts, but current
-agents use targeted search and one-schema lookup instead of receiving a whole
-provider catalog in one model turn.
-
-All four read the per-run :class:`~strix.tools.mcp.registry.McpRegistry` from the
+All three read the per-run :class:`~strix.tools.mcp.registry.McpRegistry` from the
 run context under :data:`~strix.tools.mcp.registry.MCP_REGISTRY_CONTEXT_KEY`. They
 are ordinary ``FunctionTool`` objects placed in the agent factory's base tool set,
 so the factory's output-bounding and disk-spill wrapping apply to their results
@@ -64,11 +61,12 @@ async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
     """List the MCP connections available this run, so you can discover them.
 
     Read-only. Returns one entry per connection with its ``id`` (the exact name
-    you pass to the other MCP tools), ``name``, ``description``, and
-    ``tool_count``. The response does not include tool schemas. Call
-    ``search_mcp_tools`` next. Then call ``get_mcp_tool_schema`` for one selected
-    tool before you call ``call_mcp``. Returns an empty ``connections`` list when
-    the run has no MCP connections.
+    you pass to ``describe_mcp`` and ``call_mcp``), ``name``, ``description``, and
+    ``tool_count`` — no tool schemas. The three MCP tools work in order: call
+    ``list_mcps`` to discover the available connections, then ``describe_mcp`` on
+    one connection to inspect its tools and their input schemas, then ``call_mcp``
+    to run one of its tools. Returns an empty ``connections`` list when the run has
+    no MCP connections.
     """
     registry = _registry_from_ctx(ctx)
     if registry is None or not registry:
@@ -82,7 +80,6 @@ async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
                 "description": summary.purpose,
                 "tool_count": summary.tool_count,
                 "dead": dead_by_name.get(summary.name, False),
-                "state": summary.state,
             }
             for summary in registry.summaries()
         ]
@@ -91,12 +88,13 @@ async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
 
 @function_tool(timeout=60)
 async def describe_mcp(ctx: RunContextWrapper, connection: str) -> str:
-    """Return one connection's full tool catalog as a compatibility fallback.
+    """List the tools one MCP connection offers, with their input schemas.
 
-    Read-only. This result can be large because it includes every tool schema.
-    First use ``search_mcp_tools`` and ``get_mcp_tool_schema``. Use this fallback
-    only when targeted search cannot identify an expected tool. Nothing is read
-    from the connected account.
+    Read-only. Look up a connection by the id ``list_mcps`` reported for it; this
+    returns each of its tools with the tool's name, description, and JSON input
+    schema — the argument shape you pass to ``call_mcp``. Call this before
+    ``call_mcp`` on any connection you have not used yet. Nothing is fetched from
+    or run against the connection's data.
 
     Args:
         connection: The connection name exactly as reported by ``list_mcps``.
@@ -108,7 +106,7 @@ async def describe_mcp(ctx: RunContextWrapper, connection: str) -> str:
     if entry is None:
         return _unknown_connection(connection, registry)
     try:
-        tools = await entry.ensure_catalog()
+        tools = await entry.session.list_tools()
     except McpConnectionUnavailableError as exc:
         return str(exc)
     if not tools:
@@ -118,121 +116,6 @@ async def describe_mcp(ctx: RunContextWrapper, connection: str) -> str:
     return f"{header}\n{body}"
 
 
-def _search_score(tool: MCPTool, query_terms: list[str], active: bool) -> tuple[int, str]:
-    name = tool.name.lower()
-    description = (tool.description or "").lower()
-    if not query_terms:
-        return (100 if active else 0, name)
-    score = 100 if active else 0
-    matched_terms = 0
-    for term in query_terms:
-        if name == term:
-            score += 60
-            matched_terms += 1
-        elif name.startswith(term):
-            score += 35
-            matched_terms += 1
-        elif term in name:
-            score += 25
-            matched_terms += 1
-        elif term in description:
-            score += 10
-            matched_terms += 1
-    if matched_terms == 0:
-        return (-1, name)
-    return (score + (matched_terms * 5), name)
-
-
-@function_tool(timeout=60)
-async def search_mcp_tools(
-    ctx: RunContextWrapper,
-    connection: str,
-    query: str,
-    limit: int = 8,
-) -> dict[str, Any] | str:
-    """Search one connection's tools without returning their full schemas.
-
-    The first search lazily connects the provider and caches its catalog. Results
-    contain only names, short descriptions, and whether each tool is in the
-    scan's active set. Call ``get_mcp_tool_schema`` for the selected tool.
-
-    Args:
-        connection: The connection name exactly as reported by ``list_mcps``.
-        query: Capability words such as ``page content`` or ``workspace identity``.
-        limit: Maximum candidates to return, from 1 through 20.
-    """
-    registry = _registry_from_ctx(ctx)
-    if registry is None or not registry:
-        return _NO_CONNECTIONS
-    entry = registry.get(connection)
-    if entry is None:
-        return _unknown_connection(connection, registry)
-    try:
-        tools = await entry.ensure_catalog()
-    except McpConnectionUnavailableError as exc:
-        return str(exc)
-    terms = [term for term in query.lower().split() if term]
-    bounded_limit = min(max(limit, 1), 20)
-    ranked = [
-        (score, tool)
-        for tool in tools
-        if (score := _search_score(tool, terms, tool.name in entry.active_tools))[0] >= 0
-    ]
-    if not ranked:
-        ranked = [
-            ((100, tool.name.lower()), tool) for tool in tools if tool.name in entry.active_tools
-        ]
-    ranked.sort(key=lambda item: (-item[0][0], item[0][1]))
-    return {
-        "connection": connection,
-        "query": query,
-        "matches": [
-            {
-                "name": tool.name,
-                "description": (tool.description or "").strip() or None,
-                "active": tool.name in entry.active_tools,
-            }
-            for _, tool in ranked[:bounded_limit]
-        ],
-    }
-
-
-@function_tool(timeout=60)
-async def get_mcp_tool_schema(
-    ctx: RunContextWrapper,
-    connection: str,
-    tool: str,
-) -> dict[str, Any] | str:
-    """Return one MCP tool's exact input schema.
-
-    Args:
-        connection: The connection name exactly as reported by ``list_mcps``.
-        tool: One exact tool name returned by ``search_mcp_tools``.
-    """
-    registry = _registry_from_ctx(ctx)
-    if registry is None or not registry:
-        return _NO_CONNECTIONS
-    entry = registry.get(connection)
-    if entry is None:
-        return _unknown_connection(connection, registry)
-    try:
-        tools = await entry.ensure_catalog()
-    except McpConnectionUnavailableError as exc:
-        return str(exc)
-    match = next((candidate for candidate in tools if candidate.name == tool), None)
-    if match is None:
-        return (
-            f"Unknown tool {tool!r} on MCP connection {connection!r}. "
-            "Call search_mcp_tools to find an available tool."
-        )
-    return {
-        "connection": connection,
-        "name": match.name,
-        "description": (match.description or "").strip() or None,
-        "input_schema": match.inputSchema or {"type": "object"},
-    }
-
-
 @function_tool(timeout=120, strict_mode=False)
 async def call_mcp(
     ctx: RunContextWrapper,
@@ -242,18 +125,19 @@ async def call_mcp(
 ) -> Any:
     """Call one tool on one MCP connection and return its result.
 
-    Use the connection id from ``list_mcps``. Use the tool name from
-    ``search_mcp_tools``. Pass an object that matches the schema from
-    ``get_mcp_tool_schema``. Omit the arguments for a tool that takes no
-    arguments.
+    Address the tool by the connection id from ``list_mcps`` and the tool name
+    from ``describe_mcp`` on that connection. Pass the tool's arguments as an
+    object matching the input schema ``describe_mcp`` showed for it (omit it, or
+    pass an empty object, for a tool that takes no arguments).
 
     Args:
         connection: The connection name exactly as reported by ``list_mcps``.
-        tool: The tool name exactly as reported by ``search_mcp_tools``.
+        tool: The tool name, exactly as reported by ``describe_mcp``.
         arguments: The tool's arguments as a JSON object of names to values (for
             example ``{"path": "app.py"}``), or omitted/empty for a tool that
-            takes none. Pass an object, not a stringified object. Match the shape
-            that ``get_mcp_tool_schema`` returned.
+            takes none. Pass an object, not a stringified one. Its shape is
+            whatever ``describe_mcp`` showed for the tool rather than a shape this
+            tool fixes in advance.
     """
     registry = _registry_from_ctx(ctx)
     if registry is None or not registry:
@@ -263,7 +147,7 @@ async def call_mcp(
         return _unknown_connection(connection, registry)
     invalid_arguments = (
         f"Invalid arguments for {connection!r}.{tool}: expected a JSON object of "
-        "argument names to values, or none. Call get_mcp_tool_schema for the schema."
+        "argument names to values, or none. Call describe_mcp for the input schema."
     )
     if isinstance(arguments, str):
         # The ``arguments`` parameter is schema-less (an open object is not
@@ -278,17 +162,18 @@ async def call_mcp(
     if arguments is not None and not isinstance(arguments, dict):
         return invalid_arguments
     try:
-        available = await entry.ensure_catalog()
+        available = await entry.session.list_tools()
     except McpConnectionUnavailableError as exc:
         return _errored_tool_output(str(exc))
     valid_names = {mcp_tool.name for mcp_tool in available}
     if tool not in valid_names:
+        offered = ", ".join(sorted(valid_names)) or "(none)"
         return (
             f"Unknown tool {tool!r} on MCP connection {connection!r}. "
-            "Call search_mcp_tools, then get_mcp_tool_schema."
+            f"Tools this connection offers: {offered}. "
+            "Call describe_mcp for their input schemas."
         )
-    session = await entry.ensure_connected()
-    return await session.dispatch(
+    return await entry.session.dispatch(
         tool,
         arguments or {},
         label=namespaced_tool_name(connection, tool),

strix/tools/mcp/client.py

@@ -14,7 +14,6 @@ fails the run.
 
 from __future__ import annotations
 
-import asyncio
 import contextlib
 import json
 import logging
@@ -282,10 +281,8 @@ async def _count_session_tools(config: McpConnectionConfig, session: SupervisedM
 
 async def connect_mcp_servers(
     configs: list[McpConnectionConfig],
-    *,
-    max_concurrency: int = 6,
 ) -> list[ConnectedMcpServer]:
-    """Connect MCP configs concurrently under a fixed bound.
+    """Connect each MCP config on its own supervising task and return the sessions.
 
     Each connection becomes a :class:`~strix.tools.mcp.session.SupervisedMcpSession`
     that owns ``connect()``, the held-open session, and ``cleanup()`` on one
@@ -304,43 +301,41 @@ async def connect_mcp_servers(
     :class:`~strix.tools.mcp.registry.McpRegistry` from these sessions, and the
     agent reaches each tool on demand through ``describe_mcp`` / ``call_mcp``.
     """
-    semaphore = asyncio.Semaphore(max(1, max_concurrency))
+    connected: list[ConnectedMcpServer] = []
     sessions: list[SupervisedMcpSession] = []
-
-    async def connect_one(config: McpConnectionConfig) -> ConnectedMcpServer | None:
-        async with semaphore:
+    try:
+        for config in configs:
             session = SupervisedMcpSession(config)
             sessions.append(session)
+            if not await session.start():
+                # Initial connect failed; already logged inside the session. Drop it.
+                await session.aclose()
+                sessions.remove(session)
+                continue
             try:
-                if not await session.start():
-                    await session.aclose()
-                    return None
                 tool_count = await _count_session_tools(config, session)
             except McpConnectionUnavailableError:
+                # The session died between connecting and its first listing; skip it.
                 logger.warning("MCP connection %r died before its first listing", config.name)
                 await session.aclose()
-                return None
-            except BaseException:
-                with contextlib.suppress(BaseException):
-                    await session.aclose()
-                raise
+                sessions.remove(session)
+                continue
             logger.info("Connected MCP server %r (%d tools)", config.name, tool_count)
-            return ConnectedMcpServer(
-                session=session,
-                name=config.name,
-                tool_count=tool_count,
-                notes=config.notes,
+            connected.append(
+                ConnectedMcpServer(
+                    session=session, name=config.name, tool_count=tool_count, notes=config.notes
+                )
             )
-
-    try:
-        results = await asyncio.gather(*(connect_one(config) for config in configs))
     except BaseException:
-        await asyncio.gather(
-            *(session.aclose() for session in sessions),
-            return_exceptions=True,
-        )
+        # Cancelled or errored mid-attach: close every session started so far,
+        # each on its own task, then re-raise. The runner only receives the list
+        # on a clean return, so on an abnormal exit this function owns the cleanup.
+        for session in sessions:
+            with contextlib.suppress(BaseException):
+                await session.aclose()
         raise
-    return [result for result in results if result is not None]
+
+    return connected
 
 
 async def attach_mcp_requests(

strix/tools/mcp/config.py

@@ -62,9 +62,6 @@ class McpConnectionConfig(BaseModel):
     """Tool allowlist, applied after the server lists its tools. ``None`` (the
     default) exposes every tool the server lists; a list restricts to it."""
 
-    active_tools: list[str] = Field(default_factory=list)
-    """Small scan-relevant subset ranked ahead of the broader allowed catalog."""
-
     notes: str | None = None
     """Free-text notes for the agent describing what this connection is and how
     to use it. When set, the note becomes the connection's purpose line in the

strix/tools/mcp/registry.py

@@ -22,18 +22,13 @@ single dispatch point.
 
 from __future__ import annotations
 
-import asyncio
-import contextlib
 import dataclasses
-import time
-from typing import TYPE_CHECKING, Any, Literal, NamedTuple
+from typing import TYPE_CHECKING, Any, NamedTuple
 
-from strix.tools.mcp.session import McpConnectionUnavailableError, SupervisedMcpSession
+from strix.tools.mcp.session import SupervisedMcpSession
 
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
-
     from agents.mcp import MCPServer
 
     from strix.tools.mcp.client import ResultTransform
@@ -54,60 +49,37 @@ MCP_REGISTRY_CONTEXT_KEY = "mcp_registry"
 # tracer all recognise a connection-scoped dispatch call by the same names.
 CALL_MCP_TOOL = "call_mcp"
 DESCRIBE_MCP_TOOL = "describe_mcp"
-SEARCH_MCP_TOOLS_TOOL = "search_mcp_tools"
-GET_MCP_TOOL_SCHEMA_TOOL = "get_mcp_tool_schema"
-MCP_DISPATCH_TOOLS = frozenset(
-    {
-        CALL_MCP_TOOL,
-        DESCRIBE_MCP_TOOL,
-        SEARCH_MCP_TOOLS_TOOL,
-        GET_MCP_TOOL_SCHEMA_TOOL,
-    }
-)
-
-McpConnectionState = Literal[
-    "configured",
-    "connecting",
-    "connected",
-    "catalog_loading",
-    "catalog_ready",
-    "unavailable",
-]
-_RETRY_DELAY_SECONDS = 5.0
+MCP_DISPATCH_TOOLS = frozenset({CALL_MCP_TOOL, DESCRIBE_MCP_TOOL})
 
 
-@dataclasses.dataclass
+@dataclasses.dataclass(frozen=True)
 class McpConnectionEntry:
-    """One configured MCP connection a scan may reach, keyed by ``name``.
+    """One live MCP connection a scan may reach, keyed by ``name``.
 
-    Registration is inert. The first warm-up, search, schema lookup, or call
-    creates one shared connection task; the first catalog operation creates one
-    shared listing task. Root and child agents therefore reuse the same session
-    and catalog even when they request a cold connection concurrently.
+    ``session`` is the :class:`~strix.tools.mcp.session.SupervisedMcpSession` that
+    owns the connection on its own task; the dispatch tools list tools and call
+    tools through it (``session.list_tools`` / ``session.dispatch``) so a session
+    failure is contained and can reconnect. ``purpose`` is the human label
+    ``list_mcps`` reports as the connection's description (the user's connection
+    notes, or whatever the caller supplies). ``tool_count`` is how many tools the
+    connection offers, also reported by ``list_mcps``. ``result_transform``, when
+    set, runs on each call's structured result at the single dispatch point
+    (strix-pro's sanitizer uses it). ``provider`` is an optional source label
+    (e.g. ``"supabase"``) the caller tags the connection with; the command-line
+    path leaves it ``None``, and event tagging surfaces it when set.
+
+    The connection config the session reconnects with (and its bearer token) lives
+    on ``session`` in memory only. It is reached via :attr:`config` for the
+    reconnect path and is never logged, serialized into the event stream, or
+    written to disk.
     """
 
+    session: SupervisedMcpSession
     name: str
-    connection_config: McpConnectionConfig | None = dataclasses.field(
-        default=None,
-        repr=False,
-    )
-    session: SupervisedMcpSession | None = dataclasses.field(default=None, repr=False)
     purpose: str | None = None
     tool_count: int = 0
     result_transform: ResultTransform | None = None
     provider: str | None = None
-    state: McpConnectionState = "configured"
-    _catalog: list[Any] | None = dataclasses.field(default=None, repr=False)
-    _connect_task: asyncio.Task[SupervisedMcpSession] | None = dataclasses.field(
-        default=None,
-        repr=False,
-    )
-    _catalog_task: asyncio.Task[list[Any]] | None = dataclasses.field(
-        default=None,
-        repr=False,
-    )
-    _retry_after: float = dataclasses.field(default=0.0, repr=False)
-    _status_sink: Callable[[], None] | None = dataclasses.field(default=None, repr=False)
 
     @property
     def server(self) -> MCPServer | None:
@@ -116,143 +88,12 @@ class McpConnectionEntry:
         Kept so existing callers that read ``entry.server`` keep working; new code
         should call through ``entry.session`` so reconnect and containment apply.
         """
-        return self.session.server if self.session is not None else None
+        return self.session.server
 
     @property
     def config(self) -> McpConnectionConfig | None:
-        """The reconnect config. Carries the bearer token; never log it."""
-        if self.connection_config is not None:
-            return self.connection_config
-        return self.session.config if self.session is not None else None
-
-    @property
-    def active_tools(self) -> frozenset[str]:
-        config = self.config
-        return frozenset(config.active_tools if config is not None else ())
-
-    def set_status_sink(self, sink: Callable[[], None] | None) -> None:
-        self._status_sink = sink
-        if self.session is not None:
-            self.session.set_on_dead(self._on_dead)
-
-    def _set_state(self, state: McpConnectionState) -> None:
-        if self.state == state:
-            return
-        self.state = state
-        if self._status_sink is not None:
-            self._status_sink()
-
-    def _on_dead(self) -> None:
-        self.session = None
-        self._catalog = None
-        self._catalog_task = None
-        self._retry_after = time.monotonic() + _RETRY_DELAY_SECONDS
-        self._set_state("unavailable")
-
-    async def ensure_connected(self) -> SupervisedMcpSession:
-        """Return this entry's live session, connecting it once when needed."""
-        if self.session is not None:
-            return self.session
-        if self.connection_config is None:
-            raise McpConnectionUnavailableError(
-                f"MCP connection {self.name!r} is unavailable and cannot reconnect."
-            )
-        if self._connect_task is None:
-            if self.state == "unavailable" and time.monotonic() < self._retry_after:
-                raise McpConnectionUnavailableError(
-                    f"MCP connection {self.name!r} is temporarily unavailable."
-                )
-            self._connect_task = asyncio.create_task(
-                self._connect(),
-                name=f"mcp-connect-{self.name}",
-            )
-        task = self._connect_task
-        try:
-            return await asyncio.shield(task)
-        finally:
-            if self._connect_task is task and task.done():
-                self._connect_task = None
-
-    async def _connect(self) -> SupervisedMcpSession:
-        config = self.connection_config
-        if config is None:
-            raise McpConnectionUnavailableError(
-                f"MCP connection {self.name!r} has no connection configuration."
-            )
-        self._set_state("connecting")
-        session = SupervisedMcpSession(config)
-        try:
-            started = await session.start()
-        except BaseException:
-            with contextlib.suppress(BaseException):
-                await session.aclose()
-            self._retry_after = time.monotonic() + _RETRY_DELAY_SECONDS
-            self._set_state("unavailable")
-            raise
-        if not started:
-            await session.aclose()
-            self._retry_after = time.monotonic() + _RETRY_DELAY_SECONDS
-            self._set_state("unavailable")
-            raise McpConnectionUnavailableError(f"MCP connection {self.name!r} could not connect.")
-        self.session = session
-        session.set_on_dead(self._on_dead)
-        self._retry_after = 0.0
-        self._set_state("connected")
-        return session
-
-    async def ensure_catalog(self) -> list[Any]:
-        """Return the filtered catalog, listing it once on first use."""
-        if (
-            self._catalog is not None
-            and self.session is not None
-            and not self.session.is_dead
-            and not self.session.is_unavailable
-        ):
-            return self._catalog
-        if self.session is not None and (self.session.is_dead or self.session.is_unavailable):
-            self._catalog = None
-        if self._catalog_task is None:
-            self._catalog_task = asyncio.create_task(
-                self._load_catalog(),
-                name=f"mcp-catalog-{self.name}",
-            )
-        task = self._catalog_task
-        try:
-            return await asyncio.shield(task)
-        finally:
-            if self._catalog_task is task and task.done():
-                self._catalog_task = None
-
-    async def _load_catalog(self) -> list[Any]:
-        session = await self.ensure_connected()
-        self._set_state("catalog_loading")
-        try:
-            catalog = await session.list_tools()
-        except BaseException:
-            if session.is_dead:
-                self._set_state("unavailable")
-            else:
-                self._set_state("connected")
-            raise
-        self._catalog = list(catalog)
-        self.tool_count = len(self._catalog)
-        self._set_state("catalog_ready")
-        return self._catalog
-
-    async def close(self) -> None:
-        """Cancel pending initialization and close an opened session."""
-        tasks = [task for task in (self._catalog_task, self._connect_task) if task is not None]
-        for task in tasks:
-            task.cancel()
-        for task in tasks:
-            with contextlib.suppress(BaseException):
-                await task
-        self._catalog_task = None
-        self._connect_task = None
-        if self.session is not None:
-            with contextlib.suppress(BaseException):
-                await self.session.aclose()
-        self.session = None
+        """The session's reconnect config. Carries the bearer token; never log it."""
+        return self.session.config
 
 
 @dataclasses.dataclass(frozen=True)
@@ -264,7 +105,6 @@ class McpConnectionSummary:
     purpose: str | None
     tool_count: int
     provider: str | None = None
-    state: McpConnectionState = "configured"
 
 
 @dataclasses.dataclass(frozen=True)
@@ -283,7 +123,6 @@ class McpConnectionStatus:
     provider: str | None
     tool_count: int
     dead: bool
-    state: McpConnectionState
 
 
 @dataclasses.dataclass(frozen=True)
@@ -326,21 +165,6 @@ class McpRegistry:
 
     def __init__(self) -> None:
         self._entries: dict[str, McpConnectionEntry] = {}
-        self._status_sink: Callable[[], None] | None = None
-        self._warmup_task: asyncio.Task[None] | None = None
-
-    def register(self, request: McpConnectionRequest) -> McpConnectionEntry:
-        """Register an inert request without opening a network connection."""
-        entry = McpConnectionEntry(
-            name=request.config.name,
-            connection_config=request.config,
-            purpose=request.purpose or request.config.notes,
-            result_transform=request.result_transform,
-            provider=request.provider,
-        )
-        entry.set_status_sink(self._status_sink)
-        self._entries[entry.name] = entry
-        return entry
 
     def add(
         self,
@@ -367,16 +191,13 @@ class McpRegistry:
                 raise ValueError("McpRegistry.add requires either 'session' or 'server'")
             session = SupervisedMcpSession.adopt(server, name=name, config=config)
         entry = McpConnectionEntry(
-            name=name,
-            connection_config=config or session.config,
             session=session,
+            name=name,
             purpose=purpose,
             tool_count=tool_count,
             result_transform=result_transform,
             provider=provider,
-            state="connected",
         )
-        entry.set_status_sink(self._status_sink)
         self._entries[name] = entry
         return entry
 
@@ -396,7 +217,6 @@ class McpRegistry:
                 purpose=entry.purpose,
                 tool_count=entry.tool_count,
                 provider=entry.provider,
-                state=entry.state,
             )
             for entry in self._entries.values()
         ]
@@ -413,9 +233,7 @@ class McpRegistry:
                 name=entry.name,
                 provider=entry.provider,
                 tool_count=entry.tool_count,
-                dead=entry.state == "unavailable"
-                or (entry.session is not None and entry.session.is_dead),
-                state=entry.state,
+                dead=entry.session.is_dead,
             )
             for entry in self._entries.values()
         ]
@@ -425,42 +243,6 @@ class McpRegistry:
         runner)."""
         self._entries.clear()
 
-    def set_status_sink(self, sink: Callable[[], None] | None) -> None:
-        """Receive a callback after any connection lifecycle transition."""
-        self._status_sink = sink
-        for entry in self._entries.values():
-            entry.set_status_sink(sink)
-
-    def start_warmup(self, *, max_concurrency: int = 6) -> asyncio.Task[None]:
-        """Connect every configured entry in the background with a fixed bound."""
-        if self._warmup_task is not None:
-            return self._warmup_task
-
-        async def warm() -> None:
-            semaphore = asyncio.Semaphore(max(1, max_concurrency))
-
-            async def connect(entry: McpConnectionEntry) -> None:
-                async with semaphore:
-                    with contextlib.suppress(McpConnectionUnavailableError):
-                        await entry.ensure_connected()
-
-            await asyncio.gather(*(connect(entry) for entry in self._entries.values()))
-
-        self._warmup_task = asyncio.create_task(warm(), name="mcp-warmup")
-        return self._warmup_task
-
-    async def close(self) -> None:
-        """Stop warm-up and close only sessions this registry opened."""
-        if self._warmup_task is not None:
-            self._warmup_task.cancel()
-            with contextlib.suppress(BaseException):
-                await self._warmup_task
-            self._warmup_task = None
-        await asyncio.gather(
-            *(entry.close() for entry in self._entries.values()),
-            return_exceptions=True,
-        )
-
     def __len__(self) -> int:
         return len(self._entries)
 
@@ -500,6 +282,6 @@ def resolve_mcp_call(
         if entry is None:
             return None
         provider = entry.provider
-    raw_tool = args.get("tool") if tool_name in {CALL_MCP_TOOL, GET_MCP_TOOL_SCHEMA_TOOL} else ""
+    raw_tool = args.get("tool") if tool_name == CALL_MCP_TOOL else ""
     tool = raw_tool if isinstance(raw_tool, str) else ""
     return McpCallInfo(connection=connection, tool=tool, provider=provider)

strix/tools/reporting/tool.py

@@ -11,8 +11,8 @@ import asyncio
 import json
 import logging
 import re
-from pathlib import Path, PurePosixPath
-from typing import TYPE_CHECKING, Any, cast
+from pathlib import PurePosixPath
+from typing import TYPE_CHECKING, Any
 
 from agents import RunContextWrapper, function_tool
 
@@ -21,9 +21,6 @@ from strix.tools.proxy.tools import existing_request_ids
 
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping
-
-    from strix.fix.contracts import SourceIdentity
     from strix.report.state import ReportState
 
 
@@ -345,11 +342,16 @@ def _validate_fix_verification(
     if str(fix_verification or "").strip():
         return []
     return [
-        "fix_verification is REQUIRED when any code_location carries a 'fix_after'. "
-        "Describe the checks you performed on this draft candidate. Separate executed "
-        "checks from reasoned checks and list every gap. This statement does not make "
-        "the candidate ready for automatic application. If the candidate fails a check, "
-        "revise it or remove 'fix_after'."
+        "fix_verification is REQUIRED when any code_location carries a 'fix_after' - "
+        "a suggestion a reviewer can click to apply must be verified first. State, in "
+        "order: (1) security closure - re-trace the source->sink path through the "
+        "PATCHED code and say why it is now blocked; (2) bypass review - re-read the "
+        "diff without your original rationale and name the equivalent sinks, sibling "
+        "call sites, and alternate malicious input classes you checked; (3) preserved "
+        "behavior - the legitimate inputs, APIs, and error semantics that still work; "
+        "(4) how each was checked (executed vs. reasoned), naming any unrun check as "
+        "an explicit gap. If you cannot make these statements, drop 'fix_after' and "
+        "leave the location informational."
     ]
 
 
@@ -369,110 +371,6 @@ def _finding_class_of(report: dict[str, Any]) -> str:
     return "dynamic"
 
 
-def _fix_source_context(
-    report_state: ReportState,
-) -> tuple[SourceIdentity | None, Path | None]:
-    from strix.fix.contracts import SourceIdentity, SourceIdentityKind
-
-    raw_context = report_state.get_repository_context()
-    if raw_context is None:
-        return None, None
-    context = cast("dict[str, object]", raw_context)
-    commit = context.get("commitSha")
-    if not isinstance(commit, str) or not commit:
-        return None, None
-    raw_targets = cast("object", report_state.run_record.get("targets_info"))
-    targets = cast("list[object]", raw_targets) if isinstance(raw_targets, list) else []
-    if len(targets) != 1:
-        return SourceIdentity(kind=SourceIdentityKind.COMMIT, value=commit), None
-    raw_target = targets[0]
-    target = cast("dict[str, object]", raw_target) if isinstance(raw_target, dict) else {}
-    raw_details = target.get("details")
-    details = cast("dict[str, object]", raw_details) if isinstance(raw_details, dict) else {}
-    repo_path = details.get("cloned_repo_path")
-    repository = context.get("repositoryUri")
-    return (
-        SourceIdentity(
-            kind=SourceIdentityKind.COMMIT,
-            value=commit,
-            repository=repository if isinstance(repository, str) else None,
-        ),
-        Path(repo_path) if isinstance(repo_path, str) and repo_path else None,
-    )
-
-
-def _build_fix_candidate(
-    report_state: ReportState,
-    report_fields: Mapping[str, object],
-) -> dict[str, object] | None:
-    from strix.fix.contracts import candidate_from_legacy_report
-    from strix.fix.locations import AnchorStatus, anchor_candidate
-
-    source_identity, repo_path = _fix_source_context(report_state)
-    candidate = candidate_from_legacy_report(report_fields, source_identity=source_identity)
-    if candidate is None:
-        return None
-    if repo_path is not None:
-        try:
-            anchored, results = anchor_candidate(repo_path, candidate)
-        except Exception as exc:  # noqa: BLE001
-            # Anchoring must never block the finding from being stored.
-            candidate = candidate.model_copy(
-                update={"known_gaps": [*candidate.known_gaps, f"Candidate anchoring failed: {exc}"]}
-            )
-        else:
-            candidate = anchored
-            gaps = [
-                f"{result.location.file}: {result.status}"
-                for result in results
-                if result.status is not AnchorStatus.UNIQUE
-            ]
-            if gaps:
-                candidate = candidate.model_copy(
-                    update={"known_gaps": [*candidate.known_gaps, *gaps]}
-                )
-    return cast("dict[str, object]", candidate.model_dump(mode="json"))
-
-
-_FIX_CANDIDATE_FIELDS = frozenset(
-    {
-        "code_locations",
-        "remediation_steps",
-        "technical_analysis",
-        "poc_description",
-        "evidence",
-        "fix_verification",
-    }
-)
-
-
-def _refresh_fix_candidate(
-    report_state: ReportState,
-    report_id: str,
-    changes: dict[str, Any],
-) -> None:
-    if not _FIX_CANDIDATE_FIELDS.intersection(changes):
-        return
-    existing = next(
-        (
-            report
-            for report in report_state.get_existing_vulnerabilities()
-            if report.get("id") == report_id
-        ),
-        None,
-    )
-    if existing is None:
-        return
-    candidate = _build_fix_candidate(report_state, {**existing, **changes})
-    if candidate is not None:
-        changes["fix_candidate"] = candidate
-    if candidate is not None or existing.get("fix_candidate") or existing.get("fix_preparation"):
-        changes["fix_preparation"] = {
-            "state": "stale",
-            "stop_reason": "The finding or draft candidate changed after preparation.",
-        }
-
-
 _UPDATE_TEXT_FIELDS = (
     "title",
     "description",
@@ -753,7 +651,6 @@ def _do_update(
     class_error = _fit_revision_to_class(report_state, report_id, changes)
     if class_error is not None:
         return class_error
-    _refresh_fix_candidate(report_state, report_id, changes)
 
     try:
         updated = report_state.update_vulnerability_report(
@@ -1015,9 +912,6 @@ async def _do_create(
             "fix_pr_body": fix_pr_body,
             "http_exchange_ids": normalized_http_exchange_ids,
         }
-        fix_candidate = _build_fix_candidate(report_state, report_fields)
-        if fix_candidate:
-            report_fields["fix_candidate"] = fix_candidate
 
         dedupe = await check_duplicate(candidate, existing)
         if dedupe.get("is_duplicate"):
@@ -1365,10 +1259,11 @@ async def create_vulnerability_report(
             ``update_vulnerability_report`` once the proxy responds.
             Keep IDs out of ``evidence`` and all other report text.
 
-            **How ``fix_before`` / ``fix_after`` work**: they describe an
-            initial fix candidate. A later preparation stage applies,
-            tests, repairs, and independently verifies the candidate before
-            Strix can offer an automatic pull request. Therefore:
+            **How ``fix_before`` / ``fix_after`` work**: they're used as
+            literal GitHub/GitLab PR suggestion blocks. When a reviewer
+            accepts the suggestion, the platform replaces the **exact
+            lines from ``start_line`` to ``end_line``** with
+            ``fix_after``. Therefore:
 
             1. ``fix_before`` must be a **VERBATIM** copy of the source
                at those lines — same whitespace, indentation, line
@@ -1389,9 +1284,9 @@ async def create_vulnerability_report(
             before SQL"``). Order primary fix first, supporting
             changes (imports, config) after.
 
-            **Informational vs candidate**:
+            **Informational vs actionable**:
             - With ``fix_before`` / ``fix_after``: actionable fix
-              candidate for the preparation stage.
+              (renders as a PR suggestion block).
             - Without them: informational context (e.g. showing the
               source of tainted data, or a sink that doesn't need
               direct editing).
@@ -1426,10 +1321,11 @@ async def create_vulnerability_report(
               that aren't part of the fix.
             - Duplicating the same change across multiple locations.
         fix_verification: REQUIRED whenever any ``code_locations`` entry
-            carries a ``fix_after``. This field records the reporting
-            agent's checks on the draft candidate. It is not an independent
-            verification result. Before writing this field, work the gates
-            **in order**:
+            carries a ``fix_after``. A reviewer can apply that
+            suggestion with one click, so an unverified fix ships
+            straight into the codebase. Before writing this field, work
+            the gates **in order** and never trade an earlier one for a
+            later one:
 
             1. **Security closure** — re-trace the source → sink path
                through the *patched* code and state why it is now
@@ -1448,9 +1344,11 @@ async def create_vulnerability_report(
 
             Then write what you did: the commands you ran and their
             results, and every gate you could only reason about rather
-            than execute, marked explicitly as a gap. Do not claim that a
-            gate passed because the candidate looks correct. If a gate
-            fails, revise the candidate or drop ``fix_after``.
+            than execute, marked explicitly as a gap. Do not claim a
+            gate passed because it looks right. If a gate fails, revise
+            the patch or drop ``fix_after`` and leave the location
+            informational — never compensate for a failed security
+            closure with a smaller diff or extra prose.
 
             Also use this field to record the narrowest-complete-change
             judgement: prefer the smallest repository-native fix that

tests/test_disable_streaming.py

@@ -32,7 +32,6 @@ from openai.types.responses import (
 from strix.config import codex, loader
 from strix.config.loader import load_settings
 from strix.config.models import StrixProvider, _NonStreamingModel, _TurnGuardModel
-from strix.llm.request_log import RequestLoggingModel
 
 
 if TYPE_CHECKING:
@@ -313,8 +312,7 @@ def test_get_model_keeps_streaming_by_default(
 
     model = StrixProvider().get_model("openai/gpt-4o-mini")
     assert isinstance(model, _TurnGuardModel)
-    assert isinstance(model._inner, RequestLoggingModel)
-    assert model._inner._inner is inner
+    assert model._inner is inner
 
 
 def test_get_model_guards_subscription_model_but_keeps_it_streaming(

tests/test_execution_request_log.py

@@ -1,134 +0,0 @@
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, Any, cast
-
-import httpx
-import pytest
-from agents import RunConfig, Runner
-from openai import APIError, PermissionDeniedError
-
-from strix.core import execution
-from strix.core.agents import AgentCoordinator
-from strix.llm import request_log
-
-
-if TYPE_CHECKING:
-    from collections.abc import AsyncIterator
-
-
-def _request() -> httpx.Request:
-    return httpx.Request("POST", "https://api.openai.com/v1/responses")
-
-
-def _midstream_api_error() -> APIError:
-    return APIError("An error occurred while processing the request.", _request(), body=None)
-
-
-def _blocked_error() -> PermissionDeniedError:
-    response = httpx.Response(
-        403, request=_request(), headers={"x-request-id": "req_blocked_hdr"}, text="blocked"
-    )
-    return PermissionDeniedError("Output blocked by policy", response=response, body=None)
-
-
-class _FakeStream:
-    def __init__(self, exc: BaseException | None = None) -> None:
-        self._exc = exc
-        self.run_loop_exception: BaseException | None = None
-        self.seen_context: request_log.LlmCallContext | None = None
-
-    async def stream_events(self) -> AsyncIterator[Any]:
-        self.seen_context = request_log.current_call_context()
-        if self._exc is not None:
-            raise self._exc
-        items: tuple[Any, ...] = ()
-        for item in items:
-            yield item
-
-
-async def _run(
-    monkeypatch: pytest.MonkeyPatch,
-    streams: list[_FakeStream],
-    coordinator: AgentCoordinator | None = None,
-) -> Any:
-    monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0)
-    monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0)
-    calls = {"n": 0}
-
-    def _fake_run_streamed(*_args: Any, **_kwargs: Any) -> _FakeStream:
-        stream = streams[calls["n"]]
-        calls["n"] += 1
-        return stream
-
-    monkeypatch.setattr(Runner, "run_streamed", _fake_run_streamed)
-    coordinator = coordinator or AgentCoordinator()
-    await coordinator.register("root", "strix", parent_id=None)
-    return await execution._run_cycle(
-        object(),
-        coordinator,
-        "root",
-        input_data="task",
-        run_config=cast("RunConfig", object()),
-        context={},
-        max_turns=5,
-        session=None,
-        interactive=False,
-        event_sink=None,
-        hooks=None,
-    )
-
-
-@pytest.mark.asyncio
-async def test_each_transient_replay_is_stamped_with_its_attempt_number(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    streams = [
-        _FakeStream(exc=_midstream_api_error()),
-        _FakeStream(exc=_midstream_api_error()),
-        _FakeStream(),
-    ]
-    await _run(monkeypatch, streams)
-    assert [s.seen_context.retry_attempt for s in streams if s.seen_context] == [0, 1, 2]
-
-
-@pytest.mark.asyncio
-async def test_blocked_provider_failure_text_carries_request_id(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    coordinator = AgentCoordinator()
-    with pytest.raises(PermissionDeniedError):
-        await _run(monkeypatch, [_FakeStream(exc=_blocked_error())], coordinator)
-    assert coordinator.statuses["root"] == "failed"
-    error = coordinator.errors["root"]
-    assert "Output blocked by policy" in error
-    assert error.endswith("[provider request id: req_blocked_hdr]")
-
-
-@pytest.mark.asyncio
-async def test_run_agent_loop_binds_agent_context_and_resets(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    seen: dict[str, request_log.LlmCallContext] = {}
-
-    async def _fake_loop(**_kwargs: Any) -> None:
-        seen["ctx"] = request_log.current_call_context()
-
-    monkeypatch.setattr(execution, "_run_agent_loop", _fake_loop)
-
-    class _Agent:
-        name = "Recon Agent"
-
-    coordinator = AgentCoordinator()
-    await execution.run_agent_loop(
-        agent=_Agent(),
-        initial_input="task",
-        run_config=cast("RunConfig", object()),
-        context={},
-        max_turns=1,
-        coordinator=coordinator,
-        agent_id="agent-42",
-        interactive=False,
-    )
-    assert seen["ctx"].agent_id == "agent-42"
-    assert seen["ctx"].agent_name == "Recon Agent"
-    assert request_log.current_call_context().agent_id is None

tests/test_fix_preparation.py

@@ -1,854 +0,0 @@
-"""Tests for fix candidate anchoring and repository preparation."""
-
-from __future__ import annotations
-
-import hashlib
-import subprocess
-import sys
-from typing import TYPE_CHECKING
-
-import pytest
-
-from strix.fix.contracts import (
-    BlockerKind,
-    CandidateLocation,
-    CheckResult,
-    CheckStatus,
-    CommandSpec,
-    FixCandidateV1,
-    FixEdit,
-    FixPreparationRequestV1,
-    PreparationBlocker,
-    PreparationState,
-    RepairOutcome,
-    RepairStatus,
-    ReproductionSpec,
-    SourceIdentity,
-    SourceIdentityKind,
-    VerificationDecision,
-    VerificationTarget,
-    VerifierResult,
-    candidate_from_legacy_report,
-)
-from strix.fix.locations import AnchorStatus, anchor_location
-from strix.fix.prepare import (
-    PreparationContext,
-    _network_isolation_prefix,
-    build_git_manifest,
-    prepare_fix,
-    run_command,
-)
-
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
-
-def _git(workspace: Path, *args: str) -> str:
-    return subprocess.run(  # noqa: S603
-        ["/usr/bin/git", *args],
-        cwd=workspace,
-        check=True,
-        capture_output=True,
-        text=True,
-    ).stdout.strip()
-
-
-def _workspace(tmp_path: Path) -> tuple[Path, str]:
-    workspace = tmp_path / "repo"
-    workspace.mkdir()
-    _git(workspace, "init")
-    _git(workspace, "config", "user.email", "[email protected]")
-    _git(workspace, "config", "user.name", "Test")
-    (workspace / "app.py").write_text("def result():\n    return 'unsafe'\n", encoding="utf-8")
-    _git(workspace, "add", "app.py")
-    _git(workspace, "commit", "-m", "initial")
-    return workspace, _git(workspace, "rev-parse", "HEAD")
-
-
-def _candidate(
-    commit: str,
-    *,
-    reproduction: ReproductionSpec | None = None,
-) -> FixCandidateV1:
-    if reproduction is None:
-        reproduction = ReproductionSpec(
-            instructions="Confirm that result returns safe.",
-            command=CommandSpec(
-                name="security reproduction",
-                argv=[
-                    sys.executable,
-                    "-c",
-                    "from app import result; assert result() == 'safe'",
-                ],
-            ),
-        )
-    return FixCandidateV1(
-        source_identity=SourceIdentity(kind=SourceIdentityKind.COMMIT, value=commit),
-        security_invariant="Return a safe value.",
-        finding_locations=[
-            CandidateLocation(
-                file="app.py",
-                start_line=1,
-                end_line=2,
-                snippet="def result():\n    return 'unsafe'",
-            )
-        ],
-        draft_edits=[
-            FixEdit(
-                file="app.py",
-                start_line=2,
-                end_line=2,
-                before="    return 'unsafe'",
-                after="    return 'safe'",
-            )
-        ],
-        reproduction=reproduction,
-    )
-
-
-def _request(candidate: FixCandidateV1, *, attempts: int = 2) -> FixPreparationRequestV1:
-    return FixPreparationRequestV1(
-        scan_id="scan-1",
-        finding_id="finding-1",
-        candidate=candidate,
-        checks=[
-            CommandSpec(
-                name="compile",
-                argv=[
-                    sys.executable,
-                    "-c",
-                    "compile(open('app.py', encoding='utf-8').read(), 'app.py', 'exec')",
-                ],
-            )
-        ],
-        max_repair_attempts=attempts,
-    )
-
-
-async def _noop_repair(
-    context: PreparationContext,
-    _checks: list[CheckResult],
-) -> RepairOutcome:
-    path = context.workspace / "app.py"
-    if path.exists():
-        path.write_text(
-            path.read_text(encoding="utf-8").replace("'unsafe'", "'safe'"),
-            encoding="utf-8",
-        )
-    return RepairOutcome(
-        status=RepairStatus.COMPLETE,
-        summary="The repository fix is ready for independent evaluation.",
-    )
-
-
-async def _verified(
-    _context: PreparationContext,
-    _checks: list[CheckResult],
-) -> VerifierResult:
-    return VerifierResult(
-        decision=VerificationDecision.VERIFIED,
-        summary="The invariant is closed.",
-        security_invariant_closed=True,
-        reproduction_executed=True,
-        reproduction_summary="The vulnerable input is rejected.",
-        sibling_paths_reviewed=["app.py"],
-        preserved_behaviors=["The module compiles."],
-        security_tests=[
-            CheckResult(
-                name="security reproduction",
-                argv=[sys.executable, "-c", "assert True"],
-                status=CheckStatus.PASSED,
-                exit_code=0,
-                duration_seconds=0,
-                target=VerificationTarget.PATCHED,
-            )
-        ],
-    )
-
-
-def test_candidate_from_legacy_report_preserves_draft_and_checks() -> None:
-    candidate = candidate_from_legacy_report(
-        {
-            "remediation_steps": "Reject unsafe input.",
-            "poc_description": "Call the vulnerable function.",
-            "fix_verification": "Reasoned about the patched branch.",
-            "code_locations": [
-                {
-                    "file": "src/app.py",
-                    "start_line": 9,
-                    "end_line": 9,
-                    "snippet": "sink(value)",
-                    "fix_before": "sink(value)",
-                    "fix_after": "sink(clean(value))",
-                }
-            ],
-        }
-    )
-
-    assert candidate is not None
-    assert candidate.security_invariant == "Reject unsafe input."
-    assert candidate.draft_edits[0].after == "sink(clean(value))"
-    assert candidate.reported_checks[0].executed is False
-    assert candidate.known_gaps == ["The reporting-agent verification is not independent."]
-
-
-def test_anchor_location_rewrites_invented_line_numbers(tmp_path: Path) -> None:
-    (tmp_path / "app.py").write_text("first\nsecond\ntarget\nlast\n", encoding="utf-8")
-    location = CandidateLocation(
-        file="app.py",
-        start_line=99,
-        end_line=99,
-        snippet="target",
-    )
-
-    result = anchor_location(tmp_path, location)
-
-    assert result.status is AnchorStatus.UNIQUE
-    assert result.location.start_line == 3
-    assert result.location.end_line == 3
-
-
-@pytest.mark.parametrize(
-    ("content", "expected"),
-    [
-        ("first\nlast\n", AnchorStatus.MISSING),
-        ("target\nmiddle\ntarget\n", AnchorStatus.AMBIGUOUS),
-    ],
-)
-def test_anchor_location_rejects_non_unique_source(
-    tmp_path: Path,
-    content: str,
-    expected: AnchorStatus,
-) -> None:
-    (tmp_path / "app.py").write_text(content, encoding="utf-8")
-    edit = FixEdit(
-        file="app.py",
-        start_line=1,
-        end_line=1,
-        before="target",
-        after="safe",
-    )
-
-    assert anchor_location(tmp_path, edit).status is expected
-
-
-def test_anchor_location_detects_stale_file_digest(tmp_path: Path) -> None:
-    original = "target\n"
-    (tmp_path / "app.py").write_text("changed\n", encoding="utf-8")
-    edit = FixEdit(
-        file="app.py",
-        start_line=1,
-        end_line=1,
-        before="target",
-        after="safe",
-        original_sha256=hashlib.sha256(original.encode()).hexdigest(),
-    )
-
-    assert anchor_location(tmp_path, edit).status is AnchorStatus.STALE
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_returns_ready_with_manifest(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=_noop_repair,
-        verify=_verified,
-    )
-
-    assert result.state is PreparationState.READY
-    assert result.attempts == 1
-    assert [entry.path for entry in result.final_file_manifest] == ["app.py"]
-    assert result.security_reproduction is not None
-    assert result.security_reproduction.target is VerificationTarget.PATCHED
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_does_not_apply_draft_before_repair(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    observed_source: list[str] = []
-
-    async def repair(
-        context: PreparationContext,
-        _checks: list[CheckResult],
-    ) -> RepairOutcome:
-        observed_source.append((context.workspace / "app.py").read_text(encoding="utf-8"))
-        (context.workspace / "app.py").write_text(
-            "def result():\n    return 'safe'\n",
-            encoding="utf-8",
-        )
-        return RepairOutcome(
-            status=RepairStatus.COMPLETE,
-            summary="Implemented the repair from repository context.",
-        )
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=repair,
-        verify=_verified,
-    )
-
-    assert result.state is PreparationState.READY
-    assert observed_source == ["def result():\n    return 'unsafe'\n"]
-
-
-@pytest.mark.asyncio
-async def test_quality_gate_retries_once_before_verification(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    compile_calls = 0
-    repair_calls = 0
-    verification_calls = 0
-
-    async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
-        nonlocal compile_calls
-        compile_calls += 1
-        failed = compile_calls == 1
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.FAILED if failed else CheckStatus.PASSED,
-            exit_code=1 if failed else 0,
-            duration_seconds=0,
-            required=command.required,
-        )
-
-    async def repair(
-        context: PreparationContext,
-        checks: list[CheckResult],
-    ) -> RepairOutcome:
-        nonlocal repair_calls
-        repair_calls += 1
-        assert checks == [] if repair_calls == 1 else checks[0].status is CheckStatus.FAILED
-        (context.workspace / "app.py").write_text(
-            "def result():\n    return 'safe'\n",
-            encoding="utf-8",
-        )
-        return RepairOutcome(status=RepairStatus.COMPLETE, summary="Repair complete.")
-
-    async def verify(
-        context: PreparationContext,
-        checks: list[CheckResult],
-    ) -> VerifierResult:
-        nonlocal verification_calls
-        verification_calls += 1
-        return await _verified(context, checks)
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=repair,
-        verify=verify,
-        command_runner=runner,
-    )
-
-    assert result.state is PreparationState.READY
-    assert repair_calls == 2
-    assert compile_calls == 2
-    assert verification_calls == 1
-
-
-@pytest.mark.asyncio
-async def test_failed_quality_gate_never_runs_security_verifier(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    verifier_called = False
-
-    async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.FAILED,
-            exit_code=1,
-            duration_seconds=0,
-            required=command.required,
-        )
-
-    async def verify(
-        context: PreparationContext,
-        checks: list[CheckResult],
-    ) -> VerifierResult:
-        nonlocal verifier_called
-        verifier_called = True
-        return await _verified(context, checks)
-
-    result = await prepare_fix(
-        _request(_candidate(commit), attempts=1),
-        workspace,
-        repair=_noop_repair,
-        verify=verify,
-        command_runner=runner,
-    )
-
-    assert result.state is PreparationState.FAILED
-    assert verifier_called is False
-    assert "quality gate" in result.stop_reason
-
-
-@pytest.mark.asyncio
-async def test_repository_baseline_failure_is_a_typed_blocker(
-    tmp_path: Path,
-) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.FAILED,
-            exit_code=1,
-            duration_seconds=0,
-            output="existing failure",
-            required=command.required,
-            baseline_status=CheckStatus.FAILED,
-            baseline_output="existing failure",
-        )
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=_noop_repair,
-        verify=_verified,
-        command_runner=runner,
-    )
-
-    assert result.state is PreparationState.BLOCKED
-    assert result.blocker is not None
-    assert result.blocker.kind is BlockerKind.REPOSITORY_BASELINE
-    assert result.attempts == 1
-
-
-@pytest.mark.asyncio
-async def test_different_candidate_and_baseline_failures_are_repairable(
-    tmp_path: Path,
-) -> None:
-    workspace, commit = _workspace(tmp_path)
-    repairs = 0
-
-    async def repair(
-        _context: PreparationContext,
-        _feedback: list[CheckResult],
-    ) -> RepairOutcome:
-        nonlocal repairs
-        repairs += 1
-        return await _noop_repair(_context, _feedback)
-
-    async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.FAILED,
-            exit_code=1,
-            duration_seconds=0,
-            output="candidate-specific failure",
-            required=command.required,
-            baseline_status=CheckStatus.FAILED,
-            baseline_output="different pre-existing failure",
-        )
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=repair,
-        verify=_verified,
-        command_runner=runner,
-    )
-
-    assert result.state is PreparationState.FAILED
-    assert result.blocker is None
-    assert result.attempts == 2
-    assert repairs == 2
-
-
-@pytest.mark.asyncio
-async def test_unavailable_required_check_is_a_typed_blocker(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def runner(_workspace: Path, command: CommandSpec) -> CheckResult:
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.UNAVAILABLE,
-            duration_seconds=0,
-            output="compiler unavailable",
-            required=command.required,
-        )
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=_noop_repair,
-        verify=_verified,
-        command_runner=runner,
-    )
-
-    assert result.state is PreparationState.BLOCKED
-    assert result.blocker is not None
-    assert result.blocker.kind is BlockerKind.ENVIRONMENT
-    assert result.attempts == 1
-
-
-@pytest.mark.asyncio
-async def test_repairable_security_rejection_gets_one_retry(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    repair_feedback: list[list[str]] = []
-    verifier_calls = 0
-
-    async def repair(
-        context: PreparationContext,
-        _checks: list[CheckResult],
-    ) -> RepairOutcome:
-        repair_feedback.append(
-            [
-                gap
-                for attempt in context.feedback
-                if attempt.verifier is not None
-                for gap in attempt.verifier.gaps
-            ]
-        )
-        (context.workspace / "app.py").write_text(
-            "def result():\n    return 'safe'\n",
-            encoding="utf-8",
-        )
-        return RepairOutcome(status=RepairStatus.COMPLETE, summary="Repair complete.")
-
-    async def verify(
-        context: PreparationContext,
-        checks: list[CheckResult],
-    ) -> VerifierResult:
-        nonlocal verifier_calls
-        verifier_calls += 1
-        if verifier_calls == 1:
-            return VerifierResult(
-                decision=VerificationDecision.REJECTED,
-                summary="A sibling path remains vulnerable.",
-                gaps=["Harden the sibling path."],
-                repairable=True,
-            )
-        return await _verified(context, checks)
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=repair,
-        verify=verify,
-    )
-
-    assert result.state is PreparationState.READY
-    assert result.attempts == 2
-    assert repair_feedback == [[], ["Harden the sibling path."]]
-
-
-@pytest.mark.asyncio
-async def test_security_blocker_stops_without_repair_retry(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    repair_calls = 0
-    blocker = PreparationBlocker(
-        kind=BlockerKind.EXTERNAL_CONFIGURATION,
-        summary="A production account mapping is unavailable.",
-        user_action="Provide the production account mapping.",
-    )
-
-    async def repair(
-        context: PreparationContext,
-        checks: list[CheckResult],
-    ) -> RepairOutcome:
-        nonlocal repair_calls
-        repair_calls += 1
-        return await _noop_repair(context, checks)
-
-    async def verify(
-        _context: PreparationContext,
-        _checks: list[CheckResult],
-    ) -> VerifierResult:
-        return VerifierResult(
-            decision=VerificationDecision.INCONCLUSIVE,
-            summary=blocker.summary,
-            blocker=blocker,
-        )
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=repair,
-        verify=verify,
-    )
-
-    assert result.state is PreparationState.BLOCKED
-    assert result.blocker == blocker
-    assert repair_calls == 1
-
-
-@pytest.mark.asyncio
-async def test_budget_exhaustion_fails_without_verification(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    verifier_called = False
-
-    async def exhausted(
-        context: PreparationContext,
-        _checks: list[CheckResult],
-    ) -> RepairOutcome:
-        (context.workspace / "app.py").write_text(
-            "def result():\n    return 'safe'\n",
-            encoding="utf-8",
-        )
-        return RepairOutcome(
-            status=RepairStatus.BUDGET_EXHAUSTED,
-            summary="The repair hit its emergency turn ceiling after editing the patch.",
-            turns_used=40,
-        )
-
-    async def verify(
-        context: PreparationContext,
-        checks: list[CheckResult],
-    ) -> VerifierResult:
-        nonlocal verifier_called
-        verifier_called = True
-        return await _verified(context, checks)
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=exhausted,
-        verify=verify,
-    )
-
-    assert result.state is PreparationState.FAILED
-    assert verifier_called is False
-    assert result.changed_files == ["app.py"]
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_preserves_explicit_blocked_outcome(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    blocker = PreparationBlocker(
-        kind=BlockerKind.CREDENTIAL,
-        summary="A required credential is unavailable.",
-        user_action="Provide a repository-scoped test credential.",
-    )
-
-    async def blocked(
-        _context: PreparationContext,
-        _checks: list[CheckResult],
-    ) -> RepairOutcome:
-        return RepairOutcome(
-            status=RepairStatus.BLOCKED,
-            summary=blocker.summary,
-            blocker=blocker,
-        )
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=blocked,
-        verify=_verified,
-    )
-
-    assert result.state is PreparationState.BLOCKED
-    assert result.blocker == blocker
-    assert result.verifier is None
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_requires_verifier_security_test(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def verify_without_test(
-        _context: PreparationContext,
-        _checks: list[CheckResult],
-    ) -> VerifierResult:
-        return VerifierResult(
-            decision=VerificationDecision.VERIFIED,
-            summary="The verifier did not execute the issue-specific test.",
-            security_invariant_closed=True,
-            reproduction_executed=False,
-        )
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=_noop_repair,
-        verify=verify_without_test,
-    )
-
-    assert result.state is PreparationState.FAILED
-    assert "did not execute a security test" in " ".join(result.gaps)
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_requires_closed_security_invariant(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def incomplete(
-        context: PreparationContext,
-        checks: list[CheckResult],
-    ) -> VerifierResult:
-        verified = await _verified(context, checks)
-        return verified.model_copy(update={"security_invariant_closed": False})
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=_noop_repair,
-        verify=incomplete,
-    )
-
-    assert result.state is PreparationState.FAILED
-    assert "invariant" in " ".join(result.gaps)
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_requires_a_source_change(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def no_change(
-        _context: PreparationContext,
-        _checks: list[CheckResult],
-    ) -> RepairOutcome:
-        return RepairOutcome(status=RepairStatus.COMPLETE, summary="No change.")
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=no_change,
-        verify=_verified,
-    )
-
-    assert result.state is PreparationState.FAILED
-    assert result.final_file_manifest == []
-    assert "change repository source" in result.stop_reason
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_rejects_wrong_source_commit(tmp_path: Path) -> None:
-    workspace, _commit = _workspace(tmp_path)
-
-    result = await prepare_fix(
-        _request(_candidate("0" * 40)),
-        workspace,
-        repair=_noop_repair,
-        verify=_verified,
-    )
-
-    assert result.state is PreparationState.STALE
-    assert result.blocker is not None
-    assert result.blocker.kind is BlockerKind.SOURCE
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_does_not_apply_escaping_draft_edit(tmp_path: Path) -> None:
-    workspace, _commit = _workspace(tmp_path)
-    outside = tmp_path / "outside.txt"
-    outside.write_text("target\n", encoding="utf-8")
-    (workspace / "link.txt").symlink_to(outside)
-    _git(workspace, "add", "link.txt")
-    _git(workspace, "commit", "-m", "add link")
-    candidate = _candidate(_git(workspace, "rev-parse", "HEAD")).model_copy(
-        update={
-            "draft_edits": [
-                FixEdit(
-                    file="link.txt",
-                    start_line=1,
-                    end_line=1,
-                    before="target",
-                    after="safe",
-                )
-            ]
-        }
-    )
-
-    result = await prepare_fix(
-        _request(candidate),
-        workspace,
-        repair=_noop_repair,
-        verify=_verified,
-    )
-
-    assert result.state is PreparationState.READY
-    assert outside.read_text(encoding="utf-8") == "target\n"
-    assert "link.txt" not in result.changed_files
-
-
-@pytest.mark.asyncio
-async def test_prepare_fix_rejects_dirty_workspace(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    (workspace / "stray.txt").write_text("unrelated\n", encoding="utf-8")
-
-    result = await prepare_fix(
-        _request(_candidate(commit)),
-        workspace,
-        repair=_noop_repair,
-        verify=_verified,
-    )
-
-    assert result.state is PreparationState.STALE
-
-
-@pytest.mark.asyncio
-async def test_build_git_manifest_lists_files_inside_new_directory(tmp_path: Path) -> None:
-    workspace, _commit = _workspace(tmp_path)
-    package = workspace / "pkg"
-    package.mkdir()
-    (package / "mod.py").write_text("x = 1\n", encoding="utf-8")
-
-    entries, _summary, _artifact = await build_git_manifest(workspace)
-
-    paths = {entry.path for entry in entries}
-    assert "pkg/mod.py" in paths
-    entry = next(entry for entry in entries if entry.path == "pkg/mod.py")
-    assert entry.operation == "add"
-    assert entry.resulting_sha256 is not None
-
-
-@pytest.mark.asyncio
-async def test_run_command_drops_ambient_credentials(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    monkeypatch.setenv("STRIX_AMBIENT_TOKEN", "hunter2")
-    command = CommandSpec(
-        name="env probe",
-        argv=[
-            sys.executable,
-            "-c",
-            "import os; print(os.environ.get('STRIX_AMBIENT_TOKEN', '<absent>'))",
-        ],
-    )
-
-    sealed = await run_command(tmp_path, command, network_allowed=True)
-    assert sealed.status is CheckStatus.PASSED
-    assert "<absent>" in sealed.output
-
-    granted = await run_command(
-        tmp_path,
-        command,
-        credentials_allowed=["STRIX_AMBIENT_TOKEN"],
-        network_allowed=True,
-    )
-    assert granted.status is CheckStatus.PASSED
-    assert "hunter2" in granted.output
-
-
-@pytest.mark.asyncio
-async def test_run_command_blocks_egress_when_network_not_allowed(tmp_path: Path) -> None:
-    command = CommandSpec(
-        name="egress probe",
-        argv=[
-            sys.executable,
-            "-c",
-            (
-                "import socket, sys; s = socket.socket(); s.settimeout(3); "
-                "sys.exit(0 if s.connect_ex(('1.1.1.1', 53)) == 0 else 1)"
-            ),
-        ],
-    )
-
-    result = await run_command(tmp_path, command)
-
-    if _network_isolation_prefix() is None:
-        assert result.status is CheckStatus.UNAVAILABLE
-        assert "not run" in result.output
-    else:
-        assert result.status is CheckStatus.FAILED

tests/test_mcp_client.py

@@ -36,15 +36,12 @@ from strix.tools.mcp import (
     attach_mcp_requests,
     call_mcp,
     describe_mcp,
-    get_mcp_tool_schema,
     list_mcps,
     load_user_mcp_configs,
     namespaced_tool_name,
     resolve_mcp_call,
-    search_mcp_tools,
 )
 from strix.tools.mcp import client as mcp_client
-from strix.tools.mcp import registry as mcp_registry_mod
 from strix.tools.mcp import session as mcp_session_mod
 
 
@@ -437,16 +434,8 @@ async def test_list_mcps_returns_connections_with_ids_and_descriptions() -> None
                 "description": "local files",
                 "tool_count": 2,
                 "dead": False,
-                "state": "connected",
-            },
-            {
-                "id": "db",
-                "name": "db",
-                "description": None,
-                "tool_count": 1,
-                "dead": False,
-                "state": "connected",
             },
+            {"id": "db", "name": "db", "description": None, "tool_count": 1, "dead": False},
         ]
     }
 
@@ -496,247 +485,6 @@ async def test_describe_mcp_without_any_connections() -> None:
     assert out == "No MCP connections are configured for this run."
 
 
-@pytest.mark.asyncio
-async def test_search_then_get_one_schema_without_returning_other_schemas() -> None:
-    registry = McpRegistry()
-    registry.add(
-        name="docs",
-        server=FakeMCPServer(
-            "docs",
-            [
-                _mcp_tool("find_pages"),
-                MCPTool(
-                    name="fetch_page",
-                    description="Fetch page content",
-                    inputSchema={
-                        "type": "object",
-                        "properties": {"page_id": {"type": "string"}},
-                    },
-                ),
-            ],
-        ),
-        config=McpConnectionConfig(
-            name="docs",
-            url="https://example.invalid/mcp",
-            active_tools=["fetch_page"],
-        ),
-    )
-
-    searched = await search_mcp_tools.on_invoke_tool(
-        _ctx(registry),
-        json.dumps({"connection": "docs", "query": "page content"}),
-    )
-    assert [match["name"] for match in searched["matches"]] == [
-        "fetch_page",
-        "find_pages",
-    ]
-    assert all("input_schema" not in match for match in searched["matches"])
-
-    schema = await get_mcp_tool_schema.on_invoke_tool(
-        _ctx(registry),
-        json.dumps({"connection": "docs", "tool": "fetch_page"}),
-    )
-    assert schema["input_schema"]["properties"] == {"page_id": {"type": "string"}}
-
-
-@pytest.mark.asyncio
-async def test_search_matches_any_query_term_and_prioritizes_active_tools() -> None:
-    registry = McpRegistry()
-    registry.add(
-        name="docs",
-        server=FakeMCPServer(
-            "docs",
-            [
-                MCPTool(
-                    name="fetch_page",
-                    description="Read page content",
-                    inputSchema={"type": "object"},
-                ),
-                MCPTool(
-                    name="search_documents",
-                    description="Search documents",
-                    inputSchema={"type": "object"},
-                ),
-                MCPTool(
-                    name="list_records",
-                    description="List records",
-                    inputSchema={"type": "object"},
-                ),
-            ],
-        ),
-        config=McpConnectionConfig(
-            name="docs",
-            url="https://example.invalid/mcp",
-            active_tools=["fetch_page"],
-        ),
-    )
-
-    searched = await search_mcp_tools.on_invoke_tool(
-        _ctx(registry),
-        json.dumps(
-            {
-                "connection": "docs",
-                "query": "read list search document",
-            }
-        ),
-    )
-
-    assert [match["name"] for match in searched["matches"]] == [
-        "fetch_page",
-        "search_documents",
-        "list_records",
-    ]
-
-
-@pytest.mark.asyncio
-async def test_search_returns_active_tools_when_text_does_not_match() -> None:
-    registry = McpRegistry()
-    registry.add(
-        name="docs",
-        server=FakeMCPServer(
-            "docs",
-            [
-                _mcp_tool("fetch_page"),
-                _mcp_tool("search_documents"),
-            ],
-        ),
-        config=McpConnectionConfig(
-            name="docs",
-            url="https://example.invalid/mcp",
-            active_tools=["fetch_page"],
-        ),
-    )
-
-    searched = await search_mcp_tools.on_invoke_tool(
-        _ctx(registry),
-        json.dumps(
-            {
-                "connection": "docs",
-                "query": "unrelated capability",
-            }
-        ),
-    )
-
-    assert searched["matches"] == [
-        {
-            "name": "fetch_page",
-            "description": "remote tool fetch_page",
-            "active": True,
-        }
-    ]
-
-
-@pytest.mark.asyncio
-async def test_registered_entry_connects_and_lists_once_for_concurrent_catalog_requests(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    calls = {"build": 0, "list": 0}
-
-    class _CountingServer(FakeMCPServer):
-        async def list_tools(
-            self,
-            run_context: Any = None,
-            agent: Any = None,
-        ) -> list[MCPTool]:
-            calls["list"] += 1
-            await asyncio.sleep(0)
-            return await super().list_tools(run_context, agent)
-
-    server = _CountingServer("docs", [_mcp_tool("fetch_page")])
-
-    def _build(_config: McpConnectionConfig) -> mcp_client.BuiltMcpServer:
-        calls["build"] += 1
-        return _built_server(server)
-
-    monkeypatch.setattr(mcp_client, "_build_server", _build)
-    registry = McpRegistry()
-    entry = registry.register(McpConnectionRequest(config=_config("docs", ["fetch_page"])))
-    assert entry.state == "configured"
-    assert calls == {"build": 0, "list": 0}
-
-    first, second = await asyncio.gather(entry.ensure_catalog(), entry.ensure_catalog())
-
-    assert [tool.name for tool in first] == ["fetch_page"]
-    assert second is first
-    assert calls == {"build": 1, "list": 1}
-    assert registry.statuses()[0].state == "catalog_ready"
-    await registry.close()
-
-
-@pytest.mark.asyncio
-async def test_registered_entry_replaces_a_terminally_dead_session(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    config = _config("docs", ["fetch_page"])
-    original = SupervisedMcpSession.adopt(
-        FakeMCPServer("docs", [_mcp_tool("fetch_page")]),
-        name="docs",
-        config=config,
-    )
-    replacement_server = FakeMCPServer("docs", [_mcp_tool("fetch_page")])
-    monkeypatch.setattr(
-        mcp_client,
-        "_build_server",
-        lambda _config: _built_server(replacement_server),
-    )
-    monkeypatch.setattr(mcp_registry_mod, "_RETRY_DELAY_SECONDS", 0)
-    registry = McpRegistry()
-    entry = registry.add(name="docs", session=original, config=config)
-
-    original._mark_dead()
-
-    assert entry.session is None
-    replacement = await entry.ensure_connected()
-    assert replacement is not original
-    assert replacement.server is replacement_server
-    assert entry.state == "connected"
-    await registry.close()
-
-
-@pytest.mark.asyncio
-async def test_registry_warmup_bounds_parallel_connections(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    active = 0
-    maximum = 0
-    two_started = asyncio.Event()
-    release = asyncio.Event()
-
-    class _BlockedConnectServer(FakeMCPServer):
-        async def connect(self) -> None:
-            nonlocal active, maximum
-            active += 1
-            maximum = max(maximum, active)
-            if active == 2:
-                two_started.set()
-            try:
-                await release.wait()
-            finally:
-                active -= 1
-
-    servers = {
-        f"docs-{index}": _BlockedConnectServer(f"docs-{index}", [_mcp_tool("fetch_page")])
-        for index in range(4)
-    }
-    monkeypatch.setattr(
-        mcp_client,
-        "_build_server",
-        lambda config: _built_server(servers[config.name]),
-    )
-    registry = McpRegistry()
-    for name in servers:
-        registry.register(McpConnectionRequest(config=_config(name, ["fetch_page"])))
-
-    warmup = registry.start_warmup(max_concurrency=2)
-    await asyncio.wait_for(two_started.wait(), timeout=1)
-    assert maximum == 2
-    release.set()
-    await warmup
-
-    assert [summary.state for summary in registry.summaries()] == ["connected"] * 4
-    await registry.close()
-
-
 # --- call_mcp ----------------------------------------------------------------
 
 
@@ -825,8 +573,7 @@ async def test_call_mcp_errors_on_unknown_tool() -> None:
     )
 
     assert "Unknown tool 'delete_everything'" in out
-    assert "search_mcp_tools" in out
-    assert "read_file" not in out
+    assert "read_file" in out
     # A rejected tool name never reaches the server.
     assert server.calls == []
 
@@ -884,27 +631,21 @@ async def test_call_mcp_flags_an_errored_result_failed_for_the_tui() -> None:
     assert out == {"type": "text", "text": "boom:read_file", "success": False}
 
 
-# --- generic MCP tools are the only MCP surface every agent gets -------------
+# --- the two tools are the only MCP surface every agent gets -----------------
 
 
-def test_agent_carries_exactly_the_generic_mcp_tools_regardless_of_connections() -> None:
+def test_agent_carries_exactly_the_dispatch_tools_regardless_of_connections() -> None:
     """No matter how many MCP connections a run makes, an agent's tool list gains
-    exactly the generic MCP tools and never a per-connection provider tool."""
+    exactly list_mcps, describe_mcp, and call_mcp and never a per-connection
+    provider tool."""
     root = factory.build_strix_agent(is_root=True)
     child = factory.build_strix_agent(is_root=False)
 
     root_names = [t.name for t in root.tools]
     child_names = [t.name for t in child.tools]
 
-    expected = {
-        "list_mcps",
-        "search_mcp_tools",
-        "get_mcp_tool_schema",
-        "describe_mcp",
-        "call_mcp",
-    }
-    assert expected <= set(root_names)
-    assert expected <= set(child_names)
+    assert {"list_mcps", "describe_mcp", "call_mcp"} <= set(root_names)
+    assert {"list_mcps", "describe_mcp", "call_mcp"} <= set(child_names)
 
     # Five hypothetical connections would once have added ~all their tools as
     # namespaced provider tools; none of those names may appear now.
@@ -925,25 +666,16 @@ def test_agent_carries_exactly_the_generic_mcp_tools_regardless_of_connections()
 # --- prompt guidance replaces the old per-connection inventory ---------------
 
 
-def test_prompt_renders_targeted_tool_guidance_when_mcp_available() -> None:
+def test_prompt_renders_static_three_tool_guidance_when_mcp_available() -> None:
     prompt = render_system_prompt(system_prompt_context={"mcp_available": True})
 
     assert "MCP CONNECTIONS" in prompt
-    # Targeted discovery, one-schema lookup, dispatch, and compatibility are named.
+    # The three discovery/dispatch tools are named as the way in.
     assert "list_mcps" in prompt
-    assert "search_mcp_tools" in prompt
-    assert "get_mcp_tool_schema" in prompt
     assert "describe_mcp" in prompt
     assert "call_mcp" in prompt
 
 
-def test_model_facing_mcp_guidance_uses_targeted_discovery() -> None:
-    assert "describe_mcp" not in list_mcps.description
-    assert "describe_mcp" not in call_mcp.description
-    assert "full tool catalog" in describe_mcp.description
-    assert "compatibility fallback" in describe_mcp.description
-
-
 def test_prompt_has_no_mcp_section_without_availability() -> None:
     assert "MCP CONNECTIONS" not in render_system_prompt(system_prompt_context={})
 
@@ -951,7 +683,7 @@ def test_prompt_has_no_mcp_section_without_availability() -> None:
 def test_prompt_renders_named_connection_inventory() -> None:
     """With mcp_available set, the prompt names each connected server (name, tool
     count, purpose) so every agent sees what is available at the start, alongside
-    the discovery and dispatch tools for re-listing and inspecting them at run time."""
+    the three dispatch tools for re-listing and inspecting them at run time."""
     prompt = render_system_prompt(
         system_prompt_context={
             "mcp_available": True,

tests/test_models.py

@@ -23,7 +23,6 @@ from strix.config.models import (
     uses_chat_completions_tool_schema,
 )
 from strix.config.settings import Settings
-from strix.llm.request_log import RequestLoggingModel
 
 
 @pytest.mark.parametrize("model_name", RECOMMENDED_MODEL_NAMES)
@@ -173,7 +172,7 @@ def test_routes_through_litellm_matches_the_provider(
         # proves the route is not LiteLLM's.
         assert not litellm
         return
-    while isinstance(model, _NonStreamingModel | _TurnGuardModel | RequestLoggingModel):
+    while isinstance(model, _NonStreamingModel | _TurnGuardModel):
         model = model._inner
     assert isinstance(model, LitellmModel) is litellm
 
@@ -216,6 +215,6 @@ def test_api_type_overrides_the_api_base_route(
         monkeypatch.setenv("STRIX_API_TYPE", api_type)
     configure_sdk_model_defaults(Settings())
     model = StrixProvider().get_model("gpt-5")
-    while isinstance(model, _NonStreamingModel | _TurnGuardModel | RequestLoggingModel):
+    while isinstance(model, _NonStreamingModel | _TurnGuardModel):
         model = model._inner
     assert isinstance(model, expected)

tests/test_report_writer.py

@@ -264,4 +264,4 @@ def test_render_vulnerability_md_surfaces_calibration_metadata() -> None:
     assert "Egress appears filtered at the network layer." in md
     assert "## Confidence Rationale" in md
     assert "## What Would Change This Severity" in md
-    assert "## Reported Candidate Checks" in md
+    assert "## Fix Verification" in md

tests/test_reporting_fields.py

@@ -1981,30 +1981,6 @@ def test_update_refuses_code_locations_it_cannot_use(report_state: ReportState)
     assert any("start_line" in error for error in result["errors"])
 
 
-def test_update_marks_preparation_stale_when_candidate_cannot_be_rebuilt(
-    report_state: ReportState,
-) -> None:
-    """A revision whose locations cannot form a candidate still supersedes the
-    preparation recorded for the draft it replaced."""
-    _seed_weak_report(report_state)
-    report = report_state.vulnerability_reports[0]
-    candidate = {"security_invariant": "Parametrize the query.", "draft_edits": []}
-    report["fix_candidate"] = candidate
-    report["fix_preparation"] = {"state": "ready", "candidate_digest": "0" * 64}
-
-    result = _do_update(
-        report_id="vuln-0009",
-        update_reason="Repointing the finding to the validated handler.",
-        fields={"code_locations": [{"file": "./files.py", "start_line": 4, "end_line": 9}]},
-    )
-
-    assert result["success"] is True
-    assert report["code_locations"] == [{"file": "./files.py", "start_line": 4, "end_line": 9}]
-    assert report["fix_preparation"]["state"] == "stale"
-    assert "changed after preparation" in report["fix_preparation"]["stop_reason"]
-    assert report["fix_candidate"] == candidate
-
-
 def _seed_saved_report(report_state: ReportState) -> Path:
     """The weak report, written to disk the way a filed finding is."""
     _seed_weak_report(report_state)

tests/test_runner_mcp.py

@@ -80,14 +80,13 @@ async def test_none_default_attaches_from_the_user_config_file(
     )
     monkeypatch.setattr(mcp_pkg, "load_user_mcp_configs", lambda: [file_config])
 
-    captured: list[McpConnectionRequest] = []
-    original_register = mcp_pkg.McpRegistry.register
+    captured: list[list[McpConnectionRequest]] = []
 
-    def _capture(registry: Any, request: McpConnectionRequest) -> Any:
-        captured.append(request)
-        return original_register(registry, request)
+    async def _capture(requests: list[McpConnectionRequest], _registry: Any) -> list[Any]:
+        captured.append(requests)
+        return []
 
-    monkeypatch.setattr(mcp_pkg.McpRegistry, "register", _capture)
+    monkeypatch.setattr(mcp_pkg, "attach_mcp_requests", _capture)
 
     await runner.run_strix_scan(
         scan_config={"targets": [], "scan_mode": "deep"},
@@ -98,11 +97,12 @@ async def test_none_default_attaches_from_the_user_config_file(
 
     # Each config from the file is wrapped in a bare request: no provider, no
     # transform, no explicit purpose (purpose falls back to notes at attach time).
-    (request,) = captured
-    assert request.config is file_config
-    assert request.provider is None
-    assert request.result_transform is None
-    assert request.purpose is None
+    (requests,) = captured
+    assert len(requests) == 1
+    assert requests[0].config is file_config
+    assert requests[0].provider is None
+    assert requests[0].result_transform is None
+    assert requests[0].purpose is None
 
 
 @pytest.mark.asyncio
@@ -116,14 +116,13 @@ async def test_supplied_requests_are_attached_and_the_user_file_is_not_read(
 
     monkeypatch.setattr(mcp_pkg, "load_user_mcp_configs", _fail_if_read)
 
-    captured: list[McpConnectionRequest] = []
-    original_register = mcp_pkg.McpRegistry.register
+    captured: list[list[McpConnectionRequest]] = []
 
-    def _capture(registry: Any, request: McpConnectionRequest) -> Any:
-        captured.append(request)
-        return original_register(registry, request)
+    async def _capture(requests: list[McpConnectionRequest], _registry: Any) -> list[Any]:
+        captured.append(requests)
+        return []
 
-    monkeypatch.setattr(mcp_pkg.McpRegistry, "register", _capture)
+    monkeypatch.setattr(mcp_pkg, "attach_mcp_requests", _capture)
 
     supplied = [
         McpConnectionRequest(
@@ -140,7 +139,7 @@ async def test_supplied_requests_are_attached_and_the_user_file_is_not_read(
         mcp_connection_requests=supplied,
     )
 
-    assert captured == supplied
+    assert captured == [supplied]
 
 
 @pytest.mark.asyncio
@@ -148,8 +147,8 @@ async def test_roster_is_persisted_even_without_a_status_sink(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Any
 ) -> None:
     """The viewer reads the roster off disk, so persistence must not depend on the
-    interface status sink: with ``mcp_status_sink=None`` the configured roster is
-    still written, carrying only non-secret lifecycle fields."""
+    interface status sink: with ``mcp_status_sink=None`` the connect-time roster is
+    still written, carrying only the non-secret name/provider/tool_count/dead."""
     _wire_runner(monkeypatch, tmp_path)
     monkeypatch.setattr(
         mcp_pkg,
@@ -157,6 +156,19 @@ async def test_roster_is_persisted_even_without_a_status_sink(
         lambda: [McpConnectionConfig(name="local_fs", transport="stdio", command="npx")],
     )
 
+    class _FakeSession:
+        is_dead = False
+
+        def set_on_dead(self, _callback: Any) -> None:
+            return None
+
+    async def _attach(_requests: list[McpConnectionRequest], registry: Any) -> list[Any]:
+        registry.add(name="local_fs", session=_FakeSession(), tool_count=3, provider=None)
+        entry = registry.get("local_fs")
+        return [types.SimpleNamespace(name="local_fs", tool_count=3, session=entry.session)]
+
+    monkeypatch.setattr(mcp_pkg, "attach_mcp_requests", _attach)
+
     persisted: list[list[dict[str, Any]]] = []
 
     def _capture_persist(roster: list[dict[str, Any]]) -> None:
@@ -173,12 +185,4 @@ async def test_roster_is_persisted_even_without_a_status_sink(
     )
 
     assert persisted, "roster must persist even when no status sink is attached"
-    assert persisted[0] == [
-        {
-            "name": "local_fs",
-            "provider": None,
-            "tool_count": 0,
-            "dead": False,
-            "state": "configured",
-        }
-    ]
+    assert persisted[-1] == [{"name": "local_fs", "provider": None, "tool_count": 3, "dead": False}]

tests/test_runner_root_prompt.py

@@ -194,12 +194,21 @@ async def test_mcp_available_flag_set_when_a_connection_attaches(
     scope_context: dict[str, Any] = {"scope": "built-in"}
     captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
 
+    async def _aclose() -> None:
+        return None
+
+    async def _attach(_requests: Any, registry: Any) -> list[Any]:
+        registry.add(name="fs", server=object(), purpose="local files", tool_count=2)
+        session = types.SimpleNamespace(aclose=_aclose)
+        return [types.SimpleNamespace(name="fs", tool_count=2, session=session)]
+
+    monkeypatch.setattr(mcp_pkg, "attach_mcp_requests", _attach)
+
     request = McpConnectionRequest(
         config=McpConnectionConfig(
             name="fs",
             url="https://mcp.example.com",
             auth=BearerAuth(token="run-token"),
-            notes="local files",
         )
     )
 
@@ -215,7 +224,7 @@ async def test_mcp_available_flag_set_when_a_connection_attaches(
     assert kwargs["system_prompt_context"]["mcp_available"] is True
     # The named inventory names each connected server for the prompt.
     assert kwargs["system_prompt_context"]["mcp_connections"] == [
-        {"name": "fs", "purpose": "local files", "tool_count": 0}
+        {"name": "fs", "purpose": "local files", "tool_count": 2}
     ]
 
 

tests/test_sarif.py

@@ -5,7 +5,6 @@ from __future__ import annotations
 import json
 from typing import TYPE_CHECKING, Any
 
-from strix.fix.contracts import FixCandidateV1
 from strix.report.sarif import write_sarif
 
 
@@ -130,37 +129,14 @@ def test_write_sarif_never_embeds_poc_script(tmp_path: Path) -> None:
     assert poc["description"] == "Send a crafted request to trigger the sink."
 
 
-def _fix_candidate(**overrides: Any) -> dict[str, Any]:
-    candidate: dict[str, Any] = {
-        "security_invariant": "Use a parameterized query.",
-        "draft_edits": [
-            {
-                "file": "app.py",
-                "start_line": 4,
-                "end_line": 4,
-                "before": 'query = "SELECT * FROM u WHERE id=" + uid',
-                "after": 'query = "SELECT * FROM u WHERE id=%s"',
-            }
-        ],
-    }
-    candidate.update(overrides)
-    return candidate
-
-
-def test_write_sarif_builds_fixes_from_ready_preparation(tmp_path: Path) -> None:
-    # SARIF fixes are automatically applicable, so only a prepared result can
-    # expose the verified candidate as an artifactChange.
-    candidate = _fix_candidate()
+def test_write_sarif_builds_fixes_from_code_location_fix_pairs(tmp_path: Path) -> None:
+    # A code location carrying fix_before/fix_after must surface as a SARIF
+    # fix (artifactChange/replacement) so consumers can offer a one-click fix.
     write_sarif(
         tmp_path,
         [
             _finding(
                 remediation_steps="Use a parameterized query.",
-                fix_candidate=candidate,
-                fix_preparation={
-                    "state": "ready",
-                    "candidate_digest": FixCandidateV1.model_validate(candidate).digest(),
-                },
                 code_locations=[
                     {
                         "file": "app.py",
@@ -183,91 +159,6 @@ def test_write_sarif_builds_fixes_from_ready_preparation(tmp_path: Path) -> None
     assert replacement["insertedContent"]["text"] == 'query = "SELECT * FROM u WHERE id=%s"'
 
 
-def test_write_sarif_emits_verified_candidate_from_preparation(tmp_path: Path) -> None:
-    # Preparation re-anchors the draft to the lines its block actually occupies,
-    # so the result carries the verified candidate and SARIF emits those lines
-    # rather than the stale positions the report's draft still records.
-    verified = _fix_candidate(
-        draft_edits=[
-            {
-                "file": "app.py",
-                "start_line": 9,
-                "end_line": 9,
-                "before": 'query = "SELECT * FROM u WHERE id=" + uid',
-                "after": 'query = "SELECT * FROM u WHERE id=%s"',
-            }
-        ]
-    )
-    write_sarif(
-        tmp_path,
-        [
-            _finding(
-                fix_candidate=_fix_candidate(),
-                fix_preparation={
-                    "state": "ready",
-                    "candidate": verified,
-                    "candidate_digest": FixCandidateV1.model_validate(verified).digest(),
-                },
-            )
-        ],
-    )
-    result = _read(tmp_path)["runs"][0]["results"][0]
-    replacement = result["fixes"][0]["artifactChanges"][0]["replacements"][0]
-    assert replacement["deletedRegion"]["startLine"] == 9
-    assert replacement["deletedRegion"]["endLine"] == 9
-
-
-def test_write_sarif_suppresses_fixes_when_prepared_candidate_diverged(
-    tmp_path: Path,
-) -> None:
-    # A ready result whose digest no longer matches the recorded candidate
-    # means repair changed what was verified — emitting the draft would offer
-    # the wrong replacement, so SARIF stays silent.
-    write_sarif(
-        tmp_path,
-        [
-            _finding(
-                fix_candidate=_fix_candidate(),
-                fix_preparation={"state": "ready", "candidate_digest": "0" * 64},
-                code_locations=[
-                    {
-                        "file": "app.py",
-                        "start_line": 4,
-                        "end_line": 4,
-                        "fix_before": "unsafe(value)",
-                        "fix_after": "safe(value)",
-                    }
-                ],
-            )
-        ],
-    )
-
-    result = _read(tmp_path)["runs"][0]["results"][0]
-    assert "fixes" not in result
-
-
-def test_write_sarif_hides_unprepared_fix_candidate(tmp_path: Path) -> None:
-    write_sarif(
-        tmp_path,
-        [
-            _finding(
-                code_locations=[
-                    {
-                        "file": "app.py",
-                        "start_line": 4,
-                        "end_line": 4,
-                        "fix_before": "unsafe(value)",
-                        "fix_after": "safe(value)",
-                    }
-                ],
-            )
-        ],
-    )
-
-    result = _read(tmp_path)["runs"][0]["results"][0]
-    assert "fixes" not in result
-
-
 def test_write_sarif_omits_fixes_without_fix_pairs(tmp_path: Path) -> None:
     write_sarif(tmp_path, [_finding()])
     assert "fixes" not in _read(tmp_path)["runs"][0]["results"][0]

tests/test_tool_call_ids.py

@@ -1,12 +1,11 @@
-"""Tests for tool-call id presence and uniqueness.
+"""Tests for tool-call id uniqueness.
 
 Providers that number tool calls per turn (``exec_command:0``, ``:1``, ...)
 restart the counter on every turn, so the same id eventually appears twice in
-one conversation. Others hand back a tool call with no id at all, leaving the
-paired tool message with a blank ``tool_call_id``. Strict providers reject the
-whole request either way, and because the history is replayed on every retry
-the agent can never recover. Gateways that validate ids the way those
-providers do prove both failures and their fixes.
+one conversation. Strict providers then reject the whole request, and because
+the history is replayed on every retry the agent can never recover. A gateway
+that validates id uniqueness the way those providers do proves both the
+failure and the fix.
 """
 
 from __future__ import annotations
@@ -126,51 +125,10 @@ class _StrictHandler(BaseHTTPRequestHandler):
         self.wfile.write(encoded)
 
 
-class _BlankIdHandler(BaseHTTPRequestHandler):
-    """Gateway that hands out an id-less tool call and rejects blank ids, like GLM does."""
-
-    def log_message(self, *args: Any) -> None:
-        pass
-
-    def do_POST(self) -> None:
-        length = int(self.headers.get("Content-Length", 0))
-        body = json.loads(self.rfile.read(length) or b"{}")
-        messages = body.get("messages", [])
-        _REQUESTS.append(messages)
-
-        for index, message in enumerate(messages):
-            if message.get("role") == "tool" and not message.get("tool_call_id"):
-                self._respond(
-                    400,
-                    {
-                        "error": {
-                            "message": (
-                                f"messages[{index}]: tool messages must include "
-                                "a non-empty string tool_call_id"
-                            ),
-                            "code": 400,
-                        }
-                    },
-                )
-                return
-
-        if len(_REQUESTS) == 1:
-            self._respond(200, _tool_call_completion(""))
-        else:
-            self._respond(200, _text_completion("all done"))
-
-    def _respond(self, status: int, payload: dict[str, Any]) -> None:
-        encoded = json.dumps(payload).encode()
-        self.send_response(status)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Content-Length", str(len(encoded)))
-        self.end_headers()
-        self.wfile.write(encoded)
-
-
-def _serve(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]:
+@pytest.fixture
+def strict_gateway() -> Iterator[str]:
     _REQUESTS.clear()
-    server = HTTPServer(("127.0.0.1", 0), handler)
+    server = HTTPServer(("127.0.0.1", 0), _StrictHandler)
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
     try:
@@ -180,16 +138,6 @@ def _serve(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]:
         server.server_close()
 
 
-@pytest.fixture
-def strict_gateway() -> Iterator[str]:
-    yield from _serve(_StrictHandler)
-
-
-@pytest.fixture
-def blank_id_gateway() -> Iterator[str]:
-    yield from _serve(_BlankIdHandler)
-
-
 def _model(base_url: str) -> Model:
     # The gateway answers plain JSON, so the run loop's streamed turns are
     # served non-streamed; the ids on the wire are the same either way.
@@ -274,61 +222,6 @@ def test_history_dedupe_pairs_parallel_calls_by_order() -> None:
     assert rebuilt[1]["call_id"] != "dup"
 
 
-@pytest.mark.asyncio
-async def test_blank_call_id_is_rejected_by_the_provider_without_the_wrapper(
-    blank_id_gateway: str,
-) -> None:
-    # Repro: the model answers with a tool call carrying no id, so the paired
-    # tool result goes back as a ``tool`` message with a blank ``tool_call_id``
-    # and the provider rejects the whole replayed history with a 400.
-    with pytest.raises(Exception, match="non-empty string tool_call_id"):
-        await _run_agent(blank_id_gateway, wrap=False)
-
-
-@pytest.mark.asyncio
-async def test_blank_call_id_is_filled_in_so_the_history_stays_valid(
-    blank_id_gateway: str,
-) -> None:
-    result = await _run_agent(blank_id_gateway, wrap=True)
-
-    assert result.final_output == "all done"
-    call_ids = _assistant_call_ids(_REQUESTS[-1])
-    assert len(call_ids) == 1
-    assert call_ids[0].startswith("call_")
-    assert _tool_results(_REQUESTS[-1]) == ["did 1"]
-
-
-def test_history_dedupe_fills_in_blank_ids_and_keeps_outputs_paired() -> None:
-    items = [
-        {"type": "function_call", "call_id": "", "name": "a", "arguments": "{}"},
-        {"type": "function_call", "call_id": "", "name": "b", "arguments": "{}"},
-        {"type": "function_call_output", "call_id": "", "output": "for-a"},
-        {"type": "function_call_output", "call_id": "", "output": "for-b"},
-    ]
-
-    rebuilt, changed = dedupe_history_call_ids(items)
-
-    assert changed
-    ids = [item["call_id"] for item in rebuilt]
-    assert all(call_id.startswith("call_") for call_id in ids)
-    assert ids[0] == ids[2]
-    assert ids[1] == ids[3]
-    assert ids[0] != ids[1]
-
-
-def test_history_dedupe_fills_in_a_missing_call_id_key() -> None:
-    items = [
-        {"type": "function_call", "name": "a", "arguments": "{}"},
-        {"type": "function_call_output", "output": "x"},
-    ]
-
-    rebuilt, changed = dedupe_history_call_ids(items)
-
-    assert changed
-    assert rebuilt[0]["call_id"] == rebuilt[1]["call_id"]
-    assert rebuilt[0]["call_id"].startswith("call_")
-
-
 def test_history_dedupe_leaves_unique_ids_alone() -> None:
     items = [
         {"type": "function_call", "call_id": "call_a", "name": "a", "arguments": "{}"},
@@ -354,29 +247,3 @@ def test_turn_rewriter_is_stable_across_repeated_sightings() -> None:
 
     assert first.call_id != "exec_command:0"
     assert second.call_id == first.call_id
-
-
-def test_turn_rewriter_fills_in_a_blank_id_stably() -> None:
-    rewriter = TurnCallIdRewriter([])
-    call = ResponseFunctionToolCall(
-        id="fc_1", call_id="", name="a", arguments="{}", type="function_call"
-    )
-
-    first = rewriter.rewrite_item(call)
-    second = rewriter.rewrite_item(call)
-
-    assert first.call_id.startswith("call_")
-    assert second.call_id == first.call_id
-    assert rewriter.rewrite_item(first).call_id == first.call_id
-
-
-def test_turn_rewriter_gives_parallel_blank_calls_distinct_ids() -> None:
-    rewriter = TurnCallIdRewriter([])
-    a = ResponseFunctionToolCall(
-        id="fc_1", call_id="", name="a", arguments="{}", type="function_call"
-    )
-    b = ResponseFunctionToolCall(
-        id="fc_2", call_id="", name="b", arguments="{}", type="function_call"
-    )
-
-    assert rewriter.rewrite_item(a).call_id != rewriter.rewrite_item(b).call_id

tests/test_tool_call_ids_providers.py

@@ -1,379 +0,0 @@
-"""Tool-call id repair across provider routes, models, and streaming modes.
-
-Every model Strix resolves goes through ``StrixProvider``, which picks the
-OpenAI SDK for ``openai/...`` and LiteLLM for every other prefix, then wraps the
-result in the turn guard that repairs tool-call ids. Each route parses a
-provider's tool calls on its own code path, so a blank or missing id — or a
-per-turn counter that repeats — has to be repaired no matter which route
-carried it, streamed or not.
-
-A local OpenAI-compatible gateway stands in for the provider. It hands out
-tool calls with whatever ids the scenario calls for and rejects a history the
-way strict providers do: any ``tool`` message with an empty ``tool_call_id``,
-or two assistant tool calls sharing one id.
-"""
-
-from __future__ import annotations
-
-import json
-import threading
-from contextlib import contextmanager
-from dataclasses import dataclass
-from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import TYPE_CHECKING, Any, ClassVar
-
-import litellm
-import pytest
-from agents import Agent, ModelSettings, Runner, function_tool
-from agents.run import RunConfig
-
-from strix.config import loader
-from strix.config import models as strix_models
-from strix.config.models import StrixProvider
-
-
-if TYPE_CHECKING:
-    from collections.abc import Iterator
-
-
-_MISSING = object()
-
-
-@dataclass(frozen=True)
-class _Scenario:
-    """Tool-call ids the gateway hands out, one list per tool-calling turn."""
-
-    turns: tuple[tuple[Any, ...], ...]
-
-    @property
-    def calls(self) -> int:
-        return sum(len(turn) for turn in self.turns)
-
-    @property
-    def has_null_id(self) -> bool:
-        return any(
-            call_id is None or call_id is _MISSING for turn in self.turns for call_id in turn
-        )
-
-
-SCENARIOS = {
-    "empty-id": _Scenario(turns=(("",),)),
-    "null-id": _Scenario(turns=((None,),)),
-    "missing-id": _Scenario(turns=((_MISSING,),)),
-    "parallel-empty-ids": _Scenario(turns=(("", ""),)),
-    "empty-id-every-turn": _Scenario(turns=(("",), ("",), ("",))),
-    "recycled-counter": _Scenario(turns=(("exec_command:0",), ("exec_command:0",))),
-    "parallel-recycled-counter": _Scenario(
-        turns=(("exec_command:0", "exec_command:1"), ("exec_command:0", "exec_command:1"))
-    ),
-    "mixed-blank-and-recycled": _Scenario(
-        turns=(("exec_command:0", ""), ("exec_command:0", None), (_MISSING,))
-    ),
-    "valid-ids": _Scenario(turns=(("call_a",), ("call_b", "call_c"))),
-}
-
-# One model per provider family Strix recommends or documents, each on the
-# route ``StrixProvider`` gives it: the OpenAI SDK for ``openai/``, LiteLLM's
-# OpenAI-compatible adapters for the rest, and a generic OpenAI-compatible
-# endpoint through ``litellm/openai/``.
-MODELS = [
-    "openai/gpt-5.4",
-    "openrouter/z-ai/glm-5.3",
-    "openrouter/anthropic/claude-sonnet-4.6",
-    "openrouter/google/gemini-3-pro-preview",
-    "openrouter/qwen/qwen3-coder",
-    "zai/glm-5.3",
-    "zai/glm-5.3-flash",
-    "deepseek/deepseek-chat",
-    "moonshot/kimi-k2.5",
-    "xai/grok-4",
-    "mistral/mistral-large-latest",
-    "together_ai/Qwen/Qwen3-235B-A22B",
-    "fireworks_ai/accounts/fireworks/models/kimi-k2",
-    "dashscope/qwen3-max",
-    "deepinfra/Qwen/Qwen3-32B",
-    "nebius/Qwen/Qwen3-32B",
-    "hosted_vllm/Qwen/Qwen3-32B",
-    "litellm/openai/gw-model",
-]
-
-
-class _Gateway(BaseHTTPRequestHandler):
-    scenario: ClassVar[_Scenario]
-    requests: ClassVar[list[dict[str, Any]]]
-    lock: ClassVar[threading.Lock]
-
-    def log_message(self, *args: Any) -> None:
-        pass
-
-    def do_POST(self) -> None:  # noqa: N802
-        length = int(self.headers.get("Content-Length", 0))
-        body = json.loads(self.rfile.read(length) or b"{}")
-        messages = body.get("messages", [])
-        with self.lock:
-            self.requests.append(body)
-            turn = len(self.requests) - 1
-
-        error = _history_error(messages)
-        if error:
-            self._send_json(400, {"error": {"message": error, "code": 400}})
-            return
-
-        if turn < len(self.scenario.turns):
-            ids = self.scenario.turns[turn]
-            message = _tool_call_message(ids, first_n=turn * 10)
-            finish = "tool_calls"
-        else:
-            message = {"role": "assistant", "content": "all done"}
-            finish = "stop"
-
-        if body.get("stream"):
-            self._send_stream(message, finish)
-        else:
-            self._send_json(
-                200,
-                {
-                    "id": f"chatcmpl-{turn}",
-                    "object": "chat.completion",
-                    "created": 0,
-                    "model": body.get("model", "gw-model"),
-                    "choices": [{"index": 0, "finish_reason": finish, "message": message}],
-                    "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7},
-                },
-            )
-
-    def _send_json(self, status: int, payload: dict[str, Any]) -> None:
-        encoded = json.dumps(payload).encode()
-        self.send_response(status)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Content-Length", str(len(encoded)))
-        self.end_headers()
-        self.wfile.write(encoded)
-
-    def _send_stream(self, message: dict[str, Any], finish: str) -> None:
-        self.send_response(200)
-        self.send_header("Content-Type", "text/event-stream")
-        self.end_headers()
-
-        def chunk(delta: dict[str, Any], finish_reason: str | None = None) -> None:
-            payload = {
-                "id": "chatcmpl-s",
-                "object": "chat.completion.chunk",
-                "created": 0,
-                "model": "gw-model",
-                "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}],
-            }
-            self.wfile.write(f"data: {json.dumps(payload)}\n\n".encode())
-
-        chunk({"role": "assistant", "content": ""})
-        if message.get("content"):
-            chunk({"content": message["content"]})
-        for index, call in enumerate(message.get("tool_calls") or []):
-            head: dict[str, Any] = {
-                "index": index,
-                "type": "function",
-                "function": {"name": call["function"]["name"], "arguments": ""},
-            }
-            if "id" in call:
-                head["id"] = call["id"]
-            chunk({"tool_calls": [head]})
-            arguments = call["function"]["arguments"]
-            middle = len(arguments) // 2
-            for part in (arguments[:middle], arguments[middle:]):
-                chunk({"tool_calls": [{"index": index, "function": {"arguments": part}}]})
-        chunk({}, finish)
-        usage = {
-            "id": "chatcmpl-s",
-            "object": "chat.completion.chunk",
-            "created": 0,
-            "model": "gw-model",
-            "choices": [],
-            "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7},
-        }
-        self.wfile.write(f"data: {json.dumps(usage)}\n\n".encode())
-        self.wfile.write(b"data: [DONE]\n\n")
-        self.wfile.flush()
-
-
-def _tool_call_message(ids: tuple[Any, ...], *, first_n: int) -> dict[str, Any]:
-    calls = []
-    for offset, call_id in enumerate(ids):
-        call: dict[str, Any] = {
-            "type": "function",
-            "function": {"name": "do_thing", "arguments": json.dumps({"n": first_n + offset})},
-        }
-        if call_id is not _MISSING:
-            call["id"] = call_id
-        calls.append(call)
-    return {"role": "assistant", "content": None, "tool_calls": calls}
-
-
-def _history_error(messages: list[dict[str, Any]]) -> str | None:
-    seen: set[str] = set()
-    for index, message in enumerate(messages):
-        if message.get("role") == "tool":
-            call_id = message.get("tool_call_id")
-            if not isinstance(call_id, str) or not call_id:
-                return (
-                    f"messages[{index}]: tool messages must include a non-empty string tool_call_id"
-                )
-        for call in message.get("tool_calls") or []:
-            call_id = call.get("id")
-            if not isinstance(call_id, str) or not call_id:
-                return f"messages[{index}]: assistant tool_calls must include a non-empty id"
-            if call_id in seen:
-                return f"messages[{index}]: duplicate tool_call id {call_id!r}"
-            seen.add(call_id)
-    return None
-
-
-def _serve(scenario: _Scenario) -> tuple[HTTPServer, type[_Gateway]]:
-    handler = type(
-        "_ScenarioGateway",
-        (_Gateway,),
-        {"scenario": scenario, "requests": [], "lock": threading.Lock()},
-    )
-    server = HTTPServer(("127.0.0.1", 0), handler)
-    threading.Thread(target=server.serve_forever, daemon=True).start()
-    return server, handler
-
-
-@pytest.fixture(autouse=True)
-def _reset_settings(monkeypatch: pytest.MonkeyPatch) -> None:
-    for key in ("STRIX_LLM", "LLM_DISABLE_STREAMING", "LLM_MAX_TOOL_CALLS_PER_TURN"):
-        monkeypatch.delenv(key, raising=False)
-    monkeypatch.setattr(loader, "_cached", None)
-    monkeypatch.setattr(loader, "_override", None)
-    # As ``configure_sdk_model_defaults`` sets it, so routes like ``zai/`` that
-    # reject ``parallel_tool_calls`` still get a request out.
-    monkeypatch.setattr(litellm, "drop_params", True)
-
-
-@contextmanager
-def _gateway(scenario: _Scenario) -> Iterator[tuple[str, type[_Gateway]]]:
-    server, handler = _serve(scenario)
-    try:
-        yield f"http://127.0.0.1:{server.server_address[1]}/v1", handler
-    finally:
-        server.shutdown()
-        server.server_close()
-
-
-async def _run(
-    monkeypatch: pytest.MonkeyPatch, model: str, base_url: str, *, stream: bool, parallel: bool
-) -> tuple[Any, list[int]]:
-    monkeypatch.setenv("LLM_DISABLE_STREAMING", "false" if stream else "true")
-    monkeypatch.setattr(loader, "_cached", None)
-    # Binding the provider to the gateway sends every route there, the way a
-    # custom endpoint would, while each prefix keeps its own adapter.
-    provider = StrixProvider(api_key="tok", base_url=base_url)
-    ran: list[int] = []
-
-    @function_tool
-    def do_thing(n: int) -> str:
-        ran.append(n)
-        return f"did {n}"
-
-    agent = Agent(
-        name="t",
-        instructions="use the tool",
-        tools=[do_thing],
-        model=model,
-        model_settings=ModelSettings(parallel_tool_calls=parallel),
-    )
-    result = Runner.run_streamed(
-        agent, input="please", max_turns=10, run_config=RunConfig(model_provider=provider)
-    )
-    async for _ in result.stream_events():
-        pass
-    return result, ran
-
-
-def _final_history(handler: type[_Gateway]) -> list[dict[str, Any]]:
-    messages: list[dict[str, Any]] = handler.requests[-1]["messages"]
-    return messages
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("stream", [True, False], ids=["streamed", "non-streamed"])
-@pytest.mark.parametrize("scenario_name", list(SCENARIOS))
-@pytest.mark.parametrize("model", MODELS)
-async def test_tool_call_ids_are_repaired_on_every_route(
-    request: pytest.FixtureRequest,
-    monkeypatch: pytest.MonkeyPatch,
-    model: str,
-    scenario_name: str,
-    *,
-    stream: bool,
-) -> None:
-    scenario = SCENARIOS[scenario_name]
-    if model.startswith("openai/") and not stream and scenario.has_null_id:
-        request.applymarker(
-            pytest.mark.xfail(
-                strict=True,
-                reason=(
-                    "the Agents SDK's Chat Completions converter rejects a null tool-call id "
-                    "in a non-streamed response before the turn guard sees it"
-                ),
-            )
-        )
-    with _gateway(scenario) as (base_url, handler):
-        result, ran = await _run(
-            monkeypatch,
-            model,
-            base_url,
-            stream=stream,
-            parallel=any(len(t) > 1 for t in scenario.turns),
-        )
-
-    assert result.final_output == "all done"
-    assert len(handler.requests) == len(scenario.turns) + 1
-    assert all(bool(body.get("stream")) is stream for body in handler.requests)
-
-    history = _final_history(handler)
-    call_ids = [call["id"] for m in history for call in m.get("tool_calls") or []]
-    assert len(call_ids) == scenario.calls
-    assert all(isinstance(call_id, str) and call_id for call_id in call_ids)
-    assert len(set(call_ids)) == len(call_ids)
-
-    outputs = {m["tool_call_id"]: m["content"] for m in history if m.get("role") == "tool"}
-    assert set(outputs) == set(call_ids)
-
-    originals = {
-        turn * 10 + offset: call_id
-        for turn, ids in enumerate(scenario.turns)
-        for offset, call_id in enumerate(ids)
-    }
-    assert sorted(ran) == sorted(originals)
-
-    # Each output still answers the call that produced it.
-    id_by_n = {
-        json.loads(call["function"]["arguments"])["n"]: call["id"]
-        for m in history
-        for call in m.get("tool_calls") or []
-    }
-    assert {id_by_n[n]: f"did {n}" for n in originals} == outputs
-
-    # A usable id is kept the first time it appears; only blanks and repeats change.
-    seen: set[str] = set()
-    for n, original in sorted(originals.items()):
-        if isinstance(original, str) and original and original not in seen:
-            assert id_by_n[n] == original
-            seen.add(original)
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("stream", [True, False], ids=["streamed", "non-streamed"])
-@pytest.mark.parametrize("model", MODELS)
-async def test_blank_call_id_is_rejected_on_every_route_without_the_guard(
-    monkeypatch: pytest.MonkeyPatch, model: str, *, stream: bool
-) -> None:
-    # Repro: with the turn guard removed, an empty id reaches the provider on
-    # every route, so the matrix above is exercising the repair and not a
-    # route that happens to fill ids in on its own.
-    monkeypatch.setattr(strix_models, "_TurnGuardModel", lambda model, **_: model)
-    with (
-        _gateway(SCENARIOS["empty-id"]) as (base_url, _handler),
-        pytest.raises(Exception, match="must include a non-empty"),
-    ):
-        await _run(monkeypatch, model, base_url, stream=stream, parallel=False)