From fa3ebb4450562e18f3c8e1ab5badfaebbd7c69c4 Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Tue, 6 Oct 2026 10:32:25 -0700 Subject: [PATCH] fix(libsy): preserve plan-execute sessions at capacity Signed-off-by: Ryan Lempka --- crates/libsy/src/algorithms/plan_execute.rs | 109 +++++++++++++++--- .../plan_execute_routing.md | 16 +++ 2 files changed, 109 insertions(+), 16 deletions(-) diff --git a/crates/libsy/src/algorithms/plan_execute.rs b/crates/libsy/src/algorithms/plan_execute.rs index 74e4063a2..422003e0b 100644 --- a/crates/libsy/src/algorithms/plan_execute.rs +++ b/crates/libsy/src/algorithms/plan_execute.rs @@ -80,40 +80,45 @@ impl PlanExecute { }) } - fn phase(&self, request: &Request) -> Phase { + fn phase(&self, request: &Request) -> Result { let signals = ToolSignals::from_request(request, None); let mutation_seen = signals.edit_count > 0 || signals.write_count > 0; let Some(identity) = RoutingIdentity::from_request(request) else { - return if mutation_seen { + return Ok(if mutation_seen { Phase::Handoff } else { Phase::Plan - }; + }); }; + let is_session_final = request + .metadata + .as_ref() + .and_then(|metadata| metadata.session_final) + == Some(true); let mut sessions = self.executing_sessions.lock(); let phase = if sessions.contains(&identity) { Phase::Execute } else if mutation_seen { - if sessions.len() >= MAX_EXECUTING_SESSIONS - && let Some(evicted) = sessions.iter().next().cloned() - { - sessions.remove(&evicted); + if !is_session_final { + if sessions.len() >= MAX_EXECUTING_SESSIONS { + return Err(LibsyError::AlgorithmError { + message: format!( + "plan_execute reached its limit of {MAX_EXECUTING_SESSIONS} executing sessions. \ + Finish an existing session with session_final before retrying the handoff" + ), + }); + } + sessions.insert(identity.clone()); } - sessions.insert(identity.clone()); Phase::Handoff } else { Phase::Plan }; - if request - .metadata - .as_ref() - .and_then(|metadata| metadata.session_final) - == Some(true) - { + if is_session_final { sessions.remove(&identity); } - phase + Ok(phase) } fn replay_planner_reasoning_as_text(request: &mut Request) -> usize { @@ -166,7 +171,7 @@ impl Algorithm for PlanExecute { driver: Driver, mut request: Request, ) -> Result { - match self.phase(&request) { + match self.phase(&request)? { Phase::Plan => { prepend_system_prompt(&mut request, &self.config.planning_prompt); tracing::debug!(phase = "plan", "plan-execute selected capable tier"); @@ -347,6 +352,78 @@ mod tests { assert_eq!(selected, CAPABLE); } + #[tokio::test] + async fn capacity_preserves_execution_after_compaction() { + let algorithm = algorithm(PlanExecuteConfig::default()); + let sessions: Vec<_> = (0..MAX_EXECUTING_SESSIONS) + .map(|index| format!("task-{index}")) + .collect(); + for session in &sessions { + let first_edit = request( + vec![tool_call("Write", json!({"file_path": "task.py"}))], + Some(session), + ); + let (selected, _) = route_and_capture(Arc::clone(&algorithm), first_edit).await; + assert_eq!(selected, EFFICIENT); + } + + let overflow = request( + vec![tool_call("Write", json!({"file_path": "task.py"}))], + Some("overflow"), + ); + let (driver, _) = Driver::new("plan_execute", Arc::new(models())); + let result = Arc::clone(&algorithm).route(driver, overflow.clone()).await; + assert!(matches!( + result, + Err(LibsyError::AlgorithmError { message }) + if message.contains("limit of 4096 executing sessions") + )); + + for session in &sessions { + let mut compacted = request( + vec![Message::text(Role::User, "Continue after compaction")], + Some(session), + ); + if Some(session) == sessions.last() { + compacted.metadata.as_mut().unwrap().session_final = Some(true); + } + let (selected, routed) = route_and_capture(Arc::clone(&algorithm), compacted).await; + assert_eq!(selected, EFFICIENT, "session {session} lost its latch"); + assert!(routed.llm_request.instructions.is_empty()); + } + + let (selected, _) = route_and_capture(Arc::clone(&algorithm), overflow).await; + assert_eq!(selected, EFFICIENT); + let compacted = request( + vec![Message::text(Role::User, "Continue after compaction")], + Some("overflow"), + ); + let (selected, routed) = route_and_capture(algorithm, compacted).await; + assert_eq!(selected, EFFICIENT); + assert!(routed.llm_request.instructions.is_empty()); + } + + #[tokio::test] + async fn final_handoff_does_not_evict_an_executing_session_at_capacity() { + let algorithm = Arc::new(PlanExecute::new(PlanExecuteConfig::default()).unwrap()); + algorithm.executing_sessions.lock().extend( + (0..MAX_EXECUTING_SESSIONS) + .map(|index| RoutingIdentity::Session(format!("task-{index}"))), + ); + let mut final_request = request( + vec![tool_call("Write", json!({"file_path": "task.py"}))], + Some("final-handoff"), + ); + final_request.metadata.as_mut().unwrap().session_final = Some(true); + + let (selected, routed) = route_and_capture(algorithm.clone(), final_request).await; + assert_eq!(selected, EFFICIENT); + assert!(routed.llm_request.instructions.is_empty()); + let sessions = algorithm.executing_sessions.lock(); + assert_eq!(sessions.len(), MAX_EXECUTING_SESSIONS); + assert!(!sessions.contains(&RoutingIdentity::Session("final-handoff".to_string()))); + } + #[tokio::test] async fn mutation_without_a_session_uses_the_efficient_tier() { let messages = vec![tool_call( diff --git a/docs/routing_algorithms/plan_execute_routing.md b/docs/routing_algorithms/plan_execute_routing.md index 6efab06b1..a94bb46c4 100644 --- a/docs/routing_algorithms/plan_execute_routing.md +++ b/docs/routing_algorithms/plan_execute_routing.md @@ -32,6 +32,22 @@ The first edit or write routes the full trajectory to the efficient target and latches that choice by session ID. A failed edit still triggers the handoff. Without a session ID, the first mutation must remain in the request history. +## Session capacity + +Each router instance retains up to 4,096 executing identities in memory. Root +requests use the session ID. Subagent requests use the session ID and agent ID. +Once retained, an identity stays on the efficient target after history compaction +until a request marks it with `session_final: true` or the router restarts. + +At capacity, a new identity's handoff returns an error before calling a model. +Existing identities keep their execution phase. Mark an existing identity's last +request with `session_final: true` to free its slot, then retry the handoff with +the mutation history intact. The server accepts this flag through the +`x-switchyard-session-final: true` header. + +A handoff marked final needs no saved slot and can still run at capacity. +Requests that are still planning or have no stable identity also use no slots. + ## Responses API history requirement Plan/execute needs the conversation history to detect edits and hand the task to