mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-27 21:07:01 -06:00
joshavant
1015 lines
37 KiB
TypeScript
1015 lines
37 KiB
TypeScript
// Msteams tests cover reply stream controller plugin behavior.
|
|
import { describe, expect, it, vi } from "vitest";
|
|
import { createTeamsReplyStreamController } from "./reply-stream-controller.js";
|
|
|
|
type StreamCloseResult = { id: string } | undefined;
|
|
|
|
function makeStream() {
|
|
return {
|
|
emit: vi.fn(),
|
|
update: vi.fn(),
|
|
clearText: vi.fn(),
|
|
close: vi.fn<() => Promise<StreamCloseResult>>(async () => ({ id: "stream-final" })),
|
|
canceled: false,
|
|
};
|
|
}
|
|
|
|
function makeAcknowledgedStream() {
|
|
type ChunkActivity = {
|
|
id?: string;
|
|
type?: string;
|
|
text?: string;
|
|
channelData?: { streamType?: string };
|
|
};
|
|
const handlers = new Map<number, (activity: ChunkActivity) => void>();
|
|
let nextSubscriptionId = 0;
|
|
const stream = {
|
|
...makeStream(),
|
|
events: {
|
|
on: vi.fn((_event: "chunk", handler: (activity: ChunkActivity) => void) => {
|
|
const subscriptionId = nextSubscriptionId++;
|
|
handlers.set(subscriptionId, handler);
|
|
return subscriptionId;
|
|
}),
|
|
off: vi.fn((subscriptionId: number) => {
|
|
handlers.delete(subscriptionId);
|
|
}),
|
|
},
|
|
acknowledge(text: string, overrides: Partial<ChunkActivity> = {}) {
|
|
const activity: ChunkActivity = {
|
|
id: "stream-acknowledged",
|
|
type: "typing",
|
|
text,
|
|
channelData: { streamType: "streaming" },
|
|
...overrides,
|
|
};
|
|
for (const handler of handlers.values()) {
|
|
handler(activity);
|
|
}
|
|
},
|
|
};
|
|
return stream;
|
|
}
|
|
|
|
function makeContext(stream?: ReturnType<typeof makeStream>) {
|
|
return { activity: { type: "message" }, stream } as never;
|
|
}
|
|
|
|
function makeController(
|
|
opts: {
|
|
allowProviderPreview?: boolean;
|
|
conversationType?: string;
|
|
stream?: ReturnType<typeof makeStream>;
|
|
} = {},
|
|
) {
|
|
const stream = opts.stream;
|
|
return createTeamsReplyStreamController({
|
|
allowProviderPreview: opts.allowProviderPreview ?? true,
|
|
conversationType: opts.conversationType ?? "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
});
|
|
}
|
|
|
|
describe("createTeamsReplyStreamController", () => {
|
|
it("emits chunks via stream.emit when tokens arrive", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
expect(stream.emit).toHaveBeenCalledWith("hello");
|
|
});
|
|
|
|
it("falls back to normal delivery when provider previews are disabled", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ allowProviderPreview: false, stream });
|
|
|
|
ctrl.onPartialReply({ text: "original partial" });
|
|
|
|
expect(ctrl.hasStream()).toBe(false);
|
|
expect(stream.emit).not.toHaveBeenCalled();
|
|
expect(ctrl.preparePayload({ text: "authoritative final" })).toEqual({
|
|
text: "authoritative final",
|
|
});
|
|
});
|
|
|
|
it("emits only the delta when openclaw sends cumulative text on each chunk", () => {
|
|
// openclaw's reply pipeline calls onPartialReply with the cumulative
|
|
// text-so-far on every chunk. The SDK's HttpStream APPENDS each emit() to
|
|
// its internal text buffer (this.text += activity.text). Without delta
|
|
// conversion, the SDK accumulates "chunk1 + chunk2 + chunk3" and the user
|
|
// sees the message duplicated on each progress update (real bug observed
|
|
// 2026-05-06: a sonnet rendered with each line repeated alongside the
|
|
// previous full state).
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "Here's one for you:\nThe morning" });
|
|
ctrl.onPartialReply({ text: "Here's one for you:\nThe morning light" });
|
|
ctrl.onPartialReply({ text: "Here's one for you:\nThe morning light breaks" });
|
|
expect(stream.emit).toHaveBeenNthCalledWith(1, "Here's one for you:\nThe morning");
|
|
expect(stream.emit).toHaveBeenNthCalledWith(2, " light");
|
|
expect(stream.emit).toHaveBeenNthCalledWith(3, " breaks");
|
|
});
|
|
|
|
it("keeps the next chunk after cumulative trailing whitespace is normalized", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "Intro\n\n " });
|
|
ctrl.onPartialReply({ text: "Intro\n\nNext" });
|
|
|
|
expect(stream.emit).toHaveBeenNthCalledWith(1, "Intro\n\n ");
|
|
expect(stream.emit).toHaveBeenNthCalledWith(2, "Next");
|
|
});
|
|
|
|
it("replaces non-whitespace rewrites through the native stream", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
ctrl.onPartialReply({ text: "abXYZ" });
|
|
|
|
expect(stream.emit).toHaveBeenCalledTimes(1);
|
|
expect(stream.emit).toHaveBeenCalledWith("abcde");
|
|
expect(ctrl.preparePayload({ text: "abXYZ" })).toBeUndefined();
|
|
expect(stream.emit).toHaveBeenCalledTimes(2);
|
|
expect(stream.emit).toHaveBeenLastCalledWith(
|
|
expect.objectContaining({ type: "message", text: "abXYZ" }),
|
|
);
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-final",
|
|
content: "abXYZ",
|
|
logicalContent: "abXYZ",
|
|
});
|
|
expect(stream.clearText).toHaveBeenCalledTimes(1);
|
|
expect(stream.close).toHaveBeenCalled();
|
|
});
|
|
|
|
it("retains an acknowledged replacement when stream close produces no activity", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
stream.acknowledge("abcde");
|
|
ctrl.onPartialReply({ text: "provider replacement" });
|
|
expect(
|
|
ctrl.preparePayload({
|
|
text: "provider replacement",
|
|
mediaUrl: "https://example.test/replacement.png",
|
|
}),
|
|
).toBeUndefined();
|
|
stream.acknowledge("provider replacement");
|
|
stream.close.mockResolvedValueOnce(undefined);
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-acknowledged",
|
|
content: "provider replacement",
|
|
logicalContent: "provider replacement",
|
|
postNativePayloads: [
|
|
{
|
|
text: undefined,
|
|
mediaUrl: "https://example.test/replacement.png",
|
|
},
|
|
],
|
|
});
|
|
});
|
|
|
|
it("falls back to the full replacement when close fails before acknowledgement", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
stream.acknowledge("abcde");
|
|
ctrl.onPartialReply({ text: "provider replacement" });
|
|
expect(ctrl.preparePayload({ mediaUrl: "https://example.test/before.png" })).toBeUndefined();
|
|
expect(
|
|
ctrl.preparePayload({
|
|
text: "provider replacement",
|
|
mediaUrl: "https://example.test/replacement.png",
|
|
}),
|
|
).toBeUndefined();
|
|
stream.close.mockRejectedValueOnce(new Error("close failed"));
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-acknowledged",
|
|
content: "abcde",
|
|
logicalContent: "provider replacement",
|
|
postNativePayloads: [
|
|
{ mediaUrl: "https://example.test/before.png" },
|
|
{
|
|
text: "provider replacement",
|
|
mediaUrl: "https://example.test/replacement.png",
|
|
},
|
|
],
|
|
});
|
|
});
|
|
|
|
it("ignores a delayed old chunk that is only a prefix of the replacement", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
stream.acknowledge("abcde");
|
|
ctrl.onPartialReply({ text: "ab replacement" });
|
|
expect(ctrl.preparePayload({ text: "ab replacement" })).toBeUndefined();
|
|
stream.acknowledge("ab");
|
|
stream.close.mockRejectedValueOnce(new Error("close failed"));
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-acknowledged",
|
|
content: "abcde",
|
|
logicalContent: "ab replacement",
|
|
postNativePayloads: [{ text: "ab replacement" }],
|
|
});
|
|
});
|
|
|
|
it("rejects a delayed common-prefix chunk while awaiting replacement acknowledgement", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcdef" });
|
|
ctrl.onPartialReply({ text: "abcXYZ" });
|
|
expect(ctrl.preparePayload({ text: "abcXYZ" })).toBeUndefined();
|
|
stream.acknowledge("abc");
|
|
stream.close.mockRejectedValueOnce(new Error("close failed"));
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: false,
|
|
logicalContent: "abcXYZ",
|
|
postNativePayloads: [{ text: "abcXYZ" }],
|
|
});
|
|
});
|
|
|
|
it("uses the latest partial when no final text payload follows a rewrite", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
stream.acknowledge("abcde");
|
|
ctrl.onPartialReply({ text: "abXYZ" });
|
|
ctrl.onPartialReply({ text: "abcdef" });
|
|
expect(ctrl.preparePayload({ mediaUrl: "https://example.test/final.png" })).toBeUndefined();
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-final",
|
|
content: "abcdef",
|
|
logicalContent: "abcdef",
|
|
postNativePayloads: [{ mediaUrl: "https://example.test/final.png" }],
|
|
});
|
|
expect(stream.emit).toHaveBeenLastCalledWith(
|
|
expect.objectContaining({ type: "message", text: "abcdef" }),
|
|
);
|
|
});
|
|
|
|
it("suppresses a replacement when emit synchronously discovers Stop", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
stream.acknowledge("abcde");
|
|
ctrl.onPartialReply({ text: "provider replacement" });
|
|
stream.emit.mockImplementation(() => {
|
|
const error = new Error("stream canceled");
|
|
error.name = "StreamCancelledError";
|
|
throw error;
|
|
});
|
|
|
|
expect(ctrl.preparePayload({ text: "provider replacement" })).toBeUndefined();
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-acknowledged",
|
|
content: "abcde",
|
|
});
|
|
expect(stream.close).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("holds payloads around replacement text until native settlement", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
stream.acknowledge("abcde");
|
|
ctrl.onPartialReply({ text: "provider replacement" });
|
|
|
|
expect(ctrl.preparePayload({ mediaUrl: "https://example.test/before.png" })).toBeUndefined();
|
|
expect(ctrl.preparePayload({ text: "provider replacement" })).toBeUndefined();
|
|
expect(
|
|
ctrl.preparePayload({
|
|
text: "second payload",
|
|
mediaUrl: "https://example.test/after.png",
|
|
}),
|
|
).toBeUndefined();
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-final",
|
|
content: "provider replacement",
|
|
logicalContent: "provider replacement\nsecond payload",
|
|
postNativePayloads: [
|
|
{ mediaUrl: "https://example.test/before.png" },
|
|
{ text: "second payload", mediaUrl: "https://example.test/after.png" },
|
|
],
|
|
});
|
|
});
|
|
|
|
it("preserves held payload order after replacement emit fails", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "abcde" });
|
|
stream.acknowledge("abcde");
|
|
ctrl.onPartialReply({ text: "provider replacement" });
|
|
expect(ctrl.preparePayload({ mediaUrl: "https://example.test/before.png" })).toBeUndefined();
|
|
stream.emit.mockImplementationOnce(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
expect(ctrl.preparePayload({ text: "provider replacement" })).toBeUndefined();
|
|
expect(ctrl.preparePayload({ text: "later payload" })).toBeUndefined();
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-final",
|
|
content: "abcde",
|
|
logicalContent: "provider replacement\nlater payload",
|
|
postNativePayloads: [
|
|
{ mediaUrl: "https://example.test/before.png" },
|
|
{ text: "provider replacement" },
|
|
{ text: "later payload" },
|
|
],
|
|
});
|
|
});
|
|
|
|
it("ignores duplicate or out-of-order partial replies that don't extend the text", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "abcdef" });
|
|
ctrl.onPartialReply({ text: "abc" }); // shorter — could be edit-in-place semantics
|
|
ctrl.onPartialReply({ text: "abcdef" }); // back to known length
|
|
expect(stream.emit).toHaveBeenCalledTimes(1);
|
|
expect(stream.emit).toHaveBeenCalledWith("abcdef");
|
|
});
|
|
|
|
it("does not touch native stream on reply start before text or progress work", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
await ctrl.onReplyStart();
|
|
await ctrl.onReplyStart();
|
|
|
|
expect(stream.update).not.toHaveBeenCalled();
|
|
expect(stream.emit).not.toHaveBeenCalled();
|
|
expect(ctrl.preparePayload({ text: "tool-only response" })).toEqual({
|
|
text: "tool-only response",
|
|
});
|
|
await ctrl.finalize();
|
|
expect(stream.close).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("suppresses block delivery when text was streamed", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "streamed" });
|
|
expect(ctrl.preparePayload({ text: "streamed" })).toBeUndefined();
|
|
});
|
|
|
|
it("strips text but keeps media when text was streamed and payload has media", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "streamed" });
|
|
expect(ctrl.preparePayload({ text: "streamed", mediaUrl: "https://x/y.png" })).toEqual({
|
|
text: undefined,
|
|
mediaUrl: "https://x/y.png",
|
|
});
|
|
});
|
|
|
|
it("allows fallback delivery for second text segment after tool calls", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "First segment" });
|
|
expect(ctrl.preparePayload({ text: "First segment" })).toBeUndefined();
|
|
|
|
const result = ctrl.preparePayload({ text: "Second segment after tools" });
|
|
expect(result).toEqual({ text: "Second segment after tools" });
|
|
});
|
|
|
|
it("uses fallback even when onPartialReply fires after stream finalization is pending", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "First segment" });
|
|
expect(ctrl.preparePayload({ text: "First segment" })).toBeUndefined();
|
|
|
|
ctrl.onPartialReply({ text: "Second segment" });
|
|
expect(stream.emit).toHaveBeenCalledTimes(1);
|
|
expect(ctrl.preparePayload({ text: "Second segment" })).toEqual({ text: "Second segment" });
|
|
});
|
|
|
|
it("delivers all later segments across 3+ tool call rounds", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "Segment 1" });
|
|
expect(ctrl.preparePayload({ text: "Segment 1" })).toBeUndefined();
|
|
|
|
ctrl.onPartialReply({ text: "Segment 2" });
|
|
expect(ctrl.preparePayload({ text: "Segment 2" })).toEqual({ text: "Segment 2" });
|
|
|
|
ctrl.onPartialReply({ text: "Segment 3" });
|
|
expect(ctrl.preparePayload({ text: "Segment 3" })).toEqual({ text: "Segment 3" });
|
|
});
|
|
|
|
it("passes media+text payload through fully after stream finalization is pending", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "Streamed text" });
|
|
expect(ctrl.preparePayload({ text: "Streamed text" })).toBeUndefined();
|
|
|
|
expect(
|
|
ctrl.preparePayload({
|
|
text: "Post-tool text with image",
|
|
mediaUrl: "https://example.com/tool-output.png",
|
|
}),
|
|
).toEqual({
|
|
text: "Post-tool text with image",
|
|
mediaUrl: "https://example.com/tool-output.png",
|
|
});
|
|
});
|
|
|
|
it("drops the payload after the stream is canceled (e.g. user Stop)", () => {
|
|
// After the user presses Stop in Teams, the streamed prefix is already
|
|
// visible. Returning the full payload here would render as a SECOND
|
|
// message containing everything — defeating the cancel intent.
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "partial" });
|
|
stream.canceled = true;
|
|
expect(ctrl.preparePayload({ text: "partial complete" })).toBeUndefined();
|
|
});
|
|
|
|
it("drops the payload even when it carries media after cancel", () => {
|
|
// Cancel honored consistently — no leftover media bubble lands either.
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "partial" });
|
|
stream.canceled = true;
|
|
expect(
|
|
ctrl.preparePayload({ text: "partial complete", mediaUrl: "https://x/y.png" }),
|
|
).toBeUndefined();
|
|
});
|
|
|
|
it("falls back to block delivery when no tokens were streamed", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
expect(ctrl.preparePayload({ text: "tool-only response" })).toEqual({
|
|
text: "tool-only response",
|
|
});
|
|
});
|
|
|
|
it("closes the stream in finalize after streamed text payload was suppressed", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "streamed" });
|
|
expect(ctrl.preparePayload({ text: "streamed" })).toBeUndefined();
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
messageId: "stream-final",
|
|
content: "streamed",
|
|
});
|
|
expect(stream.close).toHaveBeenCalled();
|
|
});
|
|
|
|
it("returns suppressed final payload when stream close produces no final activity", async () => {
|
|
const stream = makeStream();
|
|
stream.close.mockResolvedValueOnce(undefined);
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "streamed" });
|
|
expect(ctrl.preparePayload({ text: "streamed final" })).toBeUndefined();
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: false,
|
|
fallbackPayload: { text: "streamed final" },
|
|
});
|
|
});
|
|
|
|
it("returns text-only fallback when stream close no-ops after media already queued", async () => {
|
|
const stream = makeStream();
|
|
stream.close.mockResolvedValueOnce(undefined);
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "streamed" });
|
|
expect(ctrl.preparePayload({ text: "streamed final", mediaUrl: "https://x/y.png" })).toEqual({
|
|
text: undefined,
|
|
mediaUrl: "https://x/y.png",
|
|
});
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: false,
|
|
fallbackPayload: {
|
|
text: "streamed final",
|
|
mediaUrl: undefined,
|
|
mediaUrls: undefined,
|
|
},
|
|
});
|
|
});
|
|
|
|
it("returns suppressed final payload when stream close throws", async () => {
|
|
const stream = makeStream();
|
|
stream.close.mockRejectedValueOnce(new Error("close failed"));
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "streamed" });
|
|
expect(ctrl.preparePayload({ text: "streamed final" })).toBeUndefined();
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: false,
|
|
fallbackPayload: { text: "streamed final" },
|
|
});
|
|
});
|
|
|
|
it("does not close the stream in finalize when no tokens were emitted", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
await ctrl.finalize();
|
|
expect(stream.close).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("streams compact Teams progress lines when tool progress is enabled", async () => {
|
|
vi.useFakeTimers();
|
|
const stream = makeStream();
|
|
try {
|
|
const ctrl = createTeamsReplyStreamController({
|
|
allowProviderPreview: true,
|
|
conversationType: "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
log: { debug: vi.fn() } as never,
|
|
msteamsConfig: {
|
|
streaming: {
|
|
mode: "progress",
|
|
progress: {
|
|
label: "Working",
|
|
maxLines: 3,
|
|
},
|
|
},
|
|
} as never,
|
|
});
|
|
|
|
await ctrl.pushProgressLine("tool: search");
|
|
await ctrl.pushProgressLine("tool: exec");
|
|
expect(stream.update).not.toHaveBeenCalled();
|
|
|
|
await vi.advanceTimersByTimeAsync(5_000);
|
|
|
|
expect(stream.update).toHaveBeenLastCalledWith("Working\n\n- tool: search\n- tool: exec");
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
});
|
|
|
|
it("replaces Teams plan snapshots and keeps the explanation", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = createTeamsReplyStreamController({
|
|
allowProviderPreview: true,
|
|
conversationType: "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
msteamsConfig: {
|
|
streaming: { mode: "progress", progress: { label: false } },
|
|
} as never,
|
|
});
|
|
|
|
await ctrl.pushPlanProgress([{ step: "Inspect", status: "in_progress" }], {
|
|
explanation: "Initial plan",
|
|
});
|
|
await ctrl.pushPlanProgress(
|
|
[
|
|
{ step: "Inspect", status: "completed" },
|
|
{ step: "Patch", status: "in_progress" },
|
|
],
|
|
{ explanation: "Revised plan" },
|
|
);
|
|
|
|
expect(stream.update).toHaveBeenLastCalledWith("Revised plan\n\n✅ Inspect\n▸ Patch");
|
|
});
|
|
|
|
it("cancels the pending progress gate at finalize so no stale card posts after close", async () => {
|
|
vi.useFakeTimers();
|
|
const stream = makeStream();
|
|
try {
|
|
const ctrl = createTeamsReplyStreamController({
|
|
allowProviderPreview: true,
|
|
conversationType: "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
log: { debug: vi.fn() } as never,
|
|
msteamsConfig: {
|
|
streaming: { mode: "progress", progress: { label: "Working" } },
|
|
} as never,
|
|
});
|
|
|
|
// One work event schedules the delayed start; the turn finishes first.
|
|
await ctrl.pushProgressLine("tool: search");
|
|
ctrl.preparePayload({ text: "done" });
|
|
await ctrl.finalize();
|
|
expect(stream.update).not.toHaveBeenCalled();
|
|
|
|
// The gate timer must be dead: firing it against the closed stream
|
|
// would post a fresh stale "working" card below the final answer.
|
|
await vi.advanceTimersByTimeAsync(5_000);
|
|
expect(stream.update).not.toHaveBeenCalled();
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
});
|
|
|
|
it("suppresses block delivery when progress final text is emitted to the stream", () => {
|
|
const stream = makeStream();
|
|
const ctrl = createTeamsReplyStreamController({
|
|
allowProviderPreview: true,
|
|
conversationType: "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
msteamsConfig: { streaming: { mode: "progress" } } as never,
|
|
});
|
|
|
|
expect(ctrl.preparePayload({ text: "complete final answer" })).toBeUndefined();
|
|
expect(stream.emit).toHaveBeenCalledWith("complete final answer");
|
|
});
|
|
|
|
it("ignores plan updates after final answer streaming starts", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = createTeamsReplyStreamController({
|
|
allowProviderPreview: true,
|
|
conversationType: "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
msteamsConfig: { streaming: { mode: "progress" } } as never,
|
|
});
|
|
|
|
expect(ctrl.preparePayload({ text: "complete final answer" })).toBeUndefined();
|
|
await ctrl.pushPlanProgress([{ step: "Late plan", status: "in_progress" }]);
|
|
|
|
expect(stream.update).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("falls back to normal delivery when progress final streaming fails", () => {
|
|
const stream = makeStream();
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("progress final failed");
|
|
});
|
|
const ctrl = createTeamsReplyStreamController({
|
|
allowProviderPreview: true,
|
|
conversationType: "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
log: { debug: vi.fn() } as never,
|
|
msteamsConfig: { streaming: { mode: "progress" } } as never,
|
|
});
|
|
|
|
expect(ctrl.preparePayload({ text: "complete final answer" })).toEqual({
|
|
text: "complete final answer",
|
|
});
|
|
});
|
|
|
|
it("does not close a canceled stream in finalize", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "partial" });
|
|
stream.canceled = true;
|
|
await ctrl.finalize();
|
|
expect(stream.close).not.toHaveBeenCalled();
|
|
});
|
|
|
|
describe("StreamCancelledError handling", () => {
|
|
function makeCancelError(): Error {
|
|
const err = new Error("stream canceled");
|
|
err.name = "StreamCancelledError";
|
|
return err;
|
|
}
|
|
|
|
it("swallows StreamCancelledError thrown from stream.emit (Stop button race)", () => {
|
|
const stream = makeStream();
|
|
stream.emit.mockImplementation(() => {
|
|
throw makeCancelError();
|
|
});
|
|
const ctrl = makeController({ stream });
|
|
// Must not throw — the SDK throws this synchronously when _canceled
|
|
// flipped between our pre-check and the emit call (or when no pre-check
|
|
// happens at all). An uncaught throw here crashes the gateway process
|
|
// since it surfaces as an unhandled promise rejection in async paths.
|
|
expect(() => ctrl.onPartialReply({ text: "after stop" })).not.toThrow();
|
|
});
|
|
|
|
it("swallows StreamCancelledError thrown from progress stream.update", async () => {
|
|
const stream = makeStream();
|
|
stream.update.mockImplementation(() => {
|
|
throw makeCancelError();
|
|
});
|
|
const ctrl = createTeamsReplyStreamController({
|
|
allowProviderPreview: true,
|
|
conversationType: "personal",
|
|
context: makeContext(stream),
|
|
feedbackLoopEnabled: false,
|
|
msteamsConfig: { streaming: { mode: "progress" } } as never,
|
|
});
|
|
await expect(ctrl.noteProgressWork({ toolName: "exec" })).resolves.toBeUndefined();
|
|
});
|
|
|
|
it("swallows StreamCancelledError thrown from stream.emit during finalize", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "partial" });
|
|
expect(ctrl.preparePayload({ text: "partial" })).toBeUndefined();
|
|
// Cancel after we've started streaming, then make the final emit throw.
|
|
stream.emit.mockImplementation(() => {
|
|
throw makeCancelError();
|
|
});
|
|
// Must not throw — finalize's pre-check on stream.canceled may miss
|
|
// the cancellation that happens between check and emit.
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: false,
|
|
});
|
|
});
|
|
|
|
it("latches streamFailed (and does not throw) on non-cancel errors from stream.emit", () => {
|
|
const stream = makeStream();
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
const ctrl = makeController({ stream });
|
|
// Must not propagate — the rest of the reply pipeline needs to keep
|
|
// running so preparePayload can fall back to block delivery.
|
|
expect(() => ctrl.onPartialReply({ text: "boom" })).not.toThrow();
|
|
// Stream is no longer considered active once it has failed.
|
|
expect(ctrl.isStreamActive()).toBe(false);
|
|
});
|
|
|
|
it("falls back to block delivery when stream.emit fails after tokens were emitted", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
// First chunk succeeds — tokensEmitted goes true.
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
expect(stream.emit).toHaveBeenCalledTimes(1);
|
|
// Second chunk fails for a non-cancel reason.
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
ctrl.onPartialReply({ text: "hello world" });
|
|
// Without the streamFailed latch, preparePayload would suppress the
|
|
// payload because tokens were emitted; the user would see only "hello".
|
|
// With the latch, block delivery sends the full final reply.
|
|
const result = ctrl.preparePayload({ text: "hello world final" });
|
|
expect(result).toEqual(expect.objectContaining({ text: "hello world final" }));
|
|
});
|
|
|
|
it("redelivers only the prefix Teams actually acknowledged after a later stream failure", () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello");
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
ctrl.onPartialReply({ text: "hello world" });
|
|
|
|
expect(ctrl.preparePayload({ text: "hello world final" })).toEqual({
|
|
text: " world final",
|
|
});
|
|
});
|
|
|
|
it("preserves the full fallback when a failed stream has no provider acknowledgement", () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
ctrl.onPartialReply({ text: "hello world" });
|
|
|
|
expect(ctrl.preparePayload({ text: "hello world final" })).toEqual({
|
|
text: "hello world final",
|
|
});
|
|
});
|
|
|
|
it("does not trim an independent later payload using a previous stream acknowledgement", () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello");
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
ctrl.onPartialReply({ text: "hello world" });
|
|
|
|
expect(ctrl.preparePayload({ text: "hello world" })).toEqual({
|
|
text: " world",
|
|
});
|
|
expect(ctrl.preparePayload({ text: "hello again" })).toEqual({
|
|
text: "hello again",
|
|
});
|
|
expect(stream.events.off).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("ignores unrelated, informative, and out-of-order stream acknowledgements", () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello", { type: "message" });
|
|
stream.acknowledge("hello", { channelData: { streamType: "informative" } });
|
|
stream.acknowledge("unrelated");
|
|
stream.acknowledge("he");
|
|
stream.acknowledge("hello", { id: "different-stream" });
|
|
stream.acknowledge("h");
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
ctrl.onPartialReply({ text: "hello world" });
|
|
|
|
expect(ctrl.preparePayload({ text: "hello world" })).toEqual({
|
|
text: "llo world",
|
|
});
|
|
});
|
|
|
|
it("suppresses a failed fallback when Teams already acknowledged the entire text", () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello");
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
ctrl.onPartialReply({ text: "hello world" });
|
|
|
|
expect(ctrl.preparePayload({ text: "hello" })).toBeUndefined();
|
|
});
|
|
|
|
it("retains media when Teams already acknowledged all fallback text", () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello");
|
|
stream.emit.mockImplementation(() => {
|
|
throw new Error("network failure");
|
|
});
|
|
ctrl.onPartialReply({ text: "hello world" });
|
|
|
|
expect(
|
|
ctrl.preparePayload({ text: "hello", mediaUrl: "https://example.com/image.png" }),
|
|
).toEqual({
|
|
text: undefined,
|
|
mediaUrl: "https://example.com/image.png",
|
|
});
|
|
});
|
|
|
|
it("redelivers only unacknowledged text when stream close fails", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello");
|
|
expect(ctrl.preparePayload({ text: "hello world" })).toBeUndefined();
|
|
stream.close.mockRejectedValueOnce(new Error("close failed"));
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
content: "hello",
|
|
messageId: "stream-acknowledged",
|
|
fallbackPayload: { text: " world" },
|
|
});
|
|
expect(stream.events.off).toHaveBeenCalledWith(0);
|
|
});
|
|
|
|
it("does not redeliver an acknowledged final when stream close produces no activity", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello");
|
|
expect(ctrl.preparePayload({ text: "hello" })).toBeUndefined();
|
|
stream.close.mockResolvedValueOnce(undefined);
|
|
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
content: "hello",
|
|
messageId: "stream-acknowledged",
|
|
});
|
|
expect(stream.events.off).toHaveBeenCalledWith(0);
|
|
});
|
|
|
|
it("honors cancellation after an acknowledged stream prefix", async () => {
|
|
const stream = makeAcknowledgedStream();
|
|
const ctrl = makeController({ stream });
|
|
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
stream.acknowledge("hello");
|
|
stream.canceled = true;
|
|
|
|
expect(ctrl.preparePayload({ text: "hello world" })).toBeUndefined();
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: true,
|
|
content: "hello",
|
|
messageId: "stream-acknowledged",
|
|
});
|
|
expect(stream.events.off).toHaveBeenCalledWith(0);
|
|
});
|
|
|
|
it("preserves the no-duplicate behavior for the active streamed segment", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "hello" });
|
|
// No failure — preparePayload should still suppress block delivery for
|
|
// the active streamed segment so the streamed text isn't duplicated.
|
|
expect(ctrl.preparePayload({ text: "hello world" })).toBeUndefined();
|
|
});
|
|
|
|
it("swallows non-cancel errors from stream.close during finalize", async () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "partial" });
|
|
expect(ctrl.preparePayload({ text: "partial final" })).toBeUndefined();
|
|
stream.close.mockImplementation(async () => {
|
|
throw new Error("close failed");
|
|
});
|
|
// Finalize must not propagate; it returns the retained payload so the
|
|
// dispatcher can fall back to normal Teams delivery.
|
|
await expect(ctrl.finalize()).resolves.toEqual({
|
|
visibleReplySent: false,
|
|
fallbackPayload: { text: "partial final" },
|
|
});
|
|
});
|
|
|
|
it("treats post-cancel stream as inactive without further emit attempts", () => {
|
|
const stream = makeStream();
|
|
stream.emit.mockImplementationOnce(() => {
|
|
throw makeCancelError();
|
|
});
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "first chunk after stop" });
|
|
// Subsequent partial replies should short-circuit and not call emit
|
|
// again (the SDK would throw on every call once canceled).
|
|
ctrl.onPartialReply({ text: "second chunk" });
|
|
ctrl.onPartialReply({ text: "third chunk" });
|
|
expect(stream.emit).toHaveBeenCalledTimes(1);
|
|
expect(ctrl.isStreamActive()).toBe(false);
|
|
});
|
|
});
|
|
|
|
describe("non-personal conversation", () => {
|
|
it("does not stream in channels — onPartialReply is a no-op", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ conversationType: "channel", stream });
|
|
ctrl.onPartialReply({ text: "anything" });
|
|
expect(stream.emit).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("hasStream returns false for channels", () => {
|
|
const ctrl = makeController({ conversationType: "channel", stream: makeStream() });
|
|
expect(ctrl.hasStream()).toBe(false);
|
|
});
|
|
|
|
it("preparePayload returns payload unchanged for channels", () => {
|
|
const ctrl = makeController({ conversationType: "channel", stream: makeStream() });
|
|
expect(ctrl.preparePayload({ text: "hi" })).toEqual({ text: "hi" });
|
|
});
|
|
});
|
|
|
|
describe("isStreamActive", () => {
|
|
it("returns false before any tokens arrive", () => {
|
|
expect(makeController({ stream: makeStream() }).isStreamActive()).toBe(false);
|
|
});
|
|
|
|
it("returns true while receiving tokens", () => {
|
|
const ctrl = makeController({ stream: makeStream() });
|
|
ctrl.onPartialReply({ text: "tokens" });
|
|
expect(ctrl.isStreamActive()).toBe(true);
|
|
});
|
|
|
|
it("returns false when stream is canceled", () => {
|
|
const stream = makeStream();
|
|
const ctrl = makeController({ stream });
|
|
ctrl.onPartialReply({ text: "tokens" });
|
|
stream.canceled = true;
|
|
expect(ctrl.isStreamActive()).toBe(false);
|
|
});
|
|
|
|
it("returns false for non-personal conversations", () => {
|
|
const ctrl = makeController({ conversationType: "channel", stream: makeStream() });
|
|
ctrl.onPartialReply({ text: "tokens" });
|
|
expect(ctrl.isStreamActive()).toBe(false);
|
|
});
|
|
});
|
|
});
|