fix(gateway): validate hook delivery accounts

2026-08-13 17:07:40 +00:00 · 2026-07-29 19:46:12 -05:00
parent a67ad03eea
commit 54a6dba0d3
8 changed files with 377 additions and 62 deletions
--- a/docs/automation/cron-jobs.md
+++ b/docs/automation/cron-jobs.md
@@ -545,10 +545,11 @@ Query-string tokens are rejected.
    - Announce delivery requires a concrete channel; webhook hooks never inherit the main session's `last` channel or recipient.
    - Setting `deliver: false` keeps the run completion-only and ignores any delivery destination.
    - Supplying both a concrete `channel` and `to` enables direct announce delivery.
-    - Set `accountId` with `channel` and `to` to select a configured account on multi-account channels.
+    - Set `accountId` with `channel` and `to` to select a configured, enabled account on multi-account channels. Unknown, disabled, or invalid account IDs return `400` and schedule no run.

    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:

+    - `400` when delivery coordinates or account selection are invalid; correct the request before retrying.
    - `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.
--- a/docs/gateway/configuration-reference.md
+++ b/docs/gateway/configuration-reference.md
@@ -934,7 +934,7 @@ Validation and safety notes:
  - `sessionKey` from request payload is accepted only when `hooks.allowRequestSessionKey=true` (default: `false`).
  - `sessionMode` is `"isolated"` by default. `"persistent"` reuses the resolved session and requires an explicit request `sessionKey`, `hooks.allowRequestSessionKey=true`, and non-empty `hooks.allowedSessionKeyPrefixes`.
  - Direct announce delivery requires both a concrete `channel` and `to`; supplying only one fails before the run is scheduled.
-  - `accountId` selects a configured account for direct announce delivery and requires both `channel` and `to`.
+  - `accountId` selects a configured, enabled account for direct announce delivery and requires both `channel` and `to`; invalid selections return `400` before a run starts.
  - 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.
--- a/extensions/qa-lab/src/hook-agent-account-routing.e2e.test.ts
+++ b/extensions/qa-lab/src/hook-agent-account-routing.e2e.test.ts
@@ -9,6 +9,7 @@ import { createQaChannelTransport } from "./qa-channel-transport.js";

 const HOOK_TOKEN = "qa-hook-account-routing-token";
 const MARKER = "QA_HOOK_ACCOUNT_ROUTING_OK";
+const DEFAULT_MARKER = "QA_HOOK_DEFAULT_ACCOUNT_ROUTING_OK";
 const MODEL = "mock-openai/gpt-5.6-luna";

 async function postJson(url: string, body: unknown, headers: Record<string, string>) {
@@ -27,65 +28,119 @@ async function postJson(url: string, body: unknown, headers: Record<string, stri
  };
 }

+async function startHookAccountFixture() {
+  const repoRoot = fileURLToPath(new URL("../../../", import.meta.url));
+  const lab = await startQaLabServer({
+    repoRoot,
+    embeddedGateway: "disabled",
+  });
+  const mock = await startQaProviderServer("mock-openai", {
+    modelRefs: [MODEL],
+  });
+  let gateway: Awaited<ReturnType<typeof startQaGatewayChild>> | undefined;
+
+  try {
+    if (!mock) {
+      throw new Error("mock-openai provider server did not start");
+    }
+    const transport = createQaChannelTransport(lab.state);
+    gateway = await startQaGatewayChild({
+      repoRoot,
+      useRepoCli: true,
+      providerBaseUrl: `${mock.baseUrl}/v1`,
+      providerMode: "mock-openai",
+      primaryModel: MODEL,
+      alternateModel: MODEL,
+      transportBaseUrl: lab.listenUrl,
+      transport,
+      controlUiEnabled: false,
+      mutateConfig: (config) => {
+        const qaChannel = config.channels?.["qa-channel"];
+        if (!qaChannel) {
+          throw new Error("qa-channel transport config missing");
+        }
+        return {
+          ...config,
+          hooks: {
+            enabled: true,
+            token: HOOK_TOKEN,
+            path: "/hooks",
+          },
+          channels: {
+            ...config.channels,
+            "qa-channel": {
+              ...qaChannel,
+              defaultAccount: "default",
+              accounts: {
+                default: { name: "Default" },
+                work: { name: "Work" },
+                disabled: { name: "Disabled", enabled: false },
+              },
+            },
+          },
+        };
+      },
+    });
+  } catch (error) {
+    await gateway?.stop().catch(() => undefined);
+    await mock?.stop().catch(() => undefined);
+    await lab.stop().catch(() => undefined);
+    throw error;
+  }
+  if (!gateway || !mock) {
+    throw new Error("hook account fixture did not start");
+  }
+
+  return {
+    gateway,
+    lab,
+    stop: async () => {
+      await gateway.stop().catch(() => undefined);
+      await mock.stop().catch(() => undefined);
+      await lab.stop().catch(() => undefined);
+    },
+  };
+}
+
 describe("hook agent account routing product proof", () => {
+  it(
+    "rejects invalid explicit accounts before running the agent",
+    { timeout: 180_000 },
+    async () => {
+      const fixture = await startHookAccountFixture();
+
+      try {
+        for (const accountId of ["missing", "disabled", "__proto__"]) {
+          const response = await postJson(
+            `${fixture.gateway.baseUrl}/hooks/agent`,
+            {
+              message: `Reply exactly: ${MARKER}`,
+              deliver: true,
+              channel: "qa-channel",
+              to: "dm:hook-recipient",
+              accountId,
+            },
+            { Authorization: `Bearer ${HOOK_TOKEN}` },
+          );
+
+          expect(response.status, JSON.stringify(response.json)).toBe(400);
+        }
+        expect(fixture.lab.state.getSnapshot().messages).toHaveLength(0);
+      } finally {
+        await fixture.stop();
+      }
+    },
+  );
+
  it(
    "delivers exactly once through the selected qa-channel account",
    { timeout: 180_000 },
    async () => {
-      const repoRoot = fileURLToPath(new URL("../../../", import.meta.url));
-      const lab = await startQaLabServer({
-        repoRoot,
-        embeddedGateway: "disabled",
-      });
-      const mock = await startQaProviderServer("mock-openai", {
-        modelRefs: [MODEL],
-      });
-      let gateway: Awaited<ReturnType<typeof startQaGatewayChild>> | undefined;
+      const fixture = await startHookAccountFixture();

      try {
-        if (!mock) {
-          throw new Error("mock-openai provider server did not start");
-        }
-        const transport = createQaChannelTransport(lab.state);
-        gateway = await startQaGatewayChild({
-          repoRoot,
-          useRepoCli: true,
-          providerBaseUrl: `${mock.baseUrl}/v1`,
-          providerMode: "mock-openai",
-          primaryModel: MODEL,
-          alternateModel: MODEL,
-          transportBaseUrl: lab.listenUrl,
-          transport,
-          controlUiEnabled: false,
-          mutateConfig: (config) => {
-            const qaChannel = config.channels?.["qa-channel"];
-            if (!qaChannel) {
-              throw new Error("qa-channel transport config missing");
-            }
-            return {
-              ...config,
-              hooks: {
-                enabled: true,
-                token: HOOK_TOKEN,
-                path: "/hooks",
-              },
-              channels: {
-                ...config.channels,
-                "qa-channel": {
-                  ...qaChannel,
-                  defaultAccount: "default",
-                  accounts: {
-                    default: { name: "Default" },
-                    work: { name: "Work" },
-                  },
-                },
-              },
-            };
-          },
-        });
-
        const response = await postJson(
-          `${gateway.baseUrl}/hooks/agent`,
+          `${fixture.gateway.baseUrl}/hooks/agent`,
          {
            message: `Reply exactly: ${MARKER}`,
            deliver: true,
@@ -103,7 +158,7 @@ describe("hook agent account routing product proof", () => {

        await vi.waitFor(
          () => {
-            const outbound = lab.state
+            const outbound = fixture.lab.state
              .getSnapshot()
              .messages.filter((message) => message.direction === "outbound");
            expect(outbound).toHaveLength(1);
@@ -117,15 +172,40 @@ describe("hook agent account routing product proof", () => {
        );
        await sleep(500);

-        const outbound = lab.state
+        const outbound = fixture.lab.state
          .getSnapshot()
          .messages.filter((message) => message.direction === "outbound");
        expect(outbound).toHaveLength(1);
        expect(outbound.filter((message) => message.accountId === "default")).toHaveLength(0);
+
+        const defaultResponse = await postJson(
+          `${fixture.gateway.baseUrl}/hooks/agent`,
+          {
+            message: `Reply exactly: ${DEFAULT_MARKER}`,
+            deliver: true,
+            channel: "qa-channel",
+            to: "dm:hook-default-recipient",
+          },
+          { Authorization: `Bearer ${HOOK_TOKEN}` },
+        );
+        expect(defaultResponse.status, JSON.stringify(defaultResponse.json)).toBe(200);
+
+        await vi.waitFor(
+          () => {
+            const currentOutbound = fixture.lab.state
+              .getSnapshot()
+              .messages.filter((message) => message.direction === "outbound");
+            expect(currentOutbound).toHaveLength(2);
+            expect(currentOutbound[1]).toMatchObject({
+              accountId: "default",
+              conversation: { id: "hook-default-recipient", kind: "direct" },
+              text: DEFAULT_MARKER,
+            });
+          },
+          { interval: 50, timeout: 60_000 },
+        );
      } finally {
-        await gateway?.stop().catch(() => undefined);
-        await mock?.stop().catch(() => undefined);
-        await lab.stop().catch(() => undefined);
+        await fixture.stop();
      }
    },
  );
--- a/src/gateway/server.hooks-session-mode.test.ts
+++ b/src/gateway/server.hooks-session-mode.test.ts
@@ -3,6 +3,7 @@ import nodePath from "node:path";
 import { afterEach, describe, expect, test, vi } from "vitest";
 import { resolveMainSessionKeyFromConfig } from "../config/sessions.js";
 import { drainSystemEvents } from "../infra/system-events.js";
+import { createChannelTestPluginBase, createTestRegistry } from "../test-utils/channel-plugins.js";
 import {
  cronIsolatedRun,
  installGatewayTestHooks,
@@ -10,6 +11,7 @@ import {
  withGatewayServer,
  waitForSystemEvent,
 } from "./test-helpers.js";
+import { setTestPluginRegistry } from "./test-helpers.plugin-registry.js";

 installGatewayTestHooks({ scope: "suite" });

@@ -251,6 +253,21 @@ describe("gateway hook session mode", () => {
      token: HOOK_TOKEN,
    };
    await withGatewayServer(async ({ port }) => {
+      setTestPluginRegistry(
+        createTestRegistry([
+          {
+            pluginId: "discord",
+            source: "test",
+            plugin: createChannelTestPluginBase({
+              id: "discord",
+              config: {
+                listAccountIds: () => ["work", "personal"],
+                resolveAccount: (_cfg, accountId) => ({ accountId }),
+              },
+            }),
+          },
+        ]),
+      );
      mockRunsOk();
      const headers = { "Idempotency-Key": "hook-idem-account-id" };
      const basePayload = {
--- a/src/gateway/server/hooks-request-handler.ts
+++ b/src/gateway/server/hooks-request-handler.ts
@@ -61,7 +61,7 @@ type HookDispatchers = {

 export type HookAgentDispatchResult =
  | { ok: true; runId: string }
-  | { ok: false; statusCode: 409 | 502 | 503; error: string; runId?: string };
+  | { ok: false; statusCode: 400 | 409 | 502 | 503; error: string; runId?: string };

 type HookReplayEntry = {
  ts: number;
--- a/src/gateway/server/hooks.agent-trust.test.ts
+++ b/src/gateway/server/hooks.agent-trust.test.ts
@@ -20,6 +20,11 @@ const mainRosterConfig = (): OpenClawConfig => ({
 const loadConfigMock = vi.fn(mainRosterConfig);
 const logHooksInfoMock = vi.fn();
 const logHooksWarnMock = vi.fn();
+const validateExplicitMessageAccountSelectionMock = vi.fn(
+  ({ accountId }: { accountId?: unknown }) => accountId as string | undefined,
+);
+const resolveOutboundChannelPluginMock = vi.fn(() => ({ id: "telegram" }));
+const resolveChannelDefaultAccountIdMock = vi.fn(() => "default");

 vi.mock("../../infra/system-events.js", () => ({
  enqueueSystemEvent: enqueueSystemEventMock,
@@ -30,6 +35,15 @@ vi.mock("../../infra/heartbeat-wake.js", () => ({
 vi.mock("../../cron/isolated-agent.js", () => ({
  runCronIsolatedAgentTurn: runCronIsolatedAgentTurnMock,
 }));
+vi.mock("../../infra/outbound/message-account-selection.js", () => ({
+  validateExplicitMessageAccountSelection: validateExplicitMessageAccountSelectionMock,
+}));
+vi.mock("../../infra/outbound/channel-resolution.js", () => ({
+  resolveOutboundChannelPlugin: resolveOutboundChannelPluginMock,
+}));
+vi.mock("../../channels/plugins/helpers.js", () => ({
+  resolveChannelDefaultAccountId: resolveChannelDefaultAccountIdMock,
+}));
 vi.mock("../../config/sessions.js", () => ({
  resolveMainSessionKeyFromConfig: resolveMainSessionKeyMock,
  resolveMainSessionKey: vi.fn(
@@ -158,6 +172,11 @@ describe("dispatchAgentHook trust handling", () => {
    resetGatewayWorkAdmission();
    vi.clearAllMocks();
    loadConfigMock.mockImplementation(mainRosterConfig);
+    validateExplicitMessageAccountSelectionMock.mockImplementation(
+      ({ accountId }: { accountId?: unknown }) => accountId as string | undefined,
+    );
+    resolveOutboundChannelPluginMock.mockReturnValue({ id: "telegram" });
+    resolveChannelDefaultAccountIdMock.mockReturnValue("default");
    capturedDispatchAgentHook = undefined;
    createGatewayHooksRequestHandler(buildMinimalParams());
  });
@@ -196,6 +215,93 @@ describe("dispatchAgentHook trust handling", () => {
    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.');
+    });
+
+    const result = await dispatchAgentHook({
+      ...buildAgentPayload("Invalid account"),
+      deliver: true,
+      channel: "telegram",
+      to: "123456",
+      accountId: "missing",
+      delivery: {
+        mode: "announce",
+        channel: "telegram",
+        to: "123456",
+        accountId: "missing",
+      },
+    });
+
+    expect(result).toMatchObject({
+      ok: false,
+      statusCode: 400,
+      error: 'Unknown account "missing" for channel telegram.',
+      runId: expect.any(String),
+    });
+    expect(runCronIsolatedAgentTurnMock).not.toHaveBeenCalled();
+  });
+
+  it("binds omitted delivery accounts to the channel default", async () => {
+    runCronIsolatedAgentTurnMock.mockResolvedValueOnce({
+      status: "ok",
+      summary: "done",
+      delivered: true,
+    });
+
+    const result = await dispatchAgentHook({
+      ...buildAgentPayload("Default account"),
+      deliver: true,
+      channel: "telegram",
+      to: "123456",
+      delivery: {
+        mode: "announce",
+        channel: "telegram",
+        to: "123456",
+      },
+    });
+
+    expect(result).toMatchObject({ ok: true });
+    expect(runCronIsolatedAgentTurnMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        job: expect.objectContaining({
+          delivery: expect.objectContaining({ accountId: "default" }),
+        }),
+      }),
+    );
+  });
+
+  it("revalidates an explicit delivery account against queued-run config", async () => {
+    validateExplicitMessageAccountSelectionMock
+      .mockImplementationOnce(({ accountId }: { accountId?: unknown }) => accountId as string)
+      .mockImplementationOnce(() => {
+        throw new Error('Unknown account "removed" for channel telegram.');
+      });
+
+    const result = await dispatchAgentHook({
+      ...buildAgentPayload("Removed account"),
+      deliver: true,
+      channel: "telegram",
+      to: "123456",
+      accountId: "removed",
+      delivery: {
+        mode: "announce",
+        channel: "telegram",
+        to: "123456",
+        accountId: "removed",
+      },
+    });
+
+    expect(result).toMatchObject({
+      ok: false,
+      statusCode: 400,
+      error: 'Unknown account "removed" for channel telegram.',
+      runId: expect.any(String),
+    });
+    expect(runCronIsolatedAgentTurnMock).not.toHaveBeenCalled();
+  });
+
  it("retains detached agent work after the hook request releases admission", async () => {
    let continueRun = () => {};
    let subordinateAdmissionClosed: boolean | undefined;
--- a/src/gateway/server/hooks.ts
+++ b/src/gateway/server/hooks.ts
@@ -7,6 +7,7 @@ import {
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
 import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
 import { resolveDefaultAgentId } from "../../agents/agent-scope.js";
+import { resolveChannelDefaultAccountId } from "../../channels/plugins/helpers.js";
 import type { CliDeps } from "../../cli/deps.types.js";
 import { getRuntimeConfig } from "../../config/io.js";
 import {
@@ -22,7 +23,10 @@ import type {
 } from "../../cron/isolated-agent/run.types.js";
 import { resolveCronAgentSessionKey } from "../../cron/isolated-agent/session-key.js";
 import type { CronJob } from "../../cron/types.js";
+import { formatErrorMessage } from "../../infra/errors.js";
 import { requestHeartbeat } from "../../infra/heartbeat-wake.js";
+import { resolveOutboundChannelPlugin } from "../../infra/outbound/channel-resolution.js";
+import { validateExplicitMessageAccountSelection } from "../../infra/outbound/message-account-selection.js";
 import { enqueueSystemEvent } from "../../infra/system-events.js";
 import type { createSubsystemLogger } from "../../logging/subsystem.js";
 import { runWithGatewayIndependentRootWorkContinuation } from "../../process/gateway-work-admission.js";
@@ -150,6 +154,45 @@ function createSessionKeyedHookDispatchQueue() {
  };
 }

+function validateHookAgentDeliveryAccount(params: {
+  cfg: OpenClawConfig;
+  value: HookAgentDispatchPayload;
+}): HookAgentDispatchPayload {
+  // Mapped hooks can defer partial/last targets to cron and cannot select an account.
+  // Bind only direct hook announces whose destination is already complete.
+  if (
+    params.value.delivery.mode !== "announce" ||
+    params.value.delivery.channel === "last" ||
+    !params.value.delivery.to
+  ) {
+    return params.value;
+  }
+  const accountId = params.value.delivery.accountId
+    ? validateExplicitMessageAccountSelection({
+        cfg: params.cfg,
+        channel: params.value.delivery.channel,
+        accountId: params.value.delivery.accountId,
+      })
+    : (() => {
+        const plugin = resolveOutboundChannelPlugin({
+          channel: params.value.delivery.channel,
+          cfg: params.cfg,
+        });
+        if (!plugin) {
+          throw new Error(`Channel ${params.value.delivery.channel} is unavailable.`);
+        }
+        return resolveChannelDefaultAccountId({ plugin, cfg: params.cfg });
+      })();
+  if (!accountId) {
+    throw new Error(`Channel ${params.value.delivery.channel} did not resolve an account.`);
+  }
+  return {
+    ...params.value,
+    accountId,
+    delivery: { ...params.value.delivery, accountId },
+  };
+}
+
 /** Creates the HTTP handler used by gateway hook endpoints. */
 export function createGatewayHooksRequestHandler(params: {
  deps: CliDeps;
@@ -280,7 +323,19 @@ export function createGatewayHooksRequestHandler(params: {
      void runWithGatewayIndependentRootWorkContinuation(async () => reportHookFailure(err));
      return createHookAdmissionFailure({ runId });
    }
-    const agentId = value.agentId ?? resolveDefaultAgentId(dispatchCfg);
+    let acceptedValue: HookAgentDispatchPayload;
+    try {
+      acceptedValue = validateHookAgentDeliveryAccount({ cfg: dispatchCfg, value });
+      job.delivery = acceptedValue.delivery;
+    } catch (err) {
+      return {
+        ok: false,
+        statusCode: 400,
+        error: formatErrorMessage(err),
+        runId,
+      };
+    }
+    const agentId = acceptedValue.agentId ?? resolveDefaultAgentId(dispatchCfg);
    const queueKey = resolveCronAgentSessionKey({
      sessionKey,
      agentId,
@@ -329,11 +384,22 @@ export function createGatewayHooksRequestHandler(params: {
        }
        try {
          const cfg = getRuntimeConfig();
+          try {
+            validateHookAgentDeliveryAccount({ cfg, value: acceptedValue });
+          } catch (err) {
+            settleAdmission({
+              ok: false,
+              statusCode: 400,
+              error: formatErrorMessage(err),
+              runId,
+            });
+            return;
+          }
          // 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,
+            agentId: acceptedValue.agentId,
          });
          const { runCronIsolatedAgentTurn } = await loadIsolatedAgentModule();
          // Lazy module loading is the last Gateway-owned async boundary before
@@ -345,7 +411,7 @@ export function createGatewayHooksRequestHandler(params: {
            cfg,
            deps,
            job,
-            message: value.message,
+            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.
--- a/src/plugins/runtime/subagent-requester-context.test.ts
+++ b/src/plugins/runtime/subagent-requester-context.test.ts
@@ -79,4 +79,49 @@ describe("plugin subagent requester context", () => {
    await expect(detachedRead).resolves.toBeUndefined();
    expect(getActiveRequester()).toBeUndefined();
  });
+
+  it("isolates concurrent requester scopes", async () => {
+    const first = createPluginSubagentRequesterContext({
+      sessionKey: "agent:main:telegram:direct:first",
+      origin: { channel: "telegram", to: "telegram:first", accountId: "first" },
+    });
+    const second = createPluginSubagentRequesterContext({
+      sessionKey: "agent:main:telegram:direct:second",
+      origin: { channel: "telegram", to: "telegram:second", accountId: "second" },
+    });
+    if (!first || !second) {
+      throw new Error("expected valid requester contexts");
+    }
+
+    let release: (() => void) | undefined;
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    let firstReady: (() => void) | undefined;
+    let secondReady: (() => void) | undefined;
+    const firstStarted = new Promise<void>((resolve) => {
+      firstReady = resolve;
+    });
+    const secondStarted = new Promise<void>((resolve) => {
+      secondReady = resolve;
+    });
+
+    const firstRun = withPluginSubagentRequesterContext(first, async () => {
+      expect(getActiveRequester()).toBe(first);
+      firstReady?.();
+      await gate;
+      expect(getActiveRequester()).toBe(first);
+    });
+    const secondRun = withPluginSubagentRequesterContext(second, async () => {
+      expect(getActiveRequester()).toBe(second);
+      secondReady?.();
+      await gate;
+      expect(getActiveRequester()).toBe(second);
+    });
+
+    await Promise.all([firstStarted, secondStarted]);
+    release?.();
+    await Promise.all([firstRun, secondRun]);
+    expect(getActiveRequester()).toBeUndefined();
+  });
 });