fix(agent): invalidate flush-scan cursor when finalizer pops db marker

agent/turn_finalizer.py

@@ -339,6 +339,12 @@ def finalize_turn(
                 # otherwise ``/resume`` reloads ``content=""`` and the bug
                 # resurfaces cross-session.
                 _tail.pop("_db_persisted", None)
+                # The bounded flush-scan cursor (run_agent.py) skips the
+                # identity-matched prefix of its previous snapshot on the
+                # assumption that no live dict loses the marker in place —
+                # this pop is the one place that does. Invalidate it so the
+                # filled row is re-examined instead of skipped.
+                agent._db_flush_scan_prefix = None
 
         # The model has completed its request, so replace API-local
         # voice/model/skill guidance with the clean user input before writing the

tests/agent/test_turn_finalizer_final_response_persistence.py

@@ -180,3 +180,46 @@ def test_final_response_fills_pure_tool_call_tail(monkeypatch):
 
 
 
+
+
+def test_final_response_fill_invalidates_flush_scan_cursor():
+    """The fill's marker pop must invalidate the bounded flush-scan cursor.
+
+    The cursor (run_agent.py) skips the identity-matched prefix of its
+    previous snapshot assuming no live dict loses ``_db_persisted`` in place
+    — the fill is the one path that pops it. Without invalidation, the
+    turn-end flush skips the filled row as 'already stamped' and the
+    delivered answer never reaches state.db (the #43849 class resurfacing).
+    """
+    agent = FakeAgent()
+    agent._db_flush_scan_prefix = ["prior-snapshot"]
+    messages = [
+        {"role": "user", "content": "q"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {"id": "t1", "type": "function",
+                 "function": {"name": "f", "arguments": "{}"}}
+            ],
+            "_db_persisted": True,
+        },
+    ]
+
+    finalize_turn(
+        agent,
+        final_response="Here is your answer.",
+        api_call_count=3,
+        interrupted=False,
+        failed=False,
+        messages=messages,
+        conversation_history=[],
+        effective_task_id="t",
+        turn_id="tid",
+        user_message="q",
+        original_user_message="q",
+        _should_review_memory=False,
+        _turn_exit_reason="text_response(final)",
+    )
+
+    assert agent._db_flush_scan_prefix is None

tests/run_agent/test_identity_flush.py

@@ -197,3 +197,41 @@ def test_stale_seed_id_from_prior_flush_cannot_suppress_new_message(self):
                 assert new_assistant.get("_db_persisted") is True
             finally:
                 db.close()
+
+
+class TestFlushCursorMarkerPop:
+    def test_filled_row_repersists_after_marker_pop_and_cursor_invalidation(self):
+        """End-to-end: incremental flush stamps the tool-call tail; the
+        finalizer fills its content and pops the marker; with the cursor
+        invalidated (as the pop site now does), the turn-end flush must
+        persist the filled answer — not skip the row as already-stamped."""
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "t.db")
+            try:
+                agent = _make_agent(db)
+                tool_row = {
+                    "role": "assistant",
+                    "content": "",
+                    "tool_calls": [
+                        {"id": "t1", "type": "function",
+                         "function": {"name": "f", "arguments": "{}"}}
+                    ],
+                }
+                messages = [tool_row]
+                agent._flush_messages_to_session_db_unlocked(messages)
+                assert messages[0].get("_db_persisted") is True
+                assert agent._db_flush_scan_prefix is not None
+
+                # What finalize_turn's fill does at the pop site:
+                messages[0]["content"] = "the final answer"
+                messages[0].pop("_db_persisted", None)
+                agent._db_flush_scan_prefix = None
+
+                messages.append({"role": "user", "content": "next question"})
+                agent._flush_messages_to_session_db_unlocked(messages)
+
+                assert "the final answer" in _contents(db)
+            finally:
+                db.close()