Persist submitted ACP user messages

docs/providers.md

@@ -30,6 +30,8 @@ OpenCode MCP injection is dynamic and session-scoped. Call OpenCode's `mcp.add`
 
 OpenCode owns user message IDs. Do not pass Paseo-generated IDs to OpenCode prompt APIs; let OpenCode create `msg*` IDs and record the user timeline item from the `message.updated` event.
 
+Every provider adapter owns its canonical user-message timeline rows. When a foreground prompt is accepted, the adapter must emit exactly one `user_message` timeline item for that submitted prompt, using the same message ID it gives to or receives from the provider runtime. Optimistic client messages are UI-only and provider transcript echoes are optional; neither is allowed to be the only source of truth. If the provider later echoes the same submitted user message, dedupe by provider-visible message ID, not by text.
+
 Draft metadata lookups should avoid creating provider sessions when the upstream provider has top-level APIs for that metadata. Prefer `AgentClient.listModels`, `listModes`, `listCommands`, or `listFeatures` over creating a scratch `AgentSession`; scratch sessions can show up as empty native sessions in provider import/history UIs.
 
 ---

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

@@ -16,8 +16,10 @@ import type {
   AgentCreateSessionOptions,
   AgentFeature,
   AgentLaunchContext,
+  AgentPromptInput,
   AgentProvider,
   AgentPersistenceHandle,
+  AgentRunOptions,
   AgentRunResult,
   AgentSession,
   AgentSessionConfig,
@@ -5443,6 +5445,63 @@ test("provider user_message is recorded from the live stream", async () => {
   expect(userMessages[0].text).toBe("continuation prompt");
 });
 
+test("authoritative timeline includes provider-emitted submitted user prompt", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-submitted-prompt-"));
+  const storagePath = join(workdir, "agents");
+  const storage = new AgentStorage(storagePath, logger);
+
+  class SubmittedUserMessageSession extends TestAgentSession {
+    override async startTurn(
+      prompt: AgentPromptInput,
+      options?: AgentRunOptions,
+    ): Promise<{ turnId: string }> {
+      const turnId = "turn-submitted-user-message";
+      const text = typeof prompt === "string" ? prompt : "";
+      setTimeout(() => {
+        this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
+        this.pushEvent({
+          type: "timeline",
+          provider: this.provider,
+          turnId,
+          item: { type: "user_message", text, messageId: options?.messageId },
+        });
+        this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
+      }, 0);
+      return { turnId };
+    }
+  }
+
+  class SubmittedUserMessageClient extends TestAgentClient {
+    override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
+      return new SubmittedUserMessageSession(config);
+    }
+  }
+
+  const manager = new AgentManager({
+    clients: { codex: new SubmittedUserMessageClient() },
+    registry: storage,
+    logger,
+    idFactory: () => "00000000-0000-4000-8000-000000000402",
+  });
+
+  try {
+    const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir });
+
+    await manager.runAgent(snapshot.id, "hello from composer", { messageId: "msg-client-1" });
+
+    const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows;
+    expect(timeline.map((row) => row.item)).toContainEqual({
+      type: "user_message",
+      text: "hello from composer",
+      messageId: "msg-client-1",
+    });
+  } finally {
+    await manager.flush().catch(() => undefined);
+    await storage.flush().catch(() => undefined);
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});
+
 test("replaceAgentRun succeeds when foreground turn terminal event is never delivered", async () => {
   const workdir = mkdtempSync(join(tmpdir(), "agent-manager-stale-fg-"));
   const storagePath = join(workdir, "agents");

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

@@ -6,6 +6,7 @@ import { AgentStorage } from "./agent-storage.js";
 import {
   formatSystemNotificationPrompt,
   isSystemInjectedEnvelope,
+  sendPromptToAgent,
   setupFinishNotification,
 } from "./agent-prompt.js";
 import type { AgentManagerEvent, ManagedAgent } from "./agent-manager.js";
@@ -15,6 +16,45 @@ test("isSystemInjectedEnvelope matches the envelope formatSystemNotificationProm
   expect(isSystemInjectedEnvelope("hello world")).toBe(false);
 });
 
+test("sendPromptToAgent forwards the client message id as run options", async () => {
+  const agent: ManagedAgent = Object.create(null);
+  Reflect.set(agent, "id", "agent-1");
+  Reflect.set(agent, "provider", "codex");
+
+  const streamAgentSpy = vi.fn(() => (async function* noop() {})());
+  const agentManager: AgentManager = Object.create(AgentManager.prototype);
+  Reflect.set(
+    agentManager,
+    "getAgent",
+    vi.fn(() => agent),
+  );
+  Reflect.set(agentManager, "tryRunOutOfBand", vi.fn().mockReturnValue(false));
+  Reflect.set(agentManager, "hasInFlightRun", vi.fn().mockReturnValue(false));
+  Reflect.set(agentManager, "streamAgent", streamAgentSpy);
+
+  const agentStorage: AgentStorage = Object.create(AgentStorage.prototype);
+  Reflect.set(
+    agentStorage,
+    "get",
+    vi.fn(async () => null),
+  );
+
+  await sendPromptToAgent({
+    agentManager,
+    agentStorage,
+    agentId: "agent-1",
+    prompt: "hello",
+    messageId: "msg-client-1",
+    runOptions: { outputSchema: { type: "object" } },
+    logger: createTestLogger(),
+  });
+
+  expect(streamAgentSpy).toHaveBeenCalledWith("agent-1", "hello", {
+    outputSchema: { type: "object" },
+    messageId: "msg-client-1",
+  });
+});
+
 it("does not notify archived callers", async () => {
   let subscriber: ((event: AgentManagerEvent) => void) | null = null;
 

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

@@ -201,9 +201,13 @@ export async function sendPromptToAgent(
     await params.agentManager.setAgentMode(params.agentId, params.sessionMode);
   }
 
+  const runOptions = params.messageId
+    ? { ...params.runOptions, messageId: params.messageId }
+    : params.runOptions;
+
   return startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, {
     replaceRunning: true,
-    runOptions: params.runOptions,
+    runOptions,
   });
 }
 

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

@@ -0,0 +1,47 @@
+import { expect, test, vi } from "vitest";
+
+import { createTestLogger } from "../../../test-utils/test-logger.js";
+import { createAgentCommand } from "./create.js";
+import type { ManagedAgent } from "../agent-manager.js";
+
+test("session create forwards clientMessageId to the initial prompt run options", async () => {
+  const snapshot = {
+    id: "agent-1",
+    provider: "codex",
+    cwd: "/tmp/paseo-create-test",
+    runtimeInfo: null,
+  } as ManagedAgent;
+  const streamAgent = vi.fn(() => (async function* noop() {})());
+  const dependencies: Parameters<typeof createAgentCommand>[0] = {
+    agentManager: {
+      createAgent: vi.fn(async () => snapshot),
+      getAgent: vi.fn(() => snapshot),
+      tryRunOutOfBand: vi.fn(() => false),
+      hasInFlightRun: vi.fn(() => false),
+      streamAgent,
+      waitForAgentRunStart: vi.fn(async () => undefined),
+    } as unknown as Parameters<typeof createAgentCommand>[0]["agentManager"],
+    agentStorage: {} as Parameters<typeof createAgentCommand>[0]["agentStorage"],
+    logger: createTestLogger(),
+    providerSnapshotManager: {} as Parameters<
+      typeof createAgentCommand
+    >[0]["providerSnapshotManager"],
+  };
+
+  await createAgentCommand(dependencies, {
+    kind: "session",
+    config: { provider: "codex", cwd: "/tmp/paseo-create-test" },
+    initialPrompt: "hello from create",
+    clientMessageId: "msg-create-1",
+    labels: {},
+    provisionalTitle: null,
+    explicitTitle: "Explicit title",
+    firstAgentContext: { attachments: [] },
+    buildSessionConfig: async (config) => ({ sessionConfig: config }),
+    resolveWorkspace: async () => ({ workspaceId: "workspace-1" }),
+  });
+
+  expect(streamAgent).toHaveBeenCalledWith("agent-1", "hello from create", {
+    messageId: "msg-create-1",
+  });
+});

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

@@ -25,7 +25,7 @@ import type {
 import type { AgentStorage } from "../agent-storage.js";
 import type { ProviderSnapshotManager } from "../provider-snapshot-manager.js";
 import { setupFinishNotification, startCreatedAgentInitialPrompt } from "../agent-prompt.js";
-import { resolveClientMessageId } from "../../client-message-id.js";
+import { normalizeClientMessageId, resolveClientMessageId } from "../../client-message-id.js";
 import { resolveRequiredProviderModel } from "../mcp-shared.js";
 import {
   appendTimelineItemIfAgentKnown,
@@ -202,6 +202,14 @@ async function resolveSessionCreateAgent(
   });
   const prompt = buildAgentPrompt(trimmedPrompt ?? "", input.images, input.attachments);
   const hasPromptContent = Array.isArray(prompt) ? prompt.length > 0 : prompt.length > 0;
+  const clientMessageId = normalizeClientMessageId(input.clientMessageId);
+  const runOptions: AgentRunOptions | undefined =
+    input.outputSchema || clientMessageId
+      ? {
+          ...(input.outputSchema ? { outputSchema: input.outputSchema } : {}),
+          ...(clientMessageId ? { messageId: clientMessageId } : {}),
+        }
+      : undefined;
 
   return {
     config: sessionConfig,
@@ -214,7 +222,7 @@ async function resolveSessionCreateAgent(
     },
     metadataInitialPrompt: trimmedPrompt,
     prompt: hasPromptContent ? prompt : undefined,
-    runOptions: input.outputSchema ? { outputSchema: input.outputSchema } : undefined,
+    runOptions,
     explicitTitle: input.explicitTitle,
     setupContinuation,
     background: true,

packages/server/src/server/agent/providers/acp-agent.test.ts

@@ -1641,6 +1641,72 @@ describe("ACPAgentSession", () => {
     expect(asInternals<ACPSessionInternals>(session).activeForegroundTurnId).toBeNull();
   });
 
+  test("startTurn emits the submitted user message even when ACP does not echo it", async () => {
+    const session = createSession();
+    const events: AgentStreamEvent[] = [];
+    let resolvePrompt!: (value: PromptResponse) => void;
+    const prompt = vi.fn(
+      () =>
+        new Promise<PromptResponse>((resolve) => {
+          resolvePrompt = resolve;
+        }),
+    );
+
+    asInternals<ACPSessionInternals>(session).sessionId = "session-1";
+    asInternals<ACPSessionInternals>(session).connection = { prompt };
+
+    session.subscribe((event) => {
+      events.push(event);
+    });
+
+    const { turnId } = await session.startTurn("hello", { messageId: "msg-client-1" });
+
+    expect(prompt).toHaveBeenCalledWith({
+      sessionId: "session-1",
+      messageId: "msg-client-1",
+      prompt: [{ type: "text", text: "hello" }],
+    });
+    expect(
+      events.filter((event) => event.type === "timeline" && event.item.type === "user_message"),
+    ).toEqual([
+      {
+        type: "timeline",
+        provider: "claude-acp",
+        turnId,
+        item: { type: "user_message", text: "hello", messageId: "msg-client-1" },
+      },
+    ]);
+
+    resolvePrompt({ stopReason: "end_turn" });
+  });
+
+  test("startTurn dedupes ACP user echo chunks for the submitted message", async () => {
+    const session = createSession();
+    const events: AgentStreamEvent[] = [];
+    const prompt = vi.fn(() => new Promise<PromptResponse>(() => {}));
+
+    asInternals<ACPSessionInternals>(session).sessionId = "session-1";
+    asInternals<ACPSessionInternals>(session).connection = { prompt };
+
+    session.subscribe((event) => {
+      events.push(event);
+    });
+
+    await session.startTurn("hello", { messageId: "msg-client-1" });
+    await session.sessionUpdate({
+      sessionId: "session-1",
+      update: {
+        sessionUpdate: "user_message_chunk",
+        messageId: "msg-client-1",
+        content: { type: "text", text: "hello" },
+      } as SessionUpdate,
+    });
+
+    expect(
+      events.filter((event) => event.type === "timeline" && event.item.type === "user_message"),
+    ).toHaveLength(1);
+  });
+
   test("startTurn converts background prompt rejections into turn_failed events", async () => {
     const session = createSession();
     const events: Array<{ type: string; turnId?: string; error?: string }> = [];

packages/server/src/server/agent/providers/acp-agent.ts

@@ -929,6 +929,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
   private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
   private readonly pendingPermissions = new Map<string, PendingPermission>();
   private readonly messageAssemblies = new Map<string, MessageAssemblyState>();
+  private readonly submittedUserMessageIds = new Set<string>();
   private readonly toolCalls = new Map<string, ACPToolSnapshot>();
   private readonly terminalEntries = new Map<string, TerminalEntry>();
   private readonly persistedHistory: AgentTimelineItem[] = [];
@@ -957,8 +958,6 @@ export class ACPAgentSession implements AgentSession, ACPClient {
   private closed = false;
   private historyPending = false;
   private replayingHistory = false;
-  private suppressUserEchoMessageId: string | null = null;
-  private suppressUserEchoText: string | null = null;
   private bootstrapThreadEventPending = false;
 
   constructor(config: AgentSessionConfig, options: ACPAgentSessionOptions) {
@@ -1066,7 +1065,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
 
   async startTurn(
     prompt: AgentPromptInput,
-    _options?: AgentRunOptions,
+    options?: AgentRunOptions,
   ): Promise<{ turnId: string }> {
     if (this.closed) {
       throw new Error(`${this.provider} session is closed`);
@@ -1079,12 +1078,11 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     }
 
     const turnId = randomUUID();
-    const messageId = randomUUID();
+    const messageId = options?.messageId ?? randomUUID();
     this.activeForegroundTurnId = turnId;
-    this.suppressUserEchoMessageId = messageId;
-    this.suppressUserEchoText = extractPromptText(prompt);
     this.emitBootstrapThreadEvent();
     this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
+    this.emitSubmittedUserMessage(prompt, messageId, turnId);
 
     void this.connection
       .prompt({
@@ -1945,12 +1943,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
         if (!item) {
           return [];
         }
-        const shouldSuppress =
-          this.suppressUserEchoMessageId &&
-          update.messageId === this.suppressUserEchoMessageId &&
-          this.suppressUserEchoText &&
-          item.text === this.suppressUserEchoText;
-        if (shouldSuppress) {
+        if (update.messageId && this.submittedUserMessageIds.has(update.messageId)) {
           return [];
         }
         return [this.wrapTimeline(item)];
@@ -2154,6 +2147,24 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     }
   }
 
+  private emitSubmittedUserMessage(
+    prompt: AgentPromptInput,
+    messageId: string,
+    turnId: string,
+  ): void {
+    const text = extractPromptText(prompt);
+    if (text.trim().length === 0) {
+      return;
+    }
+    this.submittedUserMessageIds.add(messageId);
+    this.pushEvent({
+      type: "timeline",
+      provider: this.provider,
+      turnId,
+      item: { type: "user_message", text, messageId },
+    });
+  }
+
   private runtimeInfo(): AgentRuntimeInfo {
     return {
       provider: this.provider,
@@ -2172,8 +2183,6 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     event: Extract<AgentStreamEvent, { type: "turn_completed" | "turn_failed" | "turn_canceled" }>,
   ): void {
     this.activeForegroundTurnId = null;
-    this.suppressUserEchoMessageId = null;
-    this.suppressUserEchoText = null;
     this.pushEvent(event);
   }