[WS2][WS3] 接通私有登录取消和独立 token 计数

crates/cranibox-llm/src/auth/signin.rs

@@ -171,7 +171,7 @@ impl OAuth {
             flows: Default::default(),
         }
     }
-    /// In-process cancellation. The C3 management contract has no cancel operation yet.
+    /// Cancellation shared by in-process callers and the private management API.
     /// An exchange already in progress is allowed to persist its rotating credentials first.
     pub async fn cancel(&self, flow_id: &str) -> Result<(), MgmtError> {
         let flow = self
@@ -622,6 +622,9 @@ impl OAuth {
 }
 #[async_trait::async_trait]
 impl SignIn for OAuth {
+    async fn cancel(&self, flow_id: &str) -> Result<(), MgmtError> {
+        OAuth::cancel(self, flow_id).await
+    }
     async fn start(&self, provider: &str, opts: SignInOpts) -> Result<SignInStart, MgmtError> {
         if opts
             .region

crates/cranibox-llm/src/mgmt.rs

@@ -238,6 +238,14 @@ impl LlmMgmt for LocalMgmt {
             )
             .await
     }
+    async fn signin_cancel(&self, flow: &str) -> Result<(), MgmtError> {
+        self.engine
+            .sign_in
+            .as_ref()
+            .ok_or(MgmtError::NotFound)?
+            .cancel(flow)
+            .await
+    }
     async fn signin_complete(
         &self,
         flow: &str,
@@ -334,6 +342,12 @@ async fn signin_complete(
 ) -> Response {
     answer(m.signin_complete(&flow, v).await)
 }
+async fn signin_cancel(State(m): State<LocalMgmt>, Path(flow): Path<String>) -> Response {
+    match m.signin_cancel(&flow).await {
+        Ok(()) => StatusCode::NO_CONTENT.into_response(),
+        Err(e) => failure(e),
+    }
+}
 async fn models(State(m): State<LocalMgmt>) -> Response {
     answer(m.models().await)
 }
@@ -362,6 +376,7 @@ fn router(mgmt: LocalMgmt) -> Router {
         .route(ACCOUNT_ENABLE, post(enable))
         .route(SIGNIN_START, post(signin_start))
         .route(SIGNIN_COMPLETE, post(signin_complete))
+        .route(SIGNIN_CANCEL, post(signin_cancel))
         .route(MODELS, get(models))
         .route(CLAUDE_SET_NODE, post(set_node))
         .route(PRESETS, get(presets))
@@ -495,6 +510,9 @@ impl Client {
         if status == StatusCode::NOT_FOUND {
             return Err(MgmtError::NotFound);
         }
+        if status == StatusCode::NOT_IMPLEMENTED {
+            return Err(UpstreamError::Unsupported.into());
+        }
         if !status.is_success() {
             if let Ok(value) = serde_json::from_slice::<UnavailableBody>(&body) {
                 return Err(MgmtError::Unavailable(value.error.code));
@@ -532,6 +550,11 @@ fn account_path(id: &str, suffix: &str) -> Result<String, MgmtError> {
 }
 #[async_trait::async_trait]
 impl LlmMgmt for Client {
+    async fn signin_cancel(&self, flow: &str) -> Result<(), MgmtError> {
+        account_path(flow, "")?;
+        self.call(http::Method::POST, &format!("/signin/{flow}/cancel"), None)
+            .await
+    }
     async fn status(&self) -> Result<Status, MgmtError> {
         self.call(http::Method::GET, STATUS, None).await
     }

crates/cranibox-llm/src/route/mod.rs

@@ -19,4 +19,13 @@ pub trait Router: Send + Sync {
         provider: &str,
         request: CanonicalRequest,
     ) -> Result<UpstreamResponse, UpstreamError>;
+    /// Forward the count operation to the peer count endpoint, never to generation.
+    async fn forward_count_tokens(
+        &self,
+        _node: &str,
+        _provider: &str,
+        _request: CanonicalRequest,
+    ) -> Result<UpstreamResponse, UpstreamError> {
+        Err(UpstreamError::Unsupported)
+    }
 }

crates/cranibox-llm/src/server.rs

@@ -38,6 +38,14 @@ impl Handle {
     ) -> Result<UpstreamResponse, crate::ProxyError> {
         self.engine.send_local(provider, request, false).await
     }
+    /// Authorized mesh ingress for token counts; never re-enters remote routing.
+    pub async fn dispatch_count_local(
+        &self,
+        provider: &str,
+        request: CanonicalRequest,
+    ) -> Result<UpstreamResponse, crate::ProxyError> {
+        self.engine.send_local(provider, request, true).await
+    }
     pub async fn shutdown(mut self) {
         self.cancel.cancel();
         for mut task in self.tasks.drain(..) {

crates/cranibox-llm/src/server/engine.rs

@@ -2,7 +2,7 @@
 use crate::{
     Config, catalog, proto,
     route::local::{Pool, retryable},
-    upstream::apikey::{ApiKey, Credential},
+    upstream::apikey::Credential,
 };
 use cranibox_contracts::{
     error::ErrorCode, llm::mgmt::MgmtEvent, llm::*, net::BoxBody, vault::AccountStore,
@@ -309,34 +309,51 @@ impl Engine {
     pub async fn send(
         self: &Arc<Self>,
         provider: &str,
-        request: CanonicalRequest,
+        mut request: CanonicalRequest,
         count: bool,
     ) -> Result<UpstreamResponse, CallError> {
+        if count {
+            Self::normalize_count(&mut request)?;
+        }
         if let Some(router) = &self.routing {
             match router.decide(provider, &request).await {
                 crate::route::RouteDecision::Local => {}
                 crate::route::RouteDecision::Unavailable { code } => {
                     return Err(CallError::Unavailable(code));
                 }
                 crate::route::RouteDecision::Remote { node } => {
-                    if count {
-                        return Err(CallError::Upstream(UpstreamError::Unsupported));
+                    return if count {
+                        router.forward_count_tokens(&node, provider, request).await
+                    } else {
+                        router.forward(&node, provider, request).await
                     }
-                    return router
-                        .forward(&node, provider, request)
-                        .await
-                        .map_err(CallError::Upstream);
+                    .map_err(CallError::Upstream);
                 }
             }
         }
         self.send_local(provider, request, count).await
     }
+    fn normalize_count(request: &mut CanonicalRequest) -> Result<(), CallError> {
+        if request.wire != Wire::Messages {
+            return Err(CallError::Upstream(UpstreamError::Unsupported));
+        }
+        request.stream = false;
+        request
+            .body
+            .as_object_mut()
+            .ok_or(CallError::Upstream(UpstreamError::Unsupported))?
+            .remove("stream");
+        Ok(())
+    }
     pub async fn send_local(
         self: &Arc<Self>,
         provider: &str,
-        request: CanonicalRequest,
+        mut request: CanonicalRequest,
         count: bool,
     ) -> Result<UpstreamResponse, CallError> {
+        if count {
+            Self::normalize_count(&mut request)?;
+        }
         let session = request.session_id.clone().unwrap_or_else(|| {
             let user = request
                 .body
@@ -373,7 +390,9 @@ impl Engine {
                 let mut cx = self.context(&account.id, None).await?;
                 let endpoints = Credential::decode(&cx.secret)
                     .map(|c| crate::upstream::apikey::endpoints(&cx, &c))?;
-                let target = if account.kind == AccountKind::ApiKey {
+                let target = if count {
+                    Wire::Messages
+                } else if account.kind == AccountKind::ApiKey {
                     if endpoints.contains_key(&request.wire) {
                         request.wire
                     } else {
@@ -388,12 +407,6 @@ impl Engine {
                 } else {
                     *adapter.wires().first().ok_or(UpstreamError::Unsupported)?
                 };
-                if count
-                    && (account.kind != AccountKind::ApiKey
-                        || !endpoints.contains_key(&Wire::Messages))
-                {
-                    return Err(UpstreamError::Unsupported);
-                }
                 let mut outbound = request.clone();
                 outbound.wire = if count { Wire::Messages } else { target };
                 outbound.body = if count {
@@ -410,7 +423,7 @@ impl Engine {
                     node: cx.node.clone(),
                 };
                 let mut response = if count {
-                    ApiKey.send_count(&initial_cx, outbound.clone()).await
+                    adapter.count_tokens(&initial_cx, outbound.clone()).await
                 } else {
                     adapter.send(&initial_cx, outbound.clone()).await
                 };
@@ -419,7 +432,7 @@ impl Engine {
                 {
                     cx = self.context(&account.id, Some(cx.secret.clone())).await?;
                     response = if count {
-                        ApiKey.send_count(&cx, outbound).await
+                        adapter.count_tokens(&cx, outbound).await
                     } else {
                         adapter.send(&cx, outbound).await
                     };

crates/cranibox-llm/src/upstream/apikey.rs

@@ -313,6 +313,13 @@ impl ApiKey {
 }
 #[async_trait::async_trait]
 impl Upstream for ApiKey {
+    async fn count_tokens(
+        &self,
+        cx: &UpstreamCx,
+        req: CanonicalRequest,
+    ) -> Result<UpstreamResponse, UpstreamError> {
+        self.send_count(cx, req).await
+    }
     fn provider(&self) -> &'static str {
         "apikey"
     }