diff --git a/aesel/src/whistlegraph-thread.mjs b/aesel/src/whistlegraph-thread.mjs index a8853a6974..6d3d78c318 100644 --- a/aesel/src/whistlegraph-thread.mjs +++ b/aesel/src/whistlegraph-thread.mjs @@ -13,9 +13,9 @@ export function threadIdentity(storage,key,uuid=()=>crypto.randomUUID()) { const value={id:uuid(),code:null};storage.setItem(key+'-thread',JSON.stringify(value));return value; } export class WhistlegraphThread { - constructor({storage,key,token,ledger,state,onStatus,onCommand,receipts=null,WebSocketImpl=globalThis.WebSocket,url='wss://aesthetic.computer/api/whistlegraph-stream',heartbeatMs=15000,maxIdleMs=45000,reconnectMs=3000}) { + constructor({storage,key,token,ledger,state,onStatus,onCommand,onAdopt=null,receipts=null,WebSocketImpl=globalThis.WebSocket,url='wss://aesthetic.computer/api/whistlegraph-stream',heartbeatMs=15000,maxIdleMs=45000,reconnectMs=3000}) { this.receipts=receipts; - Object.assign(this,{storage,key,token,ledger,state,onStatus,onCommand,WebSocketImpl,url,heartbeatMs,maxIdleMs,reconnectMs}); + Object.assign(this,{storage,key,token,ledger,state,onStatus,onCommand,onAdopt,WebSocketImpl,url,heartbeatMs,maxIdleMs,reconnectMs}); this.identity=threadIdentity(storage,key);this.revision=Number(storage.getItem(key+'-cloud-revision')||0);this.last=storage.getItem(key+'-cloud-ledger')||''; this.active=false;this.sending=false;this.ready=false;this.connectURL=url; } @@ -34,12 +34,25 @@ export class WhistlegraphThread { this.receiptSupport=m.capabilities?.includes('attempt-receipts-v1')===true; this.identity.code=m.thread.code;this.storage.setItem(this.key+'-thread',JSON.stringify(this.identity)); const cloud=ledgerText(m.thread.ledger),local=ledgerText(this.ledger()); - // Never silently replace local work with another device's history. - if(m.thread.ledger&&cloud!==local&&m.thread.revision!==this.revision){this.onStatus(this.identity.code,'History conflict');return;} + // Never silently replace local work with another device's history. A + // device with nothing unsynced (its local ledger is the last cloud + // ledger it saw) follows the server forward: that is a turn that ran + // off the phone, not another device's history. + if(m.thread.ledger&&cloud!==local&&m.thread.revision!==this.revision){ + if(this.canFollow(local,m.thread)){await this.follow(m.thread);} + else {this.onStatus(this.identity.code,'History conflict');return;} + } this.revision=m.thread.revision;this.ready=true; if(cloud===local){this.last=local;this.storage.setItem(this.key+'-cloud-revision',String(this.revision));this.storage.setItem(this.key+'-cloud-ledger',local);} this.onStatus(this.identity.code,'Connected');this.sync();this.update();this.flushReceipts(); } + if(m.type==='updated'&&m.thread?.ledger) { + if(this.sending)return; + const cloud=ledgerText(m.thread.ledger),local=ledgerText(this.ledger()); + if(cloud===local){this.revision=m.thread.revision;this.last=cloud;this.storage.setItem(this.key+'-cloud-revision',String(this.revision));this.storage.setItem(this.key+'-cloud-ledger',cloud);return;} + if(this.canFollow(local,m.thread)){await this.follow(m.thread);this.update();} + else this.onStatus(this.identity.code,'History conflict'); + } if(m.type==='saved') { this.revision=m.thread.revision;this.last=ledgerText(m.thread.ledger);this.sending=false; this.storage.setItem(this.key+'-cloud-revision',String(this.revision));this.storage.setItem(this.key+'-cloud-ledger',this.last);this.sync(); @@ -68,6 +81,13 @@ export class WhistlegraphThread { ws.onerror=()=>{}; } send(value){if(this.ws?.readyState===1)this.ws.send(JSON.stringify(value));} + canFollow(local,thread){return !!this.onAdopt&&local===this.last&&Number.isSafeInteger(thread.revision)&&thread.revision>this.revision;} + async follow(thread){ + this.revision=thread.revision;this.last=ledgerText(thread.ledger); + this.storage.setItem(this.key+'-cloud-revision',String(this.revision));this.storage.setItem(this.key+'-cloud-ledger',this.last); + await this.onAdopt(thread.ledger,thread); + this.onStatus(this.identity.code,'Updated'); + } sync(){if(!this.ready||this.sending)return;const ledger=this.ledger(),next=ledgerText(ledger);if(next===this.last)return;this.sending=true;this.send({type:'sync',revision:this.revision,ledger});} update(){if(this.ready)this.send({type:'state',state:this.state()});} flushReceipts(){ diff --git a/apple/whistlegraph/Resources/Web/engine.mjs b/apple/whistlegraph/Resources/Web/engine.mjs index 5981dc9cf8..5f930033d6 100644 --- a/apple/whistlegraph/Resources/Web/engine.mjs +++ b/apple/whistlegraph/Resources/Web/engine.mjs @@ -357,7 +357,20 @@ function compileStream() { } function saved() { try { localStorage.setItem(storageKey,source); } catch {} updateFeed(); } function review(show) { $('speak').disabled=busy; $('speak-label').textContent=busy?'Working…':'Hold to talk'; } -function end() { clearTimeout(compileTimer);compileTimer=null; if(provisional && previewSource!==source){render(source||'export function paint({wipe}) {wipe("black");}');provisional='';} busy=false;activeModel='';void refreshBraincells(); updateFeed(); clearInterval(timer); $('live-stop').hidden=true; review(source!==previous); window.whistlegraphWorkFinished?.(); } +function end() { clearTimeout(compileTimer);compileTimer=null; if(provisional && previewSource!==source){render(source||'export function paint({wipe}) {wipe("black");}');provisional='';} busy=false;activeModel='';void refreshBraincells(); updateFeed(); clearInterval(timer); $('live-stop').hidden=true; review(source!==previous); window.whistlegraphWorkFinished?.(); if(pendingAdopt){const next=pendingAdopt;pendingAdopt=null;adoptLedger(next);} } +let pendingAdopt=null; +function adoptLedger(ledger){ + try{ + if(!versions||ledger?.format!==1||!Array.isArray(ledger.versions))return; + versions.persist(ledger); + source=previous=versions.head.source;vfs.mount(file,source);saved(); + render(source||'export function paint({wipe}) {wipe("black");}'); + lastAttempt={request:versions.head.request||'',status:'completed',error:'',startedAt:versions.head.createdAt||''}; + localStorage.removeItem(storageKey+'-inflight');localStorage.removeItem(storageKey+'-attempt');localStorage.removeItem(storageKey+'-draft'); + phase(`v${versions.head.id} · Made while you were away`);log('Adopted v'+versions.head.id+' from the knot'); + updateFeed();threadUpdate();nativeSnapshot(); + }catch(error){log('Could not adopt the server version: '+error.message);} +} function delta(text) { outputStream=(outputStream+text).slice(-6000); if (!firstDelta) {firstDelta=true; log('First model output');benchmark('firstModelOutput');} @@ -758,6 +771,9 @@ if(versions&&!window.__whistlegraphSequence&&!window.__whistlegraphBenchmark&&!w thread=new WhistlegraphThread({storage:localStorage,key:storageKey,receipts,token:()=>token,ledger:()=>versions.value, state:()=>({busy,phase:$('live-phase').textContent,head:versions.head.id,source:versions.head.source,errors:runtimeErrors,attempt:lastAttempt}), onStatus:(code,status)=>{label.textContent=code?'/'+code:'';label.title=status;label.dataset.status=status;post({action:'threadStatus',code:code||'',threadID:thread.identity.id,status});nativeSnapshot();}, + // A turn that ran off the phone: take the server's ledger as our own. + // While a turn runs here, it waits until that turn ends. + onAdopt:async ledger=>{if(busy){pendingAdopt=ledger;return;}adoptLedger(ledger);}, onCommand:async command=>{ if(command.action==='layout'){ if(typeof command.css!=='string'||new TextEncoder().encode(command.css).length>100000)throw Error('Invalid layout'); diff --git a/lith/whistlegraph-socket.mjs b/lith/whistlegraph-socket.mjs index b34f604bd3..aa7601c1cc 100644 --- a/lith/whistlegraph-socket.mjs +++ b/lith/whistlegraph-socket.mjs @@ -2,6 +2,7 @@ import {WebSocketServer,WebSocket} from 'ws'; import {attachSocketRoute} from './socket-routing.mjs'; import {randomUUID} from 'node:crypto'; import {publicThread,sourceHash} from '../system/backend/whistlegraph.mjs'; +import {onThreadUpdated} from '../system/backend/whistlegraph-live.mjs'; export function attachWhistlegraphSocket(server,{authenticate,store,authMs=5000,lifetimeMs=600000}={}) { const wss=new WebSocketServer({noServer:true,maxPayload:8_100_000,perMessageDeflate:false}); @@ -13,6 +14,11 @@ export function attachWhistlegraphSocket(server,{authenticate,store,authMs=5000, }; // Older installed clients retain their authenticated connection route. const detach=attachSocketRoute(server,['/api/whistlegraph-stream','/api/walkieware-stream'],upgrade); + // A turn that finished on the server (a worker's commit) reaches every + // client in the thread's room as 'updated'; a device with nothing unsynced + // follows it, one with local work reports a conflict as it always has. + const notify=(threadID,payload)=>{const room=rooms.get(threadID);let n=0;for(const client of room?.clients||[]){send(client,{type:'updated',...payload});n++;}return n;}; + const unsubscribe=onThreadUpdated(notify); wss.on('connection',ws=>{ let owner,row,room,role,closed=false,chain=Promise.resolve(),pending=0,alive=true,messages=0,windowAt=Date.now(); const authTimer=setTimeout(()=>ws.close(1008,'Authenticate first'),authMs); @@ -89,5 +95,5 @@ export function attachWhistlegraphSocket(server,{authenticate,store,authMs=5000, if(room&&!room.clients.size)rooms.delete(row._id); }); }); - return {wss,close(){detach();for(const ws of wss.clients)ws.terminate();wss.close();}}; + return {wss,notify,close(){unsubscribe();detach();for(const ws of wss.clients)ws.terminate();wss.close();}}; } diff --git a/lith/whistlegraph-socket.test.mjs b/lith/whistlegraph-socket.test.mjs index b44fe60892..d29da40d24 100644 --- a/lith/whistlegraph-socket.test.mjs +++ b/lith/whistlegraph-socket.test.mjs @@ -159,3 +159,28 @@ test('receipt client defers old servers, retries after lost acknowledgement, and assert.equal(receipts.pending(),null);assert.equal(phone.ready,true); }finally{phone.suspend();} }); + +test('a device with nothing unsynced follows a server-side turn; one with local work still reports a conflict',async()=>{ + const values=new Map(),storage={getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}; + const base={format:1,head:1,versions:[ledger.versions[0],{id:1,parent:0,source:'export function paint({wipe}){wipe(1);}',request:'one',createdAt:'t1',layers:0}]}; + const ahead={...base,head:2,versions:[...base.versions,{id:2,parent:1,source:'export function paint({wipe}){wipe(2);}',request:'two (worker)',createdAt:'t2',layers:0}]}; + let local=base,adopted=null,statuses=[];let sockets=[]; + class FakeSocket{constructor(){sockets.push(this);this.readyState=1;queueMicrotask(()=>this.onopen?.());}send(raw){const m=JSON.parse(raw);if(m.type==='authenticate')queueMicrotask(()=>this.onmessage?.({data:JSON.stringify({type:'ready',thread:{code:'wgRemot',revision:5,ledger:ahead}})}));}close(){this.readyState=3;}} + storage.setItem('k-cloud-revision','4');storage.setItem('k-cloud-ledger',JSON.stringify(base)); + const phone=new WhistlegraphThread({storage,key:'k',token:()=>'t',ledger:()=>local,state:()=>({}),onStatus:(_,s)=>statuses.push(s),onCommand:async()=>({ok:true}),onAdopt:async l=>{adopted=l;local=l;},WebSocketImpl:FakeSocket}); + await phone.resume();await new Promise(r=>setTimeout(r,20)); + assert.equal(adopted?.head,2,'the ledger the server moved to is adopted on connect'); + assert.equal(phone.revision,5);assert.equal(storage.getItem('k-cloud-revision'),'5'); + assert.ok(statuses.includes('Updated')&&statuses.includes('Connected'),statuses.join(',')); + // Live: a later 'updated' while connected and clean follows too. + const later={...ahead,head:3,versions:[...ahead.versions,{id:3,parent:2,source:'export function paint({wipe}){wipe(3);}',request:'three (worker)',createdAt:'t3',layers:0}]}; + await sockets[0].onmessage({data:JSON.stringify({type:'updated',thread:{code:'wgRemot',revision:6,ledger:later},versionID:3,source:'worker'})}); + assert.equal(adopted.head,3);assert.equal(phone.revision,6); + // With local work the server has not seen, nothing is replaced. + local={...later,head:4,versions:[...later.versions,{id:4,parent:3,source:'export function paint({wipe}){wipe(4);}',request:'mine',createdAt:'t4',layers:0}]}; + const further={...later,head:5,versions:[...later.versions,{id:5,parent:3,source:'x',request:'theirs',createdAt:'t5',layers:0}]}; + statuses.length=0; + await sockets[0].onmessage({data:JSON.stringify({type:'updated',thread:{code:'wgRemot',revision:7,ledger:further},versionID:5,source:'worker'})}); + assert.equal(adopted.head,3,'local work is never replaced');assert.ok(statuses.includes('History conflict')); + phone.suspend(); +}); diff --git a/lith/whistlegraph-worker.mjs b/lith/whistlegraph-worker.mjs index 6038f728d4..fb9501021b 100644 --- a/lith/whistlegraph-worker.mjs +++ b/lith/whistlegraph-worker.mjs @@ -150,6 +150,12 @@ async function work(queue, store, job, pool) { const result = await commitVersion(store, job, fresh, turn.source, [...turn.findings.map(f => f.code), ...turn.notes], {acceptance: turn.acceptance, checks: turn.checks}); await queue.complete(job._id, WORKER, {...result, repairs: turn.repairs, elapsedMs: Date.now() - started}); log('done', job.code, 'v' + result.versionID, turn.acceptance, `${Date.now() - started} ms`, turn.repairs ? 'after a repair' : ''); + // Best effort: lith broadcasts the new revision to the thread's room and pushes the owner. + try { + const r = await fetch(`${SITE}/api/whistlegraph-turn-done`, {method: 'POST', headers: {'Content-Type': 'application/json', 'x-ac-worker': process.env.WHISTLEGRAPH_WORKER_SECRET}, + body: JSON.stringify({owner: job.owner, code: job.code, versionID: result.versionID, displayText: job.request.displayText || ''}), signal: AbortSignal.timeout(15000)}); + log('told lith', job.code, r.status, (await r.text()).slice(0, 120)); + } catch (error) { log('could not tell lith', job.code, error.message); } } catch (error) { const draft = error.draft || job.checkpoint || null; await queue.fail(job._id, WORKER, error.message, {draft}); diff --git a/system/backend/whistlegraph-live.mjs b/system/backend/whistlegraph-live.mjs new file mode 100644 index 0000000000..cb22871095 --- /dev/null +++ b/system/backend/whistlegraph-live.mjs @@ -0,0 +1,11 @@ +// The in-process line between a turn finishing on the server and the phone +// hearing about it (apple/whistlegraph/TURNS.md, slice 4). The socket module +// listens; the turn-done route speaks. Both live in the lith process, so a +// module-level set is the whole bus. +const listeners = new Set(); +export function onThreadUpdated(fn) { listeners.add(fn); return () => listeners.delete(fn); } +export function threadUpdated(threadID, payload) { + let delivered = 0; + for (const fn of listeners) { try { delivered += fn(threadID, payload) || 0; } catch {} } + return delivered; +} diff --git a/system/netlify/functions/whistlegraph-turn-done.mjs b/system/netlify/functions/whistlegraph-turn-done.mjs new file mode 100644 index 0000000000..8f8c88997a --- /dev/null +++ b/system/netlify/functions/whistlegraph-turn-done.mjs @@ -0,0 +1,31 @@ +// A turn worker finished a turn: tell the thread's room and push the owner. +// POST {owner, code, versionID, displayText} x-ac-worker: +// Workers only; the phone never calls this. +import {timingSafeEqual} from 'node:crypto'; +import {connect} from '../../backend/database.mjs'; +import {whistlegraphStore} from './whistlegraph.mjs'; +import {publicThread} from '../../backend/whistlegraph.mjs'; +import {threadUpdated} from '../../backend/whistlegraph-live.mjs'; +export async function handler(event) { + const headers={'Content-Type':'application/json','Cache-Control':'no-store'}; + const reply=(statusCode,value)=>({statusCode,headers,body:JSON.stringify(value)}); + if(event.httpMethod!=='POST')return reply(405,{error:'Use POST'}); + const secret=process.env.WHISTLEGRAPH_WORKER_SECRET||'',presented=String(event.headers?.['x-ac-worker']||''); + if(secret.length<32||presented.length!==secret.length||!timingSafeEqual(Buffer.from(presented),Buffer.from(secret)))return reply(401,{error:'Workers only'}); + let body;try{body=JSON.parse(event.body||'{}');}catch{return reply(400,{error:'Send JSON'});} + const {owner,code,versionID,displayText=''}=body; + if(typeof owner!=='string'||!owner||typeof code!=='string'||!Number.isSafeInteger(versionID))return reply(400,{error:'Specify owner, code and versionID'}); + try{ + const store=await whistlegraphStore(); + const row=await store.read(owner,code); + if(!row)return reply(404,{error:'Thread unavailable'}); + const listeners=threadUpdated(row._id,{thread:publicThread(row),versionID,source:'worker'}); + let push=null; + try{ + const {db}=await connect(); + const {sendToTarget}=await import('../../../shared/push.mjs'); + push=await sendToTarget(db,{user:owner,app:'whistlegraph'},{title:'Your piece is ready',body:(String(displayText).trim()||'/'+code).slice(0,120)+' · v'+versionID,data:{url:'/'+code},thread:'whistlegraph-'+code,collapse:'whistlegraph-'+code}); + }catch(error){push={error:error.message};} + return reply(200,{told:listeners,push}); + }catch(error){return reply(503,{error:'Could not tell the thread: '+error.message});} +}