diff --git a/docs/plugins/webhooks.md b/docs/plugins/webhooks.md index d0f12efb8423..8a6e2ccfb4c1 100644 --- a/docs/plugins/webhooks.md +++ b/docs/plugins/webhooks.md @@ -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 diff --git a/extensions/acpx/test/fixtures/codex-app-server.mjs b/extensions/acpx/test/fixtures/codex-app-server.mjs index 276142e18a44..77e5cb04ab19 100755 --- a/extensions/acpx/test/fixtures/codex-app-server.mjs +++ b/extensions/acpx/test/fixtures/codex-app-server.mjs @@ -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() }; } diff --git a/extensions/webhooks/src/http.test.ts b/extensions/webhooks/src/http.test.ts index 0461aa858a6a..d24559dd8dd2 100644 --- a/extensions/webhooks/src/http.test.ts +++ b/extensions/webhooks/src/http.test.ts @@ -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", }, diff --git a/src/acp/control-plane/manager.background-task.test.ts b/src/acp/control-plane/manager.background-task.test.ts index c7c827becb45..a30d2031cf30 100644 --- a/src/acp/control-plane/manager.background-task.test.ts +++ b/src/acp/control-plane/manager.background-task.test.ts @@ -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) { diff --git a/src/acp/control-plane/manager.background-task.ts b/src/acp/control-plane/manager.background-task.ts index 3b3d0999350b..2f77d28303e3 100644 --- a/src/acp/control-plane/manager.background-task.ts +++ b/src/acp/control-plane/manager.background-task.ts @@ -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( diff --git a/src/acp/control-plane/manager.cancel-session.test.ts b/src/acp/control-plane/manager.cancel-session.test.ts index 0951be999537..15e81913233d 100644 --- a/src/acp/control-plane/manager.cancel-session.test.ts +++ b/src/acp/control-plane/manager.cancel-session.test.ts @@ -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((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((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((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; + }); + }); }); diff --git a/src/acp/control-plane/manager.cancel-session.ts b/src/acp/control-plane/manager.cancel-session.ts index 9707be1dc8e9..f9b06e6a18f7 100644 --- a/src/acp/control-plane/manager.cancel-session.ts +++ b/src/acp/control-plane/manager.cancel-session.ts @@ -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; withSessionActor: WithManagerSessionActor; resolveSession: ResolveManagerSession; @@ -28,15 +31,47 @@ export async function runManagerCancelSession(params: { }): Promise { 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 { + params.revalidate?.(); params.activeTurn.abortController.abort(); if (!params.activeTurn.cancelPromise) { params.activeTurn.cancelPromise = params.activeTurn.runtime.cancel({ diff --git a/src/acp/control-plane/manager.core.ts b/src/acp/control-plane/manager.core.ts index df59e532eec0..a51bb6e066b6 100644 --- a/src/acp/control-plane/manager.core.ts +++ b/src/acp/control-plane/manager.core.ts @@ -340,6 +340,9 @@ export class AcpSessionManager { cfg: OpenClawConfig; sessionKey: string; reason?: string; + expectedRunId?: string; + expectedInstanceId?: string; + expectedOwnerKey?: string; }): Promise { 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), diff --git a/src/acp/control-plane/manager.turn-runner.ts b/src/acp/control-plane/manager.turn-runner.ts index 5f5330d8c2e1..e0feca7ae6d1 100644 --- a/src/acp/control-plane/manager.turn-runner.ts +++ b/src/acp/control-plane/manager.turn-runner.ts @@ -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, diff --git a/src/acp/control-plane/manager.types.ts b/src/acp/control-plane/manager.types.ts index 0e562c1560ff..748768d16dad 100644 --- a/src/acp/control-plane/manager.types.ts +++ b/src/acp/control-plane/manager.types.ts @@ -140,6 +140,8 @@ export type AcpStartupIdentityReconcileResult = { }; export type ActiveTurnState = { + requestId: string; + instanceId: string; runtime: AcpRuntime; handle: AcpRuntimeHandle; abortController: AbortController; diff --git a/src/agents/subagents/registry/subagent-control-kill-runtime.ts b/src/agents/subagents/registry/subagent-control-kill-runtime.ts index 89200daecbdd..faca64384b72 100644 --- a/src/agents/subagents/registry/subagent-control-kill-runtime.ts +++ b/src/agents/subagents/registry/subagent-control-kill-runtime.ts @@ -439,12 +439,20 @@ export async function killLatestSubagentRun(params: { entry: SubagentRunRecord; cache: Map>; suppressTaskDelivery?: boolean; + expectedRunId?: string; + expectedGeneration?: number; }): Promise<{ entry: SubagentRunRecord; result: Awaited>; }> { 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 }; } diff --git a/src/agents/subagents/registry/subagent-control-kill.ts b/src/agents/subagents/registry/subagent-control-kill.ts index 0b488f5879df..f8f528f49243 100644 --- a/src/agents/subagents/registry/subagent-control-kill.ts +++ b/src/agents/subagents/registry/subagent-control-kill.ts @@ -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>(); @@ -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) { diff --git a/src/agents/subagents/registry/subagent-control.test.ts b/src/agents/subagents/registry/subagent-control.test.ts index a9a79dcf29a2..b0802e5d1d2f 100644 --- a/src/agents/subagents/registry/subagent-control.test.ts +++ b/src/agents/subagents/registry/subagent-control.test.ts @@ -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; diff --git a/src/agents/subagents/registry/subagent-registry-run-launch.ts b/src/agents/subagents/registry/subagent-registry-run-launch.ts index aecd239a54e3..4b756e1b8510 100644 --- a/src/agents/subagents/registry/subagent-registry-run-launch.ts +++ b/src/agents/subagents/registry/subagent-registry-run-launch.ts @@ -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) diff --git a/src/agents/subagents/registry/subagent-registry-run-recovery.ts b/src/agents/subagents/registry/subagent-registry-run-recovery.ts index d18b24d5fa9a..124c6c431843 100644 --- a/src/agents/subagents/registry/subagent-registry-run-recovery.ts +++ b/src/agents/subagents/registry/subagent-registry-run-recovery.ts @@ -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) { diff --git a/src/gateway/server-methods/tasks.test.ts b/src/gateway/server-methods/tasks.test.ts index 92d2af09a5cf..61508518fd3b 100644 --- a/src/gateway/server-methods/tasks.test.ts +++ b/src/gateway/server-methods/tasks.test.ts @@ -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); diff --git a/src/plugins/runtime/runtime-taskflow.test.ts b/src/plugins/runtime/runtime-taskflow.test.ts index 6e3ce40bb321..a64fcd997b60 100644 --- a/src/plugins/runtime/runtime-taskflow.test.ts +++ b/src/plugins/runtime/runtime-taskflow.test.ts @@ -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", diff --git a/src/plugins/runtime/runtime-tasks.test.ts b/src/plugins/runtime/runtime-tasks.test.ts index 9765c6b801ba..2268c6997e0e 100644 --- a/src/plugins/runtime/runtime-tasks.test.ts +++ b/src/plugins/runtime/runtime-tasks.test.ts @@ -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(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", }); }); }); diff --git a/src/tasks/task-backing-authority-write.ts b/src/tasks/task-backing-authority-write.ts new file mode 100644 index 000000000000..7d3ec5a5911a --- /dev/null +++ b/src/tasks/task-backing-authority-write.ts @@ -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"; + } +} diff --git a/src/tasks/task-backing-authority.test-support.ts b/src/tasks/task-backing-authority.test-support.ts new file mode 100644 index 000000000000..c607a595aad3 --- /dev/null +++ b/src/tasks/task-backing-authority.test-support.ts @@ -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 }; +} diff --git a/src/tasks/task-backing-authority.ts b/src/tasks/task-backing-authority.ts new file mode 100644 index 000000000000..eee91be6caf0 --- /dev/null +++ b/src/tasks/task-backing-authority.ts @@ -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), + ); +} diff --git a/src/tasks/task-executor.test.ts b/src/tasks/task-executor.test.ts index f6d51c8cf317..ad4458e4d3df 100644 --- a/src/tasks/task-executor.test.ts +++ b/src/tasks/task-executor.test.ts @@ -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[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. */ diff --git a/src/tasks/task-executor.ts b/src/tasks/task-executor.ts index d0e8453dcc85..83b833493f06 100644 --- a/src/tasks/task-executor.ts +++ b/src/tasks/task-executor.ts @@ -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, diff --git a/src/tasks/task-registry-cancel.ts b/src/tasks/task-registry-cancel.ts index eba95a6ce7be..a19ca4203d3a 100644 --- a/src/tasks/task-registry-cancel.ts +++ b/src/tasks/task-registry-cancel.ts @@ -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) { diff --git a/src/tasks/task-registry-control.types.ts b/src/tasks/task-registry-control.types.ts index 6caf10822ccd..18a31af884ed 100644 --- a/src/tasks/task-registry-control.types.ts +++ b/src/tasks/task-registry-control.types.ts @@ -11,6 +11,9 @@ type CancelAcpSessionAdmin = (params: { cfg: OpenClawConfig; sessionKey: string; reason: string; + expectedRunId?: string; + expectedInstanceId?: string; + expectedOwnerKey?: string; }) => Promise; type KillSubagentRunAdminResult = @@ -28,6 +31,9 @@ type KillSubagentRunAdminResult = type KillSubagentRunAdmin = (params: { cfg: OpenClawConfig; sessionKey: string; + expectedRunId?: string; + expectedGeneration?: number; + expectedOwnerKey?: string; }) => Promise; export type TaskRegistryControlRuntime = { diff --git a/src/tasks/task-registry-create-helpers.ts b/src/tasks/task-registry-create-helpers.ts index 22b7937790e7..2ec4aaba2195 100644 --- a/src/tasks/task-registry-create-helpers.ts +++ b/src/tasks/task-registry-create-helpers.ts @@ -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) ?? "") === diff --git a/src/tasks/task-registry-lifecycle.ts b/src/tasks/task-registry-lifecycle.ts index b9db25223a08..e9d941362814 100644 --- a/src/tasks/task-registry-lifecycle.ts +++ b/src/tasks/task-registry-lifecycle.ts @@ -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)) { diff --git a/src/tasks/task-registry-record-api.ts b/src/tasks/task-registry-record-api.ts index 4a242a01984c..cd3bb3f0dc35 100644 --- a/src/tasks/task-registry-record-api.ts +++ b/src/tasks/task-registry-record-api.ts @@ -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 = {}; const nextStatus = params.status ? normalizeTaskStatus(params.status) : current.status; if ( diff --git a/test/e2e/qa-lab/runtime/webhooks-taskflow-cancellation-authz.e2e.test.ts b/test/e2e/qa-lab/runtime/webhooks-taskflow-cancellation-authz.e2e.test.ts new file mode 100644 index 000000000000..f1df4fad7e52 --- /dev/null +++ b/test/e2e/qa-lab/runtime/webhooks-taskflow-cancellation-authz.e2e.test.ts @@ -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, +): Promise { + 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 { + 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 { + 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(); + const releaseElicitation = createDeferred(); + 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(); + const releaseQueuedTarget = createDeferred(); + 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(); + const releaseQueuedSuccessor = createDeferred(); + 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); +});