diff --git a/cloud/sandbox.py b/cloud/sandbox.py index ff801ff..f862113 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -14,11 +14,13 @@ """ import json +import logging import os import re import shutil import subprocess import sys +import time from cloud.marker_tools import extract_task_cancels as _extract_task_cancels from cloud.marker_tools import extract_task_creates as _extract_task_creates @@ -29,6 +31,12 @@ from cloud.peer_messages import extract_peer_messages as _extract_peer_messages MAX_STDOUT_CHARS = 2000 +# 启动前分段日志（2026-10-29 任务 A）：docker run 是阻塞调用， +# 容器输出要等返回才回传，中途卡死时 worker 侧没有任何标记； +# 这里的开始/结束/异常行进 worker 日志（stderr，不缓冲丢行）， +# 只记容器名与镜像，绝不记内层命令（内含任务 prompt）。 +_DOCKER_LOGGER = logging.getLogger("cloud.docker") + class Executor: """执行器基类。 @@ -1190,11 +1198,20 @@ class DockerExecutor(Executor): extra_mounts=extra_mounts, network=network, extra_env=extra_env) + _docker_t0 = time.time() + _DOCKER_LOGGER.info( + "docker_run_start container=%s image=%s", + self.container_name(task), self.image) try: result = self._call_runner( cmd, timeout=timeout, env=self._docker_cli_env(extra_env)) except Exception as exc: + _DOCKER_LOGGER.warning( + "docker_run_error container=%s elapsed_ms=%d error=%s", + self.container_name(task), + int((time.time() - _docker_t0) * 1000), + exc.__class__.__name__) if exc.__class__.__name__ == "TimeoutExpired": self.kill_container(task) return { @@ -1214,6 +1231,11 @@ class DockerExecutor(Executor): parsed = _parse_host_result(task_id, fake) parsed["executor"] = "docker" parsed["container"] = self.container_name(task) + _DOCKER_LOGGER.info( + "docker_run_done container=%s elapsed_ms=%d returncode=%s", + self.container_name(task), + int((time.time() - _docker_t0) * 1000), + parsed.get("returncode")) return parsed def dispatch_task(self, task, timeout=None, extra_mounts=None, diff --git a/cloud/tests/test_prestart_stages.py b/cloud/tests/test_prestart_stages.py new file mode 100644 index 0000000..ccdd2b8 --- /dev/null +++ b/cloud/tests/test_prestart_stages.py @@ -0,0 +1,164 @@ +"""启动前分段日志（2026-10-29 任务 A）。 + +任务 A 认领后 61 秒 interrupted，但阻塞的 docker run 中途 +没有任何标记、容器 --rm 后输出也没了，事后无法定位卡在哪 +一段。本文件断言三层标记： +- worker 派发开始/结束进审计（PG 落盘，容器没了也丢不了）； +- docker run 开始/结束/异常进 worker 日志，只记容器名与镜像； +- 失败/中断时容器输出尾巴记进 worker 日志； +- 看门狗杀容器时记一笔（之前是静默杀）。 +标记绝不含任务 prompt 与密钥值。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.sandbox import DockerExecutor + from cloud.server import init_db + from cloud.worker import Worker +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.sandbox import DockerExecutor + from cloud.server import init_db + from cloud.worker import Worker + + +def _audit_actions(db_path): + conn = sqlite3.connect(db_path) + try: + rows = conn.execute( + "SELECT action FROM audit_log ORDER BY id").fetchall() + finally: + conn.close() + return [r[0] for r in rows] + + +class FakeExecutor: + """非 Docker/非 Local 的假执行器：走 _run_attempt 同步分支。""" + + def __init__(self, result=None, error=None): + self._result = result + self._error = error + self.killed = [] + + def dispatch_task(self, task): + if self._error is not None: + raise self._error + return dict(self._result) + + def kill_container(self, name): + self.killed.append(name) + + +class PrestartStagesTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "test.db") + init_db(self.db_path) + self.worker = Worker( + self.db_path, "/fake/host.py", env={"XCM_API_KEY": "x"}, + task_timeout=1800) + + def tearDown(self): + self.tmp.cleanup() + + def _task(self): + return {"id": "T1", "project_id": "p1", "prompt": "SECRET_PROMPT"} + + def test_exec_start_and_done_audited(self): + self.worker.executor = FakeExecutor(result={ + "executor": "fake", "task_id": "T1", "status": "completed", + "stdout": "", "returncode": 0, "host_status": None, + "stderr": ""}) + result, error = self.worker._run_attempt(self._task()) + self.assertIsNone(error) + self.assertEqual(result["status"], "completed") + actions = _audit_actions(self.db_path) + self.assertIn("worker_exec_start", actions) + self.assertIn("worker_exec_done", actions) + self.assertNotIn("worker_exec_error", actions) + + def test_exec_error_audited_not_done(self): + self.worker.executor = FakeExecutor( + error=RuntimeError("boom")) + result, error = self.worker._run_attempt(self._task()) + self.assertEqual(result["status"], "failed") + self.assertEqual(error, "boom") + actions = _audit_actions(self.db_path) + self.assertIn("worker_exec_start", actions) + self.assertIn("worker_exec_error", actions) + self.assertNotIn("worker_exec_done", actions) + + def test_failed_stdout_tail_logged_not_audited(self): + self.worker.executor = FakeExecutor(result={ + "executor": "fake", "task_id": "T1", "status": "failed", + "stdout": "partial container output before death", + "returncode": 1, "host_status": None, "stderr": ""}) + with self.assertLogs("cloud.worker", level="INFO") as logs: + self.worker._run_attempt(self._task()) + tail_lines = [r for r in logs.output + if "worker_exec_stdout_tail" in r] + self.assertEqual(len(tail_lines), 1) + self.assertIn("partial container output before death", + tail_lines[0]) + self.assertNotIn("SECRET_PROMPT", "\n".join(logs.output)) + + def test_cancel_watchdog_kill_logged(self): + self.worker.executor = FakeExecutor() + self.worker._cancel_requested = {"T1"} + with self.assertLogs("cloud.worker", level="INFO") as logs: + self.worker._docker_cancel_watchdog( + "T1", "codex-T1", threading.Event()) + self.assertEqual(self.worker.executor.killed, ["codex-T1"]) + # log_event 把事件名放在记录的 event 属性里，不在消息文本里。 + self.assertTrue(any(getattr(r, "event", None) + == "worker_docker_killed" + for r in logs.records)) + + +class DockerRunStagesTestCase(unittest.TestCase): + def _executor(self, runner): + ex = DockerExecutor(runner=runner, env={"XCM_API_KEY": "x"}) + ex._check_docker = lambda: None + return ex + + def _task(self): + return {"id": "a" * 32, "project_id": "p1", + "prompt": "SECRET_PROMPT"} + + def test_docker_run_start_and_done_logged(self): + def runner(cmd, timeout=None, env=None): + return SimpleNamespace( + returncode=0, stdout=json.dumps({"status": "completed"}), + stderr="") + ex = self._executor(runner) + with self.assertLogs("cloud.docker", level="INFO") as logs: + result = ex._run_in_container(self._task(), ["python3", "x"]) + self.assertEqual(result["status"], "completed") + joined = "\n".join(logs.output) + self.assertIn("docker_run_start", joined) + self.assertIn("docker_run_done", joined) + self.assertIn("codex-" + "a" * 32, joined) + self.assertNotIn("SECRET_PROMPT", joined) + + def test_docker_run_error_logged_and_failed(self): + def runner(cmd, timeout=None, env=None): + raise RuntimeError("docker daemon down") + ex = self._executor(runner) + with self.assertLogs("cloud.docker", level="WARNING") as logs: + result = ex._run_in_container(self._task(), ["python3", "x"]) + self.assertEqual(result["status"], "failed") + self.assertTrue(any("docker_run_error" in r + for r in logs.output)) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 05878dc..bd86ad4 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -1983,6 +1983,11 @@ class Worker: kill = getattr(self.executor, "kill_container", None) if kill is not None and container_name: kill(container_name) + log_event(self.logger, "worker_docker_killed", + task_id=task_id, + message="cancel watchdog killed container " + "%s (task interrupted/cancelled)" + % container_name) except Exception: pass return @@ -2018,6 +2023,10 @@ class Worker: kill = getattr(self.executor, "kill_container", None) if kill is not None and container_name: kill(container_name) + log_event(self.logger, "worker_docker_killed", + task_id=task_id, + message="quota watchdog killed container %s " + "(quota exceeded)" % container_name) except Exception: pass return @@ -3534,6 +3543,33 @@ class Worker: _watchdog_stop, quota_check), daemon=True) _quota_watchdog_thread.start() + # 启动前分段日志（2026-10-29 任务 A）：认领后 61 秒中断， + # 但阻塞的 docker run 中途没有任何标记、容器 --rm 后输出 + # 也没了，事后完全无法定位卡在哪一段。派发开始/结束进 + # 审计（PG 落盘，容器没了也丢不了），下次卡死时最后一个 + # 标记直接说明卡在哪一段；kind/容器名/超时不含 prompt/key。 + _exec_kind = ("goal-resume-breakpoint" + if task.get("_goal_resume_breakpoint") + else "goal-resume" if task.get("_resume") + else "goal" if goal_mode else "dispatch") + _exec_container = None + if hasattr(self.executor, "container_name"): + try: + _exec_container = self.executor.container_name(task) + except Exception: + _exec_container = None + _exec_t0 = time.time() + _exec_error = None + _exec_detail = "kind=%s executor=%s timeout=%s" % ( + _exec_kind, self._executor_name(), + "None" if _effective_task_timeout is None + else _effective_task_timeout) + if _exec_container: + _exec_detail += " container=%s" % _exec_container + self._audit(task.get("project_id"), "worker_exec_start", + _exec_detail) + log_event(self.logger, "worker_exec_start", task_id=task_id, + message="worker exec start: %s" % _exec_detail) try: gateway_provider, gateway_env = self._ensure_gateway(task) if self._use_popen(): @@ -3726,6 +3762,7 @@ class Worker: "status": "failed", "stdout": "", "returncode": None, "host_status": None, "stderr": ""} error = str(exc) + _exec_error = exc finally: if _watchdog_stop is not None: _watchdog_stop.set() @@ -3753,6 +3790,34 @@ class Worker: entry = self._running_snapshot().get(task_id) or {} if entry.get("proc") is None: self._running_clear(task_id) + # 启动前分段日志（续）：派发结束标记与失败输出尾巴。结束 + # 标记与开始标记配对：只有开始没有结束 = 卡在执行器调用 + # 里；失败/中断时的容器输出尾巴记进 worker 日志——任务 A + # 当时这部分输出随 --rm 消失，是事后无法定位的另一半。 + _exec_elapsed_ms = int((time.time() - _exec_t0) * 1000) + if _exec_error is not None: + self._audit(task.get("project_id"), "worker_exec_error", + "elapsed_ms=%d error=%s" + % (_exec_elapsed_ms, + _exec_error.__class__.__name__)) + log_event(self.logger, "worker_exec_error", task_id=task_id, + message="worker exec error: elapsed_ms=%d error=%s" + % (_exec_elapsed_ms, + _exec_error.__class__.__name__)) + else: + self._audit(task.get("project_id"), "worker_exec_done", + "elapsed_ms=%d status=%s returncode=%s" + % (_exec_elapsed_ms, result.get("status"), + result.get("returncode"))) + log_event(self.logger, "worker_exec_done", task_id=task_id, + status=result.get("status"), + message="worker exec done: elapsed_ms=%d status=%s" + % (_exec_elapsed_ms, result.get("status"))) + _exec_out = result.get("stdout", "") or "" + if result.get("status") in ("failed", "interrupted") and _exec_out: + self.logger.info( + "worker_exec_stdout_tail task_id=%s tail=%s", + task_id, _exec_out[-2000:]) # Docker 超时标记：执行器已在超时时 kill+rm 容器并返回 # failed（stdout 含 timeout），这里把状态统一成 timeout， # 调用方据此置 failed 并记 worker_timeout 审计。