mirror of
https://github.com/moltbot/moltbot.git
synced 2026-03-21 16:41:56 +00:00
a54dc7fe80
Verified: - pnpm build - pnpm check - pnpm test:macmini Co-authored-by: Takhoffman <781889+Takhoffman@users.noreply.github.com> Co-authored-by: Tak Hoffman <781889+Takhoffman@users.noreply.github.com>
843 lines
32 KiB
TypeScript
843 lines
32 KiB
TypeScript
import {
|
|
resolveAgentConfig,
|
|
resolveAgentDir,
|
|
resolveAgentModelFallbacksOverride,
|
|
resolveAgentWorkspaceDir,
|
|
resolveDefaultAgentId,
|
|
} from "../../agents/agent-scope.js";
|
|
import { resolveSessionAuthProfileOverride } from "../../agents/auth-profiles/session-override.js";
|
|
import { runCliAgent } from "../../agents/cli-runner.js";
|
|
import { getCliSessionId, setCliSessionId } from "../../agents/cli-session.js";
|
|
import { lookupContextTokens } from "../../agents/context.js";
|
|
import { resolveCronStyleNow } from "../../agents/current-time.js";
|
|
import { DEFAULT_CONTEXT_TOKENS, DEFAULT_MODEL, DEFAULT_PROVIDER } from "../../agents/defaults.js";
|
|
import { loadModelCatalog } from "../../agents/model-catalog.js";
|
|
import { runWithModelFallback } from "../../agents/model-fallback.js";
|
|
import {
|
|
getModelRefStatus,
|
|
isCliProvider,
|
|
resolveAllowedModelRef,
|
|
resolveConfiguredModelRef,
|
|
resolveHooksGmailModel,
|
|
resolveThinkingDefault,
|
|
} from "../../agents/model-selection.js";
|
|
import type { MessagingToolSend } from "../../agents/pi-embedded-messaging.js";
|
|
import { runEmbeddedPiAgent } from "../../agents/pi-embedded.js";
|
|
import { runSubagentAnnounceFlow } from "../../agents/subagent-announce.js";
|
|
import { countActiveDescendantRuns } from "../../agents/subagent-registry.js";
|
|
import { resolveAgentTimeoutMs } from "../../agents/timeout.js";
|
|
import { deriveSessionTotalTokens, hasNonzeroUsage } from "../../agents/usage.js";
|
|
import { ensureAgentWorkspace } from "../../agents/workspace.js";
|
|
import {
|
|
normalizeThinkLevel,
|
|
normalizeVerboseLevel,
|
|
supportsXHighThinking,
|
|
} from "../../auto-reply/thinking.js";
|
|
import { SILENT_REPLY_TOKEN } from "../../auto-reply/tokens.js";
|
|
import { createOutboundSendDeps, type CliDeps } from "../../cli/outbound-send-deps.js";
|
|
import type { OpenClawConfig } from "../../config/config.js";
|
|
import {
|
|
resolveAgentMainSessionKey,
|
|
resolveSessionTranscriptPath,
|
|
updateSessionStore,
|
|
} from "../../config/sessions.js";
|
|
import type { AgentDefaultsConfig } from "../../config/types.js";
|
|
import { registerAgentRunContext } from "../../infra/agent-events.js";
|
|
import { deliverOutboundPayloads } from "../../infra/outbound/deliver.js";
|
|
import { resolveAgentOutboundIdentity } from "../../infra/outbound/identity.js";
|
|
import { resolveOutboundSessionRoute } from "../../infra/outbound/outbound-session.js";
|
|
import { logWarn } from "../../logger.js";
|
|
import { buildAgentMainSessionKey, normalizeAgentId } from "../../routing/session-key.js";
|
|
import {
|
|
buildSafeExternalPrompt,
|
|
detectSuspiciousPatterns,
|
|
getHookType,
|
|
isExternalHookSession,
|
|
} from "../../security/external-content.js";
|
|
import { resolveCronDeliveryPlan } from "../delivery.js";
|
|
import type { CronJob, CronRunOutcome, CronRunTelemetry } from "../types.js";
|
|
import { resolveDeliveryTarget } from "./delivery-target.js";
|
|
import {
|
|
isHeartbeatOnlyResponse,
|
|
pickLastDeliverablePayload,
|
|
pickLastNonEmptyTextFromPayloads,
|
|
pickSummaryFromOutput,
|
|
pickSummaryFromPayloads,
|
|
resolveHeartbeatAckMaxChars,
|
|
} from "./helpers.js";
|
|
import { resolveCronSession } from "./session.js";
|
|
import { resolveCronSkillsSnapshot } from "./skills-snapshot.js";
|
|
import {
|
|
expectsSubagentFollowup,
|
|
isLikelyInterimCronMessage,
|
|
readDescendantSubagentFallbackReply,
|
|
waitForDescendantSubagentSummary,
|
|
} from "./subagent-followup.js";
|
|
|
|
function matchesMessagingToolDeliveryTarget(
|
|
target: MessagingToolSend,
|
|
delivery: { channel?: string; to?: string; accountId?: string },
|
|
): boolean {
|
|
if (!delivery.channel || !delivery.to || !target.to) {
|
|
return false;
|
|
}
|
|
const channel = delivery.channel.trim().toLowerCase();
|
|
const provider = target.provider?.trim().toLowerCase();
|
|
if (provider && provider !== "message" && provider !== channel) {
|
|
return false;
|
|
}
|
|
if (target.accountId && delivery.accountId && target.accountId !== delivery.accountId) {
|
|
return false;
|
|
}
|
|
return target.to === delivery.to;
|
|
}
|
|
|
|
function resolveCronDeliveryBestEffort(job: CronJob): boolean {
|
|
if (typeof job.delivery?.bestEffort === "boolean") {
|
|
return job.delivery.bestEffort;
|
|
}
|
|
if (job.payload.kind === "agentTurn" && typeof job.payload.bestEffortDeliver === "boolean") {
|
|
return job.payload.bestEffortDeliver;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
async function resolveCronAnnounceSessionKey(params: {
|
|
cfg: OpenClawConfig;
|
|
agentId: string;
|
|
fallbackSessionKey: string;
|
|
delivery: {
|
|
channel: Parameters<typeof resolveOutboundSessionRoute>[0]["channel"];
|
|
to?: string;
|
|
accountId?: string;
|
|
threadId?: string | number;
|
|
};
|
|
}): Promise<string> {
|
|
const to = params.delivery.to?.trim();
|
|
if (!to) {
|
|
return params.fallbackSessionKey;
|
|
}
|
|
try {
|
|
const route = await resolveOutboundSessionRoute({
|
|
cfg: params.cfg,
|
|
channel: params.delivery.channel,
|
|
agentId: params.agentId,
|
|
accountId: params.delivery.accountId,
|
|
target: to,
|
|
threadId: params.delivery.threadId,
|
|
});
|
|
const resolved = route?.sessionKey?.trim();
|
|
if (resolved) {
|
|
return resolved;
|
|
}
|
|
} catch {
|
|
// Fall back to main session routing if announce session resolution fails.
|
|
}
|
|
return params.fallbackSessionKey;
|
|
}
|
|
|
|
export type RunCronAgentTurnResult = {
|
|
/** Last non-empty agent text output (not truncated). */
|
|
outputText?: string;
|
|
/**
|
|
* `true` when the isolated run already delivered its output to the target
|
|
* channel (via outbound payloads, the subagent announce flow, or a matching
|
|
* messaging-tool send). Callers should skip posting a summary to the main
|
|
* session to avoid duplicate
|
|
* messages. See: https://github.com/openclaw/openclaw/issues/15692
|
|
*/
|
|
delivered?: boolean;
|
|
} & CronRunOutcome &
|
|
CronRunTelemetry;
|
|
|
|
export async function runCronIsolatedAgentTurn(params: {
|
|
cfg: OpenClawConfig;
|
|
deps: CliDeps;
|
|
job: CronJob;
|
|
message: string;
|
|
abortSignal?: AbortSignal;
|
|
signal?: AbortSignal;
|
|
sessionKey: string;
|
|
agentId?: string;
|
|
lane?: string;
|
|
}): Promise<RunCronAgentTurnResult> {
|
|
const abortSignal = params.abortSignal ?? params.signal;
|
|
const isAborted = () => abortSignal?.aborted === true;
|
|
const abortReason = () => {
|
|
const reason = abortSignal?.reason;
|
|
return typeof reason === "string" && reason.trim()
|
|
? reason.trim()
|
|
: "cron: job execution timed out";
|
|
};
|
|
const isFastTestEnv = process.env.OPENCLAW_TEST_FAST === "1";
|
|
const defaultAgentId = resolveDefaultAgentId(params.cfg);
|
|
const requestedAgentId =
|
|
typeof params.agentId === "string" && params.agentId.trim()
|
|
? params.agentId
|
|
: typeof params.job.agentId === "string" && params.job.agentId.trim()
|
|
? params.job.agentId
|
|
: undefined;
|
|
const normalizedRequested = requestedAgentId ? normalizeAgentId(requestedAgentId) : undefined;
|
|
const agentConfigOverride = normalizedRequested
|
|
? resolveAgentConfig(params.cfg, normalizedRequested)
|
|
: undefined;
|
|
const { model: overrideModel, ...agentOverrideRest } = agentConfigOverride ?? {};
|
|
// Use the requested agentId even when there is no explicit agent config entry.
|
|
// This ensures auth-profiles, workspace, and agentDir all resolve to the
|
|
// correct per-agent paths (e.g. ~/.openclaw/agents/<agentId>/agent/).
|
|
const agentId = normalizedRequested ?? defaultAgentId;
|
|
const agentCfg: AgentDefaultsConfig = Object.assign(
|
|
{},
|
|
params.cfg.agents?.defaults,
|
|
agentOverrideRest as Partial<AgentDefaultsConfig>,
|
|
);
|
|
// Merge agent model override with defaults instead of replacing, so that
|
|
// `fallbacks` from `agents.defaults.model` are preserved when the agent
|
|
// (or its per-cron model pin) only specifies `primary`.
|
|
const existingModel = agentCfg.model && typeof agentCfg.model === "object" ? agentCfg.model : {};
|
|
if (typeof overrideModel === "string") {
|
|
agentCfg.model = { ...existingModel, primary: overrideModel };
|
|
} else if (overrideModel) {
|
|
agentCfg.model = { ...existingModel, ...overrideModel };
|
|
}
|
|
const cfgWithAgentDefaults: OpenClawConfig = {
|
|
...params.cfg,
|
|
agents: Object.assign({}, params.cfg.agents, { defaults: agentCfg }),
|
|
};
|
|
|
|
const baseSessionKey = (params.sessionKey?.trim() || `cron:${params.job.id}`).trim();
|
|
const agentSessionKey = buildAgentMainSessionKey({
|
|
agentId,
|
|
mainKey: baseSessionKey,
|
|
});
|
|
|
|
const workspaceDirRaw = resolveAgentWorkspaceDir(params.cfg, agentId);
|
|
const agentDir = resolveAgentDir(params.cfg, agentId);
|
|
const workspace = await ensureAgentWorkspace({
|
|
dir: workspaceDirRaw,
|
|
ensureBootstrapFiles: !agentCfg?.skipBootstrap && !isFastTestEnv,
|
|
});
|
|
const workspaceDir = workspace.dir;
|
|
|
|
const resolvedDefault = resolveConfiguredModelRef({
|
|
cfg: cfgWithAgentDefaults,
|
|
defaultProvider: DEFAULT_PROVIDER,
|
|
defaultModel: DEFAULT_MODEL,
|
|
});
|
|
let provider = resolvedDefault.provider;
|
|
let model = resolvedDefault.model;
|
|
let catalog: Awaited<ReturnType<typeof loadModelCatalog>> | undefined;
|
|
const loadCatalog = async () => {
|
|
if (!catalog) {
|
|
catalog = await loadModelCatalog({ config: cfgWithAgentDefaults });
|
|
}
|
|
return catalog;
|
|
};
|
|
// Resolve model - prefer hooks.gmail.model for Gmail hooks.
|
|
const isGmailHook = baseSessionKey.startsWith("hook:gmail:");
|
|
let hooksGmailModelApplied = false;
|
|
const hooksGmailModelRef = isGmailHook
|
|
? resolveHooksGmailModel({
|
|
cfg: params.cfg,
|
|
defaultProvider: DEFAULT_PROVIDER,
|
|
})
|
|
: null;
|
|
if (hooksGmailModelRef) {
|
|
const status = getModelRefStatus({
|
|
cfg: params.cfg,
|
|
catalog: await loadCatalog(),
|
|
ref: hooksGmailModelRef,
|
|
defaultProvider: resolvedDefault.provider,
|
|
defaultModel: resolvedDefault.model,
|
|
});
|
|
if (status.allowed) {
|
|
provider = hooksGmailModelRef.provider;
|
|
model = hooksGmailModelRef.model;
|
|
hooksGmailModelApplied = true;
|
|
}
|
|
}
|
|
const modelOverrideRaw =
|
|
params.job.payload.kind === "agentTurn" ? params.job.payload.model : undefined;
|
|
const modelOverride = typeof modelOverrideRaw === "string" ? modelOverrideRaw.trim() : undefined;
|
|
if (modelOverride !== undefined && modelOverride.length > 0) {
|
|
const resolvedOverride = resolveAllowedModelRef({
|
|
cfg: cfgWithAgentDefaults,
|
|
catalog: await loadCatalog(),
|
|
raw: modelOverride,
|
|
defaultProvider: resolvedDefault.provider,
|
|
defaultModel: resolvedDefault.model,
|
|
});
|
|
if ("error" in resolvedOverride) {
|
|
return { status: "error", error: resolvedOverride.error };
|
|
}
|
|
provider = resolvedOverride.ref.provider;
|
|
model = resolvedOverride.ref.model;
|
|
}
|
|
const now = Date.now();
|
|
const cronSession = resolveCronSession({
|
|
cfg: params.cfg,
|
|
sessionKey: agentSessionKey,
|
|
agentId,
|
|
nowMs: now,
|
|
// Isolated cron runs must not carry prior turn context across executions.
|
|
forceNew: params.job.sessionTarget === "isolated",
|
|
});
|
|
const runSessionId = cronSession.sessionEntry.sessionId;
|
|
const runSessionKey = baseSessionKey.startsWith("cron:")
|
|
? `${agentSessionKey}:run:${runSessionId}`
|
|
: agentSessionKey;
|
|
const persistSessionEntry = async () => {
|
|
if (isFastTestEnv) {
|
|
return;
|
|
}
|
|
cronSession.store[agentSessionKey] = cronSession.sessionEntry;
|
|
if (runSessionKey !== agentSessionKey) {
|
|
cronSession.store[runSessionKey] = cronSession.sessionEntry;
|
|
}
|
|
await updateSessionStore(cronSession.storePath, (store) => {
|
|
store[agentSessionKey] = cronSession.sessionEntry;
|
|
if (runSessionKey !== agentSessionKey) {
|
|
store[runSessionKey] = cronSession.sessionEntry;
|
|
}
|
|
});
|
|
};
|
|
const withRunSession = (
|
|
result: Omit<RunCronAgentTurnResult, "sessionId" | "sessionKey">,
|
|
): RunCronAgentTurnResult => ({
|
|
...result,
|
|
sessionId: runSessionId,
|
|
sessionKey: runSessionKey,
|
|
});
|
|
if (!cronSession.sessionEntry.label?.trim() && baseSessionKey.startsWith("cron:")) {
|
|
const labelSuffix =
|
|
typeof params.job.name === "string" && params.job.name.trim()
|
|
? params.job.name.trim()
|
|
: params.job.id;
|
|
cronSession.sessionEntry.label = `Cron: ${labelSuffix}`;
|
|
}
|
|
|
|
// Respect session model override — check session.modelOverride before falling
|
|
// back to the default config model. This ensures /model changes are honoured
|
|
// by cron and isolated agent runs.
|
|
if (!modelOverride && !hooksGmailModelApplied) {
|
|
const sessionModelOverride = cronSession.sessionEntry.modelOverride?.trim();
|
|
if (sessionModelOverride) {
|
|
const sessionProviderOverride =
|
|
cronSession.sessionEntry.providerOverride?.trim() || resolvedDefault.provider;
|
|
const resolvedSessionOverride = resolveAllowedModelRef({
|
|
cfg: cfgWithAgentDefaults,
|
|
catalog: await loadCatalog(),
|
|
raw: `${sessionProviderOverride}/${sessionModelOverride}`,
|
|
defaultProvider: resolvedDefault.provider,
|
|
defaultModel: resolvedDefault.model,
|
|
});
|
|
if (!("error" in resolvedSessionOverride)) {
|
|
provider = resolvedSessionOverride.ref.provider;
|
|
model = resolvedSessionOverride.ref.model;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Resolve thinking level - job thinking > hooks.gmail.thinking > agent default
|
|
const hooksGmailThinking = isGmailHook
|
|
? normalizeThinkLevel(params.cfg.hooks?.gmail?.thinking)
|
|
: undefined;
|
|
const thinkOverride = normalizeThinkLevel(agentCfg?.thinkingDefault);
|
|
const jobThink = normalizeThinkLevel(
|
|
(params.job.payload.kind === "agentTurn" ? params.job.payload.thinking : undefined) ??
|
|
undefined,
|
|
);
|
|
let thinkLevel = jobThink ?? hooksGmailThinking ?? thinkOverride;
|
|
if (!thinkLevel) {
|
|
thinkLevel = resolveThinkingDefault({
|
|
cfg: cfgWithAgentDefaults,
|
|
provider,
|
|
model,
|
|
catalog: await loadCatalog(),
|
|
});
|
|
}
|
|
if (thinkLevel === "xhigh" && !supportsXHighThinking(provider, model)) {
|
|
logWarn(
|
|
`[cron:${params.job.id}] Thinking level "xhigh" is not supported for ${provider}/${model}; downgrading to "high".`,
|
|
);
|
|
thinkLevel = "high";
|
|
}
|
|
|
|
const timeoutMs = resolveAgentTimeoutMs({
|
|
cfg: cfgWithAgentDefaults,
|
|
overrideSeconds:
|
|
params.job.payload.kind === "agentTurn" ? params.job.payload.timeoutSeconds : undefined,
|
|
});
|
|
|
|
const agentPayload = params.job.payload.kind === "agentTurn" ? params.job.payload : null;
|
|
const deliveryPlan = resolveCronDeliveryPlan(params.job);
|
|
const deliveryRequested = deliveryPlan.requested;
|
|
|
|
const resolvedDelivery = await resolveDeliveryTarget(cfgWithAgentDefaults, agentId, {
|
|
channel: deliveryPlan.channel ?? "last",
|
|
to: deliveryPlan.to,
|
|
sessionKey: params.job.sessionKey,
|
|
});
|
|
|
|
const { formattedTime, timeLine } = resolveCronStyleNow(params.cfg, now);
|
|
const base = `[cron:${params.job.id} ${params.job.name}] ${params.message}`.trim();
|
|
|
|
// SECURITY: Wrap external hook content with security boundaries to prevent prompt injection
|
|
// unless explicitly allowed via a dangerous config override.
|
|
const isExternalHook = isExternalHookSession(baseSessionKey);
|
|
const allowUnsafeExternalContent =
|
|
agentPayload?.allowUnsafeExternalContent === true ||
|
|
(isGmailHook && params.cfg.hooks?.gmail?.allowUnsafeExternalContent === true);
|
|
const shouldWrapExternal = isExternalHook && !allowUnsafeExternalContent;
|
|
let commandBody: string;
|
|
|
|
if (isExternalHook) {
|
|
// Log suspicious patterns for security monitoring
|
|
const suspiciousPatterns = detectSuspiciousPatterns(params.message);
|
|
if (suspiciousPatterns.length > 0) {
|
|
logWarn(
|
|
`[security] Suspicious patterns detected in external hook content ` +
|
|
`(session=${baseSessionKey}, patterns=${suspiciousPatterns.length}): ${suspiciousPatterns.slice(0, 3).join(", ")}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
if (shouldWrapExternal) {
|
|
// Wrap external content with security boundaries
|
|
const hookType = getHookType(baseSessionKey);
|
|
const safeContent = buildSafeExternalPrompt({
|
|
content: params.message,
|
|
source: hookType,
|
|
jobName: params.job.name,
|
|
jobId: params.job.id,
|
|
timestamp: formattedTime,
|
|
});
|
|
|
|
commandBody = `${safeContent}\n\n${timeLine}`.trim();
|
|
} else {
|
|
// Internal/trusted source - use original format
|
|
commandBody = `${base}\n${timeLine}`.trim();
|
|
}
|
|
if (deliveryRequested) {
|
|
commandBody =
|
|
`${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 existingSkillsSnapshot = cronSession.sessionEntry.skillsSnapshot;
|
|
const skillsSnapshot = resolveCronSkillsSnapshot({
|
|
workspaceDir,
|
|
config: cfgWithAgentDefaults,
|
|
agentId,
|
|
existingSnapshot: existingSkillsSnapshot,
|
|
isFastTestEnv,
|
|
});
|
|
if (!isFastTestEnv && skillsSnapshot !== existingSkillsSnapshot) {
|
|
cronSession.sessionEntry = {
|
|
...cronSession.sessionEntry,
|
|
updatedAt: Date.now(),
|
|
skillsSnapshot,
|
|
};
|
|
await persistSessionEntry();
|
|
}
|
|
|
|
// Persist systemSent before the run, mirroring the inbound auto-reply behavior.
|
|
cronSession.sessionEntry.systemSent = true;
|
|
await persistSessionEntry();
|
|
|
|
// Resolve auth profile for the session, mirroring the inbound auto-reply path
|
|
// (get-reply-run.ts). Without this, isolated cron sessions fall back to env-var
|
|
// auth which may not match the configured auth-profiles, causing 401 errors.
|
|
const authProfileId = await resolveSessionAuthProfileOverride({
|
|
cfg: cfgWithAgentDefaults,
|
|
provider,
|
|
agentDir,
|
|
sessionEntry: cronSession.sessionEntry,
|
|
sessionStore: cronSession.store,
|
|
sessionKey: agentSessionKey,
|
|
storePath: cronSession.storePath,
|
|
isNewSession: cronSession.isNewSession,
|
|
});
|
|
const authProfileIdSource = cronSession.sessionEntry.authProfileOverrideSource;
|
|
|
|
let runResult: Awaited<ReturnType<typeof runEmbeddedPiAgent>>;
|
|
let fallbackProvider = provider;
|
|
let fallbackModel = model;
|
|
const runStartedAt = Date.now();
|
|
let runEndedAt = runStartedAt;
|
|
try {
|
|
const sessionFile = resolveSessionTranscriptPath(cronSession.sessionEntry.sessionId, agentId);
|
|
const resolvedVerboseLevel =
|
|
normalizeVerboseLevel(cronSession.sessionEntry.verboseLevel) ??
|
|
normalizeVerboseLevel(agentCfg?.verboseDefault) ??
|
|
"off";
|
|
registerAgentRunContext(cronSession.sessionEntry.sessionId, {
|
|
sessionKey: agentSessionKey,
|
|
verboseLevel: resolvedVerboseLevel,
|
|
});
|
|
const messageChannel = resolvedDelivery.channel;
|
|
const fallbackResult = await runWithModelFallback({
|
|
cfg: cfgWithAgentDefaults,
|
|
provider,
|
|
model,
|
|
agentDir,
|
|
fallbacksOverride: resolveAgentModelFallbacksOverride(params.cfg, agentId),
|
|
run: (providerOverride, modelOverride) => {
|
|
if (abortSignal?.aborted) {
|
|
throw new Error(abortReason());
|
|
}
|
|
if (isCliProvider(providerOverride, cfgWithAgentDefaults)) {
|
|
const cliSessionId = getCliSessionId(cronSession.sessionEntry, providerOverride);
|
|
return runCliAgent({
|
|
sessionId: cronSession.sessionEntry.sessionId,
|
|
sessionKey: agentSessionKey,
|
|
agentId,
|
|
sessionFile,
|
|
workspaceDir,
|
|
config: cfgWithAgentDefaults,
|
|
prompt: commandBody,
|
|
provider: providerOverride,
|
|
model: modelOverride,
|
|
thinkLevel,
|
|
timeoutMs,
|
|
runId: cronSession.sessionEntry.sessionId,
|
|
cliSessionId,
|
|
});
|
|
}
|
|
return runEmbeddedPiAgent({
|
|
sessionId: cronSession.sessionEntry.sessionId,
|
|
sessionKey: agentSessionKey,
|
|
agentId,
|
|
messageChannel,
|
|
agentAccountId: resolvedDelivery.accountId,
|
|
sessionFile,
|
|
agentDir,
|
|
workspaceDir,
|
|
config: cfgWithAgentDefaults,
|
|
skillsSnapshot,
|
|
prompt: commandBody,
|
|
lane: params.lane ?? "cron",
|
|
provider: providerOverride,
|
|
model: modelOverride,
|
|
authProfileId,
|
|
authProfileIdSource,
|
|
thinkLevel,
|
|
verboseLevel: resolvedVerboseLevel,
|
|
timeoutMs,
|
|
runId: cronSession.sessionEntry.sessionId,
|
|
requireExplicitMessageTarget: true,
|
|
disableMessageTool: deliveryRequested,
|
|
abortSignal,
|
|
});
|
|
},
|
|
});
|
|
runResult = fallbackResult.result;
|
|
fallbackProvider = fallbackResult.provider;
|
|
fallbackModel = fallbackResult.model;
|
|
runEndedAt = Date.now();
|
|
} catch (err) {
|
|
return withRunSession({ status: "error", error: String(err) });
|
|
}
|
|
|
|
if (isAborted()) {
|
|
return withRunSession({ status: "error", error: abortReason() });
|
|
}
|
|
|
|
const payloads = runResult.payloads ?? [];
|
|
|
|
// Update token+model fields in the session store.
|
|
// Also collect best-effort telemetry for the cron run log.
|
|
let telemetry: CronRunTelemetry | undefined;
|
|
{
|
|
const usage = runResult.meta?.agentMeta?.usage;
|
|
const promptTokens = runResult.meta?.agentMeta?.promptTokens;
|
|
const modelUsed = runResult.meta?.agentMeta?.model ?? fallbackModel ?? model;
|
|
const providerUsed = runResult.meta?.agentMeta?.provider ?? fallbackProvider ?? provider;
|
|
const contextTokens =
|
|
agentCfg?.contextTokens ?? lookupContextTokens(modelUsed) ?? DEFAULT_CONTEXT_TOKENS;
|
|
|
|
cronSession.sessionEntry.modelProvider = providerUsed;
|
|
cronSession.sessionEntry.model = modelUsed;
|
|
cronSession.sessionEntry.contextTokens = contextTokens;
|
|
if (isCliProvider(providerUsed, cfgWithAgentDefaults)) {
|
|
const cliSessionId = runResult.meta?.agentMeta?.sessionId?.trim();
|
|
if (cliSessionId) {
|
|
setCliSessionId(cronSession.sessionEntry, providerUsed, cliSessionId);
|
|
}
|
|
}
|
|
if (hasNonzeroUsage(usage)) {
|
|
const input = usage.input ?? 0;
|
|
const output = usage.output ?? 0;
|
|
const totalTokens =
|
|
deriveSessionTotalTokens({
|
|
usage,
|
|
contextTokens,
|
|
promptTokens,
|
|
}) ?? input;
|
|
cronSession.sessionEntry.inputTokens = input;
|
|
cronSession.sessionEntry.outputTokens = output;
|
|
cronSession.sessionEntry.totalTokens = totalTokens;
|
|
cronSession.sessionEntry.totalTokensFresh = true;
|
|
cronSession.sessionEntry.cacheRead = usage.cacheRead ?? 0;
|
|
cronSession.sessionEntry.cacheWrite = usage.cacheWrite ?? 0;
|
|
|
|
telemetry = {
|
|
model: modelUsed,
|
|
provider: providerUsed,
|
|
usage: {
|
|
input_tokens: input,
|
|
output_tokens: output,
|
|
total_tokens: totalTokens,
|
|
},
|
|
};
|
|
} else {
|
|
telemetry = {
|
|
model: modelUsed,
|
|
provider: providerUsed,
|
|
};
|
|
}
|
|
await persistSessionEntry();
|
|
}
|
|
|
|
if (isAborted()) {
|
|
return withRunSession({ status: "error", error: abortReason(), ...telemetry });
|
|
}
|
|
const firstText = payloads[0]?.text ?? "";
|
|
let summary = pickSummaryFromPayloads(payloads) ?? pickSummaryFromOutput(firstText);
|
|
let outputText = pickLastNonEmptyTextFromPayloads(payloads);
|
|
let synthesizedText = outputText?.trim() || summary?.trim() || undefined;
|
|
const deliveryPayload = pickLastDeliverablePayload(payloads);
|
|
let deliveryPayloads =
|
|
deliveryPayload !== undefined
|
|
? [deliveryPayload]
|
|
: synthesizedText
|
|
? [{ text: synthesizedText }]
|
|
: [];
|
|
const deliveryPayloadHasStructuredContent =
|
|
Boolean(deliveryPayload?.mediaUrl) ||
|
|
(deliveryPayload?.mediaUrls?.length ?? 0) > 0 ||
|
|
Object.keys(deliveryPayload?.channelData ?? {}).length > 0;
|
|
const deliveryBestEffort = resolveCronDeliveryBestEffort(params.job);
|
|
|
|
// Skip delivery for heartbeat-only responses (HEARTBEAT_OK with no real content).
|
|
const ackMaxChars = resolveHeartbeatAckMaxChars(agentCfg);
|
|
const skipHeartbeatDelivery = deliveryRequested && isHeartbeatOnlyResponse(payloads, ackMaxChars);
|
|
const skipMessagingToolDelivery =
|
|
deliveryRequested &&
|
|
runResult.didSendViaMessagingTool === true &&
|
|
(runResult.messagingToolSentTargets ?? []).some((target) =>
|
|
matchesMessagingToolDeliveryTarget(target, {
|
|
channel: resolvedDelivery.channel,
|
|
to: resolvedDelivery.to,
|
|
accountId: resolvedDelivery.accountId,
|
|
}),
|
|
);
|
|
|
|
// `true` means we confirmed at least one outbound send reached the target.
|
|
// Keep this strict so timer fallback can safely decide whether to wake main.
|
|
let delivered = skipMessagingToolDelivery;
|
|
const failDeliveryTarget = (error: string) =>
|
|
withRunSession({
|
|
status: "error",
|
|
error,
|
|
errorKind: "delivery-target",
|
|
summary,
|
|
outputText,
|
|
...telemetry,
|
|
});
|
|
if (deliveryRequested && !skipHeartbeatDelivery && !skipMessagingToolDelivery) {
|
|
if (resolvedDelivery.error) {
|
|
if (!deliveryBestEffort) {
|
|
return failDeliveryTarget(resolvedDelivery.error.message);
|
|
}
|
|
logWarn(`[cron:${params.job.id}] ${resolvedDelivery.error.message}`);
|
|
return withRunSession({ status: "ok", summary, outputText, ...telemetry });
|
|
}
|
|
const failOrWarnMissingDeliveryField = (message: string) => {
|
|
if (!deliveryBestEffort) {
|
|
return failDeliveryTarget(message);
|
|
}
|
|
logWarn(`[cron:${params.job.id}] ${message}`);
|
|
return withRunSession({ status: "ok", summary, outputText, ...telemetry });
|
|
};
|
|
if (!resolvedDelivery.channel) {
|
|
return failOrWarnMissingDeliveryField("cron delivery channel is missing");
|
|
}
|
|
if (!resolvedDelivery.to) {
|
|
return failOrWarnMissingDeliveryField("cron delivery target is missing");
|
|
}
|
|
const identity = resolveAgentOutboundIdentity(cfgWithAgentDefaults, agentId);
|
|
|
|
// Route text-only cron announce output back through the main session so it
|
|
// follows the same system-message injection path as subagent completions.
|
|
// Keep direct outbound delivery only for structured payloads (media/channel
|
|
// data), which cannot be represented by the shared announce flow.
|
|
//
|
|
// Forum/topic targets should also use direct delivery. Announce flow can
|
|
// be swallowed by ANNOUNCE_SKIP/NO_REPLY in the target agent turn, which
|
|
// silently drops cron output for topic-bound sessions.
|
|
const useDirectDelivery =
|
|
deliveryPayloadHasStructuredContent || resolvedDelivery.threadId != null;
|
|
if (useDirectDelivery) {
|
|
try {
|
|
const payloadsForDelivery =
|
|
deliveryPayloads.length > 0
|
|
? deliveryPayloads
|
|
: synthesizedText
|
|
? [{ text: synthesizedText }]
|
|
: [];
|
|
if (payloadsForDelivery.length > 0) {
|
|
if (isAborted()) {
|
|
return withRunSession({ status: "error", error: abortReason(), ...telemetry });
|
|
}
|
|
const deliveryResults = await deliverOutboundPayloads({
|
|
cfg: cfgWithAgentDefaults,
|
|
channel: resolvedDelivery.channel,
|
|
to: resolvedDelivery.to,
|
|
accountId: resolvedDelivery.accountId,
|
|
threadId: resolvedDelivery.threadId,
|
|
payloads: payloadsForDelivery,
|
|
agentId,
|
|
identity,
|
|
bestEffort: deliveryBestEffort,
|
|
deps: createOutboundSendDeps(params.deps),
|
|
abortSignal,
|
|
});
|
|
delivered = deliveryResults.length > 0;
|
|
}
|
|
} catch (err) {
|
|
if (!deliveryBestEffort) {
|
|
return withRunSession({
|
|
status: "error",
|
|
summary,
|
|
outputText,
|
|
error: String(err),
|
|
...telemetry,
|
|
});
|
|
}
|
|
}
|
|
} else if (synthesizedText) {
|
|
const announceMainSessionKey = resolveAgentMainSessionKey({
|
|
cfg: params.cfg,
|
|
agentId,
|
|
});
|
|
const announceSessionKey = await resolveCronAnnounceSessionKey({
|
|
cfg: cfgWithAgentDefaults,
|
|
agentId,
|
|
fallbackSessionKey: announceMainSessionKey,
|
|
delivery: {
|
|
channel: resolvedDelivery.channel,
|
|
to: resolvedDelivery.to,
|
|
accountId: resolvedDelivery.accountId,
|
|
threadId: resolvedDelivery.threadId,
|
|
},
|
|
});
|
|
const taskLabel =
|
|
typeof params.job.name === "string" && params.job.name.trim()
|
|
? params.job.name.trim()
|
|
: `cron:${params.job.id}`;
|
|
const initialSynthesizedText = synthesizedText.trim();
|
|
let activeSubagentRuns = countActiveDescendantRuns(agentSessionKey);
|
|
const expectedSubagentFollowup = expectsSubagentFollowup(initialSynthesizedText);
|
|
const hadActiveDescendants = activeSubagentRuns > 0;
|
|
if (activeSubagentRuns > 0 || expectedSubagentFollowup) {
|
|
let finalReply = await waitForDescendantSubagentSummary({
|
|
sessionKey: agentSessionKey,
|
|
initialReply: initialSynthesizedText,
|
|
timeoutMs,
|
|
observedActiveDescendants: activeSubagentRuns > 0 || expectedSubagentFollowup,
|
|
});
|
|
activeSubagentRuns = countActiveDescendantRuns(agentSessionKey);
|
|
if (
|
|
!finalReply &&
|
|
activeSubagentRuns === 0 &&
|
|
(hadActiveDescendants || expectedSubagentFollowup)
|
|
) {
|
|
finalReply = await readDescendantSubagentFallbackReply({
|
|
sessionKey: agentSessionKey,
|
|
runStartedAt,
|
|
});
|
|
}
|
|
if (finalReply && activeSubagentRuns === 0) {
|
|
outputText = finalReply;
|
|
summary = pickSummaryFromOutput(finalReply) ?? summary;
|
|
synthesizedText = finalReply;
|
|
deliveryPayloads = [{ text: finalReply }];
|
|
}
|
|
}
|
|
if (activeSubagentRuns > 0) {
|
|
// Parent orchestration is still in progress; avoid announcing a partial
|
|
// update to the main requester.
|
|
return withRunSession({ status: "ok", summary, outputText, ...telemetry });
|
|
}
|
|
if (
|
|
(hadActiveDescendants || expectedSubagentFollowup) &&
|
|
synthesizedText.trim() === initialSynthesizedText &&
|
|
isLikelyInterimCronMessage(initialSynthesizedText) &&
|
|
initialSynthesizedText.toUpperCase() !== SILENT_REPLY_TOKEN.toUpperCase()
|
|
) {
|
|
// Descendants existed but no post-orchestration synthesis arrived, so
|
|
// suppress stale parent text like "on it, pulling everything together".
|
|
return withRunSession({ status: "ok", summary, outputText, ...telemetry });
|
|
}
|
|
if (synthesizedText.toUpperCase() === SILENT_REPLY_TOKEN.toUpperCase()) {
|
|
return withRunSession({ status: "ok", summary, outputText, delivered: true, ...telemetry });
|
|
}
|
|
try {
|
|
if (isAborted()) {
|
|
return withRunSession({ status: "error", error: abortReason(), ...telemetry });
|
|
}
|
|
const didAnnounce = await runSubagentAnnounceFlow({
|
|
childSessionKey: agentSessionKey,
|
|
childRunId: `${params.job.id}:${runSessionId}`,
|
|
requesterSessionKey: announceSessionKey,
|
|
requesterOrigin: {
|
|
channel: resolvedDelivery.channel,
|
|
to: resolvedDelivery.to,
|
|
accountId: resolvedDelivery.accountId,
|
|
threadId: resolvedDelivery.threadId,
|
|
},
|
|
requesterDisplayKey: announceSessionKey,
|
|
task: taskLabel,
|
|
timeoutMs,
|
|
cleanup: params.job.deleteAfterRun ? "delete" : "keep",
|
|
roundOneReply: synthesizedText,
|
|
waitForCompletion: false,
|
|
startedAt: runStartedAt,
|
|
endedAt: runEndedAt,
|
|
outcome: { status: "ok" },
|
|
announceType: "cron job",
|
|
signal: abortSignal,
|
|
});
|
|
if (didAnnounce) {
|
|
delivered = true;
|
|
} else {
|
|
const message = "cron announce delivery failed";
|
|
if (!deliveryBestEffort) {
|
|
return withRunSession({
|
|
status: "error",
|
|
summary,
|
|
outputText,
|
|
error: message,
|
|
...telemetry,
|
|
});
|
|
}
|
|
logWarn(`[cron:${params.job.id}] ${message}`);
|
|
}
|
|
} catch (err) {
|
|
if (!deliveryBestEffort) {
|
|
return withRunSession({
|
|
status: "error",
|
|
summary,
|
|
outputText,
|
|
error: String(err),
|
|
...telemetry,
|
|
});
|
|
}
|
|
logWarn(`[cron:${params.job.id}] ${String(err)}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
return withRunSession({ status: "ok", summary, outputText, delivered, ...telemetry });
|
|
}
|