diff --git a/apps/admin-api/src/dto.rs b/apps/admin-api/src/dto.rs new file mode 100644 index 0000000..e2c7611 --- /dev/null +++ b/apps/admin-api/src/dto.rs @@ -0,0 +1,458 @@ +use crank_core::{ + AgentId, AgentStatus, AuthConfig, AuthKind, ExecutionMode, ExportMode, GeneratedDraft, + InvocationLevel, InvocationSource, InvocationStatus, OperationSecurityLevel, OperationStatus, + PlatformApiKeyScope, Protocol, SecretKind, Target, UsagePeriod, WizardState, WorkspaceId, + WorkspaceStatus, +}; +use crank_mapping::MappingSet; +use crank_registry::{ + PlatformApiKeyRecord, RegistryOperation, UsageAgentBreakdown, UsageOperationBreakdown, + UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord, +}; +use crank_schema::Schema; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +#[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, +} + +#[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 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 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, +} + +#[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, +} + +fn default_operation_category() -> String { + "general".to_owned() +} + +fn default_export_mode() -> ExportMode { + ExportMode::Portable +} + +fn default_enabled() -> bool { + true +} + +pub(crate) struct InvocationRecordRequest<'a> { + pub workspace_id: &'a WorkspaceId, + pub agent_id: Option<&'a AgentId>, + pub operation: &'a RegistryOperation, + pub request_id: Option<&'a str>, + pub source: InvocationSource, + pub level: InvocationLevel, + pub status: InvocationStatus, + pub message: String, + pub status_code: Option, + pub error_kind: Option, + pub duration_ms: u64, + pub request_preview: Value, + pub response_preview: Value, +} diff --git a/apps/admin-api/src/lib.rs b/apps/admin-api/src/lib.rs index 7cf82c5..b556a1c 100644 --- a/apps/admin-api/src/lib.rs +++ b/apps/admin-api/src/lib.rs @@ -1,5 +1,6 @@ pub mod app; pub mod auth; +pub mod dto; pub mod error; pub mod import_guidance; pub mod rate_limit; diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 95b7956..0a83f42 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -4,28 +4,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, EditionCapabilities, ExecutionMode, ExportMode, - GeneratedDraft, GeneratedDraftStatus, IdentityError, IdentityProvider, InvocationLevel, - InvocationLog, InvocationLogId, InvocationSource, InvocationStatus, MembershipRole, - NoopAuditSink, OperationId, OperationSecurityLevel, OperationStatus, OwnerOnlyPolicyEngine, - PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, PlatformApiKeyStatus, PolicyEngine, - ProductEdition, Protocol, ResponseCachePolicy, SampleId, SecretKind, Target, - ToolQualityMappingRule, ToolQualityMappingSet, ToolQualitySchemaKind, ToolQualitySchemaNode, - UsagePeriod, WizardState, WorkspaceId, WorkspaceStatus, + AgentId, AuditSink, AuthProfile, AuthProfileId, CapabilityProfile, CommunityCapabilityProfile, + EditionCapabilities, ExecutionMode, GeneratedDraft, GeneratedDraftStatus, IdentityError, + IdentityProvider, InvocationLevel, InvocationLog, InvocationLogId, InvocationSource, + InvocationStatus, MembershipRole, NoopAuditSink, OperationId, OperationSecurityLevel, + OperationStatus, OwnerOnlyPolicyEngine, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, + PlatformApiKeyStatus, PolicyEngine, ProductEdition, Protocol, ResponseCachePolicy, SampleId, + Target, ToolQualityMappingRule, ToolQualityMappingSet, ToolQualitySchemaKind, + ToolQualitySchemaNode, UsagePeriod, WizardState, WorkspaceId, }; use crank_mapping::{JsonPathRoot, MappingRule, MappingSet, infer_mapping_from_samples}; use crank_registry::{ AgentSummary, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, ListInvocationLogsQuery, OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, PlatformApiKeyRecord, PostgresRegistry, RegistryOperation, SampleKind, - SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, UsageAgentBreakdown, UsageBucket, - UsageOperationBreakdown, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, - WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, + SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, UsageBucket, WorkspaceUpstream, + WorkspaceUpstreamId, }; use crank_runtime::{PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, SecretCrypto}; use crank_schema::{Schema, SchemaKind}; -use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; @@ -67,441 +64,7 @@ pub struct AdminServiceBuilder { 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, -} - -#[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 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 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, -} - -fn default_operation_category() -> String { - "general".to_owned() -} +pub use crate::dto::*; impl AdminService { #[cfg(test)] @@ -1538,10 +1101,6 @@ fn latest_sample_ref( .cloned() } -fn default_export_mode() -> ExportMode { - ExportMode::Portable -} - fn demo_revops_agent_payload() -> AgentPayload { AgentPayload { slug: "revops-copilot".to_owned(), @@ -1706,10 +1265,6 @@ 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()) } diff --git a/apps/admin-api/src/service/operations.rs b/apps/admin-api/src/service/operations.rs index efc69dd..f667ec0 100644 --- a/apps/admin-api/src/service/operations.rs +++ b/apps/admin-api/src/service/operations.rs @@ -5,7 +5,9 @@ use crank_core::{ use crank_registry::{ CreateVersionRequest, OperationVersionRecord, PublishRequest, RegistryOperation, }; -use crank_runtime::{RuntimeError, RuntimeOperation, RuntimeRequestContext}; +use crank_runtime::{ + RuntimeError, RuntimeExecutionRequest, RuntimeOperation, RuntimeRequestContext, +}; use serde_json::{Value, json}; use time::OffsetDateTime; use tracing::{info, instrument}; @@ -468,11 +470,10 @@ impl AdminService { 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), + .execute_request( + RuntimeExecutionRequest::new(&runtime, &payload.input) + .with_optional_auth(resolved_auth.as_ref()) + .with_context(&runtime_request_context), ) .await } diff --git a/crates/crank-community-mcp/src/app.rs b/crates/crank-community-mcp/src/app.rs index bc3562a..18de66e 100644 --- a/crates/crank-community-mcp/src/app.rs +++ b/crates/crank-community-mcp/src/app.rs @@ -10,12 +10,9 @@ use axum::{ extract::{Path, State}, http::{ HeaderMap, HeaderValue, StatusCode, - header::{self, ACCEPT, AUTHORIZATION, HeaderName, RETRY_AFTER}, - }, - response::{ - IntoResponse, Response, - sse::{Event, KeepAlive, Sse}, + header::{AUTHORIZATION, RETRY_AFTER}, }, + response::{IntoResponse, Response, sse::Event}, routing::get, }; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; @@ -25,8 +22,8 @@ use crank_core::{ }; use crank_registry::{CreateInvocationLogRequest, PostgresRegistry, PublishedAgentTool}; use crank_runtime::{ - RateLimitRejection, RequestRateLimiter, ResolvedAuth, RuntimeError, RuntimeExecutor, - RuntimeOperation, RuntimeRequestContext, SecretCrypto, + RateLimitRejection, RequestRateLimiter, ResolvedAuth, RuntimeError, RuntimeExecutionRequest, + RuntimeExecutor, RuntimeOperation, RuntimeRequestContext, SecretCrypto, }; use futures_util::stream; use serde::{Deserialize, Serialize}; @@ -39,9 +36,8 @@ use crate::{ auth::{SharedMachineCredentialVerifier, VerifiedMachineCredential}, catalog::PublishedToolCatalog, jsonrpc::{ - CURRENT_PROTOCOL_VERSION, DEFAULT_PROTOCOL_VERSION, is_notification, is_request, - is_response, jsonrpc_error, jsonrpc_result, method_name, negotiated_protocol_version, - params, request_id, + DEFAULT_PROTOCOL_VERSION, is_notification, is_request, is_response, jsonrpc_error, + jsonrpc_result, method_name, negotiated_protocol_version, params, request_id, }, manifest::tool_definitions, session::{SessionState, SharedSessionStore}, @@ -49,20 +45,16 @@ use crate::{ ToolErrorContract, generic_tool_error_contract, runtime_error_code, tool_error_contract_from_runtime, tool_error_text, tool_error_value, }, + transport::{ + AllowedOrigins, HEADER_MCP_SESSION_ID, ResponseMode, json_response, + negotiate_post_response_mode, protocol_version_from_headers, resolve_request_id, + session_id_from_headers, sse_response, transport_response, validate_get_accept_header, + validate_origin, validate_session_protocol_version, with_request_id_header, + }, }; -const HEADER_MCP_SESSION_ID: &str = "MCP-Session-Id"; -const HEADER_MCP_PROTOCOL_VERSION: &str = "MCP-Protocol-Version"; -const HEADER_X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id"); -const MAX_REQUEST_ID_LEN: usize = 128; const TRANSPORT_SESSION_TTL_MS: u64 = 86_400_000; -#[derive(Clone, Copy)] -enum ResponseMode { - Json, - Sse, -} - #[derive(Clone)] pub struct AppState { registry: PostgresRegistry, @@ -75,11 +67,6 @@ pub struct AppState { allowed_origins: AllowedOrigins, } -#[derive(Clone)] -pub struct AllowedOrigins { - public_origin: Option, -} - #[derive(Debug, Serialize, Deserialize)] struct InitializeParams { #[serde(rename = "protocolVersion")] @@ -623,11 +610,10 @@ async fn handle_base_tool_call( Ok(resolved_auth) => { state .runtime - .execute_with_auth_and_context( - &operation, - &arguments, - resolved_auth.as_ref(), - Some(&runtime_request_context), + .execute_request( + RuntimeExecutionRequest::new(&operation, &arguments) + .with_optional_auth(resolved_auth.as_ref()) + .with_context(&runtime_request_context), ) .await } @@ -1022,141 +1008,6 @@ fn serialize_machine_access_mode(mode: crank_core::MachineAccessMode) -> &'stati } } -fn validate_origin( - allowed_origins: &AllowedOrigins, - headers: &HeaderMap, -) -> Result<(), StatusCode> { - let Some(origin) = headers.get(header::ORIGIN) else { - return Ok(()); - }; - let Ok(origin) = origin.to_str() else { - return Err(StatusCode::FORBIDDEN); - }; - - if allowed_origins.is_allowed(origin) { - return Ok(()); - } - - Err(StatusCode::FORBIDDEN) -} - -fn negotiate_post_response_mode(headers: &HeaderMap) -> Result { - let Some(accept) = headers.get(ACCEPT) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(accept) = accept.to_str() else { - return Err(StatusCode::BAD_REQUEST); - }; - - let mut saw_json = false; - let mut saw_sse = false; - let mut preferred = None; - - for part in accept.split(',') { - let media_type = part - .split(';') - .next() - .unwrap_or_default() - .trim() - .to_ascii_lowercase(); - - match media_type.as_str() { - "application/json" => { - saw_json = true; - if preferred.is_none() { - preferred = Some(ResponseMode::Json); - } - } - "text/event-stream" => { - saw_sse = true; - if preferred.is_none() { - preferred = Some(ResponseMode::Sse); - } - } - _ => {} - } - } - - if saw_json || saw_sse { - return preferred.ok_or(StatusCode::NOT_ACCEPTABLE); - } - - Err(StatusCode::NOT_ACCEPTABLE) -} - -fn validate_get_accept_header(headers: &HeaderMap) -> Result<(), StatusCode> { - let Some(accept) = headers.get(ACCEPT) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(accept) = accept.to_str() else { - return Err(StatusCode::BAD_REQUEST); - }; - - if accept - .split(',') - .map(|part| { - part.split(';') - .next() - .unwrap_or_default() - .trim() - .to_ascii_lowercase() - }) - .any(|media_type| media_type == "text/event-stream") - { - return Ok(()); - } - - Err(StatusCode::NOT_ACCEPTABLE) -} - -fn protocol_version_from_headers(headers: &HeaderMap) -> Result { - let Some(version) = headers.get(HEADER_MCP_PROTOCOL_VERSION) else { - return Ok(DEFAULT_PROTOCOL_VERSION.to_owned()); - }; - let Ok(version) = version.to_str() else { - return Err(StatusCode::BAD_REQUEST); - }; - - if negotiated_protocol_version(version).is_some() { - return Ok(version.to_owned()); - } - - Err(StatusCode::BAD_REQUEST) -} - -fn validate_session_protocol_version( - headers: &HeaderMap, - negotiated_session_version: &str, -) -> Result<(), StatusCode> { - let Some(version) = headers.get(HEADER_MCP_PROTOCOL_VERSION) else { - return Ok(()); - }; - let Ok(version) = version.to_str() else { - return Err(StatusCode::BAD_REQUEST); - }; - - if negotiated_protocol_version(version).is_none() { - return Err(StatusCode::BAD_REQUEST); - } - - if version != negotiated_session_version { - return Err(StatusCode::BAD_REQUEST); - } - - Ok(()) -} - -fn session_id_from_headers(headers: &HeaderMap) -> Result, StatusCode> { - let Some(session_id) = headers.get(HEADER_MCP_SESSION_ID) else { - return Ok(None); - }; - let Ok(session_id) = session_id.to_str() else { - return Err(StatusCode::BAD_REQUEST); - }; - - Ok(Some(session_id.to_owned())) -} - fn internal_jsonrpc_error(message: &Value, error: impl std::fmt::Display) -> Response { transport_response( StatusCode::INTERNAL_SERVER_ERROR, @@ -1308,109 +1159,6 @@ fn hash_access_secret(secret: &str) -> String { URL_SAFE_NO_PAD.encode(digest) } -fn json_response( - status: StatusCode, - payload: Value, - session_id: Option<&str>, - protocol_version: Option<&str>, -) -> Response { - let mut response = (status, Json(payload)).into_response(); - - if let Some(session_id) = session_id { - response.headers_mut().insert( - HEADER_MCP_SESSION_ID, - HeaderValue::from_str(session_id) - .unwrap_or_else(|_| HeaderValue::from_static("invalid")), - ); - } - - if let Some(protocol_version) = protocol_version { - response.headers_mut().insert( - HEADER_MCP_PROTOCOL_VERSION, - HeaderValue::from_str(protocol_version) - .unwrap_or_else(|_| HeaderValue::from_static(CURRENT_PROTOCOL_VERSION)), - ); - } - - response -} - -fn transport_response( - status: StatusCode, - payload: Value, - response_mode: ResponseMode, - session_id: Option<&str>, - protocol_version: Option<&str>, -) -> Response { - if status == StatusCode::OK && matches!(response_mode, ResponseMode::Sse) { - let payload = payload.to_string(); - let stream = stream::once(async move { Ok(Event::default().data(payload)) }); - - return sse_response(status, stream, session_id, protocol_version); - } - - json_response(status, payload, session_id, protocol_version) -} - -fn with_request_id_header(mut response: Response, request_id: &str) -> Response { - if let Ok(value) = HeaderValue::from_str(request_id) { - response.headers_mut().insert(HEADER_X_REQUEST_ID, value); - } - response -} - -fn sse_response( - status: StatusCode, - stream: S, - session_id: Option<&str>, - protocol_version: Option<&str>, -) -> Response -where - S: futures_util::stream::Stream> + Send + 'static, -{ - let mut response = ( - status, - Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))), - ) - .into_response(); - - if let Some(session_id) = session_id { - response.headers_mut().insert( - HEADER_MCP_SESSION_ID, - HeaderValue::from_str(session_id) - .unwrap_or_else(|_| HeaderValue::from_static("invalid")), - ); - } - - if let Some(protocol_version) = protocol_version { - response.headers_mut().insert( - HEADER_MCP_PROTOCOL_VERSION, - HeaderValue::from_str(protocol_version) - .unwrap_or_else(|_| HeaderValue::from_static(CURRENT_PROTOCOL_VERSION)), - ); - } - - response -} - -fn resolve_request_id(headers: &HeaderMap) -> String { - headers - .get(&HEADER_X_REQUEST_ID) - .and_then(|value| value.to_str().ok()) - .map(str::trim) - .filter(|value| is_valid_request_id(value)) - .map(ToOwned::to_owned) - .unwrap_or_else(|| uuid::Uuid::now_v7().to_string()) -} - -fn is_valid_request_id(value: &str) -> bool { - !value.is_empty() - && value.len() <= MAX_REQUEST_ID_LEN - && value - .bytes() - .all(|byte| matches!(byte, 0x21..=0x7e) && byte != b',' && byte != b';') -} - fn resolve_generated_tool( tools: &[PublishedAgentTool], tool_name: &str, @@ -1432,42 +1180,6 @@ fn runtime_operation(tool: &PublishedAgentTool) -> RuntimeOperation { operation } -impl AllowedOrigins { - fn new(public_base_url: Option) -> Self { - Self { - public_origin: public_base_url.and_then(|value| extract_origin(&value)), - } - } - - fn is_allowed(&self, origin: &str) -> bool { - if origin.starts_with("http://localhost") - || origin.starts_with("http://127.0.0.1") - || origin.starts_with("https://localhost") - || origin.starts_with("https://127.0.0.1") - { - return true; - } - - match &self.public_origin { - Some(public_origin) => public_origin == origin, - None => false, - } - } -} - -fn extract_origin(url: &str) -> Option { - let mut parts = url.split('/'); - let scheme = parts.next()?; - let empty = parts.next()?; - let authority = parts.next()?; - - if empty.is_empty() { - return Some(format!("{scheme}//{authority}")); - } - - None -} - #[cfg(test)] mod tests { use axum::body::to_bytes; diff --git a/crates/crank-community-mcp/src/lib.rs b/crates/crank-community-mcp/src/lib.rs index 4104352..7ebf2e3 100644 --- a/crates/crank-community-mcp/src/lib.rs +++ b/crates/crank-community-mcp/src/lib.rs @@ -5,5 +5,6 @@ pub mod jsonrpc; pub mod manifest; pub mod session; pub mod tool_error; +mod transport; pub use app::build_app; diff --git a/crates/crank-community-mcp/src/transport.rs b/crates/crank-community-mcp/src/transport.rs new file mode 100644 index 0000000..69b54d4 --- /dev/null +++ b/crates/crank-community-mcp/src/transport.rs @@ -0,0 +1,311 @@ +use std::{convert::Infallible, time::Duration}; + +use axum::{ + Json, + http::{ + HeaderMap, HeaderValue, StatusCode, + header::{self, ACCEPT, HeaderName}, + }, + response::{ + IntoResponse, Response, + sse::{Event, KeepAlive, Sse}, + }, +}; +use futures_util::stream; +use serde_json::Value; + +use crate::jsonrpc::{ + CURRENT_PROTOCOL_VERSION, DEFAULT_PROTOCOL_VERSION, negotiated_protocol_version, +}; + +pub(super) const HEADER_MCP_SESSION_ID: &str = "MCP-Session-Id"; +pub(super) const HEADER_MCP_PROTOCOL_VERSION: &str = "MCP-Protocol-Version"; +pub(super) const HEADER_X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id"); +const MAX_REQUEST_ID_LEN: usize = 128; + +#[derive(Clone, Copy)] +pub(super) enum ResponseMode { + Json, + Sse, +} + +#[derive(Clone)] +pub(super) struct AllowedOrigins { + public_origin: Option, +} + +impl AllowedOrigins { + pub(super) fn new(public_base_url: Option) -> Self { + Self { + public_origin: public_base_url.and_then(|value| extract_origin(&value)), + } + } + + pub(super) fn is_allowed(&self, origin: &str) -> bool { + if origin.starts_with("http://localhost") + || origin.starts_with("http://127.0.0.1") + || origin.starts_with("https://localhost") + || origin.starts_with("https://127.0.0.1") + { + return true; + } + + match &self.public_origin { + Some(public_origin) => public_origin == origin, + None => false, + } + } +} + +pub(super) fn validate_origin( + allowed_origins: &AllowedOrigins, + headers: &HeaderMap, +) -> Result<(), StatusCode> { + let Some(origin) = headers.get(header::ORIGIN) else { + return Ok(()); + }; + let Ok(origin) = origin.to_str() else { + return Err(StatusCode::FORBIDDEN); + }; + + if allowed_origins.is_allowed(origin) { + return Ok(()); + } + + Err(StatusCode::FORBIDDEN) +} + +pub(super) fn negotiate_post_response_mode( + headers: &HeaderMap, +) -> Result { + let Some(accept) = headers.get(ACCEPT) else { + return Err(StatusCode::BAD_REQUEST); + }; + let Ok(accept) = accept.to_str() else { + return Err(StatusCode::BAD_REQUEST); + }; + + let mut saw_json = false; + let mut saw_sse = false; + let mut preferred = None; + + for part in accept.split(',') { + let media_type = part + .split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + + match media_type.as_str() { + "application/json" => { + saw_json = true; + if preferred.is_none() { + preferred = Some(ResponseMode::Json); + } + } + "text/event-stream" => { + saw_sse = true; + if preferred.is_none() { + preferred = Some(ResponseMode::Sse); + } + } + _ => {} + } + } + + if saw_json || saw_sse { + return preferred.ok_or(StatusCode::NOT_ACCEPTABLE); + } + + Err(StatusCode::NOT_ACCEPTABLE) +} + +pub(super) fn validate_get_accept_header(headers: &HeaderMap) -> Result<(), StatusCode> { + let Some(accept) = headers.get(ACCEPT) else { + return Err(StatusCode::BAD_REQUEST); + }; + let Ok(accept) = accept.to_str() else { + return Err(StatusCode::BAD_REQUEST); + }; + + if accept + .split(',') + .map(|part| { + part.split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase() + }) + .any(|media_type| media_type == "text/event-stream") + { + return Ok(()); + } + + Err(StatusCode::NOT_ACCEPTABLE) +} + +pub(super) fn protocol_version_from_headers(headers: &HeaderMap) -> Result { + let Some(version) = headers.get(HEADER_MCP_PROTOCOL_VERSION) else { + return Ok(DEFAULT_PROTOCOL_VERSION.to_owned()); + }; + let Ok(version) = version.to_str() else { + return Err(StatusCode::BAD_REQUEST); + }; + + if negotiated_protocol_version(version).is_some() { + return Ok(version.to_owned()); + } + + Err(StatusCode::BAD_REQUEST) +} + +pub(super) fn validate_session_protocol_version( + headers: &HeaderMap, + negotiated_session_version: &str, +) -> Result<(), StatusCode> { + let Some(version) = headers.get(HEADER_MCP_PROTOCOL_VERSION) else { + return Ok(()); + }; + let Ok(version) = version.to_str() else { + return Err(StatusCode::BAD_REQUEST); + }; + + if negotiated_protocol_version(version).is_none() { + return Err(StatusCode::BAD_REQUEST); + } + + if version != negotiated_session_version { + return Err(StatusCode::BAD_REQUEST); + } + + Ok(()) +} + +pub(super) fn session_id_from_headers(headers: &HeaderMap) -> Result, StatusCode> { + let Some(session_id) = headers.get(HEADER_MCP_SESSION_ID) else { + return Ok(None); + }; + let Ok(session_id) = session_id.to_str() else { + return Err(StatusCode::BAD_REQUEST); + }; + + Ok(Some(session_id.to_owned())) +} + +pub(super) fn json_response( + status: StatusCode, + payload: Value, + session_id: Option<&str>, + protocol_version: Option<&str>, +) -> Response { + let mut response = (status, Json(payload)).into_response(); + + if let Some(session_id) = session_id { + response.headers_mut().insert( + HEADER_MCP_SESSION_ID, + HeaderValue::from_str(session_id) + .unwrap_or_else(|_| HeaderValue::from_static("invalid")), + ); + } + + if let Some(protocol_version) = protocol_version { + response.headers_mut().insert( + HEADER_MCP_PROTOCOL_VERSION, + HeaderValue::from_str(protocol_version) + .unwrap_or_else(|_| HeaderValue::from_static(CURRENT_PROTOCOL_VERSION)), + ); + } + + response +} + +pub(super) fn transport_response( + status: StatusCode, + payload: Value, + response_mode: ResponseMode, + session_id: Option<&str>, + protocol_version: Option<&str>, +) -> Response { + if status == StatusCode::OK && matches!(response_mode, ResponseMode::Sse) { + let payload = payload.to_string(); + let stream = stream::once(async move { Ok(Event::default().data(payload)) }); + + return sse_response(status, stream, session_id, protocol_version); + } + + json_response(status, payload, session_id, protocol_version) +} + +pub(super) fn with_request_id_header(mut response: Response, request_id: &str) -> Response { + if let Ok(value) = HeaderValue::from_str(request_id) { + response.headers_mut().insert(HEADER_X_REQUEST_ID, value); + } + response +} + +pub(super) fn sse_response( + status: StatusCode, + stream: S, + session_id: Option<&str>, + protocol_version: Option<&str>, +) -> Response +where + S: futures_util::stream::Stream> + Send + 'static, +{ + let mut response = ( + status, + Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))), + ) + .into_response(); + + if let Some(session_id) = session_id { + response.headers_mut().insert( + HEADER_MCP_SESSION_ID, + HeaderValue::from_str(session_id) + .unwrap_or_else(|_| HeaderValue::from_static("invalid")), + ); + } + + if let Some(protocol_version) = protocol_version { + response.headers_mut().insert( + HEADER_MCP_PROTOCOL_VERSION, + HeaderValue::from_str(protocol_version) + .unwrap_or_else(|_| HeaderValue::from_static(CURRENT_PROTOCOL_VERSION)), + ); + } + + response +} + +pub(super) fn resolve_request_id(headers: &HeaderMap) -> String { + headers + .get(&HEADER_X_REQUEST_ID) + .and_then(|value| value.to_str().ok()) + .map(str::trim) + .filter(|value| is_valid_request_id(value)) + .map(ToOwned::to_owned) + .unwrap_or_else(|| uuid::Uuid::now_v7().to_string()) +} + +pub(super) fn is_valid_request_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_REQUEST_ID_LEN + && value + .bytes() + .all(|byte| matches!(byte, 0x21..=0x7e) && byte != b',' && byte != b';') +} + +pub(super) fn extract_origin(url: &str) -> Option { + let mut parts = url.split('/'); + let scheme = parts.next()?; + let empty = parts.next()?; + let authority = parts.next()?; + + if empty.is_empty() { + return Some(format!("{scheme}//{authority}")); + } + + None +} diff --git a/crates/crank-core/src/lib.rs b/crates/crank-core/src/lib.rs index facf59a..d4cc5bc 100644 --- a/crates/crank-core/src/lib.rs +++ b/crates/crank-core/src/lib.rs @@ -12,6 +12,77 @@ pub mod secret; pub mod tool_quality; pub mod workspace; +pub mod domain { + pub use crate::access::{ + InvitationStatus, InvitationToken, Membership, MembershipRole, PlatformApiKey, + PlatformApiKeyScope, PlatformApiKeyStatus, User, UserStatus, + }; + pub use crate::agent::{Agent, AgentOperationBinding, AgentStatus, AgentVersion}; + pub use crate::auth::{ + ApiKeyHeaderAuthConfig, ApiKeyQueryAuthConfig, AuthConfig, AuthProfile, BasicAuthConfig, + BearerAuthConfig, + }; + pub use crate::cache::{ + CacheBackend, CacheScope, CachedHeader, CachedResponse, ParseCacheBackendError, + RateLimitBucketState, + }; + pub use crate::edition::{ + EditionCapabilities, EditionLimits, MachineAccessMode, OperationSecurityLevel, + ProductEdition, + }; + pub use crate::ids::{ + AgentId, AuditEventId, AuthProfileId, DescriptorId, InvitationId, InvocationLogId, + OperationId, PlatformApiKeyId, SampleId, SecretId, ToolId, UserId, UserSessionId, + WorkspaceId, + }; + pub use crate::observability::{ + InvocationLevel, InvocationLog, InvocationSource, InvocationStatus, UsagePeriod, + UsageRollup, + }; + pub use crate::operation::{ + ConfigExport, ConfirmationPolicy, ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, + IdempotencyMode, IdempotencyPolicy, Operation, OperationSafetyClass, OperationSafetyPolicy, + OperationStatus, ResponseCachePolicy, RestTarget, RetryPolicy, Samples, Target, + ToolDescription, ToolExample, WizardState, + }; + pub use crate::protocol::{AuthKind, ExportMode, HttpMethod, Protocol}; + pub use crate::secret::{Secret, SecretKind, SecretStatus, SecretVersion}; + pub use crate::tool_quality::{ + ToolQualityCatalogTool, ToolQualityFinding, ToolQualityMappingRule, ToolQualityMappingSet, + ToolQualityReport, ToolQualitySchemaKind, ToolQualitySchemaNode, ToolQualitySeverity, + }; + pub use crate::workspace::{Workspace, WorkspaceStatus}; +} + +pub mod ports { + pub use crate::cache::{ + CacheStoreError, CoordinationStateStore, CoordinationStateValue, RateLimitStateStore, + ReplayGuardStatus, ReplayGuardStore, ResponseCacheStore, + }; + pub use crate::ext::access::{ + OwnerOnlyPolicyEngine, PolicyAction, PolicyDecision, PolicyEngine, PolicyScope, + SessionActor, + }; + pub use crate::ext::audit::{ + AuditActor, AuditError, AuditEvent, AuditEventPage, AuditQuery, AuditSink, AuditTarget, + AuditTargetKind, NoopAuditSink, SharedAuditSink, + }; + pub use crate::ext::auth::{ + AuthenticatedIdentity, IdentityError, IdentityProvider, IdentityProviderKind, LoginOutcome, + LoginPayload, MachineCredentialVerifier, MachineCredentialVerifierError, + SharedIdentityProvider, SharedMachineCredentialVerifier, VerifiedMachineCredential, + }; + pub use crate::ext::capability::{CapabilityProfile, CommunityCapabilityProfile}; + pub use crate::ext::metering::{ + MeteringEvent, MeteringSink, NoopMeteringSink, SharedMeteringSink, + }; + pub use crate::ext::protocol::{ + AdapterRegistry, AdapterResponse, ExecutionMode, MeteringContext, PreparedRequest, + ProtocolAdapter, ProtocolAdapterError, ResponseCacheScope, RuntimeRequestContext, + SharedProtocolAdapter, + }; +} + pub use access::{ InvitationStatus, InvitationToken, Membership, MembershipRole, PlatformApiKey, PlatformApiKeyScope, PlatformApiKeyStatus, User, UserStatus, diff --git a/crates/crank-registry/src/lib.rs b/crates/crank-registry/src/lib.rs index e2ae129..e19f452 100644 --- a/crates/crank-registry/src/lib.rs +++ b/crates/crank-registry/src/lib.rs @@ -6,6 +6,37 @@ mod postgres; pub use error::RegistryError; pub use ext::{ExtensionMigration, RegistryExtension, apply_extension_migrations}; + +pub mod records { + pub use crate::model::{ + AgentSummary, AgentVersionRecord, AuthUserRecord, DescriptorKind, DescriptorMetadata, + InvitationRecord, InvocationLogRecord, MembershipRecord, OperationAgentRef, + OperationSampleMetadata, OperationSummary, OperationUsageSummary, OperationVersionRecord, + Page, PlatformApiKeyRecord, PublishedAgentTool, RegistryOperation, SampleKind, + SecretRecord, SecretVersionRecord, SessionRecord, UsageAgentBreakdown, UsageBucket, + UsageOperationBreakdown, UsageRollupRecord, UsageSummary, UsageTimelinePoint, + WorkspaceMembershipRecord, WorkspaceRecord, WorkspaceUpstream, WorkspaceUpstreamId, + YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, + }; +} + +pub mod requests { + pub use crate::model::{ + CreateAgentDraftVersionRequest, CreateAgentRequest, CreateInvitationRequest, + CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest, + CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest, + ListInvocationLogsQuery, PublishAgentRequest, PublishRequest, RotateSecretRequest, + SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, + SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, UpdateWorkspaceRequest, + UsageQuery, + }; +} + +pub mod infrastructure { + pub use crate::ext::{ExtensionMigration, RegistryExtension, apply_extension_migrations}; + pub use crate::postgres::{PostgresPoolConfig, PostgresPoolConfigError, PostgresRegistry}; +} + pub use model::{ AgentSummary, AgentVersionRecord, AuthUserRecord, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateInvitationRequest, CreateInvocationLogRequest, diff --git a/crates/crank-registry/src/postgres/connection.rs b/crates/crank-registry/src/postgres/connection.rs new file mode 100644 index 0000000..698236c --- /dev/null +++ b/crates/crank-registry/src/postgres/connection.rs @@ -0,0 +1,70 @@ +use std::time::Duration; + +use sqlx::{ + PgPool, + postgres::{PgConnectOptions, PgPoolOptions}, +}; + +use crate::{error::RegistryError, migrations}; + +use super::{PostgresPoolConfig, PostgresRegistry}; + +impl PostgresRegistry { + pub async fn connect(database_url: &str) -> Result { + Self::connect_in_schema(database_url, None).await + } + + pub async fn connect_with_options( + connect_options: PgConnectOptions, + ) -> Result { + Self::connect_with_options_and_pool_config(connect_options, PostgresPoolConfig::default()) + .await + } + + pub async fn connect_with_options_and_pool_config( + connect_options: PgConnectOptions, + pool_config: PostgresPoolConfig, + ) -> Result { + let pool = PgPoolOptions::new() + .max_connections(pool_config.max_connections) + .min_connections(pool_config.min_connections) + .acquire_timeout(Duration::from_millis(pool_config.acquire_timeout_ms)) + .idle_timeout(Duration::from_millis(pool_config.idle_timeout_ms)) + .max_lifetime(Duration::from_millis(pool_config.max_lifetime_ms)) + .connect_with(connect_options) + .await?; + migrations::apply_postgres(&pool).await?; + Ok(Self { pool }) + } + + pub fn pool(&self) -> &PgPool { + &self.pool + } + + pub async fn migrate(&self) -> Result<(), RegistryError> { + migrations::apply_postgres(&self.pool).await?; + Ok(()) + } + + async fn connect_in_schema( + database_url: &str, + schema: Option<&str>, + ) -> Result { + let pool = PgPoolOptions::new() + .min_connections(0) + .max_connections(1) + .idle_timeout(std::time::Duration::from_secs(1)) + .connect(database_url) + .await?; + + if let Some(schema) = schema { + sqlx::query(&format!("set search_path to {schema}")) + .execute(&pool) + .await?; + } + + let registry = Self { pool }; + registry.migrate().await?; + Ok(registry) + } +} diff --git a/crates/crank-registry/src/postgres/mod.rs b/crates/crank-registry/src/postgres/mod.rs index 64b4158..def377a 100644 --- a/crates/crank-registry/src/postgres/mod.rs +++ b/crates/crank-registry/src/postgres/mod.rs @@ -1,8 +1,10 @@ mod agent; mod api_key; mod auth; +mod connection; mod observability; mod operation; +mod pool_config; mod secret; mod upstream; mod workspace; @@ -16,18 +18,13 @@ use crank_core::{ }; use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; -use sqlx::{ - PgPool, Postgres, Row, Transaction, - postgres::{PgConnectOptions, PgPoolOptions, PgRow}, - types::Json, -}; -use std::{collections::BTreeMap, env, time::Duration}; -use thiserror::Error; +use sqlx::{PgPool, Postgres, Row, Transaction, postgres::PgRow, types::Json}; use time::OffsetDateTime; +pub use pool_config::{PostgresPoolConfig, PostgresPoolConfigError}; + use crate::{ error::RegistryError, - migrations, model::{ AgentSummary, AgentVersionRecord, AuthUserRecord, CreateAgentDraftVersionRequest, CreateAgentRequest, CreateInvitationRequest, CreateInvocationLogRequest, @@ -50,156 +47,7 @@ pub struct PostgresRegistry { pool: PgPool, } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct PostgresPoolConfig { - pub max_connections: u32, - pub min_connections: u32, - pub acquire_timeout_ms: u64, - pub idle_timeout_ms: u64, - pub max_lifetime_ms: u64, -} - -#[derive(Debug, Error, PartialEq, Eq)] -pub enum PostgresPoolConfigError { - #[error("invalid postgres pool setting {name}={value}")] - InvalidValue { name: &'static str, value: String }, - #[error("POSTGRES_MAX_CONNECTIONS must be greater than zero")] - ZeroMaxConnections, - #[error("POSTGRES_MIN_CONNECTIONS must not exceed POSTGRES_MAX_CONNECTIONS")] - MinConnectionsExceedMax, -} - -impl Default for PostgresPoolConfig { - fn default() -> Self { - Self { - max_connections: 20, - min_connections: 2, - acquire_timeout_ms: 5_000, - idle_timeout_ms: 600_000, - max_lifetime_ms: 1_800_000, - } - } -} - -impl PostgresPoolConfig { - pub fn from_env() -> Result { - Self::from_vars(env::vars()) - } - - fn from_vars(vars: I) -> Result - where - I: IntoIterator, - K: AsRef, - V: AsRef, - { - let vars = vars - .into_iter() - .map(|(name, value)| (name.as_ref().to_owned(), value.as_ref().to_owned())) - .collect::>(); - let defaults = Self::default(); - let config = Self { - max_connections: parse_u32_setting( - &vars, - "POSTGRES_MAX_CONNECTIONS", - defaults.max_connections, - )?, - min_connections: parse_u32_setting( - &vars, - "POSTGRES_MIN_CONNECTIONS", - defaults.min_connections, - )?, - acquire_timeout_ms: parse_u64_setting( - &vars, - "POSTGRES_ACQUIRE_TIMEOUT_MS", - defaults.acquire_timeout_ms, - )?, - idle_timeout_ms: parse_u64_setting( - &vars, - "POSTGRES_IDLE_TIMEOUT_MS", - defaults.idle_timeout_ms, - )?, - max_lifetime_ms: parse_u64_setting( - &vars, - "POSTGRES_MAX_LIFETIME_MS", - defaults.max_lifetime_ms, - )?, - }; - - config.validate()?; - Ok(config) - } - - fn validate(self) -> Result { - if self.max_connections == 0 { - return Err(PostgresPoolConfigError::ZeroMaxConnections); - } - if self.min_connections > self.max_connections { - return Err(PostgresPoolConfigError::MinConnectionsExceedMax); - } - - Ok(self) - } -} - impl PostgresRegistry { - pub async fn connect(database_url: &str) -> Result { - Self::connect_in_schema(database_url, None).await - } - - pub async fn connect_with_options( - connect_options: PgConnectOptions, - ) -> Result { - Self::connect_with_options_and_pool_config(connect_options, PostgresPoolConfig::default()) - .await - } - - pub async fn connect_with_options_and_pool_config( - connect_options: PgConnectOptions, - pool_config: PostgresPoolConfig, - ) -> Result { - let pool = PgPoolOptions::new() - .max_connections(pool_config.max_connections) - .min_connections(pool_config.min_connections) - .acquire_timeout(Duration::from_millis(pool_config.acquire_timeout_ms)) - .idle_timeout(Duration::from_millis(pool_config.idle_timeout_ms)) - .max_lifetime(Duration::from_millis(pool_config.max_lifetime_ms)) - .connect_with(connect_options) - .await?; - migrations::apply_postgres(&pool).await?; - Ok(Self { pool }) - } - - pub fn pool(&self) -> &PgPool { - &self.pool - } - - pub async fn migrate(&self) -> Result<(), RegistryError> { - migrations::apply_postgres(&self.pool).await?; - Ok(()) - } - - async fn connect_in_schema( - database_url: &str, - schema: Option<&str>, - ) -> Result { - let pool = PgPoolOptions::new() - .min_connections(0) - .max_connections(1) - .idle_timeout(std::time::Duration::from_secs(1)) - .connect(database_url) - .await?; - - if let Some(schema) = schema { - sqlx::query(&format!("set search_path to {schema}")) - .execute(&pool) - .await?; - } - - let registry = Self { pool }; - registry.migrate().await?; - Ok(registry) - } - async fn list_agent_bindings( &self, agent_id: &AgentId, @@ -226,42 +74,6 @@ impl PostgresRegistry { } } -fn parse_u32_setting( - vars: &BTreeMap, - name: &'static str, - default: u32, -) -> Result { - vars.get(name) - .map(|value| { - value - .parse::() - .map_err(|_| PostgresPoolConfigError::InvalidValue { - name, - value: value.clone(), - }) - }) - .transpose() - .map(|value| value.unwrap_or(default)) -} - -fn parse_u64_setting( - vars: &BTreeMap, - name: &'static str, - default: u64, -) -> Result { - vars.get(name) - .map(|value| { - value - .parse::() - .map_err(|_| PostgresPoolConfigError::InvalidValue { - name, - value: value.clone(), - }) - }) - .transpose() - .map(|value| value.unwrap_or(default)) -} - async fn insert_version_row( tx: &mut Transaction<'_, Postgres>, snapshot: &RegistryOperation, @@ -930,79 +742,3 @@ fn usage_bucket_sql(bucket: crate::model::UsageBucket) -> &'static str { crate::model::UsageBucket::Month => "month", } } - -#[cfg(test)] -mod pool_config_tests { - use super::{PostgresPoolConfig, PostgresPoolConfigError}; - - #[test] - fn pool_config_uses_explicit_defaults() { - let config = PostgresPoolConfig::from_vars(std::iter::empty::<(&str, &str)>()).unwrap(); - - assert_eq!( - config, - PostgresPoolConfig { - max_connections: 20, - min_connections: 2, - acquire_timeout_ms: 5_000, - idle_timeout_ms: 600_000, - max_lifetime_ms: 1_800_000, - } - ); - } - - #[test] - fn pool_config_parses_overrides() { - let config = PostgresPoolConfig::from_vars([ - ("POSTGRES_MAX_CONNECTIONS", "32"), - ("POSTGRES_MIN_CONNECTIONS", "4"), - ("POSTGRES_ACQUIRE_TIMEOUT_MS", "7000"), - ("POSTGRES_IDLE_TIMEOUT_MS", "900000"), - ("POSTGRES_MAX_LIFETIME_MS", "3600000"), - ]) - .unwrap(); - - assert_eq!( - config, - PostgresPoolConfig { - max_connections: 32, - min_connections: 4, - acquire_timeout_ms: 7_000, - idle_timeout_ms: 900_000, - max_lifetime_ms: 3_600_000, - } - ); - } - - #[test] - fn pool_config_rejects_invalid_numeric_value() { - let error = - PostgresPoolConfig::from_vars([("POSTGRES_MAX_CONNECTIONS", "abc")]).unwrap_err(); - - assert_eq!( - error, - PostgresPoolConfigError::InvalidValue { - name: "POSTGRES_MAX_CONNECTIONS", - value: "abc".to_owned(), - } - ); - } - - #[test] - fn pool_config_rejects_zero_max_connections() { - let error = PostgresPoolConfig::from_vars([("POSTGRES_MAX_CONNECTIONS", "0")]).unwrap_err(); - - assert_eq!(error, PostgresPoolConfigError::ZeroMaxConnections); - } - - #[test] - fn pool_config_rejects_min_connections_above_max() { - let error = PostgresPoolConfig::from_vars([ - ("POSTGRES_MAX_CONNECTIONS", "2"), - ("POSTGRES_MIN_CONNECTIONS", "3"), - ]) - .unwrap_err(); - - assert_eq!(error, PostgresPoolConfigError::MinConnectionsExceedMax); - } -} diff --git a/crates/crank-registry/src/postgres/pool_config.rs b/crates/crank-registry/src/postgres/pool_config.rs new file mode 100644 index 0000000..d32f98e --- /dev/null +++ b/crates/crank-registry/src/postgres/pool_config.rs @@ -0,0 +1,206 @@ +use std::{collections::BTreeMap, env}; + +use thiserror::Error; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct PostgresPoolConfig { + pub max_connections: u32, + pub min_connections: u32, + pub acquire_timeout_ms: u64, + pub idle_timeout_ms: u64, + pub max_lifetime_ms: u64, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum PostgresPoolConfigError { + #[error("invalid postgres pool setting {name}={value}")] + InvalidValue { name: &'static str, value: String }, + #[error("POSTGRES_MAX_CONNECTIONS must be greater than zero")] + ZeroMaxConnections, + #[error("POSTGRES_MIN_CONNECTIONS must not exceed POSTGRES_MAX_CONNECTIONS")] + MinConnectionsExceedMax, +} + +impl Default for PostgresPoolConfig { + fn default() -> Self { + Self { + max_connections: 20, + min_connections: 2, + acquire_timeout_ms: 5_000, + idle_timeout_ms: 600_000, + max_lifetime_ms: 1_800_000, + } + } +} + +impl PostgresPoolConfig { + pub fn from_env() -> Result { + Self::from_vars(env::vars()) + } + + fn from_vars(vars: I) -> Result + where + I: IntoIterator, + K: AsRef, + V: AsRef, + { + let vars = vars + .into_iter() + .map(|(name, value)| (name.as_ref().to_owned(), value.as_ref().to_owned())) + .collect::>(); + let defaults = Self::default(); + let config = Self { + max_connections: parse_u32_setting( + &vars, + "POSTGRES_MAX_CONNECTIONS", + defaults.max_connections, + )?, + min_connections: parse_u32_setting( + &vars, + "POSTGRES_MIN_CONNECTIONS", + defaults.min_connections, + )?, + acquire_timeout_ms: parse_u64_setting( + &vars, + "POSTGRES_ACQUIRE_TIMEOUT_MS", + defaults.acquire_timeout_ms, + )?, + idle_timeout_ms: parse_u64_setting( + &vars, + "POSTGRES_IDLE_TIMEOUT_MS", + defaults.idle_timeout_ms, + )?, + max_lifetime_ms: parse_u64_setting( + &vars, + "POSTGRES_MAX_LIFETIME_MS", + defaults.max_lifetime_ms, + )?, + }; + + config.validate()?; + Ok(config) + } + + fn validate(self) -> Result { + if self.max_connections == 0 { + return Err(PostgresPoolConfigError::ZeroMaxConnections); + } + if self.min_connections > self.max_connections { + return Err(PostgresPoolConfigError::MinConnectionsExceedMax); + } + + Ok(self) + } +} + +fn parse_u32_setting( + vars: &BTreeMap, + name: &'static str, + default: u32, +) -> Result { + vars.get(name) + .map(|value| { + value + .parse::() + .map_err(|_| PostgresPoolConfigError::InvalidValue { + name, + value: value.clone(), + }) + }) + .transpose() + .map(|value| value.unwrap_or(default)) +} + +fn parse_u64_setting( + vars: &BTreeMap, + name: &'static str, + default: u64, +) -> Result { + vars.get(name) + .map(|value| { + value + .parse::() + .map_err(|_| PostgresPoolConfigError::InvalidValue { + name, + value: value.clone(), + }) + }) + .transpose() + .map(|value| value.unwrap_or(default)) +} + +#[cfg(test)] +mod pool_config_tests { + use super::{PostgresPoolConfig, PostgresPoolConfigError}; + + #[test] + fn pool_config_uses_explicit_defaults() { + let config = PostgresPoolConfig::from_vars(std::iter::empty::<(&str, &str)>()).unwrap(); + + assert_eq!( + config, + PostgresPoolConfig { + max_connections: 20, + min_connections: 2, + acquire_timeout_ms: 5_000, + idle_timeout_ms: 600_000, + max_lifetime_ms: 1_800_000, + } + ); + } + + #[test] + fn pool_config_parses_overrides() { + let config = PostgresPoolConfig::from_vars([ + ("POSTGRES_MAX_CONNECTIONS", "32"), + ("POSTGRES_MIN_CONNECTIONS", "4"), + ("POSTGRES_ACQUIRE_TIMEOUT_MS", "7000"), + ("POSTGRES_IDLE_TIMEOUT_MS", "900000"), + ("POSTGRES_MAX_LIFETIME_MS", "3600000"), + ]) + .unwrap(); + + assert_eq!( + config, + PostgresPoolConfig { + max_connections: 32, + min_connections: 4, + acquire_timeout_ms: 7_000, + idle_timeout_ms: 900_000, + max_lifetime_ms: 3_600_000, + } + ); + } + + #[test] + fn pool_config_rejects_invalid_numeric_value() { + let error = + PostgresPoolConfig::from_vars([("POSTGRES_MAX_CONNECTIONS", "abc")]).unwrap_err(); + + assert_eq!( + error, + PostgresPoolConfigError::InvalidValue { + name: "POSTGRES_MAX_CONNECTIONS", + value: "abc".to_owned(), + } + ); + } + + #[test] + fn pool_config_rejects_zero_max_connections() { + let error = PostgresPoolConfig::from_vars([("POSTGRES_MAX_CONNECTIONS", "0")]).unwrap_err(); + + assert_eq!(error, PostgresPoolConfigError::ZeroMaxConnections); + } + + #[test] + fn pool_config_rejects_min_connections_above_max() { + let error = PostgresPoolConfig::from_vars([ + ("POSTGRES_MAX_CONNECTIONS", "2"), + ("POSTGRES_MIN_CONNECTIONS", "3"), + ]) + .unwrap_err(); + + assert_eq!(error, PostgresPoolConfigError::MinConnectionsExceedMax); + } +} diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index f75d1fe..17f0252 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -31,6 +31,48 @@ pub struct RuntimeExecutor { metering_sink: SharedMeteringSink, } +#[derive(Clone, Copy)] +pub struct RuntimeExecutionRequest<'a> { + pub operation: &'a RuntimeOperation, + pub input: &'a Value, + pub resolved_auth: Option<&'a ResolvedAuth>, + pub request_context: Option<&'a RuntimeRequestContext>, +} + +impl<'a> RuntimeExecutionRequest<'a> { + pub fn new(operation: &'a RuntimeOperation, input: &'a Value) -> Self { + Self { + operation, + input, + resolved_auth: None, + request_context: None, + } + } + + pub fn with_auth(mut self, resolved_auth: &'a ResolvedAuth) -> Self { + self.resolved_auth = Some(resolved_auth); + self + } + + pub fn with_optional_auth(mut self, resolved_auth: Option<&'a ResolvedAuth>) -> Self { + self.resolved_auth = resolved_auth; + self + } + + pub fn with_context(mut self, request_context: &'a RuntimeRequestContext) -> Self { + self.request_context = Some(request_context); + self + } + + pub fn with_optional_context( + mut self, + request_context: Option<&'a RuntimeRequestContext>, + ) -> Self { + self.request_context = request_context; + self + } +} + impl Default for RuntimeExecutor { fn default() -> Self { Self::new() @@ -86,7 +128,7 @@ impl RuntimeExecutor { operation: &RuntimeOperation, input: &Value, ) -> Result { - self.execute_with_auth_and_context(operation, input, None, None) + self.execute_request(RuntimeExecutionRequest::new(operation, input)) .await } @@ -96,8 +138,10 @@ impl RuntimeExecutor { input: &Value, resolved_auth: Option<&ResolvedAuth>, ) -> Result { - self.execute_with_auth_and_context(operation, input, resolved_auth, None) - .await + self.execute_request( + RuntimeExecutionRequest::new(operation, input).with_optional_auth(resolved_auth), + ) + .await } pub async fn execute_with_context( @@ -106,8 +150,10 @@ impl RuntimeExecutor { input: &Value, request_context: Option<&RuntimeRequestContext>, ) -> Result { - self.execute_with_auth_and_context(operation, input, None, request_context) - .await + self.execute_request( + RuntimeExecutionRequest::new(operation, input).with_optional_context(request_context), + ) + .await } pub async fn execute_with_auth_and_context( @@ -117,16 +163,38 @@ impl RuntimeExecutor { resolved_auth: Option<&ResolvedAuth>, request_context: Option<&RuntimeRequestContext>, ) -> Result { - log_runtime_event("unary.execute", operation, request_context); - let _permit = self.acquire_unary_permit(operation)?; + self.execute_request( + RuntimeExecutionRequest::new(operation, input) + .with_optional_auth(resolved_auth) + .with_optional_context(request_context), + ) + .await + } + + pub async fn execute_request( + &self, + request: RuntimeExecutionRequest<'_>, + ) -> Result { + log_runtime_event("unary.execute", request.operation, request.request_context); + let _permit = self.acquire_unary_permit(request.operation)?; let started_at = Instant::now(); - let prepared_request = self.prepare_request(operation, input)?; - let prepared_request = apply_resolved_auth(prepared_request, resolved_auth); + let prepared_request = self.prepare_request(request.operation, request.input)?; + let prepared_request = apply_resolved_auth(prepared_request, request.resolved_auth); let result = self - .execute_prepared(operation, input, prepared_request, request_context) - .await; - self.record_metering(operation, request_context, &result, started_at) + .execute_prepared( + request.operation, + request.input, + prepared_request, + request.request_context, + ) .await; + self.record_metering( + request.operation, + request.request_context, + &result, + started_at, + ) + .await; result } diff --git a/crates/crank-runtime/src/lib.rs b/crates/crank-runtime/src/lib.rs index bb23a15..a9dd7a6 100644 --- a/crates/crank-runtime/src/lib.rs +++ b/crates/crank-runtime/src/lib.rs @@ -24,7 +24,7 @@ pub use cache_factory::{ BuiltinCacheBackendFactory, CacheBackendFactory, SharedCacheBackendFactory, }; pub use error::RuntimeError; -pub use executor::RuntimeExecutor; +pub use executor::{RuntimeExecutionRequest, RuntimeExecutor}; pub use executor_builder::{RuntimeExecutorBuilder, community_default}; pub use limits::{RuntimeLimits, RuntimeLimitsConfigError}; pub use model::{AdapterResponse, PreparedRequest, RuntimeOperation}; diff --git a/docs/architecture.md b/docs/architecture.md index 74a1b17..f5e5b26 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -39,6 +39,28 @@ MCP client Input mapping переводит MCP arguments в REST request. Output mapping переводит REST response в MCP tool result. +## Rust boundaries + +В Community-коде закреплены следующие границы: + +- `crank-core::domain` — доменные типы: operations, agents, users, secrets, observability, ids. +- `crank-core::ports` — интерфейсы внешних зависимостей: policy, audit, identity, protocol adapters, cache stores, metering. +- `crank-registry::records` — read/write records, которые возвращает storage layer. +- `crank-registry::requests` — request-структуры для persistence операций. +- `crank-registry::infrastructure` — PostgreSQL registry facade, pool config и extension migrations. +- `apps/admin-api/src/dto.rs` — HTTP payloads и view models. Service layer не должен владеть DTO-типами. +- `apps/admin-api/src/service/*` — application use-cases. Эти модули не импортируют `axum`. +- `crates/crank-community-mcp/src/transport.rs` — MCP Streamable HTTP transport: headers, accept negotiation, session id, JSON/SSE responses. +- `RuntimeExecutionRequest` — единая точка расширения runtime execution parameters. Новые auth/context/cache/metring параметры добавляются туда, а не через новые `execute_with_*` методы. + +Границы проверяются скриптами: + +```text +scripts/check-rust-boundaries.sh +scripts/check-rust-module-boundaries.sh +scripts/check-rust-code-health.sh +``` + ## Хранилище PostgreSQL хранит: diff --git a/docs/rust-code-health.md b/docs/rust-code-health.md index 3f43387..45a89c5 100644 --- a/docs/rust-code-health.md +++ b/docs/rust-code-health.md @@ -21,6 +21,8 @@ - `cargo clippy --workspace --all-targets --all-features -- -D warnings` — локальные ошибки, smell-и и часть API-антипаттернов. - `cargo test --workspace --all-targets` — unit и integration-style тесты. - `scripts/check-rust-code-health.sh` — локальный repo-level gate для размера Rust-файлов. +- `scripts/check-rust-boundaries.sh` — crate-level dependency direction и module-level import rules. +- `scripts/check-rust-module-boundaries.sh` — запрет на опасные imports внутри слоев. Что можно добавить позже: @@ -39,16 +41,21 @@ - выделять route/service/runtime helper в отдельный модуль; - оставлять рядом с production-кодом только короткие unit-тесты приватной логики. -Текущие крупные файлы: +Текущие крупные production-файлы считаются debt baseline. Они не должны расти, а при изменении их нужно постепенно разбирать: - `apps/admin-api/src/service.rs` -- `apps/admin-api/src/app.rs` -- `apps/mcp-server/src/main.rs` -- `crates/crank-runtime/src/executor.rs` - `crates/crank-registry/src/postgres/mod.rs` - `crates/crank-community-mcp/src/app.rs` - `crates/crank-registry/src/postgres/operation.rs` +Уже выделенные направления: + +- `apps/admin-api/src/dto.rs` содержит HTTP payloads и view models. +- `crates/crank-registry/src/postgres/pool_config.rs` содержит env parsing и validation pool settings. +- `crates/crank-registry/src/postgres/connection.rs` содержит connection/migration wiring. +- `crates/crank-community-mcp/src/transport.rs` содержит Streamable HTTP transport helpers. +- `crates/crank-runtime::RuntimeExecutionRequest` является основным способом передавать execution параметры. + ## Правила тестов Рядом с кодом допустимы: @@ -88,3 +95,11 @@ crank-core -> no app crates ``` Если потребуется строгая автоматическая проверка границ, добавим отдельный script на основе `cargo metadata` или `cargo-guppy`. + +Module-level правила уже проверяются: + +- `apps/admin-api/src/service*` не импортирует `axum`; +- `apps/admin-api/src/routes*` не импортирует `crank_runtime`; +- `crank-core/src` не импортирует framework/storage/network runtime crates; +- `crank-registry/src` не импортирует HTTP framework/client crates; +- `crank-runtime/src` не импортирует HTTP framework или SQL storage crates. diff --git a/scripts/check-rust-boundaries.sh b/scripts/check-rust-boundaries.sh index 3eea7f0..5d5ddb9 100755 --- a/scripts/check-rust-boundaries.sh +++ b/scripts/check-rust-boundaries.sh @@ -4,3 +4,4 @@ set -euo pipefail ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" python3 "$ROOT_DIR/scripts/check-rust-boundaries.py" "$@" +"$ROOT_DIR/scripts/check-rust-module-boundaries.sh" diff --git a/scripts/check-rust-module-boundaries.sh b/scripts/check-rust-module-boundaries.sh new file mode 100755 index 0000000..cb01f22 --- /dev/null +++ b/scripts/check-rust-module-boundaries.sh @@ -0,0 +1,65 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="${CRANK_RUST_BOUNDARY_ROOT:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}" +status=0 + +check_no_match() { + local description="$1" + local pattern="$2" + shift 2 + local output + + output="$(rg -n "$pattern" "$@" 2>/dev/null || true)" + if [[ -n "$output" ]]; then + echo "error: $description" >&2 + echo "$output" >&2 + status=1 + fi +} + +echo "Rust module boundary check: checking module-level imports" + +check_no_match \ + "admin-api service modules must not depend on axum HTTP types" \ + '^\s*use\s+axum(::|[;\{])' \ + "$ROOT_DIR/apps/admin-api/src/service" \ + "$ROOT_DIR/apps/admin-api/src/service.rs" + +check_no_match \ + "admin-api route modules must not call runtime directly" \ + '^\s*use\s+crank_runtime(::|[;\{])' \ + "$ROOT_DIR/apps/admin-api/src/routes" \ + "$ROOT_DIR/apps/admin-api/src/routes.rs" + +check_no_match \ + "crank-core production modules must not depend on axum, sqlx, reqwest, tokio or tracing" \ + '^\s*use\s+(axum|sqlx|reqwest|tokio|tracing)(::|[;\{])' \ + "$ROOT_DIR/crates/crank-core/src" + +check_no_match \ + "crank-registry production modules must not depend on axum or reqwest" \ + '^\s*use\s+(axum|reqwest)(::|[;\{])' \ + "$ROOT_DIR/crates/crank-registry/src" + +check_no_match \ + "crank-runtime production modules must not depend on axum or sqlx" \ + '^\s*use\s+(axum|sqlx)(::|[;\{])' \ + "$ROOT_DIR/crates/crank-runtime/src" + +if (( status != 0 )); then + cat >&2 <<'EOF' + +Rust module boundary check failed. + +Rules: +- routes own HTTP extraction/response details; +- service modules own application use-cases and must not import axum; +- core remains framework/storage agnostic; +- registry remains storage-only and HTTP-client agnostic; +- runtime remains execution-only and storage/framework agnostic. + +EOF +fi + +exit "$status"