fix(qa): consume Crabline events without recorder polling (#99679)

* fix(qa): consume Crabline events in process

* chore(qa): use Crabline 0.1.9

extensions/qa-lab/package.json

@@ -16,7 +16,7 @@
     "@openclaw/plugin-sdk": "workspace:*",
     "@openclaw/slack": "workspace:*",
     "@openclaw/whatsapp": "workspace:*",
-    "@openclaw/crabline": "0.1.8",
+    "@openclaw/crabline": "0.1.9",
     "openclaw": "workspace:*"
   },
   "peerDependencies": {

extensions/qa-lab/src/crabline-transport.test.ts

@@ -104,83 +104,6 @@ describe("crabline transport", () => {
     });
   });
 
-  it("defers a partial recorder tail until Crabline finishes the JSONL record", async () => {
-    await withTempDir("qa-crabline-transport-", async (outputDir) => {
-      const transport = await createQaCrablineTransportAdapter({
-        outputDir,
-        selection: createSelection(),
-        state: createQaBusState(),
-      });
-      const manifest = JSON.parse(
-        await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"),
-      ) as { recorderPath: string };
-      const recorderLine = JSON.stringify({
-        body: {
-          chat_id: "100001",
-          text: `partial recorder write ${"x".repeat(240)}`,
-        },
-        path: "/bot<redacted>/sendMessage",
-        type: "api",
-      });
-      const splitIndex = 191;
-
-      try {
-        await transport.reset();
-        await fs.appendFile(manifest.recorderPath, recorderLine.slice(0, splitIndex), "utf8");
-
-        await expect(transport.reset()).resolves.toBeUndefined();
-
-        await fs.appendFile(manifest.recorderPath, `${recorderLine.slice(splitIndex)}\n`, "utf8");
-        await expect(
-          transport.state.searchMessages({ query: "partial recorder write" }),
-        ).resolves.toHaveLength(1);
-      } finally {
-        await transport.cleanup?.();
-      }
-    });
-  });
-
-  it("rejects a malformed newline-terminated recorder record", async () => {
-    await withTempDir("qa-crabline-transport-", async (outputDir) => {
-      const transport = await createQaCrablineTransportAdapter({
-        outputDir,
-        selection: createSelection(),
-        state: createQaBusState(),
-      });
-      const manifest = JSON.parse(
-        await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"),
-      ) as { recorderPath: string };
-
-      try {
-        await transport.reset();
-        await fs.appendFile(manifest.recorderPath, '{"broken":\n', "utf8");
-
-        await expect(transport.reset()).rejects.toThrow(SyntaxError);
-        await fs.writeFile(manifest.recorderPath, "", "utf8");
-      } finally {
-        await transport.cleanup?.();
-      }
-    });
-  });
-
-  it("rejects a permanently truncated recorder tail during cleanup", async () => {
-    await withTempDir("qa-crabline-transport-", async (outputDir) => {
-      const transport = await createQaCrablineTransportAdapter({
-        outputDir,
-        selection: createSelection(),
-        state: createQaBusState(),
-      });
-      const manifest = JSON.parse(
-        await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"),
-      ) as { recorderPath: string };
-
-      await transport.reset();
-      await fs.appendFile(manifest.recorderPath, '{"unfinished":"recorder tail', "utf8");
-
-      await expect(transport.cleanup?.()).rejects.toThrow(SyntaxError);
-    });
-  });
-
   it("observes Telegram preview edits through the shared transport adapter", async () => {
     await withTempDir("qa-crabline-transport-", async (outputDir) => {
       const transport = await createQaCrablineTransportAdapter({
@@ -213,6 +136,9 @@ describe("crabline transport", () => {
           message_thread_id: 42,
           text: "preview text",
         });
+        expect(transport.state.searchMessages({ query: "preview text" })).toEqual([
+          expect.objectContaining({ text: "preview text" }),
+        ]);
         await postTelegram("editMessageText", {
           chat_id: "-1001234567890",
           message_id: sent.result.message_id,

extensions/qa-lab/src/crabline-transport.ts

@@ -37,31 +37,19 @@ import type {
   QaBusInboundMessageInput,
   QaBusMessage,
   QaBusOutboundMessageInput,
-  QaBusSearchMessagesInput,
-  QaBusWaitForInput,
 } from "./runtime-api.js";
 
 const CRABLINE_TRANSPORT_ID = "crabline";
-const RECORDER_SYNC_INTERVAL_MS = 50;
 
 type QaCrablineTransportState = QaTransportState & {
   cleanup: () => Promise<void>;
   getOutboundEvents: () => Promise<readonly QaTransportOutboundEvent[]>;
+  observeEvent: (event: unknown) => void;
   rememberProviderTarget: (providerTargetKey: string, qaTarget: string) => void;
 };
 
 const TELEGRAM_LIFECYCLE_METHOD_RE = /\/(sendMessage|editMessageText|deleteMessage)$/u;
 
-function readRecorderLines(text: string, options: { allowIncompleteTail: boolean }): string[] {
-  // Crabline appends each JSONL record asynchronously, so a concurrent read can end mid-record.
-  // Defer only the unterminated tail; newline-terminated malformed records must still fail parsing.
-  const lines = text.split(/\r?\n/u);
-  if (options.allowIncompleteTail && !text.endsWith("\n")) {
-    lines.pop();
-  }
-  return lines.filter((line) => line.trim().length > 0);
-}
-
 function readTelegramLifecycleEvent(params: {
   cursor: number;
   event: unknown;
@@ -224,25 +212,24 @@ function createCrablineState(params: {
   const telegramMessageByProviderId = new Map<string, QaBusMessage>();
   const pendingTelegramMessagesByChat = new Map<string, QaBusMessage[]>();
   const outboundEvents: QaTransportOutboundEvent[] = [];
-  let recorderLineCursor = 0;
-  let syncPromise: Promise<void> | null = null;
 
-  const syncRecorderSnapshot = async (options: { allowIncompleteTail: boolean }) => {
-    const text = await fs
-      .readFile(params.adapter.manifest.recorderPath, "utf8")
-      .catch((error: unknown) => {
-        if ((error as NodeJS.ErrnoException).code === "ENOENT") {
-          return "";
-        }
-        throw error;
-      });
-    const lines = readRecorderLines(text, options);
-    for (const line of lines.slice(recorderLineCursor)) {
-      const parsed = JSON.parse(line) as unknown;
+  return {
+    reset() {
+      baseState.reset();
+      targetByProviderTarget.clear();
+      telegramMessageByProviderId.clear();
+      pendingTelegramMessagesByChat.clear();
+      outboundEvents.length = 0;
+    },
+    getSnapshot: baseState.getSnapshot.bind(baseState),
+    async getOutboundEvents() {
+      return outboundEvents;
+    },
+    observeEvent(event) {
       if (params.adapter.channel === "telegram") {
         const lifecycle = readTelegramLifecycleEvent({
           cursor: outboundEvents.length + 1,
-          event: parsed,
+          event,
           messageByProviderId: telegramMessageByProviderId,
           pendingByChat: pendingTelegramMessagesByChat,
         });
@@ -251,50 +238,12 @@ function createCrablineState(params: {
         }
       }
       const outbound = params.adapter.createOutboundFromRecorderEvent({
-        event: parsed,
+        event,
         targetByProviderTarget,
       }) as QaBusOutboundMessageInput | null;
       if (outbound) {
         baseState.addOutboundMessage(outbound);
       }
-    }
-    recorderLineCursor = lines.length;
-  };
-
-  const syncRecorder = async () => {
-    if (syncPromise) {
-      return await syncPromise;
-    }
-    syncPromise = syncRecorderSnapshot({ allowIncompleteTail: true });
-    try {
-      await syncPromise;
-    } finally {
-      syncPromise = null;
-    }
-  };
-
-  const interval = setInterval(() => {
-    void syncRecorder().catch(() => undefined);
-  }, RECORDER_SYNC_INTERVAL_MS);
-  interval.unref?.();
-
-  return {
-    async reset() {
-      await syncRecorder();
-      baseState.reset();
-      targetByProviderTarget.clear();
-      telegramMessageByProviderId.clear();
-      pendingTelegramMessagesByChat.clear();
-      outboundEvents.length = 0;
-      recorderLineCursor = await fs
-        .readFile(params.adapter.manifest.recorderPath, "utf8")
-        .then((text) => readRecorderLines(text, { allowIncompleteTail: true }).length)
-        .catch(() => 0);
-    },
-    getSnapshot: baseState.getSnapshot.bind(baseState),
-    async getOutboundEvents() {
-      await syncRecorder();
-      return outboundEvents;
     },
     async addInboundMessage(input: QaBusInboundMessageInput) {
       const providerInbound = params.adapter.createInbound({ input });
@@ -315,21 +264,10 @@ function createCrablineState(params: {
     },
     addOutboundMessage: baseState.addOutboundMessage.bind(baseState),
     readMessage: baseState.readMessage.bind(baseState),
-    async searchMessages(input: QaBusSearchMessagesInput) {
-      await syncRecorder();
-      return baseState.searchMessages(input);
-    },
-    async waitFor(input: QaBusWaitForInput) {
-      await syncRecorder();
-      return await baseState.waitFor(input);
-    },
+    searchMessages: baseState.searchMessages.bind(baseState),
+    waitFor: baseState.waitFor.bind(baseState),
     async cleanup() {
-      clearInterval(interval);
       await params.adapter.close();
-      if (syncPromise) {
-        await syncPromise;
-      }
-      await syncRecorderSnapshot({ allowIncompleteTail: false });
     },
   };
 }
@@ -429,8 +367,10 @@ export async function createQaCrablineTransportAdapter(params: {
     `${params.selection.channel}-fake-provider.jsonl`,
   );
   await fs.mkdir(path.dirname(recorderPath), { recursive: true });
+  let observeEvent = (_event: unknown) => {};
   const adapter = await startOpenClawCrablineAdapter({
     channel: params.selection.channel,
+    onEvent: (event) => observeEvent(event),
     openclawConfig: {},
     recorderPath,
   });
@@ -440,12 +380,10 @@ export async function createQaCrablineTransportAdapter(params: {
     "utf8",
   );
 
-  return new QaCrablineTransport({
+  const state = createCrablineState({
     adapter,
-    selection: params.selection,
-    state: createCrablineState({
-      adapter,
-      state: params.state ?? createQaBusState(),
-    }),
+    state: params.state ?? createQaBusState(),
   });
+  observeEvent = state.observeEvent;
+  return new QaCrablineTransport({ adapter, selection: params.selection, state });
 }

pnpm-lock.yaml

@@ -1350,8 +1350,8 @@ importers:
         version: 4.4.3
     devDependencies:
       '@openclaw/crabline':
-        specifier: 0.1.8
-        version: 0.1.8
+        specifier: 0.1.9
+        version: 0.1.9
       '@openclaw/discord':
         specifier: workspace:*
         version: link:../discord
@@ -3165,8 +3165,8 @@ packages:
     cpu: [x64]
     os: [win32]
 
-  '@openclaw/crabline@0.1.8':
-    resolution: {integrity: sha512-a6r4vkPEsDaMdznLVcxiVdKA3oI3k2IUU0sJjCd699pFWR401qfvvQMbgFOb403eA7/vaRkpN0E9tBbFycl6Ug==}
+  '@openclaw/crabline@0.1.9':
+    resolution: {integrity: sha512-i3RZ/qAMM1dnNEaj4KwcO1iJRXJIrvfTRIXdMiy7SgZh8A7EUCFNPNviC3URLirX9wIYrkuVTyk9QHKOFPcz7g==}
     engines: {node: '>=22'}
     hasBin: true
 
@@ -9198,7 +9198,7 @@ snapshots:
   '@openai/codex@0.142.4-win32-x64':
     optional: true
 
-  '@openclaw/crabline@0.1.8':
+  '@openclaw/crabline@0.1.9':
     dependencies:
       commander: 15.0.0
       curve25519-js: 0.0.4

pnpm-workspace.yaml

@@ -7,7 +7,7 @@ packages:
 minimumReleaseAge: 2880
 
 minimumReleaseAgeExclude:
-  - "@openclaw/crabline@0.1.8"
+  - "@openclaw/crabline@0.1.9"
   - "@openclaw/fs-safe@0.3.0"
   - "@openclaw/proxyline@0.3.3"
   - "acpx"