|
| 1 | +// MSTeams delivery trace goldens: replayable wire-level lifecycle recordings. |
| 2 | +// |
| 3 | +// Drives the real createMSTeamsReplyDispatcher + createTeamsReplyStreamController |
| 4 | +// wiring (typing keepalive, append-only native stream, block batching) against a |
| 5 | +// mocked Bot Framework turn context. OUT events are the raw SDK surface calls: |
| 6 | +// context.sendActivity activities and stream.emit/update/close writes. |
| 7 | +// |
| 8 | +// msteams cancel has no inbound event: user Stop makes Teams 403 the next chunk |
| 9 | +// and the SDK throws StreamCancelledError synchronously from the write, so the |
| 10 | +// cancel scenario arms a stream write fault at setup and maps the scripted |
| 11 | +// `cancel` step to nothing. A non-cancel write fault latches streamFailed and |
| 12 | +// the full reply is intentionally re-delivered as blocks even though a prefix |
| 13 | +// already streamed (duplication over truncation — see reply-stream-controller). |
| 14 | +// Refresh goldens with OPENCLAW_TRACE_UPDATE=1 (see delivery-trace harness docs). |
| 15 | +import { |
| 16 | + deliveryTraceScenarios, |
| 17 | + expectDeliveryTraceMatchesGolden, |
| 18 | + runDeliveryTraceScenario, |
| 19 | + type DeliveryTraceInStep, |
| 20 | + type DeliveryTraceScenarioName, |
| 21 | + type WireRecorder, |
| 22 | +} from "openclaw/plugin-sdk/channel-contract-testing"; |
| 23 | +import type { PluginRuntime } from "openclaw/plugin-sdk/core"; |
| 24 | +import { resolveMarkdownTableMode } from "openclaw/plugin-sdk/markdown-table-runtime"; |
| 25 | +import { chunkMarkdownTextWithMode, resolveChunkMode } from "openclaw/plugin-sdk/reply-chunking"; |
| 26 | +import { convertMarkdownTables } from "openclaw/plugin-sdk/text-chunking"; |
| 27 | +import { describe, it } from "vitest"; |
| 28 | +import type { OpenClawConfig, ReplyPayload } from "../runtime-api.js"; |
| 29 | +import { createMSTeamsReplyDispatcher } from "./reply-dispatcher.js"; |
| 30 | +import { setMSTeamsRuntime } from "./runtime.js"; |
| 31 | +import type { MSTeamsTurnContext } from "./sdk-types.js"; |
| 32 | + |
| 33 | +/** Options msteams passes into core createReplyDispatcherWithTyping (capture seam). */ |
| 34 | +type CapturedDispatcherOptions = { |
| 35 | + onReplyStart?: () => Promise<void> | void; |
| 36 | + deliver: (payload: ReplyPayload, info: { kind: string }) => Promise<void> | void; |
| 37 | + typingCallbacks?: { onIdle?: () => void }; |
| 38 | +}; |
| 39 | + |
| 40 | +/** |
| 41 | + * Deterministic stream write fault. Writes are counted across emit/update/close; |
| 42 | + * the fault fires at `atWrite` and every later write, matching a broken or |
| 43 | + * canceled SDK HttpStream that stays broken for the rest of the turn. |
| 44 | + */ |
| 45 | +type StreamWriteFault = { atWrite: number; kind: "cancelled" | "broken" }; |
| 46 | + |
| 47 | +function createStreamCancelledError(): Error { |
| 48 | + // The controller matches by err.name (the SDK class re-export is not |
| 49 | + // resolvable), so the mock only needs the name to be contract-accurate. |
| 50 | + const err = new Error("stream cancelled by user Stop"); |
| 51 | + err.name = "StreamCancelledError"; |
| 52 | + return err; |
| 53 | +} |
| 54 | + |
| 55 | +function createRecordingStream(recorder: WireRecorder, fault?: StreamWriteFault) { |
| 56 | + let writes = 0; |
| 57 | + let canceled = false; |
| 58 | + const applyFault = (method: string, payload?: unknown): void => { |
| 59 | + writes += 1; |
| 60 | + if (!fault || writes < fault.atWrite) { |
| 61 | + return; |
| 62 | + } |
| 63 | + if (fault.kind === "cancelled") { |
| 64 | + // SDK behavior on user Stop: the chunk POST 403s, the streamer flips its |
| 65 | + // canceled flag, and this and every later write throws StreamCancelledError. |
| 66 | + canceled = true; |
| 67 | + recorder.recordWireCall({ |
| 68 | + method, |
| 69 | + ...(payload !== undefined ? { payload } : {}), |
| 70 | + result: { error: "StreamCancelledError" }, |
| 71 | + }); |
| 72 | + throw createStreamCancelledError(); |
| 73 | + } |
| 74 | + recorder.recordWireCall({ |
| 75 | + method, |
| 76 | + ...(payload !== undefined ? { payload } : {}), |
| 77 | + result: { error: "stream write failed" }, |
| 78 | + }); |
| 79 | + throw new Error("Teams stream write failed"); |
| 80 | + }; |
| 81 | + return { |
| 82 | + emit(activity: unknown): void { |
| 83 | + const payload = typeof activity === "string" ? { text: activity } : activity; |
| 84 | + applyFault("stream.emit", payload); |
| 85 | + recorder.recordWireCall({ method: "stream.emit", payload }); |
| 86 | + }, |
| 87 | + update(text: string): void { |
| 88 | + applyFault("stream.update", { text }); |
| 89 | + recorder.recordWireCall({ method: "stream.update", payload: { text } }); |
| 90 | + }, |
| 91 | + async close(): Promise<unknown> { |
| 92 | + applyFault("stream.close"); |
| 93 | + recorder.recordWireCall({ method: "stream.close", result: { id: "stream-final" } }); |
| 94 | + return { id: "stream-final" }; |
| 95 | + }, |
| 96 | + get canceled(): boolean { |
| 97 | + return canceled; |
| 98 | + }, |
| 99 | + }; |
| 100 | +} |
| 101 | + |
| 102 | +function createRecordingTurnContext(params: { |
| 103 | + recorder: WireRecorder; |
| 104 | + conversationId: string; |
| 105 | + conversationType: string; |
| 106 | + stream: ReturnType<typeof createRecordingStream>; |
| 107 | +}): MSTeamsTurnContext { |
| 108 | + let activityCount = 0; |
| 109 | + const reject = (method: string) => async () => { |
| 110 | + throw new Error(`unexpected turn-context call: ${method}`); |
| 111 | + }; |
| 112 | + return { |
| 113 | + activity: { |
| 114 | + type: "message", |
| 115 | + conversation: { id: params.conversationId, conversationType: params.conversationType }, |
| 116 | + }, |
| 117 | + sendActivity: async (activity) => { |
| 118 | + activityCount += 1; |
| 119 | + const id = `activity-${activityCount}`; |
| 120 | + params.recorder.recordWireCall({ |
| 121 | + method: "context.sendActivity", |
| 122 | + target: params.conversationId, |
| 123 | + payload: typeof activity === "string" ? { text: activity } : activity, |
| 124 | + result: { id }, |
| 125 | + }); |
| 126 | + return { id }; |
| 127 | + }, |
| 128 | + sendActivities: reject("sendActivities"), |
| 129 | + updateActivity: reject("updateActivity"), |
| 130 | + deleteActivity: reject("deleteActivity"), |
| 131 | + stream: params.stream, |
| 132 | + }; |
| 133 | +} |
| 134 | + |
| 135 | +function createTraceRuntimeStub( |
| 136 | + recorder: WireRecorder, |
| 137 | + captureDispatcherOptions: (options: CapturedDispatcherOptions) => void, |
| 138 | +): PluginRuntime { |
| 139 | + return { |
| 140 | + channel: { |
| 141 | + // Real text helpers: chunking behavior is part of the recorded lifecycle. |
| 142 | + text: { |
| 143 | + resolveChunkMode, |
| 144 | + chunkMarkdownTextWithMode, |
| 145 | + convertMarkdownTables, |
| 146 | + resolveMarkdownTableMode, |
| 147 | + }, |
| 148 | + reply: { |
| 149 | + createReplyDispatcherWithTyping: (options: CapturedDispatcherOptions) => { |
| 150 | + captureDispatcherOptions(options); |
| 151 | + return { |
| 152 | + dispatcher: {}, |
| 153 | + replyOptions: {}, |
| 154 | + // Mirror core createReplyDispatcherWithTyping.markDispatchIdle: with |
| 155 | + // no typing controller bound it stops typing via the callbacks, which |
| 156 | + // ends the 8s keepalive loop and the 10min TTL timer. |
| 157 | + markDispatchIdle: () => { |
| 158 | + options.typingCallbacks?.onIdle?.(); |
| 159 | + }, |
| 160 | + markRunComplete: () => {}, |
| 161 | + }; |
| 162 | + }, |
| 163 | + resolveHumanDelayConfig: () => undefined, |
| 164 | + }, |
| 165 | + }, |
| 166 | + system: { |
| 167 | + // Delivery-failure escalation is agent-visible choreography; record it so |
| 168 | + // goldens catch drift (none of the adopted scenarios should emit one). |
| 169 | + enqueueSystemEvent: (text: string, opts?: { contextKey?: string }) => { |
| 170 | + recorder.recordWireCall({ |
| 171 | + method: "system.enqueueSystemEvent", |
| 172 | + payload: { text, contextKey: opts?.contextKey }, |
| 173 | + }); |
| 174 | + }, |
| 175 | + }, |
| 176 | + } as unknown as PluginRuntime; |
| 177 | +} |
| 178 | + |
| 179 | +type MSTeamsTraceCase = { |
| 180 | + golden: string; |
| 181 | + scenario: DeliveryTraceScenarioName; |
| 182 | + conversationType: "personal" | "channel"; |
| 183 | + conversationId: string; |
| 184 | + streamWriteFault?: StreamWriteFault; |
| 185 | +}; |
| 186 | + |
| 187 | +const MSTEAMS_TRACE_CASES: readonly MSTeamsTraceCase[] = [ |
| 188 | + { |
| 189 | + // Native append-sink streaming in a DM: cumulative pipeline text becomes |
| 190 | + // delta emits; the AI-label chrome + feedback channelData merge into the |
| 191 | + // closing activity at finalize. |
| 192 | + golden: "streaming-happy-dm", |
| 193 | + scenario: "streaming-happy", |
| 194 | + conversationType: "personal", |
| 195 | + conversationId: "a:1dm-trace-conversation", |
| 196 | + }, |
| 197 | + { |
| 198 | + // Same script in a channel: native streaming is DM-only and typing is |
| 199 | + // unsupported for channel conversations, so everything batches as block |
| 200 | + // sends that flush at markDispatchIdle. |
| 201 | + golden: "block-fallback-channel", |
| 202 | + scenario: "streaming-happy", |
| 203 | + conversationType: "channel", |
| 204 | + conversationId: "19:[email protected];messageid=1000", |
| 205 | + }, |
| 206 | + { |
| 207 | + // User Stop: the second stream write throws StreamCancelledError. Terminal |
| 208 | + // state is canceled — no finalize chrome, no block redelivery, and typing |
| 209 | + // pulses are suppressed for the rest of the turn. |
| 210 | + golden: "cancel-via-write-error", |
| 211 | + scenario: "cancel-mid-stream", |
| 212 | + conversationType: "personal", |
| 213 | + conversationId: "a:1dm-trace-conversation", |
| 214 | + streamWriteFault: { atWrite: 2, kind: "cancelled" }, |
| 215 | + }, |
| 216 | + { |
| 217 | + // Mid-stream non-cancel write failure latches streamFailed: the streamed |
| 218 | + // prefix stays visible AND the full reply re-delivers as blocks. The |
| 219 | + // duplication is the contract (truncation is the worse outcome). |
| 220 | + golden: "stream-failure-redeliver-full", |
| 221 | + scenario: "streaming-happy", |
| 222 | + conversationType: "personal", |
| 223 | + conversationId: "a:1dm-trace-conversation", |
| 224 | + streamWriteFault: { atWrite: 2, kind: "broken" }, |
| 225 | + }, |
| 226 | +]; |
| 227 | + |
| 228 | +function setupMSTeamsTrace(recorder: WireRecorder, traceCase: MSTeamsTraceCase) { |
| 229 | + let captured: CapturedDispatcherOptions | undefined; |
| 230 | + setMSTeamsRuntime( |
| 231 | + createTraceRuntimeStub(recorder, (options) => { |
| 232 | + captured = options; |
| 233 | + }), |
| 234 | + ); |
| 235 | + const stream = createRecordingStream(recorder, traceCase.streamWriteFault); |
| 236 | + const context = createRecordingTurnContext({ |
| 237 | + recorder, |
| 238 | + conversationId: traceCase.conversationId, |
| 239 | + conversationType: traceCase.conversationType, |
| 240 | + stream, |
| 241 | + }); |
| 242 | + const created = createMSTeamsReplyDispatcher({ |
| 243 | + cfg: { channels: { msteams: {} } } as OpenClawConfig, |
| 244 | + agentId: "agent", |
| 245 | + sessionKey: "agent:msteams:trace", |
| 246 | + runtime: { error: () => {} } as never, |
| 247 | + log: { info: () => {}, error: () => {} }, |
| 248 | + app: {} as never, |
| 249 | + appId: "app-trace", |
| 250 | + conversationRef: { |
| 251 | + activityId: "inbound-activity", |
| 252 | + user: { id: "29:trace-user", name: "Trace User" }, |
| 253 | + agent: { id: "28:trace-bot", name: "OpenClaw" }, |
| 254 | + conversation: { |
| 255 | + id: traceCase.conversationId, |
| 256 | + conversationType: traceCase.conversationType, |
| 257 | + tenantId: "tenant-trace", |
| 258 | + }, |
| 259 | + tenantId: "tenant-trace", |
| 260 | + channelId: "msteams", |
| 261 | + serviceUrl: "https://smba.trafficmanager.net/amer/", |
| 262 | + }, |
| 263 | + context, |
| 264 | + // Both the DM policy and the channel default (requireMention=true) resolve |
| 265 | + // to "thread", which keeps sends on the live turn context. |
| 266 | + replyStyle: "thread", |
| 267 | + textLimit: 4000, |
| 268 | + }); |
| 269 | + const options = captured; |
| 270 | + if (!options) { |
| 271 | + throw new Error("dispatcher options were not captured"); |
| 272 | + } |
| 273 | + |
| 274 | + return async (step: DeliveryTraceInStep) => { |
| 275 | + switch (step.kind) { |
| 276 | + case "reply-start": |
| 277 | + await options.onReplyStart?.(); |
| 278 | + break; |
| 279 | + case "partial": |
| 280 | + // Only present when the controller owns a native stream (DM); the |
| 281 | + // channel case has no onPartialReply and partials are dropped. |
| 282 | + created.replyOptions.onPartialReply?.({ text: step.text }); |
| 283 | + break; |
| 284 | + case "block-final": |
| 285 | + await options.deliver({ text: step.text }, { kind: "block" }); |
| 286 | + break; |
| 287 | + case "tool-progress": |
| 288 | + // Progress lines only render in streaming.mode=progress; with the |
| 289 | + // default partial mode this is a controller-side no-op. |
| 290 | + await created.replyOptions.onToolStart?.({ name: step.name, phase: step.phase }); |
| 291 | + break; |
| 292 | + case "final": |
| 293 | + await options.deliver( |
| 294 | + { |
| 295 | + ...(step.text !== undefined ? { text: step.text } : {}), |
| 296 | + ...(step.mediaUrls ? { mediaUrls: step.mediaUrls } : {}), |
| 297 | + ...(step.isError ? { isError: true } : {}), |
| 298 | + }, |
| 299 | + { kind: "final" }, |
| 300 | + ); |
| 301 | + break; |
| 302 | + case "cancel": |
| 303 | + // No cancel event exists on msteams: cancellation is detected via the |
| 304 | + // armed StreamCancelledError write fault, so this step maps to nothing. |
| 305 | + break; |
| 306 | + case "idle": |
| 307 | + await created.markDispatchIdle(); |
| 308 | + break; |
| 309 | + case "wire-fault": |
| 310 | + throw new Error("msteams trace scenarios arm stream write faults at setup instead"); |
| 311 | + } |
| 312 | + }; |
| 313 | +} |
| 314 | + |
| 315 | +describe("msteams delivery trace goldens", () => { |
| 316 | + for (const traceCase of MSTEAMS_TRACE_CASES) { |
| 317 | + it(`records ${traceCase.golden}`, async () => { |
| 318 | + const events = await runDeliveryTraceScenario({ |
| 319 | + scenario: deliveryTraceScenarios[traceCase.scenario], |
| 320 | + setup: (recorder) => setupMSTeamsTrace(recorder, traceCase), |
| 321 | + }); |
| 322 | + expectDeliveryTraceMatchesGolden({ |
| 323 | + goldenUrl: new URL(`./__traces__/${traceCase.golden}.trace.jsonl`, import.meta.url), |
| 324 | + events, |
| 325 | + }); |
| 326 | + }); |
| 327 | + } |
| 328 | +}); |
0 commit comments