use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::Arc; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use crank_core::{ Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AggregationMode, AsyncJobHandle, AuditSink, AuthConfig, AuthKind, AuthProfile, AuthProfileId, CapabilityProfile, CommunityCapabilityProfile, ConfigExport, EditionCapabilities, ExecutionMode, ExportMode, GeneratedDraft, GeneratedDraftStatus, IdentityError, IdentityProvider, InvitationId, InvitationStatus, InvitationToken, InvocationLevel, InvocationLog, InvocationLogId, InvocationSource, InvocationStatus, JobStatus, LoginOutcome, MachineTokenIssuer, MembershipRole, NoMachineTokenIssuer, NoopAuditSink, OperationId, OperationSecurityLevel, OperationStatus, OwnerOnlyPolicyEngine, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, PolicyEngine, ProductEdition, Protocol, ResponseCachePolicy, SampleId, Samples, Secret, SecretId, SecretKind, SecretStatus, StreamSession, StreamStatus, Target, TransportBehavior, UsagePeriod, UserId, UserSessionId, WizardState, Workspace, WorkspaceId, WorkspaceStatus, }; use crank_mapping::{JsonPathRoot, MappingSet, infer_mapping_from_samples}; use crank_registry::{ AgentSummary, AgentVersionRecord, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateAsyncJobRequest, CreateInvitationRequest, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest, CreateStreamSessionRequest, CreateVersionRequest, CreateWorkspaceRequest, InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, OperationVersionRecord, PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest, PublishRequest, RegistryError, RegistryOperation, RotateSecretRequest, SampleKind, SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, UpdateAsyncJobStatusRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, UsageQuery, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, }; use crank_runtime::{ PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, RuntimeOperation, RuntimeRequestContext, SecretCrypto, }; 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, storage::LocalArtifactStorage, }; #[derive(Clone)] pub struct AdminService { registry: PostgresRegistry, runtime: RuntimeExecutor, storage: LocalArtifactStorage, auth_settings: AuthSettings, secret_crypto: SecretCrypto, identity_provider: Option>, policy_engine: Arc, audit_sink: Arc, token_issuer: Arc, capability_profile: Arc, } pub struct AdminServiceBuilder { registry: PostgresRegistry, storage_root: PathBuf, auth_settings: AuthSettings, secret_crypto: SecretCrypto, runtime: RuntimeExecutor, identity_provider: Option>, policy_engine: Option>, audit_sink: Option>, token_issuer: Option>, capability_profile: Option>, } #[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, pub current_workspace_id: Option, } #[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, #[serde(default)] pub security_level: OperationSecurityLevel, 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, #[serde(default)] pub wizard_state: Option, } #[derive(Clone, Debug, Deserialize)] pub struct NewVersionPayload { #[serde(flatten)] pub operation: OperationPayload, #[serde(default)] pub change_note: Option, } #[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 mode: ExecutionMode, pub request_preview: Value, pub response_preview: Value, pub errors: Vec, #[serde(skip_serializing_if = "Option::is_none")] pub window: Option, #[serde(skip_serializing_if = "Option::is_none")] pub stream_session: Option, #[serde(skip_serializing_if = "Option::is_none")] pub async_job: Option, } #[derive(Clone, Debug, Serialize)] pub struct WindowTestRunView { pub window_complete: bool, pub truncated: bool, pub has_more: bool, pub cursor: Option, } #[derive(Clone, Debug, Serialize)] pub struct StreamSessionStartView { pub session_id: String, pub status: StreamStatus, #[serde(with = "time::serde::rfc3339")] pub expires_at: OffsetDateTime, pub poll_after_ms: u64, pub preview: Value, } #[derive(Clone, Debug, Serialize)] pub struct AsyncJobStartView { pub job_id: String, pub status: JobStatus, pub progress: Value, } #[derive(Debug, Serialize, Deserialize)] struct StoredSessionState { input: Value, summary: Value, items: Vec, next_index: usize, batch_size: usize, } enum TestRunOutcome { Unary { output: Value, }, Window { output: crank_runtime::WindowExecutionResult, }, Session { output: StreamSessionStartView, }, AsyncJob { output: AsyncJobStartView, }, } impl TestRunOutcome { fn into_result_views( self, ) -> ( &'static str, Value, Option, Option, Option, ) { match self { Self::Unary { output } => ("admin test run completed", output, None, None, None), Self::Window { output } => ( "admin window test run completed", json!({ "summary": output.summary, "items": output.items, "cursor": output.cursor, "window_complete": output.window_complete, "truncated": output.truncated, "has_more": output.has_more, }), Some(WindowTestRunView { window_complete: output.window_complete, truncated: output.truncated, has_more: output.has_more, cursor: output.cursor, }), None, None, ), Self::Session { output } => ( "admin session test run started", json!({ "session_id": output.session_id, "status": output.status, "expires_at": format_timestamp(output.expires_at), "poll_after_ms": output.poll_after_ms, "preview": output.preview, }), None, Some(output), None, ), Self::AsyncJob { output } => ( "admin async job test run started", json!({ "job_id": output.job_id, "status": output.status, "progress": output.progress, }), None, None, Some(output), ), } } } #[derive(Clone, Debug, Deserialize)] pub struct AuthProfilePayload { pub name: String, pub kind: AuthKind, pub config: AuthConfig, } #[derive(Clone, Debug, Deserialize)] pub struct UpstreamPayload { pub name: String, pub base_url: String, #[serde(default)] pub static_headers: Value, #[serde(default)] pub auth_profile_id: Option, } #[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, pub display_name: Option, pub status: Option, pub settings: Option, } #[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, #[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, pub created_at: String, pub updated_at: String, pub published_at: Option, pub operation_count: usize, pub operation_ids: Vec, 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, } #[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, } #[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, pub invitations: Vec, pub operations: Vec, pub agents: Vec, pub platform_api_keys: Vec, pub exported_at: String, } #[derive(Clone, Debug, Deserialize)] pub struct LogsQuery { pub level: Option, pub search: Option, pub source: Option, pub operation_id: Option, pub agent_id: Option, pub period: Option, pub limit: Option, } #[derive(Clone, Debug, Deserialize)] pub struct UsageRequestQuery { pub period: Option, pub source: Option, } #[derive(Clone, Debug, Serialize)] pub struct UsageOverviewResponse { pub summary: UsageSummary, pub timeline: Vec, pub operations: Vec, pub agents: Vec, } #[derive(Clone, Debug, Serialize)] pub struct ProtocolCapabilityView { pub protocol: Protocol, pub supports_execution_modes: Vec, pub supports_transport_behaviors: Vec, pub supports_auth_kinds: Vec, pub supports_upload_artifacts: Vec, pub supports_cursor_path: bool, pub supports_done_path: bool, pub supports_aggregation_mode: Vec, } #[derive(Clone, Debug, Deserialize)] pub struct GenerateDraftPayload { #[serde(default)] pub sources: Vec, } #[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, #[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, #[serde(default)] pub security_level: OperationSecurityLevel, 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, #[serde(default)] pub wizard_state: Option, } #[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, request_id: Option<&'a str>, source: InvocationSource, level: InvocationLevel, status: InvocationStatus, message: String, status_code: Option, error_kind: Option, 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 security_level: OperationSecurityLevel, pub target_url: String, pub target_action: String, pub status: OperationStatus, pub current_draft_version: u32, pub latest_published_version: Option, pub created_at: String, pub updated_at: String, pub published_at: Option, pub usage_summary: OperationUsageSummaryView, pub agent_refs: Vec, } #[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 security_level: OperationSecurityLevel, pub status: OperationStatus, pub current_draft_version: u32, pub latest_published_version: Option, pub created_at: String, pub updated_at: String, pub published_at: Option, pub draft_version_ref: VersionRef, pub published_version_ref: Option, pub agent_refs: Vec, } #[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, } #[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, } #[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 { #[cfg(test)] pub fn new( registry: PostgresRegistry, storage_root: PathBuf, auth_settings: AuthSettings, secret_crypto: SecretCrypto, ) -> Self { Self::new_with_runtime( registry, storage_root, auth_settings, secret_crypto, RuntimeExecutor::new(), ) } #[cfg(test)] pub fn new_with_runtime( registry: PostgresRegistry, storage_root: PathBuf, auth_settings: AuthSettings, secret_crypto: SecretCrypto, runtime: RuntimeExecutor, ) -> Self { AdminServiceBuilder::new( registry, storage_root, auth_settings, secret_crypto, runtime, ) .build() } pub fn auth_settings(&self) -> &AuthSettings { &self.auth_settings } pub fn policy_engine(&self) -> &Arc { &self.policy_engine } pub fn audit_sink(&self) -> &Arc { &self.audit_sink } pub fn token_issuer(&self) -> &Arc { &self.token_issuer } pub fn capability_profile(&self) -> &Arc { &self.capability_profile } } impl AdminServiceBuilder { pub fn new( registry: PostgresRegistry, storage_root: PathBuf, auth_settings: AuthSettings, secret_crypto: SecretCrypto, runtime: RuntimeExecutor, ) -> Self { Self { registry, storage_root, auth_settings, secret_crypto, runtime, identity_provider: None, policy_engine: None, audit_sink: None, token_issuer: None, capability_profile: None, } } pub fn with_identity_provider(mut self, provider: Arc) -> Self { self.identity_provider = Some(provider); self } #[allow(dead_code)] pub fn with_policy_engine(mut self, policy_engine: Arc) -> Self { self.policy_engine = Some(policy_engine); self } #[allow(dead_code)] pub fn with_audit_sink(mut self, audit_sink: Arc) -> Self { self.audit_sink = Some(audit_sink); self } #[allow(dead_code)] pub fn with_token_issuer(mut self, token_issuer: Arc) -> Self { self.token_issuer = Some(token_issuer); self } #[allow(dead_code)] pub fn with_capability_profile( mut self, capability_profile: Arc, ) -> Self { self.capability_profile = Some(capability_profile); self } pub fn build(self) -> AdminService { AdminService { registry: self.registry, runtime: self.runtime, storage: LocalArtifactStorage::new(self.storage_root), auth_settings: self.auth_settings, secret_crypto: self.secret_crypto, identity_provider: self.identity_provider, policy_engine: self .policy_engine .unwrap_or_else(|| Arc::new(OwnerOnlyPolicyEngine)), audit_sink: self.audit_sink.unwrap_or_else(|| Arc::new(NoopAuditSink)), token_issuer: self .token_issuer .unwrap_or_else(|| Arc::new(NoMachineTokenIssuer)), capability_profile: self .capability_profile .unwrap_or_else(|| Arc::new(CommunityCapabilityProfile)), } } } impl AdminService { 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, 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 authenticated = self.authenticate_login(&payload).await?; 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(&authenticated.user.id) .await?; let current_workspace_id = authenticated .current_workspace_id .as_ref() .map(|workspace_id| workspace_id.as_str().to_owned()) .or_else(|| { memberships .first() .map(|membership| membership.workspace.id.as_str().to_owned()) }); let current_workspace_ref = current_workspace_id .as_ref() .map(|workspace_id| WorkspaceId::new(workspace_id.clone())); self.registry .create_user_session( &session_cookie.session_id, &authenticated.user.id, current_workspace_ref.as_ref(), &secret_hash, &session_cookie.expires_at, ) .await?; Ok(( session_cookie, SessionResponse { user: authenticated.user, memberships, current_workspace_id, }, )) } async fn authenticate_login( &self, payload: &LoginPayload, ) -> Result { if let Some(identity_provider) = &self.identity_provider { return match identity_provider .login_password(crank_core::LoginPayload { email: payload.email.clone(), password: payload.password.clone(), }) .await { Ok(LoginOutcome::Authenticated(identity)) => Ok(identity), Ok(LoginOutcome::TwoFactorRequired(_)) => Err(ApiError::internal( "two-factor login flow is not configured for this service", )), Err(error) => Err(map_identity_error(error)), }; } 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")); } Ok(crank_core::AuthenticatedIdentity { user: user.user, memberships: vec![], current_workspace_id: None, }) } 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, 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 { 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 { let now = OffsetDateTime::now_utc(); 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, updated_at: now, }; self.registry .create_workspace(CreateWorkspaceRequest { workspace: &workspace, }) .await?; self.registry .ensure_membership(&workspace.id, user_id, MembershipRole::Owner) .await?; self.ensure_default_workspace_upstreams(&workspace.id) .await?; Ok(WorkspaceRecord { workspace }) } #[instrument(skip(self))] pub async fn session_response( &self, session_id: &UserSessionId, session_value: &str, ) -> Result, 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 { 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 { 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_with_context( format!("user {} was not found", user_id.as_str()), json!({ "user_id": 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_with_context( format!("user {} was not found", user_id.as_str()), json!({ "user_id": 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 { self.registry .get_workspace(workspace_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("workspace {} was not found", workspace_id.as_str()), json!({ "workspace_id": 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 { 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: OffsetDateTime::now_utc(), }; self.registry .update_workspace(UpdateWorkspaceRequest { workspace: &workspace, }) .await?; Ok(WorkspaceRecord { workspace }) } #[instrument(skip(self))] pub async fn list_memberships( &self, workspace_id: &WorkspaceId, ) -> Result, 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, 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_with_context( format!( "membership for user {} in workspace {} was not found", target_user_id.as_str(), workspace_id.as_str() ), json!({ "workspace_id": workspace_id.as_str(), "user_id": target_user_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_with_context( format!( "membership for user {} in workspace {} was not found", target_user_id.as_str(), workspace_id.as_str() ), json!({ "workspace_id": workspace_id.as_str(), "user_id": target_user_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, 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 { 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) => parse_timestamp(&expires_at)?, None => default_invitation_expiry()?, }, created_at: OffsetDateTime::now_utc(), }, }; 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 { 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.registry.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_agent_platform_api_keys( &self, workspace_id: &WorkspaceId, agent_id: &AgentId, ) -> Result, ApiError> { self.ensure_workspace_exists(workspace_id).await?; self.registry .get_agent_summary(workspace_id, agent_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("agent {} was not found", agent_id.as_str()), json!({ "agent_id": agent_id.as_str() }), ) })?; Ok(self .registry .list_platform_api_keys_for_agent(workspace_id, agent_id) .await?) } #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_name = %payload.name))] pub async fn create_agent_platform_api_key( &self, workspace_id: &WorkspaceId, agent_id: &AgentId, payload: PlatformApiKeyPayload, ) -> Result { self.ensure_workspace_exists(workspace_id).await?; self.registry .get_agent_summary(workspace_id, agent_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("agent {} was not found", agent_id.as_str()), json!({ "agent_id": agent_id.as_str() }), ) })?; 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(), agent_id: Some(agent_id.clone()), name: payload.name, prefix: secret.chars().take(16).collect(), scopes: payload.scopes, status: PlatformApiKeyStatus::Active, created_at: OffsetDateTime::now_utc(), 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(), agent_id = %agent_id.as_str(), key_id = %key_id.as_str()))] pub async fn revoke_agent_platform_api_key( &self, workspace_id: &WorkspaceId, agent_id: &AgentId, key_id: &PlatformApiKeyId, ) -> Result<(), ApiError> { self.registry .revoke_platform_api_key_for_agent( workspace_id, agent_id, key_id, &OffsetDateTime::now_utc(), ) .await?; Ok(()) } #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_id = %key_id.as_str()))] pub async fn delete_agent_platform_api_key( &self, workspace_id: &WorkspaceId, agent_id: &AgentId, key_id: &PlatformApiKeyId, ) -> Result<(), ApiError> { self.registry .delete_platform_api_key_for_agent(workspace_id, agent_id, key_id) .await?; Ok(()) } #[instrument(skip(self))] pub async fn list_logs( &self, workspace_id: &WorkspaceId, query: LogsQuery, ) -> Result, 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 { self.registry .get_invocation_log(workspace_id, log_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("invocation log {} was not found", log_id.as_str()), json!({ "log_id": log_id.as_str() }), ) }) } #[instrument(skip(self))] pub async fn get_usage_overview( &self, workspace_id: &WorkspaceId, query: UsageRequestQuery, ) -> Result { 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 { 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_with_context( format!( "usage for operation {} was not found", operation_id.as_str() ), json!({ "operation_id": operation_id.as_str() }), ) }) } #[instrument(skip(self))] pub async fn get_agent_usage( &self, workspace_id: &WorkspaceId, agent_id: &AgentId, query: UsageRequestQuery, ) -> Result { 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_with_context( format!("usage for agent {} was not found", agent_id.as_str()), json!({ "agent_id": agent_id.as_str() }), ) }) } pub async fn list_protocol_capabilities(&self) -> Vec { let capabilities = self.get_capabilities().await; capabilities .supported_protocols .into_iter() .map(|protocol| protocol_capability_view(protocol, capabilities.edition)) .collect() } pub async fn get_capabilities(&self) -> EditionCapabilities { self.capability_profile().capabilities() } pub async fn list_operations( &self, workspace_id: &WorkspaceId, ) -> Result, 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 { let summary = self .registry .get_operation_summary(workspace_id, operation_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("operation {} was not found", operation_id.as_str()), json!({ "operation_id": 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, security_level: summary.security_level, status: summary.status, current_draft_version: summary.current_draft_version, latest_published_version: summary.latest_published_version, created_at: format_timestamp(summary.created_at), updated_at: format_timestamp(summary.updated_at), published_at: summary.published_at.map(format_timestamp), 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 { self.registry .get_operation_version(workspace_id, operation_id, version) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!( "operation version {version} for {} was not found", operation_id.as_str() ), json!({ "operation_id": operation_id.as_str(), "version": version, }), ) }) } #[instrument(skip(self, payload), fields(protocol = ?payload.protocol, operation_name = %payload.name))] pub async fn create_operation( &self, workspace_id: &WorkspaceId, payload: OperationPayload, ) -> Result { 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_with_context( format!("operation with name {} already exists", payload.name), json!({ "name": payload.name }), )); } let now = OffsetDateTime::now_utc(); 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, security_level: payload.security_level, 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, }), wizard_state: payload.wizard_state, created_at: now, 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: format_timestamp(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 { self.validate_operation_payload(&payload.operation)?; let summary = self .registry .get_operation_summary(workspace_id, operation_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("operation {} was not found", operation_id.as_str()), json!({ "operation_id": operation_id.as_str() }), ) })?; let now = OffsetDateTime::now_utc(); 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, security_level: payload.operation.security_level, 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, }), wizard_state: payload.operation.wizard_state, 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: format_timestamp(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 { 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 = OffsetDateTime::now_utc(); let snapshot = RegistryOperation { id: operation_id.clone(), name: existing.snapshot.name, display_name: payload.display_name, category: payload.category, protocol: existing.snapshot.protocol, security_level: payload.security_level, 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, wizard_state: payload.wizard_state, created_at: existing.snapshot.created_at, updated_at, 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: format_timestamp(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 { let published_at = OffsetDateTime::now_utc(); 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: format_timestamp(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 { let summary = self.get_operation(workspace_id, operation_id).await?; let updated_at = OffsetDateTime::now_utc(); 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: format_timestamp(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 { 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, request_id: &str, ) -> Result { let runtime_request_context = RuntimeRequestContext::from_request_id(request_id) .with_metering_context(workspace_id.clone(), None, InvocationSource::AdminTestRun); let record = self .get_operation_version(workspace_id, operation_id, payload.version) .await?; let runtime = RuntimeOperation::from(record.snapshot.clone()); let mode = runtime .execution_config .streaming .as_ref() .map(|streaming| streaming.mode) .unwrap_or(ExecutionMode::Unary); 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, request_id: Some(request_id), 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, mode, request_preview: Value::Null, response_preview: Value::Null, errors: vec![crate::error::runtime_test_failure(&RuntimeError::Mapping( error, ))], window: None, stream_session: None, async_job: None, }); } }; let resolved_auth = self .resolve_operation_auth(workspace_id, &runtime.execution_config) .await; let started_at = std::time::Instant::now(); match match resolved_auth { Ok(resolved_auth) => match mode { ExecutionMode::Unary => self .runtime .execute_with_auth_and_context( &runtime, &payload.input, resolved_auth.as_ref(), Some(&runtime_request_context), ) .await .map(|output| TestRunOutcome::Unary { output }), ExecutionMode::Window => self .runtime .execute_window_with_auth_and_context( &runtime, &payload.input, resolved_auth.as_ref(), Some(&runtime_request_context), ) .await .map(|output| TestRunOutcome::Window { output }), ExecutionMode::Session => self .start_stream_session_test( workspace_id, &record.snapshot, &runtime, &payload.input, resolved_auth.as_ref(), &runtime_request_context, ) .await .map(|output| TestRunOutcome::Session { output }), ExecutionMode::AsyncJob => self .start_async_job_test( workspace_id, &record.snapshot, &runtime, &payload.input, &runtime_request_context, ) .await .map(|output| TestRunOutcome::AsyncJob { output }), }, Err(error) => Err(error), } { Ok(outcome) => { let duration_ms = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX); let (message, response_preview, window, stream_session, async_job) = outcome.into_result_views(); self.record_invocation(InvocationRecordRequest { workspace_id, agent_id: None, operation: &record.snapshot, request_id: Some(request_id), source: InvocationSource::AdminTestRun, level: InvocationLevel::Info, status: InvocationStatus::Ok, message: message.to_owned(), status_code: None, error_kind: None, duration_ms, request_preview: request_preview.clone(), response_preview: response_preview.clone(), }) .await?; Ok(TestRunResult { ok: true, mode, request_preview, response_preview, errors: Vec::new(), window, stream_session, async_job, }) } 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, request_id: Some(request_id), 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, mode, request_preview, response_preview: Value::Null, errors: vec![crate::error::runtime_test_failure(&error)], window: None, stream_session: None, async_job: None, }) } } } async fn resolve_operation_auth( &self, workspace_id: &WorkspaceId, execution_config: &crank_core::ExecutionConfig, ) -> Result, RuntimeError> { let Some(auth_profile_id) = execution_config.auth_profile_ref.as_ref() else { return Ok(None); }; let auth_profile = self .registry .get_auth_profile(workspace_id, auth_profile_id) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "load auth profile", details: error.to_string(), })? .ok_or_else(|| RuntimeError::MissingAuthProfile { auth_profile_id: auth_profile_id.as_str().to_owned(), })?; self.resolve_auth_profile(workspace_id, &auth_profile) .await .map(Some) } async fn start_stream_session_test( &self, workspace_id: &WorkspaceId, operation: &RegistryOperation, runtime: &RuntimeOperation, input: &Value, resolved_auth: Option<&ResolvedAuth>, request_context: &RuntimeRequestContext, ) -> Result { let streaming = runtime.execution_config.streaming.as_ref().ok_or_else(|| { RuntimeError::MissingStreamingConfig { operation_id: runtime.operation_id.as_str().to_owned(), } })?; let seed = self .runtime .execute_session_seed_with_auth_and_context( runtime, input, resolved_auth, Some(request_context), ) .await?; let batch_size = streaming.max_items.unwrap_or(10).max(1) as usize; let preview_count = seed.items.len().min(batch_size); let preview_items = seed.items[..preview_count].to_vec(); let next_index = preview_count; let created_at = OffsetDateTime::now_utc(); let expires_at = created_at .checked_add(time::Duration::milliseconds( streaming.max_session_lifetime_ms.unwrap_or(60_000) as i64, )) .ok_or_else(|| RuntimeError::SecretCrypto { operation: "compute stream session expiration", details: "failed to compute stream session expiration".to_owned(), })?; let session = StreamSession { id: crank_core::StreamSessionId::new(new_prefixed_id("sess")), workspace_id: workspace_id.clone(), agent_id: None, operation_id: operation.id.clone(), protocol: operation.protocol, mode: ExecutionMode::Session, status: StreamStatus::Running, cursor: (next_index < seed.items.len()).then(|| json!(next_index)), state: json!(StoredSessionState { input: input.clone(), summary: seed.summary.clone(), items: seed.items.clone(), next_index, batch_size, }), expires_at, last_poll_at: None, created_at, closed_at: None, }; self.registry .create_stream_session(CreateStreamSessionRequest { session: &session }) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "persist stream session", details: error.to_string(), })?; Ok(StreamSessionStartView { session_id: session.id.as_str().to_owned(), status: session.status, expires_at, poll_after_ms: streaming.poll_interval_ms.unwrap_or(1_000), preview: json!({ "summary": seed.summary, "items": preview_items, }), }) } async fn start_async_job_test( &self, workspace_id: &WorkspaceId, operation: &RegistryOperation, runtime: &RuntimeOperation, input: &Value, request_context: &RuntimeRequestContext, ) -> Result { let created_at = OffsetDateTime::now_utc(); let streaming = runtime.execution_config.streaming.as_ref().ok_or_else(|| { RuntimeError::MissingStreamingConfig { operation_id: runtime.operation_id.as_str().to_owned(), } })?; let expires_at = created_at .checked_add(time::Duration::milliseconds( streaming.max_session_lifetime_ms.unwrap_or(300_000) as i64, )) .ok_or_else(|| RuntimeError::SecretCrypto { operation: "compute async job expiration", details: "failed to compute async job expiration".to_owned(), })?; let job = AsyncJobHandle { id: crank_core::AsyncJobId::new(new_prefixed_id("job")), workspace_id: workspace_id.clone(), agent_id: None, operation_id: operation.id.clone(), status: JobStatus::Running, progress: json!({ "pct": 0 }), result: None, error: None, expires_at: Some(expires_at), last_poll_at: None, created_at, updated_at: created_at, finished_at: None, }; self.registry .create_async_job(CreateAsyncJobRequest { job: &job }) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "persist async job", details: error.to_string(), })?; let registry = self.registry.clone(); let secret_crypto = self.secret_crypto.clone(); let runtime_for_task = self.runtime.clone(); let workspace_for_task = workspace_id.clone(); let operation_for_task = runtime.clone(); let input_for_task = input.clone(); let request_context_for_task = request_context.clone(); let job_id = job.id.clone(); tokio::spawn(async move { let resolved_auth = resolve_runtime_auth_for_task( ®istry, &secret_crypto, &workspace_for_task, &operation_for_task.execution_config, ) .await; let result = match resolved_auth { Ok(resolved_auth) => { runtime_for_task .execute_with_auth_and_context( &operation_for_task, &input_for_task, resolved_auth.as_ref(), Some(&request_context_for_task), ) .await } Err(error) => Err(error), }; let finished_at = OffsetDateTime::now_utc(); let _ = match result { Ok(output) => { registry .update_async_job_status(UpdateAsyncJobStatusRequest { job_id: &job_id, current_status: JobStatus::Running, next_status: JobStatus::Completed, progress: &json!({ "pct": 100 }), result: Some(&output), error: None, expires_at: None, updated_at: &finished_at, finished_at: Some(&finished_at), }) .await } Err(error) => { registry .update_async_job_status(UpdateAsyncJobStatusRequest { job_id: &job_id, current_status: JobStatus::Running, next_status: JobStatus::Failed, progress: &json!({ "pct": 100 }), result: None, error: Some(&json!({ "code": runtime_error_code(&error), "message": error.to_string(), })), expires_at: None, updated_at: &finished_at, finished_at: Some(&finished_at), }) .await } }; }); Ok(AsyncJobStartView { job_id: job.id.as_str().to_owned(), status: job.status, progress: job.progress, }) } async fn resolve_auth_profile( &self, workspace_id: &WorkspaceId, auth_profile: &AuthProfile, ) -> Result { let mut secrets = BTreeMap::new(); let used_at = OffsetDateTime::now_utc(); for secret_id in auth_profile.config.secret_ids() { let secret = self .registry .get_secret(workspace_id, secret_id) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "load secret", details: error.to_string(), })? .ok_or_else(|| RuntimeError::MissingSecret { secret_id: secret_id.as_str().to_owned(), })?; let version = self .registry .get_current_secret_version(workspace_id, secret_id) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "load current secret version", details: error.to_string(), })? .ok_or_else(|| RuntimeError::MissingSecretVersion { secret_id: secret_id.as_str().to_owned(), version: secret.secret.current_version, })?; let plaintext = self.secret_crypto.decrypt( &version.secret_version.key_version, &version.secret_version.ciphertext, )?; self.registry .touch_secret(workspace_id, secret_id, &used_at) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "touch secret", details: error.to_string(), })?; secrets.insert(secret_id.clone(), plaintext); } ResolvedAuth::from_profile(auth_profile, &secrets) } #[instrument(skip(self))] pub async fn list_auth_profiles( &self, workspace_id: &WorkspaceId, ) -> Result, ApiError> { self.ensure_workspace_exists(workspace_id).await?; Ok(self.registry.list_auth_profiles(workspace_id).await?) } #[instrument(skip(self))] pub async fn list_workspace_upstreams( &self, workspace_id: &WorkspaceId, ) -> Result, ApiError> { self.ensure_workspace_exists(workspace_id).await?; self.ensure_default_workspace_upstreams(workspace_id) .await?; Ok(self.registry.list_workspace_upstreams(workspace_id).await?) } #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), upstream_name = %payload.name))] pub async fn save_workspace_upstream( &self, workspace_id: &WorkspaceId, upstream_id: Option<&WorkspaceUpstreamId>, payload: UpstreamPayload, ) -> Result { self.ensure_workspace_exists(workspace_id).await?; validate_upstream_payload(&payload)?; if let Some(auth_profile_id) = payload.auth_profile_id.as_deref() { self.get_auth_profile( workspace_id, &AuthProfileId::new(auth_profile_id.to_owned()), ) .await?; } let now = OffsetDateTime::now_utc(); let existing = match upstream_id { Some(id) => { self.registry .get_workspace_upstream(workspace_id, id) .await? } None => None, }; if upstream_id.is_some() && existing.is_none() { return Err(ApiError::not_found_with_context( "upstream was not found", json!({ "upstream_id": upstream_id.map(|id| id.as_str()).unwrap_or_default() }), )); } let upstream = WorkspaceUpstream { id: existing .as_ref() .map(|item| item.id.clone()) .unwrap_or_else(|| WorkspaceUpstreamId::new(new_prefixed_id("upstream"))), workspace_id: workspace_id.clone(), name: payload.name.trim().to_owned(), base_url: normalize_base_url(&payload.base_url), static_headers: payload.static_headers, auth_profile_id: payload .auth_profile_id .filter(|value| !value.trim().is_empty()), created_at: existing.as_ref().map(|item| item.created_at).unwrap_or(now), updated_at: now, }; self.registry .save_workspace_upstream(SaveWorkspaceUpstreamRequest { upstream: &upstream, }) .await?; info!(upstream_id = %upstream.id.as_str(), "workspace upstream saved"); Ok(upstream) } async fn ensure_default_workspace_upstreams( &self, workspace_id: &WorkspaceId, ) -> Result<(), ApiError> { let existing = self.registry.list_workspace_upstreams(workspace_id).await?; if existing.iter().any(|item| item.name == "Open Meteo") { return Ok(()); } let now = OffsetDateTime::now_utc(); let upstream = WorkspaceUpstream { id: WorkspaceUpstreamId::new(new_prefixed_id("upstream")), workspace_id: workspace_id.clone(), name: "Open Meteo".to_owned(), base_url: "https://api.open-meteo.com".to_owned(), static_headers: json!({}), auth_profile_id: None, created_at: now, updated_at: now, }; self.registry .save_workspace_upstream(SaveWorkspaceUpstreamRequest { upstream: &upstream, }) .await?; Ok(()) } #[instrument(skip(self))] pub async fn list_secrets(&self, workspace_id: &WorkspaceId) -> Result, 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 { 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_with_context( format!("secret {} was not found", secret_id.as_str()), json!({ "secret_id": 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 { self.ensure_workspace_exists(workspace_id).await?; validate_secret_payload(&payload)?; let now = OffsetDateTime::now_utc(); 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, updated_at: now, last_used_at: None, }; let ciphertext = self .secret_crypto .encrypt(&payload.value) .map_err(|error| ApiError::internal(error.to_string()))?; 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 { self.ensure_workspace_exists(workspace_id).await?; if payload.value.is_null() { return Err(ApiError::validation("secret value must not be null")); } let now = OffsetDateTime::now_utc(); let ciphertext = self .secret_crypto .encrypt(&payload.value) .map_err(|error| ApiError::internal(error.to_string()))?; 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?; if let Some(profile) = self .registry .list_auth_profiles_referencing_secret(workspace_id, secret_id) .await? .into_iter() .next() { return Err(RegistryError::SecretReferencedByAuthProfile { secret_id: secret_id.as_str().to_owned(), auth_profile_id: profile.id.as_str().to_owned(), } .into()); } 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, 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::>(); let key_counts = self .registry .list_platform_api_keys(workspace_id) .await? .into_iter() .fold(BTreeMap::new(), |mut counts, record| { if let Some(agent_id) = record.api_key.agent_id { *counts.entry(agent_id.as_str().to_owned()).or_insert(0usize) += 1; } counts }); 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::>(); items.push(AgentSummaryView { operation_count: operation_ids.len(), operation_ids, key_count: key_counts.get(summary.id.as_str()).copied().unwrap_or(0), 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 { 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_with_context( format!("agent {} was not found", agent_id.as_str()), json!({ "agent_id": 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::>(); 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_for_agent(workspace_id, agent_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 { self.registry .get_agent_version(workspace_id, agent_id, version) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!( "agent version {version} for {} was not found", agent_id.as_str() ), json!({ "agent_id": agent_id.as_str(), "version": version, }), ) }) } #[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 { self.ensure_workspace_exists(workspace_id).await?; if self .find_agent_by_slug(workspace_id, &payload.slug) .await? .is_some() { return Err(ApiError::conflict_with_context( format!("agent with slug {} already exists", payload.slug), json!({ "slug": payload.slug }), )); } let now = OffsetDateTime::now_utc(); 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, updated_at: now, 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 { let existing = self .registry .get_agent_summary(workspace_id, agent_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("agent {} was not found", agent_id.as_str()), json!({ "agent_id": 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_with_context( format!("agent with slug {} already exists", payload.slug), json!({ "slug": payload.slug }), )); } let updated_at = OffsetDateTime::now_utc(); 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: format_timestamp(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 { let existing = self .registry .get_agent_summary(workspace_id, agent_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("agent {} was not found", agent_id.as_str()), json!({ "agent_id": 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: format_timestamp(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, ) -> Result { 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::>(); 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 { let agent_version = self .get_agent_version(workspace_id, agent_id, version) .await?; let published_bindings = self .published_agent_bindings(workspace_id, &agent_version.bindings) .await?; if published_bindings.is_empty() { return Err(ApiError::conflict_with_context( "agent cannot be published without published enabled tools", json!({ "agent_id": agent_id.as_str(), "version": version, "binding_count": agent_version.bindings.len() }), )); } let published_at = OffsetDateTime::now_utc(); if published_bindings != agent_version.bindings { let draft_version = AgentVersion { agent_id: agent_id.clone(), version: agent_version.version + 1, status: AgentStatus::Draft, instructions: agent_version.snapshot.instructions.clone(), tool_selection_policy: agent_version.snapshot.tool_selection_policy.clone(), created_at: published_at, }; self.registry .create_agent_draft_version(CreateAgentDraftVersionRequest { workspace_id, agent_id, version: &draft_version, bindings: &agent_version.bindings, updated_at: &published_at, }) .await?; self.registry .save_agent_bindings(SaveAgentBindingsRequest { workspace_id, agent_id, agent_version: agent_version.version, bindings: &published_bindings, }) .await?; } 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: format_timestamp(published_at), }) } async fn published_agent_bindings( &self, workspace_id: &WorkspaceId, bindings: &[AgentOperationBinding], ) -> Result, ApiError> { let mut published = Vec::new(); for binding in bindings { if !binding.enabled { continue; } let Some(summary) = self .registry .get_operation_summary(workspace_id, &binding.operation_id) .await? else { continue; }; let Some(operation_version) = summary.latest_published_version else { continue; }; published.push(AgentOperationBinding { operation_id: summary.id, operation_version, tool_name: binding.tool_name.clone(), tool_title: binding.tool_title.clone(), tool_description_override: binding.tool_description_override.clone(), enabled: true, }); } Ok(published) } #[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 { self.ensure_workspace_exists(workspace_id).await?; let updated_at = OffsetDateTime::now_utc(); 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: format_timestamp(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 { self.ensure_workspace_exists(workspace_id).await?; let updated_at = OffsetDateTime::now_utc(); 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: format_timestamp(updated_at), }) } #[instrument(skip(self))] pub async fn get_auth_profile( &self, workspace_id: &WorkspaceId, auth_profile_id: &AuthProfileId, ) -> Result { self.registry .get_auth_profile(workspace_id, auth_profile_id) .await? .ok_or_else(|| { ApiError::not_found_with_context( format!("auth profile {} was not found", auth_profile_id.as_str()), json!({ "auth_profile_id": 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 { validate_auth_profile_kind(payload.kind, &payload.config)?; self.ensure_workspace_exists(workspace_id).await?; self.validate_auth_profile_secret_ids(workspace_id, &payload.config) .await?; let now = OffsetDateTime::now_utc(); 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, 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) } async fn validate_auth_profile_secret_ids( &self, workspace_id: &WorkspaceId, config: &AuthConfig, ) -> Result<(), ApiError> { for secret_id in config.secret_ids() { self.get_secret(workspace_id, secret_id).await?; } Ok(()) } #[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 { 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 { 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, security_level: document.operation.security_level, 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(), wizard_state: document.operation.wizard_state.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 { 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 = OffsetDateTime::now_utc(); 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 { 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> { self.validate_operation_capabilities(payload.protocol, payload.security_level)?; validate_protocol_target(payload.protocol, &payload.target)?; validate_streaming_policy(&payload.execution_config)?; validate_response_cache_policy(&payload.target, &payload.execution_config)?; payload.input_mapping.validate_paths()?; payload.output_mapping.validate_paths()?; Ok(()) } fn validate_registry_operation(&self, operation: &RegistryOperation) -> Result<(), ApiError> { self.validate_operation_capabilities(operation.protocol, operation.security_level)?; validate_protocol_target(operation.protocol, &operation.target)?; validate_streaming_policy(&operation.execution_config)?; validate_response_cache_policy(&operation.target, &operation.execution_config)?; operation.input_mapping.validate_paths()?; operation.output_mapping.validate_paths()?; Ok(()) } fn validate_operation_capabilities( &self, protocol: Protocol, security_level: OperationSecurityLevel, ) -> Result<(), ApiError> { let supported_protocols = [Protocol::Rest]; if !supported_protocols.contains(&protocol) { return Err(ApiError::validation_with_context( format!( "protocol {} is not supported in Community", serde_json::to_string(&protocol) .unwrap_or_else(|_| "\"unknown\"".to_owned()) .trim_matches('"') ), json!({ "protocol": protocol, "edition": "community", }), )); } if security_level != OperationSecurityLevel::Standard { return Err(ApiError::validation_with_context( format!( "security level {} is not supported in Community", serde_json::to_string(&security_level) .unwrap_or_else(|_| "\"unknown\"".to_owned()) .trim_matches('"') ), json!({ "security_level": security_level, "edition": "community", }), )); } Ok(()) } async fn find_operation_by_name( &self, workspace_id: &WorkspaceId, name: &str, ) -> Result, 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, 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 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, }], true, ) .await?; self.ensure_demo_platform_api_key( workspace_id, &AgentId::new(revops_agent.id.clone()), "Web Console Demo Key", vec![PlatformApiKeyScope::Read, PlatformApiKeyScope::Write], false, ) .await?; self.seed_demo_invocation_logs( workspace_id, &AgentId::new(revops_agent.id), &rest_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?; Ok(()) } async fn ensure_demo_workspace( &self, owner_user_id: &crank_core::UserId, payload: WorkspacePayload, ) -> Result { 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 { 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, agent_id: &AgentId, name: &str, scopes: Vec, revoke: bool, ) -> Result<(), ApiError> { let existing = self .registry .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_agent_platform_api_key( workspace_id, agent_id, PlatformApiKeyPayload { name: name.to_owned(), scopes, }, ) .await? .api_key } }; if revoke && key.api_key.status != PlatformApiKeyStatus::Revoked { self.revoke_agent_platform_api_key(workspace_id, agent_id, &key.api_key.id) .await?; } Ok(()) } async fn ensure_demo_operation( &self, workspace_id: &WorkspaceId, payload: OperationPayload, ) -> Result { 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_agent( &self, workspace_id: &WorkspaceId, payload: AgentPayload, ) -> Result { 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, 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, ) -> 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?; self.record_invocation(InvocationRecordRequest { workspace_id, agent_id: Some(revops_agent_id), operation: &rest_operation.snapshot, request_id: None, 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?; 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: request.request_id.map(ToOwned::to_owned), 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: OffsetDateTime::now_utc(), }; self.registry .create_invocation_log(CreateInvocationLogRequest { log: &log }) .await?; Ok(()) } } const DEMO_USER_PASSWORD: &str = "CrankDemoPass123!"; fn build_request_preview( mapping: &MappingSet, input: &Value, ) -> Result { 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(_)) | (Protocol::Websocket, Target::Websocket(_)) | (Protocol::Soap, Target::Soap(_)) ); 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_upstream_payload(payload: &UpstreamPayload) -> Result<(), ApiError> { if payload.name.trim().is_empty() { return Err(ApiError::validation("upstream name is required")); } let base_url = normalize_base_url(&payload.base_url); if !(base_url.starts_with("https://") || base_url.starts_with("http://")) { return Err(ApiError::validation( "upstream base_url must start with http:// or https://", )); } if !payload.static_headers.is_object() { return Err(ApiError::validation( "upstream static_headers must be a JSON object", )); } if let Some(headers) = payload.static_headers.as_object() { for (key, value) in headers { if key.trim().is_empty() || !value.is_string() { return Err(ApiError::validation( "upstream static_headers must contain string values", )); } } } Ok(()) } fn normalize_base_url(value: &str) -> String { value.trim().trim_end_matches('/').to_owned() } 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 { 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_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, security_level: OperationSecurityLevel::Standard, 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, response_cache: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: 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" }), }], }, wizard_state: Some(WizardState { input_sample: Some(input), output_sample: Some(output), test_input: Some(demo_rest_input_sample()), }), } } 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, security_level: OperationSecurityLevel::Standard, 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, response_cache: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: 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(), }, wizard_state: Some(WizardState { input_sample: Some(input), output_sample: Some(response), test_input: Some(request), }), } } 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 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 { OffsetDateTime::now_utc() .format(&Rfc3339) .map_err(|error| ApiError::internal(error.to_string())) } fn parse_timestamp(value: &str) -> Result { OffsetDateTime::parse(value, &Rfc3339) .map_err(|_| ApiError::validation("timestamp must be RFC 3339")) } fn format_timestamp(timestamp: OffsetDateTime) -> String { timestamp .format(&Rfc3339) .unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_owned()) } fn map_identity_error(error: IdentityError) -> ApiError { match error { IdentityError::BadCredentials => ApiError::unauthorized("invalid email or password"), IdentityError::AccountDisabled => ApiError::forbidden("account is disabled"), IdentityError::NotSupportedForProvider => ApiError::internal( "password login is not supported by the configured identity provider", ), IdentityError::Internal(message) => ApiError::internal(message), } } async fn resolve_runtime_auth_for_task( registry: &PostgresRegistry, secret_crypto: &SecretCrypto, workspace_id: &WorkspaceId, execution_config: &crank_core::ExecutionConfig, ) -> Result, RuntimeError> { let Some(auth_profile_id) = execution_config.auth_profile_ref.as_ref() else { return Ok(None); }; let auth_profile = registry .get_auth_profile(workspace_id, auth_profile_id) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "load auth profile", details: error.to_string(), })? .ok_or_else(|| RuntimeError::MissingAuthProfile { auth_profile_id: auth_profile_id.as_str().to_owned(), })?; let mut secrets = BTreeMap::new(); let used_at = OffsetDateTime::now_utc(); for secret_id in auth_profile.config.secret_ids() { let secret = registry .get_secret(workspace_id, secret_id) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "load secret", details: error.to_string(), })? .ok_or_else(|| RuntimeError::MissingSecret { secret_id: secret_id.as_str().to_owned(), })?; let version = registry .get_current_secret_version(workspace_id, secret_id) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "load current secret version", details: error.to_string(), })? .ok_or_else(|| RuntimeError::MissingSecretVersion { secret_id: secret_id.as_str().to_owned(), version: secret.secret.current_version, })?; let plaintext = secret_crypto.decrypt( &version.secret_version.key_version, &version.secret_version.ciphertext, )?; registry .touch_secret(workspace_id, secret_id, &used_at) .await .map_err(|error| RuntimeError::SecretCrypto { operation: "touch secret", details: error.to_string(), })?; secrets.insert(secret_id.clone(), plaintext); } ResolvedAuth::from_profile(&auth_profile, &secrets).map(Some) } fn default_invitation_expiry() -> Result { OffsetDateTime::now_utc() .checked_add(time::Duration::days(7)) .ok_or_else(|| ApiError::internal("failed to compute invitation expiry")) } 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::ProtocolAdapter(_) => "adapter_error", RuntimeError::SoapAdapter(_) => "soap_error", RuntimeError::WebsocketAdapter(_) => "websocket_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", RuntimeError::ConcurrencyLimitExceeded { .. } => "runtime_overloaded", RuntimeError::MissingStreamingConfig { .. } => "streaming_config_error", RuntimeError::UnsupportedExecutionMode { .. } => "streaming_mode_error", RuntimeError::InvalidStreamingPayload { .. } => "streaming_payload_error", RuntimeError::MissingAuthProfile { .. } => "auth_profile_not_found", RuntimeError::MissingSecret { .. } | RuntimeError::MissingSecretVersion { .. } => { "secret_not_found" } RuntimeError::InvalidAuthSecretValue { .. } => "secret_value_error", RuntimeError::SecretCrypto { .. } => "secret_crypto_error", } } fn protocol_capability_view(protocol: Protocol, edition: ProductEdition) -> ProtocolCapabilityView { let community_build = matches!(edition, ProductEdition::Community); let supports_execution_modes = if community_build { vec![ExecutionMode::Unary] } else { [ ExecutionMode::Unary, ExecutionMode::Window, ExecutionMode::Session, ExecutionMode::AsyncJob, ] .into_iter() .filter(|mode| protocol.supports_execution_mode(*mode)) .collect() }; let supports_transport_behaviors = if community_build { vec![TransportBehavior::RequestResponse] } else { [ TransportBehavior::RequestResponse, TransportBehavior::ServerStream, TransportBehavior::StatefulSession, TransportBehavior::DeferredResult, ] .into_iter() .filter(|behavior| protocol.supports_transport_behavior(*behavior)) .collect() }; let supports_upload_artifacts = Vec::new(); ProtocolCapabilityView { protocol, supports_execution_modes, supports_transport_behaviors, supports_auth_kinds: vec![ "none".to_owned(), "bearer".to_owned(), "basic".to_owned(), "api_key_header".to_owned(), "api_key_query".to_owned(), ], supports_upload_artifacts, supports_cursor_path: !matches!(protocol, Protocol::Graphql), supports_done_path: !matches!(protocol, Protocol::Graphql), supports_aggregation_mode: vec![ AggregationMode::RawItems, AggregationMode::SummaryOnly, AggregationMode::SummaryPlusSamples, AggregationMode::Stats, AggregationMode::LatestState, ], } } 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 { 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) -> BTreeMap { 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) -> BTreeMap> { let mut map = BTreeMap::>::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: format_timestamp(summary.created_at), updated_at: format_timestamp(summary.updated_at), published_at: summary.published_at.map(format_timestamp), 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 validate_streaming_policy( execution_config: &crank_core::ExecutionConfig, ) -> Result<(), ApiError> { if execution_config.streaming.is_some() { return Err(ApiError::validation_with_context( "streaming execution is not supported in Community".to_owned(), json!({ "field": "execution_config.streaming", "edition": "community", }), )); } Ok(()) } fn validate_response_cache_policy( target: &Target, execution_config: &crank_core::ExecutionConfig, ) -> Result<(), ApiError> { let Some(ResponseCachePolicy { ttl_ms }) = execution_config.response_cache.as_ref() else { return Ok(()); }; if *ttl_ms == 0 { return Err(ApiError::validation_with_context( "response cache ttl must be greater than zero".to_owned(), json!({ "field": "execution_config.response_cache.ttl_ms", }), )); } if execution_config.streaming.is_some() { return Err(ApiError::validation_with_context( "response cache is supported only for unary operations".to_owned(), json!({ "field": "execution_config.response_cache", }), )); } match target { Target::Rest(rest_target) if rest_target.method == crank_core::HttpMethod::Get => {} _ => { return Err(ApiError::validation_with_context( "response cache is supported only for REST GET operations".to_owned(), json!({ "field": "execution_config.response_cache", }), )); } } if execution_config.auth_profile_ref.is_some() { return Err(ApiError::validation_with_context( "response cache is not supported for operations with auth_profile_ref".to_owned(), json!({ "field": "execution_config.auth_profile_ref", }), )); } Ok(()) } #[cfg(test)] #[allow(clippy::items_after_test_module)] mod tests { use std::collections::BTreeMap; use crank_core::{ExecutionConfig, HttpMethod, ResponseCachePolicy, RestTarget, Target}; use super::validate_response_cache_policy; fn cacheable_execution_config() -> ExecutionConfig { ExecutionConfig { timeout_ms: 1_000, retry_policy: None, response_cache: Some(ResponseCachePolicy { ttl_ms: 5_000 }), auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: None, } } #[test] fn accepts_response_cache_for_rest_get() { let target = Target::Rest(RestTarget { base_url: "http://example.invalid".to_owned(), method: HttpMethod::Get, path_template: "/catalog".to_owned(), static_headers: BTreeMap::new(), }); let result = validate_response_cache_policy(&target, &cacheable_execution_config()); assert!(result.is_ok()); } #[test] fn rejects_response_cache_for_non_rest_target() { let target = Target::Graphql(crank_core::GraphqlTarget { endpoint: "http://example.invalid/graphql".to_owned(), operation_type: crank_core::GraphqlOperationType::Query, operation_name: "LookupLead".to_owned(), query_template: "query LookupLead($email: String!) { lookupLead(email: $email) { id } }".to_owned(), response_path: "$.response.body.data.lookupLead".to_owned(), }); let error = validate_response_cache_policy(&target, &cacheable_execution_config()).unwrap_err(); assert!(matches!(error, crate::error::ApiError::Validation { .. })); assert_eq!( error.to_string(), "response cache is supported only for REST GET operations" ); } } fn enrich_operation_summary( summary: OperationSummary, usage_summary: OperationUsageSummaryView, agent_refs: Vec, ) -> 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, security_level: summary.security_level, 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: format_timestamp(summary.created_at), updated_at: format_timestamp(summary.updated_at), published_at: summary.published_at.map(format_timestamp), usage_summary, agent_refs, } }