diff --git a/extensions/qa-lab/package.json b/extensions/qa-lab/package.json index bd65f53071f..dabd2c1bab5 100644 --- a/extensions/qa-lab/package.json +++ b/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": { diff --git a/extensions/qa-lab/src/crabline-transport.test.ts b/extensions/qa-lab/src/crabline-transport.test.ts index 59b2258657a..8fb612d6ed4 100644 --- a/extensions/qa-lab/src/crabline-transport.test.ts +++ b/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/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, diff --git a/extensions/qa-lab/src/crabline-transport.ts b/extensions/qa-lab/src/crabline-transport.ts index 38466f873a0..d12c157c167 100644 --- a/extensions/qa-lab/src/crabline-transport.ts +++ b/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; getOutboundEvents: () => Promise; + 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(); const pendingTelegramMessagesByChat = new Map(); const outboundEvents: QaTransportOutboundEvent[] = []; - let recorderLineCursor = 0; - let syncPromise: Promise | 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 }); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 04256c60cca..34613779cbd 100644 --- a/pnpm-lock.yaml +++ b/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 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index b2a752874b2..6a9d66fb460 100644 --- a/pnpm-workspace.yaml +++ b/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"