fix(hold-stream): every exit of a held turn runs its transform snapshot

A held turn (hold_stream=True transform_llm_output) only ran the transform
on the paths that reach finish_text_response / finalize_turn. Other exits
returned model text the plugin never saw:

- A reply that hit finish_reason="length" on every continuation: the
  ceiling exit (_continue_text -> _Trunc.end_turn) stitched the raw
  fragments into the final and the stored row and returned without
  finalize_turn. collapse_continuation_trail now runs the turn's held
  transform snapshot on the stitched text before it appends the row, so
  the user and the stored row get the rewrite (or the withheld notice if
  the transform fails). The same seam covers a provider error that ends
  the turn after a capped fragment (settle_delivered_partial prepends that
  fragment to the error copy).
- The iteration-budget summary appends its own assistant row; finalize
  transformed the returned text but the row kept the raw summary, so
  history showed it. A recovery tail row that still holds the untouched
  text now stores the rewrite before its first flush
  (_store_rewrite_on_tail, the #44239 contract).
- Every remaining exit (non-retryable / exhausted provider errors,
  interrupts, compression and overflow failures, invalid-response caps,
  preflight timeouts, ...) returns through run_conversation, which now
  calls settle_held_final: a held turn whose final_response no seam
  transformed runs the snapshot there, so no path hands the user text the
  plugin did not see first. The hold snapshot is taken before
  build_turn_context, so an early exit never reads the previous turn's
  hold.

Tests (tests/agent/test_hold_stream_exits.py, real AIAgent + SessionDB on
the loopback model, red on 3a5b54d980): four capped responses, a capped
fragment then a 400, the budget summary, and a 400 with a tagging plugin.

Review: https://github.com/NousResearch/hermes-agent/pull/144266#discussion_r4370678627

agent/conversation_loop.py

@@ -1666,6 +1666,12 @@ def _run_conversation_turn(
     except Exception:
         logger.debug("per-turn env credential refresh failed", exc_info=True)
 
+    # A loaded ``hold_stream`` transform in this profile: the user sees only its rewritten final, made
+    # by the transforms live now (a plugin reload mid-turn cannot drop them before the final). Set
+    # before any exit, so an early one never reads the previous turn's hold.
+    agent._held_transforms = held_stream_transforms()
+    agent._hold_stream = bool(agent._held_transforms)
+
     # Per-turn setup: build_turn_context mutates ``agent`` and returns the locals the loop reads.
     try:
         _ctx = build_turn_context(
@@ -1702,10 +1708,6 @@ def _run_conversation_turn(
     # pool refresh tallies cap same-entry refreshes on a persistent 401 (#26080); usage
     # for on_turn_complete() stays None on turns that never reach a response.
     agent._delivered_interim_texts = set()
-    # A loaded ``hold_stream`` transform in this profile: the user sees only its rewritten final, made
-    # by the transforms live now (a plugin reload mid-turn cannot drop them before the final).
-    agent._held_transforms = held_stream_transforms()
-    agent._hold_stream = bool(agent._held_transforms)
     agent._incremental_persistence_failed = False
     agent._last_persistence_error_cause = None
     agent._compression_adoption_failed = False
@@ -1812,6 +1814,7 @@ def run_conversation(
     addresses, after every history rewrite including post-turn micro-compaction.
     """
     from agent.turn_context import export_current_turn_boundary
+    from agent.turn_finalizer import settle_held_final
     from agent.voice_turn_route import end_voice_turn_route
     from tools.vision_tools_history_budget import native_turn_images
 
@@ -1840,6 +1843,7 @@ def run_conversation(
             )
         finally:
             end_voice_turn_route(agent)
+    settle_held_final(agent, result)
     result = export_current_turn_boundary(agent, result, user_message)
     if isinstance(result, dict):
         result["user_intervened"] = bool(getattr(agent, "_turn_user_intervened", False))

agent/turn_finalizer.py

@@ -284,6 +284,15 @@ def _close_transcript_tail(agent, messages, final_response, interrupted, _recove
         _apply_override(messages)
 
 
+def _store_rewrite_on_tail(messages, raw, final) -> None:
+    """A recovery exit that already appended its text as the tail row (the iteration-budget summary)
+    stores the rewrite the user gets, not the model text, while that row is still unflushed (#44239)."""
+    tail = messages[-1] if messages else None
+    if (final != raw and isinstance(tail, dict) and tail.get("role") == "assistant"
+            and tail.get("content") == raw and not tail.get(_DB_PERSISTED_MARKER)):
+        tail["content"] = final
+
+
 def _micro_compact_after_turn(agent, messages, final_response, logger, task_id) -> None:
     """Post-turn micro-compaction: absorb the oldest uncompacted exchange into the
     rolling summary before persist, amortizing compression across turns."""
@@ -547,6 +556,24 @@ def _run_held_transforms(callbacks, payload: dict, logger) -> Optional[str]:
     return HELD_TRANSFORM_FAILED_REPLY if failed else rewrite
 
 
+def settle_held_final(agent, result: Any) -> None:
+    """A held turn's reply leaves only through its transform snapshot. ``finish_text_response``,
+    ``finalize_turn``, the Codex runtime and the truncation ceiling run it before persisting; every
+    other exit (provider errors, retry exhaustion, interrupts, compression failures, ...) returns its
+    text through ``run_conversation``, which calls this, so no path hands the user text the plugin
+    did not see first, and a failed transform yields the withheld notice."""
+    if getattr(agent, "_hold_stream", False) is not True or not isinstance(result, dict):
+        return
+    text, turn_id = result.get("final_response"), getattr(agent, "_current_turn_id", "") or ""
+    recorded = getattr(agent, "_llm_output_transform", None)
+    if not isinstance(text, str) or not text or (isinstance(recorded, tuple) and recorded[:1] == (turn_id,)):
+        return
+    final, transformed, _ = apply_llm_output_transform(agent, text, turn_id=turn_id)
+    if result.get("error") == text:  # keep the mirror an error card reads
+        result["error"] = final
+    result.update(final_response=final, response_transformed=transformed, pre_transform_response=None)
+
+
 def public_pre_transform(agent, pre_transform: Optional[Any]) -> Optional[Any]:
     """``pre_transform_response`` for a turn result: a held turn's original never leaves the agent
     (receipts, replay-gap recovery, post-stream suffix painting all see only the rewrite)."""
@@ -659,7 +686,9 @@ def finalize_turn(
         # gets the recorded outcome back. Either way the tail close below writes the text the
         # user will see, never the raw model text (#44239).
         if final_response and not interrupted:
+            _raw = final_response
             final_response, _, _ = apply_llm_output_transform(agent, final_response, turn_id=turn_id, logger=logger)
+            _store_rewrite_on_tail(messages, _raw, final_response)
         _close_transcript_tail(agent, messages, final_response, interrupted, _recovered_from_stream)
         if not interrupted and not failed:
             _micro_compact_after_turn(agent, messages, final_response, logger, effective_task_id)

agent/turn_truncation.py

@@ -47,7 +47,8 @@ def collapse_continuation_trail(
     finish_reason: str, parts: Optional[list[str]] = None,
 ) -> str:
     """Drop this turn's ``_length_continuation_fragment``/``_nudge`` rows and append one
-    assistant row holding the joined, think-stripped partial; returns that text ("" none).
+    assistant row holding the joined, think-stripped partial (a held turn's rewrite); returns
+    that text ("" none).
 
     ``parts=None`` (retry exhaustion, #119001): the text comes from the fragment rows and
     nothing happens without a valid turn index or a trail — an unanswered synthetic nudge
@@ -86,6 +87,11 @@ def collapse_continuation_trail(
     partial = agent._strip_think_blocks(
         _join_truncated_parts(join_parts)
     ).strip()
+    if partial and getattr(agent, "_hold_stream", False) is True:
+        # A held turn showed none of these fragments: the stitched text reaches the user and the
+        # stored row only through the turn's transform snapshot (fail closed: the withheld notice).
+        from agent.turn_finalizer import apply_llm_output_transform
+        partial, _, _ = apply_llm_output_transform(agent, partial, turn_id=getattr(agent, "_current_turn_id", "") or "")
     if partial:
         append_message(messages, {"role": "assistant", "content": partial, "finish_reason": finish_reason})
     agent._session_messages = messages

tests/agent/test_hold_stream_exits.py

@@ -0,0 +1,99 @@
+"""Every exit of a held turn (``hold_stream=True`` transform) goes through the turn-start transform
+snapshot.
+
+Through a real ``AIAgent`` + ``SessionDB`` against the loopback model: a reply capped by the output
+limit on every continuation, an iteration-budget summary, and a provider error that ends the turn
+reach the user (and the stored row) only as the rewrite.
+"""
+
+import os
+from pathlib import Path
+
+import pytest
+
+from tests.fakes.fake_llm_provider import Error, FakeLLMServer, Text, ToolCall
+from tests.fakes.redactor_plugin import RAW, REWRITE, redactor_home
+
+
+def _agent(server, db, session_id: str, **kwargs):
+    from run_agent import AIAgent
+
+    return AIAgent(base_url=server.base_url, api_key="sk-fake-e2e", provider="custom", model="fake-model",
+                   quiet_mode=True, skip_context_files=True, skip_memory=True, enabled_toolsets=["todo"],
+                   session_id=session_id, session_db=db, **kwargs)
+
+
+@pytest.fixture
+def held(tmp_path):
+    """``run(script, plugin_body=None, **agent_kwargs) -> (result, shown, displayed history, server, db)``
+    for one held turn."""
+    from gateway.session_controls import display_projection
+    from hermes_cli import plugins
+    from hermes_state import SessionDB
+
+    home = Path(os.environ["HERMES_HOME"])
+    opened = []
+
+    def run(script, plugin_body=None, **kwargs):
+        server = FakeLLMServer(script)
+        server.start()
+        opened.append(server)
+        redactor_home(home, server.base_url, hold=True)
+        if plugin_body:
+            (home / "plugins" / "redactor" / "__init__.py").write_text(plugin_body, encoding="utf-8")
+        plugins._reset_plugin_managers_for_tests()
+        db = SessionDB(db_path=tmp_path / "held.db")
+        opened.append(db)
+        agent = _agent(server, db, "held-exit", **kwargs)
+        shown = []
+        agent.stream_delta_callback = shown.append
+        agent.interim_assistant_callback = lambda text, **_: shown.append(text)
+        result = agent.run_conversation("show config")
+        displayed = display_projection(db.get_messages("held-exit"), home)
+        return result, [text for text in shown if text], displayed, server, db
+
+    yield run
+    for item in opened:
+        (item.stop if isinstance(item, FakeLLMServer) else item.close)()
+    plugins._reset_plugin_managers_for_tests()
+
+
+def _assistant_texts(displayed) -> list:
+    return [row["text"] for row in displayed if row["role"] == "assistant"]
+
+
+def test_reply_capped_on_every_continuation_is_still_rewritten(held):
+    # Four finish_reason="length" responses: the ceiling exit keeps the stitched partial.
+    result, shown, displayed, _, _ = held([Text(RAW, finish_reason="length")] * 4)
+    assert result.get("partial") and "SECRET-" not in result["final_response"]
+    assert "[REDACTED]" in result["final_response"]
+    assert result.get("pre_transform_response") is None
+    assert shown == []
+    assert "SECRET-" not in str(displayed) and _assistant_texts(displayed) == [result["final_response"]]
+
+
+def test_iteration_budget_summary_is_stored_as_the_rewrite(held):
+    result, shown, displayed, _, _ = held([ToolCall("todo_list", {}), Text(RAW)], max_iterations=1)
+    assert result["final_response"] == REWRITE
+    assert shown == []
+    assert "SECRET-" not in str(displayed) and REWRITE in _assistant_texts(displayed)
+
+
+_TAGGING_PLUGIN = (
+    "def tag(response_text, **_):\n    return '[filtered] ' + response_text\n"
+    "def register(ctx):\n    ctx.register_hook('transform_llm_output', tag, hold_stream=True)\n"
+)
+
+
+def test_a_provider_error_exit_also_runs_the_held_transform(held):
+    result, shown, _, _, _ = held([Error(status=400, message="bad request")], plugin_body=_TAGGING_PLUGIN)
+    assert result.get("failed") and result["final_response"].startswith("[filtered] ")
+    assert shown == []
+
+
+def test_provider_error_after_a_capped_fragment_delivers_only_the_rewrite(held):
+    # The error exit prepends the fragment the turn produced; held, the user never saw it raw.
+    result, shown, displayed, _, _ = held([Text(RAW, finish_reason="length"), Error(status=400, message="bad request")])
+    assert result.get("failed") and REWRITE in result["final_response"]
+    assert "SECRET-" not in result["final_response"] and shown == []
+    assert "SECRET-" not in str(displayed) and REWRITE in _assistant_texts(displayed)