Skip to content

Commit e4556d6

Browse files
cursoragentcursoragent
committed
fix(msteams): suppress SDK final create for long streamed replies
Repro with real HttpStream shows Teams double-post comes from preview stream CREATE plus addStreamFinal CREATE (not streamFailed block fallback). For replies at/above 4000 streamed chars, clearText()+close() skips the second CREATE while the preview card already holds the full text.
1 parent 8b456d3 commit e4556d6

4 files changed

Lines changed: 245 additions & 0 deletions

File tree

‎extensions/msteams/src/reply-stream-controller.test.ts‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ function makeStream() {
88
return {
99
emit: vi.fn(),
1010
update: vi.fn(),
11+
clearText: vi.fn(),
1112
close: vi.fn<() => Promise<StreamCloseResult>>(async () => ({ id: "stream-final" })),
1213
canceled: false,
1314
};
@@ -191,6 +192,29 @@ describe("createTeamsReplyStreamController", () => {
191192
expect(stream.close).toHaveBeenCalled();
192193
});
193194

195+
it("skips streaminfo final create for long streamed replies at or above 4000 chars", async () => {
196+
const stream = makeStream();
197+
const ctrl = makeController({ stream });
198+
const longText = "x".repeat(4000);
199+
ctrl.onPartialReply({ text: longText });
200+
expect(ctrl.preparePayload({ text: longText })).toBeUndefined();
201+
await expect(ctrl.finalize()).resolves.toBeUndefined();
202+
expect(stream.clearText).toHaveBeenCalled();
203+
expect(stream.emit).toHaveBeenCalledTimes(1);
204+
expect(stream.close).toHaveBeenCalled();
205+
});
206+
207+
it("still emits streaminfo final for short streamed replies below 4000 chars", async () => {
208+
const stream = makeStream();
209+
const ctrl = makeController({ stream });
210+
ctrl.onPartialReply({ text: "short reply" });
211+
expect(ctrl.preparePayload({ text: "short reply" })).toBeUndefined();
212+
await expect(ctrl.finalize()).resolves.toBeUndefined();
213+
expect(stream.clearText).not.toHaveBeenCalled();
214+
expect(stream.emit).toHaveBeenCalledTimes(2);
215+
expect(stream.close).toHaveBeenCalled();
216+
});
217+
194218
it("returns suppressed final payload when stream close produces no final activity", async () => {
195219
const stream = makeStream();
196220
stream.close.mockResolvedValueOnce(undefined);

‎extensions/msteams/src/reply-stream-controller.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,11 @@ function isStreamCancelledError(err: unknown): boolean {
4444
return err instanceof Error && err.name === "StreamCancelledError";
4545
}
4646

47+
// Teams often fails to collapse the streamed preview card into the SDK's
48+
// addStreamFinal() message once the accumulated text crosses the usual 4000-char
49+
// Bot Framework chunk ceiling, leaving two identical full messages visible.
50+
const MSTEAMS_STREAM_FINAL_SUPPRESS_AT_CHARS = 4000;
51+
4752
/**
4853
* Bridges openclaw's reply pipeline callbacks to the SDK's `ctx.stream`.
4954
* Streaming is enabled for personal (DM) conversations only; group/channel
@@ -343,6 +348,17 @@ export function createTeamsReplyStreamController(params: {
343348
? { feedbackLoopEnabled: true }
344349
: {};
345350
try {
351+
// Long replies already show the full text on the preview card. The SDK's
352+
// close() always CREATEs a streaminfo-final message on top; when Teams
353+
// fails to collapse that pair, the user sees a duplicate. clearText() +
354+
// close() returns early without a second CREATE while the preview stands.
355+
if (emittedTextLength >= MSTEAMS_STREAM_FINAL_SUPPRESS_AT_CHARS) {
356+
stream.clearText();
357+
await stream.close();
358+
streamFinalizationPending = false;
359+
pendingFinalPayload = undefined;
360+
return undefined;
361+
}
346362
stream.emit({
347363
type: "message",
348364
entities: finalEntities,

‎extensions/msteams/src/reply-stream-longreply-repro.test.ts‎

Lines changed: 204 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,204 @@
1+
import type { ConversationReference } from "@microsoft/teams.api";
2+
import { HttpStream } from "@microsoft/teams.apps/dist/http/http-stream.js";
3+
// Repro for MS Teams long-reply double-post via real @microsoft/teams.apps HttpStream.
4+
// Only the Bot Framework conversations.activities create/update API is mocked.
5+
import { describe, expect, it } from "vitest";
6+
import { createTeamsReplyStreamController } from "./reply-stream-controller.js";
7+
8+
type SentActivity = Record<string, unknown> & {
9+
id?: string;
10+
type?: string;
11+
text?: string;
12+
entities?: Array<{ type?: string; streamType?: string }>;
13+
channelData?: { streamType?: string; streamId?: string };
14+
};
15+
16+
type ActivityCall = {
17+
kind: "create" | "update";
18+
activity: SentActivity;
19+
};
20+
21+
function makeMockClient() {
22+
const calls: ActivityCall[] = [];
23+
let nextId = 1;
24+
25+
const client = {
26+
conversations: {
27+
activities: (_convId: string) => ({
28+
create: async (activity: SentActivity) => {
29+
const id = `msg-${nextId++}`;
30+
const sent = { ...activity, id };
31+
calls.push({ kind: "create", activity: sent });
32+
return { id };
33+
},
34+
update: async (id: string, activity: SentActivity) => {
35+
const sent = { ...activity, id };
36+
calls.push({ kind: "update", activity: sent });
37+
return { id };
38+
},
39+
}),
40+
},
41+
};
42+
43+
return { client, calls };
44+
}
45+
46+
function makeRef(): ConversationReference {
47+
return {
48+
bot: { id: "28:bot", name: "OpenClaw" },
49+
conversation: { id: "19:[email protected]" },
50+
} as ConversationReference;
51+
}
52+
53+
async function settleHttpStream(stream: HttpStream): Promise<void> {
54+
// HttpStream.flush() is async and close() polls until the queue drains.
55+
for (let i = 0; i < 50; i += 1) {
56+
await stream.flush();
57+
await new Promise((resolve) => setTimeout(resolve, 0));
58+
}
59+
}
60+
61+
function hasStreamInfoEntity(activity: SentActivity): boolean {
62+
return activity.entities?.some((entity) => entity.type === "streaminfo") ?? false;
63+
}
64+
65+
function streamInfoType(activity: SentActivity): string | undefined {
66+
return activity.entities?.find((entity) => entity.type === "streaminfo")?.streamType;
67+
}
68+
69+
function summarizeCalls(calls: ActivityCall[]) {
70+
const creates = calls.filter((call) => call.kind === "create");
71+
const updates = calls.filter((call) => call.kind === "update");
72+
const typingCreates = creates.filter((call) => call.activity.type === "typing");
73+
const messageCreates = creates.filter((call) => call.activity.type === "message");
74+
const finalCreates = messageCreates.filter((call) => streamInfoType(call.activity) === "final");
75+
const streamingCreates = [
76+
...typingCreates.filter((call) => hasStreamInfoEntity(call.activity)),
77+
...messageCreates.filter((call) => streamInfoType(call.activity) === "streaming"),
78+
];
79+
return {
80+
createCount: creates.length,
81+
updateCount: updates.length,
82+
typingCreateCount: typingCreates.length,
83+
messageCreateCount: messageCreates.length,
84+
finalCreateCount: finalCreates.length,
85+
streamingCreateCount: streamingCreates.length,
86+
finalCreates,
87+
lastStreamingText: streamingCreates.at(-1)?.activity.text,
88+
finalText: finalCreates.at(-1)?.activity.text,
89+
};
90+
}
91+
92+
/** Drive openclaw's cumulative partial-reply pattern through the real controller + HttpStream. */
93+
async function driveOpenClawLongReply(params: {
94+
fullText: string;
95+
chunkChars: number;
96+
feedbackLoopEnabled?: boolean;
97+
}) {
98+
const { client, calls } = makeMockClient();
99+
const httpStream = new HttpStream(client as never, makeRef());
100+
const ctrl = createTeamsReplyStreamController({
101+
conversationType: "personal",
102+
context: { stream: httpStream } as never,
103+
feedbackLoopEnabled: params.feedbackLoopEnabled ?? false,
104+
});
105+
106+
let blockDeliveries: Array<string | undefined> = [];
107+
for (let len = params.chunkChars; len < params.fullText.length; len += params.chunkChars) {
108+
ctrl.onPartialReply({ text: params.fullText.slice(0, len) });
109+
await settleHttpStream(httpStream);
110+
}
111+
ctrl.onPartialReply({ text: params.fullText });
112+
await settleHttpStream(httpStream);
113+
114+
const prepared = ctrl.preparePayload({ text: params.fullText });
115+
if (prepared?.text !== undefined) {
116+
blockDeliveries.push(prepared.text);
117+
} else if (prepared === undefined) {
118+
blockDeliveries.push(undefined);
119+
} else {
120+
blockDeliveries.push(prepared.text);
121+
}
122+
123+
await ctrl.finalize();
124+
await settleHttpStream(httpStream);
125+
126+
return { calls, blockDeliveries, summary: summarizeCalls(calls) };
127+
}
128+
129+
describe("MS Teams HttpStream long-reply repro (real SDK, mocked transport)", () => {
130+
it("long single-shot reply (>4000 chars): preview stream create + final create both carry full text", async () => {
131+
const fullText = `Long reply ${"word ".repeat(900)}`.slice(0, 4500);
132+
expect(fullText.length).toBeGreaterThan(4000);
133+
134+
const { summary, blockDeliveries } = await driveOpenClawLongReply({
135+
fullText,
136+
chunkChars: 400,
137+
});
138+
139+
// openclaw suppresses block delivery on the happy path — duplicate is not from (c).
140+
expect(blockDeliveries).toEqual([undefined]);
141+
142+
// SDK ships streaming preview chunks as CREATEs (with streaminfo), not UPDATEs.
143+
expect(summary.updateCount).toBe(0);
144+
expect(summary.streamingCreateCount).toBeGreaterThan(1);
145+
146+
// Without the plugin fix, the SDK close() adds a second full-text CREATE that Teams
147+
// may fail to collapse into the preview card for long replies.
148+
expect(summary.finalCreateCount).toBe(0);
149+
150+
expect(summary.lastStreamingText).toBe(fullText);
151+
expect(summary.finalText).toBeUndefined();
152+
});
153+
154+
it("short reply uses the same SDK create+final pattern but with fewer preview chunks", async () => {
155+
const fullText = "Short hello from OpenClaw.";
156+
const { summary, blockDeliveries } = await driveOpenClawLongReply({
157+
fullText,
158+
chunkChars: 8,
159+
});
160+
161+
expect(blockDeliveries).toEqual([undefined]);
162+
expect(summary.updateCount).toBe(0);
163+
expect(summary.finalCreateCount).toBe(1);
164+
expect(summary.finalText).toBe(fullText);
165+
expect(summary.lastStreamingText).toBe(fullText);
166+
expect(summary.streamingCreateCount).toBeGreaterThanOrEqual(1);
167+
});
168+
169+
it("documents unfixed SDK behavior: raw HttpStream close() CREATEs a streaminfo final with full text", async () => {
170+
const fullText = "x".repeat(4200);
171+
const { client, calls } = makeMockClient();
172+
const stream = new HttpStream(client as never, makeRef());
173+
174+
// HttpStream appends each emit(); drive deltas like openclaw's controller does.
175+
let emitted = 0;
176+
for (let len = 500; len < fullText.length; len += 500) {
177+
stream.emit(fullText.slice(emitted, len));
178+
emitted = len;
179+
await settleHttpStream(stream);
180+
}
181+
stream.emit(fullText.slice(emitted));
182+
await settleHttpStream(stream);
183+
await stream.close();
184+
await settleHttpStream(stream);
185+
186+
const summary = summarizeCalls(calls);
187+
expect(summary.updateCount).toBe(0);
188+
expect(summary.finalCreateCount).toBe(1);
189+
expect(summary.finalText).toBe(fullText);
190+
expect(summary.lastStreamingText).toBe(fullText);
191+
});
192+
193+
it("openclaw long-reply fix suppresses the SDK final CREATE while short replies still finalize", async () => {
194+
const longText = "y".repeat(4100);
195+
const longResult = await driveOpenClawLongReply({ fullText: longText, chunkChars: 300 });
196+
expect(longResult.summary.finalCreateCount).toBe(0);
197+
expect(longResult.summary.lastStreamingText).toBe(longText);
198+
199+
const shortText = "Brief answer.";
200+
const shortResult = await driveOpenClawLongReply({ fullText: shortText, chunkChars: 4 });
201+
expect(shortResult.summary.finalCreateCount).toBe(1);
202+
expect(shortResult.summary.finalText).toBe(shortText);
203+
});
204+
});

‎extensions/msteams/src/sdk-types.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ export type MSTeamsActivityLike = MSTeamsActivityParams | string;
5656
export type MSTeamsStreamer = {
5757
emit(activity: MSTeamsActivityParams | string): void;
5858
update(text: string): void;
59+
clearText(): void;
5960
close(): Promise<unknown>;
6061
readonly canceled: boolean;
6162
};

0 commit comments

Comments
 (0)