# Streaming Runtime Design ## 1. Назначение документа Этот документ фиксирует, как потоковая модель должна быть разложена по crates, traits, services и функциям. Цель: - избежать god-services; - зафиксировать boundaries между `core`, `registry`, `runtime`, protocol adapters и `mcp-server`; - расписать ожидаемые функции и их ответственность до уровня инженерной спецификации. ## 2. Общая схема слоев ### 2.1. `crank-core` Отвечает за: - типы `ExecutionMode`, `StreamingConfig`, `AggregationMode`; - типы `StreamSession`, `AsyncJobHandle`, `StreamStatus`, `JobStatus`; - validation helpers без инфраструктуры. ### 2.2. `crank-registry` Отвечает за: - persistence session/job state; - optimistic transitions; - cleanup expired state. ### 2.3. `crank-runtime` Отвечает за: - orchestration execution modes; - dispatch в protocol adapter; - bounded collection; - aggregation; - state transitions; - invocation logging. ### 2.4. Protocol adapters Отвечают за: - protocol-specific upstream behavior; - connect/request/collect/close; - преобразование upstream payload в нормализованный JSON. ### 2.5. `apps/mcp-server` Отвечает за: - downstream MCP transport; - tool-family generation; - JSON-RPC lifecycle; - transport/session routing. ### 2.6. `apps/admin-api` Отвечает за: - config CRUD; - test-runs; - validation; - UI-oriented metadata and previews. ## 3. Core domain types ## 3.1. `ExecutionMode` ```rust pub enum ExecutionMode { Unary, Window, Session, AsyncJob, } ``` Методы: - `fn is_stateful(&self) -> bool` - `fn requires_tool_family(&self) -> bool` ## 3.2. `AggregationMode` ```rust pub enum AggregationMode { RawItems, SummaryOnly, SummaryPlusSamples, Stats, LatestState, } ``` Методы: - `fn needs_items(&self) -> bool` - `fn needs_summary(&self) -> bool` ## 3.3. `StreamingConfig` ```rust pub struct StreamingConfig { pub mode: ExecutionMode, pub transport_behavior: TransportBehavior, pub window_duration_ms: Option, pub poll_interval_ms: Option, pub upstream_timeout_ms: Option, pub idle_timeout_ms: Option, pub max_session_lifetime_ms: Option, pub max_items: Option, pub max_bytes: Option, pub aggregation_mode: AggregationMode, pub summary_path: Option, pub items_path: Option, pub cursor_path: Option, pub status_path: Option, pub done_path: Option, pub redacted_paths: Vec, pub truncate_item_fields: bool, pub max_field_length: Option, pub drop_duplicates: bool, pub sampling_rate: Option, pub tool_family: ToolFamilyConfig, } ``` Методы: - `fn validate_common(&self) -> Result<(), StreamingConfigError>` - `fn validate_for_protocol(&self, protocol: Protocol) -> Result<(), StreamingConfigError>` - `fn effective_max_items(&self, limits: &StreamingLimits) -> u32` - `fn effective_max_bytes(&self, limits: &StreamingLimits) -> u32` - `fn effective_timeout(&self, defaults: &StreamingDefaults) -> Duration` ## 3.4. `StreamSession` ```rust pub struct StreamSession { pub id: StreamSessionId, pub workspace_id: WorkspaceId, pub agent_id: Option, pub operation_id: OperationId, pub protocol: Protocol, pub mode: ExecutionMode, pub status: StreamStatus, pub cursor: Option, pub state: Value, pub expires_at: DateTime, pub last_poll_at: Option>, pub created_at: DateTime, pub closed_at: Option>, } ``` Методы: - `fn is_expired(&self, now: DateTime) -> bool` - `fn can_poll(&self, now: DateTime) -> bool` - `fn mark_polled(&mut self, now: DateTime)` - `fn mark_closed(&mut self, now: DateTime)` ## 3.5. `AsyncJobHandle` ```rust pub struct AsyncJobHandle { pub id: AsyncJobId, pub workspace_id: WorkspaceId, pub agent_id: Option, pub operation_id: OperationId, pub status: JobStatus, pub progress: Value, pub result: Option, pub error: Option, pub expires_at: Option>, pub created_at: DateTime, pub updated_at: DateTime, pub finished_at: Option>, } ``` Методы: - `fn is_finished(&self) -> bool` - `fn can_cancel(&self) -> bool` - `fn mark_finished(&mut self, now: DateTime, result: Value)` - `fn mark_failed(&mut self, now: DateTime, error: Value)` ## 4. Registry contracts ## 4.1. `StreamSessionRepository` ```rust #[async_trait] pub trait StreamSessionRepository { async fn create_stream_session(&self, session: NewStreamSession) -> Result; async fn get_stream_session(&self, id: &StreamSessionId) -> Result, RegistryError>; async fn update_stream_session_state(&self, update: StreamSessionStateUpdate) -> Result; async fn close_stream_session(&self, id: &StreamSessionId, now: DateTime) -> Result<(), RegistryError>; async fn list_stream_sessions(&self, filter: StreamSessionFilter) -> Result, RegistryError>; async fn delete_expired_stream_sessions(&self, now: DateTime) -> Result; } ``` ## 4.2. `AsyncJobRepository` ```rust #[async_trait] pub trait AsyncJobRepository { async fn create_async_job(&self, job: NewAsyncJob) -> Result; async fn get_async_job(&self, id: &AsyncJobId) -> Result, RegistryError>; async fn update_async_job_status(&self, update: AsyncJobStatusUpdate) -> Result; async fn cancel_async_job(&self, id: &AsyncJobId, now: DateTime) -> Result<(), RegistryError>; async fn list_async_jobs(&self, filter: AsyncJobFilter) -> Result, RegistryError>; async fn delete_expired_async_jobs(&self, now: DateTime) -> Result; } ``` ## 4.3. Storage rules - update methods должны использовать optimistic concurrency, если state transitions конфликтуют; - `poll` не должен silently reopen closed session; - cleanup job должен быть отдельным service; - session/job state не должен хранить необрезанные raw payloads без лимитов. ## 5. Runtime contracts ## 5.1. `ProtocolAdapter` Базовый trait не должен пытаться описать все streaming cases в одном методе. ```rust #[async_trait] pub trait ProtocolAdapter { async fn execute_unary(&self, ctx: UnaryExecutionContext) -> Result; async fn execute_window(&self, ctx: WindowExecutionContext) -> Result; async fn start_session(&self, ctx: SessionStartContext) -> Result; async fn poll_session(&self, ctx: SessionPollContext) -> Result; async fn stop_session(&self, ctx: SessionStopContext) -> Result<(), AdapterError>; async fn start_async_job(&self, ctx: AsyncJobStartContext) -> Result; async fn get_async_job_status(&self, ctx: AsyncJobStatusContext) -> Result; async fn get_async_job_result(&self, ctx: AsyncJobResultContext) -> Result; async fn cancel_async_job(&self, ctx: AsyncJobCancelContext) -> Result<(), AdapterError>; } ``` Не каждый adapter обязан поддерживать все методы. Capability mismatch должен проверяться выше. ## 5.2. `OperationExecutor` Главная orchestration точка в `crank-runtime`. Ожидаемые функции: - `execute_unary_operation` - `execute_window_operation` - `start_stream_session` - `poll_stream_session` - `stop_stream_session` - `start_async_job` - `get_async_job_status` - `get_async_job_result` - `cancel_async_job` ### `execute_unary_operation` Отвечает за: - schema validation; - auth resolution; - adapter dispatch; - output mapping; - observability write. ### `execute_window_operation` Отвечает за: - schema validation; - auth resolution; - adapter dispatch в `execute_window`; - aggregation; - truncation flags; - observability write. ### `start_stream_session` Отвечает за: - schema validation; - auth resolution; - adapter `start_session`; - registry `create_stream_session`; - initial preview result; - observability write. ### `poll_stream_session` Отвечает за: - load session from registry; - expiration check; - adapter `poll_session`; - registry `update_stream_session_state`; - output mapping; - observability write. ### `stop_stream_session` Отвечает за: - load session; - adapter `stop_session`; - registry close; - observability write. ### `start_async_job` Отвечает за: - schema validation; - adapter `start_async_job`; - registry `create_async_job`; - observability write. ### `get_async_job_status` Отвечает за: - registry load; - adapter `get_async_job_status`, если upstream status is live; - registry update; - normalized status result. ### `get_async_job_result` Отвечает за: - job readiness check; - adapter `get_async_job_result`, если result lazy-loaded; - output mapping; - observability write. ### `cancel_async_job` Отвечает за: - adapter cancel; - registry cancel transition; - observability write. ## 5.3. Aggregation services Нужно отделить aggregation от adapters. Отдельные services: - `WindowAggregator` - `SummaryBuilder` - `CursorTracker` - `PayloadLimiter` - `RedactionService` ### `WindowAggregator` Функции: - `collect_items` - `apply_item_limit` - `apply_byte_limit` - `mark_truncated` ### `SummaryBuilder` Функции: - `build_summary` - `build_stats_summary` - `build_latest_state_summary` - `build_summary_plus_samples` ### `PayloadLimiter` Функции: - `truncate_item_fields` - `truncate_bytes` - `enforce_max_items` ### `RedactionService` Функции: - `redact_paths` - `redact_object` - `redact_item` ## 6. Adapter responsibilities ## 6.1. REST adapter Функции: - `execute_rest_unary` - `execute_rest_window` - `start_rest_session` - `poll_rest_session` - `stop_rest_session` - `start_rest_async_job` - `get_rest_async_job_status` - `get_rest_async_job_result` - `cancel_rest_async_job` Дополнительные helpers: - `open_sse_stream` - `collect_sse_window` - `parse_sse_event` - `close_sse_stream` ## 6.2. GraphQL adapter Функции: - `execute_graphql_unary` Отдельно отложено: - `execute_graphql_subscription_window` - `start_graphql_subscription_session` ## 6.3. gRPC adapter Функции: - `execute_grpc_unary` - `execute_grpc_window` - `start_grpc_session` - `poll_grpc_session` - `stop_grpc_session` - `start_grpc_async_job` - `get_grpc_async_job_status` - `get_grpc_async_job_result` - `cancel_grpc_async_job` Helpers: - `invoke_unary_method` - `open_server_stream` - `collect_server_stream_window` - `decode_stream_item` ## 6.4. WebSocket adapter Функции: - `execute_websocket_window` - `start_websocket_session` - `poll_websocket_session` - `stop_websocket_session` - `start_websocket_async_job` - `get_websocket_async_job_status` - `cancel_websocket_async_job` Helpers: - `connect_websocket` - `send_subscribe_message` - `send_unsubscribe_message` - `read_next_frame` - `decode_text_frame` - `heartbeat_tick` - `reconnect_if_needed` ## 6.5. SOAP adapter Функции: - `execute_soap_unary` - `start_soap_async_job` - `get_soap_async_job_status` - `get_soap_async_job_result` - `cancel_soap_async_job` Helpers: - `render_soap_envelope` - `render_soap_headers` - `parse_soap_envelope` - `parse_soap_fault` - `normalize_xml_value` ## 7. MCP server design ## 7.1. Route handlers Ожидаемые handlers: - `handle_initialize` - `handle_notifications_initialized` - `handle_ping` - `handle_tools_list` - `handle_tools_call` - `handle_get_sse_stream` - `handle_delete_session` ## 7.2. Tool-family generation Функции: - `build_unary_tool_definition` - `build_window_tool_definition` - `build_session_tool_family` - `build_async_job_tool_family` Для `session`: - `build_session_start_tool` - `build_session_poll_tool` - `build_session_stop_tool` Для `async_job`: - `build_async_job_start_tool` - `build_async_job_status_tool` - `build_async_job_result_tool` - `build_async_job_cancel_tool` ## 7.3. Transport session management Функции: - `ensure_protocol_version` - `resolve_transport_mode` - `ensure_mcp_session` - `attach_session_headers` - `stream_sse_event` - `close_transport_session` ## 8. Admin API service design Ожидаемые service функции: - `validate_streaming_config` - `list_protocol_capabilities` - `list_streaming_presets` - `start_stream_test_run` - `poll_stream_test_run` - `stop_stream_test_run` - `get_stream_test_result` - `list_stream_sessions` - `get_stream_session` - `stop_stream_session` - `delete_stream_session` - `list_async_jobs` - `get_async_job` - `cancel_async_job` - `get_async_job_result` ## 9. Cleanup jobs Отдельные jobs: - `expire_stream_sessions` - `expire_async_jobs` - `reap_orphaned_transport_sessions` - `compact_stream_payloads`, если будет нужен storage optimization layer ## 10. Design rules - adapters не агрегируют product-level summaries; - runtime не знает деталей transport framing; - `mcp-server` не знает про upstream protocols; - `admin-api` не управляет live protocol connections напрямую; - session/job lifecycle должен быть явным в names и contracts; - cancellation не должна зависеть от TCP disconnect; - bounded limits должны применяться раньше, чем payload попадет в final result.