diff --git a/Agent.md b/Agent.md index 2334587..7a67b86 100644 --- a/Agent.md +++ b/Agent.md @@ -112,7 +112,7 @@ Community needs voiced in HN agent-UI discussions map directly to EMRG's design: pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.port; python -m emrg ``` -Python: `uv run pytest tests/ -v` (681) — import check: `uv run python -c "from emrg.client.app import run_client"` +Python: `uv run pytest tests/ -v` (687) — import check: `uv run python -c "from emrg.client.app import run_client"` GUI: `cd emrg/gui && npm test` (179: 43 daemon_client + 19 conn-manager + 22 app-commands + 59 renderer smoke + 15 i18n + 7 integration + 3 commands + 4 build-config + 7 gui-state) — syntax: `node --check main.js preload.js daemon_client.js renderer/js/*.js` CI: `uv run pytest` + GUI tests + **actionlint workflow lint** (`rhysd/actionlint@v1.7.12` gate, #444 — workflow 解析错误在 PR CI 即失败，如 `if:` secrets 上下文) Re-trigger: `scripts/re-trigger-ci.sh [branch]` (workflow_dispatch, #527 — 替代空 commit 重触发：Actions outage 会整段丢弃 push 事件，dispatch 走 API 路径不受影响) diff --git a/README.cn.md b/README.cn.md index 678b1fb..3fe09ff 100644 --- a/README.cn.md +++ b/README.cn.md @@ -144,7 +144,7 @@ EMRG 不只是追赶——它自己追上来。 贡献指南、源码安装、架构、详细 FAQ → [DEVELOPMENT.md](DEVELOPMENT.md)。 -快速检查：`uv run pytest tests/ -v`（当前 681 项）· `cd emrg/gui && npm test`（179 项：43 daemon_client + 19 conn-manager + 22 app-commands + 59 renderer smoke + 15 i18n + 7 integration + 3 commands + 4 build-config + 7 gui-state） +快速检查：`uv run pytest tests/ -v`（当前 687 项）· `cd emrg/gui && npm test`（179 项：43 daemon_client + 19 conn-manager + 22 app-commands + 59 renderer smoke + 15 i18n + 7 integration + 3 commands + 4 build-config + 7 gui-state） --- diff --git a/README.md b/README.md index 580b85d..886fab4 100644 --- a/README.md +++ b/README.md @@ -144,7 +144,7 @@ They're products. EMRG is an experiment in *closing the loop* — the AI improve Contributing, source installs, architecture, and the full FAQ → [DEVELOPMENT.md](DEVELOPMENT.md). -Quick checks: `uv run pytest tests/ -v` (currently 681 items) · `cd emrg/gui && npm test` (179: 43 daemon_client + 19 conn-manager + 22 app-commands + 59 renderer smoke + 15 i18n + 7 integration + 3 commands + 4 build-config + 7 gui-state) +Quick checks: `uv run pytest tests/ -v` (currently 687 items) · `cd emrg/gui && npm test` (179: 43 daemon_client + 19 conn-manager + 22 app-commands + 59 renderer smoke + 15 i18n + 7 integration + 3 commands + 4 build-config + 7 gui-state) --- diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index d72d14f..b973fb2 100644 --- a/emrg/server/daemon.py +++ b/emrg/server/daemon.py @@ -178,6 +178,10 @@ def __init__(self, llm_config: LlmConfig) -> None: # ── Phase 2 broadcast model (protocol-contract §2.6) ── self._session_subscribers: dict[str, set] = {} # session_id → set[ws] self._session_busy: dict[str, bool] = {} # session_id → active task? + # P1 queue-injection (rant 2026-08-10T21:55:37): per-session FIFO of + # (TaskRequest, allow_tools) received while a tool loop is busy — + # injected at the next round boundary (aligned with codex steer_input). + self._session_pending: dict[str, list[tuple[TaskRequest, bool]]] = {} self._all_connections: set = set() # all authenticated connections # Device-flow auth (rant 10:17 Stage 2b): background gh auth login --web task @@ -539,14 +543,10 @@ async def _handle_client(self, ws) -> None: }) continue # Phase 2 session-level lock (protocol-contract §2.6.5): - # one active task per session — concurrent clients get - # "session busy" instead of racing writes. - if self._session_busy.get(session_id): - await self._send(ws, { - "error": "session busy", - "session_id": session_id, - }) - continue + # one active task per session — concurrent tasks queue. + # P1 (rant 21:55:37): construct req + allow_tools FIRST + # (the busy branch must append req to the pending queue), + # then check busy. try: req = TaskRequest( id=data.get("id", ""), @@ -562,6 +562,16 @@ async def _handle_client(self, ws) -> None: # WorkBuddy P2 (rant 21:35): Ask mode — pure chat, no tools. # mode="ask" → LLM gets an empty tool set so it can only reply. allow_tools = data.get("mode", "auto") != "ask" + if self._session_busy.get(session_id): + # Queue the task; injected at the next round boundary. + self._session_pending.setdefault(session_id, []).append((req, allow_tools)) + await self._broadcast(session_id, { + "type": "task_queued", + "request_id": req.id, + "session_id": session_id, + "position": len(self._session_pending[session_id]), + }) + continue # Cancel previous task if still running if _tool_task and not _tool_task.done(): if _cancel_event: @@ -1465,6 +1475,14 @@ async def _process_message( return session = self._get_or_create_session(session_id, Path(cwd)) session.clear() + # P1 (rant 21:55:37) Change F: clearing a session also drops its + # pending queue (queued messages are stale after clear). + dropped = self._session_pending.pop(session_id, []) + if dropped: + await self._broadcast(session_id, { + "type": "queued_cancelled", + "session_id": session_id, + }) await self._send(ws, { "type": "clear_result", "session_id": session_id, @@ -1492,6 +1510,14 @@ async def _process_message( deleted = Session.delete(session_id, Path(cwd)) if deleted: + # P1 (rant 21:55:37) Change F: deleting the session also + # drops its pending queue. + dropped = self._session_pending.pop(session_id, []) + if dropped: + await self._broadcast(session_id, { + "type": "queued_cancelled", + "session_id": session_id, + }) await self._send(ws, { "type": "session_deleted", "session_id": session_id, @@ -1643,6 +1669,50 @@ def _build_user_content(text: str, images: list[dict] | None, vision: bool = Fal # finally runs → lock released. This also roots out the multi-connection # write race (§5): session writes are serialized by the single active task. + async def _inject_pending_messages( + self, session: Session, messages: list[dict], + ) -> tuple[int, bool]: + """Pop the session's pending queue and inject it into `messages`. + + P1 queue-injection (rant 2026-08-10T21:55:37), aligned with codex + steer_input: messages sent while the tool loop is busy are queued per + session and injected at the next round boundary (after the current + round's tools, before the next LLM request). + + Uses pop() for atomic removal — while we await broadcasts, the read + loop may append new messages to the same list; popping the whole list + means those land in a fresh list (setdefault) and are injected next + round. Never dropped. + + Each injected message is persisted (append_message) so auto-compact + rebuilding from history keeps it, and a ``steer_committed`` broadcast + tells clients the message was committed into the turn. + + Returns ``(injected_count, ask_injected)`` — ``ask_injected`` is True + when any queued message was Ask mode (mode=ask); the caller must use + an empty tool set for the round that processes it. + """ + sid = session.session_id + pending = self._session_pending.pop(sid, []) + if not pending: + return 0, False + ask_injected = any(not allow for _, allow in pending) + for preq, _ in pending: + pcontent = self._build_user_content( + preq.prompt, preq.images, self.llm.config.vision + ) + messages.append({"role": "user", "content": pcontent}) + record: dict = {"type": "message", "role": "user", "content": preq.prompt} + if preq.images: + record["images"] = preq.images + session.append_message(record) + await self._broadcast(sid, { + "type": "steer_committed", + "request_id": preq.id, + "session_id": sid, + }) + return len(pending), ask_injected + async def _run_tool_loop_locked( self, req: TaskRequest, ws, session: Session, cancel_event: asyncio.Event | None = None, @@ -1650,10 +1720,36 @@ async def _run_tool_loop_locked( ) -> None: """Run _run_tool_loop and release the session busy lock on exit.""" session_id = session.session_id + normal_end = False try: await self._run_tool_loop(req, ws, session, cancel_event, allow_tools) + normal_end = True finally: self._session_busy[session_id] = False + # P1 (rant 21:55:37): messages still queued when the loop ends are + # not lost. We do NOT start a follow-up task here (_tool_task / + # _cancel_event are read-loop locals — a hand-off would break + # cancel + busy tracking); instead: + # normal end → queued_requeue → clients auto re-send (busy is + # now released, the re-send goes through the normal path) + # cancel / error / disconnect → queued_cancelled (queue dropped) + pending = self._session_pending.pop(session_id, []) + if pending: + # A cancel (even one caught and returned from inside the loop) + # must NOT auto-requeue: the user stopped the turn. Exception / + # disconnect also drop the queue. Only a clean turn end + # re-sends the queued messages. + if normal_end and not (cancel_event and cancel_event.is_set()): + await self._broadcast(session_id, { + "type": "queued_requeue", + "session_id": session_id, + "request_ids": [r.id for r, _ in pending], + }) + else: + await self._broadcast(session_id, { + "type": "queued_cancelled", + "session_id": session_id, + }) async def _run_tool_loop( self, req: TaskRequest, ws, session: Session, @@ -1696,9 +1792,41 @@ async def _run_tool_loop( *history_messages, {"role": "user", "content": user_content}, ] - tools_openai = self.tools.to_openai_tools() if allow_tools else [] + tools_base = self.tools.to_openai_tools() if allow_tools else [] + # P1 (rant 21:55:37): injection rounds do NOT consume the round budget; + # force_ask latches "an Ask message was injected outside the round-top + # injection (stop / Case 3 / loop-end)" so the next round uses an empty + # tool set for it. + force_ask = False + round_num = 1 + while True: + if round_num > self._max_tool_rounds: + # P1 (rant 21:55:37): round budget exhausted but messages + # still queued — process them with a fresh round budget + # instead of stranding them; only fall back to the + # "Exceeded maximum" error when the queue is empty. + _n, _ask = await self._inject_pending_messages(session, messages) + if _n: + force_ask = _ask + round_num = 1 + continue + + # Exceeded max tool rounds + logger.warning("max tool rounds (%d) exceeded for task %s", + self._max_tool_rounds, req.id) + await self._broadcast(session.session_id, { + "request_id": req.id, + "content": f"Exceeded maximum tool call rounds ({self._max_tool_rounds}).", + "done": True, + "delta": False, + "session_id": session.session_id, + }) - for round_num in range(1, self._max_tool_rounds + 1): + # Fire-and-forget: reflect on whether to save memories + self._maybe_reflect_memory(session, req.prompt, full_content) + return + + tools_openai = tools_base # Check for cancellation between rounds if cancel_event and cancel_event.is_set(): logger.info("tool loop cancelled by client at round %d", round_num) @@ -1711,6 +1839,13 @@ async def _run_tool_loop( }) return + # P1 queue-injection: drain pending at the round boundary (after + # this round's tools, before the next LLM request — codex steer). + _injected, _ask = await self._inject_pending_messages(session, messages) + if _ask or force_ask: + tools_openai = [] + force_ask = False + logger.debug("tool loop round %d: %d messages, %d tools", round_num, len(messages), len(tools_openai)) @@ -1846,6 +1981,20 @@ async def _run_tool_loop( "content": full_content, }) + # Append the assistant reply to the local messages so the + # LLM context stays coherent when queued messages are + # injected after this round (mirrors Case 2's assistant + # tool_calls message). + messages.append({"role": "assistant", "content": full_content}) + + # P1 (rant 21:55:37): messages queued mid-round (after the + # round-top drain) must not end the turn — inject and continue. + # Injection round does not consume the round budget. + _n, _ask = await self._inject_pending_messages(session, messages) + if _n: + force_ask = _ask + continue + await self._broadcast(session.session_id, { "request_id": req.id, "content": "", @@ -1988,7 +2137,11 @@ async def _run_tool_loop( "error": result.error, }) - # Log LLM request for this round (before continuing) + # Log LLM request for this round (before continuing). + # Tool rounds consume the round budget; injection rounds do + # not (they `continue` from the stop / Case 3 branches + # without incrementing). + round_num += 1 continue # Case 3: Max tokens or other stop — done @@ -2003,6 +2156,19 @@ async def _run_tool_loop( "content": full_content, }) + # Append the assistant reply to the local messages so the LLM + # context stays coherent when queued messages are injected + # after this round. + messages.append({"role": "assistant", "content": full_content}) + + # P1 (rant 21:55:37): messages queued mid-round must not end the + # turn — inject and continue (injection round does not consume + # the round budget). + _n, _ask = await self._inject_pending_messages(session, messages) + if _n: + force_ask = _ask + continue + await self._broadcast(session.session_id, { "request_id": req.id, "content": full_content or "", @@ -2015,20 +2181,6 @@ async def _run_tool_loop( self._maybe_reflect_memory(session, req.prompt, full_content) return - # Exceeded max tool rounds - logger.warning("max tool rounds (%d) exceeded for task %s", - self._max_tool_rounds, req.id) - await self._broadcast(session.session_id, { - "request_id": req.id, - "content": f"Exceeded maximum tool call rounds ({self._max_tool_rounds}).", - "done": True, - "delta": False, - "session_id": session.session_id, - }) - - # Fire-and-forget: reflect on whether to save memories - self._maybe_reflect_memory(session, req.prompt, full_content) - def _log_llm_exchange( self, session: Session, messages, tools, content: str, finish_reason: str = "stop", diff --git a/tests/test_ws_e2e.py b/tests/test_ws_e2e.py index dc08e3f..35e1c9e 100644 --- a/tests/test_ws_e2e.py +++ b/tests/test_ws_e2e.py @@ -21,6 +21,7 @@ from emrg.config import LlmConfig from emrg.connect import connect_to_server +from emrg.server.tool_types import ToolResult def _make_config() -> LlmConfig: @@ -333,7 +334,9 @@ class TestWSBroadcast: """Phase 2 broadcast model (protocol-contract §2.6). Same session → all subscribed connections see the same streaming - response. Concurrent task on a busy session → 'session busy'. + response. Concurrent task on a busy session → queued (task_queued, + P1 rant 2026-08-10T21:55:37), injected at the next round boundary or + re-sent via queued_requeue when the turn ends normally. """ def test_broadcast_to_subscribers(self): @@ -384,8 +387,10 @@ async def _test(): await cleanup() asyncio.run(_test()) - def test_session_busy(self): - """A's task holds the session lock; B's task gets 'session busy'.""" + def test_task_queued_instead_of_busy_error(self): + """A's task holds the session lock; B's task is queued (task_queued, + NOT 'session busy'), then injected into A's turn at the stop boundary + (steer_committed) — the message is never lost.""" async def _test(): with tempfile.TemporaryDirectory() as tmp: cwd = Path(tmp) @@ -393,7 +398,7 @@ async def _test(): try: async def slow_chat_stream(messages, tools=None): yield {"content": "处理中", "tool_calls": None, "finish_reason": None, "usage": None} - await asyncio.sleep(0.8) + await asyncio.sleep(0.6) yield {"content": "完成", "tool_calls": None, "finish_reason": "stop", "usage": None} server.llm.chat_stream = slow_chat_stream ws_a = await connect_to_server() @@ -408,10 +413,73 @@ async def slow_chat_stream(messages, tools=None): await asyncio.sleep(0.2) # let A's task grab the lock task_b = {**task, "id": "t-busy-b"} await ws_b.send(json.dumps(task_b, ensure_ascii=False)) - frame = await asyncio.wait_for(ws_b.recv(), timeout=5) - resp = json.loads(frame) - assert resp.get("error") == "session busy" + # B must get task_queued (with position), NOT "session busy" + resp = json.loads(await asyncio.wait_for(ws_b.recv(), timeout=5)) + assert resp.get("type") == "task_queued", f"got {resp!r}" + assert resp.get("request_id") == "t-busy-b" assert resp.get("session_id") == "s_busy" + assert resp.get("position") == 1 + # A's stop branch injects B's message into the same turn + fsc = await _recv_until( + ws_b, lambda f: f.get("type") == "steer_committed", + what="steer_committed") + assert fsc.get("request_id") == "t-busy-b" + # A's turn (now containing B's message) completes + await _recv_until( + ws_a, + lambda f: f.get("done") and f.get("request_id") == "t-busy-a", + what="t-busy-a done") + finally: + await ws_a.close() + await ws_b.close() + finally: + await cleanup() + asyncio.run(_test()) + + def test_queued_requeue_on_normal_end(self): + """Turn ends with a message still queued (race window: it arrived + after the last injection drain) → the wrapper broadcasts + queued_requeue with the request_ids so clients re-send. White-box: + _inject_pending_messages is stubbed to never drain, forcing the + queue to survive until the wrapper's finally.""" + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + 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(0.5) + yield {"content": "完成", "tool_calls": None, "finish_reason": "stop", "usage": None} + server.llm.chat_stream = slow_chat_stream + async def no_drain(session, messages): + return 0, False + server._inject_pending_messages = no_drain # type: ignore[assignment] + ws_a = await connect_to_server() + ws_b = await connect_to_server() + try: + task = { + "type": "task", "id": "t-rq-a", "session_id": "s_rq", + "cwd": str(cwd), "prompt": "hi", "stream": True, + "timestamp": "2026-08-02T00:00:00", + } + await ws_a.send(json.dumps(task, ensure_ascii=False)) + await asyncio.sleep(0.2) + task_b = {**task, "id": "t-rq-b"} + await ws_b.send(json.dumps(task_b, ensure_ascii=False)) + resp = json.loads(await asyncio.wait_for(ws_b.recv(), timeout=5)) + assert resp.get("type") == "task_queued" + # Turn ends normally → queued_requeue carries the ids + requeue = await _recv_until( + ws_b, lambda f: f.get("type") == "queued_requeue", + what="queued_requeue") + assert "t-rq-b" in requeue.get("request_ids", []) + # Re-send now that the lock is released → normal execution + await ws_b.send(json.dumps(task_b, ensure_ascii=False)) + await _recv_until( + ws_b, + lambda f: f.get("done") and f.get("request_id") == "t-rq-b", + what="t-rq-b done") finally: await ws_a.close() await ws_b.close() @@ -561,3 +629,327 @@ async def _test(): finally: await cleanup() asyncio.run(_test()) + + +async def _recv_until(ws, pred, timeout=10, limit=50, what="frame"): + """Read frames (skipping unrelated broadcasts) until pred(frame) or fail.""" + for _ in range(limit): + frame = json.loads(await asyncio.wait_for(ws.recv(), timeout=timeout)) + if pred(frame): + return frame + raise AssertionError(f"expected {what} not received in {limit} frames") + + +class TestWSQueueInjection: + """P1 queue-injection (rant 2026-08-10T21:55:37, design doc + mid-turn-input-queue): messages sent while a session's tool loop is busy + are queued per session and injected at the next round boundary (after the + current round's LLM request + ALL tool executions, before the next LLM + request) — never interrupting tools, never losing messages. + + All clients subscribed to a session receive that session's broadcast + stream, so every read here filters for the expected frame type (queue + frames interleave with the active turn's deltas/tool frames). + """ + + @staticmethod + def _task(sid, tid, prompt, cwd, mode="auto"): + return { + "type": "task", "id": tid, "session_id": sid, + "cwd": str(cwd), "prompt": prompt, "stream": True, + "mode": mode, "timestamp": "2026-08-10T21:55:37", + } + + def test_pending_injected_at_round_boundary_after_tools(self): + """B's message queued while A's tool executes is injected at the + round boundary: round 2's LLM request sees it, the tool ran to + completion first (no interruption), and steer_committed is + broadcast. Nothing is lost.""" + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + server, _, cleanup = await _boot_server(cwd) + try: + loop = asyncio.get_running_loop() + seen = {"round2_user_texts": None, "round2_tools": None, + "t_inject_seen": None, "t_tool_end": None} + + class _SlowBash: + async def execute(self, args): + await asyncio.sleep(0.6) + seen["t_tool_end"] = loop.time() + return ToolResult(tool_call_id="call_1", name="bash", + content="hi", error=False) + + orig_get = server.tools.get + server.tools.get = lambda name: _SlowBash() if name == "bash" else orig_get(name) + + async def chat_stream(messages, tools=None): + user_texts = [m.get("content") for m in messages + if m.get("role") == "user"] + if any("steer-mid" in str(t) for t in user_texts): + seen["round2_user_texts"] = user_texts + seen["round2_tools"] = tools + seen["t_inject_seen"] = loop.time() + yield {"content": "收到", "tool_calls": None, + "finish_reason": "stop", "usage": None} + return + yield {"content": "开始", "tool_calls": None, + "finish_reason": None, "usage": None} + yield {"content": None, "tool_calls": [{ + "index": 0, "id": "call_1", + "function": {"name": "bash", + "arguments": '{"command":"echo hi"}'}, + }], "finish_reason": "tool_calls", "usage": None} + server.llm.chat_stream = chat_stream + + ws_a = await connect_to_server() + ws_b = await connect_to_server() + try: + await ws_a.send(json.dumps( + self._task("s_inj", "t-inj-a", "turn-one", cwd), ensure_ascii=False)) + await asyncio.sleep(0.15) # round 1 tool now executing + await ws_b.send(json.dumps( + self._task("s_inj", "t-inj-b", "steer-mid", cwd), ensure_ascii=False)) + + # B: task_queued then steer_committed + fq = await _recv_until( + ws_b, lambda f: f.get("type") == "task_queued", what="task_queued") + assert fq.get("position") == 1 + fsc = await _recv_until( + ws_b, lambda f: f.get("type") == "steer_committed", + what="steer_committed") + assert fsc.get("request_id") == "t-inj-b" + + # A: tool_end before done + tool_end_seen = done_seen = False + while not done_seen: + fa = json.loads(await asyncio.wait_for(ws_a.recv(), timeout=10)) + if fa.get("type") == "tool_end": + tool_end_seen = True + if fa.get("done"): + done_seen = True + assert tool_end_seen, "tool must run to completion" + + # round 2 LLM request received the injected message + assert seen["round2_user_texts"] is not None + assert any("steer-mid" in str(t) for t in seen["round2_user_texts"]) + # injection happened strictly after the tool finished + assert seen["t_inject_seen"] >= seen["t_tool_end"] - 0.05 + # tools preserved for auto-mode rounds + assert seen["round2_tools"] is not None + assert len(seen["round2_tools"]) > 0 + finally: + await ws_a.close() + await ws_b.close() + finally: + await cleanup() + asyncio.run(_test()) + + def test_pending_ask_injects_empty_tools(self): + """A queued Ask-mode message is injected with an empty tool set + (mode=ask → tools=[]), so the injected round can only reply.""" + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + server, _, cleanup = await _boot_server(cwd) + try: + loop = asyncio.get_running_loop() + seen = {"tools": "unset"} + + class _SlowBash: + async def execute(self, args): + await asyncio.sleep(0.5) + return ToolResult(tool_call_id="call_1", name="bash", + content="hi", error=False) + + orig_get = server.tools.get + server.tools.get = lambda name: _SlowBash() if name == "bash" else orig_get(name) + + async def chat_stream(messages, tools=None): + user_texts = [m.get("content") for m in messages + if m.get("role") == "user"] + if any("ask-me" in str(t) for t in user_texts): + seen["tools"] = tools + yield {"content": "仅回复", "tool_calls": None, + "finish_reason": "stop", "usage": None} + return + yield {"content": None, "tool_calls": [{ + "index": 0, "id": "call_1", + "function": {"name": "bash", + "arguments": '{"command":"echo hi"}'}, + }], "finish_reason": "tool_calls", "usage": None} + server.llm.chat_stream = chat_stream + + ws_a = await connect_to_server() + ws_b = await connect_to_server() + try: + await ws_a.send(json.dumps( + self._task("s_ask", "t-ask-a", "turn", cwd), ensure_ascii=False)) + await asyncio.sleep(0.15) + await ws_b.send(json.dumps( + self._task("s_ask", "t-ask-b", "ask-me", cwd, mode="ask"), + ensure_ascii=False)) + fq = await _recv_until( + ws_b, lambda f: f.get("type") == "task_queued", what="task_queued") + while seen["tools"] == "unset": + await asyncio.wait_for(ws_a.recv(), timeout=10) + assert seen["tools"] == [], f"ask round must have empty tools, got {seen['tools']}" + finally: + await ws_a.close() + await ws_b.close() + finally: + await cleanup() + asyncio.run(_test()) + + def test_queued_cancelled_on_cancel(self): + """A cancels mid-turn → the queued message is not lost silently: + clients get queued_cancelled (queue dropped, client can re-send).""" + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + server, _, cleanup = await _boot_server(cwd) + try: + async def long_stream(messages, tools=None): + yield {"content": "工作中", "tool_calls": None, + "finish_reason": None, "usage": None} + await asyncio.sleep(5) + yield {"content": "完成", "tool_calls": None, + "finish_reason": "stop", "usage": None} + server.llm.chat_stream = long_stream + ws_a = await connect_to_server() + ws_b = await connect_to_server() + try: + await ws_a.send(json.dumps( + self._task("s_can", "t-can-a", "long", cwd), ensure_ascii=False)) + await asyncio.sleep(0.2) + await ws_b.send(json.dumps( + self._task("s_can", "t-can-b", "queued", cwd), ensure_ascii=False)) + fq = await _recv_until( + ws_b, lambda f: f.get("type") == "task_queued", what="task_queued") + # A cancels + await ws_a.send(json.dumps({"type": "cancel", "session_id": "s_can"})) + # A gets cancelled frame + got_cancelled = False + while not got_cancelled: + fa = json.loads(await asyncio.wait_for(ws_a.recv(), timeout=10)) + if fa.get("type") == "cancelled": + got_cancelled = True + # B gets queued_cancelled (skipping A's cancelled-done broadcast) + fb = await _recv_until( + ws_b, lambda f: f.get("type") == "queued_cancelled", + what="queued_cancelled") + assert fb.get("session_id") == "s_can" + finally: + await ws_a.close() + await ws_b.close() + finally: + await cleanup() + asyncio.run(_test()) + + def test_clear_session_drops_pending(self): + """clear_session pops the session's pending queue and broadcasts + queued_cancelled (Change F, design doc).""" + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + server, _, cleanup = await _boot_server(cwd) + try: + async def long_stream(messages, tools=None): + yield {"content": "工作中", "tool_calls": None, + "finish_reason": None, "usage": None} + await asyncio.sleep(1.0) + yield {"content": "完成", "tool_calls": None, + "finish_reason": "stop", "usage": None} + server.llm.chat_stream = long_stream + ws_a = await connect_to_server() + ws_b = await connect_to_server() + try: + await ws_a.send(json.dumps( + self._task("s_drop", "t-drop-a", "long", cwd), ensure_ascii=False)) + await asyncio.sleep(0.2) + await ws_b.send(json.dumps( + self._task("s_drop", "t-drop-b", "queued", cwd), ensure_ascii=False)) + fq = await _recv_until( + ws_b, lambda f: f.get("type") == "task_queued", what="task_queued") + assert len(server._session_pending.get("s_drop", [])) == 1 + # B clears the session while its message is queued + await ws_b.send(json.dumps({ + "type": "clear_session", "session_id": "s_drop", "cwd": str(cwd), + })) + got_clear = got_qc = False + while not (got_clear and got_qc): + fb = json.loads(await asyncio.wait_for(ws_b.recv(), timeout=10)) + if fb.get("type") == "clear_result" and fb.get("ok"): + got_clear = True + if fb.get("type") == "queued_cancelled": + got_qc = True + assert got_clear and got_qc + assert server._session_pending.get("s_drop") in (None, []) + # A's turn ends normally → no requeue (queue was dropped) + got_done = False + while not got_done: + fa = json.loads(await asyncio.wait_for(ws_a.recv(), timeout=10)) + if fa.get("done"): + got_done = True + assert server._session_pending.get("s_drop") in (None, []) + finally: + await ws_a.close() + await ws_b.close() + finally: + await cleanup() + asyncio.run(_test()) + + def test_per_session_isolation(self): + """The pending queue is per-session: a task on a different session + runs immediately (not queued); a task on the busy session queues.""" + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + cwd = Path(tmp) + server, _, cleanup = await _boot_server(cwd) + try: + async def chat_stream(messages, tools=None): + user_texts = [m.get("content") for m in messages + if m.get("role") == "user"] + if any("iso2" in str(t) for t in user_texts): + yield {"content": "iso2-done", "tool_calls": None, + "finish_reason": "stop", "usage": None} + return + yield {"content": "iso1-start", "tool_calls": None, + "finish_reason": None, "usage": None} + await asyncio.sleep(1.2) + yield {"content": "iso1-done", "tool_calls": None, + "finish_reason": "stop", "usage": None} + server.llm.chat_stream = chat_stream + ws_a = await connect_to_server() + ws_b = await connect_to_server() + try: + # A busy on s_iso1 + await ws_a.send(json.dumps( + self._task("s_iso1", "t-iso-a", "iso1 long", cwd), ensure_ascii=False)) + await asyncio.sleep(0.2) + # B on s_iso2 → NOT busy → immediate streaming, no task_queued + await ws_b.send(json.dumps( + self._task("s_iso2", "t-iso-b2", "iso2 fast", cwd), ensure_ascii=False)) + got_done = False + saw_queued = False + while not got_done: + fb = json.loads(await asyncio.wait_for(ws_b.recv(), timeout=10)) + if fb.get("type") == "task_queued": + saw_queued = True + if fb.get("done"): + got_done = True + assert not saw_queued, "different session must not be queued" + assert got_done + # B on s_iso1 (busy) → queued + await ws_b.send(json.dumps( + self._task("s_iso1", "t-iso-b1", "iso1 queued", cwd), ensure_ascii=False)) + fq = await _recv_until( + ws_b, lambda f: f.get("type") == "task_queued", what="task_queued") + assert fq.get("session_id") == "s_iso1" + finally: + await ws_a.close() + await ws_b.close() + finally: + await cleanup() + asyncio.run(_test())