diff --git a/.agents/skills/channel-message-flows/SKILL.md b/.agents/skills/channel-message-flows/SKILL.md index 714945b7b89..710c2658435 100644 --- a/.agents/skills/channel-message-flows/SKILL.md +++ b/.agents/skills/channel-message-flows/SKILL.md @@ -6,29 +6,38 @@ description: "Use when running QA Lab channel message flow evidence." # Channel Message Flows Use this from the OpenClaw repo root to run the QA Lab evidence for Telegram -draft/final delivery sequencing. This skill no longer launches a standalone -script; the behavior is owned by the QA scenario and its Vitest-backed e2e test. +draft/final delivery sequencing. The behavior is owned by one transport-native +QA flow that can run through QA Channel or Crabline Telegram. ## QA Scenario Run the scenario through QA Lab: ```bash -pnpm openclaw qa suite --scenario channel-message-flows +OPENCLAW_BUILD_PRIVATE_QA=1 node scripts/run-node.mjs qa suite \ + --provider-mode mock-openai \ + --scenario channel-message-flows \ + --channel-driver qa-channel ``` -Run the focused e2e test directly in a Codex worktree: +Run the same YAML through the real Telegram plugin against Crabline's local +provider server: ```bash -node scripts/run-vitest.mjs extensions/telegram/src/channel-message-flows.qa.e2e.test.ts +OPENCLAW_BUILD_PRIVATE_QA=1 node scripts/run-node.mjs qa suite \ + --provider-mode mock-openai \ + --scenario channel-message-flows \ + --channel-driver crabline \ + --channel telegram ``` ## References - `qa/scenarios/channels/channel-message-flows.yaml` -- `extensions/telegram/src/channel-message-flows.qa.e2e.test.ts` -- `extensions/telegram/src/test-support/channel-message-flows.ts` +- `extensions/qa-channel/src/inbound.ts` +- `extensions/qa-lab/src/qa-transport.ts` +- `extensions/qa-lab/src/crabline-transport.ts` +- `extensions/telegram/src/draft-stream.ts` -The scenario covers `channels.streaming` as primary evidence and records -secondary coverage for thread preservation, delivery ordering, and reasoning -preview visibility. +The scenario covers `channels.streaming` as primary evidence and +`runtime.delivery` as secondary evidence. diff --git a/extensions/qa-channel/src/channel.test.ts b/extensions/qa-channel/src/channel.test.ts index 52c8658db0f..6e14683653e 100644 --- a/extensions/qa-channel/src/channel.test.ts +++ b/extensions/qa-channel/src/channel.test.ts @@ -115,7 +115,9 @@ function createMockQaRuntime(params?: { replyOptions, }: { ctx: { BodyForAgent?: string; Body?: string }; - dispatcherOptions: { deliver: (payload: { text: string }) => Promise }; + dispatcherOptions: { + deliver: (payload: { text: string }, info: { kind: string }) => Promise; + }; replyOptions?: { onToolStart?: (payload: { name?: string; @@ -128,9 +130,12 @@ function createMockQaRuntime(params?: { await replyOptions?.onToolStart?.(toolStart); } params?.onDispatch?.(ctx as Record); - await dispatcherOptions.deliver({ - text: `qa-echo: ${ctx.BodyForAgent ?? ctx.Body ?? ""}`, - }); + await dispatcherOptions.deliver( + { + text: `qa-echo: ${ctx.BodyForAgent ?? ctx.Body ?? ""}`, + }, + { kind: "final" }, + ); }, }, inbound: { diff --git a/extensions/qa-channel/src/inbound.test.ts b/extensions/qa-channel/src/inbound.test.ts index c116d092f6f..200e5edc8da 100644 --- a/extensions/qa-channel/src/inbound.test.ts +++ b/extensions/qa-channel/src/inbound.test.ts @@ -1,9 +1,20 @@ // Qa Channel tests cover inbound plugin behavior. import { createPluginRuntimeMock } from "openclaw/plugin-sdk/channel-test-helpers"; -import { describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import { setQaChannelRuntime } from "../api.js"; +import { deleteQaBusMessage, editQaBusMessage, sendQaBusMessage } from "./bus-client.js"; import { handleQaInbound, isHttpMediaUrl } from "./inbound.js"; +vi.mock("./bus-client.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + deleteQaBusMessage: vi.fn(async () => ({ message: {} })), + editQaBusMessage: vi.fn(async () => ({ message: {} })), + sendQaBusMessage: vi.fn(async () => ({ message: { id: "preview-1" } })), + }; +}); + type HandleQaInboundParams = Parameters[0]; function createQaInboundParams( @@ -66,6 +77,161 @@ describe("isHttpMediaUrl", () => { }); describe("handleQaInbound", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("publishes partial replies as one edited preview before final delivery", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound( + createQaInboundParams({ + message: { + conversation: { id: "qa-room", kind: "group" }, + threadId: "42", + }, + }), + ); + + const assembled = firstRunAssembledParams(runtime); + await assembled.replyOptions?.onPartialReply?.({ text: "preview" }); + await assembled.replyOptions?.onPartialReply?.({ text: "preview expanded" }); + await assembled.delivery.deliver({ text: "final answer" }, { kind: "final" }); + + expect(sendQaBusMessage).toHaveBeenCalledOnce(); + expect(sendQaBusMessage).toHaveBeenCalledWith( + expect.objectContaining({ + replyToId: "msg-1", + text: "preview", + threadId: "42", + to: "thread:qa-room/42", + }), + ); + expect(editQaBusMessage).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ messageId: "preview-1", text: "preview expanded" }), + ); + expect(editQaBusMessage).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ messageId: "preview-1", text: "final answer" }), + ); + }); + + it("keeps block deliveries separate and retains tool calls discovered after a preview", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.replyOptions?.onPartialReply?.({ text: "preview" }); + await assembled.replyOptions?.onToolStart?.({ + phase: "start", + name: "search", + args: { query: "qa" }, + }); + await assembled.delivery.deliver({ text: "tool result" }, { kind: "block" }); + await assembled.delivery.deliver({ text: "final answer" }, { kind: "final" }); + + expect(deleteQaBusMessage).toHaveBeenCalledOnce(); + expect(sendQaBusMessage).toHaveBeenCalledTimes(3); + expect(sendQaBusMessage).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + text: "tool result", + toolCalls: [{ name: "search", arguments: { query: "[redacted]" } }], + }), + ); + expect(sendQaBusMessage).toHaveBeenNthCalledWith( + 3, + expect.objectContaining({ + text: "final answer", + toolCalls: [{ name: "search", arguments: { query: "[redacted]" } }], + }), + ); + }); + + it("deletes an active preview when reply dispatch fails", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.replyOptions?.onPartialReply?.({ text: "unfinished preview" }); + assembled.delivery.onError?.(new Error("model failed"), { kind: "final" }); + + await vi.waitFor(() => { + expect(deleteQaBusMessage).toHaveBeenCalledWith( + expect.objectContaining({ messageId: "preview-1" }), + ); + }); + }); + + it("deletes a preview after a queued edit fails", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + vi.mocked(editQaBusMessage).mockRejectedValueOnce(new Error("edit failed")); + + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.replyOptions?.onPartialReply?.({ text: "first preview" }); + await expect( + assembled.replyOptions?.onPartialReply?.({ text: "broken preview" }), + ).rejects.toThrow("edit failed"); + assembled.delivery.onError?.(new Error("dispatch failed"), { kind: "final" }); + + await vi.waitFor(() => { + expect(deleteQaBusMessage).toHaveBeenCalledWith( + expect.objectContaining({ messageId: "preview-1" }), + ); + }); + }); + + it("escapes control characters in dispatch error logs", async () => { + const runtime = createPluginRuntimeMock(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const c1Control = String.fromCharCode(0x9b); + const lineSeparator = String.fromCodePoint(0x2028); + const paragraphSeparator = String.fromCodePoint(0x2029); + vi.mocked(deleteQaBusMessage).mockRejectedValueOnce( + new Error(`cleanup\nforged\u001b[31m${c1Control}32m${lineSeparator}next`), + ); + setQaChannelRuntime(runtime); + + try { + await handleQaInbound(createQaInboundParams()); + + const assembled = firstRunAssembledParams(runtime); + await assembled.replyOptions?.onPartialReply?.({ text: "unfinished preview" }); + assembled.delivery.onError?.(new Error(`dispatch\r\nforged${paragraphSeparator}next`), { + kind: "final", + }); + + await vi.waitFor(() => { + expect(warn).toHaveBeenCalledTimes(2); + }); + assembled.delivery.onError?.(undefined, { kind: "final" }); + await vi.waitFor(() => { + expect(warn).toHaveBeenCalledTimes(3); + }); + const output = warn.mock.calls.flat().join(" "); + expect(output).not.toContain("\r"); + expect(output).not.toContain("\n"); + expect(output).not.toContain(String.fromCharCode(0x1b)); + expect(output).not.toContain(c1Control); + expect(output).not.toContain(lineSeparator); + expect(output).not.toContain(paragraphSeparator); + expect(output).toContain("dispatch\\u000d\\u000aforged\\u2029next"); + expect(output).toContain("cleanup\\u000aforged\\u001b[31m\\u009b32m\\u2028next"); + expect(output).toContain("[object Undefined]"); + } finally { + warn.mockRestore(); + } + }); + it("marks group messages that match configured mention patterns", async () => { const runtime = createPluginRuntimeMock(); vi.mocked(runtime.channel.mentions.buildMentionRegexes).mockReturnValue([/\b@?openclaw\b/i]); diff --git a/extensions/qa-channel/src/inbound.ts b/extensions/qa-channel/src/inbound.ts index 36364d116cf..2dcb0dcdb68 100644 --- a/extensions/qa-channel/src/inbound.ts +++ b/extensions/qa-channel/src/inbound.ts @@ -2,6 +2,7 @@ import { resolveStableChannelMessageIngress } from "openclaw/plugin-sdk/channel-ingress-runtime"; import { resolveNativeCommandSessionTargets } from "openclaw/plugin-sdk/command-auth-native"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { resolveInboundRouteEnvelopeBuilderWithRuntime } from "openclaw/plugin-sdk/inbound-envelope"; import { buildAgentMediaPayload, @@ -12,7 +13,13 @@ import { sanitizeQaBusToolCallArguments, type QaBusToolCall, } from "openclaw/plugin-sdk/qa-channel-protocol"; -import { buildQaTarget, sendQaBusMessage, type QaBusMessage } from "./bus-client.js"; +import { + buildQaTarget, + deleteQaBusMessage, + editQaBusMessage, + sendQaBusMessage, + type QaBusMessage, +} from "./bus-client.js"; import { getQaChannelRuntime } from "./runtime.js"; import type { CoreConfig, ResolvedQaChannelAccount } from "./types.js"; @@ -91,6 +98,106 @@ function resolveQaGroupConfig(params: { return groups?.[params.conversationId] ?? groups?.[params.target] ?? groups?.["*"]; } +function formatQaErrorForLog(error: unknown): string { + let escaped = ""; + const message = formatErrorMessage(error) || Object.prototype.toString.call(error); + for (const character of message) { + const codePoint = character.codePointAt(0) ?? 0; + const isControl = codePoint <= 0x1f || (codePoint >= 0x7f && codePoint <= 0x9f); + const isLineSeparator = codePoint === 0x2028 || codePoint === 0x2029; + escaped += + isControl || isLineSeparator ? `\\u${codePoint.toString(16).padStart(4, "0")}` : character; + } + return escaped; +} + +function createQaReplyPreview(params: { + account: ResolvedQaChannelAccount; + inbound: QaBusMessage; + target: string; + toolCalls: QaBusToolCall[]; +}) { + let messageId: string | null = null; + let currentText = ""; + let pending = Promise.resolve(); + + const write = (text: string) => { + if (!text.trim() || text === currentText) { + return pending; + } + pending = pending.then(async () => { + if (messageId) { + await editQaBusMessage({ + baseUrl: params.account.baseUrl, + accountId: params.account.accountId, + messageId, + text, + }); + } else { + const response = await sendQaBusMessage({ + baseUrl: params.account.baseUrl, + accountId: params.account.accountId, + to: params.target, + text, + senderId: params.account.botUserId, + senderName: params.account.botDisplayName, + threadId: params.inbound.threadId, + replyToId: params.inbound.id, + toolCalls: params.toolCalls, + }); + messageId = response.message.id; + } + currentText = text; + }); + return pending; + }; + + const clear = async () => { + await pending.catch(() => undefined); + if (!messageId) { + return; + } + await deleteQaBusMessage({ + baseUrl: params.account.baseUrl, + accountId: params.account.accountId, + messageId, + }); + messageId = null; + currentText = ""; + }; + + const sendDurable = async (text: string) => { + if (!text.trim()) { + return; + } + await sendQaBusMessage({ + baseUrl: params.account.baseUrl, + accountId: params.account.accountId, + to: params.target, + text, + senderId: params.account.botUserId, + senderName: params.account.botDisplayName, + threadId: params.inbound.threadId, + replyToId: params.inbound.id, + toolCalls: params.toolCalls, + }); + }; + + return { + clear, + async deliver(text: string, kind: string) { + await pending; + if (kind === "final" && messageId && params.toolCalls.length === 0) { + await write(text); + return; + } + await clear(); + await sendDurable(text); + }, + update: write, + }; +} + export async function handleQaInbound(params: { channelId: string; channelLabel: string; @@ -106,6 +213,12 @@ export async function handleQaInbound(params: { threadId: inbound.threadId, }); const toolCalls: QaBusToolCall[] = []; + const preview = createQaReplyPreview({ + account: params.account, + inbound, + target, + toolCalls, + }); const { route, buildEnvelope } = resolveInboundRouteEnvelopeBuilderWithRuntime({ cfg: params.config as OpenClawConfig, channel: params.channelId, @@ -251,7 +364,7 @@ export async function handleQaInbound(params: { dispatchReplyWithBufferedBlockDispatcher: runtime.channel.reply.dispatchReplyWithBufferedBlockDispatcher, delivery: { - deliver: async (payload) => { + deliver: async (payload, info) => { const text = payload && typeof payload === "object" && "text" in payload ? ((payload as { text?: string }).text ?? "") @@ -259,25 +372,21 @@ export async function handleQaInbound(params: { if (!text.trim()) { return; } - await sendQaBusMessage({ - baseUrl: params.account.baseUrl, - accountId: params.account.accountId, - to: target, - text, - senderId: params.account.botUserId, - senderName: params.account.botDisplayName, - threadId: inbound.threadId, - replyToId: inbound.id, - toolCalls, - }); + await preview.deliver(text, info.kind); }, onError: (error) => { - throw error instanceof Error - ? error - : new Error(`qa-channel dispatch failed: ${String(error)}`); + void preview.clear().catch((clearError: unknown) => { + console.warn( + `[qa-channel] failed to clear reply preview after dispatch error: ${formatQaErrorForLog(clearError)}`, + ); + }); + console.warn(`[qa-channel] reply dispatch failed: ${formatQaErrorForLog(error)}`); }, }, replyOptions: { + onPartialReply: async (payload) => { + await preview.update(payload.text ?? ""); + }, onToolStart: (payload) => { if (payload.phase && payload.phase !== "start") { return; diff --git a/extensions/qa-lab/src/crabline-transport.test.ts b/extensions/qa-lab/src/crabline-transport.test.ts index eb1a9ce5339..b670b364b06 100644 --- a/extensions/qa-lab/src/crabline-transport.test.ts +++ b/extensions/qa-lab/src/crabline-transport.test.ts @@ -103,6 +103,63 @@ describe("crabline transport", () => { }); }); + it("observes Telegram preview edits through the shared transport adapter", async () => { + await withTempDir("qa-crabline-transport-", async (outputDir) => { + const transport = await createQaCrablineTransportAdapter({ + outputDir, + selection: createSelection(), + state: createQaBusState(), + }); + + try { + const manifest = JSON.parse( + await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"), + ) as { + botToken: string; + endpoints: { apiRoot: string }; + }; + const postTelegram = async (method: string, body: Record) => { + const response = await fetch( + `${manifest.endpoints.apiRoot}/bot${manifest.botToken}/${method}`, + { + body: JSON.stringify(body), + headers: { "content-type": "application/json" }, + method: "POST", + }, + ); + expect(response.ok).toBe(true); + return (await response.json()) as { result: { message_id: number } }; + }; + const sent = await postTelegram("sendMessage", { + chat_id: "-1001234567890", + message_thread_id: 42, + text: "preview text", + }); + await postTelegram("editMessageText", { + chat_id: "-1001234567890", + message_id: sent.result.message_id, + text: "final marker", + }); + + await expect( + transport.waitForOutboundSequence({ + conversationId: "-1001234567890", + finalSettleMs: 0, + finalTextIncludes: "final marker", + minimumPreviewEvents: 1, + threadId: "42", + timeoutMs: 1_000, + }), + ).resolves.toMatchObject({ + events: [{ kind: "sent" }, { kind: "edited" }], + final: { text: "final marker", threadId: "42" }, + }); + } finally { + await transport.cleanup?.(); + } + }); + }); + it("configures OpenClaw's Slack plugin against a Crabline local provider server", async () => { await withTempDir("qa-crabline-transport-", async (outputDir) => { const transport = await createQaCrablineTransportAdapter({ diff --git a/extensions/qa-lab/src/crabline-transport.ts b/extensions/qa-lab/src/crabline-transport.ts index 15385b66dc0..adfd8dfcb80 100644 --- a/extensions/qa-lab/src/crabline-transport.ts +++ b/extensions/qa-lab/src/crabline-transport.ts @@ -12,19 +12,30 @@ import { import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime"; +import { + isRecord, + normalizeStringifiedOptionalString, + readStringValue, +} from "openclaw/plugin-sdk/string-coerce-runtime"; import { createQaBusState, type QaBusState } from "./bus-state.js"; import { QaSuiteInfraError } from "./errors.js"; -import { QaStateBackedTransportAdapter } from "./qa-transport.js"; +import { + QaStateBackedTransportAdapter, + waitForQaTransportOutboundSequence, +} from "./qa-transport.js"; import type { QaTransportActionName, QaTransportGatewayClient, QaTransportGatewayConfig, QaTransportNativeCommandInput, + QaTransportOutboundEvent, + QaTransportOutboundSequenceMatch, QaTransportReportParams, QaTransportState, } from "./qa-transport.js"; import type { QaBusInboundMessageInput, + QaBusMessage, QaBusOutboundMessageInput, QaBusSearchMessagesInput, QaBusWaitForInput, @@ -35,9 +46,79 @@ const RECORDER_SYNC_INTERVAL_MS = 50; type QaCrablineTransportState = QaTransportState & { cleanup: () => Promise; + getOutboundEvents: () => Promise; rememberProviderTarget: (providerTargetKey: string, qaTarget: string) => void; }; +const TELEGRAM_LIFECYCLE_METHOD_RE = /\/(sendMessage|editMessageText|deleteMessage)$/u; + +function readTelegramLifecycleEvent(params: { + cursor: number; + event: unknown; + messageByProviderId: Map; + pendingByChat: Map; +}): QaTransportOutboundEvent | null { + if (!isRecord(params.event) || params.event.type !== "api") { + return null; + } + const pathValue = readStringValue(params.event.path); + const method = pathValue ? TELEGRAM_LIFECYCLE_METHOD_RE.exec(pathValue)?.[1] : undefined; + if (!method || !isRecord(params.event.body)) { + return null; + } + const chatId = normalizeStringifiedOptionalString(params.event.body.chat_id); + if (!chatId) { + return null; + } + const providerMessageId = normalizeStringifiedOptionalString(params.event.body.message_id); + const providerKey = providerMessageId ? `${chatId}:${providerMessageId}` : null; + let previous = providerKey ? params.messageByProviderId.get(providerKey) : undefined; + if (!previous && providerKey && providerMessageId) { + const pending = params.pendingByChat.get(chatId) ?? []; + if (pending.length === 1) { + previous = pending[0]; + previous.id = providerMessageId; + params.messageByProviderId.set(providerKey, previous); + params.pendingByChat.delete(chatId); + } + } + const text = readStringValue(params.event.body.text) ?? previous?.text ?? ""; + if (!text && method !== "deleteMessage") { + return null; + } + const threadId = + normalizeStringifiedOptionalString(params.event.body.message_thread_id) ?? previous?.threadId; + const message: QaBusMessage = { + id: providerMessageId ?? previous?.id ?? `crabline-${params.cursor}`, + accountId: "default", + direction: "outbound", + conversation: { + id: chatId, + kind: chatId.startsWith("-") ? "group" : "direct", + }, + senderId: "openclaw", + senderName: "OpenClaw QA", + text, + timestamp: Date.now(), + ...(threadId ? { threadId } : {}), + ...(method === "deleteMessage" ? { deleted: true } : {}), + ...(method === "editMessageText" ? { editedAt: Date.now() } : {}), + reactions: [], + }; + if (method === "sendMessage") { + const pending = params.pendingByChat.get(chatId) ?? []; + pending.push(message); + params.pendingByChat.set(chatId, pending); + } else if (providerKey) { + params.messageByProviderId.set(providerKey, message); + } + return { + cursor: params.cursor, + kind: method === "sendMessage" ? "sent" : method === "editMessageText" ? "edited" : "deleted", + message, + }; +} + async function waitForCrablineReady(params: { accountId: string; channel: string; @@ -130,6 +211,9 @@ function createCrablineState(params: { }): QaCrablineTransportState { const baseState = params.state; const targetByProviderTarget = new Map(); + const telegramMessageByProviderId = new Map(); + const pendingTelegramMessagesByChat = new Map(); + const outboundEvents: QaTransportOutboundEvent[] = []; let recorderLineCursor = 0; let syncPromise: Promise | null = null; @@ -149,6 +233,17 @@ function createCrablineState(params: { const lines = text.split(/\r?\n/u).filter((line) => line.trim().length > 0); for (const line of lines.slice(recorderLineCursor)) { const parsed = JSON.parse(line) as unknown; + if (params.adapter.channel === "telegram") { + const lifecycle = readTelegramLifecycleEvent({ + cursor: outboundEvents.length + 1, + event: parsed, + messageByProviderId: telegramMessageByProviderId, + pendingByChat: pendingTelegramMessagesByChat, + }); + if (lifecycle) { + outboundEvents.push(lifecycle); + } + } const outbound = params.adapter.createOutboundFromRecorderEvent({ event: parsed, targetByProviderTarget, @@ -176,12 +271,19 @@ function createCrablineState(params: { await syncRecorder(); baseState.reset(); targetByProviderTarget.clear(); + telegramMessageByProviderId.clear(); + pendingTelegramMessagesByChat.clear(); + outboundEvents.length = 0; recorderLineCursor = await fs .readFile(params.adapter.manifest.recorderPath, "utf8") .then((text) => text.split(/\r?\n/u).filter((line) => line.trim().length > 0).length) .catch(() => 0); }, getSnapshot: baseState.getSnapshot.bind(baseState), + async getOutboundEvents() { + await syncRecorder(); + return outboundEvents; + }, async addInboundMessage(input: QaBusInboundMessageInput) { const providerInbound = params.adapter.createInbound({ input }); targetByProviderTarget.set(providerInbound.providerTargetKey, providerInbound.qaTarget); @@ -275,6 +377,13 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { }); } + override async waitForOutboundSequence(input: QaTransportOutboundSequenceMatch) { + return await waitForQaTransportOutboundSequence({ + input, + readEvents: () => this.#state.getOutboundEvents(), + }); + } + handleAction = async (_params: { action: QaTransportActionName; args: Record; diff --git a/extensions/qa-lab/src/providers/mock-openai/server.test.ts b/extensions/qa-lab/src/providers/mock-openai/server.test.ts index 71ee3f22518..3ef425b5f58 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.test.ts @@ -355,6 +355,33 @@ describe("qa mock openai server", () => { expect(quietBody).toContain('"phase":"final_answer"'); expect(quietBody).toContain("QA_STREAMING_OK"); + const finalOnlyMarkerResponse = await fetch(`${server.baseUrl}/v1/responses`, { + method: "POST", + headers: { + "content-type": "application/json", + }, + body: JSON.stringify({ + stream: true, + input: [ + makeUserInput( + "Final-only marker streaming QA check. Reply exactly: QA-FINAL-ONLY-STREAMING-OK", + ), + ], + }), + }); + expect(finalOnlyMarkerResponse.status).toBe(200); + const finalOnlyMarkerBody = await finalOnlyMarkerResponse.text(); + const finalOnlyMarkerDeltaText = finalOnlyMarkerBody + .split("\n") + .filter((line) => line.startsWith("data: {")) + .map((line) => JSON.parse(line.slice("data: ".length)) as { type?: string; delta?: string }) + .filter((event) => event.type === "response.output_text.delta") + .map((event) => event.delta ?? "") + .join(""); + expect(finalOnlyMarkerDeltaText).toBe("QA streaming preview in progress"); + expect(finalOnlyMarkerDeltaText).not.toContain("QA-FINAL-ONLY-STREAMING-OK"); + expect(finalOnlyMarkerBody).toContain('"text":"QA-FINAL-ONLY-STREAMING-OK"'); + const partialResponse = await fetch(`${server.baseUrl}/v1/responses`, { method: "POST", headers: { diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index a4136faf22f..ca9b1e34a70 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -159,6 +159,7 @@ const QA_THINKING_VISIBILITY_MAX_PROMPT_RE = /qa thinking visibility check max/i const QA_EMPTY_RESPONSE_RECOVERY_PROMPT_RE = /empty response continuation qa check/i; const QA_EMPTY_RESPONSE_EXHAUSTION_PROMPT_RE = /empty response exhaustion qa check/i; const QA_STREAMING_PROMPT_RE = /(?:partial|quiet) streaming qa check/i; +const QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE = /final-only marker streaming qa check/i; const QA_BLOCK_STREAMING_PROMPT_RE = /block streaming qa check/i; const QA_TOOL_PROGRESS_ERROR_PROMPT_RE = /tool progress error qa check/i; const QA_TOOL_PROGRESS_PROMPT_RE = /tool progress qa check/i; @@ -284,6 +285,31 @@ function writeSse(res: ServerResponse, events: StreamEvent[]) { res.end(body); } +async function writeSseWithPreviewPause( + res: ServerResponse, + events: StreamEvent[], + pauseMs: number, +) { + const completionIndex = events.findIndex((event) => event.type === "response.output_text.done"); + if (completionIndex < 0) { + writeSse(res, events); + return; + } + res.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + }); + for (const event of events.slice(0, completionIndex)) { + res.write(`data: ${JSON.stringify(event)}\n\n`); + } + await sleep(pauseMs); + for (const event of events.slice(completionIndex)) { + res.write(`data: ${JSON.stringify(event)}\n\n`); + } + res.end("data: [DONE]\n\n"); +} + type AnthropicStreamEvent = Record & { type: string; }; @@ -2418,6 +2444,16 @@ async function buildResponsesPayload( }, ]); } + if (QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE.test(allInputText) && exactReplyDirective) { + return buildAssistantEvents([ + { + id: "msg_mock_final_only_marker_stream", + phase: "final_answer", + streamDeltas: splitMockStreamingText("QA streaming preview in progress"), + text: exactReplyDirective, + }, + ]); + } if (QA_STREAMING_PROMPT_RE.test(allInputText) && exactReplyDirective) { return buildAssistantEvents([ { @@ -3754,7 +3790,11 @@ export async function startQaMockOpenAiServer(params?: { host?: string; port?: n writeJson(res, 200, completion.response); return; } - writeSse(res, events); + if (QA_FINAL_ONLY_MARKER_STREAMING_PROMPT_RE.test(allInputText)) { + await writeSseWithPreviewPause(res, events, 1_500); + } else { + writeSse(res, events); + } return; } if (req.method === "POST" && url.pathname === "/v1/messages") { diff --git a/extensions/qa-lab/src/qa-transport.test.ts b/extensions/qa-lab/src/qa-transport.test.ts new file mode 100644 index 00000000000..cf3079fa08d --- /dev/null +++ b/extensions/qa-lab/src/qa-transport.test.ts @@ -0,0 +1,103 @@ +// Qa Lab tests cover shared transport behavior. +import { describe, expect, it } from "vitest"; +import { createQaBusState } from "./bus-state.js"; +import { waitForQaTransportOutboundSequence } from "./qa-transport.js"; + +describe("waitForQaTransportOutboundSequence", () => { + it("returns preview and final edit events for one threaded message", async () => { + const state = createQaBusState(); + state.createThread({ + conversationId: "qa-room", + createdBy: "alice", + title: "QA thread", + }); + const preview = state.addOutboundMessage({ + accountId: "default", + senderId: "openclaw", + text: "preview", + threadId: "42", + to: "thread:qa-room/42", + }); + state.editMessage({ + accountId: "default", + messageId: preview.id, + text: "final marker", + }); + + await expect( + waitForQaTransportOutboundSequence({ + input: { + conversationId: "qa-room", + finalSettleMs: 0, + finalTextIncludes: "final marker", + minimumPreviewEvents: 1, + threadId: "42", + timeoutMs: 100, + }, + readEvents: () => state.getSnapshot().events, + }), + ).resolves.toMatchObject({ + events: [{ kind: "sent" }, { kind: "edited" }], + final: { text: "final marker", threadId: "42" }, + }); + }); + + it("does not accept a matching preview that is deleted during final settling", async () => { + const state = createQaBusState(); + const preview = state.addOutboundMessage({ + accountId: "default", + senderId: "openclaw", + text: "preview", + to: "dm:alice", + }); + state.editMessage({ + accountId: "default", + messageId: preview.id, + text: "final marker", + }); + setTimeout(() => { + state.deleteMessage({ accountId: "default", messageId: preview.id }); + }, 5); + + await expect( + waitForQaTransportOutboundSequence({ + input: { + conversationId: "alice", + finalSettleMs: 20, + finalTextIncludes: "final marker", + minimumPreviewEvents: 1, + timeoutMs: 50, + }, + readEvents: () => state.getSnapshot().events, + }), + ).rejects.toThrow("timed out after 50ms"); + }); + + it("does not count an already-final send as a preview", async () => { + const state = createQaBusState(); + const final = state.addOutboundMessage({ + accountId: "default", + senderId: "openclaw", + text: "final marker", + to: "dm:alice", + }); + state.editMessage({ + accountId: "default", + messageId: final.id, + text: "final marker", + }); + + await expect( + waitForQaTransportOutboundSequence({ + input: { + conversationId: "alice", + finalSettleMs: 0, + finalTextIncludes: "final marker", + minimumPreviewEvents: 1, + timeoutMs: 20, + }, + readEvents: () => state.getSnapshot().events, + }), + ).rejects.toThrow("timed out after 20ms"); + }); +}); diff --git a/extensions/qa-lab/src/qa-transport.ts b/extensions/qa-lab/src/qa-transport.ts index 68330d0366c..a9b277d9002 100644 --- a/extensions/qa-lab/src/qa-transport.ts +++ b/extensions/qa-lab/src/qa-transport.ts @@ -5,6 +5,7 @@ import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime"; import type { QaProviderMode } from "./model-selection.js"; import { extractQaFailureReplyText } from "./reply-failure.js"; import type { + QaBusEvent, QaBusInboundMessageInput, QaBusMessage, QaBusOutboundMessageInput, @@ -70,6 +71,27 @@ export type QaTransportWaitForNoOutboundInput = { sinceIndex?: number; }; +export type QaTransportOutboundEvent = { + cursor: number; + kind: "sent" | "edited" | "deleted"; + message: QaBusMessage; +}; + +export type QaTransportOutboundSequenceMatch = { + conversationId?: string; + finalSettleMs?: number; + finalTextIncludes: string; + minimumPreviewEvents?: number; + sinceCursor?: number; + threadId?: string; + timeoutMs?: number; +}; + +export type QaTransportOutboundSequence = { + events: QaTransportOutboundEvent[]; + final: QaBusMessage; +}; + export type QaTransportNativeCommandInput = Omit< QaBusInboundMessageInput, "nativeCommand" | "text" @@ -191,6 +213,9 @@ export type QaTransportAdapter = { sendNativeCommand: (input: QaTransportNativeCommandInput) => Promise; waitForNoOutbound: (input?: QaTransportWaitForNoOutboundInput) => Promise; waitForOutbound: (input: QaTransportOutboundMatch) => Promise; + waitForOutboundSequence: ( + input: QaTransportOutboundSequenceMatch, + ) => Promise; createGatewayConfig: (params: { baseUrl: string }) => QaTransportGatewayConfig; waitReady: (params: { gateway: QaTransportGatewayClient; @@ -325,6 +350,13 @@ export abstract class QaStateBackedTransportAdapter implements QaTransportAdapte }, input.timeoutMs); } + async waitForOutboundSequence(input: QaTransportOutboundSequenceMatch) { + return await waitForQaTransportOutboundSequence({ + input, + readEvents: () => this.state.getSnapshot().events, + }); + } + private outboundSince(sinceIndex = 0) { return this.state .getSnapshot() @@ -332,3 +364,90 @@ export abstract class QaStateBackedTransportAdapter implements QaTransportAdapte .slice(sinceIndex); } } + +function normalizeQaBusOutboundEvent(event: QaBusEvent): QaTransportOutboundEvent | null { + switch (event.kind) { + case "outbound-message": + return { cursor: event.cursor, kind: "sent", message: event.message }; + case "message-edited": + return { cursor: event.cursor, kind: "edited", message: event.message }; + case "message-deleted": + return { cursor: event.cursor, kind: "deleted", message: event.message }; + default: + return null; + } +} + +function isQaTransportOutboundEvent( + event: QaBusEvent | QaTransportOutboundEvent, +): event is QaTransportOutboundEvent { + return event.kind === "sent" || event.kind === "edited" || event.kind === "deleted"; +} + +export async function waitForQaTransportOutboundSequence(params: { + input: QaTransportOutboundSequenceMatch; + readEvents: () => + | readonly (QaBusEvent | QaTransportOutboundEvent)[] + | Promise; +}): Promise { + const minimumPreviewEvents = params.input.minimumPreviewEvents ?? 1; + const finalSettleMs = params.input.finalSettleMs ?? 300; + let stableCursor: number | null = null; + let stableSince = 0; + return await waitForQaTransportCondition(async () => { + const events = (await params.readEvents()) + .filter((event) => event.cursor > (params.input.sinceCursor ?? 0)) + .map((event) => + isQaTransportOutboundEvent(event) ? event : normalizeQaBusOutboundEvent(event), + ) + .filter((event): event is QaTransportOutboundEvent => event !== null) + .filter(({ message }) => { + if ( + params.input.conversationId && + message.conversation.id !== params.input.conversationId + ) { + return false; + } + return !params.input.threadId || message.threadId === params.input.threadId; + }); + const finalIndex = events.findLastIndex( + ({ kind, message }) => + kind !== "deleted" && message.text.includes(params.input.finalTextIncludes), + ); + if (finalIndex < 0) { + return undefined; + } + const candidate = events[finalIndex]; + const sequenceEvents = events.filter(({ message }) => message.id === candidate.message.id); + const latest = sequenceEvents.at(-1); + if ( + !latest || + latest.kind === "deleted" || + !latest.message.text.includes(params.input.finalTextIncludes) + ) { + stableCursor = null; + return undefined; + } + const previewEvents = sequenceEvents.filter( + ({ cursor, kind, message }) => + cursor < candidate.cursor && + kind !== "deleted" && + !message.text.includes(params.input.finalTextIncludes), + ); + if (previewEvents.length < minimumPreviewEvents) { + return undefined; + } + if (stableCursor !== latest.cursor) { + stableCursor = latest.cursor; + stableSince = Date.now(); + return finalSettleMs === 0 ? { events: sequenceEvents, final: latest.message } : undefined; + } + if (Date.now() - stableSince < finalSettleMs) { + return undefined; + } + return { + events: sequenceEvents, + final: latest.message, + }; + }, params.input.timeoutMs); +} diff --git a/extensions/qa-lab/src/scenario-catalog.test.ts b/extensions/qa-lab/src/scenario-catalog.test.ts index 333f894fcf0..e871aef7d6a 100644 --- a/extensions/qa-lab/src/scenario-catalog.test.ts +++ b/extensions/qa-lab/src/scenario-catalog.test.ts @@ -792,6 +792,15 @@ describe("qa scenario catalog", () => { expect(flow).toContain("String(memoryAfter) === config.seededMemory"); }); + it("enables Telegram previews for channel streaming evidence", () => { + const scenario = readQaScenarioById("channel-message-flows"); + + expect(scenario.coverage?.primary).toContain("channels.streaming"); + expect(scenario.gatewayConfigPatch).toMatchObject({ + channels: { telegram: { streaming: { mode: "partial" } } }, + }); + }); + it("rejects malformed string matcher lists before running a flow", () => { expect(() => validateQaScenarioExecutionConfig({ diff --git a/extensions/qa-lab/src/scenario-catalog.ts b/extensions/qa-lab/src/scenario-catalog.ts index 3600efbcec4..f00c8555e95 100644 --- a/extensions/qa-lab/src/scenario-catalog.ts +++ b/extensions/qa-lab/src/scenario-catalog.ts @@ -166,6 +166,10 @@ const qaFlowTransportActionSchema = z.union([ waitForOutbound: z.unknown(), saveAs: z.string().trim().min(1).optional(), }), + z.object({ + waitForOutboundSequence: z.unknown(), + saveAs: z.string().trim().min(1).optional(), + }), z.object({ waitForNoOutbound: z.unknown(), }), diff --git a/extensions/qa-lab/src/scenario-flow-runner.test.ts b/extensions/qa-lab/src/scenario-flow-runner.test.ts index df7da7a6bf9..64a0e7c1bc3 100644 --- a/extensions/qa-lab/src/scenario-flow-runner.test.ts +++ b/extensions/qa-lab/src/scenario-flow-runner.test.ts @@ -74,6 +74,9 @@ async function runLoadedScenarioFlow( } throw new Error(`timed out after ${input.timeoutMs}ms waiting for outbound marker`); }, + waitForOutboundSequence: async () => { + throw new Error("outbound sequence not configured for this fixture"); + }, }; const api = { env: {}, diff --git a/extensions/qa-lab/src/scenario-flow-runner.ts b/extensions/qa-lab/src/scenario-flow-runner.ts index 36f64f07f9f..fdbc434518d 100644 --- a/extensions/qa-lab/src/scenario-flow-runner.ts +++ b/extensions/qa-lab/src/scenario-flow-runner.ts @@ -165,6 +165,7 @@ async function runFlowAction(action: unknown, api: QaFlowApi, vars: QaFlowVars) "sendInbound", "sendNativeCommand", "waitForOutbound", + "waitForOutboundSequence", "waitForNoOutbound", ] as const) { if (name in action) { diff --git a/extensions/qa-lab/src/scenario-runtime-api.test.ts b/extensions/qa-lab/src/scenario-runtime-api.test.ts index 773b85b40a6..90cb45b69c6 100644 --- a/extensions/qa-lab/src/scenario-runtime-api.test.ts +++ b/extensions/qa-lab/src/scenario-runtime-api.test.ts @@ -154,6 +154,9 @@ describe("createQaScenarioRuntimeApi", () => { waitForOutbound: vi.fn(async () => { throw new Error("not used"); }), + waitForOutboundSequence: vi.fn(async () => { + throw new Error("not used"); + }), capabilities: { waitForCondition, getNormalizedMessageState: state.getSnapshot.bind(state), diff --git a/extensions/qa-lab/src/suite-runtime-flow.test.ts b/extensions/qa-lab/src/suite-runtime-flow.test.ts index 5ca278fa2d2..636a5fff93d 100644 --- a/extensions/qa-lab/src/suite-runtime-flow.test.ts +++ b/extensions/qa-lab/src/suite-runtime-flow.test.ts @@ -184,6 +184,7 @@ describe("qa suite runtime flow", () => { sendNativeCommand: vi.fn(), waitForNoOutbound: vi.fn(), waitForOutbound: vi.fn(), + waitForOutboundSequence: vi.fn(), state: { reset: vi.fn(), getSnapshot: vi.fn(), diff --git a/extensions/telegram/src/channel-message-flows.qa.e2e.test.ts b/extensions/telegram/src/channel-message-flows.qa.e2e.test.ts deleted file mode 100644 index 99e1f299b36..00000000000 --- a/extensions/telegram/src/channel-message-flows.qa.e2e.test.ts +++ /dev/null @@ -1,265 +0,0 @@ -// Channel Message Flows tests cover QA Lab channel delivery evidence. -import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { describe, expect, it, vi } from "vitest"; -import { - resolveTelegramFlowThreadSpec, - runTelegramThinkingFinalFlow, - runTelegramWorkingFinalFlow, -} from "./test-support/channel-message-flows.js"; - -describe("channel message flows QA e2e", () => { - function createTestDraftStream(params?: { - update?: (text: string) => void; - flush?: () => Promise; - clear?: () => Promise; - }) { - return { - update: vi.fn(params?.update ?? (() => {})), - updatePreview: vi.fn(), - flush: vi.fn(params?.flush ?? (async () => {})), - clear: vi.fn(params?.clear ?? (async () => {})), - stop: vi.fn(async () => {}), - messageId: vi.fn(() => 17), - forceNewMessage: vi.fn(), - finalizeToPreview: vi.fn(async () => undefined), - rotateToNewMessageDeferringDelete: vi.fn(() => undefined), - }; - } - - it("streams thinking updates, clears the preview, then sends the final answer", async () => { - const events: string[] = []; - const stream = { - update: vi.fn((text: string) => { - events.push(`update:${text}`); - }), - flush: vi.fn(async () => { - events.push("flush"); - }), - clear: vi.fn(async () => { - events.push("clear"); - }), - updatePreview: vi.fn(), - stop: vi.fn(async () => {}), - messageId: vi.fn(() => 17), - forceNewMessage: vi.fn(), - finalizeToPreview: vi.fn(async () => undefined), - rotateToNewMessageDeferringDelete: vi.fn(() => undefined), - }; - const sendFinal = vi.fn(async () => { - events.push("final"); - return { messageId: "99", chatId: "123" }; - }); - - const result = await runTelegramThinkingFinalFlow( - { - accountId: "sut", - cfg: {} as OpenClawConfig, - delayMs: 0, - target: "123", - threadId: 42, - thinkingUpdates: ["Checking the request.", "Reading the Telegram code.", "Ready."], - }, - { - createDraftStream: vi.fn(() => stream), - sendFinal, - sleep: vi.fn(async () => {}), - }, - ); - - expect(stream.update).toHaveBeenCalledTimes(3); - expect(stream.update.mock.calls[0]?.[0]).toContain("Thinking"); - expect(stream.update.mock.calls[0]?.[0]).toContain("_Checking the request._"); - expect(events.at(-2)).toBe("clear"); - expect(events.at(-1)).toBe("final"); - expect(sendFinal).toHaveBeenCalledWith({ - accountId: "sut", - cfg: {}, - target: "123", - text: "Final answer: the Telegram thinking preview cleared and this durable reply landed.", - threadId: 42, - }); - expect(result).toEqual({ finalMessageId: "99", previewUpdates: 3 }); - }); - - it("clears thinking previews when streaming fails before the final answer", async () => { - const stream = { - update: vi.fn(() => {}), - flush: vi.fn(async () => { - throw new Error("flush failed"); - }), - clear: vi.fn(async () => {}), - updatePreview: vi.fn(), - stop: vi.fn(async () => {}), - messageId: vi.fn(() => 17), - forceNewMessage: vi.fn(), - finalizeToPreview: vi.fn(async () => undefined), - rotateToNewMessageDeferringDelete: vi.fn(() => undefined), - }; - const sendFinal = vi.fn(async () => ({ messageId: "99", chatId: "123" })); - - await expect( - runTelegramThinkingFinalFlow( - { - cfg: {} as OpenClawConfig, - delayMs: 0, - target: "123", - thinkingUpdates: ["Checking the request."], - }, - { - createDraftStream: vi.fn(() => stream), - sendFinal, - sleep: vi.fn(async () => {}), - }, - ), - ).rejects.toThrow("flush failed"); - - expect(stream.clear).toHaveBeenCalledOnce(); - expect(sendFinal).not.toHaveBeenCalled(); - }); - - it("fails thinking-final when the final send does not return a message id", async () => { - const stream = { - update: vi.fn(() => {}), - flush: vi.fn(async () => {}), - clear: vi.fn(async () => {}), - updatePreview: vi.fn(), - stop: vi.fn(async () => {}), - messageId: vi.fn(() => 17), - forceNewMessage: vi.fn(), - finalizeToPreview: vi.fn(async () => undefined), - rotateToNewMessageDeferringDelete: vi.fn(() => undefined), - }; - - await expect( - runTelegramThinkingFinalFlow( - { - cfg: {} as OpenClawConfig, - delayMs: 0, - target: "123", - thinkingUpdates: ["Checking the request."], - }, - { - createDraftStream: vi.fn(() => stream), - sendFinal: vi.fn(async () => ({})), - sleep: vi.fn(async () => {}), - }, - ), - ).rejects.toThrow("thinking-final final send did not return a durable Telegram message id"); - }); - - it("streams working updates through rich message drafts before the final answer", async () => { - const stream = createTestDraftStream(); - const sendFinal = vi.fn(async () => ({ messageId: "100", chatId: "123" })); - - const result = await runTelegramWorkingFinalFlow( - { - cfg: {} as OpenClawConfig, - delayMs: 0, - durationMs: 12_000, - target: "123", - }, - { - createDraftStream: vi.fn(() => stream), - sendFinal, - sleep: vi.fn(async () => {}), - }, - ); - - expect(stream.update).toHaveBeenNthCalledWith(1, "Working"); - expect(stream.update.mock.calls[2]?.[0]).toContain("🛠️ pgrep -fl Discord || true (agent)"); - expect(stream.update.mock.calls[2]?.[0]).toContain( - "🛠️ list files in /Applications/Discord.app -> run true (agent)", - ); - expect(stream.update.mock.calls[4]?.[0]).toContain( - "• Discord is installed as a normal '/Applications/Discord.app'", - ); - expect(stream.update).toHaveBeenCalledWith( - expect.stringContaining("Working\n\n🛠️ pgrep -fl Discord || true (agent)"), - ); - expect(stream.clear).toHaveBeenCalledBefore(sendFinal); - expect(sendFinal).toHaveBeenCalledWith({ - accountId: undefined, - cfg: {}, - target: "123", - text: "Final answer: the Telegram working preview cleared and this durable reply landed.", - threadId: undefined, - }); - expect(stream.update).not.toHaveBeenCalledWith(expect.stringContaining("Working for")); - expect(result).toEqual({ finalMessageId: "100", previewUpdates: 6 }); - }); - - it("clears rich working drafts when progress updates fail before the final answer", async () => { - const stream = createTestDraftStream({ - update: () => { - throw new Error("draft update failed"); - }, - }); - const sendFinal = vi.fn(async () => ({ messageId: "100", chatId: "123" })); - - await expect( - runTelegramWorkingFinalFlow( - { - cfg: {} as OpenClawConfig, - delayMs: 0, - durationMs: 12_000, - target: "123", - }, - { - createDraftStream: vi.fn(() => stream), - sendFinal, - sleep: vi.fn(async () => {}), - }, - ), - ).rejects.toThrow("draft update failed"); - - expect(stream.clear).toHaveBeenCalledOnce(); - expect(sendFinal).not.toHaveBeenCalled(); - }); - - it("fails working-final when the final send does not return a message id", async () => { - const stream = createTestDraftStream(); - - await expect( - runTelegramWorkingFinalFlow( - { - cfg: {} as OpenClawConfig, - delayMs: 0, - durationMs: 12_000, - target: "123", - }, - { - createDraftStream: vi.fn(() => stream), - sendFinal: vi.fn(async () => ({})), - sleep: vi.fn(async () => {}), - }, - ), - ).rejects.toThrow("working-final final send did not return a durable Telegram message id"); - }); - - it("uses two second progress update cadence by default", async () => { - const stream = createTestDraftStream(); - const sleep = vi.fn(async () => {}); - - const result = await runTelegramWorkingFinalFlow( - { - cfg: {} as OpenClawConfig, - durationMs: 20_000, - target: "123", - }, - { - createDraftStream: vi.fn(() => stream), - sendFinal: vi.fn(async () => ({ messageId: "101", chatId: "123" })), - sleep, - }, - ); - - expect(sleep).toHaveBeenCalledTimes(9); - expect(sleep).toHaveBeenCalledWith(2_000); - expect(result.previewUpdates).toBe(7); - }); - - it("maps flow thread ids to Telegram forum topic specs", () => { - expect(resolveTelegramFlowThreadSpec(42)).toEqual({ id: 42, scope: "forum" }); - expect(resolveTelegramFlowThreadSpec()).toBeUndefined(); - }); -}); diff --git a/extensions/telegram/src/test-support/channel-message-flows.ts b/extensions/telegram/src/test-support/channel-message-flows.ts deleted file mode 100644 index a61417bf3be..00000000000 --- a/extensions/telegram/src/test-support/channel-message-flows.ts +++ /dev/null @@ -1,359 +0,0 @@ -// Channel Message Flows runtime supports QA Lab channel delivery evidence. -import { setTimeout as sleep } from "node:timers/promises"; -import type { Bot } from "grammy"; -import type { Message } from "grammy/types"; -import { formatReasoningMessage } from "openclaw/plugin-sdk/agent-runtime"; -import { formatChannelProgressDraftText } from "openclaw/plugin-sdk/channel-outbound"; -import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import type { TelegramThreadSpec } from "../bot/helpers.js"; -import { createTelegramDraftStream, type TelegramDraftStream } from "../draft-stream.js"; -import { - buildTelegramRichMarkdown, - type TelegramEditRichMessageTextParams, - type TelegramInputRichMessage, - type TelegramSendRichMessageParams, -} from "../rich-message.js"; -import { deleteMessageTelegram, editMessageTelegram, sendMessageTelegram } from "../send.js"; - -type TelegramApi = Bot["api"]; -type TelegramSendMessageParams = Parameters; -type TelegramEditMessageTextParams = Parameters; -type TelegramDeleteMessageParams = Parameters; - -type SupportedFlow = "thinking-final" | "working-final"; - -type TelegramSendFinalParams = { - accountId?: string; - cfg: OpenClawConfig; - target: string; - text: string; - threadId?: number; -}; - -type TelegramFlowResult = { - finalMessageId?: string; - previewUpdates: number; -}; - -type TelegramFlowDeps = { - createDraftStream?: (params: { - accountId?: string; - cfg: OpenClawConfig; - target: string; - threadId?: number; - }) => TelegramDraftStream; - sendFinal?: (params: TelegramSendFinalParams) => Promise<{ messageId?: string }>; - sleep?: (ms: number) => Promise; -}; - -export type TelegramThinkingFinalFlowOptions = { - accountId?: string; - cfg: OpenClawConfig; - delayMs?: number; - finalText?: string; - target: string; - threadId?: number; - thinkingUpdates?: readonly string[]; -}; - -export type TelegramWorkingFinalFlowOptions = TelegramThinkingFinalFlowOptions & { - durationMs?: number; -}; - -const DEFAULT_THINKING_FINAL_UPDATES = [ - "I'll inspect the Telegram stream surface first.", - "I found the reasoning preview path and I’m checking final delivery.", - "The preview should clear before the durable final answer lands.", -] as const; - -const DEFAULT_THINKING_FINAL_TEXT = - "Final answer: the Telegram thinking preview cleared and this durable reply landed."; -const DEFAULT_WORKING_FINAL_TEXT = - "Final answer: the Telegram working preview cleared and this durable reply landed."; -const DEFAULT_WORKING_PROGRESS_TIMELINE = [ - { - atMs: 2_000, - line: "🛠️ pgrep -fl Discord || true (agent)", - }, - { - atMs: 5_000, - line: "🛠️ list files in /Applications/Discord.app -> run true (agent)", - }, - { - atMs: 7_000, - line: "🛠️ sw_vers (agent)", - }, - { - atMs: 8_000, - line: "Discord is installed as a normal '/Applications/Discord.app', not as a Homebrew-managed cask, and it's currently running.", - }, - { - atMs: 11_000, - line: "🛠️ osascript -e 'tell application \"Discord\" to quit' || true sleep 3 pgrep -fl Discord || true (agent)", - }, - { - atMs: 14_000, - line: "🛠️ brew install --cask --force discord (agent)", - }, - { - atMs: 17_000, - line: "Homebrew found Discord as an outdated cask after updating its metadata, so this is doing a real cask reinstall.", - }, -] as const; - -function toError(value: unknown): Error { - return value instanceof Error ? value : new Error(String(value)); -} - -function requireFinalMessageId(final: { messageId?: string }, flow: SupportedFlow): string { - const messageId = final.messageId?.trim(); - if (!messageId) { - throw new Error(`${flow} final send did not return a durable Telegram message id`); - } - return messageId; -} - -function resolveWorkingProgressLines(elapsedMs: number): string[] { - return DEFAULT_WORKING_PROGRESS_TIMELINE.filter((entry) => entry.atMs <= elapsedMs).map( - (entry) => entry.line, - ); -} - -function formatWorkingProgressPreview(elapsedMs: number): string { - return formatChannelProgressDraftText({ - entry: { streaming: { progress: { label: "Working", toolProgress: false } } }, - lines: resolveWorkingProgressLines(elapsedMs), - }); -} - -function richMessageText(richMessage: TelegramInputRichMessage): { - text: string; - textMode: "markdown" | "html"; -} { - return richMessage.html !== undefined - ? { text: richMessage.html, textMode: "html" } - : { text: richMessage.markdown, textMode: "markdown" }; -} - -function createTelegramFlowApi(params: { accountId?: string; cfg: OpenClawConfig }): Bot["api"] { - const api = { - raw: { - sendRichMessage: async (sendParams: TelegramSendRichMessageParams) => { - const richText = richMessageText(sendParams.rich_message); - const result = await sendMessageTelegram(String(sendParams.chat_id), richText.text, { - accountId: params.accountId, - cfg: params.cfg, - messageThreadId: sendParams.message_thread_id, - textMode: richText.textMode, - }); - return { message_id: Number(result.messageId) } as Message; - }, - editMessageText: async (editParams: TelegramEditRichMessageTextParams) => { - if (typeof editParams.message_id !== "number") { - throw new Error("Telegram flow rich edit requires message_id."); - } - const richText = richMessageText(editParams.rich_message); - await editMessageTelegram( - String(editParams.chat_id), - editParams.message_id, - richText.text, - { - accountId: params.accountId, - cfg: params.cfg, - textMode: richText.textMode, - }, - ); - return true; - }, - }, - sendMessage: async ( - chatId: TelegramSendMessageParams[0], - text: TelegramSendMessageParams[1], - sendParams: TelegramSendMessageParams[2], - ) => { - const result = await sendMessageTelegram(String(chatId), text, { - accountId: params.accountId, - cfg: params.cfg, - messageThreadId: sendParams?.message_thread_id, - textMode: sendParams?.parse_mode === "HTML" ? "html" : "markdown", - }); - return { message_id: Number(result.messageId) }; - }, - editMessageText: async ( - chatId: TelegramEditMessageTextParams[0], - messageId: TelegramEditMessageTextParams[1], - text: TelegramEditMessageTextParams[2], - editParams: TelegramEditMessageTextParams[3], - ) => { - await editMessageTelegram(String(chatId), messageId, text, { - accountId: params.accountId, - cfg: params.cfg, - textMode: editParams?.parse_mode === "HTML" ? "html" : "markdown", - }); - return true; - }, - deleteMessage: async ( - chatId: TelegramDeleteMessageParams[0], - messageId: TelegramDeleteMessageParams[1], - ) => { - await deleteMessageTelegram(String(chatId), messageId, { - accountId: params.accountId, - cfg: params.cfg, - }); - return true; - }, - }; - return api as unknown as Bot["api"]; -} - -export function resolveTelegramFlowThreadSpec(threadId?: number): TelegramThreadSpec | undefined { - return typeof threadId === "number" ? { id: threadId, scope: "forum" } : undefined; -} - -function createDefaultTelegramDraftStream(params: { - accountId?: string; - cfg: OpenClawConfig; - target: string; - threadId?: number; -}): TelegramDraftStream { - return createTelegramDraftStream({ - api: createTelegramFlowApi(params), - chatId: params.target, - minInitialChars: 0, - renderText: (text) => ({ text, richMessage: buildTelegramRichMarkdown(text) }), - thread: resolveTelegramFlowThreadSpec(params.threadId), - throttleMs: 250, - }); -} - -async function sendTelegramFinal(params: TelegramSendFinalParams): Promise<{ messageId?: string }> { - return await sendMessageTelegram(params.target, params.text, { - accountId: params.accountId, - cfg: params.cfg, - messageThreadId: params.threadId, - }); -} - -export async function runTelegramThinkingFinalFlow( - options: TelegramThinkingFinalFlowOptions, - deps: TelegramFlowDeps = {}, -): Promise { - const delayMs = options.delayMs ?? 900; - const thinkingUpdates = options.thinkingUpdates ?? DEFAULT_THINKING_FINAL_UPDATES; - const stream = (deps.createDraftStream ?? createDefaultTelegramDraftStream)({ - accountId: options.accountId, - cfg: options.cfg, - target: options.target, - threadId: options.threadId, - }); - const wait = deps.sleep ?? sleep; - - let previewStarted = false; - let flowError: unknown; - try { - for (const update of thinkingUpdates) { - previewStarted = true; - stream.update(formatReasoningMessage(update)); - await stream.flush(); - if (delayMs > 0) { - await wait(delayMs); - } - } - } catch (error) { - flowError = error; - } - let cleanupError: unknown; - if (previewStarted) { - try { - await stream.clear(); - } catch (error) { - cleanupError = error; - } - } - if (flowError) { - throw toError(flowError); - } - if (cleanupError) { - throw toError(cleanupError); - } - - const final = await (deps.sendFinal ?? sendTelegramFinal)({ - accountId: options.accountId, - cfg: options.cfg, - target: options.target, - text: options.finalText ?? DEFAULT_THINKING_FINAL_TEXT, - threadId: options.threadId, - }); - - const finalMessageId = requireFinalMessageId(final, "thinking-final"); - return { - finalMessageId, - previewUpdates: thinkingUpdates.length, - }; -} - -export async function runTelegramWorkingFinalFlow( - options: TelegramWorkingFinalFlowOptions, - deps: TelegramFlowDeps = {}, -): Promise { - const delayMs = options.delayMs ?? 2_000; - const durationMs = options.durationMs ?? 12_000; - const stream = (deps.createDraftStream ?? createDefaultTelegramDraftStream)({ - accountId: options.accountId, - cfg: options.cfg, - target: options.target, - threadId: options.threadId, - }); - const wait = deps.sleep ?? sleep; - - let previewUpdates = 0; - let lastPreviewText = ""; - const updateIntervalMs = delayMs > 0 ? delayMs : 1_000; - let draftStarted = false; - let flowError: unknown; - try { - for (let elapsedMs = 0; elapsedMs < durationMs; elapsedMs += updateIntervalMs) { - const previewText = formatWorkingProgressPreview(elapsedMs); - if (previewText !== lastPreviewText) { - draftStarted = true; - stream.update(previewText); - await stream.flush(); - lastPreviewText = previewText; - previewUpdates += 1; - } - if (delayMs > 0 && elapsedMs + updateIntervalMs < durationMs) { - await wait(delayMs); - } - } - } catch (error) { - flowError = error; - } - let cleanupError: unknown; - if (draftStarted) { - try { - await stream.clear(); - } catch (error) { - cleanupError = error; - } - } - if (flowError) { - throw toError(flowError); - } - if (cleanupError) { - throw toError(cleanupError); - } - - const final = await (deps.sendFinal ?? sendTelegramFinal)({ - accountId: options.accountId, - cfg: options.cfg, - target: options.target, - text: options.finalText ?? DEFAULT_WORKING_FINAL_TEXT, - threadId: options.threadId, - }); - - const finalMessageId = requireFinalMessageId(final, "working-final"); - return { - finalMessageId, - previewUpdates, - }; -} diff --git a/qa/scenarios/channels/channel-message-flows.yaml b/qa/scenarios/channels/channel-message-flows.yaml index 8e8a2721d04..40e2b21525f 100644 --- a/qa/scenarios/channels/channel-message-flows.yaml +++ b/qa/scenarios/channels/channel-message-flows.yaml @@ -1,4 +1,4 @@ -title: Channel message flow evidence +title: Channel streaming message flow scenario: id: channel-message-flows @@ -7,21 +7,75 @@ scenario: primary: - channels.streaming secondary: - - channels.threads - runtime.delivery - - runtime.reasoning-visibility - objective: Exercise Telegram draft/final delivery sequencing through QA Lab evidence. + objective: Verify streaming channel replies produce visible previews that resolve to one final answer. + gatewayConfigPatch: + channels: + telegram: + streaming: + mode: partial successCriteria: - - Thinking updates flush, clear, and send the final answer in order. - - Working previews are cleared before final delivery. - - Thread IDs are preserved in Telegram flow thread specs. + - The selected transport exposes at least one preview event before final delivery. + - The final answer replaces or follows the preview without losing the requested text. docsRefs: + - docs/channels/qa-channel.md - docs/channels/telegram.md - docs/concepts/qa-e2e-automation.md codeRefs: - - extensions/telegram/src/channel-message-flows.qa.e2e.test.ts - - extensions/telegram/src/test-support/channel-message-flows.ts + - extensions/qa-channel/src/inbound.ts + - extensions/qa-lab/src/crabline-transport.ts + - extensions/qa-lab/src/providers/mock-openai/server.ts + - extensions/qa-lab/src/qa-transport.ts + - extensions/telegram/src/draft-stream.ts execution: - kind: vitest - path: extensions/telegram/src/channel-message-flows.qa.e2e.test.ts - summary: Vitest coverage for channel message flow sequencing. + kind: flow + channel: telegram + summary: Stream a deterministic answer through QA Channel or Crabline Telegram and assert its preview lifecycle. + config: + requiredProviderMode: mock-openai + conversationId: "-1001234567890" + senderId: "100001" + finalMarker: QA-CHANNEL-STREAMING-PREVIEW-FINAL-OK-1234567890 + prompt: "Final-only marker streaming QA check. Reply exactly: QA-CHANNEL-STREAMING-PREVIEW-FINAL-OK-1234567890" + +flow: + steps: + - name: streams a preview into one final reply + actions: + - assert: + expr: env.providerMode === config.requiredProviderMode + message: this deterministic streaming proof requires mock-openai + - call: waitForGatewayHealthy + args: + - ref: env + - 60000 + - call: waitForTransportReady + args: + - ref: env + - 60000 + - resetTransport: true + - sendInbound: + conversation: + id: + ref: config.conversationId + kind: group + senderId: + ref: config.senderId + senderName: QA Streaming Operator + text: + ref: config.prompt + - waitForOutboundSequence: + conversationId: + ref: config.conversationId + finalTextIncludes: + ref: config.finalMarker + finalSettleMs: 500 + minimumPreviewEvents: 1 + timeoutMs: + expr: liveTurnTimeoutMs(env, 45000) + saveAs: sequence + - assert: + expr: sequence.events.length >= 2 + message: + expr: "`expected a preview followed by the final marker; events=${JSON.stringify(sequence.events)}`" + detailsExpr: "`${sequence.events.map((event) => event.kind).join(' -> ')}: ${sequence.final.text}`"