Refactor large Rust integration tests

apps/admin-api/tests/integration.rs

@@ -0,0 +1,7 @@
+mod integration {
+    mod auth_rate_limit;
+    mod common;
+    mod community_access_usage;
+    mod operations_agents;
+    mod secrets_import_auth;
+}

apps/admin-api/tests/integration/auth_rate_limit.rs

@@ -0,0 +1,193 @@
+#![allow(dead_code, unused_imports)]
+
+use super::common::*;
+
+use std::{
+    collections::BTreeMap,
+    env, fmt,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use async_trait::async_trait;
+use axum::{Json, Router, routing::post};
+use crank_core::{
+    ExecutionConfig, HttpMethod, MembershipRole, OperationSecurityLevel, Protocol,
+    ResponseCachePolicy, RestTarget, SecretKind, Target, ToolDescription, WorkspaceId,
+};
+use crank_core::{IdentityError, IdentityProvider, IdentityProviderKind, LoginOutcome};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::PostgresRegistry;
+use crank_runtime::SecretCrypto;
+use crank_schema::{Schema, SchemaKind};
+use serde_json::{Value, json};
+use serial_test::serial;
+use tokio::net::TcpListener;
+
+use admin_api::{
+    app::build_app,
+    auth::{AuthSettings, BootstrapAdminConfig, hash_password},
+    service::{AdminService, AdminServiceBuilder, OperationPayload},
+    state::AppState,
+};
+
+const DEFAULT_WORKSPACE_ID: &str = "ws_default";
+const TEST_AUTH_EMAIL: &str = "owner@crank.local";
+const TEST_AUTH_PASSWORD: &str = "test-password";
+const TEST_PASSWORD_PEPPER: &str = "test-password-pepper";
+const TEST_SESSION_SECRET: &str = "test-session-secret";
+const TEST_MASTER_KEY: &str = "test-master-key";
+
+struct TestServer {
+    base_url: String,
+    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        let shutdown = self.shutdown.take();
+        let handle = self.handle.take();
+
+        tokio::task::block_in_place(|| {
+            if let Some(shutdown) = shutdown {
+                let _ = shutdown.send(());
+            }
+            if let Some(handle) = handle {
+                let _ = tokio::runtime::Handle::current().block_on(handle);
+            }
+        });
+    }
+}
+
+impl fmt::Display for TestServer {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.base_url)
+    }
+}
+
+impl AsRef<str> for TestServer {
+    fn as_ref(&self) -> &str {
+        &self.base_url
+    }
+}
+
+struct RejectingIdentityProvider;
+
+#[async_trait]
+impl IdentityProvider for RejectingIdentityProvider {
+    fn id(&self) -> &str {
+        "rejecting-test-provider"
+    }
+
+    fn kind(&self) -> IdentityProviderKind {
+        IdentityProviderKind::Password
+    }
+
+    async fn login_password(
+        &self,
+        _payload: crank_core::LoginPayload,
+    ) -> Result<LoginOutcome, IdentityError> {
+        Err(IdentityError::BadCredentials)
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn rejects_rapid_login_requests_with_429() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("login_rate_limit");
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let app = build_app(AppState {
+        service: test_service(
+            registry,
+            storage_root,
+            test_auth_settings(),
+            test_secret_crypto(),
+        ),
+        api_rate_limiter: crank_runtime::RequestRateLimiter::new(
+            crank_runtime::RequestRateLimitConfig::new(1, 1).unwrap(),
+        ),
+    });
+    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
+    let handle = tokio::spawn(async move {
+        axum::serve(listener, app)
+            .with_graceful_shutdown(async move {
+                let _ = shutdown_rx.await;
+            })
+            .await
+            .unwrap();
+    });
+
+    let client = reqwest::Client::new();
+    let root_url = format!("http://{address}");
+    let first_response = client
+        .post(format!("{root_url}/api/auth/login"))
+        .header("x-request-id", "req_admin_rate_01")
+        .json(&json!({
+            "email": TEST_AUTH_EMAIL,
+            "password": TEST_AUTH_PASSWORD,
+        }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first_response.status(), reqwest::StatusCode::OK);
+
+    let second_response = client
+        .post(format!("{root_url}/api/auth/login"))
+        .header("x-request-id", "req_admin_rate_02")
+        .json(&json!({
+            "email": TEST_AUTH_EMAIL,
+            "password": TEST_AUTH_PASSWORD,
+        }))
+        .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_admin_rate_02"
+    );
+    let payload = second_response.json::<Value>().await.unwrap();
+    assert_eq!(payload["error"]["code"], "rate_limited");
+    let retry_after_ms = payload["error"]["context"]["retry_after_ms"]
+        .as_u64()
+        .unwrap();
+    assert!((1..=1000).contains(&retry_after_ms));
+
+    let _ = shutdown_tx.send(());
+    let _ = handle.await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn login_uses_identity_provider_when_configured() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("identity_provider_login");
+    let service = AdminServiceBuilder::new(
+        registry,
+        storage_root,
+        test_auth_settings(),
+        test_secret_crypto(),
+        crank_runtime::community_default().build(),
+    )
+    .with_identity_provider(Arc::new(RejectingIdentityProvider))
+    .build();
+
+    let error = match service
+        .login(admin_api::service::LoginPayload {
+            email: TEST_AUTH_EMAIL.to_owned(),
+            password: TEST_AUTH_PASSWORD.to_owned(),
+        })
+        .await
+    {
+        Ok(_) => panic!("login should delegate to identity provider"),
+        Err(error) => error,
+    };
+    assert_eq!(error.to_string(), "invalid email or password");
+}

apps/admin-api/tests/integration/common.rs

@@ -0,0 +1,338 @@
+#![allow(dead_code, unused_imports)]
+
+use std::{
+    collections::BTreeMap,
+    env, fmt,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use async_trait::async_trait;
+use axum::{Json, Router, routing::post};
+use crank_core::{
+    ExecutionConfig, HttpMethod, MembershipRole, OperationSecurityLevel, Protocol,
+    ResponseCachePolicy, RestTarget, SecretKind, Target, ToolDescription, WorkspaceId,
+};
+use crank_core::{IdentityError, IdentityProvider, IdentityProviderKind, LoginOutcome};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::PostgresRegistry;
+use crank_runtime::SecretCrypto;
+use crank_schema::{Schema, SchemaKind};
+use serde_json::{Value, json};
+use serial_test::serial;
+use tokio::net::TcpListener;
+
+use admin_api::{
+    app::build_app,
+    auth::{AuthSettings, BootstrapAdminConfig, hash_password},
+    service::{AdminService, AdminServiceBuilder, OperationPayload},
+    state::AppState,
+};
+
+const DEFAULT_WORKSPACE_ID: &str = "ws_default";
+const TEST_AUTH_EMAIL: &str = "owner@crank.local";
+const TEST_AUTH_PASSWORD: &str = "test-password";
+const TEST_PASSWORD_PEPPER: &str = "test-password-pepper";
+const TEST_SESSION_SECRET: &str = "test-session-secret";
+const TEST_MASTER_KEY: &str = "test-master-key";
+
+pub(super) struct TestServer {
+    base_url: String,
+    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        let shutdown = self.shutdown.take();
+        let handle = self.handle.take();
+
+        tokio::task::block_in_place(|| {
+            if let Some(shutdown) = shutdown {
+                let _ = shutdown.send(());
+            }
+            if let Some(handle) = handle {
+                let _ = tokio::runtime::Handle::current().block_on(handle);
+            }
+        });
+    }
+}
+
+impl fmt::Display for TestServer {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.base_url)
+    }
+}
+
+impl AsRef<str> for TestServer {
+    fn as_ref(&self) -> &str {
+        &self.base_url
+    }
+}
+
+pub(super) struct RejectingIdentityProvider;
+
+#[async_trait]
+impl IdentityProvider for RejectingIdentityProvider {
+    fn id(&self) -> &str {
+        "rejecting-test-provider"
+    }
+
+    fn kind(&self) -> IdentityProviderKind {
+        IdentityProviderKind::Password
+    }
+
+    async fn login_password(
+        &self,
+        _payload: crank_core::LoginPayload,
+    ) -> Result<LoginOutcome, IdentityError> {
+        Err(IdentityError::BadCredentials)
+    }
+}
+
+pub(super) fn build_test_app(
+    registry: PostgresRegistry,
+    storage_root: std::path::PathBuf,
+) -> Router {
+    build_app(AppState {
+        service: test_service(
+            registry,
+            storage_root,
+            test_auth_settings(),
+            test_secret_crypto(),
+        ),
+        api_rate_limiter: crank_runtime::RequestRateLimiter::new(
+            crank_runtime::RequestRateLimitConfig::new(10_000, 10_000).unwrap(),
+        ),
+    })
+}
+
+pub(super) fn test_service(
+    registry: PostgresRegistry,
+    storage_root: std::path::PathBuf,
+    auth_settings: AuthSettings,
+    secret_crypto: SecretCrypto,
+) -> AdminService {
+    AdminServiceBuilder::new(
+        registry,
+        storage_root,
+        auth_settings,
+        secret_crypto,
+        crank_runtime::RuntimeExecutor::new(),
+    )
+    .build()
+}
+
+pub(super) async fn spawn_upstream_server() -> String {
+    let app = Router::new().route("/crm/leads", post(create_lead));
+    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)
+}
+
+pub(super) async fn spawn_admin_api(app: Router) -> TestServer {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
+
+    let handle = tokio::spawn(async move {
+        axum::serve(listener, app)
+            .with_graceful_shutdown(async move {
+                let _ = shutdown_rx.await;
+            })
+            .await
+            .unwrap();
+    });
+
+    TestServer {
+        base_url: format!(
+            "http://{}/api/admin/workspaces/{}",
+            address, DEFAULT_WORKSPACE_ID
+        ),
+        shutdown: Some(shutdown_tx),
+        handle: Some(handle),
+    }
+}
+
+pub(super) async fn authorized_client(workspace_base_url: impl AsRef<str>) -> reqwest::Client {
+    let root_url = workspace_base_url
+        .as_ref()
+        .split("/api/admin/workspaces/")
+        .next()
+        .unwrap();
+    let client = reqwest::Client::builder()
+        .cookie_store(true)
+        .build()
+        .unwrap();
+
+    client
+        .post(format!("{root_url}/api/auth/login"))
+        .json(&json!({
+            "email": TEST_AUTH_EMAIL,
+            "password": TEST_AUTH_PASSWORD,
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+
+    client
+}
+
+pub(super) async fn assert_success_json(response: reqwest::Response) -> Value {
+    let status = response.status();
+    let body = response.text().await.unwrap();
+
+    assert!(
+        status.is_success(),
+        "request failed with status {status}: {body}"
+    );
+
+    serde_json::from_str(&body).unwrap()
+}
+
+pub(super) async fn create_lead(Json(payload): Json<Value>) -> Json<Value> {
+    Json(json!({
+        "id": "lead_123",
+        "status": "created",
+        "email": payload["email"]
+    }))
+}
+
+pub(super) async fn test_registry() -> PostgresRegistry {
+    let database_url = crank_test_support::postgres_schema_url("test_admin_api").await;
+    let registry = PostgresRegistry::connect(&database_url).await.unwrap();
+    let password_hash = hash_password(TEST_AUTH_PASSWORD, TEST_PASSWORD_PEPPER).unwrap();
+    let user_id = registry
+        .upsert_bootstrap_user(TEST_AUTH_EMAIL, "Test Owner", &password_hash)
+        .await
+        .unwrap();
+    registry
+        .ensure_membership(
+            &WorkspaceId::new(DEFAULT_WORKSPACE_ID),
+            &user_id,
+            MembershipRole::Owner,
+        )
+        .await
+        .unwrap();
+
+    registry
+}
+
+pub(super) fn test_storage_root(name: &str) -> std::path::PathBuf {
+    env::temp_dir().join(format!(
+        "crank_admin_api_{name}_{}_{}",
+        std::process::id(),
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    ))
+}
+
+pub(super) fn test_auth_settings() -> AuthSettings {
+    AuthSettings {
+        session_secret: TEST_SESSION_SECRET.to_owned(),
+        password_pepper: TEST_PASSWORD_PEPPER.to_owned(),
+        session_ttl_hours: 24,
+        cookie_secure: false,
+        bootstrap_admin: BootstrapAdminConfig {
+            email: TEST_AUTH_EMAIL.to_owned(),
+            password: TEST_AUTH_PASSWORD.to_owned(),
+            display_name: "Test Owner".to_owned(),
+        },
+    }
+}
+
+pub(super) fn test_secret_crypto() -> SecretCrypto {
+    SecretCrypto::new(TEST_MASTER_KEY).unwrap()
+}
+
+pub(super) fn test_operation_payload(base_url: &str, name: &str) -> OperationPayload {
+    OperationPayload {
+        name: name.to_owned(),
+        display_name: "Create Lead".to_owned(),
+        category: "sales".to_owned(),
+        protocol: Protocol::Rest,
+        security_level: OperationSecurityLevel::Standard,
+        target: Target::Rest(RestTarget {
+            base_url: base_url.to_owned(),
+            method: HttpMethod::Post,
+            path_template: "/crm/leads".to_owned(),
+            static_headers: BTreeMap::new(),
+        }),
+        input_schema: object_schema("email"),
+        output_schema: object_schema("id"),
+        input_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.mcp.email".to_owned(),
+                target: "$.request.body.email".to_owned(),
+                required: true,
+                default_value: None,
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        },
+        output_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.response.body.id".to_owned(),
+                target: "$.output.id".to_owned(),
+                required: true,
+                default_value: None,
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        },
+        execution_config: ExecutionConfig {
+            timeout_ms: 1_000,
+            retry_policy: None,
+            response_cache: None,
+            idempotency: None,
+            safety: None,
+            auth_profile_ref: None,
+            headers: BTreeMap::new(),
+        },
+        tool_description: ToolDescription {
+            title: "Create Lead".to_owned(),
+            description: "Creates a CRM lead".to_owned(),
+            tags: vec!["crm".to_owned()],
+            examples: Vec::new(),
+        },
+        wizard_state: None,
+    }
+}
+
+pub(super) fn object_schema(field_name: &str) -> Schema {
+    Schema {
+        kind: SchemaKind::Object,
+        description: None,
+        required: true,
+        nullable: false,
+        default_value: None,
+        fields: BTreeMap::from([(
+            field_name.to_owned(),
+            Schema {
+                kind: SchemaKind::String,
+                description: None,
+                required: true,
+                nullable: false,
+                default_value: None,
+                fields: BTreeMap::new(),
+                items: None,
+                enum_values: Vec::new(),
+                variants: Vec::new(),
+            },
+        )]),
+        items: None,
+        enum_values: Vec::new(),
+        variants: Vec::new(),
+    }
+}

apps/admin-api/tests/integration/community_access_usage.rs

@@ -0,0 +1,803 @@
+#![allow(dead_code, unused_imports)]
+
+use super::common::*;
+
+use std::{
+    collections::BTreeMap,
+    env, fmt,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use async_trait::async_trait;
+use axum::{Json, Router, routing::post};
+use crank_core::{
+    ExecutionConfig, HttpMethod, MembershipRole, OperationSecurityLevel, Protocol,
+    ResponseCachePolicy, RestTarget, SecretKind, Target, ToolDescription, WorkspaceId,
+};
+use crank_core::{IdentityError, IdentityProvider, IdentityProviderKind, LoginOutcome};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::PostgresRegistry;
+use crank_runtime::SecretCrypto;
+use crank_schema::{Schema, SchemaKind};
+use serde_json::{Value, json};
+use serial_test::serial;
+use tokio::net::TcpListener;
+
+use admin_api::{
+    app::build_app,
+    auth::{AuthSettings, BootstrapAdminConfig, hash_password},
+    service::{AdminService, AdminServiceBuilder, OperationPayload},
+    state::AppState,
+};
+
+const DEFAULT_WORKSPACE_ID: &str = "ws_default";
+const TEST_AUTH_EMAIL: &str = "owner@crank.local";
+const TEST_AUTH_PASSWORD: &str = "test-password";
+const TEST_PASSWORD_PEPPER: &str = "test-password-pepper";
+const TEST_SESSION_SECRET: &str = "test-session-secret";
+const TEST_MASTER_KEY: &str = "test-master-key";
+
+struct TestServer {
+    base_url: String,
+    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        let shutdown = self.shutdown.take();
+        let handle = self.handle.take();
+
+        tokio::task::block_in_place(|| {
+            if let Some(shutdown) = shutdown {
+                let _ = shutdown.send(());
+            }
+            if let Some(handle) = handle {
+                let _ = tokio::runtime::Handle::current().block_on(handle);
+            }
+        });
+    }
+}
+
+impl fmt::Display for TestServer {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.base_url)
+    }
+}
+
+impl AsRef<str> for TestServer {
+    fn as_ref(&self) -> &str {
+        &self.base_url
+    }
+}
+
+struct RejectingIdentityProvider;
+
+#[async_trait]
+impl IdentityProvider for RejectingIdentityProvider {
+    fn id(&self) -> &str {
+        "rejecting-test-provider"
+    }
+
+    fn kind(&self) -> IdentityProviderKind {
+        IdentityProviderKind::Password
+    }
+
+    async fn login_password(
+        &self,
+        _payload: crank_core::LoginPayload,
+    ) -> Result<LoginOutcome, IdentityError> {
+        Err(IdentityError::BadCredentials)
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn rejects_workspace_access_management_in_community() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("platform_access");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let members_status = client
+        .get(format!("{base_url}/members"))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    let create_invitation_status = client
+        .post(format!("{base_url}/invitations"))
+        .json(&json!({
+            "email": "operator@example.com",
+            "role": "operator"
+        }))
+        .send()
+        .await
+        .unwrap()
+        .status();
+
+    assert_eq!(members_status, reqwest::StatusCode::NOT_FOUND);
+    assert_eq!(create_invitation_status, reqwest::StatusCode::NOT_FOUND);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn manages_agent_platform_api_keys() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("agent_platform_keys");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created_agent = assert_success_json(
+        client
+            .post(format!("{base_url}/agents"))
+            .json(&json!({
+                "slug": "sales-routing",
+                "display_name": "Sales Routing",
+                "description": "Routing agent",
+                "instructions": {},
+                "tool_selection_policy": {}
+            }))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let agent_id = created_agent["agent_id"].as_str().unwrap().to_owned();
+
+    let created_key = assert_success_json(
+        client
+            .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
+            .json(&json!({
+                "name": "sales-routing-primary",
+                "scopes": ["read", "write"]
+            }))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let key_id = created_key["api_key"]["api_key"]["id"]
+        .as_str()
+        .unwrap()
+        .to_owned();
+
+    let listed_keys = assert_success_json(
+        client
+            .get(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+
+    let revoke_status = client
+        .post(format!(
+            "{base_url}/agents/{agent_id}/platform-api-keys/{key_id}/revoke"
+        ))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    let delete_status = client
+        .delete(format!(
+            "{base_url}/agents/{agent_id}/platform-api-keys/{key_id}"
+        ))
+        .send()
+        .await
+        .unwrap()
+        .status();
+
+    assert_eq!(created_key["api_key"]["api_key"]["agent_id"], agent_id);
+    assert_eq!(
+        listed_keys["items"][0]["api_key"]["agent_id"],
+        json!(agent_id)
+    );
+    assert_eq!(
+        listed_keys["items"][0]["api_key"]["name"],
+        "sales-routing-primary"
+    );
+    assert!(created_key["secret"].as_str().unwrap().starts_with("crk_"));
+    assert_eq!(revoke_status, reqwest::StatusCode::NO_CONTENT);
+    assert_eq!(delete_status, reqwest::StatusCode::NO_CONTENT);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn returns_community_capabilities() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("community_capabilities");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+    let root_url = base_url
+        .as_ref()
+        .split("/api/admin/workspaces/")
+        .next()
+        .unwrap();
+
+    let response = assert_success_json(
+        client
+            .get(format!("{root_url}/api/admin/capabilities"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+
+    assert_eq!(response["edition"], "community");
+    assert_eq!(response["supported_protocols"], json!(["rest"]));
+    assert_eq!(response["supported_security_levels"], json!(["standard"]));
+    assert_eq!(
+        response["machine_access_modes"],
+        json!(["static_agent_key"])
+    );
+    assert_eq!(response["limits"]["max_workspaces"], 1);
+    assert_eq!(response["limits"]["max_users_per_workspace"], 1);
+    assert_eq!(response["limits"]["max_agents_per_workspace"], Value::Null);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn rejects_response_cache_for_non_get_rest_operation_create() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("community_response_cache_post_reject");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let mut payload = test_operation_payload(&upstream_base_url, "crm_cache_post");
+    payload.execution_config.response_cache = Some(ResponseCachePolicy { ttl_ms: 5_000 });
+
+    let response = client
+        .post(format!("{base_url}/operations"))
+        .json(&payload)
+        .send()
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response.json::<Value>().await.unwrap();
+
+    assert_eq!(status, reqwest::StatusCode::BAD_REQUEST);
+    assert_eq!(body["error"]["code"], "validation_error");
+    assert_eq!(
+        body["error"]["context"]["field"],
+        "execution_config.response_cache"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn rejects_response_cache_for_operation_with_auth_profile() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("community_response_cache_auth_reject");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let mut payload = test_operation_payload(&upstream_base_url, "crm_cache_auth");
+    payload.target = Target::Rest(RestTarget {
+        base_url: upstream_base_url,
+        method: HttpMethod::Get,
+        path_template: "/crm/leads".to_owned(),
+        static_headers: BTreeMap::new(),
+    });
+    payload.execution_config.response_cache = Some(ResponseCachePolicy { ttl_ms: 5_000 });
+    payload.execution_config.auth_profile_ref = Some("auth_profile_01".into());
+
+    let response = client
+        .post(format!("{base_url}/operations"))
+        .json(&payload)
+        .send()
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response.json::<Value>().await.unwrap();
+
+    assert_eq!(status, reqwest::StatusCode::BAD_REQUEST);
+    assert_eq!(body["error"]["code"], "validation_error");
+    assert_eq!(
+        body["error"]["context"]["field"],
+        "execution_config.auth_profile_ref"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn exports_single_workspace_but_rejects_access_lifecycle() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("workspace_access");
+    let base_url = spawn_admin_api(build_test_app(registry.clone(), storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let second_user_id = registry
+        .upsert_bootstrap_user("operator-2@crank.local", "Operator Two", "external-hash")
+        .await
+        .unwrap();
+    registry
+        .ensure_membership(
+            &WorkspaceId::new(DEFAULT_WORKSPACE_ID),
+            &second_user_id,
+            MembershipRole::Viewer,
+        )
+        .await
+        .unwrap();
+
+    let update_member_status = client
+        .patch(format!("{base_url}/members/{}", second_user_id.as_str()))
+        .json(&json!({ "role": "admin" }))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(update_member_status, reqwest::StatusCode::NOT_FOUND);
+
+    let exported = assert_success_json(
+        client
+            .get(format!("{base_url}/export"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(
+        exported["workspace"]["workspace"]["id"],
+        DEFAULT_WORKSPACE_ID
+    );
+    assert!(exported.get("memberships").is_none());
+    assert!(exported.get("invitations").is_none());
+
+    let delete_member_status = client
+        .delete(format!("{base_url}/members/{}", second_user_id.as_str()))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(delete_member_status, reqwest::StatusCode::NOT_FOUND);
+
+    let delete_workspace_status = client
+        .delete(base_url.as_ref())
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(
+        delete_workspace_status,
+        reqwest::StatusCode::METHOD_NOT_ALLOWED
+    );
+
+    let root_url = base_url
+        .as_ref()
+        .split("/api/admin/workspaces/")
+        .next()
+        .unwrap();
+    let still_available = assert_success_json(
+        client
+            .get(format!(
+                "{root_url}/api/admin/workspaces/{DEFAULT_WORKSPACE_ID}"
+            ))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(still_available["workspace"]["id"], DEFAULT_WORKSPACE_ID);
+}
+
+#[tokio::test]
+#[serial]
+async fn seeds_demo_assets_for_live_ui() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("demo_seed");
+    let service = test_service(
+        registry.clone(),
+        storage_root,
+        test_auth_settings(),
+        test_secret_crypto(),
+    );
+
+    service.bootstrap_admin_user().await.unwrap();
+    service.seed_demo_assets().await.unwrap();
+    service.seed_demo_assets().await.unwrap();
+
+    let owner = registry
+        .get_auth_user_by_email(TEST_AUTH_EMAIL)
+        .await
+        .unwrap()
+        .unwrap();
+    let workspaces = service
+        .list_workspaces_for_user(&owner.user.id)
+        .await
+        .unwrap();
+    assert!(
+        workspaces
+            .iter()
+            .any(|item| item.workspace.slug == "default")
+    );
+    assert_eq!(workspaces.len(), 1);
+
+    let default_workspace_id = WorkspaceId::new(DEFAULT_WORKSPACE_ID);
+    let operations = service
+        .list_operations(&default_workspace_id)
+        .await
+        .unwrap();
+    assert!(operations.len() >= 2);
+
+    let agents = service.list_agents(&default_workspace_id).await.unwrap();
+    assert!(!agents.is_empty());
+
+    assert!(agents.iter().any(|agent| agent.key_count > 0));
+
+    let logs = service
+        .list_logs(
+            &default_workspace_id,
+            admin_api::service::LogsQuery {
+                level: None,
+                search: None,
+                source: None,
+                operation_id: None,
+                agent_id: None,
+                period: None,
+                limit: Some(20),
+            },
+        )
+        .await
+        .unwrap();
+    assert!(!logs.is_empty());
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn updates_profile_and_changes_password() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("settings_profile");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let root_url = base_url
+        .as_ref()
+        .split("/api/admin/workspaces/")
+        .next()
+        .unwrap()
+        .to_owned();
+    let client = authorized_client(&base_url).await;
+
+    let profile = assert_success_json(
+        client
+            .get(format!("{root_url}/api/auth/profile"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(profile["user"]["email"], TEST_AUTH_EMAIL);
+
+    let updated_profile = assert_success_json(
+        client
+            .patch(format!("{root_url}/api/auth/profile"))
+            .json(&json!({
+                "display_name": "Updated Owner",
+                "email": "updated-owner@crank.local"
+            }))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(updated_profile["user"]["display_name"], "Updated Owner");
+    assert_eq!(
+        updated_profile["user"]["email"],
+        "updated-owner@crank.local"
+    );
+
+    let password_status = client
+        .post(format!("{root_url}/api/auth/password"))
+        .json(&json!({
+            "current_password": TEST_AUTH_PASSWORD,
+            "new_password": "updated-password-123"
+        }))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(password_status, reqwest::StatusCode::NO_CONTENT);
+
+    let relogin_client = reqwest::Client::builder()
+        .cookie_store(true)
+        .build()
+        .unwrap();
+    let relogin_status = relogin_client
+        .post(format!("{root_url}/api/auth/login"))
+        .json(&json!({
+            "email": "updated-owner@crank.local",
+            "password": "updated-password-123"
+        }))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(relogin_status, reqwest::StatusCode::OK);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn rejects_multi_workspace_session_switching_in_community() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("session_workspace");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let root_url = base_url
+        .as_ref()
+        .split("/api/admin/workspaces/")
+        .next()
+        .unwrap()
+        .to_owned();
+    let client = authorized_client(&base_url).await;
+
+    let create_workspace_status = client
+        .post(format!("{root_url}/api/admin/workspaces"))
+        .json(&json!({
+            "slug": "growth-lab",
+            "display_name": "Growth Lab",
+            "settings": {}
+        }))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(
+        create_workspace_status,
+        reqwest::StatusCode::METHOD_NOT_ALLOWED
+    );
+
+    let switch_status = client
+        .post(format!("{root_url}/api/auth/current-workspace"))
+        .json(&json!({ "workspace_id": DEFAULT_WORKSPACE_ID }))
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(switch_status, reqwest::StatusCode::NOT_FOUND);
+
+    let session = assert_success_json(
+        client
+            .get(format!("{root_url}/api/auth/session"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(session["current_workspace_id"], DEFAULT_WORKSPACE_ID);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn exposes_logs_and_usage_from_real_test_runs() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("observability");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_observability",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+
+    client
+        .post(format!("{base_url}/operations/{operation_id}/test-runs"))
+        .json(&json!({
+            "version": 1,
+            "input": { "email": "user@example.com" }
+        }))
+        .send()
+        .await
+        .unwrap();
+
+    let logs = client
+        .get(format!("{base_url}/logs?period=7d"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let log_id = logs["items"][0]["log"]["id"].as_str().unwrap().to_owned();
+    let log_detail = client
+        .get(format!("{base_url}/logs/{log_id}"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let usage = client
+        .get(format!("{base_url}/usage?period=7d"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_usage = client
+        .get(format!(
+            "{base_url}/usage/operations/{operation_id}?period=7d"
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(logs["items"][0]["log"]["source"], "admin_test_run");
+    assert_eq!(logs["items"][0]["operation_name"], "crm_observability");
+    assert_eq!(log_detail["log"]["status"], "ok");
+    assert_eq!(usage["summary"]["rollup"]["calls_total"], 1);
+    assert_eq!(usage["summary"]["rollup"]["calls_ok"], 1);
+    assert_eq!(
+        usage["operations"][0]["operation_name"],
+        "crm_observability"
+    );
+    assert_eq!(operation_usage["rollup"]["calls_total"], 1);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn preserves_request_id_for_test_run_invocations() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("observability_request_id");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_observability_request_id",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+
+    let response = client
+        .post(format!("{base_url}/operations/{operation_id}/test-runs"))
+        .header("x-request-id", "req_test_123")
+        .json(&json!({
+            "version": 1,
+            "input": { "email": "user@example.com" }
+        }))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(
+        response.headers()["x-request-id"].to_str().unwrap(),
+        "req_test_123"
+    );
+    response.error_for_status().unwrap();
+
+    let logs = client
+        .get(format!("{base_url}/logs?period=7d"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(logs["items"][0]["log"]["request_id"], "req_test_123");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn generates_request_id_for_test_run_invocations() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("observability_generated_request_id");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_observability_generated_request_id",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+
+    let response = client
+        .post(format!("{base_url}/operations/{operation_id}/test-runs"))
+        .json(&json!({
+            "version": 1,
+            "input": { "email": "user@example.com" }
+        }))
+        .send()
+        .await
+        .unwrap();
+    let request_id = response
+        .headers()
+        .get("x-request-id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_owned();
+
+    assert!(!request_id.is_empty());
+    response.error_for_status().unwrap();
+
+    let logs = client
+        .get(format!("{base_url}/logs?period=7d"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(logs["items"][0]["log"]["request_id"], request_id);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn returns_structured_context_for_missing_operation_usage() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("missing_operation_usage");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_missing_usage",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+
+    let response = client
+        .get(format!(
+            "{base_url}/usage/operations/{operation_id}?period=7d"
+        ))
+        .send()
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response.json::<Value>().await.unwrap();
+
+    assert_eq!(status, reqwest::StatusCode::NOT_FOUND);
+    assert_eq!(body["error"]["code"], "not_found");
+    assert_eq!(
+        body["error"]["message"],
+        format!("usage for operation {operation_id} was not found")
+    );
+    assert_eq!(
+        body["error"]["context"],
+        json!({
+            "operation_id": operation_id
+        })
+    );
+}

apps/admin-api/tests/integration/operations_agents.rs

@@ -0,0 +1,792 @@
+#![allow(dead_code, unused_imports)]
+
+use super::common::*;
+
+use std::{
+    collections::BTreeMap,
+    env, fmt,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use async_trait::async_trait;
+use axum::{Json, Router, routing::post};
+use crank_core::{
+    ExecutionConfig, HttpMethod, MembershipRole, OperationSecurityLevel, Protocol,
+    ResponseCachePolicy, RestTarget, SecretKind, Target, ToolDescription, WorkspaceId,
+};
+use crank_core::{IdentityError, IdentityProvider, IdentityProviderKind, LoginOutcome};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::PostgresRegistry;
+use crank_runtime::SecretCrypto;
+use crank_schema::{Schema, SchemaKind};
+use serde_json::{Value, json};
+use serial_test::serial;
+use tokio::net::TcpListener;
+
+use admin_api::{
+    app::build_app,
+    auth::{AuthSettings, BootstrapAdminConfig, hash_password},
+    service::{AdminService, AdminServiceBuilder, OperationPayload},
+    state::AppState,
+};
+
+const DEFAULT_WORKSPACE_ID: &str = "ws_default";
+const TEST_AUTH_EMAIL: &str = "owner@crank.local";
+const TEST_AUTH_PASSWORD: &str = "test-password";
+const TEST_PASSWORD_PEPPER: &str = "test-password-pepper";
+const TEST_SESSION_SECRET: &str = "test-session-secret";
+const TEST_MASTER_KEY: &str = "test-master-key";
+
+struct TestServer {
+    base_url: String,
+    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        let shutdown = self.shutdown.take();
+        let handle = self.handle.take();
+
+        tokio::task::block_in_place(|| {
+            if let Some(shutdown) = shutdown {
+                let _ = shutdown.send(());
+            }
+            if let Some(handle) = handle {
+                let _ = tokio::runtime::Handle::current().block_on(handle);
+            }
+        });
+    }
+}
+
+impl fmt::Display for TestServer {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.base_url)
+    }
+}
+
+impl AsRef<str> for TestServer {
+    fn as_ref(&self) -> &str {
+        &self.base_url
+    }
+}
+
+struct RejectingIdentityProvider;
+
+#[async_trait]
+impl IdentityProvider for RejectingIdentityProvider {
+    fn id(&self) -> &str {
+        "rejecting-test-provider"
+    }
+
+    fn kind(&self) -> IdentityProviderKind {
+        IdentityProviderKind::Password
+    }
+
+    async fn login_password(
+        &self,
+        _payload: crank_core::LoginPayload,
+    ) -> Result<LoginOutcome, IdentityError> {
+        Err(IdentityError::BadCredentials)
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn creates_publishes_and_tests_rest_operation() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("lifecycle");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_create_lead",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+
+    let listed = client
+        .get(format!("{base_url}/operations"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let published = client
+        .post(format!("{base_url}/operations/{operation_id}/publish"))
+        .json(&json!({ "version": 1 }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let test_run = client
+        .post(format!("{base_url}/operations/{operation_id}/test-runs"))
+        .json(&json!({
+            "version": 1,
+            "input": { "email": "user@example.com" }
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(listed["items"][0]["name"], "crm_create_lead");
+    assert_eq!(
+        listed["items"][0]["target_url"],
+        format!("{upstream_base_url}/crm/leads")
+    );
+    assert_eq!(listed["items"][0]["target_action"], "POST");
+    assert_eq!(published["published_version"], 1);
+    assert_eq!(test_run["ok"], true);
+    assert_eq!(
+        test_run["request_preview"]["body"]["email"],
+        "user@example.com"
+    );
+    assert_eq!(test_run["response_preview"]["id"], "lead_123");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn updates_archives_and_deletes_operation() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("operation_mutations");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_mutable_operation",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+
+    let listed = client
+        .get(format!("{base_url}/operations"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(listed["total"], 1);
+    assert_eq!(listed["items"][0]["category"], "sales");
+    assert_eq!(
+        listed["items"][0]["target_url"],
+        format!("{upstream_base_url}/crm/leads")
+    );
+    assert_eq!(listed["items"][0]["target_action"], "POST");
+
+    let updated = client
+        .patch(format!("{base_url}/operations/{operation_id}"))
+        .json(&json!({
+            "display_name": "Create Lead Updated",
+            "category": "marketing",
+            "target": {
+                "kind": "rest",
+                "base_url": upstream_base_url,
+                "method": "POST",
+                "path_template": "/crm/leads",
+                "static_headers": {}
+            },
+            "input_schema": {
+                "type": "object",
+                "required": true,
+                "nullable": false,
+                "fields": {
+                    "email": {
+                        "type": "string",
+                        "required": true,
+                        "nullable": false
+                    }
+                }
+            },
+            "output_schema": {
+                "type": "object",
+                "required": true,
+                "nullable": false,
+                "fields": {
+                    "id": {
+                        "type": "string",
+                        "required": true,
+                        "nullable": false
+                    }
+                }
+            },
+            "input_mapping": {
+                "rules": [
+                    {
+                        "source": "$.mcp.email",
+                        "target": "$.request.body.email",
+                        "required": true
+                    }
+                ]
+            },
+            "output_mapping": {
+                "rules": [
+                    {
+                        "source": "$.response.body.id",
+                        "target": "$.output.id",
+                        "required": true
+                    }
+                ]
+            },
+            "execution_config": {
+                "timeout_ms": 1000,
+                "headers": {}
+            },
+            "tool_description": {
+                "title": "Create Lead Updated",
+                "description": "Creates a CRM lead",
+                "tags": ["crm"]
+            }
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(updated["status"], "draft");
+
+    let detail = client
+        .get(format!("{base_url}/operations/{operation_id}"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(detail["display_name"], "Create Lead Updated");
+    assert_eq!(detail["category"], "marketing");
+
+    let archived = client
+        .post(format!("{base_url}/operations/{operation_id}/archive"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(archived["status"], "archived");
+
+    let deleted = client
+        .delete(format!("{base_url}/operations/{operation_id}"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(deleted["operation_id"], operation_id);
+
+    let missing = client
+        .get(format!("{base_url}/operations/{operation_id}"))
+        .send()
+        .await
+        .unwrap();
+    let missing_status = missing.status();
+    let missing = missing.json::<Value>().await.unwrap();
+    assert_eq!(missing_status, reqwest::StatusCode::NOT_FOUND);
+    assert_eq!(missing["error"]["code"], "not_found");
+    assert_eq!(
+        missing["error"]["context"],
+        json!({
+            "operation_id": operation_id
+        })
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn creates_binds_and_publishes_agent() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("agent_lifecycle");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let operation = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_create_lead_agent",
+        ))
+        .send()
+        .await
+        .unwrap();
+    let operation = assert_success_json(operation).await;
+    let operation_id = operation["operation_id"].as_str().unwrap().to_owned();
+
+    client
+        .post(format!("{base_url}/operations/{operation_id}/publish"))
+        .json(&json!({ "version": 1 }))
+        .send()
+        .await
+        .unwrap();
+
+    let agent = client
+        .post(format!("{base_url}/agents"))
+        .json(&json!({
+            "slug": "sales-assistant",
+            "display_name": "Sales Assistant",
+            "description": "Curated sales toolset",
+            "instructions": {},
+            "tool_selection_policy": {}
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let agent_id = agent["agent_id"].as_str().unwrap().to_owned();
+
+    let bindings = client
+        .post(format!("{base_url}/agents/{agent_id}/bindings"))
+        .json(&json!([
+            {
+                "operation_id": operation_id,
+                "operation_version": 1,
+                "tool_name": "crm_create_lead_agent",
+                "tool_title": "Create Lead",
+                "enabled": true
+            }
+        ]))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let published = client
+        .post(format!("{base_url}/agents/{agent_id}/publish"))
+        .json(&json!({ "version": 1 }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(
+        bindings["bindings"][0]["tool_name"],
+        "crm_create_lead_agent"
+    );
+    assert_eq!(published["published_version"], 1);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn agent_publish_skips_draft_operation_bindings_and_preserves_draft() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("agent_publish_filters_drafts");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let published_operation = assert_success_json(
+        client
+            .post(format!("{base_url}/operations"))
+            .json(&test_operation_payload(
+                &upstream_base_url,
+                "crm_published_tool",
+            ))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let published_operation_id = published_operation["operation_id"]
+        .as_str()
+        .unwrap()
+        .to_owned();
+    assert_success_json(
+        client
+            .post(format!(
+                "{base_url}/operations/{published_operation_id}/publish"
+            ))
+            .json(&json!({ "version": 1 }))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+
+    let draft_operation = assert_success_json(
+        client
+            .post(format!("{base_url}/operations"))
+            .json(&test_operation_payload(
+                &upstream_base_url,
+                "crm_draft_tool",
+            ))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let draft_operation_id = draft_operation["operation_id"].as_str().unwrap().to_owned();
+
+    let agent = assert_success_json(
+        client
+            .post(format!("{base_url}/agents"))
+            .json(&json!({
+                "slug": "collaborative-agent",
+                "display_name": "Collaborative Agent",
+                "description": "Agent with published and draft tools",
+                "instructions": {},
+                "tool_selection_policy": {}
+            }))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let agent_id = agent["agent_id"].as_str().unwrap().to_owned();
+
+    assert_success_json(
+        client
+            .post(format!("{base_url}/agents/{agent_id}/bindings"))
+            .json(&json!([
+                {
+                    "operation_id": published_operation_id,
+                    "operation_version": 1,
+                    "tool_name": "crm_published_tool",
+                    "tool_title": "Published Tool",
+                    "tool_description_override": null,
+                    "enabled": true
+                },
+                {
+                    "operation_id": draft_operation_id,
+                    "operation_version": 1,
+                    "tool_name": "crm_draft_tool",
+                    "tool_title": "Draft Tool",
+                    "tool_description_override": null,
+                    "enabled": true
+                }
+            ]))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+
+    let published = assert_success_json(
+        client
+            .post(format!("{base_url}/agents/{agent_id}/publish"))
+            .json(&json!({ "version": 1 }))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let agent_detail = assert_success_json(
+        client
+            .get(format!("{base_url}/agents/{agent_id}"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let published_version = assert_success_json(
+        client
+            .get(format!("{base_url}/agents/{agent_id}/versions/1"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let draft_version = assert_success_json(
+        client
+            .get(format!("{base_url}/agents/{agent_id}/versions/2"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+
+    assert_eq!(published["published_version"], 1);
+    assert_eq!(agent_detail["current_draft_version"], 2);
+    assert_eq!(agent_detail["latest_published_version"], 1);
+    assert_eq!(published_version["bindings"].as_array().unwrap().len(), 1);
+    assert_eq!(
+        published_version["bindings"][0]["operation_id"],
+        published_operation_id
+    );
+    assert_eq!(draft_version["bindings"].as_array().unwrap().len(), 2);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn updates_lists_and_deletes_agent() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("agent_mutations");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let operation = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_create_lead_agents_page",
+        ))
+        .send()
+        .await
+        .unwrap();
+    let operation = assert_success_json(operation).await;
+    let operation_id = operation["operation_id"].as_str().unwrap().to_owned();
+
+    client
+        .post(format!("{base_url}/operations/{operation_id}/publish"))
+        .json(&json!({ "version": 1 }))
+        .send()
+        .await
+        .unwrap();
+
+    let created = client
+        .post(format!("{base_url}/agents"))
+        .json(&json!({
+            "slug": "support-team",
+            "display_name": "Support Team",
+            "description": "Support workflows",
+            "instructions": {},
+            "tool_selection_policy": {}
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let agent_id = created["agent_id"].as_str().unwrap().to_owned();
+
+    client
+        .post(format!("{base_url}/agents/{agent_id}/bindings"))
+        .json(&json!([
+            {
+                "operation_id": operation_id,
+                "operation_version": 1,
+                "tool_name": "crm_create_lead_agents_page",
+                "tool_title": "Create Lead",
+                "tool_description_override": null,
+                "enabled": true
+            }
+        ]))
+        .send()
+        .await
+        .unwrap();
+
+    client
+        .post(format!("{base_url}/agents/{agent_id}/publish"))
+        .json(&json!({ "version": 1 }))
+        .send()
+        .await
+        .unwrap();
+
+    let listed = client
+        .get(format!("{base_url}/agents"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(listed["items"][0]["operation_count"], 1);
+    assert_eq!(listed["items"][0]["operation_ids"][0], operation_id);
+    assert_eq!(
+        listed["items"][0]["mcp_endpoint"],
+        "/mcp/v1/default/support-team"
+    );
+    assert_eq!(listed["items"][0]["status"], "published");
+
+    let updated = client
+        .patch(format!("{base_url}/agents/{agent_id}"))
+        .json(&json!({
+            "slug": "support-escalation",
+            "display_name": "Support Escalation",
+            "description": "Escalation workflows"
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(updated["agent_id"], agent_id);
+
+    let detail = client
+        .get(format!("{base_url}/agents/{agent_id}"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(detail["slug"], "support-escalation");
+    assert_eq!(detail["display_name"], "Support Escalation");
+    assert_eq!(detail["operation_count"], 1);
+    assert_eq!(detail["mcp_endpoint"], "/mcp/v1/default/support-escalation");
+
+    let deleted = client
+        .delete(format!("{base_url}/agents/{agent_id}"))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(deleted["agent_id"], agent_id);
+
+    let missing = client
+        .get(format!("{base_url}/agents/{agent_id}"))
+        .send()
+        .await
+        .unwrap();
+    let missing_status = missing.status();
+    let missing = missing.json::<Value>().await.unwrap();
+    assert_eq!(missing_status, reqwest::StatusCode::NOT_FOUND);
+    assert_eq!(missing["error"]["code"], "not_found");
+    assert_eq!(
+        missing["error"]["context"],
+        json!({
+            "agent_id": agent_id
+        })
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn unpublishes_and_archives_agent() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("agent_statuses");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let operation = assert_success_json(
+        client
+            .post(format!("{base_url}/operations"))
+            .json(&test_operation_payload(
+                &upstream_base_url,
+                "crm_create_lead_agent_status",
+            ))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let operation_id = operation["operation_id"].as_str().unwrap().to_owned();
+
+    client
+        .post(format!("{base_url}/operations/{operation_id}/publish"))
+        .json(&json!({ "version": 1 }))
+        .send()
+        .await
+        .unwrap();
+
+    let created = assert_success_json(
+        client
+            .post(format!("{base_url}/agents"))
+            .json(&json!({
+                "slug": "sales-routing",
+                "display_name": "Sales Routing",
+                "description": "Routing agent",
+                "instructions": {},
+                "tool_selection_policy": {}
+            }))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    let agent_id = created["agent_id"].as_str().unwrap().to_owned();
+
+    client
+        .post(format!("{base_url}/agents/{agent_id}/bindings"))
+        .json(&json!([
+            {
+                "operation_id": operation_id,
+                "operation_version": 1,
+                "tool_name": "crm_create_lead_agent_status",
+                "tool_title": "Create Lead",
+                "tool_description_override": null,
+                "enabled": true
+            }
+        ]))
+        .send()
+        .await
+        .unwrap();
+
+    client
+        .post(format!("{base_url}/agents/{agent_id}/publish"))
+        .json(&json!({ "version": 1 }))
+        .send()
+        .await
+        .unwrap();
+
+    let unpublished = assert_success_json(
+        client
+            .post(format!("{base_url}/agents/{agent_id}/unpublish"))
+            .json(&json!({}))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(unpublished["agent_id"], agent_id);
+
+    let draft_detail = assert_success_json(
+        client
+            .get(format!("{base_url}/agents/{agent_id}"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(draft_detail["status"], "draft");
+    assert_eq!(draft_detail["latest_published_version"], 1);
+
+    let archived = assert_success_json(
+        client
+            .post(format!("{base_url}/agents/{agent_id}/archive"))
+            .json(&json!({}))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(archived["agent_id"], agent_id);
+
+    let archived_detail = assert_success_json(
+        client
+            .get(format!("{base_url}/agents/{agent_id}"))
+            .send()
+            .await
+            .unwrap(),
+    )
+    .await;
+    assert_eq!(archived_detail["status"], "archived");
+}

apps/admin-api/tests/integration/secrets_import_auth.rs

@@ -0,0 +1,452 @@
+#![allow(dead_code, unused_imports)]
+
+use super::common::*;
+
+use std::{
+    collections::BTreeMap,
+    env, fmt,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use async_trait::async_trait;
+use axum::{Json, Router, routing::post};
+use crank_core::{
+    ExecutionConfig, HttpMethod, MembershipRole, OperationSecurityLevel, Protocol,
+    ResponseCachePolicy, RestTarget, SecretKind, Target, ToolDescription, WorkspaceId,
+};
+use crank_core::{IdentityError, IdentityProvider, IdentityProviderKind, LoginOutcome};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::PostgresRegistry;
+use crank_runtime::SecretCrypto;
+use crank_schema::{Schema, SchemaKind};
+use serde_json::{Value, json};
+use serial_test::serial;
+use tokio::net::TcpListener;
+
+use admin_api::{
+    app::build_app,
+    auth::{AuthSettings, BootstrapAdminConfig, hash_password},
+    service::{AdminService, AdminServiceBuilder, OperationPayload},
+    state::AppState,
+};
+
+const DEFAULT_WORKSPACE_ID: &str = "ws_default";
+const TEST_AUTH_EMAIL: &str = "owner@crank.local";
+const TEST_AUTH_PASSWORD: &str = "test-password";
+const TEST_PASSWORD_PEPPER: &str = "test-password-pepper";
+const TEST_SESSION_SECRET: &str = "test-session-secret";
+const TEST_MASTER_KEY: &str = "test-master-key";
+
+struct TestServer {
+    base_url: String,
+    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        let shutdown = self.shutdown.take();
+        let handle = self.handle.take();
+
+        tokio::task::block_in_place(|| {
+            if let Some(shutdown) = shutdown {
+                let _ = shutdown.send(());
+            }
+            if let Some(handle) = handle {
+                let _ = tokio::runtime::Handle::current().block_on(handle);
+            }
+        });
+    }
+}
+
+impl fmt::Display for TestServer {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.base_url)
+    }
+}
+
+impl AsRef<str> for TestServer {
+    fn as_ref(&self) -> &str {
+        &self.base_url
+    }
+}
+
+struct RejectingIdentityProvider;
+
+#[async_trait]
+impl IdentityProvider for RejectingIdentityProvider {
+    fn id(&self) -> &str {
+        "rejecting-test-provider"
+    }
+
+    fn kind(&self) -> IdentityProviderKind {
+        IdentityProviderKind::Password
+    }
+
+    async fn login_password(
+        &self,
+        _payload: crank_core::LoginPayload,
+    ) -> Result<LoginOutcome, IdentityError> {
+        Err(IdentityError::BadCredentials)
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn manages_auth_profiles_and_yaml_upsert() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("yaml");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+    let secret = client
+        .post(format!("{base_url}/secrets"))
+        .json(&json!({
+            "name": "crm-api-token",
+            "kind": SecretKind::Token,
+            "value": { "token": "super-secret-token" }
+        }))
+        .send()
+        .await
+        .unwrap();
+    let secret = assert_success_json(secret).await;
+    let secret_id = secret["id"].as_str().unwrap();
+
+    let auth_profile = client
+        .post(format!("{base_url}/auth-profiles"))
+        .json(&json!({
+            "name": "crm-header",
+            "kind": "api_key_header",
+            "config": {
+                "api_key_header": {
+                    "header_name": "X-Api-Key",
+                    "secret_id": secret_id
+                }
+            }
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_export_target",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap();
+    let yaml = client
+        .get(format!("{base_url}/operations/{operation_id}/export"))
+        .send()
+        .await
+        .unwrap()
+        .text()
+        .await
+        .unwrap();
+    let imported = client
+        .post(format!("{base_url}/operations/import?mode=upsert"))
+        .body(yaml)
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(auth_profile["kind"], "api_key_header");
+    assert_eq!(
+        auth_profile["config"]["api_key_header"]["secret_id"],
+        secret_id
+    );
+    assert_eq!(imported["operation_id"], operation_id);
+    assert_eq!(imported["version"], 2);
+    assert_eq!(imported["import_mode"], "upsert");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn manages_workspace_secrets_without_exposing_plaintext() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("secrets");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/secrets"))
+        .json(&json!({
+            "name": "crm-api-token",
+            "kind": SecretKind::Token,
+            "value": { "token": "super-secret-token" }
+        }))
+        .send()
+        .await
+        .unwrap();
+    let created = assert_success_json(created).await;
+    let secret_id = created["id"].as_str().unwrap().to_owned();
+
+    let listed = client
+        .get(format!("{base_url}/secrets"))
+        .send()
+        .await
+        .unwrap();
+    let listed = assert_success_json(listed).await;
+
+    let fetched = client
+        .get(format!("{base_url}/secrets/{secret_id}"))
+        .send()
+        .await
+        .unwrap();
+    let fetched = assert_success_json(fetched).await;
+
+    let rotated = client
+        .post(format!("{base_url}/secrets/{secret_id}/rotate"))
+        .json(&json!({
+            "value": { "token": "rotated-token" }
+        }))
+        .send()
+        .await
+        .unwrap();
+    let rotated = assert_success_json(rotated).await;
+
+    let deleted = client
+        .delete(format!("{base_url}/secrets/{secret_id}"))
+        .send()
+        .await
+        .unwrap();
+    let deleted = assert_success_json(deleted).await;
+
+    let missing = client
+        .get(format!("{base_url}/secrets/{secret_id}"))
+        .send()
+        .await
+        .unwrap();
+    let missing_status = missing.status();
+    let missing = missing.json::<Value>().await.unwrap();
+
+    assert_eq!(created["name"], "crm-api-token");
+    assert_eq!(created["kind"], "token");
+    assert_eq!(created["current_version"], 1);
+    assert!(created.get("value").is_none());
+    assert_eq!(listed["items"].as_array().unwrap().len(), 1);
+    assert_eq!(fetched["id"], secret_id);
+    assert!(fetched.get("value").is_none());
+    assert_eq!(rotated["current_version"], 2);
+    assert_eq!(deleted["ok"], true);
+    assert_eq!(missing_status, reqwest::StatusCode::NOT_FOUND);
+    assert_eq!(missing["error"]["code"], "not_found");
+    assert_eq!(
+        missing["error"]["context"],
+        json!({
+            "secret_id": secret_id
+        })
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn returns_structured_context_for_missing_operation_version() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("missing_operation_version");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_missing_operation_version",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap().to_owned();
+
+    let response = client
+        .get(format!("{base_url}/operations/{operation_id}/versions/99"))
+        .send()
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response.json::<Value>().await.unwrap();
+
+    assert_eq!(status, reqwest::StatusCode::NOT_FOUND);
+    assert_eq!(body["error"]["code"], "not_found");
+    assert_eq!(
+        body["error"]["message"],
+        format!("operation version 99 for {operation_id} was not found")
+    );
+    assert_eq!(
+        body["error"]["context"],
+        json!({
+            "operation_id": operation_id,
+            "version": 99
+        })
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn rejects_auth_profile_with_missing_secret() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("missing_secret_auth");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let response = client
+        .post(format!("{base_url}/auth-profiles"))
+        .json(&json!({
+            "name": "crm-header",
+            "kind": "api_key_header",
+            "config": {
+                "api_key_header": {
+                    "header_name": "X-Api-Key",
+                    "secret_id": "secret_missing"
+                }
+            }
+        }))
+        .send()
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response.json::<Value>().await.unwrap();
+
+    assert_eq!(status, reqwest::StatusCode::NOT_FOUND);
+    assert_eq!(body["error"]["code"], "not_found");
+    assert_eq!(
+        body["error"]["message"],
+        "secret secret_missing was not found"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn rejects_deleting_secret_referenced_by_auth_profile() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("secret_references");
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let secret = client
+        .post(format!("{base_url}/secrets"))
+        .json(&json!({
+            "name": "crm-api-token",
+            "kind": SecretKind::Token,
+            "value": { "token": "super-secret-token" }
+        }))
+        .send()
+        .await
+        .unwrap();
+    let secret = assert_success_json(secret).await;
+    let secret_id = secret["id"].as_str().unwrap();
+
+    let auth_profile = client
+        .post(format!("{base_url}/auth-profiles"))
+        .json(&json!({
+            "name": "crm-header",
+            "kind": "api_key_header",
+            "config": {
+                "api_key_header": {
+                    "header_name": "X-Api-Key",
+                    "secret_id": secret_id
+                }
+            }
+        }))
+        .send()
+        .await
+        .unwrap();
+    let auth_profile = assert_success_json(auth_profile).await;
+    let auth_profile_id = auth_profile["id"].as_str().unwrap();
+
+    let response = client
+        .delete(format!("{base_url}/secrets/{secret_id}"))
+        .send()
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response.json::<Value>().await.unwrap();
+
+    assert_eq!(status, reqwest::StatusCode::CONFLICT);
+    assert_eq!(body["error"]["code"], "conflict");
+    assert_eq!(
+        body["error"]["message"],
+        format!("secret {secret_id} is referenced by auth profile {auth_profile_id}")
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn uploads_samples_and_generates_draft() {
+    let registry = test_registry().await;
+    let storage_root = test_storage_root("draft");
+    let upstream_base_url = spawn_upstream_server().await;
+    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
+    let client = authorized_client(&base_url).await;
+
+    let created = client
+        .post(format!("{base_url}/operations"))
+        .json(&test_operation_payload(
+            &upstream_base_url,
+            "crm_draft_target",
+        ))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+    let operation_id = created["operation_id"].as_str().unwrap();
+
+    client
+        .post(format!(
+            "{base_url}/operations/{operation_id}/samples/input-json"
+        ))
+        .json(&json!({ "email": "user@example.com", "name": "Ada" }))
+        .send()
+        .await
+        .unwrap();
+    client
+        .post(format!(
+            "{base_url}/operations/{operation_id}/samples/output-json"
+        ))
+        .json(&json!({ "id": "lead_123", "status": "created" }))
+        .send()
+        .await
+        .unwrap();
+
+    let generated = client
+        .post(format!(
+            "{base_url}/operations/{operation_id}/drafts/generate"
+        ))
+        .json(&json!({
+            "sources": ["input_json_sample", "output_json_sample"]
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(generated["generated_draft"]["status"], "available");
+    assert_eq!(
+        generated["input_schema"]["fields"]["email"]["type"],
+        "string"
+    );
+    assert_eq!(
+        generated["output_mapping"]["rules"][0]["target"],
+        "$.output.id"
+    );
+}

apps/mcp-server/tests/integration.rs

@@ -0,0 +1,5 @@
+mod integration {
+    mod catalog_access;
+    mod common;
+    mod transport_protocol;
+}

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

@@ -0,0 +1,636 @@
+#![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, ExecutionConfig, HttpMethod,
+    Operation, OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope,
+    PlatformApiKeyStatus, Protocol, RestTarget, Target, ToolDescription, WorkspaceId,
+};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::{
+    CreateAgentRequest, 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_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));
+}

apps/mcp-server/tests/integration/common.rs

@@ -0,0 +1,491 @@
+#![allow(dead_code, unused_imports)]
+
+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, ExecutionConfig, HttpMethod,
+    Operation, OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope,
+    PlatformApiKeyStatus, Protocol, RestTarget, Target, ToolDescription, WorkspaceId,
+};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::{
+    CreateAgentRequest, 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,
+    )
+}
+
+pub(super) async fn initialize_session(
+    client: &reqwest::Client,
+    mcp_url: &str,
+    api_key: &str,
+) -> String {
+    let initialize_response = client
+        .post(mcp_url)
+        .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!(initialize_response.status(), reqwest::StatusCode::OK);
+    let session_id = initialize_response
+        .headers()
+        .get("MCP-Session-Id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_owned();
+
+    let initialized_response = client
+        .post(mcp_url)
+        .header(header::ACCEPT, "application/json, text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &session_id)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .json(&json!({
+            "jsonrpc": "2.0",
+            "method": "notifications/initialized",
+            "params": {}
+        }))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(initialized_response.status(), reqwest::StatusCode::ACCEPTED);
+
+    session_id
+}
+
+pub(super) async fn post_jsonrpc(
+    client: &reqwest::Client,
+    mcp_url: &str,
+    api_key: &str,
+    session_id: Option<&str>,
+    payload: Value,
+) -> Value {
+    post_jsonrpc_response(client, mcp_url, api_key, session_id, None, payload)
+        .await
+        .json::<Value>()
+        .await
+        .unwrap()
+}
+
+pub(super) async fn post_jsonrpc_response(
+    client: &reqwest::Client,
+    mcp_url: &str,
+    api_key: &str,
+    session_id: Option<&str>,
+    request_id: Option<&str>,
+    payload: Value,
+) -> reqwest::Response {
+    let mut request = client
+        .post(mcp_url)
+        .header(header::ACCEPT, "application/json, text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Protocol-Version", "2025-11-25");
+
+    if let Some(session_id) = session_id {
+        request = request.header("MCP-Session-Id", session_id);
+    }
+
+    if let Some(request_id) = request_id {
+        request = request.header("x-request-id", request_id);
+    }
+
+    request.json(&payload).send().await.unwrap()
+}
+
+pub(super) async fn create_platform_api_key(
+    registry: &PostgresRegistry,
+    agent_slug: &str,
+    name: &str,
+    scopes: &[PlatformApiKeyScope],
+) -> String {
+    let secret = format!("crk_{}_{}", name, uuid::Uuid::now_v7().simple());
+    let api_key = PlatformApiKey {
+        id: PlatformApiKeyId::new(format!("pk_{name}")),
+        workspace_id: test_workspace_id(),
+        agent_id: Some(test_agent_id(agent_slug)),
+        name: name.to_owned(),
+        prefix: secret.chars().take(16).collect(),
+        scopes: scopes.to_vec(),
+        status: PlatformApiKeyStatus::Active,
+        created_at: OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
+        last_used_at: None,
+    };
+
+    registry
+        .create_platform_api_key(CreatePlatformApiKeyRequest {
+            api_key: &api_key,
+            secret_hash: &hash_access_secret(&secret),
+        })
+        .await
+        .unwrap();
+
+    secret
+}
+
+pub(super) fn hash_access_secret(secret: &str) -> String {
+    let digest = Sha256::digest(secret.as_bytes());
+    URL_SAFE_NO_PAD.encode(digest)
+}
+
+pub(super) async fn spawn_upstream_server() -> String {
+    let app = Router::new()
+        .route("/sse/logs", get(stream_logs))
+        .route("/crm/leads", post(create_lead))
+        .route("/crm/slow-leads", post(create_slow_lead));
+    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)
+}
+
+pub(super) async fn spawn_mcp_server(app: Router) -> String {
+    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)
+}
+
+pub(super) fn agent_mcp_url(base_url: &str, agent_slug: &str) -> String {
+    format!("{base_url}/v1/{}/{}", test_workspace_slug(), agent_slug)
+}
+
+pub(super) async fn publish_agent_for_operation(
+    registry: &PostgresRegistry,
+    operation: &Operation<Schema, MappingSet>,
+    agent_slug: &str,
+) {
+    publish_agent_with_bindings(registry, agent_slug, vec![binding_for_operation(operation)]).await;
+}
+
+pub(super) async fn publish_agent_with_bindings(
+    registry: &PostgresRegistry,
+    agent_slug: &str,
+    bindings: Vec<AgentOperationBinding>,
+) {
+    let agent_id = AgentId::new(format!("agent_{agent_slug}"));
+    let agent = Agent {
+        id: agent_id.clone(),
+        workspace_id: test_workspace_id(),
+        slug: agent_slug.to_owned(),
+        display_name: format!("Agent {agent_slug}"),
+        description: "Curated MCP toolset".to_owned(),
+        status: AgentStatus::Draft,
+        current_draft_version: 1,
+        latest_published_version: None,
+        created_at: OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
+        updated_at: OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
+        published_at: None,
+    };
+    let version = AgentVersion {
+        agent_id: agent_id.clone(),
+        version: 1,
+        status: AgentStatus::Draft,
+        instructions: json!({}),
+        tool_selection_policy: json!({}),
+        created_at: OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
+    };
+
+    registry
+        .create_agent(CreateAgentRequest {
+            agent: &agent,
+            version: &version,
+            bindings: &bindings,
+        })
+        .await
+        .unwrap();
+    registry
+        .publish_agent(PublishAgentRequest {
+            workspace_id: &test_workspace_id(),
+            agent_id: &agent_id,
+            version: 1,
+            published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
+            published_by: Some("alice"),
+        })
+        .await
+        .unwrap();
+}
+
+pub(super) fn binding_for_operation(
+    operation: &Operation<Schema, MappingSet>,
+) -> AgentOperationBinding {
+    AgentOperationBinding {
+        operation_id: operation.id.clone(),
+        operation_version: 1,
+        tool_name: operation.name.clone(),
+        tool_title: operation.tool_description.title.clone(),
+        tool_description_override: None,
+        enabled: true,
+    }
+}
+
+pub(super) async fn create_lead(Json(payload): Json<Value>) -> Json<Value> {
+    Json(json!({
+        "id": "lead_123",
+        "email": payload["email"]
+    }))
+}
+
+pub(super) async fn create_slow_lead(Json(payload): Json<Value>) -> Json<Value> {
+    sleep(Duration::from_millis(250)).await;
+
+    Json(json!({
+        "id": "lead_123",
+        "email": payload["email"]
+    }))
+}
+
+pub(super) async fn stream_logs()
+-> Sse<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
+    let events = vec![
+        json!({ "level": "info", "message": "sync started" }),
+        json!({ "level": "warn", "message": "cache warmup slow" }),
+        json!({ "level": "error", "message": "upstream timeout" }),
+    ];
+    let stream = stream::iter(events.into_iter().map(|payload| {
+        Ok::<_, std::convert::Infallible>(Event::default().data(payload.to_string()))
+    }));
+    Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
+}
+
+pub(super) async fn test_registry() -> PostgresRegistry {
+    let database_url = crank_test_support::postgres_schema_url("test_mcp_server").await;
+    PostgresRegistry::connect(&database_url).await.unwrap()
+}
+
+pub(super) fn test_operation(base_url: &str, name: &str) -> Operation<Schema, MappingSet> {
+    Operation {
+        id: OperationId::new(format!("op_{name}")),
+        name: name.to_owned(),
+        display_name: "Create Lead".to_owned(),
+        category: "sales".to_owned(),
+        protocol: Protocol::Rest,
+        security_level: crank_core::OperationSecurityLevel::Standard,
+        status: OperationStatus::Published,
+        version: 1,
+        target: Target::Rest(RestTarget {
+            base_url: base_url.to_owned(),
+            method: HttpMethod::Post,
+            path_template: "/crm/leads".to_owned(),
+            static_headers: BTreeMap::new(),
+        }),
+        input_schema: object_schema("email"),
+        output_schema: object_schema("id"),
+        input_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.mcp.email".to_owned(),
+                target: "$.request.body.email".to_owned(),
+                required: true,
+                default_value: None,
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        },
+        output_mapping: MappingSet {
+            rules: vec![MappingRule {
+                source: "$.response.body.id".to_owned(),
+                target: "$.output.id".to_owned(),
+                required: true,
+                default_value: None,
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        },
+        execution_config: ExecutionConfig {
+            timeout_ms: 1_000,
+            retry_policy: None,
+            response_cache: None,
+            idempotency: None,
+            safety: None,
+            auth_profile_ref: None,
+            headers: BTreeMap::new(),
+        },
+        tool_description: ToolDescription {
+            title: "Create Lead".to_owned(),
+            description: "Creates a CRM lead".to_owned(),
+            tags: Vec::new(),
+            examples: Vec::new(),
+        },
+        samples: None,
+        generated_draft: None,
+        config_export: None,
+        wizard_state: None,
+        created_at: OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
+        updated_at: OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(),
+        published_at: Some(OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap()),
+    }
+}
+
+pub(super) fn object_schema(field_name: &str) -> Schema {
+    Schema {
+        kind: SchemaKind::Object,
+        description: None,
+        required: true,
+        nullable: false,
+        default_value: None,
+        fields: BTreeMap::from([(
+            field_name.to_owned(),
+            Schema {
+                kind: SchemaKind::String,
+                description: None,
+                required: true,
+                nullable: false,
+                default_value: None,
+                fields: BTreeMap::new(),
+                items: None,
+                enum_values: Vec::new(),
+                variants: Vec::new(),
+            },
+        )]),
+        items: None,
+        enum_values: Vec::new(),
+        variants: Vec::new(),
+    }
+}

apps/mcp-server/tests/integration/transport_protocol.rs

@@ -0,0 +1,930 @@
+#![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, ExecutionConfig, HttpMethod,
+    Operation, OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope,
+    PlatformApiKeyStatus, Protocol, RestTarget, Target, ToolDescription, WorkspaceId,
+};
+use crank_mapping::{MappingRule, MappingSet};
+use crank_registry::{
+    CreateAgentRequest, 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 initializes_lists_and_calls_published_tool_via_mcp() {
+    let registry = test_registry().await;
+    let upstream_base_url = spawn_upstream_server().await;
+    let operation = test_operation(&upstream_base_url, "crm_create_lead");
+
+    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-rest").await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-rest",
+        "mcp-rest",
+        &[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 mcp_url = agent_mcp_url(&base_url, "sales-rest");
+    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+    let tools = post_jsonrpc(
+        &client,
+        &mcp_url,
+        &api_key,
+        Some(&initialized_session),
+        json!({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "method": "tools/list",
+            "params": {}
+        }),
+    )
+    .await;
+    let call_result = post_jsonrpc(
+        &client,
+        &mcp_url,
+        &api_key,
+        Some(&initialized_session),
+        json!({
+            "jsonrpc": "2.0",
+            "id": 3,
+            "method": "tools/call",
+            "params": {
+                "name": "crm_create_lead",
+                "arguments": {
+                    "email": "user@example.com"
+                }
+            }
+        }),
+    )
+    .await;
+
+    assert_eq!(tools["result"]["tools"][0]["name"], "crm_create_lead");
+    assert_eq!(
+        call_result["result"]["structuredContent"],
+        json!({ "id": "lead_123" })
+    );
+    assert_eq!(call_result["result"]["isError"], false);
+
+    let logs = registry
+        .list_invocation_logs(ListInvocationLogsQuery {
+            workspace_id: &test_workspace_id(),
+            level: None,
+            search_text: None,
+            source: Some(crank_core::InvocationSource::AgentToolCall),
+            operation_id: Some(&operation.id),
+            agent_id: None,
+            created_after: None,
+            limit: 10,
+        })
+        .await
+        .unwrap();
+
+    assert_eq!(logs.len(), 1);
+    assert_eq!(
+        logs[0].log.source,
+        crank_core::InvocationSource::AgentToolCall
+    );
+    assert_eq!(logs[0].log.status, crank_core::InvocationStatus::Ok);
+    assert_eq!(logs[0].log.tool_name, "crm_create_lead");
+
+    let keys = registry
+        .list_platform_api_keys(&test_workspace_id())
+        .await
+        .unwrap();
+    assert!(keys[0].api_key.last_used_at.is_some());
+}
+
+#[tokio::test]
+async fn preserves_request_id_for_tool_call_invocations() {
+    let registry = test_registry().await;
+    let upstream_base_url = spawn_upstream_server().await;
+    let operation = test_operation(&upstream_base_url, "crm_request_id");
+
+    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-request-id").await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-request-id",
+        "mcp-request-id",
+        &[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 mcp_url = agent_mcp_url(&base_url, "sales-request-id");
+    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+    let response = post_jsonrpc_response(
+        &client,
+        &mcp_url,
+        &api_key,
+        Some(&initialized_session),
+        Some("req_test_123"),
+        json!({
+            "jsonrpc": "2.0",
+            "id": 3,
+            "method": "tools/call",
+            "params": {
+                "name": "crm_request_id",
+                "arguments": {
+                    "email": "user@example.com"
+                }
+            }
+        }),
+    )
+    .await;
+
+    assert_eq!(
+        response.headers()["x-request-id"].to_str().unwrap(),
+        "req_test_123"
+    );
+    let call_result = response.json::<Value>().await.unwrap();
+    assert_eq!(call_result["result"]["isError"], false);
+
+    let logs = registry
+        .list_invocation_logs(ListInvocationLogsQuery {
+            workspace_id: &test_workspace_id(),
+            level: None,
+            search_text: None,
+            source: Some(crank_core::InvocationSource::AgentToolCall),
+            operation_id: Some(&operation.id),
+            agent_id: None,
+            created_after: None,
+            limit: 10,
+        })
+        .await
+        .unwrap();
+
+    assert_eq!(logs.len(), 1);
+    assert_eq!(logs[0].log.request_id.as_deref(), Some("req_test_123"));
+}
+
+#[tokio::test]
+async fn generates_request_id_for_tool_call_responses_and_logs() {
+    let registry = test_registry().await;
+    let upstream_base_url = spawn_upstream_server().await;
+    let operation = test_operation(&upstream_base_url, "crm_generated_request_id");
+
+    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-generated-request-id").await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-generated-request-id",
+        "mcp-generated-request-id",
+        &[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 mcp_url = agent_mcp_url(&base_url, "sales-generated-request-id");
+    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+    let response = post_jsonrpc_response(
+        &client,
+        &mcp_url,
+        &api_key,
+        Some(&initialized_session),
+        None,
+        json!({
+            "jsonrpc": "2.0",
+            "id": 3,
+            "method": "tools/call",
+            "params": {
+                "name": "crm_generated_request_id",
+                "arguments": {
+                    "email": "user@example.com"
+                }
+            }
+        }),
+    )
+    .await;
+    let request_id = response
+        .headers()
+        .get("x-request-id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_owned();
+    assert!(!request_id.is_empty());
+
+    let call_result = response.json::<Value>().await.unwrap();
+    assert_eq!(call_result["result"]["isError"], false);
+
+    let logs = registry
+        .list_invocation_logs(ListInvocationLogsQuery {
+            workspace_id: &test_workspace_id(),
+            level: None,
+            search_text: None,
+            source: Some(crank_core::InvocationSource::AgentToolCall),
+            operation_id: Some(&operation.id),
+            agent_id: None,
+            created_after: None,
+            limit: 10,
+        })
+        .await
+        .unwrap();
+
+    assert_eq!(logs.len(), 1);
+    assert_eq!(logs[0].log.request_id.as_deref(), Some(request_id.as_str()));
+}
+
+#[tokio::test]
+async fn emits_request_id_in_mcp_ingress_logs() {
+    let registry = test_registry().await;
+    let upstream_base_url = spawn_upstream_server().await;
+    let operation = test_operation(&upstream_base_url, "crm_request_trace");
+
+    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-request-trace").await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-request-trace",
+        "mcp-request-trace",
+        &[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 mcp_url = agent_mcp_url(&base_url, "sales-request-trace");
+    let writer = SharedLogWriter::default();
+    let subscriber = tracing_subscriber::registry().with(
+        tracing_subscriber::fmt::layer()
+            .with_writer(writer.clone())
+            .without_time()
+            .with_ansi(false)
+            .with_target(false)
+            .compact()
+            .with_filter(LevelFilter::INFO),
+    );
+
+    let _ = tracing::subscriber::set_global_default(subscriber);
+    let response = post_jsonrpc_response(
+        &client,
+        &mcp_url,
+        &api_key,
+        None,
+        Some("req_mcp_trace_123"),
+        json!({
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "initialize",
+            "params": {
+                "protocolVersion": "2025-03-26"
+            }
+        }),
+    )
+    .await;
+
+    assert_eq!(
+        response.headers()["x-request-id"].to_str().unwrap(),
+        "req_mcp_trace_123"
+    );
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+
+    let logs = writer.output();
+    assert!(
+        logs.contains("mcp request received"),
+        "captured logs did not include ingress marker: {logs}"
+    );
+    assert!(logs.contains("req_mcp_trace_123"));
+    assert!(logs.contains("sales-request-trace"));
+    assert!(logs.contains("default"));
+    assert!(logs.contains("initialize"));
+}
+
+#[tokio::test]
+async fn requires_initialized_notification_before_tool_methods() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-init", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-init",
+        "mcp-init",
+        &[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-init");
+    let initialize_response = client
+        .post(&mcp_url)
+        .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();
+    let session_id = initialize_response
+        .headers()
+        .get("MCP-Session-Id")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .to_owned();
+    let tools_list = client
+        .post(&mcp_url)
+        .header(header::ACCEPT, "application/json, text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", session_id)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .json(&json!({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "method": "tools/list",
+            "params": {}
+        }))
+        .send()
+        .await
+        .unwrap()
+        .json::<Value>()
+        .await
+        .unwrap();
+
+    assert_eq!(tools_list["error"]["code"], -32002);
+}
+
+#[tokio::test]
+async fn initialize_can_return_sse_response_when_client_prefers_event_stream() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-sse-init", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-sse-init",
+        "mcp-sse-init",
+        &[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-sse-init"))
+        .header(header::ACCEPT, "text/event-stream, application/json")
+        .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::OK);
+    assert_eq!(
+        response
+            .headers()
+            .get(header::CONTENT_TYPE)
+            .unwrap()
+            .to_str()
+            .unwrap(),
+        "text/event-stream"
+    );
+    assert!(response.headers().get("MCP-Session-Id").is_some());
+
+    let body = response.text().await.unwrap();
+    assert!(body.contains("\"jsonrpc\":\"2.0\""));
+    assert!(body.contains("\"protocolVersion\":\"2025-11-25\""));
+}
+
+#[tokio::test]
+async fn get_opens_sse_stream_for_initialized_session() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-get-sse", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-get-sse",
+        "mcp-get-sse",
+        &[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 mcp_url = agent_mcp_url(&base_url, "sales-get-sse");
+    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+    let response = client
+        .get(&mcp_url)
+        .header(header::ACCEPT, "text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &initialized_session)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+    assert_eq!(
+        response
+            .headers()
+            .get(header::CONTENT_TYPE)
+            .unwrap()
+            .to_str()
+            .unwrap(),
+        "text/event-stream"
+    );
+    assert_eq!(
+        response
+            .headers()
+            .get("MCP-Session-Id")
+            .unwrap()
+            .to_str()
+            .unwrap(),
+        initialized_session
+    );
+}
+
+#[tokio::test]
+async fn get_returns_not_found_for_expired_transport_session() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-expired-session", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-expired-session",
+        "mcp-expired-session",
+        &[PlatformApiKeyScope::Read],
+    )
+    .await;
+    let session_store = Arc::new(InMemorySessionStore::default());
+    let session_id = session_store
+        .create(
+            "2025-11-25",
+            test_workspace_slug(),
+            "sales-expired-session",
+            OffsetDateTime::parse("2026-05-01T10:00:00Z", &Rfc3339).unwrap(),
+            Some(OffsetDateTime::parse("2026-05-01T10:00:01Z", &Rfc3339).unwrap()),
+        )
+        .await
+        .unwrap();
+    session_store
+        .mark_initialized(
+            &session_id,
+            OffsetDateTime::parse("2026-05-01T10:00:01Z", &Rfc3339).unwrap(),
+        )
+        .await
+        .unwrap();
+    let base_url = spawn_mcp_server(build_test_app_with_store(
+        registry,
+        Duration::from_millis(0),
+        Some("https://crank.example.com".to_owned()),
+        RequestRateLimitConfig::new(10_000, 10_000).unwrap(),
+        session_store,
+        std::sync::Arc::new(CommunityMachineCredentialVerifier),
+    ))
+    .await;
+    let client = reqwest::Client::new();
+
+    let response = client
+        .get(agent_mcp_url(&base_url, "sales-expired-session"))
+        .header(header::ACCEPT, "text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &session_id)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), reqwest::StatusCode::NOT_FOUND);
+}
+
+#[tokio::test]
+async fn rejects_rapid_transport_get_requests_with_429() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-get-rate-limit", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-get-rate-limit",
+        "mcp-get-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, 2).unwrap(),
+    ))
+    .await;
+    let client = reqwest::Client::new();
+    let mcp_url = agent_mcp_url(&base_url, "sales-get-rate-limit");
+    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+    let first_response = client
+        .get(&mcp_url)
+        .header(header::ACCEPT, "text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &initialized_session)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(first_response.status(), reqwest::StatusCode::OK);
+
+    let second_response = client
+        .get(&mcp_url)
+        .header(header::ACCEPT, "text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &initialized_session)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(
+        second_response.status(),
+        reqwest::StatusCode::TOO_MANY_REQUESTS
+    );
+    assert!(second_response.headers().get(header::RETRY_AFTER).is_some());
+}
+
+#[tokio::test]
+async fn get_requires_session_header() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-get-sse-missing", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-get-sse-missing",
+        "mcp-get-sse-missing",
+        &[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 response = client
+        .get(agent_mcp_url(&base_url, "sales-get-sse-missing"))
+        .header(header::ACCEPT, "text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
+}
+
+#[tokio::test]
+async fn delete_terminates_transport_session() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-delete-session", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-delete-session",
+        "mcp-delete-session",
+        &[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-delete-session");
+    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+    let delete_response = client
+        .delete(&mcp_url)
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &initialized_session)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(delete_response.status(), reqwest::StatusCode::NO_CONTENT);
+
+    let after_delete = client
+        .get(&mcp_url)
+        .header(header::ACCEPT, "text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &initialized_session)
+        .header("MCP-Protocol-Version", "2025-11-25")
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(after_delete.status(), reqwest::StatusCode::NOT_FOUND);
+}
+
+#[tokio::test]
+async fn initialize_accepts_json_only_response_negotiation() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-json-accept", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-json-accept",
+        "mcp-json-accept",
+        &[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 response = client
+        .post(agent_mcp_url(&base_url, "sales-json-accept"))
+        .header(header::ACCEPT, "application/json")
+        .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::OK);
+    assert_eq!(
+        response
+            .headers()
+            .get(header::CONTENT_TYPE)
+            .and_then(|value| value.to_str().ok())
+            .unwrap(),
+        "application/json"
+    );
+}
+
+#[tokio::test]
+async fn rejects_get_with_protocol_version_mismatch() {
+    let registry = test_registry().await;
+    publish_agent_with_bindings(&registry, "sales-get-bad-version", vec![]).await;
+    let api_key = create_platform_api_key(
+        &registry,
+        "sales-get-bad-version",
+        "mcp-get-bad-version",
+        &[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-get-bad-version");
+    let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+    let response = client
+        .get(&mcp_url)
+        .header(header::ACCEPT, "text/event-stream")
+        .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
+        .header("MCP-Session-Id", &initialized_session)
+        .header("MCP-Protocol-Version", "2025-06-18")
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
+}