diff --git a/openspec/changes/add-sse-replay-cursors/design.md b/openspec/changes/add-sse-replay-cursors/design.md new file mode 100644 index 0000000..1b8bafc --- /dev/null +++ b/openspec/changes/add-sse-replay-cursors/design.md @@ -0,0 +1,27 @@ +# Design: SSE Replay Cursors + +## Contract and compatibility + +`subscribe_events` remains the one generated HTTP/CLI SSE operation. It gains one optional string query parameter, `cursor`, and the generated CLI gains `--cursor`; no source name, provider route, hand-written alternate endpoint, or MCP projection is introduced. The existing provider and `thread_id` filters retain their meanings. With no cursor the stream remains future-only, preserving present consumer behavior. + +Each `event: message` frame has an SSE `id: ` line followed by the existing message JSON payload. The opaque cursor is `.`: an unpadded base64url 128-bit random process incarnation generated at server startup and a positive decimal `u64` sequence. The process incarnation is retained only in broker memory, so a pre-restart cursor can never equal a cursor from the new process even when both sequences begin at one. It is never accepted as a provider checkpoint, a forward-poll cursor, or a durable resume token. The CLI continues to print JSONL message payloads by default. `--include-cursor` wraps each emitted message as `{ "cursor": "…", "message": }` so a consumer can persist the value without parsing raw SSE. + +Malformed syntax (missing/invalid incarnation, missing/non-decimal/zero/overflowing sequence, or extra separators) returns HTTP 400 with `{ "error": "invalid_replay_cursor" }` before an SSE response opens. A syntactically valid cursor with an unrecognized incarnation or a sequence outside the retained window returns HTTP 409 with `{ "error": "replay_cursor_expired", "oldest_cursor": "…" }`; this includes a cursor from before server restart. The response does not disclose provider data. If no event has ever been retained, every syntactically valid cursor from the current incarnation is expired and `oldest_cursor` is omitted. + +## Broker ownership and retention + +`iris-server` owns a `ReplayBroker` in `AppState`, separate from any provider's realtime hub or private event history. It generates one random process incarnation at startup and assigns a strictly increasing `u64` sequence when a normalized message is accepted for SSE fan-out, then records the combined cursor, provider instance ID, thread ID, and exact outbound message required by current rendering. Retention capacity is a documented constant of 512 entries (matching the existing bounded realtime scale); it evicts the oldest entry first. Retention is process memory only: restart generates a new incarnation and invalidates all prior cursors. + +The broker is the sole HTTP-subscriber fan-out owner. Provider selection and current 422 unsupported / 503 unavailable / sanitized provider-error behavior remain in the generated SSE handler before broker registration. For each selected configured provider instance, the broker owns at most one upstream `subscribe_realtime()` task while it has one or more HTTP subscribers. That task appends each normalized provider message exactly once to the broker; it never fans out directly to HTTP connections. It starts when the first subscriber for that provider is registered and is cancelled/joined when the final subscriber leaves. Thus no event is assigned a cursor once per HTTP connection, and no new upstream event is retained while no subscriber requires that provider stream. + +Appending an event and registering a subscriber are serialized under one broker lock. A resumed subscription validates its cursor, then under that lock proves it is within the retained window, collects all retained entries strictly after it that match its filters, and registers the subscriber at the same commit point. The provider task begins after registration. The handler releases the lock, emits the collected replay in broker sequence order, then emits the subscriber's queued live events. The live queue preserves append order after registration. An event is consequently either in the retained replay snapshot or arrives in the registered live queue, never neither or both. + +A future-only subscriber registers without collecting historical entries. Replay filtering is identical to live filtering: provider is an exact configured provider instance and `thread_id` is an exact ID match. A cursor is global to the broker, not filter-specific; changing filters may yield a legal empty replay, but a cursor cannot cause a message outside the requested filters to be emitted. + +Slow-consumer behavior remains bounded and explicit. If a subscriber's queue fills after registration, its existing terminal error/close behavior is preserved; replay is collected before streaming and does not bypass the queue's bounded live path. Broker state is cleaned up when stream bodies drop, exactly as the current realtime subscription lifecycle requires. + +## Projection and verification + +`api/operations.yaml` is the source of truth. It declares `cursor` explicitly as a query input on the existing SSE operation and declares the structured 400 `invalid_replay_cursor` and 409 `replay_cursor_expired` response shapes (including optional `oldest_cursor`) in its generated error contract. Its SSE CLI-projection metadata declares `--include-cursor` as a CLI-only output-mode flag; it does not alter HTTP input/output schema. Regenerated CLI and HTTP artifacts are committed and codegen freshness remains required. MCP does not receive this SSE-only operation. + +Server integration tests cover ID framing; no-cursor future-only behavior; retained replay order; replay-to-live seam correctness; exact provider and thread filtering for both paths; malformed cursor 400; expired/evicted/restarted cursor 409; bounded eviction; and concurrent consumers. CLI tests cover `--cursor` URL construction, parsing `id:` fields, default JSONL compatibility, and `--include-cursor` output. Verification includes the full workspace build/test/clippy/fmt/codegen gates plus a generated HTTP/CLI consumer loop that reconnects using only a saved cursor. diff --git a/openspec/changes/add-sse-replay-cursors/proposal.md b/openspec/changes/add-sse-replay-cursors/proposal.md new file mode 100644 index 0000000..e9ed9ff --- /dev/null +++ b/openspec/changes/add-sse-replay-cursors/proposal.md @@ -0,0 +1,24 @@ +# Proposal: SSE Replay Cursors + +## Problem + +`subscribe_events` exposes normalized messages only while a consumer remains connected. Its SSE frames have no `id:` field and the server retains no public replay sequence, so a network disconnect silently loses messages. Consumers cannot distinguish an empty period from a gap, nor can they recover deterministically. + +## Proposal + +Add bounded, process-local replay to the existing generated `subscribe_events` operation: + +1. Every delivered `message` SSE frame receives an opaque monotonic cursor in its `id:` field. +2. The operation accepts an optional explicit `cursor` query input. Without it, the operation remains future-only. With a retained cursor, it replays matching messages strictly after that cursor, then continues live without a gap or duplicate. +3. A server-owned bounded replay broker stores normalized wire messages and their provider/thread routing data. It is independent of a provider's private cache and is not durable across process restart. +4. Malformed cursors fail before a stream opens with `invalid_replay_cursor`; cursors older than retained history (including a prior process) fail explicitly with `replay_cursor_expired` rather than silently starting over. +5. The generated HTTP and CLI surfaces expose the same cursor contract. MCP remains out of scope because this is an SSE-only operation; unary forward polling remains the separate recovery mechanism for MCP consumers. + +## Scope + +- In scope: successor OpenSpec contract, additive generated query/CLI inputs, server-owned bounded replay broker, deterministic replay-to-live handoff, HTTP/CLI integration coverage, and generated-artifact freshness. +- Out of scope: durable replay, acknowledgements, consumer checkpoint storage, provider-source cache exposure, changing provider polling, MCP streaming, or a production deployment. + +## Motivation + +An agent can persist the last SSE `id`, reconnect with `cursor`, and either receive each retained matching message once or an unambiguous recovery error. The bounded in-memory boundary makes loss explicit without turning Iris into a durable event log. diff --git a/openspec/changes/add-sse-replay-cursors/specs/sse-replay-cursors/spec.md b/openspec/changes/add-sse-replay-cursors/specs/sse-replay-cursors/spec.md new file mode 100644 index 0000000..c9c23c9 --- /dev/null +++ b/openspec/changes/add-sse-replay-cursors/specs/sse-replay-cursors/spec.md @@ -0,0 +1,67 @@ +# SSE Replay Cursors Specification + +## ADDED Requirements + +### Requirement: The generated SSE operation exposes an explicit replay cursor + +Iris SHALL retain `subscribe_events` as one generated SSE operation with HTTP and CLI surfaces only. It SHALL declare an optional `cursor` query input in `api/operations.yaml`, and its CLI projection SHALL accept the equivalent `--cursor` input. The operation SHALL NOT gain an MCP projection, a provider-specific name, an alternate replay route, or a hand-written parallel transport contract. + +Every delivered `message` SSE frame SHALL include an opaque `id:` cursor as well as the existing message payload. Without a cursor input, subscription behavior SHALL remain future-only. + +#### Scenario: Existing future-only consumer + +- **WHEN** a consumer subscribes without `cursor` +- **THEN** it receives only messages accepted after its subscription is registered +- **AND** each received message frame includes an opaque `id:` value + +### Requirement: Retained replay is ordered and seamless + +Iris SHALL maintain a bounded, process-memory, server-owned replay broker over normalized outbound message frames. At server start, it SHALL generate one unpadded base64url 128-bit process incarnation; it SHALL assign a strictly increasing positive `u64` sequence when it accepts a message for SSE delivery. The opaque frame cursor SHALL be `.`. The broker SHALL retain at most 512 entries, each with the cursor, message, provider-instance, and thread routing attributes, evicting the oldest entry first. + +The broker SHALL be the sole HTTP-subscriber fan-out owner. For each selected configured provider instance, it SHALL own at most one upstream `subscribe_realtime()` task while at least one HTTP subscriber requires it. That task SHALL append each provider message exactly once to the broker and SHALL NOT fan out directly to HTTP connections. Existing generated-handler provider-selection and unsupported/unavailable/provider-error status behavior SHALL occur before broker registration. The upstream task SHALL start after the first subscriber is registered and be cancelled/joined when the final subscriber leaves. + +For a valid retained cursor, Iris SHALL replay every retained, filter-matching message strictly after that cursor in broker order, then continue live delivery with no gap or duplicate across the replay-to-live handoff. Subscriber registration and replay snapshot collection SHALL occur atomically under the broker lock; an event accepted at that seam SHALL appear either in the replay snapshot or the registered subscriber's live queue, never neither or both. Provider and thread filters SHALL apply identically to retained and live messages. The broker SHALL be independent of provider-private caches and SHALL NOT provide durable replay, acknowledgements, or consumer checkpoint storage. + +#### Scenario: Reconnect within retention + +- **GIVEN** a consumer saved cursor `c` from a message frame +- **AND** matching messages were accepted after `c` and remain retained +- **WHEN** it reconnects with `cursor=c` +- **THEN** it receives each matching retained message exactly once in ascending broker order +- **AND** messages accepted during registration are delivered once after replay + +#### Scenario: Filtered reconnect + +- **GIVEN** a retained cursor and messages for multiple configured provider instances or threads after it +- **WHEN** a consumer reconnects with an exact provider or thread filter +- **THEN** replay and subsequent live delivery contain only messages matching that filter + +### Requirement: Replay failures are explicit before stream open + +A cursor is an opaque `.` value meaningful only for the running Iris process. Missing/invalid incarnation, missing/non-decimal/zero/overflowing sequence, or extra separators SHALL be rejected before provider I/O or SSE stream creation with HTTP 400 `{ "error": "invalid_replay_cursor" }`. + +A syntactically valid cursor with an unrecognized incarnation, a sequence older than the retained window, or a sequence not known to the current process SHALL be rejected before stream creation with HTTP 409 `{ "error": "replay_cursor_expired" }`. When an oldest retained cursor exists, the response SHALL additionally expose it as `oldest_cursor`; it SHALL NOT expose message content or provider-private state. Iris SHALL NOT silently downgrade an expired cursor to a future-only subscription. `api/operations.yaml` SHALL declare both structured error response shapes, including optional `oldest_cursor`, alongside the explicit cursor query input. + +#### Scenario: Evicted cursor + +- **GIVEN** a cursor older than the broker's bounded retained window +- **WHEN** a consumer reconnects with that cursor +- **THEN** Iris returns `replay_cursor_expired` with HTTP 409 +- **AND** it does not open an SSE stream + +#### Scenario: Restarted process + +- **GIVEN** a consumer saved a cursor before Iris restarted +- **WHEN** it reconnects to the new process with that cursor +- **THEN** Iris returns `replay_cursor_expired` with HTTP 409 +- **AND** it does not represent the old cursor as a provider or durable checkpoint + +### Requirement: CLI checkpoint output is structured and backwards compatible + +The CLI projection metadata in `api/operations.yaml` SHALL declare `--include-cursor` as an SSE CLI-only output-mode flag; it SHALL NOT alter the HTTP request or message schema. The generated CLI SHALL pass `--cursor` as the declared query parameter. Its default output SHALL remain one unmodified message JSON object per line. With `--include-cursor`, it SHALL instead emit one JSON object per message containing `cursor` and the unmodified message object under `message`, allowing a consumer to persist the checkpoint without parsing SSE framing. + +#### Scenario: CLI saves a checkpoint + +- **WHEN** a consumer invokes the generated CLI with `--include-cursor` +- **THEN** each emitted JSONL value contains the frame's opaque cursor and unchanged normalized message +- **AND** the saved cursor can be supplied to a later `--cursor` invocation diff --git a/openspec/changes/add-sse-replay-cursors/tasks.md b/openspec/changes/add-sse-replay-cursors/tasks.md new file mode 100644 index 0000000..141b3e6 --- /dev/null +++ b/openspec/changes/add-sse-replay-cursors/tasks.md @@ -0,0 +1,18 @@ +# Tasks: SSE Replay Cursors + +## Contract gate + +- [ ] T1: Approve this successor OpenSpec contract before implementation. The prior `add-realtime-subscriptions` proposal remains frozen. + +## Broker and generated surface + +- [ ] T2: Add the bounded, process-local `ReplayBroker` to `iris-server` state, including startup-generated process incarnation, monotonic cursor assignment, 512-entry exact wire-message retention, one upstream provider subscription per active configured provider, subscriber lifecycle, and bounded eviction. +- [ ] T3: Add explicit optional `cursor` input plus structured 400/409 replay-error shapes to `api/operations.yaml`; declare `--include-cursor` through SSE CLI-projection metadata and regenerate HTTP and CLI artifacts without adding an MCP projection or an alternate route. +- [ ] T4: Render incarnation-qualified SSE `id:` values; implement cursor validation, retained-window/restart expiry responses, exact replay filtering, and atomic replay-to-live registration. +- [ ] T5: Add CLI `--cursor` and `--include-cursor` behavior while preserving default message JSONL output. + +## Verification + +- [ ] T6: Add deterministic server integration coverage for ID framing, future-only behavior, ordering, handoff, filtering, malformed/expired/evicted/restarted cursors, retention bounds, and concurrent consumers. +- [ ] T7: Add CLI/parser/URL/output tests and a generated HTTP/CLI reconnecting-consumer loop. +- [ ] T8: Run `cargo build --all-targets`, `cargo test --all-targets`, `cargo clippy --all-targets -- -D warnings`, `cargo fmt --all -- --check`, and `cargo run -p iris-codegen --bin iris-codegen -- check`.