Compare commits

..
Author SHA1 Message Date
Alex Schapiro 08813c4708 fix(models): treat over-nested tool-call arguments as malformed 2026-10-14 21:55:41 +00:00
Alex Schapiro 3232409d1c test(cost): pin model resolution in the provider-prefixed cost test 2026-10-14 21:30:33 +00:00
Alex Schapiro b6f38bc90c fix(models): treat NaN/Infinity tool-call arguments as malformed 2026-10-14 21:17:10 +00:00
Alex Schapiro 6753a5610a Merge remote-tracking branch 'origin/main' into fix/malformed-tool-call-arguments
# Conflicts:
#	strix/config/models.py
2026-10-14 21:11:18 +00:00
ian-at-strix e21f9ce4cc test(resume-picker): put the pty in cbreak so the key-reader test can't hang on macOS (#1527) 2026-10-14 15:55:22 -04:00
yoni-at-strix 87f2d783d6 fix(reporting): drop the static-analysis disclaimer from dependency evidence (#1520) 2026-10-14 10:59:11 -04:00
Ahmed Allam 956a26715b feat(runtime): accept an awaitable extra_files in create_or_reuse so the sandbox create does not wait for it 2026-10-14 03:32:29 +03:00
Ahmed Allam 0f8db1fca8 fix(runtime): keep the default path blocking on the proxy endpoint and discard a sandbox created after a cancelled deferred bring-up 2026-10-14 00:48:53 +03:00
Ahmed Allam 4bf7541599 feat(runtime): defer the sandbox bring-up behind a DeferredSandboxSession when create_or_reuse is called with defer=True 2026-10-14 00:48:53 +03:00
Ian d8c00a8f4e fix(proxy): log all Caido SDK errors as warnings 2026-10-12 22:18:42 +03:00
Ian c5854e7be2 fix(proxy): log Caido timeouts as warnings 2026-10-12 22:18:42 +03:00
Alex Schapiro 9f869c3f3f fix(agents): tell the model to re-issue tool calls whose arguments are not valid JSON
A streamed turn that ends mid-tool-call leaves the client with a JSON prefix
as the arguments. Instead of the SDK's generic parse-error result, the tool
wrapper now short-circuits with a clear 'not executed, re-issue with complete
arguments' message so the model recovers in one turn.
2026-09-17 14:15:02 +00:00
Alex Schapiro b6668d13b3 fix(models): replay malformed tool-call arguments as valid JSON
A model that emits a tool call with truncated or otherwise invalid JSON
arguments fails that one call, but the raw string is recorded in the
session. Strict OpenAI-compatible servers (vLLM, SGLang) validate every
assistant tool call in the request, so each later turn is rejected with
"Assistant tool call function.arguments must be valid JSON" and the agent
never recovers. The turn guard now rewrites such arguments on the way out
to a JSON object carrying the original text, alongside the existing
duplicate call-id repair.
2026-09-17 03:53:28 +00:00
13 changed files with 1110 additions and 44 deletions

No files matched your search

+5
View File
@@ -18,6 +18,7 @@ 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 +268,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
+13 -2
View File
@@ -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],
+100
View File
@@ -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
+2 -2
View File
@@ -8,7 +8,7 @@ import io
import json
import logging
import uuid
from collections.abc import Callable
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import TYPE_CHECKING, Any
@@ -203,7 +203,7 @@ async def run_strix_scan(
scan_id: str | None = None,
image: str,
local_sources: list[dict[str, Any]] | None = None,
extra_files: list[dict[str, Any]] | None = None,
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
coordinator: AgentCoordinator | None = None,
interactive: bool = False,
max_turns: int = DEFAULT_MAX_TURNS,
+255
View File
@@ -0,0 +1,255 @@
from __future__ import annotations
import asyncio
import contextlib
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from agents.sandbox.session.manager import Instrumentation
from agents.sandbox.session.sandbox_session import SandboxSession
from agents.sandbox.session.sandbox_session_state import SandboxSessionState
from agents.sandbox.snapshot import NoopSnapshot
if TYPE_CHECKING:
import io
from collections.abc import Sequence
from agents.sandbox.entries import BaseEntry
from agents.sandbox.manifest import Manifest
from agents.sandbox.materialization import MaterializedFile
from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
from agents.sandbox.types import ExecResult, ExposedPortEndpoint, User
logger = logging.getLogger(__name__)
class _PendingSessionState(SandboxSessionState):
type: str = "pending"
class DeferredSandboxSession(SandboxSession):
def __init__(
self,
bring_up: asyncio.Task[tuple[Any, BaseSandboxSession]],
*,
manifest: Manifest,
exposed_ports: tuple[int, ...] = (),
) -> None:
self._bring_up = bring_up
self._pending_state: SandboxSessionState = _PendingSessionState(
snapshot=NoopSnapshot(id="pending"),
manifest=manifest,
exposed_ports=exposed_ports,
)
self._pending_limits: Any = None
self._pending_archive_limits: Any = None
self._resolved: BaseSandboxSession | None = None
self._client: Any = None
self._instrumentation: Instrumentation = Instrumentation()
self._seq = 0
@property
def pending(self) -> bool:
return self._resolved is None
@property
def bring_up(self) -> asyncio.Task[tuple[Any, BaseSandboxSession]]:
return self._bring_up
async def resolve(self) -> BaseSandboxSession:
if self._resolved is not None:
return self._resolved
client, session = await asyncio.shield(self._bring_up)
self._adopt(client, session)
return session
def _adopt(self, client: Any, session: BaseSandboxSession) -> None:
if self._resolved is not None:
return
self._client = client
self._resolved = session
self._inner = session
if self._pending_limits is not None:
session._set_concurrency_limits(self._pending_limits)
if self._pending_archive_limits is not None:
session._set_archive_limits(self._pending_archive_limits)
if self._pending_state.manifest != session.state.manifest:
session.state = session.state.model_copy(
update={"manifest": self._pending_state.manifest}
)
@property
def client(self) -> Any:
return self._client
@property
def state(self) -> SandboxSessionState:
if self._resolved is None:
return self._pending_state
return self._resolved.state
@state.setter
def state(self, value: SandboxSessionState) -> None:
if self._resolved is None:
self._pending_state = value
else:
self._resolved.state = value
def _set_concurrency_limits(self, limits: Any) -> None:
if self._resolved is None:
self._pending_limits = limits
else:
self._resolved._set_concurrency_limits(limits)
def _set_archive_limits(self, limits: Any) -> None:
if self._resolved is None:
self._pending_archive_limits = limits
else:
self._resolved._set_archive_limits(limits)
def _manifest_base_dir(self) -> Path:
if self._resolved is None:
return Path.cwd()
return self._resolved._manifest_base_dir()
def normalize_path(self, path: Path | str, *, for_write: bool = False) -> Path:
if self._resolved is None:
return Path(path)
return self._resolved.normalize_path(path, for_write=for_write)
def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
if self._resolved is None:
return Path(path)
return self._resolved.register_persist_workspace_skip_path(path)
def supports_pty(self) -> bool:
return self._resolved.supports_pty() if self._resolved is not None else False
async def running(self) -> bool:
if self._resolved is None:
if not self._bring_up.done():
return True
return not self._bring_up.cancelled() and self._bring_up.exception() is None
return await self._resolved.running()
async def start(self) -> None:
if self._resolved is None:
return
await self._resolved.start()
async def stop(self) -> None:
if self._resolved is None:
return
await self._resolved.stop()
async def shutdown(self) -> None:
if self._resolved is None:
return
await self._resolved.shutdown()
async def aclose(self) -> None:
if self._resolved is None:
return
await self._resolved.aclose()
async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None:
session = await self.resolve()
await session._validate_manifest_application(only_ephemeral=only_ephemeral)
async def apply_manifest(self, *, only_ephemeral: bool = False) -> Any:
session = await self.resolve()
return await session.apply_manifest(only_ephemeral=only_ephemeral)
async def _apply_entry_batch(
self,
entries: Sequence[tuple[Path, BaseEntry]],
*,
base_dir: Path,
) -> list[MaterializedFile]:
del base_dir
session = await self.resolve()
return await session._apply_entry_batch(entries, base_dir=session._manifest_base_dir())
async def exec(self, *command: str | Path, **kwargs: Any) -> ExecResult:
await self.resolve()
return await super().exec(*command, **kwargs)
async def ls(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().ls(*args, **kwargs)
async def rm(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().rm(*args, **kwargs)
async def mkdir(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().mkdir(*args, **kwargs)
async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
await self.resolve()
return await super().read(path, user=user)
async def write(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().write(*args, **kwargs)
async def resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
await self.resolve()
return await super().resolve_exposed_port(port)
async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
session = await self.resolve()
return await session._resolve_exposed_port(port)
async def pty_exec_start(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().pty_exec_start(*args, **kwargs)
async def pty_write_stdin(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().pty_write_stdin(*args, **kwargs)
async def pty_terminate_all(self) -> None:
if self._resolved is None:
return
await super().pty_terminate_all()
async def persist_workspace(self) -> io.IOBase:
await self.resolve()
return await super().persist_workspace()
async def hydrate_workspace(self, data: io.IOBase) -> None:
await self.resolve()
await super().hydrate_workspace(data)
async def teardown(self) -> None:
if self._resolved is None:
self._bring_up.cancel()
with contextlib.suppress(BaseException):
client, session = await self._bring_up
self._adopt(client, session)
if self._resolved is None:
return
await self._client.delete(self._resolved)
class DeferredSandboxClient:
def __init__(self, session: DeferredSandboxSession) -> None:
self._session = session
def __getattr__(self, name: str) -> Any:
client = self._session.client
if client is None:
raise AttributeError(f"sandbox client unavailable before bring-up finishes ({name})")
return getattr(client, name)
async def delete(self, session: Any) -> None:
if isinstance(session, DeferredSandboxSession):
await session.teardown()
return
real = await self._session.resolve()
await self._session.client.delete(real if session is self._session else session)
+93 -32
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
import asyncio
import contextlib
import inspect
import io
import logging
import os
@@ -19,9 +21,12 @@ from strix.config import load_settings
from strix.runtime.backends import backend_supports_bind_mounts, get_backend
from strix.runtime.caido_bootstrap import bootstrap_caido
from strix.runtime.caido_handle import CaidoBootstrapHandle
from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
if TYPE_CHECKING:
from collections.abc import Awaitable
from agents.sandbox.session import BaseSandboxSession
from strix.runtime.status import StatusSink
@@ -255,13 +260,22 @@ def _gitdir_from_pointer(git_file: Path) -> Path | None:
return None
async def _resolve_extra_file_archive(
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None,
local_sources: list[dict[str, Any]],
) -> bytes | None:
files = await extra_files if inspect.isawaitable(extra_files) else extra_files
return build_extra_file_archive(files, local_sources) if files else None
async def create_or_reuse(
scan_id: str,
*,
image: str,
local_sources: list[dict[str, Any]],
extra_files: list[dict[str, Any]] | None = None,
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
status_sink: StatusSink | None = None,
defer: bool = False,
) -> dict[str, Any]:
"""Return the existing session bundle for ``scan_id`` or create a new one.
@@ -271,7 +285,9 @@ async def create_or_reuse(
Each ``extra_files`` entry (``{"workspace_path": "/workspace/<rel>",
"content": bytes | str}``) lands as a regular, agent-writable file at its
``workspace_path`` on every backend: the files are uploaded as one archive
and unpacked inside the sandbox right after bring-up.
and unpacked inside the sandbox right after bring-up. ``extra_files`` may
also be an awaitable resolving to that list: provisioning starts at once and
the files are awaited only when the sandbox is ready to receive them.
"""
def report(phase: str) -> None:
@@ -292,9 +308,6 @@ async def create_or_reuse(
else:
bind_mounts = []
entries = build_manifest_entries(local_sources)
extra_file_archive = (
build_extra_file_archive(extra_files, local_sources) if extra_files else None
)
# Caido runs as an in-container sidecar; HTTP(S) traffic from any
# process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -323,41 +336,69 @@ async def create_or_reuse(
backend_name,
image,
)
report("Starting sandbox container")
client, session = await backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
if extra_file_archive is not None:
report("Placing workspace files")
async def _bring_up() -> tuple[Any, Any]:
report("Starting sandbox container")
client, session = await _create_or_discard(
backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
)
try:
await stage_extra_files(session, extra_file_archive)
extra_file_archive = await _resolve_extra_file_archive(extra_files, local_sources)
if extra_file_archive is not None:
report("Placing workspace files")
await stage_extra_files(session, extra_file_archive)
except BaseException:
await _discard_session(client, session)
raise
return client, session
report("Setting up the proxy")
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
async def _caido_url(session: Any) -> str:
report("Setting up the proxy")
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
return host_caido_url
# The Caido login + project setup polls the guest for a couple of seconds
# and nothing needs the client before the first proxy tool call, so it
# runs concurrently with the rest of scan start; consumers resolve the
# handle at first use (see CaidoBootstrapHandle).
caido_client = CaidoBootstrapHandle(
asyncio.create_task(
bootstrap_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
),
name=f"caido-bootstrap-{scan_id}",
def _caido(session: Any, host_caido_url: str) -> Any:
return bootstrap_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
)
if defer:
bring_up = asyncio.create_task(_bring_up(), name=f"sandbox-bring-up-{scan_id}")
deferred = DeferredSandboxSession(
bring_up,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
)
async def _caido_when_up() -> Any:
session = await deferred.resolve()
return await _caido(session, await _caido_url(session))
bundle = {
"client": DeferredSandboxClient(deferred),
"session": deferred,
"caido_client": CaidoBootstrapHandle(
asyncio.create_task(_caido_when_up(), name=f"caido-bootstrap-{scan_id}")
),
}
_SESSION_CACHE[scan_id] = bundle
logger.info("Sandbox session for scan %s deferred and cached", scan_id)
return bundle
client, session = await _bring_up()
host_caido_url = await _caido_url(session)
caido_client = CaidoBootstrapHandle(
asyncio.create_task(_caido(session, host_caido_url), name=f"caido-bootstrap-{scan_id}")
)
bundle = {
@@ -370,6 +411,26 @@ async def create_or_reuse(
return bundle
_DISCARD_TASKS: set[asyncio.Task[None]] = set()
async def _create_or_discard(create: Awaitable[tuple[Any, Any]]) -> tuple[Any, Any]:
pending = asyncio.ensure_future(create)
try:
return await asyncio.shield(pending)
except asyncio.CancelledError:
discard = asyncio.create_task(_discard_when_created(pending))
_DISCARD_TASKS.add(discard)
discard.add_done_callback(_DISCARD_TASKS.discard)
raise
async def _discard_when_created(create: asyncio.Future[tuple[Any, Any]]) -> None:
with contextlib.suppress(BaseException):
client, session = await create
await _discard_session(client, session)
async def _discard_session(client: Any, session: Any) -> None:
"""Best-effort teardown of a session that never made it into the cache."""
try:
-4
View File
@@ -1876,10 +1876,6 @@ def _build_dependency_evidence(
evidence += f"\n\n**Usage analysis:** {label}."
if reachability_evidence and reachability_evidence.strip():
evidence += f" {reachability_evidence.strip()}"
evidence += (
" This is a prioritization signal from static analysis, not a"
" proof of exploitability or of safety."
)
return evidence
+21 -3
View File
@@ -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
+1
View File
@@ -136,6 +136,7 @@ def test_cost_callback_estimates_cost_with_provider_prefixed_model() -> None:
with (
patch("strix.report.state.get_global_report_state", return_value=report_state),
patch("strix.report.state.resolve_litellm_model", side_effect=lambda model: model),
patch("litellm.completion_cost", side_effect=fake_completion_cost),
):
litellm_cost_callback(kwargs, response)
+316
View File
@@ -0,0 +1,316 @@
from __future__ import annotations
import asyncio
import io
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pytest
from agents.run_config import SandboxRunConfig
from agents.sandbox import SandboxAgent
from agents.sandbox.capabilities import Filesystem, Shell
from agents.sandbox.runtime_session_manager import SandboxRuntimeSessionManager
from strix.runtime import session_manager
from strix.runtime.backends import _BACKENDS, _BIND_MOUNT_BACKENDS, register_backend
from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
if TYPE_CHECKING:
from collections.abc import Iterator
@dataclass
class _RuntimeSettings:
backend: str
@dataclass
class _Settings:
runtime: _RuntimeSettings
@dataclass
class _Endpoint:
host: str = "127.0.0.1"
port: int = 8080
tls: bool = False
@dataclass
class _ExecResult:
exit_code: int = 0
stdout: bytes = b""
stderr: bytes = b""
def ok(self) -> bool:
return self.exit_code == 0
class _State:
def __init__(self) -> None:
self.manifest: Any = None
self.session_id = "00000000-0000-4000-8000-000000000001"
def model_copy(self, *, update: dict[str, Any]) -> _State:
copy = _State()
copy.manifest = update.get("manifest", self.manifest)
return copy
class _Session:
def __init__(self) -> None:
self.state = _State()
self.execs: list[tuple[str, ...]] = []
self.writes: list[tuple[Path, bytes]] = []
self.limits: list[Any] = []
def _set_concurrency_limits(self, limits: Any) -> None:
self.limits.append(limits)
def _set_archive_limits(self, limits: Any) -> None:
self.limits.append(limits)
async def running(self) -> bool:
return True
async def resolve_exposed_port(self, _port: int) -> _Endpoint:
return _Endpoint()
async def exec(self, *argv: str, timeout: float | None = None, **_kw: Any) -> _ExecResult:
del timeout
self.execs.append(argv)
return _ExecResult()
async def write(self, path: Path, data: io.IOBase, **_kw: Any) -> None:
self.writes.append((path, data.read()))
class _Client:
def __init__(self) -> None:
self.deleted: list[Any] = []
async def delete(self, session: Any) -> None:
self.deleted.append(session)
async def _no_caido(*_args: Any, **_kwargs: Any) -> None:
return None
class _Backend:
def __init__(self, *, fail: bool = False) -> None:
self.gate = asyncio.Event()
self.calls = 0
self.fail = fail
self.client = _Client()
self.session = _Session()
async def __call__(self, **_kwargs: Any) -> tuple[Any, Any]:
self.calls += 1
await self.gate.wait()
if self.fail:
raise RuntimeError("bring-up exploded")
return self.client, self.session
@pytest.fixture
def backend(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Iterator[_Backend]:
name = f"deferred-{request.node.name}"
fake = _Backend(fail="fail" in request.node.name)
register_backend(name, fake, supports_bind_mounts=True)
monkeypatch.setattr(session_manager, "bootstrap_caido", _no_caido)
monkeypatch.setattr(
session_manager, "load_settings", lambda: _Settings(runtime=_RuntimeSettings(backend=name))
)
yield fake
_BACKENDS.pop(name, None)
_BIND_MOUNT_BACKENDS.discard(name)
async def _settle() -> None:
for _ in range(10):
await asyncio.sleep(0)
async def _create(scan_id: str, **kwargs: Any) -> dict[str, Any]:
return await session_manager.create_or_reuse(
scan_id, image="img", local_sources=[], defer=True, **kwargs
)
@pytest.mark.asyncio
async def test_deferred_bundle_returns_before_the_backend_finishes(backend: _Backend) -> None:
scan_id = "deferred-returns"
try:
bundle = await asyncio.wait_for(_create(scan_id), timeout=1)
session = bundle["session"]
assert isinstance(session, DeferredSandboxSession)
assert isinstance(bundle["client"], DeferredSandboxClient)
assert session.pending
await _settle()
assert backend.calls == 1
assert await session.running()
assert session.state.manifest.environment.value["PYTHONUNBUFFERED"] == "1"
again = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
assert again is bundle
assert backend.calls == 1
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_first_io_resolves_the_sandbox_then_delegates(backend: _Backend) -> None:
scan_id = "deferred-io"
try:
bundle = await _create(scan_id)
session = bundle["session"]
session._set_concurrency_limits("limits")
exec_task = asyncio.create_task(session.exec("echo", "hi"))
await asyncio.sleep(0.05)
assert not exec_task.done()
backend.gate.set()
result = await asyncio.wait_for(exec_task, timeout=1)
assert result.ok()
assert backend.session.execs == [("echo", "hi")]
assert not session.pending
assert backend.session.limits == ["limits"]
assert backend.session.state.manifest is session.state.manifest
await session.write(Path("/workspace/x"), io.BytesIO(b"data"))
assert backend.session.writes == [(Path("/workspace/x"), b"data")]
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_failed_bring_up_surfaces_on_first_io(backend: _Backend) -> None:
scan_id = "deferred-fail"
try:
bundle = await _create(scan_id)
session = bundle["session"]
backend.gate.set()
with pytest.raises(RuntimeError, match="bring-up exploded"):
await asyncio.wait_for(session.exec("true"), timeout=1)
assert not await session.running()
with pytest.raises(RuntimeError, match="bring-up exploded"):
await bundle["caido_client"].get()
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == []
@pytest.mark.asyncio
async def test_cleanup_while_pending_cancels_the_bring_up(backend: _Backend) -> None:
scan_id = "deferred-cancel"
bundle = await _create(scan_id)
session = bundle["session"]
await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
assert session.bring_up.cancelled()
assert backend.client.deleted == []
assert session_manager._SESSION_CACHE.get(scan_id) is None
backend.gate.set()
await _settle()
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_cleanup_while_pending_ignores_a_failed_late_bring_up(backend: _Backend) -> None:
scan_id = "deferred-cancel-fail"
await _create(scan_id)
await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
backend.gate.set()
await _settle()
assert backend.client.deleted == []
@pytest.mark.asyncio
async def test_extra_files_are_staged_inside_the_deferred_bring_up(backend: _Backend) -> None:
scan_id = "deferred-extra"
try:
bundle = await _create(
scan_id,
extra_files=[{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}],
)
assert backend.session.writes == []
backend.gate.set()
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert len(backend.session.writes) == 1
assert backend.session.execs
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_awaitable_extra_files_do_not_delay_the_backend_call(backend: _Backend) -> None:
scan_id = "deferred-extra-awaitable"
files: asyncio.Future[list[dict[str, Any]] | None] = asyncio.get_running_loop().create_future()
try:
bundle = await _create(scan_id, extra_files=files)
backend.gate.set()
await _settle()
assert backend.calls == 1
assert backend.session.writes == []
files.set_result([{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}])
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert len(backend.session.writes) == 1
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_awaitable_extra_files_resolving_to_none_stage_nothing(backend: _Backend) -> None:
scan_id = "deferred-extra-none"
async def _nothing() -> list[dict[str, Any]] | None:
return None
try:
bundle = await _create(scan_id, extra_files=_nothing())
backend.gate.set()
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert backend.session.writes == []
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_default_path_still_blocks_until_the_sandbox_is_up(backend: _Backend) -> None:
scan_id = "deferred-off"
backend.gate.set()
try:
bundle = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
assert bundle["session"] is backend.session
assert bundle["client"] is backend.client
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_sdk_session_prep_does_not_wait_for_the_bring_up(backend: _Backend) -> None:
scan_id = "deferred-sdk-prep"
try:
bundle = await _create(scan_id)
agent = SandboxAgent(
name="root", instructions="x", tools=[], capabilities=[Filesystem(), Shell()]
)
manager = SandboxRuntimeSessionManager(
starting_agent=agent,
sandbox_config=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
run_state=None,
)
session = await asyncio.wait_for(
manager.ensure_session(
agent=agent, capabilities=list(agent.capabilities), is_resumed_state=False
),
timeout=1,
)
assert session is bundle["session"]
assert session.pending
backend.gate.set()
await asyncio.wait_for(session.resolve(), timeout=1)
assert not session.pending
finally:
await session_manager.cleanup(scan_id)
-1
View File
@@ -663,7 +663,6 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
== "src/render.ts:14 calls `_.template()`."
)
assert "**Usage analysis:**" in report["evidence"]
assert "not a proof of exploitability or of safety" in report["evidence"]
# The level must never influence the rating — that comes from the contextual
# breakdown, or from advisory_cvss when no breakdown applies.
assert report["severity"] == "high"
+4
View File
@@ -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())
+300
View File
@@ -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