feat(cron): introduce origin context and delivery mode for cron jobs (#6)

- Added `originContext` to `createCronTool` for routing replies back to the originating session/channel.
- Updated `CronJob` and related types to include `origin` and `deliveryMode` properties.
- Enhanced `resolveDeliveryTarget` to utilize origin context for determining delivery targets.
- Modified various components to support the new origin context and delivery mode functionality, ensuring replies are routed correctly based on the job's origin.

src/agents/moltbot-tools.ts

@@ -88,6 +88,12 @@ export function createMoltbotTools(options?: {
     }),
     createCronTool({
       agentSessionKey: options?.agentSessionKey,
+      originContext: {
+        channel: options?.agentChannel,
+        to: options?.agentTo,
+        accountId: options?.agentAccountId,
+        threadId: options?.agentThreadId,
+      },
     }),
     createMessageTool({
       agentAccountId: options?.agentAccountId,

src/agents/tools/cron-tool.ts

@@ -42,6 +42,13 @@ const CronToolSchema = Type.Object({
 
 type CronToolOptions = {
   agentSessionKey?: string;
+  /** Origin context for routing cron job replies back to where the job was created */
+  originContext?: {
+    channel?: string;
+    to?: string;
+    accountId?: string;
+    threadId?: string | number;
+  };
 };
 
 type ChatMessage = {
@@ -210,6 +217,18 @@ Use jobId as the canonical identifier; id is accepted for compatibility. Use con
               (job as { agentId?: string }).agentId = agentId;
             }
           }
+          // Inject origin context for reply routing (unless already specified)
+          if (job && typeof job === "object" && !("origin" in job) && opts?.originContext) {
+            const origin: Record<string, unknown> = {};
+            if (opts.originContext.channel) origin.channel = opts.originContext.channel;
+            if (opts.originContext.to) origin.to = opts.originContext.to;
+            if (opts.originContext.accountId) origin.accountId = opts.originContext.accountId;
+            if (opts.originContext.threadId != null) origin.threadId = opts.originContext.threadId;
+            if (opts.agentSessionKey) origin.sessionKey = opts.agentSessionKey;
+            if (Object.keys(origin).length > 0) {
+              (job as { origin?: Record<string, unknown> }).origin = origin;
+            }
+          }
           const contextMessages =
             typeof params.contextMessages === "number" && Number.isFinite(params.contextMessages)
               ? params.contextMessages

src/cron/isolated-agent/delivery-target.ts

@@ -12,6 +12,7 @@ import {
   resolveOutboundTarget,
   resolveSessionDeliveryTarget,
 } from "../../infra/outbound/targets.js";
+import type { CronDeliveryMode, CronOrigin } from "../types.js";
 
 export async function resolveDeliveryTarget(
   cfg: MoltbotConfig,
@@ -20,15 +21,67 @@ export async function resolveDeliveryTarget(
     channel?: "last" | ChannelId;
     to?: string;
   },
+  options?: {
+    /** Origin context from when the job was created */
+    origin?: CronOrigin;
+    /** Delivery mode: "origin" (default) or "current" */
+    deliveryMode?: CronDeliveryMode;
+  },
 ): Promise<{
   channel: Exclude<OutboundChannel, "none">;
   to?: string;
   accountId?: string;
+  threadId?: string | number;
   mode: "explicit" | "implicit";
   error?: Error;
 }> {
-  const requestedChannel = typeof jobPayload.channel === "string" ? jobPayload.channel : "last";
-  const explicitTo = typeof jobPayload.to === "string" ? jobPayload.to : undefined;
+  const origin = options?.origin;
+  const deliveryMode = options?.deliveryMode ?? "origin";
+
+  // Explicit payload values always take precedence
+  const hasExplicitChannel =
+    typeof jobPayload.channel === "string" && jobPayload.channel !== "last";
+  const hasExplicitTo = typeof jobPayload.to === "string" && jobPayload.to.trim() !== "";
+
+  // Determine if we should use origin context
+  // Use origin when: deliveryMode is "origin" (default), origin exists, and no explicit overrides
+  const useOrigin =
+    deliveryMode === "origin" &&
+    origin &&
+    !hasExplicitChannel &&
+    !hasExplicitTo &&
+    (origin.channel || origin.to);
+
+  if (useOrigin && origin.channel && origin.to) {
+    // Route directly to origin - no need to look up main session
+    const docked = resolveOutboundTarget({
+      channel: origin.channel,
+      to: origin.to,
+      cfg,
+      accountId: origin.accountId,
+      mode: "explicit",
+    });
+    return {
+      channel: origin.channel as Exclude<OutboundChannel, "none">,
+      to: docked.ok ? docked.to : undefined,
+      accountId: origin.accountId,
+      threadId: origin.threadId,
+      mode: "explicit",
+      error: docked.ok ? undefined : docked.error,
+    };
+  }
+
+  // Fall back to main session lookup (current behavior)
+  const requestedChannel = hasExplicitChannel
+    ? (jobPayload.channel as ChannelId)
+    : useOrigin && origin?.channel
+      ? origin.channel
+      : "last";
+  const explicitTo = hasExplicitTo
+    ? jobPayload.to
+    : useOrigin && origin?.to
+      ? origin.to
+      : undefined;
 
   const sessionCfg = cfg.session;
   const mainSessionKey = resolveAgentMainSessionKey({ cfg, agentId });
@@ -45,11 +98,16 @@ export async function resolveDeliveryTarget(
 
   let fallbackChannel: Exclude<OutboundChannel, "none"> | undefined;
   if (!preliminary.channel) {
-    try {
-      const selection = await resolveMessageChannelSelection({ cfg });
-      fallbackChannel = selection.channel;
-    } catch {
-      fallbackChannel = preliminary.lastChannel ?? DEFAULT_CHAT_CHANNEL;
+    // Try origin channel as fallback before config/default
+    if (useOrigin && origin?.channel) {
+      fallbackChannel = origin.channel as Exclude<OutboundChannel, "none">;
+    } else {
+      try {
+        const selection = await resolveMessageChannelSelection({ cfg });
+        fallbackChannel = selection.channel;
+      } catch {
+        fallbackChannel = preliminary.lastChannel ?? DEFAULT_CHAT_CHANNEL;
+      }
     }
   }
 
@@ -67,22 +125,25 @@ export async function resolveDeliveryTarget(
   const channel = resolved.channel ?? fallbackChannel ?? DEFAULT_CHAT_CHANNEL;
   const mode = resolved.mode as "explicit" | "implicit";
   const toCandidate = resolved.to;
+  const accountId = useOrigin && origin?.accountId ? origin.accountId : resolved.accountId;
+  const threadId = useOrigin && origin?.threadId != null ? origin.threadId : resolved.threadId;
 
   if (!toCandidate) {
-    return { channel, to: undefined, accountId: resolved.accountId, mode };
+    return { channel, to: undefined, accountId, threadId, mode };
   }
 
   const docked = resolveOutboundTarget({
     channel,
     to: toCandidate,
     cfg,
-    accountId: resolved.accountId,
+    accountId,
     mode,
   });
   return {
     channel,
     to: docked.ok ? docked.to : undefined,
-    accountId: resolved.accountId,
+    accountId,
+    threadId,
     mode,
     error: docked.ok ? undefined : docked.error,
   };

src/cron/isolated-agent/run.ts

@@ -234,10 +234,18 @@ export async function runCronIsolatedAgentTurn(params: {
     deliveryMode === "explicit" || (deliveryMode === "auto" && hasExplicitTarget);
   const bestEffortDeliver = agentPayload?.bestEffortDeliver === true;
 
-  const resolvedDelivery = await resolveDeliveryTarget(cfgWithAgentDefaults, agentId, {
-    channel: agentPayload?.channel ?? "last",
-    to: agentPayload?.to,
-  });
+  const resolvedDelivery = await resolveDeliveryTarget(
+    cfgWithAgentDefaults,
+    agentId,
+    {
+      channel: agentPayload?.channel ?? "last",
+      to: agentPayload?.to,
+    },
+    {
+      origin: params.job.origin,
+      deliveryMode: params.job.deliveryMode,
+    },
+  );
 
   const userTimezone = resolveUserTimezone(params.cfg.agents?.defaults?.userTimezone);
   const userTimeFormat = resolveUserTimeFormat(params.cfg.agents?.defaults?.timeFormat);

src/cron/service/jobs.ts

@@ -95,6 +95,8 @@ export function createJob(state: CronServiceState, input: CronJobCreate): CronJo
     state: {
       ...input.state,
     },
+    origin: input.origin,
+    deliveryMode: input.deliveryMode,
   };
   assertSupportedJobSpec(job);
   job.state.nextRunAtMs = computeJobNextRunAtMs(job, now);
@@ -115,6 +117,8 @@ export function applyJobPatch(job: CronJob, patch: CronJobPatch) {
   if ("agentId" in patch) {
     job.agentId = normalizeOptionalAgentId((patch as { agentId?: unknown }).agentId);
   }
+  if (patch.origin) job.origin = patch.origin;
+  if (patch.deliveryMode) job.deliveryMode = patch.deliveryMode;
   assertSupportedJobSpec(job);
 }
 

src/cron/service/state.ts

@@ -1,5 +1,12 @@
 import type { HeartbeatRunResult } from "../../infra/heartbeat-wake.js";
-import type { CronJob, CronJobCreate, CronJobPatch, CronStoreFile } from "../types.js";
+import type {
+  CronDeliveryMode,
+  CronJob,
+  CronJobCreate,
+  CronJobPatch,
+  CronOrigin,
+  CronStoreFile,
+} from "../types.js";
 
 export type CronEvent = {
   jobId: string;
@@ -24,7 +31,16 @@ export type CronServiceDeps = {
   log: Logger;
   storePath: string;
   cronEnabled: boolean;
-  enqueueSystemEvent: (text: string, opts?: { agentId?: string }) => void;
+  enqueueSystemEvent: (
+    text: string,
+    opts?: {
+      agentId?: string;
+      /** Origin context for routing replies back to where the job was created */
+      origin?: CronOrigin;
+      /** Delivery mode: "origin" (default) or "current" */
+      deliveryMode?: CronDeliveryMode;
+    },
+  ) => void;
   requestHeartbeatNow: (opts?: { reason?: string }) => void;
   runHeartbeatOnce?: (opts?: { reason?: string }) => Promise<HeartbeatRunResult>;
   runIsolatedAgentJob: (params: { job: CronJob; message: string }) => Promise<{

src/cron/service/timer.ts

@@ -149,7 +149,11 @@ export async function executeJob(
         );
         return;
       }
-      state.deps.enqueueSystemEvent(text, { agentId: job.agentId });
+      state.deps.enqueueSystemEvent(text, {
+        agentId: job.agentId,
+        origin: job.origin,
+        deliveryMode: job.deliveryMode,
+      });
       if (job.wakeMode === "now" && state.deps.runHeartbeatOnce) {
         const reason = `cron:${job.id}`;
         const delay = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));

src/cron/types.ts

@@ -53,6 +53,26 @@ export type CronIsolation = {
   postToMainMaxChars?: number;
 };
 
+/**
+ * Origin context: where the cron job was created.
+ * Used for routing replies back to the originating session/channel.
+ */
+export type CronOrigin = {
+  /** The channel where the job was created (telegram, whatsapp, etc.) */
+  channel?: ChannelId;
+  /** The chat/conversation ID where the job was created */
+  to?: string;
+  /** The account ID (for multi-account setups) */
+  accountId?: string;
+  /** The thread ID (for threaded conversations) */
+  threadId?: string | number;
+  /** The session key where the job was created */
+  sessionKey?: string;
+};
+
+/** Delivery routing mode for cron job replies */
+export type CronDeliveryMode = "origin" | "current";
+
 export type CronJobState = {
   nextRunAtMs?: number;
   runningAtMs?: number;
@@ -77,6 +97,17 @@ export type CronJob = {
   payload: CronPayload;
   isolation?: CronIsolation;
   state: CronJobState;
+  /**
+   * Origin context: where the job was created.
+   * Replies are routed back here by default.
+   */
+  origin?: CronOrigin;
+  /**
+   * Delivery routing mode:
+   * - "origin" (default): Route replies to where the job was created
+   * - "current": Route replies to the session's current lastChannel/lastTo
+   */
+  deliveryMode?: CronDeliveryMode;
 };
 
 export type CronStoreFile = {

src/gateway/protocol/schema/cron.ts

@@ -85,6 +85,27 @@ export const CronIsolationSchema = Type.Object(
   { additionalProperties: false },
 );
 
+/**
+ * Origin context: where the cron job was created.
+ * Used for routing replies back to the originating session/channel.
+ */
+export const CronOriginSchema = Type.Object(
+  {
+    channel: Type.Optional(NonEmptyString),
+    to: Type.Optional(Type.String()),
+    accountId: Type.Optional(Type.String()),
+    threadId: Type.Optional(Type.Union([Type.String(), Type.Integer()])),
+    sessionKey: Type.Optional(Type.String()),
+  },
+  { additionalProperties: false },
+);
+
+/** Delivery routing mode for cron job replies */
+export const CronDeliveryModeSchema = Type.Union([
+  Type.Literal("origin"),
+  Type.Literal("current"),
+]);
+
 export const CronJobStateSchema = Type.Object(
   {
     nextRunAtMs: Type.Optional(Type.Integer({ minimum: 0 })),
@@ -115,6 +136,8 @@ export const CronJobSchema = Type.Object(
     payload: CronPayloadSchema,
     isolation: Type.Optional(CronIsolationSchema),
     state: CronJobStateSchema,
+    origin: Type.Optional(CronOriginSchema),
+    deliveryMode: Type.Optional(CronDeliveryModeSchema),
   },
   { additionalProperties: false },
 );
@@ -140,6 +163,8 @@ export const CronAddParamsSchema = Type.Object(
     wakeMode: Type.Union([Type.Literal("next-heartbeat"), Type.Literal("now")]),
     payload: CronPayloadSchema,
     isolation: Type.Optional(CronIsolationSchema),
+    origin: Type.Optional(CronOriginSchema),
+    deliveryMode: Type.Optional(CronDeliveryModeSchema),
   },
   { additionalProperties: false },
 );
@@ -157,6 +182,8 @@ export const CronJobPatchSchema = Type.Object(
     payload: Type.Optional(CronPayloadPatchSchema),
     isolation: Type.Optional(CronIsolationSchema),
     state: Type.Optional(Type.Partial(CronJobStateSchema)),
+    origin: Type.Optional(CronOriginSchema),
+    deliveryMode: Type.Optional(CronDeliveryModeSchema),
   },
   { additionalProperties: false },
 );

src/gateway/server-cron.ts

@@ -1,17 +1,24 @@
 import { resolveDefaultAgentId } from "../agents/agent-scope.js";
 import type { CliDeps } from "../cli/deps.js";
 import { loadConfig } from "../config/config.js";
-import { resolveAgentMainSessionKey } from "../config/sessions.js";
+import {
+  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;
@@ -52,6 +59,46 @@ 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;
+      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 });
     },
     requestHeartbeatNow,