diff --git a/src/auto-reply/reply/dispatch-acp.test.ts b/src/auto-reply/reply/dispatch-acp.test.ts index 52fc592fda6e..82c49e7e2f4f 100644 --- a/src/auto-reply/reply/dispatch-acp.test.ts +++ b/src/auto-reply/reply/dispatch-acp.test.ts @@ -194,6 +194,11 @@ vi.mock("./dispatch-acp-session.runtime.js", () => ({ vi.mock("../../logging/diagnostic.js", () => ({ markDiagnosticSessionProgress: diagnosticMocks.markDiagnosticSessionProgress, + isStuckSessionRecoveryEnabled: (config?: { diagnostics?: { enabled?: boolean } }) => + config?.diagnostics?.enabled !== false, + requestStuckDiagnosticSessionRecovery: vi.fn(), + resolveStuckSessionWarnMs: () => 120_000, + resolveStuckSessionAbortMs: () => 360_000, })); vi.mock("./dispatch-acp-transcript.runtime.js", () => ({ diff --git a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts index 8035ed23d9d0..3d27a70aa702 100644 --- a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts +++ b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts @@ -2,6 +2,7 @@ import { vi } from "vitest"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { SessionBindingRecord } from "../../infra/outbound/session-binding-service.js"; +import type { StuckSessionRecoveryOutcome } from "../../logging/diagnostic-session-recovery.js"; import type { PluginHookBeforeDispatchResult, PluginHookReplyDispatchResult, @@ -32,6 +33,13 @@ const diagnosticMocks = vi.hoisted(() => ({ logMessageProcessed: vi.fn(), logSessionStateChange: vi.fn(), markDiagnosticSessionProgress: vi.fn(), + requestStuckDiagnosticSessionRecovery: vi.fn<() => Promise>( + async () => ({ + status: "skipped" as const, + action: "keep_lane" as const, + reason: "active_reply_work" as const, + }), + ), })); const hookMocks = vi.hoisted(() => ({ registry: { @@ -205,6 +213,19 @@ vi.mock("../../logging/diagnostic.js", () => ({ logMessageProcessed: diagnosticMocks.logMessageProcessed, logSessionStateChange: diagnosticMocks.logSessionStateChange, markDiagnosticSessionProgress: diagnosticMocks.markDiagnosticSessionProgress, + isStuckSessionRecoveryEnabled: (config?: { diagnostics?: { enabled?: boolean } }) => + config?.diagnostics?.enabled !== false, + requestStuckDiagnosticSessionRecovery: diagnosticMocks.requestStuckDiagnosticSessionRecovery, + resolveStuckSessionWarnMs: (config?: { diagnostics?: { stuckSessionWarnMs?: number } }) => + config?.diagnostics?.stuckSessionWarnMs ?? 120_000, + resolveStuckSessionAbortMs: ( + config: { diagnostics?: { stuckSessionAbortMs?: number } } | undefined, + stuckSessionWarnMs: number, + ) => + Math.max( + stuckSessionWarnMs, + config?.diagnostics?.stuckSessionAbortMs ?? Math.max(300_000, stuckSessionWarnMs * 3), + ), })); vi.mock("../../config/sessions/thread-info.js", () => ({ parseSessionThreadInfo: (sessionKey: string | undefined) => diff --git a/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts b/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts new file mode 100644 index 000000000000..22331911219c --- /dev/null +++ b/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts @@ -0,0 +1,690 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import type { ReplyPayload } from "../types.js"; +import { + createDispatcher, + diagnosticMocks, + emptyConfig, + mocks, + noAbortResult, + resetPluginTtsAndThreadMocks, + runtimePluginMocks, +} from "./dispatch-from-config.shared.test-harness.js"; +import { buildTestCtx } from "./test-ctx.js"; + +let dispatchReplyFromConfig: typeof import("./dispatch-from-config.js").dispatchReplyFromConfig; +let createReplyOperation: typeof import("./reply-run-registry.js").createReplyOperation; +let replyRunTesting: typeof import("./reply-run-registry.js").__testing; +let resetInboundDedupe: typeof import("./inbound-dedupe.js").resetInboundDedupe; + +function setNoAbort() { + mocks.tryFastAbortFromMessage.mockResolvedValue(noAbortResult); +} + +describe("dispatchReplyFromConfig stale visible admission recovery", () => { + beforeEach(async () => { + ({ dispatchReplyFromConfig } = await import("./dispatch-from-config.js")); + ({ createReplyOperation, __testing: replyRunTesting } = + await import("./reply-run-registry.js")); + ({ resetInboundDedupe } = await import("./inbound-dedupe.js")); + replyRunTesting.resetReplyRunRegistry(); + resetInboundDedupe(); + resetPluginTtsAndThreadMocks(); + runtimePluginMocks.ensureRuntimePluginsLoaded.mockReset(); + mocks.routeReply.mockReset(); + mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.tryFastAbortFromMessage.mockReset(); + setNoAbort(); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockReset(); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "skipped", + action: "keep_lane", + reason: "active_reply_work", + }); + }); + + afterEach(() => { + vi.useRealTimers(); + replyRunTesting.resetReplyRunRegistry(); + resetInboundDedupe(); + }); + + it("recovers stale visible reply work and retries dispatch admission", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:1"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: "active-session", + sessionKey, + queueDepth: 1, + staleActiveProgressAbortMs: 1_000, + }), + ); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("reclaims a pure stale reply registry lock when recovery finds no active work", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:pure-stale-registry"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "noop", + action: "none", + reason: "no_active_work", + sessionId: "active-session", + sessionKey, + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + expect(activeOperation.result).toMatchObject({ + kind: "failed", + code: "run_failed", + }); + }); + + it("does not clear a fresh reply operation with the same session id after recovery", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:fresh-same-session"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + let freshOperation: ReturnType | undefined; + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockImplementationOnce(async () => { + activeOperation.complete(); + freshOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + freshOperation.setPhase("running"); + return { + status: "noop", + action: "none", + reason: "no_active_work", + sessionId: "active-session", + sessionKey, + }; + }) + .mockImplementationOnce(async () => { + freshOperation?.fail("run_failed", new Error("fresh operation later became stale")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(freshOperation?.result).toBeNull(); + expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("keeps waiting when recovery observes active reply work", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:active-reply-work"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockResolvedValueOnce({ + status: "skipped", + action: "keep_lane", + reason: "active_reply_work", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + }) + .mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(activeOperation.result).toBeNull(); + expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("keeps waiting when another recovery is already in flight", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:in-flight"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockResolvedValueOnce({ + status: "skipped", + action: "observe_only", + reason: "already_in_flight", + sessionId: "active-session", + sessionKey, + }) + .mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("keeps waiting when recovery observes an active lane task", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:active-lane-task"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockResolvedValueOnce({ + status: "skipped", + action: "keep_lane", + reason: "active_lane_task", + sessionId: "active-session", + sessionKey, + activeCount: 1, + queuedCount: 1, + }) + .mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("does not clear active reply work when recovery fails", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:recovery-failed"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "failed", + action: "none", + reason: "exception", + sessionId: "active-session", + sessionKey, + error: "recovery failed", + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(1); + expect(result).toMatchObject({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(activeOperation.result).toBeNull(); + expect(replyResolver).not.toHaveBeenCalled(); + expect(dispatcher.sendFinalReply).not.toHaveBeenCalled(); + }); + + it("clears stale reply work after recovery releases lane state", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:released-lane"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "released", + action: "release_lane", + sessionId: "active-session", + sessionKey, + released: 1, + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(1); + const result = await resultPromise; + + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + expect(activeOperation.result).toMatchObject({ + kind: "failed", + code: "run_failed", + }); + }); + + it("does not run visible stuck recovery when diagnostics are disabled", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:diagnostics-disabled"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + enabled: false, + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).not.toHaveBeenCalled(); + expect(replyResolver).not.toHaveBeenCalled(); + + activeOperation.complete(); + const result = await resultPromise; + + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("releases inbound dedupe when active reply admission is aborted before processing", async () => { + const sessionKey = "agent:main:telegram:direct:dedupe"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const abortController = new AbortController(); + const ctx = buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "telegram:user-1", + To: "telegram:user-1", + ChatType: "direct", + SessionKey: sessionKey, + MessageSid: "message-1", + BodyForAgent: "second visible turn", + }); + const firstDispatcher = createDispatcher(); + const firstReplyResolver = vi.fn( + async () => ({ text: "should not run" }) satisfies ReplyPayload, + ); + + const firstResult = dispatchReplyFromConfig({ + ctx, + cfg: emptyConfig, + dispatcher: firstDispatcher, + replyOptions: { abortSignal: abortController.signal }, + replyResolver: firstReplyResolver, + }); + setTimeout(() => abortController.abort(), 0); + + await expect(firstResult).resolves.toMatchObject({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(firstReplyResolver).not.toHaveBeenCalled(); + expect(firstDispatcher.sendFinalReply).not.toHaveBeenCalled(); + + activeOperation.complete(); + + const secondDispatcher = createDispatcher(); + const secondReplyResolver = vi.fn( + async () => ({ text: "runs after dedupe release" }) satisfies ReplyPayload, + ); + await expect( + dispatchReplyFromConfig({ + ctx, + cfg: emptyConfig, + dispatcher: secondDispatcher, + replyResolver: secondReplyResolver, + }), + ).resolves.toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(secondReplyResolver).toHaveBeenCalledTimes(1); + expect(secondDispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index 596333a1e4c8..a0c082316b7d 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -13,6 +13,7 @@ import { runWithDiagnosticTraceContext, } from "../../infra/diagnostic-trace-context.js"; import type { SessionBindingRecord } from "../../infra/outbound/session-binding-service.js"; +import type { StuckSessionRecoveryOutcome } from "../../logging/diagnostic-session-recovery.js"; import type { AcpRuntime, AcpRuntimeEnsureInput, @@ -64,6 +65,13 @@ const diagnosticMocks = vi.hoisted(() => ({ logMessageProcessed: vi.fn(), logSessionStateChange: vi.fn(), markDiagnosticSessionProgress: vi.fn(), + requestStuckDiagnosticSessionRecovery: vi.fn<() => Promise>( + async () => ({ + status: "skipped" as const, + action: "keep_lane" as const, + reason: "active_reply_work" as const, + }), + ), })); const hookMocks = vi.hoisted(() => ({ registry: { @@ -408,6 +416,19 @@ vi.mock("../../logging/diagnostic.js", () => ({ logMessageProcessed: diagnosticMocks.logMessageProcessed, logSessionStateChange: diagnosticMocks.logSessionStateChange, markDiagnosticSessionProgress: diagnosticMocks.markDiagnosticSessionProgress, + isStuckSessionRecoveryEnabled: (config?: { diagnostics?: { enabled?: boolean } }) => + config?.diagnostics?.enabled !== false, + requestStuckDiagnosticSessionRecovery: diagnosticMocks.requestStuckDiagnosticSessionRecovery, + resolveStuckSessionWarnMs: (config?: { diagnostics?: { stuckSessionWarnMs?: number } }) => + config?.diagnostics?.stuckSessionWarnMs ?? 120_000, + resolveStuckSessionAbortMs: ( + config: { diagnostics?: { stuckSessionAbortMs?: number } } | undefined, + stuckSessionWarnMs: number, + ) => + Math.max( + stuckSessionWarnMs, + config?.diagnostics?.stuckSessionAbortMs ?? Math.max(300_000, stuckSessionWarnMs * 3), + ), })); vi.mock("../../config/sessions/thread-info.js", () => ({ parseSessionThreadInfo: (sessionKey: string | undefined) => @@ -950,6 +971,12 @@ describe("dispatchReplyFromConfig", () => { diagnosticMocks.logMessageProcessed.mockClear(); diagnosticMocks.logSessionStateChange.mockClear(); diagnosticMocks.markDiagnosticSessionProgress.mockClear(); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockReset(); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "skipped", + action: "keep_lane", + reason: "active_reply_work", + }); diagnosticMocks.logMessageDispatchStarted.mockClear(); diagnosticMocks.logMessageDispatchCompleted.mockClear(); hookMocks.runner.hasHooks.mockClear(); diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index 98a7bdca1470..ee30a5236ea2 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -61,10 +61,15 @@ import { measureDiagnosticsTimelineSpan } from "../../infra/diagnostics-timeline import { formatErrorMessage } from "../../infra/errors.js"; import { getSessionBindingService } from "../../infra/outbound/session-binding-service.js"; import { isAbortError } from "../../infra/unhandled-rejections.js"; +import type { StuckSessionRecoveryOutcome } from "../../logging/diagnostic-session-recovery.js"; import { logMessageDispatchCompleted, logMessageDispatchStarted, + isStuckSessionRecoveryEnabled, markDiagnosticSessionProgress, + requestStuckDiagnosticSessionRecovery, + resolveStuckSessionAbortMs, + resolveStuckSessionWarnMs, } from "../../logging/diagnostic.js"; import { createDiagnosticMessageLifecycle } from "../../logging/message-lifecycle.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; @@ -143,7 +148,11 @@ import type { ReplyDispatcher, } from "./reply-dispatcher.types.js"; import { readDispatcherFailedCounts } from "./reply-dispatcher.types.js"; -import { replyRunRegistry, type ReplyOperation } from "./reply-run-registry.js"; +import { + forceClearReplyRunBySessionId, + replyRunRegistry, + type ReplyOperation, +} from "./reply-run-registry.js"; import { isReplyProfilerEnabled } from "./reply-timing-tracker.js"; import { admitReplyTurn, resolveReplyTurnKind } from "./reply-turn-admission.js"; import { resolveRoutedDeliveryThreadId } from "./routed-delivery-thread.js"; @@ -764,6 +773,31 @@ export function getDispatcherFinalOutcomeCounts(dispatcher: DispatcherOutcomeCou }; } +function visibleRecoveryClearedActiveWork(outcome: StuckSessionRecoveryOutcome): boolean { + return ( + outcome.status === "aborted" || + outcome.status === "released" || + (outcome.status === "noop" && outcome.reason === "no_active_work") + ); +} + +function isSameReplyOperation( + left: ReplyOperation | undefined, + right: ReplyOperation | undefined, +): boolean { + return Boolean(left && right && left === right); +} + +function visibleRecoveryShouldKeepWaiting(outcome: StuckSessionRecoveryOutcome): boolean { + return ( + outcome.status === "skipped" && + (outcome.reason === "active_reply_work" || + outcome.reason === "active_embedded_run" || + outcome.reason === "active_lane_task" || + outcome.reason === "already_in_flight") + ); +} + function sourceReplyTranscriptMirrorForDeliveredPayload( metadata: SourceReplyTranscriptMirror, payload: ReplyPayload, @@ -1151,6 +1185,10 @@ export async function dispatchReplyFromConfig( markDiagnosticSessionProgress({ sessionKey: acpDispatchSessionKey }); } }; + const visibleReplyRecoveryWaitMs = (() => { + const warnMs = resolveStuckSessionWarnMs(cfg); + return resolveStuckSessionAbortMs(cfg, warnMs); + })(); const sessionStoreEntry = boundAcpDispatchSessionKey ? resolveSessionStoreLookup({ ...ctx, SessionKey: boundAcpDispatchSessionKey }, cfg) : initialSessionStoreEntry; @@ -1214,7 +1252,7 @@ export async function dispatchReplyFromConfig( if (!dispatchOperationSessionKey) { return { status: "ready" }; } - const operationSessionId = + let operationSessionId = dispatchAbortOperation?.sessionId ?? initialSessionStoreEntry.entry?.sessionId ?? sessionStoreEntry.entry?.sessionId ?? @@ -1228,7 +1266,23 @@ export async function dispatchReplyFromConfig( ctx, routeThreadId, }); - const admission = await admitReplyTurn({ + const shouldRecoverStaleVisibleOperation = + phase === "dispatch" && + replyTurnKind === "visible" && + !allowSlackRoutedThreadBypass && + isStuckSessionRecoveryEnabled(cfg) && + params.replyOptions?.abortSignal?.aborted !== true; + const recoverStaleVisibleOperation = async ( + activeOperation: ReplyOperation, + ): Promise => + requestStuckDiagnosticSessionRecovery({ + sessionId: activeOperation.sessionId, + sessionKey: dispatchOperationSessionKey, + ageMs: visibleReplyRecoveryWaitMs, + queueDepth: 1, + staleActiveProgressAbortMs: visibleReplyRecoveryWaitMs, + }); + let admission = await admitReplyTurn({ sessionKey: dispatchOperationSessionKey, sessionId: operationSessionId, kind: replyTurnKind, @@ -1236,7 +1290,55 @@ export async function dispatchReplyFromConfig( routeThreadId, upstreamAbortSignal: params.replyOptions?.abortSignal, waitForActive: !allowActivePreDispatch && !allowSlackRoutedThreadBypass, + ...(shouldRecoverStaleVisibleOperation ? { waitTimeoutMs: visibleReplyRecoveryWaitMs } : {}), }); + if (shouldRecoverStaleVisibleOperation) { + while ( + admission.status === "skipped" && + admission.reason === "active-run" && + admission.activeOperation + ) { + operationSessionId = admission.activeOperation.sessionId; + const recovery = await recoverStaleVisibleOperation(admission.activeOperation); + let activeAfterRecovery = replyRunRegistry.get(dispatchOperationSessionKey); + if ( + recovery && + visibleRecoveryClearedActiveWork(recovery) && + isSameReplyOperation(activeAfterRecovery, admission.activeOperation) + ) { + forceClearReplyRunBySessionId( + admission.activeOperation.sessionId, + new Error("Stale visible reply operation recovered without clearing reply registry"), + ); + activeAfterRecovery = replyRunRegistry.get(dispatchOperationSessionKey); + if (isSameReplyOperation(activeAfterRecovery, admission.activeOperation)) { + break; + } + } + const replyOperationStillActive = Boolean(activeAfterRecovery); + if ( + replyOperationStillActive && + (!recovery || + (!visibleRecoveryClearedActiveWork(recovery) && + !visibleRecoveryShouldKeepWaiting(recovery))) + ) { + break; + } + if (activeAfterRecovery) { + operationSessionId = activeAfterRecovery.sessionId; + } + admission = await admitReplyTurn({ + sessionKey: dispatchOperationSessionKey, + sessionId: operationSessionId, + kind: replyTurnKind, + resetTriggered: false, + routeThreadId, + upstreamAbortSignal: params.replyOptions?.abortSignal, + waitForActive: replyOperationStillActive, + waitTimeoutMs: visibleReplyRecoveryWaitMs, + }); + } + } if (admission.status === "skipped") { if (allowActivePreDispatch && admission.reason === "active-run") { preDispatchAbortOperation = admission.activeOperation; @@ -1748,7 +1850,13 @@ export async function dispatchReplyFromConfig( commitInboundDedupe(inboundDedupeClaim.key); } }; + const releaseInboundDedupeIfClaimed = () => { + if (inboundDedupeClaim.status === "claimed") { + releaseInboundDedupe(inboundDedupeClaim.key); + } + }; const finishReplyOperationBusyDispatch = (opts?: { + dedupeDisposition?: "commit" | "release"; recordAgentDispatchCompleted?: boolean; sessionMetadataChanges?: DispatchFromConfigResult["sessionMetadataChanges"]; }): DispatchFromConfigResult => { @@ -1757,7 +1865,11 @@ export async function dispatchReplyFromConfig( } recordProcessed("skipped", { reason: "reply-operation-active" }); markIdle("message_completed"); - commitInboundDedupeIfClaimed(); + if (opts?.dedupeDisposition === "release") { + releaseInboundDedupeIfClaimed(); + } else { + commitInboundDedupeIfClaimed(); + } return attachSourceReplyDeliveryMode({ queuedFinal: false, counts: dispatcher.getQueuedCounts(), @@ -1973,7 +2085,7 @@ export async function dispatchReplyFromConfig( // Register the dispatch-owned operation before any plugin hook or model work // so /stop can abort pre-run and in-run stalls through the same session lane. if ((await ensureDispatchReplyOperation("pre_dispatch")).status === "busy") { - return finishReplyOperationBusyDispatch(); + return finishReplyOperationBusyDispatch({ dedupeDisposition: "release" }); } const shouldSuppressDefaultToolProgressMessages = () => !shouldEmitVerboseProgress(); @@ -2213,7 +2325,7 @@ export async function dispatchReplyFromConfig( } if ((await ensureDispatchReplyOperation("dispatch")).status === "busy") { - return finishReplyOperationBusyDispatch(); + return finishReplyOperationBusyDispatch({ dedupeDisposition: "release" }); } // When automatic source delivery is suppressed, still let the agent process diff --git a/src/logging/diagnostic-session-recovery-coordinator.ts b/src/logging/diagnostic-session-recovery-coordinator.ts index 40762e0e3e47..c53545196370 100644 --- a/src/logging/diagnostic-session-recovery-coordinator.ts +++ b/src/logging/diagnostic-session-recovery-coordinator.ts @@ -27,6 +27,12 @@ export type RecoverStuckSession = ( params: StuckSessionRecoveryRequest, ) => void | StuckSessionRecoveryOutcome | Promise; +export type RequestStuckSessionRecoveryParams = { + recover: RecoverStuckSession; + request: StuckSessionRecoveryRequest; + classification: SessionAttentionClassification; +}; + const recoveryRequestsInFlight = new Set(); function emitSessionRecoveryRequested(params: { @@ -167,25 +173,21 @@ function applyRecoveryOutcomeToDiagnosticState(params: { markActivity(); } -export function requestStuckSessionRecovery(params: { - recover: RecoverStuckSession; - request: StuckSessionRecoveryRequest; - classification: SessionAttentionClassification; -}): void { +export function requestStuckSessionRecoveryOutcome( + params: RequestStuckSessionRecoveryParams, +): Promise { const inFlightKey = recoveryRequestKey(params.request); if (inFlightKey && recoveryRequestsInFlight.has(inFlightKey)) { - emitSessionRecoveryCompleted({ - request: params.request, - outcome: { - status: "skipped", - action: "observe_only", - reason: "already_in_flight", - sessionId: params.request.sessionId, - sessionKey: params.request.sessionKey, - activeWorkKind: params.classification.activeWorkKind, - }, - }); - return; + const outcome: StuckSessionRecoveryOutcome = { + status: "skipped", + action: "observe_only", + reason: "already_in_flight", + sessionId: params.request.sessionId, + sessionKey: params.request.sessionKey, + activeWorkKind: params.classification.activeWorkKind, + }; + emitSessionRecoveryCompleted({ request: params.request, outcome }); + return Promise.resolve(outcome); } if (inFlightKey) { recoveryRequestsInFlight.add(inFlightKey); @@ -201,53 +203,56 @@ export function requestStuckSessionRecovery(params: { recoveryRequestsInFlight.delete(inFlightKey); } }; - const failRecovery = (err: unknown) => { + const completeRecovery = (outcome: StuckSessionRecoveryOutcome | undefined) => { applyRecoveryOutcomeToDiagnosticState({ request: params.request, - outcome: { - status: "failed", - action: "none", - reason: "exception", - sessionId: params.request.sessionId, - sessionKey: params.request.sessionKey, - error: String(err), - }, + outcome, recoveryStartedAfterEmbeddedRunSequence, recoveryStartedAfterDiagnosticEventSequence, }); + return outcome; + }; + const failRecovery = (err: unknown) => { + const outcome: StuckSessionRecoveryOutcome = { + status: "failed", + action: "none", + reason: "exception", + sessionId: params.request.sessionId, + sessionKey: params.request.sessionKey, + error: String(err), + }; + applyRecoveryOutcomeToDiagnosticState({ + request: params.request, + outcome, + recoveryStartedAfterEmbeddedRunSequence, + recoveryStartedAfterDiagnosticEventSequence, + }); + return outcome; }; try { const result = params.recover(params.request); if (isRecoveryPromiseLike(result)) { - void result - .then((outcome) => { - applyRecoveryOutcomeToDiagnosticState({ - request: params.request, - outcome: outcome ?? undefined, - recoveryStartedAfterEmbeddedRunSequence, - recoveryStartedAfterDiagnosticEventSequence, - }); - }) + return result + .then((outcome) => completeRecovery(outcome ?? undefined)) .catch(failRecovery) .finally(clearInFlight); - return; } - applyRecoveryOutcomeToDiagnosticState({ - request: params.request, - outcome: result ?? undefined, - recoveryStartedAfterEmbeddedRunSequence, - recoveryStartedAfterDiagnosticEventSequence, - }); + const outcome = completeRecovery(result ?? undefined); clearInFlight(); + return Promise.resolve(outcome); } catch (err) { try { - failRecovery(err); + return Promise.resolve(failRecovery(err)); } finally { clearInFlight(); } } } +export function requestStuckSessionRecovery(params: RequestStuckSessionRecoveryParams): void { + void requestStuckSessionRecoveryOutcome(params); +} + export function resetDiagnosticSessionRecoveryCoordinatorForTest(): void { recoveryRequestsInFlight.clear(); } diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts index ec5411b5f526..2c8f1ad2ba98 100644 --- a/src/logging/diagnostic.ts +++ b/src/logging/diagnostic.ts @@ -38,6 +38,7 @@ import { } from "./diagnostic-session-context.js"; import { requestStuckSessionRecovery, + requestStuckSessionRecoveryOutcome, resetDiagnosticSessionRecoveryCoordinatorForTest, type RecoverStuckSession, } from "./diagnostic-session-recovery-coordinator.js"; @@ -176,6 +177,26 @@ async function recoverStuckSession( }); } +export function isStuckSessionRecoveryEnabled(config?: OpenClawConfig): boolean { + return areDiagnosticsEnabledForProcess() && isDiagnosticsEnabled(config); +} + +export async function requestStuckDiagnosticSessionRecovery( + params: StuckSessionRecoveryRequest, +): Promise { + return requestStuckSessionRecoveryOutcome({ + recover: recoverStuckSession, + classification: { + eventType: "session.stalled", + reason: "visible_reply_wait_timeout", + classification: "stalled_agent_run", + activeWorkKind: "embedded_run", + recoveryEligible: false, + }, + request: params, + }); +} + function formatDiagnosticWorkLabel( state: { sessionId?: string;