Address LINE webhook ack policy review

NianJiuZst · NianJiuZst · commit 8a1bf706a0fa · 2026-07-08T18:28:27.000+08:00
diff --git a/docs/channels/line.md b/docs/channels/line.md
@@ -42,9 +42,11 @@ openclaw plugins install ./path/to/local/line-plugin
 https://gateway-host/line/webhook
 ```
 
-The Gateway answers LINE's webhook verification (GET) and acknowledges signed
-inbound events (POST) immediately after signature and payload validation; agent
-processing continues asynchronously.
+The Gateway answers LINE's webhook verification (GET). For signed inbound events
+(POST), it verifies the signature and payload, starts local event dispatch, and
+acknowledges as soon as dispatch is accepted; if dispatch fails immediately,
+OpenClaw returns an error so LINE can redeliver. Agent processing continues
+asynchronously after acknowledgement.
 If you need a custom path, set `channels.line.webhookPath` or
 `channels.line.accounts.<id>.webhookPath` and update the URL accordingly.
 
diff --git a/extensions/line/src/channel.sendPayload.test.ts b/extensions/line/src/channel.sendPayload.test.ts
@@ -624,6 +624,11 @@ describe("line outbound sendPayload", () => {
       adapterName: "line",
       adapter: linePlugin.message!,
       proofs: {
+        after_receive_record: () => {
+          expect(linePlugin.message?.receive?.supportedAckPolicies).toContain(
+            "after_receive_record",
+          );
+        },
         after_agent_dispatch: () => {
           expect(linePlugin.message?.receive?.defaultAckPolicy).toBe("after_agent_dispatch");
           expect(linePlugin.message?.receive?.supportedAckPolicies).toContain(
@@ -637,7 +642,7 @@ describe("line outbound sendPayload", () => {
       "verified",
     );
     expect(proofResults.find((result) => result.policy === "after_receive_record")?.status).toBe(
-      "not_declared",
+      "verified",
     );
   });
 });
diff --git a/extensions/line/src/monitor.lifecycle.test.ts b/extensions/line/src/monitor.lifecycle.test.ts
@@ -452,6 +452,48 @@ describe("monitorLineProvider lifecycle", () => {
     monitor.stop();
   });
 
+  it("supports after_receive_record acknowledgement on shared-path POST requests", async () => {
+    const monitor = await monitorLineProvider({
+      channelAccessToken: "token",
+      channelSecret: "secret", // pragma: allowlist secret
+      accountId: "default",
+      ackPolicy: "after_receive_record",
+      config: {} as OpenClawConfig,
+      runtime: {} as RuntimeEnv,
+    });
+
+    let releaseWebhook: (() => void) | undefined;
+    const bot = createLineBotMock.mock.results[0]?.value as {
+      handleWebhook: ReturnType<typeof vi.fn<LineHandleWebhook>>;
+    };
+    bot.handleWebhook.mockImplementation(
+      () =>
+        new Promise<void>((resolve) => {
+          releaseWebhook = resolve;
+        }),
+    );
+
+    const route = requireRegisteredRoute();
+    const payload = JSON.stringify({ events: [{ type: "message" }] });
+    const signature = crypto.createHmac("SHA256", "secret").update(payload).digest("base64");
+    const req = Object.assign(createMockIncomingRequest([payload]), {
+      method: "POST",
+      headers: { "x-line-signature": signature },
+    }) as unknown as IncomingMessage;
+    const res = createRouteResponse();
+
+    await route.handler(req, res);
+
+    expect(bot.handleWebhook).toHaveBeenCalledTimes(1);
+    expect(res.statusCode).toBe(200);
+    expect(res.headersSent).toBe(true);
+    if (!releaseWebhook) {
+      throw new Error("expected pending LINE webhook handler");
+    }
+    releaseWebhook();
+    monitor.stop();
+  });
+
   it("returns 500 for shared-path POST requests when matched event processing fails", async () => {
     const runtime = {
       error: vi.fn(),
diff --git a/extensions/line/src/monitor.ts b/extensions/line/src/monitor.ts
@@ -48,7 +48,9 @@ import {
 import { buildTemplateMessageFromPayload } from "./template-messages.js";
 import type { LineChannelData, ResolvedLineAccount } from "./types.js";
 import {
+  type LineWebhookAckPolicy,
   logLineWebhookDispatchError,
+  normalizeLineWebhookAckPolicy,
   waitForLineWebhookDispatchAcceptance,
 } from "./webhook-ack.js";
 import { createLineNodeWebhookHandler, readLineWebhookRequestBody } from "./webhook-node.js";
@@ -63,6 +65,7 @@ interface MonitorLineProviderOptions {
   abortSignal?: AbortSignal;
   webhookUrl?: string;
   webhookPath?: string;
+  ackPolicy?: LineWebhookAckPolicy;
 }
 
 interface LineProviderMonitor {
@@ -91,6 +94,7 @@ type LineWebhookTarget = {
   bot: ReturnType<typeof createLineBot>;
   channelSecret: string;
   path: string;
+  ackPolicy: LineWebhookAckPolicy;
   runtime: RuntimeEnv;
 };
 
@@ -171,7 +175,9 @@ export async function monitorLineProvider(
     runtime,
     abortSignal,
     webhookPath,
+    ackPolicy,
   } = opts;
+  const normalizedAckPolicy = normalizeLineWebhookAckPolicy(ackPolicy);
   const resolvedAccountId = accountId ?? resolveDefaultLineAccountId(config);
   const token = channelAccessToken.trim();
   const secret = channelSecret.trim();
@@ -353,6 +359,7 @@ export async function monitorLineProvider(
     createLineNodeWebhookHandler({
       channelSecret: target.channelSecret,
       bot: target.bot,
+      ackPolicy: target.ackPolicy,
       runtime: target.runtime,
     });
   const { unregister: unregisterHttp } = registerWebhookTargetWithPluginRoute({
@@ -362,6 +369,7 @@ export async function monitorLineProvider(
       bot,
       channelSecret: secret,
       path: normalizedPath,
+      ackPolicy: normalizedAckPolicy,
       runtime,
     },
     route: {
@@ -447,13 +455,16 @@ export async function monitorLineProvider(
             id: `${Date.now()}:line:webhook`,
             channel: "line",
             message: body,
-            ackPolicy: "after_agent_dispatch",
+            ackPolicy: match.target.ackPolicy,
             onAck: () => {
               res.statusCode = 200;
               res.setHeader("Content-Type", "application/json");
               res.end(JSON.stringify({ status: "ok" }));
             },
           });
+          if (receiveContext.shouldAckAfter("receive_record")) {
+            await receiveContext.ack();
+          }
 
           requestLifecycle.release();
           if (!body.events || body.events.length === 0) {
@@ -463,6 +474,12 @@ export async function monitorLineProvider(
 
           logVerbose(`line: received ${body.events.length} webhook events`);
           const dispatch = Promise.resolve().then(() => match.target.bot.handleWebhook(body));
+          if (receiveContext.shouldAckAfter("receive_record")) {
+            void dispatch.catch((err: unknown) =>
+              logLineWebhookDispatchError(match.target.runtime, err),
+            );
+            return;
+          }
           const acceptance = await waitForLineWebhookDispatchAcceptance(dispatch);
           if (acceptance.status === "failed") {
             throw acceptance.error;
diff --git a/extensions/line/src/outbound.ts b/extensions/line/src/outbound.ts
@@ -445,6 +445,6 @@ export const lineMessageAdapter = defineChannelMessageAdapter({
   },
   receive: {
     defaultAckPolicy: "after_agent_dispatch",
-    supportedAckPolicies: ["after_agent_dispatch"],
+    supportedAckPolicies: ["after_receive_record", "after_agent_dispatch"],
   },
 });
diff --git a/extensions/line/src/webhook-ack.ts b/extensions/line/src/webhook-ack.ts
@@ -1,11 +1,23 @@
 // Line plugin module implements webhook acknowledgement helpers.
+import type { MessageAckPolicy } from "openclaw/plugin-sdk/channel-outbound";
 import { danger, type RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
 
+export type LineWebhookAckPolicy = Extract<
+  MessageAckPolicy,
+  "after_receive_record" | "after_agent_dispatch"
+>;
+
 type LineWebhookDispatchAcceptance =
   | { status: "completed" }
   | { status: "failed"; error: unknown }
   | { status: "pending" };
 
+export function normalizeLineWebhookAckPolicy(policy: unknown): LineWebhookAckPolicy {
+  return policy === "after_receive_record" || policy === "after_agent_dispatch"
+    ? policy
+    : "after_agent_dispatch";
+}
+
 export function logLineWebhookDispatchError(runtime: RuntimeEnv | undefined, err: unknown): void {
   runtime?.error?.(danger(`line webhook dispatch failed: ${String(err)}`));
 }
diff --git a/extensions/line/src/webhook-node.test.ts b/extensions/line/src/webhook-node.test.ts
@@ -105,6 +105,7 @@ const runSignedPost = async (params: {
   );
 
 async function invokeWebhook(params: {
+  ackPolicy?: "after_agent_dispatch" | "after_receive_record";
   body: unknown;
   headers?: Record<string, string>;
   onEvents?: ReturnType<typeof vi.fn>;
@@ -115,6 +116,7 @@ async function invokeWebhook(params: {
   const middleware = createLineWebhookMiddleware({
     channelSecret: SECRET,
     onEvents: onEventsMock as never,
+    ackPolicy: params.ackPolicy,
     runtime: params.runtime,
   });
 
@@ -475,6 +477,70 @@ describe("createLineNodeWebhookHandler", () => {
     releaseAuthenticated();
   });
 
+  it("supports after_receive_record acknowledgement before event processing completes", async () => {
+    const rawBody = JSON.stringify({ events: [{ type: "message" }] });
+    let releaseAuthenticated: (() => void) | undefined;
+    const bot = {
+      handleWebhook: vi.fn(
+        async () =>
+          await new Promise<void>((resolve) => {
+            releaseAuthenticated = resolve;
+          }),
+      ),
+    };
+    const runtime = createRuntimeMock();
+    const handler = createLineNodeWebhookHandler({
+      channelSecret: SECRET,
+      bot,
+      runtime,
+      readBody: async () => rawBody,
+      ackPolicy: "after_receive_record",
+    });
+
+    const { res } = createRes();
+    await runSignedPost({ handler, rawBody, secret: SECRET, res });
+
+    expect(bot.handleWebhook).toHaveBeenCalledTimes(1);
+    expect(res.statusCode).toBe(200);
+    expect(res.headersSent).toBe(true);
+    if (!releaseAuthenticated) {
+      throw new Error("Expected LINE authenticated request release callback to be initialized");
+    }
+    releaseAuthenticated();
+  });
+
+  it("normalizes unsupported acknowledgement policies to dispatch acceptance", async () => {
+    const rawBody = JSON.stringify({ events: [{ type: "message" }] });
+    let releaseAuthenticated: (() => void) | undefined;
+    const bot = {
+      handleWebhook: vi.fn(
+        async () =>
+          await new Promise<void>((resolve) => {
+            releaseAuthenticated = resolve;
+          }),
+      ),
+    };
+    const runtime = createRuntimeMock();
+    const handler = createLineNodeWebhookHandler({
+      channelSecret: SECRET,
+      bot,
+      runtime,
+      readBody: async () => rawBody,
+      ackPolicy: "manual" as never,
+    });
+
+    const { res } = createRes();
+    await runSignedPost({ handler, rawBody, secret: SECRET, res });
+
+    expect(bot.handleWebhook).toHaveBeenCalledTimes(1);
+    expect(res.statusCode).toBe(200);
+    expect(res.headersSent).toBe(true);
+    if (!releaseAuthenticated) {
+      throw new Error("Expected LINE authenticated request release callback to be initialized");
+    }
+    releaseAuthenticated();
+  });
+
   it("returns 400 for invalid JSON payload even when signature is valid", async () => {
     const rawBody = "not json";
     const { bot, handler, secret } = createPostWebhookTestHarness(rawBody);
@@ -572,6 +638,29 @@ describe("createLineWebhookMiddleware", () => {
     });
   });
 
+  it("supports middleware after_receive_record acknowledgement before event processing completes", async () => {
+    let releaseEvents: (() => void) | undefined;
+    const onEvents = vi.fn(
+      async () =>
+        await new Promise<void>((resolve) => {
+          releaseEvents = resolve;
+        }),
+    );
+
+    const { res } = await invokeWebhook({
+      ackPolicy: "after_receive_record",
+      body: JSON.stringify({ events: [{ type: "message" }] }),
+      onEvents,
+    });
+
+    expect(onEvents).toHaveBeenCalledTimes(1);
+    expect(res.status).toHaveBeenCalledWith(200);
+    if (!releaseEvents) {
+      throw new Error("expected pending LINE middleware handler");
+    }
+    releaseEvents();
+  });
+
   it("rejects invalid signed raw JSON even when req.body is a valid object", async () => {
     const onEvents = vi.fn(async () => {});
     const rawBody = "not-json";
diff --git a/extensions/line/src/webhook-node.ts b/extensions/line/src/webhook-node.ts
@@ -12,7 +12,9 @@ import {
   requestBodyErrorToText,
 } from "openclaw/plugin-sdk/webhook-request-guards";
 import {
+  type LineWebhookAckPolicy,
   logLineWebhookDispatchError,
+  normalizeLineWebhookAckPolicy,
   waitForLineWebhookDispatchAcceptance,
 } from "./webhook-ack.js";
 import { parseLineWebhookBody, validateLineSignature } from "./webhook-utils.js";
@@ -40,10 +42,12 @@ export function createLineNodeWebhookHandler(params: {
   runtime: RuntimeEnv;
   readBody?: ReadBodyFn;
   maxBodyBytes?: number;
+  ackPolicy?: LineWebhookAckPolicy;
   onRequestAuthenticated?: () => void;
 }): (req: IncomingMessage, res: ServerResponse) => Promise<void> {
   const maxBodyBytes = params.maxBodyBytes ?? LINE_WEBHOOK_MAX_BODY_BYTES;
   const readBody = params.readBody ?? readLineWebhookRequestBody;
+  const ackPolicy = normalizeLineWebhookAckPolicy(params.ackPolicy);
 
   return async (req: IncomingMessage, res: ServerResponse) => {
     if (req.method === "GET" || req.method === "HEAD") {
@@ -113,13 +117,16 @@ export function createLineNodeWebhookHandler(params: {
         id: `${Date.now()}:line:webhook`,
         channel: "line",
         message: body,
-        ackPolicy: "after_agent_dispatch",
+        ackPolicy,
         onAck: () => {
           res.statusCode = 200;
           res.setHeader("Content-Type", "application/json");
           res.end(JSON.stringify({ status: "ok" }));
         },
       });
+      if (receiveContext.shouldAckAfter("receive_record")) {
+        await receiveContext.ack();
+      }
 
       if (!body.events || body.events.length === 0) {
         await receiveContext.ack();
@@ -128,6 +135,10 @@ export function createLineNodeWebhookHandler(params: {
 
       logVerbose(`line: received ${body.events.length} webhook events`);
       const dispatch = Promise.resolve().then(() => params.bot.handleWebhook(body));
+      if (receiveContext.shouldAckAfter("receive_record")) {
+        void dispatch.catch((err: unknown) => logLineWebhookDispatchError(params.runtime, err));
+        return;
+      }
       const acceptance = await waitForLineWebhookDispatchAcceptance(dispatch);
       if (acceptance.status === "failed") {
         throw acceptance.error;
diff --git a/extensions/line/src/webhook.ts b/extensions/line/src/webhook.ts
