@sigitcodereview / sigit / commits / b698202

fix(acp): end a turn stopped at a permission prompt when its session closes

session/close signals the session's running turn and then waits for that turn's session lock. A turn waiting on a permission answer never looked at the signal, so a client that closed the thread without answering the request left the close waiting for good, and every later request for that thread behind it. session/cancel had the same problem with a client that does not send the cancelled outcome. The permission wait now ends on the cancellation too. The flag is read again once the turn has the workspace back, so an approval that crossed the cancellation on the wire runs nothing. resume_workspace cancels the turn when its session cannot be reinstalled. It used to log a warning and let the turn go on in whichever directory was current. Nothing makes the reinstall fail today, because close removes the session only after the turn has let go of the session lock, but the turn no longer depends on that. A SessionGuard replaces the lock and guard pair in the seven handlers that take a session's lock. On drop it removes the session's entry from session_locks when the session is gone and no other request is queued on the lock. The map used to keep an entry for every closed thread and for every id a request named without a session behind it. Tests: close while the turn is mid-stream (this passed before the change and is there to hold the lock ordering), close at a permission prompt (timed out before), the lock entry's lifetime with a request queued behind the close, and a turn whose session is gone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Seto Elkahfi committed Oct 7, 2026 at 00:48 UTC b6982025195fc1327f8e63c4a59a49d609a0bcb3

3 files changed +359 -31

CHANGELOG.md

+28

@@ -1,5 +1,33 @@

1 # Changelog

3 +## Unreleased

4 +

5 +### Changed

6 +

7 +- **A turn that cannot get its session back stops.** A turn on an HTTP backend

8 + gives up the process-wide working directory while it waits and reinstalls

9 + its session before a tool runs. If the reinstall failed, the turn logged a

10 + warning and carried on in whichever directory was current. Nothing makes it

11 + fail today, since a close waits for the turn, but the turn now ends as

12 + cancelled instead of depending on that.

13 +

14 +### Fixed

15 +

16 +- **Closing a thread that is waiting at a permission prompt no longer hangs.**

17 + `session/close` cancels the session's running turn and waits for it to end,

18 + but a turn stopped at a permission prompt only went on when the client

19 + answered the request. A client that closed the thread without answering

20 + left the close, and every later request for that thread, waiting for good.

21 + The same held for `session/cancel` with a client that did not send the

22 + cancelled outcome the protocol asks for. The permission wait now ends on

23 + the cancellation as well, and an approval that arrives after the turn was

24 + cancelled runs nothing.

25 +

26 +- **A closed thread no longer leaves its request lock behind.** The process

27 + kept one lock per session id for as long as it ran, for closed threads and

28 + for ids no session ever had. The entry now goes with the last request that

29 + used it.

30 +

31 ## 1.6.4

33 ### Added

src/main.rs

+181 -28

@@ -416,6 +416,17 @@ struct PromptCancellation {

416 notify: tokio::sync::Notify,

417 }

418

419 +impl PromptCancellation {

420 + /// Resolves once the turn has been cancelled. `notify_one` leaves a permit

421 + /// behind when nobody is waiting, so a signal sent just before this is

422 + /// polled is not lost.

423 + async fn cancelled(&self) {

424 + while !self.cancelled.load(Ordering::Acquire) {

425 + self.notify.notified().await;

426 + }

427 + }

428 +}

429 +

430 enum DrainTurnError {

431 Backend(backend::BackendError),

432 Cancelled,

@@ -1024,6 +1035,39 @@ impl WorkspaceHold {

1035 }

1036 }

1037

1038 +/// One session's lock, held for a request on it.

1039 +///

1040 +/// Dropping it also drops the session's entry in `session_locks` when the

1041 +/// request leaves no such session behind, which is what `session/close` does

1042 +/// and what any request naming an id nobody opened does.

1043 +struct SessionGuard {

1044 + guard: tokio::sync::OwnedMutexGuard<()>,

1045 + agent: Arc<SiGitAgent>,

1046 + key: String,

1047 +}

1048 +

1049 +impl Drop for SessionGuard {

1050 + fn drop(&mut self) {

1051 + if self.agent.sessions.lock().unwrap().contains_key(&self.key) {

1052 + return;

1053 + }

1054 + let held = tokio::sync::OwnedMutexGuard::mutex(&self.guard);

1055 + let mut locks = self.agent.session_locks.lock().unwrap();

1056 + // Two handles mean the map's and this guard's. A request still queued

1057 + // on the lock holds a third, and then the entry stays: taking it out

1058 + // would give the next request for this id a fresh lock, and the two

1059 + // would no longer wait for each other. The queued request drops the

1060 + // entry when its own guard goes. `lock_session` clones under this

1061 + // mutex, so the count cannot grow before the entry is removed.

1062 + if locks

1063 + .get(&self.key)

1064 + .is_some_and(|lock| Arc::ptr_eq(lock, held) && Arc::strong_count(lock) == 2)

1065 + {

1066 + locks.remove(&self.key);

1067 + }

1068 + }

1069 +}

1070 +

1071 struct SiGitAgent {

1072 engine: Arc<ChatEngine>,

1073 /// The active inference backend. `LocalBackend` by default; swapped to an

@@ -1071,7 +1115,8 @@ struct SiGitAgent {

1115 workspace_lock: Arc<tokio::sync::Mutex<()>>,

1116 /// One lock per session id, held for a whole request on that session, so

1117 /// two requests on the same thread keep the order they arrived in. Always

1074 - /// taken before `workspace_lock`, never while holding it.

1118 + /// taken before `workspace_lock`, never while holding it. A closed

1119 + /// session's entry is dropped by `SessionGuard`.

1120 session_locks: std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Mutex<()>>>>,

1121 }

1122

@@ -2014,15 +2059,21 @@ impl SiGitAgent {

2059 }

2060 }

2061

2017 - /// The lock that orders requests on one session (see `session_locks`).

2018 - fn session_lock(&self, session_id: &SessionId) -> Arc<tokio::sync::Mutex<()>> {

2019 - Arc::clone(

2062 + /// Take the lock that orders requests on one session (see `session_locks`).

2063 + async fn lock_session(self: &Arc<Self>, session_id: &SessionId) -> SessionGuard {

2064 + let key = session_id.to_string();

2065 + let lock = Arc::clone(

2066 self.session_locks

2067 .lock()

2068 .unwrap()

2023 - .entry(session_id.to_string())

2069 + .entry(key.clone())

2070 .or_default(),

2025 - )

2071 + );

2072 + SessionGuard {

2073 + guard: lock.lock_owned().await,

2074 + agent: Arc::clone(self),

2075 + key,

2076 + }

2077 }

2078

2079 /// Keep the active session's parked metadata in step with a successful

@@ -2121,14 +2172,26 @@ impl SiGitAgent {

2172

2173 /// Take `workspace_lock` back after a turn gave it up, and make the turn's

2174 /// session the installed one again if another was installed meanwhile.

2124 - async fn resume_workspace(&self, workspace: &mut WorkspaceHold, session_id: &SessionId) {

2175 + ///

2176 + /// A turn whose session cannot be reinstalled is cancelled: whatever it ran

2177 + /// next would run in another session's directory and roots. The turn holds

2178 + /// its session's lock, which `session/close` waits for, so nothing takes

2179 + /// the session away today. The turn does not rely on that.

2180 + async fn resume_workspace(

2181 + &self,

2182 + workspace: &mut WorkspaceHold,

2183 + session_id: &SessionId,

2184 + cancellation: &PromptCancellation,

2185 + ) {

2186 if workspace.reacquire().await

2187 && let Err(error) = self.activate_session(session_id).await

2188 {

2189 log::warn!(

2129 - "session({session_id}): could not reinstall mid-turn: {}",

2190 + "session({session_id}): could not reinstall mid-turn, cancelling the turn: {}",

2191 error.message

2192 );

2193 + cancellation.cancelled.store(true, Ordering::Release);

2194 + cancellation.notify.notify_one();

2195 }

2196 }

2197

@@ -2969,7 +3032,8 @@ impl SiGitAgent {

3032 &cancellation,

3033 )

3034 .await;

2972 - self.resume_workspace(&mut workspace, &session_id).await;

3035 + self.resume_workspace(&mut workspace, &session_id, &cancellation)

3036 + .await;

3037 let mut result = match outcome {

3038 Ok(result) => result,

3039 Err(DrainTurnError::Cancelled) => {

@@ -3158,16 +3222,28 @@ impl SiGitAgent {

3222 if release_while_waiting {

3223 workspace.release();

3224 }

3161 - let verdict = self

3162 - .request_tool_permission(

3225 + // A client that cancels the turn answers the

3226 + // request itself. One that closes the session

3227 + // need not, so the wait ends on either.

3228 + let verdict = tokio::select! {

3229 + verdict = self.request_tool_permission(

3230 cx,

3231 &session_id,

3232 &tc.id,

3233 &tc.name,

3234 &tc.arguments,

3168 - )

3235 + ) => verdict,

3236 + _ = cancellation.cancelled() => PermissionVerdict::TurnCancelled,

3237 + };

3238 + self.resume_workspace(&mut workspace, &session_id, &cancellation)

3239 .await;

3170 - self.resume_workspace(&mut workspace, &session_id).await;

3240 + // An approval can cross a cancellation on the wire,

3241 + // or arrive while this turn waited for the lock.

3242 + let verdict = if cancellation.cancelled.load(Ordering::Acquire) {

3243 + PermissionVerdict::TurnCancelled

3244 + } else {

3245 + verdict

3246 + };

3247 match verdict {

3248 PermissionVerdict::Approved => {

3249 // The call the user just approved leaves

@@ -3308,7 +3384,8 @@ impl SiGitAgent {

3384 &cancellation,

3385 )

3386 .await;

3311 - self.resume_workspace(&mut workspace, &session_id).await;

3387 + self.resume_workspace(&mut workspace, &session_id, &cancellation)

3388 + .await;

3389 result = match outcome {

3390 Ok(result) => result,

3391 Err(DrainTurnError::Cancelled) => {

@@ -5371,8 +5448,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5448 let state = Arc::clone(&state);

5449 let task_cx = cx.clone();

5450 cx.spawn(async move {

5374 - let session = state.session_lock(&req.session_id);

5375 - let _session = session.lock().await;

5451 + let _session = state.lock_session(&req.session_id).await;

5452 let _workspace = state.workspace_lock.lock().await;

5453 handle_response(responder, state.handle_load_session(&task_cx, req).await)

5454 })

@@ -5387,8 +5463,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5463 let state = Arc::clone(&state);

5464 let task_cx = cx.clone();

5465 cx.spawn(async move {

5390 - let session = state.session_lock(&req.session_id);

5391 - let _session = session.lock().await;

5466 + let _session = state.lock_session(&req.session_id).await;

5467 let _workspace = state.workspace_lock.lock().await;

5468 handle_response(responder, state.handle_resume_session(&task_cx, req).await)

5469 })

@@ -5405,8 +5480,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5480 state.begin_close(&req.session_id);

5481 let state = Arc::clone(&state);

5482 cx.spawn(async move {

5408 - let session = state.session_lock(&req.session_id);

5409 - let _session = session.lock().await;

5483 + let _session = state.lock_session(&req.session_id).await;

5484 let _workspace = state.workspace_lock.lock().await;

5485 handle_response(responder, state.handle_close_session(req).await)

5486 })

@@ -5423,8 +5497,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5497 cx.spawn(async move {

5498 // The source's lock: a fork copies its conversation,

5499 // which has to be between turns.

5426 - let session = state.session_lock(&req.session_id);

5427 - let _session = session.lock().await;

5500 + let _session = state.lock_session(&req.session_id).await;

5501 let _workspace = state.workspace_lock.lock().await;

5502 handle_response(responder, state.handle_fork_session(&task_cx, req).await)

5503 })

@@ -5453,8 +5526,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5526 let state = Arc::clone(&state);

5527 let task_cx = cx.clone();

5528 cx.spawn(async move {

5456 - let session = state.session_lock(&req.session_id);

5457 - let _session = session.lock().await;

5529 + let _session = state.lock_session(&req.session_id).await;

5530 handle_response(responder, state.handle_prompt(&task_cx, req).await)

5531 })

5532 }

@@ -5470,8 +5542,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5542 let state = Arc::clone(&state);

5543 let task_cx = cx.clone();

5544 cx.spawn(async move {

5473 - let session = state.session_lock(&req.session_id);

5474 - let _session = session.lock().await;

5545 + let _session = state.lock_session(&req.session_id).await;

5546 let _workspace = state.workspace_lock.lock().await;

5547 handle_response(

5548 responder,

@@ -5489,8 +5560,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5560 let state = Arc::clone(&state);

5561 let task_cx = cx.clone();

5562 cx.spawn(async move {

5492 - let session = state.session_lock(&req.session_id);

5493 - let _session = session.lock().await;

5563 + let _session = state.lock_session(&req.session_id).await;

5564 let _workspace = state.workspace_lock.lock().await;

5565 handle_response(

5566 responder,

@@ -6390,6 +6460,89 @@ mod tests {

6460 std::fs::remove_dir_all(&temp).ok();

6461 }

6462

6463 + #[tokio::test]

6464 + #[allow(clippy::await_holding_lock)] // the lock guards process-global env for the whole test

6465 + async fn a_session_lock_entry_goes_once_its_session_and_its_waiters_are_gone() {

6466 + let _guard = crate::ENV_TEST_LOCK

6467 + .lock()

6468 + .unwrap_or_else(|poisoned| poisoned.into_inner());

6469 + let agent = Arc::new(SiGitAgent::new(

6470 + Arc::new(ChatEngine::new()),

6471 + default_local_model_config(),

6472 + Arc::new(AtomicBool::new(true)),

6473 + Arc::new(AtomicBool::new(false)),

6474 + Arc::new(std::sync::Mutex::new(None)),

6475 + false,

6476 + false,

6477 + ));

6478 + let held = |key: &str| agent.session_locks.lock().unwrap().contains_key(key);

6479 + let thread = SessionId::new("lock-thread");

6480 + agent.sessions.lock().unwrap().insert(

6481 + "lock-thread".into(),

6482 + agent.session_state(&std::env::temp_dir(), &[]),

6483 + );

6484 +

6485 + // An open session keeps its entry between requests.

6486 + drop(agent.lock_session(&thread).await);

6487 + assert!(held("lock-thread"));

6488 +

6489 + // A request naming an id nobody opened leaves nothing behind.

6490 + drop(agent.lock_session(&SessionId::new("lock-stranger")).await);

6491 + assert!(!held("lock-stranger"));

6492 +

6493 + // The session closes with another request queued behind the close.

6494 + // That request still has to find the same lock, so the entry stays

6495 + // until it has had its turn.

6496 + let closing = agent.lock_session(&thread).await;

6497 + let queued = agent.lock_session(&thread);

6498 + tokio::pin!(queued);

6499 + tokio::select! {

6500 + biased;

6501 + _ = &mut queued => panic!("the queued request ran beside the one holding the lock"),

6502 + _ = std::future::ready(()) => {}

6503 + }

6504 + agent.sessions.lock().unwrap().remove("lock-thread");

6505 + drop(closing);

6506 + assert!(held("lock-thread"));

6507 + drop(queued.await);

6508 + assert!(!held("lock-thread"));

6509 + }

6510 +

6511 + #[tokio::test]

6512 + #[allow(clippy::await_holding_lock)] // the lock guards process-global env for the whole test

6513 + async fn a_turn_whose_session_cannot_be_reinstalled_is_cancelled() {

6514 + let _guard = crate::ENV_TEST_LOCK

6515 + .lock()

6516 + .unwrap_or_else(|poisoned| poisoned.into_inner());

6517 + let agent = SiGitAgent::new(

6518 + Arc::new(ChatEngine::new()),

6519 + default_local_model_config(),

6520 + Arc::new(AtomicBool::new(true)),

6521 + Arc::new(AtomicBool::new(false)),

6522 + Arc::new(std::sync::Mutex::new(None)),

6523 + false,

6524 + false,

6525 + );

6526 + let gone = SessionId::new("resume-thread-gone");

6527 + let cancellation = PromptCancellation::default();

6528 + let mut workspace = WorkspaceHold::acquire(&agent.workspace_lock).await;

6529 +

6530 + // The lock was never given up, so there is nothing to reinstall.

6531 + agent

6532 + .resume_workspace(&mut workspace, &gone, &cancellation)

6533 + .await;

6534 + assert!(!cancellation.cancelled.load(Ordering::Acquire));

6535 +

6536 + workspace.release();

6537 + agent

6538 + .resume_workspace(&mut workspace, &gone, &cancellation)

6539 + .await;

6540 + assert!(cancellation.cancelled.load(Ordering::Acquire));

6541 + // The signal is there for whatever waits on it next.

6542 + cancellation.cancelled().await;

6543 + assert!(agent.active_session.lock().unwrap().is_none());

6544 + }

6545 +

6546 #[test]

6547 fn session_context_message_lists_every_root_of_a_multi_root_project() {

6548 let primary = PathBuf::from("/tmp/primary");

tests/acp_concurrent_sessions.rs

+150 -3

@@ -22,6 +22,10 @@ const TIMEOUT: Duration = Duration::from_secs(30);

22 /// The prompt text that makes the endpoint hold its reply open.

23 const HOLD: &str = "HOLD-THIS-TURN";

25 +/// The prompt text that makes the model ask for a command instead of a read,

26 +/// which is what the permission gate stops on.

27 +const RUN: &str = "RUN-A-COMMAND";

28 +

29 fn sse(events: &[Value]) -> String {

30 events

31 .iter()

@@ -47,13 +51,25 @@ fn read_marker_call() -> Value {

51 })

52 }

54 +/// Writes `ran.txt` into the working directory. Plain redirection, so the

55 +/// same line works under `sh` and `cmd`.

56 +fn run_command_call() -> Value {

57 + json!({

58 + "choices": [{"delta": {"tool_calls": [{

59 + "index": 0,

60 + "id": "call_run",

61 + "function": {"name": "run_command", "arguments": "{\"command\":\"echo ran > ran.txt\"}"},

62 + }]}}]

63 + })

64 +}

65 +

66 /// An endpoint that answers each connection on its own thread.

67 ///

68 /// A request that already carries a tool result gets a closing line of text.

69 /// Otherwise the model "asks" to read `marker.txt`, a relative path, so the

70 /// result shows which directory the agent resolved it against. A first request

71 /// mentioning [`HOLD`] streams one fragment and then waits for `release`

56 -/// before asking for the file.

72 +/// before asking for the file. One mentioning [`RUN`] asks for a command.

73 struct Endpoint {

74 port: u16,

75 requests: Arc<Mutex<Vec<Value>>>,

@@ -131,7 +147,12 @@ fn start_endpoint() -> Endpoint {

147 std::thread::sleep(Duration::from_millis(20));

148 }

149 }

134 - sse(&[read_marker_call(), finish("tool_calls")])

150 + let call = if mentions(&request, RUN) {

151 + run_command_call()

152 + } else {

153 + read_marker_call()

154 + };

155 + sse(&[call, finish("tool_calls")])

156 };

157 let _ = stream.write_all(body.as_bytes());

158 let _ = stream.write_all(b"data: [DONE]\n\n");

@@ -157,13 +178,21 @@ struct AgentUnderTest {

178 }

179

180 fn spawn_agent(port: u16, config_dir: &std::path::Path) -> AgentUnderTest {

181 + spawn_agent_with_permissions(port, config_dir, "allow")

182 +}

183 +

184 +fn spawn_agent_with_permissions(

185 + port: u16,

186 + config_dir: &std::path::Path,

187 + permissions: &str,

188 +) -> AgentUnderTest {

189 let mut child = Command::new(env!("CARGO_BIN_EXE_sigit"))

190 .env("OPENAI_BASE_URL", format!("http://127.0.0.1:{port}"))

191 .env("OPENAI_API_KEY", "test-key")

192 .env("SIGIT_MODEL", "scripted-model")

193 .env("SIGIT_CONFIG_DIR", config_dir)

194 .env("SIGIT_MCP", "off")

166 - .env("SIGIT_PERMISSIONS", "allow")

195 + .env("SIGIT_PERMISSIONS", permissions)

196 .env_remove("SIGIT_LOCAL_INFERENCE")

197 .env_remove("SIGIT_MAX_TOOL_ROUNDS")

198 .stdin(Stdio::piped())

@@ -234,6 +263,18 @@ impl AgentUnderTest {

263 message

264 }

265

266 + /// Answer a request the agent sent, such as a permission request.

267 + fn respond(&mut self, id: &Value, result: Value) {

268 + let mut line = json!({"jsonrpc": "2.0", "id": id, "result": result}).to_string();

269 + line.push('\n');

270 + self.stdin.write_all(line.as_bytes()).expect("write stdin");

271 + self.stdin.flush().expect("flush stdin");

272 + }

273 +

274 + fn close_session(&mut self, session_id: &str) -> u64 {

275 + self.request("session/close", json!({"sessionId": session_id}))

276 + }

277 +

278 fn responded(&self, id: u64) -> bool {

279 self.seen

280 .iter()

@@ -341,3 +382,109 @@ fn a_second_session_opens_runs_and_finishes_while_the_first_is_streaming() {

382

383 std::fs::remove_dir_all(&dir).ok();

384 }

385 +

386 +fn initialized_agent(port: u16, config_dir: &std::path::Path, permissions: &str) -> AgentUnderTest {

387 + std::fs::create_dir_all(config_dir).unwrap();

388 + let mut agent = spawn_agent_with_permissions(port, config_dir, permissions);

389 + let id = agent.request(

390 + "initialize",

391 + json!({"protocolVersion": 1, "clientCapabilities": {}}),

392 + );

393 + agent.wait_for_response(id);

394 + agent

395 +}

396 +

397 +#[test]

398 +fn closing_a_session_mid_stream_ends_its_turn_and_runs_no_tool() {

399 + let dir = std::env::temp_dir().join(format!("sigit_acp_close_stream_{}", std::process::id()));

400 + let (work_a, work_b) = (dir.join("work-a"), dir.join("work-b"));

401 + for (work, marker) in [(&work_a, "marker-from-a"), (&work_b, "marker-from-b")] {

402 + std::fs::create_dir_all(work).unwrap();

403 + std::fs::write(work.join("marker.txt"), marker).unwrap();

404 + }

405 +

406 + let endpoint = start_endpoint();

407 + let mut agent = initialized_agent(endpoint.port, &dir.join("config"), "allow");

408 +

409 + // Thread B exists first, so thread A is the installed one when it closes.

410 + let session_b = agent.new_session(&work_b);

411 + let session_a = agent.new_session(&work_a);

412 + let prompt_a = agent.send_prompt(&session_a, HOLD);

413 + agent.wait_for("thread A's first streamed fragment", |message| {

414 + message["params"]["sessionId"] == session_a.as_str()

415 + && message["params"]["update"]["sessionUpdate"] == "agent_message_chunk"

416 + });

417 +

418 + // The editor closes the thread while its turn is waiting on the endpoint.

419 + let close_a = agent.close_session(&session_a);

420 + let response_a = agent.wait_for_response(prompt_a);

421 + assert_eq!(response_a["result"]["stopReason"], "cancelled");

422 + agent.wait_for_response(close_a);

423 +

424 + // The endpoint now asks for the file. Nobody is left to read it.

425 + endpoint.release.store(true, Ordering::Release);

426 +

427 + // The id is gone, and a request for it gets an answer instead of waiting

428 + // behind anything the closed turn left held.

429 + let late = agent.send_prompt(&session_a, "anyone there");

430 + let refused = agent.wait_for("the answer to a prompt on the closed thread", |message| {

431 + message["id"] == late && message.get("method").is_none()

432 + });

433 + assert!(refused.get("error").is_some(), "{refused}");

434 +

435 + // Thread B still runs its tool in its own directory.

436 + let prompt_b = agent.send_prompt(&session_b, "read the marker");

437 + let response_b = agent.wait_for_response(prompt_b);

438 + assert_eq!(response_b["result"]["stopReason"], "end_turn");

439 +

440 + let requests = endpoint.requests.lock().unwrap().clone();

441 + let follow_ups: Vec<&Value> = requests.iter().filter(|r| has_tool_result(r)).collect();

442 + assert_eq!(

443 + follow_ups.len(),

444 + 1,

445 + "only thread B ran a tool: {requests:?}"

446 + );

447 + let result = tool_result(follow_ups[0]);

448 + assert!(result.contains("marker-from-b"), "thread B read {result}");

449 +

450 + std::fs::remove_dir_all(&dir).ok();

451 +}

452 +

453 +#[test]

454 +fn closing_a_session_at_a_permission_prompt_ends_its_turn_without_an_answer() {

455 + let dir = std::env::temp_dir().join(format!("sigit_acp_close_ask_{}", std::process::id()));

456 + let work = dir.join("work");

457 + std::fs::create_dir_all(&work).unwrap();

458 +

459 + let endpoint = start_endpoint();

460 + let mut agent = initialized_agent(endpoint.port, &dir.join("config"), "ask");

461 +

462 + let session = agent.new_session(&work);

463 + let prompt = agent.send_prompt(&session, RUN);

464 + let permission = agent.wait_for("the permission request", |message| {

465 + message["method"] == "session/request_permission"

466 + });

467 +

468 + // The client closes the thread and never answers the request. The turn

469 + // must not keep the close waiting on an answer that is not coming.

470 + let close = agent.close_session(&session);

471 + let response = agent.wait_for_response(prompt);

472 + assert_eq!(response["result"]["stopReason"], "cancelled");

473 + agent.wait_for_response(close);

474 +

475 + // An approval that arrives after the close runs nothing.

476 + agent.respond(

477 + &permission["id"],

478 + json!({"outcome": {"outcome": "selected", "optionId": "allow_once"}}),

479 + );

480 + let late = agent.send_prompt(&session, "anyone there");

481 + agent.wait_for("the answer to a prompt on the closed thread", |message| {

482 + message["id"] == late && message.get("method").is_none()

483 + });

484 + assert!(

485 + !work.join("ran.txt").exists(),

486 + "the command ran after the close"

487 + );

488 +

489 + std::fs::remove_dir_all(&dir).ok();

490 +}