diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e02e52..3664c0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,17 @@ The version is carried in `pyproject.toml` and `src/assistant/__init__.py` and reminders to finish older pending work before starting another task. ### Fixed +- Ping about an idle workspace only when it needs you: the agent's last lines + ask you something, a permission or plan prompt is up, or the session stopped + on an API error. Claude's idle alert fires about a minute after every turn, + and about 75% of recent turns ended with a status update instead. +- Match the watcher's patterns ("CI red", "PR opened", "tests passed") against + what the agent said this turn, not 50 lines of scrollback, so old tool output + no longer triggers them. Name workspaces cmux left as "Terminal" by Claude's own + session title, or the folder and branch. +- Mute the "stranded", "awaiting-review", and "emit-card" word lists by default + (168 pings in two weeks, mostly on status prose), and stop a muted pattern + from hiding a real one in the same message. - Stop closing a healthy warm comms session whenever cmux is slow to answer. A refused or timed-out check now leaves the session alone; only cmux saying the workspace doesn't exist, or a successful workspace list without it, diff --git a/bin/cmux-watcher.py b/bin/cmux-watcher.py index e04eb3c..360e5cd 100644 --- a/bin/cmux-watcher.py +++ b/bin/cmux-watcher.py @@ -98,9 +98,14 @@ "patterns": [ {"id": "pr-opened", "regex": r"PR #\d+ (opened|created|shipped)", "signal": "work_complete", "priority": "high"}, + # The three needs-input word lists below matched status prose far more + # than real asks ("Awaiting the adversary, code-review…", "blocked" in a + # PR review): 168 pings in two weeks. A real ask now arrives as an + # AskUserQuestion or an idle alert whose last line asks something, and + # the pulse's Observer handles stalled workspaces itself. {"id": "awaiting-review", "regex": r"(awaiting|waiting for|needs).{0,30}(review|approval|your input)", - "signal": "needs_input", "priority": "high"}, + "signal": "needs_input", "priority": "high", "suppress": True}, {"id": "ci-green", "regex": r"(all CI (checks )?green|CI (is )?green|✓.*CI|Jenkins.*SUCCESS)", "signal": "work_complete", "priority": "medium", "suppress": True}, @@ -108,9 +113,9 @@ "regex": r"(CI (is )?red|CI fail|Jenkins.*FAIL|OURS.*failure)", "signal": "needs_input", "priority": "high"}, {"id": "emit-card", "regex": r"(needs_user|emit.card|awaiting your)", - "signal": "needs_input", "priority": "high"}, + "signal": "needs_input", "priority": "high", "suppress": True}, {"id": "stranded", "regex": r"(stranded|stuck|blocked|timed out|API error)", - "signal": "needs_input", "priority": "medium"}, + "signal": "needs_input", "priority": "medium", "suppress": True}, {"id": "tests-pass", "regex": r"\d+ (tests?|specs?) (passing|passed|green)", "signal": "work_complete", "priority": "low"}, {"id": "committed", "regex": r"\[main [a-f0-9]{7}\]", @@ -452,9 +457,30 @@ def pending_question(records: list[dict]) -> str | None: return None +def _is_prompt(rec: dict) -> bool: + """A user turn someone typed, as opposed to a tool result or hook output.""" + if agent_session.record_role(rec) != "user" or rec.get("isMeta"): + return False + msg = rec.get("message") + content = msg.get("content") if isinstance(msg, dict) else None + if isinstance(content, str): + return not content.startswith("<") + blocks = _content_blocks(rec) + return bool(blocks) and not any(b.get("type") == "tool_result" for b in blocks) + + +def current_turn(records: list[dict]) -> list[dict]: + """The records after the newest typed prompt: this turn only, so an earlier + turn's words can't be judged again.""" + for i in range(len(records) - 1, -1, -1): + if _is_prompt(records[i]): + return records[i + 1:] + return records + + def last_assistant_text(records: list[dict]) -> str | None: - """The newest non-empty text block the agent wrote.""" - for rec in reversed(records): + """The newest non-empty text block the agent wrote in the current turn.""" + for rec in reversed(current_turn(records)): if agent_session.record_role(rec) != "assistant": continue for block in reversed(_content_blocks(rec)): @@ -464,6 +490,31 @@ def last_assistant_text(records: list[dict]) -> str | None: return None +def pending_tool(records: list[dict]) -> str | None: + """Name of a tool call in the current turn that has no result yet. At an + idle alert every call has its result, so a pending one means Claude is + showing a prompt — a permission request or plan approval.""" + answered = {b.get("tool_use_id") for rec in records for b in _content_blocks(rec) + if b.get("type") == "tool_result"} + for rec in reversed(current_turn(records)): + if agent_session.record_role(rec) != "assistant": + continue + for block in _content_blocks(rec): + if block.get("type") == "tool_use" and block.get("id") not in answered: + name = block.get("name") + return name if isinstance(name, str) and name else "a tool" + return None + + +def ended_on_api_error(records: list[dict]) -> bool: + """True if the agent's newest message is an API error Claude wrote into the + transcript (an expired login, an overloaded API): the session has stopped.""" + for rec in reversed(current_turn(records)): + if agent_session.record_role(rec) == "assistant": + return rec.get("isApiErrorMessage") is True + return False + + def trim_words(text: str, limit: int = LAST_MESSAGE_CHARS) -> str: """Collapse whitespace and cut to `limit` chars on a word boundary.""" flat = " ".join(text.split()) @@ -472,21 +523,76 @@ def trim_words(text: str, limit: int = LAST_MESSAGE_CHARS) -> str: return flat[:limit].rsplit(" ", 1)[0] + "…" -def read_last_message(cwd: str | None, session_id: str | None, *, - question: bool) -> str | None: - """What the agent last said, for the Slack ping: the pending question when - `question` (an AskUserQuestion event — the hook payload redacts its text), - else the last assistant text. Returns None on any failure so a missing or - malformed transcript never blocks the drop.""" +# Words that ask the user for something. Most turns end in a status update +# ("Standing by for the gate agents…") that needs nobody; the ones that need +# the user ask a question or ask them to act ("say the word", "reply +# `cleanup`", "please run /login"). +_ASKS_RE = re.compile( + r"\?|\b(let me know|your call|want me to|should i|shall i|say the word|awaiting your" + r"|your (approval|go-ahead|review|decision|sign-off)" + r"|please (run|paste|approve|confirm|reply))\b|\breply `|/login\b", re.I) +# An ask can sit a line or two above a closing note, so the last few lines count. +ASK_TAIL_LINES = 3 + + +# A link's query string (`?config=squirrel`) isn't a question. +_URL_RE = re.compile(r"\b[a-z][a-z0-9+.-]*://\S+", re.I) + + +def asks_user(text: str | None) -> bool: + """True if the message's last few non-empty lines ask the user something.""" + lines = [ln.strip() for ln in (text or "").splitlines() if ln.strip()] + return any(_ASKS_RE.search(_URL_RE.sub("", ln)) for ln in lines[-ASK_TAIL_LINES:]) + + +def session_title(records: list[dict]) -> str | None: + """A name for the session when its workspace title is cmux's default: + Claude's own session title, else the working folder and git branch.""" + for rec in reversed(records): + title = rec.get("aiTitle") if rec.get("type") == "ai-title" else None + if isinstance(title, str) and title.strip(): + return title.strip() + for rec in reversed(records): + cwd = rec.get("cwd") + if isinstance(cwd, str) and cwd: + branch = rec.get("gitBranch") + name = Path(cwd).name or cwd + return f"{name} ({branch})" if isinstance(branch, str) and branch not in ("", "HEAD") else name + return None + + +def prompt_message(session: dict) -> str | None: + """What an idle-alert ping should quote: the open question or prompt when + one is up, else the agent's last words.""" + tool = session.get("pending_tool") + last_text = session.get("last_text") + if tool == "AskUserQuestion" and session.get("question"): + return session["question"] + if tool == "ExitPlanMode": + return "Waiting for you to approve its plan." + if tool and not asks_user(last_text): + return f"Waiting for your OK to use {tool}." + return last_text + + +def read_session(cwd: str | None, session_id: str | None) -> dict | None: + """What the session's transcript says right now: the pending question (an + AskUserQuestion's text — the hook payload redacts it), the agent's last + message this turn, any tool call still waiting on a prompt, whether the + turn ended on an API error, and a title. None on any failure so a missing + or malformed transcript never blocks the drop.""" try: path = transcript_path(cwd, session_id) if path is None: return None records = tail_records(path) - text = pending_question(records) if question else last_assistant_text(records) + return {"question": pending_question(records), + "last_text": last_assistant_text(records), + "pending_tool": pending_tool(records), + "api_error": ended_on_api_error(records), + "title": session_title(records)} except Exception: # noqa: BLE001 — transcripts are external; the ping must still go return None - return trim_words(text) if text else None # ─── event classification (pure) ────────────────────────────────────────────── @@ -625,10 +731,10 @@ def cooled_down(self, ws_key: str, signal_type: str) -> bool: def handle_event(evt: dict, bank: PatternBank, state: WatcherState, resolver: WsRefResolver, *, screen_reader=read_screen, - message_reader=read_last_message) -> dict | None: + session_reader=read_session) -> dict | None: """Process one parsed event end-to-end. Returns the dropped item dict (for tests/logging) or None when nothing was dropped. `screen_reader` and - `message_reader` are injectable so tests don't shell out to cmux or read + `session_reader` are injectable so tests don't shell out to cmux or read real transcripts.""" cls = classify_event(evt) if cls is None: @@ -638,42 +744,59 @@ def handle_event(evt: dict, bank: PatternBank, state: WatcherState, workspace_id = cls["workspace_id"] ws_ref = resolver.resolve(workspace_id) - ws_title = resolver.title(workspace_id) ws_key = ws_ref or workspace_id or "unknown" + session = session_reader(cls["cwd"], cls["session_id"]) or {} + ws_title = resolver.title(workspace_id) or session.get("title") + last_text = session.get("last_text") if cls["signal"] == "needs_input": - # Always a signal — the agent is blocked on the user. Cooldown only. + pattern_matched = cls["event_name"].split(".")[-1] # Notification / AskUserQuestion + asking = pattern_matched == "AskUserQuestion" + # cmux sends every Claude alert as this one Notification event, with no + # message: a permission prompt looks the same as the idle "waiting for + # your input" alert that fires about a minute after every turn. Ping + # when a prompt is up (a tool call still waiting), the session died on + # an API error, or the agent's last words ask the user something; an + # unreadable transcript still pings, as before. + needs_user = (not session or session.get("pending_tool") + or session.get("api_error") or asks_user(last_text)) + if not asking and not needs_user: + log(f"skip idle needs_input ws={ws_ref or workspace_id}: last message asks nothing") + return None if not state.cooled_down(ws_key, "needs_input"): return None snippet = last_lines(screen_reader(workspace_id or ws_ref or "")) - pattern_matched = cls["event_name"].split(".")[-1] # Notification / AskUserQuestion - last_message = message_reader(cls["cwd"], cls["session_id"], - question=pattern_matched == "AskUserQuestion") + message = session.get("question") if asking else prompt_message(session) item = drop_inbox_item(ws_ref, "needs_input", pattern_matched, snippet, - ws_title=ws_title, last_message=last_message) + ws_title=ws_title, + last_message=trim_words(message) if message else None) record_fired(pattern_matched, ws_ref, "needs_input") log(f"drop needs_input ws={ws_ref or workspace_id} via={pattern_matched} → {item.name}") return {"path": str(item), "signal_type": "needs_input", "pattern_matched": pattern_matched, "ws_ref": ws_ref} - # turn_end: read the screen and pattern-match. Drop only on a non-muted hit. + # turn_end: pattern-match what the agent just said, not 50 lines of screen, + # where old scrollback and tool output matched words like "blocked" or + # "API error" that had nothing to do with this turn. The screen is only the + # fallback when the transcript can't be read. screen = screen_reader(workspace_id or ws_ref or "") - if not screen: - return None # dead/headless workspace or read failure — nothing to judge - hits = bank.match(screen) + text = last_text or screen + if not text: + return None # nothing said this turn and no screen to read — nothing to judge + # A suppressed pattern doesn't count, so it can't hide a lower-priority + # real hit (a message saying both "awaiting review" and "CI is red"). + hits = [h for h in bank.match(text) if not h.get("suppress")] if not hits: return None # plain turn-end with nothing notable — the noise floor top = hits[0] - if top.get("suppress"): - return None pat_signal = top.get("signal") signal_type = pat_signal if pat_signal in ("needs_input", "work_complete") else "pattern_match" if not state.cooled_down(ws_key, signal_type): return None - snippet = last_lines(screen) - last_message = message_reader(cls["cwd"], cls["session_id"], question=False) + snippet = last_lines(screen) or (trim_words(last_text) if last_text else "") item = drop_inbox_item(ws_ref, signal_type, top.get("id", ""), snippet, - ws_title=ws_title, last_message=last_message) + ws_title=ws_title, + last_message=trim_words(last_text) if last_text else None) record_fired(top.get("id", ""), ws_ref, signal_type) log(f"drop {signal_type} ws={ws_ref or workspace_id} pattern={top.get('id')} → {item.name}") return {"path": str(item), "signal_type": signal_type, diff --git a/docs/assistant-comms-onboarding.md b/docs/assistant-comms-onboarding.md index bc96b95..0b425a9 100644 --- a/docs/assistant-comms-onboarding.md +++ b/docs/assistant-comms-onboarding.md @@ -37,7 +37,8 @@ Replies come from a **warm cmux Claude session** (Sonnet, scoped `--add-dir`) th |---|---| | Assistant appends a verified action to its ledger | `I asked a workspace to merge its PR.` with the refs in a trailing italic line | | Same, but evidence is `screen_read` (Assistant rejects this as weak) | `Heads up: I only confirmed this by reading the screen, which isn't reliable proof.` | -| A workspace asks you a question / needs input / finishes work | `** is asking you: ` or `** needs your input.` with the agent's last message quoted | +| A workspace asks you a question, shows a permission or plan prompt, stops on an API error, or goes idle on a message that asks you something | `** is asking you: ` or `** needs your input.` with the agent's last message quoted. Idle workspaces whose last message is a status update don't ping. | +| The agent's own message says CI is red, a PR opened, or tests passed | `** needs your input.` or `** looks done.` with the message quoted | | Assistant heartbeat stale (>20 min) or status flips to `frozen`/`stale_world`/`respawn-requested` | `*Assistant's main loop has stopped* — no run for 25m …`, then `*Assistant's main loop is running again* after 3h.` | Messages are Slack `mrkdwn`. diff --git a/tests/test_cmux_watcher.py b/tests/test_cmux_watcher.py index 99c0a68..3721668 100644 --- a/tests/test_cmux_watcher.py +++ b/tests/test_cmux_watcher.py @@ -123,9 +123,38 @@ def test_old_bank_ci_green_turn_end_is_silent(self): _evt("agent.hook.Stop", request_id="rCI"), bank, self.mod.WatcherState(cooldown_sec=0), FakeResolver(), screen_reader=lambda ws: "All done — CI is green", - message_reader=MessageReaderSpy()) + session_reader=SessionReaderSpy()) self.assertIsNone(res, "a suppressed default must stay quiet in an old bank") + def test_noisy_needs_input_word_lists_are_muted_by_default(self): + """Even an old on-disk bank without `suppress` keys stays quiet for + stranded / awaiting-review / emit-card, which fired on status prose.""" + self.bank_path.write_text(json.dumps({"version": 1, "patterns": [ + {"id": pid, "regex": rx, "signal": "needs_input", "priority": "high"} + for pid, rx in (("stranded", "blocked"), ("awaiting-review", "awaiting.{0,30}review"), + ("emit-card", "needs_user"))]})) + bank = self._bank() + for text in ("the rollout is blocked on infra", + "Awaiting the standing adversary, code-review, and G3", + "card state needs_user"): + res = self.mod.handle_event( + _evt("agent.hook.Stop", request_id=text), bank, + self.mod.WatcherState(cooldown_sec=0), FakeResolver(), + screen_reader=lambda ws: "", session_reader=SessionReaderSpy(last_text=text)) + self.assertIsNone(res, text) + + def test_a_suppressed_hit_doesnt_hide_a_real_one(self): + self.bank_path.write_text(json.dumps({"version": 1, "patterns": [ + {"id": "awaiting-review", "regex": "awaiting review", "signal": "needs_input", + "priority": "high", "suppress": True}, + {"id": "ci-red", "regex": "CI is red", "signal": "needs_input", "priority": "medium"}]})) + res = self.mod.handle_event( + _evt("agent.hook.Stop", request_id="both"), self._bank(), + self.mod.WatcherState(cooldown_sec=0), FakeResolver(), + screen_reader=lambda ws: "", + session_reader=SessionReaderSpy(last_text="PR awaiting review, but CI is red")) + self.assertEqual(res["pattern_matched"], "ci-red") + def test_explicit_suppress_in_file_wins(self): self._write_june_bank({"suppress": False}) by_id = {p["id"]: p for p in self._bank().patterns} @@ -217,15 +246,18 @@ def title(self, uuid): return self._title if uuid else None -class MessageReaderSpy: - """Stands in for read_last_message; records each call's arguments.""" +class SessionReaderSpy: + """Stands in for read_session; records each call's arguments.""" - def __init__(self, reply=None): - self.reply = reply + def __init__(self, question=None, last_text=None, title=None, unreadable=False, + pending_tool=None, api_error=False): + self.reply = None if unreadable else { + "question": question, "last_text": last_text, "title": title, + "pending_tool": pending_tool, "api_error": api_error} self.calls = [] - def __call__(self, cwd, session_id, *, question): - self.calls.append((cwd, session_id, question)) + def __call__(self, cwd, session_id): + self.calls.append((cwd, session_id)) return self.reply @@ -273,40 +305,119 @@ def test_classify_passes_session_id_through(self): def test_ask_user_question_reads_pending_question(self): bank, state, resolver = self._components(title="Fix archself deferral door") - reader = MessageReaderSpy(reply="Should I rebase or merge main?") + reader = SessionReaderSpy(question="Should I rebase or merge main?", + last_text="Two ways to land this.") res = self.mod.handle_event( _evt("agent.hook.AskUserQuestion", cwd="/w/repo", session_id="S-1"), bank, state, resolver, screen_reader=lambda ws: "Which option?", - message_reader=reader) - self.assertEqual(reader.calls, [("/w/repo", "S-1", True)]) + session_reader=reader) + self.assertEqual(reader.calls, [("/w/repo", "S-1")]) item = json.loads(Path(res["path"]).read_text()) self.assertEqual(item["ws_title"], "Fix archself deferral door") self.assertEqual(item["last_message"], "Should I rebase or merge main?") - def test_notification_reads_last_text_not_question(self): + def test_notification_pings_when_the_agent_asks_something(self): bank, state, resolver = self._components() - reader = MessageReaderSpy() + reader = SessionReaderSpy(question="an old question", + last_text="Fixed it.\n\nWant me to open the PR?", + title="assistant (fix/comms)") res = self.mod.handle_event( _evt("agent.hook.Notification", request_id="rN2", session_id="S-2"), bank, state, resolver, screen_reader=lambda ws: "waiting", - message_reader=reader) - self.assertEqual(reader.calls, [("/x", "S-2", False)]) + session_reader=reader) + item = json.loads(Path(res["path"]).read_text()) + self.assertEqual(item["last_message"], "Fixed it. Want me to open the PR?") + self.assertEqual(item["ws_title"], "assistant (fix/comms)", + "a default-titled workspace falls back to the session's title") + + def test_idle_notification_after_a_status_update_is_skipped(self): + """Claude's idle alert fires about a minute after every turn. When the + turn ended with a status update, nobody needs to act, so no ping + (2026-09-28: 451 such pings in two weeks). Mutation probe: drop the + asks_user gate and this drops an item.""" + bank, state, resolver = self._components() + reader = SessionReaderSpy(last_text="Standing by for the four gate agents. I'll resume.") + res = self.mod.handle_event( + _evt("agent.hook.Notification", request_id="rN3"), bank, state, resolver, + screen_reader=lambda ws: "waiting", session_reader=reader) + self.assertIsNone(res) + self.assertEqual(list(self.inbox.glob("cmux-*.json")) if self.inbox.exists() else [], []) + + def test_idle_notification_with_an_unreadable_transcript_still_pings(self): + bank, state, resolver = self._components() + res = self.mod.handle_event( + _evt("agent.hook.Notification", request_id="rN4"), bank, state, resolver, + screen_reader=lambda ws: "waiting", session_reader=SessionReaderSpy(unreadable=True)) item = json.loads(Path(res["path"]).read_text()) self.assertNotIn("ws_title", item) self.assertNotIn("last_message", item) + def test_skipped_idle_notification_doesnt_use_up_the_cooldown(self): + bank = self.mod.PatternBank(self.assistant / "pattern_bank.json") + state = self.mod.WatcherState(cooldown_sec=600) + resolver = FakeResolver() + self.mod.handle_event( + _evt("agent.hook.Notification", request_id="a"), bank, state, resolver, + screen_reader=lambda ws: "", session_reader=SessionReaderSpy(last_text="Done.")) + res = self.mod.handle_event( + _evt("agent.hook.Notification", request_id="b"), bank, state, resolver, + screen_reader=lambda ws: "", session_reader=SessionReaderSpy(last_text="Merge it?")) + self.assertIsNotNone(res) + def test_turn_end_drop_carries_title_and_last_text(self): bank, state, resolver = self._components(title="Green E2E Suite") - reader = MessageReaderSpy(reply="Opened the PR; CI is running.") + reader = SessionReaderSpy(last_text="Done. PR #321 opened; CI is running.") res = self.mod.handle_event( _evt("agent.hook.Stop", request_id="rS2", session_id="S-3"), bank, state, resolver, - screen_reader=lambda ws: "Done. PR #321 opened for review.", - message_reader=reader) - self.assertEqual(reader.calls, [("/x", "S-3", False)]) + screen_reader=lambda ws: "some screen", + session_reader=reader) + self.assertEqual(reader.calls, [("/x", "S-3")]) item = json.loads(Path(res["path"]).read_text()) self.assertEqual(item["ws_title"], "Green E2E Suite") - self.assertEqual(item["last_message"], "Opened the PR; CI is running.") + self.assertEqual(item["last_message"], "Done. PR #321 opened; CI is running.") + + + def test_turn_end_matches_the_agents_message_not_old_scrollback(self): + """Patterns match what the agent just said, not 50 lines of screen where + old output ("CI is red" from an earlier run) sits. Mutation probe: + match the screen, or screen plus message, and the first call drops.""" + bank, state, resolver = self._components() + res = self.mod.handle_event( + _evt("agent.hook.Stop", request_id="rS9"), bank, state, resolver, + screen_reader=lambda ws: "earlier run: CI is red\nPR #12 opened", + session_reader=SessionReaderSpy(last_text="Summarized the design docs.")) + self.assertIsNone(res) + res = self.mod.handle_event( + _evt("agent.hook.Stop", request_id="rS10"), bank, state, resolver, + screen_reader=lambda ws: "", + session_reader=SessionReaderSpy(last_text="Heads up: CI is red on main.")) + self.assertEqual(res["pattern_matched"], "ci-red") + item = json.loads(Path(res["path"]).read_text()) + self.assertEqual(item["screen_snippet"], "Heads up: CI is red on main.", + "an empty screen falls back to the message for the snippet") + + def test_permission_prompt_pings_even_after_a_status_line(self): + """cmux sends a permission prompt as the same bare Notification as the + idle alert. A tool call still waiting means a prompt is up. Mutation + probe: drop the pending_tool check and this is skipped.""" + bank, state, resolver = self._components() + res = self.mod.handle_event( + _evt("agent.hook.Notification", request_id="rP"), bank, state, resolver, + screen_reader=lambda ws: "", + session_reader=SessionReaderSpy(last_text="Committing the fix now.", + pending_tool="Bash")) + item = json.loads(Path(res["path"]).read_text()) + self.assertEqual(item["last_message"], "Waiting for your OK to use Bash.") + + def test_session_that_died_on_an_api_error_pings(self): + bank, state, resolver = self._components() + res = self.mod.handle_event( + _evt("agent.hook.Notification", request_id="rE"), bank, state, resolver, + screen_reader=lambda ws: "", + session_reader=SessionReaderSpy(last_text="API Error: 529 overloaded", api_error=True)) + item = json.loads(Path(res["path"]).read_text()) + self.assertEqual(item["last_message"], "API Error: 529 overloaded") def test_notification_drops_needs_input(self): bank, state, resolver = self._components() @@ -565,35 +676,82 @@ def test_pending_question_none_for_missing_or_blank_text(self): def test_last_assistant_text_skips_user_and_non_text_blocks(self): records = [ + _assistant({"type": "text", "text": "an earlier turn's words"}), + _user({"type": "text", "text": "the user's prompt"}), _assistant({"type": "text", "text": "Opened the PR."}), _assistant({"type": "text", "text": " "}, {"type": "tool_use", "name": "Bash", "input": {}}), {"type": "assistant", "message": {"role": "assistant", "content": "plain string"}}, {"type": "summary", "summary": "not a turn"}, _assistant({"type": "tool_result", "text": "tool output, not the agent"}), - _user({"type": "text", "text": "the user's reply"}), + _user({"type": "tool_result", "tool_use_id": "t1", "content": "ok"}), ] self.assertEqual(self.mod.last_assistant_text(records), "Opened the PR.") self.assertIsNone(self.mod.last_assistant_text([_user({"type": "text", "text": "x"})])) + def test_last_assistant_text_ignores_earlier_turns(self): + """A turn with only tool calls must not re-judge the previous turn's + words ("PR #12 opened" would fire again).""" + records = [_assistant({"type": "text", "text": "PR #12 opened."}), + _user({"type": "text", "text": "next task"}), + _assistant({"type": "tool_use", "id": "t2", "name": "Bash", "input": {}}), + _user({"type": "tool_result", "tool_use_id": "t2", "content": "done"})] + self.assertIsNone(self.mod.last_assistant_text(records)) + typed = {"type": "user", "message": {"role": "user", "content": "plain prompt"}} + hook = {"type": "user", "message": {"role": "user", "content": ""}} + meta = {"type": "user", "isMeta": True, "message": {"role": "user", "content": "x"}} + self.assertEqual(self.mod.current_turn([_assistant(), typed, hook, meta]), [hook, meta]) + + def test_pending_tool_names_the_call_waiting_on_a_prompt(self): + waiting = [_user({"type": "text", "text": "go"}), + _assistant({"type": "text", "text": "Committing now."}, + {"type": "tool_use", "id": "b1", "name": "Bash", "input": {}})] + self.assertEqual(self.mod.pending_tool(waiting), "Bash") + done = [*waiting, _user({"type": "tool_result", "tool_use_id": "b1", "content": "ok"})] + self.assertIsNone(self.mod.pending_tool(done)) + nameless = [_assistant({"type": "tool_use", "id": "b2", "input": {}})] + self.assertEqual(self.mod.pending_tool(nameless), "a tool") + + def test_ended_on_api_error(self): + err = {"type": "assistant", "isApiErrorMessage": True, + "message": {"role": "assistant", "content": [{"type": "text", "text": "API Error: 403"}]}} + self.assertTrue(self.mod.ended_on_api_error([_user({"type": "text", "text": "go"}), err])) + self.assertFalse(self.mod.ended_on_api_error([err, _assistant({"type": "text", "text": "ok"})])) + self.assertFalse(self.mod.ended_on_api_error([_user({"type": "text", "text": "go"})])) + after_result = [_user({"type": "text", "text": "go"}), err, + _user({"type": "tool_result", "tool_use_id": "x", "content": "ok"})] + self.assertTrue(self.mod.ended_on_api_error(after_result)) + + def test_prompt_message_quotes_the_open_prompt(self): + pm = self.mod.prompt_message + self.assertEqual(pm({"pending_tool": "AskUserQuestion", "question": "Merge?"}), "Merge?") + self.assertEqual(pm({"pending_tool": "ExitPlanMode"}), "Waiting for you to approve its plan.") + self.assertEqual(pm({"pending_tool": "Bash", "last_text": "Committing."}), + "Waiting for your OK to use Bash.") + self.assertEqual(pm({"pending_tool": "Bash", "last_text": "Run it?"}), "Run it?") + self.assertEqual(pm({"last_text": "Done."}), "Done.") + def test_trim_words(self): self.assertEqual(self.mod.trim_words("a\n\n b"), "a b") long = "word " * 100 trimmed = self.mod.trim_words(long, limit=23) self.assertEqual(trimmed, "word word word word…") - def test_read_last_message_end_to_end(self): + def test_read_session_end_to_end(self): cwd = "/Users/me/dev/proj" records = [ + {"type": "ai-title", "aiTitle": "Rebase the ruler fix", "sessionId": "sess-9"}, _assistant({"type": "text", "text": "I found two ways.\n\nPick one."}), _assistant(_ask("t9", "Should I rebase or merge main?")), ] self._transcript(self.mod.agent_session.claude_project_slug(cwd), "sess-9", records) - self.assertEqual(self.mod.read_last_message(cwd, "sess-9", question=True), - "Should I rebase or merge main?") - self.assertEqual(self.mod.read_last_message(cwd, "sess-9", question=False), - "I found two ways. Pick one.") - self.assertIsNone(self.mod.read_last_message(cwd, "sess-404", question=False)) + self.assertEqual(self.mod.read_session(cwd, "sess-9"), { + "question": "Should I rebase or merge main?", + "last_text": "I found two ways.\n\nPick one.", + "pending_tool": "AskUserQuestion", + "api_error": False, + "title": "Rebase the ruler fix"}) + self.assertIsNone(self.mod.read_session(cwd, "sess-404")) def test_live_payload_shape_finds_the_question(self): # Live cmux payloads prefix the id (`claude-`) while the file is @@ -610,17 +768,50 @@ def test_live_payload_shape_finds_the_question(self): item = json.loads(Path(res["path"]).read_text()) self.assertEqual(item["last_message"], "Should I rebase or merge main?") - def test_read_last_message_none_when_nothing_to_say(self): + def test_read_session_with_nothing_to_say(self): cwd = "/Users/me/dev/quiet" self._transcript(self.mod.agent_session.claude_project_slug(cwd), "sess-q", [_user({"type": "text", "text": "hello?"})]) - self.assertIsNone(self.mod.read_last_message(cwd, "sess-q", question=False)) + self.assertEqual(self.mod.read_session(cwd, "sess-q"), + {"question": None, "last_text": None, "pending_tool": None, + "api_error": False, "title": None}) - def test_read_last_message_never_raises(self): + def test_read_session_never_raises(self): # No ~/.claude/projects at all: the scan fails, the drop must not. - self.assertIsNone(self.mod.read_last_message("/nowhere", "sess-x", question=False)) + self.assertIsNone(self.mod.read_session("/nowhere", "sess-x")) self.assertFalse(self.projects.exists()) + def test_session_title_prefers_claudes_title_then_folder_and_branch(self): + title = self.mod.session_title + self.assertEqual(title([{"type": "ai-title", "aiTitle": " Fix ruler "}, + {"cwd": "/w/repo", "gitBranch": "main"}]), "Fix ruler") + self.assertEqual(title([{"cwd": "/w/architect-ffp", "gitBranch": "fix/x"}]), + "architect-ffp (fix/x)") + self.assertEqual(title([{"cwd": "/w/repo", "gitBranch": "HEAD"}]), "repo") + self.assertEqual(title([{"cwd": "/", "gitBranch": ""}]), "/") + self.assertEqual(title([{"type": "ai-title", "aiTitle": " "}, {"cwd": ""}]), None) + self.assertIsNone(title([])) + + def test_asks_user_reads_the_last_few_lines(self): + asks = self.mod.asks_user + self.assertTrue(asks("Done.\n\nWant me to open the PR?")) + self.assertTrue(asks("Two options below.\nLet me know which one.")) + self.assertTrue(asks("Should I merge")) + self.assertTrue(asks("Tear down the worktree (`cleanup`)?\n- The 3 deferrals remain open"), + "an ask above a closing note still counts") + self.assertTrue(asks("Paste the ticket text, or say the word and I'll open a browser")) + self.assertTrue(asks("Reply `cleanup` to tear down the batch4 worktree")) + self.assertTrue(asks("Please run /login · API Error: 403")) + self.assertTrue(asks("Ready for your approval.")) + self.assertFalse(asks("Is it fixed?\nYes.\nGates green.\nStanding by for CI."), + "a question four lines up is history, not an ask") + self.assertFalse(asks("Standing by for the four gate agents.")) + self.assertFalse(asks("Dev server is up: https://localhost.adobe.com:3000/?config=squirrel"), + "a link's query string isn't a question") + self.assertTrue(asks("Open https://x.test/?a=1 — does it look right?")) + self.assertFalse(asks("")) + self.assertFalse(asks(None)) + # ─── pattern hot-reload ────────────────────────────────────────────────────────