test(muse): preserve timeout reproductions and patch validation

Retain the controlled protocol reproductions for #6865 without changing
production behavior. Keep malformed patch responses distinguishable from
unavailable output when evaluating the proposed recovery behavior.

Co-authored-by: Blah <michaelaguilera12345@gmail.com>

packages/plugin/src/server/lifecycle.ts

@@ -50,6 +50,10 @@ export interface PluginLifecycleEvents {
     agent: PluginHookAgent;
     turnId: string | null;
     outcome: PluginTurnOutcome;
+    /**
+     * @deprecated Always empty. Fetch history on demand with
+     * context.paseo.agents.ref(agent.id).timeline.refetch({ direction: "tail", limit: 20 }).
+     */
     timeline: readonly AgentTimelineItem[];
   };
   "agent.permission_requested": { agent: PluginHookAgent; request: AgentPermissionRequest };

packages/server/src/server/agent/agent-manager.test.ts

@@ -3,7 +3,7 @@ import {
   formatAgentMessage,
   formatSystemNotificationPrompt,
 } from "./agent-messages/index.js";
-import { expect, test, vi } from "vitest";
+import { expect, test, vi, onTestFinished } from "vitest";
 import { spawn } from "node:child_process";
 import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { join } from "node:path";
@@ -11892,3 +11892,102 @@ test.each([false, true])(
     }
   },
 );
+
+test.each([
+  {
+    event: {
+      type: "turn_completed",
+      provider: "codex",
+      turnId: "hook-turn",
+    } satisfies AgentStreamEvent,
+    outcome: { kind: "completed" },
+  },
+  {
+    event: {
+      type: "turn_failed",
+      provider: "codex",
+      turnId: "hook-turn",
+      error: "Failed",
+      code: "test",
+    } satisfies AgentStreamEvent,
+    outcome: { kind: "failed", error: { message: "Failed", code: "test" } },
+  },
+  {
+    event: {
+      type: "turn_canceled",
+      provider: "codex",
+      turnId: "hook-turn",
+      reason: "Stopped",
+    } satisfies AgentStreamEvent,
+    outcome: { kind: "canceled", reason: "Stopped" },
+  },
+])(
+  "lifecycle hooks receive no conversation for $event.type and stream dispatch never reads timeline items",
+  async ({ event: terminalEvent, outcome }) => {
+    class CountingTimelineStore extends InMemoryAgentTimelineStore {
+      itemReads = 0;
+      override getItems(agentId: string): AgentTimelineItem[] {
+        this.itemReads += 1;
+        return super.getItems(agentId);
+      }
+    }
+    const workdir = mkdtempSync(join(tmpdir(), "agent-hook-payload-"));
+    const client = new SessionRecordingAgentClient();
+    const timelineStore = new CountingTimelineStore();
+    const events: Array<{ name: string; event: unknown }> = [];
+    const manager = new AgentManager({
+      clients: { codex: client },
+      registry: new AgentStorage(join(workdir, "agents"), logger),
+      logger,
+      timelineStore,
+      pluginLifecycle: {
+        before: async (_name, request) => request,
+        emit: (name, event) => {
+          events.push({ name, event });
+        },
+      },
+    });
+    onTestFinished(async () => {
+      await manager.closeAgent(manager.listAgents()[0]!.id);
+      await manager.flush();
+      rmSync(workdir, { recursive: true, force: true });
+    });
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    await manager.appendTimelineItem(agent.id, {
+      type: "assistant_message",
+      text: "A long conversation".repeat(10_000),
+    });
+    expect(manager.getTimeline(agent.id)).toHaveLength(1);
+    const readsBeforeDispatch = timelineStore.itemReads;
+    const session = client.sessions[0]!;
+    session.pushEvent({ type: "turn_started", provider: "codex", turnId: "hook-turn" });
+    await expect
+      .poll(() => events.filter(({ name }) => name === "agent.turn_started").length)
+      .toBe(1);
+    expect(timelineStore.itemReads).toBe(readsBeforeDispatch);
+    session.pushEvent(terminalEvent);
+    await expect
+      .poll(() => events.filter(({ name }) => name === "agent.turn_ended"))
+      .toEqual([
+        {
+          name: "agent.turn_ended",
+          event: {
+            agent: {
+              id: agent.id,
+              workspaceId: null,
+              parentAgentId: null,
+              provider: "codex",
+              cwd: workdir,
+              title: null,
+            },
+            turnId: "hook-turn",
+            outcome,
+            timeline: [],
+          },
+        },
+      ]);
+    expect(timelineStore.itemReads).toBe(readsBeforeDispatch);
+  },
+);

packages/server/src/server/agent/agent-manager.ts

@@ -326,6 +326,7 @@ export interface AgentManagerOptions {
   registry?: AgentStorage;
   onAgentAttention?: AgentAttentionCallback;
   onWorkspaceStateMayHaveChanged?: (params: { cwd: string }) => void;
+  timelineStore?: InMemoryAgentTimelineStore;
   durableTimelineStore?: AgentTimelineStore;
   terminalManager?: TerminalManager | null;
   mcpBaseUrl?: string;
@@ -741,7 +742,7 @@ export class AgentManager {
   private readonly providerEnabled = new Map<AgentProvider, boolean>();
   private readonly providerDefinitions = new Map<AgentProvider, ProviderEnabledFlag>();
   private readonly agents = new Map<string, LiveManagedAgent>();
-  private readonly timelineStore = new InMemoryAgentTimelineStore();
+  private readonly timelineStore: InMemoryAgentTimelineStore;
   private readonly providerSubagents = new ProviderSubagentStore();
   private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
   private readonly sessionEventTails = new Map<string, Promise<void>>();
@@ -778,7 +779,8 @@ export class AgentManager {
 
   constructor(options: AgentManagerOptions) {
     this.pluginLifecycle = options.pluginLifecycle;
-    this.idFactory = options?.idFactory ?? (() => randomUUID());
+    this.timelineStore = options.timelineStore ?? new InMemoryAgentTimelineStore();
+    this.idFactory = options.idFactory ?? (() => randomUUID());
     this.registry = options?.registry;
     this.durableTimelineStore = options?.durableTimelineStore;
     this.onAgentAttention = options?.onAgentAttention;
@@ -5164,7 +5166,6 @@ export class AgentManager {
         this.pluginLifecycle,
         describeHookAgent({ ...agent, title: agent.config.title }),
         event,
-        this.timelineStore.getItems(agentId),
       );
     }
   }

packages/server/src/server/plugins/lifecycle.e2e.test.ts

@@ -146,10 +146,12 @@ export default function contribute(server) {
         { hook: "agent.permission_resolved", event: { resolution: { behavior: "deny" } } },
         {
           hook: "agent.turn_ended",
-          event: { outcome: { kind: "completed" }, timeline: expect.any(Array) },
+          event: { outcome: { kind: "completed" }, timeline: [] },
         },
         { hook: "agent.turn_started", event: { agent: { id: agent.id } } },
       ]);
+    const page = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 20 });
+    expect(page.entries.length).toBeGreaterThan(0);
     await client.archiveAgent(agent.id);
     await expect
       .poll(async () => {

packages/server/src/server/plugins/lifecycle/index.ts

@@ -1,4 +1,4 @@
-import type { AgentStreamEvent, AgentTimelineItem } from "../../agent/agent-sdk-types.js";
+import type { AgentStreamEvent } from "../../agent/agent-sdk-types.js";
 import { z } from "zod";
 import { CreateAgentRequestMessageSchema } from "@getpaseo/protocol/messages";
 import type {
@@ -94,7 +94,6 @@ export function publishAgentStream(
   lifecycle: PluginLifecycle,
   agent: PluginHookAgent,
   event: AgentStreamEvent,
-  timeline: readonly AgentTimelineItem[],
 ): void {
   if (event.type === "turn_started") {
     lifecycle.emit("agent.turn_started", { agent, turnId: event.turnId ?? null });
@@ -102,21 +101,21 @@ export function publishAgentStream(
     lifecycle.emit("agent.turn_ended", {
       agent,
       turnId: event.turnId ?? null,
-      timeline,
+      timeline: [],
       outcome: { kind: "completed" },
     });
   } else if (event.type === "turn_failed") {
     lifecycle.emit("agent.turn_ended", {
       agent,
       turnId: event.turnId ?? null,
-      timeline,
+      timeline: [],
       outcome: { kind: "failed", error: { message: event.error, code: event.code } },
     });
   } else if (event.type === "turn_canceled") {
     lifecycle.emit("agent.turn_ended", {
       agent,
       turnId: event.turnId ?? null,
-      timeline,
+      timeline: [],
       outcome: { kind: "canceled", reason: event.reason },
     });
   } else if (event.type === "permission_requested") {

plugin-examples/lifecycle-actions/index.server.ts

@@ -21,9 +21,10 @@ export default function contribute(server: PluginServerContext) {
     if (event.outcome.kind === "canceled") {
       return;
     }
-    const text = latestOutputText(event.timeline);
+    const agent = context.paseo.agents.ref(event.agent.id);
+    const page = await agent.timeline.refetch({ direction: "tail", limit: 20 });
+    const text = latestOutputText(page.entries.map(({ item }) => item));
     if (/out of credits/i.test(text)) {
-      const agent = context.paseo.agents.ref(event.agent.id);
       console.log("Sending a follow-up after out of credits", event.agent.id);
       await agent.send("Try again.");
     }

plugins/muse-provider/test/fake-muse.mjs

@@ -1,4 +1,4 @@
-import { appendFileSync, readFileSync } from "node:fs";
+import { appendFileSync, existsSync, readFileSync } from "node:fs";
 import { createInterface } from "node:readline";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
@@ -298,7 +298,38 @@ function validateCommand(frame, recorded) {
   )
     throw new Error("Wrong approval stage");
 }
+function failedOrDelayedResponse(recordedMessage, recorded, frame) {
+  if (recordedMessage.method || recordedMessage.id !== recorded.id) return false;
+  if (frame.method === "item/readOutput" && process.env.MUSE_TEST_OUTPUT_FAILURE) {
+    if (process.env.MUSE_TEST_OUTPUT_FAILURE === "malformed") {
+      process.stdout.write(
+        JSON.stringify({
+          jsonrpc: "2.0",
+          id: frame.id,
+          result: { encoding: "utf8", eof: true, content: null },
+        }) + "\n",
+      );
+      return true;
+    }
+    // Only the patch response fails; subsequent turn notifications still arrive.
+    if (process.env.MUSE_TEST_OUTPUT_FAILURE === "error")
+      rpcError(frame, -32000, "outputUnavailable", "Stored patch unavailable");
+    return true;
+  }
+  if (frame.method === "session/resume" && process.env.MUSE_TEST_RESUME_GATE) {
+    const gate = process.env.MUSE_TEST_RESUME_GATE;
+    delete process.env.MUSE_TEST_RESUME_GATE;
+    function release() {
+      if (existsSync(gate)) emitFixtureMessage(recordedMessage, recorded, frame);
+      else setTimeout(release, 10);
+    }
+    release();
+    return true;
+  }
+  return false;
+}
 function emitFixtureMessage(recordedMessage, recorded, frame) {
+  if (failedOrDelayedResponse(recordedMessage, recorded, frame)) return;
   const message = replace(recordedMessage);
   if (!message.method && message.id === recorded.id) message.id = frame.id;
   if (

plugins/muse-provider/test/provider.test.ts

@@ -138,6 +138,14 @@ async function harness(
       .filter(Boolean)
       .map((line) => JSON.parse(line));
   }
+  async function waitForRequest(method: string) {
+    const deadline = Date.now() + 5000;
+    while (Date.now() < deadline) {
+      if ((await recorded()).some((frame) => frame.method === method)) return;
+      await delay(10);
+    }
+    throw new Error(`Missing request: ${method}`);
+  }
   return {
     provider,
     launch,
@@ -148,6 +156,7 @@ async function harness(
     open,
     prompt,
     recorded,
+    waitForRequest,
     root,
     usageSource,
   };
@@ -260,6 +269,58 @@ test("tool calls classify read, shell, and edit with fetched JSON patch converte
   );
 });
 
+test.each(["error", "timeout"])(
+  "patch fetch %s preserves the edit and completes the turn",
+  async (failure) => {
+    const h = await harness("tools-edit", { MUSE_TEST_OUTPUT_FAILURE: failure });
+    expect(await h.open()).toMatchObject({ type: "session.ready" });
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+    await h.prompt();
+    await h.waitForRequest("item/readOutput");
+    if (failure === "timeout") await vi.advanceTimersByTimeAsync(60000);
+    expect(
+      await h.wait(
+        (event) =>
+          event.type === "session.runtime_failed" ||
+          (event.type === "session.turn" && event.state === "completed"),
+      ),
+    ).toMatchObject({ type: "session.turn", state: "completed" });
+    const edit = h.events.findLast(
+      (event) =>
+        event.type === "timeline.item" &&
+        event.item.type === "tool_call" &&
+        event.item.name === "edit_file",
+    );
+    expect(edit).toMatchObject({
+      type: "timeline.item",
+      item: {
+        type: "tool_call",
+        name: "edit_file",
+        status: "completed",
+        error: null,
+        detail: {
+          type: "edit",
+          filePath: "f",
+          oldString: "beta",
+          newString: "gamma",
+          unifiedDiff: undefined,
+        },
+      },
+    });
+    expect(
+      h.events.findLast(
+        (event) => event.type === "timeline.item" && event.item.type === "assistant_message",
+      ),
+    ).toMatchObject({
+      item: { text: "Done. `f` had 2 lines; changed line 2 from `beta` to `gamma`.\n" },
+    });
+    expect(h.events.filter((event) => event.type === "session.runtime_failed")).toEqual([]);
+    expect(
+      h.events.filter((event) => event.type === "session.turn" && event.state === "completed"),
+    ).toHaveLength(1);
+  },
+);
+
 test("multi-stage approvals surface Stage 2 then Stage 3 and send the current requirement guard", async () => {
   const h = await harness("multi-stage-approval");
   await h.open();
@@ -377,6 +438,36 @@ test("a fresh connection restores earlier messages from persisted session state"
   await reopened.wait((event) => event.type === "session.turn" && event.state === "completed");
 });
 
+test("resume succeeds after more than ten seconds and can complete a new turn", async () => {
+  const root = await mkdtemp(path.join(os.tmpdir(), "muse-resume-gate-"));
+  roots.push(root);
+  const gate = path.join(root, "release");
+  const h = await harness("resume-without-cursor", { MUSE_TEST_RESUME_GATE: gate });
+  vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+  const opened = h.open({ version: 1, data: { sessionId: "saved-session" } });
+  await h.waitForRequest("session/resume");
+  // The child holds its real MSP response while the provider's request clock advances.
+  await vi.advanceTimersByTimeAsync(15000);
+  await writeFile(gate, "release");
+  expect(await opened).toMatchObject({ type: "session.ready" });
+  expect(h.events).toEqual(
+    expect.arrayContaining([
+      expect.objectContaining({
+        type: "timeline.item",
+        item: expect.objectContaining({ type: "assistant_message", text: "RESUME_MARKER_OK" }),
+      }),
+    ]),
+  );
+  await h.prompt();
+  await h.wait((event) => event.type === "session.turn" && event.state === "completed");
+  expect(
+    h.events.filter(
+      (event) => event.type === "request.failed" || event.type === "session.runtime_failed",
+    ),
+  ).toEqual([]);
+  expect((await h.recorded()).filter((frame) => frame.method === "session/start")).toEqual([]);
+});
+
 for (const cursor of [undefined, "v:old:7"]) {
   test(`resume ${cursor ? "with a legacy persistence cursor" : "without a persistence cursor"} replays inline history`, async () => {
     const h = await harness("resume-without-cursor");
@@ -1743,3 +1834,16 @@ test("Muse reports belong only to the session's account and remain available glo
   expect(await registry.listReports({ agentId: "muse-unknown" })).toEqual([]);
   expect((await registry.listReports()).map((report) => report.id)).toEqual(accountIds);
 });
+
+test("malformed patch responses remain protocol errors", async () => {
+  const h = await harness("tools-edit", { MUSE_TEST_OUTPUT_FAILURE: "malformed" });
+  expect(await h.open()).toMatchObject({ type: "session.ready" });
+  await h.prompt();
+  expect(
+    await h.wait(
+      (event) =>
+        event.type === "session.runtime_failed" ||
+        (event.type === "session.turn" && event.state === "completed"),
+    ),
+  ).toMatchObject({ type: "session.runtime_failed" });
+});

public-docs/plugins/reference.md

@@ -654,9 +654,11 @@ export default function contribute(server: PluginServerContext) {
       return;
     }
 
-    const text = latestOutputText(event.timeline);
+    const agent = context.paseo.agents.ref(event.agent.id);
+    const page = await agent.timeline.refetch({ direction: "tail", limit: 20 });
+    const text = latestOutputText(page.entries.map(({ item }) => item));
     if (/out of credits/i.test(text)) {
-      await context.paseo.agents.ref(event.agent.id).send("Try again.");
+      await agent.send("Try again.");
     }
   });
 
@@ -670,6 +672,9 @@ Turn ends: "out of credits"
   → a new turn starts
 ```
 
+The handler fetches only the last 20 timeline items. Output before that page is not inspected.
+Use `cursor` and `direction` to page further when your plugin needs more history.
+
 This sends a new message with the existing SDK. Persistent matches keep sending follow-ups;
 add limits or delays in your plugin when needed. Attachments and tool effects are not replayed.
 
@@ -721,22 +726,26 @@ plans, and mode changes; requesting permission does not end the turn.
 
 ### Events
 
-| Name                         | Event fields                             | Trigger                                            |
-| ---------------------------- | ---------------------------------------- | -------------------------------------------------- |
-| `agent.created`              | `agent`                                  | Ordinary creation finishes; excludes import/resume |
-| `agent.closed`               | `agent`                                  | A live agent runtime closes                        |
-| `agent.turn_started`         | `agent`, `turnId`                        | Live turn starts                                   |
-| `agent.turn_ended`           | `agent`, `turnId`, `outcome`, `timeline` | Live turn completes, fails, or is canceled         |
-| `agent.permission_requested` | `agent`, `request`                       | Permission or question becomes pending             |
-| `agent.permission_resolved`  | `agent`, `requestId`, `resolution`       | Pending request is answered or cleared             |
-| `agent.archived`             | `agent`, `archivedAt`                    | Archive state is saved                             |
-| `workspace.created`          | `workspace`                              | Record created; directory available                |
-| `workspace.archived`         | `workspace`                              | Archive state is saved                             |
+| Name                         | Event fields                                            | Trigger                                            |
+| ---------------------------- | ------------------------------------------------------- | -------------------------------------------------- |
+| `agent.created`              | `agent`                                                 | Ordinary creation finishes; excludes import/resume |
+| `agent.closed`               | `agent`                                                 | A live agent runtime closes                        |
+| `agent.turn_started`         | `agent`, `turnId`                                       | Live turn starts                                   |
+| `agent.turn_ended`           | `agent`, `turnId`, `outcome`, deprecated `timeline: []` | Live turn completes, fails, or is canceled         |
+| `agent.permission_requested` | `agent`, `request`                                      | Permission or question becomes pending             |
+| `agent.permission_resolved`  | `agent`, `requestId`, `resolution`                      | Pending request is answered or cleared             |
+| `agent.archived`             | `agent`, `archivedAt`                                   | Archive state is saved                             |
+| `workspace.created`          | `workspace`                                             | Record created; directory available                |
+| `workspace.archived`         | `workspace`                                             | Archive state is saved                             |
 
 Agent events exclude the daemon's private disposable helpers. Public background workspaces and
 their ordinary agents run the same lifecycle hooks as other workspaces. Archive events can precede
 runtime/worktree cleanup; observe lifecycle completion separately when cleanup matters.
 
+`agent.turn_ended.timeline` is deprecated and always empty. Fetch history on demand with
+`context.paseo.agents.ref(event.agent.id).timeline.refetch({ direction: "tail", limit: 20 })`.
+The hook payload size does not grow with conversation history.
+
 **Shared payload shapes** (`@getpaseo/plugin/server`):
 
 ```ts