agent.ts 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306
  1. import { randomUUID, createHash } from 'node:crypto'
  2. import { DatabaseSync } from 'node:sqlite'
  3. import { realpath,access } from 'node:fs/promises'
  4. import { resolve, sep, join } from 'node:path'
  5. import {LocalVmImages} from './computers/localVm.js'
  6. import {ComputerError} from './computers/container.js'
  7. import {z} from 'zod'
  8. import {ComputerRuntime,ComputerGateway,type ComputerTarget} from './worker/gateway.js'
  9. import { HttpError } from '../server/errors.js'
  10. import { UpstreamClient } from '../server/upstream.js'
  11. import { UpstreamServiceSession } from '../server/localAuth.js'
  12. import { WorkspaceGateway, type GatewayTarget, type GatewayFrame } from '../server/workspaceGateway.js'
  13. import { createWorkspaceToolLease, type WorkspaceToolLease } from '../server/workspaceToolLease.js'
  14. import {UNCONFIGURED_COMPUTER_IMAGE} from '../shared/runner.js'
  15. import type { RunnerCommand, RunnerConfiguration } from '../shared/runner.js'
  16. import { LoopbackTransport, isLocalAuthorizationTarget } from '../server/loopbackAuthorization.js'
  17. const commands=new Set(['session.create','session.resume','session.close','session.usage','session.interrupt','session.cwd.set','prompt.submit','session.steer','image.attach_bytes','file.attach','approval.respond','clarify.respond'])
  18. interface Connection {cleanupOnly?:boolean;gateway:WorkspaceGateway|ComputerGateway;sessions:Map<string,string>;running:Set<string>;events:Promise<void>}
  19. interface Lease {connectionId:string;profile:string;lease:WorkspaceToolLease;controller:AbortController;session:{runtimeId:string;storedId:string}}
  20. export class RunnerAgent {
  21. readonly instance=randomUUID()
  22. private active=true
  23. private connected=false
  24. private disconnectRevision=0
  25. private disconnecting?:Promise<void>
  26. private connections=new Map<string,Connection>()
  27. private leases=new Map<string,Lease>()
  28. private closedConnections=new Map<string,number>()
  29. private closedLeases=new Map<string,number>()
  30. private inflight=new Map<string,Promise<unknown>>()
  31. private db:DatabaseSync
  32. private computers?:ComputerRuntime
  33. private localVm?:LocalVmImages
  34. private target:GatewayTarget
  35. private controlTransport=new LoopbackTransport()
  36. private serverEpoch?:string
  37. private controlAbort=new AbortController()
  38. constructor(readonly config:RunnerConfiguration,home:string,readonly fetchImpl:typeof fetch=fetch,readonly onStatus:(message:string)=>void=()=>{}) {
  39. const web=new URL(config.serverURL),hermes=new URL(config.hermesURL)
  40. if(web.username||web.password||web.search||web.hash||!['http:','https:'].includes(web.protocol))throw new Error('Web 地址无效')
  41. if(web.protocol==='http:'&&!['127.0.0.1','localhost','[::1]'].includes(web.hostname)&&!config.allowInsecureLan)throw new Error('远程执行节点必须使用 HTTPS,或明确配置可信局域网 HTTP')
  42. if(hermes.username||hermes.password||hermes.search||hermes.hash||!['127.0.0.1','[::1]'].includes(hermes.hostname)||!['http:','https:'].includes(hermes.protocol))throw new Error('Runner 必须连接本机 Hermes')
  43. const client=new UpstreamClient(hermes)
  44. this.target={url:hermes,client,session:new UpstreamServiceSession(client,()=>config.hermesCredentials)}
  45. this.db=new DatabaseSync(join(home,'runner-commands.sqlite3'))
  46. this.db.exec('CREATE TABLE IF NOT EXISTS commands(id TEXT PRIMARY KEY,fingerprint TEXT NOT NULL,state TEXT NOT NULL,result TEXT,created INTEGER NOT NULL)')
  47. if(config.computers){this.computers=new ComputerRuntime(this.db,config,home,undefined,(id,operation,body)=>this.api('desktop',{id,operation,...body}));this.localVm=new LocalVmImages(this.computers,this.db)}
  48. this.compactReceipts()
  49. }
  50. private async api(path:string,body?:unknown):Promise<any> {
  51. const url=new URL(`/api/runner/v1/${this.config.runnerId}/${path}`,this.config.serverURL)
  52. if(url.hostname==='localhost')url.hostname='127.0.0.1'
  53. const response=await (isLocalAuthorizationTarget(url)&&this.fetchImpl===fetch?this.controlTransport.fetch.bind(this.controlTransport):this.fetchImpl)(url,{
  54. method:body===undefined?'GET':'POST',redirect:'error',signal:AbortSignal.any([this.controlAbort.signal,AbortSignal.timeout(path==='desktop'?75000:25000)]),
  55. headers:{Authorization:`Bearer ${this.config.token}`,'x-runner-instance':this.instance,'x-runner-protocol':'1','x-runner-features':this.computers?'computer-worker-v1,artifact-chunks-v1,computer-control-v1,shared-computer-v1,local-vm-v1'+(this.computers.provider.fixedCapacity?',compose-desktops-v1':',helper-retirement-v1')+(this.computers.config.imageId!==UNCONFIGURED_COMPUTER_IMAGE?',image-ready-v1':''):'',...(this.serverEpoch?{'x-runner-epoch':this.serverEpoch}:{}),'Content-Type':'application/json',...(path==='poll'&&!this.connected?{'x-runner-reset':'1'}:{})},
  56. ...(body===undefined?{}:{body:JSON.stringify(body)}),
  57. })
  58. if(!response.ok){
  59. if(path==='desktop'){const error=await response.json().catch(()=>({})) as {error?:string;code?:string};throw new HttpError(response.status,error.error?.slice(0,300)||'Compose 桌面连接失败',error.code||'compose_desktop_error')}
  60. throw new HttpError(response.status,`执行节点连接被拒绝(HTTP ${response.status})`,response.status===403?'runner_unauthorized':'runner_request_failed')
  61. }
  62. const reader=response.body?.getReader();const chunks:Uint8Array[]=[];let bytes=0
  63. if(!reader)throw new Error('执行节点响应为空')
  64. try{for(;;){const part=await reader.read();if(part.done)break;bytes+=part.value.length;if(bytes>40*1024*1024)throw new Error('执行节点响应超过大小限制');chunks.push(part.value)}}finally{await reader.cancel()}
  65. return JSON.parse(Buffer.concat(chunks).toString())
  66. }
  67. private requireProfile(profile:unknown):string {
  68. if(typeof profile!=='string'||!this.config.allowedProfiles.includes(profile))throw new HttpError(403,'Profile 未授权','runner_profile_forbidden')
  69. return profile
  70. }
  71. private async closeConnection(id:string,interrupt=true) {
  72. this.closedConnections.set(id,Date.now()+60000)
  73. const connection=this.connections.get(id)
  74. if(!connection)return
  75. this.connections.delete(id)
  76. for(const [leaseId,lease] of this.leases)if(lease.connectionId===id){this.closedLeases.set(leaseId,Date.now()+60000);lease.controller.abort();this.leases.delete(leaseId);void lease.lease.dispose()}
  77. if(interrupt){let timer:ReturnType<typeof setTimeout>|undefined;try{await Promise.race([Promise.all([...connection.running].map(session_id=>connection.gateway.rpc('session.interrupt',{session_id}).catch(()=>{}))),new Promise(resolve=>{timer=setTimeout(resolve,3000)})])}finally{clearTimeout(timer)}}
  78. await connection.gateway.close()
  79. }
  80. private disconnect():Promise<void> {
  81. if(this.disconnecting)return this.disconnecting
  82. this.disconnectRevision++;this.connected=false
  83. const leases=[...this.leases.values()];this.leases.clear()
  84. for(const value of leases)value.controller.abort()
  85. const pending=Promise.all([
  86. ...leases.map(value=>value.lease.dispose()),
  87. ...[...this.connections.keys()].map(id=>this.closeConnection(id)),
  88. ]).then(()=>{}).finally(()=>{if(this.disconnecting===pending)this.disconnecting=undefined})
  89. this.disconnecting=pending;return pending
  90. }
  91. private async execute(command:RunnerCommand):Promise<any> {
  92. const p=command.payload
  93. if(!this.active||!this.connected)throw new Error('执行节点已断开')
  94. if(command.kind==='local-vm.manage'){
  95. if(!this.localVm)throw new HttpError(409,'本地虚拟机尚未配置','computer_unavailable')
  96. const input=z.object({owner:z.string().regex(/^[a-f0-9]{64}$/),op:z.enum(['status','prepare','policy','instance']),id:z.string().uuid().optional(),mode:z.enum(['shared','per-bot']).optional(),maxInstances:z.number().int().min(1).max(4).optional(),environmentId:z.string().uuid().optional(),action:z.enum(['stop','remove']).optional()}).strict().parse(p)
  97. const check=async()=>{if(!this.connected||!this.active||(await this.api('local-vm-check',{id:input.id})).allowed!==true)throw new ComputerError('local_vm_authorization_revoked','本地虚拟机配置授权已失效')}
  98. if(input.op==='status')return this.localVm.status(input.owner)
  99. if(input.op==='instance'){
  100. await check()
  101. const spec=this.computers!.pool.definition(input.owner,input.environmentId!)
  102. if(spec)await this.computers!.pool.desktop(spec,input.action!,()=>{if(!this.active||!this.connected)throw new Error('节点已断开')})
  103. await check();return {ok:true}
  104. }
  105. if(input.op==='prepare')return this.localVm.prepare(input.id!,check)
  106. await check();return this.localVm.policy(input.mode!,input.maxInstances!)
  107. }
  108. if(command.kind==='computer.control'){
  109. if(!this.computers)throw new HttpError(409,'电脑服务未配置','computer_unavailable')
  110. const input=z.object({target:z.object({environmentId:z.string().uuid(),agentId:z.string().uuid(),ownerKey:z.string().regex(/^[a-f0-9]{64}$/)}).strict(),profile:z.string(),op:z.enum(['status','detail','lifecycle','frame','take','input','giveback']),controlId:z.string().uuid().optional(),requestId:z.string().uuid().optional(),generation:z.number().int().optional(),frameId:z.string().uuid().optional(),action:z.unknown().optional(),notes:z.string().max(4000).optional()}).strict().parse(p)
  111. this.requireProfile(input.profile)
  112. const check=async()=>{if(!this.connected||!input.controlId||(await this.api('control-check',{controlId:input.controlId})).allowed!==true)throw new HttpError(403,'电脑控制授权已失效','computer_control_expired')}
  113. const authorize=()=>{if(!this.connected||!this.active||Date.now()>command.expiresAt)throw new HttpError(403,'电脑请求已失效','computer_control_expired')}
  114. await this.computers.ready
  115. if(input.op==='detail'){
  116. const vm=await this.localVm!.status(),spec=this.computers.pool.definition(input.target.ownerKey,input.target.environmentId)??(this.computers.provider.fixedCapacity?{id:input.target.environmentId,ownerKey:input.target.ownerKey,imageId:this.computers.config.imageId}:undefined)
  117. const state=await this.computers.controls.status(input.target)
  118. const actual=spec?await this.computers.provider.inspect(spec):undefined
  119. let ready=false
  120. if(actual?.running)try{await this.computers.provider.health(spec!,authorize);ready=true}catch{}
  121. return {...state,container:actual?.running?'running':spec?'stopped':'missing',ready,inUse:['agent','pausing','human','resuming'].includes(state.mode),mode:vm.mode,maxInstances:vm.maxInstances,image:vm.image,controlMode:state.mode,fixedCapacity:vm.fixedCapacity}
  122. }
  123. if(input.op==='lifecycle'){
  124. const action=z.enum(['create','start','stop','recreate','remove']).parse(input.action)
  125. await this.computers.desktop(input.target,input.profile,action,authorize)
  126. return {ok:true}
  127. }
  128. if(input.op==='status')return this.computers.controls.status(input.target)
  129. if(input.op==='frame')return this.computers.controls.frame(input.target,authorize)
  130. if(input.op==='take')return this.computers.controls.take(input.target,input.profile,input.controlId!,check)
  131. if(input.op==='input')return this.computers.controls.input(input.target,input.controlId!,input.requestId!,input.generation!,input.frameId!,input.action)
  132. return this.computers.controls.giveBack(input.target,input.controlId!,input.notes??'')
  133. }
  134. if(command.kind==='computer.retire'){
  135. const meta=p as unknown as ComputerTarget
  136. if(!this.computers||![meta.environmentId,meta.agentId].every(value=>/^[0-9a-f-]{36}$/.test(value))||!/^[a-f0-9]{64}$/.test(meta.ownerKey))throw new HttpError(400,'电脑清理请求无效','computer_cleanup_invalid')
  137. await this.computers.retire(meta);return {ok:true}
  138. }
  139. if(command.kind==='http') {
  140. const path=String(p.path),search=new URLSearchParams(String(p.search??''))
  141. if(path==='/api/computer/capabilities'||(p.computer&&path==='/api/plugins/yaoyao-bot-bridge/capabilities')){
  142. this.requireProfile(search.get('profile')??'default')
  143. if(p.method!=='GET'||!this.computers)throw new HttpError(409,'执行节点未配置隔离电脑','computer_unavailable')
  144. await Promise.all([access(this.computers.config.python),access(this.computers.config.hermesSource),access(this.computers.script)])
  145. return {status:200,headers:{},body:Buffer.from(JSON.stringify({version:1,ready:true,native_tools:true,in_process:true,network:this.config.computers?.network??'none'})).toString('base64')}
  146. }
  147. if(p.computer&&path==='/api/files/download')throw new HttpError(409,'隔离电脑产物需通过产物导出接口读取','computer_artifact_required')
  148. if(p.computer&&/^\/api\/sessions\/[^/]+\/messages$/.test(path)){
  149. if(p.method!=='GET'||!this.computers)throw new HttpError(403,'电脑历史不可用','computer_unavailable')
  150. const profile=this.requireProfile(search.get('profile')??'default'),session=this.computers.session(decodeURIComponent(path.split('/')[3]!),p.computer as ComputerTarget,profile)
  151. const offset=Math.max(0,Number(search.get('offset')??0)),history=session.history
  152. return {status:200,headers:{},body:Buffer.from(JSON.stringify({messages:history.slice(Math.max(0,history.length-offset-500),history.length-offset)})).toString('base64')}
  153. }
  154. if(p.method!=='GET'||!(/^\/api\/(?:status|profiles|config)$/.test(path)||/^\/api\/sessions\/[^/]+\/messages$/.test(path)||path==='/api/plugins/yaoyao-bot-bridge/capabilities'||path==='/api/files/download'))throw new HttpError(403,'Runner 不支持此 HTTP 操作','runner_http_forbidden')
  155. if(!['/api/status','/api/profiles'].includes(path))this.requireProfile(search.get('profile')??'default')
  156. if(path==='/api/files/download') {
  157. const requested=await realpath(String(search.get('path')??''))
  158. const roots=await Promise.all(this.config.artifactRoots.map(root=>realpath(root)))
  159. if(!roots.some(root=>requested===root||requested.startsWith(root+sep)))throw new HttpError(403,'文件不在允许的产物目录','runner_artifact_forbidden')
  160. }
  161. const response=await this.target.session.request(path,{search,maxResponseBytes:8*1024*1024})
  162. return {status:response.status,headers:Object.fromEntries(response.headers.entries()),body:response.body.toString('base64')}
  163. }
  164. const connectionId=String(p.connectionId??'')
  165. if(command.kind==='gateway.open') {
  166. if(this.closedConnections.has(connectionId)||this.connections.has(connectionId)||!/^[0-9a-f-]{36}$/.test(connectionId)||this.connections.size>=64)throw new Error('执行通道无效或已达上限')
  167. let gateway:WorkspaceGateway|ComputerGateway
  168. if(p.computer){
  169. const meta=p.computer as ComputerTarget
  170. if(!this.computers||![meta.environmentId,meta.agentId,String(p.workId)].every(value=>/^[0-9a-f-]{36}$/.test(value))||!/^[a-f0-9]{64}$/.test(meta.ownerKey))throw new HttpError(409,'电脑任务配置或授权无效','computer_unavailable')
  171. gateway=new ComputerGateway(this.computers,meta,String(p.workId),async()=>{if(!this.active||!this.connected||(await this.api('check',{connectionId})).allowed!==true)throw new HttpError(403,'电脑任务授权已失效','computer_authorization_revoked')},body=>this.api('artifact',{connectionId,...body}))
  172. }else gateway=new WorkspaceGateway(this.target)
  173. const connection:Connection={cleanupOnly:p.cleanupOnly===true,gateway,sessions:new Map(),running:new Set(),events:Promise.resolve()}
  174. this.connections.set(connectionId,connection)
  175. gateway.onEvent=(frame:GatewayFrame)=>{
  176. if(frame.session_id&&!connection.sessions.has(frame.session_id))return
  177. if(frame.session_id&&['message.complete','run.completed','run.failed'].includes(frame.type))connection.running.delete(frame.session_id)
  178. if(frame.type==='session.info'&&frame.session_id)for(const lease of this.leases.values())if(lease.session.runtimeId===frame.session_id&&typeof frame.payload?.stored_session_id==='string')lease.session.storedId=frame.payload.stored_session_id
  179. connection.events=connection.events.then(()=>this.api('event',{connectionId,frame})).then(()=>{},error=>{if(error instanceof HttpError&&error.status===410)void this.closeConnection(connectionId);else void this.disconnect()})
  180. }
  181. gateway.onDisconnect=()=>{void this.api('event',{connectionId,closed:true}).catch(()=>{})}
  182. try {await gateway.connect();if(!this.active||!this.connected||this.connections.get(connectionId)!==connection)throw new Error('执行节点已断开');return {ok:true}}catch(error){this.connections.delete(connectionId);gateway.close();throw error}
  183. }
  184. if(command.kind==='gateway.close'){await this.closeConnection(connectionId);return {ok:true}}
  185. if(command.kind==='gateway.rpc') {
  186. const connection=this.connections.get(connectionId),method=String(p.method),params=p.params as Record<string,unknown>
  187. if(!connection||!commands.has(method)||!params||typeof params!=='object')throw new HttpError(403,'执行通道或命令无效','runner_command_forbidden')
  188. if(connection.cleanupOnly&&!['session.resume','session.interrupt','session.close'].includes(method))throw new HttpError(403,'清理通道不允许执行新工作','runner_cleanup_forbidden')
  189. if(method==='session.create'||method==='session.resume'){if(!connection.cleanupOnly)this.requireProfile(params.profile)}
  190. else if(!connection.sessions.has(String(params.session_id)))throw new HttpError(403,'会话不属于这个通道','runner_session_forbidden')
  191. if(method==='prompt.submit')connection.running.add(String(params.session_id))
  192. const result=await connection.gateway.rpc(method,params)
  193. if(method==='session.create'||method==='session.resume'){
  194. if(this.connections.get(connectionId)!==connection)throw new HttpError(409,'执行通道已关闭','runner_command_uncertain')
  195. connection.sessions.set(String(result.session_id),String(params.profile))
  196. }
  197. if(method==='session.interrupt')connection.running.delete(String(params.session_id))
  198. return result
  199. }
  200. const leaseId=String(p.leaseId??'')
  201. if(command.kind==='lease.create') {
  202. this.requireProfile(p.profile)
  203. if(this.closedLeases.has(leaseId)||this.leases.has(leaseId)||!/^[0-9a-f-]{36}$/.test(leaseId)||this.leases.size>=64)throw new Error('工具授权无效或已达上限')
  204. const session=p.session as {runtimeId:string;storedId:string}
  205. const source=[...this.connections].find(([,c])=>c.sessions.get(session?.runtimeId)===p.profile)
  206. if(!source)throw new HttpError(403,'工具会话不属于此节点','runner_session_forbidden')
  207. const controller=new AbortController()
  208. const virtual=source[1].gateway
  209. if(virtual instanceof ComputerGateway){
  210. virtual.installTeamLease(leaseId,p.catalog as any[],async(toolId,args,callId)=>(await this.api('tool',{leaseId,toolId,arguments:args,callId})).result)
  211. const lease={bind:async()=>{},dispose:async()=>{virtual.removeTeamLease(leaseId)}}
  212. this.leases.set(leaseId,{lease,controller,session,connectionId:source[0],profile:String(p.profile)});return {ok:true}
  213. }
  214. const lease=await createWorkspaceToolLease({target:this.target,profile:String(p.profile),workId:String(p.workId),session:()=>session,signal:controller.signal,
  215. assertActive:()=>{if(!this.connected||controller.signal.aborted)throw new Error('工具授权已失效')},
  216. catalog:()=>p.catalog as any[],call:async(toolId,args,callId)=>(await this.api('tool',{leaseId,toolId,arguments:args,callId:callId??randomUUID()})).result,
  217. onFailure:()=>{controller.abort()},
  218. })
  219. if(!this.active||!this.connected||this.closedLeases.has(leaseId)||!this.connections.has(source[0])){controller.abort();await lease.dispose();throw new Error('工具授权已停止')}
  220. this.leases.set(leaseId,{lease,controller,session,connectionId:source[0],profile:String(p.profile)});return {ok:true}
  221. }
  222. if(command.kind==='lease.bind') {
  223. const lease=this.leases.get(leaseId)
  224. if(!lease)throw new HttpError(410,'工具授权已关闭','runner_lease_expired')
  225. const update=p.session as {runtimeId?:string;storedId?:string}|undefined
  226. if(update?.runtimeId!==lease.session.runtimeId||!this.connections.get(lease.connectionId)?.sessions.has(update.runtimeId))throw new HttpError(403,'工具不能改绑其他会话','runner_session_forbidden')
  227. if(typeof update.storedId!=='string')throw new Error('存储会话编号无效')
  228. lease.session.storedId=update.storedId;await lease.lease.bind();return {ok:true}
  229. }
  230. if(command.kind==='lease.close') {
  231. this.closedLeases.set(leaseId,Date.now()+60000)
  232. const lease=this.leases.get(leaseId);this.leases.delete(leaseId)
  233. if(lease){lease.controller.abort();await lease.lease.dispose()}return {ok:true}
  234. }
  235. throw new HttpError(400,'未知执行命令','runner_command_unknown')
  236. }
  237. private compactReceipts(){
  238. // Keep fingerprints as replay tombstones; large HTTP/file replies expire.
  239. this.db.prepare("UPDATE commands SET state='retired',result=NULL WHERE state='complete' AND created<?").run(Date.now()-3600000)
  240. }
  241. private handled=0
  242. async handle(command:RunnerCommand):Promise<unknown> {
  243. if(++this.handled%256===0)this.compactReceipts()
  244. for(const map of [this.closedConnections,this.closedLeases])for(const [id,expires] of map)if(expires<Date.now())map.delete(id)
  245. const existing=this.inflight.get(command.id)
  246. const fingerprint=createHash('sha256').update(JSON.stringify(command)).digest('hex')
  247. const previous=this.db.prepare('SELECT * FROM commands WHERE id=?').get(command.id)
  248. if(previous){
  249. if(previous.fingerprint!==fingerprint)throw new HttpError(409,'命令编号已用于其他内容','idempotency_conflict')
  250. if(existing)return existing
  251. if(previous.state==='complete')return JSON.parse(String(previous.result))
  252. throw new HttpError(409,'原命令结果不确定,禁止重复执行','runner_command_uncertain')
  253. }
  254. this.db.prepare("INSERT INTO commands VALUES(?,?,'started',NULL,?)").run(command.id,fingerprint,Date.now())
  255. const pending=Promise.resolve().then(async()=>{
  256. if(command.expiresAt<=Date.now()||!(await this.api('admit',{id:command.id})).allowed||command.expiresAt<=Date.now())throw new HttpError(403,'命令授权已经失效,未执行','runner_command_not_admitted')
  257. return this.execute(command)
  258. }).then(result=>({result}),error=>({error:{message:error instanceof HttpError||error instanceof ComputerError?error.message:'执行节点操作失败',code:error instanceof HttpError||error instanceof ComputerError?error.code:'runner_error'}}))
  259. .then(reply=>{this.db.prepare("UPDATE commands SET state='complete',result=? WHERE id=?").run(JSON.stringify(reply),command.id);return reply})
  260. .finally(()=>this.inflight.delete(command.id))
  261. this.inflight.set(command.id,pending);return pending
  262. }
  263. private lastStatus=''
  264. private status(message:string){if(message!==this.lastStatus){this.lastStatus=message;this.onStatus(message)}}
  265. async run(signal:AbortSignal):Promise<void> {
  266. const deliveries=new Map<string,Promise<void>>()
  267. const cleanupTimer=setInterval(()=>{void this.computers?.pool.expire().catch(()=>{})},5000);cleanupTimer.unref()
  268. const abort=()=>{this.active=false;this.controlAbort.abort();void this.disconnect()};signal.addEventListener('abort',abort,{once:true})
  269. try {
  270. while(this.active&&!signal.aborted) {
  271. try {
  272. const revision=this.disconnectRevision
  273. const response=await this.api('poll')
  274. await this.disconnecting
  275. if(revision!==this.disconnectRevision)continue
  276. if(this.serverEpoch&&this.serverEpoch!==response.epoch)await this.disconnect()
  277. this.serverEpoch=response.epoch;this.connected=true;this.status('执行节点已连接')
  278. for(const command of response.commands as RunnerCommand[]){
  279. if(deliveries.has(command.id))continue
  280. const delivery=(async()=>{
  281. let reply
  282. try {reply=await this.handle(command)}catch(error){reply={error:{message:'原命令需要核对',code:error instanceof HttpError?error.code:'runner_error'}}}
  283. await this.api('result',{id:command.id,...reply as object})
  284. })().catch(()=>{}).finally(()=>deliveries.delete(command.id))
  285. deliveries.set(command.id,delivery)
  286. }
  287. }catch(error){
  288. this.status(error instanceof HttpError?error.message:'连接暂时中断,正在重连')
  289. await this.disconnect()
  290. if(error instanceof HttpError&&error.code==='runner_unauthorized')throw error
  291. if(this.active)await new Promise(resolve=>setTimeout(resolve,1000))
  292. }
  293. }
  294. }finally{
  295. clearInterval(cleanupTimer)
  296. signal.removeEventListener('abort',abort)
  297. try{
  298. await this.disconnect().catch(()=>{})
  299. await Promise.allSettled(deliveries.values());await Promise.allSettled(this.inflight.values())
  300. await this.localVm?.close();await this.computers?.controls.close();await this.computers?.pool.close()
  301. }finally{this.target.client.close();this.controlTransport.close();this.db.close();this.status('执行节点已停止')}
  302. }
  303. }
  304. }