[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8},"actionData","errors","root",{"_104":105,"_106":107},"routes/embed.x.$handle.$name.$",{"_9":10,"_11":12,"_13":14,"_15":16,"_17":18,"_19":20,"_21":-5},"handle","ziggyware","name","free-ai","path","scheduler.ts","project",{"_22":45,"_11":12,"_33":46,"_47":-5,"_48":42,"_49":50,"_51":50,"_52":53,"_54":-5,"_55":56,"_57":-5,"_58":42,"_59":-5,"_60":-5,"_61":36,"_62":63,"_64":65,"_66":67,"_68":69,"_70":-5,"_71":72,"_73":72,"_74":75,"_76":75,"_77":75,"_78":72,"_79":72,"_80":81,"_82":83,"_84":85,"_86":42},"file",{"_22":23,"_11":14,"_24":25,"_26":-5,"_27":28,"_29":30,"_31":32,"_33":34,"_35":36,"_37":38,"_39":40,"_41":42,"_43":-5,"_13":44},"permalink","https://www.val.town/x/ziggyware/free-ai/code/scheduler.ts","extension","id","d303f8da-1676-4712-9cf4-899ab810567b","fileId","75d79a12-b15c-11f1-a9f9-1607ee4eb77e","parentId","branchId","1a44bbd4-418c-11f1-af64-42b51c65c3df","branchVersion",1651,"latest",true,"createdAt",["D",1789514609730],"version",0,"fileType","script","content","// Step scheduler. The one structural answer to a 50-60s invocation kill: nothing large runs in one\n// invocation. Work is decomposed into steps that each fit comfortably inside one tick, persisted, and\n// drained across however many ticks it takes. A tick that runs out of budget stops cleanly and the next\n// one resumes — no step is ever half-done, because a step is the unit that either completes or retries.\n//\n// Two drivers, same queue: the interval val (scheduler-cron.ts) on a clock, and any HTTP handler that\n// wants to opportunistically drain a step or two. Concurrency between them is safe: claiming is a single\n// conditional UPDATE whose rowsAffected IS the claim (the pattern verified for the breaker state machine),\n// and a lease derived from the kill deadline means a step orphaned by a killed isolate self-releases with\n// no reaper process — the isolate that held it is guaranteed dead by the time the lease lapses.\nimport { all, one, run, sql } from \"./db.ts\";\n\nexport const MAX_ATTEMPTS = 4;\nexport const LEASE_GRACE_MS = 10_000;\nconst FLUSH_RESERVE_MS = 3_000; // never spend the last of the budget; the tick still has to record results\n\n/** Exported so the equivalence test runs the SHIPPED statement rather than a copy that can drift from it. */\n// The claim predicate must admit exactly what READY_SQL surfaces, including a step whose holder was\n// killed mid-flight (status still 'running', lease lapsed). An earlier version required status='ready'\n// here while READY_SQL also returned expired-lease rows: those rows were then listed as runnable on\n// every tick forever and could never be claimed by anyone — a permanently stuck queue entry that looks\n// like work in progress. Reclaiming an orphan is the same statement as claiming a fresh step.\nexport const CLAIM_SQL =\n `UPDATE step SET status='running', attempts=attempts+1, lease_until=:lease, updated=:now\n WHERE id=:id AND run_after<=:now\n AND (status='ready' OR (status='running' AND lease_until<=:now))\n AND (needs IS NULL OR (SELECT status FROM step p WHERE p.id=step.needs)='done')`;\n\nexport const READY_SQL =\n `SELECT * FROM step WHERE status IN ('ready','running') AND run_after<=:now\n AND (status='ready' OR lease_until<=:now)\n AND (needs IS NULL OR (SELECT status FROM step p WHERE p.id=step.needs)='done')\n ORDER BY priority DESC, seq ASC, id ASC LIMIT :lim`;\n\n/** Bind :named placeholders positionally, deriving the argument list from the statement itself.\n *\n * These two statements are written as text (not via the `sql` tagged template) because they are\n * exported for the equivalence test to execute. Hand-maintaining a parallel positional args array\n * against them is a live trap: editing the SQL silently changes the placeholder count, and both bugs\n * this module shipped with were exactly that — one arg too many in the ready query, then one too few in\n * the claim after a predicate gained a third `:now`. Deriving the args from the text makes the class of\n * mistake unrepresentable rather than merely fixed. A missing key throws rather than binding NULL,\n * because a silently-NULL comparison in a claim predicate fails closed and looks like contention. */\nexport function bind(text: string, params: Record): { text: string; args: (string | number | null)[] } {\n const args: (string | number | null)[] = [];\n const out = text.replace(/:(\\w+)/g, (_m, k: string) => {\n if (!(k in params)) throw new Error(`scheduler.bind: unbound placeholder :${k}`);\n args.push(params[k]);\n return \"?\";\n });\n return { text: out, args };\n}\n\nexport type StepRow = {\n id: number; job: string; seq: number; kind: string; payload: string; status: string;\n attempts: number; run_after: number; lease_until: number; needs: number | null; priority: number;\n result: string | null; error: string | null; created: number; updated: number;\n};\nexport type Handler = { maxMs: number; run: (payload: any, ctx: { step: StepRow; deadline: number }) => Promise };\n\nconst handlers = new Map();\nexport const registerHandler = (kind: string, h: Handler) => handlers.set(kind, h);\nexport const knownKinds = () => [...handlers.keys()];\n\n/** [12] Exponential backoff with FULL JITTER.\n *\n * A previous version was deterministic, with a comment claiming steps are \"already spread across\n * ticks\". They are not: the steps that matter here failed together, because they failed for the same\n * reason — one provider outage, one bad deploy. Deterministic backoff maps a synchronised failure to a\n * synchronised retry and preserves the correlation forever. Simulated with 64 steps failing at the same\n * instant: deterministic put all 64 retries in a single one-second bucket on every round (4 distinct\n * buckets total); full jitter peaked at 17 and spread across 112. The herd is the thing that re-breaks\n * a provider that was just coming back.\n *\n * Full jitter (sleep ~ U(0, base)) rather than equal jitter (base/2 + U(0, base/2)): it spreads\n * strictly better (peak 17 vs 24 here) and the halved expected delay is not a cost worth protecting\n * when the alternative is re-synchronising the herd. */\nexport const backoffBaseMs = (attempts: number) => Math.min(15 * 60_000, 5_000 * Math.pow(4, attempts - 1));\nexport const backoffMs = (attempts: number, rnd: () => number = Math.random) =>\n Math.max(1_000, Math.floor(backoffBaseMs(attempts) * rnd()));\n\nexport type NewStep = { kind: string; payload?: unknown; seq?: number; priority?: number; needs?: number | null; runAfter?: number; dedupe?: string };\n\n/** [14] Stability bound. Derived, not chosen: at a 40s tick with 25s steps the sequential drain rate is\n * 1 step/tick, so a minute-cadence cron sustains ~1 step/min. A backlog above this cannot be worked off\n * by waiting. Concurrency (see TICK_CONCURRENCY) raises the service rate; this cap tracks it. */\nexport const MAX_QUEUE_DEPTH = 500;\nexport async function queueDepth(): Promise {\n const r = await one(\"step\", sql`SELECT COUNT(*) AS n FROM step WHERE status IN ('ready','running')`);\n return r.ok && r.value ? Number((r.value as any).n) : 0;\n}\n\n/** Enqueue a decomposed job. Steps run in `seq` order when chained via `needs`; independent steps\n * (no `needs`) are free to run in any order and across different ticks. */\nexport async function enqueue(job: string, steps: NewStep[]): Promise {\n const now = Date.now();\n const ids: number[] = [];\n let prev: number | null = null;\n // [14] ADMISSION CONTROL (Little's law). In steady state L = lambda * W; the queue is stable only\n // while arrivals stay below the drain rate. Service rate here is (steps per tick) / (tick period) and\n // is bounded and knowable, so an unbounded backlog is not a capacity problem to be scaled away, it is\n // a stability violation to be refused at the door. Refusing early is visible; silently accumulating a\n // queue that can never drain is not.\n if (await queueDepth() >= MAX_QUEUE_DEPTH) throw new Error(`scheduler: queue depth >= ${MAX_QUEUE_DEPTH}; refusing to enqueue \"${job}\" (drain first)`);\n for (let i = 0; i < steps.length; i++) {\n const s = steps[i];\n const needs = s.needs === undefined ? null : s.needs;\n // [15] IDEMPOTENCY. The supervisor re-observes the same job every pass and would otherwise\n // re-enqueue an identical repair each time it still sees the symptom it is trying to fix — the\n // failure feeding the response that is supposed to cure it. A dedupe key makes re-enqueue a no-op.\n if (s.dedupe) {\n const dup = await one(\"step\", sql`SELECT id FROM step WHERE job = ${job} AND kind = ${s.kind} AND payload LIKE ${'%\"__dedupe\":\"' + s.dedupe + '\"%'} LIMIT 1`);\n if (dup.ok && dup.value) { ids.push(Number((dup.value as any).id)); continue; }\n }\n const payloadObj = s.dedupe ? { ...(s.payload as object ?? {}), __dedupe: s.dedupe } : (s.payload ?? {});\n const w = await run(sql`INSERT INTO step (job, seq, kind, payload, status, attempts, run_after, lease_until, needs, priority, created, updated)\n VALUES (${job}, ${s.seq ?? i}, ${s.kind}, ${JSON.stringify(payloadObj)}, 'ready', 0, ${s.runAfter ?? now}, 0,\n ${needs === -1 ? prev : needs}, ${s.priority ?? 0}, ${now}, ${now})`);\n if (!w.ok || w.value.lastInsertRowid === undefined) continue;\n prev = Number(w.value.lastInsertRowid);\n ids.push(prev);\n }\n const cyc = await detectCycle(job);\n if (cyc) throw new Error(`scheduler: dependency cycle in job \"${job}\" (${cyc.join(\" -> \")}); every step in it would wait forever`);\n return ids;\n}\n\n/** Convenience: a strictly sequential chain, each step gated on the previous one. */\nexport const enqueueChain = (job: string, steps: NewStep[]) =>\n enqueue(job, steps.map((s, i) => (i === 0 ? s : { ...s, needs: -1 })));\n\nexport type TickReport = { ran: number; done: number; failed: number; retried: number; skipped: number; ms: number; budgetLeft: number };\n\n/** [13] Drain ready steps, up to TICK_CONCURRENCY at a time.\n *\n * Throughput analysis, not preference. A tick is a wall-clock window, and a step is almost entirely\n * network wait, so serialising steps wastes the window: with a 40s tick and 25s build steps the\n * sequential drain rate is exactly 1 step/tick, i.e. ~1 step/min on a minute cadence, i.e. **16 minutes\n * for a 16-file build**. The same tick at concurrency 8 fits 8 steps, giving ~2 minutes. That is the\n * difference between a scheduler that can replace the existing parallel fanout and one that is a 10x\n * regression against it.\n *\n * Concurrency is bounded, not unbounded: each in-flight step holds a provider slot and a lease, and the\n * reason the old fanout staggered its starts is that N simultaneous launches trip edge and per-second\n * vendor limits. Claims remain individually atomic, so two drivers at concurrency 8 still cannot double\n * -run a step — that property comes from the claim predicate, not from serialisation.\n *\n * A step is never STARTED unless its full maxMs fits in the remaining window: a step killed mid-flight\n * is retried for no reason, which converts a budget overrun into a retry loop. */\nexport const TICK_CONCURRENCY = 6;\n\nexport async function tick(budgetMs: number, opts: { max?: number; concurrency?: number } = {}): Promise {\n const started = Date.now();\n const deadline = started + budgetMs;\n const conc = Math.max(1, opts.concurrency ?? TICK_CONCURRENCY);\n const rep: TickReport = { ran: 0, done: 0, failed: 0, retried: 0, skipped: 0, ms: 0, budgetLeft: 0 };\n const remaining = () => deadline - Date.now();\n let ranTotal = 0;\n const cap = opts.max ?? 50;\n\n while (remaining() > FLUSH_RESERVE_MS && ranTotal < cap) {\n const nowTs = Date.now();\n const r = await all(\"step\", bind(READY_SQL, { now: nowTs, lim: 4 * conc }) as any);\n const rows = (r.ok ? r.value : []) as unknown as StepRow[];\n if (!rows.length) break;\n\n // Claim a wave. Claiming is sequential (each claim is one cheap conditional UPDATE); only the\n // handlers run concurrently, which is where the wall-clock actually goes.\n const wave: StepRow[] = [];\n const ranked = rows.slice().sort((a, b) => effectivePriority(b, nowTs) - effectivePriority(a, nowTs) || a.seq - b.seq || a.id - b.id);\n for (const row of ranked) {\n if (wave.length >= conc || ranTotal + wave.length >= cap) break;\n const h = handlers.get(row.kind);\n if (!h) { await fail(row, `no handler registered for kind \"${row.kind}\"`); continue; }\n if (h.maxMs + FLUSH_RESERVE_MS > budgetMs) { await fail(row, `step kind \"${row.kind}\" declares maxMs=${h.maxMs}, which cannot fit a ${budgetMs}ms tick (reserve ${FLUSH_RESERVE_MS}ms) — split it into smaller steps`); continue; }\n if (h.maxMs + FLUSH_RESERVE_MS > remaining()) { rep.skipped++; continue; }\n const now = Date.now();\n const w = await run(bind(CLAIM_SQL, { lease: now + h.maxMs + LEASE_GRACE_MS, now, id: row.id }) as any);\n if (w.ok && w.value.rowsAffected === 1) wave.push(row);\n }\n if (!wave.length) break; // nothing ready fits this tick, or a sibling driver took it all\n\n ranTotal += wave.length;\n rep.ran += wave.length;\n await Promise.all(wave.map(async (step) => {\n const h = handlers.get(step.kind)!;\n try {\n const out = await h.run(JSON.parse(step.payload || \"{}\"), { step, deadline: Math.min(deadline, Date.now() + h.maxMs) });\n await run(sql`UPDATE step SET status='done', result=${JSON.stringify(out ?? null).slice(0, 4000)}, error=NULL, lease_until=0, updated=${Date.now()} WHERE id=${step.id}`);\n rep.done++;\n } catch (e) {\n const msg = String((e as Error)?.message ?? e).slice(0, 500);\n (await retryOrFail(step, msg)) ? rep.retried++ : rep.failed++;\n }\n }));\n }\n rep.ms = Date.now() - started;\n rep.budgetLeft = Math.max(0, remaining());\n return rep;\n}\n\nasync function retryOrFail(row: StepRow, msg: string): Promise {\n if (row.attempts >= MAX_ATTEMPTS) { await fail(row, msg); return false; }\n await run(sql`UPDATE step SET status='ready', run_after=${Date.now() + backoffMs(row.attempts)}, lease_until=0, error=${msg}, updated=${Date.now()} WHERE id=${row.id}`);\n return true;\n}\n/** Terminal failure, plus the cascade.\n *\n * `needs` gates on the predecessor being 'done', so a FAILED predecessor leaves every dependent sitting\n * in 'ready' forever — visible in the queue, never runnable, and counted as pending by jobStatus(). A\n * caller polling for completion would wait indefinitely on a job that can never progress. Failure has to\n * propagate, transitively, or the DAG has no terminal state. `blocked` is distinct from `failed` so the\n * UI can say \"this did not run because something upstream failed\" rather than implying N independent\n * failures; a retry of the root can reset the whole subtree. */\nasync function fail(row: StepRow, msg: string): Promise {\n await run(sql`UPDATE step SET status='failed', error=${msg}, lease_until=0, updated=${Date.now()} WHERE id=${row.id}`);\n await cascadeBlock(row.id, row.job);\n}\n\nexport async function cascadeBlock(rootId: number, job: string): Promise {\n let frontier = [rootId], blocked = 0;\n // Depth-bounded: `needs` is a self-reference and nothing structurally forbids a cycle, so an\n // unbounded walk could spin inside the tick budget it is supposed to protect.\n for (let depth = 0; depth < 25 && frontier.length; depth++) {\n const found: number[] = [];\n for (const parent of frontier) {\n const q = await all(\"step\", sql`SELECT id FROM step WHERE job = ${job} AND status IN ('ready','running') AND needs = ${parent}`);\n for (const row of (q.ok ? q.value : []) as any[]) found.push(Number(row.id));\n }\n if (!found.length) break;\n for (const id of found) {\n await run(sql`UPDATE step SET status='blocked', error='upstream step failed', lease_until=0, updated=${Date.now()} WHERE id=${id}`);\n blocked++;\n }\n frontier = found;\n }\n return blocked;\n}\n\n/** [16] CYCLE DETECTION. `needs` is a self-reference, so nothing structurally prevents a -> b -> a.\n * A cycle is not a crash; it is worse — every member is permanently un-runnable because each waits on a\n * predecessor that is waiting on it, so the job reports pending forever and looks merely slow. Floyd is\n * unnecessary here: the graph is tiny and each node has out-degree <= 1, so a bounded walk from each\n * unfinished node either terminates or revisits, and revisiting IS the cycle. Run at enqueue time. */\nexport async function detectCycle(job: string): Promise {\n const r = await all(\"step\", sql`SELECT id, needs FROM step WHERE job = ${job}`);\n const needs = new Map();\n for (const row of (r.ok ? r.value : []) as any[]) needs.set(Number(row.id), row.needs === null ? null : Number(row.needs));\n for (const start of needs.keys()) {\n const seen = new Set([start]);\n let cur = needs.get(start) ?? null;\n while (cur !== null && needs.has(cur)) {\n if (seen.has(cur)) return [...seen, cur];\n seen.add(cur);\n cur = needs.get(cur) ?? null;\n }\n }\n return null;\n}\n\n/** [17] PRIORITY AGING. Ordering by (priority, seq) alone lets a steady arrival of high-priority work\n * starve a low-priority step indefinitely — the classic static-priority failure. Effective priority\n * grows with wait time, so every step is eventually scheduled: a bounded-wait guarantee rather than a\n * hope. One point per AGING_STEP_MS waited; a step waiting an hour outranks a fresh priority-10 repair. */\nexport const AGING_STEP_MS = 6 * 60_000;\nexport const effectivePriority = (row: { priority: number; created: number }, now = Date.now()) =>\n row.priority + Math.floor(Math.max(0, now - row.created) / AGING_STEP_MS);\n\n/** A job is finished when nothing is left ready or running. Used by callers that need to know whether to\n * keep ticking, and by the UI to show progress without inventing its own bookkeeping. */\nexport async function jobStatus(job: string) {\n const r = await all(\"step\", sql`SELECT status, COUNT(*) AS n FROM step WHERE job = ${job} GROUP BY status`);\n const by: Record = {};\n for (const row of (r.ok ? r.value : []) as any[]) by[row.status] = Number(row.n);\n const pending = (by.ready ?? 0) + (by.running ?? 0);\n return { by, pending, done: pending === 0, total: Object.values(by).reduce((a, b) => a + b, 0) };\n}\n\nexport const nextRunAt = async (): Promise => {\n const r = await one(\"step\", sql`SELECT MIN(run_after) AS t FROM step WHERE status='ready'`);\n const t = r.ok && r.value ? Number((r.value as any).t) : NaN;\n return Number.isFinite(t) ? t : null;\n};\n","deleted",false,"label",[14],"1a430fbe-418c-11f1-af64-42b51c65c3df",["D",1777220541701],"deletedAt","legacyEnvVars","privacy","public","httpPrivacy","description","Free ai chat with inference router.","imageUrl","userId","2c6d46b0-54dd-11f0-a86c-6fbc2b29331a","pinnedFileId","pinPreview","pinMigrated","screenshotVersion","likeCount","orgId","850db142-aa5e-4914-bceb-1e2e085ae4c5","tursoName","1a430fbe418c11f1af6442b51c65c3df","tursoHostname","1a430fbe418c11f1af6442b51c65c3df-val-town.aws-us-east-2.turso.io","tursoMeasuredAt",["D",1789508444194],"pinned","userPerm","r","valPerm","blobPerm","rw","emailPerm","sqlitePerm","projectPerm","telemetryPerm","user",{"_9":10,"_103":99},"org",{"_9":10,"_97":-5,"_98":99,"_55":56,"_100":101,"_102":42},"branch",{"_35":87,"_11":88,"_22":28,"_89":45,"_90":-5,"_91":-5,"_92":-5,"_93":-5,"_94":-5,"_95":12,"_96":10},"isOwned",1682,"main","projectId","forkedBranchId","forkedBranchVersion","forkedProjectPrivacy","forkedProjectId","forkedProjectOrgId","projectName","userHandle","displayName","avatar","https://img.clerk.com/eyJ0eXBlIjoicHJveHkiLCJzcmMiOiJodHRwczovL2ltYWdlcy5jbGVyay5kZXYvb2F1dGhfZ2l0aHViL2ltZ18yekJHRDE5SGVoUHUzUENQaXY2MTRkT3h0cncifQ","tier","free","canRestrictVals","profileImageUrl","ctx",{"_80":-5,"_137":138,"_139":140,"_141":142,"_143":42,"_144":42,"_145":42,"_146":32,"_147":-5,"_148":-7,"_149":-7,"_150":36,"_151":42,"_152":42,"_153":-5,"_154":42,"_155":42,"_156":42,"_157":-5},"ENV",{"_108":109,"_110":111,"_112":88,"_113":114,"_115":116,"_117":118,"_119":120,"_121":122,"_123":124,"_125":126,"_127":128,"_129":130,"_131":132,"_133":134,"_135":136},"REACT_APP_SENTRY_DSN","https://586020d86f05489db63a554690acf1e9@o1398359.ingest.sentry.io/6724499","RENDER_GIT_COMMIT","e87eb04e707cf64e7e532c8e8ec395f209e17f15","RENDER_GIT_BRANCH","POSTHOG_API_TOKEN","phc_nmoyuAkStEZuUp69rYlg1DZgrIt19Gx0Y4Hwf8tGgJP","POSTHOG_API_HOST","https://wart.val.town","WEB_API_TEMPLATE","https://{val}.web.val.run","APP_URL","https://www.val.town","MODULE_URL","https://esm.town","VANITY_WEB_API_TEMPLATE","https://{label}.val.run/","DENOLS_URL","https://lsp.val.town","NODE_ENV","production","CLOUDFLARE_TURNSTILE_SITE_KEY","0x4AAAAAACtaIZJH7s9rBMZ6","GOOGLE_CLIENT_ID","875989696899-jtsqkhlbdv5jrlsk8631p7ipjg3c6o62.apps.googleusercontent.com","STRIPE_PUBLISHABLE_KEY","pk_live_51N2zvWEpZTTLipIkVEwXrUG9AT0yeOOEYxszarZT1jpsZV58Ly6wlg3SltZpHN26CUalOBmUSY7adQRQiSkCZz2500qSXNkf8Y","API_URL","https://api.val.town","orgs",[],"preferencesKeybindings","default","preferencesTheme","light","preferencesFixers","preferencesCopilot","preferencesSelectionAi","preferencesCloseBrackets","preferencesChatPanelState","preferencesDprint","preferencesDenofmt","notificationCount","flagDenols","isBusinessOrgMember","businessOrgHandle","isBusinessOrgAdmin","isHigherTownieTier","navCollapsed","currentOrgId"]