diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index 3c10b811a..8e590d319 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -1,24 +1,17 @@ import { resolveDefaultAgentId } from "../agents/agent-scope.js"; import type { CliDeps } from "../cli/deps.js"; import { loadConfig } from "../config/config.js"; -import { - loadSessionStore, - resolveAgentMainSessionKey, - resolveStorePath, - updateSessionStore, -} from "../config/sessions.js"; +import { resolveAgentMainSessionKey } from "../config/sessions.js"; import { runCronIsolatedAgentTurn } from "../cron/isolated-agent.js"; import { appendCronRunLog, resolveCronRunLogPath } from "../cron/run-log.js"; import { CronService } from "../cron/service.js"; import { resolveCronStorePath } from "../cron/store.js"; -import type { CronDeliveryMode, CronOrigin } from "../cron/types.js"; import { runHeartbeatOnce } from "../infra/heartbeat-runner.js"; import { requestHeartbeatNow } from "../infra/heartbeat-wake.js"; import { enqueueSystemEvent } from "../infra/system-events.js"; import { getChildLogger } from "../logging.js"; import { normalizeAgentId } from "../routing/session-key.js"; import { defaultRuntime } from "../runtime.js"; -import { normalizeSessionDeliveryFields } from "../utils/delivery-context.js"; export type GatewayCronState = { cron: CronService; @@ -59,47 +52,10 @@ export function buildGatewayCronService(params: { cfg: runtimeConfig, agentId, }); - - // If origin is provided and deliveryMode is "origin" (default), update the - // session's delivery context so the heartbeat routes replies to the origin. - const origin = opts?.origin; + // Pass origin through to the system event - heartbeat runner will use it for routing const deliveryMode = opts?.deliveryMode ?? "origin"; - if (origin && deliveryMode === "origin" && (origin.channel || origin.to)) { - const storePath = resolveStorePath(runtimeConfig.session?.store, { agentId }); - const store = loadSessionStore(storePath); - const entry = store[sessionKey]; - if (entry) { - const deliveryFields = normalizeSessionDeliveryFields({ - deliveryContext: { - channel: origin.channel, - to: origin.to, - accountId: origin.accountId, - threadId: origin.threadId, - }, - }); - // Update session entry with origin delivery context - const updatedEntry = { - ...entry, - updatedAt: Date.now(), - deliveryContext: deliveryFields.deliveryContext, - lastChannel: deliveryFields.lastChannel ?? entry.lastChannel, - lastTo: deliveryFields.lastTo ?? entry.lastTo, - lastAccountId: deliveryFields.lastAccountId ?? entry.lastAccountId, - lastThreadId: deliveryFields.lastThreadId ?? entry.lastThreadId, - }; - store[sessionKey] = updatedEntry; - void updateSessionStore(storePath, (s) => { - s[sessionKey] = updatedEntry; - }).catch((err) => { - cronLogger.warn( - { err: String(err), sessionKey }, - "cron: failed to update session delivery context for origin routing", - ); - }); - } - } - - enqueueSystemEvent(text, { sessionKey }); + const origin = deliveryMode === "origin" ? opts?.origin : undefined; + enqueueSystemEvent(text, { sessionKey, origin }); }, requestHeartbeatNow, runHeartbeatOnce: async (opts) => { diff --git a/src/infra/heartbeat-runner.ts b/src/infra/heartbeat-runner.ts index 8e0c9a4ee..cb8dbe03f 100644 --- a/src/infra/heartbeat-runner.ts +++ b/src/infra/heartbeat-runner.ts @@ -34,8 +34,9 @@ import { updateSessionStore, } from "../config/sessions.js"; import type { AgentDefaultsConfig } from "../config/types.agent-defaults.js"; +import type { CronOrigin } from "../cron/types.js"; import { formatErrorMessage } from "../infra/errors.js"; -import { peekSystemEvents } from "../infra/system-events.js"; +import { peekSystemEventEntries, peekSystemEvents } from "../infra/system-events.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { getQueueSize } from "../process/command-queue.js"; import { CommandLane } from "../process/lanes.js"; @@ -51,6 +52,7 @@ import { } from "./heartbeat-wake.js"; import type { OutboundSendDeps } from "./outbound/deliver.js"; import { deliverOutboundPayloads } from "./outbound/deliver.js"; +import type { OutboundTarget } from "./outbound/targets.js"; import { resolveHeartbeatDeliveryTarget, resolveHeartbeatSenderContext, @@ -97,6 +99,36 @@ const EXEC_EVENT_PROMPT = "Please relay the command output to the user in a helpful way. If the command succeeded, share the relevant output. " + "If it failed, explain what went wrong."; +/** + * Apply cron job origin context to override heartbeat delivery target. + * This routes replies back to where the cron job was created. + */ +function applyOriginToDelivery( + delivery: OutboundTarget, + origin: CronOrigin, + _cfg: MoltbotConfig, +): OutboundTarget { + // Only override if origin has usable channel/to + if (!origin.channel && !origin.to) { + return delivery; + } + + const channel = origin.channel ?? delivery.channel; + const to = origin.to ?? delivery.to; + + // If we still don't have a valid channel, keep the original delivery + if (channel === "none" || !channel) { + return delivery; + } + + return { + ...delivery, + channel, + to, + accountId: origin.accountId ?? delivery.accountId, + }; +} + function resolveActiveHoursTimezone(cfg: MoltbotConfig, raw?: string): string { const trimmed = raw?.trim(); if (!trimmed || trimmed === "user") { @@ -480,7 +512,16 @@ export async function runHeartbeatOnce(opts: { const { entry, sessionKey, storePath } = resolveHeartbeatSession(cfg, agentId, heartbeat); const previousUpdatedAt = entry?.updatedAt; - const delivery = resolveHeartbeatDeliveryTarget({ cfg, entry, heartbeat }); + let delivery = resolveHeartbeatDeliveryTarget({ cfg, entry, heartbeat }); + + // Check if any pending system events have origin context (from cron jobs). + // If so, override delivery target to route replies back to the origin. + const pendingEventEntries = peekSystemEventEntries(sessionKey); + const originFromEvent = pendingEventEntries.find((e) => e.origin)?.origin; + if (originFromEvent && (originFromEvent.channel || originFromEvent.to)) { + delivery = applyOriginToDelivery(delivery, originFromEvent, cfg); + } + const visibility = delivery.channel !== "none" ? resolveHeartbeatVisibility({ diff --git a/src/infra/system-events.ts b/src/infra/system-events.ts index b2842142d..43bd3d132 100644 --- a/src/infra/system-events.ts +++ b/src/infra/system-events.ts @@ -2,7 +2,14 @@ // prefixed to the next prompt. We intentionally avoid persistence to keep // events ephemeral. Events are session-scoped and require an explicit key. -export type SystemEvent = { text: string; ts: number }; +import type { CronOrigin } from "../cron/types.js"; + +export type SystemEvent = { + text: string; + ts: number; + /** Origin context for routing replies (from cron jobs) */ + origin?: CronOrigin; +}; const MAX_EVENTS = 20; @@ -17,6 +24,8 @@ const queues = new Map(); type SystemEventOptions = { sessionKey: string; contextKey?: string | null; + /** Origin context for routing replies (from cron jobs) */ + origin?: CronOrigin; }; function requireSessionKey(key?: string | null): string { @@ -62,7 +71,11 @@ export function enqueueSystemEvent(text: string, options: SystemEventOptions) { entry.lastContextKey = normalizeContextKey(options?.contextKey); if (entry.lastText === cleaned) return; // skip consecutive duplicates entry.lastText = cleaned; - entry.queue.push({ text: cleaned, ts: Date.now() }); + const event: SystemEvent = { text: cleaned, ts: Date.now() }; + if (options?.origin) { + event.origin = options.origin; + } + entry.queue.push(event); if (entry.queue.length > MAX_EVENTS) entry.queue.shift(); } @@ -87,6 +100,11 @@ export function peekSystemEvents(sessionKey: string): string[] { return queues.get(key)?.queue.map((e) => e.text) ?? []; } +export function peekSystemEventEntries(sessionKey: string): SystemEvent[] { + const key = requireSessionKey(sessionKey); + return queues.get(key)?.queue.slice() ?? []; +} + export function hasSystemEvents(sessionKey: string) { const key = requireSessionKey(sessionKey); return (queues.get(key)?.queue.length ?? 0) > 0;