@sigitcodereview / sigit / commits / 52f546a

Let ACP sessions on an HTTP backend run at the same time (#176)

An editor runs one sigit process for every thread it has open, and a single lock ordered every turn-affecting request in it. A second thread could not open until the first thread's turn ended, and two prompts never overlapped. The lock was there because a session installs process-wide state: the working directory, workspace roots, session MCP servers and the live backend. A turn does not use any of that while it waits on an HTTP endpoint, which is most of its time. Each session on an HTTP backend now keeps a backend of its own, so its conversation no longer has to be swapped in and out of a shared one. The process-wide lock becomes workspace_lock. A prompt turn releases it while it waits on the endpoint or on a permission answer, and takes it back, reinstalling its session, before any tool runs. A per-session lock keeps requests on one thread in order. On-device turns are unchanged: the engine holds one conversation, so they keep the workspace for the whole turn. Tool execution, subagents included, still runs one session at a time. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: siGit Code v1.6.0 <noreply@sigit.si>

Seto Elkahfi committed Oct 6, 2026 at 20:35 UTC 52f546a8d8b436dca42d8bd2825695597e1c095b

4 files changed +676 -48

.agents/AGENTS.md

+15 -6

@@ -113,9 +113,11 @@ feeds results back. Neither the loop nor ACP/TUI surfaces depend on a concrete b

113 - **`src/main.rs`** — entry point, mode dispatch, the full ACP `Agent` impl (session lifecycle:

114 new/load/fork/prompt/cancel, config options, slash-command advertisement), and the `SYSTEM_PROMPT`.

115 ACP session state owns its roots, conversation, and selected model even though the process has

116 - one live backend; activating a thread parks and restores all three. Unknown session ids are

117 - rejected instead of silently borrowing the active thread's cwd. Prompt cancellation is routed

118 - outside `turn_lock`, which lets a client cancel the turn currently holding that lock. The

116 + one working directory and one live backend slot; activating a thread parks and restores all

117 + three. A thread on an HTTP backend keeps a backend of its own (`InferenceBackend::fresh`,

118 + `SessionState::remote`), so its turn can wait on the endpoint while another thread is

119 + installed. Unknown session ids are rejected instead of silently borrowing the active thread's

120 + cwd. Prompt cancellation takes no lock, which lets a client cancel a turn whatever it holds. The

121 `SYSTEM_PROMPT` bakes in smbCloud-specific context the agent should use when the repo is clearly

122 smbCloud, and stay general otherwise.

123 - **`src/backend.rs`** — the `InferenceBackend` trait and neutral types (`ToolSpec`, `ToolCall`,

@@ -273,9 +275,16 @@ feeds results back. Neither the loop nor ACP/TUI surfaces depend on a concrete b

275 the spec is retiring but some clients still render instead. Both are built from

276 `PERMISSION_MODES` in `main.rs` and read the same state, so whatever changes the mode has to

277 refresh both (`ConfigOptionUpdate` and `send_current_mode`). Note: ACP turn-affecting handlers run in `cx.spawn`ed tasks

276 - serialized by `SiGitAgent::turn_lock` so the dispatch loop can route the client's permission

277 - answer mid-turn — don't move them back inline, and don't await client requests from inline

278 - handlers (deadlock).

278 + so the dispatch loop can route the client's permission answer mid-turn — don't move them back

279 + inline, and don't await client requests from inline handlers (deadlock). Two locks order them.

280 + A per-session lock (`SiGitAgent::session_lock`) is held for a whole request, so requests on one

281 + thread keep their order. `SiGitAgent::workspace_lock` guards what a session installs

282 + process-wide (cwd, workspace roots, session MCP servers, the live backend slot, the engine's

283 + conversation); lifecycle and config handlers hold it throughout, and `handle_prompt` holds it

284 + through a `WorkspaceHold` that it releases while waiting on an HTTP endpoint or a permission

285 + answer and retakes (reinstalling its session via `resume_workspace`) before any tool runs. An

286 + on-device turn never releases it. Take the session lock first, never the other way round. Tool

287 + execution, subagents included, still runs one session at a time.

288 - **`src/instructions.rs`** — project instruction files, the always-on counterpart to skills.

289 Reads `AGENTS.md` (the cross-tool [agents.md](https://agents.md) standard) and `CLAUDE.md`,

290 walking from the session cwd up to the repo root (nearest ancestor with `.git`, never above it),

src/backend.rs

+22

@@ -445,6 +445,15 @@ pub trait InferenceBackend: Send + Sync {

445 /// local model while requests actually go to the cloud.

446 fn is_remote(&self) -> bool;

447

448 + /// A backend for the same endpoint and model with an empty conversation,

449 + /// or `None` when the conversation cannot be separated from the backend

450 + /// (on-device, the engine holds it). An editor keeps several threads open

451 + /// in one process; this is what lets each of them own its history instead

452 + /// of taking turns on a shared one.

453 + fn fresh(&self) -> Option<Arc<dyn InferenceBackend>> {

454 + None

455 + }

456 +

457 /// A serializable snapshot of the conversation history, one JSON object per

458 /// message (`{"role": ..., "content": ...}` at minimum). The snapshot is

459 /// what the session store persists; it includes any seeded system message

@@ -1554,6 +1563,19 @@ impl InferenceBackend for OpenAiBackend {

1563 true

1564 }

1565

1566 + fn fresh(&self) -> Option<Arc<dyn InferenceBackend>> {

1567 + Some(Arc::new(Self {

1568 + base_url: self.base_url.clone(),

1569 + api_key: self.api_key.clone(),

1570 + model: self.model.clone(),

1571 + // The client is a handle on one connection pool; sharing it keeps

1572 + // connections to the endpoint warm across sessions.

1573 + http: self.http.clone(),

1574 + history: Mutex::new(Vec::new()),

1575 + accepts_images: self.accepts_images,

1576 + }))

1577 + }

1578 +

1579 async fn history_snapshot(&self) -> Vec<serde_json::Value> {

1580 self.history.lock().await.clone()

1581 }

src/main.rs

+296 -42

@@ -924,11 +924,15 @@ impl client_fs::ClientFileSystem for AcpClientFs {

924 /// What one ACP session owns, kept per session id.

925 ///

926 /// An editor runs a single sigit process for every thread it has open, but the

927 -/// process has one working directory, one set of workspace roots, and one

928 -/// backend conversation. Only one session can be live at a time (turns are

929 -/// serialized by `turn_lock`), so each session's roots and conversation are

930 -/// parked here while another session holds the process, and swapped back in

931 -/// by `activate_session` before its next request runs.

927 +/// process has one working directory and one set of workspace roots. Only one

928 +/// session can be installed at a time (see `workspace_lock`), so each session's

929 +/// roots are parked here while another session holds the process, and swapped

930 +/// back in by `activate_session` before its next request runs.

931 +///

932 +/// The conversation is parked the same way on-device, where the engine holds

933 +/// it. A session on an HTTP backend keeps a backend of its own instead (see

934 +/// `remote`), which is what lets its turn wait on the endpoint while another

935 +/// session is installed.

936 #[derive(Clone, Debug)]

937 struct SessionState {

938 cwd: PathBuf,

@@ -946,6 +950,57 @@ struct SessionState {

950 /// The MCP servers the client passed in `mcpServers` for this session,

951 /// connected. Installed as the live set along with the roots.

952 mcp_servers: mcp::SessionServers,

953 + /// This session's own HTTP backend, set while the session is parked and

954 + /// was last served by one. `None` for a session that has not run yet, was

955 + /// last on-device, or is the installed session (the agent's `backend`

956 + /// holds it then).

957 + remote: Option<RemoteBackend>,

958 +}

959 +

960 +/// An HTTP backend that belongs to one session, with the model selection that

961 +/// was live when the session was parked.

962 +#[derive(Clone)]

963 +struct RemoteBackend {

964 + backend: Arc<dyn InferenceBackend>,

965 + model: GgufModelConfig,

966 +}

967 +

968 +impl std::fmt::Debug for RemoteBackend {

969 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

970 + f.debug_struct("RemoteBackend")

971 + .field("model", &self.model.model_id)

972 + .finish_non_exhaustive()

973 + }

974 +}

975 +

976 +/// Holds `workspace_lock` for a prompt turn, which gives it up while it waits

977 +/// on an HTTP endpoint or on the user and takes it back before touching

978 +/// anything process-wide.

979 +struct WorkspaceHold {

980 + lock: Arc<tokio::sync::Mutex<()>>,

981 + guard: Option<tokio::sync::OwnedMutexGuard<()>>,

982 +}

983 +

984 +impl WorkspaceHold {

985 + async fn acquire(lock: &Arc<tokio::sync::Mutex<()>>) -> Self {

986 + Self {

987 + lock: Arc::clone(lock),

988 + guard: Some(Arc::clone(lock).lock_owned().await),

989 + }

990 + }

991 +

992 + fn release(&mut self) {

993 + self.guard = None;

994 + }

995 +

996 + /// Whether the lock was given up since it was last held.

997 + async fn reacquire(&mut self) -> bool {

998 + if self.guard.is_some() {

999 + return false;

1000 + }

1001 + self.guard = Some(Arc::clone(&self.lock).lock_owned().await);

1002 + true

1003 + }

1004 }

1005

1006 struct SiGitAgent {

@@ -961,8 +1016,8 @@ struct SiGitAgent {

1016 /// The session whose roots and conversation are installed right now.

1017 active_session: std::sync::Mutex<Option<String>>,

1018 /// Cancellation signals for prompts currently waiting on inference. The

964 - /// notification handler runs outside `turn_lock`, so it can interrupt the

965 - /// turn that currently owns that lock and let another session proceed.

1019 + /// notification handler takes no lock, so it can interrupt a turn whatever

1020 + /// that turn is holding.

1021 prompt_cancellations:

1022 std::sync::Mutex<std::collections::HashMap<String, Arc<PromptCancellation>>>,

1023 current_model: std::sync::Mutex<GgufModelConfig>,

@@ -981,12 +1036,19 @@ struct SiGitAgent {

1036 startup_model_name: String,

1037 /// for download-progress polling

1038 startup_model_id: String,

984 - /// Serializes turn-affecting handlers (prompt, session lifecycle, config

985 - /// changes). They run in `cx.spawn`ed tasks so the JSON-RPC dispatch loop

986 - /// stays free to route client responses (e.g. permission answers) mid-turn;

987 - /// this lock reproduces the strict ordering the dispatch loop used to give

988 - /// them for free.

989 - turn_lock: Arc<tokio::sync::Mutex<()>>,

1039 + /// Guards everything one session installs process-wide: the working

1040 + /// directory, workspace roots, session MCP servers, the live `backend`

1041 + /// slot and the on-device engine's conversation. Session lifecycle and

1042 + /// config handlers hold it throughout. A prompt turn holds it while it

1043 + /// prepares and while its tools run, and gives it up while it waits on an

1044 + /// HTTP endpoint or on a permission answer, so another session can be

1045 + /// installed in the meantime. An on-device turn never gives it up: the

1046 + /// engine serves one conversation at a time.

1047 + workspace_lock: Arc<tokio::sync::Mutex<()>>,

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

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

1050 + /// taken before `workspace_lock`, never while holding it.

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

1052 }

1053

1054 impl SiGitAgent {

@@ -1017,7 +1079,8 @@ impl SiGitAgent {

1079 auto_load_local_model,

1080 startup_model_name,

1081 startup_model_id,

1020 - turn_lock: Arc::new(tokio::sync::Mutex::new(())),

1082 + workspace_lock: Arc::new(tokio::sync::Mutex::new(())),

1083 + session_locks: std::sync::Mutex::new(std::collections::HashMap::new()),

1084 }

1085 }

1086

@@ -1854,9 +1917,21 @@ impl SiGitAgent {

1917 model_id,

1918 conversation: Vec::new(),

1919 mcp_servers: mcp::SessionServers::default(),

1920 + remote: None,

1921 }

1922 }

1923

1924 + /// The lock that orders requests on one session (see `session_locks`).

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

1926 + Arc::clone(

1927 + self.session_locks

1928 + .lock()

1929 + .unwrap()

1930 + .entry(session_id.to_string())

1931 + .or_default(),

1932 + )

1933 + }

1934 +

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

1936 /// model/backend switch. Its conversation remains stale while active.

1937 fn record_active_model(&self) {

@@ -1870,14 +1945,22 @@ impl SiGitAgent {

1945 }

1946

1947 /// Move the live conversation into the state of the session that owns it,

1873 - /// so installing another session can't lose or leak it.

1948 + /// so installing another session can't lose or leak it. An HTTP backend

1949 + /// goes with it: the session may be waiting on that backend right now, and

1950 + /// its next request has to find the same one.

1951 async fn park_active_session(&self) {

1952 let Some(active) = self.active_session.lock().unwrap().clone() else {

1953 return;

1954 };

1878 - let snapshot = self.backend.lock().await.history_snapshot().await;

1955 + let live = self.backend.lock().await.clone();

1956 + let snapshot = live.history_snapshot().await;

1957 + let remote = live.is_remote().then(|| RemoteBackend {

1958 + backend: live,

1959 + model: self.current_model.lock().unwrap().clone(),

1960 + });

1961 if let Some(state) = self.sessions.lock().unwrap().get_mut(&active) {

1962 state.conversation = backend::carryover_history(snapshot);

1963 + state.remote = remote;

1964 }

1965 }

1966

@@ -1888,7 +1971,8 @@ impl SiGitAgent {

1971 /// backend installed here starts from this session's conversation and

1972 /// system prompt and nobody else's. Reopening an existing id deliberately

1973 /// replaces its parked state (notably `session/load`); every caller holds

1891 - /// `turn_lock`, so parking and replacement cannot race another turn.

1974 + /// `workspace_lock`, so parking and replacement cannot race another

1975 + /// session being installed.

1976 async fn open_session(&self, session_id: &SessionId, state: SessionState) {

1977 self.park_active_session().await;

1978 self.open_parked_session(session_id, state).await;

@@ -1942,6 +2026,19 @@ impl SiGitAgent {

2026 Ok(())

2027 }

2028

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

2030 + /// session the installed one again if another was installed meanwhile.

2031 + async fn resume_workspace(&self, workspace: &mut WorkspaceHold, session_id: &SessionId) {

2032 + if workspace.reacquire().await

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

2034 + {

2035 + log::warn!(

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

2037 + error.message

2038 + );

2039 + }

2040 + }

2041 +

2042 /// Point the process at `state`'s roots and load its conversation.

2043 ///

2044 /// The process follows `cwd` because tool calls may use relative paths; the

@@ -1952,9 +2049,10 @@ impl SiGitAgent {

2049 let SessionState {

2050 cwd,

2051 additional_roots,

1955 - model_id: _,

2052 + model_id,

2053 conversation,

2054 mcp_servers,

2055 + remote,

2056 } = state;

2057

2058 if let Ok(mut guard) = self.session_cwd.lock() {

@@ -1969,11 +2067,44 @@ impl SiGitAgent {

2067 }

2068 *self.active_session.lock().unwrap() = Some(key.to_string());

2069 tools::set_active_session(Some(key));

2070 + if let Some(parked) = self.sessions.lock().unwrap().get_mut(key) {

2071 + parked.remote = None;

2072 + }

2073

2074 + // A session that already has its own HTTP backend goes back onto it

2075 + // untouched. Its conversation lives there and may be mid-turn, so it

2076 + // is not reseeded from the parked copy. The engine still gets this

2077 + // session's context, for a later switch to an on-device model.

2078 + if let Some(RemoteBackend { backend, model }) = remote

2079 + && model.model_id == model_id

2080 + {

2081 + self.seed_engine_context(&session_context_message(&cwd, &additional_roots))

2082 + .await;

2083 + *self.backend.lock().await = backend;

2084 + *self.current_model.lock().unwrap() = model;

2085 + return;

2086 + }

2087 +

2088 + // Otherwise the session starts from its parked conversation. The live

2089 + // HTTP backend belongs to whichever session was installed before, and

2090 + // that session may still be waiting on it, so this one gets a backend

2091 + // of its own for the same endpoint.

2092 + let fresh = self.backend.lock().await.fresh();

2093 + if let Some(fresh) = fresh {

2094 + *self.backend.lock().await = fresh;

2095 + }

2096 self.seed_conversation(&cwd, &additional_roots, conversation)

2097 .await;

2098 }

2099

2100 + /// Reset the on-device engine's conversation to just `context`.

2101 + async fn seed_engine_context(&self, context: &str) {

2102 + self.engine.clear_history().await;

2103 + self.engine

2104 + .push_history(onde::inference::ChatMessage::system(context.to_string()))

2105 + .await;

2106 + }

2107 +

2108 /// Restore the active session's model/backend after its conversation has

2109 /// been installed on the currently live backend. Model switches carry that

2110 /// freshly installed conversation across to the target backend.

@@ -2034,10 +2165,7 @@ impl SiGitAgent {

2165 let backend = self.backend.lock().await.clone();

2166

2167 if backend.is_remote() {

2037 - self.engine.clear_history().await;

2038 - self.engine

2039 - .push_history(onde::inference::ChatMessage::system(context.clone()))

2040 - .await;

2168 + self.seed_engine_context(&context).await;

2169 }

2170

2171 let system_prompt = if backend.is_remote() {

@@ -2340,6 +2468,7 @@ impl SiGitAgent {

2468

2469 // Everything below (slash commands included) acts on the live session,

2470 // so it has to be this one. Resource links are read relative to its cwd.

2471 + let mut workspace = WorkspaceHold::acquire(&self.workspace_lock).await;

2472 self.activate_session(&session_id).await?;

2473

2474 // log every block so we can debug @ references and file context

@@ -2520,6 +2649,10 @@ impl SiGitAgent {

2649 // The active backend drives the turn. Snapshot it once so a mid-turn

2650 // model switch doesn't split the conversation across backends.

2651 let backend = self.backend.lock().await.clone();

2652 + // Waiting on an HTTP endpoint uses nothing process-wide, so another

2653 + // session may be installed meanwhile. The on-device engine holds one

2654 + // conversation, so an on-device turn keeps the workspace throughout.

2655 + let release_while_waiting = backend.is_remote();

2656

2657 // Only on-device inference needs a local model in memory. Cloud tiers run

2658 // over the network, so they never need a local model. Xcode's explicit

@@ -2602,7 +2735,10 @@ impl SiGitAgent {

2735 None

2736 };

2737

2605 - let mut result = match self

2738 + if release_while_waiting {

2739 + workspace.release();

2740 + }

2741 + let outcome = self

2742 .drain_turn(

2743 cx,

2744 &session_id,

@@ -2611,8 +2747,9 @@ impl SiGitAgent {

2747 &mut reply,

2748 &cancellation,

2749 )

2614 - .await

2615 - {

2750 + .await;

2751 + self.resume_workspace(&mut workspace, &session_id).await;

2752 + let mut result = match outcome {

2753 Ok(result) => result,

2754 Err(DrainTurnError::Cancelled) => {

2755 backend::forget_trailing_user_message(backend.as_ref()).await;

@@ -2796,7 +2933,11 @@ impl SiGitAgent {

2933 reason

2934 }

2935 permissions::Decision::Ask => {

2799 - match self

2936 + // The answer can take as long as the user likes.

2937 + if release_while_waiting {

2938 + workspace.release();

2939 + }

2940 + let verdict = self

2941 .request_tool_permission(

2942 cx,

2943 &session_id,

@@ -2804,8 +2945,9 @@ impl SiGitAgent {

2945 &tc.name,

2946 &tc.arguments,

2947 )

2807 - .await

2808 - {

2948 + .await;

2949 + self.resume_workspace(&mut workspace, &session_id).await;

2950 + match verdict {

2951 PermissionVerdict::Approved => {

2952 // The call the user just approved leaves

2953 // `pending` here. The title goes back to

@@ -2932,7 +3074,10 @@ impl SiGitAgent {

3074 sent_before_last_round = reply.sent.len();

3075

3076 let cancelled_results = tool_results.clone();

2935 - result = match self

3077 + if release_while_waiting {

3078 + workspace.release();

3079 + }

3080 + let outcome = self

3081 .drain_turn(

3082 cx,

3083 &session_id,

@@ -2941,8 +3086,9 @@ impl SiGitAgent {

3086 &mut reply,

3087 &cancellation,

3088 )

2944 - .await

2945 - {

3089 + .await;

3090 + self.resume_workspace(&mut workspace, &session_id).await;

3091 + result = match outcome {

3092 Ok(result) => result,

3093 Err(DrainTurnError::Cancelled) => {

3094 backend

@@ -4986,11 +5132,13 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

5132 },

5133 agent_client_protocol::on_receive_request!(),

5134 )

4989 - // Turn-affecting handlers below run in spawned tasks, serialized by

4990 - // `turn_lock`, so the dispatch loop stays free to route client

4991 - // responses (permission answers) while a turn is in flight. Awaiting a

4992 - // client request from *inside* a handler would deadlock: the dispatch

4993 - // loop can't read the response while the handler blocks it.

5135 + // Turn-affecting handlers below run in spawned tasks, so the dispatch

5136 + // loop stays free to route client responses (permission answers) while

5137 + // a turn is in flight. Awaiting a client request from *inside* a

5138 + // handler would deadlock: the dispatch loop can't read the response

5139 + // while the handler blocks it. Requests on one session keep their

5140 + // order through that session's lock; whatever touches process-wide

5141 + // state takes `workspace_lock` (a prompt does so itself, in pieces).

5142 .on_receive_request(

5143 {

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

@@ -4998,7 +5146,9 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

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

5147 let task_cx = cx.clone();

5148 cx.spawn(async move {

5001 - let _turn = state.turn_lock.lock().await;

5149 + let session = state.session_lock(&req.session_id);

5150 + let _session = session.lock().await;

5151 + let _workspace = state.workspace_lock.lock().await;

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

5153 })

5154 }

@@ -5012,7 +5162,11 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

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

5163 let task_cx = cx.clone();

5164 cx.spawn(async move {

5015 - let _turn = state.turn_lock.lock().await;

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

5166 + // which has to be between turns.

5167 + let session = state.session_lock(&req.session_id);

5168 + let _session = session.lock().await;

5169 + let _workspace = state.workspace_lock.lock().await;

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

5171 })

5172 }

@@ -5026,7 +5180,7 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

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

5181 let task_cx = cx.clone();

5182 cx.spawn(async move {

5029 - let _turn = state.turn_lock.lock().await;

5183 + let _workspace = state.workspace_lock.lock().await;

5184 handle_response(responder, state.handle_new_session(&task_cx, req).await)

5185 })

5186 }

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

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

5195 let task_cx = cx.clone();

5196 cx.spawn(async move {

5043 - let _turn = state.turn_lock.lock().await;

5197 + let session = state.session_lock(&req.session_id);

5198 + let _session = session.lock().await;

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

5200 })

5201 }

@@ -5056,7 +5211,9 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

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

5212 let task_cx = cx.clone();

5213 cx.spawn(async move {

5059 - let _turn = state.turn_lock.lock().await;

5214 + let session = state.session_lock(&req.session_id);

5215 + let _session = session.lock().await;

5216 + let _workspace = state.workspace_lock.lock().await;

5217 handle_response(

5218 responder,

5219 state.handle_set_session_config_option(&task_cx, req).await,

@@ -5073,7 +5230,9 @@ async fn run_acp_server(auto_load_local_model: bool) -> anyhow::Result<()> {

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

5231 let task_cx = cx.clone();

5232 cx.spawn(async move {

5076 - let _turn = state.turn_lock.lock().await;

5233 + let session = state.session_lock(&req.session_id);

5234 + let _session = session.lock().await;

5235 + let _workspace = state.workspace_lock.lock().await;

5236 handle_response(

5237 responder,

5238 state.handle_set_session_mode(&task_cx, req).await,

@@ -5671,6 +5830,101 @@ mod tests {

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

5831 }

5832

5833 + /// A turn that is waiting on its endpoint holds on to its backend while

5834 + /// another session is installed. Coming back, the session has to find that

5835 + /// same backend, with whatever the turn added to it, and the other session

5836 + /// must never have been given it.

5837 + #[tokio::test]

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

5839 + async fn a_session_on_an_http_backend_keeps_that_backend_across_other_sessions() {

5840 + let _guard = crate::ENV_TEST_LOCK

5841 + .lock()

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

5843 + let temp = std::env::temp_dir().join(format!("sigit-own-backend-{}", uuid::Uuid::new_v4()));

5844 + let (repo_a, repo_b) = (temp.join("repo-a"), temp.join("repo-b"));

5845 + for dir in [&repo_a, &repo_b] {

5846 + std::fs::create_dir_all(dir).unwrap();

5847 + }

5848 + let prev_cwd = std::env::current_dir().unwrap();

5849 + // SAFETY: serialized by ENV_TEST_LOCK and restored below.

5850 + unsafe {

5851 + std::env::set_var("SIGIT_CONFIG_DIR", temp.join("config"));

5852 + std::env::set_var("SIGIT_LOCAL_INFERENCE", "on");

5853 + }

5854 +

5855 + let agent = SiGitAgent::new(

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

5857 + default_local_model_config(),

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

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

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

5861 + false,

5862 + false,

5863 + );

5864 + // No request is ever sent.

5865 + *agent.backend.lock().await = Arc::new(OpenAiBackend::new(

5866 + "http://127.0.0.1:9",

5867 + "",

5868 + "m",

5869 + Some("base prompt".into()),

5870 + ));

5871 + let thread_a = SessionId::new("thread-a");

5872 + let thread_b = SessionId::new("thread-b");

5873 +

5874 + agent

5875 + .open_session(&thread_a, agent.session_state(&repo_a, &[]))

5876 + .await;

5877 + say(&agent, "first turn in a").await;

5878 + let backend_a = agent.backend.lock().await.clone();

5879 +

5880 + agent

5881 + .open_session(&thread_b, agent.session_state(&repo_b, &[]))

5882 + .await;

5883 + let backend_b = agent.backend.lock().await.clone();

5884 + assert!(

5885 + !Arc::ptr_eq(&backend_a, &backend_b),

5886 + "thread B was handed thread A's backend"

5887 + );

5888 + let history = history_text(&live(&agent).await);

5889 + assert!(!history.contains("first turn in a"), "{history}");

5890 + say(&agent, "first turn in b").await;

5891 +

5892 + // Thread A's turn, still in flight, adds to its own backend while

5893 + // thread B is the installed session.

5894 + let mut in_flight = backend_a.history_snapshot().await;

5895 + in_flight.push(serde_json::json!({ "role": "assistant", "content": "mid-turn in a" }));

5896 + backend_a.restore_history(in_flight).await;

5897 + let history = history_text(&live(&agent).await);

5898 + assert!(!history.contains("mid-turn in a"), "{history}");

5899 +

5900 + agent.activate_session(&thread_a).await.unwrap();

5901 + assert!(Arc::ptr_eq(&*agent.backend.lock().await, &backend_a));

5902 + let history = history_text(&live(&agent).await);

5903 + assert!(history.contains("first turn in a"), "{history}");

5904 + assert!(history.contains("mid-turn in a"), "{history}");

5905 + assert!(!history.contains("first turn in b"), "{history}");

5906 + assert_eq!(

5907 + workspace::canonical_key(&std::env::current_dir().unwrap()),

5908 + workspace::canonical_key(&repo_a)

5909 + );

5910 +

5911 + agent.activate_session(&thread_b).await.unwrap();

5912 + assert!(Arc::ptr_eq(&*agent.backend.lock().await, &backend_b));

5913 + let history = history_text(&live(&agent).await);

5914 + assert!(history.contains("first turn in b"), "{history}");

5915 + assert!(!history.contains("mid-turn in a"), "{history}");

5916 +

5917 + std::env::set_current_dir(prev_cwd).unwrap();

5918 + tools::set_active_session(None);

5919 + workspace::replace_additional_roots(Vec::new());

5920 + // SAFETY: serialized by ENV_TEST_LOCK.

5921 + unsafe {

5922 + std::env::remove_var("SIGIT_CONFIG_DIR");

5923 + std::env::remove_var("SIGIT_LOCAL_INFERENCE");

5924 + }

5925 + std::fs::remove_dir_all(&temp).ok();

5926 + }

5927 +

5928 #[test]

5929 fn session_context_message_lists_every_root_of_a_multi_root_project() {

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

tests/acp_concurrent_sessions.rs new

+343

@@ -0,0 +1,343 @@

1 +//! Two ACP sessions in one agent process, on an HTTP backend.

2 +//!

3 +//! An editor runs a single sigit process for every thread it has open. A turn

4 +//! spends most of its time waiting on the endpoint, and nothing in that wait

5 +//! needs the process's working directory or workspace roots, so a second

6 +//! thread should be able to open, run a tool, and finish while the first is

7 +//! still streaming. Each thread also has to find its own working directory

8 +//! again when its next tool call runs.

9 +

10 +use std::io::{BufRead, BufReader, Read, Write};

11 +use std::net::{TcpListener, TcpStream};

12 +use std::process::{Child, ChildStdin, Command, Stdio};

13 +use std::sync::atomic::{AtomicBool, Ordering};

14 +use std::sync::mpsc::{Receiver, channel};

15 +use std::sync::{Arc, Mutex};

16 +use std::time::{Duration, Instant};

17 +

18 +use serde_json::{Value, json};

19 +

20 +const TIMEOUT: Duration = Duration::from_secs(30);

21 +

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

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

24 +

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

26 + events

27 + .iter()

28 + .map(|event| format!("data: {event}\n\n"))

29 + .collect()

30 +}

31 +

32 +fn text_delta(text: &str) -> Value {

33 + json!({"choices": [{"delta": {"content": text}}]})

34 +}

35 +

36 +fn finish(reason: &str) -> Value {

37 + json!({"choices": [{"delta": {}, "finish_reason": reason}]})

38 +}

39 +

40 +fn read_marker_call() -> Value {

41 + json!({

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

43 + "index": 0,

44 + "id": "call_marker",

45 + "function": {"name": "read_file", "arguments": "{\"path\":\"marker.txt\"}"},

46 + }]}}]

47 + })

48 +}

49 +

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

51 +///

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

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

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

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

56 +/// before asking for the file.

57 +struct Endpoint {

58 + port: u16,

59 + requests: Arc<Mutex<Vec<Value>>>,

60 + release: Arc<AtomicBool>,

61 +}

62 +

63 +fn read_request(stream: &TcpStream) -> Option<Value> {

64 + let mut reader = BufReader::new(stream.try_clone().ok()?);

65 + let mut content_length = 0usize;

66 + loop {

67 + let mut line = String::new();

68 + if reader.read_line(&mut line).ok()? == 0 {

69 + return None;

70 + }

71 + let line = line.trim();

72 + if line.is_empty() {

73 + break;

74 + }

75 + if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {

76 + content_length = length.trim().parse().unwrap_or(0);

77 + }

78 + }

79 + let mut body = vec![0u8; content_length];

80 + reader.read_exact(&mut body).ok()?;

81 + serde_json::from_slice(&body).ok()

82 +}

83 +

84 +fn mentions(request: &Value, needle: &str) -> bool {

85 + request["messages"]

86 + .as_array()

87 + .is_some_and(|messages| messages.iter().any(|m| m.to_string().contains(needle)))

88 +}

89 +

90 +fn has_tool_result(request: &Value) -> bool {

91 + request["messages"]

92 + .as_array()

93 + .is_some_and(|messages| messages.iter().any(|m| m["role"] == "tool"))

94 +}

95 +

96 +fn start_endpoint() -> Endpoint {

97 + let listener = TcpListener::bind("127.0.0.1:0").expect("bind endpoint");

98 + let port = listener.local_addr().unwrap().port();

99 + let requests = Arc::new(Mutex::new(Vec::new()));

100 + let release = Arc::new(AtomicBool::new(false));

101 +

102 + let (seen, gate) = (Arc::clone(&requests), Arc::clone(&release));

103 + std::thread::spawn(move || {

104 + for stream in listener.incoming() {

105 + let Ok(mut stream) = stream else { continue };

106 + let (seen, gate) = (Arc::clone(&seen), Arc::clone(&gate));

107 + std::thread::spawn(move || {

108 + let Some(request) = read_request(&stream) else {

109 + return;

110 + };

111 + seen.lock().unwrap().push(request.clone());

112 +

113 + // No content-length: the body ends when the connection closes,

114 + // which lets a fragment go out before the rest is decided.

115 + let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\

116 + connection: close\r\n\r\n";

117 + if stream.write_all(head.as_bytes()).is_err() {

118 + return;

119 + }

120 +

121 + let body = if has_tool_result(&request) {

122 + sse(&[text_delta("done"), finish("stop")])

123 + } else {

124 + if mentions(&request, HOLD) {

125 + let first = sse(&[text_delta("holding")]);

126 + if stream.write_all(first.as_bytes()).is_err() || stream.flush().is_err() {

127 + return;

128 + }

129 + let deadline = Instant::now() + TIMEOUT;

130 + while !gate.load(Ordering::Acquire) && Instant::now() < deadline {

131 + std::thread::sleep(Duration::from_millis(20));

132 + }

133 + }

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

135 + };

136 + let _ = stream.write_all(body.as_bytes());

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

138 + let _ = stream.flush();

139 + });

140 + }

141 + });

142 +

143 + Endpoint {

144 + port,

145 + requests,

146 + release,

147 + }

148 +}

149 +

150 +struct AgentUnderTest {

151 + child: Child,

152 + stdin: ChildStdin,

153 + incoming: Receiver<Value>,

154 + next_id: u64,

155 + /// Messages read while waiting for something else, oldest first.

156 + seen: Vec<Value>,

157 +}

158 +

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

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

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

162 + .env("OPENAI_API_KEY", "test-key")

163 + .env("SIGIT_MODEL", "scripted-model")

164 + .env("SIGIT_CONFIG_DIR", config_dir)

165 + .env("SIGIT_MCP", "off")

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

167 + .env_remove("SIGIT_LOCAL_INFERENCE")

168 + .env_remove("SIGIT_MAX_TOOL_ROUNDS")

169 + .stdin(Stdio::piped())

170 + .stdout(Stdio::piped())

171 + .stderr(Stdio::null())

172 + .spawn()

173 + .expect("spawn sigit in ACP mode");

174 +

175 + let stdout = child.stdout.take().unwrap();

176 + let (message_tx, incoming) = channel();

177 + std::thread::spawn(move || {

178 + for line in BufReader::new(stdout).lines() {

179 + let Ok(line) = line else { break };

180 + if let Ok(message) = serde_json::from_str::<Value>(&line)

181 + && message_tx.send(message).is_err()

182 + {

183 + break;

184 + }

185 + }

186 + });

187 +

188 + let stdin = child.stdin.take().unwrap();

189 + AgentUnderTest {

190 + child,

191 + stdin,

192 + incoming,

193 + next_id: 0,

194 + seen: Vec::new(),

195 + }

196 +}

197 +

198 +impl AgentUnderTest {

199 + fn request(&mut self, method: &str, params: Value) -> u64 {

200 + self.next_id += 1;

201 + let id = self.next_id;

202 + let mut line =

203 + json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}).to_string();

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

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

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

207 + id

208 + }

209 +

210 + /// Read messages until one satisfies `wanted`, keeping every message read

211 + /// on the way in `seen`. Panics with `what` when none arrives in time.

212 + fn wait_for(&mut self, what: &str, wanted: impl Fn(&Value) -> bool) -> Value {

213 + let deadline = Instant::now() + TIMEOUT;

214 + loop {

215 + let remaining = deadline.saturating_duration_since(Instant::now());

216 + let Ok(message) = self.incoming.recv_timeout(remaining) else {

217 + panic!("timed out waiting for {what}");

218 + };

219 + self.seen.push(message.clone());

220 + if wanted(&message) {

221 + return message;

222 + }

223 + }

224 + }

225 +

226 + fn wait_for_response(&mut self, id: u64) -> Value {

227 + let message = self.wait_for(&format!("the response to request {id}"), |message| {

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

229 + });

230 + assert!(

231 + message.get("error").is_none(),

232 + "request {id} failed: {message}"

233 + );

234 + message

235 + }

236 +

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

238 + self.seen

239 + .iter()

240 + .any(|message| message["id"] == id && message.get("method").is_none())

241 + }

242 +

243 + fn new_session(&mut self, cwd: &std::path::Path) -> String {

244 + let id = self.request("session/new", json!({"cwd": cwd, "mcpServers": []}));

245 + self.wait_for_response(id)["result"]["sessionId"]

246 + .as_str()

247 + .expect("session id")

248 + .to_string()

249 + }

250 +

251 + fn send_prompt(&mut self, session_id: &str, text: &str) -> u64 {

252 + self.request(

253 + "session/prompt",

254 + json!({"sessionId": session_id, "prompt": [{"type": "text", "text": text}]}),

255 + )

256 + }

257 +}

258 +

259 +impl Drop for AgentUnderTest {

260 + fn drop(&mut self) {

261 + let _ = self.child.kill();

262 + let _ = self.child.wait();

263 + }

264 +}

265 +

266 +/// The tool result a follow-up request carries.

267 +fn tool_result(request: &Value) -> String {

268 + request["messages"]

269 + .as_array()

270 + .into_iter()

271 + .flatten()

272 + .filter(|message| message["role"] == "tool")

273 + .map(|message| message["content"].to_string())

274 + .collect()

275 +}

276 +

277 +#[test]

278 +fn a_second_session_opens_runs_and_finishes_while_the_first_is_streaming() {

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

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

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

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

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

284 + }

285 + std::fs::create_dir_all(dir.join("config")).unwrap();

286 +

287 + let endpoint = start_endpoint();

288 + let mut agent = spawn_agent(endpoint.port, &dir.join("config"));

289 + let id = agent.request(

290 + "initialize",

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

292 + );

293 + agent.wait_for_response(id);

294 +

295 + // Thread A starts a turn and the endpoint holds it open mid-stream.

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

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

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

299 + let update = &message["params"]["update"];

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

301 + && update["sessionUpdate"] == "agent_message_chunk"

302 + && update["content"]["text"]

303 + .as_str()

304 + .is_some_and(|text| text.contains("holding"))

305 + });

306 +

307 + // Thread B opens and runs a whole turn, tool call included, meanwhile.

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

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

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

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

312 + assert!(

313 + !agent.responded(prompt_a),

314 + "thread A finished before it was released, so it was never mid-turn"

315 + );

316 +

317 + // Thread A picks up where it was, in its own directory.

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

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

320 + assert_eq!(response_a["result"]["stopReason"], "end_turn");

321 +

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

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

324 + assert_eq!(

325 + follow_ups.len(),

326 + 2,

327 + "one tool round per thread: {requests:?}"

328 + );

329 + for request in follow_ups {

330 + let result = tool_result(request);

331 + if mentions(request, HOLD) {

332 + assert!(result.contains("marker-from-a"), "thread A read {result}");

333 + } else {

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

335 + }

336 + }

337 + // Neither thread's conversation reached the other's requests.

338 + for request in requests.iter().filter(|r| !mentions(r, HOLD)) {

339 + assert!(!mentions(request, "holding"), "{request}");

340 + }

341 +

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

343 +}