fix: align postgresql redis migration contracts

Co-authored-by: Codex <noreply@openai.com>
2026-11-07 00:05:49 +00:00 · 2026-07-10 19:35:58 +08:00
parent 45e4b36225
commit f06530278d
27 changed files with 501 additions and 563 deletions
@@ -20,8 +20,8 @@ domain_paths=(
  src/accounts src/models src/dispatch
 )

-deny "domain -> api/bootstrap" 'crate::(api|bootstrap)\b' "${domain_paths[@]}"
-deny "upstream -> domain" 'crate::(accounts|dispatch|telemetry|keys|auth|settings|update|models)\b' src/upstream
+deny "domain -> api/bootstrap" '(^|[^[:alnum:]_])(api|bootstrap)::' "${domain_paths[@]}"
+deny "upstream -> domain" '(^|[^[:alnum:]_])(accounts|dispatch|telemetry|keys|auth|settings|update|models)::' src/upstream
 deny "axum outside api/bootstrap" '\baxum(::|\b)' "${domain_paths[@]}"
 deny "sqlx in api" '\bsqlx(::|\b)' src/api
 deny "redis in api" '\bredis::' src/api
@@ -33,6 +33,38 @@ deny "persistence repository naming" '\b[A-Za-z0-9_]*repository[A-Za-z0-9_]*\b'
  src/upstream/openai/fingerprint
 deny "Arc alias" '\bArc[[:space:]]+as[[:space:]]+' src

+storage_command_check() {
+  local kind=$1
+  local pattern=$2
+  local file
+  while IFS= read -r file; do
+    case "$file" in
+      */store.rs | */store/*)
+        continue
+        ;;
+    esac
+    if [ "$kind" = sql ]; then
+      case "$file" in
+        src/infra/database.rs | src/bootstrap/import_sqlite.rs | src/bootstrap/import_sqlite/* | src/telemetry/rebuild.rs | */query.rs)
+          continue
+          ;;
+      esac
+    else
+      case "$file" in
+        src/infra/redis.rs | src/accounts/refresh/lease.rs)
+          continue
+          ;;
+      esac
+    fi
+    echo "$file"
+    echo "ARCH VIOLATION: $kind command outside an allowed storage adapter" >&2
+    fail=1
+  done < <(grep -rl --include='*.rs' -E "$pattern" src || true)
+}
+
+storage_command_check sql 'sqlx::(query|query_as|query_scalar|raw_sql)'
+storage_command_check redis 'redis::(cmd|pipe)|[.]query_async[[:space:]]*[(]|use redis::.*AsyncCommands'
+
 banned=$(find src \( -name 'importing.rs' -o -name 'exporting.rs' -o -name 'testing.rs' \
  -o -name 'util.rs' -o -name 'helper.rs' -o -name 'common.rs' -o -name 'misc.rs' \))
 if [ -n "$banned" ]; then
@@ -36,13 +36,13 @@ use crate::{
        },
    },
    telemetry::{
-        account_usage::query::AccountUsageTimeBucket,
+        account_usage::query::{AccountUsageSummary, AccountUsageTimeBucket},
        diagnostics::{
            fingerprint_diagnostics, AccountCapacityDiagnostics, AccountPoolDiagnostics,
        },
        usage::{
            query::UsageQueryFilter,
-            store::{UsageRecordAccountUsage, UsageRecordSummary},
+            store::UsageRecordAccountUsage,
            types::{metadata_string, UsageRecord},
        },
    },
@@ -292,12 +292,7 @@ pub(crate) async fn dashboard_summary(
    let capacity = state.services.account_pool.capacity_summary_now().await;
    let now = Utc::now();
    let today_filter = today_usage_record_filter(now);
-    let usage_summary = state
-        .services
-        .usage_records
-        .summary(UsageQueryFilter::default())
-        .await
-        .unwrap_or_else(|_| UsageRecordSummary::default());
+    let lifetime_usage = state.services.usage.summary().await.unwrap_or_default();
    let account_usage_records = state
        .services
        .usage_records
@@ -327,7 +322,7 @@ pub(crate) async fn dashboard_summary(
    let trend = dashboard_trend_data(&time_buckets, query.kind.unwrap_or_default());
    let quota_used_by_account =
        account_quota_used_percent_by_id(&state, &account_usage_records).await;
-    let settings = state.services.settings.current();
+    let settings = state.services.settings.current().await;
    let recent_usage_records = recent_events;
    let account_emails = state
        .services
@@ -341,7 +336,7 @@ pub(crate) async fn dashboard_summary(
    Ok(AdminResponse::new(
        StatusCode::OK,
        AdminEnvelope::ok(DashboardSummaryData {
-            cards: dashboard_cards(&accounts, &time_buckets, &usage_summary),
+            cards: dashboard_cards(&accounts, &time_buckets, &lifetime_usage),
            trend,
            health_timeline: dashboard_health_timeline_data(&time_buckets),
            account_usage: account_usage_data(
@@ -3,15 +3,15 @@ use super::*;
 pub(super) fn dashboard_cards(
    accounts: &[Account],
    buckets: &[AccountUsageTimeBucket],
-    usage_summary: &UsageRecordSummary,
+    lifetime_usage: &AccountUsageSummary,
 ) -> DashboardCardsData {
-    dashboard_cards_at(accounts, buckets, usage_summary, Utc::now())
+    dashboard_cards_at(accounts, buckets, lifetime_usage, Utc::now())
 }

 pub(super) fn dashboard_cards_at(
    accounts: &[Account],
    buckets: &[AccountUsageTimeBucket],
-    usage_summary: &UsageRecordSummary,
+    lifetime_usage: &AccountUsageSummary,
    now: DateTime<Utc>,
 ) -> DashboardCardsData {
    let today_start = china_day_start(now);
@@ -19,10 +19,10 @@ pub(super) fn dashboard_cards_at(
    let today = usage_window(buckets, today_start, now);
    let yesterday = usage_window(buckets, yesterday_start, today_start);
    let today_cost = cost_window(buckets, today_start, now).unwrap_or(0.0);
-    let total_requests = usage_summary.total_requests;
-    let total_tokens = usage_summary.total_tokens;
-    let total_cached_tokens = usage_summary.cached_tokens;
-    let total_hit_rate = summary_cache_hit_rate(usage_summary);
+    let total_requests = nonnegative_i64_to_u64(lifetime_usage.request_count);
+    let total_tokens = nonnegative_i64_to_u64(lifetime_usage.total_tokens);
+    let total_cached_tokens = nonnegative_i64_to_u64(lifetime_usage.cached_tokens);
+    let total_hit_rate = summary_cache_hit_rate(lifetime_usage);

    let total_accounts = accounts.len() as u64;
    let enabled_accounts = accounts
@@ -77,10 +77,10 @@ pub(super) fn dashboard_cards_at(
    }
 }

-pub(super) fn summary_cache_hit_rate(summary: &UsageRecordSummary) -> Option<f64> {
-    let input_tokens = summary.input_tokens;
+pub(super) fn summary_cache_hit_rate(summary: &AccountUsageSummary) -> Option<f64> {
+    let input_tokens = nonnegative_i64_to_u64(summary.input_tokens);
    if input_tokens > 0 {
-        Some(summary.cached_tokens as f64 / input_tokens as f64)
+        Some(nonnegative_i64_to_u64(summary.cached_tokens) as f64 / input_tokens as f64)
    } else {
        None
    }
@@ -52,7 +52,7 @@ pub(crate) async fn settings(
    State(state): State<AppState>,
    _auth: AdminAuth,
 ) -> Result<impl IntoResponse, AdminError> {
-    let config = state.services.settings.current();
+    let config = state.services.settings.current().await;
    Ok(AdminResponse::new(
        StatusCode::OK,
        AdminEnvelope::ok(AdminSettingsData::from_settings(&config)),
@@ -68,11 +68,9 @@ pub(crate) async fn update_settings(
    let patch = parse_settings_patch(&body)?;
    let config = state
        .services
-        .settings
-        .update(patch)
+        .update_settings(patch)
        .await
        .map_err(settings_service_error)?;
-    state.services.apply_settings(&config).await;

    Ok(AdminResponse::new(
        StatusCode::OK,
@@ -447,11 +447,9 @@ fn optional_datetime(

 fn log_error(error: &UsageQueryError) -> AdminError {
    match error {
-        UsageQueryError::List
-        | UsageQueryError::Get
-        | UsageQueryError::Clear
-        | UsageQueryError::Append
-        | UsageQueryError::Retention => AdminError::internal(error.to_string()),
+        UsageQueryError::List | UsageQueryError::Get | UsageQueryError::Clear => {
+            AdminError::internal(error.to_string())
+        }
        UsageQueryError::Accounts => AdminError::usage_record_accounts_failed(error.to_string()),
    }
 }
@@ -7,9 +7,10 @@ use axum::{
 };
 use serde_json::{json, Value};

-use crate::dispatch::{errors::DispatchFailureClass, responses::errors::ResponseDispatchError};
-
-use super::responses::response_failed_sse_event as encode_response_failed_sse_event;
+use crate::{
+    dispatch::{errors::DispatchFailureClass, responses::errors::ResponseDispatchError},
+    upstream::openai::protocol::sse::response_failed_sse_event as encode_response_failed_sse_event,
+};

 const NO_ACTIVE_UPSTREAM_ACCOUNT_MESSAGE: &str = "No active upstream account is available";
 const NO_AVAILABLE_RESPONSES_ACCOUNTS_MESSAGE: &str =
@@ -8,7 +8,7 @@ use axum::{
    response::{IntoResponse, Response},
    Extension, Json,
 };
-use serde_json::{json, Map, Value};
+use serde_json::{Map, Value};

 use crate::{
    api::middleware::request_id::{ClientIp, RequestId},
@@ -16,7 +16,7 @@ use crate::{
    dispatch::responses::{errors::ResponseDispatchError, service::ResponseDispatchStream},
    upstream::openai::protocol::{
        responses::{CodexCompactRequest, CodexResponsesRequest},
-        sse::{encode_sse_event, sse_body_has_done},
+        sse::sse_body_has_done,
    },
 };

@@ -191,41 +191,6 @@ impl ContextField {
    }
 }

-/// 编码 OpenAI Responses `response.failed` SSE 事件。
-pub fn response_failed_sse_event(error_type: &str, code: &str, message: &str) -> String {
-    response_failed_sse_event_with_id(None, error_type, code, message)
-}
-
-/// 使用指定 response id 编码 OpenAI Responses `response.failed` SSE 事件。
-pub fn response_failed_sse_event_with_id(
-    response_id: Option<&str>,
-    error_type: &str,
-    code: &str,
-    message: &str,
-) -> String {
-    let error = json!({
-        "type": error_type,
-        "code": code,
-        "message": message,
-    });
-    let response_id = response_id
-        .filter(|value| !value.trim().is_empty())
-        .map_or_else(
-            || format!("resp_proxy_{}", uuid::Uuid::new_v4().simple()),
-            ToString::to_string,
-        );
-    let data = json!({
-        "type": "response.failed",
-        "response": {
-            "id": response_id,
-            "status": "failed",
-            "error": error,
-        },
-        "error": error,
-    });
-    encode_sse_event("response.failed", &data.to_string())
-}
-
 // ====================================================================
 // HTTP 处理器
 // ====================================================================
@@ -112,7 +112,7 @@ async fn import_runtime_settings(
        let aliases: String = row.try_get("model_aliases_json")?;
        let admin_key = row
            .try_get::<Option<String>, _>("admin_api_key")?
-            .map(|key| hash_credential(&key));
+            .and_then(|key| (!key.trim().is_empty()).then(|| hash_credential(&key)));
        sqlx::query(
            "insert into runtime_settings
             (id, model_aliases_json, refresh_margin_seconds, refresh_concurrency,
@@ -29,13 +29,14 @@ use crate::{
        service::{ModelService, ModelServiceError},
        store::{ModelSnapshotStore, RedisModelSnapshotStore},
    },
-    settings::{service::RuntimeSettingsService, SettingsSnapshot},
+    settings::{service::RuntimeSettingsService, SettingsError, SettingsPatch, SettingsSnapshot},
    telemetry::{
        account_usage::query::AccountUsageQueryService,
        account_usage::store::PgAccountUsageStore,
        buckets::store::PgRequestBucketStore,
        ops::query::OpsQueryService,
        ops::store::PgOpsErrorLogStore,
+        recorder::Recorder,
        usage::query::UsageQueryService,
        usage::store::{PgUsageRecordStore, DEFAULT_USAGE_RECORD_CAPTURE_BODY},
    },
@@ -109,6 +110,7 @@ pub struct Services {
    pub installation_id: Option<String>,
    pub background_tasks: BackgroundTaskStores,
    account_pool_static: AccountPoolStaticSettings,
+    settings_update_lock: Arc<tokio::sync::Mutex<()>>,
 }

 #[derive(Debug, Clone)]
@@ -203,13 +205,12 @@ impl Services {
                Arc::new(client)
            }
        };
-        let usage_records = Arc::new(UsageQueryService::new(
+        let usage_records = Arc::new(UsageQueryService::new(stores.usage_records.clone()));
+        let ops_errors = Arc::new(OpsQueryService::new(stores.ops_errors.clone()));
+        let recorder = Arc::new(Recorder::new(
            stores.usage_records.clone(),
-            usage_record_options.enabled,
-            usage_record_options.capture_body,
-        ));
-        let ops_errors = Arc::new(OpsQueryService::new(
            stores.ops_errors.clone(),
+            usage_record_options.enabled,
            usage_record_options.capture_body,
        ));
        let account_pool_static = AccountPoolStaticSettings {
@@ -287,8 +288,7 @@ impl Services {
            models: models.clone(),
            codex: codex.clone(),
            session_affinity: session_affinity.clone(),
-            usage_records: usage_records.clone(),
-            ops_errors: ops_errors.clone(),
+            recorder,
            installation_id: installation_id.clone(),
            cloudflare: cloudflare_recovery,
        }));
@@ -317,6 +317,7 @@ impl Services {
            installation_id,
            background_tasks: stores,
            account_pool_static,
+            settings_update_lock: Arc::new(tokio::sync::Mutex::new(())),
        })
    }

@@ -326,8 +327,18 @@ impl Services {
        Ok(())
    }

-    /// 将已持久化的设置快照同步到各运行时消费者。
-    pub async fn apply_settings(&self, settings: &SettingsSnapshot) {
+    /// 持久化设置并同步到所有运行时消费者。
+    pub async fn update_settings(
+        &self,
+        patch: SettingsPatch,
+    ) -> Result<Arc<SettingsSnapshot>, SettingsError> {
+        let _guard = self.settings_update_lock.lock().await;
+        let settings = self.settings.update(patch).await?;
+        self.apply_settings(&settings).await;
+        Ok(settings)
+    }
+
+    async fn apply_settings(&self, settings: &SettingsSnapshot) {
        self.models.update_config(ModelConfig {
            model_aliases: settings.model_aliases.clone(),
        });
@@ -9,12 +9,11 @@ use crate::{
    },
    infra::time::elapsed_millis_i64,
    telemetry::{
-        ops::query::OpsQueryService,
        ops::types::OpsErrorLog,
        recorder::{
            enrich_event_route_metadata, enrich_usage_record_identity,
            event_kind as response_event_kind, reasoning_effort_from_request,
-            record_dispatch_error_event, DispatchErrorLogRecord,
+            record_dispatch_error_event, DispatchErrorLogRecord, Recorder,
        },
        usage::types::UsageRecordLevel,
    },
@@ -37,7 +36,7 @@ use super::{
 };

 pub(super) struct ResponseUpstreamErrorEventRecord<'a> {
-    pub(super) ops_errors: &'a OpsQueryService,
+    pub(super) recorder: &'a Recorder,
    pub(super) request_id: &'a str,
    pub(super) account_id: &'a str,
    pub(super) account_email: Option<&'a str>,
@@ -53,7 +52,7 @@ pub(super) struct ResponseUpstreamErrorEventRecord<'a> {
 }

 pub(super) struct ResponseStreamFailureEventRecord<'a> {
-    pub(super) ops_errors: &'a OpsQueryService,
+    pub(super) recorder: &'a Recorder,
    pub(super) request_id: &'a str,
    pub(super) account_id: &'a str,
    pub(super) route: &'a str,
@@ -71,7 +70,7 @@ pub(super) struct ResponseStreamFailureEventRecord<'a> {
 }

 pub(super) struct ResponseDispatchErrorEventRecord<'a> {
-    pub(super) ops_errors: &'a OpsQueryService,
+    pub(super) recorder: &'a Recorder,
    pub(super) request_id: &'a str,
    pub(super) client_api_key_id: Option<&'a str>,
    pub(super) account_id: Option<&'a str>,
@@ -111,7 +110,7 @@ pub(super) async fn record_response_dispatch_error_event(
        );
    }
    record_dispatch_error_event(DispatchErrorLogRecord {
-        ops_errors: record.ops_errors,
+        recorder: record.recorder,
        request_id: record.request_id,
        client_api_key_id: record.client_api_key_id,
        provider: Some("openai"),
@@ -177,7 +176,7 @@ pub(super) async fn record_response_upstream_error_event(
        record.request.service_tier(),
    );
    event.metadata = metadata;
-    if let Err(error) = record.ops_errors.record(event).await {
+    if let Err(error) = record.recorder.record_error(event).await {
        tracing::warn!(account_id = %record.account_id, error = %error, "failed to record upstream error event");
    }
 }
@@ -244,7 +243,7 @@ pub(super) async fn record_prefetched_response_stream_failure_event(
        }
    }
    event.metadata = metadata;
-    if let Err(error) = record.ops_errors.record(event).await {
+    if let Err(error) = record.recorder.record_error(event).await {
        tracing::warn!(account_id = %record.account_id, error = %error, "failed to record prefetched stream error event");
    }
 }
@@ -506,7 +505,7 @@ pub(super) async fn record_live_response_stream_event(
    if is_success_usage_event(level, status_code) {
        crate::telemetry::recorder::record_response_event(
            crate::telemetry::usage::types::ResponseUsageRecord {
-                usage_records: &context.usage_records,
+                recorder: &context.recorder,
                request_id: &context.request_id,
                client_api_key_id: context.request.client_api_key_id.as_deref(),
                account_id: &context.account_id,
@@ -536,7 +535,7 @@ pub(super) async fn record_live_response_stream_event(
        event.status_code = Some(status_code);
        event.latency_ms = Some(elapsed_millis_i64(context.started_at));
        event.metadata = metadata;
-        if let Err(error) = context.ops_errors.record(event).await {
+        if let Err(error) = context.recorder.record_error(event).await {
            tracing::warn!(account_id = %context.account_id, error = %error, "failed to record live response stream error event");
        }
    }
@@ -50,12 +50,10 @@ use crate::{
    },
    models::service::ModelService,
    telemetry::{
-        ops::query::OpsQueryService,
        recorder::{
            reasoning_effort_from_compact_request, reasoning_effort_from_request,
-            record_response_event,
+            record_response_event, Recorder,
        },
-        usage::query::UsageQueryService,
        usage::types::ResponseUsageRecord,
    },
    upstream::openai::{
@@ -107,8 +105,7 @@ pub struct ResponseDispatchService {
    codex: Arc<CodexBackendClient>,
    session_affinity: Arc<RuntimeSessionAffinityService>,
    reasoning_replay: Arc<Mutex<ReasoningReplayCache>>,
-    usage_records: Arc<UsageQueryService>,
-    ops_errors: Arc<OpsQueryService>,
+    recorder: Arc<Recorder>,
    installation_id: Option<String>,
    cloudflare: CloudflareRecovery,
 }
@@ -118,8 +115,7 @@ pub(crate) struct ResponseDispatchServiceParts {
    pub models: Arc<ModelService>,
    pub codex: Arc<CodexBackendClient>,
    pub session_affinity: Arc<RuntimeSessionAffinityService>,
-    pub usage_records: Arc<UsageQueryService>,
-    pub ops_errors: Arc<OpsQueryService>,
+    pub recorder: Arc<Recorder>,
    pub installation_id: Option<String>,
    pub cloudflare: CloudflareRecovery,
 }
@@ -175,8 +171,7 @@ impl ResponseDispatchService {
            reasoning_replay: Arc::new(Mutex::new(ReasoningReplayCache::new(Duration::seconds(
                DEFAULT_REASONING_REPLAY_TTL_SECS,
            )))),
-            usage_records: parts.usage_records,
-            ops_errors: parts.ops_errors,
+            recorder: parts.recorder,
            installation_id: parts.installation_id,
            cloudflare: parts.cloudflare,
        }
@@ -138,7 +138,7 @@ impl ResponseDispatchService {
                    insert_response_upstream_diagnostics(&mut metadata, &response.diagnostics);
                    insert_response_trace_metadata(&mut metadata, &trace, Some(&attempt));
                    record_response_event(ResponseUsageRecord {
-                        usage_records: &self.usage_records,
+                        recorder: &self.recorder,
                        request_id,
                        client_api_key_id: request.client_api_key_id.as_deref(),
                        account_id: &account.id,
@@ -273,7 +273,7 @@ impl ResponseDispatchService {
        error: &ResponseDispatchError,
    ) {
        record_response_dispatch_error_event(ResponseDispatchErrorEventRecord {
-            ops_errors: &self.ops_errors,
+            recorder: &self.recorder,
            request_id,
            client_api_key_id,
            account_id,
@@ -397,7 +397,7 @@ impl ResponseDispatchService {
                }
                Err(error) => {
                    record_response_upstream_error_event(ResponseUpstreamErrorEventRecord {
-                        ops_errors: &self.ops_errors,
+                        recorder: &self.recorder,
                        request_id,
                        account_id: &release_account_id,
                        account_email: account.email.as_deref(),
@@ -456,7 +456,7 @@ impl ResponseDispatchService {
                insert_response_trace_metadata(&mut metadata, &trace, Some(&attempt));
                insert_websocket_pool_decision(&mut metadata, response.websocket_pool_decision);
                record_response_event(ResponseUsageRecord {
-                    usage_records: &self.usage_records,
+                    recorder: &self.recorder,
                    request_id,
                    client_api_key_id: request.client_api_key_id.as_deref(),
                    account_id: &account.id,
@@ -546,7 +546,7 @@ impl ResponseDispatchService {
        error: &ResponseDispatchError,
    ) {
        record_response_dispatch_error_event(ResponseDispatchErrorEventRecord {
-            ops_errors: &self.ops_errors,
+            recorder: &self.recorder,
            request_id,
            client_api_key_id: details.client_api_key_id,
            account_id: details.account_id,
@@ -322,7 +322,7 @@ impl ResponseDispatchService {
                            if let ResponseDispatchError::Upstream(upstream_error) = &error {
                                record_response_upstream_error_event(
                                    ResponseUpstreamErrorEventRecord {
-                                        ops_errors: &self.ops_errors,
+                                        recorder: &self.recorder,
                                        request_id,
                                        account_id: &release_account_id,
                                        account_email: account.email.as_deref(),
@@ -422,7 +422,7 @@ impl ResponseDispatchService {
                        self.account_pool.release(&release_account_id).await;
                        record_prefetched_response_stream_failure_event(
                            ResponseStreamFailureEventRecord {
-                                ops_errors: &self.ops_errors,
+                                recorder: &self.recorder,
                                request_id,
                                account_id: &release_account_id,
                                route,
@@ -447,8 +447,7 @@ impl ResponseDispatchService {
                        account_pool: Arc::clone(&self.account_pool),
                        session_affinity: Arc::clone(&self.session_affinity),
                        reasoning_replay: Arc::clone(&self.reasoning_replay),
-                        usage_records: Arc::clone(&self.usage_records),
-                        ops_errors: Arc::clone(&self.ops_errors),
+                        recorder: Arc::clone(&self.recorder),
                        cloudflare: self.cloudflare.clone(),
                        account_id: account.id,
                        account_plan_type: account.plan_type,
@@ -580,7 +579,7 @@ impl ResponseDispatchService {
                Err(error) => {
                    self.account_pool.release(&release_account_id).await;
                    record_response_upstream_error_event(ResponseUpstreamErrorEventRecord {
-                        ops_errors: &self.ops_errors,
+                        recorder: &self.recorder,
                        request_id,
                        account_id: &release_account_id,
                        account_email: account.email.as_deref(),
@@ -5,7 +5,6 @@ use tokio::sync::Mutex;

 use crate::{
    accounts::pool::RuntimeAccountPoolService,
-    api::client::responses::response_failed_sse_event_with_id,
    dispatch::{
        affinity::{
            resolve::{evict_reasoning_replay, record_response_affinity},
@@ -15,15 +14,12 @@ use crate::{
        reasoning_replay::ReasoningReplayCache,
    },
    infra::time::elapsed_millis_i64,
-    telemetry::{
-        ops::query::OpsQueryService, usage::query::UsageQueryService,
-        usage::types::UsageRecordLevel,
-    },
+    telemetry::{recorder::Recorder, usage::types::UsageRecordLevel},
    upstream::openai::{
        protocol::{
            events::extract_sse_usage,
            responses::{response_from_codex_sse, CodexResponsesRequest, CollectedResponse},
-            sse::parse_sse_events,
+            sse::{parse_sse_events, response_failed_sse_event_with_id},
        },
        transport::{
            CodexBackendTransport, CodexRateLimitHeaderUpdates, CodexTurnStateUpdate,
@@ -50,8 +46,7 @@ pub(super) struct LiveResponseStreamContext {
    pub(super) account_pool: Arc<RuntimeAccountPoolService>,
    pub(super) session_affinity: Arc<RuntimeSessionAffinityService>,
    pub(super) reasoning_replay: Arc<Mutex<ReasoningReplayCache>>,
-    pub(super) usage_records: Arc<UsageQueryService>,
-    pub(super) ops_errors: Arc<OpsQueryService>,
+    pub(super) recorder: Arc<Recorder>,
    pub(super) cloudflare: CloudflareRecovery,
    pub(super) account_id: String,
    pub(super) account_plan_type: Option<String>,
@@ -1,9 +1,9 @@
-//! 运行时设置校验、持久化与快照广播。
+//! 运行时设置校验、持久化与当前快照。

 use std::{collections::BTreeMap, sync::Arc};

 use subtle::ConstantTimeEq;
-use tokio::sync::watch;
+use tokio::sync::RwLock;

 use crate::infra::identity::{generate_admin_api_key, hash_credential};

@@ -48,19 +48,18 @@ impl SettingsService {
    }
 }

-/// 持久化设置的当前快照与变更广播。
+/// 持久化设置的当前快照。
 #[derive(Clone)]
 pub struct RuntimeSettingsService {
    store: PgSettingsStore,
-    sender: watch::Sender<Arc<SettingsSnapshot>>,
+    snapshot: Arc<RwLock<Arc<SettingsSnapshot>>>,
 }

 impl RuntimeSettingsService {
    pub fn new(settings: SettingsSnapshot, pool: sqlx::PgPool) -> Self {
-        let (sender, _) = watch::channel(Arc::new(settings));
        Self {
            store: PgSettingsStore::new(pool),
-            sender,
+            snapshot: Arc::new(RwLock::new(Arc::new(settings))),
        }
    }

@@ -73,23 +72,20 @@ impl RuntimeSettingsService {
            .await
    }

-    pub fn current(&self) -> Arc<SettingsSnapshot> {
-        self.sender.borrow().clone()
-    }
-
-    pub fn subscribe(&self) -> watch::Receiver<Arc<SettingsSnapshot>> {
-        self.sender.subscribe()
+    pub async fn current(&self) -> Arc<SettingsSnapshot> {
+        self.snapshot.read().await.clone()
    }

    pub async fn update(
        &self,
        patch: SettingsPatch,
    ) -> Result<Arc<SettingsSnapshot>, SettingsError> {
-        let mut next = (*self.current()).clone();
+        let mut current = self.snapshot.write().await;
+        let mut next = (**current).clone();
        SettingsService::apply_patch(&mut next, patch)?;
        self.store.save(&next).await?;
        let next = Arc::new(next);
-        self.sender.send_replace(next.clone());
+        *current = next.clone();
        Ok(next)
    }

@@ -98,7 +94,8 @@ impl RuntimeSettingsService {
    }

    pub async fn regenerate_admin_api_key(&self) -> Result<String, SettingsError> {
-        self.store.ensure(&self.current()).await?;
+        let current = self.current().await;
+        self.store.ensure(&current).await?;
        let key = generate_admin_api_key();
        self.store
            .set_admin_api_key_hash(&hash_credential(&key))
@@ -151,7 +151,7 @@ pub struct AccountUsageRecord {
 }

 /// 管理端账号用量汇总。
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub struct AccountUsageSummary {
    pub account_count: i64,
    pub request_count: i64,
@@ -1,53 +1,30 @@
 //! 管理端运维错误事件服务。

-use chrono::Utc;
 use thiserror::Error;

 use crate::telemetry::{
    ops::store::{OpsErrorFilter, PgOpsErrorLogStore},
    ops::types::OpsErrorLog,
-    usage::types::{metadata_i64, metadata_service_tier, metadata_string},
 };

 /// 管理端运维错误事件错误。
 #[derive(Debug, Error)]
 pub enum OpsQueryError {
-    /// 写入失败。
-    #[error("failed to append ops error log")]
-    Append,
    /// 查询失败。
    #[error("failed to list ops error logs")]
    List,
-    /// 保留期清理失败。
-    #[error("failed to trim expired ops error logs")]
-    Retention,
 }

 /// 管理端运维错误事件服务。
 #[derive(Clone)]
 pub struct OpsQueryService {
    store: PgOpsErrorLogStore,
-    capture_body: bool,
 }

 impl OpsQueryService {
    /// 构造管理端运维错误事件服务。
-    pub fn new(store: PgOpsErrorLogStore, capture_body: bool) -> Self {
-        Self {
-            store,
-            capture_body,
-        }
-    }
-
-    /// 记录运维错误事件。
-    pub async fn record(&self, mut event: OpsErrorLog) -> Result<(), OpsQueryError> {
-        lift_error_fact_fields(&mut event);
-        apply_capture_body_policy(&mut event, self.capture_body);
-        self.store
-            .append(&event)
-            .await
-            .map_err(|_| OpsQueryError::Append)?;
-        Ok(())
+    pub fn new(store: PgOpsErrorLogStore) -> Self {
+        Self { store }
    }

    pub async fn list(
@@ -73,97 +50,4 @@ impl OpsQueryService {
            .await
            .map_err(|_| OpsQueryError::List)
    }
-
-    /// 周期清理超过保留期的错误事实。
-    pub async fn trim_to_retention(
-        &self,
-        now: chrono::DateTime<Utc>,
-    ) -> Result<u64, OpsQueryError> {
-        self.store
-            .trim_to_retention(now)
-            .await
-            .map_err(|_| OpsQueryError::Retention)
-    }
-}
-
-fn lift_error_fact_fields(event: &mut OpsErrorLog) {
-    event.client_status_code = event
-        .client_status_code
-        .or_else(|| metadata_i64(&event.metadata, &["clientStatusCode", "client_status_code"]));
-    event.upstream_status_code = event
-        .upstream_status_code
-        .or_else(|| metadata_i64(&event.metadata, &["upstreamStatusCode", "upstreamStatus"]));
-    event.transport = event
-        .transport
-        .take()
-        .or_else(|| metadata_string(&event.metadata, &["transport"]));
-    event.attempt_index = event
-        .attempt_index
-        .or_else(|| metadata_i64(&event.metadata, &["attemptIndex", "attempt_index"]));
-    event.failure_class = event
-        .failure_class
-        .take()
-        .or_else(|| metadata_string(&event.metadata, &["failureClass", "failure_class"]));
-    event.response_id = event
-        .response_id
-        .take()
-        .or_else(|| metadata_string(&event.metadata, &["responseId", "response_id"]));
-    event.upstream_request_id = event.upstream_request_id.take().or_else(|| {
-        metadata_string(
-            &event.metadata,
-            &[
-                "upstreamRequestId",
-                "upstream_request_id",
-                "openaiRequestId",
-            ],
-        )
-    });
-    event.service_tier = event
-        .service_tier
-        .take()
-        .or_else(|| metadata_service_tier(&event.metadata).map(ToString::to_string));
-
-    let Some(metadata) = event.metadata.as_object_mut() else {
-        return;
-    };
-    for key in [
-        "clientStatusCode",
-        "client_status_code",
-        "upstreamStatusCode",
-        "upstreamStatus",
-        "transport",
-        "attemptIndex",
-        "attempt_index",
-        "failureClass",
-        "failure_class",
-        "responseId",
-        "response_id",
-        "upstreamRequestId",
-        "upstream_request_id",
-        "openaiRequestId",
-        "serviceTier",
-    ] {
-        metadata.remove(key);
-    }
-}
-
-fn apply_capture_body_policy(event: &mut OpsErrorLog, capture_body: bool) {
-    let Some(metadata) = event.metadata.as_object_mut() else {
-        return;
-    };
-    for key in [
-        "body",
-        "rawBody",
-        "requestBody",
-        "responseBody",
-        "upstreamBody",
-    ] {
-        if capture_body {
-            if let Some(serde_json::Value::String(value)) = metadata.get_mut(key) {
-                value.truncate(4096);
-            }
-        } else {
-            metadata.remove(key);
-        }
-    }
 }
@@ -7,15 +7,86 @@ use serde_json::Value;
 use crate::{
    infra::time::elapsed_millis_i64,
    telemetry::{
-        ops::query::OpsQueryService,
+        ops::store::{PgOpsErrorLogStore, PgOpsErrorLogStoreError},
        ops::types::OpsErrorLog,
-        usage::types::{ResponseUsageRecord, UsageRecord},
+        usage::{
+            store::{PgUsageRecordStore, PgUsageRecordStoreError},
+            types::{
+                metadata_i64, metadata_service_tier, metadata_string as metadata_string_any,
+                ResponseUsageRecord, UsageRecord,
+            },
+        },
    },
    upstream::openai::protocol::responses::{CodexCompactRequest, CodexResponsesRequest},
 };

+/// 成功与失败事实的唯一写入入口。
+#[derive(Clone)]
+pub struct Recorder {
+    usage_records: PgUsageRecordStore,
+    ops_errors: PgOpsErrorLogStore,
+    usage_enabled: bool,
+    capture_body: bool,
+}
+
+/// 遥测事实写入错误。
+#[derive(Debug, thiserror::Error)]
+pub enum RecorderError {
+    #[error("invalid success usage fact")]
+    InvalidUsageFact,
+    #[error(transparent)]
+    Usage(#[from] PgUsageRecordStoreError),
+    #[error(transparent)]
+    Ops(#[from] PgOpsErrorLogStoreError),
+}
+
+impl Recorder {
+    pub fn new(
+        usage_records: PgUsageRecordStore,
+        ops_errors: PgOpsErrorLogStore,
+        usage_enabled: bool,
+        capture_body: bool,
+    ) -> Self {
+        Self {
+            usage_records,
+            ops_errors,
+            usage_enabled,
+            capture_body,
+        }
+    }
+
+    /// 校验并写入一条成功事实及其时间桶。
+    pub async fn record_usage(&self, mut event: UsageRecord) -> Result<(), RecorderError> {
+        if !self.usage_enabled {
+            return Ok(());
+        }
+        if !is_usage_fact(&event) {
+            tracing::warn!(
+                usage_record_id = %event.id,
+                request_id = event.request_id.as_deref().unwrap_or(""),
+                account_id = event.account_id,
+                model = event.model,
+                status_code = event.status_code,
+                "rejected invalid success usage fact"
+            );
+            return Err(RecorderError::InvalidUsageFact);
+        }
+        apply_success_capture_body_policy(&mut event, self.capture_body);
+        self.usage_records.append(&event).await?;
+        Ok(())
+    }
+
+    /// 规范化并写入一条失败事实及其时间桶。
+    pub async fn record_error(&self, mut event: OpsErrorLog) -> Result<(), RecorderError> {
+        lift_error_fact_fields(&mut event);
+        apply_error_capture_body_policy(&mut event, self.capture_body);
+        self.ops_errors.append(&event).await?;
+        Ok(())
+    }
+}
+
 pub(crate) struct DispatchErrorLogRecord<'a> {
-    pub ops_errors: &'a OpsQueryService,
+    pub recorder: &'a Recorder,
    pub request_id: &'a str,
    pub client_api_key_id: Option<&'a str>,
    pub provider: Option<&'a str>,
@@ -43,7 +114,7 @@ pub(crate) async fn record_dispatch_error_event(record: DispatchErrorLogRecord<'
    event.latency_ms = Some(elapsed_millis_i64(record.started_at));
    event.metadata = metadata;

-    if let Err(error) = record.ops_errors.record(event).await {
+    if let Err(error) = record.recorder.record_error(event).await {
        tracing::warn!(
            account_id = record.account_id.unwrap_or(""),
            error = %error,
@@ -91,11 +162,120 @@ pub(crate) async fn record_response_event(record: ResponseUsageRecord<'_>) {
    lift_success_fact_fields(&mut event, &mut metadata);
    event.metadata = metadata;

-    if let Err(error) = record.usage_records.record(event).await {
+    if let Err(error) = record.recorder.record_usage(event).await {
        tracing::warn!(account_id = %record.account_id, error = %error, "failed to record response event");
    }
 }

+fn is_usage_fact(event: &UsageRecord) -> bool {
+    (200..=399).contains(&event.status_code)
+        && !event.provider.trim().is_empty()
+        && !event.account_id.trim().is_empty()
+        && !event.model.trim().is_empty()
+}
+
+fn apply_success_capture_body_policy(event: &mut UsageRecord, capture_body: bool) {
+    if capture_body {
+        return;
+    }
+    remove_body_fields(&mut event.metadata);
+}
+
+fn apply_error_capture_body_policy(event: &mut OpsErrorLog, capture_body: bool) {
+    let Some(metadata) = event.metadata.as_object_mut() else {
+        return;
+    };
+    for key in body_fields() {
+        if capture_body {
+            if let Some(Value::String(value)) = metadata.get_mut(key) {
+                value.truncate(4096);
+            }
+        } else {
+            metadata.remove(key);
+        }
+    }
+}
+
+fn remove_body_fields(metadata: &mut Value) {
+    let Some(metadata) = metadata.as_object_mut() else {
+        return;
+    };
+    for key in body_fields() {
+        metadata.remove(key);
+    }
+}
+
+fn body_fields() -> [&'static str; 5] {
+    [
+        "body",
+        "rawBody",
+        "requestBody",
+        "responseBody",
+        "upstreamBody",
+    ]
+}
+
+fn lift_error_fact_fields(event: &mut OpsErrorLog) {
+    event.client_status_code = event
+        .client_status_code
+        .or_else(|| metadata_i64(&event.metadata, &["clientStatusCode", "client_status_code"]));
+    event.upstream_status_code = event
+        .upstream_status_code
+        .or_else(|| metadata_i64(&event.metadata, &["upstreamStatusCode", "upstreamStatus"]));
+    event.transport = event
+        .transport
+        .take()
+        .or_else(|| metadata_string_any(&event.metadata, &["transport"]));
+    event.attempt_index = event
+        .attempt_index
+        .or_else(|| metadata_i64(&event.metadata, &["attemptIndex", "attempt_index"]));
+    event.failure_class = event
+        .failure_class
+        .take()
+        .or_else(|| metadata_string_any(&event.metadata, &["failureClass", "failure_class"]));
+    event.response_id = event
+        .response_id
+        .take()
+        .or_else(|| metadata_string_any(&event.metadata, &["responseId", "response_id"]));
+    event.upstream_request_id = event.upstream_request_id.take().or_else(|| {
+        metadata_string_any(
+            &event.metadata,
+            &[
+                "upstreamRequestId",
+                "upstream_request_id",
+                "openaiRequestId",
+            ],
+        )
+    });
+    event.service_tier = event
+        .service_tier
+        .take()
+        .or_else(|| metadata_service_tier(&event.metadata).map(ToString::to_string));
+
+    let Some(metadata) = event.metadata.as_object_mut() else {
+        return;
+    };
+    for key in [
+        "clientStatusCode",
+        "client_status_code",
+        "upstreamStatusCode",
+        "upstreamStatus",
+        "transport",
+        "attemptIndex",
+        "attempt_index",
+        "failureClass",
+        "failure_class",
+        "responseId",
+        "response_id",
+        "upstreamRequestId",
+        "upstream_request_id",
+        "openaiRequestId",
+        "serviceTier",
+    ] {
+        metadata.remove(key);
+    }
+}
+
 fn lift_success_fact_fields(event: &mut UsageRecord, metadata: &mut Value) {
    event.transport = metadata_string(metadata, "transport");
    event.attempt_index = metadata_nonnegative_i64(metadata, "attemptIndex");
@@ -1,6 +1,6 @@
 //! 管理端使用记录服务。

-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;

 use chrono::{DateTime, Utc};
 use thiserror::Error;
@@ -110,12 +110,6 @@ pub enum UsageQueryError {
    /// 清空失败。
    #[error("failed to clear usage records")]
    Clear,
-    /// 写入失败。
-    #[error("failed to append usage record")]
-    Append,
-    /// 保留期清理失败。
-    #[error("failed to trim expired usage records")]
-    Retention,
    /// 账号关联信息读取失败。
    #[error("failed to load usage record accounts")]
    Accounts,
@@ -125,25 +119,12 @@ pub enum UsageQueryError {
 #[derive(Clone)]
 pub struct UsageQueryService {
    store: PgUsageRecordStore,
-    settings: Arc<tokio::sync::RwLock<UsageQuerySettings>>,
-}
-
-#[derive(Debug, Clone, Copy)]
-struct UsageQuerySettings {
-    enabled: bool,
-    capture_body: bool,
 }

 impl UsageQueryService {
    /// 构造管理端使用记录服务。
-    pub fn new(store: PgUsageRecordStore, enabled: bool, capture_body: bool) -> Self {
-        Self {
-            store,
-            settings: Arc::new(tokio::sync::RwLock::new(UsageQuerySettings {
-                enabled,
-                capture_body,
-            })),
-        }
+    pub fn new(store: PgUsageRecordStore) -> Self {
+        Self { store }
    }

    /// 分页查询日志。
@@ -254,72 +235,6 @@ impl UsageQueryService {
            .map(|cleared| ClearUsageRecords { cleared })
            .map_err(|_| UsageQueryError::Clear)
    }
-
-    /// 记录使用记录。
-    pub async fn record(&self, event: UsageRecord) -> Result<(), UsageQueryError> {
-        let settings = *self.settings.read().await;
-        if !settings.enabled {
-            return Ok(());
-        }
-        if !is_usage_log_event(&event) {
-            tracing::warn!(
-                usage_record_id = %event.id,
-                request_id = event.request_id.as_deref().unwrap_or(""),
-                account_id = event.account_id,
-                model = event.model,
-                status_code = event.status_code,
-                "rejected invalid success usage fact"
-            );
-            return Err(UsageQueryError::Append);
-        }
-        self.append_with_settings(event, settings).await
-    }
-
-    async fn append_with_settings(
-        &self,
-        mut event: UsageRecord,
-        settings: UsageQuerySettings,
-    ) -> Result<(), UsageQueryError> {
-        apply_capture_body_policy(&mut event, settings.capture_body);
-        self.store
-            .append(&event)
-            .await
-            .map_err(|_| UsageQueryError::Append)?;
-        Ok(())
-    }
-
-    /// 周期清理超过保留期的成功事实。
-    pub async fn trim_to_retention(&self, now: DateTime<Utc>) -> Result<u64, UsageQueryError> {
-        self.store
-            .trim_to_retention(now)
-            .await
-            .map_err(|_| UsageQueryError::Retention)
-    }
-}
-
-fn is_usage_log_event(event: &UsageRecord) -> bool {
-    (200..=399).contains(&event.status_code)
-        && !event.provider.trim().is_empty()
-        && !event.account_id.trim().is_empty()
-        && !event.model.trim().is_empty()
-}
-
-fn apply_capture_body_policy(event: &mut UsageRecord, capture_body: bool) {
-    if capture_body {
-        return;
-    }
-    let Some(metadata) = event.metadata.as_object_mut() else {
-        return;
-    };
-    for key in [
-        "body",
-        "rawBody",
-        "requestBody",
-        "responseBody",
-        "upstreamBody",
-    ] {
-        metadata.remove(key);
-    }
 }

 pub(crate) fn usage_record_cost_details(
@@ -141,7 +141,7 @@ pub(crate) fn metadata_i64(metadata: &Value, keys: &[&str]) -> Option<i64> {

 /// 响应事件记录（供 dispatch 模块使用）。
 pub struct ResponseUsageRecord<'a> {
-    pub usage_records: &'a crate::telemetry::usage::query::UsageQueryService,
+    pub recorder: &'a crate::telemetry::recorder::Recorder,
    pub request_id: &'a str,
    pub client_api_key_id: Option<&'a str>,
    pub account_id: &'a str,
@@ -1,5 +1,6 @@
 //! SSE 事件解析与编码。

+use serde_json::json;
 use thiserror::Error;

 /// 单条 SSE 事件。
@@ -21,6 +22,41 @@ pub const MAX_SSE_EVENT_BUFFER_BYTES: usize = 64 * 1024 * 1024;
 /// SSE 流结束标记帧。
 pub const DONE_SSE_FRAME: &str = "data: [DONE]\n\n";

+/// 编码 OpenAI Responses `response.failed` SSE 事件。
+pub fn response_failed_sse_event(error_type: &str, code: &str, message: &str) -> String {
+    response_failed_sse_event_with_id(None, error_type, code, message)
+}
+
+/// 使用指定 response id 编码 OpenAI Responses `response.failed` SSE 事件。
+pub fn response_failed_sse_event_with_id(
+    response_id: Option<&str>,
+    error_type: &str,
+    code: &str,
+    message: &str,
+) -> String {
+    let error = json!({
+        "type": error_type,
+        "code": code,
+        "message": message,
+    });
+    let response_id = response_id
+        .filter(|value| !value.trim().is_empty())
+        .map_or_else(
+            || format!("resp_proxy_{}", uuid::Uuid::new_v4().simple()),
+            ToString::to_string,
+        );
+    let data = json!({
+        "type": "response.failed",
+        "response": {
+            "id": response_id,
+            "status": "failed",
+            "error": error,
+        },
+        "error": error,
+    });
+    encode_sse_event("response.failed", &data.to_string())
+}
+
 /// 判断 SSE 文本是否已经包含结束标记。
 pub fn sse_body_has_done(body: &str) -> bool {
    body.trim_end_matches(['\r', '\n'])
@@ -54,7 +54,7 @@ async fn dashboard_summary_should_render_dash_when_fingerprint_updated_at_is_mis
 #[tokio::test]
 async fn dashboard_summary_should_calculate_today_traffic_and_average_first_token_latency_from_time_buckets(
 ) {
-    let (app, store, _pool, _dir) = dashboard_test_app(
+    let (app, store, pool, _dir) = dashboard_test_app(
        "dashboard-traffic-latency",
        crate::support::fingerprint::test_fingerprint(),
    )
@@ -80,22 +80,23 @@ async fn dashboard_summary_should_calculate_today_traffic_and_average_first_toke
        .append(&usage_record_with_tokens(yesterday_record_at, 30))
        .await
        .unwrap();
+    seed_lifetime_usage_summary(&pool, 7, 400, 500, 100).await;

    let body = dashboard_summary(app).await;

    assert_eq!(body["data"]["cards"]["traffic"]["todayRequests"], "2");
    assert_eq!(body["data"]["cards"]["traffic"]["todayRequestsValue"], 2);
-    assert_eq!(body["data"]["cards"]["traffic"]["totalRequests"], "3");
+    assert_eq!(body["data"]["cards"]["traffic"]["totalRequests"], "7");
    assert_eq!(
        body["data"]["cards"]["traffic"]["yesterdayRequestsValue"],
        1
    );
    assert_eq!(body["data"]["cards"]["tokens"]["todayTokens"], "30");
    assert_eq!(body["data"]["cards"]["tokens"]["todayTokensValue"], 30);
-    assert_eq!(body["data"]["cards"]["tokens"]["totalTokens"], "60");
+    assert_eq!(body["data"]["cards"]["tokens"]["totalTokens"], "900");
    assert_eq!(body["data"]["cards"]["tokens"]["yesterdayTokensValue"], 30);
-    assert_eq!(body["data"]["cards"]["cache"]["totalHitRate"], "0.0%");
-    assert_eq!(body["data"]["cards"]["cache"]["totalCachedTokens"], "0");
+    assert_eq!(body["data"]["cards"]["cache"]["totalHitRate"], "25.0%");
+    assert_eq!(body["data"]["cards"]["cache"]["totalCachedTokens"], "100");
    assert_eq!(
        body["data"]["cards"]["cache"]["averageFirstTokenLatencyMs"],
        "300 ms"
@@ -700,6 +701,30 @@ async fn seed_usage_summary(pool: &PgPool, input_tokens: u64, output_tokens: u64
        .await;
 }

+async fn seed_lifetime_usage_summary(
+    pool: &PgPool,
+    request_count: u64,
+    input_tokens: u64,
+    output_tokens: u64,
+    cached_tokens: u64,
+) {
+    seed_dashboard_account(pool, "acct_lifetime", "lifetime@example.com").await;
+    sqlx::query(
+        "insert into account_usage
+         (account_id, request_count, input_tokens, output_tokens, cached_tokens, total_tokens)
+         values ($1, $2, $3, $4, $5, $6)",
+    )
+    .bind("acct_lifetime")
+    .bind(request_count as i64)
+    .bind(input_tokens as i64)
+    .bind(output_tokens as i64)
+    .bind(cached_tokens as i64)
+    .bind((input_tokens + output_tokens) as i64)
+    .execute(pool)
+    .await
+    .unwrap();
+}
+
 async fn seed_usage_summary_with_last_used(
    pool: &PgPool,
    input_tokens: u64,
@@ -152,6 +152,30 @@ async fn import_sqlite_should_reject_nonempty_target() {
    assert!(matches!(error, ImportSqliteError::TargetNotEmpty));
 }

+#[tokio::test]
+async fn import_sqlite_should_keep_blank_admin_api_key_disabled() {
+    let (_source_guard, source_path) = sqlite_v3_source("blank-admin-api-key").await;
+    let source = open_sqlite(&source_path, false).await;
+    sqlx::query(
+        "insert into runtime_settings values
+         (1, '{}', 3600, 2, 3, 50, 'smart', '   ', '2026-01-01T00:00:00Z')",
+    )
+    .execute(&source)
+    .await
+    .unwrap();
+    source.close().await;
+    let (target, _target_guard) = init_test_db("import-sqlite-blank-admin-api-key").await;
+
+    import_sqlite(&target, &source_path).await.unwrap();
+
+    let admin_api_key_hash: Option<String> =
+        sqlx::query_scalar("select admin_api_key_hash from runtime_settings where id = 1")
+            .fetch_one(&target)
+            .await
+            .unwrap();
+    assert!(admin_api_key_hash.is_none());
+}
+
 async fn sqlite_v3_source(label: &str) -> (tempfile::TempDir, PathBuf) {
    let guard = tempfile::tempdir().unwrap();
    let path = guard.path().join(format!("{label}.sqlite"));
@@ -1,7 +1,5 @@
 use axum::http::HeaderMap;
-use codex_proxy_rs::api::client::responses::{
-    build_codex_request, build_compact_request, response_failed_sse_event,
-};
+use codex_proxy_rs::api::client::responses::{build_codex_request, build_compact_request};
 use codex_proxy_rs::upstream::openai::protocol::events::{
    extract_sse_usage, extract_usage, parse_rate_limit_headers, parse_rate_limits_event,
    rate_limit_quota, retry_after_seconds_from_body, RateLimitWindow, TokenUsage,
@@ -11,7 +9,9 @@ use codex_proxy_rs::upstream::openai::protocol::responses::{
    response_sse_event_is_terminal, transport_for_request, CodexResponsesRequest, CodexTransport,
    CollectedResponse,
 };
-use codex_proxy_rs::upstream::openai::protocol::sse::parse_sse_events;
+use codex_proxy_rs::upstream::openai::protocol::sse::{
+    parse_sse_events, response_failed_sse_event,
+};
 use codex_proxy_rs::upstream::openai::protocol::websocket::{
    classify_websocket_error_frame, is_terminal_websocket_event,
    retry_after_seconds_from_wrapped_error_headers,
@@ -39,16 +39,16 @@
 | --- | --- | --- | --- |
 | L0 | `infra` | PostgreSQL 连接/迁移、Redis 连接/键前缀、时间、JSON、日志、身份生成、路径、格式化 | 出现任何业务概念 |
 | L1 | `upstream` | 纯上游客户端：协议编解码、HTTP/SSE/WebSocket 传输、指纹、OAuth token client。**每家 provider 一个子目录**（`openai/`，规划中 `cloudflare/`） | 依赖任何领域模块；出现账号池/调度/遥测概念 |
-| L1 | `telemetry` | 遥测域：usage_records / ops_error_logs / request_time_buckets / account_usage 的**写入与查询**、billing、dashboard 聚合、rebuild | 被冠以 admin 语义；HTTP handler |
-| L1 | `keys` | 客户端 API key 域：store、内存缓存、鉴权校验、管理操作 | — |
+| L1 | `telemetry` | 遥测域：usage_records / ops_error_logs / request_time_buckets / account_usage 的**写入与查询**、billing、rebuild | 被冠以 admin 语义；HTTP handler |
+| L1 | `keys` | 客户端 API key 域：store、鉴权校验、管理操作 | — |
 | L1 | `auth` | 管理端认证域：admin 用户（PG）、会话（Redis，token_hash 键，database.md §4B.1） | 客户端 key 概念（那是 `keys`） |
-| L1 | `settings` | 运行时设置域：持久化、快照、变更广播（watch）、admin API key 校验 | 直接调用上层服务推送变更（§2.3） |
+| L1 | `settings` | 运行时设置域：校验、持久化、并发串行化的当前快照、admin API key 校验 | 持有账号池、模型或刷新服务引用（§2.3） |
 | L1 | `update` | 自更新域：release 查询、下载、校验、解压、替换、状态 | — |
 | L2 | `accounts` | 账号域：实体、池、调度器、store、quota、token 刷新、Cookie、导入导出、管理操作 | — |
 | L2 | `models` | 模型目录域：catalog、别名、plan 快照 | — |
 | L3 | `dispatch` | 请求编排域：账号获取、affinity、上游调用、流转发、失败恢复、重试、遥测记录 | HTTP 类型（axum）出现在签名 |
 | L4 | `api` | 全部入站 HTTP：`api/client`（/v1）、`api/admin`（/api/admin）、路由组合、中间件、SPA 资源 | 写 SQL；承载领域逻辑 |
-| L5 | `bootstrap` | 进程装配：启动配置、服务构造、AppState、后台任务、关闭 | 被任何层依赖 |
+| 装配层 | `bootstrap` | 进程装配：启动配置、服务构造、AppState、后台任务、关闭；`api` 只读取其 AppState 门面 | 被 L0–L3 领域层依赖 |

 ### 2.2 依赖方向

@@ -57,8 +57,6 @@
 | 允许的同层/特殊依赖 | 理由 |
 | --- | --- |
 | `accounts` → `models` | 候选过滤需要模型允许列表；单向 |
-| `accounts` → `settings` | 池参数/刷新策略订阅设置快照（watch 类型）；单向 |
-| `auth` → `settings` | admin API key 哈希存放在 runtime_settings 表，经 settings 的 store 方法读取；单向 |

 **全库硬禁**（CI 脚本断言，§9）：

@@ -70,7 +68,7 @@

 ### 2.3 控制反转两处（消除向上依赖）

-1. **设置传播**：现状 `RuntimeSettingsService` 持有账号池与刷新策略的引用、主动推送更新（低层持有高层引用）。终态：`settings` 只暴露 `tokio::sync::watch::Receiver<SettingsSnapshot>`；`accounts`（池、刷新策略）、`models`（别名）各自订阅。装配在 `bootstrap` 完成。
+1. **设置传播**：`RuntimeSettingsService` 只负责校验、持久化和当前快照，不持有任何上层消费者。`bootstrap::Services::update_settings` 在持久化成功后同步把同一快照应用到账号池、刷新策略与模型配置；更新串行执行，不保留无人消费的 watch 通道，也不存在低层向上回调。
 2. **遥测写入**：现状 proxy 热路径 import `admin::monitoring` 的 Admin* 服务。终态：`dispatch` 依赖 `telemetry::Recorder`（L3 → L1，方向正确），admin 侧只是同一域的查询消费者。

 ---
@@ -79,117 +77,58 @@

 ```text
 backend/src/
-├── main.rs                       # 子命令入口：serve（默认）/ import-sqlite / rebuild-buckets
-├── lib.rs
-├── infra/                        # L0
-│   ├── mod.rs
-│   ├── database.rs               # PgPool 连接/池参数、迁移框架（PG 谱系 0001 终态基线）、ping
-│   ├── redis.rs                  # Redis ConnectionManager、键前缀、ping（database.md §1）
-│   ├── migrations/               # *.sql（PG）
-│   ├── identity.rs  time.rs  json.rs  format.rs  logging.rs  paths.rs
-├── upstream/                     # L1 纯上游客户端（每家 provider 一个子目录）
-│   ├── mod.rs                    # 第二家 provider 出现时，UpstreamClient trait 落此
-│   └── openai/
-│       ├── mod.rs
-│       ├── token_client.rs
-│       ├── protocol/
-│       │   ├── mod.rs  schema.rs  responses.rs  events.rs  sse.rs  websocket.rs
-│       ├── transport/
-│       │   ├── mod.rs  client.rs  endpoints.rs  headers.rs  tls.rs
-│       │   ├── response_meta.rs  usage.rs  diagnostics.rs
-│       │   ├── websocket.rs  websocket_pool.rs  websocket_pump.rs
-│       └── fingerprint/
-│           ├── mod.rs  types.rs  store.rs  runtime.rs  updater.rs
-├── telemetry/                    # L1 遥测域（原 admin/monitoring + proxy/dispatch/usage_events）
-│   ├── mod.rs
-│   ├── recorder.rs               # 唯一写入口：成功/失败两条 PG 事务（database.md §5.2）
-│   ├── billing.rs
-│   ├── rebuild.rs                # rebuild-buckets（database.md §7）
-│   ├── dashboard.rs              # Dashboard 聚合查询（无 HTTP；口径 database.md §5.3）
-│   ├── usage/                    # usage_records
-│   │   ├── mod.rs  types.rs  store.rs  query.rs
-│   ├── ops/                      # ops_error_logs（含 /api/admin/ops/errors 的查询面）
-│   │   ├── mod.rs  types.rs  store.rs  query.rs
-│   ├── buckets/                  # request_time_buckets
-│   │   ├── mod.rs  store.rs  query.rs
-│   └── account_usage/            # account_usage / account_model_usage
-│       ├── mod.rs  store.rs  query.rs
-├── keys/                         # L1 客户端 key 域（原 admin/keys/service.rs 拆分）
-│   ├── mod.rs  types.rs  store.rs  service.rs  manage.rs
-│   │                             # 鉴权 = 完整 key → PG unique 点查（database.md §4.3），无进程内鉴权缓存
-├── auth/                         # L1 管理端认证域（原 admin/auth/service.rs）
-│   ├── mod.rs  types.rs  store.rs  service.rs   # store = PG admin_users + Redis 会话键（§4B.1）
-├── settings/                     # L1 运行时设置域（原 config/settings.rs）
-│   ├── mod.rs  types.rs  store.rs  service.rs   # service 持 watch 广播
-├── update/                       # L1 自更新域（原 admin/system/*）
-│   ├── mod.rs  types.rs  state.rs  release.rs  download.rs  archive.rs  service.rs
-├── accounts/                     # L2 账号域（原 upstream/accounts + upstream/scheduler + admin/accounts/service）
-│   ├── mod.rs
-│   ├── account.rs                # 实体（原 model.rs）
-│   ├── window.rs
-│   ├── pool.rs                   # 池：存储态/槽位/过滤（§5.4 行数预算内需拆出 pool/ 子文件）
-│   ├── scheduler.rs              # 调度门面（原 scheduler/mod.rs 的类型）
-│   ├── scheduler/
-│   │   ├── candidates.rs  feedback.rs
-│   │   └── strategy/
-│   │       ├── mod.rs  smart.rs  quota_reset.rs  round_robin.rs  sticky.rs
-│   ├── store.rs                  # AccountStore trait + PG 门面
-│   ├── store/
-│   │   ├── queries.rs  rows.rs
-│   ├── quota/
-│   │   ├── mod.rs  service.rs  runtime.rs
-│   ├── refresh/                  # 原 token_refresh
-│   │   ├── mod.rs  policy.rs  lease.rs  service.rs   # lease = Redis SET NX PX（§4B.2）
-│   ├── cookies/
-│   │   ├── mod.rs  store.rs
-│   ├── import.rs                 # 导入 payload 解析（原 upstream/accounts/importing.rs）
-│   └── manage/                   # 管理操作（原 admin/accounts/service/*）
-│       ├── mod.rs  types.rs  service.rs  lifecycle.rs
-│       ├── import.rs  export.rs  oauth.rs  probe.rs  quota.rs  quota_view.rs
-├── models/                       # L2 模型目录域
-│   ├── mod.rs  types.rs  catalog.rs  service.rs  store.rs   # store = Redis 快照 HASH（§4B.4）
-├── dispatch/                     # L3 请求编排域（原 proxy/dispatch）
-│   ├── mod.rs
-│   ├── service.rs                # 编排入口（原 responses/service.rs，按 §5.4 拆分）
-│   ├── upstream_call.rs          # 原 dispatch/upstream.rs
-│   ├── errors.rs                 # 原 dispatch/errors.rs + responses/errors.rs 合并
-│   ├── recording.rs              # 原 responses/event_recording.rs，调 telemetry::recorder
-│   ├── affinity/                 # 原 session_affinity.rs（683 行三合一）+ responses/affinity.rs
-│   │   ├── mod.rs  types.rs  identity.rs  store.rs  service.rs  resolve.rs
-│   │   │                         # store = Redis 键 + 二级索引（§4B.3），无内存映射副本
-│   ├── recovery/
-│   │   ├── mod.rs  auth.rs  cloudflare.rs  exhaustion.rs
-│   │   ├── implicit_resume.rs  reasoning_replay.rs
-│   └── stream/
-│       ├── mod.rs  live.rs  lifecycle.rs  prefetch.rs  sse_failure.rs  trace.rs
-├── api/                          # L4 全部入站 HTTP（原 http + web + proxy 入口 + admin routes）
-│   ├── mod.rs
-│   ├── router.rs                 # 总路由 + healthz（infra 的 PG ping + Redis ping，任一失败 503）
-│   ├── assets.rs                 # SPA 静态资源（原 web/assets.rs）
-│   ├── middleware/
-│   │   ├── mod.rs  request_id.rs  trace.rs
-│   ├── client/                   # /v1（原 proxy/auth + proxy/openai）
-│   │   ├── mod.rs  router.rs  auth.rs
-│   │   ├── responses.rs  models.rs  errors.rs  sse.rs
-│   └── admin/                    # /api/admin（原 admin 的全部 routes 与 DTO）
-│       ├── mod.rs  router.rs  response.rs  session.rs      # session = 鉴权提取器/中间件
-│       ├── auth_routes.rs  settings_routes.rs  system_routes.rs
-│       ├── accounts_routes.rs  keys_routes.rs
-│       ├── usage_routes.rs  ops_routes.rs  dashboard_routes.rs   # ops_routes = /api/admin/ops/errors
-└── bootstrap/                    # L5 进程装配（原 runtime + config/schema,loader）
-    ├── mod.rs
-    ├── config.rs                 # 启动配置 schema + loader（原 config/schema.rs + loader.rs；含 redis.url）
-    ├── services.rs  state.rs  shutdown.rs
-    ├── import_sqlite.rs          # 一次性 SQLite v3 → PG 导入命令（database.md §6；§5.5 的唯一跨域写豁免）
-    └── tasks/
-        ├── mod.rs  coordinator.rs  periodic.rs  cleanup.rs
-        ├── cookie_cleanup.rs  retention_trim.rs   # retention_trim 新增（database.md §7）
-        ├── model_refresh.rs  token_refresh.rs  quota_refresh.rs
-        ├── fingerprint_update.rs
-        # session_cleanup / session_affinity_cleanup 不复存在：Redis TTL 接管（database.md §4B）
+├── main.rs  lib.rs
+├── infra/                        # database.rs、redis.rs、migrations/ 与通用基础能力
+├── upstream/openai/              # 纯 OpenAI/Codex client
+│   ├── token_client.rs
+│   ├── protocol/                 # schema / responses / events / SSE / WebSocket
+│   ├── transport/                # HTTP、TLS、WebSocket 与连接池；大文件可按职责建子文件
+│   └── fingerprint/              # types / store / runtime / updater
+├── telemetry/                    # 四类账本与查询
+│   ├── recorder.rs               # 成功/失败事实的唯一领域写入口
+│   ├── billing.rs  diagnostics.rs  rebuild.rs
+│   ├── usage/                    # types.rs、query.rs、store.rs、store/query.rs
+│   ├── ops/                      # types.rs、query.rs、store.rs
+│   ├── buckets/                  # query.rs、store.rs
+│   └── account_usage/            # query.rs、store.rs
+├── keys/                         # types / PG store / verifier / manage
+├── auth/                         # types / PG+Redis store / session service
+├── settings/                     # types / PG store / serialized current-snapshot service
+├── update/                       # types / state / release / download / archive / service
+├── accounts/
+│   ├── account.rs  service.rs  import.rs  window.rs
+│   ├── pool.rs  pool/{runtime,types}.rs
+│   ├── scheduler.rs  scheduler/{candidates,feedback,strategy/...}.rs
+│   ├── store.rs  store/{queries,rows,write}.rs
+│   ├── quota/{service,runtime}.rs
+│   ├── refresh/{lease,service,runtime}.rs  refresh/runtime/types.rs
+│   ├── cookies/store.rs
+│   └── manage/{service,types,lifecycle,import,export,oauth,probe,quota,quota_view}.rs
+├── models/                       # config / info / snapshot / catalog / service / Redis store
+├── dispatch/
+│   ├── errors.rs  upstream.rs  cloudflare.rs  exhaustion.rs
+│   ├── implicit_resume.rs  reasoning_replay.rs
+│   ├── affinity/{types,identity,store,service,resolve}.rs
+│   └── responses/                # Responses 编排是一个完整子能力，不做无收益扁平化
+│       ├── service.rs  service/{complete,stream,compact}.rs
+│       ├── errors.rs  event_recording.rs  live_stream.rs
+│       └── prefetch.rs  sse_failure.rs  stream_lifecycle.rs  trace.rs
+├── api/
+│   ├── router.rs  assets.rs  middleware/
+│   ├── client/                   # /v1 路由、鉴权与协议响应
+│   └── admin/                    # /api/admin handlers 与 DTO
+│       ├── accounts_routes.rs  accounts_routes/{actions,stats}.rs
+│       ├── dashboard_routes.rs  dashboard_routes/data.rs
+│       └── keys/ops/settings/system/usage 等 routes
+└── bootstrap/
+    ├── application.rs  config.rs  services.rs  state.rs  shutdown.rs
+    ├── import_sqlite.rs  import_sqlite/{core,telemetry}.rs
+    └── tasks/                    # periodic、retention、token/quota/model refresh 等装配任务
 ```

-`backend/tests/` 目录结构**镜像 `src/`**：模块搬家时测试目录同步搬家，`fixtures/`、`support/` 不动。
+这是已落地结构，不是待办蓝图。为控制单文件行数而增加的同名子目录属于同一能力，不构成抽象层。Dashboard 的跨域只读组合保留在 `api/admin/dashboard_routes/data.rs`：它只组装 presentation read model，不写 SQL、不改变领域状态；把它下沉到 `telemetry` 反而会制造 telemetry → accounts 的向上依赖。
+
+`backend/tests/` 顶层按 `src/` 领域镜像；子目录可按公开路由或行为场景组织，因此 `monitoring`、`token_refresh` 等历史行为名称可保留在测试中。`fixtures/`、`support/` 是测试基础设施，不要求与生产文件逐一同名。

 ---

@@ -248,7 +187,7 @@ backend/src/

 ### 5.2 存储边界（SQL 与 Redis 同规）

-SQL 文本与 Redis 命令只允许出现在 `store.rs` / `store/` / `query.rs`。`api` 层与 `service.rs` 出现 `sqlx::query` 或 `redis::` 调用即为违规（现状 `http/router.rs` 的 healthz 直接拿某个 store 的连接池执行裸 SQL——终态 `infra::database` 提供 `ping(pool)`、`infra::redis` 提供 `ping(conn)`，router 只调它们）。
+业务 SQL 与 Redis 命令只允许出现在 `store.rs` / `store/` / `query.rs`。明确的基础设施/维护例外只有：`infra/database.rs`（迁移与 ping）、`infra/redis.rs`（连接与 ping）、`bootstrap/import_sqlite*`（一次性跨库导入）、`telemetry/rebuild.rs`（账本重建命令）、`accounts/refresh/lease.rs`（Redis 租约适配器）。`api` 层与领域 `service.rs` 出现存储命令即为违规；healthz 只调用 infra 的 ping 门面。

 ### 5.3 HTTP 边界

@@ -257,20 +196,20 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名
 ### 5.4 文件行数预算

 - 软上限 **400 行**（超过即在 review 中说明理由）；硬上限 **800 行**（超过必须拆，CI 直接阻断，§9）。
- 现状超硬线清单（11 个，全部有既定拆法）：
+- 迁移前超硬线清单（11 个）及最终落点：

 | 文件（现路径） | 行数 | 拆法 |
 | --- | --- | --- |
-| `proxy/dispatch/responses/service.rs` | 1913 | → `dispatch/service.rs`（编排骨架）+ `upstream_call.rs` + `stream/*` + `recovery/*` 已分走的逻辑归位 |
-| `upstream/transport/websocket.rs` | 1470 | → `websocket.rs`（连接/握手）+ `websocket_frames.rs`（帧编解码） |
-| `admin/accounts/routes.rs` | 1455 | → `api/admin/accounts_routes.rs` 按子资源拆（accounts / oauth / quota / import-export） |
-| `upstream/accounts/pool.rs` | 1420 | → `accounts/pool.rs`（门面）+ `pool/state.rs`（槽位/在途）+ `pool/filters.rs` |
-| `upstream/transport/client.rs` | 1251 | → `client.rs` + `client_sse.rs`（SSE 请求路径） |
-| `upstream/accounts/store/mod.rs` | 1126 | → `store.rs` 门面 + `store/queries.rs` 按读写再分 |
-| `admin/monitoring/usage_record_store.rs` | 1114 | PR0 原地重写（拆表后 metadata 聚合全数删除，行数大减），PR1 搬家时拆 `telemetry/usage/{store,query}.rs` |
-| `admin/monitoring/dashboard.rs` | 1080 | → `telemetry/dashboard.rs`（聚合）+ `api/admin/dashboard_routes.rs`（HTTP） |
-| `upstream/protocol/websocket.rs` | 1033 | → `websocket.rs`（事件转换）+ `websocket_errors.rs`（错误帧解析） |
-| `upstream/accounts/token_refresh/runtime.rs` | 953 | → `accounts/refresh/{service,lease,policy}.rs` |
+| `proxy/dispatch/responses/service.rs` | 1913 | → `dispatch/responses/service.rs` + `service/{complete,stream,compact}.rs` |
+| `upstream/transport/websocket.rs` | 1470 | → `upstream/openai/transport/websocket.rs` + `websocket/{stream,types}.rs` + `websocket_pump.rs` |
+| `admin/accounts/routes.rs` | 1455 | → `api/admin/accounts_routes.rs` + `accounts_routes/{actions,stats}.rs`；领域操作进入 `accounts/manage` |
+| `upstream/accounts/pool.rs` | 1420 | → `accounts/pool.rs` + `pool/{runtime,types}.rs` |
+| `upstream/transport/client.rs` | 1251 | → `upstream/openai/transport/client.rs` + `client/requests.rs` |
+| `upstream/accounts/store/mod.rs` | 1126 | → `accounts/store.rs` + `store/{queries,rows,write}.rs` |
+| `admin/monitoring/usage_record_store.rs` | 1114 | → `telemetry/usage/store.rs` + `store/query.rs`；查询门面独立为 `usage/query.rs` |
+| `admin/monitoring/dashboard.rs` | 1080 | → 账本查询进入 `telemetry`，跨域 presentation 组合进入 `api/admin/dashboard_routes/{data}.rs` |
+| `upstream/protocol/websocket.rs` | 1033 | → `upstream/openai/protocol/websocket.rs` + `websocket/validation.rs` |
+| `upstream/accounts/token_refresh/runtime.rs` | 953 | → `accounts/refresh/{service,lease,runtime}.rs` + `runtime/types.rs` |
 | `admin/system/updater/mod.rs` | 811 | → `update/{service,state}.rs` |

 ### 5.5 一域一账本
@@ -294,7 +233,7 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名
 | `cpr:affinity:*` | `dispatch`（affinity/store.rs） |
 | `cpr:models:plan_snapshots` | `models` |

-跨域读通过属主域的 query/service 方法，不直接跨域写。**唯一豁免**：`bootstrap/import_sqlite.rs`（一次性导入命令，database.md §6）按依赖序直写全部 PG 表；v3 源库退役后该文件可删。
+跨域读通过属主域的 query/service 方法，不直接跨域写。**跨域写唯一豁免**：`bootstrap/import_sqlite*`（一次性导入命令，database.md §6）按依赖序直写全部 PG 表；`telemetry/rebuild.rs` 仍只写 telemetry 自有账本，不是跨域豁免。v3 源库退役后导入模块可删。

 ---

@@ -304,7 +243,7 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名

 1. **热路径反向依赖管理端**：`proxy/dispatch/usage_events.rs`、`event_recording.rs` import `admin::monitoring` 的 `Admin*Service` 写账本；`runtime/services.rs` 用 Admin 命名的服务装配核心链路。遥测是领域层，管理端只是它的一个读者。→ 建 `telemetry` 域（PR1）。
 2. **核心编排寄居 proxy**：`proxy/dispatch/`（约 5600 行，含 1913 行 service）是系统心脏，却挂在客户端 HTTP 入口下；`SqliteSessionAffinityStore`（DB store）住在 proxy 里。→ 建 `dispatch` 域（PR4）。
-3. **config 模块越权**：`config/settings.rs`（601 行）= 持久化 + 热更新推送 + admin API key 校验，还持有账号池引用向上推送。config 应只剩启动 schema + loader。→ 建 `settings` 域 + watch 反转（PR2）。
+3. **config 模块越权**：`config/settings.rs`（601 行）= 持久化 + 热更新推送 + admin API key 校验，还持有账号池引用向上推送。config 应只剩启动 schema + loader。→ 建 `settings` 域，由 bootstrap 在持久化成功后统一应用快照（PR2）。
 4. **admin/keys 四合一**：`admin/keys/service.rs`（698 行）同时装着 proxy 鉴权用的 `ClientKeyService`、运行时缓存、SQLite store、管理端 service——客户端鉴权路径穿过 admin 模块。→ 建 `keys` 域（PR2）。
 5. **web/http 假分家**：`web/` 只有 69 行 assets，是路由的 fallback 组成部分。→ 并入 `api`（PR5）。
 6. healthz 穿透三层拿 store 连接池执行裸 SQL（§5.2）。
@@ -352,17 +291,17 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名
 | `proxy/dispatch/usage_events.rs` | `telemetry/recorder.rs` †（Admin* 类型消失） |
 | `proxy/dispatch/session_affinity.rs` | `dispatch/affinity/{types,store,service}.rs` † |
 | `proxy/dispatch/responses/affinity.rs` | `dispatch/affinity/resolve.rs` |
-| `proxy/dispatch/responses/service.rs` | `dispatch/service.rs` † |
-| `proxy/dispatch/responses/{live_stream,stream_lifecycle,prefetch,sse_failure,trace}.rs` | `dispatch/stream/{live,lifecycle,prefetch,sse_failure,trace}.rs` |
-| `proxy/dispatch/responses/event_recording.rs` | `dispatch/recording.rs` |
-| `proxy/dispatch/responses/errors.rs` + `proxy/dispatch/errors.rs` | `dispatch/errors.rs` † |
-| `proxy/dispatch/{auth_recovery,cloudflare,exhaustion,implicit_resume,reasoning_replay}.rs` | `dispatch/recovery/{auth,cloudflare,exhaustion,implicit_resume,reasoning_replay}.rs` |
-| `proxy/dispatch/upstream.rs` | `dispatch/upstream_call.rs` |
+| `proxy/dispatch/responses/service.rs` | `dispatch/responses/service.rs` + `service/{complete,stream,compact}.rs` † |
+| `proxy/dispatch/responses/{live_stream,stream_lifecycle,prefetch,sse_failure,trace}.rs` | `dispatch/responses/` 下同名文件 |
+| `proxy/dispatch/responses/event_recording.rs` | `dispatch/responses/event_recording.rs`（只依赖 `telemetry::Recorder`） |
+| `proxy/dispatch/responses/errors.rs` + `proxy/dispatch/errors.rs` | `dispatch/responses/errors.rs` + `dispatch/errors.rs`（协议编排错误与通用调度错误分开） |
+| `proxy/dispatch/{auth_recovery,cloudflare,exhaustion,implicit_resume,reasoning_replay}.rs` | 通用逻辑保留在 `dispatch/` 根；Responses 专属恢复留在 `dispatch/responses/` |
+| `proxy/dispatch/upstream.rs` | `dispatch/upstream.rs` |
 | `admin/monitoring/usage_record_{model,store,service}.rs` | `telemetry/usage/{types,store,query}.rs` †（重写已在 PR0 完成，此处纯搬家） |
 | `admin/monitoring/ops_error_{model,store,service}.rs` | `telemetry/ops/{types,store,query}.rs` † |
 | `admin/monitoring/account_usage_{store,service}.rs` | `telemetry/account_usage/{store,query}.rs` † |
 | `admin/monitoring/billing.rs` | `telemetry/billing.rs` |
-| `admin/monitoring/dashboard.rs` | `telemetry/dashboard.rs` + `api/admin/dashboard_routes.rs` † |
+| `admin/monitoring/dashboard.rs` | 各账本查询进入 `telemetry/*/query.rs`；跨域只读组合进入 `api/admin/dashboard_routes/{data}.rs` † |
 | `admin/monitoring/diagnostics.rs` | `telemetry/diagnostics.rs` |
 | `admin/monitoring/usage_record_routes.rs` | `api/admin/usage_routes.rs` |
 | `admin/keys/service.rs` | `keys/{types,store,service,manage}.rs` †（进程内鉴权缓存消亡，database.md §4.3） |
@@ -372,7 +311,7 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名
 | `admin/accounts/routes.rs` | `api/admin/accounts_routes.rs` †（按子资源拆） |
 | `admin/accounts/quota_view.rs` | `accounts/manage/quota_view.rs` |
 | `admin/accounts/service/*` | `accounts/manage/*`（`importing`→`import`、`exporting`→`export`、`testing`→`probe`、`contracts`→`types`） |
-| `admin/{response,update_payload}.rs` | `api/admin/response.rs`、`update/types.rs` |
+| `admin/{response,update_payload}.rs` | `api/admin/{response,update_payload}.rs` |
 | `admin/router.rs` | `api/admin/router.rs` |
 | `admin/settings/routes.rs` | `api/admin/settings_routes.rs` |
 | `admin/system/{routes,state}.rs`、`admin/system/updater/*` | `api/admin/system_routes.rs`、`update/{state,service,release,download,archive}.rs` † |
@@ -380,12 +319,12 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名
 | `upstream/accounts/{pool,window}.rs` | `accounts/{pool,window}.rs`（pool 拆 †） |
 | `upstream/accounts/store/*` | `accounts/store.rs` + `accounts/store/{queries,rows}.rs` † |
 | `upstream/accounts/quota/*` | `accounts/quota/{service,runtime}.rs` † |
-| `upstream/accounts/token_refresh/*` | `accounts/refresh/{policy,lease,service}.rs` † |
+| `upstream/accounts/token_refresh/*` | `accounts/refresh/{lease,service,runtime}.rs` + `runtime/types.rs` † |
 | `upstream/accounts/cookies.rs` | `accounts/cookies/store.rs` |
 | `upstream/accounts/importing.rs` | `accounts/import.rs` |
-| `upstream/accounts/service.rs` | 并入 `accounts/mod.rs` 门面导出（32 行） |
+| `upstream/accounts/service.rs` | `accounts/service.rs` |
 | `upstream/scheduler/*` | `accounts/scheduler.rs` + `accounts/scheduler/*`（mod.rs 门面类型移出） |
-| `upstream/models/*` | `models/{types,catalog,service,store}.rs`（`backend_entry`+`info`+`snapshot`+`config` 并入 `types.rs`/`service.rs`） |
+| `upstream/models/*` | `models/{config,info,snapshot,catalog,service,store}.rs`（各文件保留真实职责） |
 | `upstream/fingerprint.rs` | `upstream/openai/fingerprint/{types,store,runtime,updater}.rs` †（Repository→Store） |
 | `upstream/{protocol,transport}/*`、`upstream/token_client.rs` | `upstream/openai/` 下同名路径（websocket/client 按 §5.4 拆） |
 | `infra/*` | 原地保留 |
@@ -401,9 +340,9 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名
 | --- | --- | --- |
 | **PR0** | **存储终态切换**（database.md §7 全部 12 条，在现有目录结构内完成） | Cargo：sqlx 加 postgres、新增 redis；`infra/database.rs` 改 PG + 0001 终态基线、新增 `infra/redis.rs`；全部 store 换 PG 方言 / 运行态四键空间改 Redis；成功/失败事实拆表 + provider 维度 + 两条写入事务；client key 可取回、session/admin-key 哈希化 + `client_api_key_id` 归因贯通；Dashboard 口径换源；trim 周期任务化；`import-sqlite` / `rebuild-buckets` 子命令；tests support 换 PG+Redis;deploy compose 加双服务。验收对齐 database.md §6.4 |
 | **PR1** | `telemetry` 域成立 | monitoring 全部迁出 admin；`usage_events.rs` → `recorder.rs`；Admin* 遥测类型按 §4.3 更名；dispatch/api 改依赖 telemetry；启用"禁止 `crate::admin::monitoring`"检查 |
-| **PR2** | `keys` / `auth` / `settings` / `update` 域成立 | 四个域从 admin/config 拆出；settings 改 watch 广播（§2.3）；config 只剩 schema+loader |
+| **PR2** | `keys` / `auth` / `settings` / `update` 域成立 | 四个域从 admin/config 拆出；settings 只保留持久化与串行快照，消费者传播由 bootstrap 统一执行（§2.3）；config 只剩启动 schema+loader |
 | **PR3** | `accounts` / `models` 域成立 | `upstream/accounts`、`upstream/scheduler`、`admin/accounts/service` 三处合并为 `accounts`；`upstream` 缩为纯客户端并落成 `openai/` 子目录；动名词文件更名；Repository→Store |
-| **PR4** | `dispatch` 域成立 | proxy/dispatch 整体迁出并按目标树重组；affinity store 独立成文件；`proxy` 只剩 api/client 的料 |
+| **PR4** | `dispatch` 域成立 | proxy/dispatch 整体迁出；保留内聚的 `responses` 子能力并按行数拆 service 子文件；affinity store 独立成文件；`proxy` 只剩 api/client 的料 |
 | **PR5** | `api` / `bootstrap` 壳成立 | http + web + proxy 入口 + admin routes 合并为 `api`；runtime + config 装配部分改名 `bootstrap`；删除 `crate::{admin,proxy,runtime,web,http,config}` 六个旧根模块；启用全部依赖检查 |
 | **PR6** | 行数预算清账 | §5.4 清单内 11 个文件拆到硬线以下；启用行数预算 CI 阻断 |

@@ -413,62 +352,12 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名

 ## 9. CI 强制

-新增 `backend/scripts/check_architecture.sh`，进 `.github/workflows/ci.yml`（在 clippy 之后）：
+`backend/scripts/check_architecture.sh` 是可执行规则的唯一来源，并由 `.github/workflows/ci.yml` 在 clippy 后调用；本文不复制脚本，避免规则双写漂移。当前强制项包括：

-```bash
-#!/usr/bin/env bash
-# 架构依赖与命名断言。违规即失败。
-set -euo pipefail
-cd "$(dirname "$0")/.."
-fail=0
-deny() { # deny <描述> <grep-pattern> <路径...>
-  if grep -rn --include='*.rs' -E "$2" "${@:3}" ; then
-    echo "ARCH VIOLATION: $1" >&2; fail=1
-  fi
-}
-
-# 分层：领域层不得依赖 api / bootstrap
-deny "domain -> api/bootstrap" 'crate::(api|bootstrap)\b' \
-  src/infra src/upstream src/telemetry src/keys src/auth src/settings src/update \
-  src/accounts src/models src/dispatch
-# upstream 是纯客户端
-deny "upstream -> domain" 'crate::(accounts|dispatch|telemetry|keys|auth|settings|update|models)\b' src/upstream
-# HTTP 类型只在 api / bootstrap
-deny "axum outside api" '\baxum(::|\b)' \
-  src/infra src/upstream src/telemetry src/keys src/auth src/settings src/update \
-  src/accounts src/models src/dispatch
-# SQL 与 Redis 命令只在 store / query
-deny "sqlx outside store/query" '\bsqlx(::|\b)' \
-  $(find src/api -name '*.rs')
-deny "redis outside store" '\bredis::' \
-  $(find src/api -name '*.rs')
-# 旧模块根不得复活
-deny "legacy module path" 'crate::(admin|proxy|runtime|web|http|config)::' src
-# 领域类型与命名纪律
-deny "Admin-prefixed domain type" '\bAdmin[A-Z][A-Za-z0-9]*\b' \
-  src/infra src/upstream src/telemetry src/keys src/auth src/settings src/update \
-  src/accounts src/models src/dispatch
-deny "Repository naming" '\b[A-Za-z0-9]*Repository\b' src
-deny "persistence repository naming" '\b[A-Za-z0-9_]*repository[A-Za-z0-9_]*\b' \
-  src/accounts src/auth src/dispatch src/keys src/models src/settings src/telemetry \
-  src/upstream/openai/fingerprint
-deny "Arc alias" '\bArc[[:space:]]+as[[:space:]]+' src
-# 禁用文件名
-banned=$(find src -name 'importing.rs' -o -name 'exporting.rs' -o -name 'testing.rs' \
-  -o -name 'util.rs' -o -name 'helper.rs' -o -name 'common.rs' -o -name 'misc.rs')
-[ -n "$banned" ] && { echo "ARCH VIOLATION: banned file names: $banned" >&2; fail=1; }
-# mod.rs 纪律
-while read -r f; do
-  lines=$(wc -l < "$f")
-  [ "$lines" -gt 50 ] && { echo "ARCH VIOLATION: $f has $lines lines (mod.rs <= 50)" >&2; fail=1; }
-done < <(find src -name mod.rs)
-# 行数硬上限（PR6 已完成，违规即失败）
-while read -r f; do
-  lines=$(wc -l < "$f")
-  [ "$lines" -gt 800 ] && { echo "ARCH VIOLATION: $f has $lines lines (> 800)" >&2; fail=1; }
-done < <(find src -name '*.rs' ! -name mod.rs)
-exit $fail
-```
+- 领域层不得依赖 `api` / `bootstrap`，且同时识别 `crate::api::...` 与 `use crate::{ api::... }` 两种写法；`upstream` 不得依赖任何领域模块。
+- axum 只进入 `api` / `bootstrap`；API 不得执行 SQL 或 Redis 命令。
+- 存储命令只能位于 §5.2 列出的 adapter/query/维护入口；文件名、类型前缀、旧模块路径、Repository 命名与 Arc 别名均有阻断。
+- `mod.rs` 不超过 50 行，其他 Rust 文件不超过 800 行。

 全部规则已启用；后续新增代码一旦恢复旧路径、越层依赖、违规命名或突破硬行数上限，CI 直接失败。

@@ -476,10 +365,10 @@ exit $fail

 ## 10. 仓库级约定

- `backend/tests/` 镜像 `src/`，不在 `src/` 写测试模块。测试依赖本机（或 CI 服务容器）的 PostgreSQL 与 Redis：`CPR_TEST_DATABASE_URL` 下每测试建独立数据库、`CPR_TEST_REDIS_URL` 下每测试随机键前缀（database.md §7）。
- `docs/` 只保留权威文档：`architecture.md`（本文）、`database.md`。行为细节（调度策略、WS 池、更新流程等）以模块文档注释为准（`scheduler/mod.rs` 是范本），不再在 docs 里维护第二份会腐化的副本。
+- `backend/tests/` 顶层镜像 `src/` 领域，不在 `src/` 写测试模块。测试依赖本机（或 CI 服务容器）的 PostgreSQL 与 Redis：`CPR_TEST_DATABASE_URL` 下每测试建独立数据库、`CPR_TEST_REDIS_URL` 下每测试随机键前缀（database.md §7）。
+- `architecture.md`（本文）与 `database.md` 是架构/存储权威；`docs/` 可保留故障复盘或修复记录，但不得另行定义架构、schema 或运行契约。行为实现细节以模块文档注释为准。
 - `deploy/` 随 PR0 扩容（compose 增 postgres/redis 服务与健康检查依赖）；`release/`、`skills/` 现状合格，不动。
- 前端联动（全部随 PR0）：keys 列表继续返回完整 key，并长期提供复制与 CCSwitch 导入；admin API key 状态不再有脱敏值（哈希不可逆，只余"已启用"）；成功记录响应不再有 `level` 字段；错误明细/筛选改走 `/api/admin/ops/errors`；Dashboard "总请求/总 token" 卡片改为 account_usage 来源并标注"成功累计"。
+- 前端联动（全部随 PR0）：keys 列表继续返回完整 key，复制与 CCSwitch 导入长期保持可用且交互/文案不变；admin API key 状态不再有脱敏值（哈希不可逆，只余"已启用"）；成功记录响应不再有 `level` 字段；错误明细/筛选改走 `/api/admin/ops/errors`；Dashboard 只修正取数来源，既有卡片文案不变。

 ---

@@ -66,7 +66,7 @@
 | 布尔 | boolean | — |
 | 快照/模板/调试 JSON | jsonb | 见 §2.7 |

-**枚举用 text + check,不用 PG 原生 enum**(有意修订旧文档 §8 的一句话设想):原生 enum 增删值要 `alter type` 且值永远无法删除;check 约束换一条 `drop/add constraint` 即完成枚举演进(SQLite 0002 改名先例证明枚举确实会变)。**不使用 real/double 存业务数据,不存储金额**。成本一律查询期由 token × 单价计算(`admin/monitoring/billing.rs`):单价会变、需追溯重算,落库的浮点成本既漂移又过期。
+**枚举用 text + check,不用 PG 原生 enum**(有意修订旧文档 §8 的一句话设想):原生 enum 增删值要 `alter type` 且值永远无法删除;check 约束换一条 `drop/add constraint` 即完成枚举演进(SQLite 0002 改名先例证明枚举确实会变)。**不使用 real/double 存业务数据,不存储金额**。成本一律查询期由 token × 单价计算(`telemetry/billing.rs`):单价会变、需追溯重算,落库的浮点成本既漂移又过期。

 ### 2.3 时间戳

@@ -114,7 +114,7 @@
 2. **可丢弃性是准入条件**:FLUSHALL 之后系统必须无数据损失地继续运行(会话重登、租约重竞争、亲和退化为普通调度、缓存重拉)。不满足即不属于 Redis。
 3. **互斥必须原子**:获取用 `SET NX PX`,续约/释放用 Lua 比较 owner 后操作,禁止 GET-判断-SET 三段式。
 4. 二级索引(亲和的 conversation ZSET / account SET)由写入方维护,读取方容忍成员悬垂(主键已过期),遇悬垂惰性清理。
-5. 值统一 JSON 文本;不用 Redis Hash 拆业务对象字段(整读整写,无字段级更新需求)。
+5. 单个业务对象统一编码为完整 JSON 文本,不把对象属性拆成 Redis Hash 字段。集合缓存可使用 HASH（如 `cpr:models:plan_snapshots`），但每个 field 的 value 仍是一个完整 JSON 对象，整读整写。

 ### 2.10 保留与可重建性(诚实边界)

@@ -699,8 +699,8 @@ errors:    错误分布 / 排查                                   ← ops_error

 ### 5.3 Dashboard 口径决策(钉死)

- 所有 traffic 卡片(今日请求、区间请求、错误率、QPS 趋势)**一律来自 request_time_buckets**,语义为"bucket 保留期内"。卡片文案如实标注区间,**不提供也不暗示"历史总量"**——bucket 有 90 天保留期,任何"total"字样都是口径谎言。
- 生命周期累计仅有一处合法来源:`account_usage` 累计列(成功口径),使用时必须标注"成功累计"。
+- 周期 traffic 卡片(今日请求、区间请求、错误率、QPS 趋势)**一律来自 request_time_buckets**,语义为"bucket 保留期内"；生命周期累计卡片不从 bucket 推导。
+- 生命周期累计仅有一处合法来源:`account_usage` 累计列(成功口径)。这是后端数据语义，不要求改动既有前端文案。
 - 同一张卡片禁止混排两种口径的数字(v3 的 `todayRequests` 来自 bucket、`totalRequests` 来自 usage_records 全表扫描就是这个病)。
 - 错误明细与错误分布来自 ops_error_logs;成功事件列表不再混排错误行。
 - 真正的长期趋势需求 → 日粒度归档表 `request_day_buckets`(§8),不是延长 15 分钟桶保留期。
@@ -761,7 +761,7 @@ codex-proxy-rs import-sqlite <旧库路径.sqlite>
 | 3 | 抽查 `select count(*) from client_api_keys where key is null or key = ''` = 0 | 旧 key 值不变；鉴权、管理端复制与 CCSwitch 导入均可用 |
 | 3a | 抽查 `select count(*) from accounts where status = 'refreshing'` = 0 | 旧瞬态状态已规范化为 `expired`，且 `next_refresh_at is null` |
 | 4 | 执行 `rebuild-buckets`(§7),用新口径重算事实保留期内的桶 | 保留期内桶 token 列 = usage_records 聚合值 |
-| 5 | 核对 Dashboard 卡片全部走 bucket 口径,生命周期数字标注"成功累计" | 同卡片无混合口径数字 |
+| 5 | 核对 Dashboard 周期卡片走 bucket、生命周期累计走 account_usage，前端既有文案不变 | 同卡片无混合口径数字 |

 ---

@@ -774,7 +774,7 @@ codex-proxy-rs import-sqlite <旧库路径.sqlite>
 3. **鉴权归因**(`keys/{store,service,manage}.rs`、`api/client/auth.rs`):key 创建落唯一 `key` 列；鉴权按完整 key 做 PG unique 点查(删除进程内鉴权缓存);鉴权返回 key `id`,调用链把 `client_api_key_id` 装进请求上下文直至事件写入。管理端列表持续返回完整 key，前端长期保留复制与 CCSwitch 导入入口。
 4. **事件结构**(`telemetry/{usage,ops}`、`telemetry/recorder.rs`、`dispatch/responses/event_recording.rs`):成功/失败事件携带 `client_api_key_id` 与 `provider`;token、requested/upstream_model、service_tier、first_token_ms 从 metadata 移到一等字段;metadata 不再写已提升字段;UsageRecord 删除 `level`(成功事实无等级)。
 5. **写入路径**(`telemetry/usage/store.rs`、`telemetry/ops/store.rs`、`telemetry/buckets/store.rs`):§5.2 的两条 PG 事务;错误路径对桶只 `error_count + 1`;`__unknown__` sentinel;min/max 延迟使用 nullable-safe 的 `least()`/`greatest()`;内联 trim 移除。
-6. **查询路径**(`telemetry/{usage,ops,account_usage}`、`api/admin/{usage,ops,dashboard,accounts}_routes*`):summary/分布/趋势全部走列,删除所有 `->>` 聚合与 metadata LIKE;新增 `/api/admin/ops/errors` 错误明细查询面;Dashboard 按 §5.3 改卡片来源与文案。
+6. **查询路径**(`telemetry/{usage,ops,account_usage}`、`api/admin/{usage,ops,dashboard,accounts}_routes*`):summary/分布/趋势全部走列,删除所有 `->>` 聚合与 metadata LIKE;新增 `/api/admin/ops/errors` 错误明细查询面;Dashboard 按 §5.3 只改取数来源，不改既有文案。
 7. **运行态改 Redis**(`auth/store.rs` 会话、`accounts/refresh/lease.rs`、`dispatch/affinity/store.rs`、`models/store.rs`):按 §4B 契约实现;删除 admin 会话/亲和清理任务与亲和重启恢复;账号删除路径挂 Redis 亲和级联(§4B.3)。
 8. **维护命令**(`main.rs` CLI):`import-sqlite <path>`(§6);`rebuild-buckets`——删除事实保留期内的桶并从两张事实表重算(15 分钟槽对齐在 SQL 侧 `floor(epoch/900)*900`,§2.3);保留期外只读不动。范围明确**不含** `account_usage` / `account_model_usage`(§2.10,不可重建)。
 9. **清理任务**(`bootstrap/tasks/`):三张增长表的周期 trim(读 `runtime_settings` 三列);cookie 清理保留;admin 会话与亲和清理任务删除(TTL 接管)。