Просмотр исходного кода

继承 Worker 自动压缩配置并保留上下文错误后的虚拟机

Samien 1 месяц назад
Родитель
Сommit
870bfdbde3

+ 10 - 0
docs/local-vm.md

@@ -33,6 +33,16 @@ Web/桌面端的 Bot 电脑面板在选择 **本地虚拟机** 或 **云端 ·
 
 `terminal.cwd` 默认的 `.`(以及 `auto`、`cwd`、未设置)指向虚拟机的持久工作区 `/home/cua/workspace`;相对路径以该工作区为起点,显式绝对路径继续沿用配置。创建、启动和人工接管只读取工作目录,不要求模型可用;停止和移除已有实例也不依赖 Profile 配置。模型仅在运行 Agent 时解析,Profile、目录和模型错误分别提示。
 
+## 上下文压缩与错误恢复
+
+本地 Worker 保留独立执行架构,但会在每次创建或恢复会话时读取基础 Profile 的 `compression` 开关、阈值、保留消息规则、重试与超时设置,以及 `model.context_length` / `max_tokens`。未设置时采用 Hermes 默认值;明确关闭自动压缩的 Profile 继续保持关闭。人工接管后继续当前轮次沿用原配置快照,下一轮再读取最新配置。
+
+压缩完成、工具执行进度和轮次结束都会保存会话历史。失败返回的压缩历史也会保留,下一轮从保存后的历史继续;已结束或已被替换的 Worker 不能用迟到的检查点覆盖新状态。Hermes 在压缩边界刷新工具时仍使用当前轮次由父进程授权的工具清单,不会丢失虚拟机工具或重新发现本机原生工具。
+
+上下文已满且自动压缩关闭,或自动压缩已耗尽重试/超时,属于会话错误:本轮显示失败原因,持久虚拟机释放控制权进入空闲,按配置的空闲时间停止;选择“永不自动停止”时继续运行。取消、授权失效、工具执行状态不确定、Worker 异常退出及未分类的失败仍会停止旧运行实例。错误类型采用 Hermes 返回的结构化标记,不根据模型回复中的“上下文超限”文字判断。
+
+更新 Web/App 的配套 Runner 后生效,不修改 Grok Bot 云端会话路径。验证覆盖真实 Hermes 自动压缩、压缩后续聊和工具清单、压缩关闭/耗尽时的错误分类,以及真实 Docker 实例和文件保留、之后按空闲策略回收。
+
 ## 镜像与持久状态
 
 - 标准桌面沿用现有 XFCE 镜像;Cursor Universal 基于 `public.ecr.aws/k0i0n2g5/cursorenvironments/universal:sand-box-latest` 的固定 amd64 摘要,并添加相同的 CUA 0.20.0 驱动和私有桥接协议。原始镜像不能直接填入现有桌面镜像配置。

+ 29 - 8
src/runner/worker/gateway.ts

@@ -15,13 +15,13 @@ 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 WorkerTool,type WorkerFrame} from './process.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}
 export interface WorkerSession {agentId?:string;id:string;profile:string;ownerKey:string;environmentId:string;cwd:string;configuredCwd:string;history:any[];outcome?:'complete'|'failed'|'uncertain'}
-interface Live {workMarker?:string;paused?:boolean;pauseJob?:Promise<ComputerLease>;segment?:number;toolCalls:Set<Promise<unknown>>;journal:Map<string,{name:string;result:unknown}>;proxy?:ComputerPublicProxy;session:WorkerSession;model:WorkerModel;proxyEnv?:WorkerProxyEnvironment;lease:ComputerLease;controller:AbortController;worker?:HermesWorkerProcess;finishing?:Promise<void>;stopping?:Promise<void>;releasing?:Promise<void>;images:any[];running:boolean;received:boolean}
+interface Live {workMarker?:string;paused?:boolean;pauseJob?:Promise<ComputerLease>;segment?:number;toolCalls:Set<Promise<unknown>>;journal:Map<string,{name:string;result:unknown}>;proxy?:ComputerPublicProxy;session:WorkerSession;model:WorkerModel;contextConfig?:WorkerContextConfiguration;proxyEnv?:WorkerProxyEnvironment;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 text={type:'string'}
 const TOOLS:WorkerTool[]=[
@@ -152,7 +152,7 @@ export class ComputerGateway {
       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,proxyEnv:resolved.proxyEnv,images:[],running:false,received:false,toolCalls:new Set(),journal:new Map()}
+      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
@@ -199,14 +199,16 @@ export class ComputerGateway {
       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,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})
+      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})
@@ -295,15 +297,34 @@ export class ComputerGateway {
     if(this.closed||live.paused||live.controller.signal.aborted)return
     await this.authorized()
     this.runtime.pool.renew(live.lease)
+    const contextErrors:Record<string,string>={
+      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
-    if(success&&Array.isArray(frame.messages)){live.session.history=frame.messages;live.session.outcome='complete'}else live.session.outcome='failed'
+    // 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)
-    await this.release(live,success&&this.runtime.retainDesktops)
+    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
-    this.event('message.complete',{text:String(frame.text??''),status:success?'complete':'failed',...(!success?{error:'隔离 Worker 未完成本轮'}:{})})
+    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 release(live:Live,keepRunning=false){clearInterval(this.timer);live.releasing??=(async()=>{await live.proxy?.close();await this.runtime.pool.release(live.lease,keepRunning)})();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()

+ 77 - 5
src/runner/worker/hermes_worker.py

@@ -6,7 +6,9 @@ Only private stdin/stdout carry protocol frames. Ordinary logging uses stderr.
 """
 from __future__ import annotations
 import contextlib
+import copy
 import json
+import math
 import os
 from pathlib import Path
 import re
@@ -21,6 +23,41 @@ NONCE = BOOT["nonce"]
 PROXY_ENV_KEYS = ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY",
                   "http_proxy", "https_proxy", "all_proxy", "no_proxy")
 
+# Only context-management settings cross the Profile boundary. Plugins, tools,
+# arbitrary environment variables and unrelated credentials remain excluded.
+COMPRESSION_KEYS = {
+    "enabled", "threshold", "threshold_tokens", "target_ratio",
+    "protect_first_n", "protect_last_n", "min_tail_user_messages", "tail_mode",
+    "max_attempts", "abort_on_summary_failure", "in_place", "checkpoint_required",
+    "codex_gpt55_autoraise", "codex_gpt55_autoraise_notice",
+    "codex_responses_native", "codex_responses_compact_threshold",
+    "proactive_prune_tokens", "proactive_prune_min_result_chars",
+    "proactive_prune_min_reclaim_tokens", "micro_compact",
+    "micro_compact_every_n_turns", "micro_compact_defrag_threshold_tokens",
+    "idle_compact_after_seconds", "context_timeout_seconds", "context_total_ceiling_seconds",
+}
+
+def context_configuration(config):
+    raw = config.get("compression", {})
+    if not isinstance(raw, dict):
+        raw = {}
+    compression = {key: value for key, value in raw.items()
+                   if key in COMPRESSION_KEYS
+                   and isinstance(value, (str, bool, int, float))
+                   and (not isinstance(value, float) or math.isfinite(value))}
+    thresholds = raw.get("model_thresholds")
+    if isinstance(thresholds, dict):
+        compression["model_thresholds"] = {
+            key: value for key, value in thresholds.items()
+            if isinstance(key, str) and isinstance(value, (int, float))
+            and not isinstance(value, bool) and math.isfinite(value) and 0 < value <= 1
+        }
+    model = config.get("model", {})
+    limits = {key: model[key] for key in ("context_length", "max_tokens")
+              if isinstance(model, dict) and isinstance(model.get(key), int)
+              and not isinstance(model[key], bool) and model[key] > 0}
+    return {"compression": compression, "model": limits}
+
 def emit(kind, **values):
     with WRITE_LOCK:
         WIRE.write(json.dumps({"nonce": NONCE, "type": kind, **values}, ensure_ascii=False) + "\n")
@@ -95,6 +132,7 @@ def resolve_model():
     if runtime.get("api_mode") not in ("chat_completions", "anthropic_messages", "codex_responses"):
         raise ConfigurationError("computer_model_unsupported")
     emit("resolved", model={key: runtime[key] for key in ("provider", "api_mode", "base_url", "api_key") if key in runtime} | {"model": model["default"]}, **workspace,
+         contextConfig=context_configuration(config),
          **({"proxyEnv": proxy_env} if proxy_env else {}))
 
 
@@ -106,19 +144,45 @@ def run_model():
     os.environ["HERMES_HOME"] = str(home)
     os.environ["HERMES_CWD"] = str(home)
     sys.path.insert(0, str(Path(BOOT["hermesSource"]).resolve()))
-    (home / "config.yaml").write_text("plugins:\n  enabled: []\ncompression:\n  enabled: false\n", encoding="utf-8")
+    context_config = context_configuration(BOOT.get("contextConfig", {}))
+    # JSON is a YAML subset. An absent compression setting uses Hermes' own
+    # defaults, while an explicit Profile false remains false.
+    (home / "config.yaml").write_text(json.dumps({
+        "plugins": {"enabled": []}, **context_config,
+        "model": {**context_config["model"], "default": BOOT["model"]["model"],
+                  **{key: BOOT["model"][key] for key in ("provider", "base_url") if key in BOOT["model"]}},
+    }), encoding="utf-8")
     (home / ".env").touch(mode=0o600)
     # Prevent the installed project's dotenv fallback from importing host secrets.
     import hermes_cli.env_loader as env_loader
     env_loader.load_hermes_dotenv = lambda *args, **kwargs: []
     import run_agent
-    # The registry remains empty even if Hermes refreshes its tool snapshot.
-    run_agent.get_tool_definitions = lambda *args, **kwargs: []
+    import model_tools
+    supplied_tools = [{"type": "function", "function": {"name": item["name"], "description": item["description"], "parameters": item["inputSchema"]}} for item in BOOT["tools"]]
+    # Hermes refreshes tools at compaction boundaries, including through
+    # model_tools directly. Keep the parent-owned catalog on every rebuild;
+    # neither drop the VM tools nor discover native handlers from the host.
+    def closed_tool_definitions(*args, **kwargs):
+        return copy.deepcopy(supplied_tools)
+    run_agent.get_tool_definitions = closed_tool_definitions
+    model_tools.get_tool_definitions = closed_tool_definitions
     pending = {}
     stopped = threading.Event()
     allowed = {item["name"] for item in BOOT["tools"]}
 
     class ClosedAgent(run_agent.AIAgent):
+        def _compress_context(self, *args, **kwargs):
+            result = super()._compress_context(*args, **kwargs)
+            if not stopped.is_set() and isinstance(result, tuple) and result and isinstance(result[0], list):
+                emit("checkpoint", messages=result[0])
+            return result
+
+        def _persist_session(self, messages, conversation_history=None):
+            result = super()._persist_session(messages, conversation_history)
+            if not stopped.is_set():
+                emit("checkpoint", messages=messages)
+            return result
+
         def _execute_tool_calls(self, assistant_message, messages, effective_task_id, api_call_count=0):
             # Intercept before Hermes' sequential, parallel and inline branches.
             # In particular message_agent/delegate_task must never reach native code.
@@ -172,7 +236,7 @@ def run_model():
         max_iterations=32, run_budget_seconds=900,
         stream_delta_callback=lambda text, **kwargs: emit("delta", text=str(text)) if text is not None else None,
         reasoning_callback=lambda text, **kwargs: emit("reasoning", text=str(text)) if text is not None else None)
-    agent.tools = [{"type": "function", "function": {"name": item["name"], "description": item["description"], "parameters": item["inputSchema"]}} for item in BOOT["tools"]]
+    agent.tools = closed_tool_definitions()
     agent.valid_tool_names = set(allowed)
 
     def receive():
@@ -201,7 +265,15 @@ def run_model():
         environment += "Host files and commands are not available through computer tools. Never attempt to execute a host command through them. "
     result = agent.run_conversation(BOOT["prompt"], conversation_history=BOOT.get("history", []), task_id=BOOT["taskId"],
         system_message="You are a controlled Hermes worker. " + environment + "Other supplied application and team tools follow their descriptions. Use only supplied tools and follow the user's task and permission scope.")
-    emit("complete", text=str(result.get("final_response") or ""), messages=result.get("messages", []), interrupted=stopped.is_set(), completed=result.get("completed", True))
+    # These are engine result flags, never inferred from model-generated text.
+    failure_code = ("context_compaction_disabled" if result.get("compaction_disabled") is True
+                    else "context_compaction_failed" if result.get("compression_exhausted") is True
+                    else None)
+    interrupted = stopped.is_set() or result.get("interrupted") is True
+    completed = result.get("completed", True) is not False and not result.get("failed") and failure_code is None and not interrupted
+    emit("complete", text=str(result.get("final_response") or ""), messages=result.get("messages"),
+         interrupted=interrupted, completed=completed,
+         **({"failureCode": failure_code} if failure_code and not interrupted and not pending else {}))
 
 try:
     with contextlib.redirect_stdout(sys.stderr):

+ 1 - 0
src/runner/worker/process.ts

@@ -5,6 +5,7 @@ import {HttpError} from '../../server/errors.js'
 
 export interface WorkerTool {id?:string;name:string;description:string;inputSchema:Record<string,unknown>}
 export interface WorkerModel {provider:string;api_mode:string;base_url?:string;api_key?:string;model:string}
+export interface WorkerContextConfiguration {compression?:Record<string,unknown>;model?:{context_length?:number;max_tokens?:number}}
 const proxyEnvKeys=['HTTP_PROXY','HTTPS_PROXY','ALL_PROXY','NO_PROXY','http_proxy','https_proxy','all_proxy','no_proxy'] as const
 export type WorkerProxyEnvironment=Partial<Record<typeof proxyEnvKeys[number],string>>
 export interface WorkerFrame {type:string;[key:string]:any}

+ 103 - 0
tests/live/workerCompaction.live.test.ts

@@ -0,0 +1,103 @@
+// @vitest-environment node
+import {it,expect,vi} from 'vitest'
+import {createServer} from 'node:http'
+import {DatabaseSync} from 'node:sqlite'
+import {mkdtemp,mkdir,writeFile,readFile,rm} from 'node:fs/promises'
+import {join,resolve} from 'node:path'
+import {tmpdir} from 'node:os'
+import {randomUUID} from 'node:crypto'
+import {ComputerRuntime,ComputerGateway} from '../../src/runner/worker/gateway'
+import {LocalVmImages} from '../../src/runner/computers/localVm'
+
+const available=!!process.env.YAOYAO_HERMES_PYTHON&&!!process.env.YAOYAO_HERMES_SOURCE
+async function fixture(enabled:boolean,docker=false,overflow=!enabled,inPlace=true){
+ const home=await mkdtemp(join(tmpdir(),'yaoyao-context-live-')),profile=join(home,'profile'),db=new DatabaseSync(':memory:')
+ await mkdir(profile);const requests:any[]=[],gateways:ComputerGateway[]=[]
+ const summary='## Task\nContinue the fixture task.\n## Progress\nCOMPACTED_WORKER_HISTORY\n'+'Earlier inspection is complete. Keep the user request and continue from the saved result.\n'.repeat(25)
+ const server=createServer(async(req,res)=>{
+  res.setHeader('Content-Type','application/json')
+  if(req.method==='GET'){res.end(JSON.stringify({data:[{id:'fixture-context-model',context_length:65536}]}));return}
+  let body='';for await(const chunk of req)body+=chunk
+  const input=JSON.parse(body);requests.push(input)
+  const modelCall=Array.isArray(input.tools)&&input.tools.length>0
+  if(overflow&&modelCall){res.statusCode=400;res.end(JSON.stringify({error:{message:'maximum context length is 65536 tokens, requested 80000 tokens',code:'context_length_exceeded',type:'invalid_request_error'}}));return}
+  const content=modelCall?'压缩后继续完成。':summary
+  if(input.stream){res.setHeader('Content-Type','text/event-stream');res.end('data: '+JSON.stringify({id:randomUUID(),object:'chat.completion.chunk',model:'fixture-context-model',choices:[{index:0,delta:{role:'assistant',content},finish_reason:'stop'}],usage:{prompt_tokens:100,completion_tokens:20,total_tokens:120}})+'\n\ndata: [DONE]\n\n')}
+  else res.end(JSON.stringify({id:randomUUID(),object:'chat.completion',model:'fixture-context-model',choices:[{index:0,message:{role:'assistant',content},finish_reason:'stop'}],usage:{prompt_tokens:100,completion_tokens:20,total_tokens:120}}))
+ })
+ await new Promise<void>(done=>server.listen(0,'127.0.0.1',done));const base=`http://127.0.0.1:${(server.address() as {port:number}).port}`
+ await writeFile(join(profile,'config.yaml'),JSON.stringify({model:{default:'fixture-context-model',provider:'custom',base_url:base+'/v1',api_key:'fixture-secret',context_length:65536,max_tokens:4096},compression:{enabled,in_place:inPlace,threshold:0.5,target_ratio:0.2,protect_first_n:0,protect_last_n:2,tail_mode:'legacy',max_attempts:1},terminal:{cwd:'.'}}))
+ const script=join(home,'diagnostic-worker.py')
+ await writeFile(script,(await readFile(resolve('src/runner/worker/hermes_worker.py'),'utf8')).replace('NONCE = BOOT["nonce"]','NONCE = BOOT["nonce"]\nBOOT["diagnostic"] = True'))
+ const runtime=new ComputerRuntime(db,{protocol:1,runnerId:randomUUID(),serverURL:base,hermesURL:base,token:'fixture',allowedProfiles:['default'],artifactRoots:[],computers:{runtime:'docker',imageId:process.env.YAOYAO_COMPUTER_IMAGE??'sha256:'+'a'.repeat(64),python:process.env.YAOYAO_HERMES_PYTHON!,hermesSource:process.env.YAOYAO_HERMES_SOURCE!,hermesHome:profile,network:'none'}},home,script)
+ await runtime.ready;runtime.retainDesktops=true
+ const id=randomUUID(),meta={agentId:id,environmentId:id,ownerKey:'context-live-fixture'}
+ if(!docker){
+  vi.spyOn(runtime.provider,'ensure').mockResolvedValue({id,containerId:id,running:true,workspace:home,isolation:'container'})
+  vi.spyOn(runtime.provider,'stop').mockResolvedValue();vi.spyOn(runtime.provider,'remove').mockResolvedValue()
+ }
+ async function open(sessionId?:string){
+  const gateway=new ComputerGateway(runtime,meta,randomUUID(),async()=>{},async()=>({}));gateways.push(gateway)
+  await gateway.connect();const opened=await gateway.rpc(sessionId?'session.resume':'session.create',{profile:'default',...(sessionId?{session_id:sessionId}:{})})
+  return {gateway,id:opened.session_id as string}
+ }
+ async function send(gateway:ComputerGateway,id:string,prompt:string){
+  let diagnostics=''
+  const completed=new Promise<any>((done,reject)=>{gateway.onEvent=frame=>{if(frame.type==='message.complete')done(frame.payload)};gateway.onDisconnect=()=>reject(new Error('unexpected Worker disconnection\n'+diagnostics))})
+  await gateway.rpc('prompt.submit',{session_id:id,text:prompt})
+  runtime.workers.get(gateway.workId)?.process.child.stderr.on('data',chunk=>{diagnostics+=String(chunk)})
+  return completed
+ }
+ return {runtime,db,meta,requests,open,send,async close(){await Promise.all(gateways.map(g=>g.close()));await runtime.controls.close();await runtime.pool.close();db.close();vi.restoreAllMocks();await new Promise<void>(done=>{server.close(()=>done());server.closeAllConnections()});await rm(home,{recursive:true,force:true})}}
+}
+
+it.skipIf(!available).each([true,false])('inherits Profile compression and resumes compacted history with in_place=%s',async inPlace=>{
+ const f=await fixture(true,false,false,inPlace)
+ try{
+  const seed=await f.open(),session=f.runtime.session(seed.id,f.meta,'default')
+  session.history=Array.from({length:50},(_,i)=>({role:i%2?'assistant':'user',content:`HISTORY_PAYLOAD_${i}: `+'Long earlier fixture investigation with completed tool results. '.repeat(180)}))
+  const initialSize=JSON.stringify(session.history).length;f.runtime.save(session)
+  await seed.gateway.close();const first=await f.open(seed.id)
+  const result=await f.send(first.gateway,first.id,'Continue the pending fixture task.')
+  expect(result).toMatchObject({status:'complete',text:'压缩后继续完成。'})
+  const history=f.runtime.session(first.id,f.meta,'default').history
+  expect(JSON.stringify(history)).toContain('COMPACTED_WORKER_HISTORY')
+  expect(JSON.stringify(history).length).toBeLessThan(initialSize/2)
+  expect(f.requests.some(r=>!r.tools?.length)).toBe(true)
+  await first.gateway.close()
+  const next=await f.open(first.id);const again=await f.send(next.gateway,next.id,'Continue another step from the compacted history.')
+  expect(again.status).toBe('complete')
+  const last=f.requests.filter(r=>r.tools?.length).at(-1)!
+  expect(JSON.stringify(last.messages)).toContain('COMPACTED_WORKER_HISTORY')
+  expect(JSON.stringify(last.messages).length).toBeLessThan(initialSize/2)
+  expect(last.tools.map((t:any)=>t.function.name)).toEqual(expect.arrayContaining(['computer_shell','computer_export']))
+  expect(last.tools.map((t:any)=>t.function.name)).not.toContain('terminal')
+ }finally{await f.close()}
+},120000)
+
+it.skipIf(!available||!process.env.YAOYAO_COMPUTER_IMAGE)('keeps a real Docker desktop running on a typed context error, then honors the idle policy',async()=>{
+ const f=await fixture(false,true),manager=new LocalVmImages(f.runtime,f.db)
+ try{
+  await manager.ready;await manager.idlePolicy(0)
+  const turn=await f.open(),spec=f.runtime.pool.definition(f.meta.ownerKey,f.meta.environmentId)!,before=await f.runtime.provider.inspect(spec)
+  await f.runtime.provider.execute(spec,['/bin/sh','-c','printf context-proof > retained.txt'],{authorize:()=>{}})
+  const result=await f.send(turn.gateway,turn.id,'Trigger the fixture context overflow.')
+  expect(result).toMatchObject({status:'failed',code:'context_compaction_disabled',error:expect.stringContaining('虚拟机已进入空闲')})
+  await turn.gateway.close();await f.runtime.pool.expire()
+  expect(await f.runtime.provider.inspect(spec)).toMatchObject({running:true,containerId:before!.containerId})
+  expect((await f.runtime.provider.execute(spec,['cat','retained.txt'],{authorize:()=>{}})).stdout).toBe('context-proof')
+  await manager.idlePolicy(1)
+  f.db.prepare("UPDATE computer_environments SET value=json_set(value,'$.updatedAt',?)").run(Date.now()-61000)
+  await f.runtime.pool.expire();expect(await f.runtime.provider.inspect(spec)).toBeUndefined()
+ }finally{await manager.close();await f.close()}
+},120000)
+
+it.skipIf(!available)('classifies exhausted automatic compression as a session error without retiring the desktop',async()=>{
+ const f=await fixture(true,false,true)
+ try{
+  const turn=await f.open(),result=await f.send(turn.gateway,turn.id,'The fixture provider always rejects its context size.')
+  expect(result).toMatchObject({status:'failed',code:'context_compaction_failed'})
+  expect(f.runtime.pool.status(f.meta.ownerKey)[0]?.status).toBe('idle')
+  expect(f.runtime.provider.stop).not.toHaveBeenCalled()
+ }finally{await f.close()}
+},120000)

+ 69 - 0
tests/server/workerCompaction.test.ts

@@ -0,0 +1,69 @@
+// @vitest-environment node
+import {expect,it,vi} from 'vitest'
+import {DatabaseSync} from 'node:sqlite'
+import {mkdtemp,writeFile,rm} from 'node:fs/promises'
+import {tmpdir} from 'node:os'
+import {join} from 'node:path'
+import {randomUUID} from 'node:crypto'
+import {ComputerGateway,ComputerRuntime} from '../../src/runner/worker/gateway'
+
+async function fixture(){
+ const home=await mkdtemp(join(tmpdir(),'yaoyao-compaction-')),db=new DatabaseSync(':memory:'),script=join(home,'worker.py')
+ await writeFile(script,`import json,sys
+boot=json.loads(sys.stdin.readline())
+def emit(kind, **data):
+    print(json.dumps({"nonce":boot["nonce"],"type":kind,**data}),flush=True)
+payload=json.loads(boot["prompt"])
+emit("checkpoint",messages=payload.pop("checkpoint",boot["history"]))
+emit("probe",boot=boot)
+emit("complete",**payload)
+for line in sys.stdin:
+    if json.loads(line).get("type")=="interrupt":break
+`)
+ const runtime=new ComputerRuntime(db,{protocol:1,runnerId:randomUUID(),serverURL:'http://127.0.0.1',hermesURL:'http://127.0.0.1',token:'fixture',allowedProfiles:['default'],artifactRoots:[],computers:{runtime:'docker',imageId:'sha256:'+'a'.repeat(64),python:'python3',hermesSource:home,hermesHome:home,network:'none'}},home,script)
+ await runtime.ready;runtime.retainDesktops=true
+ const id=randomUUID(),meta={environmentId:id,agentId:id,ownerKey:'owner'},gateways:ComputerGateway[]=[]
+ vi.spyOn(runtime,'resolve').mockResolvedValue({type:'resolved',cwd:'/home/cua/workspace',configuredCwd:'.',model:{provider:'custom',api_mode:'chat_completions',model:'fixture'},contextConfig:{compression:{enabled:true,threshold:0.5}}})
+ vi.spyOn(runtime.provider,'ensure').mockResolvedValue({id,containerId:'fixture',running:true,workspace:home,isolation:'container'})
+ const stop=vi.spyOn(runtime.provider,'stop').mockResolvedValue(),remove=vi.spyOn(runtime.provider,'remove').mockResolvedValue()
+ async function turn(payload:Record<string,unknown>,sessionId?:string){
+  const gateway=new ComputerGateway(runtime,meta,randomUUID(),async()=>{},async()=>({}));gateways.push(gateway)
+  const completed=new Promise<any>((resolve,reject)=>{gateway.onEvent=frame=>{if(frame.type==='message.complete')resolve(frame.payload)};gateway.onDisconnect=()=>reject(new Error('unexpected disconnection'))})
+  await gateway.connect();const opened=await gateway.rpc(sessionId?'session.resume':'session.create',{profile:'default',...(sessionId?{session_id:sessionId}:{})})
+  await gateway.rpc('prompt.submit',{session_id:opened.session_id,text:JSON.stringify(payload)})
+  const worker=runtime.workers.get(gateway.workId)!.process,probe=worker.wait('probe',5000)
+  return {gateway,sessionId:opened.session_id,result:await completed,boot:(await probe).boot,worker}
+ }
+ return {db,runtime,meta,stop,remove,turn,async close(){await Promise.all(gateways.map(g=>g.close()));await runtime.controls.close();await runtime.pool.close();db.close();vi.restoreAllMocks();await rm(home,{recursive:true,force:true})}}
+}
+
+it.each(['context_compaction_disabled','context_compaction_failed'])('keeps a desktop idle after %s and resumes the saved compacted history',async failureCode=>{
+ const f=await fixture(),checkpoint=[{role:'system',content:'compacted checkpoint'}],history=[...checkpoint,{role:'user',content:'pending task'}]
+ try{
+  const first=await f.turn({completed:false,failureCode,checkpoint,messages:history,text:'engine context failure'})
+  expect(first.result).toMatchObject({status:'failed',code:failureCode,error:expect.stringContaining('虚拟机已进入空闲')})
+  expect(f.runtime.pool.status('owner')[0]?.status).toBe('idle');expect(f.stop).not.toHaveBeenCalled();expect(f.remove).not.toHaveBeenCalled()
+  expect(f.runtime.session(first.sessionId,f.meta,'default')).toMatchObject({history,outcome:'failed'})
+  first.worker.onEvent({type:'checkpoint',messages:[{role:'user',content:'late frame'}]})
+  expect(f.runtime.session(first.sessionId,f.meta,'default').history).toEqual(history)
+  await first.gateway.close();expect(f.stop).not.toHaveBeenCalled()
+  const nextHistory=[...history,{role:'assistant',content:'continued'}],next=await f.turn({completed:true,messages:nextHistory,text:'continued'},first.sessionId)
+  expect(next.boot.history).toEqual(history);expect(next.boot.contextConfig.compression.enabled).toBe(true)
+  expect(f.runtime.session(first.sessionId,f.meta,'default')).toMatchObject({history:nextHistory,outcome:'complete'})
+ }finally{await f.close()}
+})
+
+it.each([
+ {completed:false,text:'Context overflow and auto-compaction is disabled'},
+ {completed:false,failureCode:'unknown_failure'},
+ {completed:false,failureCode:'context_compaction_failed',interrupted:true},
+])('stops for unclassified or interrupted failures: %j',async payload=>{
+ const f=await fixture()
+ try{
+  const checkpoint=[{role:'system',content:'latest compacted history'}],result=await f.turn({...payload,checkpoint})
+  expect(result.result.status).toBe('failed');expect(result.result).not.toHaveProperty('code')
+  expect(f.stop).toHaveBeenCalledOnce();expect(f.remove).toHaveBeenCalledOnce()
+  expect(f.runtime.pool.status('owner')[0]?.status).toBe('free')
+  expect(f.runtime.session(result.sessionId,f.meta,'default').history).toEqual(checkpoint)
+ }finally{await f.close()}
+})

+ 10 - 2
tests/server/workerProxy.test.ts

@@ -23,6 +23,7 @@ for line in sys.stdin:
  const first={HTTPS_PROXY:'http://user:first-private-proxy@127.0.0.1:7890',NO_PROXY:'localhost'}
  const second={https_proxy:'http://user:second-private-proxy@127.0.0.1:8890',no_proxy:'internal.test'}
  let current:WorkerProxyEnvironment|undefined=first
+ let contextConfig={compression:{enabled:true,threshold:0.5}}
  const createGateway=()=>{
   const gateway=new ComputerGateway(runtime,meta,randomUUID(),async()=>{},async()=>({}))
   gateway.onEvent=frame=>events.push(frame);gateways.push(gateway);return gateway
@@ -30,7 +31,7 @@ for line in sys.stdin:
  const probe=(gateway:ComputerGateway)=>runtime.workers.get(gateway.workId)!.process.wait('probe',5000)
  try{
   await runtime.ready
-  const resolver=vi.spyOn(runtime,'resolve').mockImplementation(async()=>({type:'resolved',cwd:'/home/cua/workspace',configuredCwd:'.',model:{provider:'custom',api_mode:'chat_completions',model:'fixture'},...(current?{proxyEnv:{...current}}:{})}))
+  const resolver=vi.spyOn(runtime,'resolve').mockImplementation(async()=>({type:'resolved',cwd:'/home/cua/workspace',configuredCwd:'.',model:{provider:'custom',api_mode:'chat_completions',model:'fixture'},contextConfig:structuredClone(contextConfig),...(current?{proxyEnv:{...current}}:{})}))
   const state={id,containerId:'fixture',running:true,workspace:home,isolation:'container' as const}
   const ensure=vi.spyOn(runtime.provider,'ensure').mockResolvedValue(state)
   vi.spyOn(runtime.provider,'inspect').mockResolvedValue(state)
@@ -39,11 +40,16 @@ for line in sys.stdin:
   const created=await gateway.rpc('session.create',{profile:'default'})
   await gateway.rpc('prompt.submit',{session_id:created.session_id,text:'first turn'})
   expect((await probe(gateway)).env).toEqual(first)
+  const oldWorker=runtime.workers.get(gateway.workId)!.process
   current=second
+  contextConfig={compression:{enabled:false,threshold:0.7}}
   await gateway.takeControl(randomUUID(),()=>{})
   await gateway.giveBack('continue')
   const continued=await probe(gateway)
   expect(continued.env).toEqual(first)
+  expect(continued.boot.contextConfig).toEqual({compression:{enabled:true,threshold:0.5}})
+  oldWorker.onEvent({type:'checkpoint',messages:[{role:'user',content:'stale worker history'}]})
+  expect(JSON.stringify(runtime.session(created.session_id,meta,'default').history)).not.toContain('stale worker history')
   expect(continued.boot.history).toContainEqual({role:'user',content:'first turn'})
   expect(resolver).toHaveBeenCalledTimes(1)
   await gateway.close()
@@ -51,7 +57,9 @@ for line in sys.stdin:
   const resumed=createGateway();await resumed.connect()
   expect((await resumed.rpc('session.resume',{profile:'default',session_id:created.session_id})).session_id).toBe(created.session_id)
   await resumed.rpc('prompt.submit',{session_id:created.session_id,text:'next turn'})
-  expect((await probe(resumed)).env).toEqual(second)
+  const resumedProbe=await probe(resumed)
+  expect(resumedProbe.env).toEqual(second)
+  expect(resumedProbe.boot.contextConfig).toEqual(contextConfig)
   expect(resolver).toHaveBeenCalledTimes(2)
   await resumed.close()
 

+ 11 - 0
tests/server/workerResolution.test.ts

@@ -98,3 +98,14 @@ it('starts Python with only explicit proxy keys and never inherits Runner secret
   expect(frame.boot).not.toHaveProperty('proxyEnv')
  }
 })
+it('inherits only compression controls and model limits, preserving explicit disable and Hermes defaults',async()=>{
+ await writeFile(join(source,'hermes_cli','runtime_provider.py'),'def resolve_runtime_provider(**kwargs): return {"provider":"custom","api_mode":"chat_completions","api_key":"private-model-key"}\n')
+ const model={default:'fixture',provider:'custom',context_length:32768,max_tokens:4096,api_key:'private-config-key'}
+ await writeFile(join(home,'config.yaml'),JSON.stringify({model,compression:{enabled:true,threshold:0.65,target_ratio:0.25,protect_last_n:8,in_place:false,context_timeout_seconds:60,model_thresholds:{fixture:0.75},api_key:'private-other-key'},plugins:{enabled:['unsafe']},terminal:{cwd:'.'}}))
+ expect((await read('resolve')).contextConfig).toEqual({compression:{enabled:true,threshold:0.65,target_ratio:0.25,protect_last_n:8,in_place:false,context_timeout_seconds:60,model_thresholds:{fixture:0.75}},model:{context_length:32768,max_tokens:4096}})
+ await writeFile(join(home,'config.yaml'),JSON.stringify({model,compression:{enabled:false,threshold:0.5}}))
+ expect((await read('resolve')).contextConfig.compression).toEqual({enabled:false,threshold:0.5})
+ await writeFile(join(home,'config.yaml'),JSON.stringify({model:{default:'fixture'}}))
+ expect((await read('resolve')).contextConfig).toEqual({compression:{},model:{}})
+ expect(await read('resolve-workspace')).not.toHaveProperty('contextConfig')
+})