Show unmanaged Codex panes in acw status table

Panes running Codex but not managed by acw now appear in the status
table as "unmanaged" (dimmed), making it easy to see all running agents
and identify candidates for `acw start`.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: thomasahle

bin/acw_runtime.py

@@ -574,6 +574,42 @@ def build_status_entries_from_snapshot(
             )
         )
 
+    for pane in snapshot.panes:
+        if pane.pane_id in watched_panes:
+            continue
+        if not pane.has_codex_process:
+            continue
+        if pane.detected_thread and context.is_thread_id(pane.detected_thread):
+            if runtime_session_for_thread(model, pane.detected_thread):
+                continue  # already covered above as unwatched_pane
+        window_name = pane.window_label or pane.window_name or ""
+        entries.append(
+            (
+                {
+                    "entity_kind": "unmanaged_pane",
+                    "runtime_id": "",
+                    "pid": "",
+                    "cwd": "",
+                    "pane": pane.pane_id,
+                    "thread": pane.detected_thread or "",
+                    "name": window_name,
+                    "session_name": "",
+                    "state": "",
+                    "watch": "",
+                    "msg_inline": "",
+                    "paused": "",
+                    "runtime_status": "",
+                    "runtime_detail": "",
+                    "generation": "",
+                    "worker_label": "",
+                    "anomalies": (),
+                    "anomaly_detail": "",
+                },
+                pane.pane_id,
+                pane.window_label or pane.window_name or "",
+            )
+        )
+
     for session in snapshot.sessions:
         if session.thread_id in claimed_session_threads:
             continue

bin/acw_status.py

@@ -18,6 +18,7 @@
     "warn": "dark_orange",
     "error": "bold red",
     "dead": "dim",
+    "unmanaged": "dim",
 }
 
 
@@ -142,6 +143,8 @@ def compute_state(
 ) -> str:
     if row.get("entity_kind") == "dead_session":
         return "dead"
+    if row.get("entity_kind") == "unmanaged_pane":
+        return "unmanaged"
     if row.get("anomalies"):
         return "error"
     runtime_status = row.get("runtime_status", "")

test/support/manager_scenarios.py

@@ -250,8 +250,8 @@ def test_manager_stop_forgets_managed_session(self):
         self.assertIsNone(self.harness.manager_session_row(first_turn.thread_id), self.harness.diagnostics())
 
         status = self.harness.run_manager("status")
-        self.assertNotIn("stopcase", status.stdout, self.harness.diagnostics())
-        self.assertNotIn(first_turn.thread_id, status.stdout, self.harness.diagnostics())
+        # After stop, the pane still runs Codex so it shows as unmanaged
+        self.assertIn("unmanaged", status.stdout, self.harness.diagnostics())
 
     def test_manager_stop_without_target_forgets_all_managed_sessions(self):
         self.harness.rename_window("formal")
@@ -289,8 +289,8 @@ def test_manager_stop_without_target_forgets_all_managed_sessions(self):
         self.assertIsNone(self.harness.manager_runtime_row(aot_turn.thread_id), self.harness.diagnostics())
 
         status = self.harness.run_manager("status")
-        self.assertNotIn("formal", status.stdout, self.harness.diagnostics())
-        self.assertNotIn("aot", status.stdout, self.harness.diagnostics())
+        # After stop, panes still run Codex so they show as unmanaged
+        self.assertIn("unmanaged", status.stdout, self.harness.diagnostics())
 
     def test_manager_status_details_reads_live_sqlite_state_ref(self):
         self.harness.rename_window("details")

test/test_watchd_unit.py

@@ -208,6 +208,7 @@ def test_main_help_prints_command_summary(self):
         self.assertIn("Usage:", text)
         self.assertIn("Commands:", text)
         self.assertIn("doctor", text)
+        self.assertIn("recover", text)
         self.assertIn("Examples:", text)
         self.assertEqual("", err.getvalue())
 
@@ -895,7 +896,7 @@ def test_status_summary_includes_duplicate_live_watchers_hidden_by_thread_merge(
         self.assertIn("0:6:formal/%6", text)
         self.assertIn("0:10:aot2/%10", text)
         self.assertIn("Invariant error: thread 11111111…1111 has managed runtimes on panes %10, %6", text)
-        self.assertIn("Hint: run acw repair --dry-run for live invariant fixes.", text)
+        self.assertIn("Hint: run acw repair for live invariant fixes.", text)
 
     def test_status_summary_flags_duplicate_agent_name_on_multiple_threads(self):
         snapshot = _snapshot(
@@ -985,7 +986,7 @@ def test_status_summary_prints_doctor_recommendation_for_warn(self):
         text = out.getvalue()
         self.assertIn("Recommendation: run acw doctor %7", text)
 
-    def test_status_summary_prints_cleanup_hint_for_dead_sessions(self):
+    def test_status_summary_prints_recover_and_cleanup_hint_for_dead_sessions(self):
         snapshot = _snapshot(sessions=(_session_record(THREAD, name="formal"),))
         with patch.object(acw, "_build_runtime_snapshot", return_value=snapshot):
             with patch.object(acw, "_read_state_json", return_value={"health": "ok"}):
@@ -995,7 +996,10 @@ def test_status_summary_prints_cleanup_hint_for_dead_sessions(self):
                             out = io.StringIO()
                             with redirect_stdout(out):
                                 acw.cmd_status([])
-        self.assertIn("Hint: run acw cleanup to remove dead sessions.", out.getvalue())
+        self.assertIn(
+            "Hint: run acw recover to restart dead sessions, or acw cleanup to remove them.",
+            out.getvalue(),
+        )
 
     def test_status_summary_prefers_repair_for_pane_local_invariant_errors(self):
         snapshot = _snapshot(
@@ -1903,7 +1907,7 @@ def test_doctor_report_errors_when_live_pane_thread_cannot_be_detected(self):
             rendered,
         )
 
-    def test_status_summary_skips_unmanaged_unresolved_live_codex_pane(self):
+    def test_status_summary_shows_unmanaged_unresolved_live_codex_pane(self):
         snapshot = _snapshot(
             panes=(_pane_record(
                 "%7",
@@ -1923,8 +1927,8 @@ def test_status_summary_skips_unmanaged_unresolved_live_codex_pane(self):
                             with redirect_stdout(out):
                                 acw.cmd_status([])
         text = out.getvalue()
-        self.assertIn("Sessions: 0", text)
-        self.assertIn("(none)", text)
+        self.assertIn("Sessions: 1", text)
+        self.assertIn("unmanaged", text)
 
     def test_doctor_plain_output_shows_recommended_command(self):
         report = acw.DoctorReport(
@@ -2210,18 +2214,21 @@ def test_status_target_matches_saved_session_name_without_live_pane(self):
         self.assertIn("Sessions: 1", text)
         self.assertIn("formal", text)
 
-    def test_status_skips_unmanaged_live_codex_pane_without_saved_session(self):
+    def test_status_shows_unmanaged_live_codex_pane_without_saved_session(self):
         snapshot = _snapshot(
-            panes=(_pane_record("%9", THREAD, window_label="0:9:acw", window_name="acw"),),
+            panes=(_pane_record("%9", THREAD, window_label="0:9:acw", window_name="acw", has_codex_process=True),),
         )
         with patch.object(acw, "_build_runtime_snapshot", return_value=snapshot):
-            out = io.StringIO()
-            with redirect_stdout(out):
-                acw.cmd_status([])
+            with patch.object(acw, "_read_state_json", return_value={}):
+                with patch.object(acw, "_thread_times", return_value=("-", "-")):
+                    with patch.object(acw, "_last_agent_snippet_for_pane", return_value=""):
+                        with patch.object(acw, "_is_pid_stopped", return_value=False):
+                            out = io.StringIO()
+                            with redirect_stdout(out):
+                                acw.cmd_status([])
         text = out.getvalue()
-        self.assertIn("Sessions: 0", text)
-        self.assertIn("(none)", text)
-        self.assertNotIn("acw/%9", text)
+        self.assertIn("Sessions: 1", text)
+        self.assertIn("unmanaged", text)
 
     def test_status_shows_unwatched_live_codex_pane_when_saved_session_exists(self):
         snapshot = _snapshot(
@@ -2621,6 +2628,139 @@ def test_restart_target_with_restart_codex_fails_when_thread_unknown(self):
         stop_watchers.assert_not_called()
         self.assertIn("could not determine thread_id for pane=%7", err.getvalue())
 
+    def test_recover_target_restarts_saved_dead_session_on_saved_pane(self):
+        model = acw._build_runtime_model(_snapshot(
+            panes=(_pane_record(
+                "%50",
+                "",
+                window_label="0:50:bugs",
+                window_name="bugs",
+                has_codex_process=False,
+            ),),
+            sessions=(_session_record(THREAD, name="bugs", pane="%50", message="continue through docs"),),
+        ))
+
+        with patch.object(acw, "_build_runtime_model", return_value=model):
+            with patch.object(acw, "_daemon_rpc_restart_codex") as restart_codex:
+                with patch.object(
+                    acw,
+                    "_daemon_rpc_upsert_session",
+                    return_value={
+                        "thread": THREAD,
+                        "pane": "%50",
+                        "runtime_status": "running",
+                        "runtime_id": "runtime-9999",
+                        "daemon_pid": "9999",
+                    },
+                ) as upsert:
+                    out = io.StringIO()
+                    with redirect_stdout(out), redirect_stderr(io.StringIO()):
+                        acw.cmd_recover(["bugs"])
+
+        restart_codex.assert_called_once_with("%50", THREAD)
+        upsert.assert_called_once_with(
+            THREAD,
+            pane="%50",
+            name="bugs",
+            message="continue through docs",
+            paused=False,
+            generation=1,
+        )
+        text = out.getvalue()
+        self.assertIn("recover: preparing pane=%50", text)
+        self.assertIn(f"recover: relaunching Codex in pane=%50 thread_id={THREAD}", text)
+        self.assertIn("recover: starting runtime for pane=%50", text)
+        self.assertIn(f"recovered: pane=%50 thread_id={THREAD} daemon_pid=9999", text)
+
+    def test_recover_target_fails_without_saved_pane(self):
+        model = acw._build_runtime_model(_snapshot(
+            sessions=(_session_record(THREAD, name="bugs", message="continue through docs"),),
+        ))
+
+        err = io.StringIO()
+        with patch.object(acw, "_build_runtime_model", return_value=model):
+            with redirect_stderr(err):
+                with self.assertRaises(SystemExit) as ctx:
+                    acw.cmd_recover(["bugs"])
+
+        self.assertEqual(1, ctx.exception.code)
+        self.assertIn("has no saved pane", err.getvalue())
+
+    def test_recover_target_fails_when_saved_pane_runs_other_thread(self):
+        model = acw._build_runtime_model(_snapshot(
+            panes=(_pane_record("%50", OTHER_THREAD, window_label="0:50:bugs", window_name="bugs"),),
+            sessions=(_session_record(THREAD, name="bugs", pane="%50", message="continue through docs"),),
+        ))
+
+        err = io.StringIO()
+        with patch.object(acw, "_build_runtime_model", return_value=model):
+            with redirect_stderr(err):
+                with self.assertRaises(SystemExit) as ctx:
+                    acw.cmd_recover(["bugs"])
+
+        self.assertEqual(1, ctx.exception.code)
+        self.assertIn("is already running thread", err.getvalue())
+        self.assertIn("use acw restart or acw repair", err.getvalue())
+
+    def test_recover_without_target_recovers_both_sessions_including_missing_pane(self):
+        model = acw._build_runtime_model(_snapshot(
+            panes=(_pane_record(
+                "%50",
+                "",
+                window_label="0:50:bugs",
+                window_name="bugs",
+                has_codex_process=False,
+            ),),
+            sessions=(
+                _session_record(THREAD, name="bugs", pane="%50", message="continue through docs"),
+                _session_record(OTHER_THREAD, name="aot", pane="%60", message="continue aot"),
+            ),
+        ))
+
+        with patch.object(acw, "_build_runtime_model", return_value=model):
+            with patch.object(acw, "_daemon_rpc_restart_codex") as restart_codex:
+                with patch.object(
+                    acw,
+                    "_daemon_rpc_upsert_session",
+                    return_value={
+                        "thread": THREAD,
+                        "pane": "%50",
+                        "runtime_status": "running",
+                        "runtime_id": "runtime-9999",
+                        "daemon_pid": "9999",
+                    },
+                ) as upsert:
+                    with patch.object(acw, "run_tmux", return_value="%58\n"):
+                        with patch.object(acw, "_read_session_state", return_value={"name": "aot"}):
+                            with patch.object(acw, "_write_session_state"):
+                                out = io.StringIO()
+                                err = io.StringIO()
+                                with redirect_stdout(out), redirect_stderr(err):
+                                    acw.cmd_recover([])
+
+        self.assertEqual(restart_codex.call_count, 2)
+        self.assertEqual(upsert.call_count, 2)
+
+    def test_recover_without_target_skips_sessions_with_managed_runtime_state_on_saved_pane(self):
+        model = acw._build_runtime_model(_snapshot(
+            watchers=(_watcher_record("%60", OTHER_THREAD, pid="2222"),),
+            sessions=(
+                _session_record(THREAD, name="bugs", pane="%60", message="continue through docs"),
+            ),
+        ))
+
+        with patch.object(acw, "_build_runtime_model", return_value=model):
+            with patch.object(acw, "_daemon_rpc_restart_codex") as restart_codex:
+                out = io.StringIO()
+                err = io.StringIO()
+                with redirect_stdout(out), redirect_stderr(err):
+                    acw.cmd_recover([])
+
+        restart_codex.assert_not_called()
+        self.assertEqual("", err.getvalue())
+        self.assertIn("recover: skipped 1 session(s) that are not auto-recoverable", out.getvalue())
+        self.assertIn("managed runtime state", out.getvalue())
+
     def test_codex_top_level_descendants_returns_direct_child_of_shell(self):
         graph = (
             {