chore: deslop pass over the wave 3.5 delta

.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.

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
 

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<ArtifactEpochResult> {
     const current = runPromise(this.#buildPermit.withPermit(

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<symbol, StagingRecord>();
-  /** 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 }));

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<void, unknown> {
     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();

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 = <E>(cause: Cause.Cause<E>): Error => {
@@ -100,9 +100,6 @@ export const runPromiseExit = async <A, E>(
 export const runSync = <A, E>(effect: Effect.Effect<A, E>): A =>
   throwExitFailure(Effect.runSyncExit(effect));
 
-export const runSyncExit = <A, E>(effect: Effect.Effect<A, E>): Exit.Exit<A, E> =>
-  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().

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 = <E>(cause: Cause.Cause<E>): Error => {
@@ -100,21 +100,18 @@ export const runPromiseExit = async <A, E>(
 export const runSync = <A, E>(effect: Effect.Effect<A, E>): A =>
   throwExitFailure(Effect.runSyncExit(effect));
 
-export const runSyncExit = <A, E>(effect: Effect.Effect<A, E>): Exit.Exit<A, E> =>
-  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<R, E> {
+export interface ScopedEffectRuntime<R> {
   close(): Promise<void>;
   run<A, E2>(effect: Effect.Effect<A, E2, R>, options?: RunPromiseOptions): Promise<A>;
 }
 
 export const makeScopedEffectRuntime = <R, E>(
   layer: Layer.Layer<R, E>,
-): ScopedEffectRuntime<R, E> => {
+): ScopedEffectRuntime<R> => {
   const runtime = ManagedRuntime.make(layer);
   let closing: Promise<void> | undefined;
   return Object.freeze({

packages/rsc-runtime/src/effect/render-stream.ts

@@ -37,6 +37,15 @@ export const createFlightDemand = (): FlightDemand => {
   };
 };
 
+export const emitBoundRenderEvent = (
+  sequence: ReturnType<typeof createAgentRenderEventSequence>,
+  input: AgentRenderEventInput,
+): Effect.Effect<AgentRenderEvent, Error> =>
+  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<AgentRenderEventInput, E, R>,
 ) => Stream.Stream<AgentRenderEvent, E | Error, R> => {
   const sequence = createAgentRenderEventSequence(limits);
-  return <E, R>(stream: Stream.Stream<AgentRenderEventInput, E, R>) =>
-    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<typeof createAgentRenderEventSequence>,
-  input: AgentRenderEventInput,
-): Effect.Effect<AgentRenderEvent, Error> =>
-  Effect.try({
-    catch: (error) => toRuntimeError(error),
-    try: () => sequence.emit(input),
-  });

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(

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 = <TState, TEvents extends AgentStateEventSchemas>(
   );
   const runStore = <A>(effect: Effect.Effect<A, AgentStateError>): Promise<A> =>
     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<string, MemoryStoreEntry<unknown, AgentStateEventSchemas>>();
   const openStores = new Set<MemoryStoreEntry<unknown, AgentStateEventSchemas>>();
-  const pendingOpens = new Set<Promise<void>>();
+  const pendingOpens = createPendingOpenTracker();
   let closed = false;
   let closing: Promise<void> | undefined;
 
-  const trackPendingOpen = <T>(operation: Promise<T>): Promise<T> => {
-    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<TState, TEvents extends AgentStateEventSchemas>(
       definition: AgentStateDefinition<TState, TEvents>,
     ): Promise<AgentStateStore<TState, TEvents>> {
-      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;
         })(),

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: <T>(operation: Promise<T>) => Promise<T>;
+  readonly settle: () => Promise<void>;
+}
+
+export const createPendingOpenTracker = (): PendingOpenTracker => {
+  const pending = new Set<Promise<void>>();
+  return Object.freeze({
+    track<T>(operation: Promise<T>): Promise<T> {
+      const settled = operation.then(
+        () => undefined,
+        () => undefined,
+      );
+      pending.add(settled);
+      void settled.then(() => {
+        pending.delete(settled);
+      });
+      return operation;
+    },
+    async settle(): Promise<void> {
+      while (pending.size > 0) {
+        await Promise.all([...pending]);
+      }
+    },
+  });
+};