fix(runtime): release 2.7.1 with consistent ADB and failure diagnostics

.env.example

@@ -9,6 +9,10 @@ OPENAI_TIMEOUT_SECONDS=300
 # 模型请求和模型凭据检查默认直连；设为 true 才读取 Windows/HTTP(S)_PROXY/ALL_PROXY/NO_PROXY。
 # 仅环境配置，不显示在网页中；修改后重启服务生效。TLS 校验及自定义 CA 不受影响。
 OPENAI_USE_SYSTEM_PROXY=false
+# 默认 websocket，要求网关支持 Responses WebSocket；不支持时可在网页切回 http。
+OPENAI_TRANSPORT=websocket
+# 内部迁移标记。升级时一次性将旧 HTTP 配置切换为 WebSocket；保留此标记以记住后续手动选择。
+ARTEMIS_TRANSPORT_DEFAULT_VERSION=1
 # 可选的后台步骤截图摘要单次超时，独立于上面的主模型请求超时。
 ARTEMIS_STEP_SUMMARIZER_TIMEOUT_SECONDS=120
 # 失败后最多重试 5 次，由 Artemis 统一管理，SDK 不叠加重试。

README.md

@@ -2,7 +2,7 @@
 
 基于 [ARTEMIS](https://github.com/google/artemis) 扩展的 Android AI 自动化测试平台。用自然语言描述任务，即可通过简体中文网页或 MCP 驱动真机、模拟器执行操作并收集结果。模型统一使用 **OpenAI Responses API**。
 
-**2.6.0**：新增 Android 原生控件读写、条件等待、容器滚动、双指缩放与连续多指手势；支持状态确保、稳定数值反馈与无效动作保护，修复多窗口原生查询和选择器像素边界，并补齐前端中文动作名称。见 [2.6.0 发行说明](docs/releases/2.6.0.md)。包含 [2.5.0 的动作序列与缓存统计](docs/releases/2.5.0.md)及 [2.4.5 的插件开发 SDK](docs/releases/2.4.5.md)；SDK 与插件继续独立版本化，升级主程序不会自动更新或授权插件。
+**2.7.1**：统一任务操作与录屏使用的 ADB，修复模型空结果引起的应用查找误报，并在任务资料 ZIP 中保留运行环境诊断和批次原暂停原因。见 [2.7.1 发行说明](docs/releases/2.7.1.md)。继续包含 [2.7.0 的 Responses WebSocket 增量传输](docs/releases/2.7.0.md)、[2.6.0 的 Android 原生控件与多指手势](docs/releases/2.6.0.md)、[2.5.0 的动作序列与缓存统计](docs/releases/2.5.0.md)及 [2.4.5 的插件开发 SDK](docs/releases/2.4.5.md)；SDK 与插件继续独立版本化，升级主程序不会自动更新或授权插件。
 
 ## 主要功能
 
@@ -61,6 +61,8 @@ cd Artemis-Eapil
 
 也可在项目 `.env` 中设置 `OPENAI_API_KEY`、`OPENAI_BASE_URL` 和 `OPENAI_MODEL`，其他参数见 [.env.example](.env.example)。新安装默认接口地址为 `https://api-n-cd.playones.com/v1`；已保存的地址不变。直接编辑环境文件后需重启相应服务。不要提交自己的密钥。
 
+模型传输默认 **WebSocket（增量传输）**。升级后首次启动会将旧的全局 HTTP 配置一并迁移为 WebSocket；之后仍可在“模型配置 → 传输协议”手动切回 HTTP，后续启动保留手动选择。Flash、Pro、探索及辅助模型的异步请求复用 Responses 连接，历史一致时只上传新增输入；网关需支持 `/v1/responses` 的 WebSocket 升级与响应链。详细行为与流量诊断见[传输说明](docs/ai-automation.md#responses-websocket-增量传输)。
+
 “功能开关”默认将任务截图长边限制为 2048 像素；关闭后保留设备原始分辨率。可在 [config/artemis.jsonc](config/artemis.jsonc) 的 `agent.screenshot` 中覆盖 `enabled` 和 `max_long_edge`，重启后对新任务生效。
 
 首页“功能配置”提供“Flash 最终 AI 判定与分析”，默认关闭。切换后自动保存并提示结果，失败恢复原值；无需重启服务。开启后，Flash 在执行结束时根据用例需求和执行步骤进行独立复核；关闭后显示“执行完成 · 未做 AI 判定”，保留执行步骤、截图和报告，不自动记为测试通过。此设置适用于所有 Flash 任务，包括批量、精选、MCP/SDK 和探索提炼用例的 Flash 复跑；保存后对尚未开始的任务生效，运行中的任务保持原设置。Pro 和主动探索不受影响。其余功能配置仍需点击底部保存按钮。

apps/admin_console/services/automation_service.py

@@ -123,9 +123,18 @@ def recover_batches():
         if batch["status"] in {"running", "pausing", "paused"}:
             if any(str(job["session_id"]) in live_sessions for job in store.jobs(batch["id"])):
                 continue
+            recovery_reason = "服务已重启；请检查中断任务，点击继续执行未开始任务"
+            batch.setdefault("recovery_events", []).append(
+                {
+                    "message": recovery_reason,
+                    "recovered_at": time.time(),
+                    "previous_reason": batch.get("reason"),
+                    "previous_pause_source": batch.get("pause_source"),
+                }
+            )
             batch.update(
                 status="paused",
-                reason="服务已重启；请检查中断任务，点击继续执行未开始任务",
+                reason=batch.get("reason") or recovery_reason,
                 pause_source="recovery",
             )
             for job in store.jobs(batch["id"]):
@@ -330,7 +339,7 @@ async def tick_batches():
             else "无法确认所选设备在线且已授权，请恢复设备连接后继续批次"
         )
         if rejection:
-            batch.update(status="paused", reason=str(rejection))
+            batch.update(status="paused", reason=str(rejection), pause_source="device")
             store.put("batch", batch["id"], batch)
             continue
         if (store.get("batch", batch["id"]) or {}).get("status") != "running":

apps/admin_console/services/model_settings_service.py

@@ -1,20 +1,18 @@
 """Persist the single Responses endpoint without returning credentials to the browser."""
 
-import os
-import tempfile
-import threading
-from pathlib import Path
 from typing import Literal
 from urllib.parse import urlsplit
 
 from pydantic import BaseModel, Field, SecretStr, field_validator
 
 from artemis.config.agent import load_agent_config
+from artemis.config.environment import (
+    persist_environment as write_environment,
+    TRANSPORT_DEFAULT_VERSION_KEY,
+)
 from artemis.config.paths import get_env_file
 from artemis.config.settings import settings
 
-_write_lock = threading.Lock()
-
 
 class ModelSettingsUpdate(BaseModel):
     base_url: str = "https://api-n-cd.playones.com/v1"
@@ -22,6 +20,7 @@ class ModelSettingsUpdate(BaseModel):
     api_key: SecretStr | None = None  # Omitted/blank keeps the existing key.
     timeout_seconds: float = Field(default=300, gt=0, le=3600)
     max_retries: int = Field(default=5, ge=0, le=10)
+    transport: Literal["http", "websocket"] | None = None
     retry_interval_seconds: float = Field(default=10, ge=0, le=300)
     rate_limit_retry_seconds: float | None = Field(default=None, ge=0, le=3600)
     tool_fallback_model: str | None = Field(default=None, min_length=1, max_length=200)
@@ -72,6 +71,7 @@ def get_model_settings() -> dict:
         "api_key_configured": bool(settings.OPENAI_API_KEY),
         "timeout_seconds": settings.OPENAI_TIMEOUT_SECONDS,
         "max_retries": settings.OPENAI_MAX_RETRIES,
+        "transport": settings.OPENAI_TRANSPORT,
         "retry_interval_seconds": settings.OPENAI_RETRY_INTERVAL_SECONDS,
         "rate_limit_retry_seconds": settings.OPENAI_RATE_LIMIT_RETRY_SECONDS,
         "tool_fallback_model": settings.OPENAI_TOOL_FALLBACK_MODEL,
@@ -102,6 +102,7 @@ def save_model_settings(update: ModelSettingsUpdate) -> dict:
         "OPENAI_REASONING_EFFORT": update.reasoning_effort,
     }
     optional_settings = {
+        "transport": ("OPENAI_TRANSPORT", str),
         "rate_limit_retry_seconds": ("OPENAI_RATE_LIMIT_RETRY_SECONDS", float),
         "tool_fallback_model": ("OPENAI_TOOL_FALLBACK_MODEL", str),
         "tool_fallback_reasoning_effort": ("OPENAI_TOOL_FALLBACK_REASONING_EFFORT", str),
@@ -115,6 +116,8 @@ def save_model_settings(update: ModelSettingsUpdate) -> dict:
         value = getattr(update, field_name)
         if value is not None:
             values[env_name] = str(value)
+    if update.transport is not None:
+        values[TRANSPORT_DEFAULT_VERSION_KEY] = "1"
     if update.api_key and update.api_key.get_secret_value().strip():
         values["OPENAI_API_KEY"] = update.api_key.get_secret_value().strip()
     persist_environment(values)
@@ -139,31 +142,7 @@ def save_model_settings(update: ModelSettingsUpdate) -> dict:
 
 
 def persist_environment(values: dict[str, str]) -> None:
-    with _write_lock:
-        env_path = get_env_file()
-        env_path.parent.mkdir(parents=True, exist_ok=True)
-        original = env_path.read_text(encoding="utf-8") if env_path.exists() else ""
-        # Preserve comments and unrelated variables; quote values to avoid dotenv interpolation.
-        kept = [
-            line
-            for line in original.splitlines()
-            if line.removeprefix("export ").split("=", 1)[0].strip() not in values
-        ]
-        for key, value in values.items():
-            escaped = value.replace("\\", "\\\\").replace("'", "\\'")
-            kept.append(f"{key}='{escaped}'")
-        content = "\n".join(kept) + "\n"
-        fd, temp_name = tempfile.mkstemp(prefix=".env-", dir=env_path.parent)
-        try:
-            with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as stream:
-                stream.write(content)
-                stream.flush()
-                os.fsync(stream.fileno())
-            os.replace(temp_name, env_path)
-        finally:
-            Path(temp_name).unlink(missing_ok=True)
-        for key, value in values.items():
-            os.environ[key] = value
+    write_environment(get_env_file(), values)
 
 
 class CapabilitySettings(BaseModel):
@@ -261,10 +240,19 @@ async def test_model_settings(update: ModelSettingsUpdate) -> dict:
         api_base=update.base_url,
         timeout_seconds=update.timeout_seconds,
         max_retries=0,
+        transport=update.transport or settings.OPENAI_TRANSPORT,
         reasoning_effort=None if update.reasoning_effort == "default" else update.reasoning_effort,
     )
     model = ModelFactory.create_model(endpoint)
-    result = await model.ainvoke([HumanMessage(content="Reply with OK only.")])
+    try:
+        result = await model.ainvoke([HumanMessage(content="Reply with OK only.")])
+    finally:
+        await model.http_async_client.aclose()
+        model.http_client.close()
     if not result.content:
         raise ValueError("Responses 接口返回空内容，请检查模型配置")
-    return {"success": True, "message": "Responses 连接测试成功", "model": update.model}
+    return {
+        "success": True,
+        "message": f"Responses {endpoint.transport.upper()} 连接测试成功",
+        "model": update.model,
+    }

apps/admin_console/services/report_export.py

@@ -36,7 +36,7 @@ from artemis.runtime.case_result import RESULT_LABELS, resolve_case_result
 from apps.admin_console.services.case_review import reviewed_result
 
 EXPORT_ROOT = WORKSPACE_ROOT / "artifacts" / "reports"
-BATCH_ARCHIVE_SIGNATURE = "artemis-batch-layout-3"
+BATCH_ARCHIVE_SIGNATURE = "artemis-batch-layout-4"
 
 
 def batch_archive_signature(batch_id: str | None = None) -> str:
@@ -197,6 +197,7 @@ def session_report(session_id: str, db_path=None) -> dict:
 
     identifier(session_id)
     meta = store.job(session_id) or {}
+    batch = store.get("batch", meta["batch_id"]) if meta.get("batch_id") else None
     repo = SessionRepository(db_path) if db_path else SessionRepository()
     session = repo.get_session_by_id(session_id) or {}
     if not session and not meta:
@@ -270,6 +271,15 @@ def session_report(session_id: str, db_path=None) -> dict:
         "original_case": case.get("original_case"),
         "task_config": case.get("task_config") or {},
         "task_extensions": task_extensions.report_sections(session_id, store),
+        "runtime_diagnostics": read_json(directory / "runtime_diagnostics.json"),
+        "batch_diagnostics": (
+            {
+                key: batch.get(key)
+                for key in ("id", "status", "reason", "pause_source", "recovery_events")
+            }
+            if batch
+            else None
+        ),
     }
 
 
@@ -1081,6 +1091,17 @@ def export_session_media(session_id: str, kind: str, *, report: dict | None = No
                 )
             if kind == "all":
                 archive.writestr(prefix + "模型请求诊断.json", diagnostics or "{}")
+                archive.writestr(
+                    prefix + "运行环境诊断.json",
+                    json.dumps(
+                        {
+                            "runtime": report.get("runtime_diagnostics"),
+                            "batch": report.get("batch_diagnostics"),
+                        },
+                        ensure_ascii=False,
+                        indent=2,
+                    ),
+                )
                 archive.writestr(
                     prefix + "证据索引.json",
                     json.dumps(
@@ -1202,6 +1223,17 @@ def export_batch(batch_id: str, progress: Callable[[int, int, str], None] | None
                         compress_type=compression(source),
                     )
                 archive.writestr(f"{prefix}/模型请求诊断.json", diagnostics_json(report["id"]))
+                archive.writestr(
+                    f"{prefix}/运行环境诊断.json",
+                    json.dumps(
+                        {
+                            "runtime": report.get("runtime_diagnostics"),
+                            "batch": report.get("batch_diagnostics"),
+                        },
+                        ensure_ascii=False,
+                        indent=2,
+                    ),
+                )
                 archive.writestr(
                     f"{prefix}/证据索引.json",
                     json.dumps(

apps/showcase_ui/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "artemis-showcase-ui",
-  "version": "2.6.0",
+  "version": "2.7.1",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "artemis-showcase-ui",
-      "version": "2.6.0",
+      "version": "2.7.1",
       "dependencies": {
         "@angular/cdk": "^22.2.4",
         "@angular/common": "^22.2.3",

apps/showcase_ui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "artemis-showcase-ui",
-  "version": "2.6.0",
+  "version": "2.7.1",
   "scripts": {
     "ng": "ng",
     "start": "ng serve --proxy-config proxy.conf.json",

apps/showcase_ui/src/app/components/model-settings/model-settings.component.spec.ts

@@ -30,6 +30,31 @@ describe('持久化模型与功能设置', () => {
     http.expectNone('/api/system/model-settings');
   }));
 
+  it('缺少协议的配置默认 WebSocket，允许手动切回 HTTP', fakeAsync(() => {
+    const fixture = TestBed.createComponent(ModelSettingsComponent);
+    fixture.detectChanges();
+    http.expectOne('/api/system/model-settings').flush(config);
+    tick();
+    fixture.detectChanges();
+    const component = fixture.componentInstance;
+    expect(component.transport).toBe('websocket');
+    const selector: HTMLSelectElement = fixture.nativeElement.querySelector('#model-transport');
+    selector.value = 'http';
+    selector.dispatchEvent(new Event('change'));
+    tick();
+    fixture.detectChanges();
+    component.test();
+    const check = http.expectOne('/api/system/model-settings/test');
+    expect(check.request.body.transport).toBe('http');
+    check.flush({message: 'Responses HTTP 连接测试成功'});
+    component.save();
+    const save = http.expectOne('/api/system/model-settings');
+    expect(save.request.body.transport).toBe('http');
+    expect(save.request.body.api_key).toBeUndefined();
+    save.flush({...config, transport: 'http'});
+    expect(component.transport).toBe('http');
+  }));
+
   it('加载配置后保留密钥，保存新模型且不发送空密钥', () => {
     const fixture = TestBed.createComponent(ModelSettingsComponent);
     fixture.detectChanges();

apps/showcase_ui/src/app/components/model-settings/model-settings.component.ts

@@ -8,6 +8,7 @@ import { ModelSelectorComponent } from '../model-selector/model-selector.compone
 import { MODEL_PRESETS } from '../../core/model-presets';
 
 interface ModelSettings {
+  transport?: 'http' | 'websocket';
   base_url: string;
   model: string;
   api_key_configured: boolean;
@@ -36,6 +37,12 @@ interface ModelSettings {
       <input id="model-base-url" name="baseUrl" type="url" [(ngModel)]="baseUrl" required
         placeholder="https://api-n-cd.playones.com/v1" [disabled]="busy() || loading()" />
       <small>填写接口根地址，通常包含 /v1，无需添加 /responses。</small>
+      <label for="model-transport">传输协议</label>
+      <select id="model-transport" name="transport" [(ngModel)]="transport" [disabled]="busy() || loading()">
+        <option value="http">HTTP（兼容模式）</option>
+        <option value="websocket">WebSocket（增量传输）</option>
+      </select>
+      <small>WebSocket 可减少重复历史与截图的上传，需要接口支持。保存后对新任务生效。</small>
       <app-model-selector label="模型名称" [(value)]="model" [presets]="modelPresets" [disabled]="!!busy() || loading()" />
       <small>选择预设模型，或选择“自定义模型”填写名称。</small>
       <app-setting-slider label="思考强度" [(value)]="reasoning" [values]="reasoningValues" [labels]="reasoningLabels" [disabled]="!!busy() || loading()" hint="更高的强度适合复杂推理；需接口支持，也可选择模型默认。" />
@@ -102,6 +109,7 @@ interface ModelSettings {
   `]
 })
 export class ModelSettingsComponent implements OnInit {
+  transport: 'http' | 'websocket' = 'websocket';
   readonly reasoningValues = REASONING_VALUES;
   readonly reasoningLabels = REASONING_LABELS;
   private http = inject(HttpClient);
@@ -128,7 +136,7 @@ export class ModelSettingsComponent implements OnInit {
       && Number.isInteger(this.toolFallbackNoProgress) && this.toolFallbackNoProgress >= 1 && this.toolFallbackNoProgress <= 10;
   }
   private payload() {
-    return { base_url: this.baseUrl.trim(), model: this.model.trim(), api_key: this.apiKey.trim() || undefined,
+    return { base_url: this.baseUrl.trim(), transport: this.transport, model: this.model.trim(), api_key: this.apiKey.trim() || undefined,
       timeout_seconds: this.timeout, max_retries: this.retries, retry_interval_seconds: this.retryInterval, reasoning_effort: this.reasoning,
       rate_limit_retry_seconds: this.rateLimitRetry, tool_fallback_model: this.toolFallbackModel.trim(),
       tool_fallback_reasoning_effort: this.toolFallbackReasoning, tool_fallback_max_retries: this.toolFallbackRetries,
@@ -136,6 +144,7 @@ export class ModelSettingsComponent implements OnInit {
   }
   private apply(config: ModelSettings): void {
     this.baseUrl = config.base_url; this.model = config.model; this.timeout = config.timeout_seconds;
+    this.transport = config.transport ?? 'websocket';
     this.modelPresets = config.model_presets;
     this.retries = config.max_retries; this.retryInterval = config.retry_interval_seconds ?? 10; this.reasoning = config.reasoning_effort ?? 'high'; this.keyConfigured.set(config.api_key_configured); this.envPath.set(config.env_path);
     this.rateLimitRetry = config.rate_limit_retry_seconds ?? 60;

artemis/agents/hopper/hopper.py

@@ -23,11 +23,15 @@ from artemis.context import ArtemisContext
 from artemis.data_engine.trace import trace
 from artemis.services.llm import get_llm, invoke_llm_with_timeout_message, with_fallback
 from artemis.utils.logger import get_logger
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 logger = get_logger(__name__)
 
 
+class HopperError(RuntimeError):
+    """Extraction failed; this says nothing about whether the requested data exists."""
+
+
 class _CyFunctionDetectorMeta(type):
     def __instancecheck__(self, instance):
         name = type(instance).__name__
@@ -55,6 +59,12 @@ class HopperOutput(BaseModel):
         + " and how you decided what to extract."
     )
 
+    @model_validator(mode="after")
+    def validate_found_output(self):
+        if self.found and (not self.output or not self.output.strip()):
+            raise ValueError("A successful extraction must contain a nonempty output")
+        return self
+
 
 @trace(type="agent", name="hopper")
 async def hopper(
@@ -72,25 +82,22 @@ async def hopper(
         HumanMessage(content=f"{request}\nHere is the data you must dig:\n{data}"),
     ]
 
-    llm = get_llm(ctx=ctx, name="hopper", is_utils=True).with_structured_output(HopperOutput)
+    async def extract(model) -> HopperOutput:
+        response = await invoke_llm_with_timeout_message(model.ainvoke(messages))
+        return HopperOutput.model_validate(response)
+
     try:
+        llm = get_llm(ctx=ctx, name="hopper", is_utils=True).with_structured_output(HopperOutput)
         if use_fallback:
             llm_fallback = get_llm(
                 ctx=ctx, name="hopper", is_utils=True, use_fallback=True
             ).with_structured_output(HopperOutput)
-            response: HopperOutput = await with_fallback(
-                main_call=lambda: invoke_llm_with_timeout_message(llm.ainvoke(messages)),
-                fallback_call=lambda: invoke_llm_with_timeout_message(
-                    llm_fallback.ainvoke(messages)
-                ),
-            )  # type: ignore
+            return await with_fallback(
+                main_call=lambda: extract(llm),
+                fallback_call=lambda: extract(llm_fallback),
+            )
         else:
-            response: HopperOutput = await invoke_llm_with_timeout_message(llm.ainvoke(messages))
-        return response
+            return await extract(llm)
     except Exception as e:
         logger.error(f"Hopper LLM invocation failed: {e}")
-        return HopperOutput(
-            found=False,
-            output=None,
-            reason=f"Failed due to LLM error: {str(e)[:500]}",
-        )
+        raise HopperError("模型未能完成数据提取，请查看模型请求诊断。") from e

artemis/config/environment.py

@@ -0,0 +1,64 @@
+"""Atomic environment persistence and the one-time WebSocket default migration."""
+
+import os
+from pathlib import Path
+import tempfile
+
+from dotenv import dotenv_values
+from filelock import FileLock
+
+TRANSPORT_DEFAULT_VERSION_KEY = "ARTEMIS_TRANSPORT_DEFAULT_VERSION"
+
+
+def _lock(path: Path) -> FileLock:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    return FileLock(str(path) + ".artemis.lock", timeout=5)
+
+
+def _write(path: Path, values: dict[str, str]) -> None:
+    original = path.read_text(encoding="utf-8") if path.exists() else ""
+    kept = [
+        line
+        for line in original.splitlines()
+        if line.strip().removeprefix("export ").split("=", 1)[0].strip() not in values
+    ]
+    for key, value in values.items():
+        escaped = value.replace("\\", "\\\\").replace("'", "\\'")
+        kept.append(f"{key}='{escaped}'")
+    fd, temp_name = tempfile.mkstemp(prefix=".env-", dir=path.parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as stream:
+            stream.write("\n".join(kept) + "\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temp_name, path)
+    finally:
+        Path(temp_name).unlink(missing_ok=True)
+
+
+def persist_environment(path: Path, values: dict[str, str]) -> None:
+    with _lock(path):
+        _write(path, values)
+        os.environ.update(values)
+
+
+def _transport_default_is_current(path: Path) -> bool:
+    saved = dotenv_values(path, interpolate=False) if path.exists() else {}
+    try:
+        return int(saved.get(TRANSPORT_DEFAULT_VERSION_KEY) or "0") >= 1
+    except ValueError:
+        return False
+
+
+def migrate_transport_default(path: Path) -> bool:
+    """Migrate unversioned HTTP preferences once, preserving later opt-outs."""
+    # Completed migrations need no write access, including read-only deployments.
+    if _transport_default_is_current(path):
+        return False
+    with _lock(path):
+        if _transport_default_is_current(path):
+            return False
+        values = {"OPENAI_TRANSPORT": "websocket", TRANSPORT_DEFAULT_VERSION_KEY: "1"}
+        _write(path, values)
+        os.environ.update(values)
+        return True

artemis/config/llm.py

@@ -62,6 +62,9 @@ class LLM(BaseModel):
     provider: LLMProvider = "openai"
     model: str = Field(default_factory=lambda: settings.OPENAI_MODEL)
     api_base: str | None = None
+    transport: Literal["http", "websocket"] = Field(
+        default_factory=lambda: settings.OPENAI_TRANSPORT
+    )
     timeout: float = Field(default_factory=lambda: settings.OPENAI_TIMEOUT_SECONDS, gt=0)
     max_retries: int = Field(default_factory=lambda: settings.OPENAI_MAX_RETRIES, ge=0)
     max_tokens: int | None = Field(default=None, gt=0)

artemis/config/settings.py

@@ -16,8 +16,10 @@
 
 import os
 from pathlib import Path
+from typing import Literal
 
 from dotenv import load_dotenv
+from filelock import Timeout
 from pydantic import Field, SecretStr, model_validator
 from pydantic_settings import BaseSettings
 
@@ -31,6 +33,7 @@ from artemis.config.constants import (
     ENV_OPENAI_API_KEY,
     ENV_VISION_API_KEY,
 )
+from artemis.config.environment import migrate_transport_default
 from artemis.config.paths import (
     GLOBAL_APP_DIR,
     get_data_engine_db_path,
@@ -42,13 +45,18 @@ from artemis.utils.logger import get_logger
 
 # Installed wheels load .env from the user directory, outside site-packages.
 _canonical_env = get_env_file()
+logger = get_logger(__name__)
+try:
+    if migrate_transport_default(_canonical_env):
+        logger.info("已将全局模型传输协议迁移为 WebSocket；之后可在模型配置中切回 HTTP。")
+except (OSError, Timeout):
+    os.environ["OPENAI_TRANSPORT"] = "websocket"
+    logger.warning("无法保存 WebSocket 默认迁移；本次使用 WebSocket，下次启动重试。")
 load_dotenv(dotenv_path=_canonical_env, verbose=True)
 _global_env = GLOBAL_APP_DIR / ".env"
 if _global_env.exists() and _global_env.resolve() != _canonical_env.resolve():
     load_dotenv(dotenv_path=_global_env, verbose=True)
 
-logger = get_logger(__name__)
-
 
 def is_placeholder_key(val: str | SecretStr | None) -> bool:
     """Check if an API credential is empty or an unconfigured placeholder value."""
@@ -109,6 +117,7 @@ class Settings(BaseSettings):
     OPENAI_BASE_URL: str | None = "https://api-n-cd.playones.com/v1"
     OPENAI_MODEL: str = Field(default=DEFAULT_MODEL, min_length=1)
     OPENAI_TIMEOUT_SECONDS: float = Field(default=300.0, gt=0)
+    OPENAI_TRANSPORT: Literal["http", "websocket"] = "websocket"
     # Environment only; deliberately absent from the saved UI/task configuration.
     OPENAI_USE_SYSTEM_PROXY: bool = False
     OPENAI_MAX_RETRIES: int = Field(default=5, ge=0, le=10)

artemis/controllers/unified_controller.py

@@ -36,6 +36,8 @@ from artemis.controllers.types import (
     SwipeStartEndPercentagesRequest,
     TapOutput,
 )
+from artemis.runtime.adb_endpoint import AdbSession, current_adb_endpoint
+from artemis.toolchain import find_scrcpy
 from artemis.utils.logger import get_logger
 from artemis.utils.video import (
     ANDROID_RECORDING_SEGMENT_SECONDS,
@@ -82,9 +84,13 @@ class UnifiedMobileController:
 
     @staticmethod
     async def _spawn_scrcpy(command: list[str]) -> asyncio.subprocess.Process:
+        adb = AdbSession(current_adb_endpoint())
+        if not adb.adb_path:
+            raise FileNotFoundError("Android Platform Tools (adb) could not be found.")
         kwargs: dict[str, Any] = {
             "stdout": asyncio.subprocess.PIPE,
             "stderr": asyncio.subprocess.PIPE,
+            "env": adb.environment(),
         }
         if os.name == "nt":
             kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
@@ -606,7 +612,7 @@ class UnifiedMobileController:
         config = recording_config(self.ctx)
         bitrate = config.h265_kbps if codec == "h265" else config.h264_kbps
         return build_scrcpy_record_command(
-            "scrcpy",
+            find_scrcpy(),
             device_id,
             path,
             f"{bitrate}K",

artemis/drivers/android/adb_driver.py

@@ -39,6 +39,7 @@ from artemis.runtime.observation_budget import (
     current_observation,
 )
 from artemis.toolchain import find_ffmpeg, find_scrcpy
+from artemis.runtime.adb_endpoint import AdbSession, current_adb_endpoint
 from artemis.utils.video import build_scrcpy_record_command
 from artemis.utils.ui_filter import filter_ui_hierarchy
 from artemis.utils.logger import get_logger
@@ -608,6 +609,9 @@ class AndroidAdbDriver(BaseDeviceDriver):
         self._recording_output_path = out_dir / "recording.mp4"
         logger.info(f"Starting scrcpy video recording to {self._recording_mkv_path}...")
         try:
+            adb = AdbSession(current_adb_endpoint())
+            if not adb.adb_path:
+                raise FileNotFoundError("Android Platform Tools (adb) could not be found.")
             scrcpy_bin = find_scrcpy()
             cmd = build_scrcpy_record_command(
                 scrcpy_bin,
@@ -617,6 +621,7 @@ class AndroidAdbDriver(BaseDeviceDriver):
             )
             self._scrcpy_process = await asyncio.create_subprocess_exec(
                 *cmd,
+                env=adb.environment(),
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
             )

artemis/llm/diagnostics.py

@@ -390,7 +390,12 @@ async def observe_response_headers(response):
     """Correlate allowlisted headers and passively observe bounded SSE frames."""
     attempt = ACTIVE_ATTEMPT.get()
     if attempt is not None:
-        attempt.headers(response)
+        websocket = response.extensions.get("artemis_websocket", False)
+        if websocket:
+            # 200/SSE is an in-process SDK facade, not a wire HTTP response.
+            attempt.data.update(headers_at=time.time(), phase="waiting_output")
+        else:
+            attempt.headers(response)
         content_type = response.headers.get("content-type", "").split(";", 1)[0].lower()
         if (
             response.status_code == 200
@@ -407,5 +412,7 @@ async def observe_response_headers(response):
                 )
                 response.stream = stream
                 attempt.wire_streams.append(stream)
+                if websocket:
+                    stream.observation.data["observation_boundary"] = "websocket_event_sse_adapter"
             else:
                 attempt.data["wire_streams_truncated"] = True

artemis/llm/router.py

@@ -17,7 +17,7 @@
 from enum import StrEnum
 import hashlib
 import os
-from typing import Any
+from typing import Any, Literal
 
 from langchain_core.language_models.chat_models import BaseChatModel
 from pydantic import BaseModel, Field
@@ -40,6 +40,9 @@ class ModelEndpoint(BaseModel):
     model_name: str = Field(default_factory=lambda: settings.OPENAI_MODEL)
     api_key: str | None = Field(default=None, repr=False)
     api_base: str | None = None
+    transport: Literal["http", "websocket"] = Field(
+        default_factory=lambda: settings.OPENAI_TRANSPORT
+    )
     temperature: float | None = None
     max_tokens: int | None = Field(default=None, gt=0)
     timeout_seconds: float = Field(default_factory=lambda: settings.OPENAI_TIMEOUT_SECONDS, gt=0)
@@ -63,6 +66,7 @@ class ModelEndpoint(BaseModel):
             self.enable_grounding,
             self.api_base or settings.OPENAI_BASE_URL,
             settings.OPENAI_USE_SYSTEM_PROXY,
+            self.transport,
             hashlib.sha256(key.encode()).hexdigest(),
         )
 
@@ -119,13 +123,32 @@ class ModelFactory:
             "http_client": DefaultHttpxClient(
                 timeout=endpoint.timeout_seconds, **model_http_options()
             ),
-            "http_async_client": DefaultAsyncHttpxClient(
+        }
+        if endpoint.transport == "websocket":
+            import httpx
+
+            from artemis.llm.websocket import ResponsesWebSocketTransport
+
+            # Disable HTTP proxy mounts on this facade: they would bypass the
+            # WebSocket transport. Each actual transport owns its proxy policy.
+            kwargs["http_async_client"] = httpx.AsyncClient(
+                timeout=endpoint.timeout_seconds,
+                trust_env=False,
+                transport=ResponsesWebSocketTransport(
+                    timeout=endpoint.timeout_seconds, use_proxy=settings.OPENAI_USE_SYSTEM_PROXY
+                ),
+                event_hooks={
+                    "request": [observe_request_headers],
+                    "response": [observe_response_headers],
+                },
+            )
+        else:
+            kwargs["http_async_client"] = DefaultAsyncHttpxClient(
                 timeout=endpoint.timeout_seconds,
                 **model_http_options(),
                 event_hooks={
                     "request": [observe_request_headers],
                     "response": [observe_response_headers],
                 },
-            ),
-        }
+            )
         return ChatOpenAI(**{k: v for k, v in kwargs.items() if v is not None})

artemis/llm/websocket.py

@@ -0,0 +1,449 @@
+"""Responses WebSocket transport beneath the existing SDK and safety guards.
+
+The SDK still serializes complete, authoritative requests. Only this wire boundary
+omits a verified prefix. No model messages, tool results, or screenshots are lost
+from local history, so any new connection can start from the full request.
+"""
+
+import asyncio
+from collections import OrderedDict
+from contextlib import suppress
+import hashlib
+import json
+import time
+from uuid import uuid4
+
+import httpx
+from websockets.asyncio.client import connect
+from websockets.exceptions import ConnectionClosed, InvalidStatus, SecurityError
+from websockets.protocol import State
+
+from artemis.llm.diagnostics import ACTIVE_ATTEMPT
+from artemis.llm.http import model_http_options
+from artemis.llm.responses_diagnostics import identifier
+
+IDLE_SECONDS = 60
+MAX_SESSIONS = 32
+MAX_CONNECTION_SECONDS = 55 * 60
+
+
+def _json(value):
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
+
+
+def _fingerprint(item):
+    # Assistant message IDs/status are transport metadata; store=false strips
+    # message IDs on replay. Keep every semantic field, including call_id/phase.
+    if isinstance(item, dict):
+        item = dict(item)
+        item.pop("status", None)
+        if item.get("type") == "message" or item.get("role") == "assistant":
+            item.pop("id", None)
+            item.setdefault("type", "message")
+            # Completion events may include token probabilities; LangChain
+            # correctly omits those response-only statistics from replay input.
+            if isinstance(item.get("content"), list):
+                item["content"] = [
+                    {k: v for k, v in block.items() if k != "logprobs"}
+                    if isinstance(block, dict) and block.get("type") == "output_text"
+                    else block
+                    for block in item["content"]
+                ]
+        if item.get("type") == "function_call" and isinstance(item.get("arguments"), str):
+            with suppress(ValueError):
+                item["arguments"] = json.loads(item["arguments"])
+    return hashlib.sha256(_json(item).encode()).digest()
+
+
+class _Connect(connect):
+    def process_redirect(self, exc):
+        # Never forward the API credential to a redirected endpoint.
+        if not isinstance(exc, InvalidStatus) or exc.response.status_code not in {
+            301,
+            302,
+            303,
+            307,
+            308,
+        }:
+            return exc
+        return SecurityError("Responses WebSocket redirects are disabled")
+
+
+class _Session:
+    def __init__(self):
+        self.lock = asyncio.Lock()
+        self.users = 0
+        self.connection = None
+        self.opened_at = 0.0
+        self.previous_id = None
+        self.history = ()
+        self.options = None
+        self.timer = None
+        self.handshake = {}
+
+    async def close(self):
+        if self.timer:
+            self.timer.cancel()
+            self.timer = None
+        self.previous_id = None
+        self.history = ()
+        self.options = None
+        connection, self.connection = self.connection, None
+        if connection is not None:
+            with suppress(ConnectionClosed, OSError, TimeoutError):
+                async with asyncio.timeout(1):
+                    await connection.close()
+
+
+class ResponsesWebSocketTransport(httpx.AsyncBaseTransport):
+    """One serial socket per conversation/role, independent sockets for concurrency."""
+
+    def __init__(self, *, timeout, use_proxy):
+        self.timeout = timeout
+        self.use_proxy = use_proxy
+        self.http = httpx.AsyncClient(timeout=timeout, **model_http_options())
+        self.sessions = OrderedDict()
+        self.cleanup_tasks = set()
+
+    async def aclose(self):
+        sessions = list(self.sessions.values())
+        self.sessions.clear()
+        await asyncio.gather(*(session.close() for session in sessions))
+        if self.cleanup_tasks:
+            await asyncio.gather(*tuple(self.cleanup_tasks), return_exceptions=True)
+        await self.http.aclose()
+
+    def _close_background(self, session):
+        task = asyncio.create_task(session.close())
+        self.cleanup_tasks.add(task)
+        task.add_done_callback(self.cleanup_tasks.discard)
+
+    async def _session(self, request, body):
+        key = body.get("prompt_cache_key")
+        # Header identity prevents credentials, routing, and different gateways
+        # from sharing a socket. Per-request correlation headers aren't affinity.
+        headers = {
+            k: v
+            for k, v in request.headers.items()
+            if k not in {"content-length", "x-client-request-id", "x-client-attempt-id"}
+        }
+        identity = hashlib.sha256(_json([str(request.url), key, headers]).encode()).digest()
+        reusable = isinstance(key, str) and bool(key)
+        if reusable and identity in self.sessions:
+            session = self.sessions.pop(identity)
+            self.sessions[identity] = session
+        else:
+            if len(self.sessions) >= MAX_SESSIONS:
+                for old_key, old in list(self.sessions.items()):
+                    if not old.users:
+                        del self.sessions[old_key]
+                        # Reserve the replacement before yielding: concurrent
+                        # requests for its scope must find the same session.
+                        self._close_background(old)
+                        break
+            session = _Session()
+            reusable = reusable and len(self.sessions) < MAX_SESSIONS
+            if reusable:
+                self.sessions[identity] = session
+        acquired = False
+        session.users += 1
+        try:
+            await session.lock.acquire()
+            acquired = True
+        finally:
+            if not acquired:
+                session.users -= 1
+                self._idle(identity, session, reusable)
+        if session.timer:
+            session.timer.cancel()
+            session.timer = None
+        return identity, session, reusable
+
+    def release(self, identity, session, reusable):
+        session.lock.release()
+        session.users -= 1
+        self._idle(identity, session, reusable)
+
+    def _idle(self, identity, session, reusable):
+        if not reusable or session.users or self.sessions.get(identity) is not session:
+            return
+
+        def expire():
+            if session.users or self.sessions.get(identity) is not session:
+                return
+            del self.sessions[identity]
+            self._close_background(session)
+
+        session.timer = asyncio.get_running_loop().call_later(IDLE_SECONDS, expire)
+
+    async def handle_async_request(self, request):
+        if request.method != "POST" or not request.url.path.rstrip("/").endswith("/responses"):
+            return await self.http.send(request, stream=True)
+        body = json.loads(await request.aread())
+        identity, session, reusable = await self._session(request, body)
+        exchange = _Exchange(self, request, body, identity, session, reusable)
+        stream_handed_off = False
+        try:
+            await exchange.start()
+            first = await exchange.read()
+            if first.get("type") == "error":
+                status = first.get("status", 400)
+                status = status if type(status) is int and 400 <= status <= 599 else 400
+                return httpx.Response(status, json={"error": first.get("error", {})})
+            if not body.get("stream"):
+                event = first
+                while not exchange.done:
+                    event = await exchange.read()
+                    if event.get("type") == "error":
+                        raise httpx.ReadError("Responses WebSocket request failed", request=request)
+                return httpx.Response(
+                    200, json=event["response"], extensions={"artemis_websocket": True}
+                )
+            response = httpx.Response(
+                200,
+                headers={"content-type": "text/event-stream"},
+                stream=_EventStream(exchange, first),
+                extensions={"artemis_websocket": True},
+            )
+            stream_handed_off = True
+            return response
+        except InvalidStatus as error:
+            return httpx.Response(
+                error.response.status_code,
+                json={
+                    "error": {
+                        "message": "Responses WebSocket 升级失败，请检查网关 WebSocket 支持或选择 HTTP。"
+                    }
+                },
+                headers={
+                    k: v
+                    for k, v in error.response.headers.raw_items()
+                    if k.lower() in {"retry-after", "x-request-id"}
+                },
+            )
+        finally:
+            if not stream_handed_off:
+                await exchange.aclose()
+
+
+class _Exchange:
+    def __init__(self, transport, request, body, identity, session, reusable):
+        self.transport = transport
+        self.request = request
+        self.body = body
+        self.identity = identity
+        self.session = session
+        self.reusable = reusable
+        self.done = False
+        self.closed = False
+        self.recovered = False
+        self.output_seen = False
+        self.input = body.get("input")
+        self.input_hashes = (
+            tuple(map(_fingerprint, self.input)) if isinstance(self.input, list) else ()
+        )
+        self.options = hashlib.sha256(
+            _json(
+                {
+                    k: v
+                    for k, v in body.items()
+                    if k not in {"input", "stream", "background", "previous_response_id"}
+                }
+            ).encode()
+        ).digest()
+        self.explicit_state = bool(body.get("previous_response_id") or body.get("conversation"))
+        self.attempt = ACTIVE_ATTEMPT.get()
+        self.stats = {
+            "mode": "websocket",
+            "full_request_bytes": len(request.content),
+            "sent_payload_bytes": 0,
+            "received_payload_bytes": 0,
+            "create_events": 0,
+            "incremental": False,
+            "reused_input_items": 0,
+            "connection_reused": session.connection is not None,
+        }
+        if self.attempt is not None:
+            self.attempt.data["transport"] = self.stats
+
+    async def start(self):
+        session = self.session
+        if session.connection is not None and (
+            session.connection.state is not State.OPEN
+            or time.monotonic() - session.opened_at >= MAX_CONNECTION_SECONDS
+        ):
+            await session.close()
+            self.stats["connection_reused"] = False
+        if session.connection is None:
+            url = self.request.url.copy_with(
+                scheme="wss" if self.request.url.scheme == "https" else "ws"
+            )
+            headers = {
+                k: v
+                for k, v in self.request.headers.items()
+                if k
+                not in {"host", "content-length", "content-type", "connection", "accept-encoding"}
+            }
+            options = {"ssl": model_http_options()["verify"]} if url.scheme == "wss" else {}
+            session.connection = await _Connect(
+                str(url),
+                additional_headers=headers,
+                user_agent_header=None,
+                proxy=True if self.transport.use_proxy else None,
+                open_timeout=self.transport.timeout,
+                close_timeout=1,
+                max_size=64 * 1024 * 1024,
+                **options,
+            )
+            session.opened_at = time.monotonic()
+            session.handshake = {
+                "client_request_id": identifier(headers.get("x-client-request-id")),
+                "client_attempt_id": identifier(headers.get("x-client-attempt-id")),
+            }
+        if self.attempt:
+            self.attempt.data.update(session.handshake, request_header_scope="websocket_handshake")
+            for field, local in (
+                ("client_request_id", "request_id"),
+                ("client_attempt_id", "attempt_id"),
+            ):
+                self.attempt.data[field + "_matches"] = session.handshake.get(
+                    field
+                ) == self.attempt.data.get(local)
+        self.stats["handshake_http_status"] = 101
+        await self.send(incremental=True)
+
+    async def send(self, *, incremental):
+        session = self.session
+        payload = {k: v for k, v in self.body.items() if k not in {"stream", "background"}}
+        count = len(session.history)
+        can_continue = (
+            incremental
+            and not self.explicit_state
+            and session.previous_id
+            and session.options == self.options
+            and count
+            and len(self.input_hashes) > count
+            and self.input_hashes[:count] == session.history
+        )
+        if can_continue:
+            payload["previous_response_id"] = session.previous_id
+            payload["input"] = self.input[count:]
+            self.stats.update(incremental=True, reused_input_items=count)
+        else:
+            self.stats.update(incremental=False, reused_input_items=0)
+            self.stats["full_context_reason"] = (
+                "explicit_state"
+                if self.explicit_state
+                else "history_or_options_changed"
+                if session.previous_id
+                else "new_connection_or_chain"
+            )
+        payload["type"] = "response.create"
+        encoded = _json(payload)
+        await session.connection.send(encoded)
+        self.stats["sent_payload_bytes"] += len(encoded.encode())
+        self.stats["create_events"] += 1
+        self.stats["payload_bytes_saved"] = max(
+            0, self.stats["full_request_bytes"] - self.stats["sent_payload_bytes"]
+        )
+        if self.attempt:
+            self.attempt.data["request_sent_at"] = time.time()
+            self.attempt.publish_if_due()
+
+    async def read(self):
+        try:
+            async with asyncio.timeout(self.transport.timeout):
+                raw = await self.session.connection.recv()
+            self.stats["received_payload_bytes"] += len(
+                raw.encode() if isinstance(raw, str) else raw
+            )
+            event = json.loads(raw)
+            if not isinstance(event, dict):
+                raise ValueError("Responses WebSocket event is not an object")
+        except ConnectionClosed as error:
+            raise httpx.ReadError(
+                "Responses WebSocket connection closed before completion", request=self.request
+            ) from error
+        except (ValueError, UnicodeError) as error:
+            raise httpx.ReadError(
+                "Invalid Responses WebSocket event", request=self.request
+            ) from error
+        kind = event.get("type", "")
+        error = event.get("error") or {}
+        if (
+            kind == "error"
+            and error.get("code") == "previous_response_not_found"
+            and self.stats["incremental"]
+            and not self.recovered
+            and not self.output_seen
+        ):
+            # An explicit rejection means no model output was generated. Recover
+            # once from the authoritative input; never replay an ambiguous drop.
+            self.recovered = True
+            self.session.previous_id = None
+            self.stats["chain_recovered"] = True
+            await self.send(incremental=False)
+            return await self.read()
+        if kind not in {"error", "response.created", "response.in_progress", "response.queued"}:
+            self.output_seen = True
+        if kind in {
+            "response.completed",
+            "response.failed",
+            "response.incomplete",
+            "response.cancelled",
+        }:
+            self.done = True
+            response = event.get("response") or {}
+            self.session.previous_id = None
+            if (
+                kind == "response.completed"
+                and response.get("status") == "completed"
+                and isinstance(response.get("id"), str)
+                and response["id"]
+                and isinstance(response.get("output"), list)
+                and self.input_hashes
+                and not self.explicit_state
+            ):
+                output = [
+                    item
+                    for item in response["output"]
+                    if not (
+                        item.get("type") == "reasoning"
+                        and not item.get("encrypted_content")
+                        and self.body.get("store") is False
+                    )
+                ]
+                self.session.previous_id = response["id"]
+                self.session.history = self.input_hashes + tuple(map(_fingerprint, output))
+                self.session.options = self.options
+        return event
+
+    async def aclose(self):
+        if self.closed:
+            return
+        self.closed = True
+        try:
+            if not self.done or not self.reusable:
+                await self.session.close()
+        finally:
+            self.transport.release(self.identity, self.session, self.reusable)
+
+
+class _EventStream(httpx.AsyncByteStream):
+    def __init__(self, exchange, first):
+        self.exchange = exchange
+        self.first = first
+
+    async def __aiter__(self):
+        try:
+            event = self.first
+            while True:
+                yield b"data: " + _json(event).encode() + b"\n\n"
+                if self.exchange.done or event.get("type") == "error":
+                    return
+                event = await self.exchange.read()
+        finally:
+            await self.aclose()
+
+    async def aclose(self):
+        await self.exchange.aclose()

artemis/resources/config/artemis.jsonc

@@ -2,6 +2,7 @@
 {
   "default": {
     "provider": "openai",
+    "transport": "${env:OPENAI_TRANSPORT:-websocket}",
     "model": "${env:OPENAI_MODEL:-gpt-6-luna}",
     "timeout": "${env:OPENAI_TIMEOUT_SECONDS:-300}",
     "max_retries": "${env:OPENAI_MAX_RETRIES:-5}",

artemis/runtime/adb_endpoint.py

@@ -167,6 +167,8 @@ class AdbSession:
     def environment(self, base: Mapping[str, str] | None = None) -> dict[str, str]:
         environment = dict(base if base is not None else os.environ)
         self.endpoint.apply_to_environment(environment)
+        if self.adb_path:
+            environment["ADB"] = self.adb_path
         return environment
 
     def run(self, arguments: Sequence[str], **kwargs: Any) -> subprocess.CompletedProcess:

artemis/runtime/diagnostics.py

@@ -0,0 +1,207 @@
+"""Small, task-owned runtime snapshots; no polling or readiness decisions."""
+
+import asyncio
+import ctypes
+from ctypes import wintypes
+from functools import lru_cache
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import platform
+import subprocess
+import sys
+import time
+
+import psutil
+
+from artemis._version import get_version
+from artemis.config.paths import ROOT_DIR
+from artemis.runtime.adb_endpoint import current_adb_endpoint
+from artemis.toolchain import toolchain
+
+logger = logging.getLogger(__name__)
+FILENAME = "runtime_diagnostics.json"
+
+
+def _command(arguments: list[str]) -> str | None:
+    try:
+        result = subprocess.run(
+            arguments,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=3,
+            creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0,
+            check=False,
+        )
+        return result.stdout.strip()[:4096] if result.returncode == 0 else None
+    except (OSError, subprocess.SubprocessError):
+        return None
+
+
+def _tool_identity(name: str) -> dict:
+    path = toolchain.resolve(name)
+    result = {"path": path, "version": None, "sha256": None}
+    if path:
+        result["version"] = _command([path, "version" if name == "adb" else "--version"])
+        try:
+            with Path(path).open("rb") as stream:
+                result["sha256"] = hashlib.file_digest(stream, "sha256").hexdigest()
+        except OSError:
+            logger.debug("Could not hash %s", name)
+    return result
+
+
+@lru_cache(maxsize=2)
+def runtime_identity(android: bool) -> dict:
+    """Read identity once per worker, before later source updates can change it."""
+    git = ["git", "-C", str(ROOT_DIR)]
+    checkout = (ROOT_DIR / ".git").exists()
+    commit = _command([*git, "rev-parse", "--verify", "HEAD"]) if checkout else None
+    changes = (
+        _command([*git, "status", "--porcelain", "--untracked-files=no"]) if checkout else None
+    )
+    return {
+        "artemis_version": get_version(),
+        "source_commit": commit,
+        "source_modified": bool(changes) if changes is not None else None,
+        "installation_archive_sha256": None,
+        "python_version": platform.python_version(),
+        "platform": platform.system(),
+        "tools": {name: _tool_identity(name) for name in ("adb", "scrcpy")} if android else {},
+    }
+
+
+class _PerformanceInformation(ctypes.Structure):
+    # https://learn.microsoft.com/windows/win32/api/psapi/ns-psapi-performance_information
+    _fields_ = (
+        [("cb", wintypes.DWORD)]
+        + [
+            (name, ctypes.c_size_t)
+            for name in (
+                "CommitTotal",
+                "CommitLimit",
+                "CommitPeak",
+                "PhysicalTotal",
+                "PhysicalAvailable",
+                "SystemCache",
+                "KernelTotal",
+                "KernelPaged",
+                "KernelNonpaged",
+                "PageSize",
+            )
+        ]
+        + [(name, wintypes.DWORD) for name in ("HandleCount", "ProcessCount", "ThreadCount")]
+    )
+
+
+def _windows_commit() -> dict | None:
+    if sys.platform != "win32":
+        return None
+    values = _PerformanceInformation()
+    values.cb = ctypes.sizeof(values)
+    query = ctypes.WinDLL("psapi", use_last_error=True).GetPerformanceInfo
+    query.argtypes = [ctypes.POINTER(_PerformanceInformation), wintypes.DWORD]
+    query.restype = wintypes.BOOL
+    if not query(ctypes.byref(values), values.cb):
+        raise ctypes.WinError(ctypes.get_last_error())
+    committed = values.CommitTotal * values.PageSize
+    limit = values.CommitLimit * values.PageSize
+    return {
+        "committed_bytes": committed,
+        "limit_bytes": limit,
+        "available_bytes": max(0, limit - committed),
+        "note": "系统提交额度，不能当作页面文件大小。",
+    }
+
+
+def _memory_snapshot() -> dict:
+    result = {"physical": None, "system_commit": None, "worker": None}
+    try:
+        memory = psutil.virtual_memory()
+        result["physical"] = {"total_bytes": memory.total, "available_bytes": memory.available}
+        result["system_commit"] = _windows_commit()
+    except (OSError, psutil.Error):
+        logger.debug("Host memory snapshot unavailable")
+    try:
+        memory = psutil.Process().memory_info()
+        result["worker"] = {
+            "pid": os.getpid(),
+            "rss_bytes": memory.rss,
+            "private_bytes": getattr(memory, "private", None),
+        }
+    except (OSError, psutil.Error):
+        logger.debug("Worker memory snapshot unavailable")
+    return result
+
+
+def _adb_server() -> dict:
+    endpoint = current_adb_endpoint()
+    result = {"endpoint": endpoint.to_dict(), "local_listener": None}
+    if endpoint.host.lower() not in {"localhost", "127.0.0.1", "::1"}:
+        return result  # A remote server's process cannot be observed on this host.
+    try:
+        for connection in psutil.net_connections(kind="tcp"):
+            if (
+                connection.status == psutil.CONN_LISTEN
+                and connection.laddr.port == endpoint.port
+                and connection.laddr.ip
+                in {
+                    endpoint.host,
+                    "0.0.0.0",
+                    "::",
+                    *(("127.0.0.1", "::1") if endpoint.host.lower() == "localhost" else ()),
+                }
+                and connection.pid
+            ):
+                process = psutil.Process(connection.pid)
+                result["local_listener"] = {
+                    "pid": process.pid,
+                    "name": process.name(),
+                    "path": process.exe(),
+                }
+                break
+    except (OSError, psutil.Error):
+        logger.debug("ADB listener identity unavailable")
+    return result
+
+
+def write_runtime_diagnostics(directory: Path, phase: str, *, android: bool = True) -> None:
+    """Keep a start snapshot and the latest failure snapshot without changing task results."""
+    if phase not in {"start", "failure"}:
+        raise ValueError("Unknown runtime snapshot phase")
+    path = directory / FILENAME
+    try:
+        payload = {"schema_version": 1}
+        if path.is_file() and path.stat().st_size <= 256 * 1024:
+            try:
+                saved = json.loads(path.read_text(encoding="utf-8"))
+                if isinstance(saved, dict):
+                    payload = saved
+            except (OSError, ValueError):
+                logger.debug("Previous runtime snapshot unreadable")
+        if phase == "start" and phase in payload:
+            return
+        payload[phase] = {
+            "captured_at": time.time(),
+            "runtime": runtime_identity(android),
+            "memory": _memory_snapshot(),
+            "adb_server": _adb_server() if android else None,
+        }
+        directory.mkdir(parents=True, exist_ok=True)
+        temporary = path.with_suffix(".tmp")
+        temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+        temporary.replace(path)
+    except (OSError, psutil.Error, ValueError, TypeError):
+        logger.warning("Could not save runtime snapshot (%s)", phase, exc_info=True)
+
+
+async def capture_runtime_diagnostics(directory: Path, phase: str, *, android: bool = True) -> None:
+    """Keep optional evidence failures outside the task's execution result."""
+    try:
+        await asyncio.to_thread(write_runtime_diagnostics, directory, phase, android=android)
+    except (OSError, RuntimeError):
+        logger.warning("Could not start runtime snapshot (%s)", phase, exc_info=True)

artemis/sdk/agent.py

@@ -654,6 +654,13 @@ class Agent:
                 # publish a new active session while the current task is still
                 # finishing.
                 self._prepare_tracing(task=task, context=context)
+                from artemis.runtime.diagnostics import capture_runtime_diagnostics
+
+                await capture_runtime_diagnostics(
+                    context.data_engine.base_dir,
+                    "start",
+                    android=request.execution_target != "computer",
+                )
                 if device_lock is not None and os.environ.get("ARTEMIS_TASK_WORKER") == "1":
                     from artemis.runtime.task_control import TaskControl
 
@@ -1333,6 +1340,14 @@ class Agent:
 
     async def _finalize_tracing(self, task: Task, context: ArtemisContext):
         if context.data_engine:
+            if task.status == "failed":
+                from artemis.runtime.diagnostics import capture_runtime_diagnostics
+
+                await capture_runtime_diagnostics(
+                    context.data_engine.base_dir,
+                    "failure",
+                    android=context.execution_target != "computer",
+                )
             session_status = "completed"
             if task.status == "failed":
                 session_status = "failed"

artemis/services/llm.py

@@ -748,10 +748,15 @@ class RobustChatModelWrapper:
         if observation is not None and (usage := usage_from_message(response)) is not None:
             observation.data["usage"] = usage
         ensure_response_complete(response)
-        if self.structured_mode == "parsed" and isinstance(response, dict) and "raw" in response:
-            if response.get("parsing_error") is not None:
-                raise response["parsing_error"]
-            return response["parsed"]
+        if self.structured_mode == "parsed":
+            if isinstance(response, dict) and "raw" in response:
+                if response.get("parsing_error") is not None:
+                    raise ResponseIntegrityError("invalid_structured_output") from response[
+                        "parsing_error"
+                    ]
+                response = response.get("parsed")
+            if response is None:
+                raise ResponseIntegrityError("missing_structured_output")
         return response
 
     async def ainvoke(self, *args, **kwargs):
@@ -1174,6 +1179,7 @@ def _resolve_endpoint(
         else settings.OPENAI_MAX_RETRIES,
         max_tokens=_get_val(cfg, "max_tokens", int),
         api_base=_get_val(cfg, "api_base", str),
+        transport=_get_val(cfg, "transport", str) or settings.OPENAI_TRANSPORT,
         reasoning_effort=_get_val(cfg, "reasoning_effort", str),
         enable_grounding=_get_val(cfg, "enable_grounding", bool) or False,
     )

artemis/tools/mobile/launch_app.py

@@ -18,7 +18,7 @@ from langchain_core.messages import ToolMessage
 from langchain_core.tools import BaseTool
 from pydantic import BaseModel, Field
 
-from artemis.agents.hopper.hopper import HopperOutput, hopper
+from artemis.agents.hopper.hopper import HopperError, HopperOutput, hopper
 from artemis.context import ArtemisContext
 from artemis.controllers.platform_specific_commands_controller import (
     list_packages_async,
@@ -53,7 +53,7 @@ LAUNCH_APP_DOCSTRING = (
 async def find_package(ctx: ArtemisContext, app_name: str, use_fallback: bool = True) -> str | None:
     """Finds the package name for a given application name.
 
-    Returns None if package not found or on error.
+    Returns None if no installed package matches. Model failures raise HopperError.
     """
     package_cache = getattr(ctx, "package_cache", None)
     if package_cache is None or not isinstance(package_cache, dict):
@@ -63,7 +63,7 @@ async def find_package(ctx: ArtemisContext, app_name: str, use_fallback: bool =
         except Exception:  # pylint: disable=broad-exception-caught
             package_cache = {}
 
-    if isinstance(package_cache, dict) and app_name in package_cache:
+    if isinstance(package_cache, dict) and package_cache.get(app_name):
         logger.info(f"Cache hit for app '{app_name}': {package_cache[app_name]}")
         return package_cache[app_name]
 
@@ -84,8 +84,6 @@ async def find_package(ctx: ArtemisContext, app_name: str, use_fallback: bool =
             use_fallback=use_fallback,
         )
         if not hopper_output.found or not hopper_output.output:
-            if isinstance(package_cache, dict):
-                package_cache[app_name] = None
             return None
 
         package_name = hopper_output.output.strip()
@@ -94,13 +92,13 @@ async def find_package(ctx: ArtemisContext, app_name: str, use_fallback: bool =
                 f"Hopper returned package '{package_name}' for '{app_name}', "
                 "but it is NOT physically installed on the device!"
             )
-            if isinstance(package_cache, dict):
-                package_cache[app_name] = None
             return None
 
         if isinstance(package_cache, dict):
             package_cache[app_name] = package_name
         return package_name
+    except HopperError:
+        raise
     except Exception as e:  # pylint: disable=broad-exception-caught
         logger.error(f"Failed to find package for '{app_name}': {e}")
         return None

config/artemis.jsonc

@@ -2,6 +2,7 @@
 {
   "default": {
     "provider": "openai",
+    "transport": "${env:OPENAI_TRANSPORT:-websocket}",
     "model": "${env:OPENAI_MODEL:-gpt-6-luna}",
     "timeout": "${env:OPENAI_TIMEOUT_SECONDS:-300}",
     "max_retries": "${env:OPENAI_MAX_RETRIES:-5}",

docs/ai-automation.md

@@ -169,6 +169,20 @@ Flash、Pro（启用 transcript）和探索共用分阶段上下文。一个阶
 
 辅助摘要继续独立处理各自片段，避免为了提高命中比例累积不相关的历史。视觉摘要中完全重复的插件执行结果与控件树只发送一份；独立错误、状态、不同观察及原始 trace 均保留。
 
+## Responses WebSocket 增量传输
+
+模型传输默认 `websocket`。首次启动时，未带迁移标记的全局配置（包括已保存的 `OPENAI_TRANSPORT=http`）统一切换为 WebSocket，并在安装模式对应的 `.env` 原子写入 `ARTEMIS_TRANSPORT_DEFAULT_VERSION=1`。迁移只调整协议及此内部标记，保留凭据、接口地址、其他设置和注释；跨进程文件锁避免启动与网页保存相互覆盖。保存失败保留原文件、明确告警，本次进程仍使用 WebSocket，下次启动重试。
+
+之后可在“模型配置 → 传输协议”手动切回 `HTTP（兼容模式）`；保存会保留迁移标记，不会在重启时再次改成 WebSocket。直接编辑 `.env` 时保留 `ARTEMIS_TRANSPORT_DEFAULT_VERSION=1`，设置 `OPENAI_TRANSPORT=http` 后重启即可切回。网页保存对新任务生效；升级迁移不会改动正在运行的任务。JSON 模型/节点配置仍可用 `transport` 单独覆盖。MCP 使用 `mobile_config(section="model", changes={"transport":"websocket"}, persist_defaults=true)` 保存，仍须用户明确要求修改默认值。
+
+异步任务请求在现有 SDK 序列化与完整性保护之下接入 `/v1/responses` WebSocket。首轮发送完整上下文；同一任务（或探索）、角色、模型、路由头及凭据范围内复用连接。只在当前输入严格包含上一轮输入和完整输出、生成选项一致时，发送 `previous_response_id` 与新增输入。仅忽略消息 ID、状态、输出概率等不影响历史语义的响应字段，不按文本相似度截取。工具结果保留 `call_id`，文本、图片、结构化输出与 usage 继续走原有解析和安全检查。同步底层模型调用仍使用 HTTP；Flash、Pro、探索和辅助模型运行路径均为异步。
+
+本地历史、截图替换或压缩后重新发送当前完整上下文；不会为了续传保留已经移除的旧界面索引。连接重建或服务端明确返回 `previous_response_not_found` 时，从本地完整历史恢复，后者最多补发一次且必须尚未出现输出。连接在输出中途断开时，不在传输层盲目重发，交给现有模型重试预算；取消关闭未完成连接，不采纳半截工具参数或结论。不同会话并行、同一连接串行；连接池最多保留 32 个会话，空闲 60 秒释放，连接满 55 分钟主动重建。`store=false` 保持不变。
+
+网关必须支持 WebSocket Upgrade、双向长连接、`response.create` 与响应 ID 续接；只透传会话头不能证明支持。握手传递现有授权和路由头，保持 TLS/自定义 CA 验证与 `OPENAI_USE_SYSTEM_PROXY` 策略，拒绝携带凭据跨地址重定向。不支持升级时明确报错，可在网页改回 HTTP，不会静默改成 HTTP 全量请求。
+
+请求诊断的 `transport` 记录 `full_request_bytes`、`sent_payload_bytes`、`received_payload_bytes`、`payload_bytes_saved`、`incremental`、`reused_input_items`、`connection_reused`、`create_events` 和重建原因。字节数衡量 JSON 应用载荷，不包含 TLS/WebSocket 帧和压缩开销；SDK 内部兼容 SSE 的字节不冒充网络流量或 HTTP 200。握手上的请求标识属于建连尝试，后续轮次以本地请求诊断及响应 ID 关联。减少上传不等于减少模型上下文 token，也不等于缓存命中，usage 仍以服务端实报为准。协议见 [OpenAI 官方说明](https://developers.openai.com/api/docs/guides/websocket-mode)。
+
 ## 指定任务诊断、复盘与管理
 
 流式请求中的 `click`、`long_press`、`swipe`、`press_key`、`manage_app`、`wait_for_delay` 每次调用参数超过 16,384 字符时，会立即关闭该请求并丢弃未完成输出，避免持续生成异常参数直到超时。该保护只针对小型设备动作，不限制输入文本、笔记、报告或插件参数，也不改变模型思考强度与输出 token 配置。异常最多额外尝试一次（配置为不重试时不重试）；仍异常时进入现有备用模型或暂停恢复流程。诊断记录保留 `invalid_output` 分类及 `stream_guard` 中的工具名、实际字符数和保护上限，不保存参数正文；无法据此断定异常来自模型生成还是 API 中转。

docs/automation-workspace.md

@@ -50,6 +50,8 @@ Flash 开关保存在全局配置中，网页切换后自动保存，失败恢
 
 诊断可单独下载，也会收录到单任务“全部资料”和批次 ZIP 各任务目录的 `模型请求诊断.json`。请求状态复用任务 trace 数据库，每次尝试一条记录；同一阶段的流式更新至多每 5 秒保存一次，阶段变化立即记录，不额外调用模型、抓取屏幕或记录认证头。强制退出导致缺少结束事件时保留最后阶段，不推断为成功。
 
+新任务还会在开始和失败收尾时保存一次运行环境快照，单任务“全部资料”和批次 ZIP 各任务目录收录为 `运行环境诊断.json`。内容包括 ARTEMIS/Python 版本、可取得的源码提交与本地修改标记、实际解析的 ADB/scrcpy 路径、版本与 SHA256，以及采集时的内存和本机 ADB 监听进程信息；仅 PC 任务不查询安卓工具。Windows 的系统提交量、提交上限与余量按字节记录，不代表页面文件单独大小。快照不持续监控、不控制任务派发，也不能还原采集间隔内的内存变化；远程 ADB 进程、未保留的安装 ZIP 哈希及旧任务缺失字段均保留未知，导出时不会用当前环境补写历史。批次重启仍需手动继续，原暂停原因保留，恢复事件另行留档。
+
 Flash、Pro 和探索的最终验收使用独立总预算，不再复用中间检查点的 180 秒时限。预算按实际 Checker 的单次请求超时 `T`、重试次数 `R`，以及普通/限流重试间隔的较大值 `D` 自动计算：`min(3600, T × (3 + R) + D × R + 60)` 秒。它为取证、结论及必要的结构化整理预留三次正常请求时间，再共用一份重试时间余量和 60 秒工具读取余量；不会按 20 轮循环逐轮放大。每个请求仍最多重试 `R` 次，但全部请求和重试必须在总预算内完成。默认配置的上限为 2760 秒，提前完成就立即结束；用户停止、探索指导及任务本身的执行时限仍生效。
 
 页面在验收期间单独显示证据准备/核对/读取、结论整理、完整性校验阶段，以及已用时间、剩余总预算、请求与实际重试次数。诊断 JSON 的 `final_reviews` 记录每次验收的预算、截止时间和结束原因，相关模型请求通过 `final_review.review_id` 关联。请求超时仅重试同一模型请求，保留已取得的工具证据；耗尽请求重试或验收总预算时明确记为未验证，不再进入模型故障后的人工恢复等待。用户停止或探索指导引发的取消单独记录为中断，不重试。超时或中断后的半截输出会从流式展示中移除，只有完整响应经解析、校验后才可能确认通过。

docs/releases/2.7.0.md

@@ -0,0 +1,30 @@
+# ARTEMIS 2.7.0
+
+本版在正式 2.6.0 基础上新增 OpenAI Responses WebSocket 增量传输，减少多轮自动化任务重复上传完整历史的流量，并将 WebSocket 设为默认协议。
+
+## WebSocket 增量传输
+
+- Flash、Pro、探索及辅助模型的异步请求支持复用 Responses WebSocket 连接。首轮发送完整上下文；同一会话、角色、模型、凭据及路由范围内，后续输入严格延续上一轮完整历史时，仅发送响应 ID 和新增输入。图片、工具调用、结构化输出及用量统计保留原有解析与校验。
+- 历史压缩、截图替换、生成选项或连接变化后重新发送当前完整上下文。服务端明确表示响应 ID 不存在且尚无输出时，最多恢复一次完整请求；流式输出中途断开或取消时，不在传输层盲目重发，不采纳半截工具参数或结论。
+- 不同会话可并行，同一连接串行执行；连接池有数量、空闲和连接寿命限制。握手保持既有授权、路由、TLS 与代理策略，拒绝携带凭据跨地址重定向。
+- 请求诊断增加实际应用载荷字节、增量输入、连接复用及重建原因。上传减少不等于模型上下文 token 减少，也不等于缓存命中；用量和缓存仍按服务端实报统计。不承诺固定流量节省比例。
+
+## 默认协议与旧配置迁移
+
+- 新安装默认使用 WebSocket。升级后首次启动，也会将旧用户已保存的全局 `OPENAI_TRANSPORT=http` 一次性改为 `websocket`，并写入内部迁移标记。保留接口地址、密钥及其它配置；失败时保留原文件并明确告警，下次启动重试。
+- 迁移后可在“模型配置 → 传输协议”手动切回 HTTP，后续重启保留选择。网页、MCP 与环境配置共用此设置；JSON 模型或节点的显式协议覆盖继续有效。
+- 网关必须支持 Responses WebSocket Upgrade、双向长连接及响应 ID 续接。仅透传会话请求头不代表支持；不支持时明确报错，可手动选择 HTTP，不会静默退回全量 HTTP 请求。
+- 同步底层模型调用仍使用 HTTP。完整行为、配置方式与诊断字段见[传输说明](../ai-automation.md#responses-websocket-增量传输)。
+
+## 版本与验证
+
+- 主程序与前端根版本升级到 2.7.0；SDK、插件和 Android 助手保持现有独立版本。WebSocket 实现所需的 `websockets` 最低版本提高到 15，锁定的第三方包版本不变；此次升版只调整应用版本字段。
+- 后端回归覆盖真实本地 WebSocket/SDK 多轮与工具调用、结构化输出、取消、重连、隔离、生命周期和诊断；配置迁移覆盖旧 HTTP、新安装、重复启动、手动切回与保存失败。
+- 最终源码 ZIP 使用两个独立解压目录和空 npm 缓存，分别验证官方源、国内镜像的锁定安装及生产构建。实际运行时、源码提交、SHA256、测试结果和未测范围见发布附件 `INSTALL-VERIFICATION.md`。
+- 此次不新增付费模型或设备业务任务，不替换 CUA 插件，也不接入 agent-desktop。此前网关联调记录不冒充本次发布的新增外部验证。
+
+## 升级
+
+已有安装等待任务结束后运行 `update.bat` 或 `update-cn.bat`，重启 ARTEMIS；Codex 用户重新加载 MCP。新安装下载 `artemis-2.7.0-source.zip`，对照 `SHA256SUMS.txt` 校验后解压启动，不覆盖正在运行的安装目录。
+
+**升级会将旧的全局 HTTP 配置迁移为 WebSocket。** 网关暂不支持时，在升级后切回 HTTP；之后不会再自动重置。直接编辑 `.env` 时，保留 `ARTEMIS_TRANSPORT_DEFAULT_VERSION=1` 并设置 `OPENAI_TRANSPORT=http`，重启后生效。

docs/releases/2.7.1.md

@@ -0,0 +1,37 @@
+# ARTEMIS 2.7.1
+
+本版在正式 2.7.0 基础上改进 ADB 一致性、模型提取失败处理和故障证据留存，便于排查设备离线及批次中断。
+
+## 任务操作与录屏共用 ADB
+
+- scrcpy 录屏显式使用 ARTEMIS 选定的 ADB 可执行文件和服务端地址，覆盖普通录屏、分段及驱动直接录屏入口，减少各自选择不同 ADB 的情况。
+- 运行环境诊断记录解析到的 ADB/scrcpy 路径、版本和 SHA256，以及可取得的本机 ADB 监听进程信息；远程服务端进程保留未知。
+
+## 模型空结果与应用查找
+
+- 结构化模型输出为空或解析失败时，进入现有的有限重试和兜底流程；Hopper 对“已找到但提取内容为空”的结果也执行校验，取消请求仍直接中断。
+- 模型提取或初始化失败明确报错，与有效查询后未找到已安装应用分开；仅缓存已确认安装的包名，避免失败结果影响后续查找。
+
+## 故障留档与批次恢复
+
+- 新任务在开始和失败收尾时分别保存运行环境快照，包含程序/Python 版本、可取得的源码提交与本地修改标记、工具来源、宿主内存及任务进程内存。Windows 系统提交量、提交上限和余量按字节记录，不能当作页面文件单独大小。
+- 单任务“全部资料”及批次 ZIP 各任务目录增加 `运行环境诊断.json`。升级前的批次 ZIP 缓存会重新打包；旧任务未采集的运行信息保留未知，导出时不使用当前环境补写历史。
+- 服务重启后保留原暂停原因，恢复事件单独记录，批次仍需手动继续。诊断采集失败仅记日志，不改变任务结果。
+- 快照仅在上述时点采集，不增加持续内存监控、保护阈值、自动暂停或录屏处理并发限制。它们用于减少干扰和保留证据，尚不能证明本次 ADB 崩溃的唯一原因，也不代表宿主内存压力已经解决。
+
+详见[工作区诊断说明](../automation-workspace.md)。
+
+## 版本与验证
+
+- 主程序与前端根版本升级到 2.7.1；SDK、插件、Android 助手及锁定的第三方依赖保持现有独立版本。
+- 回归覆盖录屏 ADB 环境隔离、模型空结果与解析错误重试、Hopper 兜底与取消、包名缓存、批次恢复、诊断导出和旧 ZIP 缓存升级。
+- 最终源码 ZIP 使用独立解压目录和空 npm 缓存，分别验证官方源和国内镜像的锁定安装与生产构建。实际运行时、源码提交、SHA256、检查结果和未测范围见附件 `INSTALL-VERIFICATION.md`。
+- 本次不恢复用户批次，不新增实机业务测试或付费模型调用；自动化回归不能代替现场内存压力与 ADB 崩溃复现。
+
+## 升级
+
+已有安装等待任务结束后运行 `update.bat` 或 `update-cn.bat`，再重启 ARTEMIS；Codex 用户重新加载 MCP。升级程序会终止本项目运行进程，请保留需要继续的暂停任务。
+
+新安装下载 `artemis-2.7.1-source.zip`，对照 `SHA256SUMS.txt` 校验后解压启动，不覆盖正在运行的安装目录。
+
+继续沿用 [2.7.0 的 WebSocket 默认协议及一次性配置迁移](2.7.0.md#默认协议与旧配置迁移)。已经手动切回 HTTP 的配置保持原选择。

mcp_server/catalog.json

@@ -437,7 +437,7 @@
     },
     {
       "name": "mobile_config",
-      "description": "Read configuration or explicitly save non-secret defaults for FUTURE tasks.\n\nOnly set persist_defaults=true when the user requests saved/global configuration.\nmodel changes: model (preset or custom), base_url, reasoning_effort\n(default/none/minimal/low/medium/high/xhigh/max), timeout_seconds, max_retries,\nretry_interval_seconds, rate_limit_retry_seconds, tool_fallback_model,\ntool_fallback_reasoning_effort, tool_fallback_max_retries,\ntool_fallback_no_progress_attempts. 'default' omits reasoning;\nunsupported values fail visibly.\ncapabilities: video_enabled and recording_enabled are independent; outputter_enabled,\nflash_final_check_enabled (false; Flash exit-only AI review),\nstep_summarizer_enabled, screenshot_limit_enabled, task_timeout_seconds (default 1800),\nobservation_timeout_seconds (60), observation_max_attempts (3),\ndevice_command_timeout_seconds (10), action_failure_attempts (3),\nworker_cleanup_timeout_seconds (30), adb_auto_connect (true),\nadb_discovery_interval_seconds (10), adb_discovery_timeout_seconds (5).\nreports: formats=['docx','xlsx'] (at least one).\nexploration: model (default gpt-6.1-terra), reasoning_effort (high), timeout_seconds\n(1800 active seconds), demonstration_timeout_seconds (900), demonstration_warning_seconds (60).\nAPI keys are never accepted or returned: use the Web UI or .env for credentials.\nSaving settings does not change an already running task. Flash/Pro are execution\nprofiles, not model names. Model retries do not repeat device test cases.\n",
+      "description": "Read configuration or explicitly save non-secret defaults for FUTURE tasks.\n\nOnly set persist_defaults=true when the user requests saved/global configuration.\nmodel changes: model (preset or custom), base_url, transport (http/websocket), reasoning_effort\n(default/none/minimal/low/medium/high/xhigh/max), timeout_seconds, max_retries,\nretry_interval_seconds, rate_limit_retry_seconds, tool_fallback_model,\ntool_fallback_reasoning_effort, tool_fallback_max_retries,\ntool_fallback_no_progress_attempts. 'default' omits reasoning;\nunsupported values fail visibly.\ncapabilities: video_enabled and recording_enabled are independent; outputter_enabled,\nflash_final_check_enabled (false; Flash exit-only AI review),\nstep_summarizer_enabled, screenshot_limit_enabled, task_timeout_seconds (default 1800),\nobservation_timeout_seconds (60), observation_max_attempts (3),\ndevice_command_timeout_seconds (10), action_failure_attempts (3),\nworker_cleanup_timeout_seconds (30), adb_auto_connect (true),\nadb_discovery_interval_seconds (10), adb_discovery_timeout_seconds (5).\nreports: formats=['docx','xlsx'] (at least one).\nexploration: model (default gpt-6.1-terra), reasoning_effort (high), timeout_seconds\n(1800 active seconds), demonstration_timeout_seconds (900), demonstration_warning_seconds (60).\nAPI keys are never accepted or returned: use the Web UI or .env for credentials.\nSaving settings does not change an already running task. Flash/Pro are execution\nprofiles, not model names. Model retries do not repeat device test cases.\n",
       "inputSchema": {
         "properties": {
           "section": {

mcp_server/tools/automation.py

@@ -14,6 +14,7 @@ from mcp_server.base import mcp
 
 EXPORT_DIR = ROOT_DIR / "artifacts" / "mcp_exports"
 MODEL_FIELDS = {
+    "transport",
     "base_url",
     "model",
     "timeout_seconds",
@@ -186,7 +187,7 @@ async def mobile_config(
     """Read configuration or explicitly save non-secret defaults for FUTURE tasks.
 
     Only set persist_defaults=true when the user requests saved/global configuration.
-    model changes: model (preset or custom), base_url, reasoning_effort
+    model changes: model (preset or custom), base_url, transport (http/websocket), reasoning_effort
     (default/none/minimal/low/medium/high/xhigh/max), timeout_seconds, max_retries,
     retry_interval_seconds, rate_limit_retry_seconds, tool_fallback_model,
     tool_fallback_reasoning_effort, tool_fallback_max_retries,

pyproject.toml

@@ -14,7 +14,7 @@
 
 [project]
 name = "artemis"
-version = "2.6.0"
+version = "2.7.1"
 description = "Autonomous AI Assistant for Mobile Automation"
 
 authors = [
@@ -57,7 +57,7 @@ dependencies = [
     "uvicorn[standard]>=0.30.1",
     "httpx>=0.28.1",
     "requests>=2.31.0",
-    "websockets>=12.0",
+    "websockets>=15.0",
     "sseclient-py>=1.8.0",
     # Image, Video & Multimodal Perception
     "pillow>=13.0.0",

skills/artemis-use/references/modes.md

@@ -16,6 +16,8 @@
 
 ## 模式、模型与配置
 
+网页模型配置和 `mobile_config(section="model")` 提供 `transport=http/websocket`，默认 WebSocket；首次升级启动也会将旧全局 HTTP 配置一次性迁移为 WebSocket，之后手动切回 HTTP 会被保留。用户明确要求保存默认值时可设置 `changes={"transport":"websocket"}, persist_defaults=true`；新任务的 Flash、Pro、探索和辅助请求共用此选择。WebSocket 在历史一致时仅上传新增输入，网关须支持 Responses WebSocket；压缩、重连会发送完整上下文，上传减少不能当作 token 减少或缓存命中。不支持协议时应查看错误并确认网关能力，不自动改写用户配置。
+
 - `mobile_run_task.model` 取 `Flash` / `Pro`。API 模型名放 `runtime_overrides.model`；思考强度放 `runtime_overrides.reasoning_effort`。批次使用 `task_config.profile` 选择模式，`task_config.model` 选择 API 模型。
 - 探索独立配置，默认 `gpt-6.1-terra/high`，支持 `gpt-6-astra` 及已配置网关支持的自定义模型。执行、示范分析、提炼共用该探索配置；不改变 Flash/Pro 默认值。
 - 默认 high、单次模型请求时限 300 秒。`default` 表示不发送思考强度，`none` 是显式关闭，不能混为一谈；接口不支持参数时如实报告，不擅自降级。

tests/unit/admin_console/test_automation.py

@@ -181,18 +181,23 @@ def test_completed_batch_reuses_existing_archive(client, isolated, monkeypatch):
     exporter.assert_not_called()
 
 
-def test_legacy_batch_archive_is_rebuilt_with_current_recording_layout(client, isolated):
+@pytest.mark.parametrize("signature", [None, "artemis-batch-layout-3"])
+def test_legacy_batch_archive_is_rebuilt_with_runtime_diagnostics(client, isolated, signature):
     batch = service.create_batch(payload())
     batch_id = batch["id"]
     isolated.put("batch", batch_id, {**batch, "status": "completed", "report_ready": True})
     path = reports.EXPORT_ROOT / "batches" / batch_id / "批次报告.zip"
     path.parent.mkdir(parents=True)
     path.write_bytes(b"legacy archive")
+    if signature:
+        reports.remember(path, signature)
     base = f"/api/automation/batches/{batch_id}/archive"
     assert client.get(base + "/status").json()["status"] == "idle"
     response = client.get(base)
     assert response.status_code == 200
     assert zipfile.is_zipfile(BytesIO(response.content))
+    with zipfile.ZipFile(BytesIO(response.content)) as archive:
+        assert len([name for name in archive.namelist() if name.endswith("运行环境诊断.json")]) == 2
     assert client.get(base + "/status").json()["status"] == "ready"
 
 
@@ -721,7 +726,7 @@ async def test_fail_fast_includes_skipped_reports(isolated, monkeypatch):
     assert [j["status"] for j in isolated.jobs(batch["id"])] == ["failed", "skipped"]
     archive = reports.export_batch(batch["id"])
     with zipfile.ZipFile(archive) as z:
-        assert len(z.namelist()) == 10  # summaries + reports, diagnostics and evidence per task
+        assert len(z.namelist()) == 12  # summaries + reports, diagnostics and evidence per task
         diagnostics = [name for name in z.namelist() if name.endswith("模型请求诊断.json")]
         assert len(diagnostics) == 2
         for name in diagnostics:
@@ -1304,6 +1309,69 @@ def test_batch_zip_contains_each_tasks_android_artifacts(isolated):
         assert {archive.read(name).decode() for name in names} == {"任务0", "任务1"}
 
 
+@pytest.mark.parametrize("pause_source", ["device", "initialization", "user", "model"])
+def test_restart_retains_original_pause_reason_and_archives_recovery(isolated, pause_source):
+    batch = service.create_batch(payload())
+    reason = "原始暂停原因"
+    isolated.put(
+        "batch",
+        batch["id"],
+        {**batch, "status": "paused", "pause_source": pause_source, "reason": reason},
+    )
+    service.recover_batches()
+    recovered = isolated.get("batch", batch["id"])
+    assert recovered["status"] == "paused"
+    assert recovered["pause_source"] == "recovery"
+    assert recovered["reason"] == reason
+    event = recovered["recovery_events"][0]
+    assert event["previous_reason"] == reason
+    assert event["previous_pause_source"] == pause_source
+    assert "服务已重启" in event["message"]
+    assert all(job["status"] == "pending" for job in isolated.jobs(batch["id"]))
+    service.recover_batches()
+    assert isolated.get("batch", batch["id"])["reason"] == reason
+
+
+def test_runtime_evidence_is_exported_from_the_task_without_resampling(isolated, monkeypatch):
+    batch = service.create_batch(payload())
+    first, second = isolated.jobs(batch["id"])
+    snapshot = {
+        "schema_version": 1,
+        "failure": {
+            "captured_at": 1,
+            "runtime": {"source_commit": "worker-commit"},
+            "memory": {"available_bytes": 163},
+        },
+    }
+    directory = reports.TRACES_PATH / first["session_id"]
+    directory.mkdir(parents=True)
+    (directory / "runtime_diagnostics.json").write_text(json.dumps(snapshot), encoding="utf-8")
+    isolated.put(
+        "batch",
+        batch["id"],
+        {**batch, "status": "paused", "reason": "设备离线", "pause_source": "device"},
+    )
+    service.recover_batches()
+
+    def assert_snapshots(archive_path):
+        with zipfile.ZipFile(archive_path) as archive:
+            records = [
+                json.loads(archive.read(name))
+                for name in archive.namelist()
+                if name.endswith("运行环境诊断.json")
+            ]
+        recorded = next(record for record in records if record["runtime"])
+        assert recorded["runtime"] == snapshot
+        assert recorded["batch"]["reason"] == "设备离线"
+        assert recorded["batch"]["recovery_events"][0]["previous_pause_source"] == "device"
+        return records
+
+    assert_snapshots(reports.export_session_media(first["session_id"], "all"))
+    records = assert_snapshots(reports.export_batch(batch["id"]))
+    assert len(records) == 2
+    assert next(record for record in records if record["runtime"] is None)["runtime"] is None
+
+
 def test_all_material_cache_reuses_reports_and_invalidates_on_review_or_new_files(
     isolated, monkeypatch
 ):

tests/unit/admin_console/test_model_settings.py

@@ -9,6 +9,7 @@ from apps.admin_console.services import model_settings_service as service
 from artemis.config.settings import Settings, settings
 from artemis.config.agent import load_agent_config
 from artemis.config.llm import parse_llm_config
+from artemis.config.environment import migrate_transport_default, TRANSPORT_DEFAULT_VERSION_KEY
 
 
 @pytest.fixture
@@ -21,6 +22,7 @@ def isolated_settings(monkeypatch, tmp_path):
         "OPENAI_MODEL",
         "OPENAI_BASE_URL",
         "OPENAI_TIMEOUT_SECONDS",
+        "OPENAI_TRANSPORT",
         "OPENAI_MAX_RETRIES",
         "OPENAI_RETRY_INTERVAL_SECONDS",
         "OPENAI_RATE_LIMIT_RETRY_SECONDS",
@@ -44,12 +46,14 @@ def isolated_settings(monkeypatch, tmp_path):
     for key in keys:
         monkeypatch.setattr(settings, key, getattr(settings, key))
         monkeypatch.delenv(key, raising=False)
+    monkeypatch.delenv(TRANSPORT_DEFAULT_VERSION_KEY, raising=False)
     monkeypatch.setattr(settings, "OPENAI_API_KEY", SecretStr("existing-test-secret"))
     return env
 
 
 def test_save_reload_and_blank_key_preserves_secret(isolated_settings):
     update = service.ModelSettingsUpdate(
+        transport="websocket",
         base_url="https://gateway.example/v1/",
         model="my-model",
         api_key="new-test-secret",
@@ -63,6 +67,9 @@ def test_save_reload_and_blank_key_preserves_secret(isolated_settings):
     )
     result = service.save_model_settings(update)
     assert result["api_key_configured"]
+    assert result["transport"] == "websocket"
+    assert dotenv_values(isolated_settings)["OPENAI_TRANSPORT"] == "websocket"
+    assert service.ModelSettingsUpdate(model="test").transport is None
     assert "new-test-secret" not in repr(result)
     saved = dotenv_values(isolated_settings)
     assert saved["OPENAI_API_KEY"] == "new-test-secret"
@@ -81,6 +88,16 @@ def test_save_reload_and_blank_key_preserves_secret(isolated_settings):
     assert dotenv_values(isolated_settings)["OPENAI_API_KEY"] == "new-test-secret"
 
 
+def test_saved_http_opt_out_survives_restart_migration(isolated_settings):
+    result = service.save_model_settings(
+        service.ModelSettingsUpdate(model="test", transport="http")
+    )
+    assert result["transport"] == "http"
+    assert dotenv_values(isolated_settings)[TRANSPORT_DEFAULT_VERSION_KEY] == "1"
+    assert migrate_transport_default(isolated_settings) is False
+    assert Settings(_env_file=isolated_settings).OPENAI_TRANSPORT == "http"
+
+
 def test_new_install_uses_playones_endpoint_without_overriding_saved_url(monkeypatch):
     monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
     expected = "https://api-n-cd.playones.com/v1"
@@ -226,9 +243,7 @@ def test_screenshot_json_can_override_environment(monkeypatch, tmp_path):
 
 def test_failed_write_keeps_running_settings(monkeypatch, isolated_settings):
     previous = settings.OPENAI_MODEL
-    monkeypatch.setattr(
-        service.os, "replace", lambda *args: (_ for _ in ()).throw(OSError("read only"))
-    )
+    monkeypatch.setattr(os, "replace", lambda *args: (_ for _ in ()).throw(OSError("read only")))
     with pytest.raises(OSError):
         service.save_model_settings(service.ModelSettingsUpdate(model="not-saved"))
     assert settings.OPENAI_MODEL == previous

tests/unit/agents/test_hopper.py

@@ -0,0 +1,64 @@
+"""Hopper failures must not become successful negative lookups."""
+
+import asyncio
+from importlib import import_module
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from artemis.agents.hopper.hopper import HopperError, HopperOutput
+
+module = import_module("artemis.agents.hopper.hopper")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invalid", [None, {}, {"found": True, "output": " ", "reason": "Empty"}])
+async def test_invalid_extraction_switches_to_fallback(monkeypatch, invalid):
+    good = HopperOutput(found=True, output="com.example.app", reason="Matched")
+    primary = SimpleNamespace(ainvoke=AsyncMock(return_value=invalid))
+    secondary = SimpleNamespace(ainvoke=AsyncMock(return_value=good))
+    factory = Mock(
+        side_effect=[
+            SimpleNamespace(with_structured_output=lambda _: primary),
+            SimpleNamespace(with_structured_output=lambda _: secondary),
+        ]
+    )
+    monkeypatch.setattr(module, "get_llm", factory)
+    assert (
+        await module.hopper(SimpleNamespace(data_engine=None), "Find app", "com.example.app")
+        == good
+    )
+    assert primary.ainvoke.await_count == secondary.ainvoke.await_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_fallback", [False, True])
+async def test_exhausted_extraction_raises_a_distinct_error(monkeypatch, use_fallback):
+    model = SimpleNamespace(ainvoke=AsyncMock(return_value=None))
+    monkeypatch.setattr(
+        module, "get_llm", lambda **_: SimpleNamespace(with_structured_output=lambda _: model)
+    )
+    with pytest.raises(HopperError, match="模型未能完成数据提取"):
+        await module.hopper(
+            SimpleNamespace(data_engine=None), "Find app", "packages", use_fallback=use_fallback
+        )
+    assert model.ainvoke.await_count == (2 if use_fallback else 1)
+
+
+@pytest.mark.asyncio
+async def test_cancelled_extraction_never_uses_fallback(monkeypatch):
+    model = SimpleNamespace(ainvoke=AsyncMock(side_effect=asyncio.CancelledError))
+    monkeypatch.setattr(
+        module, "get_llm", lambda **_: SimpleNamespace(with_structured_output=lambda _: model)
+    )
+    with pytest.raises(asyncio.CancelledError):
+        await module.hopper(SimpleNamespace(data_engine=None), "Find app", "packages")
+    assert model.ainvoke.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_model_setup_failure_is_not_reported_as_missing_data(monkeypatch):
+    monkeypatch.setattr(module, "get_llm", Mock(side_effect=ValueError("invalid model settings")))
+    with pytest.raises(HopperError, match="模型未能完成数据提取"):
+        await module.hopper(SimpleNamespace(data_engine=None), "Find app", "packages")

tests/unit/core/test_transport_defaults.py

@@ -0,0 +1,117 @@
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+from dotenv import dotenv_values
+import pytest
+
+from artemis.config.environment import (
+    migrate_transport_default,
+    persist_environment,
+    TRANSPORT_DEFAULT_VERSION_KEY,
+)
+
+
+@pytest.fixture(autouse=True)
+def isolate_transport(monkeypatch):
+    monkeypatch.setenv("OPENAI_TRANSPORT", "http")
+    monkeypatch.delenv(TRANSPORT_DEFAULT_VERSION_KEY, raising=False)
+
+
+@pytest.mark.parametrize("old", ["", "OPENAI_TRANSPORT=http\n", "OPENAI_TRANSPORT=websocket\n"])
+def test_upgrade_migrates_legacy_preferences_once_and_preserves_config(tmp_path, old):
+    path = tmp_path / ".env"
+    preserved = (
+        "# 用户注释\nOPENAI_API_KEY='test-secret'\nOPENAI_BASE_URL=https://test.invalid/v1\n"
+    )
+    path.write_text(preserved + old, encoding="utf-8")
+    assert migrate_transport_default(path) is True
+    saved = dotenv_values(path)
+    assert saved["OPENAI_TRANSPORT"] == os.environ["OPENAI_TRANSPORT"] == "websocket"
+    assert saved[TRANSPORT_DEFAULT_VERSION_KEY] == "1"
+    assert path.read_text(encoding="utf-8").startswith(preserved)
+    previous = path.read_bytes()
+    assert migrate_transport_default(path) is False
+    assert path.read_bytes() == previous
+    persist_environment(path, {"OPENAI_TRANSPORT": "http"})
+    assert migrate_transport_default(path) is False
+    assert dotenv_values(path)["OPENAI_TRANSPORT"] == "http"
+
+
+def test_failed_replace_preserves_original_and_can_retry(tmp_path, monkeypatch):
+    from artemis.config import environment
+
+    path = tmp_path / ".env"
+    original = "OPENAI_TRANSPORT=http\nUNRELATED=keep\n"
+    path.write_text(original, encoding="utf-8")
+
+    def fail(*_):
+        raise OSError("simulated disk error")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(environment.os, "replace", fail)
+        with pytest.raises(OSError):
+            migrate_transport_default(path)
+    assert path.read_text(encoding="utf-8") == original
+    assert os.environ["OPENAI_TRANSPORT"] == "http"
+    assert not list(tmp_path.glob(".env-*"))
+    assert migrate_transport_default(path) is True
+
+
+@pytest.mark.parametrize("version", [1, 2])
+def test_completed_migration_preserves_http_without_write_access(tmp_path, monkeypatch, version):
+    from artemis.config import environment
+
+    path = tmp_path / ".env"
+    path.write_text(
+        f"OPENAI_TRANSPORT=http\n{TRANSPORT_DEFAULT_VERSION_KEY}={version}\n", encoding="utf-8"
+    )
+
+    def denied(*_, **__):
+        raise PermissionError("read-only deployment")
+
+    monkeypatch.setattr(environment, "FileLock", denied)
+    assert migrate_transport_default(path) is False
+    assert dotenv_values(path)["OPENAI_TRANSPORT"] == "http"
+
+
+@pytest.mark.parametrize("new_install", [False, True])
+def test_fresh_process_defaults_and_saved_opt_out(tmp_path, new_install):
+    path = tmp_path / ".env"
+    if not new_install:
+        path.write_text("OPENAI_TRANSPORT=http\n", encoding="utf-8")
+    env = {
+        k: v
+        for k, v in os.environ.items()
+        if k not in {"OPENAI_TRANSPORT", TRANSPORT_DEFAULT_VERSION_KEY}
+    }
+    env["ARTEMIS_APP_DIR"] = str(tmp_path)
+    env["PYTHONUTF8"] = "1"
+    env["PYTHONPATH"] = str(Path(__file__).resolve().parents[3])
+    code = """
+import json
+from artemis.config.settings import settings
+from artemis.config.llm import LLM, parse_llm_config
+from artemis.llm.router import ModelEndpoint
+print(json.dumps([settings.OPENAI_TRANSPORT, LLM().transport, ModelEndpoint().transport, parse_llm_config().operator.transport]))
+"""
+
+    def start():
+        result = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd=tmp_path,
+            env=env,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            timeout=20,
+            check=True,
+        )
+        return json.loads(result.stdout.splitlines()[-1])
+
+    assert start() == ["websocket"] * 4
+    assert dotenv_values(path)[TRANSPORT_DEFAULT_VERSION_KEY] == "1"
+    persist_environment(path, {"OPENAI_TRANSPORT": "http"})
+    assert start() == ["http"] * 4

tests/unit/mcp/test_automation_tools.py

@@ -72,6 +72,8 @@ async def test_config_requires_explicit_persistence_and_excludes_secrets(monkeyp
     payload = request.call_args.kwargs["json"]
     assert payload["rate_limit_retry_seconds"] == 90
     assert payload["tool_fallback_model"] == "gpt-6-luna"
+    await tools.mobile_config(changes={"transport": "websocket"}, persist_defaults=True)
+    assert request.call_args.kwargs["json"]["transport"] == "websocket"
 
 
 @pytest.mark.asyncio

tests/unit/runtime/test_runtime_diagnostics.py

@@ -0,0 +1,144 @@
+"""Runtime snapshots are bounded, read-only evidence, never a task gate."""
+
+import ctypes
+import hashlib
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import psutil
+import pytest
+
+from artemis.runtime import diagnostics as module
+from artemis.runtime.adb_endpoint import AdbEndpoint
+
+
+@pytest.fixture(autouse=True)
+def isolated_identity_cache():
+    identity = module.runtime_identity
+    identity.cache_clear()
+    yield
+    identity.cache_clear()
+
+
+def test_tool_identity_records_exact_binary_hash_and_version(monkeypatch, tmp_path):
+    adb = tmp_path / "platform tools" / "adb.exe"
+    adb.parent.mkdir()
+    adb.write_bytes(b"selected adb")
+    monkeypatch.setattr(module.toolchain, "resolve", lambda _: str(adb))
+    command = Mock(return_value="Android Debug Bridge version test")
+    monkeypatch.setattr(module, "_command", command)
+    record = module._tool_identity("adb")
+    assert record["path"] == str(adb)
+    assert record["sha256"] == hashlib.sha256(b"selected adb").hexdigest()
+    assert record["version"] == command.return_value
+    command.assert_called_once_with([str(adb), "version"])
+
+
+def test_unknown_source_and_archive_identity_are_not_invented(monkeypatch, tmp_path):
+    monkeypatch.setattr(module, "ROOT_DIR", tmp_path)
+    monkeypatch.setattr(module, "get_version", lambda: "test-version")
+    command = Mock()
+    monkeypatch.setattr(module, "_command", command)
+    monkeypatch.setenv("OPENAI_API_KEY", "private-test-key")
+    record = module.runtime_identity(False)
+    assert record["source_commit"] is None
+    assert record["source_modified"] is None
+    assert record["installation_archive_sha256"] is None
+    assert record["artemis_version"] == "test-version"
+    assert record["tools"] == {}
+    assert "private-test-key" not in json.dumps(record)
+    command.assert_not_called()
+
+
+def test_version_commands_are_bounded_and_never_repair_devices(monkeypatch):
+    run = Mock(return_value=SimpleNamespace(stdout="version", returncode=0))
+    monkeypatch.setattr(module.subprocess, "run", run)
+    assert module._command(["adb", "version"]) == "version"
+    assert run.call_args.kwargs["timeout"] == 3
+    assert run.call_args.kwargs["check"] is False
+    assert run.call_args.args == (["adb", "version"],)
+    run.side_effect = OSError("Cannot create process")
+    assert module._command(["adb", "version"]) is None
+
+
+@pytest.mark.skipif(module.sys.platform != "win32", reason="Windows PSAPI structure")
+def test_system_commit_converts_pages_to_bytes(monkeypatch):
+    def query(pointer, size):
+        values = pointer._obj
+        assert size == ctypes.sizeof(module._PerformanceInformation)
+        values.PageSize = 4096
+        values.CommitTotal = 900
+        values.CommitLimit = 1000
+        return True
+
+    dll = SimpleNamespace(GetPerformanceInfo=Mock(side_effect=query))
+    monkeypatch.setattr(module.ctypes, "WinDLL", lambda *args, **kwargs: dll)
+    result = module._windows_commit()
+    assert result["committed_bytes"] == 900 * 4096
+    assert result["limit_bytes"] == 1000 * 4096
+    assert result["available_bytes"] == 100 * 4096
+
+
+def test_remote_adb_process_is_explicitly_unknown(monkeypatch):
+    monkeypatch.setattr(
+        module, "current_adb_endpoint", lambda: AdbEndpoint.create("192.0.2.10", 5037)
+    )
+    connections = Mock()
+    monkeypatch.setattr(module.psutil, "net_connections", connections)
+    result = module._adb_server()
+    assert result["local_listener"] is None
+    assert result["endpoint"]["host"] == "192.0.2.10"
+    connections.assert_not_called()
+
+
+def test_local_adb_listener_identity_excludes_unrelated_bindings(monkeypatch):
+    monkeypatch.setattr(module, "current_adb_endpoint", lambda: AdbEndpoint.local())
+    connections = [
+        SimpleNamespace(status=psutil.CONN_LISTEN, laddr=SimpleNamespace(ip=ip, port=5037), pid=pid)
+        for ip, pid in [("192.0.2.10", 5), ("127.0.0.1", 6)]
+    ]
+    monkeypatch.setattr(module.psutil, "net_connections", lambda **_: connections)
+    process = SimpleNamespace(pid=6, name=lambda: "adb.exe", exe=lambda: "selected/adb.exe")
+    lookup = Mock(return_value=process)
+    monkeypatch.setattr(module.psutil, "Process", lookup)
+    assert module._adb_server()["local_listener"] == {
+        "pid": 6,
+        "name": "adb.exe",
+        "path": "selected/adb.exe",
+    }
+    lookup.assert_called_once_with(6)
+
+
+def test_start_snapshot_survives_failure_and_repeated_start(monkeypatch, tmp_path):
+    identity = Mock(return_value={"source_commit": "test-commit"})
+    memory = Mock(side_effect=[{"available_bytes": 100}, {"available_bytes": 1}])
+    monkeypatch.setattr(module, "runtime_identity", identity)
+    monkeypatch.setattr(module, "_memory_snapshot", memory)
+    monkeypatch.setattr(module, "_adb_server", lambda: {"local_listener": None})
+    module.write_runtime_diagnostics(tmp_path, "start")
+    module.write_runtime_diagnostics(tmp_path, "start")
+    module.write_runtime_diagnostics(tmp_path, "failure")
+    payload = json.loads((tmp_path / module.FILENAME).read_text(encoding="utf-8"))
+    assert payload["start"]["memory"]["available_bytes"] == 100
+    assert payload["failure"]["memory"]["available_bytes"] == 1
+    assert payload["failure"]["runtime"]["source_commit"] == "test-commit"
+    assert memory.call_count == 2
+    assert not (tmp_path / "runtime_diagnostics.tmp").exists()
+
+
+def test_snapshot_failure_cannot_fail_a_task(monkeypatch, tmp_path):
+    monkeypatch.setattr(module, "runtime_identity", lambda _: {})
+    monkeypatch.setattr(module, "_memory_snapshot", Mock(side_effect=OSError("commit exhausted")))
+    module.write_runtime_diagnostics(tmp_path, "failure")
+    assert not (tmp_path / module.FILENAME).exists()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error", [OSError("commit exhausted"), RuntimeError("cannot start thread")]
+)
+async def test_snapshot_thread_creation_failure_is_optional(monkeypatch, tmp_path, error):
+    monkeypatch.setattr(module.asyncio, "to_thread", AsyncMock(side_effect=error))
+    await module.capture_runtime_diagnostics(tmp_path, "start")
+    assert not (tmp_path / module.FILENAME).exists()

tests/unit/sdk/test_agent_trace_finalization.py

@@ -22,6 +22,22 @@ from artemis.runtime.device_lock import DeviceBusyError
 from artemis.sdk.types.exceptions import AgentError
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["failed", "completed", "cancelled"])
+async def test_failure_snapshot_is_written_before_tracing_ends(tmp_path, status):
+    agent = object.__new__(Agent)
+    task = MagicMock(status=status)
+    context = MagicMock(execution_target="computer", execution_setup=None)
+    context.data_engine.base_dir = tmp_path
+    with patch("artemis.runtime.diagnostics.write_runtime_diagnostics") as snapshot:
+        await agent._finalize_tracing(task, context)
+    if status == "failed":
+        snapshot.assert_called_once_with(tmp_path, "failure", android=False)
+    else:
+        snapshot.assert_not_called()
+    context.data_engine.end_session.assert_called_once_with(status=status)
+
+
 @pytest.mark.asyncio
 async def test_trace_finalization_failure_does_not_escape():
     agent = object.__new__(Agent)

tests/unit/test_llm_gateway.py

@@ -1,6 +1,7 @@
 """Tests for the LLM gateway: complete(), classified recovery, and fallback."""
 
 from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage
 import pytest
@@ -16,6 +17,55 @@ from artemis.services.llm import (
 )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "invalid",
+    [
+        None,
+        {"raw": AIMessage(content=""), "parsed": None, "parsing_error": None},
+        {
+            "raw": AIMessage(content=""),
+            "parsed": None,
+            "parsing_error": ValueError("Invalid schema"),
+        },
+    ],
+)
+async def test_missing_structured_output_retries_inside_existing_budget(invalid):
+    expected = {"found": True, "output": "com.example.app", "reason": "Matched"}
+    base = SimpleNamespace(
+        ainvoke=AsyncMock(
+            side_effect=[
+                invalid,
+                {"raw": AIMessage(content=""), "parsed": expected, "parsing_error": None},
+            ]
+        )
+    )
+    model = RobustChatModelWrapper(base, structured_mode="parsed")
+    assert await model.ainvoke([]) == expected
+    assert base.ainvoke.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_empty_structured_output_exhausts_then_uses_fallback():
+    primary = SimpleNamespace(
+        ainvoke=AsyncMock(return_value={"raw": AIMessage(content=""), "parsed": None})
+    )
+    expected = {"found": False, "output": None, "reason": "No match"}
+    secondary = SimpleNamespace(
+        ainvoke=AsyncMock(return_value={"raw": AIMessage(content=""), "parsed": expected})
+    )
+    main = RobustChatModelWrapper(
+        primary, endpoint=ModelEndpoint(model_name="primary"), structured_mode="parsed"
+    )
+    fallback = RobustChatModelWrapper(
+        secondary, endpoint=ModelEndpoint(model_name="fallback"), structured_mode="parsed"
+    )
+    result = await with_fallback(lambda: main.ainvoke([]), lambda: fallback.ainvoke([]))
+    assert result == expected
+    assert primary.ainvoke.await_count == 2
+    assert secondary.ainvoke.await_count == 1
+
+
 @pytest.fixture(autouse=True)
 def _isolate_gateway_state(monkeypatch, tmp_path):
     llm_service._ENDPOINT_BREAKER._states.clear()

tests/unit/test_llm_response_integrity.py

@@ -123,7 +123,7 @@ async def transport_case(monkeypatch, tmp_path, request):
     )
     llm_service._ENDPOINT_BREAKER._states.clear()
     llm_service._NON_STREAMING_ENDPOINTS.clear()
-    endpoint = ModelEndpoint(model_name="test-model", max_retries=0)
+    endpoint = ModelEndpoint(model_name="test-model", transport="http", max_retries=0)
     model = ModelFactory.create_model(endpoint)
     assert isinstance(model, ChatOpenAI)
     yield SimpleNamespace(

tests/unit/test_model_proxy.py

@@ -17,7 +17,7 @@ async def test_sync_and_async_clients_follow_env_only_proxy_policy(monkeypatch,
     monkeypatch.delenv("ARTEMIS_FAKE_LLM", raising=False)
     monkeypatch.setattr(settings, "OPENAI_USE_SYSTEM_PROXY", use_proxy)
     model = ModelFactory.create_model(
-        ModelEndpoint(api_key="test-only", api_base="https://example.invalid/v1")
+        ModelEndpoint(api_key="test-only", api_base="https://example.invalid/v1", transport="http")
     )
     try:
         for client in (model.http_client, model.http_async_client):

tests/unit/test_openai_responses.py

@@ -43,6 +43,9 @@ def transport_model(monkeypatch):
     from langchain_openai import ChatOpenAI
     import openai
 
+    # These fixtures exercise the explicitly selected HTTP compatibility path.
+    monkeypatch.setattr(settings, "OPENAI_TRANSPORT", "http")
+
     requests = []
     outputs = []
 

tests/unit/test_responses_websocket.py

@@ -0,0 +1,607 @@
+"""Real local WebSocket/SDK round trips; no device, credential, or paid endpoint."""
+
+import asyncio
+from contextlib import asynccontextmanager
+from copy import deepcopy
+import json
+
+import httpx
+from langchain_core.messages import HumanMessage, ToolMessage
+import pytest
+import pytest_asyncio
+from websockets.asyncio.server import serve
+
+from artemis.llm.diagnostics import ACTIVE_ATTEMPT
+from artemis.llm.response_guard import ResponseIntegrityError, ensure_response_complete
+from artemis.llm.router import ModelEndpoint, ModelFactory
+from artemis.services.llm import RobustChatModelWrapper
+
+
+def text_output(text="OK"):
+    return [
+        {
+            "id": "msg_test",
+            "type": "message",
+            "role": "assistant",
+            "status": "completed",
+            "content": [{"type": "output_text", "text": text, "annotations": [], "logprobs": []}],
+        }
+    ]
+
+
+class Wire:
+    def __init__(self):
+        self.requests = []
+        self.connections = []
+        self.outputs = []
+        self.reject_previous = False
+        self.drop = False
+        self.stall = False
+        self.status = "completed"
+        self.sent = asyncio.Event()
+        self.gate = None
+
+    async def handle(self, socket):
+        self.connections.append(socket)
+        async for raw in socket:
+            body = json.loads(raw)
+            self.requests.append(body)
+            self.sent.set()
+            if self.reject_previous and body.get("previous_response_id"):
+                await socket.send(
+                    json.dumps(
+                        {
+                            "type": "error",
+                            "status": 400,
+                            "error": {"code": "previous_response_not_found"},
+                        }
+                    )
+                )
+                continue
+            response = {
+                "id": f"resp_{len(self.requests)}",
+                "object": "response",
+                "created_at": 1,
+                "status": "in_progress",
+                "model": "test-model",
+                "output": [],
+            }
+            await socket.send(
+                json.dumps({"type": "response.created", "sequence_number": 0, "response": response})
+            )
+            if self.drop:
+                await socket.close()
+                continue
+            if self.stall:
+                await socket.wait_closed()
+                return
+            if self.gate:
+                await self.gate.wait()
+            output = self.outputs.pop(0) if self.outputs else text_output()
+            seq = 0
+            for index, item in enumerate(output):
+                seq += 1
+                await socket.send(
+                    json.dumps(
+                        {
+                            "type": "response.output_item.added",
+                            "sequence_number": seq,
+                            "output_index": index,
+                            "item": {**item, "content": [], "arguments": ""},
+                        }
+                    )
+                )
+                for content_index, block in enumerate(item.get("content", [])):
+                    if block.get("type") != "output_text":
+                        continue
+                    seq += 1
+                    await socket.send(
+                        json.dumps(
+                            {
+                                "type": "response.output_text.delta",
+                                "sequence_number": seq,
+                                "output_index": index,
+                                "content_index": content_index,
+                                "item_id": item["id"],
+                                "delta": block["text"],
+                            }
+                        )
+                    )
+                    seq += 1
+                    await socket.send(
+                        json.dumps(
+                            {
+                                "type": "response.output_text.done",
+                                "sequence_number": seq,
+                                "output_index": index,
+                                "content_index": content_index,
+                                "item_id": item["id"],
+                                "text": block["text"],
+                            }
+                        )
+                    )
+                seq += 1
+                await socket.send(
+                    json.dumps(
+                        {
+                            "type": "response.output_item.done",
+                            "sequence_number": seq,
+                            "output_index": index,
+                            "item": item,
+                        }
+                    )
+                )
+            response.update(
+                status=self.status,
+                output=output,
+                usage={
+                    "input_tokens": 20,
+                    "output_tokens": 3,
+                    "total_tokens": 23,
+                    "input_tokens_details": {"cached_tokens": 10},
+                },
+            )
+            await socket.send(
+                json.dumps(
+                    {
+                        "type": f"response.{self.status}",
+                        "sequence_number": seq + 1,
+                        "response": response,
+                    }
+                )
+            )
+
+
+@pytest_asyncio.fixture
+async def wire(monkeypatch):
+    monkeypatch.delenv("ARTEMIS_FAKE_LLM", raising=False)
+    instance = Wire()
+    async with serve(instance.handle, "127.0.0.1", 0) as server:
+        instance.base = f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}/v1"
+        yield instance
+
+
+@asynccontextmanager
+async def model(wire, **kwargs):
+    endpoint = ModelEndpoint(
+        model_name="test-model",
+        api_key="test-secret",
+        api_base=wire.base,
+        transport="websocket",
+        max_retries=0,
+        timeout_seconds=2,
+        **kwargs,
+    )
+    value = ModelFactory.create_model(endpoint)
+    try:
+        yield value
+    finally:
+        await value.http_async_client.aclose()
+        value.http_client.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stream", [False, True])
+async def test_sdk_roundtrip_incremental_history_and_usage(wire, stream):
+    async with model(wire) as raw:
+        messages = [
+            HumanMessage(
+                content=[
+                    {"type": "text", "text": "Inspect"},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": "data:image/png;base64," + "A" * 80000},
+                    },
+                ]
+            )
+        ]
+
+        async def call(history):
+            if not stream:
+                return await raw.ainvoke(history, prompt_cache_key="task:operator")
+            chunks = [
+                chunk async for chunk in raw.astream(history, prompt_cache_key="task:operator")
+            ]
+            return sum(chunks[1:], chunks[0])
+
+        first = await call(messages)
+        ensure_response_complete(first)
+        second = await call([*messages, first, HumanMessage(content="Continue")])
+        ensure_response_complete(second)
+        assert second.usage_metadata["input_token_details"]["cache_read"] == 10
+        assert wire.requests[1].get("previous_response_id") == "resp_1", {
+            "first": first.content,
+            "tail": wire.requests[1]["input"][1:],
+        }
+        assert wire.requests[1]["input"] == [
+            {"type": "message", "role": "user", "content": "Continue"}
+        ]
+        assert "stream" not in wire.requests[0]
+        assert wire.requests[0]["type"] == "response.create"
+        assert wire.requests[0]["store"] is False
+        assert len(json.dumps(wire.requests[1])) < len(json.dumps(wire.requests[0])) / 100
+        assert len(wire.connections) == 1
+        assert wire.connections[0].request.path == "/v1/responses"
+        assert wire.connections[0].request.headers["Authorization"] == "Bearer test-secret"
+
+
+@pytest.mark.asyncio
+async def test_tool_call_chain_preserves_call_id(wire):
+    wire.outputs.append(
+        [
+            {
+                "id": "fc_test",
+                "type": "function_call",
+                "call_id": "call_test",
+                "name": "lookup",
+                "arguments": '{"x":1}',
+                "status": "completed",
+            }
+        ]
+    )
+    async with model(wire) as raw:
+        messages = [HumanMessage(content="Look up")]
+        first = await raw.ainvoke(messages, prompt_cache_key="task:operator")
+        await raw.ainvoke(
+            [*messages, first, ToolMessage(content="found", tool_call_id="call_test")],
+            prompt_cache_key="task:operator",
+        )
+        assert wire.requests[1]["previous_response_id"] == "resp_1"
+        assert wire.requests[1]["input"] == [
+            {"type": "function_call_output", "call_id": "call_test", "output": "found"}
+        ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "change", ["history", "options", "scope", "new_connection", "server_closed"]
+)
+async def test_chain_resets_on_context_or_connection_change(wire, change):
+    async with model(wire) as raw:
+        messages = [HumanMessage(content="Original")]
+        first = await raw.ainvoke(messages, prompt_cache_key="task:operator")
+        history = [*messages, first, HumanMessage(content="Next")]
+        options = {"prompt_cache_key": "task:operator"}
+        if change == "history":
+            history[0] = HumanMessage(content="Compacted")
+        if change == "options":
+            options["temperature"] = 0.3
+        if change == "scope":
+            options["prompt_cache_key"] = "task:checker"
+        if change == "new_connection":
+            await raw.http_async_client._transport.sessions[
+                next(iter(raw.http_async_client._transport.sessions))
+            ].close()
+        if change == "server_closed":
+            await wire.connections[0].close()
+            await wire.connections[0].wait_closed()
+        await raw.ainvoke(history, **options)
+        assert "previous_response_id" not in wire.requests[1]
+        assert len(wire.requests[1]["input"]) == 3
+
+
+@pytest.mark.asyncio
+async def test_uncached_id_recovers_once_with_full_context(wire):
+    async with model(wire) as raw:
+        history = [HumanMessage(content="First")]
+        first = await raw.ainvoke(history, prompt_cache_key="task:operator")
+        wire.reject_previous = True
+        result = await raw.ainvoke(
+            [*history, first, HumanMessage(content="Next")], prompt_cache_key="task:operator"
+        )
+        ensure_response_complete(result)
+        assert len(wire.requests) == 3
+        assert wire.requests[1]["previous_response_id"] == "resp_1"
+        assert "previous_response_id" not in wire.requests[2]
+        assert len(wire.requests[2]["input"]) == 3
+
+
+@pytest.mark.asyncio
+async def test_missing_affinity_never_shares_connections(wire):
+    async with model(wire) as raw:
+        history = [HumanMessage(content="First")]
+        first = await raw.ainvoke(history)
+        await raw.ainvoke([*history, first, HumanMessage(content="Next")])
+        assert len(wire.connections) == 2
+        assert "previous_response_id" not in wire.requests[1]
+
+
+@pytest.mark.asyncio
+async def test_cancel_closes_socket_and_next_call_starts_full(wire):
+    async with model(wire) as raw:
+        wire.stall = True
+        task = asyncio.create_task(
+            raw.ainvoke([HumanMessage(content="Wait")], prompt_cache_key="task:operator")
+        )
+        await asyncio.wait_for(wire.sent.wait(), 2)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        await asyncio.wait_for(wire.connections[0].wait_closed(), 2)
+        wire.stall = False
+        await raw.ainvoke([HumanMessage(content="Next")], prompt_cache_key="task:operator")
+        assert len(wire.connections) == 2
+        assert "previous_response_id" not in wire.requests[1]
+
+
+@pytest.mark.asyncio
+async def test_disconnect_never_silently_resends_model_request(wire):
+    async with model(wire) as raw:
+        wire.drop = True
+        with pytest.raises(Exception, match="connection|Connection"):
+            await raw.ainvoke([HumanMessage(content="Test")], prompt_cache_key="task:operator")
+        assert len(wire.requests) == 1
+
+
+@pytest.mark.asyncio
+async def test_incomplete_result_does_not_advance_chain(wire):
+    async with model(wire) as raw:
+        wire.status = "incomplete"
+        first = await raw.ainvoke([HumanMessage(content="First")], prompt_cache_key="task:operator")
+        with pytest.raises(ResponseIntegrityError):
+            ensure_response_complete(first)
+        wire.status = "completed"
+        await raw.ainvoke(
+            [HumanMessage(content="First"), first, HumanMessage(content="Next")],
+            prompt_cache_key="task:operator",
+        )
+        assert "previous_response_id" not in wire.requests[1]
+
+
+@pytest.mark.asyncio
+async def test_diagnostics_measure_actual_wire_payload_without_logging_content(wire):
+    class Attempt:
+        data = {}
+
+        def publish_if_due(self):
+            pass
+
+    attempt = Attempt()
+    async with model(wire) as raw:
+        # Disable the unrelated HTTP trace hook: this checks the transport's
+        # own byte counters, alongside the real SDK request serialization.
+        raw.http_async_client.event_hooks = {"request": [], "response": []}
+        first = await raw.ainvoke(
+            [HumanMessage(content="A" * 5000)], prompt_cache_key="task:operator"
+        )
+        token = ACTIVE_ATTEMPT.set(attempt)
+        try:
+            await raw.ainvoke(
+                [HumanMessage(content="A" * 5000), first, HumanMessage(content="Next")],
+                prompt_cache_key="task:operator",
+            )
+        finally:
+            ACTIVE_ATTEMPT.reset(token)
+        stats = deepcopy(attempt.data["transport"])
+        assert stats["incremental"] is True
+        assert stats["sent_payload_bytes"] == len(
+            json.dumps(
+                wire.requests[1], ensure_ascii=False, separators=(",", ":"), sort_keys=True
+            ).encode()
+        )
+        assert stats["payload_bytes_saved"] > 4000
+        assert "AAAA" not in json.dumps(stats)
+
+
+@pytest.mark.asyncio
+async def test_gateway_rejects_upgrade_with_explicit_error(wire):
+    import openai
+    from websockets.http11 import Response
+    from websockets.datastructures import Headers
+
+    async with serve(
+        wire.handle,
+        "127.0.0.1",
+        0,
+        process_request=lambda *_: Response(404, "Not Found", Headers(), b""),
+    ) as server:
+        wire.base = f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}/v1"
+        async with model(wire) as raw:
+            with pytest.raises(openai.NotFoundError, match="WebSocket"):
+                await raw.ainvoke([HumanMessage(content="Test")])
+
+
+@pytest.mark.asyncio
+async def test_wrapper_retains_stream_guards_and_diagnostic_events(wire, monkeypatch):
+    from artemis.services import llm
+    from tests.unit.test_responses_wire_lifecycle import Engine
+
+    engine = Engine()
+    monkeypatch.setattr(llm, "_get_current_data_engine", lambda: engine)
+    async with model(wire) as raw:
+        wrapper = RobustChatModelWrapper(
+            raw,
+            endpoint=ModelEndpoint(model_name="test-model", transport="websocket", max_retries=0),
+        )
+        result = await wrapper.ainvoke(
+            [HumanMessage(content="Test")], prompt_cache_key="task:operator"
+        )
+        ensure_response_complete(result)
+        assert engine.data["transport"]["mode"] == "websocket"
+        assert engine.data["transport"]["received_payload_bytes"] > 0
+        assert engine.data.get("http_status") is None
+
+
+@pytest.mark.asyncio
+async def test_parallel_scopes_keep_independent_response_chains(wire):
+    async with model(wire) as raw:
+
+        async def conversation(scope):
+            history = [HumanMessage(content=f"First {scope}")]
+            first = await raw.ainvoke(history, prompt_cache_key=scope)
+            await raw.ainvoke(
+                [*history, first, HumanMessage(content=f"Next {scope}")], prompt_cache_key=scope
+            )
+            return scope, first.response_metadata["id"]
+
+        results = await asyncio.gather(
+            conversation("task-a:operator"), conversation("task-b:operator")
+        )
+        assert len(wire.connections) == 2
+        for scope, response_id in results:
+            calls = [item for item in wire.requests if item["prompt_cache_key"] == scope]
+            assert len(calls) == 2
+            assert calls[1]["previous_response_id"] == response_id
+            assert len(calls[1]["input"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_idle_connections_are_closed_and_removed(wire, monkeypatch):
+    from artemis.llm import websocket
+
+    monkeypatch.setattr(websocket, "IDLE_SECONDS", 0.01)
+    async with model(wire) as raw:
+        await raw.ainvoke([HumanMessage(content="Test")], prompt_cache_key="task:operator")
+        await asyncio.wait_for(wire.connections[0].wait_closed(), 2)
+        assert not raw.http_async_client._transport.sessions
+
+
+@pytest.mark.asyncio
+async def test_cancel_waiting_turn_keeps_active_chain_and_idle_cleanup(wire, monkeypatch):
+    from artemis.llm import websocket
+
+    # Only shorten the final idle interval; scheduler latency must not break the first chain.
+    monkeypatch.setattr(websocket, "IDLE_SECONDS", 60)
+    wire.gate = asyncio.Event()
+    async with model(wire) as raw:
+        history = [HumanMessage(content="First")]
+        first = asyncio.create_task(raw.ainvoke(history, prompt_cache_key="task:operator"))
+        await asyncio.wait_for(wire.sent.wait(), 2)
+        queued = asyncio.create_task(raw.ainvoke(history, prompt_cache_key="task:operator"))
+        async with asyncio.timeout(2):
+            while next(iter(raw.http_async_client._transport.sessions.values())).users != 2:
+                await asyncio.sleep(0)
+        queued.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await queued
+        wire.gate.set()
+        result = await first
+        monkeypatch.setattr(websocket, "IDLE_SECONDS", 0.01)
+        await raw.ainvoke(
+            [*history, result, HumanMessage(content="Next")], prompt_cache_key="task:operator"
+        )
+        assert len(wire.connections) == 1
+        assert len(wire.requests) == 2
+        assert wire.requests[1]["previous_response_id"] == result.response_metadata["id"]
+        await asyncio.wait_for(wire.connections[0].wait_closed(), 2)
+        assert not raw.http_async_client._transport.sessions
+
+
+@pytest.mark.asyncio
+async def test_pool_eviction_does_not_split_concurrent_scope(wire, monkeypatch):
+    from artemis.llm import websocket
+
+    monkeypatch.setattr(websocket, "MAX_SESSIONS", 1)
+    closing, resume = asyncio.Event(), asyncio.Event()
+    async with model(wire) as raw:
+        await raw.ainvoke([HumanMessage(content="Old")], prompt_cache_key="old")
+        old = next(iter(raw.http_async_client._transport.sessions.values()))
+        close = old.close
+
+        async def delayed_close():
+            closing.set()
+            await resume.wait()
+            await close()
+
+        monkeypatch.setattr(old, "close", delayed_close)
+        first = asyncio.create_task(
+            raw.ainvoke([HumanMessage(content="New first")], prompt_cache_key="new")
+        )
+        try:
+            await asyncio.wait_for(closing.wait(), 2)
+            await asyncio.wait_for(
+                asyncio.gather(
+                    first,
+                    raw.ainvoke([HumanMessage(content="New second")], prompt_cache_key="new"),
+                ),
+                2,
+            )
+            assert len(wire.connections) == 2
+            assert len(wire.requests) == 3
+        finally:
+            resume.set()
+
+
+@pytest.mark.asyncio
+async def test_structured_output_uses_same_websocket_path(wire):
+    from pydantic import BaseModel
+
+    class Result(BaseModel):
+        ok: bool
+
+    wire.outputs.append(
+        [
+            {
+                "id": "fc_test",
+                "type": "function_call",
+                "call_id": "call_test",
+                "name": "Result",
+                "arguments": '{"ok":true}',
+                "status": "completed",
+            }
+        ]
+    )
+    async with model(wire) as raw:
+        result = (
+            await RobustChatModelWrapper(
+                raw, endpoint=ModelEndpoint(transport="websocket", max_retries=0)
+            )
+            .with_structured_output(Result)
+            .ainvoke([HumanMessage(content="Test")])
+        )
+        assert result.ok is True
+        assert len(wire.requests) == 1
+
+
+@pytest.mark.asyncio
+async def test_invalid_tool_arguments_are_rejected_before_use(wire, monkeypatch):
+    from artemis.services import llm
+    from artemis.llm.reliability import LLMExhaustedError
+
+    monkeypatch.setattr(llm, "_get_current_data_engine", lambda: None)
+    wire.outputs.append(
+        [
+            {
+                "id": "fc_test",
+                "type": "function_call",
+                "call_id": "call_test",
+                "name": "lookup",
+                "arguments": '{"x":',
+                "status": "completed",
+            }
+        ]
+    )
+    async with model(wire) as raw:
+        wrapper = RobustChatModelWrapper(
+            raw, endpoint=ModelEndpoint(transport="websocket", max_retries=0)
+        )
+        with pytest.raises(LLMExhaustedError):
+            await wrapper.ainvoke([HumanMessage(content="Test")], prompt_cache_key="task:operator")
+        assert len(wire.requests) == 1
+
+
+@pytest.mark.asyncio
+async def test_redirect_does_not_forward_authorization(wire):
+    from websockets.http11 import Response
+    from websockets.datastructures import Headers
+
+    async with serve(
+        wire.handle,
+        "127.0.0.1",
+        0,
+        process_request=lambda *_: Response(
+            307,
+            "Redirect",
+            Headers({"Location": wire.base.replace("http://", "ws://") + "/responses"}),
+            b"",
+        ),
+    ) as server:
+        redirect_base = f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}/v1"
+        original_base, wire.base = wire.base, redirect_base
+        try:
+            async with model(wire) as raw:
+                with pytest.raises(Exception, match="redirects are disabled"):
+                    await raw.ainvoke([HumanMessage(content="Test")])
+            assert not wire.connections
+        finally:
+            wire.base = original_base

tests/unit/test_responses_wire_lifecycle.py

@@ -124,7 +124,10 @@ async def case(monkeypatch, tmp_path):
 
     def wrapper(timeout=30, max_retries=0):
         endpoint = ModelEndpoint(
-            model_name="test-model", timeout_seconds=timeout, max_retries=max_retries
+            model_name="test-model",
+            transport="http",
+            timeout_seconds=timeout,
+            max_retries=max_retries,
         )
         model = ModelFactory.create_model(endpoint)
         assert isinstance(model, ChatOpenAI)

tests/unit/test_unified_controller_video.py

@@ -25,6 +25,7 @@ import pytest
 import cv2
 
 from artemis.context import ArtemisContext
+from artemis.config.settings import settings
 from artemis.controllers.unified_controller import UnifiedMobileController
 from artemis.drivers.mock.mock_driver import MockDeviceDriver
 from artemis.utils.video import (
@@ -42,6 +43,55 @@ from artemis.utils.video import (
 )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("host,port", [("127.0.0.1", 5037), ("192.0.2.10", 5038)])
+async def test_scrcpy_uses_selected_adb_binary_and_endpoint(monkeypatch, tmp_path, host, port):
+    import os
+
+    adb = str(tmp_path / "platform tools" / "adb.exe")
+    monkeypatch.setattr("artemis.runtime.adb_endpoint.toolchain.resolve", lambda _: adb)
+    monkeypatch.setattr(settings, "ADB_HOST", host)
+    monkeypatch.setattr(settings, "ADB_PORT", port)
+    monkeypatch.setenv("ADB", "other-adb")
+    monkeypatch.setenv("ADB_SERVER_SOCKET", "tcp:unrelated:5999")
+    command = ["scrcpy", "--no-playback"]
+    with patch("asyncio.create_subprocess_exec", AsyncMock()) as spawn:
+        await UnifiedMobileController._spawn_scrcpy(command)
+    environment = spawn.call_args.kwargs["env"]
+    assert environment["ADB"] == adb
+    assert environment["ADB_SERVER_SOCKET"] == f"tcp:{'localhost' if port == 5037 else host}:{port}"
+    assert environment["ADB_HOST"] == host
+    assert environment["ADB_PORT"] == str(port)
+    assert spawn.call_args.args == tuple(command)
+    assert os.environ["ADB"] == "other-adb"
+    assert os.environ["ADB_SERVER_SOCKET"] == "tcp:unrelated:5999"
+
+
+@pytest.mark.asyncio
+async def test_scrcpy_does_not_fall_back_to_an_unselected_adb(monkeypatch):
+    monkeypatch.setattr("artemis.runtime.adb_endpoint.toolchain.resolve", lambda _: None)
+    with patch("asyncio.create_subprocess_exec", AsyncMock()) as spawn:
+        with pytest.raises(FileNotFoundError, match="adb"):
+            await UnifiedMobileController._spawn_scrcpy(["scrcpy"])
+    spawn.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_direct_android_recording_also_uses_selected_adb(monkeypatch, tmp_path):
+    from artemis.drivers.android.adb_driver import AndroidAdbDriver
+
+    adb = str(tmp_path / "platform-tools" / "adb.exe")
+    monkeypatch.setattr("artemis.runtime.adb_endpoint.toolchain.resolve", lambda _: adb)
+    monkeypatch.setattr(settings, "ADB_HOST", "192.0.2.10")
+    monkeypatch.setattr(settings, "ADB_PORT", 5038)
+    driver = AndroidAdbDriver("fixture-device", MagicMock())
+    with patch("asyncio.create_subprocess_exec", AsyncMock()) as spawn:
+        await driver.start_video_recording(tmp_path)
+    assert spawn.await_count == 1
+    assert spawn.call_args.kwargs["env"]["ADB"] == adb
+    assert spawn.call_args.kwargs["env"]["ADB_SERVER_SOCKET"] == "tcp:192.0.2.10:5038"
+
+
 @pytest.mark.asyncio
 async def test_audio_extraction_rejects_silent_video_without_running_ffmpeg(tmp_path):
     silent_video = tmp_path / "silent.mp4"

tests/unit/tools/test_launch_app.py

@@ -13,9 +13,9 @@
 # limitations under the License.
 
 from unittest.mock import MagicMock, patch
-from artemis.agents.hopper.hopper import HopperOutput
+from artemis.agents.hopper.hopper import HopperError, HopperOutput
 from artemis.context import ArtemisContext
-from artemis.tools.mobile.launch_app import find_package
+from artemis.tools.mobile.launch_app import find_package, launch_app
 import pytest
 
 
@@ -55,7 +55,7 @@ async def test_find_package_cache_hit(mock_hopper, mock_list_packages, mock_cont
 @pytest.mark.asyncio
 @patch("artemis.tools.mobile.launch_app.list_packages_async")
 @patch("artemis.tools.mobile.launch_app.hopper")
-async def test_find_package_not_found_caches_none(mock_hopper, mock_list_packages, mock_context):
+async def test_find_package_not_found_is_not_cached(mock_hopper, mock_list_packages, mock_context):
     mock_list_packages.return_value = "com.example.app1\ncom.example.app2"
 
     # Hopper doesn't find it
@@ -64,13 +64,14 @@ async def test_find_package_not_found_caches_none(mock_hopper, mock_list_package
     pkg1 = await find_package(mock_context, "App 3")
     assert pkg1 is None
     assert mock_hopper.call_count == 1
-    assert mock_context.package_cache["App 3"] is None
+    assert "App 3" not in mock_context.package_cache
 
-    # Second call: cache hit (None), does not call hopper again
+    # An earlier miss must not prevent a later successful lookup.
     mock_hopper.reset_mock()
+    mock_hopper.return_value = HopperOutput(found=True, output="com.example.app1", reason="Found")
     pkg2 = await find_package(mock_context, "App 3")
-    assert pkg2 is None
-    mock_hopper.assert_not_called()
+    assert pkg2 == "com.example.app1"
+    mock_hopper.assert_called_once()
 
 
 @pytest.mark.asyncio
@@ -90,3 +91,32 @@ async def test_find_package_passes_use_fallback(mock_hopper, mock_list_packages,
         data="com.example.app1\ncom.example.app2",
         use_fallback=False,
     )
+
+
+@pytest.mark.asyncio
+@patch("artemis.tools.mobile.launch_app.list_packages_async")
+@patch("artemis.tools.mobile.launch_app.hopper")
+async def test_model_failure_is_reported_without_poisoning_package_cache(
+    mock_hopper, mock_list_packages, mock_context
+):
+    mock_list_packages.return_value = "com.example.app1"
+    mock_hopper.side_effect = HopperError("模型未能完成数据提取，请查看模型请求诊断。")
+    result = await launch_app.execute(ctx=mock_context, app_name="App 1")
+    assert "模型未能完成数据提取" in result
+    assert "Package not found" not in result
+    assert "App 1" not in mock_context.package_cache
+    mock_hopper.side_effect = None
+    mock_hopper.return_value = HopperOutput(found=True, output="com.example.app1", reason="Found")
+    assert await find_package(mock_context, "App 1") == "com.example.app1"
+
+
+@pytest.mark.asyncio
+@patch("artemis.tools.mobile.launch_app.list_packages_async")
+@patch("artemis.tools.mobile.launch_app.hopper")
+async def test_invented_package_is_never_cached(mock_hopper, mock_list_packages, mock_context):
+    mock_list_packages.return_value = "com.example.app1"
+    mock_hopper.return_value = HopperOutput(
+        found=True, output="com.example.missing", reason="Guess"
+    )
+    assert await find_package(mock_context, "App 1") is None
+    assert "App 1" not in mock_context.package_cache

uv.lock

@@ -248,7 +248,7 @@ wheels = [
 
 [[package]]
 name = "artemis"
-version = "2.6.0"
+version = "2.7.1"
 source = { editable = "." }
 dependencies = [
     { name = "adbutils" },
@@ -369,7 +369,7 @@ requires-dist = [
     { name = "uiautomator2", specifier = ">=3.5.0" },
     { name = "uuid-utils", specifier = ">=0.9.0" },
     { name = "uvicorn", extras = ["standard"], specifier = ">=0.30.1" },
-    { name = "websockets", specifier = ">=12.0" },
+    { name = "websockets", specifier = ">=15.0" },
     { name = "windows-capture", marker = "platform_machine == 'AMD64' and sys_platform == 'win32'", specifier = ">=2.0.1,<3" },
 ]
 provides-extras = ["dev"]