fix(sessions): count and classify the settling batch on the compaction path

Two defects in how a settling held batch meets a compaction-aware session:

- The compaction-deferral branch returned the run-item count alone, and it is
  the branch every held settle with outputs takes on such a backend, so exactly
  the sessions that defer were the ones whose settled turns undercounted. The
  count gates the resumed-safety refusal and slices later saves of the same
  turn, so it must equal what the append wrote. The branch now returns the
  combined count.

- The local-continuation classification knew the mapped tool outputs but not
  the hosted MCP approval response, which is the locally produced half of its
  approval pair and must stay associated with the response chain that carried
  the request. Compacting that response before the model consumes the approval
  drops it in previous_response_id mode. The constant is now
  _LOCAL_CONTINUATION_OUTPUT_TYPES and covers both carriers: the settled dict
  and the MCPApprovalResponseItem the non-deferred resume commits, because
  classifying one and not the other would defer or compact the same response
  depending on which path persisted it.

The four-stage scenario behind the count (partial approval, held settlement
with a lapsed gate, gate re-enable, remaining approval) is pinned end to end:
it must end in the documented fail-fast refusal with nothing duplicated. Each
fix is also pinned at the unit boundary and proven red by mutation.

Author: dixso

src/agents/run_internal/session_persistence.py

@@ -21,6 +21,7 @@
     HandoffOutputItem,
     InputItem,
     ItemHelpers,
+    MCPApprovalResponseItem,
     ModelResponse,
     RunItem,
     ToolCallOutputItem,
@@ -95,9 +96,14 @@
 
 _SESSION_LIMIT_UNSET = object()
 
-# Serialized item types that represent a locally produced tool output, i.e. the output
-# kinds of the canonical call-to-output map.
-_LOCAL_TOOL_OUTPUT_TYPES = frozenset(_TOOL_CALL_TO_OUTPUT_TYPE.values())
+# Serialized item types produced locally as the continuation of a model response: the
+# output kinds of the canonical call-to-output map, plus the hosted MCP approval
+# response, which is the locally produced half of its approval pair. Compaction for
+# the response that carried the request must be deferred while any of these still
+# needs to be associated with that response chain.
+_LOCAL_CONTINUATION_OUTPUT_TYPES = frozenset(_TOOL_CALL_TO_OUTPUT_TYPE.values()) | {
+    "mcp_approval_response"
+}
 
 
 async def admit_pending_input(
@@ -776,11 +782,12 @@ async def save_result_to_session(
         # landed. Only a settle reads that slot: on an ordinary save it holds the
         # caller's input, whose earlier outputs say nothing about this response.
         has_local_tool_outputs = any(
-            isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items
+            isinstance(item, ToolCallOutputItem | HandoffOutputItem | MCPApprovalResponseItem)
+            for item in new_items
         ) or (
             settling_held_batch
             and any(
-                isinstance(item, dict) and item.get("type") in _LOCAL_TOOL_OUTPUT_TYPES
+                isinstance(item, dict) and item.get("type") in _LOCAL_CONTINUATION_OUTPUT_TYPES
                 for item in items_to_save
             )
         )
@@ -797,7 +804,7 @@ async def save_result_to_session(
                 "skip: deferring compaction for response %s due to local tool outputs",
                 response_id,
             )
-            return saved_run_items_count
+            return saved_run_items_count + settled_batch_items
 
         deferred_response_id = None
         get_deferred = getattr(session, "_get_deferred_compaction_response_id", None)

tests/test_deferred_interrupted_session_write.py

@@ -1525,6 +1525,134 @@ async def test_the_compaction_deferral_reads_the_settling_batch_not_the_callers_
     assert [entry for entry in session.compactions if "deferred" in entry] == []
 
 
+@pytest.mark.asyncio
+async def test_the_settled_count_survives_the_compaction_deferral_branch() -> None:
+    # The deferral branch is the one every held settle with outputs takes on a
+    # compaction-aware backend, so returning the run-item count alone there reports a
+    # turn that persisted less than it wrote. That count gates the final sweep's
+    # re-append protection on a later gate-enabled resume.
+    from agents.run_internal.session_persistence import save_result_to_session
+
+    session = _CompactionRecordingSession()
+    held: list[TResponseInputItem] = [
+        {"type": "function_call", "call_id": "call_PARKED", "name": "t", "arguments": "{}"},
+        {"type": "function_call_output", "call_id": "call_PARKED", "output": "ok"},
+    ]
+
+    count = await save_result_to_session(
+        session, held, [], None, response_id="resp_parked", settling_held_batch=True
+    )
+
+    assert [entry for entry in session.compactions if "deferred" in entry] == [
+        {"deferred": "resp_parked", "store": None}
+    ]
+    assert count == len(await session.get_items())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streamed", [False, True])
+async def test_a_partial_settle_on_a_compaction_session_still_fails_the_gated_resume_fast(
+    streamed: bool,
+) -> None:
+    # Park two calls, approve one, and let the gate lapse for that resume so the batch
+    # settles into a compaction-aware session mid-run. Re-enable the gate and approve
+    # the rest: the settled turn's count must cover what the settle wrote, or the
+    # final sweep treats the turn as unpersisted and appends the stored items again.
+    from agents.exceptions import UserError
+
+    session = _CompactionRecordingSession()
+    agent = _make_multi_approval_agent()
+
+    first = await _run(agent, "go", session, streamed=streamed)
+    state = await _serialized_round_trip(first, agent)
+    state.approve(
+        next(
+            interruption
+            for interruption in state.get_interruptions()
+            if getattr(interruption.raw_item, "call_id", None) == "call_PARKED"
+        )
+    )
+    gate = agent.output_guardrails
+    agent.output_guardrails = []
+    second = await _run(agent, state, session, streamed=streamed)
+    assert len(second.interruptions) == 1
+    agent.output_guardrails = gate
+
+    state = await _serialized_round_trip(second, agent)
+    for interruption in state.get_interruptions():
+        state.approve(interruption)
+    # The settled turn persisted items, so the re-enabled gate must refuse the resume
+    # outright; an undercounted turn is what would let it proceed and re-append the
+    # stored items through the final sweep.
+    with pytest.raises(UserError, match="output guardrails after current-turn items"):
+        await _run(agent, state, session, streamed=streamed)
+
+    items = await session.get_items()
+    assert _call_ids(items).count("call_PARKED") == 1
+    assert _call_ids(items).count("call_PARKED_2") == 1
+    outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"}
+    # Only the second call may still be awaiting its output; nothing is duplicated.
+    assert outputs == {"call_PARKED"}
+
+
+@pytest.mark.asyncio
+async def test_a_held_mcp_approval_pair_defers_compaction_when_it_settles() -> None:
+    # The approval response is the locally produced half of its pair and must stay
+    # associated with the response chain that carried the request; compacting that
+    # response before the model consumes the approval drops it in
+    # ``previous_response_id`` mode.
+    from agents.run_internal.session_persistence import save_result_to_session
+
+    session = _CompactionRecordingSession()
+    held: list[TResponseInputItem] = [
+        {
+            "type": "mcp_approval_request",
+            "id": "mcpr_1",
+            "server_label": "srv",
+            "name": "do_it",
+            "arguments": "{}",
+        },
+        {"type": "mcp_approval_response", "approval_request_id": "mcpr_1", "approve": True},
+    ]
+
+    count = await save_result_to_session(
+        session, held, [], None, response_id="resp_parked", settling_held_batch=True
+    )
+
+    assert [entry for entry in session.compactions if "deferred" in entry] == [
+        {"deferred": "resp_parked", "store": None}
+    ]
+    assert [entry for entry in session.compactions if "response_id" in entry] == []
+    assert count == len(await session.get_items())
+
+
+@pytest.mark.asyncio
+async def test_an_ordinary_mcp_approval_response_defers_compaction_too() -> None:
+    # The non-deferred resume commits the approval response as a run item, and the
+    # classification must treat both carriers alike: deferring for the settled dict
+    # but not for the run item would leave the same response compacted or not
+    # depending on which path persisted it.
+    from agents.items import MCPApprovalResponseItem
+    from agents.run_internal.session_persistence import save_result_to_session
+
+    session = _CompactionRecordingSession()
+    agent = _make_deferring_agent()
+    response_item = MCPApprovalResponseItem(
+        agent=agent,
+        raw_item={
+            "type": "mcp_approval_response",
+            "approval_request_id": "mcpr_1",
+            "approve": True,
+        },
+    )
+
+    await save_result_to_session(session, [], [response_item], None, response_id="resp_live")
+
+    assert [entry for entry in session.compactions if "deferred" in entry] == [
+        {"deferred": "resp_live", "store": None}
+    ]
+
+
 @pytest.mark.asyncio
 async def test_the_entry_settle_runs_the_compaction_bookkeeping() -> None:
     # The entry settle goes through the canonical persistence path, so a