From b07ebe3ff3645c4ccfd7985c07a6d2abaaea7367 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 10 Jul 2026 08:42:55 +0100 Subject: [PATCH] feat: correlate managed agent requests (#103476) --- docs/channels/clickclack.md | 15 ++++ docs/providers/clawrouter.md | 23 +++-- extensions/clawrouter/index.test.ts | 73 ++++++++++++++- extensions/clawrouter/stream.ts | 46 +++++++++- extensions/clickclack/src/gateway.test.ts | 89 ++++++++++++++++++- extensions/clickclack/src/gateway.ts | 19 +++- extensions/clickclack/src/http-client.test.ts | 38 +++++++- extensions/clickclack/src/http-client.ts | 24 +++++ extensions/clickclack/src/inbound.test.ts | 75 ++++++++++++++-- extensions/clickclack/src/inbound.ts | 55 ++++++++---- extensions/clickclack/src/outbound.test.ts | 32 +++++-- extensions/clickclack/src/outbound.ts | 8 +- packages/llm-core/src/types.ts | 5 ++ .../attempt.model-diagnostic-events.test.ts | 1 + .../run/attempt.model-diagnostic-events.ts | 9 +- test/clawrouter-managed-gateway.e2e.test.ts | 4 + 16 files changed, 463 insertions(+), 53 deletions(-) diff --git a/docs/channels/clickclack.md b/docs/channels/clickclack.md index 9d76c0c8a738..8b066ffdde63 100644 --- a/docs/channels/clickclack.md +++ b/docs/channels/clickclack.md @@ -130,6 +130,21 @@ bit: Keep the trust bit off if you only use the default `agent` reply mode; it is not needed there. +Use `agent` mode for cross-service correlation evidence. For an authoritative +ClickClack message id in its canonical `msg_` shape, the channel derives +the deterministic OpenClaw run id `clickclack:`. Each model call is +then visible in diagnostics as `clickclack::model:`; when that +turn uses ClawRouter, the same model-call id is sent as `X-Request-ID`. +`model` mode bypasses the normal agent run/session diagnostics and is therefore +not suitable for this evidence path. + +When a realtime event contains a validated `payload.correlation_id`, the +channel carries it as `X-Correlation-ID` on the authoritative message fetch and +the resulting ClickClack reply requests. Values use ClickClack's safe +128-character set (`A-Z`, `a-z`, `0-9`, `.`, `_`, `:`, and `-`); invalid values +are omitted. These joins contain identifiers only, never message bodies, +prompts, completions, credentials, or tool output. + ## Agent activity rows By default a ClickClack channel shows nothing while an agent turn runs; only the final reply lands. Set `agentActivity: true` on an account to publish durable `agent_commentary` and `agent_tool` message rows while the turn is in progress: diff --git a/docs/providers/clawrouter.md b/docs/providers/clawrouter.md index 93e18c19cefc..43f1a03fb20a 100644 --- a/docs/providers/clawrouter.md +++ b/docs/providers/clawrouter.md @@ -163,12 +163,21 @@ The existing metadata-only model transport diagnostics emit lines shaped like: ``` The plugin sends bounded `X-ClawRouter-Client`, `X-ClawRouter-Agent-Id`, and -`X-ClawRouter-Session-Id` headers when those identifiers are available. Static -deployment metadata such as `X-ClawRouter-Project-Id` can be set in the -provider `headers` map. Explicit configured headers win over automatic values. -The transport diagnostic records routing and response metadata; it does not log -credentials, request ids, prompts, or completions. ClawRouter's own audit event -provides the selected upstream provider and content-retention state. +`X-ClawRouter-Session-Id` headers when those identifiers are available. It also +maps the model call's diagnostic `callId` (`:model:`) to +`X-Request-ID`, so an OpenClaw model-call event can be joined to ClawRouter's +metadata-only audit trail. Values within the 128-character request-id budget are +identical. Longer values retain the `:model:` suffix and a deterministic +hash so distinct calls remain bounded and joinable. Static deployment metadata +such as `X-ClawRouter-Project-Id` can be set in the provider `headers` map. +Agent and session attribution headers retain their separate 256-character +limit. Automatic request ids containing characters outside ClawRouter's ASCII +identifier set use the same deterministic bounded form. +Explicit configured headers, including any case variant of `X-Request-ID`, win +over automatic values. The transport diagnostic records routing and response +metadata; it does not log credentials, request ids, prompts, or completions. +ClawRouter's own audit event provides the selected upstream provider and +content-retention state. ## Model discovery @@ -242,7 +251,7 @@ the same ClawRouter policy can change the remaining percentage. - Catalog discovery is scoped to the configured proxy key and cached per credential scope (agent dir, workspace dir, auth profile id, and base URL). - The proxy key is attached only at request dispatch; it is not stored in model metadata. -- Automatic attribution values are trimmed, control-character rejected, and bounded to 256 characters before dispatch. +- Automatic attribution and request-correlation values are trimmed and control-character rejected before dispatch. Attribution values are bounded to 256 characters; request ids are bounded to 128. - Model transport diagnostics contain metadata only and never include the proxy key or model content. - Native Anthropic and Gemini model ids are rewritten to their upstream ids only at dispatch. - Unsupported or ungranted catalog rows fail closed and are not selectable. diff --git a/extensions/clawrouter/index.test.ts b/extensions/clawrouter/index.test.ts index 248e07250a74..438e6dcc87ed 100644 --- a/extensions/clawrouter/index.test.ts +++ b/extensions/clawrouter/index.test.ts @@ -87,7 +87,7 @@ describe("ClawRouter plugin", () => { provider: "clawrouter", api: "anthropic-messages", id: "anthropic/claude-sonnet-4-6", - headers: { "X-Request-ID": "request-1" }, + headers: { "x-request-id": "request-1" }, params: { clawrouterRoute: { api: "anthropic-messages", @@ -97,11 +97,11 @@ describe("ClawRouter plugin", () => { }, } as never, {} as never, - { apiKey: "runtime-proxy-key" } as never, + { apiKey: "runtime-proxy-key", requestId: "automatic-call-id" } as never, ); expect(calls[0]?.headers).toEqual({ - "X-Request-ID": "request-1", + "x-request-id": "request-1", "X-ClawRouter-Client": "openclaw", Authorization: "Bearer runtime-proxy-key", }); @@ -122,6 +122,7 @@ describe("ClawRouter plugin", () => { streamFn: baseStreamFn, } as never); + const longRunId = `run-${"r".repeat(300)}`; void wrapped?.( { provider: "clawrouter", @@ -135,9 +136,32 @@ describe("ClawRouter plugin", () => { {} as never, { apiKey: "runtime-proxy-key", + requestId: `${longRunId}:model:1`, sessionId: `session-${"x".repeat(300)}`, } as never, ); + void wrapped?.( + { + provider: "clawrouter", + api: "openai-responses", + id: "openai/gpt-5.5", + } as never, + {} as never, + { + requestId: `${longRunId}:model:2`, + } as never, + ); + void wrapped?.( + { + provider: "clawrouter", + api: "openai-responses", + id: "openai/gpt-5.5", + } as never, + {} as never, + { + requestId: "turn-😀:model:3", + } as never, + ); expect(calls[0]?.headers).toMatchObject({ "x-clawrouter-client": "managed-openclaw", @@ -146,6 +170,43 @@ describe("ClawRouter plugin", () => { Authorization: "Bearer runtime-proxy-key", }); expect(calls[0]?.headers?.["X-ClawRouter-Session-Id"]).toHaveLength(256); + expect(calls[0]?.headers?.["X-Request-ID"]).toHaveLength(128); + expect(calls[0]?.headers?.["X-Request-ID"]).toMatch(/~[a-f0-9]{16}:model:1$/u); + expect(calls[1]?.headers?.["X-Request-ID"]).toMatch(/~[a-f0-9]{16}:model:2$/u); + expect(calls[1]?.headers?.["X-Request-ID"]).not.toBe(calls[0]?.headers?.["X-Request-ID"]); + expect(calls[2]?.headers?.["X-Request-ID"]).toMatch(/^turn-_~[a-f0-9]{16}:model:3$/u); + }); + + it("keeps an explicit per-request header ahead of the automatic model-call id", () => { + const calls: Array<{ + model: Parameters[0]; + options: Parameters[2]; + }> = []; + const baseStreamFn: StreamFn = (model, _context, options) => { + calls.push({ model, options }); + return {} as ReturnType; + }; + const wrapped = wrapClawRouterProviderStream({ + provider: "clawrouter", + modelId: "openai/gpt-5.5", + streamFn: baseStreamFn, + } as never); + + void wrapped?.( + { + provider: "clawrouter", + api: "openai-responses", + id: "openai/gpt-5.5", + } as never, + {} as never, + { + headers: { "x-request-id": "operator-request" }, + requestId: "automatic-call-id", + } as never, + ); + + expect(calls[0]?.model.headers).not.toHaveProperty("X-Request-ID"); + expect(calls[0]?.options?.headers).toEqual({ "x-request-id": "operator-request" }); }); it("omits unsafe attribution header values", () => { @@ -168,7 +229,11 @@ describe("ClawRouter plugin", () => { id: "openai/gpt-5.5", } as never, {} as never, - { apiKey: "runtime-proxy-key", sessionId: "bad\rsession" } as never, + { + apiKey: "runtime-proxy-key", + requestId: "bad\nrequest", + sessionId: "bad\rsession", + } as never, ); expect(calls[0]?.headers).toEqual({ diff --git a/extensions/clawrouter/stream.ts b/extensions/clawrouter/stream.ts index 3898097df1dd..f9689eca6763 100644 --- a/extensions/clawrouter/stream.ts +++ b/extensions/clawrouter/stream.ts @@ -1,12 +1,18 @@ +import { createHash } from "node:crypto"; import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; import type { ProviderWrapStreamFnContext } from "openclaw/plugin-sdk/plugin-entry"; import { prepareClawRouterRequestModel } from "./provider-catalog.js"; const ENV_API_KEY_MARKER = "CLAWROUTER_API_KEY"; const ATTRIBUTION_VALUE_MAX_LENGTH = 256; +const REQUEST_ID_MAX_LENGTH = 128; const CLIENT_HEADER = "X-ClawRouter-Client"; const AGENT_HEADER = "X-ClawRouter-Agent-Id"; const SESSION_HEADER = "X-ClawRouter-Session-Id"; +const REQUEST_ID_HEADER = "X-Request-ID"; +const REQUEST_ID_HASH_LENGTH = 16; +const REQUEST_ID_PATTERN = /^[A-Za-z0-9._~:/+@=-]+$/u; +const REQUEST_ID_UNSAFE_CHARACTER_PATTERN = /[^A-Za-z0-9._~:/+@=-]/gu; function hasControlCharacter(value: string): boolean { for (let index = 0; index < value.length; index += 1) { @@ -18,14 +24,46 @@ function hasControlCharacter(value: string): boolean { return false; } -function sanitizeAttributionValue(value: string | undefined): string | undefined { +function normalizeAttributionValue(value: string | undefined): string | undefined { const normalized = value?.trim(); if (!normalized || hasControlCharacter(normalized)) { return undefined; } + return normalized; +} + +function sanitizeAttributionValue(value: string | undefined): string | undefined { + const normalized = normalizeAttributionValue(value); + if (!normalized) { + return undefined; + } return normalized.slice(0, ATTRIBUTION_VALUE_MAX_LENGTH); } +function sanitizeRequestId(value: string | undefined): string | undefined { + const normalized = normalizeAttributionValue(value); + if (!normalized) { + return normalized; + } + if (normalized.length <= REQUEST_ID_MAX_LENGTH && REQUEST_ID_PATTERN.test(normalized)) { + return normalized; + } + // Retain the per-call suffix while bounding long run ids; the hash keeps + // distinct long prefixes from collapsing onto the same audit identifier. + const hash = createHash("sha256") + .update(normalized) + .digest("hex") + .slice(0, REQUEST_ID_HASH_LENGTH); + const modelSuffix = normalized.match(/:model:\d+$/u)?.[0] ?? ""; + const rawPrefix = modelSuffix ? normalized.slice(0, -modelSuffix.length) : normalized; + const safePrefix = rawPrefix.replace(REQUEST_ID_UNSAFE_CHARACTER_PATTERN, "_"); + const boundedSuffix = `~${hash}${modelSuffix}`; + if (boundedSuffix.length >= REQUEST_ID_MAX_LENGTH) { + return `${safePrefix.slice(0, REQUEST_ID_MAX_LENGTH - hash.length - 1)}~${hash}`; + } + return `${safePrefix.slice(0, REQUEST_ID_MAX_LENGTH - boundedSuffix.length)}${boundedSuffix}`; +} + function findHeader(headers: Record, target: string): string | undefined { const normalizedTarget = target.toLowerCase(); for (const [name, value] of Object.entries(headers)) { @@ -48,7 +86,7 @@ function setHeaderDefault( function withClawRouterHeaders( headers: Record | undefined, - params: { agentId?: string; apiKey?: string; sessionId?: string }, + params: { agentId?: string; apiKey?: string; requestId?: string; sessionId?: string }, ): Record { const next: Record = {}; for (const [name, value] of Object.entries(headers ?? {})) { @@ -59,6 +97,7 @@ function withClawRouterHeaders( setHeaderDefault(next, CLIENT_HEADER, "openclaw"); setHeaderDefault(next, AGENT_HEADER, sanitizeAttributionValue(params.agentId)); setHeaderDefault(next, SESSION_HEADER, sanitizeAttributionValue(params.sessionId)); + setHeaderDefault(next, REQUEST_ID_HEADER, sanitizeRequestId(params.requestId)); if (params.apiKey) { next.Authorization = `Bearer ${params.apiKey}`; } @@ -73,12 +112,15 @@ function createClawRouterStreamWrapper(ctx: ProviderWrapStreamFnContext): Stream return (model, context, options) => { const apiKey = options?.apiKey?.trim(); const preparedModel = prepareClawRouterRequestModel(model); + const hasExplicitRequestId = + findHeader(options?.headers ?? {}, REQUEST_ID_HEADER) !== undefined; return underlying( { ...preparedModel, headers: withClawRouterHeaders(preparedModel.headers, { agentId: ctx.agentId, apiKey: apiKey && apiKey !== ENV_API_KEY_MARKER ? apiKey : undefined, + requestId: hasExplicitRequestId ? undefined : options?.requestId, sessionId: options?.sessionId, }), }, diff --git a/extensions/clickclack/src/gateway.test.ts b/extensions/clickclack/src/gateway.test.ts index 1be7f9b62c55..b2fbb886c847 100644 --- a/extensions/clickclack/src/gateway.test.ts +++ b/extensions/clickclack/src/gateway.test.ts @@ -16,6 +16,7 @@ class FakeSocket extends EventEmitter { } const mocks = vi.hoisted(() => ({ + createClickClackClient: vi.fn(), client: { me: vi.fn(), events: vi.fn(), @@ -34,7 +35,9 @@ vi.mock("./access.js", () => ({ })); vi.mock("./http-client.js", () => ({ - createClickClackClient: vi.fn(() => mocks.client), + createClickClackClient: mocks.createClickClackClient, + normalizeClickClackCorrelationId: (value: unknown) => + typeof value === "string" && /^[A-Za-z0-9._:-]{1,128}$/u.test(value) ? value : undefined, })); vi.mock("./inbound.js", () => ({ @@ -79,6 +82,7 @@ function createGatewayContext( describe("ClickClack gateway", () => { beforeEach(() => { vi.clearAllMocks(); + mocks.createClickClackClient.mockReturnValue(mocks.client); mocks.client.me.mockResolvedValue({ id: "bot-user", display_name: "Bot", @@ -156,6 +160,8 @@ describe("ClickClack gateway", () => { shouldDispatch: true, commandAuthorized: true, }); + expect(mocks.createClickClackClient).toHaveBeenCalledTimes(1); + expect(mocks.handleClickClackInbound.mock.calls[0]?.[0]).not.toHaveProperty("correlationId"); abort.abort(); await run; expect(runError).toBeUndefined(); @@ -196,6 +202,87 @@ describe("ClickClack gateway", () => { await run; }); + it("carries validated event correlation through the authoritative fetch and inbound turn", async () => { + const socket = new FakeSocket(); + mocks.client.websocket.mockReturnValue(socket); + const abort = new AbortController(); + const ctx = createGatewayContext(abort.signal); + const run = startClickClackGatewayAccount(ctx); + + await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(1)); + + socket.emit( + "message", + Buffer.from( + JSON.stringify({ + id: "evt-1", + cursor: "cursor-1", + type: "message.created", + workspace_id: "workspace-1", + channel_id: "chan-1", + seq: 2, + created_at: "2026-01-01T00:00:00.000Z", + payload: { + message_id: "msg-1", + author_id: "human-1", + correlation_id: "fakeco.case_1", + }, + }), + ), + ); + + await vi.waitFor(() => expect(mocks.handleClickClackInbound).toHaveBeenCalledTimes(1)); + expect(mocks.createClickClackClient).toHaveBeenLastCalledWith({ + baseUrl: "https://clickclack.example", + token: "test-token", + correlationId: "fakeco.case_1", + }); + expect(mocks.client.channelMessages).toHaveBeenCalledWith("chan-1", 1, 10); + expect(mocks.handleClickClackInbound).toHaveBeenCalledWith( + expect.objectContaining({ correlationId: "fakeco.case_1" }), + ); + + abort.abort(); + await run; + }); + + it("omits invalid payload correlation without dropping the event", async () => { + const socket = new FakeSocket(); + mocks.client.websocket.mockReturnValue(socket); + const abort = new AbortController(); + const ctx = createGatewayContext(abort.signal); + const run = startClickClackGatewayAccount(ctx); + + await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(1)); + + socket.emit( + "message", + Buffer.from( + JSON.stringify({ + id: "evt-1", + cursor: "cursor-1", + type: "message.created", + workspace_id: "workspace-1", + channel_id: "chan-1", + seq: 2, + created_at: "2026-01-01T00:00:00.000Z", + payload: { + message_id: "msg-1", + author_id: "human-1", + correlation_id: "bad correlation", + }, + }), + ), + ); + + await vi.waitFor(() => expect(mocks.handleClickClackInbound).toHaveBeenCalledTimes(1)); + expect(mocks.createClickClackClient).toHaveBeenCalledTimes(1); + expect(mocks.handleClickClackInbound.mock.calls[0]?.[0]).not.toHaveProperty("correlationId"); + + abort.abort(); + await run; + }); + it("reconnects after ClickClack websocket errors", async () => { const firstSocket = new FakeSocket(); firstSocket.emitErrorOnClose = true; diff --git a/extensions/clickclack/src/gateway.ts b/extensions/clickclack/src/gateway.ts index f13db5873262..766e737a1099 100644 --- a/extensions/clickclack/src/gateway.ts +++ b/extensions/clickclack/src/gateway.ts @@ -7,7 +7,7 @@ import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import type { RawData } from "ws"; import { resolveClickClackInboundAccess } from "./access.js"; import { resolveClickClackAccount } from "./accounts.js"; -import { createClickClackClient } from "./http-client.js"; +import { createClickClackClient, normalizeClickClackCorrelationId } from "./http-client.js"; import { handleClickClackInbound } from "./inbound.js"; import { resolveWorkspaceId } from "./resolve.js"; import type { @@ -22,6 +22,10 @@ function payloadString(event: ClickClackEvent, key: string): string { return typeof value === "string" ? value : ""; } +function eventCorrelationId(event: ClickClackEvent): string | undefined { + return normalizeClickClackCorrelationId(event.payload?.correlation_id); +} + async function resolveEventMessage(params: { client: ReturnType; event: ClickClackEvent; @@ -94,7 +98,17 @@ async function processEvent(params: { if (payloadString(params.event, "author_id") === params.botUserId) { return; } - const message = await resolveEventMessage({ client: params.client, event: params.event }); + const correlationId = eventCorrelationId(params.event); + // The event body is only a routing hint. Re-fetch the authoritative message + // under the same safe correlation id before dispatching any model work. + const messageClient = correlationId + ? createClickClackClient({ + baseUrl: params.account.baseUrl, + token: params.account.token, + correlationId, + }) + : params.client; + const message = await resolveEventMessage({ client: messageClient, event: params.event }); if (!message || message.author_id === params.botUserId) { return; } @@ -114,6 +128,7 @@ async function processEvent(params: { config: params.config, message, access, + ...(correlationId ? { correlationId } : {}), }); } diff --git a/extensions/clickclack/src/http-client.test.ts b/extensions/clickclack/src/http-client.test.ts index 971ddbae1f7d..7aeecd123902 100644 --- a/extensions/clickclack/src/http-client.test.ts +++ b/extensions/clickclack/src/http-client.test.ts @@ -1,7 +1,7 @@ import { createServer, type Server } from "node:http"; import { describe, expect, it, vi } from "vitest"; import { WebSocketServer } from "ws"; -import { createClickClackClient } from "./http-client.js"; +import { createClickClackClient, normalizeClickClackCorrelationId } from "./http-client.js"; const LOOPBACK_RESPONSE_BYTES = 18 * 1024 * 1024; const CLICKCLACK_REQUEST_BODY_LIMIT_BYTES = 1024 * 1024; @@ -125,6 +125,42 @@ function streamedErrorResponse(body: string, limit: number) { } describe("ClickClack HTTP client", () => { + it("sends only safe bounded request correlation", async () => { + const fetchMock = vi.fn(async (_input: string | URL | Request, _init?: RequestInit) => + Response.json({ user: { id: "usr_1" } }), + ); + const client = createClickClackClient({ + baseUrl: "https://clickclack.example", + token: "test-token", + correlationId: " fakeco.case_1 ", + fetch: fetchMock as unknown as typeof fetch, + }); + + await client.me(); + + const headers = new Headers(fetchMock.mock.calls[0]?.[1]?.headers); + expect(headers.get("X-Correlation-ID")).toBe("fakeco.case_1"); + expect(normalizeClickClackCorrelationId("bad\ncorrelation")).toBeUndefined(); + expect(normalizeClickClackCorrelationId("x".repeat(129))).toBeUndefined(); + }); + + it("omits invalid request correlation instead of constructing an unsafe header", async () => { + const fetchMock = vi.fn(async (_input: string | URL | Request, _init?: RequestInit) => + Response.json({ user: { id: "usr_1" } }), + ); + const client = createClickClackClient({ + baseUrl: "https://clickclack.example", + token: "test-token", + correlationId: "bad\rcorrelation", + fetch: fetchMock as unknown as typeof fetch, + }); + + await client.me(); + + const headers = new Headers(fetchMock.mock.calls[0]?.[1]?.headers); + expect(headers.has("X-Correlation-ID")).toBe(false); + }); + it("bounds oversized success JSON responses and closes the stream early", async () => { const { server, closed } = createOversizedJsonServer(); const port = await listenLoopbackServer(server); diff --git a/extensions/clickclack/src/http-client.ts b/extensions/clickclack/src/http-client.ts index 968c945183ce..e77a2cbf6884 100644 --- a/extensions/clickclack/src/http-client.ts +++ b/extensions/clickclack/src/http-client.ts @@ -38,21 +38,42 @@ function provenanceFields(provenance?: ClickClackMessageProvenance): Record CLICKCLACK_CORRELATION_ID_MAX_LENGTH || + !CLICKCLACK_CORRELATION_ID_PATTERN.test(normalized) + ) { + return undefined; + } + return normalized; +} + /** * Creates a typed client for the ClickClack API using bearer-token auth. */ export function createClickClackClient(options: ClientOptions) { const baseUrl = options.baseUrl.replace(/\/$/, ""); const fetcher = options.fetch ?? fetch; + const correlationId = normalizeClickClackCorrelationId(options.correlationId); const headers = { Authorization: `Bearer ${options.token}`, Accept: "application/json", @@ -63,6 +84,9 @@ export function createClickClackClient(options: ClientOptions) { for (const [key, value] of Object.entries(headers)) { requestHeaders.set(key, value); } + if (correlationId) { + requestHeaders.set(CLICKCLACK_CORRELATION_ID_HEADER, correlationId); + } if (init.body && !(init.body instanceof FormData)) { requestHeaders.set("Content-Type", "application/json"); } diff --git a/extensions/clickclack/src/inbound.test.ts b/extensions/clickclack/src/inbound.test.ts index ddf26eb16028..edb615609fd0 100644 --- a/extensions/clickclack/src/inbound.test.ts +++ b/extensions/clickclack/src/inbound.test.ts @@ -2,12 +2,14 @@ import { createPluginRuntimeMock } from "openclaw/plugin-sdk/channel-test-helpers"; import type { PluginRuntime } from "openclaw/plugin-sdk/core"; import { buildAgentSessionKey, resolveAgentRoute } from "openclaw/plugin-sdk/routing"; -import { describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import { handleClickClackInbound } from "./inbound.js"; import { setClickClackRuntime } from "./runtime.js"; import type { ClickClackMessage, CoreConfig, ResolvedClickClackAccount } from "./types.js"; const sendClickClackTextMock = vi.hoisted(() => vi.fn()); +const VALID_MESSAGE_ID = "msg_01arz3ndektsv4rrffq69g5fav"; +const SECOND_VALID_MESSAGE_ID = "msg_01arz3ndektsv4rrffq69g5faw"; type LlmCompleteMock = ReturnType< typeof vi.fn< @@ -114,8 +116,11 @@ function createMessage(overrides: Partial = {}): ClickClackMe } describe("handleClickClackInbound", () => { - it("runs model-mode bot accounts without tools and posts the bot reply", async () => { + beforeEach(() => { sendClickClackTextMock.mockReset(); + }); + + it("runs model-mode bot accounts without tools and posts the bot reply", async () => { const runtime = createRuntime(); setClickClackRuntime(runtime); const cfg = { @@ -164,6 +169,7 @@ describe("handleClickClackInbound", () => { created_at: "2026-05-09T12:00:00.000Z", }, }, + correlationId: "fakeco.case_1", }); expect(runtime.channel.inbound.dispatchReply).not.toHaveBeenCalled(); @@ -180,6 +186,7 @@ describe("handleClickClackInbound", () => { expect(sendRequest?.to).toBe("channel:chn_1"); expect(sendRequest?.text).toBe("service bot online"); expect(sendRequest?.replyToId).toBe("msg_1"); + expect(sendRequest?.correlationId).toBe("fakeco.case_1"); }); it("marks agent turns command-authorized for allowlisted senders", async () => { @@ -254,32 +261,39 @@ describe("handleClickClackInbound", () => { await handleClickClackInbound({ account: createAgentAccount(), config: cfg, - message: createMessage(), + message: createMessage({ + id: VALID_MESSAGE_ID, + thread_root_id: VALID_MESSAGE_ID, + }), }); await handleClickClackInbound({ account: createAgentAccount({ agentActivity: true }), config: cfg, - message: createMessage({ id: "msg_2" }), + message: createMessage({ + id: SECOND_VALID_MESSAGE_ID, + thread_root_id: SECOND_VALID_MESSAGE_ID, + }), }); const dispatchReply = vi.mocked(runtime.channel.inbound.dispatchReply); expect(dispatchReply).toHaveBeenCalledTimes(2); const withoutOptIn = dispatchReply.mock.calls[0]?.[0] as { - replyOptions?: { onItemEvent?: unknown; onModelSelected?: unknown }; + replyOptions?: { runId?: unknown; onItemEvent?: unknown; onModelSelected?: unknown }; }; const withOptIn = dispatchReply.mock.calls[1]?.[0] as { replyOptions?: { onItemEvent?: unknown; onModelSelected?: unknown; + runId?: unknown; commentaryProgressEnabled?: unknown; suppressDefaultToolProgressMessages?: unknown; allowProgressCallbacksWhenSourceDeliverySuppressed?: unknown; }; }; - // Provenance capture and activity item events both ride the agentActivity - // opt-in: with the flag off, reply wire payloads stay byte-identical to - // pre-activity builds. - expect(withoutOptIn.replyOptions).toBeUndefined(); + expect(withoutOptIn.replyOptions).toEqual({ + runId: `clickclack:${VALID_MESSAGE_ID}`, + }); + expect(withOptIn.replyOptions?.runId).toBe(`clickclack:${SECOND_VALID_MESSAGE_ID}`); expect(typeof withOptIn.replyOptions?.onModelSelected).toBe("function"); expect(withOptIn.replyOptions?.commentaryProgressEnabled).toBe(true); // Channel-owned progress rendering: item events must flow even when @@ -289,6 +303,49 @@ describe("handleClickClackInbound", () => { expect(typeof withOptIn.replyOptions?.onItemEvent).toBe("function"); }); + it("maps the authoritative message id to the agent run and correlates the final reply", async () => { + const runtime = createRuntime(); + setClickClackRuntime(runtime); + + await handleClickClackInbound({ + account: createAgentAccount(), + config: {} as CoreConfig, + message: createMessage({ + id: VALID_MESSAGE_ID, + thread_root_id: VALID_MESSAGE_ID, + }), + correlationId: "fakeco.case_2", + }); + + const dispatchParams = vi.mocked(runtime.channel.inbound.dispatchReply).mock.calls[0]?.[0]; + expect(dispatchParams?.replyOptions?.runId).toBe(`clickclack:${VALID_MESSAGE_ID}`); + + await dispatchParams?.delivery.deliver({ text: "correlated reply" }, {} as never); + + expect(sendClickClackTextMock).toHaveBeenCalledWith( + expect.objectContaining({ + correlationId: "fakeco.case_2", + replyToId: VALID_MESSAGE_ID, + text: "correlated reply", + }), + ); + }); + + it("does not derive a run id from a noncanonical message id", async () => { + const runtime = createRuntime(); + setClickClackRuntime(runtime); + + await handleClickClackInbound({ + account: createAgentAccount(), + config: {} as CoreConfig, + message: createMessage({ id: "msg_invalid" }), + }); + + expect(vi.mocked(runtime.channel.inbound.dispatchReply).mock.calls[0]?.[0].replyOptions).toBe( + undefined, + ); + }); + it("accepts ClickClack DM target syntax in allowFrom", async () => { const runtime = createRuntime(); vi.mocked(runtime.channel.commands.shouldComputeCommandAuthorized).mockReturnValue(true); diff --git a/extensions/clickclack/src/inbound.ts b/extensions/clickclack/src/inbound.ts index cdd511b00cd3..fd64a2761d34 100644 --- a/extensions/clickclack/src/inbound.ts +++ b/extensions/clickclack/src/inbound.ts @@ -18,6 +18,11 @@ import type { } from "./types.js"; const CHANNEL_ID = "clickclack" as const; +const CLICKCLACK_MESSAGE_ID_PATTERN = /^msg_[0-9a-hjkmnp-tv-z]{26}$/u; + +function resolveClickClackAgentRunId(messageId: string): string | undefined { + return CLICKCLACK_MESSAGE_ID_PATTERN.test(messageId) ? `${CHANNEL_ID}:${messageId}` : undefined; +} function resolveAccountAgentRoute(params: { cfg: OpenClawConfig; @@ -78,6 +83,7 @@ async function dispatchModelReply(params: { message: ClickClackMessage; route: { agentId: string }; target: string; + correlationId?: string; }) { const runtime = getClickClackRuntime(); const result = await runtime.llm.complete({ @@ -104,6 +110,7 @@ async function dispatchModelReply(params: { text, threadId: params.message.parent_message_id ? params.message.thread_root_id : undefined, replyToId: params.message.id, + correlationId: params.correlationId, }); } @@ -116,6 +123,7 @@ export async function handleClickClackInbound(params: { config: CoreConfig; message: ClickClackMessage; access?: ClickClackInboundAccess; + correlationId?: string; }) { const runtime = getClickClackRuntime(); const message = params.message; @@ -148,6 +156,7 @@ export async function handleClickClackInbound(params: { message, route, target, + correlationId: params.correlationId, }); return; } @@ -164,6 +173,7 @@ export async function handleClickClackInbound(params: { client: createClickClackClient({ baseUrl: params.account.baseUrl, token: params.account.token, + correlationId: params.correlationId, }), target: message.channel_id ? { channelId: message.channel_id } @@ -224,6 +234,25 @@ export async function handleClickClackInbound(params: { OriginatingTo: target, CommandAuthorized: access.commandAuthorized, }); + const runId = resolveClickClackAgentRunId(message.id); + const activityReplyOptions = activity + ? { + onModelSelected: (ctx: { provider: string; model: string; thinkLevel?: string }) => { + turnProvenance = { + model: ctx.provider && ctx.model ? `${ctx.provider}/${ctx.model}` : ctx.model, + thinking: ctx.thinkLevel, + }; + activity?.setProvenance(turnProvenance); + }, + onItemEvent: activity.onItemEvent, + commentaryProgressEnabled: true, + // The durable activity rows are ClickClack's own progress + // rendering, so item events must flow even when session verbose + // mode is off and the default tool-progress texts stay suppressed. + suppressDefaultToolProgressMessages: true, + allowProgressCallbacksWhenSourceDeliverySuppressed: true, + } + : undefined; const dispatchPromise = runtime.channel.inbound.dispatchReply({ cfg: params.config as OpenClawConfig, channel: CHANNEL_ID, @@ -239,24 +268,13 @@ export async function handleClickClackInbound(params: { // Provenance stamping shares the agentActivity opt-in: with the flag off // the extension's wire payloads stay byte-identical to pre-activity // builds, which is the documented contract for stock setups. - replyOptions: activity - ? { - onModelSelected: (ctx: { provider: string; model: string; thinkLevel?: string }) => { - turnProvenance = { - model: ctx.provider && ctx.model ? `${ctx.provider}/${ctx.model}` : ctx.model, - thinking: ctx.thinkLevel, - }; - activity?.setProvenance(turnProvenance); - }, - onItemEvent: activity.onItemEvent, - commentaryProgressEnabled: true, - // The durable activity rows are ClickClack's own progress - // rendering, so item events must flow even when session verbose - // mode is off and the default tool-progress texts stay suppressed. - suppressDefaultToolProgressMessages: true, - allowProgressCallbacksWhenSourceDeliverySuppressed: true, - } - : undefined, + replyOptions: + runId || activityReplyOptions + ? { + ...(runId ? { runId } : {}), + ...activityReplyOptions, + } + : undefined, delivery: { deliver: async (payload) => { const text = @@ -274,6 +292,7 @@ export async function handleClickClackInbound(params: { threadId: message.parent_message_id ? message.thread_root_id : undefined, replyToId: message.id, provenance: turnProvenance, + correlationId: params.correlationId, }); }, onError: (error) => { diff --git a/extensions/clickclack/src/outbound.test.ts b/extensions/clickclack/src/outbound.test.ts index 4ebe35b038b4..3a43c4bfff01 100644 --- a/extensions/clickclack/src/outbound.test.ts +++ b/extensions/clickclack/src/outbound.test.ts @@ -8,6 +8,7 @@ const createChannelMessage = vi.hoisted(() => vi.fn(async () => ({ id: "msg_out" const createThreadReply = vi.hoisted(() => vi.fn(async () => ({ id: "msg_out" }))); const createDirectMessage = vi.hoisted(() => vi.fn(async () => ({ id: "msg_out" }))); const createDirectConversation = vi.hoisted(() => vi.fn(async () => ({ id: "dm_1" }))); +const createClientOptions = vi.hoisted(() => vi.fn()); vi.mock("./accounts.js", () => ({ resolveClickClackAccount: () => ({ @@ -18,12 +19,15 @@ vi.mock("./accounts.js", () => ({ })); vi.mock("./http-client.js", () => ({ - createClickClackClient: () => ({ - createChannelMessage, - createThreadReply, - createDirectMessage, - createDirectConversation, - }), + createClickClackClient: (options: unknown) => { + createClientOptions(options); + return { + createChannelMessage, + createThreadReply, + createDirectMessage, + createDirectConversation, + }; + }, })); vi.mock("./resolve.js", () => ({ @@ -39,6 +43,7 @@ describe("sendClickClackText routing", () => { createThreadReply.mockClear(); createDirectMessage.mockClear(); createDirectConversation.mockClear(); + createClientOptions.mockClear(); }); it("delivers a reply to a top-level channel message as an in-channel quote-reply", async () => { @@ -69,6 +74,21 @@ describe("sendClickClackText routing", () => { expect(createThreadReply).not.toHaveBeenCalled(); }); + it("uses the inbound correlation id for outbound ClickClack HTTP calls", async () => { + await sendClickClackText({ + cfg, + to: "channel:general", + text: "hi", + correlationId: "fakeco.case_1", + }); + + expect(createClientOptions).toHaveBeenCalledWith({ + baseUrl: "https://clickclack.example", + token: "test-token", + correlationId: "fakeco.case_1", + }); + }); + it("keeps replies inside a genuine thread (explicit threadId)", async () => { await sendClickClackText({ cfg, diff --git a/extensions/clickclack/src/outbound.ts b/extensions/clickclack/src/outbound.ts index efe801892554..8de6522d7ad8 100644 --- a/extensions/clickclack/src/outbound.ts +++ b/extensions/clickclack/src/outbound.ts @@ -19,11 +19,17 @@ export async function sendClickClackText(params: { text: string; threadId?: string | number | null; replyToId?: string | number | null; + /** Safe request correlation inherited from an inbound ClickClack event. */ + correlationId?: string; /** Optional model/thinking attribution stamped onto the created message. */ provenance?: ClickClackMessageProvenance; }) { const account = resolveClickClackAccount({ cfg: params.cfg, accountId: params.accountId }); - const client = createClickClackClient({ baseUrl: account.baseUrl, token: account.token }); + const client = createClickClackClient({ + baseUrl: account.baseUrl, + token: account.token, + correlationId: params.correlationId, + }); const workspaceId = await resolveWorkspaceId(client, account.workspace); const parsed = parseClickClackTarget(params.to); const explicitThreadId = params.threadId == null ? "" : String(params.threadId); diff --git a/packages/llm-core/src/types.ts b/packages/llm-core/src/types.ts index 646ceba91639..83e6b4c0652a 100644 --- a/packages/llm-core/src/types.ts +++ b/packages/llm-core/src/types.ts @@ -91,6 +91,11 @@ export interface StreamOptions { * session-aware features. Ignored by providers that don't support it. */ sessionId?: string; + /** + * Opaque per-model-call identifier for provider transport correlation. + * Providers that do not expose request correlation ignore it. + */ + requestId?: string; /** * Optional provider prompt-cache affinity key, distinct from transcript/session identity. * Providers that do not support separate cache affinity ignore it. diff --git a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.test.ts b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.test.ts index c27634291836..46f6beaa4e2c 100644 --- a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.test.ts @@ -973,6 +973,7 @@ describe("wrapStreamFnWithDiagnosticModelCallEvents", () => { expect(capturedOptions[0]).not.toBe(callerOptions); const capturedOption = requireRecord(capturedOptions[0], "captured stream options"); expect(capturedOption.sessionId).toBe("provider-session"); + expect(capturedOption.requestId).toBe("call-traceparent"); const headers = readRecordField(capturedOption, "headers", "captured stream headers"); expect(headers["X-Custom"]).toBe("kept"); expect(typeof headers.traceparent).toBe("string"); diff --git a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.ts b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.ts index e60c12461881..894a6e9fd9ae 100644 --- a/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.ts +++ b/src/agents/embedded-agent-runner/run/attempt.model-diagnostic-events.ts @@ -622,10 +622,11 @@ function emitModelCallError( }); } -function withDiagnosticTraceparentHeader( +function withDiagnosticRequestContext( options: ModelCallStreamOptions, trace: DiagnosticTraceContext, state: ModelCallObservationState, + callId: string, ): ModelCallStreamOptions { const traceparent = formatDiagnosticTraceparent(trace); const originalOnPayload = options?.onPayload; @@ -648,6 +649,7 @@ function withDiagnosticTraceparentHeader( if (!traceparent) { return { ...options, + requestId: callId, onPayload, }; } @@ -662,6 +664,7 @@ function withDiagnosticTraceparentHeader( headers[TRACEPARENT_HEADER_NAME] = traceparent; return { ...options, + requestId: callId, headers, onPayload, }; @@ -867,7 +870,9 @@ export function wrapStreamFnWithDiagnosticModelCallEvents( modelContent, contentCapture: ctx.contentCapture, }; - const propagatedOptions = withDiagnosticTraceparentHeader(options, trace, state); + // Provider wrappers consume this same call id for transport correlation, + // keeping external request evidence joined to the emitted diagnostics. + const propagatedOptions = withDiagnosticRequestContext(options, trace, state, callId); try { const result = streamFn(model, streamContext, propagatedOptions); diff --git a/test/clawrouter-managed-gateway.e2e.test.ts b/test/clawrouter-managed-gateway.e2e.test.ts index 8c468d1af8f0..6469923f1177 100644 --- a/test/clawrouter-managed-gateway.e2e.test.ts +++ b/test/clawrouter-managed-gateway.e2e.test.ts @@ -183,10 +183,14 @@ describe("ClawRouter managed gateway contract", () => { }, }); const sessionId = inferenceRequests.at(-1)?.headers["x-clawrouter-session-id"]; + const requestId = inferenceRequests.at(-1)?.headers["x-request-id"]; expect(JSON.stringify(inferenceRequests.at(-1)?.body)).toContain(SUCCESS_MARKER); expect(typeof sessionId).toBe("string"); expect(String(sessionId).length).toBeGreaterThan(0); expect(String(sessionId).length).toBeLessThanOrEqual(256); + expect(typeof requestId).toBe("string"); + expect(String(requestId)).toMatch(/:model:\d+$/u); + expect(String(requestId).length).toBeLessThanOrEqual(128); expect(instance.logs()).toContain( `[model-fetch] start provider=clawrouter api=openai-responses model=${MODEL_ID} method=POST url=${router.baseUrl}/v1/responses`,