From 3adc26a8c50adc1b73620c6116b70ee0d5f1fdbc Mon Sep 17 00:00:00 2001 From: Christian Bromann Date: Thu, 13 Aug 2026 09:17:13 -0700 Subject: [PATCH] feat(protocol): add Event.runId and sinceEventId Formalize producing-run identity on envelopes and a durable SSE seek cursor. Bump JS/Python packages to 0.0.19. --- streaming/README.md | 30 ++++++++++++++++----- streaming/js/package.json | 2 +- streaming/js/protocol.ts | 14 +++++++++- streaming/protocol.cddl | 10 ++++++- streaming/py/langchain_protocol/protocol.py | 4 ++- streaming/py/pyproject.toml | 2 +- 6 files changed, 50 insertions(+), 12 deletions(-) diff --git a/streaming/README.md b/streaming/README.md index ec567e0..2d76989 100644 --- a/streaming/README.md +++ b/streaming/README.md @@ -54,10 +54,16 @@ The `events` request body is an `EventStreamRequest`: "channels": ["messages", "updates", "lifecycle"], "namespaces": [[]], "depth": 2, - "since": 123 + "since": 123, + "sinceEventId": "evt_abc" } ``` +`since` resumes within the same connection's ring buffer (session `seq`). +`sinceEventId` seeks the durable event tape across reconnects (same id space +as WebSocket `ReconnectParams.lastEventId`). When both are set, +`sinceEventId` takes precedence for durable replay. + Each SSE connection is its own subscription. Closing the connection unsubscribes from that stream. A client may open multiple event streams for the same thread, for example one stream for low-latency model tokens and another for state or @@ -395,12 +401,22 @@ commands provide explicit reads and time-travel operations. ## Replay and Reconnection -Servers may keep a ring buffer of recent events per thread. Clients use sequence -numbers to recover missed events: - -- SSE clients pass `since` in `EventStreamRequest`. -- WebSocket clients call `subscription.reconnect` with `lastEventId` and the - subscriptions they want restored. +Servers may keep a ring buffer of recent events per thread. Clients recover +missed events with two different cursors: + +- `seq` / `since` — **connection-scoped**. Each new SSE/WebSocket session + renumbers from the start of that session's ring buffer. Use `since` only to + resume within the same connection after a brief gap. +- `eventId` / `runId` — **durable**. `eventId` is stable across reconnects + (and often anchored to an upstream stream entry). `runId` identifies which + run produced the event so clients can ignore replay from older runs after + hydrate or a deferred first subscribe. + +SSE clients pass `since` (session seq) and optionally `sinceEventId` (durable +cursor) in `EventStreamRequest`. WebSocket clients call +`subscription.reconnect` with `lastEventId` and the subscriptions they want +restored. When `sinceEventId` / `lastEventId` is set, the server seeks the +durable tape; session `since` remains for same-connection ring-buffer resume. The server replays matching buffered events after the requested point and then switches to live delivery. If the requested event is no longer buffered, servers diff --git a/streaming/js/package.json b/streaming/js/package.json index 7cc1249..f93b5cd 100644 --- a/streaming/js/package.json +++ b/streaming/js/package.json @@ -1,6 +1,6 @@ { "name": "@langchain/protocol", - "version": "0.0.17", + "version": "0.0.19", "description": "TypeScript bindings for the LangChain agent streaming protocol", "license": "MIT", "keywords": [ diff --git a/streaming/js/protocol.ts b/streaming/js/protocol.ts index 4760fa6..57bd5c4 100644 --- a/streaming/js/protocol.ts +++ b/streaming/js/protocol.ts @@ -318,6 +318,11 @@ export type Event = EventData & Extensible & { * Monotonic sequence number for ordering */ seq?: JsUint; + /** + * Producing run. Durable across connection-local `seq` resets so clients + * can ignore replayed events from older runs after hydrate / reconnect. + */ + run_id?: string; }; export type ResultData = RunResult | SubscriptionResult | AgentResult | InputResult | StateResult | EmptyResult; @@ -457,9 +462,16 @@ export type EventStreamRequest = Extensible & { */ depth?: number; /** - * Replay events after this seq number + * Connection-local seq resume (same session) */ since?: JsUint; + /** + * Durable cursor: replay events with eventId strictly after this value, + * then switch to live. Prefer this across reconnects; `since` remains for + * same-connection ring-buffer resume only. Same id space as + * ReconnectParams.lastEventId. + */ + since_event_id?: string; }; export type SubscriptionCommand = SubscriptionSubscribe | SubscriptionUnsubscribe | SubscriptionReconnect; diff --git a/streaming/protocol.cddl b/streaming/protocol.cddl index 51049ed..ed71af9 100644 --- a/streaming/protocol.cddl +++ b/streaming/protocol.cddl @@ -363,6 +363,9 @@ Event = { type: "event", ? eventId: text, ; Unique ID for reconnection (maps to SSE id:) ? seq: js-uint, ; Monotonic sequence number for ordering + ; Producing run. Durable across connection-local `seq` resets so clients + ; can ignore replayed events from older runs after hydrate / reconnect. + ? runId: text, EventData, Extensible, } @@ -539,7 +542,12 @@ EventStreamRequest = { channels: [+ Channel], ? namespaces: [* Namespace], ; Prefix-match these namespace paths ? depth: uint, ; Max depth below namespace prefix - ? since: js-uint, ; Replay events after this seq number + ? since: js-uint, ; Connection-local seq resume (same session) + ; Durable cursor: replay events with eventId strictly after this value, + ; then switch to live. Prefer this across reconnects; `since` remains for + ; same-connection ring-buffer resume only. Same id space as + ; ReconnectParams.lastEventId. + ? sinceEventId: text, Extensible, } diff --git a/streaming/py/langchain_protocol/protocol.py b/streaming/py/langchain_protocol/protocol.py index 0befe8d..dbddf02 100644 --- a/streaming/py/langchain_protocol/protocol.py +++ b/streaming/py/langchain_protocol/protocol.py @@ -304,6 +304,7 @@ class _EventFields(TypedDict): type: Literal["event"] event_id: NotRequired[str] # Unique ID for reconnection (maps to SSE id:) seq: NotRequired[JsUint] # Monotonic sequence number for ordering + run_id: NotRequired[str] class _EventVariant0(_EventFields, LifecycleEvent): pass @@ -387,7 +388,8 @@ class EventStreamRequest(TypedDict): channels: list[Channel] namespaces: NotRequired[list[Namespace]] # Prefix-match these namespace paths depth: NotRequired[int] # Max depth below namespace prefix - since: NotRequired[JsUint] # Replay events after this seq number + since: NotRequired[JsUint] # Connection-local seq resume (same session) + since_event_id: NotRequired[str] class SubscribeParams(TypedDict): channels: list[Channel] diff --git a/streaming/py/pyproject.toml b/streaming/py/pyproject.toml index 3f2ad1d..e6d0300 100644 --- a/streaming/py/pyproject.toml +++ b/streaming/py/pyproject.toml @@ -7,7 +7,7 @@ name = "langchain-protocol" description = "Python bindings for the LangChain agent streaming protocol" license = { text = "MIT" } readme = "README.md" -version = "0.0.17" +version = "0.0.19" requires-python = ">=3.10.0,<4.0.0" classifiers = [ "Development Status :: 3 - Alpha",