Compaction unusable in Runner.run_streamed: responses.compact runs before tool outputs are in chain

[Filimoa]

Please read this first

• Have you read the docs?Agents SDK docs
  Yes
• Have you searched for related issues? Others may have faced similar issues.
  Yes

Describe the bug

When using OpenAIResponsesCompactionSession in a streamed run (Runner.run_streamed), compaction can be triggered at a point where the previous_response_id passed into client.responses.compact() corresponds to a model response that includes a function_call, but the corresponding function_call_output has not yet been incorporated into the server-side response chain. The OpenAI API then returns:

400 invalid_request_error: No tool output found for function call <call_id>.

This appears to happen when compaction is evaluated/run “between” a tool call and the subsequent model request that submits the tool output.

Expected behavior

Compaction should either:

• only run on response IDs that are safe to compact (i.e., tool calls have corresponding tool outputs in the chain), or
• gracefully skip/defer compaction when the response chain isn’t in a compactable state (instead of failing the whole run).

Actual behavior
A streamed run crashes with openai.BadRequestError from responses.compact:

• No tool output found for function call call_...

Why this seems like an SDK/library issue
The local session history can already contain the tool output item (because the SDK executed the tool and stored the output), but responses.compact(previous_response_id=...) is called using a response id that—on the server—still has a pending tool call with no matching output yet.

Reproduction

1. Run the attached script repro_compaction_missing_tool_output.py (below).
2. It:
   • forces a tool call (tool_choice="required" forces at least one tool call)
   • returns a large tool output so a token-based compaction trigger fires
   • uses Runner.run_streamed + OpenAIResponsesCompactionSession
3. Observe compaction trigger log and then 400 No tool output found….

2) Minimal repro script

import asyncio
import json
import os
from collections.abc import Callable
from typing import Any

import tiktoken
from loguru import logger

from agents import (
    Agent,
    OpenAIResponsesCompactionSession,
    Runner,
    SQLiteSession,
    function_tool,
    model_settings,
)

# --- Token estimation (rough) ---
_O200K = tiktoken.get_encoding("o200k_base")


def _stable_json(obj: Any) -> str:
    try:
        return json.dumps(obj, separators=(",", ":"), sort_keys=True, ensure_ascii=False)
    except TypeError:
        return str(obj)


def _count(text: str) -> int:
    return len(_O200K.encode(text))


def estimate_item_tokens(item: Any) -> int:
    """
    Rough token estimator for Responses API items.
    Good enough to trigger compaction in a reproducible way.
    """
    if isinstance(item, dict):
        content = item.get("content")
        if isinstance(content, str):
            return _count(content)
        if isinstance(content, list):
            total = 0
            for part in content:
                if isinstance(part, dict) and part.get("type") == "input_text":
                    total += _count(part.get("text", ""))
                else:
                    total += _count(_stable_json(part))
            return total
        return _count(_stable_json(item))
    return _count(_stable_json(item))


def estimate_items_tokens(items: list[Any]) -> int:
    return sum(estimate_item_tokens(it) for it in items)


def make_token_aware_should_trigger_compaction(*, token_limit: int) -> Callable[[dict[str, Any]], bool]:
    def _hook(ctx: dict[str, Any]) -> bool:
        session_items = ctx.get("session_items") or []
        est = estimate_items_tokens(session_items)
        trigger = est >= token_limit

        if trigger:
            # Useful debugging: show tail item types, response_id, etc.
            tail_types = []
            for it in session_items[-6:]:
                if isinstance(it, dict):
                    tail_types.append(it.get("type") or ("message" if "role" in it else "unknown"))
                else:
                    tail_types.append(type(it).__name__)

            logger.info(
                "Triggering compaction: est_tokens={} token_limit={} items={} candidates={} response_id={} tail_types={}",
                est,
                token_limit,
                len(session_items),
                len(ctx.get("compaction_candidate_items") or []),
                ctx.get("response_id"),
                tail_types,
            )

        return trigger

    return _hook


# --- Tool that returns a large payload so we exceed token_limit quickly ---
@function_tool
def big_payload() -> str:
    # ~a few thousand tokens, cheap but enough to trigger compaction
    return ("lorem ipsum " * 4000).strip()


async def main() -> None:
    # NOTE: ensure OPENAI_API_KEY is set
    if not os.getenv("OPENAI_API_KEY"):
        raise RuntimeError("Set OPENAI_API_KEY first")

    # Underlying session store
    underlying = SQLiteSession(":memory:")

    # Compaction session with a low token limit for easy repro
    session = OpenAIResponsesCompactionSession(
        session_id="repro-session",
        underlying_session=underlying,
        model="gpt-4.1",  
        should_trigger_compaction=make_token_aware_should_trigger_compaction(token_limit=1500),
    )

    agent = Agent(
        name="Repro Agent",
        instructions=(
            "You must call the big_payload tool before responding. "
            "After you receive the tool output, summarize it in one sentence."
        ),
        model="gpt-4.1",
        tools=[big_payload],
        model_settings=model_settings.ModelSettings(
            # Force at least one tool call per response.
            tool_choice="required",
            # Store is needed for response_id chaining; for Responses API it is typically enabled
            # automatically when not specified, but leaving explicit store=True is OK.
            store=True,  
        ),
    )

    logger.info("Starting streamed run...")
    run = Runner.run_streamed(
        starting_agent=agent,
        input="Call the tool now.",
        session=session,
        max_turns=5,
    )

    # Just drain stream; the crash (if it happens) will surface here.
    async for _ in run.stream_events():
        pass

    logger.info("Done.")


if __name__ == "__main__":
    asyncio.run(main())

Debug information

• Agents SDK version: 0.6.6
• Python version: 3.13

Expected behavior

This should not crash