ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-26 04:15:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(webhooks): keep TaskFlow child actions within the owning session (#129633)

Browse Source 
* fix(webhooks): bind TaskFlow child actions to owning runs

* fix(tasks): preserve task flow import boundaries

* test(tasks): cover exact-run cancellation callers

* test(webhooks): prove ACP replacement cancellation fence

* fix(tasks): preserve authority across lifecycle races

* fix(tasks): bind TaskFlow children to runtime instances

* fix(acp): revalidate queued cancellation authority
This commit is contained in:
Josh Avant
2026-08-25 23:13:50 -07:00
committed by GitHub
parent 96de6f405b
commit eedc554f6e
29 changed files with 1846 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plugins/webhooks.md
+5
View File
@@ -142,10 +142,15 @@ with any other status returns `400 invalid_request`.
"flowId": "flow_123",
"runtime": "acp",
"childSessionKey": "agent:main:acp:worker",
"runId": "run_123",
"task": "Inspect the next message batch"
}
```
`childSessionKey` identifies the backing run but does not grant authority over it. For automatic
lifecycle tracking and cancellation, include the exact `runId`; the backing task must be owned by
the route's configured session. Foreign, stale, or replaced runs are rejected at use time.
## Response shape
```json
extensions/acpx/test/fixtures/codex-app-server.mjs
Vendored
+9
View File
@@ -1,8 +1,16 @@
#!/usr/bin/env node
import fs from "node:fs";
import readline from "node:readline";
let nextRequestId = 1;
const pending = new Map();
const tracePath = process.env.OPENCLAW_ACPX_PROCESS_FIXTURE_TRACE;
function trace(method) {
if (tracePath) {
fs.appendFileSync(tracePath, `${JSON.stringify({ method })}\n`, "utf8");
}
}
function write(message) {
process.stdout.write(`${JSON.stringify(message)}\n`);
@@ -33,6 +41,7 @@ const model = {
};
async function handle(method, params) {
trace(method);
if (method === "initialize") {
return { userAgent: "openclaw-acpx-process-fixture", codexHome: process.cwd() };
}
extensions/webhooks/src/http.test.ts
+1 -4
View File
@@ -235,7 +235,6 @@ describe("createTaskFlowWebhookRequestHandler", () => {
action: "run_task",
flowId: flow.flowId,
runtime: "acp",
childSessionKey: "agent:main:subagent:child",
task: "Inspect the next message batch",
status: "running",
startedAt: 10,
@@ -248,7 +247,7 @@ describe("createTaskFlowWebhookRequestHandler", () => {
expect(parsed.ok).toBe(true);
expect(parsed.result.created).toBe(true);
expect(parsed.result.task.parentFlowId).toBe(flow.flowId);
expect(parsed.result.task.childSessionKey).toBe("agent:main:subagent:child");
expect(parsed.result.task.childSessionKey).toBeUndefined();
expect(parsed.result.task.runtime).toBe("acp");
expect(parsed.result.task.ownerKey).toBeUndefined();
expect(parsed.result.task.requesterSessionKey).toBeUndefined();
@@ -362,7 +361,6 @@ describe("createTaskFlowWebhookRequestHandler", () => {
action: "run_task",
flowId: flow.flowId,
runtime: "acp",
childSessionKey: "agent:main:subagent:child",
runId: "retry-me",
task: "Inspect the next message batch",
},
@@ -375,7 +373,6 @@ describe("createTaskFlowWebhookRequestHandler", () => {
action: "run_task",
flowId: flow.flowId,
runtime: "acp",
childSessionKey: "agent:main:subagent:child",
runId: "retry-me",
task: "Inspect the next message batch",
},
src/acp/control-plane/manager.background-task.test.ts
+2
View File
@@ -130,6 +130,7 @@ describe("ACP background task execution binding", () => {
task: "private",
},
100,
admitted.operationalRunInstance.instanceId,
);
const task = findTaskByRunId("run-acp");
if (!record || !task?.parentFlowId) {
@@ -189,6 +190,7 @@ describe("ACP background task execution binding", () => {
task: "private",
},
100,
admitted.operationalRunInstance.instanceId,
);
const task = findTaskByRunId("run-acp");
if (!record || !task?.parentFlowId) {
src/acp/control-plane/manager.background-task.ts
+6
View File
@@ -10,6 +10,7 @@ import {
failTaskRunByRunId,
startTaskRunByRunId,
} from "../../tasks/detached-task-runtime.js";
import { createNextAcpTaskBackingDetail } from "../../tasks/task-backing-authority.js";
import { resolveRequiredCompletionTerminalResult } from "../../tasks/task-completion-contract.js";
import { bindTaskFlowExecution } from "../../tasks/task-flow-registry.store.sqlite.js";
import { bindTaskRunExecution } from "../../tasks/task-registry.store.sqlite.js";
@@ -142,6 +143,7 @@ export function resolveBackgroundTaskContext(params: {
export function createBackgroundTaskRecord(
context: BackgroundTaskContext,
startedAt: number,
instanceId: string,
): BackgroundTaskRecord | undefined {
try {
const task = createRunningTaskRun({
@@ -155,6 +157,10 @@ export function createBackgroundTaskRecord(
label: context.label,
task: context.task,
startedAt,
detail: createNextAcpTaskBackingDetail({
childSessionKey: context.childSessionKey,
instanceId,
}),
});
if (!task) {
logVerbose(
src/acp/control-plane/manager.cancel-session.test.ts
+153
View File
@@ -1,5 +1,6 @@
/** Tests ACP manager cancellation of active turns and idle sessions. */
import type { AcpRuntimeEvent } from "@openclaw/acp-core/runtime/types";
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
import { describe, expect, it, vi } from "vitest";
import {
requireTaskByRunId,
@@ -64,11 +65,17 @@ describe("AcpSessionManager cancelSession", () => {
},
{ interval: 1 },
);
const taskDetail = asOptionalRecord(requireTaskByRunId("run-1").detail);
const instanceId = typeof taskDetail?.instanceId === "string" ? taskDetail.instanceId : "";
expect(instanceId).not.toBe("");
await manager.cancelSession({
cfg: baseCfg,
sessionKey: "agent:codex:acp:child-1",
reason: "manual-cancel",
expectedRunId: "run-1",
expectedInstanceId: instanceId,
expectedOwnerKey: "agent:main:main",
});
await runPromise;
@@ -92,4 +99,150 @@ describe("AcpSessionManager cancelSession", () => {
expect(states).not.toContain("error");
});
});
it("keeps a queued same-id successor outside the active-turn cancellation", async () => {
await withAcpManagerTaskStateDir(async () => {
const runtimeState = createRuntime();
hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
id: "acpx",
runtime: runtimeState.runtime,
});
mockParentedAcpSessionEntries({
childSessionKey: "agent:codex:acp:child-1",
parentSessionKey: "agent:main:main",
});
let runCount = 0;
let firstEntered = false;
let secondEntered = false;
let secondSignal: AbortSignal | undefined;
let releaseSecond: (() => void) | undefined;
runtimeState.runTurn.mockImplementation(async function* (input: { signal?: AbortSignal }) {
runCount += 1;
if (runCount === 1) {
firstEntered = true;
await new Promise<void>((resolve) => {
if (input.signal?.aborted) {
resolve();
return;
}
input.signal?.addEventListener("abort", () => resolve(), { once: true });
});
yield { type: "done" as const, stopReason: "cancel" };
return;
}
secondEntered = true;
secondSignal = input.signal;
await new Promise<void>((resolve) => {
releaseSecond = resolve;
});
yield { type: "done" as const, stopReason: "end_turn" };
});
const manager = new AcpSessionManager();
const firstRun = manager.runTurn({
provenance: "system",
cfg: baseCfg,
sessionKey: "agent:codex:acp:child-1",
text: "first task",
mode: "prompt",
requestId: "run-shared",
});
await vi.waitFor(() => expect(firstEntered).toBe(true), { interval: 1 });
const taskDetail = asOptionalRecord(requireTaskByRunId("run-shared").detail);
const instanceId = typeof taskDetail?.instanceId === "string" ? taskDetail.instanceId : "";
expect(instanceId).not.toBe("");
const successorRun = manager.runTurn({
provenance: "system",
cfg: baseCfg,
sessionKey: "agent:codex:acp:child-1",
text: "queued successor",
mode: "prompt",
requestId: "run-shared",
});
await Promise.resolve();
expect(secondEntered).toBe(false);
await manager.cancelSession({
cfg: baseCfg,
sessionKey: "agent:codex:acp:child-1",
reason: "cancel-first",
expectedRunId: "run-shared",
expectedInstanceId: instanceId,
expectedOwnerKey: "agent:main:main",
});
await firstRun;
await vi.waitFor(() => expect(secondEntered).toBe(true), { interval: 1 });
expect(runtimeState.cancel).toHaveBeenCalledTimes(1);
expect(secondSignal?.aborted).toBe(false);
releaseSecond?.();
await successorRun;
expect(runtimeState.cancel).toHaveBeenCalledTimes(1);
});
});
it("does not cancel a replacement active turn", async () => {
await withAcpManagerTaskStateDir(async () => {
const runtimeState = createRuntime();
hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
id: "acpx",
runtime: runtimeState.runtime,
});
mockParentedAcpSessionEntries({
childSessionKey: "agent:codex:acp:child-1",
parentSessionKey: "agent:main:main",
});
let enteredRun = false;
let releaseRun: (() => void) | undefined;
runtimeState.runTurn.mockImplementation(async function* () {
enteredRun = true;
await new Promise<void>((resolve) => {
releaseRun = resolve;
});
yield { type: "done" as const, stopReason: "end_turn" };
});
const manager = new AcpSessionManager();
const runPromise = manager.runTurn({
provenance: "system",
cfg: baseCfg,
sessionKey: "agent:codex:acp:child-1",
text: "replacement task",
mode: "prompt",
requestId: "run-current",
});
await vi.waitFor(() => expect(enteredRun).toBe(true), { interval: 1 });
const taskDetail = asOptionalRecord(requireTaskByRunId("run-current").detail);
const instanceId = typeof taskDetail?.instanceId === "string" ? taskDetail.instanceId : "";
expect(instanceId).not.toBe("");
await expect(
manager.cancelSession({
cfg: baseCfg,
sessionKey: "agent:codex:acp:child-1",
reason: "foreign-owner-cancel",
expectedRunId: "run-current",
expectedInstanceId: instanceId,
expectedOwnerKey: "agent:main:other",
}),
).rejects.toThrow("ACP task owner could not be verified.");
expect(runtimeState.cancel).not.toHaveBeenCalled();
await expect(
manager.cancelSession({
cfg: baseCfg,
sessionKey: "agent:codex:acp:child-1",
reason: "stale-task-cancel",
expectedRunId: "run-current",
expectedInstanceId: "instance-from-prior-turn",
expectedOwnerKey: "agent:main:main",
}),
).rejects.toThrow("ACP task is no longer the active run.");
expect(runtimeState.cancel).not.toHaveBeenCalled();
releaseRun?.();
await runPromise;
});
});
});
src/acp/control-plane/manager.cancel-session.ts
+38 -1
View File
@@ -2,7 +2,7 @@
import type { AcpRuntime, AcpRuntimeHandle } from "@openclaw/acp-core/runtime/types";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
type AcpRuntimeError,
AcpRuntimeError,
toAcpRuntimeError,
withAcpRuntimeErrorBoundary,
} from "../runtime/errors.js";
@@ -20,6 +20,9 @@ export async function runManagerCancelSession(params: {
cfg: OpenClawConfig;
sessionKey: string;
reason?: string;
expectedRunId?: string;
expectedInstanceId?: string;
expectedOwnerKey?: string;
activeTurnBySession: Map<string, ActiveTurnState>;
withSessionActor: WithManagerSessionActor;
resolveSession: ResolveManagerSession;
@@ -28,15 +31,47 @@ export async function runManagerCancelSession(params: {
}): Promise<void> {
const actorKey = normalizeActorKey(params.sessionKey);
const activeTurn = params.activeTurnBySession.get(actorKey);
const expectedRunId = params.expectedRunId?.trim();
const expectedInstanceId = params.expectedInstanceId?.trim();
const expectedOwnerKey = params.expectedOwnerKey?.trim();
const requireExpectedTurn = (current: ActiveTurnState | undefined) => {
if (
(expectedRunId && current?.requestId !== expectedRunId) ||
(expectedInstanceId && current?.instanceId !== expectedInstanceId)
) {
throw new AcpRuntimeError("ACP_TURN_FAILED", "ACP task is no longer the active run.");
}
return current;
};
const requireExpectedOwner = () => {
if (!expectedOwnerKey) {
return;
}
const resolution = params.resolveSession({ cfg: params.cfg, sessionKey: params.sessionKey });
const entry = resolution.kind === "ready" ? resolution.entry : undefined;
const ownerKey = entry?.spawnedBy?.trim() || entry?.parentSessionKey?.trim();
if (ownerKey !== expectedOwnerKey) {
throw new AcpRuntimeError("ACP_TURN_FAILED", "ACP task owner could not be verified.");
}
};
requireExpectedTurn(activeTurn);
if (activeTurn) {
await cancelManagerActiveTurn({
activeTurn,
reason: params.reason,
revalidate: () => {
requireExpectedTurn(params.activeTurnBySession.get(actorKey));
requireExpectedOwner();
},
});
return;
}
await params.withSessionActor(params.sessionKey, async () => {
// The actor wait may admit queued work. Recheck exact authority only after
// that wait, immediately before the idle-handle cancellation boundary.
requireExpectedTurn(params.activeTurnBySession.get(actorKey));
requireExpectedOwner();
const resolution = params.resolveSession({
cfg: params.cfg,
sessionKey: params.sessionKey,
@@ -76,7 +111,9 @@ export async function runManagerCancelSession(params: {
export async function cancelManagerActiveTurn(params: {
activeTurn: ActiveTurnState;
reason?: string;
revalidate?: () => void;
}): Promise<void> {
params.revalidate?.();
params.activeTurn.abortController.abort();
if (!params.activeTurn.cancelPromise) {
params.activeTurn.cancelPromise = params.activeTurn.runtime.cancel({
src/acp/control-plane/manager.core.ts
+6
View File
@@ -340,6 +340,9 @@ export class AcpSessionManager {
cfg: OpenClawConfig;
sessionKey: string;
reason?: string;
expectedRunId?: string;
expectedInstanceId?: string;
expectedOwnerKey?: string;
}): Promise<void> {
const sessionKey = canonicalizeAcpSessionKey(params);
if (!sessionKey) {
@@ -350,6 +353,9 @@ export class AcpSessionManager {
cfg: params.cfg,
sessionKey,
reason: params.reason,
expectedRunId: params.expectedRunId,
expectedInstanceId: params.expectedInstanceId,
expectedOwnerKey: params.expectedOwnerKey,
activeTurnBySession: this.activeTurnBySession,
withSessionActor: this.withSessionActor.bind(this),
resolveSession: this.resolveSession.bind(this),
src/acp/control-plane/manager.turn-runner.ts
+7 -1
View File
@@ -89,7 +89,11 @@ export async function runManagerTurn(params: {
})
: null;
const taskRecord = taskContext
? createBackgroundTaskRecord(taskContext, turnStartedAt)
? createBackgroundTaskRecord(
taskContext,
turnStartedAt,
input.admittedRunContext.operationalRunInstance.instanceId,
)
: undefined;
let taskExecutionBound = false;
let taskProgressSummary = "";
@@ -243,6 +247,8 @@ export async function runManagerTurn(params: {
}
activeTurn = {
requestId: input.requestId,
instanceId: input.admittedRunContext.operationalRunInstance.instanceId,
runtime,
handle,
abortController: internalAbortController,
src/acp/control-plane/manager.types.ts
+2
View File
@@ -140,6 +140,8 @@ export type AcpStartupIdentityReconcileResult = {
};
export type ActiveTurnState = {
requestId: string;
instanceId: string;
runtime: AcpRuntime;
handle: AcpRuntimeHandle;
abortController: AbortController;
src/agents/subagents/registry/subagent-control-kill-runtime.ts
+13
View File
@@ -439,12 +439,20 @@ export async function killLatestSubagentRun(params: {
entry: SubagentRunRecord;
cache: Map<string, Record<string, SessionEntry>>;
suppressTaskDelivery?: boolean;
expectedRunId?: string;
expectedGeneration?: number;
}): Promise<{
entry: SubagentRunRecord;
result: Awaited<ReturnType<typeof killSubagentRun>>;
}> {
let entry = params.entry;
for (let attempt = 0; attempt < 3; attempt += 1) {
if (params.expectedGeneration !== undefined && entry.generation !== params.expectedGeneration) {
return {
entry,
result: { killed: false, superseded: true },
};
}
const result = await killSubagentRun({ ...params, entry });
if (!result.superseded) {
return { entry, result };
@@ -453,6 +461,11 @@ export async function killLatestSubagentRun(params: {
if (!latest || latest === entry) {
return { entry, result };
}
// A task-scoped kill is bound to its exact run. Recovery may replace that
// run while cancellation awaits lifecycle admission, but cannot inherit its authority.
if (params.expectedRunId) {
return { entry, result };
}
if (entry.execution.restartRecovery?.idempotencyKey !== latest.runId) {
return { entry, result };
}
src/agents/subagents/registry/subagent-control-kill.ts
+17 -2
View File
@@ -273,14 +273,27 @@ export async function killSubagentRunAdmin(params: {
cfg: OpenClawConfig;
sessionKey: string;
agentId?: string;
expectedRunId?: string;
expectedGeneration?: number;
expectedOwnerKey?: string;
}) {
const targetSessionKey = params.sessionKey.trim();
if (!targetSessionKey) {
return { found: false as const, killed: false };
return { found: false as const, killed: false as const };
}
const entry = getLatestOwnedSubagentRun(targetSessionKey, params.agentId, params.cfg);
if (!entry) {
return { found: false as const, killed: false };
return { found: false as const, killed: false as const };
}
if (params.expectedRunId?.trim() && entry.runId !== params.expectedRunId.trim()) {
return { found: false as const, killed: false as const };
}
if (
(params.expectedGeneration !== undefined && entry.generation !== params.expectedGeneration) ||
(params.expectedOwnerKey?.trim() &&
entry.requesterSessionKey !== params.expectedOwnerKey.trim())
) {
return { found: false as const, killed: false as const };
}
const killCache = new Map<string, Record<string, SessionEntry>>();
@@ -288,6 +301,8 @@ export async function killSubagentRunAdmin(params: {
cfg: params.cfg,
entry,
cache: killCache,
expectedRunId: params.expectedRunId?.trim() || undefined,
expectedGeneration: params.expectedGeneration,
});
const stopResult = stopped.result;
if (stopResult.error) {
src/agents/subagents/registry/subagent-control.test.ts
+198
View File
@@ -696,6 +696,204 @@ describe("killSubagentRunAdmin", () => {
expect(result).toEqual({ found: false, killed: false });
});
it("does not kill a replacement run when an exact run id is required", async () => {
const childSessionKey = "agent:main:subagent:replacement";
addSubagentRunForTests({
runId: "run-current",
childSessionKey,
controllerSessionKey: "agent:main:main",
requesterSessionKey: "agent:main:main",
requesterDisplayKey: "main",
task: "replacement work",
cleanup: "keep",
createdAt: Date.now() - 1_000,
startedAt: Date.now() - 900,
});
const result = await killSubagentRunAdmin({
cfg: cfgWithSessionStore(),
sessionKey: childSessionKey,
expectedRunId: "run-stale",
});
expect(result).toEqual({ found: false, killed: false });
expect(getSubagentRunByChildSessionKey(childSessionKey)?.execution.endedAt).toBeUndefined();
});
it("does not kill a same-id replacement generation", async () => {
const childSessionKey = "agent:main:subagent:same-id-replacement";
addSubagentRunForTests({
runId: "run-reused",
childSessionKey,
controllerSessionKey: "agent:main:main",
requesterSessionKey: "agent:main:main",
requesterDisplayKey: "main",
task: "replacement work",
cleanup: "keep",
generation: 2,
createdAt: Date.now() - 1_000,
startedAt: Date.now() - 900,
});
const result = await killSubagentRunAdmin({
cfg: cfgWithSessionStore(),
sessionKey: childSessionKey,
expectedRunId: "run-reused",
expectedGeneration: 1,
expectedOwnerKey: "agent:main:main",
});
const foreignOwner = await killSubagentRunAdmin({
cfg: cfgWithSessionStore(),
sessionKey: childSessionKey,
expectedRunId: "run-reused",
expectedGeneration: 2,
expectedOwnerKey: "agent:main:other",
});
expect(result).toEqual({ found: false, killed: false });
expect(foreignOwner).toEqual({ found: false, killed: false });
expect(getSubagentRunByChildSessionKey(childSessionKey)?.execution.endedAt).toBeUndefined();
});
it("does not adopt a restart-recovery successor when an exact run id is required", async () => {
const childSessionKey = "agent:main:subagent:fenced-recovery-successor";
const sessionId = "sess-fenced-recovery-successor";
const recoveryRunId = "run-fenced-recovery-successor";
const receipt = {
sessionId,
sessionMarker: `${sessionId}:1`,
idempotencyKey: recoveryRunId,
phase: "accepted" as const,
};
const source = createSubagentRunRecord({
runId: "run-fenced-recovery-source",
childSessionKey,
controllerSessionKey: "agent:main:controller",
requesterSessionKey: "agent:main:requester",
requesterDisplayKey: "requester",
task: "source recovery task",
cleanup: "keep",
generation: 1,
createdAt: Date.now() - 2_000,
execution: {
status: "interrupted",
startedAt: Date.now() - 1_000,
restartRecovery: receipt,
},
});
addSubagentRunForTests(source);
const storePath = await writeSessionStoreFixture("fenced-recovery-successor", {
[childSessionKey]: { sessionId, updatedAt: Date.now(), abortedLastRun: true },
});
const admission = await beginSessionWorkAdmission({
scope: storePath,
identities: [childSessionKey, sessionId],
assertAllowed: () => {},
});
const handoffId = admission.createHandoff();
const abort = vi.fn(() => true);
setSubagentControlDepsForTest({
isEmbeddedAgentRunActive: () => true,
abortEmbeddedAgentRun: abort,
clearSessionQueues: () => ({ followupCleared: 0, laneCleared: 0, keys: [] }),
});
const pendingKill = killSubagentRunAdmin({
cfg: cfgWithSessionStore(storePath),
sessionKey: childSessionKey,
expectedRunId: source.runId,
});
await vi.waitFor(() => expect(getActiveSessionLifecycleMutationCount()).toBeGreaterThan(0));
const adopted = consumeSessionWorkAdmissionHandoff({
handoffId,
scope: storePath,
identities: [childSessionKey, sessionId],
onInterrupt: () => undefined,
});
expect(
replaceSubagentRunAfterSteerCore({
previousRunId: source.runId,
nextRunId: recoveryRunId,
expected: source,
restartRecovery: receipt,
persistenceFailure: "return-false",
}),
).toBe(true);
expect(adopted).toBeDefined();
adopted?.release();
await expect(pendingKill).resolves.toMatchObject({
found: true,
killed: false,
runId: source.runId,
});
expect(getSubagentRunByChildSessionKey(childSessionKey)).toMatchObject({
runId: recoveryRunId,
execution: { status: "running" },
});
expect(getSubagentRunByChildSessionKey(childSessionKey)?.execution.endedAt).toBeUndefined();
expect(abort).not.toHaveBeenCalled();
});
it("does not adopt a same-id successor when an exact run id is required", async () => {
const childSessionKey = "agent:main:subagent:fenced-same-id-successor";
const runId = "run-fenced-same-id-successor";
const source = createSubagentRunRecord({
runId,
childSessionKey,
controllerSessionKey: "agent:main:controller",
requesterSessionKey: "agent:main:requester",
requesterDisplayKey: "requester",
task: "same-id recovery source",
cleanup: "keep",
generation: 1,
createdAt: Date.now() - 2_000,
execution: {
status: "interrupted",
startedAt: Date.now() - 1_000,
restartRecovery: {
sessionId: "sess-fenced-same-id-successor",
sessionMarker: "sess-fenced-same-id-successor:1",
idempotencyKey: runId,
phase: "accepted",
},
},
});
addSubagentRunForTests(source);
const abort = vi.fn(() => false);
setSubagentControlDepsForTest({
isEmbeddedAgentRunActive: () => false,
abortEmbeddedAgentRun: abort,
clearSessionQueues: () => ({ followupCleared: 0, laneCleared: 0, keys: [] }),
});
const pendingKill = killSubagentRunAdmin({
cfg: cfgWithSessionStore(),
sessionKey: childSessionKey,
expectedRunId: runId,
});
addSubagentRunForTests({
...source,
task: "same-id recovery successor",
generation: 2,
createdAt: Date.now(),
execution: { status: "running", startedAt: Date.now() },
});
await expect(pendingKill).resolves.toMatchObject({
found: true,
killed: false,
runId,
});
expect(getSubagentRunByChildSessionKey(childSessionKey)).toMatchObject({
runId,
generation: 2,
execution: { status: "running" },
});
expect(getSubagentRunByChildSessionKey(childSessionKey)?.execution.endedAt).toBeUndefined();
expect(abort).not.toHaveBeenCalled();
});
it("retries task reconciliation for an already-killed run", async () => {
const childSessionKey = "agent:main:subagent:already-killed";
const endedAt = Date.now() - 1_000;
src/agents/subagents/registry/subagent-registry-run-launch.ts
+2
View File
@@ -12,6 +12,7 @@ import {
finalizeTaskRunByRunId,
startTaskRunByRunId,
} from "../../../tasks/detached-task-runtime.js";
import { createSubagentTaskBackingDetail } from "../../../tasks/task-backing-authority.js";
import { normalizeDeliveryContext } from "../../../utils/delivery-context.shared.js";
import type { DeliveryContext } from "../../../utils/delivery-context.types.js";
import { resolveSubagentRequesterAgentId } from "../../subagent-requester-owner.js";
@@ -236,6 +237,7 @@ export class SubagentLaunchManager extends SubagentRecoveryManager {
requesterAgentId: resolveSubagentRequesterAgentId(cfg, registerParams),
deliveryStatus:
registerParams.expectsCompletionMessage === false ? "not_applicable" : "pending",
detail: createSubagentTaskBackingDetail(generation),
} as const;
const task = queued
? createQueuedTaskRun(taskParams)
src/agents/subagents/registry/subagent-registry-run-recovery.ts
+25
View File
@@ -10,6 +10,8 @@ import {
getGatewayContextResolver,
} from "../../../plugins/runtime/gateway-request-scope.js";
import { finalizeTaskRunByRunId } from "../../../tasks/detached-task-runtime.js";
import { setCanonicalTaskBackingDetail } from "../../../tasks/task-backing-authority-write.js";
import { createSubagentTaskBackingDetail } from "../../../tasks/task-backing-authority.js";
import { removeInternalSessionEffectsSession } from "../../internal-session-effects.js";
import type { AgentRunSessionTarget } from "../../run-session-target.js";
import {
@@ -292,6 +294,29 @@ export class SubagentRecoveryManager extends SubagentWaitManager {
nextRunId,
...[...killReconciliationSnapshots.keys()].map((entry) => entry.runId),
];
// Revoke the prior task projection before the successor becomes durable.
// A crash between stores then fails closed instead of preserving stale authority.
const taskBindingResult =
source.expectsCompletionMessage === false
? "missing"
: setCanonicalTaskBackingDetail({
runtime: "subagent",
childSessionKey: next.childSessionKey,
runId: next.taskRunId ?? next.runId,
detail: createSubagentTaskBackingDetail(generation),
});
if (taskBindingResult === "persist_failed") {
this.restoreKillReconciliationSnapshots(killReconciliationSnapshots);
this.options.runs.delete(nextRunId);
this.options.runs.set(previousRunId, source);
log.warn("failed to bind replacement subagent task generation; restored source lease", {
runId: next.runId,
});
if (replaceParams.persistenceFailure === "throw") {
throw new Error(`failed to bind replacement subagent task generation for ${next.runId}`);
}
return false;
}
try {
this.options.persistOrThrow(...changedRunIds);
} catch (error) {
src/gateway/server-methods/tasks.test.ts
+1
View File
@@ -1044,6 +1044,7 @@ describe("tasks gateway handlers", () => {
cfg: {},
sessionKey: "agent:codex:acp:child",
reason: "operator requested stop",
expectedRunId: "run-cancel-acp-gateway",
});
expect(payload?.found).toBe(true);
expect(payload?.cancelled).toBe(true);
src/plugins/runtime/runtime-taskflow.test.ts
+13
View File
@@ -1,5 +1,7 @@
// Runtime task-flow tests cover plugin task-flow registration and execution behavior.
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { createAcpTaskBackingDetailForTest } from "../../tasks/task-backing-authority.test-support.js";
import { createRunningTaskRunCore } from "../../tasks/task-executor.js";
import { createTaskFlowForTask, getTaskFlowById } from "../../tasks/task-flow-registry.js";
import { getTaskById } from "../../tasks/task-registry.js";
import {
@@ -109,6 +111,17 @@ describe("runtime TaskFlow", () => {
expect(otherTaskFlow.get(created.flowId)).toBeUndefined();
expect(otherTaskFlow.list()).toStrictEqual([]);
createRunningTaskRunCore({
runtime: "acp",
ownerKey: "agent:main:main",
scopeKind: "session",
childSessionKey: "agent:main:subagent:child",
runId: "runtime-taskflow-child",
task: "Inspect PR 1",
startedAt: 10,
detail: createAcpTaskBackingDetailForTest("instance:runtime-taskflow-child"),
});
const child = ownerTaskFlow.runTask({
flowId: created.flowId,
runtime: "acp",
src/plugins/runtime/runtime-tasks.test.ts
+29 -1
View File
@@ -2,6 +2,8 @@
import { createRequireRecord } from "openclaw/plugin-sdk/test-fixtures";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { getDetachedTaskLifecycleRuntime } from "../../tasks/detached-task-runtime.js";
import { createAcpTaskBackingDetailForTest } from "../../tasks/task-backing-authority.test-support.js";
import { createRunningTaskRunCore } from "../../tasks/task-executor.js";
import { createTaskRecord } from "../../tasks/task-registry.js";
import { setDetachedTaskLifecycleRuntime } from "../../tasks/task-runtime.test-helpers.js";
import {
@@ -37,6 +39,24 @@ function requireCreatedFlow<T>(flow: T | null): T {
return flow;
}
function createCanonicalAcpTask(runId: string) {
const task = createRunningTaskRunCore({
runtime: "acp",
ownerKey: "agent:main:main",
scopeKind: "session",
childSessionKey: "agent:main:subagent:child",
runId,
task: "Canonical child",
startedAt: 10,
deliveryStatus: "pending",
detail: createAcpTaskBackingDetailForTest(`instance:${runId}`),
});
if (!task) {
throw new Error("expected canonical backing task creation to succeed");
}
return task;
}
describe("runtime tasks", () => {
beforeEach(() => {
installRuntimeTaskDeliveryMock();
@@ -74,6 +94,7 @@ describe("runtime tasks", () => {
stateJson: { lane: "priority" },
}),
);
createCanonicalAcpTask("runtime-task-run");
const child = legacyTaskFlow.runTask({
flowId: created.flowId,
runtime: "acp",
@@ -123,7 +144,7 @@ describe("runtime tasks", () => {
expect(taskRun.title).toBe("Review PR 1");
expect(taskRun.progressSummary).toBe("Inspecting");
expect(taskRuns.findLatest()?.id).toBe(child.task.taskId);
expect(taskRuns.resolve("runtime-task-run")?.id).toBe(child.task.taskId);
expect(taskRuns.resolve(child.task.taskId)?.id).toBe(child.task.taskId);
const summary = requireRecord(taskFlows.getTaskSummary(created.flowId));
expect(summary.total).toBe(1);
expect(summary.active).toBe(1);
@@ -159,6 +180,7 @@ describe("runtime tasks", () => {
goal: "Cancel active task",
}),
);
createCanonicalAcpTask("runtime-task-cancel");
const child = legacyTaskFlow.runTask({
flowId: created.flowId,
runtime: "acp",
@@ -182,6 +204,9 @@ describe("runtime tasks", () => {
cfg: {},
sessionKey: "agent:main:subagent:child",
reason: "task-cancel",
expectedRunId: "runtime-task-cancel",
expectedInstanceId: "instance:runtime-task-cancel",
expectedOwnerKey: "agent:main:main",
});
expect(result.found).toBe(true);
expect(result.cancelled).toBe(true);
@@ -208,6 +233,7 @@ describe("runtime tasks", () => {
goal: "Cancel through runtime seam",
}),
);
createCanonicalAcpTask("runtime-task-cancel-seam");
const child = legacyTaskFlow.runTask({
flowId: created.flowId,
runtime: "acp",
@@ -260,6 +286,7 @@ describe("runtime tasks", () => {
goal: "Keep owner isolation",
}),
);
createCanonicalAcpTask("runtime-task-isolation");
const child = legacyTaskFlow.runTask({
flowId: created.flowId,
runtime: "acp",
@@ -368,6 +395,7 @@ describe("runtime tasks", () => {
cfg: {},
sessionKey: "agent:ops:acp:child",
reason: "task-cancel",
expectedRunId: "ops-global-run",
});
});
});
src/tasks/task-backing-authority-write.ts
+29
View File
@@ -0,0 +1,29 @@
import { getTaskFlowById } from "./task-flow-runtime-internal.js";
import { updateTask } from "./task-registry-mutation.js";
import { getTasksByRunScope } from "./task-registry-state.js";
import type { JsonValue, TaskRuntime } from "./task-registry.types.js";
/** Rebinds only the runtime-owned canonical task when its operational generation changes. */
export function setCanonicalTaskBackingDetail(params: {
runtime: TaskRuntime;
childSessionKey: string;
runId: string;
detail: JsonValue;
}): "updated" | "missing" | "persist_failed" {
try {
const task = getTasksByRunScope({
runId: params.runId,
runtime: params.runtime,
sessionKey: params.childSessionKey,
}).find((candidate) => {
const flowId = candidate.parentFlowId?.trim();
return flowId && getTaskFlowById(flowId)?.syncMode === "task_mirrored";
});
if (!task) {
return "missing";
}
return updateTask(task.taskId, { detail: params.detail }) ? "updated" : "persist_failed";
} catch {
return "persist_failed";
}
}
src/tasks/task-backing-authority.test-support.ts
+11
View File
@@ -0,0 +1,11 @@
export function createAcpTaskBackingDetailForTest(
instanceId: string,
generation = 1,
): {
kind: "task_backing_instance";
runtime: "acp";
instanceId: string;
generation: number;
} {
return { kind: "task_backing_instance", runtime: "acp", instanceId, generation };
}
src/tasks/task-backing-authority.ts
+188
View File
@@ -0,0 +1,188 @@
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
import { getTaskFlowById } from "./task-flow-runtime-internal.js";
import {
ensureTaskRegistryReady,
taskIdsByRelatedSessionKey,
tasks,
} from "./task-registry-state.js";
import type { JsonValue, TaskRecord, TaskRuntime, TaskScopeKind } from "./task-registry.types.js";
const TASK_BACKING_DETAIL_KIND = "task_backing_instance";
/** Owner-minted identity persisted in canonical tasks and copied into managed projections. */
export type TaskBackingInstance =
| { runtime: "acp"; instanceId: string; generation: number }
| { runtime: "subagent"; generation: number };
type TaskBackingDetail = TaskBackingInstance & { kind: typeof TASK_BACKING_DETAIL_KIND };
type ManagedTaskBacking = { taskId: string; instance: TaskBackingInstance };
function readTaskBackingInstance(value: unknown): TaskBackingInstance | undefined {
const detail = asOptionalRecord(value);
if (detail?.kind !== TASK_BACKING_DETAIL_KIND) {
return undefined;
}
if (detail.runtime === "acp") {
const instanceId = typeof detail.instanceId === "string" ? detail.instanceId.trim() : "";
return instanceId &&
typeof detail.generation === "number" &&
Number.isSafeInteger(detail.generation) &&
detail.generation > 0
? { runtime: "acp", instanceId, generation: detail.generation }
: undefined;
}
if (
detail.runtime === "subagent" &&
typeof detail.generation === "number" &&
Number.isSafeInteger(detail.generation) &&
detail.generation > 0
) {
return { runtime: "subagent", generation: detail.generation };
}
return undefined;
}
function readManagedTaskBacking(value: unknown): ManagedTaskBacking | undefined {
const detail = asOptionalRecord(value);
const taskId = typeof detail?.taskId === "string" ? detail.taskId.trim() : "";
const instance = readTaskBackingInstance(detail);
return taskId && instance ? { taskId, instance } : undefined;
}
function sameTaskBackingInstance(left: TaskBackingInstance, right: TaskBackingInstance): boolean {
return left.runtime === "acp" && right.runtime === "acp"
? left.instanceId === right.instanceId && left.generation === right.generation
: left.runtime === "subagent" && right.runtime === "subagent"
? left.generation === right.generation
: false;
}
function isCanonicalBackingTask(task: TaskRecord): boolean {
const flowId = task.parentFlowId?.trim();
return Boolean(flowId && getTaskFlowById(flowId)?.syncMode === "task_mirrored");
}
function resolveCurrentCanonicalBacking(params: {
runtime: TaskRuntime;
scopeKind: TaskScopeKind;
ownerKey: string;
childSessionKey: string;
runId: string;
}): { task: TaskRecord; instance: TaskBackingInstance } | undefined {
ensureTaskRegistryReady();
const candidates = [...(taskIdsByRelatedSessionKey.get(params.childSessionKey) ?? [])]
.flatMap((taskId) => {
const task = tasks.get(taskId);
return task ? [task] : [];
})
.flatMap((task) => {
const instance = readTaskBackingInstance(task.detail);
return instance &&
instance.runtime === params.runtime &&
task.runtime === params.runtime &&
task.scopeKind === params.scopeKind &&
task.childSessionKey?.trim() === params.childSessionKey &&
isCanonicalBackingTask(task)
? [{ task, instance }]
: [];
})
.toSorted((left, right) => {
const generationDelta = right.instance.generation - left.instance.generation;
if (generationDelta !== 0) {
return generationDelta;
}
return (
right.task.createdAt - left.task.createdAt ||
right.task.taskId.localeCompare(left.task.taskId)
);
});
const current = candidates[0];
return current?.task.ownerKey === params.ownerKey && current.task.runId?.trim() === params.runId
? current
: undefined;
}
function createAcpTaskBackingDetail(instanceId: string, generation = 1): TaskBackingDetail {
return { kind: TASK_BACKING_DETAIL_KIND, runtime: "acp", instanceId, generation };
}
export function createNextAcpTaskBackingDetail(params: {
childSessionKey: string;
instanceId: string;
}): JsonValue {
ensureTaskRegistryReady();
// ACP serializes turns per child session. Persisting the next generation here
// keeps same-run-id replacements distinguishable after restart.
let generation = 0;
for (const taskId of taskIdsByRelatedSessionKey.get(params.childSessionKey) ?? []) {
const task = tasks.get(taskId);
const instance = task ? readTaskBackingInstance(task.detail) : undefined;
if (task && instance?.runtime === "acp" && isCanonicalBackingTask(task)) {
generation = Math.max(generation, instance.generation);
}
}
return createAcpTaskBackingDetail(params.instanceId, generation + 1);
}
export function createSubagentTaskBackingDetail(generation: number): TaskBackingDetail {
return { kind: TASK_BACKING_DETAIL_KIND, runtime: "subagent", generation };
}
export function resolveManagedTaskBackingDetail(params: {
runtime: TaskRuntime;
scopeKind: TaskScopeKind;
ownerKey: string;
childSessionKey: string;
runId: string;
}): JsonValue | undefined {
const current = resolveCurrentCanonicalBacking(params);
return current
? current.instance.runtime === "acp"
? {
...createAcpTaskBackingDetail(current.instance.instanceId, current.instance.generation),
taskId: current.task.taskId,
}
: {
...createSubagentTaskBackingDetail(current.instance.generation),
taskId: current.task.taskId,
}
: undefined;
}
export function getManagedTaskBackingInstance(task: TaskRecord): TaskBackingInstance | undefined {
const flowId = task.parentFlowId?.trim();
return flowId && getTaskFlowById(flowId)?.syncMode === "managed"
? readManagedTaskBacking(task.detail)?.instance
: undefined;
}
/** A managed projection may control a child only while its exact canonical instance is current. */
export function hasAuthoritativeTaskBacking(task: TaskRecord): boolean {
if (task.runtime !== "acp" && task.runtime !== "subagent") {
return true;
}
const flowId = task.parentFlowId?.trim();
if (!flowId || getTaskFlowById(flowId)?.syncMode !== "managed") {
return true;
}
const childSessionKey = task.childSessionKey?.trim();
if (!childSessionKey) {
return true;
}
const runId = task.runId?.trim();
const managed = readManagedTaskBacking(task.detail);
if (!runId || !managed) {
return false;
}
const current = resolveCurrentCanonicalBacking({
runtime: task.runtime,
scopeKind: task.scopeKind,
ownerKey: task.ownerKey,
childSessionKey,
runId,
});
return Boolean(
current &&
current.task.taskId === managed.taskId &&
sameTaskBackingInstance(current.instance, managed.instance),
);
}
src/tasks/task-executor.test.ts
+285 -16
View File
@@ -1,11 +1,14 @@
// Covers task executor runtime selection, lifecycle updates, and error paths.
import { afterEach, describe, expect, it, vi } from "vitest";
import { resetAgentEventsForTest } from "../infra/agent-events.js";
import { emitAgentEvent, resetAgentEventsForTest } from "../infra/agent-events.js";
import { resetSystemEventsForTest } from "../infra/system-events.js";
import { withStateDirEnv } from "../test-helpers/state-dir-env.js";
import { captureEnv } from "../test-utils/env.js";
import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js";
import { getDetachedTaskLifecycleRuntime } from "./detached-task-runtime.js";
import { setCanonicalTaskBackingDetail } from "./task-backing-authority-write.js";
import { createSubagentTaskBackingDetail } from "./task-backing-authority.js";
import { createAcpTaskBackingDetailForTest } from "./task-backing-authority.test-support.js";
import {
cancelFlowById,
cancelFlowByIdForOwner,
@@ -23,6 +26,7 @@ import {
createManagedTaskFlow as createManagedTaskFlowOrNull,
getTaskFlowById,
listTaskFlowRecords,
requestFlowCancel,
} from "./task-flow-registry.js";
import type { TaskFlowRecord } from "./task-flow-registry.types.js";
import {
@@ -57,7 +61,17 @@ function createQueuedTaskRun(params: Parameters<typeof createQueuedTaskRunOrNull
function createRunningTaskRun(
params: Parameters<typeof createRunningTaskRunOrNull>[0],
): TaskRecord {
const task = createRunningTaskRunOrNull(params);
const detail =
params.detail ??
(params.runtime === "acp"
? createAcpTaskBackingDetailForTest(`instance:${params.runId ?? "unknown"}`)
: params.runtime === "subagent"
? createSubagentTaskBackingDetail(1)
: undefined);
const task = createRunningTaskRunOrNull({
...params,
...(detail !== undefined ? { detail } : {}),
});
if (!task) {
throw new Error("expected running task creation to succeed");
}
@@ -212,6 +226,7 @@ function expectCancelledAcpChildTask(
cfg: {} as never,
sessionKey: "agent:codex:acp:child",
reason: "task-cancel",
expectedRunId: child.runId,
});
}
@@ -483,17 +498,18 @@ describe("task-executor", () => {
to: "notifychat:123",
},
});
const child = createRunningTaskRun({
runtime: "acp",
ownerKey: "agent:main:main",
scopeKind: "session",
parentFlowId: flow.flowId,
childSessionKey: "agent:codex:acp:child",
runId: "run-linear-cancel",
task: "Inspect a PR",
startedAt: 10,
deliveryStatus: "pending",
});
createRunningAcpChildTaskRun({ runId: "run-linear-cancel" });
const child = requireCreatedFlowTask(
runTaskInFlow({
flowId: flow.flowId,
runtime: "acp",
childSessionKey: "agent:codex:acp:child",
runId: "run-linear-cancel",
task: "Inspect a PR",
status: "running",
startedAt: 10,
}),
);
const cancelled = await cancelFlowById({
cfg: {} as never,
@@ -502,7 +518,7 @@ describe("task-executor", () => {
expect(cancelled.found).toBe(true);
expect(cancelled.cancelled).toBe(true);
const task = findTaskByRunId("run-linear-cancel");
const task = getTaskById(child.taskId);
expect(task?.taskId).toBe(child.taskId);
expect(task?.status).toBe("cancelled");
const cancelledFlow = getTaskFlowById(flow.flowId);
@@ -518,6 +534,15 @@ describe("task-executor", () => {
controllerId: "tests/managed-flow",
goal: "Cancel a killed child",
});
createRunningTaskRun({
runtime: "subagent",
ownerKey: "agent:main:main",
scopeKind: "session",
childSessionKey: "agent:worker:subagent:flow-killed",
runId: "run-flow-provisional-kill",
task: "Stop the child",
startedAt: 10,
});
const created = runTaskInFlow({
flowId: flow.flowId,
runtime: "subagent",
@@ -577,6 +602,7 @@ describe("task-executor", () => {
to: "notifychat:123",
},
});
createRunningAcpChildTaskRun({ runId: "run-flow-child" });
const created = runTaskInFlow({
flowId: flow.flowId,
@@ -687,6 +713,7 @@ describe("task-executor", () => {
controllerId: "tests/managed-flow",
goal: "Long running batch",
});
createRunningAcpChildTaskRun({ runId: "run-flow-sticky-cancel" });
const created = runTaskInFlow({
flowId: flow.flowId,
runtime: "acp",
@@ -777,6 +804,50 @@ describe("task-executor", () => {
});
});
it("does not let a managed flow cancel another owner's backing run", async () => {
await withTaskExecutorStateDir(async () => {
const backing = createRunningTaskRun({
runtime: "acp",
ownerKey: "agent:main:victim",
scopeKind: "session",
childSessionKey: "agent:main:acp:victim-child",
runId: "run-foreign-child",
task: "Victim task",
startedAt: 10,
});
const flow = createManagedTaskFlow({
ownerKey: "agent:main:main",
controllerId: "tests/managed-flow",
goal: "Protected flow",
});
const linked = createRunningTaskRun({
runtime: "acp",
ownerKey: "agent:main:main",
scopeKind: "session",
parentFlowId: flow.flowId,
childSessionKey: "agent:main:acp:victim-child",
runId: "run-foreign-child",
task: "Forged projection",
startedAt: 10,
detail: {
...createAcpTaskBackingDetailForTest("instance:run-foreign-child"),
taskId: backing.taskId,
},
});
expect(linked.parentFlowId).toBe(flow.flowId);
const cancelled = await cancelFlowById({ cfg: {} as never, flowId: flow.flowId });
expect(cancelled).toMatchObject({
found: true,
cancelled: false,
reason: "Child task ownership could not be verified; no cancellation was performed.",
});
expect(getTaskFlowById(flow.flowId)?.cancelRequestedAt).toBeUndefined();
expect(hoisted.cancelSessionMock).not.toHaveBeenCalled();
});
});
it("cancels active ACP child tasks", async () => {
await withTaskExecutorStateDir(async () => {
hoisted.cancelSessionMock.mockResolvedValue(undefined);
@@ -1084,6 +1155,7 @@ describe("task-executor", () => {
expect(hoisted.killSubagentRunAdminMock).toHaveBeenCalledWith({
cfg: {} as never,
sessionKey: "agent:codex:subagent:child",
expectedRunId: "run-subagent-cancel",
});
});
});
@@ -1097,6 +1169,7 @@ describe("task-executor", () => {
controllerId: "tests/cancel-flow",
goal: "Cancel linked tasks",
});
createRunningAcpChildTaskRun({ runId: "run-flow-cancel-via-runtime" });
const child = runTaskInFlow({
flowId: flow.flowId,
runtime: "acp",
@@ -1136,7 +1209,7 @@ describe("task-executor", () => {
await withTaskExecutorStateDir(async () => {
const victim = createRunningTaskRun({
runtime: "acp",
ownerKey: "agent:victim:main",
ownerKey: "agent:main:victim",
scopeKind: "session",
childSessionKey: "agent:victim:acp:child",
runId: "run-shared-executor-scope",
@@ -1145,7 +1218,7 @@ describe("task-executor", () => {
});
const attacker = createRunningTaskRun({
runtime: "cli",
ownerKey: "agent:attacker:main",
ownerKey: "agent:main:attacker",
scopeKind: "session",
childSessionKey: "agent:attacker:main",
runId: "run-shared-executor-scope",
@@ -1168,5 +1241,201 @@ describe("task-executor", () => {
expect(getTaskById(victim.taskId)?.status).toBe("running");
});
});
it("does not deliver backing lifecycle updates to a foreign managed projection", async () => {
await withTaskExecutorStateDir(async () => {
const backing = createRunningTaskRun({
runtime: "acp",
ownerKey: "agent:main:victim",
scopeKind: "session",
childSessionKey: "agent:main:acp:victim-child",
runId: "run-shared-child",
task: "Victim ACP task",
deliveryStatus: "pending",
});
const flow = createManagedTaskFlow({
ownerKey: "agent:main:main",
controllerId: "tests/foreign-projection",
goal: "Foreign projection",
});
const projection = createRunningTaskRun({
runtime: "acp",
ownerKey: "agent:main:main",
scopeKind: "session",
parentFlowId: flow.flowId,
childSessionKey: "agent:main:acp:victim-child",
runId: "run-shared-child",
task: "Forged projection",
detail: {
...createAcpTaskBackingDetailForTest("instance:run-shared-child"),
taskId: backing.taskId,
},
});
failTaskRunByRunId({
runId: "run-shared-child",
runtime: "acp",
sessionKey: "agent:main:acp:victim-child",
endedAt: 40,
error: "victim failure",
});
expect(getTaskById(backing.taskId)?.status).toBe("failed");
expect(getTaskById(projection.taskId)?.status).toBe("running");
});
});
it("does not apply agent lifecycle events to a foreign managed projection", async () => {
await withTaskExecutorStateDir(async () => {
const backing = createRunningTaskRun({
runtime: "subagent",
ownerKey: "agent:main:victim",
scopeKind: "session",
childSessionKey: "agent:main:subagent:victim-child",
runId: "run-agent-event-foreign-child",
task: "Victim subagent task",
deliveryStatus: "pending",
});
const flow = createManagedTaskFlow({
ownerKey: "agent:main:main",
controllerId: "tests/foreign-agent-event",
goal: "Foreign agent-event projection",
});
const projection = createRunningTaskRun({
runtime: "subagent",
ownerKey: "agent:main:main",
scopeKind: "session",
parentFlowId: flow.flowId,
childSessionKey: "agent:main:subagent:victim-child",
runId: "run-agent-event-foreign-child",
task: "Forged agent-event projection",
detail: { ...createSubagentTaskBackingDetail(1), taskId: backing.taskId },
});
const currentFlow = getTaskFlowById(flow.flowId);
if (!currentFlow) {
throw new Error("Expected managed flow");
}
const cancelRequest = requestFlowCancel({
flowId: flow.flowId,
expectedRevision: currentFlow.revision,
});
if (!cancelRequest.applied) {
throw new Error(cancelRequest.reason);
}
const beforeFlow = getTaskFlowById(flow.flowId);
emitAgentEvent({
runId: "run-agent-event-foreign-child",
sessionKey: "agent:main:subagent:victim-child",
stream: "lifecycle",
data: { phase: "end", endedAt: 40 },
});
expect(getTaskFlowById(flow.flowId)).toMatchObject({
status: beforeFlow?.status,
revision: beforeFlow?.revision,
cancelRequestedAt: beforeFlow?.cancelRequestedAt,
});
expect(getTaskById(projection.taskId)?.status).toBe("running");
});
});
it("applies agent lifecycle events to an owner-matched managed projection", async () => {
await withTaskExecutorStateDir(async () => {
createRunningTaskRun({
runtime: "subagent",
ownerKey: "agent:main:main",
scopeKind: "session",
childSessionKey: "agent:main:subagent:owned-child",
runId: "run-agent-event-owned-child",
task: "Owned subagent task",
deliveryStatus: "pending",
});
const flow = createManagedTaskFlow({
ownerKey: "agent:main:main",
controllerId: "tests/owned-agent-event",
goal: "Owned agent-event projection",
});
const projected = runTaskInFlow({
flowId: flow.flowId,
runtime: "subagent",
childSessionKey: "agent:main:subagent:owned-child",
runId: "run-agent-event-owned-child",
task: "Owned agent-event projection",
status: "running",
});
if (!projected.created) {
throw new Error(projected.reason);
}
const projection = requireCreatedFlowTask(projected);
emitAgentEvent({
runId: "run-agent-event-owned-child",
sessionKey: "agent:main:subagent:owned-child",
stream: "lifecycle",
data: { phase: "end", endedAt: 40 },
});
expect(getTaskById(projection.taskId)).toMatchObject({
status: "succeeded",
endedAt: 40,
});
});
});
it("revokes a managed projection when its backing generation is replaced", async () => {
await withTaskExecutorStateDir(async () => {
createRunningTaskRun({
runtime: "subagent",
ownerKey: "agent:main:main",
scopeKind: "session",
childSessionKey: "agent:main:subagent:replaced-child",
runId: "run-reused-after-recovery",
task: "Original subagent generation",
deliveryStatus: "pending",
});
const flow = createManagedTaskFlow({
ownerKey: "agent:main:main",
controllerId: "tests/replaced-generation",
goal: "Reject replaced generation",
});
const projection = requireCreatedFlowTask(
runTaskInFlow({
flowId: flow.flowId,
runtime: "subagent",
childSessionKey: "agent:main:subagent:replaced-child",
runId: "run-reused-after-recovery",
task: "Managed original generation",
status: "running",
}),
);
expect(
setCanonicalTaskBackingDetail({
runtime: "subagent",
childSessionKey: "agent:main:subagent:replaced-child",
runId: "run-reused-after-recovery",
detail: createSubagentTaskBackingDetail(2),
}),
).toBe("updated");
const cancelled = await cancelFlowById({ cfg: {} as never, flowId: flow.flowId });
emitAgentEvent({
runId: "run-reused-after-recovery",
sessionKey: "agent:main:subagent:replaced-child",
stream: "lifecycle",
data: { phase: "end", endedAt: 40 },
});
expect(cancelled).toMatchObject({
found: true,
cancelled: false,
reason: "Child task ownership could not be verified; no cancellation was performed.",
});
expect(getTaskFlowById(flow.flowId)?.cancelRequestedAt).toBeUndefined();
expect(getTaskById(projection.taskId)?.status).toBe("running");
expect(hoisted.killSubagentRunAdminMock).not.toHaveBeenCalled();
});
});
});
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */
src/tasks/task-executor.ts
+41 -2
View File
@@ -21,6 +21,10 @@ import {
recordTaskProgressByRunId,
setTaskRunDeliveryStatusByRunId,
} from "./runtime-internal.js";
import {
hasAuthoritativeTaskBacking,
resolveManagedTaskBackingDetail,
} from "./task-backing-authority.js";
import {
isProvisionalSubagentKillTask,
isTaskFlowCancellationPending,
@@ -356,6 +360,31 @@ function runTaskInFlow(params: RunTaskInFlowParams): RunTaskInFlowResult {
};
}
const childSessionKey = params.childSessionKey?.trim();
const runId = params.runId?.trim();
const managedBackingDetail =
childSessionKey && runId && (params.runtime === "acp" || params.runtime === "subagent")
? resolveManagedTaskBackingDetail({
runtime: params.runtime,
scopeKind: "session",
ownerKey: flow.ownerKey,
childSessionKey,
runId,
})
: undefined;
if (
childSessionKey &&
(params.runtime === "acp" || params.runtime === "subagent") &&
!managedBackingDetail
) {
return {
found: true,
created: false,
reason: "Task backing ownership could not be verified.",
flow,
};
}
const common = {
runtime: params.runtime,
sourceId: params.sourceId,
@@ -372,6 +401,7 @@ function runTaskInFlow(params: RunTaskInFlowParams): RunTaskInFlowResult {
preferMetadata: params.preferMetadata,
notifyPolicy: params.notifyPolicy,
deliveryStatus: params.deliveryStatus ?? "pending",
...(managedBackingDetail !== undefined ? { detail: managedBackingDetail } : {}),
};
let task: TaskRecord | null;
try {
@@ -498,6 +528,17 @@ export async function cancelFlowById(params: {
tasks: listTasksForFlowId(flow.flowId),
};
}
const linkedTasks = listTasksForFlowId(flow.flowId);
const activeTasks = linkedTasks.filter(isTaskFlowCancellationPending);
if (activeTasks.some((task) => !hasAuthoritativeTaskBacking(task))) {
return {
found: true,
cancelled: false,
reason: "Child task ownership could not be verified; no cancellation was performed.",
flow,
tasks: linkedTasks,
};
}
const cancelRequestedFlow = markFlowCancelRequested(flow);
if ("reason" in cancelRequestedFlow) {
return {
@@ -508,8 +549,6 @@ export async function cancelFlowById(params: {
tasks: listTasksForFlowId(flow.flowId),
};
}
const linkedTasks = listTasksForFlowId(flow.flowId);
const activeTasks = linkedTasks.filter(isTaskFlowCancellationPending);
for (const task of activeTasks) {
await cancelDetachedTaskRunById({
cfg: params.cfg,
src/tasks/task-registry-cancel.ts
+21
View File
@@ -4,6 +4,10 @@ import { isBackgroundExecTask } from "./background-exec-task-contract.js";
import { CRON_TASK_KIND } from "./cron-task-contract.js";
import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js";
import { isHarnessOwnedSubagentTask } from "./harness-owned-subagent-task.js";
import {
getManagedTaskBackingInstance,
hasAuthoritativeTaskBacking,
} from "./task-backing-authority.js";
import { isProvisionalSubagentKillTask } from "./task-cancellation-state.js";
import { isTerminalTaskStatus } from "./task-executor-policy.js";
import { ensureLinkedTaskFlowRegistryReady } from "./task-registry-common.js";
@@ -70,6 +74,15 @@ export async function cancelTaskById(params: {
}
const childSessionKey = task.childSessionKey?.trim();
try {
if (!hasAuthoritativeTaskBacking(task)) {
return {
found: true,
cancelled: false,
reason: "Task backing ownership could not be verified.",
task: cloneTaskRecord(task),
};
}
const managedBacking = getManagedTaskBackingInstance(task);
ensureTaskCancellationReady(task);
// A direct kill is only a provisional terminal projection. Re-read the
// owning subagent run before promotion so its canonical completion can win.
@@ -126,12 +139,20 @@ export async function cancelTaskById(params: {
cfg: params.cfg,
sessionKey: childSessionKey,
reason: params.reason?.trim() || "task-cancel",
expectedRunId: task.runId,
...(managedBacking?.runtime === "acp"
? { expectedInstanceId: managedBacking.instanceId, expectedOwnerKey: task.ownerKey }
: {}),
});
} else if (task.runtime === "subagent") {
const { killSubagentRunAdmin } = await loadTaskRegistryControlRuntime();
const result = await killSubagentRunAdmin({
cfg: params.cfg,
sessionKey: childSessionKey,
expectedRunId: task.runId,
...(managedBacking?.runtime === "subagent"
? { expectedGeneration: managedBacking.generation, expectedOwnerKey: task.ownerKey }
: {}),
});
const current = tasks.get(task.taskId);
if (current?.status === "cancelled" && current.error === SUBAGENT_KILL_TASK_ERROR) {
src/tasks/task-registry-control.types.ts
+6
View File
@@ -11,6 +11,9 @@ type CancelAcpSessionAdmin = (params: {
cfg: OpenClawConfig;
sessionKey: string;
reason: string;
expectedRunId?: string;
expectedInstanceId?: string;
expectedOwnerKey?: string;
}) => Promise<void>;
type KillSubagentRunAdminResult =
@@ -28,6 +31,9 @@ type KillSubagentRunAdminResult =
type KillSubagentRunAdmin = (params: {
cfg: OpenClawConfig;
sessionKey: string;
expectedRunId?: string;
expectedGeneration?: number;
expectedOwnerKey?: string;
}) => Promise<KillSubagentRunAdminResult>;
export type TaskRegistryControlRuntime = {
src/tasks/task-registry-create-helpers.ts
+4 -4
View File
@@ -1,6 +1,7 @@
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { normalizeAgentId, parseAgentSessionKey } from "../routing/session-key.js";
import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js";
import { getTaskFlowById } from "./task-flow-runtime-internal.js";
import {
assertParentFlowLinkAllowed,
ensureLinkedTaskFlowRegistryReady,
@@ -46,10 +47,9 @@ export function findExistingTaskForCreate(params: {
) {
return false;
}
if (params.runtime === "acp") {
// ACP one-task flow ids can be derived after creation; they must not
// split one logical ACP run into duplicate task rows.
return true;
if (params.runtime === "acp" && !params.parentFlowId?.trim()) {
const existingFlowId = task.parentFlowId?.trim();
return !existingFlowId || getTaskFlowById(existingFlowId)?.syncMode === "task_mirrored";
}
return (
(normalizeOptionalString(task.parentFlowId) ?? "") ===
src/tasks/task-registry-lifecycle.ts
+2 -1
View File
@@ -1,5 +1,6 @@
import { buildAgentRunTerminalOutcomeFromLifecycleEvent } from "../agents/agent-run-terminal-outcome.js";
import { onAgentEvent } from "../infra/agent-events.js";
import { hasAuthoritativeTaskBacking } from "./task-backing-authority.js";
import { isTerminalTaskStatus } from "./task-executor-policy.js";
import { recordTaskActivityEvent } from "./task-registry-activity.js";
import {
@@ -39,7 +40,7 @@ function ensureListener() {
}
const now = evt.ts || Date.now();
for (const current of scopedTasks) {
if (isTerminalTaskStatus(current.status)) {
if (isTerminalTaskStatus(current.status) || !hasAuthoritativeTaskBacking(current)) {
continue;
}
if (recordTaskActivityEvent(current, evt)) {
src/tasks/task-registry-record-api.ts
+7
View File
@@ -1,6 +1,7 @@
import crypto from "node:crypto";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js";
import { hasAuthoritativeTaskBacking } from "./task-backing-authority.js";
import { isTerminalTaskStatus } from "./task-executor-policy.js";
import {
appendTaskEvent,
@@ -136,6 +137,9 @@ function updateTasksByRunId(params: {
}
const updated: TaskRecord[] = [];
for (const match of matches) {
if (!hasAuthoritativeTaskBacking(match)) {
continue;
}
const task = updateTask(match.taskId, params.patch);
if (task) {
updated.push(task);
@@ -320,6 +324,9 @@ export function updateTaskStateByRunId(params: {
}
const updated: TaskRecord[] = [];
for (const current of matches) {
if (!hasAuthoritativeTaskBacking(current)) {
continue;
}
const patch: Partial<TaskRecord> = {};
const nextStatus = params.status ? normalizeTaskStatus(params.status) : current.status;
if (
test/e2e/qa-lab/runtime/webhooks-taskflow-cancellation-authz.e2e.test.ts
+725
View File
@@ -0,0 +1,725 @@
// Webhooks TaskFlow E2E covers route-bound child cancellation on a real Gateway listener.
import fs from "node:fs/promises";
import path from "node:path";
import type { OpenClawPluginService } from "openclaw/plugin-sdk/core";
import {
createPluginStateKeyedStoreForTests,
resetPluginStateStoreForTests,
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
import { createTestPluginApi } from "openclaw/plugin-sdk/plugin-test-api";
import { createPluginRuntimeMock } from "openclaw/plugin-sdk/plugin-test-runtime";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import acpxPlugin from "../../../../extensions/acpx/index.js";
import webhooksPlugin from "../../../../extensions/webhooks/index.js";
import {
getAcpSessionManager,
testing as acpManagerTesting,
} from "../../../../src/acp/control-plane/manager.js";
import { createTestAdmittedRunContext } from "../../../../src/agents/admitted-run-context.test-support.js";
import { cancelBackgroundExecSession } from "../../../../src/agents/bash-process-control.js";
import { killSubagentRunAdmin } from "../../../../src/agents/subagents/registry/subagent-control.js";
import { testing as subagentControlTesting } from "../../../../src/agents/subagents/registry/subagent-control.test-support.js";
import { getSubagentRunByRunId } from "../../../../src/agents/subagents/registry/subagent-registry.js";
import {
addSubagentRunForTests,
resetSubagentRegistryForTests,
testing as subagentRegistryTesting,
} from "../../../../src/agents/subagents/registry/subagent-registry.test-helpers.js";
import { clearConfigCache, clearRuntimeConfigSnapshot } from "../../../../src/config/config.js";
import { resolveSessionStorePathCore } from "../../../../src/config/sessions/paths.js";
import { replaceSessionEntrySync } from "../../../../src/config/sessions/session-accessor.js";
import type { OpenClawConfig } from "../../../../src/config/types.openclaw.js";
import { cancelActiveCronTaskRun } from "../../../../src/cron/service/active-run-cancellation.js";
import { startGatewayServer } from "../../../../src/gateway/server.js";
import { getGatewayE2ePortBlock } from "../../../../src/gateway/test-helpers.e2e.js";
import { snapshotGatewayStartupEnv } from "../../../../src/gateway/test-helpers.env.js";
import { registerPluginHttpRoute } from "../../../../src/plugins/http-registry.js";
import {
getActivePluginRegistry,
resetPluginRuntimeStateForTest,
} from "../../../../src/plugins/runtime.js";
import { createPluginRuntime } from "../../../../src/plugins/runtime/index.js";
import { createSubagentTaskBackingDetail } from "../../../../src/tasks/task-backing-authority.js";
import { createAcpTaskBackingDetailForTest } from "../../../../src/tasks/task-backing-authority.test-support.js";
import { createRunningTaskRunCore } from "../../../../src/tasks/task-executor.js";
import { getTaskFlowById } from "../../../../src/tasks/task-flow-registry.js";
import { findTaskByRunId, listTasksForFlowId } from "../../../../src/tasks/task-registry.js";
import {
resetTaskFlowRegistryForTests,
setTaskRegistryControlRuntimeForTests,
resetTaskRegistryForTests,
} from "../../../../src/tasks/task-runtime.test-helpers.js";
import { withEnvAsync } from "../../../../src/test-utils/env.js";
import { createDeferred } from "../../../helpers/promise.js";
import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js";
const TOKEN = "webhooks-taskflow-e2e-token";
const SECRET = "webhooks-taskflow-route-secret";
const ROUTE_PATH = "/plugins/webhooks/authority-proof";
const ROUTE_OWNER = "agent:main:webhook-authority-proof";
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
type WebhookResponse = {
status: number;
body: {
ok?: boolean;
code?: string;
error?: string;
result?: {
flow?: { flowId?: string; status?: string };
tasks?: Array<{ status?: string }>;
};
};
};
beforeEach(() => {
// Keep the real registry and kill lifecycle while injecting the process facts
// that this isolated Gateway has no embedded model run or persisted session for.
subagentControlTesting.setDepsForTest({
abortEmbeddedAgentRun: () => false,
isEmbeddedAgentRunActive: () => false,
clearSessionQueues: () => ({ followupCleared: 0, laneCleared: 0, keys: [] }),
});
subagentRegistryTesting.setDepsForTest({
persistSubagentRunsToDisk: () => {},
persistSubagentRunsToDiskOrThrow: () => {},
restoreSubagentRunsFromDisk: () => 0,
});
});
afterEach(() => {
clearConfigCache();
clearRuntimeConfigSnapshot();
acpManagerTesting.resetAcpSessionManagerForTests();
resetSubagentRegistryForTests({ persist: false });
resetTaskRegistryForTests({ persist: false });
resetTaskFlowRegistryForTests({ persist: false });
resetPluginStateStoreForTests();
resetPluginRuntimeStateForTest();
subagentControlTesting.setDepsForTest();
subagentRegistryTesting.setDepsForTest();
});
function registerRunningSubagent(params: {
runId: string;
childSessionKey: string;
ownerKey: string;
}) {
const startedAt = Date.now();
const generation = (getSubagentRunByRunId(params.runId)?.generation ?? 0) + 1;
addSubagentRunForTests({
runId: params.runId,
childSessionKey: params.childSessionKey,
controllerSessionKey: params.ownerKey,
requesterSessionKey: params.ownerKey,
requesterDisplayKey: params.ownerKey,
task: `Running child ${params.runId}`,
cleanup: "keep",
generation,
createdAt: startedAt,
startedAt,
});
const task = createRunningTaskRunCore({
runtime: "subagent",
ownerKey: params.ownerKey,
scopeKind: "session",
childSessionKey: params.childSessionKey,
runId: params.runId,
task: `Running child ${params.runId}`,
startedAt,
deliveryStatus: "pending",
detail: createSubagentTaskBackingDetail(generation),
});
if (!task) {
throw new Error(`failed to create canonical task for ${params.runId}`);
}
return { generation, task };
}
async function postWebhook(
origin: string,
body: Record<string, unknown>,
): Promise<WebhookResponse> {
const response = await fetch(`${origin}${ROUTE_PATH}`, {
method: "POST",
headers: {
"content-type": "application/json",
"x-openclaw-webhook-secret": SECRET,
},
body: JSON.stringify(body),
});
return {
status: response.status,
body: (await response.json()) as WebhookResponse["body"],
};
}
async function createFlow(origin: string, goal: string): Promise<string> {
const response = await postWebhook(origin, { action: "create_flow", goal });
expect(response).toMatchObject({ status: 200, body: { ok: true } });
const flowId = response.body.result?.flow?.flowId;
if (!flowId) {
throw new Error("webhook create_flow returned no flow id");
}
return flowId;
}
async function projectChild(params: {
origin: string;
flowId: string;
childSessionKey: string;
runId: string;
runtime?: "acp" | "subagent";
}) {
const response = await postWebhook(params.origin, {
action: "run_task",
flowId: params.flowId,
runtime: params.runtime ?? "subagent",
childSessionKey: params.childSessionKey,
runId: params.runId,
task: `Managed projection ${params.runId}`,
});
expect(response).toMatchObject({ status: 200, body: { ok: true } });
}
async function readAcpTraceMethods(tracePath: string): Promise<string[]> {
return (await fs.readFile(tracePath, "utf8"))
.trim()
.split("\n")
.filter(Boolean)
.map((line) => (JSON.parse(line) as { method: string }).method);
}
describe("webhooks TaskFlow child cancellation authority", () => {
it("allows the owner and rejects foreign or replaced backing runs before termination", async () => {
const root = tempDirs.make("openclaw-webhooks-taskflow-authz-");
const stateDir = path.join(root, "state");
const acpxStateDir = path.join(root, "acpx-state");
const acpxTracePath = path.join(root, "acpx-process-trace.jsonl");
const configPath = path.join(root, "openclaw.json");
await fs.mkdir(stateDir, { recursive: true });
const config: OpenClawConfig = {
gateway: {
mode: "local",
bind: "loopback",
auth: { mode: "token", token: TOKEN },
},
acp: {
enabled: true,
backend: "acpx",
dispatch: { enabled: true },
allowedAgents: ["codex"],
},
};
await fs.writeFile(configPath, `${JSON.stringify(config)}\n`, "utf8");
await withEnvAsync(
{
...snapshotGatewayStartupEnv(),
HOME: root,
CODEX_PATH: path.resolve("extensions/acpx/test/fixtures/codex-app-server.mjs"),
OPENCLAW_ACPX_PROCESS_FIXTURE_TRACE: acpxTracePath,
OPENCLAW_ACPX_RUNTIME_STARTUP_PROBE: "0",
OPENCLAW_CONFIG_PATH: configPath,
OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1",
OPENCLAW_HOME: root,
OPENCLAW_SKIP_BROWSER_CONTROL_SERVER: "1",
OPENCLAW_SKIP_CHANNELS: "1",
OPENCLAW_SKIP_CRON: "1",
OPENCLAW_SKIP_GMAIL_WATCHER: "1",
OPENCLAW_SKIP_PROVIDERS: "1",
OPENCLAW_SKIP_ACPX_RUNTIME: undefined,
OPENCLAW_SKIP_ACPX_RUNTIME_PROBE: "1",
OPENCLAW_STATE_DIR: stateDir,
},
async () => {
clearConfigCache();
clearRuntimeConfigSnapshot();
const port = await getGatewayE2ePortBlock();
const server = await startGatewayServer(port, {
auth: { mode: "token", token: TOKEN },
bind: "loopback",
controlUiEnabled: false,
sidecarStartup: "defer",
});
await server.startupSettled;
const registry = getActivePluginRegistry();
if (!registry) {
throw new Error("gateway did not publish an active plugin registry");
}
setTaskRegistryControlRuntimeForTests({
cancelActiveCronTaskRun,
cancelBackgroundExecSession,
getAcpSessionManager,
killSubagentRunAdmin,
});
const routeCleanups: Array<() => void> = [];
const acpxServices: OpenClawPluginService[] = [];
const acpxRuntime = createPluginRuntimeMock({
state: {
openKeyedStore: (options) => createPluginStateKeyedStoreForTests("acpx", options),
},
});
acpxPlugin.register(
createTestPluginApi({
id: "acpx",
name: "ACPX Runtime",
config,
pluginConfig: {
cwd: root,
stateDir: acpxStateDir,
permissionMode: "deny-all",
timeoutSeconds: 15,
agents: {
codex: {
command: process.execPath,
args: [path.resolve("node_modules/@agentclientprotocol/codex-acp/dist/index.js")],
},
},
},
runtime: acpxRuntime,
registerService: (service) => {
acpxServices.push(service);
},
}),
);
const acpxService = acpxServices.at(0);
if (!acpxService) {
throw new Error("ACPX plugin did not register its runtime service");
}
const acpxServiceContext = {
config,
workspaceDir: root,
stateDir,
logger: { info() {}, warn() {}, error() {}, debug() {} },
};
await acpxService.start(acpxServiceContext);
webhooksPlugin.register(
createTestPluginApi({
id: "webhooks",
name: "Webhooks",
config,
pluginConfig: {
routes: {
authorityProof: {
path: ROUTE_PATH,
sessionKey: ROUTE_OWNER,
secret: SECRET,
},
},
},
runtime: createPluginRuntime(),
registerHttpRoute: (route) => {
routeCleanups.push(
registerPluginHttpRoute({
...route,
pluginId: "webhooks",
registry,
source: "extensions/webhooks/index.ts",
}),
);
},
}),
);
try {
const origin = `http://127.0.0.1:${port}`;
const allowedRunId = "run-webhook-owned";
const allowedChild = "agent:main:subagent:webhook-owned";
registerRunningSubagent({
runId: allowedRunId,
childSessionKey: allowedChild,
ownerKey: ROUTE_OWNER,
});
const allowedFlowId = await createFlow(origin, "Cancel owned child");
await projectChild({
origin,
flowId: allowedFlowId,
childSessionKey: allowedChild,
runId: allowedRunId,
});
const allowed = await postWebhook(origin, {
action: "cancel_flow",
flowId: allowedFlowId,
});
expect(allowed).toMatchObject({ status: 200, body: { ok: true } });
expect(getSubagentRunByRunId(allowedRunId)).toMatchObject({
endedReason: "subagent-killed",
execution: { status: "terminal", endedAt: expect.any(Number) },
});
expect(getTaskFlowById(allowedFlowId)?.status).toBe("cancelled");
expect(listTasksForFlowId(allowedFlowId)).toEqual([
expect.objectContaining({ status: "cancelled" }),
]);
const foreignRunId = "run-webhook-foreign";
const foreignChild = "agent:main:subagent:webhook-foreign";
const foreignBacking = registerRunningSubagent({
runId: foreignRunId,
childSessionKey: foreignChild,
ownerKey: "agent:main:foreign-owner",
});
const foreignFlowId = await createFlow(origin, "Reject foreign child");
const foreignAdmission = await postWebhook(origin, {
action: "run_task",
flowId: foreignFlowId,
runtime: "subagent",
childSessionKey: foreignChild,
runId: foreignRunId,
task: "Reject foreign child projection",
});
expect(foreignAdmission).toMatchObject({
status: 409,
body: { ok: false, code: "task_not_created" },
});
const forgedProjection = createRunningTaskRunCore({
runtime: "subagent",
ownerKey: ROUTE_OWNER,
scopeKind: "session",
parentFlowId: foreignFlowId,
childSessionKey: foreignChild,
runId: foreignRunId,
task: "Persisted foreign child projection",
startedAt: Date.now(),
detail: {
...createSubagentTaskBackingDetail(foreignBacking.generation),
taskId: foreignBacking.task.taskId,
},
});
if (!forgedProjection) {
throw new Error("failed to create persisted foreign child projection");
}
const foreign = await postWebhook(origin, {
action: "cancel_flow",
flowId: foreignFlowId,
});
expect(foreign).toMatchObject({
status: 409,
body: { ok: false, code: "cancel_rejected" },
});
expect(getSubagentRunByRunId(foreignRunId)).toMatchObject({
execution: { status: "running" },
});
expect(getSubagentRunByRunId(foreignRunId)?.execution.endedAt).toBeUndefined();
expect(getTaskFlowById(foreignFlowId)).toMatchObject({ status: "queued" });
expect(getTaskFlowById(foreignFlowId)?.cancelRequestedAt).toBeUndefined();
const replacedRunId = "run-webhook-replaced";
const replacementRunId = "run-webhook-replacement";
const replacedChild = "agent:main:subagent:webhook-replaced";
registerRunningSubagent({
runId: replacedRunId,
childSessionKey: replacedChild,
ownerKey: ROUTE_OWNER,
});
const replacedFlowId = await createFlow(origin, "Reject replaced child");
await projectChild({
origin,
flowId: replacedFlowId,
childSessionKey: replacedChild,
runId: replacedRunId,
});
registerRunningSubagent({
runId: replacementRunId,
childSessionKey: replacedChild,
ownerKey: ROUTE_OWNER,
});
const replaced = await postWebhook(origin, {
action: "cancel_flow",
flowId: replacedFlowId,
});
expect(replaced).toMatchObject({
status: 409,
body: { ok: false, code: "cancel_rejected" },
});
expect(getSubagentRunByRunId(replacementRunId)).toMatchObject({
execution: { status: "running" },
});
expect(getSubagentRunByRunId(replacementRunId)?.execution.endedAt).toBeUndefined();
const reusedRunId = "run-webhook-reused";
const reusedChild = "agent:main:subagent:webhook-reused";
registerRunningSubagent({
runId: reusedRunId,
childSessionKey: reusedChild,
ownerKey: ROUTE_OWNER,
});
const reusedFlowId = await createFlow(origin, "Reject reused-id replacement child");
await projectChild({
origin,
flowId: reusedFlowId,
childSessionKey: reusedChild,
runId: reusedRunId,
});
registerRunningSubagent({
runId: reusedRunId,
childSessionKey: reusedChild,
ownerKey: ROUTE_OWNER,
});
const reused = await postWebhook(origin, {
action: "cancel_flow",
flowId: reusedFlowId,
});
expect(reused).toMatchObject({
status: 409,
body: { ok: false, code: "cancel_rejected" },
});
expect(getSubagentRunByRunId(reusedRunId)).toMatchObject({
execution: { status: "running" },
});
expect(getSubagentRunByRunId(reusedRunId)?.execution.endedAt).toBeUndefined();
const acpChild = "agent:main:acp:webhook-replacement";
const reusedAcpRunId = "run-webhook-acp-reused";
const acpManager = getAcpSessionManager();
replaceSessionEntrySync(
{
sessionKey: acpChild,
storePath: resolveSessionStorePathCore(config.session?.store, { agentId: "main" }),
},
{
sessionId: "session-webhook-acp-replacement",
updatedAt: Date.now(),
spawnedBy: ROUTE_OWNER,
parentSessionKey: ROUTE_OWNER,
},
);
await acpManager.initializeSession({
cfg: config,
sessionKey: acpChild,
agent: "codex",
mode: "persistent",
backendId: "acpx",
});
const firstAcpAdmission = createTestAdmittedRunContext(reusedAcpRunId);
await acpManager.runTurn({
admittedRunContext: firstAcpAdmission,
cfg: config,
sessionKey: acpChild,
provenance: "system",
text: "Complete the first same-id turn before replacement.",
mode: "prompt",
requestId: reusedAcpRunId,
onElicitation: async () => ({
action: "accept",
content: { question: "complete normally" },
}),
});
const firstAcpTask = findTaskByRunId(reusedAcpRunId);
if (!firstAcpTask) {
throw new Error("first ACP turn created no canonical task");
}
const acpReplacementFlowId = await createFlow(origin, "Reject same-id ACP replacement");
const staleAcpProjection = createRunningTaskRunCore({
runtime: "acp",
ownerKey: ROUTE_OWNER,
scopeKind: "session",
parentFlowId: acpReplacementFlowId,
childSessionKey: acpChild,
runId: reusedAcpRunId,
task: "Persisted first-generation ACP projection",
startedAt: Date.now(),
detail: {
...createAcpTaskBackingDetailForTest(
firstAcpAdmission.operationalRunInstance.instanceId,
1,
),
taskId: firstAcpTask.taskId,
},
});
if (!staleAcpProjection) {
throw new Error("failed to create persisted ACP projection");
}
const elicitationEntered = createDeferred<void>();
const releaseElicitation = createDeferred<void>();
const replacementAcpTurn = acpManager.runTurn({
admittedRunContext: createTestAdmittedRunContext(reusedAcpRunId),
cfg: config,
sessionKey: acpChild,
provenance: "system",
text: "Keep the same-id replacement active for cancellation fencing proof.",
mode: "prompt",
requestId: reusedAcpRunId,
onElicitation: async () => {
elicitationEntered.resolve();
await releaseElicitation.promise;
return { action: "accept", content: { question: "complete normally" } };
},
});
let acpReplacement: WebhookResponse | undefined;
let acpxMethodsBeforeRelease: string[] = [];
try {
await elicitationEntered.promise;
acpReplacement = await postWebhook(origin, {
action: "cancel_flow",
flowId: acpReplacementFlowId,
});
expect(acpReplacement).toMatchObject({
status: 409,
body: { ok: false, code: "cancel_rejected" },
});
expect(getTaskFlowById(acpReplacementFlowId)).toMatchObject({ status: "queued" });
expect(getTaskFlowById(acpReplacementFlowId)?.cancelRequestedAt).toBeUndefined();
acpxMethodsBeforeRelease = await readAcpTraceMethods(acpxTracePath);
expect(acpxMethodsBeforeRelease).toContain("turn/start");
expect(acpxMethodsBeforeRelease).not.toContain("turn/interrupt");
} finally {
releaseElicitation.resolve();
await replacementAcpTurn;
}
if (!acpReplacement) {
throw new Error("missing ACP replacement cancellation response");
}
const queuedAcpChild = "agent:main:acp:webhook-queued-successor";
const queuedAcpRunId = "run-webhook-acp-queued";
replaceSessionEntrySync(
{
sessionKey: queuedAcpChild,
storePath: resolveSessionStorePathCore(config.session?.store, { agentId: "main" }),
},
{
sessionId: "session-webhook-acp-queued-successor",
updatedAt: Date.now(),
spawnedBy: ROUTE_OWNER,
parentSessionKey: ROUTE_OWNER,
},
);
await acpManager.initializeSession({
cfg: config,
sessionKey: queuedAcpChild,
agent: "codex",
mode: "persistent",
backendId: "acpx",
});
const queuedTargetEntered = createDeferred<void>();
const releaseQueuedTarget = createDeferred<void>();
const queuedTargetTurn = acpManager.runTurn({
admittedRunContext: createTestAdmittedRunContext(queuedAcpRunId),
cfg: config,
sessionKey: queuedAcpChild,
provenance: "system",
text: "Keep the target active while its same-id successor queues.",
mode: "prompt",
requestId: queuedAcpRunId,
onElicitation: async () => {
queuedTargetEntered.resolve();
await releaseQueuedTarget.promise;
return { action: "accept", content: { question: "cancel target" } };
},
});
await queuedTargetEntered.promise;
const queuedFlowId = await createFlow(origin, "Cancel target before queued successor");
await projectChild({
origin,
flowId: queuedFlowId,
runtime: "acp",
childSessionKey: queuedAcpChild,
runId: queuedAcpRunId,
});
const interruptsBeforeQueuedCancel = (await readAcpTraceMethods(acpxTracePath)).filter(
(method) => method === "turn/interrupt",
).length;
const queuedSuccessorEntered = createDeferred<void>();
const releaseQueuedSuccessor = createDeferred<void>();
const queuedSuccessorTurn = acpManager.runTurn({
admittedRunContext: createTestAdmittedRunContext(queuedAcpRunId),
cfg: config,
sessionKey: queuedAcpChild,
provenance: "system",
text: "Complete the same-id successor without inheriting cancellation.",
mode: "prompt",
requestId: queuedAcpRunId,
onElicitation: async () => {
queuedSuccessorEntered.resolve();
await releaseQueuedSuccessor.promise;
return { action: "accept", content: { question: "complete successor" } };
},
});
const queuedCancelPromise = postWebhook(origin, {
action: "cancel_flow",
flowId: queuedFlowId,
});
await vi.waitFor(
async () => {
const interruptCount = (await readAcpTraceMethods(acpxTracePath)).filter(
(method) => method === "turn/interrupt",
).length;
expect(interruptCount - interruptsBeforeQueuedCancel).toBeGreaterThan(0);
},
{ interval: 10, timeout: 10_000 },
);
releaseQueuedTarget.resolve();
const queuedCancel = await queuedCancelPromise;
expect(queuedCancel).toMatchObject({ status: 200, body: { ok: true } });
const interruptsAfterTargetCancel = (await readAcpTraceMethods(acpxTracePath)).filter(
(method) => method === "turn/interrupt",
).length;
expect(interruptsAfterTargetCancel - interruptsBeforeQueuedCancel).toBeGreaterThan(0);
await queuedTargetTurn;
await queuedSuccessorEntered.promise;
const interruptsWhileSuccessorActive = (await readAcpTraceMethods(acpxTracePath)).filter(
(method) => method === "turn/interrupt",
).length;
expect(interruptsWhileSuccessorActive - interruptsAfterTargetCancel).toBe(0);
releaseQueuedSuccessor.resolve();
await queuedSuccessorTurn;
console.info(
"webhooks-taskflow-authority-proof",
JSON.stringify({
allowed: {
httpStatus: allowed.status,
flowStatus: getTaskFlowById(allowedFlowId)?.status,
childStatus: getSubagentRunByRunId(allowedRunId)?.execution.status,
},
foreign: {
admissionStatus: foreignAdmission.status,
httpStatus: foreign.status,
code: foreign.body.code,
flowStatus: getTaskFlowById(foreignFlowId)?.status,
childStatus: getSubagentRunByRunId(foreignRunId)?.execution.status,
},
replaced: {
httpStatus: replaced.status,
code: replaced.body.code,
replacementStatus: getSubagentRunByRunId(replacementRunId)?.execution.status,
},
reusedId: {
httpStatus: reused.status,
code: reused.body.code,
replacementStatus: getSubagentRunByRunId(reusedRunId)?.execution.status,
},
acpReplacement: {
transport: "process",
httpStatus: acpReplacement.status,
code: acpReplacement.body.code,
interruptRequests: acpxMethodsBeforeRelease.filter(
(method) => method === "turn/interrupt",
).length,
replacementStatus: "completed",
},
acpQueuedSuccessor: {
transport: "process",
httpStatus: queuedCancel.status,
targetInterruptRequests: interruptsAfterTargetCancel - interruptsBeforeQueuedCancel,
successorInterruptRequests:
interruptsWhileSuccessorActive - interruptsAfterTargetCancel,
successorStatus: "completed",
},
}),
);
} finally {
for (const cleanup of routeCleanups.toReversed()) {
cleanup();
}
await acpxService.stop?.(acpxServiceContext);
await server.close();
}
},
);
}, 90_000);
});