refactor(dev): finish the coordinator's Effect orchestration; fan out epoch-store reads with Effect.forEach - #513
Conversation
…out epoch-store reads with Effect.forEach
Startup races each blocking step against a Deferred close signal with
Effect.raceFirst instead of Promise.race, and a failed startup releases
the watcher and lock through Effect.forEach + Effect.exit before
re-raising the original error. Shutdown captures the in-flight build's
Exit and every resource release's Exit the same way and aggregates them
into the unchanged DevCoordinatorCloseError. The build pass is an
Effect.fnUntraced generator that lifts only its leaf I/O and exposes each
phase with Effect.result.
EpochStore.recoverStaging and #readAllEpochMetadata fan out with
Effect.forEach({ concurrency: 'unbounded' }) instead of Promise.all inside
lifted thunks; the durable-fs protocol (lstat containment, inode identity,
dir fsync, atomic publication) is untouched.
ScriptedAlchemy
added
the
skip-changeset
|
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. |
commit: |
1 participant
Summary
Effect-conformance cleanup 2 of the prior audit (
dev/coordinator.ts+dev/epoch-store.ts): finish the Effect orchestration around theSemaphore/Deferredthat Stage 3 already introduced, so startup, shutdown, and the build pass are one Effect program each instead of Promise combinators wrapped around fibers. Behavior-preserving: same rejection values ('DevCoordinator is closed.', the original startup error,DevCoordinatorCloseErrorwith the samefailuresorder), sameAB7200/AB7201diagnostics, same events.What changed
packages/agent-bundle/src/dev/coordinator.ts#awaitStartup(wasPromise.race([operation, cancellation.then(throw)]), audit rowcoordinator.ts:370–376) is nowEffect.raceFirst(liftPromise(operation), Deferred.await(this.#startupClosed)). The startup-cancellationPromise+ resolver pair became#startupClosed: Deferred<never, Error>, failed by#cancelStartup()with the sameError('DevCoordinator is closed.'). The losing step is interrupted; the underlying Promise (e.g. a blocked lock acquisition) keeps running exactly as before, and#acquireStartupLockstill drains a lock that resolves after cancellation.#start(wasasync+try/catch+Promise.allSettled([releaseWatcher, releaseLock]), audit row:343–346) is#startEffect(): anEffect.genwhose sync steps (#assertOpen,#createWatcher) are lifted withliftTryso a throw is a typed failure — not a defect that would skip cleanup — and whose failure handler runsEffect.forEach([releaseWatcher, releaseLock], (r) => Effect.exit(liftPromise(r)), { concurrency: 'unbounded' })before re-raising the original error.start()runs it through the boundary'srunPromiseonce.#close(was twoPromise.allSettled, audit row:561–570) waits for the in-flight build/startup withEffect.exit(liftPromise(...)), then releases every resource withEffect.forEach(resources, ({ close }) => Effect.exit(liftPromise(close)), { concurrency: 'unbounded' }), and aggregatesExitfailures viaCause.squashinto the sameDevCoordinatorCloseError(build first, then resources in declaration order) — the same settle-then-aggregate shape asMcpSessionServiceCloseError/EpochCleanupError.#performBuild(was anasyncfunction lifted whole byliftPromise, audit row:401–405) is nowEffect.fnUntraced(function* (this: DevCoordinator, invalidation) { … }). Only the leaf I/O is lifted —projectService.prepare, theonPreparedProjecthook,diagnosticService.lint,artifactService.build,packageBuildService.build— and each phase is exposed withEffect.resultsoResult.isFailurecompletes the attempt as a failed build result, exactly where thetry/catchblocks did. Status/event bookkeeping stays synchronous inside the fiber;#startBuildnow composesthis.#performBuild(invalidation)directly underwithPermit+onExit.packages/agent-bundle/src/dev/epoch-store.tsrecoverStaging(audit row:480): thePromise.allinside the lifted thunk isEffect.forEach(entries, rm, { concurrency: 'unbounded', discard: true }); thereaddir+ENOENTtolerance moved into a small#readDirectoryEntries(path)Effect (liftPromise+Effect.catch).#readAllEpochMetadata(audit row:1004) is an Effect method:#readDirectoryEntries+Effect.forEach(..., { concurrency: 'unbounded' })overliftPromise(() => this.#readEpochMetadata(id)), with the path-safety check asEffect.fail(EpochStoreError('EPOCH_METADATA_INVALID', …)).listEpochsand#cleanupUnderLeasecompose it directly instead of lifting it.:774and:848. Those twoPromise.all([realpath, realpath])pairs live inside#verifyStaging(which also checksdev/inoidentity) and#validateActiveEpoch(lstat containment). Both are imperative leaf verification helpers of the durable-fs protocol that the brief keeps raw ("Do NOT touch the durable-fs protocol (lstat, dir fsync, inode)"), and the Stage 3 convention keeps leaf helpers imperative and identity-lifted as a unit (docs/effect-conventions.md§ Stage 3). Rewriting them as Effect generators would lift everylstat/realpathindividually for no orchestration gain. If the maintainer wants them converted anyway,Effect.all([a, b], { concurrency: 'unbounded' })is the fixed-tuple form (agent-patterns/effect-concurrency.md"Bounded work").#syncPath,#writeJsonAtomically,lstat, dir fsync, inode checks) are untouched.Idioms and citations
repos/effect/LLMS.md/ repo sectionEffect.raceFirst— first completion wins, loser interrupted (instead ofPromise.racearound Effects)agent-patterns/effect-concurrency.md"Racing and cancellation: Do notPromise.racearound Effects"Deferredas the one-shot close signalagent-patterns/effect-concurrency.md"Bounded work";docs/effect-conventions.mdStage 3 helped (Deferred+runSyncadmission)Effect.forEach(..., { concurrency: 'unbounded' })+ per-elementEffect.exitas thePromise.allSettledanaloguedocs/effect-conventions.mdStage 3 helped: "Effect.forEach(..., { concurrency: 'unbounded' })with per-elementEffect.exitis the exact analogue ofPromise.allSettled"Effect.fnUntracedfor a reusable library-internal generator (withthis: Self)Effect.fnandEffect.fnUntraced;docs/effect-conventions.md§ Generator styleEffect.result+Result.isFailureto expose a phase failure as a valueagent-patterns/effect-errors.md"Defect (bug):Effect.die— not for expected fail-closed states"Tests
tests/dev-coordinator.test.ts: re-raises a startup failure after releasing the watcher and lock it acquired (rejectedwatcher.ready, same error re-raised, both releases run once, laterstart()returns the settled startup,rebuildfails closed withAB7200,close()does not re-release); fails a synchronous watcher construction error closed and releases the lock (theliftTrypath); turns a rejected prepared-project hook into a failed prepare attempt (AB7201"Prepare failed during development rebuild: hook rejection", no artifact build,build.started/build.failed/artifact.statusevents).tests/epoch-store.test.ts: recovers every abandoned staging directory at once and tolerates a store that never published (three staging roots removed concurrently, an unrelated directory untouched,recoverStaging/listEpochson a store with noepochsdirectory).cancels blocked startup readiness…,…blocked startup recovery…,…blocked active epoch recovery…) still assert'DevCoordinator is closed.', one watcher/lock close each, and no later watcher/build state.pnpm typecheck,pnpm lintgreen;pnpm test:unitgreen (3155 passed); integrationdev-artifact-service+dev-package-build+dev-workbenchgreen (45 passed).Changeset
skip-changeset: internal orchestration refactor; every Promise-side value, diagnostic code, message, and event is unchanged, and nothing new is exported.Review status
No PR comments are posted by the author; every review thread is answered in this section.
6becd76Verify (Node 24)attempt failed only inpackages/workbench/tests/mcp-app-real.e2e.test.ts(PlaywrightwaitForRequesttimeout on the "Close MCP session" click — a Workbench e2e untouched by this PR; the same job's unit/route-unit/projection pools passed); the failed job was re-run and is green.