|
|
@@ -7,6 +7,7 @@ import type {BrowserUpload} from '../runner/browser/types.js'
|
|
|
import {type DesktopEnvironments} from './desktopEnvironments.js'
|
|
|
import {type GrokCloud} from './grokCloud.js'
|
|
|
import {VmToolSession} from './vmComputer.js'
|
|
|
+import {waitForHermesSessionIdle} from './sessionStop.js'
|
|
|
import { randomUUID } from 'node:crypto'
|
|
|
import { setTimeout as delay } from 'node:timers/promises'
|
|
|
import { readFileSync } from 'node:fs'
|
|
|
@@ -127,7 +128,7 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
|
|
|
private operationJournal:ToolOperationJournal
|
|
|
assertCanSubmit:()=>void=()=>{}
|
|
|
get idleForUpdate(): boolean {
|
|
|
- return this.live.size === 0 && this.executing.size === 0 && !this.store.db.prepare("SELECT 1 FROM workspace_entities WHERE kind IN ('run','turn') AND COALESCE(json_extract(data,'$.status'),'unknown') NOT IN ('complete','failed','interrupted') LIMIT 1").get()
|
|
|
+ return this.live.size === 0 && this.executing.size === 0 && !this.store.db.prepare("SELECT 1 FROM workspace_entities WHERE kind IN ('run','turn') AND (COALESCE(json_extract(data,'$.status'),'unknown') NOT IN ('complete','failed','interrupted') OR coalesce(json_extract(data,'$.cleanupPending'),0)=1) LIMIT 1").get()
|
|
|
}
|
|
|
desktopEnvironments?:DesktopEnvironments
|
|
|
cloud?:GrokCloud
|
|
|
@@ -135,6 +136,8 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
|
|
|
computerAvailable?:(owner:string,agent:Agent)=>boolean
|
|
|
inspector?:import('./workspaceInspector.js').WorkspaceInspector
|
|
|
private live = new Map<string, LiveTurn>()
|
|
|
+ private memoryCleanup?: Promise<void>
|
|
|
+ private memoryCleanupRetries=new Map<string,{attempt:number;after:number}>()
|
|
|
readonly teamTools: WorkspaceTeamTools
|
|
|
routineTools?: WorkspaceRoutineTools
|
|
|
readonly tasks: WorkspaceTaskCoordinator
|
|
|
@@ -146,6 +149,7 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
|
|
|
super(store, userActive)
|
|
|
this.operationJournal=new ToolOperationJournal(store)
|
|
|
this.openViking = openViking
|
|
|
+ if(openViking)openViking.onConfigurationChanged=()=>{if(!this.closing)this.wake()}
|
|
|
this.teamTools = new WorkspaceTeamTools(store, nodes, this)
|
|
|
this.tasks = new WorkspaceTaskCoordinator(store, this)
|
|
|
this.knowledge = new WorkspaceKnowledge(store.home, store, openViking)
|
|
|
@@ -160,9 +164,43 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
|
|
|
if (!this.openViking?.enabled) return
|
|
|
await this.openViking.ensureUser(owner, { id, temporaryGoalId: undefined })
|
|
|
}
|
|
|
+ protected override maintenance(): boolean {
|
|
|
+ if(!this.openViking?.enabled)return false
|
|
|
+ const pending=new Map<string,{owner:string;id:string}>()
|
|
|
+ for(const owner of this.store.owners()){
|
|
|
+ for(const item of this.store.list<{id:string;memoryCleanupPending?:boolean}>(owner,'agent-deletion'))
|
|
|
+ if(item.memoryCleanupPending)pending.set(`${owner}:${item.id}`,{owner,id:item.id})
|
|
|
+ for(const binding of this.store.list<{agentId:string;namespace:string;status:string;pendingRemoval?:boolean}>(owner,'openviking-binding'))
|
|
|
+ if(binding.namespace===this.openViking.namespace&&binding.status==='removed'&&binding.pendingRemoval)
|
|
|
+ pending.set(`${owner}:${binding.agentId}`,{owner,id:binding.agentId})
|
|
|
+ }
|
|
|
+ if(!this.memoryCleanup){
|
|
|
+ const next=[...pending.entries()].find(([key])=>(this.memoryCleanupRetries.get(key)?.after??0)<=Date.now())
|
|
|
+ if(next){
|
|
|
+ const [key,{owner,id}]=next
|
|
|
+ this.memoryCleanup=this.cleanupAgent(owner,id).then(()=>{this.memoryCleanupRetries.delete(key)})
|
|
|
+ .catch(error=>{
|
|
|
+ if(this.closing)return
|
|
|
+ const attempt=(this.memoryCleanupRetries.get(key)?.attempt??0)+1
|
|
|
+ this.memoryCleanupRetries.set(key,{attempt,after:Date.now()+Math.min(30_000,500*2**Math.min(attempt-1,6))})
|
|
|
+ const deletion=this.store.get<{id:string;memoryCleanupPending:boolean;cleanupError?:string}>(owner,'agent-deletion',id)
|
|
|
+ if(deletion){deletion.cleanupError=error instanceof Error?error.message:'远端记忆清理失败';this.store.put(owner,'agent-deletion',id,deletion)}
|
|
|
+ }).finally(()=>{this.memoryCleanup=undefined;this.wake()})
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return pending.size>0
|
|
|
+ }
|
|
|
async cleanupAgent(owner: string, id: string): Promise<void> {
|
|
|
- if (!this.openViking) return
|
|
|
- await this.openViking.removeUser(owner, { id, temporaryGoalId: undefined })
|
|
|
+ await this.openViking?.removeUser(owner, { id, temporaryGoalId: undefined })
|
|
|
+ const deletion=this.store.get<{id:string;memoryCleanupPending:boolean;cleanupError?:string}>(owner,'agent-deletion',id)
|
|
|
+ if(deletion){deletion.memoryCleanupPending=this.openViking?.binding(owner,id)?.pendingRemoval===true;delete deletion.cleanupError;this.store.put(owner,'agent-deletion',id,deletion)}
|
|
|
+ }
|
|
|
+ async deleteAgent(owner: string, id: string): Promise<void> {
|
|
|
+ // Commit all local validation/deletion before any destructive remote action.
|
|
|
+ // The durable tombstone also makes a repeated DELETE retry remote cleanup.
|
|
|
+ if(this.store.get(owner,'agent',id))this.store.deleteAgent(owner,id)
|
|
|
+ else if(!this.store.get(owner,'agent-deletion',id))this.store.require(owner,'agent',id)
|
|
|
+ try{await this.cleanupAgent(owner,id)}finally{this.wake()}
|
|
|
}
|
|
|
send(owner: string, conversationId: string, input: unknown): Run {
|
|
|
return this.submit(owner, conversationId, input)
|
|
|
@@ -214,6 +252,7 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
|
|
|
if (projectId) for (const id of source?.targetAgentId ? [source.targetAgentId] : c.memberIds) this.knowledge.requireProjectMember(owner, projectId, id)
|
|
|
const conversationTask = this.store.resolveTask(owner, conversationId, body.taskId),
|
|
|
conversationTaskId = conversationTask?.id
|
|
|
+ if(this.store.hasPendingCleanup(owner,{conversationId,conversationTaskId}))throw new HttpError(409,'原执行尚未确认停止,请等待停止完成后再发消息','conversation_stopping')
|
|
|
if (conversationTaskId) {
|
|
|
const alreadyActive = this.store.list<Run>(owner, 'run').some(run => run.conversationTaskId === conversationTaskId && !['complete', 'failed', 'interrupted'].includes(run.status))
|
|
|
if (!alreadyActive && this.store.activeTaskCount(owner, conversationId) >= WORKSPACE_CONCURRENCY_LIMIT)
|
|
|
@@ -1088,7 +1127,11 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
|
|
|
const key = this.bindingKey(work.conversationId, work.conversationTaskId, work.agentId)
|
|
|
const live = this.live.get(key)
|
|
|
if (live?.taskId === work.id) {
|
|
|
- try { await live.gateway.rpc('session.interrupt', { session_id: live.runtimeId }) }
|
|
|
+ try {
|
|
|
+ await live.gateway.rpc('session.interrupt', { session_id: live.runtimeId })
|
|
|
+ const binding=this.store.require<WorkspaceBinding>(owner,'binding',key)
|
|
|
+ if(binding.execution!=='computer')await waitForHermesSessionIdle((method,params)=>live.gateway.rpc(method,params),live.runtimeId,binding.profile)
|
|
|
+ }
|
|
|
catch (error) { if (!live.gateway.connected) live.done(new Error('已停止')); throw error }
|
|
|
const current = this.getWork(owner, work.id)
|
|
|
if (['complete', 'failed', 'interrupted'].includes(current.status)) {
|
|
|
@@ -1110,6 +1153,7 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
|
|
|
await gateway.connect()
|
|
|
const opened = await gateway.rpc('session.resume', { profile: binding.profile, session_id: binding.storedId, omit_messages: true, close_on_disconnect: false,...(binding.execution==='computer'?{recoverOnly:true}:{}) })
|
|
|
if (opened.running) await gateway.rpc('session.interrupt', { session_id: opened.session_id })
|
|
|
+ if(binding.execution!=='computer')await waitForHermesSessionIdle((method,params)=>gateway.rpc(method,params),String(opened.session_id),binding.profile)
|
|
|
} finally { gateway.close() }
|
|
|
}
|
|
|
const current = this.getWork(owner, work.id)
|