Python: [Bug]: CompactionProvider.after_run fires once per AgentLoopMiddleware iteration, not once per real user turn - digests same-task tool results mid-task #7236

Description

Problem

create_harness_agent(loop_should_continue=..., after_compaction_strategy=...) is the harness's own documented pattern for a todo-driven agent with bounded persisted history (docs/example: AgentLoopMiddleware + CompactionProvider). But CompactionProvider.after_run fires once per agent run, and AgentLoopMiddleware treats every loop iteration as a full agent run — including re-running tool approval and, critically, re-running after_run on the shared persisted history:

_harness/_agent.py (create_harness_agent, ~L582-584): "When should_continue is supplied, the loop is prepended ahead of tool approval so it sits outermost of all: each loop iteration is a full agent run (including tool approval)..."

Concretely: AgentLoopMiddleware._process_non_streaming/_process_streaming (_harness/_loop.py) call await call_next() once per iteration in a while True loop; each call_next() re-enters the pipeline down to _call_chat_client → _parse_non_streaming_response / _parse_streaming_response, both of which call self._run_after_providers(...) (_agents.py:1146, :1179) at the end of every completed run. _run_after_providers invokes provider.after_run(...) for every context provider, including CompactionProvider — so CompactionProvider.after_run runs at the end of every loop iteration, not once when the loop (and the real user turn) actually finishes.

after_run's own docstring says it should compact "so the next turn starts with a smaller context" — i.e. it's designed as an end-of-turn operation. Nothing prevents it from also firing mid-turn, once per internal iteration, whenever a should_continue loop is attached.

Impact

ToolResultCompactionStrategy(keep_last_tool_call_groups=N) collapses everything but the newest N tool-call groups into flat [Tool results: ...] text. When it runs every loop iteration instead of once per turn, a task with more than N tool-using iterations starts flattening its own, still-in-progress tool results — output the model produced this task, moments ago — into inert summary text before the task is even done. This is not a large-context edge case: create_harness_agent's own loop_max_iterations default is 10 (our app's harness config raises it to 100), so any reasonably long todo list crosses a normal keep-N.

Observed live (deployed harness app, openrouter/hy3, keep-N=2 at the time): the model's own visible reasoning was "tool results come back but they appear as my own assistant text repeating" / "the tool call results are 'no results'" — it concluded its own tools were broken and re-issued the same searches in a loop, because by the time it looked back at a recent-but-already-digested tool call, the verbatim function-call/function-result pair was gone, replaced by flat text it read as evidence nothing came back. This is a real correctness failure (a stuck run), not just wasted context.

Raising N delays this (we currently run N=6) but does not fix it — any task with more than N tool-using iterations still reproduces it, and loop_max_iterations is 10-100 by design specifically to allow long tasks.

Reproduction (minimal, no application code)

import asyncio
from agent_framework import Content, Message
from agent_framework._compaction import CompactionProvider, ToolResultCompactionStrategy

KEEP = 6  # generous keep-N -- the bug reproduces regardless of the exact number

def tool_call_group(call_id, result_text):
    return [
        Message(role="assistant", contents=[Content.from_function_call(call_id=call_id, name="search", arguments={})]),
        Message(role="tool", contents=[Content.from_function_result(call_id=call_id, result=result_text)]),
    ]

class FakeSession:
    def __init__(self, messages):
        self.state = {"in_memory": {"messages": messages}}

async def main():
    provider = CompactionProvider(
        after_strategy=ToolResultCompactionStrategy(keep_last_tool_call_groups=KEEP),
        history_source_id="in_memory",
    )
    messages = [Message(role="user", contents=[Content.from_text(text="Research 10 topics and report back.")])]
    session = FakeSession(messages)

    for i in range(1, 11):
        # One AgentLoopMiddleware iteration = one todo = one tool call + its fresh result.
        messages.extend(tool_call_group(f"todo-{i}", f"finding for todo {i}"))
        # Exactly what _run_after_providers calls at the end of every "run" --
        # and AgentLoopMiddleware treats every loop iteration as a full run.
        await provider.after_run(agent=None, session=session, context=None, state={})
        digested = sum(
            1 for m in messages
            if m.role == "assistant"
            and any(str(getattr(c, "text", "")).startswith("[Tool results:") for c in m.contents)
        )
        print(f"iteration {i:2d}: {digested} tool group(s) already flattened")

asyncio.run(main())

Output:

iteration  1: 0 tool group(s) already flattened
...
iteration  6: 0 tool group(s) already flattened
iteration  7: 1 tool group(s) already flattened   # todo-1's result, produced 6 iterations ago in THIS task
iteration  8: 2 tool group(s) already flattened
iteration  9: 3 tool group(s) already flattened
iteration 10: 4 tool group(s) already flattened

By iteration 7 of a single, still-in-progress 10-todo task, todo-1's tool result — produced 6 iterations earlier in service of the same user request — is already flattened. This uses CompactionProvider directly (no create_harness_agent/AgentLoopMiddleware needed to demonstrate it): the bug is that after_run has no way to know it's being called mid-turn versus at a genuine turn boundary, and the harness's own loop guarantees it's called mid-turn on every iteration but the last.

Why the obvious fixes don't work

• Raise keep_last_tool_call_groups. Only raises the iteration count needed to reproduce it; doesn't change the mechanism. Any task longer than N tool-using iterations still hits it, and loop_max_iterations exists specifically to allow long tasks.
• Run compaction once, after the whole loop returns, from outside agent.run(). Looks natural but is unsafe for any deployment where session state is persisted externally between requests (Redis, blob storage, multi-replica hosting — e.g. via agent-framework-ag-ui's snapshot store): persistence of the run's final state typically happens inside the same call, before control returns to the caller. Compacting after the fact mutates local state after the durable copy was already written with the uncompacted history, so the compaction has no effect in a multi-replica deployment even though it appears to work in-process.

Code Sample

Error Messages / Stack Traces

Package Versions

agent-framework-core: 1.11.0

Python Version

No response

Additional Context

Proposed fix

AgentLoopMiddleware already knows, at the point it decides to continue or stop (_evaluate_stop, _harness/_loop.py), whether another iteration is coming — that's exactly the information after_run needs and can't otherwise obtain. Options, roughly in order of preference:

1. Have the harness compose the two correctly by default. create_harness_agent builds both AgentLoopMiddleware (from loop_should_continue) and the after-phase CompactionProvider (from after_compaction_strategy) itself — it's in a unique position to wire a "loop is continuing" signal from the former into the latter automatically whenever both are supplied, with no new public API needed.
2. Expose a "will continue" signal on the run context/session state that CompactionProvider.after_run (or any ContextProvider.after_run) can read, set by AgentLoopMiddleware before each call_next() reflecting the previous iteration's stop decision (one iteration of lag is fine — the goal is "don't run on iterations 1..N-1", not perfect prediction).
3. At minimum, document the interaction on both AgentLoopMiddleware and CompactionProvider/create_harness_agent's after_compaction_strategy docstrings, so it's a known tradeoff rather than a silent surprise.

Workaround (application-side, today)

ats.compaction.LoopAwareCompactionProvider: subclasses CompactionProvider, and in after_run re-derives the loop's own continuation decision by calling the same should_continue predicate object passed to loop_should_continue (e.g. todos_remaining(looping_modes=["execute"])) before delegating to the real strategy — skipping entirely while it says "continuing," compacting once it says "stop." Backed by a token-count safety valve (force-compact even mid-loop once persisted history crosses a threshold) to cover loop_max_iterations forcing a stop while todos are still open, since that path never sees "not continuing." This is necessarily a proxy — it only works because we happen to have the identical predicate object in hand — which is exactly why we think this belongs in the harness itself: any other should_continue (a judge-based loop via AgentLoopMiddleware.with_judge, a custom predicate) hits the same bug with no equivalent workaround available.