diff --git a/packages/cli/src/commands/agent/logs.test.ts b/packages/cli/src/commands/agent/logs.test.ts new file mode 100644 index 0000000000..b687aefd92 --- /dev/null +++ b/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"); + }); +}); diff --git a/packages/cli/src/commands/agent/logs.ts b/packages/cli/src/commands/agent/logs.ts index 9ac2d4b238..ed77d4be62 100644 --- a/packages/cli/src/commands/agent/logs.ts +++ b/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, + 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(); } diff --git a/packages/server/package.json b/packages/server/package.json index 53288d2acb..33bd91fd72 100644 --- a/packages/server/package.json +++ b/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", diff --git a/packages/server/src/server/cli-agent-logs.test.ts b/packages/server/src/server/cli-agent-logs.test.ts index be492f6168..634d0615c0 100644 --- a/packages/server/src/server/cli-agent-logs.test.ts +++ b/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 { 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 new file mode 100644 index 0000000000..d8388f9dd2 --- /dev/null +++ b/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 | 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);