From 6a0381b8e58b5af766a0e757f59a322f52676492 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Mon, 6 Apr 2026 09:57:28 +0300 Subject: [PATCH] feat: add streaming core domain model --- TASKS.md | 14 +- apps/admin-api/src/app.rs | 3 + apps/admin-api/src/service.rs | 4 + apps/mcp-server/src/main.rs | 3 + crates/crank-core/src/ids.rs | 2 + crates/crank-core/src/lib.rs | 12 +- crates/crank-core/src/operation.rs | 9 +- crates/crank-core/src/protocol.rs | 40 +++ crates/crank-core/src/stream_session.rs | 210 ++++++++++++++ crates/crank-core/src/streaming.rs | 364 ++++++++++++++++++++++++ crates/crank-registry/src/postgres.rs | 1 + crates/crank-runtime/src/executor.rs | 3 + 12 files changed, 654 insertions(+), 11 deletions(-) create mode 100644 crates/crank-core/src/stream_session.rs create mode 100644 crates/crank-core/src/streaming.rs diff --git a/TASKS.md b/TASKS.md index 9413669..603ebb3 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,20 +2,20 @@ ## Current -### `feat/mcp-streamable-http-alignment` +### `feat/streaming-core-model` Status: completed DoD: -- `mcp-server` transport semantics match MCP Streamable HTTP -- `GET` SSE stream is supported for initialized sessions -- `POST` can return JSON or SSE depending on negotiated response mode -- session and protocol headers are validated consistently -- transport tests cover JSON mode, SSE mode, GET stream and session deletion +- core domain types exist for window/session/job execution +- `ExecutionConfig` can carry `streaming` +- protocol-aware validation follows documented capability matrix +- streaming types are exported from `crank-core` +- core tests cover serde and validation rules ## Next -- `feat/streaming-core-model` +- `feat/stream-session-store` ## Backlog diff --git a/apps/admin-api/src/app.rs b/apps/admin-api/src/app.rs index 3337f38..b975275 100644 --- a/apps/admin-api/src/app.rs +++ b/apps/admin-api/src/app.rs @@ -1961,6 +1961,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Create Lead".to_owned(), @@ -2016,6 +2017,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Create Lead GraphQL".to_owned(), @@ -2070,6 +2072,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Unary Echo gRPC".to_owned(), diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 607bbf9..3ac6230 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -3468,6 +3468,7 @@ fn demo_rest_operation_payload() -> OperationPayload { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: crank_core::ToolDescription { title: "Create CRM Lead".to_owned(), @@ -3521,6 +3522,7 @@ fn demo_graphql_operation_payload() -> OperationPayload { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: crank_core::ToolDescription { title: "Get Invoice Status".to_owned(), @@ -3574,6 +3576,7 @@ fn demo_grpc_operation_payload() -> OperationPayload { protocol_options: Some(crank_core::ProtocolOptions { grpc: Some(crank_core::GrpcProtocolOptions { use_tls: false }), }), + streaming: None, }, tool_description: crank_core::ToolDescription { title: "Lookup Support Ticket".to_owned(), @@ -3631,6 +3634,7 @@ fn demo_archived_operation_payload() -> OperationPayload { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: crank_core::ToolDescription { title: "Archive Marketing Contact".to_owned(), diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs index 7fc4fe9..cd034f2 100644 --- a/apps/mcp-server/src/main.rs +++ b/apps/mcp-server/src/main.rs @@ -1128,6 +1128,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Create Lead".to_owned(), @@ -1192,6 +1193,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Create Lead GraphQL".to_owned(), @@ -1255,6 +1257,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Unary Echo gRPC".to_owned(), diff --git a/crates/crank-core/src/ids.rs b/crates/crank-core/src/ids.rs index be5a2b4..62bee94 100644 --- a/crates/crank-core/src/ids.rs +++ b/crates/crank-core/src/ids.rs @@ -48,3 +48,5 @@ define_id!(InvitationId); define_id!(PlatformApiKeyId); define_id!(InvocationLogId); define_id!(SecretId); +define_id!(StreamSessionId); +define_id!(AsyncJobId); diff --git a/crates/crank-core/src/lib.rs b/crates/crank-core/src/lib.rs index 688f7e3..1006a79 100644 --- a/crates/crank-core/src/lib.rs +++ b/crates/crank-core/src/lib.rs @@ -6,6 +6,8 @@ pub mod observability; pub mod operation; pub mod protocol; pub mod secret; +pub mod stream_session; +pub mod streaming; pub mod workspace; pub use access::{ @@ -18,8 +20,9 @@ pub use auth::{ BearerAuthConfig, SecretRef, }; pub use ids::{ - AgentId, AuthProfileId, DescriptorId, InvitationId, InvocationLogId, OperationId, - PlatformApiKeyId, SampleId, SecretId, ToolId, UserId, UserSessionId, WorkspaceId, + AgentId, AsyncJobId, AuthProfileId, DescriptorId, InvitationId, InvocationLogId, OperationId, + PlatformApiKeyId, SampleId, SecretId, StreamSessionId, ToolId, UserId, UserSessionId, + WorkspaceId, }; pub use observability::{ InvocationLevel, InvocationLog, InvocationSource, InvocationStatus, UsagePeriod, UsageRollup, @@ -31,4 +34,9 @@ pub use operation::{ }; pub use protocol::{AuthKind, ExportMode, GraphqlOperationType, HttpMethod, Protocol}; pub use secret::{Secret, SecretKind, SecretStatus, SecretVersion}; +pub use stream_session::{AsyncJobHandle, JobStatus, StreamSession, StreamStatus}; +pub use streaming::{ + AggregationMode, ExecutionMode, StreamingConfig, StreamingConfigError, ToolFamilyConfig, + TransportBehavior, +}; pub use workspace::{Workspace, WorkspaceStatus}; diff --git a/crates/crank-core/src/operation.rs b/crates/crank-core/src/operation.rs index 99c42a3..709e673 100644 --- a/crates/crank-core/src/operation.rs +++ b/crates/crank-core/src/operation.rs @@ -6,6 +6,7 @@ use serde_json::Value; use crate::{ ids::{AuthProfileId, DescriptorId, OperationId, SampleId}, protocol::{ExportMode, GraphqlOperationType, HttpMethod, Protocol}, + streaming::StreamingConfig, }; fn default_operation_category() -> String { @@ -73,7 +74,7 @@ pub struct ProtocolOptions { pub grpc: Option, } -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct ExecutionConfig { pub timeout_ms: u64, #[serde(skip_serializing_if = "Option::is_none")] @@ -84,6 +85,8 @@ pub struct ExecutionConfig { pub headers: BTreeMap, #[serde(skip_serializing_if = "Option::is_none")] pub protocol_options: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub streaming: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -143,7 +146,7 @@ pub struct ConfigExport { pub export_mode: ExportMode, } -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct Operation { pub id: OperationId, pub name: String, @@ -267,6 +270,7 @@ mod tests { auth_profile_ref: Some(AuthProfileId::new("auth_01")), headers: BTreeMap::new(), protocol_options: Some(ProtocolOptions::default()), + streaming: None, }, tool_description: ToolDescription { title: "Create CRM lead".to_owned(), @@ -342,6 +346,7 @@ mod tests { auth_profile_ref: Some(AuthProfileId::new("auth_01")), headers: BTreeMap::new(), protocol_options: Some(ProtocolOptions::default()), + streaming: None, }, tool_description: ToolDescription { title: "Create CRM lead".to_owned(), diff --git a/crates/crank-core/src/protocol.rs b/crates/crank-core/src/protocol.rs index 2258ce6..1f59eed 100644 --- a/crates/crank-core/src/protocol.rs +++ b/crates/crank-core/src/protocol.rs @@ -1,5 +1,7 @@ use serde::{Deserialize, Serialize}; +use crate::streaming::{ExecutionMode, TransportBehavior}; + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum Protocol { @@ -40,3 +42,41 @@ pub enum ExportMode { Portable, Bundle, } + +impl Protocol { + pub fn supports_execution_mode(self, mode: ExecutionMode) -> bool { + match self { + Self::Rest => true, + Self::Graphql => matches!(mode, ExecutionMode::Unary), + Self::Grpc => true, + } + } + + pub fn supports_transport_behavior(self, behavior: TransportBehavior) -> bool { + match self { + Self::Rest => true, + Self::Graphql => matches!(behavior, TransportBehavior::RequestResponse), + Self::Grpc => !matches!(behavior, TransportBehavior::DeferredResult), + } + } +} + +#[cfg(test)] +mod tests { + use super::Protocol; + use crate::streaming::{ExecutionMode, TransportBehavior}; + + #[test] + fn graphql_support_matrix_is_restricted() { + assert!(Protocol::Graphql.supports_execution_mode(ExecutionMode::Unary)); + assert!(!Protocol::Graphql.supports_execution_mode(ExecutionMode::Window)); + assert!(!Protocol::Graphql.supports_execution_mode(ExecutionMode::Session)); + assert!(!Protocol::Graphql.supports_transport_behavior(TransportBehavior::ServerStream)); + } + + #[test] + fn grpc_supports_session_but_not_deferred_result() { + assert!(Protocol::Grpc.supports_execution_mode(ExecutionMode::Session)); + assert!(!Protocol::Grpc.supports_transport_behavior(TransportBehavior::DeferredResult)); + } +} diff --git a/crates/crank-core/src/stream_session.rs b/crates/crank-core/src/stream_session.rs new file mode 100644 index 0000000..b510936 --- /dev/null +++ b/crates/crank-core/src/stream_session.rs @@ -0,0 +1,210 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::{ + ids::{AgentId, AsyncJobId, OperationId, StreamSessionId, WorkspaceId}, + protocol::Protocol, + streaming::ExecutionMode, +}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StreamStatus { + Created, + Running, + Stopped, + Failed, + Expired, +} + +impl StreamStatus { + pub fn is_terminal(self) -> bool { + matches!(self, Self::Stopped | Self::Failed | Self::Expired) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct StreamSession { + pub id: StreamSessionId, + pub workspace_id: WorkspaceId, + #[serde(skip_serializing_if = "Option::is_none")] + pub agent_id: Option, + pub operation_id: OperationId, + pub protocol: Protocol, + pub mode: ExecutionMode, + pub status: StreamStatus, + #[serde(skip_serializing_if = "Option::is_none")] + pub cursor: Option, + pub state: Value, + pub expires_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_poll_at: Option, + pub created_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub closed_at: Option, +} + +impl StreamSession { + pub fn is_expired(&self, now: &str) -> bool { + self.expires_at.as_str() <= now + } + + pub fn can_poll(&self, now: &str) -> bool { + !self.status.is_terminal() && !self.is_expired(now) + } + + pub fn mark_polled(&mut self, now: impl Into) { + self.last_poll_at = Some(now.into()); + } + + pub fn mark_closed(&mut self, now: impl Into) { + let now = now.into(); + self.status = StreamStatus::Stopped; + self.last_poll_at = Some(now.clone()); + self.closed_at = Some(now); + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum JobStatus { + Created, + Running, + Completed, + Failed, + Cancelled, + Expired, +} + +impl JobStatus { + pub fn is_terminal(self) -> bool { + matches!( + self, + Self::Completed | Self::Failed | Self::Cancelled | Self::Expired + ) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct AsyncJobHandle { + pub id: AsyncJobId, + pub workspace_id: WorkspaceId, + #[serde(skip_serializing_if = "Option::is_none")] + pub agent_id: Option, + pub operation_id: OperationId, + pub status: JobStatus, + pub progress: Value, + #[serde(skip_serializing_if = "Option::is_none")] + pub result: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub expires_at: Option, + pub created_at: String, + pub updated_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub finished_at: Option, +} + +impl AsyncJobHandle { + pub fn is_finished(&self) -> bool { + self.status.is_terminal() + } + + pub fn can_cancel(&self) -> bool { + matches!(self.status, JobStatus::Created | JobStatus::Running) + } + + pub fn mark_finished(&mut self, now: impl Into, result: Value) { + let now = now.into(); + self.status = JobStatus::Completed; + self.result = Some(result); + self.error = None; + self.updated_at = now.clone(); + self.finished_at = Some(now); + } + + pub fn mark_failed(&mut self, now: impl Into, error: Value) { + let now = now.into(); + self.status = JobStatus::Failed; + self.error = Some(error); + self.updated_at = now.clone(); + self.finished_at = Some(now); + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::{AsyncJobHandle, JobStatus, StreamSession, StreamStatus}; + use crate::{ + ids::{AsyncJobId, OperationId, StreamSessionId, WorkspaceId}, + protocol::Protocol, + streaming::ExecutionMode, + }; + + #[test] + fn stream_session_tracks_poll_and_close_transitions() { + let mut session = StreamSession { + id: StreamSessionId::new("stream_01"), + workspace_id: WorkspaceId::new("ws_01"), + agent_id: None, + operation_id: OperationId::new("op_01"), + protocol: Protocol::Rest, + mode: ExecutionMode::Session, + status: StreamStatus::Running, + cursor: None, + state: json!({"cursor":"abc"}), + expires_at: "2026-04-06T12:05:00Z".to_owned(), + last_poll_at: None, + created_at: "2026-04-06T12:00:00Z".to_owned(), + closed_at: None, + }; + + assert!(session.can_poll("2026-04-06T12:01:00Z")); + + session.mark_polled("2026-04-06T12:01:00Z"); + session.mark_closed("2026-04-06T12:02:00Z"); + + assert_eq!(session.status, StreamStatus::Stopped); + assert_eq!(session.closed_at.as_deref(), Some("2026-04-06T12:02:00Z")); + assert!(!session.can_poll("2026-04-06T12:03:00Z")); + } + + #[test] + fn async_job_tracks_finish_and_failure() { + let mut job = AsyncJobHandle { + id: AsyncJobId::new("job_01"), + workspace_id: WorkspaceId::new("ws_01"), + agent_id: None, + operation_id: OperationId::new("op_01"), + status: JobStatus::Running, + progress: json!({"percent": 60}), + result: None, + error: None, + expires_at: Some("2026-04-06T12:05:00Z".to_owned()), + created_at: "2026-04-06T12:00:00Z".to_owned(), + updated_at: "2026-04-06T12:00:00Z".to_owned(), + finished_at: None, + }; + + assert!(job.can_cancel()); + + job.mark_finished("2026-04-06T12:01:00Z", json!({"ok": true})); + assert!(job.is_finished()); + assert_eq!(job.status, JobStatus::Completed); + + let mut failed_job = job.clone(); + failed_job.status = JobStatus::Running; + failed_job.result = None; + failed_job.finished_at = None; + + failed_job.mark_failed("2026-04-06T12:02:00Z", json!({"message": "boom"})); + assert_eq!(failed_job.status, JobStatus::Failed); + assert_eq!( + failed_job.finished_at.as_deref(), + Some("2026-04-06T12:02:00Z") + ); + } +} diff --git a/crates/crank-core/src/streaming.rs b/crates/crank-core/src/streaming.rs new file mode 100644 index 0000000..2ff562e --- /dev/null +++ b/crates/crank-core/src/streaming.rs @@ -0,0 +1,364 @@ +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::protocol::Protocol; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ExecutionMode { + Unary, + Window, + Session, + AsyncJob, +} + +impl ExecutionMode { + pub fn is_stateful(self) -> bool { + matches!(self, Self::Session | Self::AsyncJob) + } + + pub fn requires_tool_family(self) -> bool { + self.is_stateful() + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AggregationMode { + RawItems, + SummaryOnly, + SummaryPlusSamples, + Stats, + LatestState, +} + +impl AggregationMode { + pub fn needs_items(self) -> bool { + matches!(self, Self::RawItems | Self::SummaryPlusSamples) + } + + pub fn needs_summary(self) -> bool { + !matches!(self, Self::RawItems) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TransportBehavior { + RequestResponse, + ServerStream, + StatefulSession, + DeferredResult, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)] +pub struct ToolFamilyConfig { + #[serde(skip_serializing_if = "Option::is_none")] + pub start_tool_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub poll_tool_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub stop_tool_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub status_tool_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub result_tool_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cancel_tool_name: Option, +} + +impl ToolFamilyConfig { + pub fn validate_for_mode(&self, mode: ExecutionMode) -> Result<(), StreamingConfigError> { + match mode { + ExecutionMode::Unary | ExecutionMode::Window => { + if self.start_tool_name.is_some() + || self.poll_tool_name.is_some() + || self.stop_tool_name.is_some() + || self.status_tool_name.is_some() + || self.result_tool_name.is_some() + || self.cancel_tool_name.is_some() + { + return Err(StreamingConfigError::UnexpectedToolFamily(mode)); + } + } + ExecutionMode::Session => { + if self.start_tool_name.is_none() + || self.poll_tool_name.is_none() + || self.stop_tool_name.is_none() + { + return Err(StreamingConfigError::MissingSessionToolNames); + } + } + ExecutionMode::AsyncJob => { + if self.start_tool_name.is_none() + || self.status_tool_name.is_none() + || self.result_tool_name.is_none() + || self.cancel_tool_name.is_none() + { + return Err(StreamingConfigError::MissingAsyncJobToolNames); + } + } + } + + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct StreamingConfig { + pub mode: ExecutionMode, + pub transport_behavior: TransportBehavior, + #[serde(skip_serializing_if = "Option::is_none")] + pub window_duration_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub poll_interval_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub upstream_timeout_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub idle_timeout_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_session_lifetime_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_items: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_bytes: Option, + pub aggregation_mode: AggregationMode, + #[serde(skip_serializing_if = "Option::is_none")] + pub summary_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub items_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cursor_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub status_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub done_path: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub redacted_paths: Vec, + #[serde(default)] + pub truncate_item_fields: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_field_length: Option, + #[serde(default)] + pub drop_duplicates: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub sampling_rate: Option, + #[serde(default)] + pub tool_family: ToolFamilyConfig, +} + +impl StreamingConfig { + pub fn validate_common(&self) -> Result<(), StreamingConfigError> { + self.tool_family.validate_for_mode(self.mode)?; + + if self.max_items.is_some_and(|value| value == 0) { + return Err(StreamingConfigError::InvalidMaxItems); + } + + if self.max_bytes.is_some_and(|value| value == 0) { + return Err(StreamingConfigError::InvalidMaxBytes); + } + + if self.max_field_length.is_some_and(|value| value == 0) { + return Err(StreamingConfigError::InvalidMaxFieldLength); + } + + if self + .sampling_rate + .is_some_and(|value| value <= 0.0 || value > 1.0) + { + return Err(StreamingConfigError::InvalidSamplingRate); + } + + match self.mode { + ExecutionMode::Unary => { + if self.window_duration_ms.is_some() + || self.poll_interval_ms.is_some() + || self.idle_timeout_ms.is_some() + || self.max_session_lifetime_ms.is_some() + { + return Err(StreamingConfigError::UnexpectedStatefulLimits( + ExecutionMode::Unary, + )); + } + } + ExecutionMode::Window => { + if self.window_duration_ms.is_none() { + return Err(StreamingConfigError::MissingWindowDuration); + } + } + ExecutionMode::Session => { + if self.poll_interval_ms.is_none() || self.max_session_lifetime_ms.is_none() { + return Err(StreamingConfigError::MissingSessionLimits); + } + } + ExecutionMode::AsyncJob => {} + } + + Ok(()) + } + + pub fn validate_for_protocol(&self, protocol: Protocol) -> Result<(), StreamingConfigError> { + self.validate_common()?; + + if !protocol.supports_execution_mode(self.mode) { + return Err(StreamingConfigError::UnsupportedExecutionMode { + protocol, + mode: self.mode, + }); + } + + if !protocol.supports_transport_behavior(self.transport_behavior) { + return Err(StreamingConfigError::UnsupportedTransportBehavior { + protocol, + behavior: self.transport_behavior, + }); + } + + Ok(()) + } +} + +#[derive(Clone, Debug, Error, PartialEq)] +pub enum StreamingConfigError { + #[error("window mode requires window_duration_ms")] + MissingWindowDuration, + #[error("session mode requires poll_interval_ms and max_session_lifetime_ms")] + MissingSessionLimits, + #[error("max_items must be greater than zero")] + InvalidMaxItems, + #[error("max_bytes must be greater than zero")] + InvalidMaxBytes, + #[error("max_field_length must be greater than zero")] + InvalidMaxFieldLength, + #[error("sampling_rate must be in range (0, 1]")] + InvalidSamplingRate, + #[error("{0:?} mode cannot use session/window-only limits")] + UnexpectedStatefulLimits(ExecutionMode), + #[error("{mode:?} is not supported for protocol {protocol:?}")] + UnsupportedExecutionMode { + protocol: Protocol, + mode: ExecutionMode, + }, + #[error("{behavior:?} is not supported for protocol {protocol:?}")] + UnsupportedTransportBehavior { + protocol: Protocol, + behavior: TransportBehavior, + }, + #[error("session mode requires start, poll and stop tool names")] + MissingSessionToolNames, + #[error("async_job mode requires start, status, result and cancel tool names")] + MissingAsyncJobToolNames, + #[error("{0:?} mode cannot define tool-family names")] + UnexpectedToolFamily(ExecutionMode), +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::{ + AggregationMode, ExecutionMode, StreamingConfig, StreamingConfigError, ToolFamilyConfig, + TransportBehavior, + }; + use crate::protocol::Protocol; + + fn session_config() -> StreamingConfig { + StreamingConfig { + mode: ExecutionMode::Session, + transport_behavior: TransportBehavior::StatefulSession, + window_duration_ms: None, + poll_interval_ms: Some(1_000), + upstream_timeout_ms: Some(5_000), + idle_timeout_ms: Some(15_000), + max_session_lifetime_ms: Some(60_000), + max_items: Some(100), + max_bytes: Some(65_536), + aggregation_mode: AggregationMode::SummaryPlusSamples, + summary_path: Some("$.summary".to_owned()), + items_path: Some("$.items".to_owned()), + cursor_path: Some("$.cursor".to_owned()), + status_path: Some("$.status".to_owned()), + done_path: Some("$.done".to_owned()), + redacted_paths: vec!["$.items[*].token".to_owned()], + truncate_item_fields: true, + max_field_length: Some(256), + drop_duplicates: true, + sampling_rate: Some(0.5), + tool_family: ToolFamilyConfig { + start_tool_name: Some("logs_start".to_owned()), + poll_tool_name: Some("logs_poll".to_owned()), + stop_tool_name: Some("logs_stop".to_owned()), + status_tool_name: None, + result_tool_name: None, + cancel_tool_name: None, + }, + } + } + + #[test] + fn streaming_config_roundtrips_through_json() { + let config = session_config(); + + let value = serde_json::to_value(&config).unwrap(); + let decoded: StreamingConfig = serde_json::from_value(value.clone()).unwrap(); + + assert_eq!(decoded, config); + assert_eq!(value["mode"], json!("session")); + assert_eq!(value["tool_family"]["start_tool_name"], json!("logs_start")); + } + + #[test] + fn unary_mode_rejects_session_specific_fields() { + let config = StreamingConfig { + mode: ExecutionMode::Unary, + transport_behavior: TransportBehavior::RequestResponse, + window_duration_ms: None, + poll_interval_ms: Some(1_000), + upstream_timeout_ms: None, + idle_timeout_ms: None, + max_session_lifetime_ms: None, + max_items: None, + max_bytes: None, + aggregation_mode: AggregationMode::SummaryOnly, + 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: ToolFamilyConfig::default(), + }; + + assert_eq!( + config.validate_common(), + Err(StreamingConfigError::UnexpectedStatefulLimits( + ExecutionMode::Unary + )) + ); + } + + #[test] + fn protocol_validation_rejects_graphql_session() { + let config = session_config(); + + assert_eq!( + config.validate_for_protocol(Protocol::Graphql), + Err(StreamingConfigError::UnsupportedExecutionMode { + protocol: Protocol::Graphql, + mode: ExecutionMode::Session, + }) + ); + } + + #[test] + fn protocol_validation_accepts_rest_session() { + let config = session_config(); + + assert!(config.validate_for_protocol(Protocol::Rest).is_ok()); + } +} diff --git a/crates/crank-registry/src/postgres.rs b/crates/crank-registry/src/postgres.rs index ca6d103..1344383 100644 --- a/crates/crank-registry/src/postgres.rs +++ b/crates/crank-registry/src/postgres.rs @@ -4147,6 +4147,7 @@ mod tests { auth_profile_ref: Some("auth_crank".into()), headers: BTreeMap::from([("X-Request-Id".to_owned(), "static".to_owned())]), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Create lead".to_owned(), diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index 08ffbff..32e5d94 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -474,6 +474,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Create Lead".to_owned(), @@ -549,6 +550,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Create Lead GraphQL".to_owned(), @@ -623,6 +625,7 @@ mod tests { auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, + streaming: None, }, tool_description: ToolDescription { title: "Unary Echo gRPC".to_owned(),