/** * Fire path: new root Session, wrapped prompt, hide from workspace list, history. * Collaborators are injected so tests drive the same functions as Host apply. */ import { randomUUID } from 'node:crypto' import { wrapScheduledPrompt, scheduledPromptSource } from './prompt.js' import { decideDispatch } from './scheduler.js' import { appendRun, getJob, patchRun, upsertJob } from './store.js' import { ACTIVE_RUN_STATUSES } from './fire-status.js' import { findLabelOverlapRun, normalizeLabels, projectJobState, } from './monitor.js' export { ACTIVE_RUN_STATUSES } from './fire-status.js' export function createRunRecord(job, decision, now, trigger) { return { id: randomUUID(), jobId: job.id, scheduledAt: decision.scheduledAt ?? now, actualAt: now, status: 'queued', stateEnteredAt: now, exitedAt: null, trigger: trigger || 'schedule', sessionId: null, error: null, summary: '', reason: decision.reason || null, outputRef: null, } } export const TITLE_PREFIX = '定时任务 · ' export function extractAssistantText(messages, maxChars = 4000) { if (!Array.isArray(messages)) return '' for (let i = messages.length - 1; i >= 0; i--) { const message = messages[i] if (!message || message.role !== 'assistant') continue const parts = [] for (const block of message.content || []) { if (block && block.type === 'text' && typeof block.text === 'string' && block.text.trim()) { parts.push(block.text.trim()) } } const text = parts.join('\n').trim() if (!text) continue return text.length > maxChars ? `${text.slice(0, maxChars)}…` : text } return '' } export function publicJob(job, runs = null) { if (!job) return null const jobRuns = Array.isArray(runs) ? runs.filter((run) => run?.jobId === job.id) : null const projection = jobRuns ? projectJobState(job, jobRuns) : projectJobState(job, job.lastStatus ? [{ status: job.lastStatus, stateEnteredAt: job.lastRunAt, actualAt: job.lastRunAt, exitedAt: job.lastRunAt }] : []) // When runs not provided, prefer lastStatus-based active hint only for running/queued tip. let stateInfo = projection if (!jobRuns && (job.lastStatus === 'running' || job.lastStatus === 'queued')) { stateInfo = { state: job.lastStatus === 'queued' ? 'pending' : 'running', stateEnteredAt: job.lastRunAt || job.updatedAt || null, activeRunId: null, stuck: false, } } else if (!jobRuns && job.enabled === false) { stateInfo = { state: 'paused', stateEnteredAt: job.updatedAt || job.createdAt || null, activeRunId: null, stuck: false, } } else if (!jobRuns && job.schedule?.kind === 'at' && job.nextRunAt == null && (job.lastStatus === 'succeeded' || job.lastStatus === 'failed' || job.lastStatus === 'skipped')) { stateInfo = { state: job.lastStatus === 'succeeded' ? 'succeeded' : 'failed', stateEnteredAt: job.lastRunAt || null, activeRunId: null, stuck: false, } } else if (!jobRuns) { stateInfo = { state: 'idle', stateEnteredAt: job.lastRunAt || job.updatedAt || job.createdAt || null, activeRunId: null, stuck: false, } } return { id: job.id, name: job.name, prompt: job.prompt, enabled: job.enabled !== false, cwd: job.cwd || '', timeoutMinutes: job.timeoutMinutes || 10, provider: job.provider || '', model: job.model || '', reasoningEffort: job.reasoningEffort || '', agentPreset: job.agentPreset || '', ownerEmpNo: job.ownerEmpNo || '', ownerDisplayName: job.ownerDisplayName || '', delivery: job.delivery && job.delivery.kind === 'im' ? { kind: 'im', botId: job.delivery.botId, targetId: job.delivery.targetId, ...job.delivery.mentionJid ? { mentionJid: job.delivery.mentionJid } : {}, ...job.delivery.mentionName ? { mentionName: job.delivery.mentionName } : {}, } : { kind: 'dsh' }, mirrorToSession: job.mirrorToSession === true, labels: normalizeLabels(job.labels), ...job.watch ? { watch: job.watch } : {}, ...job.progress ? { progress: job.progress } : {}, ...job.report ? { report: job.report } : {}, persistHistory: job.persistHistory || { kind: 'retain', limit: 200 }, state: stateInfo.state, stateEnteredAt: stateInfo.stateEnteredAt, stuck: stateInfo.stuck === true, ...job.origin ? { origin: job.origin } : {}, schedule: job.schedule, createdAt: job.createdAt, updatedAt: job.updatedAt, lastRunAt: job.lastRunAt, lastStatus: job.lastStatus, nextRunAt: job.nextRunAt, } } /** * Claim one occurrence: overlap → skipped history row; otherwise queued run. * Also skips when another job with the same role+task labels is already active. */ export function claimOccurrence(state, jobId, now, trigger, policies) { const job = getJob(state, jobId) if (!job) { const error = new Error('job not found') error.code = 'NOT_FOUND' throw error } const runs = (state.runs || []).filter((run) => run.jobId === jobId) let decision = trigger === 'run-now' ? decideRunNow(job, runs, now, policies) : decideDispatch({ job, runs, now, overlapPolicy: policies?.overlapPolicy || state.settings?.overlapPolicy, misfirePolicy: policies?.misfirePolicy || state.settings?.misfirePolicy, graceMs: policies?.graceMs, }) if (decision.action === 'fire') { const overlap = findLabelOverlapRun(state, job) if (overlap) { decision = { action: 'skip', reason: 'label_overlap', scheduledAt: decision.scheduledAt ?? now, nextRunAt: Object.hasOwn(decision, 'nextRunAt') ? decision.nextRunAt : job.nextRunAt, } } } if (decision.action === 'wait') { return { state, job: publicJob(job, state.runs), decision, run: null } } const run = createRunRecord(job, decision, now, trigger) if (decision.action === 'skip') { run.status = 'skipped' run.stateEnteredAt = now run.exitedAt = now run.reason = decision.reason const oneshot = job.schedule?.kind === 'at' if (decision.reason === 'overlap') { run.summary = 'Skipped because a run is already queued or running' } else if (decision.reason === 'label_overlap') { run.summary = 'Skipped because another job with the same role+task labels is already running' } else if (decision.reason === 'misfire' && oneshot) { run.summary = 'Skipped: one-shot time missed (host downtime or delayed tick; no backlog). nextRunAt cleared.' run.error = 'misfire:one-shot' } else if (decision.reason === 'misfire') { run.summary = 'Skipped missed occurrence after host downtime (no backlog)' run.error = 'misfire' } else { run.summary = `Skipped (${decision.reason || 'policy'})` } const alreadySkipped = runs.some((row) => ( row && row.status === 'skipped' && row.reason === decision.reason && row.scheduledAt === decision.scheduledAt )) const nextJob = { ...job, updatedAt: now, lastRunAt: now, lastStatus: 'skipped', nextRunAt: Object.hasOwn(decision, 'nextRunAt') ? decision.nextRunAt : job.nextRunAt, } let next = upsertJob(state, nextJob) if (!alreadySkipped) next = appendRun(next, run) return { state: next, job: publicJob(nextJob, next.runs), decision, run: alreadySkipped ? null : run } } run.status = 'queued' run.stateEnteredAt = now const nextJob = { ...job, updatedAt: now, nextRunAt: trigger === 'run-now' ? job.nextRunAt : (Object.hasOwn(decision, 'nextRunAt') ? decision.nextRunAt : job.nextRunAt), } let next = upsertJob(state, nextJob) next = appendRun(next, run) return { state: next, job: publicJob(nextJob, next.runs), decision, run } } function decideRunNow(job, runs, now, policies) { const overlapPolicy = policies?.overlapPolicy || 'skip' if (overlapPolicy === 'skip' && runs.some((run) => run.status === 'queued' || run.status === 'running')) { return { action: 'skip', reason: 'overlap', scheduledAt: now, nextRunAt: job.nextRunAt } } return { action: 'fire', scheduledAt: now, nextRunAt: job.nextRunAt } } function failRun(state, runId, job, now, message) { return settleRun(state, runId, { status: 'failed', error: message, summary: message, }, now) } /** Persist a terminal status so overlap skip can release the job. */ export function settleRun(state, runId, terminal, now) { const run = (state.runs || []).find((row) => row.id === runId) if (!run) return state const job = getJob(state, run.jobId) const status = terminal?.status && !ACTIVE_RUN_STATUSES.has(terminal.status) ? terminal.status : 'succeeded' const prevStatus = run.status let next = patchRun(state, runId, { status, summary: terminal?.summary || '', error: terminal?.error || null, actualAt: now, stateEnteredAt: prevStatus === status ? (run.stateEnteredAt || now) : now, exitedAt: now, sessionId: run.sessionId, ...terminal?.outputRef ? { outputRef: terminal.outputRef } : {}, }) if (job) { next = upsertJob(next, { ...job, lastRunAt: now, lastStatus: status, updatedAt: now, }) } return next } /** * Execute an already-claimed queued run via injected session/archive helpers. * Performs IO through `deps`; the caller persists each returned state. * * @param {object} state * @param {string} runId * @param {object} deps * @param {() => number} [deps.now] * @param {(args: object) => Promise<{ sessionId: string, status?: string, summary?: string }>} deps.createAndPrompt * @param {(sessionId: string) => Promise} [deps.archiveSession] * @param {(sessionId: string) => Promise<{ status?: string, summary?: string, error?: string }>} [deps.waitForTurn] */ export async function executeClaimedRun(state, runId, deps) { const run = (state.runs || []).find((row) => row.id === runId) if (!run) { const error = new Error('run not found') error.code = 'NOT_FOUND' throw error } const job = getJob(state, run.jobId) if (!job) { const error = new Error('job not found') error.code = 'NOT_FOUND' throw error } const clock = deps.now || (() => Date.now()) const mark = clock() let next = patchRun(state, runId, { status: 'running', actualAt: mark, stateEnteredAt: mark, exitedAt: null, }) next = upsertJob(next, { ...job, lastRunAt: mark, lastStatus: 'running', updatedAt: mark }) const text = wrapScheduledPrompt(job.prompt, { jobName: job.name, jobId: job.id, runId, scheduledAt: new Date(run.scheduledAt).toISOString(), }) const source = scheduledPromptSource() if (typeof deps.createAndPrompt !== 'function') { next = failRun(next, runId, job, clock(), 'session port unavailable') return { state: next, run: (next.runs || []).find((row) => row.id === runId), job: publicJob(getJob(next, job.id), next.runs), } } try { const created = await deps.createAndPrompt({ job, run: { ...run, status: 'running' }, text, source, }) const sessionId = created?.sessionId if (!sessionId) throw new Error('createAndPrompt did not return sessionId') next = patchRun(next, runId, { sessionId }) if (typeof deps.archiveSession === 'function') { await deps.archiveSession(sessionId) } const reported = created.status || 'succeeded' if (typeof deps.waitForTurn === 'function' && ACTIVE_RUN_STATUSES.has(reported)) { const waited = await deps.waitForTurn(sessionId) const ended = clock() next = settleRun(next, runId, waited || { status: 'succeeded', summary: created.summary }, ended) return { state: next, run: (next.runs || []).find((row) => row.id === runId), job: publicJob(getJob(next, job.id), next.runs), } } if (ACTIVE_RUN_STATUSES.has(reported)) { // Leave running so overlap skip still works while a turn is in flight. // Caller (dispatchRun / live waitForTurn) must settle to a terminal status. return { state: next, run: (next.runs || []).find((row) => row.id === runId), job: publicJob(getJob(next, job.id), next.runs), } } const ended = clock() next = settleRun(next, runId, { status: reported, summary: created.summary || 'Dispatched to a new session', error: created.error || null, }, ended) return { state: next, run: (next.runs || []).find((row) => row.id === runId), job: publicJob(getJob(next, job.id), next.runs), } } catch (error) { const message = error instanceof Error ? error.message : String(error) next = failRun(next, runId, job, clock(), message) return { state: next, run: (next.runs || []).find((row) => row.id === runId), job: publicJob(getJob(next, job.id), next.runs), } } } export function interruptActiveRuns(state, now, reason = 'host_interrupted') { let next = state for (const run of state.runs || []) { if (run.status !== 'queued' && run.status !== 'running') continue next = patchRun(next, run.id, { status: 'failed', error: reason, summary: 'Host stopped before this run finished', actualAt: now, stateEnteredAt: now, exitedAt: now, }) const job = getJob(next, run.jobId) if (job) { next = upsertJob(next, { ...job, lastStatus: 'failed', lastRunAt: now, updatedAt: now }) } } return next } export { wrapScheduledPrompt }