Repository navigation
J3: HttpSseTransport event IDs, replay dedupe, 410 gap -> resume - #33
Merged
Merged
Conversation
- Records the id of each processed event as the cursor (lastEventId getter), sends it on reconnect as the Last-Event-ID header (for EventSource polyfills that take headers) and the last_event_id query parameter, plus the legacy lastEventId name. Authorization is passed to such polyfills too. - Dedupes events by id (bounded window) so a replay is not processed twice; an undecodable event still advances the cursor. - A closed stream with a cursor is probed; a 410 discards the cursor, opens a stream with none and emits "gap" once it is open. Peer resumes the session on "gap" (last_received_id = last event id) and reports onSessionLost when that is impossible or fails. - Test support: a Node http server stub following transport-http vectors and a vector runner that executes every transport-http vector over HTTP; client tests drive the real transport against the stub. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved issues can misroute EventSource URLs, carry replay state across sessions, repeat terminal callbacks, and allow stream tests to pass on non-stream responses.
4 open issues
What changed in this PR
Adds SSE replay and gap recovery to the HTTP transport, building on J2’s completion validation and replay redaction.
Changes:
- Tracks event IDs, sends replay cursors, and deduplicates replayed events within a bounded window.
- Detects HTTP 410 gaps, opens a cursorless stream, and triggers session resume or session-loss reporting.
- Adds an HTTP server stub, transport-vector runner, and client recovery tests.
| File | Description |
|---|---|
src/transport.ts |
Adds cursor tracking, replay deduplication, and gap detection. |
src/peer.ts |
Resumes sessions after gaps and reports recovery failures. |
test/support/eventsource.ts |
Provides a fetch-based EventSource for controlled tests. |
test/support/sse-stub.ts |
Simulates session-bound SSE replay, gaps, and recovery. |
test/support/transport-vector-runner.ts |
Executes transport vectors over HTTP and checks responses. |
test/transport-sse-client.test.ts |
Tests cursor handling, deduplication, and session recovery. |
test/transport-vectors.test.ts |
Runs transport-http vectors against the server stub. |
Review effort: Balanced
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
…, URL, runner - Peer ignores a terminal notification (complete/error) of a flow it already processed, which a resume re-sends with a fresh event ID; per session, cleared on a new/closed session and for a flow ID that starts a new flow. - HttpSseTransport.setSessionId drops the cursor and seen IDs when another session is bound (IDs are unique only within a session); a seeded cursor and a same-session rebind are kept. - The event stream URL is built by touching only the query string, so relative, protocol-relative and fragment-carrying URLs resolve as supplied. - The vector runner rejects a non-stream response where a stream is expected and checks content_type/body of expected non-stream responses. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🟡 Changes recommended
Terminal-flow tracking can discard valid events or repeat callbacks, while stale cursor parameters can block automatic recovery after a 410 response.
5 open issues
4 resolved since last review
Review effort: Balanced
Merges j2-completion (flow context via beginFlow) and reworks terminal dedupe on top of it: - A flow is marked terminal only after its notification passed validation (or was refused, whose synthesized error is delivered); a notification the validator rejected no longer discards the next valid one. - A cancellation already processed is acknowledged again without invoking the handler. - Reusing a flow id clears the old marker only once the start is not refused as duplicate/over capacity, and a failed start restores it, while terminal events of an accepted replacement flow that arrive before its start response are processed. - Terminal markers are never evicted (the set lives as long as the session). - Reserved cursor parameters in the supplied events URL are removed when there is no cursor (after a 410), so recovery does not draw another 410. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
…ls, marker identity, serialized recovery (merges j2-completion) - Terminal markers record HOW a flow ended; only a cancelled flow's repeat is acknowledged without the handler, a flow that ended by complete/error still reaches onFlowCancel on its first cancel. - Cancellations in flight are shared per flow: a repeat arriving meanwhile gets the first one's outcome (handler runs once); a failure clears the state so a retry is possible. - A failed start restores the old marker only while its own record is still the flow's, so a later flow's terminal is never discarded as a repeat. - Gap recoveries are serialized (one follow-up at most), so a second resume never presents the token the first already rotated. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🟡 Changes recommended
A delayed cancellation can incorrectly terminate a replacement flow and suppress its completion or error, while synchronous handler failures can prevent cancellation retries.
2 open issues
4 resolved since last review
Review effort: Balanced
…merges j2-completion) Merges j2-completion (which now owns the terminal/flow-state bookkeeping) and: - A pending cancellation is tied to the flow generation it was issued for: when it finishes it finalizes only that generation, so a replacement flow under the same id keeps its context and is not marked terminal, and a replacement's own cancellation does not share the old one's cached promise. - The in-flight entry is installed before the handler runs, so a synchronous throw also clears it and a retry reaches the handler. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🟡 Changes recommended
Explicit and automatic resumes can race on the same token, unsuccessful resumes bypass failure handling, and stale cancellation state can suppress flow outcomes after a session reset.
3 open issues
2 resolved since last review
Review effort: Balanced
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
…sume (merges j2-completion) Merges j2-completion (terminal history now only for known flows) and: - resumeSession itself is serialized, so an explicit resume and gap recovery (or two explicit ones) never present a token the previous one rotated. - A resumed:false result throws (SessionExpired), so recovery reports onSessionLost and nothing is taken from it. - Pending cancellations are tied to the flow-state epoch (incremented by every reset) as well as the flow record, and cleared on reset: a same-id cancellation in a new session does not share the old promise and the old handler finishing late cannot re-label a flow of the new session. A cancel of an already ended flow upgrades its marker without growing the history. - Tests that relied on unknown flows being remembered now start the flows first. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🟡 Changes recommended
Critical race conditions allow stale recovery to corrupt a replacement session or report spurious session loss, while unstable flow-generation tracking permits duplicate cancellation and mislabels replacement flows as cancelled.
2 open issues
3 resolved since last review
Review effort: Balanced
- Cancellation bookkeeping uses a per-flow-id generation that survives the flow's completion and changes only when a new flow starts under the id (and on reset): a replay shares a pending cancellation even if the flow completed meanwhile, a finished cancellation records its outcome, and a cancellation that had no record cannot relabel a replacement flow. - Gap recovery is bound to the session it was queued for (the flow-state epoch): the epoch is checked when the recovery gets its turn on the resume chain and after the response, a stale result throws an internal error that is not reported as a loss, and recovery queued before close() is discarded. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🔵 Needs a closer look
Human assessment is needed to evaluate unbounded flow-generation retention, cancellation replay races during restart rollback, and a test gap that allows duplicate SSE event IDs.
No open issues 🎉
2 resolved since last review
Review effort: Balanced
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
…Session (merges j2-completion) - Flow generations are reclaimed when nothing needs them (flow not active, no pending cancellation), so locally finished flows and succeeded cancellations leave no entries behind. - An explicit resume notes the flow epoch when it is requested and is dropped if the session was closed or replaced while it queued (no resume sent for the replacement with the old session's lastReceivedId). - A 410 probe whose session is no longer the bound one is discarded (no gap, no error for the new session). - HttpSseTransport.clearSession(): forget the binding, cursor and seen ids and close the stream when a session ends. - Test stub: a flow.start under a reused id clears that id's retained progress and terminal notification. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
…s, lost-session binding - HttpSseTransport: a stream is tied to the session it was opened for; once the session is no longer the bound one (or the stream was replaced/closed) nothing it delivers reaches the application or the cursor/seen ids, and handling a message that ends the session no longer leaves that event's id as the next session's cursor. setSessionId to another session closes the old stream and reopens one for the new session if one was open; clearSession remembers that a stream was open and reopens it when the next session is bound (clean cursor). - Peer drops the transport's session binding before it reports a lost session (no token, failed resume), so the documented "create a new session" fallback does not send the dead session's id. - A flow id's generation is kept while ANY cancellation for it is outstanding (count, not just the latest map entry), so an older pending cancellation can never match a later flow's reused generation. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🟡 Changes recommended
Critical session-isolation bugs allow stale cancellation cleanup to corrupt a new session’s flow state and cause an SSE stream opened before session binding to silently drop subsequent messages.
2 open issues
3 resolved since last review
Review effort: Balanced
…sion header, epoch-guarded cleanup - A stream opened before the first session binding is closed and reopened for that session (events are delivered; a seeded cursor survives the first bind). - A session.create POST carries no Wmp-Session-Id even while a session is bound (the server refuses one, Transport 4.3.1); the binding itself is kept in case the create fails. - A cancellation of a reset session finishing late no longer decrements the new session's cancellation count or reclaims its generation (guarded by the epoch). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
There was a problem hiding this comment.
🔵 Needs a closer look
Human assessment is needed to evaluate critical session-recovery and cancellation races that can create duplicate replacement sessions or invoke cancellation callbacks more than once.
No open issues 🎉
2 resolved since last review
Review effort: Balanced
HttpSseTransport.send puts the requested session id of a wmp.session.resume in Wmp-Session-Id (header and body name the same session) without moving the bound session; creates still carry none. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
|
Comment on lines
+1075
to
+1076
| if (typeof t.clearSession === "function") t.clearSession(); | ||
| this.onSessionLost?.(err); |
Comment on lines
+383
to
+393
| const id = typeof ev.lastEventId === "string" && ev.lastEventId !== "" ? ev.lastEventId : undefined; | ||
| if (id !== undefined && this.seenIds.has(id)) return; | ||
| try { | ||
| this.emit("message", decodeMessage(ev.data, { maxSize: 4 * 1024 * 1024 })); | ||
| } catch (err) { | ||
| this.emit("error", err instanceof Error ? err : new Error(String(err))); | ||
| } finally { | ||
| // Handling the message may have ended the session (a session.close event): | ||
| // its id must not become the next session's cursor. | ||
| if (id !== undefined && current()) this.recordEventId(id); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Parity chunk J3, stacked on #32 (base j2-completion; retarget as the stack merges).
HttpSseTransport: records the last event id (lastEventIdgetter), sends it asLast-Event-IDheader (polyfills) +last_event_idquery + legacylastEventId, dedupes replays by event id (bounded window), and on a 410 discards the cursor, opens a cursorless stream and emitsgaponce open.Peerresumes ongap(and reportsonSessionLostwhen it cannot).transport-http.jsonand a runner executing all 15 transport-http vectors over HTTP (runner proven non-vacuous by breaking the stub); client tests run the real transport against it.Non-vacuity: every client behaviour test was proven by reverting the behaviour.
🤖 Generated with Claude Code
https://claude.ai/code/session_011tAXPr6RCod7mUPqXo1pRn