From 6f0eec71b6722d28d8ef32b7637c88d1442a879d Mon Sep 17 00:00:00 2001 From: host Date: Thu, 29 Oct 2026 17:46:12 +0800 Subject: [PATCH] worker: cover goal docker paths with prestart stage logging PR #145 added docker_run_* lines only to _run_in_container, but goal/goal-resume/goal-resume-breakpoint call _call_runner directly (to keep full stdout for goal conclusions), so the production live test (a goal task) produced zero docker_run_ lines. Extract a _docker_call_staged helper and route all four task paths through it; error lines log the class name then re-raise so each caller's kill+failed handling is unchanged. Also fix audit attribution: audit_log has no task_id column, and worker_exec_done/error details carried no task id, so per-task queries missed them (the rows existed). All three exec details now start with task_id=. Tests: 2 new goal-path stage cases + a done detail task-id assertion (8 total in test_prestart_stages). --- cloud/sandbox.py | 61 ++++++++++++++++++----------- cloud/tests/test_prestart_stages.py | 31 +++++++++++++++ cloud/worker.py | 12 +++--- 3 files changed, 76 insertions(+), 28 deletions(-) diff --git a/cloud/sandbox.py b/cloud/sandbox.py index f862113..1233ce0 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -1113,6 +1113,37 @@ class DockerExecutor(Executor): raise RuntimeError( "docker runner 不支持 env，无法按名注入任务密钥") + def _docker_call_staged(self, task, cmd, timeout=None, env=None): + """带分段日志的 docker run 调用（2026-10-29 任务 A 补漏）。 + + goal/goal-resume/goal-resume-breakpoint 不走 _run_in_container + （直接调 _call_runner 保完整 stdout 结论），上一批只给 + _run_in_container 加的标记漏了这三条真正在用的路径—— + 生产实测任务就是 goal 路径，日志里一条 docker_run_ 都没有。 + 本 helper 统一四条任务路径的开始/结束/异常行；异常只记 + 类名后原样重抛，各调用点的 kill+failed 处理不变。只记 + 容器名与镜像，绝不记命令（内含 prompt）。 + """ + _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=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() - _t0) * 1000), + exc.__class__.__name__) + raise + _DOCKER_LOGGER.info( + "docker_run_done container=%s elapsed_ms=%d returncode=%s", + self.container_name(task), + int((time.time() - _t0) * 1000), + getattr(result, "returncode", None)) + return result + def list_containers(self, prefix="codex-"): """列出名字以 prefix 开头的容器（默认 codex-），失败返回 []。 @@ -1198,20 +1229,11 @@ 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, + result = self._docker_call_staged( + task, 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 { @@ -1231,11 +1253,6 @@ 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, @@ -1357,8 +1374,8 @@ class DockerExecutor(Executor): network=network, extra_env=extra_env) try: - result = self._call_runner( - cmd, timeout=timeout, + result = self._docker_call_staged( + task, cmd, timeout=timeout, env=self._docker_cli_env(extra_env)) except Exception as exc: if exc.__class__.__name__ == "TimeoutExpired": @@ -1472,8 +1489,8 @@ class DockerExecutor(Executor): network=network, extra_env=extra_env) try: - result = self._call_runner( - cmd, timeout=timeout, + result = self._docker_call_staged( + task, cmd, timeout=timeout, env=self._docker_cli_env(extra_env)) except Exception as exc: if exc.__class__.__name__ == "TimeoutExpired": @@ -1582,8 +1599,8 @@ class DockerExecutor(Executor): network=network, extra_env=extra_env) try: - result = self._call_runner( - cmd, timeout=timeout, + result = self._docker_call_staged( + task, cmd, timeout=timeout, env=self._docker_cli_env(extra_env)) except Exception as exc: if exc.__class__.__name__ == "TimeoutExpired": diff --git a/cloud/tests/test_prestart_stages.py b/cloud/tests/test_prestart_stages.py index ccdd2b8..c3c6751 100644 --- a/cloud/tests/test_prestart_stages.py +++ b/cloud/tests/test_prestart_stages.py @@ -85,6 +85,15 @@ class PrestartStagesTestCase(unittest.TestCase): self.assertIn("worker_exec_start", actions) self.assertIn("worker_exec_done", actions) self.assertNotIn("worker_exec_error", actions) + # 审计表没有任务 id 列，detail 必须自带任务 id 才能对应。 + conn = sqlite3.connect(self.db_path) + try: + detail = conn.execute( + "SELECT detail FROM audit_log" + " WHERE action = 'worker_exec_done'").fetchone()[0] + finally: + conn.close() + self.assertIn("T1", detail) def test_exec_error_audited_not_done(self): self.worker.executor = FakeExecutor( @@ -159,6 +168,28 @@ class DockerRunStagesTestCase(unittest.TestCase): self.assertTrue(any("docker_run_error" in r for r in logs.output)) + def test_goal_task_stage_lines_logged(self): + # goal 路径不走 _run_in_container（直接调 runner 保完整 + # stdout 结论），上一批的标记漏了这条真正在用的路径。 + def runner(cmd, timeout=None, env=None): + return SimpleNamespace(returncode=0, stdout="", stderr="") + ex = self._executor(runner) + with self.assertLogs("cloud.docker", level="INFO") as logs: + ex.goal_task(self._task()) + joined = "\n".join(logs.output) + self.assertIn("docker_run_start", joined) + self.assertIn("docker_run_done", joined) + + def test_goal_resume_breakpoint_stage_lines_logged(self): + def runner(cmd, timeout=None, env=None): + return SimpleNamespace(returncode=0, stdout="", stderr="") + ex = self._executor(runner) + with self.assertLogs("cloud.docker", level="INFO") as logs: + ex.goal_resume_breakpoint_task(self._task()) + joined = "\n".join(logs.output) + self.assertIn("docker_run_start", joined) + self.assertIn("docker_run_done", joined) + if __name__ == "__main__": unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index bd86ad4..797535d 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -3560,8 +3560,8 @@ class Worker: _exec_container = None _exec_t0 = time.time() _exec_error = None - _exec_detail = "kind=%s executor=%s timeout=%s" % ( - _exec_kind, self._executor_name(), + _exec_detail = "task_id=%s kind=%s executor=%s timeout=%s" % ( + task_id, _exec_kind, self._executor_name(), "None" if _effective_task_timeout is None else _effective_task_timeout) if _exec_container: @@ -3797,8 +3797,8 @@ class Worker: _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, + "task_id=%s elapsed_ms=%d error=%s" + % (task_id, _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" @@ -3806,8 +3806,8 @@ class Worker: _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"), + "task_id=%s elapsed_ms=%d status=%s returncode=%s" + % (task_id, _exec_elapsed_ms, result.get("status"), result.get("returncode"))) log_event(self.logger, "worker_exec_done", task_id=task_id, status=result.get("status"),