mirror of
https://github.com/usestrix/strix.git
synced 2026-11-09 05:19:39 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
98707a785a | ||
|
|
b58d3ac3c6 |
No files matched your search
@@ -18,7 +18,6 @@ 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,
|
||||
@@ -268,10 +267,6 @@ 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
|
||||
|
||||
+2
-13
@@ -41,7 +41,6 @@ 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
|
||||
@@ -256,12 +255,6 @@ 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.
|
||||
@@ -320,7 +313,7 @@ class _TurnGuardModel(Model):
|
||||
conversation_id: str | None,
|
||||
prompt: ResponsePromptParam | None,
|
||||
) -> ModelResponse:
|
||||
sanitized = _sanitize_input(input)
|
||||
sanitized = dedupe_input(input)
|
||||
rewriter = TurnCallIdRewriter(sanitized)
|
||||
system_instructions, sanitized = _split_cached_prefix(
|
||||
system_instructions, sanitized, model_settings
|
||||
@@ -356,7 +349,7 @@ class _TurnGuardModel(Model):
|
||||
conversation_id: str | None,
|
||||
prompt: ResponsePromptParam | None,
|
||||
) -> AsyncIterator[TResponseStreamEvent]:
|
||||
sanitized = _sanitize_input(input)
|
||||
sanitized = dedupe_input(input)
|
||||
rewriter = TurnCallIdRewriter(sanitized)
|
||||
system_instructions, sanitized = _split_cached_prefix(
|
||||
system_instructions, sanitized, model_settings
|
||||
@@ -386,10 +379,6 @@ 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],
|
||||
|
||||
@@ -1,100 +0,0 @@
|
||||
"""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
|
||||
@@ -8,7 +8,7 @@ import io
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
from collections.abc import Awaitable, Callable
|
||||
from collections.abc import 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]] | Awaitable[list[dict[str, Any]] | None] | None = None,
|
||||
extra_files: list[dict[str, Any]] | None = None,
|
||||
coordinator: AgentCoordinator | None = None,
|
||||
interactive: bool = False,
|
||||
max_turns: int = DEFAULT_MAX_TURNS,
|
||||
|
||||
@@ -1,255 +0,0 @@
|
||||
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)
|
||||
@@ -3,8 +3,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import inspect
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
@@ -21,12 +19,9 @@ 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
|
||||
@@ -260,22 +255,13 @@ 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]] | Awaitable[list[dict[str, Any]] | None] | None = None,
|
||||
extra_files: list[dict[str, Any]] | 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.
|
||||
|
||||
@@ -285,9 +271,7 @@ 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. ``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.
|
||||
and unpacked inside the sandbox right after bring-up.
|
||||
"""
|
||||
|
||||
def report(phase: str) -> None:
|
||||
@@ -308,6 +292,9 @@ 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.)
|
||||
@@ -336,69 +323,41 @@ 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,
|
||||
)
|
||||
|
||||
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,
|
||||
)
|
||||
)
|
||||
if extra_file_archive is not None:
|
||||
report("Placing workspace files")
|
||||
try:
|
||||
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)
|
||||
await stage_extra_files(session, extra_file_archive)
|
||||
except BaseException:
|
||||
await _discard_session(client, session)
|
||||
raise
|
||||
return client, session
|
||||
|
||||
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
|
||||
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)
|
||||
|
||||
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)
|
||||
# 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(_caido(session, host_caido_url), name=f"caido-bootstrap-{scan_id}")
|
||||
asyncio.create_task(
|
||||
bootstrap_caido(
|
||||
session,
|
||||
host_url=host_caido_url,
|
||||
container_url=container_caido_url,
|
||||
),
|
||||
name=f"caido-bootstrap-{scan_id}",
|
||||
)
|
||||
)
|
||||
|
||||
bundle = {
|
||||
@@ -411,26 +370,6 @@ 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:
|
||||
|
||||
@@ -1876,6 +1876,10 @@ 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
|
||||
|
||||
|
||||
|
||||
@@ -38,26 +38,11 @@ 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
|
||||
@@ -151,15 +136,12 @@ async def test_unknown_and_null_arguments_are_untouched() -> None:
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("arguments", ["not json", '{"n": NaN}', OVER_NESTED])
|
||||
async def test_non_object_payloads_are_reported_instead_of_invoked(arguments: str) -> None:
|
||||
async def test_non_object_payloads_pass_through_unchanged() -> None:
|
||||
captured: dict[str, str] = {}
|
||||
wrapped = factory._with_coerced_arguments(_capturing_tool(captured, _ARRAY))
|
||||
|
||||
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
|
||||
assert await wrapped.on_invoke_tool(cast("Any", None), "not json") == "ok"
|
||||
assert captured["raw_input"] == "not json"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -136,7 +136,6 @@ 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)
|
||||
|
||||
@@ -1,316 +0,0 @@
|
||||
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)
|
||||
@@ -663,6 +663,7 @@ 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"
|
||||
|
||||
@@ -263,11 +263,7 @@ 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())
|
||||
|
||||
@@ -1,300 +0,0 @@
|
||||
"""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
|
||||
Reference in new issue
Block a user