ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-24 19:35:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(cron): preserve explicit delivery and timeout semantics

Browse Source
This commit is contained in:
Vincent Koc
2026-06-15 14:58:39 +08:00
parent 1c30bb8ce6
commit efbefceb0e
16 changed files with 106 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/cli-runner/prepare.test.ts
+4
View File
@@ -126,6 +126,8 @@ function createTestMcpLoopbackServerConfig(port: number) {
"x-openclaw-current-inbound-audio": "${OPENCLAW_MCP_CURRENT_INBOUND_AUDIO}",
"x-openclaw-inbound-event-kind": "${OPENCLAW_MCP_INBOUND_EVENT_KIND}",
"x-openclaw-source-reply-delivery-mode": "${OPENCLAW_MCP_SOURCE_REPLY_DELIVERY_MODE}",
"x-openclaw-require-explicit-message-target":
"${OPENCLAW_MCP_REQUIRE_EXPLICIT_MESSAGE_TARGET}",
},
},
},
@@ -1586,6 +1588,7 @@ describe("shouldSkipLocalCliCredentialEpoch", () => {
currentMessageId: "reply-message-1",
currentInboundAudio: true,
sourceReplyDeliveryMode: "message_tool_only",
requireExplicitMessageTarget: true,
});
expect(context.preparedBackend.env).toMatchObject({
@@ -1596,6 +1599,7 @@ describe("shouldSkipLocalCliCredentialEpoch", () => {
OPENCLAW_MCP_CURRENT_INBOUND_AUDIO: "true",
OPENCLAW_MCP_INBOUND_EVENT_KIND: "room_event",
OPENCLAW_MCP_SOURCE_REPLY_DELIVERY_MODE: "message_tool_only",
OPENCLAW_MCP_REQUIRE_EXPLICIT_MESSAGE_TARGET: "true",
});
} finally {
fs.rmSync(dir, { recursive: true, force: true });
src/agents/cli-runner/prepare.ts
+2
View File
@@ -385,6 +385,8 @@ export async function prepareCliRunContext(
OPENCLAW_MCP_CURRENT_INBOUND_AUDIO: params.currentInboundAudio === true ? "true" : "",
OPENCLAW_MCP_INBOUND_EVENT_KIND: params.currentInboundEventKind ?? "",
OPENCLAW_MCP_SOURCE_REPLY_DELIVERY_MODE: params.sourceReplyDeliveryMode ?? "",
OPENCLAW_MCP_REQUIRE_EXPLICIT_MESSAGE_TARGET:
params.requireExplicitMessageTarget === true ? "true" : "",
}
: undefined,
warn: (message) => cliBackendLog.warn(message),
src/agents/cli-runner/types.ts
+1
View File
@@ -75,6 +75,7 @@ export type RunCliAgentParams = {
jobId?: string;
extraSystemPrompt?: string;
sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
requireExplicitMessageTarget?: boolean;
silentReplyPromptMode?: SilentReplyPromptMode;
allowEmptyAssistantReplyAsSilent?: boolean;
/** Static portion of extraSystemPrompt (excluding per-message inbound metadata) for session reuse hashing. */
src/cron/isolated-agent/run-executor.ts
+1
View File
@@ -249,6 +249,7 @@ export function createCronPromptExecutor(params: {
skillsSnapshot: params.skillsSnapshot,
messageChannel,
sourceReplyDeliveryMode,
requireExplicitMessageTarget: params.sourceDelivery.messageTool.requireExplicitTarget,
abortSignal: params.abortSignal,
onExecutionStarted: params.onExecutionStarted,
onExecutionPhase: params.onExecutionPhase,
src/cron/isolated-agent/run.message-tool-policy.test.ts
+30
View File
@@ -9,6 +9,7 @@ import {
cleanupDirectCronSessionMock,
dispatchCronDeliveryMock,
getChannelPluginMock,
isCliProviderMock,
isHeartbeatOnlyResponseMock,
loadRunCronIsolatedAgentTurn,
makeCronSession,
@@ -20,6 +21,7 @@ import {
resolveCronDeliveryPlanMock,
resolveDeliveryTargetMock,
restoreFastTestEnv,
runCliAgentMock,
runEmbeddedAgentMock,
} from "./run.test-harness.js";
@@ -347,6 +349,7 @@ describe("runCronIsolatedAgentTurn message tool policy", () => {
},
messageToolEnabled: true,
messageToolForced: false,
requireExplicitMessageTarget: true,
requireExplicitMessageTargetEvidence: true,
directFallback: true,
}),
@@ -705,10 +708,37 @@ describe("runCronIsolatedAgentTurn message tool policy", () => {
expectEmbeddedRunFields({
sourceReplyDeliveryMode: undefined,
forceMessageTool: false,
requireExplicitMessageTarget: true,
messageChannel: "messagechat",
messageTo: "123",
currentChannelId: "123",
});
expect(expectEmbeddedRunPrompt()).toContain("with an explicit target");
});
it("requires explicit message targets for CLI-backed announce delivery", async () => {
mockRunCronFallbackPassthrough();
resolveCronDeliveryPlanMock.mockReturnValue(makeAnnounceDeliveryPlan());
isCliProviderMock.mockReturnValue(true);
runCliAgentMock.mockResolvedValue({
payloads: [{ text: "done" }],
meta: { agentMeta: { usage: { input: 10, output: 20 } } },
});
await runCronIsolatedAgentTurn({
...makeParams(),
job: makeAnnounceMessageToolJob(),
});
expect(runCliAgentMock).toHaveBeenCalledTimes(1);
expectRecordFields(
getMockCallArg(runCliAgentMock, 0, 0, "CLI run"),
{
messageChannel: "messagechat",
requireExplicitMessageTarget: true,
},
"CLI run params",
);
});
it("keeps automatic exec completion notifications when announce delivery is active", async () => {
src/cron/isolated-agent/run.meta-error-status.test.ts
+5 -1
View File
@@ -69,7 +69,11 @@ describe("runCronIsolatedAgentTurn - meta.error status propagation", () => {
it("keeps cron timeout result when executor rejects after the cron abort signal fires", async () => {
const abortController = new AbortController();
abortController.abort("cron: job execution timed out (last phase: model_call_started)");
const timeoutError = new Error(
"cron: job execution timed out (last phase: model_call_started)",
);
timeoutError.name = "TimeoutError";
abortController.abort(timeoutError);
runWithModelFallbackMock.mockRejectedValueOnce(
new Error(
'All models failed (2): openai/gpt-5.5: Command lane "cron-nested" task timed out after 330000ms (timeout)',
src/cron/isolated-agent/run.ts
+10 -9
View File
@@ -46,6 +46,7 @@ import {
createCronRunDiagnosticsFromError,
mergeCronRunDiagnostics,
} from "../run-diagnostics.js";
import { resolveCronAbortReasonText } from "../service/execution-errors.js";
import type {
CronAgentExecutionPhaseUpdate,
CronAgentExecutionStarted,
@@ -341,6 +342,7 @@ function resolveCronSourceDeliveryPlan(params: {
target,
messageToolEnabled: true,
messageToolForced: false,
requireExplicitMessageTarget: true,
requireExplicitMessageTargetEvidence: true,
directFallback: true,
skipFallbackWhenMessageToolSentToTarget: params.resolvedDelivery.ok,
@@ -425,14 +427,16 @@ function appendCronDeliveryInstruction(params: {
deliveryRequested: boolean;
messageToolEnabled: boolean;
resolvedDeliveryOk: boolean;
requireExplicitMessageTarget: boolean;
}) {
if (!params.deliveryRequested) {
return params.commandBody;
}
if (params.messageToolEnabled) {
const targetHint = params.resolvedDeliveryOk
? "for the current chat"
: "with an explicit target";
const targetHint =
params.requireExplicitMessageTarget || !params.resolvedDeliveryOk
? "with an explicit target"
: "for the current chat";
return `${params.commandBody}\n\nUse the message tool if you need to notify the user directly ${targetHint}. If you do not send directly, your final plain-text reply will be delivered automatically.`.trim();
}
return `${params.commandBody}\n\nReturn your response as plain text; it will be delivered automatically. If the task explicitly calls for messaging a specific external recipient, note who/where it should go instead of sending it yourself.`.trim();
@@ -833,6 +837,7 @@ async function prepareCronRunContext(params: {
toolsAllow: agentPayload?.toolsAllow,
}),
resolvedDeliveryOk: resolvedDelivery.ok,
requireExplicitMessageTarget: sourceDelivery.messageTool.requireExplicitTarget,
});
const skillsSnapshot = await resolveCronSkillsSnapshot({
@@ -1271,12 +1276,8 @@ export async function runCronIsolatedAgentTurn(params: {
const admittedLifecycleGeneration = getAgentEventLifecycleGeneration();
const abortSignal = params.abortSignal ?? params.signal;
const isAborted = () => abortSignal?.aborted === true;
const abortReason = () => {
const reason = abortSignal?.reason;
return typeof reason === "string" && reason.trim()
? reason.trim()
: "cron: job execution timed out";
};
const abortReason = () =>
resolveCronAbortReasonText(abortSignal?.reason) ?? "cron: job execution timed out";
const isFastTestEnv = process.env.OPENCLAW_TEST_FAST === "1";
const prepared = await prepareCronRunContext({ input: params, isFastTestEnv });
if (!prepared.ok) {
src/cron/service/execution-errors.ts
+10 -3
View File
@@ -39,12 +39,19 @@ export function preExecutionTimeoutErrorMessage(execution?: CronAgentExecutionSt
}
/** Extracts a human timeout/abort reason, falling back to the canonical cron timeout text. */
export function abortErrorMessage(signal?: AbortSignal): string {
const reason = signal?.reason;
export function resolveCronAbortReasonText(reason: unknown): string | undefined {
if (typeof reason === "string" && reason.trim()) {
return reason.trim();
}
return timeoutErrorMessage();
if (reason instanceof Error && reason.message.trim()) {
return reason.message.trim();
}
return undefined;
}
/** Extracts a human timeout/abort reason, falling back to the canonical cron timeout text. */
export function abortErrorMessage(signal?: AbortSignal): string {
return resolveCronAbortReasonText(signal?.reason) ?? timeoutErrorMessage();
}
function isAbortError(err: unknown): boolean {
src/cron/service/timer.regression.test.ts
+13
View File
@@ -2527,6 +2527,7 @@ describe("cron service timer regressions", () => {
let now = scheduledAt;
const wallStart = Date.now();
let abortWallMs: number | undefined;
let abortReason: unknown;
const started = createDeferred<void>();
const state = createCronServiceState({
@@ -2553,6 +2554,7 @@ describe("cron service timer regressions", () => {
}
if (abortSignal.aborted) {
abortWallMs = Date.now();
abortReason = abortSignal.reason;
resolve();
return;
}
@@ -2560,6 +2562,7 @@ describe("cron service timer regressions", () => {
"abort",
() => {
abortWallMs = Date.now();
abortReason = abortSignal.reason;
resolve();
},
{ once: true },
@@ -2582,6 +2585,10 @@ describe("cron service timer regressions", () => {
const elapsedMs = (abortWallMs ?? Date.now()) - wallStart;
expect(elapsedMs).toBeGreaterThanOrEqual(timeoutSeconds * 1_000);
expect(abortReason).toMatchObject({
name: "TimeoutError",
message: "cron: job execution timed out",
});
const job = state.store?.jobs.find((entry) => entry.id === "timeout-fraction-29774");
expect(job?.state.lastStatus).toBe("error");
@@ -2872,6 +2879,7 @@ describe("cron service timer regressions", () => {
let now = scheduledAt;
const started = createDeferred<void>();
let abortObserved = false;
let abortReason: unknown;
const cleanupTimedOutAgentRun = vi.fn(async () => {});
const onIsolatedAgentSetupTimeout = vi.fn();
const state = createCronServiceState({
@@ -2912,6 +2920,7 @@ describe("cron service timer regressions", () => {
"abort",
() => {
abortObserved = true;
abortReason = abortSignal.reason;
},
{ once: true },
);
@@ -2931,6 +2940,10 @@ describe("cron service timer regressions", () => {
expect(job.state.lastStatus).toBe("error");
expect(job.state.lastError).toContain("stalled before execution start");
expect(job.state.lastError).toContain("context-engine");
expect(abortReason).toMatchObject({
name: "TimeoutError",
message: expect.stringContaining("context-engine"),
});
expect(cleanupTimedOutAgentRun).toHaveBeenCalledTimes(1);
const cleanupArgs = requireRecord(firstMockArg(cleanupTimedOutAgentRun));
expect(requireRecord(cleanupArgs.job).id).toBe("isolated-pre-model-timeout-74803");
src/cron/service/timer.ts
+3 -1
View File
@@ -227,7 +227,9 @@ export async function executeJobCoreWithTimeout(
const triggerTimeout = (reason: string) => {
timeoutReason = reason;
if (!runAbortController.signal.aborted) {
runAbortController.abort(reason);
const timeoutError = new Error(reason);
timeoutError.name = "TimeoutError";
runAbortController.abort(timeoutError);
}
resolveTimeout?.(timeoutMarker);
};
src/gateway/mcp-http.loopback-runtime.ts
+2
View File
@@ -51,6 +51,8 @@ export function createMcpLoopbackServerConfig(port: number) {
"x-openclaw-current-inbound-audio": "${OPENCLAW_MCP_CURRENT_INBOUND_AUDIO}",
"x-openclaw-inbound-event-kind": "${OPENCLAW_MCP_INBOUND_EVENT_KIND}",
"x-openclaw-source-reply-delivery-mode": "${OPENCLAW_MCP_SOURCE_REPLY_DELIVERY_MODE}",
"x-openclaw-require-explicit-message-target":
"${OPENCLAW_MCP_REQUIRE_EXPLICIT_MESSAGE_TARGET}",
},
},
},
src/gateway/mcp-http.request.ts
+6 -2
View File
@@ -58,6 +58,7 @@ type McpRequestContext = {
accountId: string | undefined;
inboundEventKind: InboundEventKind | undefined;
sourceReplyDeliveryMode: SourceReplyDeliveryMode | undefined;
requireExplicitMessageTarget: boolean | undefined;
senderIsOwner: boolean | undefined;
};
@@ -78,7 +79,7 @@ function normalizeMcpSourceReplyDeliveryMode(
return trimmed === "automatic" || trimmed === "message_tool_only" ? trimmed : undefined;
}
function normalizeMcpCurrentInboundAudio(value: string | undefined): boolean | undefined {
function normalizeMcpBooleanHeader(value: string | undefined): boolean | undefined {
const trimmed = normalizeOptionalString(value);
return trimmed ? isTruthyEnvValue(trimmed) : undefined;
}
@@ -365,7 +366,7 @@ export function resolveMcpRequestContext(
currentChannelId: normalizeOptionalString(getHeader(req, "x-openclaw-current-channel-id")),
currentThreadTs: normalizeOptionalString(getHeader(req, "x-openclaw-current-thread-ts")),
currentMessageId: normalizeOptionalString(getHeader(req, "x-openclaw-current-message-id")),
currentInboundAudio: normalizeMcpCurrentInboundAudio(
currentInboundAudio: normalizeMcpBooleanHeader(
getHeader(req, "x-openclaw-current-inbound-audio"),
),
accountId: normalizeOptionalString(getHeader(req, "x-openclaw-account-id")),
@@ -373,6 +374,9 @@ export function resolveMcpRequestContext(
sourceReplyDeliveryMode: normalizeMcpSourceReplyDeliveryMode(
getHeader(req, "x-openclaw-source-reply-delivery-mode"),
),
requireExplicitMessageTarget: normalizeMcpBooleanHeader(
getHeader(req, "x-openclaw-require-explicit-message-target"),
),
senderIsOwner: auth.senderIsOwner,
};
}
src/gateway/mcp-http.runtime.ts
+2
View File
@@ -36,6 +36,7 @@ type McpLoopbackScopeParams = {
accountId: string | undefined;
inboundEventKind: InboundEventKind | undefined;
sourceReplyDeliveryMode: SourceReplyDeliveryMode | undefined;
requireExplicitMessageTarget?: boolean;
senderIsOwner: boolean | undefined;
};
@@ -70,6 +71,7 @@ export class McpLoopbackToolCache {
params.accountId ?? "",
params.inboundEventKind ?? "",
params.sourceReplyDeliveryMode ?? "",
params.requireExplicitMessageTarget === true ? "explicit-message-target" : "",
params.senderIsOwner === true
? "owner"
: params.senderIsOwner === false
src/gateway/mcp-http.test.ts
+14 -2
View File
@@ -31,6 +31,7 @@ type ScopedToolsCall = {
currentInboundAudio?: boolean;
inboundEventKind?: string;
sourceReplyDeliveryMode?: string;
requireExplicitMessageTarget?: boolean;
senderIsOwner?: boolean;
surface?: string;
excludeToolNames?: Iterable<string>;
@@ -680,6 +681,7 @@ describe("mcp loopback server", () => {
"x-openclaw-current-inbound-audio": "true",
"x-openclaw-inbound-event-kind": "room_event",
"x-openclaw-source-reply-delivery-mode": "message_tool_only",
"x-openclaw-require-explicit-message-target": "true",
}),
body: mcpToolsListBody(),
});
@@ -695,6 +697,7 @@ describe("mcp loopback server", () => {
expect(call.currentInboundAudio).toBe(true);
expect(call.inboundEventKind).toBe("room_event");
expect(call.sourceReplyDeliveryMode).toBe("message_tool_only");
expect(call.requireExplicitMessageTarget).toBe(true);
expect(call.surface).toBe("loopback");
expect(Array.from(call.excludeToolNames ?? [])).toEqual([
"read",
@@ -706,12 +709,13 @@ describe("mcp loopback server", () => {
]);
});
it("keeps loopback tool cache entries separate by inbound event kind, delivery mode, and inbound audio", async () => {
it("keeps loopback tool cache entries separate by inbound event, delivery, audio, and target policy", async () => {
const { runtime } = await startLoopbackServerForTest();
const sendToolsList = async (
inboundEventKind: string,
sourceReplyDeliveryMode?: string,
currentInboundAudio?: boolean,
requireExplicitMessageTarget?: boolean,
) =>
await sendLoopbackToolsList({
token: runtime?.ownerToken,
@@ -723,6 +727,9 @@ describe("mcp loopback server", () => {
? { "x-openclaw-source-reply-delivery-mode": sourceReplyDeliveryMode }
: {}),
...(currentInboundAudio ? { "x-openclaw-current-inbound-audio": "true" } : {}),
...(requireExplicitMessageTarget
? { "x-openclaw-require-explicit-message-target": "true" }
: {}),
},
});
@@ -730,12 +737,14 @@ describe("mcp loopback server", () => {
expect((await sendToolsList("room_event")).status).toBe(200);
expect((await sendToolsList("room_event", "message_tool_only")).status).toBe(200);
expect((await sendToolsList("room_event", "message_tool_only", true)).status).toBe(200);
expect((await sendToolsList("room_event", "message_tool_only", true, true)).status).toBe(200);
expect(resolveGatewayScopedToolsMock).toHaveBeenCalledTimes(4);
expect(resolveGatewayScopedToolsMock).toHaveBeenCalledTimes(5);
expect(getScopedToolsCall(0).inboundEventKind).toBe("user_request");
expect(getScopedToolsCall(1).inboundEventKind).toBe("room_event");
expect(getScopedToolsCall(2).sourceReplyDeliveryMode).toBe("message_tool_only");
expect(getScopedToolsCall(3).currentInboundAudio).toBe(true);
expect(getScopedToolsCall(4).requireExplicitMessageTarget).toBe(true);
});
it("keeps explicit non-owner and unknown-owner loopback cache entries separate", () => {
@@ -1277,6 +1286,9 @@ describe("createMcpLoopbackServerConfig", () => {
expect(config.mcpServers?.openclaw?.headers?.["x-openclaw-source-reply-delivery-mode"]).toBe(
"${OPENCLAW_MCP_SOURCE_REPLY_DELIVERY_MODE}",
);
expect(
config.mcpServers?.openclaw?.headers?.["x-openclaw-require-explicit-message-target"],
).toBe("${OPENCLAW_MCP_REQUIRE_EXPLICIT_MESSAGE_TARGET}");
expect(config.mcpServers?.openclaw?.headers).not.toHaveProperty("x-openclaw-sender-is-owner");
});
src/gateway/mcp-http.ts
+1
View File
@@ -199,6 +199,7 @@ export async function startMcpLoopbackServer(port = 0): Promise<{
accountId: requestContext.accountId,
inboundEventKind: requestContext.inboundEventKind,
sourceReplyDeliveryMode: requestContext.sourceReplyDeliveryMode,
requireExplicitMessageTarget: requestContext.requireExplicitMessageTarget,
senderIsOwner: requestContext.senderIsOwner,
});
src/gateway/tool-resolution.ts
+2
View File
@@ -48,6 +48,7 @@ export function resolveGatewayScopedTools(params: {
accountId?: string;
inboundEventKind?: InboundEventKind;
sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
requireExplicitMessageTarget?: boolean;
agentTo?: string;
agentThreadId?: string;
senderIsOwner?: boolean;
@@ -168,6 +169,7 @@ export function resolveGatewayScopedTools(params: {
currentThreadTs: params.currentThreadTs ?? params.agentThreadId,
currentMessageId: params.currentMessageId,
currentInboundAudio: params.currentInboundAudio,
requireExplicitMessageTarget: params.requireExplicitMessageTarget,
senderIsOwner: params.senderIsOwner,
allowGatewaySubagentBinding: params.allowGatewaySubagentBinding,
allowMediaInvokeCommands: params.allowMediaInvokeCommands,