Bring the warm Slack session up on Claude Code 2.1.303 and cmux 0.65

CHANGELOG.md

@@ -57,6 +57,13 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
   and reminders to finish older pending work before starting another task.
 
 ### Fixed
+- Bring the warm Slack session up again on Claude Code 2.1.303 and cmux 0.65.
+  Claude Code now writes several hundred KB of records right after each prompt,
+  so the check that read only the last 256 KB of the transcript never saw the
+  boot prompt. Every spawn was abandoned, and cmux 0.65 refused to close the
+  abandoned workspace, so each retry left another one open. The check now
+  searches the whole transcript, and the daemon retries the close with
+  `--force` when cmux asks to confirm, still only for its own warm workspace.
 - Stop the dashboard from crashing on sessions with no working folder. cmux
   0.65 lists a `restore_record` before the `resume_binding`, so
   `world-scanner.py` read the folder from a record that names it

bin/comms_session.py

@@ -301,9 +301,6 @@ def should_clear(transcript_path: str | Path,
 
 # --------------------------------------------------------------------------- submission (pure)
 
-# Only the transcript tail is read: a submitted prompt is always among the
-# newest records, and warm transcripts grow to megabytes.
-TRANSCRIPT_TAIL_BYTES = 262_144
 _RULE_RE = re.compile(r"^\s*─{8,}\s*$")
 _WS_RE = re.compile(r"\s+")
 
@@ -360,29 +357,38 @@ def transcript_has_submission(path: str | Path, marker: str) -> bool:
 
     Headless `claude -p` transcripts never count: the proofgate Stop hook runs
     one in this cwd after every warm turn, and it quotes the warm session's
-    prompts back."""
+    prompts back.
+
+    The whole file is searched. Claude Code writes several hundred KB of
+    records right after each prompt, so a prompt can be far from the end
+    within a second of being submitted."""
+    needle = marker.encode()
     try:
-        with open(path, "rb") as f:
-            f.seek(0, os.SEEK_END)
-            f.seek(max(0, f.tell() - TRANSCRIPT_TAIL_BYTES))
-            tail = f.read().decode("utf-8", "replace")
+        data = Path(path).read_bytes()
     except OSError:
         return False
-    for line in tail.splitlines():
-        if marker not in line:
-            continue
-        try:
-            rec = json.loads(line)
-        except json.JSONDecodeError:
-            continue
-        if not isinstance(rec, dict) or rec.get("entrypoint") == "sdk-cli":
-            continue
-        text = _prompt_text(rec)
-        if text and marker in text:
+    hit = data.find(needle)
+    while hit != -1:
+        start = data.rfind(b"\n", 0, hit) + 1
+        end = data.find(b"\n", hit)
+        end = len(data) if end == -1 else end
+        if _is_submission(data[start:end], marker):
             return True
+        hit = data.find(needle, end)
     return False
 
 
+def _is_submission(line: bytes, marker: str) -> bool:
+    try:
+        rec = json.loads(line.decode("utf-8", "replace"))
+    except json.JSONDecodeError:
+        return False
+    if not isinstance(rec, dict) or rec.get("entrypoint") == "sdk-cli":
+        return False
+    text = _prompt_text(rec)
+    return bool(text and marker in text)
+
+
 def find_submission(project_dir: Path, marker: str, since: float) -> str | None:
     """Path of the newest transcript in `project_dir` changed at or after
     `since` that records a prompt containing `marker`, or None.
@@ -555,8 +561,11 @@ def close_own_workspace(paths: comms_lib.Paths, ws_ref: str, log=lambda m: None)
     if not is_warm:
         log(f"skip close {ws_ref}: not a '{SESSION_TITLE}' workspace (ref reissued?)")
         return
-    rc, _, err = comms_lib.run_cmd(
-        [str(paths.cmux_bin), "close-workspace", "--workspace", ws_ref], timeout=15)
+    argv = [str(paths.cmux_bin), "close-workspace", "--workspace", ws_ref]
+    rc, _, err = comms_lib.run_cmd(argv, timeout=15)
+    if rc != 0 and "confirmation_required" in err:
+        # cmux 0.65 won't close a workspace with a running process unless forced.
+        rc, _, err = comms_lib.run_cmd([*argv, "--force"], timeout=15)
     log(f"closed prior warm workspace {ws_ref}" if rc == 0
         else f"close {ws_ref} rc={rc}: {err.strip()[:120]}")
 

tests/test_comms_submission.py

@@ -124,14 +124,48 @@ def test_submission_skips_corrupt_lines_and_missing_files(tmp_path):
     assert not cs.transcript_has_submission(tmp_path / "missing.jsonl", MARK)
 
 
-def test_submission_reads_only_the_tail(tmp_path, monkeypatch):
-    monkeypatch.setattr(cs, "TRANSCRIPT_TAIL_BYTES", 200)
-    rec = {"type": "user", "message": {"role": "user", "content": f"old {MARK}"}}
+def test_submission_found_behind_the_records_claude_writes_after_it(tmp_path):
+    """2026-10-19: Claude Code 2.1.303 wrote ~516 KB of listings and request
+    records within 0.7 s of the boot prompt, so a check that read only the
+    last 256 KB never saw it and every warm spawn was abandoned."""
+    t = _jsonl(tmp_path / "t.jsonl", [
+        {"type": "mode"},
+        {"type": "user", "entrypoint": "cli",
+         "message": {"role": "user", "content": f"Read the prompt. {MARK}"}},
+        *({"type": "attachment", "attachment": {"type": kind, "content": "x" * 200_000}}
+          for kind in ("agent_listing_delta", "skill_listing", "instructions", "prompt_snapshot")),
+    ])
+    assert t.stat().st_size > 800_000
+    assert cs.transcript_has_submission(t, MARK)
+
+
+def test_submission_keeps_searching_past_a_non_prompt_match(tmp_path):
+    t = _jsonl(tmp_path / "t.jsonl", [
+        {"type": "assistant", "message": {"role": "assistant", "content": MARK}},
+        {"type": "user", "message": {"role": "user", "content": f"hi {MARK}"}},
+    ])
+    assert cs.transcript_has_submission(t, MARK)
+
+
+def test_submission_found_after_a_marker_line_that_is_not_utf8(tmp_path):
+    t = tmp_path / "t.jsonl"
+    good = json.dumps({"type": "user", "message": {"role": "user", "content": f"hi {MARK}"}})
+    t.write_bytes(f'{{"type": "user", "x": "{MARK}'.encode() + b"\xff\xfe\"}\n" + good.encode() + b"\n")
+    assert cs.transcript_has_submission(t, MARK)
+
+
+def test_submission_false_for_an_empty_transcript(tmp_path):
     t = tmp_path / "t.jsonl"
-    t.write_text(json.dumps(rec) + "\n" + ("x" * 400) + "\n")
+    t.write_text("")
     assert not cs.transcript_has_submission(t, MARK)
 
 
+def test_submission_matches_a_last_line_without_a_newline(tmp_path):
+    t = tmp_path / "t.jsonl"
+    t.write_text(json.dumps({"type": "user", "message": {"role": "user", "content": MARK}}))
+    assert cs.transcript_has_submission(t, MARK)
+
+
 def test_submission_droid_schema(tmp_path):
     t = _jsonl(tmp_path / "t.jsonl", [
         {"type": "message", "message": {"role": "user", "content": f"go {MARK}"}}])
@@ -430,3 +464,38 @@ def fake_run(argv, timeout=30):
     assert closed == [] and "skip close workspace:25" in logs[0]
     cs.close_own_workspace(PATHS, "workspace:258", log=logs.append)
     assert closed == ["workspace:258"]
+
+
+def _close_with(monkeypatch, close_results):
+    """Run close_own_workspace on a listed warm workspace whose close attempts
+    return `close_results` in turn; returns (close argvs, log lines)."""
+    calls: list = []
+    results = iter(close_results)
+
+    def fake_run(argv, timeout=30):
+        if argv[1] == "list-workspaces":
+            return 0, "  workspace:1000010024  assistant-comms (warm) #ea290b [2]\n", ""
+        calls.append(argv[1:])
+        return next(results)
+
+    monkeypatch.setattr(cl, "run_cmd", fake_run)
+    logs: list = []
+    cs.close_own_workspace(PATHS, "workspace:1000010024", log=logs.append)
+    return calls, logs
+
+
+def test_close_own_workspace_forces_when_cmux_asks_to_confirm(monkeypatch):
+    """cmux 0.65 refuses to close a workspace with a running process unless
+    forced, which left every failed warm spawn open (2026-10-19)."""
+    refused = (1, "", "Error: confirmation_required: Workspace has a running process; "
+                      "retry with --force")
+    calls, logs = _close_with(monkeypatch, [refused, (0, "OK workspace:1000010024", "")])
+    assert calls == [["close-workspace", "--workspace", "workspace:1000010024"],
+                     ["close-workspace", "--workspace", "workspace:1000010024", "--force"]]
+    assert logs == ["closed prior warm workspace workspace:1000010024"]
+
+
+def test_close_own_workspace_never_forces_past_other_errors(monkeypatch):
+    calls, logs = _close_with(monkeypatch, [(1, "", "Error: not_found")])
+    assert calls == [["close-workspace", "--workspace", "workspace:1000010024"]]
+    assert logs == ["close workspace:1000010024 rc=1: Error: not_found"]

tests/test_no_close_workspace.py

@@ -40,6 +40,9 @@ class TestNoCloseWorkspace(unittest.TestCase):
     # invoke the action.
     FORBIDDEN_INVOCATIONS = [
         re.compile(r'cmux\s+close-workspace\b'),                     # bare command
+        re.compile(r'cmux\s+workspace\s+close\b'),                   # cmux 0.65 canonical form
+        re.compile(r'"workspace"\s*,\s*"close"'),
+        re.compile(r"'workspace'\s*,\s*'close'"),
         re.compile(r'"close-workspace"\s*,?\s*"--workspace"'),        # subprocess argv
         re.compile(r"'close-workspace'\s*,?\s*'--workspace'"),
     ]
@@ -53,7 +56,7 @@ class TestNoCloseWorkspace(unittest.TestCase):
     # close-workspace call (even elsewhere in the same file) still fails.
     ALLOWED_INVOCATION_SUBSTRINGS = {
         'bin/comms_session.py': (
-            '[str(paths.cmux_bin), "close-workspace", "--workspace", ws_ref]',
+            'argv = [str(paths.cmux_bin), "close-workspace", "--workspace", ws_ref]',
         ),
     }
 
@@ -127,7 +130,7 @@ def test_allowlist_is_exact_not_a_file_wide_pass(self):
         self.assertFalse(self._is_allowed(p, rogue),
                          "a non-guarded close-workspace line must NOT be allowed")
         # The real guarded line IS allowed.
-        guarded = '[str(paths.cmux_bin), "close-workspace", "--workspace", ws_ref]'
+        guarded = 'argv = [str(paths.cmux_bin), "close-workspace", "--workspace", ws_ref]'
         self.assertTrue(self._is_allowed(p, guarded))
         # And the exception does not leak to other files.
         self.assertFalse(self._is_allowed(Path("/repo/bin/pulse.py"), guarded))