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

use std::path::PathBuf;

use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use crank_core::{
    Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AuthConfig, AuthKind,
    AuthProfile, AuthProfileId, ConfigExport, ExportMode, GeneratedDraft, GeneratedDraftStatus,
    InvitationId, InvitationStatus, InvitationToken, MembershipRole, OperationId, OperationStatus,
    PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, Protocol,
    SampleId, Samples, Target, 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,
    CreatePlatformApiKeyRequest, CreateVersionRequest, CreateWorkspaceRequest, InvitationRecord,
    MembershipRecord, OperationSampleMetadata, OperationSummary, OperationVersionRecord,
    PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest, PublishRequest, RegistryOperation,
    SampleKind, SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
    SaveSampleMetadataRequest, UpdateWorkspaceRequest, 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::{error::ApiError, storage::LocalArtifactStorage};

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

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct OperationPayload {
    pub name: String,
    pub display_name: 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 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 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, Deserialize)]
pub struct InvitationPayload {
    pub email: String,
    pub role: MembershipRole,
    pub expires_at: Option<String>,
}

#[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, 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,
}

#[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, 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,
}

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

    #[instrument(skip(self))]
    pub async fn list_workspaces(&self) -> Result<Vec<WorkspaceRecord>, ApiError> {
        Ok(self.registry.list_workspaces().await?)
    }

    #[instrument(skip(self, payload), fields(workspace_slug = %payload.slug))]
    pub async fn create_workspace(
        &self,
        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?;

        Ok(WorkspaceRecord { workspace })
    }

    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?)
    }

    #[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(())
    }

    #[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(())
    }

    pub async fn list_operations(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<OperationSummary>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        Ok(self.registry.list_operations(workspace_id).await?)
    }

    #[instrument(skip(self))]
    pub async fn get_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<OperationSummary, ApiError> {
        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()))
            })
    }

    #[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,
            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,
        })
    }

    #[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,
            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,
        })
    }

    #[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, 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) => {
                    return Ok(TestRunResult {
                        ok: false,
                        request_preview: Value::Null,
                        response_preview: Value::Null,
                        errors: vec![crate::error::runtime_test_failure(&RuntimeError::Mapping(
                            error,
                        ))],
                    });
                }
            };

        match self.runtime.execute(&runtime, &payload.input).await {
            Ok(output) => Ok(TestRunResult {
                ok: true,
                request_preview,
                response_preview: output,
                errors: Vec::new(),
            }),
            Err(error) => 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_agents(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<AgentSummary>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        Ok(self.registry.list_agents(workspace_id).await?)
    }

    #[instrument(skip(self))]
    pub async fn get_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
    ) -> Result<AgentSummary, ApiError> {
        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()))
            })
    }

    #[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 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, 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(),
            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(())
    }

    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(|_| ())
    }
}

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 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 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 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,
    })
}