Keep words intact in followed agent logs (#6528)

* fix(cli): keep streamed message fragments together in logs --follow

`paseo logs` without `-f` fetches the timeline projected, so
`projectTimelineRows` merges consecutive assistant and reasoning fragments by
exact concatenation before the transcript is rendered. Follow mode never got
there: it called `formatAgentActivityTranscript([item])` per arriving event, so
every fragment was projected alone and printed on its own line. Words broke at
fragment boundaries, a reader could not tell an author newline from a transport
one, and `[Thought]` was reprinted for each reasoning chunk.

Follow mode now keeps a fragment chain open using the same merge rule the
projection uses (same kind, same `turnId`, matching `messageId` when the event
carries one) and writes the text as it arrives instead of one line per event.
Only trailing whitespace is held back, so a message still ends trimmed the way
the one-shot transcript ends it, and the chain closes when another kind of item
arrives or the follow stops.

* test(cli): place daemon follow regression in its owning suite

* fix(cli): preserve filtered log boundaries and run the follow regression in CI

* test(cli): wait for a completed log line before terminating follow

---------

Co-authored-by: Lorenzo Fiore <fiorelorenzo.fl@gmail.com>
Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

packages/cli/src/commands/agent/logs.test.ts

@@ -0,0 +1,139 @@
+import { describe, expect, it } from "vitest";
+import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types";
+import { createFollowTranscriptWriter, formatAgentActivityTranscript } from "./logs.js";
+
+function collect(items: readonly AgentTimelineItem[], turnId?: string): string {
+  let output = "";
+  const writer = createFollowTranscriptWriter((chunk) => {
+    output += chunk;
+  });
+  for (const item of items) {
+    writer.push(item, turnId);
+  }
+  writer.end();
+  return output;
+}
+
+describe("createFollowTranscriptWriter", () => {
+  it("keeps a word that arrives split across fragments intact", () => {
+    const output = collect([
+      { type: "assistant_message", text: "the paseo-pr", messageId: "msg_1" },
+      { type: "assistant_message", text: "-cycle label\n", messageId: "msg_1" },
+    ]);
+
+    expect(output).toBe("the paseo-pr-cycle label\n");
+  });
+
+  it("streams text as it arrives without inventing a line break", () => {
+    const chunks: string[] = [];
+    const writer = createFollowTranscriptWriter((chunk) => {
+      chunks.push(chunk);
+    });
+
+    writer.push({
+      type: "assistant_message",
+      text: "first line",
+      messageId: "msg_1",
+    });
+    expect(chunks.join("")).toBe("first line");
+
+    writer.push({
+      type: "assistant_message",
+      text: "\nsecond",
+      messageId: "msg_1",
+    });
+    expect(chunks.join("")).toBe("first line\nsecond");
+
+    writer.push({
+      type: "assistant_message",
+      text: " line\n",
+      messageId: "msg_1",
+    });
+    expect(chunks.join("")).toBe("first line\nsecond line");
+
+    writer.end();
+    expect(chunks.join("")).toBe("first line\nsecond line\n");
+  });
+
+  it("emits the same transcript the one-shot command prints for the same items", () => {
+    const items: AgentTimelineItem[] = [
+      {
+        type: "assistant_message",
+        text: "Plan for paseo-pr",
+        messageId: "msg_1",
+      },
+      {
+        type: "assistant_message",
+        text: "-cycle:\n- build\n- test\n",
+        messageId: "msg_1",
+      },
+      { type: "error", message: "boom" },
+      { type: "assistant_message", text: "done", messageId: "msg_2" },
+    ];
+
+    expect(collect(items).trimEnd()).toBe(formatAgentActivityTranscript(items));
+  });
+
+  it("starts a new message when the messageId changes", () => {
+    const output = collect([
+      { type: "assistant_message", text: "first", messageId: "msg_1" },
+      { type: "assistant_message", text: "second", messageId: "msg_2" },
+    ]);
+
+    expect(output).toBe("first\nsecond\n");
+  });
+
+  it("does not merge fragments across turns", () => {
+    let output = "";
+    const writer = createFollowTranscriptWriter((chunk) => {
+      output += chunk;
+    });
+
+    writer.push({ type: "assistant_message", text: "first" }, "turn_1");
+    writer.push({ type: "assistant_message", text: "second" }, "turn_2");
+    writer.end();
+
+    expect(output).toBe("first\nsecond\n");
+  });
+
+  it("prints the thought prefix once for a merged reasoning stream", () => {
+    const output = collect([
+      { type: "reasoning", text: "checking the" },
+      { type: "reasoning", text: " lockfile\nthen the build\n" },
+    ]);
+
+    expect(output).toBe("[Thought] checking the lockfile\nthen the build\n");
+  });
+
+  it("keeps reasoning blocks separate across hidden timeline items", () => {
+    let output = "";
+    const writer = createFollowTranscriptWriter((chunk) => {
+      output += chunk;
+    }, "reasoning");
+    writer.push({ type: "reasoning", text: "checking files" });
+    writer.push({
+      type: "tool_call",
+      callId: "tool_1",
+      name: "Bash",
+      status: "completed",
+      error: null,
+      detail: { type: "unknown", input: "ls", output: "files" },
+    });
+    writer.push({ type: "reasoning", text: "running tests" });
+    writer.end();
+
+    expect(output).toBe("[Thought] checking files\n[Thought] running tests\n");
+  });
+
+  it("keeps interior blank lines and drops trailing whitespace", () => {
+    const output = collect([
+      {
+        type: "assistant_message",
+        text: "  # Title\n\nBody\n\n  ",
+        messageId: "msg_1",
+      },
+    ]);
+
+    expect(output).toBe("# Title\n\nBody\n");
+  });
+});

packages/cli/src/commands/agent/logs.ts

@@ -70,6 +70,134 @@ export function formatAgentActivityTranscript(
   );
 }
 
+type StreamingTextType = "assistant_message" | "reasoning";
+
+const THOUGHT_PREFIX = "[Thought] ";
+
+interface PendingStreamingText {
+  type: StreamingTextType;
+  turnId: string | undefined;
+  messageId: string | undefined;
+  buffer: string;
+  emitted: boolean;
+}
+
+export interface FollowTranscriptWriter {
+  push(item: AgentTimelineItem, turnId?: string): void;
+  end(): void;
+}
+
+function lastNonWhitespaceIndex(text: string): number {
+  for (let index = text.length - 1; index >= 0; index -= 1) {
+    if (!/\s/.test(text[index])) {
+      return index;
+    }
+  }
+  return -1;
+}
+
+/**
+ * Follow mode receives one streaming fragment per event, while the one-shot
+ * transcript sees the whole timeline and merges consecutive assistant and
+ * reasoning fragments with exact concatenation (`projectTimelineRows`). Printing
+ * each fragment on its own line splits words at fragment boundaries and
+ * reprints the reasoning prefix, so a chain of fragments is written as one
+ * stream instead: text goes out as it arrives, and only the trailing whitespace
+ * is held back so the message ends trimmed the way the transcript does.
+ */
+export function createFollowTranscriptWriter(
+  write: (chunk: string) => void = (chunk) => void process.stdout.write(chunk),
+  filter?: string,
+): FollowTranscriptWriter {
+  let pending: PendingStreamingText | null = null;
+
+  const emit = (chain: PendingStreamingText, text: string): void => {
+    if (!chain.emitted) {
+      chain.emitted = true;
+      if (chain.type === "reasoning") {
+        write(THOUGHT_PREFIX);
+      }
+    }
+    write(text);
+  };
+
+  const flushContent = (chain: PendingStreamingText): void => {
+    if (!chain.emitted) {
+      chain.buffer = chain.buffer.replace(/^\s+/, "");
+    }
+    const lastContent = lastNonWhitespaceIndex(chain.buffer);
+    if (lastContent < 0) {
+      return;
+    }
+    emit(chain, chain.buffer.slice(0, lastContent + 1));
+    chain.buffer = chain.buffer.slice(lastContent + 1);
+  };
+
+  const closePending = (): void => {
+    const chain = pending;
+    pending = null;
+    if (!chain) {
+      return;
+    }
+    const tail = chain.emitted ? chain.buffer.trimEnd() : chain.buffer.trim();
+    if (tail) {
+      emit(chain, tail);
+    }
+    if (chain.emitted) {
+      write("\n");
+    }
+  };
+
+  const continuesChain = (item: AgentTimelineItem, turnId: string | undefined): boolean => {
+    if (!pending || pending.type !== item.type || pending.turnId !== turnId) {
+      return false;
+    }
+    if (item.type !== "assistant_message") {
+      return true;
+    }
+    return item.messageId === undefined || item.messageId === pending.messageId;
+  };
+
+  const startChain = (
+    item: Extract<AgentTimelineItem, { type: StreamingTextType }>,
+    turnId: string | undefined,
+  ): PendingStreamingText => {
+    closePending();
+    const chain: PendingStreamingText = {
+      type: item.type,
+      turnId,
+      messageId: item.type === "assistant_message" ? item.messageId : undefined,
+      buffer: "",
+      emitted: false,
+    };
+    pending = chain;
+    return chain;
+  };
+
+  return {
+    push(item, turnId) {
+      if (!matchesFilter(item, filter)) {
+        closePending();
+        return;
+      }
+      if (item.type === "assistant_message" || item.type === "reasoning") {
+        const chain = continuesChain(item, turnId) && pending ? pending : startChain(item, turnId);
+        chain.buffer += item.text;
+        flushContent(chain);
+        return;
+      }
+      closePending();
+      const transcript = formatAgentActivityTranscript([item]);
+      if (transcript && transcript !== NO_ACTIVITY_MESSAGE) {
+        write(`${transcript}\n`);
+      }
+    },
+    end() {
+      closePending();
+    },
+  };
+}
+
 function parseTailCount(raw: string | undefined): number | undefined {
   if (raw === undefined) return undefined;
   const parsed = Number.parseInt(raw, 10);
@@ -149,12 +277,19 @@ export async function runLogsCommand(
         await client.close().catch(() => {});
         process.exit(1);
       }
-      await runFollowMode({ client, agentId: resolvedId, options, sinceTimestampMs });
+      await runFollowMode({
+        client,
+        agentId: resolvedId,
+        options,
+        sinceTimestampMs,
+      });
       return;
     }
 
     // Fetch timeline directly via cursor RPC.
-    let timelineItems = await fetchAgentTimelineItems(client, resolvedId, { sinceTimestampMs });
+    let timelineItems = await fetchAgentTimelineItems(client, resolvedId, {
+      sinceTimestampMs,
+    });
 
     // Apply filter
     if (options.filter) {
@@ -227,17 +362,22 @@ async function runFollowMode({
   const tailLabel =
     tailCount === 0 ? "no history" : `last ${tailCount} entr${tailCount === 1 ? "y" : "ies"}`;
 
+  const writer = createFollowTranscriptWriter(undefined, options.filter);
+
   const unsubscribe = client.subscribeAgentTimeline(agentId, (message) => {
     if (message.type === "agent.timeline.replacement") {
+      writer.end();
       console.log("\n[Timeline replaced; earlier output is no longer current]");
       return;
     }
 
     if (message.type === "agent.timeline.error") {
+      writer.end();
       console.error(`Timeline observation stopped: ${message.payload.error}`);
       return;
     }
     if (message.type === "agent.timeline.subscription_restored") {
+      writer.end();
       console.log("\n[Reconnected; live output resumed. Events may have been missed.]");
       return;
     }
@@ -248,15 +388,13 @@ async function runFollowMode({
         return;
       }
       const item = message.payload.event.item;
-      // Apply filter
-      if (options.filter && !matchesFilter(item, options.filter)) {
-        return;
-      }
-      // Print each timeline item as it arrives using the curator format
-      const transcript = formatAgentActivityTranscript([item]);
-      if (transcript !== NO_ACTIVITY_MESSAGE) {
-        console.log(transcript);
-      }
+      writer.push(item, message.payload.event.turnId);
+    } else if (
+      message.payload.event.type === "turn_completed" ||
+      message.payload.event.type === "turn_failed" ||
+      message.payload.event.type === "turn_canceled"
+    ) {
+      writer.end();
     }
   });
 
@@ -264,6 +402,7 @@ async function runFollowMode({
   console.log(`\n--- Following logs (${tailLabel}; Ctrl+C to stop) ---\n`);
 
   await waitForStop();
+  writer.end();
   unsubscribe();
   await client.close();
 }

packages/server/package.json

@@ -89,7 +89,7 @@
     "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts",
     "test": "npm run test:unit && npm run test:integration",
     "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"",
-    "test:integration": "vitest run --maxWorkers=1 src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\" && vitest run --maxWorkers=1 src/server/agent/providers/opencode-bridge.local.e2e.test.ts -t \"versioned runtime\"",
+    "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\" && vitest run --maxWorkers=1 src/server/agent/providers/opencode-bridge.local.e2e.test.ts -t \"versioned runtime\"",
     "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts",
     "test:integration:all": "npm run test:e2e",
     "test:integration:real": "vitest run real.e2e.test.ts",

packages/server/src/server/cli-agent-logs.test.ts

@@ -253,13 +253,19 @@ test("logs --since filters history and live output through the CLI and daemon",
       try {
         await waitForCliOutput(pastFollow, "--- Following logs");
         await Promise.all([
-          waitForCliOutput(pastFollow, "LIVE_REPLY"),
+          // Wait for a later visible item to close the streamed reply.
+          // Windows SIGINT forcibly exits without running shutdown handlers.
+          waitForCliOutput(pastFollow, "[User] LIVE_AFTER_REPLY\n"),
           (async () => {
             await manager.appendTimelineItem(agent.id, { type: "user_message", text: "LIVE_USER" });
             await manager.appendTimelineItem(agent.id, {
               type: "assistant_message",
               text: "LIVE_REPLY",
             });
+            await manager.appendTimelineItem(agent.id, {
+              type: "user_message",
+              text: "LIVE_AFTER_REPLY",
+            });
           })(),
         ]);
       } finally {

packages/server/src/server/daemon-e2e/cli-logs-follow.e2e.test.ts

@@ -0,0 +1,89 @@
+import { spawn } from "node:child_process";
+import { once } from "node:events";
+import { fileURLToPath } from "node:url";
+import { expect, test } from "vitest";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+
+// A real daemon and CLI with a deterministic provider emitting two text chunks.
+// The live Claude reproduction is recorded in the PR; this locks its CLI behavior.
+test("logs --follow preserves the same assistant text as one-shot logs", async () => {
+  const daemon = await createTestPaseoDaemon();
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  let follow: ReturnType<typeof spawn> | undefined;
+  let stdout = "";
+  let stderr = "";
+  try {
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    const agent = await client.createAgent({
+      provider: "claude",
+      cwd: daemon.paseoHome,
+      title: "Follow transcript regression",
+      modeId: "bypassPermissions",
+    });
+    follow = spawn(
+      process.execPath,
+      [
+        "--conditions=source",
+        "--import",
+        "tsx",
+        fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)),
+        "--host",
+        `127.0.0.1:${daemon.port}`,
+        "logs",
+        "--follow",
+        agent.id,
+        "--tail",
+        "0",
+        "--filter",
+        "assistant_message",
+      ],
+      {
+        env: { ...process.env, PASEO_HOME: daemon.paseoHome },
+        stdio: ["ignore", "pipe", "pipe"],
+      },
+    );
+    const exited = once(follow, "exit");
+    follow.stdout?.on("data", (chunk: Buffer) => {
+      stdout += chunk.toString();
+    });
+    follow.stderr?.on("data", (chunk: Buffer) => {
+      stderr += chunk.toString();
+    });
+    await expect
+      .poll(
+        () => {
+          if (follow?.exitCode !== null) {
+            throw new Error(`CLI exited before following logs: ${stderr}`);
+          }
+          return stdout;
+        },
+        { timeout: 20_000 },
+      )
+      .toContain("Following logs");
+    await client.sendMessage(agent.id, "respond with exactly: paseo-pr-cycle 한국어 보고서");
+    await client.waitForAgentUpsert(agent.id, (snapshot) => snapshot.status === "idle", 20_000);
+    const transcript = await client.fetchAgentTimeline(agent.id, {
+      direction: "tail",
+      limit: 100,
+      projection: "projected",
+    });
+    const assistantText = transcript.entries
+      .filter((row) => row.item.type === "assistant_message")
+      .map((row) => (row.item.type === "assistant_message" ? row.item.text : ""))
+      .join("");
+    expect(assistantText).not.toBe("");
+    // Completion closes the message while follow mode is still running.
+    // Then stop through the CLI's normal signal path.
+    await expect.poll(() => stdout, { timeout: 10_000 }).toContain(`${assistantText.trim()}\n`);
+    follow.kill("SIGINT");
+    await exited;
+    const output = stdout.slice(stdout.indexOf("---\n") + 4).trim();
+    expect(output, stderr).toBe(assistantText.trim());
+  } finally {
+    follow?.kill("SIGTERM");
+    await client.close();
+    await daemon.close();
+  }
+}, 60_000);