Port workspaces and the durable lander into the world

extensions/workspaces/README.md

@@ -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.

extensions/workspaces/api.ts

@@ -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')

extensions/workspaces/lander.ts

@@ -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)
+  }}
+}

extensions/workspaces/server.ts

@@ -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??{})}]}
+    }})],
+  })
+}

extensions/workspaces/tree.ts

@@ -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)}

extensions/workspaces/workspaces.ts

@@ -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
+    },
+  }
+}