fix(qa): align runner inbound dispatch contract (#111507)

steipete · web-flow · commit 5a6a19c13393 · 2026-07-19T11:51:24.000-07:00
diff --git a/extensions/qa-lab/src/harness-runtime.integration.test.ts b/extensions/qa-lab/src/harness-runtime.integration.test.ts
@@ -0,0 +1,78 @@
+// Qa Lab integration tests cover the real QA Channel runtime contract.
+import { qaChannelPlugin, setQaChannelRuntime } from "@openclaw/qa-channel/api.js";
+import { describe, expect, it, vi } from "vitest";
+import { startQaBusServer } from "./bus-server.js";
+import { createQaBusState } from "./bus-state.js";
+import { createQaRunnerRuntime } from "./harness-runtime.js";
+import { createQaChannelGatewayConfig } from "./qa-channel-transport.js";
+
+describe("QA runner runtime integration", () => {
+  it("dispatches a QA Channel inbound turn through the embedded runner", async () => {
+    const state = createQaBusState();
+    const bus = await startQaBusServer({ state });
+    const runtime = createQaRunnerRuntime();
+    setQaChannelRuntime(runtime);
+    const config = createQaChannelGatewayConfig({ baseUrl: bus.baseUrl });
+    const account = qaChannelPlugin.config.resolveAccount(config, "default");
+    const abort = new AbortController();
+    const startAccount = qaChannelPlugin.gateway?.startAccount;
+    if (!startAccount) {
+      throw new Error("QA Channel gateway is unavailable");
+    }
+    const gatewayTask = startAccount({
+      accountId: account.accountId,
+      account,
+      cfg: config,
+      runtime: {
+        log: () => undefined,
+        error: () => undefined,
+        exit: () => undefined,
+      },
+      abortSignal: abort.signal,
+      log: {
+        info: () => undefined,
+        warn: () => undefined,
+        error: () => undefined,
+        debug: () => undefined,
+      },
+      getStatus: () => ({
+        accountId: account.accountId,
+        configured: true,
+        enabled: true,
+        running: true,
+      }),
+      setStatus: () => undefined,
+    });
+
+    try {
+      state.addInboundMessage({
+        accountId: "default",
+        conversation: { kind: "direct", id: "alice" },
+        senderId: "alice",
+        senderName: "Alice",
+        text: "ping",
+      });
+
+      await Promise.race([
+        vi.waitFor(
+          () => {
+            expect(state.getSnapshot().messages).toContainEqual(
+              expect.objectContaining({ direction: "outbound", text: "qa-echo: ping" }),
+            );
+          },
+          { interval: 25, timeout: 2_000 },
+        ),
+        gatewayTask.then(() => {
+          throw new Error("QA Channel gateway stopped before delivering the turn");
+        }),
+      ]);
+    } finally {
+      abort.abort();
+      try {
+        await gatewayTask;
+      } finally {
+        await bus.stop();
+      }
+    }
+  });
+});
diff --git a/extensions/qa-lab/src/harness-runtime.ts b/extensions/qa-lab/src/harness-runtime.ts
@@ -15,6 +15,19 @@ type SessionRecord = {
 
 export function createQaRunnerRuntime(): PluginRuntime {
   const sessions = new Map<string, SessionRecord>();
+  const dispatchReplyWithBufferedBlockDispatcher: PluginRuntime["channel"]["reply"]["dispatchReplyWithBufferedBlockDispatcher"] =
+    async ({ ctx, dispatcherOptions }) => {
+      await dispatcherOptions.deliver(
+        {
+          text: `qa-echo: ${ctx.BodyForAgent ?? ctx.Body ?? ""}`,
+        },
+        { kind: "final" },
+      );
+      return {
+        queuedFinal: false,
+        counts: { tool: 0, block: 0, final: 1 },
+      };
+    };
   return {
     channel: {
       routing: {
@@ -73,39 +86,28 @@ export function createQaRunnerRuntime(): PluginRuntime {
         finalizeInboundContext(ctx: Record<string, unknown>) {
           return ctx as typeof ctx & { CommandAuthorized: boolean };
         },
-        async dispatchReplyWithBufferedBlockDispatcher({
-          ctx,
-          dispatcherOptions,
-        }: {
-          ctx: { BodyForAgent?: string; Body?: string };
-          dispatcherOptions: { deliver: (payload: { text: string }) => Promise<void> };
-        }) {
-          await dispatcherOptions.deliver({
-            text: `qa-echo: ${ctx.BodyForAgent ?? ctx.Body ?? ""}`,
-          });
-        },
+        dispatchReplyWithBufferedBlockDispatcher,
       },
       inbound: {
-        async dispatchReply(
-          params: Parameters<PluginRuntime["channel"]["inbound"]["dispatchReply"]>[0],
-        ) {
+        async dispatch(params: Parameters<PluginRuntime["channel"]["inbound"]["dispatch"]>[0]) {
           const sessionKey =
             typeof params.ctxPayload.SessionKey === "string"
               ? params.ctxPayload.SessionKey
-              : params.routeSessionKey;
-          await params.recordInboundSession({
-            storePath: params.storePath,
+              : params.route.sessionKey;
+          sessions.set(sessionKey, {
             sessionKey,
-            ctx: params.ctxPayload,
-            onRecordError: params.record?.onRecordError ?? (() => undefined),
+            body: params.ctxPayload.BodyForAgent ?? params.ctxPayload.Body ?? "",
           });
-          const dispatchResult = await params.dispatchReplyWithBufferedBlockDispatcher({
+          const delivery =
+            params.admission?.kind === "observeOnly"
+              ? async () => ({ visibleReplySent: false })
+              : params.delivery.deliver;
+          const dispatchResult = await dispatchReplyWithBufferedBlockDispatcher({
             ctx: params.ctxPayload,
             cfg: params.cfg,
             dispatcherOptions: {
-              ...params.dispatcherOptions,
               deliver: async (payload, info) => {
-                await params.delivery.deliver(payload, info);
+                await delivery(payload, info);
               },
               onError: params.delivery.onError,
             },
@@ -116,7 +118,7 @@ export function createQaRunnerRuntime(): PluginRuntime {
             admission: params.admission ?? { kind: "dispatch" },
             dispatched: true,
             ctxPayload: params.ctxPayload,
-            routeSessionKey: params.routeSessionKey,
+            routeSessionKey: params.route.sessionKey,
             dispatchResult,
           };
         },
