From 6d7d92880633aacaf2ac11e6c2ecc1bfe6e8fc25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Thu, 23 Jul 2026 13:58:42 +0000 Subject: [PATCH] jobs: queue scheduler + REST job API (phase 1c + 2) Co-Authored-By: Claude Opus 4.8 --- docs/jobs-unification.md | 9 +- src/databases/officer_db/src/index.ts | 1 + .../officer_db/src/queries/pipeline-jobs.ts | 17 ++- src/servers/api/tasks/pipeline-job-manager.ts | 60 +++++++++- src/servers/api/tasks/pipeline-jobs-routes.ts | 108 +++++++++++++++--- src/servers/hono.ts | 1 + 6 files changed, 172 insertions(+), 24 deletions(-) diff --git a/docs/jobs-unification.md b/docs/jobs-unification.md index fd22edbc..8fc54367 100644 --- a/docs/jobs-unification.md +++ b/docs/jobs-unification.md @@ -63,7 +63,12 @@ Favor power-user affordances over guardrails. See memory `sole-user-assume-compe abort poll, returns exitCode), `process-tree.ts` (shared killTree), `pipeline-job-manager` now dispatches by `mode` and finalizes script jobs by exit code. *Compiles; runtime-untested until a REST caller + restart exist.* -- [ ] 1c scheduler / queue (FIFO pending → promote on finalize; promoteNext on startup) -- [ ] 2 REST job API + WS consolidation +- [x] 1c scheduler / queue — `enqueueJob(action)` (start now / queue behind running), `promoteNext()` + on finalize + startup, `getOldestPendingJob`, `markInterruptedJobs` now running-only (pending + queue survives restart). `startJob` kept as a `enqueueJob(...,'start')` wrapper. +- [x] 2 REST job API — `POST /jobs` (create script|pipeline, action start/queue), `GET /jobs` (+`?live=1`, + now returns mode/exitCode/isLive), `GET /jobs/:id`, `GET /jobs/:id/log?offset=`, `POST /jobs/:id/stop`. + Router mounted at `/jobs` and `/pipeline-jobs`. *Needs a restart to deploy; then curl/phone-testable.* + WS consolidation still pending (old `/api/tasks/run/ws` + `/api/tasks/pipeline/ws` still live). - [ ] 3 /jobs/new page + JobDetail script branch + retire modal + header indicator - [ ] 4 push notifications diff --git a/src/databases/officer_db/src/index.ts b/src/databases/officer_db/src/index.ts index 97ff1048..d59ed752 100644 --- a/src/databases/officer_db/src/index.ts +++ b/src/databases/officer_db/src/index.ts @@ -77,6 +77,7 @@ export { getPipelineJob, updatePipelineJob, getPipelineJobsForUser, + getOldestPendingJob, markInterruptedJobs, } from './queries/pipeline-jobs'; diff --git a/src/databases/officer_db/src/queries/pipeline-jobs.ts b/src/databases/officer_db/src/queries/pipeline-jobs.ts index 4aa3638c..bc12c83d 100644 --- a/src/databases/officer_db/src/queries/pipeline-jobs.ts +++ b/src/databases/officer_db/src/queries/pipeline-jobs.ts @@ -1,4 +1,4 @@ -import { eq, and, inArray, desc } from 'drizzle-orm'; +import { eq, asc, desc } from 'drizzle-orm'; import { db } from '../db'; import { pipelineJobs } from '../schema/pipeline-jobs'; import type { PipelineJobInsert } from '../types'; @@ -26,11 +26,24 @@ export async function getPipelineJobsForUser(userId: number, limit = 50) { .limit(limit); } +// Oldest queued job across everything (single-user → global queue). Used to promote the next job. +export async function getOldestPendingJob() { + const rows = await db + .select() + .from(pipelineJobs) + .where(eq(pipelineJobs.status, 'pending')) + .orderBy(asc(pipelineJobs.createdAt)) + .limit(1); + return rows[0] ?? null; +} + +// On restart, only RUNNING jobs are orphaned (their process died) → interrupted. PENDING jobs are the +// queue backlog and must survive to be promoted after startup. export async function markInterruptedJobs() { const result = await db .update(pipelineJobs) .set({ status: 'interrupted', completedAt: new Date() }) - .where(inArray(pipelineJobs.status, ['pending', 'running'])) + .where(eq(pipelineJobs.status, 'running')) .returning({ id: pipelineJobs.id }); return result.length; } diff --git a/src/servers/api/tasks/pipeline-job-manager.ts b/src/servers/api/tasks/pipeline-job-manager.ts index 9f624c27..3bc9d8fa 100644 --- a/src/servers/api/tasks/pipeline-job-manager.ts +++ b/src/servers/api/tasks/pipeline-job-manager.ts @@ -6,8 +6,11 @@ import { getPipelineJob, updatePipelineJob, getPipelineJobsForUser, + getOldestPendingJob, markInterruptedJobs, + getUserById, } from 'officerdb'; +import { toShellUsername } from '../../data-path'; import { executePipeline } from './pipeline-executor'; import type { OutMessage } from './pipeline-executor'; import { executeScript, type ScriptEvent } from './execute-script'; @@ -79,24 +82,39 @@ type StartJobParams = { startAt?: number; }; -export async function startJob(params: StartJobParams): Promise { +export function runningCount(): number { + return liveJobs.size; +} + +// Create a job. action 'start' runs it now; 'queue' runs it only if nothing is running, else it stays +// 'pending' and gets promoted when the running job finishes. (Single user → one global queue.) +export async function enqueueJob(params: StartJobParams, action: 'start' | 'queue'): Promise<{ jobId: string; status: 'running' | 'pending' }> { const jobId = randomUUID(); const mode: JobMode = params.mode ?? 'pipeline'; - - // Create DB record + const run = action === 'start' || runningCount() === 0; await createPipelineJob({ id: jobId, userId: params.userId, taskDirName: params.taskDirName, taskName: params.taskName, mode, - status: 'running', + status: run ? 'running' : 'pending', inputs: params.inputs, cwd: params.cwd ?? null, config: params.config, - startedAt: new Date(), + startedAt: run ? new Date() : null, }); + if (run) launch(jobId, mode, params); + return { jobId, status: run ? 'running' : 'pending' }; +} +// Back-compat: the pipeline WS caller expects startJob(params) -> jobId, running immediately. +export async function startJob(params: StartJobParams): Promise { + return (await enqueueJob(params, 'start')).jobId; +} + +// Build live state + run the executor for a job whose DB row is already status 'running'. +function launch(jobId: string, mode: JobMode, params: StartJobParams) { // Create live state const job: LiveJob = { jobId, @@ -191,6 +209,7 @@ export async function startJob(params: StartJobParams): Promise { completedAt: new Date(), }).catch((err) => console.error(`[pipeline-jobs] failed to finalize ${jobId}:`, err)); liveJobs.delete(jobId); + void promoteNext(); }).catch(async (err) => { clearInterval(flushInterval); const message = err instanceof Error ? err.message : String(err); @@ -204,9 +223,36 @@ export async function startJob(params: StartJobParams): Promise { completedAt: new Date(), }).catch((e) => console.error(`[pipeline-jobs] failed to finalize ${jobId}:`, e)); liveJobs.delete(jobId); + void promoteNext(); }); +} - return jobId; +// When a job finishes (and nothing else is running), promote the oldest queued job. Also called on +// startup to resume a backlog. Rebuilds the user context from userId since there's no request here. +async function promoteNext(): Promise { + if (runningCount() > 0) return; + const next = await getOldestPendingJob(); + if (!next) return; + const user = await getUserById(next.userId); + if (!user) { + await updatePipelineJob(next.id, { status: 'failed', error: 'user not found', completedAt: new Date() }).catch(() => {}); + return promoteNext(); + } + await updatePipelineJob(next.id, { status: 'running', startedAt: new Date() }); + const nextMode = (next.mode as JobMode) ?? 'pipeline'; + launch(next.id, nextMode, { + userId: next.userId, + email: user.email, + username: toShellUsername(user.username ?? '', user.email), + role: user.role ?? '', + mode: nextMode, + sandboxed: (user.role ?? '') !== 'Super Admin', + taskDirName: next.taskDirName, + taskName: next.taskName, + inputs: next.inputs as Record, + cwd: next.cwd ?? undefined, + config: next.config, + }); } export function attachViewer(jobId: string, ws: ServerWebSocket) { @@ -271,6 +317,8 @@ export async function cleanupOnStartup() { if (count > 0) { console.log(`[pipeline-jobs] marked ${count} interrupted jobs on startup`); } + // Resume any queued backlog that survived the restart. + await promoteNext(); } /** Extract minimal progress info from events for DB storage. Script events fall through to `prev`. */ diff --git a/src/servers/api/tasks/pipeline-jobs-routes.ts b/src/servers/api/tasks/pipeline-jobs-routes.ts index c856f31e..6c8a2e95 100644 --- a/src/servers/api/tasks/pipeline-jobs-routes.ts +++ b/src/servers/api/tasks/pipeline-jobs-routes.ts @@ -1,26 +1,106 @@ +import { stat, open } from 'node:fs/promises'; import { createRouter } from '../../create-router'; +import * as errors from '../../custom-errors'; import { getPipelineJobsForUser, getPipelineJob } from 'officerdb'; +import { getTaskByDirName } from './task-files'; +import * as jobManager from './pipeline-job-manager'; +import { jobLogPath } from './execute-script'; export const pipelineJobsRouter = createRouter(); -// GET /pipeline-jobs — list user's pipeline jobs +// GET / — list the user's jobs (optionally only the live ones). pipelineJobsRouter.get('/', async (c) => { const user = c.get('user'); - const jobs = await getPipelineJobsForUser(user.id); - return c.json(jobs.map((j) => ({ - id: j.id, - taskDirName: j.taskDirName, - taskName: j.taskName, - status: j.status, - totalCost: j.totalCost, - createdAt: j.createdAt, - startedAt: j.startedAt, - completedAt: j.completedAt, - error: j.error, - }))); + const liveOnly = c.req.query('live') === '1'; + const jobs = await jobManager.getJobsForUser(user.id); + return c.json( + jobs + .filter((j) => (liveOnly ? j.isLive : true)) + .map((j) => ({ + id: j.id, + mode: j.mode, + taskDirName: j.taskDirName, + taskName: j.taskName, + status: j.status, + isLive: j.isLive, + exitCode: j.exitCode, + totalCost: j.totalCost, + createdAt: j.createdAt, + startedAt: j.startedAt, + completedAt: j.completedAt, + error: j.error, + })), + ); }); -// GET /pipeline-jobs/:id — single job detail +// POST / — create a job (script or pipeline). action 'start' runs now; 'queue' waits behind a running +// job. Returns { jobId, status }. This is the REST creation path the phone / unattended runs use. +pipelineJobsRouter.post('/', async (c) => { + const user = c.get('user'); + const body = await c.req.json<{ taskDirName: string; inputs?: Record; cwd?: string; action?: 'start' | 'queue' }>(); + if (!body.taskDirName) throw errors.BAD_REQUEST('taskDirName is required'); + + const task = await getTaskByDirName(body.taskDirName); + if (!task) throw errors.NOT_FOUND(`Task not found: ${body.taskDirName}`); + const mode = task.mode as 'pipeline' | 'script' | 'agentic'; + if (mode !== 'script' && mode !== 'pipeline') throw errors.BAD_REQUEST(`Task mode '${mode}' can't run as a job yet`); + + const { jobId, status } = await jobManager.enqueueJob( + { + userId: user.id, + email: user.email, + username: user.username ?? '', + role: user.role ?? '', + mode, + sandboxed: (user.role ?? '') !== 'Super Admin', + taskDirName: body.taskDirName, + taskName: task.name, + inputs: body.inputs ?? {}, + cwd: body.cwd, + config: task.config, + }, + body.action === 'queue' ? 'queue' : 'start', + ); + + return c.json({ jobId, status }); +}); + +// GET /:id/log?offset= — tail the persisted output log (script jobs). Returns text from `offset`. +pipelineJobsRouter.get('/:id/log', async (c) => { + const user = c.get('user'); + const job = await getPipelineJob(c.req.param('id')); + if (!job || job.userId !== user.id) return c.json({ error: 'Not found' }, 404); + + const offset = Math.max(0, parseInt(c.req.query('offset') ?? '0', 10) || 0); + const path = jobLogPath(job.id); + let size = 0; + try { + size = (await stat(path)).size; + } catch { + return c.json({ text: '', offset, size: 0 }); // no log yet + } + if (offset >= size) return c.json({ text: '', offset: size, size }); + + const fh = await open(path, 'r'); + try { + const buf = Buffer.alloc(size - offset); + await fh.read(buf, 0, buf.length, offset); + return c.json({ text: buf.toString('utf8'), offset: size, size }); + } finally { + await fh.close(); + } +}); + +// POST /:id/stop — request cancellation (cooperative abort). +pipelineJobsRouter.post('/:id/stop', async (c) => { + const user = c.get('user'); + const job = await getPipelineJob(c.req.param('id')); + if (!job || job.userId !== user.id) return c.json({ error: 'Not found' }, 404); + const stopped = jobManager.stopJob(job.id); + return c.json({ ok: true, wasLive: stopped }); +}); + +// GET /:id — single job detail (full row). pipelineJobsRouter.get('/:id', async (c) => { const user = c.get('user'); const job = await getPipelineJob(c.req.param('id')); diff --git a/src/servers/hono.ts b/src/servers/hono.ts index cfee0dba..ac8b43f2 100644 --- a/src/servers/hono.ts +++ b/src/servers/hono.ts @@ -104,6 +104,7 @@ protectedRouter.route('/apps', appsRouter); protectedRouter.route('/bug-report', bugReportRouter); protectedRouter.route('/saved-sessions', savedSessionsRouter); protectedRouter.route('/pipeline-jobs', pipelineJobsRouter); +protectedRouter.route('/jobs', pipelineJobsRouter); // unified jobs API (script + pipeline); /pipeline-jobs kept for the existing UI desktopRouter.use(superAdminMiddleware); protectedRouter.route('/desktop', desktopRouter); protectedRouter.route('/', piRestRouter);