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, +);