runnerHub.ts 45 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432
  1. import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto'
  2. import Router from '@koa/router'
  3. import type Koa from 'koa'
  4. import {createReadStream,existsSync} from 'node:fs'
  5. import {dirname,resolve} from 'node:path'
  6. import {fileURLToPath} from 'node:url'
  7. import { z } from 'zod'
  8. import { HttpError } from './errors.js'
  9. import { parse, type WorkspaceStore } from './workspaceStore.js'
  10. import type { LocalAuthStore } from './localAuth.js'
  11. import type { GatewayFrame, GatewayTarget } from './workspaceGateway.js'
  12. import type { UpstreamRequestOptions, UpstreamResponse } from './upstream.js'
  13. import type { LeaseInput, WorkspaceToolLease } from './workspaceToolLease.js'
  14. import { RUNNER_PROTOCOL, type RunnerCommand, type RunnerRecord } from '../shared/runner.js'
  15. import {ComposeDesktops} from './composeDesktops.js'
  16. import {readHostTools} from './hostToolSettings.js'
  17. const hash = (value: string) => createHash('sha256').update(value).digest('hex')
  18. const runnerBundle=()=>[resolve(dirname(fileURLToPath(import.meta.url)),'runner-bundle.tar.gz'),resolve(dirname(fileURLToPath(import.meta.url)),'../../runner-bundle.tar.gz')].find(existsSync)
  19. const input = z.object({ name:z.string().trim().min(1).max(100),sourceNodeId:z.string().min(1).max(100).default('local'),
  20. allowedProfiles:z.array(z.string().min(1).max(100)).min(1).max(256) }).strict()
  21. interface Pending { command:RunnerCommand; resolve(value:any):void; reject(error:Error):void; timer:ReturnType<typeof setTimeout>; nextDelivery:number; valid():boolean }
  22. interface GatewayConnection { taskEnvironment?:()=>import('../shared/executionEnvironment.js').TaskEnvironment; runnerId:string;computer?:{environmentId:string;ownerKey:string;agentId:string};cleanupOnly?:boolean;publishArtifact?:(name:string,bytes:Buffer)=>Promise<unknown>; valid():boolean; onEvent(frame:GatewayFrame):void; onDisconnect():void }
  23. interface Lease { runnerId:string; input:LeaseInput; calls:Map<string,{fingerprint:string;result:Promise<unknown>}> }
  24. // Only expose known diagnostics; arbitrary transport errors can contain credentials.
  25. function networkFailure(error:unknown,fallback:string){
  26. if(!(error instanceof HttpError))return fallback
  27. if(error.code==='computer_network_upgrade_required')return '缺少虚拟机网络网关,请在 Bot 设置 → 本地虚拟机中重新准备网络镜像;Compose 节点请更新网络网关'
  28. if(error.code==='computer_network_cleanup_failed')return '旧网络任务未完成清理,请重试保存代理'
  29. return fallback
  30. }
  31. /** Outbound-only machine transport. Browser auth and machine auth remain separate. */
  32. export class RunnerHub {
  33. executionSettings?:import('./executionSettings.js').ExecutionSettings
  34. private networkApplied=new Map<string,{revision:number;error?:string}>()
  35. networkStatus(){return this.records().filter(r=>r.enabled).map(r=>{const s=this.summary(r),applied=this.networkApplied.get(r.id);return {id:r.id,name:r.name,online:s.online,status:!s.online?'离线':!s.features.includes('vm-egress-v1')?'需要更新 Runner':applied?.error?`未生效:${applied.error}`:applied?.revision===this.executionSettings?.proxy().revision?'已同步,启动时校验出口':'待更新'}})}
  36. private networkSupported(id:string){return !this.executionSettings?.proxy().enabled||this.online.get(id)?.features.includes('vm-egress-v1')===true}
  37. private networkCommandAllowed(id:string,kind:RunnerCommand['kind'],payload:Record<string,unknown>){return kind!=='computer.control'||this.networkSupported(id)||!(payload.op==='autostart'||payload.op==='take'||payload.op==='input'||payload.op==='lifecycle'&&!['stop','remove'].includes(String(payload.action)))}
  38. async configureNetworks(){await Promise.all(this.records().filter(r=>r.enabled&&this.summary(r).online).map(async r=>{
  39. if(!this.summary(r).features.includes('vm-egress-v1')){
  40. if(!this.networkSupported(r.id))await Promise.all([...this.connections].filter(([,c])=>c.runnerId===r.id&&c.computer).map(async([id,c])=>{this.connections.delete(id);c.onDisconnect();await this.request(r.id,'gateway.close',{connectionId:id}).catch(()=>{})}))
  41. return
  42. }
  43. try{const result=await this.request(r.id,'network.configure',{});this.networkApplied.set(r.id,{revision:result.revision})}catch(error){this.networkApplied.set(r.id,{revision:-1,error:networkFailure(error,'网络配置未生效,请检查执行节点后重试保存')})}
  44. }))}
  45. async testNetworks(){const nodes=this.records().filter(r=>r.enabled&&this.summary(r).online&&this.summary(r).features.includes('vm-egress-v1'));if(!nodes.length)throw new HttpError(409,'没有支持代理的在线执行节点','proxy_node_unavailable');return {nodes:await Promise.all(nodes.map(async r=>{try{await this.request(r.id,'network.test',{});return {id:r.id,name:r.name,ok:true}}catch(error){return {id:r.id,name:r.name,ok:false,error:networkFailure(error,'代理连接失败,请核对地址、认证和执行节点网络')}}}))}}
  46. composeDesktops=new ComposeDesktops([])
  47. get idleForUpdate(): boolean { return this.localVmActivity.size === 0 && [...this.pending.values()].every(commands => commands.size === 0) && [...this.artifacts.values()].every(upload => upload.result !== undefined) }
  48. private localVmActivity = new Map<string, { owner: string; runnerId: string }>()
  49. private localVmChecks = new Set<string>()
  50. browserAllowed:(runnerId:string,scope:unknown,grantId?:string)=>boolean=()=>false
  51. controlAllowed:(id:string,runnerId:string)=>boolean=()=>false
  52. readonly epoch=randomUUID()
  53. private readonly credentialServerId=randomUUID()
  54. private online = new Map<string,{instance:string;seen:number;epoch:string;features:string[];wake?:()=>void}>()
  55. private retired = new Map<string,Set<string>>()
  56. private pending = new Map<string,Map<string,Pending>>()
  57. private connections = new Map<string,GatewayConnection>()
  58. private artifacts=new Map<string,{runnerId:string;connectionId:string;name:string;size:number;digest:string;chunks:Buffer[];bytes:number;next:number;result?:unknown}>()
  59. private leases = new Map<string,Lease>()
  60. private watchdog: ReturnType<typeof setInterval>
  61. /** Origin of the server Hermes. Satellite configs must use this, never loopback. */
  62. hermesUpstream=''
  63. constructor(readonly store:WorkspaceStore,readonly auth:LocalAuthStore,readonly local:GatewayTarget) {
  64. for (const grant of store.list<{ owner: string; runnerId: string; expiresAt: number }>('_system', 'local-vm-grant'))
  65. if (grant.expiresAt > Date.now()) this.localVmActivity.set(`${grant.owner}:${grant.runnerId}`, { owner: grant.owner, runnerId: grant.runnerId })
  66. this.watchdog=setInterval(()=>{
  67. for(const [id,state] of this.online) if(Date.now()-state.seen>35_000)this.disconnect(id)
  68. for(const [id,upload] of this.artifacts)if(!this.connections.has(upload.connectionId))this.artifacts.delete(id)
  69. for (const [key, activity] of this.localVmActivity) {
  70. if (this.localVmChecks.has(key) || !this.online.has(activity.runnerId)) continue
  71. this.localVmChecks.add(key)
  72. void this.request(activity.runnerId, 'local-vm.manage', { op: 'status', owner: hash(activity.owner) }, () => this.store.get<RunnerRecord>('_system', 'runner', activity.runnerId)?.enabled === true)
  73. .then(result => { if (result.busy === false && this.localVmActivity.get(key) === activity) this.localVmActivity.delete(key) })
  74. .catch(() => {}).finally(() => this.localVmChecks.delete(key))
  75. }
  76. },5000);this.watchdog.unref()
  77. }
  78. localVmAllowed:(id:string,runnerId:string)=>boolean=()=>false
  79. async localVm(owner:string,id:string,payload:Record<string,unknown>){
  80. const record=this.store.require<RunnerRecord>('_system','runner',id),version=this.auth.pushAuthorizationVersion(owner)
  81. if(!record.enabled||!this.online.get(id)?.features.includes('local-vm-v1'))throw new HttpError(409,'本地虚拟机执行环境离线或版本过旧,请重新连接或更新','local_vm_unavailable')
  82. if(payload.imageKey&&!this.online.get(id)?.features.includes('image-options-v1'))throw new HttpError(409,'执行节点版本不支持镜像选择,请更新 Runner','computer_image_options_unavailable')
  83. if(payload.op==='idle-policy'&&!this.online.get(id)?.features.includes('idle-stop-policy-v1'))throw new HttpError(409,'执行节点版本不支持空闲停止设置,请更新 Runner','local_vm_idle_policy_unavailable')
  84. const valid=()=>this.auth.isAdminActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version&&this.store.get<RunnerRecord>('_system','runner',id)?.enabled===true
  85. if(!valid())throw new HttpError(403,'本地虚拟机设置需要管理员权限','admin_required')
  86. const key = `${owner}:${id}`
  87. if (payload.op === 'prepare') this.localVmActivity.set(key, { owner, runnerId: id })
  88. const activity = this.localVmActivity.get(key)
  89. const result=await this.request(id,'local-vm.manage',{...payload,owner:hash(owner)},valid)
  90. if(!valid())throw new HttpError(403,'本地虚拟机设置授权已失效','admin_required')
  91. if (payload.op === 'status' && result.busy === false && this.localVmActivity.get(key) === activity) this.localVmActivity.delete(key)
  92. else if (payload.op === 'status' && result.busy === true && !activity) this.localVmActivity.set(key, { owner, runnerId: id })
  93. return result
  94. }
  95. records():RunnerRecord[] {return this.store.list('_system','runner')}
  96. credentialBinding(owner:string,agent:import('../shared/workspace.js').WorkspaceAgent,location:'node'|'server'='node') {
  97. if(agent.archived||!this.auth.canUseSource(owner,agent.nodeId,agent.profile))throw new HttpError(403,'Bot 凭据授权已结束','vault_agent_forbidden')
  98. const record=this.records().find(r=>r.enabled&&r.sourceNodeId===agent.nodeId&&(r.sourceOwner==='_system'||r.sourceOwner===owner))
  99. const online=record&&this.online.get(record.id)
  100. if(location==='server'){
  101. if(record&&(!online||Date.now()-online.seen>=35000))throw new HttpError(503,'Bot 所在节点离线,请等待节点连接','vault_runner_offline')
  102. if(record&&!record.allowedProfiles.includes(agent.profile))throw new HttpError(403,'执行节点未授权当前 Profile','vault_agent_forbidden')
  103. // No Runner is needed for direct Hermes tasks. Existing Runner reconnects
  104. // still change the binding, even though secrets remain in this server.
  105. return {runnerId:this.credentialServerId,runnerInstance:this.epoch,
  106. runnerEpoch:hash(JSON.stringify([agent.nodeId,agent.profile,record?.id,online?.instance,online?.epoch]))}
  107. }
  108. if(!record||!online||Date.now()-online.seen>=35000)throw new HttpError(503,'执行节点离线,请等待节点连接','vault_runner_offline')
  109. if(!record.allowedProfiles.includes(agent.profile))throw new HttpError(403,'执行节点未授权当前 Profile','vault_agent_forbidden')
  110. if(!online.features.includes('credential-ref-v1'))throw new HttpError(409,'请更新 Runner,旧节点不支持凭据引用授权','vault_runner_upgrade_required')
  111. return {runnerId:record.id,runnerInstance:online.instance,runnerEpoch:online.epoch}
  112. }
  113. enroll(owner:string,value:unknown) {
  114. const body=parse(input,value)
  115. body.allowedProfiles=this.auth.validateAssignedProfiles(body.allowedProfiles)
  116. if(body.sourceNodeId!=='local'&&this.store.require<import('./workspaceGateway.js').WorkspaceNode>(owner,'node',body.sourceNodeId).transport==='paired-web')throw new HttpError(409,'配对 Web 节点应在它自己的服务中注册 Runner','runner_source_invalid')
  117. const sourceOwner=body.sourceNodeId==='local'?'_system':owner
  118. if(this.records().some(r=>r.enabled&&r.sourceOwner===sourceOwner&&r.sourceNodeId===body.sourceNodeId))
  119. throw new HttpError(409,'该来源已有执行节点,请先停用原节点','runner_source_in_use')
  120. const token=randomBytes(32).toString('base64url')
  121. const record:RunnerRecord={...body,id:randomUUID(),sourceOwner,tokenHash:hash(token),enabled:true,createdAt:Date.now()}
  122. this.store.put('_system','runner',record.id,record)
  123. return {runner:this.summary(record),token}
  124. }
  125. summary(record:RunnerRecord) {
  126. const {tokenHash:_secret,sourceOwner:_owner,...publicRecord}=record
  127. return {...publicRecord,features:this.online.get(record.id)?.features??[],online:!!this.online.get(record.id)&&Date.now()-this.online.get(record.id)!.seen<35_000}
  128. }
  129. remove(id:string) {
  130. const record=this.store.require<RunnerRecord>('_system','runner',id)
  131. record.enabled=false;this.store.put('_system','runner',id,record);this.disconnect(id)
  132. }
  133. private disconnect(id:string) {
  134. this.online.get(id)?.wake?.();this.online.delete(id)
  135. for(const [key,connection] of this.connections) if(connection.runnerId===id){this.connections.delete(key);connection.onDisconnect()}
  136. for(const [key,lease] of this.leases) if(lease.runnerId===id)this.leases.delete(key)
  137. for(const pending of this.pending.get(id)?.values()??[]){clearTimeout(pending.timer);pending.reject(new HttpError(503,'执行节点已断开,需核对原执行','runner_offline'))}
  138. this.pending.delete(id)
  139. }
  140. private request(id:string,kind:RunnerCommand['kind'],payload:Record<string,unknown>,valid:()=>boolean=()=>true):Promise<any> {
  141. if(!this.online.has(id))return Promise.reject(new HttpError(503,'执行节点未连接','runner_offline'))
  142. if(!this.networkCommandAllowed(id,kind,payload))return Promise.reject(new HttpError(409,'全局代理需要新版 Runner,请更新后操作虚拟机','proxy_runner_required'))
  143. const queue=this.pending.get(id)??new Map<string,Pending>();this.pending.set(id,queue)
  144. if(queue.size>=64)return Promise.reject(new HttpError(429,'执行节点请求过多','runner_busy'))
  145. // Allow Hermes' 60-second binding budget plus transport overhead.
  146. const timeout = kind === 'lease.bind' ? 70000 : kind === 'http' && payload.path === '/api/plugins/yaoyao-bot-bridge/memory-extract' ? 120000 : 30000
  147. const command:RunnerCommand={id:randomUUID(),kind,payload,expiresAt:Date.now()+timeout}
  148. const bytes=Buffer.byteLength(JSON.stringify(command))
  149. if(bytes>36*1024*1024)return Promise.reject(new HttpError(413,'执行命令超过传输限制','runner_payload_limit'))
  150. if([...queue.values()].reduce((size,p)=>size+Buffer.byteLength(JSON.stringify(p.command)),bytes)>64*1024*1024)return Promise.reject(new HttpError(429,'执行节点待传输数据过多','runner_busy'))
  151. return new Promise((resolve,reject)=>{
  152. const timer=setTimeout(()=>{queue.delete(command.id);reject(new HttpError(504,'执行节点回应超时,请核对执行状态','runner_command_uncertain'))},timeout)
  153. queue.set(command.id,{command,resolve,reject,timer,nextDelivery:0,valid});this.online.get(id)?.wake?.()
  154. })
  155. }
  156. target(owner:string,sourceNodeId:string,computer?:import('../runner/worker/gateway.js').ComputerTarget):GatewayTarget|undefined {
  157. const record=this.records().find(r=>r.enabled&&r.sourceNodeId===sourceNodeId&&(r.sourceOwner==='_system'||r.sourceOwner===owner))
  158. if(!record)return undefined
  159. const requireProfile=(profile:string)=>{
  160. if(!this.auth.canUseSource(owner,sourceNodeId,profile)||!record.allowedProfiles.includes(profile))throw new HttpError(403,'执行节点未授权这个基础 Profile','runner_profile_forbidden')
  161. }
  162. const requireComputer=()=>{
  163. if(computer?.profileSession&&!this.online.get(record.id)?.features.includes('profile-computer-v1'))throw new HttpError(409,'执行节点尚不支持本机协作模式,请更新配套 Runner','computer_profile_unavailable')
  164. if(computer?.hostAccess&&!this.online.get(record.id)?.features.includes('host-computer-tools-v1'))throw new HttpError(409,'执行节点版本不支持同时使用本机和虚拟机,请更新 Runner','computer_host_tools_unavailable')
  165. if(computer&&this.composeDesktops.desktops.length){
  166. if(!this.online.get(record.id)?.features.includes('compose-desktops-v1'))throw new HttpError(409,'当前部署需要 Compose 桌面执行节点','compose_runner_required')
  167. const claim=this.store.get<{owner:string;runnerId:string}>('_system','compose-desktop-owner',computer.environmentId)
  168. if(!this.composeDesktops.desktops.some(d=>d.id===computer.environmentId)||claim?.owner!==owner||claim.runnerId!==record.id)throw new HttpError(409,'请先为机器人选择已有 Compose 共享桌面','compose_desktop_required')
  169. }
  170. if(computer&&!this.online.get(record.id)?.features.includes('computer-worker-v1'))throw new HttpError(409,'执行节点未提供隔离 Worker 能力,请检查配置或更新 Runner','computer_unavailable')
  171. if(computer&&!this.online.get(record.id)?.features.includes('hermes-computer-v2'))throw new HttpError(409,'请更新配套 Runner,虚拟机 Bot 需要 Hermes 托管会话能力。','computer_runner_upgrade_required')
  172. if(computer&&computer.environmentId!==computer.agentId&&!this.online.get(record.id)?.features.includes('shared-computer-v1'))throw new HttpError(409,'执行节点不支持共享电脑,请更新 Runner','shared_computer_unavailable')
  173. }
  174. if(computer?.hostAccess)requireComputer()
  175. const requestHTTP=async(path:string,options:UpstreamRequestOptions={}):Promise<UpstreamResponse>=>{
  176. requireComputer()
  177. const profile=options.search?.get('profile')??'default'
  178. if(path!=='/api/profiles'&&path!=='/api/status')requireProfile(profile)
  179. const version=this.auth.pushAuthorizationVersion(owner)
  180. const valid=()=>this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version&&this.store.get<RunnerRecord>('_system','runner',record.id)?.enabled===true
  181. const result=await this.request(record.id,'http',{path,method:options.method??'GET',search:options.search?.toString(),body:options.body,computer},valid)
  182. if(!valid())throw new HttpError(403,'读取权限已失效','runner_authorization_revoked')
  183. if(path==='/api/profiles') {
  184. const parsed=JSON.parse(Buffer.from(result.body,'base64').toString())
  185. parsed.profiles=(parsed.profiles??[]).filter((p:any)=>{const name=typeof p==='string'?p:p.name;return record.allowedProfiles.includes(name)&&this.auth.canUseSource(owner,sourceNodeId,name)})
  186. result.body=Buffer.from(JSON.stringify(parsed)).toString('base64')
  187. }
  188. return {status:result.status,headers:new Headers(result.headers),body:Buffer.from(result.body,'base64')}
  189. }
  190. return {url:this.local.url,client:this.local.client,session:{request:requestHTTP,webSocketCredential:async()=>{throw new Error('Runner uses its own authenticated gateway')}},
  191. runner:{id:record.id,computer:!!computer,hermesComputer:!!computer&&this.online.get(record.id)?.features.includes('hermes-computer-v2')===true,helperRetirement:this.online.get(record.id)?.features.includes('helper-retirement-v1')===true,fileTransfer:this.online.get(record.id)?.features.includes('file-transfer-v1')===true,
  192. open:async(onEvent,onDisconnect,scope)=>{
  193. requireComputer()
  194. if(computer&&!scope?.cleanupOnly&&!this.networkSupported(record.id))throw new HttpError(409,'全局代理需要新版 Runner,请更新后使用虚拟机','proxy_runner_required')
  195. if(scope?.taskEnvironment&&!this.online.get(record.id)?.features.includes('execution-context-v1'))throw new HttpError(409,'请更新 Runner 以提供系统环境与授权','execution_runner_required')
  196. if(computer&&!scope)throw new HttpError(403,'隔离执行缺少任务授权','computer_scope_required')
  197. const version=this.auth.pushAuthorizationVersion(owner)
  198. const valid=()=>{
  199. try{
  200. scope?.authorize()
  201. if(computer&&!scope?.cleanupOnly&&!this.networkSupported(record.id))return false
  202. if(computer&&!scope?.cleanupOnly){
  203. const agent=this.store.get<import('../shared/workspace.js').WorkspaceAgent>(owner,'agent',computer.agentId)
  204. if(!agent)return false
  205. }
  206. if(computer?.hostAccess){
  207. const agent=this.store.get<import('../shared/workspace.js').WorkspaceAgent>(owner,'agent',computer.agentId)
  208. if(!agent||agent.archived||agent.temporaryGoalId||agent.allowHostEnvironment!==true||(agent.computer!=='vm'&&agent.computer!==undefined)||agent.execution!=='computer')return false
  209. }
  210. return (scope?.cleanupOnly||(this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version))&&this.store.get<RunnerRecord>('_system','runner',record.id)?.enabled===true
  211. }catch{return false}
  212. }
  213. const id=randomUUID();this.connections.set(id,{runnerId:record.id,computer,valid,onEvent,onDisconnect,taskEnvironment:scope?.taskEnvironment,publishArtifact:scope?.publishArtifact,cleanupOnly:scope?.cleanupOnly})
  214. try {await this.request(record.id,'gateway.open',{connectionId:id,computer,workId:scope?.workId,cleanupOnly:scope?.cleanupOnly})} catch(error){this.connections.delete(id);void this.request(record.id,'gateway.close',{connectionId:id}).catch(()=>{});throw error}
  215. let closed=false
  216. return {rpc:async(method,params)=>{
  217. if(closed)throw new Error('Runner gateway closed')
  218. const cleanup=['session.interrupt','session.close','session.active_list'].includes(method)
  219. if(method==='computer.transfer'&&!this.online.get(record.id)?.features.includes('file-transfer-v1'))throw new HttpError(409,'请更新 Runner 以使用虚拟机文件传输','computer_transfer_upgrade_required')
  220. if(scope?.cleanupOnly&&(!['session.resume','session.interrupt','session.close','session.active_list'].includes(method)||(method==='session.resume'&&params.session_id!==scope.sessionId)))throw new HttpError(403,'清理通道只能操作原绑定会话','runner_cleanup_forbidden')
  221. if(!cleanup&&!valid())throw new HttpError(403,'本轮执行授权已失效','runner_command_not_admitted')
  222. if(!scope?.cleanupOnly&&(method==='session.create'||method==='session.resume'))requireProfile(String(params.profile??'default'))
  223. return this.request(record.id,'gateway.rpc',{connectionId:id,method,params})
  224. },close:()=>{
  225. closed=true;this.connections.delete(id)
  226. for(const [key,pending] of this.pending.get(record.id)??[])if(pending.command.payload.connectionId===id){clearTimeout(pending.timer);pending.reject(new HttpError(409,'执行通道已关闭','runner_command_not_admitted'));this.pending.get(record.id)!.delete(key)}
  227. void this.request(record.id,'gateway.close',{connectionId:id}).catch(()=>{})
  228. }}
  229. },
  230. lease:async(options)=>{
  231. requireComputer()
  232. requireProfile(options.profile);options.assertActive()
  233. if(options.computerPolicy&&['none','virtual'].includes(options.computerPolicy.mode)&&!this.online.get(record.id)?.features.includes('execution-context-v1'))throw new HttpError(409,'请更新 Runner 以执行环境限制','execution_runner_required')
  234. if(options.workspaceMemory&&!this.online.get(record.id)?.features.includes('workspace-memory-bind-v1'))throw new HttpError(409,'请更新配套 Runner,以支持工具桥会话的独立记忆。','workspace_memory_runner_required')
  235. const id=randomUUID();this.leases.set(id,{runnerId:record.id,input:options,calls:new Map()})
  236. let disposed=false
  237. const dispose=async()=>{if(disposed)return;disposed=true;this.leases.delete(id);options.signal.removeEventListener('abort',abort);await this.request(record.id,'lease.close',{leaseId:id}).catch(()=>{})}
  238. const abort=()=>{void dispose()};options.signal.addEventListener('abort',abort,{once:true})
  239. try {
  240. await this.request(record.id,'lease.create',{leaseId:id,profile:options.profile,workId:options.workId,session:options.session(),catalog:options.catalog(),workspaceMemory:options.workspaceMemory===true,computerPolicy:options.computerPolicy})
  241. options.assertActive()
  242. if(options.signal.aborted)throw new Error('执行已停止')
  243. return {bind:async()=>{options.assertActive();await this.request(record.id,'lease.bind',{leaseId:id,session:options.session()});options.assertActive()},dispose} satisfies WorkspaceToolLease
  244. } catch(error){await dispose();throw error}
  245. },
  246. }}
  247. }
  248. computerRunner(owner:string,agent:import('../shared/workspace.js').WorkspaceAgent){
  249. const record=this.records().find(record=>record.enabled&&record.sourceNodeId===agent.nodeId&&(record.sourceOwner==='_system'||record.sourceOwner===owner))
  250. if(!record)throw new HttpError(409,'本地虚拟机执行节点尚未启用,请在本地虚拟机设置中完成配置','computer_control_unavailable')
  251. const online=this.online.get(record.id)
  252. if(!online)throw new HttpError(503,'本地虚拟机执行节点未连接,请检查 Runner 是否运行','runner_offline')
  253. if(!online.features.includes('computer-control-v1'))throw new HttpError(409,'本地虚拟机执行节点不支持电脑查看与控制,请启用隔离电脑 Worker 或更新 Runner','computer_control_unavailable')
  254. if(this.composeDesktops.desktops.length&&!this.online.get(record.id)?.features.includes('compose-desktops-v1'))throw new HttpError(409,'当前部署需要 Compose 桌面执行节点','compose_runner_required')
  255. if(!record.allowedProfiles.includes(agent.profile)||!this.auth.canUseSource(owner,agent.nodeId,agent.profile))throw new HttpError(403,'电脑 Profile 未授权','computer_profile_forbidden')
  256. return record
  257. }
  258. sharedComputerRunner(owner:string,agent:import('../shared/workspace.js').WorkspaceAgent){const record=this.computerRunner(owner,agent);if(agent.computerEnvironmentId&&this.store.require<import('./sharedComputers.js').SharedComputer>(owner,'shared-computer',agent.computerEnvironmentId).runnerId!==record.id)throw new HttpError(409,'共享电脑的原执行节点已变化','shared_runner_changed');if(!this.online.get(record.id)?.features.includes('shared-computer-v1'))throw new HttpError(409,'执行节点不支持共享电脑,请更新 Runner','shared_computer_unavailable');return record}
  259. async computer(owner:string,agent:import('../shared/workspace.js').WorkspaceAgent,op:string,data:Record<string,unknown>,authorize:()=>void):Promise<any>{
  260. const record=agent.computerEnvironmentId?this.sharedComputerRunner(owner,agent):this.computerRunner(owner,agent),version=this.auth.pushAuthorizationVersion(owner)
  261. if(op==='image'&&!this.online.get(record.id)?.features.includes('image-options-v1'))throw new HttpError(409,'执行节点版本不支持镜像选择,请更新 Runner','computer_image_options_unavailable')
  262. const valid=()=>{try{authorize();return this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version}catch{return false}}
  263. const result=await this.request(record.id,'computer.control',{target:{environmentId:agent.computerEnvironmentId??agent.id,agentId:agent.id,ownerKey:hash(owner)},profile:agent.profile,op,...data},valid)
  264. if(!valid())throw new HttpError(403,'电脑请求授权已失效','computer_control_expired')
  265. return result
  266. }
  267. browserRunner(owner:string,agent:import('../shared/workspace.js').WorkspaceAgent){
  268. const record=this.records().find(r=>r.enabled&&r.sourceNodeId===agent.nodeId&&(r.sourceOwner==='_system'||r.sourceOwner===owner))
  269. if(!record)throw new HttpError(409,'尚未找到此 Bot 的执行节点,请检查服务器的节点连接','browser_runner_missing')
  270. const online=this.online.get(record.id)
  271. if(!online)throw new HttpError(503,'执行节点暂未连接,节点上线后将自动检测浏览器','runner_offline')
  272. if(online.features.includes('managed-browser-disabled-v1'))throw new HttpError(409,'节点管理员已关闭托管浏览器,请在该节点重新启用','browser_node_disabled')
  273. if(!online.features.some(feature=>['managed-browser-v1','managed-browser-setup-v1'].includes(feature)))throw new HttpError(409,'执行节点版本尚不支持自动准备浏览器,请更新节点上的夭夭并重新启动','browser_runner_upgrade_required')
  274. if(!record.allowedProfiles.includes(agent.profile)||!this.auth.canUseSource(owner,agent.nodeId,agent.profile))throw new HttpError(403,'浏览器 Profile 未授权','browser_profile_forbidden')
  275. return {...record,browserEpoch:online.epoch}
  276. }
  277. async browser(owner:string,agent:import('../shared/workspace.js').WorkspaceAgent,payload:Record<string,unknown>,authorize:()=>void):Promise<any>{
  278. const record=this.browserRunner(owner,agent),version=this.auth.pushAuthorizationVersion(owner)
  279. const valid=()=>{try{authorize();return this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version&&this.store.get<RunnerRecord>('_system','runner',record.id)?.enabled===true}catch{return false}}
  280. if(!valid())throw new HttpError(403,'浏览器授权已失效','browser_authorization_revoked')
  281. const result=await this.request(record.id,'browser.invoke',{...payload,profile:agent.profile},valid)
  282. if(!valid())throw new HttpError(403,'浏览器授权已失效','browser_authorization_revoked')
  283. return result
  284. }
  285. supportsBrowserRetention(owner:string,agent:import('../shared/workspace.js').WorkspaceAgent){
  286. const runner=this.browserRunner(owner,agent)
  287. return this.online.get(runner.id)?.features.includes('managed-browser-retention-v1')===true
  288. }
  289. async retireHelper(owner:string,helper:import('../shared/workspace.js').WorkspaceAgent):Promise<void>{
  290. if(!helper.temporaryGoalId||!helper.archived||!helper.helperRunnerId)throw new HttpError(403,'只能清理已退役的任务助手','helper_cleanup_forbidden')
  291. const current=this.store.require<import('../shared/workspace.js').WorkspaceAgent>(owner,'agent',helper.id)
  292. if(!current.archived||current.temporaryGoalId!==helper.temporaryGoalId)throw new HttpError(409,'助手状态已改变','helper_cleanup_forbidden')
  293. const record=this.store.require<RunnerRecord>('_system','runner',helper.helperRunnerId)
  294. if(record.sourceNodeId!==helper.nodeId||(record.sourceOwner!=='_system'&&record.sourceOwner!==owner))throw new HttpError(403,'清理节点不匹配','helper_cleanup_forbidden')
  295. if(!this.online.get(record.id)?.features.includes('helper-retirement-v1'))throw new HttpError(503,'等待支持清理的原执行节点上线','runner_offline')
  296. await this.request(record.id,'computer.retire',{environmentId:helper.id,agentId:helper.id,ownerKey:hash(owner)})
  297. }
  298. adminRouter():Router {
  299. const router=new Router()
  300. router.get('/api/app/admin/runners',ctx=>{const actor=this.auth.requireAdmin(ctx);ctx.body={protocol:RUNNER_PROTOCOL,hermesUpstream:this.hermesUpstream,fixedDesktops:this.composeDesktops.desktops.length>0,bundleAvailable:!!runnerBundle(),runners:this.records().map(r=>this.summary(r)),sources:[{id:'local',name:'本地来源'},...this.store.list<import('./workspaceGateway.js').WorkspaceNode>(actor.id,'node').filter(n=>n.transport!=='paired-web').map(n=>({id:n.id,name:n.name}))]}})
  301. router.get('/api/app/admin/runners/bundle',ctx=>{this.auth.requireAdmin(ctx);const path=runnerBundle();if(!path)throw new HttpError(404,'当前服务未打包执行节点程序,请使用配套 App 或从源码构建 Runner','runner_bundle_missing');ctx.set('Cache-Control','no-store');ctx.attachment('yaoyao-runner.tar.gz');ctx.type='application/gzip';ctx.body=createReadStream(path)})
  302. router.post('/api/app/admin/runners',ctx=>{const actor=this.auth.requireAdmin(ctx);ctx.body=this.enroll(actor.id,(ctx.request as any).body);ctx.status=201})
  303. router.delete('/api/app/admin/runners/:id',ctx=>{this.auth.requireAdmin(ctx);this.remove(ctx.params.id);ctx.body={ok:true}})
  304. return router
  305. }
  306. middleware():Koa.Middleware {
  307. return async(ctx,next)=>{
  308. const match=/^\/api\/runner\/v1\/([0-9a-f-]{36})\/(poll|admit|result|event|tool|check|artifact|control-check|local-vm-check|browser-check|desktop|execution-context|network-settings)$/.exec(ctx.path)
  309. if(!match)return next()
  310. const record=this.store.get<RunnerRecord>('_system','runner',match[1]!),token=ctx.get('authorization').replace(/^Bearer /,'')
  311. if(ctx.get('origin')||!record?.enabled||!ctx.get('authorization').startsWith('Bearer ')||!timingSafeEqual(Buffer.from(record.tokenHash),Buffer.from(hash(token))))
  312. throw new HttpError(403,'执行节点授权无效','runner_unauthorized')
  313. const instance=ctx.get('x-runner-instance')
  314. if(!/^[0-9a-f-]{36}$/.test(instance)||ctx.get('x-runner-protocol')!==String(RUNNER_PROTOCOL))throw new HttpError(409,'执行节点协议不兼容','runner_protocol_mismatch')
  315. let previous=this.online.get(record.id)
  316. if(match[2]==='poll'&&previous?.instance===instance&&ctx.get('x-runner-reset')==='1'){this.disconnect(record.id);previous=undefined}
  317. if(match[2]!=='poll'&&previous?.instance!==instance)throw new HttpError(409,'执行节点会话已更换','runner_instance_changed')
  318. if(this.retired.get(record.id)?.has(instance))throw new HttpError(409,'执行节点实例已被替换','runner_instance_retired')
  319. if(previous&&previous.instance!==instance){
  320. const retired=this.retired.get(record.id)??new Set<string>()
  321. if(retired.size>=256)throw new HttpError(409,'执行节点更换过于频繁,请重新注册','runner_instance_limit')
  322. retired.add(previous.instance);this.retired.set(record.id,retired);this.disconnect(record.id);previous=undefined
  323. }
  324. if(match[2]!=='poll'&&ctx.get('x-runner-epoch')!==previous?.epoch)throw new HttpError(409,'执行连接代次已改变','runner_epoch_changed')
  325. const state=this.online.get(record.id)??{instance,seen:Date.now(),features:[],epoch:`${this.epoch}:${randomUUID()}`};state.seen=Date.now();if(ctx.get('x-runner-features'))state.features=ctx.get('x-runner-features').split(',').filter(value=>['execution-context-v1','vm-egress-v1','workspace-memory-bind-v1','hermes-computer-v2','profile-computer-v1','host-computer-tools-v1','file-transfer-v1','idle-stop-policy-v1','computer-worker-v1','artifact-chunks-v1','helper-retirement-v1','computer-control-v1','shared-computer-v1','local-vm-v1','image-options-v1','image-ready-v1','compose-desktops-v1','managed-browser-v1','managed-browser-setup-v1','managed-browser-disabled-v1','managed-browser-retention-v1','credential-ref-v1'].includes(value));this.online.set(record.id,state)
  326. ctx.set('Cache-Control','no-store')
  327. if(match[2]==='poll') {
  328. if(ctx.method!=='GET')throw new HttpError(405,'仅允许 GET','method_not_allowed')
  329. const available=()=>[...(this.pending.get(record.id)?.values()??[])].filter(p=>p.nextDelivery<=Date.now())
  330. if(previous&&!available().length)await new Promise<void>(resolve=>{
  331. const done=()=>{clearTimeout(timer);ctx.res.off('close',done);if(state.wake===done)state.wake=undefined;resolve()}
  332. const retry=Math.min(15000,...[...(this.pending.get(record.id)?.values()??[])].map(p=>Math.max(1,p.nextDelivery-Date.now())))
  333. const timer=setTimeout(done,retry);state.wake?.();state.wake=done;ctx.res.once('close',done)
  334. })
  335. if(this.online.get(record.id)!==state)throw new HttpError(409,'执行节点会话已更换','runner_instance_changed')
  336. let bytes=0
  337. const batch=available().filter(item=>{const size=Buffer.byteLength(JSON.stringify(item.command));if(bytes+size>38*1024*1024)return false;bytes+=size;return true});for(const item of batch)item.nextDelivery=Date.now()+5000
  338. ctx.body={epoch:state.epoch,commands:batch.map(p=>p.command)};return
  339. }
  340. if(ctx.method!=='POST')throw new HttpError(405,'仅允许 POST','method_not_allowed')
  341. let size=0;const chunks:Buffer[]=[]
  342. for await(const chunk of ctx.req){size+=chunk.length;if(size>(match[2]==='desktop'?40:16)*1024*1024)throw new HttpError(413,'执行节点响应过大','runner_payload_limit');chunks.push(Buffer.from(chunk))}
  343. let body:any
  344. try {body=JSON.parse(Buffer.concat(chunks).toString())}catch{throw new HttpError(400,'JSON 无效','invalid_json')}
  345. if(!body||typeof body!=='object'||Array.isArray(body))throw new HttpError(400,'请求必须是对象','invalid_json')
  346. if(match[2]==='network-settings'){if(Number.isInteger(body.appliedRevision)&&body.appliedRevision===this.executionSettings?.proxy().revision)this.networkApplied.set(record.id,{revision:body.appliedRevision});ctx.set('Cache-Control','no-store');ctx.body={proxy:this.executionSettings?.proxy()};return}
  347. if(match[2]==='execution-context'){const connection=this.connections.get(String(body.connectionId));if(!connection?.computer||connection.runnerId!==record.id||!connection.valid())throw new HttpError(403,'任务环境授权已结束','execution_grant_revoked');ctx.set('Cache-Control','no-store');ctx.body={environment:connection.taskEnvironment?.()};return}
  348. if(match[2]==='desktop'){
  349. if(record.sourceNodeId!=='local'||record.sourceOwner!=='_system'||!state.features.includes('compose-desktops-v1'))throw new HttpError(403,'该节点不能访问 Compose 桌面','compose_desktop_forbidden')
  350. const {id,operation,ownerKey,...payload}=parse(z.object({id:z.string(),operation:z.enum(['list','health','frame','acquire','renew','release','execute','cancel','skills-install','network-open','network-exchange','network-close','network-health']),ownerKey:z.string().optional()}).passthrough(),body)
  351. if(operation==='list'){ctx.body=await this.composeDesktops.status();return}
  352. if(operation==='network-health'){ctx.body=await this.composeDesktops.call(id,operation);return}
  353. const claim=this.store.get<{owner:string;runnerId:string}>('_system','compose-desktop-owner',id)
  354. if(!claim||claim.runnerId!==record.id||hash(claim.owner)!==ownerKey)throw new HttpError(403,'没有该共享桌面的使用权限','compose_desktop_forbidden')
  355. const valid=()=>this.auth.isUserActive(claim.owner)&&this.store.list<import('../shared/workspace.js').WorkspaceAgent>(claim.owner,'agent').some(a=>!a.archived&&a.computerEnvironmentId===id&&this.auth.canUseSource(claim.owner,a.nodeId,a.profile))
  356. const active=()=>[...this.connections.values()].some(c=>c.runnerId===record.id&&c.computer?.environmentId===id&&c.computer.ownerKey===ownerKey&&c.valid())||this.store.list<any>('_system','computer-control').some(g=>g.owner===claim.owner&&g.runnerId===record.id&&g.environmentId===id&&this.controlAllowed(g.id,record.id))
  357. if(!['release','cancel','network-close'].includes(operation)&&(!valid()||(!['health','frame'].includes(operation)&&!active())))throw new HttpError(403,'桌面操作缺少当前任务或接管授权','compose_desktop_forbidden')
  358. const result=await this.composeDesktops.call(id,operation,{...payload,owner:ownerKey})
  359. if(operation==='acquire'&&(!valid()||!active())){await this.composeDesktops.call(id,'release',{...result,cancel:true}).catch(()=>{});throw new HttpError(410,'桌面操作已取消','computer_control_expired')}
  360. ctx.body=result;return
  361. }
  362. if(match[2]==='browser-check'){ctx.body={allowed:state.features.some(feature=>['managed-browser-v1','managed-browser-setup-v1'].includes(feature))&&this.browserAllowed(record.id,body.scope,typeof body.grantId==='string'?body.grantId:undefined)};return}
  363. if(match[2]==='local-vm-check'){ctx.body={allowed:typeof body.id==='string'&&this.localVmAllowed(body.id,record.id)};return}
  364. if(match[2]==='control-check'){ctx.body={allowed:this.networkSupported(record.id)&&typeof body.controlId==='string'&&this.controlAllowed(body.controlId,record.id)};return}
  365. if(match[2]==='artifact'){
  366. const value=parse(z.object({connectionId:z.string().uuid(),id:z.string().uuid(),phase:z.enum(['begin','append','finish']),name:z.string().min(1).max(240).optional(),size:z.number().int().min(0).max(25*1024*1024).optional(),digest:z.string().regex(/^[a-f0-9]{64}$/).optional(),index:z.number().int().min(0).optional(),data:z.string().max(1024*1024).optional()}).strict(),body)
  367. const connection=this.connections.get(value.connectionId)
  368. if(connection?.runnerId!==record.id||!connection.valid()||!connection.publishArtifact)throw new HttpError(403,'产物不属于当前有效任务','artifact_forbidden')
  369. let upload=this.artifacts.get(value.id)
  370. if(value.phase==='begin'){
  371. if(!value.name||value.size===undefined||!value.digest)throw new HttpError(400,'产物信息不完整','artifact_invalid')
  372. if(upload&&(upload.connectionId!==value.connectionId||upload.name!==value.name||upload.size!==value.size||upload.digest!==value.digest))throw new HttpError(409,'产物编号已用于其他内容','idempotency_conflict')
  373. if(!upload){
  374. const current=[...this.artifacts.values()]
  375. if(current.filter(item=>item.connectionId===value.connectionId).length>=8||current.filter(item=>item.runnerId===record.id&&item.result===undefined).reduce((sum,item)=>sum+item.size,0)+value.size>64*1024*1024)throw new HttpError(429,'待传产物过多','artifact_limit')
  376. upload={runnerId:record.id,connectionId:value.connectionId,name:value.name,size:value.size,digest:value.digest,chunks:[],bytes:0,next:0};this.artifacts.set(value.id,upload)
  377. }
  378. }else{
  379. if(!upload||upload.connectionId!==value.connectionId||upload.runnerId!==record.id)throw new HttpError(404,'产物传输不存在','artifact_missing')
  380. if(value.phase==='append'){
  381. if(value.index===undefined||value.data===undefined||upload.result!==undefined)throw new HttpError(400,'产物分块无效','artifact_invalid')
  382. const bytes=Buffer.from(value.data,'base64')
  383. if(value.index<upload.next){if(!upload.chunks[value.index]?.equals(bytes))throw new HttpError(409,'重复分块内容不一致','idempotency_conflict')}
  384. else{if(value.index!==upload.next||upload.bytes+bytes.length>upload.size)throw new HttpError(409,'分块顺序或大小无效','artifact_invalid');upload.chunks.push(bytes);upload.bytes+=bytes.length;upload.next++}
  385. }else if(upload.result===undefined){
  386. if(upload.bytes!==upload.size)throw new HttpError(409,'产物尚未传输完整','artifact_incomplete')
  387. const bytes=Buffer.concat(upload.chunks)
  388. if(createHash('sha256').update(bytes).digest('hex')!==upload.digest)throw new HttpError(409,'产物校验失败','artifact_digest_mismatch')
  389. upload.result=await connection.publishArtifact(upload.name,bytes);upload.chunks=[]
  390. }
  391. }
  392. ctx.body={ok:true,next:upload!.next,result:upload!.result};return
  393. }
  394. if(match[2]==='check'){const connection=this.connections.get(body.connectionId);ctx.body={allowed:connection?.runnerId===record.id&&connection.valid(),fileTransferMaxBytes:readHostTools(this.store.home).fileTransferMaxMiB*1024*1024};return}
  395. if(match[2]==='admit') {
  396. const pending=this.pending.get(record.id)?.get(body.id),command=pending?.command
  397. let allowed=!!command&&command.expiresAt>Date.now()&&pending!.valid()&&this.networkCommandAllowed(record.id,command.kind,command.payload)
  398. if(command?.kind==='gateway.rpc'||command?.kind==='gateway.open'){const connection=this.connections.get(String(command.payload.connectionId));allowed&&=connection?.runnerId===record.id&&(command.kind==='gateway.rpc'&&['session.interrupt','session.close','session.active_list'].includes(String(command.payload.method))||connection.valid())}
  399. if(command?.kind==='lease.create'||command?.kind==='lease.bind') {
  400. const lease=this.leases.get(String(command.payload.leaseId))
  401. try {if(lease?.runnerId!==record.id)allowed=false;else lease.input.assertActive()}catch{allowed=false}
  402. }
  403. ctx.body={allowed};return
  404. }
  405. if(match[2]==='result') {
  406. const waiter=this.pending.get(record.id)?.get(body.id)
  407. if(waiter){clearTimeout(waiter.timer);this.pending.get(record.id)!.delete(body.id);body.error?waiter.reject(new HttpError(502,String(body.error.message??'执行节点失败').slice(0,1000),String(body.error.code??'runner_error'))):waiter.resolve(body.result)}
  408. ctx.body={ok:true};return
  409. }
  410. if(match[2]==='event') {
  411. const connection=this.connections.get(body.connectionId)
  412. if(connection?.runnerId!==record.id)throw new HttpError(410,'执行通道已关闭','runner_connection_closed')
  413. if(!connection.valid()){this.connections.delete(body.connectionId);connection.onDisconnect();throw new HttpError(410,'执行通道授权已失效','runner_connection_closed')}
  414. if(body.closed){this.connections.delete(body.connectionId);connection.onDisconnect()}
  415. else if(body.frame&&typeof body.frame.type==='string')connection.onEvent(body.frame)
  416. ctx.body={ok:true};return
  417. }
  418. const value=parse(z.object({leaseId:z.string().uuid(),toolId:z.string().min(1).max(100),callId:z.string().min(1).max(200),arguments:z.record(z.string(),z.unknown())}).strict(),body)
  419. const lease=this.leases.get(value.leaseId)
  420. if(lease?.runnerId!==record.id)throw new HttpError(410,'工具授权已失效','runner_lease_expired')
  421. lease.input.assertActive()
  422. const fingerprint=hash(JSON.stringify([value.toolId,value.arguments]));let call=lease.calls.get(value.callId)
  423. if(call&&call.fingerprint!==fingerprint)throw new HttpError(409,'重复调用参数不一致','idempotency_conflict')
  424. if(!call){if(lease.calls.size>=128)throw new HttpError(429,'本轮工具调用已达上限','runner_call_limit');call={fingerprint,result:Promise.resolve().then(()=>lease.input.call(value.toolId,value.arguments,value.callId))};lease.calls.set(value.callId,call)}
  425. const result=await call.result;lease.input.assertActive();ctx.body={result}
  426. }
  427. }
  428. close(){clearInterval(this.watchdog);for(const id of [...this.online.keys()])this.disconnect(id)}
  429. }