emrg: a cancel reports only a turn it stopped

emrg/server/daemon.py

@@ -1000,7 +1000,13 @@ async def _handle_client(self, ws) -> None:
                     if event:
                         event.set()
                     cancel_task = self._session_turn_task.get(cancel_sid) or _tool_task
-                    if cancel_task and not cancel_task.done():
+                    # "Is there a turn here to stop?" is read off the task, not off
+                    # `event`: the connection's locals outlive their turn — a finished
+                    # turn leaves `_cancel_event` set and `_tool_task` assigned until
+                    # the next turn replaces it — so a guard written on `event` reports
+                    # an interruption that stopped nothing (issue #1470).
+                    stopped = bool(cancel_task and not cancel_task.done())
+                    if stopped:
                         cancel_task.cancel()
                         # Only await a task this coroutine owns: awaiting a peer's turn
                         # would park this connection's read loop until that turn unwinds.
@@ -1012,10 +1018,16 @@ async def _handle_client(self, ws) -> None:
                     # The receipt is a statement about the *session*, so every client
                     # watching it gets the same one. A connection-local ack let a peer
                     # read "cancelled" while the turn it never reached kept running.
-                    await self._broadcast(cancel_sid, {
-                        "type": "cancelled",
-                        "session_id": cancel_sid,
-                    })
+                    # It is also a statement that something *stopped*: a turn that had
+                    # already finished is not one, and a client that narrates every
+                    # receipt (GUI) would print "interrupted" under a completed answer,
+                    # reachable whenever Esc lands after the last chunk (issue #1470).
+                    # A cancel that reached no live turn has nothing to report.
+                    if stopped:
+                        await self._broadcast(cancel_sid, {
+                            "type": "cancelled",
+                            "session_id": cancel_sid,
+                        })
                     if cancel_task is _tool_task:
                         _tool_task = None
                         _cancel_event = None

tests/test_ws_e2e.py

@@ -729,11 +729,28 @@ async def spy_execute(arguments):
         asyncio.run(_test())
 
     def test_cancel_stops_task(self):
+        """A cancel on a *live* turn answers with the `cancelled` receipt.
+
+        The turn has to still be running when the cancel lands: the receipt now
+        says something stopped, so a turn that ended on its own earns none
+        (issue #1470). The default fake stream finishes in milliseconds, so
+        without the slow stream below this test would be waiting on a turn that
+        was already over — which is how it passed while the receipt was sent
+        unconditionally, proving nothing about a cancel stopping anything.
+        """
         async def _test():
             with tempfile.TemporaryDirectory() as tmp:
                 cwd = Path(tmp)
-                _, _, cleanup = await _boot_server(cwd)
+                server, _, cleanup = await _boot_server(cwd)
                 try:
+                    async def slow_chat_stream(messages, tools=None):
+                        yield {"content": "工作中", "tool_calls": None,
+                               "finish_reason": None, "usage": None}
+                        await asyncio.sleep(30)
+                        yield {"content": "太晚了", "tool_calls": None,
+                               "finish_reason": "stop", "usage": None}
+
+                    server.llm.chat_stream = slow_chat_stream
                     ws = await connect_to_server()
                     try:
                         task = {
@@ -827,6 +844,59 @@ async def slow_chat_stream(messages, tools=None):
                     await cleanup()
         asyncio.run(_test())
 
+    def test_a_cancel_that_stopped_nothing_is_not_reported(self):
+        """Issue #1470: the receipt is also a statement that something *stopped*.
+
+        Both clients narrate the receipt, so a receipt for an interruption that
+        did not happen prints "interrupted" under an answer that had already
+        completed. The trap is that `_cancel_event`/`_tool_task` are read-loop
+        locals which outlive their turn: after a turn ends on its own, the
+        finished turn's event is still set and its task object still assigned,
+        so a daemon that asks "was there an event?" still says yes — the
+        question has to be put to the *session*, which retracts its handles
+        when the turn unwinds.
+
+        The pong is what makes the absence measurable rather than assumed: the
+        read loop handles frames in order, so a receipt the daemon decided to
+        send would have to arrive before the probe's answer, and the answer
+        itself proves the connection was alive and being read throughout.
+        """
+        async def _test():
+            with tempfile.TemporaryDirectory() as tmp:
+                cwd = Path(tmp)
+                _, _, cleanup = await _boot_server(cwd)
+                try:
+                    ws = await connect_to_server()
+                    try:
+                        sid = "s_late_cancel"
+                        await ws.send(json.dumps({
+                            "type": "task", "id": "t-late", "session_id": sid,
+                            "cwd": str(cwd), "prompt": "你好", "stream": True,
+                            "timestamp": "2026-09-21T00:00:00",
+                        }, ensure_ascii=False))
+                        # Wait for `turn_end`, not `done`: the turn's handles are
+                        # retracted before that frame is broadcast, so its arrival
+                        # is the proof the cancel below really has nothing to stop.
+                        await _recv_until(ws, lambda f: f.get("type") == "turn_end",
+                                          what="turn_end")
+                        await ws.send(json.dumps({"type": "cancel", "session_id": sid}))
+                        await ws.send(json.dumps({"type": "ping"}))
+                        seen = []
+                        while True:
+                            frame = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
+                            seen.append(frame.get("type"))
+                            if frame.get("type") == "pong":
+                                break
+                        assert "cancelled" not in seen, (
+                            "a cancel that reached no live turn was reported as an "
+                            f"interruption anyway; frames in order: {seen}"
+                        )
+                    finally:
+                        await ws.close()
+                finally:
+                    await cleanup()
+        asyncio.run(_test())
+
     def test_round_complete_marker_written_on_normal_completion(
             self, tmp_path, monkeypatch):
         """Issue #1114: a genuinely completed round (final text answer after