/** * Durable JSON store for jobs, runs, and hidden session ids. * Atomic write (tmp + rename). No Cordis imports. */ import { mkdir, readFile, rename, writeFile } from 'node:fs/promises' import { dirname, join } from 'node:path' import { randomUUID } from 'node:crypto' import { applyRunIsolation, recordHiddenSession } from './isolation.js' import { nextFire, validateSchedule } from './scheduler.js' import { normalizeDelivery, normalizeOrigin } from './delivery.js' import { normalizeAgentPresetId } from './preset.js' import { normalizeOwnerEmpNo, UNASSIGNED_OWNER } from './ownership.js' import { normalizeLabels, normalizePersistHistory, normalizeProgress, normalizeReport, normalizeWatch, persistHistoryLimit, } from './monitor.js' import { ACTIVE_RUN_STATUSES } from './fire-status.js' export { UNASSIGNED_OWNER } export const STORE_VERSION = 1 export const DEFAULT_SETTINGS = { enabled: true, timezone: 'Asia/Shanghai', historyLimit: 200, overlapPolicy: 'skip', misfirePolicy: 'skip', /** * When true (default), an explicit job cwd that lies OUTSIDE the multi-tenant * provision forest (workspaceRoot/) is kept — e.g. D:\\code\\gpt. * Paths under the forest but not the owner's tree are still clamped. */ allowExternalCwd: true, } export function emptyState() { return { version: STORE_VERSION, settings: { ...DEFAULT_SETTINGS }, jobs: [], runs: [], hiddenSessionIds: [], } } export function normalizeSettings(input = {}, fallback = DEFAULT_SETTINGS) { const src = input && typeof input === 'object' ? input : {} const historyLimit = Number(src.historyLimit) const allowExternalCwd = src.allowExternalCwd !== undefined ? src.allowExternalCwd !== false : fallback.allowExternalCwd !== false return { enabled: src.enabled !== false, timezone: typeof src.timezone === 'string' && src.timezone.trim() ? src.timezone.trim() : fallback.timezone, historyLimit: Number.isInteger(historyLimit) && historyLimit >= 10 ? Math.min(2000, historyLimit) : fallback.historyLimit, overlapPolicy: src.overlapPolicy === 'skip' ? 'skip' : fallback.overlapPolicy, misfirePolicy: src.misfirePolicy === 'skip' ? 'skip' : fallback.misfirePolicy, allowExternalCwd, } } function cloneState(state) { return structuredClone(state) } export function newId() { return randomUUID() } /** * Whether agents may mutate this job via cron_* tools. * Default allow (= current behavior). Only the sidebar should set deny. * @param {unknown} input job or raw value * @returns {'allow'|'deny'} */ export function normalizeAgentAccess(input) { const raw = input && typeof input === 'object' && !Array.isArray(input) ? (input.agentAccess ?? input.agent_access) : input if (raw === false || raw === 0 || raw === 'deny' || raw === 'human' || raw === 'locked' || raw === 'off') { return 'deny' } return 'allow' } export function agentMayMutateJob(job) { return normalizeAgentAccess(job) === 'allow' } /** Host permission-preset ids (DSH 访问模式). Empty = inherit Host default for new sessions. */ export const PERMISSION_PRESET_IDS = Object.freeze([ 'read-only', 'workspace-write', 'danger-full-access', ]) /** * @param {unknown} input job or raw value * @returns {''|'read-only'|'workspace-write'|'danger-full-access'} */ export function normalizePermissionPreset(input) { const raw = input && typeof input === 'object' && !Array.isArray(input) ? (input.permissionPreset ?? input.permission_preset ?? input.accessMode ?? input.access_mode) : input const value = String(raw || '').trim().toLowerCase() if (!value || value === 'default' || value === 'inherit' || value === 'host') return '' if (value === 'readonly' || value === 'read_only' || value === 'view' || value === 'read-only') { return 'read-only' } if (value === 'workspace' || value === 'workspace_write' || value === 'write' || value === 'workspace-write') { return 'workspace-write' } if (value === 'full' || value === 'fullaccess' || value === 'full_access' || value === 'danger-full-access') { return 'danger-full-access' } if (PERMISSION_PRESET_IDS.includes(value)) return value const error = new Error(`permissionPreset must be empty|read-only|workspace-write|danger-full-access (got ${JSON.stringify(raw)})`) error.code = 'INVALID_PERMISSION_PRESET' throw error } /** * Repair provider/model pairs when LLMs or hosts pass a combined "provider/model" * route in one or both fields (e.g. provider=model="zte/Qwen3-…" → zte + Qwen3-…). * Leaves legitimate model ids that contain "/" alone when provider is a distinct id. * @param {unknown} provider * @param {unknown} model * @returns {{ provider: string, model: string }} */ export function splitProviderModel(provider, model) { let p = String(provider || '').trim() let m = String(model || '').trim() if (!p && !m) return { provider: '', model: '' } if (p && m && p === m && p.includes('/')) { const at = p.indexOf('/') return { provider: p.slice(0, at).trim(), model: p.slice(at + 1).trim() } } if (!p && m.includes('/')) { const at = m.indexOf('/') return { provider: m.slice(0, at).trim(), model: m.slice(at + 1).trim() } } if (p.includes('/') && !m) { const at = p.indexOf('/') return { provider: p.slice(0, at).trim(), model: p.slice(at + 1).trim() } } if (p && m.startsWith(`${p}/`)) { return { provider: p, model: m.slice(p.length + 1).trim() } } if (p.includes('/') && m && p.endsWith(`/${m}`)) { const at = p.indexOf('/') return { provider: p.slice(0, at).trim(), model: m } } return { provider: p, model: m } } export function normalizeJobModel(input = {}) { const reasoningEffort = typeof input.reasoningEffort === 'string' ? input.reasoningEffort.trim() : (typeof input.reasoning_effort === 'string' ? input.reasoning_effort.trim() : '') const { provider, model } = splitProviderModel(input.provider, input.model) if (!provider && !model) { return { provider: '', model: '', reasoningEffort: '' } } if (!provider || !model) { const error = new Error('provider and model must be set together') error.code = 'INVALID_JOB' throw error } return { provider, model, reasoningEffort } } /** * @param {object} input * @param {object} state * @param {number} now */ export function createJobRecord(input, state, now) { const name = String(input?.name || '').trim() const prompt = String(input?.prompt || '').trim() if (!name) { const error = new Error('name is required') error.code = 'INVALID_JOB' throw error } if (!prompt) { const error = new Error('prompt is required') error.code = 'INVALID_JOB' throw error } const settings = state.settings || DEFAULT_SETTINGS const schedule = validateSchedule(input.schedule, settings.timezone) const timeoutMinutes = Number(input.timeoutMinutes) const model = normalizeJobModel(input) const delivery = normalizeDelivery(input.delivery) const origin = normalizeOrigin(input.origin) const agentPreset = normalizeAgentPresetId(input.agentPreset ?? input.agent_preset) const mirrorToSession = input.mirrorToSession === true || input.mirror_to_session === true const ownerEmpNo = normalizeOwnerEmpNo(input.ownerEmpNo) const ownerDisplayName = typeof input.ownerDisplayName === 'string' ? input.ownerDisplayName.trim().slice(0, 80) : '' const labels = normalizeLabels(input.labels) const watch = input.watch !== undefined ? normalizeWatch(input.watch) : null const progress = input.progress !== undefined ? normalizeProgress(input.progress) : null const report = input.report !== undefined ? normalizeReport(input.report, delivery) : null const persistHistory = normalizePersistHistory( input.persistHistory ?? input.persist_history, settings.historyLimit, ) const agentAccess = normalizeAgentAccess(input) const permissionPreset = normalizePermissionPreset(input) const job = { id: String(input.id || newId()), name, prompt, enabled: input.enabled !== false, cwd: typeof input.cwd === 'string' ? input.cwd.trim() : '', timeoutMinutes: Number.isFinite(timeoutMinutes) && timeoutMinutes >= 1 ? Math.min(240, Math.round(timeoutMinutes)) : 10, provider: model.provider, model: model.model, reasoningEffort: model.reasoningEffort, agentPreset, agentAccess, permissionPreset, delivery, mirrorToSession, ownerEmpNo, ownerDisplayName, labels, ...watch ? { watch } : {}, ...progress ? { progress } : {}, ...report ? { report } : {}, persistHistory, ...origin ? { origin } : {}, schedule: schedule.kind === 'cron' ? { kind: 'cron', expr: schedule.expr, timezone: schedule.timezone } : { kind: 'at', at: new Date(schedule.at).toISOString(), timezone: schedule.timezone }, createdAt: now, updatedAt: now, lastRunAt: null, lastStatus: null, nextRunAt: nextFire(schedule, now, schedule.timezone), } if (schedule.kind === 'at') { const atMs = Date.parse(job.schedule.at) // Slightly past (within 60s): fire on next tick instead of rejecting or freezing. if (Number.isFinite(atMs) && atMs <= now && atMs >= now - 60_000) { job.nextRunAt = now } else if (job.nextRunAt == null || job.nextRunAt < now - 60_000) { const error = new Error(`that one-shot time is already in the past (now is ${new Date(now).toISOString()})`) error.code = 'INVALID_AT' throw error } } return job } export function pruneRuns(runs, historyLimit, jobs = []) { const list = Array.isArray(runs) ? runs : [] const defaultLimit = Number.isInteger(historyLimit) ? historyLimit : DEFAULT_SETTINGS.historyLimit const byJob = new Map() for (const job of jobs || []) { if (job?.id) byJob.set(job.id, job) } const active = [] /** @type {Map} */ const terminalsByJob = new Map() const orphanTerminal = [] for (const run of list) { if (!run) continue if (ACTIVE_RUN_STATUSES.has(run.status)) { active.push(run) continue } const jobId = run.jobId || '' if (!jobId || !byJob.has(jobId)) { orphanTerminal.push(run) continue } if (!terminalsByJob.has(jobId)) terminalsByJob.set(jobId, []) terminalsByJob.get(jobId).push(run) } const keptTerminal = [] for (const [jobId, rows] of terminalsByJob) { const job = byJob.get(jobId) const policy = job?.persistHistory || normalizePersistHistory(undefined, defaultLimit) const limit = persistHistoryLimit(policy, defaultLimit) rows.sort((a, b) => (b.actualAt || b.scheduledAt || 0) - (a.actualAt || a.scheduledAt || 0)) if (!Number.isFinite(limit)) { keptTerminal.push(...rows) } else { keptTerminal.push(...rows.slice(0, Math.max(10, limit))) } } orphanTerminal.sort((a, b) => (b.actualAt || b.scheduledAt || 0) - (a.actualAt || a.scheduledAt || 0)) keptTerminal.push(...orphanTerminal.slice(0, Math.max(10, defaultLimit))) return [...active, ...keptTerminal] } export function listJobs(state) { return [...(state?.jobs || [])] } export function getJob(state, jobId) { return (state?.jobs || []).find((job) => job.id === jobId) || null } /** * Filter history with optional state / time / cursor pagination. * @returns {{ runs: object[], nextCursor: string|null, total: number }} */ export function listHistoryPage(state, jobId, opts = {}) { const runs = state?.runs || [] let filtered = jobId ? runs.filter((run) => run.jobId === jobId) : [...runs] const stateFilter = typeof opts.state === 'string' ? opts.state.trim() : '' if (stateFilter) { filtered = filtered.filter((run) => { if (stateFilter === 'pending') return run.status === 'queued' if (stateFilter === 'running') return run.status === 'running' if (stateFilter === 'succeeded') return run.status === 'succeeded' if (stateFilter === 'failed') return run.status === 'failed' || run.status === 'skipped' return run.status === stateFilter }) } const from = Number(opts.from) const to = Number(opts.to) if (Number.isFinite(from)) { filtered = filtered.filter((run) => (run.actualAt || run.scheduledAt || 0) >= from) } if (Number.isFinite(to)) { filtered = filtered.filter((run) => (run.actualAt || run.scheduledAt || 0) <= to) } filtered.sort((a, b) => (b.actualAt || b.scheduledAt || 0) - (a.actualAt || a.scheduledAt || 0)) const cursor = typeof opts.cursor === 'string' ? opts.cursor.trim() : '' if (cursor) { const idx = filtered.findIndex((run) => run.id === cursor) if (idx >= 0) filtered = filtered.slice(idx + 1) } const limit = Number(opts.limit) const take = Number.isInteger(limit) && limit > 0 ? Math.min(500, limit) : filtered.length const page = filtered.slice(0, take) const nextCursor = page.length && filtered.length > page.length ? page[page.length - 1].id : null return { runs: page, nextCursor, total: filtered.length } } /** @deprecated-compatible: returns sorted run array (no pagination). */ export function listHistory(state, jobId) { return listHistoryPage(state, jobId).runs } export function upsertJob(state, job) { const jobs = [...(state.jobs || [])] const index = jobs.findIndex((row) => row.id === job.id) if (index === -1) jobs.unshift(job) else jobs[index] = job return { ...state, jobs } } export function removeJob(state, jobId) { return { ...state, jobs: (state.jobs || []).filter((job) => job.id !== jobId), } } export function appendRun(state, run) { const isolated = applyRunIsolation(state, run) isolated.runs = pruneRuns(isolated.runs, state.settings?.historyLimit, isolated.jobs) return isolated } export function patchRun(state, runId, patch) { const runs = (state.runs || []).map((run) => ( run.id === runId ? { ...run, ...patch } : run )) let next = { ...state, runs } const updated = runs.find((run) => run.id === runId) if (updated?.sessionId) next = recordHiddenSession(next, updated.sessionId) next.runs = pruneRuns(next.runs, next.settings?.historyLimit, next.jobs) return next } export function createStore(options = {}) { const filePath = options.filePath if (!filePath) throw new Error('store filePath is required') let state = emptyState() let loaded = false let queue = Promise.resolve() async function load() { try { const raw = await readFile(filePath, 'utf8') // Reject embedded NUL — JSON.parse would throw an opaque "Unexpected token". if (raw.includes('\u0000')) { const error = new Error(`ops-cron store is corrupt (NUL byte) at ${filePath}; moved aside and starting empty`) error.code = 'STORE_CORRUPT' try { const bak = `${filePath}.corrupt.${Date.now()}.bak` await rename(filePath, bak) error.message = `${error.message} (backup: ${bak})` } catch { // best-effort quarantine; persist() below overwrites either way } console.warn(`[dsh-ops-cron] ${error.message}`) state = emptyState() loaded = true await persist() return snapshot() } const parsed = JSON.parse(raw) state = hydrate(parsed) } catch (error) { if (error && error.code === 'ENOENT') { state = emptyState() } else if (error instanceof SyntaxError) { const wrapped = new Error(`ops-cron store JSON parse failed at ${filePath}: ${error.message}`) wrapped.code = 'STORE_CORRUPT' wrapped.cause = error try { const raw = await readFile(filePath, 'utf8').catch(() => '') if (raw) await writeFile(`${filePath}.corrupt.${Date.now()}.bak`, raw, 'utf8') } catch { // ignore } console.warn(`[dsh-ops-cron] ${wrapped.message}; starting empty store`) state = emptyState() loaded = true await persist() return snapshot() } else { throw error } } loaded = true return snapshot() } async function persist() { await mkdir(dirname(filePath), { recursive: true }) const tmp = join(dirname(filePath), `.store.${randomUUID()}.tmp`) const body = `${JSON.stringify(state, null, 2)}\n` await writeFile(tmp, body, 'utf8') await rename(tmp, filePath) } function snapshot() { return cloneState(state) } function mutate(fn) { const run = async () => { if (!loaded) await load() const next = await fn(snapshot()) if (!next || typeof next !== 'object') throw new Error('store mutator must return state') state = next await persist() return snapshot() } const pending = queue.then(run, run) queue = pending.then(() => undefined, () => undefined) return pending } return { filePath, snapshot, load, mutate, async read() { if (!loaded) await load() return snapshot() }, } } function hydrateJob(row, settings) { if (!row || !row.id) return null const labels = normalizeLabels(row.labels) let watch = null let progress = null let report = null try { watch = row.watch ? normalizeWatch(row.watch) : null } catch { watch = null } try { progress = row.progress ? normalizeProgress(row.progress) : null } catch { progress = null } try { report = row.report ? normalizeReport(row.report, row.delivery) : null } catch { report = null } let persistHistory try { persistHistory = normalizePersistHistory( row.persistHistory ?? row.persist_history, settings?.historyLimit, ) } catch { persistHistory = normalizePersistHistory(undefined, settings?.historyLimit) } const hydrated = { ...row, labels, persistHistory, } if (watch) hydrated.watch = watch else delete hydrated.watch if (progress) hydrated.progress = progress else delete hydrated.progress if (report) hydrated.report = report else delete hydrated.report return hydrated } function hydrateRun(row) { if (!row || !row.id) return null return { ...row, stateEnteredAt: row.stateEnteredAt ?? row.actualAt ?? row.scheduledAt ?? null, exitedAt: row.exitedAt ?? ( row.status && !ACTIVE_RUN_STATUSES.has(row.status) ? (row.actualAt || null) : null ), outputRef: row.outputRef || null, } } function hydrate(parsed) { const base = emptyState() if (!parsed || typeof parsed !== 'object') return base const settings = normalizeSettings(parsed.settings) const jobs = Array.isArray(parsed.jobs) ? parsed.jobs.map((row) => hydrateJob(row, settings)).filter(Boolean) : [] const runs = Array.isArray(parsed.runs) ? parsed.runs.map(hydrateRun).filter(Boolean) : [] return { version: STORE_VERSION, settings, jobs, runs, hiddenSessionIds: Array.isArray(parsed.hiddenSessionIds) ? parsed.hiddenSessionIds.map(String) : [], } } export function storePath(dshHome) { return join(dshHome, 'ops-cron', 'store.json') }