feat: add sessions_spawn sub-agent tool
This commit is contained in:
204
src/agents/clawdbot-tools.subagents.test.ts
Normal file
204
src/agents/clawdbot-tools.subagents.test.ts
Normal file
@@ -0,0 +1,204 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
const callGatewayMock = vi.fn();
|
||||
vi.mock("../gateway/call.js", () => ({
|
||||
callGateway: (opts: unknown) => callGatewayMock(opts),
|
||||
}));
|
||||
|
||||
vi.mock("../config/config.js", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("../config/config.js")>();
|
||||
return {
|
||||
...actual,
|
||||
loadConfig: () => ({
|
||||
session: {
|
||||
mainKey: "main",
|
||||
scope: "per-sender",
|
||||
},
|
||||
}),
|
||||
resolveGatewayPort: () => 18789,
|
||||
};
|
||||
});
|
||||
|
||||
import { createClawdbotTools } from "./clawdbot-tools.js";
|
||||
|
||||
describe("subagents", () => {
|
||||
it("sessions_spawn announces back to the requester group surface", async () => {
|
||||
callGatewayMock.mockReset();
|
||||
const calls: Array<{ method?: string; params?: unknown }> = [];
|
||||
let agentCallCount = 0;
|
||||
let lastWaitedRunId: string | undefined;
|
||||
const replyByRunId = new Map<string, string>();
|
||||
let sendParams: { to?: string; provider?: string; message?: string } = {};
|
||||
let deletedKey: string | undefined;
|
||||
|
||||
callGatewayMock.mockImplementation(async (opts: unknown) => {
|
||||
const request = opts as { method?: string; params?: unknown };
|
||||
calls.push(request);
|
||||
if (request.method === "agent") {
|
||||
agentCallCount += 1;
|
||||
const runId = `run-${agentCallCount}`;
|
||||
const params = request.params as
|
||||
| { message?: string; sessionKey?: string }
|
||||
| undefined;
|
||||
const message = params?.message ?? "";
|
||||
const reply =
|
||||
message === "Sub-agent announce step." ? "announce now" : "result";
|
||||
replyByRunId.set(runId, reply);
|
||||
return {
|
||||
runId,
|
||||
status: "accepted",
|
||||
acceptedAt: 1000 + agentCallCount,
|
||||
};
|
||||
}
|
||||
if (request.method === "agent.wait") {
|
||||
const params = request.params as { runId?: string } | undefined;
|
||||
lastWaitedRunId = params?.runId;
|
||||
return { runId: params?.runId ?? "run-1", status: "ok" };
|
||||
}
|
||||
if (request.method === "chat.history") {
|
||||
const text =
|
||||
(lastWaitedRunId && replyByRunId.get(lastWaitedRunId)) ?? "";
|
||||
return {
|
||||
messages: [{ role: "assistant", content: [{ type: "text", text }] }],
|
||||
};
|
||||
}
|
||||
if (request.method === "send") {
|
||||
const params = request.params as
|
||||
| { to?: string; provider?: string; message?: string }
|
||||
| undefined;
|
||||
sendParams = {
|
||||
to: params?.to,
|
||||
provider: params?.provider,
|
||||
message: params?.message,
|
||||
};
|
||||
return { messageId: "m-announce" };
|
||||
}
|
||||
if (request.method === "sessions.delete") {
|
||||
const params = request.params as { key?: string } | undefined;
|
||||
deletedKey = params?.key;
|
||||
return { ok: true };
|
||||
}
|
||||
return {};
|
||||
});
|
||||
|
||||
const tool = createClawdbotTools({
|
||||
agentSessionKey: "discord:group:req",
|
||||
agentSurface: "discord",
|
||||
}).find((candidate) => candidate.name === "sessions_spawn");
|
||||
if (!tool) throw new Error("missing sessions_spawn tool");
|
||||
|
||||
const result = await tool.execute("call1", {
|
||||
task: "do thing",
|
||||
timeoutSeconds: 1,
|
||||
});
|
||||
expect(result.details).toMatchObject({ status: "ok", reply: "result" });
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
const agentCalls = calls.filter((call) => call.method === "agent");
|
||||
expect(agentCalls).toHaveLength(2);
|
||||
const first = agentCalls[0]?.params as
|
||||
| { lane?: string; deliver?: boolean; sessionKey?: string }
|
||||
| undefined;
|
||||
expect(first?.lane).toBe("subagent");
|
||||
expect(first?.deliver).toBe(false);
|
||||
expect(first?.sessionKey?.startsWith("subagent:")).toBe(true);
|
||||
|
||||
expect(sendParams).toMatchObject({
|
||||
provider: "discord",
|
||||
to: "channel:req",
|
||||
message: "announce now",
|
||||
});
|
||||
expect(deletedKey?.startsWith("subagent:")).toBe(true);
|
||||
});
|
||||
|
||||
it("sessions_spawn resolves main announce target from sessions.list", async () => {
|
||||
callGatewayMock.mockReset();
|
||||
const calls: Array<{ method?: string; params?: unknown }> = [];
|
||||
let agentCallCount = 0;
|
||||
let lastWaitedRunId: string | undefined;
|
||||
const replyByRunId = new Map<string, string>();
|
||||
let sendParams: { to?: string; provider?: string; message?: string } = {};
|
||||
|
||||
callGatewayMock.mockImplementation(async (opts: unknown) => {
|
||||
const request = opts as { method?: string; params?: unknown };
|
||||
calls.push(request);
|
||||
if (request.method === "sessions.list") {
|
||||
return {
|
||||
sessions: [
|
||||
{
|
||||
key: "main",
|
||||
lastChannel: "whatsapp",
|
||||
lastTo: "+123",
|
||||
},
|
||||
],
|
||||
};
|
||||
}
|
||||
if (request.method === "agent") {
|
||||
agentCallCount += 1;
|
||||
const runId = `run-${agentCallCount}`;
|
||||
const params = request.params as
|
||||
| { message?: string; sessionKey?: string }
|
||||
| undefined;
|
||||
const message = params?.message ?? "";
|
||||
const reply =
|
||||
message === "Sub-agent announce step." ? "hello from sub" : "done";
|
||||
replyByRunId.set(runId, reply);
|
||||
return {
|
||||
runId,
|
||||
status: "accepted",
|
||||
acceptedAt: 2000 + agentCallCount,
|
||||
};
|
||||
}
|
||||
if (request.method === "agent.wait") {
|
||||
const params = request.params as { runId?: string } | undefined;
|
||||
lastWaitedRunId = params?.runId;
|
||||
return { runId: params?.runId ?? "run-1", status: "ok" };
|
||||
}
|
||||
if (request.method === "chat.history") {
|
||||
const text =
|
||||
(lastWaitedRunId && replyByRunId.get(lastWaitedRunId)) ?? "";
|
||||
return {
|
||||
messages: [{ role: "assistant", content: [{ type: "text", text }] }],
|
||||
};
|
||||
}
|
||||
if (request.method === "send") {
|
||||
const params = request.params as
|
||||
| { to?: string; provider?: string; message?: string }
|
||||
| undefined;
|
||||
sendParams = {
|
||||
to: params?.to,
|
||||
provider: params?.provider,
|
||||
message: params?.message,
|
||||
};
|
||||
return { messageId: "m1" };
|
||||
}
|
||||
if (request.method === "sessions.delete") {
|
||||
return { ok: true };
|
||||
}
|
||||
return {};
|
||||
});
|
||||
|
||||
const tool = createClawdbotTools({
|
||||
agentSessionKey: "main",
|
||||
agentSurface: "whatsapp",
|
||||
}).find((candidate) => candidate.name === "sessions_spawn");
|
||||
if (!tool) throw new Error("missing sessions_spawn tool");
|
||||
|
||||
const result = await tool.execute("call2", {
|
||||
task: "do thing",
|
||||
timeoutSeconds: 1,
|
||||
});
|
||||
expect(result.details).toMatchObject({ status: "ok", reply: "done" });
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
expect(sendParams).toMatchObject({
|
||||
provider: "whatsapp",
|
||||
to: "+123",
|
||||
message: "hello from sub",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -10,6 +10,7 @@ import { createNodesTool } from "./tools/nodes-tool.js";
|
||||
import { createSessionsHistoryTool } from "./tools/sessions-history-tool.js";
|
||||
import { createSessionsListTool } from "./tools/sessions-list-tool.js";
|
||||
import { createSessionsSendTool } from "./tools/sessions-send-tool.js";
|
||||
import { createSessionsSpawnTool } from "./tools/sessions-spawn-tool.js";
|
||||
import { createSlackTool } from "./tools/slack-tool.js";
|
||||
|
||||
export function createClawdbotTools(options?: {
|
||||
@@ -33,6 +34,10 @@ export function createClawdbotTools(options?: {
|
||||
agentSessionKey: options?.agentSessionKey,
|
||||
agentSurface: options?.agentSurface,
|
||||
}),
|
||||
createSessionsSpawnTool({
|
||||
agentSessionKey: options?.agentSessionKey,
|
||||
agentSurface: options?.agentSurface,
|
||||
}),
|
||||
...(imageTool ? [imageTool] : []),
|
||||
];
|
||||
}
|
||||
|
||||
@@ -116,6 +116,36 @@ describe("createClawdbotCodingTools", () => {
|
||||
expect(slack.some((tool) => tool.name === "slack")).toBe(true);
|
||||
});
|
||||
|
||||
it("filters session tools for sub-agent sessions by default", () => {
|
||||
const tools = createClawdbotCodingTools({ sessionKey: "subagent:test" });
|
||||
const names = new Set(tools.map((tool) => tool.name));
|
||||
expect(names.has("sessions_list")).toBe(false);
|
||||
expect(names.has("sessions_history")).toBe(false);
|
||||
expect(names.has("sessions_send")).toBe(false);
|
||||
expect(names.has("sessions_spawn")).toBe(false);
|
||||
|
||||
expect(names.has("read")).toBe(true);
|
||||
expect(names.has("bash")).toBe(true);
|
||||
expect(names.has("process")).toBe(true);
|
||||
});
|
||||
|
||||
it("supports allow-only sub-agent tool policy", () => {
|
||||
const tools = createClawdbotCodingTools({
|
||||
sessionKey: "subagent:test",
|
||||
// Intentionally partial config; only fields used by pi-tools are provided.
|
||||
config: {
|
||||
agent: {
|
||||
subagents: {
|
||||
tools: {
|
||||
allow: ["read"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(tools.map((tool) => tool.name)).toEqual(["read"]);
|
||||
});
|
||||
|
||||
it("keeps read tool image metadata intact", async () => {
|
||||
const tools = createClawdbotCodingTools();
|
||||
const readTool = tools.find((tool) => tool.name === "read");
|
||||
|
||||
@@ -333,6 +333,28 @@ function normalizeToolNames(list?: string[]) {
|
||||
return list.map((entry) => entry.trim().toLowerCase()).filter(Boolean);
|
||||
}
|
||||
|
||||
const DEFAULT_SUBAGENT_TOOL_DENY = [
|
||||
"sessions_list",
|
||||
"sessions_history",
|
||||
"sessions_send",
|
||||
"sessions_spawn",
|
||||
];
|
||||
|
||||
function isSubagentSessionKey(sessionKey?: string): boolean {
|
||||
const key = sessionKey?.trim().toLowerCase() ?? "";
|
||||
return key.startsWith("subagent:");
|
||||
}
|
||||
|
||||
function resolveSubagentToolPolicy(cfg?: ClawdbotConfig): SandboxToolPolicy {
|
||||
const configured = cfg?.agent?.subagents?.tools;
|
||||
const deny = [
|
||||
...DEFAULT_SUBAGENT_TOOL_DENY,
|
||||
...(Array.isArray(configured?.deny) ? configured.deny : []),
|
||||
];
|
||||
const allow = Array.isArray(configured?.allow) ? configured.allow : undefined;
|
||||
return { allow, deny };
|
||||
}
|
||||
|
||||
function filterToolsByPolicy(
|
||||
tools: AnyAgentTool[],
|
||||
policy?: SandboxToolPolicy,
|
||||
@@ -553,7 +575,14 @@ export function createClawdbotCodingTools(options?: {
|
||||
const sandboxed = sandbox
|
||||
? filterToolsByPolicy(globallyFiltered, sandbox.tools)
|
||||
: globallyFiltered;
|
||||
const subagentFiltered =
|
||||
isSubagentSessionKey(options?.sessionKey) && options?.sessionKey
|
||||
? filterToolsByPolicy(
|
||||
sandboxed,
|
||||
resolveSubagentToolPolicy(options.config),
|
||||
)
|
||||
: sandboxed;
|
||||
// Always normalize tool JSON Schemas before handing them to pi-agent/pi-ai.
|
||||
// Without this, some providers (notably OpenAI) will reject root-level union schemas.
|
||||
return sandboxed.map(normalizeToolParameters);
|
||||
return subagentFiltered.map(normalizeToolParameters);
|
||||
}
|
||||
|
||||
@@ -165,6 +165,11 @@
|
||||
"title": "Session Send",
|
||||
"detailKeys": ["sessionKey", "timeoutSeconds"]
|
||||
},
|
||||
"sessions_spawn": {
|
||||
"emoji": "🧑🔧",
|
||||
"title": "Sub-agent",
|
||||
"detailKeys": ["label", "timeoutSeconds", "cleanup"]
|
||||
},
|
||||
"whatsapp_login": {
|
||||
"emoji": "🟢",
|
||||
"title": "WhatsApp Login",
|
||||
|
||||
56
src/agents/tools/agent-step.ts
Normal file
56
src/agents/tools/agent-step.ts
Normal file
@@ -0,0 +1,56 @@
|
||||
import crypto from "node:crypto";
|
||||
|
||||
import { callGateway } from "../../gateway/call.js";
|
||||
import { extractAssistantText, stripToolMessages } from "./sessions-helpers.js";
|
||||
|
||||
export async function readLatestAssistantReply(params: {
|
||||
sessionKey: string;
|
||||
limit?: number;
|
||||
}): Promise<string | undefined> {
|
||||
const history = (await callGateway({
|
||||
method: "chat.history",
|
||||
params: { sessionKey: params.sessionKey, limit: params.limit ?? 50 },
|
||||
})) as { messages?: unknown[] };
|
||||
const filtered = stripToolMessages(
|
||||
Array.isArray(history?.messages) ? history.messages : [],
|
||||
);
|
||||
const last = filtered.length > 0 ? filtered[filtered.length - 1] : undefined;
|
||||
return last ? extractAssistantText(last) : undefined;
|
||||
}
|
||||
|
||||
export async function runAgentStep(params: {
|
||||
sessionKey: string;
|
||||
message: string;
|
||||
extraSystemPrompt: string;
|
||||
timeoutMs: number;
|
||||
lane?: string;
|
||||
}): Promise<string | undefined> {
|
||||
const stepIdem = crypto.randomUUID();
|
||||
const response = (await callGateway({
|
||||
method: "agent",
|
||||
params: {
|
||||
message: params.message,
|
||||
sessionKey: params.sessionKey,
|
||||
idempotencyKey: stepIdem,
|
||||
deliver: false,
|
||||
lane: params.lane ?? "nested",
|
||||
extraSystemPrompt: params.extraSystemPrompt,
|
||||
},
|
||||
timeoutMs: 10_000,
|
||||
})) as { runId?: string; acceptedAt?: number };
|
||||
|
||||
const stepRunId =
|
||||
typeof response?.runId === "string" && response.runId ? response.runId : "";
|
||||
const resolvedRunId = stepRunId || stepIdem;
|
||||
const stepWaitMs = Math.min(params.timeoutMs, 60_000);
|
||||
const wait = (await callGateway({
|
||||
method: "agent.wait",
|
||||
params: {
|
||||
runId: resolvedRunId,
|
||||
timeoutMs: stepWaitMs,
|
||||
},
|
||||
timeoutMs: stepWaitMs + 2000,
|
||||
})) as { status?: string };
|
||||
if (wait?.status !== "ok") return undefined;
|
||||
return await readLatestAssistantReply({ sessionKey: params.sessionKey });
|
||||
}
|
||||
36
src/agents/tools/sessions-announce-target.ts
Normal file
36
src/agents/tools/sessions-announce-target.ts
Normal file
@@ -0,0 +1,36 @@
|
||||
import { callGateway } from "../../gateway/call.js";
|
||||
import type { AnnounceTarget } from "./sessions-send-helpers.js";
|
||||
import { resolveAnnounceTargetFromKey } from "./sessions-send-helpers.js";
|
||||
|
||||
export async function resolveAnnounceTarget(params: {
|
||||
sessionKey: string;
|
||||
displayKey: string;
|
||||
}): Promise<AnnounceTarget | null> {
|
||||
const parsed = resolveAnnounceTargetFromKey(params.sessionKey);
|
||||
if (parsed) return parsed;
|
||||
const parsedDisplay = resolveAnnounceTargetFromKey(params.displayKey);
|
||||
if (parsedDisplay) return parsedDisplay;
|
||||
|
||||
try {
|
||||
const list = (await callGateway({
|
||||
method: "sessions.list",
|
||||
params: {
|
||||
includeGlobal: true,
|
||||
includeUnknown: true,
|
||||
limit: 200,
|
||||
},
|
||||
})) as { sessions?: Array<Record<string, unknown>> };
|
||||
const sessions = Array.isArray(list?.sessions) ? list.sessions : [];
|
||||
const match =
|
||||
sessions.find((entry) => entry?.key === params.sessionKey) ??
|
||||
sessions.find((entry) => entry?.key === params.displayKey);
|
||||
const channel =
|
||||
typeof match?.lastChannel === "string" ? match.lastChannel : undefined;
|
||||
const to = typeof match?.lastTo === "string" ? match.lastTo : undefined;
|
||||
if (channel && to) return { channel, to };
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
@@ -4,8 +4,10 @@ import { Type } from "@sinclair/typebox";
|
||||
|
||||
import { loadConfig } from "../../config/config.js";
|
||||
import { callGateway } from "../../gateway/call.js";
|
||||
import { readLatestAssistantReply, runAgentStep } from "./agent-step.js";
|
||||
import type { AnyAgentTool } from "./common.js";
|
||||
import { jsonResult, readStringParam } from "./common.js";
|
||||
import { resolveAnnounceTarget } from "./sessions-announce-target.js";
|
||||
import {
|
||||
extractAssistantText,
|
||||
resolveDisplaySessionKey,
|
||||
@@ -14,13 +16,11 @@ import {
|
||||
stripToolMessages,
|
||||
} from "./sessions-helpers.js";
|
||||
import {
|
||||
type AnnounceTarget,
|
||||
buildAgentToAgentAnnounceContext,
|
||||
buildAgentToAgentMessageContext,
|
||||
buildAgentToAgentReplyContext,
|
||||
isAnnounceSkip,
|
||||
isReplySkip,
|
||||
resolveAnnounceTargetFromKey,
|
||||
resolvePingPongTurns,
|
||||
} from "./sessions-send-helpers.js";
|
||||
|
||||
@@ -83,87 +83,6 @@ export function createSessionsSendTool(opts?: {
|
||||
const requesterSurface = opts?.agentSurface;
|
||||
const maxPingPongTurns = resolvePingPongTurns(cfg);
|
||||
|
||||
const resolveAnnounceTarget =
|
||||
async (): Promise<AnnounceTarget | null> => {
|
||||
const parsed = resolveAnnounceTargetFromKey(resolvedKey);
|
||||
if (parsed) return parsed;
|
||||
try {
|
||||
const list = (await callGateway({
|
||||
method: "sessions.list",
|
||||
params: {
|
||||
includeGlobal: true,
|
||||
includeUnknown: true,
|
||||
limit: 200,
|
||||
},
|
||||
})) as { sessions?: Array<Record<string, unknown>> };
|
||||
const sessions = Array.isArray(list?.sessions) ? list.sessions : [];
|
||||
const match =
|
||||
sessions.find((entry) => entry?.key === resolvedKey) ??
|
||||
sessions.find((entry) => entry?.key === displayKey);
|
||||
const channel =
|
||||
typeof match?.lastChannel === "string"
|
||||
? match.lastChannel
|
||||
: undefined;
|
||||
const to =
|
||||
typeof match?.lastTo === "string" ? match.lastTo : undefined;
|
||||
if (channel && to) return { channel, to };
|
||||
} catch {
|
||||
// ignore; fall through to null
|
||||
}
|
||||
return null;
|
||||
};
|
||||
|
||||
const readLatestAssistantReply = async (
|
||||
sessionKeyToRead: string,
|
||||
): Promise<string | undefined> => {
|
||||
const history = (await callGateway({
|
||||
method: "chat.history",
|
||||
params: { sessionKey: sessionKeyToRead, limit: 50 },
|
||||
})) as { messages?: unknown[] };
|
||||
const filtered = stripToolMessages(
|
||||
Array.isArray(history?.messages) ? history.messages : [],
|
||||
);
|
||||
const last =
|
||||
filtered.length > 0 ? filtered[filtered.length - 1] : undefined;
|
||||
return last ? extractAssistantText(last) : undefined;
|
||||
};
|
||||
|
||||
const runAgentStep = async (step: {
|
||||
sessionKey: string;
|
||||
message: string;
|
||||
extraSystemPrompt: string;
|
||||
timeoutMs: number;
|
||||
}): Promise<string | undefined> => {
|
||||
const stepIdem = crypto.randomUUID();
|
||||
const response = (await callGateway({
|
||||
method: "agent",
|
||||
params: {
|
||||
message: step.message,
|
||||
sessionKey: step.sessionKey,
|
||||
idempotencyKey: stepIdem,
|
||||
deliver: false,
|
||||
lane: "nested",
|
||||
extraSystemPrompt: step.extraSystemPrompt,
|
||||
},
|
||||
timeoutMs: 10_000,
|
||||
})) as { runId?: string; acceptedAt?: number };
|
||||
const stepRunId =
|
||||
typeof response?.runId === "string" && response.runId
|
||||
? response.runId
|
||||
: stepIdem;
|
||||
const stepWaitMs = Math.min(step.timeoutMs, 60_000);
|
||||
const wait = (await callGateway({
|
||||
method: "agent.wait",
|
||||
params: {
|
||||
runId: stepRunId,
|
||||
timeoutMs: stepWaitMs,
|
||||
},
|
||||
timeoutMs: stepWaitMs + 2000,
|
||||
})) as { status?: string };
|
||||
if (wait?.status !== "ok") return undefined;
|
||||
return readLatestAssistantReply(step.sessionKey);
|
||||
};
|
||||
|
||||
const runAgentToAgentFlow = async (
|
||||
roundOneReply?: string,
|
||||
runInfo?: { runId: string },
|
||||
@@ -182,12 +101,17 @@ export function createSessionsSendTool(opts?: {
|
||||
timeoutMs: waitMs + 2000,
|
||||
})) as { status?: string };
|
||||
if (wait?.status === "ok") {
|
||||
primaryReply = await readLatestAssistantReply(resolvedKey);
|
||||
primaryReply = await readLatestAssistantReply({
|
||||
sessionKey: resolvedKey,
|
||||
});
|
||||
latestReply = primaryReply;
|
||||
}
|
||||
}
|
||||
if (!latestReply) return;
|
||||
const announceTarget = await resolveAnnounceTarget();
|
||||
const announceTarget = await resolveAnnounceTarget({
|
||||
sessionKey: resolvedKey,
|
||||
displayKey,
|
||||
});
|
||||
const targetChannel = announceTarget?.channel ?? "unknown";
|
||||
if (
|
||||
maxPingPongTurns > 0 &&
|
||||
@@ -216,6 +140,7 @@ export function createSessionsSendTool(opts?: {
|
||||
message: incomingMessage,
|
||||
extraSystemPrompt: replyPrompt,
|
||||
timeoutMs: announceTimeoutMs,
|
||||
lane: "nested",
|
||||
});
|
||||
if (!replyText || isReplySkip(replyText)) {
|
||||
break;
|
||||
@@ -241,6 +166,7 @@ export function createSessionsSendTool(opts?: {
|
||||
message: "Agent-to-agent announce step.",
|
||||
extraSystemPrompt: announcePrompt,
|
||||
timeoutMs: announceTimeoutMs,
|
||||
lane: "nested",
|
||||
});
|
||||
if (
|
||||
announceTarget &&
|
||||
|
||||
348
src/agents/tools/sessions-spawn-tool.ts
Normal file
348
src/agents/tools/sessions-spawn-tool.ts
Normal file
@@ -0,0 +1,348 @@
|
||||
import crypto from "node:crypto";
|
||||
|
||||
import { Type } from "@sinclair/typebox";
|
||||
|
||||
import { loadConfig } from "../../config/config.js";
|
||||
import { callGateway } from "../../gateway/call.js";
|
||||
import { readLatestAssistantReply, runAgentStep } from "./agent-step.js";
|
||||
import type { AnyAgentTool } from "./common.js";
|
||||
import { jsonResult, readStringParam } from "./common.js";
|
||||
import { resolveAnnounceTarget } from "./sessions-announce-target.js";
|
||||
import {
|
||||
resolveDisplaySessionKey,
|
||||
resolveInternalSessionKey,
|
||||
resolveMainSessionAlias,
|
||||
} from "./sessions-helpers.js";
|
||||
import { isAnnounceSkip } from "./sessions-send-helpers.js";
|
||||
|
||||
const SessionsSpawnToolSchema = Type.Object({
|
||||
task: Type.String(),
|
||||
label: Type.Optional(Type.String()),
|
||||
timeoutSeconds: Type.Optional(Type.Integer({ minimum: 0 })),
|
||||
cleanup: Type.Optional(
|
||||
Type.Union([Type.Literal("delete"), Type.Literal("keep")]),
|
||||
),
|
||||
});
|
||||
|
||||
function buildSubagentSystemPrompt(params: {
|
||||
requesterSessionKey?: string;
|
||||
requesterSurface?: string;
|
||||
childSessionKey: string;
|
||||
label?: string;
|
||||
}) {
|
||||
const lines = [
|
||||
"Sub-agent context:",
|
||||
params.label ? `Label: ${params.label}` : undefined,
|
||||
params.requesterSessionKey
|
||||
? `Requester session: ${params.requesterSessionKey}.`
|
||||
: undefined,
|
||||
params.requesterSurface
|
||||
? `Requester surface: ${params.requesterSurface}.`
|
||||
: undefined,
|
||||
`Your session: ${params.childSessionKey}.`,
|
||||
"Run the task. Provide a clear final answer (plain text).",
|
||||
'After you finish, you may be asked to produce an "announce" message to post back to the requester chat.',
|
||||
].filter(Boolean);
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
function buildSubagentAnnouncePrompt(params: {
|
||||
requesterSessionKey?: string;
|
||||
requesterSurface?: string;
|
||||
announceChannel: string;
|
||||
task: string;
|
||||
subagentReply?: string;
|
||||
}) {
|
||||
const lines = [
|
||||
"Sub-agent announce step:",
|
||||
params.requesterSessionKey
|
||||
? `Requester session: ${params.requesterSessionKey}.`
|
||||
: undefined,
|
||||
params.requesterSurface
|
||||
? `Requester surface: ${params.requesterSurface}.`
|
||||
: undefined,
|
||||
`Post target surface: ${params.announceChannel}.`,
|
||||
`Original task: ${params.task}`,
|
||||
params.subagentReply
|
||||
? `Sub-agent result: ${params.subagentReply}`
|
||||
: "Sub-agent result: (not available).",
|
||||
'Reply exactly "ANNOUNCE_SKIP" to stay silent.',
|
||||
"Any other reply will be posted to the requester chat surface.",
|
||||
].filter(Boolean);
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
async function runSubagentAnnounceFlow(params: {
|
||||
childSessionKey: string;
|
||||
childRunId: string;
|
||||
requesterSessionKey: string;
|
||||
requesterSurface?: string;
|
||||
requesterDisplayKey: string;
|
||||
task: string;
|
||||
timeoutMs: number;
|
||||
cleanup: "delete" | "keep";
|
||||
roundOneReply?: string;
|
||||
}) {
|
||||
try {
|
||||
let reply = params.roundOneReply;
|
||||
if (!reply) {
|
||||
const waitMs = Math.min(params.timeoutMs, 60_000);
|
||||
const wait = (await callGateway({
|
||||
method: "agent.wait",
|
||||
params: {
|
||||
runId: params.childRunId,
|
||||
timeoutMs: waitMs,
|
||||
},
|
||||
timeoutMs: waitMs + 2000,
|
||||
})) as { status?: string };
|
||||
if (wait?.status !== "ok") return;
|
||||
reply = await readLatestAssistantReply({
|
||||
sessionKey: params.childSessionKey,
|
||||
});
|
||||
}
|
||||
|
||||
const announceTarget = await resolveAnnounceTarget({
|
||||
sessionKey: params.requesterSessionKey,
|
||||
displayKey: params.requesterDisplayKey,
|
||||
});
|
||||
if (!announceTarget) return;
|
||||
|
||||
const announcePrompt = buildSubagentAnnouncePrompt({
|
||||
requesterSessionKey: params.requesterSessionKey,
|
||||
requesterSurface: params.requesterSurface,
|
||||
announceChannel: announceTarget.channel,
|
||||
task: params.task,
|
||||
subagentReply: reply,
|
||||
});
|
||||
|
||||
const announceReply = await runAgentStep({
|
||||
sessionKey: params.childSessionKey,
|
||||
message: "Sub-agent announce step.",
|
||||
extraSystemPrompt: announcePrompt,
|
||||
timeoutMs: params.timeoutMs,
|
||||
lane: "nested",
|
||||
});
|
||||
|
||||
if (
|
||||
!announceReply ||
|
||||
!announceReply.trim() ||
|
||||
isAnnounceSkip(announceReply)
|
||||
)
|
||||
return;
|
||||
|
||||
await callGateway({
|
||||
method: "send",
|
||||
params: {
|
||||
to: announceTarget.to,
|
||||
message: announceReply.trim(),
|
||||
provider: announceTarget.channel,
|
||||
idempotencyKey: crypto.randomUUID(),
|
||||
},
|
||||
timeoutMs: 10_000,
|
||||
});
|
||||
} catch {
|
||||
// Best-effort follow-ups; ignore failures to avoid breaking the caller response.
|
||||
} finally {
|
||||
if (params.cleanup === "delete") {
|
||||
try {
|
||||
await callGateway({
|
||||
method: "sessions.delete",
|
||||
params: { key: params.childSessionKey, deleteTranscript: true },
|
||||
timeoutMs: 10_000,
|
||||
});
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function createSessionsSpawnTool(opts?: {
|
||||
agentSessionKey?: string;
|
||||
agentSurface?: string;
|
||||
}): AnyAgentTool {
|
||||
return {
|
||||
label: "Sessions",
|
||||
name: "sessions_spawn",
|
||||
description:
|
||||
"Spawn a background sub-agent run in an isolated session and announce the result back to the requester chat.",
|
||||
parameters: SessionsSpawnToolSchema,
|
||||
execute: async (_toolCallId, args) => {
|
||||
const params = args as Record<string, unknown>;
|
||||
const task = readStringParam(params, "task", { required: true });
|
||||
const label = typeof params.label === "string" ? params.label.trim() : "";
|
||||
const cleanup =
|
||||
params.cleanup === "keep" || params.cleanup === "delete"
|
||||
? (params.cleanup as "keep" | "delete")
|
||||
: "delete";
|
||||
const timeoutSeconds =
|
||||
typeof params.timeoutSeconds === "number" &&
|
||||
Number.isFinite(params.timeoutSeconds)
|
||||
? Math.max(0, Math.floor(params.timeoutSeconds))
|
||||
: 0;
|
||||
const timeoutMs = timeoutSeconds * 1000;
|
||||
|
||||
const cfg = loadConfig();
|
||||
const { mainKey, alias } = resolveMainSessionAlias(cfg);
|
||||
const requesterSessionKey = opts?.agentSessionKey;
|
||||
const requesterInternalKey = requesterSessionKey
|
||||
? resolveInternalSessionKey({
|
||||
key: requesterSessionKey,
|
||||
alias,
|
||||
mainKey,
|
||||
})
|
||||
: alias;
|
||||
const requesterDisplayKey = resolveDisplaySessionKey({
|
||||
key: requesterInternalKey,
|
||||
alias,
|
||||
mainKey,
|
||||
});
|
||||
|
||||
const childSessionKey = `subagent:${crypto.randomUUID()}`;
|
||||
const childSystemPrompt = buildSubagentSystemPrompt({
|
||||
requesterSessionKey,
|
||||
requesterSurface: opts?.agentSurface,
|
||||
childSessionKey,
|
||||
label: label || undefined,
|
||||
});
|
||||
|
||||
const childIdem = crypto.randomUUID();
|
||||
let childRunId: string = childIdem;
|
||||
try {
|
||||
const response = (await callGateway({
|
||||
method: "agent",
|
||||
params: {
|
||||
message: task,
|
||||
sessionKey: childSessionKey,
|
||||
idempotencyKey: childIdem,
|
||||
deliver: false,
|
||||
lane: "subagent",
|
||||
extraSystemPrompt: childSystemPrompt,
|
||||
},
|
||||
timeoutMs: 10_000,
|
||||
})) as { runId?: string };
|
||||
if (typeof response?.runId === "string" && response.runId) {
|
||||
childRunId = response.runId;
|
||||
}
|
||||
} catch (err) {
|
||||
const messageText =
|
||||
err instanceof Error
|
||||
? err.message
|
||||
: typeof err === "string"
|
||||
? err
|
||||
: "error";
|
||||
return jsonResult({
|
||||
status: "error",
|
||||
error: messageText,
|
||||
childSessionKey,
|
||||
runId: childRunId,
|
||||
});
|
||||
}
|
||||
|
||||
if (timeoutSeconds === 0) {
|
||||
void runSubagentAnnounceFlow({
|
||||
childSessionKey,
|
||||
childRunId,
|
||||
requesterSessionKey: requesterInternalKey,
|
||||
requesterSurface: opts?.agentSurface,
|
||||
requesterDisplayKey,
|
||||
task,
|
||||
timeoutMs: 30_000,
|
||||
cleanup,
|
||||
});
|
||||
return jsonResult({
|
||||
status: "accepted",
|
||||
childSessionKey,
|
||||
runId: childRunId,
|
||||
});
|
||||
}
|
||||
|
||||
let waitStatus: string | undefined;
|
||||
let waitError: string | undefined;
|
||||
try {
|
||||
const wait = (await callGateway({
|
||||
method: "agent.wait",
|
||||
params: {
|
||||
runId: childRunId,
|
||||
timeoutMs,
|
||||
},
|
||||
timeoutMs: timeoutMs + 2000,
|
||||
})) as { status?: string; error?: string };
|
||||
waitStatus = typeof wait?.status === "string" ? wait.status : undefined;
|
||||
waitError = typeof wait?.error === "string" ? wait.error : undefined;
|
||||
} catch (err) {
|
||||
const messageText =
|
||||
err instanceof Error
|
||||
? err.message
|
||||
: typeof err === "string"
|
||||
? err
|
||||
: "error";
|
||||
return jsonResult({
|
||||
status: messageText.includes("gateway timeout") ? "timeout" : "error",
|
||||
error: messageText,
|
||||
childSessionKey,
|
||||
runId: childRunId,
|
||||
});
|
||||
}
|
||||
|
||||
if (waitStatus === "timeout") {
|
||||
void runSubagentAnnounceFlow({
|
||||
childSessionKey,
|
||||
childRunId,
|
||||
requesterSessionKey: requesterInternalKey,
|
||||
requesterSurface: opts?.agentSurface,
|
||||
requesterDisplayKey,
|
||||
task,
|
||||
timeoutMs: 30_000,
|
||||
cleanup,
|
||||
});
|
||||
return jsonResult({
|
||||
status: "timeout",
|
||||
error: waitError,
|
||||
childSessionKey,
|
||||
runId: childRunId,
|
||||
});
|
||||
}
|
||||
if (waitStatus === "error") {
|
||||
void runSubagentAnnounceFlow({
|
||||
childSessionKey,
|
||||
childRunId,
|
||||
requesterSessionKey: requesterInternalKey,
|
||||
requesterSurface: opts?.agentSurface,
|
||||
requesterDisplayKey,
|
||||
task,
|
||||
timeoutMs: 30_000,
|
||||
cleanup,
|
||||
});
|
||||
return jsonResult({
|
||||
status: "error",
|
||||
error: waitError ?? "agent error",
|
||||
childSessionKey,
|
||||
runId: childRunId,
|
||||
});
|
||||
}
|
||||
|
||||
const replyText = await readLatestAssistantReply({
|
||||
sessionKey: childSessionKey,
|
||||
});
|
||||
void runSubagentAnnounceFlow({
|
||||
childSessionKey,
|
||||
childRunId,
|
||||
requesterSessionKey: requesterInternalKey,
|
||||
requesterSurface: opts?.agentSurface,
|
||||
requesterDisplayKey,
|
||||
task,
|
||||
timeoutMs: 30_000,
|
||||
cleanup,
|
||||
roundOneReply: replyText,
|
||||
});
|
||||
|
||||
return jsonResult({
|
||||
status: "ok",
|
||||
childSessionKey,
|
||||
runId: childRunId,
|
||||
reply: replyText,
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -806,6 +806,16 @@ export type ClawdbotConfig = {
|
||||
};
|
||||
/** Max concurrent agent runs across all conversations. Default: 1 (sequential). */
|
||||
maxConcurrent?: number;
|
||||
/** Sub-agent defaults (spawned via sessions_spawn). */
|
||||
subagents?: {
|
||||
/** Max concurrent sub-agent runs (global lane: "subagent"). Default: 1. */
|
||||
maxConcurrent?: number;
|
||||
/** Tool allow/deny policy for sub-agent sessions (deny wins). */
|
||||
tools?: {
|
||||
allow?: string[];
|
||||
deny?: string[];
|
||||
};
|
||||
};
|
||||
/** Bash tool defaults. */
|
||||
bash?: {
|
||||
/** Default time (ms) before a bash command auto-backgrounds. */
|
||||
|
||||
@@ -465,6 +465,17 @@ export const ClawdbotSchema = z.object({
|
||||
typingIntervalSeconds: z.number().int().positive().optional(),
|
||||
heartbeat: HeartbeatSchema,
|
||||
maxConcurrent: z.number().int().positive().optional(),
|
||||
subagents: z
|
||||
.object({
|
||||
maxConcurrent: z.number().int().positive().optional(),
|
||||
tools: z
|
||||
.object({
|
||||
allow: z.array(z.string()).optional(),
|
||||
deny: z.array(z.string()).optional(),
|
||||
})
|
||||
.optional(),
|
||||
})
|
||||
.optional(),
|
||||
bash: z
|
||||
.object({
|
||||
backgroundMs: z.number().int().positive().optional(),
|
||||
|
||||
@@ -671,6 +671,10 @@ export async function startGatewayServer(
|
||||
>();
|
||||
setCommandLaneConcurrency("cron", cfgAtStart.cron?.maxConcurrentRuns ?? 1);
|
||||
setCommandLaneConcurrency("main", cfgAtStart.agent?.maxConcurrent ?? 1);
|
||||
setCommandLaneConcurrency(
|
||||
"subagent",
|
||||
cfgAtStart.agent?.subagents?.maxConcurrent ?? 1,
|
||||
);
|
||||
|
||||
const cronLogger = getChildLogger({
|
||||
module: "cron",
|
||||
@@ -1757,6 +1761,10 @@ export async function startGatewayServer(
|
||||
|
||||
setCommandLaneConcurrency("cron", nextConfig.cron?.maxConcurrentRuns ?? 1);
|
||||
setCommandLaneConcurrency("main", nextConfig.agent?.maxConcurrent ?? 1);
|
||||
setCommandLaneConcurrency(
|
||||
"subagent",
|
||||
nextConfig.agent?.subagents?.maxConcurrent ?? 1,
|
||||
);
|
||||
|
||||
if (plan.hotReasons.length > 0) {
|
||||
logReload.info(
|
||||
|
||||
Reference in New Issue
Block a user