fix: honor WebSocket response storage with bounded sessions

ARCHITECTURE.md

@@ -178,7 +178,7 @@ reads `response.create` messages off the socket and drives the *same*
 `stream_id` values run concurrently, while requests in the same lane remain FIFO;
 requests without a `stream_id` share a default FIFO lane. The session admits at most
 64 active or queued requests and 12 MiB of aggregate request data. WebSocket sessions
-always force `stream: true, store: true`. Because axum's built-in graceful shutdown
+force `stream: true` and honor the requested `store` value. Because axum's built-in graceful shutdown
 doesn't wait for upgraded connections, `AppState` carries a separate
 `WebSocketTracker` so shutdown can drain in-flight sessions.
 
@@ -197,6 +197,59 @@ Errors are modeled by a dedicated `WsError` enum (`handler/websocket/error.rs`)
 than reusing the HTTP JSON-error path, since some failure modes (a dead socket) must
 not attempt to write a response.
 
+### Opt-in core continuation sessions (`executor/session.rs`)
+
+Core callers can use `ExecuteRequest::with_session` or `rehydrate_in_session` to
+retain response state without durable storage. The WebSocket multiplexer owns one
+`ResponseSessionGroup` per connection and keeps one session per lane, including idle
+lanes. No-session HTTP and split execution keep their existing behavior.
+
+The connection retains at most 128 lanes (including the default lane), 32,768 items
+and 16 MiB per checkpoint, and 32 MiB of aggregate serialized checkpoints. These are
+retention ceilings, not measured process-memory bounds. New lanes receive an
+immediate 429 once the lifetime lane limit is reached. Request-count and request-byte
+overloads also return immediate 429 responses without mutating retained state, so
+rejected work cannot invalidate an earlier accepted queued continuation. Other
+validation errors execute in lane order and evict only a matching referenced parent
+when routing is valid. Parent lookup happens when execution begins; accepted queues
+do not reserve parent snapshots. Existing fork and execution-failure eviction rules
+still apply. Disconnect aborts and joins request tasks, waits for active leases to
+release pinned state, and drops the entire connection group before the close handshake.
+
+A `ResponseSession` owns one latest canonical checkpoint and one execution slot.
+The executor pins a parent before inference and publishes completed or incomplete
+state before exposing terminal completion. Failed continuations discard only a
+referenced checkpoint owned by that session. Dropping the owner closes the session
+and rejects late publication. Callers still cancel and join active work explicitly;
+`wait_until_idle` waits for the execution lease to end, but does not cancel it.
+
+`ResponseSessionGroup` allows independent serial members to find and pin each
+other's latest checkpoints. Failed forks cannot evict the source member's state.
+The group bounds lifetime member count, each checkpoint's items and serialized
+bytes, and aggregate retained bytes. Shared parent references count once; replaced
+parents still pinned by active work and prepared checkpoints awaiting persistence
+remain charged until their last reference is released. Reservation happens before
+durable writes, and failure or cancellation returns unused capacity. Replacement
+requires room for both old and new snapshots until publication. These retention
+budgets are not a bound on temporary allocations, execution copies or process
+memory. Scheduling, FIFO queues and active-work limits remain caller concerns.
+
+Response-scoped session history records reasoning, messages and calls in inference-round order,
+then built-in call outputs. Public output is accumulated separately and is not
+appended to retained history twice. Compaction replaces the canonical window while
+preserving MCP discovery records needed for orchestration. Durable restoration
+and replayed compaction input select the effective compacted window before validating current calls;
+obsolete stored rows are not deleted or charged to that retained window.
+
+For response-scoped session execution, `store: false` has no durable writes or
+database fallback. A stored child of a transient parent persists the complete
+canonical window without creating a row or dangling database reference for that
+parent. Explicit `conversation_id` requests keep the existing durable Conversations
+policy and append output only through the conversation handler. Their session lease
+still serializes execution without recording a second copy. Core commit supports
+prewarming without inference; WebSocket `generate: false` uses that same session
+rehydration and commit path.
+
 ### `handler/common.rs`
 
 Transport helpers shared by the HTTP and WS handlers: body reading with a shared size

TERMINOLOGY.md

@@ -92,7 +92,8 @@ round.
 ### conversation state
 
 The prior items and metadata made available to a later turn. State may be managed with a conversation, chained with
-`previous_response_id`, or replayed manually.
+`previous_response_id`, or replayed manually. WebSocket response state may also be
+retained transiently by the active connection; it is distinct from durable conversation storage.
 
 ### stored response
 
@@ -106,7 +107,8 @@ mechanism could be either `previous_response_id` or a conversation.
 
 ### previous response ID
 
-The response identifier passed in the `previous_response_id` field to continue from a prior stored response. In prose,
+The response identifier passed in the `previous_response_id` field to continue from prior response state. That state
+may be durable or cached on the active WebSocket connection. In prose,
 write **previous response ID**; in code and wire-format discussion, use `previous_response_id`.
 
 ### rehydration
@@ -127,7 +129,8 @@ Describes a flow in which the service retains or resolves prior state, such as R
 ### stateless
 
 Describes a flow in which the request supplies all required context and the service does not rely on retained response
-or conversation state. `store: false` disables stored-response state, although callers may still replay prior items.
+or conversation state. `store: false` disables durable response storage, although callers may still replay prior items
+or continue from the active WebSocket connection's transient checkpoint. An explicit conversation remains durable.
 
 ### compaction
 

crates/agentic-llm-d/src/context.rs

@@ -65,6 +65,7 @@ impl From<SplitContext> for RequestContext {
             conversation_id: wire.conversation_id,
             // Conversation mode is rejected, so there is no version to resume.
             conversation_version: None,
+            continuation: None,
         }
     }
 }

crates/agentic-llm-d/tests/split_execution_integration.rs

@@ -287,12 +287,12 @@ async fn a_function_call_stream_passes_strict_validation() {
 
 #[tokio::test]
 async fn relayed_json_tool_call_ids_are_validated_before_persistence() {
-    assert_relayed_tool_call_ids_are_validated(false).await;
+    Box::pin(assert_relayed_tool_call_ids_are_validated(false)).await;
 }
 
 #[tokio::test]
 async fn relayed_sse_tool_call_ids_are_validated_before_persistence() {
-    assert_relayed_tool_call_ids_are_validated(true).await;
+    Box::pin(assert_relayed_tool_call_ids_are_validated(true)).await;
 }
 
 async fn assert_relayed_tool_call_ids_are_validated(stream: bool) {
@@ -402,12 +402,12 @@ async fn assert_relayed_tool_call_ids_are_validated(stream: bool) {
 
 #[tokio::test]
 async fn relayed_json_call_id_cannot_reuse_continued_history() {
-    assert_relayed_call_id_cannot_reuse_continued_history(false).await;
+    Box::pin(assert_relayed_call_id_cannot_reuse_continued_history(false)).await;
 }
 
 #[tokio::test]
 async fn relayed_sse_call_id_cannot_reuse_continued_history() {
-    assert_relayed_call_id_cannot_reuse_continued_history(true).await;
+    Box::pin(assert_relayed_call_id_cannot_reuse_continued_history(true)).await;
 }
 
 async fn assert_relayed_call_id_cannot_reuse_continued_history(stream: bool) {

crates/agentic-server-core/src/executor/compaction.rs

@@ -417,6 +417,7 @@ pub(crate) async fn compact_items(
         response_id: uuid7_str("resp_"),
         conversation_id: None,
         conversation_version: None,
+        continuation: None,
     };
     let mut agent = agent_pipeline(ctx, None, None);
     let response =
@@ -469,6 +470,9 @@ pub(crate) async fn maybe_compact_context(
     let (compacted, usage) = compact_items(&model, input, instructions.as_deref(), exec_ctx, auth).await?;
     ctx.enriched_request.input = ResponsesInput::Items(compacted.clone());
     ctx.new_input_items = compacted;
+    if let Some(continuation) = &mut ctx.continuation {
+        continuation.mark_history_replaced();
+    }
     Ok(Some(usage))
 }
 
@@ -605,6 +609,7 @@ mod tests {
             response_id: "resp_test".to_owned(),
             conversation_id: None,
             conversation_version: None,
+            continuation: None,
         }
     }
 
@@ -622,7 +627,7 @@ mod tests {
     async fn mock_execution_context(response_store: ResponseStore) -> (ExecutionContext, tokio::task::JoinHandle<()>) {
         let app = Router::new().route(
             "/v1/responses",
-            post(|| async {
+            post(|_body: axum::body::Bytes| async {
                 axum::Json(serde_json::json!({
                     "id": "resp_upstream",
                     "object": "response",

crates/agentic-server-core/src/executor/engine.rs

@@ -25,7 +25,7 @@ use crate::executor::inference::DONE_MARKER;
 use crate::executor::persist::persist_if_needed;
 use crate::executor::pipeline::{AgentPipeline, emit_deferred_stream_events};
 use crate::executor::prepare::prepare_request_tools;
-use crate::executor::rehydrate::{prepare_reasoning_for_vllm, rehydrate_conversation, validate_reasoning_for_vllm};
+use crate::executor::rehydrate::{prepare_reasoning_for_vllm, validate_reasoning_for_vllm};
 use crate::executor::request::{ExecutionContext, RequestContext};
 use crate::executor::response_budget::ExecutorResponseBudget;
 #[cfg(test)]
@@ -199,6 +199,34 @@ fn prepare_initial_reasoning_for_vllm(input: &mut ResponsesInput, round: usize,
     Ok(())
 }
 
+fn record_round_history(
+    ctx: &mut RequestContext,
+    output_items: &[OutputItem],
+    registry: &ToolRegistry,
+    public_output_count: usize,
+) {
+    // Explicit conversations append public output through their durable handler;
+    // the session lease still serializes execution but must not record it twice.
+    if let Some(continuation) = ctx
+        .continuation
+        .as_mut()
+        .filter(|_| ctx.original_request.conversation_id.is_none())
+    {
+        // The canonical sequence includes reasoning and intermediate messages in
+        // their original positions, followed by this round's tool call outputs.
+        // Discovery records are appended separately from the public response.
+        ctx.new_input_items.extend(
+            output_items
+                .iter()
+                .filter(|item| !matches!(item, OutputItem::McpListTools(_)))
+                .filter_map(OutputItem::to_input_item),
+        );
+        continuation.mark_outputs_recorded(public_output_count);
+    } else {
+        append_gateway_calls_to_new_input(ctx, output_items, registry);
+    }
+}
+
 /// Request-scoped owner of registry-backed tool orchestration and its shared byte budget.
 struct EngineOrchestration<'a> {
     agent: &'a mut AgentPipeline,
@@ -282,11 +310,17 @@ impl<'a> EngineOrchestration<'a> {
                 .await?;
             let public_output = public_output_items(&current_output, &self.registry, &gateway_results)?;
             combined_output.extend(public_output);
+            record_round_history(
+                &mut self.agent.request,
+                &current_output,
+                &self.registry,
+                combined_output.len(),
+            );
 
             // A terminal incomplete response may still contain completed gateway
             // calls. Record those results, but never start another inference round.
             if payload.status == "incomplete" {
-                self.record_round_input(&current_output, gateway_results);
+                self.record_gateway_results(gateway_results);
                 finalize_loop(&mut payload, combined_output, combined_usage, &self.agent.request);
                 let tool_search_metadata = self.agent.take_tool_search_metadata();
                 return Ok((payload, tool_search_metadata));
@@ -297,7 +331,7 @@ impl<'a> EngineOrchestration<'a> {
                 // are handed back to the caller. Gateway calls in the same round are
                 // still recorded so the returned conversation is complete.
                 LoopDecision::RequiresClientAction => {
-                    self.record_round_input(&current_output, gateway_results);
+                    self.record_gateway_results(gateway_results);
                     finalize_loop(&mut payload, combined_output, combined_usage, &self.agent.request);
                     let tool_search_metadata = self.agent.take_tool_search_metadata();
                     return Ok((payload, tool_search_metadata));
@@ -314,7 +348,7 @@ impl<'a> EngineOrchestration<'a> {
                 // The final round's gateway calls and outputs are recorded so a
                 // continuation is not fed a dangling tool call.
                 LoopDecision::Incomplete(reason) => {
-                    self.record_round_input(&current_output, gateway_results);
+                    self.record_gateway_results(gateway_results);
                     finalize_loop(&mut payload, combined_output, combined_usage, &self.agent.request);
                     "incomplete".clone_into(&mut payload.status);
                     payload.incomplete_details = Some(IncompleteDetails { reason: Some(reason) });
@@ -325,16 +359,15 @@ impl<'a> EngineOrchestration<'a> {
                 LoopDecision::Continue => {
                     self.agent.request.enriched_request.tool_choice = Some(ToolChoice::Auto);
                     append_output_items_to_input(&mut self.agent.request.enriched_request.input, &current_output);
-                    self.record_round_input(&current_output, gateway_results);
+                    self.record_gateway_results(gateway_results);
                 }
             }
         }
 
         unreachable!("the final round returns Done, RequiresClientAction, or Incomplete");
     }
 
-    fn record_round_input(&mut self, output: &[OutputItem], results: Vec<GatewayCallResult>) {
-        append_gateway_calls_to_new_input(&mut self.agent.request, output, &self.registry);
+    fn record_gateway_results(&mut self, results: Vec<GatewayCallResult>) {
         append_tool_outputs(
             &mut self.agent.request,
             results.into_iter().map(|result| result.input_item).collect(),
@@ -450,6 +483,9 @@ async fn run_compaction_trigger(
         unreachable!("compact_items always appends a compaction item");
     };
     ctx.new_input_items = compacted;
+    if let Some(continuation) = &mut ctx.continuation {
+        continuation.mark_history_replaced();
+    }
     let mut payload = ResponsePayload {
         id: ctx.response_id.clone(),
         object: "response".to_owned(),
@@ -683,6 +719,7 @@ pub struct ExecuteRequest {
     payload: RequestPayload,
     exec_ctx: Arc<ExecutionContext>,
     client_auth: Option<String>,
+    continuation: Option<super::session::ResponseContinuation>,
 }
 
 impl ExecuteRequest {
@@ -692,6 +729,7 @@ impl ExecuteRequest {
             payload,
             exec_ctx,
             client_auth: None,
+            continuation: None,
         }
     }
 
@@ -702,6 +740,15 @@ impl ExecuteRequest {
         self
     }
 
+    /// Retain this turn's continuation state in the supplied serial session.
+    ///
+    /// # Errors
+    /// Returns an error when the session is busy or closed.
+    pub fn with_session(mut self, session: &super::ResponseSession) -> ExecutorResult<Self> {
+        self.continuation = Some(session.begin(self.payload.previous_response_id.as_deref())?);
+        Ok(self)
+    }
+
     /// Execute one stateful conversation turn.
     ///
     /// Returns `Either::Left(ResponsePayload)` for non-streaming requests, or
@@ -720,7 +767,8 @@ impl ExecuteRequest {
             tools = self.payload.tools.as_ref().map_or(0, Vec::len),
             "executor received responses request"
         );
-        let ctx = rehydrate_conversation(self.payload, &self.exec_ctx).await?;
+        let ctx =
+            super::rehydrate::rehydrate_with_continuation(self.payload, &self.exec_ctx, self.continuation).await?;
         if !ctx.enriched_request.input.has_compaction_trigger() {
             validate_reasoning_for_vllm(&ctx.enriched_request.input)?;
         }
@@ -838,10 +886,18 @@ mod tests {
             "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_upstream\",\"status\":\"completed\",\"usage\":null}}\n\n",
             "data: [DONE]\n\n",
         );
-        let app = axum::Router::new().route(
-            "/v1/responses",
-            axum::routing::post(|| async { ([(axum::http::header::CONTENT_TYPE, "text/event-stream")], UPSTREAM_SSE) }),
-        );
+        let app = axum::Router::new()
+            .route(
+                "/v1/responses",
+                axum::routing::post(|_body: axum::body::Bytes| async {
+                    ([(axum::http::header::CONTENT_TYPE, "text/event-stream")], UPSTREAM_SSE)
+                }),
+            )
+            // Read the oversized test request before replying, allowing JSON framing
+            // above the response budget without relying on an early HTTP response.
+            .layer(axum::extract::DefaultBodyLimit::max(
+                MAX_EXECUTOR_RESPONSE_BYTES + 64 * 1024,
+            ));
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
             .await
             .expect("bind streaming mock inference server");
@@ -874,6 +930,7 @@ mod tests {
             response_id: "resp_test".to_owned(),
             conversation_id: None,
             conversation_version: None,
+            continuation: None,
         };
         let mut exec_ctx = ExecutionContext::new(
             ConversationHandler::new(ConversationStore::disabled()),
@@ -934,6 +991,7 @@ mod tests {
             response_id: "resp_mcp".to_owned(),
             conversation_id: None,
             conversation_version: None,
+            continuation: None,
         };
         let plain_payload: RequestPayload = serde_json::from_value(serde_json::json!({
             "model": "test-model",
@@ -948,6 +1006,7 @@ mod tests {
             response_id: "resp_plain".to_owned(),
             conversation_id: None,
             conversation_version: None,
+            continuation: None,
         };
         let mut exec_ctx = ExecutionContext::new(
             ConversationHandler::new(ConversationStore::disabled()),