fix(host): drain the dispatcher queue to empty when a step throws

src/extension/session/effect-executor.ts

@@ -672,8 +672,9 @@ export function createEffectExecutor<TEdit>(deps: EffectExecutorDeps<TEdit>): Ef
             // a settlement UNVERIFIED — so the ack for an unverified landing is
             // the effect most likely to re-run a broken read. Unwinding
             // `runEffects` here would (a) escape the FULFILMENT arm as an
-            // unhandled rejection, since `createDrainingDispatcher` has
-            // `try/finally` and NO `catch`, and (b) abandon the rest of the effect
+            // unhandled rejection, since `createDrainingDispatcher` catches a
+            // throwing `step` only to finish its drain and RETHROWS afterwards
+            // (its failure policy), and (b) abandon the rest of the effect
             // list. It would NOT skip the barrier release — the panel's `step`
             // settles unconditionally — but the ack Document is exactly the effect
             // worth keeping, so contain the throw here rather than relying on that
@@ -767,9 +768,10 @@ export function createEffectExecutor<TEdit>(deps: EffectExecutorDeps<TEdit>): Ef
           // `execute-write.ts`'s `settle()` became total: the correlated case —
           // a failure tag whose settle read ALSO threw — used to land in the
           // rejection arm's `try/catch`, and now resolves through the UNWRAPPED
-          // fulfilment arm. `createDrainingDispatcher` has `try`/`finally` and no
-          // `catch`, so a SYNCHRONOUS `window.showErrorMessage` throw would both
-          // escape as an unhandled rejection and abandon the rest of this effect
+          // fulfilment arm. `createDrainingDispatcher` rethrows a throwing `step`
+          // once its drain is empty, so a SYNCHRONOUS `window.showErrorMessage`
+          // throw would both escape as an unhandled rejection (only later, at the
+          // end of that drain) and abandon the rest of this effect
           // list — including the ack `postDocument`. The barrier release survives
           // either way (the panel's `step` settles unconditionally), so this guard
           // is about the effect list, not the barrier. No latch: this is

src/extension/session/host-session-core.ts

@@ -1314,22 +1314,79 @@ export function createHostSessionCore(context: HostSessionContext, deps: HostSes
 /** Queue-draining, non-recursive event dispatcher (Codex R2). `step(event)`
  *  runs one transition + its effects; an effect that synchronously
  *  re-dispatches enqueues behind the active loop and is drained AFTER
- *  `step` returns — flat, FIFO, never a recursive stack. */
+ *  `step` returns — flat, FIFO, never a recursive stack.
+ *
+ *  FAILURE POLICY — a throwing `step` does NOT cancel the rest of the drain.
+ *  Scheduling is this primitive's ONLY job, so its contract is "every accepted
+ *  event gets exactly one `step` ATTEMPT" — an attempt, not a success: this is
+ *  scheduling, not transactional recovery. A sibling event's failure is not a
+ *  reason to break it. The two rejected alternatives:
+ *    - ABANDON the queue (today's shape: reset `draining`, keep the entries).
+ *      The residue is not lost, it is DEFERRED — the next external dispatch
+ *      drains it first, arbitrarily later, against a state it was never
+ *      computed for. A stale replay is the worst of the three outcomes.
+ *    - CLEAR the queue. Dropping accepted events is silent state loss, and for
+ *      the host session it is unsafe by construction: `applyEditSettled` is
+ *      the write lock's ONLY release site while the panel is alive (this
+ *      file's own `disposed` case also clears `pendingApplyBaseVersion`, but
+ *      only on teardown — see `isWriteLockHeld` above and `effect-executor.ts`'s
+ *      header comment), so dropping an `applyEditSettled` strands the lock and
+ *      the side channels deferred behind it for the rest of a still-alive
+ *      session.
+ *  Continuing leaves a WELL-FORMED state from either throw site inside `step`:
+ *  a throwing TRANSITION leaves the committed state untouched, and a throwing
+ *  EFFECT runs after the transition has already committed. See
+ *  `host-session-step.ts`. What continuing does NOT do — and must not be read
+ *  as doing — is REPAIR the failed event: the rest of that event's effect list
+ *  stays abandoned, and a throw from an `applyEditSettled` TRANSITION still
+ *  strands the write lock (that cost is owned and tracked by
+ *  `HostSessionStepDeps.commitTransition`, and no queue policy can pay it).
+ *  Draining on is simply the least-bad of the three, not a rescue.
+ *
+ *  ⚠️ LIVENESS is unchanged and still the caller's to keep: a `step` that
+ *  re-dispatches on every pass never empties the queue and this loop never
+ *  returns — now also on the failure path, where the accumulated errors are
+ *  never rethrown either. A bounded drain would trade that for the silent
+ *  event loss this policy exists to avoid, so the bound stays where it always
+ *  was: no effect may re-dispatch unconditionally.
+ *
+ *  Errors are neither swallowed nor allowed to displace each other: the drain
+ *  finishes first, then a lone failure is rethrown AS-IS (callers keep the
+ *  error identity and its triage payload) and several are rethrown together as
+ *  an `AggregateError`. The throw still escapes to the caller, so the
+ *  unhandled-rejection reasoning in `effect-executor.ts` is unchanged — only
+ *  its timing moves to the end of the drain. */
 export function createDrainingDispatcher<Ev>(step: (event: Ev) => void): (event: Ev) => void {
   const queue: Ev[] = [];
   let draining = false;
   return (event: Ev): void => {
     queue.push(event);
     if (draining) {
+      // Already inside a drain — including a drain that is currently unwinding
+      // a `step` throw, since the loop below catches per step and keeps going.
       return;
     }
     draining = true;
+    const errors: unknown[] = [];
     try {
       while (queue.length > 0) {
-        step(queue.shift() as Ev);
+        try {
+          step(queue.shift() as Ev);
+        } catch (err) {
+          errors.push(err);
+        }
       }
     } finally {
+      // Released BEFORE the rethrow below, so a caller that dispatches from its
+      // own catch handler starts a fresh drain rather than silently enqueueing
+      // behind a loop that has already exited.
       draining = false;
     }
+    if (errors.length === 1) {
+      throw errors[0];
+    }
+    if (errors.length > 1) {
+      throw new AggregateError(errors, "[quoll] host session drain: multiple steps threw");
+    }
   };
 }

test/extension/session/host-session-core.test.ts

@@ -1077,6 +1077,98 @@ describe("createDrainingDispatcher", () => {
     dispatch("two");
     expect(seen).toEqual(["one", "two"]);
   });
+
+  // FAILURE POLICY on a throwing `step` (see the dispatcher's own comment). The
+  // queue is drained to EMPTY before the error leaves the dispatcher, so an
+  // event a doomed step already enqueued can never be replayed later against a
+  // diverged state. Before this policy the drain abandoned the queue and `seen`
+  // stopped at ["a"].
+  it("drains the queue to empty when a step throws, then rethrows that error", () => {
+    const seen: string[] = [];
+    const boom = new Error("step threw");
+    let dispatch!: (e: string) => void;
+    dispatch = createDrainingDispatcher<string>((event) => {
+      seen.push(event);
+      if (event === "a") {
+        dispatch("b"); // enqueued behind the active drain...
+        throw boom; // ...and abandoned by the throw, before this policy
+      }
+    });
+    let thrown: unknown = "NOTHING THROWN";
+    try {
+      dispatch("a");
+    } catch (err) {
+      thrown = err;
+    }
+    // Identity, not just shape: a single failure must reach the caller as the
+    // very error the step threw, so existing handlers keep their triage payload.
+    expect(thrown).toBe(boom);
+    expect(seen).toEqual(["a", "b"]);
+    // ...and NO residue survives into the next dispatch (the released `draining`
+    // guard starts a fresh drain that sees only its own event).
+    dispatch("c");
+    expect(seen).toEqual(["a", "b", "c"]);
+  });
+
+  it("aggregates when more than one step throws in the same drain", () => {
+    const first = new Error("first");
+    const second = new Error("second");
+    let dispatch!: (e: string) => void;
+    dispatch = createDrainingDispatcher<string>((event) => {
+      if (event === "a") {
+        dispatch("b");
+        throw first;
+      }
+      throw second;
+    });
+    let thrown: unknown = "NOTHING THROWN";
+    try {
+      dispatch("a");
+    } catch (err) {
+      thrown = err;
+    }
+    // Every failure survives: swallowing the later ones would hide a fault that
+    // only the completed drain can produce.
+    expect(thrown).toBeInstanceOf(AggregateError);
+    expect((thrown as AggregateError).errors).toEqual([first, second]);
+    expect((thrown as AggregateError).message).toBe(
+      "[quoll] host session drain: multiple steps threw"
+    );
+  });
+
+  // The rethrow counts ENTRIES, not truthiness, so a step that throws a falsy
+  // value still reaches the caller as that value rather than as "no failure".
+  it("rethrows a falsy thrown value instead of treating the drain as clean", () => {
+    const dispatch = createDrainingDispatcher<string>(() => {
+      // A non-Error throw is the ASSERTION here, not sloppiness: it is exactly
+      // the value a truthiness-based rethrow would swallow.
+      // biome-ignore lint/style/useThrowOnlyError: the non-Error throw is the fixture
+      throw undefined;
+    });
+    let caught = false;
+    let thrown: unknown = "NOTHING THROWN";
+    try {
+      dispatch("a");
+    } catch (err) {
+      caught = true;
+      thrown = err;
+    }
+    expect(caught).toBe(true);
+    expect(thrown).toBeUndefined();
+  });
+
+  it("throws nothing when every queued step succeeds", () => {
+    const seen: string[] = [];
+    let dispatch!: (e: string) => void;
+    dispatch = createDrainingDispatcher<string>((event) => {
+      seen.push(event);
+      if (event === "a") {
+        dispatch("b");
+      }
+    });
+    expect(() => dispatch("a")).not.toThrow();
+    expect(seen).toEqual(["a", "b"]);
+  });
 });
 
 describe("host-session-core: stale-version resync", () => {

test/extension/session/host-session-step.test.ts

@@ -303,12 +303,14 @@ describe("createHostSessionStep", () => {
     expect(reported).toEqual([settleErr]);
   });
 
-  // Also pins today's dispatcher behaviour on a throwing step: a re-entrant
-  // dispatch issued before the throw stays QUEUED and is only drained by the
-  // next external dispatch. That residue is a separate, pre-existing gap
-  // (tracked as its own TODO); these assertions are its measured baseline, so
-  // changing that policy must consciously update them.
-  it("settles through the real dispatcher and pins today's queue residue on a throw", () => {
+  // Also pins the dispatcher's FAILURE POLICY end-to-end (the policy itself is
+  // unit-tested in host-session-core.test.ts): a re-entrant dispatch issued
+  // before the throw is drained inside the SAME dispatch, so it can never be
+  // replayed later against a diverged state — and every drained step still
+  // settles the barrier. Until PR #405 the drain abandoned that event and the
+  // next external dispatch drained it first; these assertions were the measured
+  // baseline of that residue.
+  it("settles through the real dispatcher and drains the throw's residue in the same dispatch", () => {
     const settles: boolean[] = [];
     const seen: string[] = [];
     let dispatch!: (event: HostSessionEvent) => void;
@@ -335,14 +337,12 @@ describe("createHostSessionStep", () => {
       })
     );
     expect(() => dispatch(settled({ kind: "refused" }))).toThrow();
-    expect(settles).toEqual([false]); // settled despite the throw
-    expect(seen).toEqual(["applyEditSettled"]); // the re-entrant event is still queued
+    // The throw reaches the caller only AFTER the queue is empty: the settle for
+    // the failed apply, then the re-entrant event's own step and settle.
+    expect(seen).toEqual(["applyEditSettled", "themeChanged:light"]);
+    expect(settles).toEqual([false, true]); // settled despite the throw
     dispatch(themeChanged); // the drain guard was released by the dispatcher's finally
-    expect(seen).toEqual([
-      "applyEditSettled",
-      "themeChanged:light", // the stale re-entrant event, drained FIRST
-      "themeChanged:dark",
-    ]);
+    expect(seen).toEqual(["applyEditSettled", "themeChanged:light", "themeChanged:dark"]);
     expect(settles).toEqual([false, true, true]);
   });