crank/apps/mcp-server/tests/integration/catalog_access.rs

#![allow(dead_code, unused_imports)]

use super::common::*;

use std::{
    collections::BTreeMap,
    io,
    sync::{Arc, Mutex},
    time::Duration,
};

use axum::{
    Json, Router,
    http::header,
    response::sse::{Event, KeepAlive, Sse},
    routing::{get, post},
};
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use crank_core::{
    Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, ApprovalRequest,
    ApprovalRequestId, ApprovalRequestStatus, ExecutionConfig, HttpMethod, Operation,
    OperationApprovalPayloadPreviewMode, OperationApprovalPolicy, OperationApprovalRiskLevel,
    OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope,
    PlatformApiKeyStatus, Protocol, RestTarget, Target, ToolDescription, WorkspaceId,
};
use crank_mapping::{MappingRule, MappingSet};
use crank_registry::{
    CreateAgentRequest, CreateApprovalRequest, CreatePlatformApiKeyRequest,
    ListInvocationLogsQuery, PostgresRegistry, PublishAgentRequest, PublishRequest,
    SaveAgentBindingsRequest,
};
use crank_runtime::{
    InMemoryCoordinationStateStore, RequestRateLimitConfig, RequestRateLimiter, RuntimeExecutor,
    SecretCrypto,
};
use crank_schema::{Schema, SchemaKind};
use futures_util::stream;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
use tokio::net::TcpListener;
use tokio::time::sleep;
use tracing_subscriber::{filter::LevelFilter, fmt::MakeWriter, prelude::*};

use crank_community_mcp::{
    auth::{CommunityMachineCredentialVerifier, SharedMachineCredentialVerifier},
    build_app,
    catalog::PublishedToolCatalog,
    session::{InMemorySessionStore, SharedSessionStore, TransportSessionStore},
};

fn test_workspace_id() -> WorkspaceId {
    WorkspaceId::new("ws_default")
}

#[derive(Clone, Default)]
struct SharedLogWriter {
    buffer: Arc<Mutex<Vec<u8>>>,
}

impl SharedLogWriter {
    fn output(&self) -> String {
        String::from_utf8(self.buffer.lock().unwrap().clone()).unwrap()
    }
}

impl<'a> MakeWriter<'a> for SharedLogWriter {
    type Writer = SharedLogGuard;

    fn make_writer(&'a self) -> Self::Writer {
        SharedLogGuard {
            buffer: Arc::clone(&self.buffer),
        }
    }
}

struct SharedLogGuard {
    buffer: Arc<Mutex<Vec<u8>>>,
}

impl io::Write for SharedLogGuard {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.buffer.lock().unwrap().extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

fn test_workspace_slug() -> &'static str {
    "default"
}

fn test_agent_id(agent_slug: &str) -> AgentId {
    AgentId::new(format!("agent_{agent_slug}"))
}

fn build_test_app(
    registry: PostgresRegistry,
    refresh_interval: Duration,
    public_base_url: Option<String>,
) -> axum::Router {
    build_test_app_with_rate_limit(
        registry,
        refresh_interval,
        public_base_url,
        RequestRateLimitConfig::new(10_000, 10_000).unwrap(),
    )
}

fn build_test_app_with_rate_limit(
    registry: PostgresRegistry,
    refresh_interval: Duration,
    public_base_url: Option<String>,
    rate_limit_config: RequestRateLimitConfig,
) -> axum::Router {
    build_test_app_with_store(
        registry,
        refresh_interval,
        public_base_url,
        rate_limit_config,
        std::sync::Arc::new(InMemorySessionStore::default()),
        std::sync::Arc::new(CommunityMachineCredentialVerifier),
    )
}

fn build_test_app_with_store(
    registry: PostgresRegistry,
    refresh_interval: Duration,
    public_base_url: Option<String>,
    rate_limit_config: RequestRateLimitConfig,
    sessions: SharedSessionStore,
    credential_verifier: SharedMachineCredentialVerifier,
) -> axum::Router {
    build_app(
        registry,
        refresh_interval,
        public_base_url,
        SecretCrypto::new("test-master-key").unwrap(),
        RuntimeExecutor::new(),
        RequestRateLimiter::new(rate_limit_config),
        std::sync::Arc::new(InMemoryCoordinationStateStore::default()),
        sessions,
        credential_verifier,
    )
}

#[tokio::test]
async fn refreshes_published_tools_without_restart() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let base_url = spawn_mcp_server(build_test_app(
        registry.clone(),
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let mcp_url = agent_mcp_url(&base_url, "sales-refresh");
    publish_agent_with_bindings(&registry, "sales-refresh", vec![]).await;
    let api_key = create_platform_api_key(
        &registry,
        "sales-refresh",
        "mcp-refresh",
        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
    )
    .await;
    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;

    let before_publish = post_jsonrpc(
        &client,
        &mcp_url,
        &api_key,
        Some(&initialized_session),
        json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list",
            "params": {}
        }),
    )
    .await;

    let operation = test_operation(&upstream_base_url, "crm_publish_later");
    registry
        .create_operation(&test_workspace_id(), &operation, Some("alice"))
        .await
        .unwrap();
    registry
        .publish_operation(PublishRequest {
            workspace_id: &test_workspace_id(),
            operation_id: &operation.id,
            version: 1,
            published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
            published_by: Some("alice"),
        })
        .await
        .unwrap();
    registry
        .save_agent_bindings(SaveAgentBindingsRequest {
            workspace_id: &test_workspace_id(),
            agent_id: &test_agent_id("sales-refresh"),
            agent_version: 1,
            bindings: &[binding_for_operation(&operation)],
        })
        .await
        .unwrap();

    let after_publish = post_jsonrpc(
        &client,
        &mcp_url,
        &api_key,
        Some(&initialized_session),
        json!({
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/list",
            "params": {}
        }),
    )
    .await;

    assert_eq!(before_publish["result"]["tools"], json!([]));
    assert_eq!(
        after_publish["result"]["tools"][0]["name"],
        "crm_publish_later"
    );
}

#[tokio::test]
async fn shares_published_catalog_snapshot_across_instances() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let operation = test_operation(&upstream_base_url, "crm_catalog_shared");
    let coordination_store = std::sync::Arc::new(InMemoryCoordinationStateStore::default());

    registry
        .create_operation(&test_workspace_id(), &operation, Some("alice"))
        .await
        .unwrap();
    registry
        .publish_operation(PublishRequest {
            workspace_id: &test_workspace_id(),
            operation_id: &operation.id,
            version: 1,
            published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
            published_by: Some("alice"),
        })
        .await
        .unwrap();
    publish_agent_for_operation(&registry, &operation, "sales-shared-catalog").await;

    let catalog_a = PublishedToolCatalog::new(
        registry.clone(),
        Duration::from_secs(60),
        coordination_store.clone(),
    );
    let tools_a = catalog_a
        .list_tools(test_workspace_slug(), "sales-shared-catalog")
        .await
        .unwrap();
    assert_eq!(tools_a.len(), 1);

    registry
        .unpublish_agent(
            &test_workspace_id(),
            &test_agent_id("sales-shared-catalog"),
            &OffsetDateTime::parse("2026-03-26T10:05:00Z", &Rfc3339).unwrap(),
        )
        .await
        .unwrap();

    let catalog_b = PublishedToolCatalog::new(
        registry.clone(),
        Duration::from_secs(60),
        coordination_store,
    );
    let tools_b = catalog_b
        .list_tools(test_workspace_slug(), "sales-shared-catalog")
        .await
        .unwrap();

    assert_eq!(tools_b.len(), 1);
    assert_eq!(tools_b[0].tool_name, operation.name);
}

#[tokio::test]
async fn lists_only_bound_tools_for_agent_context() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let operation_a = test_operation(&upstream_base_url, "crm_create_lead");
    let operation_b = test_operation(&upstream_base_url, "crm_update_lead");

    for operation in [&operation_a, &operation_b] {
        registry
            .create_operation(&test_workspace_id(), operation, Some("alice"))
            .await
            .unwrap();
        registry
            .publish_operation(PublishRequest {
                workspace_id: &test_workspace_id(),
                operation_id: &operation.id,
                version: 1,
                published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
                published_by: Some("alice"),
            })
            .await
            .unwrap();
    }

    publish_agent_with_bindings(
        &registry,
        "sales-a",
        vec![binding_for_operation(&operation_a)],
    )
    .await;
    publish_agent_with_bindings(
        &registry,
        "sales-b",
        vec![binding_for_operation(&operation_b)],
    )
    .await;
    let api_key_a = create_platform_api_key(
        &registry,
        "sales-a",
        "mcp-agent-a",
        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
    )
    .await;
    let api_key_b = create_platform_api_key(
        &registry,
        "sales-b",
        "mcp-agent-b",
        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
    )
    .await;

    let base_url = spawn_mcp_server(build_test_app(
        registry,
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let agent_a_url = agent_mcp_url(&base_url, "sales-a");
    let agent_b_url = agent_mcp_url(&base_url, "sales-b");
    let session_a = initialize_session(&client, &agent_a_url, &api_key_a).await;
    let session_b = initialize_session(&client, &agent_b_url, &api_key_b).await;

    let tools_a = post_jsonrpc(
        &client,
        &agent_a_url,
        &api_key_a,
        Some(&session_a),
        json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list",
            "params": {}
        }),
    )
    .await;
    let tools_b = post_jsonrpc(
        &client,
        &agent_b_url,
        &api_key_b,
        Some(&session_b),
        json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list",
            "params": {}
        }),
    )
    .await;

    assert_eq!(tools_a["result"]["tools"][0]["name"], "crm_create_lead");
    assert_eq!(tools_b["result"]["tools"][0]["name"], "crm_update_lead");
}

#[tokio::test]
async fn rejects_initialize_with_key_from_different_agent() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let operation_a = test_operation(&upstream_base_url, "crm_create_lead");
    let operation_b = test_operation(&upstream_base_url, "crm_update_lead");

    for operation in [&operation_a, &operation_b] {
        registry
            .create_operation(&test_workspace_id(), operation, Some("alice"))
            .await
            .unwrap();
        registry
            .publish_operation(PublishRequest {
                workspace_id: &test_workspace_id(),
                operation_id: &operation.id,
                version: 1,
                published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
                published_by: Some("alice"),
            })
            .await
            .unwrap();
    }

    publish_agent_with_bindings(
        &registry,
        "sales-a",
        vec![binding_for_operation(&operation_a)],
    )
    .await;
    publish_agent_with_bindings(
        &registry,
        "sales-b",
        vec![binding_for_operation(&operation_b)],
    )
    .await;
    let api_key_a = create_platform_api_key(
        &registry,
        "sales-a",
        "mcp-agent-a-only",
        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
    )
    .await;

    let base_url = spawn_mcp_server(build_test_app(
        registry,
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let response = client
        .post(agent_mcp_url(&base_url, "sales-b"))
        .header(header::ACCEPT, "application/json, text/event-stream")
        .header(header::AUTHORIZATION, format!("Bearer {api_key_a}"))
        .json(&json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25"
            }
        }))
        .send()
        .await
        .unwrap();

    assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
}

#[tokio::test]
async fn rejects_initialize_without_platform_api_key() {
    let registry = test_registry().await;
    publish_agent_with_bindings(&registry, "sales-auth", vec![]).await;
    let base_url = spawn_mcp_server(build_test_app(
        registry,
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let response = client
        .post(agent_mcp_url(&base_url, "sales-auth"))
        .header(header::ACCEPT, "application/json, text/event-stream")
        .json(&json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25"
            }
        }))
        .send()
        .await
        .unwrap();

    assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
}

#[tokio::test]
async fn rejects_initialize_with_approval_platform_api_key() {
    let registry = test_registry().await;
    publish_agent_with_bindings(&registry, "sales-approval-key", vec![]).await;
    let api_key =
        create_approval_platform_api_key(&registry, "sales-approval-key", "approval-only").await;
    let base_url = spawn_mcp_server(build_test_app(
        registry,
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let response = client
        .post(agent_mcp_url(&base_url, "sales-approval-key"))
        .header(header::ACCEPT, "application/json, text/event-stream")
        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
        .json(&json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25"
            }
        }))
        .send()
        .await
        .unwrap();

    assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
}

#[tokio::test]
async fn approval_key_lists_and_decides_pending_requests() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let operation = test_operation(&upstream_base_url, "crm_human_approval");
    registry
        .create_operation(&test_workspace_id(), &operation, Some("alice"))
        .await
        .unwrap();
    publish_agent_with_bindings(
        &registry,
        "sales-human-approval",
        vec![binding_for_operation(&operation)],
    )
    .await;
    let approval = ApprovalRequest {
        id: ApprovalRequestId::new("approval_mcp_01"),
        workspace_id: test_workspace_id(),
        agent_id: test_agent_id("sales-human-approval"),
        operation_id: operation.id.clone(),
        operation_version: 1,
        status: ApprovalRequestStatus::Pending,
        risk_level: OperationApprovalRiskLevel::Dangerous,
        confirmation_title: "Подтвердите создание лида".to_owned(),
        confirmation_body: "Проверьте email перед отправкой в CRM.".to_owned(),
        request_payload: json!({"email": "ada@example.com"}),
        response_payload: None,
        created_at: OffsetDateTime::now_utc(),
        expires_at: OffsetDateTime::now_utc() + time::Duration::minutes(5),
        decided_at: None,
        decided_by_key_id: None,
        decision_note: None,
    };
    registry
        .create_approval_request(CreateApprovalRequest {
            approval: &approval,
        })
        .await
        .unwrap();

    let approval_key =
        create_approval_platform_api_key(&registry, "sales-human-approval", "approval-http").await;
    let mcp_key = create_platform_api_key(
        &registry,
        "sales-human-approval",
        "mcp-human-approval",
        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
    )
    .await;
    let base_url = spawn_mcp_server(build_test_app(
        registry.clone(),
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let approvals_url = format!(
        "{}/approvals",
        agent_mcp_url(&base_url, "sales-human-approval")
    );

    let rejected = client
        .get(&approvals_url)
        .header(header::AUTHORIZATION, format!("Bearer {mcp_key}"))
        .send()
        .await
        .unwrap();
    assert_eq!(rejected.status(), reqwest::StatusCode::UNAUTHORIZED);

    let pending = client
        .get(&approvals_url)
        .header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
        .send()
        .await
        .unwrap();
    assert_eq!(pending.status(), reqwest::StatusCode::OK);
    let pending_body = pending.json::<Value>().await.unwrap();
    assert_eq!(pending_body["items"].as_array().unwrap().len(), 1);
    assert_eq!(
        pending_body["items"][0]["approval"]["confirmation_title"],
        "Подтвердите создание лида"
    );

    let approve_url = format!(
        "{}/approvals/{}/approve",
        agent_mcp_url(&base_url, "sales-human-approval"),
        approval.id
    );
    let approved = client
        .post(&approve_url)
        .header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
        .json(&json!({ "approve": "yes", "note": "confirmed by test" }))
        .send()
        .await
        .unwrap();
    assert_eq!(approved.status(), reqwest::StatusCode::OK);
    let approved_body = approved.json::<Value>().await.unwrap();
    assert_eq!(
        approved_body["approval"]["status"],
        Value::String("completed".to_owned())
    );
    assert_eq!(
        approved_body["approval"]["response_payload"],
        json!({ "id": "lead_123" })
    );

    let status_url = format!(
        "{}/approvals/{}",
        agent_mcp_url(&base_url, "sales-human-approval"),
        approval.id
    );
    let current = client
        .get(&status_url)
        .header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
        .send()
        .await
        .unwrap();
    assert_eq!(current.status(), reqwest::StatusCode::OK);
    let current_body = current.json::<Value>().await.unwrap();
    assert_eq!(
        current_body["approval"]["status"],
        Value::String("completed".to_owned())
    );
    assert_eq!(
        current_body["approval"]["response_payload"],
        json!({ "id": "lead_123" })
    );

    let pending_after = client
        .get(&approvals_url)
        .header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    assert!(pending_after["items"].as_array().unwrap().is_empty());
}

#[tokio::test]
async fn tool_call_with_approval_policy_creates_pending_request() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let mut operation = test_operation(&upstream_base_url, "crm_requires_human_approval");
    operation.execution_config.approval_policy = Some(OperationApprovalPolicy {
        required: true,
        risk_level: OperationApprovalRiskLevel::Dangerous,
        confirmation_title: "Подтвердите создание лида".to_owned(),
        confirmation_body_template: "Проверьте email перед отправкой в CRM.".to_owned(),
        ttl_seconds: 300,
        show_payload_preview: true,
        payload_preview_mode: OperationApprovalPayloadPreviewMode::MaskedJson,
    });
    registry
        .create_operation(&test_workspace_id(), &operation, Some("alice"))
        .await
        .unwrap();
    registry
        .publish_operation(PublishRequest {
            workspace_id: &test_workspace_id(),
            operation_id: &operation.id,
            version: 1,
            published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
            published_by: Some("alice"),
        })
        .await
        .unwrap();
    publish_agent_for_operation(&registry, &operation, "sales-gated").await;
    let api_key = create_platform_api_key(
        &registry,
        "sales-gated",
        "mcp-gated",
        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
    )
    .await;
    let approval_key =
        create_approval_platform_api_key(&registry, "sales-gated", "approval-gated").await;

    let base_url = spawn_mcp_server(build_test_app(
        registry,
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let mcp_url = agent_mcp_url(&base_url, "sales-gated");
    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;

    let tool_result = post_jsonrpc(
        &client,
        &mcp_url,
        &api_key,
        Some(&initialized_session),
        json!({
            "jsonrpc": "2.0",
            "id": 9,
            "method": "tools/call",
            "params": {
                "name": "crm_requires_human_approval",
                "arguments": {
                    "email": "ada@example.com"
                }
            }
        }),
    )
    .await;

    assert_eq!(
        tool_result["result"]["structuredContent"]["status"],
        "approval_required"
    );
    assert_eq!(tool_result["result"]["isError"], false);
    let approval_id = tool_result["result"]["structuredContent"]["approval_id"]
        .as_str()
        .unwrap();
    assert!(approval_id.starts_with("approval_"));

    let approvals_url = format!("{}/approvals", agent_mcp_url(&base_url, "sales-gated"));
    let pending = client
        .get(&approvals_url)
        .header(header::AUTHORIZATION, format!("Bearer {approval_key}"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    assert_eq!(pending["items"].as_array().unwrap().len(), 1);
    assert_eq!(pending["items"][0]["approval"]["id"], approval_id);
    assert_eq!(
        pending["items"][0]["approval"]["request_payload"]["email"],
        "ada@example.com"
    );
}

#[tokio::test]
async fn rejects_tool_call_with_read_only_platform_api_key() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let operation = test_operation(&upstream_base_url, "crm_read_only");

    registry
        .create_operation(&test_workspace_id(), &operation, Some("alice"))
        .await
        .unwrap();
    registry
        .publish_operation(PublishRequest {
            workspace_id: &test_workspace_id(),
            operation_id: &operation.id,
            version: 1,
            published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
            published_by: Some("alice"),
        })
        .await
        .unwrap();
    publish_agent_for_operation(&registry, &operation, "sales-read-only").await;
    let api_key = create_platform_api_key(
        &registry,
        "sales-read-only",
        "mcp-read",
        &[PlatformApiKeyScope::Read],
    )
    .await;

    let base_url = spawn_mcp_server(build_test_app(
        registry,
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
    ))
    .await;
    let client = reqwest::Client::new();
    let mcp_url = agent_mcp_url(&base_url, "sales-read-only");
    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
    let response = client
        .post(&mcp_url)
        .header(header::ACCEPT, "application/json, text/event-stream")
        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
        .header("MCP-Session-Id", initialized_session)
        .header("MCP-Protocol-Version", "2025-11-25")
        .json(&json!({
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/call",
            "params": {
                "name": "crm_read_only",
                "arguments": {
                    "email": "user@example.com"
                }
            }
        }))
        .send()
        .await
        .unwrap();

    assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
}

#[tokio::test]
async fn rejects_rapid_initialize_requests_with_429() {
    let registry = test_registry().await;
    let upstream_base_url = spawn_upstream_server().await;
    let operation = test_operation(&upstream_base_url, "crm_rate_limited");

    registry
        .create_operation(&test_workspace_id(), &operation, Some("alice"))
        .await
        .unwrap();
    registry
        .publish_operation(PublishRequest {
            workspace_id: &test_workspace_id(),
            operation_id: &operation.id,
            version: 1,
            published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
            published_by: Some("alice"),
        })
        .await
        .unwrap();
    publish_agent_for_operation(&registry, &operation, "sales-rate-limited").await;
    let api_key = create_platform_api_key(
        &registry,
        "sales-rate-limited",
        "mcp-rate-limit",
        &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
    )
    .await;

    let base_url = spawn_mcp_server(build_test_app_with_rate_limit(
        registry,
        Duration::from_millis(0),
        Some("https://crank.example.com".to_owned()),
        RequestRateLimitConfig::new(1, 1).unwrap(),
    ))
    .await;
    let client = reqwest::Client::new();
    let mcp_url = agent_mcp_url(&base_url, "sales-rate-limited");

    let first_response = client
        .post(&mcp_url)
        .header(header::ACCEPT, "application/json, text/event-stream")
        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
        .header("x-request-id", "req_rate_limit_01")
        .json(&json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25"
            }
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(first_response.status(), reqwest::StatusCode::OK);

    let second_response = client
        .post(&mcp_url)
        .header(header::ACCEPT, "application/json, text/event-stream")
        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
        .header("x-request-id", "req_rate_limit_02")
        .json(&json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25"
            }
        }))
        .send()
        .await
        .unwrap();

    assert_eq!(
        second_response.status(),
        reqwest::StatusCode::TOO_MANY_REQUESTS
    );
    assert_eq!(
        second_response.headers()["x-request-id"].to_str().unwrap(),
        "req_rate_limit_02"
    );
    assert_eq!(
        second_response.headers()["retry-after"].to_str().unwrap(),
        "1"
    );

    let payload = second_response.json::<Value>().await.unwrap();
    assert_eq!(payload["error"]["code"], json!(-32029));
    assert_eq!(
        payload["error"]["data"]["code"],
        json!("request_rate_limited")
    );
    let retry_after_ms = payload["error"]["data"]["retry_after_ms"].as_u64().unwrap();
    assert!((1..=1000).contains(&retry_after_ms));
}