diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 6df6e55f854a..a104254a0b6e 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -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 diff --git a/extensions/diagnostics-otel/src/service.test.ts b/extensions/diagnostics-otel/src/service.test.ts index 48b3d409d74a..e427a6d3b5c5 100644 --- a/extensions/diagnostics-otel/src/service.test.ts +++ b/extensions/diagnostics-otel/src/service.test.ts @@ -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 }>(); const histograms = new Map }>(); const spans: Array<{ @@ -9,7 +14,7 @@ const telemetryState = vi.hoisted(() => { end: ReturnType; setAttributes: ReturnType; setStatus: ReturnType; - spanContext: ReturnType; + spanContext: ReturnType 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 } | 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((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); }); diff --git a/extensions/diagnostics-otel/src/service.ts b/extensions/diagnostics-otel/src/service.ts index 966f8c347b14..4fac83490acf 100644 --- a/extensions/diagnostics-otel/src/service.ts +++ b/extensions/diagnostics-otel/src/service.ts @@ -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; +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>(); - const activeTrustedSpanAliases = new Map>(); - const pendingTrustedRunFinalizers = new Map>(); + const activeTrustedSpanAliases = new Map< + string, + { span: ReturnType; spanId: string; owner: TrustedSpanAliasOwner } + >(); + const retainedTrustedSpanContexts = new Map< + string, + { spanContext: SpanContext; token: symbol; owner?: TrustedSpanAliasOwner } + >(); + const retainedTrustedSpanContextCleanupTimers = new Set>(); 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, + ) => { + 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, attributes: Record, ) => { 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 | undefined; + let timeoutHandle: ReturnType | 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, 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, + 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, + 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, + 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, + 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); diff --git a/qa/scenarios/runtime/otel-trace-smoke.md b/qa/scenarios/runtime/otel-trace-smoke.md index 7b4ce41ccd5f..022f5619dd9f 100644 --- a/qa/scenarios/runtime/otel-trace-smoke.md +++ b/qa/scenarios/runtime/otel-trace-smoke.md @@ -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 diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index 72791b83dcc0..19392699461b 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/src/agents/embedded-agent-runner/run/attempt.ts @@ -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), diff --git a/src/agents/harness/lifecycle.test.ts b/src/agents/harness/lifecycle.test.ts new file mode 100644 index 000000000000..a792aa960cde --- /dev/null +++ b/src/agents/harness/lifecycle.test.ts @@ -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 { + await new Promise((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) + | 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>(() => { + 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) + | 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>(() => "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>(() => "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(); + }); +}); diff --git a/src/agents/harness/lifecycle.ts b/src/agents/harness/lifecycle.ts new file mode 100644 index 000000000000..adce712d5798 --- /dev/null +++ b/src/agents/harness/lifecycle.ts @@ -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 { + 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; +} diff --git a/src/agents/harness/selection.ts b/src/agents/harness/selection.ts index 936e711710e8..d59f2b378a7e 100644 --- a/src/agents/harness/selection.ts +++ b/src/agents/harness/selection.ts @@ -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 { + 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, diff --git a/src/agents/harness/v2.test.ts b/src/agents/harness/v2.test.ts deleted file mode 100644 index 1d131c6b982d..000000000000 --- a/src/agents/harness/v2.test.ts +++ /dev/null @@ -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 { - await new Promise((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) - | 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) - | 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>(() => "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>(() => "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(); - }); -}); diff --git a/src/agents/harness/v2.ts b/src/agents/harness/v2.ts deleted file mode 100644 index d6f58c84ded5..000000000000 --- a/src/agents/harness/v2.ts +++ /dev/null @@ -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; - start(prepared: AgentHarnessV2PreparedRun): Promise; - resume?(session: AgentHarnessV2Session): Promise; - send(session: AgentHarnessV2Session): Promise; - handleToolCall?(session: AgentHarnessV2Session, call: AgentHarnessV2ToolCall): Promise; - resolveOutcome( - session: AgentHarnessV2Session, - result: AgentHarnessAttemptResult, - ): Promise; - cleanup(params: AgentHarnessV2CleanupParams): Promise; - compact?(params: AgentHarnessCompactParams): Promise; - reset?(params: AgentHarnessResetParams): Promise | void; - dispose?(): Promise | 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 { - 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; -} diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index a09b4c52a7ff..3c1506960963 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -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; diff --git a/src/channels/turn/kernel.test.ts b/src/channels/turn/kernel.test.ts index fb51eaab56d1..a5485bc8879b 100644 --- a/src/channels/turn/kernel.test.ts +++ b/src/channels/turn/kernel.test.ts @@ -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): 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; @@ -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(); diff --git a/src/channels/turn/kernel.ts b/src/channels/turn/kernel.ts index 4b4ff311cc32..2220823c863d 100644 --- a/src/channels/turn/kernel.ts +++ b/src/channels/turn/kernel.ts @@ -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, options: { suppressObserveOnlyDispatch: boolean }, +): Promise> { + const trace = createDiagnosticTraceContextFromActiveScope(); + return await runWithDiagnosticTraceContext(trace, () => + runPreparedChannelTurnCoreInTrace(params, options), + ); +} + +async function runPreparedChannelTurnCoreInTrace< + TDispatchResult = DispatchedChannelTurnResult["dispatchResult"], +>( + params: PreparedChannelTurn, + options: { suppressObserveOnlyDispatch: boolean }, ): Promise> { const admission = params.admission ?? ({ kind: "dispatch" } as const); const botLoopDrop = resolveBotLoopProtectionDrop(params); diff --git a/src/infra/diagnostic-events.ts b/src/infra/diagnostic-events.ts index ae546cdceee5..21529feb89ee 100644 --- a/src/infra/diagnostic-events.ts +++ b/src/infra/diagnostic-events.ts @@ -768,7 +768,6 @@ const ASYNC_DIAGNOSTIC_EVENT_TYPES = new Set([ "model.call.completed", "model.call.error", "run.progress", - "harness.run.started", "harness.run.completed", "harness.run.error", "context.assembled", diff --git a/src/plugin-sdk/plugin-test-runtime.ts b/src/plugin-sdk/plugin-test-runtime.ts index c72841634387..4b488bd7bc11 100644 --- a/src/plugin-sdk/plugin-test-runtime.ts +++ b/src/plugin-sdk/plugin-test-runtime.ts @@ -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,