mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-09 15:59:30 +00:00
test(qa): migrate channel streaming evidence to transport flow (#99310)
* test(qa): migrate channel streaming evidence to transport flow * test(qa): enable Telegram previews for streaming smoke * test(qa): stabilize streaming preview evidence * fix(qa): sanitize channel dispatch logs
This commit is contained in:
Dallin Romney
GitHub
parent
df350e6720
commit
8604dbdc93
19 changed files with 864 additions and 669 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -115,7 +115,9 @@ function createMockQaRuntime(params?: {
|
|||
replyOptions,
|
||||
}: {
|
||||
ctx: { BodyForAgent?: string; Body?: string };
|
||||
dispatcherOptions: { deliver: (payload: { text: string }) => Promise<void> };
|
||||
dispatcherOptions: {
|
||||
deliver: (payload: { text: string }, info: { kind: string }) => Promise<void>;
|
||||
};
|
||||
replyOptions?: {
|
||||
onToolStart?: (payload: {
|
||||
name?: string;
|
||||
|
|
@ -128,9 +130,12 @@ function createMockQaRuntime(params?: {
|
|||
await replyOptions?.onToolStart?.(toolStart);
|
||||
}
|
||||
params?.onDispatch?.(ctx as Record<string, unknown>);
|
||||
await dispatcherOptions.deliver({
|
||||
text: `qa-echo: ${ctx.BodyForAgent ?? ctx.Body ?? ""}`,
|
||||
});
|
||||
await dispatcherOptions.deliver(
|
||||
{
|
||||
text: `qa-echo: ${ctx.BodyForAgent ?? ctx.Body ?? ""}`,
|
||||
},
|
||||
{ kind: "final" },
|
||||
);
|
||||
},
|
||||
},
|
||||
inbound: {
|
||||
|
|
|
|||
|
|
@ -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<typeof import("./bus-client.js")>();
|
||||
return {
|
||||
...actual,
|
||||
deleteQaBusMessage: vi.fn(async () => ({ message: {} })),
|
||||
editQaBusMessage: vi.fn(async () => ({ message: {} })),
|
||||
sendQaBusMessage: vi.fn(async () => ({ message: { id: "preview-1" } })),
|
||||
};
|
||||
});
|
||||
|
||||
type HandleQaInboundParams = Parameters<typeof handleQaInbound>[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]);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>) => {
|
||||
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({
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
getOutboundEvents: () => Promise<readonly QaTransportOutboundEvent[]>;
|
||||
rememberProviderTarget: (providerTargetKey: string, qaTarget: string) => void;
|
||||
};
|
||||
|
||||
const TELEGRAM_LIFECYCLE_METHOD_RE = /\/(sendMessage|editMessageText|deleteMessage)$/u;
|
||||
|
||||
function readTelegramLifecycleEvent(params: {
|
||||
cursor: number;
|
||||
event: unknown;
|
||||
messageByProviderId: Map<string, QaBusMessage>;
|
||||
pendingByChat: Map<string, QaBusMessage[]>;
|
||||
}): 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<string, string>();
|
||||
const telegramMessageByProviderId = new Map<string, QaBusMessage>();
|
||||
const pendingTelegramMessagesByChat = new Map<string, QaBusMessage[]>();
|
||||
const outboundEvents: QaTransportOutboundEvent[] = [];
|
||||
let recorderLineCursor = 0;
|
||||
let syncPromise: Promise<void> | 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<string, unknown>;
|
||||
|
|
|
|||
|
|
@ -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: {
|
||||
|
|
|
|||
|
|
@ -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<string, unknown> & {
|
||||
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") {
|
||||
|
|
|
|||
103
extensions/qa-lab/src/qa-transport.test.ts
Normal file
103
extensions/qa-lab/src/qa-transport.test.ts
Normal file
|
|
@ -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");
|
||||
});
|
||||
});
|
||||
|
|
@ -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<void>;
|
||||
waitForNoOutbound: (input?: QaTransportWaitForNoOutboundInput) => Promise<void>;
|
||||
waitForOutbound: (input: QaTransportOutboundMatch) => Promise<QaBusMessage>;
|
||||
waitForOutboundSequence: (
|
||||
input: QaTransportOutboundSequenceMatch,
|
||||
) => Promise<QaTransportOutboundSequence>;
|
||||
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<readonly (QaBusEvent | QaTransportOutboundEvent)[]>;
|
||||
}): Promise<QaTransportOutboundSequence> {
|
||||
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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -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: {},
|
||||
|
|
|
|||
|
|
@ -165,6 +165,7 @@ async function runFlowAction(action: unknown, api: QaFlowApi, vars: QaFlowVars)
|
|||
"sendInbound",
|
||||
"sendNativeCommand",
|
||||
"waitForOutbound",
|
||||
"waitForOutboundSequence",
|
||||
"waitForNoOutbound",
|
||||
] as const) {
|
||||
if (name in action) {
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
clear?: () => Promise<void>;
|
||||
}) {
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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<TelegramApi["sendMessage"]>;
|
||||
type TelegramEditMessageTextParams = Parameters<TelegramApi["editMessageText"]>;
|
||||
type TelegramDeleteMessageParams = Parameters<TelegramApi["deleteMessage"]>;
|
||||
|
||||
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<void>;
|
||||
};
|
||||
|
||||
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<TelegramFlowResult> {
|
||||
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<TelegramFlowResult> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
|
@ -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}`"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue