diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 85d46d9c7e46..669948de45fc 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -542,9 +542,15 @@ Query-string tokens are rejected. - Setting `deliver: false` keeps the run completion-only and ignores any delivery destination. - Supplying both a concrete `channel` and `to` enables direct announce delivery. + The HTTP response waits only for runner admission, not for the agent turn to finish. A `200` may take up to 15 seconds and means the run entered its agent runner. Pre-run failures return `{ ok: false, error, runId }` with: + + - `409` when the target session changed or otherwise rejects new work; retry after resolving the session conflict. + - `502` when Gateway or cron preparation fails before runner entry. + - `503` when runner admission does not complete within 15 seconds. Timed-out queued work is canceled and does not start later. + - Custom hook names resolve via `hooks.mappings` in config. Mappings can transform arbitrary payloads into `wake` or `agent` actions with templates or code transforms. + Custom hook names resolve via `hooks.mappings` in config. Mappings can transform arbitrary payloads into `wake` or `agent` actions with templates or code transforms. Mapped `agent` actions use the same 15-second admission and `200`/`409`/`502`/`503` response contract as `POST /hooks/agent`. diff --git a/docs/gateway/configuration-reference.md b/docs/gateway/configuration-reference.md index 27b0e28f6fab..9b91d2a944fe 100644 --- a/docs/gateway/configuration-reference.md +++ b/docs/gateway/configuration-reference.md @@ -931,8 +931,11 @@ Validation and safety notes: - `sessionKey` from request payload is accepted only when `hooks.allowRequestSessionKey=true` (default: `false`). - Direct announce delivery requires both a concrete `channel` and `to`; supplying only one fails before the run is scheduled. - Omit both delivery fields for completion-only hooks, or set `deliver: false` to ignore supplied destination data. + - The request waits up to 15 seconds for runner admission, not run completion. `200` means the agent runner was entered. + - Pre-run failures return `{ ok: false, error, runId }`: `409` for session admission conflicts, `502` for other preparation failures, and `503` when the 15-second admission deadline expires. Timed-out queued work is canceled and will not start later. - `POST /hooks/` → resolved via `hooks.mappings` - Template-rendered mapping `sessionKey` values are treated as externally supplied and also require `hooks.allowRequestSessionKey=true`. + - Mapped `agent` actions use the same admission wait and `200`/`409`/`502`/`503` outcomes. diff --git a/src/cron/isolated-agent/run-prepare.ts b/src/cron/isolated-agent/run-prepare.ts index 5c172b1524c3..363792a0f87e 100644 --- a/src/cron/isolated-agent/run-prepare.ts +++ b/src/cron/isolated-agent/run-prepare.ts @@ -276,11 +276,11 @@ export async function prepareCronRunContext(params: { ) : Boolean(currentEntry); if (changed) { - throw new Error(`Session "${agentSessionKey}" changed while starting work. Retry.`); + throw new CronSessionLifecycleClaimError(agentSessionKey); } const archivedSessionError = resolveSessionWorkStartError(agentSessionKey, currentEntry); if (archivedSessionError) { - throw new Error(archivedSessionError); + throw new CronSessionLifecycleClaimError(agentSessionKey, archivedSessionError); } }, }); diff --git a/src/cron/isolated-agent/run-session-state.ts b/src/cron/isolated-agent/run-session-state.ts index cebc80a3f91c..28c259fee070 100644 --- a/src/cron/isolated-agent/run-session-state.ts +++ b/src/cron/isolated-agent/run-session-state.ts @@ -51,8 +51,13 @@ export type CronRunContinuationSession = { }; export class CronSessionLifecycleClaimError extends Error { - constructor(sessionKey: string) { - super(`Session "${sessionKey}" changed while starting work. Retry.`); + readonly admissionDisposition = "session-conflict" as const; + + constructor( + sessionKey: string, + message = `Session "${sessionKey}" changed while starting work. Retry.`, + ) { + super(message); this.name = "CronSessionLifecycleClaimError"; } } diff --git a/src/cron/isolated-agent/run.session-lifecycle.test.ts b/src/cron/isolated-agent/run.session-lifecycle.test.ts index ef292d4e96a8..890ee2ed2461 100644 --- a/src/cron/isolated-agent/run.session-lifecycle.test.ts +++ b/src/cron/isolated-agent/run.session-lifecycle.test.ts @@ -69,9 +69,13 @@ describe("runCronIsolatedAgentTurn session lifecycle", () => { ...initialSessionEntry, sessionId: "session-after-setup", }); - await expect(runCronIsolatedAgentTurn(makePersistentCronParams(sessionKey))).rejects.toThrow( - `Session "${sessionKey}" changed while starting work. Retry.`, - ); + await expect( + runCronIsolatedAgentTurn(makePersistentCronParams(sessionKey)), + ).resolves.toMatchObject({ + status: "error", + error: `Session "${sessionKey}" changed while starting work. Retry.`, + admissionDisposition: "session-conflict", + }); expect(preflightCronModelProviderMock).not.toHaveBeenCalled(); expect(runEmbeddedAgentMock).not.toHaveBeenCalled(); }); @@ -183,9 +187,13 @@ describe("runCronIsolatedAgentTurn session lifecycle", () => { ); loadSessionEntryMock.mockReturnValue(undefined); - await expect(runCronIsolatedAgentTurn(makePersistentCronParams(sessionKey))).rejects.toThrow( - `Session "${sessionKey}" changed while starting work. Retry.`, - ); + await expect( + runCronIsolatedAgentTurn(makePersistentCronParams(sessionKey)), + ).resolves.toMatchObject({ + status: "error", + error: `Session "${sessionKey}" changed while starting work. Retry.`, + admissionDisposition: "session-conflict", + }); expect(runEmbeddedAgentMock).not.toHaveBeenCalled(); }); diff --git a/src/cron/isolated-agent/run.ts b/src/cron/isolated-agent/run.ts index 668a85f01d60..ee83f82db14c 100644 --- a/src/cron/isolated-agent/run.ts +++ b/src/cron/isolated-agent/run.ts @@ -28,7 +28,7 @@ import type { } from "../types.js"; import { finalizeCronRun } from "./run-finalize.js"; import { prepareCronRunContext } from "./run-prepare.js"; -import type { MutableCronSession } from "./run-session-state.js"; +import { CronSessionLifecycleClaimError, type MutableCronSession } from "./run-session-state.js"; import { logWarn } from "./run.runtime.js"; import type { RunCronAgentTurnResult } from "./run.types.js"; import { cleanupCronRunSessionAfterRun } from "./session-cleanup.js"; @@ -100,13 +100,25 @@ export async function runCronIsolatedAgentTurn(params: { const abortReason = () => resolveCronAbortReasonText(abortSignal?.reason) ?? "cron: job execution timed out"; const isFastTestEnv = isFastTestRuntimeEnv(); - const prepared = await prepareCronRunContext({ - input: { ...params, abortSignal }, - isFastTestEnv, - onLifecycleInterrupt: () => lifecycleAbortController.abort(createAgentRunRestartAbortError()), - }); + let prepared: Awaited>; + try { + prepared = await prepareCronRunContext({ + input: { ...params, abortSignal }, + isFastTestEnv, + onLifecycleInterrupt: () => lifecycleAbortController.abort(createAgentRunRestartAbortError()), + }); + } catch (err) { + if (err instanceof CronSessionLifecycleClaimError) { + return { + status: "error", + error: err.message, + admissionDisposition: err.admissionDisposition, + }; + } + throw err; + } if (!prepared.ok) { - return prepared.result; + return { ...prepared.result, admissionDisposition: "rejected" }; } // Capture the stable run id before execution can rotate its persisted session. const initialSessionId = prepared.context.cronSession.sessionEntry.sessionId; @@ -271,6 +283,14 @@ export async function runCronIsolatedAgentTurn(params: { status: "error", error, executionStarted, + ...(!executionStarted + ? { + admissionDisposition: + err instanceof CronSessionLifecycleClaimError + ? err.admissionDisposition + : ("rejected" as const), + } + : {}), // Carry the already-resolved run model into the error/timeout row so // Task-run history keeps provider/model attribution instead of looking like // an un-attributed cron timeout. finalizeCronRun does the same via diff --git a/src/cron/isolated-agent/run.types.ts b/src/cron/isolated-agent/run.types.ts index b13af5535eed..6444d8a79943 100644 --- a/src/cron/isolated-agent/run.types.ts +++ b/src/cron/isolated-agent/run.types.ts @@ -6,8 +6,13 @@ import type { CronRunTelemetry, } from "../types.js"; +/** Pre-run disposition returned when isolated cron work never enters an agent runner. */ +export type CronAgentAdmissionDisposition = "session-conflict" | "rejected"; + /** Final isolated cron turn result merged into service state and run logs. */ export type RunCronAgentTurnResult = { + /** Typed pre-run rejection so callers never infer admission state from error prose. */ + admissionDisposition?: CronAgentAdmissionDisposition; /** Last non-empty agent text output (not truncated). */ outputText?: string; /** diff --git a/src/gateway/server-http.hooks-delivery.test.ts b/src/gateway/server-http.hooks-delivery.test.ts index e4a0c63249fa..e0f634c42dd0 100644 --- a/src/gateway/server-http.hooks-delivery.test.ts +++ b/src/gateway/server-http.hooks-delivery.test.ts @@ -34,7 +34,10 @@ vi.mock("./hooks.js", async () => { }); function createDeliveryHandler(params?: { mappings?: HookMappingResolved[] }) { - const dispatchAgentHook = vi.fn((_value: HookAgentDispatchPayload) => "run-1"); + const dispatchAgentHook = vi.fn((_value: HookAgentDispatchPayload) => ({ + ok: true as const, + runId: "run-1", + })); const hooksConfig = { ...createHooksConfig(), mappings: params?.mappings ?? [], diff --git a/src/gateway/server-http.hooks-request-timeout.test.ts b/src/gateway/server-http.hooks-request-timeout.test.ts index fb0d14dd6858..c43d88698ea1 100644 --- a/src/gateway/server-http.hooks-request-timeout.test.ts +++ b/src/gateway/server-http.hooks-request-timeout.test.ts @@ -38,7 +38,7 @@ describe("createHooksRequestHandler timeout status mapping", () => { test("returns 408 for request body timeout", async () => { readJsonBodyMock.mockResolvedValue({ ok: false, error: "request body timeout" }); const dispatchWakeHook = vi.fn(); - const dispatchAgentHook = vi.fn(() => "run-1"); + const dispatchAgentHook = vi.fn(() => ({ ok: true as const, runId: "run-1" })); const handler = createHooksHandler({ dispatchWakeHook, dispatchAgentHook }); const req = createHookRequest(); const { res, end } = createResponse(); @@ -52,6 +52,29 @@ describe("createHooksRequestHandler timeout status mapping", () => { expect(dispatchAgentHook).not.toHaveBeenCalled(); }); + test.each([ + [409, "session changed"], + [502, "provider preparation failed"], + [503, "hook agent run did not start before admission timeout"], + ] as const)("returns %s for typed agent admission failures", async (statusCode, error) => { + readJsonBodyMock.mockResolvedValue({ ok: true, value: { message: "Dispatch" } }); + const dispatchAgentHook = vi.fn(async () => ({ + ok: false as const, + statusCode, + error, + runId: "run-1", + })); + const handler = createHooksHandler({ dispatchAgentHook }); + const req = createHookRequest({ url: "/hooks/agent" }); + const { res, end } = createResponse(); + + const handled = await handler(req, res); + + expect(handled).toBe(true); + expect(res.statusCode).toBe(statusCode); + expect(end).toHaveBeenCalledWith(JSON.stringify({ ok: false, error, runId: "run-1" })); + }); + test("shares hook auth rate-limit bucket across ipv4 and ipv4-mapped ipv6 forms", async () => { const handler = createHooksHandler({ bindHost: "127.0.0.1" }); diff --git a/src/gateway/server-http.test-harness.ts b/src/gateway/server-http.test-harness.ts index 626b555f2104..d9bc60c15682 100644 --- a/src/gateway/server-http.test-harness.ts +++ b/src/gateway/server-http.test-harness.ts @@ -221,7 +221,7 @@ export function createHooksHandler( } as unknown as ReturnType, getClientIpConfig: options.getClientIpConfig, dispatchWakeHook: options.dispatchWakeHook ?? (() => {}), - dispatchAgentHook: options.dispatchAgentHook ?? (() => "run-1"), + dispatchAgentHook: options.dispatchAgentHook ?? (() => ({ ok: true, runId: "run-1" })), }); } diff --git a/src/gateway/server.hooks-admission.test.ts b/src/gateway/server.hooks-admission.test.ts new file mode 100644 index 000000000000..299df0352f63 --- /dev/null +++ b/src/gateway/server.hooks-admission.test.ts @@ -0,0 +1,170 @@ +/** Focused HTTP coverage for hook admission feedback and pending replay behavior. */ +import { afterEach, describe, expect, test, vi } from "vitest"; +import { resolveMainSessionKeyFromConfig } from "../config/sessions.js"; +import { drainSystemEvents } from "../infra/system-events.js"; +import { + cronIsolatedRun, + installGatewayTestHooks, + testState, + withGatewayServer, +} from "./test-helpers.js"; + +installGatewayTestHooks({ scope: "suite" }); + +await import("./server.js"); + +const HOOK_TOKEN = "hook-secret"; + +afterEach(() => { + drainSystemEvents(resolveMainSessionKeyFromConfig()); + vi.restoreAllMocks(); +}); + +async function postHook( + port: number, + path: string, + body: Record, + idempotencyKey: string, +): Promise { + return await fetch(`http://127.0.0.1:${port}${path}`, { + method: "POST", + headers: { + Authorization: `Bearer ${HOOK_TOKEN}`, + "Content-Type": "application/json", + "Idempotency-Key": idempotencyKey, + }, + body: JSON.stringify(body), + }); +} + +async function waitForCronIsolatedRuns(count: number): Promise { + await expect + .poll(() => cronIsolatedRun.mock.calls.length, { timeout: 2_000, interval: 10 }) + .toBe(count); +} + +function createDeferred() { + let resolve!: () => void; + const promise = new Promise((innerResolve) => { + resolve = innerResolve; + }); + return { promise, resolve }; +} + +async function waitForDuplicateRequest(): Promise { + await new Promise((resolve) => { + setTimeout(resolve, 25); + }); +} + +describe("gateway hook admission", () => { + test("shares one pending direct dispatch across simultaneous duplicates", async () => { + testState.hooksConfig = { enabled: true, token: HOOK_TOKEN }; + await withGatewayServer(async ({ port }) => { + const runnerAdmission = createDeferred(); + cronIsolatedRun.mockClear(); + cronIsolatedRun.mockImplementationOnce(async (params: unknown) => { + await runnerAdmission.promise; + (params as { onExecutionStarted?: () => void }).onExecutionStarted?.(); + return { status: "ok", summary: "done" }; + }); + const request = () => + postHook(port, "/hooks/agent", { message: "Dispatch" }, "pending-direct-idem"); + + const firstResponse = request(); + await waitForCronIsolatedRuns(1); + const duplicateResponse = request(); + await waitForDuplicateRequest(); + expect(cronIsolatedRun).toHaveBeenCalledTimes(1); + runnerAdmission.resolve(); + + const [first, duplicate] = await Promise.all([firstResponse, duplicateResponse]); + expect(first.status).toBe(200); + expect(duplicate.status).toBe(200); + const firstBody = (await first.json()) as { runId?: string }; + const duplicateBody = (await duplicate.json()) as { runId?: string }; + expect(duplicateBody.runId).toBe(firstBody.runId); + }); + }); + + test("shares one pending mapped dispatch across simultaneous duplicates", async () => { + testState.hooksConfig = { + enabled: true, + token: HOOK_TOKEN, + mappings: [ + { + match: { path: "mapped-pending" }, + action: "agent", + messageTemplate: "Mapped: {{payload.subject}}", + }, + ], + }; + await withGatewayServer(async ({ port }) => { + const runnerAdmission = createDeferred(); + cronIsolatedRun.mockClear(); + cronIsolatedRun.mockImplementationOnce(async (params: unknown) => { + await runnerAdmission.promise; + (params as { onExecutionStarted?: () => void }).onExecutionStarted?.(); + return { status: "ok", summary: "done" }; + }); + const request = () => + postHook(port, "/hooks/mapped-pending", { subject: "Email" }, "pending-mapped-idem"); + + const firstResponse = request(); + await waitForCronIsolatedRuns(1); + const duplicateResponse = request(); + await waitForDuplicateRequest(); + expect(cronIsolatedRun).toHaveBeenCalledTimes(1); + runnerAdmission.resolve(); + + const [first, duplicate] = await Promise.all([firstResponse, duplicateResponse]); + expect(first.status).toBe(200); + expect(duplicate.status).toBe(200); + const firstBody = (await first.json()) as { runId?: string }; + const duplicateBody = (await duplicate.json()) as { runId?: string }; + expect(duplicateBody.runId).toBe(firstBody.runId); + }); + }); + + test("returns typed admission failures and leaves the idempotency key retryable", async () => { + testState.hooksConfig = { enabled: true, token: HOOK_TOKEN }; + await withGatewayServer(async ({ port }) => { + cronIsolatedRun.mockClear(); + cronIsolatedRun + .mockResolvedValueOnce({ + status: "error", + error: "session changed", + admissionDisposition: "session-conflict", + }) + .mockResolvedValueOnce({ + status: "error", + error: "provider preparation failed", + admissionDisposition: "rejected", + }) + .mockImplementationOnce(async (params: unknown) => { + (params as { onExecutionStarted?: () => void }).onExecutionStarted?.(); + return { status: "ok", summary: "done" }; + }); + const request = () => + postHook(port, "/hooks/agent", { message: "Dispatch" }, "admission-retry"); + + const conflict = await request(); + expect(conflict.status).toBe(409); + const conflictBody = (await conflict.json()) as { ok?: boolean; runId?: string }; + expect(conflictBody.ok).toBe(false); + + const gatewayFailure = await request(); + expect(gatewayFailure.status).toBe(502); + const gatewayFailureBody = (await gatewayFailure.json()) as { + ok?: boolean; + runId?: string; + }; + expect(gatewayFailureBody.ok).toBe(false); + expect(gatewayFailureBody.runId).not.toBe(conflictBody.runId); + + const admitted = await request(); + expect(admitted.status).toBe(200); + expect(cronIsolatedRun).toHaveBeenCalledTimes(3); + }); + }); +}); diff --git a/src/gateway/server.hooks.test.ts b/src/gateway/server.hooks.test.ts index 98bacc668408..3c28b337e2dd 100644 --- a/src/gateway/server.hooks.test.ts +++ b/src/gateway/server.hooks.test.ts @@ -87,6 +87,17 @@ function mockIsolatedRunOk(): void { }); } +function mockIsolatedRunAfterStartOnce(result: { + status: "ok" | "error" | "skipped"; + summary: string; + delivered?: boolean; +}) { + cronIsolatedRun.mockImplementationOnce(async (params: unknown) => { + (params as { onExecutionStarted?: () => void }).onExecutionStarted?.(); + return result; + }); +} + async function waitForCronIsolatedRuns(count: number, timeoutMs = 2_000): Promise { await expect .poll(() => cronIsolatedRun.mock.calls.length, { timeout: timeoutMs, interval: 10 }) @@ -422,7 +433,7 @@ describe("gateway server hooks", () => { await waitForCronIsolatedRuns(2); expect(peekSystemEventEntries(resolveMainKey())).toStrictEqual([]); - cronIsolatedRun.mockResolvedValueOnce({ + mockIsolatedRunAfterStartOnce({ status: "error", summary: "boom", delivered: false, @@ -445,7 +456,7 @@ describe("gateway server hooks", () => { await withGatewayServer(async ({ port }) => { cronIsolatedRun.mockClear(); - cronIsolatedRun.mockResolvedValueOnce({ + mockIsolatedRunAfterStartOnce({ status: "error", summary: "boom", delivered: false, diff --git a/src/gateway/server/hooks-request-handler.ts b/src/gateway/server/hooks-request-handler.ts index 3529ae25dfb3..e0430be49dbf 100644 --- a/src/gateway/server/hooks-request-handler.ts +++ b/src/gateway/server/hooks-request-handler.ts @@ -49,9 +49,15 @@ export type HooksRequestHandler = (req: IncomingMessage, res: ServerResponse) => type HookDispatchers = { dispatchWakeHook: (value: { text: string; mode: "now" | "next-heartbeat" }) => void; - dispatchAgentHook: (value: HookAgentDispatchPayload) => string; + dispatchAgentHook: ( + value: HookAgentDispatchPayload, + ) => HookAgentDispatchResult | Promise; }; +export type HookAgentDispatchResult = + | { ok: true; runId: string } + | { ok: false; statusCode: 409 | 502 | 503; error: string; runId?: string }; + type HookReplayEntry = { ts: number; runId: string; @@ -82,6 +88,7 @@ export function createHooksRequestHandler( ): HooksRequestHandler { const { getHooksConfig, logHooks, dispatchAgentHook, dispatchWakeHook, getClientIpConfig } = opts; const hookReplayCache = new Map(); + const pendingHookReplays = new Map>(); const hookAuthLimiter = createAuthRateLimiter({ maxAttempts: HOOK_AUTH_FAILURE_LIMIT, windowMs: HOOK_AUTH_FAILURE_WINDOW_MS, @@ -162,6 +169,62 @@ export function createHooksRequestHandler( pruneHookReplayCache(now); }; + const resolveHookReplay = ( + key: string | undefined, + now: number, + ): HookAgentDispatchResult | Promise | undefined => { + if (!key) { + return undefined; + } + const cachedRunId = resolveCachedHookRunId(key, now); + if (cachedRunId) { + return { ok: true, runId: cachedRunId }; + } + return pendingHookReplays.get(key); + }; + + const dispatchAgentHookWithReplay = ( + key: string | undefined, + now: number, + dispatch: () => HookAgentDispatchResult | Promise, + ): HookAgentDispatchResult | Promise => { + if (!key) { + return dispatch(); + } + const existing = resolveHookReplay(key, now); + if (existing) { + return existing; + } + const pending = Promise.resolve() + .then(dispatch) + .then((result) => { + if (result.ok) { + rememberHookRunId(key, result.runId, now); + } + return result; + }) + .finally(() => { + // Failed admission stays retryable; identity guards against deleting a newer replay. + if (pendingHookReplays.get(key) === pending) { + pendingHookReplays.delete(key); + } + }); + pendingHookReplays.set(key, pending); + return pending; + }; + + const sendAgentDispatchResult = (res: ServerResponse, result: HookAgentDispatchResult) => { + if (result.ok) { + sendJson(res, 200, { ok: true, runId: result.runId }); + return; + } + sendJson(res, result.statusCode, { + ok: false, + error: result.error, + ...(result.runId ? { runId: result.runId } : {}), + }); + }; + return async (req, res) => { const hooksConfig = getHooksConfig(); if (!hooksConfig) { @@ -310,9 +373,9 @@ export function createHooksRequestHandler( timeoutSeconds: normalized.value.timeoutSeconds ?? null, }, }); - const cachedRunId = resolveCachedHookRunId(replayKey, now); - if (cachedRunId) { - sendJson(res, 200, { ok: true, runId: cachedRunId }); + const replay = resolveHookReplay(replayKey, now); + if (replay) { + sendAgentDispatchResult(res, await replay); return true; } const dispatchSessionKey = resolveDispatchSessionKeyOrRespond( @@ -322,16 +385,17 @@ export function createHooksRequestHandler( if (dispatchSessionKey === null) { return true; } - const runId = dispatchAgentHook({ - ...normalized.value, - idempotencyKey, - sessionKey: dispatchSessionKey, - sourcePath: `${basePath}/agent`, - agentId: targetAgentId, - externalContentSource: "webhook", - }); - rememberHookRunId(replayKey, runId, now); - sendJson(res, 200, { ok: true, runId }); + const dispatched = await dispatchAgentHookWithReplay(replayKey, now, () => + dispatchAgentHook({ + ...normalized.value, + idempotencyKey, + sessionKey: dispatchSessionKey, + sourcePath: `${basePath}/agent`, + agentId: targetAgentId, + externalContentSource: "webhook", + }), + ); + sendAgentDispatchResult(res, dispatched); return true; } @@ -361,37 +425,37 @@ export function createHooksRequestHandler( sendJson(res, 200, { ok: true, mode: mapped.action.mode }); return true; } - const channel = resolveHookChannel(mapped.action.channel); + const action = mapped.action; + const channel = resolveHookChannel(action.channel); if (!channel) { sendJson(res, 400, { ok: false, error: getHookChannelError() }); return true; } - const deliver = resolveHookDeliver(mapped.action.deliver); + const deliver = resolveHookDeliver(action.deliver); const delivery = deliver ? { mode: "announce" as const, channel, - to: mapped.action.to, + to: action.to, } : { mode: "none" as const }; - if (!isHookAgentAllowed(hooksConfig, mapped.action.agentId)) { + if (!isHookAgentAllowed(hooksConfig, action.agentId)) { sendJson(res, 400, { ok: false, error: getHookAgentPolicyError() }); return true; } const sessionKey = resolveHookSessionKey({ hooksConfig, - source: - mapped.action.sessionKeySource === "static" ? "mapping-static" : "mapping-templated", - sessionKey: mapped.action.sessionKey, + source: action.sessionKeySource === "static" ? "mapping-static" : "mapping-templated", + sessionKey: action.sessionKey, }); if (!sessionKey.ok) { sendJson(res, 400, { ok: false, error: sessionKey.error }); return true; } - const targetAgentId = resolveHookTargetAgentId(hooksConfig, mapped.action.agentId); + const targetAgentId = resolveHookTargetAgentId(hooksConfig, action.agentId); const effectiveTargetAgentId = resolveEffectiveHookTargetAgentId( hooksConfig, - mapped.action.agentId, + action.agentId, ); const dispatchSessionKey = resolveDispatchSessionKeyOrRespond( sessionKey.value, @@ -406,47 +470,47 @@ export function createHooksRequestHandler( idempotencyKey, dispatchScope: { agentId: effectiveTargetAgentId, - sessionKey: - mapped.action.sessionKey ?? hooksConfig.sessionPolicy.defaultSessionKey ?? null, - message: mapped.action.message, - name: mapped.action.name ?? "Hook", - wakeMode: mapped.action.wakeMode, + sessionKey: action.sessionKey ?? hooksConfig.sessionPolicy.defaultSessionKey ?? null, + message: action.message, + name: action.name ?? "Hook", + wakeMode: action.wakeMode, deliver, channel, - to: mapped.action.to ?? null, - model: mapped.action.model ?? null, - thinking: mapped.action.thinking ?? null, - timeoutSeconds: mapped.action.timeoutSeconds ?? null, + to: action.to ?? null, + model: action.model ?? null, + thinking: action.thinking ?? null, + timeoutSeconds: action.timeoutSeconds ?? null, }, }); - const cachedRunId = resolveCachedHookRunId(replayKey, now); - if (cachedRunId) { - sendJson(res, 200, { ok: true, runId: cachedRunId }); + const replay = resolveHookReplay(replayKey, now); + if (replay) { + sendAgentDispatchResult(res, await replay); return true; } - const runId = dispatchAgentHook({ - message: mapped.action.message, - name: mapped.action.name ?? "Hook", - idempotencyKey, - agentId: targetAgentId, - wakeMode: mapped.action.wakeMode, - sessionKey: dispatchSessionKey, - sourcePath: `${basePath}/${subPath}`, - deliver, - channel, - to: mapped.action.to, - delivery, - model: mapped.action.model, - thinking: mapped.action.thinking, - timeoutSeconds: mapped.action.timeoutSeconds, - allowUnsafeExternalContent: mapped.action.allowUnsafeExternalContent, - externalContentSource: resolveMappedHookExternalContentSource({ - subPath, - sessionKey: sessionKey.value, + const dispatched = await dispatchAgentHookWithReplay(replayKey, now, () => + dispatchAgentHook({ + message: action.message, + name: action.name ?? "Hook", + idempotencyKey, + agentId: targetAgentId, + wakeMode: action.wakeMode, + sessionKey: dispatchSessionKey, + sourcePath: `${basePath}/${subPath}`, + deliver, + channel, + to: action.to, + delivery, + model: action.model, + thinking: action.thinking, + timeoutSeconds: action.timeoutSeconds, + allowUnsafeExternalContent: action.allowUnsafeExternalContent, + externalContentSource: resolveMappedHookExternalContentSource({ + subPath, + sessionKey: sessionKey.value, + }), }), - }); - rememberHookRunId(replayKey, runId, now); - sendJson(res, 200, { ok: true, runId }); + ); + sendAgentDispatchResult(res, dispatched); return true; } } catch (err) { diff --git a/src/gateway/server/hooks.agent-trust.test.ts b/src/gateway/server/hooks.agent-trust.test.ts index 7c4b1ac0424a..f674b8ccb060 100644 --- a/src/gateway/server/hooks.agent-trust.test.ts +++ b/src/gateway/server/hooks.agent-trust.test.ts @@ -62,7 +62,7 @@ function waitForFast( return vi.waitFor(callback, { interval: 1, ...options }); } -function buildMinimalParams() { +function buildMinimalParams(overrides: { agentStartAdmissionTimeoutMs?: number } = {}) { return { deps: {} as never, getHooksConfig: () => null, @@ -75,6 +75,7 @@ function buildMinimalParams() { info: logHooksInfoMock, error: vi.fn(), } as never, + ...overrides, }; } @@ -369,14 +370,19 @@ describe("dispatchAgentHook trust handling", () => { await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); }); - it("reports runtime-config failures after returning a run id", async () => { + it("reports runtime-config failures as failed admission", async () => { loadConfigMock.mockImplementationOnce(() => { throw new Error("config exploded"); }); - const runId = dispatchAgentHook(buildAgentPayload("Config")); + const result = await dispatchAgentHook(buildAgentPayload("Config")); - expect(runId).toEqual(expect.any(String)); + expect(result).toMatchObject({ + ok: false, + statusCode: 502, + error: "hook agent run failed before entering the agent runner", + runId: expect.any(String), + }); await waitForFast(() => expect(enqueueSystemEventMock).toHaveBeenCalledWith( "Hook Config (error): Error: config exploded", @@ -386,6 +392,69 @@ describe("dispatchAgentHook trust handling", () => { await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); }); + it("keeps cron admission details behind stable public errors", async () => { + runCronIsolatedAgentTurnMock.mockResolvedValueOnce({ + status: "error", + error: 'Session "agent:private:canonical" changed while starting work. Retry.', + admissionDisposition: "session-conflict", + }); + + const result = await dispatchAgentHook(buildAgentPayload("Conflict")); + + expect(result).toMatchObject({ + ok: false, + statusCode: 409, + error: "hook agent run was rejected because the target session changed", + runId: expect.any(String), + }); + await waitForFast(() => + expect(enqueueSystemEventMock).toHaveBeenCalledWith( + 'Hook Conflict (error): Session "agent:private:canonical" changed while starting work. Retry.', + { sessionKey: "agent:main:main" }, + ), + ); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + }); + + it("does not start same-session work after its admission timeout", async () => { + capturedDispatchAgentHook = undefined; + createGatewayHooksRequestHandler(buildMinimalParams({ agentStartAdmissionTimeoutMs: 10 })); + const firstRunStarted = createDeferred(); + const releaseFirstRun = createDeferred(); + runCronIsolatedAgentTurnMock.mockImplementationOnce( + async (params: { onExecutionStarted?: () => void }) => { + params.onExecutionStarted?.(); + firstRunStarted.resolve(); + await releaseFirstRun.promise; + return { status: "ok", summary: "first done", delivered: false }; + }, + ); + + const firstAdmission = dispatchAgentHook({ + ...buildAgentPayload("First"), + message: "first", + sessionKey: "shared-session", + }); + await firstRunStarted.promise; + await expect(firstAdmission).resolves.toMatchObject({ ok: true }); + + const timedOutAdmission = dispatchAgentHook({ + ...buildAgentPayload("Second"), + message: "second", + sessionKey: "shared-session", + }); + await expect(timedOutAdmission).resolves.toMatchObject({ + ok: false, + statusCode: 503, + error: "hook agent run did not start before admission timeout", + }); + expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(1); + + releaseFirstRun.resolve(); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledTimes(1); + }); + 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 0a6c0475b456..7780fe2c6897 100644 --- a/src/gateway/server/hooks.ts +++ b/src/gateway/server/hooks.ts @@ -15,7 +15,10 @@ import { resolveMainSessionKeyFromConfig, } from "../../config/sessions.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; -import type { RunCronAgentTurnResult } from "../../cron/isolated-agent/run.types.js"; +import type { + CronAgentAdmissionDisposition, + 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"; @@ -23,7 +26,11 @@ import { enqueueSystemEvent } from "../../infra/system-events.js"; import type { createSubsystemLogger } from "../../logging/subsystem.js"; import { runWithGatewayIndependentRootWorkContinuation } from "../../process/gateway-work-admission.js"; import type { HookAgentDispatchPayload, HooksConfigResolved } from "../hooks.js"; -import { createHooksRequestHandler, type HookClientIpConfig } from "./hooks-request-handler.js"; +import { + createHooksRequestHandler, + type HookAgentDispatchResult, + type HookClientIpConfig, +} from "./hooks-request-handler.js"; /** * Gateway hook HTTP handler factory. @@ -32,6 +39,13 @@ import { createHooksRequestHandler, type HookClientIpConfig } from "./hooks-requ */ type SubsystemLogger = ReturnType; +const HOOK_AGENT_START_ADMISSION_TIMEOUT_MS = 15_000; +const HOOK_AGENT_START_ADMISSION_TIMEOUT_ERROR = + "hook agent run did not start before admission timeout"; +const HOOK_AGENT_SESSION_CONFLICT_ERROR = + "hook agent run was rejected because the target session changed"; +const HOOK_AGENT_PREPARATION_ERROR = "hook agent run failed before entering the agent runner"; + function resolveHookEventSessionKey(params: { cfg: OpenClawConfig; agentId?: string }): string { return params.agentId ? resolveAgentMainSessionKey({ cfg: params.cfg, agentId: params.agentId }) @@ -93,6 +107,25 @@ function formatHookRunWarningConsoleMessage(params: { return parts.join(" "); } +function createHookAdmissionFailure(params: { + runId: string; + disposition?: CronAgentAdmissionDisposition; + statusCode?: 409 | 502 | 503; +}): HookAgentDispatchResult { + const statusCode = params.statusCode ?? (params.disposition === "session-conflict" ? 409 : 502); + return { + ok: false, + statusCode, + error: + statusCode === 409 + ? HOOK_AGENT_SESSION_CONFLICT_ERROR + : statusCode === 503 + ? HOOK_AGENT_START_ADMISSION_TIMEOUT_ERROR + : HOOK_AGENT_PREPARATION_ERROR, + runId: params.runId, + }; +} + function createSessionKeyedHookDispatchQueue() { const hookAgentDispatchTails = new Map>(); @@ -123,8 +156,17 @@ export function createGatewayHooksRequestHandler(params: { bindHost: string; port: number; logHooks: SubsystemLogger; + agentStartAdmissionTimeoutMs?: number; }) { - const { deps, getHooksConfig, getClientIpConfig, bindHost, port, logHooks } = params; + const { + deps, + getHooksConfig, + getClientIpConfig, + bindHost, + port, + logHooks, + agentStartAdmissionTimeoutMs = HOOK_AGENT_START_ADMISSION_TIMEOUT_MS, + } = params; const enqueueHookAgentDispatch = createSessionKeyedHookDispatchQueue(); let isolatedAgentModulePromise: | Promise @@ -142,7 +184,9 @@ export function createGatewayHooksRequestHandler(params: { } }; - const dispatchAgentHook = (value: HookAgentDispatchPayload) => { + const dispatchAgentHook = async ( + value: HookAgentDispatchPayload, + ): Promise => { const sessionKey = value.sessionKey; // A hook name is a single-line label: it lands in logs, in cron job `name` fields, // and inside prompt-bound system-event text. Reuse the console sanitizer so control @@ -191,10 +235,8 @@ export function createGatewayHooksRequestHandler(params: { 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; + return createHookAdmissionFailure({ runId }); } const agentId = value.agentId ?? resolveDefaultAgentId(dispatchCfg); const queueKey = resolveCronAgentSessionKey({ @@ -203,13 +245,47 @@ export function createGatewayHooksRequestHandler(params: { mainKey: dispatchCfg.session?.mainKey, cfg: dispatchCfg, }); + let settleAdmission!: (result: HookAgentDispatchResult) => void; + let admissionSettled = false; + let admissionTimedOut = false; + let admissionTimer: ReturnType | undefined; + const admission = new Promise((resolve) => { + settleAdmission = (result) => { + if (admissionSettled) { + return; + } + admissionSettled = true; + if (admissionTimer) { + clearTimeout(admissionTimer); + admissionTimer = undefined; + } + resolve(result); + }; + }); + const admissionTimeoutError = new Error(HOOK_AGENT_START_ADMISSION_TIMEOUT_ERROR); + const startupAbortController = new AbortController(); + admissionTimer = setTimeout(() => { + admissionTimedOut = true; + startupAbortController.abort(admissionTimeoutError); + settleAdmission( + createHookAdmissionFailure({ + runId, + statusCode: 503, + }), + ); + }, agentStartAdmissionTimeoutMs); + admissionTimer.unref?.(); + // 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 () => { + // The admission deadline starts before this same-session queue. Expired + // work must never enter cron preparation after an HTTP 503 was returned. + if (startupAbortController.signal.aborted) { + return; + } 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. @@ -218,6 +294,11 @@ export function createGatewayHooksRequestHandler(params: { agentId: value.agentId, }); const { runCronIsolatedAgentTurn } = await loadIsolatedAgentModule(); + // Lazy module loading is the last Gateway-owned async boundary before + // cron preparation, so recheck the deadline after it settles. + if (startupAbortController.signal.aborted) { + return; + } const result = await runCronIsolatedAgentTurn({ cfg, deps, @@ -228,8 +309,28 @@ export function createGatewayHooksRequestHandler(params: { // already-stable cron: key), so accepted agentId closes reload drift. agentId, lane: "cron", + abortSignal: startupAbortController.signal, + onExecutionStarted: () => { + // Existing runner-entry callbacks are the final owner-boundary fence: + // a deadline that wins this race prevents the runner call itself. + startupAbortController.signal.throwIfAborted(); + settleAdmission({ ok: true, runId }); + }, }); + if (admissionTimedOut) { + return; + } const summary = resolveHookRunSummary(result); + if (!admissionSettled) { + settleAdmission( + result.status === "ok" || result.executionStarted === true + ? { ok: true, runId } + : createHookAdmissionFailure({ + runId, + disposition: result.admissionDisposition, + }), + ); + } const prefix = result.status === "ok" ? `Hook ${safeName}` : `Hook ${safeName} (${result.status})`; const shouldAnnounce = shouldAnnounceHookRunResult({ deliver: value.deliver, result }); @@ -271,12 +372,22 @@ export function createGatewayHooksRequestHandler(params: { }); } } catch (err) { + if (admissionTimedOut) { + return; + } + settleAdmission(createHookAdmissionFailure({ runId })); reportHookFailure(err); } }), - ); + ).catch((err: unknown) => { + if (admissionTimedOut) { + return; + } + settleAdmission(createHookAdmissionFailure({ runId })); + reportHookFailure(err); + }); - return runId; + return await admission; }; return createHooksRequestHandler({