From 3df6551608c0b4db6ef5630a0eac2d212a01f609 Mon Sep 17 00:00:00 2001 From: "prompt.ac/@jeffrey" Date: Fri, 9 Oct 2026 21:12:00 -0700 Subject: [PATCH] Whistlegraph turns, slice 2: a worker that runs the model loop and commits lith/whistlegraph-worker.mjs claims turns, compiles the same edit contract the phone compiles, runs AcServer against /api/easel-inference under the worker secret as the owner, checks the source, repairs once through runEditExperiment, and appends the version to the thread ledger with store.save (a revision that moved means the phone won; the turn fails 'moved' rather than forking). Checkpoints ride on the heartbeat so a takeover resumes from the last streamed code. No pixels yet: versions are marked unreviewed. DELETE /api/whistlegraph-turn?id= takes back a waiting turn. Tested: commit, race, and stale-base refusal. --- lith/whistlegraph-worker.mjs | 130 ++++++++++++++++++ lith/whistlegraph-worker.test.mjs | 35 +++++ system/backend/whistlegraph-turns.mjs | 5 + system/backend/whistlegraph-turns.test.mjs | 4 + .../netlify/functions/whistlegraph-turn.mjs | 10 +- 5 files changed, 182 insertions(+), 2 deletions(-) create mode 100644 lith/whistlegraph-worker.mjs create mode 100644 lith/whistlegraph-worker.test.mjs diff --git a/lith/whistlegraph-worker.mjs b/lith/whistlegraph-worker.mjs new file mode 100644 index 0000000000..33bbd50d24 --- /dev/null +++ b/lith/whistlegraph-worker.mjs @@ -0,0 +1,130 @@ +#!/usr/bin/env node +// A Whistlegraph turn worker (apple/whistlegraph/TURNS.md, slice 2). +// +// Claims turns from the queue, runs the model loop with the engine's own +// modules, checks the source, commits the version to the thread ledger, and +// moves on. Nothing here knows about the phone. No pixels yet: every version +// this worker saves is marked unreviewed; slice 3 adds the paint check and +// the picture review in a worker-thread isolate. +// +// Env: MONGODB_CONNECTION_STRING, MONGODB_NAME, WHISTLEGRAPH_WORKER_SECRET, +// AC_SITE (default https://aesthetic.computer), WORKER_NAME, TURN_CONCURRENCY (default 4) +import {MongoClient} from 'mongodb'; +import {mkdtempSync, readFileSync, writeFileSync, rmSync} from 'node:fs'; +import {join} from 'node:path'; +import {tmpdir, hostname} from 'node:os'; +import {mongoTurnQueue} from '../system/backend/whistlegraph-turns.mjs'; +import {mongoWhistlegraphStore, sourceHash} from '../system/backend/whistlegraph.mjs'; +import {AcServer} from '../aesel/src/ac-server.mjs'; +import {sourceChecks, compileEditContract, runEditExperiment} from '../aesel/src/edit-contract.mjs'; +import {selectedBranch} from '../apple/whistlegraph/Resources/Web/branch-context.mjs'; +import {GENERATION_INSTRUCTIONS, generationProfile} from '../apple/whistlegraph/Resources/Web/generation-policy.mjs'; +import {WARE_INSTRUCTIONS} from '../apple/whistlegraph/Resources/Web/wares.mjs'; + +const SITE = process.env.AC_SITE || 'https://aesthetic.computer'; +const WORKER = process.env.WORKER_NAME || `${hostname()}:${process.pid}`; +const CONCURRENCY = Math.max(1, Number(process.env.TURN_CONCURRENCY) || 4); +const HEARTBEAT_MS = 30_000; +const BASE_PIECE = 'export function paint({wipe}) { wipe("black"); }'; +const log = (...args) => console.log(new Date().toISOString(), `[${WORKER}]`, ...args); + +// The engine's fetch, minus the phone: the worker secret names the owner, and +// max_tokens follows the profile like the app's makeServer does. +function workerFetch(owner, settings) { + return async (url, options) => { + const body = JSON.parse(options.body); body.max_tokens = settings.maxTokens; + const headers = {...(options.headers || {}), 'x-ac-worker': process.env.WHISTLEGRAPH_WORKER_SECRET, 'x-ac-owner': owner}; + delete headers.Authorization; delete headers.authorization; + return fetch(url, {...options, headers, body: JSON.stringify(body)}); + }; +} + +// One turn: the same contract the phone compiles, the same model loop, the +// same single repair. Returns {source, findings, repairs, completed, error}. +export async function runTurn({job, thread, onCheckpoint = () => {}, signal}) { + const head = thread.ledger?.versions.find(v => v.id === thread.ledger.head); + if (!head || head.id !== job.baseVersion || sourceHash(head.source) !== job.baseHash) throw Object.assign(Error('The piece moved on before this turn ran'), {code: 'moved'}); + const cwd = mkdtempSync(join(tmpdir(), 'wg-turn-')); + const file = join(cwd, 'whistlegraph.mjs'); + try { + let source = head.source || BASE_PIECE; + writeFileSync(file, source.endsWith('\n') ? source : source + '\n'); + const prompt = compileEditContract({request: job.request.text, ...selectedBranch(thread.ledger), source}) + + '\nCurrent preview: 390 × 520 CSS points. Compose for this shape using screen.width and screen.height; keep subjects within the canvas and remain responsive when it resizes.'; + let completed = false, error = '', server = null, checkpointAt = 0; + const generate = async (task, repair) => { + const settings = generationProfile('', {repair, model: job.model || '', image: !!job.request.drawing}); + server = new AcServer({cwd, piece: {file, checkpoint: async () => { source = readFileSync(file, 'utf8'); const now = Date.now(); if (now - checkpointAt > 5000) { checkpointAt = now; await onCheckpoint(source); } }}, + token: 'worker', fetch: workerFetch(job.owner, settings), site: SITE, preview: false, frameCapture: false, model: settings.model, + rounds: settings.rounds, outputContinuations: settings.outputContinuations, reasoning: settings.reasoning, thinking: settings.thinking, + developerInstructions: GENERATION_INSTRUCTIONS + '\n' + WARE_INSTRUCTIONS}); + completed = false; error = ''; + server.on('notification', ({method, params}) => { + if (method === 'turn/completed') { completed = !params.turn.error && params.turn.status === 'completed'; error = params.turn.error ? String(params.turn.error.message || params.turn.error) : ''; } + }); + signal?.addEventListener('abort', () => server?.interrupt(), {once: true}); + await server.startTurn(task); + source = readFileSync(file, 'utf8'); + return completed; + }; + const inspect = async () => { const findings = sourceChecks(source); return {passed: findings.length === 0, findings, sourceHash: sourceHash(source)}; }; + const result = await runEditExperiment({prompt, generate, inspect, cancelled: () => signal?.aborted === true, onRepair: () => log('repairing', job.code)}); + if (result.cancelled) throw Object.assign(Error('Stopped'), {code: 'cancelled'}); + if (!result.completed) throw Error(error || 'The model did not finish'); + if (source.trim() === (head.source || '').trim()) throw Error('No change to the piece'); + return {source, findings: result.validation?.findings || [], repairs: result.repairs, completed: true}; + } finally { rmSync(cwd, {recursive: true, force: true}); } +} + +// Append the version to the thread, as the phone's commit does. store.save +// refuses if the revision moved, so a race with the phone cannot fork. +export async function commitVersion(store, job, thread, source, notes) { + const ledger = thread.ledger; + const id = Math.max(...ledger.versions.map(v => v.id)) + 1; + const version = {id, parent: ledger.head, source, request: job.request.text, createdAt: new Date().toISOString(), layers: 0}; + const saved = await store.save(job.owner, thread._id, thread.revision, {...ledger, head: id, versions: [...ledger.versions, version]}); + if (!saved) throw Object.assign(Error('The piece moved on while the turn ran'), {code: 'moved'}); + return {versionID: id, revision: saved.revision, sourceHash: sourceHash(source), notes, acceptance: 'unreviewed'}; +} + +async function work(queue, store, job) { + let beat = setInterval(() => queue.heartbeat(job._id, WORKER).catch(() => {}), HEARTBEAT_MS); + const started = Date.now(); + try { + const thread = await store.read(job.owner, job.code); + if (!thread) throw Object.assign(Error('Thread unavailable'), {code: 'moved'}); + const turn = await runTurn({job, thread, onCheckpoint: source => queue.heartbeat(job._id, WORKER, source)}); + const fresh = await store.read(job.owner, job.code); + const result = await commitVersion(store, job, fresh, turn.source, turn.findings.map(f => f.code)); + await queue.complete(job._id, WORKER, {...result, repairs: turn.repairs, elapsedMs: Date.now() - started}); + log('done', job.code, 'v' + result.versionID, `${Date.now() - started} ms`, turn.repairs ? 'after a repair' : ''); + } catch (error) { + const draft = job.checkpoint || null; + await queue.fail(job._id, WORKER, error.message, {draft}); + log('failed', job.code, error.message); + } finally { clearInterval(beat); } +} + +export async function main() { + for (const name of ['MONGODB_CONNECTION_STRING', 'MONGODB_NAME', 'WHISTLEGRAPH_WORKER_SECRET']) if (!process.env[name]) { console.error(`${name} is not set`); process.exit(2); } + const client = new MongoClient(process.env.MONGODB_CONNECTION_STRING, {serverSelectionTimeoutMS: 15000}); + await client.connect(); + const db = client.db(process.env.MONGODB_NAME); + const queue = mongoTurnQueue(db.collection('walkieware-turns')); + const store = mongoWhistlegraphStore(db.collection('walkieware-threads')); + let running = 0, stopping = false; + const stop = () => { stopping = true; log('stopping after current turns'); }; + process.on('SIGTERM', stop); process.on('SIGINT', stop); + log('ready', `concurrency ${CONCURRENCY}`, SITE); + while (!stopping) { + if (running >= CONCURRENCY) { await new Promise(r => setTimeout(r, 500)); continue; } + let job = null; + try { job = await queue.claim(WORKER); } catch (error) { log('claim failed', error.message); } + if (!job) { await new Promise(r => setTimeout(r, 2000)); continue; } + running++; log('claimed', job.code, job._id, `attempt ${job.attempts}`); + work(queue, store, job).finally(() => running--); + } + while (running > 0) await new Promise(r => setTimeout(r, 500)); + await client.close(); +} +if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) main().catch(error => { console.error(error); process.exit(1); }); diff --git a/lith/whistlegraph-worker.test.mjs b/lith/whistlegraph-worker.test.mjs new file mode 100644 index 0000000000..6772ba8a6b --- /dev/null +++ b/lith/whistlegraph-worker.test.mjs @@ -0,0 +1,35 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {mongoWhistlegraphStore, sourceHash} from '../system/backend/whistlegraph.mjs'; +import {commitVersion, runTurn} from './whistlegraph-worker.mjs'; + +function collection() { + const rows = []; + const matches = (row, query) => Object.entries(query).every(([k, v]) => row[k] === v); + return { + rows, createIndex: async () => {}, + insertOne: async row => { rows.push({...row}); }, + findOne: async q => { const r = rows.find(r => matches(r, q)); return r ? structuredClone(r) : null; }, + updateOne: async (q, update) => { const row = rows.find(r => matches(r, q)); if (!row) return {modifiedCount: 0}; Object.assign(row, update.$set || {}); for (const [k, n] of Object.entries(update.$inc || {})) row[k] = (row[k] || 0) + n; return {modifiedCount: 1}; }, + }; +} +const ledger = (head, source) => ({format: 1, head, versions: Array.from({length: head + 1}, (_, id) => ({id, parent: id ? id - 1 : null, source: id === head ? source : 'export function paint({wipe}){wipe(' + id + ');}', request: id ? 'req ' + id : null, createdAt: '2026-10-10T00:00:0' + id + 'Z', layers: 0}))}); + +test('a worker commit appends the next version and loses the race if the phone moved first', async () => { + const c = collection(), store = mongoWhistlegraphStore(c, {name: () => 'wgWorka'}); + const row = await store.open('alice', '11111111-1111-4111-8111-111111111111'); + const base = 'export function paint({wipe}){wipe("blue");}'; + await store.save('alice', row._id, 0, ledger(2, base)); + const thread = await store.read('alice', 'wgWorka'); + const job = {owner: 'alice', code: 'wgWorka', request: {text: 'make it red'}, baseVersion: 2, baseHash: sourceHash(base)}; + const result = await commitVersion(store, job, thread, 'export function paint({wipe}){wipe("red");}', ['api-x']); + assert.equal(result.versionID, 3); assert.equal(result.acceptance, 'unreviewed'); assert.deepEqual(result.notes, ['api-x']); + const after = await store.read('alice', 'wgWorka'); + assert.equal(after.ledger.head, 3); assert.equal(after.ledger.versions[3].request, 'make it red'); assert.equal(after.ledger.versions[3].parent, 2); + await assert.rejects(commitVersion(store, job, thread, 'export function paint({wipe}){wipe("green");}', []), {code: 'moved'}, 'a stale revision cannot fork the thread'); +}); + +test('a turn refuses to run on a piece that moved past its base', async () => { + const thread = {_id: 't', revision: 3, ledger: ledger(2, 'export function paint({wipe}){wipe(1);}')}; + await assert.rejects(runTurn({job: {baseVersion: 1, baseHash: 'x', request: {text: 'x'}}, thread}), {code: 'moved'}); +}); diff --git a/system/backend/whistlegraph-turns.mjs b/system/backend/whistlegraph-turns.mjs index 1362b79458..073c7d8a0a 100644 --- a/system/backend/whistlegraph-turns.mjs +++ b/system/backend/whistlegraph-turns.mjs @@ -104,6 +104,11 @@ export function mongoTurnQueue(collection, {now = () => new Date()} = {}) { return r.modifiedCount === 1; }, async read(owner, id) { return collection.findOne({_id: id, owner}); }, + // A waiting turn can be taken back; a running one finishes (its lease holder decides). + async cancel(owner, id) { + const r = await collection.updateOne({_id: id, owner, status: 'queued'}, {$set: {status: 'failed', finishedAt: now().toISOString(), result: {error: 'Cancelled', draft: null}}}); + return r.modifiedCount === 1; + }, async listOpen(owner, threadID) { return collection.find({owner, threadID, status: {$in: ['queued','running']}}).sort({createdAt: 1}).toArray(); }, async recent(owner, limit = 20) { return collection.find({owner}).sort({createdAt: -1}).limit(Math.min(100, Math.max(1, limit))).toArray(); }, // Queue depth is the autoscale signal. diff --git a/system/backend/whistlegraph-turns.test.mjs b/system/backend/whistlegraph-turns.test.mjs index 67f95cfa43..138bf15fa3 100644 --- a/system/backend/whistlegraph-turns.test.mjs +++ b/system/backend/whistlegraph-turns.test.mjs @@ -83,4 +83,8 @@ test('the same request is one row, and a person cannot stack more than the cap', await assert.rejects(q.enqueue('bob', thread('t2', 'wgBobbo'), validateTurnRequest({...good, code: 'wgBobbo', requestID: id})), {statusCode: 404}); for (let i = 1; i < MAX_QUEUED_PER_OWNER; i++) await q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, text: 'more ' + i})); await assert.rejects(q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, text: 'one too many'})), {statusCode: 429}); + assert.equal(await q.cancel('bob', first._id), false, 'only the owner cancels'); + assert.ok(await q.cancel('alice', first._id)); + assert.equal(await q.cancel('alice', first._id), false, 'a cancelled turn is not waiting'); + await q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, text: 'room again'})); }); diff --git a/system/netlify/functions/whistlegraph-turn.mjs b/system/netlify/functions/whistlegraph-turn.mjs index 8aa8fce70c..061fcb86e6 100644 --- a/system/netlify/functions/whistlegraph-turn.mjs +++ b/system/netlify/functions/whistlegraph-turn.mjs @@ -2,6 +2,7 @@ // POST {code, text, displayText?, drawing?, baseVersion, baseHash, model?, requestID?} → the queued turn // GET ?id= → one turn GET ?code= → that thread's open turns // GET ?recent=1 → the owner's last turns +// DELETE ?id= → take back a turn that has not started // Workers claim and finish turns through the queue module directly, never this route. import {connect} from '../../backend/database.mjs'; import {authenticateMusical} from './easel-musical-jev.mjs'; @@ -10,16 +11,21 @@ import {mongoTurnQueue, validateTurnRequest, publicTurn} from '../../backend/whi let pending; export const turnQueue = () => pending ??= (async () => { const {db} = await connect(); return mongoTurnQueue(db.collection('walkieware-turns')); })().catch(error => { pending = null; throw error; }); export async function handler(event) { - const headers = {'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Headers': 'Authorization, Content-Type', 'Access-Control-Allow-Methods': 'GET, POST, OPTIONS'}; + const headers = {'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Headers': 'Authorization, Content-Type', 'Access-Control-Allow-Methods': 'GET, POST, DELETE, OPTIONS'}; const reply = (statusCode, value) => ({statusCode, headers, body: JSON.stringify(value)}); if (event.httpMethod === 'OPTIONS') return reply(204, {}); - if (!['GET', 'POST'].includes(event.httpMethod)) return reply(405, {error: 'Use GET or POST'}); + if (!['GET', 'POST', 'DELETE'].includes(event.httpMethod)) return reply(405, {error: 'Use GET, POST or DELETE'}); let owner; try { owner = await authenticateMusical(event.headers || {}); } catch { return reply(503, {error: 'Authentication unavailable'}); } if (!owner) return reply(401, {error: 'Sign in to your AC account'}); let queue, store; try { [queue, store] = await Promise.all([turnQueue(), whistlegraphStore()]); } catch { return reply(503, {error: 'Turn queue unavailable'}); } try { + if (event.httpMethod === 'DELETE') { + const id = event.queryStringParameters?.id; + if (!id) return reply(400, {error: 'Specify the turn id'}); + return (await queue.cancel(owner, id)) ? reply(200, {cancelled: true}) : reply(409, {error: 'That turn is not waiting'}); + } if (event.httpMethod === 'GET') { const q = event.queryStringParameters || {}; if (q.id) { const row = await queue.read(owner, q.id); return row ? reply(200, publicTurn(row)) : reply(404, {error: 'No such turn'}); } -- 2.51.2