From 1a4f0ea6f3b6e0352d545bba71835f49a83c37f1 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Tue, 7 Apr 2026 18:30:08 +0300 Subject: [PATCH] feat: expose streaming test run results --- TASKS.md | 10 +- apps/admin-api/src/app.rs | 270 +++++++++++++++- apps/admin-api/src/service.rs | 435 +++++++++++++++++++++++++- apps/ui/js/i18n.js | 16 + apps/ui/js/wizard.js | 50 ++- crates/crank-registry/src/postgres.rs | 4 +- docs/admin-api.md | 5 + docs/streaming-admin-api.md | 158 ++++++---- docs/streaming-ui-contract.md | 8 + 9 files changed, 865 insertions(+), 91 deletions(-) diff --git a/TASKS.md b/TASKS.md index dcfeea8..ada9d74 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,18 +2,18 @@ ## Current -### `feat/websocket-wizard-foundation` +### `feat/streaming-test-run-results` Status: completed DoD: -- wizard step 1 exposes WebSocket as a selectable protocol -- wizard step 3 collects WebSocket endpoint, subprotocols and subscribe/unsubscribe templates -- execution config serializes `ProtocolOptions.websocket` runtime tuning +- `POST /operations/{operation_id}/test-runs` returns mode-aware streaming metadata +- window test-runs expose bounded result flags for the wizard status block +- session and async-job test-runs create persisted runtime state and return references for follow-up pages ## Next -- `feat/live-authenticated-staging-entry` +- `feat/soap-test-run-polish` ## Backlog diff --git a/apps/admin-api/src/app.rs b/apps/admin-api/src/app.rs index f3b2711..78c300e 100644 --- a/apps/admin-api/src/app.rs +++ b/apps/admin-api/src/app.rs @@ -221,9 +221,10 @@ mod tests { use axum::{Json, Router, routing::post}; use crank_adapter_grpc::test_support as grpc_test_support; use crank_core::{ - DescriptorId, ExecutionConfig, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod, - MembershipRole, Protocol, RestTarget, SecretKind, SoapBindingStyle, SoapOperationMetadata, - SoapTarget, SoapVersion, Target, ToolDescription, WorkspaceId, + DescriptorId, ExecutionConfig, ExecutionMode, GraphqlOperationType, GraphqlTarget, + GrpcTarget, HttpMethod, MembershipRole, Protocol, RestTarget, SecretKind, SoapBindingStyle, + SoapOperationMetadata, SoapTarget, SoapVersion, Target, ToolDescription, TransportBehavior, + WorkspaceId, }; use crank_mapping::{MappingRule, MappingSet}; use crank_registry::PostgresRegistry; @@ -1441,6 +1442,177 @@ mod tests { assert_eq!(test_run["response_preview"]["message"], "hello"); } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn returns_window_metadata_for_streaming_test_runs() { + let registry = test_registry().await; + let storage_root = test_storage_root("grpc_window_test_run"); + let server_addr = grpc_test_support::spawn_unary_echo_server().await; + let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await; + let client = authorized_client(&base_url).await; + + let created = client + .post(format!("{base_url}/operations")) + .json(&test_grpc_window_operation_payload( + &server_addr, + "echo_window_runtime", + )) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let operation_id = created["operation_id"].as_str().unwrap().to_owned(); + + let test_run = client + .post(format!("{base_url}/operations/{operation_id}/test-runs")) + .json(&json!({ + "version": 1, + "input": { "message": "hello" } + })) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + + assert_eq!(test_run["ok"], true); + assert_eq!(test_run["mode"], "window"); + assert!(test_run["window"].is_object()); + assert!(test_run["window"]["window_complete"].is_boolean()); + assert!(test_run["window"]["truncated"].is_boolean()); + assert!(test_run["window"]["has_more"].is_boolean()); + assert!(test_run["response_preview"]["items"].is_array()); + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn starts_stream_session_from_streaming_test_runs() { + let registry = test_registry().await; + let storage_root = test_storage_root("grpc_session_test_run"); + let server_addr = grpc_test_support::spawn_unary_echo_server().await; + let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await; + let client = authorized_client(&base_url).await; + + let created = client + .post(format!("{base_url}/operations")) + .json(&test_grpc_session_operation_payload( + &server_addr, + "echo_session_runtime", + )) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let operation_id = created["operation_id"].as_str().unwrap().to_owned(); + + let test_run = client + .post(format!("{base_url}/operations/{operation_id}/test-runs")) + .json(&json!({ + "version": 1, + "input": { "message": "hello" } + })) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let session_id = test_run["stream_session"]["session_id"] + .as_str() + .unwrap() + .to_owned(); + + let session = client + .get(format!("{base_url}/stream-sessions/{session_id}")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + + assert_eq!(test_run["ok"], true); + assert_eq!(test_run["mode"], "session"); + assert_eq!(test_run["stream_session"]["status"], "running"); + assert!(test_run["response_preview"]["preview"]["items"].is_array()); + assert_eq!(session["id"], session_id); + assert_eq!(session["status"], "running"); + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn starts_async_job_from_streaming_test_runs() { + let registry = test_registry().await; + let storage_root = test_storage_root("async_job_test_run"); + let upstream_base_url = spawn_upstream_server().await; + let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await; + let client = authorized_client(&base_url).await; + + let created = client + .post(format!("{base_url}/operations")) + .json(&test_rest_async_job_operation_payload( + &upstream_base_url, + "crm_async_job_runtime", + )) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let operation_id = created["operation_id"].as_str().unwrap().to_owned(); + + let test_run = client + .post(format!("{base_url}/operations/{operation_id}/test-runs")) + .json(&json!({ + "version": 1, + "input": { "email": "user@example.com" } + })) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let job_id = test_run["async_job"]["job_id"].as_str().unwrap().to_owned(); + + let mut job = Value::Null; + for _ in 0..20 { + job = client + .get(format!("{base_url}/async-jobs/{job_id}")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + if job["status"] == "completed" { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + + let result = client + .get(format!("{base_url}/async-jobs/{job_id}/result")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + + assert_eq!(test_run["ok"], true); + assert_eq!(test_run["mode"], "async_job"); + assert_eq!(test_run["async_job"]["status"], "running"); + assert_eq!(job["status"], "completed"); + assert_eq!(result["id"], "lead_123"); + } + #[tokio::test(flavor = "multi_thread")] #[serial] async fn uploads_wsdl_and_tests_soap_operation() { @@ -2317,6 +2489,98 @@ mod tests { } } + fn test_grpc_window_operation_payload(server_addr: &str, name: &str) -> OperationPayload { + let mut payload = test_grpc_operation_payload(server_addr, name); + payload.display_name = "Server Echo Window".to_owned(); + payload.target = Target::Grpc(GrpcTarget { + server_addr: server_addr.to_owned(), + package: "echo".to_owned(), + service: "EchoService".to_owned(), + method: "ServerEcho".to_owned(), + descriptor_ref: DescriptorId::new("desc_echo_stream"), + descriptor_set_b64: grpc_test_support::descriptor_set_b64(), + }); + payload.execution_config.streaming = Some(crank_core::StreamingConfig { + mode: ExecutionMode::Window, + transport_behavior: TransportBehavior::ServerStream, + window_duration_ms: Some(100), + poll_interval_ms: None, + upstream_timeout_ms: Some(1_000), + idle_timeout_ms: None, + max_session_lifetime_ms: None, + max_items: Some(2), + max_bytes: None, + aggregation_mode: crank_core::AggregationMode::RawItems, + summary_path: None, + items_path: Some("$.response.body.items".to_owned()), + cursor_path: None, + status_path: None, + done_path: Some("$.response.body.done".to_owned()), + redacted_paths: Vec::new(), + truncate_item_fields: false, + max_field_length: None, + drop_duplicates: false, + sampling_rate: None, + tool_family: crank_core::ToolFamilyConfig::default(), + }); + payload + } + + fn test_grpc_session_operation_payload(server_addr: &str, name: &str) -> OperationPayload { + let mut payload = test_grpc_window_operation_payload(server_addr, name); + payload.display_name = "Server Echo Session".to_owned(); + if let Some(streaming) = payload.execution_config.streaming.as_mut() { + streaming.mode = ExecutionMode::Session; + streaming.poll_interval_ms = Some(1_000); + streaming.max_session_lifetime_ms = Some(60_000); + streaming.tool_family = crank_core::ToolFamilyConfig { + start_tool_name: Some("echo_session_start".to_owned()), + poll_tool_name: Some("echo_session_poll".to_owned()), + stop_tool_name: Some("echo_session_stop".to_owned()), + status_tool_name: None, + result_tool_name: None, + cancel_tool_name: None, + }; + } + payload + } + + fn test_rest_async_job_operation_payload(base_url: &str, name: &str) -> OperationPayload { + let mut payload = test_operation_payload(base_url, name); + payload.display_name = "Create Lead Async Job".to_owned(); + payload.execution_config.streaming = Some(crank_core::StreamingConfig { + mode: ExecutionMode::AsyncJob, + transport_behavior: TransportBehavior::DeferredResult, + window_duration_ms: None, + poll_interval_ms: Some(1_000), + upstream_timeout_ms: Some(1_000), + idle_timeout_ms: None, + max_session_lifetime_ms: Some(300_000), + max_items: None, + max_bytes: None, + aggregation_mode: crank_core::AggregationMode::SummaryPlusSamples, + summary_path: None, + items_path: None, + cursor_path: None, + status_path: None, + done_path: None, + redacted_paths: Vec::new(), + truncate_item_fields: false, + max_field_length: None, + drop_duplicates: false, + sampling_rate: None, + tool_family: crank_core::ToolFamilyConfig { + start_tool_name: Some("crm_async_start".to_owned()), + poll_tool_name: None, + stop_tool_name: None, + status_tool_name: Some("crm_async_status".to_owned()), + result_tool_name: Some("crm_async_result".to_owned()), + cancel_tool_name: Some("crm_async_cancel".to_owned()), + }, + }); + payload + } + fn test_soap_operation_payload(endpoint: &str, name: &str) -> OperationPayload { OperationPayload { name: name.to_owned(), diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 1b1e429..f69bad8 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -20,14 +20,15 @@ use crank_core::{ use crank_mapping::{JsonPathRoot, MappingSet, infer_mapping_from_samples}; use crank_proto::{ProtoService, services_from_descriptor_set_bytes}; use crank_registry::{ - AgentSummary, AgentVersionRecord, AsyncJobFilter, CreateAgentRequest, CreateInvitationRequest, - CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest, - CreateVersionRequest, CreateWorkspaceRequest, InvitationRecord, InvocationLogRecord, - ListInvocationLogsQuery, MembershipRecord, OperationAgentRef, OperationSampleMetadata, - OperationSummary, OperationUsageSummary, OperationVersionRecord, Page, PlatformApiKeyRecord, - PostgresRegistry, PublishAgentRequest, PublishRequest, RegistryError, RegistryOperation, - RotateSecretRequest, SampleKind, SaveAgentBindingsRequest, SaveAuthProfileRequest, - SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, StreamSessionFilter, + AgentSummary, AgentVersionRecord, AsyncJobFilter, CreateAgentRequest, CreateAsyncJobRequest, + CreateInvitationRequest, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, + CreateSecretRequest, CreateStreamSessionRequest, CreateVersionRequest, CreateWorkspaceRequest, + InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, + OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, + OperationVersionRecord, Page, PlatformApiKeyRecord, PostgresRegistry, PublishAgentRequest, + PublishRequest, RegistryError, RegistryOperation, RotateSecretRequest, SampleKind, + SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, + SaveSampleMetadataRequest, StreamSessionFilter, UpdateAsyncJobStatusRequest, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageBucket, UsageOperationBreakdown, UsageQuery, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord, }; @@ -128,9 +129,123 @@ pub struct TestRunPayload { #[derive(Clone, Debug, Serialize)] pub struct TestRunResult { pub ok: bool, + pub mode: ExecutionMode, pub request_preview: Value, pub response_preview: Value, pub errors: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub window: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub stream_session: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub async_job: Option, +} + +#[derive(Clone, Debug, Serialize)] +pub struct WindowTestRunView { + pub window_complete: bool, + pub truncated: bool, + pub has_more: bool, + pub cursor: Option, +} + +#[derive(Clone, Debug, Serialize)] +pub struct StreamSessionStartView { + pub session_id: String, + pub status: StreamStatus, + pub expires_at: String, + pub poll_after_ms: u64, + pub preview: Value, +} + +#[derive(Clone, Debug, Serialize)] +pub struct AsyncJobStartView { + pub job_id: String, + pub status: JobStatus, + pub progress: Value, +} + +#[derive(Debug, Serialize, Deserialize)] +struct StoredSessionState { + input: Value, + summary: Value, + items: Vec, + next_index: usize, + batch_size: usize, +} + +enum TestRunOutcome { + Unary { + output: Value, + }, + Window { + output: crank_runtime::WindowExecutionResult, + }, + Session { + output: StreamSessionStartView, + }, + AsyncJob { + output: AsyncJobStartView, + }, +} + +impl TestRunOutcome { + fn into_result_views( + self, + ) -> ( + &'static str, + Value, + Option, + Option, + Option, + ) { + match self { + Self::Unary { output } => ("admin test run completed", output, None, None, None), + Self::Window { output } => ( + "admin window test run completed", + json!({ + "summary": output.summary, + "items": output.items, + "cursor": output.cursor, + "window_complete": output.window_complete, + "truncated": output.truncated, + "has_more": output.has_more, + }), + Some(WindowTestRunView { + window_complete: output.window_complete, + truncated: output.truncated, + has_more: output.has_more, + cursor: output.cursor, + }), + None, + None, + ), + Self::Session { output } => ( + "admin session test run started", + json!({ + "session_id": output.session_id, + "status": output.status, + "expires_at": output.expires_at, + "poll_after_ms": output.poll_after_ms, + "preview": output.preview, + }), + None, + Some(output), + None, + ), + Self::AsyncJob { output } => ( + "admin async job test run started", + json!({ + "job_id": output.job_id, + "status": output.status, + "progress": output.progress, + }), + None, + None, + Some(output), + ), + } + } } #[derive(Clone, Debug, Deserialize)] @@ -1889,6 +2004,12 @@ impl AdminService { .get_operation_version(workspace_id, operation_id, payload.version) .await?; let runtime = RuntimeOperation::from(record.snapshot.clone()); + let mode = runtime + .execution_config + .streaming + .as_ref() + .map(|streaming| streaming.mode) + .unwrap_or(ExecutionMode::Unary); let request_preview = match build_request_preview(&record.snapshot.input_mapping, &payload.input) { Ok(preview) => preview, @@ -1910,11 +2031,15 @@ impl AdminService { .await?; return Ok(TestRunResult { ok: false, + mode, request_preview: Value::Null, response_preview: Value::Null, errors: vec![crate::error::runtime_test_failure(&RuntimeError::Mapping( error, ))], + window: None, + stream_session: None, + async_job: None, }); } }; @@ -1924,16 +2049,39 @@ impl AdminService { .await; let started_at = std::time::Instant::now(); match match resolved_auth { - Ok(resolved_auth) => { - self.runtime + Ok(resolved_auth) => match mode { + ExecutionMode::Unary => self + .runtime .execute_with_auth(&runtime, &payload.input, resolved_auth.as_ref()) .await - } + .map(|output| TestRunOutcome::Unary { output }), + ExecutionMode::Window => self + .runtime + .execute_window_with_auth(&runtime, &payload.input, resolved_auth.as_ref()) + .await + .map(|output| TestRunOutcome::Window { output }), + ExecutionMode::Session => self + .start_stream_session_test( + workspace_id, + &record.snapshot, + &runtime, + &payload.input, + resolved_auth.as_ref(), + ) + .await + .map(|output| TestRunOutcome::Session { output }), + ExecutionMode::AsyncJob => self + .start_async_job_test(workspace_id, &record.snapshot, &runtime, &payload.input) + .await + .map(|output| TestRunOutcome::AsyncJob { output }), + }, Err(error) => Err(error), } { - Ok(output) => { + Ok(outcome) => { let duration_ms = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + let (message, response_preview, window, stream_session, async_job) = + outcome.into_result_views(); self.record_invocation(InvocationRecordRequest { workspace_id, agent_id: None, @@ -1941,19 +2089,23 @@ impl AdminService { source: InvocationSource::AdminTestRun, level: InvocationLevel::Info, status: InvocationStatus::Ok, - message: "admin test run completed".to_owned(), + message: message.to_owned(), status_code: None, error_kind: None, duration_ms, request_preview: request_preview.clone(), - response_preview: output.clone(), + response_preview: response_preview.clone(), }) .await?; Ok(TestRunResult { ok: true, + mode, request_preview, - response_preview: output, + response_preview, errors: Vec::new(), + window, + stream_session, + async_job, }) } Err(error) => { @@ -1976,9 +2128,13 @@ impl AdminService { .await?; Ok(TestRunResult { ok: false, + mode, request_preview, response_preview: Value::Null, errors: vec![crate::error::runtime_test_failure(&error)], + window: None, + stream_session: None, + async_job: None, }) } } @@ -2009,6 +2165,187 @@ impl AdminService { .map(Some) } + async fn start_stream_session_test( + &self, + workspace_id: &WorkspaceId, + operation: &RegistryOperation, + runtime: &RuntimeOperation, + input: &Value, + resolved_auth: Option<&ResolvedAuth>, + ) -> Result { + let streaming = runtime.execution_config.streaming.as_ref().ok_or_else(|| { + RuntimeError::MissingStreamingConfig { + operation_id: runtime.operation_id.as_str().to_owned(), + } + })?; + let seed = self + .runtime + .execute_session_seed_with_auth(runtime, input, resolved_auth) + .await?; + let batch_size = streaming.max_items.unwrap_or(10).max(1) as usize; + let preview_count = seed.items.len().min(batch_size); + let preview_items = seed.items[..preview_count].to_vec(); + let next_index = preview_count; + let created_at = now_string().map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + let expires_at = add_millis( + &created_at, + streaming.max_session_lifetime_ms.unwrap_or(60_000), + ) + .map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + let session = StreamSession { + id: crank_core::StreamSessionId::new(new_prefixed_id("sess")), + workspace_id: workspace_id.clone(), + agent_id: None, + operation_id: operation.id.clone(), + protocol: operation.protocol, + mode: ExecutionMode::Session, + status: StreamStatus::Running, + cursor: (next_index < seed.items.len()).then(|| json!(next_index)), + state: json!(StoredSessionState { + input: input.clone(), + summary: seed.summary.clone(), + items: seed.items.clone(), + next_index, + batch_size, + }), + expires_at: expires_at.clone(), + last_poll_at: Some(created_at.clone()), + created_at, + closed_at: None, + }; + self.registry + .create_stream_session(CreateStreamSessionRequest { session: &session }) + .await + .map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + + Ok(StreamSessionStartView { + session_id: session.id.as_str().to_owned(), + status: session.status, + expires_at, + poll_after_ms: streaming.poll_interval_ms.unwrap_or(1_000), + preview: json!({ + "summary": seed.summary, + "items": preview_items, + }), + }) + } + + async fn start_async_job_test( + &self, + workspace_id: &WorkspaceId, + operation: &RegistryOperation, + runtime: &RuntimeOperation, + input: &Value, + ) -> Result { + let created_at = now_string().map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + let expires_at = + add_millis(&created_at, 300_000).map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + let job = AsyncJobHandle { + id: crank_core::AsyncJobId::new(new_prefixed_id("job")), + workspace_id: workspace_id.clone(), + agent_id: None, + operation_id: operation.id.clone(), + status: JobStatus::Running, + progress: json!({ "pct": 0 }), + result: None, + error: None, + expires_at: Some(expires_at), + created_at: created_at.clone(), + updated_at: created_at.clone(), + finished_at: None, + }; + self.registry + .create_async_job(CreateAsyncJobRequest { job: &job }) + .await + .map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + + let registry = self.registry.clone(); + let secret_crypto = self.secret_crypto.clone(); + let runtime_for_task = self.runtime.clone(); + let workspace_for_task = workspace_id.clone(); + let operation_for_task = runtime.clone(); + let input_for_task = input.clone(); + let job_id = job.id.clone(); + tokio::spawn(async move { + let resolved_auth = resolve_runtime_auth_for_task( + ®istry, + &secret_crypto, + &workspace_for_task, + &operation_for_task.execution_config, + ) + .await; + let result = match resolved_auth { + Ok(resolved_auth) => { + runtime_for_task + .execute_with_auth( + &operation_for_task, + &input_for_task, + resolved_auth.as_ref(), + ) + .await + } + Err(error) => Err(error), + }; + let finished_at = match now_string() { + Ok(value) => value, + Err(_) => return, + }; + let _ = match result { + Ok(output) => { + registry + .update_async_job_status(UpdateAsyncJobStatusRequest { + job_id: &job_id, + current_status: JobStatus::Running, + next_status: JobStatus::Completed, + progress: &json!({ "pct": 100 }), + result: Some(&output), + error: None, + expires_at: None, + updated_at: &finished_at, + finished_at: Some(&finished_at), + }) + .await + } + Err(error) => { + registry + .update_async_job_status(UpdateAsyncJobStatusRequest { + job_id: &job_id, + current_status: JobStatus::Running, + next_status: JobStatus::Failed, + progress: &json!({ "pct": 100 }), + result: None, + error: Some(&json!({ + "code": runtime_error_code(&error), + "message": error.to_string(), + })), + expires_at: None, + updated_at: &finished_at, + finished_at: Some(&finished_at), + }) + .await + } + }; + }); + + Ok(AsyncJobStartView { + job_id: job.id.as_str().to_owned(), + status: job.status, + progress: job.progress, + }) + } + async fn resolve_auth_profile( &self, workspace_id: &WorkspaceId, @@ -4267,6 +4604,74 @@ fn now_string() -> Result { .map_err(|error| ApiError::internal(error.to_string())) } +fn add_millis(timestamp: &str, millis: u64) -> Result { + let parsed = OffsetDateTime::parse(timestamp, &Rfc3339) + .map_err(|error| ApiError::internal(error.to_string()))?; + parsed + .checked_add(time::Duration::milliseconds(millis as i64)) + .ok_or_else(|| ApiError::internal("failed to add millis to timestamp"))? + .format(&Rfc3339) + .map_err(|error| ApiError::internal(error.to_string())) +} + +async fn resolve_runtime_auth_for_task( + registry: &PostgresRegistry, + secret_crypto: &SecretCrypto, + workspace_id: &WorkspaceId, + execution_config: &crank_core::ExecutionConfig, +) -> Result, RuntimeError> { + let Some(auth_profile_id) = execution_config.auth_profile_ref.as_ref() else { + return Ok(None); + }; + + let auth_profile = registry + .get_auth_profile(workspace_id, auth_profile_id) + .await + .map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })? + .ok_or_else(|| RuntimeError::MissingAuthProfile { + auth_profile_id: auth_profile_id.as_str().to_owned(), + })?; + + let mut secrets = BTreeMap::new(); + let used_at = now_string().map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + + for secret_id in auth_profile.config.secret_ids() { + let secret = registry + .get_secret(workspace_id, secret_id) + .await + .map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })? + .ok_or_else(|| RuntimeError::MissingSecret { + secret_id: secret_id.as_str().to_owned(), + })?; + let version = registry + .get_current_secret_version(workspace_id, secret_id) + .await + .map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })? + .ok_or_else(|| RuntimeError::MissingSecretVersion { + secret_id: secret_id.as_str().to_owned(), + version: secret.secret.current_version, + })?; + let plaintext = secret_crypto.decrypt(&version.secret_version.ciphertext)?; + registry + .touch_secret(workspace_id, secret_id, &used_at) + .await + .map_err(|error| RuntimeError::SecretCrypto { + details: error.to_string(), + })?; + secrets.insert(secret_id.clone(), plaintext); + } + + ResolvedAuth::from_profile(&auth_profile, &secrets).map(Some) +} + fn default_invitation_expiry() -> Result { OffsetDateTime::now_utc() .checked_add(time::Duration::days(7)) diff --git a/apps/ui/js/i18n.js b/apps/ui/js/i18n.js index 90c8579..2daa5e0 100644 --- a/apps/ui/js/i18n.js +++ b/apps/ui/js/i18n.js @@ -831,10 +831,18 @@ var TRANSLATIONS = { 'wizard.test.failed': 'Test run returned errors', 'wizard.test.completed_body': 'The current draft executed successfully against the upstream.', 'wizard.test.failed_body': 'The draft executed, but the backend returned validation or runtime errors.', + 'wizard.test.window_completed': 'Window test completed', + 'wizard.test.window_completed_body': 'Window complete: {window_complete}. Truncated: {truncated}. Has more: {has_more}.', + 'wizard.test.session_started': 'Session test started', + 'wizard.test.session_started_body': 'Session {session_id} was created. Poll after {poll_after_ms} ms from the stream sessions view.', + 'wizard.test.async_job_started': 'Async job test started', + 'wizard.test.async_job_started_body': 'Async job {job_id} was created. Track status and result from the async jobs view.', 'wizard.test.no_response': 'No test response yet', 'wizard.test.no_response_body': 'Run a test first, then copy the response preview into the output sample editor.', 'wizard.test.response_copied': 'Response copied', 'wizard.test.response_copied_body': 'The latest test response was copied into the output sample editor.', + 'wizard.boolean.yes': 'yes', + 'wizard.boolean.no': 'no', 'wizard.yaml.exported': 'YAML exported', 'wizard.yaml.exported_body': 'The current draft YAML was downloaded from the backend.', 'wizard.yaml.none': 'No YAML provided', @@ -1813,10 +1821,18 @@ var TRANSLATIONS = { 'wizard.test.failed': 'Тестовый прогон вернул ошибки', 'wizard.test.completed_body': 'Текущий черновик успешно выполнился против upstream.', 'wizard.test.failed_body': 'Черновик выполнился, но backend вернул ошибки валидации или runtime.', + 'wizard.test.window_completed': 'Оконный тест завершен', + 'wizard.test.window_completed_body': 'Окно завершено: {window_complete}. Усечено: {truncated}. Есть продолжение: {has_more}.', + 'wizard.test.session_started': 'Session-тест запущен', + 'wizard.test.session_started_body': 'Создана session {session_id}. Следующий poll можно делать через {poll_after_ms} мс из представления stream sessions.', + 'wizard.test.async_job_started': 'Async job тест запущен', + 'wizard.test.async_job_started_body': 'Создан async job {job_id}. Отслеживайте статус и результат через представление async jobs.', 'wizard.test.no_response': 'Тестового ответа пока нет', 'wizard.test.no_response_body': 'Сначала выполните тест, затем скопируйте response preview в editor output sample.', 'wizard.test.response_copied': 'Ответ скопирован', 'wizard.test.response_copied_body': 'Последний тестовый ответ скопирован в editor output sample.', + 'wizard.boolean.yes': 'да', + 'wizard.boolean.no': 'нет', 'wizard.yaml.exported': 'YAML экспортирован', 'wizard.yaml.exported_body': 'YAML текущего черновика скачан из backend.', 'wizard.yaml.none': 'YAML не предоставлен', diff --git a/apps/ui/js/wizard.js b/apps/ui/js/wizard.js index 76a2668..30d1f9d 100644 --- a/apps/ui/js/wizard.js +++ b/apps/ui/js/wizard.js @@ -2546,6 +2546,47 @@ function setTextareaValue(id, value) { element.value = safeStringify(value); } +function describeWizardTestResult(result) { + if (result.stream_session && result.stream_session.session_id) { + return { + title: tKey('wizard.test.session_started'), + body: tfKey('wizard.test.session_started_body', { + session_id: result.stream_session.session_id, + poll_after_ms: result.stream_session.poll_after_ms + }), + isError: false, + }; + } + + if (result.async_job && result.async_job.job_id) { + return { + title: tKey('wizard.test.async_job_started'), + body: tfKey('wizard.test.async_job_started_body', { + job_id: result.async_job.job_id + }), + isError: false, + }; + } + + if (result.window) { + return { + title: result.ok ? tKey('wizard.test.window_completed') : tKey('wizard.test.failed'), + body: tfKey('wizard.test.window_completed_body', { + truncated: result.window.truncated ? tKey('wizard.boolean.yes') : tKey('wizard.boolean.no'), + has_more: result.window.has_more ? tKey('wizard.boolean.yes') : tKey('wizard.boolean.no'), + window_complete: result.window.window_complete ? tKey('wizard.boolean.yes') : tKey('wizard.boolean.no') + }), + isError: !result.ok, + }; + } + + return { + title: result.ok ? tKey('wizard.test.completed') : tKey('wizard.test.failed'), + body: result.ok ? tKey('wizard.test.completed_body') : tKey('wizard.test.failed_body'), + isError: !result.ok, + }; +} + async function uploadInputSampleFromWizard() { await persistCurrentDraft(true); await window.CrankApi.uploadInputSample( @@ -2586,12 +2627,11 @@ async function runWizardTest() { setTextareaValue('wizard-test-request-preview', result.request_preview); setTextareaValue('wizard-test-response-preview', result.response_preview); setTextareaValue('wizard-test-errors', result.errors && result.errors.length ? result.errors : []); + var status = describeWizardTestResult(result); showWizardLiveStatus( - result.ok ? tKey('wizard.test.completed') : tKey('wizard.test.failed'), - result.ok - ? tKey('wizard.test.completed_body') - : tKey('wizard.test.failed_body'), - !result.ok + status.title, + status.body, + status.isError ); } diff --git a/crates/crank-registry/src/postgres.rs b/crates/crank-registry/src/postgres.rs index 8cbe1ca..cac5446 100644 --- a/crates/crank-registry/src/postgres.rs +++ b/crates/crank-registry/src/postgres.rs @@ -3872,7 +3872,9 @@ fn map_secret_version_record(row: &PgRow) -> Result, _>("created_by")?.map(UserId::new), + created_by: row + .try_get::, _>("created_by")? + .map(UserId::new), }, }) } diff --git a/docs/admin-api.md b/docs/admin-api.md index 4e86c14..e3cf3e3 100644 --- a/docs/admin-api.md +++ b/docs/admin-api.md @@ -127,6 +127,11 @@ - `POST /descriptors/wsdl` принимает raw WSDL payload и сохраняет inspection metadata для SOAP; - `POST /descriptors/xsd` принимает supporting XSD artifacts для SOAP operation; - `GET /soap/services` возвращает нормализованный список `service -> port -> operation` из последнего WSDL текущего draft version. +- `POST /operations/{operation_id}/test-runs` остается единым test-run endpoint для `unary`, `window`, `session` и `async_job` execution modes; +- test-run response всегда возвращает `mode`, `request_preview`, `response_preview`, `errors`, а для streaming modes дополняется: + - `window` с `window_complete`, `truncated`, `has_more`, `cursor`; + - `stream_session` с `session_id`, `status`, `expires_at`, `poll_after_ms`, `preview`; + - `async_job` с `job_id`, `status`, `progress`. ### 5.5. Upstream auth profiles diff --git a/docs/streaming-admin-api.md b/docs/streaming-admin-api.md index 45a9496..9045e72 100644 --- a/docs/streaming-admin-api.md +++ b/docs/streaming-admin-api.md @@ -264,11 +264,11 @@ ## 8. Streaming operation test-runs -### `POST /api/admin/workspaces/{workspace_id}/operations/{operation_id}/stream-test-runs` +### `POST /api/admin/workspaces/{workspace_id}/operations/{operation_id}/test-runs` Назначение: -- выполнить `window`, `session` или `async_job` test-run для draft version; +- выполнить `unary`, `window`, `session` или `async_job` test-run для draft version; - показать оператору runtime behavior до publish. Тело: @@ -280,30 +280,57 @@ "service": "billing", "level": "error" }, - "test_mode": "window", - "overrides": { - "streaming": { - "window_duration_ms": 3000, - "max_items": 50 - } + "version": 4, + "input": { + "service": "billing", + "level": "error" } } ``` +Успех для `unary`: + +```json +{ + "ok": true, + "mode": "unary", + "request_preview": { + "service": "billing", + "level": "error" + }, + "response_preview": { + "summary": "completed" + }, + "errors": [], + "window": null, + "stream_session": null, + "async_job": null +} +``` + Успех для `window`: ```json { - "run_id": "strun_01j0test", + "ok": true, "mode": "window", - "status": "completed", - "window_complete": true, - "truncated": false, - "has_more": false, - "summary": {}, - "items": [], - "cursor": null, - "duration_ms": 2871 + "request_preview": { + "service": "billing", + "level": "error" + }, + "response_preview": { + "summary": {}, + "items": [] + }, + "errors": [], + "window": { + "window_complete": true, + "truncated": false, + "has_more": false, + "cursor": null + }, + "stream_session": null, + "async_job": null } ``` @@ -311,18 +338,29 @@ ```json { - "run_id": "strun_01j0test", + "ok": true, "mode": "session", - "status": "running", - "session": { - "session_id": "sess_01j0stream", - "expires_at": "2026-04-06T12:05:00Z", - "poll_after_ms": 2000 + "request_preview": { + "service": "billing", + "level": "error" }, - "preview": { + "response_preview": { "summary": {}, "items": [] - } + }, + "errors": [], + "window": null, + "stream_session": { + "session_id": "sess_01j0stream", + "status": "running", + "expires_at": "2026-04-06T12:05:00Z", + "poll_after_ms": 2000, + "preview": { + "summary": {}, + "items": [] + } + }, + "async_job": null } ``` @@ -330,10 +368,19 @@ ```json { - "run_id": "strun_01j0test", + "ok": true, "mode": "async_job", - "status": "running", - "job": { + "request_preview": { + "service": "billing", + "level": "error" + }, + "response_preview": { + "job": "started" + }, + "errors": [], + "window": null, + "stream_session": null, + "async_job": { "job_id": "job_01j0deploy", "status": "running", "progress": { @@ -343,47 +390,34 @@ } ``` -### `POST /api/admin/workspaces/{workspace_id}/operations/{operation_id}/stream-test-runs/{run_id}/poll` - -Назначение: - -- продолжить session-oriented test-run. - -Тело: +Ошибка: ```json { - "session_id": "sess_01j0stream" + "ok": false, + "mode": "window", + "request_preview": { + "service": "billing", + "level": "error" + }, + "response_preview": null, + "errors": [ + { + "code": "runtime_test_failure", + "message": "upstream timeout" + } + ], + "window": null, + "stream_session": null, + "async_job": null } ``` -Ответ: +Примечания: -```json -{ - "run_id": "strun_01j0test", - "mode": "session", - "status": "running", - "window_complete": false, - "truncated": false, - "has_more": true, - "summary": {}, - "items": [], - "cursor": "next_cursor" -} -``` - -### `POST /api/admin/workspaces/{workspace_id}/operations/{operation_id}/stream-test-runs/{run_id}/stop` - -Назначение: - -- остановить test session. - -### `GET /api/admin/workspaces/{workspace_id}/operations/{operation_id}/stream-test-runs/{run_id}/result` - -Назначение: - -- получить финальный результат async-job test-run. +- endpoint не использует отдельные `stream-test-runs/*` subresources; +- session и async-job test-runs создают обычные runtime resources в `stream_sessions` и `async_jobs`; +- дальнейшее наблюдение идет через `GET /stream-sessions`, `GET /stream-sessions/{session_id}`, `POST /stream-sessions/{session_id}/stop`, `GET /async-jobs`, `GET /async-jobs/{job_id}`, `GET /async-jobs/{job_id}/result`, `POST /async-jobs/{job_id}/cancel`. ## 9. Runtime session resources diff --git a/docs/streaming-ui-contract.md b/docs/streaming-ui-contract.md index 4c6e9aa..f0001cf 100644 --- a/docs/streaming-ui-contract.md +++ b/docs/streaming-ui-contract.md @@ -27,6 +27,14 @@ - `Stream Sessions` - `Async Jobs` +Для test-run UX: + +- wizard status block должен различать `unary`, `window`, `session`, `async_job`; +- `window` показывает bounded flags `window_complete`, `truncated`, `has_more`; +- `session` показывает `session_id`, `poll_after_ms` и переводит оператора к странице `Stream Sessions`; +- `async_job` показывает `job_id` и переводит оператора к странице `Async Jobs`; +- request/response preview textareas остаются общими для всех режимов. + ## 3. Wizard information architecture ## 3.1. Shared wizard structure