🤖 feat: stream engine becomes the AppFiberScope occupant — dispose() aborts and awaits in-flight streams (Wave 4 PR 1) - #4070
Conversation
This comment has been minimized.
This comment has been minimized.
Dogfooding evidence (PR 1 notes — long form)Setup: Branch, two flowing streams in flight (run 2)Branch, one flowing stream (run 1)(The per-stream Baseline
|
|
@codex review |
|
@codex security review |
|
Codex Review: Didn't find any major issues. More of your lovely PRs please. Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
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". |
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
|
CI status note: both Codex loops are clean on |
…tics for the engine supervisor
…tch cancels; guard completed streams from abort bookkeeping
…once the final message is committed
…ettles every stream exactly once
…efore the explicit teardown
…per supervised stream
728df0d to
b1d2a55
Compare
|
Rebased onto @codex review |
|
@codex security review |
This comment has been minimized.
This comment has been minimized.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b1d2a55767
ℹ️ 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".
… treat post-abort iterator rejections as cancellation
|
@codex security review |
|
Codex Review: Didn't find any major issues. Delightful! Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
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". |
Summary
StreamManagerbecomes the first (and intended) occupant ofAppFiberScope: every started stream is wrapped in one supervisor fiber, soServiceContainer.dispose()(desktop,xum server, ACP) and the CLI cleanup lists now abort and await in-flight streams — partial flushed with usage,stream-abort(abortReason: "system") delivered, the interrupted assistant message committed intochat.jsonl,partial.jsonremoved — before the bridges and sessions are torn down. The same PR closes the two adjacent cancel races the shutdown path would have widened (Wave 4 plan D3): a cancel landing after the stream loop finished no longer resurrectspartial.jsonor emits a second terminal event, and concurrent cancellers (user stop racing dispose) join one cleanup.Background
Until now
dispose()had nostreamManagerstep: an in-flight stream died with the process and was reconciled frompartial.json(≤ 500 ms stale) on the next load, leaving an empty assistant placeholder row inchat.jsonluntil then.AppFiberScope(Phase 11) was built for exactly this and had no occupant. This is PR 1 of the Effect migration Wave 4 plan (appended below): D1 — the fiber is the ownership/supervision unit, theAbortSignalstays the cancellation transport; D2 — supervisor topology; D3 — adjacent cancel races; D4 — the 2 s close bound stays (outer budgets are 5 s).Implementation
StreamManager.superviseEngine(D2): afterstreamInfo.processingPromiseis assigned (byte-identical) it forksEffect.promise(() => processingPromise)(zero-arity thunk → rc.112 allocates no internal AbortController) intoengineScopewith{ startImmediately: true }.onInterrupt→Effect.uninterruptible(Effect.promise(async () => { await cancelStreamSafely(ws, info, "system"); await info.completionController.promise; }));catchDefect→log.warn. A stream that finishes on its own exits the fiber, which removes its scope finalizer (no residue). A stream started after the scope closed is interrupted synchronously byforkInand aborted right away (fail-closed during shutdown, pinned inappFiberScope.test.ts). The pre-registration window (pendingStreamStarts) stays unsupervised (documented).engineScope?: Scope.Closeableis a trailing optional 6th constructor parameter, wired fromAppFiberScopeTaginStreamManagerLive(di/layers/core.ts;AppFiberScopeTagadded toCoreInputTags— both roots already provide it viaruntimeSeams). Defaultundefined= today's behavior for every direct construction (tests,aiService.tscompat path).cleanupAbortedStream: afterawait processingPromise, return ifterminalCompletion !== undefined(completed/failed while the cancel was in flight). Plus apartialRetiredmarker set in the completion path right beforedeletePartial:flushPartialWritebecomes a no-op afterwards, because the pre-abort flush incancelStreamSafely(not only the abort bookkeeping) re-createdpartial.jsonwhen the cancel landed betweendeletePartialandCOMPLETED.cancelStreamSafely: per-streamcancelPromiselatch, checked and assigned synchronously at entry (after the existingCOMPLETEDearly return, before the firstawait).checkSoftCancelStreamshares the latch (??=) so a soft interrupt racing a hard cancel/dispose also yields exactly onestream-abort.appRuntime.tscontract ("Deliberately not done" updated; step 2 now says what it aborts),appFiberScope.ts,serviceContainer.tsstep comment. New[shutdown] streamManager.abortStream { workspaceId, messageId, ms }debug line per supervised stream (shutdownStep style).Net product change: 5 files, +196/−42 (≈ 70 non-comment lines added).
PR 1 notes
Pre-work findings
processStreamWithCleanupnever rejects — body istry { … } catch { await handleStreamFailure } finally { … }; a throw fromhandleStreamFailure/finallycould still reject, butstartStreamassignsprocessingPromise = processStreamWithCleanup(...).catch(log.error), so the promise the supervisor wraps never rejects.catchDefectstays as belt-and-braces.appFiberScope.test.ts, +2 cases):closeScopeBoundedinterrupts a fiber suspended onEffect.promisewithout settling the wrapped promise, runs the asynconInterruptfinalizer to completion, and resolves only afterwards (["finalizer-start", "finalizer-end"]).completionController.promise, i.e. aftercleanupAbortedStream(writePartialwith usage) →emitStreamAbort→ AIService sink (readPartial→commitPartial→deletePartial→emit("stream-abort")to agentSession/taskService/analytics listeners) →.finally(settle). The only work left after it is the EventEmitter listeners' own async continuations (turn-phase transitions, task-handle settlement) — the same as today'ssystemaborts fromtaskService. Durable orderwritePartial → commitPartial → deletePartialholds; a force-exit between any two steps leaves eitherpartial.json(recovered byagentSession.init'scommitPartial) or an already-committed row whosehistorySequenceupdate-or-append is idempotent. No OFF-RAMP condition fired.[shutdown] AppFiberScope closed { ms }(xum server, SIGTERM,XUM_LOG_LEVEL=debug,script -f): idle 1 ms; one flowing Sonnet 5 stream 56 ms; two flowing streams 95 ms (per-streamabortStream91 / 94 ms, parallel); wedged provider (scratch env-gatedfullStreamthat never yields and ignores abort, not committed) 2050 ms with theteardown timed outwarning,AppRuntime disposed5 ms (idempotent re-close), process exit 2.10 s after SIGTERM — inside the 5 s force-exit. Baselinemain: 1 ms (nothing supervised).forkInpinned (appFiberScope.test.ts):startImmediately: trueinto an already-closed scope runs the body to its first async boundary and then theonInterruptfinalizer synchronously (["body-started", "interrupted"], fiber exit is a failure).streamManager.test.tspins the product consequence (late start →{ status: "aborted", abortReason: "system" }, onestream-abort, registry empty).Deviations from the plan (all additive)
completionController.promise(plan:cancelStreamSafelyonly).cancelStreamSafelyreturns before abort delivery — thestream-abortsink is where AIService commits the partial — so without this the close would resolve withpartial.jsonstill present and the commit racingdesktopBridgeServer.stop(). This is what makes STOP criterion Better authentication UX #1 ("partial.jsonabsent immediately after exit") true.partialRetired(see D3(a) above): the plan's guard alone left the pre-abort flush as a resurrection path; the D3(a) test was red against the guard-only variant on exactly that assertion.checkSoftCancelStreamjoins the latch: dispose during a pending soft interrupt (queued-message preemption at tool-end) is the same double-cleanup shape D3(b) fixes for hard cancels.[shutdown] streamManager.abortStreamdebug line — there was no log line for the abort at all, so the transcript could not show the plan'sstream-abort → AppFiberScope closed → desktopBridgeServer.stoporder; it now does.xum runCtrl-C transcript:xum runinstalls no SIGINT handler (onlycli/server.tsdoes), so Ctrl-C is Node's default immediate termination (exit 130) and the cleanup list — includingappFiberScope.close— runs only on normal completion. The CLI-root path is pinned by the newcoreServicesRoot.test.tscase instead; adding a signal handler toxum runis out of scope.Dogfooding evidence (transcripts + before/after in the first comment)
xum server, SIGTERM 3 s after first text)AppFiberScope closedpartial.jsonright after exitchat.jsonlrowmain@ 3c06630 (baseline)partial: true, 1172 charsRestart UX (screenshot below, branch): the persisted interrupted text is shown above the
INTERRUPTEDbarrier; the workspace then auto-resumed the turn on startup (agentSession startup auto-retry —"system"aborts do not suppress it). Verified the baseline does the same after it commits the partial on load ([STREAM MESSAGE]right after startup onmain), so the only behavioral difference is when the partial is committed (at shutdown vs. next load). Not exercised headless: Electronbefore-quit(samedispose(); CI e2e).Lessons for PR 3 (
initialize()as a startup effect)runPromise/facades reject with the raw error, so error identity is free — but anyEffect.promisethunk wrapping anawaited step must beasyncand the whole pipeline needscatchDefectif the facade must never reject.dispose()are cheap (disposeAppRuntimeafter a timed-out scope close took 5 ms becauseScope.closeis idempotent); a startup timeout should equally not try to unwind the abandoned step — parity with the existing "throwing step" path is the whole design.[startup] <step> { ms }lines are already the right measurement surface; this run's cold start showedtaskService.initialize114 ms andworkspaceService.initialize83 ms, so a 60 s per-step bound is ≥ 500× the slowest observed step.xum runparity caveat in mind: the CLI roots have no signal handling, so any "dispose on failed initialize()" work only applies where a dispose path exists (cli/server.tsis the one PR 3 adds).Validation
streamManager.test.ts(6: flowing close, wedged bound, late start, D3(a), D3(b) race, 50-stream residue),streamManager.chaos.test.ts(1 new fuzz variant, existing cases untouched; also ran 7 extra seeds locally),serviceContainer.test.ts(dispose aborts + awaits a real stream beforedesktopBridgeServer.stop(), checkspartial.json/chat.jsonl),coreServicesRoot.test.ts(same viacloseScopeBounded(appFiberScope)),appFiberScope.test.ts(2 probes).stream-aborts).streamManager.test.ts,streamManager.chaos.test.ts,streamManager.modelOnlyNotifications.test.ts,aiService.test.ts,agentSession.disposeRace.test.ts,agentSession.sinceReplayContract.test.ts,serviceContainer.test.ts,coreServicesRoot.test.ts,di/*.test.ts,taskService.test.ts,workspaceService.test.ts(one unrelatedFakeAIServicemetadata test flaked once in a combined run, green isolated and on rerun),turnRequestBuilder.test.ts;make static-check.catchDefect; async thunks), spy seams unchanged (processStreamWithCleanup,createStreamResult,createStreamAtomically,startStream; constructor arity trailing-optional;Reflect.set/gettargets valid), sync-start (processingPromiseassigned before the fork;runSync(forkIn)synchronous), no constructor side effects, zero-suspension latch.Risks
xum server/ACP): a flowing stream now costs one chunk (~50–100 ms) at dispose; a wedged provider costs the existing 2 s bound and a warning — identical outcome to today's process exit. Rollback: revert the onecore.tswiring line (engineScopeundefined → today's behavior); the D3 guards are independent bug fixes.abortReason/abandonPartialwin; a discard racing a system cancel by a few ms keeps the partial instead of dropping it (previously both cleanups ran and emitted two aborts)."system"(fail-closed): the caller sees a normalOk(handle)whose completion settlesaborted, and the turn is recovered on next load exactly like any othersystemabort.📋 Implementation Plan
Effect migration — Wave 4: finish the concurrency/lifecycle core
Bounded wave: 4 PRs (PR 4 optional), explicit STOP criterion, explicit OFF-RAMPs. Plan only; nothing here is implemented.
0. Thesis check (coordinator's judgment vs. evidence)
Thesis: Effect's payoff in this app is structured concurrency + interruption-safe lifecycles in the orchestration core (still Promise + AbortController).
Verdict: holds for the stream engine; only half-holds for turn handles.
ServiceContainer.dispose()never stops or awaits in-flight streams (serviceContainer.ts:478–537has nostreamManagerstep); an in-flight stream dies with the process and is recovered on next load frompartial.json(≤ 500 ms stale,PARTIAL_WRITE_THROTTLE_MS,streamManager.ts:776).AppFiberScopeexists precisely for this and has no occupant. A supervised per-stream fiber is the right tool.interruptWorkspaceTurnFromUncorrelatedStreamEnd,workspaceTurnManager.ts:4220–4307: any uncorrelated stream-end after the prompt index settles the handleinterrupted), not a Promise-vs-fiber structure bug. Turn handles are persisted records (taskHandleStore.upsertWorkspaceTurn) spanning multiple streams (tool-call continuations are deferred viahasSameTurnContinuation,:4491) and surviving restarts; a fiber/Deferred can only model the in-process waiter and would not fix correlation. Open PR [task-service] 🤖 fix: preserve turns across synthetic wake ends #3949 fixes the predicate in Promise idiom and is Codex-green. Wave 4's turn-handle PR therefore becomes "codify the settlement invariant + prove the class is gone", not "fiberize handles" (D5 below).Corrected baseline numbers (measured this workspace): 46/470
src/nodenon-test files importeffect(coordinator said 35); 113 directEffect.run*sites outsidedi/in 16 files; 226Effect.gen; 9TaggedErrorclasses; effect4.0.0-rc.112,@orpc/*1.14.11; effect v4 is not GA (rc line still current).1. Verified current state (evidence the design rests on)
Stream engine (streamManager.ts)
startStream(:4723–4901): per-workspace mutex →new AbortController()+linkAbortSignal(:4771–4772) →resourceScope = Scope.makeUnsafe()(:4777) → temp-dirEffect.acquireRelease(:4802–4824) →createStreamAtomically→streamText(:2244,abortSignal: abortController.signal:2250) → registered inworkspaceStreams(:2463) →streamInfo.processingPromise = this.processStreamWithCleanup(...)fire-and-forget (:4876–4882) → returnsOk({ messageId, completion }).processStreamWithCleanup(:3331–4089, plain async):while(true)retry loop;for await (part of fullStream)(:3358–3837) with abort check at loop head (:3361); post-loopif (!signal.aborted)gate (:3849) → completion path (deletePartial:3981,updateHistory:3989,recordSessionUsage:4001,state = COMPLETED:4017, emitstream-end:4023,terminalCompletion:4024); error path →handleStreamFailure(:4094–4112) →persistStreamErrorwrites error partial;finally(:4052–4088): release MCP lease,Effect.runFork(Scope.close(resourceScope))(:4064–4066), unlink abort,workspaceStreams.delete,eventSpine.emit("stream.end"),completionController.settle.stopStream(:5043–5111) →cancelStreamSafely(:1766–1800):if (state === COMPLETED) { await processingPromise; return }→state = STOPPING→flushPartialWrite→abortController.abort()→cleanupAbortedStream(:1828–1951):await processingPromise→ usage →writePartial(:1876–1910) →emitStreamAbort→settle({status:"aborted"}). No completed-guard after the await (verified:1838–1951): a cancel landing between:3849and:4017re-writespartial.jsonafterdeletePartialand emitsstream-abortafterstream-end(pre-existing window; dispose() will widen its exposure).cancelStreamSafelyis also not idempotent for concurrent callers (onlyCOMPLETEDis checked).stream-abort(aiService.ts:355–377):abandonPartial ? deletePartial : commitPartial → deletePartial(fire-and-forget listener).HistoryService.commitPartial(historyService.ts:1963–2061) — strips error metadata,hasCommitWorthyParts, stale-epoch check, update-or-append byhistorySequence, delete partial; invoked fromagentSession.init(:5002),aiService.streamMessage(:886), stream-abort (:364),duplicateWorkspace.StreamAbortReason = "user" | "startup" | "system"(src/common/orpc/schemas/stream.ts:295).Reflect.set(streamManager, "tokenTracker" | "createStreamResult")(streamManager.chaos.test.ts:130–134, 238–242);streamManager.test.tsReflect.setonprocessStreamWithCleanup(:2787),createStreamAtomically(:2783),createTempDirForStream,cleanupStreamTempDir,Reflect.getonworkspaceStreams,schedulePartialWrite, …;modelOnlyNotifications.test.tscallsprocessStreamWithCleanupdirectly (:93, :187);aiService.test.tsspiesstartStream,generateStreamToken,createTempDirForStream,isResponseIdLost. Constructor:(historyService, sessionUsageService?, getProvidersConfig?, eventSink = noop, runner = defaultEffectRunner)(:801–813);effectRunnerused at:1152, :1154, :1169only.workspaceId+messageId;stream-end/stream-abort/errorcarrymetadata.muxMetadatawhen the prompt had it.DI / shutdown / startup
AppFiberScopeLiveis inCoreLive'sruntimeSeams(di/layers/core.ts:644), so both roots have it;StreamManagerLive(core.ts:226–238, stage S2b) already yieldsEffectRunnerTag. CLI cleanup lists includeappFiberScope.close(cli/run.ts:1579,cli/workflow.ts:286).APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS = 2000,APP_RUNTIME_DISPOSE_TIMEOUT_MS = 2000; outer budgets are 5000 ms on both desktop (desktop/main.ts:1297Promise.racevssetTimeout(5000)) andxum server(cli/server.ts:236–243force-exit). The scope bound cannot grow without changing outer budgets.node_modules/effect/dist/internal/effect.js:2264:forkInregisters a scope finalizer and removes it when the fiber completes (no leak), and interrupts immediately if the scope is already closed (streams starting mid-shutdown fail closed).Effect.promise(evaluate: (signal) => PromiseLike),Effect.onInterrupt,Effect.forkIn(_, scope, { startImmediately? }),Stream.toAsyncIterableWith(context),Stream.provideContextall exist.ServiceContainer.initialize()(serviceContainer.ts:297–362): six awaitedinitialize()s wrapped inrecordStep(durations only, no catch, no timeout) + three syncstart()s + two fire-and-forget sweeps. Failure handling: desktopStartup Faileddialog +app.quit()(desktop/main.ts:1249–1265);cli/server.ts:136uncontained; ACPserverConnection.ts:205–216dispose + rethrow;tests/ipc/setup.ts:85no catch. No outer timeout anywhere.streamBridge.subscriptionIterable(orpc/streamBridge.ts:176) →Stream.toAsyncIterable(...)on the global runtime, 19 call sites inrouterSubscriptions.ts; heartbeat viaEffect.sleepinforkScoped(:145–152).streamBridge.test.tshas 11 real-time waits, but only 3 are clock-bound (:2071 ms initial delay,:241heartbeat 10 ms,:25510 ms laziness); 8 arewaitFor(listenerCount…)readiness polls that TestClock cannot replace.Turn handles + open PRs
{ handleId "wst_…", ownerWorkspaceId, workspaceId, turnId, messageId, status, attentionPolicy, disposableWorkspace }; prompt carriesmuxMetadata: { type:"workspace-turn-task", taskHandleId, ownerWorkspaceId, turnId }(workspaceTurnManager.ts:1420).TaskServiceforwardsaiServicestream-end/stream-abort/errortofinalizeWorkspaceTurnFromStreamEnd(:4442–4544): correlated branch matchesrecord.workspaceId && record.turnId(:4472); uncorrelated branch (metadata == null, notagentId === "compact") →interruptWorkspaceTurnFromUncorrelatedStreamEnd→ settlesinterruptedwheneverstreamEndIndex >= promptIndex(:4293–4305). Producers of such uncorrelated ends: bash-monitor wake continuations, child terminal-attention deliveries, heartbeat, peer messages, parent auto-resume.cleanupDisposableWorkspaceTurn→workspaceService.remove(…, true)kills its background processes; persistent child → parent seesinterrupted→task_stop→backgroundProcessManager.stopMonitor(…, "canceled"). This is the observed "monitors died afterwards".settleWorkspaceTurn(params)(:2085), 11 callers, guarded byworkspaceTurnSettlementLocks.withLock(handleId); waiters inpendingWorkspaceTurnWaitersByHandleIdwithsetTimeouttimeouts (:2425–2496).isManualChildWorkspaceInput); otherwise ignores the end. Touches:281–295, :4217–4355+ tests (+286/−31). Codex: "Didn't find any major issues" + clean security onf9baa2fc9.mergeable: MERGEABLE, butTest / UnitandCodex Commentsred, 19 commits behind main.getWorkspaceTurnLiveness/getWorkspaceTurnRuntimeActivity(identity-matches the active stream'smuxMetadataagainst the record) for staleness/capacity. Touches:442–486, :1298, :2573, :3761, :3800–4064(+494/−60).BLOCKED, latest Codex review has open findings,Test / Unitred, 19 behind.2. Design decisions
D1 — Fibers WRAP the AbortController; they do not replace it.
The AI SDK is cancelled only via
AbortSignal; thefor awaitloop, soft-interrupt at step boundaries, retry/fallback re-creation ofstreamResult, and ~30 abort touchpoints (#4032) all key off the signal. Converting the 750-line loop toStream.fromAsyncIterable+ fiber interruption would touch hundreds ofWorkspaceStreamInfotransitions and break theprocessStreamWithCleanup/createStreamResultspy seams. Instead: the fiber is the ownership/supervision unit; the signal stays the cancellation transport. The dual-cancellation glue #4032 feared is confined to one point — the supervisor'sonInterrupt— which routes through the existing user-stop path (cancelStreamSafely), so shutdown ≡ "user pressed stop" semantically (partial flushed with usage,stream-abortemitted,completionsettlesaborted, AIService commits the partial).D2 — Supervisor topology: one supervisor fiber per stream in
AppFiberScope, wrapping the already-startedprocessingPromise.streamInfo.processingPromise = this.processStreamWithCleanup(...)stays byte-identical (sync-start preserved;Reflect.set(processStreamWithCleanup)seam preserved;cleanupAbortedStream'sawait processingPromiseunchanged). Immediately after it:Effect.promiseis interruptible while suspended (internal/effect.js:741–801, Async op);onInterrupt=onErrorFilter(causeFilterInterruptors, …)(:1762);forkInregistersfiberInterrupt(fiber)as the scope finalizer (:2264–2275),fiberInterruptawaits the fiber (:635–642), and parallelscopeCloseawaits all finalizers viafiberAwaitAll(:1590–1601) →closeScopeBoundedat dispose step 2 gives "interrupt and await" whilehistoryService/sessionUsage/eventSink → AIService → bridge serversare still alive (bridges stop in step 3, so clients receivestream-abort).forkIn's observer removes the scope finalizer (verified) → no per-stream residue.forkInon a closed scope callsfiber.interruptUnsafesynchronously and returns the fiber (:2272–2274,runSyncdoes not defect). WithstartImmediately: true,forkUnsaferunschild.evaluatesynchronously (:2233–2247) up to theEffect.promiseAsync op (:772–801), so the fiber is suspended (_running=false) when the interrupt lands andinterruptUnsafe(:391–409) unwinds the stack through theonInterrupthandler → the stream is aborted assystem(fail-closed during shutdown). Verified in rc.112 internals (effect.js:2233–2247, 391–409); pin with a test ("stream started after scope close is aborted") so an RC bump cannot silently change it."system"is semantically exact:"user"/"startup"suppress next-startup recovery (retryEligibility.ts:114–118, 284–287),"system"marks an involuntary backend interruption (astaskService.ts:8100, 8223use it). No in-session retry loop is possible: thestream-aborthandler (agentSession.ts:6010) routes{ type: "aborted" }toretryManager.handleStreamFailure, and"aborted"is inNON_RETRYABLE_STREAM_ERRORS(retryEligibility.ts:49–59, 106) →retryManager.ts:99–104abandons immediately, never schedules a fiber. Dogfooding still checks the restart UX (the recovered partial is shown as interrupted; note whether any next-startup recovery re-sends — same class as today'ssystemaborts fromtaskService).engineScopearrives as an optional 6th constructor parameter (engineScope?: Scope.Closeable), wired fromAppFiberScopeTaginStreamManagerLive(core.ts:226). Defaultundefinedkeeps every direct-construction test andaiService.ts:174path identical (I4).AppFiberScopeLivealready sits beneath S2b inruntimeSeams, so no staging change (I6)."system"— no wire/schema change; UI copy forsystemalready exists.pendingStreamStarts, before registration) is not supervised: nothing is persisted for it yet, andstopStreamalready aborts pending controllers. Documented, not fixed.D3 — Fix the two adjacent cancel races in the same PR (closely-related bugs, not deferrals).
(a)
cleanupAbortedStream: afterawait processingPromise, ifstreamInfo.terminalCompletion !== undefined(completed/failed while the cancel was in flight) → return without abort bookkeeping (preventspartial.jsonresurrection afterdeletePartialand astream-abortafterstream-end). (b)cancelStreamSafely(:1766): latch a per-streamcancelPromiseso concurrent cancellers (user stop racing dispose) join one cleanup → exactly onestream-abort, onesettle. Zero-suspension requirement: the latch must be checked and assigned synchronously at function entry, before anyawait(the current first await isflushPartialWriteat:1789) — otherwise racing callers can both entercleanupAbortedStream. Shape:Both are ≤ 10 LoC and get behavioral tests.
D4 — Shutdown bound stays 2 s; the finalizer must be fast or abandoned.
Outer budgets are 5 s; 2 s + 2 s already consume 4 s. A flowing stream aborts within one chunk; a wedged provider (no chunks, ignores abort) hits the existing
boundedTeardowntimeout: warning, continue, process exit — identical to today's outcome. Dogfooding measures the actual[shutdown] AppFiberScope closed { ms }with a live stream.D5 — Turn handles: codify the settlement invariant; do not fiberize.
Invariant: a workspace-turn handle settles terminally only by (i) a stream terminal event whose
muxMetadatacorrelates{taskHandleId, ownerWorkspaceId, turnId}to the record; (ii) an explicit interrupt (task_stop/interruptWorkspaceTurn); (iii) manual supersession — a manual child input after the turn anchor; (iv) stale-liveness reconciliation. An uncorrelated stream-end is never terminal by itself. #3949 makes (iii) the only uncorrelated outcome; #3915 implements (iv) by identity. Wave 4 adds acausediscriminant tosettleWorkspaceTurn(the single chokepoint) with a runtime assertion, plus the regression harness. Rationale for not converting waiters toDeferred/fibers: no behavioral gain, 4.9k-line file, and the coordinator's "settle only on the owning stream's termination" is over-specified — a turn owns several streams.D6 — Startup:
initialize()stays a Promise facade over a runtime-run startup effect; timeout ⇒ same failure path as a thrown step.Each step is
Effect.tryPromise({ try: async () => step(), catch: identity }).pipe(Effect.timeoutOrElse({ duration: STARTUP_STEP_TIMEOUT_MS, orElse: () => Effect.fail(new StartupStepTimeoutError(name, ms)) }))(timeoutOrElseexists in rc.112,Effect.d.ts:7833; chosen overtimeout+catchTagbecause the step's error channel isunknown, whichcatchTagcannot narrow). NoforkDetachneeded: a Promise step keeps running on its own when the waiting fiber times out (not inside an uninterruptible region, so the timeout interrupts the wait directly).StartupStepTimeoutError extends Errorwithname = "StartupStepTimeoutError"set in the constructor and message"<step> exceeded <ms> ms"(so the desktop dialog's error formatting shows both the class and the step name) → desktop shows it in the existingStartup Faileddialog; CLI/ACP/tests paths unchanged. Step errors keep their identity (v4runPromiserejects with the raw failure). Downgrading any step to best-effort is a policy change, out of scope (audit of the six implementations: extensionMetadata/telemetry/experiments are local fs, <50 ms; policy has its own 10 s fetch timeout; workspaceService bounds its sync internally; onlytaskService.initialize— config scan +editConfig+ recoverysendMessages — is potentially unbounded). The threestart()s stay sync (Effect.sync), the two fire-and-forget sweeps stay outside the effect.stepDurationsMsis preserved.Abandon-and-quit safety: an abandoned
taskService.initializemay be mid-editConfigwhen the root exits. Parity requirement for PR 3: after a rejectedinitialize(), every root runs the boundeddispose()before exiting. Verified: desktop already does —servicesis assigned before the await (main.ts:653–656), the catch callsapp.quit(), and thebefore-quitlistener (:1271–1305, guardif (isDisposing || !services) return) racesservices.dispose()against 5 s; ACP does (serverConnection.ts:205–216);cli/server.tsdoes not (:133–136awaited at top level,main().catchat:282only logs) → PR 3 adds a boundeddispose()there (≤ 10 LoC, same 5 s budget).D7 — streamBridge: thread the runtime context, not a runner.
subscriptionIterablegainscontext?: Context.Context<never>→Stream.toAsyncIterableWith(context);routerSubscriptionspasses the handler's"effect/context". Production behavior identical; heartbeat sleeps on the runtimeClock; tests can run the 3 clock-bound waits onTestClock. Honest scope: the 8 readiness polls stay.3. PRs (ordered by value ÷ risk; each independently mergeable)
PR 1 — StreamManager engine core becomes the first
AppFiberScopeoccupantValue: high (the only remaining shutdown data-integrity gap; the reason
AppFiberScopeexists). Risk: medium → low with D1/D2. Net product LoC ≈ +55 (superviseEngine~25, ctor param/field ~5,engineFiberfield ~2, D3 guards ~12,core.tswiring ~2, doc updates inappRuntime.ts/appFiberScope.ts"occupant" text ~10).Files:
src/node/services/streamManager.ts,src/node/services/di/layers/core.ts,src/node/services/di/appRuntime.ts+appFiberScope.ts(docs), tests below.Pre-work (before writing product code; each yields a note in the PR body):
processStreamWithCleanupnever rejects (try/catch/finally shape:3331–4089); else the supervisor must fold rejections (it alreadycatchDefects).Effect.promiseinterruption +onInterruptawait ordering underScope.closein a 20-line probe test (pattern ofappFiberScope.test.ts:27–47).stream-abortlistener →commitPartial; agentSession completion continuations) and confirm the durable order (writePartial→ commit →deletePartial) makes a mid-flightprocess.exitrecoverable on next load (it is: partial survives until commit completes).[shutdown] AppFiberScope closed { ms }with a live stream in the sandbox (D4).startImmediately: trueinto an already-closed scope still runs itsonInterruptfinalizer (reviewer-verified in rc.112 internals; the test guards RC bumps).Acceptance (behavioral tests only):
streamManager.test.ts(new cases; existing cases untouched): withengineScope = Scope.makeUnsafe("parallel")and a fakecreateStreamResultwhosefullStreamyields onetext-deltathen blocks until itsAbortSignalfires —closeScopeBounded(engineScope)resolves;writePartialwas called with the streamed text; exactly onestream-abort(abortReason: "system") and zerostream-end;completionsettles{status:"aborted"};workspaceStreamsis empty.closeScopeBoundedresolves within the bound, never rejects, warns once (assert the returned promise resolves and no throw; do not assert log text).COMPLETED→ history has exactly one final message,partial.jsonabsent, event orderstream-endonly.stopStream+closeScopeBoundedracing on one stream → exactly onestream-abort, one settle.closeScopeBounded(engineScope)emits zerostream-abortand completes in the same tick class as an empty scope (assert no aborts andworkspaceStreams.size === 0).streamManager.chaos.test.ts— existing cases byte-identical; one new fuzz variant constructs with an engine scope and closes it at a random iteration: every stream settles exactly once (count terminal events permessageId≤ 1, allcompletionpromises settle).serviceContainer.test.ts: "dispose() aborts and awaits an in-flight stream beforedesktopBridgeServer.stop()" (extend the ordering harness at:295–323);coreServicesRoot.test.ts:xum runcleanup list does the same viaappFiberScope.close.Gate suites:
streamManager.test.ts,streamManager.chaos.test.ts,streamManager.modelOnlyNotifications.test.ts,aiService.test.ts,agentSession.disposeRace.test.ts,agentSession.sinceReplayContract.test.ts,serviceContainer.test.ts,coreServicesRoot.test.ts,di/*.test.ts,taskService.test.ts,workspaceService.test.ts,turnRequestBuilder.test.ts;make static-check.House pre-review audits: interruption posture (supervisor's only suspension is the promise; finalizer uninterruptible end-to-end incl.
cancelStreamSafely→cleanupAbortedStream); no defect escapes (catchDefecton the supervisor;Effect.promisethunksasync); spy-seam check (processStreamWithCleanup,createStreamResult,createStreamAtomically,startStreamsignatures unchanged; constructor arity unchanged, trailing optional); sync-start (processingPromiseassigned before fork;runSync(forkIn)completes synchronously); no constructor side-effects added; zero-suspension check on D3(b) latch (cancelPromisechecked-and-assigned synchronously atcancelStreamSafelyentry, before the firstawaitat:1789; review the diff for any insertedawait/lookup ahead of the assignment).Rollback: revert the
core.tswiring line →engineScopeundefined → today's behavior; D3 guards can stay (independent bug fixes).PR 2 — Turn-settlement invariant + false-settle regression harness (gated on #3949)
Value: high (7× production race). Risk: low. Net product LoC ≈ +40 (
WorkspaceTurnSettlementCauseunion +causeonsettleWorkspaceTurnparams + assert ~10; 11 call sites × 1–3 lines).Relationship to open PRs — explicit:
interruptWorkspaceTurnFromUncorrelatedStreamEnd,isWorkspaceTurnAnchorForRecord,isManualChildWorkspaceInput), and adds the invariant + proof on top. If [task-service] 🤖 fix: preserve turns across synthetic wake ends #3949 has not merged when PRs 1/3 are done: do not fork a competing fix; report to the coordinator, offer the regression test file to [task-service] 🤖 fix: preserve turns across synthetic wake ends #3949's author as a review artifact, and hold PR 2 (it is not on any other PR's critical path).:3800–4064incl.settleStaleWorkspaceTurn, asettleWorkspaceTurncaller) overlaps PR 2's one-line-per-caller change. Prefer landing after it; if PR 2 must go first, the conflict is a one-linecause:addition per caller. PR 2 never edits liveness/reservation code.workspaceTurnManager.ts.Design:
type WorkspaceTurnSettlementCauseenumerated from the 11 callers (audited atmain@b87f62729)::1373creation validation failure;:1476pre-stream interrupt during launch;:1500/:1518pre-stream send failure;:3834/:3863stale-liveness recovery / restart timeout (settleStaleWorkspaceTurn— #3915's region);:4301uncorrelated-stream-end manual supersession (the only uncorrelated settle in the codebase; the path #3949 rewrites);:4529correlated terminal;:4571stream-abort;:4675deferred stream error;:4736terminal stream error.settleWorkspaceTurnassertsparams.causeis a member and, formanual-supersession, that the superseding input'smessageIdis supplied — turning D5 into an exhaustiveRecord<Cause, …>check rather than prose, so a future "settle on uncorrelated end" cannot be added without naming (and justifying) a cause.Acceptance:
workspaceTurnManager.uncorrelatedStreamEnd.test.ts(realWorkspaceTurnManager+TaskHandleStore+ fakeaiServiceemitter, following the existing suite's harness): (1) create turn → correlatedstream-start→ synthetic wake stream on the same child ends uncorrelated after the anchor → handle staysrunning, waiter unresolved, no disposable cleanup, no terminal attention → correlatedstream-end→completed. (2) same withfinishReason:"tool-calls"continuation in between. (3) manual child input between anchor and end →interruptedwithcause: manual-supersession. (4) explicitinterruptWorkspaceTurn→interrupted,cause: explicit-interrupt. Case (1) is the scripted reproduction: it must fail on the pre-[task-service] 🤖 fix: preserve turns across synthetic wake ends #3949 merge-base (run the file from a sibling worktree atgit merge-base origin/main <#3949 head>; record the failing assertion in the PR body) and pass after.workspaceTurnManager.test.tscases andtaskService.test.tsturn cases unchanged.Gate suites:
workspaceTurnManager.test.ts,taskService.test.ts,taskHandleStore.test.ts,tools/task*.test.ts;make static-check.Audits: spy-seam (
getWorkspaceTurn,listAllWorkspaceTurns,enqueueTerminalAttention,deliverPersistentChildWorkspaceTurnResultuntouched); settlement lock held across the assert; no new suspension insidewithLock.Rollback: revert; the test file stays valid against #3949 alone (drop the
causeassertions).PR 3 —
ServiceContainer.initialize()as a runtime-run startup effect with per-step timeoutsValue: medium (a hung
taskService.initialize()currently pins the splash screen forever; deterministic TestClock tests of startup). Risk: low–medium. Net product LoC ≈ +80 (step table ~20, timed-step helper ~15,StartupStepTimeoutError~8, constant ~3, facade ~10, root dispose-on-failure parity ≤ 10, doc update ~10).Files:
serviceContainer.ts,src/constants/terminationTimeouts.ts(keep with the termination constants so the budget doc stays in one place),cli/server.ts(dispose in the startup catch if missing),di/appRuntime.tsdoc ("Deliberately not done" → remove the initialize() line; add startup contract).Design (D6):
initialize(): Promise<void>→this.runtime.managed.runPromise(this.startupEffect()).startupEffect = Effect.genover an orderedreadonly steps: ReadonlyArray<{ name, run: () => Promise<void> }>(assert names unique); each step:recordSteptiming kept,Effect.tryPromise({ try: async () => run(), catch: identity }).pipe(Effect.timeoutOrElse({ duration: STARTUP_STEP_TIMEOUT_MS, orElse: () => Effect.fail(new StartupStepTimeoutError(name, STARTUP_STEP_TIMEOUT_MS)) })). ThenEffect.syncfor the threestart()s; the sweeps remain afterrunPromise. ConstantSTARTUP_STEP_TIMEOUT_MS— pre-work measures[startup] <step> { ms }across sandbox cold starts and picks ≥ 10× the slowest observed (propose 60 s; must be generous — a false timeout turns a slow-but-fine start into a crash). Roots dispose after a rejectedinitialize()(D6 abandon-and-quit safety).Acceptance (all in
serviceContainer.test.ts, TestClock via the existingAppLivespy at:355–395):initialize()rejects withStartupStepTimeoutErrornaming the step after exactlyTestClock.adjust(STARTUP_STEP_TIMEOUT_MS); later steps did not run.initialize()rejects with the same error object (identity), later steps did not run (parity with today).stepDurationsMshas all six keys;start()s called once each; secondinitialize()call behavior unchanged from today (verify whether re-entry is guarded today; preserve).tests/ipcharness and ACP entry still pass unchanged.Gate suites:
serviceContainer.test.ts,coreServicesRoot.test.ts,di/*.test.ts,src/node/acp/*.test.ts,TEST_INTEGRATION=1 bun x jest tests/ipc(smoke subset);make static-check.Audits: I1 untouched (no layer body changes); I2 (only the composition root touches the runtime); error identity preserved (no wrapping); abandoned-step safety — every root runs bounded
dispose()after a rejectedinitialize()(D6;cli/server.tsgains it in this PR); no sweep moved into the effect; the six-step order and[startup] <step>names unchanged.Rollback: revert; constant removal.
PR 4 (optional — cut if budget is exhausted) —
streamBridgeon the runtime contextValue: low (closes the last documented "global runtime" exception; enables TestClock for the heartbeat). Risk: low. Net product LoC ≈ +30 (
context?option +toAsyncIterableWith~8; 19 call sites × 1 line via one shared helper inrouterSubscriptions.ts~3).Acceptance: the 3 clock-bound waits (
:207, :241, :255) run onTestClock; the heartbeat test asserts N heartbeats afterTestClock.adjust(N × interval)with zero real time; existing behavioral assertions unchanged;tests/ipcsubscription tests pass. Do not rewrite the 8 readiness polls.Gate suites:
streamBridge.test.ts,routerSubscriptions*.test.ts,orpc/*.test.ts,TEST_INTEGRATION=1 bun x jest tests/ipc(subscription subset);make static-check.Audits:
Stream.toAsyncIterableWithpreserves double-close safety (pin with the existing test);Cause.Donetyping unchanged; noScope/MemoMap/Schedulercaptured (pass the oRPCeffect/context, which the DI layer already strips perEffectRunnerLive);contextstays optional so direct callers/tests without a runtime keep today's global-runtime path.Rollback: revert; the optional
contextdefault (Context.empty()) is exactly today'stoAsyncIterable, so a partial revert of call sites is also safe.Execution order and size
Net product LoC for the wave ≈ +205 (PR 1 ≈ +55, PR 2 ≈ +40, PR 3 ≈ +80, PR 4 ≈ +30); tests ≈ +600–800. PR 1 starts immediately. PR 2 starts the moment #3949 merges (parallel with PR 1/3 — disjoint files). PR 3 after PR 1 merges (both touch
appRuntime.tsdocs; PR 3 also touchesserviceContainer.ts). PR 4 last, only if PRs 1–3 landed and no OFF-RAMP fired. Each PR: Codex dual review,Codex Commentsminimization, merge queue; commit WIP early (/tmpwipes).4. STOP criterion (measurable) and OFF-RAMPs
Wave 4 is done — and the Effect migration line stops without a new RFC — when all hold:
serviceContainer.test.tsordering test +coreServicesRoot.test.tspass on main; a sandboxscript -ftranscript ofxum serverreceiving SIGTERM mid-stream showsstream-abort→[shutdown] AppFiberScope closed { ms }→[shutdown] desktopBridgeServer.stop, and immediately after exitpartial.jsonis absent whilechat.jsonlcontains the interrupted assistant message (baseline onmain:partial.jsonpresent, message absent until next load).{ ms }< 2000 in the flowing-stream case.settleWorkspaceTurnrejects any settlement without an enumerated cause; the coordinator's own Mux sessions show zero "superseded by an uncorrelated workspace stream-end" in the two weeks after PR 2 (soft signal, logged in the wave summary).[startup]per-step lines unchanged in the sandbox transcript; a throwaway build with the constant set to 1 ms showsStartup failed: StartupStepTimeoutError: <step> exceeded 1 msand a clean exit.Test / Unitruns onmainafter the last Wave 4 merge — querygh run list --workflow pr.yml --branch main --limit 60 --json databaseId,status,conclusion,event,headSha(note:gh run list --jsonserializes these fields in lowercase, e.g.{"status":"completed","conclusion":"success"}, unlikestatusCheckRollup), keepstatus === "completed"(pending runs have an emptyconclusion, not null), take the newest 20, and for any run withconclusion !== "success"(case-insensitive normalization acceptable) inspect the failing job's log for the touched suite names (jobtimeout/cancelledfrom the 15-min budget is not a flake); plus green merge-queue runs for each PR. Any attributable flake → fix or revert before declaring done.OFF-RAMP (PR 1): fires if pre-work 1–3 shows (a) routing shutdown through
cancelStreamSafelycannot preserve crash-recovery semantics without changingcleanupAbortedStream's contract beyond D3, (b) the chaos variant exposes a double-settle not closable by D3(b), or (c) the finalizer cannot fit the 2 s bound for flowing streams. Then: stop PR 1, keepAppFiberScopeunoccupied, updateappRuntime.ts"Deliberately not done" with the concrete blocker and the measured evidence, land D3 alone as a bug-fix PR. PRs 2–4 are independent and proceed.OFF-RAMP (PR 3): if error identity or the
tests/ipc/ACP paths cannot be preserved, keepinitialize()as is and record why.OFF-RAMP (PR 2): #3949 not merged → hold (see PR 2).
5. Risk register
commitPartial'shistorySequenceupdate-or-append is idempotent (historyService.ts:2036–2041)errorchunk → error path instead of abort pathboundedTeardownalready bounds; transcript measures; no budget change possible (5 s outer)stream-abortlistener (commitPartial) still in flight whenprocess.exitrunspartial.jsonis already gone whencli/server.tslogs its final cleanup line beforeprocess.exit(0)(:252–266)createStreamResult,tokenTracker)cause:conflicts onlyforkIn/onInterrupt/toAsyncIterableWith)streamManager.ts/streamBridge.ts; pins fixed; GA upgrade is a separate lockstep PR (§6)Reflect.set(processStreamWithCleanup)runSyncsynchronousshutdown()(desktop secondbefore-quitlistener) still does not await streamsshutdown()never touches the runtime; desktop's dispose race is the covered pathforkInon closed scope interrupts immediately →systemabort (startImmediatelysemantics verified in rc.112; pinned by test)systemabort triggers an in-session RetryManager retry during shutdown"aborted"∈NON_RETRYABLE_STREAM_ERRORS→retryManager.ts:99–104abandons; no fiber scheduledtaskService.initializemid-editConfigwhen the root exits after a timeoutcli/server.tsdispose; config writes are lock/journal-protectedstartImmediately/onInterruptsemantics differ in a later RCappFiberScope.test.tsstyle; RC bumps are a separate lockstep PR6. Standing item — effect v4 GA +
@orpc/experimental-effectlockstep (analysis only)v4 is not GA (rc.112 is current; v3
3.xremains the stable line). No PR this wave. When GA ships: one lockstep PR bumpingeffect+ all@orpc/*(1.14.11today; check the GA-compatible@orpc/experimental-effect), canary gates =di/*.test.ts,streamBridge.test.ts,streamManager.test.ts,serviceContainer.test.ts,TEST_INTEGRATION=1 bun x jest tests/ipc,make static-check. TheContext → ServiceMaprename risk is firewalled:Context.Servicetags,Context.omit/get,Layer,ManagedRuntime,TestClocklive only underdi/+orpc/effectContext.ts;streamManager.ts/streamBridge.tsuseEffect/Scope/Fiber/Exit/Stream/Queue/Causeonly. PR 4 adds oneContext.Context<never>type reference tostreamBridge.ts— keep it as a type-only import so a rename is a one-line fix.7. Dogfooding (per PR; evidence attached to the PR with
gh … --attach)Common setup:
make dev-server-sandbox DEV_SERVER_SANDBOX_ARGS="--clean-projects"(orxum serveron a tempXUM_ROOT) withXUM_LOG_LEVEL=debug, run underscript -f ~/wave4-scratch/<pr>-<scenario>.log; scratch under$HOME/wave4-scratch/(never/tmp). Drive the UI withagent-browser(open→snapshot -i→ click the explicit "Send message" ref; re-snapshot after typing). Screenshots are primary evidence; record WebM and finalize withffmpeg -c copy.kill -TERM <server pid>; transcript must show the order in STOP Better authentication UX #1 and[shutdown] AppFiberScope closed { ms }; (2)ls <XUM_ROOT>/sessions/<ws>/partial.json(absent) +tail -n 1 chat.jsonl(interrupted assistant message); (3) restart, open the workspace in agent-browser, screenshot the persisted interrupted message; (4) same scenario onmainfor the baseline diff; (5)xum runCtrl-C mid-stream transcript (CLI root parity); (6) quality gate between phases: gate suites green before the sandbox run, sandbox evidence before requesting review.taskkind=workspace to a child that arms a background bash monitor firing within ~10 s and keeps working ~60 s; screenshot the parent's task result (baselinemain:interrupted … uncorrelated workspace stream-end; after:completed) and the child's log lines.xum servercold start transcript with[startup] <step> { ms }for the six steps (parity); throwaway worktree build withSTARTUP_STEP_TIMEOUT_MS = 1→ transcript ofStartup failed: StartupStepTimeoutError …and exit code (do not ship); desktop dialog cannot be shown headless — cite the unchangeddesktop/main.ts:1249–1265catch.streamBridge.test.tson TestClock.8. Non-goals (restated; out of this wave)
Typed-error propagation sweep / removing the ~113 facades; converting services to
yield*-based Effect services; PubSub for the internal EventEmitter bus; Schema at persistence boundaries; Effect observability; converting sync read paths, AI-SDK per-request callbacks, cross-process lock interiors, or deterministic try-lock funnels; replacingAbortControlleras the SDK cancellation transport; converting thefullStreamloop toStream; fiberizing turn-handle waiters; downgrading startup steps to best-effort; changing outer quit budgets; supervising the pre-registration stream-start window;shutdown()semantics; the effect GA bump (standing analysis only).Generated with
xum• Model:anthropic:claude-fable-5-1• Thinking:xhigh• Cost:$34.90