jobs: queue scheduler + REST job API (phase 1c + 2)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/databases/officer_db/src/index.ts

@@ -77,6 +77,7 @@ export {
   getPipelineJob,
   updatePipelineJob,
   getPipelineJobsForUser,
+  getOldestPendingJob,
   markInterruptedJobs,
 } from './queries/pipeline-jobs';
 

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;
 }

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<string> {
+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<string> {
+  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<string> {
       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<string> {
       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<void> {
+  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<string, string>,
+    cwd: next.cwd ?? undefined,
+    config: next.config,
+  });
 }
 
 export function attachViewer(jobId: string, ws: ServerWebSocket<WSData>) {
@@ -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`. */

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<string, string>; 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'));

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);