fix(cli): decline questions from interrupted turns

opencollab/adapters/cli/main.py

@@ -13,6 +13,7 @@
 import sys
 import uuid
 from collections.abc import Awaitable
+from contextvars import ContextVar
 from typing import Any, Optional
 
 import typer
@@ -366,10 +367,18 @@ async def _run(
     # ``--yolo`` only auto-approves risky commands; a human is still present, so
     # the ask-user tool stays interactive — its question takes the shared input
     # line ahead of whatever the user was typing.
+    turn_cancel: ContextVar[asyncio.Event | None] = ContextVar("cli_turn_cancel", default=None)
+
+    async def read_question(text: str) -> str:
+        cancel = turn_cancel.get()
+        if cancel is not None and cancel.is_set():
+            raise EOFError
+        return await prompt.ask(text)
+
     permission_policy = None
     if not yolo:
-        permission_policy = TuiPermissionPolicy(read_line=prompt.ask)
-    ask_policy = TuiAskUserPolicy(read_line=prompt.ask)
+        permission_policy = TuiPermissionPolicy(read_line=read_question)
+    ask_policy = TuiAskUserPolicy(read_line=read_question)
 
     event_sink = TuiEventSink(tui)
     events_file = os.environ.get("OPENCOLLAB_EVENTS_FILE")
@@ -418,13 +427,15 @@ async def turn(line: str, target_aid: int, cancel: asyncio.Event) -> None:
             tui.record_user_message(target_aid, line)
             tui.start_turn(target_aid)
             failure: SchedulerTurnError | None = None
+            cancel_token = turn_cancel.set(cancel)
             try:
                 await scheduler.run_turn(target_aid, line, cancel_event=cancel)
             except SchedulerTurnError as exc:
                 failure = exc
             except KeyboardInterrupt:
                 pass
             finally:
+                turn_cancel.reset(cancel_token)
                 # Flush the answering agent's unprinted tail when focus moved to a teammate.
                 # Switching focus would reprint rows already in scrollback.
                 tui.settle_turn(

tests/interfaces/test_cli_queued_focus.py

@@ -235,3 +235,108 @@ async def respond(request):
 
     assert tui.selected_aid == 1
     assert output.getvalue().count("Final lead answer after inspection") == 1
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("question_count", "interrupted"), [(2, True), (1, True), (2, False)])
+async def test_question_batch_releases_the_input_after_interrupt(
+    monkeypatch, tmp_path, question_count, interrupted,
+):
+    tui, output, team, cfg = _install_cli(monkeypatch, tmp_path)
+    team.write_text(team.read_text().replace(
+        "tools: [spawn_agent]", "tools: [spawn_agent, ask_user, file_write]",
+    ))
+    entered, release = asyncio.Event(), asyncio.Event()
+    requests = []
+
+    def tool(name, arguments, call_id):
+        return {"id": call_id, "type": "function", "function": {
+            "name": name, "arguments": json.dumps(arguments),
+        }}
+
+    async def respond(request):
+        body = json.loads(request.content)
+        requests.append(body)
+        if "OFFLINE_ANALYST_ROLE" in body["messages"][0]["content"]:
+            return completion_http_response(request, text="Analyst completed")
+        user = next(message["content"] for message in reversed(body["messages"]) if message["role"] == "user")
+        if body["messages"][-1]["role"] != "user":
+            return completion_http_response(request, text="Lead completed")
+        if "setup_team" in user:
+            calls = [tool("spawn_agent", {"role": "analyst", "task": "initial_child_task"}, "spawn")]
+        elif "ask_batch" in user:
+            entered.set()
+            await release.wait()
+            calls = [tool("ask_user", {"question": f"Question {index}?"}, f"question_{index}")
+                     for index in range(question_count)]
+            calls.append(tool("file_write", {
+                "path": "batch-completed.txt", "mode": "create", "content": "Batch completed",
+            }, "write"))
+        else:
+            return completion_http_response(request, text="Lead completed")
+        return http_response(request, 200, json={
+            "id": "chatcmpl_questions", "object": "chat.completion", "created": 1, "model": body["model"],
+            "choices": [{"index": 0, "finish_reason": "tool_calls", "message": {
+                "role": "assistant", "content": None, "tool_calls": calls,
+            }}],
+            "usage": {"prompt_tokens": 20, "completion_tokens": 10, "total_tokens": 30},
+        })
+
+    install_sdk_transport(monkeypatch, "openai", respond)
+    with create_pipe_input() as keyboard:
+        session = PromptSession(input=keyboard, output=DummyOutput(), erase_when_done=True)
+        _pin_status_row_under_input(session)
+        def prompt_session(prompt):
+            prompt._session = session
+            return session
+
+        monkeypatch.setattr(LivePrompt, "_ensure_session", prompt_session)
+        run = asyncio.create_task(cli_main._run(
+            str(tmp_path), cfg, None, False, True, False, team_config_path=str(team),
+        ))
+        try:
+            await _wait_for(lambda: session.app.is_running)
+            keyboard.send_text("setup_team\r")
+            await _wait_for(lambda: _snapshot(tmp_path, 0)
+                            and _snapshot(tmp_path, 0)["session_state"]["phase"] == "done")
+            keyboard.send_text("ask_batch\r")
+            await asyncio.wait_for(entered.wait(), timeout=5)
+            draft = "Continue the draft café"
+            keyboard.send_text(draft)
+            await _wait_for(lambda: session.default_buffer.text == draft)
+            release.set()
+            await _wait_for(lambda: "Question 0?" in "".join(text for _, text in session.message()))
+            keyboard.send_text("\t")
+            await _wait_for(lambda: tui.selected_aid == 1)
+            if interrupted:
+                keyboard.send_bytes(b"\x03")
+                await _wait_for(lambda: "Question 1?" in "".join(text for _, text in session.message())
+                                or _snapshot(tmp_path, 0)["session_state"]["phase"] == "stopped")
+                assert "Question 1?" not in "".join(text for _, text in session.message())
+            else:
+                keyboard.send_text("/help\r")
+                await _wait_for(lambda: "Question 1?" in "".join(text for _, text in session.message()))
+                keyboard.send_text("src/第二个.py\r")
+            phase = "stopped" if interrupted else "done"
+            await _wait_for(lambda: _snapshot(tmp_path, 0)["session_state"]["phase"] == phase)
+            await _wait_for(lambda: session.app.is_running and session.default_buffer.text == draft)
+            keyboard.send_text("\r")
+            await _wait_for(lambda: _snapshot(tmp_path, 1)["session_state"]["phase"] == "done"
+                            and any(draft in (message.get("content") or "")
+                                    for message in _snapshot(tmp_path, 1)["messages"]))
+            keyboard.send_text("/exit\r")
+            await asyncio.wait_for(run, timeout=5)
+        finally:
+            release.set()
+            if not run.done():
+                run.cancel()
+            await asyncio.gather(run, return_exceptions=True)
+
+    messages = _snapshot(tmp_path, 0)["messages"]
+    answers = {message["tool_call_id"]: message["content"] for message in messages
+               if message["role"] == "tool" and message["tool_call_id"].startswith("question_")}
+    assert answers == ({f"question_{index}": "User declined to answer." for index in range(question_count)}
+                       if interrupted else {"question_0": "/help", "question_1": "src/第二个.py"})
+    assert (tmp_path / "batch-completed.txt").read_text() == "Batch completed"
+    assert "Interactive team help" not in output.getvalue()
+    assert sum("OFFLINE_ANALYST_ROLE" in body["messages"][0]["content"]
+               and draft in body["messages"][-1]["content"] for body in requests) == 1