diff --git a/src/gateway/server/hooks.agent-trust.test.ts b/src/gateway/server/hooks.agent-trust.test.ts index 1493f62bdef4..f1b51952b57e 100644 --- a/src/gateway/server/hooks.agent-trust.test.ts +++ b/src/gateway/server/hooks.agent-trust.test.ts @@ -95,10 +95,22 @@ function buildAgentPayload(name: string, agentId?: string) { } function dispatchAgentHook(payload: unknown): unknown { + return resolveDispatchAgentHook()(payload); +} + +function resolveDispatchAgentHook(): (...args: unknown[]) => unknown { if (!capturedDispatchAgentHook) { throw new Error("dispatchAgentHook missing"); } - return capturedDispatchAgentHook(payload); + return capturedDispatchAgentHook; +} + +function createDeferred() { + let resolve!: () => void; + const promise = new Promise((innerResolve) => { + resolve = innerResolve; + }); + return { promise, resolve }; } type HookLogMeta = { @@ -139,6 +151,7 @@ describe("dispatchAgentHook trust handling", () => { beforeEach(() => { resetGatewayWorkAdmission(); vi.clearAllMocks(); + loadConfigMock.mockImplementation(() => ({})); capturedDispatchAgentHook = undefined; createGatewayHooksRequestHandler(buildMinimalParams()); }); @@ -180,6 +193,167 @@ describe("dispatchAgentHook trust handling", () => { expect(getActiveGatewayRootWorkCount()).toBe(0); }); + it("serializes canonical aliases for the same session in dispatch order", async () => { + const dispatch = resolveDispatchAgentHook(); + const firstGate = createDeferred(); + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + await firstGate.promise; + return { status: "ok", summary: "first done", delivered: false }; + }); + runCronIsolatedAgentTurnMock.mockResolvedValueOnce({ + status: "ok", + summary: "second done", + delivered: false, + }); + + dispatch({ + ...buildAgentPayload("First"), + message: "first", + sessionKey: "main", + }); + dispatch({ + ...buildAgentPayload("Second"), + message: "second", + sessionKey: "agent:main:main", + }); + + await waitForFast(() => expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(1)); + expect(runCronIsolatedAgentTurnMock.mock.calls[0]?.[0]).toMatchObject({ + message: "first", + sessionKey: "main", + }); + + firstGate.resolve(); + + await waitForFast(() => expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(2)); + expect(runCronIsolatedAgentTurnMock.mock.calls[1]?.[0]).toMatchObject({ + message: "second", + sessionKey: "agent:main:main", + }); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + }); + + it("runs different sessions in parallel", async () => { + const dispatch = resolveDispatchAgentHook(); + const firstGate = createDeferred(); + const secondGate = createDeferred(); + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + await firstGate.promise; + return { status: "ok", summary: "first done", delivered: false }; + }); + + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + await secondGate.promise; + return { status: "ok", summary: "second done", delivered: false }; + }); + + dispatch({ + ...buildAgentPayload("First"), + message: "first", + sessionKey: "agent:main:session-a", + }); + dispatch({ + ...buildAgentPayload("Second"), + message: "second", + sessionKey: "agent:main:session-b", + }); + + expect(getActiveGatewayRootWorkCount()).toBe(2); + + try { + await waitForFast(() => expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(2)); + } finally { + firstGate.resolve(); + secondGate.resolve(); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + } + }); + + it("uses fresh config when a queued hook starts after reload", async () => { + const dispatch = resolveDispatchAgentHook(); + let currentConfig: { session?: { mainKey?: string } } = {}; + loadConfigMock.mockImplementation(() => currentConfig); + const firstGate = createDeferred(); + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + await firstGate.promise; + return { status: "ok", summary: "first done", delivered: false }; + }); + runCronIsolatedAgentTurnMock.mockResolvedValueOnce({ + status: "ok", + summary: "second done", + delivered: false, + }); + + dispatch({ ...buildAgentPayload("First"), message: "first", sessionKey: "main" }); + dispatch({ ...buildAgentPayload("Second"), message: "second", sessionKey: "main" }); + await waitForFast(() => expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(1)); + + currentConfig = { session: { mainKey: "reloaded" } }; + firstGate.resolve(); + + await waitForFast(() => expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(2)); + expect(runCronIsolatedAgentTurnMock.mock.calls[1]?.[0]).toMatchObject({ + agentId: "main", + cfg: currentConfig, + message: "second", + sessionKey: "main", + }); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + }); + + it("continues a same-session hook queue after a failed run", async () => { + const dispatch = resolveDispatchAgentHook(); + runCronIsolatedAgentTurnMock.mockRejectedValueOnce(new Error("agent exploded")); + runCronIsolatedAgentTurnMock.mockResolvedValueOnce({ + status: "ok", + summary: "second done", + delivered: false, + }); + + dispatch({ + ...buildAgentPayload("First"), + message: "first", + sessionKey: "shared-session", + }); + dispatch({ + ...buildAgentPayload("Second"), + message: "second", + sessionKey: "shared-session", + }); + + await waitForFast(() => + expect(enqueueSystemEventMock).toHaveBeenCalledWith( + "Hook First (error): Error: agent exploded", + { + sessionKey: "agent:main:main", + }, + ), + ); + await waitForFast(() => expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(2)); + expect(runCronIsolatedAgentTurnMock.mock.calls[1]?.[0]).toMatchObject({ + message: "second", + sessionKey: "shared-session", + }); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + }); + + it("reports runtime-config failures after returning a run id", async () => { + loadConfigMock.mockImplementationOnce(() => { + throw new Error("config exploded"); + }); + + const runId = dispatchAgentHook(buildAgentPayload("Config")); + + expect(runId).toEqual(expect.any(String)); + await waitForFast(() => + expect(enqueueSystemEventMock).toHaveBeenCalledWith( + "Hook Config (error): Error: config exploded", + { sessionKey: "main-session" }, + ), + ); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + }); + it("does not announce successful deliver:false hook results", async () => { runCronIsolatedAgentTurnMock.mockResolvedValueOnce({ status: "ok", diff --git a/src/gateway/server/hooks.ts b/src/gateway/server/hooks.ts index 107134143878..564080eb1c22 100644 --- a/src/gateway/server/hooks.ts +++ b/src/gateway/server/hooks.ts @@ -6,6 +6,7 @@ import { } from "@openclaw/normalization-core/number-coercion"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; +import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; import { sanitizeInboundSystemTags } from "../../auto-reply/reply/inbound-text.js"; import type { CliDeps } from "../../cli/deps.types.js"; import { getRuntimeConfig } from "../../config/io.js"; @@ -16,6 +17,7 @@ import { } from "../../config/sessions.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { RunCronAgentTurnResult } from "../../cron/isolated-agent/run.types.js"; +import { resolveCronAgentSessionKey } from "../../cron/isolated-agent/session-key.js"; import type { CronJob } from "../../cron/types.js"; import { requestHeartbeat } from "../../infra/heartbeat-wake.js"; import { enqueueSystemEvent } from "../../infra/system-events.js"; @@ -93,6 +95,28 @@ function formatHookRunWarningConsoleMessage(params: { return parts.join(" "); } +function createSessionKeyedHookDispatchQueue() { + const hookAgentDispatchTails = new Map>(); + + return (sessionKey: string, operation: () => Promise) => { + const previousTail = hookAgentDispatchTails.get(sessionKey); + const run = previousTail ? previousTail.catch(() => undefined).then(operation) : operation(); + const tail = run.then( + () => undefined, + () => undefined, + ); + hookAgentDispatchTails.set(sessionKey, tail); + // Same-session hook agent runs append to one agent session. Serializing avoids + // optimistic lifecycle-claim races while preserving parallelism across sessions. + void tail.finally(() => { + if (hookAgentDispatchTails.get(sessionKey) === tail) { + hookAgentDispatchTails.delete(sessionKey); + } + }); + return run; + }; +} + /** Creates the HTTP handler used by gateway hook endpoints. */ export function createGatewayHooksRequestHandler(params: { deps: CliDeps; @@ -103,6 +127,12 @@ export function createGatewayHooksRequestHandler(params: { logHooks: SubsystemLogger; }) { const { deps, getHooksConfig, getClientIpConfig, bindHost, port, logHooks } = params; + const enqueueHookAgentDispatch = createSessionKeyedHookDispatchQueue(); + let isolatedAgentModulePromise: + | Promise + | undefined; + const loadIsolatedAgentModule = () => + (isolatedAgentModulePromise ??= import("../../cron/isolated-agent.js")); const dispatchWakeHook = (value: { text: string; mode: "now" | "next-heartbeat" }) => { const sessionKey = resolveMainSessionKeyFromConfig(); @@ -149,81 +179,108 @@ export function createGatewayHooksRequestHandler(params: { delivery, state: { nextRunAtMs: nowMs }, }; - let hookEventSessionKey: string | undefined; - void runWithGatewayIndependentRootWorkContinuation(async () => { - try { - // Agent hooks run after the HTTP response path has returned, so failure - // handling must record a system event instead of throwing to the caller. - const cfg = getRuntimeConfig(); - hookEventSessionKey = resolveHookEventSessionKey({ - cfg, - agentId: value.agentId, + const reportHookFailure = (err: unknown) => { + logHooks.warn(`hook agent failed: ${String(err)}`); + enqueueSystemEvent(`Hook ${safeName} (error): ${String(err)}`, { + sessionKey: hookEventSessionKey ?? resolveMainSessionKeyFromConfig(), + }); + if (value.wakeMode === "now") { + requestHeartbeat({ + source: "hook", + intent: "immediate", + reason: `hook:${jobId}:error`, }); - const { runCronIsolatedAgentTurn } = await import("../../cron/isolated-agent.js"); - const result = await runCronIsolatedAgentTurn({ - cfg, - deps, - job, - message: value.message, - sessionKey, - lane: "cron", - }); - const summary = resolveHookRunSummary(result); - const prefix = - result.status === "ok" ? `Hook ${safeName}` : `Hook ${safeName} (${result.status})`; - const shouldAnnounce = shouldAnnounceHookRunResult({ deliver: value.deliver, result }); - if (result.status !== "ok") { - logHooks.warn("hook agent run returned non-ok status", { - sourcePath: value.sourcePath, - name: safeName, - runId, - jobId, + } + }; + let dispatchCfg: OpenClawConfig; + try { + dispatchCfg = getRuntimeConfig(); + } catch (err) { + // Config resolution historically failed after the hook response returned. + // Preserve that detached failure contract while queue keys stay canonical. + void runWithGatewayIndependentRootWorkContinuation(async () => reportHookFailure(err)); + return runId; + } + const agentId = value.agentId ?? resolveDefaultAgentId(dispatchCfg); + const queueKey = resolveCronAgentSessionKey({ + sessionKey, + agentId, + mainKey: dispatchCfg.session?.mainKey, + cfg: dispatchCfg, + }); + // Queue identity is fixed when accepted; the isolated runner still receives + // the original session expression and fresh config, preserving hook routing. + void runWithGatewayIndependentRootWorkContinuation(() => + enqueueHookAgentDispatch(queueKey, async () => { + try { + // Agent hooks run after the HTTP response path has returned, so failure + // handling must record a system event instead of throwing to the caller. + const cfg = getRuntimeConfig(); + // Keep an omitted agent omitted for event routing so global session scope + // stays global; runner identity is frozen separately via accepted agentId. + hookEventSessionKey = resolveHookEventSessionKey({ + cfg, agentId: value.agentId, + }); + const { runCronIsolatedAgentTurn } = await loadIsolatedAgentModule(); + const result = await runCronIsolatedAgentTurn({ + cfg, + deps, + job, + message: value.message, sessionKey, - status: result.status, - model: value.model, - summary, - consoleMessage: formatHookRunWarningConsoleMessage({ + // Isolated runs derive their lifecycle key from random jobId (or an + // already-stable cron: key), so accepted agentId closes reload drift. + agentId, + lane: "cron", + }); + const summary = resolveHookRunSummary(result); + const prefix = + result.status === "ok" ? `Hook ${safeName}` : `Hook ${safeName} (${result.status})`; + const shouldAnnounce = shouldAnnounceHookRunResult({ deliver: value.deliver, result }); + if (result.status !== "ok") { + logHooks.warn("hook agent run returned non-ok status", { + sourcePath: value.sourcePath, + name: safeName, + runId, + jobId, + agentId: value.agentId, + sessionKey, status: result.status, model: value.model, summary, - }), - }); - } - if (shouldAnnounce) { - const eventSessionKey = hookEventSessionKey ?? resolveMainSessionKeyFromConfig(); - enqueueSystemEvent(`${prefix}: ${summary}`.trim(), { - sessionKey: eventSessionKey, - }); - if (value.wakeMode === "now") { - requestHeartbeat({ source: "hook", intent: "immediate", reason: `hook:${jobId}` }); + consoleMessage: formatHookRunWarningConsoleMessage({ + status: result.status, + model: value.model, + summary, + }), + }); } - } else if (result.status === "ok" && !value.deliver) { - logHooks.info("hook agent run completed without announcement", { - sourcePath: value.sourcePath, - name: safeName, - runId, - jobId, - agentId: value.agentId, - sessionKey, - completedAt: new Date().toISOString(), - }); + if (shouldAnnounce) { + const eventSessionKey = hookEventSessionKey ?? resolveMainSessionKeyFromConfig(); + enqueueSystemEvent(`${prefix}: ${summary}`.trim(), { + sessionKey: eventSessionKey, + }); + if (value.wakeMode === "now") { + requestHeartbeat({ source: "hook", intent: "immediate", reason: `hook:${jobId}` }); + } + } else if (result.status === "ok" && !value.deliver) { + logHooks.info("hook agent run completed without announcement", { + sourcePath: value.sourcePath, + name: safeName, + runId, + jobId, + agentId: value.agentId, + sessionKey, + completedAt: new Date().toISOString(), + }); + } + } catch (err) { + reportHookFailure(err); } - } catch (err) { - logHooks.warn(`hook agent failed: ${String(err)}`); - enqueueSystemEvent(`Hook ${safeName} (error): ${String(err)}`, { - sessionKey: hookEventSessionKey ?? resolveMainSessionKeyFromConfig(), - }); - if (value.wakeMode === "now") { - requestHeartbeat({ - source: "hook", - intent: "immediate", - reason: `hook:${jobId}:error`, - }); - } - } - }); + }), + ); return runId; };