emrg: a cancel names the session, and its receipt ends the turn

emrg/client/app.py

@@ -163,6 +163,121 @@ def _task_open_switch(pending: tuple[str, str] | None, resumed_sid: str) -> str
     return pending[0]
 
 
+# ── Stopping a turn: ask the session, wait for its receipt ───
+#
+# Rant 2026-09-20T12:50:13. A cancel used to be two statements about one event:
+# the client cleared busy, stopped the timer and printed "⏸ Interrupted" the
+# moment Esc was pressed, then sent a `cancel` that named no session; the daemon
+# answered the *connection* that asked. So a client could read "interrupted"
+# about a turn it never stopped (a peer client's Esc reached nothing, because the
+# turn belongs to the session, not to the connection that started it), and the
+# client that did stop one had already said so before the daemon knew.
+#
+# The daemon now resolves the turn from the session's own registration and
+# broadcasts one receipt to every client watching that session (PR #1474), so the
+# two halves below are the client's side of that: the request names the session,
+# and the receipt — not the keystroke — is what ends the turn this client shows.
+
+#: What every client prints when a cancel's receipt arrives. One wording, because
+#: the client that asked is not privileged: the same cancel from a peer reads the
+#: same here, which is the property the old optimistic branch could not have.
+CANCELLED_LINE = "⏸ Interrupted — response stopped. You can continue."
+
+
+async def request_cancel(conn, session_id: str) -> None:
+    """Ask the daemon to stop ``session_id``'s turn; send nothing if unknown.
+
+    The session is named because that is what a cancel is about: the daemon
+    resolves the turn from the session's registration, so any client watching
+    that session can stop the same turn with the same frame. A bare
+    ``{"type": "cancel"}`` is not a fallback but the defect — the daemon answers
+    it by resolving *this connection's* last session, i.e. a client that cannot
+    say which session it is stopping gets whichever one it happened to touch
+    last. With no session to name there is nothing to ask for, so nothing is
+    sent (the turn, if any, ends on its own and its `done` frame says so).
+    """
+    if not session_id:
+        return
+    await conn.send_command("cancel", session_id=session_id)
+
+
+def cancelled_line(data: dict, session_id: str) -> str | None:
+    """The line a ``cancelled`` receipt earns on the client showing that session.
+
+    ``None`` means "not this client's to narrate", and it covers two shapes a
+    plain type check would confuse: a frame that is not a receipt at all, and a
+    receipt for a *different* session — the daemon broadcasts to every client
+    watching the cancelled session, and a client that has moved to another one
+    must not print an interruption about a turn it is not showing.
+
+    A receipt carrying no ``session_id`` is narrated: the daemon stamps every
+    one it sends (`daemon.py` cancel branch), so a missing stamp means an older
+    daemon, whose broadcast went to the session's subscribers and no further.
+    """
+    if data.get("type") != "cancelled":
+        return None
+    stamp = data.get("session_id") or ""
+    if stamp and session_id and stamp != session_id:
+        return None
+    return CANCELLED_LINE
+
+
+def receipt_is_about_a_turn_this_client_shows(
+    *, busy: bool, turn_ended_cancelled: bool
+) -> bool:
+    """True when the receipt's turn is the one on screen, so the line is true.
+
+    Two facts answer it together, because the receipt and the ending it reports
+    reach a client in *either* order and both have to narrate (measured on the
+    in-process daemon harness, cycles cyc20260921-010110 and -031413):
+
+    * the asker's own cancel — the daemon awaits the turn *it owns* before
+      broadcasting the receipt, and the unwinding tool loop broadcasts
+      ``done{cancelled: true}`` on the way out, so that frame lands **first** and
+      has already cleared ``busy``. Asking only ``busy`` printed nothing at all:
+      the host pressed Esc, the response stopped, and the TUI said nothing;
+    * a peer's cancel — the same cancel branch does not await a task this
+      connection does not own, so the receipt is broadcast immediately and lands
+      **first**, with nothing known about the ending yet.
+
+    `busy` is this client's own claim that a turn is in flight; `turn_ended_cancelled`
+    is set from the ``done`` frame's own ``cancelled`` field — the frame that
+    ended the turn, and the only place that says *why* it ended. The flag has to
+    be read rather than assumed false: a receipt can also arrive after a turn
+    that **completed**, when the cancel reached the daemon too late to stop it
+    (the task is already done, so the branch does not cancel it but still
+    broadcasts). Narrating that one would print "response stopped" under a
+    response that finished.
+
+    The second order is the one this cannot answer alone — at that moment all
+    four combinations are still open — so the receipt is held instead of dropped
+    and the question is asked again where the ending lands:
+    `the_ending_completes_a_receipt_this_client_holds`.
+    """
+    return busy or turn_ended_cancelled
+
+
+def the_ending_completes_a_receipt_this_client_holds(
+    *, receipt_held: bool, turn_ended_cancelled: bool
+) -> bool:
+    """Whether the turn that just ended is the one a held receipt was about.
+
+    The peer's order, and the reason a receipt that arrives with ``busy`` false
+    and no ending known yet cannot simply be dropped: the daemon broadcasts it
+    before the cancelled turn unwinds, so the ``done{cancelled: true}`` that says
+    what happened lands *after* it. Holding the receipt and printing the line
+    where the second fact arrives makes the two orders one rule — one statement,
+    from the receipt, whichever frame makes it true.
+
+    Both halves are load-bearing. Without the held receipt the client that
+    pressed Esc is told nothing; without the flag, a receipt held for a cancel
+    that arrived too late would narrate under a turn that **completed** when the
+    next ending came in. So a held receipt is spent by the first ``done`` whatever
+    it says (see the frame loop) and dropped by the start of a new turn.
+    """
+    return receipt_held and turn_ended_cancelled
+
+
 # ── Clipboard image support (platform-adaptive) ─────────────
 
 # Reading a clipboard *file path* must not go through the console locale codec.
@@ -525,11 +640,55 @@ def _status_left(
         """Format left status: version + session title + short ID + model."""
         return _format_status_left(title, sid, model, vision)
     busy = False; server_id = ""; need_new_assistant = False; session_title = ""
+    # Set from a `done` frame whose `cancelled` field is true, so a receipt that
+    # arrives after the ending still finds a turn this client is showing (rant
+    # 2026-09-20T12:50:13; both frame orders are measured and named in
+    # receipt_is_about_a_turn_this_client_shows). Cleared when the line is told and
+    # when a new turn starts.
+    #
+    # ⚠️ One binding, owned here and shared with the frame loop and the send path
+    # (`nonlocal` at both) — the first version of this declared it in each function,
+    # so `read_server`'s receipt branch read a local that no `done` had assigned
+    # yet and raised UnboundLocalError out of the loop, killing the reader task
+    # silently (measured, cycle cyc20260921-031413). A flag the loop reads has to
+    # live where the loop and the keystrokes both look.
+    turn_ended_cancelled = False
+    # The other order: a peer's cancel is broadcast before its turn unwinds, so the
+    # receipt arrives with nothing yet known about the ending. It is held here
+    # rather than dropped, and the line is printed where the second fact lands.
+    cancel_receipt_held = False
     current_model = ""  # model name tracked independently of server_id (rant 2026-08-11T20:02:43)
     # Effective image capability, as the daemon reports it (rant 2026-09-17T16:53:02).
     # None until a pong or model_set frame says — an older daemon never does.
     current_vision: bool | None = None
 
+    def _narrate_the_stop() -> None:
+        """Say that the turn stopped, from whichever frame made that knowable.
+
+        The one place the line is printed, and the one place the two facts are
+        spent: it is reached by the receipt this client was still busy for, and by
+        the `done{cancelled: true}` the receipt arrived *before*
+        (`the_ending_completes_a_receipt_this_client_holds`). A second site would be
+        a second wording waiting to drift, and two clients of one session are
+        supposed to read the same statement.
+        """
+        nonlocal busy, _elapsed_task, _last_center
+        nonlocal turn_ended_cancelled, cancel_receipt_held
+        turn_ended_cancelled = False  # spent: nothing later can borrow this ending
+        cancel_receipt_held = False
+        busy = False
+        if _elapsed_task:
+            _elapsed_task.cancel()
+            _elapsed_task = None
+        status.elapsed = ""
+        # Rant 2026-09-02T10:31:11：与 done 路径（:617）一致的标题复位——中断
+        # 漏清 term.set_title 会让标题残留 [m:ss] 计时停住。
+        term.set_title(f"{session_title or session_id} @ {project_name}")
+        chat.add("system", CANCELLED_LINE)
+        _last_center = server_id or "emrg"
+        status.update(center=_last_center)
+        chat.dirty = True; term.render()
+
     status = StatusLine(left=_status_left(session_title, session_id, current_model, current_vision), center="connecting...")
     inp = InputWidget(); chat = ChatHistory()
     term.mount(status=status, composer=inp, chat=chat)
@@ -619,6 +778,7 @@ def _render_throttled():
 
     async def read_server():
         nonlocal stream_buffer, status, history, chat, busy, server_id, need_new_assistant, session_id, session_title, msg_count, tool_args, _welcomed
+        nonlocal turn_ended_cancelled, cancel_receipt_held
         nonlocal current_model, current_vision
         nonlocal _last_center, _elapsed_task, conn
         nonlocal _request_start
@@ -726,6 +886,8 @@ async def _reconnect():
                     _queued_sends.clear()
                     if to_resend:
                         was_busy = busy
+                        turn_ended_cancelled = False  # a new turn is this client's to show
+                        cancel_receipt_held = False   # and no earlier receipt is this turn's
                         busy = True; need_new_assistant = True; stream_buffer = ""
                         _request_start = time.time()
                         if _elapsed_task is None:
@@ -823,10 +985,39 @@ async def _reconnect():
                     term.render()
                     continue
 
+                if data.get("type") == "cancelled":
+                    # The receipt, not the Esc that asked for it, is what ends the
+                    # turn (rant 2026-09-20T12:50:13). It reaches this client in
+                    # either order relative to the ending it reports — the asker's
+                    # own cancel broadcasts it *after* the turn unwinds and clears
+                    # `busy` first, a peer's broadcasts it *before* — so the two
+                    # facts are asked here and in the `done` branch below, and the
+                    # line is printed by whichever frame finishes the sentence.
+                    line = cancelled_line(data, session_id)
+                    if line:
+                        if receipt_is_about_a_turn_this_client_shows(
+                            busy=busy, turn_ended_cancelled=turn_ended_cancelled
+                        ):
+                            _narrate_the_stop()
+                        else:
+                            # Nothing here says a turn stopped yet, and the frame
+                            # that would is still in flight: keep the receipt.
+                            cancel_receipt_held = True
+                    continue
+
                 if data.get("type") == "turn_start":
                     # Rant 2026-09-02T10:36:26：daemon 权威 turn 开始帧——把本地计时
                     # 对齐到实际执行时刻（排队请求目前从发送时刻起算，计时偏大）。
                     # 仅当本端 busy（该 turn 属于本会话）时对齐；started_at 为 epoch 秒。
+                    #
+                    # A held receipt expires at this boundary (the peer order above): it
+                    # belongs to the turn that just ended, and the daemon broadcasts
+                    # `turn_start` for every turn of the session, so a hold can never
+                    # outlive it. Without this it would also answer the cancelled `done`
+                    # of a *later* turn cut off by a disconnect — the one cancel that
+                    # broadcasts no receipt of its own — and print an interruption under
+                    # that turn instead.
+                    cancel_receipt_held = False
                     started = data.get("started_at")
                     if busy and isinstance(started, (int, float)) and started > 0:
                         _request_start = float(started)
@@ -857,6 +1048,13 @@ async def _reconnect():
                 if resp.done:
                     logger.info("response complete, %d chars", len(stream_buffer) if stream_buffer else 0)
                     logger.debug("DONE: stream_buffer=%r", stream_buffer[:80])
+                    # Why this turn ended, read off the frame that ended it
+                    # (rant 2026-09-20T12:50:13): a cancel's receipt arrives *after*
+                    # this frame, because the daemon awaits the turn it owns before
+                    # broadcasting it, so the receipt's own turn is already past
+                    # `busy` by then. False for every other outcome, so an unrelated
+                    # receipt cannot borrow this turn's ending.
+                    turn_ended_cancelled = data.get("cancelled") is True
                     busy = False
                     # Cancel elapsed timer
                     if _elapsed_task:
@@ -885,6 +1083,19 @@ async def _reconnect():
                         msg_count += 1
                     _update_left_extra()
                     term.render()
+                    # The other half of the receipt's sentence: a peer's cancel
+                    # broadcasts the receipt before its turn unwinds, so this is the
+                    # frame that says *why* the turn ended, and the line becomes true
+                    # here. Printed after the buffer flush above, so the system row it
+                    # adds is the last row and cannot be overwritten by that flush.
+                    if the_ending_completes_a_receipt_this_client_holds(
+                        receipt_held=cancel_receipt_held,
+                        turn_ended_cancelled=turn_ended_cancelled,
+                    ):
+                        _narrate_the_stop()
+                    # Spent either way: a held receipt is about one ending, and a
+                    # cancel that arrived too late must not narrate under the next one.
+                    cancel_receipt_held = False
                 if "error" in data:
                     err = data["error"]; logger.error("server error: %s", err)
                     chat.add("system", f"Error: {err}"); term.render()
@@ -1516,6 +1727,7 @@ def _handle_selector_nav(data: bytes, widget) -> bool:
 
     async def handle_key(data: bytes) -> bool:
         nonlocal inp, status, history, paste_mode, stream_buffer, conn, chat, busy, need_new_assistant, session_id, session_title, msg_count, cwd
+        nonlocal turn_ended_cancelled, cancel_receipt_held
         nonlocal current_model, project_name
         nonlocal session_sel, delete_sel, project_sel, model_sel, rewind_sel, task_sel
         nonlocal _task_list_intent, _task_open_pending
@@ -1564,21 +1776,12 @@ async def handle_key(data: bytes) -> bool:
         # Mimics Claude Code: Esc stops the current response mid-turn,
         # keeping work done so far. Dialogs (selector/autocomplete) are
         # handled below — this only fires when LLM is actively responding.
+        # Nothing is cleared or narrated here (rant 2026-09-20T12:50:13): the
+        # daemon's `cancelled` receipt is the only statement of what stopped, so
+        # a keypress cannot claim a stop the session never took. The typing
+        # indicator keeps turning until that receipt arrives.
         if data == b"\x1b" and busy:
-            busy = False
-            if _elapsed_task:
-                _elapsed_task.cancel()
-                _elapsed_task = None
-            status.elapsed = ""
-            # Rant 2026-09-02T10:31:11：与 done 路径（:617）一致的标题复位——ESC 中断
-            # 漏清 term.set_title 会让标题残留 [m:ss] 计时停住。
-            term.set_title(f"{session_title or session_id} @ {project_name}")
-            # Send cancel to daemon so it stops tool/LLM processing
-            await conn.send_command("cancel")
-            chat.add("system", "⏸ Interrupted — response stopped. You can continue.")
-            _last_center = server_id or "emrg"
-            status.update(center=_last_center)
-            chat.dirty = True; term.render()
+            await request_cancel(conn, session_id)
             return True
 
         # ── Session selector mode ──────────────────────────
@@ -2374,6 +2577,8 @@ def _is_image_token(s, i):
                 # when the turn ends. Track the send for requeue.
                 was_busy = busy
 
+                turn_ended_cancelled = False  # a new turn is this client's to show
+                cancel_receipt_held = False   # and no earlier receipt is this turn's
                 busy = True; need_new_assistant = True  # rant #32: force new StreamingMarkdown per response
                 _request_start = time.time()
                 # Cancel any stale timer and start a new one