Skip to content

Commit 1cab479

Browse files
RomneyDaRomneyDa
authored
fix(qa): defer partial Crabline recorder rows (#99649)
1 parent f0da000 commit 1cab479

2 files changed

Lines changed: 127 additions & 35 deletions

File tree

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

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,83 @@ describe("crabline transport", () => {
103103
});
104104
});
105105

106+
it("defers a partial recorder tail until Crabline finishes the JSONL record", async () => {
107+
await withTempDir("qa-crabline-transport-", async (outputDir) => {
108+
const transport = await createQaCrablineTransportAdapter({
109+
outputDir,
110+
selection: createSelection(),
111+
state: createQaBusState(),
112+
});
113+
const manifest = JSON.parse(
114+
await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"),
115+
) as { recorderPath: string };
116+
const recorderLine = JSON.stringify({
117+
body: {
118+
chat_id: "100001",
119+
text: `partial recorder write ${"x".repeat(240)}`,
120+
},
121+
path: "/bot<redacted>/sendMessage",
122+
type: "api",
123+
});
124+
const splitIndex = 191;
125+
126+
try {
127+
await transport.reset();
128+
await fs.appendFile(manifest.recorderPath, recorderLine.slice(0, splitIndex), "utf8");
129+
130+
await expect(transport.reset()).resolves.toBeUndefined();
131+
132+
await fs.appendFile(manifest.recorderPath, `${recorderLine.slice(splitIndex)}\n`, "utf8");
133+
await expect(
134+
transport.state.searchMessages({ query: "partial recorder write" }),
135+
).resolves.toHaveLength(1);
136+
} finally {
137+
await transport.cleanup?.();
138+
}
139+
});
140+
});
141+
142+
it("rejects a malformed newline-terminated recorder record", async () => {
143+
await withTempDir("qa-crabline-transport-", async (outputDir) => {
144+
const transport = await createQaCrablineTransportAdapter({
145+
outputDir,
146+
selection: createSelection(),
147+
state: createQaBusState(),
148+
});
149+
const manifest = JSON.parse(
150+
await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"),
151+
) as { recorderPath: string };
152+
153+
try {
154+
await transport.reset();
155+
await fs.appendFile(manifest.recorderPath, '{"broken":\n', "utf8");
156+
157+
await expect(transport.reset()).rejects.toThrow(SyntaxError);
158+
await fs.writeFile(manifest.recorderPath, "", "utf8");
159+
} finally {
160+
await transport.cleanup?.();
161+
}
162+
});
163+
});
164+
165+
it("rejects a permanently truncated recorder tail during cleanup", async () => {
166+
await withTempDir("qa-crabline-transport-", async (outputDir) => {
167+
const transport = await createQaCrablineTransportAdapter({
168+
outputDir,
169+
selection: createSelection(),
170+
state: createQaBusState(),
171+
});
172+
const manifest = JSON.parse(
173+
await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"),
174+
) as { recorderPath: string };
175+
176+
await transport.reset();
177+
await fs.appendFile(manifest.recorderPath, '{"unfinished":"recorder tail', "utf8");
178+
179+
await expect(transport.cleanup?.()).rejects.toThrow(SyntaxError);
180+
});
181+
});
182+
106183
it("observes Telegram preview edits through the shared transport adapter", async () => {
107184
await withTempDir("qa-crabline-transport-", async (outputDir) => {
108185
const transport = await createQaCrablineTransportAdapter({

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

Lines changed: 50 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,16 @@ type QaCrablineTransportState = QaTransportState & {
5252

5353
const TELEGRAM_LIFECYCLE_METHOD_RE = /\/(sendMessage|editMessageText|deleteMessage)$/u;
5454

55+
function readRecorderLines(text: string, options: { allowIncompleteTail: boolean }): string[] {
56+
// Crabline appends each JSONL record asynchronously, so a concurrent read can end mid-record.
57+
// Defer only the unterminated tail; newline-terminated malformed records must still fail parsing.
58+
const lines = text.split(/\r?\n/u);
59+
if (options.allowIncompleteTail && !text.endsWith("\n")) {
60+
lines.pop();
61+
}
62+
return lines.filter((line) => line.trim().length > 0);
63+
}
64+
5565
function readTelegramLifecycleEvent(params: {
5666
cursor: number;
5767
event: unknown;
@@ -217,43 +227,45 @@ function createCrablineState(params: {
217227
let recorderLineCursor = 0;
218228
let syncPromise: Promise<void> | null = null;
219229

220-
const syncRecorder = async () => {
221-
if (syncPromise) {
222-
return await syncPromise;
223-
}
224-
syncPromise = (async () => {
225-
const text = await fs
226-
.readFile(params.adapter.manifest.recorderPath, "utf8")
227-
.catch((error: unknown) => {
228-
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
229-
return "";
230-
}
231-
throw error;
232-
});
233-
const lines = text.split(/\r?\n/u).filter((line) => line.trim().length > 0);
234-
for (const line of lines.slice(recorderLineCursor)) {
235-
const parsed = JSON.parse(line) as unknown;
236-
if (params.adapter.channel === "telegram") {
237-
const lifecycle = readTelegramLifecycleEvent({
238-
cursor: outboundEvents.length + 1,
239-
event: parsed,
240-
messageByProviderId: telegramMessageByProviderId,
241-
pendingByChat: pendingTelegramMessagesByChat,
242-
});
243-
if (lifecycle) {
244-
outboundEvents.push(lifecycle);
245-
}
230+
const syncRecorderSnapshot = async (options: { allowIncompleteTail: boolean }) => {
231+
const text = await fs
232+
.readFile(params.adapter.manifest.recorderPath, "utf8")
233+
.catch((error: unknown) => {
234+
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
235+
return "";
246236
}
247-
const outbound = params.adapter.createOutboundFromRecorderEvent({
237+
throw error;
238+
});
239+
const lines = readRecorderLines(text, options);
240+
for (const line of lines.slice(recorderLineCursor)) {
241+
const parsed = JSON.parse(line) as unknown;
242+
if (params.adapter.channel === "telegram") {
243+
const lifecycle = readTelegramLifecycleEvent({
244+
cursor: outboundEvents.length + 1,
248245
event: parsed,
249-
targetByProviderTarget,
250-
}) as QaBusOutboundMessageInput | null;
251-
if (outbound) {
252-
baseState.addOutboundMessage(outbound);
246+
messageByProviderId: telegramMessageByProviderId,
247+
pendingByChat: pendingTelegramMessagesByChat,
248+
});
249+
if (lifecycle) {
250+
outboundEvents.push(lifecycle);
253251
}
254252
}
255-
recorderLineCursor = lines.length;
256-
})();
253+
const outbound = params.adapter.createOutboundFromRecorderEvent({
254+
event: parsed,
255+
targetByProviderTarget,
256+
}) as QaBusOutboundMessageInput | null;
257+
if (outbound) {
258+
baseState.addOutboundMessage(outbound);
259+
}
260+
}
261+
recorderLineCursor = lines.length;
262+
};
263+
264+
const syncRecorder = async () => {
265+
if (syncPromise) {
266+
return await syncPromise;
267+
}
268+
syncPromise = syncRecorderSnapshot({ allowIncompleteTail: true });
257269
try {
258270
await syncPromise;
259271
} finally {
@@ -276,7 +288,7 @@ function createCrablineState(params: {
276288
outboundEvents.length = 0;
277289
recorderLineCursor = await fs
278290
.readFile(params.adapter.manifest.recorderPath, "utf8")
279-
.then((text) => text.split(/\r?\n/u).filter((line) => line.trim().length > 0).length)
291+
.then((text) => readRecorderLines(text, { allowIncompleteTail: true }).length)
280292
.catch(() => 0);
281293
},
282294
getSnapshot: baseState.getSnapshot.bind(baseState),
@@ -313,8 +325,11 @@ function createCrablineState(params: {
313325
},
314326
async cleanup() {
315327
clearInterval(interval);
316-
await syncRecorder();
317328
await params.adapter.close();
329+
if (syncPromise) {
330+
await syncPromise;
331+
}
332+
await syncRecorderSnapshot({ allowIncompleteTail: false });
318333
},
319334
};
320335
}

0 commit comments

Comments
 (0)