diff --git a/CHANGELOG.md b/CHANGELOG.md index 95c9c8c70..75f1485e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ - macOS: treat location permission as always-only to avoid iOS-only enums. (#165) — thanks @Nachx639 - macOS: make generated gateway protocol models `Sendable` for Swift 6 strict concurrency. (#195) — thanks @andranik-sahakyan - WhatsApp: suppress typing indicator during heartbeat background tasks. (#190) — thanks @mcinteerj +- Discord: avoid duplicate replies when a provider emits late streaming `text_end` events (OpenAI/GPT). - Env: load global `$CLAWDBOT_STATE_DIR/.env` (`~/.clawdbot/.env`) as a fallback after CWD `.env`. - Agent tools: OpenAI-compatible tool JSON Schemas (fix `browser`, normalize union schemas). - Onboarding: when running from source, auto-build missing Control UI assets (`pnpm ui:build`). diff --git a/src/agents/pi-embedded-subscribe.test.ts b/src/agents/pi-embedded-subscribe.test.ts index 8799ff70b..ca0782909 100644 --- a/src/agents/pi-embedded-subscribe.test.ts +++ b/src/agents/pi-embedded-subscribe.test.ts @@ -182,6 +182,110 @@ describe("subscribeEmbeddedPiSession", () => { expect(subscription.assistantTexts).toEqual(["Hello block"]); }); + it("does not duplicate when message_end flushes and a late text_end arrives", () => { + let handler: ((evt: unknown) => void) | undefined; + const session: StubSession = { + subscribe: (fn) => { + handler = fn; + return () => {}; + }, + }; + + const onBlockReply = vi.fn(); + + const subscription = subscribeEmbeddedPiSession({ + session: session as unknown as Parameters< + typeof subscribeEmbeddedPiSession + >[0]["session"], + runId: "run", + onBlockReply, + blockReplyBreak: "text_end", + }); + + handler?.({ type: "message_start", message: { role: "assistant" } }); + + handler?.({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { + type: "text_delta", + delta: "Hello block", + }, + }); + + const assistantMessage = { + role: "assistant", + content: [{ type: "text", text: "Hello block" }], + } as AssistantMessage; + + // Simulate a provider that ends the message without emitting text_end. + handler?.({ type: "message_end", message: assistantMessage }); + + expect(onBlockReply).toHaveBeenCalledTimes(1); + expect(subscription.assistantTexts).toEqual(["Hello block"]); + + // Some providers can still emit a late text_end; this must not re-emit. + handler?.({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { + type: "text_end", + content: "Hello block", + }, + }); + + expect(onBlockReply).toHaveBeenCalledTimes(1); + expect(subscription.assistantTexts).toEqual(["Hello block"]); + }); + + it("clears block reply state on message_start", () => { + let handler: ((evt: unknown) => void) | undefined; + const session: StubSession = { + subscribe: (fn) => { + handler = fn; + return () => {}; + }, + }; + + const onBlockReply = vi.fn(); + + subscribeEmbeddedPiSession({ + session: session as unknown as Parameters< + typeof subscribeEmbeddedPiSession + >[0]["session"], + runId: "run", + onBlockReply, + blockReplyBreak: "text_end", + }); + + handler?.({ type: "message_start", message: { role: "assistant" } }); + handler?.({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { type: "text_delta", delta: "OK" }, + }); + handler?.({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { type: "text_end" }, + }); + expect(onBlockReply).toHaveBeenCalledTimes(1); + + // New assistant message with identical output should still emit. + handler?.({ type: "message_start", message: { role: "assistant" } }); + handler?.({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { type: "text_delta", delta: "OK" }, + }); + handler?.({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { type: "text_end" }, + }); + expect(onBlockReply).toHaveBeenCalledTimes(2); + }); + it("does not emit duplicate block replies when text_end repeats", () => { let handler: ((evt: unknown) => void) | undefined; const session: StubSession = { diff --git a/src/agents/pi-embedded-subscribe.ts b/src/agents/pi-embedded-subscribe.ts index 7997885e6..2d32f4fe3 100644 --- a/src/agents/pi-embedded-subscribe.ts +++ b/src/agents/pi-embedded-subscribe.ts @@ -317,6 +317,20 @@ export function subscribeEmbeddedPiSession(params: { const unsubscribe = params.session.subscribe( (evt: AgentEvent | { type: string; [k: string]: unknown }) => { + if (evt.type === "message_start") { + const msg = (evt as AgentEvent & { message: AgentMessage }).message; + if (msg?.role === "assistant") { + // Start-of-message is a safer reset point than message_end: some providers + // may deliver late text_end updates after message_end, which would + // otherwise re-trigger block replies. + deltaBuffer = ""; + blockBuffer = ""; + lastStreamedAssistant = undefined; + lastBlockReplyText = undefined; + assistantTextBaseline = assistantTexts.length; + } + } + if (evt.type === "tool_execution_start") { const toolName = String( (evt as AgentEvent & { toolName: string }).toolName, @@ -585,7 +599,6 @@ export function subscribeEmbeddedPiSession(params: { deltaBuffer = ""; blockBuffer = ""; lastStreamedAssistant = undefined; - lastBlockReplyText = undefined; } }