From 91634ca6f23b19a6772f680367dd662932234acb Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 25 Feb 2026 19:53:04 +0700 Subject: [PATCH] fix(server): dedupe opencode assistant timeline echoes --- .../server/agent/providers/opencode-agent.ts | 423 ++++++++++-------- .../opencode/event-translator.test.ts | 174 +++++++ 2 files changed, 410 insertions(+), 187 deletions(-) create mode 100644 packages/server/src/server/agent/providers/opencode/event-translator.test.ts diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index d359948ae..320269ebc 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -53,6 +53,7 @@ const DEFAULT_MODES: AgentMode[] = [ ]; type OpenCodeAgentConfig = AgentSessionConfig & { provider: "opencode" }; +type OpenCodeMessageRole = "user" | "assistant"; type OpenCodeMcpConfig = | { @@ -214,6 +215,21 @@ async function findAvailablePort(): Promise { }); } +function resolvePartDedupeKey( + part: AgentMetadata, + partType: "text" | "reasoning" +): string | null { + const partId = part.id; + if (typeof partId === "string" && partId.trim().length > 0) { + return `${partType}:${partId}`; + } + const messageId = part.messageID; + if (typeof messageId === "string" && messageId.trim().length > 0) { + return `${partType}:message:${messageId}`; + } + return null; +} + export class OpenCodeServerManager { private static instance: OpenCodeServerManager | null = null; private static exitHandlerRegistered = false; @@ -523,6 +539,211 @@ export class OpenCodeAgentClient implements AgentClient { } } +export type OpenCodeEventTranslationState = { + sessionId: string; + messageRoles: Map; + accumulatedUsage: AgentUsage; + streamedPartKeys: Set; +}; + +export function translateOpenCodeEvent( + event: unknown, + state: OpenCodeEventTranslationState +): AgentStreamEvent[] { + const events: AgentStreamEvent[] = []; + + if (!event || typeof event !== "object") { + return events; + } + + const e = event as { type?: string; properties?: AgentMetadata }; + const type = e.type; + const props = e.properties ?? {}; + + switch (type) { + case "session.created": + case "session.updated": { + const sessionId = props.id as string | undefined; + if (sessionId === state.sessionId) { + events.push({ + type: "thread_started", + sessionId: state.sessionId, + provider: "opencode", + }); + } + break; + } + + case "message.updated": { + const info = props.info as AgentMetadata | undefined; + if (!info) { + break; + } + const messageId = info.id as string | undefined; + const messageSessionId = info.sessionID as string | undefined; + const role = info.role as OpenCodeMessageRole | undefined; + + if (messageId && messageSessionId === state.sessionId && role) { + state.messageRoles.set(messageId, role); + } + break; + } + + case "message.part.updated": { + const part = props.part as AgentMetadata | undefined; + const delta = props.delta as string | undefined; + if (!part) { + break; + } + + const partSessionId = part.sessionID as string | undefined; + if (partSessionId !== state.sessionId) { + break; + } + + const messageId = part.messageID as string | undefined; + const messageRole = messageId ? state.messageRoles.get(messageId) : undefined; + const partType = part.type as string | undefined; + const partTime = part.time as { start?: number; end?: number } | undefined; + + if (partType === "text") { + const partKey = resolvePartDedupeKey(part, "text"); + if (messageRole === "user") { + break; + } + if (!messageRole && !delta) { + break; + } + if (delta) { + if (partKey) { + state.streamedPartKeys.add(partKey); + } + events.push({ + type: "timeline", + provider: "opencode", + item: { type: "assistant_message", text: delta }, + }); + } else if (partTime?.end) { + if (partKey && state.streamedPartKeys.delete(partKey)) { + break; + } + const text = part.text as string | undefined; + if (text) { + events.push({ + type: "timeline", + provider: "opencode", + item: { type: "assistant_message", text }, + }); + } + } + } else if (partType === "reasoning") { + const partKey = resolvePartDedupeKey(part, "reasoning"); + if (delta) { + if (partKey) { + state.streamedPartKeys.add(partKey); + } + events.push({ + type: "timeline", + provider: "opencode", + item: { type: "reasoning", text: delta }, + }); + } else if (partTime?.end) { + if (partKey && state.streamedPartKeys.delete(partKey)) { + break; + } + const text = part.text as string | undefined; + if (text) { + events.push({ + type: "timeline", + provider: "opencode", + item: { type: "reasoning", text }, + }); + } + } + } else if (partType === "tool") { + const parsedToolPart = OpencodeToolPartToTimelineItemSchema.safeParse(part); + if (parsedToolPart.success && parsedToolPart.data) { + events.push({ + type: "timeline", + provider: "opencode", + item: parsedToolPart.data, + }); + } + } else if (partType === "step-finish") { + const tokens = part.tokens as { input?: number; output?: number; reasoning?: number } | undefined; + const cost = part.cost as number | undefined; + + if (tokens) { + state.accumulatedUsage.inputTokens = (state.accumulatedUsage.inputTokens ?? 0) + (tokens.input ?? 0); + state.accumulatedUsage.outputTokens = (state.accumulatedUsage.outputTokens ?? 0) + (tokens.output ?? 0); + } + if (cost !== undefined) { + state.accumulatedUsage.totalCostUsd = (state.accumulatedUsage.totalCostUsd ?? 0) + cost; + } + } + break; + } + + case "permission.asked": { + const sessionId = props.sessionID as string | undefined; + if (sessionId !== state.sessionId) { + break; + } + + const requestId = props.id as string; + const permission = props.permission as string; + const metadata = props.metadata as AgentMetadata | undefined; + const patterns = props.patterns as string[] | undefined; + + const permRequest: AgentPermissionRequest = { + id: requestId, + provider: "opencode", + name: permission, + kind: "tool", + title: permission, + description: patterns?.join(", "), + input: metadata, + }; + + events.push({ + type: "permission_requested", + provider: "opencode", + request: permRequest, + }); + break; + } + + case "session.idle": { + const sessionId = props.sessionID as string | undefined; + if (sessionId === state.sessionId) { + state.streamedPartKeys.clear(); + events.push({ + type: "turn_completed", + provider: "opencode", + usage: undefined, + }); + } + break; + } + + case "session.error": { + const sessionId = props.sessionID as string | undefined; + if (sessionId === state.sessionId) { + state.streamedPartKeys.clear(); + const error = props.error as string | undefined; + events.push({ + type: "turn_failed", + provider: "opencode", + error: error ?? "Unknown error", + }); + } + break; + } + } + + return events; +} + class OpenCodeAgentSession implements AgentSession { readonly provider: "opencode" = "opencode"; readonly capabilities = OPENCODE_CAPABILITIES; @@ -537,7 +758,9 @@ class OpenCodeAgentSession implements AgentSession { private mcpConfigured = false; private mcpSetupPromise: Promise | null = null; /** Tracks the role of each message by ID to distinguish user from assistant messages */ - private messageRoles = new Map(); + private messageRoles = new Map(); + /** Tracks streamed textual part IDs to suppress final full-text echoes from OpenCode. */ + private streamedPartKeys = new Set(); constructor( config: OpenCodeAgentConfig, @@ -886,197 +1109,23 @@ class OpenCodeAgentSession implements AgentSession { } private translateEvent(event: unknown): AgentStreamEvent[] { - const events: AgentStreamEvent[] = []; + const translated = translateOpenCodeEvent(event, { + sessionId: this.sessionId, + messageRoles: this.messageRoles, + accumulatedUsage: this.accumulatedUsage, + streamedPartKeys: this.streamedPartKeys, + }); - if (!event || typeof event !== "object") { - return events; - } - - const e = event as { type?: string; properties?: AgentMetadata }; - const type = e.type; - const props = e.properties ?? {}; - - switch (type) { - case "session.created": - case "session.updated": { - const sessionId = props.id as string | undefined; - if (sessionId === this.sessionId) { - events.push({ - type: "thread_started", - sessionId: this.sessionId, - provider: "opencode", - }); - } - break; + for (const translatedEvent of translated) { + if (translatedEvent.type === "permission_requested") { + this.pendingPermissions.set(translatedEvent.request.id, translatedEvent.request); } - - case "message.updated": { - // Track message roles by ID so we can distinguish user from assistant parts - const info = props.info as AgentMetadata | undefined; - if (!info) { - break; - } - const messageId = info.id as string | undefined; - const messageSessionId = info.sessionID as string | undefined; - const role = info.role as "user" | "assistant" | undefined; - - if (messageId && messageSessionId === this.sessionId && role) { - this.messageRoles.set(messageId, role); - } - break; - } - - case "message.part.updated": { - // Structure: { part: { id, sessionID, messageID, type, text?, ... }, delta?: string } - const part = props.part as AgentMetadata | undefined; - const delta = props.delta as string | undefined; - if (!part) { - break; - } - - const partSessionId = part.sessionID as string | undefined; - if (partSessionId !== this.sessionId) { - break; - } - - const messageId = part.messageID as string | undefined; - const messageRole = messageId ? this.messageRoles.get(messageId) : undefined; - const partType = part.type as string | undefined; - const partTime = part.time as { start?: number; end?: number } | undefined; - - if (partType === "text") { - // Skip user messages - agent-manager emits user_message via recordUserMessage - if (messageRole === "user") { - break; - } - // Skip if role unknown AND no delta (likely user message before role is known) - if (!messageRole && !delta) { - break; - } - // Emit delta for streaming, or full text only when complete (has time.end) - if (delta) { - events.push({ - type: "timeline", - provider: "opencode", - item: { type: "assistant_message", text: delta }, - }); - } else if (partTime?.end) { - // Final complete text - only emit if we haven't been streaming deltas - // (In practice, if delta was present, we've already streamed it all) - const text = part.text as string | undefined; - if (text) { - events.push({ - type: "timeline", - provider: "opencode", - item: { type: "assistant_message", text }, - }); - } - } - } else if (partType === "reasoning") { - // Emit delta for streaming reasoning - if (delta) { - events.push({ - type: "timeline", - provider: "opencode", - item: { type: "reasoning", text: delta }, - }); - } else if (partTime?.end) { - const text = part.text as string | undefined; - if (text) { - events.push({ - type: "timeline", - provider: "opencode", - item: { type: "reasoning", text }, - }); - } - } - } else if (partType === "tool") { - // Tool parts: { tool: string, state: { status, input, output?, error? } } - const parsedToolPart = OpencodeToolPartToTimelineItemSchema.safeParse(part); - if (parsedToolPart.success) { - if (parsedToolPart.data) { - events.push({ - type: "timeline", - provider: "opencode", - item: parsedToolPart.data, - }); - } - } - } else if (partType === "step-finish") { - // Extract usage from step-finish parts - const tokens = part.tokens as { input?: number; output?: number; reasoning?: number } | undefined; - const cost = part.cost as number | undefined; - - if (tokens) { - this.accumulatedUsage.inputTokens = (this.accumulatedUsage.inputTokens ?? 0) + (tokens.input ?? 0); - this.accumulatedUsage.outputTokens = (this.accumulatedUsage.outputTokens ?? 0) + (tokens.output ?? 0); - } - if (cost !== undefined) { - this.accumulatedUsage.totalCostUsd = (this.accumulatedUsage.totalCostUsd ?? 0) + cost; - } - } - break; - } - - case "permission.asked": { - // props IS the PermissionRequest directly - const sessionId = props.sessionID as string | undefined; - if (sessionId !== this.sessionId) { - break; - } - - const requestId = props.id as string; - const permission = props.permission as string; - const metadata = props.metadata as AgentMetadata | undefined; - const patterns = props.patterns as string[] | undefined; - - const permRequest: AgentPermissionRequest = { - id: requestId, - provider: "opencode", - name: permission, - kind: "tool", - title: permission, - description: patterns?.join(", "), - input: metadata, - }; - - this.pendingPermissions.set(requestId, permRequest); - events.push({ - type: "permission_requested", - provider: "opencode", - request: permRequest, - }); - break; - } - - case "session.idle": { - const sessionId = props.sessionID as string | undefined; - if (sessionId === this.sessionId) { - const usage = this.extractAndResetUsage(); - events.push({ - type: "turn_completed", - provider: "opencode", - usage, - }); - } - break; - } - - case "session.error": { - const sessionId = props.sessionID as string | undefined; - if (sessionId === this.sessionId) { - const error = props.error as string | undefined; - events.push({ - type: "turn_failed", - provider: "opencode", - error: error ?? "Unknown error", - }); - } - break; + if (translatedEvent.type === "turn_completed") { + translatedEvent.usage = this.extractAndResetUsage(); } } - return events; + return translated; } private extractAndResetUsage(): AgentUsage | undefined { diff --git a/packages/server/src/server/agent/providers/opencode/event-translator.test.ts b/packages/server/src/server/agent/providers/opencode/event-translator.test.ts new file mode 100644 index 000000000..6e1e9746c --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode/event-translator.test.ts @@ -0,0 +1,174 @@ +import { describe, expect, it } from "vitest"; + +import { + translateOpenCodeEvent, + type OpenCodeEventTranslationState, +} from "../opencode-agent.js"; + +function createState(sessionId = "session-1"): OpenCodeEventTranslationState { + return { + sessionId, + messageRoles: new Map(), + accumulatedUsage: {}, + streamedPartKeys: new Set(), + }; +} + +describe("translateOpenCodeEvent", () => { + it("does not duplicate assistant output when completed part echoes streamed delta", () => { + const state = createState(); + + translateOpenCodeEvent( + { + type: "message.updated", + properties: { + info: { + id: "message-1", + sessionID: "session-1", + role: "assistant", + }, + }, + }, + state + ); + + const streamed = translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + delta: "hey! what can I help with?", + part: { + id: "part-1", + sessionID: "session-1", + messageID: "message-1", + type: "text", + time: { start: 1 }, + }, + }, + }, + state + ); + + const completed = translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + part: { + id: "part-1", + sessionID: "session-1", + messageID: "message-1", + type: "text", + text: "hey! what can I help with?", + time: { start: 1, end: 2 }, + }, + }, + }, + state + ); + + const assistantEvents = [...streamed, ...completed].filter( + (event) => event.type === "timeline" && event.item.type === "assistant_message" + ); + + expect(assistantEvents).toEqual([ + { + type: "timeline", + provider: "opencode", + item: { type: "assistant_message", text: "hey! what can I help with?" }, + }, + ]); + }); + + it("emits completed assistant text when no delta was streamed", () => { + const state = createState(); + + translateOpenCodeEvent( + { + type: "message.updated", + properties: { + info: { + id: "message-2", + sessionID: "session-1", + role: "assistant", + }, + }, + }, + state + ); + + const completed = translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + part: { + id: "part-2", + sessionID: "session-1", + messageID: "message-2", + type: "text", + text: "final text", + time: { start: 3, end: 4 }, + }, + }, + }, + state + ); + + expect(completed).toEqual([ + { + type: "timeline", + provider: "opencode", + item: { type: "assistant_message", text: "final text" }, + }, + ]); + }); + + it("does not duplicate reasoning output when completed part echoes streamed delta", () => { + const state = createState(); + + const streamed = translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + delta: "The user said hello.", + part: { + id: "reasoning-part-1", + sessionID: "session-1", + messageID: "message-3", + type: "reasoning", + time: { start: 10 }, + }, + }, + }, + state + ); + + const completed = translateOpenCodeEvent( + { + type: "message.part.updated", + properties: { + part: { + id: "reasoning-part-1", + sessionID: "session-1", + messageID: "message-3", + type: "reasoning", + text: "The user said hello.", + time: { start: 10, end: 11 }, + }, + }, + }, + state + ); + + const reasoningEvents = [...streamed, ...completed].filter( + (event) => event.type === "timeline" && event.item.type === "reasoning" + ); + + expect(reasoningEvents).toEqual([ + { + type: "timeline", + provider: "opencode", + item: { type: "reasoning", text: "The user said hello." }, + }, + ]); + }); +});