let5see/clawdbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: multi-agent routing + multi-account providers

Browse Source
This commit is contained in:
Peter Steinberger
2026-01-06 18:25:37 +00:00
parent 50d4b17417
commit dbfa316d19
129 changed files with 3760 additions and 1126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/cron/cron-protocol-conformance.test.ts
View File

@@ -9,22 +9,22 @@ type SchemaLike = {
const?: unknown;
};
type ChannelSchema = {
type ProviderSchema = {
anyOf?: Array<{ const?: unknown }>;
};
function extractCronChannels(schema: SchemaLike): string[] {
function extractCronProviders(schema: SchemaLike): string[] {
const union = schema.anyOf ?? [];
const payloadWithChannel = union.find((entry) =>
Boolean(entry?.properties && "channel" in entry.properties),
const payloadWithProvider = union.find((entry) =>
Boolean(entry?.properties && "provider" in entry.properties),
);
const channelSchema = payloadWithChannel?.properties
? (payloadWithChannel.properties.channel as ChannelSchema)
const providerSchema = payloadWithProvider?.properties
? (payloadWithProvider.properties.provider as ProviderSchema)
: undefined;
const channels = (channelSchema?.anyOf ?? [])
const providers = (providerSchema?.anyOf ?? [])
.map((entry) => entry?.const)
.filter((value): value is string => typeof value === "string");
return channels;
return providers;
}
const UI_FILES = [
@@ -36,28 +36,28 @@ const UI_FILES = [
const SWIFT_FILES = ["apps/macos/Sources/Clawdbot/GatewayConnection.swift"];
describe("cron protocol conformance", () => {
it("ui + swift include all cron channels from gateway schema", async () => {
const channels = extractCronChannels(CronPayloadSchema as SchemaLike);
expect(channels.length).toBeGreaterThan(0);
it("ui + swift include all cron providers from gateway schema", async () => {
const providers = extractCronProviders(CronPayloadSchema as SchemaLike);
expect(providers.length).toBeGreaterThan(0);
const cwd = process.cwd();
for (const relPath of UI_FILES) {
const content = await fs.readFile(path.join(cwd, relPath), "utf-8");
for (const channel of channels) {
for (const provider of providers) {
expect(
content.includes(`"${channel}"`),
`${relPath} missing ${channel}`,
content.includes(`"${provider}"`),
`${relPath} missing ${provider}`,
).toBe(true);
}
}
for (const relPath of SWIFT_FILES) {
const content = await fs.readFile(path.join(cwd, relPath), "utf-8");
for (const channel of channels) {
const pattern = new RegExp(`\\bcase\\s+${channel}\\b`);
for (const provider of providers) {
const pattern = new RegExp(`\\bcase\\s+${provider}\\b`);
expect(
pattern.test(content),
`${relPath} missing case ${channel}`,
`${relPath} missing case ${provider}`,
).toBe(true);
}
}

22
src/cron/isolated-agent.test.ts
View File

@@ -42,10 +42,10 @@ async function writeSessionStore(home: string) {
storePath,
JSON.stringify(
{
main: {
"agent:main:main": {
sessionId: "main-session",
updatedAt: Date.now(),
lastChannel: "webchat",
lastProvider: "webchat",
lastTo: "",
},
},
@@ -224,7 +224,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "whatsapp",
provider: "whatsapp",
bestEffortDeliver: false,
}),
message: "do it",
@@ -264,7 +264,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "whatsapp",
provider: "whatsapp",
bestEffortDeliver: true,
}),
message: "do it",
@@ -309,7 +309,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "telegram",
provider: "telegram",
to: "123",
}),
message: "do it",
@@ -361,7 +361,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "discord",
provider: "discord",
to: "channel:1122",
}),
message: "do it",
@@ -406,7 +406,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "telegram",
provider: "telegram",
to: "123",
}),
message: "do it",
@@ -450,7 +450,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "whatsapp",
provider: "whatsapp",
to: "+1234",
}),
message: "do it",
@@ -493,7 +493,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "telegram",
provider: "telegram",
to: "123",
}),
message: "do it",
@@ -537,7 +537,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "telegram",
provider: "telegram",
to: "123",
}),
message: "do it",
@@ -585,7 +585,7 @@ describe("runCronIsolatedAgentTurn", () => {
kind: "agentTurn",
message: "do it",
deliver: true,
channel: "telegram",
provider: "telegram",
to: "123",
}),
message: "do it",

71
src/cron/isolated-agent.ts
View File

@@ -29,6 +29,8 @@ import type { ClawdbotConfig } from "../config/config.js";
import {
DEFAULT_IDLE_MINUTES,
loadSessionStore,
resolveAgentIdFromSessionKey,
resolveMainSessionKey,
resolveSessionTranscriptPath,
resolveStorePath,
type SessionEntry,
@@ -87,7 +89,7 @@ function isHeartbeatOnlyResponse(
function resolveDeliveryTarget(
cfg: ClawdbotConfig,
jobPayload: {
channel?:
provider?:
| "last"
| "whatsapp"
| "telegram"
@@ -98,36 +100,37 @@ function resolveDeliveryTarget(
to?: string;
},
) {
const requestedChannel =
typeof jobPayload.channel === "string" ? jobPayload.channel : "last";
const requestedProvider =
typeof jobPayload.provider === "string" ? jobPayload.provider : "last";
const explicitTo =
typeof jobPayload.to === "string" && jobPayload.to.trim()
? jobPayload.to.trim()
: undefined;
const sessionCfg = cfg.session;
const mainKey = (sessionCfg?.mainKey ?? "main").trim() || "main";
const storePath = resolveStorePath(sessionCfg?.store);
const mainSessionKey = resolveMainSessionKey(cfg);
const agentId = resolveAgentIdFromSessionKey(mainSessionKey);
const storePath = resolveStorePath(sessionCfg?.store, { agentId });
const store = loadSessionStore(storePath);
const main = store[mainKey];
const lastChannel =
main?.lastChannel && main.lastChannel !== "webchat"
? main.lastChannel
const main = store[mainSessionKey];
const lastProvider =
main?.lastProvider && main.lastProvider !== "webchat"
? main.lastProvider
: undefined;
const lastTo = typeof main?.lastTo === "string" ? main.lastTo.trim() : "";
const channel = (() => {
const provider = (() => {
if (
requestedChannel === "whatsapp" ||
requestedChannel === "telegram" ||
requestedChannel === "discord" ||
requestedChannel === "slack" ||
requestedChannel === "signal" ||
requestedChannel === "imessage"
requestedProvider === "whatsapp" ||
requestedProvider === "telegram" ||
requestedProvider === "discord" ||
requestedProvider === "slack" ||
requestedProvider === "signal" ||
requestedProvider === "imessage"
) {
return requestedChannel;
return requestedProvider;
}
return lastChannel ?? "whatsapp";
return lastProvider ?? "whatsapp";
})();
const to = (() => {
@@ -136,7 +139,7 @@ function resolveDeliveryTarget(
})();
const sanitizedWhatsappTo = (() => {
if (channel !== "whatsapp") return to;
if (provider !== "whatsapp") return to;
const rawAllow = cfg.whatsapp?.allowFrom ?? [];
if (rawAllow.includes("*")) return to;
const allowFrom = rawAllow
@@ -150,8 +153,8 @@ function resolveDeliveryTarget(
})();
return {
channel,
to: channel === "whatsapp" ? sanitizedWhatsappTo : to,
provider,
to: provider === "whatsapp" ? sanitizedWhatsappTo : to,
};
}
@@ -181,7 +184,7 @@ function resolveCronSession(params: {
model: entry?.model,
contextTokens: entry?.contextTokens,
sendPolicy: entry?.sendPolicy,
lastChannel: entry?.lastChannel,
lastProvider: entry?.lastProvider,
lastTo: entry?.lastTo,
};
return { storePath, store, sessionEntry, systemSent, isNewSession: !fresh };
@@ -251,9 +254,9 @@ export async function runCronIsolatedAgentTurn(params: {
params.job.payload.bestEffortDeliver === true;
const resolvedDelivery = resolveDeliveryTarget(params.cfg, {
channel:
provider:
params.job.payload.kind === "agentTurn"
? params.job.payload.channel
? params.job.payload.provider
: "last",
to:
params.job.payload.kind === "agentTurn"
@@ -302,7 +305,7 @@ export async function runCronIsolatedAgentTurn(params: {
registerAgentRunContext(cronSession.sessionEntry.sessionId, {
sessionKey: params.sessionKey,
});
const surface = resolvedDelivery.channel;
const messageProvider = resolvedDelivery.provider;
const fallbackResult = await runWithModelFallback({
cfg: params.cfg,
provider,
@@ -311,7 +314,7 @@ export async function runCronIsolatedAgentTurn(params: {
runEmbeddedPiAgent({
sessionId: cronSession.sessionEntry.sessionId,
sessionKey: params.sessionKey,
surface,
messageProvider,
sessionFile,
workspaceDir,
config: params.cfg,
@@ -380,7 +383,7 @@ export async function runCronIsolatedAgentTurn(params: {
delivery && isHeartbeatOnlyResponse(payloads, Math.max(0, ackMaxChars));
if (delivery && !skipHeartbeatDelivery) {
if (resolvedDelivery.channel === "whatsapp") {
if (resolvedDelivery.provider === "whatsapp") {
if (!resolvedDelivery.to) {
if (!bestEffortDeliver)
return {
@@ -415,7 +418,7 @@ export async function runCronIsolatedAgentTurn(params: {
return { status: "error", summary, error: String(err) };
return { status: "ok", summary };
}
} else if (resolvedDelivery.channel === "telegram") {
} else if (resolvedDelivery.provider === "telegram") {
if (!resolvedDelivery.to) {
if (!bestEffortDeliver)
return {
@@ -459,14 +462,14 @@ export async function runCronIsolatedAgentTurn(params: {
return { status: "error", summary, error: String(err) };
return { status: "ok", summary };
}
} else if (resolvedDelivery.channel === "discord") {
} else if (resolvedDelivery.provider === "discord") {
if (!resolvedDelivery.to) {
if (!bestEffortDeliver)
return {
status: "error",
summary,
error:
"Cron delivery to Discord requires --channel discord and --to <channelId|user:ID>",
"Cron delivery to Discord requires --provider discord and --to <channelId|user:ID>",
};
return {
status: "skipped",
@@ -503,14 +506,14 @@ export async function runCronIsolatedAgentTurn(params: {
return { status: "error", summary, error: String(err) };
return { status: "ok", summary };
}
} else if (resolvedDelivery.channel === "slack") {
} else if (resolvedDelivery.provider === "slack") {
if (!resolvedDelivery.to) {
if (!bestEffortDeliver)
return {
status: "error",
summary,
error:
"Cron delivery to Slack requires --channel slack and --to <channelId|user:ID>",
"Cron delivery to Slack requires --provider slack and --to <channelId|user:ID>",
};
return {
status: "skipped",
@@ -543,7 +546,7 @@ export async function runCronIsolatedAgentTurn(params: {
return { status: "error", summary, error: String(err) };
return { status: "ok", summary };
}
} else if (resolvedDelivery.channel === "signal") {
} else if (resolvedDelivery.provider === "signal") {
if (!resolvedDelivery.to) {
if (!bestEffortDeliver)
return {
@@ -582,7 +585,7 @@ export async function runCronIsolatedAgentTurn(params: {
return { status: "error", summary, error: String(err) };
return { status: "ok", summary };
}
} else if (resolvedDelivery.channel === "imessage") {
} else if (resolvedDelivery.provider === "imessage") {
if (!resolvedDelivery.to) {
if (!bestEffortDeliver)
return {

2
src/cron/types.ts
View File

@@ -14,7 +14,7 @@ export type CronPayload =
thinking?: string;
timeoutSeconds?: number;
deliver?: boolean;
channel?:
provider?:
| "last"
| "whatsapp"
| "telegram"