crank/apps/admin-api/src/service.rs

use std::collections::BTreeMap;
use std::path::PathBuf;

use base64::{
    Engine as _,
    engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD},
};
use crank_adapter_grpc::test_support as grpc_test_support;
use crank_core::{
    Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AuthConfig, AuthKind,
    AuthProfile, AuthProfileId, ConfigExport, ExportMode, GeneratedDraft, GeneratedDraftStatus,
    InvitationId, InvitationStatus, InvitationToken, InvocationLevel, InvocationLog,
    InvocationLogId, InvocationSource, InvocationStatus, MembershipRole, OperationId,
    OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus,
    Protocol, SampleId, Samples, Secret, SecretId, SecretKind, SecretStatus, Target, UsagePeriod,
    UserId, UserSessionId, Workspace, WorkspaceId, WorkspaceStatus,
};
use crank_mapping::{JsonPathRoot, MappingSet, infer_mapping_from_samples};
use crank_proto::{ProtoService, services_from_descriptor_set_bytes};
use crank_registry::{
    AgentSummary, AgentVersionRecord, CreateAgentRequest, CreateInvitationRequest,
    CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest,
    CreateVersionRequest, CreateWorkspaceRequest, InvitationRecord, InvocationLogRecord,
    ListInvocationLogsQuery, MembershipRecord, OperationAgentRef, OperationSampleMetadata,
    OperationSummary, OperationUsageSummary, OperationVersionRecord, PlatformApiKeyRecord,
    PostgresRegistry, PublishAgentRequest, PublishRequest, RegistryOperation, RotateSecretRequest,
    SampleKind, SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
    SaveSampleMetadataRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket,
    UsageOperationBreakdown, UsageQuery, UsageSummary, UsageTimelinePoint,
    WorkspaceMembershipRecord, WorkspaceRecord,
};
use crank_runtime::{PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeOperation};
use crank_schema::Schema;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
use tracing::{info, instrument};
use uuid::Uuid;

use crate::{
    auth::{
        AuthSettings, AuthenticatedSession, SessionCookie, create_session_cookie, hash_password,
        hash_session_secret, verify_password,
    },
    error::ApiError,
    secret_crypto::SecretCrypto,
    storage::LocalArtifactStorage,
};

#[derive(Clone)]
pub struct AdminService {
    registry: PostgresRegistry,
    runtime: RuntimeExecutor,
    storage: LocalArtifactStorage,
    auth_settings: AuthSettings,
    secret_crypto: SecretCrypto,
}

#[derive(Clone, Debug, Deserialize)]
pub struct LoginPayload {
    pub email: String,
    pub password: String,
}

#[derive(Clone, Debug, Serialize)]
pub struct SessionResponse {
    pub user: crank_core::User,
    pub memberships: Vec<WorkspaceMembershipRecord>,
    pub current_workspace_id: Option<String>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct UpdateProfilePayload {
    pub display_name: String,
    pub email: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct ChangePasswordPayload {
    pub current_password: String,
    pub new_password: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct UpdateCurrentWorkspacePayload {
    pub workspace_id: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct OperationPayload {
    pub name: String,
    pub display_name: String,
    #[serde(default = "default_operation_category")]
    pub category: String,
    pub protocol: Protocol,
    pub target: Target,
    pub input_schema: Schema,
    pub output_schema: Schema,
    pub input_mapping: MappingSet,
    pub output_mapping: MappingSet,
    pub execution_config: crank_core::ExecutionConfig,
    pub tool_description: crank_core::ToolDescription,
}

#[derive(Clone, Debug, Deserialize)]
pub struct NewVersionPayload {
    #[serde(flatten)]
    pub operation: OperationPayload,
    #[serde(default)]
    pub change_note: Option<String>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct PublishPayload {
    pub version: u32,
}

#[derive(Clone, Debug, Deserialize)]
pub struct TestRunPayload {
    pub version: u32,
    pub input: Value,
}

#[derive(Clone, Debug, Serialize)]
pub struct TestRunResult {
    pub ok: bool,
    pub request_preview: Value,
    pub response_preview: Value,
    pub errors: Vec<Value>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct AuthProfilePayload {
    pub name: String,
    pub kind: AuthKind,
    pub config: AuthConfig,
}

#[derive(Clone, Debug, Deserialize)]
pub struct SecretPayload {
    pub name: String,
    pub kind: SecretKind,
    pub value: Value,
}

#[derive(Clone, Debug, Deserialize)]
pub struct RotateSecretPayload {
    pub value: Value,
}

#[derive(Clone, Debug, Deserialize)]
pub struct WorkspacePayload {
    pub slug: String,
    pub display_name: String,
    #[serde(default)]
    pub settings: Value,
}

#[derive(Clone, Debug, Deserialize)]
pub struct UpdateWorkspacePayload {
    pub slug: Option<String>,
    pub display_name: Option<String>,
    pub status: Option<WorkspaceStatus>,
    pub settings: Option<Value>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct AgentPayload {
    pub slug: String,
    pub display_name: String,
    pub description: String,
    #[serde(default)]
    pub instructions: Value,
    #[serde(default)]
    pub tool_selection_policy: Value,
}

#[derive(Clone, Debug, Deserialize)]
pub struct UpdateAgentPayload {
    pub slug: String,
    pub display_name: String,
    pub description: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct AgentBindingPayload {
    pub operation_id: String,
    pub operation_version: u32,
    pub tool_name: String,
    pub tool_title: String,
    pub tool_description_override: Option<String>,
    #[serde(default = "default_enabled")]
    pub enabled: bool,
}

#[derive(Clone, Debug, Serialize)]
pub struct CreatedAgentResponse {
    pub agent_id: String,
    pub workspace_id: String,
    pub version: u32,
    pub status: AgentStatus,
}

#[derive(Clone, Debug, Serialize)]
pub struct PublishAgentResponse {
    pub agent_id: String,
    pub workspace_id: String,
    pub published_version: u32,
    pub published_at: String,
}

#[derive(Clone, Debug, Serialize)]
pub struct AgentSummaryView {
    pub id: String,
    pub workspace_id: String,
    pub slug: String,
    pub display_name: String,
    pub description: String,
    pub status: AgentStatus,
    pub current_draft_version: u32,
    pub latest_published_version: Option<u32>,
    pub created_at: String,
    pub updated_at: String,
    pub published_at: Option<String>,
    pub operation_count: usize,
    pub operation_ids: Vec<String>,
    pub key_count: usize,
    pub calls_today: u64,
    pub mcp_endpoint: String,
}

#[derive(Clone, Debug, Serialize)]
pub struct AgentMutationResult {
    pub agent_id: String,
    pub workspace_id: String,
    pub updated_at: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct InvitationPayload {
    pub email: String,
    pub role: MembershipRole,
    pub expires_at: Option<String>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct UpdateMembershipPayload {
    pub role: MembershipRole,
}

#[derive(Clone, Debug, Serialize)]
pub struct CreatedInvitationResponse {
    pub invitation: InvitationRecord,
    pub invite_token: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct PlatformApiKeyPayload {
    pub name: String,
    pub scopes: Vec<PlatformApiKeyScope>,
}

#[derive(Clone, Debug, Serialize)]
pub struct CreatedPlatformApiKeyResponse {
    pub api_key: PlatformApiKeyRecord,
    pub secret: String,
}

#[derive(Clone, Debug, Serialize)]
pub struct WorkspaceExportResponse {
    pub workspace: WorkspaceRecord,
    pub memberships: Vec<MembershipRecord>,
    pub invitations: Vec<Value>,
    pub operations: Vec<OperationSummaryView>,
    pub agents: Vec<AgentSummaryView>,
    pub platform_api_keys: Vec<PlatformApiKeyRecord>,
    pub exported_at: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct LogsQuery {
    pub level: Option<InvocationLevel>,
    pub search: Option<String>,
    pub source: Option<InvocationSource>,
    pub operation_id: Option<String>,
    pub agent_id: Option<String>,
    pub period: Option<UsagePeriod>,
    pub limit: Option<u32>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct UsageRequestQuery {
    pub period: Option<UsagePeriod>,
    pub source: Option<InvocationSource>,
}

#[derive(Clone, Debug, Serialize)]
pub struct UsageOverviewResponse {
    pub summary: UsageSummary,
    pub timeline: Vec<UsageTimelinePoint>,
    pub operations: Vec<UsageOperationBreakdown>,
    pub agents: Vec<UsageAgentBreakdown>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct GenerateDraftPayload {
    #[serde(default)]
    pub sources: Vec<String>,
}

#[derive(Clone, Debug, Serialize)]
pub struct DraftGenerationResult {
    pub generated_draft: GeneratedDraft,
    pub input_schema: Schema,
    pub output_schema: Schema,
    pub input_mapping: MappingSet,
    pub output_mapping: MappingSet,
}

#[derive(Clone, Debug, Deserialize)]
pub struct ImportQuery {
    #[serde(default)]
    pub mode: ImportMode,
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ImportMode {
    #[default]
    Create,
    Upsert,
}

#[derive(Clone, Debug, Deserialize)]
pub struct ExportQuery {
    pub version: Option<u32>,
    #[serde(default = "default_export_mode")]
    pub mode: ExportMode,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct YamlOperationDocument {
    pub format_version: String,
    pub kind: String,
    pub operation: RegistryOperation,
}

#[derive(Clone, Debug, Serialize)]
pub struct CreatedOperationResponse {
    pub operation_id: String,
    pub workspace_id: String,
    pub version: u32,
    pub status: OperationStatus,
    pub updated_at: String,
}

#[derive(Clone, Debug, Serialize)]
pub struct PublishResponse {
    pub operation_id: String,
    pub workspace_id: String,
    pub published_version: u32,
    pub published_at: String,
}

#[derive(Clone, Debug, Deserialize)]
pub struct UpdateOperationPayload {
    pub display_name: String,
    #[serde(default = "default_operation_category")]
    pub category: String,
    pub target: Target,
    pub input_schema: Schema,
    pub output_schema: Schema,
    pub input_mapping: MappingSet,
    pub output_mapping: MappingSet,
    pub execution_config: crank_core::ExecutionConfig,
    pub tool_description: crank_core::ToolDescription,
}

#[derive(Clone, Debug, Serialize)]
pub struct OperationUsageSummaryView {
    pub calls_today: u64,
    pub error_rate_pct: f64,
    pub avg_latency_ms: u64,
}

#[derive(Clone, Debug, Serialize)]
pub struct OperationAgentRefView {
    pub agent_id: String,
    pub agent_slug: String,
    pub display_name: String,
}

struct InvocationRecordRequest<'a> {
    workspace_id: &'a WorkspaceId,
    agent_id: Option<&'a AgentId>,
    operation: &'a RegistryOperation,
    source: InvocationSource,
    level: InvocationLevel,
    status: InvocationStatus,
    message: String,
    status_code: Option<u16>,
    error_kind: Option<String>,
    duration_ms: u64,
    request_preview: Value,
    response_preview: Value,
}

#[derive(Clone, Debug, Serialize)]
pub struct OperationSummaryView {
    pub id: String,
    pub workspace_id: String,
    pub name: String,
    pub display_name: String,
    pub category: String,
    pub protocol: Protocol,
    pub target_url: String,
    pub target_action: String,
    pub status: OperationStatus,
    pub current_draft_version: u32,
    pub latest_published_version: Option<u32>,
    pub created_at: String,
    pub updated_at: String,
    pub published_at: Option<String>,
    pub usage_summary: OperationUsageSummaryView,
    pub agent_refs: Vec<OperationAgentRefView>,
}

#[derive(Clone, Debug, Serialize)]
pub struct OperationDetailView {
    pub id: String,
    pub workspace_id: String,
    pub name: String,
    pub display_name: String,
    pub category: String,
    pub protocol: Protocol,
    pub status: OperationStatus,
    pub current_draft_version: u32,
    pub latest_published_version: Option<u32>,
    pub created_at: String,
    pub updated_at: String,
    pub published_at: Option<String>,
    pub draft_version_ref: VersionRef,
    pub published_version_ref: Option<VersionRef>,
    pub agent_refs: Vec<OperationAgentRefView>,
}

#[derive(Clone, Debug, Serialize)]
pub struct VersionRef {
    pub version: u32,
    pub status: OperationStatus,
}

#[derive(Clone, Debug, Serialize)]
pub struct OperationMutationResult {
    pub operation_id: String,
    pub workspace_id: String,
    pub version: u32,
    pub status: OperationStatus,
    pub updated_at: String,
}

#[derive(Clone, Debug, Serialize)]
pub struct ImportResponse {
    pub operation_id: String,
    pub workspace_id: String,
    pub version: u32,
    pub import_mode: ImportMode,
    pub warnings: Vec<String>,
}

#[derive(Clone, Debug, Serialize)]
pub struct DescriptorUploadResponse {
    pub descriptor_id: String,
    pub version: u32,
}

#[derive(Clone, Debug, Serialize)]
pub struct GrpcServiceSummary {
    pub package: String,
    pub service: String,
    pub methods: Vec<GrpcMethodSummary>,
}

#[derive(Clone, Debug, Serialize)]
pub struct GrpcMethodSummary {
    pub name: String,
    pub kind: String,
    pub input_schema: Schema,
    pub output_schema: Schema,
}

fn default_operation_category() -> String {
    "general".to_owned()
}

impl AdminService {
    pub fn new(
        registry: PostgresRegistry,
        storage_root: PathBuf,
        auth_settings: AuthSettings,
        secret_crypto: SecretCrypto,
    ) -> Self {
        Self {
            registry,
            runtime: RuntimeExecutor::new(),
            storage: LocalArtifactStorage::new(storage_root),
            auth_settings,
            secret_crypto,
        }
    }

    pub fn auth_settings(&self) -> &AuthSettings {
        &self.auth_settings
    }

    pub async fn bootstrap_admin_user(&self) -> Result<(), ApiError> {
        let password_hash = hash_password(
            &self.auth_settings.bootstrap_admin.password,
            &self.auth_settings.password_pepper,
        )?;
        let user_id = self
            .registry
            .upsert_bootstrap_user(
                &self.auth_settings.bootstrap_admin.email,
                &self.auth_settings.bootstrap_admin.display_name,
                &password_hash,
            )
            .await?;
        self.registry
            .ensure_membership(
                &WorkspaceId::new("ws_default"),
                &user_id,
                MembershipRole::Owner,
            )
            .await?;

        Ok(())
    }

    pub async fn seed_demo_assets(&self) -> Result<(), ApiError> {
        let admin_user = self
            .registry
            .get_auth_user_by_email(&self.auth_settings.bootstrap_admin.email)
            .await?
            .ok_or_else(|| ApiError::internal("bootstrap admin user was not found"))?;
        let admin_user_id = admin_user.user.id.clone();
        let default_workspace_id = WorkspaceId::new("ws_default");

        self.seed_default_workspace_demo(&admin_user_id, &default_workspace_id)
            .await?;
        self.seed_growth_workspace_demo(&admin_user_id).await?;

        Ok(())
    }

    pub async fn get_session(
        &self,
        session_id: &UserSessionId,
        session_value: &str,
    ) -> Result<Option<AuthenticatedSession>, ApiError> {
        let secret_hash = hash_session_secret(
            session_id,
            session_value,
            &self.auth_settings.session_secret,
        );
        let session = self
            .registry
            .get_user_session(session_id, &secret_hash)
            .await?
            .map(|record| AuthenticatedSession {
                session_id: record.session_id,
                user: record.user,
                memberships: record.memberships,
                current_workspace_id: record.current_workspace_id,
            });

        Ok(session)
    }

    pub async fn touch_session(&self, session_id: &UserSessionId) -> Result<(), ApiError> {
        self.registry.touch_user_session(session_id).await?;
        Ok(())
    }

    pub async fn login(
        &self,
        payload: LoginPayload,
    ) -> Result<(SessionCookie, SessionResponse), ApiError> {
        let user = self
            .registry
            .get_auth_user_by_email(&payload.email)
            .await?
            .ok_or_else(|| ApiError::unauthorized("invalid email or password"))?;

        if !verify_password(
            &payload.password,
            &self.auth_settings.password_pepper,
            &user.password_hash,
        ) {
            return Err(ApiError::unauthorized("invalid email or password"));
        }

        let session_cookie = create_session_cookie(&self.auth_settings)?;
        let secret_hash = hash_session_secret(
            &session_cookie.session_id,
            &session_cookie.value,
            &self.auth_settings.session_secret,
        );
        let memberships = self
            .registry
            .list_workspaces_for_user(&user.user.id)
            .await?;
        let current_workspace_id = memberships
            .first()
            .map(|membership| membership.workspace.id.as_str().to_owned());
        self.registry
            .create_user_session(
                &session_cookie.session_id,
                &user.user.id,
                memberships
                    .first()
                    .map(|membership| &membership.workspace.id),
                &secret_hash,
                &session_cookie.expires_at,
            )
            .await?;

        Ok((
            session_cookie,
            SessionResponse {
                user: user.user,
                memberships,
                current_workspace_id,
            },
        ))
    }

    pub async fn logout(
        &self,
        session_id: &UserSessionId,
        _session_value: &str,
    ) -> Result<(), ApiError> {
        self.registry.revoke_user_session(session_id).await?;
        Ok(())
    }

    pub async fn list_workspaces_for_user(
        &self,
        user_id: &crank_core::UserId,
    ) -> Result<Vec<WorkspaceMembershipRecord>, ApiError> {
        Ok(self.registry.list_workspaces_for_user(user_id).await?)
    }

    pub async fn user_has_workspace_access(
        &self,
        user_id: &crank_core::UserId,
        workspace_id: &WorkspaceId,
    ) -> Result<bool, ApiError> {
        Ok(self
            .registry
            .user_has_workspace_access(user_id, workspace_id)
            .await?)
    }

    #[instrument(skip(self, payload), fields(workspace_slug = %payload.slug, user_id = %user_id.as_str()))]
    pub async fn create_workspace(
        &self,
        user_id: &crank_core::UserId,
        payload: WorkspacePayload,
    ) -> Result<WorkspaceRecord, ApiError> {
        let now = now_string()?;
        let workspace = Workspace {
            id: WorkspaceId::new(new_prefixed_id("ws")),
            slug: payload.slug,
            display_name: payload.display_name,
            status: WorkspaceStatus::Active,
            settings: payload.settings,
            created_at: now.clone(),
            updated_at: now,
        };

        self.registry
            .create_workspace(CreateWorkspaceRequest {
                workspace: &workspace,
            })
            .await?;
        self.registry
            .ensure_membership(&workspace.id, user_id, MembershipRole::Owner)
            .await?;

        Ok(WorkspaceRecord { workspace })
    }

    #[instrument(skip(self))]
    pub async fn session_response(
        &self,
        session_id: &UserSessionId,
        session_value: &str,
    ) -> Result<Option<SessionResponse>, ApiError> {
        Ok(self
            .get_session(session_id, session_value)
            .await?
            .map(|session| SessionResponse {
                user: session.user,
                memberships: session.memberships,
                current_workspace_id: session
                    .current_workspace_id
                    .map(|id| id.as_str().to_owned()),
            }))
    }

    pub async fn update_profile(
        &self,
        user_id: &crank_core::UserId,
        current_workspace_id: Option<&WorkspaceId>,
        payload: UpdateProfilePayload,
    ) -> Result<SessionResponse, ApiError> {
        let display_name = payload.display_name.trim();
        let email = payload.email.trim().to_ascii_lowercase();

        if display_name.is_empty() {
            return Err(ApiError::validation("display name is required"));
        }
        if email.is_empty() || !email.contains('@') {
            return Err(ApiError::validation("a valid email address is required"));
        }

        let user = self
            .registry
            .update_user_profile(user_id, &email, display_name)
            .await?;
        let memberships = self.registry.list_workspaces_for_user(user_id).await?;

        Ok(SessionResponse {
            user,
            memberships,
            current_workspace_id: current_workspace_id.map(|id| id.as_str().to_owned()),
        })
    }

    pub async fn set_current_workspace(
        &self,
        session_id: &UserSessionId,
        user_id: &crank_core::UserId,
        workspace_id: &WorkspaceId,
    ) -> Result<SessionResponse, ApiError> {
        if !self
            .user_has_workspace_access(user_id, workspace_id)
            .await?
        {
            return Err(ApiError::forbidden("workspace access denied"));
        }

        self.registry
            .set_user_session_current_workspace(session_id, workspace_id)
            .await?;

        let user = self
            .registry
            .get_auth_user_by_id(user_id)
            .await?
            .ok_or_else(|| ApiError::not_found(format!("user {} was not found", user_id.as_str())))?
            .user;
        let memberships = self.registry.list_workspaces_for_user(user_id).await?;

        Ok(SessionResponse {
            user,
            memberships,
            current_workspace_id: Some(workspace_id.as_str().to_owned()),
        })
    }

    pub async fn change_password(
        &self,
        user_id: &crank_core::UserId,
        payload: ChangePasswordPayload,
    ) -> Result<(), ApiError> {
        if payload.new_password.len() < 12 {
            return Err(ApiError::validation(
                "new password must be at least 12 characters long",
            ));
        }

        let user = self
            .registry
            .get_auth_user_by_id(user_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!("user {} was not found", user_id.as_str()))
            })?;

        if !verify_password(
            &payload.current_password,
            &self.auth_settings.password_pepper,
            &user.password_hash,
        ) {
            return Err(ApiError::unauthorized("current password is invalid"));
        }

        let password_hash =
            hash_password(&payload.new_password, &self.auth_settings.password_pepper)?;
        self.registry
            .update_user_password(user_id, &password_hash)
            .await?;

        Ok(())
    }

    pub async fn get_workspace(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<WorkspaceRecord, ApiError> {
        self.registry
            .get_workspace(workspace_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!("workspace {} was not found", workspace_id.as_str()))
            })
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str()))]
    pub async fn update_workspace(
        &self,
        workspace_id: &WorkspaceId,
        payload: UpdateWorkspacePayload,
    ) -> Result<WorkspaceRecord, ApiError> {
        let existing = self.get_workspace(workspace_id).await?.workspace;
        let workspace = Workspace {
            id: existing.id,
            slug: payload.slug.unwrap_or(existing.slug),
            display_name: payload.display_name.unwrap_or(existing.display_name),
            status: payload.status.unwrap_or(existing.status),
            settings: payload.settings.unwrap_or(existing.settings),
            created_at: existing.created_at,
            updated_at: now_string()?,
        };

        self.registry
            .update_workspace(UpdateWorkspaceRequest {
                workspace: &workspace,
            })
            .await?;

        Ok(WorkspaceRecord { workspace })
    }

    #[instrument(skip(self))]
    pub async fn list_memberships(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<MembershipRecord>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        Ok(self.registry.list_memberships(workspace_id).await?)
    }

    pub async fn update_membership_role(
        &self,
        workspace_id: &WorkspaceId,
        actor_user_id: &crank_core::UserId,
        target_user_id: &crank_core::UserId,
        payload: UpdateMembershipPayload,
    ) -> Result<Vec<MembershipRecord>, ApiError> {
        let memberships = self.list_memberships(workspace_id).await?;
        let actor_membership = memberships
            .iter()
            .find(|membership| &membership.user.id == actor_user_id)
            .ok_or_else(|| ApiError::forbidden("workspace access denied"))?;
        let target_membership = memberships
            .iter()
            .find(|membership| &membership.user.id == target_user_id)
            .ok_or_else(|| {
                ApiError::not_found(format!(
                    "membership for user {} in workspace {} was not found",
                    target_user_id.as_str(),
                    workspace_id.as_str()
                ))
            })?;

        if !matches!(
            actor_membership.role,
            MembershipRole::Owner | MembershipRole::Admin
        ) {
            return Err(ApiError::forbidden(
                "only owners and admins can manage workspace members",
            ));
        }

        if matches!(target_membership.role, MembershipRole::Owner)
            && !matches!(payload.role, MembershipRole::Owner)
        {
            let owner_count = memberships
                .iter()
                .filter(|membership| matches!(membership.role, MembershipRole::Owner))
                .count();
            if owner_count <= 1 {
                return Err(ApiError::validation(
                    "workspace must keep at least one owner",
                ));
            }
        }

        self.registry
            .update_membership_role(workspace_id, target_user_id, payload.role)
            .await?;
        self.list_memberships(workspace_id).await
    }

    pub async fn remove_membership(
        &self,
        workspace_id: &WorkspaceId,
        actor_user_id: &crank_core::UserId,
        target_user_id: &crank_core::UserId,
    ) -> Result<(), ApiError> {
        let memberships = self.list_memberships(workspace_id).await?;
        let actor_membership = memberships
            .iter()
            .find(|membership| &membership.user.id == actor_user_id)
            .ok_or_else(|| ApiError::forbidden("workspace access denied"))?;
        let target_membership = memberships
            .iter()
            .find(|membership| &membership.user.id == target_user_id)
            .ok_or_else(|| {
                ApiError::not_found(format!(
                    "membership for user {} in workspace {} was not found",
                    target_user_id.as_str(),
                    workspace_id.as_str()
                ))
            })?;

        if !matches!(
            actor_membership.role,
            MembershipRole::Owner | MembershipRole::Admin
        ) {
            return Err(ApiError::forbidden(
                "only owners and admins can manage workspace members",
            ));
        }

        if matches!(target_membership.role, MembershipRole::Owner) {
            let owner_count = memberships
                .iter()
                .filter(|membership| matches!(membership.role, MembershipRole::Owner))
                .count();
            if owner_count <= 1 {
                return Err(ApiError::validation(
                    "workspace must keep at least one owner",
                ));
            }
        }

        self.registry
            .delete_membership(workspace_id, target_user_id)
            .await?;
        Ok(())
    }

    #[instrument(skip(self))]
    pub async fn list_invitations(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<InvitationRecord>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        Ok(self.registry.list_invitations(workspace_id).await?)
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), email = %payload.email))]
    pub async fn create_invitation(
        &self,
        workspace_id: &WorkspaceId,
        payload: InvitationPayload,
    ) -> Result<CreatedInvitationResponse, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;

        let invite_token = generate_access_secret("invite");
        let invitation = InvitationRecord {
            invitation: InvitationToken {
                id: InvitationId::new(new_prefixed_id("inv")),
                workspace_id: workspace_id.clone(),
                email: payload.email,
                role: payload.role,
                status: InvitationStatus::Pending,
                token_hash: hash_access_secret(&invite_token),
                expires_at: match payload.expires_at {
                    Some(expires_at) => expires_at,
                    None => default_invitation_expiry()?,
                },
                created_at: now_string()?,
            },
        };

        self.registry
            .create_invitation(CreateInvitationRequest {
                invitation: &invitation.invitation,
            })
            .await?;

        Ok(CreatedInvitationResponse {
            invitation,
            invite_token,
        })
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), invitation_id = %invitation_id.as_str()))]
    pub async fn delete_invitation(
        &self,
        workspace_id: &WorkspaceId,
        invitation_id: &InvitationId,
    ) -> Result<(), ApiError> {
        self.registry
            .delete_invitation(workspace_id, invitation_id)
            .await?;
        Ok(())
    }

    pub async fn export_workspace(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<WorkspaceExportResponse, ApiError> {
        let workspace = self.get_workspace(workspace_id).await?;
        let memberships = self.list_memberships(workspace_id).await?;
        let invitations = self
            .list_invitations(workspace_id)
            .await?
            .into_iter()
            .map(|record| {
                json!({
                    "id": record.invitation.id,
                    "email": record.invitation.email,
                    "role": record.invitation.role,
                    "status": record.invitation.status,
                    "expires_at": record.invitation.expires_at,
                    "created_at": record.invitation.created_at,
                })
            })
            .collect();
        let operations = self.list_operations(workspace_id).await?;
        let agents = self.list_agents(workspace_id).await?;
        let platform_api_keys = self.list_platform_api_keys(workspace_id).await?;

        Ok(WorkspaceExportResponse {
            workspace,
            memberships,
            invitations,
            operations,
            agents,
            platform_api_keys,
            exported_at: now_string()?,
        })
    }

    pub async fn delete_workspace(
        &self,
        workspace_id: &WorkspaceId,
        actor_user_id: &crank_core::UserId,
    ) -> Result<(), ApiError> {
        let memberships = self.list_memberships(workspace_id).await?;
        let actor_membership = memberships
            .iter()
            .find(|membership| &membership.user.id == actor_user_id)
            .ok_or_else(|| ApiError::forbidden("workspace access denied"))?;

        if !matches!(actor_membership.role, MembershipRole::Owner) {
            return Err(ApiError::forbidden(
                "only workspace owners can delete a workspace",
            ));
        }

        self.registry.delete_workspace(workspace_id).await?;
        Ok(())
    }

    #[instrument(skip(self))]
    pub async fn list_platform_api_keys(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<PlatformApiKeyRecord>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        Ok(self.registry.list_platform_api_keys(workspace_id).await?)
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), key_name = %payload.name))]
    pub async fn create_platform_api_key(
        &self,
        workspace_id: &WorkspaceId,
        payload: PlatformApiKeyPayload,
    ) -> Result<CreatedPlatformApiKeyResponse, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;

        let secret = generate_access_secret("crk");
        let api_key = PlatformApiKeyRecord {
            api_key: PlatformApiKey {
                id: PlatformApiKeyId::new(new_prefixed_id("pk")),
                workspace_id: workspace_id.clone(),
                name: payload.name,
                prefix: secret.chars().take(16).collect(),
                scopes: payload.scopes,
                status: PlatformApiKeyStatus::Active,
                created_at: now_string()?,
                last_used_at: None,
            },
        };

        self.registry
            .create_platform_api_key(CreatePlatformApiKeyRequest {
                api_key: &api_key.api_key,
                secret_hash: &hash_access_secret(&secret),
            })
            .await?;

        Ok(CreatedPlatformApiKeyResponse { api_key, secret })
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), key_id = %key_id.as_str()))]
    pub async fn revoke_platform_api_key(
        &self,
        workspace_id: &WorkspaceId,
        key_id: &PlatformApiKeyId,
    ) -> Result<(), ApiError> {
        self.registry
            .revoke_platform_api_key(workspace_id, key_id, &now_string()?)
            .await?;
        Ok(())
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), key_id = %key_id.as_str()))]
    pub async fn delete_platform_api_key(
        &self,
        workspace_id: &WorkspaceId,
        key_id: &PlatformApiKeyId,
    ) -> Result<(), ApiError> {
        self.registry
            .delete_platform_api_key(workspace_id, key_id)
            .await?;
        Ok(())
    }

    #[instrument(skip(self))]
    pub async fn list_logs(
        &self,
        workspace_id: &WorkspaceId,
        query: LogsQuery,
    ) -> Result<Vec<InvocationLogRecord>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let operation_id = query.operation_id.as_deref().map(OperationId::new);
        let agent_id = query.agent_id.as_deref().map(AgentId::new);
        let (_, created_after, _) = usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;

        self.registry
            .list_invocation_logs(ListInvocationLogsQuery {
                workspace_id,
                level: query.level,
                search_text: query.search.as_deref(),
                source: query.source,
                operation_id: operation_id.as_ref(),
                agent_id: agent_id.as_ref(),
                created_after: Some(&created_after),
                limit: query.limit.unwrap_or(100),
            })
            .await
            .map_err(ApiError::from)
    }

    #[instrument(skip(self))]
    pub async fn get_log(
        &self,
        workspace_id: &WorkspaceId,
        log_id: &InvocationLogId,
    ) -> Result<InvocationLogRecord, ApiError> {
        self.registry
            .get_invocation_log(workspace_id, log_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!("invocation log {} was not found", log_id.as_str()))
            })
    }

    #[instrument(skip(self))]
    pub async fn get_usage_overview(
        &self,
        workspace_id: &WorkspaceId,
        query: UsageRequestQuery,
    ) -> Result<UsageOverviewResponse, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let (period, created_after, bucket) =
            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;
        let usage_query = UsageQuery {
            workspace_id,
            period,
            source: query.source,
            created_after: &created_after,
            bucket,
        };

        let summary = self.registry.summarize_usage(usage_query.clone()).await?;
        let timeline = self
            .registry
            .list_usage_timeline(usage_query.clone())
            .await?;
        let operations = self
            .registry
            .list_usage_by_operation(usage_query.clone())
            .await?;
        let agents = self.registry.list_usage_by_agent(usage_query).await?;

        Ok(UsageOverviewResponse {
            summary,
            timeline,
            operations,
            agents,
        })
    }

    #[instrument(skip(self))]
    pub async fn get_operation_usage(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        query: UsageRequestQuery,
    ) -> Result<crank_registry::UsageRollupRecord, ApiError> {
        self.get_operation(workspace_id, operation_id).await?;
        let (period, created_after, bucket) =
            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;

        self.registry
            .get_usage_for_operation(
                UsageQuery {
                    workspace_id,
                    period,
                    source: query.source,
                    created_after: &created_after,
                    bucket,
                },
                operation_id,
            )
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!(
                    "usage for operation {} was not found",
                    operation_id.as_str()
                ))
            })
    }

    #[instrument(skip(self))]
    pub async fn get_agent_usage(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        query: UsageRequestQuery,
    ) -> Result<crank_registry::UsageRollupRecord, ApiError> {
        self.get_agent(workspace_id, agent_id).await?;
        let (period, created_after, bucket) =
            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;

        self.registry
            .get_usage_for_agent(
                UsageQuery {
                    workspace_id,
                    period,
                    source: query.source,
                    created_after: &created_after,
                    bucket,
                },
                agent_id,
            )
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!(
                    "usage for agent {} was not found",
                    agent_id.as_str()
                ))
            })
    }

    pub async fn list_operations(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<OperationSummaryView>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let summaries = self.registry.list_operations(workspace_id).await?;
        let usage = self
            .registry
            .list_operation_usage_summaries(workspace_id, &today_start_utc()?)
            .await?;
        let agent_refs = self
            .registry
            .list_operation_agent_refs(workspace_id)
            .await?;
        let usage_by_operation = usage_map(usage);
        let refs_by_operation = agent_ref_map(agent_refs);

        Ok(summaries
            .into_iter()
            .map(|summary| {
                let operation_id = summary.id.as_str().to_owned();
                enrich_operation_summary(
                    summary,
                    usage_by_operation
                        .get(&operation_id)
                        .cloned()
                        .unwrap_or_else(default_usage_summary),
                    refs_by_operation
                        .get(&operation_id)
                        .cloned()
                        .unwrap_or_default(),
                )
            })
            .collect())
    }

    #[instrument(skip(self))]
    pub async fn get_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<OperationDetailView, ApiError> {
        let summary = self
            .registry
            .get_operation_summary(workspace_id, operation_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!("operation {} was not found", operation_id.as_str()))
            })?;
        let agent_refs = self
            .registry
            .list_operation_agent_refs(workspace_id)
            .await?;
        let refs = agent_ref_map(agent_refs)
            .remove(operation_id.as_str())
            .unwrap_or_default();

        Ok(OperationDetailView {
            id: summary.id.as_str().to_owned(),
            workspace_id: summary.workspace_id.as_str().to_owned(),
            name: summary.name,
            display_name: summary.display_name,
            category: summary.category,
            protocol: summary.protocol,
            status: summary.status,
            current_draft_version: summary.current_draft_version,
            latest_published_version: summary.latest_published_version,
            created_at: summary.created_at,
            updated_at: summary.updated_at,
            published_at: summary.published_at,
            draft_version_ref: VersionRef {
                version: summary.current_draft_version,
                status: summary.status,
            },
            published_version_ref: summary.latest_published_version.map(|version| VersionRef {
                version,
                status: OperationStatus::Published,
            }),
            agent_refs: refs,
        })
    }

    #[instrument(skip(self))]
    pub async fn get_operation_version(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        version: u32,
    ) -> Result<OperationVersionRecord, ApiError> {
        self.registry
            .get_operation_version(workspace_id, operation_id, version)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!(
                    "operation version {version} for {} was not found",
                    operation_id.as_str()
                ))
            })
    }

    #[instrument(skip(self, payload), fields(protocol = ?payload.protocol, operation_name = %payload.name))]
    pub async fn create_operation(
        &self,
        workspace_id: &WorkspaceId,
        payload: OperationPayload,
    ) -> Result<CreatedOperationResponse, ApiError> {
        self.validate_operation_payload(&payload)?;
        self.ensure_workspace_exists(workspace_id).await?;

        if self
            .find_operation_by_name(workspace_id, &payload.name)
            .await?
            .is_some()
        {
            return Err(ApiError::conflict(format!(
                "operation with name {} already exists",
                payload.name
            )));
        }

        let now = now_string()?;
        let operation_id = OperationId::new(new_prefixed_id("op"));
        let snapshot = RegistryOperation {
            id: operation_id.clone(),
            name: payload.name,
            display_name: payload.display_name,
            category: payload.category,
            protocol: payload.protocol,
            status: OperationStatus::Draft,
            version: 1,
            target: payload.target,
            input_schema: payload.input_schema,
            output_schema: payload.output_schema,
            input_mapping: payload.input_mapping,
            output_mapping: payload.output_mapping,
            execution_config: payload.execution_config,
            tool_description: payload.tool_description,
            samples: Some(Samples::default()),
            generated_draft: None,
            config_export: Some(ConfigExport {
                format_version: "1".to_owned(),
                export_mode: ExportMode::Portable,
            }),
            created_at: now.clone(),
            updated_at: now,
            published_at: None,
        };

        self.registry
            .create_operation(workspace_id, &snapshot, None)
            .await?;
        info!(operation_id = %operation_id.as_str(), version = 1, "operation created");

        Ok(CreatedOperationResponse {
            operation_id: operation_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            version: 1,
            status: OperationStatus::Draft,
            updated_at: snapshot.updated_at,
        })
    }

    #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), protocol = ?payload.operation.protocol))]
    pub async fn create_version(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        payload: NewVersionPayload,
    ) -> Result<CreatedOperationResponse, ApiError> {
        self.validate_operation_payload(&payload.operation)?;

        let summary = self.get_operation(workspace_id, operation_id).await?;
        let now = now_string()?;
        let version = summary.current_draft_version + 1;
        let snapshot = RegistryOperation {
            id: operation_id.clone(),
            name: payload.operation.name,
            display_name: payload.operation.display_name,
            category: payload.operation.category,
            protocol: payload.operation.protocol,
            status: OperationStatus::Draft,
            version,
            target: payload.operation.target,
            input_schema: payload.operation.input_schema,
            output_schema: payload.operation.output_schema,
            input_mapping: payload.operation.input_mapping,
            output_mapping: payload.operation.output_mapping,
            execution_config: payload.operation.execution_config,
            tool_description: payload.operation.tool_description,
            samples: Some(Samples::default()),
            generated_draft: None,
            config_export: Some(ConfigExport {
                format_version: "1".to_owned(),
                export_mode: ExportMode::Portable,
            }),
            created_at: summary.created_at,
            updated_at: now,
            published_at: None,
        };

        self.registry
            .create_version(CreateVersionRequest {
                workspace_id,
                snapshot: &snapshot,
                change_note: payload.change_note.as_deref(),
                created_by: None,
            })
            .await?;
        info!(operation_id = %operation_id.as_str(), version, "operation version created");

        Ok(CreatedOperationResponse {
            operation_id: operation_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            version,
            status: OperationStatus::Draft,
            updated_at: snapshot.updated_at,
        })
    }

    #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str()))]
    pub async fn update_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        payload: UpdateOperationPayload,
    ) -> Result<OperationMutationResult, ApiError> {
        let existing = self
            .get_operation_version(
                workspace_id,
                operation_id,
                self.get_operation(workspace_id, operation_id)
                    .await?
                    .current_draft_version,
            )
            .await?;

        let updated_at = now_string()?;
        let snapshot = RegistryOperation {
            id: operation_id.clone(),
            name: existing.snapshot.name,
            display_name: payload.display_name,
            category: payload.category,
            protocol: existing.snapshot.protocol,
            status: OperationStatus::Draft,
            version: existing.version,
            target: payload.target,
            input_schema: payload.input_schema,
            output_schema: payload.output_schema,
            input_mapping: payload.input_mapping,
            output_mapping: payload.output_mapping,
            execution_config: payload.execution_config,
            tool_description: payload.tool_description,
            samples: existing.snapshot.samples,
            generated_draft: existing.snapshot.generated_draft,
            config_export: existing.snapshot.config_export,
            created_at: existing.snapshot.created_at,
            updated_at: updated_at.clone(),
            published_at: existing.snapshot.published_at,
        };

        self.validate_registry_operation(&snapshot)?;
        self.registry
            .update_operation_draft(workspace_id, &snapshot)
            .await?;

        Ok(OperationMutationResult {
            operation_id: operation_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            version: snapshot.version,
            status: snapshot.status,
            updated_at,
        })
    }

    #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))]
    pub async fn publish_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        version: u32,
    ) -> Result<PublishResponse, ApiError> {
        let published_at = now_string()?;
        self.registry
            .publish_operation(PublishRequest {
                workspace_id,
                operation_id,
                version,
                published_at: &published_at,
                published_by: None,
            })
            .await?;
        info!(operation_id = %operation_id.as_str(), version, "operation published");

        Ok(PublishResponse {
            operation_id: operation_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            published_version: version,
            published_at,
        })
    }

    #[instrument(skip(self), fields(operation_id = %operation_id.as_str()))]
    pub async fn archive_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<OperationMutationResult, ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        let updated_at = now_string()?;
        self.registry
            .archive_operation(workspace_id, operation_id, &updated_at)
            .await?;

        Ok(OperationMutationResult {
            operation_id: operation_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            version: summary.current_draft_version,
            status: OperationStatus::Archived,
            updated_at,
        })
    }

    #[instrument(skip(self), fields(operation_id = %operation_id.as_str()))]
    pub async fn delete_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<OperationMutationResult, ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        let updated_at = now_string()?;
        self.registry
            .delete_operation(workspace_id, operation_id)
            .await?;

        Ok(OperationMutationResult {
            operation_id: operation_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            version: summary.current_draft_version,
            status: summary.status,
            updated_at,
        })
    }

    #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), version = payload.version))]
    pub async fn run_test(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        payload: TestRunPayload,
    ) -> Result<TestRunResult, ApiError> {
        let record = self
            .get_operation_version(workspace_id, operation_id, payload.version)
            .await?;
        let runtime = RuntimeOperation::from(record.snapshot.clone());
        let request_preview =
            match build_request_preview(&record.snapshot.input_mapping, &payload.input) {
                Ok(preview) => preview,
                Err(error) => {
                    self.record_invocation(InvocationRecordRequest {
                        workspace_id,
                        agent_id: None,
                        operation: &record.snapshot,
                        source: InvocationSource::AdminTestRun,
                        level: InvocationLevel::Error,
                        status: InvocationStatus::Error,
                        message: "mapping preview failed".to_owned(),
                        status_code: None,
                        error_kind: Some("mapping".to_owned()),
                        duration_ms: 0,
                        request_preview: Value::Null,
                        response_preview: Value::Null,
                    })
                    .await?;
                    return Ok(TestRunResult {
                        ok: false,
                        request_preview: Value::Null,
                        response_preview: Value::Null,
                        errors: vec![crate::error::runtime_test_failure(&RuntimeError::Mapping(
                            error,
                        ))],
                    });
                }
            };

        let started_at = std::time::Instant::now();
        match self.runtime.execute(&runtime, &payload.input).await {
            Ok(output) => {
                let duration_ms =
                    u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
                self.record_invocation(InvocationRecordRequest {
                    workspace_id,
                    agent_id: None,
                    operation: &record.snapshot,
                    source: InvocationSource::AdminTestRun,
                    level: InvocationLevel::Info,
                    status: InvocationStatus::Ok,
                    message: "admin test run completed".to_owned(),
                    status_code: None,
                    error_kind: None,
                    duration_ms,
                    request_preview: request_preview.clone(),
                    response_preview: output.clone(),
                })
                .await?;
                Ok(TestRunResult {
                    ok: true,
                    request_preview,
                    response_preview: output,
                    errors: Vec::new(),
                })
            }
            Err(error) => {
                let duration_ms =
                    u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
                self.record_invocation(InvocationRecordRequest {
                    workspace_id,
                    agent_id: None,
                    operation: &record.snapshot,
                    source: InvocationSource::AdminTestRun,
                    level: InvocationLevel::Error,
                    status: InvocationStatus::Error,
                    message: error.to_string(),
                    status_code: None,
                    error_kind: Some(runtime_error_code(&error).to_owned()),
                    duration_ms,
                    request_preview: request_preview.clone(),
                    response_preview: Value::Null,
                })
                .await?;
                Ok(TestRunResult {
                    ok: false,
                    request_preview,
                    response_preview: Value::Null,
                    errors: vec![crate::error::runtime_test_failure(&error)],
                })
            }
        }
    }

    #[instrument(skip(self))]
    pub async fn list_auth_profiles(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<AuthProfile>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        Ok(self.registry.list_auth_profiles(workspace_id).await?)
    }

    #[instrument(skip(self))]
    pub async fn list_secrets(&self, workspace_id: &WorkspaceId) -> Result<Vec<Secret>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        Ok(self
            .registry
            .list_secrets(workspace_id)
            .await?
            .into_iter()
            .map(|record| record.secret)
            .collect())
    }

    #[instrument(skip(self))]
    pub async fn get_secret(
        &self,
        workspace_id: &WorkspaceId,
        secret_id: &SecretId,
    ) -> Result<Secret, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        self.registry
            .get_secret(workspace_id, secret_id)
            .await?
            .map(|record| record.secret)
            .ok_or_else(|| {
                ApiError::not_found(format!("secret {} was not found", secret_id.as_str()))
            })
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), secret_name = %payload.name))]
    pub async fn create_secret(
        &self,
        workspace_id: &WorkspaceId,
        created_by: Option<&UserId>,
        payload: SecretPayload,
    ) -> Result<Secret, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        validate_secret_payload(&payload)?;

        let now = now_string()?;
        let secret = Secret {
            id: SecretId::new(new_prefixed_id("secret")),
            workspace_id: workspace_id.clone(),
            name: payload.name.trim().to_owned(),
            kind: payload.kind,
            status: SecretStatus::Active,
            current_version: 1,
            created_at: now.clone(),
            updated_at: now,
            last_used_at: None,
        };
        let ciphertext = self.secret_crypto.encrypt(&payload.value)?;

        self.registry
            .create_secret(CreateSecretRequest {
                secret: &secret,
                ciphertext: &ciphertext,
                key_version: self.secret_crypto.key_version(),
                created_by,
            })
            .await?;
        info!(secret_id = %secret.id.as_str(), "secret created");

        Ok(secret)
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), secret_id = %secret_id.as_str()))]
    pub async fn rotate_secret(
        &self,
        workspace_id: &WorkspaceId,
        secret_id: &SecretId,
        created_by: Option<&UserId>,
        payload: RotateSecretPayload,
    ) -> Result<Secret, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        if payload.value.is_null() {
            return Err(ApiError::validation("secret value must not be null"));
        }

        let now = now_string()?;
        let ciphertext = self.secret_crypto.encrypt(&payload.value)?;
        self.registry
            .rotate_secret(RotateSecretRequest {
                workspace_id,
                secret_id,
                ciphertext: &ciphertext,
                key_version: self.secret_crypto.key_version(),
                created_at: &now,
                updated_at: &now,
                created_by,
            })
            .await?;
        info!(secret_id = %secret_id.as_str(), "secret rotated");

        self.get_secret(workspace_id, secret_id).await
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), secret_id = %secret_id.as_str()))]
    pub async fn delete_secret(
        &self,
        workspace_id: &WorkspaceId,
        secret_id: &SecretId,
    ) -> Result<(), ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        self.registry.delete_secret(workspace_id, secret_id).await?;
        info!(secret_id = %secret_id.as_str(), "secret deleted");
        Ok(())
    }

    #[instrument(skip(self))]
    pub async fn list_agents(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<AgentSummaryView>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let workspace = self.get_workspace(workspace_id).await?;
        let summaries = self.registry.list_agents(workspace_id).await?;
        let usage = self
            .registry
            .list_usage_by_agent(UsageQuery {
                workspace_id,
                period: UsagePeriod::Last24Hours,
                source: None,
                created_after: &today_start_utc()?,
                bucket: crank_registry::UsageBucket::Hour,
            })
            .await?;
        let calls_today = usage
            .into_iter()
            .map(|item| (item.agent_id.as_str().to_owned(), item.calls_total))
            .collect::<BTreeMap<_, _>>();
        let key_count = self
            .registry
            .list_platform_api_keys(workspace_id)
            .await?
            .len();

        let mut items = Vec::with_capacity(summaries.len());
        for summary in summaries {
            let version = self
                .get_agent_version(workspace_id, &summary.id, summary.current_draft_version)
                .await?;
            let operation_ids = version
                .bindings
                .iter()
                .map(|binding| binding.operation_id.as_str().to_owned())
                .collect::<Vec<_>>();
            items.push(AgentSummaryView {
                operation_count: operation_ids.len(),
                operation_ids,
                key_count,
                calls_today: calls_today.get(summary.id.as_str()).copied().unwrap_or(0),
                mcp_endpoint: agent_mcp_endpoint(
                    workspace.workspace.slug.as_str(),
                    summary.slug.as_str(),
                ),
                ..map_agent_summary_view(summary)
            });
        }

        Ok(items)
    }

    #[instrument(skip(self))]
    pub async fn get_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
    ) -> Result<AgentSummaryView, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let workspace = self.get_workspace(workspace_id).await?;
        let summary = self
            .registry
            .get_agent_summary(workspace_id, agent_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!("agent {} was not found", agent_id.as_str()))
            })?;
        let version = self
            .get_agent_version(workspace_id, agent_id, summary.current_draft_version)
            .await?;
        let operation_ids = version
            .bindings
            .iter()
            .map(|binding| binding.operation_id.as_str().to_owned())
            .collect::<Vec<_>>();
        let usage = self
            .registry
            .get_usage_for_agent(
                UsageQuery {
                    workspace_id,
                    period: UsagePeriod::Last24Hours,
                    source: None,
                    created_after: &today_start_utc()?,
                    bucket: crank_registry::UsageBucket::Hour,
                },
                agent_id,
            )
            .await?;
        let key_count = self
            .registry
            .list_platform_api_keys(workspace_id)
            .await?
            .len();

        Ok(AgentSummaryView {
            operation_count: operation_ids.len(),
            operation_ids,
            key_count,
            calls_today: usage.map(|item| item.rollup.calls_total).unwrap_or(0),
            mcp_endpoint: agent_mcp_endpoint(
                workspace.workspace.slug.as_str(),
                summary.slug.as_str(),
            ),
            ..map_agent_summary_view(summary)
        })
    }

    #[instrument(skip(self))]
    pub async fn get_agent_version(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        version: u32,
    ) -> Result<AgentVersionRecord, ApiError> {
        self.registry
            .get_agent_version(workspace_id, agent_id, version)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!(
                    "agent version {version} for {} was not found",
                    agent_id.as_str()
                ))
            })
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_slug = %payload.slug))]
    pub async fn create_agent(
        &self,
        workspace_id: &WorkspaceId,
        payload: AgentPayload,
    ) -> Result<CreatedAgentResponse, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;

        if self
            .find_agent_by_slug(workspace_id, &payload.slug)
            .await?
            .is_some()
        {
            return Err(ApiError::conflict(format!(
                "agent with slug {} already exists",
                payload.slug
            )));
        }

        let now = now_string()?;
        let agent_id = AgentId::new(new_prefixed_id("agent"));
        let agent = Agent {
            id: agent_id.clone(),
            workspace_id: workspace_id.clone(),
            slug: payload.slug,
            display_name: payload.display_name,
            description: payload.description,
            status: AgentStatus::Draft,
            current_draft_version: 1,
            latest_published_version: None,
            created_at: now.clone(),
            updated_at: now.clone(),
            published_at: None,
        };
        let version = AgentVersion {
            agent_id: agent_id.clone(),
            version: 1,
            status: AgentStatus::Draft,
            instructions: payload.instructions,
            tool_selection_policy: payload.tool_selection_policy,
            created_at: now,
        };

        self.registry
            .create_agent(CreateAgentRequest {
                agent: &agent,
                version: &version,
                bindings: &[],
            })
            .await?;
        info!(agent_id = %agent_id.as_str(), version = 1, "agent created");

        Ok(CreatedAgentResponse {
            agent_id: agent_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            version: 1,
            status: AgentStatus::Draft,
        })
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
    pub async fn update_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        payload: UpdateAgentPayload,
    ) -> Result<AgentMutationResult, ApiError> {
        let existing = self
            .registry
            .get_agent_summary(workspace_id, agent_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!("agent {} was not found", agent_id.as_str()))
            })?;

        if payload.slug != existing.slug
            && self
                .find_agent_by_slug(workspace_id, &payload.slug)
                .await?
                .is_some()
        {
            return Err(ApiError::conflict(format!(
                "agent with slug {} already exists",
                payload.slug
            )));
        }

        let updated_at = now_string()?;
        self.registry
            .update_agent_summary(
                workspace_id,
                agent_id,
                &payload.slug,
                &payload.display_name,
                &payload.description,
                &updated_at,
            )
            .await?;

        Ok(AgentMutationResult {
            agent_id: agent_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            updated_at,
        })
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
    pub async fn delete_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
    ) -> Result<AgentMutationResult, ApiError> {
        let existing = self
            .registry
            .get_agent_summary(workspace_id, agent_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!("agent {} was not found", agent_id.as_str()))
            })?;

        self.registry.delete_agent(workspace_id, agent_id).await?;

        Ok(AgentMutationResult {
            agent_id: agent_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            updated_at: existing.updated_at,
        })
    }

    #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
    pub async fn save_agent_bindings(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        payload: Vec<AgentBindingPayload>,
    ) -> Result<AgentVersionRecord, ApiError> {
        let agent = self.get_agent(workspace_id, agent_id).await?;
        let bindings = payload
            .into_iter()
            .map(|binding| AgentOperationBinding {
                operation_id: OperationId::new(binding.operation_id),
                operation_version: binding.operation_version,
                tool_name: binding.tool_name,
                tool_title: binding.tool_title,
                tool_description_override: binding.tool_description_override,
                enabled: binding.enabled,
            })
            .collect::<Vec<_>>();

        self.registry
            .save_agent_bindings(SaveAgentBindingsRequest {
                workspace_id,
                agent_id,
                agent_version: agent.current_draft_version,
                bindings: &bindings,
            })
            .await?;
        info!(
            agent_id = %agent_id.as_str(),
            version = agent.current_draft_version,
            binding_count = bindings.len(),
            "agent bindings saved"
        );

        self.get_agent_version(workspace_id, agent_id, agent.current_draft_version)
            .await
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), version))]
    pub async fn publish_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        version: u32,
    ) -> Result<PublishAgentResponse, ApiError> {
        let published_at = now_string()?;
        self.registry
            .publish_agent(PublishAgentRequest {
                workspace_id,
                agent_id,
                version,
                published_at: &published_at,
                published_by: None,
            })
            .await?;
        info!(agent_id = %agent_id.as_str(), version, "agent published");

        Ok(PublishAgentResponse {
            agent_id: agent_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            published_version: version,
            published_at,
        })
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
    pub async fn unpublish_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
    ) -> Result<AgentMutationResult, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let updated_at = now_string()?;
        self.registry
            .unpublish_agent(workspace_id, agent_id, &updated_at)
            .await?;
        info!(agent_id = %agent_id.as_str(), "agent moved to draft");

        Ok(AgentMutationResult {
            agent_id: agent_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            updated_at,
        })
    }

    #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str()))]
    pub async fn archive_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
    ) -> Result<AgentMutationResult, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let updated_at = now_string()?;
        self.registry
            .archive_agent(workspace_id, agent_id, &updated_at)
            .await?;
        info!(agent_id = %agent_id.as_str(), "agent archived");

        Ok(AgentMutationResult {
            agent_id: agent_id.as_str().to_owned(),
            workspace_id: workspace_id.as_str().to_owned(),
            updated_at,
        })
    }

    #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("descriptor-set.bin")))]
    pub async fn upload_descriptor_set(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        source_name: Option<&str>,
        payload: &[u8],
    ) -> Result<DescriptorUploadResponse, ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        if summary.protocol != Protocol::Grpc {
            return Err(ApiError::validation(
                "descriptor upload is only allowed for grpc operations",
            ));
        }

        let services = services_from_descriptor_set_bytes(payload)
            .map_err(|error| ApiError::validation(error.to_string()))?;
        let descriptor_id = crank_core::DescriptorId::new(new_prefixed_id("desc"));
        let storage_ref = self
            .storage
            .write_descriptor(
                operation_id,
                summary.current_draft_version,
                crank_registry::DescriptorKind::DescriptorSet,
                &descriptor_id,
                source_name,
                payload,
            )
            .await?;

        self.registry
            .save_descriptor_metadata(SaveDescriptorMetadataRequest {
                descriptor: &crank_registry::DescriptorMetadata {
                    id: descriptor_id.clone(),
                    operation_id: Some(operation_id.clone()),
                    version: Some(summary.current_draft_version),
                    descriptor_kind: crank_registry::DescriptorKind::DescriptorSet,
                    storage_ref,
                    source_name: source_name.map(ToOwned::to_owned),
                    package_index: Some(
                        serde_json::to_value(&services)
                            .map_err(|error| ApiError::internal(error.to_string()))?,
                    ),
                    created_at: now_string()?,
                },
            })
            .await?;
        info!(
            operation_id = %operation_id.as_str(),
            descriptor_id = %descriptor_id.as_str(),
            version = summary.current_draft_version,
            "descriptor set uploaded"
        );

        Ok(DescriptorUploadResponse {
            descriptor_id: descriptor_id.as_str().to_owned(),
            version: summary.current_draft_version,
        })
    }

    #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("schema.proto")))]
    pub async fn upload_proto_file(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        source_name: Option<&str>,
        payload: &[u8],
    ) -> Result<DescriptorUploadResponse, ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        if summary.protocol != Protocol::Grpc {
            return Err(ApiError::validation(
                "proto upload is only allowed for grpc operations",
            ));
        }

        let descriptor_id = crank_core::DescriptorId::new(new_prefixed_id("desc"));
        let storage_ref = self
            .storage
            .write_descriptor(
                operation_id,
                summary.current_draft_version,
                crank_registry::DescriptorKind::ProtoUpload,
                &descriptor_id,
                source_name,
                payload,
            )
            .await?;

        self.registry
            .save_descriptor_metadata(SaveDescriptorMetadataRequest {
                descriptor: &crank_registry::DescriptorMetadata {
                    id: descriptor_id.clone(),
                    operation_id: Some(operation_id.clone()),
                    version: Some(summary.current_draft_version),
                    descriptor_kind: crank_registry::DescriptorKind::ProtoUpload,
                    storage_ref,
                    source_name: source_name.map(ToOwned::to_owned),
                    package_index: None,
                    created_at: now_string()?,
                },
            })
            .await?;
        info!(
            operation_id = %operation_id.as_str(),
            descriptor_id = %descriptor_id.as_str(),
            version = summary.current_draft_version,
            "proto file uploaded"
        );

        Ok(DescriptorUploadResponse {
            descriptor_id: descriptor_id.as_str().to_owned(),
            version: summary.current_draft_version,
        })
    }

    #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))]
    pub async fn list_grpc_services(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        version: Option<u32>,
    ) -> Result<Vec<GrpcServiceSummary>, ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        if summary.protocol != Protocol::Grpc {
            return Err(ApiError::validation(
                "grpc services are only available for grpc operations",
            ));
        }

        let version = version.unwrap_or(summary.current_draft_version);
        let descriptor = self
            .registry
            .list_descriptor_metadata(operation_id, version)
            .await?
            .into_iter()
            .rev()
            .find(|descriptor| {
                descriptor.descriptor_kind == crank_registry::DescriptorKind::DescriptorSet
            })
            .ok_or_else(|| ApiError::not_found("descriptor set was not uploaded"))?;
        let bytes = self.storage.read_bytes(&descriptor.storage_ref).await?;
        let services = services_from_descriptor_set_bytes(&bytes)
            .map_err(|error| ApiError::validation(error.to_string()))?;

        services
            .into_iter()
            .map(proto_service_summary)
            .collect::<Result<Vec<_>, _>>()
    }

    #[instrument(skip(self))]
    pub async fn get_auth_profile(
        &self,
        workspace_id: &WorkspaceId,
        auth_profile_id: &AuthProfileId,
    ) -> Result<AuthProfile, ApiError> {
        self.registry
            .get_auth_profile(workspace_id, auth_profile_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found(format!(
                    "auth profile {} was not found",
                    auth_profile_id.as_str()
                ))
            })
    }

    #[instrument(skip(self, payload), fields(auth_profile_name = %payload.name, auth_kind = ?payload.kind))]
    pub async fn create_auth_profile(
        &self,
        workspace_id: &WorkspaceId,
        payload: AuthProfilePayload,
    ) -> Result<AuthProfile, ApiError> {
        validate_auth_profile_kind(payload.kind, &payload.config)?;
        self.ensure_workspace_exists(workspace_id).await?;

        let now = now_string()?;
        let profile = AuthProfile {
            id: AuthProfileId::new(new_prefixed_id("auth")),
            workspace_id: workspace_id.clone(),
            name: payload.name,
            kind: payload.kind,
            config: payload.config,
            created_at: now.clone(),
            updated_at: now,
        };

        self.registry
            .save_auth_profile(SaveAuthProfileRequest {
                workspace_id,
                profile: &profile,
            })
            .await?;
        info!(auth_profile_id = %profile.id.as_str(), "auth profile created");

        Ok(profile)
    }

    #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))]
    pub async fn export_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        query: ExportQuery,
    ) -> Result<String, ApiError> {
        let version = match query.version {
            Some(version) => version,
            None => {
                self.get_operation(workspace_id, operation_id)
                    .await?
                    .current_draft_version
            }
        };
        let record = self
            .get_operation_version(workspace_id, operation_id, version)
            .await?;
        let document = YamlOperationDocument {
            format_version: "1".to_owned(),
            kind: "operation".to_owned(),
            operation: RegistryOperation {
                config_export: Some(ConfigExport {
                    format_version: "1".to_owned(),
                    export_mode: query.mode,
                }),
                ..record.snapshot
            },
        };

        serde_yaml::to_string(&document).map_err(|error| ApiError::internal(error.to_string()))
    }

    #[instrument(skip(self, yaml_document), fields(mode = ?query.mode))]
    pub async fn import_operation(
        &self,
        workspace_id: &WorkspaceId,
        query: ImportQuery,
        yaml_document: &str,
    ) -> Result<ImportResponse, ApiError> {
        let document: YamlOperationDocument = serde_yaml::from_str(yaml_document)
            .map_err(|error| ApiError::validation(error.to_string()))?;
        if document.kind != "operation" {
            return Err(ApiError::validation("yaml kind must be operation"));
        }

        let payload = OperationPayload {
            name: document.operation.name.clone(),
            display_name: document.operation.display_name.clone(),
            category: document.operation.category.clone(),
            protocol: document.operation.protocol,
            target: document.operation.target.clone(),
            input_schema: document.operation.input_schema.clone(),
            output_schema: document.operation.output_schema.clone(),
            input_mapping: document.operation.input_mapping.clone(),
            output_mapping: document.operation.output_mapping.clone(),
            execution_config: document.operation.execution_config.clone(),
            tool_description: document.operation.tool_description.clone(),
        };

        match query.mode {
            ImportMode::Create => {
                let created = self.create_operation(workspace_id, payload).await?;
                Ok(ImportResponse {
                    operation_id: created.operation_id,
                    workspace_id: created.workspace_id,
                    version: created.version,
                    import_mode: ImportMode::Create,
                    warnings: Vec::new(),
                })
            }
            ImportMode::Upsert => {
                if let Some(existing) = self
                    .find_operation_by_name(workspace_id, &document.operation.name)
                    .await?
                {
                    let created = self
                        .create_version(
                            workspace_id,
                            &existing.id,
                            NewVersionPayload {
                                operation: payload,
                                change_note: Some("yaml upsert".to_owned()),
                            },
                        )
                        .await?;

                    let response = ImportResponse {
                        operation_id: created.operation_id,
                        workspace_id: created.workspace_id,
                        version: created.version,
                        import_mode: ImportMode::Upsert,
                        warnings: Vec::new(),
                    };
                    info!(
                        operation_id = %response.operation_id,
                        version = response.version,
                        "operation imported by upsert"
                    );
                    Ok(response)
                } else {
                    let created = self.create_operation(workspace_id, payload).await?;
                    let response = ImportResponse {
                        operation_id: created.operation_id,
                        workspace_id: created.workspace_id,
                        version: created.version,
                        import_mode: ImportMode::Upsert,
                        warnings: Vec::new(),
                    };
                    info!(
                        operation_id = %response.operation_id,
                        version = response.version,
                        "operation imported by upsert"
                    );
                    Ok(response)
                }
            }
        }
    }

    #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))]
    pub async fn save_json_sample(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        sample_kind: SampleKind,
        payload: &Value,
    ) -> Result<OperationSampleMetadata, ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        let version = summary.current_draft_version;
        let sample_id = SampleId::new(new_prefixed_id("sample"));
        let now = now_string()?;
        let file_name = match sample_kind {
            SampleKind::InputJson => "input.json",
            SampleKind::OutputJson => "output.json",
            SampleKind::YamlImportSource => "source.yaml",
        };
        let storage_ref = self
            .storage
            .write_json_sample(operation_id, version, sample_kind, &sample_id, payload)
            .await?;
        let metadata = OperationSampleMetadata {
            id: sample_id,
            operation_id: operation_id.clone(),
            version,
            sample_kind,
            storage_ref,
            content_type: "application/json".to_owned(),
            file_name: Some(file_name.to_owned()),
            created_at: now,
        };

        self.registry
            .save_sample_metadata(SaveSampleMetadataRequest { sample: &metadata })
            .await?;
        info!(
            operation_id = %operation_id.as_str(),
            sample_id = %metadata.id.as_str(),
            version,
            "json sample saved"
        );

        Ok(metadata)
    }

    #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str()))]
    pub async fn generate_draft(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        payload: GenerateDraftPayload,
    ) -> Result<DraftGenerationResult, ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        let samples = self
            .registry
            .list_sample_metadata(operation_id, summary.current_draft_version)
            .await?;

        let input_sample = latest_sample_ref(&samples, SampleKind::InputJson)
            .ok_or_else(|| ApiError::validation("input_json sample was not found"))?;
        let output_sample = latest_sample_ref(&samples, SampleKind::OutputJson)
            .ok_or_else(|| ApiError::validation("output_json sample was not found"))?;
        let input_value = self.storage.read_json(&input_sample.storage_ref).await?;
        let output_value = self.storage.read_json(&output_sample.storage_ref).await?;

        let input_schema = Schema::from_json_sample(&input_value);
        let output_schema = Schema::from_json_sample(&output_value);
        let input_mapping = infer_mapping_from_samples(
            &input_value,
            JsonPathRoot::Mcp,
            &input_value,
            JsonPathRoot::RequestBody,
        );
        let output_mapping = infer_mapping_from_samples(
            &output_value,
            JsonPathRoot::ResponseBody,
            &output_value,
            JsonPathRoot::Output,
        );
        let source_types = if payload.sources.is_empty() {
            vec![
                "input_json_sample".to_owned(),
                "output_json_sample".to_owned(),
            ]
        } else {
            payload.sources
        };
        let generated_draft = GeneratedDraft {
            status: GeneratedDraftStatus::Available,
            source_types,
            generated_at: Some(now_string()?),
            input_schema_generated: true,
            output_schema_generated: true,
            input_mapping_generated: true,
            output_mapping_generated: true,
            warnings: Vec::new(),
        };

        let result = DraftGenerationResult {
            generated_draft,
            input_schema,
            output_schema,
            input_mapping,
            output_mapping,
        };
        info!(operation_id = %operation_id.as_str(), "draft generated from samples");

        Ok(result)
    }

    fn validate_operation_payload(&self, payload: &OperationPayload) -> Result<(), ApiError> {
        validate_protocol_target(payload.protocol, &payload.target)?;
        payload.input_mapping.validate_paths()?;
        payload.output_mapping.validate_paths()?;
        Ok(())
    }

    fn validate_registry_operation(&self, operation: &RegistryOperation) -> Result<(), ApiError> {
        validate_protocol_target(operation.protocol, &operation.target)?;
        operation.input_mapping.validate_paths()?;
        operation.output_mapping.validate_paths()?;
        Ok(())
    }

    async fn find_operation_by_name(
        &self,
        workspace_id: &WorkspaceId,
        name: &str,
    ) -> Result<Option<OperationSummary>, ApiError> {
        Ok(self
            .registry
            .list_operations(workspace_id)
            .await?
            .into_iter()
            .find(|operation| operation.name == name))
    }

    async fn find_agent_by_slug(
        &self,
        workspace_id: &WorkspaceId,
        slug: &str,
    ) -> Result<Option<AgentSummary>, ApiError> {
        Ok(self
            .registry
            .list_agents(workspace_id)
            .await?
            .into_iter()
            .find(|agent| agent.slug == slug))
    }

    async fn ensure_workspace_exists(&self, workspace_id: &WorkspaceId) -> Result<(), ApiError> {
        self.get_workspace(workspace_id).await.map(|_| ())
    }

    async fn seed_default_workspace_demo(
        &self,
        owner_user_id: &crank_core::UserId,
        workspace_id: &WorkspaceId,
    ) -> Result<(), ApiError> {
        let ops_admin_id = self
            .ensure_demo_user("ops-manager@crank.demo", "Ops Manager")
            .await?;
        let analyst_id = self
            .ensure_demo_user("analyst@crank.demo", "Revenue Analyst")
            .await?;
        let contractor_id = self
            .ensure_demo_user("contractor@crank.demo", "Delivery Contractor")
            .await?;

        self.registry
            .ensure_membership(workspace_id, owner_user_id, MembershipRole::Owner)
            .await?;
        self.registry
            .ensure_membership(workspace_id, &ops_admin_id, MembershipRole::Admin)
            .await?;
        self.registry
            .ensure_membership(workspace_id, &analyst_id, MembershipRole::Operator)
            .await?;
        self.registry
            .ensure_membership(workspace_id, &contractor_id, MembershipRole::Viewer)
            .await?;

        self.ensure_demo_invitation(
            workspace_id,
            InvitationPayload {
                email: "partner@crank.demo".to_owned(),
                role: MembershipRole::Viewer,
                expires_at: None,
            },
        )
        .await?;
        self.ensure_demo_invitation(
            workspace_id,
            InvitationPayload {
                email: "automation@crank.demo".to_owned(),
                role: MembershipRole::Operator,
                expires_at: None,
            },
        )
        .await?;

        let rest_operation = self
            .ensure_demo_operation(workspace_id, demo_rest_operation_payload())
            .await?;
        self.ensure_operation_published(workspace_id, &rest_operation)
            .await?;
        self.ensure_demo_json_samples(
            workspace_id,
            &rest_operation.id,
            &demo_rest_input_sample(),
            &demo_rest_output_sample(),
        )
        .await?;

        let graphql_operation = self
            .ensure_demo_operation(workspace_id, demo_graphql_operation_payload())
            .await?;
        self.ensure_operation_published(workspace_id, &graphql_operation)
            .await?;
        self.ensure_demo_json_samples(
            workspace_id,
            &graphql_operation.id,
            &demo_graphql_input_sample(),
            &demo_graphql_output_sample(),
        )
        .await?;

        let grpc_operation = self
            .ensure_demo_operation(workspace_id, demo_grpc_operation_payload())
            .await?;
        self.ensure_demo_json_samples(
            workspace_id,
            &grpc_operation.id,
            &demo_grpc_input_sample(),
            &demo_grpc_output_sample(),
        )
        .await?;
        self.ensure_demo_grpc_artifacts(workspace_id, &grpc_operation.id)
            .await?;

        let archived_operation = self
            .ensure_demo_operation(workspace_id, demo_archived_operation_payload())
            .await?;
        self.ensure_operation_archived(workspace_id, &archived_operation)
            .await?;

        let revops_agent = self
            .ensure_demo_agent(workspace_id, demo_revops_agent_payload())
            .await?;
        self.ensure_demo_agent_bindings(
            workspace_id,
            &AgentId::new(revops_agent.id.clone()),
            vec![
                AgentBindingPayload {
                    operation_id: rest_operation.id.as_str().to_owned(),
                    operation_version: rest_operation.current_draft_version,
                    tool_name: "create_crm_lead".to_owned(),
                    tool_title: "Create CRM Lead".to_owned(),
                    tool_description_override: Some(
                        "Create a new CRM lead in the revenue workspace.".to_owned(),
                    ),
                    enabled: true,
                },
                AgentBindingPayload {
                    operation_id: graphql_operation.id.as_str().to_owned(),
                    operation_version: graphql_operation.current_draft_version,
                    tool_name: "get_invoice_status".to_owned(),
                    tool_title: "Get Invoice Status".to_owned(),
                    tool_description_override: Some(
                        "Fetch invoice state from billing GraphQL.".to_owned(),
                    ),
                    enabled: true,
                },
            ],
            true,
        )
        .await?;

        let support_agent = self
            .ensure_demo_agent(workspace_id, demo_support_agent_payload())
            .await?;
        self.ensure_demo_agent_bindings(
            workspace_id,
            &AgentId::new(support_agent.id.clone()),
            vec![AgentBindingPayload {
                operation_id: grpc_operation.id.as_str().to_owned(),
                operation_version: grpc_operation.current_draft_version,
                tool_name: "lookup_support_ticket".to_owned(),
                tool_title: "Lookup Support Ticket".to_owned(),
                tool_description_override: Some(
                    "Draft support lookup flow backed by unary gRPC.".to_owned(),
                ),
                enabled: true,
            }],
            false,
        )
        .await?;

        self.ensure_demo_platform_api_key(
            workspace_id,
            "Web Console Demo Key",
            vec![PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
            false,
        )
        .await?;
        self.ensure_demo_platform_api_key(
            workspace_id,
            "Readonly Export Key",
            vec![PlatformApiKeyScope::Read],
            true,
        )
        .await?;

        self.seed_demo_invocation_logs(
            workspace_id,
            &AgentId::new(revops_agent.id),
            &rest_operation.id,
            &graphql_operation.id,
            &grpc_operation.id,
        )
        .await?;

        Ok(())
    }

    async fn seed_growth_workspace_demo(
        &self,
        owner_user_id: &crank_core::UserId,
    ) -> Result<(), ApiError> {
        let workspace = self
            .ensure_demo_workspace(
                owner_user_id,
                WorkspacePayload {
                    slug: "growth-lab".to_owned(),
                    display_name: "Growth Lab".to_owned(),
                    settings: json!({
                        "tier": "demo",
                        "region": "eu-central",
                        "notes": "Secondary workspace for workspace switch testing"
                    }),
                },
            )
            .await?;
        let workspace_id = workspace.workspace.id;

        let growth_pm_id = self
            .ensure_demo_user("growth.pm@crank.demo", "Growth PM")
            .await?;
        self.registry
            .ensure_membership(&workspace_id, owner_user_id, MembershipRole::Owner)
            .await?;
        self.registry
            .ensure_membership(&workspace_id, &growth_pm_id, MembershipRole::Admin)
            .await?;

        self.ensure_demo_invitation(
            &workspace_id,
            InvitationPayload {
                email: "agency@crank.demo".to_owned(),
                role: MembershipRole::Viewer,
                expires_at: None,
            },
        )
        .await?;
        self.ensure_demo_platform_api_key(
            &workspace_id,
            "Growth Readonly Key",
            vec![PlatformApiKeyScope::Read],
            false,
        )
        .await?;

        Ok(())
    }

    async fn ensure_demo_workspace(
        &self,
        owner_user_id: &crank_core::UserId,
        payload: WorkspacePayload,
    ) -> Result<WorkspaceRecord, ApiError> {
        if let Some(existing) = self
            .registry
            .list_workspaces_for_user(owner_user_id)
            .await?
            .into_iter()
            .find(|record| record.workspace.slug == payload.slug)
        {
            return Ok(WorkspaceRecord {
                workspace: existing.workspace,
            });
        }

        self.create_workspace(owner_user_id, payload).await
    }

    async fn ensure_demo_user(
        &self,
        email: &str,
        display_name: &str,
    ) -> Result<crank_core::UserId, ApiError> {
        let password_hash = hash_password(DEMO_USER_PASSWORD, &self.auth_settings.password_pepper)?;
        self.registry
            .upsert_bootstrap_user(email, display_name, &password_hash)
            .await
            .map_err(ApiError::from)
    }

    async fn ensure_demo_invitation(
        &self,
        workspace_id: &WorkspaceId,
        payload: InvitationPayload,
    ) -> Result<(), ApiError> {
        if self
            .list_invitations(workspace_id)
            .await?
            .iter()
            .any(|record| record.invitation.email == payload.email)
        {
            return Ok(());
        }

        self.create_invitation(workspace_id, payload).await?;
        Ok(())
    }

    async fn ensure_demo_platform_api_key(
        &self,
        workspace_id: &WorkspaceId,
        name: &str,
        scopes: Vec<PlatformApiKeyScope>,
        revoke: bool,
    ) -> Result<(), ApiError> {
        let existing = self
            .list_platform_api_keys(workspace_id)
            .await?
            .into_iter()
            .find(|record| record.api_key.name == name);
        let key = match existing {
            Some(record) => record,
            None => {
                self.create_platform_api_key(
                    workspace_id,
                    PlatformApiKeyPayload {
                        name: name.to_owned(),
                        scopes,
                    },
                )
                .await?
                .api_key
            }
        };

        if revoke && key.api_key.status != PlatformApiKeyStatus::Revoked {
            self.revoke_platform_api_key(workspace_id, &key.api_key.id)
                .await?;
        }

        Ok(())
    }

    async fn ensure_demo_operation(
        &self,
        workspace_id: &WorkspaceId,
        payload: OperationPayload,
    ) -> Result<OperationSummary, ApiError> {
        if let Some(existing) = self
            .find_operation_by_name(workspace_id, &payload.name)
            .await?
        {
            return Ok(existing);
        }

        let operation_name = payload.name.clone();
        self.create_operation(workspace_id, payload).await?;
        self.find_operation_by_name(workspace_id, &operation_name)
            .await?
            .ok_or_else(|| ApiError::internal("demo operation was created but not found"))
    }

    async fn ensure_operation_published(
        &self,
        workspace_id: &WorkspaceId,
        summary: &OperationSummary,
    ) -> Result<(), ApiError> {
        if summary.latest_published_version.is_some() {
            return Ok(());
        }

        self.publish_operation(workspace_id, &summary.id, summary.current_draft_version)
            .await?;
        Ok(())
    }

    async fn ensure_operation_archived(
        &self,
        workspace_id: &WorkspaceId,
        summary: &OperationSummary,
    ) -> Result<(), ApiError> {
        if summary.status == OperationStatus::Archived {
            return Ok(());
        }

        self.archive_operation(workspace_id, &summary.id).await?;
        Ok(())
    }

    async fn ensure_demo_json_samples(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        input: &Value,
        output: &Value,
    ) -> Result<(), ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        let samples = self
            .registry
            .list_sample_metadata(operation_id, summary.current_draft_version)
            .await?;

        if !samples
            .iter()
            .any(|sample| sample.sample_kind == SampleKind::InputJson)
        {
            self.save_json_sample(workspace_id, operation_id, SampleKind::InputJson, input)
                .await?;
        }

        if !samples
            .iter()
            .any(|sample| sample.sample_kind == SampleKind::OutputJson)
        {
            self.save_json_sample(workspace_id, operation_id, SampleKind::OutputJson, output)
                .await?;
        }

        Ok(())
    }

    async fn ensure_demo_grpc_artifacts(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<(), ApiError> {
        let summary = self.get_operation(workspace_id, operation_id).await?;
        let descriptors = self
            .registry
            .list_descriptor_metadata(operation_id, summary.current_draft_version)
            .await?;
        let has_proto = descriptors
            .iter()
            .any(|item| item.descriptor_kind == crank_registry::DescriptorKind::ProtoUpload);
        let has_descriptor = descriptors
            .iter()
            .any(|item| item.descriptor_kind == crank_registry::DescriptorKind::DescriptorSet);

        if !has_proto {
            self.upload_proto_file(
                workspace_id,
                operation_id,
                Some("echo.proto"),
                DEMO_GRPC_PROTO.as_bytes(),
            )
            .await?;
        }

        if !has_descriptor {
            let descriptor_set = STANDARD
                .decode(grpc_test_support::descriptor_set_b64())
                .map_err(|error| ApiError::internal(error.to_string()))?;
            self.upload_descriptor_set(
                workspace_id,
                operation_id,
                Some("echo_descriptor.bin"),
                &descriptor_set,
            )
            .await?;
        }

        Ok(())
    }

    async fn ensure_demo_agent(
        &self,
        workspace_id: &WorkspaceId,
        payload: AgentPayload,
    ) -> Result<AgentSummaryView, ApiError> {
        let summary =
            if let Some(existing) = self.find_agent_by_slug(workspace_id, &payload.slug).await? {
                existing
            } else {
                self.create_agent(workspace_id, payload.clone()).await?;
                self.find_agent_by_slug(workspace_id, &payload.slug)
                    .await?
                    .ok_or_else(|| ApiError::internal("demo agent was created but not found"))?
            };

        self.get_agent(workspace_id, &summary.id).await
    }

    async fn ensure_demo_agent_bindings(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        bindings: Vec<AgentBindingPayload>,
        publish: bool,
    ) -> Result<(), ApiError> {
        let summary = self.get_agent(workspace_id, agent_id).await?;
        self.save_agent_bindings(workspace_id, agent_id, bindings)
            .await?;
        if publish && summary.latest_published_version.is_none() {
            self.publish_agent(workspace_id, agent_id, summary.current_draft_version)
                .await?;
        }

        Ok(())
    }

    async fn seed_demo_invocation_logs(
        &self,
        workspace_id: &WorkspaceId,
        revops_agent_id: &AgentId,
        rest_operation_id: &OperationId,
        graphql_operation_id: &OperationId,
        grpc_operation_id: &OperationId,
    ) -> Result<(), ApiError> {
        if !self
            .registry
            .list_invocation_logs(ListInvocationLogsQuery {
                workspace_id,
                level: None,
                search_text: None,
                source: None,
                operation_id: None,
                agent_id: None,
                created_after: None,
                limit: 1,
            })
            .await?
            .is_empty()
        {
            return Ok(());
        }

        let rest_operation = self
            .get_operation_version(
                workspace_id,
                rest_operation_id,
                self.get_operation(workspace_id, rest_operation_id)
                    .await?
                    .current_draft_version,
            )
            .await?;
        let graphql_operation = self
            .get_operation_version(
                workspace_id,
                graphql_operation_id,
                self.get_operation(workspace_id, graphql_operation_id)
                    .await?
                    .current_draft_version,
            )
            .await?;
        let grpc_operation = self
            .get_operation_version(
                workspace_id,
                grpc_operation_id,
                self.get_operation(workspace_id, grpc_operation_id)
                    .await?
                    .current_draft_version,
            )
            .await?;

        self.record_invocation(InvocationRecordRequest {
            workspace_id,
            agent_id: Some(revops_agent_id),
            operation: &rest_operation.snapshot,
            source: InvocationSource::AgentToolCall,
            level: InvocationLevel::Info,
            status: InvocationStatus::Ok,
            message: "lead created in CRM".to_owned(),
            status_code: Some(201),
            error_kind: None,
            duration_ms: 182,
            request_preview: json!({
                "path": {},
                "query": {},
                "headers": { "x-demo-source": "crank-seed" },
                "variables": null,
                "grpc": null,
                "body": demo_rest_request_sample()
            }),
            response_preview: demo_rest_response_sample(),
        })
        .await?;
        self.record_invocation(InvocationRecordRequest {
            workspace_id,
            agent_id: Some(revops_agent_id),
            operation: &graphql_operation.snapshot,
            source: InvocationSource::AgentToolCall,
            level: InvocationLevel::Info,
            status: InvocationStatus::Ok,
            message: "invoice status resolved".to_owned(),
            status_code: Some(200),
            error_kind: None,
            duration_ms: 126,
            request_preview: json!({
                "path": {},
                "query": {},
                "headers": { "x-demo-source": "crank-seed" },
                "variables": demo_graphql_variables_sample(),
                "grpc": null,
                "body": null
            }),
            response_preview: demo_graphql_response_sample(),
        })
        .await?;
        self.record_invocation(InvocationRecordRequest {
            workspace_id,
            agent_id: Some(revops_agent_id),
            operation: &graphql_operation.snapshot,
            source: InvocationSource::AgentToolCall,
            level: InvocationLevel::Error,
            status: InvocationStatus::Error,
            message: "billing service returned operation error".to_owned(),
            status_code: Some(502),
            error_kind: Some("graphql_error".to_owned()),
            duration_ms: 412,
            request_preview: json!({
                "path": {},
                "query": {},
                "headers": { "x-demo-source": "crank-seed" },
                "variables": { "invoiceId": "inv_4099" },
                "grpc": null,
                "body": null
            }),
            response_preview: json!({
                "errors": [{ "message": "invoice not available" }]
            }),
        })
        .await?;
        self.record_invocation(InvocationRecordRequest {
            workspace_id,
            agent_id: None,
            operation: &grpc_operation.snapshot,
            source: InvocationSource::AdminTestRun,
            level: InvocationLevel::Info,
            status: InvocationStatus::Ok,
            message: "descriptor-backed unary echo tested".to_owned(),
            status_code: Some(200),
            error_kind: None,
            duration_ms: 98,
            request_preview: json!({
                "path": {},
                "query": {},
                "headers": {},
                "variables": null,
                "grpc": demo_grpc_request_sample(),
                "body": null
            }),
            response_preview: demo_grpc_response_sample(),
        })
        .await?;

        Ok(())
    }

    async fn record_invocation(
        &self,
        request: InvocationRecordRequest<'_>,
    ) -> Result<(), ApiError> {
        let log = InvocationLog {
            id: InvocationLogId::new(new_prefixed_id("log")),
            workspace_id: request.workspace_id.clone(),
            agent_id: request.agent_id.cloned(),
            operation_id: request.operation.id.clone(),
            source: request.source,
            level: request.level,
            status: request.status,
            tool_name: request.operation.name.clone(),
            message: request.message,
            request_id: None,
            status_code: request.status_code,
            duration_ms: request.duration_ms,
            error_kind: request.error_kind,
            request_preview: request.request_preview,
            response_preview: request.response_preview,
            created_at: now_string()?,
        };

        self.registry
            .create_invocation_log(CreateInvocationLogRequest { log: &log })
            .await?;

        Ok(())
    }
}

const DEMO_USER_PASSWORD: &str = "CrankDemoPass123!";
const DEMO_GRPC_PROTO: &str = r#"syntax = "proto3";
package echo;

service EchoService {
  rpc UnaryEcho (EchoRequest) returns (EchoResponse);
}

message EchoRequest {
  string message = 1;
}

message EchoResponse {
  string message = 1;
}
"#;

fn build_request_preview(
    mapping: &MappingSet,
    input: &Value,
) -> Result<Value, crank_mapping::MappingError> {
    let mapped = mapping.apply(&json!({ "mcp": input }))?;
    let prepared = PreparedRequest::from_mapping_output(&mapped).map_err(|error| {
        crank_mapping::MappingError::InvalidJsonPath {
            path: error.to_string(),
        }
    })?;

    Ok(json!({
        "path": prepared.path_params,
        "query": prepared.query_params,
        "headers": prepared.headers,
        "grpc": prepared.grpc.unwrap_or(Value::Null),
        "variables": prepared.variables.unwrap_or(Value::Null),
        "body": prepared.body.unwrap_or(Value::Null)
    }))
}

fn validate_protocol_target(protocol: Protocol, target: &Target) -> Result<(), ApiError> {
    let is_match = matches!(
        (protocol, target),
        (Protocol::Rest, Target::Rest(_))
            | (Protocol::Graphql, Target::Graphql(_))
            | (Protocol::Grpc, Target::Grpc(_))
    );

    if is_match {
        return Ok(());
    }

    Err(ApiError::validation("protocol and target kind must match"))
}

fn validate_auth_profile_kind(kind: AuthKind, config: &AuthConfig) -> Result<(), ApiError> {
    let is_match = matches!(
        (kind, config),
        (AuthKind::Bearer, AuthConfig::Bearer(_))
            | (AuthKind::Basic, AuthConfig::Basic(_))
            | (AuthKind::ApiKeyHeader, AuthConfig::ApiKeyHeader(_))
            | (AuthKind::ApiKeyQuery, AuthConfig::ApiKeyQuery(_))
    );

    if is_match {
        return Ok(());
    }

    Err(ApiError::validation("auth kind and config must match"))
}

fn validate_secret_payload(payload: &SecretPayload) -> Result<(), ApiError> {
    if payload.name.trim().is_empty() {
        return Err(ApiError::validation("secret name must not be empty"));
    }

    if payload.value.is_null() {
        return Err(ApiError::validation("secret value must not be null"));
    }

    Ok(())
}

fn latest_sample_ref(
    samples: &[OperationSampleMetadata],
    sample_kind: SampleKind,
) -> Option<OperationSampleMetadata> {
    samples
        .iter()
        .rev()
        .find(|sample| sample.sample_kind == sample_kind)
        .cloned()
}

fn default_export_mode() -> ExportMode {
    ExportMode::Portable
}

fn demo_revops_agent_payload() -> AgentPayload {
    AgentPayload {
        slug: "revops-copilot".to_owned(),
        display_name: "RevOps Copilot".to_owned(),
        description: "Revenue operations assistant with CRM and billing tools.".to_owned(),
        instructions: json!({
            "system": "Prefer CRM mutations first, then billing lookups for confirmation."
        }),
        tool_selection_policy: json!({
            "max_tools": 8,
            "prefer_tag": ["sales", "finance"]
        }),
    }
}

fn demo_support_agent_payload() -> AgentPayload {
    AgentPayload {
        slug: "support-triage".to_owned(),
        display_name: "Support Triage".to_owned(),
        description: "Draft support agent focused on unary gRPC workflows.".to_owned(),
        instructions: json!({
            "system": "Use support tools carefully and ask for confirmation before mutating state."
        }),
        tool_selection_policy: json!({
            "max_tools": 4,
            "prefer_tag": ["support"]
        }),
    }
}

fn demo_rest_operation_payload() -> OperationPayload {
    let input = demo_rest_input_sample();
    let request = demo_rest_request_sample();
    let response = demo_rest_response_sample();
    let output = demo_rest_output_sample();

    OperationPayload {
        name: "crm_create_lead".to_owned(),
        display_name: "Create CRM Lead".to_owned(),
        category: "sales".to_owned(),
        protocol: Protocol::Rest,
        target: Target::Rest(crank_core::RestTarget {
            base_url: "https://crm.demo.internal".to_owned(),
            method: crank_core::HttpMethod::Post,
            path_template: "/v1/leads".to_owned(),
            static_headers: BTreeMap::from([("x-demo-source".to_owned(), "crank-seed".to_owned())]),
        }),
        input_schema: Schema::from_json_sample(&input),
        output_schema: Schema::from_json_sample(&output),
        input_mapping: infer_mapping_from_samples(
            &input,
            JsonPathRoot::Mcp,
            &request,
            JsonPathRoot::RequestBody,
        ),
        output_mapping: infer_mapping_from_samples(
            &response,
            JsonPathRoot::ResponseBody,
            &output,
            JsonPathRoot::Output,
        ),
        execution_config: crank_core::ExecutionConfig {
            timeout_ms: 10_000,
            retry_policy: None,
            auth_profile_ref: None,
            headers: BTreeMap::new(),
            protocol_options: None,
        },
        tool_description: crank_core::ToolDescription {
            title: "Create CRM Lead".to_owned(),
            description: "Create a lead record in the CRM system.".to_owned(),
            tags: vec!["sales".to_owned(), "crm".to_owned()],
            examples: vec![crank_core::ToolExample {
                input: json!({
                    "email": "sarah.connor@example.com",
                    "company": "Cyberdyne"
                }),
            }],
        },
    }
}

fn demo_graphql_operation_payload() -> OperationPayload {
    let input = demo_graphql_input_sample();
    let variables = demo_graphql_variables_sample();
    let data = demo_graphql_data_sample();
    let output = demo_graphql_output_sample();

    OperationPayload {
        name: "billing_get_invoice".to_owned(),
        display_name: "Get Invoice Status".to_owned(),
        category: "finance".to_owned(),
        protocol: Protocol::Graphql,
        target: Target::Graphql(crank_core::GraphqlTarget {
            endpoint: "https://billing.demo.internal/graphql".to_owned(),
            operation_type: crank_core::GraphqlOperationType::Query,
            operation_name: "InvoiceById".to_owned(),
            query_template: "query InvoiceById($invoiceId: ID!) { invoice(id: $invoiceId) { id status total currency customerName } }".to_owned(),
            response_path: "$.response.body.data.invoice".to_owned(),
        }),
        input_schema: Schema::from_json_sample(&input),
        output_schema: Schema::from_json_sample(&output),
        input_mapping: infer_mapping_from_samples(
            &input,
            JsonPathRoot::Mcp,
            &variables,
            JsonPathRoot::RequestVariables,
        ),
        output_mapping: infer_mapping_from_samples(
            &data,
            JsonPathRoot::ResponseData,
            &output,
            JsonPathRoot::Output,
        ),
        execution_config: crank_core::ExecutionConfig {
            timeout_ms: 8_000,
            retry_policy: None,
            auth_profile_ref: None,
            headers: BTreeMap::new(),
            protocol_options: None,
        },
        tool_description: crank_core::ToolDescription {
            title: "Get Invoice Status".to_owned(),
            description: "Fetch invoice state and amount from billing.".to_owned(),
            tags: vec!["finance".to_owned(), "billing".to_owned()],
            examples: vec![crank_core::ToolExample {
                input: json!({ "invoiceId": "inv_1001" }),
            }],
        },
    }
}

fn demo_grpc_operation_payload() -> OperationPayload {
    let input = demo_grpc_input_sample();
    let request = demo_grpc_request_sample();
    let response = demo_grpc_response_sample();
    let output = demo_grpc_output_sample();

    OperationPayload {
        name: "support_lookup_ticket".to_owned(),
        display_name: "Lookup Support Ticket".to_owned(),
        category: "support".to_owned(),
        protocol: Protocol::Grpc,
        target: Target::Grpc(crank_core::GrpcTarget {
            server_addr: "http://grpc.demo.internal".to_owned(),
            package: "echo".to_owned(),
            service: "EchoService".to_owned(),
            method: "UnaryEcho".to_owned(),
            descriptor_ref: crank_core::DescriptorId::new("desc_echo_demo"),
            descriptor_set_b64: grpc_test_support::descriptor_set_b64(),
        }),
        input_schema: Schema::from_json_sample(&input),
        output_schema: Schema::from_json_sample(&output),
        input_mapping: infer_mapping_from_samples(
            &input,
            JsonPathRoot::Mcp,
            &request,
            JsonPathRoot::RequestGrpc,
        ),
        output_mapping: infer_mapping_from_samples(
            &response,
            JsonPathRoot::ResponseGrpc,
            &output,
            JsonPathRoot::Output,
        ),
        execution_config: crank_core::ExecutionConfig {
            timeout_ms: 5_000,
            retry_policy: None,
            auth_profile_ref: None,
            headers: BTreeMap::new(),
            protocol_options: Some(crank_core::ProtocolOptions {
                grpc: Some(crank_core::GrpcProtocolOptions { use_tls: false }),
            }),
        },
        tool_description: crank_core::ToolDescription {
            title: "Lookup Support Ticket".to_owned(),
            description: "Draft unary gRPC support lookup using a descriptor set.".to_owned(),
            tags: vec!["support".to_owned(), "grpc".to_owned()],
            examples: vec![crank_core::ToolExample {
                input: json!({ "message": "ticket: T-1001" }),
            }],
        },
    }
}

fn demo_archived_operation_payload() -> OperationPayload {
    let input = json!({
        "contactId": "contact_123",
        "reason": "Duplicate profile"
    });
    let request = json!({
        "contactId": "contact_123",
        "reason": "Duplicate profile"
    });
    let response = json!({
        "archived": true,
        "contactId": "contact_123"
    });

    OperationPayload {
        name: "marketing_archive_contact".to_owned(),
        display_name: "Archive Marketing Contact".to_owned(),
        category: "marketing".to_owned(),
        protocol: Protocol::Rest,
        target: Target::Rest(crank_core::RestTarget {
            base_url: "https://marketing.demo.internal".to_owned(),
            method: crank_core::HttpMethod::Patch,
            path_template: "/v1/contacts/archive".to_owned(),
            static_headers: BTreeMap::new(),
        }),
        input_schema: Schema::from_json_sample(&input),
        output_schema: Schema::from_json_sample(&response),
        input_mapping: infer_mapping_from_samples(
            &input,
            JsonPathRoot::Mcp,
            &request,
            JsonPathRoot::RequestBody,
        ),
        output_mapping: infer_mapping_from_samples(
            &response,
            JsonPathRoot::ResponseBody,
            &response,
            JsonPathRoot::Output,
        ),
        execution_config: crank_core::ExecutionConfig {
            timeout_ms: 6_000,
            retry_policy: None,
            auth_profile_ref: None,
            headers: BTreeMap::new(),
            protocol_options: None,
        },
        tool_description: crank_core::ToolDescription {
            title: "Archive Marketing Contact".to_owned(),
            description: "Legacy archived flow kept for audit only.".to_owned(),
            tags: vec!["marketing".to_owned()],
            examples: Vec::new(),
        },
    }
}

fn demo_rest_input_sample() -> Value {
    json!({
        "firstName": "Sarah",
        "lastName": "Connor",
        "email": "sarah.connor@example.com",
        "company": "Cyberdyne",
        "source": "website"
    })
}

fn demo_rest_request_sample() -> Value {
    demo_rest_input_sample()
}

fn demo_rest_response_sample() -> Value {
    json!({
        "id": "lead_1001",
        "status": "created",
        "owner": "revops"
    })
}

fn demo_rest_output_sample() -> Value {
    demo_rest_response_sample()
}

fn demo_graphql_input_sample() -> Value {
    json!({ "invoiceId": "inv_1001" })
}

fn demo_graphql_variables_sample() -> Value {
    json!({ "invoiceId": "inv_1001" })
}

fn demo_graphql_data_sample() -> Value {
    json!({
        "id": "inv_1001",
        "status": "paid",
        "total": 4200,
        "currency": "USD",
        "customerName": "Cyberdyne"
    })
}

fn demo_graphql_response_sample() -> Value {
    json!({
        "invoice": demo_graphql_data_sample()
    })
}

fn demo_graphql_output_sample() -> Value {
    demo_graphql_data_sample()
}

fn demo_grpc_input_sample() -> Value {
    json!({ "message": "ticket: T-1001" })
}

fn demo_grpc_request_sample() -> Value {
    demo_grpc_input_sample()
}

fn demo_grpc_response_sample() -> Value {
    json!({ "message": "ticket: T-1001 is currently open" })
}

fn demo_grpc_output_sample() -> Value {
    demo_grpc_response_sample()
}

fn default_enabled() -> bool {
    true
}

fn new_prefixed_id(prefix: &str) -> String {
    format!("{prefix}_{}", Uuid::now_v7().simple())
}

fn generate_access_secret(prefix: &str) -> String {
    let random = URL_SAFE_NO_PAD.encode(Uuid::now_v7().as_bytes());
    format!("{prefix}_{random}")
}

fn hash_access_secret(secret: &str) -> String {
    let digest = Sha256::digest(secret.as_bytes());
    URL_SAFE_NO_PAD.encode(digest)
}

fn now_string() -> Result<String, ApiError> {
    OffsetDateTime::now_utc()
        .format(&Rfc3339)
        .map_err(|error| ApiError::internal(error.to_string()))
}

fn default_invitation_expiry() -> Result<String, ApiError> {
    OffsetDateTime::now_utc()
        .checked_add(time::Duration::days(7))
        .ok_or_else(|| ApiError::internal("failed to compute invitation expiry"))?
        .format(&Rfc3339)
        .map_err(|error| ApiError::internal(error.to_string()))
}

fn runtime_error_code(error: &RuntimeError) -> &'static str {
    match error {
        RuntimeError::Schema(_) => "schema_error",
        RuntimeError::Mapping(_) => "mapping_error",
        RuntimeError::InvalidPreparedRequest { .. } => "invalid_request",
        RuntimeError::GraphqlAdapter(_) => "graphql_error",
        RuntimeError::GrpcAdapter(_) => "grpc_error",
        RuntimeError::RestAdapter(_) => "rest_error",
        RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol",
    }
}

fn usage_window(period: UsagePeriod) -> Result<(UsagePeriod, String, UsageBucket), ApiError> {
    let now = OffsetDateTime::now_utc();
    let (start, bucket) = match period {
        UsagePeriod::Last30Minutes => (
            now.checked_sub(time::Duration::minutes(30))
                .ok_or_else(|| ApiError::internal("failed to compute usage period"))?,
            UsageBucket::Hour,
        ),
        UsagePeriod::LastHour => (
            now.checked_sub(time::Duration::hours(1))
                .ok_or_else(|| ApiError::internal("failed to compute usage period"))?,
            UsageBucket::Hour,
        ),
        UsagePeriod::Last6Hours => (
            now.checked_sub(time::Duration::hours(6))
                .ok_or_else(|| ApiError::internal("failed to compute usage period"))?,
            UsageBucket::Hour,
        ),
        UsagePeriod::Last24Hours => (
            now.checked_sub(time::Duration::hours(24))
                .ok_or_else(|| ApiError::internal("failed to compute usage period"))?,
            UsageBucket::Hour,
        ),
        UsagePeriod::Last7Days => (
            now.checked_sub(time::Duration::days(7))
                .ok_or_else(|| ApiError::internal("failed to compute usage period"))?,
            UsageBucket::Day,
        ),
        UsagePeriod::Last30Days => (
            now.checked_sub(time::Duration::days(30))
                .ok_or_else(|| ApiError::internal("failed to compute usage period"))?,
            UsageBucket::Week,
        ),
        UsagePeriod::Last90Days => (
            now.checked_sub(time::Duration::days(90))
                .ok_or_else(|| ApiError::internal("failed to compute usage period"))?,
            UsageBucket::Month,
        ),
        UsagePeriod::ThisMonth => (
            OffsetDateTime::from_unix_timestamp(
                now.unix_timestamp() - i64::from(now.day() - 1) * 24 * 60 * 60,
            )
            .map_err(|error| ApiError::internal(error.to_string()))?
            .replace_time(time::Time::MIDNIGHT),
            UsageBucket::Week,
        ),
    };

    Ok((
        period,
        start
            .format(&Rfc3339)
            .map_err(|error| ApiError::internal(error.to_string()))?,
        bucket,
    ))
}

fn today_start_utc() -> Result<String, ApiError> {
    OffsetDateTime::now_utc()
        .replace_time(time::Time::MIDNIGHT)
        .format(&Rfc3339)
        .map_err(|error| ApiError::internal(error.to_string()))
}

fn default_usage_summary() -> OperationUsageSummaryView {
    OperationUsageSummaryView {
        calls_today: 0,
        error_rate_pct: 0.0,
        avg_latency_ms: 0,
    }
}

fn usage_map(items: Vec<OperationUsageSummary>) -> BTreeMap<String, OperationUsageSummaryView> {
    items
        .into_iter()
        .map(|item| {
            (
                item.operation_id.as_str().to_owned(),
                OperationUsageSummaryView {
                    calls_today: item.calls_today,
                    error_rate_pct: item.error_rate_pct,
                    avg_latency_ms: item.avg_latency_ms,
                },
            )
        })
        .collect()
}

fn agent_ref_map(items: Vec<OperationAgentRef>) -> BTreeMap<String, Vec<OperationAgentRefView>> {
    let mut map = BTreeMap::<String, Vec<OperationAgentRefView>>::new();
    for item in items {
        map.entry(item.operation_id.as_str().to_owned())
            .or_default()
            .push(OperationAgentRefView {
                agent_id: item.agent_id.as_str().to_owned(),
                agent_slug: item.agent_slug,
                display_name: item.display_name,
            });
    }
    map
}

fn map_agent_summary_view(summary: AgentSummary) -> AgentSummaryView {
    AgentSummaryView {
        id: summary.id.as_str().to_owned(),
        workspace_id: summary.workspace_id.as_str().to_owned(),
        slug: summary.slug,
        display_name: summary.display_name,
        description: summary.description,
        status: summary.status,
        current_draft_version: summary.current_draft_version,
        latest_published_version: summary.latest_published_version,
        created_at: summary.created_at,
        updated_at: summary.updated_at,
        published_at: summary.published_at,
        operation_count: 0,
        operation_ids: Vec::new(),
        key_count: 0,
        calls_today: 0,
        mcp_endpoint: String::new(),
    }
}

fn agent_mcp_endpoint(workspace_slug: &str, agent_slug: &str) -> String {
    format!("/mcp/v1/{workspace_slug}/{agent_slug}")
}

fn enrich_operation_summary(
    summary: OperationSummary,
    usage_summary: OperationUsageSummaryView,
    agent_refs: Vec<OperationAgentRefView>,
) -> OperationSummaryView {
    OperationSummaryView {
        id: summary.id.as_str().to_owned(),
        workspace_id: summary.workspace_id.as_str().to_owned(),
        name: summary.name,
        display_name: summary.display_name,
        category: summary.category,
        protocol: summary.protocol,
        target_url: summary.target_url,
        target_action: summary.target_action,
        status: summary.status,
        current_draft_version: summary.current_draft_version,
        latest_published_version: summary.latest_published_version,
        created_at: summary.created_at,
        updated_at: summary.updated_at,
        published_at: summary.published_at,
        usage_summary,
        agent_refs,
    }
}

fn proto_service_summary(service: ProtoService) -> Result<GrpcServiceSummary, ApiError> {
    let methods = service
        .unary_methods()
        .map(|method| {
            Ok(GrpcMethodSummary {
                name: method.name.clone(),
                kind: "unary".to_owned(),
                input_schema: crank_proto::message_to_schema(&method.input)
                    .map_err(|error| ApiError::validation(error.to_string()))?,
                output_schema: crank_proto::message_to_schema(&method.output)
                    .map_err(|error| ApiError::validation(error.to_string()))?,
            })
        })
        .collect::<Result<Vec<_>, ApiError>>()?;

    Ok(GrpcServiceSummary {
        package: service.package,
        service: service.name,
        methods,
    })
}