import {randomUUID} from 'node:crypto' import {join,dirname,posix} from 'node:path' import {fileURLToPath} from 'node:url' import {mkdir,rm} from 'node:fs/promises' import {existsSync} from 'node:fs' import type {DatabaseSync} from 'node:sqlite' import {z} from 'zod' import {HttpError} from '../../server/errors.js' import type {GatewayFrame,GatewayTarget} from '../../server/workspaceGateway.js' import {ProfileComputerSession} from './profileSession.js' import {UNCONFIGURED_COMPUTER_IMAGE} from '../../shared/runner.js' import type {RunnerConfiguration} from '../../shared/runner.js' import {ContainerComputerProvider,COMPUTER_WORKSPACE,CUA_DRIVER,CUA_SOCKET,type ComputerSpecification} from '../computers/container.js' import {ComputerPool,type ComputerLease} from '../computers/pool.js' import {ComposeComputerProvider} from '../computers/compose.js' import {COMPOSE_DESKTOP_IMAGE,type DesktopRelay} from '../../shared/composeDesktops.js' import {ComputerControls} from './control.js' import {ComputerPublicProxy} from '../network/computerProxy.js' import {HermesWorkerProcess,type WorkerModel,type WorkerProxyEnvironment,type WorkerContextConfiguration,type WorkerTool,type WorkerFrame} from './process.js' import {HOST_TOOLS,hostTool,hostPath,readHostFile,writeHostFile} from './hostTools.js' import {homedir,platform} from 'node:os' export interface ComputerTarget {environmentId:string;ownerKey:string;agentId:string;hostAccess?:boolean;profileSession?:boolean} export interface WorkerSession {agentId?:string;id:string;profile:string;ownerKey:string;environmentId:string;cwd:string;configuredCwd:string;history:any[];outcome?:'complete'|'failed'|'uncertain';vmExecution?:'profile';profileSessionId?:string} interface Live {workMarker?:string;paused?:boolean;pauseJob?:Promise;segment?:number;toolCalls:Set>;journal:Map;proxy?:ComputerPublicProxy;session:WorkerSession;model:WorkerModel;contextConfig?:WorkerContextConfiguration;proxyEnv?:WorkerProxyEnvironment;lease:ComputerLease;controller:AbortController;worker?:HermesWorkerProcess;finishing?:Promise;stopping?:Promise;releasing?:Promise;images:any[];running:boolean;received:boolean} const object=(properties:Record,required:string[])=>({type:'object',properties,required,additionalProperties:false}) const text={type:'string'} const TOOLS:WorkerTool[]=[ {name:'computer_export',description:'将隔离电脑中的文件作为可下载附件回传当前会话,最大 25 MiB。',inputSchema:object({path:text,name:text},['path'])}, {name:'computer_shell',description:'在当前隔离电脑和 Profile 工作目录执行 shell 命令。',inputSchema:object({command:text},['command'])}, {name:'computer_read_file',description:'读取隔离电脑内的 UTF-8 文本文件。',inputSchema:object({path:text},['path'])}, {name:'computer_write_file',description:'向隔离电脑写入 UTF-8 文本文件。',inputSchema:object({path:text,content:text},['path','content'])}, {name:'computer_desktop_state',description:'查看隔离电脑当前桌面,返回真实截图。',inputSchema:object({},[])}, {name:'computer_action',description:'在隔离电脑内调用 CUA 动作;先查看桌面并使用可用动作。',inputSchema:object({name:text,arguments:{type:'object'}},['name','arguments'])}, ] const pathField=z.string().min(1).max(4096).refine(value=>!value.includes('\0')) const PROFILE_COMPUTER_RULES='当前使用本机协作模式:沿用基础 Profile 的本机工具、配置和上下文,本地虚拟机是独立的 Linux 电脑。computer_shell/read_file/write_file/desktop_state/action/export 只操作虚拟机;本机文件、终端和程序使用 Profile 已授权的原生工具。computer_copy_file 在两个环境间显式复制文件。调用前明确目标环境,文件路径和浏览器登录互不通用;需要回传虚拟机文件时使用 computer_export。人工接管期间停止操作,交还后先检查当前状态。' export class ComputerRuntime { retainDesktops=false readonly controls:ComputerControls readonly provider:ContainerComputerProvider readonly pool:ComputerPool readonly ready:Promise readonly config:NonNullable readonly gateways=new Map() readonly workers=new Map() readonly script:string readonly proxyScript:string constructor(readonly db:DatabaseSync,config:RunnerConfiguration,readonly home:string,script?:string,relay?:DesktopRelay){ if(!config.computers)throw new HttpError(409,'执行节点尚未配置隔离电脑','computer_unavailable') this.config=config.computers this.script=script??join(dirname(fileURLToPath(import.meta.url)),'hermes_worker.py') this.proxyScript=existsSync(join(dirname(this.script),'guest_proxy.py'))?join(dirname(this.script),'guest_proxy.py'):join(dirname(this.script),'../network/guest_proxy.py') if(this.config.managedBy==='compose'){ if(!relay)throw new HttpError(409,'缺少 Compose 桌面连接','compose_desktop_unavailable') this.config.imageId=COMPOSE_DESKTOP_IMAGE;this.config.network='none' this.provider=new ComposeComputerProvider(config.runnerId,home,relay) }else this.provider=new ContainerComputerProvider(this.config.runtime,config.runnerId,home) this.pool=new ComputerPool(db,this.provider,{concurrent:this.config.maxConcurrent??2,environments:32,ttlMs:30000}) db.exec('CREATE TABLE IF NOT EXISTS retired_computers(id TEXT PRIMARY KEY,owner_key TEXT NOT NULL,spec TEXT)') db.exec('CREATE TABLE IF NOT EXISTS computer_sessions(id TEXT PRIMARY KEY,value TEXT NOT NULL)') db.exec('CREATE TABLE IF NOT EXISTS computer_image_bindings(id TEXT PRIMARY KEY,owner_key TEXT NOT NULL,image_id TEXT NOT NULL)') this.controls=new ComputerControls(this) this.ready=this.pool.recover();void this.ready.catch(()=>{}) } assertTarget(meta:ComputerTarget){if(this.db.prepare('SELECT id FROM retired_computers WHERE id=?').get(meta.environmentId))throw new HttpError(410,'临时电脑已退役','computer_retired')} imageFor(meta:ComputerTarget):string { const spec=this.pool.definition(meta.ownerKey,meta.environmentId) const binding=this.db.prepare('SELECT owner_key,image_id FROM computer_image_bindings WHERE id=?').get(meta.environmentId) as {owner_key:string;image_id:string}|undefined if(binding&&binding.owner_key!==meta.ownerKey)throw new HttpError(403,'电脑归属不匹配','computer_owner_mismatch') return spec?.imageId??binding?.image_id??this.config.imageId } async retire(meta:ComputerTarget){ const old=this.db.prepare('SELECT owner_key,spec FROM retired_computers WHERE id=?').get(meta.environmentId) as {owner_key:string;spec?:string}|undefined if(old&&old.owner_key!==meta.ownerKey)throw new HttpError(403,'电脑归属不匹配','computer_owner_mismatch') this.pool.definition(meta.ownerKey,meta.environmentId) this.db.prepare('INSERT OR IGNORE INTO retired_computers VALUES(?,?,NULL)').run(meta.environmentId,meta.ownerKey) await Promise.all([...this.workers.values()].filter(worker=>worker.environmentId===meta.environmentId&&worker.ownerKey===meta.ownerKey).map(worker=>worker.process.close())) const current=this.pool.status(meta.ownerKey).find(row=>row.environmentId===meta.environmentId) if(current?.holderId)await this.pool.stopHolder(meta.ownerKey,meta.environmentId,current.holderId) let spec=old?.spec?JSON.parse(old.spec):undefined spec??=this.pool.definition(meta.ownerKey,meta.environmentId) if(spec)this.db.prepare('UPDATE retired_computers SET spec=? WHERE id=?').run(JSON.stringify(spec),meta.environmentId) await this.pool.retire(meta.ownerKey,meta.environmentId) if(spec)await this.provider.deleteWorkspace(spec) const sessions=(this.db.prepare('SELECT id,value FROM computer_sessions').all() as {id:string;value:string}[]).filter(row=>{const session=JSON.parse(row.value);return session.environmentId===meta.environmentId&&session.ownerKey===meta.ownerKey}) for(const row of sessions){if(!/^[0-9a-f-]{36}$/.test(row.id))throw new Error('电脑会话编号无效');await rm(join(this.home,'workers',row.id),{recursive:true,force:true});this.db.prepare('DELETE FROM computer_sessions WHERE id=?').run(row.id)} this.db.prepare('UPDATE retired_computers SET spec=NULL WHERE id=?').run(meta.environmentId) } save(session:WorkerSession){this.db.prepare('INSERT INTO computer_sessions VALUES(?,?) ON CONFLICT(id) DO UPDATE SET value=excluded.value').run(session.id,JSON.stringify(session))} session(id:string,meta:ComputerTarget,profile:string):WorkerSession { const row=this.db.prepare('SELECT value FROM computer_sessions WHERE id=?').get(id) as {value:string}|undefined const session:WorkerSession|undefined=row?JSON.parse(row.value):undefined if(!session||(session.agentId??session.environmentId)!==meta.agentId||session.ownerKey!==meta.ownerKey||session.environmentId!==meta.environmentId||session.profile!==profile)throw new HttpError(404,'电脑会话不存在或不属于当前任务','computer_session_missing') if((session.vmExecution==='profile')!==(meta.profileSession===true))throw new HttpError(409,'虚拟机执行方式已改变,请使用新会话','computer_session_mode_changed') return session } async resolve(profile:string,signal?:AbortSignal){ return this.resolveProfile(profile,'resolve',signal) } async resolveWorkspace(profile:string,signal?:AbortSignal){ return this.resolveProfile(profile,'resolve-workspace',signal) } private async resolveProfile(profile:string,mode:'resolve'|'resolve-workspace',signal?:AbortSignal){ const resolver=new HermesWorkerProcess(this.config.python,this.script,{mode,profile,defaultCwd:COMPUTER_WORKSPACE,hermesSource:this.config.hermesSource,hermesHome:this.config.hermesHome}) const abort=()=>{void resolver.close()};signal?.addEventListener('abort',abort,{once:true});if(signal?.aborted)abort() try{const result=await resolver.wait('resolved');return this.provider.fixedCapacity?{...result,cwd:COMPUTER_WORKSPACE}:result}finally{signal?.removeEventListener('abort',abort);await resolver.close()} } async desktop(meta:ComputerTarget,profile:string,action:'create'|'start'|'stop'|'recreate'|'remove',authorize:()=>void){ if(this.provider.fixedCapacity)throw new HttpError(409,'桌面由 Compose 创建和管理,不能在界面增删或重建','compose_desktop_managed') this.assertTarget(meta);authorize() // Stopping/removing an existing desktop must remain possible even when // its Profile or model was removed or is temporarily misconfigured. if(action==='stop'||action==='remove'){ const spec=this.pool.definition(meta.ownerKey,meta.environmentId) if(spec)await this.pool.desktop(spec,action,authorize) return } const resolved=await this.resolveWorkspace(profile);authorize() await this.pool.desktop({id:meta.environmentId,ownerKey:meta.ownerKey,imageId:this.imageFor(meta),cwd:meta.environmentId!==meta.agentId?(this.pool.definition(meta.ownerKey,meta.environmentId)?.cwd??COMPUTER_WORKSPACE):resolved.cwd,network:this.config.network??'none'},action,authorize) } } export class ComputerGateway { onEvent:(frame:GatewayFrame)=>void=()=>{} onDisconnect:()=>void=()=>{} private live?:Live private recoverySession?:WorkerSession private closed=false private controller=new AbortController() private timer?:ReturnType private checking=false private hostCwd=homedir() private profileSession?:ProfileComputerSession private team?:{id:string;catalog:WorkerTool[];call(name:string,args:Record,callId:string):Promise} constructor(readonly runtime:ComputerRuntime,readonly meta:ComputerTarget,readonly workId:string,readonly check:()=>Promise,readonly publish:(body:Record)=>Promise,readonly profileTarget?:GatewayTarget){ } async connect(){this.runtime.assertTarget(this.meta);await this.runtime.ready;await this.authorized()} private async authorized(){try{await this.check();this.guard()}catch(error){const stopping=this.interrupt().catch(()=>{});if(!this.profileSession)await stopping;if(!this.closed)this.onDisconnect();throw error}} private guard(){if(this.closed||this.live?.controller.signal.aborted)throw new HttpError(410,'电脑任务授权已结束','computer_cancelled')} private event(type:string,payload:Record){if(!this.closed&&this.live)this.onEvent({type,session_id:this.live.session.id,payload})} private spec(session:WorkerSession):ComputerSpecification{return {id:this.meta.environmentId,ownerKey:this.meta.ownerKey,imageId:this.runtime.imageFor(this.meta),cwd:session.cwd,network:this.runtime.config.network??'none'}} async rpc(method:string,params:Record):Promise{ if((method==='session.interrupt'||method==='session.close')&¶ms.session_id===this.live?.session.id){await this.interrupt();return {status:'interrupted'}} if((method==='session.interrupt'||method==='session.close')&¶ms.session_id===this.recoverySession?.id){await this.runtime.pool.stopHolder(this.meta.ownerKey,this.meta.environmentId,this.workId);return {status:'interrupted'}} this.guard() if(method==='session.create'||method==='session.resume'){ if(this.live)throw new HttpError(409,'电脑通道已有会话','computer_session_busy') await this.authorized() if(method==='session.resume'&¶ms.recoverOnly===true){const session=this.runtime.session(String(params.session_id),this.meta,String(params.profile));this.recoverySession=session;if(this.meta.profileSession){await this.openProfile(session,params);await this.profileSession!.stop()}const resource=this.runtime.pool.status(this.meta.ownerKey).find(item=>item.environmentId===this.meta.environmentId);if(resource?.holderId===this.workId)await this.runtime.pool.stopHolder(this.meta.ownerKey,this.meta.environmentId,this.workId);return {session_id:session.id,stored_session_id:session.id,running:false,info:{profile_name:session.profile}}} if(this.runtime.imageFor(this.meta)===UNCONFIGURED_COMPUTER_IMAGE)throw new HttpError(409,'请先在应用设置的本地虚拟机页面完成准备','computer_image_required') const profile=String(params.profile),resolved=await (this.meta.profileSession?this.runtime.resolveWorkspace(profile,this.controller.signal):this.runtime.resolve(profile,this.controller.signal)) if(this.meta.hostAccess){const configured=String(resolved.configuredCwd??'.');this.hostCwd=['.','auto','cwd'].includes(configured)?homedir():hostPath(homedir(),configured)} await this.authorized();this.guard() const session:WorkerSession=method==='session.resume'?this.runtime.session(String(params.session_id),this.meta,profile):{id:randomUUID(),agentId:this.meta.agentId,ownerKey:this.meta.ownerKey,environmentId:this.meta.environmentId,profile,cwd:resolved.cwd,configuredCwd:resolved.configuredCwd,history:[]} if(this.meta.profileSession)session.vmExecution='profile' session.cwd=this.meta.environmentId!==this.meta.agentId?(this.runtime.pool.definition(this.meta.ownerKey,this.meta.environmentId)?.cwd??COMPUTER_WORKSPACE):resolved.cwd;session.configuredCwd=resolved.configuredCwd await this.runtime.pool.configure(this.spec(session),()=>this.guard()) await this.authorized() const controller=new AbortController(),lease=await this.runtime.pool.acquire(this.spec(session),this.workId,()=>{this.guard();if(controller.signal.aborted)throw new Error('cancelled')},controller.signal) if(this.closed){controller.abort();await this.runtime.pool.release(lease);throw new Error('电脑通道已关闭')} this.live={session,lease,controller,model:resolved.model as WorkerModel,contextConfig:resolved.contextConfig,proxyEnv:resolved.proxyEnv,images:[],running:false,received:false,toolCalls:new Set(),journal:new Map()} this.runtime.save(session);this.runtime.gateways.set(this.meta.environmentId,this) if(this.runtime.config.network==='public-proxy'){ const live=this.live try{live.proxy=await ComputerPublicProxy.start(this.runtime.provider,this.spec(session),this.runtime.proxyScript,()=>this.runtime.pool.authorize(live.lease),()=>this.authorized(),()=>{void this.interrupt().finally(()=>{if(!this.closed)this.onDisconnect()}).catch(()=>{})})} catch(error){await this.interrupt();throw error} } this.timer=setInterval(()=>{if(this.checking)return;this.checking=true;void this.authorized().then(()=>this.runtime.pool.renew(this.live!.lease)).catch(async()=>{await this.interrupt().catch(()=>{});if(!this.closed)this.onDisconnect()}).finally(()=>{this.checking=false})},5000);this.timer.unref() if(this.meta.profileSession){ try{const opened=await this.openProfile(session,params);return {...opened,session_id:session.id,stored_session_id:session.id}} catch(error){await this.interrupt();throw error} } return {session_id:session.id,stored_session_id:session.id,running:false,info:{profile_name:profile}} } const live=this.live if(!live||params.session_id!==live.session.id)throw new HttpError(403,'电脑会话不属于当前通道','computer_session_forbidden') if(method==='session.interrupt'||method==='session.close'){await this.interrupt();return {status:'interrupted'}} if(this.profileSession){ if(method==='session.usage')return this.profileSession.rpc(method,params) if(live.paused)throw new HttpError(409,'电脑正在等待人工操作','computer_paused') await this.authorized() if(method==='prompt.submit'){ if(live.running||live.received)throw new HttpError(409,'电脑会话已经提交过本轮','computer_session_busy') if(typeof params.workMarker==='string'&&/^\[yaoyao-run:[0-9a-f-]{36}:[0-9a-f-]{36}\]$/.test(params.workMarker))live.workMarker=params.workMarker await this.profileSession.bind();this.guard() live.running=true;live.session.outcome='uncertain';this.runtime.save(live.session) return this.profileSession.rpc(method,{...params,text:`${PROFILE_COMPUTER_RULES}\n${String(params.text??'')}`}) } return this.profileSession.rpc(method,params) } if(method==='session.usage')return {} if(method==='session.cwd.set'){ if(params.cwd!==live.session.cwd&¶ms.cwd!==live.session.configuredCwd)throw new HttpError(403,'工作目录必须采用基础 Profile 的配置','computer_cwd_forbidden') return {cwd:live.session.cwd} } if(method==='file.attach'||method==='image.attach_bytes'){ if(live.running)throw new HttpError(409,'当前轮次已开始,不能追加附件','computer_session_busy') await this.authorized() const name=String(params.name??params.filename??'attachment').replace(/[\/\\\u0000-\u001f]/g,'_').slice(0,200)||'attachment' const encoded=method==='image.attach_bytes'?String(params.content_base64??''):String(params.data_url??'').split(',')[1]??'' const bytes=Buffer.from(encoded,'base64');if(bytes.length>25*1024*1024)throw new HttpError(413,'附件超过 25 MiB','attachment_too_large') const relative=`.yaoyao-inputs/${randomUUID()}/${name}` await this.runtime.pool.use(live.lease,context=>this.runtime.provider.execute(this.spec(live.session),['python3','-c','import sys,pathlib; p=pathlib.Path(sys.argv[1]); p.parent.mkdir(parents=True,exist_ok=True); p.write_bytes(sys.stdin.buffer.read())',relative],{...context,input:bytes})) if(method==='image.attach_bytes'){ const mime=String(params.mime_type??'image/png') if(!/^image\/(?:png|jpeg|webp|gif)$/.test(mime))throw new HttpError(400,'图片格式无效','invalid_image') if(JSON.stringify(live.images).length+encoded.length>64*1024*1024)throw new HttpError(413,'本轮图片总量超过限制','image_limit') live.images.push({type:'image_url',image_url:{url:`data:${mime};base64,${encoded}`}}) } return {ref_text:`[附件 ${name}](${live.session.cwd}/${relative})`} } if(method==='prompt.submit'){ if(live.running||live.received)throw new HttpError(409,'电脑会话已经提交过本轮','computer_session_busy') await this.authorized();this.guard() if(typeof params.workMarker==='string'&&/^\[yaoyao-run:[0-9a-f-]{36}:[0-9a-f-]{36}\]$/.test(params.workMarker))live.workMarker=params.workMarker await this.startModel(live,String(params.text??''),structuredClone(live.session.history)) return {status:'streaming'} } throw new HttpError(409,'当前隔离 Worker 不支持此交互','computer_method_unavailable') } private async openProfile(session:WorkerSession,params:Record){ if(!this.profileTarget)throw new HttpError(409,'缺少本机 Profile 连接,请更新执行节点','computer_profile_unavailable') if(params.session_id&&!session.profileSessionId)throw new HttpError(409,'本机 Profile 会话尚未建立,请新建会话','computer_profile_session_invalid') const native=new ProfileComputerSession(this.profileTarget,session.profile,this.workId,()=>this.guard(), ()=>[...TOOLS,...HOST_TOOLS.filter(tool=>tool.name==='computer_copy_file'),...(this.team?.catalog??[])], async(name,args)=>{ const live=this.live if(!live||!this.active||live.paused)throw new HttpError(410,'本轮电脑工具已结束或暂停','computer_cancelled') const call=this.tool(live,name,args,randomUUID()) live.toolCalls.add(call);void call.finally(()=>live.toolCalls.delete(call)).catch(()=>{}) return call }) this.profileSession=native native.onStoredId=id=>{session.profileSessionId=id;this.runtime.save(session)} native.onDisconnect=()=>{void this.interrupt().finally(()=>{if(!this.closed)this.onDisconnect()}).catch(()=>{})} native.onEvent=frame=>{ const live=this.live if(!live||this.closed||live.controller.signal.aborted||live.paused||live.received)return if(['message.complete','run.completed','run.failed','message.error','error'].includes(frame.type)){ live.received=true live.finishing=this.finishProfile(live,frame) void live.finishing.catch(()=>this.lost(live)) }else this.event(frame.type,frame.type==='session.info'?{...frame.payload,stored_session_id:session.id,session_key:session.id}:frame.payload??{}) } return native.open(session.profileSessionId,params) } private async finishProfile(live:Live,frame:GatewayFrame){ await Promise.allSettled([...live.toolCalls]);await this.authorized() const payload=frame.payload??{},success=['message.complete','run.completed'].includes(frame.type)&&!payload.error&&!['interrupted','failed','error'].includes(String(payload.status)) live.session.outcome=success?'complete':'failed';this.runtime.save(live.session) await this.release(live,this.runtime.retainDesktops&&success);live.running=false if(this.closed||live.controller.signal.aborted)return this.event(frame.type,{...payload,...(payload.status==='error'?{status:'failed'}:{})}) } private async startModel(live:Live,prompt:string,history:any[]){ live.received=false;live.running=true;live.session.outcome='uncertain';live.session.history.push({role:'user',content:prompt});this.runtime.save(live.session) const directory=join(this.runtime.home,'workers',live.session.id,this.workId,String(live.segment=(live.segment??0)+1)) await mkdir(directory,{recursive:true,mode:0o700}) const worker=new HermesWorkerProcess(this.runtime.config.python,this.runtime.script,{mode:'run',home:directory,hermesSource:this.runtime.config.hermesSource,model:live.model,contextConfig:live.contextConfig,proxyEnv:live.proxyEnv,sessionId:live.session.id,taskId:this.workId,cwd:live.session.cwd,...(this.meta.hostAccess?{host:{cwd:this.hostCwd,platform:platform()}}:{}),network:this.runtime.config.network??'none',tools:[...TOOLS,...(this.meta.hostAccess?HOST_TOOLS:[]),...(this.team?.catalog??[])],prompt:live.images.length?[{type:'text',text:prompt},...live.images]:prompt,history}) live.worker=worker this.runtime.workers.set(this.workId,{environmentId:this.meta.environmentId,ownerKey:this.meta.ownerKey,process:worker});void worker.exited.then(()=>{if(this.runtime.workers.get(this.workId)?.process===worker)this.runtime.workers.delete(this.workId)}) worker.onTool=(name,args,id,callId)=>{ if(live.worker!==worker||live.received)throw new HttpError(410,'本轮 Worker 已结束','computer_worker_finished') const task=this.tool(live,name,args,id).then(result=>{if(callId)live.journal.set(callId,{name,result});return result},error=>{if(callId)live.journal.set(callId,{name,result:{error:'操作未完成或已中断'}});throw error}) live.toolCalls.add(task);void task.finally(()=>live.toolCalls.delete(task)).catch(()=>{});return task } worker.onEvent=frame=>{ if(this.closed||live.worker!==worker||live.received||live.controller.signal.aborted)return if(frame.type==='checkpoint'&&Array.isArray(frame.messages)){live.session.history=frame.messages;this.runtime.save(live.session);return} if(live.paused)return if(frame.type==='delta')this.event('message.delta',{text:frame.text}) if(frame.type==='reasoning')this.event('reasoning.delta',{text:frame.text}) if(frame.type==='complete'){live.received=true;live.finishing=this.finish(live,frame);void live.finishing.catch(()=>this.lost(live))} if(frame.type==='failed')void this.lost(live) } void worker.exited.then(()=>{if(!live.paused&&!live.received&&live.running&&!this.closed)void this.lost(live)}) } installTeamLease(id:string,catalog:WorkerTool[],call:(name:string,args:Record,callId:string)=>Promise){this.team={id,catalog,call}} removeTeamLease(id:string){if(this.team?.id===id)this.team=undefined} private async tool(live:Live,name:string,args:unknown,id:string):Promise{ if(live.paused)throw new HttpError(409,'电脑正在等待人工操作','computer_paused') await this.authorized();this.guard() if(live.paused)throw new HttpError(409,'电脑正在等待人工操作','computer_paused') this.event('tool.started',{tool_id:id,name}) try{ const team=this.team?.catalog.find(tool=>tool.name===name) if(team)return await this.team!.call(team.id??name,z.record(z.string(),z.unknown()).parse(args),id) return await this.runtime.pool.use(live.lease,async context=>{ const execute=(argv:string[],input?:Buffer)=>this.runtime.provider.execute(this.spec(live.session),argv,{...context,input}) if(this.meta.hostAccess&&name.startsWith('host_'))return hostTool(name,args,{...context,cwd:this.hostCwd}) if(this.meta.hostAccess&&name==='computer_copy_file'){ const body=z.object({from:z.enum(['host','vm']),source:pathField,destination:pathField}).strict().parse(args),host={...context,cwd:this.hostCwd} if(body.from==='host'){ const bytes=await readHostFile(hostPath(this.hostCwd,body.source),25*1024*1024,host) await execute(['python3','-c','import pathlib,sys; p=pathlib.Path(sys.argv[1]); p.parent.mkdir(parents=True,exist_ok=True); p.write_bytes(sys.stdin.buffer.read())',body.destination],bytes) }else{ const snapshot=`/tmp/yaoyao-copy-${randomUUID()}`,chunks:Buffer[]=[] try{ const size=Number((await execute(['python3','-c','import pathlib,sys,shutil; p=pathlib.Path(sys.argv[1]); assert p.is_file() and p.stat().st_size<=25*1024*1024; shutil.copyfile(p,sys.argv[2]); n=pathlib.Path(sys.argv[2]).stat().st_size; assert n<=25*1024*1024; print(n)',body.source,snapshot])).stdout.trim()) if(!Number.isSafeInteger(size)||size<0||size>25*1024*1024)throw new Error('文件大小无效') for(let offset=0;offset{})} } return {ok:true} } if(name==='computer_export'){ const body=z.object({path:pathField,name:z.string().min(1).max(240).optional()}).strict().parse(args) const snapshot=`/tmp/yaoyao-export-${randomUUID()}` try{ const metadata=JSON.parse((await execute(['python3','-c','import pathlib,sys,shutil,json,hashlib; p=pathlib.Path(sys.argv[1]); assert p.is_file(),"not a file"; assert p.stat().st_size<=25*1024*1024,"file too large"; shutil.copyfile(p,sys.argv[2]); b=pathlib.Path(sys.argv[2]).read_bytes(); assert len(b)<=25*1024*1024,"file too large"; print(json.dumps({"size":len(b),"digest":hashlib.sha256(b).hexdigest()}))',body.path,snapshot])).stdout) await this.publish({phase:'begin',id,name:body.name??posix.basename(body.path),size:metadata.size,digest:metadata.digest}) for(let offset=0,index=0;offset{})} } if(name==='computer_shell'){ const body=z.object({command:z.string().min(1).max(65536)}).strict().parse(args) try{return {exitCode:0,...await execute(['/bin/bash','-lc',body.command])}}catch(error){if(typeof(error as any).code==='number')return {exitCode:(error as any).code,stdout:String((error as any).stdout??''),stderr:String((error as any).stderr??'')};throw error} } if(name==='computer_read_file'){ const body=z.object({path:pathField}).strict().parse(args) return execute(['python3','-c','import sys; f=open(sys.argv[1],encoding="utf-8"); s=f.read(1000001); assert len(s)<=1000000,"file too large"; print(s,end="")',body.path]) } if(name==='computer_write_file'){ const body=z.object({path:pathField,content:z.string().max(1000000)}).strict().parse(args) await execute(['python3','-c','import pathlib,sys; p=pathlib.Path(sys.argv[1]); p.parent.mkdir(parents=True,exist_ok=True); p.write_bytes(sys.stdin.buffer.read())',body.path],Buffer.from(body.content));return {ok:true} } if(name==='computer_action'){ const body=z.object({name:z.string().regex(/^[a-zA-Z][a-zA-Z0-9_]{0,99}$/),arguments:z.record(z.string(),z.unknown())}).strict().parse(args) return execute([CUA_DRIVER,'call',body.name,JSON.stringify(body.arguments),'--socket',CUA_SOCKET]) } if(name==='computer_desktop_state'){ z.object({}).strict().parse(args) const path=`/tmp/yaoyao-frame-${randomUUID()}.png` const state=await execute([CUA_DRIVER,'call','get_desktop_state','{}','--socket',CUA_SOCKET,'--screenshot-out-file',path]) const data=(await execute(['base64','-w0',path])).stdout.trim() return {_multimodal:true,text_summary:state.stdout,content:[{type:'image_url',image_url:{url:`data:image/png;base64,${data}`}}]} } throw new HttpError(403,'本轮未授权该工具','computer_tool_forbidden') }) }finally{this.event('tool.completed',{tool_id:id,name})} } private async finish(live:Live,frame:WorkerFrame){ if(this.closed||live.paused||live.controller.signal.aborted)return await this.authorized() this.runtime.pool.renew(live.lease) const contextErrors:Record={ context_compaction_disabled:'模型上下文已超限,基础 Profile 关闭了自动压缩。请开启压缩、缩短输入或新建对话后继续。', context_compaction_failed:'模型上下文压缩未能完成,本轮已结束。请缩短输入、检查模型配置或新建对话后继续。', } const contextError=frame.completed===false&&!frame.interrupted&&Object.hasOwn(contextErrors,String(frame.failureCode)) ?contextErrors[String(frame.failureCode)]:undefined const success=frame.completed!==false&&!frame.interrupted // Failed model turns can carry a newly compacted, authoritative history too. // A missing history leaves the last durable checkpoint intact. if(Array.isArray(frame.messages))live.session.history=frame.messages live.session.outcome=success?'complete':'failed' this.runtime.save(live.session) const keepRunning=this.runtime.retainDesktops&&(success||!!contextError)&&!live.toolCalls.size await live.worker?.close() await this.release(live,keepRunning) live.running=false if(this.closed||live.controller.signal.aborted)return const error=contextError?contextError+(keepRunning?' 虚拟机已进入空闲,后续按空闲停止设置处理。':''):'隔离 Worker 未完成本轮' this.event('message.complete',{text:contextError?error:String(frame.text??''),status:success?'complete':'failed',...(!success?{error,...(contextError?{code:frame.failureCode}:{})}:{})}) } private release(live:Live,keepRunning=false){ clearInterval(this.timer) live.releasing??=(async()=>{ try{await live.proxy?.close();await this.runtime.pool.release(live.lease,keepRunning)} catch(error){await this.runtime.pool.release(live.lease).catch(()=>{});throw error} })() return live.releasing } private async lost(live:Live){ if(!live.running||live.paused||this.closed||live.controller.signal.aborted)return live.running=false;await this.interrupt().catch(()=>{});if(!this.closed)this.onDisconnect() } get active(){return !!this.live?.running&&!this.live.received&&!this.closed} get paused(){return this.live?.paused===true} get lease(){return this.live?.lease} get specification(){return this.live?this.spec(this.live.session):undefined} async takeControl(controlId:string,authorize:()=>void):Promise{ const live=this.live if(!live||!this.active)throw new HttpError(409,'机器人已不在执行,请重新打开电脑','computer_not_active') if(live.pauseJob)return live.pauseJob live.paused=true;this.event('computer.paused',{stage:'pausing'}) live.pauseJob=(async()=>{ await this.profileSession?.stop() await live.worker?.close() await Promise.allSettled([...live.toolCalls]) authorize();this.guard() const history=structuredClone(live.session.history),done=new Set(history.filter(message=>message.role==='tool').map(message=>message.tool_call_id)) for(const message of [...history])for(const tool of message.tool_calls??[])if(!done.has(tool.id)){ const entry=live.journal.get(tool.id),value=entry?.result??{error:'用户接管时此操作未完成,继续前请检查电脑状态,不要盲目重复'} const content=typeof value==='string'?value:JSON.stringify(value) history.push({role:'tool',name:tool.function?.name??entry?.name,tool_call_id:tool.id,content});done.add(tool.id) } live.session.history=history;this.runtime.save(live.session) live.lease=await this.runtime.pool.transfer(live.lease,`human:${controlId}`,authorize) this.event('computer.paused',{stage:'human',generation:live.lease.generation}) return live.lease })() return live.pauseJob } async giveBack(notes:string):Promise{ const live=this.live if(!live?.paused)throw new HttpError(409,'机器人没有处于接管状态','computer_not_paused') await live.pauseJob;await this.authorized() live.lease=await this.runtime.pool.transfer(live.lease,this.workId,()=>this.guard()) live.paused=false;live.pauseJob=undefined;live.journal.clear() this.event('computer.resumed',{generation:live.lease.generation,notes}) if(this.profileSession){ await this.profileSession.bind();this.guard() await this.profileSession.rpc('prompt.submit',{text:`${PROFILE_COMPUTER_RULES}\n用户已交还电脑控制权。请先检查当前状态,避免重复已完成的操作。交还说明:${notes||'无额外说明'}\n${live.workMarker??''}`}) return } await this.startModel(live,`用户已完成电脑接管并明确交还控制。请先检查当前电脑状态,再继续原任务;不要重复已完成的副作用。用户说明:${notes||'无额外说明'}\n${live.workMarker??''}`,structuredClone(live.session.history)) } async interrupt(){ const live=this.live;if(!live)return if(live.stopping)return live.stopping live.stopping=(async()=>{ live.controller.abort();clearInterval(this.timer) try{await this.profileSession?.stop()} finally{ await this.profileSession?.close();await live.worker?.close() try{await this.release(live)}catch(error){if((error as any).code!=='computer_lease_stale')throw error} live.running=false } })() return live.stopping } async stopControl(){await this.interrupt();this.event('message.complete',{text:'电脑控制已结束,本轮已停止',status:'interrupted'})} async close(){if(this.runtime.gateways.get(this.meta.environmentId)===this)this.runtime.gateways.delete(this.meta.environmentId);this.closed=true;this.controller.abort();this.team=undefined;clearInterval(this.timer);try{if(this.live?.running||this.live&&!this.live.releasing)await this.interrupt()}finally{await this.profileSession?.close()}} }