fix(cli): surface unknown liveness in resolve/delete/archive (win32)

src/automator/cli.py

@@ -571,6 +571,25 @@ def cmd_resume(args: argparse.Namespace) -> int:
     except runs.RunRefError as e:
         print(str(e), file=sys.stderr)
         return 1
+    args.run_id = run_dir.name  # normalize so messages show the full id
+    # Gate here, NOT in _resume_paused_run: that helper is also resolve's re-arm
+    # path, which is already gated at resolve entry. A provably-live engine blocks
+    # outright (the TUI warns in its confirm modal instead; the CLI has no confirm
+    # step). 'unknown' warns but proceeds: resume is the recovery path that
+    # rewrites engine.pid, so it must stay usable when liveness is unverifiable.
+    live = runs.engine_liveness(run_dir)
+    if live == "alive":
+        print(
+            f"run {args.run_id} is still live — resuming would double-drive it; stop it first",
+            file=sys.stderr,
+        )
+        return 1
+    if live == "unknown":
+        print(
+            f"run {args.run_id}: engine may still be live (unverifiable pid) — "
+            "resuming could double-drive this run",
+            file=sys.stderr,
+        )
     return _resume_paused_run(project, run_dir)
 
 
@@ -600,9 +619,29 @@ def cmd_resolve(args: argparse.Namespace) -> int:
             file=sys.stderr,
         )
         return 1
-    if runs.engine_alive(run_dir):
+    # Not a cleanup path, so the "unknown must not block" invariant does not apply:
+    # an unverifiable-but-live pid must not be re-driven. A provably-live engine
+    # always blocks (--force never bypasses it); unknown blocks unless the operator
+    # vouches with --force — `stop` cannot verify or clear an unverifiable pid, so
+    # without an escape hatch a squatted pid would lock resolve out forever.
+    live = runs.engine_liveness(run_dir)
+    if live == "alive":
         print(f"run {args.run_id} is still live — stop it first", file=sys.stderr)
         return 1
+    if live == "unknown":
+        if not args.force:
+            print(
+                f"run {args.run_id}: engine may still be live (unverifiable pid) — "
+                "refusing to re-arm. Confirm the engine process is gone, then re-run "
+                "with --force (`stop` cannot verify or clear an unverifiable pid).",
+                file=sys.stderr,
+            )
+            return 1
+        print(
+            f"run {args.run_id}: engine may still be live (unverifiable pid) — "
+            "proceeding anyway (--force)",
+            file=sys.stderr,
+        )
     story_key = args.story or state.paused_story_key
     task = state.tasks.get(story_key) if story_key else None
     if story_key is None or task is None or task.phase != Phase.ESCALATED:
@@ -815,6 +854,27 @@ def cmd_stop(args: argparse.Namespace) -> int:
     return 0
 
 
+def _stop_or_block_live_engine(run_dir: Path, run_id: str, force: bool) -> int | None:
+    """Shared delete/archive guard. One liveness sample drives both the warning and the
+    block, so a mid-check identity flip can't fire one without the other. An unverifiable
+    pid (``unknown``) warns but never blocks cleanup; only a provably-live engine blocks,
+    or is stopped first under ``force``. Returns an exit code to propagate, or None to
+    proceed with cleanup."""
+    live = runs.engine_liveness(run_dir)
+    if live == "unknown":
+        print(f"run {run_id}: engine may still be live (unverifiable pid)", file=sys.stderr)
+    if live == "alive":
+        if not force:
+            print(f"run {run_id} is still live — stop it first (or pass --force)", file=sys.stderr)
+            return 1
+        try:
+            runs.stop_run(run_dir)
+        except (runs.StopRunError, ProcessHostError) as e:
+            print(str(e), file=sys.stderr)
+            return 1
+    return None
+
+
 def cmd_delete(args: argparse.Namespace) -> int:
     project = _project(args)
     try:
@@ -823,18 +883,9 @@ def cmd_delete(args: argparse.Namespace) -> int:
         print(str(e), file=sys.stderr)
         return 1
     args.run_id = run_dir.name
-    if runs.engine_alive(run_dir):
-        if not args.force:
-            print(
-                f"run {args.run_id} is still live — stop it first (or pass --force)",
-                file=sys.stderr,
-            )
-            return 1
-        try:
-            runs.stop_run(run_dir)
-        except (runs.StopRunError, ProcessHostError) as e:
-            print(str(e), file=sys.stderr)
-            return 1
+    rc = _stop_or_block_live_engine(run_dir, args.run_id, args.force)
+    if rc is not None:
+        return rc
     runs.delete_run(run_dir)
     print(f"run {args.run_id} deleted")
     return 0
@@ -848,18 +899,9 @@ def cmd_archive(args: argparse.Namespace) -> int:
         print(str(e), file=sys.stderr)
         return 1
     args.run_id = run_dir.name
-    if runs.engine_alive(run_dir):
-        if not args.force:
-            print(
-                f"run {args.run_id} is still live — stop it first (or pass --force)",
-                file=sys.stderr,
-            )
-            return 1
-        try:
-            runs.stop_run(run_dir)
-        except (runs.StopRunError, ProcessHostError) as e:
-            print(str(e), file=sys.stderr)
-            return 1
+    rc = _stop_or_block_live_engine(run_dir, args.run_id, args.force)
+    if rc is not None:
+        return rc
     dest = runs.archive_run(project, run_dir)
     print(f"run {args.run_id} archived to {dest}")
     return 0
@@ -869,20 +911,26 @@ def cmd_cleanup(args: argparse.Namespace) -> int:
     from .tui import launch  # pure stdlib; no textual import
 
     project = _project(args)
-    prunable, live = runs.prunable_sessions(project)
+    # one partition sample drives the prune and every message below, so the
+    # warnings and live count always match what was actually killed/skipped
+    killed, live, unknown = runs.prune_sessions(project, dry_run=args.dry_run)
+    for run_id in sorted(unknown):
+        # warn-only: unknown never blocks cleanup (same wording as delete/archive).
+        # Pruning kills the tmux session, never the engine pid, so the warning
+        # holds after the fact too.
+        print(f"run {run_id}: engine may still be live (unverifiable pid)", file=sys.stderr)
     if args.dry_run:
         windows = launch.prunable_ctl_windows(project)
-        if not prunable and not windows:
+        if not killed and not windows:
             print("nothing to clean up")
         else:
-            for run_id in prunable:
+            for run_id in killed:
                 print(f"would kill session bmad-auto-{run_id}")
             for name in windows:
                 print(f"would close ctl window {name}")
         if live:
             print(f"leaving {len(live)} live session(s) untouched")
         return 0
-    killed = runs.prune_sessions(project)
     windows = launch.prune_ctl_windows(project)
     print(f"removed {len(killed)} session(s), {len(windows)} ctl window(s)")
     if live:
@@ -948,6 +996,12 @@ def cmd_clean(args: argparse.Namespace) -> int:
     archived: list[str] = []
     deleted: list[str] = []
     for run_dir in reclaimable:
+        if runs.engine_liveness(run_dir) == "unknown":
+            # warn-only: unknown never blocks cleanup, but say so before removal
+            print(
+                f"run {run_dir.name}: engine may still be live (unverifiable pid)",
+                file=sys.stderr,
+            )
         # measure before mutating so the reclaim estimate holds for --dry-run too
         wt_dir = run_dir / "worktrees"
         wt_bytes = _dir_size(wt_dir) if wt_dir.is_dir() else 0
@@ -1251,6 +1305,12 @@ def add(name: str, func, help: str, *, aliases=()) -> argparse.ArgumentParser:
         help="--resume: re-arm + resume without prompting; --no-resume: re-arm only "
         "(default: prompt to confirm, then resume)",
     )
+    resolve_p.add_argument(
+        "--force",
+        action="store_true",
+        help="proceed when engine liveness is unverifiable (unknown); "
+        "a provably-live engine still blocks",
+    )
 
     decisions_p = add(
         "decisions",

src/automator/process_host.py

@@ -82,13 +82,11 @@ def alive_and_ours(self, pid: int, identity: float | None) -> bool:
         conflated again.
 
         Destructive paths use this strict check; non-destructive TUI reads use
-        :meth:`liveness_of` to preserve an ``'unknown'`` state."""
-        if pid <= 0:
-            return False
-        if identity is None:
-            return self.is_alive(pid)
-        # Gone, reused, or unreadable all read not-ours here.
-        return self.identity(pid) == identity
+        :meth:`liveness_of` to preserve an ``'unknown'`` state. Derived from
+        :meth:`liveness_of` — one decision table, two projections — so the binary
+        and tri-state probes can never drift: gone, reused, or unreadable
+        (``'unknown'``) all read not-ours here."""
+        return self.liveness_of(pid, identity) == "alive"
 
     def liveness_of(self, pid: int, identity: float | None) -> str:
         """Non-destructive tri-state read of *our* engine: ``'alive'`` |

src/automator/runs.py

@@ -166,6 +166,28 @@ def engine_alive(run_dir: Path) -> bool:
     return get_process_host().alive_and_ours(pid, identity)
 
 
+def engine_liveness(run_dir: Path) -> str:
+    """Tri-state read of the local engine: ``'alive'`` | ``'dead'`` | ``'unknown'``.
+    Wraps :meth:`ProcessHost.liveness_of` so a live-but-unreadable pid (win32
+    ``ERROR_ACCESS_DENIED``) reads ``'unknown'``, not a false ``'dead'``. No pid →
+    ``'dead'`` (the session fallback lives in the TUI layer)."""
+    pid, identity = read_pid_identity(run_dir)
+    if pid is None:
+        return "dead"
+    return probe_liveness(pid, identity)
+
+
+def probe_liveness(pid: int, identity: float | None) -> str:
+    """Tri-state probe of an already-read ``(pid, identity)`` — the shared body of
+    :func:`engine_liveness` and ``tui.data.liveness``, so both read the pid file once.
+    A probe failure degrades to ``'unknown'``, never a false ``'dead'``."""
+    host = get_process_host()  # ProcessHostError (misconfig) propagates, not masked as unknown
+    try:
+        return host.liveness_of(pid, identity)
+    except Exception:
+        return "unknown"
+
+
 # ----------------------------------------------------------- stop / delete / archive
 
 
@@ -204,15 +226,18 @@ def session_project_tags() -> dict[str, str]:
     return get_multiplexer().session_options(PROJECT_OPTION)
 
 
-def prunable_sessions(project: Path) -> tuple[list[str], list[str]]:
-    """Partition the bmad-auto-<id> agent sessions into (prunable, live) run ids.
+def prunable_sessions(project: Path) -> tuple[list[str], list[str], set[str]]:
+    """Partition the bmad-auto-<id> agent sessions into (prunable, live) run ids,
+    plus the subset of prunable ids whose engine liveness read 'unknown'
+    (unverifiable pid). Unknown never blocks cleanup — those sessions stay
+    prunable — but frontends surface a warning for them.
 
     The control session (bmad-auto-ctl) is never a candidate. Pruning is scoped
     to `project` via the PROJECT_OPTION tag set at session creation:
 
     - tag == this project: ours — prunable unless a provably-alive engine pid is
       running (covers finished/stopped/crashed *and* orphans whose run dir was
-      deleted, since engine_alive is False with no pid).
+      deleted, since engine_liveness reads 'dead' with no pid).
     - tag is another project: skipped — never touched.
     - tag empty (pre-upgrade, untagged session): can't prove ownership, so fall
       back to the run dir — prunable only when the dir exists under this project
@@ -222,6 +247,7 @@ def prunable_sessions(project: Path) -> tuple[list[str], list[str]]:
     mine = project_tag(project)
     prunable: list[str] = []
     live: list[str] = []
+    unknown: set[str] = set()
     for name in tmux_sessions():
         if name == CTL_SESSION or not name.startswith(_SESSION_PREFIX):
             continue
@@ -233,21 +259,29 @@ def prunable_sessions(project: Path) -> tuple[list[str], list[str]]:
                 continue  # another project's session
         elif not is_run(run_dir):
             continue  # untagged and no run dir here — ownership unprovable
-        if engine_alive(run_dir):
+        liveness = engine_liveness(run_dir)
+        if liveness == "alive":
             live.append(run_id)
-        else:
-            prunable.append(run_id)
-    return prunable, live
+            continue
+        prunable.append(run_id)
+        if liveness == "unknown":
+            unknown.add(run_id)
+    return prunable, live, unknown
 
 
-def prune_sessions(project: Path, *, dry_run: bool = False) -> list[str]:
+def prune_sessions(
+    project: Path, *, dry_run: bool = False
+) -> tuple[list[str], list[str], set[str]]:
     """Kill every prunable bmad-auto-<id> session (see prunable_sessions);
-    returns the run ids that were (or, with dry_run, would be) killed."""
-    prunable, _ = prunable_sessions(project)
+    returns (killed, live, unknown): the run ids that were (or, with dry_run,
+    would be) killed, the live ids skipped, and the killed subset whose engine
+    liveness read 'unknown'. All three come from the same partition sample, so
+    frontend messaging built from them always describes the performed actions."""
+    prunable, live, unknown = prunable_sessions(project)
     if not dry_run:
         for run_id in prunable:
             kill_session(run_id)
-    return prunable
+    return prunable, live, unknown
 
 
 def stop_run(run_dir: Path) -> bool:

src/automator/tui/app.py

@@ -555,8 +555,17 @@ def done(ok: bool | None) -> None:
 
     @work(thread=True, group="lifecycle")
     def _cleanup_sessions_worker(self) -> None:
-        killed = runs.prune_sessions(self.project)
+        # killed and unknown come from prune_sessions' single partition sample,
+        # so the warning below only ever names sessions that were actually pruned
+        killed, _live, unknown = runs.prune_sessions(self.project)
         windows = launch.prune_ctl_windows(self.project)
+        if unknown:
+            self.call_from_thread(
+                self.notify,
+                f"{len(unknown)} pruned session(s) had an unverifiable engine pid "
+                f"(may still be live): {', '.join(sorted(unknown))}",
+                severity="warning",
+            )
         self.call_from_thread(
             self.notify,
             f"removed {len(killed)} session(s), {len(windows)} window(s)",

src/automator/tui/data.py

@@ -30,8 +30,8 @@
 from ..gates import ATTENTION_FILE
 from ..journal import JOURNAL_FILE, LOGS_DIR, STATE_FILE, load_state
 from ..model import RunState
-from ..process_host import get_process_host
-from ..runs import list_run_dirs, read_pid_identity, session_name
+from ..process_host import ProcessHostError
+from ..runs import list_run_dirs, probe_liveness, read_pid_identity, session_name
 
 # Run statuses shown by the dashboard.
 RUNNING = "running"
@@ -73,11 +73,15 @@ def liveness(run_dir: Path) -> str:
     pid, identity = read_pid_identity(run_dir)
     if pid is None:
         return _session_liveness(run_dir.name)
+    # Probe the pid we just read (shared body with runs.engine_liveness) rather than
+    # re-reading it, so a non-atomic pid rewrite can't split the two reads and flash
+    # a false 'dead' between "pid present" here and a re-read seeing an empty file.
     try:
-        # non-destructive identity-aware tri-state probe
-        return get_process_host().liveness_of(pid, identity)
-    except Exception:
-        # never falsely dead — an unexpected probe failure stays 'unknown'
+        return probe_liveness(pid, identity)
+    except ProcessHostError:
+        # A misconfigured host (bad BMAD_AUTO_PROCESS_HOST) stays a hard error on
+        # CLI decision paths, but the display layer must degrade, not crash: the
+        # dashboard poll worker has no except and would take the whole app down.
         return "unknown"
 
 

src/automator/tui/launch.py

@@ -209,6 +209,9 @@ def _ctl_window_candidates(project: Path) -> list[tuple[str, str]]:
                 continue  # another project's window
         elif not runs.is_run(run_dir):
             continue  # untagged and no run dir here — ownership unprovable
+        # boolean gate on purpose: an 'unknown' engine stays a candidate (unknown
+        # never blocks cleanup) with no per-window warning — the session-level
+        # unknown warning from prunable_sessions covers the operator surface.
         if runs.engine_alive(run_dir):
             continue
         candidates.append((win_id, name))

tests/test_cleanup.py

@@ -203,6 +203,20 @@ def test_cmd_clean_dry_run_removes_nothing(project, capsys):
     assert "would remove worktree" in out
 
 
+def test_cmd_clean_warns_unknown_liveness(project, monkeypatch, capsys):
+    # warn-only: 'unknown' stays reclaimable (classification unchanged); the
+    # frontend re-probes just to say so before removal.
+    install_bmad_config(project)
+    repo = project.project
+    run_dir = repo / ".automator" / "runs" / "20260101-000000-aaaa"
+    save_state(run_dir, RunState(run_id="r", project=str(repo), started_at="x", stopped=True))
+    monkeypatch.setattr(runs, "engine_liveness", lambda _rd: "unknown")
+
+    assert cli.cmd_clean(_clean_args(repo)) == 0
+    err = capsys.readouterr().err
+    assert "run 20260101-000000-aaaa: engine may still be live (unverifiable pid)" in err
+
+
 def test_cmd_clean_reclaims_and_keeps_protected(project, capsys):
     install_bmad_config(project)
     repo = project.project