diff --git a/backend/package/yuxi/agents/backends/sandbox/backend.py b/backend/package/yuxi/agents/backends/sandbox/backend.py index e3fdd2ba1..7c2263686 100644 --- a/backend/package/yuxi/agents/backends/sandbox/backend.py +++ b/backend/package/yuxi/agents/backends/sandbox/backend.py @@ -545,31 +545,99 @@ def execute(self, command: str, *, timeout: int | None = None) -> ExecuteRespons payload size before being returned. """ try: - kwargs: dict[str, Any] = {"command": command} - if timeout is not None: - kwargs["timeout"] = timeout - kwargs["hard_timeout"] = timeout - kwargs["request_options"] = {"timeout_in_seconds": timeout} + effective_timeout = timeout if timeout is not None else self._command_timeout_seconds + kwargs: dict[str, Any] = { + "command": command, + "timeout": effective_timeout, + "hard_timeout": effective_timeout, + "request_options": {"timeout_in_seconds": effective_timeout}, + } result = self._get_client().shell.exec_command(**kwargs) + return self._execute_response(result.data) + except Exception as exc: # noqa: BLE001 + logger.error(f"Sandbox execute failed for thread {self._thread_id}: {exc}") + return ExecuteResponse(output=f"Error: {exc}", exit_code=1, truncated=False) - output = result.data.output or "" - exit_code = result.data.exit_code - - truncated = False - encoded = output.encode("utf-8", errors="ignore") - if len(encoded) > self._max_output_bytes: - output = encoded[: self._max_output_bytes].decode("utf-8", errors="ignore") - truncated = True - - return ExecuteResponse( - output=output, - exit_code=exit_code if isinstance(exit_code, int) else None, - truncated=truncated, - ) + async def aexecute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse: + """以独立 GEM session 执行命令，取消时显式终止远端进程。""" + effective_timeout = timeout if timeout is not None else self._command_timeout_seconds + try: + connection = await asyncio.to_thread(self._get_connection) + async with httpx.AsyncClient(timeout=10, trust_env=False) as http_client: + client = self._build_async_client(connection.sandbox_url, http_client) + session_id = f"yuxi-{uuid.uuid4().hex}" + await client.shell.create_session(id=session_id, request_options={"timeout_in_seconds": 10}) + launch = asyncio.create_task( + client.shell.exec_command( + id=session_id, + command=command, + async_mode=True, + hard_timeout=effective_timeout, + request_options={"timeout_in_seconds": 10}, + ) + ) + try: + async with asyncio.timeout(effective_timeout + 10): + # 启动确认先于终止，避免 kill 到达后命令才在远端开始。 + result = await asyncio.shield(launch) + status = result.data.status + while status in {"running", "no_change_timeout"}: + waited = await client.shell.wait_for_process( + id=session_id, + seconds=1, + max_wait_seconds=1, + request_options={"timeout_in_seconds": 5}, + ) + status = waited.data.status + final = await client.shell.view(id=session_id, request_options={"timeout_in_seconds": 5}) + return self._execute_response(final.data) + except BaseException: + await self._terminate_command(client, session_id, launch) + raise except Exception as exc: # noqa: BLE001 logger.error(f"Sandbox execute failed for thread {self._thread_id}: {exc}") return ExecuteResponse(output=f"Error: {exc}", exit_code=1, truncated=False) + async def _terminate_command(self, client: Any, session_id: str, launch: asyncio.Task) -> None: + """收敛启动与 kill，重复取消不能打断这个有界的远端收尾。""" + + async def stop(): + """启动失败仍尝试 kill，处理响应丢失但远端已执行的情形。""" + with suppress(Exception): + async with asyncio.timeout(10): + await launch + async with asyncio.timeout(5): + await client.shell.kill_process(id=session_id, request_options={"timeout_in_seconds": 5}) + + cleanup = asyncio.create_task(stop()) + while not cleanup.done(): + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError: + continue + except Exception: + break + try: + cleanup.result() + except Exception: + logger.exception(f"Sandbox command termination failed: thread={self._thread_id}, session={session_id}") + + def _execute_response(self, data: Any) -> ExecuteResponse: + """归一化最终命令输出并限制返回字节数。""" + output = data.output or "" + encoded = output.encode("utf-8", errors="ignore") + truncated = len(encoded) > self._max_output_bytes + if truncated: + output = encoded[: self._max_output_bytes].decode("utf-8", errors="ignore") + exit_code = data.exit_code + if exit_code is None and getattr(data, "status", None) in {"hard_timeout", "terminated"}: + exit_code = 124 if data.status == "hard_timeout" else 130 + return ExecuteResponse( + output=output, + exit_code=exit_code if isinstance(exit_code, int) else None, + truncated=truncated, + ) + def ls(self, path: str) -> LsResult: """List direct children of an allowed sandbox path with lightweight metadata.""" try: diff --git a/backend/package/yuxi/services/agent_run_service.py b/backend/package/yuxi/services/agent_run_service.py index 7e481e456..7d5f13bfc 100644 --- a/backend/package/yuxi/services/agent_run_service.py +++ b/backend/package/yuxi/services/agent_run_service.py @@ -1051,6 +1051,6 @@ async def get_active_run_by_thread(*, thread_id: str, current_uid: str, db: Asyn .limit(1) ) run = result.scalar_one_or_none() - if run and run.status in ("pending", "running", "cancel_requested", "interrupted"): + if run and (run.runtime_cleanup_pending or run.status in ("pending", "running", "cancel_requested", "interrupted")): return {"run": run.to_dict()} return {"run": None} diff --git a/backend/package/yuxi/services/run_worker.py b/backend/package/yuxi/services/run_worker.py index ef779a0a2..d8ed478d8 100644 --- a/backend/package/yuxi/services/run_worker.py +++ b/backend/package/yuxi/services/run_worker.py @@ -494,7 +494,7 @@ async def reconcile_pending_runtime_cleanups() -> list[str]: if not await _release_runtime_if_idle(run): continue except Exception: - logger.error("Failed to reconcile execution-tree runtime cleanup: run=%s", run.id, exc_info=True) + logger.exception(f"Failed to reconcile execution-tree runtime cleanup: run={run.id}") continue if run.status in TERMINAL_RUN_STATUSES: await _append_end_event(run.id, run.status, thread_id=run.conversation_thread_id) diff --git a/backend/test/http_helpers.py b/backend/test/http_helpers.py new file mode 100644 index 000000000..9aaedfe58 --- /dev/null +++ b/backend/test/http_helpers.py @@ -0,0 +1,29 @@ +"""真实 TCP HTTP 测试的服务生命周期。""" + +import asyncio +import socket +from contextlib import asynccontextmanager + +import uvicorn + + +@asynccontextmanager +async def live_http_server(app): + """启动临时本地服务，结束后关闭监听与连接。""" + listener = socket.socket() + listener.bind(("127.0.0.1", 0)) + port = listener.getsockname()[1] + server = uvicorn.Server(uvicorn.Config(app, lifespan="off", log_level="error")) + serving = asyncio.create_task(server.serve(sockets=[listener])) + try: + async with asyncio.timeout(5): + while not server.started: + if serving.done(): + await serving + raise RuntimeError("Test HTTP server did not start") + await asyncio.sleep(0.01) + yield f"http://127.0.0.1:{port}" + finally: + server.should_exit = True + await asyncio.wait_for(serving, timeout=5) + listener.close() diff --git a/backend/test/integration/services/test_sandbox_cancel_lifecycle.py b/backend/test/integration/services/test_sandbox_cancel_lifecycle.py new file mode 100644 index 000000000..216bb0c12 --- /dev/null +++ b/backend/test/integration/services/test_sandbox_cancel_lifecycle.py @@ -0,0 +1,301 @@ +"""真实 HTTP、PostgreSQL、Redis、worker 执行链与 GEM 的取消证据。""" + +import asyncio +import importlib.util +import json +import os +import shutil +import sys +from contextlib import asynccontextmanager +from pathlib import Path +from unittest.mock import AsyncMock +from uuid import uuid4 + +import httpx +import anyio +import pytest +import pytest_asyncio +from fastapi import FastAPI +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from server.routers.agent_router import agent_router +from server.utils.auth_middleware import get_db, get_required_user +from test.http_helpers import live_http_server +from yuxi.agents.backends.sandbox import ProvisionerSandboxBackend, get_sandbox_provider +from yuxi.agents.backends.sandbox.provider import sandbox_provisioner_token +from yuxi.agents.context import BaseContext +from yuxi.services.skills.projection import get_user_skills_root_dir +from yuxi.config import get_skill_projection_dir +from yuxi.repositories.agent_run_repository import AgentRunRepository +from yuxi.services import run_worker +from yuxi.services.agent_run_manifest_service import PreparedRunExecution +from yuxi.storage.postgres.models_business import AgentRun, Base, Conversation, Message, Project, User +from yuxi.storage.redis import close_async_redis_client, get_async_redis_client +from yuxi.workspace.paths import ensure_bound_user_workdir, global_user_data_dir, workspace_uid_dirname + +from agent_run_test_helpers import create_agent_run + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +@pytest_asyncio.fixture +async def cancellation_db(): + """隔离业务表，避免测试 Run 被部署中的共享 worker 恢复扫描捕获。""" + url = os.environ["POSTGRES_URL"] + schema = f"test_sandbox_cancel_{uuid4().hex}" + admin = create_async_engine(url) + engine = create_async_engine(url, connect_args={"server_settings": {"search_path": schema}}) + try: + async with admin.begin() as connection: + await connection.execute(text(f'CREATE SCHEMA "{schema}"')) + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + async with async_sessionmaker(engine, expire_on_commit=False)() as db: + yield db + finally: + await engine.dispose() + async with admin.begin() as connection: + await connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await admin.dispose() + + +async def test_http_stop_kills_gem_before_worker_releases_fifo(cancellation_db, monkeypatch): + """模型输出替换为固定脚本，取消/lease/事件/清理和文件均使用真实边界。""" + db = cancellation_db + factory = async_sessionmaker(db.bind, expire_on_commit=False) + run_id, thread_id, message_id = await create_agent_run( + factory, prefix="sandbox-cancel", message_content="cancel command", input_payload={} + ) + async with factory() as fresh: + run = await fresh.get(AgentRun, run_id) + user = await fresh.scalar(select(User).where(User.uid == run.uid)) + conversation = await fresh.get(Conversation, run.conversation_id) + project = await fresh.get(Project, conversation.project_id) + message = await fresh.get(Message, message_id) + message.extra_metadata = {} + await fresh.commit() + ensure_bound_user_workdir(user.uid, project.workdir_path) + get_user_skills_root_dir(user.uid) + backend = ProvisionerSandboxBackend(thread_id=thread_id, uid=user.uid, workdir_path=project.workdir_path) + cleanup_entered, allow_cleanup = asyncio.Event(), asyncio.Event() + execution = None + + @asynccontextmanager + async def worker_session(): + """worker 与 HTTP 都只访问测试 Schema。""" + async with factory() as fresh: + try: + yield fresh + await fresh.commit() + except BaseException: + await fresh.rollback() + raise + + async def http_db(): + """保留真实取消路由的事务提交顺序。""" + async with worker_session() as fresh: + yield fresh + + async def prepare(*, run, user, worker_id, workdir_binding): + """固定模型准备结果，不调用收费模型；后续 worker 流程保持真实。""" + return PreparedRunExecution( + manifest={}, + backend_id="ChatbotAgent", + context=BaseContext( + uid=user.uid, + thread_id=thread_id, + run_id=run_id, + request_id=run.request_id, + worker_id=worker_id, + runtime_scope_id=thread_id, + workdir_relative_path=workdir_binding.workdir_path, + workdir_path=f"/home/gem/user-data/{workdir_binding.workdir_path}", + ), + ) + + async def command_stream(**_kwargs): + """启动真实进程，延迟副作用必须被停止操作阻止。""" + result = await backend.aexecute( + 'python3 -c "import os,time; from pathlib import Path; ' + "Path('/tmp/cancel-pid').write_text(str(os.getpid())); time.sleep(8); " + "Path('/tmp/cancel-late').write_text('unexpected')\"", + timeout=3600, + ) + assert result.exit_code == 0, result.output + yield json.dumps({"status": "finished", "thread_id": thread_id}).encode() + + release_runtime = run_worker._release_runtime_if_idle + + async def pause_before_release(run): + """在真实清理前设置屏障，回读进程和 durable fence。""" + cleanup_entered.set() + await allow_cleanup.wait() + return await release_runtime(run) + + monkeypatch.setattr(run_worker.pg_manager, "get_async_session_context", worker_session) + monkeypatch.setattr(run_worker, "prepare_and_record_run_execution", prepare) + monkeypatch.setattr(run_worker, "stream_agent_chat", command_stream) + monkeypatch.setattr(run_worker, "_read_run_token_usage_from_state", AsyncMock(return_value=None)) + monkeypatch.setattr(run_worker, "_release_runtime_if_idle", pause_before_release) + app = FastAPI() + app.include_router(agent_router, prefix="/api") + app.dependency_overrides[get_db] = http_db + app.dependency_overrides[get_required_user] = lambda: user + provider = get_sandbox_provider() + redis = await get_async_redis_client() + try: + # 预创建仅属于测试的 runtime，把镜像启动时间排除在停止时延之外。 + assert (await backend.aexecute("printf ready")).output.strip() == "ready" + execution = asyncio.create_task(run_worker.process_agent_run({}, run_id)) + async with asyncio.timeout(15): + while True: + if execution.done(): + await execution + async with factory() as fresh: + failed = await fresh.get(AgentRun, run_id) + pytest.fail(f"worker 未启动目标命令: {failed.error_type}: {failed.error_message}") + result = await asyncio.to_thread(backend.execute, "test -f /tmp/cancel-pid") + if result.exit_code == 0: + break + await asyncio.sleep(0.1) + async with live_http_server(app) as url, httpx.AsyncClient(base_url=url, timeout=15) as http: + response = await http.post(f"/api/agent/runs/{run_id}/cancel") + assert response.status_code == 200, response.text + await asyncio.wait_for(cleanup_entered.wait(), 15) + result = await asyncio.to_thread( + backend.execute, "! kill -0 $(cat /tmp/cancel-pid) 2>/dev/null && test ! -e /tmp/cancel-late" + ) + assert result.exit_code == 0, "取消已记录，但实际 GEM 进程仍存活" + state = (await http.get(f"/api/agent/runs/{run_id}")).json()["run"] + assert state["status"] == "cancelled" and state["runtime_cleanup_pending"] is True + active_state = (await http.get(f"/api/agent/thread/{thread_id}/active_run")).json()["run"] + assert active_state is not None, "刷新或新标签页必须看见尚未清理的 Run" + assert active_state["id"] == run_id and active_state["runtime_cleanup_pending"] is True + async with factory() as fresh: + active = await AgentRunRepository(fresh).get_active_run_by_thread_for_user( + uid=user.uid, agent_slug="main", conversation_thread_id=thread_id + ) + assert active.id == run_id, "清理前不得放行同一线程" + events = await redis.xrange(f"run:events:{run_id}") + assert not any( + fields.get("event_type") == "end" or fields.get(b"event_type") == b"end" for _, fields in events + ) + allow_cleanup.set() + await asyncio.wait_for(execution, 15) + assert (await http.get(f"/api/agent/thread/{thread_id}/active_run")).json()["run"] is None + async with factory() as fresh: + persisted = await fresh.get(AgentRun, run_id) + assert persisted.status == "cancelled" and persisted.runtime_cleanup_pending is False + assert persisted.worker_id is None and persisted.lease_expires_at is None + assert ( + await AgentRunRepository(fresh).get_active_run_by_thread_for_user( + uid=user.uid, agent_slug="main", conversation_thread_id=thread_id + ) + is None + ) + assert await asyncio.to_thread(provider._client.discover, backend.id) is None + events = await redis.xrange(f"run:events:{run_id}") + assert any(fields.get("event_type") == "end" or fields.get(b"event_type") == b"end" for _, fields in events) + assert (await backend.aexecute("printf next-message")).output.strip() == "next-message" + finally: + allow_cleanup.set() + if execution is not None and not execution.done(): + execution.cancel() + await asyncio.gather(execution, return_exceptions=True) + await asyncio.to_thread(provider.release, thread_id, uid=user.uid, workdir_path=project.workdir_path) + await redis.delete(f"run:events:{run_id}", f"run:cancel:{run_id}") + await close_async_redis_client() + shutil.rmtree(global_user_data_dir(user.uid), ignore_errors=True) + shutil.rmtree(get_skill_projection_dir() / workspace_uid_dirname(user.uid), ignore_errors=True) + + +async def test_busy_proxy_delete_is_bounded_and_other_runtime_stays_available(): + """真实长 HTTP 代理占用 pin 时删除返回 503，其他实例和健康检查仍可用。""" + uid = f"pytest-busy-{uuid4().hex}" + first = ProvisionerSandboxBackend(thread_id="busy", uid=uid, inherit_env=False) + other = ProvisionerSandboxBackend(thread_id="other", uid=uid, inherit_env=False) + provider = get_sandbox_provider() + command = None + client = None + try: + assert (await first.aexecute("printf ready")).exit_code == 0 + assert (await other.aexecute("printf ready")).exit_code == 0 + client = first._get_client() + await asyncio.to_thread(client.shell.create_session, id="busy-proxy") + command = asyncio.create_task( + asyncio.to_thread( + client.shell.exec_command, + id="busy-proxy", + command="touch /tmp/busy-started; sleep 12", + timeout=20, + hard_timeout=15, + ) + ) + async with asyncio.timeout(10): + while True: + if (await asyncio.to_thread(first.execute, "test -f /tmp/busy-started")).exit_code == 0: + break + await asyncio.sleep(0.1) + url = os.environ["SANDBOX_PROVISIONER_URL"].rstrip("/") + headers = {"Authorization": f"Bearer {sandbox_provisioner_token()}"} + async with httpx.AsyncClient(base_url=url, headers=headers, timeout=10) as http: + started_at = asyncio.get_running_loop().time() + deletion = asyncio.create_task(http.delete(f"/api/sandboxes/{first.id}")) + try: + for _ in range(5): + health = await http.get("/health") + assert health.status_code == 200 + assert (await other.aexecute("printf independent")).output.strip() == "independent" + response = await deletion + assert response.status_code == 503, response.text + assert response.json()["detail"] == "sandbox busy" + assert asyncio.get_running_loop().time() - started_at < 10 + assert (await http.get(f"/api/sandboxes/{first.id}")).status_code == 200 + finally: + await asyncio.gather(deletion, return_exceptions=True) + finally: + try: + if command is not None: + await asyncio.wait_for(command, 25) + finally: + for backend in (first, other): + await asyncio.to_thread(provider.release, backend._thread_id, uid=uid) + + +async def test_health_responds_when_cleanup_occupies_all_sync_threads(monkeypatch): + """同步线程池被清理占满时，真实 HTTP 健康检查仍须及时响应。""" + monkeypatch.setenv("PROVISIONER_BACKEND", "memory") + monkeypatch.setenv("SANDBOX_PROVISIONER_TOKEN", "test-busy-health-token-at-least-32-characters") + spec = importlib.util.spec_from_file_location( + "busy_health_provisioner", Path(__file__).resolve().parents[4] / "docker/sandbox_provisioner/app.py" + ) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + pins = module.sandbox_operation_pins + pins.acquire("busy") + limiter = anyio.to_thread.current_default_thread_limiter() + original_tokens = limiter.total_tokens + limiter.total_tokens = 1 + deletion = None + try: + async with live_http_server(module.app) as url, httpx.AsyncClient(base_url=url, timeout=10) as http: + deletion = asyncio.create_task( + http.delete( + "/api/sandboxes/busy", + headers={"Authorization": "Bearer test-busy-health-token-at-least-32-characters"}, + ) + ) + async with asyncio.timeout(2): + while "busy" not in pins._deleting: + await asyncio.sleep(0.01) + async with asyncio.timeout(1): + assert (await http.get("/health")).status_code == 200 + assert (await deletion).status_code == 503 + finally: + pins.release("busy") + limiter.total_tokens = original_tokens + if deletion is not None: + await asyncio.gather(deletion, return_exceptions=True) + sys.modules.pop(spec.name, None) diff --git a/backend/test/unit/backends/test_sandbox_command_cancellation.py b/backend/test/unit/backends/test_sandbox_command_cancellation.py new file mode 100644 index 000000000..74ae4e0e7 --- /dev/null +++ b/backend/test/unit/backends/test_sandbox_command_cancellation.py @@ -0,0 +1,100 @@ +"""命令取消必须到达 GEM session，不能只结束 Python 等待。""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from yuxi.agents.backends.sandbox.backend import ProvisionerSandboxBackend + + +@pytest.mark.asyncio +@pytest.mark.parametrize("cancel_during_start", [False, True]) +async def test_cancel_terminates_the_started_session_before_return(monkeypatch, cancel_during_start): + """启动确认前后的取消均收敛同一个命令，取消后不得产生延迟副作用。""" + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + backend = ProvisionerSandboxBackend(thread_id="cancel-thread", uid="cancel-user") + monkeypatch.setattr(backend, "_get_connection", lambda: SimpleNamespace(sandbox_url="http://sandbox")) + entered, started, allow_start, stopped, killing, allow_kill = (asyncio.Event() for _ in range(6)) + active = set() + + async def launch(**kwargs): + """模拟启动响应在请求到达远端后延迟返回。""" + entered.set() + if cancel_during_start: + await allow_start.wait() + active.add(kwargs["id"]) + started.set() + return SimpleNamespace(data=SimpleNamespace(status="running")) + + async def kill(*, id, **_kwargs): + """只有显式终止能清除模拟远端进程。""" + assert started.is_set() + killing.set() + await allow_kill.wait() + active.remove(id) + stopped.set() + + shell = SimpleNamespace( + create_session=AsyncMock(), + exec_command=launch, + kill_process=kill, + ) + + async def wait(**_kwargs): + """模拟长命令尚未结束。""" + await asyncio.Event().wait() + + shell.wait_for_process = wait + monkeypatch.setattr(backend, "_build_async_client", lambda *_args: SimpleNamespace(shell=shell)) + monkeypatch.setattr(backend, "execute", lambda *_args, **_kwargs: pytest.fail("异步命令仍进入同步线程")) + task = asyncio.create_task(backend.aexecute("long command", timeout=3600)) + try: + await asyncio.wait_for(entered.wait(), 1) + task.cancel() + allow_start.set() + await asyncio.wait_for(killing.wait(), 1) + task.cancel() + allow_kill.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 1) + assert stopped.is_set() + assert active == set() + finally: + allow_start.set() + allow_kill.set() + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_async_execute_returns_completed_output_and_hard_timeout(monkeypatch): + """短轮询只在命令终止后交付最终输出。""" + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + backend = ProvisionerSandboxBackend(thread_id="command-thread", uid="command-user") + monkeypatch.setattr(backend, "_get_connection", lambda: SimpleNamespace(sandbox_url="http://sandbox")) + shell = SimpleNamespace( + create_session=AsyncMock(), + exec_command=AsyncMock(return_value=SimpleNamespace(data=SimpleNamespace(status="running"))), + wait_for_process=AsyncMock(return_value=SimpleNamespace(data=SimpleNamespace(status="completed"))), + view=AsyncMock(return_value=SimpleNamespace(data=SimpleNamespace(output="done", exit_code=0))), + kill_process=AsyncMock(), + ) + monkeypatch.setattr(backend, "_build_async_client", lambda *_args: SimpleNamespace(shell=shell)) + result = await backend.aexecute("echo done", timeout=3600) + assert result.output == "done" + assert result.exit_code == 0 + assert shell.exec_command.call_args.kwargs["hard_timeout"] == 3600 + assert shell.exec_command.call_args.kwargs["async_mode"] is True + assert shell.wait_for_process.call_args.kwargs["max_wait_seconds"] <= 2 + shell.kill_process.assert_not_awaited() + + +def test_hard_timeout_without_exit_code_is_a_failed_command(monkeypatch): + """GEM 超时无退出码时不得呈现成未知或成功。""" + monkeypatch.setattr("yuxi.agents.backends.sandbox.backend.get_sandbox_provider", lambda: object()) + backend = ProvisionerSandboxBackend(thread_id="timeout-thread", uid="timeout-user") + response = backend._execute_response(SimpleNamespace(status="hard_timeout", output="partial", exit_code=None)) + assert response.exit_code == 124 + assert response.output == "partial" diff --git a/backend/test/unit/backends/test_sandbox_provisioner_config.py b/backend/test/unit/backends/test_sandbox_provisioner_config.py index 89e314cde..a2c94bd9d 100644 --- a/backend/test/unit/backends/test_sandbox_provisioner_config.py +++ b/backend/test/unit/backends/test_sandbox_provisioner_config.py @@ -1,6 +1,7 @@ from __future__ import annotations import importlib.util +import asyncio import os import sys import threading @@ -586,6 +587,95 @@ def next_request(): next_thread.join(timeout=1) +def test_operation_pin_timeout_keeps_active_request_and_reopens_discovery(): + """清理超时不能删活跃实例，也不能留下永久 deleting 标记。""" + module = _load_module() + pins = module.SandboxOperationPins(wait_timeout_seconds=0.02) + pins.acquire("busy") + with pytest.raises(module.SandboxBusyError): + pins.begin_delete("busy") + assert pins._active == {"busy": 1} + pins.acquire("busy") + pins.release("busy") + pins.release("busy") + pins.begin_delete("busy") + with pytest.raises(module.SandboxBusyError): + pins.acquire("busy") + pins.end_delete("busy") + pins.acquire("busy") + pins.release("busy") + + +def test_busy_delete_returns_retryable_http_error_without_deleting(monkeypatch): + """活跃命令阻止删除时 HTTP 有界失败，其他实例仍可发现。""" + module = _load_module() + monkeypatch.setenv("SANDBOX_PROVISIONER_TOKEN", "test-provisioner-token-32-characters-long") + pins = module.SandboxOperationPins(wait_timeout_seconds=0.02) + monkeypatch.setattr(module, "sandbox_operation_pins", pins) + deleted = [] + monkeypatch.setattr(module.backend_impl, "delete", lambda *args, **kwargs: deleted.append(args)) + pins.acquire("busy") + try: + with TestClient(module.app) as client: + response = client.delete( + "/api/sandboxes/busy", + headers={"Authorization": "Bearer test-provisioner-token-32-characters-long"}, + ) + assert response.status_code == 503 + assert response.json()["detail"] == "sandbox busy" + assert client.get("/health").status_code == 200 + assert deleted == [] + finally: + pins.release("busy") + + +@pytest.mark.asyncio +async def test_cancelled_proxy_waiter_does_not_leak_an_operation_pin(monkeypatch): + """等待删除的代理已断开时，线程晚取得的 pin 必须归还。""" + module = _load_module() + pins = module.SandboxOperationPins() + monkeypatch.setattr(module, "sandbox_operation_pins", pins) + entered = threading.Event() + acquired = threading.Event() + acquire = pins.acquire + + def blocked_acquire(sandbox_id): + """确认代理线程已进入等待，再取消其异步调用。""" + entered.set() + acquire(sandbox_id) + acquired.set() + + monkeypatch.setattr(pins, "acquire", blocked_acquire) + pins.begin_delete("busy") + proxy = asyncio.create_task(module.proxy_sandbox_request("busy", None)) + try: + assert await asyncio.to_thread(entered.wait, 1) + proxy.cancel() + with pytest.raises(asyncio.CancelledError): + await proxy + finally: + pins.end_delete("busy") + assert await asyncio.to_thread(acquired.wait, 1) + async with asyncio.timeout(1): + while pins._active: + await asyncio.sleep(0.01) + assert pins._active == {} + + +def test_second_delete_cannot_release_first_delete_ownership(): + """并发删除超时不允许第二个调用释放第一个调用的 deleting 栅栏。""" + module = _load_module() + pins = module.SandboxOperationPins(wait_timeout_seconds=0.02) + pins.begin_delete("busy") + with pytest.raises(module.SandboxBusyError): + pins.begin_delete("busy") + with pytest.raises(module.SandboxBusyError): + pins.acquire("busy") + pins.end_delete("busy") + pins.acquire("busy") + pins.release("busy") + + def test_docker_mount_checks_reject_uploads_and_outputs_mounts(monkeypatch, tmp_path): monkeypatch.setenv("PROVISIONER_BACKEND", "memory") module = _load_module() diff --git a/docker/sandbox_provisioner/app.py b/docker/sandbox_provisioner/app.py index a265d4d9e..5fc5fe205 100644 --- a/docker/sandbox_provisioner/app.py +++ b/docker/sandbox_provisioner/app.py @@ -21,7 +21,7 @@ import httpx from dotenv import dotenv_values from fastapi import Depends, FastAPI, Header, HTTPException, Request -from fastapi.responses import StreamingResponse +from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field logger = logging.getLogger(__name__) @@ -432,18 +432,26 @@ class SandboxQuiesceTimeoutError(RuntimeError): """Sandbox 全局静默未在调用方 deadline 内完成。""" +class SandboxBusyError(RuntimeError): + """操作无法在有界窗口内排空，调用方应保留清理意图并重试。""" + + class SandboxOperationPins: """让删除等待已经开始的 proxy 请求排空，并阻止新的请求穿过删除。""" - def __init__(self): + def __init__(self, *, wait_timeout_seconds: float = 5): self._condition = threading.Condition() self._active: dict[str, int] = {} self._deleting: set[str] = set() + self._wait_timeout_seconds = wait_timeout_seconds def acquire(self, sandbox_id: str) -> None: with self._condition: - while sandbox_id in self._deleting: - self._condition.wait() + if not self._condition.wait_for( + lambda: sandbox_id not in self._deleting, + timeout=self._wait_timeout_seconds, + ): + raise SandboxBusyError("sandbox busy") self._active[sandbox_id] = self._active.get(sandbox_id, 0) + 1 def release(self, sandbox_id: str) -> None: @@ -457,9 +465,20 @@ def release(self, sandbox_id: str) -> None: def begin_delete(self, sandbox_id: str) -> None: with self._condition: + deadline = time.monotonic() + self._wait_timeout_seconds + if not self._condition.wait_for( + lambda: sandbox_id not in self._deleting, + timeout=max(0, deadline - time.monotonic()), + ): + raise SandboxBusyError("sandbox busy") self._deleting.add(sandbox_id) - while self._active.get(sandbox_id, 0): - self._condition.wait() + if not self._condition.wait_for( + lambda: not self._active.get(sandbox_id, 0), + timeout=max(0, deadline - time.monotonic()), + ): + self._deleting.remove(sandbox_id) + self._condition.notify_all() + raise SandboxBusyError("sandbox busy") def end_delete(self, sandbox_id: str) -> None: with self._condition: @@ -1883,8 +1902,10 @@ def _collect_expired_sandboxes(self) -> list[tuple[str, str | None]]: ] def _delete_expired_sandbox(self, sandbox_id: str, generation: str | None) -> None: - self._operation_pins.begin_delete(sandbox_id) + acquired = False try: + self._operation_pins.begin_delete(sandbox_id) + acquired = True cutoff = time.time() - self._idle_timeout_seconds with self._lock: current = self._last_activity_at.get(sandbox_id) @@ -1896,7 +1917,8 @@ def _delete_expired_sandbox(self, sandbox_id: str, generation: str | None) -> No except Exception as exc: # noqa: BLE001 logger.warning(f"Failed to delete idle sandbox {sandbox_id}: {exc}") finally: - self._operation_pins.end_delete(sandbox_id) + if acquired: + self._operation_pins.end_delete(sandbox_id) def _run(self) -> None: while not self._stop_event.wait(self._check_interval_seconds): @@ -1960,6 +1982,16 @@ async def lifespan(app: FastAPI): app = FastAPI(title="Yuxi Sandbox Provisioner", lifespan=lifespan) +@app.exception_handler(SandboxBusyError) +async def sandbox_busy_response(_request: Request, _exc: SandboxBusyError): + """忙碌只影响当前 runtime，不把重试误报成清理成功。""" + return JSONResponse( + status_code=503, + content={"detail": "sandbox busy"}, + headers={"Retry-After": "1"}, + ) + + def sandbox_response(record: SandboxRecord) -> SandboxResponse: return SandboxResponse( sandbox_id=record.sandbox_id, @@ -1971,7 +2003,7 @@ def sandbox_response(record: SandboxRecord) -> SandboxResponse: @app.get("/health") -def health(): +async def health(): tracked = len(idle_reaper._last_activity_at) # noqa: SLF001 return { "status": "ok", @@ -2210,7 +2242,20 @@ def delete_sandbox(sandbox_id: str, expected_generation: str | None = None): dependencies=[Depends(require_provisioner_auth)], ) async def proxy_sandbox_request(sandbox_id: str, request: Request, path: str = ""): - await asyncio.to_thread(sandbox_operation_pins.acquire, sandbox_id) + acquiring = asyncio.create_task( + asyncio.to_thread(sandbox_operation_pins.acquire, sandbox_id) + ) + try: + await asyncio.shield(acquiring) + except asyncio.CancelledError: + + def release_after_acquire(task): + """线程等待不会随协程取消，取得 pin 后仍须归还。""" + if not task.cancelled() and task.exception() is None: + sandbox_operation_pins.release(sandbox_id) + + acquiring.add_done_callback(release_after_acquire) + raise try: record = await asyncio.to_thread(backend_impl.discover, sandbox_id) except asyncio.CancelledError: diff --git a/docs/advanced/sandbox-operations.md b/docs/advanced/sandbox-operations.md index 1499cbfcc..8679dbfaa 100644 --- a/docs/advanced/sandbox-operations.md +++ b/docs/advanced/sandbox-operations.md @@ -109,6 +109,8 @@ Compose 默认把 `/var/run/docker.sock`、UserWorkspace 和 Skill projection provisioner 会扫描 Docker 已占用网段，从专用地址池中为新 Sandbox 选择不重叠的独立子网；并发分配冲突时重新选择，池耗尽时返回 503。`SANDBOX_DOCKER_SUBNET_PREFIX` 留空时 provisioner 使用 `/28`，最大只能设置为 `/29`，确保网络有足够地址容纳网关、provisioner 和 Sandbox。网络删除后子网可再次使用。部署者必须确认地址池不与宿主机路由、VPN 或其他 Docker 网络重叠，并在冲突时通过 `SANDBOX_DOCKER_ADDRESS_POOL` 覆盖 Compose 默认值。 +运行中点击停止会终止当前 GEM 命令，再由 worker 清理 execution runtime。`503 sandbox busy` 表示当前实例仍有代理操作或删除正在进行，调用方保留清理意图并重试；`/health` 正常不代表该实例已经清理。判断同一会话能否继续执行，需要同时核对 Run 终态和 `runtime_cleanup_pending=false`，详见[命令取消和清理完成](../mechanisms/sandbox.md#命令取消和清理完成)。 + ## Kubernetes 后端 Kubernetes 后端由 provisioner 创建沙盒 Pod 和 NodePort Service。Compose/宿主变量与容器变量如下： diff --git a/docs/develop-guides/decisions/implemented/2026-10-10-sandbox-command-cancellation.md b/docs/develop-guides/decisions/implemented/2026-10-10-sandbox-command-cancellation.md new file mode 100644 index 000000000..d929b5068 --- /dev/null +++ b/docs/develop-guides/decisions/implemented/2026-10-10-sandbox-command-cancellation.md @@ -0,0 +1,39 @@ +# 沙盒命令取消与有界清理 + +状态：implemented +类型：bug-fix +Owner：backend/package/yuxi/agents/backends/sandbox/backend.py + +## 问题 + +取消 Agent 的异步等待后，同步线程中的 GEM 命令调用仍可等待到原始长超时。provisioner 的删除与发现无限等待操作 pin，清理请求重试积累会占用同步线程资源；前端依据 Run 终态提前显示结束，同一线程的新消息仍被待清理 runtime 阻挡。 + +## 决策 + +异步命令使用独立 GEM shell session、异步启动与短周期等待，取消时先收敛启动请求，再显式终止该 session 的进程。重复取消等待有界收尾完成。未指定工具超时时使用现有 `SANDBOX_EXEC_TIMEOUT_SECONDS`；硬超时缺少退出码时返回失败码 124。 + +provisioner 的操作锁最多等待 5 秒，忙碌返回可重试的 503，失败的删除不能移除另一个删除者的栅栏。代理等待 pin 时取消，晚取得的 pin 仍被归还。健康检查使用异步路由，不依赖同步线程池。 + +PostgreSQL 的 cleanup fence 与 FIFO 保持原有所有权。当前运行接口返回终态但待清理的 Run；Web 的看门狗、断线和刷新路径等待清理完成后确认终态。worker 的清理异常记录具体 Run 和 traceback。 + +## 替代方案 + +- 继续取消同步线程等待：已经开始的远端调用仍运行。 +- 取消时直接放行下一条消息：旧命令和旧清理可能影响新 runtime。 +- 只增加线程数或缩短所有脚本超时：不能闭合取消传递与清理语义。 + +## 后果 + +每条异步命令增加 session 创建、短轮询和最终输出读取请求。忙碌清理仍阻挡同一会话，其他 runtime 与健康检查可继续响应。取消不回滚已经完成的文件和外部副作用。GEM 终止失败会记录异常，实际 runtime 删除仍以 generation-fenced 清理结果为准。 + +## 验证 + +- 命令取消 unit 覆盖启动确认前后取消、重复取消与硬超时失败码。 +- provisioner unit 覆盖活跃 pin 下删除超时、发现恢复、并发删除所有权和已取消代理的 pin 回收。 +- Web unit 覆盖看门狗、自然断流及新标签页恢复 completed/cancelled/interrupted 待清理 Run。 +- HTTP/worker/GEM 集成测试回读实际 PID、文件、PostgreSQL 清理标记、Redis end 事件和 generation；使用隔离 Schema，固定模型准备结果，不启动额外共享队列 worker。 +- 实际部署中的人工回归观察到两次长脚本停止：终止接口约 0.2 秒响应，generation 删除和终态事件约 3 秒完成，随后新 Run 可执行。该观察不替代当前补丁快照的 PostgreSQL 和完整 assembled-path E2E 验证。 + +上游补丁快照的实际检查：后端完整 unit 2520 项通过，相关后端 unit 198 项通过，Web 队列 unit 36 项通过；忙碌删除和线程池饱和下健康检查的真实 HTTP/GEM 两项通过。工程契约检查覆盖 184 份文档，其 unit 63 项通过，改动文件 Ruff、Web ESLint/Vite build 和 VitePress build 通过。前端全量未全绿，至少 5 项失败已在未打补丁的官方基线复现，不能将该命令记录为 Passed。 + +上游 PostgreSQL 集成和完整 API/ARQ/SSE E2E 在当前环境为 Not run；相关测试可在维护者配置的 Compose 环境执行。详见[事故复盘](../../postmortems/2026-10-10-sandbox-command-cancellation.md)。 diff --git a/docs/develop-guides/postmortems/2026-10-10-sandbox-command-cancellation.md b/docs/develop-guides/postmortems/2026-10-10-sandbox-command-cancellation.md new file mode 100644 index 000000000..aba2fac06 --- /dev/null +++ b/docs/develop-guides/postmortems/2026-10-10-sandbox-command-cancellation.md @@ -0,0 +1,42 @@ +# 长命令取消后 runtime 清理阻塞 + +日期：2026-10-10 +Owner：backend/package/yuxi/agents/backends/sandbox/backend.py +关联决策：[沙盒命令取消与有界清理](../decisions/implemented/2026-10-10-sandbox-command-cancellation.md) + +## 影响 + +一次实际部署发生长脚本停止后其他会话结束显示与队列状态不一致：回答已经结束，新消息继续排队。已确认命令调用在取消后持续到原始超时，Run 的 runtime 清理未完成；provisioner 曾约 30 分钟无法通过健康检查。未发现数据泄露或越权证据。受影响用户总数未知。 + +## 事实时间线 + +- 两条命令分别设置 1800 秒和 3600 秒超时，用户在超时之前点击停止；数据库先记录取消，执行调用之后才达到原定超时。 +- 第二条调用达到超时后约 1 秒，provisioner 恢复响应；随后同批 pending runtime cleanup 标记清除并发布终态事件。 +- 修复后用户在真实浏览器再次执行两次长脚本并停止。终止接口均约 0.2 秒响应，沙盒删除和终态事件均约 3 秒完成；同一对话随后可运行新任务。 + +## 因果链 + +`SandboxBackendProtocol.aexecute()` 默认把同步 `execute()` 放入线程。取消协程不会中断线程或远端命令。provisioner 删除等待长代理请求的 pin 排空，新发现等待删除栅栏；这些同步等待没有超时。终态但清理未完成的 Run 仍拥有 FIFO fence，前端轮询只检查终态则提前显示结束。 + +反复清理和发现请求耗尽同步线程池与现象吻合，且修复增加了线程池占满时健康检查的直接负向测试。故障当时没有捕获线程栈，因此不能把线程池耗尽写成已现场证明的事实。 + +## 安全网为何漏过 + +既有取消测试关闭的是异步生成器，没有证明同步线程中的真实 GEM 命令退出；清理测试只覆盖成功排空，缺少超时和重复删除场景。前端未把 durable cleanup 纳入终态投影。 + +## 修正与验证 + +修复 Owner、取舍和证据边界见[决策记录](../decisions/implemented/2026-10-10-sandbox-command-cancellation.md)。`test_sandbox_command_cancellation.py` 的旧实现不能进入原生异步取消路径；Web 的两个 cleanup-pending 负向用例在旧实现下提前清空 active Run，修复后保留订阅。 + +`test_sandbox_cancel_lifecycle.py` 在真实 HTTP、数据库、Redis、worker 执行函数和 GEM 边界回读 PID、文件、清理标记、generation 与 end 事件；模型与身份测试依赖使用固定输入，不能充当完整 assembled-path E2E。当前环境的上游 PostgreSQL 和共享 ARQ E2E 为 Not run。 + +## 防复发措施 + +- 命令取消由明确 session 的远端终止操作拥有；负向测试拒绝取消后残留进程或启动与 kill 乱序。 +- provisioner 操作锁等待有界；负向测试拒绝忙碌删除假成功、误释放其他删除者栅栏和 pin 泄漏。 +- 健康检查独立于同步线程池；真实 HTTP 负向测试在同步池占满时仍要求及时响应。 +- API 与 Web 共同投影 runtime cleanup；回归拒绝清理完成前的空闲判断和提前终态。 + +## 未解决风险 + +没有复刻生产现场线程栈。Kubernetes、真实外部模型、上游 PostgreSQL 完整运行链路尚未在当前环境执行。已完成的文件写入和外部操作不会随取消回滚。 diff --git a/docs/mechanisms/sandbox.md b/docs/mechanisms/sandbox.md index a33021e96..5de4b050a 100644 --- a/docs/mechanisms/sandbox.md +++ b/docs/mechanisms/sandbox.md @@ -42,6 +42,16 @@ Conversation 通过 `project_id` 绑定 Project；Project 拥有这项绑定和 `uid + runtime_scope_id` 派生稳定 `sandbox_id`。同一 runtime 存活期间不能改绑到另一个 Workdir。根执行树终态后，worker 清理 runtime，但保留 UserWorkspace 文件。 +## 命令取消和清理完成 + +Agent 的异步 `execute` 为每条命令创建独立 GEM shell session，异步启动命令并用短请求等待结果。工具的 `timeout` 同时限制远端进程的硬执行时间；未指定时使用 `SANDBOX_EXEC_TIMEOUT_SECONDS`。取消会先收敛已发出的启动请求，再调用该 session 的 `kill_process`。重复取消仍等待这个有界收尾完成，避免终止请求先到、命令后启动。 + +worker 将取消写入 PostgreSQL 后继续清理根 execution runtime。Run 进入终态与 runtime 清理完成是两个事实：`runtime_cleanup_pending=true` 时，同一用户、智能体和 Conversation 的 FIFO 仍被阻挡，终态 `end` 事件等待清理成功后发布。当前运行接口也返回终态但尚未清理的 Run，前端在无事件轮询、断线和刷新恢复时检查这项标记。 + +provisioner 的操作 pin 保护正在进行的代理请求。删除等待这些请求排空，新的发现和代理请求等待删除结束；单次锁等待最多 5 秒，超时返回 `503 sandbox busy` 和 `Retry-After: 1`，不删除活跃实例、不宣称清理成功。超时会释放本次删除栅栏，worker 保留数据库中的清理意图，由周期恢复重试。健康检查使用异步路由，避免与同步锁等待竞争线程池。 + +远端进程终止失败会记录包含 thread/session 身份的异常；实际 runtime 删除仍须由 provisioner 确认，数据库清理标记不会因终止请求失败而被直接清除。已经完成的文件写入和外部请求保留，取消不回滚这些副作用。 + ## 挂载和文件 Owner | 虚拟路径 | 内容 | Agent 权限 | diff --git a/web/src/composables/useAgentRunStream.js b/web/src/composables/useAgentRunStream.js index e40dd844f..50664a160 100644 --- a/web/src/composables/useAgentRunStream.js +++ b/web/src/composables/useAgentRunStream.js @@ -332,8 +332,9 @@ export function useAgentRunStream({ if (Date.now() - lastEventAt < 45000) return try { const runRes = await agentApi.getAgentRun(runId) - const st = runRes?.run?.status - if (st && RUN_TERMINAL_STATUSES.has(st)) { + const run = runRes?.run + const st = run?.status + if (st && !run.runtime_cleanup_pending && RUN_TERMINAL_STATUSES.has(st)) { clearInterval(idleWatchdog) finalizeRunStream(threadId, runId, touchedThreadIds, { status: st }) } @@ -423,7 +424,9 @@ export function useAgentRunStream({ try { const runRes = await agentApi.getAgentRun(runId) const run = runRes?.run - if (run?.status === RUN_INTERRUPTED_STATUS) { + if (run?.runtime_cleanup_pending) { + scheduleRunReconnect(threadId, runId) + } else if (run?.status === RUN_INTERRUPTED_STATUS) { if (hasPendingInterruptInThreads(touchedThreadIds, run.id)) { await preserveInterruptedRun(threadId, run) } else { @@ -472,6 +475,7 @@ export function useAgentRunStream({ try { const runRes = await agentApi.getAgentRun(ts.activeRunId) const run = runRes?.run + if (run?.runtime_cleanup_pending) return if (run?.status === RUN_INTERRUPTED_STATUS) { stopRunStreamSubscription(threadId) const snapshot = loadActiveRunSnapshot(threadId) @@ -502,14 +506,14 @@ export function useAgentRunStream({ try { const runRes = await agentApi.getAgentRun(snapshot.run_id) const run = runRes?.run - if (run?.status === RUN_INTERRUPTED_STATUS) { + if (run?.status === RUN_INTERRUPTED_STATUS && !run.runtime_cleanup_pending) { // 仅当本地仍持有该中断时才据快照恢复；否则不能仅凭快照重放旧中断 // （可能已被回复），交由下方 active_run 做权威判定。 if (hasPendingInterruptForRun(ts, run.id)) { await preserveInterruptedRun(threadId, run, snapshot) return } - } else if (run && !RUN_TERMINAL_STATUSES.has(run.status)) { + } else if (run && (run.runtime_cleanup_pending || !RUN_TERMINAL_STATUSES.has(run.status))) { const afterSeq = resolveRunResumeAfterSeq({ snapshot, threadState: ts @@ -533,7 +537,7 @@ export function useAgentRunStream({ try { const active = await agentApi.getThreadActiveRun(threadId) const run = active?.run - if (run?.status === RUN_INTERRUPTED_STATUS) { + if (run?.status === RUN_INTERRUPTED_STATUS && !run.runtime_cleanup_pending) { if (hasPendingInterruptForRun(ts, run.id)) { await preserveInterruptedRun(threadId, run) return @@ -544,7 +548,7 @@ export function useAgentRunStream({ }) return } - if (run && !RUN_TERMINAL_STATUSES.has(run.status)) { + if (run && (run.runtime_cleanup_pending || !RUN_TERMINAL_STATUSES.has(run.status))) { resetOnGoingConv(threadId) await startRunStream(threadId, run.id, '0-0') return diff --git a/web/test/unit/agentRequestQueue.test.js b/web/test/unit/agentRequestQueue.test.js index 9d516063d..1dc1615b4 100644 --- a/web/test/unit/agentRequestQueue.test.js +++ b/web/test/unit/agentRequestQueue.test.js @@ -1,6 +1,7 @@ import assert from 'node:assert/strict' import path from 'node:path' import { after, before, test } from 'node:test' +import { setImmediate } from 'node:timers' import { fileURLToPath } from 'node:url' import { createServer } from 'vite' @@ -879,6 +880,85 @@ test('自然断流后从 PG 终态复用统一清理并刷新历史', async () = } }) +test('终态清理未完成时看门狗继续等待，清理完成后才结束', async (t) => { + const threadState = { activeRunId: null, runLastSeq: '0-0', replyLoadingVisible: true } + let clock = 1000 + let watchdog + let body + let cleanupPending = true + t.mock.method(Date, 'now', () => clock) + t.mock.method(globalThis, 'setInterval', (callback) => { + watchdog = callback + return 0 + }) + t.mock.method(agentApi, 'streamAgentRunEvents', async () => + new Response(new ReadableStream({ start: (controller) => { body = controller } })) + ) + t.mock.method(agentApi, 'getAgentRun', async () => ({ + run: { id: 'run-cleanup', status: 'completed', runtime_cleanup_pending: cleanupPending } + })) + const stream = createRunStream({ threadState, handleStreamChunk: () => {}, resetOnGoingConv: () => {} }) + const running = stream.startRunStream('thread-cleanup', 'run-cleanup') + try { + await new Promise((resolve) => setImmediate(resolve)) + clock += 60000 + await watchdog() + assert.equal(threadState.activeRunId, 'run-cleanup') + assert.equal(threadState.isStreaming, true) + assert.equal(threadState.replyLoadingVisible, true) + cleanupPending = false + await watchdog() + assert.equal(threadState.activeRunId, null) + assert.equal(threadState.replyLoadingVisible, false) + } finally { + body.close() + await running + threadState.activeRunId = null + localStorage.removeItem('active_run:thread-cleanup') + } +}) + +test('自然断流时清理未完成不能显示已结束', async (t) => { + const threadState = { activeRunId: null, runLastSeq: '0-0', replyLoadingVisible: true } + t.mock.method(agentApi, 'streamAgentRunEvents', async () => new Response('')) + t.mock.method(agentApi, 'getAgentRun', async () => ({ + run: { id: 'run-cleanup', status: 'cancelled', runtime_cleanup_pending: true } + })) + const stream = createRunStream({ threadState, handleStreamChunk: () => {}, resetOnGoingConv: () => {} }) + try { + await stream.startRunStream('thread-cleanup', 'run-cleanup') + assert.equal(threadState.activeRunId, 'run-cleanup') + assert.equal(threadState.replyLoadingVisible, true) + } finally { + threadState.activeRunId = null + localStorage.removeItem('active_run:thread-cleanup') + } +}) + +for (const status of ['completed', 'cancelled', 'interrupted']) { + test(`新标签页恢复 ${status} 且清理未完成的 Run 时继续订阅`, async (t) => { + const threadId = `cleanup-${status}` + const threadState = { activeRunId: null, runLastSeq: '0-0', replyLoadingVisible: true } + t.mock.method(agentApi, 'getThreadActiveRun', async () => ({ + run: { id: 'run-cleanup', status, runtime_cleanup_pending: true } + })) + t.mock.method(agentApi, 'getAgentRun', async () => ({ + run: { id: 'run-cleanup', status, runtime_cleanup_pending: true } + })) + t.mock.method(agentApi, 'streamAgentRunEvents', async () => new Response('')) + const stream = createRunStream({ threadState, handleStreamChunk: () => {}, resetOnGoingConv: () => {} }) + try { + await stream.resumeActiveRunForThread(threadId) + assert.equal(threadState.activeRunId, 'run-cleanup') + assert.equal(threadState.isStreaming, true) + assert.equal(threadState.replyLoadingVisible, true) + } finally { + threadState.activeRunId = null + localStorage.removeItem(`active_run:${threadId}`) + } + }) +} + test('自然断流遇到仍持有的 pending interrupt 时保留 Run 快照', async () => { const threadId = 'thread-with-pending-interrupt' const threadState = {