Merge origin/main into plan-provider-switch-sqlite-timelines

Adapts main's changes to the single asynchronous timeline owner:

- The subscribe freshness check (#9175) awaits the loaded timeline head and
  compares the client's cursor against the epoch that client sees, so legacy
  clients compare against their derived epoch after a rewind.
- History reads (#8737) pass purpose "history" when an agent with no stored
  timeline must be loaded; stored history is still read without the provider.
- Background work clears on reload, close and parent delete next to the
  async provider-child cancellation.
- Claude steered-message rewinds (#8372) report an at-target cut. Codex keeps
  main's saved thread id alongside the reported cut.
- Plugin rewinds (#8368) return an unreported outcome; a failed replay after
  the native revert leaves the rewind for the reload to settle rather than a
  retry. Manager-level rewinds target the timeline row's message id.
- Main's tests read the timeline asynchronously, and wait for the provider to
  receive a prompt instead of assuming it arrives in the dispatch tick.

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,85 +380,108 @@ 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);
+    // 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);
     }
-  }
-
-  // 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;
-    }
-    if (message.type === "agent.timeline.rewind") {
-      writer.end();
-      console.log(
-        "\n[Conversation rewound; output since the rewound message is no longer current]",
-      );
-      return;
+    // Apply filter to existing items
+    if (options.filter) {
+      existingItems = existingItems.filter((item) => matchesFilter(item, options.filter));
     }
 
-    if (message.type === "agent.timeline.error") {
-      writer.end();
-      console.error(`Timeline observation stopped: ${message.payload.error}`);
-      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.type === "agent.timeline.subscription_restored") {
-      writer.end();
-      console.log("\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) {
+
+    // 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;
+      }
+      if (message.type === "agent.timeline.rewind") {
+        writer.end();
+        notice("\n[Conversation rewound; output since the rewound message 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/commands/agent/ls.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it } from "vitest";
 import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages";
+import { render } from "../../output/index.js";
 import { buildAgentLsFetchOptions, listAgents } from "./ls.js";
 
 const daemonTarget = { kind: "endpoint" as const, host: "example.test:12345" };
@@ -130,4 +131,26 @@ describe("runLsCommand", () => {
 
     expect(result.data.map((item) => item.name)).toEqual(["agent 3", "agent 2", "agent 1"]);
   });
+
+  it("marks background agents in a BACKGROUND column only for inclusive listings", async () => {
+    const client = createPagingClient();
+    const fetchAgents: typeof client.fetchAgents = async (options) => {
+      const page = await client.fetchAgents(options);
+      return {
+        ...page,
+        entries: page.entries.map(({ agent }) => ({
+          agent: agent.title === "agent 2" ? { ...agent, background: true } : agent,
+        })),
+      };
+    };
+    const connect = async () => ({ ...client, fetchAgents });
+
+    const inclusive = await listAgents({ daemonTarget, all: true, global: true }, connect);
+    const rows = render(inclusive, { noColor: true }).split("\n");
+    expect(rows[0]).toMatch(/BACKGROUND\s*$/);
+    expect(rows.slice(1).map((row) => row.trim().split(/\s+/).at(-1))).toEqual(["-", "yes", "-"]);
+
+    const ordinary = await listAgents({ daemonTarget, global: true }, connect);
+    expect(render(ordinary, { noColor: true })).not.toContain("BACKGROUND");
+  });
 });

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

@@ -15,7 +15,7 @@ export function addLsOptions(cmd: Command): Command {
     .description("List agents. By default excludes archived agents.")
     .option("-a, --all", "Include archived and background agents")
     .option("-g, --global", "List agents across all directories")
-    .option("--background", "Include background workspace agents")
+    .option("--background", "Include background agents and agents in background workspaces")
     .option(
       "--label <key=value>",
       "Filter by label (can be used multiple times)",
@@ -35,6 +35,7 @@ export interface AgentListItem {
   status: string;
   cwd: string;
   created: string;
+  background: boolean;
 }
 
 /** Helper to get relative time string */
@@ -89,6 +90,15 @@ export const agentLsSchema: OutputSchema<AgentListItem> = {
   ],
 };
 
+/** Inclusive listings mark background agents so they read apart from visible ones. */
+const agentLsInclusiveSchema: OutputSchema<AgentListItem> = {
+  ...agentLsSchema,
+  columns: [
+    ...agentLsSchema.columns,
+    { header: "BACKGROUND", field: (item) => (item.background ? "yes" : "-"), width: 10 },
+  ],
+};
+
 /** Transform agent snapshot to AgentListItem */
 function toListItem(agent: AgentSnapshotPayload): AgentListItem {
   const model = normalizeModelId(agent.runtimeInfo?.model) ?? normalizeModelId(agent.model);
@@ -101,6 +111,7 @@ function toListItem(agent: AgentSnapshotPayload): AgentListItem {
     status: agent.status,
     cwd: shortenPath(agent.cwd),
     created: relativeTime(agent.createdAt),
+    background: agent.background === true,
   };
 }
 
@@ -259,7 +270,7 @@ export async function listAgents(
     return {
       type: "list",
       data: items,
-      schema: agentLsSchema,
+      schema: options.all || options.background ? agentLsInclusiveSchema : agentLsSchema,
     };
   } catch (err) {
     await client.close().catch(() => {});

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();
   });
 }