diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 9741ea8d0b..8eb79881ec 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -137,10 +137,13 @@ Key behaviors: - Prompt is prefixed with `[cron: ]` for traceability. - Each run starts a **fresh session id** (no prior conversation carry-over). -- Default behavior: if `delivery` is omitted, isolated jobs announce a summary immediately (`delivery.mode = "announce"`). +- Default behavior: if `delivery` is omitted, isolated jobs announce a summary (`delivery.mode = "announce"`). - `delivery.mode` (isolated-only) chooses what happens: - - `announce`: subagent-style summary delivered immediately to a chat. - - `none`: internal only (no delivery). + - `announce`: deliver a summary to the target channel and post a brief summary to the main session. + - `none`: internal only (no delivery, no main-session summary). +- `wakeMode` controls when the main-session summary posts: + - `now`: immediate heartbeat. + - `next-heartbeat`: waits for the next scheduled heartbeat. Use isolated jobs for noisy, frequent, or "background chores" that shouldn't spam your main chat history. @@ -166,10 +169,27 @@ Delivery config (isolated jobs only): - `delivery.bestEffort`: avoid failing the job if announce delivery fails. Announce delivery suppresses messaging tool sends for the run; use `delivery.channel`/`delivery.to` -to target the chat instead. +to target the chat instead. When `delivery.mode = "none"`, no summary is posted to the main session. If `delivery` is omitted for isolated jobs, OpenClaw defaults to `announce`. +#### Announce delivery flow + +When `delivery.mode = "announce"`, cron delivers directly via the outbound channel adapters. +The main agent is not spun up to craft or forward the message. + +Behavior details: + +- Content: delivery uses the isolated run's outbound payloads (text/media) with normal chunking and + channel formatting. +- Heartbeat-only responses (`HEARTBEAT_OK` with no real content) are not delivered. +- If the isolated run already sent a message to the same target via the message tool, delivery is + skipped to avoid duplicates. +- Missing or invalid delivery targets fail the job unless `delivery.bestEffort = true`. +- A short summary is posted to the main session only when `delivery.mode = "announce"`. +- The main-session summary respects `wakeMode`: `now` triggers an immediate heartbeat and + `next-heartbeat` waits for the next scheduled heartbeat. + ### Model and thinking overrides Isolated jobs (`agentTurn`) can override the model and thinking level: @@ -191,7 +211,7 @@ Resolution priority: Isolated jobs can deliver output to a channel via the top-level `delivery` config: -- `delivery.mode`: `announce` (subagent-style summary) or `none`. +- `delivery.mode`: `announce` (deliver a summary) or `none`. - `delivery.channel`: `whatsapp` / `telegram` / `discord` / `slack` / `mattermost` (plugin) / `signal` / `imessage` / `last`. - `delivery.to`: channel-specific recipient target. diff --git a/src/cron/isolated-agent.delivers-response-has-heartbeat-ok-but-includes.test.ts b/src/cron/isolated-agent.delivers-response-has-heartbeat-ok-but-includes.test.ts index b74b52d888..5d3a7caf2b 100644 --- a/src/cron/isolated-agent.delivers-response-has-heartbeat-ok-but-includes.test.ts +++ b/src/cron/isolated-agent.delivers-response-has-heartbeat-ok-but-includes.test.ts @@ -5,6 +5,9 @@ import type { CliDeps } from "../cli/deps.js"; import type { OpenClawConfig } from "../config/config.js"; import type { CronJob } from "./types.js"; import { withTempHome as withTempHomeBase } from "../../test/helpers/temp-home.js"; +import { telegramOutbound } from "../channels/plugins/outbound/telegram.js"; +import { setActivePluginRegistry } from "../plugins/runtime.js"; +import { createOutboundTestPlugin, createTestRegistry } from "../test-utils/channel-plugins.js"; vi.mock("../agents/pi-embedded.js", () => ({ abortEmbeddedPiRun: vi.fn().mockReturnValue(false), @@ -14,13 +17,9 @@ vi.mock("../agents/pi-embedded.js", () => ({ vi.mock("../agents/model-catalog.js", () => ({ loadModelCatalog: vi.fn(), })); -vi.mock("../agents/subagent-announce.js", () => ({ - runSubagentAnnounceFlow: vi.fn(), -})); import { loadModelCatalog } from "../agents/model-catalog.js"; import { runEmbeddedPiAgent } from "../agents/pi-embedded.js"; -import { runSubagentAnnounceFlow } from "../agents/subagent-announce.js"; import { runCronIsolatedAgentTurn } from "./isolated-agent.js"; async function withTempHome(fn: (home: string) => Promise): Promise { @@ -87,7 +86,15 @@ describe("runCronIsolatedAgentTurn", () => { beforeEach(() => { vi.mocked(runEmbeddedPiAgent).mockReset(); vi.mocked(loadModelCatalog).mockResolvedValue([]); - vi.mocked(runSubagentAnnounceFlow).mockReset().mockResolvedValue(true); + setActivePluginRegistry( + createTestRegistry([ + { + pluginId: "telegram", + plugin: createOutboundTestPlugin({ id: "telegram", outbound: telegramOutbound }), + source: "test", + }, + ]), + ); }); it("delivers when response has HEARTBEAT_OK but includes media", async () => { @@ -128,7 +135,7 @@ describe("runCronIsolatedAgentTurn", () => { }); expect(res.status).toBe("ok"); - expect(runSubagentAnnounceFlow).toHaveBeenCalledTimes(1); + expect(deps.sendMessageTelegram).toHaveBeenCalled(); }); }); @@ -178,7 +185,7 @@ describe("runCronIsolatedAgentTurn", () => { }); expect(res.status).toBe("ok"); - expect(runSubagentAnnounceFlow).toHaveBeenCalledTimes(1); + expect(deps.sendMessageTelegram).toHaveBeenCalled(); }); }); }); diff --git a/src/cron/isolated-agent.skips-delivery-without-whatsapp-recipient-besteffortdeliver-true.test.ts b/src/cron/isolated-agent.skips-delivery-without-whatsapp-recipient-besteffortdeliver-true.test.ts index adedfba715..6aac38f88d 100644 --- a/src/cron/isolated-agent.skips-delivery-without-whatsapp-recipient-besteffortdeliver-true.test.ts +++ b/src/cron/isolated-agent.skips-delivery-without-whatsapp-recipient-besteffortdeliver-true.test.ts @@ -4,16 +4,10 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { CliDeps } from "../cli/deps.js"; import type { OpenClawConfig } from "../config/config.js"; import type { CronJob } from "./types.js"; -import { discordPlugin } from "../../extensions/discord/src/channel.js"; -import { setDiscordRuntime } from "../../extensions/discord/src/runtime.js"; -import { telegramPlugin } from "../../extensions/telegram/src/channel.js"; -import { setTelegramRuntime } from "../../extensions/telegram/src/runtime.js"; -import { whatsappPlugin } from "../../extensions/whatsapp/src/channel.js"; -import { setWhatsAppRuntime } from "../../extensions/whatsapp/src/runtime.js"; import { withTempHome as withTempHomeBase } from "../../test/helpers/temp-home.js"; +import { telegramOutbound } from "../channels/plugins/outbound/telegram.js"; import { setActivePluginRegistry } from "../plugins/runtime.js"; -import { createPluginRuntime } from "../plugins/runtime/index.js"; -import { createTestRegistry } from "../test-utils/channel-plugins.js"; +import { createOutboundTestPlugin, createTestRegistry } from "../test-utils/channel-plugins.js"; vi.mock("../agents/pi-embedded.js", () => ({ abortEmbeddedPiRun: vi.fn().mockReturnValue(false), @@ -23,13 +17,9 @@ vi.mock("../agents/pi-embedded.js", () => ({ vi.mock("../agents/model-catalog.js", () => ({ loadModelCatalog: vi.fn(), })); -vi.mock("../agents/subagent-announce.js", () => ({ - runSubagentAnnounceFlow: vi.fn(), -})); import { loadModelCatalog } from "../agents/model-catalog.js"; import { runEmbeddedPiAgent } from "../agents/pi-embedded.js"; -import { runSubagentAnnounceFlow } from "../agents/subagent-announce.js"; import { runCronIsolatedAgentTurn } from "./isolated-agent.js"; async function withTempHome(fn: (home: string) => Promise): Promise { @@ -96,16 +86,13 @@ describe("runCronIsolatedAgentTurn", () => { beforeEach(() => { vi.mocked(runEmbeddedPiAgent).mockReset(); vi.mocked(loadModelCatalog).mockResolvedValue([]); - vi.mocked(runSubagentAnnounceFlow).mockReset().mockResolvedValue(true); - const runtime = createPluginRuntime(); - setDiscordRuntime(runtime); - setTelegramRuntime(runtime); - setWhatsAppRuntime(runtime); setActivePluginRegistry( createTestRegistry([ - { pluginId: "whatsapp", plugin: whatsappPlugin, source: "test" }, - { pluginId: "telegram", plugin: telegramPlugin, source: "test" }, - { pluginId: "discord", plugin: discordPlugin, source: "test" }, + { + pluginId: "telegram", + plugin: createOutboundTestPlugin({ id: "telegram", outbound: telegramOutbound }), + source: "test", + }, ]), ); }); @@ -143,9 +130,7 @@ describe("runCronIsolatedAgentTurn", () => { }); expect(res.status).toBe("ok"); - expect(runSubagentAnnounceFlow).toHaveBeenCalledTimes(1); - const call = vi.mocked(runSubagentAnnounceFlow).mock.calls[0]?.[0]; - expect(call?.label).toBe("Cron: job-1"); + expect(deps.sendMessageTelegram).toHaveBeenCalled(); }); }); @@ -184,7 +169,7 @@ describe("runCronIsolatedAgentTurn", () => { }); expect(res.status).toBe("ok"); - expect(runSubagentAnnounceFlow).not.toHaveBeenCalled(); + expect(deps.sendMessageTelegram).not.toHaveBeenCalled(); }); }); @@ -221,7 +206,7 @@ describe("runCronIsolatedAgentTurn", () => { }); expect(res.status).toBe("ok"); - expect(runSubagentAnnounceFlow).not.toHaveBeenCalled(); + expect(deps.sendMessageTelegram).not.toHaveBeenCalled(); }); }); @@ -230,7 +215,7 @@ describe("runCronIsolatedAgentTurn", () => { const storePath = await writeSessionStore(home); const deps: CliDeps = { sendMessageWhatsApp: vi.fn(), - sendMessageTelegram: vi.fn(), + sendMessageTelegram: vi.fn().mockRejectedValue(new Error("boom")), sendMessageDiscord: vi.fn(), sendMessageSignal: vi.fn(), sendMessageIMessage: vi.fn(), @@ -242,8 +227,6 @@ describe("runCronIsolatedAgentTurn", () => { agentMeta: { sessionId: "s", provider: "p", model: "m" }, }, }); - vi.mocked(runSubagentAnnounceFlow).mockResolvedValue(false); - const res = await runCronIsolatedAgentTurn({ cfg: makeCfg(home, storePath, { channels: { telegram: { botToken: "t-1" } }, @@ -259,7 +242,7 @@ describe("runCronIsolatedAgentTurn", () => { }); expect(res.status).toBe("error"); - expect(res.error).toBe("cron announce delivery failed"); + expect(res.error).toBe("Error: boom"); }); }); @@ -268,7 +251,7 @@ describe("runCronIsolatedAgentTurn", () => { const storePath = await writeSessionStore(home); const deps: CliDeps = { sendMessageWhatsApp: vi.fn(), - sendMessageTelegram: vi.fn(), + sendMessageTelegram: vi.fn().mockRejectedValue(new Error("boom")), sendMessageDiscord: vi.fn(), sendMessageSignal: vi.fn(), sendMessageIMessage: vi.fn(), @@ -280,8 +263,6 @@ describe("runCronIsolatedAgentTurn", () => { agentMeta: { sessionId: "s", provider: "p", model: "m" }, }, }); - vi.mocked(runSubagentAnnounceFlow).mockResolvedValue(false); - const res = await runCronIsolatedAgentTurn({ cfg: makeCfg(home, storePath, { channels: { telegram: { botToken: "t-1" } }, @@ -302,7 +283,7 @@ describe("runCronIsolatedAgentTurn", () => { }); expect(res.status).toBe("ok"); - expect(runSubagentAnnounceFlow).toHaveBeenCalledTimes(1); + expect(deps.sendMessageTelegram).toHaveBeenCalled(); }); }); }); diff --git a/src/cron/isolated-agent/run.ts b/src/cron/isolated-agent/run.ts index b0eb580de9..0ac7013a26 100644 --- a/src/cron/isolated-agent/run.ts +++ b/src/cron/isolated-agent/run.ts @@ -31,10 +31,6 @@ import { import { runEmbeddedPiAgent } from "../../agents/pi-embedded.js"; import { buildWorkspaceSkillSnapshot } from "../../agents/skills.js"; import { getSkillsSnapshotVersion } from "../../agents/skills/refresh.js"; -import { - runSubagentAnnounceFlow, - type SubagentRunOutcome, -} from "../../agents/subagent-announce.js"; import { resolveAgentTimeoutMs } from "../../agents/timeout.js"; import { hasNonzeroUsage } from "../../agents/usage.js"; import { ensureAgentWorkspace } from "../../agents/workspace.js"; @@ -44,13 +40,10 @@ import { normalizeVerboseLevel, supportsXHighThinking, } from "../../auto-reply/thinking.js"; -import { type CliDeps } from "../../cli/outbound-send-deps.js"; -import { - resolveAgentMainSessionKey, - resolveSessionTranscriptPath, - updateSessionStore, -} from "../../config/sessions.js"; +import { createOutboundSendDeps, type CliDeps } from "../../cli/outbound-send-deps.js"; +import { resolveSessionTranscriptPath, updateSessionStore } from "../../config/sessions.js"; import { registerAgentRunContext } from "../../infra/agent-events.js"; +import { deliverOutboundPayloads } from "../../infra/outbound/deliver.js"; import { getRemoteSkillEligibility } from "../../infra/skills-remote.js"; import { logWarn } from "../../logger.js"; import { buildAgentMainSessionKey, normalizeAgentId } from "../../routing/session-key.js"; @@ -314,7 +307,7 @@ export async function runCronIsolatedAgentTurn(params: { } if (deliveryRequested) { commandBody = - `${commandBody}\n\nReturn your summary as plain text; it will be delivered by the main agent. If the task explicitly calls for messaging a specific external recipient, note who/where it should go instead of sending it yourself.`.trim(); + `${commandBody}\n\nReturn your summary as plain text; it will be delivered automatically. If the task explicitly calls for messaging a specific external recipient, note who/where it should go instead of sending it yourself.`.trim(); } const existingSnapshot = cronSession.sessionEntry.skillsSnapshot; @@ -480,42 +473,21 @@ export async function runCronIsolatedAgentTurn(params: { logWarn(`[cron:${params.job.id}] ${deliveryFailure.message}`); return { status: "ok", summary, outputText }; } - const requesterSessionKey = resolveAgentMainSessionKey({ - cfg: cfgWithAgentDefaults, - agentId, - }); - const useExplicitOrigin = deliveryPlan.channel !== "last" || Boolean(deliveryPlan.to?.trim()); - const requesterOrigin = useExplicitOrigin - ? { - channel: resolvedDelivery.channel, - to: resolvedDelivery.to, - accountId: resolvedDelivery.accountId, - threadId: resolvedDelivery.threadId, - } - : undefined; - const outcome: SubagentRunOutcome = { status: "ok" }; - const taskLabel = params.job.name?.trim() || "cron job"; - const didAnnounce = await runSubagentAnnounceFlow({ - childSessionKey: agentSessionKey, - childRunId: cronSession.sessionEntry.sessionId, - requesterSessionKey, - requesterOrigin, - requesterDisplayKey: requesterSessionKey, - task: taskLabel, - timeoutMs: 30_000, - cleanup: "keep", - roundOneReply: outputText ?? summary, - waitForCompletion: false, - label: `Cron: ${taskLabel}`, - outcome, - }); - if (!didAnnounce && !deliveryBestEffort) { - return { - status: "error", - error: "cron announce delivery failed", - summary, - outputText, - }; + try { + await deliverOutboundPayloads({ + cfg: cfgWithAgentDefaults, + channel: resolvedDelivery.channel, + to: resolvedDelivery.to, + accountId: resolvedDelivery.accountId, + threadId: resolvedDelivery.threadId, + payloads, + bestEffort: deliveryBestEffort, + deps: createOutboundSendDeps(params.deps), + }); + } catch (err) { + if (!deliveryBestEffort) { + return { status: "error", summary, outputText, error: String(err) }; + } } } diff --git a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts index 9acd8fe3ad..e26e71cab7 100644 --- a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts +++ b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts @@ -211,7 +211,8 @@ describe("CronService", () => { schedule: { kind: "at", at: new Date(atMs).toISOString() }, sessionTarget: "isolated", wakeMode: "now", - payload: { kind: "agentTurn", message: "do it", deliver: false }, + payload: { kind: "agentTurn", message: "do it" }, + delivery: { mode: "announce" }, }); vi.setSystemTime(new Date("2025-12-13T00:00:01.000Z")); @@ -359,7 +360,8 @@ describe("CronService", () => { schedule: { kind: "at", at: new Date(atMs).toISOString() }, sessionTarget: "isolated", wakeMode: "now", - payload: { kind: "agentTurn", message: "do it", deliver: false }, + payload: { kind: "agentTurn", message: "do it" }, + delivery: { mode: "announce" }, }); vi.setSystemTime(new Date("2025-12-13T00:00:01.000Z")); diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index 1151974e55..a4b33bf3c3 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -188,12 +188,15 @@ export async function executeJob( // Post a short summary back to the main session so the user sees // the cron result without opening the isolated session. const summaryText = res.summary?.trim(); - if (summaryText) { + const deliveryMode = job.delivery?.mode ?? "announce"; + if (summaryText && deliveryMode !== "none") { const prefix = "Cron"; const label = res.status === "error" ? `${prefix} (error): ${summaryText}` : `${prefix}: ${summaryText}`; state.deps.enqueueSystemEvent(label, { agentId: job.agentId }); - state.deps.requestHeartbeatNow({ reason: `cron:${job.id}` }); + if (job.wakeMode === "now") { + state.deps.requestHeartbeatNow({ reason: `cron:${job.id}` }); + } } if (res.status === "ok") { diff --git a/ui/src/ui/format.test.ts b/ui/src/ui/format.test.ts index 281651357d..8e1f121ea6 100644 --- a/ui/src/ui/format.test.ts +++ b/ui/src/ui/format.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it } from "vitest"; import { formatAgo, stripThinkingTags } from "./format.ts"; describe("formatAgo", () => {