I found a deterministic Responses API server-side compaction failure involving parallel native shell calls.
With store=false and:
{
"context_management": [
{ "type": "compaction", "compact_threshold": 7000 }
]
}
the following stateless input ordering compacts correctly:
shell_call(0), shell_call_output(0), shell_call(1), shell_call_output(1)
The normal ordering produced by parallel shell calls does not compact:
shell_call(0), shell_call(1), shell_call_output(0), shell_call_output(1)
The calls are server-issued and both requests use the exact same calls, outputs, model, tools, threshold, and generation options; only item order differs. The failing request is above threshold, completes successfully with an assistant message, and returns no warning or compaction item.
The failure is sticky while that parallel batch remains in cumulative input, so a long tool loop can run into the context limit even though automatic compaction is enabled. Sequential native shell calls and parallel ordinary function calls both compact correctly.
I reproduced this using raw HTTP, independent of any client SDK. A related report is open at Bug: server-side compaction is not emitted on Responses tool-call-only turns · Issue #3075 · openai/openai-python · GitHub, but that report focuses on tool-call-only responses. Here compaction remains absent even when the response contains assistant text.
#!/usr/bin/env python3
"""Reproduce missing OpenAI Responses compaction after parallel native shell calls."""
from __future__ import annotations
import json
import os
import sys
import urllib.error
import urllib.request
from typing import TypeAlias, cast
Json: TypeAlias = None | bool | int | float | str | list["Json"] | dict[str, "Json"]
JsonObject: TypeAlias = dict[str, Json]
MODEL = os.environ.get("OPENAI_MODEL", "gpt-5.6-sol")
BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1").rstrip("/")
API_KEY = os.environ["OPENAI_API_KEY"]
THRESHOLD = 7_000
TOOLS: list[JsonObject] = [{"type": "shell", "environment": {"type": "local"}}]
def require_object(value: Json, description: str) -> JsonObject:
if not isinstance(value, dict):
raise RuntimeError(f"Expected {description} to be an object, got {type(value).__name__}")
return value
def require_list(value: Json, description: str) -> list[Json]:
if not isinstance(value, list):
raise RuntimeError(f"Expected {description} to be an array, got {type(value).__name__}")
return value
def post_responses(input_items: list[Json], instructions: str) -> tuple[JsonObject, str | None]:
body: JsonObject = {
"model": MODEL,
"input": input_items,
"instructions": instructions,
"tools": TOOLS,
"parallel_tool_calls": True,
"context_management": [{"type": "compaction", "compact_threshold": THRESHOLD}],
"reasoning": {"effort": "none"},
"max_output_tokens": 1_024,
"store": False,
"stream": False,
}
request = urllib.request.Request(
f"{BASE_URL}/responses",
data=json.dumps(body).encode(),
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=600) as response:
result = require_object(cast(Json, json.load(response)), "response")
request_id = response.headers.get("x-request-id") or response.headers.get("llm_provider-x-request-id")
return result, request_id
except urllib.error.HTTPError as error:
response_body = error.read().decode("utf-8", "replace")
raise RuntimeError(f"Responses API returned HTTP {error.code}: {response_body}") from error
def output_items(response: JsonObject) -> list[JsonObject]:
return [require_object(item, "output item") for item in require_list(response.get("output"), "response.output")]
def item_types(response: JsonObject) -> list[str]:
return [str(item.get("type")) for item in output_items(response)]
def input_tokens(response: JsonObject) -> int | None:
usage = response.get("usage")
if not isinstance(usage, dict):
return None
value = usage.get("input_tokens")
return value if isinstance(value, int) else None
def shell_call_input(call: JsonObject) -> JsonObject:
return {
"type": "shell_call",
"call_id": call.get("call_id"),
"id": call.get("id"),
"status": call.get("status"),
"action": call.get("action"),
}
def shell_output(call: JsonObject) -> JsonObject:
return {
"type": "shell_call_output",
"call_id": call.get("call_id"),
"output": [
{
"stdout": "y " * 2_000,
"stderr": "",
"outcome": {"type": "exit", "exit_code": 0},
}
],
}
def summarize(response: JsonObject, request_id: str | None) -> JsonObject:
types = item_types(response)
return {
"request_id": request_id,
"input_tokens": input_tokens(response),
"output_types": types,
"compacted": "compaction" in types,
}
def main() -> None:
bootstrap, bootstrap_id = post_responses(
[{"role": "user", "content": "Create exactly two separate shell calls in parallel: `echo one` and `echo two`."}],
"Return exactly two native shell calls and no assistant message.",
)
calls = [item for item in output_items(bootstrap) if item.get("type") == "shell_call"]
if len(calls) != 2:
raise RuntimeError(f"Expected two shell calls, got: {json.dumps(item_types(bootstrap))}")
large_user: JsonObject = {"role": "user", "content": f"{'x ' * 4_000}\nReply exactly DONE after the supplied shell results."}
call_0, call_1 = (shell_call_input(call) for call in calls)
output_0, output_1 = (shell_output(call) for call in calls)
parallel, parallel_id = post_responses(
[large_user, call_0, call_1, output_0, output_1],
"Reply exactly DONE without calling tools.",
)
sequential, sequential_id = post_responses(
[large_user, call_0, output_0, call_1, output_1],
"Reply exactly DONE without calling tools.",
)
report: JsonObject = {
"model": MODEL,
"base_url": BASE_URL,
"compact_threshold": THRESHOLD,
"bootstrap": summarize(bootstrap, bootstrap_id),
"parallel_order_call_call_output_output": summarize(parallel, parallel_id),
"sequential_order_call_output_call_output": summarize(sequential, sequential_id),
}
print(json.dumps(report, indent=2))
parallel_compacted = report["parallel_order_call_call_output_output"]
sequential_compacted = report["sequential_order_call_output_call_output"]
if not isinstance(parallel_compacted, dict) or not isinstance(sequential_compacted, dict):
raise RuntimeError("Invalid report")
reproduced = parallel_compacted["compacted"] is False and sequential_compacted["compacted"] is True
if not reproduced:
print("Expected failure/control split was not reproduced.", file=sys.stderr)
raise SystemExit(1)
if __name__ == "__main__":
main()
Bug originally noticed while using LiteLLM; the above reproduction script obtains the same result when requests are sent straight to the OpenAI API.
Has anyone found a supported workaround other than disabling parallel native tools, rewriting history into adjacent call/output pairs, or explicitly calling /responses/compact?