fix: transform_llm_output result is what the session stores and replays

A plugin's transform_llm_output replacement reached final_response only.
agent/turn_finalizer.py::finalize_turn fired the hook after the assistant
row had already been persisted — and the row is first written even earlier,
in agent/turn_final_response.py::finish_text_response's durable flush — so
messages[-1], the SQLite/JSON session, /resume and the next turn's replay all
kept the raw model text while the user had seen the rewritten one.

Writing the transformed text back after that flush cannot work: SQLite treats
a non-blank assistant row as settled (resolve_and_repair_transcript_batch
adopts the stored content instead of overwriting it), so the only correct seam
is BEFORE the row is first persisted. apply_llm_output_transform (new, in
turn_finalizer) fires the hook once per turn_id and records the outcome;
finish_text_response calls it ahead of append+flush and writes the result into
the row (api_content for the promoted-reasoning sidecar), finalize_turn's
_persist_step calls it ahead of the recovery-path tail close, and
_apply_output_hooks reads the recorded outcome (firing only when no earlier
seam saw a response) before post_llm_call. Only the current turn's not-yet-
written text changes — earlier turns and the system prompt are untouched.

post_llm_call is unchanged: it is an observer whose return is ignored by
contract, so there is nothing of it to persist (#14913/#44253's premise).

Fixes #44239
Slim redo of #44244 (AIalliAI, earliest; same sync-then-persist idea, moved to
the pre-flush seam) — also supersedes #65921 (SingleVirgin, sibling fix).

Co-authored-by: AIalliAI <285906080+AIalliAI@users.noreply.github.com>
(cherry picked from commit 5fe02a0aeecef422a4ffb4ff4385015f6a1528a0)

agent/turn_final_response.py

@@ -301,6 +301,22 @@ def finish_text_response(
         final_response = None
         return _verdict("continue")
 
+    # Plugins rewrite the reply BEFORE it is appended and flushed: SQLite treats a non-blank
+    # assistant row as settled, so a transform after this write would reach the user but never
+    # the stored/replayed transcript (#44239). finalize_turn reads the recorded outcome; like
+    # there, an interrupted turn keeps the raw text.
+    from agent.turn_finalizer import apply_llm_output_transform
+    _transformed = False
+    if not getattr(agent, "_interrupt_requested", False):
+        final_response, _transformed, _ = apply_llm_output_transform(
+            agent, final_response, turn_id=getattr(agent, "_current_turn_id", "") or "", logger=logger,
+        )
+    if _transformed:
+        if _promoted:
+            final_msg["api_content"] = final_response
+        else:
+            final_msg["content"] = final_response
+
     append_message(messages, final_msg)
     # Make the answer durable before leaving the loop (_DB_PERSISTED_MARKER keeps
     # _persist_session idempotent). Failure must NOT abort the turn: finalize retries.

agent/turn_finalizer.py

@@ -413,21 +413,16 @@ def _apply_output_hooks(
     agent, final_response, logger, *, platform, effective_task_id, turn_id, original_user_message,
     messages,
 ) -> Tuple[Any, bool, Optional[Any]]:
-    """Fire ``transform_llm_output`` then ``post_llm_call`` once per turn after the tool loop.
-    Returns ``(final_response, transformed, pre_transform_response)``."""
-    transformed, pre_transform = False, None
-    # First hook to return a string wins; None/empty leaves the text unchanged.
-    for _hook_result in _invoke_hook_safely(
-        "transform_llm_output", logger,
-        response_text=final_response,
-        session_id=agent.session_id or "",
-        model=agent.model,
-        platform=platform,
-        turn_id=turn_id,  # per-turn identity for the hook callback gate
-    ):
-        if isinstance(_hook_result, str) and _hook_result:
-            pre_transform, final_response, transformed = final_response, _hook_result, True
-            break
+    """Resolve the turn's ``transform_llm_output`` outcome, then fire ``post_llm_call`` once per
+    turn after the tool loop. Returns ``(final_response, transformed, pre_transform_response)``.
+
+    The transform itself normally already ran before the assistant row was first persisted
+    (``apply_llm_output_transform`` from ``finish_text_response`` / ``_persist_step``); this
+    call returns that recorded outcome, and only fires the hook here when no earlier seam saw a
+    response (e.g. text that only appeared through ``_explain_abnormal_exit``)."""
+    final_response, transformed, pre_transform = apply_llm_output_transform(
+        agent, final_response, turn_id=turn_id, platform=platform, logger=logger,
+    )
     # Detached forks are internal work and must not publish turns under the parent's session ID.
     if not getattr(agent, "_persist_disabled", False):
         _invoke_hook_safely(
@@ -444,6 +439,48 @@ def _apply_output_hooks(
     return final_response, transformed, pre_transform
 
 
+def apply_llm_output_transform(
+    agent, final_response, *, turn_id, platform=None, logger=None,
+) -> Tuple[Any, bool, Optional[Any]]:
+    """Fire ``transform_llm_output`` once per turn and return
+    ``(final_response, transformed, pre_transform_response)``.
+
+    Called BEFORE the final assistant row is first persisted — from ``finish_text_response``
+    ahead of its durable flush, and from ``finalize_turn._persist_step`` ahead of the
+    recovery-path tail close — so the text the user sees is the text stored in SQLite/JSON and
+    replayed next turn (#44239). SQLite treats a non-blank assistant row as settled (a re-flush
+    adopts the stored content rather than overwriting it), so transforming after that first
+    write can never reach the durable store. Idempotent per ``turn_id``: later callers in the
+    same turn get the recorded outcome instead of a second hook firing. Only the current
+    turn's not-yet-written text is touched — earlier turns and the system prompt are never
+    rewritten (prompt-cache invariant)."""
+    if logger is None:
+        from agent.conversation_loop import logger
+    recorded = getattr(agent, "_llm_output_transform", None)
+    if isinstance(recorded, tuple) and len(recorded) == 3 and recorded[0] == turn_id:
+        _, transformed, pre_transform = recorded
+        return final_response, transformed, pre_transform
+    if not final_response:
+        return final_response, False, None
+    if platform is None:
+        platform = getattr(agent, "platform", None) or ""
+    transformed, pre_transform = False, None
+    # First hook to return a string wins; None/empty leaves the text unchanged.
+    for _hook_result in _invoke_hook_safely(
+        "transform_llm_output", logger,
+        response_text=final_response,
+        session_id=agent.session_id or "",
+        model=agent.model,
+        platform=platform,
+        turn_id=turn_id,  # per-turn identity for the hook callback gate
+    ):
+        if isinstance(_hook_result, str) and _hook_result:
+            pre_transform, final_response, transformed = final_response, _hook_result, True
+            break
+    agent._llm_output_transform = (turn_id, transformed, pre_transform)
+    return final_response, transformed, pre_transform
+
+
 def finalize_turn(
     agent, *, final_response, api_call_count, interrupted, failed, messages, conversation_history,
     effective_task_id, turn_id, user_message, original_user_message, _should_review_memory,
@@ -508,6 +545,12 @@ def finalize_turn(
         final_response, _recovered_from_stream = _recover_final_from_stream(
             agent, final_response, interrupted, failed
         )
+        # Recovery paths (stream-recovered / prior-turn text) reach here with a response no
+        # earlier seam transformed; the normal text turn already did this before its flush and
+        # 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:
+            final_response, _, _ = apply_llm_output_transform(agent, final_response, turn_id=turn_id, logger=logger)
         _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)

tests/agent/test_transform_llm_output_persistence.py

@@ -0,0 +1,101 @@
+"""#44239: what ``transform_llm_output`` shows the user is what the session stores and replays.
+
+The hook used to fire after the final assistant row had been flushed to SQLite, so
+``final_response`` carried the rewritten text while ``messages[-1]``, the SQLite row and the
+next turn's replay kept the raw model text.
+"""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+
+def _rewriting_hook(calls):
+    def invoke_hook(hook_name, **kwargs):
+        calls.append(hook_name)
+        if hook_name == "transform_llm_output":
+            return ["REWRITTEN:" + kwargs["response_text"]]
+        return []
+    return invoke_hook
+
+
+def _fake_completion(text):
+    def create(**kwargs):
+        msg = SimpleNamespace(content=text, tool_calls=None, reasoning=None)
+        return SimpleNamespace(
+            choices=[SimpleNamespace(message=msg, finish_reason="stop")],
+            usage=SimpleNamespace(prompt_tokens=10, completion_tokens=5, total_tokens=15), model="fake/model",
+        )
+    return create
+
+
+@pytest.fixture
+def db_agent(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    (tmp_path / ".hermes").mkdir()
+    db = SessionDB(db_path=tmp_path / ".hermes" / "state.db")
+    with (
+        patch("model_tools.get_tool_definitions", return_value=[]),
+        patch("model_tools.check_toolset_requirements", return_value={}),
+        patch("agent.process_bootstrap.OpenAI"),
+    ):
+        agent = AIAgent(
+            api_key="test-key-1234567890", base_url="https://openrouter.ai/api/v1", model="fake/model",
+            quiet_mode=True, skip_context_files=True, skip_memory=True, platform="cli",
+            session_id="sess-44239", session_db=db,
+        )
+    agent.client = MagicMock()
+    return agent, db
+
+
+def test_transformed_reply_is_the_stored_and_replayed_text(db_agent, monkeypatch):
+    agent, db = db_agent
+    calls = []
+    monkeypatch.setattr("hermes_cli.lifecycle.invoke_hook", _rewriting_hook(calls))
+    agent.client.chat.completions.create = _fake_completion("RAW MODEL TEXT")
+
+    result = agent.run_conversation("hello")
+
+    last_assistant = next(m for m in reversed(result["messages"]) if m.get("role") == "assistant")
+    stored = [r["content"] for r in db.get_messages("sess-44239") if r["role"] == "assistant"]
+    assert result["final_response"] == "REWRITTEN:RAW MODEL TEXT"
+    assert last_assistant["content"] == result["final_response"]
+    assert stored == [result["final_response"]]
+    assert result["response_transformed"] is True
+    assert result["pre_transform_response"] == "RAW MODEL TEXT"
+    # Exactly once per turn: a second firing would nest the prefix.
+    assert calls.count("transform_llm_output") == 1
+
+
+def test_recovery_path_tail_row_carries_transformed_text(db_agent, monkeypatch):
+    """A recovery ``break`` returns ``final_response`` with no closing assistant row; the row
+    finalize_turn appends must be the transformed text, and the hook still fires once."""
+    from agent.turn_finalizer import finalize_turn
+
+    agent, _db = db_agent
+    calls = []
+    monkeypatch.setattr("hermes_cli.lifecycle.invoke_hook", _rewriting_hook(calls))
+    agent._persist_session = lambda *a, **k: None
+    agent._current_turn_id = "turn-r"
+    messages = [
+        {"role": "user", "content": "do a thing"},
+        {"role": "assistant", "content": "", "tool_calls": [{"id": "c1", "function": {"name": "read_file", "arguments": "{}"}}]},
+        {"role": "tool", "tool_call_id": "c1", "content": json.dumps({"ok": True})},
+    ]
+
+    result = finalize_turn(
+        agent, final_response="RECOVERED TEXT", api_call_count=1, interrupted=False, failed=False,
+        messages=messages, conversation_history=None, effective_task_id="task-1", turn_id="turn-r",
+        user_message="do a thing", original_user_message="do a thing", _should_review_memory=False,
+        _turn_exit_reason="partial_stream_recovery",
+    )
+
+    assert result["final_response"].startswith("REWRITTEN:RECOVERED TEXT")
+    assert messages[-1]["role"] == "assistant"
+    assert messages[-1]["content"] == "REWRITTEN:RECOVERED TEXT"
+    assert calls.count("transform_llm_output") == 1

website/docs/user-guide/features/hooks.md

@@ -1563,7 +1563,7 @@ Pairs with `transform_tool_result`, which runs afterward for every tool, includi
 
 ### `transform_llm_output`
 
-Fires **once per turn** after the tool-calling loop completes and the model has produced a final response, **before** that response is delivered to the user (CLI, gateway, or programmatic caller). Lets a plugin rewrite the assistant's final text using classical-programming methods — no extra inference tokens burned on SOUL flavor text or a skill-driven transform.
+Fires **once per turn** after the tool-calling loop completes and the model has produced a final response, **before** that response is delivered to the user (CLI, gateway, or programmatic caller) and **before** the assistant row is persisted — the replacement is what the session stores, what `/resume` shows and what the next turn replays, so the transcript never diverges from what the user saw. Hermes' own trailers (the file-mutation warning, the abnormal-exit note) are appended afterwards and are not part of `response_text`. Lets a plugin rewrite the assistant's final text using classical-programming methods — no extra inference tokens burned on SOUL flavor text or a skill-driven transform.
 
 **Callback signature:**