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";

type CollectorMode = "local" | "docker";

"openclaw.context.assembled",

"openclaw.message.delivery",

] as const;

const REQUIRED_METRIC_NAMES = [

"openclaw.harness.duration_ms",

] as const;

const REQUIRED_METRIC_NAMES = ["openclaw.harness.duration_ms"] as const;

const DISALLOWED_ATTRIBUTE_KEYS = new Set([

"openclaw.runId",

"openclaw.chatId",

"openclaw.call_id",

"openclaw.tool_call_id",

]);

const DISALLOWED_BODY_NEEDLES = [

"OTEL-QA-SECRET",

"OTEL-QA-OK",

];

const DISALLOWED_BODY_NEEDLES = ["OTEL-QA-SECRET", "OTEL-QA-OK"];

const COLLECTOR_OUTPUT_TAIL_BYTES = 16_000;

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,

);

function readPositiveIntegerEnv(name: string, fallback: number): number {

const parsed = Number.parseInt(process.env[name] ?? "", 10);

return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;

}

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] [--output-dir <path>] [--provider-mode <mode>] [--scenario <id>] [--model <ref>] [--alt-model <ref>]

return [...needles];

}

async function readRequestBody(req: IncomingMessage): Promise<Buffer> {

async function readRequestBody(

req: IncomingMessage,

maxBytes = MAX_OTLP_COMPRESSED_BODY_BYTES,

): Promise<Buffer> {

const chunks: Buffer[] = [];

let totalBytes = 0;

for await (const chunk of req) {

chunks.push(Buffer.from(chunk));

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);

}

return Array.isArray(value) ? value[0] : value;

}

function decodeRequestBody(body: Buffer, contentEncoding: string | undefined): Buffer {

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") {

return gunzipSync(body);

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<Record<OtlpSignal, string[]>>,

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 "";

}

class ProtoReader {

private readonly buffer: Uint8Array;

private offset = 0;

constructor(private readonly buffer: Uint8Array) {}

constructor(buffer: Uint8Array) {

this.buffer = buffer;

}

done(): boolean {

return this.offset >= this.buffer.length;

return records;

}

function startLocalOtlpReceiver() {

function startLocalOtlpReceiver(disallowedBodyNeedles: string[] = []) {

const capturedRequests: CapturedRequest[] = [];

const capturedSpans: CapturedSpan[] = [];

const capturedMetrics: CapturedMetric[] = [];

return;

}

const compressedBody = await readRequestBody(req);

const contentEncoding = headerValue(req.headers["content-encoding"]);

const body = decodeRequestBody(compressedBody, contentEncoding);

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;

}

const spans = signal === "traces" ? decodeTraceRequest(body) : [];

const metrics = signal === "metrics" ? decodeMetricRequest(body) : [];

const logRecords = signal === "logs" ? decodeLogRequest(body) : [];

if (logRecords.length > 0) {

capturedLogRecords.push(...logRecords);

}

capturedBodyText[signal] ??= [];

capturedBodyText[signal]?.push(body.toString("utf8"));

appendCapturedBodyText(capturedBodyText, signal, body, undefined, disallowedBodyNeedles);

capturedRequests.push({

path: requestPath,

signal,

containerName,

...(useHostNetwork

? ["--network", "host"]

: [

"--add-host=host.docker.internal:host-gateway",

"-p",

`127.0.0.1:${collectorPort}:4318`,

]),

: ["--add-host=host.docker.internal:host-gateway", "-p", `127.0.0.1:${collectorPort}:4318`]),

"-v",

`${configPath}:/etc/otelcol/config.yaml:ro`,

DEFAULT_DOCKER_COLLECTOR_IMAGE,

"--config=/etc/otelcol/config.yaml",

];

const child = spawn(

"docker",

dockerArgs,

{ stdio: ["ignore", "pipe", "pipe"] },

);

const child = spawn("docker", dockerArgs, { stdio: ["ignore", "pipe", "pipe"] });

child.stdout?.on("data", (chunk) => stdout.push(String(chunk)));

child.stderr?.on("data", (chunk) => stderr.push(String(chunk)));

child.on("error", (err) => {

REQUIRED_METRIC_NAMES.every((name) => metricNames.has(name)) &&

receiver.capturedLogRecords.length > 0 &&

["traces", "metrics", "logs"].every((signal) =>

receiver.capturedRequests.some((request) => request.signal === signal)

receiver.capturedRequests.some((request) => request.signal === signal),

)

);

}

.map((record) => record.body)

.filter((body) => body !== "log");

if (rawLogBodies.length > 0) {

failures.push(

`OTLP log records exported ${rawLogBodies.length} non-placeholder bodies`,

);

failures.push(`OTLP log records exported ${rawLogBodies.length} non-placeholder bodies`);

}

const attributeKeys = collectAttributeKeys(params.spans);

}

await mkdir(options.outputDir, { recursive: true });

const receiver = startLocalOtlpReceiver();

const receiver = startLocalOtlpReceiver(disallowedBodyNeedles(options));

const port = await receiver.listen();

process.stdout.write(

`qa-otel-smoke: local OTLP receiver listening on http://127.0.0.1:${port}\n`,

);

}

main().catch((error) => {

process.stderr.write(

`qa-otel-smoke: ${error instanceof Error ? error.stack || error.message : String(error)}\n`,

);

process.exitCode = 1;

});

export const testing = {

appendCapturedBodyText,

decodeRequestBody,

readRequestBody,

};

if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) {

main().catch((error) => {

process.stderr.write(

`qa-otel-smoke: ${error instanceof Error ? error.stack || error.message : String(error)}\n`,

);

process.exitCode = 1;

});

}