mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-28 05:16:23 -06:00
Fix stale visible reply recovery (#91840)
* fix visible reply stale recovery * fix visible recovery lint loop * fix visible reply registry recovery * test: cover failed visible recovery admission
This commit is contained in:
Josh Avant
@@ -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", () => ({
|
||||
|
||||
@@ -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<StuckSessionRecoveryOutcome>>(
|
||||
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) =>
|
||||
|
||||
@@ -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<typeof createReplyOperation> | 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);
|
||||
});
|
||||
});
|
||||
@@ -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<StuckSessionRecoveryOutcome>>(
|
||||
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();
|
||||
|
||||
@@ -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<StuckSessionRecoveryOutcome | undefined> =>
|
||||
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
|
||||
|
||||
@@ -27,6 +27,12 @@ export type RecoverStuckSession = (
|
||||
params: StuckSessionRecoveryRequest,
|
||||
) => void | StuckSessionRecoveryOutcome | Promise<void | StuckSessionRecoveryOutcome>;
|
||||
|
||||
export type RequestStuckSessionRecoveryParams = {
|
||||
recover: RecoverStuckSession;
|
||||
request: StuckSessionRecoveryRequest;
|
||||
classification: SessionAttentionClassification;
|
||||
};
|
||||
|
||||
const recoveryRequestsInFlight = new Set<string>();
|
||||
|
||||
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<StuckSessionRecoveryOutcome | undefined> {
|
||||
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();
|
||||
}
|
||||
|
||||
@@ -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<StuckSessionRecoveryOutcome | undefined> {
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user