feat: add streaming core domain model

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
 

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(),

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(),

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(),

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);

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};

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<GrpcProtocolOptions>,
 }
 
-#[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<String, String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub protocol_options: Option<ProtocolOptions>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub streaming: Option<StreamingConfig>,
 }
 
 #[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<TSchema, TMapping> {
     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(),

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));
+    }
+}

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<AgentId>,
+    pub operation_id: OperationId,
+    pub protocol: Protocol,
+    pub mode: ExecutionMode,
+    pub status: StreamStatus,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cursor: Option<Value>,
+    pub state: Value,
+    pub expires_at: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub last_poll_at: Option<String>,
+    pub created_at: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub closed_at: Option<String>,
+}
+
+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<String>) {
+        self.last_poll_at = Some(now.into());
+    }
+
+    pub fn mark_closed(&mut self, now: impl Into<String>) {
+        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<AgentId>,
+    pub operation_id: OperationId,
+    pub status: JobStatus,
+    pub progress: Value,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result: Option<Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub expires_at: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub finished_at: Option<String>,
+}
+
+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<String>, 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<String>, 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")
+        );
+    }
+}

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<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub poll_tool_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stop_tool_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub status_tool_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result_tool_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cancel_tool_name: Option<String>,
+}
+
+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<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub poll_interval_ms: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub upstream_timeout_ms: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub idle_timeout_ms: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_session_lifetime_ms: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_items: Option<u32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_bytes: Option<u32>,
+    pub aggregation_mode: AggregationMode,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub summary_path: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub items_path: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cursor_path: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub status_path: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub done_path: Option<String>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub redacted_paths: Vec<String>,
+    #[serde(default)]
+    pub truncate_item_fields: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_field_length: Option<u32>,
+    #[serde(default)]
+    pub drop_duplicates: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sampling_rate: Option<f64>,
+    #[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());
+    }
+}

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(),

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(),