fix: reject stale conversation turns

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

@@ -198,6 +198,7 @@ pub(crate) async fn compact_items(
         new_input_items: Vec::new(),
         response_id: uuid7_str("resp_"),
         conversation_id: None,
+        conversation_version: None,
     };
     let response = fetch_blocking_payload(&ctx, exec_ctx, auth).await?;
     let summary = completed_summary_text(&response)?;

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

@@ -19,6 +19,16 @@ pub enum ExecutorError {
     #[error("failed to persist response")]
     Persistence(#[source] Box<ExecutorError>),
 
+    /// A persisted conversation changed after its history was read.
+    ///
+    /// The storage source is retained for internal diagnostics while the
+    /// display message remains safe to send to API clients.
+    #[error("conversation changed while the response was being generated; retry the request")]
+    ConversationLocked {
+        #[source]
+        source: StorageError,
+    },
+
     /// The LLM backend returned a non-2xx status or was unreachable.
     #[error("LLM request failed ({status}): {body}")]
     LLMRequest { status: StatusCode, body: String },
@@ -71,33 +81,74 @@ pub enum ExecutorError {
 }
 
 impl ExecutorError {
+    fn client_visible_error(&self) -> &Self {
+        match self {
+            Self::Persistence(source) if source.contains_conversation_locked() => source.client_visible_error(),
+            _ => self,
+        }
+    }
+
+    fn contains_conversation_locked(&self) -> bool {
+        match self {
+            Self::ConversationLocked { .. } => true,
+            Self::Persistence(source) => source.contains_conversation_locked(),
+            _ => false,
+        }
+    }
+
     /// HTTP status code that best represents this error to an API caller.
     #[must_use]
     pub fn http_status(&self) -> StatusCode {
-        match self {
+        match self.client_visible_error() {
             Self::Storage(e) if e.is_not_found() => StatusCode::NOT_FOUND,
             Self::LLMRequest { status, .. } => *status,
-            Self::Tool(ToolError::Config(_)) | Self::InvalidRequest(_) | Self::JsonError(_) => StatusCode::BAD_REQUEST,
+            Self::ConversationLocked { .. }
+            | Self::Tool(ToolError::Config(_))
+            | Self::InvalidRequest(_)
+            | Self::JsonError(_) => StatusCode::BAD_REQUEST,
             Self::Tool(ToolError::Execution(_)) | Self::CompactionFailed { .. } => StatusCode::BAD_GATEWAY,
             Self::ParseError(_) => StatusCode::UNPROCESSABLE_ENTITY,
             _ => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 
-    /// Short machine-readable error code for the API error envelope.
+    /// Machine-readable error type for the API error envelope.
     #[must_use]
-    pub fn error_code(&self) -> &'static str {
-        match self {
+    pub fn error_type(&self) -> &'static str {
+        match self.client_visible_error() {
+            Self::ConversationLocked { .. }
+            | Self::Tool(ToolError::Config(_))
+            | Self::InvalidRequest(_)
+            | Self::ParseError(_)
+            | Self::JsonError(_) => "invalid_request_error",
             Self::Storage(e) if e.is_not_found() => "not_found",
             Self::LLMRequest { .. } | Self::CompactionFailed { .. } => "upstream_error",
-            Self::Tool(ToolError::Config(_)) | Self::InvalidRequest(_) | Self::ParseError(_) | Self::JsonError(_) => {
-                "invalid_request_error"
-            }
             Self::Tool(ToolError::Execution(_)) => "tool_error",
             _ => "server_error",
         }
     }
 
+    /// Short machine-readable error code for the API error envelope.
+    #[must_use]
+    pub fn error_code(&self) -> &'static str {
+        match self.client_visible_error() {
+            Self::ConversationLocked { .. } => "conversation_locked",
+            other => other.error_type(),
+        }
+    }
+
+    /// Request parameter associated with the API error, when applicable.
+    #[must_use]
+    pub fn error_param(&self) -> Option<&'static str> {
+        matches!(self.client_visible_error(), Self::ConversationLocked { .. }).then_some("conversation")
+    }
+
+    /// Client-safe message for the API error envelope.
+    #[must_use]
+    pub fn error_message(&self) -> String {
+        self.client_visible_error().to_string()
+    }
+
     /// Serialise the error into the HTTP response body bytes.
     ///
     /// `LLMRequest` bodies are forwarded verbatim; all other variants are
@@ -107,10 +158,16 @@ impl ExecutorError {
         match self {
             Self::LLMRequest { body, .. } => body.into_bytes(),
             other => {
+                let error_type = other.error_type();
                 let code = other.error_code();
-                serialize_to_vec_or_default(&serde_json::json!({
-                    "error": { "message": other.to_string(), "type": code, "code": code }
-                }))
+                let mut error = serde_json::Map::new();
+                error.insert("message".to_owned(), serde_json::json!(other.error_message()));
+                error.insert("type".to_owned(), serde_json::json!(error_type));
+                error.insert("code".to_owned(), serde_json::json!(code));
+                if let Some(param) = other.error_param() {
+                    error.insert("param".to_owned(), serde_json::json!(param));
+                }
+                serialize_to_vec_or_default(&serde_json::json!({ "error": error }))
             }
         }
     }
@@ -160,4 +217,47 @@ mod tests {
         assert!(exec_err.source().is_some(), "source should be chained");
         assert!(exec_err.to_string().contains("json error"));
     }
+
+    #[test]
+    fn conversation_locked_response_preserves_conflict_through_persistence() {
+        use std::error::Error;
+
+        let error = ExecutorError::Persistence(Box::new(ExecutorError::ConversationLocked {
+            source: StorageError::ConversationConflict {
+                conversation_id: "conv_internal".to_owned(),
+            },
+        }));
+
+        let conversation_locked = error.source().expect("persistence source must be retained");
+        let conflict = conversation_locked
+            .source()
+            .expect("conversation conflict source must be retained");
+        assert!(matches!(
+            conflict.downcast_ref::<StorageError>(),
+            Some(StorageError::ConversationConflict { conversation_id })
+                if conversation_id == "conv_internal"
+        ));
+
+        assert_eq!(error.http_status(), StatusCode::BAD_REQUEST);
+        assert_eq!(
+            serde_json::from_slice::<serde_json::Value>(&error.into_response_body())
+                .expect("valid error response JSON"),
+            serde_json::json!({
+                "error": {
+                    "message": "conversation changed while the response was being generated; retry the request",
+                    "type": "invalid_request_error",
+                    "code": "conversation_locked",
+                    "param": "conversation"
+                }
+            })
+        );
+    }
+
+    #[test]
+    fn non_conflict_response_omits_param() {
+        let body = ExecutorError::InvalidRequest("invalid input".to_owned()).into_response_body();
+        let value: serde_json::Value = serde_json::from_slice(&body).expect("valid error response JSON");
+
+        assert!(!value["error"].as_object().expect("error object").contains_key("param"));
+    }
 }

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

@@ -114,18 +114,20 @@ fn terminal_response_frame(payload: &ResponsePayload) -> ExecutorResult<EventFra
 }
 
 fn executor_error_frame(error: &ExecutorError) -> EventFrame {
+    let error_type = error.error_type();
     let code = error.error_code();
     let mut wire = WireEvent::new("error");
     wire.rest
         .insert("status".to_owned(), serde_json::json!(error.http_status().as_u16()));
-    wire.rest.insert(
-        "error".to_owned(),
-        serde_json::json!({
-            "message": error.to_string(),
-            "type": code,
-            "code": code,
-        }),
-    );
+    let mut error_details = serde_json::Map::new();
+    error_details.insert("message".to_owned(), serde_json::json!(error.error_message()));
+    error_details.insert("type".to_owned(), serde_json::json!(error_type));
+    error_details.insert("code".to_owned(), serde_json::json!(code));
+    if let Some(param) = error.error_param() {
+        error_details.insert("param".to_owned(), serde_json::json!(param));
+    }
+    wire.rest
+        .insert("error".to_owned(), serde_json::Value::Object(error_details));
     EventFrame {
         event_type: SSEEventType::Other,
         payload: EventPayload::None,
@@ -187,6 +189,7 @@ fn serialize_sse_frame(frame: &EventFrame) -> ExecutorResult<String> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::StorageError;
 
     #[test]
     fn process_sse_line_numbers_and_rebases_output_index() {
@@ -218,6 +221,34 @@ mod tests {
         assert_eq!(event["error"]["message"], "task failed: \"unexpected\"\nretry");
     }
 
+    #[test]
+    fn executor_conflict_sse_chunk_uses_client_conflict_contract() {
+        let mut accumulator = GatewayStreamAccumulator::new();
+        let error = ExecutorError::Persistence(Box::new(ExecutorError::ConversationLocked {
+            source: StorageError::ConversationConflict {
+                conversation_id: "conv_test".to_owned(),
+            },
+        }));
+        let chunk = accumulator.executor_error_chunk(&error);
+        let data = chunk
+            .trim_end_matches('\n')
+            .strip_prefix("data: ")
+            .expect("SSE data prefix");
+        let event: serde_json::Value = serde_json::from_str(data).expect("valid error event JSON");
+
+        assert_eq!(event["type"], "error");
+        assert_eq!(event["status"], 400);
+        assert_eq!(
+            event["error"],
+            serde_json::json!({
+                "message": "conversation changed while the response was being generated; retry the request",
+                "type": "invalid_request_error",
+                "code": "conversation_locked",
+                "param": "conversation"
+            })
+        );
+    }
+
     #[test]
     fn emits_in_progress_terminal_event_after_lifecycle_event() {
         let mut accumulator = GatewayStreamAccumulator::new();

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

@@ -23,7 +23,7 @@ pub use inference::call_inference;
 pub use messages_loop::run_messages_loop;
 pub use messages_stream::run_messages_stream;
 pub use modes::{ConversationHandler, ResponseHandler};
-pub use persist::persist_response;
+pub use persist::{persist_response, persist_turn};
 pub use rehydrate::rehydrate_conversation;
 pub use request::ExecutionContext;
 pub use request::RequestContext;