fix(compression): turn-start in-place compaction no longer hides the newest summarized turn (#120187)

archive_and_compact takes the newest tail_count durable rows as the carried tail's superseded
originals (active=0, compacted=0: hidden from display and session_search). The CLI and the gateway
persist a turn's user row only after the turn-start preflight, so at that compaction the row rides
in the carried tail with no durable original of its own. The positional rewind then reached one
row past the tail and flagged the newest summarized message as a superseded duplicate. Each
turn-start compaction hid one more (A5, then A16 in a two-compaction run), on the default config.
The TUI/Desktop are immune because they persist the user row at submit.

tail_count now leaves out this turn's rows that never reached state.db: no persisted marker and
no _row_id, counted within the carried tail window only. That includes the user row and any
unflushed scaffolding. It does so only while the turn holds the session turn lease. Between turns
(a manual /compress, the gateway's pre-turn hygiene) the turn anchor is left over from the last
turn, and a gateway transcript reload is durable but unmarked. Counting those rows would leave
carried originals at compacted=1, the duplicate recall #86366 fixed.

agent/conversation_compression.py

@@ -3622,6 +3622,19 @@ def _commit_compaction(
                 # Tail rows tagged by compress() are archived as superseded duplicates, not
                 # compacted=1. Count against the FINAL list — salvage may have dropped rows.
                 tail_count = sum(1 for m in compressed if id(m) in _tail_tagged_ids)
+                # The rewind takes the newest `tail_count` durable rows as the tail's originals, so a tail row
+                # with none (this turn's user row, which the CLI and gateway persist after preflight; unflushed
+                # scaffolding) would flag a summarized row superseded instead: gone from display and search.
+                # Only while a turn holds the session: between turns (manual /compress, gateway hygiene) the
+                # anchor is the last turn's, and the rows it points at are durable, just unmarked.
+                _turn_idx = getattr(agent, "_persist_user_message_idx", None)
+                if (getattr(agent, "_active_session_turn_lease_holder", None) is not None
+                        and isinstance(_turn_idx, int) and 0 <= _turn_idx < len(messages)):
+                    from agent.context_compressor import _DB_PERSISTED_MARKER
+                    tail_count -= sum(
+                        1 for m in messages[max(_turn_idx, len(messages) - tail_count):]
+                        if isinstance(m, dict) and not m.get(_DB_PERSISTED_MARKER)
+                        and not isinstance(m.get("_row_id"), int))
                 persisted = compressed
                 if verbatim_tail:
                     # The kept exchanges are durable rows under the watermark, so the archive below covers

tests/agent/test_in_place_preflight_rewind.py

@@ -0,0 +1,106 @@
+"""In-place compaction's positional rewind must land on the carried tail's own originals.
+
+``archive_and_compact`` flags the newest ``tail_count`` durable rows as the carried tail's superseded originals
+(``active=0, compacted=0``: hidden from display and search). The CLI and the gateway persist a turn's user row
+only after the turn-start preflight, so that row rides in the tail with no durable original of its own.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from agent.conversation_compression_manual import compress_now, parse_compress_args
+
+
+def _aux_llm(**kwargs):
+    text = "## Goal\nNumbered steps.\n## Progress\nEarly ones done." if kwargs.get("task") == "compression" else "Title"
+    message = SimpleNamespace(content=text, tool_calls=None)
+    return SimpleNamespace(choices=[SimpleNamespace(message=message, finish_reason="stop")], model="aux", usage=None)
+
+
+def _reply(content, prompt_tokens):
+    message = SimpleNamespace(content=content, tool_calls=None)
+    response = SimpleNamespace(choices=[SimpleNamespace(message=message, finish_reason="stop")], model="test/model")
+    response.usage = SimpleNamespace(
+        prompt_tokens=prompt_tokens, completion_tokens=100, total_tokens=prompt_tokens + 100)
+    return response
+
+
+@pytest.fixture
+def session(tmp_path, monkeypatch):
+    from hermes_state import SessionDB
+    from run_agent import AIAgent
+
+    monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
+    monkeypatch.setattr("agent.context_compressor.call_llm", _aux_llm)
+    monkeypatch.setattr("agent.title_generator.call_llm", _aux_llm)
+    db = SessionDB(db_path=tmp_path / "state.db")
+    with patch("agent.process_bootstrap.OpenAI"):
+        agent = AIAgent(api_key="test-key-1234567890", base_url="https://openrouter.ai/api/v1", model="test/model",
+                        quiet_mode=True, session_db=db, session_id="sid", skip_context_files=True, skip_memory=True)
+    agent.client, agent.tool_delay, agent.save_trajectories = MagicMock(), 0, False
+    assert agent.compression_in_place is True
+    yield db, agent
+    db.close()
+
+
+def _turn(db, agent, cli, surface, n, prompt_tokens):
+    """One turn exactly as the classic CLI (staged dict, history[:-1]) or the gateway (transcript reload) runs it."""
+    text = f"U{n} please continue with the next step"
+    agent.client.chat.completions.create.side_effect = [_reply(f"A{n} " + "lorem ipsum dolor " * 300, prompt_tokens)]
+    if surface == "gateway":
+        from gateway.run import _build_gateway_agent_history
+
+        history = _build_gateway_agent_history(db.get_messages_as_conversation("sid", repair_alternation=True))[0]
+        agent.run_conversation(user_message=text, conversation_history=history, task_id="sid", persist_user_message=text)
+        return
+    from hermes_cli.cli_chat_turn_mixin import CLIChatTurnMixin
+
+    CLIChatTurnMixin._chat_stage_user_message(cli, agent, text)
+    result = agent.run_conversation(
+        user_message=text, conversation_history=cli.conversation_history[:-1], task_id="sid", persist_user_message=None)
+    cli.conversation_history = result.get("messages", cli.conversation_history)
+
+
+def _replies_displayed(db):
+    return {m["content"].split(" ")[0] for m in db.get_resume_conversations("sid")[1]
+            if m.get("role") == "assistant" and isinstance(m.get("content"), str)}
+
+
+@pytest.mark.parametrize("surface", ["cli", "gateway"])
+def test_turn_start_compaction_hides_no_summarized_turn(session, surface):
+    db, agent = session
+    cli = SimpleNamespace(conversation_history=[])
+    for n in range(1, 14):
+        _turn(db, agent, cli, surface, n, 5_000)
+    _turn(db, agent, cli, surface, 14, 200_000)  # real usage over the threshold: the next turn compacts first
+    shown = _replies_displayed(db)
+    assert {f"A{n}" for n in range(1, 15)} <= shown
+
+    _turn(db, agent, cli, surface, 15, 20_000)
+
+    assert getattr(agent, "_last_compaction_in_place", None) is True
+    assert {f"A{n}" for n in range(1, 16)} <= _replies_displayed(db)
+
+
+def test_manual_compress_between_turns_rewinds_every_carried_original(session):
+    """Between turns the turn anchor is the last turn's, and a gateway reload is durable but unmarked: counting its
+    rows as unflushed would leave carried originals at compacted=1, so each shows (and is recalled) twice."""
+    db, agent = session
+    cli = SimpleNamespace(conversation_history=[])
+    for n in range(1, 15):
+        _turn(db, agent, cli, "gateway", n, 5_000)
+    assert agent._persist_user_message_idx is not None
+
+    from gateway.run import _build_gateway_agent_history
+
+    history = _build_gateway_agent_history(db.get_messages_as_conversation("sid", repair_alternation=True))[0]
+    assert compress_now(agent, history, parse_compress_args(""), system_message="").status == "compressed"
+
+    model_history, display_history = db.get_resume_conversations("sid")
+    live = [m["content"] for m in model_history if isinstance(m.get("content"), str)]
+    carried = live[next(i for i, c in enumerate(live) if "Numbered steps" in c) + 1:]
+    assert carried
+    display = [m["content"] for m in display_history if isinstance(m.get("content"), str)]
+    assert [display.count(content) for content in carried] == [1] * len(carried)