From 51fc5aa3db1a3f5f2df083c3bc78121334ce28b8 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 7 Aug 2026 04:33:48 +0200 Subject: [PATCH] fix(agents): reject attribution collisions before capture --- .../agent-command-execution-identity.ts | 6 +++ .../agent-command.live-model-switch.test.ts | 32 +++++++++++- .../reply/agent-runner-execution-identity.ts | 6 +++ .../agent-runner-execution-runtime.test.ts | 52 ++++++++++++++++++- .../reply/agent-runner-execution.ts | 34 +++++++++++- src/commands/agent.acp.test.ts | 5 +- src/infra/agent-run-registry.ts | 23 +++++++- 7 files changed, 150 insertions(+), 8 deletions(-) diff --git a/src/agents/agent-command-execution-identity.ts b/src/agents/agent-command-execution-identity.ts index b20068c04450..bdc53d44fe2f 100644 --- a/src/agents/agent-command-execution-identity.ts +++ b/src/agents/agent-command-execution-identity.ts @@ -5,6 +5,7 @@ import { } from "../audit/execution-identity-admission.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { captureAgentRunLifecycleGeneration } from "../infra/agent-events.js"; +import { assertAgentRunAttributionAdmissionCompatible } from "../infra/agent-run-registry.js"; import { createAgentExecutionAttribution } from "./agent-execution-attribution.js"; import type { AgentCommandGatewayIngressOpts, AgentCommandOpts } from "./command/types.js"; @@ -74,6 +75,11 @@ function resolveAgentCommandExecutionAttribution( opts.executionAttribution?.lifecycleGeneration ?? opts.lifecycleGeneration ?? captureAgentRunLifecycleGeneration(params.runId); + assertAgentRunAttributionAdmissionCompatible( + params.runId, + lifecycleGeneration, + opts.executionAttribution, + ); return { attribution: opts.executionAttribution ?? diff --git a/src/agents/agent-command.live-model-switch.test.ts b/src/agents/agent-command.live-model-switch.test.ts index a9325597d59f..7fcd825e31b5 100644 --- a/src/agents/agent-command.live-model-switch.test.ts +++ b/src/agents/agent-command.live-model-switch.test.ts @@ -4,6 +4,7 @@ import { expectDefined } from "@openclaw/normalization-core"; import { createRequireRecord } from "openclaw/plugin-sdk/test-fixtures"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import type { SessionEntry } from "../config/sessions.js"; +import * as agentRunRegistry from "../infra/agent-run-registry.js"; import { createUserTurnTranscriptRecorder } from "../sessions/user-turn-transcript.js"; import { deliveryContextFromSession, @@ -387,7 +388,10 @@ vi.mock("../infra/agent-events.js", () => ({ registerAgentEventLifecycleRotationHandler: vi.fn(), withAgentRunLifecycleGeneration: (_generation: string, run: () => unknown) => run(), })); -vi.mock("../infra/agent-run-registry.js", () => ({ +vi.mock("../infra/agent-run-registry.js", async () => ({ + ...(await vi.importActual( + "../infra/agent-run-registry.js", + )), clearAgentRunContext: (...args: unknown[]) => state.clearAgentRunContextMock(...args), registerAgentRunContext: (...args: unknown[]) => state.registerAgentRunContextMock(...args), })); @@ -1043,6 +1047,7 @@ describe("agentCommand – LiveSessionModelSwitchError retry", () => { }); afterEach(() => { + agentRunRegistry.resetAgentRunRegistryForTest(); vi.restoreAllMocks(); }); @@ -4570,6 +4575,31 @@ describe("agentCommand – LiveSessionModelSwitchError retry", () => { ); }); + it("rejects a colliding public ACP run id before recording execution identity", async () => { + setupAcpSession(); + const runId = "session-1"; + agentRunRegistry.claimAgentRunContext(runId, { + attribution: createAgentExecutionAttribution({ + runId, + lifecycleGeneration: "test-generation", + }), + lifecycleGeneration: "test-generation", + }); + + await expect( + agentCommandFromIngress({ + message: "colliding public ACP turn", + sessionKey: "agent:main:main", + runId, + allowModelOverride: false, + }), + ).rejects.toThrow("Agent run ID is already bound to host-owned execution attribution."); + + expect(state.enqueueExecutionIdentityContextAtAdmissionMock).not.toHaveBeenCalled(); + expect(state.registerAgentRunContextMock).not.toHaveBeenCalled(); + expect(state.acpRunTurnMock).not.toHaveBeenCalled(); + }); + it("allows manual ACP spawn turns when ACP dispatch is disabled", async () => { setupAcpSession(); state.resolveAcpDispatchPolicyErrorMock.mockReturnValue( diff --git a/src/auto-reply/reply/agent-runner-execution-identity.ts b/src/auto-reply/reply/agent-runner-execution-identity.ts index a4adaad22482..5ab6e3c8a741 100644 --- a/src/auto-reply/reply/agent-runner-execution-identity.ts +++ b/src/auto-reply/reply/agent-runner-execution-identity.ts @@ -9,6 +9,7 @@ import { type ExecutionIdentityAdmissionFacts, } from "../../audit/execution-identity-admission.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { assertAgentRunAttributionAdmissionCompatible } from "../../infra/agent-run-registry.js"; import type { InputProvenance } from "../../sessions/input-provenance.js"; type AutoReplyExecutionIdentityContext = { @@ -133,6 +134,11 @@ export function admitAutoReplyExecutionAttribution(params: { lifecycleGeneration: string; runId: string; }): AgentExecutionAttribution { + assertAgentRunAttributionAdmissionCompatible( + params.runId, + params.lifecycleGeneration, + params.attribution, + ); if (params.attribution) { return params.attribution; } diff --git a/src/auto-reply/reply/agent-runner-execution-runtime.test.ts b/src/auto-reply/reply/agent-runner-execution-runtime.test.ts index c32c16745583..329f8a0962b2 100644 --- a/src/auto-reply/reply/agent-runner-execution-runtime.test.ts +++ b/src/auto-reply/reply/agent-runner-execution-runtime.test.ts @@ -1,7 +1,8 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { createAgentExecutionAttribution } from "../../agents/agent-execution-attribution.js"; import { testing as cliBackendsTesting } from "../../agents/cli-backends.test-support.js"; import { installSessionPlacementAdmissionProvider } from "../../agents/session-placement-admission.js"; +import { configureExecutionIdentityAdmissionSink } from "../../audit/execution-identity-admission.js"; import type { SessionEntry } from "../../config/sessions.js"; import { getAgentEventLifecycleGeneration, @@ -13,6 +14,7 @@ import { getExecuteAgentTurnForTest, createMockTypingSignaler, createFollowupRun, + GENERIC_RUN_FAILURE_TEXT, requireRecord, requireMockCall, expectMockCallArgFields, @@ -320,12 +322,58 @@ describe("executeAgentTurn: runtime selection", () => { kind: "final", payload: { isError: true, - text: "⚠️ Something went wrong while processing your request. Please try again, or use /new to start a fresh session.", + text: GENERIC_RUN_FAILURE_TEXT, }, }); expect(state.runCliAgentMock).not.toHaveBeenCalled(); expect(agentRunRegistry.getAgentRunContext(runId)?.attribution).toBe(existingAttribution); + agentRunRegistry.resetAgentRunRegistryForTest(); + }); + + it("rejects a fresh auto-reply run id collision before recording execution identity", async () => { + const agentRunRegistry = await import("../../infra/agent-run-registry.js"); + const lifecycleGeneration = getAgentEventLifecycleGeneration(); + const runId = "fresh-auto-reply-attribution-collision"; + const existingAttribution = createAgentExecutionAttribution({ + runId, + lifecycleGeneration, + }); + agentRunRegistry.claimAgentRunContext(runId, { + attribution: existingAttribution, + lifecycleGeneration, + }); + const sink = vi.fn(() => true); + const restoreSink = configureExecutionIdentityAdmissionSink(sink); + const followupRun = createFollowupRun(); + followupRun.run.config = { + logging: { audit: { enabled: true, executionIdentity: true } }, + }; + + try { + const executeAgentTurn = await getExecuteAgentTurnForTest(); + await expect( + executeAgentTurn( + createMinimalRunAgentTurnParams({ + followupRun, + opts: { runId }, + }), + ), + ).resolves.toEqual({ + kind: "final", + payload: { + isError: true, + text: GENERIC_RUN_FAILURE_TEXT, + }, + }); + + expect(sink).not.toHaveBeenCalled(); + expect(state.runEmbeddedAgentMock).not.toHaveBeenCalled(); + expect(agentRunRegistry.getAgentRunContext(runId)?.attribution).toBe(existingAttribution); + } finally { + restoreSink(); + agentRunRegistry.resetAgentRunRegistryForTest(); + } }); it("rejects queued heartbeat CLI fallback after placement crosses a lifecycle rotation", async () => { diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index e8e0f49b4970..80c136308953 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/src/auto-reply/reply/agent-runner-execution.ts @@ -25,7 +25,11 @@ import { captureAgentRunLifecycleGeneration, withAgentRunLifecycleGeneration, } from "../../infra/agent-events.js"; -import { clearAgentRunContext, registerAgentRunContext } from "../../infra/agent-run-registry.js"; +import { + AgentRunAttributionCollisionError, + clearAgentRunContext, + registerAgentRunContext, +} from "../../infra/agent-run-registry.js"; import { emitAgentRunStatusEvent } from "../../infra/agent-run-status-events.js"; import { isDiagnosticsEnabled } from "../../infra/diagnostic-events.js"; import { formatErrorMessage } from "../../infra/errors.js"; @@ -49,6 +53,7 @@ import type { AgentTurnParams, RuntimeFallbackAttempt, } from "./agent-runner-execution.types.js"; +import { GENERIC_EXTERNAL_RUN_FAILURE_TEXT } from "./agent-runner-failure-copy.js"; import { buildTerminalAgentRunFailureReplyPayload, markAgentRunFailureReplyPayload, @@ -505,13 +510,28 @@ function resolveAgentTurnRunId(params: AgentTurnParams): string { return attributedRunId ?? requestedRunId ?? crypto.randomUUID(); } +function tryAdmitAgentTurnExecutionAttribution( + params: Parameters[0], +): + | { kind: "admitted"; attribution: ReturnType } + | { kind: "collision" } { + try { + return { kind: "admitted", attribution: admitAutoReplyExecutionAttribution(params) }; + } catch (error) { + if (error instanceof AgentRunAttributionCollisionError) { + return { kind: "collision" }; + } + throw error; + } +} + /** Runs the agent turn with provider/model fallback, retry, and closed settlement. */ export async function executeAgentTurn(params: AgentTurnParams): Promise { const runId = resolveAgentTurnRunId(params); const baseExecutionParams = params.opts?.runId === runId ? params : { ...params, opts: { ...params.opts, runId } }; const lifecycleGeneration = captureAgentRunLifecycleGeneration(runId); - const attribution = admitAutoReplyExecutionAttribution({ + const attributionAdmission = tryAdmitAgentTurnExecutionAttribution({ attribution: baseExecutionParams.attribution, config: resolveQueuedReplyRuntimeConfig(baseExecutionParams.followupRun.run.config), lifecycleGeneration, @@ -542,6 +562,16 @@ export async function executeAgentTurn(params: AgentTurnParams): Promise ({ })); vi.mock("../infra/agent-events.js", () => agentEventMocks); -vi.mock("../infra/agent-run-registry.js", () => ({ +vi.mock("../infra/agent-run-registry.js", async () => ({ + ...(await vi.importActual( + "../infra/agent-run-registry.js", + )), clearAgentRunContext: agentEventMocks.clearAgentRunContext, registerAgentRunContext: agentEventMocks.registerAgentRunContext, })); diff --git a/src/infra/agent-run-registry.ts b/src/infra/agent-run-registry.ts index 0b9c6a0885f3..c922303c07bd 100644 --- a/src/infra/agent-run-registry.ts +++ b/src/infra/agent-run-registry.ts @@ -53,6 +53,8 @@ type AgentRunRegistryState = { const AGENT_RUN_REGISTRY_STATE_KEY = Symbol.for("openclaw.agentRunRegistry.state"); +export class AgentRunAttributionCollisionError extends TypeError {} + function getAgentRunRegistryState(): AgentRunRegistryState { return resolveGlobalSingleton(AGENT_RUN_REGISTRY_STATE_KEY, () => ({ contexts: new Map(), @@ -87,7 +89,9 @@ export function assertAgentRunAttributionCompatible( attribution: AgentExecutionAttribution | undefined, ): void { if (existingAttribution && !attribution) { - throw new TypeError("Agent run ID is already bound to host-owned execution attribution."); + throw new AgentRunAttributionCollisionError( + "Agent run ID is already bound to host-owned execution attribution.", + ); } if ( existingAttribution && @@ -96,10 +100,25 @@ export function assertAgentRunAttributionCompatible( existingAttribution.executionId !== attribution.executionId || existingAttribution.createdAt !== attribution.createdAt) ) { - throw new TypeError("Agent run ID is already bound to different execution attribution."); + throw new AgentRunAttributionCollisionError( + "Agent run ID is already bound to different execution attribution.", + ); } } +/** Rejects attribution collisions before admission-owned audit capture can observe them. */ +export function assertAgentRunAttributionAdmissionCompatible( + runId: string, + lifecycleGeneration: string, + attribution: AgentExecutionAttribution | undefined, +): void { + const existing = getAgentRunRegistryState().contexts.get(runId); + if (existing?.lifecycleGeneration !== lifecycleGeneration) { + return; + } + assertAgentRunAttributionCompatible(existing.attribution, attribution); +} + function createAgentRunContext( context: AgentRunContext, lifecycleGeneration: string,