mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-22 10:25:20 -06:00
Revert "refactor(agents): centralize exact execution attribution"
This reverts commit e628b42a49.
This commit is contained in:
joshavant
@@ -19,7 +19,7 @@ function systemIngress(boundary: string): AgentCommandAdmissionIngress {
|
||||
}
|
||||
|
||||
function recordAgentCommandExecutionIdentity(params: {
|
||||
attribution?: AgentCommandOpts["executionAttribution"];
|
||||
admission?: AgentCommandOpts["executionIdentityAdmission"];
|
||||
agentId: string;
|
||||
cfg: OpenClawConfig;
|
||||
ingress: AgentCommandAdmissionIngress;
|
||||
@@ -37,17 +37,8 @@ function recordAgentCommandExecutionIdentity(params: {
|
||||
},
|
||||
{
|
||||
enabled: isExecutionIdentityCollectionEnabled(params.cfg),
|
||||
...(params.attribution
|
||||
? params.attribution.executionIdentityAdmission
|
||||
? {
|
||||
token: params.attribution.executionIdentityAdmission.token,
|
||||
retryOnly: params.attribution.executionIdentityAdmission.retryOnly,
|
||||
}
|
||||
: {
|
||||
contextId: params.attribution.contextId,
|
||||
executionId: params.attribution.executionId,
|
||||
now: params.attribution.createdAt,
|
||||
}
|
||||
...(params.admission
|
||||
? { token: params.admission.token, retryOnly: params.admission.retryOnly }
|
||||
: {}),
|
||||
},
|
||||
);
|
||||
|
||||
@@ -231,7 +231,7 @@ async function agentCommandInternal(
|
||||
});
|
||||
return await sessionWorkAdmission.run(async () => {
|
||||
executionIdentity.record({
|
||||
attribution: opts.executionAttribution,
|
||||
admission: opts.executionIdentityAdmission,
|
||||
agentId: sessionAgentId,
|
||||
cfg,
|
||||
ingress: admissionIngress,
|
||||
@@ -674,10 +674,10 @@ export async function agentCommandFromIngress(
|
||||
runtime: RuntimeEnv = defaultRuntime,
|
||||
deps?: CliDeps,
|
||||
) {
|
||||
// Plugin SDK callers may be plain JavaScript. Enforce the private execution
|
||||
// Plugin SDK callers may be plain JavaScript. Enforce the private recovery
|
||||
// boundary at runtime so extra or inherited properties cannot author audit identity.
|
||||
return await agentCommandFromIngressInternal(
|
||||
{ ...opts, executionAttribution: undefined },
|
||||
{ ...opts, executionIdentityAdmission: undefined },
|
||||
runtime,
|
||||
deps,
|
||||
);
|
||||
|
||||
@@ -1,125 +0,0 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { createExecutionIdentityAdmissionToken } from "../audit/execution-identity-admission.js";
|
||||
import {
|
||||
createAgentExecutionAttribution,
|
||||
rebindAgentExecutionAttribution,
|
||||
} from "./agent-execution-attribution.js";
|
||||
|
||||
describe("createAgentExecutionAttribution", () => {
|
||||
it("preserves required identities, normalizes optional correlation, and freezes the record", () => {
|
||||
const token = createExecutionIdentityAdmissionToken(" run-1 ", {
|
||||
contextId: "context-1",
|
||||
executionId: "execution-1",
|
||||
now: 123,
|
||||
});
|
||||
const attribution = createAgentExecutionAttribution({
|
||||
runId: " run-1 ",
|
||||
lifecycleGeneration: " generation-1 ",
|
||||
sessionKey: " agent:main:main ",
|
||||
sessionId: " session-1 ",
|
||||
agentId: " main ",
|
||||
executionIdentityAdmission: { token, retryOnly: true },
|
||||
});
|
||||
|
||||
expect(attribution).toEqual({
|
||||
runId: " run-1 ",
|
||||
contextId: "context-1",
|
||||
executionId: "execution-1",
|
||||
createdAt: 123,
|
||||
lifecycleGeneration: " generation-1 ",
|
||||
executionIdentityAdmission: { token, retryOnly: true },
|
||||
sessionKey: "agent:main:main",
|
||||
sessionId: "session-1",
|
||||
agentId: "main",
|
||||
});
|
||||
expect(Object.isFrozen(attribution)).toBe(true);
|
||||
expect(Reflect.set(attribution, "sessionId", "replacement")).toBe(false);
|
||||
});
|
||||
|
||||
it("leaves unknown optional correlation absent", () => {
|
||||
const attribution = createAgentExecutionAttribution({
|
||||
runId: "run-1",
|
||||
lifecycleGeneration: "generation-1",
|
||||
sessionKey: " ",
|
||||
sessionId: "",
|
||||
});
|
||||
expect(attribution).toMatchObject({
|
||||
runId: "run-1",
|
||||
lifecycleGeneration: "generation-1",
|
||||
});
|
||||
expect(attribution).not.toHaveProperty("sessionKey");
|
||||
expect(attribution).not.toHaveProperty("sessionId");
|
||||
expect(attribution).not.toHaveProperty("executionIdentityAdmission");
|
||||
expect(attribution.contextId).toBeTruthy();
|
||||
expect(attribution.executionId).toBeTruthy();
|
||||
expect(attribution.createdAt).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("does not apply the opt-in audit token bounds to default runtime attribution", () => {
|
||||
const runId = "r".repeat(1_024);
|
||||
const attribution = createAgentExecutionAttribution({
|
||||
runId,
|
||||
lifecycleGeneration: "generation-1",
|
||||
});
|
||||
|
||||
expect(attribution.runId).toBe(runId);
|
||||
expect(attribution).not.toHaveProperty("executionIdentityAdmission");
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
"changes only lifecycle ownership when rebound (token=%s)",
|
||||
(withToken) => {
|
||||
const runId = "run-rebound";
|
||||
const attribution = createAgentExecutionAttribution({
|
||||
runId,
|
||||
lifecycleGeneration: "generation-1",
|
||||
...(withToken
|
||||
? {
|
||||
executionIdentityAdmission: {
|
||||
token: createExecutionIdentityAdmissionToken(runId, {
|
||||
contextId: "context-1",
|
||||
executionId: "execution-1",
|
||||
now: 123,
|
||||
}),
|
||||
retryOnly: true,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
|
||||
const rebound = rebindAgentExecutionAttribution(attribution, "generation-2");
|
||||
|
||||
expect(rebound).toEqual({
|
||||
...attribution,
|
||||
lifecycleGeneration: "generation-2",
|
||||
});
|
||||
expect(rebound.contextId).toBe(attribution.contextId);
|
||||
expect(rebound.executionId).toBe(attribution.executionId);
|
||||
expect(rebound.createdAt).toBe(attribution.createdAt);
|
||||
expect(rebound.executionIdentityAdmission).toBe(attribution.executionIdentityAdmission);
|
||||
expect(Object.isFrozen(rebound)).toBe(true);
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
["runId", { runId: " ", lifecycleGeneration: "generation-1" }],
|
||||
["lifecycleGeneration", { runId: "run-1", lifecycleGeneration: "" }],
|
||||
])("rejects a missing required %s", (field, params) => {
|
||||
expect(() => createAgentExecutionAttribution(params)).toThrow(
|
||||
`Agent execution attribution requires ${field}`,
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects an admission token owned by another run", () => {
|
||||
expect(() =>
|
||||
createAgentExecutionAttribution({
|
||||
runId: "run-1",
|
||||
lifecycleGeneration: "generation-1",
|
||||
executionIdentityAdmission: {
|
||||
token: createExecutionIdentityAdmissionToken("run-2"),
|
||||
retryOnly: false,
|
||||
},
|
||||
}),
|
||||
).toThrow("Agent execution attribution token disagrees with runId");
|
||||
});
|
||||
});
|
||||
@@ -1,79 +0,0 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import {
|
||||
parseExecutionIdentityAdmissionToken,
|
||||
type ExecutionIdentityAdmissionToken,
|
||||
} from "../audit/execution-identity-admission.js";
|
||||
|
||||
export type AgentExecutionIdentityAdmission = Readonly<{
|
||||
token: ExecutionIdentityAdmissionToken;
|
||||
retryOnly: boolean;
|
||||
}>;
|
||||
|
||||
/** Host-owned correlation captured once for an admitted agent execution. */
|
||||
export type AgentExecutionAttribution = Readonly<{
|
||||
runId: string;
|
||||
contextId: string;
|
||||
executionId: string;
|
||||
createdAt: number;
|
||||
lifecycleGeneration: string;
|
||||
executionIdentityAdmission?: AgentExecutionIdentityAdmission;
|
||||
sessionKey?: string;
|
||||
sessionId?: string;
|
||||
agentId?: string;
|
||||
}>;
|
||||
|
||||
function requireAttributionField(value: string, field: "runId" | "lifecycleGeneration"): string {
|
||||
if (!value.trim()) {
|
||||
throw new TypeError(`Agent execution attribution requires ${field}`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
export function createAgentExecutionAttribution(params: {
|
||||
runId: string;
|
||||
lifecycleGeneration: string;
|
||||
sessionKey?: string;
|
||||
sessionId?: string;
|
||||
agentId?: string;
|
||||
executionIdentityAdmission?: AgentExecutionIdentityAdmission;
|
||||
}): AgentExecutionAttribution {
|
||||
const runId = requireAttributionField(params.runId, "runId");
|
||||
const token = params.executionIdentityAdmission
|
||||
? parseExecutionIdentityAdmissionToken(params.executionIdentityAdmission.token)
|
||||
: undefined;
|
||||
if (token && token.runId !== runId) {
|
||||
throw new TypeError("Agent execution attribution token disagrees with runId");
|
||||
}
|
||||
const sessionKey = normalizeOptionalString(params.sessionKey);
|
||||
const sessionId = normalizeOptionalString(params.sessionId);
|
||||
const agentId = normalizeOptionalString(params.agentId);
|
||||
return Object.freeze({
|
||||
runId,
|
||||
contextId: token?.contextId ?? randomUUID(),
|
||||
executionId: token?.executionId ?? randomUUID(),
|
||||
createdAt: token?.createdAt ?? Date.now(),
|
||||
lifecycleGeneration: requireAttributionField(params.lifecycleGeneration, "lifecycleGeneration"),
|
||||
...(token
|
||||
? {
|
||||
executionIdentityAdmission: Object.freeze({
|
||||
token,
|
||||
retryOnly: params.executionIdentityAdmission?.retryOnly === true,
|
||||
}),
|
||||
}
|
||||
: {}),
|
||||
...(sessionKey ? { sessionKey } : {}),
|
||||
...(sessionId ? { sessionId } : {}),
|
||||
...(agentId ? { agentId } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
export function rebindAgentExecutionAttribution(
|
||||
attribution: AgentExecutionAttribution,
|
||||
lifecycleGeneration: string,
|
||||
): AgentExecutionAttribution {
|
||||
return Object.freeze({
|
||||
...attribution,
|
||||
lifecycleGeneration: requireAttributionField(lifecycleGeneration, "lifecycleGeneration"),
|
||||
});
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import type { FastMode } from "@openclaw/normalization-core/string-coerce";
|
||||
import type { AgentInternalEvent } from "../../agents/internal-events.js";
|
||||
import type { SpawnedRunMetadata } from "../../agents/spawned-context.js";
|
||||
import type { PromptMode } from "../../agents/system-prompt.types.js";
|
||||
import type { ExecutionIdentityAdmissionToken } from "../../audit/execution-identity-admission.js";
|
||||
import type { SourceReplyDeliveryMode } from "../../auto-reply/get-reply-options.types.js";
|
||||
import type { ChannelOutboundTargetMode } from "../../channels/plugins/types.public.js";
|
||||
import type { MediaFact } from "../../media/media-facts.js";
|
||||
@@ -16,7 +17,6 @@ import type {
|
||||
UserTurnInput,
|
||||
UserTurnTranscriptRecorder,
|
||||
} from "../../sessions/user-turn-transcript.types.js";
|
||||
import type { AgentExecutionAttribution } from "../agent-execution-attribution.js";
|
||||
import type { ExecApprovalContinuationPromptRange } from "../bash-tools.exec-approval-output.js";
|
||||
import type { ExecElevatedDefaults } from "../bash-tools.exec-types.js";
|
||||
import type { BootstrapContextRunKind } from "../bootstrap-mode.js";
|
||||
@@ -193,8 +193,11 @@ export type AgentCommandOpts = {
|
||||
mainRestartRecoveryOwnerLease?: MainSessionRecoveryOwnerLease;
|
||||
/** Gateway already consumed this automatic recovery run's durable reservation. */
|
||||
mainRestartRecoveryAdmitted?: boolean;
|
||||
/** Private host-owned execution identity; public ingress callers cannot author it. */
|
||||
executionAttribution?: AgentExecutionAttribution;
|
||||
/** Private recovery correlation; public ingress callers cannot author identity evidence. */
|
||||
executionIdentityAdmission?: {
|
||||
token: ExecutionIdentityAdmissionToken;
|
||||
retryOnly: boolean;
|
||||
};
|
||||
/** Called when the actual run model is selected, including fallback retries. */
|
||||
onActiveModelSelected?: (ctx: { provider: string; model: string }) => void | Promise<void>;
|
||||
/** Called when every candidate in the run's model fallback chain failed. */
|
||||
@@ -218,7 +221,7 @@ export type AgentCommandOpts = {
|
||||
/** Restricted option surface for external ingress callsites. */
|
||||
export type AgentCommandIngressOpts = Omit<
|
||||
AgentCommandOpts,
|
||||
"senderIsOwner" | "allowModelOverride" | "executionAttribution"
|
||||
"senderIsOwner" | "allowModelOverride" | "executionIdentityAdmission"
|
||||
> & {
|
||||
/** Trusted sender identity bit for command/channel-action auth; defaults false for ingress. */
|
||||
senderIsOwner?: boolean;
|
||||
@@ -226,6 +229,6 @@ export type AgentCommandIngressOpts = Omit<
|
||||
allowModelOverride: boolean;
|
||||
};
|
||||
|
||||
/** Gateway-only ingress extends the public Plugin SDK surface with private execution correlation. */
|
||||
/** Gateway-only ingress extends the public Plugin SDK surface with private recovery correlation. */
|
||||
export type AgentCommandGatewayIngressOpts = AgentCommandIngressOpts &
|
||||
Pick<AgentCommandOpts, "executionAttribution">;
|
||||
Pick<AgentCommandOpts, "executionIdentityAdmission">;
|
||||
|
||||
@@ -6,7 +6,6 @@ import {
|
||||
} from "../../../infra/agent-events.js";
|
||||
import { claimAgentRunContext, getAgentRunContext } from "../../../infra/agent-run-registry.js";
|
||||
import type { CommandQueueEnqueueOptions } from "../../../process/command-queue.types.js";
|
||||
import { createAgentExecutionAttribution } from "../../agent-execution-attribution.js";
|
||||
import type { EmbeddedAgentRunResult } from "../types.js";
|
||||
import { createEmbeddedRunLaneController } from "./lane-controller.js";
|
||||
import type { RunEmbeddedAgentParams } from "./params.js";
|
||||
@@ -130,41 +129,6 @@ describe("createEmbeddedRunLaneController lifecycle admission", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("rebinds admitted attribution with foreground work across lifecycle rotation", async () => {
|
||||
const queue = deferredTaskQueue();
|
||||
const generation = getAgentEventLifecycleGeneration();
|
||||
const attribution = createAgentExecutionAttribution({
|
||||
runId: "attributed-across-restart",
|
||||
lifecycleGeneration: generation,
|
||||
sessionKey: "agent:main:session-1",
|
||||
sessionId: "session-1",
|
||||
agentId: "main",
|
||||
});
|
||||
claimAgentRunContext(attribution.runId, {
|
||||
attribution,
|
||||
...(attribution.sessionKey ? { sessionKey: attribution.sessionKey } : {}),
|
||||
...(attribution.sessionId ? { sessionId: attribution.sessionId } : {}),
|
||||
...(attribution.agentId ? { agentId: attribution.agentId } : {}),
|
||||
lifecycleGeneration: generation,
|
||||
});
|
||||
const state = createController({
|
||||
lifecycleGeneration: generation,
|
||||
enqueue: queue.enqueue as LaneParams["enqueue"],
|
||||
trigger: "user",
|
||||
runId: attribution.runId,
|
||||
});
|
||||
const run = state.controller.enqueueGlobal(async () => completedResult);
|
||||
|
||||
const currentGeneration = rotateAgentEventLifecycleGeneration();
|
||||
queue.release();
|
||||
await run;
|
||||
|
||||
expect(getAgentRunContext(attribution.runId)?.attribution).toEqual({
|
||||
...attribution,
|
||||
lifecycleGeneration: currentGeneration,
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects background work queued across lifecycle rotation", async () => {
|
||||
const queue = deferredTaskQueue();
|
||||
const generation = getAgentEventLifecycleGeneration();
|
||||
|
||||
@@ -10,7 +10,6 @@ import {
|
||||
} from "../../../infra/agent-run-registry.js";
|
||||
import { enqueueCommandInLane, getCommandLaneSnapshot } from "../../../process/command-queue.js";
|
||||
import type { CommandQueueEnqueueOptions } from "../../../process/command-queue.types.js";
|
||||
import { rebindAgentExecutionAttribution } from "../../agent-execution-attribution.js";
|
||||
import { withSessionPlacementTurnAdmission } from "../../session-placement-admission.js";
|
||||
import type { EmbeddedAgentRunResult } from "../types.js";
|
||||
import {
|
||||
@@ -169,12 +168,8 @@ export function createEmbeddedRunLaneController<TParams extends LaneParams>(opti
|
||||
assertAgentRunLifecycleGenerationCurrent(lifecycleGeneration);
|
||||
releaseQueuedContext("admitted");
|
||||
// Queue-stage rotation may rebind, but placement admitted into a retired runtime must fail.
|
||||
const attribution = existingContext?.attribution
|
||||
? rebindAgentExecutionAttribution(existingContext.attribution, lifecycleGeneration)
|
||||
: undefined;
|
||||
claimAgentRunContext(params.runId, {
|
||||
...existingContext,
|
||||
...(attribution ? { attribution } : {}),
|
||||
sessionKey: params.sessionKey ?? existingContext?.sessionKey,
|
||||
sessionId: params.sessionId ?? existingContext?.sessionId,
|
||||
lifecycleGeneration,
|
||||
|
||||
@@ -498,7 +498,7 @@ describe("agentCommand", () => {
|
||||
);
|
||||
|
||||
expect(record).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ attribution: undefined, runId: "public-ingress-run" }),
|
||||
expect.objectContaining({ admission: undefined, runId: "public-ingress-run" }),
|
||||
);
|
||||
} finally {
|
||||
record.mockRestore();
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js";
|
||||
import { AGENT_RUN_RESTART_ABORT_STOP_REASON } from "../../agents/run-termination.js";
|
||||
import { resolveAgentTimeoutMs } from "../../agents/timeout.js";
|
||||
import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js";
|
||||
@@ -104,27 +103,6 @@ export function createAgentDedupeLifecycle(params: {
|
||||
reserved = false;
|
||||
};
|
||||
|
||||
const failBeforeDispatch = (summary: string) => {
|
||||
const error = errorShape(ErrorCodes.UNAVAILABLE, summary);
|
||||
const payload = {
|
||||
runId: params.runId,
|
||||
status: "error" as const,
|
||||
summary,
|
||||
};
|
||||
// A reserved request owns its idempotency keys even when admission fails.
|
||||
// Persist the terminal failure so retries replay it instead of re-entering admission.
|
||||
accepted = true;
|
||||
setGatewayDedupeEntries({
|
||||
dedupe: params.context.dedupe,
|
||||
keys: params.agentDedupeKeys,
|
||||
entry: { ts: Date.now(), ok: false, payload, error },
|
||||
});
|
||||
params.respond(false, payload, error, {
|
||||
runId: params.runId,
|
||||
error: summary,
|
||||
});
|
||||
};
|
||||
|
||||
const abortForLifecycleRotation = (target?: { sessionKey?: string; agentId?: string }) => {
|
||||
if (params.lifecycleGeneration === getAgentEventLifecycleGeneration()) {
|
||||
return false;
|
||||
@@ -186,7 +164,6 @@ export function createAgentDedupeLifecycle(params: {
|
||||
reservationId,
|
||||
reserve,
|
||||
clearUnaccepted,
|
||||
failBeforeDispatch,
|
||||
abortForLifecycleRotation,
|
||||
isReserved: () => reserved,
|
||||
isAccepted: () => accepted,
|
||||
|
||||
@@ -1,29 +0,0 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { isPreRegistrationAbortedAgentDedupeEntryForSession } from "./agent-dedupe.js";
|
||||
|
||||
describe("agent dedupe", () => {
|
||||
it("compares admitted run identities without normalizing them", () => {
|
||||
const entry = {
|
||||
ts: 1,
|
||||
ok: true,
|
||||
payload: {
|
||||
runId: " padded-agent-run ",
|
||||
status: "timeout",
|
||||
stopReason: "rpc",
|
||||
},
|
||||
} as const;
|
||||
|
||||
expect(
|
||||
isPreRegistrationAbortedAgentDedupeEntryForSession({
|
||||
entry,
|
||||
runId: " padded-agent-run ",
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(
|
||||
isPreRegistrationAbortedAgentDedupeEntryForSession({
|
||||
entry,
|
||||
runId: "padded-agent-run",
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -72,8 +72,7 @@ export function isPreRegistrationAbortedAgentDedupeEntryForSession(params: {
|
||||
return false;
|
||||
}
|
||||
const payload = params.entry.payload;
|
||||
const payloadRunId =
|
||||
typeof payload.runId === "string" && payload.runId.trim() ? payload.runId : "";
|
||||
const payloadRunId = typeof payload.runId === "string" ? payload.runId.trim() : "";
|
||||
if (payloadRunId && payloadRunId !== params.runId) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -73,14 +73,6 @@ export function prepareAgentRequestPreflight(
|
||||
return undefined;
|
||||
}
|
||||
const request = params.params as AgentRunRequest;
|
||||
if (!request.idempotencyKey.trim()) {
|
||||
params.respond(
|
||||
false,
|
||||
undefined,
|
||||
errorShape(ErrorCodes.INVALID_REQUEST, "idempotencyKey must not be blank"),
|
||||
);
|
||||
return undefined;
|
||||
}
|
||||
const cfg = params.context.getRuntimeConfig();
|
||||
const canUseInternalRuntimeHandoff = resolveCanUseInternalRuntimeHandoff(params.client);
|
||||
const requestSessionKey = request.sessionKey?.trim();
|
||||
@@ -278,7 +270,7 @@ export function prepareAgentRequestPreflight(
|
||||
if (cached.ok && isAcceptedAgentDedupePayload(cached.payload)) {
|
||||
const cachedRunId =
|
||||
typeof cached.payload.runId === "string" && cached.payload.runId.trim()
|
||||
? cached.payload.runId
|
||||
? cached.payload.runId.trim()
|
||||
: runId;
|
||||
const cachedSessionKey =
|
||||
typeof cached.payload.sessionKey === "string" && cached.payload.sessionKey.trim()
|
||||
|
||||
@@ -1,10 +1,5 @@
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js";
|
||||
import {
|
||||
createAgentExecutionAttribution,
|
||||
type AgentExecutionIdentityAdmission,
|
||||
type AgentExecutionAttribution,
|
||||
} from "../../agents/agent-execution-attribution.js";
|
||||
import {
|
||||
clearEmbeddedAgentRunAbortabilityForRunId,
|
||||
isEmbeddedAgentRunAbortableForRunId,
|
||||
@@ -50,7 +45,6 @@ import type { GatewayRequestHandlerOptions } from "./types.js";
|
||||
export type PreparedAgentRunDispatch = {
|
||||
activeGatewayWorkAdmission: SessionWorkAdmissionLease;
|
||||
activeRunAbort: ReturnType<typeof registerChatAbortController>;
|
||||
attribution: AgentExecutionAttribution;
|
||||
effectiveProviderOverride?: string;
|
||||
effectiveModelOverride?: string;
|
||||
effectiveThinking?: string;
|
||||
@@ -92,7 +86,6 @@ export async function prepareAgentRunDispatch(params: {
|
||||
inputProvenance?: InputProvenance;
|
||||
isOneShotModelRun: boolean;
|
||||
isRestartRecoveryResumeRun: boolean;
|
||||
executionIdentityAdmission?: AgentExecutionIdentityAdmission;
|
||||
runId: string;
|
||||
agentDedupeKeys: readonly string[];
|
||||
context: GatewayRequestHandlerOptions["context"];
|
||||
@@ -259,16 +252,6 @@ export async function prepareAgentRunDispatch(params: {
|
||||
});
|
||||
return undefined;
|
||||
}
|
||||
const attribution = createAgentExecutionAttribution({
|
||||
runId: params.runId,
|
||||
lifecycleGeneration: params.lifecycleGeneration,
|
||||
sessionKey: params.resolvedSessionKey,
|
||||
sessionId: params.getAdmittedSessionId(),
|
||||
agentId: params.activeSessionAgentId,
|
||||
...(params.executionIdentityAdmission
|
||||
? { executionIdentityAdmission: params.executionIdentityAdmission }
|
||||
: {}),
|
||||
});
|
||||
if (!activeRunAbort.registered) {
|
||||
activeGatewayWorkAdmission.release();
|
||||
} else {
|
||||
@@ -280,14 +263,15 @@ export async function prepareAgentRunDispatch(params: {
|
||||
});
|
||||
}
|
||||
if (params.resolvedSessionKey) {
|
||||
claimAgentRunContext(params.runId, {
|
||||
attribution,
|
||||
...(attribution.sessionKey ? { sessionKey: attribution.sessionKey } : {}),
|
||||
...(attribution.sessionId ? { sessionId: attribution.sessionId } : {}),
|
||||
...(attribution.agentId ? { agentId: attribution.agentId } : {}),
|
||||
...(params.suppressVisibleSessionEffects ? { isControlUiVisible: false } : {}),
|
||||
lifecycleGeneration: attribution.lifecycleGeneration,
|
||||
});
|
||||
claimAgentRunContext(
|
||||
params.runId,
|
||||
params.suppressVisibleSessionEffects
|
||||
? { isControlUiVisible: false, lifecycleGeneration: params.lifecycleGeneration }
|
||||
: {
|
||||
sessionKey: params.resolvedSessionKey,
|
||||
lifecycleGeneration: params.lifecycleGeneration,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -450,7 +434,6 @@ export async function prepareAgentRunDispatch(params: {
|
||||
return {
|
||||
activeGatewayWorkAdmission,
|
||||
activeRunAbort,
|
||||
attribution,
|
||||
effectiveProviderOverride,
|
||||
effectiveModelOverride,
|
||||
effectiveThinking,
|
||||
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
} from "../../agents/main-session-recovery-store.js";
|
||||
import { resolveScheduledToolPolicyContext } from "../../agents/scheduled-tool-policy.js";
|
||||
import { resolveIngressWorkspaceOverrideForSessionRun } from "../../agents/spawned-context.js";
|
||||
import { isExecutionIdentityCollectionEnabled } from "../../audit/audit-config.js";
|
||||
import {
|
||||
setChannelSourceTurnId,
|
||||
setChannelSourceTurnSameThreadRequired,
|
||||
@@ -51,7 +52,10 @@ import {
|
||||
type RestoredCronContinuation,
|
||||
} from "./agent-handler-helpers.js";
|
||||
import type { AgentRunRequest } from "./agent-request-types.js";
|
||||
import { resolveAgentRestartRecoveryChannelContext } from "./agent-restart-recovery-context.js";
|
||||
import {
|
||||
resolveAgentRestartRecoveryChannelContext,
|
||||
resolveAgentRestartRecoveryExecutionIdentityAdmission,
|
||||
} from "./agent-restart-recovery-context.js";
|
||||
import type { PreparedAgentRunDispatch } from "./agent-run-admission-phase.js";
|
||||
import {
|
||||
resolveAbortedAgentStopReason,
|
||||
@@ -325,6 +329,13 @@ export function startAgentRunExecution(params: {
|
||||
params.client.internal.runtimePluginToolGrant?.pluginId
|
||||
? params.client.internal.runtimePluginToolGrant
|
||||
: undefined;
|
||||
const executionIdentityAdmission = resolveAgentRestartRecoveryExecutionIdentityAdmission({
|
||||
collectionEnabled: isExecutionIdentityCollectionEnabled(params.cfg),
|
||||
isRestartRecoveryResumeRun: params.isRestartRecoveryResumeRun,
|
||||
retryOnly: params.request.internalExecutionIdentityRetry,
|
||||
runId: params.runId,
|
||||
sessionEntry: params.sessionEntry,
|
||||
});
|
||||
const restartRecoveryChannelContext = resolveAgentRestartRecoveryChannelContext({
|
||||
canUseInternalRuntimeHandoff: params.canUseInternalRuntimeHandoff,
|
||||
expectedExistingSessionId: params.request.expectedExistingSessionId,
|
||||
@@ -421,7 +432,7 @@ export function startAgentRunExecution(params: {
|
||||
swarmOutputSchema: params.request.swarmOutputSchema,
|
||||
forceRestartSafeTools: params.request.forceRestartSafeTools,
|
||||
forceCodeModeTools: params.request.forceCodeModeTools,
|
||||
executionAttribution: prepared.attribution,
|
||||
...(executionIdentityAdmission ? { executionIdentityAdmission } : {}),
|
||||
internalDeliveryMediaUrls: params.client?.internal?.internalDeliveryMediaUrls,
|
||||
internalDeliverySuppressText: params.client?.internal?.internalDeliverySuppressText,
|
||||
suppressPromptPersistence:
|
||||
|
||||
@@ -5,7 +5,6 @@ import {
|
||||
releaseMainSessionRecoveryOwner,
|
||||
type MainSessionRecoveryOwnerLease,
|
||||
} from "../../agents/main-session-recovery-store.js";
|
||||
import { isExecutionIdentityCollectionEnabled } from "../../audit/audit-config.js";
|
||||
import { mergeSessionEntry, type SessionEntry } from "../../config/sessions.js";
|
||||
import type { OpenClawConfig } from "../../config/types.openclaw.js";
|
||||
import { normalizeDeliveryContext } from "../../utils/delivery-context.shared.js";
|
||||
@@ -20,7 +19,6 @@ import type { RestoredCronContinuation } from "./agent-handler-helpers.js";
|
||||
import { prepareAgentRequestPreflight } from "./agent-request-preflight.js";
|
||||
import { prepareAgentRequestRouting } from "./agent-request-routing.js";
|
||||
import { runAgentResetPhase } from "./agent-reset-phase.js";
|
||||
import { resolveAgentRestartRecoveryExecutionIdentityAdmission } from "./agent-restart-recovery-context.js";
|
||||
import { prepareAgentRunDispatch } from "./agent-run-admission-phase.js";
|
||||
import { startAgentRunExecution } from "./agent-run-execution-phase.js";
|
||||
import { buildAgentSessionPatch } from "./agent-session-patch.js";
|
||||
@@ -434,19 +432,6 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
|
||||
return;
|
||||
}
|
||||
const { activeSessionAgentId } = delivery;
|
||||
let executionIdentityAdmission;
|
||||
try {
|
||||
executionIdentityAdmission = resolveAgentRestartRecoveryExecutionIdentityAdmission({
|
||||
collectionEnabled: isExecutionIdentityCollectionEnabled(cfg),
|
||||
isRestartRecoveryResumeRun,
|
||||
retryOnly: request.internalExecutionIdentityRetry,
|
||||
runId,
|
||||
sessionEntry,
|
||||
});
|
||||
} catch (error) {
|
||||
dedupeLifecycle.failBeforeDispatch(formatForLog(error));
|
||||
return;
|
||||
}
|
||||
|
||||
const preparedDispatch = await prepareAgentRunDispatch({
|
||||
request,
|
||||
@@ -472,7 +457,6 @@ export const agentRunHandler: GatewayRequestHandlers["agent"] = async ({
|
||||
inputProvenance,
|
||||
isOneShotModelRun,
|
||||
isRestartRecoveryResumeRun,
|
||||
executionIdentityAdmission,
|
||||
runId,
|
||||
agentDedupeKeys,
|
||||
context,
|
||||
|
||||
@@ -2309,99 +2309,6 @@ describe("gateway agent handler chat.abort integration", () => {
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("replays restart recovery identity admission failures for the same idempotency key", async () => {
|
||||
const sessionKey = "agent:main:main";
|
||||
const sessionId = "recovery-session";
|
||||
const runId = "recovery-identity-token-unavailable";
|
||||
const storePath = "/tmp/sessions.json";
|
||||
const store: Record<string, SessionEntry> = {
|
||||
[sessionKey]: {
|
||||
sessionId,
|
||||
updatedAt: Date.now() - 10_000,
|
||||
status: "running",
|
||||
abortedLastRun: true,
|
||||
mainRestartRecovery: {
|
||||
cycleId: "cycle-1",
|
||||
revision: 1,
|
||||
chargedAttempts: 1,
|
||||
reservation: {
|
||||
runId,
|
||||
attempt: 1,
|
||||
lifecycleGeneration: "test-generation",
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
mocks.loadConfigReturn = {
|
||||
logging: {
|
||||
audit: {
|
||||
enabled: true,
|
||||
executionIdentity: true,
|
||||
},
|
||||
},
|
||||
};
|
||||
mocks.loadSessionEntry.mockImplementation(() => ({
|
||||
cfg: mocks.loadConfigReturn,
|
||||
storePath,
|
||||
entry: structuredClone(store[sessionKey]),
|
||||
canonicalKey: sessionKey,
|
||||
}));
|
||||
mocks.updateSessionStore.mockImplementation(async (_path, updater) => await updater(store));
|
||||
|
||||
const context = makeContext();
|
||||
const request = {
|
||||
message: "resume after restart",
|
||||
agentId: "main",
|
||||
sessionKey,
|
||||
sessionId,
|
||||
expectedExistingSessionId: sessionId,
|
||||
idempotencyKey: runId,
|
||||
internalExecutionIdentityRetry: false,
|
||||
inputProvenance: {
|
||||
kind: "internal_system" as const,
|
||||
sourceSessionKey: sessionKey,
|
||||
sourceTool: "main_session_restart_recovery",
|
||||
},
|
||||
};
|
||||
const firstRespond = vi.fn();
|
||||
await invokeAgent(request, {
|
||||
client: backendGatewayClient(),
|
||||
context,
|
||||
reqId: runId,
|
||||
respond: firstRespond,
|
||||
});
|
||||
|
||||
const summary = "Error: restart recovery execution identity token is unavailable";
|
||||
expect(firstRespond).toHaveBeenCalledWith(
|
||||
false,
|
||||
{ runId, status: "error", summary },
|
||||
expect.objectContaining({ code: "UNAVAILABLE", message: summary }),
|
||||
{ runId, error: summary },
|
||||
);
|
||||
expect(context.dedupe.get(`agent:${runId}`)).toMatchObject({
|
||||
ok: false,
|
||||
payload: { runId, status: "error", summary },
|
||||
error: { code: "UNAVAILABLE", message: summary },
|
||||
});
|
||||
expect(mocks.agentCommand).not.toHaveBeenCalled();
|
||||
|
||||
const secondRespond = vi.fn();
|
||||
await invokeAgent(request, {
|
||||
client: backendGatewayClient(),
|
||||
context,
|
||||
reqId: `${runId}-retry`,
|
||||
respond: secondRespond,
|
||||
});
|
||||
|
||||
expect(secondRespond).toHaveBeenCalledWith(
|
||||
false,
|
||||
{ runId, status: "error", summary },
|
||||
expect.objectContaining({ code: "UNAVAILABLE", message: summary }),
|
||||
{ cached: true },
|
||||
);
|
||||
expect(mocks.agentCommand).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("releases a foreground recovery owner if pre-dispatch reactivation fails", async () => {
|
||||
const sessionKey = "agent:main:main";
|
||||
const sessionId = "interrupted-session";
|
||||
|
||||
@@ -301,71 +301,10 @@ describe("gateway agent handler", () => {
|
||||
expect(callArgs.suppressPromptPersistence).toBe(true);
|
||||
expect(mocks.updateSessionStore).not.toHaveBeenCalled();
|
||||
expect(context.addChatRun).not.toHaveBeenCalled();
|
||||
const runContext = mockCallArg(mocks.registerAgentRunContext, 0, 1) as {
|
||||
attribution: Record<string, unknown>;
|
||||
};
|
||||
expect(runContext).toEqual({
|
||||
attribution: expect.objectContaining({
|
||||
runId: "test-backend-internal-effects",
|
||||
contextId: expect.any(String),
|
||||
executionId: expect.any(String),
|
||||
createdAt: expect.any(Number),
|
||||
lifecycleGeneration: "test-generation",
|
||||
sessionKey: "agent:main:main",
|
||||
sessionId: "existing-session-id",
|
||||
agentId: "main",
|
||||
}),
|
||||
sessionKey: "agent:main:main",
|
||||
sessionId: "existing-session-id",
|
||||
agentId: "main",
|
||||
expect(mocks.registerAgentRunContext).toHaveBeenCalledWith("test-backend-internal-effects", {
|
||||
isControlUiVisible: false,
|
||||
lifecycleGeneration: "test-generation",
|
||||
});
|
||||
expect(Object.isFrozen(runContext.attribution)).toBe(true);
|
||||
});
|
||||
|
||||
it("preserves the admitted idempotency key exactly in execution attribution", async () => {
|
||||
primeMainAgentRun({ cfg: mocks.loadConfigReturn });
|
||||
mocks.registerAgentRunContext.mockClear();
|
||||
const runId = " padded-agent-run ";
|
||||
|
||||
await invokeAgent({
|
||||
message: "preserve exact run identity",
|
||||
agentId: "main",
|
||||
sessionKey: "agent:main:main",
|
||||
idempotencyKey: runId,
|
||||
});
|
||||
|
||||
await waitForAgentCommandCall();
|
||||
expect(mockCallArg(mocks.registerAgentRunContext, 0, 0)).toBe(runId);
|
||||
expect(mockCallArg(mocks.registerAgentRunContext, 0, 1)).toMatchObject({
|
||||
attribution: { runId },
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects blank idempotency keys before registering run state", async () => {
|
||||
const context = makeContext();
|
||||
const respond = vi.fn();
|
||||
mocks.registerAgentRunContext.mockClear();
|
||||
mocks.agentCommand.mockClear();
|
||||
|
||||
await invokeAgent(
|
||||
{
|
||||
message: "reject blank run identity",
|
||||
agentId: "main",
|
||||
sessionKey: "agent:main:main",
|
||||
idempotencyKey: " \t ",
|
||||
},
|
||||
{ context, respond },
|
||||
);
|
||||
|
||||
expectRespondError(respond, {
|
||||
code: ErrorCodes.INVALID_REQUEST,
|
||||
message: "idempotencyKey must not be blank",
|
||||
});
|
||||
expect(context.chatAbortControllers.size).toBe(0);
|
||||
expect(mocks.registerAgentRunContext).not.toHaveBeenCalled();
|
||||
expect(mocks.agentCommand).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("allows backend internal runs without a persisted session row", async () => {
|
||||
@@ -663,27 +602,10 @@ describe("gateway agent handler", () => {
|
||||
expect(context.broadcastToConnIds).not.toHaveBeenCalled();
|
||||
expect(mocks.getLatestSubagentRunByChildSessionKey).not.toHaveBeenCalled();
|
||||
expect(mocks.replaceSubagentRunAfterSteer).not.toHaveBeenCalled();
|
||||
const runContext = mockCallArg(mocks.registerAgentRunContext, 0, 1) as {
|
||||
attribution: Record<string, unknown>;
|
||||
};
|
||||
expect(runContext).toEqual({
|
||||
attribution: expect.objectContaining({
|
||||
runId: "test-stateless-model-run",
|
||||
contextId: expect.any(String),
|
||||
executionId: expect.any(String),
|
||||
createdAt: expect.any(Number),
|
||||
lifecycleGeneration: "test-generation",
|
||||
sessionKey: "agent:main:explicit:model-run-123e4567-e89b-12d3-a456-426614174000",
|
||||
sessionId: "model-run-123e4567-e89b-12d3-a456-426614174000",
|
||||
agentId: "main",
|
||||
}),
|
||||
sessionKey: "agent:main:explicit:model-run-123e4567-e89b-12d3-a456-426614174000",
|
||||
sessionId: "model-run-123e4567-e89b-12d3-a456-426614174000",
|
||||
agentId: "main",
|
||||
expect(mocks.registerAgentRunContext).toHaveBeenCalledWith("test-stateless-model-run", {
|
||||
isControlUiVisible: false,
|
||||
lifecycleGeneration: "test-generation",
|
||||
});
|
||||
expect(Object.isFrozen(runContext.attribution)).toBe(true);
|
||||
});
|
||||
|
||||
it("respects explicit bestEffortDeliver=false for main session runs", async () => {
|
||||
|
||||
@@ -136,6 +136,7 @@ describe("gateway agent handler", () => {
|
||||
payloads: [{ text: "ok" }],
|
||||
meta: { durationMs: 100 },
|
||||
});
|
||||
|
||||
await invokeAgent(
|
||||
{
|
||||
message: "forged exec followup",
|
||||
@@ -1265,6 +1266,7 @@ describe("gateway agent handler", () => {
|
||||
payloads: [{ text: "ok" }],
|
||||
meta: { durationMs: 100 },
|
||||
});
|
||||
|
||||
await invokeAgent(
|
||||
{
|
||||
message: "resume channel session",
|
||||
@@ -2069,7 +2071,6 @@ describe("gateway agent handler", () => {
|
||||
});
|
||||
|
||||
it("infers selected-global agent id from agent-prefixed session aliases", async () => {
|
||||
mocks.registerAgentRunContext.mockClear();
|
||||
mocks.listAgentIds.mockReturnValue(["main", "work"]);
|
||||
mocks.loadConfigReturn = {
|
||||
agents: { list: [{ id: "main", default: true }, { id: "work" }] },
|
||||
@@ -2114,24 +2115,6 @@ describe("gateway agent handler", () => {
|
||||
agentId: "work",
|
||||
clone: false,
|
||||
});
|
||||
expect(mockCallArg(mocks.registerAgentRunContext, 0, 1)).toEqual(
|
||||
expect.objectContaining({
|
||||
attribution: expect.objectContaining({
|
||||
runId: "alias-global-session-agent-id",
|
||||
contextId: expect.any(String),
|
||||
executionId: expect.any(String),
|
||||
createdAt: expect.any(Number),
|
||||
lifecycleGeneration: "test-generation",
|
||||
sessionKey: "global",
|
||||
sessionId: "global-work-session-id",
|
||||
agentId: "work",
|
||||
}),
|
||||
sessionKey: "global",
|
||||
sessionId: "global-work-session-id",
|
||||
agentId: "work",
|
||||
lifecycleGeneration: "test-generation",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("registers tool event recipients for active selected-global alias runs", async () => {
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
import { expect, test } from "vitest";
|
||||
import { emitAgentEvent, onAgentEvent } from "./agent-events.js";
|
||||
import { clearAgentRunContext, registerAgentRunContext } from "./agent-run-registry.js";
|
||||
|
||||
test("clearAgentRunContext also cleans up seqByRun to prevent memory leak (#63643)", () => {
|
||||
registerAgentRunContext("run-leak", { sessionKey: "main" });
|
||||
emitAgentEvent({ runId: "run-leak", stream: "lifecycle", data: {} });
|
||||
emitAgentEvent({ runId: "run-leak", stream: "lifecycle", data: {} });
|
||||
|
||||
clearAgentRunContext("run-leak");
|
||||
|
||||
const seqs: number[] = [];
|
||||
const stop = onAgentEvent((evt) => {
|
||||
if (evt.runId === "run-leak") {
|
||||
seqs.push(evt.seq);
|
||||
}
|
||||
});
|
||||
emitAgentEvent({ runId: "run-leak", stream: "lifecycle", data: {} });
|
||||
stop();
|
||||
|
||||
expect(seqs).toEqual([1]);
|
||||
});
|
||||
@@ -1,6 +1,5 @@
|
||||
// Covers agent event sequencing and run context cleanup.
|
||||
import { beforeEach, describe, expect, test, vi } from "vitest";
|
||||
import { createAgentExecutionAttribution } from "../agents/agent-execution-attribution.js";
|
||||
import {
|
||||
type AgentEventPayload,
|
||||
captureAgentRunLifecycleGeneration,
|
||||
@@ -898,45 +897,6 @@ describe("agent-events sequencing", () => {
|
||||
expect(context?.lastActiveAt).toBe(12_345);
|
||||
});
|
||||
|
||||
test("keeps the first same-generation attribution private and immutable", () => {
|
||||
const attribution = createAgentExecutionAttribution({
|
||||
runId: "run-ctx",
|
||||
lifecycleGeneration: getAgentEventLifecycleGeneration(),
|
||||
sessionKey: "agent:main:main",
|
||||
sessionId: "session-1",
|
||||
agentId: "main",
|
||||
});
|
||||
const replacement = createAgentExecutionAttribution({
|
||||
runId: "run-ctx",
|
||||
lifecycleGeneration: attribution.lifecycleGeneration,
|
||||
sessionKey: "agent:main:other",
|
||||
sessionId: "session-2",
|
||||
agentId: "main",
|
||||
});
|
||||
registerAgentRunContext("run-ctx", {
|
||||
attribution,
|
||||
lifecycleGeneration: attribution.lifecycleGeneration,
|
||||
});
|
||||
registerAgentRunContext("run-ctx", {
|
||||
attribution: replacement,
|
||||
lifecycleGeneration: attribution.lifecycleGeneration,
|
||||
verboseLevel: "full",
|
||||
});
|
||||
|
||||
expect(getAgentRunContext("run-ctx")?.attribution).toBe(attribution);
|
||||
expect(getAgentRunContext("run-ctx")?.verboseLevel).toBe("full");
|
||||
expect(Reflect.set(getAgentRunContext("run-ctx")!, "attribution", replacement)).toBe(false);
|
||||
|
||||
let received: AgentEventPayload | undefined;
|
||||
const stop = onAgentEvent((event) => {
|
||||
received = event;
|
||||
});
|
||||
emitAgentEvent({ runId: "run-ctx", stream: "lifecycle", data: { phase: "end" } });
|
||||
stop();
|
||||
|
||||
expect(JSON.stringify(received)).not.toContain("attribution");
|
||||
});
|
||||
|
||||
test("falls back to registered sessionKey when event sessionKey is blank", () => {
|
||||
registerAgentRunContext("run-ctx", { sessionKey: "session-main" });
|
||||
|
||||
@@ -1157,3 +1117,27 @@ describe("agent-events sequencing", () => {
|
||||
clock.mockRestore();
|
||||
});
|
||||
});
|
||||
|
||||
test("clearAgentRunContext also cleans up seqByRun to prevent memory leak (#63643)", () => {
|
||||
// Regression test: seqByRun entries were never deleted when a run ended,
|
||||
// causing unbounded growth over time.
|
||||
registerAgentRunContext("run-leak", { sessionKey: "main" });
|
||||
emitAgentEvent({ runId: "run-leak", stream: "lifecycle", data: {} });
|
||||
emitAgentEvent({ runId: "run-leak", stream: "lifecycle", data: {} });
|
||||
|
||||
// After clearing run context, the sequence counter should also be removed.
|
||||
clearAgentRunContext("run-leak");
|
||||
|
||||
// Emitting a new event on the same runId should start seq from 1 again,
|
||||
// proving the old entry was deleted.
|
||||
const seqs: number[] = [];
|
||||
const stop = onAgentEvent((evt) => {
|
||||
if (evt.runId === "run-leak") {
|
||||
seqs.push(evt.seq);
|
||||
}
|
||||
});
|
||||
emitAgentEvent({ runId: "run-leak", stream: "lifecycle", data: {} });
|
||||
stop();
|
||||
|
||||
expect(seqs).toEqual([1]);
|
||||
});
|
||||
|
||||
@@ -1,14 +1,11 @@
|
||||
// Owns process-local agent run context, ownership, and projection state.
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { AgentExecutionAttribution } from "../agents/agent-execution-attribution.js";
|
||||
import type { VerboseLevel } from "../auto-reply/thinking.js";
|
||||
import { resolveGlobalSingleton } from "../shared/global-singleton.js";
|
||||
import { clearAgentRunUsage, resetAgentRunUsageForTest } from "./agent-run-usage.js";
|
||||
|
||||
/** Per-run metadata used to stamp events and gate Control UI visibility. */
|
||||
type AgentRunContext = {
|
||||
/** Immutable admission-owned correlation; later context merges cannot replace it. */
|
||||
readonly attribution?: AgentExecutionAttribution;
|
||||
sessionKey?: string;
|
||||
/** Resolved agent owner, including for unscoped session keys. */
|
||||
agentId?: string;
|
||||
@@ -66,35 +63,6 @@ function bumpAgentRunIndexVersion(): void {
|
||||
getAgentRunRegistryState().version += 1;
|
||||
}
|
||||
|
||||
function attachAgentExecutionAttribution(
|
||||
context: AgentRunContext,
|
||||
attribution: AgentExecutionAttribution | undefined,
|
||||
): void {
|
||||
if (!attribution || context.attribution) {
|
||||
return;
|
||||
}
|
||||
Object.defineProperty(context, "attribution", {
|
||||
value: attribution,
|
||||
enumerable: true,
|
||||
configurable: false,
|
||||
writable: false,
|
||||
});
|
||||
}
|
||||
|
||||
function createAgentRunContext(
|
||||
context: AgentRunContext,
|
||||
lifecycleGeneration: string,
|
||||
): AgentRunContext {
|
||||
const { attribution, ...fields } = context;
|
||||
const stored: AgentRunContext = {
|
||||
...fields,
|
||||
lifecycleGeneration,
|
||||
registeredAt: context.registeredAt ?? Date.now(),
|
||||
};
|
||||
attachAgentExecutionAttribution(stored, attribution);
|
||||
return stored;
|
||||
}
|
||||
|
||||
/** Reads the process-local version of the active-run projection inputs. */
|
||||
export function readAgentRunIndexVersion(): number {
|
||||
return getAgentRunRegistryState().version;
|
||||
@@ -137,7 +105,11 @@ export function registerAgentRunContext(
|
||||
}
|
||||
const existing = state.contexts.get(runId);
|
||||
if (!existing) {
|
||||
state.contexts.set(runId, createAgentRunContext(context, lifecycleGeneration));
|
||||
state.contexts.set(runId, {
|
||||
...context,
|
||||
lifecycleGeneration,
|
||||
registeredAt: context.registeredAt ?? Date.now(),
|
||||
});
|
||||
bumpAgentRunIndexVersion();
|
||||
return;
|
||||
}
|
||||
@@ -148,7 +120,6 @@ export function registerAgentRunContext(
|
||||
) {
|
||||
return;
|
||||
}
|
||||
attachAgentExecutionAttribution(existing, context.attribution);
|
||||
let runIndexChanged = false;
|
||||
if (context.sessionKey && existing.sessionKey !== context.sessionKey) {
|
||||
existing.sessionKey = context.sessionKey;
|
||||
@@ -270,7 +241,11 @@ export function claimAgentRunContext(
|
||||
}
|
||||
return claimId;
|
||||
}
|
||||
state.contexts.set(runId, createAgentRunContext(context, lifecycleGeneration));
|
||||
state.contexts.set(runId, {
|
||||
...context,
|
||||
lifecycleGeneration,
|
||||
registeredAt: context.registeredAt ?? Date.now(),
|
||||
});
|
||||
state.sequenceResetHandler?.(runId);
|
||||
clearAgentRunUsage(runId);
|
||||
bumpAgentRunIndexVersion();
|
||||
|
||||
@@ -7,13 +7,13 @@ const optionalRunIdCaller: PublicIngressOptions = {
|
||||
sessionKey: "agent:main:plugin-session",
|
||||
allowModelOverride: false,
|
||||
};
|
||||
const privateExecutionCorrelationIsHidden: "executionAttribution" extends keyof PublicIngressOptions
|
||||
const privateRecoveryCorrelationIsHidden: "executionIdentityAdmission" extends keyof PublicIngressOptions
|
||||
? false
|
||||
: true = true;
|
||||
|
||||
describe("public agent ingress correlation contract", () => {
|
||||
it("keeps runId optional and private execution recovery state unavailable", () => {
|
||||
expect(optionalRunIdCaller).not.toHaveProperty("runId");
|
||||
expect(privateExecutionCorrelationIsHidden).toBe(true);
|
||||
expect(privateRecoveryCorrelationIsHidden).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user