David/HERMES-GIELIE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #5520 from nesquena/release/stage-p1

Browse source 
Release — session SSE contract RFC (#5513, addresses #4812)
This commit is contained in:
nesquena-hermes 2026-07-03 23:55:45 -07:00 committed by GitHub
commit 8e26cc03d0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 549 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
docs/CONTRACTS.md
View file

@ -60,6 +60,13 @@ contributor guidance; it does not change runtime behavior or CI gates.
queued messages, interrupt replacement, steer visibility, or leftover-steer
recovery changes.
- [`docs/rfcs/README.md`](rfcs/README.md): RFC conventions and current RFC index.
- [`docs/rfcs/session-sse-contract-v1.md`](rfcs/session-sse-contract-v1.md):
proposed contract vocabulary, cursor/resume semantics, replay identity, snapshot
fallback, event taxonomy, and implementation gates for the per-session SSE
stream `GET /api/sessions/{session_id}/events` (#4812). Distinct from the
existing global session-list stream `GET /api/sessions/events`. Start here for
any work that touches per-session SSE, `Last-Event-ID` replay, or session
lifecycle event delivery. Proposed RFCs do not authorize implementation.
When a change touches streaming, recovery, replay, compression, context
reconstruction, cancellation, approval/clarify, session metadata, or run state,

3
docs/rfcs/README.md
View file

@ -66,3 +66,6 @@ First-time contributor RFCs should be discussed in an issue before opening a PR.
#3058 control-surface companion to #3400 for Queue, Steer, Stop-and-send,
Interrupt, and leftover-steer inputs submitted while a long-running agent
session is active.
- [`session-sse-contract-v1.md`](session-sse-contract-v1.md) — #4812 Proposed
contract vocabulary, replay identity, event taxonomy, cursor/resume semantics,
and implementation gates for `GET /api/sessions/{session_id}/events`.

261
docs/rfcs/session-sse-contract-v1.md Normal file
View file

@ -0,0 +1,261 @@
# Session SSE Contract v1
- **Status:** Proposed
- **Author:** @rodboev
- **Created:** 2026-07-04
- **Tracking:** #4812
Refs #4812
---
## Problem
hermes-webui has no stable, cross-client contract for observing the lifecycle
of an individual session over SSE. Five or more future consumers — WebUI
reconnect/multi-tab, Android wrapper, iOS/PWA wrapper, desktop/TWA wrapper, and
test/CLI observers — each need a resumable, dedupe-safe event stream. Without a
shared contract, every client invents its own cursor, heartbeat, and event-type
semantics, multiplying coordination cost as new producers are added.
The maintainer asked for a docs-only RFC first, holding implementation until
sequence and replay semantics are settled (comment 2026-06-24T17:14:05Z,
2026-06-25T04:50:06Z on #4812). This document settles the contract vocabulary
against current source before any route is added.
## Goals
- Define the SSE envelope and event-type vocabulary for a proposed per-session
stream `GET /api/sessions/{session_id}/events`.
- Specify replay identity using the existing run-journal cursor model.
- Specify the snapshot fallback for stale or evicted cursors.
- Document the distinction from the existing global session-list stream.
- Record open implementation gates that must be resolved before the endpoint
ships.
## Non-goals
- This RFC does **not** implement `GET /api/sessions/{session_id}/events`. No
route, handler, or related code is added in this PR.
- This RFC does **not** modify `GET /api/sessions/events` (the existing global
session-list invalidation stream routed at `api/routes.py:12345-12346` and
implemented by `_handle_session_events_stream()` at `api/routes.py:16177`).
- This RFC does **not** replace or modify existing streams: `/api/chat/stream`,
`/api/approval/stream`, or `/api/clarify/stream`.
- This RFC does **not** introduce Android, iOS, or PWA client code.
- This RFC does **not** claim Android/iOS background reconnect behavior or
production proxy delivery; those require owner-reaching proof in a later
implementation PR.
- This RFC does **not** promise a new session-global sequence counter in Phase 1.
## Current source inventory
### Existing global session-list stream
`GET /api/sessions/events` is a **different endpoint** from the one this RFC
proposes. It is routed at `api/routes.py:12345-12346` and implemented by
`_handle_session_events_stream()` at `api/routes.py:16177`. It emits bare
`sessions_changed` events and keepalives for any change to the session list. It
is a global invalidation signal, not a per-session lifecycle stream. The proposed
`GET /api/sessions/{session_id}/events` is per-session and path-distinct.
### Heartbeat
`_SSE_HEARTBEAT_INTERVAL_SECONDS = 5` (`api/routes.py:1018-1030`) is the current
heartbeat interval for SSE streams. Phase 1 reuses this constant rather than
adding a separate configurable knob.
### Run-journal cursor and replay
Current replay identity is run/stream-scoped:
Line ranges in this inventory were verified against WebUI `master` when this
RFC was written. Function, constant, and endpoint names are the stable anchors
if source layout moves later.
- `_parse_run_journal_event_id()` (`api/routes.py:15673-15686`) and
`_parse_run_journal_after_seq()` (`api/routes.py:15688-15701`) parse the replay
cursor from the `after_event_id` / `after_seq` **query params** (not the
`Last-Event-ID` header — that header is the *proposed* new-endpoint contract
below, §Reconnect).
- `_runner_event_id()` at `api/routes.py:15765-15772` constructs the event `id`
field as `stream_id:seq`.
- SSE frames carry their `id:` via the `_sse_with_id()` helper, emitted on the
live `/api/chat/stream` path at `api/routes.py:15918`, on the runner-observe
path at `api/routes.py:15811`, and during journal replay at
`api/routes.py:15721` / `15734`.
- `_replay_run_journal()` reads events by `(session_id, stream_id)` at
`api/routes.py:15703-15735`.
- `api/streaming.py:6265-6285` writes current live agent streams to
`STREAMS[stream_id]`.
- `api/streaming.py:6620-6634` appends SSE events to the run journal and carries
per-item `event_id` into the live queue.
The existing run journal represents `session_id`, `stream_id`, `seq`, and
`event_id`, but **not** a session-global monotonic sequence. Phase 1 must not
promise a session-global counter because current source does not provide one.
## Proposed endpoint
```
GET /api/sessions/{session_id}/events
```
This endpoint is **path-distinct** from `GET /api/sessions/events`. The
`{session_id}` path segment is required; the global endpoint has no such segment.
Response: `Content-Type: text/event-stream`. Authentication and session
visibility checks reuse existing mechanisms.
## Envelope
Each SSE event carries a JSON payload with this structure:
```json
{
"schema_version": 1,
"session_id": "<session_id>",
"event_type": "<string>",
"event_id": "<opaque cursor>",
"stream_id": "<stream_id>",
"seq": <integer>,
"emitted_at": "<ISO-8601 UTC>",
"payload": { ... },
"meta": { ... }
}
```
- `schema_version`: integer, always `1` for Phase 1 events.
- `session_id`: the session this event belongs to.
- `event_type`: one of the event types listed in the taxonomy below.
- `event_id`: opaque client cursor (see Cursor and resume semantics).
- `stream_id`: the run journal stream this event came from, if applicable.
- `seq`: monotonic within a stream/run (see Cursor and resume semantics).
- `emitted_at`: server-side emission timestamp in ISO-8601 UTC.
- `payload`: event-type-specific data.
- `meta`: optional; reserved for tracing and debug metadata.
Server-generated events that do not originate in the run journal, currently
`heartbeat` and `session_snapshot`, need an explicit `event_id` / `stream_id` /
`seq` rule before implementation. This RFC records that as an implementation
gate rather than inventing values without source support.
## Event taxonomy (Phase 1 draft)
| event_type | Source | Description |
|---|---|---|
| `chat_delta` | run journal / live stream | Token or chunk from an assistant reply. |
| `tool_call` | run journal / live stream | Tool invocation record. |
| `tool_result` | run journal / live stream | Tool result record. |
| `approval_request` | run journal | Approval prompt sent to the user. |
| `clarify_request` | run journal | Clarification prompt sent to the user. |
| `run_started` | run journal | Run entered active state. |
| `run_finished` | run journal | Run reached a terminal state (complete, cancelled, error). |
| `session_snapshot` | server fallback | Current session projection; emitted when replay is unavailable. |
| `heartbeat` | server | Keepalive emitted on the `_SSE_HEARTBEAT_INTERVAL_SECONDS` cadence. |
The event-type table is a draft. The final table must be confirmed during
maintainer review before implementation.
## Cursor and resume semantics
`Last-Event-ID` is the standard SSE reconnect header. Clients send the last
`event_id` value seen on reconnect; the server uses it to resume replay from that
position.
**`event_id` is opaque to clients.** Its current source-compatible form is
`stream_id:seq`, as constructed by `_runner_event_id()` at
`api/routes.py:15765-15772`. Clients must treat it as an opaque string and must
not parse or construct cursor values.
**`seq` is monotonic within a stream/run.** It is not a session-global counter
and is not promised to increase monotonically across streams or runs. Phase 1
does not claim a pre-existing session-global sequence because current source does
not provide one.
**Clients dedupe by `event_id`.** If a reconnect causes overlap with already-seen
events, clients use `event_id` to detect and skip duplicates.
## Replay source
Phase 1 uses the **durable run journal** as the replay source for replayable
events. The live `STREAMS[stream_id]` queue (`api/streaming.py:6265-6285`) is
not a reliable replay source because it holds only recent in-memory state.
A future implementation must replay from the run journal via the existing
`_replay_run_journal()` path (`api/routes.py:15703-15735`) and fall back to the
snapshot mechanism when journal entries are unavailable for a given cursor.
## Snapshot fallback
When the `Last-Event-ID` cursor is evicted, expired, unknown, or refers to a
stream that is no longer replayable, the server must:
1. Emit a `session_snapshot` event containing the current session projection.
2. Continue the live stream from the present without pretending that missed
events were replayed.
`session_snapshot` is a recovery boundary, not proof of exact missed-event
replay. Clients receiving a snapshot must treat prior cursor state as invalid and
resync from the snapshot payload.
## Heartbeat
Phase 1 reuses `_SSE_HEARTBEAT_INTERVAL_SECONDS` (`api/routes.py:1018-1030`) for
heartbeat cadence. A new per-session configurable heartbeat knob is **not** added
in Phase 1. The implementation PR must follow whatever value the constant holds
at implementation time; it must not hard-code a separate interval.
## Security and privacy
- Reuse existing auth and session visibility checks. A client must not be able to
subscribe to events for a session it does not own.
- Payloads must not include credentials, raw provider API keys, or unsanitized
internal error details.
- `meta` fields are for tracing and debug metadata and must not carry
security-sensitive values in production.
## Implementation gates (open questions)
The following decisions must be resolved before any implementation PR for this
endpoint is accepted:
1. **Sequence semantics**: Maintainer must confirm that stream/run-scoped `seq`
(not session-global) is acceptable for Phase 1 clients.
2. **Retention policy**: How long are run journal entries retained for replay?
What is the eviction boundary that triggers the snapshot fallback?
3. **Event-type table**: The taxonomy above is a draft. The complete event-type
list must be confirmed during review of this RFC.
4. **Auth behavior on reconnect**: Does `Last-Event-ID` replay require the same
auth token, or can it continue across token refresh?
5. **Client proof**: At least one browser-based client (WebUI) and at least one
non-browser client (Android wrapper or CLI) must provide owner-reaching
reconnect proof before implementation closes #4812.
6. **Proxy and keepalive**: The 5 s heartbeat choice must survive real proxy
deployments. This requires manual-owner-proof or standards-doc evidence in the
implementation PR.
7. **Server-generated event identity**: Maintainer must confirm how
`heartbeat` and `session_snapshot` populate `event_id`, `stream_id`, and
`seq`, because those events do not originate in the run journal.
## Bypass risks
Future implementation work must not:
- Introduce an in-memory-only cursor that bypasses the run journal.
- Conflate `GET /api/sessions/events` (global session-list invalidation) with
`GET /api/sessions/{session_id}/events` (per-session lifecycle).
- Promise a session-global monotonic sequence without defining a migration from
the current stream/run-scoped model.
Tests in `tests/test_issue4812_session_sse_contract_rfc.py` assert these
boundaries so review catches regressions against this contract.
## Rollout plan
1. This RFC is accepted by maintainer review on #4812.
2. Retention and event-type decisions are confirmed.
3. Client proof (browser + non-browser) is provided.
4. An implementation PR adds `GET /api/sessions/{session_id}/events` following
this contract vocabulary.
5. Implementation PR closes #4812.

278
tests/test_issue4812_session_sse_contract_rfc.py Normal file
View file

@ -0,0 +1,278 @@
"""Docs/source contract tests for issue #4812 session SSE contract RFC.
These tests assert the invariants of docs/rfcs/session-sse-contract-v1.md
without requiring a running server or endpoint implementation.
"""
import pathlib
REPO = pathlib.Path(__file__).parent.parent
RFC = REPO / "docs" / "rfcs" / "session-sse-contract-v1.md"
RFC_README = REPO / "docs" / "rfcs" / "README.md"
CONTRACTS = REPO / "docs" / "CONTRACTS.md"
def _rfc() -> str:
return RFC.read_text(encoding="utf-8")
def _readme() -> str:
return RFC_README.read_text(encoding="utf-8")
def _contracts() -> str:
return CONTRACTS.read_text(encoding="utf-8")
def _rfc_section(text: str, heading: str) -> str:
marker = f"\n## {heading}\n"
assert marker in text, f"RFC must include a '## {heading}' section"
return text.split(marker, 1)[1].split("\n## ", 1)[0]
class TestRFCExists:
def test_rfc_file_exists(self):
assert RFC.exists(), f"RFC file not found: {RFC}"
def test_rfc_has_status_proposed(self):
assert "Status:** Proposed" in _rfc(), "RFC must have 'Status: Proposed' header"
def test_rfc_has_author(self):
assert "Author:** @" in _rfc(), "RFC must have 'Author: @...' header"
def test_rfc_has_created_date(self):
assert "Created:** " in _rfc(), "RFC must have 'Created: YYYY-MM-DD' header"
def test_rfc_has_tracking_issue(self):
assert "Tracking:** #4812" in _rfc(), "RFC must have 'Tracking: #4812' header"
def test_rfc_refs_not_closes(self):
text = _rfc()
assert "Refs #4812" in text, "RFC must use 'Refs #4812', not 'Closes #4812'"
assert "Closes #4812" not in text, "RFC must not use 'Closes #4812'"
class TestRFCIndexed:
def test_readme_indexes_rfc(self):
assert "session-sse-contract-v1.md" in _readme(), (
"docs/rfcs/README.md must list session-sse-contract-v1.md"
)
def test_readme_indexes_rfc_with_issue(self):
assert "#4812" in _readme(), (
"docs/rfcs/README.md index entry must reference #4812"
)
def test_contracts_indexes_rfc(self):
assert "session-sse-contract-v1.md" in _contracts(), (
"docs/CONTRACTS.md must reference session-sse-contract-v1.md"
)
class TestEndpointDistinction:
def test_rfc_names_proposed_per_session_endpoint(self):
assert "/api/sessions/{session_id}/events" in _rfc(), (
"RFC must propose GET /api/sessions/{session_id}/events"
)
def test_rfc_distinguishes_global_sessions_events(self):
text = _rfc()
assert "/api/sessions/events" in text, (
"RFC must mention existing GET /api/sessions/events"
)
assert "/api/sessions/{session_id}/events" in text, (
"RFC must distinguish per-session endpoint from global endpoint"
)
def test_rfc_states_global_endpoint_is_different(self):
text = _rfc()
assert "path-distinct" in text or "different endpoint" in text, (
"RFC must explicitly state the two endpoints are distinct"
)
def test_rfc_cites_current_global_endpoint_source(self):
"""The RFC's source anchors for the existing global stream must be
ACCURATE against current api/routes.py — verify the cited lines actually
contain what the RFC claims, rather than string-matching a fixed number
(which silently rots and codifies wrong anchors, #5513 gate finding)."""
import re
text = _rfc()
routes = (REPO / "api" / "routes.py").read_text(encoding="utf-8").splitlines()
# Pull every `api/routes.py:<start>-<end>` or `api/routes.py:<line>`
# anchor the RFC cites and confirm the referenced span exists.
anchors = re.findall(r"api/routes\.py:(\d+)(?:-(\d+))?", text)
assert anchors, "RFC must cite at least one api/routes.py source anchor"
for start, end in anchors:
lo = int(start)
hi = int(end) if end else lo
assert 1 <= lo <= len(routes), f"RFC cites api/routes.py:{start} beyond EOF ({len(routes)} lines)"
assert 1 <= hi <= len(routes), f"RFC cites api/routes.py:{end} beyond EOF ({len(routes)} lines)"
# The two load-bearing anchors must land on the real definitions.
def _anchor_line(label):
m = re.search(r"%s.*?api/routes\.py:(\d+)" % re.escape(label), text, re.DOTALL)
assert m, f"RFC must cite a routes.py anchor near {label!r}"
return int(m.group(1))
route_line = _anchor_line("routed at")
assert "/api/sessions/events" in routes[route_line - 1], (
f"RFC's routed-at anchor api/routes.py:{route_line} must be the "
f"/api/sessions/events route; got: {routes[route_line - 1].strip()!r}"
)
handler_line = _anchor_line("_handle_session_events_stream()")
assert "def _handle_session_events_stream" in routes[handler_line - 1], (
f"RFC's handler anchor api/routes.py:{handler_line} must be the "
f"_handle_session_events_stream definition; got: {routes[handler_line - 1].strip()!r}"
)
def test_rfc_run_journal_anchors_land_on_real_source(self):
"""Every named-symbol / emission anchor the RFC cites in the run-journal
inventory must land on the actual source token (not just be in-bounds),
so a stale line number can't silently pass (#5513 gate finding 2)."""
import re
text = _rfc()
routes = (REPO / "api" / "routes.py").read_text(encoding="utf-8").splitlines()
def _first_anchor_after(label):
m = re.search(r"%s.*?api/routes\.py:(\d+)" % re.escape(label), text, re.DOTALL)
assert m, f"RFC must cite a routes.py anchor near {label!r}"
return int(m.group(1))
# (label in RFC prose, token that must appear on the cited line)
checks = [
("_parse_run_journal_event_id()", "def _parse_run_journal_event_id"),
("_parse_run_journal_after_seq()", "def _parse_run_journal_after_seq"),
("_runner_event_id()", "def _runner_event_id"),
("_replay_run_journal()", "def _replay_run_journal"),
]
for label, token in checks:
line = _first_anchor_after(label)
assert 1 <= line <= len(routes), f"{label} anchor api/routes.py:{line} beyond EOF"
assert token in routes[line - 1], (
f"RFC's {label} anchor api/routes.py:{line} must contain {token!r}; "
f"got: {routes[line - 1].strip()!r}"
)
# The live-emission bullet must cite the real _sse_with_id call site.
emit_line = _first_anchor_after("live `/api/chat/stream` path at")
assert "_sse_with_id" in routes[emit_line - 1], (
f"RFC's live-emission anchor api/routes.py:{emit_line} must be an "
f"_sse_with_id() call; got: {routes[emit_line - 1].strip()!r}"
)
def test_contracts_distinguishes_both_endpoints(self):
text = _contracts()
assert "/api/sessions/{session_id}/events" in text, (
"CONTRACTS.md must mention proposed per-session endpoint"
)
assert "/api/sessions/events" in text, (
"CONTRACTS.md must mention existing global sessions/events endpoint"
)
class TestSequenceAndReplaySemantics:
def test_rfc_includes_last_event_id(self):
assert "Last-Event-ID" in _rfc(), "RFC must include Last-Event-ID"
def test_rfc_includes_event_id(self):
assert "event_id" in _rfc(), "RFC must include event_id"
def test_rfc_includes_stream_id(self):
assert "stream_id" in _rfc(), "RFC must include stream_id"
def test_rfc_includes_seq(self):
assert '"seq"' in _rfc() or "`seq`" in _rfc(), "RFC must include seq field"
def test_rfc_includes_session_snapshot(self):
assert "session_snapshot" in _rfc(), "RFC must include session_snapshot event"
def test_rfc_states_seq_is_stream_scoped(self):
text = _rfc()
assert "monotonic within a stream" in text or "stream/run-scoped" in text, (
"RFC must state seq is monotonic within a stream/run, not session-global"
)
def test_rfc_does_not_promise_session_global_counter(self):
text = _rfc()
assert "does not claim a pre-existing session-global sequence" in text or (
"not a session-global counter" in text
), (
"RFC must explicitly state Phase 1 does not claim a session-global counter"
)
def test_rfc_states_event_id_is_opaque(self):
assert "opaque" in _rfc(), "RFC must state event_id is opaque to clients"
def test_rfc_gates_server_generated_event_identity(self):
text = _rfc()
assert "Server-generated event identity" in text, (
"RFC must gate heartbeat and snapshot event identity before implementation"
)
assert "heartbeat" in text and "session_snapshot" in text, (
"RFC must name heartbeat and session_snapshot in the event identity gate"
)
assert "event_id" in text and "stream_id" in text and "`seq`" in text, (
"RFC must gate event_id, stream_id, and seq values for server events"
)
def test_rfc_names_run_journal_as_replay_source(self):
text = _rfc()
assert "run journal" in text.lower(), (
"RFC must name the run journal as the replay source"
)
def test_rfc_defines_session_snapshot_as_fallback(self):
text = _rfc()
assert "session_snapshot" in text
assert "fallback" in text.lower() or "snapshot fallback" in text.lower(), (
"RFC must define session_snapshot as the stale-cursor fallback"
)
def test_rfc_snapshot_is_not_exact_replay(self):
text = _rfc()
assert "not proof of exact missed-event replay" in text or (
"recovery boundary" in text
), (
"RFC must state snapshot is a recovery boundary, not exact missed-event replay"
)
class TestHeartbeat:
def test_rfc_references_heartbeat_constant(self):
assert "_SSE_HEARTBEAT_INTERVAL_SECONDS" in _rfc(), (
"RFC must reference _SSE_HEARTBEAT_INTERVAL_SECONDS rather than inventing a new constant"
)
def test_rfc_does_not_add_new_heartbeat_knob(self):
text = _rfc()
heartbeat_section = _rfc_section(text, "Heartbeat")
normalized = heartbeat_section.replace("*", "").lower()
assert "new per-session configurable heartbeat knob is not added" in normalized, (
"RFC must not add a new per-session heartbeat knob in Phase 1"
)
class TestDocsOnlyScope:
def test_rfc_states_no_endpoint_implementation(self):
text = _rfc()
assert "does **not** implement" in text or "does not implement" in text, (
"RFC must state it does not implement the endpoint"
)
def test_rfc_has_non_goals_section(self):
assert "Non-goals" in _rfc(), "RFC must have a Non-goals section"
def test_rfc_lists_implementation_gates(self):
text = _rfc()
assert "implementation gate" in text.lower() or "open question" in text.lower(), (
"RFC must list open implementation gates"
)
def test_contracts_preserves_no_implementation_warning(self):
text = _contracts()
assert "not authorize implementation" in text or (
"Proposed RFCs are review guardrails, not implementation authorization" in text
), (
"CONTRACTS.md must preserve warning that proposed RFCs do not authorize implementation"
)