fix(tui): stop a finished session op from rejecting the next one

Author: TheGreatAxios

src/tui/delivery-queue.ts

@@ -461,24 +461,46 @@ export function createSessionOperationQueue(): SessionOperationQueue {
   // the await.
   let nextId = 0;
   let activeId = 0;
+  // Body promise of the op in `activeId`. Unset during that op's first
+  // synchronous turn, before `run` returns.
+  let activeResult: Promise<void> | undefined;
 
   // True only when this continuation is the op that is running right now.
   // A detached continuation still carries its parent's id; that must not
   // match a later op's activeId.
   const inThisOp = (): boolean => runningSessionOp.getStore() === activeId;
 
+  // The body promise can be fulfilled while `activeId` is still set: `finally`
+  // clears the id on a later microtask. A continuation in that gap is not the
+  // op waiting on itself.
+  const bodyStillRunning = (): boolean => {
+    if (activeResult === undefined) return true;
+    try {
+      return Bun.peek(activeResult) === activeResult;
+    } catch {
+      return false;
+    }
+  };
+
   const execute = (op: () => Promise<void>): Promise<void> => {
     const id = ++nextId;
     activeId = id;
     let result: Promise<void>;
     try {
       result = runningSessionOp.run(id, op);
     } catch (err) {
-      if (activeId === id) activeId = 0;
+      if (activeId === id) {
+        activeId = 0;
+        activeResult = undefined;
+      }
       throw err;
     }
+    activeResult = result;
     return Promise.resolve(result).finally(() => {
-      if (activeId === id) activeId = 0;
+      if (activeId === id) {
+        activeId = 0;
+        activeResult = undefined;
+      }
     });
   };
 
@@ -493,7 +515,10 @@ export function createSessionOperationQueue(): SessionOperationQueue {
     tail = scheduled;
     if (!inThisOp()) return scheduled;
     const enqueuedDuring = activeId;
-    return observeUnlessSelfWait(scheduled, () => activeId === enqueuedDuring);
+    return observeUnlessSelfWait(
+      scheduled,
+      () => inThisOp() && activeId === enqueuedDuring && bodyStillRunning(),
+    );
   };
 
   const awaitTail = (): Promise<void> => {
@@ -503,7 +528,10 @@ export function createSessionOperationQueue(): SessionOperationQueue {
     // deadlocks the same way as awaiting enqueue.
     if (!inThisOp()) return drained;
     const awaitedDuring = activeId;
-    return observeUnlessSelfWait(drained, () => activeId === awaitedDuring);
+    return observeUnlessSelfWait(
+      drained,
+      () => inThisOp() && activeId === awaitedDuring && bodyStillRunning(),
+    );
   };
 
   return { enqueue, awaitTail };

src/tui/session-operation-queue.test.ts

@@ -437,3 +437,87 @@ test("a self-wait does not leave the queued op's rejection unhandled", async ()
     process.off("unhandledRejection", onRejection);
   }
 });
+
+test("a microtask detached from a finished op body is not a self-wait", async () => {
+  const log: string[] = [];
+  const { enqueue, awaitTail } = createSessionOperationQueue();
+  let detached: Promise<void> = Promise.resolve();
+
+  const outer = enqueue(async () => {
+    detached = (async () => {
+      await Promise.resolve();
+      await enqueue(async () => {
+        log.push("inner");
+      });
+      log.push("detached:ok");
+    })();
+    log.push("outer:end");
+  });
+
+  await outer;
+  await detached;
+  await awaitTail();
+  expect(log).toEqual(["outer:end", "inner", "detached:ok"]);
+});
+
+test("queueMicrotask after the op body returns is not a self-wait", async () => {
+  const log: string[] = [];
+  const { enqueue, awaitTail } = createSessionOperationQueue();
+  let done: Promise<void> = Promise.resolve();
+
+  const outer = enqueue(async () => {
+    done = new Promise((resolve, reject) => {
+      queueMicrotask(() => {
+        void enqueue(async () => {
+          log.push("inner");
+        }).then(resolve, reject);
+      });
+    });
+    log.push("outer:end");
+  });
+
+  await outer;
+  await done;
+  await awaitTail();
+  expect(log).toEqual(["outer:end", "inner"]);
+});
+
+test("an outside waiter is not rejected while the enqueuing op is blocked", async () => {
+  const log: string[] = [];
+  const { enqueue, awaitTail } = createSessionOperationQueue();
+  let release: () => void = () => undefined;
+  const gate = new Promise<void>((resolve) => {
+    release = resolve;
+  });
+  let inner: Promise<void> | undefined;
+
+  const outer = enqueue(async () => {
+    inner = enqueue(async () => {
+      log.push("inner");
+    });
+    await gate;
+    log.push("outer:end");
+  });
+
+  await Promise.resolve();
+  await Promise.resolve();
+  if (inner === undefined) throw new Error("inner missing");
+  let observed = "pending";
+  const waiting = inner.then(
+    () => {
+      observed = "resolved";
+    },
+    (err: unknown) => {
+      observed =
+        err instanceof SessionOperationSelfWaitError ? "self-wait" : "error";
+    },
+  );
+  await Promise.resolve();
+  expect(observed).toBe("pending");
+  release();
+  await outer;
+  await waiting;
+  await awaitTail();
+  expect(observed).toBe("resolved");
+  expect(log).toEqual(["outer:end", "inner"]);
+});