diff --git a/backend/src/dispatch/controllers/cyber_policy/mod.rs b/backend/src/dispatch/controllers/cyber_policy/mod.rs index ccb50fa7e..5951019c8 100644 --- a/backend/src/dispatch/controllers/cyber_policy/mod.rs +++ b/backend/src/dispatch/controllers/cyber_policy/mod.rs @@ -11,8 +11,11 @@ mod types; use crate::{ dispatch::{ - affinity::SessionAffinityService, controllers::ControllerFailureFact, - errors::ClientFailure, transport::observation::UpstreamFailureFacts, + affinity::SessionAffinityService, + controllers::ControllerFailureFact, + errors::ClientFailure, + lifecycle::contract::{AttemptDecision, AttemptObservation, AttemptReturnKind}, + transport::observation::UpstreamFailureFacts, }, upstream::openai::protocol::responses::{CodexResponsesRequest, ResponsesSseFailure}, }; @@ -48,6 +51,11 @@ pub(in crate::dispatch) struct CyberPolicyController { store: Store, } +/// 已由 Cyber owner 认领的失败;调用方只能请求 owner 产生 effect 或 decision。 +pub(super) struct ClassifiedCyberPolicyFailure<'a> { + fact: ControllerFailureFact<'a>, +} + impl CyberPolicyController { pub(in crate::dispatch) fn new(session_affinity: Arc) -> Self { Self { @@ -55,18 +63,46 @@ impl CyberPolicyController { } } - pub(super) fn client_failure(failure: ResponsesSseFailure) -> ClientFailure { - ClientFailure::new(failure, 400, true) + pub(super) fn classify( + fact: ControllerFailureFact<'_>, + ) -> Option> { + let owned = match fact { + ControllerFailureFact::Upstream(facts) => { + facts + .status_code + .is_some_and(|status| (400..=499).contains(&status)) + && is_cyber_policy_code(facts.code.as_deref()) + } + ControllerFailureFact::Response(failure) => is_cyber_policy_failure(failure), + }; + owned.then_some(ClassifiedCyberPolicyFailure { fact }) } - pub(super) fn attempt_client_failure(failure: ControllerFailureFact<'_>) -> ClientFailure { - match failure { + pub(super) fn client_failure(classified: &ClassifiedCyberPolicyFailure<'_>) -> ClientFailure { + match classified.fact { ControllerFailureFact::Upstream(facts) => ClientFailure::new( failure_from_upstream_facts(facts), facts.status_code.unwrap_or(400), true, ), - ControllerFailureFact::Response(failure) => Self::client_failure(failure.clone()), + ControllerFailureFact::Response(failure) => { + ClientFailure::new(failure.clone(), 400, true) + } + } + } + + pub(super) fn decision( + observation: &AttemptObservation, + classified: ClassifiedCyberPolicyFailure<'_>, + ) -> AttemptDecision { + let client_failure = Self::client_failure(&classified); + if observation.routing.can_retry_next_candidate { + AttemptDecision::RetryNextCandidate { + exhaustion: None, + on_exhaustion: Some(client_failure), + } + } else { + AttemptDecision::Return(AttemptReturnKind::Failed(client_failure)) } } @@ -122,18 +158,6 @@ impl CyberPolicyController { } } - pub(super) fn owns_failure(&self, failure: ControllerFailureFact<'_>) -> bool { - match failure { - ControllerFailureFact::Upstream(facts) => { - facts - .status_code - .is_some_and(|status| (400..=499).contains(&status)) - && is_cyber_policy_code(facts.code.as_deref()) - } - ControllerFailureFact::Response(failure) => is_cyber_policy_failure(failure), - } - } - pub(in crate::dispatch) async fn observe_success(&self, plan: &CyberPolicyScope) { if !plan.has_failures() { return; diff --git a/backend/src/dispatch/controllers/mod.rs b/backend/src/dispatch/controllers/mod.rs index c9db57ad9..7ee8bc322 100644 --- a/backend/src/dispatch/controllers/mod.rs +++ b/backend/src/dispatch/controllers/mod.rs @@ -32,7 +32,9 @@ use crate::{ use crate::{ dispatch::{ controllers::cloudflare::CloudflareRecovery, - controllers::cyber_policy::{CyberPolicyController, CyberPolicyScope}, + controllers::cyber_policy::{ + ClassifiedCyberPolicyFailure, CyberPolicyController, CyberPolicyScope, + }, controllers::{ account_failure::AccountFailureController, history::HistoryController, history::HistoryState, quota::QuotaController, @@ -168,6 +170,30 @@ enum ControllerFailureFact<'a> { Response(&'a ResponsesSseFailure), } +#[derive(Clone, Copy)] +enum FailureObservation<'a> { + Attempt(&'a AttemptObservation), + Stream { + account_id: &'a str, + failure: &'a ResponsesSseFailure, + }, +} + +impl<'a> FailureObservation<'a> { + fn fact(self) -> Option> { + match self { + Self::Attempt(observation) => ControllerFailureFact::from_attempt(observation), + Self::Stream { failure, .. } => Some(ControllerFailureFact::Response(failure)), + } + } +} + +enum SharedFailureClassification<'a> { + CyberPolicy(ClassifiedCyberPolicyFailure<'a>), + AccountFailure(account_failure::ClassifiedFailure), + Quota(quota::ClassifiedQuotaFailure), +} + impl<'a> ControllerFailureFact<'a> { fn from_attempt(observation: &'a AttemptObservation) -> Option { match &observation.kind { @@ -183,40 +209,30 @@ impl<'a> ControllerFailureFact<'a> { /// 单次 attempt 的唯一分类结果;副作用与 decision 必须共同消费这个值。 enum AttemptClassification<'a> { - CyberPolicy(ControllerFailureFact<'a>), + Shared(SharedFailureClassification<'a>), Cloudflare(cloudflare::CloudflareFailure), History(AttemptDecision), - AccountFailure(account_failure::ClassifiedFailure), - Quota(quota::ClassifiedQuotaFailure), Default(AttemptDecision), } -enum StreamFailureClassification { - CyberPolicy(ClientFailure), - AccountFailure { - classified: account_failure::ClassifiedFailure, - client_failure: ClientFailure, - }, - Quota { - classified: quota::ClassifiedQuotaFailure, - client_failure: ClientFailure, - }, - Default(ClientFailure), +struct StreamFailureClassification<'a> { + failure: &'a ResponsesSseFailure, + owner: Option>, } -impl StreamFailureClassification { - fn client_failure(&self) -> &ClientFailure { - match self { - Self::CyberPolicy(failure) - | Self::AccountFailure { - client_failure: failure, - .. +impl StreamFailureClassification<'_> { + fn client_failure(&self) -> ClientFailure { + match &self.owner { + Some(SharedFailureClassification::CyberPolicy(classified)) => { + CyberPolicyController::client_failure(classified) } - | Self::Quota { - client_failure: failure, - .. + Some(SharedFailureClassification::AccountFailure(classified)) => { + AccountFailureController::client_failure(classified, self.failure) } - | Self::Default(failure) => failure, + Some(SharedFailureClassification::Quota(_)) => { + QuotaController::client_failure(self.failure) + } + None => default_response_client_failure(self.failure), } } } @@ -523,60 +539,47 @@ impl ControllerSet { } } - fn classify_stream_failure( + fn classify_stream_failure<'a>( &self, - account_id: &str, - failure: &ResponsesSseFailure, - ) -> StreamFailureClassification { - let fact = ControllerFailureFact::Response(failure); - if self.cyber_policy.owns_failure(fact) { - return StreamFailureClassification::CyberPolicy( - CyberPolicyController::client_failure(failure.clone()), - ); + account_id: &'a str, + failure: &'a ResponsesSseFailure, + ) -> StreamFailureClassification<'a> { + StreamFailureClassification { + failure, + owner: Self::classify_shared_failure(FailureObservation::Stream { + account_id, + failure, + }), } - if let Some(classified) = AccountFailureController::classify_failure(account_id, fact) { - let client_failure = AccountFailureController::client_failure(&classified, failure); - return StreamFailureClassification::AccountFailure { - classified, - client_failure, - }; - } - if let Some(classified) = QuotaController::classify_failure(account_id, fact) { - return StreamFailureClassification::Quota { - classified, - client_failure: QuotaController::client_failure(failure), - }; - } - StreamFailureClassification::Default(default_response_client_failure(failure)) } async fn apply_stream_failure( &self, scope: &ControllerRequestScope, account_id: &str, - classification: &StreamFailureClassification, + classification: &StreamFailureClassification<'_>, ) { - match classification { - StreamFailureClassification::CyberPolicy(_) => { + match classification.owner.as_ref() { + Some(SharedFailureClassification::CyberPolicy(_)) => { self.cyber_policy .exclude_account(&scope.cyber_policy, account_id) .await; } - StreamFailureClassification::AccountFailure { classified, .. } => { + Some(SharedFailureClassification::AccountFailure(classified)) => { let _ = best_effort_controller_io( "account_failure.stream", AccountFailureController::apply_effect(&self.account_pool, classified), ) .await; } - StreamFailureClassification::Quota { classified, .. } => { + Some(SharedFailureClassification::Quota(classified)) => { let _ = best_effort_controller_io( "quota.stream", QuotaController::apply_effect(&self.account_pool, classified), ) .await; } - StreamFailureClassification::Default(_) => {} + None => {} } } @@ -714,17 +717,33 @@ impl ControllerSet { } match &classification { - AttemptClassification::CyberPolicy(_) => { - if let Some(account_id) = observation - .account - .as_ref() - .map(|account| account.id.as_str()) - { - self.cyber_policy - .exclude_account(&scope.cyber_policy, account_id) - .await; + AttemptClassification::Shared(shared) => match shared { + SharedFailureClassification::CyberPolicy(_) => { + if let Some(account_id) = observation + .account + .as_ref() + .map(|account| account.id.as_str()) + { + self.cyber_policy + .exclude_account(&scope.cyber_policy, account_id) + .await; + } } - } + SharedFailureClassification::AccountFailure(classified) => { + let _ = best_effort_controller_io( + "account_failure.attempt", + AccountFailureController::apply_effect(&self.account_pool, classified), + ) + .await; + } + SharedFailureClassification::Quota(classified) => { + let _ = best_effort_controller_io( + "quota.attempt", + QuotaController::apply_effect(&self.account_pool, classified), + ) + .await; + } + }, AttemptClassification::Cloudflare(failure) => { let _ = best_effort_controller_io( "cloudflare.attempt", @@ -736,20 +755,6 @@ impl ControllerSet { ) .await; } - AttemptClassification::AccountFailure(classified) => { - let _ = best_effort_controller_io( - "account_failure.attempt", - AccountFailureController::apply_effect(&self.account_pool, classified), - ) - .await; - } - AttemptClassification::Quota(classified) => { - let _ = best_effort_controller_io( - "quota.attempt", - QuotaController::apply_effect(&self.account_pool, classified), - ) - .await; - } AttemptClassification::History(_) | AttemptClassification::Default(_) => {} } @@ -760,44 +765,83 @@ impl ControllerSet { &self, observation: &'a AttemptObservation, ) -> AttemptClassification<'a> { - let failure = ControllerFailureFact::from_attempt(observation); - if let Some(failure) = failure - && self.cyber_policy.owns_failure(failure) - { - return AttemptClassification::CyberPolicy(failure); - } + let shared = Self::classify_shared_failure(FailureObservation::Attempt(observation)); + let shared = match shared { + Some(shared @ SharedFailureClassification::CyberPolicy(_)) => { + return AttemptClassification::Shared(shared); + } + shared => shared, + }; if let Some(classified) = cloudflare::CloudflareController::classify(observation) { return AttemptClassification::Cloudflare(classified); } if let Some(decision) = HistoryController::decide(observation) { return AttemptClassification::History(decision); } - if let Some(classified) = AccountFailureController::classify(observation) { - return AttemptClassification::AccountFailure(classified); - } - if let Some(classified) = QuotaController::classify(observation) { - return AttemptClassification::Quota(classified); + if let Some(shared) = shared { + return AttemptClassification::Shared(shared); } AttemptClassification::Default(Self::default_attempt_decision(observation)) } + /// complete、prefetch 与 committed stream 共用这一份 feature owner 优先级。 + fn classify_shared_failure<'a>( + observation: FailureObservation<'a>, + ) -> Option> { + let fact = observation.fact(); + if let Some(classified) = fact.and_then(CyberPolicyController::classify) { + return Some(SharedFailureClassification::CyberPolicy(classified)); + } + + let account_failure = match observation { + FailureObservation::Attempt(observation) => { + AccountFailureController::classify(observation) + } + FailureObservation::Stream { account_id, .. } => { + fact.and_then(|fact| AccountFailureController::classify_failure(account_id, fact)) + } + }; + if let Some(classified) = account_failure { + return Some(SharedFailureClassification::AccountFailure(classified)); + } + + let quota = match observation { + FailureObservation::Attempt(observation) => QuotaController::classify(observation), + FailureObservation::Stream { account_id, .. } => { + fact.and_then(|fact| QuotaController::classify_failure(account_id, fact)) + } + }; + quota.map(SharedFailureClassification::Quota) + } + fn attempt_decision( &self, observation: &AttemptObservation, classification: AttemptClassification<'_>, ) -> AttemptDecision { match classification { - AttemptClassification::CyberPolicy(failure) => { - let client_failure = CyberPolicyController::attempt_client_failure(failure); - if observation.routing.can_retry_next_candidate { - AttemptDecision::RetryNextCandidate { - exhaustion: None, - on_exhaustion: Some(client_failure), - } - } else { - AttemptDecision::Return(AttemptReturnKind::Failed(client_failure)) + AttemptClassification::Shared(shared) => match shared { + SharedFailureClassification::CyberPolicy(classified) => { + CyberPolicyController::decision(observation, classified) } - } + SharedFailureClassification::AccountFailure(classified) => { + let client_failure = response_failure(observation).map(|failure| { + AccountFailureController::client_failure(&classified, failure) + }); + with_response_client_failure( + AccountFailureController::decision(observation, classified), + client_failure, + ) + } + SharedFailureClassification::Quota(classified) => { + let client_failure = + response_failure(observation).map(QuotaController::client_failure); + with_response_client_failure( + QuotaController::decision(observation, classified), + client_failure, + ) + } + }, AttemptClassification::Cloudflare(classified) => { cloudflare::CloudflareController::decision(observation, classified) } @@ -807,22 +851,6 @@ impl ControllerSet { .map(|failure| ClientFailure::new(failure.clone(), 400, false)), ), AttemptClassification::Default(decision) => decision, - AttemptClassification::AccountFailure(classified) => { - let client_failure = response_failure(observation) - .map(|failure| AccountFailureController::client_failure(&classified, failure)); - with_response_client_failure( - AccountFailureController::decision(observation, classified), - client_failure, - ) - } - AttemptClassification::Quota(classified) => { - let client_failure = - response_failure(observation).map(QuotaController::client_failure); - with_response_client_failure( - QuotaController::decision(observation, classified), - client_failure, - ) - } } } diff --git a/backend/tests/dispatch/controllers/mod.rs b/backend/tests/dispatch/controllers/mod.rs index f0d9e238c..73e174dd5 100644 --- a/backend/tests/dispatch/controllers/mod.rs +++ b/backend/tests/dispatch/controllers/mod.rs @@ -42,6 +42,51 @@ fn cyber_policy_semantics_must_stay_inside_its_owner() { ); } +#[test] +fn cyber_policy_owner_should_construct_its_own_attempt_decision() { + let manifest = Path::new(env!("CARGO_MANIFEST_DIR")); + let composition = read(&manifest.join("src/dispatch/controllers/mod.rs")); + let cyber_policy = read(&manifest.join("src/dispatch/controllers/cyber_policy/mod.rs")); + let decision = between( + &composition, + "fn attempt_decision(", + "fn default_attempt_decision(", + ); + + assert!(cyber_policy.contains("pub(super) fn decision(")); + assert!(cyber_policy.contains("AttemptDecision::RetryNextCandidate")); + assert!(decision.contains("CyberPolicyController::decision(observation, classified)")); + assert!(!decision.contains("can_retry_next_candidate")); + assert!(!decision.contains("AttemptDecision::RetryNextCandidate")); +} + +#[test] +fn attempt_and_stream_should_share_one_feature_failure_owner_order() { + let manifest = Path::new(env!("CARGO_MANIFEST_DIR")); + let composition = read(&manifest.join("src/dispatch/controllers/mod.rs")); + let shared = between( + &composition, + "fn classify_shared_failure", + "fn attempt_decision(", + ); + + assert_eq!(composition.matches("fn classify_shared_failure").count(), 1); + assert!( + composition + .contains("Self::classify_shared_failure(FailureObservation::Attempt(observation))") + ); + assert!(composition.contains("Self::classify_shared_failure(FailureObservation::Stream {")); + assert_appears_in_order( + shared, + &[ + "fact.and_then(CyberPolicyController::classify)", + "AccountFailureController::classify(observation)", + "QuotaController::classify(observation)", + ], + ); + assert!(!composition.contains("self.cyber_policy.owns_failure")); +} + #[test] fn request_enter_io_should_be_parallel_and_bounded() { let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("src/dispatch/controllers"); @@ -272,6 +317,25 @@ fn collect_live_executor_dependencies(root: &Path, path: &Path, violations: &mut } } +fn between<'a>(source: &'a str, start: &str, end: &str) -> &'a str { + let start = source + .find(start) + .expect("source start marker should exist"); + let tail = &source[start..]; + let end = tail.find(end).expect("source end marker should exist"); + &tail[..end] +} + +fn assert_appears_in_order(source: &str, required: &[&str]) { + let mut cursor = 0; + for marker in required { + let offset = source[cursor..] + .find(marker) + .unwrap_or_else(|| panic!("ordered source marker should exist: {marker}")); + cursor += offset + marker.len(); + } +} + fn read(path: &Path) -> String { fs::read_to_string(path).unwrap_or_else(|error| panic!("read {}: {error}", path.display())) } diff --git a/docs/architecture.md b/docs/architecture.md index 9773db485..babae3d6f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -218,7 +218,7 @@ fleet/ `v1/*` 使用 Request、Attempt、Stream 三层洋葱生命周期。核心代码位于 `lifecycle/`、`controllers/`、`routing/`、`transport/` 和 `failure/`;`service.rs` 与 `stream/lifecycle.rs` 只是 complete/stream 入口适配器,`stream/live.rs` 是响应提交后的流执行器,不承载功能策略。 -架构硬规则是“单功能单归属”:功能阈值、状态机、优先级和 decision 只能由对应 controller 拥有;`lifecycle` 只管固定顺序、retry、commit、逆序退出和 exactly-once finalize。迁移不保留 compatibility re-export、feature flag 切换或新旧双路径,新路径接管后必须在同一次变更中删除旧入口和重复判断。 +架构硬规则是“单功能单归属”:功能阈值、状态机和 decision 只能由对应 controller 拥有;`ControllerSet` 只声明 controller 的固定优先级并分发其 typed classification、effect 与 decision,不能替 owner 构造功能专用 decision;`lifecycle` 只管 retry、commit、逆序退出和 exactly-once finalize。complete、prefetch 与 committed stream 必须共用一份 feature failure owner 顺序。迁移不保留 compatibility re-export、feature flag 切换或新旧双路径,新路径接管后必须在同一次变更中删除旧入口和重复判断。 ```text dispatch/ diff --git a/docs/request-lifecycle-controller-design.md b/docs/request-lifecycle-controller-design.md index a9b355b55..15ceb425b 100644 --- a/docs/request-lifecycle-controller-design.md +++ b/docs/request-lifecycle-controller-design.md @@ -44,14 +44,14 @@ 边界必须满足以下规则: -- `controller` 唯一拥有功能规则、状态机、优先级、阈值、typed facts、decision 和 scope;不能选择下一个账号、编码 HTTP/SSE/WebSocket 或直接调用其他 controller。 +- `controller` 唯一拥有本功能的规则、状态机、阈值、typed classification、decision 和 scope;不能选择具体的下一个账号、编码 HTTP/SSE/WebSocket 或直接调用其他 controller。跨 controller 的唯一固定优先级由 `ControllerSet` 声明。 - `lifecycle` 只拥有固定顺序、enter/leave、短路、逆序 unwind、retry 循环、commit 边界和 exactly-once finalize;不得出现 `if cyber_policy` 或功能阈值。 - `transport` / `upstream` 只把原始 HTTP、SSE、WebSocket 输入规范化为通用事实,例如状态码、错误码、terminal、usage;不得解释事实的业务后果。 - `api` 只解析请求,并编码 `ResponseDispatchResponse`、`ResponseDispatchStream` 或包含 `ClientFailure` 的 `ResponseDispatchError`;不得重新识别 cyber、quota、history 等功能。 - `store` 只提供原子读写、CAS、TTL 和数据映射;业务专用 store trait 可以由 controller 定义,基础设施实现不能拥有业务阈值和状态转换。 - controller 之间不得横向调用;共享内容只能是没有业务含义的协议事实和值对象,不能把共享业务规则藏进 `utils`、`common` 或 helper。 -允许的唯一挂载点是静态 composition root `ControllerSet`:它负责构造 controller、注入依赖,并通过类型化方法分发各阶段 hook。业务代码不在各个入口或尝试循环中手工挂载功能。 +允许的唯一挂载点是静态 composition root `ControllerSet`:它负责构造 controller、注入依赖、声明固定优先级,并通过类型化方法分发各阶段 hook。它只能调用 owner 暴露的分类、effect 与 decision,不能替 owner 构造功能专用 decision。业务代码不在各个入口或尝试循环中手工挂载功能。 ## 已实现目录与边界 @@ -218,7 +218,7 @@ controller 分为两类: 8. 按唯一优先级分类 history、session policy、账号隔离和 transport retry。 9. 产生 `AttemptDecision`,由唯一 pipeline 应用为 `AttemptApplyOutcome` 并记录 attempt trace。 -classifier 优先级只能存在一处。例如 `cyber_policy` 必须先于通用 403、quota 或 5xx 分类;其他模块消费分类后的类型,不能再次调用 `is_cyber_policy_*`。 +classifier 优先级只能存在一处。complete、prefetch 与 committed stream 共同经过同一个 feature failure owner classifier;`cyber_policy` 必须先于通用 403、quota 或 5xx 分类。composition root 只能把不透明的分类结果交还 owner 产生 decision,其他模块不能再次调用 `is_cyber_policy_*` 或自行拼装 Cyber retry。 ### Stream enter/leave @@ -248,6 +248,8 @@ classifier 优先级只能存在一处。例如 `cyber_policy` 必须先于通 - complete 与 stream 入口共享 `enter_request` 和 `AttemptRunner`,不存在各自维护的账号尝试主循环。 - `ControllerSet` 是静态 composition root;controller 顺序和 typed hook 在编译期可见,不使用动态 middleware 链。 +- complete、prefetch 与 committed stream 的公共失败 owner 优先级只在 `ControllerSet::classify_shared_failure` 声明一次;composition root 不实现 Cyber、quota 或账号失败规则。 +- Cyber failure 由 `CyberPolicyController` 返回不透明分类,并由该 owner 唯一构造换号或返回 decision;`ControllerSet` 只分发 effect 与 decision。 - `ControllerRequestScope` 只保存 cyber、history、usage 的真实状态;不存在通用 entered-controller marker、容量错误或空 unwind hook。 - complete 的 accepted response 归一为 `FinalOutcome` 后由 `finalize_complete` 消费 scope;retry 继续进入下一 attempt,rejection 在 controller effect 已完成后直接返回,不经过空 finalize。 - complete 的真实退出顺序固定为 Cloudflare → Usage → Affinity → Telemetry;live stream 固定为 Cloudflare → Usage → Affinity → Telemetry → Cyber,无副作用的阶段不制造空分支。