fix(sessions): preserve UTF-8 across tail reads (#108151)

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("<22>");
+  });
+
+  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("<22>");
+  });
+
+  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("<22>");
+  });
+
   it("continues following when SQLite trajectory rows are appended", async () => {
     const runtime = makeRuntime();
     await writeSessionEntry(sessionKey, {

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<TailTrajectorySource, { kind: "file" }>;
       },