diff --git a/extensions/discord/src/shared-interactive.test.ts b/extensions/discord/src/shared-interactive.test.ts index 0318e51b9f5b..642627a6d6df 100644 --- a/extensions/discord/src/shared-interactive.test.ts +++ b/extensions/discord/src/shared-interactive.test.ts @@ -223,6 +223,30 @@ describe("buildDiscordInteractiveComponents", () => { expect(buildDiscordPresentationComponents(presentation)).toBeUndefined(); }); + it("preserves authored block order around controls", () => { + expect( + buildDiscordPresentationComponents({ + blocks: [ + { type: "text", text: "First" }, + { + type: "buttons", + buttons: [{ label: "Approve", value: "approve", style: "success" }], + }, + { type: "text", text: "Last" }, + ], + }), + ).toEqual({ + blocks: [ + { type: "text", text: "First" }, + { + type: "actions", + buttons: [{ label: "Approve", style: "success", callbackData: "approve" }], + }, + { type: "text", text: "Last" }, + ], + }); + }); + it("renders typed approvals as actionable transport-private Discord controls", () => { const rendered = buildDiscordPresentationComponents({ blocks: [ diff --git a/extensions/discord/src/shared-interactive.ts b/extensions/discord/src/shared-interactive.ts index 03809b68b753..a6ce85a64caa 100644 --- a/extensions/discord/src/shared-interactive.ts +++ b/extensions/discord/src/shared-interactive.ts @@ -229,8 +229,6 @@ export function buildDiscordPresentationComponents( blocks.push({ type: "separator" }); continue; } - } - for (const block of presentation.blocks) { if (block.type === "buttons") { appendDiscordButtonBlocks(blocks, block.buttons); continue; diff --git a/extensions/qa-lab/src/cli.runtime.test.ts b/extensions/qa-lab/src/cli.runtime.test.ts index 0ce9a6207a5e..1974e3d815e0 100644 --- a/extensions/qa-lab/src/cli.runtime.test.ts +++ b/extensions/qa-lab/src/cli.runtime.test.ts @@ -787,6 +787,64 @@ describe("qa cli runtime", () => { expectWriteContains(stdoutWrite, "QA run profile: all; categories: 1; scenarios:"); }); + it.each([ + { + label: "implicit profile membership", + scenarioIds: undefined, + expectedExitCode: undefined, + explicitScenarioSelection: false, + }, + { + label: "explicit profile selection", + scenarioIds: ["runtime-tool-image-generate"], + expectedExitCode: 1, + explicitScenarioSelection: true, + }, + ])( + "keeps optional skips $label blocking semantics", + async ({ scenarioIds, expectedExitCode, explicitScenarioSelection }) => { + const priorExitCode = process.exitCode; + process.exitCode = undefined; + const optionalScenario = { + name: "Runtime tool fixture — image_generate", + status: "skip" as const, + details: "image_generate mock provider report-only: tool unavailable", + }; + await fs.writeFile( + suiteSummaryPath, + JSON.stringify({ + counts: { total: 2, passed: 1, failed: 0, skipped: 1 }, + scenarios: [QA_PASSING_SUITE_SCENARIO, optionalScenario], + }), + "utf8", + ); + runQaSuite.mockResolvedValueOnce( + flowSuiteRuntimeResult({ + reportPath: suiteReportPath, + summaryPath: suiteSummaryPath, + scenarios: [QA_PASSING_SUITE_SCENARIO, optionalScenario], + }), + ); + + try { + await runQaProfileCommand({ + repoRoot: "/tmp/openclaw-repo", + profile: "all", + surface: "media", + category: "media.media-generation", + providerMode: "mock-openai", + scenarioIds, + }); + expect(process.exitCode).toBe(expectedExitCode); + expect(mockFirstObjectArg(runQaSuite).adapterOptions).toMatchObject({ + explicitScenarioSelection, + }); + } finally { + process.exitCode = priorExitCode; + } + }, + ); + it("filters QA-channel-pinned scenarios from an implicit Crabline smoke profile", async () => { runQaSuite.mockImplementationOnce(async () => { await fs.writeFile(suiteEvidencePath, JSON.stringify(makeQaEvidence()), "utf8"); diff --git a/extensions/qa-lab/src/cli.runtime.ts b/extensions/qa-lab/src/cli.runtime.ts index df70d63e1149..8263eb639f13 100644 --- a/extensions/qa-lab/src/cli.runtime.ts +++ b/extensions/qa-lab/src/cli.runtime.ts @@ -713,6 +713,7 @@ export async function runQaProfileCommand(opts: QaProfileCommandOptions) { fastMode: opts.fastMode, failFast: opts.failFast, scenarioIds: scenarios.map((scenario) => scenario.id), + explicitScenarioSelection: requestedScenarioIds.length > 0, concurrency: opts.concurrency, allowFailures: opts.allowFailures, channelDriver: profileReport.channelDriver, @@ -815,7 +816,9 @@ function resolveQaReportOnlyOptionalScenarioNames(params: { scenarioIds: readonly string[]; explicitScenarioSelection?: boolean; }): ReadonlySet | undefined { - if (params.explicitScenarioSelection || params.scenarioIds.length > 0) { + const explicitScenarioSelection = + params.explicitScenarioSelection ?? params.scenarioIds.length > 0; + if (explicitScenarioSelection) { return undefined; } return resolveQaReportOnlyOptionalScenarioNamesFromCatalog(readQaScenarioPack().scenarios); 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 0036d0a76b75..9425f6293a0e 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.test.ts @@ -64,12 +64,25 @@ const QA_COMPACTION_RETRY_CODE_MODE_WRITE_RESULT = { const QA_COMPACTION_RETRY_PROMPT = "Compaction retry mutating tool check. Current durable context marker: QA-COMPACTION-DURABLE-MARKER. Create compaction-retry-summary.txt."; const QA_COMPACTION_RETRY_OVERFLOW_PADDING = "x".repeat(300_000); +const QA_COMPACTION_RETRY_HISTORICAL_PHRASE = "post-marker historical user block"; const QA_COMPACTION_EMPTY_RECOVERY_SUMMARY_MARKER = "QA-COMPACTION-EMPTY-RECOVERED-SUMMARY"; const QA_COMPACTION_REASONING_RECOVERY_SUMMARY_MARKER = "QA-COMPACTION-REASONING-RECOVERED-SUMMARY"; +const QA_COMPACTION_SUMMARY_HEADINGS = [ + "## Decisions", + "## Open TODOs", + "## Constraints/Rules", + "## Pending user asks", + "## Exact identifiers", +] as const; const QA_COMPACTION_SUMMARY_INSTRUCTIONS = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI assistant, then produce a structured summary following the exact format specified. Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.`; +function expectCurrentCompactionSummaryHeadings(summary: string) { + expect(summary.match(/^## .+$/gmu)).toEqual(QA_COMPACTION_SUMMARY_HEADINGS); + expect(summary).not.toContain("## Goal"); +} + afterEach(async () => { while (cleanups.length > 0) { await cleanups.pop()?.(); @@ -2395,7 +2408,7 @@ describe("qa mock openai server", () => { expect(response.status).toBe(200); const summary = outputText(await response.json()); - expect(summary).toContain("## Goal"); + expectCurrentCompactionSummaryHeadings(summary); expect(summary).not.toContain("QA-COMPACTION-DURABLE-MARKER"); expect(summary).not.toContain("QA-COMPACTION-BULKY-HISTORICAL-MARKER"); const requests = requireArray( @@ -2570,9 +2583,16 @@ describe("qa mock openai server", () => { expect(response.status).toBe(200); const body = requireRecord(await response.json(), "Anthropic summary response"); - expect(requireArray(body.content, "content")).toContainEqual( - expect.objectContaining({ type: "text", text: expect.stringContaining("## Goal") }), + const content = requireArray(body.content, "content"); + expect(content).toContainEqual( + expect.objectContaining({ type: "text", text: expect.stringContaining("## Decisions") }), ); + const summaryText = requireRecord(content[0], "summary content").text; + expect(typeof summaryText).toBe("string"); + if (typeof summaryText !== "string") { + throw new TypeError("Anthropic summary content text must be a string"); + } + expectCurrentCompactionSummaryHeadings(summaryText); expect(await getJson(server, "/debug/last-request")).toMatchObject({ requestKind: "compaction-summary", outcome: "success", @@ -2588,20 +2608,43 @@ describe("qa mock openai server", () => { "\n[Chunk 1 - oldest messages]\nunrelated historical context\n\n\nAdditional focus: preserve exact identifiers.", }); const genericChunkSummary = outputText(genericChunkPayload); - expect(genericChunkSummary).toContain("## Goal"); + expectCurrentCompactionSummaryHeadings(genericChunkSummary); expect(genericChunkSummary).not.toContain("QA-COMPACTION-DURABLE-MARKER"); const scenarioChunkPayload = await expectOpenAiNonStreamingResponsesJson(server, { model: "gpt-5.6-luna", instructions: QA_COMPACTION_SUMMARY_INSTRUCTIONS, - input: - "\n[Chunk 2 - most recent messages]\nQA-COMPACTION-BULKY-HISTORICAL-MARKER\n\n\nAdditional focus: preserve exact identifiers.", + input: ` +[Chunk 2 - most recent messages] +QA-COMPACTION-BULKY-HISTORICAL-MARKER ${QA_COMPACTION_RETRY_HISTORICAL_PHRASE} 10 + + +Additional focus: preserve exact identifiers.`, }); const scenarioChunkSummary = outputText(scenarioChunkPayload); - expect(scenarioChunkSummary).toContain("## Goal"); + expectCurrentCompactionSummaryHeadings(scenarioChunkSummary); + expect(scenarioChunkSummary).toContain(QA_COMPACTION_RETRY_HISTORICAL_PHRASE); expect(scenarioChunkSummary).not.toContain("QA-COMPACTION-DURABLE-MARKER"); expect(scenarioChunkSummary).not.toContain("QA-COMPACTION-BULKY-HISTORICAL-MARKER"); + const historyMergePayload = await expectOpenAiNonStreamingResponsesJson(server, { + model: "gpt-5.6-luna", + instructions: QA_COMPACTION_SUMMARY_INSTRUCTIONS, + input: ` +[Chunk 1 - oldest messages] +${genericChunkSummary} +[Chunk 2 - most recent messages] +${scenarioChunkSummary} + + +Update and merge these partial structured summaries.`, + }); + const historyMergedSummary = outputText(historyMergePayload); + expectCurrentCompactionSummaryHeadings(historyMergedSummary); + expect(historyMergedSummary).toContain(QA_COMPACTION_RETRY_HISTORICAL_PHRASE); + expect(historyMergedSummary).not.toContain("QA-COMPACTION-DURABLE-MARKER"); + expect(historyMergedSummary).not.toContain("QA-COMPACTION-BULKY-HISTORICAL-MARKER"); + const durableChunkPayload = await expectOpenAiNonStreamingResponsesJson(server, { model: "gpt-5.6-luna", instructions: QA_COMPACTION_SUMMARY_INSTRUCTIONS, @@ -2612,6 +2655,7 @@ Retain QA-COMPACTION-DURABLE-MARKER for the active task. Additional focus: preserve QA-COMPACTION-DURABLE-MARKER.`, }); const durableChunkSummary = outputText(durableChunkPayload); + expectCurrentCompactionSummaryHeadings(durableChunkSummary); expect(durableChunkSummary).toContain("QA-COMPACTION-DURABLE-MARKER"); const promptOnlyChunkPayload = await expectOpenAiNonStreamingResponsesJson(server, { @@ -2624,6 +2668,7 @@ Compaction retry mutating tool check. Create compaction-retry-summary.txt. Additional focus: preserve current work.`, }); const promptOnlyChunkSummary = outputText(promptOnlyChunkPayload); + expectCurrentCompactionSummaryHeadings(promptOnlyChunkSummary); expect(promptOnlyChunkSummary).not.toContain("QA-COMPACTION-DURABLE-MARKER"); const currentChunkPayload = await expectOpenAiNonStreamingResponsesJson(server, { @@ -2637,6 +2682,7 @@ Create compaction-retry-summary.txt. Additional focus: preserve current work.`, }); const currentChunkSummary = outputText(currentChunkPayload); + expectCurrentCompactionSummaryHeadings(currentChunkSummary); expect(currentChunkSummary).toContain("QA-COMPACTION-DURABLE-MARKER"); const mergePayload = await expectOpenAiNonStreamingResponsesJson(server, { @@ -2657,7 +2703,9 @@ ${durableChunkSummary} Update and merge these partial structured summaries.`, }); - expect(outputText(mergePayload)).toContain("QA-COMPACTION-DURABLE-MARKER"); + const mergedSummary = outputText(mergePayload); + expectCurrentCompactionSummaryHeadings(mergedSummary); + expect(mergedSummary).toContain("QA-COMPACTION-DURABLE-MARKER"); const unrelatedPayload = await expectOpenAiNonStreamingResponsesJson(server, { model: "gpt-5.6-luna", @@ -2665,14 +2713,15 @@ Update and merge these partial structured summaries.`, input: "\na later unrelated scenario\n\n\nCreate a structured summary.", }); - expect(outputText(unrelatedPayload)).toContain("## Goal"); - expect(outputText(unrelatedPayload)).not.toContain("QA-COMPACTION-DURABLE-MARKER"); + const unrelatedSummary = outputText(unrelatedPayload); + expectCurrentCompactionSummaryHeadings(unrelatedSummary); + expect(unrelatedSummary).not.toContain("QA-COMPACTION-DURABLE-MARKER"); const requests = requireArray( await getJson(server, "/debug/requests"), "compaction requests", ).map((request) => requireRecord(request, "compaction request")); - expect(requests).toHaveLength(7); + expect(requests).toHaveLength(8); expect( requests.every( (request) => @@ -2705,6 +2754,7 @@ Update and merge these partial structured summaries.`, ], }); const compactedSummary = outputText(summaryPayload); + expectCurrentCompactionSummaryHeadings(compactedSummary); expect(compactedSummary).toContain("QA-COMPACTION-DURABLE-MARKER"); const writePlan = await expectOpenAiStreamingResponsesText(server, { diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index 13299a405f94..f4e68efb9d34 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -186,61 +186,59 @@ const QA_COMPACTION_RETRY_OVERFLOW_THRESHOLD_BYTES = 256 * 1024; const QA_COMPACTION_OUTPUT_RECOVERY_OVERFLOW_THRESHOLD_BYTES = 96 * 1024; const QA_COMPACTION_RETRY_DURABLE_MARKER = "QA-COMPACTION-DURABLE-MARKER"; const QA_COMPACTION_RETRY_BULKY_MARKER = "QA-COMPACTION-BULKY-HISTORICAL-MARKER"; +const QA_COMPACTION_RETRY_HISTORICAL_PHRASE = "post-marker historical user block"; const QA_COMPACTION_EMPTY_OUTPUT_ONCE_MARKER_RE = /\bQA-COMPACTION-EMPTY-OUTPUT-ONCE-[A-Za-z0-9_-]+\b/u; const QA_COMPACTION_REASONING_ONLY_OUTPUT_ONCE_MARKER_RE = /\bQA-COMPACTION-REASONING-ONLY-OUTPUT-ONCE-[A-Za-z0-9_-]+\b/u; const QA_COMPACTION_EMPTY_RECOVERY_SUMMARY_MARKER = "QA-COMPACTION-EMPTY-RECOVERED-SUMMARY"; const QA_COMPACTION_REASONING_RECOVERY_SUMMARY_MARKER = "QA-COMPACTION-REASONING-RECOVERED-SUMMARY"; -const QA_COMPACTION_RETRY_SUMMARY = `## Goal -Complete the compaction retry mutating tool check. +const QA_COMPACTION_RETRY_SUMMARY = `## Decisions +- Continue the compaction retry from durable context without replaying a completed mutation. -## Constraints & Preferences +## Open TODOs +- Write compaction-retry-summary.txt exactly once. +- Return the final replay-safety marker. + +## Constraints/Rules - Preserve ${QA_COMPACTION_RETRY_DURABLE_MARKER}. +- Write exactly: Replay safety: unsafe after write. -## Progress -### Done -- [x] Historical context compacted after overflow. +## Pending user asks +- Create compaction-retry-summary.txt, then reply exactly: Protocol note: replay unsafe after write. -### In Progress -- [ ] Write compaction-retry-summary.txt exactly once. +## Exact identifiers +- ${QA_COMPACTION_RETRY_DURABLE_MARKER} +- compaction-retry-summary.txt`; +const QA_COMPACTION_RETRY_HISTORICAL_SUMMARY = `## Decisions +- Preserve the latest ${QA_COMPACTION_RETRY_HISTORICAL_PHRASE} context through staged compaction. -### Blocked -- (none) +## Open TODOs +- Continue summarizing the ${QA_COMPACTION_RETRY_HISTORICAL_PHRASE} sequence. -## Key Decisions -- **Retry once**: Continue from compacted context without replaying a completed mutation. +## Constraints/Rules +- Keep historical content distinct from live task state. +- Do not invent durable context absent from the summarized history. -## Next Steps -1. Write the required file. -2. Return the final replay-safety marker. +## Pending user asks +- Retain the ${QA_COMPACTION_RETRY_HISTORICAL_PHRASE} details. -## Critical Context -- ${QA_COMPACTION_RETRY_DURABLE_MARKER}`; -const QA_GENERIC_COMPACTION_SUMMARY = `## Goal -Preserve the active conversation context. +## Exact identifiers +- None captured.`; +const QA_GENERIC_COMPACTION_SUMMARY = `## Decisions +- Continue from the summary without restarting completed work. -## Constraints & Preferences +## Open TODOs +- Continue the active task. + +## Constraints/Rules - Keep current requirements and identifiers. -## Progress -### Done -- [x] Historical context summarized. +## Pending user asks +- Continue the active task from the retained context. -### In Progress -- [ ] Continue the active task. - -### Blocked -- (none) - -## Key Decisions -- **Continue from summary**: Do not restart completed work. - -## Next Steps -1. Continue the active task from the retained context. - -## Critical Context -- Refer to the retained recent turns for current task details.`; +## Exact identifiers +- None captured.`; const QA_COMPACTION_OUTPUT_RECOVERY_SUMMARY = `## Decisions - Retry the typed compaction-summary fault at the compaction owner. @@ -834,7 +832,10 @@ async function buildResponsesPayload( return buildAssistantEvents( hasCompactionRetryDurableContext ? QA_COMPACTION_RETRY_SUMMARY - : resolveCompactionRecoverySummary(allInputText), + : allInputText.includes(QA_COMPACTION_RETRY_BULKY_MARKER) || + allInputText.includes(QA_COMPACTION_RETRY_HISTORICAL_PHRASE) + ? QA_COMPACTION_RETRY_HISTORICAL_SUMMARY + : resolveCompactionRecoverySummary(allInputText), ); } if ( diff --git a/extensions/qa-lab/src/scenario-catalog-compaction.test.ts b/extensions/qa-lab/src/scenario-catalog-compaction.test.ts index 3c6859ea8983..b32c7a91be15 100644 --- a/extensions/qa-lab/src/scenario-catalog-compaction.test.ts +++ b/extensions/qa-lab/src/scenario-catalog-compaction.test.ts @@ -71,6 +71,7 @@ describe("qa compaction scenario catalog", () => { const postWriteContinuationsExpr = readSetExpression("postWriteContinuations"); const writeTranscriptToolCallIdExpr = readSetExpression("writeTranscriptToolCallId"); const continuationChainExpr = readSetExpression("continuationChain"); + const compactionSummaryRequestsExpr = readSetExpression("compactionSummaryRequests"); const continuationAssertIndex = actionIndex((action) => readFlowAssertExpression(action).includes("continuationChain.valid === true"), ); @@ -97,6 +98,10 @@ describe("qa compaction scenario catalog", () => { const terminalEvidenceAssertExpr = readAssertExpression( "terminalContinuations[0].providerVariant === 'openai'", ); + const compactionSummaryAssertExpr = readAssertExpression( + "compactionSummaryRequests.length > 0", + ); + const noQualityRetryAssertExpr = readAssertExpression("Previous summary failed quality checks"); const knownGap = "known-harness-gap compaction-retry-mutating-tool: provider-error recovery does not invoke Codex native compaction; native token-threshold compaction needs a separate scenario."; @@ -274,11 +279,19 @@ describe("qa compaction scenario catalog", () => { expect(flow).not.toContain("config.expectedOpenClawToolResult"); expect(flow).not.toContain("String(request.toolOutput ?? '').includes(`---"); expect(flow).not.toContain("String(request.toolOutput ?? '').includes(`+++"); - expect(flow).toContain( - "compactionSummaryRequests.length === 1 && compactionSummaryRequests[0].outcome === 'success' && compactionSummaryRequests[0].plannedToolName === undefined && compactionSummaryRequests[0].toolOutputStructuredError !== true", + expect(compactionSummaryRequestsExpr).toContain("request.requestKind === 'compaction-summary'"); + expect(compactionSummaryAssertExpr).toContain("compactionSummaryRequests.length > 0"); + expect(compactionSummaryAssertExpr).toContain( + "request.cursor > overflowRequest.cursor && request.cursor < writeRequest.cursor", ); - expect(flow).not.toContain("compactionSummaryRequests.every("); - expect(flow).not.toContain("compactionSummaryRequests.length >= 1"); + expect(compactionSummaryAssertExpr).toContain("request.outcome === 'success'"); + expect(compactionSummaryAssertExpr).toContain("request.plannedToolName === undefined"); + expect(compactionSummaryAssertExpr).toContain("request.toolOutputStructuredError !== true"); + expect(noQualityRetryAssertExpr).toContain("compactionSummaryRequests.every"); + expect(noQualityRetryAssertExpr).toContain( + "!String(request.allInputText ?? '').includes('Previous summary failed quality checks')", + ); + expect(flow).not.toContain("compactionSummaryRequests.length === 1"); expect(flow).toContain( "writeRequest.rawByteLength < config.overflowThresholdBytes && writeRequest.rawByteLength < overflowRequest.rawByteLength", ); @@ -292,6 +305,7 @@ describe("qa compaction scenario catalog", () => { expect(flow).toContain('"set":"requestEvidence"'); expect(flow).toContain("durable: String(request.allInputText ?? '')"); expect(flow).toContain("bulky: String(request.allInputText ?? '')"); + expect(flow).toContain("qualityRetry: String(request.allInputText ?? '')"); expect(flow).toContain("inputChars: String(request.allInputText ?? '').length"); expect(flow).toContain( "resolvedWireTool: request.plannedWireToolName ?? request.plannedToolName ?? null", diff --git a/extensions/qa-lab/src/scenario-catalog.test.ts b/extensions/qa-lab/src/scenario-catalog.test.ts index db9119f886c7..940a22f7c86f 100644 --- a/extensions/qa-lab/src/scenario-catalog.test.ts +++ b/extensions/qa-lab/src/scenario-catalog.test.ts @@ -792,6 +792,24 @@ describe("qa scenario catalog", () => { expect(heartbeatFlow).not.toContain("waitForNoOutbound"); }); + it.each([ + "inbound-media-store-audio-transcription", + "active-memory-cold-first-turn-trigger-recall", + "compaction-empty-response-recovery", + "compaction-reasoning-only-recovery", + "compaction-retry-mutating-tool", + "empty-response-recovery-replay-safe-read", + "empty-response-retry-budget-exhausted", + "reasoning-only-no-auto-retry-after-write", + "reasoning-only-recovery-replay-safe-read", + ])("keeps strict mock-only scenario %s on the mock-openai lane", (scenarioId) => { + const config = readQaScenarioExecutionConfig(scenarioId) as + | { requiredProviderMode?: string } + | undefined; + + expect(config?.requiredProviderMode).toBe("mock-openai"); + }); + it("includes the thinking slash model remap scenario", () => { const scenario = readQaScenarioById("thinking-slash-model-remap"); const config = readQaScenarioExecutionConfig("thinking-slash-model-remap") as diff --git a/extensions/qa-lab/src/suite-runtime-agent-process.integration.test.ts b/extensions/qa-lab/src/suite-runtime-agent-process.integration.test.ts index c5ab297f2416..374b50bbad93 100644 --- a/extensions/qa-lab/src/suite-runtime-agent-process.integration.test.ts +++ b/extensions/qa-lab/src/suite-runtime-agent-process.integration.test.ts @@ -72,4 +72,46 @@ describe("qa suite runtime CLI integration", () => { status: "ok", }); }); + + it("retains real child output when the qa cli times out", async () => { + const repoRoot = await mkdtemp(path.join(os.tmpdir(), "qa-cli-timeout-repo-")); + const tempRoot = await mkdtemp(path.join(os.tmpdir(), "qa-cli-timeout-runtime-")); + cleanups.push(async () => { + await rm(repoRoot, { recursive: true, force: true }); + await rm(tempRoot, { recursive: true, force: true }); + }); + const distDir = path.join(repoRoot, "dist"); + await mkdir(distDir, { recursive: true }); + await writeFile( + path.join(distDir, "index.js"), + [ + 'process.stdout.write("timeout stdout marker\\n");', + 'process.stderr.write("timeout stderr marker\\n");', + "setInterval(() => {}, 60_000);", + "", + ].join("\n"), + "utf8", + ); + + const error = await runQaCli( + { + repoRoot, + gateway: { + tempRoot, + runtimeEnv: process.env, + }, + primaryModel: "openai/gpt-5.6-luna", + alternateModel: "openai/gpt-5.6-luna", + providerMode: "mock-openai", + } as never, + ["qa", "suite"], + { timeoutMs: 1_000 }, + ).catch((value: unknown) => value); + + expect(error).toMatchObject({ code: "qa_cli_timeout" }); + const message = error instanceof Error ? error.message : String(error); + expect(message).toContain("qa cli timed out: openclaw qa suite"); + expect(message).toContain("stdout:\ntimeout stdout marker"); + expect(message).toContain("stderr:\ntimeout stderr marker"); + }); }); diff --git a/extensions/qa-lab/src/suite-runtime-agent-process.test.ts b/extensions/qa-lab/src/suite-runtime-agent-process.test.ts index ddbfebc5d091..047efc5d5256 100644 --- a/extensions/qa-lab/src/suite-runtime-agent-process.test.ts +++ b/extensions/qa-lab/src/suite-runtime-agent-process.test.ts @@ -174,6 +174,7 @@ describe("qa suite runtime agent process helpers", () => { it.runIf(process.platform !== "win32")("kills timed-out qa cli process groups", async () => { const killSpy = vi.spyOn(process, "kill").mockImplementation(() => true); + vi.useFakeTimers(); try { const child = createSpawnedProcess({ pid: 12345 }); const { pending } = startMockQaCli({ @@ -181,15 +182,37 @@ describe("qa suite runtime agent process helpers", () => { child, options: { timeoutMs: 1 }, }); - const timeoutAssertion = expect(pending).rejects.toThrow( - "qa cli timed out: openclaw qa suite", + const errorPromise = pending.catch((value: unknown) => value); + await Promise.resolve(); + expect(spawnMock).toHaveBeenCalledTimes(1); + child.stdout.emit( + "data", + Buffer.from( + `stdout-head-marker\n${"x".repeat(QA_CHILD_STDOUT_MAX_BYTES)}\nstdout-tail-marker`, + ), ); + child.stderr.emit( + "data", + Buffer.from( + `stderr-head-marker\n${"x".repeat(QA_CHILD_STDERR_TAIL_BYTES)}\nstderr-tail-marker`, + ), + ); + await vi.advanceTimersByTimeAsync(1); - await waitForSpawnCount(1); - await timeoutAssertion; + const error = await errorPromise; + expect(error).toMatchObject({ code: "qa_cli_timeout" }); + const message = error instanceof Error ? error.message : String(error); + expect(message).toContain("qa cli timed out: openclaw qa suite"); + expect(message).toContain("stdout:\n[qa cli stdout truncated to last"); + expect(message).toContain("stdout-tail-marker"); + expect(message).not.toContain("stdout-head-marker"); + expect(message).toContain("stderr:\n[qa cli stderr truncated to last"); + expect(message).toContain("stderr-tail-marker"); + expect(message).not.toContain("stderr-head-marker"); expect(killSpy).toHaveBeenCalledWith(-12345, "SIGKILL"); expect(child.kill).not.toHaveBeenCalled(); } finally { + vi.useRealTimers(); killSpy.mockRestore(); } }); diff --git a/extensions/qa-lab/src/suite-runtime-agent-process.ts b/extensions/qa-lab/src/suite-runtime-agent-process.ts index 3fbba57037fb..709fde09c9a1 100644 --- a/extensions/qa-lab/src/suite-runtime-agent-process.ts +++ b/extensions/qa-lab/src/suite-runtime-agent-process.ts @@ -265,6 +265,7 @@ async function runQaCli( opts?: { timeoutMs?: number; json?: boolean; env?: NodeJS.ProcessEnv }, ) { const stdout = createQaChildOutputCapture(); + const stdoutTail = createQaChildOutputTail(); const stderr = createQaChildOutputTail(); const distEntryPath = path.join(env.repoRoot, "dist", "index.js"); const nodeExecPath = await resolveQaNodeExecPath(); @@ -281,11 +282,25 @@ async function runQaCli( const timeoutMs = resolveTimerTimeoutMs(opts?.timeoutMs, 60_000); const timeout = setTimeout(() => { signalQaCliProcessTree(child, "SIGKILL"); + const stdoutText = formatQaChildOutputTail(stdoutTail, "qa cli stdout"); + const stderrText = formatQaChildOutputTail(stderr, "qa cli stderr"); + const diagnostics = [ + stdoutText ? `stdout:\n${stdoutText}` : "", + stderrText ? `stderr:\n${stderrText}` : "", + ] + .filter(Boolean) + .join("\n"); reject( - new QaSuiteInfraError("qa_cli_timeout", `qa cli timed out: openclaw ${args.join(" ")}`), + new QaSuiteInfraError( + "qa_cli_timeout", + `qa cli timed out: openclaw ${args.join(" ")}${diagnostics ? `\n${diagnostics}` : ""}`, + ), ); }, timeoutMs); - child.stdout.on("data", (chunk) => appendQaChildOutput(stdout, chunk)); + child.stdout.on("data", (chunk) => { + appendQaChildOutput(stdout, chunk); + appendQaChildOutputTail(stdoutTail, chunk); + }); child.stderr.on("data", (chunk) => appendQaChildOutputTail(stderr, chunk)); child.once("error", (error) => { clearTimeout(timeout); diff --git a/qa/scenarios/media/inbound-media-store-audio-transcription.yaml b/qa/scenarios/media/inbound-media-store-audio-transcription.yaml index aea721e8761a..96d6dde5ad7d 100644 --- a/qa/scenarios/media/inbound-media-store-audio-transcription.yaml +++ b/qa/scenarios/media/inbound-media-store-audio-transcription.yaml @@ -36,6 +36,7 @@ scenario: timeoutMs: 90000 retryCount: 0 config: + requiredProviderMode: mock-openai conversationId: qa-media-store-audio expectedMarker: WHATSAPP_QA_AUDIO_TRANSCRIPT_OK diff --git a/qa/scenarios/memory/active-memory-cold-first-turn-trigger-recall.yaml b/qa/scenarios/memory/active-memory-cold-first-turn-trigger-recall.yaml index e639e55aa9b4..5702c30f22bc 100644 --- a/qa/scenarios/memory/active-memory-cold-first-turn-trigger-recall.yaml +++ b/qa/scenarios/memory/active-memory-cold-first-turn-trigger-recall.yaml @@ -55,6 +55,7 @@ scenario: timeoutMs: 120000 retryCount: 0 config: + requiredProviderMode: mock-openai conversationId: qa-active-memory-cold-first-turn senderId: qa-active-memory-cold-user memoryFact: "Stable QA cold first-turn movie night snack preference: lemon pepper wings with blue cheese. " diff --git a/qa/scenarios/observability/diagnostics-otel-plugin-install.yaml b/qa/scenarios/observability/diagnostics-otel-plugin-install.yaml new file mode 100644 index 000000000000..909ed74e32e8 --- /dev/null +++ b/qa/scenarios/observability/diagnostics-otel-plugin-install.yaml @@ -0,0 +1,31 @@ +title: Diagnostics OTEL managed plugin install + +scenario: + id: diagnostics-otel-plugin-install + surface: telemetry + coverage: + primary: + - observability.diagnostics-otel-plugin-install + objective: Verify the exact diagnostics-otel package works through the managed npm install, enable, restart, and export lifecycle. + successCriteria: + - The release package is packed and served by the fixture npm registry. + - The real plugins install command records npm provenance and a managed install path. + - Disable and enable persist through the CLI before a hard Gateway restart. + - Signal-specific config wins over the matching environment endpoint. + - Sampling at zero drops a completed turn, then sampling at one exports the next turn. + - The clamped flush interval exports a trace before shutdown. + - A preloaded OpenTelemetry SDK still receives spans from installed-plugin listeners. + plugins: + - diagnostics-otel + docsRefs: + - docs/gateway/opentelemetry.md + - docs/tools/plugin.md + codeRefs: + - extensions/diagnostics-otel/src/service.ts + - src/cli/plugins-cli.runtime.ts + - test/e2e/qa-lab/runtime/diagnostics-otel-install-runtime.e2e.test.ts + - test/e2e/qa-lab/runtime/otel-test-support.ts + execution: + kind: vitest + path: test/e2e/qa-lab/runtime/diagnostics-otel-install-runtime.e2e.test.ts + summary: Pack, registry-install, restart, and export through the managed diagnostics-otel package. diff --git a/qa/scenarios/observability/otel-gateway-runtime.yaml b/qa/scenarios/observability/otel-gateway-runtime.yaml new file mode 100644 index 000000000000..c92a879ea17e --- /dev/null +++ b/qa/scenarios/observability/otel-gateway-runtime.yaml @@ -0,0 +1,33 @@ +title: OTEL gateway runtime + +scenario: + id: otel-gateway-runtime + surface: telemetry + coverage: + primary: + - observability.otlp-http-traces-qa-lab + - observability.telemetry-runtime-validation + - observability.telemetry-trace + - observability.telemetry-failure-recovery + objective: Verify a real mock-provider Gateway and qa-channel run exports linked OTLP traces for normal execution and honest failed-tool recovery. + successCriteria: + - A qa-channel turn with a successful read exports linked run, tool, and terminal message-processing spans. + - A missing-file tool failure is exported as an error tool span. + - The failed-tool turn still produces an honest outbound response and a successful terminal message-processing span in the same trace. + - Exporter retry behavior is not accepted as tool-failure recovery evidence. + plugins: + - diagnostics-otel + - qa-channel + - qa-lab + docsRefs: + - docs/gateway/opentelemetry.md + - docs/concepts/qa-e2e-automation.md + codeRefs: + - extensions/diagnostics-otel/src/service-recorders-tools.ts + - extensions/diagnostics-otel/src/service-recorders-usage.ts + - test/e2e/qa-lab/runtime/otel-gateway-runtime.e2e.test.ts + - test/e2e/qa-lab/runtime/otel-test-support.ts + execution: + kind: vitest + path: test/e2e/qa-lab/runtime/otel-gateway-runtime.e2e.test.ts + summary: Run the public QA gateway, qa-channel bus, mock provider, and a decoded local OTLP receiver. diff --git a/qa/scenarios/runtime/compaction-empty-response-recovery.yaml b/qa/scenarios/runtime/compaction-empty-response-recovery.yaml index e19189ea085e..0d4264b52bb6 100644 --- a/qa/scenarios/runtime/compaction-empty-response-recovery.yaml +++ b/qa/scenarios/runtime/compaction-empty-response-recovery.yaml @@ -27,6 +27,7 @@ scenario: kind: flow summary: Reject one empty overflow-compaction summary, retry it, and verify the recovered summary persists. config: + requiredProviderMode: mock-openai faultMode: empty-output-once faultMarkerPrefix: QA-COMPACTION-EMPTY-OUTPUT-ONCE summaryMarker: QA-COMPACTION-EMPTY-RECOVERED-SUMMARY diff --git a/qa/scenarios/runtime/compaction-reasoning-only-recovery.yaml b/qa/scenarios/runtime/compaction-reasoning-only-recovery.yaml index dd2dc9b059e2..4f4af0d88eed 100644 --- a/qa/scenarios/runtime/compaction-reasoning-only-recovery.yaml +++ b/qa/scenarios/runtime/compaction-reasoning-only-recovery.yaml @@ -27,6 +27,7 @@ scenario: kind: flow summary: Reject one reasoning-only overflow-compaction summary, retry it, and verify the recovered summary persists. config: + requiredProviderMode: mock-openai faultMode: reasoning-only-output-once faultMarkerPrefix: QA-COMPACTION-REASONING-ONLY-OUTPUT-ONCE summaryMarker: QA-COMPACTION-REASONING-RECOVERED-SUMMARY diff --git a/qa/scenarios/runtime/compaction-retry-mutating-tool.yaml b/qa/scenarios/runtime/compaction-retry-mutating-tool.yaml index ce6b94690484..ef7987132b33 100644 --- a/qa/scenarios/runtime/compaction-retry-mutating-tool.yaml +++ b/qa/scenarios/runtime/compaction-retry-mutating-tool.yaml @@ -27,6 +27,7 @@ scenario: summary: Force one OpenClaw context overflow, verify persisted pruning evidence, and prove exactly-once mutation after compaction. retryCount: 0 config: + requiredProviderMode: mock-openai outputFile: compaction-retry-summary.txt promptSnippet: Compaction retry mutating tool check durableMarker: QA-COMPACTION-DURABLE-MARKER @@ -102,7 +103,7 @@ flow: expr: "await fetchJson(`${env.mock.baseUrl}/debug/requests?after=${requestCursorBefore}`)" - set: requestEvidence value: - expr: "scenarioRequests.map((request) => ({ cursor: request.cursor, kind: request.requestKind, outcome: request.outcome, code: request.errorCode ?? null, bytes: request.rawByteLength, inputChars: String(request.allInputText ?? '').length, tailBlocks: [...new Set(Array.from({ length: 16 }, (_, index) => String(index).padStart(2, '0')).filter((id) => String(request.allInputText ?? '').includes(`post-marker historical user block ${id}`)))].sort().slice(0, 16), prompt: String(request.allInputText ?? '').includes(config.promptSnippet), durable: String(request.allInputText ?? '').includes(config.durableMarker), bulky: String(request.allInputText ?? '').includes(config.bulkyMarker), tool: request.plannedToolName ?? null, resolvedWireTool: request.plannedWireToolName ?? request.plannedToolName ?? null, callId: request.plannedToolCallId ?? null, itemId: request.plannedToolItemId ?? null, transcriptId: typeof request.plannedToolItemId === 'string' && request.plannedToolItemId.length > 0 ? `${request.plannedToolCallId}|${request.plannedToolItemId}` : request.plannedToolCallId ?? null }))" + expr: "scenarioRequests.map((request) => ({ cursor: request.cursor, kind: request.requestKind, outcome: request.outcome, code: request.errorCode ?? null, bytes: request.rawByteLength, inputChars: String(request.allInputText ?? '').length, tailBlocks: [...new Set(Array.from({ length: 16 }, (_, index) => String(index).padStart(2, '0')).filter((id) => String(request.allInputText ?? '').includes(`post-marker historical user block ${id}`)))].sort().slice(0, 16), prompt: String(request.allInputText ?? '').includes(config.promptSnippet), durable: String(request.allInputText ?? '').includes(config.durableMarker), bulky: String(request.allInputText ?? '').includes(config.bulkyMarker), qualityRetry: String(request.allInputText ?? '').includes('Previous summary failed quality checks'), tool: request.plannedToolName ?? null, resolvedWireTool: request.plannedWireToolName ?? request.plannedToolName ?? null, callId: request.plannedToolCallId ?? null, itemId: request.plannedToolItemId ?? null, transcriptId: typeof request.plannedToolItemId === 'string' && request.plannedToolItemId.length > 0 ? `${request.plannedToolCallId}|${request.plannedToolItemId}` : request.plannedToolCallId ?? null }))" - set: overflowRequests value: expr: "scenarioRequests.filter((request) => request.requestKind === 'agent-initial' && request.outcome === 'error' && request.errorCode === 'context_length_exceeded' && String(request.allInputText ?? '').includes(sessionId) && String(request.allInputText ?? '').includes(config.promptSnippet) && String(request.allInputText ?? '').includes(config.durableMarker))" @@ -228,9 +229,13 @@ flow: value: expr: "scenarioRequests.filter((request) => request.requestKind === 'compaction-summary')" - assert: - expr: "compactionSummaryRequests.length === 1 && compactionSummaryRequests[0].outcome === 'success' && compactionSummaryRequests[0].plannedToolName === undefined && compactionSummaryRequests[0].toolOutputStructuredError !== true" + expr: "compactionSummaryRequests.length > 0 && compactionSummaryRequests.every((request) => request.cursor > overflowRequest.cursor && request.cursor < writeRequest.cursor && request.outcome === 'success' && request.plannedToolName === undefined && request.toolOutputStructuredError !== true)" message: - expr: "`expected exactly one successful OpenClaw compaction summary request: ${JSON.stringify(requestEvidence.filter((request) => request.kind === 'compaction-summary'))}`" + expr: "`expected successful OpenClaw compaction summary requests causally between overflow and retry: ${JSON.stringify(requestEvidence.filter((request) => request.kind === 'compaction-summary'))}`" + - assert: + expr: "compactionSummaryRequests.every((request) => !String(request.allInputText ?? '').includes('Previous summary failed quality checks'))" + message: + expr: "`compaction summary unexpectedly required quality-feedback regeneration: ${JSON.stringify(requestEvidence.filter((request) => request.kind === 'compaction-summary'))}`" - call: readRawQaSessionStore saveAs: store args: diff --git a/qa/scenarios/runtime/empty-response-recovery-replay-safe-read.yaml b/qa/scenarios/runtime/empty-response-recovery-replay-safe-read.yaml index 09cc436c3e57..5955ffdbcf89 100644 --- a/qa/scenarios/runtime/empty-response-recovery-replay-safe-read.yaml +++ b/qa/scenarios/runtime/empty-response-recovery-replay-safe-read.yaml @@ -26,6 +26,7 @@ scenario: kind: flow summary: Verify empty OpenAI turns recover after a replay-safe read. config: + requiredProviderMode: mock-openai requiredProvider: mock-openai promptSnippet: Empty response continuation QA check prompt: "Empty response continuation QA check: read QA_KICKOFF_TASK.md, then answer with exactly EMPTY-RECOVERED-OK." diff --git a/qa/scenarios/runtime/empty-response-retry-budget-exhausted.yaml b/qa/scenarios/runtime/empty-response-retry-budget-exhausted.yaml index 933f62ae19de..43ed0cf53328 100644 --- a/qa/scenarios/runtime/empty-response-retry-budget-exhausted.yaml +++ b/qa/scenarios/runtime/empty-response-retry-budget-exhausted.yaml @@ -24,6 +24,7 @@ scenario: kind: flow summary: Verify empty-response retry exhaustion still surfaces a visible failure. config: + requiredProviderMode: mock-openai requiredProvider: mock-openai promptSnippet: Empty response exhaustion QA check prompt: "Empty response exhaustion QA check: read QA_KICKOFF_TASK.md, then answer with exactly EMPTY-EXHAUSTED-OK." diff --git a/qa/scenarios/runtime/reasoning-only-no-auto-retry-after-write.yaml b/qa/scenarios/runtime/reasoning-only-no-auto-retry-after-write.yaml index d7f0a3eb94cd..a1e4a00a3ec4 100644 --- a/qa/scenarios/runtime/reasoning-only-no-auto-retry-after-write.yaml +++ b/qa/scenarios/runtime/reasoning-only-no-auto-retry-after-write.yaml @@ -25,6 +25,7 @@ scenario: kind: flow summary: Verify reasoning-only turns after a write do not auto-retry. config: + requiredProviderMode: mock-openai requiredProvider: mock-openai promptSnippet: Reasoning-only after write safety check prompt: "Reasoning-only after write safety check: write reasoning-only-side-effect.txt, then answer with exactly SIDE-EFFECT-GUARD-OK." diff --git a/qa/scenarios/runtime/reasoning-only-recovery-replay-safe-read.yaml b/qa/scenarios/runtime/reasoning-only-recovery-replay-safe-read.yaml index 3354f9b6d0f0..fbb8cc630a72 100644 --- a/qa/scenarios/runtime/reasoning-only-recovery-replay-safe-read.yaml +++ b/qa/scenarios/runtime/reasoning-only-recovery-replay-safe-read.yaml @@ -23,6 +23,7 @@ scenario: kind: flow summary: Verify reasoning-only OpenAI turns recover after a replay-safe read. config: + requiredProviderMode: mock-openai requiredProvider: mock-openai promptSnippet: Reasoning-only continuation QA check prompt: "Reasoning-only continuation QA check: read QA_KICKOFF_TASK.md, then answer with exactly REASONING-RECOVERED-OK." diff --git a/scripts/lib/local-heavy-check-runtime.mjs b/scripts/lib/local-heavy-check-runtime.mjs index fea66f547525..a8b486de16bc 100644 --- a/scripts/lib/local-heavy-check-runtime.mjs +++ b/scripts/lib/local-heavy-check-runtime.mjs @@ -6,6 +6,7 @@ import path from "node:path"; const GIB = 1024 ** 3; const DEFAULT_LOCAL_GO_GC = "30"; +const DEFAULT_LOCAL_GO_MAX_PROCS = "1"; const DEFAULT_LOCAL_GO_MEMORY_LIMIT = "3GiB"; const DEFAULT_LOCAL_TSGO_BUILD_INFO_FILE = ".artifacts/tsgo-cache/root.tsbuildinfo"; const DEFAULT_LOCK_TIMEOUT_MS = 10 * 60 * 1000; @@ -141,6 +142,9 @@ export function applyLocalTsgoPolicy(args, env, hostResources) { insertBeforeSeparator(nextArgs, "--singleThreaded"); insertBeforeSeparator(nextArgs, "--checkers", "1"); + if (!nextEnv.GOMAXPROCS) { + nextEnv.GOMAXPROCS = DEFAULT_LOCAL_GO_MAX_PROCS; + } if (!nextEnv.GOGC) { nextEnv.GOGC = DEFAULT_LOCAL_GO_GC; } diff --git a/src/agents/tool-search-runtime.test.ts b/src/agents/tool-search-runtime.test.ts index adbbe8b0bde1..eaf2e2a540f1 100644 --- a/src/agents/tool-search-runtime.test.ts +++ b/src/agents/tool-search-runtime.test.ts @@ -63,6 +63,24 @@ describe("Tool Search flattened call arguments", () => { arguments: { id: "inspect_resource", command: "list", timeout_ms: 5_000 }, expected: { command: "list", timeout_ms: 5_000 }, }, + { + label: "dotted args from compatibility providers", + arguments: { + id: "inspect_resource", + "args.path": "projects/example.md", + "args.limit": 20, + }, + expected: { path: "projects/example.md", limit: 20 }, + }, + { + label: "ordinary flattened args precedence over dotted args", + arguments: { + id: "inspect_resource", + "args.path": "projects/dotted.md", + path: "projects/flattened.md", + }, + expected: { path: "projects/flattened.md" }, + }, { label: "toolId selector with a target id", arguments: { toolId: "inspect_resource", id: "record-7" }, @@ -84,8 +102,10 @@ describe("Tool Search flattened call arguments", () => { id: "inspect_resource", args: { command: "nested" }, command: "flattened", + "args.command": "dotted", + "args.path": "projects/dotted.md", }, - expected: { command: "nested" }, + expected: { command: "nested", path: "projects/dotted.md" }, }, { label: "explicit input precedence", diff --git a/src/agents/tool-search-runtime.ts b/src/agents/tool-search-runtime.ts index 032831afa711..6f3fbf204c7b 100644 --- a/src/agents/tool-search-runtime.ts +++ b/src/agents/tool-search-runtime.ts @@ -216,9 +216,17 @@ export function readToolSearchCallArgs( catalog?: ToolSearchCatalogSession, ): { id: string; input: unknown } { const params = asToolParamsRecord(args); + const dottedInput = Object.fromEntries( + Object.entries(params) + .filter(([key]) => key.startsWith("args.") && key.length > 5) + .map(([key, value]) => [key.slice(5), value]), + ); const nestedInput = params.args ?? params.input; if (nestedInput != null) { - return { id: readToolSearchId(params), input: nestedInput }; + return { + id: readToolSearchId(params), + input: isRecord(nestedInput) ? { ...dottedInput, ...nestedInput } : nestedInput, + }; } const selectorKeys = ["id", "toolId", "name"] as const; @@ -254,10 +262,11 @@ export function readToolSearchCallArgs( ...matchingSelectors.map(({ key }) => key), ...(matchingSelector ? [] : [selector ?? "id"]), ]); + const targetInputEntries = Object.entries(params).filter(([key]) => !wrapperKeys.has(key)); const flattenedInput = Object.fromEntries( - Object.entries(params).filter(([key]) => !wrapperKeys.has(key)), + targetInputEntries.filter(([key]) => !(key.startsWith("args.") && key.length > 5)), ); - return { id, input: flattenedInput }; + return { id, input: { ...dottedInput, ...flattenedInput } }; } function getTelemetry(catalog: ToolSearchCatalogSession) { diff --git a/src/commands/doctor-config-preflight-plugin-verification.test.ts b/src/commands/doctor-config-preflight-plugin-verification.test.ts new file mode 100644 index 000000000000..c2e274221d11 --- /dev/null +++ b/src/commands/doctor-config-preflight-plugin-verification.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from "vitest"; +import { formatStartupPluginVerificationFailure } from "./doctor-config-preflight-plugin-verification.js"; + +describe("formatStartupPluginVerificationFailure", () => { + it("uses install-neutral gateway restart guidance", () => { + expect( + formatStartupPluginVerificationFailure({ + kind: "plugin-verification", + messages: ['Plugin "discord" has no install path.'], + }), + ).toBe( + [ + "OpenClaw plugin verification failed; refusing to report the gateway ready.", + '- Plugin "discord" has no install path.', + "Resolve the plugin verification errors above, then restart the gateway.", + ].join("\n"), + ); + }); +}); diff --git a/src/commands/doctor-config-preflight-plugin-verification.ts b/src/commands/doctor-config-preflight-plugin-verification.ts index d338f5ab2d69..d0b719d37d1e 100644 --- a/src/commands/doctor-config-preflight-plugin-verification.ts +++ b/src/commands/doctor-config-preflight-plugin-verification.ts @@ -221,6 +221,6 @@ export function formatStartupPluginVerificationFailure( return [ "OpenClaw plugin verification failed; refusing to report the gateway ready.", ...diagnostic.messages.map((message) => `- ${message}`), - "Resolve the plugin verification errors above, then restart the container.", + "Resolve the plugin verification errors above, then restart the gateway.", ].join("\n"); } diff --git a/test/e2e/qa-lab/runtime/diagnostics-otel-install-runtime.e2e.test.ts b/test/e2e/qa-lab/runtime/diagnostics-otel-install-runtime.e2e.test.ts new file mode 100644 index 000000000000..f76fa49613cc --- /dev/null +++ b/test/e2e/qa-lab/runtime/diagnostics-otel-install-runtime.e2e.test.ts @@ -0,0 +1,476 @@ +import { execFile } from "node:child_process"; +import { spawn, type ChildProcess } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { cp, mkdir, mkdtemp, readFile, readdir, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { pathToFileURL } from "node:url"; +import { promisify } from "node:util"; +import { describe, expect, test } from "vitest"; +import { startQaGatewayChild, startQaMockOpenAiServer } from "../../../../extensions/qa-lab/api.js"; +import { readPluginInstallRecords } from "../../../../scripts/e2e/lib/plugin-index-sqlite.mjs"; +import { startLocalOtlpReceiver } from "./otel-test-support.js"; + +const execFileAsync = promisify(execFile); +const PACKAGE_NAME = "@openclaw/diagnostics-otel"; +const PACKAGE_VERSION = "2026.7.2"; + +type MutableConfig = { + diagnostics?: unknown; + plugins?: { + entries?: Record; + }; + [key: string]: unknown; +}; + +async function waitForChildExit(child: ChildProcess, timeoutMs: number): Promise { + if (child.exitCode !== null) { + return true; + } + return await new Promise((resolve) => { + const onExit = () => { + clearTimeout(timer); + resolve(true); + }; + const timer = setTimeout(() => { + child.off("exit", onExit); + resolve(false); + }, timeoutMs); + timer.unref(); + child.once("exit", onExit); + if (child.exitCode !== null) { + child.off("exit", onExit); + clearTimeout(timer); + resolve(true); + } + }); +} + +async function stopChild(child: ChildProcess | undefined): Promise { + if (!child || child.exitCode !== null) { + return; + } + child.kill("SIGTERM"); + if (await waitForChildExit(child, 5_000)) { + return; + } + child.kill("SIGKILL"); + if (!(await waitForChildExit(child, 5_000))) { + throw new Error("fixture registry did not exit after SIGKILL"); + } +} + +async function waitFor( + read: () => T | undefined | Promise, + timeoutMs = 60_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const value = await read(); + if (value !== undefined) { + return value; + } + await sleep(100); + } + throw new Error("timed out waiting for managed diagnostics-otel evidence"); +} + +async function startReceiver() { + const receiver = startLocalOtlpReceiver(); + const port = await receiver.listen(); + return { ...receiver, baseUrl: `http://127.0.0.1:${port}` }; +} + +async function runCleanup( + label: string, + cleanup: () => Promise, + timeoutMs = 30_000, +): Promise { + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`${label} cleanup timed out`)), timeoutMs); + timer.unref(); + cleanup().then( + () => { + clearTimeout(timer); + resolve(); + }, + (error) => { + clearTimeout(timer); + reject(error); + }, + ); + }); +} + +async function settleCleanup( + ...cleanups: Array Promise]> +): Promise { + const results = await Promise.allSettled( + cleanups.map(async ([label, cleanup]) => await runCleanup(label, cleanup)), + ); + const failures = results.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length > 0) { + throw new AggregateError(failures, "managed diagnostics-otel cleanup failed"); + } +} + +async function packPlugin(repoRoot: string, scratch: string) { + const outputDir = path.join(scratch, "pack"); + const pluginRoot = path.join(repoRoot, "extensions/diagnostics-otel"); + const stagingDir = path.join(scratch, "package-source"); + await cp(pluginRoot, stagingDir, { + recursive: true, + filter: (source) => { + const relative = path.relative(pluginRoot, source); + const topLevel = relative.split(path.sep)[0]; + return topLevel !== "dist" && topLevel !== "node_modules"; + }, + }); + await mkdir(outputDir, { recursive: true }); + await execFileAsync(process.execPath, ["scripts/lib/plugin-npm-runtime-build.mjs", stagingDir], { + cwd: repoRoot, + maxBuffer: 16 * 1024 * 1024, + timeout: 120_000, + }); + await execFileAsync( + process.execPath, + [ + "scripts/lib/plugin-npm-package-manifest.mjs", + "--run", + stagingDir, + "--", + "npm", + "pack", + "--json", + "--ignore-scripts", + "--pack-destination", + outputDir, + ], + { + cwd: repoRoot, + env: { + ...process.env, + OPENCLAW_PLUGIN_NPM_BUNDLE_DEPENDENCIES: "1", + }, + maxBuffer: 16 * 1024 * 1024, + timeout: 120_000, + }, + ); + const tarballName = (await readdir(outputDir)).find((name) => name.endsWith(".tgz")); + if (!tarballName) { + throw new Error("diagnostics-otel pack did not produce a tarball"); + } + return path.join(outputDir, tarballName); +} + +async function startRegistry(repoRoot: string, scratch: string, tarball: string) { + const portFile = path.join(scratch, "registry-port"); + const child = spawn( + process.execPath, + [ + "scripts/e2e/lib/plugins/npm-registry-server.mjs", + portFile, + PACKAGE_NAME, + PACKAGE_VERSION, + tarball, + ], + { + cwd: repoRoot, + env: { + ...process.env, + OPENCLAW_NPM_REGISTRY_UPSTREAM: "https://registry.npmjs.org", + }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + try { + const port = await waitFor(async () => { + try { + return (await readFile(portFile, "utf8")).trim() || undefined; + } catch { + if (child.exitCode !== null) { + throw new Error(`fixture npm registry exited early (${child.exitCode})`); + } + return undefined; + } + }); + return { baseUrl: `http://127.0.0.1:${port}`, child }; + } catch (error) { + await stopChild(child).catch((stopError) => { + throw new Error( + `fixture npm registry startup cleanup failed: ${ + stopError instanceof Error ? stopError.message : String(stopError) + }`, + { + cause: error, + }, + ); + }); + throw error; + } +} + +async function runTurn(gateway: Awaited>, marker: string) { + const started = (await gateway.call("chat.send", { + sessionKey: `agent:qa:${marker.toLowerCase()}`, + message: `Reply exactly: ${marker}`, + idempotencyKey: randomUUID(), + })) as { runId?: string; status?: string }; + expect(started.status).toBe("started"); + expect(started.runId).toBeTruthy(); + const completed = (await gateway.call( + "agent.wait", + { runId: started.runId, timeoutMs: 60_000 }, + { timeoutMs: 65_000 }, + )) as { status?: string }; + expect(completed.status).toBe("ok"); +} + +async function restartWithOtelConfig(params: { + gateway: Awaited>; + sampleRate: number; + traceEndpoint: string; +}) { + await params.gateway.restartAfterStateMutation(async ({ configPath }) => { + const current = JSON.parse(await readFile(configPath, "utf8")) as MutableConfig; + current.diagnostics = { + enabled: true, + otel: { + enabled: true, + protocol: "http/protobuf", + traces: true, + metrics: false, + logs: false, + tracesEndpoint: `${params.traceEndpoint}/v1/traces`, + sampleRate: params.sampleRate, + flushIntervalMs: 250, + captureContent: false, + }, + }; + await writeFile(configPath, `${JSON.stringify(current, null, 2)}\n`); + }); +} + +async function installAndConfigure(params: { + configTraceEndpoint: string; + envTraceEndpoint: string; + mockBaseUrl: string; + nodeOptions?: string; + registryBaseUrl: string; + repoRoot: string; + sampleRate?: number; +}) { + const gateway = await startQaGatewayChild({ + repoRoot: params.repoRoot, + providerBaseUrl: `${params.mockBaseUrl}/v1`, + providerMode: "mock-openai", + transportBaseUrl: "http://127.0.0.1:9", + controlUiEnabled: false, + mutateConfig: (cfg) => ({ + ...cfg, + plugins: { + ...cfg.plugins, + allow: [], + slots: { + ...cfg.plugins?.slots, + memory: "none", + }, + entries: {}, + }, + }), + runtimeEnvPatch: { + NPM_CONFIG_REGISTRY: params.registryBaseUrl, + OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1", + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: `${params.envTraceEndpoint}/v1/traces`, + ...(params.nodeOptions ? { NODE_OPTIONS: params.nodeOptions } : {}), + ...(params.nodeOptions ? { OPENCLAW_OTEL_PRELOADED: "1" } : {}), + }, + }); + const spec = `npm:${PACKAGE_NAME}@${PACKAGE_VERSION}`; + await gateway.runCli(["plugins", "install", spec, "--force"]); + const stateDir = gateway.runtimeEnv.OPENCLAW_STATE_DIR; + if (!stateDir) { + throw new Error("qa gateway state directory was not configured"); + } + const records = readPluginInstallRecords({ + stateDir, + configPath: gateway.configPath, + }); + expect(records["diagnostics-otel"]).toMatchObject({ + source: "npm", + spec: `${PACKAGE_NAME}@${PACKAGE_VERSION}`, + version: PACKAGE_VERSION, + resolvedName: PACKAGE_NAME, + resolvedVersion: PACKAGE_VERSION, + }); + expect(records["diagnostics-otel"]?.installPath).toContain("diagnostics-otel"); + expect(records["diagnostics-otel"]?.integrity).toMatch(/^sha512-/u); + + await gateway.runCli(["plugins", "disable", "diagnostics-otel"]); + let config = JSON.parse(await readFile(gateway.configPath, "utf8")) as MutableConfig; + expect(config.plugins?.entries?.["diagnostics-otel"]?.enabled).toBe(false); + await gateway.runCli(["plugins", "enable", "diagnostics-otel"]); + config = JSON.parse(await readFile(gateway.configPath, "utf8")) as MutableConfig; + expect(config.plugins?.entries?.["diagnostics-otel"]?.enabled).toBe(true); + + await restartWithOtelConfig({ + gateway, + sampleRate: params.sampleRate ?? 1, + traceEndpoint: params.configTraceEndpoint, + }); + const inspect = JSON.parse( + await gateway.runCli(["plugins", "inspect", "diagnostics-otel", "--runtime", "--json"]), + ) as { plugin?: { enabled?: boolean; id?: string; status?: string } }; + expect(inspect.plugin).toMatchObject({ + enabled: true, + id: "diagnostics-otel", + status: "loaded", + }); + return gateway; +} + +describe("managed diagnostics-otel install runtime", () => { + test("installs the exact package and exports with config precedence, sampling, and flush", async () => { + const repoRoot = path.resolve(import.meta.dirname, "../../../.."); + const scratch = await mkdtemp(path.join(tmpdir(), "openclaw-otel-install-")); + const configured = await startReceiver(); + const envOnly = await startReceiver(); + let registry: Awaited> | undefined; + let mock: Awaited> | undefined; + let gateway: Awaited> | undefined; + try { + const tarball = await packPlugin(repoRoot, scratch); + registry = await startRegistry(repoRoot, scratch, tarball); + mock = await startQaMockOpenAiServer(); + gateway = await installAndConfigure({ + configTraceEndpoint: configured.baseUrl, + envTraceEndpoint: envOnly.baseUrl, + mockBaseUrl: mock.baseUrl, + registryBaseUrl: registry.baseUrl, + repoRoot, + sampleRate: 0, + }); + await runTurn(gateway, "OTEL-MANAGED-SAMPLED-OUT"); + await sleep(1_500); + expect(configured.capturedRequests).toHaveLength(0); + const sampledOutRequestCursor = configured.capturedRequests.length; + const sampledOutSpanCursor = configured.capturedSpans.length; + expect(envOnly.capturedRequests).toHaveLength(0); + + await restartWithOtelConfig({ + gateway, + sampleRate: 1, + traceEndpoint: configured.baseUrl, + }); + expect(configured.capturedRequests).toHaveLength(sampledOutRequestCursor); + expect(configured.capturedSpans).toHaveLength(sampledOutSpanCursor); + const sampledInRequestCursor = configured.capturedRequests.length; + const sampledInSpanCursor = configured.capturedSpans.length; + await runTurn(gateway, "OTEL-MANAGED-INSTALL-OK"); + const sampledInExport = await waitFor(() => { + let spanOffset = sampledInSpanCursor; + for (const request of configured.capturedRequests.slice(sampledInRequestCursor)) { + const requestSpans = configured.capturedSpans.slice( + spanOffset, + spanOffset + request.spanCount, + ); + spanOffset += request.spanCount; + if ( + request.path === "/v1/traces" && + requestSpans.some((span) => span.name === "openclaw.run") + ) { + return { request, spans: requestSpans }; + } + } + return undefined; + }, 15_000); + // BatchSpanProcessor starts its timer on the first ended span. The first + // export's earliest end timestamp is the boundary that must observe the clamp. + const firstRequestEndTimes = sampledInExport.spans.flatMap((span) => + span.endTimeMs === undefined ? [] : [span.endTimeMs], + ); + expect(firstRequestEndTimes.length).toBeGreaterThan(0); + const firstSpanEndAt = Math.min(...firstRequestEndTimes); + const exportDelayMs = (sampledInExport.request.receivedAtMs ?? 0) - firstSpanEndAt; + expect(exportDelayMs).toBeGreaterThanOrEqual(1_000); + expect(exportDelayMs).toBeLessThan(4_500); + expect(envOnly.capturedRequests).toHaveLength(0); + } finally { + await settleCleanup( + ["gateway", async () => await gateway?.stop()], + ["mock provider", async () => await mock?.stop()], + ["fixture registry", async () => await stopChild(registry?.child)], + ["configured receiver", async () => await configured.close()], + ["environment receiver", async () => await envOnly.close()], + ["scratch directory", async () => await rm(scratch, { recursive: true, force: true })], + ); + } + }, 180_000); + + test("keeps installed diagnostic listeners active with a preloaded SDK", async () => { + const repoRoot = path.resolve(import.meta.dirname, "../../../.."); + const scratch = await mkdtemp(path.join(tmpdir(), "openclaw-otel-preloaded-")); + const receiver = await startReceiver(); + const ignoredConfig = await startReceiver(); + let registry: Awaited> | undefined; + let mock: Awaited> | undefined; + let gateway: Awaited> | undefined; + try { + const tarball = await packPlugin(repoRoot, scratch); + registry = await startRegistry(repoRoot, scratch, tarball); + mock = await startQaMockOpenAiServer(); + const preloadRoot = path.join(scratch, `otel-preload-${randomUUID()}`); + const preloadModules = path.join(preloadRoot, "node_modules", "@opentelemetry"); + await mkdir(preloadModules, { recursive: true }); + // The scratch preload resolves the same hoisted packages declared by the + // diagnostics plugin without making them root test dependencies. + for (const packageName of ["sdk-node", "exporter-trace-otlp-proto"]) { + await symlink( + path.join(repoRoot, "node_modules", "@opentelemetry", packageName), + path.join(preloadModules, packageName), + process.platform === "win32" ? "junction" : "dir", + ); + } + const preloadPath = path.join(preloadRoot, "preload.mjs"); + await writeFile( + preloadPath, + [ + 'import { NodeSDK } from "@opentelemetry/sdk-node";', + 'import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";', + `const sdk = new NodeSDK({ traceExporter: new OTLPTraceExporter({ url: ${JSON.stringify(`${receiver.baseUrl}/v1/traces`)} }) });`, + "sdk.start();", + "globalThis.__openclawQaPreloadedOtelSdk = sdk;", + ].join("\n"), + ); + gateway = await installAndConfigure({ + configTraceEndpoint: ignoredConfig.baseUrl, + envTraceEndpoint: receiver.baseUrl, + mockBaseUrl: mock.baseUrl, + nodeOptions: `--import=${pathToFileURL(preloadPath).href}`, + registryBaseUrl: registry.baseUrl, + repoRoot, + }); + expect(gateway.logs()).toContain("diagnostics-otel: using preloaded OpenTelemetry SDK"); + await runTurn(gateway, "OTEL-PRELOADED-INSTALL-OK"); + const runSpan = await waitFor( + () => receiver.capturedSpans.find((span) => span.name === "openclaw.run"), + 20_000, + ); + expect(runSpan.traceId).toBeTruthy(); + expect(runSpan.spanId).toBeTruthy(); + expect(ignoredConfig.capturedRequests).toHaveLength(0); + } finally { + await settleCleanup( + ["gateway", async () => await gateway?.stop()], + ["mock provider", async () => await mock?.stop()], + ["fixture registry", async () => await stopChild(registry?.child)], + ["preloaded receiver", async () => await receiver.close()], + ["ignored config receiver", async () => await ignoredConfig.close()], + ["scratch directory", async () => await rm(scratch, { recursive: true, force: true })], + ); + } + }, 180_000); +}); diff --git a/test/e2e/qa-lab/runtime/otel-gateway-runtime.e2e.test.ts b/test/e2e/qa-lab/runtime/otel-gateway-runtime.e2e.test.ts new file mode 100644 index 000000000000..00fc83e8a0e5 --- /dev/null +++ b/test/e2e/qa-lab/runtime/otel-gateway-runtime.e2e.test.ts @@ -0,0 +1,320 @@ +import path from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { describe, expect, test } from "vitest"; +import { + createQaBusState, + startQaBusServer, + startQaGatewayChild, + startQaMockOpenAiServer, +} from "../../../../extensions/qa-lab/api.js"; +import { type CapturedSpan, startLocalOtlpReceiver } from "./otel-test-support.js"; + +async function startOtlpReceiver() { + const receiver = startLocalOtlpReceiver(); + const port = await receiver.listen(); + return { ...receiver, baseUrl: `http://127.0.0.1:${port}` }; +} + +async function settleCleanup(...cleanups: Array<() => Promise>): Promise { + const results = await Promise.allSettled(cleanups.map(async (cleanup) => await cleanup())); + const failures = results.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length > 0) { + throw new AggregateError(failures, "diagnostics-otel gateway cleanup failed"); + } +} + +async function waitFor( + read: () => T | undefined, + timeoutMs = 30_000, + timeoutContext?: () => unknown, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const value = read(); + if (value !== undefined) { + return value; + } + await sleep(100); + } + const context = timeoutContext?.(); + throw new Error( + `timed out waiting for QA runtime evidence${ + context === undefined ? "" : `: ${JSON.stringify(context)}` + }`, + ); +} + +function indexSpansById(spans: CapturedSpan[]): Map { + return new Map(spans.flatMap((span) => (span.spanId ? ([[span.spanId, span]] as const) : []))); +} + +function expectResolvedParent( + span: CapturedSpan, + spansById: ReadonlyMap, +): CapturedSpan { + expect(span.parentSpanId).toBeTruthy(); + const parent = span.parentSpanId ? spansById.get(span.parentSpanId) : undefined; + expect(parent).toBeDefined(); + return parent!; +} + +describe("diagnostics-otel gateway runtime", () => { + test("exports linked success and failed-tool recovery spans from a real qa-channel run", async () => { + const repoRoot = path.resolve(import.meta.dirname, "../../../.."); + const state = createQaBusState(); + const transport = { + requiredPluginIds: ["qa-channel"], + createGatewayConfig: ({ baseUrl }: { baseUrl: string }) => ({ + channels: { + "qa-channel": { + enabled: true, + baseUrl, + botUserId: "openclaw", + botDisplayName: "OpenClaw QA", + allowFrom: ["*"], + pollTimeoutMs: 250, + }, + }, + messages: { + visibleReplies: "automatic" as const, + groupChat: { + mentionPatterns: ["\\b@?openclaw\\b"], + visibleReplies: "automatic" as const, + }, + }, + }), + }; + let bus: Awaited> | undefined; + let receiver: Awaited> | undefined; + let mock: Awaited> | undefined; + let gateway: Awaited> | undefined; + + try { + bus = await startQaBusServer({ state }); + const activeReceiver = await startOtlpReceiver(); + receiver = activeReceiver; + mock = await startQaMockOpenAiServer(); + gateway = await startQaGatewayChild({ + repoRoot, + useRepoCli: true, + providerBaseUrl: `${mock.baseUrl}/v1`, + providerMode: "mock-openai", + transport, + transportBaseUrl: bus.baseUrl, + enabledPluginIds: ["diagnostics-otel"], + controlUiEnabled: false, + mutateConfig: (cfg) => ({ + ...cfg, + tools: { + ...cfg.tools, + codeMode: { + ...(typeof cfg.tools?.codeMode === "object" ? cfg.tools.codeMode : {}), + enabled: true, + }, + }, + diagnostics: { + enabled: true, + otel: { + enabled: true, + endpoint: activeReceiver.baseUrl, + protocol: "http/protobuf", + traces: true, + metrics: false, + logs: false, + sampleRate: 1, + flushIntervalMs: 1000, + captureContent: false, + }, + }, + }), + }); + const conversation = { id: "qa-operator", kind: "direct" as const }; + const send = async (text: string) => { + const cursor = state.getSnapshot().messages.length; + state.addInboundMessage({ + conversation, + senderId: "qa-user", + senderName: "QA User", + text, + }); + return await waitFor(() => + state + .getSnapshot() + .messages.slice(cursor) + .find( + (message) => + message.direction === "outbound" && message.conversation.id === conversation.id, + ), + ); + }; + + const successful = await send( + "Tool progress QA check: use the read tool exactly once on `QA_KICKOFF_TASK.md` before answering. After that read completes, reply with only this exact marker and no other text: `OTEL-GATEWAY-SUCCESS-OK`.", + ); + expect(successful.direction).toBe("outbound"); + expect(successful.text).toContain("OTEL-GATEWAY-SUCCESS-OK"); + + const requestCursor = (await fetch(`${mock.baseUrl}/debug/request-cursor`).then((response) => + response.json(), + )) as { cursor: number }; + const recovered = await send( + "Failed tool terminal recovery QA check: read the missing workspace file, then respond with exact marker: `QA-FAILED-TOOL-FINALIZED-OK`.", + ); + expect(recovered.direction).toBe("outbound"); + expect(recovered.text).toContain("The requested file could not be read: ENOENT."); + expect(recovered.text).toContain("QA-FAILED-TOOL-FINALIZED-OK"); + + const scenarioRequests = (await fetch( + `${mock.baseUrl}/debug/requests?after=${requestCursor.cursor}`, + ).then((response) => response.json())) as Array<{ + allInputText?: string; + body?: { input?: Array>; tools?: unknown[] }; + plannedToolName?: string; + plannedWireToolName?: string; + toolOutputCallId?: string; + }>; + const readPlans = scenarioRequests.filter((request) => request.plannedToolName === "read"); + const finalizations = scenarioRequests.filter((request) => + String(request.allInputText ?? "").includes( + "The previous assistant turn completed its tool calls but did not produce a user-visible answer.", + ), + ); + expect(readPlans).toHaveLength(1); + expect(readPlans[0]?.plannedWireToolName).toBe("exec"); + expect(finalizations).toHaveLength(1); + expect(finalizations[0]?.body?.tools ?? []).toHaveLength(0); + expect(finalizations[0]?.allInputText).toContain( + "state that failure plainly and do not claim it succeeded", + ); + const finalizationInput = finalizations[0]?.body?.input ?? []; + const failedExecCalls = finalizationInput.filter( + (item) => + item.type === "function_call" && + item.name === "exec" && + String(item.arguments ?? "").includes("qa-failed-terminal-missing-file.txt"), + ); + const failedExecOutputs = finalizationInput.filter( + (item) => + item.type === "function_call_output" && + item.call_id === failedExecCalls[0]?.call_id && + /ENOENT|no such file/iu.test(String(item.output ?? "")), + ); + expect(failedExecCalls).toHaveLength(1); + expect(failedExecOutputs).toHaveLength(1); + expect(finalizations[0]?.toolOutputCallId).toBe(failedExecCalls[0]?.call_id); + + const failureEvidence = await waitFor( + () => { + const toolError = activeReceiver.capturedSpans.find( + (span) => + span.name === "openclaw.tool.execution" && + span.statusCode === 2 && + span.attributes["openclaw.toolName"] === "read" && + Boolean(span.attributes["openclaw.errorCategory"]), + ); + if (!toolError?.traceId) { + return undefined; + } + const sameTrace = activeReceiver.capturedSpans.filter( + (span) => span.traceId === toolError.traceId, + ); + const runs = sameTrace.filter((span) => span.name === "openclaw.run"); + const harnesses = sameTrace.filter((span) => span.name === "openclaw.harness.run"); + const modelCalls = sameTrace.filter((span) => span.name === "openclaw.model.call"); + // QA-channel inbound replies use the channel-owned direct callback, not + // deliver-core; the outbound bus receipt above is the delivery proof. + const terminal = sameTrace.find( + (span) => + span.name === "openclaw.message.processed" && + span.attributes["openclaw.channel"] === "qa-channel" && + span.attributes["openclaw.outcome"] === "completed", + ); + return runs.length >= 2 && harnesses.length >= 2 && modelCalls.length >= 2 && terminal + ? { harnesses, modelCalls, runs, sameTrace, terminal, toolError } + : undefined; + }, + 45_000, + () => ({ + requests: activeReceiver.capturedRequests, + spans: activeReceiver.capturedSpans.map((span) => ({ + attributes: span.attributes, + name: span.name, + parentSpanId: span.parentSpanId, + spanId: span.spanId, + statusCode: span.statusCode, + traceId: span.traceId, + })), + }), + ); + + const failureSpansById = indexSpansById(failureEvidence.sameTrace); + expect(failureEvidence.terminal.parentSpanId).toBeFalsy(); + for (const harness of failureEvidence.harnesses) { + expect(expectResolvedParent(harness, failureSpansById)).toBe(failureEvidence.terminal); + } + for (const run of failureEvidence.runs) { + expect(expectResolvedParent(run, failureSpansById).name).toBe("openclaw.harness.run"); + } + for (const modelCall of failureEvidence.modelCalls) { + expect(expectResolvedParent(modelCall, failureSpansById).name).toBe("openclaw.run"); + } + expect(expectResolvedParent(failureEvidence.toolError, failureSpansById).name).toBe( + "openclaw.run", + ); + + const successEvidence = activeReceiver.capturedSpans.find( + (span) => + span.name === "openclaw.tool.execution" && + span.statusCode !== 2 && + span.attributes["openclaw.toolName"] === "read" && + span.traceId !== failureEvidence.toolError.traceId, + ); + expect(successEvidence).toBeTruthy(); + const successTrace = activeReceiver.capturedSpans.filter( + (span) => span.traceId === successEvidence?.traceId, + ); + const successTerminal = successTrace.find( + (span) => + span.name === "openclaw.message.processed" && + span.attributes["openclaw.channel"] === "qa-channel" && + span.attributes["openclaw.outcome"] === "completed", + ); + const successHarnesses = successTrace.filter((span) => span.name === "openclaw.harness.run"); + const successRuns = successTrace.filter((span) => span.name === "openclaw.run"); + const successModelCalls = successTrace.filter((span) => span.name === "openclaw.model.call"); + expect(successTerminal).toBeDefined(); + expect(successHarnesses.length).toBeGreaterThanOrEqual(1); + expect(successRuns.length).toBeGreaterThanOrEqual(1); + expect(successModelCalls.length).toBeGreaterThanOrEqual(1); + const successSpansById = indexSpansById(successTrace); + expect(successTerminal?.parentSpanId).toBeFalsy(); + for (const harness of successHarnesses) { + expect(expectResolvedParent(harness, successSpansById)).toBe(successTerminal); + } + for (const run of successRuns) { + expect(expectResolvedParent(run, successSpansById).name).toBe("openclaw.harness.run"); + } + for (const modelCall of successModelCalls) { + expect(expectResolvedParent(modelCall, successSpansById).name).toBe("openclaw.run"); + } + expect(expectResolvedParent(successEvidence!, successSpansById).name).toBe("openclaw.run"); + } finally { + await settleCleanup( + async () => { + await gateway?.stop(); + }, + async () => { + await mock?.stop(); + }, + async () => { + await receiver?.close(); + }, + async () => { + await bus?.stop(); + }, + ); + } + }, 120_000); +}); diff --git a/test/e2e/qa-lab/runtime/otel-test-support.ts b/test/e2e/qa-lab/runtime/otel-test-support.ts new file mode 100644 index 000000000000..6e7d42b38915 --- /dev/null +++ b/test/e2e/qa-lab/runtime/otel-test-support.ts @@ -0,0 +1,756 @@ +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import { Socket } from "node:net"; +import { gunzipSync } from "node:zlib"; + +export type OtlpSignal = "logs" | "metrics" | "traces"; + +type OtlpAnyValue = { + stringValue?: string; + boolValue?: boolean; + intValue?: number | string | { toString(): string }; + doubleValue?: number; + arrayValue?: { values?: OtlpAnyValue[] }; + kvlistValue?: { values?: OtlpKeyValue[] }; + bytesValue?: Uint8Array; +}; + +type OtlpKeyValue = { + key?: string; + value?: OtlpAnyValue; +}; + +type OtlpSpan = { + attributes?: OtlpKeyValue[]; + endTimeMs?: number; + name?: string; + parentSpanId?: Uint8Array; + spanId?: Uint8Array; + statusCode?: number; + traceId?: Uint8Array; +}; + +type OtlpScopeSpans = { + spans?: OtlpSpan[]; +}; + +type OtlpResourceSpans = { + scopeSpans?: OtlpScopeSpans[]; +}; + +export type CapturedRequest = { + bytes: number; + contentEncoding?: string; + logCount: number; + metricCount: number; + path: string; + receivedAtMs?: number; + signal: OtlpSignal; + spanCount: number; + status: number; +}; + +export type CapturedSpan = { + attributes: Record; + endTimeMs?: number; + name: string; + parent: boolean; + parentSpanId?: string; + spanId?: string; + statusCode?: number; + traceId?: string; +}; + +export type CapturedMetric = { + name: string; +}; + +export type CapturedLogRecord = { + body: string | number | boolean | string[]; + spanId: string; + traceId: string; +}; + +const OTLP_SIGNAL_PATHS = new Map([ + ["/v1/traces", "traces"], + ["/v1/metrics", "metrics"], + ["/v1/logs", "logs"], +]); +const POSITIVE_INTEGER_PATTERN = /^[1-9]\d*$/u; +const MAX_OTLP_COMPRESSED_BODY_BYTES = readPositiveIntegerEnv( + "OPENCLAW_QA_OTEL_MAX_COMPRESSED_BODY_BYTES", + 2 * 1024 * 1024, +); +const MAX_OTLP_DECODED_BODY_BYTES = readPositiveIntegerEnv( + "OPENCLAW_QA_OTEL_MAX_DECODED_BODY_BYTES", + 8 * 1024 * 1024, +); +const MAX_CAPTURED_BODY_TEXT_BYTES = readPositiveIntegerEnv( + "OPENCLAW_QA_OTEL_MAX_CAPTURED_BODY_TEXT_BYTES", + 512 * 1024, +); + +export function readPositiveIntegerEnv( + name: string, + fallback: number, + env: NodeJS.ProcessEnv = process.env, +): number { + const raw = env[name]; + if (raw == null || raw.trim() === "") { + return fallback; + } + const value = raw.trim(); + if (!POSITIVE_INTEGER_PATTERN.test(value)) { + throw new Error(`${name} must be a positive integer`); + } + const parsed = Number(value); + if (!Number.isSafeInteger(parsed)) { + throw new Error(`${name} must be a safe integer`); + } + return parsed; +} + +function oversizedBodyError(label: string, actualBytes: number, maxBytes: number): Error { + const error = new Error(`${label} exceeded ${maxBytes} bytes: ${actualBytes} bytes`) as Error & { + statusCode?: number; + }; + error.statusCode = 413; + return error; +} + +export async function readRequestBody( + req: IncomingMessage, + maxBytes = MAX_OTLP_COMPRESSED_BODY_BYTES, +): Promise { + const chunks: Buffer[] = []; + let totalBytes = 0; + for await (const chunk of req) { + const buffer = Buffer.from(chunk); + totalBytes += buffer.length; + if (totalBytes > maxBytes) { + req.destroy(); + throw oversizedBodyError("compressed OTLP request body", totalBytes, maxBytes); + } + chunks.push(buffer); + } + return Buffer.concat(chunks); +} + +function headerValue(value: string | string[] | undefined): string | undefined { + return Array.isArray(value) ? value[0] : value; +} + +export function decodeRequestBody( + body: Buffer, + contentEncoding: string | undefined, + maxBytes = MAX_OTLP_DECODED_BODY_BYTES, +): Buffer { + const normalizedEncoding = contentEncoding?.trim().toLowerCase(); + if (body.length > maxBytes && (!normalizedEncoding || normalizedEncoding === "identity")) { + throw oversizedBodyError("OTLP request body", body.length, maxBytes); + } + if (!normalizedEncoding || normalizedEncoding === "identity") { + return body; + } + if (normalizedEncoding === "gzip") { + let decoded: Buffer; + try { + decoded = gunzipSync(body, { maxOutputLength: maxBytes }); + } catch (error) { + const code = (error as { code?: unknown }).code; + const message = error instanceof Error ? error.message : String(error); + if (code === "ERR_BUFFER_TOO_LARGE" || /maxOutputLength|larger than/u.test(message)) { + throw oversizedBodyError("decoded OTLP request body", maxBytes + 1, maxBytes); + } + throw error; + } + if (decoded.length > maxBytes) { + throw oversizedBodyError("decoded OTLP request body", decoded.length, maxBytes); + } + return decoded; + } + throw new Error(`unsupported OTLP content-encoding ${contentEncoding}`); +} + +export function appendCapturedBodyText( + capturedBodyText: Partial>, + signal: OtlpSignal, + body: Buffer, + maxBytes = MAX_CAPTURED_BODY_TEXT_BYTES, + disallowedNeedles: string[] = [], +): void { + const currentEntries = capturedBodyText[signal] ?? []; + const leakEntries = currentEntries.filter((entry) => entry.startsWith("[detected leak needle] ")); + const currentTail = currentEntries + .filter((entry) => !entry.startsWith("[detected leak needle] ")) + .join("\n"); + const bodyText = body.toString("utf8"); + const next = currentTail ? `${currentTail}\n${bodyText}` : bodyText; + const buffer = Buffer.from(next); + const nextLeakEntries = [ + ...leakEntries, + ...disallowedNeedles + .filter((needle) => bodyText.includes(needle)) + .map((needle) => `[detected leak needle] ${needle}`), + ].slice(-20); + const tailEntry = + buffer.length > maxBytes + ? `[captured body text truncated to last ${maxBytes} bytes]\n${buffer + .subarray(buffer.length - maxBytes) + .toString("utf8")}` + : next; + capturedBodyText[signal] = [...nextLeakEntries, tailEntry]; +} + +function normalizeOtlpValue(value: OtlpAnyValue | undefined): string | number | boolean | string[] { + if (!value) { + return ""; + } + if (typeof value.stringValue === "string") { + return value.stringValue; + } + if (typeof value.boolValue === "boolean") { + return value.boolValue; + } + if (typeof value.doubleValue === "number") { + return value.doubleValue; + } + if (value.intValue !== undefined) { + return Number(value.intValue.toString()); + } + if (value.arrayValue?.values) { + return value.arrayValue.values.map((entry) => String(normalizeOtlpValue(entry))); + } + if (value.kvlistValue?.values) { + return value.kvlistValue.values + .map((entry) => `${entry.key ?? ""}=${String(normalizeOtlpValue(entry.value))}`) + .filter(Boolean); + } + if (value.bytesValue) { + return Buffer.from(value.bytesValue).toString("hex"); + } + return ""; +} + +function spanAttributes(span: OtlpSpan): Record { + const attributes: Record = {}; + for (const attribute of span.attributes ?? []) { + const key = attribute.key?.trim(); + if (!key) { + continue; + } + attributes[key] = normalizeOtlpValue(attribute.value); + } + return attributes; +} + +class ProtoReader { + private offset = 0; + + constructor(private readonly buffer: Uint8Array) {} + + done(): boolean { + return this.offset >= this.buffer.length; + } + + tag() { + const raw = this.varint(); + return { field: raw >>> 3, wire: raw & 0x7 }; + } + + varint(): number { + let result = 0; + let shift = 0; + while (this.offset < this.buffer.length) { + const byte = this.buffer.at(this.offset); + if (byte === undefined) { + throw new Error("truncated protobuf varint"); + } + this.offset += 1; + result += (byte & 0x7f) * 2 ** shift; + if ((byte & 0x80) === 0) { + return result; + } + shift += 7; + } + throw new Error("truncated protobuf varint"); + } + + bytes(): Uint8Array { + const length = this.varint(); + const end = this.offset + length; + if (end > this.buffer.length) { + throw new Error("truncated protobuf bytes"); + } + const value = this.buffer.subarray(this.offset, end); + this.offset = end; + return value; + } + + string(): string { + return new TextDecoder().decode(this.bytes()); + } + + private advance(length: number, label: string): number { + const start = this.offset; + const end = this.offset + length; + if (end > this.buffer.length) { + throw new Error(`truncated protobuf ${label}`); + } + this.offset = end; + return start; + } + + fixed64Float(): number { + const start = this.advance(8, "fixed64"); + const view = new DataView(this.buffer.buffer, this.buffer.byteOffset + start, 8); + return view.getFloat64(0, true); + } + + fixed64Uint(): bigint { + const start = this.advance(8, "fixed64"); + const view = new DataView(this.buffer.buffer, this.buffer.byteOffset + start, 8); + return view.getBigUint64(0, true); + } + + skip(wire: number): void { + if (wire === 0) { + this.varint(); + } else if (wire === 1) { + this.advance(8, "fixed64"); + } else if (wire === 2) { + this.bytes(); + } else if (wire === 5) { + this.advance(4, "fixed32"); + } else { + throw new Error(`unsupported protobuf wire type ${wire}`); + } + } +} + +function decodeAnyValue(message: Uint8Array): OtlpAnyValue { + const reader = new ProtoReader(message); + const value: OtlpAnyValue = {}; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + value.stringValue = reader.string(); + } else if (field === 2 && wire === 0) { + value.boolValue = reader.varint() !== 0; + } else if (field === 3 && wire === 0) { + value.intValue = reader.varint(); + } else if (field === 4 && wire === 1) { + value.doubleValue = reader.fixed64Float(); + } else if (field === 5 && wire === 2) { + value.arrayValue = decodeArrayValue(reader.bytes()); + } else if (field === 6 && wire === 2) { + value.kvlistValue = decodeKeyValueList(reader.bytes()); + } else if (field === 7 && wire === 2) { + value.bytesValue = reader.bytes(); + } else { + reader.skip(wire); + } + } + return value; +} + +function decodeArrayValue(message: Uint8Array): { values?: OtlpAnyValue[] } { + const reader = new ProtoReader(message); + const values: OtlpAnyValue[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + values.push(decodeAnyValue(reader.bytes())); + } else { + reader.skip(wire); + } + } + return { values }; +} + +function decodeKeyValue(message: Uint8Array): OtlpKeyValue { + const reader = new ProtoReader(message); + const entry: OtlpKeyValue = {}; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + entry.key = reader.string(); + } else if (field === 2 && wire === 2) { + entry.value = decodeAnyValue(reader.bytes()); + } else { + reader.skip(wire); + } + } + return entry; +} + +function decodeKeyValueList(message: Uint8Array): { values?: OtlpKeyValue[] } { + const reader = new ProtoReader(message); + const values: OtlpKeyValue[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + values.push(decodeKeyValue(reader.bytes())); + } else { + reader.skip(wire); + } + } + return { values }; +} + +function decodeStatus(message: Uint8Array): number { + const reader = new ProtoReader(message); + let code = 0; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 3 && wire === 0) { + code = reader.varint(); + } else { + reader.skip(wire); + } + } + return code; +} + +function decodeSpan(message: Uint8Array): OtlpSpan { + const reader = new ProtoReader(message); + const span: OtlpSpan = {}; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + span.traceId = reader.bytes(); + } else if (field === 2 && wire === 2) { + span.spanId = reader.bytes(); + } else if (field === 4 && wire === 2) { + span.parentSpanId = reader.bytes(); + } else if (field === 5 && wire === 2) { + span.name = reader.string(); + } else if (field === 8 && wire === 1) { + span.endTimeMs = Number(reader.fixed64Uint() / 1_000_000n); + } else if (field === 9 && wire === 2) { + span.attributes ??= []; + span.attributes.push(decodeKeyValue(reader.bytes())); + } else if (field === 15 && wire === 2) { + span.statusCode = decodeStatus(reader.bytes()); + } else { + reader.skip(wire); + } + } + return span; +} + +function decodeScopeSpans(message: Uint8Array): OtlpScopeSpans { + const reader = new ProtoReader(message); + const spans: OtlpSpan[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 2 && wire === 2) { + spans.push(decodeSpan(reader.bytes())); + } else { + reader.skip(wire); + } + } + return { spans }; +} + +function decodeResourceSpans(message: Uint8Array): OtlpResourceSpans { + const reader = new ProtoReader(message); + const scopeSpans: OtlpScopeSpans[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 2 && wire === 2) { + scopeSpans.push(decodeScopeSpans(reader.bytes())); + } else { + reader.skip(wire); + } + } + return { scopeSpans }; +} + +function decodeTraceRequest(body: Buffer): CapturedSpan[] { + const reader = new ProtoReader(body); + const resourceSpans: OtlpResourceSpans[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + resourceSpans.push(decodeResourceSpans(reader.bytes())); + } else { + reader.skip(wire); + } + } + const spans: CapturedSpan[] = []; + for (const resource of resourceSpans) { + for (const scopeSpans of resource.scopeSpans ?? []) { + for (const span of scopeSpans.spans ?? []) { + const name = span.name?.trim(); + if (!name) { + continue; + } + spans.push({ + attributes: spanAttributes(span), + endTimeMs: span.endTimeMs, + name, + parent: (span.parentSpanId?.length ?? 0) > 0, + parentSpanId: span.parentSpanId ? Buffer.from(span.parentSpanId).toString("hex") : "", + spanId: span.spanId ? Buffer.from(span.spanId).toString("hex") : "", + statusCode: span.statusCode ?? 0, + traceId: span.traceId ? Buffer.from(span.traceId).toString("hex") : "", + }); + } + } + } + return spans; +} + +function decodeMetric(message: Uint8Array): CapturedMetric | undefined { + const reader = new ProtoReader(message); + let name = ""; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + name = reader.string(); + } else { + reader.skip(wire); + } + } + const normalizedName = name.trim(); + return normalizedName ? { name: normalizedName } : undefined; +} + +function decodeScopeMetrics(message: Uint8Array): CapturedMetric[] { + const reader = new ProtoReader(message); + const metrics: CapturedMetric[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 2 && wire === 2) { + const metric = decodeMetric(reader.bytes()); + if (metric) { + metrics.push(metric); + } + } else { + reader.skip(wire); + } + } + return metrics; +} + +function decodeResourceMetrics(message: Uint8Array): CapturedMetric[] { + const reader = new ProtoReader(message); + const metrics: CapturedMetric[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 2 && wire === 2) { + metrics.push(...decodeScopeMetrics(reader.bytes())); + } else { + reader.skip(wire); + } + } + return metrics; +} + +function decodeMetricRequest(body: Buffer): CapturedMetric[] { + const reader = new ProtoReader(body); + const metrics: CapturedMetric[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + metrics.push(...decodeResourceMetrics(reader.bytes())); + } else { + reader.skip(wire); + } + } + return metrics; +} + +function decodeLogRecord(message: Uint8Array): CapturedLogRecord { + const reader = new ProtoReader(message); + let body: string | number | boolean | string[] = ""; + let traceId = ""; + let spanId = ""; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 5 && wire === 2) { + body = normalizeOtlpValue(decodeAnyValue(reader.bytes())); + } else if (field === 9 && wire === 2) { + traceId = Buffer.from(reader.bytes()).toString("hex"); + } else if (field === 10 && wire === 2) { + spanId = Buffer.from(reader.bytes()).toString("hex"); + } else { + reader.skip(wire); + } + } + return { body, spanId, traceId }; +} + +function decodeScopeLogs(message: Uint8Array): CapturedLogRecord[] { + const reader = new ProtoReader(message); + const records: CapturedLogRecord[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 2 && wire === 2) { + records.push(decodeLogRecord(reader.bytes())); + } else { + reader.skip(wire); + } + } + return records; +} + +function decodeResourceLogs(message: Uint8Array): CapturedLogRecord[] { + const reader = new ProtoReader(message); + const records: CapturedLogRecord[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 2 && wire === 2) { + records.push(...decodeScopeLogs(reader.bytes())); + } else { + reader.skip(wire); + } + } + return records; +} + +function decodeLogRequest(body: Buffer): CapturedLogRecord[] { + const reader = new ProtoReader(body); + const records: CapturedLogRecord[] = []; + while (!reader.done()) { + const { field, wire } = reader.tag(); + if (field === 1 && wire === 2) { + records.push(...decodeResourceLogs(reader.bytes())); + } else { + reader.skip(wire); + } + } + return records; +} + +function closeLocalOtlpReceiverConnections( + server: ReturnType, + sockets: Set, +): void { + for (const socket of sockets) { + socket.destroy(); + } + server.closeAllConnections(); +} + +export function startLocalOtlpReceiver(disallowedBodyNeedles: string[] = []) { + const capturedRequests: CapturedRequest[] = []; + const capturedSpans: CapturedSpan[] = []; + const capturedMetrics: CapturedMetric[] = []; + const capturedLogRecords: CapturedLogRecord[] = []; + const capturedBodyText: Partial> = {}; + const sockets = new Set(); + const server = createServer((req: IncomingMessage, res: ServerResponse) => { + void (async () => { + if (req.method !== "POST" || !req.url) { + res.writeHead(404, { "content-type": "text/plain" }); + res.end("not found"); + return; + } + const requestPath = req.url; + const signal = OTLP_SIGNAL_PATHS.get(requestPath); + if (!signal) { + res.writeHead(404, { "content-type": "text/plain" }); + res.end("not found"); + return; + } + + const contentEncoding = headerValue(req.headers["content-encoding"]); + let body: Buffer; + try { + body = decodeRequestBody(await readRequestBody(req), contentEncoding); + } catch (error) { + const statusCode = + typeof (error as { statusCode?: unknown }).statusCode === "number" + ? (error as { statusCode: number }).statusCode + : 400; + capturedRequests.push({ + path: requestPath, + signal, + bytes: 0, + contentEncoding, + status: statusCode, + spanCount: 0, + metricCount: 0, + logCount: 0, + }); + res.writeHead(statusCode, { "content-type": "text/plain" }); + res.end(error instanceof Error ? error.message : String(error)); + return; + } + let spans: CapturedSpan[]; + let metrics: CapturedMetric[]; + let logRecords: CapturedLogRecord[]; + try { + spans = signal === "traces" ? decodeTraceRequest(body) : []; + metrics = signal === "metrics" ? decodeMetricRequest(body) : []; + logRecords = signal === "logs" ? decodeLogRequest(body) : []; + appendCapturedBodyText(capturedBodyText, signal, body, undefined, disallowedBodyNeedles); + } catch (error) { + appendCapturedBodyText(capturedBodyText, signal, body, undefined, disallowedBodyNeedles); + capturedRequests.push({ + path: requestPath, + signal, + bytes: body.length, + contentEncoding, + status: 400, + spanCount: 0, + metricCount: 0, + logCount: 0, + }); + res.writeHead(400, { "content-type": "text/plain" }); + res.end(error instanceof Error ? error.message : String(error)); + return; + } + capturedSpans.push(...spans); + capturedMetrics.push(...metrics); + capturedLogRecords.push(...logRecords); + capturedRequests.push({ + path: requestPath, + signal, + bytes: body.length, + contentEncoding, + receivedAtMs: Date.now(), + status: 200, + spanCount: spans.length, + metricCount: metrics.length, + logCount: logRecords.length, + }); + res.writeHead(200, { "content-type": "application/x-protobuf" }); + res.end(); + })(); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.once("close", () => { + sockets.delete(socket); + }); + }); + let closePromise: Promise | undefined; + + return { + capturedRequests, + capturedSpans, + capturedMetrics, + capturedLogRecords, + capturedBodyText, + async listen(): Promise { + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("failed to bind local OTLP receiver"); + } + return address.port; + }, + async close(): Promise { + closePromise ??= new Promise((resolve, reject) => { + closeLocalOtlpReceiverConnections(server, sockets); + server.close((error) => (error ? reject(error) : resolve())); + closeLocalOtlpReceiverConnections(server, sockets); + }); + await closePromise; + }, + }; +} diff --git a/test/e2e/qa-lab/runtime/qa-otel-smoke-runtime.ts b/test/e2e/qa-lab/runtime/qa-otel-smoke-runtime.ts index 3cac4f1b27c6..c031fc0c58de 100644 --- a/test/e2e/qa-lab/runtime/qa-otel-smoke-runtime.ts +++ b/test/e2e/qa-lab/runtime/qa-otel-smoke-runtime.ts @@ -4,12 +4,11 @@ import { spawn } from "node:child_process"; /* oxlint-disable typescript/unbound-method -- the original stream method is invoked with process.stdout through Reflect.apply below. */ import { randomUUID } from "node:crypto"; import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; -import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import { createServer } from "node:http"; import { Socket } from "node:net"; import { tmpdir } from "node:os"; import path from "node:path"; import { pathToFileURL } from "node:url"; -import { gunzipSync } from "node:zlib"; import { createDiagnosticTraceContext, emitTrustedDiagnosticEvent, @@ -21,42 +20,23 @@ import { type OpenClawPluginServiceContext, } from "../../../../extensions/diagnostics-otel/runtime-api.js"; import { onTrustedInternalDiagnosticEvent } from "../../../../src/infra/diagnostic-events.js"; +import { + appendCapturedBodyText, + type CapturedLogRecord, + type CapturedMetric, + type CapturedRequest, + type CapturedSpan, + decodeRequestBody, + type OtlpSignal, + readPositiveIntegerEnv, + readRequestBody, + startLocalOtlpReceiver, +} from "./otel-test-support.js"; import { createQaScriptEvidenceWriter } from "./script-evidence.js"; type CollectorMode = "local" | "docker"; type OtelLogsExporter = "otlp" | "stdout" | "both"; -type OtlpAnyValue = { - stringValue?: string; - boolValue?: boolean; - intValue?: number | string | { toString(): string }; - doubleValue?: number; - arrayValue?: { values?: OtlpAnyValue[] }; - kvlistValue?: { values?: OtlpKeyValue[] }; - bytesValue?: Uint8Array; -}; - -type OtlpKeyValue = { - key?: string; - value?: OtlpAnyValue; -}; - -type OtlpSpan = { - name?: string; - parentSpanId?: Uint8Array; - attributes?: OtlpKeyValue[]; -}; - -type OtlpScopeSpans = { - spans?: OtlpSpan[]; -}; - -type OtlpResourceSpans = { - scopeSpans?: OtlpScopeSpans[]; -}; - -type OtlpSignal = "logs" | "metrics" | "traces"; - type CliOptions = { collectorMode: CollectorMode; logsExporter: OtelLogsExporter; @@ -71,33 +51,6 @@ type OtelSmokeEvidenceContext = { let activeEvidenceContext: OtelSmokeEvidenceContext | undefined; -type CapturedRequest = { - path: string; - signal: OtlpSignal; - bytes: number; - contentEncoding?: string; - status: number; - spanCount: number; - metricCount: number; - logCount: number; -}; - -type CapturedSpan = { - name: string; - parent: boolean; - attributes: Record; -}; - -type CapturedMetric = { - name: string; -}; - -type CapturedLogRecord = { - body: string | number | boolean | string[]; - spanId: string; - traceId: string; -}; - type StdoutDiagnosticLogRecord = { signal: "openclaw.diagnostic.log"; ts?: unknown; @@ -114,11 +67,6 @@ type StdoutDiagnosticLogRecord = { const DEFAULT_DOCKER_COLLECTOR_IMAGE = process.env.OPENCLAW_QA_OTEL_COLLECTOR_IMAGE || "otel/opentelemetry-collector:0.104.0"; -const OTLP_SIGNAL_PATHS = new Map([ - ["/v1/traces", "traces"], - ["/v1/metrics", "metrics"], - ["/v1/logs", "logs"], -]); const REQUIRED_SPAN_NAMES = [ "openclaw.run", "openclaw.harness.run", @@ -154,19 +102,6 @@ const DISALLOWED_BODY_NEEDLES = [ DIRECT_CALL_ID, ]; const COLLECTOR_OUTPUT_TAIL_BYTES = 16_000; -const POSITIVE_INTEGER_PATTERN = /^[1-9]\d*$/u; -const MAX_OTLP_COMPRESSED_BODY_BYTES = readPositiveIntegerEnv( - "OPENCLAW_QA_OTEL_MAX_COMPRESSED_BODY_BYTES", - 2 * 1024 * 1024, -); -const MAX_OTLP_DECODED_BODY_BYTES = readPositiveIntegerEnv( - "OPENCLAW_QA_OTEL_MAX_DECODED_BODY_BYTES", - 8 * 1024 * 1024, -); -const MAX_CAPTURED_BODY_TEXT_BYTES = readPositiveIntegerEnv( - "OPENCLAW_QA_OTEL_MAX_CAPTURED_BODY_TEXT_BYTES", - 512 * 1024, -); const MAX_STDOUT_DIAGNOSTIC_LINE_BYTES = readPositiveIntegerEnv( "OPENCLAW_QA_OTEL_MAX_STDOUT_DIAGNOSTIC_LINE_BYTES", 512 * 1024, @@ -191,42 +126,10 @@ const QA_OTEL_ENV_TO_CLEAR = [ "OTEL_RESOURCE_ATTRIBUTES", ] as const; -function readPositiveIntegerEnv( - name: string, - fallback: number, - env: NodeJS.ProcessEnv = process.env, -): number { - const raw = env[name]; - if (raw == null || raw.trim() === "") { - return fallback; - } - const value = raw.trim(); - if (!POSITIVE_INTEGER_PATTERN.test(value)) { - throw new Error(`${name} must be a positive integer`); - } - const parsed = Number(value); - if (!Number.isSafeInteger(parsed)) { - throw new Error(`${name} must be a safe integer`); - } - return parsed; -} - function createOtelSmokeRunId(): string { return `${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`; } -function oversizedBodyError( - label: string, - actualBytes: number, - maxBytes: number, -): Error & { - statusCode: number; -} { - return Object.assign(new Error(`${label} exceeded ${maxBytes} bytes: ${actualBytes} bytes`), { - statusCode: 413, - }); -} - function usage(): string { return `Usage: pnpm qa:otel:smoke [--collector local|docker] [--logs-exporter otlp|stdout|both] [--output-dir ] @@ -299,632 +202,6 @@ function disallowedBodyNeedles(): string[] { return [...DISALLOWED_BODY_NEEDLES]; } -async function readRequestBody( - req: IncomingMessage, - maxBytes = MAX_OTLP_COMPRESSED_BODY_BYTES, -): Promise { - const chunks: Buffer[] = []; - let totalBytes = 0; - for await (const chunk of req) { - const buffer = Buffer.from(chunk); - totalBytes += buffer.length; - if (totalBytes > maxBytes) { - req.destroy(); - throw oversizedBodyError("compressed OTLP request body", totalBytes, maxBytes); - } - chunks.push(buffer); - } - return Buffer.concat(chunks); -} - -function headerValue(value: string | string[] | undefined): string | undefined { - return Array.isArray(value) ? value[0] : value; -} - -function decodeRequestBody( - body: Buffer, - contentEncoding: string | undefined, - maxBytes = MAX_OTLP_DECODED_BODY_BYTES, -): Buffer { - const normalizedEncoding = contentEncoding?.trim().toLowerCase(); - if (body.length > maxBytes && (!normalizedEncoding || normalizedEncoding === "identity")) { - throw oversizedBodyError("OTLP request body", body.length, maxBytes); - } - if (!normalizedEncoding || normalizedEncoding === "identity") { - return body; - } - if (normalizedEncoding === "gzip") { - let decoded: Buffer; - try { - decoded = gunzipSync(body, { maxOutputLength: maxBytes }); - } catch (error) { - const code = (error as { code?: unknown }).code; - const message = error instanceof Error ? error.message : String(error); - if (code === "ERR_BUFFER_TOO_LARGE" || /maxOutputLength|larger than/u.test(message)) { - throw oversizedBodyError("decoded OTLP request body", maxBytes + 1, maxBytes); - } - throw error; - } - if (decoded.length > maxBytes) { - throw oversizedBodyError("decoded OTLP request body", decoded.length, maxBytes); - } - return decoded; - } - throw new Error(`unsupported OTLP content-encoding ${contentEncoding}`); -} - -function appendCapturedBodyText( - capturedBodyText: Partial>, - signal: OtlpSignal, - body: Buffer, - maxBytes = MAX_CAPTURED_BODY_TEXT_BYTES, - disallowedNeedles: string[] = [], -): void { - const currentEntries = capturedBodyText[signal] ?? []; - const leakEntries = currentEntries.filter((entry) => entry.startsWith("[detected leak needle] ")); - const currentTail = currentEntries - .filter((entry) => !entry.startsWith("[detected leak needle] ")) - .join("\n"); - const bodyText = body.toString("utf8"); - const next = currentTail ? `${currentTail}\n${bodyText}` : bodyText; - const buffer = Buffer.from(next); - const nextLeakEntries = [ - ...leakEntries, - ...disallowedNeedles - .filter((needle) => bodyText.includes(needle)) - .map((needle) => `[detected leak needle] ${needle}`), - ].slice(-20); - const tailEntry = - buffer.length > maxBytes - ? `[captured body text truncated to last ${maxBytes} bytes]\n${buffer - .subarray(buffer.length - maxBytes) - .toString("utf8")}` - : next; - capturedBodyText[signal] = [...nextLeakEntries, tailEntry]; -} - -function normalizeOtlpValue(value: OtlpAnyValue | undefined): string | number | boolean | string[] { - if (!value) { - return ""; - } - if (typeof value.stringValue === "string") { - return value.stringValue; - } - if (typeof value.boolValue === "boolean") { - return value.boolValue; - } - if (typeof value.doubleValue === "number") { - return value.doubleValue; - } - if (value.intValue !== undefined) { - return Number(value.intValue.toString()); - } - if (value.arrayValue?.values) { - return value.arrayValue.values.map((entry) => String(normalizeOtlpValue(entry))); - } - if (value.kvlistValue?.values) { - return value.kvlistValue.values - .map((entry) => `${entry.key ?? ""}=${String(normalizeOtlpValue(entry.value))}`) - .filter(Boolean); - } - if (value.bytesValue) { - return Buffer.from(value.bytesValue).toString("hex"); - } - return ""; -} - -function spanAttributes(span: OtlpSpan): Record { - const attributes: Record = {}; - for (const attribute of span.attributes ?? []) { - const key = attribute.key?.trim(); - if (!key) { - continue; - } - attributes[key] = normalizeOtlpValue(attribute.value); - } - return attributes; -} - -class ProtoReader { - private readonly buffer: Uint8Array; - private offset = 0; - - constructor(buffer: Uint8Array) { - this.buffer = buffer; - } - - done(): boolean { - return this.offset >= this.buffer.length; - } - - tag() { - const raw = this.varint(); - return { field: raw >>> 3, wire: raw & 0x7 }; - } - - varint(): number { - let result = 0; - let shift = 0; - while (this.offset < this.buffer.length) { - const byte = this.buffer.at(this.offset); - if (byte === undefined) { - throw new Error("truncated protobuf varint"); - } - this.offset += 1; - result += (byte & 0x7f) * 2 ** shift; - if ((byte & 0x80) === 0) { - return result; - } - shift += 7; - } - throw new Error("truncated protobuf varint"); - } - - bytes(): Uint8Array { - const length = this.varint(); - const end = this.offset + length; - if (end > this.buffer.length) { - throw new Error("truncated protobuf bytes"); - } - const value = this.buffer.subarray(this.offset, end); - this.offset = end; - return value; - } - - string(): string { - return new TextDecoder().decode(this.bytes()); - } - - private advance(length: number, label: string): number { - const start = this.offset; - const end = this.offset + length; - if (end > this.buffer.length) { - throw new Error(`truncated protobuf ${label}`); - } - this.offset = end; - return start; - } - - fixed64(): number { - const start = this.advance(8, "fixed64"); - const view = new DataView(this.buffer.buffer, this.buffer.byteOffset + start, 8); - return view.getFloat64(0, true); - } - - skip(wire: number) { - if (wire === 0) { - this.varint(); - } else if (wire === 1) { - this.advance(8, "fixed64"); - } else if (wire === 2) { - this.bytes(); - } else if (wire === 5) { - this.advance(4, "fixed32"); - } else { - throw new Error(`unsupported protobuf wire type ${wire}`); - } - } -} - -function decodeAnyValue(message: Uint8Array): OtlpAnyValue { - const reader = new ProtoReader(message); - const value: OtlpAnyValue = {}; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - value.stringValue = reader.string(); - } else if (field === 2 && wire === 0) { - value.boolValue = reader.varint() !== 0; - } else if (field === 3 && wire === 0) { - value.intValue = reader.varint(); - } else if (field === 4 && wire === 1) { - value.doubleValue = reader.fixed64(); - } else if (field === 5 && wire === 2) { - value.arrayValue = decodeArrayValue(reader.bytes()); - } else if (field === 6 && wire === 2) { - value.kvlistValue = decodeKeyValueList(reader.bytes()); - } else if (field === 7 && wire === 2) { - value.bytesValue = reader.bytes(); - } else { - reader.skip(wire); - } - } - return value; -} - -function decodeArrayValue(message: Uint8Array): { values?: OtlpAnyValue[] } { - const reader = new ProtoReader(message); - const values: OtlpAnyValue[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - values.push(decodeAnyValue(reader.bytes())); - } else { - reader.skip(wire); - } - } - return { values }; -} - -function decodeKeyValue(message: Uint8Array): OtlpKeyValue { - const reader = new ProtoReader(message); - const entry: OtlpKeyValue = {}; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - entry.key = reader.string(); - } else if (field === 2 && wire === 2) { - entry.value = decodeAnyValue(reader.bytes()); - } else { - reader.skip(wire); - } - } - return entry; -} - -function decodeKeyValueList(message: Uint8Array): { values?: OtlpKeyValue[] } { - const reader = new ProtoReader(message); - const values: OtlpKeyValue[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - values.push(decodeKeyValue(reader.bytes())); - } else { - reader.skip(wire); - } - } - return { values }; -} - -function decodeSpan(message: Uint8Array): OtlpSpan { - const reader = new ProtoReader(message); - const span: OtlpSpan = {}; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 4 && wire === 2) { - span.parentSpanId = reader.bytes(); - } else if (field === 5 && wire === 2) { - span.name = reader.string(); - } else if (field === 9 && wire === 2) { - span.attributes ??= []; - span.attributes.push(decodeKeyValue(reader.bytes())); - } else { - reader.skip(wire); - } - } - return span; -} - -function decodeScopeSpans(message: Uint8Array): OtlpScopeSpans { - const reader = new ProtoReader(message); - const spans: OtlpSpan[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 2 && wire === 2) { - spans.push(decodeSpan(reader.bytes())); - } else { - reader.skip(wire); - } - } - return { spans }; -} - -function decodeResourceSpans(message: Uint8Array): OtlpResourceSpans { - const reader = new ProtoReader(message); - const scopeSpans: OtlpScopeSpans[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 2 && wire === 2) { - scopeSpans.push(decodeScopeSpans(reader.bytes())); - } else { - reader.skip(wire); - } - } - return { scopeSpans }; -} - -function decodeTraceRequest(body: Buffer): CapturedSpan[] { - const reader = new ProtoReader(body); - const resourceSpans: OtlpResourceSpans[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - resourceSpans.push(decodeResourceSpans(reader.bytes())); - } else { - reader.skip(wire); - } - } - const spans: CapturedSpan[] = []; - for (const resource of resourceSpans) { - for (const scopeSpans of resource.scopeSpans ?? []) { - for (const span of scopeSpans.spans ?? []) { - const name = span.name?.trim(); - if (!name) { - continue; - } - spans.push({ - name, - parent: (span.parentSpanId?.length ?? 0) > 0, - attributes: spanAttributes(span), - }); - } - } - } - return spans; -} - -function decodeMetric(message: Uint8Array): CapturedMetric | undefined { - const reader = new ProtoReader(message); - let name = ""; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - name = reader.string(); - } else { - reader.skip(wire); - } - } - const normalizedName = name.trim(); - return normalizedName ? { name: normalizedName } : undefined; -} - -function decodeScopeMetrics(message: Uint8Array): CapturedMetric[] { - const reader = new ProtoReader(message); - const metrics: CapturedMetric[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 2 && wire === 2) { - const metric = decodeMetric(reader.bytes()); - if (metric) { - metrics.push(metric); - } - } else { - reader.skip(wire); - } - } - return metrics; -} - -function decodeResourceMetrics(message: Uint8Array): CapturedMetric[] { - const reader = new ProtoReader(message); - const metrics: CapturedMetric[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 2 && wire === 2) { - metrics.push(...decodeScopeMetrics(reader.bytes())); - } else { - reader.skip(wire); - } - } - return metrics; -} - -function decodeMetricRequest(body: Buffer): CapturedMetric[] { - const reader = new ProtoReader(body); - const metrics: CapturedMetric[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - metrics.push(...decodeResourceMetrics(reader.bytes())); - } else { - reader.skip(wire); - } - } - return metrics; -} - -function decodeLogRecord(message: Uint8Array): CapturedLogRecord { - const reader = new ProtoReader(message); - let body: string | number | boolean | string[] = ""; - let traceId = ""; - let spanId = ""; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 5 && wire === 2) { - body = normalizeOtlpValue(decodeAnyValue(reader.bytes())); - } else if (field === 9 && wire === 2) { - traceId = Buffer.from(reader.bytes()).toString("hex"); - } else if (field === 10 && wire === 2) { - spanId = Buffer.from(reader.bytes()).toString("hex"); - } else { - reader.skip(wire); - } - } - return { body, spanId, traceId }; -} - -function decodeScopeLogs(message: Uint8Array): CapturedLogRecord[] { - const reader = new ProtoReader(message); - const records: CapturedLogRecord[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 2 && wire === 2) { - records.push(decodeLogRecord(reader.bytes())); - } else { - reader.skip(wire); - } - } - return records; -} - -function decodeResourceLogs(message: Uint8Array): CapturedLogRecord[] { - const reader = new ProtoReader(message); - const records: CapturedLogRecord[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 2 && wire === 2) { - records.push(...decodeScopeLogs(reader.bytes())); - } else { - reader.skip(wire); - } - } - return records; -} - -function decodeLogRequest(body: Buffer): CapturedLogRecord[] { - const reader = new ProtoReader(body); - const records: CapturedLogRecord[] = []; - while (!reader.done()) { - const { field, wire } = reader.tag(); - if (field === 1 && wire === 2) { - records.push(...decodeResourceLogs(reader.bytes())); - } else { - reader.skip(wire); - } - } - return records; -} - -function startLocalOtlpReceiver(disallowedBodyNeedlesLocal: string[] = []) { - const capturedRequests: CapturedRequest[] = []; - const capturedSpans: CapturedSpan[] = []; - const capturedMetrics: CapturedMetric[] = []; - const capturedLogRecords: CapturedLogRecord[] = []; - const capturedBodyText: Partial> = {}; - const sockets = new Set(); - const server = createServer((req: IncomingMessage, res: ServerResponse) => { - void (async () => { - if (req.method !== "POST" || !req.url) { - res.writeHead(404, { "content-type": "text/plain" }); - res.end("not found"); - return; - } - const requestPath = req.url; - const signal = OTLP_SIGNAL_PATHS.get(requestPath); - if (!signal) { - res.writeHead(404, { "content-type": "text/plain" }); - res.end("not found"); - return; - } - - const contentEncoding = headerValue(req.headers["content-encoding"]); - let body: Buffer; - try { - const compressedBody = await readRequestBody(req); - body = decodeRequestBody(compressedBody, contentEncoding); - } catch (error) { - const statusCode = - typeof (error as { statusCode?: unknown }).statusCode === "number" - ? (error as { statusCode: number }).statusCode - : 400; - capturedRequests.push({ - path: requestPath, - signal, - bytes: 0, - contentEncoding, - status: statusCode, - spanCount: 0, - metricCount: 0, - logCount: 0, - }); - res.writeHead(statusCode, { "content-type": "text/plain" }); - res.end(error instanceof Error ? error.message : String(error)); - return; - } - let spans: CapturedSpan[]; - let metrics: CapturedMetric[]; - let logRecords: CapturedLogRecord[]; - try { - spans = signal === "traces" ? decodeTraceRequest(body) : []; - metrics = signal === "metrics" ? decodeMetricRequest(body) : []; - logRecords = signal === "logs" ? decodeLogRequest(body) : []; - appendCapturedBodyText( - capturedBodyText, - signal, - body, - undefined, - disallowedBodyNeedlesLocal, - ); - } catch (error) { - appendCapturedBodyText( - capturedBodyText, - signal, - body, - undefined, - disallowedBodyNeedlesLocal, - ); - capturedRequests.push({ - path: requestPath, - signal, - bytes: body.length, - contentEncoding, - status: 400, - spanCount: 0, - metricCount: 0, - logCount: 0, - }); - res.writeHead(400, { "content-type": "text/plain" }); - res.end(error instanceof Error ? error.message : String(error)); - return; - } - if (spans.length > 0) { - capturedSpans.push(...spans); - } - if (metrics.length > 0) { - capturedMetrics.push(...metrics); - } - if (logRecords.length > 0) { - capturedLogRecords.push(...logRecords); - } - capturedRequests.push({ - path: requestPath, - signal, - bytes: body.length, - contentEncoding, - status: 200, - spanCount: spans.length, - metricCount: metrics.length, - logCount: logRecords.length, - }); - res.writeHead(200, { "content-type": "application/x-protobuf" }); - res.end(); - })(); - }); - server.on("connection", (socket) => { - sockets.add(socket); - socket.once("close", () => { - sockets.delete(socket); - }); - }); - let closePromise: Promise | undefined; - - return { - capturedRequests, - capturedSpans, - capturedMetrics, - capturedLogRecords, - capturedBodyText, - async listen(): Promise { - await new Promise((resolve) => { - server.listen(0, "127.0.0.1", resolve); - }); - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("failed to bind local OTLP receiver"); - } - return address.port; - }, - async close(): Promise { - closePromise ??= new Promise((resolve, reject) => { - closeLocalOtlpReceiverConnections(server, sockets); - server.close((err) => (err ? reject(err) : resolve())); - closeLocalOtlpReceiverConnections(server, sockets); - }); - await closePromise; - }, - }; -} - -function closeLocalOtlpReceiverConnections( - server: ReturnType, - sockets: Set, -): void { - for (const socket of sockets) { - socket.destroy(); - } - server.closeAllConnections(); -} - async function reserveLocalPort(): Promise { const server = createServer(); await new Promise((resolve) => { diff --git a/test/scripts/local-heavy-check-runtime.test.ts b/test/scripts/local-heavy-check-runtime.test.ts index 1236ad0d5a89..50ffdd068c48 100644 --- a/test/scripts/local-heavy-check-runtime.test.ts +++ b/test/scripts/local-heavy-check-runtime.test.ts @@ -148,6 +148,7 @@ describe("local-heavy-check-runtime", () => { "--checkers", "1", ]); + expect(env.GOMAXPROCS).toBe("1"); expect(env.GOGC).toBe("30"); expect(env.GOMEMLIMIT).toBe("3GiB"); }); @@ -162,6 +163,7 @@ describe("local-heavy-check-runtime", () => { const { args, env } = applyLocalTsgoPolicy( ["--checkers", "4", "--singleThreaded", "--pprofDir", "/tmp/existing"], makeEnv({ + GOMAXPROCS: "3", GOGC: "80", GOMEMLIMIT: "5GiB", OPENCLAW_TSGO_PPROF_DIR: "/tmp/profile", @@ -178,6 +180,7 @@ describe("local-heavy-check-runtime", () => { "--declaration", "false", ]); + expect(env.GOMAXPROCS).toBe("3"); expect(env.GOGC).toBe("80"); expect(env.GOMEMLIMIT).toBe("5GiB"); }); @@ -201,6 +204,7 @@ describe("local-heavy-check-runtime", () => { "--tsBuildInfoFile", ".artifacts/tsgo-cache/root.tsbuildinfo", ]); + expect(env.GOMAXPROCS).toBeUndefined(); expect(env.GOGC).toBeUndefined(); expect(env.GOMEMLIMIT).toBeUndefined(); }); @@ -253,6 +257,7 @@ describe("local-heavy-check-runtime", () => { "--checkers", "1", ]); + expect(env.GOMAXPROCS).toBe("1"); expect(env.GOGC).toBe("30"); expect(env.GOMEMLIMIT).toBe("3GiB"); }); @@ -273,6 +278,7 @@ describe("local-heavy-check-runtime", () => { "--tsBuildInfoFile", ".artifacts/tsgo-cache/root.tsbuildinfo", ]); + expect(env.GOMAXPROCS).toBeUndefined(); expect(env.GOGC).toBeUndefined(); expect(env.GOMEMLIMIT).toBeUndefined(); });