P1 security batch: quota execution checks, cost accounting, goal/coordinator semantics, PR channel, orphan signature, control-plane and console hardening

- Docker quota watchdog + fail-closed quota checks, system tasks need real owner
- max_cost_usd actually accumulated, cancel records cost, estimated rows excluded from unpriced cap, cache_write declared non-billed
- goal-resume keeps original verify on breakpoint resume, goal retry falls back to resume, goal_max_rounds capped at 100 at create and execution
- coordinator children inherit parent model/effort, orphans marked interrupted with audit
- PR jobs on docker workers return explicit needs_local_worker instead of silent empty
- worker code signature covers 6 files and is checked before claim, fail-closed
- sub2api lazy provisioning with rotation, backup whitelist + 0600, push SSRF block, repos/workers desensitized, key/project audits and limits, console buttons/quota/status wording, coordinator/automation quota reserve

cloud/worker.py

@@ -139,6 +139,7 @@ from cloud.coordinator import DEFAULT_POLL_INTERVAL as _COORD_POLL_INTERVAL
 from cloud.coordinator import DEFAULT_TIMEOUT_SECONDS as _COORD_TIMEOUT
 from cloud.coordinator import ensure_columns as _coord_ensure_columns
 from cloud.coordinator import parse_parts as _coord_parse_parts
+from cloud.coordinator import nonterminal_children as _coord_nonterminal
 from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \
     sum_usage, unpriced_token_cap
 from cloud.db import connect as _db_connect
@@ -170,19 +171,68 @@ DEFAULT_ENV_KEY = "XCM_API_KEY"
 # gpt-6.1-sol 路由当日实测持续 503，terra 同日实测 1 秒完成；
 # 任务未指定模型时 worker 与 host.py 都落到这里。
 DEFAULT_MODEL = "gpt-5.6-terra"
+# P1 执行面：goal 轮次上限（有限轮次形态）。goal_mode=1 的 0 轮
+# （模型宣告才结束）不在此限——它的上界是配额（配额看门狗与
+# 认领/边界门），不是轮次；旧验收形态的 goal_max_rounds 必须
+# 在 1..GOAL_MAX_ROUNDS_CAP 之间，超限的任务 worker 在认领后
+# 直接 failed（建任务时的 400 校验在 server.py，属另一路文件，
+# 本路只做执行侧防线，见 run_one 的 goal 上限门）。
+GOAL_MAX_ROUNDS_CAP = 100
+# P1 执行面：无配额行时的默认配额（微美元），与 server.py
+# _default_quota_usd_micros() 同值（$10/月，环境变量
+# CLOUD_DEFAULT_QUOTA_USD_MICROS 可覆盖）。P0 前“无行 =
+# 不限额”已废：worker 侧查不到行时按此默认额度算，显式
+# NULL 行才是不限额（与服务端 _quota_status 同口径）。
 # host.py dispatch 的原话是 "task 已存在: <name>，先换名或 discard 它"
 #（英文 task + 已存在），取共有子串做判断，不依赖标点。
 _ALREADY_EXISTS_MARKER = "已存在"
 
 
-def _code_signature():
-    """本文件签名（mtime_ns, size）：部署 staging 换目录后必变；
-    取不到时返回 None（此时自退检查跳过，绝不误杀）。"""
+def _default_quota_usd_micros():
+    """无配额行时的默认配额（微美元），与 server 侧同值。
+
+    server.py 的 _default_quota_usd_micros() 在另一路文件里，
+    worker 不跨模块 import 它（避免循环 import），这里复刻
+    同款解析：环境变量 CLOUD_DEFAULT_QUOTA_USD_MICROS，非法
+    值回落 10_000_000。"""
     try:
-        st = os.stat(os.path.abspath(__file__))
-        return (st.st_mtime_ns, st.st_size)
+        return max(0, int(os.environ.get(
+            "CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
+    except (TypeError, ValueError):
+        return 10000000
+
+
+def _code_signature(host_py_path=None):
+    """多文件签名：worker/sandbox/coordinator/cost/server/host
+    各取 (mtime_ns, size) 组成的元组；任一文件取不到即 None。
+
+    P1 执行面扩面：旧版只签 worker.py，部署只换了 sandbox 或
+    host.py 时旧 worker 会继续用旧代码抢任务。host.py 取
+    host_py_path（真实文件时用它，否则用 cloud/ 上一级的
+    host.py）；单测常传 /fake/host.py 这种不存在的路径，此时
+    回落到仓库 host.py，不因此判签名缺失。取不到时返回 None，
+    调用方必须 fail-closed（不认领、不执行），绝不按“没变”
+    放行——旧版“取不到当没变”是 fail-open，已废。"""
+    here = os.path.dirname(os.path.abspath(__file__))
+    candidates = [
+        os.path.join(here, "worker.py"),
+        os.path.join(here, "sandbox.py"),
+        os.path.join(here, "coordinator.py"),
+        os.path.join(here, "cost.py"),
+        os.path.join(here, "server.py"),
+    ]
+    if host_py_path and os.path.isfile(host_py_path):
+        candidates.append(os.path.abspath(host_py_path))
+    else:
+        candidates.append(os.path.join(os.path.dirname(here), "host.py"))
+    sig = []
+    try:
+        for path in candidates:
+            st = os.stat(path)
+            sig.append((st.st_mtime_ns, st.st_size))
     except OSError:
         return None
+    return tuple(sig)
 
 
 def _missing_key_error(env_key):
@@ -447,16 +497,46 @@ class Worker:
         # 注入的普通 dict 同样可读环境变量（无该键时默认 1）。
         self.capacity = resolve_capacity(capacity, env=self._env)
         # 常驻启动时的代码签名：run_loop 每轮边界比对，变了就自退
-        # （见 run_loop 的代码变化自退检查）。
-        self._code_signature = _code_signature()
+        # （见 run_loop 的代码变化自退检查）；_claim 每次认领前
+        # 也比对（见 _code_signature_valid），不要只在轮次边界查。
+        self._code_signature = _code_signature(host_py_path)
+        # 配额看门狗命中过的任务（Docker 同步路径用）：看门狗杀
+        # 容器后，同步返回的 failed 要被 _run_attempt 改写成
+        # quota_exceeded，否则 run_one 会把它当普通失败重试，
+        # 配额中断就续跑下去了。受 _quota_hit_lock 保护。
+        self._quota_hit_tasks = set()
+        self._quota_hit_lock = threading.Lock()
 
     def _code_changed(self):
-        """启动后 worker.py 是否被换过（部署换目录/手动改代码）：
-        签名取不到时一律当没变，不误杀。"""
-        now = _code_signature()
+        """启动后关键代码是否被换过（部署换目录/手动改代码）：
+        签名取不到时返回 False（是否“变了”无从判断），但这
+        不等于放行——放行与否看 _code_signature_valid。"""
+        now = _code_signature(self.host_py_path)
         return (now is not None and self._code_signature is not None
                 and now != self._code_signature)
 
+    def _code_signature_valid(self):
+        """签名是否可信且未变：启动签名与现签名都必须取得到
+        且相等，否则 False。False 时 _claim 拒绝认领、run_loop
+        自退（75）、_run_pr_jobs 不碰任务——fail-closed，绝不
+        让签名缺失的 worker 抢任务（旧版取不到当没变是 fail-open）。"""
+        now = _code_signature(self.host_py_path)
+        return (now is not None and self._code_signature is not None
+                and now == self._code_signature)
+
+    def _mark_quota_hit(self, task_id):
+        with self._quota_hit_lock:
+            self._quota_hit_tasks.add(task_id)
+
+    def _take_quota_hit(self, task_id):
+        """取走一次配额看门狗命中（命中即消费，避免旧命中污染
+        同一任务的下一轮）：返回 True 表示本轮确被看门狗杀过。"""
+        with self._quota_hit_lock:
+            if task_id in self._quota_hit_tasks:
+                self._quota_hit_tasks.discard(task_id)
+                return True
+            return False
+
     def register(self):
         """向控制面注册本 worker（写 workers 表，失败只记日志）。
 
@@ -573,7 +653,17 @@ class Worker:
         里清（本地 _execute_via_process / Docker 同步路径），
         认领后提前返回的三道门（缺 key / 仓库 / 配额）在
         run_one 里显式清占位；CAS 语义本身一句未改。
+        P1 签名门：占位之前先比对多文件签名，签名缺失或已变
+        时直接返回 False（fail-closed，不抢任务、不占槽），
+        并记日志；不要只在 run_loop 轮次边界查——并行池线程
+        直接调 run_one/_claim，轮次边界查不到它们。
         """
+        if not self._code_signature_valid():
+            log_event(self.logger, "worker_claim_refused_code_signature",
+                      task_id=task_id,
+                      message="code signature missing or changed,"
+                      " refusing claim")
+            return False
         with self._running_lock:
             already_running = task_id in self._running
             if not already_running and \
@@ -865,14 +955,51 @@ class Worker:
         _check_unpriced_token_cap 同口径）：USD 配额不计量无价
         模型，默认模型 terra 无价，若无此上限则配额对默认模型
         永远不生效；有配额行的用户当月无价 token（input +
-        output，已记账 + 本轮）达到上限即停。无配额行 = 不限额。
-        查询本身异常时按不限额处理（fail-open 并记一条警告）：
-        配额是第二道网，绝不能因查询抖动把全部任务挡在门外；
-        服务端建任务/分支/重试时的准入仍是第一道。
+        output，已记账 + 本轮）达到上限即停。无价上限只计
+        source='actual' 的行：估算行（estimated）不是真实花费，
+        计入上限会把“没花钱”算成“花超了”，且与 cost.py
+        _report 的 unpriced_actual_tokens 口径一致。
+        P1 执行面三处收口（旧版三处都是 fail-open，已废）：
+        - 属主：任务 created_by 为空时先查父任务（协调员子任务
+          行可能无 created_by，属主跟父任务走）；仍查不到时
+          拒绝新回合，不再返回 None 放行。created_by 字面为
+          "system" 的系统任务按普通属主算（无行时同样走默认
+          额度，不另设无限额的系统预算）。
+        - 无配额行：不再是不限额，按默认配额
+          _default_quota_usd_micros()（$10/月）算；只有显式
+          NULL 行才是不限额（与 P0 的服务端口径一致，管理员
+          显式设 NULL 的行仍被尊重）。
+        - 查库失败：返回 "quota_check_failed: ..."（fail-closed），
+          绝不返回 None 放行。调用方分流：新回合边界（认领门、
+          重试/fallback/后续消息边界）见此串即拒绝新回合并记
+          worker_quota_stop 审计；执行中轮询（popen 轮询、
+          Docker 配额看门狗）见此串只记警告、不中途杀当前轮
+          ——查询抖动不应杀掉已在跑的回合，但下一轮边界仍会
+          拒绝（见 run_one 的 _quota_hit 过滤）。查库失败的
+          审计由边界调用方记，这里只记日志，避免轮询高频写审计。
         """
         owner = (task or {}).get("created_by")
+        if not owner and (task or {}).get("parent_task_id"):
+            # 协调员子任务属主跟父任务走（与 _gateway_context
+            # 同款回查）；查不到时不猜，直接走下面的拒绝分支。
+            try:
+                conn = _db_connect(self.db_path)
+                try:
+                    row = conn.execute(
+                        "SELECT created_by FROM tasks WHERE id = ?",
+                        (task["parent_task_id"],)).fetchone()
+                finally:
+                    conn.close()
+                if row is not None:
+                    try:
+                        owner = row[0]
+                    except (TypeError, IndexError):
+                        owner = None
+            except Exception:
+                owner = None
         if not owner:
-            return None
+            return ("quota_exceeded: 任务缺少属主，无法查配额，"
+                    "拒绝新回合（请归属真实用户后再排队）")
         try:
             month = datetime.now(timezone.utc).strftime("%Y-%m")
             conn = _db_connect(self.db_path)
@@ -887,19 +1014,33 @@ class Worker:
                     " WHERE user_id = ?",
                     (owner,)).fetchone()
                 if row is None:
-                    return None
-                quota = row[0]
+                    # P1：无行按默认额度，不是不限额（P0 服务端
+                    # 已在建号/登录时补行，worker 侧仍可能遇到
+                    # 老库未补行的用户，这里按默认额度兜底，
+                    # 不写行——写行是服务端的职责，worker 不抢）。
+                    quota = _default_quota_usd_micros()
+                else:
+                    quota = row[0]
                 if quota is None:
                     return None
                 # cost_records 由 CostStore 懒建：全新库上它还不存
                 # 在，直接 JOIN 会报错——先实例化一次建表再查（与
                 # 服务端 _quota_status 同款做法），否则 quota=0
                 #（一分钱也不能花）在全新库上会被误放行。
+                # P1 补：建表必须落在本连接上，不只 CostStore 的
+                # 连接上——文件库两者指向同一文件所以旧代码碰巧
+                # 能用，:memory: 库两者是不同空库，本连接查 JOIN
+                # 会直接报错并（P1 fail-closed 后）误拒全部任务。
                 _cost_store = CostStore(self.db_path)
                 try:
                     _cost_store._conn.close()
                 except Exception:
                     pass
+                try:
+                    conn.execute(CostStore._COST_DDL)
+                    conn.commit()
+                except Exception:
+                    pass
                 used_row = conn.execute(
                     "SELECT COALESCE(SUM(c.usd_micros), 0)"
                     " FROM cost_records c JOIN tasks t"
@@ -912,7 +1053,7 @@ class Worker:
                     " FROM cost_records c JOIN tasks t"
                     " ON c.task_id = t.id"
                     " WHERE t.created_by = ? AND c.day LIKE ?"
-                    " AND c.usd_micros IS NULL",
+                    " AND c.usd_micros IS NULL AND c.source = 'actual'",
                     (owner, month + "%")).fetchone()
                 unpriced_used = unpriced_row[0] if unpriced_row else 0
             finally:
@@ -940,11 +1081,15 @@ class Worker:
                         % (unpriced_used, cap, month))
             return None
         except Exception as exc:
+            # P1 fail-closed：查库失败不再返回 None。调用方按
+            # 前缀分流（见本函数 docstring）：quota_check_failed
+            # 在新回合边界 = 拒绝，在执行中轮询 = 只警告不杀轮。
             try:
                 self.logger.warning("quota check failed: %s", exc)
             except Exception:
                 pass
-            return None
+            return ("quota_check_failed: 配额查询失败，拒绝新回合"
+                    "（查库异常，非配额用完）")
 
     def _ready_tasks(self):
         """当前就绪任务列表（pick_next 的批量版）。
@@ -1123,6 +1268,41 @@ class Worker:
                 pass
             return
 
+    def _docker_quota_watchdog(self, task_id, container_name,
+                               stop_event, quota_check, interval=30.0):
+        """Docker 同步路径的配额看门狗（P1，复用取消看门狗模式）。
+
+        同步路径阻塞在 docker run 里，run_one 的边界检查要等
+        本轮返回才跑得到——配额在别处被花完时，本轮会继续烧
+        到自然结束。本线程每 interval 秒（默认 30 秒，不超过
+        30 秒）用 quota_check 查一次已记账配额：命中时按容器
+        名真杀容器并记 _quota_hit_tasks，_run_attempt 在同步
+        返回后把结果改写成 quota_exceeded，run_one 走统一的
+        配额中断收尾（interrupted + 配额原因 + 已花成本照记，
+        下一轮不再续跑）。查库失败时 quota_check 返回 False
+        （run_one 的 _quota_hit 会把 quota_check_failed 前缀
+        过滤掉，见 _quota_exceeded docstring），本线程只继续
+        轮询、不杀当前轮；下一轮边界仍会拒绝。诚实边界：看
+        门狗只能看到已记账用量 + 往轮累计用量，本轮容器内尚
+        未回传的 token 它看不到——那部分在结果返回后由边界
+        门拦住，不会悄悄续跑下一轮。
+        """
+        while not stop_event.wait(interval):
+            try:
+                over = bool(quota_check()) if quota_check else False
+            except Exception:
+                over = False
+            if not over:
+                continue
+            self._mark_quota_hit(task_id)
+            try:
+                kill = getattr(self.executor, "kill_container", None)
+                if kill is not None and container_name:
+                    kill(container_name)
+            except Exception:
+                pass
+            return
+
     def cleanup_stale_containers(self):
         """启动清理兜底：清掉上一轮残留的 codex- 容器。
 
@@ -2069,6 +2249,7 @@ class Worker:
         # 按容器名真杀；非 Docker 路径不启动，避免空线程）。
         _watchdog_stop = None
         _watchdog_thread = None
+        _quota_watchdog_thread = None
         if is_docker and hasattr(self.executor, "kill_container"):
             try:
                 _container_name = self.executor.container_name(task)
@@ -2081,6 +2262,17 @@ class Worker:
                     args=(task_id, _container_name, _watchdog_stop),
                     daemon=True)
                 _watchdog_thread.start()
+                # P1 配额看门狗：只在同步 Docker 路径起（进程
+                # 模式的 _execute_via_process 已每 30 秒轮询
+                # quota_check，不重复起）。quota_check 为 None
+                # 时不起（无配额检查可做，不空转线程）。
+                if quota_check is not None and not self._use_popen():
+                    _quota_watchdog_thread = threading.Thread(
+                        target=self._docker_quota_watchdog,
+                        args=(task_id, _container_name,
+                              _watchdog_stop, quota_check),
+                        daemon=True)
+                    _quota_watchdog_thread.start()
         try:
             gateway_provider, gateway_env = self._ensure_gateway(task)
             if self._use_popen():
@@ -2123,8 +2315,14 @@ class Worker:
                                 prep_err)
                     if goal_mode:
                         if task.get("_resume"):
+                            # P1 断点续跑传 None（不传 --goal-text，
+                            # host 沿用原 goal+verify）；开新目标
+                            # 才传本轮 prompt 当新 goal_text。
+                            _resume_text = None if task.get(
+                                "_resume_kind") == "breakpoint" \
+                                else task["prompt"]
                             result = self.executor.goal_resume_task(
-                                task, task["prompt"],
+                                task, _resume_text,
                                 timeout=effective_docker_timeout,
                                 extra_mounts=tenant_mounts,
                                 network=tenant_network,
@@ -2165,8 +2363,11 @@ class Worker:
 
                         if task.get("_resume") and isinstance(
                                 self.executor, LocalExecutor):
+                            _resume_text = None if task.get(
+                                "_resume_kind") == "breakpoint" \
+                                else task["prompt"]
                             result = self.executor.goal_resume_task(
-                                task, task["prompt"],
+                                task, _resume_text,
                                 extra_env=gateway_env,
                                 model_provider=gateway_provider)
                         elif isinstance(self.executor, LocalExecutor):
@@ -2220,6 +2421,8 @@ class Worker:
                 _watchdog_stop.set()
                 if _watchdog_thread is not None:
                     _watchdog_thread.join(timeout=2)
+                if _quota_watchdog_thread is not None:
+                    _quota_watchdog_thread.join(timeout=2)
             # 同步路径执行结束：清掉取消用的容器标记（取消路径里
             # 若已清过，这里再清一次无害）。进程模式的 proc 由
             # _execute_via_process 自己的 finally 清，这里只清
@@ -2236,6 +2439,18 @@ class Worker:
         if result is not None and result.get("status") == "failed" and \
                 "docker run timeout" in (result.get("stdout", "") or ""):
             result = dict(result, status="timeout")
+        # P1 配额看门狗命中改写：同步路径被看门狗杀容器后，
+        # 执行器只返回普通 failed，run_one 无法区分“模型失败”
+        # 与“配额杀容器”——命中时改写成 quota_exceeded，run_one
+        # 走统一配额中断收尾（interrupted + 配额原因 + 成本照记，
+        # 不重试、不续跑下一轮）。改写前先取走命中标记，避免
+        # 旧命中污染同一任务的后续尝试；超时标记优先于配额
+        # 改写的情况不存在（看门狗杀容器时执行器不会同时报超时）。
+        if result is not None and self._take_quota_hit(task_id):
+            result = dict(result, status="quota_exceeded")
+            if error is None:
+                error = ("quota_exceeded: Docker 配额看门狗已杀容器"
+                         "（配额中断，下一轮不再续跑）")
         return result, error
 
     def _attempt_error(self, result, error, token):
@@ -2384,7 +2599,12 @@ class Worker:
                      task.get("priority", 0) or 0, task["created_by"],
                      now, now,
                      part.get("executor") or task.get("executor"),
-                     part.get("model"), part.get("effort"),
+                     # P1：分工未指定模型/强度时继承父任务（旧版
+                     # 直接写 part 的 None，子任务会落到 worker
+                     # 默认模型/强度，与父任务的显式选择脱节，
+                     # 既改成本也改行为，且排查时看不出继承关系）。
+                     part.get("model") or task.get("model"),
+                     part.get("effort") or task.get("effort"),
                      task.get("environment_id"), task["id"], name))
                 self._audit(
                     task["project_id"], "coordinator_child_created",
@@ -2422,6 +2642,109 @@ class Worker:
             conn.close()
         return rows
 
+    def _coord_mark_orphans(self, task, child_ids):
+        """把仍未终态的子任务标 interrupted 并记审计（P1 孤儿收口）。
+
+        只对 CAS 改到行的子任务记审计（rowcount==1）：已被别的
+        worker 终态化的子任务不动，避免覆盖它的结果。返回被标
+        记的子任务 id 列表。正在跑的子任务被标 interrupted 后，
+        跑它的 worker 会在轮询里看到库状态并杀进程/容器（popen
+        每 1 秒查库、Docker 取消看门狗每 1 秒查库），不会无声
+        继续烧钱；若那个 worker 也已死，子任务至少已是终态，
+        不会再被认领烧钱。
+        """
+        marked = []
+        rows = self._coord_child_rows(child_ids)
+        pending = _coord_nonterminal([
+            {"id": cid, "status": (rows.get(cid) or {}).get("status")}
+            for cid in child_ids])
+        if not pending:
+            return marked
+        conn = _db_connect(self.db_path)
+        try:
+            for ch in pending:
+                cur = conn.execute(
+                    "UPDATE tasks SET status = 'interrupted',"
+                    " updated_at = ? WHERE id = ? AND status NOT IN"
+                    " ('completed', 'failed', 'interrupted', 'merged',"
+                    " 'merge-failed')",
+                    (time.time(), ch["id"]))
+                if cur.rowcount == 1:
+                    marked.append(ch["id"])
+                    self._audit(
+                        task["project_id"], "coordinator_child_orphaned",
+                        "%s -> %s (parent %s)"
+                        % (ch["id"], "interrupted", task["id"]))
+            conn.commit()
+        finally:
+            conn.close()
+        return marked
+
+    def sweep_orphaned_coordinator_children(self):
+        """常驻循环每轮的孤儿扫尾（P1）：父已终态或父行失踪、子
+        仍未终态时，把子标 interrupted 并记审计。
+
+        父仍在 running/queued 时绝不动子（协调员等齐期间子任务
+        未终态是正常状态）。父行失踪（被删）同样算孤儿：没有
+        父会再来合流，子任务继续跑只是烧钱。返回被标记的
+        (parent_id, child_id) 对列表；查库失败时返回 [] 且只记
+        日志，不炸循环。
+        """
+        marked = []
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                rows = conn.execute(
+                    "SELECT id, parent_task_id, project_id FROM tasks"
+                    " WHERE parent_task_id IS NOT NULL AND status NOT IN"
+                    " ('completed', 'failed', 'interrupted', 'merged',"
+                    " 'merge-failed')").fetchall()
+                parents = {}
+                for row in rows:
+                    try:
+                        pid = row["parent_task_id"]
+                    except (KeyError, IndexError, TypeError):
+                        continue
+                    if pid not in parents:
+                        prow = conn.execute(
+                            "SELECT status FROM tasks WHERE id = ?",
+                            (pid,)).fetchone()
+                        try:
+                            parents[pid] = prow["status"] \
+                                if prow is not None else None
+                        except (KeyError, IndexError, TypeError):
+                            parents[pid] = None
+                for row in rows:
+                    try:
+                        cid = row["id"]
+                        pid = row["parent_task_id"]
+                        proj = row["project_id"]
+                    except (KeyError, IndexError, TypeError):
+                        continue
+                    pstatus = parents.get(pid)
+                    if pstatus is not None and \
+                            pstatus not in TERMINAL_STATUSES:
+                        continue
+                    cur = conn.execute(
+                        "UPDATE tasks SET status = 'interrupted',"
+                        " updated_at = ? WHERE id = ? AND status NOT IN"
+                        " ('completed', 'failed', 'interrupted', 'merged',"
+                        " 'merge-failed')",
+                        (time.time(), cid))
+                    if cur.rowcount == 1:
+                        marked.append((pid, cid))
+                        self._audit(
+                            proj, "coordinator_child_orphaned",
+                            "%s -> interrupted (parent %s, parent"
+                            " status %s)" % (cid, pid, pstatus))
+                conn.commit()
+            finally:
+                conn.close()
+        except Exception as exc:
+            log_event(self.logger, "worker_orphan_sweep_failed",
+                      message="orphan sweep failed: %s" % exc)
+        return marked
+
     def _run_coordinator(self, task, started_at, poll_interval=None,
                          timeout=None):
         """协调员执行：建子任务 → 等齐 → 合流汇总写回。
@@ -2460,6 +2783,7 @@ class Worker:
                 timed_out = True
                 break
             time.sleep(poll_interval)
+        orphaned = []
         if cancelled or timed_out:
             rows = self._coord_child_rows(child_ids)
             for cid in child_ids:
@@ -2469,6 +2793,15 @@ class Worker:
                         self.cancel(cid)
                     except Exception:
                         pass
+            # P1：cancel() 对已终态/不存在的子任务返回 False，
+            # 跨 worker 在跑的子任务也可能因竞态仍未终态——
+            # cancel 之后仍未终态的一律直接标 interrupted 并记
+            # 审计（_coord_mark_orphans），绝不留孤儿无声烧钱；
+            # 标记结果写进汇总报告，调用方在详情里看得到。
+            try:
+                orphaned = self._coord_mark_orphans(task, child_ids)
+            except Exception:
+                orphaned = []
         rows = self._coord_child_rows(child_ids)
         merged = []
         for ch in children:
@@ -2479,6 +2812,9 @@ class Worker:
                 "result_text": row.get("result_text"),
                 "error_text": row.get("error_text")})
         report, overall = _coord_build_report(merged)
+        if orphaned:
+            report += ("\n\n孤儿子任务已标 interrupted：%s"
+                       % "、".join(orphaned))
         if cancelled:
             final = "interrupted"
             error = "cancelled"
@@ -2531,6 +2867,47 @@ class Worker:
                 "executor": self._executor_name(), "stdout": report,
                 "error": error, "events": [], "cost_record_id": None}
 
+    def _record_usage_cost(self, task, actual_model, usage_total,
+                           stdout, fallback_count):
+        """记一条任务成本并累加 worker 级用量（调用方须持 _db_lock）。
+
+        P1 成本修齐：旧版 max_cost_usd 参数存在但 usage_cost_usd
+        从未累加——_budget_exceeded 的成本门永远是 0，等于没设。
+        本函数是唯一记账点（终局块与取消分支共用）：actual 行
+        按价目表算 usd_micros（无价模型为 None），有价行才累加
+        usage_cost_usd（无价行不折算成 0 美元，仍受无价上限管）；
+        无用量时回落字符估算记 estimated，绝不冒充 actual。
+        cache_write_input_tokens 不进这里：它只在 usage dict 里
+        做观测（见 cost.py 的不计费声明），不加进 total、不用
+        于定价。返回 cost record id。
+        """
+        task_id = task["id"]
+        if usage_total and (usage_total.get("total_tokens") or 0) > 0:
+            micros = price_usd_micros(
+                actual_model, usage_total.get("input_tokens", 0),
+                usage_total.get("cached_input_tokens", 0),
+                usage_total.get("output_tokens", 0))
+            cost_id = self.costs.record(
+                task_id, task["project_id"], actual_model,
+                input_tokens=usage_total.get("input_tokens", 0),
+                output_tokens=usage_total.get("output_tokens", 0),
+                cached_input_tokens=usage_total.get(
+                    "cached_input_tokens", 0),
+                usd_micros=micros,
+                source="actual", fallback_count=fallback_count)
+        else:
+            micros = None
+            total = estimate_tokens(task.get("prompt") or "", stdout)
+            cost_id = self.costs.record(
+                task_id, task["project_id"], actual_model,
+                input_tokens=total, output_tokens=0, usd_micros=None,
+                source="estimated", fallback_count=fallback_count)
+        if micros is not None:
+            # max_cost_usd 不再是死参数：只有有价花费才累加，
+            # _budget_exceeded 据此在下一认领前挡住超支 worker。
+            self.usage_cost_usd += micros / 1000000.0
+        return cost_id
+
     def run_one(self, task=None):
         """认领并执行一个任务，返回结果 dict（见模块 docstring）。
 
@@ -2552,11 +2929,32 @@ class Worker:
         # 成它（只换内存里的副本，原 prompt 不动）；_resume 标记
         # 让 _run_attempt 在 goal 模式下走 goal-resume 而不是 goal
         # （goal 会撞已存在，只有 goal-resume 能在旧线程上开新循环）。
+        # P1 goal-resume 分流（断点 vs 开新目标，三处一致的 worker 侧）：
+        # - 旧验收形态（verify_command 非空）：一律断点续跑——
+        #   绝不能把续跑消息当新 goal_text 传给 host.py，否则
+        #   host 的 goal-resume --goal-text 分支会用 verify=None
+        #   开新循环，把原 verify_command 覆盖掉；断点调用不传
+        #   --goal-text，host 沿用原 goal + 原 verify + 原轮次，
+        #   从 goal_rounds_done 的下一轮续跑。续跑消息本轮不
+        #   当 prompt 用（断点语义就是续原目标），跑完照常清掉。
+        # - goal_mode=1（无验收）：开新目标——续跑消息就是新
+        #   goal_text（服务端 /continue 带 goal_mode=1 时已把
+        #   task.goal 写成该消息，这里与之对齐）。
+        # - 非 goal 任务：沿用旧行为（prompt 换成续跑消息，走
+        #   已存在续跑 run 路径）。
         resume_message = task.get("resume_message")
         if resume_message:
             task = dict(task)
-            task["prompt"] = resume_message
-            task["_resume"] = True
+            if task.get("verify_command"):
+                task["_resume"] = True
+                task["_resume_kind"] = "breakpoint"
+            elif task.get("goal_mode"):
+                task["prompt"] = resume_message
+                task["_resume"] = True
+                task["_resume_kind"] = "new-goal"
+            else:
+                task["prompt"] = resume_message
+                task["_resume"] = True
         self._audit(task["project_id"], "worker_claim",
                     "%s executor=%s" % (task_id, self._executor_name()))
         # log_event 只接受 task_id/project_id/status/message 四个
@@ -2647,6 +3045,56 @@ class Worker:
                     "stdout": "", "error": quota_error, "events": [],
                     "cost_record_id": None}
 
+        # P1 goal 轮次上限门（执行侧防线）：旧验收形态的
+        # goal_max_rounds 超过 GOAL_MAX_ROUNDS_CAP 时直接 failed，
+        # 不进 host.py goal 循环烧轮次；goal_mode=1 的 0（模型宣告
+        # 才结束）豁免轮次上限（上界是配额），但 0 配上旧验收命令
+        # 时同样 failed（服务端建任务时 0 本就 400，老行若有 0
+        # 验收形态属于脏数据，不按不限轮次跑）。建任务时的 400
+        # 校验在 server.py（另一路文件），这里是执行侧第二道门，
+        # 错误文本写明上限与“请新建合规任务”，不静默截断轮次
+        # （截断会改任务语义，调用方看不出被改过）。
+        _goal_rounds = task.get("goal_max_rounds")
+        if (task.get("verify_command") or task.get("goal_mode")) \
+                and _goal_rounds is not None:
+            _goal_cap_error = None
+            try:
+                _goal_rounds_int = int(_goal_rounds)
+            except (TypeError, ValueError):
+                _goal_rounds_int = None
+            if _goal_rounds_int is not None \
+                    and _goal_rounds_int > GOAL_MAX_ROUNDS_CAP:
+                _goal_cap_error = (
+                    "goal_max_rounds 超过上限 %d 轮（当前 %d），"
+                    "请新建轮次合规的任务" % (
+                        GOAL_MAX_ROUNDS_CAP, _goal_rounds_int))
+            elif _goal_rounds_int == 0 and task.get("verify_command") \
+                    and not task.get("goal_mode"):
+                _goal_cap_error = (
+                    "旧验收形态 goal_max_rounds=0 表示不限轮次，"
+                    "已禁用（上限 %d 轮），请新建轮次合规的任务"
+                    % GOAL_MAX_ROUNDS_CAP)
+            if _goal_cap_error is not None:
+                self._set_status(task_id, "failed")
+                self._write_execution_result(
+                    task_id, error_text=_goal_cap_error,
+                    actual_model=self.model,
+                    started_at=started_at, finished_at=time.time(),
+                    only_if_status="failed")
+                self._audit(task["project_id"], "worker_goal_cap_rejected",
+                            "%s: %s" % (task_id, _goal_cap_error))
+                log_event(self.logger, "worker_goal_cap_rejected",
+                          task_id=task_id, project_id=task["project_id"],
+                          status="failed")
+                self._running_clear(task_id)
+                return {"dispatched": True, "task_id": task_id,
+                        "host_name": self.executor._host_name(task)
+                        if hasattr(self.executor, "_host_name") else task_id,
+                        "status": "failed",
+                        "executor": self._executor_name(),
+                        "stdout": "", "error": _goal_cap_error, "events": [],
+                        "cost_record_id": None}
+
         # 协调员分支：不跑模型回合，改为按分工建并行子任务、等齐、
         # 合流汇总。已过 key 门与配额门；仓库门已对协调员任务本身
         # 跑过（项目仓库解析失败会在前面直接 failed，不会建一堆
@@ -2721,7 +3169,14 @@ class Worker:
         def _quota_hit():
             # 闭包读当前值：usage_total 与 task 在重试/fallback
             # 时会被重绑，轮询掐断要按最新值判断。
-            return self._quota_exceeded(task, usage_total) is not None
+            # P1 分流：执行中轮询只认真正的 quota_exceeded，
+            # quota_check_failed（查库失败）不杀当前轮——查询
+            # 抖动不应杀掉已在跑的回合；但它在重试/fallback/
+            # 后续消息等新回合边界仍是拒绝（那些边界直接调
+            # _quota_exceeded，不走这个过滤闭包）。
+            err = self._quota_exceeded(task, usage_total)
+            return err is not None and \
+                not err.startswith("quota_check_failed:")
 
         # 分支任务：先 fork 再跑首轮（见 _run_fork）；首轮失败
         # 不自动重试/fallback，失败后走重试入口重跑首轮（已存在
@@ -2763,6 +3218,39 @@ class Worker:
                 # 喂回，或新形态的模型自决多轮），worker 级重试/fallback
                 # 不再叠加——同任务重跑会撞 host 任务名已存在，第二
                 # 次尝试必失败且污染错误文本。
+                # P1 唯一例外：goal 首轮撞“已存在”时（重试入口
+                # 重排同一 goal 任务的典型死法），不再直接 failed，
+                # 而是走一次断点续跑（goal-resume 不传 --goal-text，
+                # 沿用原 goal+verify，从断点续跑）；断点续跑的
+                # 结果即本轮结果（成功则完成，仍失败则以续跑的
+                # 错误为准，不再拼“已存在”）。已是续跑的任务
+                # （_resume）不再二次回落，避免断点循环。
+                # 分支任务（forked_from）不走这里：fork 首轮的
+                # 已存在语义是 dispatch 续跑，已在 _run_attempt
+                # 内处理，不叠加 goal 断点。
+                if not task.get("_resume") and not task.get("forked_from"):
+                    _combined = ((result or {}).get("stdout", "") or "") \
+                        + ((result or {}).get("stderr", "") or "")
+                    _goal_err = (result or {}).get("goal_error") or ""
+                    if _ALREADY_EXISTS_MARKER in _combined or \
+                            _ALREADY_EXISTS_MARKER in _goal_err or \
+                            _ALREADY_EXISTS_MARKER in (error or ""):
+                        self._audit(task["project_id"],
+                                    "worker_goal_resume_fallback",
+                                    "%s: goal 已存在，走断点续跑" % task_id)
+                        _resume_task = dict(task)
+                        _resume_task["_resume"] = True
+                        _resume_task["_resume_kind"] = "breakpoint"
+                        result, error = self._run_attempt(
+                            _resume_task, quota_check=_quota_hit)
+                        usage_total = sum_usage(
+                            usage_total, (result or {}).get("usage"))
+                        self._deliver_peer_messages(task, result)
+                        if (result or {}).get("status") == "quota_exceeded":
+                            quota_error = self._quota_exceeded(
+                                task, usage_total) or \
+                                "quota_exceeded: 本月配额已用完"
+                            quota_stop = True
                 break
             if result.get("status") == "completed":
                 break
@@ -2900,6 +3388,18 @@ class Worker:
                 turn_id=(result or {}).get("turn_id"),
                 root_turn_id=(result or {}).get("root_turn_id"),
                 only_if_status="interrupted")
+            # P1：取消分支先记一条成本再返回（与配额中断收尾
+            # 同款记账点）。旧版这里直接返回 cost_record_id=None，
+            # 被取消前已花的模型费用就从账上丢了；取消不等于没
+            # 花钱。记账失败不挡取消返回（costs.record 异常时
+            # cost_id 置 None，任务状态仍是 interrupted）。
+            try:
+                with self._db_lock:
+                    _cancel_cost_id = self._record_usage_cost(
+                        task, actual_model, usage_total,
+                        (result or {}).get("stdout", ""), fallback_count)
+            except Exception:
+                _cancel_cost_id = None
             log_event(self.logger, "worker_cancelled", task_id=task_id,
                       project_id=task["project_id"], status="interrupted")
             return {"dispatched": True, "task_id": task_id,
@@ -2908,7 +3408,7 @@ class Worker:
                     "status": "interrupted", "executor": self._executor_name(),
                     "stdout": (result or {}).get("stdout", ""),
                     "error": "cancelled", "events": [],
-                    "cost_record_id": None}
+                    "cost_record_id": _cancel_cost_id}
 
         stdout = result.get("stdout", "") or ""
         host_status = result.get("host_status")
@@ -2979,32 +3479,12 @@ class Worker:
         self._cleanup_task_seed(task)
         with self._db_lock:
             self.usage_turns += 1
-            # 成本：执行器带回上游真实用量时记 actual（input/output/
-            # cached 全是实测值，usd 按价目表算、无价模型记 None，
-            # 报表里进 unpriced_actual_tokens 而不是 0 美元）；
-            # 没带回用量时回落字符估算记 estimated，绝不冒充 actual。
-            # fallback_count 复用同一记录；模型记最后真正跑的那个
-            # （fallback 成功时就是 fallback 模型）。cost 记录共用
-            # 一条连接，并发下必须持锁写，否则两个线程共用同一
-            # sqlite 连接会直接报错。
-            if usage_total and (usage_total.get("total_tokens") or 0) > 0:
-                cost_id = self.costs.record(
-                    task_id, task["project_id"], actual_model,
-                    input_tokens=usage_total.get("input_tokens", 0),
-                    output_tokens=usage_total.get("output_tokens", 0),
-                    cached_input_tokens=usage_total.get(
-                        "cached_input_tokens", 0),
-                    usd_micros=price_usd_micros(
-                        actual_model, usage_total.get("input_tokens", 0),
-                        usage_total.get("cached_input_tokens", 0),
-                        usage_total.get("output_tokens", 0)),
-                    source="actual", fallback_count=fallback_count)
-            else:
-                total = estimate_tokens(task.get("prompt") or "", stdout)
-                cost_id = self.costs.record(
-                    task_id, task["project_id"], actual_model,
-                    input_tokens=total, output_tokens=0, usd_micros=None,
-                    source="estimated", fallback_count=fallback_count)
+            # 成本记账走 _record_usage_cost（终局与取消分支共用
+            # 的唯一记账点）：actual/estimated 分流、usage_cost_usd
+            # 累加都在里面；cost 记录共用一条连接，并发下必须
+            # 持锁写，否则两个线程共用同一 sqlite 连接会直接报错。
+            cost_id = self._record_usage_cost(
+                task, actual_model, usage_total, stdout, fallback_count)
         if final_written:
             self._audit(task["project_id"],
                         "worker_quota_stop" if quota_stop else
@@ -3064,9 +3544,65 @@ class Worker:
         发 review-requested/merge-failed，这里只写回任务终态。
         跳过（PrSkipped，如任务已不是 merged）与落盘失败都记
         failed + error，任务本身保持 merged、可重调 /pr 再排队。
+        P1 非 local 通道：旧版非 local worker 直接返回 []，
+        排队的 pr 任务看起来像“没人处理”且无任何痕迹——
+        静默未完成。现在改为显式 needs_local_worker：只读
+         peek 本主机 queued 的 pr 任务，逐条记审计 + 日志并
+        返回 needs_local_worker 条目，任务本身保持 merged、
+        pr 任务保持 queued（留给 local worker 认领），绝不
+        把它标 failed（failed 会让调用方以为 PR 已失败，且
+        任务仍可重调 /pr 的语义会被混淆）。为什么不直接把
+        pr_jobs 行写成 needs_local_worker 状态：pr_jobs 的认
+        领只认 queued（claim_pr_job 在 pr_exec.py，本路不改
+        该文件），写成新状态后 local worker 也认领不到，任务
+        会被永久搁置；真要落库状态，需 pr_exec.py 的认领同
+        步改成含 needs_local_worker 或 local 重排队——该改动
+        留给协调员（见汇报残留），本路不伪造一个“写了状态
+        却没人能接”的通道。签名无效时同样不碰任务（返回 []，
+        只记日志），fail-closed。
         """
-        if self._executor_name() != "local":
+        if not self._code_signature_valid():
+            log_event(self.logger, "worker_pr_refused_code_signature",
+                      message="code signature missing or changed,"
+                      " refusing pr jobs")
             return []
+        if self._executor_name() != "local":
+            results = []
+            try:
+                from cloud.pr_exec import ensure_pr_jobs_table
+                conn = _db_connect(self.db_path)
+                try:
+                    ensure_pr_jobs_table(conn)
+                    rows = conn.execute(
+                        "SELECT id, task_id, project_id FROM pr_jobs"
+                        " WHERE host = ? AND status = 'queued'",
+                        (self.host_name,)).fetchall()
+                finally:
+                    conn.close()
+                for row in rows:
+                    try:
+                        job_id = row["id"]
+                        task_id = row["task_id"]
+                        project_id = row["project_id"]
+                    except (KeyError, IndexError, TypeError):
+                        continue
+                    reason = ("PR 执行需要 local worker（host-worktrees"
+                              " 只在主机上可见），本 worker 为 %s，"
+                              "任务保持 queued 等 local worker 认领"
+                              % self._executor_name())
+                    self._audit(project_id, "pr_job_needs_local_worker",
+                                "%s: %s" % (job_id, reason))
+                    log_event(self.logger, "pr_job_needs_local_worker",
+                              task_id=task_id, status="needs_local_worker",
+                              message=reason)
+                    results.append({"pr_job_id": job_id,
+                                    "task_id": task_id,
+                                    "status": "needs_local_worker",
+                                    "error": reason})
+            except Exception as exc:
+                log_event(self.logger, "pr_job_peek_failed",
+                          message="pr job peek failed: %s" % exc)
+            return results
         results = []
         while True:
             conn = _db_connect(self.db_path)
@@ -3312,8 +3848,12 @@ class Worker:
                 # 任务；退出码 75（EX_TEMPFAIL，失败类）：unit 托管
                 # 的由 systemd on-failure 重拉新版，autoscaler 托管
                 # 的由 reconcile 重拉，孤儿直接消失。单次 run_all
-                # 不走这里（进程本来就退出）。
-                if self._code_changed():
+                # 不走这里（进程本来就退出）。P1 fail-closed：
+                # 签名缺失同样自退（旧版只在“变了”时退，取不到
+                # 签名时继续跑是 fail-open）；每次认领前的签名门
+                # 在 _claim 里（池线程不走轮次边界也拦得住）。
+                if self._code_changed() or \
+                        not self._code_signature_valid():
                     log_event(self.logger, "worker_code_changed_exit",
                               message="worker code changed, exiting "
                               "for restart")
@@ -3336,6 +3876,15 @@ class Worker:
                     except Exception as exc:
                         log_event(self.logger, "worker_sweep_failed",
                                   message="worker sweep failed: %s" % exc)
+                    # P1 孤儿扫尾（接在陈旧回收之后）：协调员本身
+                    # 被回收标 interrupted 后，它的未终态子任务
+                    # 在这里被统一标 interrupted；协调员仍在跑时
+                    # 本函数不动子任务，见 sweep 方法 docstring。
+                    try:
+                        self.sweep_orphaned_coordinator_children()
+                    except Exception as exc:
+                        log_event(self.logger, "worker_orphan_sweep_failed",
+                                  message="orphan sweep failed: %s" % exc)
                 # 每轮结束写心跳：哪怕本轮一个任务都没派，worker 活着
                 # 这件事本身也要被 server 看到（文件心跳 + 注册表心跳）。
                 self._write_heartbeat()