diff --git a/src/commands/sessions-tail.test.ts b/src/commands/sessions-tail.test.ts index 579ab800088a..0f312fb73014 100644 --- a/src/commands/sessions-tail.test.ts +++ b/src/commands/sessions-tail.test.ts @@ -391,6 +391,154 @@ describe("sessionsTailCommand", () => { expect(output).toContain("python ok"); }); + it("preserves UTF-8 characters split across follow reads", async () => { + const runtime = makeRuntime(); + await writeSessionEntry(); + await appendEvents([ + makeEvent({ + sourceSeq: 1, + type: "session.started", + ts: "2026-05-18T12:04:17.000Z", + }), + ]); + const appendedEvent = makeEvent({ + sourceSeq: 2, + type: "prompt.skipped", + ts: "2026-05-18T12:04:21.000Z", + data: { reason: "猫" }, + }); + + const run = sessionsTailCommand( + { store: storePath, sessionKey, tail: "1", follow: true }, + runtime, + ); + try { + await waitForRuntimeOutput(runtime, "session.started"); + const line = Buffer.from(`${JSON.stringify(appendedEvent)}\n`, "utf8"); + const marker = Buffer.from("猫", "utf8"); + const markerOffset = line.indexOf(marker); + expect(markerOffset).toBeGreaterThanOrEqual(0); + + fs.appendFileSync(trajectoryPath, line.subarray(0, markerOffset + 1)); + await new Promise((resolve) => { + setTimeout(resolve, 100); + }); + fs.appendFileSync(trajectoryPath, line.subarray(markerOffset + 1)); + await waitForRuntimeOutput(runtime, "prompt skipped"); + } finally { + process.emit("SIGTERM", "SIGTERM"); + await run; + } + + const output = runtimeOutput(runtime); + expect(output).toContain("prompt skipped: 猫"); + expect(output).not.toContain("�"); + }); + + it("preserves UTF-8 characters split before follow starts", async () => { + const runtime = makeRuntime(); + await writeSessionEntry(); + await appendEvents([ + makeEvent({ + sourceSeq: 1, + type: "session.started", + ts: "2026-05-18T12:04:17.000Z", + }), + ]); + const appendedEvent = makeEvent({ + sourceSeq: 2, + type: "prompt.skipped", + ts: "2026-05-18T12:04:21.000Z", + data: { reason: "猫" }, + }); + const line = Buffer.from(`${JSON.stringify(appendedEvent)}\n`, "utf8"); + const markerOffset = line.indexOf(Buffer.from("猫", "utf8")); + expect(markerOffset).toBeGreaterThanOrEqual(0); + fs.appendFileSync(trajectoryPath, line.subarray(0, markerOffset + 1)); + + const run = sessionsTailCommand( + { store: storePath, sessionKey, tail: "1", follow: true }, + runtime, + ); + try { + await waitForRuntimeOutput(runtime, "session.started"); + fs.appendFileSync(trajectoryPath, line.subarray(markerOffset + 1)); + await waitForRuntimeOutput(runtime, "prompt skipped"); + } finally { + process.emit("SIGTERM", "SIGTERM"); + await run; + } + + const output = runtimeOutput(runtime); + expect(output).toContain("prompt skipped: 猫"); + expect(output).not.toContain("�"); + }); + + it("resets split UTF-8 state when the followed file is replaced", async () => { + const runtime = makeRuntime(); + await writeSessionEntry(); + const startedEvent = makeEvent({ + sourceSeq: 1, + type: "session.started", + ts: "2026-05-18T12:04:17.000Z", + }); + await appendEvents([startedEvent]); + + const run = sessionsTailCommand( + { store: storePath, sessionKey, tail: "1", follow: true }, + runtime, + ); + try { + await waitForRuntimeOutput(runtime, "session.started"); + const partialLine = Buffer.from( + `${JSON.stringify( + makeEvent({ + sourceSeq: 2, + type: "prompt.skipped", + ts: "2026-05-18T12:04:20.000Z", + data: { reason: "猫" }, + }), + )}\n`, + "utf8", + ); + const markerOffset = partialLine.indexOf(Buffer.from("猫", "utf8")); + expect(markerOffset).toBeGreaterThanOrEqual(0); + fs.appendFileSync(trajectoryPath, partialLine.subarray(0, markerOffset + 1)); + await new Promise((resolve) => { + setTimeout(resolve, 100); + }); + + const replacementPath = `${trajectoryPath}.replacement`; + writeJsonl(replacementPath, [ + startedEvent, + makeEvent({ + sourceSeq: 3, + type: "tool.result", + ts: "2026-05-18T12:04:21.000Z", + data: { name: "rotation", success: true }, + }), + ]); + fs.renameSync(replacementPath, trajectoryPath); + await waitForRuntimeOutput(runtime, "rotation ok"); + + appendJsonl( + trajectoryPath, + makeEvent({ + sourceSeq: 4, + type: "prompt.skipped", + ts: "2026-05-18T12:04:22.000Z", + data: { reason: "clean" }, + }), + ); + await waitForRuntimeOutput(runtime, "prompt skipped: clean"); + } finally { + process.emit("SIGTERM", "SIGTERM"); + await run; + } + + expect(runtimeOutput(runtime)).not.toContain("�"); + }); + it("continues following when SQLite trajectory rows are appended", async () => { const runtime = makeRuntime(); await writeSessionEntry(sessionKey, { diff --git a/src/commands/sessions-tail.ts b/src/commands/sessions-tail.ts index c54b3e0df2b5..cad799036a5c 100644 --- a/src/commands/sessions-tail.ts +++ b/src/commands/sessions-tail.ts @@ -6,6 +6,7 @@ */ import fs from "node:fs"; import path from "node:path"; +import { StringDecoder } from "node:string_decoder"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { readAcpSessionMeta } from "../acp/runtime/session-meta.js"; import { getRuntimeConfig } from "../config/config.js"; @@ -60,6 +61,7 @@ type TailTrajectorySource = type FileFollowState = { cursor: TrajectoryCursor | null; + decoder: StringDecoder; fileState: FollowFileState | null; kind: "file"; offset: number; @@ -78,6 +80,8 @@ type FollowState = FileFollowState | SqliteFollowState; type TrajectorySnapshot = { events: TrajectoryEvent[]; + fileDecoder?: StringDecoder; + filePending?: string; fileState: FollowFileState | null; maxStorageSeq?: number; offset: number; @@ -308,11 +312,16 @@ function readTrajectorySnapshot(filePath: string): TrajectorySnapshot { filePath, maxBytes: TRAJECTORY_RUNTIME_FILE_MAX_BYTES, }); - const text = buffer.toString("utf8"); + const fileDecoder = new StringDecoder("utf8"); + const lines = fileDecoder.write(buffer).split(/\r?\n/u); + const trailing = lines.pop() ?? ""; + const trailingEvent = parseTrajectoryEventLine(trailing); return { - events: parseTrajectoryEventLines(text.split(/\r?\n/u)), + events: [...parseTrajectoryEventLines(lines), ...(trailingEvent ? [trailingEvent] : [])], + fileDecoder, + filePending: trailingEvent ? "" : trailing, fileState: fileStateFromStat(stat), - offset: Buffer.byteLength(text, "utf8"), + offset: buffer.length, }; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") { @@ -492,6 +501,7 @@ function statFileSize(filePath: string): number { function readNewFileFollowEvents(state: FileFollowState): TrajectoryEvent[] { const fileState = readFollowFileState(state.selection.source.path); if (!fileState) { + state.decoder = new StringDecoder("utf8"); state.fileState = null; state.offset = 0; state.pending = ""; @@ -507,9 +517,10 @@ function readNewFileFollowEvents(state: FileFollowState): TrajectoryEvent[] { // Log rotation, truncation, and same-size rewrites all require a full // rescan; cursor filtering prevents duplicate event output. const snapshot = readTrajectorySnapshot(state.selection.source.path); + state.decoder = snapshot.fileDecoder ?? new StringDecoder("utf8"); state.fileState = snapshot.fileState; state.offset = snapshot.offset; - state.pending = ""; + state.pending = snapshot.filePending ?? ""; return eventsAfterCursor(snapshot.events, state.cursor); } @@ -530,7 +541,7 @@ function readNewFileFollowEvents(state: FileFollowState): TrajectoryEvent[] { fs.readSync(fd, buffer, 0, buffer.length, state.offset); state.offset = fileState.size; state.fileState = fileState; - const combined = `${state.pending}${buffer.toString("utf8")}`; + const combined = `${state.pending}${state.decoder.write(buffer)}`; // Keep an incomplete trailing JSON line until the next poll, matching // append-only writers that flush in chunks. const lines = combined.split(/\r?\n/u); @@ -591,10 +602,11 @@ async function followSelections( } return { cursor: snapshot ? maxCursorFromEvents(snapshot.events) : null, + decoder: snapshot?.fileDecoder ?? new StringDecoder("utf8"), fileState: snapshot?.fileState ?? readFollowFileState(selection.source.path), kind: "file", offset: snapshot?.offset ?? statFileSize(selection.source.path), - pending: "", + pending: snapshot?.filePending ?? "", selection: selection as TailSelection & { source: Extract; },