diff --git a/src/agents/moltbot-tools.ts b/src/agents/moltbot-tools.ts index 5e0d74741..700a5b7e0 100644 --- a/src/agents/moltbot-tools.ts +++ b/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, diff --git a/src/agents/tools/cron-tool.ts b/src/agents/tools/cron-tool.ts index 739b3ada3..a37db1f1a 100644 --- a/src/agents/tools/cron-tool.ts +++ b/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 = {}; + 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 }).origin = origin; + } + } const contextMessages = typeof params.contextMessages === "number" && Number.isFinite(params.contextMessages) ? params.contextMessages diff --git a/src/cron/isolated-agent/delivery-target.ts b/src/cron/isolated-agent/delivery-target.ts index 4b1ff01f9..f5b70ef68 100644 --- a/src/cron/isolated-agent/delivery-target.ts +++ b/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; 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, + 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 | 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; + } 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, }; diff --git a/src/cron/isolated-agent/run.ts b/src/cron/isolated-agent/run.ts index b299fe9a2..fd7808bca 100644 --- a/src/cron/isolated-agent/run.ts +++ b/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); diff --git a/src/cron/service/jobs.ts b/src/cron/service/jobs.ts index 132156a0c..10f16947d 100644 --- a/src/cron/service/jobs.ts +++ b/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); } diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index ab094c20b..23ef13482 100644 --- a/src/cron/service/state.ts +++ b/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; runIsolatedAgentJob: (params: { job: CronJob; message: string }) => Promise<{ diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index 370f5d116..f2335ce41 100644 --- a/src/cron/service/timer.ts +++ b/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((resolve) => setTimeout(resolve, ms)); diff --git a/src/cron/types.ts b/src/cron/types.ts index f3fd891d6..fa8238503 100644 --- a/src/cron/types.ts +++ b/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 = { diff --git a/src/gateway/protocol/schema/cron.ts b/src/gateway/protocol/schema/cron.ts index 63ed0c209..9c8170631 100644 --- a/src/gateway/protocol/schema/cron.ts +++ b/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 }, ); diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index 9a0c0ca98..3c10b811a 100644 --- a/src/gateway/server-cron.ts +++ b/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,