From 9feb1db00d623ca150ac52ef4d116c63e4bb92d8 Mon Sep 17 00:00:00 2001 From: Marvinthebored Date: Fri, 21 Aug 2026 13:59:09 +0800 Subject: [PATCH] fix(gateway): restore scheduler-owned Gateway context (#126640) Bind scheduler-owned cron, hook, and heartbeat runs to lifecycle-fenced Gateway context so trusted built-in tools resolve after startup or reload without inheriting request client state. Co-authored-by: Marvinthebored Co-authored-by: Marvinthebored Co-authored-by: Ayaan Zaidi --- src/cron/service/run-admission.ts | 14 +- src/cron/service/state.ts | 2 + src/gateway/scheduled-run-gateway-context.ts | 51 ++++ src/gateway/server-cron-lazy.ts | 7 + src/gateway/server-cron.test.ts | 243 +++++++++++++++++++ src/gateway/server-cron.ts | 20 ++ src/gateway/server-lifecycle.ts | 1 + src/gateway/server-reload-contracts.ts | 4 + src/gateway/server-reload-handlers.test.ts | 13 + src/gateway/server-reload-hot.ts | 5 + src/gateway/server-reload-managed.ts | 3 + src/gateway/server-runtime-services.test.ts | 41 +++- src/gateway/server-runtime-services.ts | 19 ++ src/gateway/server-runtime-state.ts | 3 + src/gateway/server-startup-finish.ts | 1 + src/gateway/server/hooks.agent-trust.test.ts | 32 +++ src/gateway/server/hooks.ts | 63 +++-- src/plugins/runtime/gateway-request-scope.ts | 8 +- 18 files changed, 502 insertions(+), 28 deletions(-) create mode 100644 src/gateway/scheduled-run-gateway-context.ts diff --git a/src/cron/service/run-admission.ts b/src/cron/service/run-admission.ts index 7582e0003080..2f7935d839f6 100644 --- a/src/cron/service/run-admission.ts +++ b/src/cron/service/run-admission.ts @@ -647,11 +647,15 @@ export async function executeQueuedCronRun(params: { }; let outcome: TimedCronRunOutcome; try { - const result = await executeJobCoreWithTimeout(state, executionJob, { - runId: taskRunId, - activeJobMarker, - runReceipt: started.runReceipt, - }); + const execute = async () => + await executeJobCoreWithTimeout(state, executionJob, { + runId: taskRunId, + activeJobMarker, + runReceipt: started.runReceipt, + }); + const result = state.deps.runSchedulerOwned + ? await state.deps.runSchedulerOwned(execute) + : await execute(); outcome = { ...base, ...result, endedAt: state.deps.nowMs() }; } catch (error) { const receiptSettlementDisposition = diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index a52dfadf671d..244c59015ac5 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -157,6 +157,8 @@ export type CronServiceDeps = { sessionKey?: string; agentId?: string; }) => DeliveryContext | undefined; + /** Runs timer and startup work inside the owning Gateway's detached scope. */ + runSchedulerOwned?: (run: () => Promise) => Promise; requestHeartbeat: (opts: HeartbeatWakeRequest) => void; runHeartbeatOnce?: (opts?: { source?: HeartbeatWakeRequest["source"]; diff --git a/src/gateway/scheduled-run-gateway-context.ts b/src/gateway/scheduled-run-gateway-context.ts new file mode 100644 index 000000000000..06730beba959 --- /dev/null +++ b/src/gateway/scheduled-run-gateway-context.ts @@ -0,0 +1,51 @@ +/** + * Supplies a Gateway request context to scheduler-owned agent runs. + * + * Timer ticks, hook dispatch queues, and heartbeat wakeups have no Gateway + * request of their own, so trusted built-in tools (terminal, dashboard) resolve + * no context and fail mid-run. RPC-triggered runs already inherit a scope from + * their caller and must keep it. + */ +import { withPluginRuntimeGatewayContextResolver } from "../plugins/runtime/gateway-request-scope.js"; +import type { GatewayRequestContext } from "./server-methods/types.js"; + +type ScheduledGatewayContextResolver = () => GatewayRequestContext | undefined; + +/** + * Fences a raw context reference behind the owning Gateway instance lifecycle. + * + * The process-wide holder is not cleared on shutdown, so a queued run could + * otherwise resolve a retired context. The context's own `resolveGatewayContext` + * returns undefined once its instance is unavailable; prefer no context over a + * retired one, because a missing context fails visibly. + */ +export function fenceScheduledGatewayContextResolver( + resolveGatewayContext: ScheduledGatewayContextResolver | undefined, +): ScheduledGatewayContextResolver | undefined { + if (!resolveGatewayContext) { + return undefined; + } + return () => { + const context = resolveGatewayContext(); + return context?.resolveGatewayContext?.() ?? undefined; + }; +} + +/** + * Runs scheduler-owned work with a Gateway context. + * + * Detached work replaces any request scope inherited when it was queued or + * armed. Caller-owned work must stay outside this boundary. + */ +export async function runWithScheduledGatewayContext(params: { + resolveGatewayContext?: ScheduledGatewayContextResolver; + run: () => Promise; +}): Promise { + const resolveGatewayContext = params.resolveGatewayContext; + if (!resolveGatewayContext) { + return await params.run(); + } + return await withPluginRuntimeGatewayContextResolver(resolveGatewayContext, params.run, { + inheritRequestScope: false, + }); +} diff --git a/src/gateway/server-cron-lazy.ts b/src/gateway/server-cron-lazy.ts index 109ea8584c02..8e39fcc7c8ae 100644 --- a/src/gateway/server-cron-lazy.ts +++ b/src/gateway/server-cron-lazy.ts @@ -6,12 +6,19 @@ import { resolveCronJobsStorePathFromConfig } from "../cron/store.js"; import { createLazyPromiseLoader } from "../shared/lazy-runtime.js"; import type { GatewayCronServiceContract } from "./server-cron-contract.js"; import type { GatewayCronState } from "./server-cron.js"; +import type { GatewayRequestContext } from "./server-methods/types.js"; type LazyGatewayCronParams = { cfg: OpenClawConfig; deps: CliDeps; broadcast: (event: string, payload: unknown, opts?: { dropIfSlow?: boolean }) => void; env?: NodeJS.ProcessEnv; + /** + * Resolves the live Gateway request context for scheduler-triggered runs. + * RPC-triggered runs inherit one from the caller; timer-triggered runs have + * no request of their own, so trusted built-in tools would otherwise see none. + */ + resolveGatewayContext?: () => GatewayRequestContext | undefined; }; type LoadedGatewayCronState = { diff --git a/src/gateway/server-cron.test.ts b/src/gateway/server-cron.test.ts index 65a7bda1c603..bffc0a9aa144 100644 --- a/src/gateway/server-cron.test.ts +++ b/src/gateway/server-cron.test.ts @@ -212,12 +212,19 @@ vi.mock("../cron/trigger-script.js", () => ({ createCronScriptRuntime: createCronScriptRuntimeMock, })); +import { getInProcessGatewayToolContext } from "../agents/tools/in-process-gateway.js"; import { registerActiveCronTaskRun, trackActiveCronTaskRunSettlement, } from "../cron/service/active-run-cancellation.js"; import { resetActiveCronTaskRunsForTests } from "../cron/service/active-run-cancellation.test-support.js"; +import type { CronServiceState } from "../cron/service/state.js"; +import { armTimer } from "../cron/service/timer.js"; import type { CronJob } from "../cron/types.js"; +import { + getPluginRuntimeGatewayRequestScope, + withPluginRuntimeGatewayRequestScope, +} from "../plugins/runtime/gateway-request-scope.js"; import { buildGatewayCronService as buildGatewayCronServiceRuntime, fireOnExitJob, @@ -3617,6 +3624,242 @@ describe("buildGatewayCronService", () => { state.cron.stop(); } }); + + it("replaces the request scope inherited by a scheduler timer", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-21T01:00:00.000Z")); + const cfg = createCronConfig("server-cron-scheduled-gateway-context"); + loadConfigMock.mockReturnValue(cfg); + const gatewayContext = { + terminalSessions: {}, + resolveGatewayContext: () => gatewayContext, + } as never; + let requestContextActive = true; + const retiredRequestContext = { + terminalSessions: { retired: true }, + resolveGatewayContext: () => (requestContextActive ? retiredRequestContext : undefined), + } as never; + const retiredRequestClient = { id: "retired-request" } as never; + let observed: unknown = "never-ran"; + let observedClient: unknown = "never-ran"; + const ran = createDeferred(); + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + observed = getInProcessGatewayToolContext(); + observedClient = getPluginRuntimeGatewayRequestScope()?.client; + ran.resolve(); + return { status: "ok", text: "done" } as never; + }); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + resolveGatewayContext: () => gatewayContext, + }); + try { + await state.cron.start(); + await state.cron.add({ + name: "scheduled-isolated", + enabled: true, + deleteAfterRun: false, + schedule: { kind: "at", at: new Date(Date.now() + 60_000).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "run it" }, + }); + const cronState = (state.cron as unknown as { state: CronServiceState }).state; + withPluginRuntimeGatewayRequestScope( + { + context: retiredRequestContext, + client: retiredRequestClient, + isWebchatConnect: () => false, + } as never, + () => armTimer(cronState), + ); + requestContextActive = false; + + await vi.advanceTimersByTimeAsync(60_000); + await ran.promise; + + expect(observed).toBe(gatewayContext); + expect(observedClient).toBeUndefined(); + } finally { + state.cron.stop(); + vi.useRealTimers(); + } + }); + + it("leaves a scheduler-triggered isolated run without context when no resolver is wired", async () => { + const cfg = createCronConfig("server-cron-scheduled-gateway-context-absent"); + loadConfigMock.mockReturnValue(cfg); + let observed: unknown = "never-ran"; + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + observed = getInProcessGatewayToolContext(); + return { status: "ok", text: "done" } as never; + }); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + }); + try { + const job = await state.cron.add({ + name: "scheduled-isolated-no-resolver", + enabled: true, + deleteAfterRun: false, + schedule: { kind: "at", at: new Date(1).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "run it" }, + }); + + await state.cron.run(job.id, "force"); + + expect(observed).toBeUndefined(); + } finally { + state.cron.stop(); + } + }); + + it("withholds a retired gateway context from a scheduled run", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-21T02:00:00.000Z")); + // The process-wide context holder is not cleared on shutdown, so an + // unfenced resolver would hand a queued run a retired context. No context + // fails visibly; a retired one operates against a dead Gateway generation. + const cfg = createCronConfig("server-cron-retired-gateway-context"); + loadConfigMock.mockReturnValue(cfg); + const retiredContext = { + terminalSessions: {}, + // Instance retired: its own lifecycle resolver reports unavailable. + resolveGatewayContext: () => undefined, + } as never; + let observed: unknown = "never-ran"; + const ran = createDeferred(); + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + observed = getInProcessGatewayToolContext(); + ran.resolve(); + return { status: "ok", text: "done" } as never; + }); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + resolveGatewayContext: () => retiredContext, + }); + try { + await state.cron.start(); + await state.cron.add({ + name: "retired-context", + enabled: true, + deleteAfterRun: false, + schedule: { kind: "at", at: new Date(Date.now() + 60_000).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "run it" }, + }); + + await vi.advanceTimersByTimeAsync(60_000); + await ran.promise; + + expect(observed).toBeUndefined(); + } finally { + state.cron.stop(); + vi.useRealTimers(); + } + }); + + it("gives a scheduled heartbeat wake a resolvable gateway context", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-21T03:00:00.000Z")); + // Main-session cron jobs and heartbeat monitors reach the agent through the + // heartbeat adapter, which shares the isolated path's contextless defect. + const cfg = createCronConfig("server-cron-heartbeat-gateway-context"); + loadConfigMock.mockReturnValue(cfg); + const gatewayContext = { + terminalSessions: {}, + resolveGatewayContext: () => gatewayContext, + } as never; + let observed: unknown = "never-ran"; + const ran = createDeferred(); + runHeartbeatOnceMock.mockImplementationOnce(async () => { + observed = getInProcessGatewayToolContext(); + ran.resolve(); + return { status: "ran", durationMs: 1 }; + }); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + resolveGatewayContext: () => gatewayContext, + }); + try { + await state.cron.start(); + await state.cron.add({ + name: "scheduled-heartbeat", + enabled: true, + deleteAfterRun: false, + schedule: { kind: "at", at: new Date(Date.now() + 60_000).toISOString() }, + sessionTarget: "main", + wakeMode: "now", + payload: { kind: "systemEvent", text: "run it" }, + }); + await vi.advanceTimersByTimeAsync(60_000); + await ran.promise; + + expect(observed).toBe(gatewayContext); + } finally { + state.cron.stop(); + vi.useRealTimers(); + } + }); + + it("keeps an RPC-inherited gateway context instead of the scheduler resolver", async () => { + const cfg = createCronConfig("server-cron-rpc-gateway-context"); + loadConfigMock.mockReturnValue(cfg); + const rpcContext = { terminalSessions: { rpc: true } } as never; + const schedulerContext = { + terminalSessions: { scheduler: true }, + resolveGatewayContext: () => schedulerContext, + } as never; + const resolveGatewayContext = vi.fn(() => schedulerContext); + let observed: unknown = "never-ran"; + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + observed = getInProcessGatewayToolContext(); + return { status: "ok", text: "done" } as never; + }); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + resolveGatewayContext, + }); + try { + const job = await state.cron.add({ + name: "rpc-isolated", + enabled: true, + deleteAfterRun: false, + schedule: { kind: "at", at: new Date(1).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "run it" }, + }); + + await withPluginRuntimeGatewayRequestScope( + { context: rpcContext, isWebchatConnect: () => false } as never, + () => state.cron.run(job.id, "force"), + ); + + expect(observed).toBe(rpcContext); + expect(resolveGatewayContext).not.toHaveBeenCalled(); + } finally { + state.cron.stop(); + } + }); }); describe("fireOnExitJob (on-exit fire routing)", () => { diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index b4df94a2c1db..dba10a133fa8 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -86,6 +86,10 @@ import { type CronStreamFireDisposition, resolveStreamStopReason, } from "./cron-stream-watchers.js"; +import { + fenceScheduledGatewayContextResolver, + runWithScheduledGatewayContext, +} from "./scheduled-run-gateway-context.js"; import type { GatewayCronServiceContract } from "./server-cron-contract.js"; import { reconcileHeartbeatMonitorJobs } from "./server-cron-heartbeat-jobs.js"; import { @@ -93,6 +97,7 @@ import { sendGatewayCronWebhook, sendGatewayCronFailureAlert, } from "./server-cron-notifications.js"; +import type { GatewayRequestContext } from "./server-methods/types.js"; import { bumpSessionAutomationVersion, claimSessionAutomationEpoch, @@ -354,8 +359,14 @@ export function buildGatewayCronService(params: { deps: CliDeps; broadcast: (event: string, payload: unknown, opts?: { dropIfSlow?: boolean }) => void; env?: NodeJS.ProcessEnv; + resolveGatewayContext?: () => GatewayRequestContext | undefined; }): GatewayCronState { const cronLogger = getChildLogger({ module: "cron" }); + // Fence the raw context reference behind its Gateway instance lifecycle so a + // long-running scheduled turn cannot resolve a retired context after shutdown. + const scheduledGatewayContextResolver = fenceScheduledGatewayContextResolver( + params.resolveGatewayContext, + ); const env = params.env ?? process.env; const storePath = resolveCronJobsStorePathFromConfig(params.cfg, env); const cronEnabled = env.OPENCLAW_SKIP_CRON !== "1" && params.cfg.cron?.enabled !== false; @@ -760,6 +771,15 @@ export function buildGatewayCronService(params: { } return resolveCronStoredDeliveryContext({ cfg: runtimeConfig, sessionKey }); }, + ...(scheduledGatewayContextResolver + ? { + runSchedulerOwned: async (run: () => Promise) => + await runWithScheduledGatewayContext({ + resolveGatewayContext: scheduledGatewayContextResolver, + run, + }), + } + : {}), requestHeartbeat: (opts) => { const { agentId, sessionKey } = resolveCronTarget({ ...opts, diff --git a/src/gateway/server-lifecycle.ts b/src/gateway/server-lifecycle.ts index cf8f520614c9..67064a1b02f4 100644 --- a/src/gateway/server-lifecycle.ts +++ b/src/gateway/server-lifecycle.ts @@ -217,6 +217,7 @@ export async function prepareGatewayLifecycle(params: { cfg: cfgAtStart, deps, broadcast, + resolveGatewayContext: runtime.resolvePluginGatewayContext, }), gatewayMethods: listActiveGatewayMethods(pluginRuntime.baseGatewayMethods), }); diff --git a/src/gateway/server-reload-contracts.ts b/src/gateway/server-reload-contracts.ts index bceb7d0d5b22..02ea1d7e3685 100644 --- a/src/gateway/server-reload-contracts.ts +++ b/src/gateway/server-reload-contracts.ts @@ -142,6 +142,10 @@ export type GatewayPluginReloadResult = { export type GatewayReloadHandlerParams = { deps: CliDeps; broadcast: (event: string, payload: unknown, opts?: { dropIfSlow?: boolean }) => void; + /** Kept across cron rebuilds so a hot reload does not drop scheduler gateway context. */ + resolveGatewayContext?: () => + | import("./server-methods/types.js").GatewayRequestContext + | undefined; getState: () => GatewayHotReloadState; setState: (state: GatewayHotReloadState) => void; getPluginMetadataSnapshot?: () => PluginMetadataSnapshot | undefined; diff --git a/src/gateway/server-reload-handlers.test.ts b/src/gateway/server-reload-handlers.test.ts index dd075ecdd335..7017fc2a23f8 100644 --- a/src/gateway/server-reload-handlers.test.ts +++ b/src/gateway/server-reload-handlers.test.ts @@ -1282,6 +1282,19 @@ describe("gateway hot reload model state", () => { ); }); + it("keeps the gateway context resolver when hot reload rebuilds cron", async () => { + const resolveGatewayContext = vi.fn(() => undefined); + const { applyHotReload } = createGatewayReloadHandlers({ resolveGatewayContext }); + + await withGatewayRestartSignal(async () => { + await applyHotReload(createCronRestartPlan(), { cron: { enabled: true } }); + }); + + expect(hoisted.buildGatewayCronService).toHaveBeenCalledWith( + expect.objectContaining({ resolveGatewayContext }), + ); + }); + it("completes reload reconciliation when the replacement scheduler is disabled", async () => { const rebuiltCronState = { cron: { start: vi.fn(async () => {}), stop: vi.fn() }, diff --git a/src/gateway/server-reload-hot.ts b/src/gateway/server-reload-hot.ts index c17d36a1631e..b3d0283750a8 100644 --- a/src/gateway/server-reload-hot.ts +++ b/src/gateway/server-reload-hot.ts @@ -120,6 +120,11 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) deps: params.deps, broadcast: params.broadcast, env: publication?.runtimeEnv ?? process.env, + // Without this a cron hot reload silently drops scheduler gateway + // context, so scheduled runs regress to contextless after any reload. + ...(params.resolveGatewayContext + ? { resolveGatewayContext: params.resolveGatewayContext } + : {}), }); } diff --git a/src/gateway/server-reload-managed.ts b/src/gateway/server-reload-managed.ts index d381f59dfb82..7909ce59a2bb 100644 --- a/src/gateway/server-reload-managed.ts +++ b/src/gateway/server-reload-managed.ts @@ -134,6 +134,9 @@ export function startManagedGatewayConfigReloader( } = createGatewayReloadHandlers({ deps: params.deps, broadcast: params.broadcast, + ...(params.resolveGatewayContext + ? { resolveGatewayContext: params.resolveGatewayContext } + : {}), getState: params.getState, setState: params.setState, getPluginMetadataSnapshot: params.getPluginMetadataSnapshot, diff --git a/src/gateway/server-runtime-services.test.ts b/src/gateway/server-runtime-services.test.ts index f743bb8c30b8..6d9e196be034 100644 --- a/src/gateway/server-runtime-services.test.ts +++ b/src/gateway/server-runtime-services.test.ts @@ -18,6 +18,7 @@ function waitForFast( type StartSessionDeliveryRuntime = typeof import("../infra/session-delivery-queue-runtime.js").startSessionDeliveryRuntime; +type StartHeartbeatRunner = typeof import("../infra/heartbeat-runner.js").startHeartbeatRunner; type DrainPendingDeliveries = typeof import("../infra/outbound/delivery-queue-recovery.js").drainPendingDeliveriesCore; type RecoverPendingDeliveries = @@ -32,7 +33,8 @@ const hoisted = vi.hoisted(() => { const stopSessionDeliveryRuntime = vi.fn(); return { heartbeatRunner, - startHeartbeatRunner: vi.fn(() => heartbeatRunner), + startHeartbeatRunner: vi.fn(() => heartbeatRunner), + runHeartbeatOnce: vi.fn(async () => ({ status: "ran" as const, durationMs: 1 })), startChannelHealthMonitor: vi.fn(() => ({ stop: vi.fn(), shutdown: vi.fn(), @@ -64,6 +66,7 @@ vi.mock("../infra/heartbeat-runner.js", () => ({ { agentId: "main", heartbeat: cfg.agents?.defaults?.heartbeat }, ], startHeartbeatRunner: hoisted.startHeartbeatRunner, + runHeartbeatOnce: hoisted.runHeartbeatOnce, })); vi.mock("../sessions/session-upstream-monitor.js", () => ({ @@ -95,6 +98,11 @@ vi.mock("./channel-health-monitor.js", () => ({ startChannelHealthMonitor: hoisted.startChannelHealthMonitor, })); +import { + getPluginRuntimeGatewayRequestScope, + withPluginRuntimeGatewayRequestScope, +} from "../plugins/runtime/gateway-request-scope.js"; + const { activateGatewayScheduledServices, runGatewayPostReadyMaintenance, @@ -116,6 +124,7 @@ describe("server-runtime-services", () => { hoisted.heartbeatRunner.stop.mockClear(); hoisted.heartbeatRunner.updateConfig.mockClear(); hoisted.startHeartbeatRunner.mockClear(); + hoisted.runHeartbeatOnce.mockClear(); hoisted.startChannelHealthMonitor.mockClear(); hoisted.startSessionUpstreamMonitor.mockClear(); hoisted.stopSessionUpstreamMonitor.mockClear(); @@ -389,6 +398,36 @@ describe("server-runtime-services", () => { expect(hoisted.schedulePendingSessionDeliveries).toHaveBeenCalledTimes(1); }); + it("gives standalone scheduled heartbeats a resolvable gateway context", async () => { + vi.useFakeTimers(); + const gatewayContext = { + terminalSessions: {}, + resolveGatewayContext: () => gatewayContext, + } as never; + let observed: unknown = "never-ran"; + let observedClient: unknown = "never-ran"; + hoisted.runHeartbeatOnce.mockImplementationOnce(async () => { + const scope = getPluginRuntimeGatewayRequestScope(); + observed = scope?.resolveGatewayContext?.(); + observedClient = scope?.client; + return { status: "ran", durationMs: 1 }; + }); + const { services } = activateScheduledServicesForTest({ + resolveGatewayContext: () => gatewayContext, + }); + const runnerParams = hoisted.startHeartbeatRunner.mock.calls[0]?.[0] as + | { runOnce?: (opts: never) => Promise } + | undefined; + + await withPluginRuntimeGatewayRequestScope({ client: { id: "retired-request" } } as never, () => + runnerParams?.runOnce?.({} as never), + ); + + expect(observed).toBe(gatewayContext); + expect(observedClient).toBeUndefined(); + services.heartbeatRunner.stop(); + }); + it("waits for active startup recovery before its stop handle settles", async () => { vi.useFakeTimers(); let resolveRecovery: (() => void) | undefined; diff --git a/src/gateway/server-runtime-services.ts b/src/gateway/server-runtime-services.ts index 53df703d36b6..7d21d509791b 100644 --- a/src/gateway/server-runtime-services.ts +++ b/src/gateway/server-runtime-services.ts @@ -7,6 +7,7 @@ import { resolveHeartbeatAgents, startHeartbeatRunner, type HeartbeatRunner, + runHeartbeatOnce, } from "../infra/heartbeat-runner.js"; import { resolveHeartbeatIntervalMs } from "../infra/heartbeat-summary.js"; import { @@ -18,6 +19,10 @@ import { runWithGatewayIndependentRootWorkAdmission, } from "../process/gateway-work-admission.js"; import { startSessionUpstreamMonitor } from "../sessions/session-upstream-monitor.js"; +import { + fenceScheduledGatewayContextResolver, + runWithScheduledGatewayContext, +} from "./scheduled-run-gateway-context.js"; import type { GatewayCronReconciliation } from "./server-cron-reconciled.js"; import type { GatewayCronState } from "./server-cron.js"; import type { startGatewayMaintenanceTimers } from "./server-maintenance.js"; @@ -361,9 +366,23 @@ export function activateGatewayScheduledServices(params: { "scheduled heartbeats are disabled because the cron scheduler is disabled; enable cron and restart the gateway", ); } + // Scheduled heartbeat wakes fire from a timer with no Gateway request, so + // without this the turn runs contextless and trusted built-in tools fail. + const heartbeatGatewayContextResolver = fenceScheduledGatewayContextResolver( + params.resolveGatewayContext, + ); const heartbeatRunner = startHeartbeatRunner({ cfg: params.cfgAtStart, readCurrentConfig: getRuntimeConfig, + ...(heartbeatGatewayContextResolver + ? { + runOnce: async (opts: Parameters[0]) => + await runWithScheduledGatewayContext({ + resolveGatewayContext: heartbeatGatewayContextResolver, + run: async () => await runHeartbeatOnce(opts), + }), + } + : {}), }); const sessionUpstreamMonitor = startSessionUpstreamMonitor(); const stopSessionDeliveryRuntime = startPendingSessionDeliveryRuntime({ diff --git a/src/gateway/server-runtime-state.ts b/src/gateway/server-runtime-state.ts index 718792a1c59d..6c5be29d8099 100644 --- a/src/gateway/server-runtime-state.ts +++ b/src/gateway/server-runtime-state.ts @@ -171,6 +171,9 @@ export async function createGatewayHttpTransport(params: { bindHost: params.bindHost, port: params.port, logHooks: params.logHooks, + ...(params.getGatewayRequestContext + ? { resolveGatewayContext: params.getGatewayRequestContext } + : {}), }); } return await loadedHooksRequestHandler(req, res); diff --git a/src/gateway/server-startup-finish.ts b/src/gateway/server-startup-finish.ts index 8780890b3d34..5bff5edcb725 100644 --- a/src/gateway/server-startup-finish.ts +++ b/src/gateway/server-startup-finish.ts @@ -368,6 +368,7 @@ export async function finishGatewayStartup(params: { const { startManagedGatewayConfigReloader } = await import("./server-reload-handlers.js"); const configReloaderParams: Parameters[0] = { configRevisionProjector: gatewayRequestContext.configRevisionProjector, + resolveGatewayContext: resolvePluginGatewayContext, minimalTestGateway, initialConfig: cfgAtStart, initialCompareConfig: startupLastGoodSnapshot.sourceConfig, diff --git a/src/gateway/server/hooks.agent-trust.test.ts b/src/gateway/server/hooks.agent-trust.test.ts index 47380eaf870f..96c920b82a0e 100644 --- a/src/gateway/server/hooks.agent-trust.test.ts +++ b/src/gateway/server/hooks.agent-trust.test.ts @@ -61,6 +61,11 @@ vi.mock("../../config/io.js", () => ({ getRuntimeConfig: loadConfigMock, })); +import { + getPluginRuntimeGatewayRequestScope, + withPluginRuntimeGatewayRequestScope, +} from "../../plugins/runtime/gateway-request-scope.js"; + let capturedDispatchAgentHook: ((...args: unknown[]) => unknown) | undefined; let capturedDispatchWakeHook: ((...args: unknown[]) => unknown) | undefined; @@ -278,6 +283,33 @@ describe("dispatchAgentHook trust handling", () => { await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); }); + it("gives a queued hook run a resolvable gateway context", async () => { + const gatewayContext = { + terminalSessions: {}, + resolveGatewayContext: () => gatewayContext, + } as never; + let observed: unknown = "never-ran"; + let observedClient: unknown = "never-ran"; + runCronIsolatedAgentTurnMock.mockImplementationOnce(async () => { + const scope = getPluginRuntimeGatewayRequestScope(); + observed = scope?.resolveGatewayContext?.(); + observedClient = scope?.client; + return { status: "ok", summary: "done", delivered: false }; + }); + createGatewayHooksRequestHandler({ + ...buildMinimalParams(), + resolveGatewayContext: () => gatewayContext, + }); + + await withPluginRuntimeGatewayRequestScope({ client: { id: "retired-request" } } as never, () => + dispatchAgentHook(buildAgentPayload("Gateway context")), + ); + + expect(observed).toBe(gatewayContext); + expect(observedClient).toBeUndefined(); + await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + }); + it("rejects an invalid explicit delivery account before the agent runner", async () => { validateExplicitMessageAccountSelectionMock.mockImplementationOnce(() => { throw new Error('Unknown account "missing" for channel telegram.'); diff --git a/src/gateway/server/hooks.ts b/src/gateway/server/hooks.ts index 58b34640892f..c62578eb6839 100644 --- a/src/gateway/server/hooks.ts +++ b/src/gateway/server/hooks.ts @@ -29,6 +29,11 @@ import { runWithGatewayIndependentRootWorkContinuation } from "../../process/gat import { CommandLane } from "../../process/lanes.js"; import { isUnscopedSessionKeySentinel, toAgentStoreSessionKey } from "../../routing/session-key.js"; import type { HookAgentDispatchPayload, HooksConfigResolved } from "../hooks.js"; +import { + fenceScheduledGatewayContextResolver, + runWithScheduledGatewayContext, +} from "../scheduled-run-gateway-context.js"; +import type { GatewayRequestContext } from "../server-methods/types.js"; import { createHooksRequestHandler, type HookAgentDispatchResult, @@ -228,6 +233,12 @@ export function createGatewayHooksRequestHandler(params: { port: number; logHooks: SubsystemLogger; agentStartAdmissionTimeoutMs?: number; + /** + * Hook agent dispatch runs off a session-keyed queue, so the inbound HTTP + * request scope is already unwound by the time the turn starts. Without this + * the run is contextless and trusted built-in tools fail mid-run. + */ + resolveGatewayContext?: () => GatewayRequestContext | undefined; }) { const { deps, @@ -236,8 +247,11 @@ export function createGatewayHooksRequestHandler(params: { bindHost, port, logHooks, + resolveGatewayContext, agentStartAdmissionTimeoutMs = HOOK_AGENT_START_ADMISSION_TIMEOUT_MS, } = params; + const scheduledGatewayContextResolver = + fenceScheduledGatewayContextResolver(resolveGatewayContext); const enqueueHookAgentDispatch = createSessionKeyedHookDispatchQueue(); let isolatedAgentModulePromise: | Promise @@ -455,27 +469,34 @@ export function createGatewayHooksRequestHandler(params: { if (startupAbortController.signal.aborted) { return; } - const result = await runCronIsolatedAgentTurn({ - cfg, - deps, - job, - message: acceptedValue.message, - sessionKey, - // Isolated runs derive their lifecycle key from random jobId (or an - // already-stable cron: key), so accepted agentId closes reload drift. - agentId, - // Hook agent runs get their own lane rather than sharing - // `cron-nested` with cron inner work, so a saturated cron budget - // cannot starve them. Aggregate capacity stays bounded by the lane - // group that owns both lanes. - lane: CommandLane.HookDispatch, - 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 }); - }, + const runHookIsolatedTurn = async () => + await runCronIsolatedAgentTurn({ + cfg, + deps, + job, + message: acceptedValue.message, + sessionKey, + // Isolated runs derive their lifecycle key from random jobId (or an + // already-stable cron: key), so accepted agentId closes reload drift. + agentId, + // Hook agent runs get their own lane rather than sharing + // `cron-nested` with cron inner work, so a saturated cron budget + // cannot starve them. Aggregate capacity stays bounded by the lane + // group that owns both lanes. + lane: CommandLane.HookDispatch, + 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 }); + }, + }); + const result = await runWithScheduledGatewayContext({ + ...(scheduledGatewayContextResolver + ? { resolveGatewayContext: scheduledGatewayContextResolver } + : {}), + run: runHookIsolatedTurn, }); if (admissionTimedOut) { return; diff --git a/src/plugins/runtime/gateway-request-scope.ts b/src/plugins/runtime/gateway-request-scope.ts index 098cebc32edc..f73cbac8bd53 100644 --- a/src/plugins/runtime/gateway-request-scope.ts +++ b/src/plugins/runtime/gateway-request-scope.ts @@ -77,8 +77,14 @@ export function withPluginRuntimeGatewayRequestScope( export function withPluginRuntimeGatewayContextResolver( resolveGatewayContext: GatewayContextResolver, run: () => T, + options?: { inheritRequestScope?: boolean }, ): T { - const current = pluginRuntimeGatewayRequestScope.getStore(); + // Scheduler-owned work must not retain the request-local client or context + // that happened to exist when its timer was armed. + const current = + options?.inheritRequestScope === false + ? undefined + : pluginRuntimeGatewayRequestScope.getStore(); const scoped: PluginRuntimeGatewayRequestScope = { ...current, isWebchatConnect: current?.isWebchatConnect ?? (() => false),