diff --git a/docs/providers.md b/docs/providers.md index ca949357a..ddc4da6f2 100644 --- a/docs/providers.md +++ b/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. --- diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 465f98a4b..27d5c60e4 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/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 { + 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"); diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index d8eb8eb2a..84632dda1 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/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; diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 492e4ffd8..6331a9d7c 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/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, }); } diff --git a/packages/server/src/server/agent/create-agent/create.test.ts b/packages/server/src/server/agent/create-agent/create.test.ts new file mode 100644 index 000000000..3a97c4a1e --- /dev/null +++ b/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[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[0]["agentManager"], + agentStorage: {} as Parameters[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", + }); +}); diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index 600c09562..d38a35c43 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/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, diff --git a/packages/server/src/server/agent/providers/acp-agent.test.ts b/packages/server/src/server/agent/providers/acp-agent.test.ts index 1e13ef190..cfab810bc 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -1641,6 +1641,72 @@ describe("ACPAgentSession", () => { expect(asInternals(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((resolve) => { + resolvePrompt = resolve; + }), + ); + + asInternals(session).sessionId = "session-1"; + asInternals(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(() => {})); + + asInternals(session).sessionId = "session-1"; + asInternals(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 }> = []; diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 5b5f3657c..17ea0146f 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/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(); private readonly messageAssemblies = new Map(); + private readonly submittedUserMessageIds = new Set(); private readonly toolCalls = new Map(); private readonly terminalEntries = new Map(); 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, ): void { this.activeForegroundTurnId = null; - this.suppressUserEchoMessageId = null; - this.suppressUserEchoText = null; this.pushEvent(event); }