diff --git a/tests/tui_gateway/test_auto_continue.py b/tests/tui_gateway/test_auto_continue.py index 83b7905bc1..3b1698cbcb 100644 --- a/tests/tui_gateway/test_auto_continue.py +++ b/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) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 0acbc41820..a565df92c4 100644 --- a/tui_gateway/server.py +++ b/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 diff --git a/tui_gateway/session_auto_continue.py b/tui_gateway/session_auto_continue.py index 9590954652..05e88d97c7 100644 --- a/tui_gateway/session_auto_continue.py +++ b/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"]} diff --git a/tui_gateway/turn_marker.py b/tui_gateway/turn_marker.py index d23b92c4dd..cb3d96db0b 100644 --- a/tui_gateway/turn_marker.py +++ b/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: