diff --git a/.github/workflows/mantis-telegram-live.yml b/.github/workflows/mantis-telegram-live.yml index 87831ae9caf9..6ea1af97bc11 100644 --- a/.github/workflows/mantis-telegram-live.yml +++ b/.github/workflows/mantis-telegram-live.yml @@ -19,6 +19,14 @@ on: required: false default: telegram-status-command type: string + provider_mode: + description: QA provider mode + required: false + default: live-frontier + type: choice + options: + - live-frontier + - mock-openai crabbox_provider: description: Crabbox provider for the desktop transcript capture required: false @@ -98,6 +106,7 @@ jobs: crabbox_provider: ${{ steps.resolve.outputs.crabbox_provider }} lease_id: ${{ steps.resolve.outputs.lease_id }} pr_number: ${{ steps.resolve.outputs.pr_number }} + provider_mode: ${{ steps.resolve.outputs.provider_mode }} reaction_id: ${{ steps.add_reaction.outputs.reaction_id }} request_source: ${{ steps.resolve.outputs.request_source }} scenario: ${{ steps.resolve.outputs.scenario }} @@ -117,10 +126,16 @@ jobs: if (eventName === "workflow_dispatch") { const inputs = context.payload.inputs ?? {}; + const providerMode = inputs.provider_mode || "live-frontier"; + if (!["live-frontier", "mock-openai"].includes(providerMode)) { + core.setFailed(`Unsupported provider mode for Mantis Telegram: ${providerMode}`); + return; + } setOutput("should_run", "true"); setOutput("candidate_ref", inputs.candidate_ref || "main"); setOutput("pr_number", inputs.pr_number || ""); setOutput("scenario", inputs.scenario || "telegram-status-command"); + setOutput("provider_mode", providerMode); setOutput("crabbox_provider", inputs.crabbox_provider || "aws"); setOutput("lease_id", inputs.crabbox_lease_id || ""); setOutput("request_source", "workflow_dispatch"); @@ -158,6 +173,7 @@ jobs: setOutput("candidate_ref", ""); setOutput("pr_number", ""); setOutput("scenario", ""); + setOutput("provider_mode", ""); setOutput("crabbox_provider", ""); setOutput("lease_id", ""); setOutput("request_source", "unsupported_issue_comment"); @@ -172,6 +188,7 @@ jobs: }); const candidateMatch = body.match(/(?:candidate|head)[\s:=]+([^\s`]+)/i); const scenarioMatch = body.match(/(?:scenario|scenarios)[\s:=]+([^\s`]+)/i); + const providerModeMatch = body.match(/(?:provider_mode|provider-mode)[\s:=]+([^\s`]+)/i); const providerMatch = body.match(/(?:provider|crabbox_provider)[\s:=]+([^\s`]+)/i); const leaseMatch = body.match(/(?:lease|lease_id|crabbox_lease_id)[\s:=]+([^\s`]+)/i); const rawCandidate = candidateMatch?.[1]; @@ -184,11 +201,17 @@ jobs: core.setFailed(`Unsupported Crabbox provider for Mantis Telegram: ${provider}`); return; } + const providerMode = providerModeMatch?.[1] || "live-frontier"; + if (!["live-frontier", "mock-openai"].includes(providerMode)) { + core.setFailed(`Unsupported provider mode for Mantis Telegram: ${providerMode}`); + return; + } setOutput("should_run", "true"); setOutput("candidate_ref", candidate); setOutput("pr_number", String(issue.number)); setOutput("scenario", scenarioMatch?.[1] || "telegram-status-command"); + setOutput("provider_mode", providerMode); setOutput("crabbox_provider", provider); setOutput("lease_id", leaseMatch?.[1] || ""); setOutput("request_source", "issue_comment"); @@ -412,6 +435,7 @@ jobs: CRABBOX_CAPACITY_REGIONS: ${{ env.CRABBOX_CAPACITY_REGIONS }} CRABBOX_LEASE_ID: ${{ needs.resolve_request.outputs.lease_id }} CRABBOX_PROVIDER: ${{ needs.resolve_request.outputs.crabbox_provider }} + PROVIDER_MODE: ${{ needs.resolve_request.outputs.provider_mode }} SCENARIO_INPUT: ${{ needs.resolve_request.outputs.scenario }} CANDIDATE_SHA: ${{ needs.validate_ref.outputs.candidate_revision }} shell: bash @@ -430,7 +454,6 @@ jobs: CRABBOX_COORDINATOR_TOKEN="${CRABBOX_COORDINATOR_TOKEN:-${OPENCLAW_QA_MANTIS_CRABBOX_COORDINATOR_TOKEN:-}}" export CRABBOX_COORDINATOR CRABBOX_COORDINATOR_TOKEN - require_var OPENAI_API_KEY require_var OPENCLAW_QA_CONVEX_SITE_URL require_var OPENCLAW_QA_CONVEX_SECRET_CI require_var CRABBOX_COORDINATOR_TOKEN @@ -439,7 +462,29 @@ jobs: output_rel=".artifacts/qa-e2e/mantis/telegram-live" root="$candidate_repo/$output_rel" echo "output_dir=${root}" >> "$GITHUB_OUTPUT" - model="${OPENCLAW_CI_OPENAI_MODEL:-openai/gpt-5.6-luna}" + + qa_args=( + --repo-root "$candidate_repo" + --output-dir "$output_rel" + --provider-mode "$PROVIDER_MODE" + ) + case "$PROVIDER_MODE" in + live-frontier) + require_var OPENAI_API_KEY + model="${OPENCLAW_CI_OPENAI_MODEL:-openai/gpt-5.6-luna}" + qa_args+=(--model "$model" --alt-model "$model" --fast) + ;; + mock-openai) ;; + *) + echo "Unsupported provider mode for Mantis Telegram: ${PROVIDER_MODE}" >&2 + exit 1 + ;; + esac + qa_args+=( + --credential-source convex + --credential-role ci + --allow-failures + ) scenario_args=() if [[ -n "${SCENARIO_INPUT// }" ]]; then @@ -453,17 +498,7 @@ jobs: fi set +e - pnpm --dir "$candidate_repo" openclaw qa telegram \ - --repo-root "$candidate_repo" \ - --output-dir "$output_rel" \ - --provider-mode live-frontier \ - --model "$model" \ - --alt-model "$model" \ - --fast \ - --credential-source convex \ - --credential-role ci \ - --allow-failures \ - "${scenario_args[@]}" + pnpm --dir "$candidate_repo" openclaw qa telegram "${qa_args[@]}" "${scenario_args[@]}" telegram_exit=$? set -e diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts index bdceabe55de2..ed61b4668db6 100644 --- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import type { StreamEvent } from "./mock-openai-contracts.js"; import { buildAssistantEvents, + buildPartialFailureEvents, buildAssistantThenToolCallEvents, buildFailedResponseEvents, buildReasoningAndAssistantEvents, @@ -31,13 +32,35 @@ function readOutputItemSlots(events: StreamEvent[]) { describe("mock OpenAI Responses output item slots", () => { it("emits the provider no-details failure used by repeated-request recovery QA", () => { - expect(buildFailedResponseEvents()).toEqual([ + const events = buildFailedResponseEvents(); + expect(events).toEqual([ expect.objectContaining({ type: "response.created" }), expect.objectContaining({ type: "response.failed", response: expect.not.objectContaining({ error: expect.anything() }), }), ]); + expect(events.some((event) => event.type === "response.output_text.delta")).toBe(false); + }); + + it("emits an unfinished assistant delta before the failed response", () => { + const marker = "TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE"; + const events = buildPartialFailureEvents(marker); + + expect(events.map((event) => event.type)).toEqual([ + "response.created", + "response.output_item.added", + "response.output_text.delta", + "response.failed", + ]); + expect(events[1]).toMatchObject({ + item: { type: "message", role: "assistant", status: "in_progress" }, + }); + expect(events[2]).toMatchObject({ + type: "response.output_text.delta", + delta: marker, + }); + expect(events.some((event) => event.type === "response.output_item.done")).toBe(false); }); it("indexes preview deltas and the final answer on the same assistant slot", () => { diff --git a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts index 2cbd08467822..e8d7c86441dc 100644 --- a/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts +++ b/extensions/qa-lab/src/providers/mock-openai/mock-openai-events.ts @@ -21,6 +21,40 @@ export function buildFailedResponseEvents(): StreamEvent[] { ]; } +export function buildPartialFailureEvents(partialText: string): StreamEvent[] { + const responseId = "resp_qa_partial_failed_1"; + const itemId = "msg_qa_partial_failed_1"; + return [ + { type: "response.created", response: { id: responseId } }, + { + type: "response.output_item.added", + output_index: 0, + item: { + type: "message", + id: itemId, + role: "assistant", + phase: "final_answer", + content: [], + status: "in_progress", + }, + }, + { + type: "response.output_text.delta", + item_id: itemId, + output_index: 0, + content_index: 0, + delta: partialText, + }, + { + type: "response.failed", + response: { + id: responseId, + status: "failed", + }, + }, + ]; +} + export function buildToolCallEvents(prompt: string): StreamEvent[] { const targetPath = readTargetFromPrompt(prompt); return buildToolCallEventsWithArgs("read", { path: targetPath }); diff --git a/extensions/qa-lab/src/providers/mock-openai/server.test.ts b/extensions/qa-lab/src/providers/mock-openai/server.test.ts index e7abb64fc38f..52885416a27a 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.test.ts @@ -219,6 +219,13 @@ function expectOpenAiStreamingResponsesText(server: MockServer, body: Record line.startsWith("data: {") && line.endsWith("}")) + .map((line) => JSON.parse(line.slice("data: ".length)) as StreamEvent); +} + const requireRecord = createRequireRecord("record", "expected-label-capitalized"); function requireArray(value: unknown, label: string): unknown[] { @@ -913,6 +920,36 @@ describe("qa mock openai server", () => { expect(blockContinuationBody).not.toContain('"item_id":"msg_mock_block_1"'); }); + it("serves Telegram visible and unsent failure directives", async () => { + const server = await startMockServer(); + const visibleEvents = parseStreamingResponseEvents( + await expectOpenAiStreamingResponsesText(server, { + input: [makeUserInput("Telegram visible partial failure QA check")], + }), + ); + const unsentEvents = parseStreamingResponseEvents( + await expectOpenAiStreamingResponsesText(server, { + input: [makeUserInput("Telegram unsent failure QA check")], + }), + ); + + expect(visibleEvents.map((event) => event.type)).toEqual([ + "response.created", + "response.output_item.added", + "response.output_text.delta", + "response.failed", + ]); + expect(visibleEvents[2]).toMatchObject({ + type: "response.output_text.delta", + delta: "TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE", + }); + expect(unsentEvents.map((event) => event.type)).toEqual([ + "response.created", + "response.failed", + ]); + expect(unsentEvents.some((event) => event.type === "response.output_text.delta")).toBe(false); + }); + it("plans deterministic tool-progress reads from prompt paths", async () => { const server = await startMockServer(); diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index 62b10d6e8c3a..8feac7a08144 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -144,6 +144,7 @@ import { buildQaLongFinalText, buildAssistantThenToolCallEvents, buildAssistantEvents, + buildPartialFailureEvents, buildReasoningOnlyEvents, buildReasoningAndAssistantEvents, buildFailedResponseEvents, @@ -288,6 +289,9 @@ const QA_STREAMING_TOOL_PROGRESS_CONTINUATION_RE = /^Continue with (?:the current Matrix QA scenario|the QA scenario plan and report worked, failed, and blocked items)\.$/i; const QA_CODE_MODE_TARGET_MARKER = "qa-code-mode-target:"; const QA_FAILED_TOOL_TERMINAL_RECOVERY_PROMPT_RE = /failed tool terminal recovery qa check/i; +const QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_PROMPT_RE = /telegram visible partial failure qa check/i; +const QA_TELEGRAM_UNSENT_FAILURE_PROMPT_RE = /telegram unsent failure qa check/i; +const QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_MARKER = "TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE"; // Keep each real provider request active long enough for retries to span the // unchanged five-minute recovery bound while remaining below first-byte timeout. const QA_REPEATED_REQUEST_RESPONSE_PAUSE_MS = 110_000; @@ -868,6 +872,12 @@ async function buildResponsesPayload( if (QA_REPEATED_REQUEST_QUEUED_REPLY_PROMPT_RE.test(prompt)) { return buildAssistantEvents(QA_REPEATED_REQUEST_QUEUED_REPLY_MARKER); } + if (QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_PROMPT_RE.test(prompt)) { + return buildPartialFailureEvents(QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_MARKER); + } + if (QA_TELEGRAM_UNSENT_FAILURE_PROMPT_RE.test(prompt)) { + return buildFailedResponseEvents(); + } if (QA_REPEATED_REQUEST_RECOVERY_PROMPT_RE.test(allInputText)) { return buildFailedResponseEvents(); } diff --git a/extensions/telegram/src/bot-message-dispatch-turn.ts b/extensions/telegram/src/bot-message-dispatch-turn.ts index cf0c2c6de063..a5e3e3c762ac 100644 --- a/extensions/telegram/src/bot-message-dispatch-turn.ts +++ b/extensions/telegram/src/bot-message-dispatch-turn.ts @@ -158,7 +158,16 @@ export async function runTelegramDispatchTurn(params: { const queued = params.draft.enqueueEvent(async () => { await params.draft.ingestDraftLaneSegments(payload); }); - return queued.then(() => false); + // Queue settlement records draft intent; a numeric provider message ID + // proves operator visibility for terminal recovery. + return queued.then(async () => { + const answerStream = params.draft.answerLane.stream; + await answerStream?.waitForInFlight(); + const providerMessageId = answerStream?.messageId(); + return ( + typeof providerMessageId === "number" && Number.isFinite(providerMessageId) + ); + }); } : undefined, onBlockReplyQueued: params.draft.answerLane.stream diff --git a/extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts b/extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts index 8ffd5694a4c5..2c3ca8d250d7 100644 --- a/extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts +++ b/extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts @@ -18,6 +18,7 @@ import { setupDraftStreams, telegramProgressPreview, } from "./bot-message-dispatch.test-harness.js"; +import { createTestDraftStream } from "./draft-stream.test-helpers.js"; const draftWarn = vi.hoisted(() => vi.fn()); @@ -80,36 +81,72 @@ describeTelegramDispatch("dispatchTelegramMessage draft-failures-progress", () = }); it.each([ - { label: "direct chat", createSessionPayload: createDirectSessionPayload }, + { + label: "direct chat", + createMessageContext: () => + createContext({ + ctxPayload: createDirectSessionPayload(), + }), + }, { label: "group chat", - createSessionPayload: () => ({ - ...createDirectSessionPayload(), - SessionKey: "agent:test:telegram:group:-100123", - ChatType: "group" as const, - }), + createMessageContext: () => + createContext({ + chatId: -100123, + isGroup: true, + ctxPayload: { + ...createDirectSessionPayload(), + SessionKey: "agent:test:telegram:group:-100123", + ChatType: "group", + }, + primaryCtx: { + ...createContext().primaryCtx, + message: { + chat: { id: -100123, type: "supergroup", title: "Test group" }, + date: 0, + message_id: 456, + }, + }, + msg: { + chat: { id: -100123, type: "supergroup", title: "Test group" }, + date: 0, + message_id: 456, + message_thread_id: undefined, + }, + threadSpec: { id: undefined, scope: "none" }, + replyThreadId: undefined, + }), }, ])( "finalizes the default streamed draft in place after an unexpected reply failure in a $label", - async ({ createSessionPayload }) => { - const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); + async ({ createMessageContext }) => { + const answerDraftStream = createTestDraftStream({ + onWaitForInFlight: () => answerDraftStream.setMessageId(2001), + }); + const reasoningDraftStream = createTestDraftStream(); + createTelegramDraftStream + .mockImplementationOnce(() => answerDraftStream) + .mockImplementationOnce(() => reasoningDraftStream); + let partialAccepted: boolean | void = undefined; dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async (params) => { expect(params.replyOptions?.disableBlockStreaming).toBe(true); return await dispatchReplyWithBufferedBlockDispatcherRuntime({ ...params, replyResolver: async (_ctx, opts) => { - await opts?.onPartialReply?.({ text: "partial answer" }); + partialAccepted = await opts?.onPartialReply?.({ text: "partial answer" }); throw new Error("unexpected model failure"); }, }); }); await dispatchWithContext({ - context: createContext({ ctxPayload: createSessionPayload() }), + context: createMessageContext(), streamMode: "partial", telegramCfg: { streaming: { mode: "partial" } }, }); + expect(partialAccepted).toBeUndefined(); + expect(answerDraftStream.waitForInFlight).toHaveBeenCalledOnce(); expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "partial answer"); expect(answerDraftStream.update).toHaveBeenCalledTimes(2); expect(answerDraftStream.update).toHaveBeenLastCalledWith( @@ -122,6 +159,35 @@ describeTelegramDispatch("dispatchTelegramMessage draft-failures-progress", () = }, ); + it("clears a pending partial and sends one fallback after an unexpected reply failure", async () => { + const { answerDraftStream } = setupDraftStreams(); + let partialAccepted: boolean | void = undefined; + dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async (params) => { + return await dispatchReplyWithBufferedBlockDispatcherRuntime({ + ...params, + replyResolver: async (_ctx, opts) => { + partialAccepted = await opts?.onPartialReply?.({ text: "partial answer" }); + throw new Error("unexpected model failure"); + }, + }); + }); + + await dispatchWithContext({ + context: createContext({ ctxPayload: createDirectSessionPayload() }), + streamMode: "partial", + telegramCfg: { streaming: { mode: "partial" } }, + }); + + expect(partialAccepted).toBe(false); + expect(answerDraftStream.update).toHaveBeenCalledOnce(); + expect(answerDraftStream.update).toHaveBeenCalledWith("partial answer"); + expect(answerDraftStream.clear).toHaveBeenCalledOnce(); + expect(deliverReplies).toHaveBeenCalledOnce(); + expectDeliveredReply(0, { + text: "Something went wrong while processing your request. Please try again.", + }); + }); + it("returns retryable when dispatch fails after partial output and the fallback is not delivered", async () => { deliverReplies.mockResolvedValueOnce({ delivered: true }); deliverReplies.mockResolvedValueOnce({ delivered: false }); diff --git a/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts b/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts index 3bd6a15b1c9f..ac6fc785ee1b 100644 --- a/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts +++ b/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts @@ -187,7 +187,7 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => { expect(answerDraftStream.clear).not.toHaveBeenCalled(); expect(answerDraftStream.update).toHaveBeenCalledWith("Photo"); - expect(partialAccepted).toBe(false); + expect(partialAccepted).toBe(true); expectDeliverRepliesParams({ mediaMaxBytes }); expectDeliveredReply(0, { text: undefined, mediaUrl: "https://example.com/a.png" }); expect(emitTelegramMessageSentHooks).toHaveBeenCalledTimes(1); diff --git a/extensions/telegram/src/draft-stream.test-helpers.ts b/extensions/telegram/src/draft-stream.test-helpers.ts index 34ceff92d6f9..c889d8ea17df 100644 --- a/extensions/telegram/src/draft-stream.test-helpers.ts +++ b/extensions/telegram/src/draft-stream.test-helpers.ts @@ -11,6 +11,7 @@ type TestDraftStream = { updateLazy: ReturnType string | undefined) => void>>; updatePreview: ReturnType void>>; flush: ReturnType Promise>>; + waitForInFlight: ReturnType Promise>>; messageId: ReturnType number | undefined>>; lastDeliveredText: ReturnType string>>; currentMessageSnapshot: ReturnType TelegramDraftMessageSnapshot | undefined>>; @@ -31,6 +32,7 @@ type TestDraftStream = { export function createTestDraftStream(params?: { messageId?: number; onUpdate?: (text: string) => void; + onWaitForInFlight?: () => void | Promise; onStop?: () => void | Promise; onDiscard?: () => void | Promise; clearMessageIdOnForceNew?: boolean; @@ -64,6 +66,9 @@ export function createTestDraftStream(params?: { params?.onUpdate?.(preview.text); }), flush: vi.fn().mockResolvedValue(undefined), + waitForInFlight: vi.fn().mockImplementation(async () => { + await params?.onWaitForInFlight?.(); + }), messageId: vi.fn().mockImplementation(() => messageId), lastDeliveredText: vi.fn().mockImplementation(() => lastDeliveredText), currentMessageSnapshot: vi @@ -142,6 +147,7 @@ export function createSequencedTestDraftStream(startMessageId = 1001): TestDraft lastDeliveredText = preview.text.trimEnd(); }), flush: vi.fn().mockResolvedValue(undefined), + waitForInFlight: vi.fn().mockResolvedValue(undefined), messageId: vi.fn().mockImplementation(() => activeMessageId), lastDeliveredText: vi.fn().mockImplementation(() => lastDeliveredText), currentMessageSnapshot: vi diff --git a/extensions/telegram/src/draft-stream.test.ts b/extensions/telegram/src/draft-stream.test.ts index 8a440d037fbb..71483e1b2698 100644 --- a/extensions/telegram/src/draft-stream.test.ts +++ b/extensions/telegram/src/draft-stream.test.ts @@ -194,7 +194,7 @@ describe("createTelegramDraftStream", () => { const stream = createDraftStream(api, { validateProviderMessage }); stream.update("First preview"); - await expect(stream.flush()).rejects.toBe(validationError); + await expect(stream.waitForInFlight()).rejects.toBe(validationError); stream.update("Second preview"); await expect(stream.flush()).rejects.toBe(validationError); diff --git a/extensions/telegram/src/draft-stream.ts b/extensions/telegram/src/draft-stream.ts index b4d50256d2fc..c1acef630328 100644 --- a/extensions/telegram/src/draft-stream.ts +++ b/extensions/telegram/src/draft-stream.ts @@ -65,6 +65,7 @@ export type TelegramDraftStream = { updateLazy: (resolveText: () => string | undefined) => void; updatePreview: (preview: TelegramDraftPreview) => void; flush: () => Promise; + waitForInFlight: () => Promise; messageId: () => number | undefined; lastDeliveredText?: () => string; currentMessageSnapshot?: () => TelegramDraftMessageSnapshot | undefined; @@ -758,8 +759,12 @@ export function createTelegramDraftStream(params: { throw terminalDeliveryError; } }; - const flush = async () => { + const waitForInFlight = async () => { await loop.waitForInFlight(); + throwTerminalDeliveryError(); + }; + const flush = async () => { + await waitForInFlight(); if (!streamState.stopped) { await loop.flush(); } @@ -1048,6 +1053,7 @@ export function createTelegramDraftStream(params: { updateLazy: requestLazyDraftUpdate, updatePreview, flush, + waitForInFlight, messageId: () => streamMessageId, lastDeliveredText: () => lastDeliveredText, currentMessageSnapshot: () => streamMessageSnapshot, diff --git a/extensions/telegram/src/outbound-adapter.web-app.test.ts b/extensions/telegram/src/outbound-adapter.web-app.test.ts index 458dbf061ec7..fb800a0b46c2 100644 --- a/extensions/telegram/src/outbound-adapter.web-app.test.ts +++ b/extensions/telegram/src/outbound-adapter.web-app.test.ts @@ -19,10 +19,11 @@ describe("Telegram outbound web app presentation", () => { it.each(["-1001234567890", "@channelname"])( "falls back to a link for non-DM target %s", async (to) => { + const payload = { text: "Open app:" }; const rendered = await telegramOutbound.renderPresentation?.({ - payload: { text: "Open app:" }, + payload, presentation: webAppPresentation, - ctx: { to } as never, + ctx: { cfg: {}, to, text: payload.text, payload }, }); expect(rendered).toEqual({ diff --git a/qa/scenarios/channels/telegram-partial-failure-recovery.yaml b/qa/scenarios/channels/telegram-partial-failure-recovery.yaml new file mode 100644 index 000000000000..3f438c40596f --- /dev/null +++ b/qa/scenarios/channels/telegram-partial-failure-recovery.yaml @@ -0,0 +1,107 @@ +title: Telegram partial failure recovery + +scenario: + id: telegram-partial-failure-recovery + surface: channels + category: channels.conversation-routing-and-delivery + coverage: + primary: + - channels.streaming-final-reply + regressionRefs: + - openclaw/openclaw#120626 + objective: Verify Telegram recovers provider failures in one logical message whether or not a streaming preview became visible. + successCriteria: + - A visible partial is retained and the terminal recovery edits the same Telegram message. + - A failure before any visible partial produces one fallback Telegram message. + codeRefs: + - extensions/telegram/src/bot-message-dispatch-turn.ts + - extensions/telegram/src/draft-stream.ts + - extensions/qa-lab/src/providers/mock-openai/server.ts + - extensions/qa-lab/src/live-transports/telegram/adapter.runtime.ts + gatewayConfigPatch: + agents: + defaults: + model: + fallbacks: [] + entries: + qa: + model: + fallbacks: [] + channels: + telegram: + streaming: + mode: partial + execution: + kind: flow + channel: telegram + summary: Inject provider failures before and after Telegram accepts a streaming preview, then verify one logical recovery message remains. + config: + requiredProviderMode: mock-openai + partialMarker: TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE + +flow: + steps: + - name: preserves a visible partial in one recovered message + actions: + - assert: + expr: env.providerMode === config.requiredProviderMode + message: this Telegram recovery scenario requires mock-openai + - call: waitForGatewayHealthy + args: [{ ref: env }, 60000] + - call: waitForTransportReady + args: [{ ref: env }, 60000] + - resetTransport: true + - set: visibleOutboundStart + value: + expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').length" + - sendInbound: + conversation: { id: telegram-partial-failure-room, kind: group } + senderId: qa-telegram-recovery-operator + senderName: Telegram Recovery Operator + text: "@openclaw Telegram visible partial failure QA check" + - waitForOutbound: + conversation: { id: telegram-partial-failure-room, kind: group } + sinceIndex: { ref: visibleOutboundStart } + textIncludes: { ref: config.partialMarker } + timeoutMs: 75000 + saveAs: visiblePartial + - set: visiblePartialText + value: + expr: "visiblePartial.text" + - call: sleep + args: [4000] + - set: visibleMessages + value: + expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').slice(visibleOutboundStart)" + - assert: + expr: "visibleMessages.length === 1 && visiblePartialText.includes(config.partialMarker) && visibleMessages[0].text !== visiblePartialText && visibleMessages[0].text.trim().length > 0" + message: + expr: "`expected one Telegram message edited away from the visible partial; saw ${visibleMessages.length}: initial=${visiblePartialText} final=${visibleMessages.map((message) => message.text).join(' | ')}`" + detailsExpr: "JSON.stringify({ count: visibleMessages.length, initialText: visiblePartialText, finalText: visibleMessages.map((message) => message.text).join(' | ') })" + + - name: sends one fallback when no partial became visible + actions: + - resetTransport: true + - set: unsentOutboundStart + value: + expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').length" + - sendInbound: + conversation: { id: telegram-partial-failure-room, kind: group } + senderId: qa-telegram-recovery-operator + senderName: Telegram Recovery Operator + text: "@openclaw Telegram unsent failure QA check" + - waitForOutbound: + conversation: { id: telegram-partial-failure-room, kind: group } + sinceIndex: { ref: unsentOutboundStart } + timeoutMs: 75000 + saveAs: unsentReply + - call: sleep + args: [4000] + - set: unsentMessages + value: + expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').slice(unsentOutboundStart)" + - assert: + expr: "unsentMessages.length === 1 && unsentReply.text.trim().length > 0 && unsentMessages[0].text.trim().length > 0 && !unsentMessages[0].text.includes(config.partialMarker)" + message: + expr: "`expected one non-empty Telegram fallback without partial text; saw ${unsentMessages.length}: ${unsentMessages.map((message) => message.text).join(' | ')}`" + detailsExpr: "JSON.stringify({ count: unsentMessages.length, text: unsentMessages.map((message) => message.text).join(' | ') })" diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts b/src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts index acee5e676f15..f5558f6d096d 100644 --- a/src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt-stream-finalize.test.ts @@ -12,6 +12,7 @@ vi.mock("./attempt-stream-settle.js", () => ({ settleEmbeddedAttemptStream: mocks.settleStream, })); +import { createSubscribedSessionHarness } from "../../embedded-agent-subscribe.e2e-harness.js"; import { SessionManager } from "../../sessions/index.js"; import { finalizeEmbeddedAttemptStreamPhase } from "./attempt-stream-finalize.js"; @@ -102,6 +103,66 @@ beforeEach(() => { }); describe("finalizeEmbeddedAttemptStreamPhase", () => { + it("does not settle a provider failure before partial presentation finishes", async () => { + let resolvePartial: (() => void) | undefined; + const onPartialReply = vi.fn( + () => + new Promise((resolve) => { + resolvePartial = resolve; + }), + ); + const { emit, subscription } = createSubscribedSessionHarness({ + runId: "run-partial-provider-failure", + onBeforeTerminalDelivery: async () => undefined, + onPartialReply, + }); + const failedAssistant = { + role: "assistant", + content: [{ type: "text", text: "partial answer" }], + stopReason: "error", + errorMessage: "provider failed after partial", + provider: "test-provider", + model: "test-model", + }; + emit({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { type: "text_delta", delta: "partial answer" }, + }); + emit({ type: "message_end", message: failedAssistant }); + emit({ type: "agent_end", messages: [failedAssistant], willRetry: false }); + + const fixture = createFixture({ + waitForPendingEvents: subscription.waitForPendingEvents, + getBeforeAgentFinalizeRevisionReason: () => undefined, + }); + mocks.settleStream.mockResolvedValue({ + promptError: new Error("provider failed after partial"), + promptErrorSource: "prompt", + timedOutDuringCompaction: false, + compactionOccurredThisAttempt: false, + messagesSnapshot: [failedAssistant], + sessionIdUsed: "session-1", + lastAssistant: failedAssistant, + currentAttemptAssistant: failedAssistant, + currentAttemptCompletedAssistant: failedAssistant, + attemptUsage: undefined, + cacheBreak: null, + lastCallUsage: undefined, + promptCache: undefined, + }); + mocks.completeAfterTurn.mockResolvedValue({ sessionIdUsed: "session-1" }); + + const finalize = finalizeEmbeddedAttemptStreamPhase(fixture.input); + await vi.waitFor(() => expect(onPartialReply).toHaveBeenCalledOnce()); + await Promise.resolve(); + expect(mocks.settleStream).not.toHaveBeenCalled(); + + resolvePartial?.(); + await finalize; + expect(mocks.settleStream).toHaveBeenCalledOnce(); + }); + it("rewinds the exact rejected branch before the hidden retry can choose NO_REPLY", async () => { const sessionManager = SessionManager.inMemory(); const promptId = sessionManager.appendMessage({ diff --git a/src/agents/embedded-agent-subscribe.partial-reply-lifecycle.test.ts b/src/agents/embedded-agent-subscribe.partial-reply-lifecycle.test.ts new file mode 100644 index 000000000000..e4106da1bd0d --- /dev/null +++ b/src/agents/embedded-agent-subscribe.partial-reply-lifecycle.test.ts @@ -0,0 +1,99 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const logger = vi.hoisted(() => ({ + debug: vi.fn(), + error: vi.fn(), + fatal: vi.fn(), + info: vi.fn(), + isEnabled: vi.fn(() => false), + trace: vi.fn(), + warn: vi.fn(), +})); + +vi.mock("../logging/subsystem.js", () => ({ + createSubsystemLogger: () => logger, +})); + +import { createSubscribedSessionHarness } from "./embedded-agent-subscribe.e2e-harness.js"; + +function emitPartialThenProviderFailure(emit: (event: unknown) => void): void { + emit({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { type: "text_delta", delta: "partial answer" }, + }); + const failedAssistant = { + role: "assistant", + content: [{ type: "text", text: "partial answer" }], + stopReason: "error", + errorMessage: "provider failed after partial", + provider: "test-provider", + model: "test-model", + }; + emit({ type: "message_end", message: failedAssistant }); + emit({ type: "agent_end", messages: [failedAssistant], willRetry: false }); +} + +describe("subscribeEmbeddedAgentSession partial reply lifecycle", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("joins a partial reply task created while terminal events settle", async () => { + let resolvePartial: (() => void) | undefined; + const onPartialReply = vi.fn( + () => + new Promise((resolve) => { + resolvePartial = resolve; + }), + ); + const { emit, subscription } = createSubscribedSessionHarness({ + runId: "run-partial-provider-failure", + onBeforeTerminalDelivery: async () => undefined, + onPartialReply, + }); + + emitPartialThenProviderFailure(emit); + let settled = false; + const settlement = subscription.waitForPendingEvents().then(() => { + settled = true; + }); + + await vi.waitFor(() => expect(onPartialReply).toHaveBeenCalledOnce()); + await Promise.resolve(); + expect(settled).toBe(false); + + resolvePartial?.(); + await settlement; + expect(settled).toBe(true); + }); + + it("contains and logs a rejected partial reply after unsubscribe", async () => { + const callbackError = new Error("draft send rejected"); + let rejectPartial: ((reason: unknown) => void) | undefined; + const onPartialReply = vi.fn( + () => + new Promise((_resolve, reject) => { + rejectPartial = reject; + }), + ); + const { emit, subscription } = createSubscribedSessionHarness({ + runId: "run-partial-rejection", + onPartialReply, + }); + + emit({ + type: "message_update", + message: { role: "assistant" }, + assistantMessageEvent: { type: "text_delta", delta: "partial answer" }, + }); + + await vi.waitFor(() => expect(onPartialReply).toHaveBeenCalledOnce()); + subscription.unsubscribe(); + rejectPartial?.(callbackError); + await expect(subscription.waitForPendingEvents()).resolves.toBeUndefined(); + expect(logger.warn).toHaveBeenCalledWith( + `assistant partial reply callback failed: ${String(callbackError)}`, + ); + }); +}); diff --git a/src/agents/embedded-agent-subscribe.ts b/src/agents/embedded-agent-subscribe.ts index 545f205826c0..d509742fb27f 100644 --- a/src/agents/embedded-agent-subscribe.ts +++ b/src/agents/embedded-agent-subscribe.ts @@ -308,6 +308,7 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess const pendingMessagingTexts = state.pendingMessagingTexts; const pendingMessagingTargets = state.pendingMessagingTargets; const pendingBlockReplyTasks = new Set>(); + const pendingPartialReplyTasks = new Set>(); const replyDirectiveAccumulator = createStreamingDirectiveAccumulator(); const partialReplyDirectiveAccumulator = createStreamingDirectiveAccumulator(); const shouldAllowSilentTurnText = (text: string | undefined) => @@ -333,11 +334,22 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess }); } if (delivery.emitPartialReply && params.onPartialReply && state.shouldEmitPartialReplies) { - runBestEffortCallback({ - label: "assistant partial reply", - log, - callback: () => params.onPartialReply?.(data), - }); + try { + const maybeTask = params.onPartialReply(data); + if (isPromiseLike(maybeTask)) { + const task = Promise.resolve(maybeTask) + .then(() => undefined) + .catch((error: unknown) => { + log.warn(`assistant partial reply callback failed: ${String(error)}`); + }); + pendingPartialReplyTasks.add(task); + void task.finally(() => { + pendingPartialReplyTasks.delete(task); + }); + } + } catch (error) { + log.warn(`assistant partial reply callback failed: ${String(error)}`); + } } }; const emitAssistantStreamData = ( @@ -1585,7 +1597,16 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess getCompactionCount: () => compactionCount, getLastCompactionTokensAfter: () => state.lastCompactionTokensAfter, getAssistantTurnCount: () => state.assistantTurnCount, - waitForPendingEvents: () => state.pendingEventChain ?? Promise.resolve(), + waitForPendingEvents: async () => { + // Partial presentation stays concurrent with provider events, but terminal + // settlement must observe callbacks launched while the event chain drains. + while (state.pendingEventChain || pendingPartialReplyTasks.size > 0) { + await Promise.allSettled([ + ...(state.pendingEventChain ? [state.pendingEventChain] : []), + ...pendingPartialReplyTasks, + ]); + } + }, getItemLifecycle: () => ({ startedCount: state.itemStartedCount, completedCount: state.itemCompletedCount,