alkdev/alknet
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(arch): ADR-049 — streaming handler for subscription operations

Browse Source 
The call protocol spec describes streaming (call.responded*N +
call.completed, PendingRequestMap::Subscribe, CallConnection::subscribe),
but the server-side Handler type returned a single ResponseEnvelope —
a Subscription op had no way to produce a stream. The TS predecessor
(@alkdev/operations) had separate OperationHandler / SubscriptionHandler
types; the Rust port collapsed them, losing the streaming path. This
restores it end-to-end: StreamingHandler type, HandlerKind on
HandlerRegistration validated against op_type, invoke_streaming() on
OperationRegistry, server-side dispatch branches on op_type, new
INVALID_OPERATION_TYPE protocol code for wrong-dispatch-path misuse,
GatewayDispatch::invoke_streaming() for /subscribe SSE, from_call stream
forwarding via CallConnection::subscribe(), from_openapi SSE forwarding.
OperationEnv::invoke() stays request/response-only (stream composition is
handler-level, not protocol-level). Amends ADR-023's protocol-code list
(five → six). Tracks the stream-operators library as OQ-41 (feature
extension, not an unmade decision).
This commit is contained in:
glm-5.2
2026-07-02 07:43:01 +00:00
parent 139c651eaa
commit 7ecc11610a
10 changed files with 602 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
docs/architecture/crates/call/call-protocol.md
View File

@@ -1,6 +1,6 @@
---
status: draft
last_updated: 2026-06-23
last_updated: 2026-07-02
---
# Call Protocol
@@ -275,6 +275,7 @@ Error codes use an extensible string enum. The protocol defines the following **
- `NOT_FOUND` — operation not in registry (or Internal op called from wire)
- `FORBIDDEN` — access denied (insufficient scopes or unauthenticated)
- `INVALID_INPUT` — input doesn't match the operation's JSON Schema
- `INVALID_OPERATION_TYPE` — wrong dispatch path for the operation's type (`invoke()` called on a `Subscription`, or `invoke_streaming()` on a `Query`/`Mutation`, or `OperationEnv::invoke()` on a `Subscription` during composition — ADR-049)
- `INTERNAL` — handler error, panic, connection failure
- `TIMEOUT` — request timed out (retryable: true)
@@ -309,7 +310,7 @@ Local dispatch produces `ResponseEnvelope { request_id, result: Result<Value, Ca
| `Ok(value)` | `{ type: "call.responded", id: request_id, payload: { output: value } }` |
| `Err(call_error)` | `{ type: "call.error", id: request_id, payload: <serialized CallError> }` |
The `request_id` becomes the `id` field. For subscriptions, each `call.responded` is a separate `EventEnvelope` with the same `id`; `call.completed` is `{ type: "call.completed", id, payload: {} }`.
The `request_id` becomes the `id` field. For subscriptions, each `call.responded` is a separate `EventEnvelope` with the same `id`; `call.completed` is `{ type: "call.completed", id, payload: {} }`. The streaming dispatch path (`invoke_streaming()` → write each → write `call.completed`) produces these frames from a `StreamingHandler`'s stream; the single-response path (`invoke()` → write one) produces them from a `Handler`'s future. See ADR-049 and [operation-registry.md](operation-registry.md#handler).
### Protocol Operations
@@ -405,10 +406,14 @@ The `CallAdapter::handle()` method:
1. Spawns a task that continuously calls `connection.accept_bi()` to receive incoming streams
2. For each accepted stream, reads `EventEnvelope` frames using `FrameFramedReader`
3. Dispatches `call.requested` events to the operation registry
3. Dispatches `call.requested` events to the operation registry, **branching on `op_type`** (ADR-049):
- **`Query` / `Mutation`** → `OperationRegistry::invoke()` → write one `call.responded` (or `call.error`) `EventEnvelope` frame
- **`Subscription`** → `OperationRegistry::invoke_streaming()` → write each `call.responded` `EventEnvelope` as the stream yields → write `call.completed` on natural stream end (or `call.error` if the stream yields an `Err`). `deadline: None` for subscriptions (unbounded — see Timeouts below). Abort (`call.aborted` arriving for the request ID, or the stream being dropped) cascades per ADR-016: the stream future is dropped, `Drop` guards release the handler's resources, and descendants are aborted.
4. Writes response `EventEnvelope` frames using `FrameFramedWriter`
5. Manages `PendingRequestMap` for outgoing calls initiated by the server
The streaming branch is the server-side path that makes `Subscription` operations work end-to-end. Without it, a `Subscription` op registered with a `StreamingHandler` had no server-side dispatch path — the handler produced a stream but the dispatcher only read one `ResponseEnvelope` and closed. ADR-049 adds the `StreamingHandler` type and the `invoke_streaming()` dispatch path; this section wires them into the accept loop. See [operation-registry.md](operation-registry.md#handler) for the `Handler` / `StreamingHandler` / `HandlerKind` types.
For outgoing calls (server → client), the adapter:
1. Opens a bidirectional stream with `connection.open_bi()`
2. Sends `call.requested` on that stream
@@ -562,6 +567,7 @@ Handlers clean up resources when their call is cancelled (in Rust, the future is
| Peer-graph routing model (supersedes ADR-028) | [ADR-029](../../decisions/029-peer-graph-routing-model.md) | Peer-keyed overlays + `PeerRef` routing; `AccessControl`-based peer authorization; retires `remote_safe`/`trusted_peer` |
| Forwarded-for identity | [ADR-032](../../decisions/032-forwarded-for-identity.md) | `forwarded_for` field on `call.requested` and `OperationContext`; metadata only — `AccessControl::check` never reads it; the `from_call` handler populates it |
| Operation error schemas | [ADR-023](../../decisions/023-operation-error-schemas.md) | Operations declare domain errors; `call.error` carries typed `details` |
| Streaming handler for subscriptions | [ADR-049](../../decisions/049-streaming-handler-for-subscriptions.md) | `StreamingHandler` type, `invoke_streaming()` dispatch path, `INVALID_OPERATION_TYPE` protocol code; the server-side streaming branch in `handle_stream` |
## Open Questions
@@ -615,4 +621,5 @@ See [open-questions.md](../../open-questions.md) for full details.
- ADR-030: PeerEntry and Identity.id decoupling (`PeerId` source)
- ADR-032: Forwarded-for identity (`forwarded_for` on `call.requested` and `OperationContext`)
- ADR-034: Outgoing-only X.509 and the three peer roles
- ADR-049: Streaming handler for subscriptions (server-side streaming dispatch path)
- Reference implementation: `/workspace/@alkdev/alknet-main/crates/alknet-core/src/call/`

20
docs/architecture/crates/call/client-and-adapters.md
View File

@@ -1,6 +1,6 @@
---
status: draft
last_updated: 2026-06-28
last_updated: 2026-07-02
---
# alknet-call — Client and Adapters
@@ -323,8 +323,21 @@ The flow (ADR-017 §3):
3. For each discovered op, construct a `HandlerRegistration`:
- `spec` mirrors the remote op's name (with optional prefix), namespace,
type, schemas, access control.
- `handler` is a forwarding handler: sends `call.requested` through the
`CallConnection`, awaits `call.responded` (or streams for subscriptions).
- `handler` is a forwarding handler, **branched on `op_type`** (ADR-049):
- `Query` / `Mutation` → a `Handler` (registered as `HandlerKind::Once`):
sends `call.requested` via `CallConnection::call_with_payload()`, awaits
the single `call.responded` (or `call.error`), returns the
`ResponseEnvelope`.
- `Subscription` → a `StreamingHandler` (registered as
`HandlerKind::Stream`): calls `CallConnection::subscribe()`, which
returns `impl Stream<Item = ResponseEnvelope>` (the client-side
streaming path, already implemented), maps it to a
`BoxStream<ResponseEnvelope>`. The remote stream flows end-to-end:
each `call.responded` the remote sends becomes a stream item; the
remote's `call.completed` ends the stream (→ wire `call.completed`);
`call.aborted` drops the stream (cascade per ADR-016). No truncation,
no first-value fallback — a `from_call`-imported subscription forwards
the full remote stream.
- `provenance: FromCall`, `composition_authority: None`, `scoped_env: None`
(leaf — ADR-022).
4. The caller registers the bundles via
@@ -668,6 +681,7 @@ Based on the gap analysis and the downstream unblock chain:
| Privilege model and authority context | [ADR-015](../../decisions/015-privilege-model-and-authority-context.md) | Adapter-registered ops are `Internal` by default; default-deny posture |
| Abort cascade for nested calls | [ADR-016](../../decisions/016-abort-cascade-for-nested-calls.md) | Cross-node abort through `from_call` forwarding handler's `parent_request_id` |
| Operation error schemas | [ADR-023](../../decisions/023-operation-error-schemas.md) | `error_schemas` mirrored by `from_call` from remote op's spec |
| Streaming handler for subscriptions | [ADR-049](../../decisions/049-streaming-handler-for-subscriptions.md) | `from_call` `Subscription` ops register a `StreamingHandler` (`HandlerKind::Stream`) that calls `CallConnection::subscribe()` and forwards the remote stream; `Query`/`Mutation` stay `HandlerKind::Once` |
| TLS identity redesign | [ADR-027](../../decisions/027-tls-identity-redesign-acme-rawkey-decoupling.md) | RFC 7250 raw key / X.509 cert dimensions of `CallCredentials` |
| Outgoing-only X.509 and three peer roles | [ADR-034](../../decisions/034-outgoing-only-x509-and-three-peer-roles.md) | Public X.509 endpoint is not a `PeerEntry` on the client side (no `PeerId`, not in peer graph); client-side verifier by `PeerEntry` presence (CA vs fingerprint pin); hub = mixed-fingerprint `PeerEntry` |
| HD derivation for encryption keys | [ADR-020](../../decisions/020-hd-derivation-for-encryption-keys.md) | Vault-derived TLS identity material |

108
docs/architecture/crates/call/operation-registry.md
View File

@@ -1,6 +1,6 @@
---
status: draft
last_updated: 2026-06-27
last_updated: 2026-07-02
---
# Operation Registry
@@ -91,19 +91,75 @@ Operations with empty `AccessControl` (no required scopes, no resource checks) a
### Handler
There are two handler types, one per dispatch shape — mirroring the
TypeScript prior art (`@alkdev/operations/src/types.ts:62-78`:
`OperationHandler` returns a single value; `SubscriptionHandler` returns an
`AsyncGenerator`). The split is locked by ADR-049.
```rust
pub type Handler = Arc<dyn Fn(Value, OperationContext) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>> + Send + Sync>;
/// Request/response handler — Query and Mutation operations.
pub type Handler = Arc<
dyn Fn(Value, OperationContext) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>
+ Send + Sync,
>;
/// Streaming handler — Subscription operations. Returns a stream of
/// ResponseEnvelopes: each Ok(value) → call.responded, an Err → call.error
/// (terminal — stream ends), natural stream end → call.completed.
pub type StreamingHandler = Arc<
dyn Fn(Value, OperationContext)
-> Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>
+ Send + Sync,
>;
/// Type alias for the boxed stream shape used by `invoke_streaming()` and
/// `StreamingHandler` return values. The concrete library
/// (`futures::stream::BoxStream<'static, T>` = `Pin<Box<dyn Stream<Item = T>
/// + Send>>`) is a two-way-door implementation detail (ADR-049); the alias
/// exists so the two spellings (the expanded form in `StreamingHandler` and
/// the short form in `invoke_streaming()`) refer to the same type.
pub type ResponseStream = Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>;
```
Handlers are async — many operations (file I/O, HTTP service calls, irpc service calls) are inherently asynchronous. The handler receives an `async` runtime context and returns a `Future<Output = ResponseEnvelope>`.
Both handlers are async — many operations (file I/O, HTTP service calls,
irpc service calls, LLM streaming) are inherently asynchronous. A handler
receives:
A handler receives:
- `input: Value` — the deserialized `payload` from the `call.requested` event (always `serde_json::Value`)
- `input: Value` — the deserialized `payload` from the `call.requested` event
(always `serde_json::Value`)
- `context: OperationContext` — request ID, identity, metadata, env
And returns a `ResponseEnvelope` containing the result or an error. `ResponseEnvelope` is defined in [call-protocol.md](call-protocol.md#responseenvelope) — it carries the request ID and a `Result<Value, CallError>`. Local dispatch produces it with no serialization overhead; the `CallAdapter` converts it to `EventEnvelope` for the wire.
The **`Handler`** (request/response) returns a single `ResponseEnvelope`
containing the result or an error. `ResponseEnvelope` is defined in
[call-protocol.md](call-protocol.md#responseenvelope) — it carries the request
ID and a `Result<Value, CallError>`. Local dispatch produces it with no
serialization overhead; the `CallAdapter` converts it to `EventEnvelope` for
the wire.
When a handler returns an error, the `CallError.code` is matched against the operation's declared `error_schemas` (ADR-023). If the code matches a declared `ErrorDefinition`, the `call.error` event carries that code and the error's detail payload. If it doesn't match, the `call.error` carries `INTERNAL`. This is how handler failures become typed errors on the wire instead of string-matched messages.
The **`StreamingHandler`** (streaming) returns a `Pin<Box<dyn Stream<Item =
ResponseEnvelope> + Send>>` — the stream analogue of `Handler`'s
`Pin<Box<dyn Future<...>>>`. Each `Ok(value)` in the stream becomes a
`call.responded` event; an `Err` becomes a `call.error` event (terminal — the
stream ends after it); natural stream end becomes `call.completed`. The
dispatch path converts each `ResponseEnvelope` to `EventEnvelope` exactly as
it does for the single-response case — no new wire-format concept is
introduced. See ADR-049 and [call-protocol.md](call-protocol.md) §"CallAdapter
Stream Handling".
When a handler returns an error, the `CallError.code` is matched against the operation's declared `error_schemas` (ADR-023). If the code matches a declared `ErrorDefinition`, the `call.error` event carries that code and the error's detail payload. If it doesn't match, the `call.error` carries `INTERNAL`. This is how handler failures become typed errors on the wire instead of string-matched messages. The same matching applies to `Err` values yielded by a `StreamingHandler`.
A `make_streaming_handler()` helper (analogue of `make_handler()`) wraps a
stream-producing closure into a `StreamingHandler`:
```rust
pub fn make_streaming_handler<S, St>(f: S) -> StreamingHandler
where
S: Fn(Value, OperationContext) -> St + Send + Sync + 'static,
St: Stream<Item = ResponseEnvelope> + Send + 'static,
{
Arc::new(move |input, context| Box::pin(f(input, context)))
}
```
### OperationContext
@@ -196,9 +252,10 @@ pub struct OperationRegistry {
The registry maps operation names to `HandlerRegistration` bundles. The curated layer (Layer 0) is a `HashMap<String, HandlerRegistration>`; session and connection overlays (Layers 1 and 2) are separate maps that the `CallAdapter` composes into the per-call `OperationContext.env` (ADR-024). See ADR-022 for the full registration model and ADR-024 for the layering model. Key methods:
- `register(registration)`: Add an operation to the curated layer at startup
- `registration(name)`: Find a registration by operation name (checks active overlays first, then curated base — ADR-024). Returns spec, handler, provenance, composition authority, scoped env, capabilities.
- `invoke(name, input, context)`: Look up, check ACL, invoke handler, return result
- `register(registration)`: Add an operation to the curated layer at startup. Validates `handler` is the right `HandlerKind` for `spec.op_type` (Once for Query/Mutation, Stream for Subscription — ADR-049). Mismatch is a startup error.
- `registration(name)`: Find a registration by operation name (checks active overlays first, then curated base — ADR-024). Returns spec, handler (`HandlerKind`), provenance, composition authority, scoped env, capabilities.
- `invoke(name, input, context)`: Look up, check ACL, invoke handler, return a single `ResponseEnvelope` (request/response path — Query/Mutation). **Errors with `INVALID_OPERATION_TYPE` if the op is a `Subscription`** — `invoke()` is the wrong dispatch path for streaming ops; use `invoke_streaming()` (ADR-049).
- `invoke_streaming(name, input, context)`: Look up, check ACL, invoke streaming handler, return a `ResponseStream` (the boxed stream alias — ADR-049) (streaming path — Subscription). Pre-handler errors (not-found, forbidden, `INVALID_OPERATION_TYPE` for a non-Subscription op) yield a single error `ResponseEnvelope` and end the stream. See ADR-049.
- `list_operations()`: Return all registered specs (for `/services/list` — returns curated + active overlay ops)
### Request ID Generation
@@ -229,15 +286,23 @@ The registration bundle carries everything the dispatch path needs to construct
```rust
pub struct HandlerRegistration {
pub spec: OperationSpec,
pub handler: Handler,
pub handler: HandlerKind, // Once or Stream — validated against spec.op_type (ADR-049)
pub provenance: OperationProvenance,
pub composition_authority: Option<CompositionAuthority>, // None for leaves
pub scoped_env: Option<ScopedOperationEnv>, // None for leaves
pub scoped_env: Option<ScopedPeerEnv>, // None for leaves
pub capabilities: Capabilities,
// NOTE: ADR-028 added `remote_safe: bool` here; ADR-029 supersedes it and
// removes the field. Peer authorization is `AccessControl::check(peer_identity)`,
// not a per-op boolean. See ADR-029 §3.
}
/// Which dispatch path a handler uses — locked by ADR-049.
/// Validated against `spec.op_type` at registration:
/// Query/Mutation → Once; Subscription → Stream. Mismatch is a startup error.
pub enum HandlerKind {
Once(Handler),
Stream(StreamingHandler),
}
```
#### OperationProvenance
@@ -291,19 +356,22 @@ impl CompositionAuthority {
- `scoped_env`: The set of operations this handler may reach via `env.invoke()`. `None` for leaves (empty env). The reachability control from ADR-015.
- `capabilities`: Outbound credentials (decrypted API keys, signing keys). Populated by the assembly layer from the vault at registration time. See [Capability Injection](#capability-injection).
The `OperationRegistryBuilder` provides a fluent API with convenience methods for common cases:
The `OperationRegistryBuilder` provides a fluent API with convenience methods for common cases. The builder absorbs the `HandlerKind` wrapping internally — `.with_local()` and `.with_leaf()` take the raw `Handler` (or `StreamingHandler`) and wrap it in the right `HandlerKind` based on `spec.op_type` (ADR-049):
```rust
// with_local: Local provenance, full bundle — all 5 args required.
// with_local(spec, handler, composition_authority, scoped_env, capabilities)
// The builder inspects spec.op_type and wraps in HandlerKind::Once
// (Query/Mutation) or HandlerKind::Stream (Subscription) automatically.
let registry = OperationRegistryBuilder::new()
// Built-in service discovery (Local, no composition — empty authority, empty env, empty caps)
.with_local(services_list_spec(), Arc::new(services_list_handler),
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
.with_local(services_schema_spec(), Arc::new(schema_handler),
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
// Agent handler (Local, composes — authority + scoped env + capabilities)
.with_local(agent_chat_spec(), Arc::new(agent_chat_handler),
// Agent handler (Local, Subscription — streams call.responded as the
// LLM generates tokens; builder wraps in HandlerKind::Stream)
.with_local(agent_chat_spec(), Arc::new(agent_chat_streaming_handler),
CompositionAuthority::new("agent-chat", ["llm:call", "fs:read", "vastai:query"]),
ScopedOperationEnv::new(["fs/readFile", "vastai/listMachines", "llm/generate"]),
Capabilities::new().with_api_key("google", google_api_key))
@@ -318,6 +386,8 @@ The CLI binary (or assembly layer) constructs the registry and passes it to the
The `OperationEnv` trait is the universal composition mechanism. A handler calls `context.env.invoke("fs", "readFile", input, &context)` and gets a `ResponseEnvelope` back — regardless of whether the operation runs locally, via an irpc service, or on a remote node.
**`OperationEnv` is request/response-only** (ADR-049). It returns a single `ResponseEnvelope` — no streaming variant exists. Calling `invoke()` on a `Subscription` op produces `CallError { code: "INVALID_OPERATION_TYPE", ... }` — composition cannot truncate a stream to its first value. Stream composition (filter, map, combine, window, dedupe) is a handler-level concern, not a protocol composition concern; see ADR-049 for the rationale and the `@alkdev/pubsub` `operators.ts` prior art.
```rust
/// The composition dispatch trait. A handler composes child operations
/// through its `OperationContext.env` (which implements this trait).
@@ -673,10 +743,11 @@ let registry = OperationRegistryBuilder::new()
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
.with_local(services_schema_spec(), Arc::new(schema_handler),
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
// Agent handler (Local, composes — full bundle via .with())
// Agent handler (Local, Subscription — composes; streaming handler
// wrapped in HandlerKind::Stream by the builder per ADR-049)
.with(HandlerRegistration {
spec: agent_chat_spec(),
handler: Arc::new(agent_chat_handler),
handler: HandlerKind::Stream(Arc::new(agent_chat_streaming_handler)),
provenance: OperationProvenance::Local,
composition_authority: Some(CompositionAuthority::new(
"agent-chat", ["llm:call", "fs:read", "vastai:query"])),
@@ -750,6 +821,7 @@ The `Capabilities` type holds non-serializable, zeroized secret material. It doe
- **The call protocol carries no secret material.** Secret material (private keys, API keys, mnemonics, decrypted credentials) must not appear in `call.requested` payloads, `call.responded` payloads, or `OperationContext.metadata`. See ADR-014.
- **Metadata does not propagate through composition.** `OperationEnv::invoke()` constructs fresh metadata for nested calls (`HashMap::new()`), not the parent's metadata. This prevents a handler that accidentally places a secret in metadata from leaking it to child operations — and if a child is a `from_call` operation (ADR-017), across the wire to a remote node. The tracing link is `parent_request_id`, not metadata propagation. See ADR-014.
- **Provenance determines composition capability.** Only `Local` and `Session` ops can compose. Leaves (`FromOpenAPI`, `FromMCP`, `FromCall`) get `composition_authority: None` and `scoped_env: None` — they don't compose, so they don't need authority or reachability bounds. See ADR-022.
- **`HandlerKind` matches `op_type`** (ADR-049). `Query`/`Mutation` ops register a `HandlerKind::Once(Handler)`; `Subscription` ops register a `HandlerKind::Stream(StreamingHandler)`. Mismatch is a startup error. `invoke()` on a `Subscription` and `invoke_streaming()` on a `Query`/`Mutation` both return `INVALID_OPERATION_TYPE`. `OperationEnv::invoke()` (composition) is request/response-only and errors with `INVALID_OPERATION_TYPE` on `Subscription` ops — stream composition is a handler-level concern, not a protocol composition concern.
## Design Decisions
@@ -768,6 +840,7 @@ The `Capabilities` type holds non-serializable, zeroized secret material. It doe
| Peer-graph routing model (supersedes ADR-028) | [ADR-029](../../decisions/029-peer-graph-routing-model.md) | Peer-keyed overlays + `PeerRef` routing; peer authorization via `AccessControl::check(peer_identity)`; retires `remote_safe`/`trusted_peer` (the field this doc's `HandlerRegistration` previously gained) |
| Forwarded-for identity | [ADR-032](../../decisions/032-forwarded-for-identity.md) | `forwarded_for` field on `OperationContext` and `call.requested`; metadata only — `AccessControl::check` never reads it; the `from_call` handler populates it |
| ~~Peer-scoped registry filtering~~ (superseded) | ~~[ADR-028](../../decisions/028-callclient-peer-scoped-registry-filtering.md)~~ | ~~`remote_safe` marking on `HandlerRegistration`~~ — superseded by ADR-029 |
| Streaming handler for subscriptions | [ADR-049](../../decisions/049-streaming-handler-for-subscriptions.md) | `StreamingHandler` type alongside `Handler`; `HandlerKind` enum on `HandlerRegistration` validated against `op_type`; `invoke_streaming()` on `OperationRegistry`; `invoke()` and `OperationEnv::invoke()` error with `INVALID_OPERATION_TYPE` on `Subscription` ops; composition stays request/response-only, stream composition is handler-level |
## Open Questions
@@ -814,4 +887,5 @@ See [open-questions.md](../../open-questions.md) for full details.
- ADR-029: Peer-graph routing model (peer-keyed overlays + `PeerRef` routing; `PeerCompositeEnv` supersedes the singular-connection `CompositeOperationEnv`)
- ADR-030: PeerEntry and Identity.id decoupling (`PeerId` source = `Identity.id` = `PeerEntry.peer_id`)
- ADR-032: Forwarded-for identity (`forwarded_for` on `OperationContext` and `call.requested`; metadata only)
- ADR-049: Streaming handler for subscriptions (`StreamingHandler`, `HandlerKind`, `invoke_streaming()`, `INVALID_OPERATION_TYPE`)
- Reference implementation: `/workspace/@alkdev/alknet-main/crates/alknet-core/src/call/`

27
docs/architecture/crates/http/http-adapters.md
View File

@@ -1,6 +1,6 @@
---
status: draft
last_updated: 2026-07-01
last_updated: 2026-07-02
---
# HTTP Adapters — from_openapi and to_openapi
@@ -123,8 +123,8 @@ The adapter:
### Forwarding handler
The forwarding handler is the `Arc<dyn Handler>` stored in the
`HandlerRegistration`. At call time, it:
The forwarding handler is stored in the `HandlerRegistration` as a
`HandlerKind` (ADR-049). At call time, it:
1. Reads the call input (`serde_json::Value`).
2. Builds the outbound HTTP request:
@@ -138,15 +138,23 @@ The forwarding handler is the `Arc<dyn Handler>` stored in the
below).
4. For a `Query`/`Mutation`: parses the response body (JSON, text, or
binary — same content-type branching as the TS `createHTTPOperation`),
wraps it in a `ResponseEnvelope`, returns.
wraps it in a `ResponseEnvelope`, returns. Registered as
`HandlerKind::Once` — a `Handler` returning a single
`ResponseEnvelope`.
5. For a `Subscription` (`text/event-stream` response): streams
`call.responded` events as the SSE chunks arrive (same SSE parsing as
the TS `parseSSEFrames`), then `call.completed` on stream end.
the TS `parseSSEFrames`), then the stream ends on SSE close (which
becomes `call.completed` on the wire). Registered as
`HandlerKind::Stream` — a `StreamingHandler` returning a
`BoxStream<ResponseEnvelope>` (ADR-049). Each SSE `data:` frame becomes
a `ResponseEnvelope::ok()`; an HTTP error (non-2xx) becomes a single
`ResponseEnvelope::error()` and ends the stream.
6. On HTTP error (non-2xx): maps to the declared `ErrorDefinition` by
HTTP status code (see Error Fidelity below), returns a `CallError`.
The handler is opaque to the `CallAdapter` — it's an `Arc<dyn Handler>`
the registry dispatches. `alknet-call` never sees `reqwest`.
The handler is opaque to the `CallAdapter` — it's a `HandlerKind` the
registry dispatches (via `invoke()` for `Once`, `invoke_streaming()` for
`Stream`). `alknet-call` never sees `reqwest`.
### HTTP client (reqwest)
@@ -319,9 +327,9 @@ factoring recommendation (thin shared struct, not a trait).
`from_openapi` maps OpenAPI non-2xx response status codes to
`ErrorDefinition`s (ADR-023 §5). The normative rule (review #002 W20):
`from_openapi` must not produce error codes that collide with the five
`from_openapi` must not produce error codes that collide with the six
protocol-level codes (`NOT_FOUND`, `FORBIDDEN`, `INVALID_INPUT`,
`INTERNAL`, `TIMEOUT`). The adapter prefixes imported error codes with
`INVALID_OPERATION_TYPE`, `INTERNAL`, `TIMEOUT`). The adapter prefixes imported error codes with
`HTTP_` and the status number:
```rust
@@ -423,6 +431,7 @@ once published, the 5-endpoint gateway shape is one-way.
| HTTP path = operation path (~~direct-call surface~~) | [ADR-036](../../decisions/036-http-to-call-operation-mapping.md) → superseded by [ADR-047](../../decisions/047-remove-direct-call-http-surface.md) | ~~`POST /{service}/{op}` → `call.requested`~~ — removed; the gateway `/call` with `{ operation, input }` is the sole invoke path; `to_openapi` describes the gateway, not a per-operation surface |
| `to_openapi` gateway pattern | [ADR-042](../../decisions/042-openapi-gateway-pattern.md) | 5 fixed gateway endpoints (search/schema/call/batch/subscribe), not one path per operation; per-caller AccessControl-filtered. Supersedes ADR-036's original `to_openapi` "paths mirror `/{service}/{op}`" clause |
| `to_openapi` published-spec versioning | [ADR-045](../../decisions/045-to-openapi-gateway-spec-versioning.md) | `info.version` semver tracks the gateway endpoint contract, not the operation set; consumers detect breaking changes via the major version |
| Streaming handler for subscriptions | [ADR-049](../../decisions/049-streaming-handler-for-subscriptions.md) | `from_openapi` `Subscription` ops register a `StreamingHandler` (`HandlerKind::Stream`); SSE response → `BoxStream<ResponseEnvelope>`; `Query`/`Mutation` stay `HandlerKind::Once` |
## Open Questions

51
docs/architecture/crates/http/http-mcp.md
View File

@@ -1,6 +1,6 @@
---
status: draft
last_updated: 2026-07-01
last_updated: 2026-07-02
---
# HTTP MCP — from_mcp and to_mcp
@@ -78,10 +78,12 @@ The adapter:
namespace prefix is configured — same local-naming sugar as
`from_call`'s `FromCallConfig::namespace_prefix`, ADR-029 §5).
- `spec.namespace` = the configured `namespace`.
- `spec.op_type` = `Mutation` (MCP tools are call/response; the MCP
spec doesn't have a native streaming/tool-subscription distinction
— `tools/call` returns a result. If MCP adds a streaming-tool
extension, a `Subscription` mapping would be added.)
- `spec.op_type` = `Mutation` (MCP tools are call/response; the MCP
spec doesn't have a native streaming/tool-subscription distinction
— `tools/call` returns a result. If MCP adds a streaming-tool
extension, a `Subscription` mapping would be added.) All `from_mcp`
handlers are `HandlerKind::Once` (ADR-049); `from_mcp` never
produces a `StreamingHandler`.
- `spec.visibility` = `Internal` (adapter-registered, ADR-015).
- `spec.input_schema` = the tool's `inputSchema` (JSON Schema).
- `spec.output_schema` = depends on whether the tool declares
@@ -128,8 +130,9 @@ At call time, the `from_mcp` forwarding handler:
registration (the MCP server is a persistent streamable HTTP
endpoint, not a per-call connection).
The handler is opaque to the `CallAdapter` — `Arc<dyn Handler>` the
registry dispatches. `alknet-call` never sees rmcp.
The handler is opaque to the `CallAdapter` — a `HandlerKind::Once`
wrapping an `Arc<dyn Handler>` that the registry dispatches. `alknet-call`
never sees rmcp.
### Output handling (structuredContent vs content blocks)
@@ -222,7 +225,12 @@ The gateway exposes only `Query` and `Mutation` operations
(request/response). `Subscription` operations (streaming) are filtered
out of `search` results and cannot be invoked via `call` — MCP tool
calls are request/response by protocol design; streaming subscriptions
don't fit the LLM tool-call pattern. See ADR-041 §2.
don't fit the LLM tool-call pattern. This is unaffected by ADR-049
(streaming handlers): the `StreamingHandler` type and `invoke_streaming()`
dispatch path exist in `alknet-call` and are used by `to_openapi`'s
`/subscribe` endpoint, but `to_mcp` does not expose them — it filters by
`op_type` and only dispatches `Query`/`Mutation` via `invoke()`. See
ADR-041 §2.
#### `to_mcp` service behavior
@@ -263,19 +271,19 @@ axum route handlers) are genuinely per-gateway and are not shared.
Research findings
(`docs/research/alknet-http-gateway-factoring/findings.md`) recommend
extracting a **thin shared spine** (a concrete struct holding
`Arc<OperationRegistry>` + `Arc<dyn IdentityProvider>` with a
`resolve + build_context + invoke` method returning a
`ResponseEnvelope`), **not** a `GatewayDispatch` trait or gateway
abstraction. The spine is small (~15–30 lines per endpoint), but it is
the one place where a divergence bug (identity resolved differently,
`OperationContext.internal` set inconsistently, `CallError` mapped
asymmetrically) would be a security/correctness issue. The
server-integration and wire-framing layers stay per-gateway; a third
gateway (GraphQL, gRPC) is not on the horizon, and if one appears its
server-integration layer needs its own shape anyway. This is an
implementation factoring note, not an ADR — the decision is internal to
`alknet-http` and does not cross crate boundaries.
extracting a **thin shared spine** (the concrete `GatewayDispatch` struct
holding `Arc<OperationRegistry>` + `Arc<dyn IdentityProvider>` with a
`resolve + build_context + invoke` method returning a `ResponseEnvelope`,
named in ADR-049 and extended with `invoke_streaming()` for the streaming
path), **not** a trait or gateway abstraction. The spine is small (~15–30
lines per endpoint), but it is the one place where a divergence bug
(identity resolved differently, `OperationContext.internal` set
inconsistently, `CallError` mapped asymmetrically) would be a
security/correctness issue. The server-integration and wire-framing layers
stay per-gateway; a third gateway (GraphQL, gRPC) is not on the horizon,
and if one appears its server-integration layer needs its own shape anyway.
This is an implementation factoring note, not an ADR — the decision is
internal to `alknet-http` and does not cross crate boundaries.
### No-Env-Vars
@@ -340,6 +348,7 @@ every other HTTP request.
| Error fidelity | [ADR-023](../../decisions/023-operation-error-schemas.md) | MCP tool errors mapped to `ErrorDefinition`s |
| No-env-vars credential injection | [ADR-014](../../decisions/014-secret-material-flow-and-capability-injection.md) | Handler reads `context.capabilities`, not env vars |
| MCP clients are not alknet peers | [ADR-034](../../decisions/034-outgoing-only-x509-and-three-peer-roles.md) | Bearer token, no `PeerId` |
| Streaming handler for subscriptions | [ADR-049](../../decisions/049-streaming-handler-for-subscriptions.md) | `from_mcp` handlers are always `HandlerKind::Once` (MCP tools are request/response); `to_mcp` excludes `Subscription` ops (unchanged by the streaming handler) |
## Open Questions

36
docs/architecture/crates/http/http-server.md
View File

@@ -1,6 +1,6 @@
---
status: draft
last_updated: 2026-07-01
last_updated: 2026-07-02
---
# HTTP Server
@@ -194,13 +194,22 @@ The request body is `{ operation, input }` (the same flat JSON shape as
`Accept: text/event-stream` on the `POST`). The axum route handler:
- Sets `Content-Type: text/event-stream`.
- For each `call.responded` event, writes an SSE `data:` frame (the
event's `output` serialized as JSON).
- On `call.completed`, closes the SSE stream (normal end).
- On `call.aborted`, closes the stream with an SSE error event.
- On HTTP client disconnect (detected as the response writer closing),
sends `call.aborted` for the in-flight subscription, which cascades
to descendants per ADR-016.
- Calls `GatewayDispatch::invoke_streaming()` (ADR-049) — the streaming
analogue of `invoke()`, returning a `BoxStream<ResponseEnvelope>`. The
security invariants are identical to `invoke()`: `internal: false`,
`forwarded_for: None`, same capabilities, same `scoped_env`, same ACL
check before dispatch. The two methods diverge only on the return shape
(stream vs single envelope).
- For each `ResponseEnvelope` the stream yields, writes an SSE `data:` frame:
`Ok(value)` → `data:` frame with the output serialized as JSON; `Err` →
SSE error event with the `CallError` serialized, then close (an `Err` is
terminal — the stream ends after it, matching the wire protocol's
`call.error` semantics).
- On natural stream end (the `StreamingHandler`'s stream completes), closes
the SSE stream (normal end — corresponds to `call.completed` on the wire).
- On `call.aborted` or HTTP client disconnect (detected as the response
writer closing), drops the stream future — `Drop` guards release the
handler's resources, and the abort cascade runs per ADR-016.
This is the HTTP/1.1 + HTTP/2 streaming projection. Over WebSocket
([websocket.md](websocket.md)), the subscription projects directly
@@ -209,6 +218,16 @@ no SSE framing. WebTransport (`h3`, deferred per ADR-044) would project
onto WebTransport bidirectional streams; see
[webtransport.md](webtransport.md).
**The streaming dispatch path.** Pre-ADR-049, `subscribe_handler` called
`GatewayDispatch::invoke()` (single response) and wrapped the one
`ResponseEnvelope` in a one-event SSE stream — a placeholder that couldn't
stream a real `Subscription` op. ADR-049 adds `GatewayDispatch::
invoke_streaming()` and the underlying `OperationRegistry::
invoke_streaming()`, giving `/subscribe` a real streaming dispatch path
to call. See ADR-049 and [http-adapters.md](http-adapters.md) for the
`from_openapi` SSE forwarding handler that feeds `StreamingHandler`s from
external `text/event-stream` responses.
### One-directional projection (HTTP request/response)
The HTTP/1.1 + HTTP/2 surface is a **lossy, one-directional projection**
@@ -446,6 +465,7 @@ two-way door (add/remove freely). See
| Browsers are not alknet peers | [ADR-034](../../decisions/034-outgoing-only-x509-and-three-peer-roles.md) §4 (amended by ADR-044 §5) | Bearer token, no `PeerId`, connection-local overlay (addressability vs. bidirectionality) — full rationale in [websocket.md](websocket.md) |
| Error mapping (call codes → HTTP status) | [ADR-023](../../decisions/023-operation-error-schemas.md) | Protocol/operation codes distinct; `HTTP_<status>` prefix for imported |
| Custom HTTP routes from the assembly layer | [ADR-046](../../decisions/046-assembly-layer-custom-http-routes.md) | `extra_routes: Option<Router>` at construction; raw HTTP, not operations; default surface takes precedence on collision |
| Streaming handler for subscriptions (`invoke_streaming()`) | [ADR-049](../../decisions/049-streaming-handler-for-subscriptions.md) | `GatewayDispatch::invoke_streaming()` returns `BoxStream<ResponseEnvelope>`; `/subscribe` pipes it to SSE; replaces the one-event placeholder with the real streaming dispatch path |
## Open Questions