Reject unsupported message-thread continue requests before provider dispatch

src/server.rs

@@ -1535,6 +1535,46 @@ async fn dispatch_request(
         }
     };
 
+    if let Some(thread) = body.extra.remove("thread")
+        && thread.get("type").and_then(Value::as_str) != Some("create")
+    {
+        let message = "capability_rejected: beta_header:message-threads-2026-08-12";
+        let response = json_error(StatusCode::BAD_REQUEST, "invalid_request_error", message);
+        log_request_completed(
+            &log,
+            RequestLogContext {
+                req_id: &req_id,
+                provider: None,
+                model: None,
+                count_tokens,
+                status: response.status(),
+                started_at,
+            },
+        );
+        let (response, details) = record_failed_response(
+            &log,
+            FailedResponseLogContext {
+                req_id: &req_id,
+                provider: None,
+                model: None,
+                count_tokens,
+                started_at,
+            },
+            response,
+        )
+        .await;
+        monitor_failed(
+            state.monitor.as_ref(),
+            &req_id,
+            Some(response.status()),
+            details
+                .as_ref()
+                .map(|details| details.message.as_str())
+                .unwrap_or(message),
+        );
+        return with_request_id(response, &req_id);
+    }
+
     if let Some(project) = project::name_from_request(
         body.extra.get("system"),
         body.messages.iter().rev().map(|message| &message.content),

tests/server.rs

@@ -196,6 +196,42 @@ struct IdentityCaptureProvider {
     captured: Arc<Mutex<Vec<CapturedIdentity>>>,
 }
 
+struct RequestCaptureProvider {
+    captured: Arc<Mutex<Vec<MessagesRequest>>>,
+}
+
+#[async_trait]
+impl Provider for RequestCaptureProvider {
+    fn name(&self) -> &'static str {
+        "codex"
+    }
+
+    fn supported_models(&self) -> Vec<String> {
+        vec!["gpt-5.5".to_string()]
+    }
+
+    fn cli(&self) -> &'static dyn CliHandlers {
+        &FAKE_CLI
+    }
+
+    async fn handle_messages(
+        &self,
+        body: MessagesRequest,
+        _ctx: RequestContext,
+    ) -> axum::response::Response {
+        self.captured.lock().unwrap().push(body);
+        (StatusCode::OK, "captured").into_response()
+    }
+
+    async fn handle_count_tokens(
+        &self,
+        body: MessagesRequest,
+        ctx: RequestContext,
+    ) -> axum::response::Response {
+        self.handle_messages(body, ctx).await
+    }
+}
+
 #[async_trait]
 impl Provider for IdentityCaptureProvider {
     fn name(&self) -> &'static str {
@@ -473,6 +509,164 @@ async fn invalid_json_request_is_json_error() {
     assert_eq!(error_type, "invalid_request_error");
 }
 
+async fn message_thread_response(body: Value) -> (axum::response::Response, Vec<MessagesRequest>) {
+    let captured = Arc::new(Mutex::new(Vec::new()));
+    let registry = Arc::new(Registry::from_providers(
+        AliasProvider::Codex,
+        [Arc::new(RequestCaptureProvider {
+            captured: captured.clone(),
+        }) as Arc<dyn Provider>],
+    ));
+    let response = app(registry)
+        .oneshot(
+            Request::builder()
+                .method(Method::POST)
+                .uri("/v1/messages")
+                .header("content-type", "application/json")
+                .header("anthropic-beta", "message-threads-2026-08-12")
+                .body(Body::from(body.to_string()))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    let requests = captured.lock().unwrap().clone();
+    (response, requests)
+}
+
+async fn assert_message_thread_rejected(body: Value) {
+    let (response, captured) = message_thread_response(body).await;
+    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(captured.len(), 0);
+    let body: Value = serde_json::from_slice(
+        &axum::body::to_bytes(response.into_body(), usize::MAX)
+            .await
+            .unwrap(),
+    )
+    .unwrap();
+    assert_eq!(body["type"], "error");
+    assert_eq!(body["error"]["type"], "invalid_request_error");
+    assert!(
+        body["error"]["message"]
+            .as_str()
+            .unwrap()
+            .contains("capability_rejected: beta_header:message-threads-2026-08-12")
+    );
+}
+
+async fn assert_message_thread_dispatched(mut body: Value) {
+    let (response, captured) = message_thread_response(body.clone()).await;
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(captured.len(), 1);
+    assert!(!captured[0].extra.contains_key("thread"));
+    body.as_object_mut().unwrap().remove("thread");
+    let expected: MessagesRequest = serde_json::from_value(body).unwrap();
+    assert_eq!(
+        serde_json::to_value(&captured[0]).unwrap(),
+        serde_json::to_value(expected).unwrap()
+    );
+}
+
+#[tokio::test]
+async fn message_thread_continue_tool_result_is_rejected_before_dispatch() {
+    assert_message_thread_rejected(json!({
+        "model": "gpt-5.5",
+        "messages": [{"role": "user", "content": [{
+            "type": "tool_result", "tool_use_id": "call_test", "content": "done"
+        }]}],
+        "thread": {"type": "continue", "previous_message_id": "msg_test"}
+    }))
+    .await;
+}
+
+#[tokio::test]
+async fn message_thread_continue_text_is_rejected_before_dispatch() {
+    assert_message_thread_rejected(json!({
+        "model": "gpt-5.5",
+        "messages": [{"role": "user", "content": "next question"}],
+        "thread": {"type": "continue", "previous_message_id": "msg_test"}
+    }))
+    .await;
+}
+
+#[tokio::test]
+async fn message_thread_unknown_type_is_rejected_before_dispatch() {
+    assert_message_thread_rejected(json!({
+        "model": "gpt-5.5",
+        "messages": [{"role": "user", "content": "hello"}],
+        "thread": {"type": "fork"}
+    }))
+    .await;
+}
+
+#[tokio::test]
+async fn message_thread_missing_type_is_rejected_before_dispatch() {
+    assert_message_thread_rejected(json!({
+        "model": "gpt-5.5",
+        "messages": [{"role": "user", "content": "hello"}],
+        "thread": {}
+    }))
+    .await;
+}
+
+#[tokio::test]
+async fn message_thread_create_tool_result_dispatches_without_thread() {
+    assert_message_thread_dispatched(json!({
+        "model": "gpt-5.5",
+        "max_tokens": 32,
+        "system": "Keep the full history.",
+        "messages": [
+            {"role": "user", "content": "run the tool"},
+            {"role": "assistant", "content": [{
+                "type": "tool_use", "id": "call_test", "name": "test", "input": {}
+            }]},
+            {"role": "user", "content": [{
+                "type": "tool_result", "tool_use_id": "call_test", "content": "done"
+            }]}
+        ],
+        "thread": {"type": "create"}
+    }))
+    .await;
+}
+
+#[tokio::test]
+async fn message_thread_create_text_dispatches_without_thread() {
+    assert_message_thread_dispatched(json!({
+        "model": "gpt-5.5",
+        "max_tokens": 32,
+        "system": "Keep the full history.",
+        "messages": [
+            {"role": "user", "content": "hello"},
+            {"role": "assistant", "content": "hi"},
+            {"role": "user", "content": "next question"}
+        ],
+        "thread": {"type": "create"}
+    }))
+    .await;
+}
+
+#[tokio::test]
+async fn message_thread_absent_text_dispatches_unchanged() {
+    assert_message_thread_dispatched(json!({
+        "model": "gpt-5.5",
+        "max_tokens": 32,
+        "system": "Keep the full history.",
+        "messages": [{"role": "user", "content": "hello"}]
+    }))
+    .await;
+}
+
+#[tokio::test]
+async fn message_thread_absent_tool_result_dispatches_unchanged() {
+    assert_message_thread_dispatched(json!({
+        "model": "gpt-5.5",
+        "max_tokens": 32,
+        "messages": [{"role": "user", "content": [{
+            "type": "tool_result", "tool_use_id": "call_test", "content": "done"
+        }]}]
+    }))
+    .await;
+}
+
 #[tokio::test]
 async fn empty_body_is_invalid_json() {
     let app = app(Arc::new(Registry::with_default_alias()));