146 lines
10 KiB
TypeScript
146 lines
10 KiB
TypeScript
/** A durable FIFO per source. A recorded landing always resumes at push/report, never at merge. */
|
|
import { BACKGROUND_CONTEXT } from '@earendil-works/chord/context'
|
|
import { configure, defineTask, ROOT_CONVERSATION_ID as ROOT } from '@earendil-works/pi-durable'
|
|
import { CheckRan, SubmissionLanded, Workspaces } from './api.ts'
|
|
import { MessagesApi } from '../messages/api.ts'
|
|
import { starterOf } from '../messages/addresses.ts'
|
|
import { chain, git, hash, mkdir, registered } from './tree.ts'
|
|
export function lander(host:any,workspaces:any) {
|
|
const task=defineTask<any,any,null>({name:'workspaces.lander',version:1,initial:()=>({phase:'next'}),phases:{
|
|
next:(_task:any,runtime:any,ctx:any)=>runtime.commit(async(tx:any)=>{
|
|
const index=await tx.doc(Workspaces,ROOT)
|
|
const entry=index.queue.find((entry:any)=>entry.source===_task.input.source)
|
|
if(!entry){delete index.landers[_task.input.source];return {status:'terminal',outcome:{status:'completed',result:null}}}
|
|
return {status:'running',checkpoint:{phase:'land',id:entry.id}}
|
|
},ctx),
|
|
land:async(task:any,runtime:any,ctx:any)=>{
|
|
const id=task.state.checkpoint.id
|
|
const entry={...(await runtime.snapshot(Workspaces,ROOT,ctx)).queue.find((entry:any)=>entry.id===id)}
|
|
const identity={task:task.id,conversations:entry.conversations,signal:runtime.signal}
|
|
const path=`/worktrees/lander-${hash(entry.source)}`
|
|
const run=(cwd:string,...args:string[])=>git(host,cwd,args,{...identity,timeout:['fetch','push'].includes(args[0])?300:30})
|
|
const save=async(update:any)=>runtime.commit(async(tx:any)=>{Object.assign((await tx.doc(Workspaces,ROOT)).queue.find((e:any)=>e.id===id),update);return {status:'running',checkpoint:{phase:'land',id}}},ctx)
|
|
let failure:string|undefined
|
|
let pushed:string|undefined
|
|
try {
|
|
if(!entry.landed)await workspaces.serial(entry.source,async()=>{
|
|
if(await run(entry.source,'symbolic-ref','--short','HEAD')!==entry.target)throw new Error('Target branch changed')
|
|
if(await run(entry.source,'status','--porcelain'))throw new Error('Target worktree is dirty')
|
|
const base=await run(entry.source,'rev-parse','HEAD')
|
|
if(!entry.prepared) {
|
|
await mkdir(host,'/worktrees')
|
|
if(!await registered(host,{source:entry.source,path}))await run(entry.source,'worktree','add','--detach',path,base)
|
|
await run(path,'reset','--hard',base)
|
|
const remote=host.settings().lander?.push
|
|
if(remote)await run(path,'fetch',remote,entry.target)
|
|
if(remote)await run(path,'merge','--no-edit','FETCH_HEAD')
|
|
await run(path,'merge','--no-ff','--no-edit','-m',`Merge from #${entry.conversationId} into ${entry.target}\n\nSausage-Conversation: ${entry.conversationId}`,entry.tip)
|
|
const sha=await run(path,'rev-parse','HEAD')
|
|
for(const configured of host.settings().lander?.checks??[]) {
|
|
const check=typeof configured==='string'?{command:configured}:configured
|
|
const start=Date.now()
|
|
let result:any
|
|
let error:any
|
|
try {result=await host.exec(check.command,{cwd:path,timeout:check.timeout??600,...identity})}catch(cause){error=cause}
|
|
const passed=!error&&result.code===0
|
|
await runtime.commit(async(tx:any)=>{
|
|
await host.emit(tx,CheckRan,{submission:id,command:check.command,result:passed?'passed':error?'timeout':'failed',ms:Date.now()-start})
|
|
return {status:'running',checkpoint:{phase:'land',id}}
|
|
},ctx)
|
|
if(!passed)throw new Error(`Check ${check.command} failed: ${String(error??(result.stderr||result.stdout)).slice(-4000)}`)
|
|
}
|
|
if(await run(path,'status','--porcelain')||await run(path,'rev-parse','HEAD')!==sha)throw new Error('Check changed the lander checkout')
|
|
entry.prepared={base,sha}
|
|
await save({prepared:entry.prepared})
|
|
}
|
|
const {sha,base:previous}=entry.prepared
|
|
// Recover a fast-forward before its document commit, even when another commit followed it.
|
|
const current=await run(entry.source,'rev-parse','HEAD')
|
|
if(current===previous)await run(entry.source,'merge','--ff-only',sha)
|
|
else if(current!==sha)await run(entry.source,'merge-base','--is-ancestor',sha,'HEAD')
|
|
const commits=(await run(path,'log','--format=%H %s',`${previous}..${sha}`)).split('\n').filter(Boolean).map(line=>({sha:line.slice(0,40),subject:line.slice(41)}))
|
|
const paths=(await run(path,'diff','--name-only',previous,sha)).split('\n').filter(Boolean)
|
|
entry.landed={base:previous,sha,commits,paths}
|
|
await save({landed:entry.landed})
|
|
})
|
|
const remote=host.settings().lander?.push
|
|
if(remote) {
|
|
// World git accepts HEAD, not raw object IDs. Pin the detached lander checkout, never the moving source branch.
|
|
await run(path,'checkout','--detach',entry.landed.sha)
|
|
await run(path,'push',remote,`HEAD:refs/heads/${entry.target}`)
|
|
pushed=`${remote}/${entry.target}`
|
|
}
|
|
}catch(error){if(runtime.signal.aborted)throw error;failure=String(error)}
|
|
const outcome=entry.landed?{outcome:'landed',...entry.landed,...(pushed?{pushed}:{}),...(failure?{reason:`Push failed: ${failure}`}:{})}:{outcome:'failed',reason:failure}
|
|
const text=entry.landed?`[lander] Submission ${id} landed as ${entry.landed.sha} on ${entry.target}. ${pushed?`Pushed to ${pushed}.`:failure?`Push failed: ${failure}`:'Not pushed.'}`:`[lander] Submission ${id} was not landed: ${failure}`
|
|
await runtime.commit(async(tx:any)=>{
|
|
const index=await tx.doc(Workspaces,ROOT)
|
|
const recorded=index.results[id]
|
|
index.results[id]=outcome
|
|
if(entry.landed){
|
|
index.items[entry.conversationId].landed=entry.landed.sha
|
|
// Lineage records the landing in the root tally and distills its memory.
|
|
if(!recorded)await host.emit(tx,SubmissionLanded,{submission:id,conversationId:entry.conversationId,at:entry.at,sha:entry.landed.sha,base:entry.landed.base,target:entry.target,lander:runtime.conversationId,divergence:entry.divergence,title:entry.title,commits:entry.landed.commits.slice(0,10),commitCount:entry.landed.commits.length,paths:entry.landed.paths.slice(0,8),...(pushed?{pushed}:{})})
|
|
}
|
|
return {status:'running',checkpoint:{phase:'report',id,conversationId:entry.conversationId,text,outcome}}
|
|
},ctx)
|
|
},
|
|
report:async(task:any,runtime:any,ctx:any)=>{
|
|
const {id,conversationId,text,outcome}=task.state.checkpoint
|
|
const conversation=await runtime.conversation(conversationId,ctx)
|
|
const entry={kind:'workspaces.lander-report',data:{submission:id,...outcome}}
|
|
if(outcome.outcome==='failed')
|
|
await conversation.submit({type:'input',content:text,entry,whenBusy:'followUp',requestId:`lander:${id}`},ctx)
|
|
else {
|
|
await conversation.submit({type:'write',entry:{...entry,model:[{role:'user',content:text,timestamp:Date.now()}]},requestId:`lander:${id}`},ctx)
|
|
const starter=await starterOf(host,conversationId,ctx)
|
|
if(starter!==undefined&&starter!==conversationId)
|
|
await host.need(MessagesApi).send(conversationId,{to:`#${starter}`,text,reply:false},`lander:${id}`,ctx)
|
|
}
|
|
await runtime.commit(async(tx:any)=>{
|
|
const index=await tx.doc(Workspaces,ROOT)
|
|
index.queue=index.queue.filter((entry:any)=>entry.id!==id)
|
|
return {status:'running',checkpoint:{phase:'next'}}
|
|
},ctx)
|
|
},
|
|
},abort:(_task:any,runtime:any,ctx:any)=>runtime.commit(async(tx:any)=>{delete (await tx.doc(Workspaces,ROOT)).landers[_task.input.source];return {status:'terminal',outcome:{status:'aborted'}}},ctx)})
|
|
const ensure=async(tx:any,index:any,source:string,known?:any)=>{
|
|
const existing=known===undefined?(index.landers[source]&&await tx.task(index.landers[source])):known
|
|
if(existing&&existing.state.status!=='terminal')return
|
|
if(!index.landerConversations[source]){
|
|
const conversation=await tx.createConversation({ownership:{kind:'ownerless'}})
|
|
index.landerConversations[source]=conversation.id
|
|
await configure(tx,conversation.id,{cwd:`/worktrees/lander-${hash(source)}`})
|
|
Object.assign(await tx.doc(host.metadata,conversation.id),{title:'@lander',label:'lander'})
|
|
}
|
|
index.landers[source]=await tx.createTask(task,{source},{conversationId:index.landerConversations[source],ownership:{kind:'conversation'},background:true})
|
|
}
|
|
host.onOpen(()=>host.harness().commit(async(tx:any)=>{
|
|
const index=await tx.doc(Workspaces,ROOT)
|
|
const sources=[...new Set<string>(index.queue.map((entry:any)=>entry.source))]
|
|
const tasks=await Promise.all(sources.map(source=>index.landers[source]?tx.task(index.landers[source]):null))
|
|
for(let i=0;i<sources.length;i++)await ensure(tx,index,sources[i],tasks[i]??null)
|
|
},BACKGROUND_CONTEXT))
|
|
return {task,async submit(id:number,divergence:string,ctx:any,requestId?:string){
|
|
const unavailable=await workspaces.unavailable(id,ctx)
|
|
if(unavailable)throw new Error(unavailable)
|
|
const workspace=await workspaces.get(id,ctx)
|
|
if(!workspace)throw new Error('This conversation has no workspace')
|
|
const conversations=await chain(host,id,ctx)
|
|
const identity={task:`submit-${id}`,conversations}
|
|
if(await git(host,workspace.path,['status','--porcelain'],identity))throw new Error('Commit work before submitting')
|
|
const tip=await git(host,workspace.path,['rev-parse','HEAD'],identity)
|
|
return host.harness().commit(async(tx:any)=>{
|
|
const index=await tx.doc(Workspaces,ROOT)
|
|
if(requestId&&index.results[requestId])return {id:requestId,position:0}
|
|
let entry=index.queue.find((entry:any)=>entry.conversationId===id)
|
|
if(!entry){
|
|
entry={id:requestId??globalThis.crypto.randomUUID(),conversationId:id,source:workspace.source,target:workspace.target,tip,conversations,divergence,title:(await tx.doc(host.metadata,id)).title??'',at:await tx.latestEntryId(id)??0}
|
|
index.queue.push(entry)
|
|
}
|
|
await ensure(tx,index,workspace.source)
|
|
return {id:entry.id,position:index.queue.filter((item:any)=>item.source===workspace.source).findIndex((item:any)=>item.id===entry.id)+1}
|
|
},ctx)
|
|
}}
|
|
}
|