trace: Correlate channel diagnostics into one trace

Correlates channel receive, agent lifecycle, model attempt diagnostics, and outbound delivery diagnostics into one trace waterfall so channel message runs can be inspected end-to-end.

Maintainer follow-up removed the internal `AgentHarnessV2` adapter surface and kept the harness path canonical through `src/agents/harness/lifecycle.ts`.

Proof:
- PR checks passed on `04e9189c15480d53663d533a04c9883164b4dd54`.
- `node scripts/run-vitest.mjs src/agents/harness/lifecycle.test.ts src/agents/harness/selection.test.ts src/channels/turn/kernel.test.ts`
- `pnpm check:changed` Testbox `tbx_01kt3xtrm70qc7nb90cqv5rah1`

Thanks @bek91.

Co-authored-by: Bek <bek.akhmedov@gmail.com>
This commit is contained in:
Bek
2026-06-02 06:38:00 -04:00
committed by GitHub
parent ad9f7f9a59
commit bce3d5bf92
15 changed files with 2135 additions and 984 deletions
@@ -1,2 +1,2 @@
63d49032a9b4dc4874a0ca17be73ecc97a2df5d1f47b4e72db34868423370558 plugin-sdk-api-baseline.json
af79f7d711afa0a8563782b8f5cdd7e46b9aea245f5e7ebc464327a8969ed65e plugin-sdk-api-baseline.jsonl
f3e0379cbe0e584a8c9658253d4a808356fe80fb5ec775bbee9e968e8d815380 plugin-sdk-api-baseline.json
601b55acafbd1e00b850c9b0c15d587029050906960071d448d37538b223e226 plugin-sdk-api-baseline.jsonl
+870 -18
View File
@@ -1,6 +1,11 @@
import { afterAll, afterEach, beforeEach, describe, expect, test, vi } from "vitest";
const telemetryState = vi.hoisted(() => {
type TestSpanContext = {
traceId: string;
spanId: string;
traceFlags: number;
};
const counters = new Map<string, { add: ReturnType<typeof vi.fn> }>();
const histograms = new Map<string, { record: ReturnType<typeof vi.fn> }>();
const spans: Array<{
@@ -9,7 +14,7 @@ const telemetryState = vi.hoisted(() => {
end: ReturnType<typeof vi.fn>;
setAttributes: ReturnType<typeof vi.fn>;
setStatus: ReturnType<typeof vi.fn>;
spanContext: ReturnType<typeof vi.fn>;
spanContext: ReturnType<typeof vi.fn<() => TestSpanContext>>;
}> = [];
const tracer = {
startSpan: vi.fn((name: string, _opts?: unknown, _ctx?: unknown) => {
@@ -20,7 +25,7 @@ const telemetryState = vi.hoisted(() => {
end: vi.fn(),
setAttributes: vi.fn(),
setStatus: vi.fn(),
spanContext: vi.fn(() => ({
spanContext: vi.fn<() => TestSpanContext>(() => ({
traceId: "4bf92f3577b34da6a3ce929d0e0e4736",
spanId,
traceFlags: 1,
@@ -156,13 +161,21 @@ vi.mock("@opentelemetry/semantic-conventions", () => ({
}));
import {
createDiagnosticTraceContext,
emitTrustedDiagnosticEvent,
emitTrustedDiagnosticEventWithPrivateData,
onInternalDiagnosticEvent,
resetDiagnosticEventsForTest,
waitForDiagnosticEventsDrained,
type DiagnosticEventPrivateData,
} from "openclaw/plugin-sdk/diagnostic-runtime";
import { onTrustedInternalDiagnosticEvent } from "openclaw/plugin-sdk/plugin-test-runtime";
import {
emitInternalDiagnosticEventForTest,
logMessageDispatchStarted,
logMessageProcessed,
onTrustedInternalDiagnosticEvent,
runWithDiagnosticTraceContext,
} from "openclaw/plugin-sdk/plugin-test-runtime";
import type { OpenClawPluginServiceContext } from "../api.js";
import { emitDiagnosticEvent } from "../api.js";
import { createDiagnosticsOtelService } from "./service.js";
@@ -175,6 +188,12 @@ const SPAN_ID = "00f067aa0ba902b7";
const CHILD_SPAN_ID = "1111111111111111";
const GRANDCHILD_SPAN_ID = "2222222222222222";
const TOOL_SPAN_ID = "3333333333333333";
const MODEL_CALL_SPAN_ID = "4444444444444444";
const MODEL_USAGE_SPAN_ID = "5555555555555555";
function numberedSpanId(index: number) {
return (index + 0x1000).toString(16).padStart(16, "0");
}
const PROTO_KEY = "__proto__";
const MAX_TEST_OTEL_CONTENT_ATTRIBUTE_CHARS = 128 * 1024;
const OTEL_TRUNCATED_SUFFIX_MAX_CHARS = 20;
@@ -249,6 +268,27 @@ function startedSpanOptions(name: string) {
return startedSpanCall(name)?.[1];
}
function startedSpanParentContexts(name: string) {
return telemetryState.tracer.startSpan.mock.calls
.filter((call) => call[0] === name)
.map(
(call) =>
(call[2] as { spanContext?: { traceId?: string; spanId?: string } } | undefined)
?.spanContext,
);
}
function startedSpanParentContextsByName(name: string) {
return telemetryState.tracer.startSpan.mock.calls
.filter((call) => call[0] === name)
.map((call) => ({
attributes: (call[1] as { attributes?: Record<string, unknown> } | undefined)?.attributes,
parentContext: (
call[2] as { spanContext?: { traceId?: string; spanId?: string } } | undefined
)?.spanContext,
}));
}
function mockCall(mock: { mock: { calls: unknown[][] } }, callIndex = 0): unknown[] {
const call = mock.mock.calls.at(callIndex);
if (!call) {
@@ -2563,7 +2603,581 @@ describe("diagnostics-otel service", () => {
await service.stop?.(ctx);
});
test("keeps trusted run spans alive long enough for post-completion usage parenting", async () => {
test("correlates one channel message waterfall across message, harness, usage, and model spans", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "message.dispatch.started",
channel: "slack",
source: "replyResolver",
sessionKey: "agent:main:slack:channel:c1",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "harness.run.started",
runId: "run-1",
harnessId: "codex",
pluginId: "codex",
provider: "openai",
model: "gpt-5.5",
channel: "slack",
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-1",
provider: "openai",
model: "gpt-5.5",
channel: "slack",
trace: {
traceId: TRACE_ID,
spanId: TOOL_SPAN_ID,
parentSpanId: GRANDCHILD_SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "model.call.started",
runId: "run-1",
callId: "call-1",
provider: "openai",
model: "gpt-5.5",
api: "openai-codex-responses",
transport: "stdio",
trace: {
traceId: TRACE_ID,
spanId: MODEL_CALL_SPAN_ID,
parentSpanId: TOOL_SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "model.call.completed",
runId: "run-1",
callId: "call-1",
provider: "openai",
model: "gpt-5.5",
api: "openai-codex-responses",
transport: "stdio",
durationMs: 80,
trace: {
traceId: TRACE_ID,
spanId: MODEL_CALL_SPAN_ID,
parentSpanId: TOOL_SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "harness.run.completed",
runId: "run-1",
harnessId: "codex",
pluginId: "codex",
provider: "openai",
model: "gpt-5.5",
channel: "slack",
durationMs: 100,
outcome: "completed",
itemLifecycle: { startedCount: 1, completedCount: 1, activeCount: 0 },
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "model.usage",
sessionKey: "agent:main:slack:channel:c1",
channel: "slack",
agentId: "main",
provider: "openai",
model: "gpt-5.5",
usage: { input: 3, output: 2, total: 5 },
durationMs: 10,
trace: {
traceId: TRACE_ID,
spanId: MODEL_USAGE_SPAN_ID,
parentSpanId: GRANDCHILD_SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "message.processed",
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 120,
outcome: "completed",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
await flushDiagnosticEvents();
const messageSpan = spanByName("openclaw.message.processed");
const harnessSpan = spanByName("openclaw.harness.run");
const runSpan = spanByName("openclaw.run");
const usageSpan = spanByName("openclaw.model.usage");
const modelCallSpan = spanByName("openclaw.model.call");
const messageSpanContext = messageSpan.spanContext();
const harnessSpanContext = harnessSpan.spanContext();
const runSpanContext = runSpan.spanContext();
const usageSpanContext = usageSpan.spanContext();
const modelCallSpanContext = modelCallSpan.spanContext();
const parentBySpanName = Object.fromEntries(
telemetryState.tracer.startSpan.mock.calls.map((call) => [
call[0],
(call[2] as { spanContext?: { traceId?: string; spanId?: string } } | undefined)
?.spanContext,
]),
);
expect(messageSpanContext.traceId).toBe(TRACE_ID);
expect(harnessSpanContext.traceId).toBe(TRACE_ID);
expect(usageSpanContext.traceId).toBe(TRACE_ID);
expect(modelCallSpanContext.traceId).toBe(TRACE_ID);
expect(parentBySpanName["openclaw.message.processed"]?.spanId).toBe(SPAN_ID);
expect(parentBySpanName["openclaw.harness.run"]?.spanId).toBe(messageSpanContext.spanId);
expect(parentBySpanName["openclaw.run"]?.spanId).toBe(harnessSpanContext.spanId);
expect(parentBySpanName["openclaw.model.usage"]?.spanId).toBe(harnessSpanContext.spanId);
expect(parentBySpanName["openclaw.model.call"]?.spanId).toBe(runSpanContext.spanId);
await service.stop?.(ctx);
});
test("uses production message lifecycle helpers as the message span anchor", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
const messageTrace = createDiagnosticTraceContext({
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
});
runWithDiagnosticTraceContext(messageTrace, () => {
logMessageDispatchStarted({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
source: "replyResolver",
});
emitTrustedDiagnosticEvent({
type: "harness.run.started",
runId: "run-1",
harnessId: "codex",
pluginId: "codex",
provider: "openai",
model: "gpt-5.5",
channel: "slack",
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "model.usage",
sessionKey: "agent:main:slack:channel:c1",
channel: "slack",
agentId: "main",
provider: "openai",
model: "gpt-5.5",
usage: { input: 3, output: 2, total: 5 },
durationMs: 10,
trace: {
traceId: TRACE_ID,
spanId: MODEL_USAGE_SPAN_ID,
parentSpanId: GRANDCHILD_SPAN_ID,
traceFlags: "01",
},
});
logMessageProcessed({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 120,
outcome: "completed",
});
});
await flushDiagnosticEvents();
const messageSpan = spanByName("openclaw.message.processed");
const harnessSpan = spanByName("openclaw.harness.run");
const messageSpanContext = messageSpan.spanContext();
const harnessSpanContext = harnessSpan.spanContext();
const parentBySpanName = Object.fromEntries(
telemetryState.tracer.startSpan.mock.calls.map((call) => [
call[0],
(call[2] as { spanContext?: { traceId?: string; spanId?: string } } | undefined)
?.spanContext,
]),
);
expect(parentBySpanName["openclaw.message.processed"]?.spanId).toBe(SPAN_ID);
expect(parentBySpanName["openclaw.harness.run"]?.spanId).toBe(messageSpanContext.spanId);
expect(parentBySpanName["openclaw.model.usage"]?.spanId).toBe(harnessSpanContext.spanId);
expect(messageSpanContext.traceId).toBe(TRACE_ID);
expect(harnessSpanContext.traceId).toBe(TRACE_ID);
await service.stop?.(ctx);
});
test("does not force a remote parent for root message lifecycle helpers", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
const messageTrace = createDiagnosticTraceContext({
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
traceFlags: "01",
});
runWithDiagnosticTraceContext(messageTrace, () => {
logMessageDispatchStarted({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
source: "replyResolver",
});
logMessageProcessed({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 120,
outcome: "completed",
});
});
await flushDiagnosticEvents();
expect(spanByName("openclaw.message.processed").spanContext().traceId).toBe(TRACE_ID);
expect(startedSpanParentContexts("openclaw.message.processed")[0]).toBeUndefined();
await service.stop?.(ctx);
});
test("parents outbound delivery spans under the active message lifecycle span", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
const messageTrace = createDiagnosticTraceContext({
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
});
runWithDiagnosticTraceContext(messageTrace, () => {
logMessageDispatchStarted({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
source: "replyResolver",
});
emitInternalDiagnosticEventForTest({
type: "message.delivery.completed",
channel: "slack",
deliveryKind: "text",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 15,
resultCount: 1,
});
emitInternalDiagnosticEventForTest({
type: "message.delivery.error",
channel: "slack",
deliveryKind: "media",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 25,
errorCategory: "network",
});
logMessageProcessed({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 120,
outcome: "completed",
});
});
await flushDiagnosticEvents();
const messageSpanContext = spanByName("openclaw.message.processed").spanContext();
const deliveryParentContexts = startedSpanParentContexts("openclaw.message.delivery");
expect(deliveryParentContexts).toHaveLength(2);
expect(deliveryParentContexts[0]?.traceId).toBe(TRACE_ID);
expect(deliveryParentContexts[0]?.spanId).toBe(messageSpanContext.spanId);
expect(deliveryParentContexts[1]?.traceId).toBe(TRACE_ID);
expect(deliveryParentContexts[1]?.spanId).toBe(messageSpanContext.spanId);
await service.stop?.(ctx);
});
test("parents multi-batch late delivery spans from the retained message context", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
const messageTrace = createDiagnosticTraceContext({
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
});
runWithDiagnosticTraceContext(messageTrace, () => {
logMessageDispatchStarted({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
source: "replyResolver",
});
for (let index = 0; index < 125; index += 1) {
emitInternalDiagnosticEventForTest({
type: "message.delivery.completed",
channel: "slack",
deliveryKind: "text",
sessionKey: `agent:main:slack:channel:c${index}`,
durationMs: 15,
resultCount: 1,
});
}
logMessageProcessed({
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 120,
outcome: "completed",
});
});
const messageSpan = spanByName("openclaw.message.processed");
const messageSpanContext = messageSpan.spanContext();
expect(messageSpan.end).toHaveBeenCalledTimes(1);
await waitForDiagnosticEventsDrained();
const deliveryParentContexts = startedSpanParentContexts("openclaw.message.delivery");
expect(deliveryParentContexts).toHaveLength(125);
expect(deliveryParentContexts.every((parent) => parent?.traceId === TRACE_ID)).toBe(true);
expect(
deliveryParentContexts.every((parent) => parent?.spanId === messageSpanContext.spanId),
).toBe(true);
await service.stop?.(ctx);
});
test("correlates skipped duplicate message lifecycle helpers to the active inbound trace", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
const messageTrace = createDiagnosticTraceContext({
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
});
runWithDiagnosticTraceContext(messageTrace, () => {
logMessageProcessed({
channel: "slack",
messageId: "msg-duplicate",
chatId: "c1",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 5,
outcome: "skipped",
reason: "duplicate",
});
});
await flushDiagnosticEvents();
const messageSpan = spanByName("openclaw.message.processed");
const messageSpanContext = messageSpan.spanContext();
const parentContext = startedSpanParentContexts("openclaw.message.processed")[0];
expect(messageSpanContext.traceId).toBe(TRACE_ID);
expect(parentContext?.traceId).toBe(TRACE_ID);
expect(parentContext?.spanId).toBe(SPAN_ID);
expect(firstSpanAttributes("openclaw.message.processed")["openclaw.reason"]).toBe("duplicate");
expect(messageSpan.end).toHaveBeenCalledTimes(1);
await service.stop?.(ctx);
});
test("does not force a remote parent for fallback root message processed spans", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "message.processed",
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 25,
outcome: "skipped",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
await flushDiagnosticEvents();
expect(spanByName("openclaw.message.processed").spanContext().traceId).toBe(TRACE_ID);
expect(startedSpanParentContexts("openclaw.message.processed")[0]).toBeUndefined();
await service.stop?.(ctx);
});
test("does not retain fallback message processed spans as active parents", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "message.processed",
channel: "slack",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 25,
outcome: "skipped",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
expect(spanByName("openclaw.message.processed").end).toHaveBeenCalledTimes(1);
telemetryState.tracer.setSpanContext.mockClear();
emitTrustedDiagnosticEvent({
type: "harness.run.started",
runId: "run-1",
harnessId: "codex",
pluginId: "codex",
provider: "openai",
model: "gpt-5.5",
channel: "slack",
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
expect(telemetryState.tracer.setSpanContext).not.toHaveBeenCalled();
expect(startedSpanCall("openclaw.harness.run")?.[2]).toBeUndefined();
await service.stop?.(ctx);
});
test("retains trusted run context long enough for exact post-completion usage parenting", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "run.completed",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
outcome: "completed",
durationMs: 100,
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
await Promise.resolve();
emitTrustedDiagnosticEvent({
type: "model.usage",
provider: "openai",
model: "gpt-5.4",
usage: { input: 3, output: 2, total: 5 },
durationMs: 10,
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
await flushDiagnosticEvents();
const runSpan = telemetryState.spans.find((span) => span.name === "openclaw.run");
const runSpanId = runSpan?.spanContext.mock.results[0]?.value?.spanId;
const modelUsageCall = telemetryState.tracer.startSpan.mock.calls.find(
(call) => call[0] === "openclaw.model.usage",
);
const linkedSpanContext = firstSetSpanContext();
expect(linkedSpanContext.traceId).toBe(TRACE_ID);
expect(linkedSpanContext.spanId).toBe(runSpanId);
expect(
(modelUsageCall?.[2] as { spanContext?: { spanId?: string } } | undefined)?.spanContext
?.spanId,
).toBe(runSpanId);
expect(firstSpanEndTime("openclaw.run")).toBeTypeOf("number");
await service.stop?.(ctx);
});
test("does not parent sibling active runs through shared upstream aliases", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-2",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
const runContexts = startedSpanParentContextsByName("openclaw.run");
expect(runContexts).toHaveLength(2);
expect(runContexts[0]?.parentContext).toBeUndefined();
expect(runContexts[1]?.parentContext).toBeUndefined();
await service.stop?.(ctx);
});
test("does not parent sibling runs through retained upstream aliases", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
@@ -2594,6 +3208,201 @@ describe("diagnostics-otel service", () => {
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-2",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
const runContexts = startedSpanParentContextsByName("openclaw.run");
expect(runContexts).toHaveLength(2);
expect(runContexts[0]?.parentContext).toBeUndefined();
expect(runContexts[1]?.parentContext).toBeUndefined();
await service.stop?.(ctx);
});
test("parents retained upstream alias events only when the owner matches", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "model.call.completed",
runId: "run-1",
callId: "call-1",
provider: "openai",
model: "gpt-5.4",
durationMs: 80,
trace: {
traceId: TRACE_ID,
spanId: MODEL_CALL_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "run.completed",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
outcome: "completed",
durationMs: 100,
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
await flushDiagnosticEvents();
const runSpanContext = spanByName("openclaw.run").spanContext();
const modelParentContext = startedSpanParentContexts("openclaw.model.call")[0];
expect(modelParentContext?.traceId).toBe(TRACE_ID);
expect(modelParentContext?.spanId).toBe(runSpanContext.spanId);
await service.stop?.(ctx);
});
test("parents multi-batch late model spans from the retained run context", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
for (let index = 0; index < 125; index += 1) {
emitTrustedDiagnosticEvent({
type: "model.call.completed",
runId: "run-1",
callId: `call-${index}`,
provider: "openai",
model: "gpt-5.4",
durationMs: 80,
trace: {
traceId: TRACE_ID,
spanId: numberedSpanId(index),
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
}
emitTrustedDiagnosticEvent({
type: "run.completed",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
outcome: "completed",
durationMs: 100,
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
const runSpan = spanByName("openclaw.run");
const runSpanContext = runSpan.spanContext();
expect(runSpan.end).toHaveBeenCalledTimes(1);
await waitForDiagnosticEventsDrained();
const modelParentContexts = startedSpanParentContexts("openclaw.model.call");
expect(modelParentContexts).toHaveLength(125);
expect(modelParentContexts.every((parent) => parent?.traceId === TRACE_ID)).toBe(true);
expect(modelParentContexts.every((parent) => parent?.spanId === runSpanContext.spanId)).toBe(
true,
);
await service.stop?.(ctx);
});
test("removes retained run contexts after queued diagnostics drain", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
for (let index = 0; index < 125; index += 1) {
emitTrustedDiagnosticEvent({
type: "model.call.completed",
runId: "run-1",
callId: `call-${index}`,
provider: "openai",
model: "gpt-5.4",
durationMs: 80,
trace: {
traceId: TRACE_ID,
spanId: numberedSpanId(index),
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
}
emitTrustedDiagnosticEvent({
type: "run.completed",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
outcome: "completed",
durationMs: 100,
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
await waitForDiagnosticEventsDrained();
await new Promise<void>((resolve) => {
setTimeout(resolve, 0);
});
await waitForDiagnosticEventsDrained();
await Promise.resolve();
telemetryState.tracer.setSpanContext.mockClear();
telemetryState.tracer.startSpan.mockClear();
emitTrustedDiagnosticEvent({
type: "model.usage",
provider: "openai",
@@ -2603,26 +3412,69 @@ describe("diagnostics-otel service", () => {
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
expect(telemetryState.tracer.setSpanContext).not.toHaveBeenCalled();
expect(startedSpanCall("openclaw.model.usage")?.[2]).toBeUndefined();
await service.stop?.(ctx);
});
test("clears retained run contexts when the service stops", async () => {
const service = createDiagnosticsOtelService();
const ctx = createOtelContext(OTEL_TEST_ENDPOINT, { traces: true, metrics: true });
await service.start(ctx);
emitTrustedDiagnosticEvent({
type: "run.started",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
emitTrustedDiagnosticEvent({
type: "run.completed",
runId: "run-1",
provider: "openai",
model: "gpt-5.4",
outcome: "completed",
durationMs: 100,
trace: {
traceId: TRACE_ID,
spanId: CHILD_SPAN_ID,
parentSpanId: SPAN_ID,
traceFlags: "01",
},
});
await flushDiagnosticEvents();
const runSpan = telemetryState.spans.find((span) => span.name === "openclaw.run");
const runSpanId = runSpan?.spanContext.mock.results[0]?.value?.spanId;
const modelUsageCall = telemetryState.tracer.startSpan.mock.calls.find(
(call) => call[0] === "openclaw.model.usage",
);
await service.stop?.(ctx);
await service.start(ctx);
telemetryState.tracer.setSpanContext.mockClear();
telemetryState.tracer.startSpan.mockClear();
const linkedSpanContext = firstSetSpanContext();
expect(linkedSpanContext.traceId).toBe(TRACE_ID);
expect(linkedSpanContext.spanId).toBe(runSpanId);
expect(
(modelUsageCall?.[2] as { spanContext?: { spanId?: string } } | undefined)?.spanContext
?.spanId,
).toBe(runSpanId);
expect(firstSpanEndTime("openclaw.run")).toBeTypeOf("number");
emitTrustedDiagnosticEvent({
type: "model.usage",
provider: "openai",
model: "gpt-5.4",
usage: { input: 3, output: 2, total: 5 },
durationMs: 10,
trace: {
traceId: TRACE_ID,
spanId: GRANDCHILD_SPAN_ID,
parentSpanId: CHILD_SPAN_ID,
traceFlags: "01",
},
});
expect(telemetryState.tracer.setSpanContext).not.toHaveBeenCalled();
expect(startedSpanCall("openclaw.model.usage")?.[2]).toBeUndefined();
await service.stop?.(ctx);
});
+302 -42
View File
@@ -6,6 +6,7 @@ import {
SpanStatusCode,
TraceFlags,
} from "@opentelemetry/api";
import type { SpanContext } from "@opentelemetry/api";
import type { LogRecord, SeverityNumber } from "@opentelemetry/api-logs";
import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-proto";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-proto";
@@ -26,6 +27,7 @@ import {
ATTR_GEN_AI_SYSTEM_INSTRUCTIONS,
ATTR_GEN_AI_TOOL_DEFINITIONS,
} from "@opentelemetry/semantic-conventions/incubating";
import { waitForDiagnosticEventsDrained } from "openclaw/plugin-sdk/diagnostic-runtime";
import { registerUnhandledRejectionHandler } from "openclaw/plugin-sdk/runtime-env";
import type {
DiagnosticEventMetadata,
@@ -86,6 +88,8 @@ const GEN_AI_TOKEN_USAGE_BUCKETS = [
const GEN_AI_OPERATION_DURATION_BUCKETS = [
0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92,
];
const MAX_RETAINED_TRUSTED_SPAN_CONTEXTS = 1024;
const RETAINED_TRUSTED_SPAN_CONTEXT_TIMEOUT_MS = 5_000;
type OtelContentCapturePolicy = {
inputMessages: boolean;
@@ -128,6 +132,7 @@ type SessionRecoveryDiagnosticEvent = Extract<
{ type: "session.recovery.requested" | "session.recovery.completed" }
>;
type TalkDiagnosticEvent = Extract<DiagnosticEventPayload, { type: "talk.event" }>;
type TrustedSpanAliasOwner = { kind: "run"; id: string };
const NO_CONTENT_CAPTURE: OtelContentCapturePolicy = {
inputMessages: false,
@@ -1240,17 +1245,25 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
const meter = metrics.getMeter("openclaw");
const tracer = trace.getTracer("openclaw");
const activeTrustedSpans = new Map<string, ReturnType<typeof tracer.startSpan>>();
const activeTrustedSpanAliases = new Map<string, ReturnType<typeof tracer.startSpan>>();
const pendingTrustedRunFinalizers = new Map<string, ReturnType<typeof setImmediate>>();
const activeTrustedSpanAliases = new Map<
string,
{ span: ReturnType<typeof tracer.startSpan>; spanId: string; owner: TrustedSpanAliasOwner }
>();
const retainedTrustedSpanContexts = new Map<
string,
{ spanContext: SpanContext; token: symbol; owner?: TrustedSpanAliasOwner }
>();
const retainedTrustedSpanContextCleanupTimers = new Set<ReturnType<typeof setTimeout>>();
stopActiveTrustedSpans = () => {
const stopAt = Date.now();
for (const handle of pendingTrustedRunFinalizers.values()) {
clearImmediate(handle);
for (const handle of retainedTrustedSpanContextCleanupTimers) {
clearTimeout(handle);
}
pendingTrustedRunFinalizers.clear();
retainedTrustedSpanContextCleanupTimers.clear();
retainedTrustedSpanContexts.clear();
for (const span of new Set([
...activeTrustedSpans.values(),
...activeTrustedSpanAliases.values(),
...Array.from(activeTrustedSpanAliases.values(), (entry) => entry.span),
])) {
span.end(stopAt);
}
@@ -1679,20 +1692,139 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
) => (metadata.trusted ? normalizeTraceContext(evt.trace) : undefined);
const internalOrTrustedTraceContext = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
) => (metadata.trusted || metadata.internal ? normalizeTraceContext(evt.trace) : undefined);
const trustedSpanAliasOwner = (
evt: DiagnosticEventPayload,
): TrustedSpanAliasOwner | undefined => {
if ("runId" in evt && evt.runId) {
return { kind: "run", id: evt.runId };
}
return undefined;
};
const sameTrustedSpanAliasOwner = (
left: TrustedSpanAliasOwner | undefined,
right: TrustedSpanAliasOwner | undefined,
) => Boolean(left && right && left.kind === right.kind && left.id === right.id);
const trustedSpanAliasKey = (spanId: string, owner: TrustedSpanAliasOwner) =>
`${spanId}:${owner.kind}:${owner.id}`;
const retainedTrustedSpanContextKey = (
traceId: string,
spanId: string,
owner?: TrustedSpanAliasOwner,
) => `${traceId}:${owner ? trustedSpanAliasKey(spanId, owner) : spanId}`;
const retainedTrustedSpanContext = (
traceContext: DiagnosticTraceContext | undefined,
spanId: string | undefined,
owner?: TrustedSpanAliasOwner,
) => {
if (!traceContext?.traceId || !spanId) {
return undefined;
}
const retained =
(owner
? retainedTrustedSpanContexts.get(
retainedTrustedSpanContextKey(traceContext.traceId, spanId, owner),
)
: undefined) ??
retainedTrustedSpanContexts.get(
retainedTrustedSpanContextKey(traceContext.traceId, spanId),
);
if (retained?.spanContext.traceId !== traceContext.traceId) {
return undefined;
}
if (retained.owner && !sameTrustedSpanAliasOwner(retained.owner, owner)) {
return undefined;
}
return retained.spanContext;
};
const activeTrustedSpanAlias = (spanId: string, owner: TrustedSpanAliasOwner | undefined) => {
if (!owner) {
return undefined;
}
const alias = activeTrustedSpanAliases.get(trustedSpanAliasKey(spanId, owner));
if (!alias || !sameTrustedSpanAliasOwner(alias.owner, owner)) {
return undefined;
}
return alias.span;
};
const internalOrTrustedParentContext = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
) => {
const traceContext = internalOrTrustedTraceContext(evt, metadata);
const parentSpanId = traceContext?.parentSpanId ?? traceContext?.spanId;
if (!traceContext || !parentSpanId) {
return undefined;
}
return contextForTraceContext({
...traceContext,
spanId: parentSpanId,
});
};
const internalOrTrustedExplicitParentContext = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
) => {
const traceContext = internalOrTrustedTraceContext(evt, metadata);
if (!traceContext?.parentSpanId) {
return undefined;
}
return contextForTraceContext({
...traceContext,
spanId: traceContext.parentSpanId,
});
};
const activeTrustedParentContext = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
) => {
const parentSpanId = trustedTraceContext(evt, metadata)?.parentSpanId;
const traceContext = trustedTraceContext(evt, metadata);
const parentSpanId = traceContext?.parentSpanId;
if (!parentSpanId) {
return undefined;
}
const owner = trustedSpanAliasOwner(evt);
const activeParentSpan =
activeTrustedSpans.get(parentSpanId) ?? activeTrustedSpanAliases.get(parentSpanId);
if (!activeParentSpan) {
activeTrustedSpans.get(parentSpanId) ?? activeTrustedSpanAlias(parentSpanId, owner);
const spanContext =
activeParentSpan?.spanContext() ??
retainedTrustedSpanContext(traceContext, parentSpanId, owner);
if (!spanContext) {
return undefined;
}
return trace.setSpanContext(otelContextApi.active(), activeParentSpan.spanContext());
return trace.setSpanContext(otelContextApi.active(), spanContext);
};
const activeInternalOrTrustedContext = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
) => {
const traceContext = internalOrTrustedTraceContext(evt, metadata);
if (!traceContext) {
return undefined;
}
const owner = trustedSpanAliasOwner(evt);
const activeSpan =
(traceContext.spanId
? (activeTrustedSpans.get(traceContext.spanId) ??
activeTrustedSpanAlias(traceContext.spanId, owner))
: undefined) ??
(traceContext.parentSpanId
? (activeTrustedSpans.get(traceContext.parentSpanId) ??
activeTrustedSpanAlias(traceContext.parentSpanId, owner))
: undefined);
if (activeSpan) {
return trace.setSpanContext(otelContextApi.active(), activeSpan.spanContext());
}
const retainedSpanContext =
retainedTrustedSpanContext(traceContext, traceContext.spanId, owner) ??
retainedTrustedSpanContext(traceContext, traceContext.parentSpanId, owner);
if (retainedSpanContext) {
return trace.setSpanContext(otelContextApi.active(), retainedSpanContext);
}
return internalOrTrustedParentContext(evt, metadata);
};
const trackTrustedSpan = (
evt: DiagnosticEventPayload,
@@ -1705,6 +1837,17 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
}
return span;
};
const trackInternalOrTrustedSpan = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
span: ReturnType<typeof tracer.startSpan>,
) => {
const spanId = internalOrTrustedTraceContext(evt, metadata)?.spanId;
if (spanId) {
activeTrustedSpans.set(spanId, span);
}
return span;
};
const takeTrackedTrustedSpan = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
@@ -1719,33 +1862,109 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
}
return span;
};
const getTrackedInternalOrTrustedSpan = (
evt: DiagnosticEventPayload,
metadata: DiagnosticEventMetadata,
) => {
const spanId = internalOrTrustedTraceContext(evt, metadata)?.spanId;
if (!spanId) {
return undefined;
}
return activeTrustedSpans.get(spanId);
};
const setSpanAttrs = (
span: ReturnType<typeof tracer.startSpan>,
attributes: Record<string, string | number | boolean>,
) => {
span.setAttributes?.(redactOtelAttributes(attributes));
};
const scheduleTrackedRunSpanFinalize = (
const retainTrustedSpanContext = (
traceId: string,
spanId: string,
spanContext: SpanContext,
token: symbol,
owner?: TrustedSpanAliasOwner,
) => {
retainedTrustedSpanContexts.set(retainedTrustedSpanContextKey(traceId, spanId, owner), {
spanContext,
token,
...(owner ? { owner } : {}),
});
while (retainedTrustedSpanContexts.size > MAX_RETAINED_TRUSTED_SPAN_CONTEXTS) {
const oldestKey = retainedTrustedSpanContexts.keys().next().value;
if (!oldestKey) {
break;
}
retainedTrustedSpanContexts.delete(oldestKey);
}
};
const scheduleRetainedTrustedSpanContextCleanup = (token: symbol) => {
let drainHandle: ReturnType<typeof setTimeout> | undefined;
let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
const cleanup = () => {
if (drainHandle) {
clearTimeout(drainHandle);
retainedTrustedSpanContextCleanupTimers.delete(drainHandle);
drainHandle = undefined;
}
if (timeoutHandle) {
clearTimeout(timeoutHandle);
retainedTrustedSpanContextCleanupTimers.delete(timeoutHandle);
timeoutHandle = undefined;
}
for (const [key, retained] of retainedTrustedSpanContexts) {
if (retained.token === token) {
retainedTrustedSpanContexts.delete(key);
}
}
};
drainHandle = setTimeout(() => {
if (drainHandle) {
retainedTrustedSpanContextCleanupTimers.delete(drainHandle);
drainHandle = undefined;
}
void waitForDiagnosticEventsDrained().then(cleanup, cleanup);
}, 0);
(drainHandle as { unref?: () => void }).unref?.();
retainedTrustedSpanContextCleanupTimers.add(drainHandle);
timeoutHandle = setTimeout(cleanup, RETAINED_TRUSTED_SPAN_CONTEXT_TIMEOUT_MS);
(timeoutHandle as { unref?: () => void }).unref?.();
retainedTrustedSpanContextCleanupTimers.add(timeoutHandle);
};
const completeTrackedLifecycleSpan = (
spanId: string,
parentSpanId: string | undefined,
span: ReturnType<typeof tracer.startSpan>,
endTimeMs: number,
) => {
const existingHandle = pendingTrustedRunFinalizers.get(spanId);
if (existingHandle) {
clearImmediate(existingHandle);
const spanContext = span.spanContext();
const retainedKeys: Array<{ spanId: string; owner?: TrustedSpanAliasOwner }> = [{ spanId }];
const retainedAliasKeys: string[] = [];
for (const [aliasKey, alias] of activeTrustedSpanAliases) {
if (alias.span === span) {
retainedKeys.push({ spanId: alias.spanId, owner: alias.owner });
retainedAliasKeys.push(aliasKey);
}
}
const handle = setImmediate(() => {
pendingTrustedRunFinalizers.delete(spanId);
if (activeTrustedSpans.get(spanId) === span) {
activeTrustedSpans.delete(spanId);
if (activeTrustedSpans.get(spanId) === span) {
activeTrustedSpans.delete(spanId);
}
for (const aliasKey of retainedAliasKeys) {
if (activeTrustedSpanAliases.get(aliasKey)?.span === span) {
activeTrustedSpanAliases.delete(aliasKey);
}
if (parentSpanId && activeTrustedSpanAliases.get(parentSpanId) === span) {
activeTrustedSpanAliases.delete(parentSpanId);
}
span.end(endTimeMs);
});
pendingTrustedRunFinalizers.set(spanId, handle);
}
span.end(endTimeMs);
const token = Symbol("retainedTrustedSpanContext");
for (const retainedKey of retainedKeys) {
retainTrustedSpanContext(
spanContext.traceId,
retainedKey.spanId,
spanContext,
token,
retainedKey.owner,
);
}
scheduleRetainedTrustedSpanContextCleanup(token);
};
const addRunAttrs = (
@@ -1962,11 +2181,28 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
const recordMessageDispatchStarted = (
evt: Extract<DiagnosticEventPayload, { type: "message.dispatch.started" }>,
metadata: DiagnosticEventMetadata,
) => {
messageDispatchStartedCounter.add(1, {
const attrs = {
"openclaw.channel": lowCardinalityAttr(evt.channel),
"openclaw.source": lowCardinalityAttr(evt.source),
});
};
messageDispatchStartedCounter.add(1, attrs);
if (!tracesEnabled) {
return;
}
const traceContext = internalOrTrustedTraceContext(evt, metadata);
if (!traceContext?.spanId || activeTrustedSpans.has(traceContext.spanId)) {
return;
}
trackInternalOrTrustedSpan(
evt,
metadata,
spanWithDuration("openclaw.message.processed", attrs, undefined, {
parentContext: internalOrTrustedExplicitParentContext(evt, metadata),
startTimeMs: evt.ts,
}),
);
};
const recordMessageDispatchCompleted = (
@@ -1984,6 +2220,7 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
const recordMessageProcessed = (
evt: Extract<DiagnosticEventPayload, { type: "message.processed" }>,
metadata: DiagnosticEventMetadata,
) => {
const attrs = {
"openclaw.channel": lowCardinalityAttr(evt.channel),
@@ -2000,11 +2237,23 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
if (evt.reason) {
spanAttrs["openclaw.reason"] = lowCardinalityAttr(evt.reason, "unknown");
}
const span = spanWithDuration("openclaw.message.processed", spanAttrs, evt.durationMs);
const trackedSpan = getTrackedInternalOrTrustedSpan(evt, metadata);
const span =
trackedSpan ??
spanWithDuration("openclaw.message.processed", spanAttrs, evt.durationMs, {
parentContext: internalOrTrustedExplicitParentContext(evt, metadata),
endTimeMs: evt.ts,
});
setSpanAttrs(span, spanAttrs);
if (evt.outcome === "error" && evt.error) {
span.setStatus({ code: SpanStatusCode.ERROR, message: redactSensitiveText(evt.error) });
}
span.end();
const traceContext = internalOrTrustedTraceContext(evt, metadata);
if (trackedSpan && traceContext?.spanId) {
completeTrackedLifecycleSpan(traceContext.spanId, trackedSpan, evt.ts);
return;
}
span.end(evt.ts);
};
const messageDeliveryAttrs = (
@@ -2022,6 +2271,7 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
const recordMessageDeliveryCompleted = (
evt: Extract<DiagnosticEventPayload, { type: "message.delivery.completed" }>,
metadata: DiagnosticEventMetadata,
) => {
const attrs = {
...messageDeliveryAttrs(evt),
@@ -2038,13 +2288,14 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
"openclaw.delivery.result_count": evt.resultCount,
},
evt.durationMs,
{ endTimeMs: evt.ts },
{ parentContext: activeInternalOrTrustedContext(evt, metadata), endTimeMs: evt.ts },
);
span.end(evt.ts);
};
const recordMessageDeliveryError = (
evt: Extract<DiagnosticEventPayload, { type: "message.delivery.error" }>,
metadata: DiagnosticEventMetadata,
) => {
const attrs = {
...messageDeliveryAttrs(evt),
@@ -2056,6 +2307,7 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
return;
}
const span = spanWithDuration("openclaw.message.delivery", attrs, evt.durationMs, {
parentContext: activeInternalOrTrustedContext(evt, metadata),
endTimeMs: evt.ts,
});
span.setStatus({
@@ -2084,7 +2336,12 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
);
const parentSpanId = trustedTraceContext(evt, metadata)?.parentSpanId;
if (parentSpanId && !activeTrustedSpans.has(parentSpanId)) {
activeTrustedSpanAliases.set(parentSpanId, span);
const owner: TrustedSpanAliasOwner = { kind: "run", id: evt.runId };
activeTrustedSpanAliases.set(trustedSpanAliasKey(parentSpanId, owner), {
span,
spanId: parentSpanId,
owner,
});
}
};
@@ -2363,12 +2620,7 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
});
}
if (trackedSpan && trustedTrace?.spanId) {
scheduleTrackedRunSpanFinalize(
trustedTrace.spanId,
trustedTrace.parentSpanId,
trackedSpan,
evt.ts,
);
completeTrackedLifecycleSpan(trustedTrace.spanId, trackedSpan, evt.ts);
return;
}
span.end(evt.ts);
@@ -2428,8 +2680,12 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
spanAttrs["openclaw.harness.items.completed"] = evt.itemLifecycle.completedCount;
spanAttrs["openclaw.harness.items.active"] = evt.itemLifecycle.activeCount;
}
const trustedTrace = trustedTraceContext(evt, metadata);
const trackedSpan = trustedTrace?.spanId
? activeTrustedSpans.get(trustedTrace.spanId)
: undefined;
const span =
takeTrackedTrustedSpan(evt, metadata) ??
trackedSpan ??
spanWithDuration("openclaw.harness.run", spanAttrs, evt.durationMs, {
parentContext: activeTrustedParentContext(evt, metadata),
endTimeMs: evt.ts,
@@ -2441,6 +2697,10 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
message: "error",
});
}
if (trackedSpan && trustedTrace?.spanId) {
completeTrackedLifecycleSpan(trustedTrace.spanId, trackedSpan, evt.ts);
return;
}
span.end(evt.ts);
};
@@ -3076,22 +3336,22 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
recordMessageReceived(evt);
return;
case "message.dispatch.started":
recordMessageDispatchStarted(evt);
recordMessageDispatchStarted(evt, metadata);
return;
case "message.dispatch.completed":
recordMessageDispatchCompleted(evt);
return;
case "message.processed":
recordMessageProcessed(evt);
recordMessageProcessed(evt, metadata);
return;
case "message.delivery.started":
recordMessageDeliveryStarted(evt);
return;
case "message.delivery.completed":
recordMessageDeliveryCompleted(evt);
recordMessageDeliveryCompleted(evt, metadata);
return;
case "message.delivery.error":
recordMessageDeliveryError(evt);
recordMessageDeliveryError(evt, metadata);
return;
case "talk.event":
recordTalkEvent(evt, metadata);
+1 -1
View File
@@ -38,7 +38,7 @@ docsRefs:
- docs/concepts/qa-e2e-automation.md
codeRefs:
- extensions/diagnostics-otel/src/service.ts
- src/agents/harness/v2.ts
- src/agents/harness/lifecycle.ts
- extensions/qa-lab/src/suite.ts
execution:
kind: flow
@@ -27,7 +27,8 @@ import { emitTrustedDiagnosticEvent } from "../../../infra/diagnostic-events.js"
import { resolveDiagnosticModelContentCapturePolicy } from "../../../infra/diagnostic-llm-content.js";
import {
createChildDiagnosticTraceContext,
createDiagnosticTraceContextFromActiveScope,
createDiagnosticTraceContext,
getActiveDiagnosticTraceContext,
freezeDiagnosticTraceContext,
} from "../../../infra/diagnostic-trace-context.js";
import { isEmbeddedMode } from "../../../infra/embedded-mode.js";
@@ -1058,7 +1059,7 @@ export async function runEmbeddedAttempt(
resolveContextEngineOwnerPluginId(activeContextEngine);
const agentDir = params.agentDir ?? resolveAgentDir(params.config ?? {}, sessionAgentId);
const diagnosticTrace = freezeDiagnosticTraceContext(
createDiagnosticTraceContextFromActiveScope(),
getActiveDiagnosticTraceContext() ?? createDiagnosticTraceContext(),
);
const runTrace = freezeDiagnosticTraceContext(
createChildDiagnosticTraceContext(diagnosticTrace),
+464
View File
@@ -0,0 +1,464 @@
import type { Model } from "openclaw/plugin-sdk/llm";
import { afterEach, describe, expect, it, vi } from "vitest";
import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../context-engine/host-compat.js";
import type { ContextEngine } from "../../context-engine/types.js";
import {
onInternalDiagnosticEvent,
resetDiagnosticEventsForTest,
type DiagnosticEventMetadata,
type DiagnosticEventPayload,
} from "../../infra/diagnostic-events.js";
import {
getActiveDiagnosticTraceContext,
resetDiagnosticTraceContextForTest,
runWithDiagnosticTraceContext,
type DiagnosticTraceContext,
} from "../../infra/diagnostic-trace-context.js";
import type { EmbeddedRunAttemptResult } from "../embedded-agent-runner/run/types.js";
import { createOpenClawAgentHarness } from "./builtin-openclaw.js";
import { runAgentHarnessLifecycleAttempt } from "./lifecycle.js";
import type { AgentHarness, AgentHarnessAttemptParams } from "./types.js";
function createAttemptParams(): AgentHarnessAttemptParams {
return {
prompt: "hello",
sessionId: "session-1",
sessionKey: "session-key",
runId: "run-1",
sessionFile: "/tmp/session.jsonl",
workspaceDir: "/tmp/workspace",
timeoutMs: 5_000,
provider: "codex",
modelId: "gpt-5.4",
model: { id: "gpt-5.4", provider: "codex" } as Model,
authStorage: {} as never,
authProfileStore: { version: 1, profiles: {} },
modelRegistry: {} as never,
thinkLevel: "low",
messageChannel: "qa",
trigger: "manual",
} as AgentHarnessAttemptParams;
}
function createDiagnosticTrace() {
return {
traceId: "11111111111111111111111111111111",
spanId: "2222222222222222",
traceFlags: "01",
};
}
function createAttemptResult(): EmbeddedRunAttemptResult {
return {
aborted: false,
externalAbort: false,
timedOut: false,
idleTimedOut: false,
timedOutDuringCompaction: false,
timedOutDuringToolExecution: false,
promptError: null,
promptErrorSource: null,
sessionIdUsed: "session-1",
diagnosticTrace: createDiagnosticTrace(),
messagesSnapshot: [],
assistantTexts: ["ok"],
toolMetas: [],
lastAssistant: undefined,
didSendViaMessagingTool: false,
messagingToolSentTexts: [],
messagingToolSentMediaUrls: [],
messagingToolSentTargets: [],
cloudCodeAssistFormatError: false,
replayMetadata: { hadPotentialSideEffects: false, replaySafe: true },
itemLifecycle: { startedCount: 0, completedCount: 0, activeCount: 0 },
};
}
function createContextEngineRequiringAssembly(): ContextEngine {
return {
info: {
id: "lossless-claw",
name: "Lossless",
hostRequirements: {
"agent-run": {
requiredCapabilities: ["assemble-before-prompt"],
},
},
},
async ingest() {
return { ingested: true };
},
async assemble({ messages }) {
return { messages, estimatedTokens: 0 };
},
async compact() {
return { ok: true, compacted: false };
},
};
}
async function flushDiagnosticEvents(): Promise<void> {
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
}
function captureDiagnosticEvents(
filter: (event: DiagnosticEventPayload) => boolean = (event) =>
event.type.startsWith("harness.run."),
): {
events: Array<{ event: DiagnosticEventPayload; metadata: DiagnosticEventMetadata }>;
unsubscribe: () => void;
} {
const events: Array<{ event: DiagnosticEventPayload; metadata: DiagnosticEventMetadata }> = [];
const unsubscribe = onInternalDiagnosticEvent((event, metadata) => {
if (filter(event)) {
events.push({ event, metadata });
}
});
return { events, unsubscribe };
}
describe("AgentHarness lifecycle runner", () => {
afterEach(() => {
resetDiagnosticEventsForTest();
resetDiagnosticTraceContextForTest();
});
it("runs a harness attempt without changing attempt params", async () => {
const params = createAttemptParams();
const result = createAttemptResult();
const runAttempt = vi.fn(async () => result);
const harness: AgentHarness = {
id: "codex",
label: "Codex",
pluginId: "codex-plugin",
supports: () => ({ supported: true, priority: 100 }),
runAttempt,
};
const attemptResult = await runAgentHarnessLifecycleAttempt(harness, params);
expect(attemptResult).toEqual({ ...result, agentHarnessId: "codex" });
expect(runAttempt).toHaveBeenCalledWith(params);
});
it("rejects harnesses that do not advertise required context-engine capabilities", async () => {
const params = createAttemptParams();
params.contextEngine = createContextEngineRequiringAssembly();
const runAttempt = vi.fn(async () => createAttemptResult());
const harness: AgentHarness = {
id: "custom",
label: "Custom",
supports: () => ({ supported: true }),
runAttempt,
};
await expect(runAgentHarnessLifecycleAttempt(harness, params)).rejects.toThrow(
'Context engine "lossless-claw" cannot run operation "agent-run" on agent harness "custom".',
);
expect(runAttempt).not.toHaveBeenCalled();
});
it("allows harnesses that advertise required context-engine capabilities", async () => {
const params = createAttemptParams();
params.contextEngine = createContextEngineRequiringAssembly();
const result = createAttemptResult();
const runAttempt = vi.fn(async () => result);
const harness: AgentHarness = {
id: "codex",
label: "Codex",
contextEngineHostCapabilities: ["assemble-before-prompt"],
supports: () => ({ supported: true }),
runAttempt,
};
await expect(runAgentHarnessLifecycleAttempt(harness, params)).resolves.toEqual({
...result,
agentHarnessId: "codex",
});
expect(runAttempt).toHaveBeenCalledOnce();
});
it("advertises OpenClaw embedded host capabilities", async () => {
const harness = createOpenClawAgentHarness();
expect(harness.contextEngineHostCapabilities).toEqual(
OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST.capabilities,
);
});
it("emits trusted harness lifecycle diagnostics for successful attempts", async () => {
resetDiagnosticEventsForTest();
const params = createAttemptParams();
const result = {
...createAttemptResult(),
agentHarnessResultClassification: "reasoning-only",
yieldDetected: true,
itemLifecycle: { startedCount: 3, completedCount: 2, activeCount: 1 },
} as EmbeddedRunAttemptResult;
const harness: AgentHarness = {
id: "codex",
label: "Codex",
pluginId: "codex-plugin",
supports: () => ({ supported: true }),
runAttempt: async () => result,
};
const diagnostics = captureDiagnosticEvents();
try {
await runAgentHarnessLifecycleAttempt(harness, params);
await flushDiagnosticEvents();
} finally {
diagnostics.unsubscribe();
}
expect(diagnostics.events.map(({ event }) => event.type)).toEqual([
"harness.run.started",
"harness.run.completed",
]);
expect(diagnostics.events.every(({ metadata }) => metadata.trusted)).toBe(true);
const completedEvent = diagnostics.events[1]?.event as
| (DiagnosticEventPayload & Record<string, unknown>)
| undefined;
expect(completedEvent?.type).toBe("harness.run.completed");
expect(completedEvent?.runId).toBe("run-1");
expect(completedEvent?.sessionKey).toBe("session-key");
expect(completedEvent?.sessionId).toBe("session-1");
expect(completedEvent?.provider).toBe("codex");
expect(completedEvent?.model).toBe("gpt-5.4");
expect(completedEvent?.channel).toBe("qa");
expect(completedEvent?.trigger).toBe("manual");
expect(completedEvent?.harnessId).toBe("codex");
expect(completedEvent?.pluginId).toBe("codex-plugin");
expect(completedEvent?.outcome).toBe("completed");
expect(completedEvent?.resultClassification).toBe("reasoning-only");
expect(completedEvent?.yieldDetected).toBe(true);
expect(completedEvent?.itemLifecycle).toEqual({
startedCount: 3,
completedCount: 2,
activeCount: 1,
});
expect(typeof completedEvent?.durationMs).toBe("number");
});
it("scopes plugin harness run diagnostics under a child run trace", async () => {
resetDiagnosticEventsForTest();
const params = createAttemptParams();
params.messageChannel = undefined;
params.messageProvider = "discord-voice";
const harnessTrace = createDiagnosticTrace();
const result = createAttemptResult();
result.diagnosticTrace = undefined;
let attemptResult: EmbeddedRunAttemptResult | undefined;
let runAttemptTrace: DiagnosticTraceContext | undefined;
let classifyTrace: DiagnosticTraceContext | undefined;
const classify = vi.fn<NonNullable<AgentHarness["classify"]>>(() => {
classifyTrace = getActiveDiagnosticTraceContext();
return "ok";
});
const harness: AgentHarness = {
id: "codex",
label: "Codex",
pluginId: "codex-plugin",
supports: () => ({ supported: true }),
runAttempt: async () => {
runAttemptTrace = getActiveDiagnosticTraceContext();
return result;
},
classify,
};
const diagnostics = captureDiagnosticEvents(
(event) =>
event.type === "harness.run.started" ||
event.type === "run.started" ||
event.type === "run.completed" ||
event.type === "harness.run.completed",
);
try {
attemptResult = await runWithDiagnosticTraceContext(harnessTrace, () =>
runAgentHarnessLifecycleAttempt(harness, params),
);
await flushDiagnosticEvents();
} finally {
diagnostics.unsubscribe();
}
expect(diagnostics.events.map(({ event }) => event.type)).toEqual([
"harness.run.started",
"run.started",
"run.completed",
"harness.run.completed",
]);
expect(diagnostics.events.every(({ metadata }) => metadata.trusted)).toBe(true);
const runStarted = diagnostics.events[1]?.event as
| (DiagnosticEventPayload & { trace?: DiagnosticTraceContext })
| undefined;
const runCompleted = diagnostics.events[2]?.event as
| (DiagnosticEventPayload & {
channel?: string;
trace?: DiagnosticTraceContext;
outcome?: string;
})
| undefined;
const harnessCompleted = diagnostics.events[3]?.event as
| (DiagnosticEventPayload & { channel?: string; trace?: DiagnosticTraceContext })
| undefined;
expect(runStarted?.trace?.traceId).toBe(harnessTrace.traceId);
expect(runStarted?.trace?.parentSpanId).toBe(harnessTrace.spanId);
expect(runAttemptTrace).toEqual(runStarted?.trace);
expect(classifyTrace).toEqual(runStarted?.trace);
expect(runCompleted?.trace).toEqual(runStarted?.trace);
expect(runCompleted?.outcome).toBe("completed");
expect(runCompleted?.channel).toBe("discord-voice");
expect(harnessCompleted?.trace).toEqual(harnessTrace);
expect(harnessCompleted?.channel).toBe("discord-voice");
expect(attemptResult?.diagnosticTrace).toEqual(harnessTrace);
});
it("emits plugin before-agent-run hook blocks as blocked run completions", async () => {
resetDiagnosticEventsForTest();
const params = createAttemptParams();
const harnessTrace = createDiagnosticTrace();
const result = {
...createAttemptResult(),
promptError: new Error("blocked by policy"),
promptErrorSource: "hook:before_agent_run",
} as EmbeddedRunAttemptResult;
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: async () => result,
};
const diagnostics = captureDiagnosticEvents((event) => event.type === "run.completed");
try {
await runWithDiagnosticTraceContext(harnessTrace, () =>
runAgentHarnessLifecycleAttempt(harness, params),
);
await flushDiagnosticEvents();
} finally {
diagnostics.unsubscribe();
}
const completed = diagnostics.events[0]?.event as
| (DiagnosticEventPayload & {
blockedBy?: string;
errorCategory?: string;
outcome?: string;
})
| undefined;
expect(completed?.outcome).toBe("blocked");
expect(completed?.blockedBy).toBe("before_agent_run");
expect(completed?.errorCategory).toBeUndefined();
});
it("emits trusted harness error diagnostics with the failing lifecycle phase", async () => {
resetDiagnosticEventsForTest();
const params = createAttemptParams();
const sendError = new Error("codex app-server send failed");
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: async () => {
throw sendError;
},
};
const diagnostics = captureDiagnosticEvents();
try {
await expect(runAgentHarnessLifecycleAttempt(harness, params)).rejects.toThrow(
"codex app-server send failed",
);
await flushDiagnosticEvents();
} finally {
diagnostics.unsubscribe();
}
expect(diagnostics.events.map(({ event }) => event.type)).toEqual([
"harness.run.started",
"harness.run.error",
]);
expect(diagnostics.events.every(({ metadata }) => metadata.trusted)).toBe(true);
const errorEvent = diagnostics.events[1]?.event as
| (DiagnosticEventPayload & Record<string, unknown>)
| undefined;
expect(errorEvent?.type).toBe("harness.run.error");
expect(errorEvent?.phase).toBe("send");
expect(errorEvent?.errorCategory).toBe("Error");
expect(errorEvent).not.toHaveProperty("cleanupFailed");
expect(errorEvent?.harnessId).toBe("codex");
expect(typeof errorEvent?.durationMs).toBe("number");
});
it("keeps result classification as an explicit outcome stage", async () => {
const params = createAttemptParams();
const result = createAttemptResult();
const classify = vi.fn<NonNullable<AgentHarness["classify"]>>(() => "empty");
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => result),
classify,
};
const outcome = await runAgentHarnessLifecycleAttempt(harness, params);
expect(outcome.agentHarnessId).toBe("codex");
expect(outcome.agentHarnessResultClassification).toBe("empty");
expect(harness["classify"]).toHaveBeenCalledWith(result, params);
});
it("preserves harness-supplied classification when no classify hook is registered", async () => {
const params = createAttemptParams();
const result = {
...createAttemptResult(),
agentHarnessResultClassification: "reasoning-only",
} as EmbeddedRunAttemptResult;
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => result),
};
const outcome = await runAgentHarnessLifecycleAttempt(harness, params);
expect(outcome.agentHarnessId).toBe("codex");
expect(outcome.agentHarnessResultClassification).toBe("reasoning-only");
});
it("clears stale non-ok classification when classification resolves to ok", async () => {
const params = createAttemptParams();
const result = {
...createAttemptResult(),
agentHarnessResultClassification: "empty",
} as EmbeddedRunAttemptResult;
const classify = vi.fn<NonNullable<AgentHarness["classify"]>>(() => "ok");
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => result),
classify,
};
const classified = await runAgentHarnessLifecycleAttempt(harness, params);
expect(classified.agentHarnessId).toBe("codex");
expect(classified).not.toHaveProperty("agentHarnessResultClassification");
});
it("does not dispose harnesses after individual attempts", async () => {
const dispose = vi.fn();
const harness: AgentHarness = {
id: "custom",
label: "Custom",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => createAttemptResult()),
dispose,
};
await runAgentHarnessLifecycleAttempt(harness, createAttemptParams());
expect(dispose).not.toHaveBeenCalled();
});
});
+269
View File
@@ -0,0 +1,269 @@
import {
assertContextEngineHostSupport,
type ContextEngineHostSupport,
} from "../../context-engine/host-compat.js";
import { diagnosticErrorCategory } from "../../infra/diagnostic-error-metadata.js";
import {
emitTrustedDiagnosticEvent,
type DiagnosticHarnessRunErrorEvent,
type DiagnosticHarnessRunOutcome,
} from "../../infra/diagnostic-events.js";
import {
createChildDiagnosticTraceContext,
freezeDiagnosticTraceContext,
getActiveDiagnosticTraceContext,
runWithDiagnosticTraceContext,
type DiagnosticTraceContext,
} from "../../infra/diagnostic-trace-context.js";
import { applyAgentHarnessResultClassification } from "./result-classification.js";
import type {
AgentHarness,
AgentHarnessAttemptParams,
AgentHarnessAttemptResult,
} from "./types.js";
type AgentHarnessLifecyclePhase = DiagnosticHarnessRunErrorEvent["phase"];
type AgentRunCompletedOutcome = "completed" | "aborted" | "blocked" | "error";
type AgentRunCompletion = {
outcome: AgentRunCompletedOutcome;
blockedBy?: string;
error?: unknown;
};
function buildAgentHarnessContextEngineHostSupport(
harness: AgentHarness,
): ContextEngineHostSupport {
return {
id: `agent-harness:${harness.id}`,
label: `agent harness "${harness.id}"`,
capabilities: harness.contextEngineHostCapabilities ?? [],
};
}
function assertAgentHarnessContextEngineSupport(
harness: AgentHarness,
params: AgentHarnessAttemptParams,
): void {
if (!params.contextEngine || params.contextEngine.info.id === "legacy") {
return;
}
assertContextEngineHostSupport({
contextEngine: params.contextEngine,
operation: "agent-run",
host: buildAgentHarnessContextEngineHostSupport(harness),
});
}
function agentHarnessDiagnosticBase(
harness: AgentHarness,
params: AgentHarnessAttemptParams,
trace?: DiagnosticTraceContext,
) {
const diagnosticTrace = trace ?? getActiveDiagnosticTraceContext();
const channel = diagnosticChannel(params);
return {
runId: params.runId,
sessionId: params.sessionId,
provider: params.provider,
model: params.modelId,
harnessId: harness.id,
...(harness.pluginId ? { pluginId: harness.pluginId } : {}),
...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
...(params.trigger ? { trigger: params.trigger } : {}),
...(channel ? { channel } : {}),
...(diagnosticTrace ? { trace: freezeDiagnosticTraceContext(diagnosticTrace) } : {}),
};
}
function agentHarnessRunOutcome(result: AgentHarnessAttemptResult): DiagnosticHarnessRunOutcome {
if (result.promptError) {
return "error";
}
if (result.externalAbort || result.aborted) {
return "aborted";
}
if (result.timedOut || result.idleTimedOut || result.timedOutDuringCompaction) {
return "timed_out";
}
return "completed";
}
function shouldEmitAgentRunDiagnostics(harness: AgentHarness): boolean {
return harness.id !== "openclaw";
}
function diagnosticChannel(params: AgentHarnessAttemptParams): string | undefined {
return params.messageChannel ?? params.messageProvider;
}
function agentRunDiagnosticBase(params: AgentHarnessAttemptParams, trace: DiagnosticTraceContext) {
const channel = diagnosticChannel(params);
return {
runId: params.runId,
...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
...(params.sessionId ? { sessionId: params.sessionId } : {}),
provider: params.provider,
model: params.modelId,
...(params.trigger ? { trigger: params.trigger } : {}),
...(channel ? { channel } : {}),
trace,
};
}
function agentRunCompletion(result: AgentHarnessAttemptResult): AgentRunCompletion {
if (result.promptErrorSource === "hook:before_agent_run") {
return { outcome: "blocked", blockedBy: "before_agent_run" };
}
if (result.promptError) {
return { outcome: "error", error: result.promptError };
}
if (
result.externalAbort ||
result.aborted ||
result.timedOut ||
result.idleTimedOut ||
result.timedOutDuringCompaction
) {
return { outcome: "aborted" };
}
return { outcome: "completed" };
}
function withFallbackDiagnosticTrace(
result: AgentHarnessAttemptResult,
trace: DiagnosticTraceContext | undefined,
): AgentHarnessAttemptResult {
if (result.diagnosticTrace || !trace) {
return result;
}
return {
...result,
diagnosticTrace: freezeDiagnosticTraceContext(trace),
};
}
function emitAgentHarnessRunStarted(
harness: AgentHarness,
params: AgentHarnessAttemptParams,
trace?: DiagnosticTraceContext,
): void {
emitTrustedDiagnosticEvent({
type: "harness.run.started",
...agentHarnessDiagnosticBase(harness, params, trace),
});
}
function emitAgentHarnessRunCompleted(params: {
harness: AgentHarness;
attemptParams: AgentHarnessAttemptParams;
result: AgentHarnessAttemptResult;
startedAt: number;
trace?: DiagnosticTraceContext;
}): void {
const { harness, attemptParams, result, startedAt, trace } = params;
emitTrustedDiagnosticEvent({
type: "harness.run.completed",
...agentHarnessDiagnosticBase(harness, attemptParams, trace ?? result.diagnosticTrace),
durationMs: Date.now() - startedAt,
outcome: agentHarnessRunOutcome(result),
...(result.agentHarnessResultClassification
? { resultClassification: result.agentHarnessResultClassification }
: {}),
...(typeof result.yieldDetected === "boolean" ? { yieldDetected: result.yieldDetected } : {}),
itemLifecycle: { ...result.itemLifecycle },
});
}
function emitAgentHarnessRunError(params: {
harness: AgentHarness;
attemptParams: AgentHarnessAttemptParams;
startedAt: number;
phase: AgentHarnessLifecyclePhase;
error: unknown;
trace?: DiagnosticTraceContext;
}): void {
const { harness, attemptParams, startedAt, phase, error, trace } = params;
emitTrustedDiagnosticEvent({
type: "harness.run.error",
...agentHarnessDiagnosticBase(harness, attemptParams, trace),
durationMs: Date.now() - startedAt,
phase,
errorCategory: diagnosticErrorCategory(error),
});
}
export async function runAgentHarnessLifecycleAttempt(
harness: AgentHarness,
params: AgentHarnessAttemptParams,
): Promise<AgentHarnessAttemptResult> {
let result: AgentHarnessAttemptResult;
let phase: AgentHarnessLifecyclePhase = "prepare";
const startedAt = Date.now();
const activeHarnessTrace = getActiveDiagnosticTraceContext();
let agentRunTrace: DiagnosticTraceContext | undefined;
let agentRunStartedAt = 0;
let agentRunCompleted = false;
const emitAgentRunCompleted = (completion: AgentRunCompletion): void => {
if (!agentRunTrace || agentRunCompleted) {
return;
}
agentRunCompleted = true;
emitTrustedDiagnosticEvent({
type: "run.completed",
...agentRunDiagnosticBase(params, agentRunTrace),
durationMs: Date.now() - agentRunStartedAt,
outcome: completion.outcome,
...(completion.blockedBy ? { blockedBy: completion.blockedBy } : {}),
...(completion.error && completion.outcome === "error"
? { errorCategory: diagnosticErrorCategory(completion.error) }
: {}),
});
};
emitAgentHarnessRunStarted(harness, params, activeHarnessTrace);
try {
phase = "prepare";
assertAgentHarnessContextEngineSupport(harness, params);
if (shouldEmitAgentRunDiagnostics(harness) && activeHarnessTrace) {
agentRunTrace = freezeDiagnosticTraceContext(
createChildDiagnosticTraceContext(activeHarnessTrace),
);
agentRunStartedAt = Date.now();
emitTrustedDiagnosticEvent({
type: "run.started",
...agentRunDiagnosticBase(params, agentRunTrace),
});
}
const runAndClassify = async () => {
phase = "send";
const rawResult = await harness.runAttempt(params);
phase = "resolve";
return applyAgentHarnessResultClassification(harness, rawResult, params);
};
result = agentRunTrace
? await runWithDiagnosticTraceContext(agentRunTrace, runAndClassify)
: await runAndClassify();
result = withFallbackDiagnosticTrace(result, activeHarnessTrace);
} catch (error) {
emitAgentHarnessRunError({
harness,
attemptParams: params,
startedAt,
phase,
error,
trace: activeHarnessTrace,
});
emitAgentRunCompleted({ outcome: "error", error });
throw error;
}
emitAgentRunCompleted(agentRunCompletion(result));
emitAgentHarnessRunCompleted({
harness,
attemptParams: params,
result,
startedAt,
trace: activeHarnessTrace,
});
return result;
}
+15 -4
View File
@@ -1,4 +1,11 @@
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
createChildDiagnosticTraceContext,
createDiagnosticTraceContext,
freezeDiagnosticTraceContext,
getActiveDiagnosticTraceContext,
runWithDiagnosticTraceContext,
} from "../../infra/diagnostic-trace-context.js";
import { formatErrorMessage } from "../../infra/errors.js";
import { createSubsystemLogger } from "../../logging/subsystem.js";
import { parseAgentSessionKey } from "../../routing/session-key.js";
@@ -29,13 +36,13 @@ import {
import { expandToolGroups, normalizeToolName } from "../tool-policy.js";
import { createOpenClawAgentHarness } from "./builtin-openclaw.js";
import { MissingAgentHarnessError } from "./errors.js";
import { runAgentHarnessLifecycleAttempt } from "./lifecycle.js";
import {
resolveAgentHarnessPolicy as resolveConfiguredAgentHarnessPolicy,
type AgentHarnessPolicy,
} from "./policy.js";
import { getRegisteredAgentHarness, listRegisteredAgentHarnesses } from "./registry.js";
import type { AgentHarness, AgentHarnessSupport } from "./types.js";
import { adaptAgentHarnessToV2, runAgentHarnessV2LifecycleAttempt } from "./v2.js";
const log = createSubsystemLogger("agents/harness");
export { resolveAgentHarnessPolicy } from "./policy.js";
@@ -262,6 +269,10 @@ function selectAgentHarnessDecision(params: {
export async function runAgentHarnessAttempt(
params: EmbeddedRunAttemptParams,
): Promise<EmbeddedRunAttemptResult> {
const activeTrace = getActiveDiagnosticTraceContext();
const harnessTrace = freezeDiagnosticTraceContext(
activeTrace ? createChildDiagnosticTraceContext(activeTrace) : createDiagnosticTraceContext(),
);
const selection = selectAgentHarnessDecision({
provider: params.provider,
modelId: params.modelId,
@@ -280,13 +291,13 @@ export async function runAgentHarnessAttempt(
sessionKey: params.sessionKey,
agentId: params.agentId,
});
const v2Harness = adaptAgentHarnessToV2(harness);
const runAttempt = () => runAgentHarnessLifecycleAttempt(harness, attemptParams);
if (harness.id === "openclaw") {
return await runAgentHarnessV2LifecycleAttempt(v2Harness, attemptParams);
return await runWithDiagnosticTraceContext(harnessTrace, runAttempt);
}
try {
return await runAgentHarnessV2LifecycleAttempt(v2Harness, attemptParams);
return await runWithDiagnosticTraceContext(harnessTrace, runAttempt);
} catch (error) {
log.warn(`${harness.label} failed; not falling back to embedded OpenClaw backend`, {
harnessId: harness.id,
-632
View File
@@ -1,632 +0,0 @@
import type { Model } from "openclaw/plugin-sdk/llm";
import { afterEach, describe, expect, it, vi } from "vitest";
import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../context-engine/host-compat.js";
import type { ContextEngine } from "../../context-engine/types.js";
import {
onInternalDiagnosticEvent,
resetDiagnosticEventsForTest,
type DiagnosticEventMetadata,
type DiagnosticEventPayload,
} from "../../infra/diagnostic-events.js";
import type { EmbeddedRunAttemptResult } from "../embedded-agent-runner/run/types.js";
import { createOpenClawAgentHarness } from "./builtin-openclaw.js";
import type { AgentHarness, AgentHarnessAttemptParams } from "./types.js";
import type { AgentHarnessV2 } from "./v2.js";
import { adaptAgentHarnessToV2, runAgentHarnessV2LifecycleAttempt } from "./v2.js";
function createAttemptParams(): AgentHarnessAttemptParams {
return {
prompt: "hello",
sessionId: "session-1",
sessionKey: "session-key",
runId: "run-1",
sessionFile: "/tmp/session.jsonl",
workspaceDir: "/tmp/workspace",
timeoutMs: 5_000,
provider: "codex",
modelId: "gpt-5.4",
model: { id: "gpt-5.4", provider: "codex" } as Model,
authStorage: {} as never,
authProfileStore: { version: 1, profiles: {} },
modelRegistry: {} as never,
thinkLevel: "low",
messageChannel: "qa",
trigger: "manual",
} as AgentHarnessAttemptParams;
}
function createDiagnosticTrace() {
return {
traceId: "11111111111111111111111111111111",
spanId: "2222222222222222",
traceFlags: "01",
};
}
function createAttemptResult(): EmbeddedRunAttemptResult {
return {
aborted: false,
externalAbort: false,
timedOut: false,
idleTimedOut: false,
timedOutDuringCompaction: false,
timedOutDuringToolExecution: false,
promptError: null,
promptErrorSource: null,
sessionIdUsed: "session-1",
diagnosticTrace: createDiagnosticTrace(),
messagesSnapshot: [],
assistantTexts: ["ok"],
toolMetas: [],
lastAssistant: undefined,
didSendViaMessagingTool: false,
messagingToolSentTexts: [],
messagingToolSentMediaUrls: [],
messagingToolSentTargets: [],
cloudCodeAssistFormatError: false,
replayMetadata: { hadPotentialSideEffects: false, replaySafe: true },
itemLifecycle: { startedCount: 0, completedCount: 0, activeCount: 0 },
};
}
function createContextEngineRequiringAssembly(): ContextEngine {
return {
info: {
id: "lossless-claw",
name: "Lossless",
hostRequirements: {
"agent-run": {
requiredCapabilities: ["assemble-before-prompt"],
},
},
},
async ingest() {
return { ingested: true };
},
async assemble({ messages }) {
return { messages, estimatedTokens: 0 };
},
async compact() {
return { ok: true, compacted: false };
},
};
}
async function flushDiagnosticEvents(): Promise<void> {
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
}
function captureDiagnosticEvents(): {
events: Array<{ event: DiagnosticEventPayload; metadata: DiagnosticEventMetadata }>;
unsubscribe: () => void;
} {
const events: Array<{ event: DiagnosticEventPayload; metadata: DiagnosticEventMetadata }> = [];
const unsubscribe = onInternalDiagnosticEvent((event, metadata) => {
if (event.type.startsWith("harness.run.")) {
events.push({ event, metadata });
}
});
return { events, unsubscribe };
}
function mockCallArg(mock: { mock: { calls: unknown[][] } }, index = 0): unknown {
const call = mock.mock.calls[index];
if (!call) {
throw new Error(`Expected mock call at index ${index}`);
}
return call[0];
}
describe("AgentHarness V2 compatibility adapter", () => {
afterEach(() => {
resetDiagnosticEventsForTest();
});
it("executes prepare/start/send/outcome/cleanup as one bounded lifecycle", async () => {
const params = createAttemptParams();
const result = createAttemptResult();
const events: string[] = [];
const harness: AgentHarnessV2 = {
id: "native-v2",
label: "Native V2",
supports: () => ({ supported: true }),
prepare: async (attemptParams) => {
events.push("prepare");
expect(attemptParams).toBe(params);
return {
harnessId: "native-v2",
label: "Native V2",
params,
lifecycleState: "prepared",
};
},
start: async (prepared) => {
events.push(`start:${prepared.lifecycleState}`);
return { ...prepared, lifecycleState: "started" };
},
send: async (session) => {
events.push(`send:${session.lifecycleState}`);
return result;
},
resolveOutcome: async (session, rawResult) => {
events.push(`outcome:${session.lifecycleState}`);
return { ...rawResult, agentHarnessId: session.harnessId };
},
cleanup: async ({ prepared, session, result: cleanupResult, error }) => {
expect(prepared?.lifecycleState).toBe("prepared");
expect(session?.lifecycleState).toBe("started");
if (!session) {
throw new Error("expected started session during successful cleanup");
}
events.push(`cleanup:${session.lifecycleState}`);
expect((cleanupResult as { agentHarnessId?: string }).agentHarnessId).toBe("native-v2");
expect(error).toBeUndefined();
},
};
const attemptResult = await runAgentHarnessV2LifecycleAttempt(harness, params);
expect((attemptResult as { agentHarnessId?: string }).agentHarnessId).toBe("native-v2");
expect(attemptResult.sessionIdUsed).toBe("session-1");
expect(events).toEqual([
"prepare",
"start:prepared",
"send:started",
"outcome:started",
"cleanup:started",
]);
});
it("rejects V1-adapted harnesses that do not advertise required context-engine capabilities", async () => {
const params = createAttemptParams();
params.contextEngine = createContextEngineRequiringAssembly();
const runAttempt = vi.fn(async () => createAttemptResult());
const harness = adaptAgentHarnessToV2({
id: "custom",
label: "Custom",
supports: () => ({ supported: true }),
runAttempt,
});
await expect(runAgentHarnessV2LifecycleAttempt(harness, params)).rejects.toThrow(
'Context engine "lossless-claw" cannot run operation "agent-run" on agent harness "custom".',
);
expect(runAttempt).not.toHaveBeenCalled();
});
it("allows V1-adapted harnesses that advertise required context-engine capabilities", async () => {
const params = createAttemptParams();
params.contextEngine = createContextEngineRequiringAssembly();
const result = createAttemptResult();
const runAttempt = vi.fn(async () => result);
const harness = adaptAgentHarnessToV2({
id: "codex",
label: "Codex",
contextEngineHostCapabilities: ["assemble-before-prompt"],
supports: () => ({ supported: true }),
runAttempt,
});
await expect(runAgentHarnessV2LifecycleAttempt(harness, params)).resolves.toEqual({
...result,
agentHarnessId: "codex",
});
expect(runAttempt).toHaveBeenCalledOnce();
});
it("advertises OpenClaw embedded host capabilities through the V1 adapter", async () => {
const harness = createOpenClawAgentHarness();
expect(harness.contextEngineHostCapabilities).toEqual(
OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST.capabilities,
);
});
it("emits trusted harness lifecycle diagnostics for successful attempts", async () => {
resetDiagnosticEventsForTest();
const params = createAttemptParams();
const result = {
...createAttemptResult(),
agentHarnessResultClassification: "reasoning-only",
yieldDetected: true,
itemLifecycle: { startedCount: 3, completedCount: 2, activeCount: 1 },
} as EmbeddedRunAttemptResult;
const harness: AgentHarnessV2 = {
id: "codex",
label: "Codex",
pluginId: "codex-plugin",
supports: () => ({ supported: true }),
prepare: async () => ({
harnessId: "codex",
label: "Codex",
pluginId: "codex-plugin",
params,
lifecycleState: "prepared",
}),
start: async (prepared) => ({ ...prepared, lifecycleState: "started" }),
send: async () => result,
resolveOutcome: async (_session, rawResult) => rawResult,
cleanup: async () => {},
};
const diagnostics = captureDiagnosticEvents();
try {
await runAgentHarnessV2LifecycleAttempt(harness, params);
await flushDiagnosticEvents();
} finally {
diagnostics.unsubscribe();
}
expect(diagnostics.events.map(({ event }) => event.type)).toEqual([
"harness.run.started",
"harness.run.completed",
]);
expect(diagnostics.events.every(({ metadata }) => metadata.trusted)).toBe(true);
const completedEvent = diagnostics.events[1]?.event as
| (DiagnosticEventPayload & Record<string, unknown>)
| undefined;
expect(completedEvent?.type).toBe("harness.run.completed");
expect(completedEvent?.runId).toBe("run-1");
expect(completedEvent?.sessionKey).toBe("session-key");
expect(completedEvent?.sessionId).toBe("session-1");
expect(completedEvent?.provider).toBe("codex");
expect(completedEvent?.model).toBe("gpt-5.4");
expect(completedEvent?.channel).toBe("qa");
expect(completedEvent?.trigger).toBe("manual");
expect(completedEvent?.harnessId).toBe("codex");
expect(completedEvent?.pluginId).toBe("codex-plugin");
expect(completedEvent?.outcome).toBe("completed");
expect(completedEvent?.resultClassification).toBe("reasoning-only");
expect(completedEvent?.yieldDetected).toBe(true);
expect(completedEvent?.itemLifecycle).toEqual({
startedCount: 3,
completedCount: 2,
activeCount: 1,
});
expect(typeof completedEvent?.durationMs).toBe("number");
});
it("emits trusted harness error diagnostics with the failing lifecycle phase", async () => {
resetDiagnosticEventsForTest();
const params = createAttemptParams();
const sendError = new Error("codex app-server send failed");
const harness: AgentHarnessV2 = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
prepare: async () => ({
harnessId: "codex",
label: "Codex",
params,
lifecycleState: "prepared",
}),
start: async (prepared) => ({ ...prepared, lifecycleState: "started" }),
send: async () => {
throw sendError;
},
resolveOutcome: async (_session, rawResult) => rawResult,
cleanup: async () => {
throw new Error("cleanup failed");
},
};
const diagnostics = captureDiagnosticEvents();
try {
await expect(runAgentHarnessV2LifecycleAttempt(harness, params)).rejects.toThrow(
"codex app-server send failed",
);
await flushDiagnosticEvents();
} finally {
diagnostics.unsubscribe();
}
expect(diagnostics.events.map(({ event }) => event.type)).toEqual([
"harness.run.started",
"harness.run.error",
]);
expect(diagnostics.events.every(({ metadata }) => metadata.trusted)).toBe(true);
const errorEvent = diagnostics.events[1]?.event as
| (DiagnosticEventPayload & Record<string, unknown>)
| undefined;
expect(errorEvent?.type).toBe("harness.run.error");
expect(errorEvent?.phase).toBe("send");
expect(errorEvent?.errorCategory).toBe("Error");
expect(errorEvent?.cleanupFailed).toBe(true);
expect(errorEvent?.harnessId).toBe("codex");
expect(typeof errorEvent?.durationMs).toBe("number");
});
it("runs cleanup with the original failure and preserves that failure", async () => {
const params = createAttemptParams();
const sendError = new Error("codex app-server send failed");
const cleanup = vi.fn(async () => {
throw new Error("cleanup should not mask send failure");
});
const harness: AgentHarnessV2 = {
id: "native-v2",
label: "Native V2",
supports: () => ({ supported: true }),
prepare: async () => ({
harnessId: "native-v2",
label: "Native V2",
params,
lifecycleState: "prepared",
}),
start: async (prepared) => ({ ...prepared, lifecycleState: "started" }),
send: async () => {
throw sendError;
},
resolveOutcome: async (_session, rawResult) => rawResult,
cleanup,
};
await expect(runAgentHarnessV2LifecycleAttempt(harness, params)).rejects.toThrow(
"codex app-server send failed",
);
const cleanupInput = mockCallArg(cleanup) as {
error?: unknown;
prepared?: { lifecycleState?: string };
session?: { lifecycleState?: string };
};
expect(cleanupInput.error).toBe(sendError);
expect(cleanupInput.prepared?.lifecycleState).toBe("prepared");
expect(cleanupInput.session?.lifecycleState).toBe("started");
});
it("runs cleanup for failed prepare/start lifecycle stages", async () => {
const params = createAttemptParams();
const startError = new Error("codex app-server start failed");
const cleanup = vi.fn(async () => {});
const harness: AgentHarnessV2 = {
id: "native-v2",
label: "Native V2",
supports: () => ({ supported: true }),
prepare: async () => ({
harnessId: "native-v2",
label: "Native V2",
params,
lifecycleState: "prepared",
}),
start: async () => {
throw startError;
},
send: async () => createAttemptResult(),
resolveOutcome: async (_session, rawResult) => rawResult,
cleanup,
};
await expect(runAgentHarnessV2LifecycleAttempt(harness, params)).rejects.toThrow(
"codex app-server start failed",
);
const cleanupInput = mockCallArg(cleanup) as {
error?: unknown;
prepared?: { lifecycleState?: string };
session?: unknown;
};
expect(cleanupInput.error).toBe(startError);
expect(cleanupInput.prepared?.lifecycleState).toBe("prepared");
expect(cleanupInput.session).toBeUndefined();
});
it("passes raw send results to cleanup when outcome resolution fails", async () => {
const params = createAttemptParams();
const rawResult = createAttemptResult();
const outcomeError = new Error("outcome classification failed");
const cleanup = vi.fn(async () => {});
const harness: AgentHarnessV2 = {
id: "native-v2",
label: "Native V2",
supports: () => ({ supported: true }),
prepare: async () => ({
harnessId: "native-v2",
label: "Native V2",
params,
lifecycleState: "prepared",
}),
start: async (prepared) => ({ ...prepared, lifecycleState: "started" }),
send: async () => rawResult,
resolveOutcome: async () => {
throw outcomeError;
},
cleanup,
};
await expect(runAgentHarnessV2LifecycleAttempt(harness, params)).rejects.toThrow(
"outcome classification failed",
);
const cleanupInput = mockCallArg(cleanup) as {
error?: unknown;
result?: unknown;
prepared?: { lifecycleState?: string };
session?: { lifecycleState?: string };
};
expect(cleanupInput.error).toBe(outcomeError);
expect(cleanupInput.result).toBe(rawResult);
expect(cleanupInput.prepared?.lifecycleState).toBe("prepared");
expect(cleanupInput.session?.lifecycleState).toBe("started");
});
it("surfaces cleanup failures after successful outcomes", async () => {
const params = createAttemptParams();
const harness: AgentHarnessV2 = {
id: "native-v2",
label: "Native V2",
supports: () => ({ supported: true }),
prepare: async () => ({
harnessId: "native-v2",
label: "Native V2",
params,
lifecycleState: "prepared",
}),
start: async (prepared) => ({ ...prepared, lifecycleState: "started" }),
send: async () => createAttemptResult(),
resolveOutcome: async (_session, rawResult) => rawResult,
cleanup: async () => {
throw new Error("cleanup failed");
},
};
await expect(runAgentHarnessV2LifecycleAttempt(harness, params)).rejects.toThrow(
"cleanup failed",
);
});
it("runs a V1 harness through prepare/start/send without changing attempt params", async () => {
const params = createAttemptParams();
const result = createAttemptResult();
const runAttempt = vi.fn(async () => result);
const harness: AgentHarness = {
id: "codex",
label: "Codex",
pluginId: "codex-plugin",
supports: () => ({ supported: true, priority: 100 }),
runAttempt,
};
const v2 = adaptAgentHarnessToV2(harness);
const prepared = await v2.prepare(params);
const session = await v2.start(prepared);
expect(v2["resume"]).toBeUndefined();
expect(await v2.send(session)).toBe(result);
expect(runAttempt).toHaveBeenCalledWith(params);
expect(session.harnessId).toBe("codex");
expect(session.label).toBe("Codex");
expect(session.pluginId).toBe("codex-plugin");
expect(session.params).toBe(params);
expect(session.lifecycleState).toBe("started");
expect(prepared.lifecycleState).toBe("prepared");
});
it("keeps result classification as an explicit outcome stage", async () => {
const params = createAttemptParams();
const result = createAttemptResult();
const classify = vi.fn<NonNullable<AgentHarness["classify"]>>(() => "empty");
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => result),
classify,
};
const v2 = adaptAgentHarnessToV2(harness);
const session = await v2.start(await v2.prepare(params));
const outcome = await v2.resolveOutcome(session, result);
expect(outcome.agentHarnessId).toBe("codex");
expect(outcome.agentHarnessResultClassification).toBe("empty");
expect(harness["classify"]).toHaveBeenCalledWith(result, params);
});
it("preserves harness-supplied classification when no classify hook is registered", async () => {
const params = createAttemptParams();
const result = {
...createAttemptResult(),
agentHarnessResultClassification: "reasoning-only",
} as EmbeddedRunAttemptResult;
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => result),
};
const v2 = adaptAgentHarnessToV2(harness);
const session = await v2.start(await v2.prepare(params));
const outcome = await v2.resolveOutcome(session, result);
expect(outcome.agentHarnessId).toBe("codex");
expect(outcome.agentHarnessResultClassification).toBe("reasoning-only");
});
it("clears stale non-ok classification when classification resolves to ok", async () => {
const params = createAttemptParams();
const result = {
...createAttemptResult(),
agentHarnessResultClassification: "empty",
} as EmbeddedRunAttemptResult;
const classify = vi.fn<NonNullable<AgentHarness["classify"]>>(() => "ok");
const harness: AgentHarness = {
id: "codex",
label: "Codex",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => result),
classify,
};
const v2 = adaptAgentHarnessToV2(harness);
const session = await v2.start(await v2.prepare(params));
const classified = await v2.resolveOutcome(session, result);
expect(classified.agentHarnessId).toBe("codex");
expect(classified).not.toHaveProperty("agentHarnessResultClassification");
});
it("preserves existing compact/reset/dispose hook this binding as compatibility methods", async () => {
const harness: AgentHarness & {
compactCalls: number;
resetCalls: number;
disposeCalls: number;
} = {
id: "custom",
label: "Custom",
compactCalls: 0,
resetCalls: 0,
disposeCalls: 0,
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => createAttemptResult()),
async compact() {
this.compactCalls += 1;
return {
ok: true,
compacted: true,
result: {
summary: "done",
firstKeptEntryId: "entry-1",
tokensBefore: 100,
},
};
},
reset(params) {
expect(params).toEqual({ reason: "reset" });
this.resetCalls += 1;
},
dispose() {
this.disposeCalls += 1;
},
};
const v2 = adaptAgentHarnessToV2(harness);
await expect(
v2.compact?.({
sessionId: "session-1",
sessionFile: "/tmp/session.jsonl",
workspaceDir: "/tmp/workspace",
}),
).resolves.toHaveProperty("compacted", true);
await v2.reset?.({ reason: "reset" });
await v2.dispose?.();
expect(harness.compactCalls).toBe(1);
expect(harness.resetCalls).toBe(1);
expect(harness.disposeCalls).toBe(1);
});
it("does not dispose V1 harnesses during per-attempt cleanup", async () => {
const dispose = vi.fn();
const harness: AgentHarness = {
id: "custom",
label: "Custom",
supports: () => ({ supported: true }),
runAttempt: vi.fn(async () => createAttemptResult()),
dispose,
};
const v2 = adaptAgentHarnessToV2(harness);
const session = await v2.start(await v2.prepare(createAttemptParams()));
await v2.cleanup({ session, result: createAttemptResult() });
expect(dispose).not.toHaveBeenCalled();
});
});
-281
View File
@@ -1,281 +0,0 @@
import {
assertContextEngineHostSupport,
type ContextEngineHostSupport,
} from "../../context-engine/host-compat.js";
import { diagnosticErrorCategory } from "../../infra/diagnostic-error-metadata.js";
import {
emitTrustedDiagnosticEvent,
type DiagnosticHarnessRunErrorEvent,
type DiagnosticHarnessRunOutcome,
} from "../../infra/diagnostic-events.js";
import type { DiagnosticTraceContext } from "../../infra/diagnostic-trace-context.js";
import { formatErrorMessage } from "../../infra/errors.js";
import { createSubsystemLogger } from "../../logging/subsystem.js";
import { applyAgentHarnessResultClassification } from "./result-classification.js";
import type {
AgentHarness,
AgentHarnessAttemptParams,
AgentHarnessAttemptResult,
AgentHarnessCompactParams,
AgentHarnessCompactResult,
AgentHarnessResetParams,
AgentHarnessSupport,
AgentHarnessSupportContext,
} from "./types.js";
const log = createSubsystemLogger("agents/harness/v2");
type AgentHarnessV2LifecyclePhase = DiagnosticHarnessRunErrorEvent["phase"];
type AgentHarnessV2RunBase = {
harnessId: string;
label: string;
pluginId?: string;
params: AgentHarnessAttemptParams;
contextEngineHost?: ContextEngineHostSupport;
};
export type AgentHarnessV2PreparedRun = AgentHarnessV2RunBase & {
lifecycleState: "prepared";
};
export type AgentHarnessV2Session = AgentHarnessV2RunBase & {
lifecycleState: "started";
};
export type AgentHarnessV2ToolCall = {
id?: string;
name: string;
input?: unknown;
};
export type AgentHarnessV2CleanupParams = {
prepared?: AgentHarnessV2PreparedRun;
session?: AgentHarnessV2Session;
result?: AgentHarnessAttemptResult;
error?: unknown;
};
export type AgentHarnessV2 = {
id: string;
label: string;
pluginId?: string;
supports(ctx: AgentHarnessSupportContext): AgentHarnessSupport;
prepare(params: AgentHarnessAttemptParams): Promise<AgentHarnessV2PreparedRun>;
start(prepared: AgentHarnessV2PreparedRun): Promise<AgentHarnessV2Session>;
resume?(session: AgentHarnessV2Session): Promise<AgentHarnessV2Session>;
send(session: AgentHarnessV2Session): Promise<AgentHarnessAttemptResult>;
handleToolCall?(session: AgentHarnessV2Session, call: AgentHarnessV2ToolCall): Promise<unknown>;
resolveOutcome(
session: AgentHarnessV2Session,
result: AgentHarnessAttemptResult,
): Promise<AgentHarnessAttemptResult>;
cleanup(params: AgentHarnessV2CleanupParams): Promise<void>;
compact?(params: AgentHarnessCompactParams): Promise<AgentHarnessCompactResult | undefined>;
reset?(params: AgentHarnessResetParams): Promise<void> | void;
dispose?(): Promise<void> | void;
};
export function adaptAgentHarnessToV2(harness: AgentHarness): AgentHarnessV2 {
return {
id: harness.id,
label: harness.label,
pluginId: harness.pluginId,
supports: (ctx) => harness.supports(ctx),
prepare: async (params) => ({
harnessId: harness.id,
label: harness.label,
pluginId: harness.pluginId,
params,
contextEngineHost: buildAgentHarnessContextEngineHostSupport(harness),
lifecycleState: "prepared",
}),
start: async (prepared) => ({
harnessId: prepared.harnessId,
label: prepared.label,
pluginId: prepared.pluginId,
params: prepared.params,
contextEngineHost: prepared.contextEngineHost,
lifecycleState: "started",
}),
send: async (session) => {
if (session.params.contextEngine && session.params.contextEngine.info.id !== "legacy") {
assertContextEngineHostSupport({
contextEngine: session.params.contextEngine,
operation: "agent-run",
host: session.contextEngineHost ?? buildAgentHarnessContextEngineHostSupport(harness),
});
}
return harness.runAttempt(session.params);
},
resolveOutcome: async (session, result) =>
applyAgentHarnessResultClassification(harness, result, session.params),
cleanup: async (_params) => {
// V1 harnesses have no per-attempt cleanup hook. Global cleanup remains
// on dispose(), which must not run after every attempt.
},
compact: harness.compact ? (params) => harness.compact!(params) : undefined,
reset: harness.reset ? (params) => harness.reset!(params) : undefined,
dispose: harness.dispose ? () => harness.dispose!() : undefined,
};
}
function buildAgentHarnessContextEngineHostSupport(
harness: AgentHarness,
): ContextEngineHostSupport {
return {
id: `agent-harness:${harness.id}`,
label: `agent harness "${harness.id}"`,
capabilities: harness.contextEngineHostCapabilities ?? [],
};
}
function agentHarnessDiagnosticBase(
harness: AgentHarnessV2,
params: AgentHarnessAttemptParams,
trace?: DiagnosticTraceContext,
) {
return {
runId: params.runId,
sessionId: params.sessionId,
provider: params.provider,
model: params.modelId,
harnessId: harness.id,
...(harness.pluginId ? { pluginId: harness.pluginId } : {}),
...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
...(params.trigger ? { trigger: params.trigger } : {}),
...(params.messageChannel ? { channel: params.messageChannel } : {}),
...(trace ? { trace } : {}),
};
}
function agentHarnessRunOutcome(result: AgentHarnessAttemptResult): DiagnosticHarnessRunOutcome {
if (result.promptError) {
return "error";
}
if (result.externalAbort || result.aborted) {
return "aborted";
}
if (result.timedOut || result.idleTimedOut || result.timedOutDuringCompaction) {
return "timed_out";
}
return "completed";
}
function emitAgentHarnessRunStarted(
harness: AgentHarnessV2,
params: AgentHarnessAttemptParams,
): void {
emitTrustedDiagnosticEvent({
type: "harness.run.started",
...agentHarnessDiagnosticBase(harness, params),
});
}
function emitAgentHarnessRunCompleted(params: {
harness: AgentHarnessV2;
attemptParams: AgentHarnessAttemptParams;
result: AgentHarnessAttemptResult;
startedAt: number;
}): void {
const { harness, attemptParams, result, startedAt } = params;
emitTrustedDiagnosticEvent({
type: "harness.run.completed",
...agentHarnessDiagnosticBase(harness, attemptParams, result.diagnosticTrace),
durationMs: Date.now() - startedAt,
outcome: agentHarnessRunOutcome(result),
...(result.agentHarnessResultClassification
? { resultClassification: result.agentHarnessResultClassification }
: {}),
...(typeof result.yieldDetected === "boolean" ? { yieldDetected: result.yieldDetected } : {}),
itemLifecycle: { ...result.itemLifecycle },
});
}
function emitAgentHarnessRunError(params: {
harness: AgentHarnessV2;
attemptParams: AgentHarnessAttemptParams;
startedAt: number;
phase: AgentHarnessV2LifecyclePhase;
error: unknown;
cleanupFailed?: boolean;
}): void {
const { harness, attemptParams, startedAt, phase, error, cleanupFailed } = params;
emitTrustedDiagnosticEvent({
type: "harness.run.error",
...agentHarnessDiagnosticBase(harness, attemptParams),
durationMs: Date.now() - startedAt,
phase,
errorCategory: diagnosticErrorCategory(error),
...(cleanupFailed ? { cleanupFailed: true } : {}),
});
}
export async function runAgentHarnessV2LifecycleAttempt(
harness: AgentHarnessV2,
params: AgentHarnessAttemptParams,
): Promise<AgentHarnessAttemptResult> {
let prepared: AgentHarnessV2PreparedRun | undefined;
let session: AgentHarnessV2Session | undefined;
let rawResult: AgentHarnessAttemptResult | undefined;
let result: AgentHarnessAttemptResult;
let phase: AgentHarnessV2LifecyclePhase = "prepare";
const startedAt = Date.now();
emitAgentHarnessRunStarted(harness, params);
try {
phase = "prepare";
prepared = await harness.prepare(params);
phase = "start";
session = await harness.start(prepared);
phase = "send";
rawResult = await harness.send(session);
phase = "resolve";
result = await harness.resolveOutcome(session, rawResult);
} catch (error) {
let cleanupFailed = false;
try {
await harness.cleanup({
prepared,
session,
error,
...(rawResult === undefined ? {} : { result: rawResult }),
});
} catch (cleanupError) {
cleanupFailed = true;
// Preserve the user-visible harness failure. Cleanup errors after a
// failed lifecycle stage must not mask the actionable runtime error.
log.warn("agent harness cleanup failed after attempt failure", {
harnessId: harness.id,
provider: params.provider,
modelId: params.modelId,
error: formatErrorMessage(cleanupError),
originalError: formatErrorMessage(error),
});
}
emitAgentHarnessRunError({
harness,
attemptParams: params,
startedAt,
phase,
error,
cleanupFailed,
});
throw error;
}
try {
phase = "cleanup";
await harness.cleanup({ prepared, session, result });
} catch (error) {
emitAgentHarnessRunError({
harness,
attemptParams: params,
startedAt,
phase,
error,
});
throw error;
}
emitAgentHarnessRunCompleted({ harness, attemptParams: params, result, startedAt });
return result;
}
@@ -6,6 +6,11 @@ import {
clearApprovalNativeRouteStateForTest,
createApprovalNativeRouteReporter,
} from "../../infra/approval-native-route-coordinator.js";
import {
createDiagnosticTraceContext,
getActiveDiagnosticTraceContext,
runWithDiagnosticTraceContext,
} from "../../infra/diagnostic-trace-context.js";
import type { SessionBindingRecord } from "../../infra/outbound/session-binding-service.js";
import type {
AcpRuntime,
@@ -6184,6 +6189,53 @@ describe("dispatchReplyFromConfig", () => {
expect(skippedEvent?.reason).toBe("duplicate");
});
it("keeps duplicate skip diagnostics inside the active inbound trace", async () => {
setNoAbort();
const cfg = { diagnostics: { enabled: true } } as OpenClawConfig;
const ctx = buildTestCtx({
Provider: "whatsapp",
OriginatingChannel: "whatsapp",
OriginatingTo: "whatsapp:+15555550123",
MessageSid: "msg-dup-trace",
});
const replyResolver = vi.fn(async () => ({ text: "hi" }) as ReplyPayload);
const inboundTrace = createDiagnosticTraceContext();
const processedTraces: Array<{
outcome?: unknown;
reason?: unknown;
traceId?: string;
spanId?: string;
}> = [];
diagnosticMocks.logMessageProcessed.mockImplementation((event) => {
const activeTrace = getActiveDiagnosticTraceContext();
processedTraces.push({
outcome: event.outcome,
reason: event.reason,
traceId: activeTrace?.traceId,
spanId: activeTrace?.spanId,
});
});
try {
await runWithDiagnosticTraceContext(inboundTrace, () =>
dispatchTwiceWithFreshDispatchers({
ctx,
cfg,
replyResolver,
}),
);
} finally {
diagnosticMocks.logMessageProcessed.mockReset();
}
const skippedEvent = processedTraces.find((event) => event.outcome === "skipped");
expect(replyResolver).toHaveBeenCalledTimes(1);
expect(skippedEvent?.reason).toBe("duplicate");
expect(skippedEvent?.traceId).toBe(inboundTrace.traceId);
expect(skippedEvent?.spanId).toBe(inboundTrace.spanId);
});
it("releases inbound dedupe when dispatch fails before completion", async () => {
setNoAbort();
const cfg = { diagnostics: { enabled: true } } as OpenClawConfig;
+138 -1
View File
@@ -1,9 +1,24 @@
import { beforeEach, describe, expect, expectTypeOf, it, vi } from "vitest";
import { afterEach, beforeEach, describe, expect, expectTypeOf, it, vi } from "vitest";
import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
import type { HistoryEntry } from "../../auto-reply/reply/history.types.js";
import type { DispatchReplyWithBufferedBlockDispatcher } from "../../auto-reply/reply/provider-dispatcher.types.js";
import type { FinalizedMsgContext } from "../../auto-reply/templating.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
emitTrustedDiagnosticEvent,
onInternalDiagnosticEvent,
resetDiagnosticEventsForTest,
waitForDiagnosticEventsDrained,
type DiagnosticEventPayload,
} from "../../infra/diagnostic-events.js";
import {
createChildDiagnosticTraceContext,
freezeDiagnosticTraceContext,
getActiveDiagnosticTraceContext,
resetDiagnosticTraceContextForTest,
} from "../../infra/diagnostic-trace-context.js";
import { logMessageProcessed } from "../../logging/diagnostic.js";
import { getChildLogger, resetLogger, setLoggerOverride } from "../../logging/logger.js";
import type { RecordInboundSession } from "../session.types.js";
import type { ChannelTurnResult, DispatchedChannelTurnResult } from "./kernel.js";
import {
@@ -177,10 +192,19 @@ function loggedEvents(log: ReturnType<typeof vi.fn>): TurnLogEvent[] {
describe("channel turn kernel", () => {
beforeEach(() => {
vi.clearAllMocks();
resetDiagnosticEventsForTest();
resetDiagnosticTraceContextForTest();
resetLogger();
setLoggerOverride({ level: "info" });
clearChannelBotPairLoopGuardForTests();
resolveOutboundDurableFinalDeliverySupport.mockResolvedValue({ ok: true });
});
afterEach(() => {
setLoggerOverride(null);
resetLogger();
});
it("types optionally guarded prepared turns as drop-capable", () => {
type DispatchResult = { queuedFinal: true };
const guarded = {} as PreparedChannelTurn<DispatchResult>;
@@ -666,6 +690,119 @@ describe("channel turn kernel", () => {
]);
});
it("keeps channel message, harness, usage, and model diagnostics in one trace scope", async () => {
const diagnostics: DiagnosticEventPayload[] = [];
const unsubscribe = onInternalDiagnosticEvent((event) => {
if (
event.type === "message.processed" ||
event.type === "harness.run.started" ||
event.type === "model.usage" ||
event.type === "model.call.started" ||
event.type === "log.record"
) {
diagnostics.push(event);
}
});
const recordInboundSession = createRecordInboundSession();
const runDispatch = vi.fn(async () => {
const messageTrace = getActiveDiagnosticTraceContext();
if (!messageTrace) {
throw new Error("expected active channel message trace");
}
const harnessTrace = freezeDiagnosticTraceContext(
createChildDiagnosticTraceContext(messageTrace),
);
const runTrace = freezeDiagnosticTraceContext(
createChildDiagnosticTraceContext(harnessTrace),
);
const modelCallTrace = freezeDiagnosticTraceContext(
createChildDiagnosticTraceContext(runTrace),
);
const usageTrace = freezeDiagnosticTraceContext(
createChildDiagnosticTraceContext(harnessTrace),
);
getChildLogger({ subsystem: "diagnostic" }).info({ runId: "run-1" }, "channel lifecycle log");
emitTrustedDiagnosticEvent({
type: "harness.run.started",
runId: "run-1",
harnessId: "codex",
pluginId: "codex",
provider: "openai",
model: "gpt-5.5",
channel: "slack",
trace: harnessTrace,
});
emitTrustedDiagnosticEvent({
type: "model.call.started",
runId: "run-1",
callId: "call-1",
provider: "openai",
model: "gpt-5.5",
api: "openai-codex-responses",
transport: "stdio",
trace: modelCallTrace,
});
emitTrustedDiagnosticEvent({
type: "model.usage",
sessionKey: "agent:main:slack:channel:c1",
channel: "slack",
agentId: "main",
provider: "openai",
model: "gpt-5.5",
usage: { input: 10, output: 5, total: 15 },
durationMs: 25,
trace: usageTrace,
});
logMessageProcessed({
channel: "slack",
messageId: "msg-1",
chatId: "c1",
sessionKey: "agent:main:slack:channel:c1",
durationMs: 50,
outcome: "completed",
});
return {
queuedFinal: true,
counts: { tool: 0, block: 0, final: 1 },
};
});
try {
await runPreparedChannelTurn({
channel: "slack",
routeSessionKey: "agent:main:slack:channel:c1",
storePath: "/tmp/sessions.json",
ctxPayload: createCtx({ SessionKey: "agent:main:slack:channel:c1" }),
recordInboundSession,
runDispatch,
messageId: "msg-1",
});
await waitForDiagnosticEventsDrained();
} finally {
unsubscribe();
}
const message = diagnostics.find((event) => event.type === "message.processed");
const harness = diagnostics.find((event) => event.type === "harness.run.started");
const usage = diagnostics.find((event) => event.type === "model.usage");
const modelCall = diagnostics.find((event) => event.type === "model.call.started");
const logRecord = diagnostics.find(
(event) => event.type === "log.record" && event.message === "channel lifecycle log",
);
const traceId = message?.trace?.traceId;
expect(traceId).toBeTruthy();
expect(harness?.trace?.traceId).toBe(traceId);
expect(usage?.trace?.traceId).toBe(traceId);
expect(modelCall?.trace?.traceId).toBe(traceId);
expect(harness?.trace?.parentSpanId).toBe(message?.trace?.spanId);
expect(usage?.trace?.parentSpanId).toBe(harness?.trace?.spanId);
expect(modelCall?.trace?.parentSpanId).toBeTruthy();
expect(modelCall?.trace?.parentSpanId).not.toBe(message?.trace?.spanId);
expect(logRecord?.trace?.traceId).toBe(traceId);
});
it("drops direct prepared turns with bot-loop protection before record and dispatch", async () => {
const events: string[] = [];
const log = vi.fn();
+16
View File
@@ -3,6 +3,10 @@ import {
clearHistoryEntriesIfEnabled,
recordPendingHistoryEntryWithMedia,
} from "../../auto-reply/reply/history.js";
import {
createDiagnosticTraceContextFromActiveScope,
runWithDiagnosticTraceContext,
} from "../../infra/diagnostic-trace-context.js";
import { toHistoryMediaEntries } from "../inbound-event/media.js";
import { createChannelReplyPipeline } from "../message/reply-pipeline.js";
import type { CreateChannelReplyPipelineParams } from "../message/reply-pipeline.js";
@@ -455,6 +459,18 @@ async function runPreparedChannelTurnCore<
>(
params: PreparedChannelTurn<TDispatchResult>,
options: { suppressObserveOnlyDispatch: boolean },
): Promise<ChannelTurnResult<TDispatchResult>> {
const trace = createDiagnosticTraceContextFromActiveScope();
return await runWithDiagnosticTraceContext(trace, () =>
runPreparedChannelTurnCoreInTrace(params, options),
);
}
async function runPreparedChannelTurnCoreInTrace<
TDispatchResult = DispatchedChannelTurnResult["dispatchResult"],
>(
params: PreparedChannelTurn<TDispatchResult>,
options: { suppressObserveOnlyDispatch: boolean },
): Promise<ChannelTurnResult<TDispatchResult>> {
const admission = params.admission ?? ({ kind: "dispatch" } as const);
const botLoopDrop = resolveBotLoopProtectionDrop(params);
-1
View File
@@ -768,7 +768,6 @@ const ASYNC_DIAGNOSTIC_EVENT_TYPES = new Set<DiagnosticEventPayload["type"]>([
"model.call.completed",
"model.call.error",
"run.progress",
"harness.run.started",
"harness.run.completed",
"harness.run.error",
"context.assembled",
+3
View File
@@ -14,6 +14,9 @@ export {
resolveWebSearchProviderContractEntriesForPluginId,
} from "../plugins/contracts/registry.js";
export { loadPluginManifestRegistry } from "../plugins/manifest-registry.js";
export { emitInternalDiagnosticEvent as emitInternalDiagnosticEventForTest } from "../infra/diagnostic-events.js";
export { runWithDiagnosticTraceContext } from "../infra/diagnostic-trace-context.js";
export { logMessageDispatchStarted, logMessageProcessed } from "../logging/diagnostic.js";
export { resolveBundledExplicitProviderContractsFromPublicArtifacts } from "../plugins/provider-contract-public-artifacts.js";
export {
initializeGlobalHookRunner,