mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
test(qa): cover diagnostic event boundary (#118844)
This commit is contained in:
Vincent Koc
@@ -0,0 +1,41 @@
|
||||
title: Diagnostic events composed boundary
|
||||
|
||||
scenario:
|
||||
id: diagnostic-events-boundary
|
||||
surface: telemetry
|
||||
coverage:
|
||||
primary:
|
||||
- observability.async-dispatch
|
||||
- observability.diagnostic-event-types
|
||||
- observability.model-call-diagnostic-events
|
||||
- observability.plugin-sdk-diagnostic-runtime-exports
|
||||
- observability.trusted-diagnostic-event-subscription
|
||||
- observability.trusted-trace-context
|
||||
- observability.w3c-trace-context-creation
|
||||
objective: Compose the real model-call diagnostic producer with public Plugin SDK subscribers and the trusted private subscription boundary.
|
||||
successCriteria:
|
||||
- Model-call lifecycle events remain queued until the async dispatcher drains.
|
||||
- Shared and trusted subscribers receive immutable typed events, while private model content stays trusted-only.
|
||||
- Public subscribers receive the untrusted control event but no trusted model-call events.
|
||||
- The model-call child trace replaces caller-supplied trace headers with a parseable W3C traceparent.
|
||||
- The script emits a bounded summary and standard QA evidence package.
|
||||
docsRefs:
|
||||
- docs/gateway/opentelemetry.md
|
||||
- docs/logging.md
|
||||
- docs/plugins/sdk-subpaths.md
|
||||
codeRefs:
|
||||
- src/infra/diagnostic-events.ts
|
||||
- src/infra/diagnostic-trace-context.ts
|
||||
- src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.ts
|
||||
- src/plugin-sdk/diagnostic-runtime.ts
|
||||
- test/e2e/qa-lab/runtime/diagnostic-events-boundary-runtime.ts
|
||||
execution:
|
||||
kind: script
|
||||
path: test/e2e/qa-lab/runtime/diagnostic-events-boundary-runtime.ts
|
||||
config:
|
||||
requiredProviderMode: mock-openai
|
||||
args:
|
||||
- --output-dir
|
||||
- ${outputDir}
|
||||
timeoutMs: 120000
|
||||
summary: Composed diagnostic dispatcher, trusted subscription, model-call, and W3C trace evidence.
|
||||
@@ -0,0 +1,52 @@
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import {
|
||||
runDiagnosticEventsBoundaryRuntime,
|
||||
testing,
|
||||
} from "./diagnostic-events-boundary-runtime.js";
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { force: true, recursive: true })));
|
||||
});
|
||||
|
||||
describe("diagnostic events boundary runtime", () => {
|
||||
it("composes async dispatch, trusted subscriptions, and model-call trace propagation", async () => {
|
||||
const artifactBase = await fs.mkdtemp(path.join(os.tmpdir(), "diagnostic-events-boundary-"));
|
||||
tempDirs.push(artifactBase);
|
||||
|
||||
const { evidence, summary } = await runDiagnosticEventsBoundaryRuntime({
|
||||
artifactBase,
|
||||
repoRoot: process.cwd(),
|
||||
});
|
||||
|
||||
expect(evidence.entries[0]?.result.status).toBe("pass");
|
||||
expect(summary).toMatchObject({
|
||||
deliveredBeforeDrain: 0,
|
||||
eventTypes: ["model.call.started", "model.call.completed"],
|
||||
failures: [],
|
||||
immutableEventCopies: true,
|
||||
passed: true,
|
||||
pendingBeforeDrain: true,
|
||||
privateEventCount: 2,
|
||||
publicEventTypes: ["message.queued"],
|
||||
trustedEventCount: 2,
|
||||
});
|
||||
expect(summary.propagatedTraceparent).toMatch(
|
||||
/^00-4bf92f3577b34da6a3ce929d0e0e4736-[0-9a-f]{16}-01$/u,
|
||||
);
|
||||
const summaryText = await fs.readFile(
|
||||
path.join(artifactBase, "diagnostic-events-boundary-summary.json"),
|
||||
"utf8",
|
||||
);
|
||||
expect(summaryText).not.toContain("diagnostic-boundary-private-input");
|
||||
await expect(fs.stat(path.join(artifactBase, "qa-evidence.json"))).resolves.toBeDefined();
|
||||
});
|
||||
|
||||
it("rejects missing output-dir values", () => {
|
||||
expect(() => testing.parseOptions(["--output-dir"])).toThrow("--output-dir requires a value");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,344 @@
|
||||
// Diagnostic event boundary evidence composes the real model-call producer with SDK subscribers.
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import type { StreamFn } from "openclaw/plugin-sdk/agent-core";
|
||||
import {
|
||||
type DiagnosticEventMetadata,
|
||||
type DiagnosticEventPayload,
|
||||
type DiagnosticEventPrivateData,
|
||||
createDiagnosticTraceContext,
|
||||
emitDiagnosticEvent,
|
||||
formatDiagnosticTraceparent,
|
||||
hasPendingInternalDiagnosticEvent,
|
||||
onDiagnosticEvent,
|
||||
onInternalDiagnosticEvent,
|
||||
parseDiagnosticTraceparent,
|
||||
resetDiagnosticEventsForTest,
|
||||
waitForDiagnosticEventsDrained,
|
||||
} from "openclaw/plugin-sdk/diagnostic-runtime";
|
||||
import { wrapStreamFnWithDiagnosticModelCallEvents } from "../../../../src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.js";
|
||||
import { onTrustedInternalDiagnosticEvent } from "../../../../src/infra/diagnostic-events.js";
|
||||
import { formatErrorMessage } from "../../../../src/infra/errors.js";
|
||||
import { createQaScriptEvidenceWriter } from "./script-evidence.js";
|
||||
|
||||
const RUN_ID = "qa-diagnostic-events-run";
|
||||
const CALL_ID = "qa-diagnostic-events-call";
|
||||
const PRIVATE_MODEL_INPUT = "diagnostic-boundary-private-input";
|
||||
const CALLER_TRACEPARENT = `00-${"a".repeat(32)}-${"b".repeat(16)}-00`;
|
||||
|
||||
type RuntimeOptions = {
|
||||
artifactBase: string;
|
||||
repoRoot: string;
|
||||
};
|
||||
|
||||
type ModelCallEvent = Extract<
|
||||
DiagnosticEventPayload,
|
||||
{ type: "model.call.started" | "model.call.completed" | "model.call.error" }
|
||||
>;
|
||||
|
||||
export type DiagnosticEventsBoundarySummary = {
|
||||
deliveredBeforeDrain: number;
|
||||
eventTypes: string[];
|
||||
failures: string[];
|
||||
immutableEventCopies: boolean;
|
||||
passed: boolean;
|
||||
pendingBeforeDrain: boolean;
|
||||
privateEventCount: number;
|
||||
propagatedTraceparent?: string;
|
||||
publicEventTypes: string[];
|
||||
sdkExportsExercised: string[];
|
||||
trustedEventCount: number;
|
||||
};
|
||||
|
||||
function parseOptions(argv: string[], repoRoot = process.cwd()): RuntimeOptions {
|
||||
let artifactBase = path.join(repoRoot, ".artifacts", "qa-e2e", "diagnostic-events-boundary");
|
||||
for (let index = 0; index < argv.length; index += 1) {
|
||||
const arg = argv[index];
|
||||
if (arg === "--output-dir") {
|
||||
const value = argv[++index];
|
||||
if (!value) {
|
||||
throw new Error("--output-dir requires a value");
|
||||
}
|
||||
artifactBase = path.resolve(repoRoot, value);
|
||||
continue;
|
||||
}
|
||||
if (arg === "--") {
|
||||
continue;
|
||||
}
|
||||
throw new Error(`Unknown argument: ${arg}`);
|
||||
}
|
||||
return { artifactBase, repoRoot };
|
||||
}
|
||||
|
||||
function isTargetModelCallEvent(event: DiagnosticEventPayload): event is ModelCallEvent {
|
||||
return (
|
||||
(event.type === "model.call.started" ||
|
||||
event.type === "model.call.completed" ||
|
||||
event.type === "model.call.error") &&
|
||||
event.callId === CALL_ID
|
||||
);
|
||||
}
|
||||
|
||||
function createWriter(options: RuntimeOptions) {
|
||||
return createQaScriptEvidenceWriter({
|
||||
artifactBase: options.artifactBase,
|
||||
logFileName: "diagnostic-events-boundary.log",
|
||||
primaryModel: "diagnostic-runtime/model-call-wrapper",
|
||||
providerMode: "mock-openai",
|
||||
repoRoot: options.repoRoot,
|
||||
target: {
|
||||
id: "diagnostic-events-boundary",
|
||||
title: "Diagnostic events composed boundary",
|
||||
sourcePath: "test/e2e/qa-lab/runtime/diagnostic-events-boundary-runtime.ts",
|
||||
docsRefs: [
|
||||
"docs/gateway/opentelemetry.md",
|
||||
"docs/logging.md",
|
||||
"docs/plugins/sdk-subpaths.md",
|
||||
],
|
||||
codeRefs: [
|
||||
"src/infra/diagnostic-events.ts",
|
||||
"src/infra/diagnostic-trace-context.ts",
|
||||
"src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.ts",
|
||||
"src/plugin-sdk/diagnostic-runtime.ts",
|
||||
],
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function probeDiagnosticEventsBoundary(): Promise<DiagnosticEventsBoundarySummary> {
|
||||
resetDiagnosticEventsForTest();
|
||||
const sharedEvents: Array<{ event: ModelCallEvent; metadata: DiagnosticEventMetadata }> = [];
|
||||
const trustedEvents: Array<{
|
||||
event: ModelCallEvent;
|
||||
metadata: DiagnosticEventMetadata;
|
||||
privateData: DiagnosticEventPrivateData;
|
||||
}> = [];
|
||||
const publicEventTypes: string[] = [];
|
||||
const stopShared = onInternalDiagnosticEvent((event, metadata) => {
|
||||
if (isTargetModelCallEvent(event)) {
|
||||
sharedEvents.push({ event, metadata });
|
||||
}
|
||||
});
|
||||
const stopTrusted = onTrustedInternalDiagnosticEvent((event, metadata, privateData) => {
|
||||
if (isTargetModelCallEvent(event)) {
|
||||
trustedEvents.push({ event, metadata, privateData });
|
||||
}
|
||||
});
|
||||
const stopPublic = onDiagnosticEvent((event) => {
|
||||
publicEventTypes.push(event.type);
|
||||
});
|
||||
|
||||
let propagatedTraceparent: string | undefined;
|
||||
const rootTrace = createDiagnosticTraceContext({
|
||||
traceId: "4BF92F3577B34DA6A3CE929D0E0E4736",
|
||||
spanId: "00F067AA0BA902B7",
|
||||
traceFlags: "01",
|
||||
});
|
||||
const rootTraceparent = formatDiagnosticTraceparent(rootTrace);
|
||||
const parsedRootTraceparent = parseDiagnosticTraceparent(rootTraceparent);
|
||||
const wrapped = wrapStreamFnWithDiagnosticModelCallEvents(
|
||||
((
|
||||
model: Parameters<StreamFn>[0],
|
||||
_streamContext: Parameters<StreamFn>[1],
|
||||
options: Parameters<StreamFn>[2],
|
||||
) => {
|
||||
const header = options?.headers?.traceparent;
|
||||
propagatedTraceparent = typeof header === "string" ? header : undefined;
|
||||
options?.onPayload?.({ input: PRIVATE_MODEL_INPUT }, model);
|
||||
return Promise.resolve({ role: "assistant", content: [] });
|
||||
}) as unknown as StreamFn,
|
||||
{
|
||||
runId: RUN_ID,
|
||||
provider: "openai",
|
||||
model: "gpt-5.6-luna",
|
||||
api: "responses",
|
||||
transport: "direct",
|
||||
trace: rootTrace,
|
||||
contentCapture: {
|
||||
inputMessages: true,
|
||||
outputMessages: false,
|
||||
toolInputs: false,
|
||||
toolOutputs: false,
|
||||
systemPrompt: false,
|
||||
toolDefinitions: false,
|
||||
anyModelContent: true,
|
||||
},
|
||||
nextCallId: () => CALL_ID,
|
||||
suppressPluginHooks: true,
|
||||
},
|
||||
);
|
||||
|
||||
try {
|
||||
emitDiagnosticEvent({
|
||||
type: "message.queued",
|
||||
source: "qa-diagnostic-events-boundary",
|
||||
trace: rootTrace,
|
||||
});
|
||||
await Promise.resolve(
|
||||
wrapped(
|
||||
{ id: "gpt-5.6-luna" } as never,
|
||||
{
|
||||
messages: [{ role: "user", content: PRIVATE_MODEL_INPUT }],
|
||||
} as never,
|
||||
{
|
||||
headers: { TraceParent: CALLER_TRACEPARENT },
|
||||
} as never,
|
||||
),
|
||||
);
|
||||
|
||||
const deliveredBeforeDrain = sharedEvents.length;
|
||||
const pendingBeforeDrain = hasPendingInternalDiagnosticEvent(
|
||||
(event, metadata) =>
|
||||
metadata.trusted && event.type === "model.call.completed" && event.callId === CALL_ID,
|
||||
);
|
||||
await waitForDiagnosticEventsDrained();
|
||||
|
||||
const failures: string[] = [];
|
||||
const eventTypes = sharedEvents.map(({ event }) => event.type);
|
||||
if (
|
||||
!rootTraceparent ||
|
||||
parsedRootTraceparent?.traceId !== rootTrace.traceId ||
|
||||
parsedRootTraceparent.spanId !== rootTrace.spanId ||
|
||||
parsedRootTraceparent.traceFlags !== rootTrace.traceFlags
|
||||
) {
|
||||
failures.push("W3C root trace context did not format and parse losslessly");
|
||||
}
|
||||
if (deliveredBeforeDrain !== 0) {
|
||||
failures.push(`model-call events delivered before async drain: ${deliveredBeforeDrain}`);
|
||||
}
|
||||
if (!pendingBeforeDrain) {
|
||||
failures.push("model-call completion was not pending before async drain");
|
||||
}
|
||||
if (eventTypes.join(",") !== "model.call.started,model.call.completed") {
|
||||
failures.push(`unexpected model-call event sequence: ${eventTypes.join(",")}`);
|
||||
}
|
||||
if (!sharedEvents.every(({ metadata }) => metadata.trusted)) {
|
||||
failures.push("model-call events were not dispatcher-trusted");
|
||||
}
|
||||
if (publicEventTypes.join(",") !== "message.queued") {
|
||||
failures.push(
|
||||
`public subscription observed unexpected events: ${publicEventTypes.join(",")}`,
|
||||
);
|
||||
}
|
||||
const immutableEventCopies = sharedEvents.every(
|
||||
({ event, metadata }) =>
|
||||
Object.isFrozen(event) && Object.isFrozen(event.trace) && Object.isFrozen(metadata),
|
||||
);
|
||||
if (!immutableEventCopies) {
|
||||
failures.push("diagnostic subscribers did not receive immutable event copies");
|
||||
}
|
||||
const started = sharedEvents.find(({ event }) => event.type === "model.call.started")?.event;
|
||||
const modelTrace = started?.trace;
|
||||
if (
|
||||
!modelTrace?.spanId ||
|
||||
modelTrace.traceId !== rootTrace.traceId ||
|
||||
modelTrace.parentSpanId !== rootTrace.spanId
|
||||
) {
|
||||
failures.push("model-call event did not create a child of the trusted root trace");
|
||||
}
|
||||
const expectedTraceparent = formatDiagnosticTraceparent(modelTrace);
|
||||
if (!expectedTraceparent || propagatedTraceparent !== expectedTraceparent) {
|
||||
failures.push("provider traceparent did not match the emitted model-call child trace");
|
||||
}
|
||||
if (propagatedTraceparent === CALLER_TRACEPARENT) {
|
||||
failures.push("caller-supplied traceparent was not replaced by trusted runtime context");
|
||||
}
|
||||
const parsedProviderTraceparent = parseDiagnosticTraceparent(propagatedTraceparent);
|
||||
if (
|
||||
!modelTrace?.spanId ||
|
||||
!parsedProviderTraceparent ||
|
||||
parsedProviderTraceparent.traceId !== modelTrace.traceId ||
|
||||
parsedProviderTraceparent.spanId !== modelTrace.spanId ||
|
||||
parsedProviderTraceparent.traceFlags !== modelTrace.traceFlags
|
||||
) {
|
||||
failures.push("provider traceparent did not parse back to the model-call trace");
|
||||
}
|
||||
if (trustedEvents.length !== 2) {
|
||||
failures.push(`trusted subscription observed ${trustedEvents.length} model-call events`);
|
||||
}
|
||||
if (
|
||||
!trustedEvents.every(
|
||||
({ metadata, privateData }) =>
|
||||
metadata.trusted &&
|
||||
Object.isFrozen(privateData) &&
|
||||
JSON.stringify(privateData).includes(PRIVATE_MODEL_INPUT),
|
||||
)
|
||||
) {
|
||||
failures.push("trusted subscription did not receive immutable private model content");
|
||||
}
|
||||
if (JSON.stringify(sharedEvents).includes(PRIVATE_MODEL_INPUT)) {
|
||||
failures.push("private model content leaked into shared diagnostic event payloads");
|
||||
}
|
||||
|
||||
return {
|
||||
deliveredBeforeDrain,
|
||||
eventTypes,
|
||||
failures,
|
||||
immutableEventCopies,
|
||||
passed: failures.length === 0,
|
||||
pendingBeforeDrain,
|
||||
privateEventCount: trustedEvents.length,
|
||||
propagatedTraceparent,
|
||||
publicEventTypes,
|
||||
sdkExportsExercised: [
|
||||
"createDiagnosticTraceContext",
|
||||
"emitDiagnosticEvent",
|
||||
"formatDiagnosticTraceparent",
|
||||
"hasPendingInternalDiagnosticEvent",
|
||||
"onDiagnosticEvent",
|
||||
"onInternalDiagnosticEvent",
|
||||
"parseDiagnosticTraceparent",
|
||||
"resetDiagnosticEventsForTest",
|
||||
"waitForDiagnosticEventsDrained",
|
||||
],
|
||||
trustedEventCount: sharedEvents.filter(({ metadata }) => metadata.trusted).length,
|
||||
};
|
||||
} finally {
|
||||
stopPublic();
|
||||
stopShared();
|
||||
stopTrusted();
|
||||
resetDiagnosticEventsForTest();
|
||||
}
|
||||
}
|
||||
|
||||
export async function runDiagnosticEventsBoundaryRuntime(options: RuntimeOptions) {
|
||||
const writer = createWriter(options);
|
||||
const startedAt = Date.now();
|
||||
const summary = await probeDiagnosticEventsBoundary();
|
||||
const summaryPath = path.join(options.artifactBase, "diagnostic-events-boundary-summary.json");
|
||||
await fs.mkdir(options.artifactBase, { recursive: true });
|
||||
await fs.writeFile(summaryPath, `${JSON.stringify(summary, null, 2)}\n`, "utf8");
|
||||
writer.appendLog(
|
||||
`diagnostic-events-boundary: ${summary.passed ? "passed" : "failed"} ` +
|
||||
`events=${summary.eventTypes.join(",")} traceparent=${summary.propagatedTraceparent ?? "missing"}\n`,
|
||||
);
|
||||
const evidence = await writer.write({
|
||||
artifacts: [{ kind: "summary", filePath: summaryPath }],
|
||||
details: summary.passed
|
||||
? `trusted model-call events=${summary.trustedEventCount}`
|
||||
: summary.failures.join("\n"),
|
||||
durationMs: Math.max(1, Date.now() - startedAt),
|
||||
status: summary.passed ? "pass" : "fail",
|
||||
});
|
||||
return { evidence, summary };
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const result = await runDiagnosticEventsBoundaryRuntime(parseOptions(process.argv.slice(2)));
|
||||
process.stdout.write(
|
||||
`diagnostic-events-boundary: ${result.summary.passed ? "passed" : "failed"}\n`,
|
||||
);
|
||||
if (!result.summary.passed) {
|
||||
process.exitCode = 1;
|
||||
}
|
||||
}
|
||||
|
||||
export const testing = { parseOptions, probeDiagnosticEventsBoundary };
|
||||
|
||||
if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) {
|
||||
main().catch((error: unknown) => {
|
||||
process.stderr.write(`diagnostic-events-boundary: ${formatErrorMessage(error)}\n`);
|
||||
process.exitCode = 1;
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user