fix(tui_gateway): record a turn marker's writer and defer to it while alive (#94778)

A marker proved a turn had started, never that its writer had died, so a
second backend resuming the same session over a shared HERMES_HOME read a
sibling's live marker as crash evidence and started a duplicate turn over it.

record_turn_start now stamps writer_pid and, when readable, writer_start_time
(process create time, so a recycled pid cannot pass as the original writer)
and logs the identity — never the prompt. New marker_writer_state(entry)
answers alive/dead/unknown from active_sessions._pid_liveness, and
_maybe_schedule_auto_continue holds back when the writer is alive and is not
this process: no continuation, no "Resuming interrupted turn…" frame, and the
marker stays for its owner to clear.

Adds the two-backend regression: a real second process writes the marker and
keeps running while B resumes S (nothing scheduled, marker intact), then exits
and the same resume does schedule — so the live-writer gate is what defers B,
not some other switch.

tests/tui_gateway/test_auto_continue.py

@@ -13,20 +13,28 @@ time is positive proof the turn never finished. Contract pinned here:
 * ``_maybe_schedule_auto_continue`` re-submits a fresh interrupted prompt as
   a continuation note (display_kind ``auto_continue``), refuses stale /
   disabled / crash-looping / already-running cases, and bounds attempts via
-  the marker's attempt counter.
+  the marker's attempt counter;
+* a marker whose writer is still alive is ownership evidence, not crash
+  evidence: a second backend resuming the same session over one HERMES_HOME
+  schedules nothing and leaves the marker for its owner to clear (#94778).
 """
 
 from __future__ import annotations
 
+import os
+import subprocess
+import sys
 import threading
 import time
 import types
+from pathlib import Path
 
 import pytest
 
 from tui_gateway import server
 from tui_gateway.turn_marker import (
     clear_turn_marker,
+    marker_writer_state,
     read_turn_marker,
     record_turn_start,
 )
@@ -511,3 +519,107 @@ def test_failed_agent_build_leaves_marker_for_retry(
 
 # ── End to end: continuation runs a real turn and clears the marker ────
 
+
+# ── Marker writer identity: a live writer is an owner, not a corpse ────
+#
+# The scheduler used to treat every marker it found as proof the writing
+# process died. Two backends over one HERMES_HOME break that assumption: A is
+# mid-turn on session S while B resumes S — B read A's live marker as a crash
+# and started a second turn over it (#94778). Ownership now comes from the
+# writer's pid + create time.
+
+_CHILD_WRITER = """
+import os, sys, time
+from pathlib import Path
+from tui_gateway.turn_marker import record_turn_start
+
+home = Path(sys.argv[1])
+record_turn_start(home, "session-key", "interrupted prompt")
+print(f"ready {os.getpid()}", flush=True)
+# Linger until the test drops a sentinel. Self-exit, not terminate(): a venv
+# python.exe on Windows re-execs the real interpreter, so Popen.pid is the
+# launcher and killing it does not reliably reach the process that holds the
+# marker.
+deadline = time.time() + 120
+while not (home / "writer-exit").exists() and time.time() < deadline:
+    time.sleep(0.05)
+"""
+
+
+def _wait_for_marker(home, key, timeout=30.0):
+    """Bounded poll — the marker is written by another process, so a fixed
+    sleep in the parent would just be a flaky test."""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if read_turn_marker(home, key) is not None:
+            return
+        time.sleep(0.05)
+    raise AssertionError(f"child writer produced no marker within {timeout}s")
+
+
+def test_marker_writer_state_rejects_a_recycled_pid():
+    """Same pid, different process: the create time is what makes the pid an
+    identity. A start time from yesterday cannot be this process."""
+    assert marker_writer_state(
+        {"writer_pid": os.getpid(), "writer_start_time": time.time() - 86400}
+    ) != "alive"
+
+
+def test_second_backend_defers_to_a_live_marker_writer(emits, schedule_env, marker_home):
+    """Two backends, one HERMES_HOME, one session: A is mid-turn (alive writer)
+
+    B resumes S and must read the marker as ownership evidence, not crash
+    evidence — no continuation, no misleading "Resuming interrupted turn…"
+    frame, no duplicate turn, and A's marker left for A to clear. Once A is
+    really gone the same call does schedule, so the live-writer gate is what
+    held B back and not some other switch.
+    """
+    repo_root = Path(server.__file__).resolve().parents[1]
+    child = subprocess.Popen(
+        [sys.executable, "-c", _CHILD_WRITER, str(marker_home)],
+        cwd=str(repo_root),
+        env={**os.environ, "PYTHONPATH": str(repo_root)},
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        ready = child.stdout.readline().strip()
+        if not ready.startswith("ready "):
+            child.kill()
+            _, stderr = child.communicate(timeout=10)
+            raise AssertionError(f"child writer never started: {ready!r} / {stderr!r}")
+        writer_pid = int(ready.split()[1])
+        _wait_for_marker(marker_home, "session-key")
+        assert writer_pid != os.getpid()
+
+        written = read_turn_marker(marker_home, "session-key")
+        assert written["writer_pid"] == writer_pid
+        assert marker_writer_state(written) == "alive"
+
+        session = _session()
+        assert server._maybe_schedule_auto_continue("sid", session, "session-key") is None
+        assert not schedule_env  # nothing queued behind the live writer
+        assert session.get("_auto_continue_scheduled") is None  # not even claimed
+        assert not [e for e in emits if e[0] in ("status.update", "message.start")]
+        assert read_turn_marker(marker_home, "session-key") is not None  # A's marker intact
+
+        child.terminate()  # the launcher; the real writer exits on the sentinel below
+        child.wait(timeout=10)
+        (marker_home / "writer-exit").write_text("go", encoding="utf-8")
+        deadline = time.time() + 30
+        while time.time() < deadline and marker_writer_state(
+            read_turn_marker(marker_home, "session-key")
+        ) != "dead":
+            time.sleep(0.05)
+        assert marker_writer_state(read_turn_marker(marker_home, "session-key")) == "dead", (
+            f"writer pid {writer_pid} (child pid {child.pid}) still reads live"
+        )
+
+        assert server._maybe_schedule_auto_continue("sid", _session(), "session-key") is not None
+        assert len(schedule_env) == 1
+    finally:
+        (marker_home / "writer-exit").touch()  # release the child even on an early failure
+        if child.poll() is None:
+            child.kill()
+        child.wait(timeout=10)

tui_gateway/server.py

@@ -36,7 +36,7 @@ from agent.skill_commands import describe_skill_invocation  # noqa: F401
 from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX  # noqa: F401
 from tui_gateway import git_probe
 from tui_gateway._env import env_float, env_int
-from tui_gateway.turn_marker import clear_turn_marker, read_turn_marker, record_turn_start  # noqa: F401
+from tui_gateway.turn_marker import clear_turn_marker, marker_writer_state, read_turn_marker, record_turn_start  # noqa: F401
 from tui_gateway.contracts import registry as _contracts
 # User-facing copy shared with the split method modules (they close over this namespace).
 from tui_gateway.user_messages import (  # noqa: F401

tui_gateway/session_auto_continue.py

@@ -5,6 +5,7 @@ busy-submit handling. Bodies are rebound onto server.py's globals at install tim
 from __future__ import annotations
 
 import contextlib
+import os
 
 from .method_ctx import bind_module
 
@@ -67,6 +68,15 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
         return None
     if not marker.get("auto_continue", True):
         return None  # The mailbox owns recovery and receipt identity for imported turns.
+    # Ownership, not forensics: a sibling backend sharing this HERMES_HOME can be mid-turn on this very session, so
+    # its live marker says "someone is working on it", never "someone crashed". Leave the marker for its writer —
+    # clearing it would cancel the live turn's own account of itself. See #94778.
+    writer_state = marker_writer_state(marker)
+    if writer_state == "alive" and marker.get("writer_pid") != os.getpid():
+        logger.info("auto-continue for %s held back: marker writer pid %s is still alive (reader pid %s); the "
+                    "marker is ownership evidence and the owner clears it", session_key,
+                    marker.get("writer_pid"), os.getpid())
+        return None
     enabled, freshness_secs, max_attempts = _auto_continue_config()
     age = time.time() - marker["started_at"]
     if not enabled or age > freshness_secs or marker["attempts"] >= max_attempts:
@@ -124,7 +134,8 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
     if _start_session_work(kickoff, name=f"auto-continue-{sid}") is None:
         session["_auto_continue_scheduled"] = False
         return None
-    logger.info("auto-continue scheduled for session %s (attempt %d, interrupted %.0fs ago)", session_key, attempt, age)
+    logger.info("auto-continue scheduled for session %s (attempt %d, interrupted %.0fs ago, writer pid %s: %s, "
+                "reader pid %s)", session_key, attempt, age, marker.get("writer_pid"), writer_state, os.getpid())
     return {"attempt": attempt, "interrupted_at": marker["started_at"]}
 
 

tui_gateway/turn_marker.py

@@ -4,12 +4,14 @@ start and cleared on any conclusion — only a process death leaves one behind,
 reads it (``_maybe_schedule_auto_continue``). Stored per ``HERMES_HOME`` (profile-aware); writes prune
 entries older than ``_MAX_AGE_SECS`` and cap the count so a crash streak can't grow the file. Every
 function is best-effort — marker bookkeeping must never break a turn — so I/O errors degrade to "no
-marker" instead of raising."""
+marker" instead of raising. A marker also carries its writer's pid + start time (``marker_writer_state``): "a
+marker exists" only implies the writer died when no live sibling backend wrote it (#94778)."""
 
 from __future__ import annotations
 
 import json
 import logging
+import os
 import threading
 import time
 from pathlib import Path
@@ -34,6 +36,44 @@ def _started_at(entry: dict) -> float:
     return float(entry.get("started_at") or 0)
 
 
+def _writer_identity() -> dict:
+    """Best-effort identity of the process writing this marker. ``writer_pid`` alone already answers "who wrote
+    this, and are they still alive?" (``marker_writer_state``); ``writer_start_time`` pairs the pid with its create
+    time so a recycled pid cannot pass as the original writer. Identity is bookkeeping, never turn-critical, so
+    every failure degrades to a bare pid."""
+    identity = {"writer_pid": os.getpid()}
+    try:
+        from hermes_cli.active_sessions import _own_start_time
+        start = _own_start_time()
+        if start is not None:
+            identity["writer_start_time"] = float(start)
+    except Exception:
+        pass
+    return identity
+
+
+def marker_writer_state(entry: dict) -> str:
+    """``"alive"`` / ``"dead"`` / ``"unknown"``: is the process that wrote this marker still running?
+
+    A marker is durable proof a turn started — never proof its writer died. Two backends sharing one HERMES_HOME
+    break that assumption: A is mid-turn on session S while B resumes S, and B used to read A's marker as crash
+    evidence and start a second turn over it (#94778). Liveness comes from ``active_sessions._pid_liveness``
+    (pid + start time, so a reused pid reads dead), and "unknown" is the safe answer: it leaves the marker alone
+    without claiming its writer is gone.
+    """
+    if not isinstance(entry, dict):
+        return "unknown"
+    pid = entry.get("writer_pid")
+    if not isinstance(pid, int) or isinstance(pid, bool) or pid <= 0:
+        return "unknown"
+    try:
+        from hermes_cli.active_sessions import _pid_liveness
+        live = _pid_liveness(pid, entry.get("writer_start_time"))
+    except Exception:
+        return "unknown"
+    return "unknown" if live is None else ("alive" if live else "dead")
+
+
 def _load(path: Path) -> dict[str, dict]:
     try:
         with open(path, encoding="utf-8") as f:
@@ -80,9 +120,11 @@ def record_turn_start(home: Path | str, session_key: str, prompt: str, *, attemp
         return
     now = time.time()
     entry = {"attempts": max(0, int(attempts)), "prompt": prompt[:_MAX_PROMPT_CHARS], "started_at": now,
-             "auto_continue": bool(auto_continue)}
+             "auto_continue": bool(auto_continue), **_writer_identity()}
     if notification_category == "diagnostic":
         entry["notification_category"] = notification_category
+    # Identity only — never the prompt: this log is read on crash triage and must not carry turn content.
+    logger.debug("turn marker recorded for session %s by writer pid %s", session_key, entry["writer_pid"])
     _update(home, session_key, lambda entries: {**_prune(entries, now), session_key: entry}, "record")
 
 
@@ -104,6 +146,8 @@ def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | Non
             return None
         return {"attempts": max(0, int(entry.get("attempts") or 0)), "prompt": prompt, "started_at": _started_at(entry),
                 "auto_continue": bool(entry.get("auto_continue", True)),
+                # Writer identity when present: extra keys only, so a marker written by an older build still reads.
+                **{k: entry[k] for k in ("writer_pid", "writer_start_time") if entry.get(k) is not None},
                 **({"notification_category": "diagnostic"}
                    if entry.get("notification_category") == "diagnostic" else {})}
     except Exception: