diff --git a/cloud/README.md b/cloud/README.md index 820f0e9..b1810ed 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -1252,6 +1252,55 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 仍是串行一轮一任务）。两个端点都要 X-API-Key（与其它已认证 端点一致）；worker 直接写库注册时不需要 key，HTTP 心跳是给有 key 的外部 worker 用的另一条路。 +- 沙箱规模（本轮新增，见 `cloud/tests/test_sandbox_scale.py`）： + - /tmp 隔离分层表（唯一口径 `sandbox.tmp_isolation_status()`）： + Docker 任务容器 `--tmpfs /tmp` + `--tmpfs /var/tmp`（容器随 + `--rm` 消失，/tmp 也随之消失，绝不落宿主 /tmp）；且 tmpfs + 挂载在全部 `-v` 之前，附件 seed 的 `/tmp/codex-seed:ro` 子挂载 + 压在 tmpfs 之上（旧顺序把 tmpfs 放最后，本 VM 无 docker 未 + 证实旧顺序确实遮挡，故写为顺序加严而非已证实 bug）。 + 宿主级 codex workspace-write（bwrap）仍不隔离：挂载由 codex + 二进制内部决定，host.py 只传 sandboxPolicy 类型、无挂载旋钮， + 本仓库改不了它；需要 /tmp 隔离的任务请走 docker 执行器， + 不要拿 Docker 层的隔离给宿主级任务背书。 + - workers 注册表寻址：workers 表新增 host（worker 的 + `--host-name`，容器内 worker 绝不能用容器名冒充宿主名）、 + pid（worker 进程号）、address（CLI `--worker-address`，显式给 + 才有，不含任何凭证）；老库幂等补列，老行三列为 NULL 按未知 + 处理。`POST /workers/heartbeat` 同样接受三字段（host 非字符串、 + pid 非整数、address 非字符串时 400）；心跳未给时保留旧值， + worker 每次心跳都显式刷新本机 host 与本进程 pid，重启复用 + 同 worker_id 时不残留旧 pid。 + - 失联容器精确清理（`workers.orphaned_containers()` + + `Worker.cleanup_orphaned_containers()`，只写识别与杀容器， + 不改任务状态）：容器名必须精确匹配 `codex-<32 位 hex 任务 id>`， + 且任务行存在、状态为 running/interrupted、认领记录 + `tasks.claimed_by_worker`（`_claim` 在 CAS 认领时写）指向 + 心跳过期的 worker，才算失联；基础设施容器（如 codex-postgres）、 + 非任务名、已完成任务、无认领记录、owner 仍存活的一律排除， + 绝不按 `codex-` 前缀删。默认只杀 pid_verified 的容器（本机 host + + 已知 pid 且 pid 明确已死）；远端 host / pid 未知 / 未验证的 + 只列不杀，显式 `verified_only=False` 才杀（仅单机语义已确认 + 时开，跨主机场景绝不能默认开）。旧的 + `cleanup_stale_containers()`（任务正则兜底、不区分 owner）保留 + 不改行为，供认领记录缺失的老任务容器用；任务状态本身由 + `sweep_stale_tasks`（心跳过期 -> interrupted）负责，两者分工。 + 本轮只写代码与测试，绝不在本轮动生产容器、不杀任何生产任务。 + - 长任务天花板与并行验证的代码支撑（不宣称已大规模实测）： + 超时边界抽成纯函数 `worker.docker_run_timeout()`（取较小值、 + None 语义、no_time_limit 语义、非法值回落，行为与原内联一致）； + 取消 -> interrupted（执行结果绝不覆盖）、超时 -> failed、 + 恢复靠断点续跑（goal-resume / run 续跑），三者状态语义不变。 + 天花板验证方法（待真实大规模跑时按此执行，非本轮已做）： + 睡眠循环任务（每轮 sleep + 一行回复）测轮次/墙钟上限、 + 取消与超时各走一遍终态、owner 崩溃后 sweep + 精确清理各走 + 一遍；多端并行只做 web 自适应跨视口：console 已含 viewport + meta 与小屏 media 查询（单测只守基线），手工验证矩阵为 + 320 / 375 / 768 / 1024 / 1440 宽度（无横向滚动、导航可收起、 + 任务列表与详情可读），iOS/Android 原生明确不在本轮范围。 + 未验证项（诚实清单）：真实 40 人并行、真实大规模长任务天花板、 + 真实 docker 下 tmpfs 顺序是否曾遮挡 seed、宿主级 bwrap /tmp + 隔离（本仓库无修复手段，见上）。 - 任务按执行器池派发：tasks 表新增 `executor` 列（可空，老库幂等 补列）；`POST /tasks` 接受可选 executor（local/docker/cloud， 非法值 400），未传时为 None 表示任意 worker 可认领；worker diff --git a/cloud/db.py b/cloud/db.py index 1b3f286..bc33792 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -1115,6 +1115,7 @@ def init_postgres_schema(conn): for _name, _type in (("executor", "TEXT"), ("pr_url", "TEXT"), ("pr_number", "INTEGER"), ("worktree_host", "TEXT"), + ("claimed_by_worker", "TEXT"), ("goal_mode", "INTEGER"), ("environment_id", "TEXT"), ("child_model", "TEXT"), diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 5d26e94..c522367 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -41,6 +41,33 @@ from cloud.marker_tools import extract_wait_threads as _extract_wait_threads from cloud.peer_messages import extract_peer_messages as _extract_peer_messages MAX_STDOUT_CHARS = 2000 + + +def tmp_isolation_status(): + """各层 /tmp 隔离现状的唯一口径（README 与单测都以它为准）。 + + 返回 {docker: bool, host_bwrap: bool, ...}： + - docker=True：DockerExecutor 给每个任务容器挂 --tmpfs /tmp + 与 --tmpfs /var/tmp（容器随 --rm 消失，/tmp 也随之消失， + 绝不落宿主 /tmp；见 build_run_command 的顺序注释）。 + - host_bwrap=False：宿主级 codex workspace-write（bwrap） + 的挂载由 codex 二进制内部决定，host.py 只传 + sandboxPolicy 类型、没有任何挂载旋钮，所以本仓库改不 + 了它；2026-10-26 宿主实测 agent 写 /tmp 落到宿主 /tmp， + 本轮未在宿主重测、也不宣称已修复。需要 /tmp 隔离的 + 任务请走 docker 执行器，不要走宿主级沙箱 worker。 + 本函数是状态宣告，不是修复开关：调用方绝不能拿它的 + docker=True 去给宿主级任务背书。 + """ + return { + "docker": True, + "host_bwrap": False, + "docker_detail": "per-container tmpfs /tmp + /var/tmp, removed with --rm", + "host_bwrap_detail": "bwrap mounts are owned by the codex binary;" + " host.py passes only a sandboxPolicy type" + " and has no mount knob; /tmp is shared" + " with the host (2026-10-26 host test)", + } # 任务结果文本上限：2000 字截断的是 host.py 打印的整段 JSON， # text 超过 2000 字时 JSON 本身被截断，worker 再解析 stdout # 取 text 必然失败，存库的就变成截断的 JSON（2026-10-30 @@ -1111,23 +1138,33 @@ class DockerExecutor(Executor): cmd += ["--runtime", self.runtime] if network: cmd += ["--network", network] - cmd += ["-v", "%s:/work" % self._worktree_for(task), - "-v", "%s:/app:ro" % self._app_dir(), - "-w", "/work"] - # 额外挂载放在基础挂载之后：/app/work 这类子路径挂载会 - # 覆盖 /app:ro 的对应子路径，让容器内 host.py 能写 - # work/ 和 host-state.json，同时代码主体仍只读。 - mounts = self.extra_mounts if extra_mounts is None \ - else self._parse_extra_mounts(extra_mounts) - for host_p, cont_p, mode in mounts: - cmd += ["-v", "%s:%s:%s" % (host_p, cont_p, mode)] - # /tmp 固定为 tmpfs：容器内 /tmp 绝不落在宿主 /tmp，也不受 - # 额外挂载影响（/tmp 目标的额外挂载已在解析时拒绝）。 + # /tmp 固定为 tmpfs，且必须放在所有 -v 挂载之前：父挂载 + # 在前、子挂载在后是 Linux 挂载顺序的基本要求——附件 + # seed 以 /tmp/codex-seed:ro 挂进容器（见 dispatch_task）， + # 它是 /tmp 之下的子挂载，只有 tmpfs 先挂、seed 后挂， + # seed 才不会被后挂的 tmpfs 遮掉。旧顺序把 --tmpfs 放在 + # 全部 -v 之后，本 VM 无 docker、未在真实 docker 上证实 + # 旧顺序确实遮挡（docker 可能按父子关系重排），所以这里 + # 写为顺序加严而非“已证实 bug 修复”；单测只断言顺序， + # 不宣称旧顺序一定失效。 + # 容器内 /tmp 绝不落在宿主 /tmp，也不受额外挂载影响 + # （/tmp 目标的额外挂载已在解析时拒绝）。 # 诚实边界：这是 Docker 外层能做的；内层 codex 的 bwrap # workspace-write 仍由 codex 二进制控制，本仓库改不了它 # 的 --proc 挂载，宿主级 bwrap 的 /tmp 不隔离问题也不在 # 本仓库的修复范围内——不要把本改动说成“内层沙箱已修”。 cmd += ["--tmpfs", "/tmp", "--tmpfs", "/var/tmp"] + cmd += ["-v", "%s:/work" % self._worktree_for(task), + "-v", "%s:/app:ro" % self._app_dir(), + "-w", "/work"] + # 额外挂载放在基础挂载之后：/app/work 这类子路径挂载会 + # 覆盖 /app:ro 的对应子路径，让容器内 host.py 能写 + # work/ 和 host-state.json，同时代码主体仍只读；seed 的 + # /tmp/codex-seed:ro 同样在此后挂载，压在 tmpfs 之上。 + mounts = self.extra_mounts if extra_mounts is None \ + else self._parse_extra_mounts(extra_mounts) + for host_p, cont_p, mode in mounts: + cmd += ["-v", "%s:%s:%s" % (host_p, cont_p, mode)] if self.env_key: cmd += ["-e", self.env_key] merged_env = dict(self.extra_env) diff --git a/cloud/server.py b/cloud/server.py index ed3ec10..29ed1c5 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -1189,6 +1189,10 @@ def migrate_tasks_table(conn): # HA 阶段 1：认领任务的 worker 所在主机名；/pr 据此判断 # worktree 在本机（同步跑）还是别机（排 pr 任务给该主机）。 "worktree_host": "TEXT", + # 沙箱规模：认领任务的 worker_id（worker._claim 在 CAS + # 认领时写）；失联容器识别的 owner 依据，见 + # workers.orphaned_containers。老行 NULL = 无 owner 记录。 + "claimed_by_worker": "TEXT", # 任务级选项（console 选模型/强度/权限用）：NULL = 跟随 # worker 配置默认值；model 不做白名单（中转目录会变， # 下拉只是建议值），effort/approval 由 _create_task 校验。 @@ -1959,6 +1963,7 @@ def _task_to_dict(row, attachments=None, total_tokens=None): "attachments": list(attachments or []), "resume_message": _row_get(row, "resume_message"), "worktree_host": _row_get(row, "worktree_host"), + "claimed_by_worker": _row_get(row, "claimed_by_worker"), "model": _row_get(row, "model"), "effort": _row_get(row, "effort"), "child_model": _row_get(row, "child_model"), @@ -9068,9 +9073,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): executor/capacity/running/host/pid/address。心跳本身就是 注册：worker_id 不存在时自动注册，绝不 404。running 是 worker 上报的当前占用槽数（启动中占位也算）；未给时保留 - 旧值。host/pid/address 只存调用方显式给的值：远程 worker - 的 host/pid 绝不由 server 本机推导（server 的主机名/进程 - 号不是该 worker 的），未给时保留旧值。""" +旧值。host/pid/address 只存调用方显式给的值：远程 worker +的 host/pid 绝不由 server 本机推导（server 的主机名/进程 +号不是该 worker 的），未给时保留旧值；host 须为字符串、 +pid 须为整数、address 须为字符串，否则 400（注册表里的 +寻址信息错了比没有更危险，失联判断会拿错 host/pid 当证据）。""" body, err = self._read_json() if err: self._send_json(400, {"error": err}) @@ -9103,9 +9110,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): return pid = body.get("pid") if pid is not None: - try: - pid = int(pid) - except (TypeError, ValueError): + if isinstance(pid, bool) or not isinstance(pid, int): self._send_json(400, {"error": "pid must be an integer"}) return if pid <= 0: diff --git a/cloud/tests/test_sandbox_scale.py b/cloud/tests/test_sandbox_scale.py new file mode 100644 index 0000000..4d94ced --- /dev/null +++ b/cloud/tests/test_sandbox_scale.py @@ -0,0 +1,435 @@ +"""沙箱规模单测：/tmp 隔离顺序、注册表寻址、失联识别、超时边界。 + +运行（在 ~/workspace/codex-app-server-sandboxscale 下）： + python3 -m unittest discover -s cloud/tests -p "test_sandbox_scale.py" -v + +说明：全部用注入的假 runner / 假 pid 检查，不调用真实 docker、 +不跑真实模型回合、不联网；env 映射里的值只是占位假名， +绝不是真实密钥。任务 id 用 32 位 hex（容器名正则的精确 +形状），不用短 id——短 id 的容器名本来就不在失联识别范围内。 + +覆盖（与本轮改动一一对应）： +- Docker /tmp tmpfs 在所有 -v 挂载之前（父挂载在前，seed 子挂载 + 在后，不被遮掉）；host_bwrap 仍宣告未隔离（tmp_isolation_status）。 +- workers 注册/心跳：host/pid/address 写进库、心跳未给保留旧值、 + 老库（无三列）迁移补列；stale_workers 只列心跳过期行。 +- 失联容器识别：只认精确 codex-<32 位 hex> + 认领到失联 worker + + running/interrupted；基础设施容器、非任务名、已完成任务、 + 无认领、owner 存活的一律排除；pid_verified 只在“本机 host + + 已知 pid + pid 明确已死”时为真。 +- Worker 精确清理：verified_only 默认只杀已验证的精确容器名， + 未验证的只列不杀；显式 verified_only=False 才杀未验证的。 +- 认领写 claimed_by_worker（CAS 语义不变：二次认领仍失败）。 +- 超时边界纯函数 docker_run_timeout：取较小值、None 语义、 + 不限时语义、非法值回落（行为与原内联一致）。 +- web 自适应基线：console.html 含 viewport meta 与小屏 media + 查询（跨视口手工验证矩阵见 cloud/README.md 新增节，不宣称 + 本文件已做浏览器实测）。 +""" + +import http.client +import json +import os +import socket +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +try: + from cloud.auth import AuthStore + from cloud.sandbox import DockerExecutor, tmp_isolation_status + from cloud.server import create_server, init_db, sweep_stale_tasks + from cloud.worker import Worker, docker_run_timeout + from cloud.workers import (default_pid_alive, heartbeat_worker, + list_workers, orphaned_containers, + register_worker, stale_workers) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.sandbox import DockerExecutor, tmp_isolation_status + from cloud.server import create_server, init_db, sweep_stale_tasks + from cloud.worker import Worker, docker_run_timeout + from cloud.workers import (default_pid_alive, heartbeat_worker, + list_workers, orphaned_containers, + register_worker, stale_workers) + +DEAD_TASK = "aa" * 16 +LIVE_TASK = "bb" * 16 +DONE_TASK = "cc" * 16 +UNCLAIMED_TASK = "dd" * 16 +MISSING_TASK = "ee" * 16 + + +def _insert_task(db_path, task_id, status="queued", claimed=None, + heartbeat_ago=None): + now = time.time() + heartbeat = None + if heartbeat_ago is not None: + heartbeat = now - heartbeat_ago + elif status == "running": + heartbeat = now + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, claimed_by_worker)" + " VALUES(?, 'p1', 'do work', ?, NULL, '[]', 0, 'tester'," + " ?, ?, ?, ?)", + (task_id, status, now, now, heartbeat, claimed), + ) + conn.commit() + finally: + conn.close() + + +def _set_worker_old(db_path, worker_id, seconds_ago): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "UPDATE workers SET last_seen = ? WHERE worker_id = ?", + (time.time() - seconds_ago, worker_id)) + conn.commit() + finally: + conn.close() + + +def _seed_orphan_db(db_path, local_host): + """失联场景：dead worker（本机 host、pid 已死）与 live worker + 各认领一个 running 任务，另有已完成、无认领任务各一。""" + init_db(db_path) + register_worker(db_path, "dead-worker", executor="docker", + host=local_host, pid=999999, address="h1:9001") + register_worker(db_path, "live-worker", executor="docker", + host=local_host, pid=123456, address="h1:9002") + _set_worker_old(db_path, "dead-worker", 10000) + _insert_task(db_path, DEAD_TASK, status="running", + claimed="dead-worker", heartbeat_ago=10000) + _insert_task(db_path, LIVE_TASK, status="running", + claimed="live-worker") + _insert_task(db_path, DONE_TASK, status="completed", + claimed="dead-worker") + _insert_task(db_path, UNCLAIMED_TASK, status="running", + claimed=None) + + +class TmpIsolationTestCase(unittest.TestCase): + def test_tmpfs_before_all_volume_mounts(self): + seed_dir = tempfile.mkdtemp() + try: + with mock.patch("cloud.sandbox.shutil.which", + return_value="/usr/bin/docker"): + ex = DockerExecutor( + "/fake/host.py", + runner=lambda cmd, timeout=None: SimpleNamespace( + returncode=0, stdout="{}", stderr=""), + env={"XCM_API_KEY": "placeholder-not-a-real-key"}) + cmd = ex.build_run_command( + {"id": "t1", "worktree": "/tmp/wt"}, + ["echo", "hi"], + extra_mounts=["%s:/tmp/codex-seed:ro" % seed_dir]) + tmpfs_idx = cmd.index("--tmpfs") + self.assertEqual(cmd[tmpfs_idx + 1], "/tmp") + seed_idx = cmd.index("%s:/tmp/codex-seed:ro" % seed_dir) + self.assertLess(tmpfs_idx, seed_idx) + self.assertIn("/var/tmp", cmd) + finally: + os.rmdir(seed_dir) + + def test_host_bwrap_still_declared_not_isolated(self): + status = tmp_isolation_status() + self.assertTrue(status["docker"]) + self.assertFalse(status["host_bwrap"]) + + +class RegistryAddressTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + + def tearDown(self): + self.tmp.cleanup() + + def test_register_and_heartbeat_address_fields(self): + row = register_worker(self.db_path, "w1", executor="docker", + host="host-a", pid=111, address="h:1") + self.assertEqual(row["host"], "host-a") + self.assertEqual(row["pid"], 111) + self.assertEqual(row["address"], "h:1") + # 心跳未给 host/pid/address 时保留旧值，不洗成 None。 + row = heartbeat_worker(self.db_path, "w1", executor="docker") + self.assertEqual(row["host"], "host-a") + self.assertEqual(row["pid"], 111) + self.assertEqual(row["address"], "h:1") + # 心跳显式给新 pid 时刷新（重启后同 worker_id 复用不残留旧 pid）。 + row = heartbeat_worker(self.db_path, "w1", pid=222) + self.assertEqual(row["pid"], 222) + self.assertEqual(row["host"], "host-a") + + def test_old_workers_table_migrates_address_columns(self): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE workers(worker_id TEXT PRIMARY KEY," + " executor TEXT, capacity INTEGER," + " last_seen DOUBLE PRECISION," + " created_at DOUBLE PRECISION)") + conn.commit() + finally: + conn.close() + row = register_worker(self.db_path, "w-old", host="host-b", + pid=333) + self.assertEqual(row["host"], "host-b") + self.assertEqual(row["pid"], 333) + self.assertIsNone(row["address"]) + + def test_stale_workers_only_expired(self): + register_worker(self.db_path, "fresh", host="h") + register_worker(self.db_path, "old", host="h") + _set_worker_old(self.db_path, "old", 10000) + stale = stale_workers(self.db_path, alive_seconds=120) + self.assertEqual([w["worker_id"] for w in stale], ["old"]) + self.assertFalse(stale[0]["alive"]) + live = list_workers(self.db_path, alive_seconds=120) + self.assertTrue([w for w in live + if w["worker_id"] == "fresh"][0]["alive"]) + + def test_default_pid_alive_current_process(self): + self.assertTrue(default_pid_alive(os.getpid())) + self.assertIsNone(default_pid_alive("not-a-pid")) + self.assertIsNone(default_pid_alive(None)) + + +class OrphanIdentifyTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.host = socket.gethostname() + _seed_orphan_db(self.db_path, self.host) + + def tearDown(self): + self.tmp.cleanup() + + def _names(self): + return ["codex-" + DEAD_TASK, "codex-" + LIVE_TASK, + "codex-" + DONE_TASK, "codex-" + UNCLAIMED_TASK, + "codex-" + MISSING_TASK, "codex-postgres", + "codex-not-a-task", "other-" + DEAD_TASK] + + def test_only_dead_owner_exact_container_verified(self): + found = orphaned_containers( + self.db_path, self._names(), alive_seconds=120, + pid_alive=lambda pid: False) + self.assertEqual(len(found), 1) + self.assertEqual(found[0]["container"], "codex-" + DEAD_TASK) + self.assertEqual(found[0]["task_id"], DEAD_TASK) + self.assertEqual(found[0]["worker_id"], "dead-worker") + self.assertTrue(found[0]["pid_verified"]) + + def test_live_pid_not_verified_but_still_listed(self): + found = orphaned_containers( + self.db_path, self._names(), alive_seconds=120, + pid_alive=lambda pid: True) + self.assertEqual(len(found), 1) + self.assertFalse(found[0]["pid_verified"]) + + def test_remote_host_never_pid_verified(self): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE workers SET host = ? WHERE worker_id = ?", + ("some-other-host", "dead-worker")) + conn.commit() + finally: + conn.close() + found = orphaned_containers( + self.db_path, self._names(), alive_seconds=120, + pid_alive=lambda pid: False) + self.assertEqual(len(found), 1) + self.assertFalse(found[0]["pid_verified"]) + + def test_sweep_marks_dead_owner_task_interrupted(self): + swept = sweep_stale_tasks(self.db_path, threshold_seconds=300) + self.assertIn(DEAD_TASK, swept) + self.assertNotIn(LIVE_TASK, swept) + # sweep 后失联容器仍可识别（interrupted 状态也算失联）。 + found = orphaned_containers( + self.db_path, self._names(), alive_seconds=120, + pid_alive=lambda pid: False) + self.assertEqual(len(found), 1) + + +class WorkerOrphanCleanupTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.host = socket.gethostname() + _seed_orphan_db(self.db_path, self.host) + self.killed = [] + + def fake(cmd, timeout=None): + if len(cmd) >= 2 and cmd[1] == "ps": + return SimpleNamespace( + returncode=0, + stdout="\n".join( + ["codex-" + DEAD_TASK, "codex-" + LIVE_TASK, + "codex-postgres"]) + "\n", + stderr="") + if len(cmd) >= 2 and cmd[1] in ("kill", "rm"): + self.killed.append(cmd[-1]) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + self.ex = DockerExecutor( + "/fake/host.py", runner=fake, + env={"XCM_API_KEY": "placeholder-not-a-real-key"}) + self.worker = Worker( + self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "placeholder-not-a-real-key"}, + executor=self.ex, worker_id="cleaner", + worker_address="cleaner-host:9009") + + def tearDown(self): + self.tmp.cleanup() + + def test_verified_only_kills_exact_dead_container(self): + cleaned = self.worker.cleanup_orphaned_containers( + pid_alive=lambda pid: False, alive_seconds=120) + self.assertEqual(cleaned, ["codex-" + DEAD_TASK]) + self.assertIn("codex-" + DEAD_TASK, self.killed) + self.assertNotIn("codex-" + LIVE_TASK, self.killed) + self.assertNotIn("codex-postgres", self.killed) + + def test_unverified_skipped_by_default(self): + cleaned = self.worker.cleanup_orphaned_containers( + pid_alive=lambda pid: True, alive_seconds=120) + self.assertEqual(cleaned, []) + self.assertEqual(self.killed, []) + + def test_explicit_unverified_opt_in_kills(self): + cleaned = self.worker.cleanup_orphaned_containers( + verified_only=False, pid_alive=lambda pid: True, + alive_seconds=120) + self.assertEqual(cleaned, ["codex-" + DEAD_TASK]) + + def test_claim_writes_claimed_by_worker(self): + _insert_task(self.db_path, "f1" * 16, status="queued") + worker = Worker( + self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "placeholder-not-a-real-key"}, + worker_id="claimer-1") + self.assertTrue(worker._claim("f1" * 16)) + task = worker.get_task("f1" * 16) + self.assertEqual(task["claimed_by_worker"], "claimer-1") + # CAS 语义不变：已 running 的任务二次认领仍失败。 + self.assertFalse(worker._claim("f1" * 16)) + + +class TimeoutBoundaryTestCase(unittest.TestCase): + def test_min_wins_and_none_semantics(self): + self.assertEqual(docker_run_timeout(1800, None), 1800) + self.assertEqual(docker_run_timeout(1800, 600), 600) + self.assertEqual(docker_run_timeout(600, 1800), 600) + self.assertIsNone(docker_run_timeout(None, 600)) + self.assertIsNone(docker_run_timeout(None, None)) + + def test_no_time_limit_overrides_both(self): + self.assertIsNone(docker_run_timeout(1800, 600, + no_time_limit=True)) + + def test_invalid_value_falls_back_to_task_timeout(self): + self.assertEqual(docker_run_timeout(1800, "bad"), 1800) + + +class ViewportBaselineTestCase(unittest.TestCase): + def test_console_has_viewport_and_small_screen_query(self): + console = (Path(__file__).resolve().parents[2] + / "cloud" / "console.html") + text = console.read_text(encoding="utf-8") + self.assertIn('name="viewport"', text) + self.assertIn("@media(max-width:", text) + + +class HeartbeatHttpTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS") + os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + if self._old_admins is None: + os.environ.pop("CODEX_GLOBAL_ADMINS", None) + else: + os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, + timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def test_heartbeat_address_fields_roundtrip(self): + status, body = self._req( + "POST", "/workers/heartbeat", + {"worker_id": "hw1", "host": "host-x", "pid": 4242, + "address": "host-x:9001"}, + self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["host"], "host-x") + self.assertEqual(body["pid"], 4242) + self.assertEqual(body["address"], "host-x:9001") + status, body = self._req("GET", "/workers", + key=self.owner_key) + self.assertEqual(status, 200, body) + row = [w for w in body["workers"] + if w["worker_id"] == "hw1"][0] + self.assertEqual(row["host"], "host-x") + self.assertEqual(row["pid"], 4242) + + def test_heartbeat_invalid_pid_400(self): + status, _ = self._req( + "POST", "/workers/heartbeat", + {"worker_id": "hw2", "pid": "not-an-int"}, + self.owner_key) + self.assertEqual(status, 400) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index cb18809..92a446a 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -199,7 +199,8 @@ from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor from cloud.uploads import upload_root as _upload_root from cloud.sandbox import _default_runner as _sandbox_default_runner from cloud.caps import effective_capacity, resolve_capacity -from cloud.workers import heartbeat_worker, register_worker +from cloud.workers import (heartbeat_worker, orphaned_containers, + register_worker) # 与 server.TERMINAL_STATUSES 对齐：这些状态的任务不再被认领或取消。 TERMINAL_STATUSES = frozenset( @@ -232,6 +233,39 @@ def _uuid_hex(): return uuid.uuid4().hex +def docker_run_timeout(task_timeout, docker_timeout, + no_time_limit=False): + """本次 docker run 超时（纯函数，可单测，长任务天花板边界）。 + + 规则（与 _run_attempt 原内联逻辑完全一致，本轮只抽取， + 不改行为）： + - no_time_limit=1（任务不限时）时一律 None：不限的是任务 + 总墙钟与总轮次，不是单轮无限——host 单轮 turn_timeout + 仍是护栏，不在本函数里。 + - task_timeout 为 None 时 None（Docker 路径此时也不再用 + executor.timeout 兜底？注意：原内联只在 task_timeout + 非 None 时算 min，task_timeout 为 None 时 effective 恒 + 为 None——本函数忠实保留该语义，绝不悄悄把 docker_timeout + 扶正为有效超时）。 + - 否则取 min(task_timeout, docker_timeout)（docker_timeout + 为 None 时取 task_timeout）；任一值非法时回落 task_timeout。 + 取消与超时的状态语义不在此：超时 -> failed（不是取消）， + 取消 -> interrupted（执行结果绝不覆盖它），见 _run_attempt + 与 cancel()；恢复靠断点续跑（goal-resume / run 续跑）， + 不是靠把超时任务悄悄重跑一遍。 + """ + if no_time_limit: + return None + if task_timeout is None: + return None + if docker_timeout is None: + return task_timeout + try: + return min(float(task_timeout), float(docker_timeout)) + except (TypeError, ValueError): + return task_timeout + + def _worker_host_name(executor, task): """host 任务名：优先 executor._host_name（与 sandbox 同款 派生），没有 executor 时回落任务 id（uuid hex 本就是安全名）。""" @@ -405,6 +439,9 @@ def _task_from_row(row): "interrupted_reason": _col("interrupted_reason"), "auto_resume_count": _col("auto_resume_count") or 0, "no_time_limit": _col("no_time_limit") or 0, + # 认领 owner：老行缺列时 None（= 认领前或老任务，无 owner + # 可查，失联识别宁可漏不猜）。 + "claimed_by_worker": _col("claimed_by_worker"), # 协调员：老行缺列时 coordinator_parts 为 []（= 普通任务）， # parent_task_id/coordinator_part 缺列时为 None。 "coordinator_parts": _coord_parse_parts(_col("coordinator_parts")), @@ -442,7 +479,8 @@ class Worker: outbox_max_attempts=3, quota_poll_interval_seconds=30, auto_resume_max=50, - ignore_host_budget=True): + ignore_host_budget=True, + worker_address=None): self.db_path = str(db_path) self.host_py_path = host_py_path self.config_path = config_path @@ -638,6 +676,11 @@ class Worker: # 自然变 stale，不要用随机 uuid 让注册表无法对应同一台机器。 self.worker_id = worker_id or "%s-%d" % ( socket.gethostname(), os.getpid()) + # 寻址串：调用方显式给才有（如运维约定的 host:port）， + # 默认 None（不猜、不从环境变量里拼密钥类地址）；注册/ + # 心跳时原样写进 workers.address，供失联排查时定位 + # owner 在哪台主机、哪个入口，绝不是访问凭证。 + self.worker_address = worker_address # 并发上限配置来源（见 cloud/caps.py）：显式构造参数 / # CLI --worker-capacity > 环境变量 CLOUD_WORKER_CAPACITY > # 默认 1；非法值归一为 1。self._env 在上面已定好，单测 @@ -700,7 +743,8 @@ class Worker: self.db_path, self.worker_id, executor=self._executor_name(), capacity=self.capacity, running=self._running_count(), - host=self.host_name, pid=os.getpid()) + host=self.host_name, pid=os.getpid(), + address=self.worker_address) except Exception as exc: log_event(self.logger, "worker_register_failed", message="worker register failed: %s" % exc) @@ -712,13 +756,17 @@ class Worker: 时退化为 socket.gethostname()，容器内即容器名，与 worktree_host 同款 caveat），pid 用本进程号；两者每轮 刷新，供失联判定做本机 + pid 已死的 pid_verified 校验。 - """ + 每次都显式传本机 host 与本进程 pid：worker 重启后 + 同 worker_id 复用时，旧 pid 不会残留在注册表里误导 + 失联判断（旧 pid 残留 + 心跳新鲜 = 活着，尚可；旧 pid + 残留 + 心跳过期 = pid 校验会误判——每次刷新杜绝此例）。 """ try: heartbeat_worker( self.db_path, self.worker_id, executor=self._executor_name(), capacity=self.capacity, running=self._running_count(), - host=self.host_name, pid=os.getpid()) + host=self.host_name, pid=os.getpid(), + address=self.worker_address) except Exception as exc: log_event(self.logger, "worker_registry_heartbeat_failed", message="worker registry heartbeat failed: %s" % exc) @@ -859,31 +907,37 @@ class Worker: conn = _db_connect(self.db_path) try: self._ensure_result_columns(conn) + # 认领同时写 claimed_by_worker：失联容器识别的唯一 + # owner 依据（见 workers.orphaned_containers）。CAS + # 语义不变（仍只在 queued 时改行），认领失败时行不 + # 动、旧认领记录也不被覆盖。 if executor is None: cur = conn.execute( "UPDATE tasks SET status = 'running'," " last_heartbeat = ?, updated_at = ?," - " worktree_host = ?" + " worktree_host = ?, claimed_by_worker = ?" " WHERE id = ? AND status = 'queued'", - (now, now, self.host_name, task_id), + (now, now, self.host_name, self.worker_id, task_id), ) elif self.claim_executor_only: cur = conn.execute( "UPDATE tasks SET status = 'running'," " last_heartbeat = ?, updated_at = ?," - " worktree_host = ?" + " worktree_host = ?, claimed_by_worker = ?" " WHERE id = ? AND status = 'queued'" " AND executor = ?", - (now, now, self.host_name, task_id, executor), + (now, now, self.host_name, self.worker_id, task_id, + executor), ) else: cur = conn.execute( "UPDATE tasks SET status = 'running'," " last_heartbeat = ?, updated_at = ?," - " worktree_host = ?" + " worktree_host = ?, claimed_by_worker = ?" " WHERE id = ? AND status = 'queued'" " AND (executor IS NULL OR executor = ?)", - (now, now, self.host_name, task_id, executor), + (now, now, self.host_name, self.worker_id, task_id, + executor), ) conn.commit() claimed = cur.rowcount == 1 @@ -1431,6 +1485,7 @@ class Worker: "goal_max_rounds": "INTEGER", "goal_mode": "INTEGER", "worktree_host": "TEXT", + "claimed_by_worker": "TEXT", "model": "TEXT", "effort": "TEXT", "child_model": "TEXT", @@ -2382,12 +2437,71 @@ class Worker: pass return + def cleanup_orphaned_containers(self, verified_only=True, + pid_alive=None, alive_seconds=None): + """精确失联清理：只杀 workers.orphaned_containers() 认定的 + 容器（按精确容器名逐个杀，绝不按前缀展开）。 + + verified_only=True（默认）：只杀 pid_verified 的容器—— + 本机 host + 已知 pid 且 pid 明确已死，owner 死亡有本地 + 证据；远端 host / pid 未知 / 未验证的只列不杀（返回里 + 不含它们，日志里记 unverified 数量供运维人工处理）。 + verified_only=False：把未验证的也杀掉——只应在“本机 + 单机部署、注册表 pid 列是老版本缺失”这类调用方已确认 + 单机语义的场景显式开，跨主机场景开它等于按心跳过期 + 猜死，绝不能默认开。 + 本方法只写识别与杀容器，不改任务状态：任务状态由 + sweep_stale_tasks（心跳过期 -> interrupted）负责，两者 + 分工，互不越权。执行器无 list_containers/kill_container + 时返回 []（Local/Cloud 执行器本来就没有容器可清）。 + 返回已清理的精确容器名列表。 + """ + list_fn = getattr(self.executor, "list_containers", None) + kill_fn = getattr(self.executor, "kill_container", None) + if list_fn is None or kill_fn is None: + return [] + try: + names = list_fn() + except Exception as exc: + log_event(self.logger, "worker_orphan_list_failed", + message="orphan container list failed: %s" % exc) + return [] + try: + orphans = orphaned_containers( + self.db_path, names, alive_seconds=alive_seconds, + pid_alive=pid_alive) + except Exception as exc: + log_event(self.logger, "worker_orphan_identify_failed", + message="orphan identify failed: %s" % exc) + return [] + cleaned = [] + unverified = 0 + for item in orphans: + if verified_only and not item.get("pid_verified"): + unverified += 1 + continue + try: + if kill_fn(item["container"]): + cleaned.append(item["container"]) + except Exception: + continue + if cleaned or unverified: + log_event(self.logger, "worker_orphan_cleanup", + message="cleaned orphan containers: %s;" + " unverified skipped: %d" + % (",".join(cleaned), unverified)) + return cleaned + def cleanup_stale_containers(self): """启动清理兜底：清掉上一轮残留的 codex- 容器。 只在执行器有 cleanup_stale_containers 时做（Docker）； 列出或删除失败只记日志，绝不抛异常炸掉 worker。 返回已清理的容器名列表（非 Docker 执行器返回 []）。 + 注意：这是旧的任务正则兜底（删所有 codex-<任务 id> + 容器，不区分 owner），精确失联清理请用 + cleanup_orphaned_containers()；本方法保留是为兼容 + 认领记录缺失的老任务容器，不改其行为。 """ cleanup = getattr(self.executor, "cleanup_stale_containers", None) if cleanup is None: @@ -4522,18 +4636,15 @@ class Worker: # Docker 超时取较小值：两个超时都不应静默延长另一个。 # 注意绝不能改共享的 executor.timeout（并发下两个任务会 # 互相覆盖对方的超时）；超时只作为本次调用的参数传下去。 - effective_docker_timeout = None + # 规则抽到模块级 docker_run_timeout()（纯函数可单测）， + # 这里只接线，行为与原内联完全一致。 is_docker = isinstance(self.executor, DockerExecutor) - if is_docker and _effective_task_timeout is not None: - if self.executor.timeout is None: - effective_docker_timeout = _effective_task_timeout - else: - try: - effective_docker_timeout = min( - float(_effective_task_timeout), - float(self.executor.timeout)) - except (TypeError, ValueError): - effective_docker_timeout = _effective_task_timeout + effective_docker_timeout = ( + docker_run_timeout( + _effective_task_timeout, + self.executor.timeout if is_docker else None, + no_time_limit=_no_time_limit) + if is_docker else None) # P0 安全批后：Docker 同步路径起取消看门狗线程 # （阻塞在 docker run 里时，库状态变 interrupted 就 # 按容器名真杀；非 Docker 路径不启动，避免空线程）。 @@ -6932,6 +7043,10 @@ def main(argv=None): parser.add_argument("--worker-id", default=None, help="多 worker 池身份（默认：主机名-进程号；" "生产部署应配稳定 id，否则重启后旧行变 stale）") + parser.add_argument("--worker-address", default=None, + help="worker 寻址串（可选，显式给才写进注册表" + " workers.address，如运维约定的 host:port；" + "不给则为 NULL，绝不猜、不含任何凭证）") parser.add_argument("--worker-capacity", type=int, default=None, help="worker 并发上限：同一 worker 内最多同时跑" "几个任务（未给时读环境变量" @@ -7065,6 +7180,7 @@ def main(argv=None): repos_dir=args.repos_dir, heartbeat_file=args.heartbeat_file, worker_id=args.worker_id, + worker_address=args.worker_address, capacity=args.worker_capacity, host_name=args.host_name, pr_jobs_only=args.pr_jobs_only, diff --git a/cloud/workers.py b/cloud/workers.py index 58c2b3f..dc60836 100644 --- a/cloud/workers.py +++ b/cloud/workers.py @@ -15,6 +15,21 @@ 存在就 UPDATE，绝不因“未注册”返回 404 报错。 - alive 定义：last_seen 在阈值内（默认 120 秒，可经环境变量 CLOUD_WORKER_ALIVE_SECONDS 调，测试可传参覆盖）。 +- 寻址信息（本轮新增）：host（worker 所在主机名，worker 的 + --host-name，容器内 worker 绝不能用容器名冒充宿主名）、 + pid（worker 进程号，owner 死活的本地验证依据）、address + （可选寻址串，如运维侧约定的 host:port，调用方显式给才 + 有，绝不从环境变量里猜密钥类地址）。老行三列为 NULL， + 按“未知”处理，绝不猜成活着或死了。 +- 失联识别（本轮新增）：stale_workers() 列心跳过期行； + orphaned_containers() 把“容器名 codex-<任务 id>”精确映射 + 到 tasks.claimed_by_worker（认领时 worker 写），只认领到 + 失联 worker 且状态仍为 running/interrupted 的任务容器才 + 算失联容器；名字不匹配任务 id 正则、任务行缺失、无认领 + 记录、owner 仍存活的一律不算——绝不按 "codex-" 前缀删。 + pid_verified 另算：本机 host + 已知 pid 时用 pid_alive + 校验（默认 os.kill(pid, 0)），远端 host 或 pid 未知时 + 为 False（列出来给运维看，不自动杀）。 - 本模块不存、不打任何密钥：注册信息只有 worker_id、executor、 capacity、last_seen，没有任何凭证字段。 - host/pid/address（本轨新增）：新库建表直接含三列；老库由 @@ -46,6 +61,7 @@ """ import os +import re import signal import socket import time @@ -78,6 +94,54 @@ def alive_threshold_seconds(override=None): return float(DEFAULT_ALIVE_SECONDS) +def _existing_worker_columns(conn): + """workers 表现有列名集合；查不到时返回 None（调用方跳过补列）。 + + PG 走 information_schema，SQLite 走 PRAGMA table_info；两种 + 查询失败都只回滚、不抛异常（老库迁移兜底，绝不因查列炸掉 + 注册/心跳主路径）。 + """ + try: + rows = conn.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_name = 'workers'").fetchall() + return {r[0] for r in rows} + except Exception: + try: + conn.rollback() + except Exception: + pass + try: + rows = conn.execute("PRAGMA table_info(workers)").fetchall() + return {r[1] for r in rows} + except Exception: + try: + conn.rollback() + except Exception: + pass + return None + + +def _ensure_worker_address_columns(conn): + """老库补 host/pid/address 三列（缺哪个补哪个，幂等）。""" + cols = _existing_worker_columns(conn) + if cols is None: + return + for name, coltype in (("host", "TEXT"), ("pid", "INTEGER"), + ("address", "TEXT")): + if name in cols: + continue + try: + conn.execute( + "ALTER TABLE workers ADD COLUMN %s %s" % (name, coltype)) + conn.commit() + except Exception: + try: + conn.rollback() + except Exception: + pass + + def ensure_workers_table(db_path): """建 workers 表（不存在时建，已存在时不动，老库补精度）。 @@ -107,6 +171,7 @@ def ensure_workers_table(db_path): "created_at DOUBLE PRECISION)" ) conn.commit() + _ensure_worker_address_columns(conn) _ensure_running_column(conn) _ensure_hostpid_columns(conn) # 老库升精度只在确实是 Postgres REAL 列时做一次：每次心跳都 @@ -501,6 +566,21 @@ def reap_stale_workers(db_path, stale_seconds=None, local_host=None, return results +def _coerce_pid(pid): + """pid 归一化：None/空串 -> None；可转 int 的转 int；否则 None。 + + 绝不把非法 pid 写进库（写进去了后续的本地存活校验会拿 + 它当进程号查，错 pid 比缺 pid 更危险）。 + """ + if pid is None or pid == "": + return None + if isinstance(pid, bool): + return None + try: + return int(pid) + except (TypeError, ValueError): + return None + def register_worker(db_path, worker_id, executor=None, capacity=1, running=None, host=None, pid=None, address=None): """注册（或刷新）一个 worker：不存在 INSERT，存在 UPDATE。 @@ -509,8 +589,7 @@ def register_worker(db_path, worker_id, executor=None, capacity=1, （server 的 heartbeat 端点）据此返回 400。新列语义：新行 按给定值写（未给为 NULL）；已存在行上给定值覆盖、未给 保留旧值——绝不因一次不带 host/pid 的注册把已有的身份 - 洗成 NULL，也不编造 host/pid/address。 - """ + 洗成 NULL，也不编造 host/pid/address。 """ if not worker_id or not isinstance(worker_id, str): raise ValueError("worker_id required") pid = _normalize_pid(pid) @@ -541,8 +620,7 @@ def register_worker(db_path, worker_id, executor=None, capacity=1, " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)", (worker_id, executor, capacity, running if running is not None else 0, - host, pid, address, now, now), - ) + host, pid, address, now, now), ) conn.commit() row = conn.execute( "SELECT * FROM workers WHERE worker_id = ?", @@ -562,6 +640,7 @@ def heartbeat_worker(db_path, worker_id, executor=None, capacity=None, 应传本机 host 与本进程 pid，见 worker._registry_heartbeat）， 未给则保留旧值；address 没有新来源，绝不由本函数推导 （例如不从 worker_id 或 socket 猜地址），只存显式给的值。 +显式给 None 与未给在心跳里无法区分（均按未给保留处理）。 """ if not worker_id or not isinstance(worker_id, str): raise ValueError("worker_id required") @@ -580,8 +659,7 @@ def heartbeat_worker(db_path, worker_id, executor=None, capacity=None, " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)", (worker_id, executor, capacity if capacity is not None else 1, running if running is not None else 0, - host, pid, address, now, now), - ) + host, pid, address, now, now), ) else: old = _worker_to_dict(row) conn.execute( @@ -594,8 +672,7 @@ def heartbeat_worker(db_path, worker_id, executor=None, capacity=None, host if host is not None else old["host"], pid if pid is not None else old["pid"], address if address is not None else old["address"], - now, worker_id), - ) + now, worker_id), ) conn.commit() row = conn.execute( "SELECT * FROM workers WHERE worker_id = ?", @@ -643,4 +720,133 @@ def _worker_to_dict(row): "address": _get("address"), "last_seen": _get("last_seen"), "created_at": _get("created_at"), + "host": _get("host"), + "pid": _get("pid"), + "address": _get("address"), } + + +def stale_workers(db_path, alive_seconds=None): + """心跳过期（失联）的 worker 行列表（每行已含 alive=False）。 + + 只按 last_seen 判，不查 pid：pid 校验是 orphaned_containers + 里对“能不能自动杀”的加严，不是失联本身的定义（失联的 + 定义与 sweep_stale_tasks 的任务心跳过期同款语义）。 + """ + return [w for w in list_workers(db_path, alive_seconds) + if not w.get("alive")] + + +def default_pid_alive(pid): + """本地进程存活检查：os.kill(pid, 0) 成功或无权限都算活着 + （PermissionError 说明进程存在，只是归属别人）；只有 + 查无此进程算死了；其它异常算未知（返回 None，不猜）。 + pid 非法时返回 None。""" + pid = _coerce_pid(pid) + if pid is None: + return None + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + except Exception: + return None + + +def _is_local_host(host): + """host 是否本机：空值按“无法确认”处理，不按本机猜。 + + 注意与 pid_verified 的关系：host 为空时即使 pid 给了也 + 不能断定该 pid 是本机的——跨主机同 pid 号毫无意义， + 所以空 host 一律 unverified。 + """ + if not host or not isinstance(host, str): + return False + try: + return host == socket.gethostname() + except Exception: + return False + + +_TASK_CONTAINER_RE = re.compile(r"^codex-([0-9a-f]{32})$") + + +def orphaned_containers(db_path, container_names, + alive_seconds=None, pid_alive=None): + """识别失联 owner 的任务容器（纯识别，不杀任何容器）。 + + 返回 list，每项 {container, task_id, worker_id, pid_verified}： + - container：精确的容器名（原样返回，调用方按此名杀， + 绝不按前缀展开）； + - task_id：从容器名正则里精确抠出的 32 位 hex（不是前缀 + 切出来的任意串）； + - worker_id：tasks.claimed_by_worker（认领时写）； + - pid_verified：True = 本机 host + 已知 pid 且 pid_alive + 明确报死（可自动杀）；False = 未验证（远端 host、pid + 未知、pid_alive 未给或返回未知），只列不自动杀。 + 排除项（任一即排除）：容器名不匹配任务正则；tasks 表或 + claimed_by_worker 列不存在（无法映射时宁可漏，不猜）； + 任务行缺失；任务状态不在 running/interrupted；owner 行 + 缺失或仍存活（alive）。调用方杀之前仍须按返回的精确 + 容器名逐个杀，绝不把本函数的返回再按前缀展开。 + """ + if pid_alive is None: + pid_alive = default_pid_alive + stale = {w["worker_id"]: w for w in stale_workers( + db_path, alive_seconds)} + if not stale: + return [] + conn = _db_connect(str(db_path)) + try: + try: + rows = conn.execute( + "SELECT id, status, claimed_by_worker FROM tasks" + ).fetchall() + except Exception: + try: + conn.rollback() + except Exception: + pass + return [] + finally: + conn.close() + tasks = {} + for row in rows: + try: + tasks[row["id"]] = (row["status"], row["claimed_by_worker"]) + except (KeyError, IndexError, TypeError): + continue + out = [] + for name in container_names or []: + if not isinstance(name, str): + continue + m = _TASK_CONTAINER_RE.match(name) + if not m: + continue + task_id = m.group(1) + info = tasks.get(task_id) + if info is None: + continue + status, claimed = info + if status not in ("running", "interrupted"): + continue + if not claimed or claimed not in stale: + continue + worker = stale[claimed] + verified = False + pid = worker.get("pid") + if _is_local_host(worker.get("host")) and pid is not None: + try: + verified = pid_alive(pid) is False + except Exception: + verified = False + out.append({ + "container": name, + "task_id": task_id, + "worker_id": claimed, + "pid_verified": verified, + }) + return out