From 3857ed336594978410e7acddb4eeb194e84ea046 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 01:52:53 +0000 Subject: [PATCH] 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 * 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 Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../cli/src/commands/agent/logs-json.test.ts | 50 ++++ packages/cli/src/commands/agent/logs.test.ts | 34 ++- packages/cli/src/commands/agent/logs.ts | 203 +++++++++++----- packages/cli/src/utils/wait-for-stop.ts | 5 +- .../daemon-e2e/cli-logs-follow.e2e.test.ts | 225 ++++++++++++++++++ 5 files changed, 451 insertions(+), 66 deletions(-) create mode 100644 packages/cli/src/commands/agent/logs-json.test.ts diff --git a/packages/cli/src/commands/agent/logs-json.test.ts b/packages/cli/src/commands/agent/logs-json.test.ts new file mode 100644 index 0000000000..bbb69f9bd7 --- /dev/null +++ b/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>; + 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, + ); +}); diff --git a/packages/cli/src/commands/agent/logs.test.ts b/packages/cli/src/commands/agent/logs.test.ts index e551aa3791..afbf97acb6 100644 --- a/packages/cli/src/commands/agent/logs.test.ts +++ b/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]); + }); +}); diff --git a/packages/cli/src/commands/agent/logs.ts b/packages/cli/src/commands/agent/logs.ts index ed77d4be62..6b40a92536 100644 --- a/packages/cli/src/commands/agent/logs.ts +++ b/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 { - 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(); + } } diff --git a/packages/cli/src/utils/wait-for-stop.ts b/packages/cli/src/utils/wait-for-stop.ts index 8430c39726..7988d8f436 100644 --- a/packages/cli/src/utils/wait-for-stop.ts +++ b/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 { +export function waitForStop(signal?: AbortSignal): Promise { 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 { process.on("SIGINT", stop); process.on("SIGTERM", stop); process.stdout.on("error", stopOnClosedStdout); + signal?.addEventListener("abort", stop, { once: true }); + if (signal?.aborted) stop(); }); } diff --git a/packages/server/src/server/daemon-e2e/cli-logs-follow.e2e.test.ts b/packages/server/src/server/daemon-e2e/cli-logs-follow.e2e.test.ts index 55750831c8..63cdf2363b 100644 --- a/packages/server/src/server/daemon-e2e/cli-logs-follow.e2e.test.ts +++ b/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 | 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 | 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 | 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((done) => proxy.close(() => done())); + await client.close(); + await daemon.close(); + } + }, + 30_000, +);