From c1e714e3df9bec0b4850232cc07ddec225b2ad9f Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 19 Jul 2026 00:40:22 +0100 Subject: [PATCH] refactor(agents): shared spawn orchestration pipeline behind backend adapters (#111007) * refactor(agents): share spawn orchestration pipeline * fix(agents): preserve ACP spawn failure contract * fix(agents): progress hooks keep per-backend session-key semantics * fix(agents): ACP registration keeps the resolved requester agent id * chore(agents): drop orphaned exports surfaced by the spawn dedup --- src/agents/acp-spawn.test.ts | 47 +- src/agents/acp-spawn.ts | 741 ++++++------------- src/agents/spawn-pipeline.ts | 96 +++ src/agents/spawn-plan.ts | 396 ++++++++++ src/agents/subagent-spawn.runtime.ts | 4 - src/agents/subagent-spawn.ts | 680 +++++------------ src/agents/tools/sessions-spawn-tool.test.ts | 72 +- src/agents/tools/sessions-spawn-tool.ts | 102 +-- src/agents/tools/sessions-spawn-visible.ts | 26 +- 9 files changed, 975 insertions(+), 1189 deletions(-) create mode 100644 src/agents/spawn-pipeline.ts create mode 100644 src/agents/spawn-plan.ts diff --git a/src/agents/acp-spawn.test.ts b/src/agents/acp-spawn.test.ts index 63a6eea79333..b55d3ce9848f 100644 --- a/src/agents/acp-spawn.test.ts +++ b/src/agents/acp-spawn.test.ts @@ -70,7 +70,7 @@ const hoisted = vi.hoisted(() => { return normalized || null; }); const cleanupFailedAcpSpawnMock = vi.fn(); - const createRunningTaskRunMock = vi.fn(); + const registerSubagentRunMock = vi.fn(); const countActiveRunsForSessionMock = vi.fn(); const getSubagentRunByChildSessionKeyMock = vi.fn(); const listTasksForOwnerKeyMock = vi.fn(); @@ -158,7 +158,7 @@ const hoisted = vi.hoisted(() => { getLoadedChannelPluginMock, normalizeChannelIdMock, cleanupFailedAcpSpawnMock, - createRunningTaskRunMock, + registerSubagentRunMock, countActiveRunsForSessionMock, getSubagentRunByChildSessionKeyMock, listTasksForOwnerKeyMock, @@ -222,10 +222,6 @@ vi.mock("../infra/heartbeat-wake.js", () => ({ areHeartbeatsEnabled: hoisted.areHeartbeatsEnabledMock, })); -vi.mock("../tasks/detached-task-runtime.js", () => ({ - createRunningTaskRun: hoisted.createRunningTaskRunMock, -})); - vi.mock("./acp-spawn-parent-stream.js", () => ({ startAcpSpawnParentStreamRelay: hoisted.startAcpSpawnParentStreamRelayMock, })); @@ -233,6 +229,8 @@ vi.mock("./acp-spawn-parent-stream.js", () => ({ vi.mock("./subagent-registry.js", () => ({ countActiveRunsForSession: hoisted.countActiveRunsForSessionMock, getSubagentRunByChildSessionKey: hoisted.getSubagentRunByChildSessionKeyMock, + // ACP registration deliberately moved behind the shared spawn pipeline. + registerSubagentRun: hoisted.registerSubagentRunMock, })); vi.mock("../tasks/runtime-internal.js", () => ({ @@ -703,7 +701,7 @@ describe("spawnAcpDirect", () => { hoisted.getChannelPluginMock.mockReset().mockReturnValue(undefined); hoisted.getLoadedChannelPluginMock.mockReset().mockReturnValue(undefined); hoisted.cleanupFailedAcpSpawnMock.mockReset().mockResolvedValue(undefined); - hoisted.createRunningTaskRunMock.mockReset().mockReturnValue(undefined); + hoisted.registerSubagentRunMock.mockReset(); hoisted.countActiveRunsForSessionMock.mockReset().mockReturnValue(0); hoisted.getSubagentRunByChildSessionKeyMock.mockReset().mockReturnValue(null); hoisted.listTasksForOwnerKeyMock.mockReset().mockReturnValue([]); @@ -2262,6 +2260,15 @@ describe("spawnAcpDirect", () => { accountId: "bot-alpha", to: `room:${boundRoom}`, }); + expect(hoisted.registerSubagentRunMock).toHaveBeenCalledWith( + expect.objectContaining({ + requesterOrigin: expect.objectContaining({ + channel: "matrix", + accountId: "bot-alpha", + to: `room:${boundRoom}`, + }), + }), + ); }); it.each([ @@ -2885,9 +2892,9 @@ describe("spawnAcpDirect", () => { ); expectAcceptedSpawn(result); - expect(hoisted.createRunningTaskRunMock).toHaveBeenCalledWith( + expect(hoisted.registerSubagentRunMock).toHaveBeenCalledWith( expect.objectContaining({ - ownerKey: "global", + requesterSessionKey: "global", childSessionKey: expect.stringMatching(/^agent:codex:acp:/), agentId: "codex", requesterAgentId: "research", @@ -3213,6 +3220,28 @@ describe("spawnAcpDirect", () => { ); }); + it("preserves the ACP failure code when run registration fails", async () => { + hoisted.registerSubagentRunMock.mockImplementationOnce(() => { + throw new Error("registry unavailable"); + }); + + const result = await spawnAcpDirect( + { + task: "Investigate flaky tests", + agentId: "codex", + }, + { + agentSessionKey: "agent:main:main", + }, + ); + + const failed = expectFailedSpawn(result, "error"); + expect(failed.errorCode).toBe("spawn_failed"); + expect(failed.error).toContain("registry unavailable"); + expect(failed.runId).toBe("run-1"); + expect(hoisted.cleanupFailedAcpSpawnMock).toHaveBeenCalledTimes(1); + }); + it('rejects streamTo="parent" without requester session context', async () => { const result = await spawnAcpDirect( { diff --git a/src/agents/acp-spawn.ts b/src/agents/acp-spawn.ts index 5157c6f48f59..7454c3774007 100644 --- a/src/agents/acp-spawn.ts +++ b/src/agents/acp-spawn.ts @@ -6,10 +6,7 @@ import { resolveAcpThreadSessionDetailLines, } from "@openclaw/acp-core/runtime/session-identifiers"; import type { AcpRuntimeSessionMode } from "@openclaw/acp-core/runtime/types"; -import { - normalizeOptionalLowercaseString, - normalizeOptionalString, -} from "@openclaw/normalization-core/string-coerce"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { getAcpSessionManager } from "../acp/control-plane/manager.js"; import type { AcpTurnAttachment } from "../acp/control-plane/manager.types.js"; import { @@ -20,10 +17,6 @@ import { isAcpEnabledByPolicy, resolveAcpAgentPolicyError } from "../acp/policy. import { readAcpSessionMeta } from "../acp/runtime/session-meta.js"; import { DEFAULT_HEARTBEAT_EVERY } from "../auto-reply/heartbeat.js"; import { formatThinkingLevels } from "../auto-reply/thinking.js"; -import { - resolveChannelDefaultBindingPlacement, - resolveInboundConversationResolution, -} from "../channels/conversation-resolution.js"; import { formatConversationTarget, routeFromBindingRecord, @@ -34,17 +27,10 @@ import { resolveThreadBindingThreadName, } from "../channels/thread-bindings-messages.js"; import { - formatThreadBindingDisabledError, - formatThreadBindingSpawnDisabledError, resolveThreadBindingIdleTimeoutMsForChannel, resolveThreadBindingMaxAgeMsForChannel, - resolveThreadBindingSpawnPolicy, } from "../channels/thread-bindings-policy.js"; import { parseDurationMs } from "../cli/parse-duration.js"; -import { - DEFAULT_SUBAGENT_MAX_CHILDREN_PER_AGENT, - DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH, -} from "../config/agent-limits.js"; import { getRuntimeConfig } from "../config/config.js"; import { resolveStorePath } from "../config/sessions/paths.js"; import { @@ -64,6 +50,7 @@ import { type SessionBindingRecord, } from "../infra/outbound/session-binding-service.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; +import { getGlobalHookRunner } from "../plugins/hook-runner-global.js"; import { isSubagentSessionKey, normalizeAgentId, @@ -72,7 +59,6 @@ import { resolveAgentIdFromSessionKey, } from "../routing/session-key.js"; import { recordSubagentSpawned } from "../sessions/session-state-events.js"; -import { createRunningTaskRun } from "../tasks/detached-task-runtime.js"; import { listTasksForOwnerKey } from "../tasks/runtime-internal.js"; import { deliveryContextFromSession, normalizeDeliveryContext } from "../utils/delivery-context.js"; import { @@ -94,22 +80,35 @@ import { resolveThinkingDefault, } from "./model-selection.js"; import { resolveSandboxRuntimeStatus } from "./sandbox/runtime-status.js"; +import { + runSpawnPipeline, + type SpawnBackendAdapter, + summarizeSpawnError, +} from "./spawn-pipeline.js"; +import { + mintSpawnSessionKey, + prepareSpawnThreadBinding, + resolveConversationRefForThreadBinding, + resolveSpawnAdmission, + resolveSpawnChannelAccountId, + resolveSpawnMode, + resolveSpawnSandboxError, + type PreparedSpawnThreadBinding, +} from "./spawn-plan.js"; import { resolveRequesterOriginForChild } from "./spawn-requester-origin.js"; import { resolveSpawnedWorkspaceInheritance } from "./spawned-context.js"; import { isSubagentEnvelopeSession, - resolveSubagentCapabilities, resolveSubagentCapabilityStore, type SessionCapabilityStore, } from "./subagent-capabilities.js"; -import { getSubagentDepthFromSessionStore } from "./subagent-depth.js"; -import { countActiveRunsForSession, getSubagentRunByChildSessionKey } from "./subagent-registry.js"; +import { getSubagentRunByChildSessionKey } from "./subagent-registry.js"; +import { resolveSubagentSpawnOwnership } from "./subagent-spawn-ownership.js"; import { resolveConfiguredSubagentRunTimeoutSeconds, splitModelRef, } from "./subagent-spawn-plan.js"; import { resolveSubagentThinkingOverride } from "./subagent-spawn-thinking.js"; -import { resolveSubagentTargetPolicy } from "./subagent-target-policy.js"; import { resolveInternalSessionKey, resolveMainSessionAlias } from "./tools/sessions-helpers.js"; const log = createSubsystemLogger("agents/acp-spawn"); @@ -125,6 +124,7 @@ type SpawnAcpStreamTarget = (typeof ACP_SPAWN_STREAM_TARGETS)[number]; type SpawnAcpParams = { task: string; + taskName?: string; label?: string; agentId?: string; resumeSessionId?: string; @@ -135,6 +135,8 @@ type SpawnAcpParams = { mode?: SpawnAcpMode; thread?: boolean; sandbox?: SpawnAcpSandboxMode; + cleanup?: "delete" | "keep"; + expectsCompletionMessage?: boolean; streamTo?: SpawnAcpStreamTarget; attachments?: AcpTurnAttachment[]; }; @@ -166,11 +168,16 @@ function toGatewayImageAttachments( export type SpawnAcpContext = { agentSessionKey?: string; + requesterTurnRunId?: string; + completionOwnerKey?: string; requesterAgentIdOverride?: string; agentChannel?: string; agentAccountId?: string; agentTo?: string; agentThreadId?: string | number; + currentMessagingTarget?: string; + currentChannelId?: string; + currentMessageId?: string | number; /** Group chat ID for channels that distinguish group vs. topic (e.g. Telegram). */ agentGroupId?: string; /** Group space label (guild/team id) from the originating channel context. */ @@ -244,23 +251,13 @@ export function resolveAcpSpawnRuntimePolicyError(params: { sessionKey: params.requesterSessionKey, }); const requesterSandboxed = params.requesterSandboxed === true || requesterRuntime.sandboxed; - if (requesterSandboxed) { - return 'Sandboxed sessions cannot spawn ACP sessions because runtime="acp" runs on the host. Use runtime="subagent" from sandboxed sessions.'; - } - if (sandboxMode === "require") { - return 'sessions_spawn sandbox="require" is unsupported for runtime="acp" because ACP sessions run outside the sandbox. Use runtime="subagent" or sandbox="inherit".'; - } - return undefined; + return resolveSpawnSandboxError({ + backend: "acp", + requesterSandboxed, + sandbox: sandboxMode, + }); } -type PreparedAcpThreadBinding = { - channel: string; - accountId: string; - placement: "current" | "child"; - conversationId: string; - parentConversationId?: string; -}; - type AcpSpawnInitializedSession = Awaited< ReturnType["initializeSession"]> >; @@ -288,15 +285,6 @@ type AcpSpawnStreamPlan = { effectiveStreamToParent: boolean; }; -type AcpSubagentEnvelopeState = { - childSessionPatch?: { - spawnDepth: number; - subagentRole: "orchestrator" | "leaf" | null; - subagentControlScope: "children" | "none"; - }; - error?: string; -}; - function isActiveTaskStatus(status: string | undefined): boolean { return status === "queued" || status === "running"; } @@ -342,23 +330,6 @@ type AcpSpawnBootstrapDeliveryPlan = { threadId?: string; }; -function resolvePlacementWithoutChannelPlugin(params: { - capabilities: { placements: Array<"current" | "child"> }; -}): "current" | "child" { - return params.capabilities.placements.includes("child") ? "child" : "current"; -} - -function resolveSpawnMode(params: { - requestedMode?: SpawnAcpMode; - threadRequested: boolean; -}): SpawnAcpMode { - if (params.requestedMode === "run" || params.requestedMode === "session") { - return params.requestedMode; - } - // Thread-bound spawns should default to persistent sessions. - return params.threadRequested ? "session" : "run"; -} - function resolveAcpSessionMode(mode: SpawnAcpMode): AcpRuntimeSessionMode { return mode === "session" ? "persistent" : "oneshot"; } @@ -545,12 +516,14 @@ function createAcpSpawnFailure(params: { errorCode: SpawnAcpErrorCode; error: string; childSessionKey?: string; + runId?: string; }): SpawnAcpFailedResult { return { status: params.status, errorCode: params.errorCode, error: params.error, ...(params.childSessionKey ? { childSessionKey: params.childSessionKey } : {}), + ...(params.runId ? { runId: params.runId } : {}), }; } @@ -627,143 +600,6 @@ async function persistAcpSpawnSessionFileBestEffort(params: { } } -function resolveConversationRefForThreadBinding(params: { - cfg: OpenClawConfig; - channel?: string; - accountId?: string; - to?: string; - threadId?: string | number; - groupId?: string; -}): { conversationId: string; parentConversationId?: string } | null { - const resolution = resolveInboundConversationResolution({ - cfg: params.cfg, - channel: params.channel, - accountId: params.accountId, - to: params.to, - threadId: params.threadId, - groupId: params.groupId, - isGroup: true, - }); - return resolution?.canonical ?? null; -} - -function resolveAcpSpawnChannelAccountId(params: { - cfg: OpenClawConfig; - channel?: string; - accountId?: string; -}): string | undefined { - const channel = normalizeOptionalLowercaseString(params.channel); - const explicitAccountId = normalizeOptionalString(params.accountId); - if (explicitAccountId) { - return explicitAccountId; - } - if (!channel) { - return undefined; - } - const channels = params.cfg.channels as Record; - const configuredDefaultAccountId = channels?.[channel]?.defaultAccount; - return normalizeOptionalString(configuredDefaultAccountId) ?? "default"; -} - -function prepareAcpThreadBinding(params: { - cfg: OpenClawConfig; - channel?: string; - accountId?: string; - to?: string; - threadId?: string | number; - groupId?: string; -}): { ok: true; binding: PreparedAcpThreadBinding } | { ok: false; error: string } { - const channel = normalizeOptionalLowercaseString(params.channel); - if (!channel) { - return { - ok: false, - error: "thread=true for ACP sessions requires a channel context.", - }; - } - - const accountId = resolveAcpSpawnChannelAccountId({ - cfg: params.cfg, - channel, - accountId: params.accountId, - }); - const policy = resolveThreadBindingSpawnPolicy({ - cfg: params.cfg, - channel, - accountId, - kind: "acp", - }); - if (!policy.enabled) { - return { - ok: false, - error: formatThreadBindingDisabledError({ - channel: policy.channel, - accountId: policy.accountId, - kind: "acp", - }), - }; - } - if (!policy.spawnEnabled) { - return { - ok: false, - error: formatThreadBindingSpawnDisabledError({ - channel: policy.channel, - accountId: policy.accountId, - kind: "acp", - }), - }; - } - const bindingService = getSessionBindingService(); - const capabilities = bindingService.getCapabilities({ - channel: policy.channel, - accountId: policy.accountId, - }); - if (!capabilities.adapterAvailable) { - return { - ok: false, - error: `Thread bindings are unavailable for ${policy.channel}.`, - }; - } - const pluginPlacement = resolveChannelDefaultBindingPlacement(policy.channel); - const placementToUse = - pluginPlacement ?? - resolvePlacementWithoutChannelPlugin({ - capabilities, - }); - if (!capabilities.bindSupported || !capabilities.placements.includes(placementToUse)) { - return { - ok: false, - error: `Thread bindings do not support ${placementToUse} placement for ${policy.channel}.`, - }; - } - const conversationRef = resolveConversationRefForThreadBinding({ - cfg: params.cfg, - channel: policy.channel, - accountId: policy.accountId, - to: params.to, - threadId: params.threadId, - groupId: params.groupId, - }); - if (!conversationRef?.conversationId) { - return { - ok: false, - error: `Could not resolve a ${policy.channel} conversation for ACP thread spawn.`, - }; - } - - return { - ok: true, - binding: { - channel: policy.channel, - accountId: policy.accountId, - placement: placementToUse, - conversationId: conversationRef.conversationId, - ...(conversationRef.parentConversationId - ? { parentConversationId: conversationRef.parentConversationId } - : {}), - }, - }; -} - function resolveAcpSpawnRequesterState(params: { cfg: OpenClawConfig; parentSessionKey?: string; @@ -817,89 +653,6 @@ function resolveAcpSpawnRequesterState(params: { }; } -function resolveAcpSubagentEnvelopeState(params: { - cfg: OpenClawConfig; - requesterSessionKey?: string; - requesterAgentId: string; - targetAgentId: string; - requestedAgentId?: string; - subagentStore?: SessionCapabilityStore; -}): AcpSubagentEnvelopeState { - const requesterSessionKey = normalizeOptionalString(params.requesterSessionKey); - if (!requesterSessionKey) { - return {}; - } - if ( - !isSubagentEnvelopeSession(requesterSessionKey, { - cfg: params.cfg, - store: params.subagentStore, - }) - ) { - return {}; - } - - const callerDepth = getSubagentDepthFromSessionStore(requesterSessionKey, { - cfg: params.cfg, - }); - const maxSpawnDepth = - params.cfg.agents?.defaults?.subagents?.maxSpawnDepth ?? DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH; - if (callerDepth >= maxSpawnDepth) { - return { - error: `sessions_spawn is not allowed at this depth (current depth: ${callerDepth}, max: ${maxSpawnDepth})`, - }; - } - - const maxChildren = - params.cfg.agents?.defaults?.subagents?.maxChildrenPerAgent ?? - DEFAULT_SUBAGENT_MAX_CHILDREN_PER_AGENT; - const activeChildren = - countActiveRunsForSession(requesterSessionKey) + - countUntrackedActiveAcpRunsForOwner(requesterSessionKey); - if (activeChildren >= maxChildren) { - return { - error: `sessions_spawn has reached max active children for this session (${activeChildren}/${maxChildren})`, - }; - } - - const requireAgentId = - resolveAgentConfig(params.cfg, params.requesterAgentId)?.subagents?.requireAgentId ?? - params.cfg.agents?.defaults?.subagents?.requireAgentId ?? - false; - if (requireAgentId && !params.requestedAgentId?.trim()) { - return { - error: - "sessions_spawn requires explicit agentId when requireAgentId is configured. Use agents_list to see allowed agent ids.", - }; - } - - const targetPolicy = resolveSubagentTargetPolicy({ - requesterAgentId: params.requesterAgentId, - targetAgentId: params.targetAgentId, - requestedAgentId: params.requestedAgentId, - allowAgents: - resolveAgentConfig(params.cfg, params.requesterAgentId)?.subagents?.allowAgents ?? - params.cfg.agents?.defaults?.subagents?.allowAgents, - configuredAgentIds: resolveConfiguredAcpSubagentTargetIds(params.cfg), - }); - if (!targetPolicy.ok) { - return { - error: targetPolicy.error, - }; - } - - const childCapabilities = resolveSubagentCapabilities({ - depth: callerDepth + 1, - maxSpawnDepth, - }); - return { - childSessionPatch: { - spawnDepth: childCapabilities.depth, - subagentRole: childCapabilities.role === "main" ? null : childCapabilities.role, - subagentControlScope: childCapabilities.controlScope, - }, - }; -} - function resolveAcpSpawnStreamPlan(params: { spawnMode: SpawnAcpMode; requestThreadBinding: boolean; @@ -1120,7 +873,7 @@ async function bindPreparedAcpThread(params: { sessionKey: string; targetAgentId: string; label?: string; - preparedBinding: PreparedAcpThreadBinding; + preparedBinding: PreparedSpawnThreadBinding; initializedRuntime: AcpSpawnInitializedRuntime; }): Promise<{ binding: SessionBindingRecord; @@ -1217,7 +970,7 @@ function resolveAcpSpawnBootstrapDeliveryPlan(params: { threadId: fallbackThreadId, to: params.requester.origin?.to, }); - const requesterAccountId = resolveAcpSpawnChannelAccountId({ + const requesterAccountId = resolveSpawnChannelAccountId({ cfg: params.cfg, channel: params.requester.origin?.channel, accountId: params.requester.origin?.accountId, @@ -1382,19 +1135,27 @@ export async function spawnAcpDirect( ctx, subagentStore, }); - const subagentEnvelopeState = resolveAcpSubagentEnvelopeState({ + const hasSubagentEnvelope = isSubagentEnvelopeSession(requesterInternalKey, { cfg, + store: subagentStore, + }); + const admission = resolveSpawnAdmission({ + cfg, + enabled: hasSubagentEnvelope, requesterSessionKey: requesterInternalKey, requesterAgentId, targetAgentId, requestedAgentId: params.agentId, - subagentStore, + configuredAgentIds: resolveConfiguredAcpSubagentTargetIds(cfg), + additionalActiveChildren: hasSubagentEnvelope + ? countUntrackedActiveAcpRunsForOwner(requesterInternalKey) + : 0, }); - if (subagentEnvelopeState.error) { + if (!admission.ok) { return createAcpSpawnFailure({ status: "forbidden", errorCode: "subagent_policy", - error: subagentEnvelopeState.error, + error: admission.error, }); } const resumeAuthorization = validateAcpResumeSessionOwnership({ @@ -1432,7 +1193,7 @@ export async function spawnAcpDirect( requester: requesterState, }); - const sessionKey = `agent:${targetAgentId}:acp:${crypto.randomUUID()}`; + const sessionKey = mintSpawnSessionKey({ targetAgentId, backend: "acp" }); const runtimeMode = resolveAcpSessionMode(spawnMode); const resolvedCwd = resolveSpawnedWorkspaceInheritance({ config: cfg, @@ -1454,10 +1215,13 @@ export async function spawnAcpDirect( }); } - let preparedBinding: PreparedAcpThreadBinding | null = null; + let preparedBinding: PreparedSpawnThreadBinding | null = null; if (requestThreadBinding) { - const prepared = prepareAcpThreadBinding({ + const prepared = prepareSpawnThreadBinding({ cfg, + kind: "acp", + mode: spawnMode, + bindingService: getSessionBindingService(), channel: requesterState.origin?.channel, accountId: requesterState.origin?.accountId, to: requesterState.origin?.to, @@ -1474,80 +1238,9 @@ export async function spawnAcpDirect( preparedBinding = prepared.binding; } - let binding: SessionBindingRecord | null = null; let sessionCreated = false; - let childSessionId: string | undefined; let initializedRuntime: AcpSpawnRuntimeCloseHandle | undefined; - try { - await callGateway({ - method: "sessions.patch", - params: { - key: sessionKey, - spawnedBy: requesterInternalKey, - ...subagentEnvelopeState.childSessionPatch, - ...inheritedToolAllowPatch(ctx.inheritedToolAllowlist), - ...inheritedToolDenyPatch(ctx.inheritedToolDenylist), - ...(params.label ? { label: params.label } : {}), - }, - timeoutMs: 10_000, - }); - sessionCreated = true; - const initializedSession = await initializeAcpSpawnRuntime({ - cfg, - sessionKey, - targetAgentId, - runtimeMode, - resumeSessionId: params.resumeSessionId, - runtimeOptions: runtimeOptionsResult.runtimeOptions, - modelExplicit: runtimeOptionsResult.modelExplicit, - cwd: runtimeCwd, - }); - initializedRuntime = initializedSession.runtimeCloseHandle; - childSessionId = initializedSession.sessionId; - - if (preparedBinding) { - ({ binding } = await bindPreparedAcpThread({ - cfg, - sessionKey, - targetAgentId, - label: params.label, - preparedBinding, - initializedRuntime: initializedSession, - })); - } - } catch (err) { - await cleanupFailedAcpSpawn({ - cfg, - sessionKey, - shouldDeleteSession: sessionCreated, - deleteTranscript: true, - runtimeCloseHandle: initializedRuntime, - }); - return createAcpSpawnFailure({ - status: "error", - errorCode: isSessionBindingError(err) ? "thread_binding_invalid" : "spawn_failed", - error: isSessionBindingError(err) ? err.message : summarizeError(err), - }); - } - - const deliveryPlan = resolveAcpSpawnBootstrapDeliveryPlan({ - cfg, - spawnMode, - requestThreadBinding, - effectiveStreamToParent, - requester: requesterState, - binding, - }); const childIdem = crypto.randomUUID(); - let childRunId: string = childIdem; - // ACP children take this branch instead of spawnSubagentDirect; without this the - // signal log has no child_spawned event and the parent cursor is never seeded. - recordSubagentSpawned({ - childSessionKey: sessionKey, - childRunId: childIdem, - requesterSessionKey: requesterInternalKey, - agentId: targetAgentId, - }); const parentAgentId = parentSessionKey ? resolveAgentIdFromSessionKey(parentSessionKey) : undefined; @@ -1564,160 +1257,214 @@ export async function spawnAcpDirect( ) : undefined; - let parentRelay: AcpSpawnParentRelayHandle | undefined; const parentRelayStateEnv = { ...process.env }; const parentEventRouting = parentSessionKey ? resolveEventSessionRoutingPolicy({ cfg, sessionKey: parentSessionKey }) : undefined; - if (effectiveStreamToParent && parentSessionKey) { - // Register relay before dispatch so fast lifecycle failures are not missed. - parentRelay = startAcpSpawnParentStreamRelay({ - runId: childIdem, - parentSessionKey, - childSessionKey: sessionKey, - childSessionId, - agentId: targetAgentId, - env: parentRelayStateEnv, - mainKey: cfg.session?.mainKey, - sessionScope: cfg.session?.scope, - eventRouting: parentEventRouting, - deliveryContext: parentDeliveryCtx, - emitStartNotice: false, - cfg, - }); - } const gatewayAttachments = toGatewayImageAttachments(params.attachments); - try { - const response = await callGateway({ - method: "agent", - params: { - message: params.task, - sessionKey, - channel: deliveryPlan.channel, - to: deliveryPlan.to, - accountId: deliveryPlan.accountId, - threadId: deliveryPlan.threadId, - idempotencyKey: childIdem, - deliver: deliveryPlan.useInlineDelivery, - lane: AGENT_LANE_SUBAGENT, - acpTurnSource: "manual_spawn", - timeout: runTimeoutSeconds, - label: params.label || undefined, - ...(gatewayAttachments ? { attachments: gatewayAttachments } : {}), - }, - timeoutMs: 10_000, - }); - const responseRunId = normalizeOptionalString(response?.runId); - if (responseRunId) { - childRunId = responseRunId; - } - } catch (err) { - parentRelay?.dispose(); - await cleanupFailedAcpSpawn({ - cfg, - sessionKey, - shouldDeleteSession: true, - deleteTranscript: true, - runtimeCloseHandle: initializedRuntime, - }); - return createAcpSpawnFailure({ - status: "error", - errorCode: "dispatch_failed", - error: summarizeError(err), - childSessionKey: sessionKey, - }); - } - - if (effectiveStreamToParent && parentSessionKey) { - if (parentRelay && childRunId !== childIdem) { - parentRelay.dispose(); - // Defensive fallback if gateway returns a runId that differs from idempotency key. - parentRelay = startAcpSpawnParentStreamRelay({ - runId: childRunId, - parentSessionKey, - childSessionKey: sessionKey, - childSessionId, - agentId: targetAgentId, - env: parentRelayStateEnv, - mainKey: cfg.session?.mainKey, - sessionScope: cfg.session?.scope, - eventRouting: parentEventRouting, - deliveryContext: parentDeliveryCtx, - emitStartNotice: false, + const ownership = resolveSubagentSpawnOwnership({ + cfg, + agentSessionKey: ctx.agentSessionKey, + completionOwnerKey: ctx.completionOwnerKey, + }); + const requesterOrigin = requesterState.origin; + const progressOrigin = { + channel: requesterOrigin?.channel, + accountId: requesterOrigin?.accountId, + to: ctx.currentMessagingTarget ?? ctx.currentChannelId ?? requesterOrigin?.to, + threadId: requesterOrigin?.threadId, + channelId: ctx.currentChannelId, + messageId: ctx.currentMessageId, + }; + type AcpBackendState = { + initializedSession: AcpSpawnInitializedRuntime; + binding: SessionBindingRecord | null; + deliveryPlan?: AcpSpawnBootstrapDeliveryPlan; + parentRelay?: AcpSpawnParentRelayHandle; + }; + const adapter: SpawnBackendAdapter = { + async initialize() { + await callGateway({ + method: "sessions.patch", + params: { + key: sessionKey, + spawnedBy: requesterInternalKey, + ...admission.childSessionPatch, + ...inheritedToolAllowPatch(ctx.inheritedToolAllowlist), + ...inheritedToolDenyPatch(ctx.inheritedToolDenylist), + ...(params.label ? { label: params.label } : {}), + }, + timeoutMs: 10_000, + }); + sessionCreated = true; + const initializedSession = await initializeAcpSpawnRuntime({ cfg, + sessionKey, + targetAgentId, + runtimeMode, + resumeSessionId: params.resumeSessionId, + runtimeOptions: runtimeOptionsResult.runtimeOptions, + modelExplicit: runtimeOptionsResult.modelExplicit, + cwd: runtimeCwd, }); - } - parentRelay?.notifyStarted(); - try { - const task = createRunningTaskRun({ - runtime: "acp", - sourceId: childRunId, - ownerKey: requesterInternalKey, - scopeKind: "session", - requesterOrigin: requesterState.origin, + initializedRuntime = initializedSession.runtimeCloseHandle; + const binding = preparedBinding + ? ( + await bindPreparedAcpThread({ + cfg, + sessionKey, + targetAgentId, + label: params.label, + preparedBinding, + initializedRuntime: initializedSession, + }) + ).binding + : null; + return { initializedSession, binding }; + }, + async dispatchTurn(state) { + state.deliveryPlan = resolveAcpSpawnBootstrapDeliveryPlan({ + cfg, + spawnMode, + requestThreadBinding, + effectiveStreamToParent, + requester: requesterState, + binding: state.binding, + }); + // ACP bypasses the native adapter, so seed the same child lineage before dispatch. + recordSubagentSpawned({ childSessionKey: sessionKey, + childRunId: childIdem, + requesterSessionKey: requesterInternalKey, agentId: targetAgentId, - requesterAgentId, - runId: childRunId, - label: params.label, - task: params.task, - preferMetadata: true, - deliveryStatus: requesterInternalKey ? "pending" : "parent_missing", - startedAt: Date.now(), }); - if (!task) { - log.warn("Failed to persist background task for ACP spawn", { - sessionKey, - runId: childRunId, + if (effectiveStreamToParent && parentSessionKey) { + state.parentRelay = startAcpSpawnParentStreamRelay({ + runId: childIdem, + parentSessionKey, + childSessionKey: sessionKey, + childSessionId: state.initializedSession.sessionId, + agentId: targetAgentId, + env: parentRelayStateEnv, + mainKey: cfg.session?.mainKey, + sessionScope: cfg.session?.scope, + eventRouting: parentEventRouting, + deliveryContext: parentDeliveryCtx, + emitStartNotice: false, + cfg, }); } - } catch (error) { - log.warn("Failed to create background task for ACP spawn", { + const response = await callGateway({ + method: "agent", + params: { + message: params.task, + sessionKey, + channel: state.deliveryPlan.channel, + to: state.deliveryPlan.to, + accountId: state.deliveryPlan.accountId, + threadId: state.deliveryPlan.threadId, + idempotencyKey: childIdem, + deliver: state.deliveryPlan.useInlineDelivery, + lane: AGENT_LANE_SUBAGENT, + acpTurnSource: "manual_spawn", + timeout: runTimeoutSeconds, + label: params.label || undefined, + ...(gatewayAttachments ? { attachments: gatewayAttachments } : {}), + }, + timeoutMs: 10_000, + }); + const runId = normalizeOptionalString(response?.runId) ?? childIdem; + if (state.parentRelay && runId !== childIdem && parentSessionKey) { + state.parentRelay.dispose(); + state.parentRelay = startAcpSpawnParentStreamRelay({ + runId, + parentSessionKey, + childSessionKey: sessionKey, + childSessionId: state.initializedSession.sessionId, + agentId: targetAgentId, + env: parentRelayStateEnv, + mainKey: cfg.session?.mainKey, + sessionScope: cfg.session?.scope, + eventRouting: parentEventRouting, + deliveryContext: parentDeliveryCtx, + emitStartNotice: false, + cfg, + }); + } + state.parentRelay?.notifyStarted(); + return { runId }; + }, + async cleanupOnFailure({ state }) { + state?.parentRelay?.dispose(); + await cleanupFailedAcpSpawn({ + cfg, sessionKey, - runId: childRunId, - error, + shouldDeleteSession: sessionCreated, + deleteTranscript: true, + runtimeCloseHandle: initializedRuntime, + }); + }, + }; + const pipelineResult = await runSpawnPipeline({ + adapter, + hookRunner: getGlobalHookRunner(), + progressOrigin, + progressSessionKey: ownership.completionRequesterSessionKey, + buildRegistration: (state, runId) => { + const inlineDelivery = state.deliveryPlan?.useInlineDelivery === true; + return { + runId, + requesterTurnRunId: ctx.requesterTurnRunId, + childSessionKey: sessionKey, + controllerSessionKey: ownership.controllerSessionKey, + requesterSessionKey: ownership.completionRequesterSessionKey, + requesterOrigin, + progressOrigin, + requesterDisplayKey: ownership.completionRequesterDisplayKey, + task: params.task, + taskName: params.taskName, + agentId: targetAgentId, + requesterAgentId, + cleanup: spawnMode === "session" ? "keep" : params.cleanup === "delete" ? "delete" : "keep", + label: params.label, + runTimeoutSeconds, + expectsCompletionMessage: inlineDelivery + ? false + : params.expectsCompletionMessage !== false, + spawnMode, + }; + }, + }); + if (!pipelineResult.ok) { + if (pipelineResult.phase === "initialize") { + return createAcpSpawnFailure({ + status: "error", + errorCode: isSessionBindingError(pipelineResult.error) + ? "thread_binding_invalid" + : "spawn_failed", + error: isSessionBindingError(pipelineResult.error) + ? pipelineResult.error.message + : summarizeSpawnError(pipelineResult.error), }); } - return { - status: "accepted", - childSessionKey: sessionKey, - runId: childRunId, - mode: spawnMode, - runTimeoutSeconds, - note: spawnMode === "session" ? ACP_SPAWN_SESSION_ACCEPTED_NOTE : ACP_SPAWN_ACCEPTED_NOTE, - }; - } - - try { - const task = createRunningTaskRun({ - runtime: "acp", - sourceId: childRunId, - ownerKey: requesterInternalKey, - scopeKind: "session", - requesterOrigin: requesterState.origin, - childSessionKey: sessionKey, - agentId: targetAgentId, - requesterAgentId, - runId: childRunId, - label: params.label, - task: params.task, - preferMetadata: true, - deliveryStatus: requesterInternalKey ? "pending" : "parent_missing", - startedAt: Date.now(), - }); - if (!task) { - log.warn("Failed to persist background task for ACP spawn", { - sessionKey, - runId: childRunId, + if (pipelineResult.phase === "dispatch") { + return createAcpSpawnFailure({ + status: "error", + errorCode: "dispatch_failed", + error: summarizeSpawnError(pipelineResult.error), + childSessionKey: sessionKey, }); } - } catch (error) { - log.warn("Failed to create background task for ACP spawn", { - sessionKey, - runId: childRunId, - error, + return createAcpSpawnFailure({ + status: "error", + errorCode: "spawn_failed", + error: `Failed to register ACP run: ${summarizeSpawnError(pipelineResult.error)}. Cleanup was attempted, but the already-started ACP run may still finish in the background.`, + childSessionKey: sessionKey, + runId: pipelineResult.runId, }); } + const childRunId = pipelineResult.runId; + const deliveryPlan = pipelineResult.state.deliveryPlan; return { status: "accepted", @@ -1725,7 +1472,7 @@ export async function spawnAcpDirect( runId: childRunId, mode: spawnMode, runTimeoutSeconds, - ...(deliveryPlan.useInlineDelivery ? { inlineDelivery: true } : {}), + ...(deliveryPlan?.useInlineDelivery ? { inlineDelivery: true } : {}), note: spawnMode === "session" ? ACP_SPAWN_SESSION_ACCEPTED_NOTE : ACP_SPAWN_ACCEPTED_NOTE, }; } diff --git a/src/agents/spawn-pipeline.ts b/src/agents/spawn-pipeline.ts new file mode 100644 index 000000000000..f81d017f6747 --- /dev/null +++ b/src/agents/spawn-pipeline.ts @@ -0,0 +1,96 @@ +import type { SubagentLifecycleHookRunner } from "../plugins/hooks.js"; +import { registerSubagentRun } from "./subagent-registry.js"; + +type SpawnPipelinePhase = "initialize" | "dispatch" | "register"; + +export type SpawnBackendAdapter = { + initialize(): Promise; + dispatchTurn(state: TState): Promise<{ runId: string }>; + cleanupOnFailure(params: { + phase: SpawnPipelinePhase; + state?: TState; + error: unknown; + }): Promise; +}; + +type RegisterSubagentRunInput = Parameters[0]; + +type SpawnProgressOrigin = { + channel?: string; + accountId?: string; + to?: string; + threadId?: string | number; + channelId?: string; + messageId?: string | number; +}; + +type SpawnPipelineResult = + | { ok: true; state: TState; runId: string } + | { + ok: false; + phase: SpawnPipelinePhase; + error: unknown; + state?: TState; + runId?: string; + }; + +export function summarizeSpawnError(error: unknown): string { + return error instanceof Error ? error.message : typeof error === "string" ? error : "error"; +} + +export async function runSpawnPipeline(params: { + adapter: SpawnBackendAdapter; + buildRegistration: (state: TState, runId: string) => RegisterSubagentRunInput; + hookRunner?: SubagentLifecycleHookRunner | null; + progressOrigin?: SpawnProgressOrigin; + /** Session key the started-progress hook fires against. Backends differ on + purpose: native passes the controller-side requester key, ACP its + historical completion-owner key; do not collapse them. */ + progressSessionKey: string; +}): Promise> { + let state: TState; + try { + state = await params.adapter.initialize(); + } catch (error) { + await params.adapter.cleanupOnFailure({ phase: "initialize", error }); + return { ok: false, phase: "initialize", error }; + } + + let runId: string; + try { + ({ runId } = await params.adapter.dispatchTurn(state)); + } catch (error) { + await params.adapter.cleanupOnFailure({ phase: "dispatch", state, error }); + return { ok: false, phase: "dispatch", state, error }; + } + + const registration = params.buildRegistration(state, runId); + try { + registerSubagentRun(registration); + } catch (error) { + await params.adapter.cleanupOnFailure({ phase: "register", state, error }); + return { ok: false, phase: "register", state, runId, error }; + } + + if (params.hookRunner?.hasHooks("subagent_progress")) { + try { + await params.hookRunner.runSubagentProgress( + { + phase: "started", + runId, + childSessionKey: registration.childSessionKey, + requester: params.progressOrigin, + }, + { + runId, + childSessionKey: registration.childSessionKey, + requesterSessionKey: params.progressSessionKey, + }, + ); + } catch { + // Presentation hooks are best-effort after the run is durably registered. + } + } + + return { ok: true, state, runId }; +} diff --git a/src/agents/spawn-plan.ts b/src/agents/spawn-plan.ts new file mode 100644 index 000000000000..1c8510132b94 --- /dev/null +++ b/src/agents/spawn-plan.ts @@ -0,0 +1,396 @@ +import crypto from "node:crypto"; +import { + normalizeOptionalLowercaseString, + normalizeOptionalString, +} from "@openclaw/normalization-core/string-coerce"; +import { + resolveChannelDefaultBindingPlacement, + resolveInboundConversationResolution, +} from "../channels/conversation-resolution.js"; +import { + formatThreadBindingDisabledError, + formatThreadBindingSpawnDisabledError, + resolveThreadBindingSpawnPolicy, +} from "../channels/thread-bindings-policy.js"; +import { + DEFAULT_SUBAGENT_MAX_CHILDREN_PER_AGENT, + DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH, +} from "../config/agent-limits.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import type { getSessionBindingService } from "../infra/outbound/session-binding-service.js"; +import { resolveAgentConfig } from "./agent-scope.js"; +import { resolveSubagentCapabilities } from "./subagent-capabilities.js"; +import { getSubagentDepthFromSessionStore } from "./subagent-depth.js"; +import { countActiveRunsForSession } from "./subagent-registry.js"; +import { resolveSubagentTargetPolicy } from "./subagent-target-policy.js"; + +type SpawnMode = "run" | "session"; +type SpawnBackendKind = "subagent" | "acp"; + +export type PreparedSpawnThreadBinding = { + channel: string; + accountId: string; + placement: "current" | "child"; + conversationId: string; + parentConversationId?: string; +}; + +type SessionBindingService = ReturnType; + +export function resolveSpawnMode(params: { + requestedMode?: SpawnMode; + threadRequested: boolean; +}): SpawnMode { + if (params.requestedMode === "run" || params.requestedMode === "session") { + return params.requestedMode; + } + return params.threadRequested ? "session" : "run"; +} + +export function mintSpawnSessionKey(params: { + targetAgentId: string; + backend: SpawnBackendKind; +}): string { + const kind = params.backend === "acp" ? "acp" : "subagent"; + return `agent:${params.targetAgentId}:${kind}:${crypto.randomUUID()}`; +} + +export function resolveSpawnChannelAccountId(params: { + cfg: OpenClawConfig; + channel?: string; + accountId?: string; +}): string | undefined { + const channel = normalizeOptionalLowercaseString(params.channel); + const explicitAccountId = normalizeOptionalString(params.accountId); + if (explicitAccountId) { + return explicitAccountId; + } + if (!channel) { + return undefined; + } + const channels = params.cfg.channels as Record; + return normalizeOptionalString(channels?.[channel]?.defaultAccount) ?? "default"; +} + +export function resolveConversationRefForThreadBinding(params: { + cfg: OpenClawConfig; + channel?: string; + accountId?: string; + to?: string; + threadId?: string | number; + groupId?: string; +}): { conversationId: string; parentConversationId?: string } | null { + const resolution = resolveInboundConversationResolution({ + cfg: params.cfg, + channel: params.channel, + accountId: params.accountId, + to: params.to, + threadId: params.threadId, + groupId: params.groupId, + isGroup: true, + }); + return resolution?.canonical ?? null; +} + +function resolveRequesterBoundConversationRef(params: { + bindingService: SessionBindingService; + requesterSessionKey?: string; + channel: string; + accountId: string; + fallback?: { conversationId: string; parentConversationId?: string } | null; +}): { conversationId: string; parentConversationId?: string } | null | undefined { + const requesterSessionKey = normalizeOptionalString(params.requesterSessionKey); + if (!requesterSessionKey) { + return undefined; + } + const activeBindings = params.bindingService + .listBySession(requesterSessionKey) + .filter( + (record) => + record.status !== "ended" && + record.conversation.channel === params.channel && + (record.conversation.accountId ?? params.accountId) === params.accountId, + ); + if (activeBindings.length === 0) { + return undefined; + } + if (activeBindings.length === 1) { + const conversation = activeBindings[0]?.conversation; + return conversation + ? { + conversationId: conversation.conversationId, + ...(conversation.parentConversationId + ? { parentConversationId: conversation.parentConversationId } + : {}), + } + : undefined; + } + if (!params.fallback?.conversationId) { + return null; + } + const matched = activeBindings.filter( + (record) => + record.conversation.conversationId === params.fallback?.conversationId && + normalizeOptionalString(record.conversation.parentConversationId) === + normalizeOptionalString(params.fallback?.parentConversationId), + ); + const conversation = matched.length === 1 ? matched[0]?.conversation : undefined; + return conversation + ? { + conversationId: conversation.conversationId, + ...(conversation.parentConversationId + ? { parentConversationId: conversation.parentConversationId } + : {}), + } + : null; +} + +function buildThreadBindingUnavailableError(kind: SpawnBackendKind, mode: SpawnMode): string { + if (kind === "acp") { + return "thread=true for ACP sessions requires a channel context."; + } + if (mode === "session") { + return ( + 'sessions_spawn(mode="session") is only available on channels that expose thread bindings (e.g. Discord threads, Slack threads, Telegram forum topics). ' + + "This request is not running on a channel that can bind a subagent thread. " + + 'Use mode="run" for one-shot subagent work, or sessions_send(sessionKey=...) to keep talking to a persistent session without thread binding.' + ); + } + return ( + "thread=true is only available on channels that expose thread bindings (e.g. Discord threads, Slack threads, Telegram forum topics). " + + "This request is not running on a channel that can bind a subagent thread. " + + "Retry without thread=true, or re-run sessions_spawn from a channel that supports threads." + ); +} + +export function prepareSpawnThreadBinding(params: { + cfg: OpenClawConfig; + kind: SpawnBackendKind; + mode: SpawnMode; + bindingService: SessionBindingService; + requesterSessionKey?: string; + channel?: string; + accountId?: string; + to?: string; + threadId?: string | number; + groupId?: string; +}): { ok: true; binding: PreparedSpawnThreadBinding } | { ok: false; error: string } { + const channel = normalizeOptionalLowercaseString(params.channel); + if (!channel) { + return { ok: false, error: buildThreadBindingUnavailableError(params.kind, params.mode) }; + } + const accountId = resolveSpawnChannelAccountId({ + cfg: params.cfg, + channel, + accountId: params.accountId, + }); + const policy = resolveThreadBindingSpawnPolicy({ + cfg: params.cfg, + channel, + accountId, + kind: params.kind, + }); + if (!policy.enabled) { + return { + ok: false, + error: formatThreadBindingDisabledError({ + channel: policy.channel, + accountId: policy.accountId, + kind: params.kind, + }), + }; + } + if (!policy.spawnEnabled) { + return { + ok: false, + error: formatThreadBindingSpawnDisabledError({ + channel: policy.channel, + accountId: policy.accountId, + kind: params.kind, + }), + }; + } + const capabilities = params.bindingService.getCapabilities({ + channel: policy.channel, + accountId: policy.accountId, + }); + if (!capabilities.adapterAvailable) { + return { + ok: false, + error: + params.kind === "acp" + ? `Thread bindings are unavailable for ${policy.channel}.` + : buildThreadBindingUnavailableError(params.kind, params.mode), + }; + } + const placement = + resolveChannelDefaultBindingPlacement(policy.channel) ?? + (capabilities.placements.includes("child") ? "child" : "current"); + if (!capabilities.bindSupported || !capabilities.placements.includes(placement)) { + return { + ok: false, + error: `Thread bindings do not support ${placement} placement for ${policy.channel}.`, + }; + } + const fallback = resolveConversationRefForThreadBinding({ + cfg: params.cfg, + channel: policy.channel, + accountId: policy.accountId, + to: params.to, + threadId: params.threadId, + groupId: params.groupId, + }); + const requesterConversation = + params.kind === "subagent" + ? resolveRequesterBoundConversationRef({ + bindingService: params.bindingService, + requesterSessionKey: params.requesterSessionKey, + channel: policy.channel, + accountId: policy.accountId, + fallback, + }) + : undefined; + if (requesterConversation === null) { + return { + ok: false, + error: `Could not resolve a unique ${policy.channel} requester conversation for subagent thread spawn.`, + }; + } + const conversation = requesterConversation ?? fallback; + if (!conversation?.conversationId) { + return { + ok: false, + error: `Could not resolve a ${policy.channel} conversation for ${params.kind} thread spawn.`, + }; + } + return { + ok: true, + binding: { + channel: policy.channel, + accountId: policy.accountId, + placement, + conversationId: conversation.conversationId, + ...(conversation.parentConversationId + ? { parentConversationId: conversation.parentConversationId } + : {}), + }, + }; +} + +export function resolveSpawnAdmission(params: { + cfg: OpenClawConfig; + enabled?: boolean; + requesterSessionKey: string; + requesterAgentId: string; + targetAgentId: string; + requestedAgentId?: string; + configuredAgentIds: string[]; + additionalActiveChildren?: number; +}): + | { + ok: true; + callerDepth?: number; + maxSpawnDepth?: number; + childSessionPatch?: { + spawnDepth: number; + subagentRole: "orchestrator" | "leaf" | null; + subagentControlScope: "children" | "none"; + }; + } + | { ok: false; error: string } { + if (params.enabled === false) { + return { ok: true }; + } + const callerDepth = getSubagentDepthFromSessionStore(params.requesterSessionKey, { + cfg: params.cfg, + }); + const maxSpawnDepth = + params.cfg.agents?.defaults?.subagents?.maxSpawnDepth ?? DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH; + if (callerDepth >= maxSpawnDepth) { + return { + ok: false, + error: `sessions_spawn is not allowed at this depth (current depth: ${callerDepth}, max: ${maxSpawnDepth})`, + }; + } + const maxChildren = + params.cfg.agents?.defaults?.subagents?.maxChildrenPerAgent ?? + DEFAULT_SUBAGENT_MAX_CHILDREN_PER_AGENT; + const activeChildren = + countActiveRunsForSession(params.requesterSessionKey) + (params.additionalActiveChildren ?? 0); + if (activeChildren >= maxChildren) { + return { + ok: false, + error: `sessions_spawn has reached max active children for this session (${activeChildren}/${maxChildren})`, + }; + } + const requesterSubagentConfig = resolveAgentConfig( + params.cfg, + params.requesterAgentId, + )?.subagents; + const requireAgentId = + requesterSubagentConfig?.requireAgentId ?? + params.cfg.agents?.defaults?.subagents?.requireAgentId ?? + false; + if (requireAgentId && !params.requestedAgentId?.trim()) { + return { + ok: false, + error: + "sessions_spawn requires explicit agentId when requireAgentId is configured. Use agents_list to see allowed agent ids.", + }; + } + const targetPolicy = resolveSubagentTargetPolicy({ + requesterAgentId: params.requesterAgentId, + targetAgentId: params.targetAgentId, + requestedAgentId: params.requestedAgentId, + allowAgents: + requesterSubagentConfig?.allowAgents ?? params.cfg.agents?.defaults?.subagents?.allowAgents, + configuredAgentIds: params.configuredAgentIds, + }); + if (!targetPolicy.ok) { + return { ok: false, error: targetPolicy.error }; + } + const capabilities = resolveSubagentCapabilities({ + depth: callerDepth + 1, + maxSpawnDepth, + }); + return { + ok: true, + callerDepth, + maxSpawnDepth, + childSessionPatch: { + spawnDepth: capabilities.depth, + subagentRole: capabilities.role === "main" ? null : capabilities.role, + subagentControlScope: capabilities.controlScope, + }, + }; +} + +export function resolveSpawnSandboxError( + params: + | { + backend: "acp"; + requesterSandboxed: boolean; + sandbox: "inherit" | "require"; + } + | { + backend: "subagent"; + requesterSandboxed: boolean; + childSandboxed: boolean; + sandbox: "inherit" | "require"; + }, +): string | undefined { + if (params.backend === "acp") { + if (params.requesterSandboxed) { + return 'Sandboxed sessions cannot spawn ACP sessions because runtime="acp" runs on the host. Use runtime="subagent" from sandboxed sessions.'; + } + return params.sandbox === "require" + ? 'sessions_spawn sandbox="require" is unsupported for runtime="acp" because ACP sessions run outside the sandbox. Use runtime="subagent" or sandbox="inherit".' + : undefined; + } + if (params.childSandboxed || (!params.requesterSandboxed && params.sandbox !== "require")) { + return undefined; + } + return params.requesterSandboxed + ? "Sandboxed sessions cannot spawn unsandboxed subagents. Set a sandboxed target agent or use the same agent runtime." + : 'sessions_spawn sandbox="require" needs a sandboxed target runtime. Pick a sandboxed agentId or use sandbox="inherit".'; +} diff --git a/src/agents/subagent-spawn.runtime.ts b/src/agents/subagent-spawn.runtime.ts index ccbc338091c1..c4fb05b37da8 100644 --- a/src/agents/subagent-spawn.runtime.ts +++ b/src/agents/subagent-spawn.runtime.ts @@ -3,10 +3,6 @@ * single module lets spawn tests replace runtime seams without loading the * entire gateway/channel stack. */ -export { - DEFAULT_SUBAGENT_MAX_CHILDREN_PER_AGENT, - DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH, -} from "../config/agent-limits.js"; export { getRuntimeConfig } from "../config/config.js"; export { loadSessionEntry, upsertSessionEntry } from "../config/sessions/session-accessor.js"; export { forkSessionEntryFromParent } from "../auto-reply/reply/session-fork.js"; diff --git a/src/agents/subagent-spawn.ts b/src/agents/subagent-spawn.ts index b00c97a33f33..c9149bc8e13c 100644 --- a/src/agents/subagent-spawn.ts +++ b/src/agents/subagent-spawn.ts @@ -6,23 +6,14 @@ import crypto from "node:crypto"; import { promises as fs } from "node:fs"; import { finiteSecondsToTimerSafeMilliseconds } from "@openclaw/normalization-core/number-coercion"; -import { - normalizeOptionalLowercaseString, - normalizeOptionalString, -} from "@openclaw/normalization-core/string-coerce"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { isAcpRuntimeSpawnAvailable } from "../acp/runtime/availability.js"; -import { - resolveChannelDefaultBindingPlacement, - resolveInboundConversationResolution, -} from "../channels/conversation-resolution.js"; import { routeFromBindingRecord, routeToDeliveryFields } from "../channels/route-projection.js"; import { resolveThreadBindingIntroText, resolveThreadBindingThreadName, } from "../channels/thread-bindings-messages.js"; import { - formatThreadBindingDisabledError, - formatThreadBindingSpawnDisabledError, resolveThreadBindingIdleTimeoutMsForChannel, resolveThreadBindingMaxAgeMsForChannel, resolveThreadBindingSpawnPolicy, @@ -51,6 +42,18 @@ import { resolvePersistedSelectedModelRef, } from "./model-selection.js"; import { resolveThinkingDefault } from "./model-thinking-default.js"; +import { + runSpawnPipeline, + type SpawnBackendAdapter, + summarizeSpawnError, +} from "./spawn-pipeline.js"; +import { + mintSpawnSessionKey, + prepareSpawnThreadBinding, + resolveSpawnAdmission, + resolveSpawnMode, + resolveSpawnSandboxError, +} from "./spawn-plan.js"; import { resolveRequesterOriginForChild } from "./spawn-requester-origin.js"; import { mapToolContextToSpawnedRunMetadata, @@ -61,10 +64,7 @@ import { materializeSubagentAttachments, type SubagentAttachmentReceiptFile, } from "./subagent-attachments.js"; -import { resolveSubagentCapabilities } from "./subagent-capabilities.js"; -import { getSubagentDepthFromSessionStore } from "./subagent-depth.js"; import { buildSubagentInitialUserMessage } from "./subagent-initial-user-message.js"; -import { countActiveRunsForSession, registerSubagentRun } from "./subagent-registry.js"; import { resolveSubagentRunTimerDelayMs } from "./subagent-run-timeout.js"; import { resolveSubagentSpawnAcceptedNote } from "./subagent-spawn-accepted-note.js"; import { resolveSubagentSpawnOwnership } from "./subagent-spawn-ownership.js"; @@ -76,8 +76,6 @@ import { import { ADMIN_SCOPE, AGENT_LANE_SUBAGENT, - DEFAULT_SUBAGENT_MAX_CHILDREN_PER_AGENT, - DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH, buildSubagentSystemPrompt, callGateway, dispatchGatewayMethodInProcess, @@ -105,7 +103,6 @@ import type { SpawnSubagentMode, SpawnSubagentSandboxMode, } from "./subagent-spawn.types.js"; -import { resolveSubagentTargetPolicy } from "./subagent-target-policy.js"; import { normalizeSubagentTaskName } from "./subagent-task-name.js"; export { SUBAGENT_SPAWN_CONTEXT_MODES, SUBAGENT_SPAWN_MODES } from "./subagent-spawn.types.js"; @@ -668,17 +665,6 @@ async function cleanupFailedSpawnBeforeAgentStart(params: { }); } -function resolveSpawnMode(params: { - requestedMode?: SpawnSubagentMode; - threadRequested: boolean; -}): SpawnSubagentMode { - if (params.requestedMode === "run" || params.requestedMode === "session") { - return params.requestedMode; - } - // Thread-bound spawns should default to persistent sessions. - return params.threadRequested ? "session" : "run"; -} - function resolveSubagentContextMode(params: { requestedContext?: SpawnSubagentContextMode; threadRequested: boolean; @@ -712,242 +698,6 @@ function summarizeError(err: unknown): string { return "error"; } -function buildThreadBindingUnavailableError(mode: SpawnSubagentMode): string { - if (mode === "session") { - return ( - 'sessions_spawn(mode="session") is only available on channels that expose thread bindings (e.g. Discord threads, Slack threads, Telegram forum topics). ' + - "This request is not running on a channel that can bind a subagent thread. " + - 'Use mode="run" for one-shot subagent work, or sessions_send(sessionKey=...) to keep talking to a persistent session without thread binding.' - ); - } - return ( - "thread=true is only available on channels that expose thread bindings (e.g. Discord threads, Slack threads, Telegram forum topics). " + - "This request is not running on a channel that can bind a subagent thread. " + - "Retry without thread=true, or re-run sessions_spawn from a channel that supports threads." - ); -} - -type PreparedSubagentThreadBinding = { - channel: string; - accountId: string; - placement: "current" | "child"; - conversationId: string; - parentConversationId?: string; -}; - -function resolvePlacementWithoutChannelPlugin(params: { - capabilities: { placements: Array<"current" | "child"> }; -}): "current" | "child" { - return params.capabilities.placements.includes("child") ? "child" : "current"; -} - -function resolveSubagentSpawnChannelAccountId(params: { - cfg: OpenClawConfig; - channel?: string; - accountId?: string; -}): string | undefined { - const channel = normalizeOptionalLowercaseString(params.channel); - const explicitAccountId = normalizeOptionalString(params.accountId); - if (explicitAccountId) { - return explicitAccountId; - } - if (!channel) { - return undefined; - } - const channels = params.cfg.channels as Record; - return normalizeOptionalString(channels?.[channel]?.defaultAccount) ?? "default"; -} - -function resolveConversationRefForThreadBinding(params: { - cfg: OpenClawConfig; - channel?: string; - accountId?: string; - to?: string; - threadId?: string | number; -}): { conversationId: string; parentConversationId?: string } | null { - const resolution = resolveInboundConversationResolution({ - cfg: params.cfg, - channel: params.channel, - accountId: params.accountId, - to: params.to, - threadId: params.threadId, - isGroup: true, - }); - return resolution?.canonical ?? null; -} - -function resolveRequesterBoundConversationRef(params: { - requesterSessionKey?: string; - channel: string; - accountId: string; - fallback?: { conversationId: string; parentConversationId?: string } | null; -}): { conversationId: string; parentConversationId?: string } | null | undefined { - const requesterSessionKey = normalizeOptionalString(params.requesterSessionKey); - if (!requesterSessionKey) { - return undefined; - } - const activeBindings = getSessionBindingService() - .listBySession(requesterSessionKey) - .filter( - (record) => - record.status !== "ended" && - record.conversation.channel === params.channel && - (record.conversation.accountId ?? params.accountId) === params.accountId, - ); - if (activeBindings.length === 0) { - return undefined; - } - if (activeBindings.length === 1) { - const conversation = activeBindings.at(0)?.conversation; - if (!conversation) { - return undefined; - } - return { - conversationId: conversation.conversationId, - ...(conversation.parentConversationId - ? { parentConversationId: conversation.parentConversationId } - : {}), - }; - } - if (params.fallback?.conversationId) { - const matched = activeBindings.filter( - (record) => - record.conversation.conversationId === params.fallback?.conversationId && - normalizeOptionalString(record.conversation.parentConversationId) === - normalizeOptionalString(params.fallback?.parentConversationId), - ); - if (matched.length === 1) { - const conversation = matched.at(0)?.conversation; - if (!conversation) { - return undefined; - } - return { - conversationId: conversation.conversationId, - ...(conversation.parentConversationId - ? { parentConversationId: conversation.parentConversationId } - : {}), - }; - } - } - return null; -} - -function prepareSubagentThreadBinding(params: { - cfg: OpenClawConfig; - mode: SpawnSubagentMode; - requesterSessionKey?: string; - requester: { - channel?: string; - accountId?: string; - to?: string; - threadId?: string | number; - }; -}): { ok: true; binding: PreparedSubagentThreadBinding } | { ok: false; error: string } { - const channel = normalizeOptionalLowercaseString(params.requester.channel); - if (!channel) { - return { - ok: false, - error: buildThreadBindingUnavailableError(params.mode), - }; - } - - const accountId = resolveSubagentSpawnChannelAccountId({ - cfg: params.cfg, - channel, - accountId: params.requester.accountId, - }); - const policy = resolveThreadBindingSpawnPolicy({ - cfg: params.cfg, - channel, - accountId, - kind: "subagent", - }); - if (!policy.enabled) { - return { - ok: false, - error: formatThreadBindingDisabledError({ - channel: policy.channel, - accountId: policy.accountId, - kind: "subagent", - }), - }; - } - if (!policy.spawnEnabled) { - return { - ok: false, - error: formatThreadBindingSpawnDisabledError({ - channel: policy.channel, - accountId: policy.accountId, - kind: "subagent", - }), - }; - } - - const bindingService = getSessionBindingService(); - const capabilities = bindingService.getCapabilities({ - channel: policy.channel, - accountId: policy.accountId, - }); - if (!capabilities.adapterAvailable) { - return { - ok: false, - error: buildThreadBindingUnavailableError(params.mode), - }; - } - const pluginPlacement = resolveChannelDefaultBindingPlacement(policy.channel); - const placementToUse = - pluginPlacement ?? - resolvePlacementWithoutChannelPlugin({ - capabilities, - }); - if (!capabilities.bindSupported || !capabilities.placements.includes(placementToUse)) { - return { - ok: false, - error: `Thread bindings do not support ${placementToUse} placement for ${policy.channel}.`, - }; - } - - const fallbackConversationRef = resolveConversationRefForThreadBinding({ - cfg: params.cfg, - channel: policy.channel, - accountId: policy.accountId, - to: params.requester.to, - threadId: params.requester.threadId, - }); - const requesterConversationRef = resolveRequesterBoundConversationRef({ - requesterSessionKey: params.requesterSessionKey, - channel: policy.channel, - accountId: policy.accountId, - fallback: fallbackConversationRef, - }); - if (requesterConversationRef === null) { - return { - ok: false, - error: `Could not resolve a unique ${policy.channel} requester conversation for subagent thread spawn.`, - }; - } - const conversationRef = requesterConversationRef ?? fallbackConversationRef; - if (!conversationRef?.conversationId) { - return { - ok: false, - error: `Could not resolve a ${policy.channel} conversation for subagent thread spawn.`, - }; - } - - return { - ok: true, - binding: { - channel: policy.channel, - accountId: policy.accountId, - placement: placementToUse, - conversationId: conversationRef.conversationId, - ...(conversationRef.parentConversationId - ? { parentConversationId: conversationRef.parentConversationId } - : {}), - }, - }; -} - async function bindThreadForSubagentSpawn(params: { cfg: OpenClawConfig; childSessionKey: string; @@ -968,11 +718,16 @@ async function bindThreadForSubagentSpawn(params: { error: string; } > { - const prepared = prepareSubagentThreadBinding({ + const prepared = prepareSpawnThreadBinding({ cfg: params.cfg, + kind: "subagent", mode: params.mode, + bindingService: getSessionBindingService(), requesterSessionKey: params.requesterSessionKey, - requester: params.requester, + channel: params.requester.channel, + accountId: params.requester.accountId, + to: params.requester.to, + threadId: params.requester.threadId, }); if (!prepared.ok) { return { @@ -1130,41 +885,23 @@ export async function spawnSubagentDirect( completionOwnerKey: ctx.completionOwnerKey, }); - const callerDepth = getSubagentDepthFromSessionStore(requesterInternalKey, { cfg }); - const maxSpawnDepth = - cfg.agents?.defaults?.subagents?.maxSpawnDepth ?? DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH; - if (callerDepth >= maxSpawnDepth) { - return { - status: "forbidden", - error: `sessions_spawn is not allowed at this depth (current depth: ${callerDepth}, max: ${maxSpawnDepth})`, - }; - } - - const maxChildren = - cfg.agents?.defaults?.subagents?.maxChildrenPerAgent ?? DEFAULT_SUBAGENT_MAX_CHILDREN_PER_AGENT; - const activeChildren = countActiveRunsForSession(requesterInternalKey); - if (activeChildren >= maxChildren) { - return { - status: "forbidden", - error: `sessions_spawn has reached max active children for this session (${activeChildren}/${maxChildren})`, - }; - } - const requesterAgentId = normalizeAgentId( ctx.requesterAgentIdOverride ?? parseAgentSessionKey(requesterInternalKey)?.agentId, ); - const requireAgentId = - resolveAgentConfig(cfg, requesterAgentId)?.subagents?.requireAgentId ?? - cfg.agents?.defaults?.subagents?.requireAgentId ?? - false; - if (requireAgentId && !requestedAgentId?.trim()) { - return { - status: "forbidden", - error: - "sessions_spawn requires explicit agentId when requireAgentId is configured. Use agents_list to see allowed agent ids.", - }; - } const targetAgentId = requestedAgentId ? normalizeAgentId(requestedAgentId) : requesterAgentId; + const admission = resolveSpawnAdmission({ + cfg, + requesterSessionKey: requesterInternalKey, + requesterAgentId, + targetAgentId, + requestedAgentId, + configuredAgentIds: resolveConfiguredAgentIds(cfg), + }); + if (!admission.ok) { + return { status: "forbidden", error: admission.error }; + } + const childDepth = admission.childSessionPatch?.spawnDepth ?? 1; + const maxSpawnDepth = admission.maxSpawnDepth ?? childDepth; const requestedCwd = normalizeOptionalString(params.cwd); const spawnedCwd = requestedCwd ? resolveUserPath(requestedCwd) : undefined; const toolSpawnMetadata = mapToolContextToSpawnedRunMetadata({ @@ -1199,22 +936,7 @@ export async function spawnSubagentDirect( requesterGroupSpace: ctx.agentGroupSpace, requesterMemberRoleIds: ctx.agentMemberRoleIds, }); - const targetPolicy = resolveSubagentTargetPolicy({ - requesterAgentId, - targetAgentId, - requestedAgentId, - allowAgents: - resolveAgentConfig(cfg, requesterAgentId)?.subagents?.allowAgents ?? - cfg?.agents?.defaults?.subagents?.allowAgents, - configuredAgentIds: resolveConfiguredAgentIds(cfg), - }); - if (!targetPolicy.ok) { - return { - status: "forbidden", - error: targetPolicy.error, - }; - } - const childSessionKey = `agent:${targetAgentId}:subagent:${crypto.randomUUID()}`; + const childSessionKey = mintSpawnSessionKey({ targetAgentId, backend: "subagent" }); const requesterRuntime = resolveSandboxRuntimeStatus({ cfg, sessionKey: requesterInternalKey, @@ -1223,19 +945,14 @@ export async function spawnSubagentDirect( cfg, sessionKey: childSessionKey, }); - if (!childRuntime.sandboxed && (requesterRuntime.sandboxed || sandboxMode === "require")) { - if (requesterRuntime.sandboxed) { - return { - status: "forbidden", - error: - "Sandboxed sessions cannot spawn unsandboxed subagents. Set a sandboxed target agent or use the same agent runtime.", - }; - } - return { - status: "forbidden", - error: - 'sessions_spawn sandbox="require" needs a sandboxed target runtime. Pick a sandboxed agentId or use sandbox="inherit".', - }; + const sandboxError = resolveSpawnSandboxError({ + backend: "subagent", + requesterSandboxed: requesterRuntime.sandboxed, + childSandboxed: childRuntime.sandboxed, + sandbox: sandboxMode, + }); + if (sandboxError) { + return { status: "forbidden", error: sandboxError }; } const spawnedWorkspaceCwd = spawnedWorkspaceDir ? resolveUserPath(spawnedWorkspaceDir) @@ -1247,12 +964,7 @@ export async function spawnSubagentDirect( "cwd override is not supported for sandboxed subagent runs; omit cwd or use the target agent workspace as cwd", }; } - const childDepth = callerDepth + 1; const spawnedByKey = requesterInternalKey; - const childCapabilities = resolveSubagentCapabilities({ - depth: childDepth, - maxSpawnDepth, - }); const targetAgentDir = resolveAgentDir(cfg, targetAgentId); const requesterAgentConfig = resolveAgentConfig(cfg, requesterAgentId); const targetAgentConfig = resolveAgentConfig(cfg, targetAgentId); @@ -1299,9 +1011,7 @@ export async function spawnSubagentDirect( }; const initialChildSessionPatch: Record = { - spawnDepth: childDepth, - subagentRole: childCapabilities.role === "main" ? null : childCapabilities.role, - subagentControlScope: childCapabilities.controlScope, + ...admission.childSessionPatch, ...inheritedToolAllowPatch(ctx.inheritedToolAllowlist), ...inheritedToolDenyPatch(ctx.inheritedToolDenylist), ...plan.initialSessionPatch, @@ -1492,140 +1202,11 @@ export async function spawnSubagentDirect( requesterSessionKey: requesterInternalKey, agentId: targetAgentId, }); - const contextEnginePrepareResult = - params.lightContext && preparedSpawnContext.mode === "isolated" - ? ({ status: "ok", preparation: undefined } as const) - : await prepareContextEngineSubagentSpawn({ - cfg, - context: preparedSpawnContext, - requesterInternalKey, - childSessionKey, - runTimeoutSeconds, - }); - if (contextEnginePrepareResult.status === "error") { - await cleanupFailedSpawnBeforeAgentStart({ - childSessionKey, - attachmentAbsDir, - emitLifecycleHooks: threadBindingReady, - deleteTranscript: true, - }); - return { - status: "error", - error: contextEnginePrepareResult.error, - childSessionKey, - }; - } - const contextEnginePreparation = contextEnginePrepareResult.preparation; - const deliverInitialChildRunDirectly = requestThreadBinding && spawnMode === "session" && hasBoundThreadDeliveryOrigin; const shouldAnnounceCompletion = deliverInitialChildRunDirectly ? false : expectsCompletionMessage; - try { - const { - spawnedBy: _spawnedBy, - workspaceDir: _workspaceDir, - ...publicSpawnedMetadata - } = spawnedMetadata; - const response = await callSubagentGateway({ - method: "agent", - params: { - message: childTaskMessage, - sessionKey: childSessionKey, - channel: childSessionOrigin?.channel, - to: childSessionOrigin?.to ?? undefined, - accountId: childSessionOrigin?.accountId ?? undefined, - threadId: - childSessionOrigin?.threadId != null - ? stringifyRouteThreadId(childSessionOrigin.threadId) - : undefined, - idempotencyKey: childIdem, - deliver: deliverInitialChildRunDirectly, - lane: AGENT_LANE_SUBAGENT, - disableMessageTool: true, - cleanupBundleMcpOnRunEnd: spawnMode !== "session", - extraSystemPrompt: childSystemPrompt, - thinking: thinkingOverride, - timeout: runTimeoutSeconds, - label: label || undefined, - ...(bootstrapContextMode - ? { - bootstrapContextMode, - bootstrapContextRunKind: "default" as const, - } - : {}), - ...publicSpawnedMetadata, - }, - timeoutMs: resolveSubagentAgentGatewayTimeoutMs(runTimeoutSeconds), - }); - const runId = readGatewayRunId(response); - if (runId) { - childRunId = runId; - } - } catch (err) { - await rollbackPreparedContextEngine(contextEnginePreparation); - if (attachmentAbsDir) { - try { - await fs.rm(attachmentAbsDir, { recursive: true, force: true }); - } catch { - // Best-effort cleanup only. - } - } - let emitLifecycleHooks = false; - if (threadBindingReady) { - const hasEndedHook = hookRunner?.hasHooks("subagent_ended") === true; - let endedHookEmitted = false; - if (hasEndedHook) { - try { - await hookRunner?.runSubagentEnded( - { - targetSessionKey: childSessionKey, - targetKind: "subagent", - reason: "spawn-failed", - sendFarewell: true, - accountId: childSessionOrigin?.accountId, - runId: childRunId, - outcome: "error", - error: "Session failed to start", - }, - { - runId: childRunId, - childSessionKey, - requesterSessionKey: requesterInternalKey, - }, - ); - endedHookEmitted = true; - } catch { - // Spawn should still return an actionable error even if cleanup hooks fail. - } - } - emitLifecycleHooks = !endedHookEmitted; - } - // Always delete the provisional child session after a failed spawn attempt. - // If we already emitted subagent_ended above, suppress a duplicate lifecycle hook. - try { - await callSubagentGateway({ - method: "sessions.delete", - params: { - key: childSessionKey, - deleteTranscript: true, - emitLifecycleHooks, - }, - timeoutMs: SUBAGENT_CONTROL_GATEWAY_TIMEOUT_MS, - }); - } catch { - // Best-effort only. - } - const messageText = summarizeError(err); - return { - status: "error", - error: messageText, - childSessionKey, - runId: childRunId, - }; - } - const progressOrigin = { channel: requesterOrigin?.channel, accountId: requesterOrigin?.accountId, @@ -1634,9 +1215,123 @@ export async function spawnSubagentDirect( channelId: ctx.currentChannelId, messageId: ctx.currentMessageId, }; - try { - registerSubagentRun({ - runId: childRunId, + type SubagentBackendState = { contextEnginePreparation?: SubagentSpawnPreparation }; + const adapter: SpawnBackendAdapter = { + async initialize() { + const result = + params.lightContext && preparedSpawnContext.mode === "isolated" + ? ({ status: "ok", preparation: undefined } as const) + : await prepareContextEngineSubagentSpawn({ + cfg, + context: preparedSpawnContext, + requesterInternalKey, + childSessionKey, + runTimeoutSeconds, + }); + if (result.status === "error") { + throw new Error(result.error); + } + return { contextEnginePreparation: result.preparation }; + }, + async dispatchTurn() { + const { + spawnedBy: _spawnedBy, + workspaceDir: _workspaceDir, + ...publicSpawnedMetadata + } = spawnedMetadata; + const response = await callSubagentGateway({ + method: "agent", + params: { + message: childTaskMessage, + sessionKey: childSessionKey, + channel: childSessionOrigin?.channel, + to: childSessionOrigin?.to ?? undefined, + accountId: childSessionOrigin?.accountId ?? undefined, + threadId: + childSessionOrigin?.threadId != null + ? stringifyRouteThreadId(childSessionOrigin.threadId) + : undefined, + idempotencyKey: childIdem, + deliver: deliverInitialChildRunDirectly, + lane: AGENT_LANE_SUBAGENT, + disableMessageTool: true, + cleanupBundleMcpOnRunEnd: spawnMode !== "session", + extraSystemPrompt: childSystemPrompt, + thinking: thinkingOverride, + timeout: runTimeoutSeconds, + label: label || undefined, + ...(bootstrapContextMode + ? { + bootstrapContextMode, + bootstrapContextRunKind: "default" as const, + } + : {}), + ...publicSpawnedMetadata, + }, + timeoutMs: resolveSubagentAgentGatewayTimeoutMs(runTimeoutSeconds), + }); + return { runId: readGatewayRunId(response) ?? childIdem }; + }, + async cleanupOnFailure({ phase, state }) { + if (phase === "initialize") { + await cleanupFailedSpawnBeforeAgentStart({ + childSessionKey, + attachmentAbsDir, + emitLifecycleHooks: threadBindingReady, + deleteTranscript: true, + }); + return; + } + await rollbackPreparedContextEngine(state?.contextEnginePreparation); + if (attachmentAbsDir) { + try { + await fs.rm(attachmentAbsDir, { recursive: true, force: true }); + } catch { + // Best-effort cleanup only. + } + } + let emitLifecycleHooks = threadBindingReady; + if (phase === "dispatch" && threadBindingReady) { + let endedHookEmitted = false; + if (hookRunner?.hasHooks("subagent_ended")) { + try { + await hookRunner.runSubagentEnded( + { + targetSessionKey: childSessionKey, + targetKind: "subagent", + reason: "spawn-failed", + sendFarewell: true, + accountId: childSessionOrigin?.accountId, + runId: childIdem, + outcome: "error", + error: "Session failed to start", + }, + { + runId: childIdem, + childSessionKey, + requesterSessionKey: requesterInternalKey, + }, + ); + endedHookEmitted = true; + } catch { + // Spawn cleanup continues even when presentation hooks fail. + } + } + emitLifecycleHooks = !endedHookEmitted; + } + await cleanupProvisionalSession(childSessionKey, { + emitLifecycleHooks, + deleteTranscript: true, + }); + }, + }; + const pipelineResult = await runSpawnPipeline({ + adapter, + hookRunner, + progressOrigin, + progressSessionKey: requesterInternalKey, + buildRegistration: (_state, runId) => ({ + runId, requesterTurnRunId: ctx.requesterTurnRunId, childSessionKey, controllerSessionKey: ownership.controllerSessionKey, @@ -1659,56 +1354,21 @@ export async function spawnSubagentDirect( attachmentsDir: attachmentAbsDir, attachmentsRootDir: attachmentRootDir, retainAttachmentsOnKeep: retainOnSessionKeep, - }); - } catch (err) { - await rollbackPreparedContextEngine(contextEnginePreparation); - if (attachmentAbsDir) { - try { - await fs.rm(attachmentAbsDir, { recursive: true, force: true }); - } catch { - // Best-effort cleanup only. - } - } - try { - await callSubagentGateway({ - method: "sessions.delete", - params: { - key: childSessionKey, - deleteTranscript: true, - emitLifecycleHooks: threadBindingReady, - }, - timeoutMs: SUBAGENT_CONTROL_GATEWAY_TIMEOUT_MS, - }); - } catch { - // Best-effort cleanup only. - } + }), + }); + if (!pipelineResult.ok) { + const runId = pipelineResult.runId ?? childIdem; return { status: "error", - error: `Failed to register subagent run: ${summarizeError(err)}`, + error: + pipelineResult.phase === "register" + ? `Failed to register subagent run: ${summarizeSpawnError(pipelineResult.error)}` + : summarizeSpawnError(pipelineResult.error), childSessionKey, - runId: childRunId, + ...(pipelineResult.phase === "initialize" ? {} : { runId }), }; } - - if (hookRunner?.hasHooks("subagent_progress")) { - try { - await hookRunner.runSubagentProgress( - { - phase: "started", - runId: childRunId, - childSessionKey, - requester: progressOrigin, - }, - { - runId: childRunId, - childSessionKey, - requesterSessionKey: requesterInternalKey, - }, - ); - } catch { - // Progress presentation is best-effort and must not reject an accepted spawn. - } - } + childRunId = pipelineResult.runId; if (hookRunner?.hasHooks("subagent_spawned")) { try { diff --git a/src/agents/tools/sessions-spawn-tool.test.ts b/src/agents/tools/sessions-spawn-tool.test.ts index ee488d891913..0b30371b1d30 100644 --- a/src/agents/tools/sessions-spawn-tool.test.ts +++ b/src/agents/tools/sessions-spawn-tool.test.ts @@ -1189,40 +1189,19 @@ describe("sessions_spawn tool", () => { expect(spawnArgs).not.toHaveProperty("runTimeoutSeconds"); expect(spawnArgs.thread).toBe(true); expect(spawnArgs.mode).toBe("session"); + expect(spawnArgs.cleanup).toBe("keep"); + expect(spawnArgs.expectsCompletionMessage).toBe(true); expect(spawnArgs.streamTo).toBe("parent"); const spawnContext = mockCallArg(hoisted.spawnAcpDirectMock, 0, 1, "spawnAcpDirect"); expect(spawnContext.agentSessionKey).toBe("agent:main:main"); expect(spawnContext.requesterAgentIdOverride).toBe("main"); + expect(spawnContext.currentMessagingTarget).toBe("channel:source"); + expect(spawnContext.currentChannelId).toBe("source-native"); + expect(spawnContext.currentMessageId).toBe("message-789"); expect(hoisted.spawnSubagentDirectMock).not.toHaveBeenCalled(); - const registration = mockCallArg(hoisted.registerSubagentRunMock, 0, 0, "registerSubagentRun"); - expect(registration.runId).toBe("run-acp"); - expect(registration.childSessionKey).toBe("agent:codex:acp:1"); - expect(registration.requesterSessionKey).toBe("agent:main:main"); - expect(registration.requesterAgentId).toBe("main"); - expect(registration.task).toBe("investigate the failing CI run"); - expect(registration.cleanup).toBe("keep"); - expect(registration.spawnMode).toBe("session"); - expect(registration.expectsCompletionMessage).toBe(true); - expect(hoisted.runSubagentProgressMock).toHaveBeenCalledWith( - { - phase: "started", - runId: "run-acp", - childSessionKey: "agent:codex:acp:1", - requester: { - channel: "quietchat", - accountId: "default", - to: "channel:source", - threadId: "456", - channelId: "source-native", - messageId: "message-789", - }, - }, - { - runId: "run-acp", - childSessionKey: "agent:codex:acp:1", - requesterSessionKey: "agent:main:main", - }, - ); + // Registration and progress hooks now belong to the shared backend pipeline. + expect(hoisted.registerSubagentRunMock).not.toHaveBeenCalled(); + expect(hoisted.runSubagentProgressMock).not.toHaveBeenCalled(); }); it("passes inherited tool denies to ACP spawns", async () => { @@ -1409,19 +1388,13 @@ describe("sessions_spawn tool", () => { const spawnArgs = mockCallArg(hoisted.spawnAcpDirectMock, 0, 0, "spawnAcpDirect"); expect(spawnArgs.task).toBe("investigate"); expect(spawnArgs.sandbox).toBe("require"); + expect(spawnArgs.cleanup).toBe("keep"); const spawnContext = mockCallArg(hoisted.spawnAcpDirectMock, 0, 1, "spawnAcpDirect"); expect(spawnContext.agentSessionKey).toBe("agent:main:subagent:parent"); - const registration = mockCallArg(hoisted.registerSubagentRunMock, 0, 0, "registerSubagentRun"); - expect(registration.runId).toBe("run-acp"); - expect(registration.childSessionKey).toBe("agent:codex:acp:1"); - expect(registration.requesterSessionKey).toBe("agent:main:subagent:parent"); - expect(registration.task).toBe("investigate"); - expect(registration.cleanup).toBe("keep"); - expect(registration.runTimeoutSeconds).toBe(120); - expect(registration.spawnMode).toBe("run"); + expect(hoisted.registerSubagentRunMock).not.toHaveBeenCalled(); }); - it("suppresses completion announces for inline ACP session delivery", async () => { + it("forwards completion policy for inline ACP session delivery", async () => { registerAcpBackendForTest(); hoisted.spawnAcpDirectMock.mockResolvedValueOnce({ status: "accepted", @@ -1446,14 +1419,12 @@ describe("sessions_spawn tool", () => { mode: "session", }); - const registration = mockCallArg(hoisted.registerSubagentRunMock, 0, 0, "registerSubagentRun"); - expect(registration.runId).toBe("run-acp"); - expect(registration.childSessionKey).toBe("agent:codex:acp:1"); - expect(registration.requesterSessionKey).toBe("agent:main:main"); - expect(registration.task).toBe("investigate"); - expect(registration.cleanup).toBe("keep"); - expect(registration.spawnMode).toBe("session"); - expect(registration.expectsCompletionMessage).toBe(false); + const spawnArgs = mockCallArg(hoisted.spawnAcpDirectMock, 0, 0, "spawnAcpDirect"); + expect(spawnArgs.mode).toBe("session"); + expect(spawnArgs.cleanup).toBe("keep"); + expect(spawnArgs.expectsCompletionMessage).toBe(true); + // Inline-delivery suppression is decided after the ACP adapter binds its thread. + expect(hoisted.registerSubagentRunMock).not.toHaveBeenCalled(); }); it("rejects ACP runtime calls from sandboxed requester sessions", async () => { @@ -1761,7 +1732,7 @@ describe("sessions_spawn tool", () => { expect(spawnContext.completionOwnerKey).toBe("agent:main:main"); }); - it("uses completionOwnerKey for ACP registerSubagentRun requesterSessionKey", async () => { + it("forwards completionOwnerKey to the ACP registration pipeline", async () => { registerAcpBackendForTest(); const tool = createSessionsSpawnTool({ agentSessionKey: "agent:main:telegram:default:direct:456", @@ -1777,10 +1748,9 @@ describe("sessions_spawn tool", () => { agentId: "codex", }); - const registration = mockCallArg(hoisted.registerSubagentRunMock, 0, 0, "registerSubagentRun"); - expect(registration.controllerSessionKey).toBe("agent:main:telegram:default:direct:456"); - expect(registration.requesterSessionKey).toBe("agent:main:main"); - expect(registration.requesterDisplayKey).toBe("agent:main:main"); + const spawnContext = mockCallArg(hoisted.spawnAcpDirectMock, 0, 1, "spawnAcpDirect"); + expect(spawnContext.agentSessionKey).toBe("agent:main:telegram:default:direct:456"); + expect(spawnContext.completionOwnerKey).toBe("agent:main:main"); }); }); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/agents/tools/sessions-spawn-tool.ts b/src/agents/tools/sessions-spawn-tool.ts index 14cbaa8652da..7d80469bd3db 100644 --- a/src/agents/tools/sessions-spawn-tool.ts +++ b/src/agents/tools/sessions-spawn-tool.ts @@ -12,9 +12,7 @@ import { import { getRuntimeConfig } from "../../config/config.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { resolveSnakeCaseParamKey } from "../../param-key.js"; -import { getGlobalHookRunner } from "../../plugins/hook-runner-global.js"; import { createLazyImportLoader } from "../../shared/lazy-promise.js"; -import { normalizeDeliveryContext } from "../../utils/delivery-context.shared.js"; import type { GatewayMessageChannel } from "../../utils/message-channel.js"; import { findAcpUnsupportedInheritedToolAllow, @@ -25,8 +23,6 @@ import { import { optionalStringEnum } from "../schema/typebox.js"; import type { SpawnedToolContext } from "../spawned-context.js"; import { resolveAcpSessionsSpawnImageAttachments } from "../subagent-attachments.js"; -import { registerSubagentRun } from "../subagent-registry.js"; -import { resolveSubagentSpawnOwnership } from "../subagent-spawn-ownership.js"; import { SUBAGENT_SPAWN_CONTEXT_MODES, SUBAGENT_SPAWN_MODES, @@ -46,10 +42,7 @@ import { ToolInputError, } from "./common.js"; import { - cleanupUntrackedAcpSession, maybeSpawnVisibleSession, - resolveTrackedSpawnMode, - summarizeSessionsSpawnError, type VisibleSessionsSpawnDeps, VISIBLE_SESSIONS_SPAWN_SCHEMA, } from "./sessions-spawn-visible.js"; @@ -357,7 +350,7 @@ export function createSessionsSpawnTool( : undefined; if (runtime === "acp") { - const { isSpawnAcpAcceptedResult, spawnAcpDirect } = await loadAcpSpawnModule(); + const { spawnAcpDirect } = await loadAcpSpawnModule(); const acpAttachments = resolveAcpSessionsSpawnImageAttachments({ config: opts?.config ?? getRuntimeConfig(), attachments, @@ -372,6 +365,7 @@ export function createSessionsSpawnTool( const result = await spawnAcpDirect( { task, + taskName, label: label || undefined, agentId: requestedAgentId, resumeSessionId, @@ -381,16 +375,23 @@ export function createSessionsSpawnTool( mode: mode === "run" || mode === "session" ? mode : undefined, thread, sandbox, + cleanup, + expectsCompletionMessage, streamTo, attachments: acpAttachments?.attachments, }, { agentSessionKey: opts?.agentSessionKey, + requesterTurnRunId: opts?.requesterTurnRunId, + completionOwnerKey: opts?.completionOwnerKey, requesterAgentIdOverride: opts?.requesterAgentIdOverride, agentChannel: opts?.agentChannel, agentAccountId: opts?.agentAccountId, agentTo: opts?.agentTo, agentThreadId: opts?.agentThreadId, + currentMessagingTarget: opts?.currentMessagingTarget, + currentChannelId: opts?.currentChannelId, + currentMessageId: opts?.currentMessageId, agentGroupId: opts?.agentGroupId ?? undefined, agentGroupSpace: opts?.agentGroupSpace, agentMemberRoleIds: opts?.agentMemberRoleIds, @@ -399,91 +400,6 @@ export function createSessionsSpawnTool( inheritedToolDenylist: opts?.inheritedToolDenylist, }, ); - const childSessionKey = result.childSessionKey?.trim(); - const childRunId = isSpawnAcpAcceptedResult(result) ? result.runId?.trim() : undefined; - const shouldTrackViaRegistry = - result.status === "accepted" && Boolean(childSessionKey) && Boolean(childRunId); - if (shouldTrackViaRegistry && childSessionKey && childRunId) { - const cfg = getRuntimeConfig(); - const trackedSpawnMode = resolveTrackedSpawnMode({ - requestedMode: result.mode, - threadRequested: thread, - }); - const trackedCleanup = trackedSpawnMode === "session" ? "keep" : cleanup; - const ownership = resolveSubagentSpawnOwnership({ - cfg, - agentSessionKey: opts?.agentSessionKey, - completionOwnerKey: opts?.completionOwnerKey, - }); - const requesterOrigin = normalizeDeliveryContext({ - channel: opts?.agentChannel, - accountId: opts?.agentAccountId, - to: opts?.agentTo, - threadId: opts?.agentThreadId, - }); - const progressOrigin = { - channel: requesterOrigin?.channel, - accountId: requesterOrigin?.accountId, - to: opts?.currentMessagingTarget ?? opts?.currentChannelId ?? requesterOrigin?.to, - threadId: requesterOrigin?.threadId, - channelId: opts?.currentChannelId, - messageId: opts?.currentMessageId, - }; - const shouldExpectCompletionMessage = result.inlineDelivery - ? false - : expectsCompletionMessage; - try { - registerSubagentRun({ - runId: childRunId, - requesterTurnRunId: opts?.requesterTurnRunId, - childSessionKey, - controllerSessionKey: ownership.controllerSessionKey, - requesterSessionKey: ownership.completionRequesterSessionKey, - requesterOrigin, - progressOrigin, - requesterDisplayKey: ownership.completionRequesterDisplayKey, - task, - taskName, - requesterAgentId: opts?.requesterAgentIdOverride, - cleanup: trackedCleanup, - label: label || undefined, - runTimeoutSeconds: result.runTimeoutSeconds, - expectsCompletionMessage: shouldExpectCompletionMessage, - spawnMode: trackedSpawnMode, - }); - try { - const hookRunner = getGlobalHookRunner(); - if (hookRunner?.hasHooks("subagent_progress")) { - await hookRunner.runSubagentProgress( - { - phase: "started", - runId: childRunId, - childSessionKey, - requester: progressOrigin, - }, - { - runId: childRunId, - childSessionKey, - requesterSessionKey: ownership.completionRequesterSessionKey, - }, - ); - } - } catch { - // ACP already started; presentation hooks are best-effort only. - } - } catch (err) { - // Best-effort only: the ACP turn was already started above, so deleting the - // child session record here does not guarantee the in-flight run was aborted. - await cleanupUntrackedAcpSession(childSessionKey); - return jsonResult({ - status: "error", - error: `Failed to register ACP run: ${summarizeSessionsSpawnError(err)}. Cleanup was attempted, but the already-started ACP run may still finish in the background.`, - childSessionKey, - runId: childRunId, - ...roleContext, - }); - } - } return jsonResult(addRoleToFailureResult(result, requestedAgentId)); } diff --git a/src/agents/tools/sessions-spawn-visible.ts b/src/agents/tools/sessions-spawn-visible.ts index 05a97c2c57d9..67fc3ba43bf1 100644 --- a/src/agents/tools/sessions-spawn-visible.ts +++ b/src/agents/tools/sessions-spawn-visible.ts @@ -5,7 +5,6 @@ import { } from "../../config/agent-limits.js"; import { getRuntimeConfig } from "../../config/config.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; -import { callGateway } from "../../gateway/call.js"; import { isPathInside } from "../../infra/path-guards.js"; import { isValidAgentId, @@ -62,33 +61,10 @@ type VisibleSessionsSpawnOptions = VisibleSessionsSpawnDeps & { inheritedToolDenylist?: string[]; }; -export function summarizeSessionsSpawnError(error: unknown): string { +function summarizeSessionsSpawnError(error: unknown): string { return error instanceof Error ? error.message : typeof error === "string" ? error : "error"; } -export function resolveTrackedSpawnMode(params: { - requestedMode?: "run" | "session"; - threadRequested: boolean; -}): "run" | "session" { - return params.requestedMode ?? (params.threadRequested ? "session" : "run"); -} - -export async function cleanupUntrackedAcpSession(sessionKey: string): Promise { - const key = sessionKey.trim(); - if (!key) { - return; - } - try { - await callGateway({ - method: "sessions.delete", - params: { key, deleteTranscript: true, emitLifecycleHooks: false }, - timeoutMs: 10_000, - }); - } catch { - // Best-effort cleanup only. - } -} - async function deleteVisibleSession( gatewayCall: InProcessGatewayCaller, childSessionKey: string,