fix(cli): return JSON from agent logs (#8589)

* fix(cli): print JSON from paseo logs --json

paseo logs ignored the global --json flag and always printed the text
transcript. Print { agentId, items } for one-shot logs, and one timeline
item per line in --follow mode, with the follow notices on stderr so
stdout stays parseable.

* fix(cli): honor the JSON format alias in agent logs

Verify JSON output through the real CLI and isolated daemon, including the documented --format json spelling.

Co-authored-by: hadan8977 <zc-huang23@mails.tsinghua.edu.cn>

* test(cli): verify JSON follow history and live output channels

* fix(cli): stop JSON log followers when their reader closes at startup

Observe shutdown before saved history is written and retain an early stop through subscription setup. Dispose listeners and the subscription on every follow exit.

* fix(cli): interrupt pending log follow startup

Race history and subscription setup against shutdown so Ctrl+C does not wait for a slow daemon response.

---------

Co-authored-by: hadan8977 <zc-huang23@mails.tsinghua.edu.cn>
Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

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

@@ -0,0 +1,50 @@
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { mkdir, mkdtemp } from "node:fs/promises";
+import { resolve } from "node:path";
+import { connectToDaemon } from "../../utils/client.js";
+import { runPaseoCli, startTestDaemon } from "../../../tests/helpers/test-daemon.js";
+
+describe("paseo logs JSON output", () => {
+  let daemon: Awaited<ReturnType<typeof startTestDaemon>>;
+  let agentId: string;
+  let host: string;
+
+  beforeAll(async () => {
+    await mkdir(".artifacts", { recursive: true });
+    daemon = await startTestDaemon({
+      paseoHome: await mkdtemp(resolve(".artifacts/logs-json-home-")),
+      workDir: await mkdtemp(resolve(".artifacts/logs-json-work-")),
+    });
+    host = `127.0.0.1:${daemon.port}`;
+    const client = await connectToDaemon({ target: { kind: "endpoint", host } });
+    try {
+      // Creating an idle agent needs no inference or external provider connection.
+      const agent = await client.createAgent({
+        provider: "claude",
+        cwd: daemon.workDir,
+        title: "JSON logs",
+      });
+      agentId = agent.id;
+    } finally {
+      await client.close();
+    }
+  }, 60_000);
+
+  afterAll(async () => {
+    await daemon?.stop();
+  });
+
+  it.each([
+    { label: "top-level --json", args: ["logs", "--json"] },
+    { label: "agent logs --json", args: ["agent", "logs", "--json"] },
+    { label: "--format json alias", args: ["logs", "--format", "json"] },
+  ])(
+    "prints JSON through $label",
+    async ({ args }) => {
+      const result = await runPaseoCli(daemon, [...args, agentId, "--host", host]);
+      expect(result.exitCode, result.stderr).toBe(0);
+      expect(JSON.parse(result.stdout)).toEqual({ agentId, items: [] });
+    },
+    30_000,
+  );
+});

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

@@ -1,6 +1,10 @@
 import { describe, expect, it } from "vitest";
 import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types";
-import { createFollowTranscriptWriter, formatAgentActivityTranscript } from "./logs.js";
+import {
+  createFollowJsonWriter,
+  createFollowTranscriptWriter,
+  formatAgentActivityTranscript,
+} from "./logs.js";
 
 function collect(items: readonly AgentTimelineItem[], turnId?: string): string {
   let output = "";
@@ -149,3 +153,31 @@ describe("createFollowTranscriptWriter", () => {
     expect(output).toBe("# Title\n\nBody\n");
   });
 });
+
+describe("createFollowJsonWriter", () => {
+  it("writes one JSON item per line and honours the filter", () => {
+    let output = "";
+    const writer = createFollowJsonWriter("tools", (chunk) => {
+      output += chunk;
+    });
+    const toolCall = {
+      type: "tool_call",
+      callId: "call_1",
+      name: "shell",
+      status: "completed",
+      detail: { type: "unknown", input: null, output: null },
+      error: null,
+    } as AgentTimelineItem;
+
+    writer.push({ type: "assistant_message", text: "Running it.", messageId: "msg_1" });
+    writer.push(toolCall);
+    writer.end();
+
+    expect(
+      output
+        .trimEnd()
+        .split("\n")
+        .map((line) => JSON.parse(line)),
+    ).toEqual([toolCall]);
+  });
+});

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

@@ -70,6 +70,35 @@ export function formatAgentActivityTranscript(
   );
 }
 
+/**
+ * Picks the items `--tail` keeps, the same way the transcript does: the last n
+ * timeline items, before any of them are dropped from the curated view.
+ */
+function selectTailItems(
+  timelineItems: AgentTimelineItem[],
+  tailCount?: number,
+): AgentTimelineItem[] {
+  if (tailCount === undefined) {
+    return timelineItems;
+  }
+  if (tailCount === 0) {
+    return [];
+  }
+  return timelineItems.slice(-tailCount);
+}
+
+function formatAgentLogsJson(
+  agentId: string,
+  timelineItems: AgentTimelineItem[],
+  tailCount?: number,
+): string {
+  return JSON.stringify({ agentId, items: selectTailItems(timelineItems, tailCount) }, null, 2);
+}
+
+function usesJsonOutput(options: AgentLogsOptions): boolean {
+  return options.json === true || options.format?.trim().toLowerCase() === "json";
+}
+
 type StreamingTextType = "assistant_message" | "reasoning";
 
 const THOUGHT_PREFIX = "[Thought] ";
@@ -198,6 +227,21 @@ export function createFollowTranscriptWriter(
   };
 }
 
+/** Follow mode with `--json`: one JSON timeline item per line (NDJSON). */
+export function createFollowJsonWriter(
+  filter?: string,
+  write: (chunk: string) => void = (chunk) => void process.stdout.write(chunk),
+): FollowTranscriptWriter {
+  return {
+    push(item) {
+      if (matchesFilter(item, filter)) {
+        write(`${JSON.stringify(item)}\n`);
+      }
+    },
+    end() {},
+  };
+}
+
 function parseTailCount(raw: string | undefined): number | undefined {
   if (raw === undefined) return undefined;
   const parsed = Number.parseInt(raw, 10);
@@ -306,6 +350,11 @@ export async function runLogsCommand(
 
     await client.close();
 
+    if (usesJsonOutput(options)) {
+      console.log(formatAgentLogsJson(resolvedId, timelineItems, tailCount));
+      return;
+    }
+
     // Use curateAgentActivity to format the transcript
     if (tailCount === 0) {
       return;
@@ -331,78 +380,104 @@ async function runFollowMode({
   options,
   sinceTimestampMs,
 }: FollowModeInput): Promise<void> {
-  const DEFAULT_FOLLOW_TAIL = 10;
-  const tailCount = parseTailCount(options.tail) ?? DEFAULT_FOLLOW_TAIL;
-
-  // First, get existing timeline.
-  let existingItems: AgentTimelineItem[] = [];
+  // Observe shutdown before history writes can encounter a closed reader.
+  // Keep an early stop resolved while subscription setup is in flight.
+  const stopController = new AbortController();
+  const stopped = waitForStop(stopController.signal);
+  let unsubscribe: (() => void) | undefined;
+  const json = usesJsonOutput(options);
+  const writer = json
+    ? createFollowJsonWriter(options.filter)
+    : createFollowTranscriptWriter(undefined, options.filter);
   try {
-    existingItems = await fetchAgentTimelineItems(client, agentId, {
-      timeoutMs: LIVE_HISTORY_FETCH_TIMEOUT_MS,
-      sinceTimestampMs,
-    });
-  } catch (error) {
-    console.warn("Warning: failed to fetch existing timeline", error);
-  }
+    const DEFAULT_FOLLOW_TAIL = 10;
+    const tailCount = parseTailCount(options.tail) ?? DEFAULT_FOLLOW_TAIL;
 
-  // Apply filter to existing items
-  if (options.filter) {
-    existingItems = existingItems.filter((item) => matchesFilter(item, options.filter));
-  }
-
-  // Print existing transcript (tail-like behavior)
-  if (tailCount > 0) {
-    const existingTranscript = formatAgentActivityTranscript(existingItems, tailCount);
-    if (existingTranscript !== NO_ACTIVITY_MESSAGE) {
-      console.log(existingTranscript);
-    }
-  }
-
-  // Subscribe to new events
-  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;
+    // First, get existing timeline.
+    let existingItems: AgentTimelineItem[] = [];
+    try {
+      const history = await Promise.race([
+        fetchAgentTimelineItems(client, agentId, {
+          timeoutMs: LIVE_HISTORY_FETCH_TIMEOUT_MS,
+          sinceTimestampMs,
+        }),
+        stopped,
+      ]);
+      if (!history) return;
+      existingItems = history;
+    } catch (error) {
+      console.warn("Warning: failed to fetch existing timeline", error);
     }
 
-    if (message.type === "agent.timeline.error") {
-      writer.end();
-      console.error(`Timeline observation stopped: ${message.payload.error}`);
-      return;
+    // Apply filter to existing items
+    if (options.filter) {
+      existingItems = existingItems.filter((item) => matchesFilter(item, options.filter));
     }
-    if (message.type === "agent.timeline.subscription_restored") {
-      writer.end();
-      console.log("\n[Reconnected; live output resumed. Events may have been missed.]");
-      return;
+
+    // Print existing transcript (tail-like behavior)
+    if (json) {
+      for (const item of selectTailItems(existingItems, tailCount)) {
+        console.log(JSON.stringify(item));
+      }
+    } else if (tailCount > 0) {
+      const existingTranscript = formatAgentActivityTranscript(existingItems, tailCount);
+      if (existingTranscript !== NO_ACTIVITY_MESSAGE) {
+        console.log(existingTranscript);
+      }
     }
-    if (message.payload.event.type === "timeline") {
-      const matchesSince =
-        sinceTimestampMs === undefined || Date.parse(message.payload.timestamp) >= sinceTimestampMs;
-      if (!matchesSince) {
+
+    // Subscribe to new events
+    const tailLabel =
+      tailCount === 0 ? "no history" : `last ${tailCount} entr${tailCount === 1 ? "y" : "ies"}`;
+
+    // Keep stdout parseable in JSON mode: one timeline item per line, notices on stderr.
+    const notice = json ? console.error : console.log;
+
+    const subscription = client.subscribeAgentTimeline(agentId, (message) => {
+      if (message.type === "agent.timeline.replacement") {
+        writer.end();
+        notice("\n[Timeline replaced; earlier output is no longer current]");
         return;
       }
-      const item = message.payload.event.item;
-      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();
-    }
-  });
 
-  await unsubscribe.ready;
-  console.log(`\n--- Following logs (${tailLabel}; Ctrl+C to stop) ---\n`);
+      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();
+        notice("\n[Reconnected; live output resumed. Events may have been missed.]");
+        return;
+      }
+      if (message.payload.event.type === "timeline") {
+        const matchesSince =
+          sinceTimestampMs === undefined ||
+          Date.parse(message.payload.timestamp) >= sinceTimestampMs;
+        if (!matchesSince) {
+          return;
+        }
+        const item = message.payload.event.item;
+        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();
+      }
+    });
 
-  await waitForStop();
-  writer.end();
-  unsubscribe();
-  await client.close();
+    unsubscribe = subscription;
+    const ready = await Promise.race([subscription.ready.then(() => true), stopped]);
+    if (!ready) return;
+    notice(`\n--- Following logs (${tailLabel}; Ctrl+C to stop) ---\n`);
+
+    await stopped;
+  } finally {
+    stopController.abort();
+    writer.end();
+    unsubscribe?.();
+    await client.close();
+  }
 }

packages/cli/src/utils/wait-for-stop.ts

@@ -3,9 +3,10 @@
 // their output closes its end, as `| head` does. A closed reader shows up as
 // EPIPE on the next write to stdout, so the command stops at its next line of
 // output.
-export function waitForStop(): Promise<void> {
+export function waitForStop(signal?: AbortSignal): Promise<void> {
   return new Promise((resolve) => {
     const stop = () => {
+      signal?.removeEventListener("abort", stop);
       process.off("SIGINT", stop);
       process.off("SIGTERM", stop);
       process.stdout.off("error", stopOnClosedStdout);
@@ -20,5 +21,7 @@ export function waitForStop(): Promise<void> {
     process.on("SIGINT", stop);
     process.on("SIGTERM", stop);
     process.stdout.on("error", stopOnClosedStdout);
+    signal?.addEventListener("abort", stop, { once: true });
+    if (signal?.aborted) stop();
   });
 }

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

@@ -1,7 +1,10 @@
 import { spawn } from "node:child_process";
 import { once } from "node:events";
+import { mkdir, mkdtemp } from "node:fs/promises";
+import { join, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import { expect, test } from "vitest";
+import { WebSocket, WebSocketServer } from "ws";
 import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
 import { DaemonClient } from "../test-utils/daemon-client.js";
 
@@ -137,3 +140,225 @@ test("omitted after limit pages forward while explicit zero preserves complete C
     await daemon.close();
   }
 }, 60_000);
+
+test.each([
+  { label: "--json", flags: ["--json"] },
+  { label: "--format json", flags: ["--format", "json"] },
+])(
+  "logs --follow $label keeps history, live items and notices on their channels",
+  async ({ flags }) => {
+    await mkdir(".artifacts", { recursive: true });
+    const paseoHomeRoot = await mkdtemp(resolve(".artifacts/logs-follow-json-"));
+    const staticDir = join(paseoHomeRoot, "static");
+    await mkdir(staticDir);
+    const daemon = await createTestPaseoDaemon({ paseoHomeRoot, staticDir });
+    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();
+      const agent = await client.createAgent({ provider: "claude", cwd: daemon.paseoHome });
+      const history = [
+        { type: "user_message" as const, text: "Saved JSON prompt" },
+        { type: "assistant_message" as const, text: "Saved JSON reply", messageId: "saved" },
+      ];
+      for (const item of history)
+        await daemon.daemon.agentManager.appendTimelineItem(agent.id, item);
+      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",
+          agent.id,
+          "--follow",
+          ...flags,
+        ],
+        {
+          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 + stderr;
+          },
+          { timeout: 20_000 },
+        )
+        .toContain("Following logs");
+      const live = {
+        type: "assistant_message" as const,
+        text: "Live JSON reply",
+        messageId: "live",
+      };
+      await daemon.daemon.agentManager.appendTimelineItem(agent.id, live);
+      await expect.poll(() => stdout, { timeout: 10_000 }).toContain(live.text);
+      follow.kill("SIGINT");
+      await exited;
+      expect(
+        stdout
+          .trim()
+          .split("\n")
+          .map((line) => JSON.parse(line)),
+      ).toEqual([...history, live]);
+      expect(stderr).toContain("Following logs");
+      expect(stdout).not.toContain("Following logs");
+    } finally {
+      follow?.kill("SIGTERM");
+      await client.close();
+      await daemon.close();
+    }
+  },
+  60_000,
+);
+
+test.each([
+  { label: "--json", command: ["logs"], flags: ["--json"] },
+  { label: "--format json", command: ["agent", "logs"], flags: ["--format", "json"] },
+])(
+  "logs --follow $label exits when stdout closes before saved history is written",
+  async ({ command, flags }) => {
+    await mkdir(".artifacts", { recursive: true });
+    const paseoHomeRoot = await mkdtemp(resolve(".artifacts/logs-follow-closed-"));
+    const staticDir = join(paseoHomeRoot, "static");
+    await mkdir(staticDir);
+    const daemon = await createTestPaseoDaemon({ paseoHomeRoot, staticDir });
+    const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+    let follow: ReturnType<typeof spawn> | undefined;
+    let stderr = "";
+    try {
+      await client.connect();
+      // No turn is started: the saved history must trigger shutdown on its own.
+      const agent = await client.createAgent({ provider: "claude", cwd: daemon.paseoHome });
+      await daemon.daemon.agentManager.appendTimelineItem(agent.id, {
+        type: "assistant_message",
+        text: "Saved reply before the reader closes",
+        messageId: "saved",
+      });
+      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}`,
+          ...command,
+          agent.id,
+          "--follow",
+          ...flags,
+        ],
+        {
+          env: { ...process.env, PASEO_HOME: daemon.paseoHome },
+          stdio: ["ignore", "pipe", "pipe"],
+        },
+      );
+      const exited = once(follow, "exit");
+      follow.stderr?.on("data", (chunk: Buffer) => {
+        stderr += chunk.toString();
+      });
+      follow.stdout?.destroy();
+      await expect.poll(() => follow?.exitCode, { timeout: 10_000 }).toBe(0);
+      expect(await exited, stderr).toEqual([0, null]);
+    } finally {
+      follow?.kill("SIGKILL");
+      await client.close();
+      await daemon.close();
+    }
+  },
+  30_000,
+);
+
+// Windows kill("SIGINT") force-terminates rather than delivering a console interrupt.
+test.skipIf(process.platform === "win32")(
+  "logs --follow stops on SIGINT while subscription startup is waiting",
+  async () => {
+    await mkdir(".artifacts", { recursive: true });
+    const paseoHomeRoot = await mkdtemp(resolve(".artifacts/logs-follow-startup-"));
+    const staticDir = join(paseoHomeRoot, "static");
+    await mkdir(staticDir);
+    const daemon = await createTestPaseoDaemon({ paseoHomeRoot, staticDir });
+    const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+    const proxy = new WebSocketServer({ port: 0, host: "127.0.0.1" });
+    await once(proxy, "listening");
+    const address = proxy.address();
+    if (typeof address === "string" || address === null) throw new Error("Missing proxy port");
+    const upstreams: WebSocket[] = [];
+    let waitingForSubscription = false;
+    proxy.on("connection", (downstream) => {
+      const upstream = new WebSocket(`ws://127.0.0.1:${daemon.port}/ws`);
+      upstreams.push(upstream);
+      const connected = once(upstream, "open");
+      downstream.on("message", async (data, binary) => {
+        await connected;
+        const frame = binary ? undefined : JSON.parse(data.toString());
+        // Hold only the subscription request. All other traffic uses the real daemon.
+        if (frame?.message?.type === "agent.timeline.set_subscription.request") {
+          waitingForSubscription = true;
+          return;
+        }
+        upstream.send(data, { binary });
+      });
+      upstream.on("message", (data, binary) => downstream.send(data, { binary }));
+      downstream.on("close", () => upstream.close());
+    });
+    let follow: ReturnType<typeof spawn> | undefined;
+    try {
+      await client.connect();
+      const agent = await client.createAgent({ provider: "claude", cwd: daemon.paseoHome });
+      follow = spawn(
+        process.execPath,
+        [
+          "--conditions=source",
+          "--import",
+          "tsx",
+          fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)),
+          "--host",
+          `127.0.0.1:${address.port}`,
+          "logs",
+          agent.id,
+          "--follow",
+          "--json",
+        ],
+        {
+          env: { ...process.env, PASEO_HOME: daemon.paseoHome },
+          stdio: ["ignore", "pipe", "pipe"],
+        },
+      );
+      follow.stdout?.resume();
+      follow.stderr?.resume();
+      const exited = once(follow, "exit");
+      await expect.poll(() => waitingForSubscription, { timeout: 10_000 }).toBe(true);
+      follow.kill("SIGINT");
+      await expect.poll(() => follow?.exitCode, { timeout: 5_000 }).toBe(0);
+      expect(await exited).toEqual([0, null]);
+    } finally {
+      if (follow && follow.exitCode === null && follow.signalCode === null) {
+        follow.kill("SIGKILL");
+        await once(follow, "exit");
+      }
+      for (const upstream of upstreams) upstream.terminate();
+      for (const downstream of proxy.clients) downstream.terminate();
+      await new Promise<void>((done) => proxy.close(() => done()));
+      await client.close();
+      await daemon.close();
+    }
+  },
+  30_000,
+);