fix(sessions): fence transcript writes with the turn-lease holder

Refresh-loss interrupt is cooperative, so a stalled writer could still flush after another process reclaimed the conversation. Carry the holder into append_message / append_messages_batch and reject the write in the same SQLite transaction when the lease row is missing, expired, or owned by someone else.

agent/turn_finalizer.py

@@ -699,8 +699,8 @@ def finalize_turn(
             "health (`hermes doctor`), then send your message again"
         )
         # Machine-readable cause for the gateway/desktop: exactly
-        # 'session_persistence_failed:<locked|disk|unknown>'. Never clobber a
-        # failure_reason another path already stamped on this result.
+        # 'session_persistence_failed:<locked|compression|turn_lease|disk|unknown>'.
+        # Never clobber a failure_reason another path already stamped.
         if "failure_reason" not in result:
             _cause = getattr(agent, "_last_persistence_error_cause", None)
             result["failure_reason"] = (

hermes_state.py

@@ -1492,7 +1492,13 @@ def is_disk_full_error(exc: BaseException | str | None) -> bool:
 # enumerate causes (e.g. the cron scheduler's explainer-variant suppression)
 # must iterate this tuple instead of hardcoding the list, so adding a bucket
 # can never silently desynchronize them.
-PERSISTENCE_ERROR_CAUSES = ("locked", "compression", "disk", "unknown")
+PERSISTENCE_ERROR_CAUSES = (
+    "locked",
+    "compression",
+    "turn_lease",
+    "disk",
+    "unknown",
+)
 
 
 def classify_persistence_error(exc_or_str) -> str:
@@ -1509,6 +1515,9 @@ def classify_persistence_error(exc_or_str) -> str:
       database write lock); transient, retry-later guidance applies.
     * ``"compression"`` — a live compression lease refused the transcript
       write; the database itself is healthy and unlocked.
+    * ``"turn_lease"`` — a presented session-turn-lease holder no longer
+      owns the conversation (expired, released, or reclaimed); fail-fast
+      fencing, not a storage fault.
     * ``"disk"``    — disk full / read-only / permission-shaped failures
       (delegates the disk-full patterns to :func:`is_disk_full_error` so the
       two classifiers can never drift apart — e.g. ENOSPC).
@@ -1521,9 +1530,13 @@ def classify_persistence_error(exc_or_str) -> str:
     # writer" / "Compression lease lost") contains neither "locked" nor
     # "busy", so it must be matched by type and by phrase (for strings that
     # survived RPC wrapping).
+    if isinstance(exc_or_str, SessionTurnLeaseLostError):
+        return "turn_lease"
     if isinstance(exc_or_str, CompressionSessionBusyError):
         return "compression"
     text = str(exc_or_str).lower()
+    if "turn lease" in text:
+        return "turn_lease"
     if "being compressed" in text or "compression lease" in text:
         return "compression"
     if (
@@ -2323,6 +2336,16 @@ class SessionCompressionInProgressError(CompressionSessionBusyError):
     """
 
 
+class SessionTurnLeaseLostError(RuntimeError):
+    """A transcript write presented a turn-lease holder that no longer owns it.
+
+    Fail-fast fencing: do not retry inside ``_execute_write``. The caller
+    either still thinks it owns the conversation after expiry/reclaim, or
+    the lease row is gone. A later writer may already be persisting a
+    newer turn; landing this write would interleave a stale reply.
+    """
+
+
 def _connect_tracked_db(path, tracking_path=None, **kwargs):
     """``sqlite3.connect`` that registers the open fd for lock-safety.
 
@@ -8240,7 +8263,11 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         return None
 
     def _check_transcript_write_guards(
-        self, conn, session_id: str, compression_lock_holder: Optional[str]
+        self,
+        conn,
+        session_id: str,
+        compression_lock_holder: Optional[str],
+        turn_lease_holder: Optional[str] = None,
     ) -> None:
         """Transcript-append admission checks, run INSIDE the write txn.
 
@@ -8261,6 +8288,22 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             raise SessionCompressionInProgressError(
                 f"Session {session_id!r} is being compressed by another writer"
             )
+        if turn_lease_holder:
+            conversation_id = self._session_turn_lease_key_on_conn(conn, session_id)
+            lease = conn.execute(
+                "SELECT holder, expires_at FROM session_turn_leases "
+                "WHERE conversation_id = ?",
+                (conversation_id,),
+            ).fetchone()
+            if (
+                lease is None
+                or lease["holder"] != turn_lease_holder
+                or float(lease["expires_at"]) <= time.time()
+            ):
+                raise SessionTurnLeaseLostError(
+                    f"Session turn lease lost; refusing transcript write "
+                    f"for {session_id!r}"
+                )
         session = conn.execute(
             "SELECT ended_at, end_reason FROM sessions WHERE id = ?",
             (session_id,),
@@ -8340,6 +8383,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         display_kind: Optional[str] = None,
         display_metadata: Optional[Dict[str, Any]] = None,
         compression_lock_holder: Optional[str] = None,
+        turn_lease_holder: Optional[str] = None,
     ) -> int:
         """
         Append a message to a session. Returns the message row ID.
@@ -8398,7 +8442,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
 
         def _do(conn):
             self._check_transcript_write_guards(
-                conn, session_id, compression_lock_holder
+                conn,
+                session_id,
+                compression_lock_holder,
+                turn_lease_holder=turn_lease_holder,
             )
             cursor = conn.execute(
                 """INSERT INTO messages (session_id, role, content, tool_call_id,
@@ -8460,6 +8507,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         session_id: str,
         messages: List[Dict[str, Any]],
         compression_lock_holder: Optional[str] = None,
+        turn_lease_holder: Optional[str] = None,
         chunk_rows: Optional[int] = None,
     ) -> int:
         """Append multiple messages atomically in ONE write transaction.
@@ -8499,12 +8547,16 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                     session_id,
                     messages[start:start + chunk_rows],
                     compression_lock_holder=compression_lock_holder,
+                    turn_lease_holder=turn_lease_holder,
                 )
             return inserted_total
 
         def _do(conn):
             self._check_transcript_write_guards(
-                conn, session_id, compression_lock_holder
+                conn,
+                session_id,
+                compression_lock_holder,
+                turn_lease_holder=turn_lease_holder,
             )
             inserted, tool_calls_total = self._insert_message_rows(
                 conn, session_id, messages

run_agent.py

@@ -2290,6 +2290,9 @@ class AIAgent:
                     compression_lock_holder=getattr(
                         self, "_active_compression_lock_holder", None
                     ),
+                    turn_lease_holder=getattr(
+                        self, "_active_session_turn_lease_holder", None
+                    ),
                 )
                 for _written in _batch_msgs:
                     _written[_DB_PERSISTED_MARKER] = True
@@ -3729,6 +3732,14 @@ class AIAgent:
                     "compressing this session. Your message should already be "
                     "saved — please send it again after compression completes."
                 )
+            if cause == "turn_lease":
+                return (
+                    prefix
+                    + "the turn was stopped because another Hermes process "
+                    "took over this session. Your reply was not saved — wait "
+                    "for the other process to finish, then send your message "
+                    "again."
+                )
             if cause == "locked":
                 return (
                     prefix
@@ -8212,8 +8223,11 @@ class AIAgent:
                     }
 
                 # Assign only after admission so finally release cannot target a
-                # holder string that never owned the row.
+                # holder string that never owned the row. Persist paths read
+                # the agent attr so a late flush after reclaim is fenced in
+                # the same SQLite write transaction as the transcript insert.
                 durable_turn_lease = _durable_holder
+                self._active_session_turn_lease_holder = _durable_holder
                 if _lease_waited:
                     self._emit_status(
                         "Session is free; loading the latest transcript..."
@@ -8396,6 +8410,11 @@ class AIAgent:
                                 session_id,
                                 exc_info=True,
                             )
+                        if (
+                            getattr(self, "_active_session_turn_lease_holder", None)
+                            == durable_turn_lease
+                        ):
+                            self._active_session_turn_lease_holder = None
                     # Always clear mid-turn labels when the turn exits — including
                     # interrupted early returns that skip finalize_turn. Keep ts.
                     try:

tests/run_agent/test_cross_process_turn_lease.py

@@ -5,7 +5,9 @@ from __future__ import annotations
 import sqlite3
 import threading
 import time
+from types import SimpleNamespace
 
+from hermes_state import SessionDB
 from run_agent import AIAgent
 
 
@@ -379,3 +381,128 @@ def test_late_refresh_miss_after_release_does_not_interrupt(monkeypatch):
     assert result["final_response"] == "ok"
     assert interrupt_calls == []
     assert agent._interrupt_requested is False
+
+
+def test_run_conversation_exposes_holder_for_fenced_flush(monkeypatch):
+    """The acquired holder is visible to persist, then cleared on release."""
+    db = _DB()
+    captured = {}
+
+    def append_messages_batch(session_id, messages, **kwargs):
+        captured["session_id"] = session_id
+        captured["turn_lease_holder"] = kwargs.get("turn_lease_holder")
+        captured["count"] = len(messages)
+        return len(messages)
+
+    db.append_messages_batch = append_messages_batch
+    agent = _agent_with_db(db)
+    agent._last_flushed_db_idx = 0
+    agent._flushed_db_message_ids = set()
+    agent._flushed_db_message_session_id = None
+    agent._db_flush_scan_prefix = None
+    agent._pending_cli_user_message = None
+    agent._session_persist_lock = None
+
+    def fake_run(_agent, _message, _system, history, *_args, **_kwargs):
+        captured["active"] = getattr(
+            _agent, "_active_session_turn_lease_holder", None
+        )
+        ok = _agent._flush_messages_to_session_db(
+            [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": "done"},
+            ],
+            [],
+        )
+        captured["flush_ok"] = ok
+        return {"final_response": "done", "messages": history, "failed": False}
+
+    monkeypatch.setattr("agent.conversation_loop.run_conversation", fake_run)
+    result = AIAgent.run_conversation(
+        agent,
+        "new message",
+        conversation_history=[{"role": "user", "content": "durable latest"}],
+    )
+
+    assert result["final_response"] == "done"
+    assert captured["flush_ok"] is True
+    assert captured["active"]
+    assert captured["active"].startswith("pid=")
+    assert captured["turn_lease_holder"] == captured["active"]
+    assert captured["session_id"] == "compressed-tip"
+    assert captured["count"] == 2
+    assert getattr(agent, "_active_session_turn_lease_holder", None) is None
+    assert [event[0] for event in db.events] == [
+        "acquire",
+        "resolve",
+        "reload",
+        "release",
+    ]
+
+
+def _flush_agent(db, session_id):
+    """Bind the real flush onto a stand-in so we can use a live SessionDB."""
+    agent = SimpleNamespace(
+        _session_db=db,
+        _session_db_created=True,
+        _persist_disabled=False,
+        session_id=session_id,
+        _session_persist_lock=None,
+        _flushed_db_message_ids=set(),
+        _flushed_db_message_session_id=None,
+        _last_flushed_db_idx=0,
+        _db_flush_scan_prefix=None,
+        _persist_user_message_idx=None,
+        _persist_user_message_override=None,
+        _persist_user_message_timestamp=None,
+        _pending_cli_user_message=None,
+        _active_session_turn_lease_holder=None,
+        _last_persistence_error_cause=None,
+    )
+    agent._ensure_db_session = lambda: None
+    agent._flush_messages_to_session_db = (
+        AIAgent._flush_messages_to_session_db.__get__(agent, AIAgent)
+    )
+    agent._flush_messages_to_session_db_unlocked = (
+        AIAgent._flush_messages_to_session_db_unlocked.__get__(agent, AIAgent)
+    )
+    return agent
+
+
+def test_flush_messages_to_session_db_fences_stale_holder_on_live_db(tmp_path):
+    """A-loses / B-acquires / A-late-flush, through the real persist path."""
+    path = tmp_path / "state.db"
+    first = SessionDB(path)
+    second = SessionDB(path)
+    first.create_session("shared", source="test")
+    stale_holder = "pid=1:turn=stale"
+    next_holder = "pid=2:turn=next"
+    assert first.try_acquire_session_turn_lease(
+        "shared", stale_holder, ttl_seconds=5
+    )
+
+    agent = _flush_agent(first, "shared")
+    agent._active_session_turn_lease_holder = stale_holder
+    owned = [{"role": "user", "content": "stale-owned"}]
+    assert agent._flush_messages_to_session_db(owned, []) is True
+    assert [m["content"] for m in first.get_messages("shared")] == ["stale-owned"]
+
+    first.release_session_turn_lease("shared", stale_holder)
+    assert second.try_acquire_session_turn_lease(
+        "shared", next_holder, ttl_seconds=5
+    )
+
+    late = [{"role": "assistant", "content": "late stale reply"}]
+    assert agent._flush_messages_to_session_db(late, []) is False
+    assert agent._last_persistence_error_cause == "turn_lease"
+    assert [m["content"] for m in second.get_messages("shared")] == ["stale-owned"]
+
+    agent._active_session_turn_lease_holder = next_holder
+    assert agent._flush_messages_to_session_db(late, []) is True
+    assert [m["content"] for m in second.get_messages("shared")] == [
+        "stale-owned",
+        "late stale reply",
+    ]
+    second.release_session_turn_lease("shared", next_holder)
+    first.close()
+    second.close()

tests/run_agent/test_turn_completion_explainer.py

@@ -119,6 +119,18 @@ def test_explanation_persistence_compression_cause_is_specific():
     assert "disk" not in lower
 
 
+def test_explanation_persistence_turn_lease_cause_is_specific():
+    out = AIAgent._format_turn_completion_explanation(
+        "session_persistence_failed", "turn_lease"
+    )
+    lower = out.lower()
+    assert "took over" in lower
+    assert "not saved" in lower
+    assert "disk" not in lower
+    assert "compression" not in lower
+    assert "hermes doctor" not in lower
+
+
 def test_explanation_persistence_disk_cause_keeps_disk_wording():
     out = AIAgent._format_turn_completion_explanation(
         "session_persistence_failed", "disk"
@@ -232,6 +244,19 @@ def test_classify_persistence_error_compression_busy_is_distinct():
     ) == "compression"
 
 
+def test_classify_persistence_error_turn_lease_lost_is_distinct():
+    from hermes_state import SessionTurnLeaseLostError, classify_persistence_error
+
+    assert classify_persistence_error(
+        SessionTurnLeaseLostError(
+            "Session turn lease lost; refusing transcript write for 'abc'"
+        )
+    ) == "turn_lease"
+    assert classify_persistence_error(
+        "Session turn lease lost; refusing transcript write for 'abc'"
+    ) == "turn_lease"
+
+
 def test_persistence_error_causes_tuple_matches_classifier():
     """PERSISTENCE_ERROR_CAUSES must cover every value the classifier can
     return (consumers like cron suppression iterate it)."""
@@ -240,6 +265,7 @@ def test_persistence_error_causes_tuple_matches_classifier():
     probes = (
         "database is locked",
         "Session 'abc' is being compressed by another writer",
+        "Session turn lease lost; refusing transcript write for 'abc'",
         "database or disk is full",
         "something else entirely",
         None,

tests/state/test_session_turn_lease.py

@@ -11,7 +11,7 @@ from types import SimpleNamespace
 import pytest
 
 import hermes_state
-from hermes_state import SessionDB
+from hermes_state import SessionDB, SessionTurnLeaseLostError
 
 
 def test_turn_lease_serializes_separate_session_db_instances(tmp_path):
@@ -436,4 +436,226 @@ def test_non_expired_turn_lease_from_dead_pid_is_reclaimed(
     assert db.try_acquire_session_turn_lease(
         "shared", fresh_holder, ttl_seconds=300
     ) is True
-    assert probed == [424242]
+    assert probed == [424242]
+
+
+def test_turn_lease_fences_stale_transcript_flush_after_reclaim(tmp_path):
+    """A lost holder cannot persist after B has taken the conversation.
+
+    Refresh-loss interrupt is cooperative; the lease itself must reject the
+    late append inside the same SQLite write transaction.
+    """
+    db = SessionDB(tmp_path / "state.db")
+    db.create_session("shared", source="test")
+    stale_holder = f"pid={os.getpid()}:turn=stale"
+    next_holder = f"pid={os.getpid()}:turn=next"
+
+    assert db.try_acquire_session_turn_lease(
+        "shared", stale_holder, ttl_seconds=5
+    )
+    assert db.append_messages_batch(
+        "shared",
+        [{"role": "user", "content": "stale-owned"}],
+        turn_lease_holder=stale_holder,
+    ) == 1
+
+    db.release_session_turn_lease("shared", stale_holder)
+    assert db.try_acquire_session_turn_lease(
+        "shared", next_holder, ttl_seconds=5
+    )
+
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_messages_batch(
+            "shared",
+            [{"role": "assistant", "content": "late stale reply"}],
+            turn_lease_holder=stale_holder,
+        )
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_message(
+            "shared",
+            "assistant",
+            "late stale single-row",
+            turn_lease_holder=stale_holder,
+        )
+
+    assert db.append_messages_batch(
+        "shared",
+        [{"role": "assistant", "content": "next reply"}],
+        turn_lease_holder=next_holder,
+    ) == 1
+    assert [m["content"] for m in db.get_messages("shared")] == [
+        "stale-owned",
+        "next reply",
+    ]
+    db.release_session_turn_lease("shared", next_holder)
+
+
+def test_turn_lease_fences_flush_when_row_is_absent_or_expired(tmp_path):
+    db = SessionDB(tmp_path / "state.db")
+    db.create_session("shared", source="test")
+    holder = f"pid={os.getpid()}:turn=owner"
+
+    assert db.try_acquire_session_turn_lease("shared", holder, ttl_seconds=0.05)
+    time.sleep(0.12)
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_messages_batch(
+            "shared",
+            [{"role": "assistant", "content": "after ttl"}],
+            turn_lease_holder=holder,
+        )
+
+    db.release_session_turn_lease("shared", holder)
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_messages_batch(
+            "shared",
+            [{"role": "assistant", "content": "after release"}],
+            turn_lease_holder=holder,
+        )
+    assert db.get_messages("shared") == []
+
+
+def test_turn_lease_fence_walks_compression_child_to_root(tmp_path):
+    """A parent-key holder still fences writes against the rotated tip."""
+    db = SessionDB(tmp_path / "state.db")
+    db.create_session("root", source="test")
+    db.end_session("root", "compression")
+    db.create_session("child", source="test", parent_session_id="root")
+
+    root_holder = f"pid={os.getpid()}:turn=root"
+    stale_holder = f"pid={os.getpid()}:turn=stale"
+    assert db.try_acquire_session_turn_lease(
+        "root", root_holder, ttl_seconds=5
+    )
+    assert db.append_messages_batch(
+        "child",
+        [{"role": "user", "content": "owner on tip"}],
+        turn_lease_holder=root_holder,
+    ) == 1
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_messages_batch(
+            "child",
+            [{"role": "assistant", "content": "impostor"}],
+            turn_lease_holder=stale_holder,
+        )
+    db.release_session_turn_lease("child", root_holder)
+
+
+def test_lost_turn_lease_flush_fails_fast_without_patience_retry(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+):
+    """Sibling of test_a_lost_compression_lease_still_fails_fast.
+
+    SessionTurnLeaseLostError is permanent fencing, not a live-busy signal.
+    Retrying it would burn transcript write patience and still fail.
+    """
+    db = SessionDB(tmp_path / "state.db")
+    db.create_session("shared", source="test")
+    stale_holder = f"pid={os.getpid()}:turn=stale"
+    next_holder = f"pid={os.getpid()}:turn=next"
+    assert db.try_acquire_session_turn_lease(
+        "shared", stale_holder, ttl_seconds=5
+    )
+    db.release_session_turn_lease("shared", stale_holder)
+    assert db.try_acquire_session_turn_lease(
+        "shared", next_holder, ttl_seconds=5
+    )
+
+    sleeps = []
+    original = db._sleep_before_write_retry
+
+    def track_sleep(deadline, patience_s):
+        sleeps.append(patience_s)
+        return original(deadline, patience_s)
+
+    monkeypatch.setattr(db, "_sleep_before_write_retry", track_sleep)
+    monkeypatch.setattr(SessionDB, "_COMPRESSION_BUSY_WAIT_S", 5.0)
+
+    started = time.monotonic()
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_messages_batch(
+            "shared",
+            [{"role": "assistant", "content": "late stale reply"}],
+            turn_lease_holder=stale_holder,
+        )
+    assert time.monotonic() - started < 0.5
+    assert sleeps == []
+    assert db.get_messages("shared") == []
+    db.release_session_turn_lease("shared", next_holder)
+
+
+def test_turn_lease_fence_walks_continuation_that_inherited_fork_markers(tmp_path):
+    """Owner flush on a rotated tip must use the parent-key lease.
+
+    Presence-only ``_delegate_from`` / ``_branched_from`` detection would
+    treat the continuation as its own conversation. The presented parent
+    holder would then miss the row and fail-close a still-valid owner.
+    """
+    db = SessionDB(tmp_path / "state.db")
+    db.create_session("original-parent", source="test")
+    db.create_session(
+        "delegate",
+        source="delegate",
+        parent_session_id="original-parent",
+        model_config={"_delegate_from": "original-parent"},
+    )
+    db.end_session("delegate", "compression")
+    db.create_session(
+        "delegate-continuation",
+        source="delegate",
+        parent_session_id="delegate",
+        model_config={"_delegate_from": "original-parent"},
+    )
+    db.create_session(
+        "branch",
+        source="test",
+        parent_session_id="original-parent",
+        model_config={"_branched_from": "original-parent"},
+    )
+    db.end_session("branch", "compression")
+    db.create_session(
+        "branch-continuation",
+        source="test",
+        parent_session_id="branch",
+        model_config={"_branched_from": "original-parent"},
+    )
+
+    delegate_holder = f"pid={os.getpid()}:turn=delegate"
+    assert db.try_acquire_session_turn_lease(
+        "delegate", delegate_holder, ttl_seconds=5
+    )
+    assert db.append_messages_batch(
+        "delegate-continuation",
+        [{"role": "user", "content": "owner on inherited tip"}],
+        turn_lease_holder=delegate_holder,
+    ) == 1
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_messages_batch(
+            "delegate-continuation",
+            [{"role": "assistant", "content": "impostor"}],
+            turn_lease_holder=f"pid={os.getpid()}:turn=impostor",
+        )
+
+    branch_holder = f"pid={os.getpid()}:turn=branch"
+    assert db.try_acquire_session_turn_lease(
+        "branch", branch_holder, ttl_seconds=5
+    )
+    assert db.append_messages_batch(
+        "branch-continuation",
+        [{"role": "user", "content": "branch owner on inherited tip"}],
+        turn_lease_holder=branch_holder,
+    ) == 1
+    with pytest.raises(SessionTurnLeaseLostError, match="turn lease lost"):
+        db.append_messages_batch(
+            "branch-continuation",
+            [{"role": "assistant", "content": "branch impostor"}],
+            turn_lease_holder=f"pid={os.getpid()}:turn=branch-impostor",
+        )
+
+    assert [m["content"] for m in db.get_messages("delegate-continuation")] == [
+        "owner on inherited tip"
+    ]
+    assert [m["content"] for m in db.get_messages("branch-continuation")] == [
+        "branch owner on inherited tip"
+    ]
+    db.release_session_turn_lease("delegate-continuation", delegate_holder)
+    db.release_session_turn_lease("branch-continuation", branch_holder)