const BLOCK_REPLY_SEND_TIMEOUT_MS = 15_000;

function scheduleFollowupDrainAfterReplyOperationClear(params: {

operation: ReplyOperation;

queueKey: string;

runFollowup: (run: FollowupRun) => Promise<void>;

}): void {

runAfterReplyOperationClear(params.operation, (admissionSessionId) => {

const completedSessionId = params.operation.sessionId;

const runFollowupAfterClear =

admissionSessionId === completedSessionId

? params.runFollowup

: (queued: FollowupRun) =>

params.runFollowup(

queued.run.sessionId === completedSessionId

? { ...queued, admissionSessionId }

: queued,

);

scheduleFollowupDrain(params.queueKey, runFollowupAfterClear);

});

}

function markBeforeAgentRunBlockedPayloads(payloads: ReplyPayload[]): ReplyPayload[] {

return payloads.map((payload) =>

setReplyPayloadMetadata(payload, { beforeAgentRunBlocked: true }),

typing.cleanup();

return undefined;

}

// Re-check liveness after enqueue so a stale active snapshot cannot leave

// the followup queue idle if the original run already finished.

const queuedBehindActiveRun = isRunActive?.() === true;

if (!queuedBehindActiveRun) {

// The queue must stay dormant while the active owner can still collect

// messages. Registering after enqueue closes the owner-clear race.

const activeReplyOperation = replyRunRegistry.get(queueKey);

if (activeReplyOperation) {

scheduleFollowupDrainAfterReplyOperationClear({

operation: activeReplyOperation,

queueKey,

runFollowup: queuedRunFollowupTurn,

});

} else {

scheduleFollowupDrain(queueKey, queuedRunFollowupTurn);

}

const queuedBehindActiveRun = isRunActive?.() === true;

await touchActiveSessionEntry();

if (queuedBehindActiveRun) {

await typingSignals.signalToolStart();

shouldDrainQueuedFollowupsAfterClear = true;

return value;

};

const drainQueuedFollowupsAfterClear = (admissionSessionId: string) => {

const completedSessionId = replyOperation.sessionId;

const runFollowupAfterClear =

admissionSessionId === completedSessionId

? runFollowupTurn

: (queued: FollowupRun) =>

runFollowupTurn(

queued.run.sessionId === completedSessionId

? { ...queued, admissionSessionId }

: queued,

);

scheduleFollowupDrain(queueKey, runFollowupAfterClear);

};

const restartRecoveryDeliveryRunId = crypto.randomUUID();

let trackedRestartRecoveryDeliveryContext = false;

const persistRestartRecoveryDeliveryContext = async (): Promise<void> => {

);

}

if (shouldDrainQueuedFollowupsAfterClear) {

if (providedReplyOperation) {

runAfterReplyOperationClear(replyOperation, drainQueuedFollowupsAfterClear);

} else {

replyOperation.completeThen(() => drainQueuedFollowupsAfterClear(replyOperation.sessionId));

scheduleFollowupDrainAfterReplyOperationClear({

operation: replyOperation,

queueKey,

runFollowup: runFollowupTurn,

});

if (!providedReplyOperation) {

replyOperation.complete();

}

} else if (!providedReplyOperation) {

replyOperation.complete();