Merge v3-t52 (T52)
# Conflicts: # host/js/package.json
This commit is contained in:
@@ -334,7 +334,7 @@ export default function subagents(host: any) {
|
||||
const spawnDefinition = async (args: any, definitions: any[]) => {
|
||||
const definition = definitions.find(d => d.name === args.agent)
|
||||
if (!definition) throw new Error(`Unknown agent: ${args.agent}. Available definitions: ${definitions.map(d => d.name).join(', ')}`)
|
||||
if ((args.isolate ?? definition.isolate ?? 'none') !== 'none') throw new Error('worktree isolation requires the workspaces extension (P3-k)')
|
||||
if ((args.isolate ?? definition.isolate ?? 'none') === 'worktree' && !host.want({name:'workspaces'})) throw new Error('worktree isolation requires the workspaces extension (P3-k)')
|
||||
return definition
|
||||
}
|
||||
const Reporter = createReporter()
|
||||
@@ -554,11 +554,11 @@ export default function subagents(host: any) {
|
||||
defineTool({
|
||||
name: 'subagent',
|
||||
description:
|
||||
'Delegate to a named definition. start returns an answer or a background child ID. message steers a busy child unless steer is false. wait returns its latest answer. Foreground waits yield on parent steering. stop ends current child work. list shows your children. Parent stop leaves children independent. fork inherits history before this call. Only isolate none is available until workspaces arrive. Background answers steer the parent exactly once.',
|
||||
'Delegate to a named definition. start returns an answer or a background child ID. message steers a busy child unless steer is false. wait returns its latest answer. Foreground waits yield on parent steering. stop ends current child work. list shows your children. Parent stop leaves children independent. fork inherits history before this call. Worktree isolation gives the child a checkout. land fast-forwards your checkout and releases the child checkout. discard removes a clean checkout and retains its branch. Background answers steer the parent exactly once.',
|
||||
// Call-id receipts deduplicate starts/messages; memoized targets and guards protect stop/workspace replays.
|
||||
replay: 'safe',
|
||||
parameters: obj({
|
||||
action: oneOf(['start', 'message', 'wait', 'stop', 'list']),
|
||||
action: oneOf(['start', 'message', 'wait', 'stop', 'list', 'land', 'discard']),
|
||||
agent: opt(
|
||||
str({
|
||||
description:
|
||||
@@ -603,6 +603,12 @@ export default function subagents(host: any) {
|
||||
(!args.id || !Object.hasOwn(index.agents, String(args.id)))
|
||||
)
|
||||
throw new Error('id must identify a child of this conversation')
|
||||
if (args.action === 'land' || args.action === 'discard') {
|
||||
const workspaces = host.need({name:'workspaces'})
|
||||
const result = await workspaces[args.action](args.id, context)
|
||||
if (args.action === 'land') await workspaces.release([args.id], context)
|
||||
return finish(result.text)
|
||||
}
|
||||
if (args.action === 'start' || args.action === 'message') {
|
||||
const text = args.action === 'start' ? args.task : args.text
|
||||
if (!text?.trim())
|
||||
@@ -632,6 +638,8 @@ export default function subagents(host: any) {
|
||||
if (forkAt === undefined) throw new Error('No parent history before this call')
|
||||
}
|
||||
|
||||
const workspaceChain = definition && (args.isolate ?? definition.isolate ?? 'none') === 'worktree'
|
||||
? await host.need({name:'workspaces'}).chain(api.conversationId, context) : []
|
||||
const action = await api.commit(async (tx) => {
|
||||
const state = await tx.doc(Subagents, api.conversationId)
|
||||
if (state.actions[api.callId]) return { ...state.actions[api.callId] }
|
||||
@@ -655,6 +663,8 @@ export default function subagents(host: any) {
|
||||
: { fork: { conversationId: api.conversationId, at: forkAt } }),
|
||||
cwd: parent.cwd ?? '/src/sausage',
|
||||
})
|
||||
if ((args.isolate ?? definition.isolate ?? 'none') === 'worktree')
|
||||
await host.need({name:'workspaces'}).create(tx, child.id, parent.cwd ?? '/src/sausage', workspaceChain)
|
||||
conversationId = child.id
|
||||
await started(host, tx, child, label!, 'child')
|
||||
const runs = await tx.doc(Runs, ROOT_CONVERSATION_ID)
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
# Workspaces
|
||||
|
||||
An isolated child uses `/worktrees/<conversation-id>`.
|
||||
The preparation task creates its Git checkout before the first tool.
|
||||
The tool gate checks the registration through the VFS. It does not spawn Git.
|
||||
Preparation and lander commands use world Git through `host.exec`.
|
||||
Each command carries its task tag and the conversation chain of its caller.
|
||||
|
||||
`subagent start` accepts `isolate: "worktree"`.
|
||||
A definition can select the same default.
|
||||
`subagent land` fast-forwards the parent checkout and releases the child checkout.
|
||||
`subagent discard` removes a clean checkout and retains its branch.
|
||||
The service refuses release when the branch retains unlanded commits.
|
||||
|
||||
`workspace submit` requires a clean checkout and a divergence description.
|
||||
Use `"None"` when the implementation matches the request.
|
||||
The lander processes one FIFO queue per source checkout.
|
||||
It merges the submitted commit in a detached checkout, runs checks, and fast-forwards the source.
|
||||
|
||||
A failed check returns an input to the submitter.
|
||||
A successful landing leaves a report without another model turn.
|
||||
It sends a one-way note to the branch starter through the messages API.
|
||||
Both reports use stable request IDs.
|
||||
The landing event lets lineage update the root tally and project memory.
|
||||
|
||||
The world settings file can contain this configuration:
|
||||
|
||||
```json
|
||||
{
|
||||
"lander": {
|
||||
"checks": ["sh -c 'test -f feature.txt'", {"command": "sh -c 'true'", "timeout": 30}],
|
||||
"push": "origin"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Checks run sequentially. The default timeout is 600 seconds.
|
||||
Omit `push` to land locally.
|
||||
Fetch, checks, and push use the submitter's conversation chain, not the lander conversation.
|
||||
A push failure does not erase a local landing.
|
||||
|
||||
The queue records the checked candidate before the fast-forward.
|
||||
It records `landed.sha` before push and report.
|
||||
After a restart, the lander pushes and reports that exact SHA without another merge.
|
||||
A later source commit does not change the SHA that the lander pushes.
|
||||
The report uses a submission-scoped request ID to prevent duplicate delivery.
|
||||
@@ -0,0 +1,12 @@
|
||||
/** Durable workspace records and the landing event consumed by coordination extensions. */
|
||||
import { defineDoc } from '@earendil-works/pi-durable'
|
||||
import { defineEvent } from '@earendil-works/sausage-events'
|
||||
export type Workspace = {source:string;target:string;path:string;branch:string;base?:string;state:'preparing'|'ready'|'failed'|'removed';preparation?:number;error?:string;landed?:string}
|
||||
export type Submission = {id:string;conversationId:number;source:string;target:string;tip:string;at:number;divergence:string;title:string;conversations:string[];prepared?:{base:string;sha:string};landed?:{base:string;sha:string;commits:{sha:string;subject:string}[];paths:string[]};outcome?:any}
|
||||
export const Workspaces = defineDoc<{items:Record<string,Workspace>;queue:Submission[];landers:Record<string,number>;landerConversations:Record<string,number>;results:Record<string,any>}>({
|
||||
kind:'ultimate.workspaces',version:1,scope:'conversation',history:'latest',fork:'initial',checkpointWhen:()=>false,
|
||||
initial:()=>({items:{},queue:[],landers:{},landerConversations:{},results:{}}),
|
||||
})
|
||||
export const WorkspacesApi = {name:'workspaces'}
|
||||
export const SubmissionLanded = defineEvent<any>('workspaces.landed')
|
||||
export const CheckRan = defineEvent<any>('workspaces.check')
|
||||
@@ -0,0 +1,145 @@
|
||||
/** 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}`,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)
|
||||
}}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
/** World-loaded worktree tools and the VFS-only gate before every guest tool. */
|
||||
import { defineExtension, defineTool, hook, ToolTask, GenerationTask, ROOT_CONVERSATION_ID, section } from '@earendil-works/pi-durable'
|
||||
import { obj, oneOf, opt, str } from '@earendil-works/pi-schema'
|
||||
import { Workspaces, WorkspacesApi } from './api.ts'
|
||||
import { workspaces } from './workspaces.ts'
|
||||
import { lander } from './lander.ts'
|
||||
export const order=-200
|
||||
export const docs=[Workspaces]
|
||||
export default function(host:any) {
|
||||
const service=workspaces(host), landing=lander(host,service)
|
||||
host.implement(WorkspacesApi,{...service,submit:landing.submit,doc:Workspaces})
|
||||
return defineExtension({name:'workspaces',tasks:[service.Preparation,landing.task],
|
||||
hooks:[hook(ToolTask,{beforeTool:async(_call:any,api:any,ctx:any)=>{
|
||||
const reason=await service.unavailable(api.conversationId,ctx)
|
||||
return reason?{block:reason}:undefined
|
||||
}}),hook(GenerationTask,{beforeInput:async(_input:any,api:any)=>{
|
||||
const index=await api.tx.doc(Workspaces,ROOT_CONVERSATION_ID)
|
||||
if(Object.values(index.landerConversations).includes(api.conversationId))throw new Error('The lander runs no model')
|
||||
}})],
|
||||
sections:[section('workspace',async(input:any,ctx:any)=>{
|
||||
await service.ready(input.conversationId,ctx)
|
||||
const workspace=await service.get(input.conversationId,ctx)
|
||||
return workspace?`Workspace: ${workspace.path} (${workspace.state}), branch ${workspace.branch}, target ${workspace.target}.`:''
|
||||
})],
|
||||
tools:[defineTool({name:'workspace',description:'Inspect your workspace or submit its clean tip to the lander. State any divergence from the request.',parameters:obj({action:oneOf(['status','submit']),divergence:opt(str())}),replay:'safe',execute:async(args:any,api:any,ctx:any)=>{
|
||||
if(args.action==='submit'&&!args.divergence?.trim())throw new Error('State divergence, or None')
|
||||
const result=args.action==='submit'?await landing.submit(api.conversationId,args.divergence,ctx,`workspace:${api.conversationId}:${api.callId}`):await service.get(api.conversationId,ctx)
|
||||
return {output:[{type:'text',text:JSON.stringify(result??{})}]}
|
||||
}})],
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
/** World Git execution, inherited caller identity, and VFS-only checkout registration checks. */
|
||||
import { Started } from '../messages/addresses.ts'
|
||||
export const resolve = (base:string,path:string) => {
|
||||
const parts:string[]=[]
|
||||
for(const part of (path.startsWith('/')?path:`${base}/${path}`).split('/')) {
|
||||
if(part==='..')parts.pop();else if(part&&part!=='.')parts.push(part)
|
||||
}
|
||||
return '/'+parts.join('/')
|
||||
}
|
||||
export async function chain(host:any,id:number,context:any):Promise<string[]> {
|
||||
const ids=new Set<number>()
|
||||
const visit=async(id:number)=>{
|
||||
if(ids.has(id))return
|
||||
ids.add(id)
|
||||
const row=await host.harness().commit((tx:any)=>tx.conversation(id),context)
|
||||
const up=row?.owner?.conversationId??row?.parent?.conversationId
|
||||
if(up!==undefined)await visit(up)
|
||||
const starter=(await host.harness().snapshot(Started,id,context))?.by
|
||||
if(starter!==undefined)await visit(starter)
|
||||
}
|
||||
await visit(id)
|
||||
return [...ids].map(String)
|
||||
}
|
||||
export const git = async(host:any,cwd:string,args:string[],identity:any={})=>{
|
||||
const result=await host.exec(['git',...args],{cwd,timeout:30,...identity})
|
||||
if(result.code)throw new Error(`${cwd}: git ${args.join(' ')}: ${result.stderr||result.stdout||`exited ${result.code}`}`)
|
||||
return result.stdout.trim()
|
||||
}
|
||||
export async function mkdir(host:any,path:string) {
|
||||
let current=''
|
||||
for(const part of path.split('/').filter(Boolean)) {
|
||||
current+='/'+part
|
||||
try {await host.fs.mkdir(current)} catch(error) {if((error as any).code!=='already_exists')throw error}
|
||||
}
|
||||
}
|
||||
export async function gitdir(host:any,source:string) {
|
||||
const path=`${source}/.git`
|
||||
return (await host.fs.stat(path)).kind==='directory'?path:resolve(source,(await host.fs.read(path)).trim().replace(/^gitdir: /,''))
|
||||
}
|
||||
export async function registered(host:any,workspace:{source:string;path:string}) {
|
||||
try {
|
||||
const admin=await gitdir(host,workspace.path)
|
||||
const source=await gitdir(host,workspace.source)
|
||||
const common=await host.fs.read(`${source}/commondir`).then((p:string)=>resolve(source,p.trim()),(e:any)=>{if(e.code==='not_found')return source;throw e})
|
||||
return admin.startsWith(`${common}/worktrees/`) &&
|
||||
resolve(admin,(await host.fs.read(`${admin}/gitdir`)).trim())===`${workspace.path}/.git` &&
|
||||
(await host.fs.stat(`${admin}/HEAD`)).kind==='file'
|
||||
} catch(error) {if((error as any).code==='not_found')return false;throw error}
|
||||
}
|
||||
export const hash=(text:string)=>{let hash=2166136261;for(const c of text)hash=Math.imul(hash^c.charCodeAt(0),16777619);return (hash>>>0).toString(16)}
|
||||
@@ -0,0 +1,137 @@
|
||||
/** Workspace preparation, safe release, and synchronous child-to-parent fast-forwards. */
|
||||
import { configure, defineTask, LiveDoc, InboxDoc, ROOT_CONVERSATION_ID as ROOT } from '@earendil-works/pi-durable'
|
||||
import { Workspaces } from './api.ts'
|
||||
import { chain, git, mkdir, registered, resolve } from './tree.ts'
|
||||
export function workspaces(host:any) {
|
||||
const queues=host.resource('source-queues',()=>new Map<string,Promise<any>>())
|
||||
const serial=async(source:string,operation:()=>Promise<any>)=>{
|
||||
const pending=(queues.get(source)??Promise.resolve()).catch(()=>{}).then(operation)
|
||||
queues.set(source,pending)
|
||||
try{return await pending}finally{if(queues.get(source)===pending)queues.delete(source)}
|
||||
}
|
||||
const get=async(id:number,ctx:any)=>(await host.harness().snapshot(Workspaces,ROOT,ctx))?.items[id]
|
||||
const Preparation=defineTask<any,any,null>({name:'workspaces.prepare',version:1,initial:()=>({phase:'prepare'}),phases:{
|
||||
prepare:async(task:any,runtime:any,ctx:any)=>{
|
||||
const workspace=await get(task.input.id,ctx)
|
||||
const identity={task:task.id,conversations:await chain(host,task.input.id,ctx),signal:runtime.signal}
|
||||
try {
|
||||
await serial(workspace.source,async()=>{
|
||||
await mkdir(host,'/worktrees')
|
||||
if(!await registered(host,workspace)) {
|
||||
// The preparation owns the branch name, but never adopts an unrelated directory.
|
||||
const exists=await host.fs.stat(workspace.path).then(()=>true,(e:any)=>{if(e.code==='not_found')return false;throw e})
|
||||
if(exists)throw new Error(`Workspace path already exists: ${workspace.path}`)
|
||||
const existing=await git(host,workspace.source,['branch','--list',workspace.branch],identity)
|
||||
if(existing)await git(host,workspace.source,['worktree','add',workspace.path,workspace.branch],identity)
|
||||
else await git(host,workspace.source,['worktree','add','-b',workspace.branch,workspace.path,workspace.base],identity)
|
||||
}
|
||||
})
|
||||
for(const command of host.settings().isolates?.worktree?.setup??[]) {
|
||||
const result=await host.exec(command,{cwd:workspace.path,timeout:300,env:[['SAUSAGE_SOURCE',workspace.source]],...identity})
|
||||
if(result.code)throw new Error(result.stderr||result.stdout||'Workspace setup failed')
|
||||
}
|
||||
await runtime.commit(async(tx:any)=>{
|
||||
const saved=(await tx.doc(Workspaces,ROOT)).items[task.input.id]
|
||||
saved.state='ready';delete saved.error
|
||||
return {status:'terminal',outcome:{status:'completed',result:null}}
|
||||
},ctx)
|
||||
}catch(error){
|
||||
if(runtime.signal.aborted)throw error
|
||||
await runtime.commit(async(tx:any)=>{
|
||||
const saved=(await tx.doc(Workspaces,ROOT)).items[task.input.id]
|
||||
saved.state='failed';saved.error=String(error)
|
||||
return {status:'terminal',outcome:{status:'completed',result:null}}
|
||||
},ctx)
|
||||
}
|
||||
},
|
||||
},abort:(_task:any,runtime:any,ctx:any)=>runtime.commit(()=>({status:'terminal',outcome:{status:'aborted'}}),ctx)})
|
||||
const ready=async(id:number,ctx:any)=>{
|
||||
const workspace=await get(id,ctx)
|
||||
if(!workspace)return
|
||||
if(workspace.state==='removed') {
|
||||
await host.harness().commit(async(tx:any)=>{
|
||||
const saved=(await tx.doc(Workspaces,ROOT)).items[id]
|
||||
if(saved.state==='removed'){
|
||||
saved.state='preparing'
|
||||
saved.preparation=await tx.createTask(Preparation,{id},{conversationId:id,ownership:{kind:'conversation'},background:true})
|
||||
}
|
||||
},ctx)
|
||||
}
|
||||
const saved=await get(id,ctx)
|
||||
if(saved.state==='preparing') {
|
||||
host.harness().resume()
|
||||
await host.harness().waitForTask(saved.preparation,ctx)
|
||||
}
|
||||
}
|
||||
const unavailable=async(id:number,ctx:any)=>{
|
||||
await ready(id,ctx)
|
||||
const workspace=await get(id,ctx)
|
||||
if(!workspace)return
|
||||
if(workspace.state!=='ready')return `Workspace is ${workspace.state}: ${workspace.error??''}`
|
||||
if(!await registered(host,workspace))return `Workspace checkout is missing or unregistered: ${workspace.path}`
|
||||
}
|
||||
const idle=async(id:number,ctx:any)=>{
|
||||
if((await host.harness().snapshot(LiveDoc,id,ctx))?.run||(await host.harness().snapshot(InboxDoc,id,ctx))?.items.length)
|
||||
throw new Error('Conversation is running or has queued input; wait or stop it first')
|
||||
}
|
||||
const discard=async(id:number,ctx:any)=>{
|
||||
await idle(id,ctx)
|
||||
const workspace=await get(id,ctx)
|
||||
if(!workspace)throw new Error('Conversation has no workspace')
|
||||
const identity={task:`discard-${id}`,conversations:await chain(host,id,ctx)}
|
||||
return serial(workspace.source,async()=>{
|
||||
if(workspace.state==='preparing')throw new Error('Workspace is preparing')
|
||||
if(workspace.state!=='removed') {
|
||||
await git(host,workspace.source,['worktree','remove',workspace.path],identity)
|
||||
await host.harness().commit(async(tx:any)=>{(await tx.doc(Workspaces,ROOT)).items[id].state='removed'},ctx)
|
||||
}
|
||||
return {text:`Discarded isolate; kept branch ${workspace.branch}.`}
|
||||
})
|
||||
}
|
||||
const release=async(ids:number[],ctx:any)=>{
|
||||
const kept:any[]=[]
|
||||
for(const id of ids) {
|
||||
const workspace=await get(id,ctx)
|
||||
if(!workspace||workspace.state==='removed')continue
|
||||
try {
|
||||
// Refuse detached work that its named branch would not retain.
|
||||
const identity={task:`release-${id}`,conversations:await chain(host,id,ctx)}
|
||||
await git(host,workspace.path,['merge-base','--is-ancestor','HEAD',workspace.branch],identity)
|
||||
await git(host,workspace.source,['merge-base','--is-ancestor',workspace.branch,workspace.target],identity)
|
||||
await discard(id,ctx)
|
||||
}catch(error){kept.push({id,reason:String(error)})}
|
||||
}
|
||||
return kept
|
||||
}
|
||||
const land=async(id:number,ctx:any)=>{
|
||||
await idle(id,ctx)
|
||||
const workspace=await get(id,ctx)
|
||||
if(workspace?.state==='removed'&&workspace.landed)return {text:`Landed commits from #${id} as ${workspace.landed}.`,landed:workspace.landed}
|
||||
if(!workspace||workspace.state!=='ready')throw new Error('Conversation has no ready workspace')
|
||||
const identity={task:`land-${id}`,conversations:await chain(host,id,ctx)}
|
||||
return serial(workspace.source,async()=>{
|
||||
if(await git(host,workspace.path,['status','--porcelain'],identity))throw new Error('Worktree is dirty')
|
||||
if(await git(host,workspace.source,['symbolic-ref','--short','HEAD'],identity)!==workspace.target)throw new Error('Target branch changed')
|
||||
if(await git(host,workspace.path,['symbolic-ref','--short','HEAD'],identity)!==workspace.branch)throw new Error('Workspace branch changed')
|
||||
const tip=await git(host,workspace.path,['rev-parse','HEAD'],identity)
|
||||
await git(host,workspace.source,['merge','--ff-only',tip],identity)
|
||||
await host.harness().commit(async(tx:any)=>{(await tx.doc(Workspaces,ROOT)).items[id].landed=tip},ctx)
|
||||
return {text:`Landed commits from #${id} as ${tip}.`,landed:tip}
|
||||
})
|
||||
}
|
||||
return {Preparation,serial,get,ready,unavailable,land,discard,release,chain:(id:number,ctx:any)=>chain(host,id,ctx),
|
||||
async create(tx:any,id:number,cwd:string,ancestors:string[]=[]) {
|
||||
const identity={task:`workspace-${id}`,conversations:[String(id),...ancestors]}
|
||||
const source=resolve(cwd,await git(host,cwd,['rev-parse','--show-toplevel'],identity))
|
||||
const target=await git(host,cwd,['symbolic-ref','--quiet','--short','HEAD'],identity)
|
||||
const base=await git(host,source,['rev-parse','HEAD'],identity)
|
||||
const index=await tx.doc(Workspaces,ROOT)
|
||||
if(index.items[id])throw new Error('Conversation already has a workspace')
|
||||
const preparation=await tx.createTask(Preparation,{id},{conversationId:id,ownership:{kind:'conversation'},background:true})
|
||||
const workspace={source,target,base,path:`/worktrees/${id}`,branch:`sausage/${id}`,state:'preparing',preparation}
|
||||
index.items[id]=workspace
|
||||
await configure(tx,id,{cwd:resolve(workspace.path,cwd.slice(source.length).replace(/^\//,''))})
|
||||
return workspace
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js && esbuild subagents-test.ts --bundle --format=esm --platform=browser --outfile=subagents-test-bundle.js && esbuild core-ext-test.ts --bundle --format=esm --platform=browser --outfile=core-ext-test-bundle.js && esbuild plan-todos-test.ts --bundle --format=esm --platform=browser --outfile=plan-todos-test-bundle.js && esbuild p2-test.ts --bundle --format=esm --platform=browser --outfile=p2-test-bundle.js && esbuild chain-test.ts --bundle --format=esm --platform=browser --outfile=chain-test-bundle.js && esbuild undo-harness-test.ts --bundle --format=esm --platform=browser --outfile=undo-harness-test-bundle.js && esbuild messages-test.ts --bundle --format=esm --platform=browser --outfile=messages-test-bundle.js && esbuild p3-test.ts --bundle --format=esm --platform=browser --outfile=p3-test-bundle.js && esbuild automations-test.ts --bundle --format=esm --platform=browser --outfile=automations-test-bundle.js && esbuild watchdog-test.ts --bundle --format=esm --platform=browser --outfile=watchdog-test-bundle.js" },
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js && esbuild subagents-test.ts --bundle --format=esm --platform=browser --outfile=subagents-test-bundle.js && esbuild core-ext-test.ts --bundle --format=esm --platform=browser --outfile=core-ext-test-bundle.js && esbuild plan-todos-test.ts --bundle --format=esm --platform=browser --outfile=plan-todos-test-bundle.js && esbuild p2-test.ts --bundle --format=esm --platform=browser --outfile=p2-test-bundle.js && esbuild chain-test.ts --bundle --format=esm --platform=browser --outfile=chain-test-bundle.js && esbuild undo-harness-test.ts --bundle --format=esm --platform=browser --outfile=undo-harness-test-bundle.js && esbuild messages-test.ts --bundle --format=esm --platform=browser --outfile=messages-test-bundle.js && esbuild p3-test.ts --bundle --format=esm --platform=browser --outfile=p3-test-bundle.js && esbuild automations-test.ts --bundle --format=esm --platform=browser --outfile=automations-test-bundle.js && esbuild watchdog-test.ts --bundle --format=esm --platform=browser --outfile=watchdog-test-bundle.js && esbuild workspaces-test.ts --bundle --format=esm --platform=browser --outfile=workspaces-test-bundle.js" },
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-durable": "1.5.0",
|
||||
"@earendil-works/pi-ai": "1.5.0",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,161 @@
|
||||
/** Scripted acceptance cases run through the world extension loader and real guest Git. */
|
||||
import { BACKGROUND_CONTEXT as context, awaitWithContext } from '@earendil-works/chord/context';
|
||||
import { createModels } from '@earendil-works/pi-ai/models';
|
||||
import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
|
||||
import { Harness, createRegistry, defineExtension, defineTool } from '@earendil-works/pi-durable';
|
||||
import { obj, str } from '@earendil-works/pi-schema';
|
||||
import { loadExtensions, fs } from './extensions';
|
||||
import { openStorage } from './sqlite';
|
||||
const assert=(value:any,message:string)=>{if(!value)throw new Error(message)};
|
||||
const models=createModels(),faux=fauxProvider({tokensPerSecond:100000});models.setProvider(faux.provider);
|
||||
const registry=createRegistry(),storage=await openStorage(globalThis.storagePath);
|
||||
const harness=await Harness.open(storage,{models,registry,settings:{get extensions(){return registry.snapshot().installed()}}},context);
|
||||
const root=await harness.root(context,{agent:{model:{provider:'faux',modelId:'faux-1'},cwd:'/project'}});
|
||||
if(globalThis.workspacesMode==='kill') {
|
||||
const install=registry.install.bind(registry);
|
||||
registry.install=(extension:any)=>install(extension.name!=='workspaces'?extension:{...extension,tasks:extension.tasks.map((task:any)=>task.definition.name!=='workspaces.lander'?task:{definition:{...task.definition,phases:{...task.definition.phases,land:async(record:any,runtime:any,ctx:any)=>{
|
||||
const commit=runtime.commit.bind(runtime);
|
||||
await taskDefinition(record,runtime,ctx);
|
||||
async function taskDefinition(record:any,rt:any,c:any){
|
||||
await task.definition.phases.land(record,{...rt,commit:async(...args:any[])=>{
|
||||
await commit(...args);
|
||||
const index=await harness.snapshot(globalThis.workspaceService.doc,root.id,context);
|
||||
if(index.queue.some((entry:any)=>entry.landed)) {
|
||||
await fs.write('/push-ready','yes');await awaitWithContext(new Promise(()=>{}),c);
|
||||
}
|
||||
}},c);
|
||||
}
|
||||
}}}})});
|
||||
}
|
||||
const probe=defineExtension({name:'guest',tools:[defineTool({name:'implement',description:'Write and commit in the prepared checkout',parameters:obj({name:str()}),replay:'safe',execute:async(args:any,api:any,ctx:any)=>{
|
||||
const cwd=(await api.agent(ctx)).cwd;
|
||||
assert(cwd.startsWith('/worktrees/'),'First tool needs isolated cwd');
|
||||
assert((await fs.stat(`${cwd}/base.txt`)).kind==='file','Checkout must exist before first tool');
|
||||
const workspace=await globalThis.workspaceService.get(api.conversationId,ctx);
|
||||
assert(workspace.state==='ready','Preparation must finish');
|
||||
await fs.write(`${cwd}/${args.name}`,'feature');
|
||||
for(const args of [['add','.'],['-c','user.name=Test','-c','user.email=test@example.com','commit','--no-gpg-sign','-m','feature']])await globalThis.workspaceHost.git(cwd,...args);
|
||||
return {output:[{type:'text',text:'committed'}]};
|
||||
}})]});
|
||||
const route=(transcript:any):any=>{
|
||||
faux.appendResponses([route]);
|
||||
const last=transcript.messages.findLast((message:any)=>message.role!=='system');
|
||||
const text=typeof last.content==='string'?last.content:last.content.filter((part:any)=>part.type==='text').map((part:any)=>part.text).join('');
|
||||
if(last.role==='user'&&text.startsWith('CALL ')){const {tool,args}=JSON.parse(text.slice(5));return fauxAssistantMessage([fauxToolCall(tool,args)],{stopReason:'toolUse'});}
|
||||
if(last.role==='user'&&text==='IMPLEMENT')return fauxAssistantMessage([fauxToolCall('implement',{name:'feature.txt'})],{stopReason:'toolUse'});
|
||||
if(last.role==='user'&&text==='BROKEN')return fauxAssistantMessage([fauxToolCall('implement',{name:'broken.txt'})],{stopReason:'toolUse'});
|
||||
return fauxAssistantMessage('Done');
|
||||
};faux.setResponses([route]);
|
||||
const loader=await loadExtensions(harness,registry,storage,models,context,[probe]);
|
||||
assert(!loader.state().error,`Load: ${loader.state().error}`);
|
||||
const service=globalThis.workspaceService,host=globalThis.workspaceHost;
|
||||
const call=async(tool:string,args:any)=>{
|
||||
await(await root.submit({type:'input',content:`CALL ${JSON.stringify({tool,args})}`},context)).wait(context);
|
||||
const result=(await root.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1)?.model?.[0] as any;
|
||||
assert(!result?.isError,`Tool error: ${JSON.stringify(result)}`);
|
||||
return result;
|
||||
};
|
||||
const submit=async(id:number)=>{
|
||||
const {Started}=await import(`file:///run/sausage/extensions/${await fs.read('/src/sausage/.git/sausage-last-good')}/extensions/messages/addresses.ts`);
|
||||
await harness.commit(async(tx:any)=>{(await tx.doc(Started,id)).by=root.id},context);
|
||||
const child=await harness.conversation(id,context);
|
||||
await(await child.submit({type:'input',content:`CALL ${JSON.stringify({tool:'workspace',args:{action:'submit',divergence:'None'}})}`},context)).wait(context);
|
||||
const result=(await child.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1)?.model?.[0] as any;
|
||||
assert(!result?.isError,`Workspace tool: ${JSON.stringify(result)}`);
|
||||
return JSON.parse(result.content.filter((part:any)=>part.type==='text').map((part:any)=>part.text).join(''));
|
||||
};
|
||||
const children=async()=>{
|
||||
const {Subagents}=await import(`file:///run/sausage/extensions/${await fs.read('/src/sausage/.git/sausage-last-good')}/extensions/subagents/api.ts`);
|
||||
return Object.values((await harness.snapshot(Subagents,root.id,context)).agents) as any[];
|
||||
};
|
||||
const waitQueue=async()=>{
|
||||
const deadline=Date.now()+30000;
|
||||
while((await harness.snapshot(service.doc,root.id,context)).queue.length) {
|
||||
if(Date.now()>deadline)throw new Error(`Lander deadline: ${JSON.stringify(await harness.snapshot(service.doc,root.id,context))}`);
|
||||
await new Promise(r=>setTimeout(r,0));
|
||||
}
|
||||
await harness.waitForIdle(context);
|
||||
};
|
||||
const reports=async(id:number)=>(await(await harness.conversation(id,context)).context(context)).entries.filter((entry:any)=>entry.kind==='workspaces.lander-report');
|
||||
const notes=async()=>{
|
||||
const {Messages}=await import(`file:///run/sausage/extensions/${await fs.read('/src/sausage/.git/sausage-last-good')}/extensions/messages/api.ts`);
|
||||
const deadline=Date.now()+30000;
|
||||
for(;;) {
|
||||
const notes=((await harness.snapshot(Messages,root.id,context)).messages as any[]).filter(message=>message.text.startsWith('[lander] Submission')&&message.text.includes(' landed as '));
|
||||
if(!notes.some(note=>note.state==='sent'))return notes;
|
||||
assert(Date.now()<deadline,`Landing note delivery: ${JSON.stringify(notes)}`);
|
||||
await new Promise(r=>setTimeout(r,0));
|
||||
}
|
||||
};
|
||||
harness.resume();
|
||||
try {
|
||||
if(globalThis.workspacesMode==='resume') {
|
||||
await waitQueue();
|
||||
const index=await harness.snapshot(service.doc,root.id,context);
|
||||
const result=Object.values(index.results)[0] as any;
|
||||
assert(result?.outcome==='landed',`Recovery result ${JSON.stringify(index)}`);
|
||||
const [child]=await children();
|
||||
const report=await reports(child.conversationId);
|
||||
assert(report.length===1,'Restart must report once');
|
||||
assert(report[0].data.sha===result.sha,'Report must retain landed.sha');
|
||||
const delivered=await notes();
|
||||
assert(delivered.length===1&&delivered[0].state==='delivered'&&delivered[0].text.includes(result.sha),`Restart must notify the starter once with landed.sha: ${JSON.stringify(delivered)}`);
|
||||
assert(await host.git('/remote.git','rev-parse','main')===result.sha,`Push must retain landed.sha: ${JSON.stringify(result)}`);
|
||||
assert((await host.git('/project','log','--format=%s')).split('\n').filter((line:string)=>line.startsWith('Merge from')).length===1,'Restart must not merge again');
|
||||
assert((await host.eventLog.recent({events:[{kind:'event.workspaces.landed'}]})).length===1,'Restart must emit once');
|
||||
} else {
|
||||
await call('subagent',{action:'start',agent:'worker',task:globalThis.workspacesMode==='normal'?'BROKEN':'IMPLEMENT'});
|
||||
if(globalThis.workspacesMode==='normal') {
|
||||
const [broken]=await children();
|
||||
const before=globalThis.workspaceExecutions.length;
|
||||
assert(!await service.unavailable(broken.conversationId,context),'Ready VFS gate');
|
||||
assert(globalThis.workspaceExecutions.length===before,'Ready gate must not spawn Git');
|
||||
const checkout=await service.get(broken.conversationId,context);
|
||||
await fs.rename(`${checkout.path}/.git`,`${checkout.path}/.git.saved`);
|
||||
assert((await service.unavailable(broken.conversationId,context)).includes('missing or unregistered'),'Gate must reject missing registration');
|
||||
const blocked=await harness.conversation(broken.conversationId,context);
|
||||
await(await blocked.submit({type:'input',content:`CALL ${JSON.stringify({tool:'implement',args:{name:'guard.txt'}})}`},context)).wait(context);
|
||||
const blockedResult=(await blocked.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1)?.model?.[0] as any;
|
||||
assert(blockedResult.isError&&JSON.stringify(blockedResult).includes('missing or unregistered'),'beforeTool must block the guest tool');
|
||||
assert(!await fs.stat(`${checkout.path}/guard.txt`).then(()=>true,(error:any)=>{if(error.code==='not_found')return false;throw error}),'Blocked tool must not write');
|
||||
await fs.rename(`${checkout.path}/.git.saved`,`${checkout.path}/.git`);
|
||||
assert(globalThis.workspaceExecutions.length===before,'Failed VFS gate must not spawn Git');
|
||||
const failed=await submit(broken.conversationId);
|
||||
await waitQueue();
|
||||
const result=(await harness.snapshot(service.doc,root.id,context)).results[failed.id];
|
||||
assert(result.outcome==='failed'&&result.reason.includes('Check sh'),`Check must fail: ${JSON.stringify(result)}`);
|
||||
assert((await reports(broken.conversationId)).length===1,'Failed check must return to submitter');
|
||||
const failedWorkspace=await service.get(broken.conversationId,context);
|
||||
assert((await service.release([broken.conversationId],context)).length===1,'Release must retain unlanded work');
|
||||
await call('subagent',{action:'discard',id:broken.conversationId});
|
||||
assert(await host.git('/project','rev-parse',failedWorkspace.branch),'Discard must retain its branch');
|
||||
await call('subagent',{action:'start',agent:'worker',task:'IMPLEMENT'});
|
||||
const child=(await children()).at(-1);
|
||||
await call('subagent',{action:'land',id:child.conversationId});
|
||||
assert(await fs.read('/project/feature.txt')==='feature','Parent fast-forward');
|
||||
assert((await service.get(child.conversationId,context)).state==='removed','Land must release checkout');
|
||||
await call('subagent',{action:'start',agent:'worker',task:'BROKEN'});
|
||||
}
|
||||
const submitter=(await children()).at(-1).conversationId;
|
||||
const submission=await submit(submitter);
|
||||
if(globalThis.workspacesMode==='kill')await new Promise(()=>{});
|
||||
await waitQueue();
|
||||
const result=(await harness.snapshot(service.doc,root.id,context)).results[submission.id];
|
||||
// feature.txt came from synchronous landing, so this submission passes its real shell check too.
|
||||
assert(result.outcome==='landed',`Landing failed: ${JSON.stringify(result)}`);
|
||||
assert(await host.git('/remote.git','rev-parse','main')===result.sha,`Local bare push: ${JSON.stringify(result)}`);
|
||||
assert((await reports(submitter)).length===1,'Submission report');
|
||||
const delivered=await notes();
|
||||
assert(delivered.length===1&&delivered[0].state==='delivered'&&delivered[0].from===submitter&&delivered[0].to===root.id&&delivered[0].reply===false&&delivered[0].text.includes(result.sha),`Landing note must use the messages API: ${JSON.stringify(delivered)}`);
|
||||
const commands=globalThis.workspaceExecutions.filter((entry:any)=>typeof entry.command==='string'||entry.command[1]==='fetch'||entry.command[1]==='push');
|
||||
const push=commands.findLast((entry:any)=>entry.command[1]==='push');
|
||||
assert(push,'Real push must run');
|
||||
assert(JSON.stringify(push.options.conversations)===JSON.stringify([String(submitter),String(root.id)]),'Push must use the exact submitter chain');
|
||||
for(const entry of commands) {
|
||||
assert(entry.options.task!==undefined,'Lander exec needs its task tag');
|
||||
if(typeof entry.command==='string')assert(entry.options.timeout===600,'Checks need a bounded default timeout');
|
||||
assert(entry.options.conversations[1]===String(root.id),'Lander exec needs the submitter ancestor');
|
||||
assert(entry.options.conversations[0]!==String(root.id),'Lander exec needs the submitter first');
|
||||
}
|
||||
}
|
||||
}finally{await loader.close();await harness.close(context)}
|
||||
@@ -41,6 +41,45 @@ async fn command(runtime: &Arc<Runtime>, args: &[&str]) -> String {
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn tools_restore_one_call_or_the_whole_turn() {
|
||||
let target = PathBuf::from(std::env::var("CARGO_TARGET_DIR").unwrap());
|
||||
let repo = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap();
|
||||
assert!(
|
||||
std::process::Command::new("nu")
|
||||
.args(["--no-config-file", "userland/cwd-init/build.nu"])
|
||||
.current_dir(repo)
|
||||
.status()
|
||||
.unwrap()
|
||||
.success()
|
||||
);
|
||||
assert!(
|
||||
std::process::Command::new("cargo")
|
||||
.args([
|
||||
"build",
|
||||
"--release",
|
||||
"--locked",
|
||||
"--target",
|
||||
"wasm32-wasip2",
|
||||
"-p",
|
||||
"git",
|
||||
"-p",
|
||||
"undo",
|
||||
"--target-dir"
|
||||
])
|
||||
.arg(&target)
|
||||
.args(["-j", "2"])
|
||||
.current_dir(repo.join("userland"))
|
||||
.status()
|
||||
.unwrap()
|
||||
.success()
|
||||
);
|
||||
assert!(
|
||||
std::process::Command::new("nu")
|
||||
.args(["--no-config-file", "userland/brush/build.nu"])
|
||||
.env("CARGO_TARGET_DIR", &target)
|
||||
.current_dir(repo)
|
||||
.status()
|
||||
.unwrap()
|
||||
.success()
|
||||
);
|
||||
for since in [false, true] {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
let world = World::open(temp.path()).unwrap();
|
||||
|
||||
@@ -0,0 +1,299 @@
|
||||
//! World worktrees, landing checks, and SIGKILL recovery before push.
|
||||
use sausage::{harness::run_module, world::World};
|
||||
use std::{
|
||||
io::Write,
|
||||
path::Path,
|
||||
process::{Command, Stdio},
|
||||
sync::Arc,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
const BUNDLE: &str = include_str!("../js/workspaces-test-bundle.js");
|
||||
fn git(path: &Path, args: &[&str]) -> anyhow::Result<()> {
|
||||
assert!(
|
||||
Command::new("git")
|
||||
.args(args)
|
||||
.current_dir(path)
|
||||
.env("GIT_AUTHOR_NAME", "Test")
|
||||
.env("GIT_AUTHOR_EMAIL", "test@example.com")
|
||||
.env("GIT_COMMITTER_NAME", "Test")
|
||||
.env("GIT_COMMITTER_EMAIL", "test@example.com")
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
fn prepare(dir: &Path) -> anyhow::Result<()> {
|
||||
let disk = dir.join("world");
|
||||
let repo = disk.join("src/sausage");
|
||||
if repo.join(".git").exists() {
|
||||
return Ok(());
|
||||
}
|
||||
let source = Path::new(env!("CARGO_MANIFEST_DIR")).join("..");
|
||||
let target = Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap();
|
||||
let userland = source.join("userland");
|
||||
if !userland.join("cwd-init/init.o").exists() {
|
||||
assert!(
|
||||
Command::new("nu")
|
||||
.args(["--no-config-file", "cwd-init/build.nu"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
Command::new("cargo")
|
||||
.args(["build", "--release", "-p", "git", "--manifest-path"])
|
||||
.arg(userland.join("Cargo.toml"))
|
||||
.args(["--target", "wasm32-wasip2", "--target-dir"])
|
||||
.arg(target)
|
||||
.args(["-j", "2", "--locked"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
if !target.join("wasm32-wasip2/release/brush.wasm").exists() {
|
||||
assert!(
|
||||
Command::new("nu")
|
||||
.arg("brush/build.nu")
|
||||
.env("CARGO_TARGET_DIR", target)
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
}
|
||||
std::fs::create_dir_all(disk.join("bin"))?;
|
||||
for (binary, name) in [("git", "git"), ("brush", "sh")] {
|
||||
std::fs::copy(
|
||||
target.join(format!("wasm32-wasip2/release/{binary}.wasm")),
|
||||
disk.join(format!("bin/{name}.wasm")),
|
||||
)?;
|
||||
}
|
||||
for extension in [
|
||||
"subagents",
|
||||
"models",
|
||||
"workspaces",
|
||||
"messages",
|
||||
"lineage",
|
||||
"conversations",
|
||||
] {
|
||||
let dest = repo.join("extensions").join(extension);
|
||||
std::fs::create_dir_all(&dest)?;
|
||||
for entry in std::fs::read_dir(source.join("extensions").join(extension))? {
|
||||
let entry = entry?;
|
||||
if entry.path().extension().is_some_and(|ext| ext == "ts") {
|
||||
std::fs::copy(entry.path(), dest.join(entry.file_name()))?;
|
||||
}
|
||||
}
|
||||
}
|
||||
let server = repo.join("extensions/workspaces/server.ts");
|
||||
let code = std::fs::read_to_string(&server)?;
|
||||
std::fs::write(server,code.replace("const service=workspaces(host)","globalThis.workspaceExecutions??=[]; const exec=host.exec; host.exec=(command,options)=>{globalThis.workspaceExecutions.push({command,options});return exec(command,options)}; const service=workspaces(host)"))?;
|
||||
std::fs::create_dir_all(repo.join("extensions/core"))?;
|
||||
std::fs::copy(
|
||||
source.join("extensions/core/agents.ts"),
|
||||
repo.join("extensions/core/agents.ts"),
|
||||
)?;
|
||||
std::fs::create_dir_all(repo.join("extensions/probe"))?;
|
||||
std::fs::write(
|
||||
repo.join("extensions/probe/server.ts"),
|
||||
"export default host => { globalThis.workspaceService=host.need({name:'workspaces'}); globalThis.workspaceHost=host; return {name:'probe'}; }",
|
||||
)?;
|
||||
std::fs::create_dir_all(repo.join("agents"))?;
|
||||
std::fs::write(
|
||||
repo.join("agents/models.json"),
|
||||
r#"{"default":{"model":"faux/faux-1"}}"#,
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("agents/worker.md"),
|
||||
"---\nname: worker\ndescription: Implement\nisolate: worktree\n---\nImplement\n",
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("settings.json"),
|
||||
r#"{"lander":{"push":"origin","checks":["sh -c 'test -f feature.txt'"]}}"#,
|
||||
)?;
|
||||
let project = disk.join("project");
|
||||
std::fs::create_dir_all(&project)?;
|
||||
std::fs::write(project.join("base.txt"), "base")?;
|
||||
for path in [&project, &repo] {
|
||||
git(path, &["init", "-b", "main"])?;
|
||||
git(path, &["config", "user.name", "Test"])?;
|
||||
git(path, &["config", "user.email", "test@example.com"])?;
|
||||
git(path, &["add", "."])?;
|
||||
git(path, &["commit", "--no-gpg-sign", "-m", "fixture"])?;
|
||||
}
|
||||
assert!(
|
||||
Command::new("git")
|
||||
.args(["clone", "--bare"])
|
||||
.arg(&project)
|
||||
.arg(disk.join("remote.git"))
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
git(&project, &["remote", "add", "origin", "/remote.git"])?;
|
||||
Ok(())
|
||||
}
|
||||
// The world Git supports smart HTTP and SSH, not file remotes.
|
||||
struct GitServer(Arc<tiny_http::Server>, Option<std::thread::JoinHandle<()>>);
|
||||
impl Drop for GitServer {
|
||||
fn drop(&mut self) {
|
||||
self.0.unblock();
|
||||
self.1.take().unwrap().join().unwrap();
|
||||
}
|
||||
}
|
||||
fn serve(dir: &Path) -> anyhow::Result<GitServer> {
|
||||
let root = dir.join("world");
|
||||
git(
|
||||
&root.join("remote.git"),
|
||||
&["config", "http.receivepack", "true"],
|
||||
)?;
|
||||
let server =
|
||||
Arc::new(tiny_http::Server::http("127.0.0.1:0").map_err(|e| anyhow::anyhow!("{e}"))?);
|
||||
let port = server.server_addr().to_ip().unwrap().port();
|
||||
git(
|
||||
&root.join("project"),
|
||||
&[
|
||||
"remote",
|
||||
"set-url",
|
||||
"origin",
|
||||
&format!("http://127.0.0.1:{port}/remote.git"),
|
||||
],
|
||||
)?;
|
||||
let serving = server.clone();
|
||||
let thread = std::thread::spawn(move || {
|
||||
while let Ok(mut request) = serving.recv() {
|
||||
let url = request.url().to_owned();
|
||||
let (path, query) = url.split_once('?').unwrap_or((&url, ""));
|
||||
let content_type = request
|
||||
.headers()
|
||||
.iter()
|
||||
.find(|h| h.field.equiv("Content-Type"))
|
||||
.map(|h| h.value.as_str().to_owned())
|
||||
.unwrap_or_default();
|
||||
let mut body = Vec::new();
|
||||
request.as_reader().read_to_end(&mut body).unwrap();
|
||||
let mut process = Command::new("git")
|
||||
.arg("http-backend")
|
||||
.env("GIT_PROJECT_ROOT", &root)
|
||||
.env("GIT_HTTP_EXPORT_ALL", "1")
|
||||
.env("REQUEST_METHOD", request.method().as_str())
|
||||
.env("PATH_INFO", path)
|
||||
.env("QUERY_STRING", query)
|
||||
.env("CONTENT_TYPE", content_type)
|
||||
.env("CONTENT_LENGTH", body.len().to_string())
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.spawn()
|
||||
.unwrap();
|
||||
process.stdin.take().unwrap().write_all(&body).unwrap();
|
||||
let output = process.wait_with_output().unwrap();
|
||||
assert!(output.status.success());
|
||||
let split = output
|
||||
.stdout
|
||||
.windows(4)
|
||||
.position(|p| p == b"\r\n\r\n")
|
||||
.unwrap();
|
||||
let headers = std::str::from_utf8(&output.stdout[..split]).unwrap();
|
||||
let mut response = tiny_http::Response::from_data(output.stdout[split + 4..].to_vec());
|
||||
for line in headers.split("\r\n") {
|
||||
let (name, value) = line.split_once(':').unwrap();
|
||||
if name.eq_ignore_ascii_case("status") {
|
||||
response = response.with_status_code(
|
||||
value
|
||||
.trim()
|
||||
.split(' ')
|
||||
.next()
|
||||
.unwrap()
|
||||
.parse::<u16>()
|
||||
.unwrap(),
|
||||
);
|
||||
} else {
|
||||
response = response
|
||||
.with_header(tiny_http::Header::from_bytes(name, value.trim()).unwrap());
|
||||
}
|
||||
}
|
||||
let _ = request.respond(response);
|
||||
}
|
||||
});
|
||||
Ok(GitServer(server, Some(thread)))
|
||||
}
|
||||
async fn run(dir: &Path, mode: &str) -> anyhow::Result<()> {
|
||||
let world = World::open(dir)?;
|
||||
prepare(dir)?;
|
||||
let exec = sausage::process::exec(World {
|
||||
root: world.root.clone(),
|
||||
host_dir: world.host_dir.clone(),
|
||||
processes: world.processes.clone(),
|
||||
})?;
|
||||
run_module(
|
||||
world,
|
||||
exec,
|
||||
&format!("globalThis.workspacesMode={mode:?};\n{BUNDLE}"),
|
||||
)
|
||||
.await
|
||||
}
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn coordination() -> anyhow::Result<()> {
|
||||
let dir = tempfile::tempdir()?;
|
||||
World::open(dir.path())?;
|
||||
prepare(dir.path())?;
|
||||
let _server = serve(dir.path())?;
|
||||
run(dir.path(), "normal").await
|
||||
}
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn restart_child() -> anyhow::Result<()> {
|
||||
if let Ok(dir) = std::env::var("WORKSPACES_DIR") {
|
||||
run(Path::new(&dir), &std::env::var("WORKSPACES_MODE")?).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
#[test]
|
||||
fn sigkill_between_fast_forward_and_push() -> anyhow::Result<()> {
|
||||
let dir = tempfile::tempdir()?;
|
||||
World::open(dir.path())?;
|
||||
prepare(dir.path())?;
|
||||
let _server = serve(dir.path())?;
|
||||
let spawn = |mode: &str| {
|
||||
Command::new(std::env::current_exe().unwrap())
|
||||
.args(["--exact", "restart_child", "--nocapture"])
|
||||
.env("WORKSPACES_DIR", dir.path())
|
||||
.env("WORKSPACES_MODE", mode)
|
||||
.spawn()
|
||||
};
|
||||
let mut child = spawn("kill")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(60);
|
||||
while !dir.path().join("world/push-ready").exists() {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
anyhow::bail!("Child exited before push: {status}");
|
||||
}
|
||||
if Instant::now() > deadline {
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
anyhow::bail!("Lander did not reach push");
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
// A later target commit must not replace the recorded SHA in the push or report.
|
||||
git(
|
||||
&dir.path().join("world/project"),
|
||||
&["commit", "--allow-empty", "--no-gpg-sign", "-m", "later"],
|
||||
)?;
|
||||
for _ in 0..2 {
|
||||
let mut child = spawn("resume")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(60);
|
||||
loop {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
assert!(status.success());
|
||||
break;
|
||||
}
|
||||
if Instant::now() > deadline {
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
anyhow::bail!("Recovery timed out");
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user