ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-28 05:16:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor(agents): shared spawn orchestration pipeline behind backend adapters (#111007)

Browse Source 
* 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
This commit is contained in:
Peter Steinberger
2026-07-19 00:40:22 +01:00
committed by GitHub
parent ec998a0f3f
commit c1e714e3df
9 changed files with 975 additions and 1189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/acp-spawn.test.ts
+38 -9
View File
@@ -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(
{
src/agents/acp-spawn.ts
+244 -497
View File
@@ -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<ReturnType<typeof getAcpSessionManager>["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<string, { defaultAccount?: unknown } | undefined>;
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<AcpBackendState> = {
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,
};
}
src/agents/spawn-pipeline.ts
+96
View File
@@ -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<TState> = {
initialize(): Promise<TState>;
dispatchTurn(state: TState): Promise<{ runId: string }>;
cleanupOnFailure(params: {
phase: SpawnPipelinePhase;
state?: TState;
error: unknown;
}): Promise<void>;
};
type RegisterSubagentRunInput = Parameters<typeof registerSubagentRun>[0];
type SpawnProgressOrigin = {
channel?: string;
accountId?: string;
to?: string;
threadId?: string | number;
channelId?: string;
messageId?: string | number;
};
type SpawnPipelineResult<TState> =
| { 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<TState>(params: {
adapter: SpawnBackendAdapter<TState>;
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<SpawnPipelineResult<TState>> {
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 };
}
src/agents/spawn-plan.ts
+396
View File
@@ -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<typeof getSessionBindingService>;
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<string, { defaultAccount?: unknown } | undefined>;
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".';
}
src/agents/subagent-spawn.runtime.ts
-4
View File
@@ -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";
src/agents/subagent-spawn.ts
+170 -510
View File
@@ -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<string, { defaultAccount?: unknown } | undefined>;
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<string, unknown> = {
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<SubagentBackendState> = {
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 {
src/agents/tools/sessions-spawn-tool.test.ts
+21 -51
View File
@@ -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. */
src/agents/tools/sessions-spawn-tool.ts
+9 -93
View File
@@ -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));
}
src/agents/tools/sessions-spawn-visible.ts
+1 -25
View File
@@ -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<void> {
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,