From 8ddcb213a35de27662c23ff31442eeaf7b571f9b Mon Sep 17 00:00:00 2001 From: github-ops Date: Thu, 14 May 2026 18:28:19 +0000 Subject: [PATCH] runtime: switch executor to adapter registry --- TASKS.md | 1 + apps/admin-api/src/error.rs | 1 + apps/admin-api/src/service.rs | 1 + apps/mcp-server/src/app.rs | 1 + crates/crank-adapter-rest/src/lib.rs | 4 +- crates/crank-core/src/ext/protocol.rs | 2 + crates/crank-runtime/Cargo.toml | 4 - crates/crank-runtime/src/error.rs | 39 +- crates/crank-runtime/src/executor.rs | 425 ++++++-------------- crates/crank-runtime/src/model.rs | 41 ++ crates/crank-runtime/src/request_context.rs | 28 ++ crates/crank-runtime/src/streaming.rs | 13 + 12 files changed, 226 insertions(+), 334 deletions(-) diff --git a/TASKS.md b/TASKS.md index 1658d58..12bee8d 100644 --- a/TASKS.md +++ b/TASKS.md @@ -151,5 +151,6 @@ Progress: - Phase 0 / task `0.7`: введены `CapabilityProfile` и `CommunityCapabilityProfile`, а `admin-api` capability payload теперь собирается через public seam вместо локального literal - Phase 0 / task `0.8`: введены `ProtocolAdapter`, `ProtocolAdapterError`, `AdapterRegistry` и базовые transport-типы (`PreparedRequest`, `AdapterResponse`, `WindowExecutionResult`, `RuntimeRequestContext`) в `crank_core::ext::protocol` - Phase 0 / task `0.9`: `crank-adapter-rest::RestAdapter` реализует новый `ProtocolAdapter` contract, а seam дополнен явным `Target` и window parameters для реального adapter wiring + - Phase 0 / task `0.10`: `RuntimeExecutor` переведен с hardcoded adapter fields на `AdapterRegistry`, `crank-runtime` больше не держит protocol feature flags, а общий `PreparedRequest` seam дополнен `timeout_ms` для корректного runtime dispatch - Phase 0 / task `0.13`: введены `RegistryExtension`, `ExtensionMigration` и `apply_extension_migrations(...)` в `crank-registry` как отдельный public seam для additive private migrations - backward-compatible alias `CommunityMachineCredentialVerifier` сохранен, поведение Community не изменено diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index ad9b492..ed8e207 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -423,6 +423,7 @@ fn runtime_test_failure_code(error: &RuntimeError) -> &'static str { RuntimeError::GraphqlAdapter(_) => "runtime_graphql_error", RuntimeError::GrpcAdapter(_) => "runtime_grpc_error", RuntimeError::RestAdapter(_) => "runtime_rest_error", + RuntimeError::ProtocolAdapter(_) => "runtime_adapter_error", RuntimeError::SoapAdapter(_) => "runtime_soap_error", RuntimeError::WebsocketAdapter(_) => "runtime_websocket_error", RuntimeError::UnsupportedProtocol { .. } => "runtime_protocol_error", diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index a0a03c0..297ca14 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -4031,6 +4031,7 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str { RuntimeError::GraphqlAdapter(_) => "graphql_error", RuntimeError::GrpcAdapter(_) => "grpc_error", RuntimeError::RestAdapter(_) => "rest_error", + RuntimeError::ProtocolAdapter(_) => "adapter_error", RuntimeError::SoapAdapter(_) => "soap_error", RuntimeError::WebsocketAdapter(_) => "websocket_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs index 52df5fe..22a98ff 100644 --- a/apps/mcp-server/src/app.rs +++ b/apps/mcp-server/src/app.rs @@ -2378,6 +2378,7 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str { RuntimeError::GraphqlAdapter(_) => "adapter_execution_error", RuntimeError::GrpcAdapter(_) => "adapter_execution_error", RuntimeError::RestAdapter(_) => "adapter_execution_error", + RuntimeError::ProtocolAdapter(_) => "adapter_execution_error", RuntimeError::SoapAdapter(_) => "adapter_execution_error", RuntimeError::WebsocketAdapter(_) => "adapter_execution_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", diff --git a/crates/crank-adapter-rest/src/lib.rs b/crates/crank-adapter-rest/src/lib.rs index 6427263..14e386b 100644 --- a/crates/crank-adapter-rest/src/lib.rs +++ b/crates/crank-adapter-rest/src/lib.rs @@ -36,7 +36,7 @@ impl ProtocolAdapter for RestAdapter { query_params: prepared.query_params.clone(), headers: prepared.headers.clone(), body: prepared.body.clone(), - timeout_ms: 30_000, + timeout_ms: prepared.timeout_ms, }; let response = self.execute(target, &request).await?; @@ -63,7 +63,7 @@ impl ProtocolAdapter for RestAdapter { query_params: prepared.query_params.clone(), headers: prepared.headers.clone(), body: prepared.body.clone(), - timeout_ms: 30_000, + timeout_ms: prepared.timeout_ms, }, window_duration_ms, max_items: max_items.map(|value| value.saturating_add(1)), diff --git a/crates/crank-core/src/ext/protocol.rs b/crates/crank-core/src/ext/protocol.rs index 144988f..b7688ff 100644 --- a/crates/crank-core/src/ext/protocol.rs +++ b/crates/crank-core/src/ext/protocol.rs @@ -71,6 +71,8 @@ pub struct PreparedRequest { pub variables: Option, #[serde(skip_serializing_if = "Option::is_none")] pub body: Option, + #[serde(default)] + pub timeout_ms: u64, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] diff --git a/crates/crank-runtime/Cargo.toml b/crates/crank-runtime/Cargo.toml index e3e9ac9..1ae22e5 100644 --- a/crates/crank-runtime/Cargo.toml +++ b/crates/crank-runtime/Cargo.toml @@ -10,10 +10,6 @@ path = "src/lib.rs" [features] default = [] -protocol-graphql = [] -protocol-grpc = [] -protocol-soap = [] -protocol-websocket = [] [dependencies] aes-gcm.workspace = true diff --git a/crates/crank-runtime/src/error.rs b/crates/crank-runtime/src/error.rs index f0dfcea..e404038 100644 --- a/crates/crank-runtime/src/error.rs +++ b/crates/crank-runtime/src/error.rs @@ -4,15 +4,6 @@ use crank_mapping::MappingError; use crank_schema::SchemaError; use thiserror::Error; -#[cfg(feature = "protocol-graphql")] -use crank_adapter_graphql::GraphqlAdapterError; -#[cfg(feature = "protocol-grpc")] -use crank_adapter_grpc::GrpcAdapterError; -#[cfg(feature = "protocol-soap")] -use crank_adapter_soap::SoapAdapterError; -#[cfg(feature = "protocol-websocket")] -use crank_adapter_websocket::WebsocketAdapterError; - #[derive(Debug, Error)] pub enum RuntimeError { #[error(transparent)] @@ -26,6 +17,8 @@ pub enum RuntimeError { #[error(transparent)] RestAdapter(#[from] RestAdapterError), #[error("{0}")] + ProtocolAdapter(String), + #[error("{0}")] SoapAdapter(String), #[error("{0}")] WebsocketAdapter(String), @@ -58,31 +51,3 @@ pub enum RuntimeError { details: String, }, } - -#[cfg(feature = "protocol-graphql")] -impl From for RuntimeError { - fn from(value: GraphqlAdapterError) -> Self { - Self::GraphqlAdapter(value.to_string()) - } -} - -#[cfg(feature = "protocol-grpc")] -impl From for RuntimeError { - fn from(value: GrpcAdapterError) -> Self { - Self::GrpcAdapter(value.to_string()) - } -} - -#[cfg(feature = "protocol-soap")] -impl From for RuntimeError { - fn from(value: SoapAdapterError) -> Self { - Self::SoapAdapter(value.to_string()) - } -} - -#[cfg(feature = "protocol-websocket")] -impl From for RuntimeError { - fn from(value: WebsocketAdapterError) -> Self { - Self::WebsocketAdapter(value.to_string()) - } -} diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index df073db..ba5c0d0 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -3,10 +3,10 @@ use std::sync::Arc; use std::time::Duration; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; -use crank_adapter_rest::{RestAdapter, RestRequest, RestWindowRequest}; +use crank_adapter_rest::RestAdapter; use crank_core::{ - CachedHeader, CachedResponse, ExecutionMode, HttpMethod, ResponseCacheStore, Target, - TransportBehavior, + AdapterRegistry, CachedHeader, CachedResponse, ExecutionMode, HttpMethod, ResponseCacheStore, + SharedProtocolAdapter, Target, TransportBehavior, }; use serde_json::{Map, Value, json}; use sha2::{Digest, Sha256}; @@ -18,26 +18,9 @@ use crate::{ RuntimeRequestContext, WindowExecutionResult, }; -#[cfg(feature = "protocol-graphql")] -use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest}; -#[cfg(feature = "protocol-grpc")] -use crank_adapter_grpc::{GrpcAdapter, GrpcRequest, GrpcWindowRequest}; -#[cfg(feature = "protocol-soap")] -use crank_adapter_soap::{SoapAdapter, SoapRequest}; -#[cfg(feature = "protocol-websocket")] -use crank_adapter_websocket::{WebsocketAdapter, WebsocketWindowRequest}; - #[derive(Clone)] pub struct RuntimeExecutor { - #[cfg(feature = "protocol-graphql")] - graphql_adapter: GraphqlAdapter, - #[cfg(feature = "protocol-grpc")] - grpc_adapter: GrpcAdapter, - rest_adapter: RestAdapter, - #[cfg(feature = "protocol-soap")] - soap_adapter: SoapAdapter, - #[cfg(feature = "protocol-websocket")] - websocket_adapter: WebsocketAdapter, + adapters: AdapterRegistry, limits: RuntimeLimits, unary_limiter: Arc, window_limiter: Arc, @@ -59,15 +42,7 @@ impl RuntimeExecutor { pub fn with_limits(limits: RuntimeLimits) -> Self { Self { - #[cfg(feature = "protocol-graphql")] - graphql_adapter: GraphqlAdapter::new(), - #[cfg(feature = "protocol-grpc")] - grpc_adapter: GrpcAdapter::new(), - rest_adapter: RestAdapter::new(), - #[cfg(feature = "protocol-soap")] - soap_adapter: SoapAdapter::new(), - #[cfg(feature = "protocol-websocket")] - websocket_adapter: WebsocketAdapter::new(), + adapters: community_adapter_registry(), unary_limiter: Arc::new(Semaphore::new(limits.max_concurrent_unary)), window_limiter: Arc::new(Semaphore::new(limits.max_concurrent_window)), session_limiter: Arc::new(Semaphore::new(limits.max_concurrent_sessions)), @@ -185,9 +160,6 @@ impl RuntimeExecutor { let adapter_response = if matches!( streaming.transport_behavior, TransportBehavior::ServerStream - ) && matches!( - operation.target, - Target::Rest(_) | Target::Grpc(_) | Target::Websocket(_) ) { self.execute_window_adapter(operation, prepared_request, request_context) .await? @@ -322,131 +294,31 @@ impl RuntimeExecutor { request_context: Option<&RuntimeRequestContext>, ) -> Result { log_runtime_event("adapter.dispatch", operation, request_context); - let context_headers = runtime_context_headers(request_context); - match &operation.target { - Target::Grpc(target) => { - #[cfg(not(feature = "protocol-grpc"))] - { - let _ = target; - return Err(RuntimeError::UnsupportedProtocol { - protocol: operation.protocol, - }); - } - #[cfg(feature = "protocol-grpc")] - { - let request = GrpcRequest { - headers: merge_headers( - &BTreeMap::new(), - &operation.execution_config.headers, - &prepared_request.headers, - &context_headers, - ), - body: prepared_request - .grpc - .clone() - .unwrap_or(Value::Object(Map::new())), - timeout_ms: operation.execution_config.timeout_ms, - }; - let response = self.grpc_adapter.execute(target, &request).await?; - - Ok(AdapterResponse { - status_code: response.status_code, - headers: response.headers, - body: response.body.clone(), - data: response.body, - }) - } - } - Target::Graphql(target) => { - #[cfg(not(feature = "protocol-graphql"))] - { - let _ = target; - return Err(RuntimeError::UnsupportedProtocol { - protocol: operation.protocol, - }); - } - #[cfg(feature = "protocol-graphql")] - { - let request = GraphqlRequest { - headers: merge_headers( - &BTreeMap::new(), - &operation.execution_config.headers, - &prepared_request.headers, - &context_headers, - ), - variables: prepared_request.variables.clone(), - timeout_ms: operation.execution_config.timeout_ms, - }; - let response = self.graphql_adapter.execute(target, &request).await?; - - Ok(AdapterResponse { - status_code: response.status_code, - headers: response.headers, - body: response.body, - data: response.data, - }) - } - } - Target::Rest(target) => { - let request = RestRequest { - path_params: prepared_request.path_params.clone(), - query_params: prepared_request.query_params.clone(), - headers: merge_headers( - &target.static_headers, - &operation.execution_config.headers, - &prepared_request.headers, - &context_headers, - ), - body: prepared_request.body.clone(), - timeout_ms: operation.execution_config.timeout_ms, - }; - let response = self.rest_adapter.execute(target, &request).await?; - - Ok(AdapterResponse { - status_code: response.status_code, - headers: response.headers, - body: response.body, - data: Value::Null, - }) - } - Target::Soap(target) => { - #[cfg(not(feature = "protocol-soap"))] - { - let _ = target; - Err(RuntimeError::UnsupportedProtocol { - protocol: operation.protocol, - }) - } - #[cfg(feature = "protocol-soap")] - { - let request = SoapRequest { - headers: merge_headers( - &BTreeMap::new(), - &operation.execution_config.headers, - &prepared_request.headers, - &context_headers, - ), - body: prepared_request - .body - .clone() - .unwrap_or(Value::Object(Map::new())), - timeout_ms: operation.execution_config.timeout_ms, - }; - let response = self.soap_adapter.execute(target, &request).await?; - - Ok(AdapterResponse { - status_code: response.status_code, - headers: response.headers, - body: response.body, - data: Value::Null, - }) - } - } - Target::Websocket(_) => Err(RuntimeError::UnsupportedExecutionMode { + let adapter = self.adapter_for(operation)?; + if !adapter.supports_mode(ExecutionMode::Unary) { + return Err(RuntimeError::UnsupportedExecutionMode { operation_id: operation.operation_id.as_str().to_owned(), mode: ExecutionMode::Unary, - }), + }); } + + let prepared_request = adapter_prepared_request( + operation, + &prepared_request, + request_context, + operation.execution_config.timeout_ms, + ); + let adapter_context = adapter_request_context(request_context); + + adapter + .invoke_unary( + &operation.target, + &prepared_request.into(), + &adapter_context, + ) + .await + .map(Into::into) + .map_err(|error| map_protocol_adapter_error(operation, error)) } async fn execute_window_adapter( @@ -456,145 +328,50 @@ impl RuntimeExecutor { request_context: Option<&RuntimeRequestContext>, ) -> Result { log_runtime_event("window.adapter.dispatch", operation, request_context); - let context_headers = runtime_context_headers(request_context); - match &operation.target { - Target::Rest(target) => { - let Some(streaming) = operation.execution_config.streaming.as_ref() else { - return Err(RuntimeError::MissingStreamingConfig { - operation_id: operation.operation_id.as_str().to_owned(), - }); - }; - let request = RestWindowRequest { - request: RestRequest { - path_params: prepared_request.path_params.clone(), - query_params: prepared_request.query_params.clone(), - headers: merge_headers( - &target.static_headers, - &operation.execution_config.headers, - &prepared_request.headers, - &context_headers, - ), - body: prepared_request.body.clone(), - timeout_ms: streaming - .upstream_timeout_ms - .unwrap_or(operation.execution_config.timeout_ms), - }, - window_duration_ms: streaming.window_duration_ms.unwrap_or_default(), - max_items: streaming.max_items.map(|value| value.saturating_add(1)), - }; - let response = self.rest_adapter.execute_window(target, &request).await?; - - Ok(AdapterResponse { - status_code: response.status_code, - headers: response.headers, - body: response.body, - data: Value::Null, - }) - } - Target::Grpc(target) => { - #[cfg(not(feature = "protocol-grpc"))] - { - let _ = target; - return Err(RuntimeError::UnsupportedProtocol { - protocol: operation.protocol, - }); - } - #[cfg(feature = "protocol-grpc")] - { - let Some(streaming) = operation.execution_config.streaming.as_ref() else { - return Err(RuntimeError::MissingStreamingConfig { - operation_id: operation.operation_id.as_str().to_owned(), - }); - }; - let request = GrpcWindowRequest { - request: GrpcRequest { - headers: merge_headers( - &BTreeMap::new(), - &operation.execution_config.headers, - &prepared_request.headers, - &context_headers, - ), - body: prepared_request - .grpc - .clone() - .unwrap_or(Value::Object(Map::new())), - timeout_ms: streaming - .upstream_timeout_ms - .unwrap_or(operation.execution_config.timeout_ms), - }, - window_duration_ms: streaming.window_duration_ms.unwrap_or_default(), - max_items: streaming.max_items.map(|value| value.saturating_add(1)), - }; - let response = self.grpc_adapter.execute_window(target, &request).await?; - - Ok(AdapterResponse { - status_code: response.status_code, - headers: response.headers, - body: response.body, - data: Value::Null, - }) - } - } - Target::Websocket(target) => { - #[cfg(not(feature = "protocol-websocket"))] - { - let _ = target; - Err(RuntimeError::UnsupportedProtocol { - protocol: operation.protocol, - }) - } - #[cfg(feature = "protocol-websocket")] - { - let Some(streaming) = operation.execution_config.streaming.as_ref() else { - return Err(RuntimeError::MissingStreamingConfig { - operation_id: operation.operation_id.as_str().to_owned(), - }); - }; - let websocket_options = operation - .execution_config - .protocol_options - .as_ref() - .and_then(|options| options.websocket.as_ref()); - let request = WebsocketWindowRequest { - headers: merge_headers( - &BTreeMap::new(), - &operation.execution_config.headers, - &prepared_request.headers, - &context_headers, - ), - window_duration_ms: streaming.window_duration_ms.unwrap_or_default(), - max_items: streaming.max_items.map(|value| value.saturating_add(1)), - heartbeat_interval_ms: websocket_options - .and_then(|options| options.heartbeat_interval_ms), - reconnect_max_attempts: websocket_options - .and_then(|options| options.reconnect_max_attempts) - .unwrap_or_default(), - reconnect_backoff_ms: websocket_options - .and_then(|options| options.reconnect_backoff_ms) - .unwrap_or_default(), - }; - let response = self - .websocket_adapter - .execute_window(target, &request) - .await?; - - Ok(AdapterResponse { - status_code: response.status_code, - headers: response.headers, - body: response.body, - data: Value::Null, - }) - } - } - Target::Soap(_) => Err(RuntimeError::UnsupportedExecutionMode { + let Some(streaming) = operation.execution_config.streaming.as_ref() else { + return Err(RuntimeError::MissingStreamingConfig { + operation_id: operation.operation_id.as_str().to_owned(), + }); + }; + let adapter = self.adapter_for(operation)?; + if !adapter.supports_mode(ExecutionMode::Window) { + return Err(RuntimeError::UnsupportedExecutionMode { operation_id: operation.operation_id.as_str().to_owned(), mode: ExecutionMode::Window, - }), - _ => { - self.execute_adapter(operation, prepared_request, request_context) - .await - } + }); } + + let prepared_request = adapter_prepared_request( + operation, + &prepared_request, + request_context, + streaming + .upstream_timeout_ms + .unwrap_or(operation.execution_config.timeout_ms), + ); + let adapter_context = adapter_request_context(request_context); + let result = adapter + .invoke_window( + &operation.target, + &prepared_request.into(), + streaming.window_duration_ms.unwrap_or_default(), + streaming.max_items, + &adapter_context, + ) + .await + .map_err(|error| map_protocol_adapter_error(operation, error))?; + + Ok(AdapterResponse { + status_code: 200, + headers: BTreeMap::new(), + body: json!({ + "summary": result.summary, + "items": result.items, + "cursor": result.cursor, + "done": result.window_complete, + }), + data: Value::Null, + }) } fn acquire_unary_permit( @@ -732,10 +509,76 @@ impl PreparedRequest { grpc: non_empty_payload(request.get("grpc").cloned()), variables: non_empty_payload(request.get("variables").cloned()), body: request.get("body").and_then(non_empty_body).cloned(), + timeout_ms: 0, }) } } +fn community_adapter_registry() -> AdapterRegistry { + AdapterRegistry::empty().register(Arc::new(RestAdapter::new()) as SharedProtocolAdapter) +} + +fn adapter_request_context( + request_context: Option<&RuntimeRequestContext>, +) -> crank_core::RuntimeRequestContext { + request_context + .map(Into::into) + .unwrap_or_else(|| crank_core::RuntimeRequestContext::new(String::new(), String::new())) +} + +fn adapter_prepared_request( + operation: &RuntimeOperation, + prepared_request: &PreparedRequest, + request_context: Option<&RuntimeRequestContext>, + timeout_ms: u64, +) -> PreparedRequest { + let empty_headers = BTreeMap::new(); + let static_headers = match &operation.target { + Target::Rest(target) => &target.static_headers, + _ => &empty_headers, + }; + + let mut prepared_request = prepared_request.clone(); + prepared_request.headers = merge_headers( + static_headers, + &operation.execution_config.headers, + &prepared_request.headers, + &runtime_context_headers(request_context), + ); + prepared_request.timeout_ms = timeout_ms; + prepared_request +} + +fn map_protocol_adapter_error( + operation: &RuntimeOperation, + error: crank_core::ProtocolAdapterError, +) -> RuntimeError { + match error { + crank_core::ProtocolAdapterError::UnsupportedMode { mode, .. } => { + RuntimeError::UnsupportedExecutionMode { + operation_id: operation.operation_id.as_str().to_owned(), + mode, + } + } + crank_core::ProtocolAdapterError::Message(message) => RuntimeError::ProtocolAdapter( + format!("operation {}: {message}", operation.operation_id), + ), + } +} + +impl RuntimeExecutor { + fn adapter_for( + &self, + operation: &RuntimeOperation, + ) -> Result { + self.adapters + .get(operation.protocol) + .ok_or(RuntimeError::UnsupportedProtocol { + protocol: operation.protocol, + }) + } +} + fn finalize_output( operation: &RuntimeOperation, response: &AdapterResponse, diff --git a/crates/crank-runtime/src/model.rs b/crates/crank-runtime/src/model.rs index 034463e..c1e97ad 100644 --- a/crates/crank-runtime/src/model.rs +++ b/crates/crank-runtime/src/model.rs @@ -53,6 +53,8 @@ pub struct PreparedRequest { pub variables: Option, #[serde(skip_serializing_if = "Option::is_none")] pub body: Option, + #[serde(default)] + pub timeout_ms: u64, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] @@ -63,3 +65,42 @@ pub struct AdapterResponse { pub body: Value, pub data: Value, } + +impl From for crank_core::PreparedRequest { + fn from(value: PreparedRequest) -> Self { + Self { + path_params: value.path_params, + query_params: value.query_params, + headers: value.headers, + grpc: value.grpc, + variables: value.variables, + body: value.body, + timeout_ms: value.timeout_ms, + } + } +} + +impl From for PreparedRequest { + fn from(value: crank_core::PreparedRequest) -> Self { + Self { + path_params: value.path_params, + query_params: value.query_params, + headers: value.headers, + grpc: value.grpc, + variables: value.variables, + body: value.body, + timeout_ms: value.timeout_ms, + } + } +} + +impl From for AdapterResponse { + fn from(value: crank_core::AdapterResponse) -> Self { + Self { + status_code: value.status_code, + headers: value.headers, + body: value.body, + data: value.data, + } + } +} diff --git a/crates/crank-runtime/src/request_context.rs b/crates/crank-runtime/src/request_context.rs index b292dc0..d57906c 100644 --- a/crates/crank-runtime/src/request_context.rs +++ b/crates/crank-runtime/src/request_context.rs @@ -51,6 +51,34 @@ impl RuntimeRequestContext { } } +impl From for crank_core::ResponseCacheScope { + fn from(value: ResponseCacheScope) -> Self { + Self { + workspace_key: value.workspace_key, + agent_key: value.agent_key, + } + } +} + +impl From<&ResponseCacheScope> for crank_core::ResponseCacheScope { + fn from(value: &ResponseCacheScope) -> Self { + Self { + workspace_key: value.workspace_key.clone(), + agent_key: value.agent_key.clone(), + } + } +} + +impl From<&RuntimeRequestContext> for crank_core::RuntimeRequestContext { + fn from(value: &RuntimeRequestContext) -> Self { + Self { + request_id: value.request_id.clone(), + correlation_id: value.correlation_id.clone(), + response_cache_scope: value.response_cache_scope.as_ref().map(Into::into), + } + } +} + #[cfg(test)] mod tests { use super::RuntimeRequestContext; diff --git a/crates/crank-runtime/src/streaming.rs b/crates/crank-runtime/src/streaming.rs index 6e19ab3..e5d9e45 100644 --- a/crates/crank-runtime/src/streaming.rs +++ b/crates/crank-runtime/src/streaming.rs @@ -11,3 +11,16 @@ pub struct WindowExecutionResult { pub truncated: bool, pub has_more: bool, } + +impl From for WindowExecutionResult { + fn from(value: crank_core::WindowExecutionResult) -> Self { + Self { + summary: value.summary, + items: value.items, + cursor: value.cursor, + window_complete: value.window_complete, + truncated: value.truncated, + has_more: value.has_more, + } + } +}