# Streaming MCP Plan

## 1. Назначение документа

Этот документ фиксирует целевую архитектуру потоковой обработки в Crank. Он описывает:

- какие transport- и upstream-протоколы поддерживаются;
- какие streaming-сценарии считаются допустимыми;
- какие ограничения обязательны для безопасности и управляемости;
- какие сущности, API, UI-поля и runtime-механизмы нужно добавить;
- в каком порядке это реализовывать.

Документ синхронизирован с MCP transport specification `2025-06-18`, где `Streamable HTTP` определен как основной HTTP transport, а SSE допускается как часть `POST`-response и как отдельный `GET`-stream.

## 2. Базовое архитектурное решение

Crank поддерживает streaming не как бесконечный текстовый поток в чат, а как управляемую инструментальную модель поверх MCP tools.

Принцип:

- transport может быть long-lived;
- tool contract обязан оставаться ограниченным и управляемым;
- upstream stream всегда преобразуется в bounded result, session poll или async job;
- оператор в UI настраивает не "трубу", а режим сбора, агрегации и завершения.

## 3. Поддерживаемые transport- и upstream-протоколы

### 3.1. Downstream MCP transport

Поддерживается:

- `Streamable HTTP` как канонический network transport;
- HTTP `POST` с ответом `application/json` или `text/event-stream`;
- HTTP `GET` для server-to-client SSE stream;
- несколько SSE streams одновременно в рамках одного MCP session;
- `Mcp-Session-Id` и `MCP-Protocol-Version` headers;
- resumable SSE streams как опциональная возможность.

Не является текущим приоритетом:

- `stdio` как обязательная часть продукта;
- собственные нестандартные transport-режимы поверх MCP.

### 3.2. Upstream protocols

Поддерживаются:

- REST unary;
- REST SSE в bounded режимах;
- GraphQL `query` и `mutation`;
- gRPC unary;
- gRPC server-streaming в bounded режимах;
- WebSocket в bounded режимах.

Отложено:

- GraphQL `subscription`;
- gRPC client-streaming;
- gRPC bidirectional streaming;
- arbitrary websocket passthrough;
- raw infinite stream forwarding в MCP client.

### 3.3. Protocol capability matrix

| Protocol | Unary | Window | Session | Async Job | Notes |
| --- | --- | --- | --- | --- | --- |
| REST | Yes | Yes | Limited | Yes | SSE and long-poll sources are supported in controlled form |
| GraphQL | Yes | No | No | Limited | `query` and `mutation` only; `subscription` is future scope |
| gRPC | Yes | Yes | Yes | Yes | server-streaming only; client/bidi deferred |
| WebSocket | No | Yes | Yes | Yes | upstream adapter only; not downstream MCP transport |
| SOAP | Yes | Limited | Limited | Yes | primarily request-response enterprise workflows |

## 4. Поддерживаемые streaming modes

Crank поддерживает четыре режима выполнения operation.

### 4.1. `unary`

Обычный request-response вызов.

Подходит для:

- REST;
- GraphQL `query` и `mutation`;
- gRPC unary;
- SOAP.

### 4.2. `window`

Runtime открывает upstream stream или repeatedly polls upstream source, собирает данные в пределах окна и возвращает один bounded ответ.

Параметры:

- `window_duration_ms`
- `max_items`
- `max_bytes`
- `upstream_timeout_ms`
- `aggregation_mode`

Подходит для:

- логи за период;
- метрики за период;
- event window;
- SSE stream snapshot;
- gRPC server-stream window;
- WebSocket event window.

### 4.3. `session`

Runtime создает stream session, после чего данные читаются по шагам через session-oriented tool family.

Обязательные операции:

- `start`
- `poll`
- `stop`

Подходит для:

- follow logs;
- telemetry follow;
- alert/event feed;
- контроль длительных stream-подписок;
- WebSocket subscriptions.

### 4.4. `async_job`

Runtime запускает long-running upstream operation и возвращает `job_id`.

Обязательные операции:

- `start`
- `status`
- `result`
- `cancel`

Подходит для:

- import/export;
- deploy/reindex;
- batch processing;
- инфраструктурные control-plane действия;
- SOAP workflows with deferred status polling.

## 5. Бизнес-кейсы

### 5.1. Logs Window

LLM запрашивает:

- ошибки сервиса за последние `30s`;
- top errors за последние `100` записей;
- логи по конкретному `correlation_id`.

Runtime:

- собирает bounded окно;
- агрегирует counts, уровни, sample lines;
- возвращает summary плюс ограниченный список items.

### 5.2. Metrics Window

LLM запрашивает:

- latency/error summary по сервису;
- CPU/memory snapshot;
- anomaly summary за окно.

Runtime:

- читает поток метрик;
- агрегирует min/max/avg/p95 или anomaly set;
- возвращает компактный JSON.

### 5.3. Event Feed

LLM запрашивает:

- audit events за период;
- queue events по фильтру;
- security alerts за окно.

Runtime:

- фильтрует events;
- ограничивает количество;
- возвращает items plus cursor.

### 5.4. Long-running Operation Status

LLM запускает:

- reindex;
- import job;
- rollout;
- repair task.

Runtime:

- создает job handle;
- возвращает `job_id`;
- позволяет дальше получать status/result/cancel.

### 5.5. Control Plane Follow

LLM инициирует:

- reboot;
- rollout;
- node drain;
- workflow transition.

Runtime:

- стартует действие;
- пишет progress в session/job state;
- возвращает snapshots по `poll`.

### 5.6. WebSocket Realtime Feeds

LLM запрашивает:

- realtime alert snapshot;
- device telemetry window;
- market data slice;
- status feed по subscription channel.

Runtime:

- открывает upstream WebSocket;
- подписывается на channel;
- собирает bounded окно или session step;
- возвращает summary и limited items.

### 5.7. SOAP Enterprise Operations

LLM запрашивает:

- создание/поиск сущности в ERP;
- запуск enterprise workflow;
- получение статуса batch operation;
- B2B request через SOAP gateway.

Runtime:

- строит SOAP envelope из MCP input;
- вызывает enterprise endpoint;
- нормализует XML response или SOAP Fault;
- возвращает JSON-oriented output.

## 6. Функциональные требования

### 6.1. Общие

- operation должна явно указывать `execution_mode`;
- streaming operation обязана быть bounded;
- runtime обязан поддерживать timeout, max items и max bytes;
- tool output обязан иметь предсказуемую схему;
- stream/session/job state должен быть наблюдаемым и логируемым;
- cancel/stop должен быть явной операцией, а не побочным эффектом disconnect.

### 6.2. Для `window`

- задать размер окна;
- задать лимит items;
- задать лимит bytes;
- выбрать aggregation mode;
- вернуть `truncated` и `window_complete` flags;
- поддерживать optional cursor для следующего окна.

### 6.3. Для `session`

- создать `session_id`;
- поддерживать `poll`;
- поддерживать `stop`;
- хранить курсор и session status;
- иметь `idle_timeout`;
- иметь `max_session_lifetime`;
- удалять expired sessions.

### 6.4. Для `async_job`

- создать `job_id`;
- хранить progress, status и final result metadata;
- поддерживать `cancel`;
- поддерживать retrieval последнего готового результата.

## 7. Нефункциональные требования

### 7.1. Безопасность

- никакого неограниченного passthrough потока;
- обязательные лимиты по времени, items и bytes;
- обязательный redact layer для secret-bearing полей;
- audit trail на `start`, `poll`, `stop`, `cancel`;
- session и job identifiers должны быть криптографически стойкими.

### 7.2. Производительность

- bounded memory per session;
- bounded upstream read buffer;
- ограничение числа параллельных sessions и jobs на workspace и agent;
- backpressure при медленных клиентах;
- возможность early cut-off после достижения лимита.

### 7.3. Надежность

- TTL и cleanup для sessions/jobs;
- устойчивость к disconnect downstream client;
- poll должен быть идемпотентным;
- long-running upstream action не должен считаться отмененным из-за SSE disconnect;
- resumability для downstream SSE допускается как следующая волна, но не является обязательной в первой реализации.

### 7.4. UX

- оператор должен видеть, что operation является `unary`, `window`, `session` или `async_job`;
- UI должен явно показывать все лимиты и режим агрегации;
- результат тестового вызова должен показывать `truncated`, `window_complete`, `has_more`, `status`.

## 8. UI contract

### 8.1. Новый блок `Execution mode`

Поля:

- `mode`: `unary | window | session | async_job`
- `transport_behavior`: `request_response | server_stream`

### 8.2. Блок `Collection limits`

Поля:

- `window_duration_ms`
- `poll_interval_ms`
- `upstream_timeout_ms`
- `idle_timeout_ms`
- `max_session_lifetime_ms`
- `max_items`
- `max_bytes`

### 8.3. Блок `Aggregation`

Поля:

- `aggregation_mode`: `raw_items | summary_only | summary_plus_samples | stats | latest_state`
- `summary_path`
- `items_path`
- `cursor_path`
- `status_path`
- `done_path`

### 8.4. Блок `Safety`

Поля:

- `truncate_item_fields`
- `max_field_length`
- `redacted_paths`
- `drop_duplicates`
- `sampling_rate`

### 8.5. Block `Tool family`

Для `session`:

- `start_tool_name`
- `poll_tool_name`
- `stop_tool_name`

Для `async_job`:

- `start_tool_name`
- `status_tool_name`
- `result_tool_name`
- `cancel_tool_name`

## 9. Domain model changes

### 9.1. `ExecutionMode`

Новый enum:

- `Unary`
- `Window`
- `Session`
- `AsyncJob`

### 9.2. `StreamingConfig`

Новая часть `execution_config`:

- `mode`
- `window_duration_ms`
- `poll_interval_ms`
- `upstream_timeout_ms`
- `idle_timeout_ms`
- `max_session_lifetime_ms`
- `max_items`
- `max_bytes`
- `aggregation_mode`
- `items_path`
- `summary_path`
- `cursor_path`
- `status_path`
- `done_path`
- `redacted_paths`

### 9.3. `StreamSession`

Новая runtime/store сущность:

- `id`
- `workspace_id`
- `agent_id`
- `operation_id`
- `protocol`
- `mode`
- `status`
- `cursor`
- `state_json`
- `expires_at`
- `last_poll_at`
- `created_at`
- `closed_at`

### 9.4. `AsyncJobHandle`

Новая runtime/store сущность:

- `id`
- `workspace_id`
- `agent_id`
- `operation_id`
- `status`
- `progress_json`
- `result_json`
- `error_json`
- `expires_at`
- `created_at`
- `updated_at`
- `finished_at`

## 10. MCP publishing model for streaming tools

### 10.1. Unary and Window

`unary` и `window` публикуются как один MCP tool:

- один input contract;
- один bounded result;
- transport может использовать `application/json` или SSE response stream до финального JSON-RPC response.

### 10.2. Session

`session` публикуется как tool family:

- `{tool}_start`
- `{tool}_poll`
- `{tool}_stop`

Причина:

- lifecycle становится явным;
- LLM получает контролируемую state machine;
- runtime не скрывает долговременное состояние за одним "магическим" вызовом.

### 10.3. Async Job

`async_job` публикуется как tool family:

- `{tool}_start`
- `{tool}_status`
- `{tool}_result`
- `{tool}_cancel`

## 11. Module decomposition and responsibilities

### 11.1. `crank-core`

Новые модули:

- `streaming`
- `stream_session`

Новые типы:

- `ExecutionMode`
- `StreamingConfig`
- `AggregationMode`
- `StreamSession`
- `AsyncJobHandle`
- `StreamStatus`
- `JobStatus`

### 11.2. `crank-registry`

Новые обязанности:

- хранение `stream_sessions`;
- хранение `async_jobs`;
- cleanup expired rows;
- optimistic updates on poll/stop/cancel.

Ожидаемые функции:

- `create_stream_session`
- `get_stream_session`
- `advance_stream_session`
- `close_stream_session`
- `create_async_job`
- `get_async_job`
- `update_async_job_status`
- `cancel_async_job`
- `delete_expired_stream_sessions`

### 11.3. `crank-runtime`

Новые orchestration функции:

- `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`

### 11.4. Protocol adapters

REST:

- unary HTTP;
- bounded SSE collection;
- bounded long-poll collection.

GraphQL:

- `query` и `mutation`;
- `subscription` отложен на отдельную protocol wave.

gRPC:

- unary;
- bounded server-stream collection;
- client/bidi отложены на отдельную protocol wave.

WebSocket:

- bounded event collection;
- subscribe/poll/stop orchestration;
- heartbeat and reconnect policy.

SOAP:

- WSDL-driven request/response adapter;
- XML normalization;
- SOAP Fault normalization;
- future WS-Security expansion.

### 11.5. `apps/mcp-server`

Новые обязанности:

- корректно вести `Streamable HTTP` lifecycle;
- принимать `POST` с `Accept: application/json, text/event-stream`;
- отдавать `application/json` или `text/event-stream`;
- поддерживать `GET` SSE stream для server-to-client messages и notifications;
- вести `Mcp-Session-Id`;
- публиковать tool families для `session` и `async_job`.

### 11.6. `apps/admin-api`

Новые обязанности:

- CRUD и versioning для streaming config;
- тестовые window/session/job runs;
- UI-oriented validation ошибок для streaming fields.

### 11.7. `apps/ui`

- конфиг execution mode;
- конфиг limits/aggregation/safety;
- test-run screen для bounded window/session/job behavior;
- отдельные предупреждения про truncation и timeouts.

## 12. Границы текущей продуктовой волны

В первой продуктовой волне входит:

- `Streamable HTTP` и SSE на MCP transport;
- `window` mode;
- `async_job` mode;
- REST SSE;
- gRPC server streaming;
- tool family generation;
- bounded session/job state.

Во второй продуктовой волне:

- WebSocket upstream adapter;
- SOAP adapter foundation;
- richer session tooling;
- expanded protocol smoke suite.

Отложено:

- GraphQL subscriptions;
- gRPC client streaming;
- gRPC bidirectional streaming;
- raw infinite stream passthrough;
- guaranteed resumability across all stream types;
- generic websocket proxy mode.

## 13. Порядок реализации

### 13.1. `feat/streaming-mcp-architecture`

- зафиксировать docs;
- обновить protocol support matrix;
- синхронизировать `TASKS.md`.

### 13.2. `feat/mcp-streamable-http-alignment`

- довести `mcp-server` до полного соответствия `Streamable HTTP`;
- session headers;
- GET SSE;
- protocol version header validation;
- explicit cancel behavior.

### 13.3. `feat/streaming-core-model`

- ввести `ExecutionMode`, `StreamingConfig`, `StreamSession`, `AsyncJobHandle`.

### 13.4. `feat/stream-session-store`

- таблицы `stream_sessions` и `async_jobs`;
- cleanup;
- optimistic state transitions.

### 13.5. `feat/runtime-window-mode`

- bounded collection для `window`;
- `truncated`, `window_complete`, `has_more`.

### 13.6. `feat/rest-sse-adapter`

- поддержка REST SSE upstream.

### 13.7. `feat/grpc-server-streaming-adapter`

- поддержка bounded gRPC server-streaming.

### 13.8. `feat/session-and-job-tools`

- генерация tool families;
- `start/poll/stop`;
- `start/status/result/cancel`.

### 13.9. `feat/websocket-upstream-adapter`

- bounded WebSocket collection;
- subscribe/unsubscribe templates;
- heartbeat/reconnect policy;
- session integration.

### 13.10. `feat/soap-architecture-and-core-model`

- WSDL/XSD-driven domain model;
- SOAP execution config;
- XML normalization strategy.

### 13.11. `feat/soap-adapter-foundation`

- runtime SOAP adapter;
- envelope builder;
- fault normalization;
- test-run support.

### 13.12. `feat/streaming-ui-config`

- новый execution mode selector;
- limits/aggregation/safety blocks;
- test-run UX.

### 13.13. `feat/streaming-e2e`

- публичные smoke targets;
- e2e сценарии;
- manual regression plan.

## 14. Практический итог

Crank должен поддерживать streaming как полнофункциональный MCP proxy, но в управляемой форме:

- transport-level SSE и `Streamable HTTP` поддерживаются;
- upstream streaming поддерживается там, где его можно bounded-ить;
- tool contract остается контролируемым;
- UI настраивает лимиты, aggregation и lifecycle;
- платформа не превращается в бесконечную data pipe.