fix(plugin): prove replay safety for managed HTTP 429 rejections (#532)

Seal managed callback observations before Core retry decisions and synchronize RPC resource revocation with stream teardown.

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/event.rs

@@ -28,6 +28,7 @@ pub(super) struct EventDecoder {
     usage: Usage,
     contents: usize,
     pub(super) completed: bool,
+    pub(super) replay_blocked: bool,
 }
 
 pub(super) struct DecodedEvent {
@@ -50,6 +51,10 @@ impl EventDecoder {
         if self.completed || message.event.facts.len() > 64 {
             return Err(invalid(sent));
         }
+        // 即使与失败同批携带，也不能把 wire、工具或其他业务事实丢弃后重放
+        self.replay_blocked |= !message.event.facts.is_empty()
+            || message.event.wire.is_some()
+            || message.continuation.is_some();
         let wire = message
             .event
             .wire

backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs

@@ -75,12 +75,34 @@ pub(super) fn execute(
         let result = state.next().await;
         match result {
             Ok(event) => Ok(Some((event, state))),
-            Err(error) => Err(state.invocation.account.record_failure(error).await),
+            Err(error) => {
+                let error = state.finish_error(error);
+                Err(state.invocation.account.record_failure(error).await)
+            }
         }
     }))
 }
 
 impl Execution {
+    fn finish_error(&mut self, mut error: ProviderError) -> ProviderError {
+        if let Some(active) = self.active.take() {
+            let managed = Arc::clone(&active.managed);
+            // RpcStream 析构同步撤销回调并封存未完成发送，证明不能早于这一步
+            drop(active);
+            if error.kind() == ProviderErrorKind::RateLimited
+                && error.upstream_status() == Some(429)
+                && error.send_state() == UpstreamSendState::Sent
+                && managed.send_state.get() == UpstreamSendState::Sent
+                && managed.replay.rejected_once()
+                && !self.decoder.replay_blocked
+                && !self.invocation.context.cancellation().is_cancelled()
+            {
+                error = error.with_replay_safe();
+            }
+        }
+        error
+    }
+
     async fn next(&mut self) -> Result<gateway_core::event::ProviderEvent, ProviderError> {
         if self.active.is_none() {
             self.active = Some(self.begin().await?);

backend/crates/gateway-plugin/runtime/src/callback/dispatch.rs

@@ -94,6 +94,18 @@ impl CallbackHandler for PluginCallbacks {
                 if method == host::LOG {
                     return log.record(&context, params, &payload);
                 }
+                // 只证明受管 HTTP 拒绝；其他能力即使只读，也不推测其业务副作用
+                if let Some(managed) = &scope.upstream
+                    && !matches!(
+                        method.as_str(),
+                        upstream_wire::HTTP_DO
+                            | upstream_wire::HTTP_DO_STREAM
+                            | upstream_wire::HTTP_STREAM_READ
+                            | upstream_wire::HTTP_STREAM_CLOSE
+                    )
+                {
+                    managed.replay.invalidate();
+                }
                 let middleware = if matches!(
                     method.as_str(),
                     middleware_wire::NEXT_METHOD
@@ -225,7 +237,13 @@ impl CallbackHandler for PluginCallbacks {
                     | upstream_wire::WEBSOCKET_READ
                     | upstream_wire::WEBSOCKET_CLOSE
             ) {
-                return upstream::dispatch(&http, &method, params, payload).await;
+                return upstream::dispatch(&http, &method, params, payload)
+                    .await
+                    .inspect_err(|_| {
+                        if let Some(managed) = &scope.upstream {
+                            managed.replay.invalidate();
+                        }
+                    });
             }
             if matches!(
                 method.as_str(),

backend/crates/gateway-plugin/runtime/src/callback/http.rs

@@ -75,10 +75,11 @@ impl HttpCallbacks<'_> {
         let timeout = call.timeout()?;
         let attempt = call.start_upstream(upstream_purpose)?;
         let response = client.open(request, proxy, network, timeout).await;
-        attempt.finish(
+        attempt.finish_http(
             response
                 .as_ref()
                 .map_or_else(|error| error.send_state, |_| UpstreamSendState::Sent),
+            response.as_ref().ok().map(|response| response.status),
         );
         let response = response.map_err(http_error)?;
         let mut body = response.body;

backend/crates/gateway-plugin/runtime/src/callback/mod.rs

@@ -20,6 +20,7 @@ mod model;
 pub(crate) mod outbound;
 mod ports;
 pub(crate) mod private_state;
+mod replay;
 pub(crate) mod request_limits;
 mod resources;
 mod scope;

backend/crates/gateway-plugin/runtime/src/callback/replay.rs

@@ -0,0 +1,59 @@
+//! 记录受管单次 HTTP 拒绝，其他调用或不明结果只会收紧重放资格
+
+use std::sync::Mutex;
+
+#[derive(Default)]
+pub(crate) struct HttpReplay(Mutex<State>);
+
+#[derive(Default)]
+enum State {
+    #[default]
+    Untouched,
+    Pending,
+    Rejected,
+    Ineligible,
+}
+
+impl HttpReplay {
+    pub(super) fn start(&self, inference: bool) {
+        let mut state = self
+            .0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        *state = if inference && matches!(*state, State::Untouched) {
+            State::Pending
+        } else {
+            State::Ineligible
+        };
+    }
+
+    pub(super) fn finish(&self, status: Option<u16>) {
+        let mut state = self
+            .0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        *state = if status == Some(429) && matches!(*state, State::Pending) {
+            State::Rejected
+        } else {
+            State::Ineligible
+        };
+    }
+
+    pub(super) fn invalidate(&self) {
+        *self
+            .0
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner) = State::Ineligible;
+    }
+
+    /// 调用撤销后读取，防止检查通过后仍有新回调发送
+    pub(crate) fn rejected_once(&self) -> bool {
+        matches!(
+            *self
+                .0
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner),
+            State::Rejected
+        )
+    }
+}

backend/crates/gateway-plugin/runtime/src/callback/scope.rs

@@ -98,10 +98,13 @@ impl CallbackScope {
         &self,
         purpose: Option<gateway_plugin_sdk::call::upstream_adapter::UpstreamPathPurpose>,
     ) -> HttpAttempt {
+        let inference = purpose
+            == Some(gateway_plugin_sdk::call::upstream_adapter::UpstreamPathPurpose::Inference);
+        if let Some(managed) = &self.upstream {
+            managed.replay.start(inference);
+        }
         HttpAttempt(Arc::new(HttpAttemptObservation {
-            effects: if purpose
-                == Some(gateway_plugin_sdk::call::upstream_adapter::UpstreamPathPurpose::Inference)
-            {
+            effects: if inference {
                 None
             } else {
                 self.execution_effects.clone()
@@ -111,6 +114,10 @@ impl CallbackScope {
                 .as_ref()
                 .map(|managed| Arc::clone(&managed.send_state)),
             completed: Mutex::new(false),
+            replay: self
+                .upstream
+                .as_ref()
+                .map(|managed| Arc::clone(&managed.replay)),
         }))
     }
 }
@@ -121,6 +128,7 @@ pub(super) struct HttpAttemptObservation {
     effects: Option<Arc<ExecutionEffects>>,
     completed: Mutex<bool>,
     upstream: Option<Arc<super::upstream::SendWatermark>>,
+    replay: Option<Arc<super::replay::HttpReplay>>,
 }
 
 impl HttpAttempt {
@@ -131,10 +139,18 @@ impl HttpAttempt {
     pub(super) fn finish(self, observed: UpstreamSendState) {
         self.0.finish(observed);
     }
+
+    pub(super) fn finish_http(self, observed: UpstreamSendState, status: Option<u16>) {
+        self.0.finish_http(observed, status);
+    }
 }
 
 impl HttpAttemptObservation {
     pub(super) fn finish(&self, observed: UpstreamSendState) {
+        self.finish_http(observed, None);
+    }
+
+    fn finish_http(&self, observed: UpstreamSendState, status: Option<u16>) {
         let mut completed = self
             .completed
             .lock()
@@ -145,6 +161,9 @@ impl HttpAttemptObservation {
         if let Some(upstream) = &self.upstream {
             upstream.observe(observed);
         }
+        if let Some(replay) = &self.replay {
+            replay.finish(status.filter(|_| observed == UpstreamSendState::Sent));
+        }
         if observed != UpstreamSendState::NotSent
             && let Some(effects) = &self.effects
         {

backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs

@@ -55,6 +55,7 @@ impl SendWatermark {
 pub(crate) struct ManagedUpstream {
     pub(crate) account: Arc<dyn UpstreamAccountConnection>,
     pub(crate) send_state: Arc<SendWatermark>,
+    pub(crate) replay: Arc<super::replay::HttpReplay>,
     pub(crate) effects: Option<Arc<ExecutionEffects>>,
     target: Arc<UpstreamTarget>,
     socket: tokio::sync::Mutex<Option<(String, Arc<ManagedWebSocket>)>>,
@@ -85,6 +86,7 @@ impl ManagedUpstream {
             owner,
             websocket_allowed,
             send_state: Arc::default(),
+            replay: Arc::default(),
             socket: tokio::sync::Mutex::new(socket),
         }))
     }

backend/crates/gateway-plugin/runtime/src/rpc/calls.rs

@@ -44,6 +44,7 @@ struct PendingCall {
     execution: Option<OwnedSemaphorePermit>,
     deadline: CallDeadline,
     handler: Arc<dyn CallbackHandler>,
+    revoked: bool,
     transmitted: bool,
     cancellation: Option<RpcError>,
     context: CallContext,
@@ -84,6 +85,10 @@ impl Drop for CallbackTask {
 impl PendingCall {
     /// 关闭资源与封存副作用必须先于任何结果通知；取消确认只延续容量租约
     fn revoke(&mut self) {
+        if self.revoked {
+            return;
+        }
+        self.revoked = true;
         self.handler.finish(&self.context);
         self.callbacks.clear();
         self.stop_resumption();
@@ -201,6 +206,7 @@ impl Shared {
                 execution: Some(execution),
                 deadline: deadline.clone(),
                 handler,
+                revoked: false,
                 transmitted: false,
                 cancellation: None,
                 context,
@@ -256,6 +262,10 @@ impl Shared {
             let observer = (!self.expected_stop.load(Ordering::Acquire))
                 .then(|| self.failure_observer.clone())
                 .flatten();
+            // 消费者可能正在析构已有流；调用对查询者不可见前必须完成同步撤销
+            for call in state.calls.values_mut() {
+                call.revoke();
+            }
             (std::mem::take(&mut state.calls), observer)
         };
         self.execution.close();
@@ -298,7 +308,8 @@ impl Shared {
             drop(state);
             let _ = initial.send(result);
         } else {
-            let pending = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?;
+            let mut pending = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?;
+            pending.revoke();
             drop(state);
             pending.finish(result);
         }
@@ -329,7 +340,8 @@ impl Shared {
         {
             return Ok(());
         }
-        let call = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?;
+        let mut call = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?;
+        call.revoke();
         drop(state);
         if call.stream.is_none() || call.result.is_some() {
             call.finish(Err(RpcError::Protocol(None)));
@@ -416,6 +428,7 @@ impl Shared {
             );
         }
         if !call.transmitted {
+            call.revoke();
             let call = state.calls.remove(&id);
             drop(state);
             if let Some(call) = call {

backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs

@@ -848,6 +848,10 @@ async fn adapter_failure_preserves_upstream_diagnostics_and_native_feedback() {
     assert_eq!(error.kind(), ProviderErrorKind::RateLimited);
     assert_eq!(error.send_state(), UpstreamSendState::Sent);
     assert_eq!(error.upstream_status(), Some(429));
+    assert!(
+        error.replay_is_safe(),
+        "宿主观测的单次 HTTP429 应允许 Core 判断重试"
+    );
     assert_eq!(
         error.upstream_code().unwrap().as_str(),
         "rate_limit_exceeded"
@@ -876,6 +880,197 @@ async fn adapter_failure_preserves_upstream_diagnostics_and_native_feedback() {
     super::wait_until_empty(cache.path()).await;
 }
 
+#[tokio::test]
+async fn adapter_replay_requires_a_single_managed_http_rejection_before_output() {
+    for scenario in [
+        "json",
+        "stream",
+        "no_request",
+        "http_200",
+        "http_503",
+        "wrong_status",
+        "wrong_kind",
+        "sequential",
+        "concurrent",
+        "auxiliary",
+        "unmanaged",
+        "started",
+        "text",
+        "tool",
+        "failure_wire",
+    ] {
+        let server = MockServer::start().await;
+        let status = match scenario {
+            "http_200" => 200,
+            "http_503" => 503,
+            _ => 429,
+        };
+        let requests = match scenario {
+            "no_request" => 0,
+            "sequential" | "concurrent" => 2,
+            _ => 1,
+        };
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(status).set_body_string("{}"))
+            .expect(requests)
+            .mount(&server)
+            .await;
+        let mut config = configuration(&server.uri(), scenario == "stream");
+        let callback = config["upstream_callbacks"][0].clone();
+        match scenario {
+            "no_request" => config["upstream_callbacks"] = json!([]),
+            "sequential" => config["upstream_callbacks"] = json!([callback, callback]),
+            "concurrent" => config["upstream_callbacks"] = json!([[callback, callback]]),
+            "auxiliary" => {
+                config["upstream_registration"]["adapters"][0]["paths"][0]["purpose"] =
+                    json!("auxiliary");
+            }
+            "unmanaged" => {
+                config["upstream_callbacks"] = json!([{
+                    "method":"host.http.do", "params":{"method":"POST","url":server.uri()},
+                    "body":"request"
+                }]);
+            }
+            _ => {}
+        }
+        let mut events = Vec::new();
+        if matches!(scenario, "started" | "text" | "tool") {
+            events.push(json!({"event":{"facts":[{"type":"started","id":"response-one"}]}}));
+        }
+        if matches!(scenario, "text" | "tool") {
+            let kind = if scenario == "text" {
+                "text"
+            } else {
+                "tool_call"
+            };
+            events
+                .push(json!({"event":{"facts":[{"type":"content_added","index":0,"kind":kind}]}}));
+            let fact = if scenario == "text" {
+                json!({"type":"text_delta","index":0,"text":"partial"})
+            } else {
+                json!({"type":"tool_call_delta","index":0,"id":"call-one","name":"write_file","arguments":"{}"})
+            };
+            events.push(json!({"event":{"facts":[fact]}}));
+        }
+        let mut failure = json!({
+            "event":{"facts":[]},
+            "failure":{"kind":"rate_limited","status":429,"message":"fixture rejection"}
+        });
+        if scenario == "wrong_status" {
+            failure["failure"]["status"] = json!(503);
+        }
+        if scenario == "wrong_kind" {
+            failure["failure"]["kind"] = json!("unavailable");
+        }
+        if scenario == "failure_wire" {
+            failure["wire"] = json!("http");
+        }
+        events.push(failure);
+        config["upstream_events"] = json!(events);
+        let (cache, _, runtime) = setup(config).await;
+        let generation =
+            ExtensionPreparationPort::prepare(&runtime, ConfigRevision::new(1).unwrap())
+                .await
+                .unwrap();
+        let account = Account::new();
+        let mut stream = execute(&runtime, &generation, Arc::clone(&account), "key-one", None);
+        let error = loop {
+            match stream.next().await.expect("fixture must report a failure") {
+                Ok(_) => {}
+                Err(error) => break error,
+            }
+        };
+        assert_eq!(
+            error.kind(),
+            if scenario == "wrong_kind" {
+                ProviderErrorKind::Unavailable
+            } else {
+                ProviderErrorKind::RateLimited
+            },
+            "{scenario}"
+        );
+        assert_eq!(
+            error.replay_is_safe(),
+            matches!(scenario, "json" | "stream"),
+            "{scenario}"
+        );
+        assert_eq!(
+            error.send_state(),
+            if requests == 0 {
+                UpstreamSendState::NotSent
+            } else {
+                UpstreamSendState::Sent
+            },
+            "{scenario}"
+        );
+        drop(stream);
+        assert_eq!(Arc::strong_count(&account), 1, "{scenario}");
+        drop(generation);
+        runtime.shutdown().await;
+        super::wait_until_empty(cache.path()).await;
+    }
+}
+
+#[tokio::test]
+async fn adapter_rejection_with_pending_http_revokes_the_call_without_replay() {
+    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .respond_with(ResponseTemplate::new(429).set_body_string("{}"))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let directory = tempfile::tempdir().unwrap();
+    let marker = directory.path().join("pending-http-received");
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let pending_url = format!("http://{}/pending", listener.local_addr().unwrap());
+    let received = marker.clone();
+    let pending = tokio::spawn(async move {
+        let (mut socket, _) = listener.accept().await.unwrap();
+        let mut bytes = [0; 4096];
+        assert_ne!(socket.read(&mut bytes).await.unwrap(), 0);
+        std::fs::write(received, b"received").unwrap();
+        // 不返回响应头；父调用结束必须释放尚未确定结果的 HTTP 请求
+        while socket.read(&mut bytes).await.unwrap() != 0 {}
+        let _ = socket.shutdown().await;
+    });
+    let mut config = configuration(&server.uri(), false);
+    let first = config["upstream_callbacks"][0].clone();
+    config["upstream_callbacks"] = json!([first, {
+        "method":"host.upstream.http.do",
+        "params":{"method":"POST","path":pending_url,"headers":[]},
+        "pending_marker":marker
+    }]);
+    config["upstream_events"] = json!([{
+        "event":{"facts":[]},
+        "failure":{"kind":"rate_limited","status":429,"message":"fixture rejection"}
+    }]);
+    let (cache, _, runtime) = setup(config).await;
+    let generation = ExtensionPreparationPort::prepare(&runtime, ConfigRevision::new(1).unwrap())
+        .await
+        .unwrap();
+    let account = Account::new();
+    let mut stream = execute(&runtime, &generation, Arc::clone(&account), "key-one", None);
+    let error = tokio::time::timeout(Duration::from_secs(10), stream.next())
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap_err();
+    assert_eq!(error.kind(), ProviderErrorKind::RateLimited);
+    assert!(!error.replay_is_safe());
+    assert_ne!(error.send_state(), UpstreamSendState::NotSent);
+    tokio::time::timeout(Duration::from_secs(5), pending)
+        .await
+        .unwrap()
+        .unwrap();
+    drop(stream);
+    drop(generation);
+    runtime.shutdown().await;
+    assert_eq!(Arc::strong_count(&account), 1);
+    super::wait_until_empty(cache.path()).await;
+}
+
 #[tokio::test]
 async fn adapter_overrides_auth_headers_and_rejects_stale_continuation() {
     let server = MockServer::start().await;
@@ -1381,6 +1576,168 @@ async fn upstream_adapter_usage_uses_native_account_pricing() {
     runtime.shutdown().await;
 }
 
+struct RotatingAdapterProvider {
+    accounts: Vec<Arc<Account>>,
+    selected: std::sync::Mutex<Vec<ProviderAccountId>>,
+}
+
+#[async_trait::async_trait]
+impl gateway_core::engine::provider::Provider for RotatingAdapterProvider {
+    fn name(&self) -> &'static str {
+        "openai"
+    }
+
+    fn catalog_generation(&self) -> gateway_core::routing::ProviderCatalogGeneration {
+        Default::default()
+    }
+
+    async fn query_model_capabilities(
+        &self,
+    ) -> Result<Vec<gateway_core::routing::ProviderModelCapabilities>, ProviderError> {
+        Ok(vec![gateway_core::routing::ProviderModelCapabilities::new(
+            UpstreamModelId::new("native-model").unwrap(),
+            gateway_core::routing::ModelCapabilities::new(
+                std::collections::BTreeSet::from([
+                    gateway_core::operation::OperationKind::Generate,
+                ]),
+                None,
+            ),
+        )])
+    }
+
+    async fn execute(
+        self: Arc<Self>,
+        request: gateway_core::engine::provider::ProviderRequest,
+        context: AttemptContext,
+    ) -> Result<gateway_core::engine::provider::ProviderStream, ProviderError> {
+        let account = self
+            .accounts
+            .iter()
+            .find(|account| !context.excluded_accounts().contains(&account.id))
+            .cloned()
+            .ok_or_else(|| {
+                ProviderError::new(
+                    ProviderErrorKind::NoEligibleAccount,
+                    UpstreamSendState::NotSent,
+                )
+            })?;
+        self.selected.lock().unwrap().push(account.id.clone());
+        let candidate = request.candidate();
+        let model = candidate.upstream_model().unwrap();
+        let adapter = context
+            .upstream_adapter_for_account(candidate.provider(), model, &account.id)?
+            .unwrap();
+        let metadata = ProviderCallMetadata::new(
+            candidate.provider().clone(),
+            model.clone(),
+            account.id.clone(),
+            UpstreamTransport::new(adapter.transport()).unwrap(),
+        );
+        let events = adapter.execute(UpstreamAdapterInvocation {
+            operation: request.operation().clone(),
+            headers: vec![],
+            context,
+            metadata: metadata.clone(),
+            account,
+        });
+        Ok(gateway_core::engine::provider::ProviderStream::new(
+            metadata,
+            events,
+            (),
+        ))
+    }
+}
+
+#[tokio::test]
+async fn core_rotates_after_a_managed_adapter_http_rejection() {
+    use gateway_core::engine::{
+        execution::{ClientTransport, ExecutionRequestMetadata, StartExecution},
+        provider::{Provider, ProviderRegistry},
+    };
+    let Some(environment) = crate::support::environment::Environment::create().await else {
+        eprintln!("SKIP: plugin integration environment absent");
+        return;
+    };
+    let server = MockServer::start().await;
+    let calls = Arc::new(AtomicUsize::new(0));
+    let observed = Arc::clone(&calls);
+    Mock::given(method("POST"))
+        .respond_with(move |_: &wiremock::Request| {
+            let status = if observed.fetch_add(1, Ordering::SeqCst) == 0 {
+                429
+            } else {
+                200
+            };
+            ResponseTemplate::new(status).set_body_string("{}")
+        })
+        .expect(2)
+        .mount(&server)
+        .await;
+    let first = environment.account(None).await;
+    let second = environment.account(None).await;
+    let provider = Arc::new(RotatingAdapterProvider {
+        accounts: vec![
+            Account::with_identity(first.as_str(), 1),
+            Account::with_identity(second.as_str(), 1),
+        ],
+        selected: Default::default(),
+    });
+    let mut config = configuration(&server.uri(), false);
+    config["upstream_registration"]["adapters"][0]["account_ids"] =
+        json!([first.as_str(), second.as_str()]);
+    config["upstream_account_events"] = json!({first.as_str(): [{
+        "event":{"facts":[]},
+        "failure":{"kind":"rate_limited","status":429,"message":"fixture rejection"}
+    }]});
+    environment.install_plugin(config).await;
+    let key = "sk-adapter-rotation-fixture";
+    environment.client_key("key_adapter_rotation", key).await;
+    let (runtime, core) = environment
+        .runtime_with_providers(
+            Default::default(),
+            ProviderRegistry::new([provider.clone() as Arc<dyn Provider>]).unwrap(),
+        )
+        .await;
+    let execution = core.execution_service();
+    let operation = Operation::Generate(GenerateRequest::from_protocol_payload(
+        ProtocolPayload::json_object(
+            "openai",
+            json!({"model":"native-model","input":"hello"})
+                .as_object()
+                .unwrap()
+                .clone(),
+        )
+        .unwrap(),
+    ));
+    let mut started = execution
+        .start(StartExecution {
+            client: execution.authenticate(key).unwrap(),
+            public_model: gateway_core::routing::PublicModelId::new("native-model").unwrap(),
+            operation,
+            metadata: ExecutionRequestMetadata {
+                protocol: "openai".into(),
+                endpoint: "/v1/responses".into(),
+                transport: ClientTransport::HttpJson,
+                stream: false,
+                client_ip: None,
+                user_agent: None,
+                previous_response_id: None,
+            },
+        })
+        .await
+        .unwrap();
+    started.session.collect_uncommitted().await.unwrap();
+    started.session.commit_downstream(Some(200)).await.unwrap();
+    started.session.detach_finalize().await;
+    assert_eq!(*provider.selected.lock().unwrap(), vec![first, second]);
+    assert_eq!(calls.load(Ordering::SeqCst), 2);
+    environment.release_plugin_accounts(&runtime);
+    drop(execution);
+    drop(core);
+    runtime.shutdown().await;
+    environment.close().await;
+}
+
 #[tokio::test]
 async fn core_published_adapter_uses_normal_attempt_usage_and_cost_ledger() {
     use crate::support::environment::Environment;

backend/crates/gateway-plugin/runtime/tests/rpc/calls.rs

@@ -1,7 +1,10 @@
 //! 验证调用表在成功、远端错误和超时的结果交付前同步撤销回调资源
 
 use std::{
-    sync::{Arc, atomic::Ordering},
+    sync::{
+        Arc, Condvar, Mutex,
+        atomic::{AtomicBool, Ordering},
+    },
     time::Duration,
 };
 
@@ -11,6 +14,116 @@ use serde_json::json;
 
 use super::session::{LifecycleCallbacks, prepared_session};
 
+struct GatedFinish {
+    ready: Arc<tokio::sync::Notify>,
+    entered: std::sync::mpsc::Sender<()>,
+    started: AtomicBool,
+    finished: AtomicBool,
+    released: Mutex<bool>,
+    release: Condvar,
+}
+
+impl gateway_plugin_runtime::CallbackHandler for GatedFinish {
+    fn finish(&self, _: &gateway_plugin_sdk::CallContext) {
+        if self.started.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        self.entered.send(()).unwrap();
+        // 测试屏障模拟撤销处理中被调度器暂停，不依赖恰好命中的线程交错
+        let released = self.released.lock().unwrap();
+        let (released, timeout) = self
+            .release
+            .wait_timeout_while(released, Duration::from_secs(5), |released| !*released)
+            .unwrap();
+        assert!(!timeout.timed_out() && *released);
+        self.finished.store(true, Ordering::Release);
+    }
+
+    fn call(
+        &self,
+        _: gateway_plugin_sdk::CallContext,
+        method: String,
+        _: serde_json::Value,
+        _: Vec<u8>,
+    ) -> futures::future::BoxFuture<
+        'static,
+        Result<gateway_plugin_runtime::RpcReply, gateway_plugin_sdk::PluginFault>,
+    > {
+        let ready = Arc::clone(&self.ready);
+        Box::pin(async move {
+            assert_eq!(method, "test.stream_ready");
+            ready.notified().await;
+            Ok(gateway_plugin_runtime::RpcReply {
+                result: json!({}),
+                payload: vec![],
+            })
+        })
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn dropping_stream_waits_for_resource_revocation_even_after_end_or_process_failure() {
+    for method in ["stream_single", "stream_single_crash"] {
+        let (_cache, prepared, handshake, processes) = prepared_session();
+        let (entered, entry) = std::sync::mpsc::channel();
+        let callbacks = Arc::new(GatedFinish {
+            ready: Arc::new(tokio::sync::Notify::new()),
+            entered,
+            started: AtomicBool::new(false),
+            finished: AtomicBool::new(false),
+            released: Mutex::new(false),
+            release: Condvar::new(),
+        });
+        let session = Arc::new(
+            RpcSession::start(
+                prepared,
+                handshake,
+                RpcLimits::default(),
+                &processes,
+                callbacks.clone(),
+            )
+            .await
+            .unwrap(),
+        );
+        let stream = session
+            .call_stream(
+                method,
+                session.context(Stage::Request, Duration::from_secs(5)),
+                json!({}),
+                vec![],
+            )
+            .await
+            .unwrap();
+        // 拿到流后才允许对端结束；整个屏障控制不依赖被暂停的 I/O 调度器
+        callbacks.ready.notify_one();
+        entry.recv_timeout(Duration::from_secs(5)).unwrap();
+        let observed = Arc::clone(&callbacks);
+        let (started, received) = std::sync::mpsc::channel();
+        let (completed, completion) = std::sync::mpsc::channel();
+        let dropped = std::thread::spawn(move || {
+            started.send(()).unwrap();
+            drop(stream);
+            completed
+                .send(observed.finished.load(Ordering::Acquire))
+                .unwrap();
+        });
+        received.recv_timeout(Duration::from_secs(1)).unwrap();
+        // 撤销屏障会暂停 I/O 调度线程，不能用同一 runtime 的计时器释放它
+        let early = completion.recv_timeout(Duration::from_millis(100));
+        *callbacks.released.lock().unwrap() = true;
+        callbacks.release.notify_one();
+        let was_early = early.is_ok();
+        let revoked =
+            early.unwrap_or_else(|_| completion.recv_timeout(Duration::from_secs(1)).unwrap());
+        dropped.join().unwrap();
+        session.shutdown(Duration::from_secs(1)).await;
+        assert!(
+            !was_early && revoked,
+            "{method}: stream drop must not outrun callback revocation"
+        );
+    }
+}
+
 #[tokio::test]
 async fn terminal_results_revoke_callback_resources_before_waking_callers() {
     for method in ["echo", "deny", "hang"] {

backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs

@@ -489,7 +489,15 @@ impl Peer {
             )
             .await;
             let credits = self.streams.lock().await.get(&id).unwrap().clone();
-            for sequence in 0..128_u64 {
+            let chunks = if method.starts_with("stream_single") {
+                self.callback(id, "test.stream_ready", json!({}))
+                    .await
+                    .unwrap();
+                1
+            } else {
+                128
+            };
+            for sequence in 0..chunks {
                 credits.take(8192).await;
                 let payload = if method == "stream_overflow" {
                     vec![b'x'; 256 * 1024 + 1]
@@ -499,6 +507,9 @@ impl Peer {
                 let sequence = sequence + u64::from(method == "stream_sequence");
                 self.send(Message::Stream { id, sequence }, payload).await;
             }
+            if method == "stream_single_crash" {
+                std::process::exit(7);
+            }
             self.send(Message::End { id, error: None }, vec![]).await;
             return;
         }
@@ -1737,6 +1748,27 @@ impl Peer {
             .into_iter()
             .flatten()
         {
+            if let Some(marker) = callback["pending_marker"].as_str() {
+                // 等真实服务器确认收到字节，再报告失败，避免依赖回调调度顺序
+                let _pending = self
+                    .send_callback(
+                        id,
+                        callback["method"].as_str().unwrap(),
+                        callback["params"].clone(),
+                        body.clone(),
+                    )
+                    .await;
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while !std::path::Path::new(marker).exists() {
+                        tokio::time::sleep(Duration::from_millis(5)).await;
+                    }
+                })
+                .await
+                .map_err(|_| {
+                    PluginFault::new(ErrorCode::Timeout, "pending HTTP fixture did not start")
+                })?;
+                continue;
+            }
             let callbacks = callback
                 .as_array()
                 .map_or_else(|| std::slice::from_ref(callback), Vec::as_slice);
@@ -1787,7 +1819,11 @@ impl Peer {
                 }
             }
         }
-        let events = self.configuration["upstream_events"].as_array().unwrap();
+        let events = self.configuration["upstream_account_events"]
+            .get(&request.account_id)
+            .unwrap_or(&self.configuration["upstream_events"])
+            .as_array()
+            .unwrap();
         let credits = self.streams.lock().await.get(&id).unwrap().clone();
         for (sequence, event) in events.iter().enumerate() {
             let mut message = UpstreamAdapterEvent::new(

backend/crates/gateway-plugin/sdk/docs/upstream-adapters.md

@@ -94,6 +94,13 @@ let (message, ()) = tokio::try_join!(
 包含标准事实与可选原始 wire；事件需符合开始、内容、工具、用量和唯一结束的顺序。
 原始 JSON / SSE / WebSocket 字节与标准事实分开传递，错误使用 `UpstreamFailure`；宿主网络观测决定发送状态，插件不能将其降为未发送
 
+宿主只对一次受管 `inference` HTTP 调用明确返回的 429 提供有限重放证明：插件必须报告
+`rate_limited` 和 `status: 429`，且此前及失败事件中均无业务事实、wire 或续接材料。
+调用结束时先撤销回调、回收资源并封存未完成发送，再将错误交给 Core；发送状态仍为 `sent`。
+多次或并发请求、辅助请求、其他宿主能力调用（日志除外）、受管回调失败以及取消都不能取得该证明。
+观测事件和该次 HTTP 的读流、关流不单独阻断；HTTP200 流内报错、WebSocket 拒绝及其他状态码不适用。
+Core 仍复核续接归属、外部副作用、交付与重试预算；这不保证一定换号，也不证明完整信任插件绕过宿主执行的操作可重放
+
 初始 RPC 回复受插件处理预算约束，默认上限为 120 秒；持续接收与解析应放在返回的流中。
 初始流式回复后，流的寿命遵循模型请求期限与取消，受管 HTTP 正文每次读取单独计时。
 持续有数据的流不会因初始 RPC 预算耗尽而结束，单次读取超时、客户端取消和流结束仍会回收资源