diff --git a/TASKS.md b/TASKS.md index 0d0620b..d7e8887 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,19 +2,20 @@ ## Current -### `feat/stream-session-store` +### `feat/runtime-window-mode` Status: completed DoD: -- persistent tables exist for `stream_sessions` and `async_jobs` -- registry supports create/get/update/close/list/cleanup for both stores -- invalid state transitions are rejected explicitly -- migration and storage tests cover lifecycle and cleanup +- runtime exposes bounded `window` execution +- item and byte limits are enforced +- `window_complete`, `truncated`, `has_more` are returned +- aggregation and redaction are covered by tests +- unary execution remains intact ## Next -- `feat/runtime-window-mode` +- `feat/rest-sse-adapter` ## Backlog diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index f9b288e..b0fe3f2 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -240,5 +240,8 @@ fn runtime_test_failure_code(error: &RuntimeError) -> &'static str { RuntimeError::RestAdapter(_) => "runtime_rest_error", RuntimeError::UnsupportedProtocol { .. } => "runtime_protocol_error", RuntimeError::InvalidPreparedRequest { .. } => "runtime_request_error", + RuntimeError::MissingStreamingConfig { .. } => "runtime_streaming_config_error", + RuntimeError::UnsupportedExecutionMode { .. } => "runtime_streaming_mode_error", + RuntimeError::InvalidStreamingPayload { .. } => "runtime_streaming_payload_error", } } diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 3ac6230..ddb9c7e 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -3756,6 +3756,9 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str { RuntimeError::GrpcAdapter(_) => "grpc_error", RuntimeError::RestAdapter(_) => "rest_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", + RuntimeError::MissingStreamingConfig { .. } => "streaming_config_error", + RuntimeError::UnsupportedExecutionMode { .. } => "streaming_mode_error", + RuntimeError::InvalidStreamingPayload { .. } => "streaming_payload_error", } } diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs index b86e677..04233e1 100644 --- a/apps/mcp-server/src/app.rs +++ b/apps/mcp-server/src/app.rs @@ -846,7 +846,10 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str { RuntimeError::GrpcAdapter(_) => "adapter_execution_error", RuntimeError::RestAdapter(_) => "adapter_execution_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", + RuntimeError::MissingStreamingConfig { .. } => "streaming_config_error", + RuntimeError::UnsupportedExecutionMode { .. } => "streaming_mode_error", RuntimeError::InvalidPreparedRequest { .. } => "runtime_error", + RuntimeError::InvalidStreamingPayload { .. } => "streaming_payload_error", } } diff --git a/crates/crank-runtime/src/aggregation.rs b/crates/crank-runtime/src/aggregation.rs new file mode 100644 index 0000000..79d9545 --- /dev/null +++ b/crates/crank-runtime/src/aggregation.rs @@ -0,0 +1,292 @@ +use serde_json::{Map, Value, json}; + +use crate::{RuntimeError, WindowExecutionResult}; + +pub fn collect_window_result( + response: &Value, + config: &crank_core::StreamingConfig, +) -> Result { + let mut items = extract_path_value(response, config.items_path.as_deref()) + .and_then(|value| value.as_array().cloned()) + .unwrap_or_default(); + let cursor = extract_path_value(response, config.cursor_path.as_deref()).cloned(); + let done = extract_path_value(response, config.done_path.as_deref()) + .and_then(Value::as_bool) + .unwrap_or(true); + + let mut truncated = false; + let mut has_more = !done; + + if let Some(max_items) = config.max_items.map(|value| value as usize) { + if items.len() > max_items { + items.truncate(max_items); + truncated = true; + has_more = true; + } + } + + let mut summary = build_summary(response, &items, config); + redact_and_truncate(&mut summary, &mut items, config); + + if let Some(max_bytes) = config.max_bytes.map(|value| value as usize) { + while payload_size_bytes(&summary, &items, cursor.as_ref()) > max_bytes && !items.is_empty() + { + items.pop(); + truncated = true; + has_more = true; + } + + if payload_size_bytes(&summary, &items, cursor.as_ref()) > max_bytes { + summary = json!({ + "notice": "window payload exceeded byte limit", + "items_returned": items.len() + }); + truncated = true; + has_more = true; + } + } + + if matches!( + config.aggregation_mode, + crank_core::AggregationMode::SummaryOnly + ) { + items.clear(); + } + + Ok(WindowExecutionResult { + summary, + items, + cursor, + window_complete: !has_more, + truncated, + has_more, + }) +} + +fn build_summary(response: &Value, items: &[Value], config: &crank_core::StreamingConfig) -> Value { + match config.aggregation_mode { + crank_core::AggregationMode::RawItems => Value::Null, + crank_core::AggregationMode::SummaryOnly + | crank_core::AggregationMode::SummaryPlusSamples => { + extract_path_value(response, config.summary_path.as_deref()) + .cloned() + .unwrap_or_else(|| stats_summary(items)) + } + crank_core::AggregationMode::Stats => stats_summary(items), + crank_core::AggregationMode::LatestState => items.last().cloned().unwrap_or(Value::Null), + } +} + +fn stats_summary(items: &[Value]) -> Value { + let mut summary = Map::new(); + summary.insert("items_total".to_owned(), Value::from(items.len() as u64)); + summary.insert("items_sampled".to_owned(), Value::from(items.len() as u64)); + Value::Object(summary) +} + +fn redact_and_truncate( + summary: &mut Value, + items: &mut [Value], + config: &crank_core::StreamingConfig, +) { + crate::redaction::redact_paths(summary, &config.redacted_paths); + + for item in items.iter_mut() { + crate::redaction::redact_paths(item, &config.redacted_paths); + } + + if config.truncate_item_fields { + if let Some(max_len) = config.max_field_length.map(|value| value as usize) { + for item in items.iter_mut() { + crate::redaction::truncate_item_fields(item, max_len); + } + } + } +} + +fn payload_size_bytes(summary: &Value, items: &[Value], cursor: Option<&Value>) -> usize { + serde_json::to_vec(&json!({ + "summary": summary, + "items": items, + "cursor": cursor + })) + .map(|value| value.len()) + .unwrap_or(usize::MAX) +} + +pub fn extract_path_value<'a>(value: &'a Value, path: Option<&str>) -> Option<&'a Value> { + let path = path?; + let path = path.strip_prefix("$.").or_else(|| path.strip_prefix('$'))?; + + if path.is_empty() { + return Some(value); + } + + let mut current = value; + for segment in path.split('.') { + current = current.get(segment)?; + } + + Some(current) +} + +#[cfg(test)] +mod tests { + use serde_json::{Value, json}; + + use crank_core::{ + AggregationMode, ExecutionMode, StreamingConfig, ToolFamilyConfig, TransportBehavior, + }; + + use super::collect_window_result; + + fn window_config() -> StreamingConfig { + StreamingConfig { + mode: ExecutionMode::Window, + transport_behavior: TransportBehavior::ServerStream, + window_duration_ms: Some(3_000), + poll_interval_ms: None, + upstream_timeout_ms: Some(5_000), + idle_timeout_ms: None, + max_session_lifetime_ms: None, + max_items: Some(10), + max_bytes: Some(10_000), + aggregation_mode: AggregationMode::RawItems, + summary_path: Some("$.summary".to_owned()), + items_path: Some("$.items".to_owned()), + cursor_path: Some("$.cursor".to_owned()), + status_path: None, + done_path: Some("$.done".to_owned()), + redacted_paths: Vec::new(), + truncate_item_fields: false, + max_field_length: None, + drop_duplicates: false, + sampling_rate: None, + tool_family: ToolFamilyConfig::default(), + } + } + + #[test] + fn collects_raw_items_mode() { + let config = window_config(); + let result = collect_window_result( + &json!({ + "items": [{ "message": "one" }, { "message": "two" }], + "summary": { "count": 2 }, + "done": true + }), + &config, + ) + .unwrap(); + + assert_eq!(result.summary, Value::Null); + assert_eq!(result.items.len(), 2); + assert!(result.window_complete); + assert!(!result.truncated); + assert!(!result.has_more); + } + + #[test] + fn builds_summary_only_mode() { + let mut config = window_config(); + config.aggregation_mode = AggregationMode::SummaryOnly; + + let result = collect_window_result( + &json!({ + "items": [{ "message": "one" }], + "summary": { "count": 1 }, + "done": true + }), + &config, + ) + .unwrap(); + + assert_eq!(result.summary, json!({ "count": 1 })); + assert!(result.items.is_empty()); + } + + #[test] + fn keeps_summary_plus_samples() { + let mut config = window_config(); + config.aggregation_mode = AggregationMode::SummaryPlusSamples; + + let result = collect_window_result( + &json!({ + "items": [{ "message": "one" }, { "message": "two" }], + "summary": { "count": 2 }, + "done": true + }), + &config, + ) + .unwrap(); + + assert_eq!(result.summary, json!({ "count": 2 })); + assert_eq!(result.items.len(), 2); + } + + #[test] + fn truncates_by_item_count() { + let mut config = window_config(); + config.max_items = Some(1); + + let result = collect_window_result( + &json!({ + "items": [{ "message": "one" }, { "message": "two" }], + "cursor": "next", + "done": false + }), + &config, + ) + .unwrap(); + + assert_eq!(result.items.len(), 1); + assert!(result.truncated); + assert!(result.has_more); + assert!(!result.window_complete); + } + + #[test] + fn truncates_by_byte_limit() { + let mut config = window_config(); + config.max_bytes = Some(120); + config.aggregation_mode = AggregationMode::SummaryPlusSamples; + + let result = collect_window_result( + &json!({ + "items": [ + { "message": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" }, + { "message": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" } + ], + "summary": { "count": 2 }, + "done": true + }), + &config, + ) + .unwrap(); + + assert!(result.truncated); + assert!(result.items.len() < 2); + } + + #[test] + fn redacts_and_truncates_values() { + let mut config = window_config(); + config.redacted_paths = vec!["$.secret".to_owned()]; + config.truncate_item_fields = true; + config.max_field_length = Some(4); + + let result = collect_window_result( + &json!({ + "items": [ + { "message": "abcdefghijklmnop", "secret": "token" } + ], + "done": true + }), + &config, + ) + .unwrap(); + + assert_eq!(result.items[0]["secret"], json!("[REDACTED]")); + assert_eq!(result.items[0]["message"], json!("abcd...")); + } +} diff --git a/crates/crank-runtime/src/error.rs b/crates/crank-runtime/src/error.rs index 85b1827..dfb00c4 100644 --- a/crates/crank-runtime/src/error.rs +++ b/crates/crank-runtime/src/error.rs @@ -1,7 +1,7 @@ use crank_adapter_graphql::GraphqlAdapterError; use crank_adapter_grpc::GrpcAdapterError; use crank_adapter_rest::RestAdapterError; -use crank_core::Protocol; +use crank_core::{ExecutionMode, Protocol}; use crank_mapping::MappingError; use crank_schema::SchemaError; use thiserror::Error; @@ -20,6 +20,15 @@ pub enum RuntimeError { RestAdapter(#[from] RestAdapterError), #[error("protocol {protocol:?} is not supported by runtime")] UnsupportedProtocol { protocol: Protocol }, + #[error("operation {operation_id} does not define streaming config")] + MissingStreamingConfig { operation_id: String }, + #[error("operation {operation_id} does not support requested execution mode {mode:?}")] + UnsupportedExecutionMode { + operation_id: String, + mode: ExecutionMode, + }, #[error("invalid prepared request: {details}")] InvalidPreparedRequest { details: String }, + #[error("invalid streaming payload: {details}")] + InvalidStreamingPayload { details: String }, } diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index 32e5d94..16d3253 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -3,10 +3,12 @@ use std::collections::BTreeMap; use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest}; use crank_adapter_grpc::{GrpcAdapter, GrpcRequest}; use crank_adapter_rest::{RestAdapter, RestRequest}; -use crank_core::Target; +use crank_core::{ExecutionMode, Target}; use serde_json::{Map, Value, json}; -use crate::{AdapterResponse, PreparedRequest, RuntimeError, RuntimeOperation}; +use crate::{ + AdapterResponse, PreparedRequest, RuntimeError, RuntimeOperation, WindowExecutionResult, +}; #[derive(Clone, Debug)] pub struct RuntimeExecutor { @@ -39,6 +41,30 @@ impl RuntimeExecutor { self.execute_prepared(operation, prepared_request).await } + pub async fn execute_window( + &self, + operation: &RuntimeOperation, + input: &Value, + ) -> Result { + let Some(streaming) = operation.execution_config.streaming.as_ref() else { + return Err(RuntimeError::MissingStreamingConfig { + operation_id: operation.operation_id.as_str().to_owned(), + }); + }; + + if streaming.mode != ExecutionMode::Window { + return Err(RuntimeError::UnsupportedExecutionMode { + operation_id: operation.operation_id.as_str().to_owned(), + mode: streaming.mode, + }); + } + + let prepared_request = self.prepare_request(operation, input)?; + let adapter_response = self.execute_adapter(operation, prepared_request).await?; + + crate::aggregation::collect_window_result(&adapter_response.body, streaming) + } + pub fn prepare_request( &self, operation: &RuntimeOperation, @@ -55,7 +81,20 @@ impl RuntimeExecutor { operation: &RuntimeOperation, prepared_request: PreparedRequest, ) -> Result { - let adapter_response = match &operation.target { + let adapter_response = self.execute_adapter(operation, prepared_request).await?; + let finalized_output = finalize_output(operation, &adapter_response)?; + + operation.output_schema.validate_shape(&finalized_output)?; + + Ok(finalized_output) + } + + async fn execute_adapter( + &self, + operation: &RuntimeOperation, + prepared_request: PreparedRequest, + ) -> Result { + match &operation.target { Target::Grpc(target) => { let request = GrpcRequest { headers: merge_headers( @@ -71,12 +110,12 @@ impl RuntimeExecutor { }; let response = self.grpc_adapter.execute(target, &request).await?; - AdapterResponse { + Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body.clone(), data: response.body, - } + }) } Target::Graphql(target) => { let request = GraphqlRequest { @@ -90,12 +129,12 @@ impl RuntimeExecutor { }; let response = self.graphql_adapter.execute(target, &request).await?; - AdapterResponse { + Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: response.data, - } + }) } Target::Rest(target) => { let request = RestRequest { @@ -111,20 +150,14 @@ impl RuntimeExecutor { }; let response = self.rest_adapter.execute(target, &request).await?; - AdapterResponse { + Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: Value::Null, - } + }) } - }; - - let finalized_output = finalize_output(operation, &adapter_response)?; - - operation.output_schema.validate_shape(&finalized_output)?; - - Ok(finalized_output) + } } } @@ -234,9 +267,10 @@ mod tests { use axum::{Json, Router, routing::post}; use crank_adapter_grpc::test_support as grpc_test_support; use crank_core::{ - DescriptorId, ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, GraphqlOperationType, - GraphqlTarget, GrpcTarget, HttpMethod, Operation, OperationId, OperationStatus, Protocol, - RestTarget, Samples, Target, ToolDescription, ToolExample, + AggregationMode, DescriptorId, ExecutionConfig, ExecutionMode, GeneratedDraft, + GeneratedDraftStatus, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod, + Operation, OperationId, OperationStatus, Protocol, RestTarget, Samples, StreamingConfig, + Target, ToolDescription, ToolExample, ToolFamilyConfig, TransportBehavior, }; use crank_mapping::{MappingRule, MappingSet}; use crank_schema::{Schema, SchemaKind}; @@ -340,8 +374,81 @@ mod tests { assert!(matches!(error, RuntimeError::GraphqlAdapter(_))); } + #[tokio::test] + async fn executes_window_mode_with_raw_items() { + let base_url = spawn_runtime_server().await; + let executor = RuntimeExecutor::new(); + let operation = test_window_operation(&base_url, AggregationMode::RawItems, None, None); + + let result = executor + .execute_window(&operation, &json!({})) + .await + .unwrap(); + + assert_eq!(result.items.len(), 3); + assert_eq!(result.summary, Value::Null); + assert!(result.window_complete); + assert!(!result.truncated); + } + + #[tokio::test] + async fn executes_window_mode_with_summary_only() { + let base_url = spawn_runtime_server().await; + let executor = RuntimeExecutor::new(); + let operation = + test_window_operation(&base_url, AggregationMode::SummaryOnly, Some(10), None); + + let result = executor + .execute_window(&operation, &json!({})) + .await + .unwrap(); + + assert_eq!(result.summary, json!({ "service": "billing", "errors": 1 })); + assert!(result.items.is_empty()); + } + + #[tokio::test] + async fn executes_window_mode_with_truncation_and_redaction() { + let base_url = spawn_runtime_server().await; + let executor = RuntimeExecutor::new(); + let operation = test_window_operation( + &base_url, + AggregationMode::SummaryPlusSamples, + Some(2), + Some(120), + ); + + let result = executor + .execute_window(&operation, &json!({})) + .await + .unwrap(); + + assert!(result.truncated); + assert!(result.has_more); + assert_eq!(result.items.len(), 1); + assert_eq!(result.items[0]["secret"], json!("[REDACTED]")); + assert_eq!(result.items[0]["message"], json!("disk...")); + } + + #[tokio::test] + async fn propagates_timeout_in_window_mode() { + let base_url = spawn_runtime_server().await; + let executor = RuntimeExecutor::new(); + let operation = test_slow_window_operation(&base_url); + + let error = executor + .execute_window(&operation, &json!({})) + .await + .unwrap_err(); + + assert!(matches!(error, RuntimeError::RestAdapter(_))); + } + async fn spawn_runtime_server() -> String { - let app = Router::new().route("/leads", post(create_lead)); + let app = Router::new() + .route("/leads", post(create_lead)) + .route("/events", post(events_window)) + .route("/slow-events", post(slow_events_window)); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); @@ -383,6 +490,27 @@ mod tests { ) } + async fn events_window() -> (axum::http::StatusCode, Json) { + ( + axum::http::StatusCode::OK, + Json(json!({ + "summary": { "service": "billing", "errors": 1 }, + "items": [ + { "message": "disk pressure detected", "secret": "token-a" }, + { "message": "error budget exhausted", "secret": "token-b" }, + { "message": "node restarted", "secret": "token-c" } + ], + "cursor": "cursor_02", + "done": true + })), + ) + } + + async fn slow_events_window() -> (axum::http::StatusCode, Json) { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + events_window().await + } + async fn graphql_handler(Json(payload): Json) -> Json { let email = payload .get("variables") @@ -653,6 +781,107 @@ mod tests { }) } + fn test_window_operation( + base_url: &str, + aggregation_mode: AggregationMode, + max_items: Option, + max_bytes: Option, + ) -> RuntimeOperation { + RuntimeOperation::from(Operation { + id: OperationId::new("op_window_runtime"), + name: "billing_log_window".to_owned(), + display_name: "Billing Log Window".to_owned(), + category: "ops".to_owned(), + protocol: Protocol::Rest, + status: OperationStatus::Published, + version: 1, + target: Target::Rest(RestTarget { + base_url: base_url.to_owned(), + method: HttpMethod::Post, + path_template: "/events".to_owned(), + static_headers: BTreeMap::new(), + }), + input_schema: Schema { + kind: SchemaKind::Object, + description: None, + required: true, + nullable: false, + default_value: None, + fields: BTreeMap::new(), + items: None, + enum_values: Vec::new(), + variants: Vec::new(), + }, + output_schema: object_schema("ignored", SchemaKind::String), + input_mapping: MappingSet { + rules: vec![MappingRule { + source: "$.mcp".to_owned(), + target: "$.request.body".to_owned(), + required: false, + default_value: None, + transform: None, + condition: None, + notes: None, + }], + }, + output_mapping: MappingSet { rules: Vec::new() }, + execution_config: ExecutionConfig { + timeout_ms: 1_000, + retry_policy: None, + auth_profile_ref: None, + headers: BTreeMap::new(), + protocol_options: None, + streaming: Some(StreamingConfig { + mode: ExecutionMode::Window, + transport_behavior: TransportBehavior::ServerStream, + window_duration_ms: Some(3_000), + poll_interval_ms: None, + upstream_timeout_ms: Some(1_000), + idle_timeout_ms: None, + max_session_lifetime_ms: None, + max_items, + max_bytes, + aggregation_mode, + summary_path: Some("$.summary".to_owned()), + items_path: Some("$.items".to_owned()), + cursor_path: Some("$.cursor".to_owned()), + status_path: None, + done_path: Some("$.done".to_owned()), + redacted_paths: vec!["$.secret".to_owned()], + truncate_item_fields: true, + max_field_length: Some(4), + drop_duplicates: false, + sampling_rate: None, + tool_family: ToolFamilyConfig::default(), + }), + }, + tool_description: ToolDescription { + title: "Billing Log Window".to_owned(), + description: "Collects bounded billing events.".to_owned(), + tags: vec!["logs".to_owned()], + examples: Vec::new(), + }, + samples: None, + generated_draft: None, + config_export: None, + created_at: "2026-04-06T12:00:00Z".to_owned(), + updated_at: "2026-04-06T12:00:00Z".to_owned(), + published_at: Some("2026-04-06T12:00:00Z".to_owned()), + }) + } + + fn test_slow_window_operation(base_url: &str) -> RuntimeOperation { + let mut operation = + test_window_operation(base_url, AggregationMode::RawItems, Some(10), None); + + if let Target::Rest(target) = &mut operation.target { + target.path_template = "/slow-events".to_owned(); + } + + operation.execution_config.timeout_ms = 10; + operation + } + fn object_schema(field_name: &str, kind: SchemaKind) -> Schema { Schema { kind: SchemaKind::Object, diff --git a/crates/crank-runtime/src/lib.rs b/crates/crank-runtime/src/lib.rs index f712205..40dd27c 100644 --- a/crates/crank-runtime/src/lib.rs +++ b/crates/crank-runtime/src/lib.rs @@ -1,7 +1,11 @@ +mod aggregation; mod error; mod executor; mod model; +mod redaction; +mod streaming; pub use error::RuntimeError; pub use executor::RuntimeExecutor; pub use model::{AdapterResponse, PreparedRequest, RuntimeOperation}; +pub use streaming::WindowExecutionResult; diff --git a/crates/crank-runtime/src/redaction.rs b/crates/crank-runtime/src/redaction.rs new file mode 100644 index 0000000..ebecc50 --- /dev/null +++ b/crates/crank-runtime/src/redaction.rs @@ -0,0 +1,124 @@ +use serde_json::Value; + +pub fn redact_paths(value: &mut Value, paths: &[String]) { + for path in paths { + let segments = parse_path(path); + if segments.is_empty() { + continue; + } + + redact_path(value, &segments); + } +} + +pub fn truncate_item_fields(value: &mut Value, max_len: usize) { + match value { + Value::String(text) => { + if text != "[REDACTED]" && text.chars().count() > max_len { + let truncated = text.chars().take(max_len).collect::(); + *text = format!("{truncated}..."); + } + } + Value::Array(items) => { + for item in items { + truncate_item_fields(item, max_len); + } + } + Value::Object(map) => { + for value in map.values_mut() { + truncate_item_fields(value, max_len); + } + } + Value::Null | Value::Bool(_) | Value::Number(_) => {} + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum PathSegment { + Field(String), + Wildcard, +} + +fn parse_path(path: &str) -> Vec { + let path = path.strip_prefix("$.").or_else(|| path.strip_prefix('$')); + let Some(path) = path else { + return Vec::new(); + }; + + path.split('.') + .flat_map(|segment| { + if let Some(field) = segment.strip_suffix("[*]") { + vec![PathSegment::Field(field.to_owned()), PathSegment::Wildcard] + } else if segment == "*" { + vec![PathSegment::Wildcard] + } else { + vec![PathSegment::Field(segment.to_owned())] + } + }) + .collect() +} + +fn redact_path(value: &mut Value, segments: &[PathSegment]) { + let Some((current, rest)) = segments.split_first() else { + *value = Value::String("[REDACTED]".to_owned()); + return; + }; + + match current { + PathSegment::Field(field) => { + if let Some(next) = value + .as_object_mut() + .and_then(|object| object.get_mut(field)) + { + redact_path(next, rest); + } + } + PathSegment::Wildcard => { + if let Some(items) = value.as_array_mut() { + for item in items { + redact_path(item, rest); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::{redact_paths, truncate_item_fields}; + + #[test] + fn redacts_simple_and_wildcard_paths() { + let mut value = json!({ + "summary": { "token": "secret" }, + "items": [ + { "token": "a", "message": "keep" }, + { "token": "b", "message": "keep" } + ] + }); + + redact_paths( + &mut value, + &["$.summary.token".to_owned(), "$.items[*].token".to_owned()], + ); + + assert_eq!(value["summary"]["token"], json!("[REDACTED]")); + assert_eq!(value["items"][0]["token"], json!("[REDACTED]")); + assert_eq!(value["items"][1]["token"], json!("[REDACTED]")); + } + + #[test] + fn truncates_long_strings_recursively() { + let mut value = json!({ + "message": "abcdefghijklmnop", + "nested": [{ "message": "qrstuvwxyz" }] + }); + + truncate_item_fields(&mut value, 4); + + assert_eq!(value["message"], json!("abcd...")); + assert_eq!(value["nested"][0]["message"], json!("qrst...")); + } +} diff --git a/crates/crank-runtime/src/streaming.rs b/crates/crank-runtime/src/streaming.rs new file mode 100644 index 0000000..6e19ab3 --- /dev/null +++ b/crates/crank-runtime/src/streaming.rs @@ -0,0 +1,13 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct WindowExecutionResult { + pub summary: Value, + pub items: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub cursor: Option, + pub window_complete: bool, + pub truncated: bool, + pub has_more: bool, +}