diff --git a/src/agent/state.rv b/src/agent/state.rv index 2bb59d5..fc2aa5b 100644 --- a/src/agent/state.rv +++ b/src/agent/state.rv @@ -346,7 +346,19 @@ impl AgentState { // ask with extra lines (a diff preview) shown under the label. fun ask_detail(self, desc: String, detail: String) -> Int { + // Esc already asked to stop: answer deny without showing a prompt. + // Askers queue on ask_lock, so with parallel subagents running the + // user would otherwise have to dismiss one prompt per waiting caller + // before the turn actually ended. + if self.interrupted() { + return 0 + } self.ask_lock.lock() + // The interrupt may have landed while this caller waited its turn. + if self.interrupted() { + self.ask_lock.unlock() + return 0 + } self.lock.lock() self.pending = desc self.pending_detail = detail @@ -381,7 +393,15 @@ impl AgentState { // ask_lock and decision channel with approvals, so only one prompt is // in front of the user at a time. fun ask_question(self, q: String, options: List) -> Int { + // Declined without prompting once Esc has asked the turn to stop. + if self.interrupted() { + return 0 - 1 + } self.ask_lock.lock() + if self.interrupted() { + self.ask_lock.unlock() + return 0 - 1 + } self.lock.lock() self.question = q self.choices = options diff --git a/src/agent/state_test.rv b/src/agent/state_test.rv index e4011e7..bc3eb64 100644 --- a/src/agent/state_test.rv +++ b/src/agent/state_test.rv @@ -118,6 +118,25 @@ fun test_interrupt_cancels_the_stream_token() { assert_false(st.token().canceled()) } +// Once Esc has asked the turn to stop, an approval is denied outright rather +// than queued in front of the user. Both calls return without blocking on a +// decision, which is what lets a turn with parallel agents actually end on +// one Esc instead of one per waiting caller. +fun test_approval_is_denied_without_prompting_after_an_interrupt() { + let st = AgentState.new() + st.request_interrupt() + assert_eq_int(st.ask_detail("write_file(a.txt)", ""), 0) + assert_eq_str(st.pending_now(), "") +} + +fun test_question_is_declined_without_prompting_after_an_interrupt() { + let st = AgentState.new() + st.request_interrupt() + let options: List = ["keep", "drop"] + assert_eq_int(st.ask_question("which?", options), 0 - 1) + assert_eq_str(st.question_now(), "") +} + fun test_pending_is_readable_without_draining_events() { let st = AgentState.new() st.emit(DisplayEvent.Started) diff --git a/src/agent/subturn.rv b/src/agent/subturn.rv index 53d61f2..1630429 100644 --- a/src/agent/subturn.rv +++ b/src/agent/subturn.rv @@ -112,6 +112,12 @@ fun run_subagent( history.push(Message.assistant_calls(reply_text, calls)) let i = 0 while i < calls.len() { + // Esc between the calls of one batch stops the subagent here, so + // the rest of the batch does not run after the user asked to stop. + if st.interrupted() { + st.emit(DisplayEvent.AgentFinished(id, agent.name, used)) + return "error: interrupted" + } let call = calls[i] let result = _sub_one(st, agent, call, allowed) st.emit(DisplayEvent.AgentTool(id, agent.name, label(call), _short(result))) diff --git a/src/agent/subturn_test.rv b/src/agent/subturn_test.rv index 581d934..d9815b4 100644 --- a/src/agent/subturn_test.rv +++ b/src/agent/subturn_test.rv @@ -156,3 +156,30 @@ fun test_subagent_write_approval_shows_and_records_the_preview() { remove_file(path) return } + +// A subagent's mutating call after Esc is denied on the spot: it must not put +// a fresh approval prompt in front of a user who already asked to stop, and +// the write must not happen. Runs inline, since nothing should block. +fun test_subagent_mutation_after_an_interrupt_is_denied_without_prompting() { + reset_hooks() + let path = "_rook_sub_interrupted.txt" + if exists(path) { + remove_file(path) + } + let st = AgentState.new() + st.request_interrupt() + let allowed: List = [] + let result = _sub_one_with_hooks( + st, + full_agent("worker"), + call("write_file", "{\"path\":\"_rook_sub_interrupted.txt\",\"content\":\"changed\"}"), + allowed, + allowing_pre_hook, + record_post_hook, + ask_permission, + ) + assert_true(result.contains("denied")) + assert_eq_str(st.pending_now(), "") + assert_true(!exists(path)) + return +} diff --git a/src/agent/turn.rv b/src/agent/turn.rv index 4e2749d..701c2ef 100644 --- a/src/agent/turn.rv +++ b/src/agent/turn.rv @@ -127,27 +127,29 @@ fun run_turn(st: AgentState, model: String, api_key: String, base_url: String) { // in one batch fan out to concurrent goroutines; everything else // runs in order on this one. st.push_assistant_calls(reply_text, calls) - // Esc may have landed while the model was streaming this tool request. - // Skip running the batch so a tool the user pre-approved this turn does - // not execute after they asked to stop, but still record a result for - // each call so the conversation stays balanced for a resume. - if st.interrupted() { - let c = 0 - while c < calls.len() { - st.push_tool_result(calls[c].id, calls[c].name, "(interrupted by user before running)") - c = c + 1 - } - st.emit(DisplayEvent.Info("Interrupted")) - _end_turn(st) - return - } let fanout = _count_dispatch(calls) >= 2 let fanout_results = ToolResults.new(calls.len()) - if fanout { + // Nothing is about to run if Esc already landed while the model was + // streaming this request, so do not announce a fan-out that will not + // happen. + if fanout && !st.interrupted() { st.emit(DisplayEvent.Info("running ${_count_dispatch(calls)} agents in parallel…")) } let i = 0 + // The index of the first call left unrun by an Esc, -1 when the whole + // batch ran and 0 when the stop arrived before any call started. + let stopped = 0 - 1 while i < calls.len() { + // Esc can land between the calls of one batch, not just before it. + // Without this the rest of the batch still runs: read-only calls + // silently, and a mutating one puts another approval prompt in + // front of a user who already asked to stop. Checking here rather + // than before the loop keeps the interrupted bookkeeping in one + // place below, since a stop that beat the batch is just stopped=0. + if st.interrupted() { + stopped = i + break + } let call = calls[i] if fanout && call.name == "dispatch" { let idx = i @@ -177,6 +179,31 @@ fun run_turn(st: AgentState, model: String, api_key: String, base_url: String) { } i = i + 1 } + // Esc stopped the batch part way. Wait only for the calls that were + // actually started (a spawned agent still owns its slot), then record a + // result for each one that never ran so the conversation stays balanced + // for a resume, and end the turn. + if stopped >= 0 { + if fanout { + while fanout_results.count() < stopped { + sleep_millis(20) + } + let d = 0 + while d < stopped { + st.push_tool_result(calls[d].id, calls[d].name, fanout_results.get(d)) + d = d + 1 + } + } + let c = stopped + while c < calls.len() { + st.push_tool_result(calls[c].id, calls[c].name, "(interrupted by user before running)") + c = c + 1 + } + st.emit(DisplayEvent.Info("Interrupted")) + _report_usage(st, model) + _end_turn(st) + return + } // Every agent's result must be in the conversation before the next // round's request is built. if fanout { diff --git a/src/agent/turn_e2e_test.rv b/src/agent/turn_e2e_test.rv index 9235d0e..5f0dbf8 100644 --- a/src/agent/turn_e2e_test.rv +++ b/src/agent/turn_e2e_test.rv @@ -18,6 +18,8 @@ let fanout_hits = 0 let ask_hits = 0 let perm_hits = 0 let plan_hits = 0 +let batch_hits = 0 +let fanout_stop_hits = 0 // Round 1: ask to read a file. Round 2: answer. fun read_handler(req: Request) -> Response { @@ -77,6 +79,56 @@ fun serve_plan() { app.listen("127.0.0.1:8469") } +// Round 1: two write_file calls in one batch, so Esc can land between them. +fun batch_handler(req: Request) -> Response { + batch_hits = batch_hits + 1 + if batch_hits == 1 { + return Response.text( + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"b1\",\"function\":{\"name\":\"write_file\",\"arguments\":\"{\\\"path\\\":\\\"_rook_e2e_batch_a.txt\\\",\\\"content\\\":\\\"a\\\"}\"}},{\"index\":1,\"id\":\"b2\",\"function\":{\"name\":\"write_file\",\"arguments\":\"{\\\"path\\\":\\\"_rook_e2e_batch_b.txt\\\",\\\"content\\\":\\\"b\\\"}\"}}]}}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\ndata: [DONE]\n\n", + ) + } + return Response.text( + "data: {\"choices\":[{\"delta\":{\"content\":\"Wrote both files.\"}}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n", + ) +} + +fun serve_batch() { + let app = Server.new() + app.post("/v1/chat/completions", batch_handler) + app.listen("127.0.0.1:8470") +} + +// Round 1: dispatch, write_file, dispatch. The write in the middle needs +// approval, which gives Esc a deterministic place to land with one agent +// already spawned and one dispatch still ahead of it. +fun fanout_stop_handler(req: Request) -> Response { + if req.body.contains("echo agent") { + return Response.text( + "data: {\"choices\":[{\"delta\":{\"content\":\"echo done\"}}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n", + ) + } + if req.body.contains("shout agent") { + return Response.text( + "data: {\"choices\":[{\"delta\":{\"content\":\"SHOUT DONE\"}}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n", + ) + } + fanout_stop_hits = fanout_stop_hits + 1 + if fanout_stop_hits == 1 { + return Response.text( + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"g1\",\"function\":{\"name\":\"dispatch\",\"arguments\":\"{\\\"agent\\\":\\\"echoer\\\",\\\"task\\\":\\\"say it\\\"}\"}},{\"index\":1,\"id\":\"g2\",\"function\":{\"name\":\"write_file\",\"arguments\":\"{\\\"path\\\":\\\"_rook_e2e_fanstop.txt\\\",\\\"content\\\":\\\"x\\\"}\"}},{\"index\":2,\"id\":\"g3\",\"function\":{\"name\":\"dispatch\",\"arguments\":\"{\\\"agent\\\":\\\"shouter\\\",\\\"task\\\":\\\"shout it\\\"}\"}}]}}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\ndata: [DONE]\n\n", + ) + } + return Response.text( + "data: {\"choices\":[{\"delta\":{\"content\":\"All three ran.\"}}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n", + ) +} + +fun serve_fanout_stop() { + let app = Server.new() + app.post("/v1/chat/completions", fanout_stop_handler) + app.listen("127.0.0.1:8471") +} + // One server plays both sides of a dispatch: the subagent's requests carry // its own system prompt ("echo agent") and get a plain non-streaming // answer; the parent's streaming requests get a dispatch call first, then @@ -419,6 +471,110 @@ fun test_plan_mode_blocks_a_write() { return } +// Esc at the approval prompt for the first of two writes in one batch must +// stop the whole batch: the second write neither runs nor re-prompts. +fun test_esc_at_an_approval_stops_the_rest_of_the_batch() { + spawn(fun() -> Unit { + serve_batch() + }) + sleep_millis(200) + let st = AgentState.new() + st.push_user("write both files") + st.begin() + spawn(fun() -> Unit { + run_turn(st, "custom/local", "", "http://127.0.0.1:8470/v1") + }) + // Play the user pressing Esc at the prompt: deny, then interrupt. + _interrupt_when_pending(st) + let got = e2e_drain(st) + assert_true(got.contains("[info Interrupted]")) + // Neither file exists: the first was denied, the second never ran. + assert_true(!exists("_rook_e2e_batch_a.txt")) + assert_true(!exists("_rook_e2e_batch_b.txt")) + // The second call was never even offered for approval. + assert_true(!got.contains("write_file(_rook_e2e_batch_b.txt)")) + // Every call still carries a tool result, so the history stays balanced + // for a resume: user, assistant(two tool_calls), two tool_results. + let msgs = st.messages() + assert_eq_int(msgs.len(), 4) + assert_eq_str(msgs[2].tool_call_id, "b1") + assert_eq_str(msgs[3].tool_call_id, "b2") + assert_true(msgs[3].content.contains("interrupted by user")) + return +} + +// Esc mid fan-out: the batch is dispatch, write_file, dispatch, so the stop +// lands at the write's approval with the first agent already spawned and the +// second dispatch still ahead. The turn must wait for the agent it started +// (rather than for a slot nothing will ever fill), skip the one it did not, +// and still leave every call with a result. +fun test_esc_mid_fanout_waits_only_for_started_agents() { + create_dir_all(".agents/subagents") + fs_write( + ".agents/subagents/echoer.md", + "---\ndescription: echoes things\n---\nYou are the echo agent.", + ) + fs_write( + ".agents/subagents/shouter.md", + "---\ndescription: shouts things\n---\nYou are the shout agent.", + ) + spawn(fun() -> Unit { + serve_fanout_stop() + }) + sleep_millis(200) + let st = AgentState.new() + st.push_user("run all three") + st.begin() + spawn(fun() -> Unit { + run_turn(st, "custom/local", "", "http://127.0.0.1:8471/v1") + }) + _interrupt_when_pending(st) + // Reaching the end of the drain at all is the point: a stopped fan-out + // that waited on the unstarted dispatch's slot would block here forever. + let got = e2e_drain(st) + assert_true(got.contains("[info Interrupted]")) + // The first agent was already running when Esc landed; the second was + // never started, and the denied write never happened. + assert_true(got.contains("[agent echoer started]")) + assert_true(!got.contains("[agent shouter started]")) + assert_true(!exists("_rook_e2e_fanstop.txt")) + // All three calls carry a result, in call order: the started agent's (its + // own answer or an interrupted one, depending on where the cancel landed), + // the denied write, then the placeholder for the dispatch that never ran. + let msgs = st.messages() + assert_eq_int(msgs.len(), 5) + assert_eq_str(msgs[2].tool_call_id, "g1") + assert_eq_str(msgs[3].tool_call_id, "g2") + assert_eq_str(msgs[4].tool_call_id, "g3") + assert_true(msgs[4].content.contains("interrupted by user")) + remove_file(".agents/subagents/echoer.md") + remove_file(".agents/subagents/shouter.md") + remove_dir(".agents/subagents") + remove_dir(".agents") + return +} + +// Poll for a pending approval and answer it the way Esc does: deny the call +// and ask the turn to stop. +fun _interrupt_when_pending(st: AgentState) { + spawn(fun() -> Unit { + let tries = 0 + let answered = false + while tries < 400 && !answered { + if st.pending_now() != "" { + // Same order the UI uses: raise the flag, then release the + // goroutine, so it cannot slip past the stop. + st.request_interrupt() + st.decide(0) + answered = true + } + sleep_millis(10) + tries = tries + 1 + } + }) + return +} + // Poll for a pending approval and allow it once, the way the UI would. Uses // pending_now so it never drains display events out from under e2e_drain. fun _approve_when_pending(st: AgentState) { diff --git a/src/app/update.rv b/src/app/update.rv index 3c87da7..1236986 100644 --- a/src/app/update.rv +++ b/src/app/update.rv @@ -902,10 +902,13 @@ fun _question_key(m: Chat, k: Key) -> Step { }, Esc -> { // Match the approval prompt: Esc backs out of the whole turn, not - // just the question, so an interrupt never feels swallowed. + // just the question, so an interrupt never feels swallowed. Raise + // the interrupt before answering: decide() unblocks the turn + // goroutine, and if the flag is not set yet it can get through the + // next round or tool call before it sees the stop. + m.agent.request_interrupt() m.agent.decide(0 - 1) m.question = "" - m.agent.request_interrupt() }, _ -> {}, } @@ -935,9 +938,12 @@ fun _approval_key(m: Chat, k: Key) -> Step { m.pending = "" }, Esc -> { + // Interrupt first, then answer: decide() releases the waiting turn + // goroutine immediately, so setting the flag afterwards races with + // it and the turn can run another tool call before noticing. + m.agent.request_interrupt() m.agent.decide(0) m.pending = "" - m.agent.request_interrupt() }, _ -> {}, }