throw error;

}

let runFollowupTurn = queuedRunFollowupTurn;

let shouldDrainFollowupsAfterReplyOperationClears = false;

const returnAfterReplyOperationClearsThenDrainFollowups = <T>(value: T): T => {

shouldDrainFollowupsAfterReplyOperationClears = true;

let shouldDrainQueuedFollowupsAfterClear = false;

const returnWithQueuedFollowupDrain = <T>(value: T): T => {

shouldDrainQueuedFollowupsAfterClear = true;

return value;

};

const drainQueuedFollowupsAfterClear = () => {

scheduleFollowupDrain(queueKey, runFollowupTurn);

};

const prePreflightCompactionCount = activeSessionEntry?.compactionCount ?? 0;

let preflightCompactionApplied = false;

if (!replyOperation.result) {

replyOperation.fail("run_failed", new Error("reply operation exited with final payload"));

}

return returnAfterReplyOperationClearsThenDrainFollowups(runOutcome.payload);

return returnWithQueuedFollowupDrain(runOutcome.payload);

}

const {

// Otherwise, a late typing trigger (e.g. from a tool callback) can outlive the run and

// keep the typing indicator stuck.

if (payloadArray.length === 0) {

return returnAfterReplyOperationClearsThenDrainFollowups(undefined);

return returnWithQueuedFollowupDrain(undefined);

}

const currentMessageId = sessionCtx.MessageSidFull ?? sessionCtx.MessageSid;

didLogHeartbeatStrip = payloadResult.didLogHeartbeatStrip;

if (replyPayloads.length === 0) {

return returnAfterReplyOperationClearsThenDrainFollowups(undefined);

return returnWithQueuedFollowupDrain(undefined);

}

const successfulCronAdds = runResult.successfulCronAdds ?? 0;

}

}

const result = returnAfterReplyOperationClearsThenDrainFollowups(

const result = returnWithQueuedFollowupDrain(

finalPayloads.length === 1 ? finalPayloads[0] : finalPayloads,

);

replyOperation.result?.kind === "aborted" &&

replyOperation.result.code === "aborted_for_restart"

) {

return returnAfterReplyOperationClearsThenDrainFollowups({

return returnWithQueuedFollowupDrain({

text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",

});

}

if (replyOperation.result?.kind === "aborted") {

return returnAfterReplyOperationClearsThenDrainFollowups({ text: SILENT_REPLY_TOKEN });

return returnWithQueuedFollowupDrain({ text: SILENT_REPLY_TOKEN });

}

if (error instanceof GatewayDrainingError) {

replyOperation.fail("gateway_draining", error);

return returnAfterReplyOperationClearsThenDrainFollowups({

return returnWithQueuedFollowupDrain({

text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",

});

}

if (error instanceof CommandLaneClearedError) {

replyOperation.fail("command_lane_cleared", error);

return returnAfterReplyOperationClearsThenDrainFollowups({

return returnWithQueuedFollowupDrain({

text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",

});

}

replyOperation.fail("run_failed", error);

// Keep the followup queue moving even when an unexpected exception escapes

// the run path; the caller still receives the original error.

returnAfterReplyOperationClearsThenDrainFollowups(undefined);

returnWithQueuedFollowupDrain(undefined);

throw error;

} finally {

replyOperation.complete();

if (shouldDrainFollowupsAfterReplyOperationClears) {

// Same-session follow-up turns create their own ReplyOperation; start them

// only after this run clears the active-run guard.

scheduleFollowupDrain(queueKey, runFollowupTurn);

if (shouldDrainQueuedFollowupsAfterClear) {

replyOperation.completeThen(drainQueuedFollowupsAfterClear);

} else {

replyOperation.complete();

}

blockReplyPipeline?.stop();

typing.markRunComplete();