codex-proxy-rs/backend/crates/gateway-store/tests/postgres/execution/mod.rs

//! 验证执行记录端口的请求、尝试、恢复租约与快照持久化

mod source_governance;

use std::time::{Duration as StdDuration, SystemTime};

use chrono::{DateTime, Duration, Utc};
use gateway_admin::{
    model::observability as admin_observability, ports::store::ObservabilityStore as _,
};
use gateway_core::diagnostics::{MAX_TRACE_SNAPSHOT_BYTES, TraceContext};
use gateway_core::engine::{
    ExecutionOutcome, ExecutionStore, ModelRequestFailureObservation,
    ModelRequestFinalization as CoreModelRequestFinalization, ModelRequestId,
    ModelRequestTimings as CoreModelRequestTimings, NewModelRequest as CoreNewModelRequest,
};
use gateway_core::error::{
    GatewayError, GatewayErrorKind, ProviderConnectionObservation, StoreErrorKind,
};
use gateway_core::metering::{CalculatedCost, CostEstimate, Usage};
use gateway_core::operation::OperationKind;
use gateway_core::policy::ClientApiKeyId;
use gateway_core::routing::{AccountRoutingSnapshot, ConfigRevision, PublicModelId};
use gateway_core::upstream::UpstreamSendState;
use gateway_store::postgres::{
    ModelRequestAttemptStart, ModelRequestRepository, NewModelRequest, ObservabilityPageSize,
    ObservabilityRange, OpsErrorFilter, OpsErrorQuery, PgExecutionStore, UsageRecordFilter,
    UsageRecordQuery,
};
use serde_json::{Value, json};
use sqlx::PgPool;

use super::{TestDatabase, admin_observability_store, observability_repository};

#[test]
fn model_request_rejects_empty_historical_client_key_ref() {
    let started_at = Utc::now();
    let request = NewModelRequest {
        admission_decision_ms: None,
        id: "request-1".to_owned(),
        client_api_key_name_snapshot: None,
        client_api_key_ref: String::new(),
        config_revision: 1,
        protocol: "openai".to_owned(),
        operation: "responses".to_owned(),
        endpoint: "/v1/responses".to_owned(),
        client_transport: "http_sse".to_owned(),
        requested_model_id: Some("coding".to_owned()),
        routing_scope: "all".to_owned(),
        routing_group_refs: Vec::new(),
        routing_group_names_snapshot: serde_json::json!([]),
        client_ip: None,
        user_agent: None,
        reasoning_effort: None,
        reasoning_preset: None,
        request_kind: None,
        subagent_kind: None,
        compact: false,
        continuation: Default::default(),
        image_generation_requested: false,
        started_at,
        lease_expires_at: started_at + Duration::seconds(30),
    };
    assert!(request.validate().is_err());
}

#[tokio::test]
async fn request_recovery_lease_is_renewed_and_abandoned_requests_are_recovered() {
    let Some(database) = TestDatabase::create("request_renewal").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let mut request = accepted_request("req_long_running");
    request.started_at = SystemTime::now() - StdDuration::from_secs(601);
    request.deadline_at = Default::default();
    store.create_model_request(request.clone()).await.unwrap();
    let initial: DateTime<Utc> = sqlx::query_scalar("update model_request_leases set lease_expires_at = now() + interval '1 second' where request_id = $1 returning lease_expires_at")
        .bind(request.id.as_str()).fetch_one(&database.pool).await.unwrap();
    let renewal = store.maintain_request(&request.id, request.deadline_at);
    let renewed: DateTime<Utc> = tokio::time::timeout(StdDuration::from_secs(2), async {
        loop {
            let at: DateTime<Utc> = sqlx::query_scalar(
                "select lease_expires_at from model_request_leases where request_id = $1",
            )
            .bind(request.id.as_str())
            .fetch_one(&database.pool)
            .await
            .unwrap();
            if at > initial + Duration::seconds(500) {
                break at;
            }
            tokio::time::sleep(StdDuration::from_millis(10)).await;
        }
    })
    .await
    .expect("live request refreshes recovery lease");
    assert_eq!(
        store
            .recover_expired(initial.into())
            .await
            .unwrap()
            .requests,
        0
    );
    drop(renewal);
    assert_eq!(
        store
            .recover_expired(renewed.into())
            .await
            .unwrap()
            .requests,
        1
    );
    let row = stored_row(&database.pool, request.id.as_str()).await;
    assert_eq!(row["error_kind"], "process_interrupted");
    database.close().await;
}

#[tokio::test]
async fn merged_model_less_first_attempt_should_match_sequential_semantics() {
    let Some(database) = TestDatabase::create("execution_merged_insert").await else {
        return;
    };
    let repository = PgExecutionStore::new(database.pool.clone());
    let started_at = Utc::now();
    sqlx::query(
        "insert into provider_accounts (
           id, provider_kind, name, email, upstream_user_id,
           upstream_account_id, plan_type, authentication_kind,
           provider_credentials_json, credential_revision,
           has_refresh_token, access_token_expires_at, enabled, credential_state,
           credential_observed_at, created_at, updated_at
         ) values (
           'acct_merged', 'openai', 'merged', null, 'user-merged', null, null, 'oauth',
           '{}'::jsonb, 1, false, $1 + interval '1 day', true, 'ready', $1, $1, $1
         )",
    )
    .bind(started_at)
    .execute(&database.pool)
    .await
    .expect("seed provider account");
    let request = NewModelRequest {
        admission_decision_ms: None,
        id: "req_merged".to_owned(),
        client_api_key_name_snapshot: None,
        client_api_key_ref: "key_merged".to_owned(),
        config_revision: 1,
        protocol: "openai".to_owned(),
        operation: "generate_image".to_owned(),
        endpoint: "/v1/images/generations".to_owned(),
        client_transport: "http_json".to_owned(),
        requested_model_id: None,
        routing_scope: "all".to_owned(),
        routing_group_refs: Vec::new(),
        routing_group_names_snapshot: serde_json::json!([]),
        client_ip: None,
        user_agent: None,
        reasoning_effort: None,
        reasoning_preset: None,
        request_kind: None,
        subagent_kind: None,
        compact: false,
        continuation: Default::default(),
        image_generation_requested: false,
        started_at,
        lease_expires_at: started_at + Duration::seconds(30),
    };
    let attempt = ModelRequestAttemptStart {
        account_selection_wait_ms: None,
        capacity_used_slots: None,
        capacity_total_slots: None,
        model_request_id: "req_merged".to_owned(),
        attempt_count: 1,
        provider_kind: "openai".to_owned(),
        account_snapshot: None,
        provider_account_ref: Some("acct_merged".to_owned()),
        upstream_model_id: None,
        upstream_transport: "http_json".to_owned(),
        http_version: None,
    };

    repository
        .insert_model_request_with_first_attempt(request, attempt)
        .await
        .expect("merged insert");

    let (attempt_count, send_state, provider_kind, outcome, requested_model, upstream_model): (
        i32,
        String,
        String,
        String,
        Option<String>,
        Option<String>,
    ) = sqlx::query_as(
        "select attempt_count, upstream_send_state, provider_kind, outcome,
                    requested_model_id, upstream_model_id
             from model_requests where id = 'req_merged'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load merged request");
    assert_eq!(
        (
            attempt_count,
            send_state.as_str(),
            provider_kind.as_str(),
            outcome.as_str(),
            requested_model,
            upstream_model,
        ),
        (1, "not_sent", "openai", "running", None, None)
    );

    // 后续 attempt 沿用常规 CAS 递增路径；已持久化的 sent 水位不被重试重置
    repository
        .mark_upstream_send_state(
            "req_merged",
            gateway_store::postgres::UpstreamSendState::Sent,
        )
        .await
        .expect("mark sent before retry");
    let second = repository
        .begin_model_request_attempt(ModelRequestAttemptStart {
            account_selection_wait_ms: None,
            capacity_used_slots: None,
            capacity_total_slots: None,
            model_request_id: "req_merged".to_owned(),
            attempt_count: 2,
            provider_kind: "openai".to_owned(),
            account_snapshot: None,
            provider_account_ref: Some("acct_merged".to_owned()),
            upstream_model_id: None,
            upstream_transport: "http_json".to_owned(),
            http_version: None,
        })
        .await
        .expect("second attempt");
    assert_eq!(second, 2);
    let send_state: String = sqlx::query_scalar(
        "select upstream_send_state from model_requests where id = 'req_merged'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load send state after retry");
    assert_eq!(send_state, "sent");
    database.close().await;
}

#[tokio::test]
async fn model_request_persists_group_routing_snapshot_without_live_group_foreign_keys() {
    let Some(database) = TestDatabase::create("execution_group_routing_history").await else {
        return;
    };
    let repository = PgExecutionStore::new(database.pool.clone());
    let started_at = Utc::now();
    repository
        .insert_model_request(NewModelRequest {
            admission_decision_ms: None,
            id: "req_group_history".to_owned(),
            client_api_key_name_snapshot: None,
            client_api_key_ref: "key_group_history".to_owned(),
            config_revision: 7,
            routing_scope: "groups".to_owned(),
            routing_group_refs: vec![
                "grp_00000000000000000000000000000001".to_owned(),
                "grp_00000000000000000000000000000002".to_owned(),
            ],
            routing_group_names_snapshot: serde_json::json!(["Production Pool", "Overflow Pool"]),
            protocol: "openai".to_owned(),
            operation: "responses".to_owned(),
            endpoint: "/v1/responses".to_owned(),
            client_transport: "http_sse".to_owned(),
            requested_model_id: Some("coding".to_owned()),
            client_ip: None,
            user_agent: None,
            reasoning_effort: None,
            reasoning_preset: None,
            request_kind: None,
            subagent_kind: None,
            compact: false,
            continuation: Default::default(),
            image_generation_requested: false,
            started_at,
            lease_expires_at: started_at + Duration::seconds(30),
        })
        .await
        .expect("insert grouped request history");

    let stored: (String, Vec<String>, serde_json::Value) = sqlx::query_as(
        "select routing_scope, routing_group_refs, routing_group_names_snapshot
         from model_request_observations where id = 'req_group_history'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load grouped request history");
    assert_eq!(stored.0, "groups");
    assert_eq!(
        stored.1,
        [
            "grp_00000000000000000000000000000001",
            "grp_00000000000000000000000000000002",
        ]
    );
    assert_eq!(
        stored.2,
        serde_json::json!(["Production Pool", "Overflow Pool"])
    );
    database.close().await;
}

#[tokio::test]
async fn downstream_commit_should_atomically_record_http_status_once() {
    let Some(database) = TestDatabase::create("execution_commit_status").await else {
        return;
    };
    seed_running_request(&database.pool, "req_commit_status")
        .await
        .expect("seed model request");
    let repository = PgExecutionStore::new(database.pool.clone());

    let committed = ModelRequestRepository::mark_downstream_committed(
        &repository,
        "req_commit_status",
        Utc::now(),
        Some(200),
    )
    .await
    .expect("commit downstream");
    let overwritten = repository
        .record_client_status_code("req_commit_status", 500)
        .await
        .expect("reject status overwrite without store failure");
    let (committed_at, status): (Option<chrono::DateTime<Utc>>, Option<i32>) = sqlx::query_as(
        "select downstream_committed_at, client_status_code
         from model_requests where id = 'req_commit_status'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load committed request");

    assert!(committed);
    assert!(!overwritten);
    assert!(committed_at.is_some());
    assert_eq!(status, Some(200));
    database.close().await;
}

#[tokio::test]
async fn terminal_failure_should_accept_exactly_one_client_status_backfill() {
    let Some(database) = TestDatabase::create("execution_terminal_status").await else {
        return;
    };
    seed_running_request(&database.pool, "req_terminal_status")
        .await
        .expect("seed model request");
    sqlx::query(
        "update model_requests
         set outcome = 'failed', completed_at = now()
         where id = 'req_terminal_status'",
    )
    .execute(&database.pool)
    .await
    .expect("finalize model request");
    let repository = PgExecutionStore::new(database.pool.clone());

    let recorded = repository
        .record_client_status_code("req_terminal_status", 429)
        .await
        .expect("record terminal client status");
    let overwritten = repository
        .record_client_status_code("req_terminal_status", 500)
        .await
        .expect("reject terminal status overwrite without store failure");
    let status: Option<i32> = sqlx::query_scalar(
        "select client_status_code from model_requests where id = 'req_terminal_status'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load terminal client status");

    assert!(recorded);
    assert!(!overwritten);
    assert_eq!(status, Some(429));
    database.close().await;
}

#[tokio::test]
async fn core_adapter_should_persist_calculated_cost_exactly() {
    let Some(database) = TestDatabase::create("execution_calculated_cost").await else {
        return;
    };
    seed_running_request(&database.pool, "req_calculated_cost")
        .await
        .expect("seed model request");
    sqlx::query(
        "update model_requests
         set provider_kind = 'xai', provider_account_ref = 'acct_xai',
             upstream_model_id = 'grok-4.5', upstream_transport = 'http_sse', attempt_count = 1
         where id = 'req_calculated_cost'",
    )
    .execute(&database.pool)
    .await
    .expect("seed model request attempt");
    let repository = PgExecutionStore::new(database.pool.clone());

    let mut finalization = successful_core_finalization("req_calculated_cost");
    finalization.downstream_committed_at = Some(std::time::SystemTime::now());
    finalization.websocket_pool = Some("reuse".to_owned());
    finalization.service_tier = Some("priority".to_owned());
    finalization.cost = CalculatedCost::from_usd_ticks(12_345)
        .expect("calculated cost")
        .into_estimate();
    ExecutionStore::finalize_model_request(&repository, finalization)
        .await
        .expect("persist calculated cost");
    let persisted: (
        String,
        String,
        String,
        String,
        String,
        String,
        Option<String>,
    ) = sqlx::query_as(
        "select cost_source, cost_amount::text, cost_currency, upstream_transport, http_version,
                websocket_pool, service_tier
         from model_request_observations where id = 'req_calculated_cost'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load calculated cost");

    assert_eq!(
        persisted,
        (
            "calculated".to_owned(),
            "0.0000012345".to_owned(),
            "USD".to_owned(),
            "websocket".to_owned(),
            "HTTP/2".to_owned(),
            "reuse".to_owned(),
            Some("priority".to_owned()),
        )
    );
    let observation = observability_repository(&database.pool);
    let detail = observation
        .usage_record_detail("req_calculated_cost")
        .await
        .expect("request detail");
    assert_eq!(
        detail.request.upstream_model_id.as_deref(),
        Some("grok-4.5")
    );
    let page = observation
        .list_usage_records(UsageRecordQuery {
            status: Default::default(),
            range: ObservabilityRange::new(
                Utc::now() - Duration::hours(1),
                Utc::now() + Duration::hours(1),
            )
            .unwrap(),
            filter: UsageRecordFilter::default(),
            current_page: 1,
            page_size: ObservabilityPageSize::new(10).unwrap(),
        })
        .await
        .expect("request list");
    assert_eq!(page.items.len(), 1);
    database.close().await;
}

#[tokio::test]
async fn billing_snapshot_survives_later_price_changes_and_usage_detail_reads() {
    use gateway_core::metering::{
        CalculatedCostAmounts, CalculatedCostBreakdown, CalculatedCostRates, CurrencyCode, Decimal,
        Money,
    };
    let Some(database) = TestDatabase::create("billing_snapshot").await else {
        return;
    };
    seed_running_request(&database.pool, "req_billing_snapshot")
        .await
        .unwrap();
    sqlx::query(
        "update model_requests
         set provider_kind = 'xai', upstream_model_id = 'grok-4'
         where id = 'req_billing_snapshot'",
    )
    .execute(&database.pool)
    .await
    .unwrap();
    let money = |ticks| {
        Money::new(
            Decimal::from_scaled(ticks).unwrap(),
            CurrencyCode::new("USD").unwrap(),
        )
    };
    let billing = CalculatedCostBreakdown::new(
        CalculatedCostAmounts::new(
            money(1_000_000),
            money(2_000_000),
            money(3_000_000),
            money(0),
            money(6_000_000),
            money(6_000_000),
        ),
        CalculatedCostRates::new(
            money(10_000_000_000),
            money(20_000_000_000),
            money(5_000_000_000),
            money(0),
        ),
        Some("default".to_owned()),
        100,
    )
    .with_long_context_billing(true)
    .with_custom_multiplier(12500)
    .unwrap();
    let mut finalization = successful_core_finalization("req_billing_snapshot");
    finalization.cost = billing.calculated_cost().into_estimate();
    ExecutionStore::finalize_model_request(
        &PgExecutionStore::new(database.pool.clone()),
        finalization,
    )
    .await
    .unwrap();
    let observation = admin_observability_store(&database.pool);
    let original = observation
        .usage_record_detail("req_billing_snapshot")
        .await
        .unwrap();
    let Some(admin_observability::UsageBilling::Calculated(saved)) = &original.request.billing
    else {
        panic!("persisted billing breakdown must be restored without provider recalculation");
    };
    assert!(saved.long_context_billing_applied);
    assert_eq!(saved.custom_multiplier_bps, 12500);
    assert_eq!(saved.total_amount.amount, "0.00075".parse().unwrap());
    sqlx::query("update runtime_settings set pricing_overrides_json = $1 where id = 1")
        .bind(json!({"xai":{"grok-4":{"multiplierBps":90000,"bands":{}}}}))
        .execute(&database.pool)
        .await
        .unwrap();
    let after = observation
        .usage_record_detail("req_billing_snapshot")
        .await
        .unwrap();
    assert_eq!(original.request.billing, after.request.billing);
    sqlx::query("update model_requests set billing_snapshot_json = billing_snapshot_json - 'longContextBillingApplied' where id = 'req_billing_snapshot'")
        .execute(&database.pool).await.unwrap();
    let legacy = observation
        .usage_record_detail("req_billing_snapshot")
        .await
        .unwrap();
    let Some(admin_observability::UsageBilling::Calculated(legacy)) = legacy.request.billing else {
        panic!("legacy billing snapshot must remain readable");
    };
    assert!(!legacy.long_context_billing_applied);
    assert_eq!(legacy.total_amount, saved.total_amount);

    database.close().await;
}

#[tokio::test]
async fn core_adapter_should_persist_image_result_and_new_websocket_pool() {
    let Some(database) = TestDatabase::create("execution_image_usage").await else {
        return;
    };
    seed_running_request(&database.pool, "req_image_usage")
        .await
        .expect("seed image request");
    sqlx::query(
        "update model_requests
         set image_generation_requested = true, provider_kind = 'openai',
             provider_account_ref = 'acct_openai', upstream_model_id = 'gpt-image',
             upstream_transport = 'websocket', attempt_count = 1
         where id = 'req_image_usage'",
    )
    .execute(&database.pool)
    .await
    .expect("seed image attempt");
    let repository = PgExecutionStore::new(database.pool.clone());
    let mut usage = Usage::new();
    usage.image_input_tokens = Some(31);
    usage.image_output_tokens = Some(9);
    let mut finalization = successful_core_finalization("req_image_usage");
    finalization.usage = usage;
    finalization.image_generation_succeeded = Some(true);
    finalization.websocket_pool = Some("new".to_owned());

    ExecutionStore::finalize_model_request(&repository, finalization)
        .await
        .expect("persist image usage");
    let persisted: (bool, Option<i64>, Option<i64>, Option<bool>, Option<String>) = sqlx::query_as(
        "select image_generation_requested, image_input_tokens, image_output_tokens,
                    image_generation_succeeded, websocket_pool
             from model_request_observations where id = 'req_image_usage'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load image usage");

    assert_eq!(
        persisted,
        (true, Some(31), Some(9), Some(true), Some("new".to_owned()))
    );
    database.close().await;
}

#[tokio::test]
async fn expired_image_request_should_be_recovered_as_failed() {
    let Some(database) = TestDatabase::create("execution_expired_image").await else {
        return;
    };
    seed_running_request(&database.pool, "req_expired_image")
        .await
        .expect("seed image request");
    sqlx::query(
        "update model_requests
         set image_generation_requested = true,
             started_at = now() - interval '2 seconds',
             recovery_lease_expires_at = now() - interval '1 second'
         where id = 'req_expired_image'",
    )
    .execute(&database.pool)
    .await
    .expect("expire image request");
    let repository = PgExecutionStore::new(database.pool.clone());

    repository
        .recover_expired_model_requests(Utc::now())
        .await
        .expect("recover expired image request");
    let persisted: (String, Option<bool>) = sqlx::query_as(
        "select outcome, image_generation_succeeded
         from model_requests where id = 'req_expired_image'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load recovered image request");

    assert_eq!(persisted, ("incomplete".to_owned(), Some(false)));
    database.close().await;
}

fn successful_core_finalization(id: &str) -> CoreModelRequestFinalization {
    CoreModelRequestFinalization {
        diagnostic_trace_json: None,
        request_id: ModelRequestId::new(id).expect("request id"),
        outcome: ExecutionOutcome::Succeeded,
        send_state: UpstreamSendState::Sent,
        attempt_count: 1,
        downstream_committed_at: None,
        client_status_code: Some(200),
        client_response_id: None,
        upstream_status_code: Some(200),
        upstream_request_id: None,
        upstream_response_id: None,
        upstream_transport: Some("websocket".to_owned()),
        http_version: Some("HTTP/2".to_owned()),
        websocket_pool: None,
        service_tier: None,
        provider_metadata_json: None,
        error: None,
        provider_error_code: None,
        error_details: None,
        failure_observation: Default::default(),
        retry_after_ms: None,
        usage: Usage::new(),
        image_generation_succeeded: None,
        cost: CostEstimate::unavailable(),
        timings: CoreModelRequestTimings::default(),
        completed_at: std::time::SystemTime::now(),
    }
}

#[tokio::test]
async fn clock_rollback_preserves_terminal_outcomes_and_errors_after_recovery() {
    let Some(database) = TestDatabase::create("execution_clock_rollback").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    for (index, outcome) in [
        ExecutionOutcome::Succeeded,
        ExecutionOutcome::Failed,
        ExecutionOutcome::Cancelled,
        ExecutionOutcome::Incomplete,
    ]
    .into_iter()
    .enumerate()
    {
        let id = format!("req_clock_rollback_{index}");
        let request = accepted_request(&id);
        store
            .create_model_request(request.clone())
            .await
            .expect("create request before clock rollback");
        store
            .begin_model_request_attempt(ModelRequestAttemptStart {
                account_selection_wait_ms: None,
                capacity_used_slots: None,
                capacity_total_slots: None,
                model_request_id: id.clone(),
                attempt_count: 1,
                provider_kind: "openai".to_owned(),
                account_snapshot: None,
                provider_account_ref: Some("acct_clock_rollback".to_owned()),
                upstream_model_id: Some("coding".to_owned()),
                upstream_transport: "websocket".to_owned(),
                http_version: None,
            })
            .await
            .expect("record real attempt before clock rollback");
        let mut finalization = successful_core_finalization(&id);
        finalization.outcome = outcome;
        finalization.timings.latency_ms = Some(125);
        finalization.completed_at = request.started_at - StdDuration::from_millis(400);
        if outcome != ExecutionOutcome::Succeeded {
            finalization.error = Some(GatewayError::new(
                GatewayErrorKind::RateLimited,
                "synthetic upstream quota error",
            ));
            finalization.provider_error_code = Some("usage_limit_reached".to_owned());
            finalization.upstream_status_code = Some(429);
        }
        ExecutionStore::finalize_model_request(&store, finalization)
            .await
            .expect("finalize despite wall clock rollback");
        let terminal = stored_row(&database.pool, &id).await;
        assert_eq!(terminal["completed_at"], terminal["started_at"]);
        assert_eq!(terminal["latency_ms"], 125);
        assert_eq!(
            terminal["outcome"],
            ["succeeded", "failed", "cancelled", "incomplete"][index]
        );
        if outcome != ExecutionOutcome::Succeeded {
            assert_eq!(terminal["error_kind"], "rate_limited");
            assert_eq!(terminal["provider_error_code"], "usage_limit_reached");
            assert_eq!(terminal["error_message"], "synthetic upstream quota error");
        }
        assert_eq!(
            store
                .recover_expired(request.deadline_at.at().unwrap())
                .await
                .expect("recovery leaves finalized request intact")
                .requests,
            0
        );
        assert_eq!(stored_row(&database.pool, &id).await, terminal);
    }
    database.close().await;
}

#[tokio::test]
async fn zero_attempt_finalization_preserves_real_error_after_clock_rollback() {
    let Some(database) = TestDatabase::create("zero_attempt_clock_rollback").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let request = accepted_request("req_zero_attempt_clock_rollback");
    store
        .create_model_request(request.clone())
        .await
        .expect("create zero attempt request");
    let mut finalization = early_failure(&request);
    finalization.completed_at = request.started_at - StdDuration::from_millis(400);
    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .expect("finalize zero attempt after clock rollback");
    assert_eq!(
        store
            .recover_expired(request.deadline_at.at().unwrap())
            .await
            .expect("recovery preserves real early failure")
            .requests,
        0
    );
    let terminal = stored_row(&database.pool, request.id.as_str()).await;
    assert_eq!(terminal["outcome"], "failed");
    assert_eq!(terminal["error_kind"], "no_available_provider");
    assert_eq!(terminal["attempt_count"], 0);
    assert_eq!(terminal["completed_at"], terminal["started_at"]);
    database.close().await;
}

#[tokio::test]
async fn core_adapter_persists_opaque_response_ids_as_bytes() {
    let Some(database) = TestDatabase::create("execution_opaque_response_id").await else {
        return;
    };
    seed_running_request(&database.pool, "req_opaque_response_id")
        .await
        .expect("seed model request");
    let store = PgExecutionStore::new(database.pool.clone());
    let response_id = format!("resp_{}\0opaque", "x".repeat(4_096));
    let mut finalization = successful_core_finalization("req_opaque_response_id");
    finalization.client_response_id = Some(response_id.clone());
    finalization.upstream_response_id = Some(response_id.clone());

    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .expect("persist opaque response IDs");

    let persisted: (Option<Vec<u8>>, Option<Vec<u8>>) = sqlx::query_as(
        "select client_response_id, upstream_response_id
         from model_requests where id = 'req_opaque_response_id'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load opaque response IDs");
    assert_eq!(persisted.0.as_deref(), Some(response_id.as_bytes()));
    assert_eq!(persisted.1.as_deref(), Some(response_id.as_bytes()));

    database.close().await;
}

#[tokio::test]
async fn core_adapter_persists_native_causes_and_verbatim_upstream_details() {
    let Some(database) = TestDatabase::create("execution_error_details").await else {
        return;
    };
    seed_running_request(&database.pool, "req_raw_upstream_error")
        .await
        .expect("seed model request");
    let store = PgExecutionStore::new(database.pool.clone());
    let raw = r#"{"error":{"message":"raw upstream body","opaque":"\u0000"}}"#;
    let provider_error = gateway_core::error::ProviderError::new(
        gateway_core::error::ProviderErrorKind::Unavailable,
        UpstreamSendState::Sent,
    )
    .with_source(std::io::Error::other("original local cause"))
    .with_raw_upstream_error(gateway_core::error::RawUpstreamError::new(raw));
    let mut finalization = successful_core_finalization("req_raw_upstream_error");
    finalization.outcome = ExecutionOutcome::Failed;
    finalization.client_status_code = Some(502);
    finalization.upstream_status_code = Some(500);
    finalization.error = Some(GatewayError::new(
        GatewayErrorKind::UpstreamUnavailable,
        "upstream service is unavailable",
    ));
    finalization.error_details = provider_error.error_details();

    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .expect("persist restricted error details");

    let persisted: Option<String> = sqlx::query_scalar(
        "select error_details from model_requests where id = 'req_raw_upstream_error'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load restricted error details");
    let persisted: Value = serde_json::from_str(persisted.as_deref().unwrap()).unwrap();
    assert_eq!(persisted["upstream"], raw);
    assert_eq!(
        persisted["causes"]["messages"],
        json!(["original local cause"])
    );
    assert_eq!(persisted["causes"]["truncated"], false);

    database.close().await;
}

#[tokio::test]
async fn core_adapter_should_persist_only_object_provider_observation() {
    let Some(database) = TestDatabase::create("execution_provider_observation").await else {
        return;
    };
    seed_running_request(&database.pool, "req_provider_observation")
        .await
        .expect("seed model request");
    let store = PgExecutionStore::new(database.pool.clone());
    let mut finalization = successful_core_finalization("req_provider_observation");
    finalization.provider_metadata_json = Some("{\"effectiveModel\":\"gpt-test\"}".to_owned());

    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .expect("persist provider observation");
    let persisted: serde_json::Value = sqlx::query_scalar(
        "select provider_observation_json from model_requests where id = 'req_provider_observation'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load provider observation");
    assert_eq!(persisted["effectiveModel"], "gpt-test");

    database.close().await;
}

#[tokio::test]
async fn core_adapter_should_reject_non_object_provider_observation() {
    let Some(database) = TestDatabase::create("execution_invalid_provider_observation").await
    else {
        return;
    };
    seed_running_request(&database.pool, "req_invalid_provider_observation")
        .await
        .expect("seed model request");
    let store = PgExecutionStore::new(database.pool.clone());
    let mut finalization = successful_core_finalization("req_invalid_provider_observation");
    finalization.provider_metadata_json = Some("[]".to_owned());

    assert!(
        ExecutionStore::finalize_model_request(&store, finalization)
            .await
            .is_err()
    );

    database.close().await;
}

#[tokio::test]
async fn core_adapter_persists_continuation_connection_failure_observation() {
    let Some(database) = TestDatabase::create("execution_continuation_observation").await else {
        return;
    };
    seed_running_request(&database.pool, "req_continuation_observation")
        .await
        .expect("seed model request");
    let store = PgExecutionStore::new(database.pool.clone());
    let mut finalization = successful_core_finalization("req_continuation_observation");
    finalization.outcome = ExecutionOutcome::Failed;
    finalization.client_status_code = Some(400);
    finalization.error = Some(GatewayError::new(
        GatewayErrorKind::InvalidRequest,
        "conversation continuation must be rebuilt",
    ));
    finalization.failure_observation = ModelRequestFailureObservation {
        continuation_unavailable_reason: Some("reused_connection_lost".to_owned()),
        upstream_connection: Some(ProviderConnectionObservation::new(
            "connection-observed",
            "tcp_reset",
            12_000,
            4_000,
        )),
    };

    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .expect("persist continuation failure observation");

    type PersistedContinuationObservation = (
        Option<String>,
        Option<String>,
        Option<String>,
        Option<String>,
        Option<i64>,
        Option<i64>,
    );
    let persisted: PersistedContinuationObservation = sqlx::query_as(
        "select error_kind, continuation_unavailable_reason,
                upstream_connection_id, upstream_connection_exit_reason,
                upstream_connection_age_ms, upstream_connection_idle_ms
           from model_request_observations where id = 'req_continuation_observation'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load continuation failure observation");
    assert_eq!(
        persisted.0.as_deref(),
        Some("continuation_recovery_required")
    );
    assert_eq!(persisted.1.as_deref(), Some("reused_connection_lost"));
    assert_eq!(persisted.2.as_deref(), Some("connection-observed"));
    assert_eq!(persisted.3.as_deref(), Some("tcp_reset"));
    assert_eq!(persisted.4, Some(12_000));
    assert_eq!(persisted.5, Some(4_000));

    database.close().await;
}

#[tokio::test]
async fn successful_new_chain_should_mark_recent_continuation_failure_recovered() {
    let Some(database) = TestDatabase::create("execution_continuation_recovered").await else {
        return;
    };
    let affinity_hash = "a".repeat(64);
    let failure_completed_at = Utc::now() - Duration::seconds(8);
    seed_continuation_failure(
        &database.pool,
        "req_continuation_failed",
        &affinity_hash,
        failure_completed_at,
    )
    .await;
    seed_recovery_request(
        &database.pool,
        "req_continuation_recovery",
        &affinity_hash,
        failure_completed_at + Duration::seconds(4),
    )
    .await;

    let store = PgExecutionStore::new(database.pool.clone());
    ExecutionStore::finalize_model_request(
        &store,
        successful_core_finalization("req_continuation_recovery"),
    )
    .await
    .expect("finalize successful recovery request");

    type PersistedRecoveryObservation = (
        Option<String>,
        Option<chrono::DateTime<Utc>>,
        i32,
        Option<i64>,
        Option<i64>,
    );
    let recovered: PersistedRecoveryObservation = sqlx::query_as(
        "select recovery_request_id, recovered_at, recovery_attempt_count,
                recovery_retry_delay_ms, recovery_total_latency_ms
           from model_request_observations where id = 'req_continuation_failed'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load recovered continuation failure");

    assert_eq!(recovered.0.as_deref(), Some("req_continuation_recovery"));
    assert!(recovered.1.is_some());
    assert_eq!(recovered.2, 1);
    assert_eq!(recovered.3, Some(4_000));
    assert!(recovered.4.is_some_and(|latency| latency >= 4_000));
    database.close().await;
}

#[tokio::test]
async fn successful_http_downgrade_should_mark_pending_websocket_failures_recovered() {
    let Some(database) = TestDatabase::create("execution_transport_downgrade_recovered").await
    else {
        return;
    };
    let affinity_hash = "d".repeat(64);
    let first_failure_at = Utc::now() - Duration::seconds(8);
    let second_failure_at = first_failure_at + Duration::seconds(1);
    seed_websocket_transport_failure(
        &database.pool,
        "req_websocket_failed_first",
        &affinity_hash,
        first_failure_at,
    )
    .await;
    seed_websocket_transport_failure(
        &database.pool,
        "req_websocket_failed_second",
        &affinity_hash,
        second_failure_at,
    )
    .await;
    seed_websocket_transport_failure(
        &database.pool,
        "req_websocket_failed_after_delivery",
        &affinity_hash,
        second_failure_at,
    )
    .await;
    sqlx::query(
        "update model_requests
            set downstream_committed_at = completed_at
          where id = 'req_websocket_failed_after_delivery'",
    )
    .execute(&database.pool)
    .await
    .expect("mark control failure delivered downstream");

    let recovery_started_at = first_failure_at + Duration::seconds(4);
    seed_transport_recovery_request(
        &database.pool,
        "req_http_fallback_succeeded",
        &affinity_hash,
        recovery_started_at,
    )
    .await;
    let store = PgExecutionStore::new(database.pool.clone());
    let mut recovery = successful_core_finalization("req_http_fallback_succeeded");
    recovery.upstream_transport = Some("http_sse".to_owned());
    recovery.downstream_committed_at = Some(std::time::SystemTime::now());
    ExecutionStore::finalize_model_request(&store, recovery)
        .await
        .expect("finalize successful HTTP fallback");

    type PersistedRecovery = (
        String,
        Option<String>,
        Option<chrono::DateTime<Utc>>,
        i32,
        Option<i64>,
        Option<i64>,
    );
    let recovered: Vec<PersistedRecovery> = sqlx::query_as(
        "select id, recovery_request_id, recovered_at, recovery_attempt_count,
                recovery_retry_delay_ms, recovery_total_latency_ms
           from model_request_observations
          where id like 'req_websocket_failed_%'
          order by id",
    )
    .fetch_all(&database.pool)
    .await
    .expect("load WebSocket recovery observations");

    assert_eq!(recovered.len(), 3);
    let delivered = &recovered[0];
    assert_eq!(delivered.0, "req_websocket_failed_after_delivery");
    assert_eq!((&delivered.1, delivered.2, delivered.3), (&None, None, 0));
    for (row, expected_delay_ms) in recovered[1..].iter().zip([4_000, 3_000]) {
        assert_eq!(row.1.as_deref(), Some("req_http_fallback_succeeded"));
        assert!(row.2.is_some());
        assert_eq!(row.3, 1);
        assert_eq!(row.4, Some(expected_delay_ms));
        assert!(row.5.is_some_and(|latency| latency >= expected_delay_ms));
    }
    let repository = super::observability_repository(&database.pool);
    let failed_detail = repository
        .usage_record_detail("req_websocket_failed_first")
        .await
        .unwrap();
    assert_eq!(
        failed_detail.related_requests[0]["requestId"],
        "req_http_fallback_succeeded"
    );
    assert_eq!(
        failed_detail.related_requests[0]["relation"],
        "recovered_by"
    );
    let recovered_detail = repository
        .usage_record_detail("req_http_fallback_succeeded")
        .await
        .unwrap();
    assert_eq!(recovered_detail.related_requests.len(), 2);
    assert!(
        recovered_detail
            .related_requests
            .iter()
            .all(|related| related["relation"] == "recovers")
    );
    database.close().await;
}

#[tokio::test]
async fn websocket_success_or_late_http_success_should_not_hide_transport_failure() {
    let Some(database) = TestDatabase::create("execution_transport_downgrade_unresolved").await
    else {
        return;
    };
    let failure_at = Utc::now() - Duration::seconds(60);
    let store = PgExecutionStore::new(database.pool.clone());

    let websocket_affinity = "e".repeat(64);
    seed_websocket_transport_failure(
        &database.pool,
        "req_websocket_control_failure",
        &websocket_affinity,
        failure_at,
    )
    .await;
    seed_transport_recovery_request(
        &database.pool,
        "req_websocket_control_success",
        &websocket_affinity,
        failure_at + Duration::seconds(4),
    )
    .await;
    ExecutionStore::finalize_model_request(
        &store,
        successful_core_finalization("req_websocket_control_success"),
    )
    .await
    .expect("finalize successful WebSocket control request");

    let late_affinity = "f".repeat(64);
    seed_websocket_transport_failure(
        &database.pool,
        "req_websocket_late_failure",
        &late_affinity,
        failure_at,
    )
    .await;
    seed_transport_recovery_request(
        &database.pool,
        "req_http_late_success",
        &late_affinity,
        failure_at + Duration::seconds(31),
    )
    .await;
    let mut late_recovery = successful_core_finalization("req_http_late_success");
    late_recovery.upstream_transport = Some("http_sse".to_owned());
    ExecutionStore::finalize_model_request(&store, late_recovery)
        .await
        .expect("finalize late HTTP request");

    type UnresolvedTransportFailure = (String, Option<String>, Option<chrono::DateTime<Utc>>, i32);
    let unresolved: Vec<UnresolvedTransportFailure> = sqlx::query_as(
        "select id, recovery_request_id, recovered_at, recovery_attempt_count
               from model_requests
              where id in ('req_websocket_control_failure', 'req_websocket_late_failure')
              order by id",
    )
    .fetch_all(&database.pool)
    .await
    .expect("load unresolved transport failures");
    assert_eq!(unresolved.len(), 2);
    assert!(
        unresolved
            .iter()
            .all(|row| row.1.is_none() && row.2.is_none() && row.3 == 0)
    );
    database.close().await;
}

#[tokio::test]
async fn failed_or_late_new_chain_should_leave_continuation_failure_unresolved() {
    let Some(database) = TestDatabase::create("execution_continuation_unresolved").await else {
        return;
    };
    let affinity_hash = "c".repeat(64);
    let failure_completed_at = Utc::now() - Duration::seconds(60);
    seed_continuation_failure(
        &database.pool,
        "req_continuation_unresolved",
        &affinity_hash,
        failure_completed_at,
    )
    .await;
    seed_recovery_request(
        &database.pool,
        "req_continuation_retry_failed",
        &affinity_hash,
        failure_completed_at + Duration::seconds(3),
    )
    .await;

    let store = PgExecutionStore::new(database.pool.clone());
    let mut failed = successful_core_finalization("req_continuation_retry_failed");
    failed.outcome = ExecutionOutcome::Failed;
    failed.client_status_code = Some(502);
    failed.upstream_status_code = Some(500);
    failed.error = Some(GatewayError::new(
        GatewayErrorKind::UpstreamUnavailable,
        "recovery request failed",
    ));
    ExecutionStore::finalize_model_request(&store, failed)
        .await
        .expect("finalize failed recovery request");

    let after_failed: (Option<String>, Option<chrono::DateTime<Utc>>, i32) = sqlx::query_as(
        "select recovery_request_id, recovered_at, recovery_attempt_count
           from model_requests where id = 'req_continuation_unresolved'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load unresolved continuation failure");
    assert_eq!(after_failed, (None, None, 1));

    seed_recovery_request(
        &database.pool,
        "req_continuation_retry_late",
        &affinity_hash,
        failure_completed_at + Duration::seconds(31),
    )
    .await;
    ExecutionStore::finalize_model_request(
        &store,
        successful_core_finalization("req_continuation_retry_late"),
    )
    .await
    .expect("finalize late recovery request");

    let after_late: (Option<String>, Option<chrono::DateTime<Utc>>, i32) = sqlx::query_as(
        "select recovery_request_id, recovered_at, recovery_attempt_count
           from model_requests where id = 'req_continuation_unresolved'",
    )
    .fetch_one(&database.pool)
    .await
    .expect("load late unresolved continuation failure");
    assert_eq!(after_late, (None, None, 1));
    database.close().await;
}

async fn seed_continuation_failure(
    pool: &sqlx::PgPool,
    id: &str,
    affinity_hash: &str,
    completed_at: chrono::DateTime<Utc>,
) {
    seed_running_request(pool, id)
        .await
        .expect("seed continuation failure request");
    sqlx::query(
        "update model_requests
            set client_api_key_ref = 'key_continuation',
           client_transport = 'websocket',
           continuation_affinity_hash = $2,
           continuation_requested = true,
           outcome = 'failed',
           error_kind = 'continuation_recovery_required',
           started_at = $4 - interval '1 second',
           recovery_lease_expires_at = $4,
           completed_at = $4,
           request_observation_json = jsonb_strip_nulls(request_observation_json || jsonb_build_object(
           'continuation', coalesce(request_observation_json #> '{continuation}', '{}'::jsonb) || jsonb_build_object(
             'previousResponseIdHash', $3::text,
             'unavailableReason', 'reused_connection_lost')))
          where id = $1",
    )
    .bind(id)
    .bind(affinity_hash)
    .bind("b".repeat(64))
    .bind(completed_at)
    .execute(pool)
    .await
    .expect("finalize seeded continuation failure");
}

async fn seed_recovery_request(
    pool: &sqlx::PgPool,
    id: &str,
    affinity_hash: &str,
    started_at: chrono::DateTime<Utc>,
) {
    seed_running_request(pool, id)
        .await
        .expect("seed continuation recovery request");
    sqlx::query(
        "update model_requests
            set client_api_key_ref = 'key_continuation', client_transport = 'websocket',
                continuation_affinity_hash = $2, continuation_requested = false,
                started_at = $3, recovery_lease_expires_at = $3 + interval '1 minute'
          where id = $1",
    )
    .bind(id)
    .bind(affinity_hash)
    .bind(started_at)
    .execute(pool)
    .await
    .expect("prepare seeded recovery request");
}

async fn seed_websocket_transport_failure(
    pool: &sqlx::PgPool,
    id: &str,
    affinity_hash: &str,
    completed_at: chrono::DateTime<Utc>,
) {
    seed_running_request(pool, id)
        .await
        .expect("seed WebSocket transport failure request");
    sqlx::query(
        "update model_requests
            set client_api_key_ref = 'key_transport_recovery',
           client_transport = 'http_sse',
           continuation_affinity_hash = $2,
           attempt_count = 1,
           upstream_send_state = 'ambiguous',
           upstream_transport = 'websocket',
           outcome = 'failed',
           error_kind = 'upstream_unavailable',
           started_at = $3 - interval '1 second',
           recovery_lease_expires_at = $3,
           completed_at = $3,
           request_observation_json = jsonb_strip_nulls(request_observation_json || jsonb_build_object(
           'error', coalesce(request_observation_json #> '{error}', '{}'::jsonb) || jsonb_build_object(
             'providerErrorCode', 'websocket_close_1000')))
          where id = $1",
    )
    .bind(id)
    .bind(affinity_hash)
    .bind(completed_at)
    .execute(pool)
    .await
    .expect("finalize seeded WebSocket transport failure");
}

async fn seed_transport_recovery_request(
    pool: &sqlx::PgPool,
    id: &str,
    affinity_hash: &str,
    started_at: chrono::DateTime<Utc>,
) {
    seed_running_request(pool, id)
        .await
        .expect("seed transport recovery request");
    sqlx::query(
        "update model_requests
            set client_api_key_ref = 'key_transport_recovery',
                client_transport = 'http_sse',
                continuation_affinity_hash = $2,
                started_at = $3, recovery_lease_expires_at = $3 + interval '1 minute'
          where id = $1",
    )
    .bind(id)
    .bind(affinity_hash)
    .bind(started_at)
    .execute(pool)
    .await
    .expect("prepare seeded transport recovery request");
}

async fn seed_running_request(pool: &sqlx::PgPool, id: &str) -> Result<(), sqlx::Error> {
    sqlx::query(
        "insert into model_requests (
           id, client_api_key_ref, operation, client_transport, requested_model_id, provider_kind, provider_account_ref, cost_source, started_at, recovery_lease_expires_at, request_observation_json
         ) values (
           $1, 'key_status', 'generate', 'http_json', 'status-model', 'openai', 'acct_status', 'unavailable', now(), now() + interval '1 minute',
           jsonb_strip_nulls(jsonb_build_object(
           'request', jsonb_build_object(
             'configRevision', 1,
             'protocol', 'openai_responses',
             'endpoint', '/v1/responses',
             'compact', false),
           'routing', jsonb_build_object(
             'scope', 'all',
             'groupRefs', '{}'::text[],
             'groupNamesSnapshot', '[]'::jsonb)))
         )",
    )
    .bind(id)
    .execute(pool)
    .await?;
    Ok(())
}

#[tokio::test]
async fn diagnostic_trace_pressure_preserves_terminal_outcomes_and_usage() {
    let Some(database) = TestDatabase::create("diagnostic_trace").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let repository = super::observability_repository(&database.pool);
    for (id, outcome, expected_outcome) in [
        (
            "req_diagnostic_success",
            ExecutionOutcome::Succeeded,
            "succeeded",
        ),
        ("req_diagnostic_failed", ExecutionOutcome::Failed, "failed"),
    ] {
        seed_running_request(&database.pool, id).await.unwrap();
        let mut finalization = successful_core_finalization(id);
        finalization.outcome = outcome;
        finalization.usage.input_tokens = Some(100_000);
        finalization.usage.output_tokens = Some(2000);
        finalization.usage.total_tokens = Some(102_000);
        finalization.cost = CalculatedCost::from_usd_ticks(10_000_000_000)
            .unwrap()
            .into_estimate();
        finalization.timings.first_token_ms = Some(120);
        finalization.timings.latency_ms = Some(42_000);
        let trace = TraceContext::new(id);
        trace.record("request.routed", json!({"scope": "all"}));
        for stage in ["provider.request.input", "provider.request.outbound"] {
            trace.record_evidence(stage, json!({"hashes": "a".repeat(6000)}));
        }
        for index in 0..500 {
            trace.record(
                "stream.sample",
                json!({"index": index, "detail": "x".repeat(2000)}),
            );
        }
        let error_message = if outcome == ExecutionOutcome::Failed {
            finalization.client_status_code = Some(502);
            let message = "OpenAI WebSocket receive idle timeout after 300s";
            let provider_error = gateway_core::error::ProviderError::new(
                gateway_core::error::ProviderErrorKind::Timeout,
                UpstreamSendState::Ambiguous,
            )
            .with_diagnostic(
                gateway_core::error::ProviderDiagnostic::new(message)
                    .with_classification("receive", "receive_idle_timeout"),
            );
            trace.record_provider_failure(&provider_error);
            finalization.error = Some(GatewayError::from_provider(&provider_error));
            Some(message)
        } else {
            None
        };
        trace.record("request.finished", json!({"outcome": expected_outcome}));
        let snapshot = trace.snapshot().unwrap();
        let encoded = snapshot.to_string();
        assert!(encoded.len() > 64 * 1024);
        assert!(encoded.len() <= MAX_TRACE_SNAPSHOT_BYTES);
        assert!(snapshot["droppedEvents"].as_u64().unwrap() > 0);
        finalization.diagnostic_trace_json = Some(encoded);
        ExecutionStore::finalize_model_request(&store, finalization)
            .await
            .expect("trace pressure must not discard the terminal result");
        let persisted: Value = sqlx::query_scalar(
            "select jsonb_build_object(
               'outcome', outcome, 'input', input_tokens, 'output', output_tokens,
               'cost', cost_amount::text, 'firstToken', first_token_ms,
               'latency', latency_ms, 'error', error_message)
             from model_request_observations where id = $1",
        )
        .bind(id)
        .fetch_one(&database.pool)
        .await
        .unwrap();
        assert_eq!(
            persisted,
            json!({
                "outcome": expected_outcome, "input": 100_000, "output": 2000,
                "cost": "1.0000000000", "firstToken": 120, "latency": 42_000,
                "error": error_message,
            })
        );
        let detail = repository.usage_record_detail(id).await.unwrap();
        assert_eq!(detail.trace, Some(snapshot));
        assert!(detail.related_requests.is_empty());
    }
    assert_eq!(
        store
            .recover_expired_model_requests(Utc::now() + Duration::minutes(20))
            .await
            .unwrap()
            .requests,
        0,
    );
    database.close().await;
}

// 只构造已接纳的入口事实；未选择账号、未出站，也没有上游用量
pub(super) fn accepted_request(id: &str) -> CoreNewModelRequest {
    let started_at = SystemTime::from(
        DateTime::from_timestamp_micros(Utc::now().timestamp_micros())
            .expect("PostgreSQL timestamp precision"),
    );
    CoreNewModelRequest {
        id: ModelRequestId::new(id).expect("request id"),
        client_api_key_name_snapshot: None,
        client_api_key_ref: ClientApiKeyId::new("key_zero_attempt").expect("client key ref"),
        config_revision: ConfigRevision::new(1).expect("revision"),
        routing: AccountRoutingSnapshot::all(),
        protocol: "openai".to_owned(),
        operation: OperationKind::Generate,
        endpoint: "/v1/responses".to_owned(),
        client_transport: "http_sse".to_owned(),
        requested_model: Some(PublicModelId::new("coding").expect("public model")),
        client_ip: None,
        user_agent: None,
        reasoning_effort: None,
        reasoning_preset: None,
        request_kind: None,
        subagent_kind: None,
        compact: false,
        continuation: Default::default(),
        image_generation_requested: false,
        admission_decision_ms: Some(2),
        started_at,
        deadline_at: (started_at + StdDuration::from_secs(30)).into(),
    }
}

pub(super) fn early_failure(request: &CoreNewModelRequest) -> CoreModelRequestFinalization {
    let trace = TraceContext::new(request.id.as_str());
    trace.record("request.started", json!({"operation": "generate"}));
    // trace 的 index 是预备阶段关联，不证明已经拿到账号、建流或实际发送
    let preparation = trace.attempt(1);
    preparation.record(
        "attempt.started",
        json!({"provider": "openai", "model": "coding", "pinnedAccount": null}),
    );
    preparation.record(
        "attempt.failed",
        json!({"kind": "no_eligible_account", "sendState": "NotSent"}),
    );
    trace.record(
        "request.finished",
        json!({"outcome": "failed", "errorKind": "no_available_provider", "attemptCount": 0}),
    );
    CoreModelRequestFinalization {
        request_id: request.id.clone(),
        outcome: ExecutionOutcome::Failed,
        send_state: UpstreamSendState::NotSent,
        attempt_count: 0,
        downstream_committed_at: None,
        client_status_code: Some(503),
        client_response_id: None,
        upstream_status_code: None,
        upstream_request_id: None,
        upstream_response_id: None,
        upstream_transport: None,
        http_version: None,
        websocket_pool: None,
        service_tier: None,
        provider_metadata_json: None,
        diagnostic_trace_json: trace.snapshot().map(|value| value.to_string()),
        error: Some(GatewayError::new(
            GatewayErrorKind::NoAvailableProvider,
            "no available provider",
        )),
        provider_error_code: None,
        error_details: None,
        failure_observation: Default::default(),
        retry_after_ms: None,
        usage: Usage::new(),
        image_generation_succeeded: None,
        cost: CostEstimate::unavailable(),
        timings: CoreModelRequestTimings {
            latency_ms: Some(1_000),
            ..Default::default()
        },
        completed_at: request.started_at + StdDuration::from_secs(1),
    }
}

async fn stored_row(pool: &PgPool, request_id: &str) -> Value {
    sqlx::query_scalar("select to_jsonb(mr) from model_request_observations mr where id = $1")
        .bind(request_id)
        .fetch_one(pool)
        .await
        .expect("load complete synthetic request row")
}

fn zero_attempt_range(request: &CoreNewModelRequest) -> ObservabilityRange {
    ObservabilityRange::new(
        DateTime::from(request.started_at - StdDuration::from_secs(60)),
        DateTime::from(request.deadline_at.at().unwrap() + StdDuration::from_secs(60)),
    )
    .expect("observation range")
}

#[tokio::test]
async fn zero_attempt_failure_is_queryable_without_fabricating_upstream_facts() {
    let Some(database) = TestDatabase::create("zero_attempt_queries").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let request = accepted_request("req_zero_attempt_queries");
    store
        .create_model_request(request.clone())
        .await
        .expect("create accepted request without attempt");
    let finalization = early_failure(&request);
    let trace: Value = serde_json::from_str(
        finalization
            .diagnostic_trace_json
            .as_deref()
            .expect("trace snapshot"),
    )
    .expect("trace JSON");
    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .expect("finalize zero-attempt failure with trace");

    let repository = observability_repository(&database.pool);
    let detail = repository
        .usage_record_detail(request.id.as_str())
        .await
        .expect("load failed request detail");
    assert_eq!(detail.request.outcome.as_str(), "failed");
    assert_eq!(detail.request.attempt_count, 0);
    assert_eq!(detail.request.upstream_send_state, "not_sent");
    assert_eq!(detail.request.client_api_key_ref, "key_zero_attempt");
    assert_eq!(detail.request.requested_model_id.as_deref(), Some("coding"));
    assert_eq!(detail.request.admission_decision_ms, Some(2));
    assert_eq!(detail.request.latency_ms, Some(1_000));
    assert_eq!(detail.request.client_status_code, Some(503));
    assert_eq!(
        detail.request.error_kind.as_deref(),
        Some("no_available_provider")
    );
    assert_eq!(
        detail.request.error_message.as_deref(),
        Some("no available provider")
    );
    assert!(detail.attempts.is_empty());
    assert!(detail.related_requests.is_empty());
    assert_eq!(detail.trace, Some(trace.clone()));
    let row = stored_row(&database.pool, request.id.as_str()).await;
    for field in [
        "provider_kind",
        "provider_account_ref",
        "provider_account_name_snapshot",
        "provider_account_email_snapshot",
        "provider_account_authentication_kind_snapshot",
        "upstream_model_id",
        "upstream_transport",
        "http_version",
        "upstream_status_code",
        "upstream_request_id",
        "upstream_response_id",
        "client_response_id",
        "downstream_committed_at",
        "service_tier",
        "provider_observation_json",
        "error_details",
        "input_tokens",
        "output_tokens",
        "total_tokens",
        "cost_amount",
        "cost_currency",
        "account_selection_wait_ms",
        "capacity_used_slots",
        "capacity_total_slots",
    ] {
        assert_eq!(
            row.get(field),
            Some(&Value::Null),
            "{field} must remain absent"
        );
    }
    assert_eq!(row["cost_source"], "unavailable");
    let events = trace["events"].as_array().expect("events");
    assert!(events.iter().all(|event| event["exchangeId"].is_null()));
    let prepared_attempts: Vec<_> = events
        .iter()
        .filter(|event| event["attemptIndex"] == 1)
        .map(|event| event["stage"].as_str().expect("stage"))
        .collect();
    assert_eq!(prepared_attempts, ["attempt.started", "attempt.failed"]);

    let query = OpsErrorQuery {
        range: zero_attempt_range(&request),
        filter: OpsErrorFilter {
            request_id: Some(request.id.as_str().to_owned()),
            ..Default::default()
        },
        current_page: 1,
        page_size: ObservabilityPageSize::new(10).expect("page size"),
    };
    let errors = repository
        .list_ops_errors(query.clone())
        .await
        .expect("ops errors");
    assert_eq!(errors.total, 1);
    assert_eq!(errors.items.len(), 1);
    let error = &errors.items[0];
    assert_eq!(error.source, "model_request");
    assert_eq!(error.request_id.as_deref(), Some(request.id.as_str()));
    assert_eq!(error.attempt_index, None);
    assert_eq!(error.failure_kind, "no_available_provider");
    assert_eq!(error.message, "no available provider");
    assert_eq!(error.upstream_send_state.as_deref(), Some("not_sent"));
    assert_eq!(error.provider_kind, None);
    assert_eq!(error.provider_account_ref, None);
    assert_eq!(error.upstream_transport, None);
    assert_eq!(error.upstream_status_code, None);
    assert_eq!(error.upstream_request_id, None);
    assert_eq!(error.error_details, None);

    // 路由入口不等于已选择 Provider；没有平台事实时不能被平台/attempt 筛选命中
    for filter in [
        OpsErrorFilter {
            provider_kind: Some("openai".to_owned()),
            ..query.filter.clone()
        },
        OpsErrorFilter {
            attempt_index: Some(1),
            ..query.filter.clone()
        },
    ] {
        let page = repository
            .list_ops_errors(OpsErrorQuery {
                filter,
                ..query.clone()
            })
            .await
            .expect("filter absent upstream facts");
        assert_eq!(page.total, 0);
        assert!(page.items.is_empty());
    }
    let admin = admin_observability_store(&database.pool);
    let admin_detail = admin
        .usage_record_detail(request.id.as_str())
        .await
        .expect("admin detail");
    assert_eq!(admin_detail.trace, Some(trace));
    assert_eq!(admin_detail.request.attempt_count, 0);
    assert!(admin_detail.attempts.is_empty());
    let admin_errors = admin
        .list_ops_errors(admin_observability::OpsErrorQuery {
            range: admin_observability::TimeRange::new(query.range.start, query.range.end)
                .expect("admin time range"),
            filter: admin_observability::OpsErrorFilter {
                request_id: Some(request.id.as_str().to_owned()),
                ..Default::default()
            },
            current_page: 1,
            page_size: ObservabilityPageSize::new(10).expect("page size"),
        })
        .await
        .expect("admin ops errors without provider facts");
    assert_eq!(admin_errors.total, 1);
    assert_eq!(admin_errors.items.len(), 1);
    assert_eq!(admin_errors.items[0].attempt_index, None);
    assert_eq!(admin_errors.items[0].provider_kind, None);
    database.close().await;
}

#[tokio::test]
async fn zero_attempt_finalization_rejects_sent_or_missing_attempt_facts_atomically() {
    let Some(database) = TestDatabase::create("zero_attempt_constraints").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let request = accepted_request("req_zero_attempt_constraints");
    store
        .create_model_request(request.clone())
        .await
        .expect("create request");
    let before = stored_row(&database.pool, request.id.as_str()).await;

    for (count, send_state) in [
        (0, UpstreamSendState::Sent),
        (0, UpstreamSendState::Ambiguous),
        (1, UpstreamSendState::NotSent),
    ] {
        let mut finalization = early_failure(&request);
        finalization.attempt_count = count;
        finalization.send_state = send_state;
        assert!(
            ExecutionStore::finalize_model_request(&store, finalization)
                .await
                .is_err()
        );
        assert_eq!(
            stored_row(&database.pool, request.id.as_str()).await,
            before
        );
    }
    for invalid_trace in [
        "not JSON".to_owned(),
        "[]".to_owned(),
        json!({"data": "x".repeat(MAX_TRACE_SNAPSHOT_BYTES)}).to_string(),
    ] {
        let mut finalization = early_failure(&request);
        finalization.diagnostic_trace_json = Some(invalid_trace);
        assert_eq!(
            ExecutionStore::finalize_model_request(&store, finalization)
                .await
                .expect_err("invalid trace")
                .kind(),
            StoreErrorKind::InvalidData,
        );
        assert_eq!(
            stored_row(&database.pool, request.id.as_str()).await,
            before
        );
    }
    ExecutionStore::finalize_model_request(&store, early_failure(&request))
        .await
        .expect("valid zero attempt");
    database.close().await;
}

#[tokio::test]
async fn zero_attempt_duplicate_create_and_finalization_never_overwrite_terminal_facts() {
    let Some(database) = TestDatabase::create("zero_attempt_idempotency").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let request = accepted_request("req_zero_attempt_idempotency");
    store
        .create_model_request(request.clone())
        .await
        .expect("create request");
    let (first, second) = tokio::join!(
        ExecutionStore::finalize_model_request(&store, early_failure(&request)),
        ExecutionStore::finalize_model_request(&store, early_failure(&request)),
    );
    assert_eq!(usize::from(first.is_ok()) + usize::from(second.is_ok()), 1);
    assert_eq!(
        first
            .err()
            .or_else(|| second.err())
            .expect("CAS loser")
            .kind(),
        StoreErrorKind::InvalidState
    );
    let terminal = stored_row(&database.pool, request.id.as_str()).await;
    assert!(store.create_model_request(request.clone()).await.is_err());
    let mut late = early_failure(&request);
    late.outcome = ExecutionOutcome::Cancelled;
    late.error = Some(GatewayError::new(
        GatewayErrorKind::Cancelled,
        "request was cancelled",
    ));
    late.diagnostic_trace_json = None;
    late.completed_at += StdDuration::from_secs(1);
    assert_eq!(
        ExecutionStore::finalize_model_request(&store, late)
            .await
            .expect_err("already finalized")
            .kind(),
        StoreErrorKind::InvalidState
    );
    assert_eq!(
        store
            .recover_expired(request.deadline_at.at().unwrap())
            .await
            .expect("recover")
            .requests,
        0
    );
    assert_eq!(
        stored_row(&database.pool, request.id.as_str()).await,
        terminal
    );
    let count: i64 = sqlx::query_scalar("select count(*) from model_requests")
        .fetch_one(&database.pool)
        .await
        .expect("one request row");
    assert_eq!(count, 1);
    let ops: i64 = sqlx::query_scalar("select count(*) from ops_events")
        .fetch_one(&database.pool)
        .await
        .expect("no extra error ledger");
    assert_eq!(ops, 0);
    database.close().await;
}

#[tokio::test]
async fn zero_attempt_recovery_respects_deadline_and_does_not_invent_a_trace() {
    let Some(database) = TestDatabase::create("zero_attempt_recovery").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let request = accepted_request("req_zero_attempt_recovery");
    let mut active = request.clone();
    active.id = ModelRequestId::new("req_zero_attempt_still_running").expect("request id");
    active.deadline_at = (active.deadline_at.at().unwrap() + StdDuration::from_secs(30)).into();
    store
        .create_model_request(request.clone())
        .await
        .expect("create interrupted request");
    store
        .create_model_request(active.clone())
        .await
        .expect("create active request");
    assert_eq!(
        store
            .recover_expired(request.deadline_at.at().unwrap() - StdDuration::from_micros(1))
            .await
            .expect("before deadline")
            .requests,
        0
    );
    assert_eq!(
        store
            .recover_expired(request.deadline_at.at().unwrap())
            .await
            .expect("at deadline")
            .requests,
        1
    );
    let recovered = stored_row(&database.pool, request.id.as_str()).await;
    assert_eq!(
        store
            .recover_expired(request.deadline_at.at().unwrap())
            .await
            .expect("repeat recovery")
            .requests,
        0
    );
    assert_eq!(
        ExecutionStore::finalize_model_request(&store, early_failure(&request))
            .await
            .expect_err("late finalize after recovery")
            .kind(),
        StoreErrorKind::InvalidState
    );
    assert_eq!(
        stored_row(&database.pool, request.id.as_str()).await,
        recovered
    );
    let repository = observability_repository(&database.pool);
    let detail = repository
        .usage_record_detail(request.id.as_str())
        .await
        .expect("recovered detail");
    assert_eq!(detail.request.outcome.as_str(), "incomplete");
    assert_eq!(
        detail.request.error_kind.as_deref(),
        Some("process_interrupted")
    );
    assert_eq!(
        detail.request.completed_at,
        Some(DateTime::from(request.deadline_at.at().unwrap()))
    );
    assert_eq!(detail.request.attempt_count, 0);
    assert_eq!(detail.request.upstream_send_state, "not_sent");
    assert_eq!(detail.trace, None);
    assert!(detail.attempts.is_empty());
    let errors = repository
        .list_ops_errors(OpsErrorQuery {
            range: zero_attempt_range(&request),
            filter: OpsErrorFilter::default(),
            current_page: 1,
            page_size: ObservabilityPageSize::new(10).expect("page size"),
        })
        .await
        .expect("recovered ops error");
    assert_eq!(errors.total, 1);
    assert_eq!(errors.items[0].failure_kind, "process_interrupted");
    assert_eq!(errors.items[0].attempt_index, None);
    assert_eq!(
        repository
            .usage_record_detail(active.id.as_str())
            .await
            .expect("active detail")
            .request
            .outcome
            .as_str(),
        "running"
    );
    database.close().await;
}

#[tokio::test]
async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates() {
    let Some(database) = TestDatabase::create("zero_attempt_metrics").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let request = accepted_request("req_zero_attempt_metrics");
    seed_running_request(&database.pool, "req_zero_attempt_success_control")
        .await
        .expect("success control");
    let mut success = successful_core_finalization("req_zero_attempt_success_control");
    success.downstream_committed_at = Some(success.completed_at);
    success.usage = Usage {
        input_tokens: Some(100),
        output_tokens: Some(20),
        total_tokens: Some(120),
        ..Usage::new()
    };
    success.timings.latency_ms = Some(200);
    success.cost = gateway_core::metering::CalculatedCost::from_usd_ticks(12_500_000_000)
        .expect("calculated cost")
        .into_estimate();
    ExecutionStore::finalize_model_request(&store, success)
        .await
        .expect("finalize success control");
    let repository = observability_repository(&database.pool);
    super::drain_analytics(&database.pool).await;
    let before = repository
        .usage_insights_observation(
            zero_attempt_range(&request),
            UsageRecordFilter::default(),
            gateway_admin::model::observability::Granularity::Hour,
            true,
        )
        .await
        .map(|snapshot| snapshot.overview)
        .expect("baseline usage");
    assert_eq!(before.requests.total_tokens, 120);
    assert!(!before.attempts.costs.is_empty());
    store
        .create_model_request(request.clone())
        .await
        .expect("create zero attempt");
    ExecutionStore::finalize_model_request(&store, early_failure(&request))
        .await
        .expect("finalize failure");
    super::drain_analytics(&database.pool).await;
    let after = repository
        .usage_insights_observation(
            zero_attempt_range(&request),
            UsageRecordFilter::default(),
            gateway_admin::model::observability::Granularity::Hour,
            true,
        )
        .await
        .map(|snapshot| snapshot.overview)
        .expect("usage after failure");
    assert_eq!(
        after.requests.request_count,
        before.requests.request_count + 1
    );
    assert_eq!(
        after.requests.failure_count,
        before.requests.failure_count + 1
    );
    assert_eq!(after.requests.success_count, before.requests.success_count);
    assert_eq!(after.requests.total_tokens, before.requests.total_tokens);
    assert_eq!(
        after.requests.latency_sum_ms,
        before.requests.latency_sum_ms
    );
    assert_eq!(after.requests.latency_count, before.requests.latency_count);
    assert_eq!(after.attempts, before.attempts);
    let successes = repository
        .usage_insights_observation(
            zero_attempt_range(&request),
            UsageRecordFilter {
                outcome: Some("succeeded".to_owned()),
                ..Default::default()
            },
            gateway_admin::model::observability::Granularity::Hour,
            true,
        )
        .await
        .map(|snapshot| snapshot.overview)
        .expect("successful aggregation unchanged");
    assert_eq!(successes.range, before.range);
    assert_eq!(successes.requests, before.requests);
    assert_eq!(successes.attempts, before.attempts);
    assert_eq!(successes.providers, before.providers);
    assert_eq!(
        successes.analytics.schema_version,
        before.analytics.schema_version
    );
    assert_eq!(
        successes.analytics.histogram_version,
        before.analytics.histogram_version
    );
    assert!(successes.analytics.generation >= before.analytics.generation);
    assert!(successes.analytics.generated_at >= before.analytics.generated_at);
    assert!(successes.analytics.coverage_start.is_some());
    assert!(successes.analytics.coverage_end >= before.analytics.coverage_end);
    let filter = UsageRecordFilter {
        request_id: Some(request.id.as_str().to_owned()),
        ..Default::default()
    };
    let failed = repository
        .usage_insights_observation(
            zero_attempt_range(&request),
            filter.clone(),
            gateway_admin::model::observability::Granularity::Hour,
            true,
        )
        .await
        .map(|snapshot| snapshot.overview)
        .expect("failed request metrics");
    assert_eq!(failed.requests.request_count, 1);
    assert_eq!(failed.requests.failure_count, 1);
    assert_eq!(failed.requests.total_tokens, 0);
    assert_eq!(failed.requests.latency_count, 0);
    assert_eq!(
        failed.attempts,
        admin_observability::AttemptMetrics::default()
    );
    let page = repository
        .list_usage_records(UsageRecordQuery {
            status: Default::default(),
            range: zero_attempt_range(&request),
            filter: filter.clone(),
            current_page: 1,
            page_size: ObservabilityPageSize::new(10).expect("page size"),
        })
        .await
        .expect("successful usage list excludes failure");
    assert_eq!(page.total, 0);
    assert!(page.items.is_empty());
    super::drain_analytics(&database.pool).await;
    let insights = repository
        .usage_insights_observation(
            zero_attempt_range(&request),
            filter,
            admin_observability::Granularity::FifteenMinutes,
            true,
        )
        .await
        .expect("billing projection");
    assert!(
        insights
            .costs
            .iter()
            .all(|point| { point.standard_amount.is_none() && point.no_cache_amount.is_none() })
    );
    let charges: i64 = sqlx::query_scalar("select count(*) from client_api_key_charge_events")
        .fetch_one(&database.pool)
        .await
        .expect("observations do not settle budgets");
    assert_eq!(charges, 0);
    database.close().await;
}

#[tokio::test]
async fn entry_rejection_is_visible_without_a_fictitious_model_execution() {
    let Some(database) = TestDatabase::create("entry_rejection").await else {
        return;
    };
    let repository = PgExecutionStore::new(database.pool.clone());
    repository
        .record_entry_rejection(gateway_core::engine::EntryRejection {
            request_id: ModelRequestId::new("req_entry_rejection").unwrap(),
            client_key_id: ClientApiKeyId::new("key_entry").unwrap(),
            error: GatewayError::new(GatewayErrorKind::NoAvailableProvider, "no route"),
            latency: StdDuration::from_millis(12),
        })
        .await
        .unwrap();
    let row: (Option<String>, String, String, i64) = sqlx::query_as(
        "select model_request_id, failure_kind, message, latency_ms from ops_events where component = 'request_entry' and operation = 'reject'"
    ).fetch_one(&database.pool).await.unwrap();
    assert!(row.0.is_none());
    assert_eq!(row.1, GatewayErrorKind::NoAvailableProvider.as_str());
    assert_eq!(
        serde_json::from_str::<Value>(&row.2).unwrap()["requestId"],
        "req_entry_rejection"
    );
    assert_eq!(row.3, 12);
    let count: i64 = sqlx::query_scalar("select count(*) from model_requests")
        .fetch_one(&database.pool)
        .await
        .unwrap();
    assert_eq!(count, 0);
    database.close().await;
}

#[tokio::test]
async fn shared_core_usage_and_timings_preserve_fields_and_reject_invalid_phases() {
    let Some(database) = TestDatabase::create("shared_execution_facts").await else {
        return;
    };
    let id = "req_shared_facts";
    seed_running_request(&database.pool, id).await.unwrap();
    let store = PgExecutionStore::new(database.pool.clone());
    let mut finalization = successful_core_finalization(id);
    finalization.usage = Usage {
        input_tokens: Some(101),
        output_tokens: Some(22),
        cached_input_tokens: Some(33),
        cache_write_input_tokens: Some(44),
        reasoning_output_tokens: Some(5),
        image_input_tokens: Some(6),
        image_output_tokens: Some(7),
        total_tokens: Some(123),
    };
    finalization.timings = CoreModelRequestTimings {
        transport_decision_wait_ms: Some(1),
        connect_ms: Some(2),
        headers_ms: Some(3),
        first_event_ms: Some(4),
        first_reasoning_ms: Some(5),
        first_text_ms: Some(6),
        first_token_ms: Some(7),
        provider_processing_ms: Some(8),
        upstream_response_ms: Some(1_000),
        upstream_api_overhead_ms: Some(120.25),
        upstream_engine_ms: Some(6400.0),
        latency_ms: Some(9),
    };
    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .unwrap();
    let persisted = stored_row(&database.pool, id).await;
    for (field, expected) in [
        ("input_tokens", 101),
        ("output_tokens", 22),
        ("cached_input_tokens", 33),
        ("cache_write_input_tokens", 44),
        ("reasoning_output_tokens", 5),
        ("image_input_tokens", 6),
        ("image_output_tokens", 7),
        ("total_tokens", 123),
        ("transport_decision_wait_ms", 1),
        ("connect_ms", 2),
        ("headers_ms", 3),
        ("first_event_ms", 4),
        ("first_reasoning_ms", 5),
        ("first_text_ms", 6),
        ("first_token_ms", 7),
        ("provider_processing_ms", 8),
        ("upstream_response_ms", 1_000),
        ("latency_ms", 9),
    ] {
        assert_eq!(persisted[field], json!(expected), "{field}");
    }
    assert_eq!(persisted["upstream_api_overhead_ms"].as_f64(), Some(120.25));
    assert_eq!(persisted["upstream_engine_ms"].as_f64(), Some(6400.0));
    let invalid_id = "req_invalid_phase";
    seed_running_request(&database.pool, invalid_id)
        .await
        .unwrap();
    let before = stored_row(&database.pool, invalid_id).await;
    let mut invalid = successful_core_finalization(invalid_id);
    invalid.timings.latency_ms = Some(10);
    invalid.timings.first_token_ms = Some(11);
    let error = ExecutionStore::finalize_model_request(&store, invalid)
        .await
        .unwrap_err();
    assert_eq!(error.kind(), StoreErrorKind::InvalidData);
    assert_eq!(stored_row(&database.pool, invalid_id).await, before);
    for invalid_ms in [-1.0, f64::NAN, f64::INFINITY, i64::MAX as f64] {
        let mut invalid = successful_core_finalization(invalid_id);
        invalid.timings.upstream_engine_ms = Some(invalid_ms);
        let error = ExecutionStore::finalize_model_request(&store, invalid)
            .await
            .expect_err("invalid official timing must not silently become missing");
        assert_eq!(error.kind(), StoreErrorKind::InvalidData);
        assert_eq!(stored_row(&database.pool, invalid_id).await, before);
    }
    database.close().await;
}

#[tokio::test]
async fn retry_after_transport_commit_preserves_delivery_facts_and_attempt_order() {
    let Some(database) = TestDatabase::create("retry_after_commit").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    for (transport, status) in [("websocket", None), ("http_sse", Some(200))] {
        let id = format!("req_retry_after_{transport}_commit");
        let mut request = accepted_request(&id);
        request.client_transport = transport.to_owned();
        store.create_model_request(request).await.unwrap();
        let mut attempt = ModelRequestAttemptStart {
            model_request_id: id.clone(),
            attempt_count: 1,
            provider_kind: "openai".to_owned(),
            account_snapshot: None,
            provider_account_ref: Some("account_before_retry".to_owned()),
            upstream_model_id: Some("model".to_owned()),
            upstream_transport: "websocket".to_owned(),
            http_version: Some("HTTP/1.1".to_owned()),
            account_selection_wait_ms: Some(5),
            capacity_used_slots: Some(1),
            capacity_total_slots: Some(10),
        };
        store
            .begin_model_request_attempt(attempt.clone())
            .await
            .unwrap();
        store
            .mark_upstream_send_state(&id, gateway_store::postgres::UpstreamSendState::Sent)
            .await
            .unwrap();
        // 可重放的 WS 前导事件或 HTTP 心跳先交付，不能替代 Core 的业务重放屏障
        let committed_at = Utc::now();
        ModelRequestRepository::mark_downstream_committed(&store, &id, committed_at, status)
            .await
            .unwrap();
        let committed = stored_row(&database.pool, &id).await;
        attempt.attempt_count = 2;
        attempt.provider_account_ref = Some("account_after_retry".to_owned());
        attempt.upstream_transport = "http_sse".to_owned();
        attempt.http_version = Some("HTTP/2".to_owned());
        attempt.account_selection_wait_ms = Some(7);
        attempt.capacity_used_slots = Some(2);
        assert_eq!(
            store
                .begin_model_request_attempt(attempt.clone())
                .await
                .unwrap(),
            2
        );
        let retried = stored_row(&database.pool, &id).await;
        assert_eq!(
            retried["downstream_committed_at"],
            committed["downstream_committed_at"]
        );
        assert_eq!(
            retried["client_status_code"],
            committed["client_status_code"]
        );
        assert_eq!(retried["upstream_send_state"], "sent");
        assert_eq!(retried["provider_account_ref"], "account_after_retry");
        assert_eq!(retried["upstream_transport"], "http_sse");
        assert_eq!(
            retried["request_observation_json"]["scheduling"]["accountSelectionWaitMs"],
            12
        );
        assert_eq!(
            retried["request_observation_json"]["scheduling"]["capacityUsedSlots"],
            2
        );
        assert_eq!(
            retried["request_observation_json"]["transport"]["httpVersion"],
            "HTTP/2"
        );
        // 已接收的序号与跳号都不能覆盖新 attempt 的账号和计时
        for count in [2, 4] {
            attempt.attempt_count = count;
            let error = store
                .begin_model_request_attempt(attempt.clone())
                .await
                .unwrap_err();
            assert!(matches!(
                error,
                gateway_store::StoreError::Conflict {
                    kind: gateway_store::ConflictKind::InvalidTransition,
                    ..
                }
            ));
            assert_eq!(stored_row(&database.pool, &id).await, retried);
        }
        let mut finalization = successful_core_finalization(&id);
        finalization.attempt_count = 2;
        finalization.downstream_committed_at = Some(committed_at.into());
        finalization.client_status_code = status;
        finalization.upstream_transport = Some("http_sse".to_owned());
        ExecutionStore::finalize_model_request(&store, finalization)
            .await
            .unwrap();
        let terminal = stored_row(&database.pool, &id).await;
        assert_eq!(terminal["outcome"], "succeeded");
        assert_eq!(terminal["attempt_count"], 2);
        assert_eq!(
            terminal["request_observation_json"]["scheduling"],
            retried["request_observation_json"]["scheduling"]
        );
        attempt.attempt_count = 3;
        assert!(store.begin_model_request_attempt(attempt).await.is_err());
        assert_eq!(stored_row(&database.pool, &id).await, terminal);
    }
    database.close().await;
}

#[tokio::test]
async fn concurrent_retry_updates_keep_request_context_and_clear_previous_transport() {
    let Some(database) = TestDatabase::create("observation_retry_cas").await else {
        return;
    };
    let store = PgExecutionStore::new(database.pool.clone());
    let request = accepted_request("req_observation_retry");
    store.create_model_request(request.clone()).await.unwrap();
    let before = stored_row(&database.pool, request.id.as_str()).await;
    let mut attempt = ModelRequestAttemptStart {
        model_request_id: request.id.as_str().to_owned(),
        attempt_count: 1,
        provider_kind: "openai".to_owned(),
        account_snapshot: None,
        provider_account_ref: Some("deleted_account".to_owned()),
        upstream_model_id: Some("model".to_owned()),
        upstream_transport: "http_sse".to_owned(),
        http_version: Some("HTTP/1.1".to_owned()),
        account_selection_wait_ms: Some(5),
        capacity_used_slots: Some(1),
        capacity_total_slots: Some(10),
    };
    store
        .begin_model_request_attempt(attempt.clone())
        .await
        .unwrap();
    sqlx::query("update model_requests set request_observation_json = request_observation_json || $1::jsonb where id = $2")
        .bind(sqlx::types::Json(
            json!({"transport": {"httpVersion":"HTTP/1.1", "connection": {
                "id":"previous", "exitReason":"peer_closed", "ageMs":10, "idleMs":0
            }}}),
        ))
        .bind(request.id.as_str())
        .execute(&database.pool)
        .await
        .unwrap();
    attempt.attempt_count = 2;
    attempt.http_version = Some("HTTP/2".to_owned());
    attempt.account_selection_wait_ms = Some(7);
    attempt.capacity_used_slots = Some(2);
    let (first, second) = tokio::join!(
        store.begin_model_request_attempt(attempt.clone()),
        store.begin_model_request_attempt(attempt),
    );
    assert_eq!(usize::from(first.is_ok()) + usize::from(second.is_ok()), 1);
    let retried = stored_row(&database.pool, request.id.as_str()).await;
    let observation = &retried["request_observation_json"];
    assert_eq!(
        observation["request"],
        before["request_observation_json"]["request"]
    );
    assert_eq!(
        observation["routing"],
        before["request_observation_json"]["routing"]
    );
    assert_eq!(observation["scheduling"]["admissionDecisionMs"], 2);
    assert_eq!(observation["scheduling"]["accountSelectionWaitMs"], 12);
    assert_eq!(observation["scheduling"]["capacityUsedSlots"], 2);
    assert_eq!(observation["transport"], json!({"httpVersion":"HTTP/2"}));
    let mut finalization = successful_core_finalization(request.id.as_str());
    finalization.attempt_count = 2;
    finalization.http_version = None;
    finalization.timings.upstream_response_ms = Some(1_000);
    ExecutionStore::finalize_model_request(&store, finalization)
        .await
        .unwrap();
    let terminal = stored_row(&database.pool, request.id.as_str()).await;
    assert_eq!(
        terminal["request_observation_json"]["scheduling"],
        observation["scheduling"]
    );
    assert_eq!(
        terminal["request_observation_json"]["request"],
        observation["request"]
    );
    assert_eq!(
        terminal["request_observation_json"]["transport"]["httpVersion"],
        "HTTP/2"
    );
    assert_eq!(
        terminal["request_observation_json"]["timings"]["upstream"]["responseMs"],
        1_000
    );
    database.close().await;
}