|
@@ -2,6 +2,7 @@ import {randomUUID} from 'node:crypto'
|
|
|
import {join,dirname,posix} from 'node:path'
|
|
import {join,dirname,posix} from 'node:path'
|
|
|
import {fileURLToPath} from 'node:url'
|
|
import {fileURLToPath} from 'node:url'
|
|
|
import {mkdir} from 'node:fs/promises'
|
|
import {mkdir} from 'node:fs/promises'
|
|
|
|
|
+import {existsSync} from 'node:fs'
|
|
|
import type {DatabaseSync} from 'node:sqlite'
|
|
import type {DatabaseSync} from 'node:sqlite'
|
|
|
import {z} from 'zod'
|
|
import {z} from 'zod'
|
|
|
import {HttpError} from '../../server/errors.js'
|
|
import {HttpError} from '../../server/errors.js'
|
|
@@ -9,11 +10,12 @@ import type {GatewayFrame} from '../../server/workspaceGateway.js'
|
|
|
import type {RunnerConfiguration} from '../../shared/runner.js'
|
|
import type {RunnerConfiguration} from '../../shared/runner.js'
|
|
|
import {ContainerComputerProvider,CUA_DRIVER,CUA_SOCKET,type ComputerSpecification} from '../computers/container.js'
|
|
import {ContainerComputerProvider,CUA_DRIVER,CUA_SOCKET,type ComputerSpecification} from '../computers/container.js'
|
|
|
import {ComputerPool,type ComputerLease} from '../computers/pool.js'
|
|
import {ComputerPool,type ComputerLease} from '../computers/pool.js'
|
|
|
|
|
+import {ComputerPublicProxy} from '../network/computerProxy.js'
|
|
|
import {HermesWorkerProcess,type WorkerModel,type WorkerTool,type WorkerFrame} from './process.js'
|
|
import {HermesWorkerProcess,type WorkerModel,type WorkerTool,type WorkerFrame} from './process.js'
|
|
|
|
|
|
|
|
export interface ComputerTarget {environmentId:string;ownerKey:string;agentId:string}
|
|
export interface ComputerTarget {environmentId:string;ownerKey:string;agentId:string}
|
|
|
export interface WorkerSession {id:string;profile:string;ownerKey:string;environmentId:string;cwd:string;configuredCwd:string;history:any[];outcome?:'complete'|'failed'|'uncertain'}
|
|
export interface WorkerSession {id:string;profile:string;ownerKey:string;environmentId:string;cwd:string;configuredCwd:string;history:any[];outcome?:'complete'|'failed'|'uncertain'}
|
|
|
-interface Live {session:WorkerSession;model:WorkerModel;lease:ComputerLease;controller:AbortController;worker?:HermesWorkerProcess;finishing?:Promise<void>;stopping?:Promise<void>;releasing?:Promise<void>;images:any[];running:boolean;received:boolean}
|
|
|
|
|
|
|
+interface Live {proxy?:ComputerPublicProxy;session:WorkerSession;model:WorkerModel;lease:ComputerLease;controller:AbortController;worker?:HermesWorkerProcess;finishing?:Promise<void>;stopping?:Promise<void>;releasing?:Promise<void>;images:any[];running:boolean;received:boolean}
|
|
|
const object=(properties:Record<string,unknown>,required:string[])=>({type:'object',properties,required,additionalProperties:false})
|
|
const object=(properties:Record<string,unknown>,required:string[])=>({type:'object',properties,required,additionalProperties:false})
|
|
|
const text={type:'string'}
|
|
const text={type:'string'}
|
|
|
const TOOLS:WorkerTool[]=[
|
|
const TOOLS:WorkerTool[]=[
|
|
@@ -31,10 +33,12 @@ export class ComputerRuntime {
|
|
|
readonly ready:Promise<void>
|
|
readonly ready:Promise<void>
|
|
|
readonly config:NonNullable<RunnerConfiguration['computers']>
|
|
readonly config:NonNullable<RunnerConfiguration['computers']>
|
|
|
readonly script:string
|
|
readonly script:string
|
|
|
|
|
+ readonly proxyScript:string
|
|
|
constructor(readonly db:DatabaseSync,config:RunnerConfiguration,readonly home:string,script?:string){
|
|
constructor(readonly db:DatabaseSync,config:RunnerConfiguration,readonly home:string,script?:string){
|
|
|
if(!config.computers)throw new HttpError(409,'执行节点尚未配置隔离电脑','computer_unavailable')
|
|
if(!config.computers)throw new HttpError(409,'执行节点尚未配置隔离电脑','computer_unavailable')
|
|
|
this.config=config.computers
|
|
this.config=config.computers
|
|
|
this.script=script??join(dirname(fileURLToPath(import.meta.url)),'hermes_worker.py')
|
|
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')
|
|
|
this.provider=new ContainerComputerProvider(this.config.runtime,config.runnerId,home)
|
|
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})
|
|
this.pool=new ComputerPool(db,this.provider,{concurrent:this.config.maxConcurrent??2,environments:32,ttlMs:30000})
|
|
|
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_sessions(id TEXT PRIMARY KEY,value TEXT NOT NULL)')
|
|
@@ -68,7 +72,7 @@ export class ComputerGateway {
|
|
|
private async authorized(){try{await this.check();this.guard()}catch(error){await this.interrupt().catch(()=>{});if(!this.closed)this.onDisconnect();throw error}}
|
|
private async authorized(){try{await this.check();this.guard()}catch(error){await this.interrupt().catch(()=>{});if(!this.closed)this.onDisconnect();throw error}}
|
|
|
private guard(){if(this.closed||this.live?.controller.signal.aborted)throw new HttpError(410,'电脑任务授权已结束','computer_cancelled')}
|
|
private guard(){if(this.closed||this.live?.controller.signal.aborted)throw new HttpError(410,'电脑任务授权已结束','computer_cancelled')}
|
|
|
private event(type:string,payload:Record<string,unknown>){if(!this.closed&&this.live)this.onEvent({type,session_id:this.live.session.id,payload})}
|
|
private event(type:string,payload:Record<string,unknown>){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.config.imageId,cwd:session.cwd}}
|
|
|
|
|
|
|
+ private spec(session:WorkerSession):ComputerSpecification{return {id:this.meta.environmentId,ownerKey:this.meta.ownerKey,imageId:this.runtime.config.imageId,cwd:session.cwd,network:this.runtime.config.network??'none'}}
|
|
|
async rpc(method:string,params:Record<string,any>):Promise<any>{
|
|
async rpc(method:string,params:Record<string,any>):Promise<any>{
|
|
|
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.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'}}
|
|
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'}}
|
|
@@ -87,7 +91,12 @@ export class ComputerGateway {
|
|
|
if(this.closed){controller.abort();await this.runtime.pool.release(lease);throw new Error('电脑通道已关闭')}
|
|
if(this.closed){controller.abort();await this.runtime.pool.release(lease);throw new Error('电脑通道已关闭')}
|
|
|
this.live={session,lease,controller,model:resolved.model as WorkerModel,images:[],running:false,received:false}
|
|
this.live={session,lease,controller,model:resolved.model as WorkerModel,images:[],running:false,received:false}
|
|
|
this.runtime.save(session)
|
|
this.runtime.save(session)
|
|
|
- this.timer=setInterval(()=>{if(this.checking)return;this.checking=true;void this.authorized().then(()=>this.runtime.pool.renew(lease)).catch(()=>this.interrupt()).finally(()=>{this.checking=false})},5000);this.timer.unref()
|
|
|
|
|
|
|
+ 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(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(lease)).catch(async()=>{await this.interrupt().catch(()=>{});if(!this.closed)this.onDisconnect()}).finally(()=>{this.checking=false})},5000);this.timer.unref()
|
|
|
return {session_id:session.id,stored_session_id:session.id,running:false,info:{profile_name:profile}}
|
|
return {session_id:session.id,stored_session_id:session.id,running:false,info:{profile_name:profile}}
|
|
|
}
|
|
}
|
|
|
const live=this.live
|
|
const live=this.live
|
|
@@ -121,7 +130,7 @@ export class ComputerGateway {
|
|
|
live.running=true;live.session.outcome='uncertain';live.session.history.push({role:'user',content:prompt});this.runtime.save(live.session)
|
|
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)
|
|
const directory=join(this.runtime.home,'workers',live.session.id,this.workId)
|
|
|
await mkdir(directory,{recursive:true,mode:0o700})
|
|
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,sessionId:live.session.id,taskId:this.workId,cwd:live.session.cwd,tools:[...TOOLS,...(this.team?.catalog??[])],prompt:live.images.length?[{type:'text',text:prompt},...live.images]:prompt,history})
|
|
|
|
|
|
|
+ const worker=new HermesWorkerProcess(this.runtime.config.python,this.runtime.script,{mode:'run',home:directory,hermesSource:this.runtime.config.hermesSource,model:live.model,sessionId:live.session.id,taskId:this.workId,cwd:live.session.cwd,network:this.runtime.config.network??'none',tools:[...TOOLS,...(this.team?.catalog??[])],prompt:live.images.length?[{type:'text',text:prompt},...live.images]:prompt,history})
|
|
|
live.worker=worker
|
|
live.worker=worker
|
|
|
worker.onTool=(name,args,id)=>this.tool(live,name,args,id)
|
|
worker.onTool=(name,args,id)=>this.tool(live,name,args,id)
|
|
|
worker.onEvent=frame=>{
|
|
worker.onEvent=frame=>{
|
|
@@ -199,7 +208,7 @@ export class ComputerGateway {
|
|
|
if(this.closed||live.controller.signal.aborted)return
|
|
if(this.closed||live.controller.signal.aborted)return
|
|
|
this.event('message.complete',{text:String(frame.text??''),status:success?'complete':'failed',...(!success?{error:'隔离 Worker 未完成本轮'}:{})})
|
|
this.event('message.complete',{text:String(frame.text??''),status:success?'complete':'failed',...(!success?{error:'隔离 Worker 未完成本轮'}:{})})
|
|
|
}
|
|
}
|
|
|
- private release(live:Live){clearInterval(this.timer);live.releasing??=this.runtime.pool.release(live.lease);return live.releasing}
|
|
|
|
|
|
|
+ private release(live:Live){clearInterval(this.timer);live.releasing??=(async()=>{await live.proxy?.close();await this.runtime.pool.release(live.lease)})();return live.releasing}
|
|
|
private async lost(live:Live){
|
|
private async lost(live:Live){
|
|
|
if(!live.running||this.closed||live.controller.signal.aborted)return
|
|
if(!live.running||this.closed||live.controller.signal.aborted)return
|
|
|
live.running=false;await this.interrupt().catch(()=>{});if(!this.closed)this.onDisconnect()
|
|
live.running=false;await this.interrupt().catch(()=>{});if(!this.closed)this.onDisconnect()
|