diff --git a/extensions/qa-channel/src/gateway.test.ts b/extensions/qa-channel/src/gateway.test.ts index 1366b91a2bd..56d67fba1fa 100644 --- a/extensions/qa-channel/src/gateway.test.ts +++ b/extensions/qa-channel/src/gateway.test.ts @@ -2,9 +2,14 @@ import { createServer } from "node:http"; import { afterEach, describe, expect, it, vi } from "vitest"; import { startQaGatewayAccount } from "./gateway.js"; +import { handleQaInbound } from "./inbound.js"; import type { ChannelGatewayContext } from "./runtime-api.js"; import type { ResolvedQaChannelAccount } from "./types.js"; +vi.mock("./inbound.js", () => ({ + handleQaInbound: vi.fn(async () => undefined), +})); + async function startJsonServer( handler: (req: { url?: string | undefined }) => { statusCode?: number; body: string }, ) { @@ -40,9 +45,92 @@ describe("qa-channel gateway", () => { const stops: Array<() => Promise> = []; afterEach(async () => { + vi.mocked(handleQaInbound).mockReset().mockResolvedValue(undefined); await Promise.all(stops.splice(0).map((stop) => stop())); }); + it("lets native commands bypass the ordered inbound queue", async () => { + const controller = new AbortController(); + const message = { + id: "msg-1", + accountId: "default", + direction: "inbound" as const, + conversation: { id: "alice", kind: "direct" as const }, + senderId: "alice", + text: "hello", + timestamp: Date.now(), + reactions: [], + }; + const server = await startJsonServer(() => ({ + body: JSON.stringify({ + cursor: 2, + events: [ + { cursor: 1, kind: "inbound-message", accountId: "default", message }, + { + cursor: 2, + kind: "inbound-message", + accountId: "default", + message: { ...message, id: "msg-2", text: "follow-up" }, + }, + { + cursor: 3, + kind: "inbound-message", + accountId: "default", + message: { + ...message, + id: "msg-3", + text: "/stop", + nativeCommand: { name: "stop" }, + }, + }, + ], + }), + })); + stops.push(() => server.stop()); + let releaseFirst: (() => void) | undefined; + const firstPending = new Promise((resolve) => { + releaseFirst = resolve; + }); + vi.mocked(handleQaInbound).mockImplementation(async ({ message: inbound }) => { + if (inbound.text === "hello") { + await firstPending; + } + if (inbound.text === "/stop") { + controller.abort(); + } + }); + const account: ResolvedQaChannelAccount = { + accountId: "default", + baseUrl: server.baseUrl, + botDisplayName: "QA Bot", + botUserId: "qa-bot", + config: {}, + configured: true, + enabled: true, + pollTimeoutMs: 1, + }; + + const gateway = startQaGatewayAccount("qa-channel", "QA Channel", { + abortSignal: controller.signal, + account, + cfg: {}, + setStatus: vi.fn(), + } as unknown as ChannelGatewayContext); + + await vi.waitFor(() => { + const handled = vi.mocked(handleQaInbound).mock.calls.map(([params]) => params.message.text); + expect(handled).toContain("hello"); + expect(handled).toContain("/stop"); + expect(handled).not.toContain("follow-up"); + }); + releaseFirst?.(); + await gateway; + const handled = vi.mocked(handleQaInbound).mock.calls.map(([params]) => params.message.text); + expect(handled).toHaveLength(3); + expect(handled).toContain("/stop"); + expect(handled.indexOf("hello")).toBeLessThan(handled.indexOf("follow-up")); + }); + it("clears running status when polling fails", async () => { const server = await startJsonServer(() => ({ statusCode: 500, @@ -84,4 +172,57 @@ describe("qa-channel gateway", () => { }, ]); }); + + it("stops the ordered inbound queue after the first dispatch failure", async () => { + const controller = new AbortController(); + const message = { + id: "msg-1", + accountId: "default", + direction: "inbound" as const, + conversation: { id: "alice", kind: "direct" as const }, + senderId: "alice", + text: "first", + timestamp: Date.now(), + reactions: [], + }; + const server = await startJsonServer(() => ({ + body: JSON.stringify({ + cursor: 2, + events: [ + { cursor: 1, kind: "inbound-message", accountId: "default", message }, + { + cursor: 2, + kind: "inbound-message", + accountId: "default", + message: { ...message, id: "msg-2", text: "second" }, + }, + ], + }), + })); + stops.push(() => server.stop()); + vi.mocked(handleQaInbound).mockImplementationOnce(async () => { + controller.abort(); + throw new Error("inbound failed"); + }); + const account: ResolvedQaChannelAccount = { + accountId: "default", + baseUrl: server.baseUrl, + botDisplayName: "QA Bot", + botUserId: "qa-bot", + config: {}, + configured: true, + enabled: true, + pollTimeoutMs: 1, + }; + + await expect( + startQaGatewayAccount("qa-channel", "QA Channel", { + abortSignal: controller.signal, + account, + cfg: {}, + setStatus: vi.fn(), + } as unknown as ChannelGatewayContext), + ).rejects.toThrow("inbound failed"); + expect(handleQaInbound).toHaveBeenCalledTimes(1); + }); }); diff --git a/extensions/qa-channel/src/gateway.ts b/extensions/qa-channel/src/gateway.ts index dd311858bb3..98975baac44 100644 --- a/extensions/qa-channel/src/gateway.ts +++ b/extensions/qa-channel/src/gateway.ts @@ -21,8 +21,36 @@ export async function startQaGatewayAccount( baseUrl: account.baseUrl, }); let cursor = 0; + let inboundError: Error | undefined; + let queuedInbound = Promise.resolve(); + const controlTasks = new Set>(); + const handleMessage = (message: Parameters[0]["message"]) => + handleQaInbound({ + channelId, + channelLabel, + account, + config: ctx.cfg as CoreConfig, + message, + }); + const captureInboundError = (error: unknown) => { + inboundError ??= error instanceof Error ? error : new Error(String(error)); + }; + const dispatchControl = (message: Parameters[0]["message"]) => { + const task = handleMessage(message) + .catch(captureInboundError) + .finally(() => controlTasks.delete(task)); + controlTasks.add(task); + }; + const enqueueInbound = (message: Parameters[0]["message"]) => { + queuedInbound = queuedInbound + .then(() => (inboundError ? undefined : handleMessage(message))) + .catch(captureInboundError); + }; try { while (!ctx.abortSignal.aborted) { + if (inboundError) { + throw inboundError; + } const result = await pollQaBus({ baseUrl: account.baseUrl, accountId: account.accountId, @@ -35,23 +63,28 @@ export async function startQaGatewayAccount( if (event.kind !== "inbound-message") { continue; } - await handleQaInbound({ - channelId, - channelLabel, - account, - config: ctx.cfg as CoreConfig, - message: event.message, - }); + if (event.message.nativeCommand) { + dispatchControl(event.message); + } else { + enqueueInbound(event.message); + } } } + if (inboundError) { + throw inboundError; + } } catch (error) { if (!(error instanceof Error) || error.name !== "AbortError") { throw error; } } finally { + await Promise.all([queuedInbound, ...controlTasks]); ctx.setStatus({ accountId: account.accountId, running: false, }); } + if (inboundError) { + throw inboundError; + } } diff --git a/extensions/qa-channel/src/inbound.test.ts b/extensions/qa-channel/src/inbound.test.ts index 83ae38b0d6c..c116d092f6f 100644 --- a/extensions/qa-channel/src/inbound.test.ts +++ b/extensions/qa-channel/src/inbound.test.ts @@ -125,6 +125,33 @@ describe("handleQaInbound", () => { expect(ctxPayload?.SenderId).toBe("alice"); }); + it("routes native commands through a separate slash session to the conversation session", async () => { + const runtime = createPluginRuntimeMock(); + setQaChannelRuntime(runtime); + + await handleQaInbound( + createQaInboundParams({ + message: { + text: "/stop", + nativeCommand: { name: "stop" }, + }, + }), + ); + + const assembled = firstRunAssembledParams(runtime); + expect(assembled.ctxPayload).toMatchObject({ + CommandAuthorized: true, + CommandSource: "native", + CommandTargetSessionKey: assembled.routeSessionKey, + CommandTurn: { + body: "/stop", + source: "native", + }, + }); + expect(assembled.ctxPayload.SessionKey).toContain("qa-channel:slash:alice"); + expect(assembled.ctxPayload.SessionKey).not.toBe(assembled.routeSessionKey); + }); + it("skips malformed inline attachment base64 without dropping the message", async () => { const runtime = createPluginRuntimeMock(); setQaChannelRuntime(runtime); diff --git a/extensions/qa-channel/src/inbound.ts b/extensions/qa-channel/src/inbound.ts index 41093403ce6..36364d116cf 100644 --- a/extensions/qa-channel/src/inbound.ts +++ b/extensions/qa-channel/src/inbound.ts @@ -1,5 +1,6 @@ // Qa Channel plugin module implements inbound behavior. 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 { resolveInboundRouteEnvelopeBuilderWithRuntime } from "openclaw/plugin-sdk/inbound-envelope"; import { @@ -179,15 +180,26 @@ export async function handleQaInbound(params: { body: inbound.text, }); const mediaPayload = await resolveQaInboundMediaPayload(inbound.attachments); + const nativeCommand = inbound.nativeCommand; + const commandTargets = nativeCommand + ? resolveNativeCommandSessionTargets({ + agentId: route.agentId, + sessionPrefix: "qa-channel:slash", + userId: inbound.senderId, + targetSessionKey: route.sessionKey, + }) + : undefined; + const commandBody = nativeCommand ? `/${nativeCommand.name}` : inbound.text; const ctxPayload = runtime.channel.reply.finalizeInboundContext({ Body: body, BodyForAgent: inbound.text, RawBody: inbound.text, - CommandBody: inbound.text, + CommandBody: commandBody, From: target, To: target, - SessionKey: route.sessionKey, + SessionKey: commandTargets?.sessionKey ?? route.sessionKey, + CommandTargetSessionKey: commandTargets?.commandTargetSessionKey, AccountId: route.accountId ?? params.account.accountId, ChatType: inbound.conversation.kind === "direct" ? "direct" : "group", WasMentioned: wasMentioned, @@ -215,6 +227,15 @@ export async function handleQaInbound(params: { OriginatingChannel: params.channelId, OriginatingTo: target, CommandAuthorized: true, + CommandSource: nativeCommand ? "native" : undefined, + CommandTurn: nativeCommand + ? { + kind: "native", + source: "native", + authorized: true, + body: commandBody, + } + : undefined, ...mediaPayload, }); diff --git a/extensions/qa-lab/package.json b/extensions/qa-lab/package.json index fee315e074f..6bccd4088e2 100644 --- a/extensions/qa-lab/package.json +++ b/extensions/qa-lab/package.json @@ -16,7 +16,7 @@ "@openclaw/plugin-sdk": "workspace:*", "@openclaw/slack": "workspace:*", "@openclaw/whatsapp": "workspace:*", - "@openclaw/crabline": "0.1.6", + "@openclaw/crabline": "0.1.7", "openclaw": "workspace:*" }, "peerDependencies": { diff --git a/extensions/qa-lab/src/bus-queries.ts b/extensions/qa-lab/src/bus-queries.ts index 3181ac140cc..0e6fd1326c6 100644 --- a/extensions/qa-lab/src/bus-queries.ts +++ b/extensions/qa-lab/src/bus-queries.ts @@ -61,6 +61,7 @@ export function cloneMessage(message: QaBusMessage): QaBusMessage { ...message, conversation: { ...message.conversation }, attachments: (message.attachments ?? []).map((attachment) => cloneAttachment(attachment)), + ...(message.nativeCommand ? { nativeCommand: { ...message.nativeCommand } } : {}), toolCalls: message.toolCalls?.map((toolCall) => cloneToolCall(toolCall)), reactions: message.reactions.map((reaction) => ({ ...reaction })), }; diff --git a/extensions/qa-lab/src/bus-state.ts b/extensions/qa-lab/src/bus-state.ts index e1a17019843..156c5c6105c 100644 --- a/extensions/qa-lab/src/bus-state.ts +++ b/extensions/qa-lab/src/bus-state.ts @@ -118,6 +118,7 @@ export function createQaBusState() { threadTitle?: string; replyToId?: string; attachments?: QaBusAttachment[]; + nativeCommand?: QaBusInboundMessageInput["nativeCommand"]; toolCalls?: QaBusToolCall[]; }): QaBusMessage => { const conversation = ensureConversation(params.conversation); @@ -135,6 +136,7 @@ export function createQaBusState() { threadTitle: params.threadTitle, replyToId: params.replyToId, attachments: params.attachments?.map((attachment) => ({ ...attachment })) ?? [], + ...(params.nativeCommand ? { nativeCommand: { ...params.nativeCommand } } : {}), ...(toolCalls ? { toolCalls } : {}), reactions: [], }; @@ -175,6 +177,7 @@ export function createQaBusState() { threadTitle: input.threadTitle, replyToId: input.replyToId, attachments: input.attachments, + nativeCommand: input.nativeCommand, toolCalls: input.toolCalls, }); pushEvent({ diff --git a/extensions/qa-lab/src/crabline-transport.test.ts b/extensions/qa-lab/src/crabline-transport.test.ts index b07072e5916..7dd2fb8864d 100644 --- a/extensions/qa-lab/src/crabline-transport.test.ts +++ b/extensions/qa-lab/src/crabline-transport.test.ts @@ -59,6 +59,47 @@ describe("crabline transport", () => { }); }); + it("injects Telegram native commands through the shared transport adapter", async () => { + await withTempDir("qa-crabline-transport-", async (outputDir) => { + const transport = await createQaCrablineTransportAdapter({ + outputDir, + selection: createSelection(), + state: createQaBusState(), + }); + + try { + await transport.sendNativeCommand({ + command: "stop", + conversation: { id: "alice", kind: "direct" }, + senderId: "alice", + senderName: "Alice", + }); + + const manifest = JSON.parse( + await fs.readFile(path.join(outputDir, OPENCLAW_CRABLINE_MANIFEST_PATH), "utf8"), + ) as { + botToken: string; + endpoints: { apiRoot: string }; + }; + const response = await fetch( + `${manifest.endpoints.apiRoot}/bot${manifest.botToken}/getUpdates`, + ); + await expect(response.json()).resolves.toMatchObject({ + result: [ + { + message: { + entities: [{ length: 5, offset: 0, type: "bot_command" }], + text: "/stop", + }, + }, + ], + }); + } 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 cf72d46033a..15385b66dc0 100644 --- a/extensions/qa-lab/src/crabline-transport.ts +++ b/extensions/qa-lab/src/crabline-transport.ts @@ -19,6 +19,7 @@ import type { QaTransportActionName, QaTransportGatewayClient, QaTransportGatewayConfig, + QaTransportNativeCommandInput, QaTransportReportParams, QaTransportState, } from "./qa-transport.js"; @@ -260,6 +261,20 @@ class QaCrablineTransport extends QaStateBackedTransportAdapter { createRuntimeEnvPatch = () => this.#adapter.createChannelDriverSmokeEnv({}); + override async sendNativeCommand(input: QaTransportNativeCommandInput): Promise { + if (this.#selection.channel !== "telegram") { + throw new Error( + `Crabline ${this.#selection.channel} does not support native command injection.`, + ); + } + const { command, ...message } = input; + await this.sendInbound({ + ...message, + text: `/${command}`, + nativeCommand: { name: command }, + }); + } + handleAction = async (_params: { action: QaTransportActionName; args: Record; diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index be7a941e51e..a4136faf22f 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -3612,6 +3612,8 @@ export async function startQaMockOpenAiServer(params?: { host?: string; port?: n }; let lastRequest: MockOpenAiRequestSnapshot | null = null; const requests: MockOpenAiRequestSnapshot[] = []; + const inflightRequests = new Map(); + let nextInflightRequestId = 1; const imageGenerationRequests: Array> = []; const server = createServer((req, res) => { void (async () => { @@ -3642,6 +3644,10 @@ export async function startQaMockOpenAiServer(params?: { host?: string; port?: n writeJson(res, 200, requests); return; } + if (req.method === "GET" && url.pathname === "/debug/inflight-requests") { + writeJson(res, 200, [...inflightRequests.values()]); + return; + } if (req.method === "GET" && url.pathname === "/debug/image-generations") { writeJson(res, 200, imageGenerationRequests); return; @@ -3708,13 +3714,22 @@ export async function startQaMockOpenAiServer(params?: { host?: string; port?: n return; } const input = Array.isArray(body.input) ? (body.input as ResponsesInputItem[]) : []; - const events = await buildResponsesPayload(body, scenarioState); + const prompt = extractLastUserText(input); + const allInputText = extractAllRequestTexts(input, body); + const inflightRequestId = nextInflightRequestId++; + inflightRequests.set(inflightRequestId, { prompt, allInputText }); + let events: StreamEvent[]; + try { + events = await buildResponsesPayload(body, scenarioState); + } finally { + inflightRequests.delete(inflightRequestId); + } const resolvedModel = typeof body.model === "string" ? body.model : ""; lastRequest = { raw, body, - prompt: extractLastUserText(input), - allInputText: extractAllRequestTexts(input, body), + prompt, + allInputText, instructions: extractInstructionsText(body) || undefined, toolOutput: extractToolOutput(input), model: resolvedModel, diff --git a/extensions/qa-lab/src/qa-channel-transport.test.ts b/extensions/qa-lab/src/qa-channel-transport.test.ts index e17d95279f1..0ddb4aaa549 100644 --- a/extensions/qa-lab/src/qa-channel-transport.test.ts +++ b/extensions/qa-lab/src/qa-channel-transport.test.ts @@ -140,6 +140,22 @@ describe("qa channel transport", () => { expect(transport.state.getSnapshot().messages).toEqual([]); }); + it("injects native commands with transport metadata", async () => { + const transport = createQaChannelTransport(createQaBusState()); + + await transport.sendNativeCommand({ + command: "stop", + conversation: { id: "alice", kind: "direct" }, + senderId: "alice", + }); + + const [message] = transport.state.getSnapshot().messages; + expect(message).toMatchObject({ + text: "/stop", + nativeCommand: { name: "stop" }, + }); + }); + it("inherits the shared failure-aware wait helper", async () => { const transport = createQaChannelTransport(createQaBusState()); let injected = false; diff --git a/extensions/qa-lab/src/qa-channel-transport.ts b/extensions/qa-lab/src/qa-channel-transport.ts index 2f499422851..cae2ae18ff9 100644 --- a/extensions/qa-lab/src/qa-channel-transport.ts +++ b/extensions/qa-lab/src/qa-channel-transport.ts @@ -10,6 +10,7 @@ import type { QaTransportActionName, QaTransportGatewayConfig, QaTransportGatewayClient, + QaTransportNativeCommandInput, QaTransportReportParams, } from "./qa-transport.js"; import { qaChannelPlugin } from "./runtime-api.js"; @@ -147,6 +148,14 @@ class QaChannelTransport extends QaStateBackedTransportAdapter { replyChannel: QA_CHANNEL_ID, replyTo: target, }); + override async sendNativeCommand(input: QaTransportNativeCommandInput): Promise { + const { command, ...message } = input; + await this.sendInbound({ + ...message, + text: `/${command}`, + nativeCommand: { name: command }, + }); + } handleAction = handleQaChannelAction; createReportNotes = createQaChannelReportNotes; } diff --git a/extensions/qa-lab/src/qa-transport.ts b/extensions/qa-lab/src/qa-transport.ts index 036d148b4b2..68330d0366c 100644 --- a/extensions/qa-lab/src/qa-transport.ts +++ b/extensions/qa-lab/src/qa-transport.ts @@ -70,6 +70,13 @@ export type QaTransportWaitForNoOutboundInput = { sinceIndex?: number; }; +export type QaTransportNativeCommandInput = Omit< + QaBusInboundMessageInput, + "nativeCommand" | "text" +> & { + command: string; +}; + export type QaTransportCapabilities = { sendInboundMessage: QaTransportState["addInboundMessage"]; injectOutboundMessage: QaTransportState["addOutboundMessage"]; @@ -181,6 +188,7 @@ export type QaTransportAdapter = { capabilities: QaTransportCapabilities; reset: () => Promise; sendInbound: (input: QaBusInboundMessageInput) => Promise; + sendNativeCommand: (input: QaTransportNativeCommandInput) => Promise; waitForNoOutbound: (input?: QaTransportWaitForNoOutboundInput) => Promise; waitForOutbound: (input: QaTransportOutboundMatch) => Promise; createGatewayConfig: (params: { baseUrl: string }) => QaTransportGatewayConfig; @@ -275,6 +283,10 @@ export abstract class QaStateBackedTransportAdapter implements QaTransportAdapte return await this.state.addInboundMessage(input); } + async sendNativeCommand(_input: QaTransportNativeCommandInput): Promise { + throw new Error(`${this.label} does not support native commands.`); + } + async waitForNoOutbound(input: QaTransportWaitForNoOutboundInput = {}) { const quietMs = resolveTimerTimeoutMs(input.quietMs, 1_200, 0); await sleep(quietMs); diff --git a/extensions/qa-lab/src/scenario-catalog.test.ts b/extensions/qa-lab/src/scenario-catalog.test.ts index 03e60661b00..333f894fcf0 100644 --- a/extensions/qa-lab/src/scenario-catalog.test.ts +++ b/extensions/qa-lab/src/scenario-catalog.test.ts @@ -750,6 +750,20 @@ describe("qa scenario catalog", () => { } }); + it("routes native command session targeting through Crabline Telegram", () => { + const scenario = readQaScenarioById("native-command-session-target"); + const config = readQaScenarioExecutionConfig("native-command-session-target") as + | { + requiredChannelDriver?: string; + requiredProviderMode?: string; + } + | undefined; + + expect(scenario.execution.channel).toBe("telegram"); + expect(config?.requiredChannelDriver).toBeUndefined(); + expect(config?.requiredProviderMode).toBe("mock-openai"); + }); + it("adds a dreaming shadow trial report scenario", () => { const scenario = readQaScenarioById("dreaming-shadow-trial-report"); const config = readQaScenarioExecutionConfig("dreaming-shadow-trial-report") as diff --git a/extensions/qa-lab/src/scenario-catalog.ts b/extensions/qa-lab/src/scenario-catalog.ts index 49e88d973aa..3600efbcec4 100644 --- a/extensions/qa-lab/src/scenario-catalog.ts +++ b/extensions/qa-lab/src/scenario-catalog.ts @@ -158,6 +158,10 @@ const qaFlowTransportActionSchema = z.union([ sendInbound: z.unknown(), saveAs: z.string().trim().min(1).optional(), }), + z.object({ + sendNativeCommand: z.unknown(), + saveAs: z.string().trim().min(1).optional(), + }), z.object({ waitForOutbound: z.unknown(), saveAs: z.string().trim().min(1).optional(), diff --git a/extensions/qa-lab/src/scenario-flow-runner.test.ts b/extensions/qa-lab/src/scenario-flow-runner.test.ts index a672b9880ef..df7da7a6bf9 100644 --- a/extensions/qa-lab/src/scenario-flow-runner.test.ts +++ b/extensions/qa-lab/src/scenario-flow-runner.test.ts @@ -40,6 +40,18 @@ async function runLoadedScenarioFlow( }, sendInbound: async (input: Parameters[0]) => state.addInboundMessage(input), + sendNativeCommand: async ( + input: Omit[0], "nativeCommand" | "text"> & { + command: string; + }, + ) => { + const { command, ...message } = input; + state.addInboundMessage({ + ...message, + text: `/${command}`, + nativeCommand: { name: command }, + }); + }, waitForNoOutbound: async () => undefined, waitForOutbound: async (input: { conversation?: { id: string; kind: string }; diff --git a/extensions/qa-lab/src/scenario-flow-runner.ts b/extensions/qa-lab/src/scenario-flow-runner.ts index 738a85985cd..36f64f07f9f 100644 --- a/extensions/qa-lab/src/scenario-flow-runner.ts +++ b/extensions/qa-lab/src/scenario-flow-runner.ts @@ -161,7 +161,12 @@ async function runFlowAction(action: unknown, api: QaFlowApi, vars: QaFlowVars) } return; } - for (const name of ["sendInbound", "waitForOutbound", "waitForNoOutbound"] as const) { + for (const name of [ + "sendInbound", + "sendNativeCommand", + "waitForOutbound", + "waitForNoOutbound", + ] as const) { if (name in action) { const callable = resolveCallable(`transport.${name}`, api, vars); const result = await callable(await resolveValue(action[name], api, vars)); diff --git a/extensions/qa-lab/src/scenario-runtime-api.test.ts b/extensions/qa-lab/src/scenario-runtime-api.test.ts index 43ffe1e27d9..773b85b40a6 100644 --- a/extensions/qa-lab/src/scenario-runtime-api.test.ts +++ b/extensions/qa-lab/src/scenario-runtime-api.test.ts @@ -138,6 +138,18 @@ describe("createQaScenarioRuntimeApi", () => { }, sendInbound: async (input: Parameters[0]) => state.addInboundMessage(input), + sendNativeCommand: async ( + input: Omit[0], "nativeCommand" | "text"> & { + command: string; + }, + ) => { + const { command, ...message } = input; + state.addInboundMessage({ + ...message, + text: `/${command}`, + nativeCommand: { name: command }, + }); + }, waitForNoOutbound: vi.fn(async () => undefined), waitForOutbound: vi.fn(async () => { throw new Error("not used"); diff --git a/extensions/qa-lab/src/scenario-runtime-api.ts b/extensions/qa-lab/src/scenario-runtime-api.ts index 3b88b883988..d6025b6cb15 100644 --- a/extensions/qa-lab/src/scenario-runtime-api.ts +++ b/extensions/qa-lab/src/scenario-runtime-api.ts @@ -8,7 +8,13 @@ type QaScenarioRuntimeFunction = (...args: never[]) => unknown; type QaScenarioTransport = Pick< QaTransportAdapter, - "capabilities" | "reset" | "sendInbound" | "state" | "waitForNoOutbound" | "waitForOutbound" + | "capabilities" + | "reset" + | "sendInbound" + | "sendNativeCommand" + | "state" + | "waitForNoOutbound" + | "waitForOutbound" >; export type QaScenarioRuntimeEnv< diff --git a/extensions/qa-lab/src/suite-runtime-flow.test.ts b/extensions/qa-lab/src/suite-runtime-flow.test.ts index f36f4efe2b3..5ca278fa2d2 100644 --- a/extensions/qa-lab/src/suite-runtime-flow.test.ts +++ b/extensions/qa-lab/src/suite-runtime-flow.test.ts @@ -181,6 +181,7 @@ describe("qa suite runtime flow", () => { createReportNotes: vi.fn(), reset: vi.fn(), sendInbound: vi.fn(), + sendNativeCommand: vi.fn(), waitForNoOutbound: vi.fn(), waitForOutbound: vi.fn(), state: { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8f19ca7fabd..3f7a2e07dd2 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1350,8 +1350,8 @@ importers: version: 4.4.3 devDependencies: '@openclaw/crabline': - specifier: 0.1.6 - version: 0.1.6 + specifier: 0.1.7 + version: 0.1.7 '@openclaw/discord': specifier: workspace:* version: link:../discord @@ -3165,8 +3165,8 @@ packages: cpu: [x64] os: [win32] - '@openclaw/crabline@0.1.6': - resolution: {integrity: sha512-nu/XD7eoly5DJOEG7krCfZY68cDcD/AC31mAMLoP36HzmetNVM17OtfnakpuWyBwdO+c4noF7b2LM4AAKvq9CQ==} + '@openclaw/crabline@0.1.7': + resolution: {integrity: sha512-FwKY45UJ4nlKA3wMEGb9KSxNvJgI4gzaLttr4wxARpurQArzVyf5z3IjWzN7b9sra0oNQTk2Y8ugdOtJqSjI4g==} engines: {node: '>=22'} hasBin: true @@ -9198,7 +9198,7 @@ snapshots: '@openai/codex@0.142.4-win32-x64': optional: true - '@openclaw/crabline@0.1.6': + '@openclaw/crabline@0.1.7': dependencies: commander: 15.0.0 curve25519-js: 0.0.4 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index b6022e7702e..56d99e5e92d 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -7,7 +7,7 @@ packages: minimumReleaseAge: 2880 minimumReleaseAgeExclude: - - "@openclaw/crabline@0.1.6" + - "@openclaw/crabline@0.1.7" - "@openclaw/fs-safe@0.3.0" - "@openclaw/proxyline@0.3.3" - "acpx" diff --git a/qa/maturity-coverage-investigation.md b/qa/maturity-coverage-investigation.md index a5ed182e6a2..f923e40c45f 100644 --- a/qa/maturity-coverage-investigation.md +++ b/qa/maturity-coverage-investigation.md @@ -106,7 +106,6 @@ Add small native scenario YAML wrappers for these rather than duplicating the te | `automation.heartbeat-scheduling` | automation-cron-hooks-tasks-polling | Heartbeat | `src/infra/heartbeat-runner.active-hours-schedule.e2e.test.ts` | | `ui.assistant-media-tickets` | browser-control-ui-and-webchat | WebChat Conversations | `src/gateway/control-ui-assistant-media.e2e.test.ts` | | `ui.browser-talk-start-stop` | browser-control-ui-and-webchat | Browser Realtime Talk | `ui/src/ui/realtime-talk-google-live.test.ts` | -| `channels.native-command-session-target` | channel-framework | Channel Actions Commands and Approvals | `src/auto-reply/reply.triggers.trigger-handling.targets-active-session-native-stop.e2e.test.ts` | | `clawhub.marketplace-list` | clawhub-and-external-plugin-distribution | Plugin Lifecycle and Health | `scripts/e2e/lib/plugins/marketplace.sh`
`scripts/e2e/lib/release-plugin-marketplace/scenario.sh` | | `clawhub.npm-pack-local-release-candidate-installs` | clawhub-and-external-plugin-distribution | Plugin Lifecycle and Health | `scripts/release-candidate-checklist.mjs`
`test/scripts/release-candidate-checklist.test.ts` | | `clawhub.skill-installs` | clawhub-and-external-plugin-distribution | Plugin Lifecycle and Health | `src/cli/skills-cli.clawhub-install.e2e.test.ts` | diff --git a/qa/scenarios/channels/native-command-session-target.yaml b/qa/scenarios/channels/native-command-session-target.yaml index d29b5e7f112..947ab8a7774 100644 --- a/qa/scenarios/channels/native-command-session-target.yaml +++ b/qa/scenarios/channels/native-command-session-target.yaml @@ -7,17 +7,113 @@ scenario: coverage: primary: - channels.native-command-session-target - objective: Link native command target-session e2e coverage to channel framework maturity accounting. + secondary: + - channels.native-commands + objective: Verify a channel-native `/stop` command aborts the active routed conversation session instead of its separate slash-command session. successCriteria: - - Native `/stop` commands use the active target session key instead of the slash-command session. - - The target embedded agent run is aborted. - - Queued follow-up work for the target session is cleared. + - A real delayed agent turn is active on the routed channel conversation session. + - The selected transport sends a provider-native command that targets the routed conversation session. + - Native `/stop` aborts the active turn, returns the abort acknowledgement, and unblocks the next turn. docsRefs: - docs/channels/qa-channel.md + - docs/channels/telegram.md - docs/help/testing.md codeRefs: - - src/auto-reply/reply.triggers.trigger-handling.targets-active-session-native-stop.e2e.test.ts + - extensions/qa-channel/src/inbound.ts + - extensions/qa-lab/src/crabline-transport.ts + - extensions/telegram/src/bot-native-commands.ts + - src/channels/native-command-session-targets.ts + - src/auto-reply/reply/abort.ts execution: - kind: vitest - path: src/auto-reply/reply.triggers.trigger-handling.targets-active-session-native-stop.e2e.test.ts - summary: Vitest e2e coverage for native command active-session targeting. + kind: flow + channel: telegram + suiteIsolation: isolated + isolationReason: Waits for the one active routed session before interrupting it with a provider-native command. + summary: Start a real delayed channel turn, abort it through native `/stop`, then prove the conversation is unblocked. + config: + requiredProviderMode: mock-openai + conversationId: native-stop-target + senderId: qa-native-operator + delayedPrompt: "Subagent recovery worker native command target proof. Wait until stopped." + abortReplyNeedle: Agent was aborted + recoveryMarker: QA-NATIVE-STOP-RECOVERY-OK + +flow: + steps: + - name: native stop targets the active conversation session + actions: + - assert: + expr: "env.providerMode === config.requiredProviderMode" + message: this deterministic active-run proof requires mock-openai + - call: waitForGatewayHealthy + args: + - ref: env + - 60000 + - call: waitForTransportReady + args: + - ref: env + - 60000 + - resetTransport: true + - sendInbound: + conversation: + id: + expr: config.conversationId + kind: direct + senderId: + expr: config.senderId + senderName: QA Native Operator + text: + expr: config.delayedPrompt + - call: waitForCondition + saveAs: activeSession + args: + - lambda: + async: true + expr: "env.gateway.call('sessions.list', {}).then((result) => result.sessions?.find((session) => session.hasActiveRun === true))" + - 5000 + - 100 + - set: startIndex + value: + expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound').length" + - sendNativeCommand: + command: stop + conversation: + id: + expr: config.conversationId + kind: direct + senderId: + expr: config.senderId + senderName: QA Native Operator + - waitForOutbound: + conversation: + id: + expr: config.conversationId + kind: direct + sinceIndex: + ref: startIndex + textIncludes: + expr: config.abortReplyNeedle + timeoutMs: 15000 + saveAs: abortReply + - sendInbound: + conversation: + id: + expr: config.conversationId + kind: direct + senderId: + expr: config.senderId + senderName: QA Native Operator + text: + expr: "`Reply exactly: ${config.recoveryMarker}`" + - waitForOutbound: + conversation: + id: + expr: config.conversationId + kind: direct + sinceIndex: + ref: startIndex + textIncludes: + expr: config.recoveryMarker + timeoutMs: 15000 + saveAs: recoveryReply + detailsExpr: "`native command reply=${abortReply.text}; recovery reply=${recoveryReply.text}`" diff --git a/src/auto-reply/reply.triggers.trigger-handling.targets-active-session-native-stop.e2e.test.ts b/src/auto-reply/reply.triggers.trigger-handling.e2e.test.ts similarity index 98% rename from src/auto-reply/reply.triggers.trigger-handling.targets-active-session-native-stop.e2e.test.ts rename to src/auto-reply/reply.triggers.trigger-handling.e2e.test.ts index c3338ccae00..f8500b7bbbd 100644 --- a/src/auto-reply/reply.triggers.trigger-handling.targets-active-session-native-stop.e2e.test.ts +++ b/src/auto-reply/reply.triggers.trigger-handling.e2e.test.ts @@ -1,4 +1,4 @@ -/** E2E tests for native /stop targeting the active auto-reply session. */ +/** E2E tests for auto-reply trigger and command handling. */ import fs from "node:fs/promises"; import { join } from "node:path"; import { describe, expect, it, vi } from "vitest"; @@ -604,10 +604,7 @@ describe("trigger handling", () => { const cfg = makeCfg(home); cfg.session = { ...cfg.session, store: join(home, "native-stop.sessions.json") }; getAbortEmbeddedAgentRunMock().mockReset().mockReturnValue(false); - const storePath = cfg.session?.store; - if (!storePath) { - throw new Error("missing session store path"); - } + const storePath = requireSessionStorePath(cfg); const targetSessionKey = "agent:main:telegram:group:123"; const targetSessionId = "session-target"; await saveSessionStore( @@ -664,8 +661,7 @@ describe("trigger handling", () => { cfg, ); - const text = Array.isArray(res) ? res[0]?.text : res?.text; - expect(text).toBe("⚙️ Agent was aborted."); + expect(maybeReplyText(res)).toBe("⚙️ Agent was aborted."); expect(getAbortEmbeddedAgentRunMock()).toHaveBeenCalledWith(targetSessionId); const store = loadSessionStore(storePath); expect(store[targetSessionKey]?.abortedLastRun).toBe(true); diff --git a/src/plugin-sdk/qa-channel-protocol.ts b/src/plugin-sdk/qa-channel-protocol.ts index e7cb90fccaf..46244e72319 100644 --- a/src/plugin-sdk/qa-channel-protocol.ts +++ b/src/plugin-sdk/qa-channel-protocol.ts @@ -33,6 +33,11 @@ export type QaBusToolCall = { arguments?: Record; }; +/** Channel-native command metadata attached to a synthetic inbound message. */ +export type QaBusNativeCommand = { + name: string; +}; + /** Stored QA bus message after defaults, reactions, and account ids are normalized. */ export type QaBusMessage = { id: string; @@ -49,6 +54,7 @@ export type QaBusMessage = { deleted?: boolean; editedAt?: number; attachments?: QaBusAttachment[]; + nativeCommand?: QaBusNativeCommand; toolCalls?: QaBusToolCall[]; reactions: Array<{ emoji: string; @@ -95,6 +101,7 @@ export type QaBusInboundMessageInput = { threadTitle?: string; replyToId?: string; attachments?: QaBusAttachment[]; + nativeCommand?: QaBusNativeCommand; toolCalls?: QaBusToolCall[]; };