Skip to content

Commit 37aaa5c

Browse files
omarshahineomarshahine
authored
fix(imessage): frame rpc stdout on LF only (#90845)
Merged via squash. Prepared head SHA: c62a2dc Co-authored-by: omarshahine <10343873+omarshahine@users.noreply.github.com> Co-authored-by: omarshahine <10343873+omarshahine@users.noreply.github.com> Reviewed-by: @omarshahine
1 parent ab7c922 commit 37aaa5c

2 files changed

Lines changed: 157 additions & 9 deletions

File tree

‎extensions/imessage/src/client.ts‎

Lines changed: 45 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
// Imessage plugin module implements client behavior.
22
import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process";
3-
import { createInterface, type Interface } from "node:readline";
3+
import { StringDecoder } from "node:string_decoder";
44
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
55
import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
66
import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime";
@@ -68,7 +68,8 @@ export class IMessageRpcClient {
6868
private readonly closed: Promise<void>;
6969
private closedResolve: (() => void) | null = null;
7070
private child: ChildProcessWithoutNullStreams | null = null;
71-
private reader: Interface | null = null;
71+
private stdoutBuffer = "";
72+
private readonly stdoutDecoder = new StringDecoder("utf8");
7273
private nextId = 1;
7374
private publicProcessError: string | null = null;
7475

@@ -97,14 +98,12 @@ export class IMessageRpcClient {
9798
stdio: ["pipe", "pipe", "pipe"],
9899
});
99100
this.child = child;
100-
this.reader = createInterface({ input: child.stdout });
101101

102-
this.reader.on("line", (line) => {
103-
const trimmed = line.trim();
104-
if (!trimmed) {
102+
child.stdout.on("data", (chunk) => {
103+
if (this.child !== child) {
105104
return;
106105
}
107-
this.handleLine(trimmed);
106+
this.handleStdoutChunk(chunk);
108107
});
109108

110109
child.stderr?.on("data", (chunk) => {
@@ -131,6 +130,9 @@ export class IMessageRpcClient {
131130
});
132131

133132
child.on("close", (code, signal) => {
133+
if (this.child === child) {
134+
this.flushStdoutBuffer();
135+
}
134136
this.failAll(this.buildCloseError(code, signal));
135137
this.closedResolve?.();
136138
});
@@ -140,8 +142,8 @@ export class IMessageRpcClient {
140142
if (!this.child) {
141143
return;
142144
}
143-
this.reader?.close();
144-
this.reader = null;
145+
this.stdoutBuffer = "";
146+
this.stdoutDecoder.end();
145147
this.child.stdin?.end();
146148
const child = this.child;
147149
this.child = null;
@@ -215,6 +217,40 @@ export class IMessageRpcClient {
215217
return await response;
216218
}
217219

220+
private handleStdoutChunk(chunk: Buffer | string) {
221+
const text = typeof chunk === "string" ? chunk : this.stdoutDecoder.write(chunk);
222+
this.stdoutBuffer += text;
223+
224+
let newlineIndex = this.stdoutBuffer.indexOf("\n");
225+
while (newlineIndex !== -1) {
226+
const line = this.stdoutBuffer.slice(0, newlineIndex);
227+
this.stdoutBuffer = this.stdoutBuffer.slice(newlineIndex + 1);
228+
this.handleStdoutLine(line);
229+
newlineIndex = this.stdoutBuffer.indexOf("\n");
230+
}
231+
}
232+
233+
private flushStdoutBuffer() {
234+
const tail = this.stdoutDecoder.end();
235+
if (tail) {
236+
this.stdoutBuffer += tail;
237+
}
238+
if (!this.stdoutBuffer) {
239+
return;
240+
}
241+
const line = this.stdoutBuffer;
242+
this.stdoutBuffer = "";
243+
this.handleStdoutLine(line);
244+
}
245+
246+
private handleStdoutLine(line: string) {
247+
const trimmed = line.trim();
248+
if (!trimmed) {
249+
return;
250+
}
251+
this.handleLine(trimmed);
252+
}
253+
218254
private handleLine(line: string) {
219255
let parsed: IMessageRpcResponse<unknown>;
220256
try {

‎extensions/imessage/src/status.test.ts‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
11
// Imessage tests cover status plugin behavior.
2+
import { EventEmitter } from "node:events";
3+
import { PassThrough } from "node:stream";
24
import { createPluginSetupWizardStatus } from "openclaw/plugin-sdk/plugin-test-runtime";
35
import * as processRuntime from "openclaw/plugin-sdk/process-runtime";
46
import * as setupRuntime from "openclaw/plugin-sdk/setup";
@@ -20,6 +22,26 @@ const getIMessageSetupStatus = createPluginSetupWizardStatus({
2022

2123
const spawnMock = vi.hoisted(() => vi.fn());
2224

25+
function createMockChildProcess() {
26+
const child = new EventEmitter() as EventEmitter & {
27+
stdin: PassThrough;
28+
stdout: PassThrough;
29+
stderr: PassThrough;
30+
killed: boolean;
31+
kill: (signal?: string) => boolean;
32+
};
33+
child.stdin = new PassThrough();
34+
child.stdout = new PassThrough();
35+
child.stderr = new PassThrough();
36+
child.killed = false;
37+
child.kill = (signal?: string) => {
38+
child.killed = true;
39+
child.emit("close", 0, signal ?? null);
40+
return true;
41+
};
42+
return child;
43+
}
44+
2345
vi.mock("node:child_process", async () => {
2446
const actual = await vi.importActual<typeof import("node:child_process")>("node:child_process");
2547
return {
@@ -67,6 +89,96 @@ describe("createIMessageRpcClient", () => {
6789

6890
expect(internals.buildCloseError(1, null).message).toBe(PUBLIC_IMESSAGE_FULL_DISK_ACCESS_ERROR);
6991
});
92+
93+
it.each([
94+
["U+2028", "\u2028"],
95+
["U+2029", "\u2029"],
96+
])(
97+
"frames stdout on LF only so raw %s inside JSON strings stays intact",
98+
async (_, separator) => {
99+
const { IMessageRpcClient } = await import("./client.js");
100+
const client = new IMessageRpcClient();
101+
const internals = client as unknown as {
102+
handleStdoutChunk: (chunk: Buffer | string) => void;
103+
pending: Map<
104+
string,
105+
{
106+
resolve: (value: unknown) => void;
107+
reject: (error: Error) => void;
108+
}
109+
>;
110+
};
111+
const result = new Promise((resolve, reject) => {
112+
internals.pending.set("1", { resolve, reject });
113+
});
114+
const text = `line one${separator}line two`;
115+
const payload = `${JSON.stringify({
116+
jsonrpc: "2.0",
117+
id: 1,
118+
result: { messages: [{ text }] },
119+
})}\n`;
120+
const bytes = Buffer.from(payload, "utf8");
121+
const separatorIndex = bytes.indexOf(Buffer.from(separator, "utf8"));
122+
123+
internals.handleStdoutChunk(bytes.subarray(0, separatorIndex + 1));
124+
internals.handleStdoutChunk(bytes.subarray(separatorIndex + 1));
125+
126+
await expect(result).resolves.toEqual({
127+
messages: [{ text }],
128+
});
129+
},
130+
);
131+
132+
it("handles multiple LF-delimited stdout responses in one chunk", async () => {
133+
const { IMessageRpcClient } = await import("./client.js");
134+
const client = new IMessageRpcClient();
135+
const internals = client as unknown as {
136+
handleStdoutChunk: (chunk: Buffer | string) => void;
137+
pending: Map<
138+
string,
139+
{
140+
resolve: (value: unknown) => void;
141+
reject: (error: Error) => void;
142+
}
143+
>;
144+
};
145+
const first = new Promise((resolve, reject) => {
146+
internals.pending.set("1", { resolve, reject });
147+
});
148+
const second = new Promise((resolve, reject) => {
149+
internals.pending.set("2", { resolve, reject });
150+
});
151+
152+
internals.handleStdoutChunk(
153+
`${JSON.stringify({ jsonrpc: "2.0", id: 1, result: { ok: "first" } })}\n${JSON.stringify({
154+
jsonrpc: "2.0",
155+
id: 2,
156+
result: { ok: "second" },
157+
})}\n`,
158+
);
159+
160+
await expect(first).resolves.toEqual({ ok: "first" });
161+
await expect(second).resolves.toEqual({ ok: "second" });
162+
});
163+
164+
it("ignores stdout from a stale child after stop so late notifications cannot leak (#89830)", async () => {
165+
vi.stubEnv("VITEST", "");
166+
vi.stubEnv("NODE_ENV", "");
167+
const child = createMockChildProcess();
168+
spawnMock.mockReturnValue(child);
169+
const onNotification = vi.fn();
170+
const { IMessageRpcClient } = await import("./client.js");
171+
const client = new IMessageRpcClient({ onNotification });
172+
173+
await client.start();
174+
await client.stop();
175+
176+
// A not-yet-exited imsg child emits a complete notification after stop().
177+
// The `this.child !== child` guard must drop it before handleStdoutChunk.
178+
child.stdout.write('{"jsonrpc":"2.0","method":"messages.changed","params":{}}\n');
179+
180+
expect(onNotification).not.toHaveBeenCalled();
181+
});
70182
});
71183

72184
describe("imessage setup status", () => {

0 commit comments

Comments
 (0)