diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 01613a1..7c8c3f8 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -5,30 +5,25 @@ use std::sync::Arc; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use crank_core::{ AgentId, AgentStatus, AuditSink, AuthConfig, AuthKind, AuthProfile, AuthProfileId, - CapabilityProfile, CommunityCapabilityProfile, ConfigExport, EditionCapabilities, - ExecutionMode, ExportMode, GeneratedDraft, GeneratedDraftStatus, IdentityError, - IdentityProvider, InvocationLevel, InvocationLog, InvocationLogId, InvocationSource, - InvocationStatus, LoginOutcome, MembershipRole, NoopAuditSink, OperationId, - OperationSecurityLevel, OperationStatus, OwnerOnlyPolicyEngine, PlatformApiKey, - PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, PolicyEngine, ProductEdition, - Protocol, ResponseCachePolicy, SampleId, Samples, SecretKind, Target, ToolQualityMappingRule, - ToolQualityMappingSet, ToolQualitySchemaKind, ToolQualitySchemaNode, UsagePeriod, - UserSessionId, WizardState, WorkspaceId, WorkspaceStatus, + CapabilityProfile, CommunityCapabilityProfile, EditionCapabilities, ExecutionMode, ExportMode, + GeneratedDraft, GeneratedDraftStatus, IdentityError, IdentityProvider, InvocationLevel, + InvocationLog, InvocationLogId, InvocationSource, InvocationStatus, LoginOutcome, + MembershipRole, NoopAuditSink, OperationId, OperationSecurityLevel, OperationStatus, + OwnerOnlyPolicyEngine, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, + PlatformApiKeyStatus, PolicyEngine, ProductEdition, Protocol, ResponseCachePolicy, SampleId, + SecretKind, Target, ToolQualityMappingRule, ToolQualityMappingSet, ToolQualitySchemaKind, + ToolQualitySchemaNode, UsagePeriod, UserSessionId, WizardState, WorkspaceId, WorkspaceStatus, }; use crank_mapping::{JsonPathRoot, MappingRule, MappingSet, infer_mapping_from_samples}; use crank_registry::{ - AgentSummary, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest, - ListInvocationLogsQuery, OperationAgentRef, OperationSampleMetadata, OperationSummary, - OperationUsageSummary, OperationVersionRecord, PlatformApiKeyRecord, PostgresRegistry, - PublishRequest, RegistryOperation, SampleKind, SaveSampleMetadataRequest, - SaveWorkspaceUpstreamRequest, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, - UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord, - WorkspaceUpstream, WorkspaceUpstreamId, -}; -use crank_runtime::{ - PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, RuntimeOperation, - RuntimeRequestContext, SecretCrypto, + AgentSummary, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, ListInvocationLogsQuery, + OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, + PlatformApiKeyRecord, PostgresRegistry, RegistryOperation, SampleKind, + SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, UsageAgentBreakdown, UsageBucket, + UsageOperationBreakdown, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, + WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, }; +use crank_runtime::{PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, SecretCrypto}; use crank_schema::{Schema, SchemaKind}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -40,6 +35,7 @@ use uuid::Uuid; mod agents; mod import_export; mod observability; +mod operations; mod secrets; mod workspaces; @@ -1001,518 +997,6 @@ impl AdminService { 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(protocol = ?payload.protocol, operation_name = %payload.name))] - pub async fn analyze_operation_quality( - &self, - workspace_id: &WorkspaceId, - payload: OperationPayload, - ) -> Result { - self.validate_operation_payload(&payload)?; - self.ensure_workspace_exists(workspace_id).await?; - - let mut findings = - crank_core::analyze_tool_identity_quality(&payload.name, &payload.tool_description) - .findings; - let input_schema = tool_quality_schema_node(&payload.input_schema); - findings.extend( - crank_core::analyze_tool_schema_quality("input_schema", &input_schema).findings, - ); - let output_mapping = tool_quality_mapping_set(&payload.output_mapping); - findings - .extend(crank_core::analyze_tool_response_projection_quality(&output_mapping).findings); - - Ok(crank_core::ToolQualityReport::new(findings)) - } - - #[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 = 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, - ))], - }); - } - }; - - 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) => { - self.runtime - .execute_with_auth_and_context( - &runtime, - &payload.input, - resolved_auth.as_ref(), - Some(&runtime_request_context), - ) - .await - } - Err(error) => Err(error), - } { - Ok(response_preview) => { - 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::Info, - status: InvocationStatus::Ok, - message: "admin test run completed".to_owned(), - status_code: None, - error_kind: None, - duration_ms, - request_preview: request_preview.clone(), - response_preview: response_preview.clone(), - }) - .await?; - Ok(TestRunResult { - ok: true, - mode, - request_preview, - response_preview, - errors: Vec::new(), - }) - } - Err(error) => { - let duration_ms = - u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX); - self.record_invocation(InvocationRecordRequest { - workspace_id, - agent_id: None, - operation: &record.snapshot, - 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)], - }) - } - } - } - async fn resolve_operation_auth( &self, workspace_id: &WorkspaceId, diff --git a/apps/admin-api/src/service/operations.rs b/apps/admin-api/src/service/operations.rs new file mode 100644 index 0000000..efc69dd --- /dev/null +++ b/apps/admin-api/src/service/operations.rs @@ -0,0 +1,537 @@ +use crank_core::{ + ConfigExport, ExecutionMode, ExportMode, InvocationLevel, InvocationSource, InvocationStatus, + OperationId, OperationStatus, Samples, WorkspaceId, +}; +use crank_registry::{ + CreateVersionRequest, OperationVersionRecord, PublishRequest, RegistryOperation, +}; +use crank_runtime::{RuntimeError, RuntimeOperation, RuntimeRequestContext}; +use serde_json::{Value, json}; +use time::OffsetDateTime; +use tracing::{info, instrument}; + +use crate::{ + error::ApiError, + service::{ + AdminService, CreatedOperationResponse, InvocationRecordRequest, NewVersionPayload, + OperationDetailView, OperationMutationResult, OperationPayload, OperationSummaryView, + PublishResponse, TestRunPayload, TestRunResult, UpdateOperationPayload, VersionRef, + agent_ref_map, build_request_preview, default_usage_summary, enrich_operation_summary, + format_timestamp, new_prefixed_id, now_string, runtime_error_code, today_start_utc, + tool_quality_mapping_set, tool_quality_schema_node, usage_map, + }, +}; + +impl AdminService { + 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(protocol = ?payload.protocol, operation_name = %payload.name))] + pub async fn analyze_operation_quality( + &self, + workspace_id: &WorkspaceId, + payload: OperationPayload, + ) -> Result { + self.validate_operation_payload(&payload)?; + self.ensure_workspace_exists(workspace_id).await?; + + let mut findings = + crank_core::analyze_tool_identity_quality(&payload.name, &payload.tool_description) + .findings; + let input_schema = tool_quality_schema_node(&payload.input_schema); + findings.extend( + crank_core::analyze_tool_schema_quality("input_schema", &input_schema).findings, + ); + let output_mapping = tool_quality_mapping_set(&payload.output_mapping); + findings + .extend(crank_core::analyze_tool_response_projection_quality(&output_mapping).findings); + + Ok(crank_core::ToolQualityReport::new(findings)) + } + + #[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 = 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, + ))], + }); + } + }; + + 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) => { + self.runtime + .execute_with_auth_and_context( + &runtime, + &payload.input, + resolved_auth.as_ref(), + Some(&runtime_request_context), + ) + .await + } + Err(error) => Err(error), + } { + Ok(response_preview) => { + 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::Info, + status: InvocationStatus::Ok, + message: "admin test run completed".to_owned(), + status_code: None, + error_kind: None, + duration_ms, + request_preview: request_preview.clone(), + response_preview: response_preview.clone(), + }) + .await?; + Ok(TestRunResult { + ok: true, + mode, + request_preview, + response_preview, + errors: Vec::new(), + }) + } + Err(error) => { + let duration_ms = + u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + self.record_invocation(InvocationRecordRequest { + workspace_id, + agent_id: None, + operation: &record.snapshot, + 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)], + }) + } + } + } +}