Skip to content

Commit 290c90e

Browse files
committed
fix(tui): reject a session-queue op that waits on itself
1 parent 0d780ee commit 290c90e

2 files changed

Lines changed: 488 additions & 9 deletions

File tree

‎src/tui/delivery-queue.ts‎

Lines changed: 127 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -8,11 +8,16 @@
88
* submit kind, not a QueueKind — tests/API only.
99
*/
1010

11+
import { AsyncLocalStorage } from "node:async_hooks";
12+
1113
import { AgentClosedError } from "@intx/agent";
12-
import type { PendingImageAttachment } from "./image-attachments.js";
13-
import type { ProductHostDeliver } from "./product-host.js";
14+
import { getLogger } from "@intx/log";
15+
16+
import { LOG_NAMESPACE_ROOT } from "../branding.js";
1417
import { ASK_DIRECTOR_WAKE_PREFIX } from "../subagent/fleet-report.js";
1518
import { MAILBOX_MAIL_WAKE_PREFIX } from "../subagent/mailbox-mail-drive.js";
19+
import type { PendingImageAttachment } from "./image-attachments.js";
20+
import type { ProductHostDeliver } from "./product-host.js";
1621

1722
export type AgentDeliveryNotDeliveredReason =
1823
| "agent-closed"
@@ -421,6 +426,24 @@ export function drainSteersOnly(state: SessionQueueState): {
421426
return { state: current, drained };
422427
}
423428

429+
const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]);
430+
431+
/**
432+
* Id of the op whose body is on this continuation. A detached promise
433+
* inherits the spawning op's id, so the store stays set after that op
434+
* settles. That is not a self-wait: the deadlock is awaiting the FIFO
435+
* from inside the op that is running right now, not merely calling
436+
* enqueue while some later op is running.
437+
*/
438+
const runningSessionOp = new AsyncLocalStorage<number>();
439+
440+
export class SessionOperationSelfWaitError extends Error {
441+
constructor() {
442+
super("session operation waited on work it queued behind itself");
443+
this.name = "SessionOperationSelfWaitError";
444+
}
445+
}
446+
424447
// Serial promise chain for session-scoped operations; failures do not
425448
// block the tail.
426449

@@ -433,16 +456,113 @@ export interface SessionOperationQueue {
433456

434457
export function createSessionOperationQueue(): SessionOperationQueue {
435458
let tail: Promise<void> = Promise.resolve();
459+
// Id of the op that has not settled, or 0. Cleared in `finally` only for
460+
// that id, so a later op's run cannot be mistaken for the one that queued
461+
// the await.
462+
let nextId = 0;
463+
let activeId = 0;
464+
465+
// True only when this continuation is the op that is running right now.
466+
// A detached continuation still carries its parent's id; that must not
467+
// match a later op's activeId.
468+
const inThisOp = (): boolean => runningSessionOp.getStore() === activeId;
469+
470+
const execute = (op: () => Promise<void>): Promise<void> => {
471+
const id = ++nextId;
472+
activeId = id;
473+
let result: Promise<void>;
474+
try {
475+
result = runningSessionOp.run(id, op);
476+
} catch (err) {
477+
if (activeId === id) activeId = 0;
478+
throw err;
479+
}
480+
return Promise.resolve(result).finally(() => {
481+
if (activeId === id) activeId = 0;
482+
});
483+
};
436484

437485
const enqueue = (op: () => Promise<void>): Promise<void> => {
438-
tail = tail.then(op, op);
439-
return tail;
486+
// `tail.then(op, op)` — a failed op must not wedge the chain. The
487+
// returned promise is not that tail: a self-wait rejection has to
488+
// stay off it, or the op queued behind this one never runs.
489+
const scheduled = tail.then(
490+
() => execute(op),
491+
() => execute(op),
492+
);
493+
tail = scheduled;
494+
if (!inThisOp()) return scheduled;
495+
const enqueuedDuring = activeId;
496+
return observeUnlessSelfWait(scheduled, () => activeId === enqueuedDuring);
440497
};
441498

442-
return {
443-
enqueue,
444-
awaitTail: () => tail.catch(() => undefined),
499+
const awaitTail = (): Promise<void> => {
500+
const drained = tail.catch(() => undefined);
501+
// Outside an op this is the real tail, even if another op is in
502+
// flight. Inside, the tail includes the running op, so awaiting it
503+
// deadlocks the same way as awaiting enqueue.
504+
if (!inThisOp()) return drained;
505+
const awaitedDuring = activeId;
506+
return observeUnlessSelfWait(drained, () => activeId === awaitedDuring);
507+
};
508+
509+
return { enqueue, awaitTail };
510+
}
511+
512+
/**
513+
* Await observes via `.then`. Reject that observation while the enqueuing
514+
* op is still the running op; if it is observed only after that op has
515+
* settled, forward to the scheduled promise. `void enqueue()` never calls
516+
* `.then`, so a fire-and-forget enqueue stays a normal queued op.
517+
*/
518+
function observeUnlessSelfWait(
519+
forward: Promise<void>,
520+
stillRunning: () => boolean,
521+
): Promise<void> {
522+
let reported: SessionOperationSelfWaitError | undefined;
523+
let guarded = false;
524+
const reject = (): Promise<undefined> => {
525+
if (reported === undefined) {
526+
reported = new SessionOperationSelfWaitError();
527+
tuiLogger.warn(reported.message);
528+
}
529+
return Promise.reject(reported);
530+
};
531+
// The self-wait thenable never forwards to `forward`, so a later
532+
// rejection of that promise (the tail, when nothing was queued after
533+
// it) would be unhandled. Attach a handler only on that path. A real
534+
// waiter that observes `forward` after the op settles still hears the
535+
// rejection; this handler does not replace theirs.
536+
const guardUnobserved = (): void => {
537+
if (guarded) return;
538+
guarded = true;
539+
forward.then(
540+
() => undefined,
541+
() => undefined,
542+
);
543+
};
544+
// Not a native Promise: Bun's await does not call an overridden
545+
// `Promise.prototype.then`, so the self-wait check has to live on the
546+
// thenable `await` actually invokes.
547+
const thenable: {
548+
then(
549+
onFulfilled?:
550+
| ((value: undefined) => undefined | PromiseLike<undefined>)
551+
| null,
552+
onRejected?:
553+
| ((reason: unknown) => undefined | PromiseLike<undefined>)
554+
| null,
555+
): Promise<undefined>;
556+
} = {
557+
then(onFulfilled, onRejected) {
558+
if (stillRunning()) {
559+
guardUnobserved();
560+
return reject().then(onFulfilled, onRejected);
561+
}
562+
return forward.then(() => undefined).then(onFulfilled, onRejected);
563+
},
445564
};
565+
return thenable as Promise<void>;
446566
}
447567

448568
/**

0 commit comments

Comments
 (0)