@@ -1,4 +1,5 @@
11import { resolveSessionAgentId } from "../agents/agent-scope.js";
2+import { REPLY_RUN_STILL_SHUTTING_DOWN_TEXT } from "../auto-reply/reply/get-reply-run-queue.js";
23import { finalizeInboundContext } from "../auto-reply/reply/inbound-context.js";
34import { dispatchReplyWithBufferedBlockDispatcher } from "../auto-reply/reply/provider-dispatcher.js";
45import type { ChatType } from "../channels/chat-type.js";
@@ -27,6 +28,7 @@ import {
2728import {
2829drainPendingSessionDeliveries,
2930enqueueSessionDelivery,
31+loadPendingSessionDelivery,
3032recoverPendingSessionDeliveries,
3133type QueuedSessionDelivery,
3234type QueuedSessionDeliveryPayload,
@@ -49,8 +51,14 @@ import { runStartupTasks, type StartupTask } from "./startup-tasks.js";
4951const log = createSubsystemLogger("gateway/restart-sentinel");
5052const OUTBOUND_RETRY_DELAY_MS = 1_000;
5153const OUTBOUND_MAX_ATTEMPTS = 45;
54+const RESTART_CONTINUATION_BUSY_RETRY_DELAY_MS = process.env.VITEST ? 1 : 6_000;
55+const RESTART_CONTINUATION_BUSY_MAX_ATTEMPTS = 5;
56+const RESTART_CONTINUATION_BUSY_RETRY_ERROR =
57+"restart continuation deferred because previous run is still shutting down";
5258let latestUpdateRestartSentinel: RestartSentinelPayload | null = null;
535960+type QueuedAgentTurnSessionDelivery = Extract<QueuedSessionDelivery, { kind: "agentTurn" }>;
61+5462function cloneRestartSentinelPayload(
5563payload: RestartSentinelPayload | null,
5664): RestartSentinelPayload | null {
@@ -203,6 +211,23 @@ function resolveRestartContinuationOutboundPayload(params: {
203211return params.replyToId ? { ...payload, replyToId: params.replyToId } : payload;
204212}
205213214+function isRestartContinuationBusyPayload(payload: OutboundReplyPayload): boolean {
215+return (
216+typeof payload.text === "string" && payload.text.trim() === REPLY_RUN_STILL_SHUTTING_DOWN_TEXT
217+);
218+}
219+220+function isRestartContinuationBusyRetry(entry: QueuedSessionDelivery | null): boolean {
221+return entry?.lastError === RESTART_CONTINUATION_BUSY_RETRY_ERROR;
222+}
223+224+function resolveQueuedRestartContinuationMessageId(entry: QueuedAgentTurnSessionDelivery): string {
225+if (isRestartContinuationBusyRetry(entry) && entry.retryCount > 0) {
226+return `${entry.messageId}:retry:${entry.retryCount}`;
227+}
228+return entry.messageId;
229+}
230+206231function resolveQueuedSessionDeliveryContext(entry: QueuedSessionDelivery):
207232| {
208233channel?: string;
@@ -270,7 +295,7 @@ async function deliverQueuedSessionDelivery(params: {
270295}
271296272297const route = params.entry.route;
273-const messageId = params.entry.messageId;
298+const messageId = resolveQueuedRestartContinuationMessageId(params.entry);
274299const userMessage = params.entry.message.trim();
275300const agentId = resolveSessionAgentId({
276301sessionKey: canonicalKey,
@@ -320,12 +345,16 @@ async function deliverQueuedSessionDelivery(params: {
320345 recordInboundSession,
321346 dispatchReplyWithBufferedBlockDispatcher,
322347delivery: {
323-preparePayload: (payload) =>
324-resolveRestartContinuationOutboundPayload({
348+preparePayload: (payload) => {
349+if (isRestartContinuationBusyPayload(payload)) {
350+throw new Error(RESTART_CONTINUATION_BUSY_RETRY_ERROR);
351+}
352+return resolveRestartContinuationOutboundPayload({
325353 payload,
326354 messageId,
327355replyToId: route.replyToId,
328-}),
356+});
357+},
329358durable: (_payload, info) =>
330359info.kind === "final"
331360 ? {
@@ -421,16 +450,30 @@ async function drainRestartContinuationQueue(params: {
421450entryId: string;
422451log: SessionDeliveryRecoveryLogger;
423452}) {
424-await drainPendingSessionDeliveries({
425-drainKey: `restart-continuation:${params.entryId}`,
426-logLabel: "restart continuation",
427-log: params.log,
428-deliver: (entry) => deliverQueuedSessionDelivery({ deps: params.deps, entry }),
429-selectEntry: (entry) => ({
430-match: entry.id === params.entryId,
431-bypassBackoff: true,
432-}),
433-});
453+for (let attempt = 1; attempt <= RESTART_CONTINUATION_BUSY_MAX_ATTEMPTS; attempt += 1) {
454+await drainPendingSessionDeliveries({
455+drainKey: `restart-continuation:${params.entryId}`,
456+logLabel: "restart continuation",
457+log: params.log,
458+deliver: (entry) => deliverQueuedSessionDelivery({ deps: params.deps, entry }),
459+selectEntry: (entry) => ({
460+match: entry.id === params.entryId,
461+bypassBackoff: true,
462+}),
463+});
464+465+const queued = await loadPendingSessionDelivery(params.entryId);
466+if (!isRestartContinuationBusyRetry(queued)) {
467+return;
468+}
469+if (attempt >= RESTART_CONTINUATION_BUSY_MAX_ATTEMPTS) {
470+return;
471+}
472+params.log.info(
473+`restart continuation: entry ${params.entryId} still waiting for the previous run to clear; retrying in ${RESTART_CONTINUATION_BUSY_RETRY_DELAY_MS}ms`,
474+);
475+await waitForOutboundRetry(RESTART_CONTINUATION_BUSY_RETRY_DELAY_MS);
476+}
434477}
435478436479export async function recoverPendingRestartContinuationDeliveries(params: {
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。