#![allow(dead_code, unused_imports)]

use super::common::*;

use std::sync::{Arc, Mutex};

use axum::{Json, Router, http::HeaderMap, routing::post};
use crank_core::{AuditError, AuditEvent, AuditSink, PlatformApiKeyKind, SecretKind};
use serde_json::{Value, json};
use serial_test::serial;
use tokio::net::TcpListener;

#[derive(Clone, Default)]
struct RecordingAuditSink {
    events: Arc<Mutex<Vec<AuditEvent>>>,
}

impl RecordingAuditSink {
    fn events(&self) -> Vec<AuditEvent> {
        self.events.lock().unwrap().clone()
    }
}

#[async_trait::async_trait]
impl AuditSink for RecordingAuditSink {
    async fn record(&self, event: AuditEvent) -> Result<(), AuditError> {
        self.events.lock().unwrap().push(event);
        Ok(())
    }
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn machine_key_raw_value_is_create_only_and_duplicate_name_is_typed_conflict() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("credential_lifecycle_keys");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let agent_id = create_agent(&client, &base_url, "credential-lifecycle-agent").await;

    let created = post_json(
        &client,
        format!("{base_url}/agents/{agent_id}/platform-api-keys"),
        json!({"name": "primary-mcp-key", "key_kind": PlatformApiKeyKind::McpClient, "scopes": ["read", "write"]}),
    )
    .await;
    let raw_key = created["secret"]
        .as_str()
        .expect("create response must disclose the raw key once")
        .to_owned();
    let random_part = raw_key
        .strip_prefix("crk_")
        .expect("MCP key must use the MCP bearer marker");
    assert_eq!(random_part.len(), 43, "32 raw bytes in unpadded base64url");
    assert!(
        random_part
            .bytes()
            .all(|byte| { byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_') })
    );
    assert_eq!(
        created["connection"]["endpoint"],
        "http://localhost:3000/mcp/v1/default/credential-lifecycle-agent"
    );
    assert!(
        created["connection"]["clients"]
            .as_array()
            .is_some_and(|clients| !clients.is_empty())
    );

    let approval = post_json(
        &client,
        format!("{base_url}/agents/{agent_id}/platform-api-keys"),
        json!({"name": "approval-without-mcp-config", "key_kind": PlatformApiKeyKind::Approval, "scopes": ["read_pending"]}),
    )
    .await;
    assert!(approval.get("connection").is_none());

    let listed_text = client
        .get(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(!listed_text.contains(&raw_key));
    assert!(!listed_text.contains("secret_hash"));
    assert!(!listed_text.contains("\"secret\""));

    let duplicate = client
        .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
        .json(&json!({
            "name": "primary-mcp-key",
            "key_kind": PlatformApiKeyKind::McpClient,
            "scopes": ["read"]
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(duplicate.status(), reqwest::StatusCode::CONFLICT);
    let duplicate_body = duplicate.text().await.unwrap();
    assert!(!duplicate_body.contains(&raw_key));
    assert!(!duplicate_body.contains("secret_hash"));
    assert!(!duplicate_body.contains("duplicate key value"));
    let duplicate_json: Value = serde_json::from_str(&duplicate_body).unwrap();
    assert_eq!(
        duplicate_json["error"]["code"],
        "platform_api_key_name_conflict"
    );

    for payload in [
        json!({"name": "n".repeat(129), "key_kind": PlatformApiKeyKind::McpClient, "scopes": ["read"]}),
        json!({"name": "duplicate-scope", "key_kind": PlatformApiKeyKind::McpClient, "scopes": ["read", "read"]}),
        json!({"name": "past-expiry", "key_kind": PlatformApiKeyKind::McpClient, "scopes": ["read"], "expires_at": "2020-01-01T00:00:00Z"}),
    ] {
        assert_post_error(
            &client,
            format!("{base_url}/agents/{agent_id}/platform-api-keys"),
            payload,
            reqwest::StatusCode::BAD_REQUEST,
            "validation_error",
        )
        .await;
    }

    let canonical = post_json(
        &client,
        format!("{base_url}/agents/{agent_id}/platform-api-keys"),
        json!({"name": "canonical-approval-origin", "key_kind": PlatformApiKeyKind::Approval, "scopes": ["read_pending"], "allowed_origins": ["HTTPS://Example.COM:443/", "http://EXAMPLE.com:80/"]}),
    )
    .await;
    assert_eq!(
        canonical["api_key"]["api_key"]["allowed_origins"],
        json!(["https://example.com", "http://example.com"])
    );
    for (index, origin) in [
        "https://[::1",
        "https://example.test:bad/",
        "https://user:pass@example.test/",
        "https://example.test/path",
        "https://example.test/?token=origin-canary",
        "https://example.test/#fragment",
    ]
    .into_iter()
    .enumerate()
    {
        assert_post_error(
            &client,
            format!("{base_url}/agents/{agent_id}/platform-api-keys"),
            json!({"name": format!("malformed-origin-{index}"), "key_kind": PlatformApiKeyKind::Approval, "scopes": ["read_pending"], "allowed_origins": [origin]}),
            reqwest::StatusCode::BAD_REQUEST,
            "validation_error",
        )
        .await;
    }
    assert_post_error(
        &client,
        format!("{base_url}/agents/{agent_id}/platform-api-keys"),
        json!({"name": "duplicate-canonical-origin", "key_kind": PlatformApiKeyKind::Approval, "scopes": ["read_pending"], "allowed_origins": ["https://EXAMPLE.test:443/", "https://example.test/"]}),
        reqwest::StatusCode::BAD_REQUEST,
        "validation_error",
    )
    .await;
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn deleted_machine_key_name_can_be_reused_and_terminal_state_cannot_regress() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("credential_lifecycle_key_reuse");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let agent_id = create_agent(&client, &base_url, "credential-key-reuse-agent").await;

    let first = assert_success_json(
        client
            .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
            .json(&json!({
                "name": "replaceable-key",
                "key_kind": PlatformApiKeyKind::McpClient,
                "scopes": ["read"]
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let first_key_id = first["api_key"]["api_key"]["id"].as_str().unwrap();

    let delete_response = client
        .delete(format!(
            "{base_url}/agents/{agent_id}/platform-api-keys/{first_key_id}"
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(delete_response.status(), reqwest::StatusCode::NO_CONTENT);

    let revoke_deleted = client
        .post(format!(
            "{base_url}/agents/{agent_id}/platform-api-keys/{first_key_id}/revoke"
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(revoke_deleted.status(), reqwest::StatusCode::CONFLICT);
    let revoke_body: Value = serde_json::from_str(&revoke_deleted.text().await.unwrap()).unwrap();
    assert_eq!(revoke_body["error"]["code"], "platform_api_key_not_active");

    let second = assert_success_json(
        client
            .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
            .json(&json!({
                "name": "replaceable-key",
                "key_kind": PlatformApiKeyKind::McpClient,
                "scopes": ["read"]
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let second_key_id = second["api_key"]["api_key"]["id"].as_str().unwrap();
    assert_ne!(first_key_id, second_key_id);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn secret_plaintext_is_write_only_and_auth_profile_stores_reference() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("credential_lifecycle_secrets");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;
    let canary = "secret-canary-1-10-do-not-leak";

    let created_response = client
        .post(format!("{base_url}/secrets"))
        .json(&json!({
            "name": "crm-bearer-token",
            "kind": SecretKind::Token,
            "value": { "token": canary }
        }))
        .send()
        .await
        .unwrap();
    let created_text = created_response.text().await.unwrap();
    assert!(!created_text.contains(canary));
    assert!(!created_text.contains("ciphertext"));
    let created: Value = serde_json::from_str(&created_text).unwrap();
    let secret_id = created["id"].as_str().unwrap();

    let listed_text = client
        .get(format!("{base_url}/secrets"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(!listed_text.contains(canary));
    assert!(!listed_text.contains("ciphertext"));

    let fetched_text = client
        .get(format!("{base_url}/secrets/{secret_id}"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(!fetched_text.contains(canary));
    assert!(!fetched_text.contains("ciphertext"));

    let auth_profile_text = client
        .post(format!("{base_url}/auth-profiles"))
        .json(&json!({
            "name": "crm-bearer",
            "kind": "bearer",
            "config": {
                "bearer": {
                    "header_name": "Authorization",
                    "secret_id": secret_id
                }
            }
        }))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(!auth_profile_text.contains(canary));
    assert!(!auth_profile_text.contains("ciphertext"));
    let auth_profile: Value = serde_json::from_str(&auth_profile_text).unwrap();
    assert_eq!(auth_profile["config"]["bearer"]["secret_id"], secret_id);

    for (name, kind, value) in [
        ("null-token", SecretKind::Token, Value::Null),
        (
            "wrong-token-shape",
            SecretKind::Token,
            json!({"token": "token", "extra": "field"}),
        ),
        (
            "wrong-header-shape",
            SecretKind::Header,
            json!({"header_name": "X-Token"}),
        ),
        (
            "wrong-username-password-shape",
            SecretKind::UsernamePassword,
            json!({"username": "user"}),
        ),
    ] {
        assert_post_error(
            &client,
            format!("{base_url}/secrets"),
            json!({"name": name, "kind": kind, "value": value}),
            reqwest::StatusCode::BAD_REQUEST,
            "validation_error",
        )
        .await;
    }
    assert_post_error(
        &client,
        format!("{base_url}/secrets"),
        json!({"name": "oversized-secret", "kind": SecretKind::Generic, "value": "x".repeat(65_537)}),
        reqwest::StatusCode::BAD_REQUEST,
        "validation_error",
    )
    .await;
    for value in [
        Value::Null,
        json!({"token": "token", "extra": "field"}),
        json!("x".repeat(65_537)),
    ] {
        assert_post_error(
            &client,
            format!("{base_url}/secrets/{secret_id}/rotate"),
            json!({"value": value}),
            reqwest::StatusCode::BAD_REQUEST,
            "validation_error",
        )
        .await;
    }
    let current = assert_success_json(
        client
            .get(format!("{base_url}/secrets/{secret_id}"))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(current["current_version"], 1);

    for payload in [
        json!({"name": "", "kind": "bearer", "config": {"bearer": {"header_name": "Authorization", "secret_id": secret_id}}}),
        json!({"name": "n".repeat(129), "kind": "bearer", "config": {"bearer": {"header_name": "Authorization", "secret_id": secret_id}}}),
        json!({"name": "invalid-header-name", "kind": "bearer", "config": {"bearer": {"header_name": "Bad Header", "secret_id": secret_id}}}),
        json!({"name": "invalid-query-name", "kind": "api_key_query", "config": {"api_key_query": {"param_name": "bad name", "secret_id": secret_id}}}),
    ] {
        assert_post_error(
            &client,
            format!("{base_url}/auth-profiles"),
            payload,
            reqwest::StatusCode::BAD_REQUEST,
            "validation_error",
        )
        .await;
    }
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn credential_mutations_emit_bounded_non_secret_audit_events() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("credential_lifecycle_audit");
    let audit_sink = RecordingAuditSink::default();
    let base_url = spawn_admin_api(build_test_app_with_audit_sink(
        registry,
        storage_root,
        Arc::new(audit_sink.clone()),
    ))
    .await;
    let client = authorized_client(&base_url).await;
    let request_id = "req_credential_audit_1";
    let traceparent = "00-11111111111111111111111111111111-2222222222222222-01";

    let secret = assert_success_json(
        client
            .post(format!("{base_url}/secrets"))
            .header("x-request-id", request_id)
            .header("traceparent", traceparent)
            .json(&json!({
                "name": "audit-token",
                "kind": SecretKind::Token,
                "value": { "token": "audit-secret-canary" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let secret_id = secret["id"].as_str().unwrap();

    assert_success_json(
        client
            .post(format!("{base_url}/secrets/{secret_id}/rotate"))
            .header("x-request-id", request_id)
            .header("traceparent", traceparent)
            .json(&json!({
                "value": { "token": "audit-rotated-canary" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;

    let auth_profile = assert_success_json(
        client
            .post(format!("{base_url}/auth-profiles"))
            .header("x-request-id", request_id)
            .header("traceparent", traceparent)
            .json(&json!({
                "name": "audit-profile",
                "kind": "bearer",
                "config": {
                    "bearer": {
                        "header_name": "Authorization",
                        "secret_id": secret_id
                    }
                }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let auth_profile_id = auth_profile["id"].as_str().unwrap();

    let delete_denied = client
        .delete(format!("{base_url}/secrets/{secret_id}"))
        .header("x-request-id", request_id)
        .header("traceparent", traceparent)
        .send()
        .await
        .unwrap();
    assert_eq!(delete_denied.status(), reqwest::StatusCode::CONFLICT);

    let created_agent = assert_success_json(
        client
            .post(format!("{base_url}/agents"))
            .header("x-request-id", request_id)
            .header("traceparent", traceparent)
            .json(&json!({
                "slug": "credential-audit-agent",
                "display_name": "Credential Audit Agent",
                "description": "Agent for credential audit tests.",
                "instructions": {},
                "tool_selection_policy": {}
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let agent_id = created_agent["agent_id"].as_str().unwrap();
    let created_key = assert_success_json(
        client
            .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
            .header("x-request-id", request_id)
            .header("traceparent", traceparent)
            .json(&json!({
                "name": "audit-mcp-key",
                "key_kind": PlatformApiKeyKind::McpClient,
                "scopes": ["read"]
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let raw_key = created_key["secret"].as_str().unwrap();
    let key_id = created_key["api_key"]["api_key"]["id"].as_str().unwrap();
    let duplicate_key = client
        .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
        .header("x-request-id", request_id)
        .header("traceparent", traceparent)
        .json(&json!({
            "name": "audit-mcp-key",
            "key_kind": PlatformApiKeyKind::McpClient,
            "scopes": ["read"]
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(duplicate_key.status(), reqwest::StatusCode::CONFLICT);
    let validation_key = client
        .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
        .header("x-request-id", request_id)
        .header("traceparent", traceparent)
        .json(&json!({
            "name": "n".repeat(129),
            "key_kind": PlatformApiKeyKind::McpClient,
            "scopes": ["read"]
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(validation_key.status(), reqwest::StatusCode::BAD_REQUEST);
    let invalid_profile = client
        .post(format!("{base_url}/auth-profiles"))
        .header("x-request-id", request_id)
        .header("traceparent", traceparent)
        .json(&json!({
            "name": "audit-invalid-profile",
            "kind": "bearer",
            "config": {
                "bearer": {
                    "header_name": "Authorization",
                    "secret_id": "secret_missing_for_audit"
                }
            }
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(invalid_profile.status(), reqwest::StatusCode::NOT_FOUND);
    client
        .post(format!(
            "{base_url}/agents/{agent_id}/platform-api-keys/{key_id}/revoke"
        ))
        .header("x-request-id", request_id)
        .header("traceparent", traceparent)
        .send()
        .await
        .unwrap();
    let delete_response = client
        .delete(format!(
            "{base_url}/agents/{agent_id}/platform-api-keys/{key_id}"
        ))
        .header("x-request-id", request_id)
        .header("traceparent", traceparent)
        .send()
        .await
        .unwrap();
    assert_eq!(delete_response.status(), reqwest::StatusCode::NO_CONTENT);

    let listed_after_delete = assert_success_json(
        client
            .get(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let deleted_metadata = listed_after_delete["items"]
        .as_array()
        .unwrap()
        .iter()
        .find(|item| item["api_key"]["id"] == key_id)
        .expect("delete must preserve key metadata for provenance");
    assert_eq!(deleted_metadata["api_key"]["status"], "deleted");
    let listed_after_delete_text = serde_json::to_string(&listed_after_delete).unwrap();
    assert!(!listed_after_delete_text.contains(raw_key));

    let events = audit_sink.events();
    let actions: Vec<&str> = events.iter().map(|event| event.action.as_str()).collect();
    for expected in [
        "credential.secret.created",
        "credential.secret.rotated",
        "credential.auth_profile.created",
        "credential.secret.delete_denied",
        "credential.platform_api_key.created",
        "credential.platform_api_key.create_failed",
        "credential.auth_profile.create_failed",
        "credential.platform_api_key.revoked",
        "credential.platform_api_key.deleted",
    ] {
        assert!(
            actions.contains(&expected),
            "missing audit action {expected}"
        );
    }

    let audit_json = serde_json::to_string(&events).unwrap();
    assert!(audit_json.contains(request_id));
    assert!(audit_json.contains("11111111111111111111111111111111"));
    assert!(audit_json.contains(auth_profile_id));
    assert!(!audit_json.contains("audit-secret-canary"));
    assert!(!audit_json.contains("audit-rotated-canary"));
    assert!(!audit_json.contains(raw_key));
    assert!(!audit_json.contains("secret_hash"));
    assert!(!audit_json.contains("ciphertext"));
    assert!(!audit_json.contains("Bearer "));

    for event in &events {
        let reason = event.payload["reason"]
            .as_str()
            .expect("credential audit events must contain a reason code");
        assert!(!reason.is_empty());
        assert!(reason.len() <= 128, "audit reason must remain bounded");
    }
    assert!(events.iter().any(|event| {
        event.action == "credential.platform_api_key.create_failed"
            && event.payload["reason"] == "validation_error"
    }));
    assert!(events.iter().any(|event| {
        event.action == "credential.platform_api_key.create_failed"
            && event.payload["reason"] == "platform_api_key_name_conflict"
    }));
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn disabled_secret_fails_closed_for_reference_rotation_and_execution() {
    let registry = test_registry().await;
    let registry_control = registry.clone();
    let storage_root = test_storage_root("credential_lifecycle_disabled_secret");
    let (upstream_base_url, observed_authorizations) = spawn_auth_capture_upstream().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let secret = assert_success_json(
        client
            .post(format!("{base_url}/secrets"))
            .json(&json!({
                "name": "disabled-token",
                "kind": SecretKind::Token,
                "value": { "token": "disabled-secret-canary" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let secret_id = secret["id"].as_str().unwrap();

    let auth_profile = assert_success_json(
        client
            .post(format!("{base_url}/auth-profiles"))
            .json(&json!({
                "name": "disabled-profile",
                "kind": "bearer",
                "config": {
                    "bearer": {
                        "header_name": "Authorization",
                        "secret_id": secret_id
                    }
                }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let auth_profile_id = auth_profile["id"].as_str().unwrap();

    let mut operation_payload = serde_json::to_value(test_operation_payload(
        &upstream_base_url,
        "disabled_secret_execution",
    ))
    .unwrap();
    operation_payload["execution_config"]["auth_profile_ref"] = json!(auth_profile_id);
    let created_operation = assert_success_json(
        client
            .post(format!("{base_url}/operations"))
            .json(&operation_payload)
            .send()
            .await
            .unwrap(),
    )
    .await;
    let operation_id = created_operation["operation_id"].as_str().unwrap();

    sqlx::query("update secrets set status = 'disabled' where id = $1")
        .bind(secret_id)
        .execute(registry_control.pool())
        .await
        .unwrap();

    let ref_disabled = client
        .post(format!("{base_url}/auth-profiles"))
        .json(&json!({
            "name": "disabled-profile-new-ref",
            "kind": "bearer",
            "config": {
                "bearer": {
                    "header_name": "Authorization",
                    "secret_id": secret_id
                }
            }
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(ref_disabled.status(), reqwest::StatusCode::CONFLICT);
    let ref_disabled_body: Value =
        serde_json::from_str(&ref_disabled.text().await.unwrap()).unwrap();
    assert_eq!(ref_disabled_body["error"]["code"], "secret_not_active");

    let rotate_disabled = client
        .post(format!("{base_url}/secrets/{secret_id}/rotate"))
        .json(&json!({
            "value": { "token": "should-not-rotate" }
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(rotate_disabled.status(), reqwest::StatusCode::CONFLICT);
    let rotate_body: Value = serde_json::from_str(&rotate_disabled.text().await.unwrap()).unwrap();
    assert_eq!(rotate_body["error"]["code"], "secret_not_active");

    let test_run = assert_success_json(
        client
            .post(format!("{base_url}/operations/{operation_id}/test-runs"))
            .json(&json!({
                "version": 1,
                "input": { "email": "disabled@example.com" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(test_run["ok"], false);
    let body = serde_json::to_string(&test_run).unwrap();
    assert!(!body.contains("disabled-secret-canary"));
    assert!(observed_authorizations.lock().unwrap().is_empty());

    let operation_get = client
        .get(format!("{base_url}/operations/{operation_id}"))
        .send()
        .await
        .unwrap();
    let operation_etag = operation_get
        .headers()
        .get(reqwest::header::ETAG)
        .unwrap()
        .to_str()
        .unwrap()
        .to_owned();
    let publish_disabled = client
        .post(format!("{base_url}/operations/{operation_id}/publish"))
        .header(reqwest::header::IF_MATCH, operation_etag)
        .json(&json!({ "version": 1 }))
        .send()
        .await
        .unwrap();
    assert_eq!(publish_disabled.status(), reqwest::StatusCode::CONFLICT);
    let publish_body: Value =
        serde_json::from_str(&publish_disabled.text().await.unwrap()).unwrap();
    assert_eq!(publish_body["error"]["code"], "secret_not_active");
    assert!(observed_authorizations.lock().unwrap().is_empty());
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn concurrent_secret_rotation_serializes_versions_without_storage_conflict() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("credential_lifecycle_concurrent_rotation");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let secret = assert_success_json(
        client
            .post(format!("{base_url}/secrets"))
            .json(&json!({
                "name": "concurrent-token",
                "kind": SecretKind::Token,
                "value": { "token": "initial-concurrent-token" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let secret_id = secret["id"].as_str().unwrap().to_owned();

    let first = client
        .post(format!("{base_url}/secrets/{secret_id}/rotate"))
        .json(&json!({ "value": { "token": "concurrent-token-a" } }))
        .send();
    let second = client
        .post(format!("{base_url}/secrets/{secret_id}/rotate"))
        .json(&json!({ "value": { "token": "concurrent-token-b" } }))
        .send();
    let (first, second) = tokio::join!(first, second);
    let first = assert_success_json(first.unwrap()).await;
    let second = assert_success_json(second.unwrap()).await;
    assert!(first["current_version"].as_u64().unwrap() >= 2);
    assert!(second["current_version"].as_u64().unwrap() >= 2);
    let current = assert_success_json(
        client
            .get(format!("{base_url}/secrets/{secret_id}"))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(current["current_version"], 3);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn auth_profile_secret_rotation_changes_next_admin_test_execution() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("credential_lifecycle_rotation");
    let (upstream_base_url, observed_authorizations) = spawn_auth_capture_upstream().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let secret = assert_success_json(
        client
            .post(format!("{base_url}/secrets"))
            .json(&json!({
                "name": "crm-rotating-token",
                "kind": SecretKind::Token,
                "value": { "token": "initial-rotation-token" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let secret_id = secret["id"].as_str().unwrap();

    let auth_profile = assert_success_json(
        client
            .post(format!("{base_url}/auth-profiles"))
            .json(&json!({
                "name": "crm-rotating-bearer",
                "kind": "bearer",
                "config": {
                    "bearer": {
                        "header_name": "Authorization",
                        "secret_id": secret_id
                    }
                }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let auth_profile_id = auth_profile["id"].as_str().unwrap();

    let mut operation_payload = serde_json::to_value(test_operation_payload(
        &upstream_base_url,
        "crm_auth_rotation",
    ))
    .unwrap();
    operation_payload["execution_config"]["auth_profile_ref"] = json!(auth_profile_id);
    let created_operation = assert_success_json(
        client
            .post(format!("{base_url}/operations"))
            .json(&operation_payload)
            .send()
            .await
            .unwrap(),
    )
    .await;
    let operation_id = created_operation["operation_id"].as_str().unwrap();

    let first_run = assert_success_json(
        client
            .post(format!("{base_url}/operations/{operation_id}/test-runs"))
            .json(&json!({
                "version": 1,
                "input": { "email": "first@example.com" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(first_run["ok"], true);

    let rotated = assert_success_json(
        client
            .post(format!("{base_url}/secrets/{secret_id}/rotate"))
            .json(&json!({
                "value": { "token": "rotated-rotation-token" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(rotated["current_version"], 2);

    let second_run = assert_success_json(
        client
            .post(format!("{base_url}/operations/{operation_id}/test-runs"))
            .json(&json!({
                "version": 1,
                "input": { "email": "second@example.com" }
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(second_run["ok"], true);

    let captured = observed_authorizations.lock().unwrap().clone();
    assert_eq!(
        captured,
        vec![
            Some("Bearer initial-rotation-token".to_owned()),
            Some("Bearer rotated-rotation-token".to_owned())
        ]
    );

    let auth_profile_after = client
        .get(format!("{base_url}/auth-profiles/{auth_profile_id}"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(auth_profile_after.contains(secret_id));
    assert!(!auth_profile_after.contains("initial-rotation-token"));
    assert!(!auth_profile_after.contains("rotated-rotation-token"));
}

async fn spawn_auth_capture_upstream() -> (String, Arc<Mutex<Vec<Option<String>>>>) {
    let observed_authorizations = Arc::new(Mutex::new(Vec::new()));
    let captured = Arc::clone(&observed_authorizations);
    let app = Router::new().route(
        "/crm/leads",
        post(move |headers: HeaderMap| {
            let captured = Arc::clone(&captured);
            async move {
                let authorization = headers
                    .get("authorization")
                    .and_then(|value| value.to_str().ok())
                    .map(str::to_owned);
                captured.lock().unwrap().push(authorization);
                Json(json!({ "id": "lead_123" }))
            }
        }),
    );
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    (format!("http://{address}"), observed_authorizations)
}

async fn assert_error_code(response: reqwest::Response, status: reqwest::StatusCode, code: &str) {
    assert_eq!(response.status(), status);
    let body = response.text().await.unwrap();
    let json: Value = serde_json::from_str(&body).unwrap();
    assert_eq!(json["error"]["code"], code, "unexpected error body: {body}");
}

async fn post_json(client: &reqwest::Client, url: String, payload: Value) -> Value {
    assert_success_json(client.post(url).json(&payload).send().await.unwrap()).await
}

async fn create_agent(client: &reqwest::Client, base_url: impl AsRef<str>, slug: &str) -> String {
    let base_url = base_url.as_ref();
    post_json(
        client,
        format!("{base_url}/agents"),
        json!({"slug": slug, "display_name": slug, "description": "Credential lifecycle test agent.", "instructions": {}, "tool_selection_policy": {}}),
    )
    .await["agent_id"]
        .as_str()
        .unwrap()
        .to_owned()
}

async fn assert_post_error(
    client: &reqwest::Client,
    url: String,
    payload: Value,
    status: reqwest::StatusCode,
    code: &str,
) {
    assert_error_code(
        client.post(url).json(&payload).send().await.unwrap(),
        status,
        code,
    )
    .await;
}