@@ -135,6 +135,26 @@ import type { TypingController } from "./typing.js";
135135136136const BLOCK_REPLY_SEND_TIMEOUT_MS = 15_000;
137137138+function scheduleFollowupDrainAfterReplyOperationClear(params: {
139+operation: ReplyOperation;
140+queueKey: string;
141+runFollowup: (run: FollowupRun) => Promise<void>;
142+}): void {
143+runAfterReplyOperationClear(params.operation, (admissionSessionId) => {
144+const completedSessionId = params.operation.sessionId;
145+const runFollowupAfterClear =
146+admissionSessionId === completedSessionId
147+ ? params.runFollowup
148+ : (queued: FollowupRun) =>
149+params.runFollowup(
150+queued.run.sessionId === completedSessionId
151+ ? { ...queued, admissionSessionId }
152+ : queued,
153+);
154+scheduleFollowupDrain(params.queueKey, runFollowupAfterClear);
155+});
156+}
157+138158function markBeforeAgentRunBlockedPayloads(payloads: ReplyPayload[]): ReplyPayload[] {
139159return payloads.map((payload) =>
140160setReplyPayloadMetadata(payload, { beforeAgentRunBlocked: true }),
@@ -1317,12 +1337,19 @@ export async function runReplyAgent(params: {
13171337typing.cleanup();
13181338return undefined;
13191339}
1320-// Re-check liveness after enqueue so a stale active snapshot cannot leave
1321-// the followup queue idle if the original run already finished.
1322-const queuedBehindActiveRun = isRunActive?.() === true;
1323-if (!queuedBehindActiveRun) {
1340+// The queue must stay dormant while the active owner can still collect
1341+// messages. Registering after enqueue closes the owner-clear race.
1342+const activeReplyOperation = replyRunRegistry.get(queueKey);
1343+if (activeReplyOperation) {
1344+scheduleFollowupDrainAfterReplyOperationClear({
1345+operation: activeReplyOperation,
1346+ queueKey,
1347+runFollowup: queuedRunFollowupTurn,
1348+});
1349+} else {
13241350scheduleFollowupDrain(queueKey, queuedRunFollowupTurn);
13251351}
1352+const queuedBehindActiveRun = isRunActive?.() === true;
13261353await touchActiveSessionEntry();
13271354if (queuedBehindActiveRun) {
13281355await typingSignals.signalToolStart();
@@ -1464,19 +1491,6 @@ export async function runReplyAgent(params: {
14641491shouldDrainQueuedFollowupsAfterClear = true;
14651492return value;
14661493};
1467-const drainQueuedFollowupsAfterClear = (admissionSessionId: string) => {
1468-const completedSessionId = replyOperation.sessionId;
1469-const runFollowupAfterClear =
1470-admissionSessionId === completedSessionId
1471- ? runFollowupTurn
1472- : (queued: FollowupRun) =>
1473-runFollowupTurn(
1474-queued.run.sessionId === completedSessionId
1475- ? { ...queued, admissionSessionId }
1476- : queued,
1477-);
1478-scheduleFollowupDrain(queueKey, runFollowupAfterClear);
1479-};
14801494const restartRecoveryDeliveryRunId = crypto.randomUUID();
14811495let trackedRestartRecoveryDeliveryContext = false;
14821496const persistRestartRecoveryDeliveryContext = async (): Promise<void> => {
@@ -2625,10 +2639,13 @@ export async function runReplyAgent(params: {
26252639);
26262640}
26272641if (shouldDrainQueuedFollowupsAfterClear) {
2628-if (providedReplyOperation) {
2629-runAfterReplyOperationClear(replyOperation, drainQueuedFollowupsAfterClear);
2630-} else {
2631-replyOperation.completeThen(() => drainQueuedFollowupsAfterClear(replyOperation.sessionId));
2642+scheduleFollowupDrainAfterReplyOperationClear({
2643+operation: replyOperation,
2644+ queueKey,
2645+runFollowup: runFollowupTurn,
2646+});
2647+if (!providedReplyOperation) {
2648+replyOperation.complete();
26322649}
26332650} else if (!providedReplyOperation) {
26342651replyOperation.complete();
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。