Split admin operations service module
This commit is contained in:
github-ops
+16
-532
@@ -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<Vec<OperationSummaryView>, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let summaries = self.registry.list_operations(workspace_id).await?;
|
||||
let usage = self
|
||||
.registry
|
||||
.list_operation_usage_summaries(workspace_id, &today_start_utc()?)
|
||||
.await?;
|
||||
let agent_refs = self
|
||||
.registry
|
||||
.list_operation_agent_refs(workspace_id)
|
||||
.await?;
|
||||
let usage_by_operation = usage_map(usage);
|
||||
let refs_by_operation = agent_ref_map(agent_refs);
|
||||
|
||||
Ok(summaries
|
||||
.into_iter()
|
||||
.map(|summary| {
|
||||
let operation_id = summary.id.as_str().to_owned();
|
||||
enrich_operation_summary(
|
||||
summary,
|
||||
usage_by_operation
|
||||
.get(&operation_id)
|
||||
.cloned()
|
||||
.unwrap_or_else(default_usage_summary),
|
||||
refs_by_operation
|
||||
.get(&operation_id)
|
||||
.cloned()
|
||||
.unwrap_or_default(),
|
||||
)
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_operation(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
operation_id: &OperationId,
|
||||
) -> Result<OperationDetailView, ApiError> {
|
||||
let summary = self
|
||||
.registry
|
||||
.get_operation_summary(workspace_id, operation_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_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<OperationVersionRecord, ApiError> {
|
||||
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<CreatedOperationResponse, ApiError> {
|
||||
self.validate_operation_payload(&payload)?;
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
|
||||
if self
|
||||
.find_operation_by_name(workspace_id, &payload.name)
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
return Err(ApiError::conflict_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<crank_core::ToolQualityReport, ApiError> {
|
||||
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<CreatedOperationResponse, ApiError> {
|
||||
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<OperationMutationResult, ApiError> {
|
||||
let existing = self
|
||||
.get_operation_version(
|
||||
workspace_id,
|
||||
operation_id,
|
||||
self.get_operation(workspace_id, operation_id)
|
||||
.await?
|
||||
.current_draft_version,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let updated_at = 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<PublishResponse, ApiError> {
|
||||
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<OperationMutationResult, ApiError> {
|
||||
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<OperationMutationResult, ApiError> {
|
||||
let summary = self.get_operation(workspace_id, operation_id).await?;
|
||||
let updated_at = now_string()?;
|
||||
self.registry
|
||||
.delete_operation(workspace_id, operation_id)
|
||||
.await?;
|
||||
|
||||
Ok(OperationMutationResult {
|
||||
operation_id: operation_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
version: summary.current_draft_version,
|
||||
status: summary.status,
|
||||
updated_at,
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), version = payload.version))]
|
||||
pub async fn run_test(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
operation_id: &OperationId,
|
||||
payload: TestRunPayload,
|
||||
request_id: &str,
|
||||
) -> Result<TestRunResult, ApiError> {
|
||||
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,
|
||||
|
||||
@@ -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<Vec<OperationSummaryView>, ApiError> {
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
let summaries = self.registry.list_operations(workspace_id).await?;
|
||||
let usage = self
|
||||
.registry
|
||||
.list_operation_usage_summaries(workspace_id, &today_start_utc()?)
|
||||
.await?;
|
||||
let agent_refs = self
|
||||
.registry
|
||||
.list_operation_agent_refs(workspace_id)
|
||||
.await?;
|
||||
let usage_by_operation = usage_map(usage);
|
||||
let refs_by_operation = agent_ref_map(agent_refs);
|
||||
|
||||
Ok(summaries
|
||||
.into_iter()
|
||||
.map(|summary| {
|
||||
let operation_id = summary.id.as_str().to_owned();
|
||||
enrich_operation_summary(
|
||||
summary,
|
||||
usage_by_operation
|
||||
.get(&operation_id)
|
||||
.cloned()
|
||||
.unwrap_or_else(default_usage_summary),
|
||||
refs_by_operation
|
||||
.get(&operation_id)
|
||||
.cloned()
|
||||
.unwrap_or_default(),
|
||||
)
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_operation(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
operation_id: &OperationId,
|
||||
) -> Result<OperationDetailView, ApiError> {
|
||||
let summary = self
|
||||
.registry
|
||||
.get_operation_summary(workspace_id, operation_id)
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
ApiError::not_found_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<OperationVersionRecord, ApiError> {
|
||||
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<CreatedOperationResponse, ApiError> {
|
||||
self.validate_operation_payload(&payload)?;
|
||||
self.ensure_workspace_exists(workspace_id).await?;
|
||||
|
||||
if self
|
||||
.find_operation_by_name(workspace_id, &payload.name)
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
return Err(ApiError::conflict_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<crank_core::ToolQualityReport, ApiError> {
|
||||
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<CreatedOperationResponse, ApiError> {
|
||||
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<OperationMutationResult, ApiError> {
|
||||
let existing = self
|
||||
.get_operation_version(
|
||||
workspace_id,
|
||||
operation_id,
|
||||
self.get_operation(workspace_id, operation_id)
|
||||
.await?
|
||||
.current_draft_version,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let updated_at = 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<PublishResponse, ApiError> {
|
||||
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<OperationMutationResult, ApiError> {
|
||||
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<OperationMutationResult, ApiError> {
|
||||
let summary = self.get_operation(workspace_id, operation_id).await?;
|
||||
let updated_at = now_string()?;
|
||||
self.registry
|
||||
.delete_operation(workspace_id, operation_id)
|
||||
.await?;
|
||||
|
||||
Ok(OperationMutationResult {
|
||||
operation_id: operation_id.as_str().to_owned(),
|
||||
workspace_id: workspace_id.as_str().to_owned(),
|
||||
version: summary.current_draft_version,
|
||||
status: summary.status,
|
||||
updated_at,
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), version = payload.version))]
|
||||
pub async fn run_test(
|
||||
&self,
|
||||
workspace_id: &WorkspaceId,
|
||||
operation_id: &OperationId,
|
||||
payload: TestRunPayload,
|
||||
request_id: &str,
|
||||
) -> Result<TestRunResult, ApiError> {
|
||||
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)],
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user