@sigitcodereview-old / sigit / commits / be21081

Fork 0

A modular local inference LLM coding agent. Core component of siGit Factory.

Harden ACP session restoration

Fail safely until pending local history can be applied, preserve desired models across transient restore failures, discard cancelled trailing prompts, and avoid redundant fork parking. Co-Authored-By: siGit Code <noreply@sigit.si>

Seto Elkahfi committed Sep 17, 2026 at 08:40 UTC be210818e08d4d7459a1e57a1d22043cbecb075a

2 files changed +86 -17

src/backend.rs

+50 -8

@@ -221,12 +221,18 @@ impl LocalBackend {

221 }

222 }

223

224 - async fn apply_pending_history(&self) {

225 - if self.engine.info().await.status != EngineStatus::Ready {

226 - return;

224 + async fn apply_pending_history(&self) -> Result<(), BackendError> {

225 + if self.pending_history.lock().await.is_none() {

226 + return Ok(());

227 + }

228 + let status = self.engine.info().await.status;

229 + if status != EngineStatus::Ready {

230 + return Err(format!(

231 + "on-device model is not ready to restore session history (status: {status})"

232 + ));

233 }

234 let Some(history) = self.pending_history.lock().await.take() else {

229 - return;

235 + return Ok(());

236 };

237

238 self.engine.clear_history().await;

@@ -248,6 +254,7 @@ impl LocalBackend {

254 };

255 self.engine.push_history(message).await;

256 }

257 + Ok(())

258 }

259 }

260

@@ -270,7 +277,7 @@ impl InferenceBackend for LocalBackend {

277 tools: &[ToolSpec],

278 sink: Option<&TokenSink>,

279 ) -> Result<TurnResult, BackendError> {

273 - self.apply_pending_history().await;

280 + self.apply_pending_history().await?;

281 // onde's tool-aware path is non-streaming: it has to buffer the whole

282 // reply to detect tool calls. We can only stream when no tools are on

283 // offer (a plain answer), which is exactly the tools-disabled case.

@@ -301,7 +308,7 @@ impl InferenceBackend for LocalBackend {

308 allow_tool_calls: bool,

309 sink: Option<&TokenSink>,

310 ) -> Result<TurnResult, BackendError> {

304 - self.apply_pending_history().await;

311 + self.apply_pending_history().await?;

312 let onde_results: Vec<onde::inference::ToolResult> = results

313 .into_iter()

314 .map(|result| onde::inference::ToolResult {

@@ -365,11 +372,15 @@ impl InferenceBackend for LocalBackend {

372

373 async fn restore_history(&self, history: Vec<serde_json::Value>) {

374 *self.pending_history.lock().await = Some(history);

368 - self.apply_pending_history().await;

375 + if self.engine.info().await.status == EngineStatus::Ready {

376 + // Ready was just observed, so failure here can only mean a status

377 + // transition; keep the pending copy for the next inference call.

378 + let _ = self.apply_pending_history().await;

379 + }

380 }

381

382 async fn compact_history(&self, keep_last: usize) -> Result<(), BackendError> {

372 - self.apply_pending_history().await;

383 + self.apply_pending_history().await?;

384 let snapshot = self.engine.history().await;

385 // One plain (tool-free) inference round produces the summary. On error

386 // history is untouched — send_message only mutates it on success, and

@@ -1240,6 +1251,17 @@ pub fn carryover_history(snapshot: Vec<serde_json::Value>) -> Vec<serde_json::Va

1251 }

1252 }

1253

1254 + // Session switches happen only after the serialized turn has ended. A

1255 + // trailing user message therefore belongs to a cancelled/failed inference

1256 + // request with no answer; carrying it would make the next activation feed

1257 + // the model an orphaned prompt that the client considers cancelled.

1258 + while carried

1259 + .last()

1260 + .is_some_and(|message| message["role"] == "user")

1261 + {

1262 + carried.pop();

1263 + }

1264 +

1265 carried

1266 }

1267

@@ -1433,6 +1455,19 @@ mod tests {

1455 assert_eq!(carried[1]["content"], "hi");

1456 }

1457

1458 + #[test]

1459 + fn carryover_drops_a_cancelled_trailing_user_message() {

1460 + let carried = carryover_history(vec![

1461 + serde_json::json!({ "role": "system", "content": "prompt" }),

1462 + serde_json::json!({ "role": "user", "content": "completed question" }),

1463 + serde_json::json!({ "role": "assistant", "content": "completed answer" }),

1464 + serde_json::json!({ "role": "user", "content": "cancelled question" }),

1465 + ]);

1466 +

1467 + assert_eq!(carried.len(), 2, "{carried:#?}");

1468 + assert_eq!(carried.last().unwrap()["content"], "completed answer");

1469 + }

1470 +

1471 #[test]

1472 fn carryover_strips_a_tool_call_that_never_got_a_result() {

1473 // Switching mid-turn: the assistant asked for a tool, nothing answered.

@@ -1613,6 +1648,13 @@ mod tests {

1648 .contains("thread A")),

1649 "restore_history must replace, not append to, the local engine: {restored:#?}"

1650 );

1651 +

1652 + let error = backend

1653 + .send_message_with_tools("must not run", &[], None)

1654 + .await

1655 + .unwrap_err();

1656 + assert!(error.contains("not ready"), "{error}");

1657 + assert_eq!(backend.history_snapshot().await, restored);

1658 }

1659

1660 /// Minimal scripted OpenAI-compatible endpoint: accepts one HTTP request on

src/main.rs

+36 -9

@@ -1526,7 +1526,6 @@ impl SiGitAgent {

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

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

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

1529 - state.model_id = self.current_model.lock().unwrap().model_id.clone();

1529 }

1530 }

1531

@@ -1539,8 +1538,15 @@ impl SiGitAgent {

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

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

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

1542 - let key = session_id.to_string();

1541 self.park_active_session().await;

1542 + self.open_parked_session(session_id, state).await;

1543 + }

1544 +

1545 + /// Install a new session after the caller has already parked the live one.

1546 + /// Fork needs this split so it can read an up-to-date source state without

1547 + /// making `open_session` snapshot the same live backend a second time.

1548 + async fn open_parked_session(&self, session_id: &SessionId, state: SessionState) {

1549 + let key = session_id.to_string();

1550 self.sessions

1551 .lock()

1552 .unwrap()

@@ -1643,7 +1649,6 @@ impl SiGitAgent {

1649 log::warn!(

1650 "session({key}): cannot restore cloud model without a signed-in account"

1651 );

1646 - self.record_active_model();

1652 }

1653 return;

1654 }

@@ -1655,7 +1660,6 @@ impl SiGitAgent {

1660 }

1661 Err(error) => {

1662 log::warn!("session({key}): could not restore model {target}: {error}");

1658 - self.record_active_model();

1663 }

1664 }

1665 }

@@ -1702,9 +1706,6 @@ impl SiGitAgent {

1706 /// wrong project is worse than one that isn't listed at all.

1707 async fn persist_session(&self, session_id: &SessionId, snapshot: &[serde_json::Value]) {

1708 let key = session_id.to_string();

1705 - if self.active_session.lock().unwrap().as_deref() == Some(key.as_str()) {

1706 - self.record_active_model();

1707 - }

1709 if let Err(error) = session_store::save(&key, snapshot) {

1710 log::warn!("session({session_id}) save failed: {error}");

1711 return;

@@ -1881,7 +1882,7 @@ impl SiGitAgent {

1882 cx.send_notification(SessionNotification::new(new_id.clone(), update))

1883 .ok();

1884 }

1884 - self.open_session(&new_id, state).await;

1885 + self.open_parked_session(&new_id, state).await;

1886 log::info!(

1887 "fork_session: copied {copied} message(s), replayed {replayed} update(s) for {new_id}"

1888 );

@@ -2414,6 +2415,9 @@ impl SiGitAgent {

2415 tool_call_id: tc.id.clone(),

2416 content: output,

2417 });

2418 + // Cancellation cannot preempt a synchronous foreground tool:

2419 + // it is observed here as soon as that tool exits. Background

2420 + // commands remain cancellable through `kill_command`.

2421 if cancellation.cancelled.load(Ordering::Acquire) {

2422 for pending in &result.tool_calls[call_index + 1..] {

2423 tool_results.push(BackendToolResult {

@@ -2892,6 +2896,12 @@ impl SiGitAgent {

2896 "set_session_config_option: {} is already the active selection, skipping",

2897 current.display_name

2898 );

2899 + // A restored preferred model may have fallen back temporarily.

2900 + // Explicitly picking the live model adopts that fallback as the

2901 + // session's new preference even though no backend switch runs.

2902 + drop(current);

2903 + self.record_active_model();

2904 + let current = self.current_model.lock().unwrap();

2905 let config_options = build_config_options(&current, &args.session_id.to_string());

2906 return Ok(SetSessionConfigOptionResponse::new(config_options));

2907 }

@@ -4655,11 +4665,12 @@ mod tests {

4665 .join("\n")

4666 }

4667

4658 - /// Append a user message to the live conversation.

4668 + /// Append one completed exchange to the live conversation.

4669 async fn say(agent: &SiGitAgent, text: &str) {

4670 let backend = agent.backend.lock().await.clone();

4671 let mut history = backend.history_snapshot().await;

4672 history.push(serde_json::json!({ "role": "user", "content": text }));

4673 + history.push(serde_json::json!({ "role": "assistant", "content": "acknowledged" }));

4674 backend.restore_history(history).await;

4675 }

4676

@@ -4851,6 +4862,22 @@ mod tests {

4862 assert_eq!(sessions["thread-b"].model_id, "sigit-cloud:test-b");

4863 drop(sessions);

4864

4865 + // A transient restore failure may leave another model live, but parking

4866 + // must not erase the session's desired selection; its next activation

4867 + // should retry that preference.

4868 + agent

4869 + .sessions

4870 + .lock()

4871 + .unwrap()

4872 + .get_mut("thread-b")

4873 + .unwrap()

4874 + .model_id = "sigit-cloud:preferred".into();

4875 + agent.park_active_session().await;

4876 + assert_eq!(

4877 + agent.sessions.lock().unwrap()["thread-b"].model_id,

4878 + "sigit-cloud:preferred"

4879 + );

4880 +

4881 std::env::set_current_dir(prev_cwd).unwrap();

4882 tools::set_active_session(None);

4883 workspace::replace_additional_roots(Vec::new());