SSE auto-reconnect for axons; AbortSignal for poll-based waits#765
Merged
Conversation
Wrap SSE auto-reconnect in StreamBackedAPIPromise so asResponse/withResponse still work. Add withStreamAutoReconnect to axon subscribe (after_sequence) mirroring execution stdout/stderr streams. Co-authored-by: albert <alb-rl@users.noreply.github.com>
Extend poll() with optional signal and wire it through scenario run scoring waits, blueprint build waits, and disk snapshot completion. ScenarioRun.awaitEnvReady forwards signal to devbox awaitRunning. Co-authored-by: albert <alb-rl@users.noreply.github.com>
✅ Object Smoke Tests & Coverage ReportTest Results✅ All smoke tests passed Coverage Results
Coverage Requirement: 100% function coverage (all public methods must be called in smoke tests) ✅ All tests passed and all object methods are covered! View detailed coverage report
|
- Lazy-start the first stream request in withStreamAutoReconnect so APIPromise construction does not hit the network (fixes worker crashes). - ScenarioRun.awaitEnvReady passes undefined when no options; only forwards longPoll/polling/signal when set. - Axon.subscribeSse with no args calls axons.subscribeSse(id, undefined). Co-authored-by: albert <alb-rl@users.noreply.github.com>
raceAbort no longer uses Promise.race with a dangling abort listener. It attaches onAbort, awaits the wrapped promise via .then, and removes the listener on both success and failure (and onAbort removes itself when firing first). Co-authored-by: albert <alb-rl@users.noreply.github.com>
alb-rl
marked this pull request as ready for review
April 1, 2026 22:00
dines-rl
reviewed
Apr 1, 2026
| query?: AxonSubscribeSseParams, | ||
| options?: Core.RequestOptions, | ||
| ): APIPromise<Stream<AxonEventView>>; | ||
| subscribeSse(id: string, options?: Core.RequestOptions): APIPromise<Stream<AxonEventView>>; |
Contributor
Author
There was a problem hiding this comment.
overloading functions in typescript
dines-rl
reviewed
Apr 1, 2026
dines-rl
reviewed
Apr 1, 2026
| export async function withStreamAutoReconnect<Item>( | ||
| streamCreator: (offset: number | undefined) => Promise<Stream<Item>>, | ||
| export function withStreamAutoReconnect<Item>( | ||
| streamCreator: (offset: number | undefined) => APIPromise<Stream<Item>>, |
Contributor
Author
There was a problem hiding this comment.
it's part of stainless's core primitives, it's our SDK’s return type for most HTTP calls
dines-rl
reviewed
Apr 1, 2026
| let dataPromiseMemo: Promise<Stream<Item>> | undefined; | ||
| const getDataPromise = () => { | ||
| if (!dataPromiseMemo) { | ||
| dataPromiseMemo = (async () => { |
Contributor
There was a problem hiding this comment.
Memo? This all looks a little to complicated, did the OG reconnector not work?
Contributor
There was a problem hiding this comment.
Why are we creating a new one?
Contributor
Author
There was a problem hiding this comment.
memo = single-flight for the first request and for the constructed reconnecting stream promise, so all APIPromise entry points stay consistent and idempotent
dines-rl
approved these changes
Apr 1, 2026
dines-rl
reviewed
Apr 1, 2026
Drop public AxonSubscribeSseParams; resume offset is only supplied by withStreamAutoReconnect via getStream(afterSequence). subscribeSse is now (id, options?) like execution stdout/stderr streams. Co-authored-by: albert <alb-rl@users.noreply.github.com>
Contributor
|
CodeAnt AI is reviewing your PR. Thanks for using CodeAnt! 🎉We're free for open-source projects. if you're enjoying it, help us grow by sharing. Share on X · |
codeant-ai
Bot
added
the
size:L
dines-rl
self-requested a review
April 1, 2026 23:19
dines-rl
requested changes
Apr 1, 2026
![codeant-ai[bot]](https://avatars.githubusercontent.com/in/646884?s=60&v=4){kind=small}
Contributor
|
CodeAnt AI finished reviewing your PR. |
…tests - poll(): signal during delays, timeout vs abort, listener balance, lastResult - withStreamAutoReconnect: empty stream, getOffset undefined, asResponse/ withResponse, single-flight, StreamBackedAPIPromise lazy getData - Axons.subscribeSse: after_sequence on reconnect, headers, first GET query - Smoketests: awaitScored + signal, runs.awaitScored + signal, axon SSE, execution stdout stream Co-authored-by: albert <alb-rl@users.noreply.github.com>
- Axon subscribeSse: merge options.query with after_sequence; spread rest options so reconnect cursor is not overwritten by caller query - Executions streamStdout/Stderr: seed reconnect with parsed query.offset - Blueprints.awaitScored path: map resolveLongPollTimeoutMs into poll() - DiskSnapshots.awaitCompleted and Runs.awaitScored: same timeout mapping - createAndAwaitBuildCompleted: pass signal through to create() Tests: axon query+reconnect merge; execution initial offset + reconnect Co-authored-by: albert <alb-rl@users.noreply.github.com>
dines-rl
reviewed
Apr 1, 2026
|
|
||
| await expect( | ||
| run.awaitScored({ | ||
| polling: { pollingIntervalMs: 500, maxAttempts: 500, timeoutMs: 30 * 60 * 1000 }, |
Contributor
There was a problem hiding this comment.
What is this configuration? I don't think we need any of these right?
Contributor
Author
There was a problem hiding this comment.
this is the default polling config, it's been here for a while
dines-rl
reviewed
Apr 1, 2026
| }), | ||
| ).rejects.toThrow(LongPollAbortError); | ||
|
|
||
| try { |
Contributor
There was a problem hiding this comment.
Need a test for reeconnect that it reconnects at the right position IE the output starts from where it left off, need maybe 5 events in the axon and then have it disconnect and reconnect showing that it works.
Contributor
There was a problem hiding this comment.
We need some way of showing that this whole reconnect thing works and resumes from the last received SSE event
dines-rl
requested changes
Apr 1, 2026
Publish 5 tagged events, subscribe, wait for server 408 idle, publish 2 more, assert strictly increasing sequence and markers 1–7 (no duplicates). Co-authored-by: albert <alb-rl@users.noreply.github.com>
dines-rl
approved these changes
Apr 2, 2026
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
3 participants
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.
User description
Summary
withStreamAutoReconnect;after_sequenceis internal (resume from lastsequenceon 408).StreamBackedAPIPromise+ deferred first fetch preserveasResponse/withResponse.poll()supportssignalwith balanced abort listeners; wired through scenario/blueprint/snapshot waits.Testing
tests/polling.test.ts(signal + delays + timeout race + listener balance +lastResult)tests/lib/streaming-reconnection.test.ts(empty stream,getOffsetundefined path,asResponse/withResponse, parallelasResponse+await,StreamBackedAPIPromiselazygetData)tests/resources/axons-subscribe-sse.test.ts(first GET has noafter_sequence; reconnect sendsafter_sequence; custom headers)tests/smoketests/polling-streaming-flow.test.ts(RUN_SMOKETESTS=1) —awaitScored+signal,runs.awaitScored+signal, axon SSE, executionstreamStdoutUpdatesRun unit:
jest tests/polling.test.ts tests/lib/streaming-reconnection.test.ts tests/resources/axons-subscribe-sse.test.tsRun smoke:
yarn test:smoke(orRUN_SMOKETESTS=1 jest tests/smoketests/polling-streaming-flow.test.ts)Slack Thread
CodeAnt-AI Description
Add abort support to long polls and keep reconnecting streams compatible with response helpers
What Changed
AbortSignalis cancelled, including the delay between polling attempts and any in-flight wait.Impact
✅ Cancel long waits on demand✅ Fewer stuck blueprint and scenario waits✅ Fewer dropped axon event streams after idle timeouts💡 Usage Guide
Checking Your Pull Request
Every time you make a pull request, our system automatically looks through it. We check for security issues, mistakes in how you're setting up your infrastructure, and common code problems. We do this to make sure your changes are solid and won't cause any trouble later.
Talking to CodeAnt AI
Got a question or need a hand with something in your pull request? You can easily get in touch with CodeAnt AI right here. Just type the following in a comment on your pull request, and replace "Your question here" with whatever you want to ask:
This lets you have a chat with CodeAnt AI about your pull request, making it easier to understand and improve your code.
Example
Preserve Org Learnings with CodeAnt
You can record team preferences so CodeAnt AI applies them in future reviews. Reply directly to the specific CodeAnt AI suggestion (in the same thread) and replace "Your feedback here" with your input:
This helps CodeAnt AI learn and adapt to your team's coding style and standards.
Example
Retrigger review
Ask CodeAnt AI to review the PR again, by typing:
Check Your Repository Health
To analyze the health of your code repository, visit our dashboard at https://app.codeant.ai. This tool helps you identify potential issues and areas for improvement in your codebase, ensuring your repository maintains high standards of code health.