Compaction session compacts a pending approval call, leaving an orphan output #5340

Activity

1. gomission commented on Oct 10, 2026

   

   gomission

   on Oct 10, 2026

   More actions

   Independent extension of the reproduction, with AI assistance: the rejection path also leaves an unpaired output on the next turn, even though the protected tool never executes. I checked both Runner.run and fully consumed Runner.run_streamed after a RunState.to_json() / RunState.from_json() round trip.

   Environment: released openai-agents==0.23.1, Python 3.12.11, openai==3.28.0, Pydantic 2.14.0. Public runner/session APIs, real SQLite, ScriptedModel, and a synthetic responses.compact response; no provider calls.

   Decision	Automatic compaction hook	Tool dispatches	Unpaired pending-c1 output in next-turn model input
   Approve	Always true	1	Present in both runner modes
   Reject	Always true	0	Present in both runner modes
   Approve	Always false	1	Absent in both runner modes
   Reject	Always false	0	Absent in both runner modes

   Eight cases executed: four next-turn pairing violations and four controls. In all eight, the immediate resumed model input still has its call/output pairing; the missing call is observable in the persisted history and the following user turn. Rejection remains enforced: this is not a rejected-tool execution.

   For the regression, I would include rejection as well as approval: a denied call still produces a tool output that needs its original call. Deferring only when a backend actually executes would miss that case. The JSON round trip keeps this tied to the public durable state boundary.

   Scope: one process, a stub compaction result and a scripted model. No real compaction API, OpenAI 400 response, separate-worker restore, SDK patch, or full-suite claim. The false compaction hook is a comparison control, not a complete fix.

   Runnable eight-case characterization

   Save as characterize_approval_compaction.py and run:

   uv run --no-project --python 3.12 --with openai-agents==0.23.1 python characterize_approval_compaction.py

   from __future__ import annotations
   
   import asyncio
   import importlib.metadata
   import json
   import platform
   import tempfile
   from pathlib import Path
   from types import SimpleNamespace
   from typing import Any
   
   from agents import Agent, Runner, RunState, SQLiteSession, function_tool
   from agents.memory import OpenAIResponsesCompactionSession
   from agents.testing import ScriptedModel, assistant_message, function_call
   from agents.tracing import set_tracing_disabled
   
   set_tracing_disabled(True)
   
   
   def normalized(items: list[Any]) -> list[dict[str, Any]]:
       return [
           item if isinstance(item, dict) else item.model_dump(exclude_unset=True)
           for item in items
       ]
   
   
   def unpaired_outputs(items: list[Any]) -> list[str]:
       values = normalized(items)
       calls = {item["call_id"] for item in values if item.get("type") == "function_call"}
       return [
           item["call_id"]
           for item in values
           if item.get("type") == "function_call_output" and item["call_id"] not in calls
       ]
   
   
   async def run_case(*, streaming: bool, allow: bool, compact: bool) -> dict[str, Any]:
       dispatches: list[str] = []
       compact_inputs: list[list[dict[str, Any]]] = []
   
       @function_tool(needs_approval=True)
       def protected(value: str) -> str:
           """Return a synthetic value after approval."""
           dispatches.append(value)
           return "synthetic:" + value
   
       async def fake_compact(**kwargs: Any) -> SimpleNamespace:
           compact_inputs.append(normalized(kwargs.get("input", [])))
           return SimpleNamespace(
               output=[
                   {
                       "type": "compaction",
                       "id": "synthetic-compaction",
                       "encrypted_content": "synthetic-summary",
                   }
               ],
               usage=None,
           )
   
       model = ScriptedModel(
           steps=[
               [function_call("protected", {"value": "v"}, call_id="pending-c1")],
               [assistant_message("finished")],
               [assistant_message("next answer")],
           ]
       )
       agent = Agent(name="synthetic-agent", model=model, tools=[protected])
   
       async def run(value: str | RunState, session: Any) -> Any:
           if not streaming:
               return await Runner.run(agent, value, session=session)
           result = Runner.run_streamed(agent, value, session=session)
           async for _ in result.stream_events():
               pass
           return result
   
       with tempfile.TemporaryDirectory(prefix="synthetic-compaction-") as directory:
           store = SQLiteSession("synthetic-session", str(Path(directory) / "history.db"))
           try:
               session = OpenAIResponsesCompactionSession(
                   "synthetic-session",
                   store,
                   client=SimpleNamespace(responses=SimpleNamespace(compact=fake_compact)),
                   compaction_mode="input",
                   should_trigger_compaction=lambda _: compact,
               )
               first = await run("first question", session)
               assert len(first.interruptions) == 1 and not dispatches
               stored_at_pause = normalized(await session.get_items())
               state = first.to_state()
               state_json = state.to_json()
               restored = await RunState.from_json(agent, state_json)
               interruption = first.interruptions[0]
               if allow:
                   restored.approve(interruption)
               else:
                   restored.reject(interruption)
               await run(restored, session)
               stored_after_resume = normalized(await session.get_items())
               resume_model_input = normalized(model.calls[-1].input)
               await run("next question", session)
               next_model_input = normalized(model.calls[-1].input)
               assert len(dispatches) == (1 if allow else 0)
               return {
                   "streaming": streaming,
                   "decision": "approve" if allow else "reject",
                   "compaction_enabled": compact,
                   "runstate_json_roundtrip": True,
                   "dispatches": len(dispatches),
                   "compact_invocations": len(compact_inputs),
                   "stored_at_pause": stored_at_pause,
                   "stored_after_resume": stored_after_resume,
                   "resume_model_input": resume_model_input,
                   "next_model_input": next_model_input,
                   "resume_unpaired_outputs": unpaired_outputs(resume_model_input),
                   "next_turn_unpaired_outputs": unpaired_outputs(next_model_input),
               }
           finally:
               store.close()
   
   
   async def main() -> None:
       cases = []
       for streaming in (False, True):
           for allow in (True, False):
               for compact in (True, False):
                   cases.append(
                       await run_case(streaming=streaming, allow=allow, compact=compact)
                   )
       violations = [case for case in cases if case["next_turn_unpaired_outputs"]]
       report = {
           "python": platform.python_version(),
           "openai_agents": importlib.metadata.version("openai-agents"),
           "openai": importlib.metadata.version("openai"),
           "pydantic": importlib.metadata.version("pydantic"),
           "cases_executed": len(cases),
           "cases_with_next_turn_unpaired_outputs": len(violations),
           "cases": cases,
           "scope": (
               "Public SDK Runner, real SQLite session and RunState JSON round trip; "
               "ScriptedModel and synthetic compaction response, one process. "
               "No model/compaction API calls, separate-process restore, SDK fix or full suite."
           ),
       }
       Path(__file__).with_name("verification.json").write_text(
           json.dumps(report, indent=2) + "\n"
       )
       print(
           json.dumps(
               {
                   **{key: value for key, value in report.items() if key != "cases"},
                   "case_summaries": [
                       {
                           key: case[key]
                           for key in (
                               "streaming",
                               "decision",
                               "compaction_enabled",
                               "dispatches",
                               "resume_unpaired_outputs",
                               "next_turn_unpaired_outputs",
                           )
                       }
                       for case in cases
                   ],
               },
               indent=2,
           )
       )
   
   
   if __name__ == "__main__":
       asyncio.run(main())