From 982f4dff71ffe6256219611b5b9b643ff88c94af Mon Sep 17 00:00:00 2001 From: fliaping <269832017@qq.com> Date: Mon, 19 Oct 2026 22:07:14 +0800 Subject: [PATCH 1/3] [WS2][WS3] add private sign-in cancellation and explicit token counting --- crates/cranibox-llm/src/auth/signin.rs | 5 +- crates/cranibox-llm/src/mgmt.rs | 23 ++ crates/cranibox-llm/src/route/mod.rs | 9 + crates/cranibox-llm/src/server.rs | 8 + crates/cranibox-llm/src/server/engine.rs | 49 ++-- crates/cranibox-llm/src/upstream/apikey.rs | 7 + crates/cranibox-llm/tests/count_tokens.rs | 232 ++++++++++++++++++ .../cranibox-llm/tests/providers/lifecycle.rs | 156 ++++++++++++ 8 files changed, 470 insertions(+), 19 deletions(-) create mode 100644 crates/cranibox-llm/tests/count_tokens.rs diff --git a/crates/cranibox-llm/src/auth/signin.rs b/crates/cranibox-llm/src/auth/signin.rs index 19d0301..9bf8384 100644 --- a/crates/cranibox-llm/src/auth/signin.rs +++ b/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 { if opts .region diff --git a/crates/cranibox-llm/src/mgmt.rs b/crates/cranibox-llm/src/mgmt.rs index 24b2c64..e97098c 100644 --- a/crates/cranibox-llm/src/mgmt.rs +++ b/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, Path(flow): Path) -> Response { + match m.signin_cancel(&flow).await { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(e) => failure(e), + } +} async fn models(State(m): State) -> 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::(&body) { return Err(MgmtError::Unavailable(value.error.code)); @@ -532,6 +550,11 @@ fn account_path(id: &str, suffix: &str) -> Result { } #[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 { self.call(http::Method::GET, STATUS, None).await } diff --git a/crates/cranibox-llm/src/route/mod.rs b/crates/cranibox-llm/src/route/mod.rs index b7bf91a..f15468c 100644 --- a/crates/cranibox-llm/src/route/mod.rs +++ b/crates/cranibox-llm/src/route/mod.rs @@ -19,4 +19,13 @@ pub trait Router: Send + Sync { provider: &str, request: CanonicalRequest, ) -> Result; + /// 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 { + Err(UpstreamError::Unsupported) + } } diff --git a/crates/cranibox-llm/src/server.rs b/crates/cranibox-llm/src/server.rs index 38d031a..4aa8db2 100644 --- a/crates/cranibox-llm/src/server.rs +++ b/crates/cranibox-llm/src/server.rs @@ -38,6 +38,14 @@ impl Handle { ) -> Result { 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 { + self.engine.send_local(provider, request, true).await + } pub async fn shutdown(mut self) { self.cancel.cancel(); for mut task in self.tasks.drain(..) { diff --git a/crates/cranibox-llm/src/server/engine.rs b/crates/cranibox-llm/src/server/engine.rs index 96817ff..ed8f519 100644 --- a/crates/cranibox-llm/src/server/engine.rs +++ b/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,9 +309,12 @@ impl Engine { pub async fn send( self: &Arc, provider: &str, - request: CanonicalRequest, + mut request: CanonicalRequest, count: bool, ) -> Result { + if count { + Self::normalize_count(&mut request)?; + } if let Some(router) = &self.routing { match router.decide(provider, &request).await { crate::route::RouteDecision::Local => {} @@ -319,24 +322,38 @@ impl Engine { 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, provider: &str, - request: CanonicalRequest, + mut request: CanonicalRequest, count: bool, ) -> Result { + 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 }; diff --git a/crates/cranibox-llm/src/upstream/apikey.rs b/crates/cranibox-llm/src/upstream/apikey.rs index 86f09ab..413ae69 100644 --- a/crates/cranibox-llm/src/upstream/apikey.rs +++ b/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 { + self.send_count(cx, req).await + } fn provider(&self) -> &'static str { "apikey" } diff --git a/crates/cranibox-llm/tests/count_tokens.rs b/crates/cranibox-llm/tests/count_tokens.rs new file mode 100644 index 0000000..c0120a1 --- /dev/null +++ b/crates/cranibox-llm/tests/count_tokens.rs @@ -0,0 +1,232 @@ +use cranibox_contracts::{llm::*, mock::MemoryAccountStore, net::BoxBody, vault::AccountStore}; +use cranibox_llm::{ + Config, Handle, + route::{RouteDecision, Router}, + upstream::apikey::Credential, +}; +use serde_json::{Value, json}; +use std::sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, +}; + +#[derive(Default)] +struct Counter { + calls: Mutex>, + refreshes: AtomicUsize, +} +#[async_trait::async_trait] +impl Upstream for Counter { + fn provider(&self) -> &'static str { + "counter" + } + fn wires(&self) -> &[Wire] { + &[Wire::Messages] + } + async fn send( + &self, + _: &UpstreamCx, + _: CanonicalRequest, + ) -> Result { + panic!("counting must never generate a completion") + } + async fn count_tokens( + &self, + cx: &UpstreamCx, + request: CanonicalRequest, + ) -> Result { + assert_eq!(request.wire, Wire::Messages); + assert!(!request.stream); + assert!(request.body.get("stream").is_none()); + assert_eq!(request.model, "model"); + let token = Credential::decode(&cx.secret)?.access_token; + self.calls.lock().unwrap().push(token.clone()); + if token == "old" { + return Err(UpstreamError::Auth); + } + Ok(UpstreamResponse { + status: 200, + headers: vec![], + wire: Wire::Messages, + body: BoxBody::full(json!({"input_tokens":7}).to_string()), + }) + } + async fn refresh(&self, _: &UpstreamCx, _: &SecretBox) -> Result { + self.refreshes.fetch_add(1, Ordering::SeqCst); + Ok(Refreshed { + secret: Credential { + access_token: "new".into(), + ..Default::default() + } + .secret(), + expires_at: None, + }) + } + async fn models(&self, _: &UpstreamCx) -> Result, UpstreamError> { + Ok(vec![]) + } +} +struct Legacy; +#[async_trait::async_trait] +impl Upstream for Legacy { + fn provider(&self) -> &'static str { + "counter" + } + fn wires(&self) -> &[Wire] { + &[Wire::Messages] + } + async fn send( + &self, + _: &UpstreamCx, + _: CanonicalRequest, + ) -> Result { + panic!("legacy adapter must not generate to count") + } + async fn models(&self, _: &UpstreamCx) -> Result, UpstreamError> { + Ok(vec![]) + } +} +fn store(token: &str) -> Arc { + let store = Arc::new(MemoryAccountStore::default()); + let account = cranibox_llm::auth::signin::summary("counter", "local", None); + store + .put( + account, + Some( + Credential { + access_token: token.into(), + ..Default::default() + } + .secret(), + ), + ) + .unwrap(); + store +} +async fn start( + store: Arc, + upstream: Arc, + router: Option>, +) -> (tempfile::TempDir, Handle) { + let dir = tempfile::tempdir().unwrap(); + let mut cfg = Config::new("local".into(), dir.path().join("run"), store); + cfg.listen = vec!["127.0.0.1:0".parse().unwrap()]; + cfg.upstreams = vec![upstream]; + cfg.routing = router; + let handle = cranibox_llm::spawn(cfg).await.unwrap(); + (dir, handle) +} +async fn count(handle: &Handle) -> reqwest::Response { + reqwest::Client::new().post(format!("http://{}/v1/messages/count_tokens", handle.addresses[0])) + .json(&json!({"model":"counter/model","stream":true,"messages":[{"role":"user","content":"hello"}]})).send().await.unwrap() +} +#[tokio::test] +async fn oauth_count_uses_count_operation_and_refreshes_after_auth_failure() { + let store = store("old"); + let adapter = Arc::new(Counter::default()); + let (_dir, handle) = start(store.clone(), adapter.clone(), None).await; + let response = count(&handle).await; + assert_eq!(response.status(), 200); + assert_eq!( + response.json::().await.unwrap(), + json!({"input_tokens":7}) + ); + assert_eq!(*adapter.calls.lock().unwrap(), ["old", "new"]); + assert_eq!(adapter.refreshes.load(Ordering::SeqCst), 1); + let id = &store.list()[0].id; + assert_eq!( + Credential::decode(&store.get_secret(id).unwrap()) + .unwrap() + .access_token, + "new" + ); + handle.shutdown().await; +} +#[tokio::test] +async fn unsupported_oauth_count_returns_501_without_generating() { + let (_dir, handle) = start(store("new"), Arc::new(Legacy), None).await; + assert_eq!(count(&handle).await.status(), 501); + handle.shutdown().await; +} +struct LegacyRouter; +#[async_trait::async_trait] +impl Router for LegacyRouter { + async fn decide(&self, _: &str, _: &CanonicalRequest) -> RouteDecision { + RouteDecision::Remote { + node: "peer".into(), + } + } + async fn forward( + &self, + _: &str, + _: &str, + _: CanonicalRequest, + ) -> Result { + panic!("count must not use generation forwarding") + } +} +#[tokio::test] +async fn legacy_router_does_not_forward_counts_to_generation() { + let (_dir, handle) = start(store("new"), Arc::new(Legacy), Some(Arc::new(LegacyRouter))).await; + assert_eq!(count(&handle).await.status(), 501); + handle.shutdown().await; +} +struct Bridge { + peer: Arc, + calls: AtomicUsize, +} +#[async_trait::async_trait] +impl Router for Bridge { + async fn decide(&self, _: &str, _: &CanonicalRequest) -> RouteDecision { + RouteDecision::Remote { + node: "peer".into(), + } + } + async fn forward( + &self, + _: &str, + _: &str, + _: CanonicalRequest, + ) -> Result { + panic!("generation forwarding must not be used") + } + async fn forward_count_tokens( + &self, + node: &str, + provider: &str, + request: CanonicalRequest, + ) -> Result { + assert_eq!(node, "peer"); + assert_eq!(provider, "counter"); + assert!(!request.stream); + assert!(request.body.get("stream").is_none()); + self.calls.fetch_add(1, Ordering::SeqCst); + self.peer + .dispatch_count_local(provider, request) + .await + .map_err(|_| UpstreamError::Network("test peer dispatch failed".into())) + } +} +#[tokio::test] +async fn remote_count_operation_and_local_dispatch_do_not_loop() { + let counter = Arc::new(Counter::default()); + // The peer's router would send it back out if local dispatch did not bypass routing. + let (_peer_dir, peer) = + start(store("new"), counter.clone(), Some(Arc::new(LegacyRouter))).await; + let bridge = Arc::new(Bridge { + peer: Arc::new(peer), + calls: AtomicUsize::new(0), + }); + let (_dir, handle) = start(Arc::default(), Arc::new(Legacy), Some(bridge.clone())).await; + let response = count(&handle).await; + assert_eq!(response.status(), 200); + assert_eq!( + response.json::().await.unwrap(), + json!({"input_tokens":7}) + ); + assert_eq!(bridge.calls.load(Ordering::SeqCst), 1); + assert_eq!(*counter.calls.lock().unwrap(), ["new"]); + handle.shutdown().await; + let Bridge { peer, .. } = Arc::try_unwrap(bridge).ok().unwrap(); + Arc::try_unwrap(peer).ok().unwrap().shutdown().await; +} diff --git a/crates/cranibox-llm/tests/providers/lifecycle.rs b/crates/cranibox-llm/tests/providers/lifecycle.rs index 5eebe1a..5814c6b 100644 --- a/crates/cranibox-llm/tests/providers/lifecycle.rs +++ b/crates/cranibox-llm/tests/providers/lifecycle.rs @@ -1,7 +1,163 @@ use super::*; +use cranibox_contracts::llm::mgmt::{LlmMgmt, SignInStartReq}; use cranibox_contracts::vault::StoreError; +use cranibox_llm::mgmt::Client; use std::sync::atomic::{AtomicBool, Ordering}; +async fn managed( + mock: &Mock, + store: Arc, +) -> (tempfile::TempDir, cranibox_llm::Handle, Client) { + let dir = tempfile::tempdir().unwrap(); + let mut config = + cranibox_llm::Config::new("local".into(), dir.path().join("run"), store.clone()); + config.listen = vec!["127.0.0.1:0".parse().unwrap()]; + config.sign_in = Some(Arc::new(OAuth::new( + "local".into(), + store, + OAuthConfig::for_loopback(&mock.base).unwrap(), + ))); + let handle = cranibox_llm::spawn(config).await.unwrap(); + let client = Client::new(&handle.management); + (dir, handle, client) +} + +#[tokio::test] +async fn management_cancel_releases_browser_and_prevents_device_polling() { + let mock = Mock::new().await; + let store = Arc::new(MemoryAccountStore::default()); + let (_dir, handle, client) = managed(&mock, store.clone()).await; + let (flow, url) = browser( + client + .signin_start(SignInStartReq { + provider: "codex".into(), + region: None, + }) + .await + .unwrap(), + ); + client.signin_cancel(&flow).await.unwrap(); + client.signin_cancel(&flow).await.unwrap(); + listener_closed(&url).await; + assert!(matches!( + client.signin_complete(&flow, no_input()).await, + Err(MgmtError::Invalid(_)) + )); + assert!(matches!( + client.signin_cancel("missing").await, + Err(MgmtError::NotFound) + )); + assert!(matches!( + client.signin_cancel("../accounts").await, + Err(MgmtError::Invalid(_)) + )); + mock.replies("/device", vec![(200, json!({"device_code":"fake-device","user_code":"TEST","verification_uri":"https://example.invalid/device","expires_in":600,"interval":1}))]); + let device = client + .signin_start(SignInStartReq { + provider: "kimi".into(), + region: None, + }) + .await + .unwrap(); + let SignInStart::DeviceCode { flow_id, .. } = device else { + panic!("device flow") + }; + client.signin_cancel(&flow_id).await.unwrap(); + tokio::time::sleep(Duration::from_millis(1050)).await; + assert!(matches!( + client.signin_complete(&flow_id, no_input()).await, + Err(MgmtError::Invalid(_)) + )); + assert_eq!(mock.calls().len(), 1); + assert!(store.list().is_empty()); + assert_eq!( + reqwest::Client::new() + .post(format!( + "http://{}/signin/{flow}/cancel", + handle.addresses[0] + )) + .send() + .await + .unwrap() + .status(), + 404 + ); + handle.shutdown().await; +} + +#[tokio::test] +async fn management_cancel_keeps_completed_signin() { + let mock = Mock::new().await; + mock.replies("/token", vec![(200, token())]); + let store = Arc::new(MemoryAccountStore::default()); + let (_dir, handle, client) = managed(&mock, store.clone()).await; + let (flow, url) = browser( + client + .signin_start(SignInStartReq { + provider: "codex".into(), + region: None, + }) + .await + .unwrap(), + ); + let account = client + .signin_complete( + &flow, + SignInInput { + callback_url: Some(callback(&url, None)), + code: None, + }, + ) + .await + .unwrap(); + client.signin_cancel(&flow).await.unwrap(); + assert_eq!( + client.signin_complete(&flow, no_input()).await.unwrap(), + account + ); + assert_eq!(store.list().len(), 1); + assert_eq!(mock.calls().len(), 1); + handle.shutdown().await; +} + +#[tokio::test] +async fn management_cancel_keeps_credentials_awaiting_retry() { + let mock = Mock::new().await; + mock.replies("/token", vec![(200, token())]); + let store = Arc::new(FlakyStore::default()); + let (_dir, handle, client) = managed(&mock, store.clone()).await; + let (flow, url) = browser( + client + .signin_start(SignInStartReq { + provider: "codex".into(), + region: None, + }) + .await + .unwrap(), + ); + store.fail.store(true, Ordering::SeqCst); + assert!(matches!( + client + .signin_complete( + &flow, + SignInInput { + callback_url: Some(callback(&url, None)), + code: None + } + ) + .await, + Err(MgmtError::Other(_)) + )); + assert!(matches!( + client.signin_cancel(&flow).await, + Err(MgmtError::Invalid(_)) + )); + let account = client.signin_complete(&flow, no_input()).await.unwrap(); + assert!(store.get_secret(&account.id).is_ok()); + assert_eq!(mock.calls().len(), 1); + handle.shutdown().await; +} + #[derive(Default)] struct FlakyStore { inner: MemoryAccountStore, From e179c388fbe34559c165e0d50f06352e3110f168 Mon Sep 17 00:00:00 2001 From: fliaping <269832017@qq.com> Date: Mon, 19 Oct 2026 22:10:16 +0800 Subject: [PATCH 2/3] [WS2] retain the reviewed integration dependency lock --- Cargo.lock | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6ae7cdf..393fc48 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2450,7 +2450,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bb0228f477c0900c880fd78c8759b95c7636dbd7842707f49e132378aa2acdc" dependencies = [ "heck 0.4.1", - "proc-macro-crate 2.0.2", + "proc-macro-crate 2.0.0", "proc-macro-error", "proc-macro2", "quote", @@ -4849,11 +4849,10 @@ dependencies = [ [[package]] name = "proc-macro-crate" -version = "2.0.2" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b00f26d3400549137f92511a46ac1cd8ce37cb5598a96d382381458b992a5d24" +checksum = "7e8366a6159044a37876a2b9817124296703c586a5c92e2c53751fa06d8d43e8" dependencies = [ - "toml_datetime 0.6.3", "toml_edit 0.20.2", ] @@ -6829,7 +6828,7 @@ checksum = "185d8ab0dfbb35cf1399a6344d8484209c088f75f8f68230da55d48d95d43e3d" dependencies = [ "serde", "serde_spanned 0.6.9", - "toml_datetime 0.6.3", + "toml_datetime 0.6.11", "toml_edit 0.20.2", ] @@ -6850,9 +6849,9 @@ dependencies = [ [[package]] name = "toml_datetime" -version = "0.6.3" +version = "0.6.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7cda73e2f1397b1262d6dfdcef8aafae14d1de7748d66822d3bfeeb6d03e5e4b" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" dependencies = [ "serde", ] @@ -6873,7 +6872,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1b5bb770da30e5cbfde35a2d7b9b8a2c4b8ef89548a7a6aeab5c9a576e3e7421" dependencies = [ "indexmap 2.14.2", - "toml_datetime 0.6.3", + "toml_datetime 0.6.11", "winnow 0.5.40", ] @@ -6886,7 +6885,7 @@ dependencies = [ "indexmap 2.14.2", "serde", "serde_spanned 0.6.9", - "toml_datetime 0.6.3", + "toml_datetime 0.6.11", "winnow 0.5.40", ] From 729fab96b4c6654fabf94f6ca4c256bc18df9ab7 Mon Sep 17 00:00:00 2001 From: fliaping <269832017@qq.com> Date: Mon, 19 Oct 2026 22:27:59 +0800 Subject: [PATCH 3/3] [WS2][WS3] document cancellation and count-token verification --- crates/cranibox-llm/tests/IMPLEMENTATION.md | 98 +++++++++------------ 1 file changed, 43 insertions(+), 55 deletions(-) diff --git a/crates/cranibox-llm/tests/IMPLEMENTATION.md b/crates/cranibox-llm/tests/IMPLEMENTATION.md index 7f0b308..ac11951 100644 --- a/crates/cranibox-llm/tests/IMPLEMENTATION.md +++ b/crates/cranibox-llm/tests/IMPLEMENTATION.md @@ -1,72 +1,60 @@ # WS2 / WS3 交付与验收记录 -日期：2026-10-19。已跟进 main `6d4a8a4`（Phase 0 契约 PR #16），Windows CI 导入修复为 `8abcd1a`，P2 厂商与 OAuth 生命周期补充为 `0edb2e7`。当前补充刷新协调器已保存凭据时避免重复写入的修复。本记录分别列出 WS2/WS3 隔离验证和包含其他工作流未提交代码的联合快照验证；联合通过不代表这些改动已随本 PR 提交或合入。WS4 首轮观测见 [Claude cch 报告](claude/CCH-FINDINGS.md)，真实 Claude 验证按用户要求后置。 +日期：2026-10-19。已跟进 main `d2e5ec2` 和代理 PR #20 分支 `eedcc29`。本次补全私有管理登录取消、独立 token 计数与 OAuth 刷新；共享接口单独提交为 [CONTRACT-REQUEST #23](https://github.com/fliaping/cranibox/pull/23)，尚待协调者评审。WS6 / node / App 已提交为 PR #22。本记录区分代理核心和多节点集成。Claude 真实验证按用户要求后置，首轮观测见 [cch 报告](claude/CCH-FINDINGS.md)。 ## 已实现 -- `spawn(Config) -> Handle`：默认监听 `127.0.0.1:3424` / `[::1]:3424`，逐请求核验连接来源和 Host；转发头、Tailscale 头、浏览器来源头统一拒绝。`cbx-*` 只标识 Agent，不是权限凭证。 -- Chat / Responses / Messages 原生直通和以 Messages 为中间格式的转换，非流式与 SSE、并行工具调用、reasoning；首字节前暂存和重试，同一提供商账号池、会话亲和、限流与额度冷却。输出开始后不重放请求。 -- C3 管理路由、类型化 socket 客户端与 SSE；Unix socket 0600，独立目录 0700，没有管理 TCP 监听。Windows 使用 owner-only named pipe 并拒绝远程客户端；同用户管理访问已在 Windows runner 通过，另增加拒绝重复服务和检查受保护 owner-only DACL 的回归。 -- API key、自定义提供商、33 个内置预设、可校验的 TOML override、模型发现和静态模型表；curl 作为数据解析，不执行命令；最小连接探测。国内覆盖 DeepSeek、智谱/z.ai、百炼 Coding/Token、千问 Token、DashScope、火山、MiniMax、Moonshot、Kimi Code，新增阶跃、小米 MiMo、百度千帆、腾讯混元与 TokenHub。[P2 官方入口依据](providers/ENDPOINTS.md)。 -- Codex：PKCE、loopback / 粘贴回调、Responses 订阅端点、刷新轮换和撤销判定；显式 CLI 登录导入要求确认原 CLI 可能失效。 -- Kimi：设备码、pending/slow_down/过期/拒绝、国内外端点、五个 X-Msh 头、402、schema 适配。 -- Gemini：Google PKCE、Code Assist 项目发现和协议适配。部署者需设置自己的桌面 OAuth client：`CRANIBOX_GOOGLE_CLIENT_ID` / `CRANIBOX_GOOGLE_CLIENT_SECRET`；账号需已开通 Code Assist 并能发现项目。没有自动导入 Gemini CLI 凭据，也没有新项目开通向导。 -- Copilot：GitHub 设备码换短期 Copilot token、模型列表、严格的 API host 校验和 token 刷新。 -- 登录生命周期：接收成功/拒绝回调、取消、过期或最后一个 OAuth 实例退出时释放回调监听；验证 state 后才处理拒绝，错误说明不反射到页面，回调响应禁止缓存和 Referer 传播。进程内 `OAuth::cancel` 可停止未交换的流程；已交换但尚未保存的凭据必须先重试保存，不能因取消而丢失。 -- OAuth 刷新按本机账号串行，HTTP 断开不取消刷新；token 已轮换但存储暂时写失败时暂存在内存，下次写入时恢复。进程崩溃恢复和多节点刷新租约由 WS6 负责。 -- 兼容 WS6 协调刷新：上游已持久化同一新 secret 时不再调用 `put`，避免重复版本和无谓的写失败；存储中已是更晚凭据时拒绝覆盖，保留协调器更新的账号元数据。仍保留普通 adapter 的写失败恢复路径。 +- 数据面默认监听 `127.0.0.1:3424` / `[::1]:3424`，逐请求检查来源与 Host，拒绝转发头、Tailscale 头和浏览器来源头。`cbx-*` 只标识 Agent，不是权限凭证。 +- Chat / Responses / Messages 原生直通及协议转换、非流式 / SSE、并行工具与 reasoning；首字节前重试、同提供商账号池、会话亲和、限流与额度冷却。输出开始后不重放请求。 +- 管理面：类型化客户端和 SSE；Unix socket 0600、目录 0700；Windows owner-only named pipe，拒绝远程访问，包含重复服务和受保护 DACL 回归；没有管理 TCP 监听。 +- API key、自定义提供商、33 个预设、TOML override、模型发现与静态目录；curl 仅解析数据，不执行命令。国内包含 DeepSeek、智谱/z.ai、百炼 Coding/Token、千问 Token、DashScope、火山、MiniMax、Moonshot、Kimi Code、阶跃、MiMo、千帆、混元和 TokenHub。[官方入口依据](providers/ENDPOINTS.md)。 +- Codex：PKCE、loopback / 粘贴回调、Responses 订阅端点、刷新轮换和撤销判定。显式 CLI 登录导入须确认原 CLI 可能失效。 +- Kimi：设备码、pending / slow_down / 过期 / 拒绝、国内外端点、五个 X-Msh 头、402 和 schema 适配。 +- Gemini：Google PKCE、Code Assist 项目发现和协议适配。部署者需配置自己的 `CRANIBOX_GOOGLE_CLIENT_ID` / `CRANIBOX_GOOGLE_CLIENT_SECRET`；账号需已开通 Code Assist 且可发现项目。没有自动导入 Gemini CLI 凭据或新项目开通向导。 +- Copilot：GitHub 设备码换短期 Copilot token、模型列表、严格 API host 校验和 token 刷新。 +- 登录生命周期：完成、拒绝、取消、过期和退出时释放回调监听；先检查 state，不反射错误详情，禁止缓存和 Referer 传播。新增 `signin_cancel(flow)`：重复取消成功、未知 flow 返回 404；完成后保留账号；已交换但未保存的凭据必须先重试保存，不能因取消丢失。数据 TCP 端口不提供该管理接口。 +- OAuth 刷新按本机账号串行，HTTP 断开不取消刷新；轮换后暂时写失败会保留内存并重试。协调器已持久化同一 secret 时不重复写入；遇到更晚凭据拒绝覆盖，保留账号元数据。跨节点租约和崩溃恢复由 WS6 负责。 +- 独立 token 计数：强制 Messages、非流式并移除 JSON `stream`，不会调用生成接口代替计数。OAuth adapter 计数可复用计数前 / 401 后刷新流程。 -## 接入说明与契约请求 +## 接入与契约请求 -共享 C1–C8 类型保持不变。生产启动必须在 node/web 生命周期中持有 Handle，并提供持久化 `Arc`；不要用开发用 `MemoryAccountStore` 启动正式服务。联合快照中的 `workspace-node/src/models.rs` 已接入 SQLite vault、Router 和协调刷新装饰器，App 层已接入管理 socket；这些属于其他工作流的未提交改动，交付时必须同时纳入。调用者提供当前用户独占的 run 目录。 +C3 增加兼容默认实现的 `Upstream::count_tokens`、`SignIn::cancel`、`LlmMgmt::signin_cancel`；序列化结构和 fixtures 不变，旧实现默认 Unsupported。私有管理路径为 `POST /signin/{flow}/cancel`，无请求体，成功 204。尚未增加 UI 取消按钮。 -```rust,ignore -let mut cfg = cranibox_llm::Config::new(node_id, private_run_dir, persistent_store); -cfg.presets = cranibox_llm::catalog::load_overrides(&presets_directory)?; -cfg.routing = Some(authenticated_mesh_router); // 可选，WS6 实现 Router -let llm = cranibox_llm::spawn(cfg).await?; -// 将 llm 保存在运行时状态，退出时 llm.shutdown().await。 -``` +生产启动必须持有 `spawn(Config) -> Handle` 并提供持久化 `Arc`、用户独占的运行目录；退出调用 `Handle::shutdown`。不要用开发用 MemoryAccountStore 启动正式服务。PR #22 的 node 已接入 SQLite vault / Router / 协调刷新，App 使用私有管理 socket，由对应工作流评审。 -WS6 可实现 `route::Router::{decide,forward}`，并在验证 mesh 身份与授权后调用 `Handle::dispatch_local`，避免再次远端路由。这个进程内入口不开放 TCP。`load_overrides` 在启动时加载，未实现文件热更新。 +WS6 实现 `Router::{decide,forward,forward_count_tokens}`，身份与授权检查后调用 `Handle::dispatch_local` / `dispatch_count_local`，避免再次远端路由。这些进程内入口不开放 TCP。TOML override 只在启动加载，尚无热更新。 -**CONTRACT-REQUEST（WS0/WS6）**： +原生 Messages API key 已接入 `POST /v1/messages/count_tokens`。**当前 Codex、Kimi、Gemini、Copilot adapter 没有原生计数实现，仍明确返回 501。** 新接口只允许 adapter 按实际能力实现计数；默认 Router 也返回 Unsupported，不能改发生成请求。 -1. C3 `CanonicalRequest` / `Upstream` 缺少 count-tokens 操作。当前 `/v1/messages/count_tokens` 仅支持本机原生 Messages API key；OAuth 和远端计数返回 501。需要显式操作类型，不能把控制信息塞进用户 JSON 或 Agent 名称。 -2. 原 refresh lease / generation 接入请求已通过 WS6 的 `vault::coordinated::Coordinated` 装饰器在联合快照中验证，无需为这条路径修改 C7。生产调用必须使用该装饰器；单独启用普通 adapter 仍只保证本进程串行，不能代替跨节点租约。 -3. Claude 节点选择需要名册持久化和旧节点清理。当前 `/claude/set-node` 只接受本机并保存在内存中；远端返回 `claude_node_offline`。没有 Claude 订阅 adapter，不代表完成 WS4 节点迁移。 -4. 运行时 store / Router 和 UI 转发已在联合快照中验证，请对应工作流提交并评审；NOTICE 仍需按 THIRD-PARTY.md 更新。上述挂载点不在本次自有提交中。 -5. C3 `SignIn` / `LlmMgmt` 尚无取消登录操作。当前只提供进程内 `OAuth::cancel`，没有另造管理路由；请协调者决定取消接口后再接入 UI。 +CONTRACT-REQUEST（WS0/WS6）：#23 单独评审共享接口。跨节点计数另行提交：RemoteRouter 使用既有 LLM_COUNT_TOKENS 路径，Coordinated 代理计数方法，mesh 入口保留 Unsupported 的 501。不能用进程内 Router 测试替代真实 mesh 联调。刷新租约沿用现有 C7；生产需要 Coordinated，普通 adapter 仅保证本进程串行。Claude 节点持久化由 WS6 / App 交付，没有 Claude adapter 不代表 WS4 完成。 ## 已执行验证 -全部运行在 `admin@192.168.217.6:22022` 的 `cranibox-validation` 容器、用户 abc；自有隔离目录为 `/config/validation-sources/llm-proxy-20261019`，联合目录为 `/config/validation-sources/llm-integration-20261019`。未在日常容器或用户指定不用的本机运行测试。所有账号和厂商上游均为 mock；三节点联调用真实 TCP + Noise 传输。 +全部在 `admin@192.168.217.6:22022` 的 `cranibox-validation` 容器、用户 abc 运行。本次核心源码目录 `/config/validation-sources/llm-completion-20261019`，独立构建目录 `/config/validation-build/ws2-completion`。没有在本机运行构建、测试或服务；账号和厂商上游均为 mock。 | 检查 | 结果 | |---|---| -| `cargo fmt --all --check` | 通过 | -| `cargo clippy --workspace --locked --all-targets -- -D warnings` | 通过 | -| `cargo test --locked`（联合快照完整工作区） | 460 通过、5 个已有显式忽略项；延迟测试此前另行运行通过 | -| `cranibox-llm`（WS2/WS3 自有代码隔离验证） | 202 通过：169 单元、12 代理、5 协议回放、1 目录、15 订阅；延迟测试默认 ignored | -| `cranibox-llm`（联合快照） | 210 通过，包含 WS6 的 6 项多节点测试及新增 2 项单元测试 | -| 3×3 回放 | 请求、非流式响应、单字节 UTF-8 SSE、并行工具与 reasoning；另验证混合换行及截断错误 | -| 本机访问控制 | 非 loopback / 缺少 peer、全部转发头、外部 Host、浏览器头；socket 权限和 TCP 隔离 | -| 账号池 | 401/429/额度/连接失败首字节前切换；输出后不切换；同会话十次；刷新并发、断开和写失败恢复 | -| 订阅与国内预设 | 四家登录/adapter mock；全部 33 个预设端点和认证（含 MiMo `api-key`、阶跃 Messages Bearer）；智谱 200 信封/1113；Kimi schema/设备码；新增取消/拒绝/回调释放/写失败后凭据保留回归 | -| 显式延迟测试（120 样本/路径） | direct p50 446 µs；proxy p50 1305 µs；额外 859 µs，低于 2 ms。该值仅代表此主机和测试负载。 | -| Claude Code 2.1.303 本地 mock | API key 3 次 + Bearer 2 次；未见 cch；真实订阅行为未证实 | -| Linux / Windows / macOS CI | `0edb2e7` 的 GitHub Actions run `39247033227` 中，三平台均通过工作区 fmt / Clippy / test；Windows 管理客户端、拒绝重复服务及受保护 owner-only DACL 回归已通过。桌面打包等其他步骤状态以 PR 检查为准。 | - -联合快照以 `8076a2d4bda85c28ba179ee6fe2124e8db1108c5` 为底，覆盖 87 个当时的工作区文件，再覆盖本次 `engine.rs` / `proxy.rs` 修复。初始源码包 SHA-256 为 `0ed76a910b4ac37feb91bfdcde874c78d02a31cdfef62b2be276a33bc2512a70`；文件清单与哈希保存在验证主机的独立源码目录中。 - -新回归已先验证旧实现失败（预期 200、实际 503），再验证修复通过。联合测试还覆盖：加密同步和本机显式启用、SQLite 重启不丢凭据、并发刷新只调用一次厂商、租约过期接管与恢复交还、撤销后的密钥轮换，以及三节点通过 Noise 同步、转发、Hub 停止后继续使用本机账号。它们不能替代真实厂商验收。 - -## 真实账号人工验收（尚未执行） - -1. 由协调者用已授权账号分别登录 Codex、Kimi、Gemini、Copilot，确认登录状态、模型列表和一条流式工具请求；确认 Codex 刷新后仍可持续请求，撤销后显示重新登录。 -2. 各用 DeepSeek、智谱、百炼、火山、MiniMax、阶跃、MiMo、千帆、混元/TokenHub 的真实 key 发一条最小请求；确认区域对应 host、套餐类型和扣费来源。测试连接会发送最小模型请求，可能产生费用。 -3. Windows 补充实际不同用户、远程 SMB 连接拒绝验证；自动测试的同用户访问与 DACL 检查不能代替这两种真实身份测试。macOS 的 UDS、双栈和 OAuth mock 已由 CI 验证，真实浏览器登录仍需人工验收。 -4. 将已经联合验证的 WS6 / node / App 改动提交并完成评审，再在实际设备上验收配对、登录、重启和网络切换；自动化加密传输、租约、持久化、Hub 离线与撤销验证已通过。 -5. 真实 Claude 订阅与默认官方 base URL 的验证后置，待恢复该任务时按 cch 报告执行。不要将本地 mock 结果作为真实上游验收。 - -没有宣称真实厂商订阅已验收、Windows 不同用户/远程身份测试已完成或 WS4/WS6 已完成。 +| cargo fmt --all --check | 通过 | +| cargo clippy --workspace --locked --all-targets -- -D warnings | 通过 | +| cargo test --workspace --locked（核心与当前 main） | **455 通过，0 失败，6 个显式忽略项** | +| cranibox-llm | **209 通过**：169 单元、12 代理、5 协议回放、1 目录、18 订阅、4 计数；延迟默认 ignored | +| cranibox-contracts | **11 通过**：单元、原有 fixtures、2 项接口兼容测试 | +| 新增管理回归 | 浏览器/设备码取消、幂等、未知/非法 flow、回调端口释放、停止轮询、完成后保留账号、待保存凭据取消被拒绝且重试不重复交换、数据 TCP 隔离 | +| 新增计数回归 | OAuth 401 后仍走计数；旧 adapter / Router 501 且从不生成；本地分发绕过再次路由 | +| 原有核心回归 | 3×3 协议回放、单字节 UTF-8 SSE、工具 / reasoning、来源与 OS socket 控制、账号池、首字节前重试、亲和、并发刷新与写失败恢复 | +| 订阅与国内厂商 | 四家登录 / adapter mock 和全部 33 个预设端点、认证、套餐差异 | +| 先前延迟测试（120 样本/路径） | direct p50 446 µs；proxy p50 1305 µs；额外 859 µs，仅代表此前主机和负载，本轮未重跑 | +| Claude Code 2.1.303 本地 mock | API key 3 次、Bearer 2 次；未见 cch，真实订阅行为未证实 | +| 跨平台 CI | 本轮 Actions 尚在运行，以对应 PR 当前检查为准，不用旧提交结果代替本次验证 | + +日志位于远端源码目录：`completion-fmt.log`、`completion-clippy-private.log`、`completion-tests-private.log`。共享缓存曾出现不属于当前源码的旧 contract 元数据，切换独立构建目录后全量通过；表中只采用独立目录结果。 + +先前联合快照的加密同步、SQLite 重启、刷新租约与撤销作为历史证据保留。本次 mesh 计数在 `/config/validation-sources/count-mesh-20261019` 单独验证，结果随其集成 PR 记录；上述 455 项不代表新多节点补丁已验收。 + +## 尚需人工验收 + +1. 使用已授权真实账号登录 Codex、Kimi、Gemini、Copilot，确认状态、模型列表、流式工具请求、刷新和撤销。Gemini 需部署者配置 OAuth client / Code Assist 项目。 +2. 国内厂商真实 key 最小请求，核对区域 host、套餐和扣费来源；连接测试会发送模型请求，可能产生费用。 +3. Windows 不同用户和远程 SMB 拒绝验证；同用户与 DACL 自动测试不能代替真实身份测试。macOS 真实浏览器登录仍需人工验收。 +4. 完成 #22 与计数集成请求评审，再在实际设备验收配对、登录、重启和网络切换；mock 上游和加密传输测试不能替代真实设备/账号验收。 +5. Claude 真实订阅与官方 base URL 验证按用户要求后置，恢复时依据 cch 报告执行。