Compare commits

...
12 changed files with 623 additions and 10 deletions
+1
View File
@@ -236,6 +236,7 @@ ignore = [
"strix/interface/auth_cli.py" = ["N802"]
"tests/test_codex_streaming.py" = ["N802"]
"tests/test_disable_streaming.py" = ["N802"]
"tests/test_tool_call_ids.py" = ["N802"]
"tests/test_unknown_tool_recovery.py" = ["N802"]
"tests/test_report_pdf.py" = ["S105", "S106"]
# Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a
+109 -4
View File
@@ -6,7 +6,7 @@ import contextlib
import inspect
import os
import time
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, cast
from agents import (
set_default_openai_api,
@@ -24,12 +24,18 @@ from agents.retry import (
RetryPolicyContext,
retry_policies,
)
from openai.types.responses import Response, ResponseCompletedEvent
from openai.types.responses import (
Response,
ResponseCompletedEvent,
ResponseOutputItemAddedEvent,
ResponseOutputItemDoneEvent,
)
from openai.types.responses.response_usage import ResponseUsage
from openai.types.shared import Reasoning
from strix.config import codex
from strix.config.loader import load_settings
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
if TYPE_CHECKING:
@@ -229,6 +235,105 @@ class _NonStreamingModel(Model):
yield _completed_stream_event(response, getattr(self._inner, "model", None))
class _UniqueToolCallIdModel(Model):
"""Keep tool-call ids unique so a recycled id can't invalidate the history.
Providers that number tool calls per turn (``exec_command:0``, ...) restart
the counter each turn, so the same id eventually appears twice in one
conversation and strict providers reject every subsequent request. Ids that
collide with the history are rewritten before the turn is recorded, and
already-corrupted histories are repaired on the way out.
"""
def __init__(self, inner: Model) -> None:
self._inner = inner
async def close(self) -> None:
await self._inner.close()
def get_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None:
return self._inner.get_retry_advice(request)
async def get_response(
self,
system_instructions: str | None,
input: str | list[TResponseInputItem], # noqa: A002
model_settings: ModelSettings,
tools: list[Tool],
output_schema: AgentOutputSchemaBase | None,
handoffs: list[Handoff],
tracing: ModelTracing,
*,
previous_response_id: str | None,
conversation_id: str | None,
prompt: ResponsePromptParam | None,
) -> ModelResponse:
sanitized = dedupe_input(input)
rewriter = TurnCallIdRewriter(sanitized)
response = await self._inner.get_response(
system_instructions,
cast("str | list[TResponseInputItem]", sanitized),
model_settings,
tools,
output_schema,
handoffs,
tracing,
previous_response_id=previous_response_id,
conversation_id=conversation_id,
prompt=prompt,
)
response.output = rewriter.rewrite_items(list(response.output))
return response
async def stream_response(
self,
system_instructions: str | None,
input: str | list[TResponseInputItem], # noqa: A002
model_settings: ModelSettings,
tools: list[Tool],
output_schema: AgentOutputSchemaBase | None,
handoffs: list[Handoff],
tracing: ModelTracing,
*,
previous_response_id: str | None,
conversation_id: str | None,
prompt: ResponsePromptParam | None,
) -> AsyncIterator[TResponseStreamEvent]:
sanitized = dedupe_input(input)
rewriter = TurnCallIdRewriter(sanitized)
stream = self._inner.stream_response(
system_instructions,
cast("str | list[TResponseInputItem]", sanitized),
model_settings,
tools,
output_schema,
handoffs,
tracing,
previous_response_id=previous_response_id,
conversation_id=conversation_id,
prompt=prompt,
)
async for event in stream:
yield _rewrite_event_call_ids(event, rewriter)
def _rewrite_event_call_ids(
event: TResponseStreamEvent, rewriter: TurnCallIdRewriter
) -> TResponseStreamEvent:
if isinstance(event, ResponseOutputItemAddedEvent | ResponseOutputItemDoneEvent):
rewritten = rewriter.rewrite_item(event.item)
if rewritten is not event.item:
return event.model_copy(update={"item": rewritten})
return event
if isinstance(event, ResponseCompletedEvent):
output = rewriter.rewrite_items(list(event.response.output))
if output != list(event.response.output):
return event.model_copy(
update={"response": event.response.model_copy(update={"output": output})}
)
return event
def _completed_stream_event(
model_response: ModelResponse, model_name: object | None
) -> TResponseStreamEvent:
@@ -305,8 +410,8 @@ class StrixProvider(MultiProvider):
)
model = super().get_model(model_name)
if llm.disable_streaming:
return _NonStreamingModel(model)
return model
model = _NonStreamingModel(model)
return _UniqueToolCallIdModel(model)
DEFAULT_MODEL_RETRY = ModelRetrySettings(
+117
View File
@@ -0,0 +1,117 @@
"""Keep tool-call ids unique within a conversation.
Some providers return per-turn tool-call ids (``exec_command:0``,
``exec_command:1``, ...) whose counter restarts on every turn. Once the same
id appears twice in one conversation, the request payload has two assistant
tool calls sharing an id and strict providers reject the whole turn, which
permanently kills the agent because the malformed history is replayed on
every retry. Rewriting duplicates to fresh unique ids keeps the history
valid for any provider.
"""
from __future__ import annotations
from collections import defaultdict, deque
from typing import Any
from uuid import uuid4
from openai.types.responses import ResponseFunctionToolCall
def new_call_id() -> str:
return f"call_{uuid4().hex}"
def collect_call_ids(items: list[Any]) -> set[str]:
used: set[str] = set()
for item in items:
if isinstance(item, dict):
call_id = item.get("call_id")
if isinstance(call_id, str):
used.add(call_id)
elif isinstance(item, ResponseFunctionToolCall):
used.add(item.call_id)
return used
def dedupe_history_call_ids(items: list[Any]) -> tuple[list[Any], bool]:
"""Rewrite duplicate call ids in a conversation history.
Outputs are paired with their call by order, so parallel calls that share
an id keep answering the right call after the rewrite.
"""
used: set[str] = set()
pending: dict[str, deque[str]] = defaultdict(deque)
rebuilt: list[Any] = []
changed = False
for item in items:
if not isinstance(item, dict):
rebuilt.append(item)
continue
call_id = item.get("call_id")
if not isinstance(call_id, str):
rebuilt.append(item)
continue
kind = item.get("type")
if kind == "function_call":
effective = call_id
if call_id in used:
effective = new_call_id()
item = {**item, "call_id": effective} # noqa: PLW2901
changed = True
used.add(effective)
pending[call_id].append(effective)
elif kind == "function_call_output":
queue = pending.get(call_id)
if queue:
effective = queue.popleft()
if effective != call_id:
item = {**item, "call_id": effective} # noqa: PLW2901
changed = True
rebuilt.append(item)
return rebuilt, changed
def dedupe_input(model_input: str | list[Any]) -> str | list[Any]:
if isinstance(model_input, str):
return model_input
rebuilt, changed = dedupe_history_call_ids(model_input)
return rebuilt if changed else model_input
class TurnCallIdRewriter:
"""Rewrite a single turn's tool-call ids that collide with the history.
A turn's items surface several times (streamed item events, then the
completed response), so the same original id must always map to the same
replacement within the turn.
"""
def __init__(self, model_input: str | list[Any]) -> None:
self._used = set() if isinstance(model_input, str) else collect_call_ids(model_input)
self._remap: dict[str, str] = {}
self._settled: set[str] = set()
def rewrite_item(self, item: Any) -> Any:
if not isinstance(item, ResponseFunctionToolCall):
return item
original = item.call_id
if original in self._settled:
return item
replacement = self._remap.get(original)
if replacement is None:
if original not in self._used:
self._used.add(original)
self._settled.add(original)
return item
replacement = new_call_id()
self._remap[original] = replacement
self._used.add(replacement)
self._settled.add(replacement)
return item.model_copy(update={"call_id": replacement})
def rewrite_items(self, items: list[Any]) -> list[Any]:
return [self.rewrite_item(item) for item in items]
@@ -74,6 +74,8 @@ func vulnerabilityMarkdownReport(v map[string]any) string {
field("Ecosystem", render.StringValue(dep["package_ecosystem"]))
field("Installed Version", render.StringValue(dep["installed_version"]))
field("Fixed Version", render.StringValue(dep["fixed_version"]))
field("Introduced By", render.StringValue(dep["introduced_by"]))
field("Dependency Chain", render.StringValue(dep["dependency_path"]))
}
field("Endpoint", render.StringValue(v["endpoint"]))
field("Method", render.StringValue(v["method"]))
@@ -298,6 +298,8 @@ func vulnerabilityBody(v map[string]any) string {
field("Ecosystem", render.StringValue(dep["package_ecosystem"]))
field("Installed Version", render.StringValue(dep["installed_version"]))
field("Fixed Version", render.StringValue(dep["fixed_version"]))
field("Introduced By", render.StringValue(dep["introduced_by"]))
field("Dependency Chain", render.StringValue(dep["dependency_path"]))
}
field("Endpoint", render.StringValue(v["endpoint"]))
field("Method", render.StringValue(v["method"]))
+4
View File
@@ -531,6 +531,10 @@ def _result_properties(
if value not in (None, ""):
strix[key] = value
dependency_metadata = report.get("dependency_metadata")
if isinstance(dependency_metadata, dict) and dependency_metadata:
strix["dependency_metadata"] = dependency_metadata
# SARIF is written for external upload (code-scanning / ASPM), so it must
# NOT carry the weaponized exploit payload — that stays a local run
# artifact (vulnerabilities.json / the finding MD). We surface the PoC
+2
View File
@@ -205,6 +205,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
("Ecosystem", dep_meta.get("package_ecosystem")),
("Installed Version", dep_meta.get("installed_version")),
("Fixed Version", dep_meta.get("fixed_version")),
("Introduced By", dep_meta.get("introduced_by")),
("Dependency Chain", dep_meta.get("dependency_path")),
("Endpoint", report.get("endpoint")),
("Method", report.get("method")),
("CVE", report.get("cve")),
+34 -2
View File
@@ -39,9 +39,11 @@ trivy version --format json 2>/dev/null | tee "$ART/trivy-version.json"
# sandbox with egress gets the freshest CVEs; if the update fails, fall back to the
# cached DB instead of failing the scan. --offline-scan keeps per-package advisory
# lookups offline.
trivy fs --scanners vuln --timeout 30m --offline-scan \
# --list-all-pkgs includes the package graph (Relationship + DependsOn) needed
# to attribute transitive CVEs to the direct dependency that introduces them.
trivy fs --scanners vuln --timeout 30m --offline-scan --list-all-pkgs \
--format json --output "$ART/trivy-sca.json" . \
|| trivy fs --scanners vuln --timeout 30m --offline-scan --skip-db-update \
|| trivy fs --scanners vuln --timeout 30m --offline-scan --skip-db-update --list-all-pkgs \
--format json --output "$ART/trivy-sca.json" . \
|| true
```
@@ -78,6 +80,36 @@ For each entry under `.Results[].Vulnerabilities[]` in `trivy-sca.json`, collect
Deduplicate by `(CVE, PkgName, InstalledVersion)`. File one
`create_dependency_report` per CVE — do not batch multiple CVEs into one report.
### Attribute transitive CVEs to the direct dependency
With `--list-all-pkgs`, each `.Results[].Packages[]` entry carries `ID`
(`name@version`), `Relationship` (`direct` / `indirect`) and `DependsOn` (the
`ID`s it resolves to). For every vulnerable package that is **indirect**, walk
the `DependsOn` graph backwards to find the `direct` package(s) whose closure
contains it, then pass to `create_dependency_report`:
- `introduced_by` — the direct dependency as `name@version` (e.g.
`express@4.18.1`). If several direct dependencies pull it in, pick the
primary one and name the rest in `technical_analysis`.
- `dependency_path` — the shortest resolution chain from that direct
dependency to the vulnerable package, joined with ` > ` (e.g.
`express@4.18.1 > body-parser@1.20.0 > qs@6.10.2`).
- Omit both when the vulnerable package is itself a direct dependency.
If the ecosystem's lockfile gives trivy no graph (`DependsOn` absent), derive
the chain from the package manager instead (`npm ls <pkg>`, `pnpm why <pkg>`,
`yarn why <pkg>`, `pipdeptree --reverse -p <pkg>`, `go mod graph`,
`mvn dependency:tree`, ...) — and if that also fails, leave the fields out
rather than guessing.
For transitive findings, `remediation_steps` must be actionable at the
**direct-dependency level**: upgrading the vulnerable package directly is
usually impossible from the app's own manifest. Say which direct dependency to
bump (a version whose closure resolves the fixed version), or how to force the
resolution (npm `overrides` / yarn `resolutions` / pnpm `pnpm.overrides` /
Maven `dependencyManagement` / Gradle resolution strategy / `go mod edit`),
not just "upgrade <vulnerable pkg> to <fixed>".
### Reachability is a confidence modifier, not a gate
Do NOT suppress or downgrade a known CVE just because you could not prove the
+34
View File
@@ -725,6 +725,8 @@ def _build_dependency_metadata(
installed_version: str,
package_ecosystem: str | None,
fixed_version: str | None,
introduced_by: str | None,
dependency_path: str | None,
) -> dict[str, str]:
metadata = {
"package_name": package_name.strip(),
@@ -734,6 +736,10 @@ def _build_dependency_metadata(
metadata["package_ecosystem"] = package_ecosystem.strip()
if fixed_version and fixed_version.strip():
metadata["fixed_version"] = fixed_version.strip()
if introduced_by and introduced_by.strip():
metadata["introduced_by"] = introduced_by.strip()
if dependency_path and dependency_path.strip():
metadata["dependency_path"] = dependency_path.strip()
return metadata
@@ -743,6 +749,8 @@ def _build_dependency_evidence(
package_name: str,
installed_version: str,
fixed_version: str | None,
introduced_by: str | None,
dependency_path: str | None,
) -> str:
evidence = (
f"**Advisory evidence:** `{cve}` applies to `{package_name}` "
@@ -750,6 +758,13 @@ def _build_dependency_evidence(
)
if fixed_version and fixed_version.strip():
evidence += f" The advisory is fixed in `{fixed_version.strip()}`."
if introduced_by and introduced_by.strip():
evidence += (
f"\n\n**Transitive dependency:** introduced by the direct "
f"dependency `{introduced_by.strip()}`."
)
if dependency_path and dependency_path.strip():
evidence += f"\n\n**Dependency chain:** `{dependency_path.strip()}`"
return evidence
@@ -770,6 +785,8 @@ async def _do_create_dependency( # noqa: PLR0912
advisory_cvss: float | None,
technical_analysis: str | None,
fix_effort: str,
introduced_by: str | None = None,
dependency_path: str | None = None,
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
@@ -824,12 +841,16 @@ async def _do_create_dependency( # noqa: PLR0912
installed_version=installed_version,
package_ecosystem=package_ecosystem,
fixed_version=fixed_version,
introduced_by=introduced_by,
dependency_path=dependency_path,
)
evidence = _build_dependency_evidence(
cve=parsed_cve,
package_name=package_name.strip(),
installed_version=installed_version.strip(),
fixed_version=fixed_version,
introduced_by=introduced_by,
dependency_path=dependency_path,
)
try:
@@ -926,6 +947,8 @@ async def create_dependency_report(
cwe: str | None = None,
technical_analysis: str | None = None,
fix_effort: str = "low",
introduced_by: str | None = None,
dependency_path: str | None = None,
) -> str:
"""File a known-CVE dependency (SCA) finding — one report per CVE x package.
@@ -978,6 +1001,15 @@ async def create_dependency_report(
technical_analysis: Optional deeper mechanism/root-cause detail.
fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``
(dependency upgrades are usually ``trivial``/``low``).
introduced_by: For a **transitive** dependency, the direct
dependency (from the project's own manifest) that pulls the
vulnerable package in, as ``name@version`` (e.g.
``express@4.18.1``). Omit when the vulnerable package is
itself a direct dependency.
dependency_path: The resolution chain from the direct dependency
to the vulnerable package, joined with `` > `` (e.g.
``express@4.18.1 > body-parser@1.20.0 > qs@6.10.2``). Omit
for direct dependencies.
"""
agent_id, agent_name = _caller_identity(ctx)
@@ -997,6 +1029,8 @@ async def create_dependency_report(
advisory_cvss=advisory_cvss,
technical_analysis=technical_analysis,
fix_effort=fix_effort,
introduced_by=introduced_by,
dependency_path=dependency_path,
agent_id=agent_id,
agent_name=agent_name,
)
+6 -4
View File
@@ -31,7 +31,7 @@ from openai.types.responses import (
from strix.config import codex, loader
from strix.config.loader import load_settings
from strix.config.models import StrixProvider, _NonStreamingModel
from strix.config.models import StrixProvider, _NonStreamingModel, _UniqueToolCallIdModel
if TYPE_CHECKING:
@@ -299,10 +299,11 @@ def test_get_model_wraps_when_disabled(
load_settings()
model = StrixProvider().get_model("openai/gpt-4o-mini")
assert isinstance(model, _NonStreamingModel)
assert isinstance(model, _UniqueToolCallIdModel)
assert isinstance(model._inner, _NonStreamingModel)
def test_get_model_unwrapped_by_default(
def test_get_model_keeps_streaming_by_default(
monkeypatch: pytest.MonkeyPatch, _reset_settings: None
) -> None:
inner = _DummyModel()
@@ -310,7 +311,8 @@ def test_get_model_unwrapped_by_default(
load_settings()
model = StrixProvider().get_model("openai/gpt-4o-mini")
assert model is inner
assert isinstance(model, _UniqueToolCallIdModel)
assert model._inner is inner
def test_get_model_does_not_wrap_subscription_model(
+63
View File
@@ -164,6 +164,69 @@ async def test_dependency_report_sets_class_and_metadata(report_state: ReportSta
}
async def test_dependency_report_records_transitive_chain(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2022-24999 in qs 6.10.2",
description="Prototype pollution in qs parsing.",
target="repo/package.json",
cve="CVE-2022-24999",
package_name="qs",
installed_version="6.10.2",
impact="Denial of service via crafted query strings.",
remediation_steps="Upgrade express to 4.18.2, which resolves qs 6.11.0.",
assumptions="qs parses all incoming query strings by default.",
package_ecosystem="npm",
fixed_version="6.10.3",
cwe="CWE-1321",
advisory_cvss=7.5,
technical_analysis=None,
fix_effort="trivial",
introduced_by="express@4.18.1",
dependency_path="express@4.18.1 > body-parser@1.20.0 > qs@6.10.2",
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["dependency_metadata"]["introduced_by"] == "express@4.18.1"
assert (
report["dependency_metadata"]["dependency_path"]
== "express@4.18.1 > body-parser@1.20.0 > qs@6.10.2"
)
assert (
"**Transitive dependency:** introduced by the direct dependency `express@4.18.1`."
in report["evidence"]
)
assert (
"**Dependency chain:** `express@4.18.1 > body-parser@1.20.0 > qs@6.10.2`"
in report["evidence"]
)
async def test_dependency_report_omits_blank_chain_fields(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Impact.",
remediation_steps="Upgrade.",
assumptions="Assumptions.",
package_ecosystem="npm",
fixed_version=None,
cwe=None,
advisory_cvss=5.0,
technical_analysis=None,
fix_effort="trivial",
introduced_by=" ",
dependency_path=None,
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert "introduced_by" not in report["dependency_metadata"]
assert "dependency_path" not in report["dependency_metadata"]
async def test_dependency_report_with_zero_cvss_remains_low_severity(
report_state: ReportState,
) -> None:
+249
View File
@@ -0,0 +1,249 @@
"""Tests for tool-call id uniqueness.
Providers that number tool calls per turn (``exec_command:0``, ``:1``, ...)
restart the counter on every turn, so the same id eventually appears twice in
one conversation. Strict providers then reject the whole request, and because
the history is replayed on every retry the agent can never recover. A gateway
that validates id uniqueness the way those providers do proves both the
failure and the fix.
"""
from __future__ import annotations
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
from openai.types.responses import ResponseFunctionToolCall
from strix.config.models import _NonStreamingModel, _UniqueToolCallIdModel
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_history_call_ids
if TYPE_CHECKING:
from collections.abc import Iterator
def _tool_call_completion(call_id: str, n: int = 1) -> 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_id,
"type": "function",
"function": {"name": "do_thing", "arguments": json.dumps({"n": n})},
}
],
},
}
],
"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_call_ids(messages: list[dict[str, Any]]) -> list[str]:
return [str(call.get("id")) 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 a history reusing a tool-call id, like strict providers 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)
call_ids = _assistant_call_ids(messages)
if len(call_ids) != len(set(call_ids)):
self._respond(
400,
{
"error": {
"message": (
"tool messages need a resolvable tool name: carry `tool`/`name`, "
"or match a preceding assistant tool_call by order"
)
}
},
)
return
turn = len(_REQUESTS)
if turn <= 2:
# The provider restarts its per-turn counter, so both turns say ":0".
self._respond(200, _tool_call_completion("exec_command:0", n=turn))
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:
# The gateway answers plain JSON, so the run loop's streamed turns are
# served non-streamed; the ids on the wire are the same either way.
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 _UniqueToolCallIdModel(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_recycled_call_id_erases_a_turn_without_the_wrapper(strict_gateway: str) -> None:
# Repro: two turns run a tool and both are labelled ``exec_command:0``, so
# the colliding call and its result are dropped as duplicates. The agent
# ends the run having silently lost a turn of its own work — and a provider
# that does not drop them instead rejects the malformed history outright.
result = await _run_agent(strict_gateway, wrap=False)
assert result.final_output == "all done"
assert _assistant_call_ids(_REQUESTS[-1]) == ["exec_command:0"]
assert _tool_results(_REQUESTS[-1]) == ["did 2"]
@pytest.mark.asyncio
async def test_recycled_call_id_is_rewritten_so_no_turn_is_lost(strict_gateway: str) -> None:
result = await _run_agent(strict_gateway, wrap=True)
assert result.final_output == "all done"
call_ids = _assistant_call_ids(_REQUESTS[-1])
assert len(call_ids) == len(set(call_ids)) == 2
assert call_ids[0] == "exec_command:0"
assert call_ids[1].startswith("call_")
assert _tool_results(_REQUESTS[-1]) == ["did 1", "did 2"]
def test_history_dedupe_keeps_outputs_paired_with_their_call() -> None:
items = [
{"type": "function_call", "call_id": "exec_command:0", "name": "a", "arguments": "{}"},
{"type": "function_call_output", "call_id": "exec_command:0", "output": "first"},
{"type": "function_call", "call_id": "exec_command:0", "name": "b", "arguments": "{}"},
{"type": "function_call_output", "call_id": "exec_command:0", "output": "second"},
]
rebuilt, changed = dedupe_history_call_ids(items)
assert changed
ids = [item["call_id"] for item in rebuilt]
assert ids[0] == ids[1] == "exec_command:0"
assert ids[2] == ids[3] != "exec_command:0"
assert rebuilt[3]["output"] == "second"
def test_history_dedupe_pairs_parallel_calls_by_order() -> None:
items = [
{"type": "function_call", "call_id": "dup", "name": "a", "arguments": "{}"},
{"type": "function_call", "call_id": "dup", "name": "b", "arguments": "{}"},
{"type": "function_call_output", "call_id": "dup", "output": "for-a"},
{"type": "function_call_output", "call_id": "dup", "output": "for-b"},
]
rebuilt, changed = dedupe_history_call_ids(items)
assert changed
assert rebuilt[0]["call_id"] == rebuilt[2]["call_id"] == "dup"
assert rebuilt[1]["call_id"] == rebuilt[3]["call_id"]
assert rebuilt[1]["call_id"] != "dup"
def test_history_dedupe_leaves_unique_ids_alone() -> None:
items = [
{"type": "function_call", "call_id": "call_a", "name": "a", "arguments": "{}"},
{"type": "function_call_output", "call_id": "call_a", "output": "x"},
{"type": "function_call", "call_id": "call_b", "name": "b", "arguments": "{}"},
]
rebuilt, changed = dedupe_history_call_ids(items)
assert not changed
assert rebuilt == items
def test_turn_rewriter_is_stable_across_repeated_sightings() -> None:
history = [{"type": "function_call", "call_id": "exec_command:0", "name": "a"}]
rewriter = TurnCallIdRewriter(history)
call = ResponseFunctionToolCall(
call_id="exec_command:0", name="a", arguments="{}", type="function_call"
)
first = rewriter.rewrite_item(call)
second = rewriter.rewrite_item(first)
assert first.call_id != "exec_command:0"
assert second.call_id == first.call_id