Show cron state/labels in sidebar and push archive history.

Surface lifecycle badges and editable monitor fields in the UI, and POST terminal runs to persist_history archive endpoints on settle/tick.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-10 15:29:04 +08:00
parent b6934b20a3
commit 2e13432b0a
4 changed files with 500 additions and 13 deletions

View file

@ -18,6 +18,12 @@ import {
readProgressSnapshot,
selectJobsByWatch,
} from './monitor.js'
import {
markRunsArchived,
pruneArchivedRuns,
pushArchiveBatch,
runsNeedingArchive,
} from './archive.js'
import { decideDispatch, nextFire, validateSchedule } from './scheduler.js'
import {
createJobRecord,
@ -369,6 +375,7 @@ function runView(run) {
summary: run.summary,
reason: run.reason,
output_ref: run.outputRef || (run.sessionId ? `session:${run.sessionId}` : null),
...run.archivedAt ? { archivedAt: run.archivedAt } : {},
}
}
@ -481,7 +488,57 @@ export function createHostService(options = {}) {
await maybeDeliverIm(job, run)
await maybeMirrorResult(job, run)
await maybeNotifyWatchers(job, run)
return { job: jobView(job, state.runs), run: runView(run), decision: claimedDecision }
if (job?.persistHistory?.kind === 'archive' && job.persistHistory.endpoint) {
try {
await archiveJobRuns(jobId, { identity: null })
} catch (error) {
logger.warn?.(`[dsh-ops-cron] archive push failed for ${jobId}: ${error instanceof Error ? error.message : error}`)
}
}
return { job: jobView(job, (await snapshot()).runs), run: runView(run), decision: claimedDecision }
}
/**
* Push unarchived terminal runs for archive-policy jobs to their endpoint.
* @param {string|null} [jobId] limit to one job; null = all archive jobs
* @param {{ identity?: object|null, fetchImpl?: typeof fetch, limit?: number }} [opts]
*/
async function archiveJobRuns(jobId = null, opts = {}) {
const identity = opts.identity || null
const limit = Number.isInteger(opts.limit) && opts.limit > 0 ? Math.min(200, opts.limit) : 50
const state = await snapshot()
let jobs = listJobs(state).filter((job) => (
job?.persistHistory?.kind === 'archive' && job.persistHistory.endpoint
))
if (jobId) jobs = jobs.filter((job) => job.id === jobId)
if (identity) {
for (const job of jobs) assertCanAccessJob(job, identity)
jobs = filterJobsForIdentity(jobs, identity)
}
const results = []
for (const job of jobs) {
const live = await snapshot()
const batch = runsNeedingArchive(live.runs, job.id)
.sort((a, b) => (a.actualAt || a.scheduledAt || 0) - (b.actualAt || b.scheduledAt || 0))
.slice(0, limit)
if (!batch.length) {
results.push({ jobId: job.id, archived: 0, skipped: true })
continue
}
await pushArchiveBatch(job.persistHistory.endpoint, {
job: jobView(job, live.runs),
runs: batch.map(runView),
}, { fetchImpl: opts.fetchImpl || options.fetchImpl, now })
const mark = now()
const ids = batch.map((run) => run.id)
await withState((current) => pruneArchivedRuns(markRunsArchived(current, ids, mark), 50))
results.push({ jobId: job.id, archived: batch.length, endpoint: job.persistHistory.endpoint })
}
return {
ok: true,
archived: results.reduce((sum, row) => sum + (row.archived || 0), 0),
results,
}
}
async function createJob(input, identity = null, opts = {}) {
@ -721,6 +778,11 @@ export function createHostService(options = {}) {
if (result.run) fired.push(result)
}
await tickWatcherReports()
try {
await archiveJobRuns(null, { limit: 20 })
} catch (error) {
logger.warn?.(`[dsh-ops-cron] archive tick failed: ${error instanceof Error ? error.message : error}`)
}
return fired
} finally {
ticking = false
@ -1080,12 +1142,14 @@ export function createHostService(options = {}) {
if (path === `${API_PREFIX}/archive` && method === 'POST') {
const identity = await requireIdentity(req, write, getUdsAuth)
if (!identity) return
// Reserved: persist_history=archive:endpoint will push here later.
write(200, {
ok: true,
archived: false,
message: 'archive endpoint reserved; set persist_history=archive:<url> on jobs for future use',
const body = await readJsonBody(req).catch(() => ({}))
const jobId = typeof body.jobId === 'string' ? body.jobId.trim()
: (typeof body.task_id === 'string' ? body.task_id.trim() : '')
const result = await archiveJobRuns(jobId || null, {
identity,
limit: Number(body.limit) > 0 ? Number(body.limit) : 50,
})
write(200, { ok: true, ...result })
return
}
@ -1118,7 +1182,7 @@ export function createHostService(options = {}) {
const locale = resolveLocale(req)
const code = error && error.code
if (code === 'NOT_FOUND') return write(404, { ...apiError('not_found', locale), error: error.message || 'not_found' })
if (code === 'INVALID_CRON' || code === 'INVALID_AT' || code === 'INVALID_SCHEDULE' || code === 'INVALID_JOB' || code === 'INVALID_TIMEZONE' || code === 'INVALID_CWD' || code === 'INVALID_DELIVERY' || code === 'IM_DELIVERY_FORBIDDEN' || code === 'IM_TARGET_FORBIDDEN' || code === 'IM_JOB_MISSING_CWD' || code === 'INVALID_WATCH' || code === 'INVALID_PROGRESS' || code === 'INVALID_REPORT' || code === 'INVALID_PERSIST_HISTORY' || code === 'INVALID_PROGRESS_PATH') {
if (code === 'INVALID_CRON' || code === 'INVALID_AT' || code === 'INVALID_SCHEDULE' || code === 'INVALID_JOB' || code === 'INVALID_TIMEZONE' || code === 'INVALID_CWD' || code === 'INVALID_DELIVERY' || code === 'IM_DELIVERY_FORBIDDEN' || code === 'IM_TARGET_FORBIDDEN' || code === 'IM_JOB_MISSING_CWD' || code === 'INVALID_WATCH' || code === 'INVALID_PROGRESS' || code === 'INVALID_REPORT' || code === 'INVALID_PERSIST_HISTORY' || code === 'INVALID_PROGRESS_PATH' || code === 'INVALID_ARCHIVE' || code === 'ARCHIVE_FAILED' || code === 'ARCHIVE_UNAVAILABLE') {
return write(400, { ok: false, error: error.message, code, message: error.message })
}
if (code === 'PAYLOAD_TOO_LARGE') return write(413, apiError('payload_too_large', locale))
@ -1304,6 +1368,7 @@ export function createHostService(options = {}) {
queryJobs,
listRuns,
getProgress,
archiveJobRuns,
async listJobs(identity = null, query = null) {
const state = await snapshot()
let jobs = listJobs(state)