LLM/moltbot
1
0
Fork 0
mirror of https://github.com/moltbot/moltbot.git synced 2026-03-08 06:54:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Add mesh auto-planning with chat command UX and hardened auth/session behavior"

Browse Source 
This reverts commit 16e59b26a6.

# Conflicts:
#	src/auto-reply/reply/commands-mesh.ts
#	src/gateway/server-methods/mesh.ts
#	src/gateway/server-methods/server-methods.test.ts
This commit is contained in:
Peter Steinberger
2026-02-18 02:09:54 +01:00
parent 6dcc052bb4
commit 01672a8f25
14 changed files with 33 additions and 1017 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -267,7 +267,6 @@ ClawHub is a minimal skill registry. With ClawHub enabled, the agent can search
Send these in WhatsApp/Telegram/Slack/Google Chat/Microsoft Teams/WebChat (group commands are owner-only):
- `/status` — compact session status (model + tokens, cost when available)
- `/mesh <goal>` — auto-plan + run a multi-step workflow (`/mesh plan|run|status|retry` available)
- `/new` or `/reset` — reset the session
- `/compact` — compact session context (summary)
- `/think <level>` — off|minimal|low|medium|high|xhigh (GPT-5.2 + Codex models only)

1
docs/tools/slash-commands.md
View File

@@ -73,7 +73,6 @@ Text + native (when enabled):
- `/commands`
- `/skill <name> [input]` (run a skill by name)
- `/status` (show current status; includes provider usage/quota for the current model provider when available)
- `/mesh <goal>` (auto-plan + run a workflow; also `/mesh plan|run|status|retry`, with `/mesh run <mesh-plan-id>` for exact plan replay in the same chat)
- `/allowlist` (list/add/remove allowlist entries)
- `/approve <id> allow-once|allow-always|deny` (resolve exec approval prompts)
- `/context [list|detail|json]` (explain “context”; `detail` shows per-file + per-tool + per-skill + system prompt size)

9
src/auto-reply/commands-registry.data.ts
View File

@@ -172,15 +172,6 @@ function buildChatCommands(): ChatCommandDefinition[] {
textAlias: "/status",
category: "status",
}),
defineChatCommand({
key: "mesh",
nativeName: "mesh",
description: "Plan and run multi-step workflows.",
textAlias: "/mesh",
category: "tools",
argsParsing: "none",
acceptsArgs: true,
}),
defineChatCommand({
key: "allowlist",
description: "List/add/remove allowlist entries.",

2
src/auto-reply/reply/commands-core.ts
View File

@@ -17,7 +17,6 @@ import {
handleStatusCommand,
handleWhoamiCommand,
} from "./commands-info.js";
import { handleMeshCommand } from "./commands-mesh.js";
import { handleModelsCommand } from "./commands-models.js";
import { handlePluginCommand } from "./commands-plugin.js";
import {
@@ -53,7 +52,6 @@ export async function handleCommands(params: HandleCommandsParams): Promise<Comm
handleHelpCommand,
handleCommandsListCommand,
handleStatusCommand,
handleMeshCommand,
handleAllowlistCommand,
handleApproveCommand,
handleContextCommand,

351
src/auto-reply/reply/commands-mesh.ts
View File

@@ -1,351 +0,0 @@
import { callGateway } from "../../gateway/call.js";
import { logVerbose } from "../../globals.js";
import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES } from "../../utils/message-channel.js";
import type { CommandHandler } from "./commands-types.js";
type MeshPlanShape = {
planId: string;
goal: string;
createdAt: number;
steps: Array<{ id: string; name?: string; prompt: string; dependsOn?: string[] }>;
};
type CachedMeshPlan = { plan: MeshPlanShape; createdAt: number };
type ParsedMeshCommand =
| { ok: true; action: "help" }
| { ok: true; action: "run" | "plan"; target: string }
| { ok: true; action: "status"; runId: string }
| { ok: true; action: "retry"; runId: string; stepIds?: string[] }
| { ok: false; message: string }
| null;
const meshPlanCache = new Map<string, CachedMeshPlan>();
const MAX_CACHED_MESH_PLANS = 200;
function trimMeshPlanCache() {
if (meshPlanCache.size <= MAX_CACHED_MESH_PLANS) {
return;
}
const oldest = [...meshPlanCache.entries()]
.toSorted((a, b) => a[1].createdAt - b[1].createdAt)
.slice(0, meshPlanCache.size - MAX_CACHED_MESH_PLANS);
for (const [key] of oldest) {
meshPlanCache.delete(key);
}
}
function parseMeshCommand(commandBody: string): ParsedMeshCommand {
const trimmed = commandBody.trim();
if (!/^\/mesh\b/i.test(trimmed)) {
return null;
}
const rest = trimmed.replace(/^\/mesh\b:?/i, "").trim();
if (!rest || /^help$/i.test(rest)) {
return { ok: true, action: "help" };
}
const tokens = rest.split(/\s+/).filter(Boolean);
if (tokens.length === 0) {
return { ok: true, action: "help" };
}
const actionCandidate = tokens[0]?.toLowerCase() ?? "";
const explicitAction =
actionCandidate === "run" ||
actionCandidate === "plan" ||
actionCandidate === "status" ||
actionCandidate === "retry"
? actionCandidate
: null;
if (!explicitAction) {
// Shorthand: `/mesh <goal>` => auto plan + run
return { ok: true, action: "run", target: rest };
}
const actionArgs = rest.slice(tokens[0]?.length ?? 0).trim();
if (explicitAction === "plan" || explicitAction === "run") {
if (!actionArgs) {
return { ok: false, message: `Usage: /mesh ${explicitAction} <goal>` };
}
return { ok: true, action: explicitAction, target: actionArgs };
}
if (explicitAction === "status") {
if (!actionArgs) {
return { ok: false, message: "Usage: /mesh status <runId>" };
}
return { ok: true, action: "status", runId: actionArgs.split(/\s+/)[0] };
}
// retry
const argsTokens = actionArgs.split(/\s+/).filter(Boolean);
if (argsTokens.length === 0) {
return { ok: false, message: "Usage: /mesh retry <runId> [step1,step2,...]" };
}
const runId = argsTokens[0];
const stepArg = argsTokens.slice(1).join(" ").trim();
const stepIds =
stepArg.length > 0
? stepArg
.split(",")
.map((entry) => entry.trim())
.filter(Boolean)
: undefined;
return { ok: true, action: "retry", runId, stepIds };
}
function cacheKeyForPlan(params: Parameters<CommandHandler>[0], planId: string) {
const sender = params.command.senderId ?? "unknown";
const channel = params.command.channel || "unknown";
return `${channel}:${sender}:${planId}`;
}
function putCachedPlan(params: Parameters<CommandHandler>[0], plan: MeshPlanShape) {
meshPlanCache.set(cacheKeyForPlan(params, plan.planId), { plan, createdAt: Date.now() });
trimMeshPlanCache();
}
function getCachedPlan(
params: Parameters<CommandHandler>[0],
planId: string,
): MeshPlanShape | null {
return meshPlanCache.get(cacheKeyForPlan(params, planId))?.plan ?? null;
}
function looksLikeMeshPlanId(value: string) {
return /^mesh-plan-[a-z0-9-]+$/i.test(value.trim());
}
function resolveMeshCommandBody(params: Parameters<CommandHandler>[0]) {
return (
params.ctx.BodyForCommands ??
params.ctx.CommandBody ??
params.ctx.RawBody ??
params.ctx.Body ??
params.command.commandBodyNormalized
);
}
function formatPlanSummary(plan: {
goal: string;
steps: Array<{ id: string; name?: string; prompt: string; dependsOn?: string[] }>;
}) {
const lines = [`🕸️ Mesh Plan`, `Goal: ${plan.goal}`, "", `Steps (${plan.steps.length}):`];
for (const step of plan.steps) {
const dependsOn = Array.isArray(step.dependsOn) && step.dependsOn.length > 0;
const depLine = dependsOn ? ` (depends on: ${step.dependsOn?.join(", ")})` : "";
lines.push(`- ${step.id}${step.name ? ` — ${step.name}` : ""}${depLine}`);
lines.push(` ${step.prompt}`);
}
return lines.join("\n");
}
function formatRunSummary(payload: {
runId: string;
status: string;
stats?: {
total?: number;
succeeded?: number;
failed?: number;
skipped?: number;
running?: number;
pending?: number;
};
}) {
const stats = payload.stats ?? {};
return [
`🕸️ Mesh Run`,
`Run: ${payload.runId}`,
`Status: ${payload.status}`,
`Steps: total=${stats.total ?? 0}, ok=${stats.succeeded ?? 0}, failed=${stats.failed ?? 0}, skipped=${stats.skipped ?? 0}, running=${stats.running ?? 0}, pending=${stats.pending ?? 0}`,
].join("\n");
}
function meshUsageText() {
return [
"🕸️ Mesh command",
"Usage:",
"- /mesh <goal> (auto plan + run)",
"- /mesh plan <goal>",
"- /mesh run <goal|mesh-plan-id>",
"- /mesh status <runId>",
"- /mesh retry <runId> [step1,step2,...]",
].join("\n");
}
function resolveMeshClientLabel(params: Parameters<CommandHandler>[0]) {
const channel = params.command.channel;
const sender = params.command.senderId ?? "unknown";
return `Chat mesh (${channel}:${sender})`;
}
export const handleMeshCommand: CommandHandler = async (params, allowTextCommands) => {
if (!allowTextCommands) {
return null;
}
const parsed = parseMeshCommand(resolveMeshCommandBody(params));
if (!parsed) {
return null;
}
if (!params.command.isAuthorizedSender) {
logVerbose(
`Ignoring /mesh from unauthorized sender: ${params.command.senderId || "<unknown>"}`,
);
return { shouldContinue: false };
}
if (!parsed.ok) {
return { shouldContinue: false, reply: { text: parsed.message } };
}
if (parsed.action === "help") {
return { shouldContinue: false, reply: { text: meshUsageText() } };
}
const clientDisplayName = resolveMeshClientLabel(params);
const commonGateway = {
clientName: GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT,
clientDisplayName,
mode: GATEWAY_CLIENT_MODES.BACKEND,
} as const;
try {
if (parsed.action === "plan") {
const planResp = await callGateway<{
plan: MeshPlanShape;
order?: string[];
source?: string;
}>({
method: "mesh.plan.auto",
params: {
goal: parsed.target,
agentId: params.agentId ?? "main",
},
...commonGateway,
});
putCachedPlan(params, planResp.plan);
const sourceLine = planResp.source ? `\nPlanner source: ${planResp.source}` : "";
return {
shouldContinue: false,
reply: {
text: `${formatPlanSummary(planResp.plan)}${sourceLine}\n\nRun exact plan: /mesh run ${planResp.plan.planId}`,
},
};
}
if (parsed.action === "run") {
let runPlan: MeshPlanShape;
if (looksLikeMeshPlanId(parsed.target)) {
const cached = getCachedPlan(params, parsed.target.trim());
if (!cached) {
return {
shouldContinue: false,
reply: {
text: `Plan ${parsed.target.trim()} not found in this chat.\nCreate one first: /mesh plan <goal>`,
},
};
}
runPlan = cached;
} else {
const planResp = await callGateway<{
plan: MeshPlanShape;
order?: string[];
source?: string;
}>({
method: "mesh.plan.auto",
params: {
goal: parsed.target,
agentId: params.agentId ?? "main",
},
...commonGateway,
});
putCachedPlan(params, planResp.plan);
runPlan = planResp.plan;
}
const runResp = await callGateway<{
runId: string;
status: string;
stats?: {
total?: number;
succeeded?: number;
failed?: number;
skipped?: number;
running?: number;
pending?: number;
};
}>({
method: "mesh.run",
params: {
plan: runPlan,
},
...commonGateway,
});
return {
shouldContinue: false,
reply: {
text: `${formatPlanSummary(runPlan)}\n\n${formatRunSummary(runResp)}`,
},
};
}
if (parsed.action === "status") {
const statusResp = await callGateway<{
runId: string;
status: string;
stats?: {
total?: number;
succeeded?: number;
failed?: number;
skipped?: number;
running?: number;
pending?: number;
};
}>({
method: "mesh.status",
params: { runId: parsed.runId },
...commonGateway,
});
return {
shouldContinue: false,
reply: { text: formatRunSummary(statusResp) },
};
}
if (parsed.action === "retry") {
const retryResp = await callGateway<{
runId: string;
status: string;
stats?: {
total?: number;
succeeded?: number;
failed?: number;
skipped?: number;
running?: number;
pending?: number;
};
}>({
method: "mesh.retry",
params: {
runId: parsed.runId,
...(parsed.stepIds && parsed.stepIds.length > 0 ? { stepIds: parsed.stepIds } : {}),
},
...commonGateway,
});
return {
shouldContinue: false,
reply: { text: `🔁 Retry submitted\n${formatRunSummary(retryResp)}` },
};
}
return null;
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return {
shouldContinue: false,
reply: {
text: `❌ Mesh command failed: ${message}`,
},
};
}
};

148
src/auto-reply/reply/commands.test.ts
View File

@@ -288,154 +288,6 @@ describe("/approve command", () => {
});
});
describe("/mesh command", () => {
beforeEach(() => {
vi.clearAllMocks();
callGatewayMock.mockReset();
});
it("shows usage for bare /mesh", async () => {
const cfg = {
commands: { text: true },
channels: { whatsapp: { allowFrom: ["*"] } },
} as OpenClawConfig;
const params = buildParams("/mesh", cfg);
const result = await handleCommands(params);
expect(result.shouldContinue).toBe(false);
expect(result.reply?.text).toContain("Mesh command");
expect(result.reply?.text).toContain("/mesh run <goal|mesh-plan-id>");
expect(callGatewayMock).not.toHaveBeenCalled();
});
it("runs auto plan + run for /mesh <goal>", async () => {
const cfg = {
commands: { text: true },
channels: { whatsapp: { allowFrom: ["*"] } },
} as OpenClawConfig;
const params = buildParams("/mesh build a landing animation", cfg);
callGatewayMock
.mockResolvedValueOnce({
plan: {
planId: "mesh-plan-1",
goal: "build a landing animation",
createdAt: Date.now(),
steps: [
{ id: "design", prompt: "Design animation" },
{ id: "mobile-test", prompt: "Test mobile", dependsOn: ["design"] },
],
},
order: ["design", "mobile-test"],
source: "llm",
})
.mockResolvedValueOnce({
runId: "mesh-run-1",
status: "completed",
stats: { total: 2, succeeded: 2, failed: 0, skipped: 0, running: 0, pending: 0 },
});
const result = await handleCommands(params);
expect(result.shouldContinue).toBe(false);
expect(result.reply?.text).toContain("Mesh Plan");
expect(result.reply?.text).toContain("Mesh Run");
expect(callGatewayMock).toHaveBeenNthCalledWith(
1,
expect.objectContaining({
method: "mesh.plan.auto",
params: expect.objectContaining({
goal: "build a landing animation",
}),
}),
);
expect(callGatewayMock).toHaveBeenNthCalledWith(
2,
expect.objectContaining({
method: "mesh.run",
}),
);
});
it("returns status via /mesh status <runId>", async () => {
const cfg = {
commands: { text: true },
channels: { whatsapp: { allowFrom: ["*"] } },
} as OpenClawConfig;
const params = buildParams("/mesh status mesh-run-77", cfg);
callGatewayMock.mockResolvedValueOnce({
runId: "mesh-run-77",
status: "failed",
stats: { total: 3, succeeded: 1, failed: 1, skipped: 1, running: 0, pending: 0 },
});
const result = await handleCommands(params);
expect(result.shouldContinue).toBe(false);
expect(result.reply?.text).toContain("Run: mesh-run-77");
expect(result.reply?.text).toContain("Status: failed");
expect(callGatewayMock).toHaveBeenCalledWith(
expect.objectContaining({
method: "mesh.status",
params: { runId: "mesh-run-77" },
}),
);
});
it("runs a previously planned mesh plan id without re-planning", async () => {
const cfg = {
commands: { text: true },
channels: { whatsapp: { allowFrom: ["*"] } },
} as OpenClawConfig;
const planParams = buildParams("/mesh plan Build Hero Animation", cfg);
callGatewayMock.mockResolvedValueOnce({
plan: {
planId: "mesh-plan-abc",
goal: "Build Hero Animation",
createdAt: Date.now(),
steps: [{ id: "design", prompt: "Design hero animation" }],
},
order: ["design"],
source: "llm",
});
const planResult = await handleCommands(planParams);
expect(planResult.shouldContinue).toBe(false);
expect(planResult.reply?.text).toContain("Run exact plan: /mesh run mesh-plan-abc");
expect(callGatewayMock).toHaveBeenCalledTimes(1);
expect(callGatewayMock).toHaveBeenCalledWith(
expect.objectContaining({
method: "mesh.plan.auto",
params: expect.objectContaining({
goal: "Build Hero Animation",
}),
}),
);
callGatewayMock.mockReset();
callGatewayMock.mockResolvedValueOnce({
runId: "mesh-run-abc",
status: "completed",
stats: { total: 1, succeeded: 1, failed: 0, skipped: 0, running: 0, pending: 0 },
});
const runParams = buildParams("/mesh run mesh-plan-abc", cfg);
const runResult = await handleCommands(runParams);
expect(runResult.shouldContinue).toBe(false);
expect(callGatewayMock).toHaveBeenCalledTimes(1);
expect(callGatewayMock).toHaveBeenCalledWith(
expect.objectContaining({
method: "mesh.run",
params: expect.objectContaining({
plan: expect.objectContaining({
planId: "mesh-plan-abc",
goal: "Build Hero Animation",
}),
}),
}),
);
});
});
describe("/compact command", () => {
beforeEach(() => {
vi.clearAllMocks();

5
src/gateway/protocol/index.ts
View File

@@ -130,8 +130,6 @@ import {
LogsTailResultSchema,
type MeshPlanParams,
MeshPlanParamsSchema,
type MeshPlanAutoParams,
MeshPlanAutoParamsSchema,
type MeshRetryParams,
MeshRetryParamsSchema,
type MeshRunParams,
@@ -371,7 +369,6 @@ export const validateExecApprovalsNodeSetParams = ajv.compile<ExecApprovalsNodeS
);
export const validateLogsTailParams = ajv.compile<LogsTailParams>(LogsTailParamsSchema);
export const validateMeshPlanParams = ajv.compile<MeshPlanParams>(MeshPlanParamsSchema);
export const validateMeshPlanAutoParams = ajv.compile<MeshPlanAutoParams>(MeshPlanAutoParamsSchema);
export const validateMeshRunParams = ajv.compile<MeshRunParams>(MeshRunParamsSchema);
export const validateMeshStatusParams = ajv.compile<MeshStatusParams>(MeshStatusParamsSchema);
export const validateMeshRetryParams = ajv.compile<MeshRetryParams>(MeshRetryParamsSchema);
@@ -435,7 +432,6 @@ export {
AgentEventSchema,
ChatEventSchema,
MeshPlanParamsSchema,
MeshPlanAutoParamsSchema,
MeshWorkflowPlanSchema,
MeshRunParamsSchema,
MeshStatusParamsSchema,
@@ -540,7 +536,6 @@ export type {
AgentWaitParams,
ChatEvent,
MeshPlanParams,
MeshPlanAutoParams,
MeshWorkflowPlan,
MeshRunParams,
MeshStatusParams,

14
src/gateway/protocol/schema/mesh.ts
View File

@@ -61,19 +61,6 @@ export const MeshRunParamsSchema = Type.Object(
{ additionalProperties: false },
);
export const MeshPlanAutoParamsSchema = Type.Object(
{
goal: NonEmptyString,
maxSteps: Type.Optional(Type.Integer({ minimum: 1, maximum: 16 })),
agentId: Type.Optional(NonEmptyString),
sessionKey: Type.Optional(NonEmptyString),
thinking: Type.Optional(Type.String()),
timeoutMs: Type.Optional(Type.Integer({ minimum: 1_000, maximum: 3_600_000 })),
lane: Type.Optional(Type.String()),
},
{ additionalProperties: false },
);
export const MeshStatusParamsSchema = Type.Object(
{
runId: NonEmptyString,
@@ -92,6 +79,5 @@ export const MeshRetryParamsSchema = Type.Object(
export type MeshPlanParams = Static<typeof MeshPlanParamsSchema>;
export type MeshWorkflowPlan = Static<typeof MeshWorkflowPlanSchema>;
export type MeshRunParams = Static<typeof MeshRunParamsSchema>;
export type MeshPlanAutoParams = Static<typeof MeshPlanAutoParamsSchema>;
export type MeshStatusParams = Static<typeof MeshStatusParamsSchema>;
export type MeshRetryParams = Static<typeof MeshRetryParamsSchema>;

2
src/gateway/protocol/schema/protocol-schemas.ts
View File

@@ -104,7 +104,6 @@ import {
LogsTailResultSchema,
} from "./logs-chat.js";
import {
MeshPlanAutoParamsSchema,
MeshPlanParamsSchema,
MeshRetryParamsSchema,
MeshRunParamsSchema,
@@ -263,7 +262,6 @@ export const ProtocolSchemas: Record<string, TSchema> = {
ChatInjectParams: ChatInjectParamsSchema,
ChatEvent: ChatEventSchema,
MeshPlanParams: MeshPlanParamsSchema,
MeshPlanAutoParams: MeshPlanAutoParamsSchema,
MeshWorkflowPlan: MeshWorkflowPlanSchema,
MeshRunParams: MeshRunParamsSchema,
MeshStatusParams: MeshStatusParamsSchema,

1
src/gateway/server-methods-list.ts
View File

@@ -86,7 +86,6 @@ const BASE_METHODS = [
"agent.identity.get",
"agent.wait",
"mesh.plan",
"mesh.plan.auto",
"mesh.run",
"mesh.status",
"mesh.retry",

1
src/gateway/server-methods.ts
View File

@@ -97,7 +97,6 @@ const WRITE_METHODS = new Set([
"chat.send",
"chat.abort",
"browser.request",
"mesh.plan.auto",
"mesh.run",
"mesh.retry",
]);

88
src/gateway/server-methods/mesh.test.ts
View File

@@ -5,7 +5,6 @@ import type { GatewayRequestContext } from "./types.js";
const mocks = vi.hoisted(() => ({
agent: vi.fn(),
agentWait: vi.fn(),
agentCommand: vi.fn(),
}));
vi.mock("./agent.js", () => ({
@@ -15,10 +14,6 @@ vi.mock("./agent.js", () => ({
},
}));
vi.mock("../../commands/agent.js", () => ({
agentCommand: (...args: unknown[]) => mocks.agentCommand(...args),
}));
const makeContext = (): GatewayRequestContext =>
({
dedupe: new Map(),
@@ -43,7 +38,6 @@ afterEach(() => {
__resetMeshRunsForTest();
mocks.agent.mockReset();
mocks.agentWait.mockReset();
mocks.agentCommand.mockReset();
});
describe("mesh handlers", () => {
@@ -147,86 +141,4 @@ describe("mesh handlers", () => {
const statusPayload = statusRes.payload as { status: string };
expect(statusPayload.status).toBe("completed");
});
it("auto planner creates multiple steps from llm json output", async () => {
mocks.agentCommand.mockResolvedValue({
payloads: [
{
text: JSON.stringify({
steps: [
{ id: "analyze", prompt: "Analyze requirements" },
{ id: "build", prompt: "Build implementation", dependsOn: ["analyze"] },
],
}),
},
],
meta: {},
});
const res = await callMesh("mesh.plan.auto", {
goal: "Create dashboard with auth",
maxSteps: 4,
});
expect(res.ok).toBe(true);
const payload = res.payload as {
source: string;
plan: { steps: Array<{ id: string }> };
order: string[];
};
expect(payload.source).toBe("llm");
expect(payload.plan.steps.map((s) => s.id)).toEqual(["analyze", "build"]);
expect(payload.order).toEqual(["analyze", "build"]);
expect(mocks.agentCommand).toHaveBeenCalledWith(
expect.objectContaining({
agentId: "main",
sessionKey: "agent:main:mesh-planner",
}),
expect.any(Object),
undefined,
);
});
it("auto planner falls back to single-step plan when llm output is invalid", async () => {
mocks.agentCommand.mockResolvedValue({
payloads: [{ text: "not valid json" }],
meta: {},
});
const res = await callMesh("mesh.plan.auto", {
goal: "Do a thing",
});
expect(res.ok).toBe(true);
const payload = res.payload as {
source: string;
plan: { steps: Array<{ id: string; prompt: string }> };
};
expect(payload.source).toBe("fallback");
expect(payload.plan.steps).toHaveLength(1);
expect(payload.plan.steps[0]?.prompt).toBe("Do a thing");
});
it("auto planner respects caller-provided planner session key", async () => {
mocks.agentCommand.mockResolvedValue({
payloads: [
{
text: JSON.stringify({
steps: [{ id: "one", prompt: "One" }],
}),
},
],
meta: {},
});
const res = await callMesh("mesh.plan.auto", {
goal: "Do a thing",
sessionKey: "agent:main:custom-planner",
});
expect(res.ok).toBe(true);
expect(mocks.agentCommand).toHaveBeenCalledWith(
expect.objectContaining({
sessionKey: "agent:main:custom-planner",
}),
expect.any(Object),
undefined,
);
});
});

257
src/gateway/server-methods/mesh.ts
View File

@@ -1,20 +1,17 @@
import { randomUUID } from "node:crypto";
import { agentCommand } from "../../commands/agent.js";
import { normalizeAgentId } from "../../routing/session-key.js";
import { defaultRuntime } from "../../runtime.js";
import type { GatewayRequestHandlerOptions, GatewayRequestHandlers, RespondFn } from "./types.js";
import {
ErrorCodes,
errorShape,
formatValidationErrors,
validateMeshPlanAutoParams,
validateMeshPlanParams,
validateMeshRetryParams,
validateMeshRunParams,
validateMeshStatusParams,
type MeshRunParams,
type MeshWorkflowPlan,
} from "../protocol/index.js";
import { agentHandlers } from "./agent.js";
import type { GatewayRequestHandlerOptions, GatewayRequestHandlers, RespondFn } from "./types.js";
type MeshStepStatus = "pending" | "running" | "succeeded" | "failed" | "skipped";
type MeshRunStatus = "pending" | "running" | "completed" | "failed";
@@ -51,51 +48,20 @@ type MeshRunRecord = {
history: Array<{ ts: number; type: string; stepId?: string; data?: Record<string, unknown> }>;
};
type MeshAutoStep = {
id?: string;
name?: string;
prompt: string;
dependsOn?: string[];
agentId?: string;
sessionKey?: string;
thinking?: string;
timeoutMs?: number;
};
type MeshAutoPlanShape = {
steps?: MeshAutoStep[];
};
const meshRuns = new Map<string, MeshRunRecord>();
const MAX_KEEP_RUNS = 200;
const AUTO_PLAN_TIMEOUT_MS = 90_000;
const PLANNER_MAIN_KEY = "mesh-planner";
function trimMap() {
if (meshRuns.size <= MAX_KEEP_RUNS) {
return;
}
const sorted = [...meshRuns.values()].toSorted((a, b) => a.startedAt - b.startedAt);
const sorted = [...meshRuns.values()].sort((a, b) => a.startedAt - b.startedAt);
const overflow = meshRuns.size - MAX_KEEP_RUNS;
for (const stale of sorted.slice(0, overflow)) {
meshRuns.delete(stale.runId);
}
}
function stringifyUnknown(value: unknown): string {
if (typeof value === "string") {
return value;
}
if (value instanceof Error) {
return value.message;
}
try {
return JSON.stringify(value);
} catch {
return String(value);
}
}
function normalizeDependsOn(dependsOn: string[] | undefined): string[] {
if (!Array.isArray(dependsOn)) {
return [];
@@ -135,7 +101,19 @@ function normalizePlan(plan: MeshWorkflowPlan): MeshWorkflowPlan {
};
}
function createPlanFromParams(params: { goal: string; steps?: MeshAutoStep[] }): MeshWorkflowPlan {
function createPlanFromParams(params: {
goal: string;
steps?: Array<{
id?: string;
name?: string;
prompt: string;
dependsOn?: string[];
agentId?: string;
sessionKey?: string;
thinking?: string;
timeoutMs?: number;
}>;
}): MeshWorkflowPlan {
const now = Date.now();
const goal = params.goal.trim();
const sourceSteps = params.steps?.length
@@ -173,9 +151,7 @@ function createPlanFromParams(params: { goal: string; steps?: MeshAutoStep[] }):
};
}
function validatePlanGraph(
plan: MeshWorkflowPlan,
): { ok: true; order: string[] } | { ok: false; error: string } {
function validatePlanGraph(plan: MeshWorkflowPlan): { ok: true; order: string[] } | { ok: false; error: string } {
const ids = new Set<string>();
for (const step of plan.steps) {
if (ids.has(step.id)) {
@@ -242,12 +218,7 @@ async function callGatewayHandler(
): Promise<{ ok: boolean; payload?: unknown; error?: unknown; meta?: Record<string, unknown> }> {
return await new Promise((resolve) => {
let settled = false;
const settle = (result: {
ok: boolean;
payload?: unknown;
error?: unknown;
meta?: Record<string, unknown>;
}) => {
const settle = (result: { ok: boolean; payload?: unknown; error?: unknown; meta?: Record<string, unknown> }) => {
if (settled) {
return;
}
@@ -328,7 +299,7 @@ async function executeStep(params: {
if (!accepted.ok) {
step.status = "failed";
step.endedAt = Date.now();
step.error = stringifyUnknown(accepted.error ?? "agent request failed");
step.error = String(accepted.error ?? "agent request failed");
run.history.push({
ts: Date.now(),
type: "step.error",
@@ -385,7 +356,7 @@ async function executeStep(params: {
step.error =
typeof waitPayload?.error === "string"
? waitPayload.error
: stringifyUnknown(waited.error ?? `agent.wait returned status ${waitStatus}`);
: String(waited.error ?? `agent.wait returned status ${waitStatus}`);
run.history.push({
ts: Date.now(),
type: "step.error",
@@ -460,7 +431,6 @@ async function runWorkflow(run: MeshRunRecord, opts: GatewayRequestHandlerOption
const inFlight = new Set<Promise<void>>();
let stopScheduling = false;
while (true) {
const failed = Object.values(run.steps).some((step) => step.status === "failed");
if (failed && !run.continueOnError) {
@@ -489,7 +459,6 @@ async function runWorkflow(run: MeshRunRecord, opts: GatewayRequestHandlerOption
if (pending.length === 0) {
break;
}
for (const step of pending) {
step.status = "skipped";
step.endedAt = Date.now();
@@ -578,130 +547,6 @@ function summarizeRun(run: MeshRunRecord) {
};
}
function extractTextFromAgentResult(result: unknown): string {
const payloads = (result as { payloads?: Array<{ text?: unknown }> } | undefined)?.payloads;
if (!Array.isArray(payloads)) {
return "";
}
const texts: string[] = [];
for (const payload of payloads) {
if (typeof payload?.text === "string" && payload.text.trim()) {
texts.push(payload.text.trim());
}
}
return texts.join("\n\n");
}
function parseJsonObjectFromText(text: string): Record<string, unknown> | null {
const trimmed = text.trim();
if (!trimmed) {
return null;
}
try {
const parsed = JSON.parse(trimmed);
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
? (parsed as Record<string, unknown>)
: null;
} catch {
// keep trying
}
const fenceMatch = trimmed.match(/```(?:json)?\s*([\s\S]*?)\s*```/i);
if (fenceMatch?.[1]) {
try {
const parsed = JSON.parse(fenceMatch[1]);
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
? (parsed as Record<string, unknown>)
: null;
} catch {
// keep trying
}
}
const start = trimmed.indexOf("{");
const end = trimmed.lastIndexOf("}");
if (start >= 0 && end > start) {
const candidate = trimmed.slice(start, end + 1);
try {
const parsed = JSON.parse(candidate);
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
? (parsed as Record<string, unknown>)
: null;
} catch {
return null;
}
}
return null;
}
function buildAutoPlannerPrompt(params: { goal: string; maxSteps: number }) {
return [
"You are a workflow planner. Convert the user's goal into executable workflow steps.",
"Return STRICT JSON only, no markdown, no prose.",
'JSON schema: {"steps": [{"id": string, "name"?: string, "prompt": string, "dependsOn"?: string[]}]}',
"Rules:",
`- Use 2 to ${params.maxSteps} steps.`,
"- Keep ids short, lowercase, kebab-case.",
"- dependsOn must reference earlier step ids when needed.",
"- prompts must be concrete and executable by an AI coding assistant.",
"- Do not include extra fields.",
`Goal: ${params.goal}`,
].join("\n");
}
async function generateAutoPlan(params: {
goal: string;
maxSteps: number;
agentId?: string;
sessionKey?: string;
thinking?: string;
timeoutMs?: number;
lane?: string;
opts: GatewayRequestHandlerOptions;
}): Promise<{ plan: MeshWorkflowPlan; source: "llm" | "fallback"; plannerText?: string }> {
const prompt = buildAutoPlannerPrompt({ goal: params.goal, maxSteps: params.maxSteps });
const timeoutSeconds = Math.ceil((params.timeoutMs ?? AUTO_PLAN_TIMEOUT_MS) / 1000);
const resolvedAgentId = normalizeAgentId(params.agentId ?? "main");
const plannerSessionKey =
params.sessionKey?.trim() || `agent:${resolvedAgentId}:${PLANNER_MAIN_KEY}`;
try {
const runResult = await agentCommand(
{
message: prompt,
deliver: false,
timeout: String(timeoutSeconds),
agentId: resolvedAgentId,
sessionKey: plannerSessionKey,
...(params.thinking ? { thinking: params.thinking } : {}),
...(params.lane ? { lane: params.lane } : {}),
},
defaultRuntime,
params.opts.context.deps,
);
const text = extractTextFromAgentResult(runResult);
const parsed = parseJsonObjectFromText(text) as MeshAutoPlanShape | null;
const rawSteps = Array.isArray(parsed?.steps) ? parsed.steps : [];
if (rawSteps.length > 0) {
const plan = normalizePlan(
createPlanFromParams({
goal: params.goal,
steps: rawSteps.slice(0, params.maxSteps),
}),
);
return { plan, source: "llm", plannerText: text };
}
const fallbackPlan = normalizePlan(createPlanFromParams({ goal: params.goal }));
return { plan: fallbackPlan, source: "fallback", plannerText: text };
} catch {
const fallbackPlan = normalizePlan(createPlanFromParams({ goal: params.goal }));
return { plan: fallbackPlan, source: "fallback" };
}
}
export const meshHandlers: GatewayRequestHandlers = {
"mesh.plan": ({ params, respond }) => {
if (!validateMeshPlanParams(params)) {
@@ -736,56 +581,6 @@ export const meshHandlers: GatewayRequestHandlers = {
undefined,
);
},
"mesh.plan.auto": async ({ params, respond, ...rest }) => {
if (!validateMeshPlanAutoParams(params)) {
respond(
false,
undefined,
errorShape(
ErrorCodes.INVALID_REQUEST,
`invalid mesh.plan.auto params: ${formatValidationErrors(validateMeshPlanAutoParams.errors)}`,
),
);
return;
}
const p = params;
const maxSteps =
typeof p.maxSteps === "number" && Number.isFinite(p.maxSteps)
? Math.max(1, Math.min(16, Math.floor(p.maxSteps)))
: 6;
const auto = await generateAutoPlan({
goal: p.goal,
maxSteps,
agentId: p.agentId,
sessionKey: p.sessionKey,
thinking: p.thinking,
timeoutMs: p.timeoutMs,
lane: p.lane,
opts: {
...rest,
params,
respond,
},
});
const graph = validatePlanGraph(auto.plan);
if (!graph.ok) {
respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, graph.error));
return;
}
respond(
true,
{
plan: auto.plan,
order: graph.order,
source: auto.source,
plannerText: auto.plannerText,
},
undefined,
);
},
"mesh.run": async (opts) => {
const { params, respond } = opts;
if (!validateMeshRunParams(params)) {
@@ -799,7 +594,7 @@ export const meshHandlers: GatewayRequestHandlers = {
);
return;
}
const p = params;
const p = params as MeshRunParams;
const plan = normalizePlan(p.plan);
const graph = validatePlanGraph(plan);
if (!graph.ok) {
@@ -845,7 +640,7 @@ export const meshHandlers: GatewayRequestHandlers = {
}
const run = meshRuns.get(params.runId.trim());
if (!run) {
respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "mesh run not found"));
respond(false, undefined, errorShape(ErrorCodes.NOT_FOUND, "mesh run not found"));
return;
}
respond(true, summarizeRun(run), undefined);
@@ -866,15 +661,11 @@ export const meshHandlers: GatewayRequestHandlers = {
const runId = params.runId.trim();
const run = meshRuns.get(runId);
if (!run) {
respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "mesh run not found"));
respond(false, undefined, errorShape(ErrorCodes.NOT_FOUND, "mesh run not found"));
return;
}
if (run.status === "running") {
respond(
false,
undefined,
errorShape(ErrorCodes.UNAVAILABLE, "mesh run is currently running"),
);
respond(false, undefined, errorShape(ErrorCodes.UNAVAILABLE, "mesh run is currently running"));
return;
}
const stepIds = resolveStepIdsForRetry(run, params.stepIds);

170
src/gateway/server-methods/server-methods.test.ts
View File

@@ -25,17 +25,6 @@ type HealthStatusHandlerParams = Parameters<
>[0];
describe("waitForAgentJob", () => {
const AGENT_RUN_ERROR_RETRY_GRACE_MS = 15_000;
beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
});
afterEach(() => {
vi.useRealTimers();
});
it("maps lifecycle end events with aborted=true to timeout", async () => {
const runId = `run-timeout-${Date.now()}-${Math.random().toString(36).slice(2)}`;
const waitPromise = waitForAgentJob({ runId, timeoutMs: 1_000 });
@@ -67,86 +56,6 @@ describe("waitForAgentJob", () => {
expect(snapshot?.startedAt).toBe(300);
expect(snapshot?.endedAt).toBe(400);
});
it("treats transient error->start->end as recovered when restart lands inside grace", async () => {
const runId = `run-recover-${Date.now()}-${Math.random().toString(36).slice(2)}`;
const waitPromise = waitForAgentJob({ runId, timeoutMs: 60_000 });
emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "start", startedAt: 100 } });
emitAgentEvent({
runId,
stream: "lifecycle",
data: { phase: "error", endedAt: 110, error: "transient" },
});
await vi.advanceTimersByTimeAsync(1_000);
emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "start", startedAt: 200 } });
emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "end", endedAt: 260 } });
const snapshot = await waitPromise;
expect(snapshot).not.toBeNull();
expect(snapshot?.status).toBe("ok");
expect(snapshot?.startedAt).toBe(200);
expect(snapshot?.endedAt).toBe(260);
});
it("resolves error only after grace expires when no recovery start arrives", async () => {
const runId = `run-error-${Date.now()}-${Math.random().toString(36).slice(2)}`;
const waitPromise = waitForAgentJob({ runId, timeoutMs: 60_000 });
emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "start", startedAt: 10 } });
emitAgentEvent({
runId,
stream: "lifecycle",
data: { phase: "error", endedAt: 20, error: "fatal" },
});
let settled = false;
void waitPromise.finally(() => {
settled = true;
});
await vi.advanceTimersByTimeAsync(AGENT_RUN_ERROR_RETRY_GRACE_MS - 1);
expect(settled).toBe(false);
await vi.advanceTimersByTimeAsync(1);
const snapshot = await waitPromise;
expect(snapshot).not.toBeNull();
expect(snapshot?.status).toBe("error");
expect(snapshot?.error).toBe("fatal");
expect(snapshot?.startedAt).toBe(10);
expect(snapshot?.endedAt).toBe(20);
});
it("honors pending error grace when waiter attaches after the error event", async () => {
const runId = `run-late-wait-${Date.now()}-${Math.random().toString(36).slice(2)}`;
emitAgentEvent({ runId, stream: "lifecycle", data: { phase: "start", startedAt: 900 } });
emitAgentEvent({
runId,
stream: "lifecycle",
data: { phase: "error", endedAt: 999, error: "late-listener" },
});
await vi.advanceTimersByTimeAsync(5_000);
const waitPromise = waitForAgentJob({ runId, timeoutMs: 60_000 });
let settled = false;
void waitPromise.finally(() => {
settled = true;
});
await vi.advanceTimersByTimeAsync(AGENT_RUN_ERROR_RETRY_GRACE_MS - 5_001);
expect(settled).toBe(false);
await vi.advanceTimersByTimeAsync(1);
const snapshot = await waitPromise;
expect(snapshot).not.toBeNull();
expect(snapshot?.status).toBe("error");
expect(snapshot?.error).toBe("late-listener");
expect(snapshot?.startedAt).toBe(900);
expect(snapshot?.endedAt).toBe(999);
});
});
describe("injectTimestamp", () => {
@@ -331,12 +240,10 @@ describe("gateway chat transcript writes (guardrail)", () => {
});
describe("exec approval handlers", () => {
const execApprovalNoop = () => false;
const execApprovalNoop = () => {};
type ExecApprovalHandlers = ReturnType<typeof createExecApprovalHandlers>;
type ExecApprovalRequestArgs = Parameters<ExecApprovalHandlers["exec.approval.request"]>[0];
type ExecApprovalResolveArgs = Parameters<ExecApprovalHandlers["exec.approval.resolve"]>[0];
type ExecApprovalRequestRespond = ExecApprovalRequestArgs["respond"];
type ExecApprovalResolveRespond = ExecApprovalResolveArgs["respond"];
const defaultExecApprovalRequestParams = {
command: "echo ok",
@@ -359,7 +266,7 @@ describe("exec approval handlers", () => {
async function requestExecApproval(params: {
handlers: ExecApprovalHandlers;
respond: ExecApprovalRequestRespond;
respond: ReturnType<typeof vi.fn>;
context: { broadcast: (event: string, payload: unknown) => void };
params?: Record<string, unknown>;
}) {
@@ -380,24 +287,14 @@ describe("exec approval handlers", () => {
async function resolveExecApproval(params: {
handlers: ExecApprovalHandlers;
id: string;
respond: ExecApprovalResolveRespond;
respond: ReturnType<typeof vi.fn>;
context: { broadcast: (event: string, payload: unknown) => void };
}) {
return params.handlers["exec.approval.resolve"]({
params: { id: params.id, decision: "allow-once" } as ExecApprovalResolveArgs["params"],
respond: params.respond,
context: toExecApprovalResolveContext(params.context),
client: {
connect: {
client: {
id: "cli",
displayName: "CLI",
version: "1.0.0",
platform: "test",
mode: "cli",
},
},
} as unknown as ExecApprovalResolveArgs["client"],
client: { connect: { client: { id: "cli", displayName: "CLI" } } },
req: { id: "req-2", type: "req", method: "exec.approval.resolve" },
isWebchatConnect: execApprovalNoop,
});
@@ -407,7 +304,7 @@ describe("exec approval handlers", () => {
const manager = new ExecApprovalManager();
const handlers = createExecApprovalHandlers(manager);
const broadcasts: Array<{ event: string; payload: unknown }> = [];
const respond = vi.fn() as unknown as ExecApprovalRequestRespond;
const respond = vi.fn();
const context = {
broadcast: (event: string, payload: unknown) => {
broadcasts.push({ event, payload });
@@ -478,7 +375,7 @@ describe("exec approval handlers", () => {
undefined,
);
const resolveRespond = vi.fn() as unknown as ExecApprovalResolveRespond;
const resolveRespond = vi.fn();
await resolveExecApproval({
handlers,
id,
@@ -501,7 +398,7 @@ describe("exec approval handlers", () => {
const manager = new ExecApprovalManager();
const handlers = createExecApprovalHandlers(manager);
const respond = vi.fn();
const resolveRespond = vi.fn() as unknown as ExecApprovalResolveRespond;
const resolveRespond = vi.fn();
const resolveContext = {
broadcast: () => {},
@@ -582,7 +479,7 @@ describe("gateway healthHandlers.status scope handling", () => {
await healthHandlers.status({
respond,
client: { connect: { role: "operator", scopes: ["operator.read"] } },
} as unknown as HealthStatusHandlerParams);
} as HealthStatusHandlerParams);
expect(vi.mocked(status.getStatusSummary)).toHaveBeenCalledWith({ includeSensitive: false });
expect(respond).toHaveBeenCalledWith(true, { ok: true }, undefined);
@@ -596,62 +493,13 @@ describe("gateway healthHandlers.status scope handling", () => {
await healthHandlers.status({
respond,
client: { connect: { role: "operator", scopes: ["operator.admin"] } },
} as unknown as HealthStatusHandlerParams);
} as HealthStatusHandlerParams);
expect(vi.mocked(status.getStatusSummary)).toHaveBeenCalledWith({ includeSensitive: true });
expect(respond).toHaveBeenCalledWith(true, { ok: true }, undefined);
});
});
describe("gateway mesh.plan.auto scope handling", () => {
it("rejects operator.read clients for mesh.plan.auto", async () => {
const { handleGatewayRequest } = await import("../server-methods.js");
const respond = vi.fn();
const handler = vi.fn();
await handleGatewayRequest({
req: { id: "req-mesh-read", type: "req", method: "mesh.plan.auto", params: {} },
respond,
context: {} as Parameters<typeof handleGatewayRequest>[0]["context"],
client: { connect: { role: "operator", scopes: ["operator.read"] } } as unknown as Parameters<
typeof handleGatewayRequest
>[0]["client"],
isWebchatConnect: () => false,
extraHandlers: { "mesh.plan.auto": handler },
});
expect(handler).not.toHaveBeenCalled();
expect(respond).toHaveBeenCalledWith(
false,
undefined,
expect.objectContaining({ message: "missing scope: operator.write" }),
);
});
it("allows operator.write clients for mesh.plan.auto", async () => {
const { handleGatewayRequest } = await import("../server-methods.js");
const respond = vi.fn();
const handler = vi.fn(
({ respond: send }: { respond: (ok: boolean, payload?: unknown) => void }) =>
send(true, { ok: true }),
);
await handleGatewayRequest({
req: { id: "req-mesh-write", type: "req", method: "mesh.plan.auto", params: {} },
respond,
context: {} as Parameters<typeof handleGatewayRequest>[0]["context"],
client: {
connect: { role: "operator", scopes: ["operator.write"] },
} as unknown as Parameters<typeof handleGatewayRequest>[0]["client"],
isWebchatConnect: () => false,
extraHandlers: { "mesh.plan.auto": handler },
});
expect(handler).toHaveBeenCalledOnce();
expect(respond).toHaveBeenCalledWith(true, { ok: true });
});
});
describe("logs.tail", () => {
const logsNoop = () => false;