refactor(cron): streamline origin handling in system events and heartbeat delivery (#7)

- Removed redundant session delivery context updates in `buildGatewayCronService`.
- Enhanced `enqueueSystemEvent` to include optional origin context for routing replies from cron jobs.
- Updated `runHeartbeatOnce` to apply origin context from pending system events, ensuring replies are directed back to the correct source.
- Introduced `peekSystemEventEntries` for accessing system event entries with origin context.

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 {
+import { resolveAgentMainSessionKey } from "../config/sessions.js";
-  loadSessionStore,
-  resolveAgentMainSessionKey,
-  resolveStorePath,
-  updateSessionStore,
-} 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,
       });
-
+      // Pass origin through to the system event - heartbeat runner will use it for routing
-      // 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;
       const deliveryMode = opts?.deliveryMode ?? "origin";
-      if (origin && deliveryMode === "origin" && (origin.channel || origin.to)) {
+      const origin = deliveryMode === "origin" ? opts?.origin : undefined;
-        const storePath = resolveStorePath(runtimeConfig.session?.store, { agentId });
+      enqueueSystemEvent(text, { sessionKey, origin });
-        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 });
     },
     requestHeartbeatNow,
     runHeartbeatOnce: async (opts) => {

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({

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<string, SessionQueue>();
 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;