diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index 8837ed77..a31bdf35 100644 --- a/emrg/server/daemon.py +++ b/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 diff --git a/tests/test_ws_e2e.py b/tests/test_ws_e2e.py index a13379ce..65740277 100644 --- a/tests/test_ws_e2e.py +++ b/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