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