fix(dictation): retain audio recorded during transcription

Detach committed audio before asynchronous transcription so later chunks retain their own buffer and segment identity. Carry the contributor fix onto current main, retain its existing finalization protection, and verify uploaded PCM through the real SDK HTTP interface.

Co-authored-by: Zhou Yurui <github@yuruiz.com>

Co-authored-by: Joey Hotz <joeyhotz1@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/dictation/dictation-stream-manager.test.ts

@@ -63,6 +63,15 @@ class FakeSttProvider implements SpeechToTextProvider {
   }
 }
 
+class SyncFinalSession extends FakeRealtimeSession {
+  override commit(): void {
+    this.commitCalls += 1;
+    const segmentId = `seg-${this.commitCalls}`;
+    this.emitCommitted(segmentId);
+    this.emitTranscript(segmentId, `transcript ${this.commitCalls}`, true);
+  }
+}
+
 const buildPcmBase64 = (sampleValue: number, sampleCount: number): string => {
   const samples = new Int16Array(sampleCount);
   samples.fill(sampleValue);
@@ -533,6 +542,33 @@ describe("DictationStreamManager (provider-agnostic provider)", () => {
       vi.useRealTimers();
     }
   });
+
+  it("emits one final transcript when synchronous STT events re-enter finalization", async () => {
+    const session = new SyncFinalSession();
+    const emitted: Array<{ type: string; payload: unknown }> = [];
+    const manager = new DictationStreamManager({
+      logger: pino({ level: "silent" }),
+      emit: (msg) => emitted.push(msg),
+      sessionId: "s1",
+      stt: new FakeSttProvider(session),
+      autoCommitSeconds: 0,
+    });
+
+    await manager.handleStart("d-sync-final", "audio/pcm;rate=24000;bits=16");
+    await manager.handleChunk({
+      dictationId: "d-sync-final",
+      seq: 0,
+      audioBase64: buildPcmBase64(2000, 2400),
+      format: "audio/pcm;rate=24000;bits=16",
+    });
+
+    await manager.handleFinish("d-sync-final", 0);
+    await tick();
+
+    const finals = emitted.filter((msg) => msg.type === "dictation_stream_final");
+    expect(finals).toHaveLength(1);
+    expect((finals[0]?.payload as { text?: string } | undefined)?.text).toBe("transcript 1");
+  });
 });
 
 it("cancellation during STT bootstrap closes the producer and never acknowledges a late connection", async () => {

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") {

packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, it } from "vitest";
+
+import { SherpaParakeetRealtimeTranscriptionSession } from "./sherpa-parakeet-realtime-session.js";
+import type { SherpaOfflineRecognizerEngine } from "./sherpa-offline-recognizer.js";
+
+function pcm16(value: number): Buffer {
+  const samples = new Int16Array([value]);
+  return Buffer.from(samples.buffer);
+}
+
+function createEngine(): SherpaOfflineRecognizerEngine {
+  return {
+    sampleRate: 16000,
+    createStream() {
+      return {
+        samples: new Float32Array(0),
+        acceptWaveform(input: { samples: Float32Array }) {
+          this.samples = input.samples;
+        },
+        free() {
+          // no-op
+        },
+      };
+    },
+    acceptWaveform(stream: { samples: Float32Array }, _sampleRate: number, samples: Float32Array) {
+      stream.samples = samples;
+    },
+    recognizer: {
+      decode() {
+        // no-op
+      },
+      getResult(stream: { samples: Float32Array }) {
+        const firstSample = Math.abs(stream.samples[0] ?? 0);
+        return { text: firstSample > 0.4 ? "second" : "first" };
+      },
+    },
+  } as unknown as SherpaOfflineRecognizerEngine;
+}
+
+describe("SherpaParakeetRealtimeTranscriptionSession", () => {
+  it("keeps audio appended after commit for the next segment", async () => {
+    const session = new SherpaParakeetRealtimeTranscriptionSession({
+      engine: createEngine(),
+      minDecodeIntervalMs: Number.MAX_SAFE_INTEGER,
+    });
+    const transcripts: string[] = [];
+
+    session.on("transcript", (event) => {
+      if (event.isFinal) {
+        transcripts.push(event.transcript);
+      }
+    });
+
+    await session.connect();
+    session.appendPcm16(pcm16(0));
+    session.commit();
+    session.appendPcm16(pcm16(24000));
+
+    await expect.poll(() => transcripts.length).toBe(1);
+
+    session.commit();
+
+    await expect.poll(() => transcripts.length).toBe(2);
+
+    expect(transcripts).toEqual(["first", "second"]);
+  });
+});

packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.ts

@@ -58,20 +58,19 @@ export class SherpaParakeetRealtimeTranscriptionSession
       return;
     }
 
+    const segmentId = this.currentSegmentId;
+    const previousSegmentId = this.previousSegmentId;
+    const committedPcm16 = this.pcm16;
+    this.previousSegmentId = segmentId;
+    this.currentSegmentId = uuidv4();
+    this.lastPartialText = "";
+    this.pcm16 = Buffer.alloc(0);
+    this.emit("committed", { segmentId, previousSegmentId });
+
     void (async () => {
       try {
-        await this.maybeDecode(true);
-        const finalText = this.lastPartialText;
-        const segmentId = this.currentSegmentId!;
-        const previousSegmentId = this.previousSegmentId;
-
-        this.emit("committed", { segmentId, previousSegmentId });
+        const finalText = await this.decodePcm16(committedPcm16);
         this.emit("transcript", { segmentId, transcript: finalText, isFinal: true });
-
-        this.previousSegmentId = segmentId;
-        this.currentSegmentId = uuidv4();
-        this.lastPartialText = "";
-        this.pcm16 = Buffer.alloc(0);
       } catch (err) {
         this.emit("error", err instanceof Error ? err : new Error(String(err)));
       }
@@ -110,12 +109,16 @@ export class SherpaParakeetRealtimeTranscriptionSession
 
     this.decoding = true;
     try {
-      const text = await this.decodeNow();
+      const segmentId = this.currentSegmentId;
+      const text = await this.decodePcm16(this.pcm16);
       this.lastDecodeAt = Date.now();
+      if (!this.connected || this.currentSegmentId !== segmentId) {
+        return;
+      }
       if (text !== this.lastPartialText) {
         this.lastPartialText = text;
         this.emit("transcript", {
-          segmentId: this.currentSegmentId,
+          segmentId,
           transcript: text,
           isFinal: false,
         });
@@ -129,12 +132,12 @@ export class SherpaParakeetRealtimeTranscriptionSession
     }
   }
 
-  private async decodeNow(): Promise<string> {
-    if (this.pcm16.length === 0) {
+  private async decodePcm16(pcm16: Buffer): Promise<string> {
+    if (pcm16.length === 0) {
       return "";
     }
 
-    const peak = pcm16lePeakAbs(this.pcm16);
+    const peak = pcm16lePeakAbs(pcm16);
     const peakFloat = peak / 32768.0;
     const targetPeak = 0.6;
     const maxGain = 50;
@@ -143,7 +146,7 @@ export class SherpaParakeetRealtimeTranscriptionSession
 
     const stream = this.engine.createStream();
     try {
-      const floatSamples = pcm16leToFloat32(this.pcm16, gain);
+      const floatSamples = pcm16leToFloat32(pcm16, gain);
       this.engine.acceptWaveform(stream, this.engine.sampleRate, floatSamples);
       this.engine.recognizer.decode(stream);
       const result = this.engine.recognizer.getResult(stream);

packages/server/src/server/speech/providers/openai/stt-segment-retention.test.ts

@@ -0,0 +1,66 @@
+import { createServer, type ServerResponse } from "node:http";
+import { once } from "node:events";
+import { expect, test } from "vitest";
+import pino from "pino";
+import { OpenAISTT } from "./stt.js";
+
+test.each(["before", "after"])(
+  "retains later PCM when stop is %s the first response",
+  async (stop) => {
+    const uploads: Buffer[] = [];
+    const responses: ServerResponse[] = [];
+    // A transcription service fake on the SDK's HTTP interface. The SDK, WAV encoding,
+    // temporary files, session buffer, and event handling all run unchanged.
+    const server = createServer(async (request, response) => {
+      const chunks: Buffer[] = [];
+      for await (const chunk of request) chunks.push(Buffer.from(chunk));
+      const body = Buffer.concat(chunks);
+      const wav = body.subarray(body.indexOf(Buffer.from("RIFF")));
+      uploads.push(wav.subarray(44, 44 + wav.readUInt32LE(40)));
+      responses.push(response);
+    });
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("Missing server address");
+    const logger = pino({ level: "silent" });
+    const session = new OpenAISTT(
+      { apiKey: "test", baseUrl: `http://127.0.0.1:${address.port}/v1` },
+      logger,
+    ).createSession({ logger });
+    const transcripts: string[] = [];
+    session.on("transcript", (event) => {
+      if (event.isFinal) transcripts.push(event.transcript);
+    });
+    const finish = (index: number, text: string) => {
+      responses[index]!.writeHead(200, { "Content-Type": "application/json" });
+      responses[index]!.end(JSON.stringify({ text }));
+    };
+    try {
+      await session.connect();
+      session.appendPcm16(Buffer.from([1, 0, 2, 0]));
+      session.commit();
+      await expect.poll(() => uploads.length).toBe(1);
+      session.appendPcm16(Buffer.from([3, 0, 4, 0]));
+      if (stop === "after") {
+        finish(0, "first");
+        await expect.poll(() => transcripts.length).toBe(1);
+      }
+      session.commit();
+      await expect.poll(() => uploads.length, { timeout: 1000 }).toBe(2);
+      expect(uploads.map((pcm) => Array.from(pcm))).toEqual([
+        [1, 0, 2, 0],
+        [3, 0, 4, 0],
+      ]);
+      if (stop === "before") finish(0, "first");
+      finish(1, "second");
+      await expect.poll(() => transcripts.length).toBe(2);
+    } finally {
+      session.close();
+      for (const response of responses)
+        if (!response.writableEnded) response.end(JSON.stringify({ text: "" }));
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  },
+);

packages/server/src/server/speech/providers/openai/stt.ts

@@ -124,11 +124,15 @@ export class OpenAISTT implements SpeechToTextProvider {
 
         const committedId = segmentId;
         const prev = previousSegmentId;
+        const committedPcm16 = pcm16;
+        previousSegmentId = committedId;
+        segmentId = v4();
+        pcm16 = Buffer.alloc(0);
         emitter.emit("committed", { segmentId: committedId, previousSegmentId: prev });
 
         void (async () => {
           try {
-            if (pcm16.length === 0) {
+            if (committedPcm16.length === 0) {
               emitter.emit("transcript", {
                 segmentId: committedId,
                 transcript: "",
@@ -139,7 +143,7 @@ export class OpenAISTT implements SpeechToTextProvider {
               return;
             }
 
-            const wav = convertPCMToWavBuffer(pcm16);
+            const wav = convertPCMToWavBuffer(committedPcm16);
             const result = await transcribeAudio(
               wav,
               "audio/wav",
@@ -159,10 +163,6 @@ export class OpenAISTT implements SpeechToTextProvider {
             });
           } catch (err) {
             emitter.emit("error", err);
-          } finally {
-            previousSegmentId = committedId;
-            segmentId = v4();
-            pcm16 = Buffer.alloc(0);
           }
         })();
       },

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.");
     }

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