refactor(agents): name the report tool via the prompt template, not string replacement

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.
@@ -123,6 +141,10 @@ affecting the agents that do the actual testing.
   Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
 </ParamField>
 
+<ParamField path="STRIX_RUN_RECORD_TIMINGS" default="false" type="boolean">
+  Record every LLM request's and tool call's timing in `run.json`, under `llm_requests` and `tool_calls`. Adds about 0.5 MB to a typical scan's `run.json`.
+</ParamField>
+
 <ParamField path="TRACELOOP_BASE_URL" type="string">
   OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only.
 </ParamField>

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
@@ -676,6 +724,7 @@ def build_strix_agent(
     instructions_override: str | None = None,
     supports_images: bool = True,
     finish_tool: Tool = finish_scan,
+    report_tool: Tool = create_vulnerability_report,
 ) -> SandboxAgent[Any]:
     """Build a SandboxAgent for either root or child use.
 
@@ -689,6 +738,8 @@ def build_strix_agent(
         instructions_override: Use this verbatim as the system prompt instead
             of rendering the built-in scan prompt.
         finish_tool: The tool that ends the run, given to the root agent.
+        report_tool: The tool every agent files findings with, in place of
+            ``create_vulnerability_report``.
     """
     if instructions_override is not None:
         instructions = instructions_override
@@ -702,16 +753,18 @@ def build_strix_agent(
             interactive=interactive,
             system_prompt_context=system_prompt_context,
             supports_images=supports_images,
+            report_tool=report_tool.name,
         )
 
     agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
     if interactive:
         # Yielding to the user is only meaningful when one is attached.
         agent_tools.append(wait_for_user)
+    base_tools = [report_tool if t is create_vulnerability_report else t for t in _BASE_TOOLS]
     if is_root:
-        tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_tool]
+        tools: list[Tool] = [*base_tools, *agent_tools, finish_tool]
     else:
-        tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
+        tools = [*base_tools, *agent_tools, agent_finish]
     _ensure_unique_tool_names(tools)
     tools = [
         _with_bounded_result(_with_strictness(_with_coerced_arguments(tool), strict_tool_schemas))
@@ -764,6 +817,7 @@ def make_child_factory(
     strict_tool_schemas: bool = True,
     system_prompt_context: dict[str, Any] | None = None,
     supports_images: bool = True,
+    report_tool: Tool = create_vulnerability_report,
 ) -> Any:
     """Return the runner-owned builder used by ``spawn_child_agent``.
 
@@ -785,6 +839,7 @@ def make_child_factory(
             strict_tool_schemas=strict_tool_schemas,
             system_prompt_context=system_prompt_context,
             supports_images=supports_images,
+            report_tool=report_tool,
         )
 
     return _factory

strix/agents/prompt.py

@@ -86,6 +86,7 @@ def render_system_prompt(
     system_prompt_context: dict[str, Any] | None = None,
     include_scope: bool = True,
     supports_images: bool = True,
+    report_tool: str = "create_vulnerability_report",
 ) -> str:
     """Render the system prompt. Returns empty string on template failure.
 
@@ -133,6 +134,7 @@ def render_system_prompt(
             is_root=is_root,
             system_prompt_context=system_prompt_context or {},
             include_scope=include_scope,
+            report_tool=report_tool,
             **skill_content,
         )
     except Exception:

strix/agents/prompts/system_prompt.jinja

@@ -119,7 +119,7 @@ WHITE-BOX TESTING (code provided):
 - Local execution, unit/integration testing, patch verification, and HTTP requests against locally started in-scope services are normal authorized white-box validation
 - If dynamically running the code proves impossible after exhaustive attempts, pivot to comprehensive static analysis.
 - Try to infer how to run the code based on its structure and content.
-- Derive the code fix as PART OF reporting, not as a separate later pass: create_vulnerability_report already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
+- Derive the code fix as PART OF reporting, not as a separate later pass: {{ report_tool }} already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
 - If you also apply and verify the patch in the repo (edit the file, re-test that the vulnerability is gone), do it in the same agent/turn while the analysis is fresh — right before or as part of filing the report — never as a second re-analysis pass.
 
 COMBINED MODE (code + deployed target present):
@@ -205,9 +205,9 @@ VALIDATION REQUIREMENTS:
 - COVERAGE: record every surface you assess with `record_coverage` (surface + risk area + outcome + evidence), including the ones that came back clean — a report that only lists findings cannot say what was reviewed and cleared. Use the `needs_follow_up` outcome for anything left in an `open_proof_gap` state, and carry the same items up in `agent_finish(open_items=[...])`. The ledger is shared and mutable: when you resolve a surface another agent left open — or find that a closed one is not — move that entry with `update_coverage` instead of recording a second one for the same surface. The root agent reconciles all of it via `list_coverage` before `finish_scan`.
 - THREAT MODEL: before you start testing, call `get_threat_model` on the target you were pointed at — it is the scan's shared answer to who the attacker is, where the trust boundaries sit, and what counts as critical here. It is scoped to this scan and nothing carries over from an earlier run, so `found: false` means no agent on this run has derived one yet. Read it instead of re-deriving trust boundaries yourself; where your testing disproves it — a boundary it calls trusted turns out to be attacker-reachable, a role it did not know about, a host or endpoint it never listed — record that with `amend_threat_model` so the agents after you inherit the correction. Amending is not optional politeness: a model nobody corrects turns the first agent's guesses into everyone's assumptions.
 - Before filing any report, run the counterevidence pass: argue the strongest case AGAINST the finding, record what you found in the `counterevidence` field, set `confidence` honestly (a static-only trace you couldn't execute is at best `medium`), and state what evidence would change the severity. See the counterevidence and severity-calibration knowledge above.
-- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
-- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
-- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
+- A vulnerability is ONLY considered reported when a reporting agent uses {{ report_tool }} (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
+- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via {{ report_tool }} (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
+- DEDUPLICATION: The {{ report_tool }} tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
 - HTTP EVIDENCE: a finding you validated through the proxy is not fully filed until `http_exchange_ids` carries the proxy request ids of the exchanges that prove it — the request that triggers the vulnerability plus the baseline/control request it differs from (an unauthenticated success next to the authenticated one, the payload response next to the benign one). Copy the ids exactly as `list_requests`/`view_request` show them, never invent or guess one, and never omit the field to bypass validation. Leave it out only when there is no captured HTTP exchange at all (static-only code findings, dependency CVEs). If you filed before the proving exchanges existed, attach them afterwards with update_vulnerability_report. Without the ids, the finding ships as prose nobody can replay.
 - REVISING A FINDING: use update_vulnerability_report (report id + the fields you want to replace + update_reason) when you learn something a finding already on file does not carry — you built the PoC after filing it, a chain raised its impact, further testing weakened it, or its counterevidence/remediation/code locations were wrong. Editing a finding needs no duplicate verdict, and it is always better than filing a second report for the same issue. Read the finding first with get_report, and pass only the fields that change.
 - WITHDRAWING A FINDING: when re-testing proves a finding you filed is not a vulnerability at all — the exploit only worked because of your own test setup (a mixed-up session, a self-inflicted state change, a misread response), the behaviour is intended, or the control is in fact enforced — delete it with delete_vulnerability_report (report id + delete_reason). Never leave a disproved finding on file as a zero-CVSS, "retracted", or "false positive" report, and never rewrite its text into a note saying it was wrong: a report that exists is read as a finding. Delete is for a finding that does not exist; a finding that is real but weaker than filed is revised, not deleted.
@@ -227,7 +227,7 @@ Every one of these tools writes to state the rest of the scan reads. Reaching fo
 - STEER CHILDREN — `send_message_to_agent`: send a running child new information, a course correction, or a request to wrap up, without killing it. Use it to answer a child's question or narrow its scope mid-run.
 - BLOCK ON CHILDREN — `wait_for_agents`: block until named children report back when your next move genuinely depends on their results. If you can keep making progress in parallel, keep working instead of waiting.
 - CANCEL CHILDREN — `stop_agent`: gracefully cancel a child whose work is redundant, misdirected, or no longer needed. Prefer `send_message_to_agent` to redirect a child that is merely off-track; reserve `stop_agent` for work that should not continue at all.
-- FINISH — subagents call `agent_finish` (with `open_items=[...]` for anything left unresolved); the root agent calls `finish_scan` exactly once, only after every child is wrapped up and coverage is reconciled. `agent_finish`/`finish_scan` are handoffs, not reporting channels — a vulnerability is reported only via `create_vulnerability_report`/`create_dependency_report`.
+- FINISH — subagents call `agent_finish` (with `open_items=[...]` for anything left unresolved); the root agent calls `finish_scan` exactly once, only after every child is wrapped up and coverage is reconciled. `agent_finish`/`finish_scan` are handoffs, not reporting channels — a vulnerability is reported only via `{{ report_tool }}`/`create_dependency_report`.
 </execution_guidelines>
 
 <vulnerability_focus>
@@ -380,7 +380,7 @@ CRITICAL RULES:
 - **REALISTIC OUTCOMES** - Some tests find nothing, some validations fail
 - **ONE AGENT = ONE TASK** - Don't let agents do multiple unrelated jobs
 - **SPAWN REACTIVELY** - Create new agents based on what you discover
-- **ONLY REPORTING AGENTS** can use create_vulnerability_report tool
+- **ONLY REPORTING AGENTS** can use {{ report_tool }} tool
 - **AGENT SPECIALIZATION MANDATORY** - Each agent must be highly specialized; prefer 1–3 skills, up to 5 for complex contexts
 - **NO GENERIC AGENTS** - Avoid creating broad, multi-purpose agents that dilute focus
 

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],
@@ -694,7 +705,11 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
     """Apply Strix config to SDK-native defaults."""
     llm = settings.llm
     set_tracing_disabled(True)
+    from strix.report.state import record_llm_request
+
     request_log.install()
+    if settings.telemetry.run_record_timings:
+        request_log.register_sink(record_llm_request)
     if codex.subscription_model(llm.model):
         return
     _configure_litellm_compatibility()

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
@@ -153,6 +163,7 @@ class TelemetrySettings(BaseSettings):
     model_config = _BASE_CONFIG
 
     enabled: bool = Field(default=True, alias="STRIX_TELEMETRY")
+    run_record_timings: bool = Field(default=False, alias="STRIX_RUN_RECORD_TIMINGS")
 
 
 WebSearchProvider = Literal["auto", "perplexity", "exa"]

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

@@ -599,7 +599,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 +612,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 +832,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)

strix/core/hooks.py

@@ -4,18 +4,21 @@ from __future__ import annotations
 
 import logging
 import math
+import time
+from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
 
 from agents.lifecycle import RunHooks
 
 from strix.core.agents import BudgetPolicy, coordinator_from_context
-from strix.report.state import get_global_report_state
+from strix.report.state import get_global_report_state, record_tool_call
 
 
 if TYPE_CHECKING:
     from agents import RunContextWrapper
     from agents.agent import Agent
     from agents.items import ModelResponse, TResponseInputItem
+    from agents.tool import Tool
 
 
 logger = logging.getLogger(__name__)
@@ -65,9 +68,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:
@@ -160,6 +166,7 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
         max_turns: int | None = None,
         interactive: bool = False,
         budget_policy: BudgetPolicy = "stop",
+        record_tool_timings: bool = False,
     ) -> None:
         _validate_budget(max_budget_usd)
         if max_turns is not None and max_turns <= 0:
@@ -172,6 +179,8 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
         self._max_turns = max_turns
         self._interactive = interactive
         self._budget_policy: BudgetPolicy = budget_policy
+        self._record_tool_timings = record_tool_timings
+        self._tool_starts: dict[str, tuple[datetime, float]] = {}
 
     @property
     def max_budget_usd(self) -> float | None:
@@ -198,6 +207,7 @@ 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
@@ -207,6 +217,15 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
         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.
 
@@ -302,6 +321,37 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
             )
         input_items.append({"role": "user", "content": content})
 
+    async def on_tool_start(
+        self,
+        context: RunContextWrapper[dict[str, Any]],
+        agent: Agent[dict[str, Any]],  # noqa: ARG002
+        tool: Tool,  # noqa: ARG002
+    ) -> None:
+        call_id = getattr(context, "tool_call_id", None)
+        if self._record_tool_timings and isinstance(call_id, str):
+            self._tool_starts[call_id] = (datetime.now(UTC), time.monotonic())
+
+    async def on_tool_end(
+        self,
+        context: RunContextWrapper[dict[str, Any]],
+        agent: Agent[dict[str, Any]],  # noqa: ARG002
+        tool: Tool,
+        result: object,  # noqa: ARG002
+    ) -> None:
+        call_id = getattr(context, "tool_call_id", None)
+        start = self._tool_starts.pop(call_id, None) if isinstance(call_id, str) else None
+        if start is None:
+            return
+        started_at, started = start
+        ctx = context.context if isinstance(context.context, dict) else {}
+        agent_id = ctx.get("agent_id")
+        record_tool_call(
+            started_at=started_at,
+            agent_id=agent_id if isinstance(agent_id, str) else None,
+            tool=tool.name,
+            duration_ms=round((time.monotonic() - started) * 1000),
+        )
+
     async def on_llm_end(
         self,
         context: RunContextWrapper[dict[str, Any]],

strix/core/inputs.py

@@ -250,6 +250,7 @@ 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)
@@ -259,13 +260,16 @@ def make_model_settings(
         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

@@ -54,6 +54,7 @@ from strix.tools.output_store import (
     WORKSPACE_SPILL_DIR,
     configure_spill_writer,
 )
+from strix.tools.reporting.tool import create_vulnerability_report
 
 
 if TYPE_CHECKING:
@@ -172,6 +173,7 @@ def _compose_root_instructions_override(
     interactive: bool,
     system_prompt_context: dict[str, Any],
     supports_images: bool,
+    report_tool: str,
 ) -> str | None:
     if root_instructions_override is None:
         return None
@@ -186,6 +188,7 @@ def _compose_root_instructions_override(
         system_prompt_context=system_prompt_context,
         include_scope=False,
         supports_images=supports_images,
+        report_tool=report_tool,
     )
     return (
         f"{base_instructions}\n\n"
@@ -218,6 +221,7 @@ async def run_strix_scan(
     mcp_connection_requests: list[McpConnectionRequest] | None = None,
     mcp_status_sink: McpStatusSink | None = None,
     root_finish_tool: Tool = finish_scan,
+    report_tool: Tool = create_vulnerability_report,
 ) -> RunResultBase | None:
     """Run or resume one Strix scan against a sandbox.
 
@@ -241,6 +245,8 @@ async def run_strix_scan(
     way the engine does the connecting, so the caller passes inert configs plus
     metadata and never live sessions.
     ``root_finish_tool`` is the tool the root agent ends the run with.
+    ``report_tool`` is the tool every agent files findings with, in place of
+    ``create_vulnerability_report``.
     """
 
     def report(phase: str) -> None:
@@ -401,6 +407,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,
@@ -418,6 +425,7 @@ async def run_strix_scan(
             max_turns=max_turns,
             interactive=interactive,
             budget_policy=budget_policy,
+            record_tool_timings=settings.telemetry.run_record_timings,
         )
         coordinator.set_budget_limit_setter(hooks.set_max_budget_usd)
         if interactive and budget_policy != "pause":
@@ -493,6 +501,7 @@ async def run_strix_scan(
             interactive=interactive,
             system_prompt_context=root_context,
             supports_images=supports_images,
+            report_tool=report_tool.name,
         )
 
         root_agent = build_strix_agent(
@@ -509,6 +518,7 @@ async def run_strix_scan(
             instructions_override=root_instructions,
             supports_images=supports_images,
             finish_tool=root_finish_tool,
+            report_tool=report_tool,
         )
 
         if not is_resume:
@@ -529,6 +539,7 @@ async def run_strix_scan(
             strict_tool_schemas=strict_tool_schemas,
             system_prompt_context=scope_context,
             supports_images=supports_images,
+            report_tool=report_tool,
         )
 
         async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]:

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/report/state.py

@@ -28,6 +28,8 @@ from strix.telemetry import posthog, scarf
 if TYPE_CHECKING:
     from agents.usage import Usage
 
+    from strix.llm.request_log import LlmRequestEvent
+
 
 logger = logging.getLogger(__name__)
 
@@ -1019,6 +1021,47 @@ class StreamedOpenRouterCosts:
 streamed_openrouter_costs = StreamedOpenRouterCosts()
 
 
+def record_llm_request(event: "LlmRequestEvent") -> None:
+    """Request-log sink: each provider attempt's timing, kept in run.json's ``llm_requests``."""
+    report_state = get_global_report_state()
+    if report_state is None:
+        return
+    details = event.details or {}
+    report_state.run_record.setdefault("llm_requests", []).append(
+        {
+            "started_at": event.started_at.isoformat(),
+            "agent_id": event.agent_id,
+            "model": event.model,
+            "provider": details.get("upstream_provider") or event.provider,
+            "outcome": event.outcome,
+            "status": event.status_code,
+            "duration_ms": event.duration_ms,
+            "ttft_ms": event.time_to_first_token_ms,
+            "input_tokens": event.input_tokens,
+            "cached_tokens": event.cached_input_tokens,
+            "output_tokens": event.output_tokens,
+            "retry_attempt": event.retry_attempt,
+        }
+    )
+
+
+def record_tool_call(
+    *, started_at: datetime, agent_id: str | None, tool: str, duration_ms: int
+) -> None:
+    """Each tool call's timing, kept in run.json's ``tool_calls``."""
+    report_state = get_global_report_state()
+    if report_state is None:
+        return
+    report_state.run_record.setdefault("tool_calls", []).append(
+        {
+            "started_at": started_at.isoformat(),
+            "agent_id": agent_id,
+            "tool": tool,
+            "duration_ms": duration_ms,
+        }
+    )
+
+
 def record_openrouter_provider(provider: Any, usage: Any) -> None:
     """Tally which upstream provider served a stream, from its final usage chunk.
 

strix/skills/analysis/counterevidence.md

@@ -20,7 +20,7 @@ state, and "I moved on" is not one of them.
 
 **1. `confirmed`** — you have a working PoC or, in white-box, a complete
 source → control → sink → impact trace plus evidence the path is
-reachable. File it with `create_vulnerability_report`.
+reachable. File it as a vulnerability report.
 
 **2. `ruled_out`** — you can name the **specific control** that makes the
 code safe, at a specific location, and you have checked that the control
@@ -144,8 +144,7 @@ endpoint is broken/unreachable for unrelated reasons".
 
 ## Before You File a Report
 
-Run this pass on every finding before calling
-`create_vulnerability_report`:
+Run this pass on every finding before filing it:
 
 1. **Argue the other side.** Spend real effort building the strongest
    case that this is *not* exploitable, or not as severe as you think.

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_agent_tool_registration.py

@@ -93,6 +93,17 @@ def test_per_call_extra_tools_reject_duplicate_registered_names() -> None:
         factory.build_strix_agent(is_root=True, extra_tools=[_tool("same_name")])
 
 
+def test_report_tool_replaces_create_vulnerability_report() -> None:
+    for is_root in (True, False):
+        agent = factory.build_strix_agent(is_root=is_root, report_tool=_tool("file_pr_finding"))
+        names = [t.name for t in agent.tools]
+        assert "file_pr_finding" in names
+        assert "create_vulnerability_report" not in names
+        assert isinstance(agent.instructions, str)
+        assert "file_pr_finding" in agent.instructions
+        assert "create_vulnerability_report" not in agent.instructions
+
+
 def test_instructions_override_is_used_verbatim() -> None:
     custom = "You are a scan agent. Follow the provided scope."
 

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

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import uuid
+from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any
 from unittest.mock import MagicMock, call, patch
@@ -18,11 +19,13 @@ from strix.config.models import (
     _configure_litellm_compatibility,
     _install_openrouter_stream_cost_capture,
 )
+from strix.core.hooks import ReportUsageHooks
 from strix.llm import request_log
 from strix.report.state import (
     ReportState,
     litellm_cost_callback,
     openrouter_stream_cost,
+    record_llm_request,
     set_global_report_state,
     streamed_openrouter_costs,
 )
@@ -136,6 +139,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)
@@ -454,3 +458,73 @@ def test_openrouter_records_upstream_of_pre_stream_error() -> None:
         request_log._http_reply.reset(token)
     assert reply.upstream_provider == "InferenceNet"
     assert reply.upstream_error_type is None
+
+
+def test_request_timings_land_in_run_record() -> None:
+    started = datetime(2026, 10, 19, 12, 0, tzinfo=UTC)
+    event = request_log.LlmRequestEvent(
+        call_id="call-1",
+        route="litellm",
+        provider="openrouter",
+        model="openrouter/deepseek/deepseek-v4.1-flash",
+        api_host="openrouter.ai",
+        streaming=True,
+        outcome="success",
+        status_code=200,
+        provider_request_id=None,
+        response_id="gen-1",
+        error_type=None,
+        error_message=None,
+        started_at=started,
+        finished_at=started + timedelta(milliseconds=4200),
+        duration_ms=4200,
+        input_tokens=90_000,
+        output_tokens=600,
+        cached_input_tokens=85_000,
+        agent_id="agent-1",
+        time_to_first_token_ms=900,
+        details={"upstream_provider": "Together"},
+    )
+    state = ReportState.__new__(ReportState)
+    state.run_record = {}
+    try:
+        set_global_report_state(state)
+        record_llm_request(event)
+    finally:
+        set_global_report_state(None)
+
+    assert state.run_record["llm_requests"] == [
+        {
+            "started_at": "2026-10-19T12:00:00+00:00",
+            "agent_id": "agent-1",
+            "model": "openrouter/deepseek/deepseek-v4.1-flash",
+            "provider": "Together",
+            "outcome": "success",
+            "status": 200,
+            "duration_ms": 4200,
+            "ttft_ms": 900,
+            "input_tokens": 90_000,
+            "cached_tokens": 85_000,
+            "output_tokens": 600,
+            "retry_attempt": 0,
+        }
+    ]
+
+
+async def test_tool_call_timings_land_in_run_record() -> None:
+    hooks = ReportUsageHooks(model="m", record_tool_timings=True)
+    context = SimpleNamespace(tool_call_id="call-1", context={"agent_id": "agent-1"})
+    tool = SimpleNamespace(name="terminal_execute")
+    state = ReportState.__new__(ReportState)
+    state.run_record = {}
+    try:
+        set_global_report_state(state)
+        await hooks.on_tool_start(context, MagicMock(), tool)  # type: ignore[arg-type]
+        await hooks.on_tool_end(context, MagicMock(), tool, "ok")  # type: ignore[arg-type]
+    finally:
+        set_global_report_state(None)
+
+    [entry] = state.run_record["tool_calls"]
+    assert entry["agent_id"] == "agent-1"
+    assert entry["tool"] == "terminal_execute"
+    assert entry["duration_ms"] >= 0

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_inputs.py

@@ -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_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