From ffbdca0f025caad833a728d1e235b643f2940815 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 19:11:52 +0000 Subject: [PATCH] chore: deslop pass over the wave 3.5 delta Rebased survivors of the stranded deslop/wave-3.5 commit (5a8723423) onto current main. Applied: - finalizers: sqlite connection close and Flight reader cancel no longer mask the original failure when teardown itself throws - state drivers: shared pending-open tracker replaces the verbatim trackPendingOpen/close-drain duplication; drop runPromise(Effect.fail) ceremony in favor of direct rejections - boundaries: remove the unused runSyncExit export from both seams, the dead ScopedEffectRuntime E type parameter from the rsc-runtime copy (matching the dev-seam copy), and the redundant string ternary in toDevError/toRuntimeError - delete dead epoch-lease-registry.ts (zero importers; #161 rewrote the same concept in epoch-store.ts); dedupe boundRenderEventStream through emitBoundRenderEvent; trim migration-narration comments Dropped as superseded: the reconciler progress-queue rework (#172 rebuilt that path with a demand-bounded design), the dev-seam trim of interruptWhenAborted/runPromiseExit (#164 fixed and kept them with tests), the sqlite #commit self-rewrite (#171 rewrote #commit), and the lint-plugin inlining (#164 expanded the plugin around those helpers). --- .changeset/effect-wave-cleanup.md | 11 ++++ docs/effect-conventions.md | 2 +- packages/agent-bundle/src/dev/coordinator.ts | 2 +- .../src/dev/epoch-lease-registry.ts | 62 ------------------- packages/agent-bundle/src/dev/epoch-store.ts | 7 +-- .../src/dev/mcp-session/mcp-session.ts | 7 +-- packages/agent-bundle/src/effect/boundary.ts | 5 +- packages/rsc-runtime/src/effect/boundary.ts | 9 +-- .../rsc-runtime/src/effect/render-stream.ts | 26 +++----- packages/rsc-runtime/src/reconciler.ts | 4 +- .../rsc-runtime/src/state/memory-driver.ts | 27 ++------ .../rsc-runtime/src/state/pending-opens.ts | 31 ++++++++++ packages/rsc-runtime/src/state/sqlite.ts | 38 +++++------- 13 files changed, 87 insertions(+), 144 deletions(-) create mode 100644 .changeset/effect-wave-cleanup.md delete mode 100644 packages/agent-bundle/src/dev/epoch-lease-registry.ts create mode 100644 packages/rsc-runtime/src/state/pending-opens.ts diff --git a/.changeset/effect-wave-cleanup.md b/.changeset/effect-wave-cleanup.md new file mode 100644 index 000000000..bed7a2da9 --- /dev/null +++ b/.changeset/effect-wave-cleanup.md @@ -0,0 +1,11 @@ +--- +"@agent-bundle/runtime": patch +"agent-bundle": patch +--- + +Cleanup pass over the Wave 3.5 Effect migration: harden the sqlite +connection and Flight reader finalizers against masking the original +failure, consolidate the state drivers' duplicated pending-open lifecycle +tracking, remove the dead epoch lease registry and unused `runSyncExit` +boundary exports, and trim migration-narration comments. No public API or +behavior change. diff --git a/docs/effect-conventions.md b/docs/effect-conventions.md index 2af53825e..4c2313652 100644 --- a/docs/effect-conventions.md +++ b/docs/effect-conventions.md @@ -26,7 +26,7 @@ Each Effect-consuming package has exactly one `src/effect/boundary.ts`: The boundary owns: -- `runPromise` / `runPromiseExit` / `runSync` / `runSyncExit` +- `runPromise` / `runSync` (plus `runPromiseExit` where a caller branches on `Exit`) - `AbortSignal` ↔ interruption (`interruptWhenAborted`, `scopedAbortSignal`, `signal` on `runPromise`) - mapping the Effect error channel onto the existing typed Error contracts diff --git a/packages/agent-bundle/src/dev/coordinator.ts b/packages/agent-bundle/src/dev/coordinator.ts index 088bae1a1..eda1b9ef7 100644 --- a/packages/agent-bundle/src/dev/coordinator.ts +++ b/packages/agent-bundle/src/dev/coordinator.ts @@ -399,7 +399,7 @@ export class DevCoordinator { /** * Runs one build pass as an Effect fiber holding the build permit; the * exit hook drains the coalesced follow-up slot before the caller's - * promise settles, exactly where the pre-Effect `finally` chain sat. + * promise settles. */ #startBuild(invalidation: Invalidation): Promise { const current = runPromise(this.#buildPermit.withPermit( diff --git a/packages/agent-bundle/src/dev/epoch-lease-registry.ts b/packages/agent-bundle/src/dev/epoch-lease-registry.ts deleted file mode 100644 index 35c30b93a..000000000 --- a/packages/agent-bundle/src/dev/epoch-lease-registry.ts +++ /dev/null @@ -1,62 +0,0 @@ -import { serialQueue } from '../core/async.ts'; - -interface LeaseQueueEntry { - pending: number; - readonly queue: ReturnType; -} - -/** - * Process-wide lease registry for epoch stores. - * - * Cross-instance sharing is intentional: every EpochStore over the same - * project must serialize lease transitions through one queue and observe one - * set of reference counts, so leases survive across store instances over one - * path (a test asserts this). Lease queues are keyed by the store's resolved - * `.agent-bundle` path and reference counts by the absolute epoch directory. - * - * Entries clean themselves up only where that is provably safe: a - * reference-count entry is deleted when its count returns to zero, and a - * lease-queue entry is deleted once its last pending transition settles — a - * later transition recreates the queue, and nothing can interleave because an - * entry only leaves the map while no transition is queued against it. - */ -export class EpochLeaseRegistry { - readonly #queues = new Map(); - readonly #references = new Map(); - - /** Current lease count for one absolute epoch directory path. */ - referenceCount(epochPath: string): number { - return this.#references.get(epochPath) ?? 0; - } - - /** Releases one lease; the entry is removed when the count reaches zero. */ - release(epochPath: string): void { - const count = this.#references.get(epochPath) ?? 0; - if (count <= 1) { - this.#references.delete(epochPath); - return; - } - this.#references.set(epochPath, count - 1); - } - - retain(epochPath: string): void { - this.#references.set(epochPath, (this.#references.get(epochPath) ?? 0) + 1); - } - - async runLeaseTransition(agentBundlePath: string, operation: () => Promise): Promise { - const entry = this.#queues.get(agentBundlePath) ?? { pending: 0, queue: serialQueue() }; - this.#queues.set(agentBundlePath, entry); - entry.pending += 1; - try { - return await entry.queue.run(operation); - } finally { - entry.pending -= 1; - if (entry.pending === 0 && this.#queues.get(agentBundlePath) === entry) { - this.#queues.delete(agentBundlePath); - } - } - } -} - -/** The one process-wide lease registry every EpochStore instance shares. */ -export const sharedEpochLeaseRegistry = new EpochLeaseRegistry(); diff --git a/packages/agent-bundle/src/dev/epoch-store.ts b/packages/agent-bundle/src/dev/epoch-store.ts index 048a2386f..484b81378 100644 --- a/packages/agent-bundle/src/dev/epoch-store.ts +++ b/packages/agent-bundle/src/dev/epoch-store.ts @@ -361,7 +361,7 @@ export class EpochStore { /** The process-wide lease mutex shared by every store over this project. */ readonly #leaseTransitions: Semaphore.Semaphore; readonly #staging = new Map(); - /** Serializes this store's state transitions, replacing the pre-Effect serial queue. */ + /** Serializes this store's state transitions. */ readonly #transitions: Semaphore.Semaphore = runSync(Semaphore.make(1)); constructor(options: EpochStoreOptions) { @@ -611,9 +611,8 @@ export class EpochStore { return yield* this.#publishVerifiedStaging(record, beforeActivate); }); // The staging root is removed whether the publish committed or failed; - // a removal failure replaces the outcome, exactly as the pre-Effect - // `finally` did (so it is deliberately not a scope finalizer, which - // would have to swallow it). + // a removal failure replaces the outcome, so it is deliberately not a + // scope finalizer (which would have to swallow it). return Effect.gen({ self: this }, function* (this: EpochStore) { const outcome = yield* Effect.exit(attempt); yield* liftPromise(() => rm(record.root, { force: true, recursive: true })); diff --git a/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts b/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts index e1083c408..53dcac693 100644 --- a/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts +++ b/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts @@ -153,7 +153,7 @@ export class McpSession { #closed = false; #connection: McpSessionConnectionState | undefined; #droppedThroughSequence = 0; - /** Serializes initialize / restart / close, replacing the pre-Effect serial queue. */ + /** Serializes initialize / restart / close. */ readonly #lifecycle: Semaphore.Semaphore = runSync(Semaphore.make(1)); #sequence = 0; #stderrOutput = ''; @@ -414,8 +414,7 @@ export class McpSession { * Session teardown as one Effect. Every release step always runs, in * order — drain the client, remove plugin data, release the epoch lease, * notify the owner — and the last failing step's error is re-raised once - * every resource has been visited (the pre-Effect nested `finally` chain - * had the same last-failure-wins contract). + * every resource has been visited (last-failure-wins). */ #closeEffect(): Effect.Effect { return Effect.suspend(() => { @@ -526,7 +525,7 @@ export class McpSession { }).pipe( // A failed connect drains the replacement client and stops its // stderr capture before the failure re-raises; a cleanup failure - // replaces the original error, exactly as the pre-Effect catch did. + // replaces the original error. Effect.catch((error) => liftPromise(async () => { try { await client.close(); diff --git a/packages/agent-bundle/src/effect/boundary.ts b/packages/agent-bundle/src/effect/boundary.ts index 99bfb8b38..1f6392739 100644 --- a/packages/agent-bundle/src/effect/boundary.ts +++ b/packages/agent-bundle/src/effect/boundary.ts @@ -62,7 +62,7 @@ export const isTypedDevError = (error: unknown): error is Error => export const toDevError = (value: unknown): Error => { if (isAbortError(value) || isTypedDevError(value)) return value; if (value instanceof Error) return value; - return new Error(typeof value === 'string' ? value : String(value)); + return new Error(String(value)); }; export const mapCause = (cause: Cause.Cause): Error => { @@ -100,9 +100,6 @@ export const runPromiseExit = async ( export const runSync = (effect: Effect.Effect): A => throwExitFailure(Effect.runSyncExit(effect)); -export const runSyncExit = (effect: Effect.Effect): Exit.Exit => - Effect.runSyncExit(effect); - /** * Internal long-lived Effect runtime. Its Layer owns one Scope, which stays * live across Promise API calls and is finalized exactly once by close(). diff --git a/packages/rsc-runtime/src/effect/boundary.ts b/packages/rsc-runtime/src/effect/boundary.ts index 68a250f7e..f3c0ddb48 100644 --- a/packages/rsc-runtime/src/effect/boundary.ts +++ b/packages/rsc-runtime/src/effect/boundary.ts @@ -62,7 +62,7 @@ export const isTypedRuntimeError = (error: unknown): error is Error => export const toRuntimeError = (value: unknown): Error => { if (isAbortError(value) || isTypedRuntimeError(value)) return value; if (value instanceof Error) return value; - return new Error(typeof value === 'string' ? value : String(value)); + return new Error(String(value)); }; export const mapCause = (cause: Cause.Cause): Error => { @@ -100,21 +100,18 @@ export const runPromiseExit = async ( export const runSync = (effect: Effect.Effect): A => throwExitFailure(Effect.runSyncExit(effect)); -export const runSyncExit = (effect: Effect.Effect): Exit.Exit => - Effect.runSyncExit(effect); - /** * Internal long-lived Effect runtime. Its Layer owns one Scope, which stays * live across Promise API calls and is finalized exactly once by close(). */ -export interface ScopedEffectRuntime { +export interface ScopedEffectRuntime { close(): Promise; run(effect: Effect.Effect, options?: RunPromiseOptions): Promise; } export const makeScopedEffectRuntime = ( layer: Layer.Layer, -): ScopedEffectRuntime => { +): ScopedEffectRuntime => { const runtime = ManagedRuntime.make(layer); let closing: Promise | undefined; return Object.freeze({ diff --git a/packages/rsc-runtime/src/effect/render-stream.ts b/packages/rsc-runtime/src/effect/render-stream.ts index ba012415c..2ef98e4c9 100644 --- a/packages/rsc-runtime/src/effect/render-stream.ts +++ b/packages/rsc-runtime/src/effect/render-stream.ts @@ -37,6 +37,15 @@ export const createFlightDemand = (): FlightDemand => { }; }; +export const emitBoundRenderEvent = ( + sequence: ReturnType, + input: AgentRenderEventInput, +): Effect.Effect => + Effect.try({ + catch: (error) => toRuntimeError(error), + try: () => sequence.emit(input), + }); + /** * Contract bounds as a stream stage: sequence numbers, elapsed / rate / * count / event-bytes, document snapshot bounds (depth / nodes / bytes), @@ -50,20 +59,5 @@ export const boundRenderEventStream = ( stream: Stream.Stream, ) => Stream.Stream => { const sequence = createAgentRenderEventSequence(limits); - return (stream: Stream.Stream) => - Stream.mapEffect(stream, (input) => - Effect.try({ - catch: (error) => toRuntimeError(error), - try: () => sequence.emit(input), - }), - ); + return (stream) => Stream.mapEffect(stream, (input) => emitBoundRenderEvent(sequence, input)); }; - -export const emitBoundRenderEvent = ( - sequence: ReturnType, - input: AgentRenderEventInput, -): Effect.Effect => - Effect.try({ - catch: (error) => toRuntimeError(error), - try: () => sequence.emit(input), - }); diff --git a/packages/rsc-runtime/src/reconciler.ts b/packages/rsc-runtime/src/reconciler.ts index 0d331a4b1..416aebe7a 100644 --- a/packages/rsc-runtime/src/reconciler.ts +++ b/packages/rsc-runtime/src/reconciler.ts @@ -417,7 +417,9 @@ const gatedFlightStream = ( Effect.gen(function*() { const reader = yield* Effect.acquireRelease( Effect.sync(() => flight.getReader()), - (handle) => Effect.promise(() => handle.cancel().then(() => undefined)), + // cancel() rejects when the source already errored; a defect inside + // this closing scope must not replace the stream's own failure. + (handle) => Effect.promise(() => handle.cancel().then(() => undefined, () => undefined)), ); return Stream.unfold(undefined, () => demand.wait.pipe( diff --git a/packages/rsc-runtime/src/state/memory-driver.ts b/packages/rsc-runtime/src/state/memory-driver.ts index ef11aa6e9..9b626d0eb 100644 --- a/packages/rsc-runtime/src/state/memory-driver.ts +++ b/packages/rsc-runtime/src/state/memory-driver.ts @@ -31,6 +31,7 @@ import { runStateMigrations, } from './journal.js'; import { stateEffect } from './effect.js'; +import { createPendingOpenTracker } from './pending-opens.js'; /** * In-memory state driver (#98). @@ -135,7 +136,7 @@ const createMemoryStore = ( ); const runStore = (effect: Effect.Effect): Promise => internals.closed - ? runPromise(Effect.fail(new AgentStateError('store-closed', `State '${internals.definition.id}' store is closed`))) + ? Promise.reject(new AgentStateError('store-closed', `State '${internals.definition.id}' store is closed`)) : runtime.run(effect); /** @@ -340,22 +341,10 @@ export const createMemoryStateDriver = (options: MemoryStateDriverOptions = {}): // retrieval site below, keyed by the definition id they were created for. const registry = new Map>(); const openStores = new Set>(); - const pendingOpens = new Set>(); + const pendingOpens = createPendingOpenTracker(); let closed = false; let closing: Promise | undefined; - const trackPendingOpen = (operation: Promise): Promise => { - const settled = operation.then( - () => undefined, - () => undefined, - ); - pendingOpens.add(settled); - void settled.then(() => { - pendingOpens.delete(settled); - }); - return operation; - }; - return Object.freeze({ durable: false, kind: 'memory', @@ -365,9 +354,7 @@ export const createMemoryStateDriver = (options: MemoryStateDriverOptions = {}): if (closing !== undefined) return closing; closed = true; closing = (async () => { - while (pendingOpens.size > 0) { - await Promise.all([...pendingOpens]); - } + await pendingOpens.settle(); for (const entry of [...openStores]) { await entry.store.close(); } @@ -380,7 +367,7 @@ export const createMemoryStateDriver = (options: MemoryStateDriverOptions = {}): open( definition: AgentStateDefinition, ): Promise> { - return trackPendingOpen( + return pendingOpens.track( (async () => { const entry = await runPromise( stateEffect(() => { @@ -427,9 +414,7 @@ export const createMemoryStateDriver = (options: MemoryStateDriverOptions = {}): await entry.activate(); if (closed) { await entry.store.close(); - return runPromise( - Effect.fail(new AgentStateError('store-closed', `State '${definition.id}' cannot open on a closed driver`)), - ); + throw new AgentStateError('store-closed', `State '${definition.id}' cannot open on a closed driver`); } return entry.store; })(), diff --git a/packages/rsc-runtime/src/state/pending-opens.ts b/packages/rsc-runtime/src/state/pending-opens.ts new file mode 100644 index 000000000..cc6ef0edb --- /dev/null +++ b/packages/rsc-runtime/src/state/pending-opens.ts @@ -0,0 +1,31 @@ +/** + * Tracks in-flight driver `open()` promises so `close()` can wait for every + * pending open — including ones that start while it waits — to settle before + * tearing down the stores they may have created. + */ +export interface PendingOpenTracker { + readonly track: (operation: Promise) => Promise; + readonly settle: () => Promise; +} + +export const createPendingOpenTracker = (): PendingOpenTracker => { + const pending = new Set>(); + return Object.freeze({ + track(operation: Promise): Promise { + const settled = operation.then( + () => undefined, + () => undefined, + ); + pending.add(settled); + void settled.then(() => { + pending.delete(settled); + }); + return operation; + }, + async settle(): Promise { + while (pending.size > 0) { + await Promise.all([...pending]); + } + }, + }); +}; diff --git a/packages/rsc-runtime/src/state/sqlite.ts b/packages/rsc-runtime/src/state/sqlite.ts index ebad92a2e..1b465b487 100644 --- a/packages/rsc-runtime/src/state/sqlite.ts +++ b/packages/rsc-runtime/src/state/sqlite.ts @@ -50,6 +50,7 @@ import { resolveResetState, runStateMigrations, } from './index.js'; +import { createPendingOpenTracker } from './pending-opens.js'; /** * Workspace-durable state driver on `node:sqlite` (#98, G3). @@ -239,14 +240,14 @@ class SqliteStore implements Age #definition: AgentStateDefinition; readonly #now: () => Date; readonly #onClose: () => void; - readonly #runtime: ScopedEffectRuntime; + readonly #runtime: ScopedEffectRuntime; constructor( definition: AgentStateDefinition, file: string, now: () => Date, onClose: () => void, - runtime: ScopedEffectRuntime, + runtime: ScopedEffectRuntime, ) { this.#definition = definition; this.location = file; @@ -511,7 +512,7 @@ class SqliteStore implements Age #run(effect: Effect.Effect): Promise { return this.#closed - ? runPromise(Effect.fail(new AgentStateError('store-closed', `State '${this.#definition.id}' store is closed`))) + ? Promise.reject(new AgentStateError('store-closed', `State '${this.#definition.id}' store is closed`)) : this.#runtime.run(effect); } @@ -724,22 +725,10 @@ export const createSqliteStateDriver = (options: SqliteStateDriverOptions): Agen } const now = options.now ?? ((): Date => new Date()); const openStores = new Set>(); - const pendingOpens = new Set>(); + const pendingOpens = createPendingOpenTracker(); let closed = false; let closing: Promise | undefined; - const trackPendingOpen = (operation: Promise): Promise => { - const settled = operation.then( - () => undefined, - () => undefined, - ); - pendingOpens.add(settled); - void settled.then(() => { - pendingOpens.delete(settled); - }); - return operation; - }; - return Object.freeze({ durable: true, kind: 'sqlite', @@ -749,9 +738,7 @@ export const createSqliteStateDriver = (options: SqliteStateDriverOptions): Agen if (closing !== undefined) return closing; closed = true; closing = (async () => { - while (pendingOpens.size > 0) { - await Promise.all([...pendingOpens]); - } + await pendingOpens.settle(); for (const store of [...openStores]) { await store.close(); } @@ -763,7 +750,7 @@ export const createSqliteStateDriver = (options: SqliteStateDriverOptions): Agen open( definition: AgentStateDefinition, ): Promise> { - return trackPendingOpen( + return pendingOpens.track( (async () => { const file = await runPromise( sqliteEffect(definition.id, 'resolve storage', () => { @@ -788,7 +775,12 @@ export const createSqliteStateDriver = (options: SqliteStateDriverOptions): Agen }, true), (db) => Effect.sync(() => { - db.close(); + try { + db.close(); + } catch { + // Closing an already-broken connection must not mask the + // caller's path (the original failure carries the cause). + } }), ); const runtime = makeScopedEffectRuntime( @@ -807,9 +799,7 @@ export const createSqliteStateDriver = (options: SqliteStateDriverOptions): Agen } if (closed) { await store.close(); - return runPromise( - Effect.fail(new AgentStateError('store-closed', `State '${definition.id}' cannot open on a closed driver`)), - ); + throw new AgentStateError('store-closed', `State '${definition.id}' cannot open on a closed driver`); } openStores.add(store as unknown as SqliteStore); return store;