@@ -46,7 +46,6 @@ import { runAgentTurnWithFallback } from "./agent-runner-execution.js";
4646import {
4747createShouldEmitToolOutput,
4848createShouldEmitToolResult,
49-finalizeWithFollowup,
5049isAudioPayload,
5150signalTypingIfNeeded,
5251} from "./agent-runner-helpers.js";
@@ -71,6 +70,7 @@ import {
7170enqueueFollowupRun,
7271refreshQueuedFollowupSession,
7372resolvePiSteeringModeForQueueMode,
73+scheduleFollowupDrain,
7474type FollowupRun,
7575type QueueSettings,
7676} from "./queue.js";
@@ -1064,7 +1064,7 @@ export async function runReplyAgent(params: {
10641064// the followup queue idle if the original run already finished.
10651065const queuedBehindActiveRun = isRunActive?.() === true;
10661066if (!queuedBehindActiveRun) {
1067-finalizeWithFollowup(undefined, queueKey, queuedRunFollowupTurn);
1067+scheduleFollowupDrain(queueKey, queuedRunFollowupTurn);
10681068}
10691069await touchActiveSessionEntry();
10701070if (queuedBehindActiveRun) {
@@ -1148,6 +1148,11 @@ export async function runReplyAgent(params: {
11481148throw error;
11491149}
11501150let runFollowupTurn = queuedRunFollowupTurn;
1151+let shouldDrainFollowupsAfterReplyOperationClears = false;
1152+const returnAfterReplyOperationClearsThenDrainFollowups = <T>(value: T): T => {
1153+shouldDrainFollowupsAfterReplyOperationClears = true;
1154+return value;
1155+};
11511156const prePreflightCompactionCount = activeSessionEntry?.compactionCount ?? 0;
11521157let preflightCompactionApplied = false;
11531158@@ -1283,7 +1288,7 @@ export async function runReplyAgent(params: {
12831288if (!replyOperation.result) {
12841289replyOperation.fail("run_failed", new Error("reply operation exited with final payload"));
12851290}
1286-return finalizeWithFollowup(runOutcome.payload, queueKey, runFollowupTurn);
1291+return returnAfterReplyOperationClearsThenDrainFollowups(runOutcome.payload);
12871292}
1288129312891294const {
@@ -1416,7 +1421,7 @@ export async function runReplyAgent(params: {
14161421// Otherwise, a late typing trigger (e.g. from a tool callback) can outlive the run and
14171422// keep the typing indicator stuck.
14181423if (payloadArray.length === 0) {
1419-return finalizeWithFollowup(undefined, queueKey, runFollowupTurn);
1424+return returnAfterReplyOperationClearsThenDrainFollowups(undefined);
14201425}
1421142614221427const currentMessageId = sessionCtx.MessageSidFull ?? sessionCtx.MessageSid;
@@ -1448,7 +1453,7 @@ export async function runReplyAgent(params: {
14481453didLogHeartbeatStrip = payloadResult.didLogHeartbeatStrip;
1449145414501455if (replyPayloads.length === 0) {
1451-return finalizeWithFollowup(undefined, queueKey, runFollowupTurn);
1456+return returnAfterReplyOperationClearsThenDrainFollowups(undefined);
14521457}
1453145814541459const successfulCronAdds = runResult.successfulCronAdds ?? 0;
@@ -1865,10 +1870,8 @@ export async function runReplyAgent(params: {
18651870}
18661871}
186718721868-const result = finalizeWithFollowup(
1873+const result = returnAfterReplyOperationClearsThenDrainFollowups(
18691874finalPayloads.length === 1 ? finalPayloads[0] : finalPayloads,
1870-queueKey,
1871-runFollowupTurn,
18721875);
1873187618741877return result;
@@ -1877,38 +1880,37 @@ export async function runReplyAgent(params: {
18771880replyOperation.result?.kind === "aborted" &&
18781881replyOperation.result.code === "aborted_for_restart"
18791882) {
1880-return finalizeWithFollowup(
1881-{ text: "⚠️ Gateway is restarting. Please wait a few seconds and try again." },
1882-queueKey,
1883-runFollowupTurn,
1884-);
1883+return returnAfterReplyOperationClearsThenDrainFollowups({
1884+text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",
1885+});
18851886}
18861887if (replyOperation.result?.kind === "aborted") {
1887-return finalizeWithFollowup({ text: SILENT_REPLY_TOKEN }, queueKey, runFollowupTurn);
1888+return returnAfterReplyOperationClearsThenDrainFollowups({ text: SILENT_REPLY_TOKEN });
18881889}
18891890if (error instanceof GatewayDrainingError) {
18901891replyOperation.fail("gateway_draining", error);
1891-return finalizeWithFollowup(
1892-{ text: "⚠️ Gateway is restarting. Please wait a few seconds and try again." },
1893-queueKey,
1894-runFollowupTurn,
1895-);
1892+return returnAfterReplyOperationClearsThenDrainFollowups({
1893+text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",
1894+});
18961895}
18971896if (error instanceof CommandLaneClearedError) {
18981897replyOperation.fail("command_lane_cleared", error);
1899-return finalizeWithFollowup(
1900-{ text: "⚠️ Gateway is restarting. Please wait a few seconds and try again." },
1901-queueKey,
1902-runFollowupTurn,
1903-);
1898+return returnAfterReplyOperationClearsThenDrainFollowups({
1899+text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",
1900+});
19041901}
19051902replyOperation.fail("run_failed", error);
19061903// Keep the followup queue moving even when an unexpected exception escapes
19071904// the run path; the caller still receives the original error.
1908-finalizeWithFollowup(undefined, queueKey, runFollowupTurn);
1905+returnAfterReplyOperationClearsThenDrainFollowups(undefined);
19091906throw error;
19101907} finally {
19111908replyOperation.complete();
1909+if (shouldDrainFollowupsAfterReplyOperationClears) {
1910+// Same-session follow-up turns create their own ReplyOperation; start them
1911+// only after this run clears the active-run guard.
1912+scheduleFollowupDrain(queueKey, runFollowupTurn);
1913+}
19121914blockReplyPipeline?.stop();
19131915typing.markRunComplete();
19141916// Safety net: the dispatcher's onIdle callback normally fires
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。