feat(agents): soft wrap-up nudge for sub-agents whose skills set wrap_up_after_turns

BudgetPausedError carries whether the operator pause caused it, so a resume that lands between the hook's refusal and the loop's handler still parks and continues instead of exiting under the stop policy. After the resume the agent drains its mailbox into the session before retrying the call, so a message queued while it was held is not skipped.

docs/advanced/configuration.mdx

@@ -33,6 +33,18 @@ Configure Strix using environment variables or a config file.
   the LiteLLM and native OpenAI routing paths.
 </ParamField>
 
+<ParamField path="LLM_EXTRA_BODY" type="string">
+  Extra fields merged into the JSON body of every request to the main model, as a JSON object
+  (e.g. `{"service_tier":"flex","safety_identifier":"9f86d081..."}`). Use it
+  for provider parameters Strix doesn't expose, such as OpenAI's `service_tier`
+  or `safety_identifier`. Applies to both the LiteLLM and native OpenAI routing
+  paths. Fields are not validated, so the provider or gateway must accept them;
+  unknown fields are usually rejected. Extra fields can override fields Strix
+  sends and break requests, so don't set request-shaping fields such as `model`, `messages`, `input`,
+  `tools` or `stream`, and use `STRIX_REASONING_EFFORT` rather than
+  `reasoning_effort`.
+</ParamField>
+
 <ParamField path="LLM_TIMEOUT" default="300" type="integer">
   Request timeout in seconds for LLM calls.
 </ParamField>
@@ -88,6 +100,12 @@ affecting the agents that do the actual testing.
   inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers.
 </ParamField>
 
+<ParamField path="DEDUPE_LLM_EXTRA_BODY" type="string">
+  Optional JSON object of extra request-body fields sent on every
+  deduplication-model request. A dedicated dedupe model never inherits
+  `LLM_EXTRA_BODY`; set this when its endpoint needs them too.
+</ParamField>
+
 <ParamField path="STRIX_DEDUPE_REASONING_EFFORT" type="string">
   Reasoning effort for the deduplication model. Defaults to the model's own
   baseline when unset.

docs/llm-providers/local.mdx

@@ -68,6 +68,21 @@ export LLM_API_KEY="your-bearer-token"          # sent as Authorization: Bearer
 export LLM_EXTRA_HEADERS='{"X-Feature-Key":"value","X-Tenant":"acme"}'
 ```
 
+### Extra request-body fields
+
+Some provider parameters go in the request body rather than a header, and
+Strix has no dedicated setting for them. Set them with `LLM_EXTRA_BODY` as a
+JSON object — they are merged into the body of every request to the model. For example, to
+use OpenAI's cheaper flex processing and attribute requests to a user with a
+safety identifier (a hash, never a raw email):
+
+```bash
+export LLM_EXTRA_BODY='{"service_tier":"flex","safety_identifier":"9f86d081884c7d65"}'
+```
+
+Fields aren't validated, so use only fields your provider accepts, and don't
+set fields Strix already sends, such as `model`, `messages` or `stream`.
+
 For endpoints behind a private CA, point Strix at your certificate bundle with
 the standard `SSL_CERT_FILE=/path/to/ca-bundle.pem` — never disable TLS
 verification against a real endpoint.

strix/agents/factory.py

@@ -13,11 +13,13 @@ from agents.agent import ToolsToFinalOutputResult
 from agents.sandbox import SandboxAgent
 from agents.sandbox.capabilities import Filesystem, Shell
 from agents.sandbox.errors import InvalidManifestPathError
+from agents.sandbox.session.pty_types import PTY_YIELD_TIME_MS_MAX
 from agents.tool import CustomTool, FunctionTool, Tool
 from pydantic import ValidationError
 
 from strix.agents.prompt import render_system_prompt
 from strix.config import load_settings
+from strix.config.tool_call_arguments import describe_malformed_arguments
 from strix.tools.agents_graph.tools import (
     agent_finish,
     create_agent,
@@ -267,6 +269,10 @@ def _with_coerced_arguments(tool: FunctionTool) -> FunctionTool:
     nullish = tool.name.startswith(_QUERY_TOOL_PREFIXES)
 
     async def invoke(ctx: Any, raw_input: str) -> Any:
+        malformed = describe_malformed_arguments(tool.name, raw_input)
+        if malformed is not None:
+            logger.debug("Tool %s got malformed arguments; asking the model to re-issue", tool.name)
+            return malformed
         return await invoke_tool(ctx, _coerce_arguments(raw_input, schema, nullish=nullish))
 
     tool.on_invoke_tool = invoke
@@ -436,8 +442,47 @@ def _apply_shell_output_cap(parsed: dict[str, Any]) -> None:
     )
 
 
+_EXEC_YIELD_CAP_S = PTY_YIELD_TIME_MS_MAX // 1000
+
+_EXEC_COMMAND_DESCRIPTION_NOTE = (
+    f"The wait for output is capped at {_EXEC_YIELD_CAP_S} s whatever yield_time_ms says. "
+    "A non-TTY command still running at the yield cannot be read later (write_stdin fails "
+    "on it and its output is lost): run anything longer in the background with its output "
+    "redirected to a file (`cmd > /tmp/out.log 2>&1 &`, then read the log), or start it with "
+    'tty=true and poll with write_stdin(session_id=<id>, chars="").'
+)
+
+_RUNNING_NON_TTY_NOTICE = (
+    "Note: this non-TTY process cannot be polled and its later output is lost. Rerun it in "
+    "the background (`cmd > /tmp/out.log 2>&1 &`, then read the log) or with tty=true."
+)
+
+_PROCESS_RUNNING_RE = re.compile(r"^Process running with session ID \d+$", re.MULTILINE)
+_OUTPUT_HEADER_RE = re.compile(r"^Output:\n", re.MULTILINE)
+
+
+def _annotate_running_non_tty(parsed: dict[str, Any], result: Any) -> Any:
+    if parsed.get("tty") or not isinstance(result, str):
+        return result
+    header = _OUTPUT_HEADER_RE.search(result)
+    head_end = len(result) if header is None else header.start()
+    if _PROCESS_RUNNING_RE.search(result, 0, head_end) is None:
+        return result
+    notice = _RUNNING_NON_TTY_NOTICE
+    requested = parsed.get("yield_time_ms")
+    if isinstance(requested, int | float) and requested > PTY_YIELD_TIME_MS_MAX:
+        notice = (
+            f"Note: yield_time_ms={int(requested)} was capped at {PTY_YIELD_TIME_MS_MAX} "
+            f"({_EXEC_YIELD_CAP_S} s). " + notice.removeprefix("Note: ")
+        )
+    if header is None:
+        return f"{result}\n{notice}"
+    return f"{result[: header.end()]}{notice}\n{result[header.end() :]}"
+
+
 def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
     invoke_tool = tool.on_invoke_tool
+    tool.description = f"{tool.description.rstrip()} {_EXEC_COMMAND_DESCRIPTION_NOTE}"
 
     async def invoke(ctx: Any, raw_input: str) -> Any:
         try:
@@ -450,7 +495,7 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
             _apply_shell_output_cap(parsed)
             raw_input = json.dumps(parsed)
         try:
-            return await invoke_tool(ctx, raw_input)
+            result = await invoke_tool(ctx, raw_input)
         except ValidationError as exc:
             return _format_validation_error(tool.name, exc)
         except InvalidManifestPathError as exc:
@@ -460,6 +505,9 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
                 "(or omitted to use the turn's cwd). "
                 f"Got: {rel!r}."
             )
+        if isinstance(parsed, dict):
+            return _annotate_running_non_tty(parsed, result)
+        return result
 
     tool.on_invoke_tool = invoke
     return tool

strix/agents/prompts/system_prompt.jinja

@@ -44,7 +44,7 @@ INTERACTIVE BEHAVIOR:
 - If you end a turn on plain text and the nudge arrives, your words already reached the user. Do not repeat them: call wait_for_user.
 - You can narrate while you work — plain text is shown to the user as you go. Narrating is free; wait_for_user is specifically the act of WAITING for the user, so do not call it just to give a status update.
 - Respond naturally when the user asks questions or gives instructions.
-- While actively working on a task, every turn should carry exactly one tool call — use think to plan, the appropriate tool to act, and wait_for_user only when you genuinely need the user.
+- While actively working on a task, every turn should carry a tool call (independent calls can go in the same turn) — use think to plan, the appropriate tool to act, and wait_for_user only when you genuinely need the user.
 - Never loop through think or other tools just to prepare, polish, confirm, or announce an answer. Once you know the answer, write it and call wait_for_user.
 {% else %}
 AUTONOMOUS BEHAVIOR:
@@ -55,6 +55,7 @@ AUTONOMOUS BEHAVIOR:
 - There is no user attached to this run, so there is nobody to ask and nothing to yield to. If there is nothing left to execute: do NOT send filler/repetitive text — either call wait_for_agents (only if you are genuinely expecting another agent to message you) or finish your work (subagents: agent_finish; root: finish_scan)
 - While the agent loop is running, almost every output MUST be a tool call. Do NOT send plain text messages; act via tools. If waiting on another agent, use wait_for_agents; when done, use agent_finish (subagents) or finish_scan (root)
 - A text-only turn does nothing: it neither ends the run nor yields — it just wastes a turn and forces a retry. The lifecycle tools (``finish_scan`` for root, ``agent_finish`` for subagents) are the ONLY way to terminate, and the report flows through them. If you find yourself wanting to say "Done!" or "Scan complete" without a tool call, call the lifecycle tool instead.
+- Read-only calls can go together in one turn: when several file reads or searches don't depend on each other's output, issue them as separate tool calls in the same turn rather than one per turn. Anything else — writing files, running commands that change state, sending requests, creating or messaging agents, reporting, or finishing — goes alone in its own turn.
 {% endif %}
 </communication_rules>
 

strix/config/loader.py

@@ -6,7 +6,8 @@ import json
 import logging
 import os
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from types import UnionType
+from typing import TYPE_CHECKING, Any, Union, get_args, get_origin
 
 from pydantic import AliasChoices, BaseModel
 
@@ -126,13 +127,36 @@ def _read_json_overrides(path: Path) -> dict[str, dict[str, Any]]:
                 continue  # env wins under some alias; skip the JSON file for this field
             for alias in aliases:
                 if alias in env_block_upper:
-                    sub_data[fname] = env_block_upper[alias]
+                    sub_data[fname] = _decode_saved_value(finfo, env_block_upper[alias])
                     break
         if sub_data:
             nested[sub_name] = sub_data
     return nested
 
 
+def _decode_saved_value(finfo: FieldInfo, value: Any) -> Any:
+    """Decode a JSON string saved for a dict field, as pydantic does for env vars.
+
+    ``persist_current`` saves env values verbatim, so a dict setting such as
+    ``LLM_EXTRA_HEADERS`` lands in the file as JSON text. Values that are
+    already objects, or aren't valid JSON, are left for validation to judge.
+    """
+    if not isinstance(value, str) or not _is_dict_field(finfo):
+        return value
+    try:
+        return json.loads(value)
+    except ValueError:
+        return value
+
+
+def _is_dict_field(finfo: FieldInfo) -> bool:
+    annotation = finfo.annotation
+    candidates = (
+        get_args(annotation) if get_origin(annotation) in (Union, UnionType) else (annotation,)
+    )
+    return any(c is dict or get_origin(c) is dict for c in candidates)
+
+
 def _first_alias_value(aliases: list[str], source: Mapping[str, Any]) -> Any | None:
     return next((source[alias] for alias in aliases if alias in source), None)
 

strix/config/models.py

@@ -41,6 +41,7 @@ from openai.types.shared import Reasoning
 from strix.agents.prompt import CACHE_POINT
 from strix.config import codex
 from strix.config.loader import load_settings
+from strix.config.tool_call_arguments import repair_input
 from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
 from strix.config.tool_call_limits import TurnToolCallLimiter
 from strix.llm import request_log
@@ -255,6 +256,12 @@ class _TurnGuardModel(Model):
     Ids that collide with the history are rewritten before the turn is
     recorded, and already-corrupted histories are repaired on the way out.
 
+    Tool-call arguments: a turn whose ``arguments`` are not valid JSON fails
+    that one call with a parse error, but the raw string is recorded and
+    strict providers then reject every request that replays it. Such
+    arguments are rewritten to a valid JSON object on the way out (see
+    :mod:`strix.config.tool_call_arguments`).
+
     Tool-call volume: a degenerate response can queue hundreds of calls that
     the run loop then honours one by one. Only the first
     ``LLM_MAX_TOOL_CALLS_PER_TURN`` calls of a response are kept.
@@ -313,7 +320,7 @@ class _TurnGuardModel(Model):
         conversation_id: str | None,
         prompt: ResponsePromptParam | None,
     ) -> ModelResponse:
-        sanitized = dedupe_input(input)
+        sanitized = _sanitize_input(input)
         rewriter = TurnCallIdRewriter(sanitized)
         system_instructions, sanitized = _split_cached_prefix(
             system_instructions, sanitized, model_settings
@@ -349,7 +356,7 @@ class _TurnGuardModel(Model):
         conversation_id: str | None,
         prompt: ResponsePromptParam | None,
     ) -> AsyncIterator[TResponseStreamEvent]:
-        sanitized = dedupe_input(input)
+        sanitized = _sanitize_input(input)
         rewriter = TurnCallIdRewriter(sanitized)
         system_instructions, sanitized = _split_cached_prefix(
             system_instructions, sanitized, model_settings
@@ -379,6 +386,10 @@ class _TurnGuardModel(Model):
         self._log_dropped(limiter)
 
 
+def _sanitize_input(model_input: str | list[TResponseInputItem]) -> str | list[Any]:
+    return repair_input(dedupe_input(model_input))
+
+
 def _split_cached_prefix(
     system_instructions: str | None,
     model_input: str | list[Any],

strix/config/settings.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Literal
+from typing import Any, Literal
 
 from pydantic import AliasChoices, Field, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -55,6 +55,11 @@ class LlmSettings(BaseSettings):
         alias="LLM_EXTRA_HEADERS",
         repr=False,
     )
+    extra_body: dict[str, Any] | None = Field(
+        default=None,
+        alias="LLM_EXTRA_BODY",
+        repr=False,
+    )
     reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
     force_required_tool_choice: bool = Field(
         default=False,
@@ -110,6 +115,11 @@ class DedupeSettings(BaseSettings):
         alias="DEDUPE_LLM_EXTRA_HEADERS",
         repr=False,
     )
+    extra_body: dict[str, Any] | None = Field(
+        default=None,
+        alias="DEDUPE_LLM_EXTRA_BODY",
+        repr=False,
+    )
 
     @field_validator("reasoning_effort", mode="before")
     @classmethod

strix/config/tool_call_arguments.py

@@ -0,0 +1,100 @@
+"""Keep replayed tool-call arguments valid JSON.
+
+A model occasionally emits a tool call whose ``arguments`` string is not a
+JSON object (truncated, unbalanced, empty). The call itself fails safely: the
+tool reports the parse error and the run continues. But the raw string is
+recorded in the session as-is, and strict OpenAI-compatible servers (vLLM,
+SGLang, ...) validate every assistant tool call in the request, so from then
+on each turn is rejected with ``Assistant tool call function.arguments must
+be valid JSON`` and the agent can never recover. Rewriting the replayed
+arguments to a JSON object that carries the original text keeps the history
+valid while the model still sees what it sent.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from openai.types.responses import ResponseFunctionToolCall
+
+
+MALFORMED_ARGUMENTS_KEY = "malformed_arguments"
+
+
+def _reject_constant(name: str) -> Any:
+    raise ValueError(f"non-standard JSON constant {name}")
+
+
+def _loads_strict(text: str) -> Any:
+    try:
+        return json.loads(text, parse_constant=_reject_constant)
+    except RecursionError as exc:
+        raise ValueError("arguments are nested too deeply to parse") from exc
+
+
+def describe_malformed_arguments(tool_name: str, arguments: str) -> str | None:
+    """Return a model-facing recovery message if ``arguments`` is not a JSON object.
+
+    A tool call whose arguments do not parse is almost always one the stream cut
+    off (the server flushed a prefix of the JSON, or the client closed early),
+    so instead of the SDK's generic parse-error result the model is told the
+    call never ran and must be re-issued whole.
+    """
+    if not arguments.strip():
+        return None
+    try:
+        parsed = _loads_strict(arguments)
+    except ValueError as exc:
+        detail = str(exc)
+    else:
+        if isinstance(parsed, dict):
+            return None
+        detail = f"expected a JSON object, got {type(parsed).__name__}"
+    return (
+        f"{tool_name}: the tool call was not executed because its arguments were "
+        f"truncated or otherwise not valid JSON ({detail}). The response was likely "
+        "cut off mid-call. Re-issue the call with complete, valid JSON arguments."
+    )
+
+
+def repair_arguments(arguments: object) -> str | None:
+    """Return replacement arguments a strict server accepts, or ``None`` if already valid."""
+    if not isinstance(arguments, str) or not arguments.strip():
+        return "{}"
+    try:
+        parsed = _loads_strict(arguments)
+    except ValueError:
+        parsed = None
+    if isinstance(parsed, dict):
+        return None
+    return json.dumps({MALFORMED_ARGUMENTS_KEY: arguments}, ensure_ascii=False)
+
+
+def repair_history_arguments(items: list[Any]) -> tuple[list[Any], bool]:
+    """Rewrite function calls in a conversation history whose arguments are not a JSON object."""
+    rebuilt: list[Any] = []
+    changed = False
+
+    for item in items:
+        if isinstance(item, dict):
+            if item.get("type") == "function_call":
+                repaired = repair_arguments(item.get("arguments"))
+                if repaired is not None:
+                    item = {**item, "arguments": repaired}  # noqa: PLW2901
+                    changed = True
+        elif isinstance(item, ResponseFunctionToolCall):
+            repaired = repair_arguments(item.arguments)
+            if repaired is not None:
+                item = item.model_copy(update={"arguments": repaired})  # noqa: PLW2901
+                changed = True
+        rebuilt.append(item)
+
+    return rebuilt, changed
+
+
+def repair_input(model_input: str | list[Any]) -> str | list[Any]:
+    if isinstance(model_input, str):
+        return model_input
+    rebuilt, changed = repair_history_arguments(model_input)
+    return rebuilt if changed else model_input

strix/core/agents.py

@@ -75,6 +75,7 @@ class AgentCoordinator:
         self._budget_stopped = False
         self._reserve_stopped = False
         self._budget_paused = False
+        self._operator_paused = False
         self._resume_epoch = 0
         self._budget_policy: BudgetPolicy = "stop"
         self._extend_budget: Callable[[], None] | None = None
@@ -105,6 +106,10 @@ class AgentCoordinator:
     def budget_paused(self) -> bool:
         return self._budget_paused
 
+    @property
+    def operator_paused(self) -> bool:
+        return self._operator_paused
+
     @property
     def resume_epoch(self) -> int:
         """Bumped by every ``resume_budget``; an agent parks against the value it read."""
@@ -150,6 +155,7 @@ class AgentCoordinator:
         """
         async with self._lock:
             self._budget_paused = True
+            self._operator_paused = True
         logger.info("scan paused by the operator")
         await self._maybe_snapshot()
 
@@ -165,6 +171,7 @@ class AgentCoordinator:
             self._set_budget_limit(max_budget_usd)
         async with self._lock:
             self._budget_paused = False
+            self._operator_paused = False
             self._resume_epoch += 1
             woken = [aid for aid, status in self.statuses.items() if status == "budget_paused"]
             for aid in woken:
@@ -232,7 +239,7 @@ class AgentCoordinator:
             self._budget_stopped = budget_stopped
             self._reserve_stopped = reserve_stopped
             if not budget_paused:
-                self._budget_paused = False
+                self._budget_paused = self._operator_paused
                 for aid, status in self.statuses.items():
                     if status == "budget_paused":
                         self.statuses[aid] = "waiting"
@@ -424,7 +431,12 @@ class AgentCoordinator:
         unknown, or it is terminal and its loop does not park for wake-ups.
         """
         from_user = message.get("from") == "user"
-        if from_user and self._budget_paused and self._budget_policy != "pause":
+        if (
+            from_user
+            and self._budget_paused
+            and not self._operator_paused
+            and self._budget_policy != "pause"
+        ):
             await self.resume_from_budget_pause(exclude=target_agent_id)
         async with self._lock:
             if target_agent_id not in self.statuses:
@@ -447,7 +459,9 @@ class AgentCoordinator:
                 self.recovery_counts.pop(target_agent_id, None)
                 self.idle_resume_counts.pop(target_agent_id, None)
                 self._parent_notified.discard(target_agent_id)
-                self.statuses[target_agent_id] = "waiting"
+                held = self._operator_paused and self.statuses[target_agent_id] == "budget_paused"
+                if not held:
+                    self.statuses[target_agent_id] = "waiting"
             runtime.wake.set()
             stream = runtime.stream
             interrupt_on_message = runtime.interrupt_on_message
@@ -640,6 +654,7 @@ class AgentCoordinator:
                 "budget_stopped": self._budget_stopped,
                 "reserve_stopped": self._reserve_stopped,
                 "budget_paused": self._budget_paused,
+                "operator_paused": self._operator_paused,
             }
 
     async def restore(self, snap: dict[str, Any]) -> None:
@@ -662,6 +677,7 @@ class AgentCoordinator:
             self._budget_stopped = bool(snap.get("budget_stopped", False))
             self._reserve_stopped = bool(snap.get("reserve_stopped", False))
             self._budget_paused = bool(snap.get("budget_paused", False))
+            self._operator_paused = bool(snap.get("operator_paused", False))
             for aid in self.statuses:
                 self.runtimes.setdefault(aid, AgentRuntime())
 

strix/core/execution.py

@@ -24,6 +24,7 @@ from openai import (
 
 from strix.config import codex
 from strix.core.hooks import (
+    WRAP_UP_AFTER_TURNS_KEY,
     BudgetExceededError,
     BudgetPausedError,
     SubagentBudgetReservedError,
@@ -39,6 +40,7 @@ from strix.core.sessions import (
 )
 from strix.llm import request_log
 from strix.llm.compaction import is_context_overflow, maybe_compact
+from strix.skills import skill_wrap_up_after_turns
 
 
 if TYPE_CHECKING:
@@ -434,6 +436,7 @@ async def spawn_child_agent(
             task=task,
             parent_history=parent_history,
         ),
+        wrap_up_after_turns=skill_wrap_up_after_turns(skills),
         event_sink=event_sink,
         hooks=hooks,
     )
@@ -512,6 +515,7 @@ async def respawn_subagents(
                 task=str(md.get("task", "")),
                 initial_input=[],
                 start_parked=start_parked,
+                wrap_up_after_turns=skill_wrap_up_after_turns(child_skills),
                 event_sink=event_sink,
                 hooks=hooks,
             )
@@ -599,7 +603,7 @@ async def _run_until_lifecycle(
                     hooks=hooks,
                 )
         except BudgetPausedError as exc:
-            if coordinator.budget_policy != "pause":
+            if coordinator.budget_policy != "pause" and not exc.operator_pause:
                 raise
             # The agent parked right before an LLM call; everything up to that
             # point is already in its session. Once resumed, the same call goes
@@ -612,6 +616,7 @@ async def _run_until_lifecycle(
                 # Stopped while parked (operator stop or a parent's stop_agent).
                 await coordinator.reset_recovery(agent_id)
                 return result
+            await coordinator.consume_pending(agent_id)
             if session is not None:
                 input_data = []
             continue
@@ -831,7 +836,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                 await coordinator.detach_stream(agent_id, stream)
         except BudgetPausedError as exc:
             logger.info("agent %s paused at the scan budget limit: %s", agent_id, exc)
-            if coordinator.budget_policy == "pause":
+            if coordinator.budget_policy == "pause" or exc.operator_pause:
                 await coordinator.park_for_budget(agent_id)
             else:
                 await coordinator.pause_for_budget(agent_id)
@@ -1003,7 +1008,7 @@ async def _append_tool_required_message(
         message = (
             "Your previous message ended a turn without a tool call. Plain text never ends "
             "execution and never hands control to the user: it is shown to the user, and the "
-            "run continues. Continue immediately and call exactly one tool. "
+            "run continues. Continue immediately and call a tool. "
             "If you have nothing to do until the user replies, call wait_for_user; your "
             "text already reached them, so do not repeat it. "
             "If you are blocked waiting for another agent, call wait_for_agents. "
@@ -1015,7 +1020,7 @@ async def _append_tool_required_message(
         message = (
             "Your previous response ended the autonomous run without a lifecycle tool "
             "call. That is invalid in non-interactive mode; plain text final answers are "
-            "ignored. Continue immediately and call exactly one tool. "
+            "ignored. Continue immediately and call a tool. "
             f"If your work is complete, call {finish_tool}. "
             "If you are blocked waiting for another agent, call wait_for_agents. "
             "Otherwise use the appropriate execution or planning tool. "
@@ -1162,6 +1167,7 @@ async def _start_child_runner(
     task: str,
     initial_input: Any,
     start_parked: bool = False,
+    wrap_up_after_turns: int | None = None,
     event_sink: StreamEventSink | None = None,
     hooks: RunHooks[dict[str, Any]] | None = None,
 ) -> None:
@@ -1173,6 +1179,7 @@ async def _start_child_runner(
     child_ctx["agent_id"] = child_id
     child_ctx["parent_id"] = parent_id
     child_ctx["task"] = task
+    child_ctx[WRAP_UP_AFTER_TURNS_KEY] = wrap_up_after_turns
 
     async def _child_loop() -> None:
         # A budget stop is a clean scan-wide shutdown, not a child failure: the

strix/core/hooks.py

@@ -22,6 +22,9 @@ logger = logging.getLogger(__name__)
 
 
 LLM_TURN_KEY = "llm_turn"
+# Set on a sub-agent's context from its skills (see ``skill_wrap_up_after_turns``).
+WRAP_UP_AFTER_TURNS_KEY = "wrap_up_after_turns"
+_WRAP_UP_REPEAT_TURNS = 5
 
 # ``BudgetPolicy`` decides what happens when the accumulated LLM cost reaches
 # ``max_budget_usd``.
@@ -35,6 +38,7 @@ LLM_TURN_KEY = "llm_turn"
 # same call when the operator raises the limit or resumes.
 __all__ = [
     "LLM_TURN_KEY",
+    "WRAP_UP_AFTER_TURNS_KEY",
     "BudgetExceededError",
     "BudgetPausedError",
     "BudgetPolicy",
@@ -65,9 +69,12 @@ class BudgetPausedError(RuntimeError):
     decided to park; the agent waits for a resume newer than that.
     """
 
-    def __init__(self, message: str, *, resume_epoch: int = 0) -> None:
+    def __init__(
+        self, message: str, *, resume_epoch: int = 0, operator_pause: bool = False
+    ) -> None:
         super().__init__(message)
         self.resume_epoch = resume_epoch
+        self.operator_pause = operator_pause
 
 
 def _validate_budget(max_budget_usd: float | None) -> None:
@@ -198,15 +205,26 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
         system_prompt: str | None,  # noqa: ARG002
         input_items: list[TResponseInputItem],
     ) -> None:
+        self._pause_if_operator_paused(context)
         if self._budget_policy == "pause":
             self._pause_if_limited(context)
         context.context[LLM_TURN_KEY] = int(context.context.get(LLM_TURN_KEY, 0)) + 1
         try:
             self._maybe_warn_turns(context, input_items)
+            self._maybe_nudge_wrap_up(context, input_items)
             self._maybe_warn_budget(context, input_items)
         except Exception:
             logger.exception("budget/turn warning injection failed")
 
+    def _pause_if_operator_paused(self, context: RunContextWrapper[dict[str, Any]]) -> None:
+        coordinator = coordinator_from_context(context.context)
+        if coordinator is not None and coordinator.operator_paused:
+            raise BudgetPausedError(
+                "scan paused; waiting for the operator to resume",
+                resume_epoch=coordinator.resume_epoch,
+                operator_pause=True,
+            )
+
     def _pause_if_limited(self, context: RunContextWrapper[dict[str, Any]]) -> None:
         """Park the agent before a paid call when the scan is at its limit or paused.
 
@@ -257,6 +275,26 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
         )
         input_items.append({"role": "user", "content": content})
 
+    def _maybe_nudge_wrap_up(
+        self,
+        context: RunContextWrapper[dict[str, Any]],
+        input_items: list[TResponseInputItem],
+    ) -> None:
+        """A soft nudge, never a stop: at the turn a skill asks for, then every few turns."""
+        after = context.context.get(WRAP_UP_AFTER_TURNS_KEY)
+        requests = getattr(getattr(context, "usage", None), "requests", None)
+        if not isinstance(after, int) or not isinstance(requests, int):
+            return
+        turns_used = requests + 1
+        if turns_used < after or (turns_used - after) % _WRAP_UP_REPEAT_TURNS:
+            return
+        content = (
+            f"[NOTICE] You have used {turns_used} turns. Wrap up soon: take only the steps "
+            "you still need to answer your task, then call agent_finish with what you have, "
+            "listing anything unresolved in open_items instead of tracing it further."
+        )
+        input_items.append({"role": "user", "content": content})
+
     def _maybe_warn_budget(
         self,
         context: RunContextWrapper[dict[str, Any]],

strix/core/inputs.py

@@ -250,22 +250,26 @@ def make_model_settings(
     request_timeout: float | None = None,
     prompt_cache: bool = True,
     extra_headers: dict[str, str] | None = None,
+    extra_body: dict[str, Any] | None = None,
     has_tools: bool = True,
 ) -> ModelSettings:
     headers = _request_headers(model_name, extra_headers)
     model_settings = ModelSettings(
-        parallel_tool_calls=False if has_tools else None,
+        parallel_tool_calls=True if has_tools else None,
         retry=DEFAULT_MODEL_RETRY,
         include_usage=True,
         extra_args=request_timeout_extra_args(request_timeout),
         extra_headers=headers,
+        extra_body=dict(extra_body) if extra_body else None,
     )
     if (
         reasoning_effort is not None
         and reasoning_effort != "none"
         and model_supports_reasoning(model_name)
     ):
-        model_settings = model_settings.resolve(_reasoning_settings(reasoning_effort))
+        model_settings = model_settings.resolve(
+            _reasoning_settings(reasoning_effort, extra_body),
+        )
     if force_required_tool_choice and _accepts_required_tool_choice(model_name):
         model_settings = model_settings.resolve(ModelSettings(tool_choice="required"))
 
@@ -290,7 +294,9 @@ def _request_headers(
     return headers or None
 
 
-def _reasoning_settings(effort: ReasoningEffort) -> ModelSettings:
+def _reasoning_settings(
+    effort: ReasoningEffort, extra_body: dict[str, Any] | None = None
+) -> ModelSettings:
     """``max`` is not in the OpenAI SDK's ``Reasoning.effort`` enum, so send it as
     a raw body field instead — also keeping it clear of LiteLLM's DeepSeek mapping,
     which collapses every ``reasoning_effort`` level to plain thinking-enabled.
@@ -298,11 +304,13 @@ def _reasoning_settings(effort: ReasoningEffort) -> ModelSettings:
 
     It goes in ``extra_body``, the field every model implementation forwards as the
     request's ``extra_body``; the same value under ``extra_args`` collides with that
-    keyword and raises before a request is ever sent.
+    keyword and raises before a request is ever sent. ``resolve`` replaces
+    ``extra_body`` wholesale, so the user's ``LLM_EXTRA_BODY`` fields are carried
+    over, with ``reasoning_effort`` taking precedence.
     """
     if effort != "max":
         return ModelSettings(reasoning=Reasoning(effort=effort))
-    return ModelSettings(extra_body={"reasoning_effort": "max"})
+    return ModelSettings(extra_body={**(extra_body or {}), "reasoning_effort": "max"})
 
 
 def _prompt_cache_extra_args(model_name: str) -> dict[str, Any] | None:

strix/core/runner.py

@@ -8,7 +8,7 @@ import io
 import json
 import logging
 import uuid
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -203,7 +203,7 @@ async def run_strix_scan(
     scan_id: str | None = None,
     image: str,
     local_sources: list[dict[str, Any]] | None = None,
-    extra_files: list[dict[str, Any]] | None = None,
+    extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
     coordinator: AgentCoordinator | None = None,
     interactive: bool = False,
     max_turns: int = DEFAULT_MAX_TURNS,
@@ -401,6 +401,7 @@ async def run_strix_scan(
             request_timeout=settings.llm.timeout,
             prompt_cache=settings.llm.prompt_cache,
             extra_headers=settings.llm.extra_headers,
+            extra_body=settings.llm.extra_body,
         )
         run_config = RunConfig(
             model=resolved_model,

strix/interface/main.py

@@ -164,12 +164,13 @@ async def warm_up_llm() -> None:
             raw_model = dedupe_model
             deduper = resolve_dedupe_model(settings.dedupe, dedupe_model)
             # A dedicated dedupe model may route to another provider, which must
-            # never receive the main endpoint's headers; it has its own
-            # DEDUPE_LLM_EXTRA_HEADERS.
+            # never receive the main endpoint's headers or body fields; it has
+            # its own DEDUPE_LLM_EXTRA_HEADERS and DEDUPE_LLM_EXTRA_BODY.
             await preflight_request(
                 deduper,
                 model_name=dedupe_model,
                 extra_headers=settings.dedupe.extra_headers,
+                extra_body=settings.dedupe.extra_body,
                 timeout=llm.preflight_timeout,
                 api_base_setting="DEDUPE_LLM_API_BASE",
             )

strix/interface/scan_setup.py

@@ -116,6 +116,7 @@ async def preflight_model_connection(
         model,
         model_name=model_name,
         extra_headers=resolved_settings.llm.extra_headers,
+        extra_body=resolved_settings.llm.extra_body,
         timeout=resolved_settings.llm.preflight_timeout,
         api_base_setting="LLM_API_BASE",
     )
@@ -126,6 +127,7 @@ async def preflight_request(
     *,
     model_name: str,
     extra_headers: dict[str, str] | None,
+    extra_body: dict[str, Any] | None,
     timeout: int,
     api_base_setting: str,
 ) -> None:
@@ -144,6 +146,7 @@ async def preflight_request(
         request_timeout=timeout,
         prompt_cache=False,
         extra_headers=extra_headers,
+        extra_body=extra_body,
         has_tools=False,
     )
     try:

strix/llm/compaction.py

@@ -307,6 +307,7 @@ async def _summarize(model: str, prompt: str, max_tokens: int) -> str | None:
         request_timeout=llm.timeout,
         prompt_cache=False,
         extra_headers=llm.extra_headers,
+        extra_body=llm.extra_body,
         has_tools=False,
     ).resolve(ModelSettings(max_tokens=max_tokens))
     try:

strix/report/dedupe.py

@@ -39,11 +39,13 @@ def _dedupe_model_settings(
         model_name=model_name,
         force_required_tool_choice=False,
         request_timeout=request_timeout,
-        # The main model's headers apply only when dedupe falls back to the main
-        # model; a dedicated dedupe model may route to another provider, which
-        # must never receive the main endpoint's credentials. A dedicated model
-        # gets its own DEDUPE_LLM_EXTRA_HEADERS instead.
+        # The main model's headers and body fields apply only when dedupe falls
+        # back to the main model; a dedicated dedupe model may route to another
+        # provider, which must never receive the main endpoint's credentials or
+        # identifiers. A dedicated model gets its own DEDUPE_LLM_EXTRA_HEADERS
+        # and DEDUPE_LLM_EXTRA_BODY instead.
         extra_headers=dedupe.extra_headers if dedupe.model else llm.extra_headers,
+        extra_body=dedupe.extra_body if dedupe.model else llm.extra_body,
         has_tools=False,
     )
 

strix/runtime/deferred_session.py

@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from agents.sandbox.session.manager import Instrumentation
+from agents.sandbox.session.sandbox_session import SandboxSession
+from agents.sandbox.session.sandbox_session_state import SandboxSessionState
+from agents.sandbox.snapshot import NoopSnapshot
+
+
+if TYPE_CHECKING:
+    import io
+    from collections.abc import Sequence
+
+    from agents.sandbox.entries import BaseEntry
+    from agents.sandbox.manifest import Manifest
+    from agents.sandbox.materialization import MaterializedFile
+    from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
+    from agents.sandbox.types import ExecResult, ExposedPortEndpoint, User
+
+
+logger = logging.getLogger(__name__)
+
+
+class _PendingSessionState(SandboxSessionState):
+    type: str = "pending"
+
+
+class DeferredSandboxSession(SandboxSession):
+    def __init__(
+        self,
+        bring_up: asyncio.Task[tuple[Any, BaseSandboxSession]],
+        *,
+        manifest: Manifest,
+        exposed_ports: tuple[int, ...] = (),
+    ) -> None:
+        self._bring_up = bring_up
+        self._pending_state: SandboxSessionState = _PendingSessionState(
+            snapshot=NoopSnapshot(id="pending"),
+            manifest=manifest,
+            exposed_ports=exposed_ports,
+        )
+        self._pending_limits: Any = None
+        self._pending_archive_limits: Any = None
+        self._resolved: BaseSandboxSession | None = None
+        self._client: Any = None
+        self._instrumentation: Instrumentation = Instrumentation()
+        self._seq = 0
+
+    @property
+    def pending(self) -> bool:
+        return self._resolved is None
+
+    @property
+    def bring_up(self) -> asyncio.Task[tuple[Any, BaseSandboxSession]]:
+        return self._bring_up
+
+    async def resolve(self) -> BaseSandboxSession:
+        if self._resolved is not None:
+            return self._resolved
+        client, session = await asyncio.shield(self._bring_up)
+        self._adopt(client, session)
+        return session
+
+    def _adopt(self, client: Any, session: BaseSandboxSession) -> None:
+        if self._resolved is not None:
+            return
+        self._client = client
+        self._resolved = session
+        self._inner = session
+        if self._pending_limits is not None:
+            session._set_concurrency_limits(self._pending_limits)
+        if self._pending_archive_limits is not None:
+            session._set_archive_limits(self._pending_archive_limits)
+        if self._pending_state.manifest != session.state.manifest:
+            session.state = session.state.model_copy(
+                update={"manifest": self._pending_state.manifest}
+            )
+
+    @property
+    def client(self) -> Any:
+        return self._client
+
+    @property
+    def state(self) -> SandboxSessionState:
+        if self._resolved is None:
+            return self._pending_state
+        return self._resolved.state
+
+    @state.setter
+    def state(self, value: SandboxSessionState) -> None:
+        if self._resolved is None:
+            self._pending_state = value
+        else:
+            self._resolved.state = value
+
+    def _set_concurrency_limits(self, limits: Any) -> None:
+        if self._resolved is None:
+            self._pending_limits = limits
+        else:
+            self._resolved._set_concurrency_limits(limits)
+
+    def _set_archive_limits(self, limits: Any) -> None:
+        if self._resolved is None:
+            self._pending_archive_limits = limits
+        else:
+            self._resolved._set_archive_limits(limits)
+
+    def _manifest_base_dir(self) -> Path:
+        if self._resolved is None:
+            return Path.cwd()
+        return self._resolved._manifest_base_dir()
+
+    def normalize_path(self, path: Path | str, *, for_write: bool = False) -> Path:
+        if self._resolved is None:
+            return Path(path)
+        return self._resolved.normalize_path(path, for_write=for_write)
+
+    def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
+        if self._resolved is None:
+            return Path(path)
+        return self._resolved.register_persist_workspace_skip_path(path)
+
+    def supports_pty(self) -> bool:
+        return self._resolved.supports_pty() if self._resolved is not None else False
+
+    async def running(self) -> bool:
+        if self._resolved is None:
+            if not self._bring_up.done():
+                return True
+            return not self._bring_up.cancelled() and self._bring_up.exception() is None
+        return await self._resolved.running()
+
+    async def start(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.start()
+
+    async def stop(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.stop()
+
+    async def shutdown(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.shutdown()
+
+    async def aclose(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.aclose()
+
+    async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None:
+        session = await self.resolve()
+        await session._validate_manifest_application(only_ephemeral=only_ephemeral)
+
+    async def apply_manifest(self, *, only_ephemeral: bool = False) -> Any:
+        session = await self.resolve()
+        return await session.apply_manifest(only_ephemeral=only_ephemeral)
+
+    async def _apply_entry_batch(
+        self,
+        entries: Sequence[tuple[Path, BaseEntry]],
+        *,
+        base_dir: Path,
+    ) -> list[MaterializedFile]:
+        del base_dir
+        session = await self.resolve()
+        return await session._apply_entry_batch(entries, base_dir=session._manifest_base_dir())
+
+    async def exec(self, *command: str | Path, **kwargs: Any) -> ExecResult:
+        await self.resolve()
+        return await super().exec(*command, **kwargs)
+
+    async def ls(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().ls(*args, **kwargs)
+
+    async def rm(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().rm(*args, **kwargs)
+
+    async def mkdir(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().mkdir(*args, **kwargs)
+
+    async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
+        await self.resolve()
+        return await super().read(path, user=user)
+
+    async def write(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().write(*args, **kwargs)
+
+    async def resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
+        await self.resolve()
+        return await super().resolve_exposed_port(port)
+
+    async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
+        session = await self.resolve()
+        return await session._resolve_exposed_port(port)
+
+    async def pty_exec_start(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().pty_exec_start(*args, **kwargs)
+
+    async def pty_write_stdin(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().pty_write_stdin(*args, **kwargs)
+
+    async def pty_terminate_all(self) -> None:
+        if self._resolved is None:
+            return
+        await super().pty_terminate_all()
+
+    async def persist_workspace(self) -> io.IOBase:
+        await self.resolve()
+        return await super().persist_workspace()
+
+    async def hydrate_workspace(self, data: io.IOBase) -> None:
+        await self.resolve()
+        await super().hydrate_workspace(data)
+
+    async def teardown(self) -> None:
+        if self._resolved is None:
+            self._bring_up.cancel()
+            with contextlib.suppress(BaseException):
+                client, session = await self._bring_up
+                self._adopt(client, session)
+        if self._resolved is None:
+            return
+        await self._client.delete(self._resolved)
+
+
+class DeferredSandboxClient:
+    def __init__(self, session: DeferredSandboxSession) -> None:
+        self._session = session
+
+    def __getattr__(self, name: str) -> Any:
+        client = self._session.client
+        if client is None:
+            raise AttributeError(f"sandbox client unavailable before bring-up finishes ({name})")
+        return getattr(client, name)
+
+    async def delete(self, session: Any) -> None:
+        if isinstance(session, DeferredSandboxSession):
+            await session.teardown()
+            return
+        real = await self._session.resolve()
+        await self._session.client.delete(real if session is self._session else session)

strix/runtime/session_manager.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
+import inspect
 import io
 import logging
 import os
@@ -19,9 +21,12 @@ from strix.config import load_settings
 from strix.runtime.backends import backend_supports_bind_mounts, get_backend
 from strix.runtime.caido_bootstrap import bootstrap_caido
 from strix.runtime.caido_handle import CaidoBootstrapHandle
+from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
 
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable
+
     from agents.sandbox.session import BaseSandboxSession
 
     from strix.runtime.status import StatusSink
@@ -255,13 +260,22 @@ def _gitdir_from_pointer(git_file: Path) -> Path | None:
     return None
 
 
+async def _resolve_extra_file_archive(
+    extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None,
+    local_sources: list[dict[str, Any]],
+) -> bytes | None:
+    files = await extra_files if inspect.isawaitable(extra_files) else extra_files
+    return build_extra_file_archive(files, local_sources) if files else None
+
+
 async def create_or_reuse(
     scan_id: str,
     *,
     image: str,
     local_sources: list[dict[str, Any]],
-    extra_files: list[dict[str, Any]] | None = None,
+    extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
     status_sink: StatusSink | None = None,
+    defer: bool = False,
 ) -> dict[str, Any]:
     """Return the existing session bundle for ``scan_id`` or create a new one.
 
@@ -271,7 +285,9 @@ async def create_or_reuse(
     Each ``extra_files`` entry (``{"workspace_path": "/workspace/<rel>",
     "content": bytes | str}``) lands as a regular, agent-writable file at its
     ``workspace_path`` on every backend: the files are uploaded as one archive
-    and unpacked inside the sandbox right after bring-up.
+    and unpacked inside the sandbox right after bring-up. ``extra_files`` may
+    also be an awaitable resolving to that list: provisioning starts at once and
+    the files are awaited only when the sandbox is ready to receive them.
     """
 
     def report(phase: str) -> None:
@@ -292,9 +308,6 @@ async def create_or_reuse(
     else:
         bind_mounts = []
         entries = build_manifest_entries(local_sources)
-    extra_file_archive = (
-        build_extra_file_archive(extra_files, local_sources) if extra_files else None
-    )
 
     # Caido runs as an in-container sidecar; HTTP(S) traffic from any
     # process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -323,41 +336,69 @@ async def create_or_reuse(
         backend_name,
         image,
     )
-    report("Starting sandbox container")
-    client, session = await backend(
-        image=image,
-        manifest=manifest,
-        exposed_ports=(_CONTAINER_CAIDO_PORT,),
-        bind_mounts=bind_mounts,
-    )
 
-    if extra_file_archive is not None:
-        report("Placing workspace files")
+    async def _bring_up() -> tuple[Any, Any]:
+        report("Starting sandbox container")
+        client, session = await _create_or_discard(
+            backend(
+                image=image,
+                manifest=manifest,
+                exposed_ports=(_CONTAINER_CAIDO_PORT,),
+                bind_mounts=bind_mounts,
+            )
+        )
         try:
-            await stage_extra_files(session, extra_file_archive)
+            extra_file_archive = await _resolve_extra_file_archive(extra_files, local_sources)
+            if extra_file_archive is not None:
+                report("Placing workspace files")
+                await stage_extra_files(session, extra_file_archive)
         except BaseException:
             await _discard_session(client, session)
             raise
+        return client, session
 
-    report("Setting up the proxy")
-    caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
-    scheme = "https" if caido_endpoint.tls else "http"
-    host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
-    logger.debug("Caido host endpoint resolved: %s", host_caido_url)
+    async def _caido_url(session: Any) -> str:
+        report("Setting up the proxy")
+        caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
+        scheme = "https" if caido_endpoint.tls else "http"
+        host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
+        logger.debug("Caido host endpoint resolved: %s", host_caido_url)
+        return host_caido_url
 
-    # The Caido login + project setup polls the guest for a couple of seconds
-    # and nothing needs the client before the first proxy tool call, so it
-    # runs concurrently with the rest of scan start; consumers resolve the
-    # handle at first use (see CaidoBootstrapHandle).
-    caido_client = CaidoBootstrapHandle(
-        asyncio.create_task(
-            bootstrap_caido(
-                session,
-                host_url=host_caido_url,
-                container_url=container_caido_url,
-            ),
-            name=f"caido-bootstrap-{scan_id}",
+    def _caido(session: Any, host_caido_url: str) -> Any:
+        return bootstrap_caido(
+            session,
+            host_url=host_caido_url,
+            container_url=container_caido_url,
         )
+
+    if defer:
+        bring_up = asyncio.create_task(_bring_up(), name=f"sandbox-bring-up-{scan_id}")
+        deferred = DeferredSandboxSession(
+            bring_up,
+            manifest=manifest,
+            exposed_ports=(_CONTAINER_CAIDO_PORT,),
+        )
+
+        async def _caido_when_up() -> Any:
+            session = await deferred.resolve()
+            return await _caido(session, await _caido_url(session))
+
+        bundle = {
+            "client": DeferredSandboxClient(deferred),
+            "session": deferred,
+            "caido_client": CaidoBootstrapHandle(
+                asyncio.create_task(_caido_when_up(), name=f"caido-bootstrap-{scan_id}")
+            ),
+        }
+        _SESSION_CACHE[scan_id] = bundle
+        logger.info("Sandbox session for scan %s deferred and cached", scan_id)
+        return bundle
+
+    client, session = await _bring_up()
+    host_caido_url = await _caido_url(session)
+    caido_client = CaidoBootstrapHandle(
+        asyncio.create_task(_caido(session, host_caido_url), name=f"caido-bootstrap-{scan_id}")
     )
 
     bundle = {
@@ -370,6 +411,26 @@ async def create_or_reuse(
     return bundle
 
 
+_DISCARD_TASKS: set[asyncio.Task[None]] = set()
+
+
+async def _create_or_discard(create: Awaitable[tuple[Any, Any]]) -> tuple[Any, Any]:
+    pending = asyncio.ensure_future(create)
+    try:
+        return await asyncio.shield(pending)
+    except asyncio.CancelledError:
+        discard = asyncio.create_task(_discard_when_created(pending))
+        _DISCARD_TASKS.add(discard)
+        discard.add_done_callback(_DISCARD_TASKS.discard)
+        raise
+
+
+async def _discard_when_created(create: asyncio.Future[tuple[Any, Any]]) -> None:
+    with contextlib.suppress(BaseException):
+        client, session = await create
+        await _discard_session(client, session)
+
+
 async def _discard_session(client: Any, session: Any) -> None:
     """Best-effort teardown of a session that never made it into the cache."""
     try:

strix/skills/__init__.py

@@ -270,6 +270,19 @@ def _candidate_skill_files(skill_name: str) -> list[Path]:
     return _bare_skill_files(skill_name)
 
 
+def skill_wrap_up_after_turns(skill_names: list[str]) -> int | None:
+    """Smallest ``wrap_up_after_turns`` set in these skills' frontmatter, if any."""
+    turns = []
+    for skill_name in skill_names:
+        candidates = _candidate_skill_files(skill_name)
+        if len(candidates) != 1:
+            continue
+        value = _read_skill_metadata(candidates[0]).get("wrap_up_after_turns", "")
+        if value.isdigit() and int(value) > 0:
+            turns.append(int(value))
+    return min(turns, default=None)
+
+
 def load_skills(skill_names: list[str], *, supports_images: bool = True) -> dict[str, str]:
     """Load skill markdown bodies (frontmatter stripped) by name.
 

strix/tools/proxy/tools.py

@@ -13,6 +13,7 @@ from datetime import datetime
 from typing import TYPE_CHECKING, Any, Literal
 
 from agents import RunContextWrapper, function_tool
+from caido_sdk_client import BaseError
 
 from strix.runtime.caido_handle import CaidoBootstrapHandle
 from strix.tools.nullish import clean_optional
@@ -124,7 +125,11 @@ def _no_client() -> str:
 
 
 def _err(name: str, exc: Exception) -> str:
-    logger.exception("%s failed", name)
+    # Caido-side failures (timeouts, bad filters, ...) are returned to the model, not our bugs.
+    if isinstance(exc, BaseError):
+        logger.warning("%s failed", name, exc_info=True)
+    else:
+        logger.exception("%s failed", name)
     return json.dumps(
         {"success": False, "error": f"{name} failed: {exc}"},
         ensure_ascii=False,

strix/tools/reporting/tool.py

@@ -1876,10 +1876,6 @@ def _build_dependency_evidence(
         evidence += f"\n\n**Usage analysis:** {label}."
         if reachability_evidence and reachability_evidence.strip():
             evidence += f" {reachability_evidence.strip()}"
-        evidence += (
-            " This is a prioritization signal from static analysis, not a"
-            " proof of exploitability or of safety."
-        )
     return evidence
 
 

tests/test_agent_factory_shell.py

@@ -80,6 +80,94 @@ async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None:
     assert json.loads(captured["raw_input"])["shell"] == shell
 
 
+def _running_exec_tool(result: str) -> FunctionTool:
+    async def invoke(_ctx: Any, _raw_input: str) -> str:
+        return result
+
+    return FunctionTool(
+        name="exec_command",
+        description="Runs a command.",
+        params_json_schema={"type": "object", "properties": {}},
+        on_invoke_tool=invoke,
+    )
+
+
+_RUNNING_RESULT = (
+    "Chunk ID: abc123\nWall time: 30.0012 seconds\nProcess running with session ID 7\nOutput:\n"
+)
+
+
+def test_wrap_exec_command_describes_the_yield_cap() -> None:
+    wrapped = factory._wrap_exec_command(_running_exec_tool("ok"))
+
+    assert wrapped.description.startswith("Runs a command. The wait for output is capped at 30 s")
+    assert "cannot be read later" in wrapped.description
+    assert "tty=true" in wrapped.description
+
+
+@pytest.mark.asyncio
+async def test_wrap_exec_command_flags_a_running_non_tty_process() -> None:
+    wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
+
+    result = await wrapped.on_invoke_tool(
+        cast("Any", None), json.dumps({"cmd": "sleep 60; curl x", "yield_time_ms": 90000})
+    )
+
+    assert result == (
+        "Chunk ID: abc123\nWall time: 30.0012 seconds\n"
+        "Process running with session ID 7\nOutput:\n"
+        "Note: yield_time_ms=90000 was capped at 30000 (30 s). this non-TTY process cannot be "
+        "polled and its later output is lost. Rerun it in the background "
+        "(`cmd > /tmp/out.log 2>&1 &`, then read the log) or with tty=true.\n"
+    )
+
+
+@pytest.mark.asyncio
+async def test_wrap_exec_command_flags_a_running_non_tty_process_within_the_cap() -> None:
+    wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
+
+    result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "sleep 60"}))
+
+    assert "Note: this non-TTY process cannot be polled" in result
+    assert "was capped" not in result
+
+
+@pytest.mark.asyncio
+async def test_wrap_exec_command_leaves_a_running_tty_process_alone() -> None:
+    wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
+
+    result = await wrapped.on_invoke_tool(
+        cast("Any", None), json.dumps({"cmd": "python3 -m http.server", "tty": True})
+    )
+
+    assert result == _RUNNING_RESULT
+
+
+@pytest.mark.asyncio
+async def test_wrap_exec_command_leaves_an_exited_process_alone() -> None:
+    exited = "Chunk ID: abc123\nWall time: 0.1 seconds\nProcess exited with code 0\nOutput:\nhi"
+    wrapped = factory._wrap_exec_command(_running_exec_tool(exited))
+
+    result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "echo hi"}))
+
+    assert result == exited
+
+
+@pytest.mark.asyncio
+async def test_wrap_exec_command_ignores_the_running_marker_inside_command_output() -> None:
+    exited = (
+        "Chunk ID: abc123\nWall time: 0.2 seconds\nProcess exited with code 0\nOutput:\n"
+        "Process running with session ID 7\nOutput:\n"
+    )
+    wrapped = factory._wrap_exec_command(_running_exec_tool(exited))
+
+    result = await wrapped.on_invoke_tool(
+        cast("Any", None), json.dumps({"cmd": "cat transcript.txt"})
+    )
+
+    assert result == exited
+
+
 @pytest.mark.asyncio
 async def test_responses_filesystem_custom_tool_output_is_bounded() -> None:
     async def invoke(_ctx: Any, _inp: str) -> str:

tests/test_agent_factory_tool_arguments.py

@@ -38,11 +38,26 @@ async def _roundtrip(
 
 
 _STRING = {"todos": {"type": "string"}}
+
+
+@pytest.mark.asyncio
+async def test_truncated_arguments_short_circuit_with_a_reissue_message() -> None:
+    captured: dict[str, str] = {}
+    wrapped = factory._with_coerced_arguments(_capturing_tool(captured, _STRING))
+
+    result = await wrapped.on_invoke_tool(cast("Any", None), '{"todos": "a, b')
+
+    assert "not executed" in result
+    assert "Re-issue the call" in result
+    assert "raw_input" not in captured
+
+
 _ARRAY = {"tags": {"type": "array", "items": {"type": "string"}}}
 _NULLABLE_ARRAY = {
     "tags": {"anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}]}
 }
 _OBJECT = {"modifications": {"type": "object"}}
+OVER_NESTED = '{"a":' + "[" * 100_000
 
 
 @pytest.mark.asyncio
@@ -136,12 +151,15 @@ async def test_unknown_and_null_arguments_are_untouched() -> None:
 
 
 @pytest.mark.asyncio
-async def test_non_object_payloads_pass_through_unchanged() -> None:
+@pytest.mark.parametrize("arguments", ["not json", '{"n": NaN}', OVER_NESTED])
+async def test_non_object_payloads_are_reported_instead_of_invoked(arguments: str) -> None:
     captured: dict[str, str] = {}
     wrapped = factory._with_coerced_arguments(_capturing_tool(captured, _ARRAY))
 
-    assert await wrapped.on_invoke_tool(cast("Any", None), "not json") == "ok"
-    assert captured["raw_input"] == "not json"
+    result = await wrapped.on_invoke_tool(cast("Any", None), arguments)
+
+    assert result.startswith("probe: the tool call was not executed")
+    assert "raw_input" not in captured
 
 
 @pytest.mark.asyncio

tests/test_budget_pause_policy.py

@@ -21,7 +21,7 @@ from strix.core.sessions import open_agent_session
 
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Callable
+    from collections.abc import AsyncIterator, Awaitable, Callable
     from pathlib import Path
 
 
@@ -31,6 +31,7 @@ _CALL_LATENCY_S = 0.005
 
 class _FakeLedger:
     def __init__(self) -> None:
+        self.on_pause: Callable[[], Awaitable[None]] | None = None
         self.cost = 0.0
         self.calls: list[str] = []
         self.remaining: dict[str, int] = {}
@@ -80,6 +81,8 @@ class _FakeStream:
                 await self._hooks.on_llm_start(ctx_wrapper, self._agent, None, input_items)
             except BudgetPausedError as exc:
                 self.run_loop_exception = exc
+                if self._ledger.on_pause is not None:
+                    await self._ledger.on_pause()
                 return
             self._ledger.warned_inputs.append(input_items)
             if self._ledger.gate is not None:
@@ -150,13 +153,16 @@ class _Scan:
         monkeypatch: pytest.MonkeyPatch,
         *,
         max_budget_usd: float | None,
+        budget_policy: str = "pause",
     ) -> None:
         self.ledger = _FakeLedger()
         self.hooks = ReportUsageHooks(
-            model="test-model", max_budget_usd=max_budget_usd, budget_policy="pause"
+            model="test-model",
+            max_budget_usd=max_budget_usd,
+            budget_policy=budget_policy,  # type: ignore[arg-type]
         )
         self.coordinator = AgentCoordinator()
-        self.coordinator.set_budget_policy("pause")
+        self.coordinator.set_budget_policy(budget_policy)  # type: ignore[arg-type]
         self.coordinator.set_budget_limit_setter(self.hooks.set_max_budget_usd)
         monkeypatch.setattr(execution, "Runner", _fake_runner(self.ledger, self.coordinator))
         monkeypatch.setattr(execution, "_compact_session", _noop_compact)
@@ -538,3 +544,140 @@ async def test_stop_policy_is_unchanged() -> None:
 def test_budget_policy_is_validated() -> None:
     with pytest.raises(ValueError, match="budget_policy"):
         ReportUsageHooks(model="m", budget_policy="later")  # type: ignore[arg-type]
+
+
+def _pause_flags(coordinator: AgentCoordinator) -> tuple[bool, bool]:
+    return coordinator.budget_paused, coordinator.operator_paused
+
+
+@pytest.mark.asyncio
+async def test_operator_pause_holds_under_the_stop_policy(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
+    agents = ("root", "child-a", "child-b")
+
+    with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
+        await scan.start_root(calls=6)
+        await scan.start_child("child-a", calls=6)
+        await scan.start_child("child-b", calls=6)
+
+        await _wait_until(lambda: scan.ledger.cost >= 3.0)
+        await scan.coordinator.pause_budget()
+        spent_at_pause = scan.ledger.cost
+        await _wait_until(lambda: _all_parked(scan.coordinator, *agents))
+        assert _pause_flags(scan.coordinator) == (True, True)
+        assert all(not task.done() for task in scan.tasks())
+
+        await scan.coordinator.send("child-a", {"from": "root", "content": "status?"})
+        await scan.coordinator.send("root", {"from": "user", "content": "keep going"})
+        await asyncio.sleep(0.05)
+        assert _pause_flags(scan.coordinator) == (True, True)
+        assert scan.ledger.cost == pytest.approx(spent_at_pause)
+        assert _all_parked(scan.coordinator, *agents)
+
+        woken = await scan.coordinator.resume_budget()
+        assert sorted(woken) == sorted(agents)
+        assert _pause_flags(scan.coordinator) == (False, False)
+        await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
+        assert scan.ledger.cost == pytest.approx(18.0)
+        assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
+            "root": "completed",
+            "child-a": "completed",
+            "child-b": "completed",
+        }
+        root_items = await scan.sessions[0].get_items()
+        assert any("keep going" in str(item) for item in root_items)
+
+    for session in scan.sessions:
+        session.close()
+
+
+@pytest.mark.asyncio
+async def test_a_resume_that_lands_before_the_park_still_continues(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
+    resumed = 0
+
+    async def resume_at_once() -> None:
+        nonlocal resumed
+        resumed += 1
+        await scan.coordinator.resume_budget()
+
+    scan.ledger.on_pause = resume_at_once
+    with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
+        await scan.start_root(calls=6)
+        await scan.start_child("child-a", calls=6)
+        await _wait_until(lambda: scan.ledger.cost >= 2.0)
+        await scan.coordinator.pause_budget()
+        await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
+        assert resumed >= 1
+        assert _pause_flags(scan.coordinator) == (False, False)
+        assert scan.ledger.cost == pytest.approx(12.0)
+        assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
+            "root": "completed",
+            "child-a": "completed",
+        }
+
+    for session in scan.sessions:
+        session.close()
+
+
+@pytest.mark.asyncio
+async def test_operator_pause_gates_the_stop_policy_hooks() -> None:
+    coordinator = AgentCoordinator()
+    hooks = ReportUsageHooks(model="m", max_budget_usd=None)
+    assert hooks.budget_policy == "stop"
+
+    await coordinator.pause_budget()
+    with pytest.raises(BudgetPausedError):
+        await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
+
+    await coordinator.resume_budget()
+    await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
+
+
+@pytest.mark.asyncio
+async def test_a_user_message_lifts_only_the_limit_pause(tmp_path: Path) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    session = open_agent_session("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)
+    extensions: list[int] = []
+    coordinator.set_budget_extender(lambda: extensions.append(1))
+
+    await coordinator.pause_budget()
+    await coordinator.park_for_budget("root")
+    assert coordinator.statuses["root"] == "budget_paused"
+    await coordinator.send("root", {"from": "user", "content": "keep going"})
+    assert _pause_flags(coordinator) == (True, True)
+    assert extensions == []
+    assert coordinator.pending_counts["root"] == 1
+
+    await coordinator.resume_budget()
+    await coordinator.pause_for_budget("root")
+    await coordinator.send("root", {"from": "user", "content": "keep going"})
+    assert _pause_flags(coordinator) == (False, False)
+    assert extensions == [1]
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_snapshot_and_reset_keep_an_operator_pause() -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.pause_budget()
+    snap = await coordinator.snapshot()
+
+    restored = AgentCoordinator()
+    await restored.restore(snap)
+    assert _pause_flags(restored) == (True, True)
+    await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
+    assert _pause_flags(restored) == (True, True)
+
+    await restored.resume_budget()
+    await restored.pause_for_budget("root")
+    await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
+    assert _pause_flags(restored) == (False, False)
+    assert restored.statuses["root"] == "waiting"

tests/test_compaction.py

@@ -147,7 +147,9 @@ def _patch_budget(monkeypatch: pytest.MonkeyPatch, *, keep_tokens: int, window:
     context.auto_compact = True
     settings = SimpleNamespace(
         context=context,
-        llm=SimpleNamespace(api_key=None, api_base=None, timeout=1, extra_headers=None),
+        llm=SimpleNamespace(
+            api_key=None, api_base=None, timeout=1, extra_headers=None, extra_body=None
+        ),
     )
     monkeypatch.setattr(compaction, "load_settings", lambda: settings)
 
@@ -228,7 +230,11 @@ async def test_summarize_routes_through_provider_with_settings(
         "load_settings",
         lambda: SimpleNamespace(
             llm=SimpleNamespace(
-                api_key=None, api_base=None, timeout=1, extra_headers={"X-Feature-Key": "svc"}
+                api_key=None,
+                api_base=None,
+                timeout=1,
+                extra_headers={"X-Feature-Key": "svc"},
+                extra_body={"service_tier": "flex"},
             )
         ),
     )
@@ -239,6 +245,7 @@ async def test_summarize_routes_through_provider_with_settings(
     assert captured["model"] == "litellm/openai/some-model"
     settings = captured["model_settings"]
     assert settings.extra_headers == {"X-Feature-Key": "svc"}
+    assert settings.extra_body == {"service_tier": "flex"}
     assert settings.max_tokens == 64
 
 

tests/test_config_loader.py

@@ -37,6 +37,12 @@ _LLM_ENV_KEYS = [
     "STRIX_RUNTIME_BACKEND",
     # TelemetrySettings
     "STRIX_TELEMETRY",
+    # JSON-object settings
+    "LLM_EXTRA_HEADERS",
+    "LLM_EXTRA_BODY",
+    "STRIX_DEDUPE_MODEL",
+    "DEDUPE_LLM_EXTRA_HEADERS",
+    "DEDUPE_LLM_EXTRA_BODY",
 ]
 
 
@@ -406,6 +412,57 @@ def test_persist_current_replaces_corrupt_file(
     assert json.loads(target.read_text(encoding="utf-8")) == {"env": {"STRIX_LLM": "env-model"}}
 
 
+@pytest.mark.parametrize(
+    ("env_var", "section", "field"),
+    [
+        ("LLM_EXTRA_HEADERS", "llm", "extra_headers"),
+        ("LLM_EXTRA_BODY", "llm", "extra_body"),
+        ("DEDUPE_LLM_EXTRA_HEADERS", "dedupe", "extra_headers"),
+        ("DEDUPE_LLM_EXTRA_BODY", "dedupe", "extra_body"),
+    ],
+)
+def test_persisted_json_object_setting_reloads_without_env(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, env_var: str, section: str, field: str
+) -> None:
+    loader.apply_config_override(tmp_path / "cli-config.json")
+    monkeypatch.setenv(env_var, '{"X-Tenant": "acme"}')
+    loader.load_settings()
+    loader.persist_current()
+    monkeypatch.delenv(env_var)
+    loader.apply_config_override(tmp_path / "cli-config.json")
+
+    settings = loader.load_settings()
+
+    assert getattr(getattr(settings, section), field) == {"X-Tenant": "acme"}
+
+
+def test_json_object_setting_saved_as_object_still_loads(tmp_path: Path) -> None:
+    target = tmp_path / "cli-config.json"
+    target.write_text(
+        json.dumps({"env": {"LLM_EXTRA_BODY": {"service_tier": "flex"}}}), encoding="utf-8"
+    )
+    loader.apply_config_override(target)
+
+    assert loader.load_settings().llm.extra_body == {"service_tier": "flex"}
+
+
+def test_invalid_saved_json_object_setting_is_rejected(tmp_path: Path) -> None:
+    target = tmp_path / "cli-config.json"
+    target.write_text(json.dumps({"env": {"LLM_EXTRA_BODY": "not json"}}), encoding="utf-8")
+    loader.apply_config_override(target)
+
+    with pytest.raises(ValidationError):
+        loader.load_settings()
+
+
+def test_saved_string_setting_is_not_json_decoded(tmp_path: Path) -> None:
+    target = tmp_path / "cli-config.json"
+    target.write_text(json.dumps({"env": {"STRIX_LLM": '{"not": "decoded"}'}}), encoding="utf-8")
+    loader.apply_config_override(target)
+
+    assert loader.load_settings().llm.model == '{"not": "decoded"}'
+
+
 def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("STRIX_LLM", "persisted-model")
     target = tmp_path / "cli-config.json"

tests/test_cost_tracking.py

@@ -136,6 +136,7 @@ def test_cost_callback_estimates_cost_with_provider_prefixed_model() -> None:
 
     with (
         patch("strix.report.state.get_global_report_state", return_value=report_state),
+        patch("strix.report.state.resolve_litellm_model", side_effect=lambda model: model),
         patch("litellm.completion_cost", side_effect=fake_completion_cost),
     ):
         litellm_cost_callback(kwargs, response)

tests/test_dedupe_model.py

@@ -148,3 +148,42 @@ def test_config_file_loads_dedupe_model(
     assert settings.dedupe.reasoning_effort == "minimal"
     # Main model stays independent of the dedupe override.
     assert settings.llm.model == "openai/root"
+
+
+def test_dedicated_dedupe_model_uses_own_body_fields_not_main(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
+    loader._cached = None
+    try:
+        dedupe = DedupeSettings(
+            STRIX_DEDUPE_MODEL="deepseek/cheap",
+            DEDUPE_LLM_EXTRA_BODY={"service_tier": "flex"},
+        )
+        settings = _dedupe_model_settings(dedupe, "deepseek/cheap", 300)
+        assert settings.extra_body == {"service_tier": "flex"}
+    finally:
+        loader._cached = None
+
+
+def test_dedicated_dedupe_model_gets_no_main_body_fields_by_default(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
+    loader._cached = None
+    try:
+        dedupe = DedupeSettings(STRIX_DEDUPE_MODEL="deepseek/cheap")
+        settings = _dedupe_model_settings(dedupe, "deepseek/cheap", 300)
+        assert settings.extra_body is None
+    finally:
+        loader._cached = None
+
+
+def test_fallback_dedupe_inherits_main_body_fields(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
+    loader._cached = None
+    try:
+        settings = _dedupe_model_settings(DedupeSettings(), "openai/main-model", 300)
+        assert settings.extra_body == {"safety_identifier": "main"}
+    finally:
+        loader._cached = None

tests/test_deferred_session.py

@@ -0,0 +1,316 @@
+from __future__ import annotations
+
+import asyncio
+import io
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from agents.run_config import SandboxRunConfig
+from agents.sandbox import SandboxAgent
+from agents.sandbox.capabilities import Filesystem, Shell
+from agents.sandbox.runtime_session_manager import SandboxRuntimeSessionManager
+
+from strix.runtime import session_manager
+from strix.runtime.backends import _BACKENDS, _BIND_MOUNT_BACKENDS, register_backend
+from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+@dataclass
+class _RuntimeSettings:
+    backend: str
+
+
+@dataclass
+class _Settings:
+    runtime: _RuntimeSettings
+
+
+@dataclass
+class _Endpoint:
+    host: str = "127.0.0.1"
+    port: int = 8080
+    tls: bool = False
+
+
+@dataclass
+class _ExecResult:
+    exit_code: int = 0
+    stdout: bytes = b""
+    stderr: bytes = b""
+
+    def ok(self) -> bool:
+        return self.exit_code == 0
+
+
+class _State:
+    def __init__(self) -> None:
+        self.manifest: Any = None
+        self.session_id = "00000000-0000-4000-8000-000000000001"
+
+    def model_copy(self, *, update: dict[str, Any]) -> _State:
+        copy = _State()
+        copy.manifest = update.get("manifest", self.manifest)
+        return copy
+
+
+class _Session:
+    def __init__(self) -> None:
+        self.state = _State()
+        self.execs: list[tuple[str, ...]] = []
+        self.writes: list[tuple[Path, bytes]] = []
+        self.limits: list[Any] = []
+
+    def _set_concurrency_limits(self, limits: Any) -> None:
+        self.limits.append(limits)
+
+    def _set_archive_limits(self, limits: Any) -> None:
+        self.limits.append(limits)
+
+    async def running(self) -> bool:
+        return True
+
+    async def resolve_exposed_port(self, _port: int) -> _Endpoint:
+        return _Endpoint()
+
+    async def exec(self, *argv: str, timeout: float | None = None, **_kw: Any) -> _ExecResult:
+        del timeout
+        self.execs.append(argv)
+        return _ExecResult()
+
+    async def write(self, path: Path, data: io.IOBase, **_kw: Any) -> None:
+        self.writes.append((path, data.read()))
+
+
+class _Client:
+    def __init__(self) -> None:
+        self.deleted: list[Any] = []
+
+    async def delete(self, session: Any) -> None:
+        self.deleted.append(session)
+
+
+async def _no_caido(*_args: Any, **_kwargs: Any) -> None:
+    return None
+
+
+class _Backend:
+    def __init__(self, *, fail: bool = False) -> None:
+        self.gate = asyncio.Event()
+        self.calls = 0
+        self.fail = fail
+        self.client = _Client()
+        self.session = _Session()
+
+    async def __call__(self, **_kwargs: Any) -> tuple[Any, Any]:
+        self.calls += 1
+        await self.gate.wait()
+        if self.fail:
+            raise RuntimeError("bring-up exploded")
+        return self.client, self.session
+
+
+@pytest.fixture
+def backend(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Iterator[_Backend]:
+    name = f"deferred-{request.node.name}"
+    fake = _Backend(fail="fail" in request.node.name)
+    register_backend(name, fake, supports_bind_mounts=True)
+    monkeypatch.setattr(session_manager, "bootstrap_caido", _no_caido)
+    monkeypatch.setattr(
+        session_manager, "load_settings", lambda: _Settings(runtime=_RuntimeSettings(backend=name))
+    )
+    yield fake
+    _BACKENDS.pop(name, None)
+    _BIND_MOUNT_BACKENDS.discard(name)
+
+
+async def _settle() -> None:
+    for _ in range(10):
+        await asyncio.sleep(0)
+
+
+async def _create(scan_id: str, **kwargs: Any) -> dict[str, Any]:
+    return await session_manager.create_or_reuse(
+        scan_id, image="img", local_sources=[], defer=True, **kwargs
+    )
+
+
+@pytest.mark.asyncio
+async def test_deferred_bundle_returns_before_the_backend_finishes(backend: _Backend) -> None:
+    scan_id = "deferred-returns"
+    try:
+        bundle = await asyncio.wait_for(_create(scan_id), timeout=1)
+        session = bundle["session"]
+        assert isinstance(session, DeferredSandboxSession)
+        assert isinstance(bundle["client"], DeferredSandboxClient)
+        assert session.pending
+        await _settle()
+        assert backend.calls == 1
+        assert await session.running()
+        assert session.state.manifest.environment.value["PYTHONUNBUFFERED"] == "1"
+        again = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
+        assert again is bundle
+        assert backend.calls == 1
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_first_io_resolves_the_sandbox_then_delegates(backend: _Backend) -> None:
+    scan_id = "deferred-io"
+    try:
+        bundle = await _create(scan_id)
+        session = bundle["session"]
+        session._set_concurrency_limits("limits")
+        exec_task = asyncio.create_task(session.exec("echo", "hi"))
+        await asyncio.sleep(0.05)
+        assert not exec_task.done()
+        backend.gate.set()
+        result = await asyncio.wait_for(exec_task, timeout=1)
+        assert result.ok()
+        assert backend.session.execs == [("echo", "hi")]
+        assert not session.pending
+        assert backend.session.limits == ["limits"]
+        assert backend.session.state.manifest is session.state.manifest
+        await session.write(Path("/workspace/x"), io.BytesIO(b"data"))
+        assert backend.session.writes == [(Path("/workspace/x"), b"data")]
+    finally:
+        await session_manager.cleanup(scan_id)
+    assert backend.client.deleted == [backend.session]
+
+
+@pytest.mark.asyncio
+async def test_failed_bring_up_surfaces_on_first_io(backend: _Backend) -> None:
+    scan_id = "deferred-fail"
+    try:
+        bundle = await _create(scan_id)
+        session = bundle["session"]
+        backend.gate.set()
+        with pytest.raises(RuntimeError, match="bring-up exploded"):
+            await asyncio.wait_for(session.exec("true"), timeout=1)
+        assert not await session.running()
+        with pytest.raises(RuntimeError, match="bring-up exploded"):
+            await bundle["caido_client"].get()
+    finally:
+        await session_manager.cleanup(scan_id)
+    assert backend.client.deleted == []
+
+
+@pytest.mark.asyncio
+async def test_cleanup_while_pending_cancels_the_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-cancel"
+    bundle = await _create(scan_id)
+    session = bundle["session"]
+    await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
+    assert session.bring_up.cancelled()
+    assert backend.client.deleted == []
+    assert session_manager._SESSION_CACHE.get(scan_id) is None
+    backend.gate.set()
+    await _settle()
+    assert backend.client.deleted == [backend.session]
+
+
+@pytest.mark.asyncio
+async def test_cleanup_while_pending_ignores_a_failed_late_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-cancel-fail"
+    await _create(scan_id)
+    await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
+    backend.gate.set()
+    await _settle()
+    assert backend.client.deleted == []
+
+
+@pytest.mark.asyncio
+async def test_extra_files_are_staged_inside_the_deferred_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-extra"
+    try:
+        bundle = await _create(
+            scan_id,
+            extra_files=[{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}],
+        )
+        assert backend.session.writes == []
+        backend.gate.set()
+        await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
+        assert len(backend.session.writes) == 1
+        assert backend.session.execs
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_awaitable_extra_files_do_not_delay_the_backend_call(backend: _Backend) -> None:
+    scan_id = "deferred-extra-awaitable"
+    files: asyncio.Future[list[dict[str, Any]] | None] = asyncio.get_running_loop().create_future()
+    try:
+        bundle = await _create(scan_id, extra_files=files)
+        backend.gate.set()
+        await _settle()
+        assert backend.calls == 1
+        assert backend.session.writes == []
+        files.set_result([{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}])
+        await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
+        assert len(backend.session.writes) == 1
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_awaitable_extra_files_resolving_to_none_stage_nothing(backend: _Backend) -> None:
+    scan_id = "deferred-extra-none"
+
+    async def _nothing() -> list[dict[str, Any]] | None:
+        return None
+
+    try:
+        bundle = await _create(scan_id, extra_files=_nothing())
+        backend.gate.set()
+        await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
+        assert backend.session.writes == []
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_default_path_still_blocks_until_the_sandbox_is_up(backend: _Backend) -> None:
+    scan_id = "deferred-off"
+    backend.gate.set()
+    try:
+        bundle = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
+        assert bundle["session"] is backend.session
+        assert bundle["client"] is backend.client
+    finally:
+        await session_manager.cleanup(scan_id)
+    assert backend.client.deleted == [backend.session]
+
+
+@pytest.mark.asyncio
+async def test_sdk_session_prep_does_not_wait_for_the_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-sdk-prep"
+    try:
+        bundle = await _create(scan_id)
+        agent = SandboxAgent(
+            name="root", instructions="x", tools=[], capabilities=[Filesystem(), Shell()]
+        )
+        manager = SandboxRuntimeSessionManager(
+            starting_agent=agent,
+            sandbox_config=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
+            run_state=None,
+        )
+        session = await asyncio.wait_for(
+            manager.ensure_session(
+                agent=agent, capabilities=list(agent.capabilities), is_resumed_state=False
+            ),
+            timeout=1,
+        )
+        assert session is bundle["session"]
+        assert session.pending
+        backend.gate.set()
+        await asyncio.wait_for(session.resolve(), timeout=1)
+        assert not session.pending
+    finally:
+        await session_manager.cleanup(scan_id)

tests/test_inputs.py

@@ -310,8 +310,8 @@ def test_make_model_settings_forces_required_for_anyllm_routed_openai_model() ->
     assert settings.tool_choice == "required"
 
 
-def test_make_model_settings_disables_parallel_tool_calls_by_default() -> None:
-    assert make_model_settings("none", model_name="gpt-4o").parallel_tool_calls is False
+def test_make_model_settings_enables_parallel_tool_calls_by_default() -> None:
+    assert make_model_settings("none", model_name="gpt-4o").parallel_tool_calls is True
 
 
 def test_make_model_settings_omits_parallel_tool_calls_without_tools() -> None:
@@ -361,6 +361,58 @@ def test_make_model_settings_extra_headers_survive_reasoning_resolve() -> None:
     assert settings.extra_headers == {"X-Feature-Key": "svc"}
 
 
+def test_make_model_settings_sets_extra_body() -> None:
+    settings = make_model_settings(
+        "none",
+        model_name="openai/gpt-5",
+        extra_body={"service_tier": "flex", "safety_identifier": "user-hash"},
+    )
+
+    assert settings.extra_body == {"service_tier": "flex", "safety_identifier": "user-hash"}
+
+
+def test_make_model_settings_omits_extra_body_when_unset() -> None:
+    assert make_model_settings("none", model_name="gpt-4o").extra_body is None
+
+
+def test_make_model_settings_extra_body_survives_reasoning_resolve() -> None:
+    settings = make_model_settings(
+        "high",
+        model_name="openai/o3",
+        extra_body={"service_tier": "flex"},
+    )
+
+    assert settings.extra_body == {"service_tier": "flex"}
+
+
+def test_make_model_settings_extra_body_merges_with_max_reasoning_field() -> None:
+    settings = make_model_settings(
+        "max",
+        model_name="deepseek/deepseek-v4-flash",
+        extra_body={"safety_identifier": "user-hash"},
+    )
+
+    assert settings.extra_body == {"safety_identifier": "user-hash", "reasoning_effort": "max"}
+
+
+def test_make_model_settings_max_reasoning_overrides_extra_body_reasoning_field() -> None:
+    settings = make_model_settings(
+        "max",
+        model_name="deepseek/deepseek-v4-flash",
+        extra_body={"reasoning_effort": "low"},
+    )
+
+    assert settings.extra_body == {"reasoning_effort": "max"}
+
+
+def test_make_model_settings_does_not_mutate_caller_extra_body() -> None:
+    extra_body = {"safety_identifier": "user-hash"}
+
+    make_model_settings("max", model_name="deepseek/deepseek-v4-flash", extra_body=extra_body)
+
+    assert extra_body == {"safety_identifier": "user-hash"}
+
+
 def test_make_model_settings_timeout_survives_reasoning_resolve() -> None:
     # Reasoning is resolved via ModelSettings.resolve(); the timeout in extra_args
     # must not be dropped when a reasoning override is merged in.

tests/test_llm_extra_body.py

@@ -0,0 +1,97 @@
+"""Tests for LLM_EXTRA_BODY: extra fields merged into every LLM request body."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+
+import httpx
+import litellm
+import openai
+import pytest
+from agents.extensions.models.litellm_model import LitellmModel
+from agents.models.interface import ModelTracing
+from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
+from agents.models.openai_responses import OpenAIResponsesModel
+
+from strix.config import loader
+from strix.config.loader import load_settings
+from strix.core.inputs import make_model_settings
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from agents.models.interface import Model
+
+
+_EXTRA_BODY = {"service_tier": "flex", "safety_identifier": "user-hash"}
+
+
+@pytest.fixture(autouse=True)
+def _reset(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    monkeypatch.delenv("LLM_EXTRA_BODY", raising=False)
+    monkeypatch.setattr(loader, "_cached", None)
+    monkeypatch.setattr(loader, "_override", None)
+    yield
+
+
+def test_extra_body_parsed_from_json_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"service_tier": "flex", "n": 1}))
+
+    assert load_settings().llm.extra_body == {"service_tier": "flex", "n": 1}
+
+
+async def _send(model: Model) -> None:
+    await model.get_response(
+        system_instructions=None,
+        input="Reply with just 'OK'.",
+        model_settings=make_model_settings(None, model_name="gpt-5", extra_body=_EXTRA_BODY),
+        tools=[],
+        output_schema=None,
+        handoffs=[],
+        tracing=ModelTracing.DISABLED,
+        previous_response_id=None,
+        conversation_id=None,
+        prompt=None,
+    )
+
+
+def _capturing_client(captured: dict[str, Any]) -> openai.AsyncOpenAI:
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured.update(json.loads(request.content))
+        return httpx.Response(400, json={"error": {"message": "captured"}})
+
+    return openai.AsyncOpenAI(
+        api_key="test",
+        base_url="https://gateway.example/v1",
+        max_retries=0,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("model_cls", [OpenAIChatCompletionsModel, OpenAIResponsesModel])
+async def test_extra_body_reaches_native_openai_request_body(model_cls: type) -> None:
+    captured: dict[str, Any] = {}
+
+    with pytest.raises(openai.BadRequestError):
+        await _send(model_cls(model="gpt-5", openai_client=_capturing_client(captured)))
+
+    assert {k: captured.get(k) for k in _EXTRA_BODY} == _EXTRA_BODY
+
+
+@pytest.mark.asyncio
+async def test_extra_body_passed_to_litellm(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_acompletion(**kwargs: Any) -> Any:
+        captured.update(kwargs)
+        raise RuntimeError("captured")
+
+    monkeypatch.setattr(litellm, "acompletion", fake_acompletion)
+
+    with pytest.raises(RuntimeError, match="captured"):
+        await _send(LitellmModel(model="openai/gpt-5", base_url="https://gateway.example/v1"))
+
+    assert captured["extra_body"] == _EXTRA_BODY

tests/test_preflight_timeout.py

@@ -60,6 +60,7 @@ def test_preflight_request_fails_after_its_own_timeout_with_a_clear_message() ->
                 model,  # type: ignore[arg-type]
                 model_name="openai/gpt-4o",
                 extra_headers=None,
+                extra_body=None,
                 timeout=1,
                 api_base_setting="DEDUPE_LLM_API_BASE",
             )
@@ -130,3 +131,52 @@ def test_warm_up_checks_the_dedupe_model_with_its_own_headers_and_the_preflight_
         "DEDUPE_LLM_API_BASE",
     ]
     assert calls[1][0] is dedupe_model
+
+
+def test_warm_up_checks_each_model_with_its_own_body_fields(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("STRIX_LLM", "openai/gpt-4o")
+    monkeypatch.setenv("LLM_API_KEY", "sk-test")
+    monkeypatch.setenv("LLM_EXTRA_BODY", '{"safety_identifier": "main"}')
+    monkeypatch.setenv("STRIX_DEDUPE_MODEL", "anthropic/claude-sonnet-4-5")
+    monkeypatch.setenv("DEDUPE_LLM_EXTRA_BODY", '{"service_tier": "flex"}')
+    _fresh_settings(monkeypatch, tmp_path)
+
+    calls: list[dict[str, Any]] = []
+
+    async def record(_model: Any, **kwargs: Any) -> None:
+        calls.append(kwargs)
+
+    monkeypatch.setattr(scan_setup, "preflight_request", record)
+    monkeypatch.setattr(cli_main, "preflight_request", record)
+    monkeypatch.setattr("strix.config.models.configure_sdk_model_defaults", lambda _settings: None)
+    monkeypatch.setattr("strix.report.dedupe.resolve_dedupe_model", lambda _dedupe, _name: object())
+
+    asyncio.run(cli_main.warm_up_llm())
+
+    assert [kwargs["extra_body"] for kwargs in calls] == [
+        {"safety_identifier": "main"},
+        {"service_tier": "flex"},
+    ]
+
+
+def test_preflight_request_sends_the_extra_body() -> None:
+    seen: list[Any] = []
+
+    class _Recorder:
+        async def get_response(self, *, model_settings: Any, **_: Any) -> None:
+            seen.append(model_settings.extra_body)
+
+    asyncio.run(
+        preflight_request(
+            _Recorder(),  # type: ignore[arg-type]
+            model_name="openai/gpt-4o",
+            extra_headers=None,
+            extra_body={"safety_identifier": "main"},
+            timeout=5,
+            api_base_setting="LLM_API_BASE",
+        )
+    )
+
+    assert seen == [{"safety_identifier": "main"}]

tests/test_proxy_client.py

@@ -13,6 +13,7 @@ from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, cast
 
 import pytest
+from caido_sdk_client.errors import OperationUserError
 
 from strix.runtime.caido_handle import CaidoBootstrapHandle
 from strix.tools.proxy import caido_api, tools
@@ -252,3 +253,11 @@ async def test_existing_request_ids_queries_current_project(
 
     assert existing == {"1042"}
     assert looked_up == ["1042", "1088"]
+
+
+def test_err_logs_caido_errors_as_warnings(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level("WARNING", logger="strix.tools.proxy.tools"):
+        tools._err("list_requests", OperationUserError(""))
+
+    [record] = caplog.records
+    assert record.levelname == "WARNING"

tests/test_reporting_fields.py

@@ -663,7 +663,6 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
         == "src/render.ts:14 calls `_.template()`."
     )
     assert "**Usage analysis:**" in report["evidence"]
-    assert "not a proof of exploitability or of safety" in report["evidence"]
     # The level must never influence the rating — that comes from the contextual
     # breakdown, or from advisory_cvss when no breakdown applies.
     assert report["severity"] == "high"

tests/test_resume_picker.py

@@ -263,7 +263,11 @@ def test_narrow_terminals_never_wrap_a_row() -> None:
 
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX key reader")
 def test_posix_key_reader_handles_multibyte_and_arrows() -> None:
+    tty = pytest.importorskip("tty")
     master, slave = os.openpty()
+    # Without this the pty echoes the writes into an output queue nobody reads,
+    # and read_raw_key's TCSADRAIN restore blocks forever on macOS.
+    tty.setcbreak(slave)
     try:
         with os.fdopen(slave, "r+b", buffering=0) as stream:
             os.write(master, "\u00e9".encode())

tests/test_runner_interrupt.py

@@ -28,6 +28,7 @@ def _wire_runner(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
             timeout=300,
             prompt_cache=True,
             extra_headers=None,
+            extra_body=None,
         ),
         runtime=types.SimpleNamespace(max_context_images=3),
     )

tests/test_runner_mcp.py

@@ -44,6 +44,7 @@ def _settings() -> Any:
             timeout=300,
             prompt_cache=True,
             extra_headers=None,
+            extra_body=None,
         ),
         runtime=types.SimpleNamespace(max_context_images=3),
     )

tests/test_runner_rate_limit.py

@@ -42,6 +42,7 @@ async def test_persistent_rate_limit_stops_gracefully(
             timeout=300,
             prompt_cache=True,
             extra_headers=None,
+            extra_body=None,
         ),
         runtime=types.SimpleNamespace(max_context_images=3),
     )

tests/test_runner_root_prompt.py

@@ -64,6 +64,7 @@ def _patch_engine_scaffold(
             timeout=300,
             prompt_cache=True,
             extra_headers=None,
+            extra_body=None,
         ),
         runtime=types.SimpleNamespace(max_context_images=3),
     )

tests/test_runner_teardown.py

@@ -52,6 +52,7 @@ def _settings() -> Any:
             timeout=300,
             prompt_cache=True,
             extra_headers=None,
+            extra_body=None,
         ),
         runtime=types.SimpleNamespace(max_context_images=3),
     )

tests/test_tool_call_arguments.py

@@ -0,0 +1,300 @@
+"""Tests for keeping replayed tool-call arguments valid JSON.
+
+A model that emits a tool call with malformed ``arguments`` fails that one
+call, but the raw string is recorded in the session. Strict OpenAI-compatible
+servers validate every assistant tool call in the request, so each later turn
+is rejected and the agent can never recover. A gateway that validates
+arguments the way those servers do proves both the failure and the fix.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from agents import Agent, Runner, function_tool
+from agents.models.interface import Model, ModelProvider
+from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
+from agents.run import RunConfig
+from openai import AsyncOpenAI, BadRequestError
+from openai.types.responses import ResponseFunctionToolCall
+
+from strix.config.models import _NonStreamingModel, _TurnGuardModel
+from strix.config.tool_call_arguments import (
+    MALFORMED_ARGUMENTS_KEY,
+    _loads_strict,
+    describe_malformed_arguments,
+    repair_arguments,
+    repair_history_arguments,
+    repair_input,
+)
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+TRUNCATED = '{"n": 1'
+OVER_NESTED = '{"a":' + "[" * 100_000
+
+
+def _tool_call_completion(arguments: str) -> dict[str, Any]:
+    return {
+        "id": "chatcmpl-1",
+        "object": "chat.completion",
+        "created": 0,
+        "model": "gw-model",
+        "choices": [
+            {
+                "index": 0,
+                "finish_reason": "tool_calls",
+                "message": {
+                    "role": "assistant",
+                    "content": None,
+                    "tool_calls": [
+                        {
+                            "id": "call_1",
+                            "type": "function",
+                            "function": {"name": "do_thing", "arguments": arguments},
+                        }
+                    ],
+                },
+            }
+        ],
+        "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7},
+    }
+
+
+def _text_completion(text: str) -> dict[str, Any]:
+    return {
+        "id": "chatcmpl-2",
+        "object": "chat.completion",
+        "created": 0,
+        "model": "gw-model",
+        "choices": [
+            {"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": text}}
+        ],
+        "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8},
+    }
+
+
+_REQUESTS: list[list[dict[str, Any]]] = []
+
+
+def _assistant_arguments(messages: list[dict[str, Any]]) -> list[str]:
+    return [
+        str(call["function"]["arguments"])
+        for message in messages
+        for call in message.get("tool_calls") or []
+    ]
+
+
+def _tool_results(messages: list[dict[str, Any]]) -> list[str]:
+    return [str(m.get("content")) for m in messages if m.get("role") == "tool"]
+
+
+class _StrictHandler(BaseHTTPRequestHandler):
+    """Gateway that rejects malformed assistant tool-call arguments, like vLLM/SGLang do."""
+
+    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 arguments in _assistant_arguments(messages):
+            try:
+                json.loads(arguments)
+            except ValueError:
+                self._respond(
+                    400,
+                    {
+                        "object": "error",
+                        "message": "Assistant tool call function.arguments must be valid JSON.",
+                        "type": "BadRequest",
+                        "param": None,
+                        "code": 400,
+                    },
+                )
+                return
+
+        if len(_REQUESTS) == 1:
+            self._respond(200, _tool_call_completion(TRUNCATED))
+        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)
+
+
+@pytest.fixture
+def strict_gateway() -> Iterator[str]:
+    _REQUESTS.clear()
+    server = HTTPServer(("127.0.0.1", 0), _StrictHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_address[1]}/v1"
+    finally:
+        server.shutdown()
+        server.server_close()
+
+
+def _model(base_url: str) -> Model:
+    client = AsyncOpenAI(api_key="tok", base_url=base_url, max_retries=0)
+    return _NonStreamingModel(OpenAIChatCompletionsModel(model="gw-model", openai_client=client))
+
+
+async def _run_agent(base_url: str, *, wrap: bool) -> Any:
+    @function_tool
+    def do_thing(n: int) -> str:
+        return f"did {n}"
+
+    class _Provider(ModelProvider):
+        def get_model(self, model_name: str | None) -> Model:  # noqa: ARG002
+            model = _model(base_url)
+            return _TurnGuardModel(model) if wrap else model
+
+    agent = Agent(name="t", instructions="use the tool", tools=[do_thing], model="gw-model")
+    result = Runner.run_streamed(
+        agent, input="please", run_config=RunConfig(model_provider=_Provider())
+    )
+    async for _ in result.stream_events():
+        pass
+    return result
+
+
+@pytest.mark.asyncio
+async def test_malformed_arguments_poison_every_later_turn_without_the_wrapper(
+    strict_gateway: str,
+) -> None:
+    # Repro: the model truncates a tool call's arguments once. The tool fails
+    # that call gracefully, but the next request replays the raw string and
+    # the provider rejects the whole conversation from then on.
+    with pytest.raises(BadRequestError, match="must be valid JSON"):
+        await _run_agent(strict_gateway, wrap=False)
+
+    assert _assistant_arguments(_REQUESTS[-1]) == [TRUNCATED]
+
+
+@pytest.mark.asyncio
+async def test_malformed_arguments_are_replayed_as_valid_json(strict_gateway: str) -> None:
+    result = await _run_agent(strict_gateway, wrap=True)
+
+    assert result.final_output == "all done"
+    (arguments,) = _assistant_arguments(_REQUESTS[-1])
+    assert json.loads(arguments) == {MALFORMED_ARGUMENTS_KEY: TRUNCATED}
+    (tool_result,) = _tool_results(_REQUESTS[-1])
+    assert "JSON" in tool_result
+
+
+@pytest.mark.parametrize(
+    ("arguments", "expected"),
+    [
+        ('{"cmd": "ls', json.dumps({MALFORMED_ARGUMENTS_KEY: '{"cmd": "ls'})),
+        ("[1, 2]", json.dumps({MALFORMED_ARGUMENTS_KEY: "[1, 2]"})),
+        ("", "{}"),
+        ("  ", "{}"),
+        (None, "{}"),
+    ],
+)
+def test_repair_arguments_rewrites_anything_but_a_json_object(
+    arguments: str | None, expected: str
+) -> None:
+    assert repair_arguments(arguments) == expected
+
+
+@pytest.mark.parametrize(
+    "arguments", ["{}", '{"cmd": "ls -la"}', '{"nested": {"a": [1]}}', '{"n": "NaN"}']
+)
+def test_repair_arguments_leaves_json_objects_alone(arguments: str) -> None:
+    assert repair_arguments(arguments) is None
+
+
+@pytest.mark.parametrize("arguments", ['{"n": NaN}', '{"n": Infinity}', '{"n": -Infinity}'])
+def test_repair_arguments_wraps_non_standard_json_constants(arguments: str) -> None:
+    repaired = repair_arguments(arguments)
+
+    assert repaired is not None
+    assert _loads_strict(repaired) == {MALFORMED_ARGUMENTS_KEY: arguments}
+
+
+def test_repair_arguments_wraps_over_nested_json() -> None:
+    repaired = repair_arguments(OVER_NESTED)
+
+    assert repaired is not None
+    assert _loads_strict(repaired) == {MALFORMED_ARGUMENTS_KEY: OVER_NESTED}
+
+
+def test_history_repair_rewrites_only_malformed_calls() -> None:
+    items = [
+        {"role": "user", "content": "go"},
+        {"type": "function_call", "call_id": "a", "name": "x", "arguments": '{"cmd": "ls'},
+        {"type": "function_call_output", "call_id": "a", "output": "invalid JSON"},
+        {"type": "function_call", "call_id": "b", "name": "y", "arguments": '{"ok": true}'},
+        ResponseFunctionToolCall(call_id="c", name="z", arguments="", type="function_call"),
+    ]
+
+    rebuilt, changed = repair_history_arguments(items)
+
+    assert changed
+    assert rebuilt[0] is items[0]
+    assert json.loads(rebuilt[1]["arguments"]) == {MALFORMED_ARGUMENTS_KEY: '{"cmd": "ls'}
+    assert rebuilt[1]["call_id"] == "a"
+    assert rebuilt[2] is items[2]
+    assert rebuilt[3] is items[3]
+    assert isinstance(rebuilt[4], ResponseFunctionToolCall)
+    assert rebuilt[4].arguments == "{}"
+    assert isinstance(items[1], dict)
+    assert items[1]["arguments"] == '{"cmd": "ls'
+
+
+def test_repair_input_returns_same_object_when_nothing_changes() -> None:
+    items = [{"type": "function_call", "call_id": "a", "name": "x", "arguments": "{}"}]
+
+    assert repair_input(items) is items
+    assert repair_input("plain prompt") == "plain prompt"
+
+
+def test_repair_input_wraps_over_nested_function_call_arguments() -> None:
+    items = [
+        {
+            "type": "function_call",
+            "call_id": "a",
+            "name": "x",
+            "arguments": OVER_NESTED,
+        }
+    ]
+
+    repaired = repair_input(items)
+
+    assert isinstance(repaired, list)
+    assert _loads_strict(repaired[0]["arguments"]) == {MALFORMED_ARGUMENTS_KEY: OVER_NESTED}
+
+
+@pytest.mark.parametrize(
+    "arguments",
+    ['{"cmd": "ls -la', "[1, 2]", "null", "not json", '{"n": NaN}', OVER_NESTED],
+)
+def test_describe_malformed_arguments_tells_the_model_to_reissue(arguments: str) -> None:
+    message = describe_malformed_arguments("exec_command", arguments)
+
+    assert message is not None
+    assert message.startswith("exec_command: the tool call was not executed")
+    assert "Re-issue the call" in message
+
+
+@pytest.mark.parametrize("arguments", ["", "   ", "{}", '{"cmd": "ls"}'])
+def test_describe_malformed_arguments_accepts_objects_and_empty_input(arguments: str) -> None:
+    assert describe_malformed_arguments("exec_command", arguments) is None