fix(sessions): ignore headless queued prompts and prefer the live resume binding

CHANGELOG.md

@@ -70,6 +70,15 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py`
   `working_directory` and saved no folder for those sessions. The renderer then
   failed on every run. The scanner now reads either name, and the open-sessions
   list shows a session even when its folder is unknown.
+- Stop the Slack daemon from following a headless run's copy of a Slack
+  message. The proofgate check's `claude -p` run records the warm session's
+  prompts as queued prompts with no entrypoint, so the daemon could bind to
+  that transcript and watch it for the reply. A queued prompt now counts only
+  in a transcript whose first entrypoint is interactive.
+- Match each terminal to its session through the `resume_binding`, which agent
+  hooks keep current, before the `restore_record` that cmux 0.65 lists first.
+  The restore record is a snapshot from when cmux restored the surface, so
+  `world-scanner.py` and `build-ws-context.py` could name an old session.
 - Stop two time-of-day test failures. The Observer audit tests failed in the
   last minute of a two-hour window, and five Strategist tests that count calls
   per UTC day failed in the last minutes of the day. Each test now starts its clock at the

bin/build-ws-context.py

@@ -396,7 +396,9 @@ def resume_binding_for_surface(surface_ref: str, ws_ref: str) -> dict | None:
         doc = json.loads(r.stdout)
     except Exception:
         return None
-    queue = [doc]
+    # Agent hooks keep resume_binding current; cmux 0.65's restore_record, listed
+    # first, is a snapshot from when cmux restored the surface.
+    queue = [doc.get("resume_binding"), doc] if isinstance(doc, dict) else [doc]
     binding = None
     while queue:
         value = queue.pop(0)

bin/comms_session.py

@@ -357,36 +357,47 @@ 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. Queued-prompt records carry no entrypoint, and a headless run
+    can write one before any record that does, so a queued prompt counts only
+    once the transcript's first entrypoint shows an interactive session.
 
     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:
         data = Path(path).read_bytes()
     except OSError:
         return False
+    queued = False
+    for rec in _records_containing(data, marker.encode()):
+        text = _prompt_text(rec)
+        if rec.get("entrypoint") == "sdk-cli" or not (text and marker in text):
+            continue
+        if rec.get("type") != "queue-operation":
+            return True
+        queued = True
+    return queued and _first_entrypoint(data) not in (None, "sdk-cli")
+
+
+def _records_containing(data: bytes, needle: bytes):
+    """Each JSON object record in `data` whose line contains `needle`."""
     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
+        try:
+            rec = json.loads(data[start:end].decode("utf-8", "replace"))
+        except json.JSONDecodeError:
+            rec = None
+        if isinstance(rec, dict):
+            yield rec
         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 _first_entrypoint(data: bytes) -> str | None:
+    return next((rec["entrypoint"] for rec in _records_containing(data, b'"entrypoint"')
+                 if isinstance(rec.get("entrypoint"), str)), None)
 
 
 def find_submission(project_dir: Path, marker: str, since: float) -> str | None:

bin/world-scanner.py

@@ -236,7 +236,9 @@ def surface_resume_binding(surface_ref, ws_ref):
         return {"unverified": True}
     if not isinstance(doc, dict):
         return {"unverified": True}
-    queue = [doc]
+    # Agent hooks keep resume_binding current; cmux 0.65's restore_record, listed
+    # first, is a snapshot from when cmux restored the surface.
+    queue = [doc.get("resume_binding"), doc]
     binding = None
     has_kind = False
     while queue:
@@ -257,8 +259,8 @@ def surface_resume_binding(surface_ref, ws_ref):
     return {
         "session_id": str(sid),
         "provider": "droid" if kind in {"factory", "droid"} else "claude",
-        # cmux 0.65 lists a restore_record, which names it working_directory,
-        # before the resume_binding.
+        # A restore_record, the fallback without a resume_binding, names it
+        # working_directory.
         "cwd": binding.get("cwd") or binding.get("working_directory"),
     }
 

tests/test_build_ws_context_in_process.py

@@ -425,6 +425,24 @@ def test_resume_binding_normalizes_factory_to_droid(self):
         self.assertEqual(binding["agent"], "droid")
         self.assertEqual(binding["session_id"][:8], "d0d01234")
 
+    def test_resume_binding_beats_a_stale_restore_record(self):
+        fake = mock.Mock(returncode=0, stdout=json.dumps({
+            "restore_record": {"kind": "claude",
+                               "checkpoint_id": "0ld00000-0000-4000-8000-000000000000"},
+            "resume_binding": {"kind": "claude",
+                               "checkpoint_id": "11ve0000-0000-4000-8000-000000000000"},
+        }))
+        with mock.patch.object(self.mod.subprocess, "run", return_value=fake):
+            binding = self.mod.resume_binding_for_surface(
+                "surface:21", "workspace:9")
+        self.assertEqual(binding["session_id"][:8], "11ve0000")
+
+    def test_resume_binding_none_for_a_non_object_reply(self):
+        fake = mock.Mock(returncode=0, stdout="[]")
+        with mock.patch.object(self.mod.subprocess, "run", return_value=fake):
+            self.assertIsNone(self.mod.resume_binding_for_surface(
+                "surface:21", "workspace:9"))
+
 
 class FindAgentPaneTests(unittest.TestCase):
     """The multi-pane trap: the agent is NOT always the focused/first pane.

tests/test_comms_submission.py

@@ -95,11 +95,37 @@ def test_submission_counts_text_blocks_and_queued_prompts(tmp_path):
         {"type": "user", "message": {"role": "user",
                                      "content": [{"type": "text", "text": f"hi {MARK}"}]}}])
     queued = _jsonl(tmp_path / "b.jsonl", [
+        {"type": "mode"},
+        {"type": "user", "entrypoint": "cli", "message": {"role": "user", "content": "boot"}},
         {"type": "queue-operation", "operation": "enqueue", "content": f"hi {MARK}"}])
     assert cs.transcript_has_submission(blocks, MARK)
     assert cs.transcript_has_submission(queued, MARK)
 
 
+def test_submission_ignores_a_headless_runs_queued_prompt(tmp_path):
+    """2026-10-19: 8 of 159 proofgate `claude -p` transcripts held a Slack
+    marker in a queue-operation record. Those records carry no entrypoint, and
+    26 of them came before the first sdk-cli record."""
+    queued = {"type": "queue-operation", "operation": "enqueue",
+              "content": f"You are a verification gate. Trace: {MARK}"}
+    before = _jsonl(tmp_path / "before.jsonl", [{"type": "atis-latch"}, queued])
+    after = _jsonl(tmp_path / "after.jsonl", [
+        {"type": "attachment", "entrypoint": "sdk-cli"}, queued])
+    assert not cs.transcript_has_submission(before, MARK)
+    assert not cs.transcript_has_submission(after, MARK)
+
+
+def test_first_entrypoint_skips_records_that_only_nest_the_key(tmp_path):
+    """Only a record's own top-level `entrypoint` counts."""
+    t = _jsonl(tmp_path / "t.jsonl", [
+        {"type": "attachment", "attachment": {"entrypoint": "cli"}},
+        {"type": "attachment", "entrypoint": ["cli"]},
+        {"type": "attachment", "entrypoint": "sdk-cli"},
+        {"type": "queue-operation", "operation": "enqueue", "content": f"hi {MARK}"}])
+    assert cs._first_entrypoint(t.read_bytes()) == "sdk-cli"
+    assert not cs.transcript_has_submission(t, MARK)
+
+
 def test_submission_ignores_tool_results_headless_runs_and_assistant_text(tmp_path):
     """The warm session greps logs that contain msg_ts markers, the proofgate
     Stop hook's headless run quotes its prompts, and the assistant may repeat

tests/test_world_scanner.py

@@ -475,17 +475,39 @@ def test_build_reads_cwd_from_the_cmux_065_restore_record(ws, monkeypatch):
     def run(args, **kwargs):
         if args[1:4] != ["surface", "resume", "show"]:
             return original_run(args, **kwargs)
-        # cmux 0.65 order: the restore_record comes before the resume_binding.
+        # A restored surface whose agent hooks haven't written a resume_binding yet.
         return fake_completed(json.dumps({
             "cleared": False, "pane_ref": "pane:1",
             "restore_record": {"checkpoint_id": sid, "kind": "claude",
                                "working_directory": "/work/repo"},
+        }))
+
+    monkeypatch.setattr(ws.subprocess, "run", run)
+    ws.build()
+    world = json.loads(ws.OUT_PATH.read_text())
+    assert [s["cwd"] for s in world["live_sessions"]] == ["/work/repo"]
+
+
+def test_build_prefers_the_resume_binding_over_a_stale_restore_record(ws, monkeypatch):
+    """Agent hooks keep resume_binding current; restore_record is a snapshot
+    from when cmux restored the surface."""
+    records, _, _, _ = _seed_identity_world(ws, monkeypatch, ("claude",))
+    sid = records[0]["session_id"]
+    original_run = ws.subprocess.run
+
+    def run(args, **kwargs):
+        if args[1:4] != ["surface", "resume", "show"]:
+            return original_run(args, **kwargs)
+        return fake_completed(json.dumps({
+            "restore_record": {"checkpoint_id": "0ld00000-0000-4000-8000-000000000000",
+                               "kind": "claude", "working_directory": "/old/repo"},
             "resume_binding": {"checkpoint_id": sid, "kind": "claude", "cwd": "/work/repo"},
         }))
 
     monkeypatch.setattr(ws.subprocess, "run", run)
     ws.build()
     world = json.loads(ws.OUT_PATH.read_text())
+    assert world["workspaces"][0]["surfaces"][0]["session_id"] == sid
     assert [s["cwd"] for s in world["live_sessions"]] == ["/work/repo"]