bluebut#6stalled

/home/philip/Documents/GitHub/bluebut-epic-6
fix/epic-6

title:	Skyvern runtime cutover — flip queue-worker dispatcher (supervised, atomic)
state:	open
author:	root
labels:	epic, priority:high
comments:	1
assignees:	root
--
Replaces the finished epic #19 (closed). The Skyvern migration's **foundation is done + on main**; only ONE thing remains — the runtime dispatcher cutover — and it's atomic + supervised. This issue tracks just that.

## Already done (in main — not part of this issue)
- ✅ `updateRunStatus` single-writer funnel (`server/jobs/queue.ts`, 42d197e) — `markFailed` routes through it; `okForStatus`/`statusPatch` unit-tested.
- ✅ Vocabulary rename job→run / batch→workflow_run live in prod (`0047`), all call-sites repointed; `0050` = contract step.
- ✅ God-file decompose (`runtime.mjs` 1842→123 → `server/http/runtime/*`), catalog (64 verbs), Hono error middleware, failure_classifier, worker already `execute_safe`.

## What remains — ONE atomic, SUPERVISED cutover (do NOT merge piecemeal)
The queue executor (`server/jobs/worker.ts` + `/api/queue`) coexists dormant beside the live path (`/exec` inline + `setInterval` scheduler). Cutting over is **runtime-coupled** — these must flip together on one proven drive:

- [ ] Repoint UI job client (`ui/src/api/jobs.ts`) off `/exec/stream` → `POST /api/queue` + `GET /api/queue/:id/events` (SSE). *Alone this hangs every job if the worker is off.*
- [ ] `BLUEBUT_QUEUE_WORKER=1` in prod env — worker becomes the dispatcher.
- [ ] Delete the now-duplicate live dispatchers: inline `spawn` in `bridge.ts` `/exec`, the `setInterval` scheduler in `runtime.mjs`. *Leaving them = double-dispatch, the inv #23 ~1400-job runaway class.*
- [ ] Contract: migrate the read-side off `run.ok` (UI reads it — `AccountJobs.tsx:38`, `taxonomy.ts:78`, `WorkflowView`, SSE `jobs.ts:45`) → then drop the column.

## Why it can't be a headless merge
`main` auto-deploys to prod. Merging any slice alone = fleet outage (double-dispatch) or dead dashboard (jobs enqueue, nothing drains) or wrong status data (ok flip). Proven in #19 thread. **Finish = one `npm run dev:all` session:** land each step, prove against a real drive + real DB rows (inv #24), then it's genuinely done.

Foundation: `docs/SKYVERN-PATTERNS.md` · gap: `docs/skyvern-gap-analysis.md`.

---
_Migrated from GitHub https://github.com/devdashco/bluebut/issues/295 (#295)_

=== structure ===
server/http/runtime:
auto-broadcast.mjs
broadcast.mjs
health.mjs
instances.mjs
job-registry.mjs
login.mjs
recordings.mjs
run-core.mjs
scheduler.mjs
spawn-env.d.mts
spawn-env.mjs
traffic.mjs
workflow.mjs

server/jobs:
catalog-integrity.test.mjs
catalog.test.mjs
catalog.ts
enqueue-broadcast.ts
enqueue-validation.test.mjs
job-id.ts
queue-status.test.mjs
queue.ts
verbs
worker.test.mjs
worker.ts

ui/src/api:
accounts.ts
catalog.ts
config-workflow.ts
_core.ts
db-types.gen.ts
entities.ts
har-traces.ts
index.ts
instances.ts
jobs.ts
listings.ts
proxy.ts
store.ts
workflow-runs.ts
workflows.ts
=== files exist? ===
-- server/jobs/worker.ts
OK
-- server/jobs/queue.ts
OK
-- ui/src/api/jobs.ts
OK
-- server/http/runtime/bridge.ts
MISSING
-- runtime.mjs
MISSING
-- ui/src/pages/AccountJobs.tsx
MISSING
-- ui/src/lib/taxonomy.ts
OK

=== /exec inline spawn + bridge ===
server/app.ts
server/http/runtime/job-registry.mjs
server/http/runtime/scheduler.mjs
server/http/routes/bridge.ts
server/jobs/queue.ts
server/http/index.mjs
--- setInterval scheduler ---
server/http/runtime.mjs
server/http/runtime/traffic.mjs
server/http/runtime/health.mjs
server/http/runtime/recordings.mjs
server/http/runtime/run-core.mjs
server/http/runtime/auto-broadcast.mjs
server/http/runtime/scheduler.mjs
server/http/routes/capture.ts
server/jobs/worker.ts
server/http/index.mjs
=== AccountJobs ===
ui/src/components/WorkflowView.tsx
ui/src/components/AccountJobs.tsx
=== run.ok read sites ===
ui/src/api/jobs.ts
ui/src/lib/taxonomy.ts
ui/src/api/accounts.ts
ui/src/api/workflow-runs.ts
ui/src/components/ProxyPanel/ReadView.tsx
ui/src/pages/broadcast/BundleBroadcast.tsx
ui/src/pages/Insights/tabs.tsx
ui/src/api/har-traces.ts
ui/src/api/jobs.ts
ui/src/api/workflows.ts
ui/src/state/jobs.tsx
ui/src/components/AccountActionsMenu.tsx
ui/src/api/proxy.ts
ui/src/api/config-workflow.ts
ui/src/api/instances.ts
ui/src/api/catalog.ts
ui/src/components/WorkflowView.tsx
ui/src/components/InfraNodeDetail.tsx
ui/src/components/ProxyPanel.tsx
ui/src/components/InfraStatusMenu.tsx
ui/src/components/AccountJobs.tsx
ui/src/components/CdpGate.tsx

###### ui/src/api/jobs.ts ######
     1	import { API, bridgeAuth, dbDelete } from "./_core";
     2	
     3	export interface RunResult { ok: boolean; [k: string]: unknown; }
     4	
     5	function extractAccount(args: string[]): string | undefined {
     6	  const idx = args.indexOf("--account");
     7	  return idx >= 0 ? args[idx + 1] : undefined;
     8	}
     9	
    10	// Batch envelope a Job belongs to (taxonomy: Batch → Job). Minted in the UI by
    11	// a bulk form (newBatchId), threaded UI → bridge → runner so the whole run is
    12	// one sliceable unit. Flattened into the four batch* opts on runCli*/engage.
    13	export interface BatchRef { id: string; label?: string; size?: number; index?: number }
    14	
    15	// runCli is a thin wrapper around runCliStream. Every adapter call streams
    16	// progress events through the backend — we just log each event to the
    17	// structured console (HyperDX picks them up) so callers that don't want a
    18	// live UI updater still get observability for free. Callers that DO want
    19	// to render progress should use runCliStream directly with a custom onLine.
    20	export async function runCli(
    21	  site: string,
    22	  cmd: string,
    23	  args: string[],
    24	  // Number = timeout (the long-standing shape). Object form additionally threads
    25	  // the caller's `jobId`/`batch` to the runner (BLUEBUT_JOB_ID/BLUEBUT_BATCH_ID).
    26	  // WITHOUT it the backend mints its own job_id, so the durable `job` row and the
    27	  // FB traffic land under an id the UI job never shows: "⧉ copy log" then yields
    28	  // a HyperDX query that finds the UI's steps and none of the adapter's trail.
    29	  // Anything running inside a Job MUST pass jobId.
    30	  // `signal` lets the caller ABORT the run — aborting cancels the SSE fetch, and
    31	  // the backend SIGKILLs the runner child on request-close, so the job stops for real.
    32	  opts: number | { timeout?: number; jobId?: string; batch?: BatchRef; signal?: AbortSignal } = 180,
    33	): Promise<RunResult> {
    34	  const { timeout = 180, jobId, batch, signal } = typeof opts === "number" ? { timeout: opts } as { timeout: number; jobId?: string; batch?: BatchRef; signal?: AbortSignal } : opts;
    35	  const account = extractAccount(args);
    36	  const tag = `${site}/${cmd}`;
    37	  return runCliStream(site, cmd, args, {
    38	    timeout, jobId, batch, signal,
    39	    onLine: (text) => {
    40	      // Adapters emit one JSON object per stderr line; non-JSON lines (rare
    41	      // — opencli update notices etc.) get logged as plain text. Either way
    42	      // the structured logger gets a per-event row, so HyperDX timelines
    43	      // show each step instead of one giant "completed" entry.
    44	      try {
    45	        const ev = JSON.parse(text);
    46	        console.info(`[cli:event] ${tag}`, { site, cmd, account, ...ev });
    47	      } catch {
    48	        console.debug(`[cli:line] ${tag} ${text}`, { site, cmd, account, line: text });
    49	      }
    50	    },
    51	  });
    52	}
    53	
    54	// Streaming variant of runCli. POSTs to /exec/stream and yields each
    55	// stderr line the adapter writes (Server-Sent Events: `data: <line>\n\n`).
    56	// The final line is `data: [RESULT] {ok, result, ...}` carrying the JSON
    57	// result; on hard error it's `data: [ERROR] <msg>`. Caller gets one
    58	// `onLine(text)` callback per stderr line and the awaited promise
    59	// resolves to the parsed final result.
    60	//
    61	// Lines that look like JSON are passed through unchanged — the consumer
    62	// can JSON.parse to get structured progress (e.g. ingest-groups emits
    63	// `{phase:"group", ok:true, index, total, fb_group_id, name, ...}` per
    64	// committed group, which the bluebut Jobs panel renders as ticks).
    65	export async function runCliStream(
    66	  site: string,
    67	  cmd: string,
    68	  args: string[],
    69	  opts: {
    70	    timeout?: number;
    71	    onLine: (text: string) => void;
    72	    signal?: AbortSignal;
    73	    // When set, the backend forwards it to the runner as BLUEBUT_JOB_ID so the
    74	    // job's FB traffic (chromebox-traffic stream, by account) is sliceable by
    75	    // job_id. Optional — account + cmd + time still brackets every run.
    76	    jobId?: string;
    77	    // Batch this job belongs to — forwarded to the backend job record (Jobs-page
    78	    // grouping) and to the runner as BLUEBUT_BATCH_ID (traffic sliceable by batch).
    79	    batch?: BatchRef;
    80	  },
    81	): Promise<RunResult> {
    82	  const t0 = performance.now();
    83	  const account = extractAccount(args);
    84	  const tag = `${site}/${cmd}`;
    85	  console.info(`[cli:stream] ${tag} started`, { site, cmd, account });
    86	
    87	  try {
    88	    const r = await fetch(`${API}/exec/stream`, {
    89	      method: "POST",
    90	      headers: { "Content-Type": "application/json", ...bridgeAuth() },
    91	      body: JSON.stringify({
    92	        site, cmd, args: [...args.filter(a => a != null)],
    93	        format: "json",
    94	        timeout: opts.timeout ?? 180,
    95	        ...(opts.jobId ? { job_id: opts.jobId } : {}),
    96	        ...(opts.batch ? {
    97	          batch_id: opts.batch.id,
    98	          batch_label: opts.batch.label,
    99	          batch_size: opts.batch.size,
   100	          batch_index: opts.batch.index,
   101	        } : {}),
   102	      }),
   103	      signal: opts.signal,
   104	    });
   105	    if (!r.ok || !r.body) throw new Error(`/exec/stream HTTP ${r.status}`);
   106	
   107	    const reader = r.body.getReader();
   108	    const decoder = new TextDecoder();
   109	    let buf = "";
   110	    let finalResult: RunResult | null = null;
   111	    let errorMsg: string | null = null;
   112	
   113	    while (true) {
   114	      const { value, done } = await reader.read();
   115	      if (done) break;
   116	      buf += decoder.decode(value, { stream: true });
   117	      let idx;
   118	      while ((idx = buf.indexOf("\n\n")) !== -1) {
   119	        const block = buf.slice(0, idx);
   120	        buf = buf.slice(idx + 2);

1	        buf = buf.slice(idx + 2);
     2	        const text = block.split("\n")
     3	          .filter(l => l.startsWith("data:"))
     4	          .map(l => l.slice(5).replace(/^ /, ""))
     5	          .join("\n");
     6	        if (!text) continue;
     7	        if (text.startsWith("[RESULT] ")) {
     8	          try {
     9	            const payload = JSON.parse(text.slice(9));
    10	            finalResult = (payload.result ?? payload) as RunResult;
    11	          } catch {
    12	            finalResult = { ok: false, error: "malformed [RESULT]" } as RunResult;
    13	          }
    14	        } else if (text.startsWith("[ERROR] ")) {
    15	          errorMsg = text.slice(8);
    16	        } else {
    17	          opts.onLine(text);
    18	        }
    19	      }
    20	    }
    21	
    22	    if (errorMsg && !finalResult) throw new Error(errorMsg);
    23	    const result = finalResult ?? ({ ok: false, error: "stream ended without [RESULT]" } as RunResult);
    24	    const ms = Math.round(performance.now() - t0);
    25	    if (result.ok) {
    26	      console.info(`[cli:stream] ${tag} ok ${ms}ms`, { site, cmd, account, duration_ms: ms, ok: true });
    27	    } else {
    28	      const reason = (result.aborted_reason as string) ?? (result.error as string) ?? errorMsg ?? "unknown";
    29	      console.warn(`[cli:stream] ${tag} FAIL ${ms}ms: ${reason}`, { site, cmd, account, duration_ms: ms, ok: false, reason });
    30	    }
    31	    return result;
    32	  } catch (e) {
    33	    const ms = Math.round(performance.now() - t0);
    34	    // A failed fetch to the backend surfaces as `TypeError: Load failed` (Safari)
    35	    // / `Failed to fetch` (Chrome) / an AbortError — none of which mention the
    36	    // bridge, so it reads like an adapter bug. It almost always means: the local
    37	    // bridge isn't running (`npm run backend`), or the dashboard is being served
    38	    // from prod (bluebut.blpk.cc) which has NO bridge and can't reach localhost.
    39	    // Rewrite it into something the operator can act on.
    40	    const isAbort = e instanceof DOMException && e.name === "AbortError";
    41	    const isFetchFail = e instanceof TypeError && /load failed|failed to fetch|networkerror/i.test(e.message);
    42	    let err = e;
    43	    if (isAbort) {
    44	      err = new Error(`Cancelled (${tag})`);
    45	    } else if (isFetchFail) {
    46	      const onProd = typeof location !== "undefined" && location.hostname !== "localhost" && location.hostname !== "127.0.0.1";
    47	      err = new Error(
    48	        onProd
    49	          ? `Bridge unreachable — the hosted dashboard is view-only. Run the dashboard locally (npm run dev + npm run backend) to execute FB actions.`
    50	          : `Bridge unreachable at ${API} — start it with \`npm run backend\` in the repo root, then retry.`,
    51	      );
    52	    }
    53	    console.error(`[cli:stream] ${tag} ERROR ${ms}ms: ${err}`, { site, cmd, account, duration_ms: ms, error: String(e), classified: isFetchFail ? "bridge_unreachable" : isAbort ? "cancelled" : "unknown" });
    54	    throw err;
    55	  }
    56	}
    57	
    58	// POST /login/stream → the ONE credential login verb. LOAD-FIRST: the backend
    59	// replays the stored jar; if it's already authenticated it returns immediately
    60	// (did_login:false, never touches credentials). Otherwise it runs the login flow,
    61	// streaming captcha/2FA progress as SSE and writing a single `job` ledger row
    62	// (shared job_id). Wire format: `data:` lines + a final `[RESULT]`.
    63	export interface LoginOpts {
    64	  account: string;
    65	  geo?: string; locale?: string; tzone?: string;  // omit → bridge resolves from the keyvault row
    66	  jobId?: string;                                  // UI Job id → threaded to the ledger + HyperDX
    67	  onLine: (text: string) => void;
    68	  signal?: AbortSignal;
    69	}
    70	export async function runLoginStream(o: LoginOpts): Promise<RunResult> {
    71	  const r = await fetch(`${API}/login/stream`, {
    72	    method: "POST",
    73	    headers: { "Content-Type": "application/json", ...bridgeAuth() },
    74	    body: JSON.stringify({ account: o.account, geo: o.geo, locale: o.locale, tzone: o.tzone, job_id: o.jobId }),
    75	    signal: o.signal,
    76	  });
    77	  if (!r.ok || !r.body) {
    78	    let msg = `/login/stream HTTP ${r.status}`;
    79	    try { const j = await r.json(); if (j?.error) msg = j.error; } catch { /* non-json */ }
    80	    throw new Error(msg);
    81	  }
    82	  const reader = r.body.getReader();
    83	  const decoder = new TextDecoder();
    84	  let buf = "", finalResult: RunResult | null = null, errorMsg: string | null = null;
    85	  while (true) {
    86	    const { value, done } = await reader.read();
    87	    if (done) break;
    88	    buf += decoder.decode(value, { stream: true });
    89	    let idx;
    90	    while ((idx = buf.indexOf("\n\n")) !== -1) {
    91	      const block = buf.slice(0, idx); buf = buf.slice(idx + 2);
    92	      const text = block.split("\n").filter(l => l.startsWith("data:")).map(l => l.slice(5).replace(/^ /, "")).join("\n");
    93	      if (!text) continue;
    94	      if (text.startsWith("[RESULT] ")) {
    95	        try { const p = JSON.parse(text.slice(9)); finalResult = (p.result ?? p) as RunResult; }
    96	        catch { finalResult = { ok: false, error: "malformed [RESULT]" } as RunResult; }
    97	      } else if (text.startsWith("[ERROR] ")) { errorMsg = text.slice(8); o.onLine(text); }
    98	      else o.onLine(text);
    99	    }
   100	  }
   101	  if (errorMsg && !finalResult) throw new Error(errorMsg);
###### queue API routes ######
server/app.ts:32:// Mounted at /api/queue/* — the durable queue API. Additive: it rides ALONGSIDE the
server/app.ts:35:// any /api/queue/* request here (see server/http/index.mjs); the UI migrates onto
server/app.ts:38:  const app = new Hono().basePath('/api/queue')
server/app.ts:84:  base: '/api/queue',
server/app.ts:88:// The node:http listener the legacy shell dispatches /api/queue/* to (twin of
server/jobs/queue.ts:59:// — worker.ts, the /api/queue routes, and the tests are untouched by the column rename.
server/http/index.mjs:64:// The durable-queue API (/api/queue/*) + its worker. Mounted ALONGSIDE this frozen
server/http/index.mjs:183:    { m: "GET", p: "/jobs", auth: true, r: "live job/queue snapshot" },
server/http/index.mjs:300:  // + /api/queue into a single doc (the cloakbox orchestrator is a separate service
server/http/index.mjs:309:  // /api/queue/* → the durable-queue Hono sub-app (owns its own auth + CORS).
server/http/index.mjs:312:  if (req.url.startsWith("/api/queue")) { queueListener(req, res); return; }
server/http/index.mjs:413:  // Durable-queue worker (drains /api/queue POSTs → claim_job → runner). It is a
server/http/index.mjs:417:  // POSTs /api/queue/jobs, and claim errors are caught, so an unset RPC just logs.
server/http/index.mjs:423:  if (!queueWorkerOn) console.log('[backend] queue worker OFF (set BLUEBUT_QUEUE_WORKER=1 to drain /api/queue jobs)');

###### server/app.ts ######
     1	// app — the API root. Mounts one thin route group per resource and owns the
     2	// cross-cutting middleware (auth, error boundary). This is the Hono replacement
     3	// for the 3591-line http.createServer if-chain; each resource is a self-contained
     4	// group, not a slice of one dispatcher.
     5	//
     6	// MIGRATION MAP — the backend's 16 resources, mounted here as they land. `jobs` is
     7	// live (queue-backed). The rest are ported verb-by-verb from server/http/*, moving
     8	// their domain logic into adapters/_lib/* and leaving only validate→call→validate
     9	// in the route. Until a group is ported it stays served by the frozen server/http shell.
    10	//
    11	//   ✅ /jobs        routes/jobs.mjs      (was /jobs + /exec/stream + in-mem Map)
    12	//   ⏳ /accounts    routes/accounts.mjs  (+ /proxy)   ← port server/http/routes/accounts.mjs
    13	//   ⏳ /broadcast   routes/broadcast.mjs  → _lib/broadcast-run.js (kill inline bcast*)
    14	//   ⏳ /workflow    routes/workflow.mjs   → _lib/workflow-run.js
    15	//   ⏳ /listings    routes/listings.mjs   ← port server/http/listings.mjs
    16	//   ⏳ /messenger   routes/messenger.mjs  (move inline CAT/socket/receipt SQL to _lib)
    17	//   ⏳ /login       folds into /jobs (login is a verb)
    18	//   ⏳ /capture     routes/capture.mjs    (/har + /traffic + /wiretap + /media)
    19	//   ⏳ /cloakbox    routes/cloakbox.mjs   ← port server/http/routes/cloakbox.mjs
    20	//   ⏳ /coolify /llm /schedule /settings /v1
    21	import { Hono } from 'hono'
    22	import { compress } from 'hono/compress'
    23	import { cors } from 'hono/cors'
    24	import { getRequestListener } from '@hono/node-server'
    25	import { ZodError } from 'zod'
    26	import { ValidationError } from '../adapters/_lib/errors.ts'
    27	import { auth } from './middleware/auth.ts'
    28	import { jobs } from './routes/jobs.ts'
    29	import { EnqueueRequest } from './contract/job.ts'
    30	import { openApiFromHono } from './http/openapi-from-hono.ts'
    31	
    32	// Mounted at /api/queue/* — the durable queue API. Additive: it rides ALONGSIDE the
    33	// frozen legacy shell (which keeps serving the in-mem /jobs snapshot + /exec/stream)
    34	// under its own prefix, the twin of storeListener's /api/store. The shell dispatches
    35	// any /api/queue/* request here (see server/http/index.mjs); the UI migrates onto
    36	// it verb-by-verb. Nothing legacy changes shape.
    37	export function createApp() {
    38	  const app = new Hono().basePath('/api/queue')
    39	
    40	  // Browser CORS (the legacy shell injects CORS on its own if-chain; this mounted
    41	  // sub-app is dispatched before that, so it owns CORS itself — mirrors store.ts).
    42	  app.use('*', cors())
    43	
    44	  // gzip every response. JSON compresses ~8-10x; the front nginx also gzips the
    45	  // proxied path (belt + braces), but any direct/localhost consumer of this app
    46	  // gets compression here without depending on the proxy.
    47	  app.use('*', compress())
    48	
    49	  // One global error boundary — no per-route try/catch. Bad args → 422, unknown
    50	  // verb → 400, everything else → a clean 500 JSON. Classify by TYPE, never by
    51	  // regexing err.message (issue #27 / inv #7): getVerb throws a typed ValidationError.
    52	  app.onError((err, c) => {
    53	    if (err instanceof ZodError) return c.json({ error: 'invalid_input', issues: err.issues }, 422)
    54	    if (err instanceof ValidationError) return c.json({ error: 'unknown_verb', message: err.message }, 400)
    55	    console.error('[api] 500', err)
    56	    return c.json({ error: 'internal', message: err.message }, 500)
    57	  })
    58	
    59	  app.get('/health', (c) => c.json({ ok: true, service: 'bluebut' }))
    60	
    61	  app.use('/jobs/*', auth())
    62	  app.route('/jobs', jobs)
    63	
    64	  // Live spec, auto-derived from the routes above — powers the `bluebut-queue`
    65	  // OpenAPI->MCP wrapper. Open (only /jobs/* is auth-gated); operations need the
    66	  // Bearer token. Registered last so app.routes is fully populated when it runs.
    67	  //
    68	  // Deliberately open (issue #163 audit): the `@ivotoby/openapi-mcp-server` bridge
    69	  // bare-`fetch()`es this spec with no headers (v1.16.1), so gating it behind the
    70	  // Bearer token would break `bluebut-queue`, and a `?token=` param would leak the
    71	  // token into access logs. The spec emits NO secrets (no examples/response schemas)
    72	  // — recon only. The credential-bearing takeover surface is the ORCHESTRATOR, which
    73	  // is hardened separately (fail-closed key + no anonymous /docs+/redoc explorers).
    74	  app.get('/openapi.json', (c) => c.json(openApiFromHono(app, QUEUE_OPENAPI_OPTS)))
    75	
    76	  return app
    77	}
    78	
    79	// Spec opts lifted to module scope so the unified root /openapi.json can rebuild
    80	// the same doc (one spec per service). Relative paths under servers[0].url.
    81	const QUEUE_OPENAPI_OPTS = {
    82	  title: 'Bluebut Queue API',
    83	  description: 'The durable job queue (enqueue · list · status stream) over HTTP. /jobs operations require Authorization: Bearer <token>; /openapi.json is open.',
    84	  base: '/api/queue',
    85	  bodies: { 'POST /jobs': EnqueueRequest },
    86	}
    87	
    88	// The node:http listener the legacy shell dispatches /api/queue/* to (twin of
    89	// storeListener). Built once at import; the standalone server/index.ts uses
    90	// createApp().fetch directly instead.
    91	const queueApp = createApp()
    92	export const queueListener = getRequestListener(queueApp.fetch)
    93	
    94	// The live queue spec object — exported for the unified root /openapi.json merge.
    95	export const queueOpenApiSpec = () => openApiFromHono(queueApp, QUEUE_OPENAPI_OPTS)

###### routes/jobs.ts (queue routes) ######
1:// routes/jobs — the jobs resource. Thin: validate against the contract, call the
4:import { streamSSE } from 'hono/streaming'
8:import { enqueue, getJob, listJobs, isLedgerTerminal } from '../jobs/queue.ts'
13:jobs.get('/verbs', (c) => c.json({ verbs: catalogSummary() }))
19:jobs.post('/', zValidator('json', EnqueueRequest), async (c) => {
24:jobs.get('/', async (c) => {
30:jobs.get('/:id', async (c) => {
31:  const job = await getJob(c.req.param('id'))
35:// Live status stream. Slice 1 polls the ledger until terminal; step-level events
37:jobs.get('/:id/events', (c) => streamSSE(c, async (stream) => {
40:  for (let i = 0; i < 600 && !stream.aborted; i += 1) {
41:    const job = await getJob(id)
42:    if (!job) { await stream.writeSSE({ event: 'error', data: JSON.stringify({ error: 'not_found' }) }); return }
45:      await stream.writeSSE({ event: 'status', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok }) })
48:    await stream.sleep(1000)

###### bridge.ts /exec inline spawn ######
5:// the two dispatch phases the shell has:
16://     Hono Response. The job-spawn catch-all (/login/stream, /exec, /exec/stream —
17://     live child_process spawn + SSE, HIGHEST RISK) stays a RAW (req,res) function
18://     bypassing Hono entirely: it manually writes/streams to the real
27://     spawn reader.
28:import { Readable } from "node:stream";
30:import { spawn } from "node:child_process";
119:// ── POST-AUTH: the Hono sub-app for every non-streaming branch ──────────────────
189:// GET /recording — streams a job's NDJSON tab-recording. Native Hono (no manual
191:// officially-supported streaming path — node-server drains it chunk-by-chunk via
200:    rs.on("error", () => { /* stream errors just truncate the response, matching the old rs.pipe(res) behaviour */ });
299:// which targets (+ unreachable/capped), NO spawn, NO browser. The operator's dry run.
406:// heard of) falls to the raw spawn catch-all below, mirroring the original
421:// ── FINAL job-spawn catch-all — /login/stream + the site/cmd adapter spawn. RAW
423://    dispatches. A request with no site/cmd gets a 400 here (the chain's terminal
433:    if (path0 === "/login/stream") { handleLogin(req, res, b); return; }
435:    // Header-only auth on the process-spawn path (issue #20): /exec runs an adapter
437:    // weak to authorize spawning code — require the Authorization header here.
438:    if ((path0 === "/exec" || path0 === "/exec/stream") && !authed(req, { allowQuery: false })) {
440:      res.end(JSON.stringify({ ok: false, error: "unauthorized — /exec requires an Authorization header" }));
470:    const isStream = path0 === "/exec/stream";
484:    if (isStream) res.writeHead(200, { ...CORS, "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive" });
489:    let child: ReturnType<typeof spawn> | null = null;
501:    // The actual run — spawned only when it's this job's turn on the account
505:      child = spawn(NODE, cliArgs, { env: childEnv(acct, jobTraceId, batch_id, site) });
544:      // POST /exec — one-shot JSON. Still tee stderr into the job log line-by-line
545:      // so even non-streamed callers get a step trail (→ Workflow view + diagnostics).
580:// ── POST-AUTH entry point — dispatches exactly the original if-chain order: the
581://    Hono sub-app for every route it owns, else the raw spawn catch-all. Always

###### scheduler.mjs setInterval ######
3:// tick fns and the SCHED_ENABLED gate live here; the BOOT triggers (the setInterval /
7:import { spawn } from "node:child_process";
15:import { RUNNER, NODE, adapterPath, childEnv, parseLoose, envFile } from "./spawn-env.mjs";
22:// The bridge is the always-on process that already spawns runner jobs and holds
26:// (newJob → account lock → childEnv spawn), and stamp run-state back to the DB.
38:// batch id), cleared when that batch drains (workflowRunExec finally). A deferred sweep
63:// directly) so a device-less / dark account is never dispatched.
74:// same newJob + withAccountLock + childEnv spawn, minus the SSE response. The
90:      const child = spawn(NODE, cliArgs, { env: childEnv(acct, jobId, s.schedule_id || null, site) });
184:// Atomic compare-and-swap claim on a schedule's interval slot (issue #43.F3). The

###### run.ok reads: AccountJobs ######
26:const COLOR: Record<Phase, string> = { running: "var(--color-success)", queued: "var(--color-warning)", ok: "var(--color-info-fg)", failed: "var(--color-danger-fg)" };
27:const PHASE_LABEL: Record<Phase, string> = { running: "Running", queued: "Queued", ok: "Done", failed: "Failed" };
30:// funnel writes), NOT the retiring `run.ok` boolean (#295: read-side off run.ok).
33:function phaseFromStatus(status: string | null | undefined, ok: boolean | null | undefined): Phase {
43:  return j.ok === false ? "failed" : "ok";
123:        phase: live ? phaseOfBridge(live) : phaseFromStatus(j.status, j.ok),
###### taxonomy.ts ######
68:// The Jobs page holds BridgeJobs (status queued|running|done + ok:boolean), so
72:  children: { status: string; ok: boolean | null }[],
78:      : j.ok === false
86:export const lifecycleTone: Record<Lifecycle, "ok" | "warn" | "bad" | "run" | "muted"> = {
87:  completed: "ok",
###### WorkflowView ######
61:    if (phase === "completed" || phase === "done") { const ok = obj.ok !== false; return { label: "completed", detail: compact(obj, ["phase", "ok"]), kind: ok ? "ok" : "fail" }; }
96:        const ok = obj.ok !== false;
###### state/jobs.tsx ######
36:  }) => Promise<{ ok: boolean; data?: Record<string, unknown>; error?: string }>;
128:      if (r.ok) {
133:        return { ok: true, data: r as Record<string, unknown> };
142:      return { ok: false, error: err };
152:        return { ok: false, error: "stopped by operator" };
159:      return { ok: false, error: err };

###### routes/jobs.ts full ######
     1	// routes/jobs — the jobs resource. Thin: validate against the contract, call the
     2	// queue engine, validate out. No domain logic here (it lives in server/jobs/*).
     3	import { Hono } from 'hono'
     4	import { streamSSE } from 'hono/streaming'
     5	import { zValidator } from '@hono/zod-validator'
     6	import { EnqueueRequest } from '../contract/job.ts'
     7	import { catalogSummary } from '../jobs/catalog.ts'
     8	import { enqueue, getJob, listJobs, isLedgerTerminal } from '../jobs/queue.ts'
     9	
    10	export const jobs = new Hono()
    11	
    12	// The catalog IS the API surface — no dir scan, no convention-guessing.
    13	jobs.get('/verbs', (c) => c.json({ verbs: catalogSummary() }))
    14	
    15	// Enqueue. Envelope validated by zValidator; the verb's own input schema is
    16	// validated inside enqueue(). The app.ts onError classifies by TYPE: an unknown
    17	// verb throws ValidationError → 400 `unknown_verb`; bad args throw a ZodError →
    18	// 422 `invalid_input` with structured issues.
    19	jobs.post('/', zValidator('json', EnqueueRequest), async (c) => {
    20	  const { verb, input } = c.req.valid('json')
    21	  return c.json(await enqueue(verb, input), 202)
    22	})
    23	
    24	jobs.get('/', async (c) => {
    25	  const n = Number(c.req.query('limit'))
    26	  const limit = Number.isFinite(n) && n > 0 ? Math.min(n, 500) : 50   // NaN/negatives → default, cap at 500
    27	  return c.json({ jobs: await listJobs({ limit }) })
    28	})
    29	
    30	jobs.get('/:id', async (c) => {
    31	  const job = await getJob(c.req.param('id'))
    32	  return job ? c.json(job) : c.json({ error: 'not_found' }, 404)
    33	})
    34	
    35	// Live status stream. Slice 1 polls the ledger until terminal; step-level events
    36	// ride the runner's channel in a later slice.
    37	jobs.get('/:id/events', (c) => streamSSE(c, async (stream) => {
    38	  const id = c.req.param('id')
    39	  let last = null
    40	  for (let i = 0; i < 600 && !stream.aborted; i += 1) {
    41	    const job = await getJob(id)
    42	    if (!job) { await stream.writeSSE({ event: 'error', data: JSON.stringify({ error: 'not_found' }) }); return }
    43	    if (job.status !== last) {
    44	      last = job.status
    45	      await stream.writeSSE({ event: 'status', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok }) })
    46	    }
    47	    if (isLedgerTerminal(job.status)) return
    48	    await stream.sleep(1000)
    49	  }
    50	}))
###### queue.ts getJob / enqueue signatures ######
21:// The durable `job` ledger's status vocabulary — the ONE source for "is this run
27:// TERMINAL must list EVERY status the runner banks on a finished row, or a reader
29:// SUCCESS as 'ok' (not 'done') and the benign terminals 'skipped'/'partial'/
36:  'ok', 'done', 'completed', 'failed', 'blocked', 'skipped', 'partial',
39:export const isLedgerTerminal = (status: string): boolean =>
40:  (LEDGER_TERMINAL as readonly string[]).includes(status)
49:  status: string
52:  ok?: boolean | null
78:export async function enqueue(
82:): Promise<{ job_id: string; status: 'queued' }> {
98:    status: 'queued',
105:  return { job_id, status: 'queued' }
125:export async function enqueueRaw(
130:): Promise<{ job_id: string; status: 'queued' }> {
138:    status: 'queued',
146:  return { job_id, status: 'queued' }
150:export async function claim(workerId: string): Promise<JobRow | null> {
156:export async function getJob(job_id: string): Promise<JobRow | null> {
162:export async function listJobs({ limit = 50 }: { limit?: number } = {}): Promise<JobRow[]> {
167:// ── The single status-writer funnel (issue #19 Phase 1) ───────────────────────
168:// updateRunStatus is the ONE place a `run` row's status transition is written in
169:// the queue path: it derives the legacy `ok` boolean FROM the status (so no caller
170:// hand-writes an inconsistent ok/status pair — the drift Phase 1 kills), stamps
171:// ended_at on a terminal status, and scopes the PATCH to non-terminal rows so a
174:// EXPAND/CONTRACT: `ok` is still WRITTEN (derived), not dropped — the read-side
175:// migration off `ok` onto `status` is the later contract step. The LIVE runner-
180:// Statuses that mean the run SUCCEEDED (ok=true). A benign terminal
182:// with ok=false — so okForStatus returns false for it, matching the ledger's data.
183:const LEDGER_SUCCESS = ['ok', 'done', 'completed'] as const
185:/** Derive the legacy `ok` boolean a ledger status implies: true for a success
187:export function okForStatus(status: string): boolean | null {
188:  if ((LEDGER_SUCCESS as readonly string[]).includes(status)) return true
189:  if (isLedgerTerminal(status)) return false
193:/** Build the PATCH body for a status transition (pure — no I/O; unit-tested). */
194:export function statusPatch(status: string, reason?: unknown): Record<string, unknown> {
196:  const body: Record<string, unknown> = { status, ok: okForStatus(status), updated_at: now }
197:  if (isLedgerTerminal(status)) body.ended_at = now
202:/** THE single writer for a `run` row's status transition (queue path). Scoped to
204: * row in that case (a safe no-op). Callers pass a status; `ok` is derived here. */
205:export async function updateRunStatus(job_id: string, status: string, reason?: unknown): Promise<void> {

(Bash completed with no output)

The user doesn't want to take this action right now. STOP what you are doing and wait for the user to tell you how to proceed.

=== branch ===
fix/epic-6
=== status ===
=== log (epic-6 vs main) ===
19d2db5c refactor(ui): finish batch→workflow_run rename across broadcast + API + tab (#50 / epic #200)
fad420bd refactor(ui): rename operator-facing Batches surface → Workflow runs (#50 / epic #200)
12058ffd Merge branch 'fix/epic-38' into 'main'
40ba11c7 fix(runtime): checkpoint block-detection, claimed-row reaper + worker SIGTERM, scheduler backoff, login freshness gate (epic #38)
=== recent ===
19d2db5c refactor(ui): finish batch→workflow_run rename across broadcast + API + tab (#50 / epic #200)
fad420bd refactor(ui): rename operator-facing Batches surface → Workflow runs (#50 / epic #200)
12058ffd Merge branch 'fix/epic-38' into 'main'
40ba11c7 fix(runtime): checkpoint block-detection, claimed-row reaper + worker SIGTERM, scheduler backoff, login freshness gate (epic #38)
c19576ff fix(db): reconcile applied-but-missing run-rename migrations onto main + fix replay order (#305 / epic #200)

###### taxonomy.ts 60-95 ######
     1	// collision only mis-groups two concurrent bulk runs, and the ts+rand makes
     2	// that vanishingly unlikely.
     3	export const BATCH_ID_PREFIX = "bat_";
     4	export function newBatchId(): string {
     5	  return `${BATCH_ID_PREFIX}${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 7)}`;
     6	}
     7	
     8	// Roll a batch's child jobs up to ONE lifecycle for the batch status chip.
     9	// The Jobs page holds BridgeJobs (status queued|running|done + ok:boolean), so
    10	// map each to a lifecycle first, then reuse rollup(). `aborted` (stop-on-block
    11	// halted the batch early) → terminated when nothing succeeded, else partial.
    12	export function batchRollup(
    13	  children: { status: string; ok: boolean | null }[],
    14	  opts: { aborted?: boolean } = {},
    15	): Lifecycle {
    16	  const lifecycles = children.map((j) =>
    17	    j.status === "running" || j.status === "queued"
    18	      ? (j.status as Lifecycle)
    19	      : j.ok === false
    20	        ? "failed"
    21	        : "completed",
    22	  );
    23	  return rollup(lifecycles, opts);
    24	}
    25	
    26	// Lifecycle → display tone, for chips/dots across the Jobs UI.
    27	export const lifecycleTone: Record<Lifecycle, "ok" | "warn" | "bad" | "run" | "muted"> = {
    28	  completed: "ok",
    29	  partial: "warn",
    30	  terminated: "warn",
    31	  failed: "bad",
    32	  canceled: "muted",
    33	  timed_out: "bad",
    34	  running: "run",
    35	  queued: "muted",
    36	  created: "muted",
###### AccountJobs 24-45 (reference) ######
     1	
     2	type Phase = "running" | "queued" | "ok" | "failed";
     3	const COLOR: Record<Phase, string> = { running: "var(--color-success)", queued: "var(--color-warning)", ok: "var(--color-info-fg)", failed: "var(--color-danger-fg)" };
     4	const PHASE_LABEL: Record<Phase, string> = { running: "Running", queued: "Queued", ok: "Done", failed: "Failed" };
     5	
     6	// Durable-ledger phase — derive from `status` (the authoritative field the single
     7	// funnel writes), NOT the retiring `run.ok` boolean (#295: read-side off run.ok).
     8	// Any terminal status that isn't `ok` is a failure; `ok` is only a fallback for a
     9	// row with no status at all (never the case for a durable row).
    10	function phaseFromStatus(status: string | null | undefined, ok: boolean | null | undefined): Phase {
    11	  if (status === "running") return "running";
    12	  if (status === "queued") return "queued";
    13	  if (status === "ok") return "ok";
    14	  if (status) return "failed";
    15	  return ok === false ? "failed" : "ok";
    16	}
    17	function phaseOfBridge(j: BridgeJob): Phase {
    18	  if (j.status === "running") return "running";
    19	  if (j.status === "queued") return "queued";
    20	  return j.ok === false ? "failed" : "ok";
    21	}
    22

The file /home/philip/Documents/GitHub/bluebut-epic-6/ui/src/lib/taxonomy.ts has been updated successfully. (file state is current in your context — no need to Read it back)

###### WorkflowView 50-105 ######
     1	  const line = String(raw).trim();
     2	  if (!line || line.startsWith("[RESULT]") || line.startsWith("[QUEUED]")) return null;
     3	  // Drop per-request GraphQL chatter — network noise, captured elsewhere.
     4	  if (line.includes("doc_id=") || line.includes('"⇄') || /["\s]⇄ /.test(line)) {
     5	    try { const o = JSON.parse(line); const s = String(o.step ?? ""); if (s.startsWith("⇄") || s.includes("doc_id=")) return null; } catch { if (line.includes("doc_id=")) return null; }
     6	  }
     7	  let obj: Record<string, unknown> | null = null;
     8	  if (line.startsWith("{")) { try { obj = JSON.parse(line); } catch { obj = null; } }
     9	  if (obj) {
    10	    const phase = String(obj.phase ?? obj.event ?? "");
    11	    if (phase === "step" && obj.step) { const t = String(obj.step); return { label: t, kind: kindOf(t) }; }
    12	    if (phase === "completed" || phase === "done") { const ok = obj.ok !== false; return { label: "completed", detail: compact(obj, ["phase", "ok"]), kind: ok ? "ok" : "fail" }; }
    13	    if (phase) { const detail = compact(obj, ["phase", "event"]); return { label: phase, detail, kind: kindOf(`${phase} ${detail}`) }; }
    14	    const detail = compact(obj, []);
    15	    return detail ? { label: detail.slice(0, 80), kind: kindOf(detail) } : null;
    16	  }
    17	  return { label: line.replace(/\x1b\[[0-9;]*m/g, ""), kind: kindOf(line) };
    18	}
    19	
    20	// Timestamped path — keeps each step's wall-clock epoch for the time gutter and
    21	// recording sync.
    22	export function parseFlowEntries(entries: LogEntry[]): FlowStep[] {
    23	  const out: FlowStep[] = [];
    24	  for (const e of entries) { const s = parseLine(e.s); if (s) out.push({ ...s, at: e.t }); }
    25	  return out;
    26	}
    27	
    28	export function parseFlow(lines: string[]): FlowStep[] {
    29	  const out: FlowStep[] = [];
    30	  for (const raw of lines) {
    31	    const line = String(raw).trim();
    32	    if (!line || line.startsWith("[RESULT]") || line.startsWith("[QUEUED]")) continue;
    33	    // Drop per-request GraphQL chatter (the runner logs every FB query as
    34	    // "⇄ <name> doc_id=…") — it's network noise (the cloakbox wiretap holds the
    35	    // real traffic) and would bury the actual workflow.
    36	    if (line.includes("doc_id=") || line.includes('"⇄') || /["\s]⇄ /.test(line)) {
    37	      try { const o = JSON.parse(line); const s = String(o.step ?? ""); if (s.startsWith("⇄") || s.includes("doc_id=")) continue; } catch { if (line.includes("doc_id=")) continue; }
    38	    }
    39	    let obj: Record<string, unknown> | null = null;
    40	    if (line.startsWith("{")) { try { obj = JSON.parse(line); } catch { obj = null; } }
    41	    if (obj) {
    42	      const phase = String(obj.phase ?? obj.event ?? "");
    43	      if (phase === "step" && obj.step) {
    44	        const t = String(obj.step);
    45	        out.push({ label: t, kind: kindOf(t) });
    46	      } else if (phase === "completed" || phase === "done") {
    47	        const ok = obj.ok !== false;
    48	        out.push({ label: "completed", detail: compact(obj, ["phase", "ok"]), kind: ok ? "ok" : "fail" });
    49	      } else if (phase) {
    50	        const detail = compact(obj, ["phase", "event"]);
    51	        out.push({ label: phase, detail, kind: kindOf(`${phase} ${detail}`) });
    52	      } else {
    53	        const detail = compact(obj, []);
    54	        if (detail) out.push({ label: detail.slice(0, 80), kind: kindOf(detail) });
    55	      }
    56	    } else {

The file /home/philip/Documents/GitHub/bluebut-epic-6/ui/src/components/WorkflowView.tsx has been updated successfully. (file state is current in your context — no need to Read it back)

###### queue.ts JobRow 44-76 ######
     1	  id?: string          // the `run` PK (epic #200 rename of job_id); normalized → job_id below
     2	  job_id: string
     3	  verb?: string | null
     4	  cmd?: string | null
     5	  account?: string | null
     6	  status: string
     7	  input?: unknown
     8	  batch_id?: string | null
     9	  ok?: boolean | null
    10	  [k: string]: unknown
    11	}
    12	
    13	// epic #200 renamed the ledger table job→run: the PK column is now `id` (was `job_id`)
    14	// and the batch FK is `workflow_run_id` (was `batch_id`). The queue's public JobRow shape
    15	// keeps job_id/batch_id, so EVERY row read back from `run` is normalized HERE at the seam
    16	// — worker.ts, the /api/queue routes, and the tests are untouched by the column rename.
    17	function normalizeRun(row: Record<string, unknown>): JobRow {
    18	  return {
    19	    ...row,
    20	    job_id: String(row.id ?? row.job_id ?? ''),
    21	    batch_id: (row.workflow_run_id ?? row.batch_id ?? null) as string | null,
    22	  } as JobRow
    23	}
    24	
    25	/** Optional ledger-level knobs an enqueue can set (issue #217): a batch grouping, the
    26	 *  `not_before` due-time (pacing — the row is invisible to claim_job_v2 until due), and
    27	 *  `priority` (claim order). All optional; omitting them = legacy FIFO immediate-claim. */
    28	export interface EnqueueOpts {
    29	  batch_id?: string | null
    30	  not_before?: string | null   // ISO timestamptz; a queued row won't be claimed before this
    31	  priority?: number            // higher = claimed first (default 0)
    32	}
    33

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/routes/jobs.ts has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/routes/jobs.ts has been updated successfully. (file state is current in your context — no need to Read it back)

###### scheduler.mjs 1-30 + around spawn (70-100) ######
     1	// runtime/scheduler — the in-process recurring scheduler (job_schedule tick) + the
     2	// runtime recognizer loop, relocated VERBATIM from runtime.mjs (issue #21 §4). The
     3	// tick fns and the SCHED_ENABLED gate live here; the BOOT triggers (the setInterval /
     4	// setTimeout that start them) are owned by the runtime.mjs barrel so their firing
     5	// order is deterministic — this module only defines, it does not self-start.
     6	
     7	import { spawn } from "node:child_process";
     8	import { rest as dbRest } from "../../../adapters/_lib/db/db.ts";
     9	import { groupsRepo } from "../../../adapters/_lib/repos/index.ts";
    10	import { mintJobId } from "../../jobs/job-id.ts";
    11	import { isDue, nextRunAt } from "../../../adapters/_lib/schedule.ts";
    12	import { resolveScheduleActor } from "../../../adapters/_lib/schedule-actor.ts";
    13	import { workflowCmd } from "../../../adapters/_lib/workflow/workflow.ts";
    14	import { API_TOKEN } from "../lib.mjs";
    15	import { RUNNER, NODE, adapterPath, childEnv, parseLoose, envFile } from "./spawn-env.mjs";
    16	import { newJob, pushLog, withAccountLock } from "./job-registry.mjs";
    17	import { sweep, usableDrivableFleet } from "./run-core.mjs";
    18	import { workflowStart, resolveWorkflowPlan } from "./workflow.mjs";
    19	import { numEnv } from "../../../adapters/_lib/num-env.ts";
    20	
    21	// ── Recurring jobs (job_schedule) — the in-process scheduler tick ─────────────
    22	// The bridge is the always-on process that already spawns runner jobs and holds
    23	// the per-account lock, so recurring jobs tick RIGHT HERE — not as an external
    24	// Coolify cron. Every SCHED_TICK_MS we read the enabled job_schedule rules, fire
    25	// each one that's DUE as a real job through the SAME machinery /exec/stream uses
    26	// (newJob → account lock → childEnv spawn), and stamp run-state back to the DB.
    27	// State lives in the DB, so a bridge restart loses nothing. The due-ness math is
    28	// pure + unit-tested in adapters/_lib/schedule.js (isDue/nextRunAt). No HTTP hop,
    29	// no bridge token, no fresh node process per minute — the killed-off cron's cost.
    30	const SCHED_TICK_MS = numEnv('BLUEBUT_SCHED_TICK_MS', 60_000);
----
     1	  let h = 2166136261;                                            // FNV-1a(schedule_id:bucket)
     2	  const key = `${s.schedule_id}:${bucket}`;
     3	  for (let i = 0; i < key.length; i++) { h ^= key.charCodeAt(i); h = Math.imul(h, 16777619); }
     4	  return Math.floor(((h >>> 0) % 100000) / 100000 * maxJ);
     5	}
     6	
     7	// Build the adapter argv for a schedule: the JIT-resolved account, optional
     8	// --groups csv, verbatim extras. `acct` is resolved at fire time (never s.account
     9	// directly) so a device-less / dark account is never dispatched.
    10	function schedArgs(s, acct) {
    11	  const out = [];
    12	  if (acct) out.push("--account", acct);
    13	  const groups = Array.isArray(s.groups) ? s.groups.filter(Boolean) : [];
    14	  if (groups.length) out.push("--groups", groups.join(","));
    15	  for (const a of Array.isArray(s.args) ? s.args : []) out.push(String(a));
    16	  return out;
    17	}
    18	
    19	// Fire ONE schedule as a real job — the headless twin of /exec/stream's task():
    20	// same newJob + withAccountLock + childEnv spawn, minus the SSE response. The
    21	// runner writes the durable ledger row (jobId = BLUEBUT_JOB_ID) for free.
    22	function fireSchedule(s, jobId, acct) {
    23	  const site = s.site || "facebook";
    24	  const cmd = s.cmd;
    25	  const adapter = adapterPath(site, cmd);
    26	  if (!adapter) return Promise.resolve({ ok: false, jobId, error: `no adapter for ${site}/${cmd}` });
    27	  const args = schedArgs(s, acct);
    28	  const job = newJob({ id: jobId, account: acct, kind: "adapter", cmd: `${site}/${cmd}`, label: `⏰ ${s.label} · ${site}/${cmd}${acct ? ` (${acct})` : ""}` });
    29	  job.traceJobId = jobId;   // == job.id (id-collapse); kept for the UI/analyze-job field
    30	  job.scheduled = true;   // mute routine schedule ticks in Telegram (BLUEBUT_TG_SCHEDULED=1 re-enables)
    31	  const cliArgs = [RUNNER, adapter, ...args.filter((a) => a != null).map(String)];
    32	  return new Promise((resolve) => {
    33	    const task = () => new Promise((done) => {
    34	      // batch_id = schedule_id → every run this schedule fires is linkable in the ledger
    35	      // (job.batch_id = schedule_id), so the UI can open each run from the schedule.
    36	      const child = spawn(NODE, cliArgs, { env: childEnv(acct, jobId, s.schedule_id || null, site) });
    37	      const killTimer = setTimeout(() => { try { child.kill("SIGKILL"); } catch { /* gone */ } }, SCHED_TIMEOUT_MS + 15_000);
    38	      let out = "", ebuf = "";
    39	      child.stdout.on("data", (d) => (out += d));
    40	      child.stderr.on("data", (d) => {
    41	        ebuf += d; let i;
    42	        while ((i = ebuf.indexOf("\n")) >= 0) { const line = ebuf.slice(0, i).replace(/\x1b\[[0-9;]*m/g, ""); ebuf = ebuf.slice(i + 1); if (line.trim()) pushLog(job, line); }
    43	      });
    44	      child.on("close", () => { clearTimeout(killTimer); const parsed = parseLoose(out); job.ok = !!parsed && parsed.ok !== false; if (parsed?.diagnostic) job.diagnostic = parsed.diagnostic; resolve({ ok: job.ok, jobId, result: parsed }); done(); });
    45	      child.on("error", (e) => { clearTimeout(killTimer); job.ok = false; resolve({ ok: false, jobId, error: e.message }); done(); });
    46	    });
    47	    withAccountLock(acct, job, task, null);
    48	  });
    49	}
    50	
    51	// Resolve the TARGET SET for a workflow schedule. A per-account-discover cmd (the

1:// runtime/scheduler — the in-process recurring scheduler (job_schedule tick) + the
2:// runtime recognizer loop, relocated VERBATIM from runtime.mjs (issue #21 §4). The
3:// tick fns and the SCHED_ENABLED gate live here; the BOOT triggers (the setInterval /
21:// ── Recurring jobs (job_schedule) — the in-process scheduler tick ─────────────
23:// the per-account lock, so recurring jobs tick RIGHT HERE — not as an external
32:const _schedInflight = new Set();   // schedule_ids currently firing — guards tick overlap
33:let _schedTicking = false;
39:// simply re-fires next tick. Broadcast batches serialize separately (_activeBatches).
46:// pending window (no early/late flip-flop as the 60s tick re-checks) yet shifts every
48:function schedFireJitterMs(s) {
64:function schedArgs(s, acct) {
76:function fireSchedule(s, jobId, acct) {
84:  job.scheduled = true;   // mute routine schedule ticks in Telegram (BLUEBUT_TG_SCHEDULED=1 re-enables)
111:// a newly-joined group self-includes on the next tick (no per-group schedule churn).
112:async function resolveScheduleGroupIds(s) {
144:async function fireWorkflowSchedule(s) {
158:    // re-fire every tick, fail identically, and — because it returns `skipped` not
185:// in-memory guards (_schedInflight/_schedTicking) only serialize WITHIN one process;
192:// last_run_at un-advanced so the schedule re-fired every tick; here the winning CAS
194:async function casClaimSchedule(scheduleId, oldLra, body) {
202:    // A failed claim PATCH must NOT be treated as a win — better to skip this tick
203:    // (the schedule stays due and re-CASes next tick) than to fire un-stamped.
209:async function tickSchedules() {
210:  if (_schedTicking) return;
211:  _schedTicking = true;
216:      if (_schedInflight.has(s.schedule_id)) continue;   // still running from a prior tick
219:      // later tick (don't stamp last_run — its jitter re-rolls and it re-fires when free).
223:      if (isBrowserSweep) sweep.batch = "starting";   // reserve synchronously (closes the same-tick race)
239:        // at START (holds the interval gate across a restart) only if no other tick/
256:            // last_run so it re-fires next tick once accounts/targets appear (no dead
260:            // instead of re-resolving the whole fleet every single tick forever (a
302:      // CAS-claim the interval slot at START (issue #43.F3) — only if no other tick/
310:      fireSchedule(s, jobId, acct)
317:  } catch (e) { console.error(`[backend] schedule tick failed: ${e?.message || e}`); }
318:  finally { _schedTicking = false; }
326:// locally with BLUEBUT_FORCE_SCHEDULER=1 if you really mean to test the tick.
328:// must NEVER tick the scheduler, which fires REAL prod job_schedule rows the instant it
329:// evaluates (tickSchedules() below runs immediately). Without this guard, `npm run
336:const SCHED_ENABLED = SCHED_TICK_MS > 0
341:// ── RUNTIME recognizer — recognize cloak_captures → fb_* continuously ─────────

1	async function tickSchedules() {
     2	  if (_schedTicking) return;
     3	  _schedTicking = true;
     4	  try {
     5	    const rows = (await dbRest("GET", "run_schedule?enabled=eq.true")) || [];
     6	    const now = new Date();
     7	    for (const s of rows) {

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/http/runtime/scheduler.mjs has been updated successfully. (file state is current in your context — no need to Read it back)

1	// ── FINAL job-spawn catch-all — /login/stream + the site/cmd adapter spawn. RAW
     2	//    (req,res), byte-for-byte identical to the original: reads the body, then
     3	//    dispatches. A request with no site/cmd gets a 400 here (the chain's terminal
     4	//    handler), so this ALWAYS handles the request.
     5	function execCatchAll(req: IncomingMessage, res: ServerResponse) {
     6	  let body = "";
     7	  req.on("data", (c) => (body += c));
     8	  req.on("end", () => {
     9	    let b: Record<string, unknown> = {};
    10	    try { b = JSON.parse(body || "{}"); } catch { /* empty */ }
    11	    const path0 = (req.url || "").split("?")[0];
    12	
    13	    if (path0 === "/login/stream") { handleLogin(req, res, b); return; }
    14	
    15	    // Header-only auth on the process-spawn path (issue #20): /exec runs an adapter
    16	    // subprocess. A query-string token (accepted by the global gate for reads) is too
    17	    // weak to authorize spawning code — require the Authorization header here.
    18	    if ((path0 === "/exec" || path0 === "/exec/stream") && !authed(req, { allowQuery: false })) {
    19	      res.writeHead(401, { ...CORS, "Content-Type": "application/json" });
    20	      res.end(JSON.stringify({ ok: false, error: "unauthorized — /exec requires an Authorization header" }));
    21	      return;
    22	    }
    23	
    24	    const { site, cmd, args = [], timeout = 180, job_id, batch_id, batch_label, batch_size, batch_index } = b as {
    25	      site?: string; cmd?: string; args?: unknown[]; timeout?: number;
    26	      job_id?: string; batch_id?: string; batch_label?: string; batch_size?: number; batch_index?: number;
    27	    };
    28	    if (!site || !cmd) {
    29	      res.writeHead(400, { ...CORS, "Content-Type": "application/json" });
    30	      res.end(JSON.stringify({ ok: false, error: "site and cmd required" }));
    31	      return;
    32	    }
    33	    if (!Array.isArray(args)) {
    34	      res.writeHead(400, { ...CORS, "Content-Type": "application/json" });
    35	      res.end(JSON.stringify({ ok: false, error: "args must be an array" }));
    36	      return;
    37	    }
    38	    const acct = accountFromArgs(args);
    39	    const adapter = adapterPath(site, cmd);
    40	    if (!adapter) {
    41	      res.writeHead(404, { ...CORS, "Content-Type": "application/json" });
    42	      res.end(JSON.stringify({ ok: false, error: `no adapter at adapters/${site}/${cmd}.js (this repo owns its adapters; fork the toolbox one if you need it)` }));
    43	      return;
    44	    }
    45	    // Adapters used to be invoked as `opencli <site> <cmd> ...args --format json`.
    46	    // Now they're plain Node scripts run through bin/bluebut-adapter.mjs, which
    47	    // drops the `--format json` flag (it always emits JSON) and the {site, cmd}
    48	    // pair (the file path encodes both). All other args pass through unchanged.
    49	    const cliArgs = [RUNNER, adapter, ...(args as unknown[]).filter((a) => a != null && a !== "--format" && a !== "json").map(String)];
    50	    const isStream = path0 === "/exec/stream";
    51	    const job = newJob({
    52	      account: acct, kind: "adapter", cmd: `${site}/${cmd}`,
    53	      label: `${site}/${cmd}${acct ? ` (${acct})` : ""}`, group: groupFromArgs(args),
    54	      batch: batch_id, batchLabel: batch_label, batchSize: batch_size, batchIndex: batch_index,
    55	    });
    56	    // The id threaded to the runner as BLUEBUT_JOB_ID — what its HyperDX traffic
    57	    // is keyed by. Prefer the caller's UI job_id (so the in-memory UI job stays
    58	    // correlated); fall back to THIS bridge job's id so an adapter fired without a
    59	    // UI job (bulk forms, CLI) is still queryable. Stored on the record so the
    60	    // Debug page can find the HyperDX trail + the per-job HAR.
    61	    const jobTraceId = job_id || job.id;
    62	    job.traceJobId = jobTraceId;
    63	
    64	    if (isStream) res.writeHead(200, { ...CORS, "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive" });
    65	
    66	    // Operator stop: kill the live child if running, else (still queued) drop it
    67	    // from the account queue and close the response now. `child` is filled in by
    68	    // task() once this job's turn comes.
    69	    let child: ReturnType<typeof spawn> | null = null;
    70	    job.cancel = () => {
    71	      job.canceled = true;
    72	      if (child) { try { child.kill("SIGKILL"); } catch { /* gone */ } return; }
    73	      if (acct) { const q = ACCOUNT_LOCK.get(acct); if (q) q.queue = q.queue.filter((fn: { __jobId?: string }) => fn.__jobId !== job.id); }
    74	      try {
    75	        if (isStream) { if (!res.writableEnded) { res.write(`data: [ERROR] canceled by operator\n\n`); res.end(); } }
    76	        else if (!res.headersSent) { res.writeHead(200, { ...CORS, "Content-Type": "application/json" }); res.end(JSON.stringify({ ok: false, error: "canceled by operator" })); }
    77	      } catch { /* response already gone */ }
    78	      job.endedAt = Date.now(); job.status = "done"; job.ok = false;
    79	    };
    80	
    81	    // The actual run — spawned only when it's this job's turn on the account
    82	    // lock, so two jobs never drive the same account's Chrome at once.
    83	    const task = () => new Promise<void>((resolve) => {
    84	      if (job.canceled) { resolve(); return; }   // canceled while queued — cancel() already closed res
    85	      child = spawn(NODE, cliArgs, { env: childEnv(acct, jobTraceId, batch_id, site) });
    86	      const killTimer = setTimeout(() => child?.kill("SIGKILL"), (Number(timeout) + 15) * 1000);
    87	      res.on("close", () => { if (!res.writableEnded) { try { child?.kill("SIGKILL"); } catch { /* gone */ } } });
    88	
    89	      // Screen recording now happens INSIDE the runner (adapters/_lib/netcapture.ts
    90	      // screencasts the job's own page over the SAME CDP socket that captures the

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/http/routes/bridge.ts has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/http/routes/bridge.ts has been updated successfully. (file state is current in your context — no need to Read it back)

###### contract/job.ts ######
     1	// contract — the ONE typed surface for the jobs API, shared by the server (validates
     2	// out) and, later, the UI (`z.infer` in, replacing the 96 hand-written interfaces in
     3	// ui/src/api/index.ts). zod is the runtime source of truth: every route validates its
     4	// input/output against these, and the OpenAPI/docs are generated from them — nothing
     5	// is hand-described twice. See docs/ARCHITECTURE-AND-DEV.md (the verbs plane).
     6	import { z } from 'zod'
     7	
     8	// A job's lifecycle status. `queued` → `claimed` → `running` → terminal. This MUST
     9	// list every status the ledger actually holds, or output validation (Phase 4) 500s
    10	// on a real row: the runner banks SUCCESS as `'ok'` (adapter-finalize.
    11	// jobLifecycleStatus) — NOT `'done'` (that is the in-mem bridge map's word) — plus
    12	// the benign terminals `'skipped'`/`'partial'`/`'content_unavailable'`. Kept in
    13	// step with server/jobs/queue.ts LEDGER_ACTIVE+LEDGER_TERMINAL (the ledger's own
    14	// vocab, distinct from taxonomy.ts RunStatus until the #19 rename lands).
    15	//
    16	// #263 BRIDGE (documented, not collapsed): this enum is now also enforced at the DB layer —
    17	// db/migrations/entity/0049_run_status_check.sql binds `run.status` to this exact value set
    18	// (plus 'created' for forward-compat with taxonomy.ts LIFECYCLE). The write-path cutover this
    19	// comment used to gesture at (`ok`→`completed`, retiring JobStatus in favor of taxonomy.ts
    20	// RunStatus) is NOT done here — success is written as 'ok' at 8+ call sites (adapter-finalize,
    21	// the broadcast/workflow runtime, the queue) and a partial rename risks a live-jobs break for
    22	// no correctness gain until the #19 single-writer `updateRunStatus()` funnel lands. JobStatus
    23	// stays the accepted superset; RunStatus stays the target. Do not add a 3rd vocabulary.
    24	export const JobStatus = z.enum([
    25	  // active
    26	  'queued', 'claimed', 'running',
    27	  // terminal (ledger vocabulary)
    28	  'ok', 'done', 'completed', 'failed', 'blocked', 'skipped', 'partial',
    29	  'content_unavailable', 'terminated', 'timed_out', 'canceled',
    30	])
    31	
    32	// The durable job row, as the API returns it (a projection of the `job` ledger table).
    33	export const Job = z.object({
    34	  job_id: z.string(),
    35	  verb: z.string().nullable().optional(),
    36	  cmd: z.string().nullable().optional(),
    37	  account: z.string().nullable().optional(),
    38	  target_type: z.string().nullable().optional(),
    39	  target_id: z.string().nullable().optional(),
    40	  status: JobStatus,
    41	  ok: z.boolean().nullable().optional(),
    42	  aborted_reason: z.string().nullable().optional(),
    43	  summary: z.record(z.string(), z.unknown()).nullable().optional(),
    44	  created_at: z.string().nullable().optional(),
    45	  started_at: z.string().nullable().optional(),
    46	  ended_at: z.string().nullable().optional(),
    47	  duration_ms: z.number().nullable().optional(),
    48	})
    49	
    50	// POST /jobs — enqueue a job. `verb` selects a catalog entry; `input` is validated
    51	// against that verb's own input schema by the route (not here — this is the envelope).
    52	export const EnqueueRequest = z.object({
    53	  verb: z.string(),
    54	  input: z.record(z.string(), z.unknown()).default({}),
    55	})
    56	
    57	export const EnqueueResponse = z.object({
    58	  job_id: z.string(),
    59	  status: JobStatus,
    60	})
    61	
    62	// A single progress event on the SSE stream (GET /jobs/:id/events). Mirrors the
    63	// runner's stderr JSONL step shape, normalized.
    64	export const JobEvent = z.object({
    65	  job_id: z.string(),
    66	  at: z.string(),
    67	  kind: z.enum(['step', 'action', 'status', 'result', 'error']),
    68	  message: z.string().optional(),
    69	  status: JobStatus.optional(),
    70	  data: z.record(z.string(), z.unknown()).optional(),
    71	})
###### catalog verb names (sample) ######
server/jobs/catalog.ts:1:// catalog — the typed verb registry. THE source of truth for "what jobs exist".
server/jobs/catalog.ts:5:// `adapters/<site>/<cmd>.js` by string. Here a verb is a plain typed object you
server/jobs/catalog.ts:6:// `import` — the command list is Object.keys(VERBS), arg validation is the verb's
server/jobs/catalog.ts:7:// own zod schema, the UI's available-actions is this catalog serialized (GET /verbs),
server/jobs/catalog.ts:10:// A verb OWNS four things: its input schema (zod), how that input maps to the runner
server/jobs/catalog.ts:12:// verbs and mutations share the same shape; the kernel treats them identically
server/jobs/catalog.ts:15:// The verb PRIMITIVE (defineVerb + the argv-flag helpers + the Verb type) lives in
server/jobs/catalog.ts:16:// ./verbs/kernel; the verb DEFINITIONS are split per-domain (./verbs/groups,
server/jobs/catalog.ts:17:// ./verbs/people) so no single file crosses the god-file budget (inv #16). This
server/jobs/catalog.ts:21:import { defineVerb, type Verb, type VerbKind, type LedgerTarget } from './verbs/kernel.ts'
server/jobs/catalog.ts:22:import { groupVerbs } from './verbs/groups.ts'
server/jobs/catalog.ts:23:import { groupAdminVerbs } from './verbs/group-admin.ts'
server/jobs/catalog.ts:24:import { pageVerbs } from './verbs/pages.ts'
server/jobs/catalog.ts:25:import { peopleVerbs } from './verbs/people.ts'
server/jobs/catalog.ts:26:import { leadVerbs } from './verbs/leads.ts'
server/jobs/catalog.ts:27:import { opsVerbs } from './verbs/ops.ts'
server/jobs/catalog.ts:40:  throw new Error('catalog: duplicate verb name across domain modules')
server/jobs/catalog.ts:43:/** Look up a verb or throw a caller-friendly error. */
server/jobs/catalog.ts:44:export function getVerb(name: string): Verb {
server/jobs/catalog.ts:46:  // Typed error (issue #27): a bad verb name is client input → ValidationError, so the

###### catalogSummary + getVerb + a verb def ######
     1	/** Look up a verb or throw a caller-friendly error. */
     2	export function getVerb(name: string): Verb {
     3	  const v = VERBS[name]
     4	  // Typed error (issue #27): a bad verb name is client input → ValidationError, so the
     5	  // queue's onError boundary branches on `instanceof`, NOT a regex over err.message
     6	  // (the inv #7 anti-pattern). Message kept for the operator + the existing test.
     7	  if (!v) throw new ValidationError(`unknown verb '${name}' — known: ${Object.keys(VERBS).join(', ')}`)
     8	  return v
     9	}
    10	
    11	/** Serialize the catalog for GET /verbs (schema rendered as JSON Schema). */
    12	export function catalogSummary() {
    13	  return Object.values(VERBS).map((v) => ({
    14	    name: v.name,
    15	    site: v.site,
    16	    kind: v.kind,
    17	    summary: v.summary,
    18	    input: z.toJSONSchema(v.input),
    19	  }))
    20	}
###### one verb def (ops or groups) ######
     1	// verbs/ops — operational verbs: the recurring health monitors (scheduler-run),
     2	// the durable-media backfills, the capture/diagnostic tools, and the e2ee
     3	// keepalive. Split from catalog.ts (inv #16 god-file budget); imports the verb
     4	// primitive from ./kernel. These are our-DB / socket / capture work, no FB
     5	// mutation. The monitors are the recurring jobs Phase-2's worker will claim from
     6	// job_schedule instead of a web-process setInterval (epic #19, inv #23).
     7	import { z } from 'zod'
     8	import { defineVerb, flag, bflag, boolish, ResultBase, type Verb } from './kernel.ts'
     9	
    10	const quietOnly = {
    11	  input: z.object({ quiet: boolish }),
    12	  toArgv: (i: { quiet?: boolean }) => [...bflag('quiet', i.quiet)],
    13	  target: () => ({ type: null, id: null }),
    14	} as const
    15	
    16	export const opsVerbs: Record<string, Verb> = {
    17	  'backfill-page-avatars': defineVerb({
    18	    name: 'backfill-page-avatars',
    19	    site: 'facebook',
    20	    kind: 'read',
    21	    summary: 'Mirror missing Page avatars into MinIO (durable bytes) → fb_page.',
    22	    adapter: 'adapters/facebook/backfill-page-avatars.js',
    23	    input: z.object({
    24	      limit: z.coerce.number().int().positive().optional(),
    25	      concurrency: z.coerce.number().int().positive().max(32).optional(),
    26	      dryRun: boolish,
    27	    }),
    28	    output: ResultBase.extend({
    29	      candidates: z.number().optional(),
    30	      fetched: z.number().optional(),
    31	      silhouette: z.number().optional(),
    32	      missed: z.number().optional(),
    33	      seconds: z.number().optional(),
    34	    }),
    35	    toArgv: (i) => [...flag('limit', i.limit), ...flag('concurrency', i.concurrency), ...bflag('dry-run', i.dryRun)],
    36	    target: () => ({ type: null, id: null }),
    37	  }),
    38	
    39	  'backfill-person-avatars': defineVerb({
    40	    name: 'backfill-person-avatars',
    41	    site: 'facebook',
    42	    kind: 'read',
    43	    summary: 'Mirror missing person avatars into MinIO (durable bytes) → fb_person.',
    44	    adapter: 'adapters/facebook/backfill-person-avatars.js',
    45	    input: z.object({
    46	      limit: z.coerce.number().int().positive().optional(),
    47	      all: boolish,
    48	      dryRun: boolish,
    49	    }),
    50	    output: ResultBase.extend({
    51	      candidates: z.number().optional(),
    52	      stored: z.number().optional(),
    53	      failed: z.number().optional(),
    54	      seconds: z.number().optional(),
    55	    }),
    56	    toArgv: (i) => [...flag('limit', i.limit), ...bflag('all', i.all), ...bflag('dry-run', i.dryRun)],
    57	    target: () => ({ type: null, id: null }),
    58	  }),
    59	
    60	  'backfill-person-faces': defineVerb({

=== remotes ===
origin	https://claude-mcp-scoped:glpat-58AJ5MLOF1IvXNhvd-fPZW86MQp1OjQH.01.0w0e896lh@gitlab.hostbun.cc/devdashco/bluebut.git (fetch)
origin	https://claude-mcp-scoped:glpat-58AJ5MLOF1IvXNhvd-fPZW86MQp1OjQH.01.0w0e896lh@gitlab.hostbun.cc/devdashco/bluebut.git (push)
=== diff stat ===
 server/http/routes/bridge.ts       | 20 ++++++++++++++++++++
 server/http/runtime/scheduler.mjs  | 11 +++++++++++
 server/routes/jobs.ts              | 11 +++++++++--
 ui/src/components/WorkflowView.tsx |  2 ++
 ui/src/lib/taxonomy.ts             | 13 ++++++++++---
 5 files changed, 52 insertions(+), 5 deletions(-)
=== worktree status ===
 M server/http/routes/bridge.ts
 M server/http/runtime/scheduler.mjs
 M server/routes/jobs.ts
 M ui/src/components/WorkflowView.tsx
 M ui/src/lib/taxonomy.ts

diff --git a/server/http/routes/bridge.ts b/server/http/routes/bridge.ts
index dfec99d7..4fd9c2e4 100644
--- a/server/http/routes/bridge.ts
+++ b/server/http/routes/bridge.ts
@@ -418,6 +418,12 @@ const HONO_ROUTES = new Set([
   "POST /accounts/sync-farm",
 ]);
 
+// epic #6 cutover: when the durable queue worker owns dispatch (BLUEBUT_QUEUE_WORKER=1),
+// this inline /exec spawn is DORMANT — the UI drives jobs through /api/queue instead.
+// Leaving it live alongside the worker = double-dispatch (inv #23 runaway). Gated, not
+// deleted: unset the env and the inline path drives again. /login/stream is untouched.
+const QUEUE_WORKER_ON = process.env.BLUEBUT_QUEUE_WORKER === "1";
+
 // ── FINAL job-spawn catch-all — /login/stream + the site/cmd adapter spawn. RAW
 //    (req,res), byte-for-byte identical to the original: reads the body, then
 //    dispatches. A request with no site/cmd gets a 400 here (the chain's terminal
@@ -441,6 +447,20 @@ function execCatchAll(req: IncomingMessage, res: ServerResponse) {
       return;
     }
 
+    // epic #6: queue worker owns dispatch → don't spawn a duplicate child here. Reply
+    // terminally so the caller never hangs (the UI now uses /api/queue; a stale caller
+    // gets a clear disabled signal instead of a silent 200 that never drains).
+    if ((path0 === "/exec" || path0 === "/exec/stream") && QUEUE_WORKER_ON) {
+      if (path0 === "/exec/stream") {
+        res.writeHead(200, { ...CORS, "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive" });
+        res.end(`data: [ERROR] legacy /exec disabled — queue worker owns dispatch (epic #6)\n\n`);
+      } else {
+        res.writeHead(410, { ...CORS, "Content-Type": "application/json" });
+        res.end(JSON.stringify({ ok: false, error: "legacy /exec disabled — use /api/queue (epic #6)" }));
+      }
+      return;
+    }
+
     const { site, cmd, args = [], timeout = 180, job_id, batch_id, batch_label, batch_size, batch_index } = b as {
       site?: string; cmd?: string; args?: unknown[]; timeout?: number;
       job_id?: string; batch_id?: string; batch_label?: string; batch_size?: number; batch_index?: number;
diff --git a/server/http/runtime/scheduler.mjs b/server/http/runtime/scheduler.mjs
index 22274d59..df418be8 100644
--- a/server/http/runtime/scheduler.mjs
+++ b/server/http/runtime/scheduler.mjs
@@ -206,7 +206,18 @@ async function casClaimSchedule(scheduleId, oldLra, body) {
   }
 }
 
+// epic #6 cutover: when the durable queue worker owns dispatch (BLUEBUT_QUEUE_WORKER=1)
+// this legacy in-process scheduler must go DORMANT — else its fireSchedule() spawn
+// double-dispatches every due schedule alongside the worker (inv #23 runaway class).
+// Gated, not deleted: unset the env and it drives again.
+const QUEUE_WORKER_ON = process.env.BLUEBUT_QUEUE_WORKER === '1';
+let _schedDormantLogged = false;
+
 async function tickSchedules() {
+  if (QUEUE_WORKER_ON) {
+    if (!_schedDormantLogged) { console.log('[backend] legacy scheduler DORMANT — queue worker owns dispatch (epic #6)'); _schedDormantLogged = true; }
+    return;
+  }
   if (_schedTicking) return;
   _schedTicking = true;
   try {
diff --git a/server/routes/jobs.ts b/server/routes/jobs.ts
index cbbac85f..52374b16 100644
--- a/server/routes/jobs.ts
+++ b/server/routes/jobs.ts
@@ -5,7 +5,7 @@ import { streamSSE } from 'hono/streaming'
 import { zValidator } from '@hono/zod-validator'
 import { EnqueueRequest } from '../contract/job.ts'
 import { catalogSummary } from '../jobs/catalog.ts'
-import { enqueue, getJob, listJobs, isLedgerTerminal } from '../jobs/queue.ts'
+import { enqueue, getJob, listJobs, isLedgerTerminal, okForStatus } from '../jobs/queue.ts'
 
 export const jobs = new Hono()
 
@@ -44,7 +44,14 @@ jobs.get('/:id/events', (c) => streamSSE(c, async (stream) => {
       last = job.status
       await stream.writeSSE({ event: 'status', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok }) })
     }
-    if (isLedgerTerminal(job.status)) return
+    if (isLedgerTerminal(job.status)) {
+      // epic #6: emit a terminal `result` frame so the repointed UI job client
+      // (ui/src/api/jobs.ts, off /exec/stream → this stream) can resolve its
+      // RunResult. `ok` is the funnel-derived boolean; `result` rides through if
+      // the ledger row carries one (the UI falls back to deriving ok from status).
+      await stream.writeSSE({ event: 'result', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok ?? okForStatus(job.status), result: job.result ?? null }) })
+      return
+    }
     await stream.sleep(1000)
   }
 }))
diff --git a/ui/src/components/WorkflowView.tsx b/ui/src/components/WorkflowView.tsx
index 7a7a1a7d..d93a670d 100644
--- a/ui/src/components/WorkflowView.tsx
+++ b/ui/src/components/WorkflowView.tsx
@@ -58,6 +58,8 @@ function parseLine(raw: string): FlowStep | null {
   if (obj) {
     const phase = String(obj.phase ?? obj.event ?? "");
     if (phase === "step" && obj.step) { const t = String(obj.step); return { label: t, kind: kindOf(t) }; }
+    // epic #6: this `ok` is the runner's completion LOG event ({phase:"completed", ok}),
+    // i.e. wire data — NOT the retiring run.ok column — so it is read as-is.
     if (phase === "completed" || phase === "done") { const ok = obj.ok !== false; return { label: "completed", detail: compact(obj, ["phase", "ok"]), kind: ok ? "ok" : "fail" }; }
     if (phase) { const detail = compact(obj, ["phase", "event"]); return { label: phase, detail, kind: kindOf(`${phase} ${detail}`) }; }
     const detail = compact(obj, []);
diff --git a/ui/src/lib/taxonomy.ts b/ui/src/lib/taxonomy.ts
index 29f31afb..1bcf4db9 100644
--- a/ui/src/lib/taxonomy.ts
+++ b/ui/src/lib/taxonomy.ts
@@ -72,12 +72,19 @@ export function batchRollup(
   children: { status: string; ok: boolean | null }[],
   opts: { aborted?: boolean } = {},
 ): Lifecycle {
+  // epic #6: read-side prefers status over the retiring run.ok. A terminal status
+  // in the success set (ok/done/completed) rolls up as completed; any other
+  // terminal status is a failure. `ok` is only a fallback for a row with no status.
   const lifecycles = children.map((j) =>
     j.status === "running" || j.status === "queued"
       ? (j.status as Lifecycle)
-      : j.ok === false
-        ? "failed"
-        : "completed",
+      : j.status === "ok" || j.status === "done" || j.status === "completed"
+        ? "completed"
+        : j.status
+          ? "failed"
+          : j.ok === false
+            ? "failed"
+            : "completed",
   );
   return rollup(lifecycles, opts);
 }

4761dfd2 feat(runtime): epic #6 cutover scaffolding — queue-worker dispatcher, gated legacy paths, read-side off run.ok

remote: 
remote: To create a merge request for fix/epic-6, visit:        
remote:   https://gitlab.hostbun.cc/devdashco/bluebut/-/merge_requests/new?merge_request%5Bsource_branch%5D=fix%2Fepic-6        
remote: 
To https://gitlab.hostbun.cc/devdashco/bluebut.git
 * [new branch]        fix/epic-6 -> fix/epic-6
branch 'fix/epic-6' set up to track 'origin/fix/epic-6'.

Creating merge request for fix/epic-6 into main in devdashco/bluebut

https://gitlab.hostbun.cc/devdashco/bluebut/-/merge_requests/15

https://gitlab.hostbun.cc/devdashco/bluebut/-/work_items/6#note_510
done

=== status ===
=== HEAD ===
4761dfd2 feat(runtime): epic #6 cutover scaffolding — queue-worker dispatcher, gated legacy paths, read-side off run.ok
=== local vs origin ===
4761dfd29fa5649ef83c3e4e8deef39936ab527a
4761dfd29fa5649ef83c3e4e8deef39936ab527a
=== unpushed? ===
=== MR list ===
Showing 1 open merge request in devdashco/bluebut that match your search. (Page 1)

!15	devdashco/bluebut!15	epic #6: Skyvern runtime cutover scaffolding (gated dispatcher, queue SSE result, read-side off run.ok)	(main) ← (fix/epic-6)

state:	open

###### kernel Verb type + defineVerb ######
9:// adapter's RICH status vocabulary (ok/posted/joined/blocked/no_session/…), NOT the
17:// diagnostic, …) the runner/finalizer stamp on AFTER the adapter returns.
18:export const ResultBase = z.object({
24:export type VerbKind = 'read' | 'mutation'
25:export interface LedgerTarget { type: string | null; id: string | null }
27:export interface Verb<T extends ZodTypeAny = ZodTypeAny> {
29:  site: string
30:  kind: VerbKind
32:  adapter: string                                   // repo-relative adapter path the runner loads
34:  toArgv: (input: z.infer<T>) => string[]           // validated input → runner argv flags
35:  target: (input: z.infer<T>) => LedgerTarget       // ledger (target_type, target_id)
39:type VerbSpec<T extends ZodTypeAny> = Omit<Verb<T>, 'target'> & {
40:  target?: (input: z.infer<T>) => LedgerTarget
44:export function defineVerb<T extends ZodTypeAny>(spec: VerbSpec<T>): Verb<T> {
45:  const { name, site, kind, summary, adapter, input, toArgv, target, output } = spec
46:  if (!name || !site || !kind || !adapter || !input || !toArgv) {
50:    name, site, kind, summary, adapter, input, toArgv, output,
60:export function validateVerbOutput(
71:// A flag helper: emit `--name value` only when value is present. Keeps toArgv terse
73:export const flag = (name: string, value: unknown): string[] =>
77:// (undefined = leave the adapter's own default in place). Adapters read String(kwargs.x),
79:export const bflag = (name: string, value: unknown): string[] =>
85:// stays undefined so bflag can leave the adapter's own default in place.
86:export const boolish = z.preprocess(
###### worker.ts execution (adapter/argv/spawn) ######
1:// worker — claims queued jobs and runs them through the EXISTING runner.
3:// This is the industry-standard replacement for the backend's spawn-per-request +
4:// in-memory ACCOUNT_LOCK Map. The worker owns no job state: it polls claim() (the
5:// atomic RPC), spawns scripts/bluebut-adapter.ts for the claimed verb, and the
6:// runner writes the durable ledger row. A dead worker leaves rows queued/claimed
7:// to be reclaimed by the periodic reaper (reapStale, wired below), no lock Map,
11:// SQL (claim_job won't hand out a second job for a busy account), so the worker
13:import { spawn } from 'node:child_process'
16:import { getVerb } from './catalog.ts'
17:import { claim, markFailed, reapStale, type JobRow } from './queue.ts'
18:import { numEnv } from '../../adapters/_lib/num-env.ts'
22:const RUNNER = process.env.BLUEBUT_ADAPTER_BIN || join(REPO, 'scripts', 'bluebut-adapter.ts')
27:// The reaper runs on its own slower cadence (not every claim poll) — sweeping stale
28:// claimed/running rows is a cleanup, not a hot path, and a per-poll DB scan across the
42: * PURE: resolve a claimed job to its runner adapter + argv. THROWS on a malformed
43: * row — an unknown/empty verb (getVerb → ValidationError) or bad input (toArgv).
46: * jobs wedge the worker), and leave the row `claimed` (account lane blocked until
47: * the reaper). Exported so that guard is unit-tested without spawning a child.
49:export function jobSpawnArgs(job: JobRow): { name: string; adapter: string; argv: string[] } {
50:  const verb = getVerb(String(job.verb || job.cmd || ''))
51:  return { name: verb.name, adapter: verb.adapter, argv: verb.toArgv(job.input ?? {}) }
54:// Run one claimed job to completion. Resolves when the child exits (any code) —
59:  let name: string, adapter: string, argv: string[]
61:    ({ name, adapter, argv } = jobSpawnArgs(job))
70:    console.log(`[worker] DRY_RUN ${job.job_id} ${name} ${argv.join(' ')}`)
75:    const child = spawn(NODE, [RUNNER, adapter, ...argv], { env, cwd: REPO, stdio: 'inherit' })
77:      void markFailed(job.job_id, `worker spawn failed: ${err.message}`).catch(() => {})
84:        // `claimed`/`running` — and claim_job treats that as an active job, so the
85:        // account is blocked forever. markFailed is status-scoped (queued/claimed/
100:      job = await claim(WORKER_ID)
102:      console.warn(`[worker] claim error: ${(err as Error).message}`)
117:  // Reclaim rows orphaned in claimed/running by a worker that died before its
119:  // stays blocked until this sweeps. Runs regardless of claim activity.
122:      .then((n) => { if (n) console.warn(`[worker] reaped ${n} stale claimed/running row(s)`) })
134:const isEntry = !!process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href
142:    stop() // idempotent: flips `stopped`, halts new claims + reaper

###### enqueueRaw + enqueue (queue.ts 78-150) ######
     1	export async function enqueue(
     2	  verbName: string,
     3	  rawInput: unknown,
     4	  opts: EnqueueOpts = {},
     5	): Promise<{ job_id: string; status: 'queued' }> {
     6	  const verb = getVerb(verbName)          // unknown verb → ValidationError → 400 (app.ts onError)
     7	  // Bad args throw a ZodError, which the app.ts onError maps to 422 'invalid_input'
     8	  // with structured `issues` — richer than a flattened string, and the design's
     9	  // intended split (unknown verb → 400, bad args → 422). Do NOT wrap it.
    10	  const input = verb.input.parse(rawInput) as Record<string, unknown>
    11	  const tgt = verb.target(input)
    12	  const job_id = mintJobId()
    13	  await rest('POST', 'run', {
    14	    id: job_id,                                // `run` PK (epic #200 rename of job_id)
    15	    verb: verb.name,
    16	    cmd: verb.name,                            // keep `cmd` populated for existing ledger consumers
    17	    site: verb.site,
    18	    account: input.account ?? null,
    19	    target_type: tgt.type,
    20	    target_id: tgt.id,
    21	    status: 'queued',
    22	    input,
    23	    workflow_run_id: opts.batch_id ?? null,    // batch FK (epic #200 rename of batch_id)
    24	    not_before: opts.not_before ?? null,
    25	    priority: opts.priority ?? 0,
    26	    created_at: new Date().toISOString(),
    27	  }, { headers: { Prefer: 'return=minimal' }, label: 'queue' })
    28	  return { job_id, status: 'queued' }
    29	}
    30	
    31	// Pull --account (or the send-path aliases) out of a raw argv so a raw job still
    32	// takes the per-account claim lane, exactly like the typed path's input.account.
    33	function accountFromArgv(argv: string[]): string | null {
    34	  for (const f of ['--account', '--from-account', '--to-account']) {
    35	    const i = argv.indexOf(f)
    36	    if (i >= 0 && argv[i + 1]) return argv[i + 1]
    37	  }
    38	  return null
    39	}
    40	
    41	/** enqueueRaw — enqueue an arbitrary (site, cmd, argv) job WITHOUT the typed-verb
    42	 *  catalog. The UI's runCli() path speaks argv, not typed input; this carries the raw
    43	 *  argv on the run row (input._raw) so the worker spawns the runner IDENTICALLY to the
    44	 *  legacy /exec dispatcher (same adapterPath + argv). It is the durable-queue equivalent
    45	 *  of a POST /exec/stream call — the missing capability the UI cutover needs, so the
    46	 *  queue can run ANY adapter the dashboard fires, not only the 64 cataloged verbs. (#295)
    47	 *  A `job_id` may be supplied so the UI's own job identity threads through unchanged. */
    48	export async function enqueueRaw(
    49	  site: string,
    50	  cmd: string,
    51	  argv: string[],
    52	  opts: EnqueueOpts & { job_id?: string; account?: string | null } = {},
    53	): Promise<{ job_id: string; status: 'queued' }> {
    54	  const job_id = opts.job_id || mintJobId()
    55	  await rest('POST', 'run', {
    56	    id: job_id,                                // `run` PK (epic #200 rename of job_id)
    57	    verb: cmd,
    58	    cmd,
    59	    site,
    60	    account: opts.account ?? accountFromArgv(argv),
    61	    status: 'queued',
    62	    input: { _raw: true, site, cmd, argv },
    63	    run_type: 'job',
    64	    workflow_run_id: opts.batch_id ?? null,    // batch FK (epic #200 rename of batch_id)
    65	    not_before: opts.not_before ?? null,
    66	    priority: opts.priority ?? 0,
    67	    created_at: new Date().toISOString(),
    68	  }, { headers: { Prefer: 'return=minimal' }, label: 'queue-raw' })
    69	  return { job_id, status: 'queued' }
    70	}
    71	
    72	/** claim — atomically take the next runnable queued job for this worker. */
    73	export async function claim(workerId: string): Promise<JobRow | null> {
###### who calls enqueueRaw / raw route ######
server/jobs/queue.ts:118:/** enqueueRaw — enqueue an arbitrary (site, cmd, argv) job WITHOUT the typed-verb
server/jobs/queue.ts:125:export async function enqueueRaw(

1	import { spawn } from 'node:child_process'
     2	import { join, dirname } from 'node:path'
     3	import { fileURLToPath, pathToFileURL } from 'node:url'
     4	import { getVerb } from './catalog.ts'
     5	import { claim, markFailed, reapStale, type JobRow } from './queue.ts'
     6	import { numEnv } from '../../adapters/_lib/num-env.ts'
     7	
     8	const HERE = dirname(fileURLToPath(import.meta.url))
     9	const REPO = join(HERE, '..', '..')
    10	const RUNNER = process.env.BLUEBUT_ADAPTER_BIN || join(REPO, 'scripts', 'bluebut-adapter.ts')
    11	const NODE = process.execPath
    12	
    13	const POLL_MS = numEnv('BLUEBUT_QUEUE_POLL_MS', 1500)
    14	const MAX_CONCURRENT = numEnv('BLUEBUT_QUEUE_CONCURRENCY', 3)
    15	// The reaper runs on its own slower cadence (not every claim poll) — sweeping stale
    16	// claimed/running rows is a cleanup, not a hot path, and a per-poll DB scan across the
    17	// fleet would be wasteful. reapStale itself is time-boxed (BLUEBUT_QUEUE_REAP_STALE_MS).
    18	const REAP_POLL_MS = numEnv('BLUEBUT_QUEUE_REAP_POLL_MS', 60_000)
    19	const DRY_RUN = /^(1|true|yes)$/i.test(process.env.BLUEBUT_QUEUE_DRYRUN || '')
    20	// Graceful-drain window for a STANDALONE worker (worker.ts run directly). When the
    21	// worker is imported by server/http/index.mjs, that barrel owns SIGTERM/SIGINT and
    22	// calls our stop fn — the entry-point guard below only self-registers signals when
    23	// this module IS the process entry, so handlers are never double-installed.
    24	const SHUTDOWN_GRACE_MS = numEnv('BLUEBUT_SHUTDOWN_GRACE_MS', 8000)
    25	
    26	let running = 0
    27	let stopped = false
    28	
    29	/**
    30	 * PURE: resolve a claimed job to its runner adapter + argv. THROWS on a malformed
    31	 * row — an unknown/empty verb (getVerb → ValidationError) or bad input (toArgv).
    32	 * runJob MUST call this inside a try: a synchronous throw here would skip tick's
    33	 * `.finally(() => running -= 1)` and leak a concurrency slot forever (3 malformed
    34	 * jobs wedge the worker), and leave the row `claimed` (account lane blocked until
    35	 * the reaper). Exported so that guard is unit-tested without spawning a child.
    36	 */
    37	export function jobSpawnArgs(job: JobRow): { name: string; adapter: string; argv: string[] } {
    38	  const verb = getVerb(String(job.verb || job.cmd || ''))
    39	  return { name: verb.name, adapter: verb.adapter, argv: verb.toArgv(job.input ?? {}) }
    40	}
    41	
    42	// Run one claimed job to completion. Resolves when the child exits (any code) —
    43	// the runner owns the terminal ledger write, so the worker only frees its slot.
    44	// NEVER throws synchronously (see jobSpawnArgs) — a malformed job resolves after
    45	// being marked failed, so tick's slot-release `.finally` always runs.
    46	function runJob(job: JobRow): Promise<void> {
    47	  let name: string, adapter: string, argv: string[]
    48	  try {
    49	    ({ name, adapter, argv } = jobSpawnArgs(job))
    50	  } catch (err) {
    51	    void markFailed(job.job_id, `worker setup failed: ${(err as Error).message}`).catch(() => {})
    52	    return Promise.resolve()
    53	  }
    54	  const env: NodeJS.ProcessEnv = { ...process.env, BLUEBUT_JOB_ID: job.job_id }
    55	  if (job.batch_id) env.BLUEBUT_BATCH_ID = String(job.batch_id)
    56	
    57	  if (DRY_RUN) {
    58	    console.log(`[worker] DRY_RUN ${job.job_id} ${name} ${argv.join(' ')}`)
    59	    return Promise.resolve()
    60	  }
    61	
    62	  return new Promise<void>((resolve) => {
    63	    const child = spawn(NODE, [RUNNER, adapter, ...argv], { env, cwd: REPO, stdio: 'inherit' })
    64	    child.on('error', (err: Error) => {
    65	      void markFailed(job.job_id, `worker spawn failed: ${err.message}`).catch(() => {})
    66	      resolve()
    67	    })
    68	    child.on('exit', (code) => {
    69	      if (code !== 0) {
    70	        // The runner OWNS the terminal write on a clean run, but a child that dies
    71	        // before writing one (crash, OOM, SIGKILL) would leave the row stuck in
    72	        // `claimed`/`running` — and claim_job treats that as an active job, so the
    73	        // account is blocked forever. markFailed is status-scoped (queued/claimed/
    74	        // running only), so it's a no-op when the runner already wrote a terminal row.
    75	        console.warn(`[worker] ${job.job_id} exited ${code} — sweeping any non-terminal row to failed`)
    76	        void markFailed(job.job_id, `child exited ${code}`).catch(() => {})
    77	      }
    78	      resolve()
    79	    })
    80	  })
    81	}
    82	
    83	async function tick(): Promise<void> {

###### adapterPath def ######
server/http/runtime/spawn-env.d.mts:6:export function adapterPath(site: string, cmd: string): string | null;
--- body ---
server/http/runtime/spawn-env.mjs:51:const adapterPath = (site, cmd) => {
server/http/runtime/spawn-env.mjs:124:  adapterPath,
     1	const adapterPath = (site, cmd) => {
     2	  if (!/^[a-z0-9_-]+$/i.test(site) || !/^[a-z0-9_-]+$/i.test(cmd)) return null;
     3	  // The Messenger box is a self-contained root subsystem: its job entrypoints
     4	  // live in `messenger/jobs/<cmd>.js` (not `adapters/messenger/`). They still
     5	  // declare cli({ site:'facebook' }) so the runner applies every FB account
     6	  // protection. Route site='messenger' to the root box; everything else is a
     7	  // normal `adapters/<site>/<cmd>.js`.
     8	  const p = site === "messenger"
     9	    ? join(REPO_ROOT, "messenger", "jobs", `${cmd}.js`)
    10	    : join(ADAPTERS_DIR, site, `${cmd}.js`);
    11	  return existsSync(p) ? p : null;
    12	};
    13

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/jobs/worker.ts has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/jobs/worker.ts has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/routes/jobs.ts has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-epic-6/server/routes/jobs.ts has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-epic-6/ui/src/api/jobs.ts has been updated successfully. (file state is current in your context — no need to Read it back)

=== diff stat ===
 server/jobs/worker.ts | 16 +++++++++++
 server/routes/jobs.ts | 19 ++++++++++++-
 ui/src/api/jobs.ts    | 74 ++++++++++++++++++++++++++++++++++++---------------
 3 files changed, 86 insertions(+), 23 deletions(-)
=== run_type column usage (does typed enqueue/schema use it?) ===
db/migrations/entity/0047_rename_run_workflow_run.sql:15:-- `run_type` (bluebut#201/#19 EXECUTION_MODEL row 1 — the Skyvern-style run-kind discriminator,
db/migrations/entity/0047_rename_run_workflow_run.sql:45:-- 2.5) normalize DEAD legacy `run.status` values BEFORE the run_type backfill below.
db/migrations/entity/0047_rename_run_workflow_run.sql:51:--    on every re-boot the `UPDATE run SET run_type…` below touched those 257 rows, Postgres
db/migrations/entity/0047_rename_run_workflow_run.sql:67:-- 3) add + backfill `run.run_type` — additive, only touches the base table (never the compat view).
db/migrations/entity/0047_rename_run_workflow_run.sql:73:    EXECUTE 'ALTER TABLE run ADD COLUMN IF NOT EXISTS run_type text';
db/migrations/entity/0047_rename_run_workflow_run.sql:75:    UPDATE run SET run_type = CASE
db/migrations/entity/0047_rename_run_workflow_run.sql:83:    WHERE run_type IS NULL;
server/jobs/queue.ts:140:    run_type: 'job',
db/migrations/entity/0051_rename_run_columns_expand.sql:36:-- touched; the `job_schedule` table is a distinct concept, untouched; `run_type` VALUES stay data.
=== does 'run' table have run_type? (migrations) ===
db/0010_rename_run_workflow_run.sql
db/schema.ts
db/meta/0013_snapshot.json
db/migrations/entity/0047_rename_run_workflow_run.sql
db/meta/0010_snapshot.json
db/meta/0012_snapshot.json
db/migrations/entity/0051_rename_run_columns_expand.sql
db/meta/0014_snapshot.json
db/meta/0011_snapshot.json

diff --git a/server/jobs/worker.ts b/server/jobs/worker.ts
index 4b68828c..7156f524 100644
--- a/server/jobs/worker.ts
+++ b/server/jobs/worker.ts
@@ -16,6 +16,7 @@ import { fileURLToPath, pathToFileURL } from 'node:url'
 import { getVerb } from './catalog.ts'
 import { claim, markFailed, reapStale, type JobRow } from './queue.ts'
 import { numEnv } from '../../adapters/_lib/num-env.ts'
+import { adapterPath } from '../http/runtime/spawn-env.mjs'
 
 const HERE = dirname(fileURLToPath(import.meta.url))
 const REPO = join(HERE, '..', '..')
@@ -47,6 +48,21 @@ let stopped = false
  * the reaper). Exported so that guard is unit-tested without spawning a child.
  */
 export function jobSpawnArgs(job: JobRow): { name: string; adapter: string; argv: string[] } {
+  const input = (job.input ?? {}) as Record<string, unknown>
+  // epic #6 cutover: RAW jobs (enqueueRaw — the UI's argv-speaking /exec path). They
+  // carry `input._raw` with the (site, cmd, argv) verbatim instead of a typed verb, so
+  // resolve the adapter by PATH and pass argv through EXACTLY like the legacy /exec
+  // dispatcher — same adapterPath + same --format/json strip. A missing adapter throws,
+  // which runJob catches → markFailed (the row never wedges the account lane).
+  if (input._raw) {
+    const site = String(input.site || '')
+    const cmd = String(input.cmd || '')
+    const adapter = adapterPath(site, cmd)
+    if (!adapter) throw new Error(`no adapter at adapters/${site}/${cmd}.js (raw job ${job.job_id})`)
+    const rawArgv = Array.isArray(input.argv) ? (input.argv as unknown[]) : []
+    const argv = rawArgv.filter((a) => a != null && a !== '--format' && a !== 'json').map(String)
+    return { name: `${site}/${cmd}`, adapter, argv }
+  }
   const verb = getVerb(String(job.verb || job.cmd || ''))
   return { name: verb.name, adapter: verb.adapter, argv: verb.toArgv(job.input ?? {}) }
 }
diff --git a/server/routes/jobs.ts b/server/routes/jobs.ts
index 52374b16..52fe54b7 100644
--- a/server/routes/jobs.ts
+++ b/server/routes/jobs.ts
@@ -1,11 +1,12 @@
 // routes/jobs — the jobs resource. Thin: validate against the contract, call the
 // queue engine, validate out. No domain logic here (it lives in server/jobs/*).
 import { Hono } from 'hono'
+import { z } from 'zod'
 import { streamSSE } from 'hono/streaming'
 import { zValidator } from '@hono/zod-validator'
 import { EnqueueRequest } from '../contract/job.ts'
 import { catalogSummary } from '../jobs/catalog.ts'
-import { enqueue, getJob, listJobs, isLedgerTerminal, okForStatus } from '../jobs/queue.ts'
+import { enqueue, enqueueRaw, getJob, listJobs, isLedgerTerminal, okForStatus } from '../jobs/queue.ts'
 
 export const jobs = new Hono()
 
@@ -21,6 +22,22 @@ jobs.post('/', zValidator('json', EnqueueRequest), async (c) => {
   return c.json(await enqueue(verb, input), 202)
 })
 
+// epic #6 cutover: RAW enqueue — the durable-queue twin of the legacy POST /exec[/stream].
+// The UI's runCli() speaks argv, not typed input, so this carries (site, cmd, args) verbatim
+// onto the run row (input._raw) for the worker to spawn IDENTICALLY to the old inline /exec.
+// The UI threads its own job_id so its job identity is unchanged; batch_id groups the run.
+const ExecRequest = z.object({
+  site: z.string(),
+  cmd: z.string(),
+  args: z.array(z.string()).default([]),
+  job_id: z.string().optional(),
+  batch_id: z.string().optional(),
+})
+jobs.post('/exec', zValidator('json', ExecRequest), async (c) => {
+  const { site, cmd, args, job_id, batch_id } = c.req.valid('json')
+  return c.json(await enqueueRaw(site, cmd, args, { job_id, batch_id }), 202)
+})
+
 jobs.get('/', async (c) => {
   const n = Number(c.req.query('limit'))
   const limit = Number.isFinite(n) && n > 0 ? Math.min(n, 500) : 50   // NaN/negatives → default, cap at 500

fe18030e feat(runtime): epic #6 — wire the raw-argv cutover bridge (UI → /api/queue)
4761dfd2 feat(runtime): epic #6 cutover scaffolding — queue-worker dispatcher, gated legacy paths, read-side off run.ok

remote:   https://gitlab.hostbun.cc/devdashco/bluebut/-/merge_requests/15        
remote: 
To https://gitlab.hostbun.cc/devdashco/bluebut.git
   4761dfd2..fe18030e  fix/epic-6 -> fix/epic-6

- Updating merge request !15
✓ updated body
https://gitlab.hostbun.cc/devdashco/bluebut/-/merge_requests/15

https://gitlab.hostbun.cc/devdashco/bluebut/-/work_items/6#note_521
done