bluebut#52done

/home/philip/Documents/GitHub/bluebut-issue-52
fix/issue-52
---MRs---
No open merge requests match your search in devdashco/bluebut.

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);
31	const SCHED_TIMEOUT_MS = numEnv('BLUEBUT_SCHED_TIMEOUT_MS', 600_000);
32	const _schedInflight = new Set();   // schedule_ids currently firing — guards tick overlap
33	let _schedTicking = false;
34	// ANTI-DETECT: at most ONE browser-driving workflow SWEEP (scrape-new-members /
35	// get-group-feed / any needs_actor workflow) runs at a time. Firing both together would
36	// double the whole fleet's live FB activity in one window — a fleet-correlation tell. This
37	// holds the batch_id of the active sweep (or the "starting" sentinel between fire and
38	// batch id), cleared when that batch drains (workflowRunExec finally). A deferred sweep
39	// simply re-fires next tick. Broadcast batches serialize separately (_activeBatches).
40	// The reservation slot itself is sweep.batch (run-core), shared with the run loops.
41	
42	// ANTI-CLOCKWORK fire jitter: a deterministic per-cycle random hold (0..maxJitter) added
43	// on top of a schedule's interval, so an FB-touching sweep never fires at the same
44	// wall-clock every cycle (the "fires at 11:00:2x daily" robot tell). Deterministic —
45	// hashed from schedule_id + the current interval bucket — so it's STABLE within one
46	// pending window (no early/late flip-flop as the 60s tick re-checks) yet shifts every
47	// cycle. Browserless DB/LLM schedules don't jitter (no FB detection surface).
48	function schedFireJitterMs(s) {
49	  const intMs = Number(s.interval_minutes) * 60_000;
50	  if (!(intMs > 0)) return 0;
51	  if (!(s.workflow || s.campaign) && s.needs_actor === false) return 0;   // browserless → no jitter (#56.1: read EITHER flag)
52	  const maxJ = Math.min(intMs * 0.5, 60 * 60_000);               // ≤ half the interval, ≤ 60 min
53	  const last = s.last_run_at ? new Date(s.last_run_at).getTime() : 0;
54	  const bucket = Math.floor(last / intMs);
55	  let h = 2166136261;                                            // FNV-1a(schedule_id:bucket)
56	  const key = `${s.schedule_id}:${bucket}`;
57	  for (let i = 0; i < key.length; i++) { h ^= key.charCodeAt(i); h = Math.imul(h, 16777619); }
58	  return Math.floor(((h >>> 0) % 100000) / 100000 * maxJ);
59	}
60	
61	// Build the adapter argv for a schedule: the JIT-resolved account, optional
62	// --groups csv, verbatim extras. `acct` is resolved at fire time (never s.account
63	// directly) so a device-less / dark account is never dispatched.
64	function schedArgs(s, acct) {
65	  const out = [];
66	  if (acct) out.push("--account", acct);
67	  const groups = Array.isArray(s.groups) ? s.groups.filter(Boolean) : [];
68	  if (groups.length) out.push("--groups", groups.join(","));
69	  for (const a of Array.isArray(s.args) ? s.args : []) out.push(String(a));
70	  return out;
71	}
72	
73	// Fire ONE schedule as a real job — the headless twin of /exec/stream's task():
74	// same newJob + withAccountLock + childEnv spawn, minus the SSE response. The
75	// runner writes the durable ledger row (jobId = BLUEBUT_JOB_ID) for free.
76	function fireSchedule(s, jobId, acct) {
77	  const site = s.site || "facebook";
78	  const cmd = s.cmd;
79	  const adapter = adapterPath(site, cmd);
80	  if (!adapter) return Promise.resolve({ ok: false, jobId, error: `no adapter for ${site}/${cmd}` });
81	  const args = schedArgs(s, acct);
82	  const job = newJob({ id: jobId, account: acct, kind: "adapter", cmd: `${site}/${cmd}`, label: `⏰ ${s.label} · ${site}/${cmd}${acct ? ` (${acct})` : ""}` });
83	  job.traceJobId = jobId;   // == job.id (id-collapse); kept for the UI/analyze-job field
84	  job.scheduled = true;   // mute routine schedule ticks in Telegram (BLUEBUT_TG_SCHEDULED=1 re-enables)
85	  const cliArgs = [RUNNER, adapter, ...args.filter((a) => a != null).map(String)];
86	  return new Promise((resolve) => {
87	    const task = () => new Promise((done) => {
88	      // batch_id = schedule_id → every run this schedule fires is linkable in the ledger
89	      // (job.batch_id = schedule_id), so the UI can open each run from the schedule.
90	      const child = spawn(NODE, cliArgs, { env: childEnv(acct, jobId, s.schedule_id || null, site) });
91	      const killTimer = setTimeout(() => { try { child.kill("SIGKILL"); } catch { /* gone */ } }, SCHED_TIMEOUT_MS + 15_000);
92	      let out = "", ebuf = "";
93	      child.stdout.on("data", (d) => (out += d));
94	      child.stderr.on("data", (d) => {
95	        ebuf += d; let i;
96	        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); }
97	      });
98	      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(); });
99	      child.on("error", (e) => { clearTimeout(killTimer); job.ok = false; resolve({ ok: false, jobId, error: e.message }); done(); });
100	    });
101	    withAccountLock(acct, job, task, null);
102	  });
103	}
104	
105	// Resolve the TARGET SET for a workflow schedule. A per-account-discover cmd (the
106	// fleet sweeps / utilities — get-user-groups, list-pages, whoami…) needs NO group
107	// set (the target IS the account). A per-account-batch / per-target group cmd
108	// (get-group-feed, scrape-new-members, get-group…) needs a group id list: honour an
109	// explicit s.groups; else auto-resolve the natural set for the intent — every group
110	// we're a MEMBER of for a feed scrape, every group we OWN for an admin scrape — so
111	// a newly-joined group self-includes on the next tick (no per-group schedule churn).
112	async function resolveScheduleGroupIds(s) {
113	  const entry = workflowCmd(s.cmd);
114	  if (!entry || entry.fanMode === "per-account-discover") return [];
115	  const explicit = (Array.isArray(s.groups) ? s.groups : []).map(String).filter(Boolean);
116	  if (explicit.length) return explicit;
117	  // BIGGEST GROUPS FIRST: order candidates by member_count desc so the highest-value
118	  // groups get scraped before a sweep can be cut short (block / stop / actor blip). The
119	  // groupIds order is preserved all the way to batch_target.idx (resolveWorkflowTargets
120	  // maps over groupIds in order) → each lane processes its biggest groups first.
121	  const dedupInOrder = (rows, key = "fb_group_id") => {
122	    const seen = new Set(); const out = [];
123	    for (const r of rows) { const id = String(r?.[key] ?? ""); if (id && !seen.has(id)) { seen.add(id); out.push(id); } }
124	    return out;
125	  };
126	  try {
127	    if (entry.intent === "scrape-admin" || entry.intent === "insights") {
128	      const rows = (await groupsRepo.select("select=fb_group_id&owned_by_us=is.true&order=member_count.desc.nullslast")) || [];
129	      return dedupInOrder(rows);
130	    }
131	    // feed-scrape / group sync — every group any of our accounts is a member of, biggest
132	    // first (member_count lives on fb_group → embed + order by the embedded column).
133	    const rows = (await dbRest("GET", "fb_account_group?select=fb_group_id,fb_group!inner(member_count)&is_member=is.true&order=fb_group(member_count).desc.nullslast", null, { silent: true })) || [];
134	    return dedupInOrder(rows);
135	  } catch { return []; }
136	}
137	
138	// Fire ONE workflow schedule: resolve the target set, PARTITION the intent across the
139	// USABLE fleet (usync-alive ∩ device-runnable; an optional single pin narrows it),
140	// then workflowStart — the durable, parallel-lane executor owns the rest. This is the
141	// agnostic model: a scheduled job is "all accounts that fit criteria", not one profile.
142	// workflowStart MINTS the batch id (passing one makes it a resume → PATCH-not-INSERT →
143	// batch_target FK fail), so we let it and return the id. Returns { ok, targets, lanes, batchId, reason }.
144	async function fireWorkflowSchedule(s) {
145	  const entry = workflowCmd(s.cmd);
146	  if (!entry) return { ok: false, targets: 0, reason: `'${s.cmd}' is not a fannable workflow cmd` };
147	  const groupIds = await resolveScheduleGroupIds(s);
148	  // Resolve WHO can run it: usync-alive fleet (fail-open to device pool on a checker
149	  // outage). A single-account pin narrows the set; null accounts ⇒ pass the whole
150	  // usable fleet as the explicit subset so resolveWorkflowPlan fans across exactly it.
151	  const usable = await usableDrivableFleet();
152	  const pin = (s.account || "").trim();
153	  let accounts;
154	  if (!pin) accounts = usable == null ? null : [...usable];             // null = fall back to device pool
155	  else if (usable == null || usable.has(pin)) accounts = [pin];         // pin alive (or checker down) → honor it
156	  else {
157	    // #56.2: a pin that isn't usync-alive must NOT brick the flow forever (it would
158	    // re-fire every tick, fail identically, and — because it returns `skipped` not
159	    // `failed` — never trip the stall monitor). Mirror resolveScheduleActor: treat the
160	    // pin as a preference, autopick the usable fleet instead of returning [].
161	    accounts = [...usable];
162	    console.log(`[backend] workflow schedule ${s.schedule_id}: pinned '${pin}' not usync-alive → autopick usable fleet (${accounts.length}) [pinned_not_alive_autopicked]`);
163	  }
164	  if (Array.isArray(accounts) && accounts.length === 0) {
165	    return { ok: false, targets: 0, reason: "no usync-alive accounts" };
166	  }
167	  let plan;
168	  try {
169	    plan = await resolveWorkflowPlan({ cmd: s.cmd, accounts, groupIds, capPerAccount: 500 });
170	  } catch (e) { return { ok: false, targets: 0, reason: `plan failed: ${e?.message || e}` }; }
171	  if (!plan.targets.length) {
172	    return { ok: false, targets: 0, reason: plan.drivableSize === 0 ? "no drivable accounts" : "no eligible targets" };
173	  }
174	  const lanes = plan.partitions.size;
175	  const batchId = await workflowStart({
176	    cmd: s.cmd, targets: plan.targets,
177	    kind: entry.intent, scheduleId: s.schedule_id,
178	    label: `⏰ ${s.label} · ${plan.targets.length} target${plan.targets.length === 1 ? "" : "s"} · ${lanes} account${lanes === 1 ? "" : "s"}`,
179	    pacing: { maxConcurrentAccounts: entry.defaultConcurrency ?? 3, capPerAccount: 500 },
180	  });
181	  return { ok: true, targets: plan.targets.length, lanes, batchId, reason: entry.intent };
182	}
183	
184	// Atomic compare-and-swap claim on a schedule's interval slot (issue #43.F3). The
185	// in-memory guards (_schedInflight/_schedTicking) only serialize WITHIN one process;
186	// across a rolling redeploy two containers overlap and both pass isDue, and the old
187	// stamp was a blind fire-and-forget PATCH with no WHERE guard → both fire the same
188	// `*/5` sweep (the ~1400-keepalive runaway). This stamps last_run_at ONLY if the row
189	// still holds the exact value we read (eq guard, or is.null for a never-run row) and
190	// returns whether WE won — the loser skips instead of double-firing. It also closes
191	// the single-process re-fire storm: a DB blip that dropped the old blind stamp left
192	// last_run_at un-advanced so the schedule re-fired every tick; here the winning CAS
193	// advance is awaited and confirmed before we fire.
194	async function casClaimSchedule(scheduleId, oldLra, body) {
195	  const guard = oldLra ? `last_run_at=eq.${encodeURIComponent(oldLra)}` : "last_run_at=is.null";
196	  try {
197	    const rows = await dbRest("PATCH",
198	      `run_schedule?schedule_id=eq.${encodeURIComponent(scheduleId)}&${guard}`,
199	      body, { silent: true, headers: { Prefer: "return=representation" } });
200	    return Array.isArray(rows) && rows.length > 0;
201	  } catch (e) {
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.
204	    console.error(`[backend] schedule CAS claim ${scheduleId} failed: ${e?.message || e}`);
205	    return false;
206	  }
207	}
208	
209	// epic #6 cutover: the UI/exec dispatch path moved to the durable queue worker
210	// (bridge.ts /exec is gated dormant under BLUEBUT_QUEUE_WORKER=1). The in-process
211	// SCHEDULER stays live for now: it fires recurring run_schedule rows via fireSchedule()
212	// (spawn), a lane the queue worker does NOT cover — there is no schedule→queue producer
213	// yet, so gating this dormant would silently stop every recurring job. Migrating the
214	// scheduler onto the queue is tracked as a follow-up; until then it keeps driving.
215	async function tickSchedules() {
216	  if (_schedTicking) return;
217	  _schedTicking = true;
218	  try {
219	    const rows = (await dbRest("GET", "run_schedule?enabled=eq.true")) || [];
220	    const now = new Date();
221	    for (const s of rows) {
222	      if (_schedInflight.has(s.schedule_id)) continue;   // still running from a prior tick
223	      if (!isDue(s, now, schedFireJitterMs(s))) continue;   // interval + anti-clockwork jitter
224	      // Serialize browser-driving sweeps: if one is already draining, defer this one to a
225	      // later tick (don't stamp last_run — its jitter re-rolls and it re-fires when free).
226	      const isBrowserSweep = (s.workflow || s.campaign) && s.needs_actor !== false;
227	      if (isBrowserSweep && sweep.batch) continue;
228	      _schedInflight.add(s.schedule_id);
229	      if (isBrowserSweep) sweep.batch = "starting";   // reserve synchronously (closes the same-tick race)
230	
231	      // ── WORKFLOW schedule — fan the intent across the whole drivable fleet ──────
232	      // The agnostic model: "run this for ALL accounts that fit the criteria", not
233	      // one pinned profile. We DON'T resolve a single actor here (workflowRunExec
234	      // partitions the targets across every drivable account itself); we just launch
235	      // the durable batch and let it drain in parallel lanes. last_run_at is stamped
236	      // at START so the interval gate holds across a restart; last_run_job_id = the
237	      // workflow batch id (the run history links via batch.meta->>schedule_id).
238	      // #56.1: read EITHER flag during the campaign→workflow expand/contract window.
239	      // Both are NOT NULL default false (schema.ts:922-923) so `??` would be a no-op;
240	      // OR fires the fleet-fan path if either is set, surviving the `campaign` drop.
241	      if (s.workflow || s.campaign) {
242	        const startIso = new Date().toISOString();
243	        const next = nextRunAt({ ...s, last_run_at: startIso }, now);
244	        // CAS-claim the interval slot BEFORE firing (issue #43.F3): stamp last_run_at
245	        // at START (holds the interval gate across a restart) only if no other tick/
246	        // process already advanced it. Lost the race → release the sweep reservation
247	        // + inflight guard and skip, so the sibling never double-fires the sweep.
248	        const won = await casClaimSchedule(s.schedule_id, s.last_run_at ?? null,
249	          { last_run_at: startIso, next_run_at: next ? next.toISOString() : null, last_run_status: "running", last_run_account: null, updated_at: startIso });
250	        if (!won) {
251	          if (isBrowserSweep && sweep.batch === "starting") sweep.batch = null;
252	          _schedInflight.delete(s.schedule_id);
253	          continue;
254	        }
255	        fireWorkflowSchedule(s)
256	          .then((res) => {
257	            console.log(`[backend] workflow schedule fire: ${s.schedule_id} (${s.label}) → ${s.cmd} · ${res.ok ? `${res.targets} target(s) across ${res.lanes} lane(s) · batch ${res.batchId}` : `SKIPPED (${res.reason})`}`);
258	            // Hand the sweep reservation the real batch id (cleared when it drains), or
259	            // RELEASE it here if the sweep never started, so a deferred sibling can run.
260	            if (isBrowserSweep) sweep.batch = res.ok ? res.batchId : null;
261	            // A workflow that resolved 0 targets isn't a run. On the FIRST miss, restore
262	            // last_run so it re-fires next tick once accounts/targets appear (no dead
263	            // 'failed' churn). But if the PRIOR fire was already a 0-target miss
264	            // (last_run_status === "skipped"), DON'T restore — keep the CAS-stamped
265	            // startIso interval gate so the schedule backs off to its normal cadence
266	            // instead of re-resolving the whole fleet every single tick forever (a
267	            // never-satisfiable schedule would otherwise hot-loop, #43.F3 follow-up).
268	            const wasMiss = s.last_run_status === "skipped";
269	            return dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
270	              res.ok ? { last_run_status: "ok", last_run_job_id: res.batchId, updated_at: new Date().toISOString() }
271	                     : wasMiss ? { last_run_status: "skipped", updated_at: new Date().toISOString() }
272	                               : { last_run_at: s.last_run_at ?? null, next_run_at: s.next_run_at ?? null, last_run_status: "skipped", updated_at: new Date().toISOString() },
273	              { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {});
274	          })
275	          .catch((e) => { if (isBrowserSweep && sweep.batch === "starting") sweep.batch = null; console.error(`[backend] workflow schedule ${s.schedule_id} error: ${e?.message || e}`); })
276	          .finally(() => _schedInflight.delete(s.schedule_id));
277	        continue;
278	      }
279	
280	      // Actor selection depends on whether the cmd drives a browser. BROWSERLESS
281	      // schedules (needs_actor=false — classify-posts / llm-analyze-and-extract, the
282	      // Strategy.PUBLIC DB+LLM funnel jobs) need NO cloakbox account: resolving one
283	      // would (a) burn a drivable-account rotation slot on a job that never opens a
284	      // browser and (b) get SKIPPED whenever the 5-account drivable pool blips —
285	      // starving the funnel for no reason. They fire with acct=null (no --account).
286	      // BROWSER schedules keep JIT actor selection — resolve WHICH drivable account
287	      // runs this now, never the pinned s.account (a device-less account is doomed at
288	      // cdp_load). A null actor → skip WITHOUT stamping last_run_at, so the schedule
289	      // re-fires the moment an account becomes runnable (no DB churn, no dead job).
290	      let acct = null;
291	      let actorReason = "browserless";
292	      if (s.needs_actor !== false) {
293	        let actor;
294	        try { actor = await resolveScheduleActor(s, { rest: dbRest, now }); }
295	        catch (e) { actor = { account: null, reason: `resolve_error: ${e?.message || e}` }; }
296	        if (!actor.account) {
297	          console.log(`[backend] schedule ${s.schedule_id} (${s.label}): no runnable actor (${actor.reason}) — skipping`);
298	          _schedInflight.delete(s.schedule_id);
299	          continue;
300	        }
301	        acct = actor.account;
302	        actorReason = actor.reason;
303	      }
304	      const jobId = mintJobId();
305	      const startIso = new Date().toISOString();
306	      const next = nextRunAt({ ...s, last_run_at: startIso }, now);
307	      console.log(`[backend] schedule fire: ${s.schedule_id} (${s.label}) → ${s.site || "facebook"}/${s.cmd} as ${acct || "(browserless)"} [${actorReason}] job ${jobId}`);
308	      // CAS-claim the interval slot at START (issue #43.F3) — only if no other tick/
309	      // process already advanced last_run_at. Holds the interval gate across a bridge
310	      // restart mid-run AND prevents a redeploy-overlap double-fire; a crash can't
311	      // cause a re-fire storm. Lost the race → skip (a sibling already fired this
312	      // interval). The terminal status/ok is patched when the job finishes.
313	      const won = await casClaimSchedule(s.schedule_id, s.last_run_at ?? null,
314	        { last_run_at: startIso, next_run_at: next ? next.toISOString() : null, last_run_status: "running", last_run_job_id: jobId, last_run_account: acct, updated_at: startIso });
315	      if (!won) { _schedInflight.delete(s.schedule_id); continue; }
316	      fireSchedule(s, jobId, acct)
317	        .then((res) => dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
318	          { last_run_status: res.ok ? "ok" : "failed", updated_at: new Date().toISOString() },
319	          { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {}))
320	        .catch((e) => console.error(`[backend] schedule ${s.schedule_id} error: ${e?.message || e}`))
321	        .finally(() => _schedInflight.delete(s.schedule_id));
322	    }
323	  } catch (e) { console.error(`[backend] schedule tick failed: ${e?.message || e}`); }
324	  finally { _schedTicking = false; }
325	}
326	// HOSTED-ONLY. The recurring scheduler fires prod job_schedule rows — including
327	// BROWSER-driven ones that cdp_load a live cloakbox account. Both the prod bridge
328	// (bluebut-bridge.blpk.cc) and a laptop `npm run backend` read the SAME table, so
329	// without this gate a local dev bridge ALSO fires them, opening a cloakbox from
330	// your machine and duplicating prod's runs. API_TOKEN is set on the hosted
331	// bridge and unset locally, so it's the hosted/local discriminator; force it on
332	// locally with BLUEBUT_FORCE_SCHEDULER=1 if you really mean to test the tick.
333	// API_NO_LISTEN=1 means this is a smoke/import-check (or CI), NOT a serving bridge — it
334	// must NEVER tick the scheduler, which fires REAL prod job_schedule rows the instant it
335	// evaluates (tickSchedules() below runs immediately). Without this guard, `npm run
336	// smoke:backend` / `npm run check` drive prod schedules on a bare import (inv #23).
337	// BLUEBUT_NO_SCH

=== BLUEBUT_QUEUE_WORKER refs ===
server/http/routes/bridge.ts
server/http/index.mjs
server/http/runtime/scheduler.mjs
=== queue worker files ===
server/jobs/queue.ts
server/jobs/queue-status.test.mjs
server/jobs/enqueue-validation.test.mjs
server/jobs/enqueue-broadcast.ts
=== updateRunStatus ===
adapters/_lib/taxonomy.ts
server/jobs/queue.ts
server/jobs/queue-status.test.mjs
ontology/ontology.ts
server/contract/job.ts

1	// queue — the job queue, built on the `job` ledger table via the db.js seam ONLY.
2	//
3	// enqueue() inserts a `queued` row; claim() calls the claim_job RPC (atomic
4	// FOR UPDATE SKIP LOCKED + per-account unique-index lock, migration 0001). No new
5	// DB client, no in-memory Map: the table IS the queue and the job identity, so it
6	// survives any process restart. The runner (spawned by the worker) merge-upserts
7	// the same job_id to running→terminal, so a claimed row upgrades in place.
8	// db.ts is the untyped legacy _lib seam — treat `rest` as loose until it's typed on touch (invariant #20)
9	import { rest as _rest } from '../../adapters/_lib/db/db.ts'
10	const rest = _rest as (method: string, path: string, body?: unknown, opts?: Record<string, unknown>) => Promise<any>
11	import { getVerb } from './catalog.ts'
12	import { mintJobId } from './job-id.ts'
13	import { numEnv } from '../../adapters/_lib/num-env.ts'
14	
15	// How long a row may sit in claimed/running before the reaper considers it stale
16	// (worker process died without ever reaching the child.on('exit') handler that
17	// normally sweeps it via markFailed — see reapStale below). Env-tunable, matching
18	// this repo's numEnv convention (adapters/_lib/num-env.ts).
19	const REAP_STALE_MS = numEnv('BLUEBUT_QUEUE_REAP_STALE_MS', 15 * 60 * 1000)
20	
21	// The durable `job` ledger's status vocabulary — the ONE source for "is this run
22	// still going / done?" across every ledger reader (SSE guard, reaper, worker).
23	// (Distinct from taxonomy.ts RunStatus, the Skyvern vocab the ledger has not yet
24	// migrated onto — see docs/skyvern-gap-analysis.md §2. Until that lands, this is
25	// the single writer/reader of the ledger's own terms — no re-hardcoding per site.)
26	//
27	// TERMINAL must list EVERY status the runner banks on a finished row, or a reader
28	// treats a done job as still-running. `adapter-finalize.jobLifecycleStatus` writes
29	// SUCCESS as 'ok' (not 'done') and the benign terminals 'skipped'/'partial'/
30	// 'content_unavailable' — omitting them made `/jobs/:id/events` poll a succeeded
31	// job for the full 600s instead of closing. 'done' is kept for the in-mem bridge
32	// map; the taxonomy terminals are included so a future write of them still reads
33	// terminal. (LEDGER_ACTIVE also scopes markFailed's non-terminal sweep, below.)
34	export const LEDGER_ACTIVE = ['queued', 'claimed', 'running'] as const
35	export const LEDGER_TERMINAL = [
36	  'ok', 'done', 'completed', 'failed', 'blocked', 'skipped', 'partial',
37	  'content_unavailable', 'terminated', 'timed_out', 'canceled',
38	] as const
39	export const isLedgerTerminal = (status: string): boolean =>
40	  (LEDGER_TERMINAL as readonly string[]).includes(status)
41	
42	// A job row as PostgREST returns it (loosely typed — the DB is the schema).
43	export interface JobRow {
44	  id?: string          // the `run` PK (epic #200 rename of job_id); normalized → job_id below
45	  job_id: string
46	  verb?: string | null
47	  cmd?: string | null
48	  account?: string | null
49	  status: string
50	  input?: unknown
51	  batch_id?: string | null
52	  ok?: boolean | null
53	  [k: string]: unknown
54	}
55	
56	// epic #200 renamed the ledger table job→run: the PK column is now `id` (was `job_id`)
57	// and the batch FK is `workflow_run_id` (was `batch_id`). The queue's public JobRow shape
58	// keeps job_id/batch_id, so EVERY row read back from `run` is normalized HERE at the seam
59	// — worker.ts, the /api/queue routes, and the tests are untouched by the column rename.
60	function normalizeRun(row: Record<string, unknown>): JobRow {
61	  return {
62	    ...row,
63	    job_id: String(row.id ?? row.job_id ?? ''),
64	    batch_id: (row.workflow_run_id ?? row.batch_id ?? null) as string | null,
65	  } as JobRow
66	}
67	
68	/** Optional ledger-level knobs an enqueue can set (issue #217): a batch grouping, the
69	 *  `not_before` due-time (pacing — the row is invisible to claim_job_v2 until due), and
70	 *  `priority` (claim order). All optional; omitting them = legacy FIFO immediate-claim. */
71	export interface EnqueueOpts {
72	  batch_id?: string | null
73	  not_before?: string | null   // ISO timestamptz; a queued row won't be claimed before this
74	  priority?: number            // higher = claimed first (default 0)
75	}
76	
77	/** enqueue — validate input against the verb, insert a queued ledger row. */
78	export async function enqueue(
79	  verbName: string,
80	  rawInput: unknown,
81	  opts: EnqueueOpts = {},
82	): Promise<{ job_id: string; status: 'queued' }> {
83	  const verb = getVerb(verbName)          // unknown verb → ValidationError → 400 (app.ts onError)
84	  // Bad args throw a ZodError, which the app.ts onError maps to 422 'invalid_input'
85	  // with structured `issues` — richer than a flattened string, and the design's
86	  // intended split (unknown verb → 400, bad args → 422). Do NOT wrap it.
87	  const input = verb.input.parse(rawInput) as Record<string, unknown>
88	  const tgt = verb.target(input)
89	  const job_id = mintJobId()
90	  await rest('POST', 'run', {
91	    id: job_id,                                // `run` PK (epic #200 rename of job_id)
92	    verb: verb.name,
93	    cmd: verb.name,                            // keep `cmd` populated for existing ledger consumers
94	    site: verb.site,
95	    account: input.account ?? null,
96	    target_type: tgt.type,
97	    target_id: tgt.id,
98	    status: 'queued',
99	    input,
100	    workflow_run_id: opts.batch_id ?? null,    // batch FK (epic #200 rename of batch_id)
101	    not_before: opts.not_before ?? null,
102	    priority: opts.priority ?? 0,
103	    created_at: new Date().toISOString(),
104	  }, { headers: { Prefer: 'return=minimal' }, label: 'queue' })
105	  return { job_id, status: 'queued' }
106	}
107	
108	// Pull --account (or the send-path aliases) out of a raw argv so a raw job still
109	// takes the per-account claim lane, exactly like the typed path's input.account.
110	function accountFromArgv(argv: string[]): string | null {
111	  for (const f of ['--account', '--from-account', '--to-account']) {
112	    const i = argv.indexOf(f)
113	    if (i >= 0 && argv[i + 1]) return argv[i + 1]
114	  }
115	  return null
116	}
117	
118	/** enqueueRaw — enqueue an arbitrary (site, cmd, argv) job WITHOUT the typed-verb
119	 *  catalog. The UI's runCli() path speaks argv, not typed input; this carries the raw
120	 *  argv on the run row (input._raw) so the worker spawns the runner IDENTICALLY to the
121	 *  legacy /exec dispatcher (same adapterPath + argv). It is the durable-queue equivalent
122	 *  of a POST /exec/stream call — the missing capability the UI cutover needs, so the
123	 *  queue can run ANY adapter the dashboard fires, not only the 64 cataloged verbs. (#295)
124	 *  A `job_id` may be supplied so the UI's own job identity threads through unchanged. */
125	export async function enqueueRaw(
126	  site: string,
127	  cmd: string,
128	  argv: string[],
129	  opts: EnqueueOpts & { job_id?: string; account?: string | null } = {},
130	): Promise<{ job_id: string; status: 'queued' }> {
131	  const job_id = opts.job_id || mintJobId()
132	  await rest('POST', 'run', {
133	    id: job_id,                                // `run` PK (epic #200 rename of job_id)
134	    verb: cmd,
135	    cmd,
136	    site,
137	    account: opts.account ?? accountFromArgv(argv),
138	    status: 'queued',
139	    input: { _raw: true, site, cmd, argv },
140	    run_type: 'job',
141	    workflow_run_id: opts.batch_id ?? null,    // batch FK (epic #200 rename of batch_id)
142	    not_before: opts.not_before ?? null,
143	    priority: opts.priority ?? 0,
144	    created_at: new Date().toISOString(),
145	  }, { headers: { Prefer: 'return=minimal' }, label: 'queue-raw' })
146	  return { job_id, status: 'queued' }
147	}
148	
149	/** claim — atomically take the next runnable queued job for this worker. */
150	export async function claim(workerId: string): Promise<JobRow | null> {
151	  const rows = await rest('POST', 'rpc/claim_job', { p_worker: workerId }, { silent: true })
152	  return Array.isArray(rows) && rows.length ? normalizeRun(rows[0]) : null
153	}
154	
155	/** Fetch one job row by id (for GET /jobs/:id). */
156	export async function getJob(job_id: string): Promise<JobRow | null> {
157	  const rows = await rest('GET', `run?id=eq.${encodeURIComponent(job_id)}&limit=1`, null, { silent: true })
158	  return Array.isArray(rows) && rows.length ? normalizeRun(rows[0]) : null
159	}
160	
161	/** Most-recent jobs (for GET /jobs). */
162	export async function listJobs({ limit = 50 }: { limit?: number } = {}): Promise<JobRow[]> {
163	  const rows = await rest('GET', `run?order=created_at.desc&limit=${Number(limit)}`, null, { silent: true })
164	  return Array.isArray(rows) ? rows.map(normalizeRun) : []
165	}
166	
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
172	// banked terminal is never clobbered.
173	//
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-
176	// finalize (scripts/_lib/adapter-finalize.ts) + pace-loop writers still write
177	// directly today; routing THEM through this funnel is the supervised-live
178	// remainder of #19 Phase 1 (inv #24 — it needs a real drive, not a headless merge).
179	
180	// Statuses that mean the run SUCCEEDED (ok=true). A benign terminal
181	// (skipped/partial/content_unavailable) is NOT a success — the runner banks it
182	// with ok=false — so okForStatus returns false for it, matching the ledger's data.
183	const LEDGER_SUCCESS = ['ok', 'done', 'completed'] as const
184	
185	/** Derive the legacy `ok` boolean a ledger status implies: true for a success
186	 *  terminal, false for any other terminal (failure or benign), null while active. */
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
190	  return null   // queued / claimed / running — undecided
191	}
192	
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> {
195	  const now = new Date().toISOString()
196	  const body: Record<string, unknown> = { status, ok: okForStatus(status), updated_at: now }
197	  if (isLedgerTerminal(status)) body.ended_at = now
198	  if (reason != null) body.aborted_reason = String(reason).slice(0, 500)
199	  return body
200	}
201	
202	/** THE single writer for a `run` row's status transition (queue path). Scoped to
203	 * non-terminal rows so a banked terminal is NEVER clobbered — the PATCH matches no
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> {
206	  await rest('PATCH', `run?id=eq.${encodeURIComponent(job_id)}&status=in.(${LEDGER_ACTIVE.join(',')})`,
207	    statusPatch(status, reason), { headers: { Prefer: 'return=minimal' }, silent: true })
208	}
209	
210	/** Mark a claimed/running job failed when the worker itself can't finish it (spawn
211	 * error, child died before writing a terminal row). Routes through the single
212	 * updateRunStatus funnel; the non-terminal scope means a runner that already wrote
213	 * a terminal row is NEVER clobbered. Guards against a job stuck in `claimed` forever
214	 * (which permanently blocks its account via claim_job). */
215	export async function markFailed(job_id: string, reason: unknown): Promise<void> {
216	  await updateRunStatus(job_id, 'failed', reason)
217	}
218	
219	/** Sweep `claimed`/`running` rows whose worker process died WITHOUT ever hitting
220	 * markFailed (the live worker's child.on('exit') path, worker.ts) — e.g. the worker
221	 * itself was killed. Those rows have no other reclaim path: LEDGER_ACTIVE scopes
222	 * them as active forever, and the account-lane unique index (job_one_active_per_account)
223	 * stays held, permanently blocking that account. Time-boxed off `claimed_at` (set by
224	 * claim_job, db/0002_claim_job.sql) — any row older than REAP_STALE_MS is reclaimed to
225	 * `failed` via the same markFailed funnel, so a row already finished by its runner is
226	 * never clobbered (updateRunStatus's non-terminal scope). Called once per worker tick
227	 * (worker.ts) — safe to call repeatedly/concurrently since each markFailed is a scoped
228	 * no-op once a row is terminal. Returns the count of rows reaped (for logging). */
229	export async function reapStale(staleMs: number = REAP_STALE_MS): Promise<number> {
230	  const cutoff = new Date(Date.now() - staleMs).toISOString()
231	  const rows = await rest(
232	    'GET',
233	    `run?status=in.(claimed,running)&claimed_at=lt.${encodeURIComponent(cutoff)}&select=id,claimed_at`,
234	    null,
235	    { silent: true },
236	  )
237	  if (!Array.isArray(rows) || !rows.length) return 0
238	  await Promise.all(rows.map((row: Record<string, unknown>) => {
239	    const job_id = String(row.id ?? row.job_id ?? '')
240	    return job_id ? markFailed(job_id, 'reaped: stale claim').catch(() => {}) : Promise.resolve()
241	  }))
242	  return rows.length
243	}
244

catalog-integrity.test.mjs
catalog.test.mjs
catalog.ts
enqueue-broadcast.ts
enqueue-validation.test.mjs
job-id.ts
job-stream.test.mjs
job-stream.ts
queue-status.test.mjs
queue.ts
verbs
worker.test.mjs
worker.ts
=== worker ===
server/jobs/worker.test.mjs
server/jobs/worker.ts
=== dormancy gate in bridge.ts ===
server/http/index.mjs:419:  // until BLUEBUT_QUEUE_WORKER=1 is set (after confirming claim_job exists + a
server/http/index.mjs:422:  const queueWorkerOn = process.env.BLUEBUT_QUEUE_WORKER === '1'
server/http/index.mjs:423:    || (process.env.NODE_ENV !== 'production' && process.env.BLUEBUT_QUEUE_WORKER !== '0');
server/http/index.mjs:427:  if (!queueWorkerOn) console.log('[backend] queue worker OFF (set BLUEBUT_QUEUE_WORKER=1 to drain /api/queue jobs)');
server/http/routes/bridge.ts:421:// epic #6 cutover: when the durable queue worker owns dispatch (BLUEBUT_QUEUE_WORKER=1),
server/http/routes/bridge.ts:425:const QUEUE_WORKER_ON = process.env.BLUEBUT_QUEUE_WORKER === "1";

1	// worker — claims queued jobs and runs them through the EXISTING runner.
2	//
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,
8	// no bridge process required.
9	//
10	// Concurrency is bounded by MAX_CONCURRENT; per-account exclusion is enforced in
11	// SQL (claim_job won't hand out a second job for a busy account), so the worker
12	// never has to reason about it.
13	import { spawn } from 'node:child_process'
14	import { join, dirname } from 'node:path'
15	import { fileURLToPath, pathToFileURL } from 'node:url'
16	import { getVerb } from './catalog.ts'
17	import { claim, markFailed, reapStale, type JobRow } from './queue.ts'
18	import { publishLine, endStream } from './job-stream.ts'
19	import { numEnv } from '../../adapters/_lib/num-env.ts'
20	import { adapterPath } from '../http/runtime/spawn-env.mjs'
21	
22	const HERE = dirname(fileURLToPath(import.meta.url))
23	const REPO = join(HERE, '..', '..')
24	const RUNNER = process.env.BLUEBUT_ADAPTER_BIN || join(REPO, 'scripts', 'bluebut-adapter.ts')
25	const NODE = process.execPath
26	
27	const POLL_MS = numEnv('BLUEBUT_QUEUE_POLL_MS', 1500)
28	const MAX_CONCURRENT = numEnv('BLUEBUT_QUEUE_CONCURRENCY', 3)
29	// The reaper runs on its own slower cadence (not every claim poll) — sweeping stale
30	// claimed/running rows is a cleanup, not a hot path, and a per-poll DB scan across the
31	// fleet would be wasteful. reapStale itself is time-boxed (BLUEBUT_QUEUE_REAP_STALE_MS).
32	const REAP_POLL_MS = numEnv('BLUEBUT_QUEUE_REAP_POLL_MS', 60_000)
33	const DRY_RUN = /^(1|true|yes)$/i.test(process.env.BLUEBUT_QUEUE_DRYRUN || '')
34	// Graceful-drain window for a STANDALONE worker (worker.ts run directly). When the
35	// worker is imported by server/http/index.mjs, that barrel owns SIGTERM/SIGINT and
36	// calls our stop fn — the entry-point guard below only self-registers signals when
37	// this module IS the process entry, so handlers are never double-installed.
38	const SHUTDOWN_GRACE_MS = numEnv('BLUEBUT_SHUTDOWN_GRACE_MS', 8000)
39	
40	let running = 0
41	let stopped = false
42	
43	/**
44	 * PURE: resolve a claimed job to its runner adapter + argv. THROWS on a malformed
45	 * row — an unknown/empty verb (getVerb → ValidationError) or bad input (toArgv).
46	 * runJob MUST call this inside a try: a synchronous throw here would skip tick's
47	 * `.finally(() => running -= 1)` and leak a concurrency slot forever (3 malformed
48	 * jobs wedge the worker), and leave the row `claimed` (account lane blocked until
49	 * the reaper). Exported so that guard is unit-tested without spawning a child.
50	 */
51	export function jobSpawnArgs(job: JobRow): { name: string; adapter: string; argv: string[] } {
52	  const input = (job.input ?? {}) as Record<string, unknown>
53	  // epic #6 cutover: RAW jobs (enqueueRaw — the UI's argv-speaking /exec path). They
54	  // carry `input._raw` with the (site, cmd, argv) verbatim instead of a typed verb, so
55	  // resolve the adapter by PATH and pass argv through EXACTLY like the legacy /exec
56	  // dispatcher — same adapterPath + same --format/json strip. A missing adapter throws,
57	  // which runJob catches → markFailed (the row never wedges the account lane).
58	  if (input._raw) {
59	    const site = String(input.site || '')
60	    const cmd = String(input.cmd || '')
61	    const adapter = adapterPath(site, cmd)
62	    if (!adapter) throw new Error(`no adapter at adapters/${site}/${cmd}.js (raw job ${job.job_id})`)
63	    const rawArgv = Array.isArray(input.argv) ? (input.argv as unknown[]) : []
64	    const argv = rawArgv.filter((a) => a != null && a !== '--format' && a !== 'json').map(String)
65	    return { name: `${site}/${cmd}`, adapter, argv }
66	  }
67	  const verb = getVerb(String(job.verb || job.cmd || ''))
68	  return { name: verb.name, adapter: verb.adapter, argv: verb.toArgv(job.input ?? {}) }
69	}
70	
71	// Run one claimed job to completion. Resolves when the child exits (any code) —
72	// the runner owns the terminal ledger write, so the worker only frees its slot.
73	// NEVER throws synchronously (see jobSpawnArgs) — a malformed job resolves after
74	// being marked failed, so tick's slot-release `.finally` always runs.
75	function runJob(job: JobRow): Promise<void> {
76	  let name: string, adapter: string, argv: string[]
77	  try {
78	    ({ name, adapter, argv } = jobSpawnArgs(job))
79	  } catch (err) {
80	    void markFailed(job.job_id, `worker setup failed: ${(err as Error).message}`).catch(() => {})
81	    return Promise.resolve()
82	  }
83	  const env: NodeJS.ProcessEnv = { ...process.env, BLUEBUT_JOB_ID: job.job_id }
84	  if (job.batch_id) env.BLUEBUT_BATCH_ID = String(job.batch_id)
85	
86	  if (DRY_RUN) {
87	    console.log(`[worker] DRY_RUN ${job.job_id} ${name} ${argv.join(' ')}`)
88	    return Promise.resolve()
89	  }
90	
91	  return new Promise<void>((resolve) => {
92	    // Capture the runner's stderr (its step-log lines) into the in-process tap so a live
93	    // HTTP tailer — the /api/queue/jobs/:id/events SSE, same process — can stream them to
94	    // the dashboard (epic #6: the worker spawns out-of-band, so the request handler has no
95	    // child to tee; the tap bridges it). stdout is discarded — the runner writes the
96	    // durable `run` row itself, so the terminal RESULT is read from the ledger, not here.
97	    const child = spawn(NODE, [RUNNER, adapter, ...argv], { env, cwd: REPO, stdio: ['ignore', 'ignore', 'pipe'] })
98	    let ebuf = ''
99	    child.stderr?.on('data', (d: Buffer) => {
100	      ebuf += d
101	      let i: number
102	      while ((i = ebuf.indexOf('\n')) >= 0) {
103	        const line = ebuf.slice(0, i).replace(/\x1b\[[0-9;]*m/g, '')
104	        ebuf = ebuf.slice(i + 1)
105	        if (line.trim()) publishLine(job.job_id, line)
106	      }
107	    })
108	    child.on('error', (err: Error) => {
109	      void markFailed(job.job_id, `worker spawn failed: ${err.message}`).catch(() => {})
110	      endStream(job.job_id)
111	      resolve()
112	    })
113	    child.on('exit', (code) => {
114	      if (ebuf.trim()) publishLine(job.job_id, ebuf.replace(/\x1b\[[0-9;]*m/g, ''))   // flush trailing partial line
115	      endStream(job.job_id)
116	      if (code !== 0) {
117	        // The runner OWNS the terminal write on a clean run, but a child that dies
118	        // before writing one (crash, OOM, SIGKILL) would leave the row stuck in
119	        // `claimed`/`running` — and claim_job treats that as an active job, so the
120	        // account is blocked forever. markFailed is status-scoped (queued/claimed/
121	        // running only), so it's a no-op when the runner already wrote a terminal row.
122	        console.warn(`[worker] ${job.job_id} exited ${code} — sweeping any non-terminal row to failed`)
123	        void markFailed(job.job_id, `child exited ${code}`).catch(() => {})
124	      }
125	      resolve()
126	    })
127	  })
128	}
129	
130	async function tick(): Promise<void> {
131	  if (stopped) return
132	  while (running < MAX_CONCURRENT) {
133	    let job: JobRow | null
134	    try {
135	      job = await claim(WORKER_ID)
136	    } catch (err) {
137	      console.warn(`[worker] claim error: ${(err as Error).message}`)
138	      break
139	    }
140	    if (!job) break
141	    running += 1
142	    void runJob(job).finally(() => { running -= 1 })
143	  }
144	}
145	
146	export const WORKER_ID = `w_${process.pid}`
147	
148	export function startWorker(): () => void {
149	  console.log(`[worker] ${WORKER_ID} up — poll ${POLL_MS}ms, concurrency ${MAX_CONCURRENT}${DRY_RUN ? ' (DRY_RUN)' : ''}`)
150	  const timer = setInterval(() => { void tick().catch((e) => console.warn(`[worker] tick: ${(e as Error).message}`)) }, POLL_MS)
151	  timer.unref?.()
152	  // Reclaim rows orphaned in claimed/running by a worker that died before its
153	  // child.on('exit') could markFailed them (queue.ts reapStale) — the account lane
154	  // stays blocked until this sweeps. Runs regardless of claim activity.
155	  const reaper = setInterval(() => {
156	    void reapStale()
157	      .then((n) => { if (n) console.warn(`[worker] reaped ${n} stale claimed/running row(s)`) })
158	      .catch((e) => console.warn(`[worker] reap: ${(e as Error).message}`))
159	  }, REAP_POLL_MS)
160	  reaper.unref?.()
161	  return () => { stopped = true; clearInterval(timer); clearInterval(reaper) }
162	}
163	
164	// Standalone entry: when worker.ts is the process (not imported by index.mjs), it
165	// must catch SIGTERM/SIGINT itself — container stop/redeploy sends SIGTERM, and a
166	// bare process.on listener would otherwise OVERRIDE Node's default-exit, wedging the
167	// container. So we only self-register when this module IS the entry point; under
168	// index.mjs the barrel's own handlers drive our stop fn instead (no double-install).
169	const isEntry = !!process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href
170	if (isEntry) {
171	  const stop = startWorker()
172	  let shuttingDown = false
173	  const shutdown = (sig: string): void => {
174	    if (shuttingDown) return
175	    shuttingDown = true
176	    console.log(`[worker] ${sig} — draining ${running} in-flight (grace ${SHUTDOWN_GRACE_MS}ms)`)
177	    stop() // idempotent: flips `stopped`, halts new claims + reaper
178	    const hard = setTimeout(() => { console.warn('[worker] grace elapsed — exiting'); process.exit(0) }, SHUTDOWN_GRACE_MS)
179	    hard.unref?.()
180	    const poll = setInterval(() => {
181	      if (running === 0) { clearTimeout(hard); clearInterval(poll); process.exit(0) }
182	    }, 100)
183	    poll.unref?.()
184	  }
185	  process.once('SIGTERM', () => shutdown('SIGTERM'))
186	  process.once('SIGINT', () => shutdown('SIGINT'))
187	}
188

=== migrations dir ===
0000_init_baseline.sql
0001_job_queue.sql
0002_claim_job.sql
0003_worried_goblin_queen.sql
0004_milky_mac_gargan.sql
0005_calm_speedball.sql
0006_parched_cammi.sql
0007_natural_colleen_wing.sql
0008_parched_johnny_storm.sql
0009_issue_186_fb_actor_identity.sql
0010_rename_run_workflow_run.sql
0011_issue_187_friendship_broadcast_config.sql
0012_lively_mystique.sql
0013_oval_typhoid_mary.sql
0014_powerful_the_anarchist.sql
meta
migrations
relations.ts
retired
schema.ts

=== files reading/writing run.ok ('ok:' or .ok or ok=) near run/job ===
scripts/_lib/adapter-finalize.ts:56:    const ok = !!(result && typeof result === 'object' && result.ok === true);
scripts/_lib/adapter-finalize.ts:58:    const severity = ok || isBenignStatus(status) ? 'INFO' : 'ERROR';
scripts/_lib/adapter-finalize.ts:59:    try { await shipLog('bluebut-traffic', severity, `◆ job-end ${cmd.name} account=${account} ${ok ? 'ok' : (isBenignStatus(status) ? status : 'FAIL')} (${Date.now() - t0}ms)`, { ...jobBase, phase: 'end', ok: ok ? 1 : 0, status, duration_ms: Date.now() - t0 }, 'bluebut.runner'); } catch { /* non-fatal */ }
scripts/_lib/adapter-finalize.ts:207:      emitStep(`access: ${acc.access || 'ok'}${acc.doc_status ? ` (HTTP ${acc.doc_status})` : ''}${acc.error_kind ? ` · error=${acc.error_kind}${acc.retryable ? ' (retryable)' : ''}` : ''}${acc.errors ? ` · ${acc.errors.length} err` : ''}${said}`);
scripts/_lib/adapter-finalize.ts:355:      const okEnd = !!(result && typeof result === 'object' && result.ok === true);
scripts/_lib/adapter-finalize.ts:371:        severity: okEnd || endBenign ? 'INFO' : 'ERROR', body: `job-end ${cmd.name} ${okEnd ? 'ok' : (endBenign ? endStatus : 'FAIL')}`,
scripts/_lib/adapter-finalize.ts:372:        attrs: { kind: 'job-end', job_id, account, cmd: cmd.name, op, ok: okEnd ? 1 : 0, status: endStatus, duration_ms: Date.now() - t0 },
scripts/_lib/adapter-finalize.ts:384:    const ok = r.ok === true;
scripts/_lib/adapter-finalize.ts:405:    if (!ok && status === 'failed') {
scripts/_lib/adapter-finalize.ts:412:      status, ok,
scripts/_lib/adapter-finalize.ts:433:    if (!ok && batch_id && process.env.BLUEBUT_TG_JOBS !== '0') {

378	
379	// DURABLE JOB LEDGER — terminal update. Prefer the target the adapter
380	// surfaced in its result (richer numeric id + name + url) over the raw arg.
381	async function updateJobLedger({ result, walled, cmd, account, target, ledgerJobId, batch_id, harKey, recKey, t0, timeline }) {
382	  try {
383	    const r = (result && typeof result === 'object') ? result : {};
384	    const ok = r.ok === true;
385	    const status = jobLifecycleStatus(r, walled);
386	    const rid   = r.group_fb_id || r.fb_group_id || r.group_id || (r.group && (r.group.fb_group_id || r.group.id)) || null;
387	    const rurl  = r.group_url  || (r.group && r.group.url)  || null;
388	    const rname = r.group_name || (r.group && r.group.name) || null;
389	    // Bank the full step/action timeline onto the durable row (`timeline` drained
390	    // above). This is what lets the in-memory JOBS Map stop being the history
391	    // store — a finished job now renders its steps from Postgres, not from the
392	    // capped, restart-lossy bridge Map. Compact + bounded (the wire trail is in
393	    // the HAR; this is the human-readable phase log).
394	    const steps = timeline.slice(-600).map((e) => ({
395	      t: e.t - t0,
396	      s: String(e.text || e.action || e.phase || '').slice(0, 300),
397	      k: e.phase === 'action' ? 'action' : 'step',
398	      src: e.source || 'adapter',
399	    }));
400	    // AUTO GITHUB ISSUE — only on a real FAILURE (not `blocked` gate-refusals, which
401	    // are expected ops states: usync-dead/frozen/velocity). Deduped by signature in
402	    // job_error_issue (one issue per distinct cause, not per run). Fail-open. Awaited
403	    // so the number lands in this same summary write; the child exits right after.
404	    let githubIssue = null;
405	    if (!ok && status === 'failed') {
406	      githubIssue = await fileErrorIssue({
407	        cmd: cmd.name, reason: r.aborted_reason ?? `${cmd.name} failed`,
408	        account, target, job_id: ledgerJobId, batch_id, status,
409	      });
410	    }
411	    await dbRest('PATCH', `run?id=eq.${encodeURIComponent(ledgerJobId)}`, {
412	      status, ok,
413	      aborted_reason: r.aborted_reason ?? null,
414	      summary: githubIssue ? { ...jobSummary(r), github_issue: githubIssue } : jobSummary(r),
415	      steps: steps.length ? steps : null,
416	      har_key: harKey,
417	      rec_key: recKey,
418	      ended_at: new Date().toISOString(),
419	      duration_ms: Date.now() - t0,
420	      updated_at: new Date().toISOString(),
421	      target_type: target.type || (rid ? 'group' : null),
422	      target_id: (target.type === 'group' && rid) ? rid : (target.id || rid),
423	      target_url: rurl || target.url,
424	      target_name: rname || target.name,
425	    }, { silent: true });
426	
427	    // EVERY FAILED JOB PINGS TELEGRAM. This is the ONE settle point every job
428	    // crosses (single, broadcast, workflow, scheduled). Non-batch jobs already
429	    // alert via the backend's notifyJobEnd, so gate on batch_id to fire for the
430	    // children that path never sees (broadcast/workflow/scheduled) — one alert
431	    // per failed job, no double. AWAITED (the child exits right after this, so a
432	    // fire-and-forget fetch would be killed mid-flight). BLUEBUT_TG_JOBS=0 mutes.
433	    if (!ok && batch_id && process.env.BLUEBUT_TG_JOBS !== '0') {
434	      const reason = r.aborted_reason ?? (status === 'blocked' ? 'blocked by Facebook' : `${cmd.name} failed`);
435	      await notifyTelegram(
436	        `❌ Job · ${cmd.name}${account ? ` · ${account}` : ''}\n${status}${target.name || target.url ? ` · ${target.name || target.url}` : ''}\n${String(reason).slice(0, 220)}\n(job ${ledgerJobId} · batch ${String(batch_id).slice(0, 14)})`,
437	      ).catch(() => {});

=== db/migrations dir ===
cloak
entity
README.md
=== schema.ts run table ok field ===
31:	renter_facebook_user_id: text(),
32:	renter_facebook_profile_url: text(),
82:	index("idx_intake_homii_renter_fb").using("btree", table.renter_facebook_user_id.asc().nullsLast().op("text_ops")).where(sql`(renter_facebook_user_id IS NOT NULL)`),
541:	cookies: jsonb(),
780:	req_cookies: jsonb(),
783:	resp_cookies: jsonb(),
836:	ok: boolean(),
881:	check("run_status_chk", sql`status = ANY (ARRAY['created'::text, 'queued'::text, 'claimed'::text, 'running'::text, 'completed'::text, 'ok'::text, 'done'::text, 'failed'::text, 'blocked'::text, 'skipped'::text, 'partial'::text, 'content_unavailable'::text, 'terminated'::text, 'canceled'::text, 'timed_out'::text])`),
955:	n_ok: integer().default(0).notNull(),
1004:	site: text().default('facebook').notNull(),
1103:	cookies_set: jsonb(),
1138:	// so pfbid lookups seq-scan (issue #44.6). Partial → only the rows carrying a pfbid.
1617:	site: text().default('facebook').notNull(),
1697:	// btree can't serve a value-only lookup → seq-scan the alias tier (issue #44.5).
1866:	check("batch_target_status_chk", sql`status = ANY (ARRAY['pending'::text, 'running'::text, 'ok'::text, 'completed'::text, 'failed'::text, 'blocked'::text, 'skipped'::text, 'partial'::text])`),
=== recent drop-column migration example ===
db/migrations/entity/0017_fix_existing_schema.sql
db/migrations/entity/0019_drop_fb_person_uuid.sql
db/migrations/entity/0039_drop_lead_replied_at.sql

820	// compat VIEW (`CREATE VIEW job AS SELECT * FROM run`) declared in the hand-authored migration
821	// db/migrations/entity/0047_rename_run_workflow_run.sql — NOT here, so drizzle never tries to
822	// manage two objects sharing one name.
823	export const run = pgTable("run", {
824		job_id: text().primaryKey().notNull(),
825		batch_id: text(),
826		site: text(),
827		cmd: text().notNull(),
828		op: text(),
829		access: text(),
830		account: text(),
831		target_type: text(),
832		target_id: text(),
833		target_url: text(),
834		target_name: text(),
835		status: text().notNull(),
836		ok: boolean(),
837		aborted_reason: text(),
838		summary: jsonb(),
839		started_at: timestamp({ withTimezone: true, mode: 'string' }),
840		ended_at: timestamp({ withTimezone: true, mode: 'string' }),
841		duration_ms: integer(),
842		created_at: timestamp({ withTimezone: true, mode: 'string' }).defaultNow(),
843		updated_at: timestamp({ withTimezone: true, mode: 'string' }).defaultNow(),
844		har_key: text(),
845		run_id: text(),   // soft join → cloak_runs.id (engine store, separate DB → no FK). id-collapse killed bridge_job_id.
846		rec_key: text(),
847		reharvested_at: timestamp({ withTimezone: true, mode: 'string' }),
848		// The COMPLETE run timeline banked on the row (Skyvern-alignment: the durable Job row
849		// is the whole record of a Run). Shape {t,s,k,src}[]. Added out-of-band by migration
850		// entity/0040_job_steps.sql; declared here so db/schema.ts (the SoT, invariant #25) matches.
851		steps: jsonb(),
852		// job-queue columns (server/jobs): the job ledger doubles as the durable work queue.
853		verb: text(),
854		input: jsonb(),
855		worker_id: text(),
856		claimed_at: timestamp({ withTimezone: true, mode: 'string' }),
857		// bluebut#201: the Skyvern-style discriminator for what KIND of run this row records
858		// (e.g. 'job' | 'workflow_run' | 'e2ee-keepalive' | …). Backfilled from cmd/op by 0047.
859		run_type: text(),
860		// Pacing primitives (issue #217): the ledger itself expresses inter-post jitter +
861		// velocity backoff. `not_before` gates a queued row from being claimed until due
862		// (claim_job_v2 adds `AND (not_before IS NULL OR not_before <= now())`); `priority`
863		// lets an urgent job jump the FIFO order (claim_job_v2 orders priority DESC, then
864		// created_at ASC). Both nullable/defaulted so legacy enqueue paths are unaffected.
865		not_before: timestamp({ withTimezone: true, mode: 'string' }),
866		priority: integer().default(0),
867	}, (table) => [
868		index("job_account_idx").using("btree", table.account.asc().nullsLast().op("timestamptz_ops"), table.created_at.desc().nullsFirst().op("text_ops")),
869		index("job_batch_idx").using("btree", table.batch_id.asc().nullsLast().op("text_ops")),
870		index("job_run_idx").using("btree", table.run_id.asc().nullsLast().op("text_ops")),
871		index("job_created_idx").using("btree", table.created_at.desc().nullsFirst().op("timestamptz_ops")),
872		index("job_queued_idx").using("btree", table.created_at.asc().nullsLast().op("timestamptz_ops")).where(sql`(status = 'queued')`),
873		// Serves claim_job_v2's ORDER BY priority DESC, created_at ASC over the queued slice (issue #217).
874		index("job_queue_pace_idx").using("btree", table.priority.desc().nullsLast().op("int4_ops"), table.created_at.asc().nullsLast().op("timestamptz_ops")).where(sql`(status = 'queued')`),
875		uniqueIndex("job_one_active_per_account").using("btree", table.account.asc().nullsLast().op("text_ops")).where(sql`(status = ANY (ARRAY['claimed'::text, 'running'::text])) AND account IS NOT NULL`),
876		index("job_target_idx").using("btree", table.target_type.asc().nullsLast().op("text_ops"), table.target_id.asc().nullsLast().op("text_ops"), table.created_at.desc().nullsFirst().op("text_ops")),
877		// bluebut#263: bind status to the canonical taxonomy.ts LIFECYCLE UNION the live ledger
878		// vocab (server/jobs/queue.ts LEDGER_ACTIVE+LEDGER_TERMINAL / server/contract/job.ts
879		// JobStatus) — see db/migrations/entity/0049_run_status_check.sql for the authoritative
880		// (NOT VALID) constraint; mirrored here per inv #25 so schema.ts matches the live DB.
881		check("run_status_chk", sql`status = ANY (ARRAY['created'::text, 'queued'::text, 'claimed'::text, 'running'::text, 'completed'::text, 'ok'::text, 'done'::text, 'failed'::text, 'blocked'::text, 'skipped'::text, 'partial'::text, 'content_unavailable'::text, 'terminated'::text, 'canceled'::text, 'timed_out'::text])`),
882	]);
883	
884	// A reusable, versioned Workflow definition (Skyvern WorkflowDefinition-as-data). The

0040_job_steps.sql
0041_rename_to_skyvern.sql
0042_capture_legacy_rpcs.sql
0043_broadcast_config_closest_km.sql
0044_job_error_issue.sql
0045_homii_resolve_claim.sql
0046_resync_workflow_campaign.sql
0047_rename_run_workflow_run.sql
0048_outreach_geo_source.sql
0049_run_status_check.sql
0050_drop_legacy_run_views.sql
0051_rename_run_columns_expand.sql
0052_drop_run_compat_columns.sql
0053_schedule_compat_and_run_grants.sql
pending
=== latest entity migration head ===
        status = ANY (ARRAY[
          'pending','running',
          'ok','completed',
          'failed','blocked','skipped','partial'
        ])
      ) NOT VALID;
  END IF;
END $$;

-- target_run has NO status column (last_run_at/last_job_id/last_result only) — nothing to
-- constrain there; see db/schema.ts target_run.

-- NOT VALID leaves these unenforced against pre-existing rows but enforced on every INSERT/
-- UPDATE from here on (Postgres always validates NOT VALID checks against new/changed rows).
-- A future migration can `VALIDATE CONSTRAINT <name>` once a one-off audit query confirms zero
-- live rows fall outside the ANY-list above (`SELECT status, count(*) FROM run GROUP BY 1` etc.)
-- — deliberately NOT done in this pass per the #263 scope note (smallest correct fix; don't risk
-- a VALIDATE scan finding a surprise value and failing this deploy).

NOTIFY pgrst, 'reload schema';
=== how migrations are applied? ===
scripts/docker-migrate-boot.sh:12:#   2. hand-SQL imperative pile (db/migrations/entity/*.sql), entity DB — the things

=== 0052 ===
-- 0052_drop_run_compat_columns — CONTRACT (drop) pass of bluebut#305 (epic #200 Phase B),
-- the follow-on to 0051's EXPAND. 0051 renamed the run-registry COLUMNS to their canonical
-- Skyvern names (job_id→id, run_id→engine_run_id, batch_id→workflow_run_id) and kept the OLD
-- names as compat columns kept writable by a mirror trigger, so a still-deployed pre-rename
-- container survived the deploy window. This deploy migrates EVERY reader/writer onto the
-- canonical names (server/runs, scripts/bluebut-adapter, adapter-finalize, health reaper,
-- the UI queryKeys), so the shims can finally go:
--
--   • drop the run_compat_mirror trigger + its function
--   • drop run.job_id / run.run_id / run.batch_id compat columns
--   • drop the batch_target passthrough compat VIEW (workflow_run_target is the base table)
--   • rename job_schedule → run_schedule
--
-- The job/batch compat VIEWs (0047) were already dropped in 0050; workflow_run's OWN `batch_id`
-- PK is a distinct concept and is NOT touched. Idempotent: every drop is IF EXISTS and the
-- rename is guarded by information_schema, so a re-run — or a run against an already-contracted
-- DB — is a no-op. No explicit BEGIN/COMMIT: the runner (scripts/apply-migrations.ts) wraps
-- each file in one transaction (matches 0047/0049/0050/0051).

-- 0) Repoint the claim_job RPC onto the run registry BEFORE dropping run.job_id. Its original
--    body (db/0002_claim_job.sql) read `FROM job ... WHERE job_id = ...`; the `job` compat view
--    was dropped in 0050 and the run PK is now `id`. queue.ts still calls rpc/claim_job (the
--    durable-queue worker path), so keep it valid. DROP+CREATE because the return rowtype moves
--    from SETOF job → SETOF run. Same atomic FOR UPDATE SKIP LOCKED + per-account active guard.
DROP FUNCTION IF EXISTS public.claim_job(text);
CREATE FUNCTION public.claim_job(p_worker text)
RETURNS SETOF public.run
LANGUAGE plpgsql AS $$
DECLARE
  v_run public.run;
BEGIN
  SELECT j.* INTO v_run
  FROM public.run j
  WHERE j.status = 'queued'
    AND (
      j.account IS NULL
      OR NOT EXISTS (
        SELECT 1 FROM public.run b
        WHERE b.account = j.account
          AND b.status IN ('claimed', 'running')
      )
    )
  ORDER BY j.created_at ASC
  FOR UPDATE OF j SKIP LOCKED
  LIMIT 1;

  IF NOT FOUND THEN
    RETURN;
  END IF;

  BEGIN
    UPDATE public.run
      SET status = 'claimed', worker_id = p_worker, claimed_at = now(), updated_at = now()
      WHERE id = v_run.id;
  EXCEPTION WHEN unique_violation THEN
    RETURN;  -- another worker just claimed this account — leave ours queued
  END;

  v_run.status := 'claimed';
  v_run.worker_id := p_worker;
  RETURN NEXT v_run;
END;
$$;

-- 1) Drop the old⇄new mirror trigger + its function — no writer touches the old names anymore.
DROP TRIGGER IF EXISTS run_compat_mirror_trg ON public.run;
DROP FUNCTION IF EXISTS public.run_compat_mirror();

-- 2) Drop the run-table compat alias columns. The canonical id (PK) / engine_run_id /
--    workflow_run_id remain; no index or view references the alias columns (the indexes were
--    renamed to run_*_idx over the canonical columns in 0051; overview_daily/overview_totals
--    read only run.created_at/status).
ALTER TABLE public.run DROP COLUMN IF EXISTS job_id;
ALTER TABLE public.run DROP COLUMN IF EXISTS run_id;
ALTER TABLE public.run DROP COLUMN IF EXISTS batch_id;

-- 3) Drop the batch_target passthrough compat VIEW — callers read/write workflow_run_target now.
DROP VIEW IF EXISTS public.batch_target;

-- 4) Rename job_schedule → run_schedule (Skyvern run vocabulary). No FK or view depends on the
--    table, and the scheduler runtime (single web replica) is redeployed with this migration, so
--    a clean rename with no compat view is safe. Index names keep their job_schedule_* labels
--    (functionally inert; a later cosmetic pass can rename them).
DO $$
BEGIN
  IF EXISTS (SELECT 1 FROM information_schema.tables
             WHERE table_schema = 'public' AND table_name = 'job_schedule' AND table_type = 'BASE TABLE')
     AND NOT EXISTS (SELECT 1 FROM information_schema.tables
                     WHERE table_schema = 'public' AND table_name = 'run_schedule' AND table_type = 'BASE TABLE')
  THEN
    EXECUTE 'ALTER TABLE public.job_schedule RENAME TO run_schedule';
  END IF;
END $$;

NOTIFY pgrst, 'reload schema';
=== pending dir ===
total 12
drwxrwxr-x 2 philip philip 4096 Jul 24 17:10 .
drwxrwxr-x 3 philip philip 4096 Jul 24 17:10 ..
-rw-rw-r-- 1 philip philip 1092 Jul 24 17:10 README.md

=== pending README ===
# Pending (NOT auto-applied) entity migrations

The migration runner (`scripts/apply-migrations.ts`) does a **non-recursive**
`readdirSync('db/migrations/entity')`, so `.sql` files in THIS subdir are **never**
auto-applied on deploy. They are held here deliberately.

_Currently empty — no held migrations._

## History

- **`0048_drop_legacy_run_views.sql`** — the CONTRACT step of epic #200 / issue #206.
  `0047` renamed `job`→`run` / `batch`→`workflow_run`, keeping the old names alive as
  passthrough compat VIEWS through the deploy window (expand/contract, inv #21). Once `0047`
  was live in prod and every call site repointed at `run`/`workflow_run` (grep = zero legacy
  refs), the drop was promoted out of `pending/` as
  `db/migrations/entity/0050_drop_legacy_run_views.sql` — renumbered off the taken `0048`
  slot (`0048_outreach_geo_source`), and extended to first `CREATE OR REPLACE` the two
  `overview_*` analytics views onto `run` so the `DROP VIEW job` no longer fails on that
  dependency. It ships in a SEPARATE, LATER deploy than `0047` (never the same one).

=== bridge.ts 415-440 ===
  "POST /workflow/plan", "POST /campaign/plan", "POST /workflow/run", "POST /campaign/run",
  "POST /schedule/run", "POST /workflow/stop", "POST /campaign/stop",
  "GET /accounts/health", "GET /accounts/usable",
  "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
//    handler), so this ALWAYS handles the request.
function execCatchAll(req: IncomingMessage, res: ServerResponse) {
  let body = "";
  req.on("data", (c) => (body += c));
  req.on("end", () => {
    let b: Record<string, unknown> = {};
    try { b = JSON.parse(body || "{}"); } catch { /* empty */ }
    const path0 = (req.url || "").split("?")[0];

    if (path0 === "/login/stream") { handleLogin(req, res, b); return; }

    // Header-only auth on the process-spawn path (issue #20): /exec runs an adapter
    // subprocess. A query-string token (accepted by the global gate for reads) is too
    // weak to authorize spawning code — require the Authorization header here.
    if ((path0 === "/exec" || path0 === "/exec/stream") && !authed(req, { allowQuery: false })) {
      res.writeHead(401, { ...CORS, "Content-Type": "application/json" });

server/http/runtime/auto-broadcast.mjs:81://   • hosted-only: gated on SCHED_ENABLED, exactly like the scheduler + boot-resume,
server/http/runtime/broadcast.mjs:219:    // cmd='create-group-post' reserves sweep.batch (tickSchedules) but
server/http/runtime.mjs:37:if (SCHED_ENABLED) {
server/http/runtime.mjs:38:  tickSchedules();
server/http/runtime.mjs:39:  setInterval(tickSchedules, SCHED_TICK_MS).unref?.();
server/http/runtime.mjs:40:} else if (SCHED_TICK_MS > 0) {
server/http/runtime.mjs:45:if (SCHED_ENABLED && RECOGNIZE_TICK_MS > 0) {
server/http/runtime.mjs:53:if (SCHED_ENABLED) setTimeout(() => { bcastResumeRunning(); }, 8_000).unref?.();
server/http/runtime.mjs:56:if (SCHED_ENABLED && AUTO_BCAST_TICK_MS > 0) {
server/http/runtime.mjs:61:  // listings) alone — NOT tickSchedules and NOT the funnel sweep — so a local run can
server/http/runtime.mjs:69:// Durable manual broadcast QUEUE drain. Unconditional (NOT SCHED_ENABLED-gated): a
server/http/runtime/run-core.mjs:7://                       _activeSweepBatch`; because BOTH the scheduler (tickSchedules)

30	import { SCHED_ENABLED, SCHED_TICK_MS, RECOGNIZE_TICK_MS, fireSchedule, fireWorkflowSchedule, tickSchedules, tickRecognize } from "./runtime/scheduler.mjs";
31	import { AUTO_BCAST_TICK_MS, bcastResumeRunning, bcastAutoSweep, bcastAutoSweepAll, bcastQueueDrain } from "./runtime/auto-broadcast.mjs";
32	import { handleLogin } from "./runtime/login.mjs";
33	
34	// ── BOOT SIDE EFFECTS — verbatim order from the pre-split runtime.mjs ──────────
35	
36	// Recurring scheduler (HOSTED-ONLY; the SCHED_ENABLED gate lives in scheduler.mjs).
37	if (SCHED_ENABLED) {
38	  tickSchedules();
39	  setInterval(tickSchedules, SCHED_TICK_MS).unref?.();
40	} else if (SCHED_TICK_MS > 0) {
41	  console.log("[backend] recurring scheduler DISABLED (no API_TOKEN → local/dev bridge). Prod fires schedules; set BLUEBUT_FORCE_SCHEDULER=1 to run them here.");
42	}
43	
44	// Runtime recognizer — recognize cloak_captures → fb_* continuously.
45	if (SCHED_ENABLED && RECOGNIZE_TICK_MS > 0) {
46	  setTimeout(tickRecognize, 10_000).unref?.();   // first sweep shortly after boot
47	  setInterval(tickRecognize, RECOGNIZE_TICK_MS).unref?.();
48	}
49	
50	// Batch boot-resume — re-run any batch left 'running' when this bridge last died.
51	// Only the always-on hosted bridge resumes (same gate as the scheduler — a local
52	// dev bridge must not pick up prod's live batches).
53	if (SCHED_ENABLED) setTimeout(() => { bcastResumeRunning(); }, 8_000).unref?.();
54	
55	// Auto-broadcast — a new bofrid/funnel listing goes live → run its broadcast.
56	if (SCHED_ENABLED && AUTO_BCAST_TICK_MS > 0) {
57	  setTimeout(() => { bcastAutoSweepAll(); }, 20_000).unref?.();   // first sweep shortly after boot
58	  setInterval(bcastAutoSweepAll, AUTO_BCAST_TICK_MS).unref?.();
59	} else if (process.env.BLUEBUT_FORCE_BROADCAST_SWEEP === "1" && AUTO_BCAST_TICK_MS > 0) {
60	  // Broadcast-ONLY on-demand drip (no full scheduler): ticks bcastAutoSweep (bofrid
61	  // listings) alone — NOT tickSchedules and NOT the funnel sweep — so a local run can
62	  // drain the armed auto_broadcast cursor without also firing every due scrape/join.
63	  // Same guardrails (serialized, paced, hourly cap, stop-on-block, repost guard).
64	  console.log("[backend] FORCE broadcast-only sweep ON (BLUEBUT_FORCE_BROADCAST_SWEEP=1) — bofrid listings only");
65	  setTimeout(() => { bcastAutoSweep(); }, 5_000).unref?.();
66	  setInterval(() => { bcastAutoSweep(); }, AUTO_BCAST_TICK_MS).unref?.();
67	}
68	
69	// Durable manual broadcast QUEUE drain. Unconditional (NOT SCHED_ENABLED-gated): a
70	// manual Run isn't gated either, and the drain is cheap — one indexed peek that
71	// returns instantly when nothing is queued.
72	setInterval(bcastQueueDrain, 30_000).unref?.();
73	
74	// (Live tab CDP screencast removed — watching a browser is noVNC-only now:

/home/philip/Documents/GitHub/bluebut-issue-52
fix/issue-52

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

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

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

=== writes: 'ok' appearing in a run/job PATCH/POST body ===
server/app.ts:59:  app.get('/health', (c) => c.json({ ok: true, service: 'bluebut' }))
server/http/listings/geo.mjs:45:    return kvJson(res, 200, { ok: true, kommun: byLevel("kommun"), lan: byLevel("lan") });
server/http/listings/geo.mjs:173:    if (!Number.isFinite(lat) || !Number.isFinite(lon)) return kvJson(res, 400, { ok: false, error: "lat + lng required" });
server/http/listings/geo.mjs:189:      ok: true, lat, lng: lon,
server/http/listings/geo.mjs:206:    if (!Number.isFinite(lat) || !Number.isFinite(lon)) return kvJson(res, 400, { ok: false, error: "lat + lng required" });
server/http/listings/geo.mjs:243:      ok: true, lat, lng: lon,
server/http/listings/geo.mjs:265:      ok: true,
server/http/listings/geo.mjs:289:      return kvJson(res, 400, { ok: false, error: "fb_group_id + geo_level (lan|kommun|area) + geo_ref required" });
server/http/listings/geo.mjs:314:          return kvJson(res, 400, { ok: false, error: `no such ${geo_level}: "${geo_ref}" (geo.bofrid.eu → 404)` });
server/http/listings/geo.mjs:327:    return kvJson(res, 200, { ok: true, fb_group_id, geo_level, geo_ref, polygon_cached, polygon_error });
server/http/listings/geo.mjs:338:      return kvJson(res, 400, { ok: false, error: "fb_group_id + geo_level + geo_ref required" });
server/http/listings/geo.mjs:344:    return kvJson(res, 200, { ok: true, fb_group_id, geo_level, geo_ref, removed: true });
server/store/profiles.ts:126:  async delete(id: string): Promise<{ ok: true; id: string }> {
server/store/profiles.ts:128:    return { ok: true, id }
server/contract/job.ts:41:  ok: z.boolean().nullable().optional(),
scripts/canary-admin.ts:77:        resolve({ ok: false, aborted_reason: 'unparseable adapter output', raw: out.slice(0, 300) });
scripts/canary-admin.ts:80:    child.on('error', (e) => resolve({ ok: false, aborted_reason: `spawn failed: ${e.message}` }));
scripts/canary-admin.ts:110:  try { res = (await fn()) || {}; } catch (e) { res = { ok: false, detail: String(e) }; }
scripts/canary-admin.ts:112:  const rec = { step: name, ok: !!res.ok, detail: res.detail || '', secs, status: res.status, diagnostic: res.diagnostic || null };
scripts/canary-admin.ts:116:    { step: name, ok: String(!!res.ok), status: res.status, group: GROUP, actor: ACTOR, target: TARGET, diag_url: res.diagnostic?.url, diag_heading: res.diagnostic?.heading });
scripts/canary-admin.ts:134:    return { ok: !!(r.logged_in || r.status === 'logged_in'), detail: `status=${r.status}`, status: r.status, stop: true };
scripts/canary-admin.ts:139:    return { ok: !!(r.logged_in || r.status === 'logged_in') && !!targetFbId, detail: `fb_id=${targetFbId}`, status: r.status, stop: true };
scripts/canary-admin.ts:156:    steps.push({ step: 'accept', ok: true, detail: 'skipped — target already admin', secs: '0.0' });
scripts/canary-admin.ts:161:      return { ok: r.ok && r.status === 'accepted', detail: r.ok ? `verified_role=${r.verified_role}` : r.aborted_reason, status: r.status, diagnostic: r.diagnostic, stop: !r.ok };
scripts/canary-admin.ts:174:    return { ok: r.ok && r.status === 'demoted', detail: r.status || r.aborted_reason, status: r.status, diagnostic: r.diagnostic };
scripts/canary-admin.ts:201:  const slim = steps.map(s => ({ step: s.step, ok: s.ok, detail: s.detail, secs: s.secs, status: s.status,
server/http/runtime/login.mjs:41:  if (!account) return j(400, { ok: false, error: "account required" });
server/http/runtime/login.mjs:72:            emit(`[RESULT] ${JSON.stringify({ result: { ok: true, did_login: false, status: "already_logged_in", fb_id: p.fb_id || null } })}`);
server/http/runtime/login.mjs:108:        const result = parseLoose(out) || { ok: rc === 0, logged_in: rc === 0 };
server/http/runtime/login.mjs:112:        emit(`[RESULT] ${JSON.stringify({ result: { did_login: true, ...result, ok: job.ok } })}`);
server/http/lib.mjs:140:  if (r?.ok && fallbackLink && r?.warming_up === false) return { ok: true, link: fallbackLink };
server/http/lib.mjs:141:  if (!seed) return fallbackLink ? { ok: true, link: fallbackLink } : { ok: false, link: null, error: String(r?.error || "focus returned no seed") };
server/http/lib.mjs:147:      if (sess?.browser_up && sess?.alive && sess?.viewer_url) return { ok: true, link: sess.viewer_url };
server/http/lib.mjs:152:  return { ok: false, link: null, error: lastErr || "focus did not become drivable" };
server/http/api-v1-tables.ts:54:  | { ok: true; params: Record<string, string>; limit: number; offset: number }
server/http/api-v1-tables.ts:55:  | { ok: false; error: string; valid_filters: string[] };
server/http/api-v1-tables.ts:68:    if (!cols.has(col)) return { ok: false, error: `unknown order column "${col}"`, valid_filters: [...cols].sort() };
server/http/api-v1-tables.ts:80:    if (!cols.has(key)) return { ok: false, error: `unknown filter "${key}"`, valid_filters: [...cols].sort() };
server/http/api-v1-tables.ts:83:  return { ok: true, params, limit, offset };
server/routes/jobs.ts:73:        await stream.writeSSE({ event: 'status', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok }) })

55	jobs.get('/:id/events', (c) => streamSSE(c, async (stream) => {
56	  const id = c.req.param('id')
57	  // epic #6: tail the worker's live step-log lines (job-stream tap, same process) so the
58	  // dashboard shows per-step progress — the durable queue SSE otherwise carries only
59	  // status/result frames (the worker runs the runner out-of-band, no child to tee). Bare
60	  // `data:` frames (no event name) render as progress lines in ui/src/api/jobs.ts (onLine).
61	  const tap = subscribe(id)
62	  const pending: string[] = [...tap.replay]
63	  tap.onLine((line) => pending.push(line))
64	  const flush = async (): Promise<void> => { while (pending.length && !stream.aborted) await stream.writeSSE({ data: pending.shift()! }) }
65	  let last = null
66	  try {
67	    for (let i = 0; i < 600 && !stream.aborted; i += 1) {
68	      await flush()
69	      const job = await getJob(id)
70	      if (!job) { await stream.writeSSE({ event: 'error', data: JSON.stringify({ error: 'not_found' }) }); return }
71	      if (job.status !== last) {
72	        last = job.status
73	        await stream.writeSSE({ event: 'status', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok }) })
74	      }
75	      if (isLedgerTerminal(job.status)) {
76	        await flush()   // drain any final lines before the terminal frame
77	        // epic #6: emit a terminal `result` frame so the repointed UI job client
78	        // (ui/src/api/jobs.ts, off /exec/stream → this stream) can resolve its
79	        // RunResult. `ok` is the funnel-derived boolean; `result` rides through if
80	        // the ledger row carries one (the UI falls back to deriving ok from status).
81	        await stream.writeSSE({ event: 'result', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok ?? okForStatus(job.status), result: job.result ?? null }) })
82	        return
83	      }
84	      await stream.sleep(1000)
85	    }
86	  } finally { tap.close() }
87	}))
88

=== ui selects mentioning ok column on run/job ===
ui/src/api/har-traces.ts:11:  if (!r.ok) throw new Error(r.status === 404 ? "no HAR captured for this job" : `/har/blob HTTP ${r.status}`);
ui/src/api/har-traces.ts:63:  if (!r.ok) throw new Error(r.status === 404 ? "no HAR captured for this job" : `/har/blob HTTP ${r.status}`);
ui/src/components/WorkflowView.tsx:62:    // i.e. wire data — NOT the retiring run.ok column — so it is read as-is.
ui/src/state/jobs.tsx:128:      if (r.ok) {
ui/src/components/AccountJobs.tsx:30:// funnel writes), NOT the retiring `run.ok` boolean (#295: read-side off run.ok).
ui/src/components/AccountJobs.tsx:43:  return j.ok === false ? "failed" : "ok";
ui/src/components/AccountJobs.tsx:123:        phase: live ? phaseOfBridge(live) : phaseFromStatus(j.status, j.ok),
ui/src/pages/JobDebug.tsx:64:  const ok = fromSnap?.ok ?? log?.jobOk ?? null;
ui/src/pages/groups/MultiGroupScrapeForm.tsx:244:                border: `1px solid ${res?.status === "running" ? "var(--color-info-fg)" : !e.ok ? "var(--color-danger-bg)" : "var(--color-surface-2)"}`, borderRadius: 8, padding: 10,
ui/src/api/jobs.ts:109:    if (!enq.ok) {
ui/src/api/jobs.ts:121:    if (!r.ok || !r.body) throw new Error(`/api/queue/jobs/${job_id}/events HTTP ${r.status}`);
ui/src/api/jobs.ts:226:  if (!r.ok || !r.body) {
ui/src/api/jobs.ts:298:  if (!r.ok) throw new Error(`/jobs HTTP ${r.status}`);
ui/src/api/jobs.ts:330:  if (!r.ok) throw new Error(`/jobs/failures HTTP ${r.status}`);
ui/src/api/jobs.ts:372:  if (!r.ok) throw new Error(`/messenger/socket-status HTTP ${r.status}`);
ui/src/api/jobs.ts:395:  if (!r.ok && !("alive" in d)) throw new Error(d.error || `/messenger/health HTTP ${r.status}`);
ui/src/api/jobs.ts:408:  if (!r.ok) throw new Error(d?.error || `/messenger/connect HTTP ${r.status}`);
ui/src/api/jobs.ts:418:  if (!r.ok) throw new Error(`/jobs/cancel HTTP ${r.status}`);
ui/src/api/jobs.ts:431:  if (!r.ok) throw new Error(`/job-log HTTP ${r.status}`);
ui/src/api/workflow-runs.ts:105:    if (r.status === "ok") c.ok++;
=== JobRow.ok / job.ok / row.ok reads in server ===
server/routes/jobs.ts:73:        await stream.writeSSE({ event: 'status', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok }) })
server/routes/jobs.ts:81:        await stream.writeSSE({ event: 'result', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok ?? okForStatus(job.status), result: job.result ?? null }) })
server/http/runtime/recordings.mjs:47:      row.ok = d.result?.success ?? d.result?.done ?? null;
server/http/runtime/login.mjs:55:    job.endedAt = Date.now(); job.status = "done"; job.ok = false;
server/http/runtime/login.mjs:71:            job.ok = true;
server/http/runtime/login.mjs:111:        job.ok = rc === 0 && result.ok !== false && result.logged_in !== false;
server/http/runtime/login.mjs:112:        emit(`[RESULT] ${JSON.stringify({ result: { did_login: true, ...result, ok: job.ok } })}`);
server/http/runtime/login.mjs:115:      child.on("error", (e) => { clearTimeout(killTimer); job.ok = false; emit(`[ERROR] ${e.message}`); res.end(); resolve(); });
server/http/runtime/job-registry.mjs:249:  if (job.ok === true) b.ok++; else if (job.ok === false) b.failed++;
server/http/runtime/job-registry.mjs:266:  const ok = job.ok === true;

The file /home/philip/Documents/GitHub/bluebut-issue-52/server/jobs/queue.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-issue-52/server/jobs/queue.ts has been updated successfully. (file state is current in your context — no need to Read it back)

=== health.mjs 240-250 ===
  try {
    // Cover BOTH orphan shapes (issue #43.F1): a row stuck `running` past its
    // started_at, AND a row stuck `claimed` past its claimed_at — a worker that
    // claimed then died before writing its first `running` row leaves started_at
    // NULL, so a status=running-only sweep never matches it and claim_job's
    // per-account guard leaves that lane permanently unclaimable.
    const rows = await dbRest("PATCH",
      `run?or=(and(status.eq.running,started_at.lt.${encodeURIComponent(cutoff)}),and(status.eq.claimed,claimed_at.lt.${encodeURIComponent(cutoff)}))`,
      { status: "failed", ok: false, aborted_reason: "runner_died (no terminal update — reaped)",
        ended_at: new Date().toISOString(), updated_at: new Date().toISOString() },
      { silent: true, headers: { Prefer: "return=representation" } });
=== health.mjs 348-358 ===
async function reapOnBoot() {
  const cutoff = new Date(Date.now() - 60_000).toISOString();
  try {
    // Same running+claimed coverage as the steady reaper (issue #43.F1): a boot
    // SIGKILLed every child this process owned, so a row left `claimed` (never
    // reached the first `running` write) is just as orphaned as a `running` one.
    const rows = await dbRest("PATCH",
      `run?or=(and(status.eq.running,started_at.lt.${encodeURIComponent(cutoff)}),and(status.eq.claimed,claimed_at.lt.${encodeURIComponent(cutoff)}))`,
      { status: "failed", ok: false, aborted_reason: "bridge_restart (orphaned running/claimed row — reaped on boot)",
        ended_at: new Date().toISOString(), updated_at: new Date().toISOString() },
      { silent: true, headers: { Prefer: "return=representation" } });
=== bluebut-adapter 145-155 ===
      const result = { ok: false, status: 'blocked', aborted_reason: reason,
        cdp_unreachable: !health.reachable, cdp_not_drivable: true, cdp_base: CLOAK_API_BASE,
        cdp_native: !!health.native, account: kwargs.account || null, cmd: cmd.name, op };
      try {
        const nowIso = new Date().toISOString();
        await dbRest('POST', 'run', {
          id: job_id, workflow_run_id: process.env.BLUEBUT_BATCH_ID || null, site: cmd.site, cmd: cmd.name,
          op, access, account: kwargs.account || null, status: 'blocked', ok: false,
          aborted_reason: reason, summary: { cdp_unreachable: !health.reachable, not_drivable: true, base: CLOAK_API_BASE, native: !!health.native, reason: health.reason },
          started_at: nowIso, ended_at: nowIso, duration_ms: 0, updated_at: nowIso,
        }, { silent: true, headers: { Prefer: 'resolution=merge-duplicates' } });
=== bluebut-adapter blocked inserts pattern (grep) ===
135:    const health = await platformHealth().catch((e) => ({ ok: false, reachable: false, drivable: false,
145:      const result = { ok: false, status: 'blocked', aborted_reason: reason,
152:          op, access, account: kwargs.account || null, status: 'blocked', ok: false,
181:      const result = { ok: false, status: 'blocked', aborted_reason: reason, frozen: true,
189:          op, access, account: acct, status: 'blocked', ok: false,
223:      const result = { ok: false, status: 'blocked', aborted_reason: reason, account: acct, cmd: cmd.name, op,
229:          op, access, account: acct, status: 'blocked', ok: false,
258:      const result = { ok: false, status: 'blocked', aborted_reason: reason, account: acct, cmd: cmd.name, op,
264:          op, access, account: acct, status: 'blocked', ok: false,
389:            ok: false,
415:        result = { ...(result && typeof result === 'object' ? result : {}), ok: false, status: 'error', aborted_reason: `output failed schema: ${chk.error}`, output_contract_failed: true, raw_result: result };
427:    result = { ...(result && typeof result === 'object' ? result : {}), ok: false, status: 'error', aborted_reason: msg, error: msg };
464:  try { process.stdout.write(JSON.stringify({ ok: false, status, aborted_reason: msg, error: msg }) + '\n'); } catch {}

=== bluebut-adapter 380-395 ===
      try {
        const e = await detectFbError(br.page);
        if (e && e.kind) {
          let diagnostic = null;
          try { diagnostic = await captureDiagnostic(br.page, { label: `runner-backstop: ${e.kind}` }); } catch { /* best-effort */ }
          const status = e.kind === 'unavailable' ? 'content_unavailable' : e.kind; // block | checkpoint
          emitStep(`⚠ runner backstop: FB ${e.kind} on final page — overriding ok:true → ok:false`);
          result = {
            ...result,
            ok: false,
            status,
            blocked: !!e.blocking,
            fb_error: e.kind,
            aborted_reason: `FB ${e.kind} on final page (runner backstop): ${e.text || 'interstitial'}`,
            diagnostic: diagnostic || result.diagnostic || null,
            runner_backstop: true,
=== accounts-mint 40-55 ===

const FB_LOGIN_URL = "https://www.facebook.com/login/device-based/regular/login/";
const CLOAK_VNC_URL = "https://cloak-vnc.hostbun.cc";

// A mint is a JOB — one deterministic ledger row per minted seed (mint-<seed>), so
// the onboarding shows up in Jobs/history like every other action. Upsert on
// job_id so start → observe → bank all update the same row. Best-effort.
async function writeMintJob(account: string, seed: string, patch: Row) {
  try {
    await dbRest("POST", "run", [{
      job_id: `mint-${seed}`, site: CLOAK_SERVICE, cmd: "mint", op: "mint", access: "write",
      account, target_type: "account", target_id: account, ...patch, updated_at: new Date().toISOString(),
    }] as any, { headers: { Prefer: "resolution=merge-duplicates,return=minimal" } } as any);
  } catch { /* ledger best-effort */ }
}

=== accounts-mint 244-274 ===
    catch (e: any) { observeErr = String(e?.message || e); }
    const hardWall = observed?.login?.hard_wall === true;
    await writeMintJob(account, seed, {
      status: hardWall ? "blocked" : "ok", ok: !hardWall,
      ended_at: new Date().toISOString(),
      summary: { banked: true, login: observed?.login ?? null, wiretap_session: observed?.session_id ?? null, ...(observeErr ? { observe_failed: observeErr } : {}) },
    });
    return { status: 200, body: { ok: true, account, seed, saved, observed } };
  } catch (e: any) {
    return { status: 502, body: { ok: false, error: `bank session failed (cdp_save): ${e?.message || e}` } };
  }
}

// POST /accounts/mint/observe {account,seed} — run the passive wires over the
// mint's wiretap session WITHOUT banking. This is how a checkpoint that BLOCKS
// login is seen + saved: the human never completes login (so never banks), but
// the wiretap already captured the /checkpoint/<id> + UFAC screen, so harvesting
// it burns the account (login-wire) and returns the verdict. Poll this from the
// dialog to show "logged in ✓" vs "checkpoint: video_selfie → burned".
export async function handleAccountMintObserve(b: Row): Promise<Resp> {
  const account = String(b.account || "").trim();
  const seed = String(b.seed || "").trim();
  if (!/^[a-z0-9][a-z0-9_-]{1,39}$/.test(account) || !seed) {
    return { status: 400, body: { ok: false, error: "account (valid slug) + seed are required" } };
  }
  try {
    const observed = await observeMintSeed(account, seed);
    if (observed?.login?.hard_wall) {
      await writeMintJob(account, seed, { status: "blocked", ok: false, ended_at: new Date().toISOString(), summary: { login: observed.login, wiretap_session: observed.session_id } });
    }
    return { status: 200, body: { ok: true, account, seed, ...observed } };
=== e2ee 82-90, 140-148 ===
}

// Write the event JOB row (running → terminal). Same ledger shape the runner writes.
async function ledger(job) {
  try {
    await rest('POST', 'run', job, { silent: true, headers: { Prefer: 'resolution=merge-duplicates' } });
  } catch { /* ledger best-effort */ }
}

  const enrolled = !!res?.registered;
  const endIso = new Date().toISOString();
  await ledger({
    job_id, site: 'messenger', cmd: 'e2ee-autoconnect', op: 'messenger.e2ee_autoconnect',
    access: 'write', account, target_type: 'account', target_id: account, target_name: account,
    status: ok ? 'ok' : 'failed', ok,
    aborted_reason: ok ? null : (err || 'stored-CAT + fetch-mint both failed — jar likely dead'),
    summary: { trigger: 'cat-harvest', mode, connected: ok, enrolled, wa_device_id: res?.wa_device_id || null },
    started_at: nowIso, ended_at: endIso, updated_at: endIso,

=== job-registry 120-145 ===
// Map is only the live in-flight supervisor. id = job_id (id-collapse: the ledger PK
// IS the bridge/UI id now — no separate bridge_job_id), so a still-running row
// reconciles with its live Map entry (same id → the UI dedupes active vs recent).
// Status is normalised back to the Map's {running|done} vocab (the ledger's lifecycle
// ok/failed/partial/blocked all render as done+`ok`), and ISO timestamps → epoch ms,
// so the wire contract the UI reads is unchanged.
async function recentFromLedger(limit, liveIds) {
  const cols = "job_id,batch_id,cmd,account,target_type,target_id,target_name,status,ok,started_at,ended_at,created_at";
  const rows = (await dbRest("GET", `run?select=${cols}&order=created_at.desc&limit=${limit}`, null, { silent: true })) || [];
  const ms = (s) => (s ? Date.parse(s) || null : null);
  const out = [];
  for (const r of rows) {
    const id = r.job_id;
    if (!id || (liveIds && liveIds.has(id))) continue;   // already shown live in active/queued
    out.push({
      id, account: r.account || null, kind: "job", cmd: r.cmd || null, label: r.cmd || "job",
      group: r.target_type === "group" ? r.target_id : null, groupName: r.target_name || null,
      traceJobId: r.job_id || null,
      batch: r.batch_id || null, batchLabel: null, batchSize: null, batchIndex: null,
      status: r.status === "running" ? "running" : "done", queuePos: 0,
      queuedAt: ms(r.created_at), startedAt: ms(r.started_at), endedAt: ms(r.ended_at),
      ok: r.ok, lastLine: null, browserSeed: null, logLen: 0,
    });
  }
  return out;
}
=== job-registry 170-182 ===
// threshold, newest first, each with its auto-filed GitHub issue number.
const FAIL_STUCK_MS = numEnv('JOB_REAP_AFTER_MS', 45 * 60_000);
async function jobFailures() {
  const since = new Date(Date.now() - 7 * 24 * 60 * 60_000).toISOString();
  const stuckBefore = new Date(Date.now() - FAIL_STUCK_MS).toISOString();
  const cols = "job_id,batch_id,cmd,account,status,ok,aborted_reason,target_name,target_url,started_at,ended_at,summary";
  const q = `run?select=${cols}` +
    `&or=(and(status.in.(failed,blocked),created_at.gte.${encodeURIComponent(since)}),and(status.eq.running,started_at.lt.${encodeURIComponent(stuckBefore)}))` +
    `&order=created_at.desc&limit=200`;
  const rows = await dbRest("GET", q, null, { silent: true });
  const failures = (Array.isArray(rows) ? rows : []).map((r) => ({
    job_id: r.job_id, batch_id: r.batch_id || null, cmd: r.cmd, account: r.account || null,
    status: r.status === "running" ? "stuck" : r.status,
=== job-registry 240-270 ===
  const b = { label: job.batchLabel || id, size: Number(job.batchSize) || 0, ok: 0, failed: 0, done: 0, timer: null };
  BATCHES.set(id, b);
  notifyTelegram(`▶️ Batch · ${b.label}\nstarting${b.size ? ` · ${b.size} target${b.size === 1 ? "" : "s"}` : ""} (${id})`).catch(() => {});
}
function batchJobEnd(job) {
  const id = job.batch;
  const b = BATCHES.get(id);
  if (!b) return;
  b.done++;
  if (job.ok === true) b.ok++; else if (job.ok === false) b.failed++;
  if (b.timer) clearTimeout(b.timer);
  if (b.size && b.done >= b.size) { flushBatch(id); return; }
  b.timer = setTimeout(() => flushBatch(id), BATCH_IDLE_MS);
  if (b.timer.unref) b.timer.unref();
}

function notifyJobStart(job) {
  if (!tgJobsOn() || !job || job.util || (job.scheduled && !TG_SCHEDULED)) return;
  if (job.batch) { batchJobStart(job); return; }   // child → rolled into the batch alert, no per-job spam
  const who = job.account ? ` · ${job.account}` : "";
  notifyTelegram(`▶️ ${job.label || job.cmd || "job"}${who}\nstarted (${job.id})`).catch(() => {});
}
function notifyJobEnd(job) {
  if (!tgJobsOn() || !job || job.util || (job.scheduled && !TG_SCHEDULED)) return;
  if (job.batch) { batchJobEnd(job); return; }   // child → tallied into the batch summary
  const secs = job.startedAt && job.endedAt ? Math.round((job.endedAt - job.startedAt) / 1000) : null;
  const ok = job.ok === true;
  const icon = ok ? "✅" : job.ok === false ? "❌" : "☑️";
  const state = ok ? "ok" : job.ok === false ? "failed" : "done";
  const who = job.account ? ` · ${job.account}` : "";
  const tail = job.aborted_reason ? `\n${String(job.aborted_reason).slice(0, 200)}`

The file /home/philip/Documents/GitHub/bluebut-issue-52/scripts/_lib/adapter-finalize.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-issue-52/server/http/runtime/health.mjs has been updated successfully. (file state is current in your context — no need to Read it back)

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

152:          op, access, account: kwargs.account || null, status: 'blocked', ok: false,
189:          op, access, account: acct, status: 'blocked', ok: false,
229:          op, access, account: acct, status: 'blocked', ok: false,
264:          op, access, account: acct, status: 'blocked', ok: false,

The file /home/philip/Documents/GitHub/bluebut-issue-52/scripts/bluebut-adapter.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-issue-52/scripts/bluebut-adapter.ts has been updated. All occurrences were successfully replaced. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-issue-52/server/http/routes/accounts-mint.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-issue-52/server/http/routes/accounts-mint.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-issue-52/adapters/_lib/e2ee-event-connect.ts has been updated successfully. (file state is current in your context — no need to Read it back)

3:import { Hono } from 'hono'
4:import { z } from 'zod'
5:import { streamSSE } from 'hono/streaming'
6:import { zValidator } from '@hono/zod-validator'
7:import { EnqueueRequest } from '../contract/job.ts'
8:import { catalogSummary } from '../jobs/catalog.ts'
9:import { enqueue, enqueueRaw, getJob, listJobs, isLedgerTerminal, okForStatus } from '../jobs/queue.ts'
10:import { subscribe } from '../jobs/job-stream.ts'
81:        await stream.writeSSE({ event: 'result', data: JSON.stringify({ job_id: id, status: job.status, ok: job.ok ?? okForStatus(job.status), result: job.result ?? null }) })

The file /home/philip/Documents/GitHub/bluebut-issue-52/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-issue-52/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-issue-52/server/jobs/queue.ts has been updated successfully. (file state is current in your context — no need to Read it back)

180	  const failures = (Array.isArray(rows) ? rows : []).map((r) => ({
181	    job_id: r.job_id, batch_id: r.batch_id || null, cmd: r.cmd, account: r.account || null,
182	    status: r.status === "running" ? "stuck" : r.status,
183	    reason: r.aborted_reason || null,
184	    target: r.target_name || r.target_url || null,
185	    started_at: r.started_at || null, ended_at: r.ended_at || null,
186	    github_issue: r.summary && typeof r.summary === "object" ? (r.summary.github_issue ?? null) : null,
187	  }));
188	  return { ok: true, failures };
189	}
190	
191	// Operator stop: kill a job by id. Running → the installed cancel() SIGKILLs the
192	// child (its close handler then writes the terminal [RESULT]/[ERROR] + ends the
193	// SSE). Queued → cancel() drops it from the account queue and closes its stream.
194	// Already finished → no-op. Returns {code, ok, ...} for the HTTP layer.

6:import { notifyTelegram } from "../../../adapters/_lib/notify/tg-notify.ts";
7:import { rest as dbRest } from "../../../adapters/_lib/db/db.ts";
8:import { mintJobId } from "../../jobs/job-id.ts";
9:import { API_TOKEN, rememberCloakCdp } from "../lib.mjs";
10:import { envFile } from "./spawn-env.mjs";
11:import { numEnv } from "../../../adapters/_lib/num-env.ts";

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

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

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

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

140	const { traceId, via, bridge } = await resolveTraceId(JOB);
141	
142	const ledgerRows = await rest(HB_URL, HB_KEY,
143	  `run?id=eq.${encodeURIComponent(traceId)}&select=*&limit=1`);
144	const ledger = ledgerRows && ledgerRows[0] || null;
145	
146	const harKey = (ledger && ledger.har_key) || `job/${traceId}.har.gz`;
147	
148	const har = await fetchHar(harKey);
149	const harA = har ? analyzeHar(har) : null;
150	
151	// ── verdict: CONFIRMED vs UNCONFIRMED ────────────────────────────────────────
152	const confirmed = [];
153	const unconfirmed = [];
154	const signals = [];
155	
156	if (ledger) {
157	  if (ledger.ok === true) confirmed.push(`ledger: terminal status='${ledger.status}', duration ${ledger.duration_ms}ms`);
158	  else unconfirmed.push(`ledger ok=${ledger.ok} status='${ledger.status}'${ledger.aborted_reason ? ` (${ledger.aborted_reason})` : ''}`);
159	} else {
160	  unconfirmed.push(`NO ledger row for ${traceId} — the runner never wrote it (job died pre-write, or .env/bofrid key missing on the runner).`);
161	}
162	if (harA) {
163	  confirmed.push(`HAR present: ${harA.entries} entries, ${harA.ws_frames} WS frames (${harA.ws_sockets} sockets), ${harA.graphql.length} graphql calls, ${harA.ls_payloads} LightSpeed payloads`);
164	  if (harA.ws_frames === 0) unconfirmed.push('0 WebSocket frames in HAR — if this job touched Messenger/MQTT, the browser-level WS tap may not have attached (or there was genuinely no WS traffic).');
165	  if (harA.e2ee_signals.length) confirmed.push(`E2EE/restore evidence on the wire: ${harA.e2ee_signals.join(', ')}`);
166	  for (const e of harA.errors.slice(0, 10)) signals.push(`HTTP ${e.status} ${e.url}`);
167	} else {
168	  unconfirmed.push(`NO HAR blob at ${harKey} — nothing lossless to confirm against.`);
169	}
170	
171	const report = {
172	  input: JOB, traceId, resolvedVia: via,
173	  bridge: bridge ? { id: bridge.id, cmd: bridge.cmd, account: bridge.account, status: bridge.status, ok: bridge.ok } : null,
174	  ledger: ledger ? { cmd: ledger.cmd, account: ledger.account, status: ledger.status, ok: ledger.ok, op: ledger.op, target: `${ledger.target_type || ''}:${ledger.target_id || ''}`, summary: ledger.summary, har_key: ledger.har_key, duration_ms: ledger.duration_ms } : null,
175	  har: harA, confirmed, unconfirmed, signals,
176	};
177	
178	if (JSON_OUT) { console.log(JSON.stringify(report, null, 2)); process.exit(0); }
179

The file /home/philip/Documents/GitHub/bluebut-issue-52/scripts/analyze-job.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-issue-52/scripts/analyze-job.ts has been updated successfully. (file state is current in your context — no need to Read it back)

=== queryKeys 172-182 ===
  groupPostsForAccount:      (account: string) => ({ queryKey: ["groupPostsFor", account],    queryFn: () => dbGet<GroupPost>("group_post", { account: `eq.${account}`, select: "id,fb_post_id,permalink_url,message,fb_group_id,group_name,account,status,skip_reason,created_at,posted_as,as_page_id", order: "created_at.desc", limit: "500" }) }),
  // The whole durable `job` ledger (most-recent N), agnostic across every
  // target/account/site — the source for the Jobs page's History & analytics
  // view (the durable counterpart to the backend's wiped-on-restart in-memory
  // list). Capped high enough to chart trends; indexed (created_at DESC).
  allJobs: (limit = 5000) => ({ queryKey: ["allJobs", limit], queryFn: () => dbGet<Job>("run", { select: "job_id,batch_id,site,cmd,op,access,account,target_type,target_id,target_url,target_name,status,ok,aborted_reason,summary,started_at,ended_at,duration_ms,har_key,created_at", order: "created_at.desc", limit: String(limit) }) }),
  // Per-group job rollup (count + last date) for the Groups index columns.
  groupJobStats: () => ({ queryKey: ["groupJobStats"], queryFn: () => dbGet<GroupJobStat>("group_job_stats", { select: "fb_group_id,jobs_count,last_job_at" }) }),
  // Per-group rollup of the posts we made (group_post, status='posted') for the
  // Groups index "Our posts" / "Last post" columns.
  groupPostStats: () => ({ queryKey: ["groupPostStats"], queryFn: () => dbGet<GroupPostStat>("group_post_stats", { select: "fb_group_id,posts_count,last_post_at,listings_count,last_listing_at" }) }),
=== GroupPostView 70-78, 190-200 ===
  // The job that made it — its verdict, not just its id.
  const jobQ = useQuery<Job | null>({
    queryKey: ["job", p?.job_id],
    enabled: !!p?.job_id,
    queryFn: async () => (await dbGet<Job>("run", { job_id: `eq.${p!.job_id}`, select: "*", limit: "1" }))[0] ?? null,
  });
  // The scraped mirror of our own post. Joined on the NUMERIC fb post id, never on the
  // permalink string: FB serves the same story as both /permalink/<id>/ and /posts/<id>/,
  // so a url equality test misses half of them.
          note={neverRan > 0 ? `${neverRan} target${neverRan === 1 ? "" : "s"} never ran` : undefined}
        />
        {Arrow}
        <Rung
          label="JOB"
          value={p.job_id ? (jobQ.data?.status ?? p.job_id.slice(0, 12)) : "—"}
          to={p.job_id ? `/runs/${p.job_id}` : undefined}
          tone={jobQ.data && !jobQ.data.ok ? "var(--color-danger-fg)" : undefined}
        />
        {Arrow}
        <Rung label="MIRROR" value={mirrorId ? "scraped fb_post" : "not scraped"} to={mirrorId ? `/post/${mirrorId}` : undefined} />
=== db-types.gen 1115-1122 ===
  target_id: string | null;
  target_url: string | null;
  target_name: string | null;
  status: string | null;
  ok: boolean | null;
  aborted_reason: string | null;
  summary: unknown | null;
  started_at: string | null;

14:export const TERMINAL: ReadonlySet<Lifecycle> = new Set<Lifecycle>([
17:export const isFinal = (s: Lifecycle | string): boolean => TERMINAL.has(s as Lifecycle);
31:export function classify(richStatus: string | null | undefined): Lifecycle {
46:export function rollup(childStatuses: (string | null | undefined)[], opts: { aborted?: boolean } = {}): Lifecycle {
71:export function batchRollup(
93:export const lifecycleTone: Record<Lifecycle, "ok" | "warn" | "bad" | "run" | "muted"> = {
=== phaseFromStatus def ===
ui/src/components/AccountJobs.tsx:33:function phaseFromStatus(status: string | null | undefined, ok: boolean | null | undefined): Phase {
ui/src/components/AccountJobs.tsx:123:        phase: live ? phaseOfBridge(live) : phaseFromStatus(j.status, j.ok),

25	type Phase = "running" | "queued" | "ok" | "failed";
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" };
28	
29	// Durable-ledger phase — derive from `status` (the authoritative field the single
30	// funnel writes), NOT the retiring `run.ok` boolean (#295: read-side off run.ok).
31	// Any terminal status that isn't `ok` is a failure; `ok` is only a fallback for a
32	// row with no status at all (never the case for a durable row).
33	function phaseFromStatus(status: string | null | undefined, ok: boolean | null | undefined): Phase {
34	  if (status === "running") return "running";
35	  if (status === "queued") return "queued";
36	  if (status === "ok") return "ok";
37	  if (status) return "failed";
38	  return ok === false ? "failed" : "ok";
39	}
40	function phaseOfBridge(j: BridgeJob): Phase {
41	  if (j.status === "running") return "running";
42	  if (j.status === "queued") return "queued";
43	  return j.ok === false ? "failed" : "ok";
44	}
45	
46	function ago(ts: number | null): string {
47	  if (!ts) return "";
48	  const s = Math.round((Date.now() - ts) / 1000);
49	  if (s < 60) return `${s}s ago`;
50	  if (s < 3600) return `${Math.round(s / 60)}m ago`;
51	  if (s < 86400) return `${Math.round(s / 3600)}h ago`;
52	  return `${Math.round(s / 86400)}d ago`;
53	}
54	const ms = (iso: string | null | undefined): number => (iso ? Date.parse(iso) || 0 : 0);
55	
56	// One unified row, whether it came from the durable ledger, live bridge, or both.
57	interface Row {
58	  key: string;          // job_id (the stable, durable id) — DataGrid row id
59	  bridgeId?: string;    // bridge job id, when a live/recent run matches → deep-link target
60	  label: string;
61	  cmd: string | null;
62	  op: string | null;
63	  phase: Phase;
64	  at: number;           // ms, sort + "ago"
65	  durationMs: number | null;
66	  live: boolean;
67	  queuePos: number | null;
68	  harKey: string | null; // durable HAR present → offer download
69	}
70	
71	// One row's HAR button — opens the in-app HAR explorer (DevTools-style request
72	// list + Headers/Cookies/Payload/Response/Timing) for this job. The viewer
73	// itself keeps a "Download .har" escape hatch for opening in real Chrome
74	// DevTools. stopPropagation so clicking it doesn't also trigger the row's
75	// navigate-to-Job-detail.
76	function HarCell({ jobId }: { jobId: string }) {
77	  const [open, setOpen] = useState(false);
78	  return (
79	    <>
80	      <Button
81	        variant="secondary"
82	        title="Explore this job's full HAR — every request, header, cookie & body — in the dashboard"
83	        onClick={(e) => { e.stopPropagation(); setOpen(true); }}
84	        style={{ display: "inline-flex", alignItems: "center", gap: 5, padding: "2px 7px", fontSize: 11 }}
85	      >
86	        <FileSearch size={12} />
87	        HAR
88	      </Button>
89	      {open && <HarViewer jobId={jobId} open={open} onClose={() => setOpen(false)} />}
90	    </>
91	  );
92	}
93	
94	export function AccountJobs({ account }: { account: string }) {
95	  // Durable history (survives bridge restarts).
96	  const { data: ledger, isError: ledgerErr } = useQuery<Job[]>({ ...Q.jobsForAccount(account), refetchInterval: 5000, retry: false });
97	  // Live overlay (instant status for in-flight runs).
98	  const { data: bridge } = useQuery<JobsSnapshot>({ queryKey: ["bridge-jobs"], queryFn: fetchBridgeJobs, refetchInterval: (q) => bridgeJobsRefetch(q.state.data), retry: false });
99	
100	  const rows = useMemo<Row[]>(() => {
101	    // Index live bridge jobs by the id that threads to the runner (traceJobId =
102	    // the UI/runner job_id = the ledger's job_id), falling back to the backend id.
103	    const liveAll = [...(bridge?.active ?? []), ...(bridge?.queued ?? []), ...(bridge?.recent ?? [])]
104	      .filter((j) => j.account === account);
105	    const liveByJobId = new Map<string, BridgeJob>();
106	    for (const j of liveAll) liveByJobId.set(j.traceJobId || j.id, j);
107	
108	    const byKey = new Map<string, Row>();
109	
110	    // Durable rows first — the authoritative history.
111	    for (const j of ledger ?? []) {
112	      const live = liveByJobId.get(j.job_id);
113	      // job_id is typed non-null, but a run-ledger row can arrive with a null
114	      // job_id (e.g. the half-applied #305 job→run rename); synthesize a stable
115	      // unique key so the DataGrid always has a valid row id (MUI X error #85).
116	      const k = j.job_id || live?.id || `run:${j.op ?? j.cmd ?? "job"}:${ms(j.ended_at) || ms(j.started_at) || ms(j.created_at)}`;
117	      byKey.set(k, {
118	        key: k,
119	        bridgeId: live?.id,
120	        label: live?.label || `${j.site ? `${j.site}/` : ""}${j.cmd}`,
121	        cmd: j.cmd,
122	        op: j.op,
123	        phase: live ? phaseOfBridge(live) : phaseFromStatus(j.status, j.ok),
124	        at: ms(j.ended_at) || ms(j.started_at) || ms(j.created_at),
125	        durationMs: j.duration_ms ?? null,
126	        live: live ? live.status === "running" || live.status === "queued" : j.status === "running",
127	        queuePos: live?.queuePos ?? null,
128	        harKey: j.har_key ?? null,
129	      });

ui/src/pages/MemberRequests.tsx:87:      select: "job_id,account,target_id,target_name,status,ok,summary,created_at,duration_ms",
ui/src/lib/util/queryKeys.ts:177:  allJobs: (limit = 5000) => ({ queryKey: ["allJobs", limit], queryFn: () => dbGet<Job>("run", { select: "job_id,batch_id,site,cmd,op,access,account,target_type,target_id,target_url,target_name,status,ok,aborted_reason,summary,started_at,ended_at,duration_ms,har_key,created_at", order: "created_at.desc", limit: String(limit) }) }),
ui/src/pages/Overview.tsx:61:      select: "job_id,cmd,op,account,target_type,target_id,target_name,status,ok,created_at",
=== jobsForAccount select ===
166:  jobsForAccount: (account: string) => ({ queryKey: ["jobsForAccount", account], queryFn: () => dbGet<Job>("run", { account: `eq.${account}`, order: "created_at.desc", limit: "500" }) }),

=== MemberRequests .ok reads ===
=== Overview .ok reads ===
165:        dbCount("run", { cmd: "eq.approve-join-request", status: "eq.ok", created_at: `gte.${day}` }),

The file /home/philip/Documents/GitHub/bluebut-issue-52/ui/src/lib/util/queryKeys.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-issue-52/ui/src/pages/MemberRequests.tsx has been updated successfully. (file state is current in your context — no need to Read it back)

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

export function classify(richStatus: string | null | undefined): Lifecycle {
  const s = String(richStatus ?? "").toLowerCase();
  if (!s) return "completed";
  if (s === "running" || s === "queued" || s === "created") return s as Lifecycle;
  if (s === "pending") return "running";
  if (TERMINATED_RICH.has(s)) return "terminated";
  if (FAILED_RICH.has(s)) return "failed";
  if (s === "timed_out" || s === "canceled" || s === "terminated") return s as Lifecycle;
  if (s === "warn") return "completed";
  if (s.endsWith("_failed")) return "failed";
  return "completed";
}

// Aggregate a Batch's lifecycle from its jobs (mirror of taxonomy.js rollup()).
// Mixed → "partial"; aborted-with-nothing-good → "terminated".
=== GroupPostView imports of taxonomy ===

93	export const lifecycleTone: Record<Lifecycle, "ok" | "warn" | "bad" | "run" | "muted"> = {
94	  completed: "ok",
95	  partial: "warn",
96	  terminated: "warn",
97	  failed: "bad",
98	  canceled: "muted",

(Bash completed with no output)

18:import { useMemo } from "react";
19:import { useQuery } from "@tanstack/react-query";
20:import { Alert, AlertTitle, Box, Chip, Divider, Link, Skeleton, Stack, Typography } from "@mui/material";
21:import { PageHeader } from "../../components/PageHeader";
22:import { ReadField } from "../../components/ui/fields";
23:import { PageLink, GroupLink } from "../../components/EntityLinks";
24:import { dbGet, fetchWorkflowRun, type DbWorkflowRun, type GroupPost, type Job } from "../../api";
25:import { Q, posterLabel } from "../../lib/util";
26:import { StatusChip } from "../../components/StatusChip";
27:import { useCloakboxOpener } from "../../components/OpenInCloakbox";
28:import { fmtStamp } from "./listing";

The file /home/philip/Documents/GitHub/bluebut-issue-52/ui/src/pages/broadcast/GroupPostView.tsx has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-issue-52/ui/src/pages/broadcast/GroupPostView.tsx has been updated successfully. (file state is current in your context — no need to Read it back)

The file /home/philip/Documents/GitHub/bluebut-issue-52/db/schema.ts has been updated successfully. (file state is current in your context — no need to Read it back)

=== any run/job select still naming ok ===
=== any remaining 'ok:' or 'ok,' in a POST/PATCH run body ===
server/http/runtime/health.mjs:166:      ok: true, probed_at: Date.now(), total: probed.length, live,
server/http/runtime/health.mjs:190:      ok: true, ...cfg,
server/http/routes/store.ts:124:    ok: true,
server/http/runtime/scheduler.mjs:114:      child.on("error", (e) => { clearTimeout(killTimer); job.ok = false; resolve({ ok: false, jobId, error: e.message }); done(); });
server/http/routes/cloakbox.ts:68:    op: "read", access: "auth", account: slug, status: "ok", ok: true,
server/http/routes/cloakbox.ts:196:            ok: false,
server/http/routes/cloakbox.ts:231:      ok: true, account, slug, seed, url, reused, navigated,
server/http/routes/api-v1.ts:222:    ok: true,
server/jobs/queue-status.test.mjs:15:  for (const r of [{ ok: true }, { ok: false }, { ok: false, status: 'skipped' },
server/jobs/queue-status.test.mjs:16:    { ok: false, status: 'blocked' }, { ok: true, status: 'partial' },

58	// every later velocity gate (the "uncounted" hole). Terminal status so it never
59	// holds the one-active-per-account lane. Reuse-navigations open no new session →
60	// no row (that call site never runs this).
61	export function uiOpenLedgerRow(
62	  slug: string,
63	  url: string,
64	  { jobId, nowIso = new Date().toISOString() }: { jobId?: string; nowIso?: string } = {},
65	) {
66	  return {
67	    job_id: jobId || `uiopen-${slug}-${Date.now()}`, site: "facebook", cmd: "ui-open",
68	    op: "read", access: "auth", account: slug, status: "ok", ok: true,
69	    summary: { ui_open: true, url }, started_at: nowIso, ended_at: nowIso,
70	    duration_ms: 0, updated_at: nowIso,
71	  };
72	}
73	
74	interface CloakboxDeps {
75	  liveInstances: () => Promise<Array<Record<string, unknown>>>;

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

=== all POST/PATCH to run (to eyeball for ok) ===
server/http/routes/cloakbox.ts:210:        await dbRest("POST", "run", uiOpenLedgerRow(slug, url) as unknown as Parameters<typeof dbRest>[2], { silent: true });
server/http/routes/accounts-mint.ts:49:    await dbRest("POST", "run", [{
server/http/runtime/health.mjs:272:          .then((num) => { if (num) dbRest("PATCH", `run?id=eq.${encodeURIComponent(r.id)}`, { summary: { github_issue: num } }, { silent: true }).catch(() => {}); })
server/jobs/queue.ts:90:  await rest('POST', 'run', {
server/jobs/queue.ts:132:  await rest('POST', 'run', {
server/jobs/queue.ts:209:  await rest('PATCH', `run?id=eq.${encodeURIComponent(job_id)}&status=in.(${LEDGER_ACTIVE.join(',')})`,
scripts/bluebut-adapter.ts:150:        await dbRest('POST', 'run', {
scripts/bluebut-adapter.ts:187:        await dbRest('POST', 'run', {
scripts/bluebut-adapter.ts:227:        await dbRest('POST', 'run', {
scripts/bluebut-adapter.ts:262:        await dbRest('POST', 'run', {
scripts/bluebut-adapter.ts:339:    await dbRest('POST', 'run', { ...ledgerBase, status: 'running', started_at: new Date(t0).toISOString() },
server/http/routes/bridge.ts:344:      dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(id)}`,
server/http/routes/bridge.ts:357:    dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(id)}`,
server/http/routes/bridge.ts:361:      .then((r: { ok: boolean }) => dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(id)}`, { last_run_status: r.ok ? "ok" : "failed", updated_at: new Date().toISOString() } as unknown as Parameters<typeof dbRest>[2], { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {}))
scripts/_lib/adapter-finalize.ts:411:    await dbRest('PATCH', `run?id=eq.${encodeURIComponent(ledgerJobId)}`, {
adapters/_lib/e2ee-event-connect.ts:87:    await rest('POST', 'run', job, { silent: true, headers: { Prefer: 'resolution=merge-duplicates' } });
adapters/_lib/post-login-orchestrator.ts:53:  try { await rest('POST', 'run', job, { silent: true, headers: { Prefer: 'resolution=merge-duplicates' } }); }
server/http/runtime/scheduler.mjs:287:            return dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
server/http/runtime/scheduler.mjs:347:          dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
server/http/runtime/scheduler.mjs:354:          .then(() => dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
server/http/runtime/scheduler.mjs:357:          .catch((e) => { console.error(`[backend] schedule ${s.schedule_id} enqueue error: ${e?.message || e}`); return dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`, { last_run_status: "failed", updated_at: new Date().toISOString() }, { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {}); })
server/http/runtime/scheduler.mjs:364:        .then((res) => dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,

=== post-login-orchestrator job ok? ===
10://     (type-login.ts), synchronous + operator-invoked, and STOPPED at save.
14:// wire) produced an fb_logins row and nothing else — cookies unbanked past the
21://   2. bank cookies (saveProfile) — best-effort, a no-op off a live window,
22://   3. arm e2ee if usync is down — reuses the health-gated on-load hook
107:      job_id: key, site: 'facebook', cmd: 'post-login-ready', op: 'facebook.post_login_ready',
118:    // 2. BANK cookies — saveProfile is idempotent + a no-op off a live window (returns
123:    // 3. ARM e2ee — the SAME health-gated on-load hook: a live usync socket ⇒ silent
135:    // cookie-bank are best-effort environmentals — recorded, but they don't gate ready.
139:      job_id: key, site: 'facebook', cmd: 'post-login-ready', op: 'facebook.post_login_ready',
141:      status: steps.stamped ? (ready ? 'ok' : 'partial') : 'failed', ok: ready,
=== ledgerBase def ===
327:  const ledgerBase = {
339:    await dbRest('POST', 'run', { ...ledgerBase, status: 'running', started_at: new Date(t0).toISOString() },

job_id: key, site: 'facebook', cmd: 'post-login-ready', op: 'facebook.post_login_ready',
      access: 'write', account, target_type: 'account', target_id: account, target_name: account,
      status: steps.stamped ? (ready ? 'ok' : 'partial') : 'failed', ok: ready,
      aborted_reason: ready ? null : (steps.stamp_error || `wiretap ${steps.wiretap?.reason || 'unconfirmed'}`),
      summary: { trigger: 'login-event', fb_id: ev.fb_id || null, ready, ...steps },
--- also line 107 block ---
  try {
    await ledger({
      job_id: key, site: 'facebook', cmd: 'post-login-ready', op: 'facebook.post_login_ready',
      access: 'write', account, target_type: 'account', target_id: account, target_name: account,
      status: 'running', summary: { trigger: 'login-event', fb_id: ev.fb_id || null, login_job: ctx.job_id || null },
      started_at: startIso, updated_at: startIso,
    });

--- ledgerBase 327-340 ---
  const ledgerBase = {
    id: ledgerJobId, workflow_run_id: batch_id, site: cmd.site || null, cmd: cmd.name, op, access,
    account: account === 'na' ? null : account,
    target_type: target.type, target_id: target.id, target_url: target.url, target_name: target.name,
    // Soft join to the engine store (id-collapse). cloak_runs lives in the SEPARATE
    // cloak Postgres (CLOAKBOX_DATABASE_URL) → no cross-DB FK; we carry the run id as
    // a plain column so a job row links to the exact browser launch that drove it
    // (its HAR/captures/egress). Set on browser jobs (setupBrowserJob → br.runId from
    // loadProfile's /api/start), null for browserless/CLI runs.
    engine_run_id: br.runId || null,
  };
  try {
    await dbRest('POST', 'run', { ...ledgerBase, status: 'running', started_at: new Date(t0).toISOString() },
      { headers: { Prefer: 'resolution=merge-duplicates,return=minimal' }, silent: true });

The file /home/philip/Documents/GitHub/bluebut-issue-52/adapters/_lib/post-login-orchestrator.ts has been updated successfully. (file state is current in your context — no need to Read it back)

=== contract/job.ts 30-50 ===
])

// The durable job row, as the API returns it (a projection of the `job` ledger table).
export const Job = z.object({
  job_id: z.string(),
  verb: z.string().nullable().optional(),
  cmd: z.string().nullable().optional(),
  account: z.string().nullable().optional(),
  target_type: z.string().nullable().optional(),
  target_id: z.string().nullable().optional(),
  status: JobStatus,
  ok: z.boolean().nullable().optional(),
  aborted_reason: z.string().nullable().optional(),
  summary: z.record(z.string(), z.unknown()).nullable().optional(),
  created_at: z.string().nullable().optional(),
  started_at: z.string().nullable().optional(),
  ended_at: z.string().nullable().optional(),
  duration_ms: z.number().nullable().optional(),
})

// POST /jobs — enqueue a job. `verb` selects a catalog entry; `input` is validated
=== queue-status.test statusPatch assertions ===
5:// SUCCESS='ok' (NOT 'done'), plus the benign terminals skipped/partial/
9:import { isLedgerTerminal, LEDGER_ACTIVE, LEDGER_TERMINAL, okForStatus, statusPatch } from './queue.ts'
15:  for (const r of [{ ok: true }, { ok: false }, { ok: false, status: 'skipped' },
16:    { ok: false, status: 'blocked' }, { ok: true, status: 'partial' },
17:    { ok: false, status: 'content_unavailable' }]) {
25:test("success ('ok') is terminal — the SSE-hang regression", () => {
26:  assert.equal(jobLifecycleStatus({ ok: true }, false), 'ok')
27:  assert.equal(isLedgerTerminal('ok'), true)
35:  for (const s of LEDGER_ACTIVE) assert.ok(!LEDGER_TERMINAL.includes(s), `${s} is in both sets`)
39:test('okForStatus: success terminals → true, other terminals → false, active → null', () => {
40:  for (const s of ['ok', 'done', 'completed']) assert.equal(okForStatus(s), true, s)
42:    assert.equal(okForStatus(s), false, s)
43:  for (const s of LEDGER_ACTIVE) assert.equal(okForStatus(s), null, s)
46:test('okForStatus agrees with the runner on the decided failure/success cases', () => {
47:  // jobLifecycleStatus is the authoritative producer; the runner banks ok=(result.ok===true).
48:  assert.equal(okForStatus(jobLifecycleStatus({ ok: true }, false)), true)   // 'ok'  → ok=true
49:  assert.equal(okForStatus(jobLifecycleStatus({ ok: false }, false)), false) // 'failed' → ok=false
50:  assert.equal(okForStatus(jobLifecycleStatus({}, true)), false)             // walled 'blocked' → ok=false
53:test('statusPatch: a terminal stamps ended_at + derived ok; an active status does not', () => {
54:  const f = statusPatch('failed', 'boom')
55:  assert.equal(f.status, 'failed'); assert.equal(f.ok, false)
56:  assert.equal(f.aborted_reason, 'boom'); assert.ok(f.ended_at); assert.ok(f.updated_at)
57:  const r = statusPatch('running')
58:  assert.equal(r.ok, null); assert.equal(r.ended_at, undefined); assert.ok(r.updated_at)
61:test('statusPatch preserves the exact old markFailed shape (behavior-preserving refactor)', () => {
62:  // Old markFailed wrote { status:'failed', ok:false, aborted_reason, ended_at, updated_at }.
63:  const p = statusPatch('failed', 'x'.repeat(999))
65:  assert.equal(p.ok, false)
67:  assert.ok(p.ended_at && p.updated_at)

53	test('statusPatch: a terminal stamps ended_at + derived ok; an active status does not', () => {
54	  const f = statusPatch('failed', 'boom')
55	  assert.equal(f.status, 'failed'); assert.equal(f.ok, false)
56	  assert.equal(f.aborted_reason, 'boom'); assert.ok(f.ended_at); assert.ok(f.updated_at)
57	  const r = statusPatch('running')
58	  assert.equal(r.ok, null); assert.equal(r.ended_at, undefined); assert.ok(r.updated_at)
59	})
60	
61	test('statusPatch preserves the exact old markFailed shape (behavior-preserving refactor)', () => {
62	  // Old markFailed wrote { status:'failed', ok:false, aborted_reason, ended_at, updated_at }.
63	  const p = statusPatch('failed', 'x'.repeat(999))
64	  assert.equal(p.status, 'failed')
65	  assert.equal(p.ok, false)
66	  assert.equal(String(p.aborted_reason).length, 500)   // truncated to 500 as before
67	  assert.ok(p.ended_at && p.updated_at)
68	})
69

The file /home/philip/Documents/GitHub/bluebut-issue-52/server/jobs/queue-status.test.mjs has been updated successfully. (file state is current in your context — no need to Read it back)

File created successfully at: /home/philip/Documents/GitHub/bluebut-issue-52/db/migrations/entity/0054_drop_run_ok.sql (file state is current in your context — no need to Read it back)

316	          _schedInflight.delete(s.schedule_id);
317	          continue;
318	        }
319	        acct = actor.account;
320	        actorReason = actor.reason;
321	      }
322	      const jobId = mintJobId();
323	      const startIso = new Date().toISOString();
324	      const next = nextRunAt({ ...s, last_run_at: startIso }, now);
325	      console.log(`[backend] schedule fire: ${s.schedule_id} (${s.label}) → ${s.site || "facebook"}/${s.cmd} as ${acct || "(browserless)"} [${actorReason}] job ${jobId}`);
326	      // CAS-claim the interval slot at START (issue #43.F3) — only if no other tick/
327	      // process already advanced last_run_at. Holds the interval gate across a bridge
328	      // restart mid-run AND prevents a redeploy-overlap double-fire; a crash can't
329	      // cause a re-fire storm. Lost the race → skip (a sibling already fired this
330	      // interval). The terminal status/ok is patched when the job finishes.
331	      const won = await casClaimSchedule(s.schedule_id, s.last_run_at ?? null,
332	        { last_run_at: startIso, next_run_at: next ? next.toISOString() : null, last_run_status: "running", last_run_job_id: jobId, last_run_account: acct, updated_at: startIso });
333	      if (!won) { _schedInflight.delete(s.schedule_id); continue; }
334	
335	      // ── PRODUCER (issue #52): when the durable worker owns dispatch, DON'T spawn the
336	      // adapter inline — enqueue a queued `run` row (jobId == run PK, batch_id =
337	      // schedule_id so every run is linkable back to its schedule) and let the worker
338	      // drain it. It resolves the same adapter by (site, cmd) + passes schedArgs verbatim,
339	      // so the runner runs IDENTICALLY to fireSchedule's spawn. last_run_status "ok" here
340	      // means ENQUEUED (launched) — mirroring the workflow lane's launch semantics; the
341	      // run's own success/failure is banked on its `run` ledger row, not on run_schedule.
342	      if (QUEUE_WORKER_ON) {
343	        const site = s.site || "facebook";
344	        const cmd = s.cmd;
345	        if (!adapterPath(site, cmd)) {
346	          console.error(`[backend] schedule ${s.schedule_id}: no adapter for ${site}/${cmd} — not enqueuing`);
347	          dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
348	            { last_run_status: "failed", updated_at: new Date().toISOString() },
349	            { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {});
350	          _schedInflight.delete(s.schedule_id);
351	          continue;
352	        }
353	        enqueueRaw(site, cmd, schedArgs(s, acct), { job_id: jobId, account: acct, batch_id: s.schedule_id })
354	          .then(() => dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
355	            { last_run_status: "ok", updated_at: new Date().toISOString() },
356	            { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {}))
357	          .catch((e) => { console.error(`[backend] schedule ${s.schedule_id} enqueue error: ${e?.message || e}`); return dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`, { last_run_status: "failed", updated_at: new Date().toISOString() }, { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {}); })
358	          .finally(() => _schedInflight.delete(s.schedule_id));
359	        continue;
360	      }
361	
362	      // LEGACY inline lane (worker off): spawn the adapter directly (the original path).
363	      fireSchedule(s, jobId, acct)
364	        .then((res) => dbRest("PATCH", `run_schedule?schedule_id=eq.${encodeURIComponent(s.schedule_id)}`,
365	          { last_run_status: res.ok ? "ok" : "failed", updated_at: new Date().toISOString() },
366	          { silent: true, headers: { Prefer: "return=minimal" } }).catch(() => {}))
367	        .catch((e) => console.error(`[backend] schedule ${s.schedule_id} error: ${e?.message || e}`))
368	        .finally(() => _schedInflight.delete(s.schedule_id));
369	    }
370	  } catch (e) { console.error(`[backend] schedule tick failed: ${e?.message || e}`); }

OK server/http/runtime/scheduler.mjs
OK server/http/runtime/job-registry.mjs
OK server/http/runtime/health.mjs

M adapters/_lib/e2ee-event-connect.ts
 M adapters/_lib/post-login-orchestrator.ts
 M db/schema.ts
 M scripts/_lib/adapter-finalize.ts
 M scripts/analyze-job.ts
 M scripts/bluebut-adapter.ts
 M server/http/routes/accounts-mint.ts
 M server/http/routes/cloakbox.ts
 M server/http/runtime/health.mjs
 M server/http/runtime/job-registry.mjs
 M server/http/runtime/scheduler.mjs
 M server/jobs/queue-status.test.mjs
 M server/jobs/queue.ts
 M server/routes/jobs.ts
 M ui/src/lib/util/queryKeys.ts
 M ui/src/pages/MemberRequests.tsx
 M ui/src/pages/Overview.tsx
 M ui/src/pages/broadcast/GroupPostView.tsx
?? db/migrations/entity/0054_drop_run_ok.sql
=== stat ===
 adapters/_lib/e2ee-event-connect.ts      |  2 +-
 adapters/_lib/post-login-orchestrator.ts |  2 +-
 db/schema.ts                             |  3 +-
 scripts/_lib/adapter-finalize.ts         |  2 +-
 scripts/analyze-job.ts                   | 11 ++++--
 scripts/bluebut-adapter.ts               |  8 ++---
 server/http/routes/accounts-mint.ts      |  4 +--
 server/http/routes/cloakbox.ts           |  2 +-
 server/http/runtime/health.mjs           |  4 +--
 server/http/runtime/job-registry.mjs     |  7 ++--
 server/http/runtime/scheduler.mjs        | 59 ++++++++++++++++++++++++++++----
 server/jobs/queue-status.test.mjs        | 15 ++++----
 server/jobs/queue.ts                     | 27 ++++++++-------
 server/routes/jobs.ts                    |  4 +--
 ui/src/lib/util/queryKeys.ts             |  2 +-
 ui/src/pages/MemberRequests.tsx          |  2 +-
 ui/src/pages/Overview.tsx                |  2 +-
 ui/src/pages/broadcast/GroupPostView.tsx |  3 +-
 18 files changed, 109 insertions(+), 50 deletions(-)

committed
efe6b264 feat(runtime): finish Skyvern cutover — scheduler→queue producer + drop run.ok (#52)

direct push to main is blocked - open a merge request (glab mr create).

fix/issue-52
efe6b264 feat(runtime): finish Skyvern cutover — scheduler→queue producer + drop run.ok (#52)
a50e9337 Merge branch 'fix/issue-53' into 'main'
18fb9f40 feat(accounts): Load is a verify-and-finalize workflow (#53)
=== remote branches ===
  origin/HEAD -> origin/main
  origin/fix/epic-207
  origin/fix/epic-6
  origin/fix/epic-77
  origin/fix/epic-79
  origin/fix/issue-185
  origin/main
  origin/verify/epic-6
=== status vs origin ===
## fix/issue-52...origin/main [ahead 1]

=== push config ===
current
refs/heads/main
=== explicit refspec push ===
remote: 
remote: To create a merge request for fix/issue-52, visit:        
remote:   https://gitlab.hostbun.cc/devdashco/bluebut/-/merge_requests/new?merge_request%5Bsource_branch%5D=fix%2Fissue-52        
remote: 
To https://gitlab.hostbun.cc/devdashco/bluebut.git
 * [new branch]        fix/issue-52 -> fix/issue-52

Creating merge request for fix/issue-52 into main in devdashco/bluebut

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

ERROR  
          
  All attempts fail:                                                                                                  
  #1: PUT https://gitlab.hostbun.cc/api/v4/projects/devdashco%2Fbluebut/merge_requests/17/merge: 400 {message: SHA    
  must be provided when merging}.

sha=efe6b264fc60911d1474a402d30d69a70472218c
! No pipeline running on fix/issue-52
✓ Merged!
https://gitlab.hostbun.cc/devdashco/bluebut/-/merge_requests/17

=== main contains our commit? ===
  origin/HEAD -> origin/main
  origin/main
LANDED on main

no result shipped