gapfill2: worker/sub90 coverage tests, G7 rollback branch, G5 caller-contract docs

H1 worker.py 92%->94% (277->210 miss): 35 real-assertion tests
covering goal-signal, tenant/env, followup, poll/wait,
process/watchdog and injected defensive branches; prod code 0 changes.
H2 sub-90 sweep: approval_timeout 100%, merge_gate_hard 100%,
worker_sweep 98% (only __main__ line 174 left); 14 tests.
H3 G7 rollback except:pass (server.py:6615-6616) covered by an
injected rollback-raising connection; original error still propagates,
status stays completed, retry succeeds; no semantics change.
H4 caller contract documented only (docstrings in _audit_scorecard
endpoint, check_baseline_stale, build_scorecard, baseline store):
current_main_head caller-supplied, any string as-is, no format checks,
missing/empty treated as not passed with no stale determination at the
endpoint, always 200, overall never rewritten; no new validation or
rejection, no response-structure change. 3 doc tests pin the wording.
H5 design notes (no decisions, no code changes) live in
verification/code-to10-gapfill2-design-notes.md, referenced by the
batch summary.
No bug found by the new tests; no existing assertion touched.

cloud/approvals.py

@@ -74,6 +74,7 @@ import json
 import time
 import uuid
 
+from cloud.peer_messages import row_get
 from cloud.timeouts import APPROVAL_DEFAULT_SECONDS as _APPROVAL_DEFAULT
 from cloud.timeouts import expires_at_for as _expires_at_for
 from cloud.timeouts import resolve_module_timeout as _resolve_timeout
@@ -86,16 +87,6 @@ MAX_SUMMARY_CHARS = 2000
 MAX_DETAIL_CHARS = 20000
 
 
-def row_get(row, name):
-    """行取值容错：sqlite3.Row 缺列报 IndexError、dict 缺键报
-    KeyError、空行报 TypeError，一律回 None(与 peer_messages
-    的同名函数同款，模块独立所以这里自带一份)。"""
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return None
-
-
 def ensure_approvals_table(conn):
     """幂等建审批队列表。调用方负责 commit(服务端入口在写
     操作后统一 commit；本函数自己不 commit，避免半截事务)。"""

cloud/arch_conformance.py

@@ -74,7 +74,6 @@ stopping at the first.
 """
 
 import ast
-import os
 from pathlib import Path
 
 FOUNDATION_MODULES = frozenset({"db", "models", "queue", "logging_config"})

cloud/audit_scorecard.py

@@ -31,6 +31,9 @@
   build_scorecard(..., baseline_head=None) 的 baseline_head 是机器字段，
   只能由调用方显式传入当前 main 头；本模块绝不在生产代码里探测
   版本控制状态、绝不读版本控制目录、绝不编造一个头。
+  接受任意字符串、原样记录，不做任何格式校验（不要求形如
+  commit hash）；本模块不验证该值是否真实、可信，可信 main 头
+  的来源仍是调用方责任。
   不传入（或传入空串）时 baseline_head 为 None，并在 missing_items
   里记 baseline_head_missing，不把缺失读成“与当前一致”。
   check_baseline_stale(scorecard, current_main_head) 是唯一的漂移判定：
@@ -916,6 +919,16 @@ def check_baseline_stale(scorecard, current_main_head):
     参数全部由调用方传入：scorecard 是 build_scorecard 的输出 dict，
     current_main_head 是调用方当前确认的 main 头。本函数不在生产
     代码里探测版本控制状态，也不读版本控制目录。
+    调用方契约（gapfill2 H4，只写文档、不改行为）：
+    current_main_head 接受任意字符串、原样比对，不做任何格式
+    校验（不要求形如 commit hash，不校验长度、字符集、前缀）；
+    本函数不验证该值是否真实、可信、是否确为 main 头——可信
+    main 头的来源仍是调用方责任。空串先归一化为 None，与未
+    传入同等处理。注意端点层与函数层的区别：评分卡端点在缺失
+    或空串时根本不调用本函数（不做判定、不附加判定字段）；
+    调用方若直接调本函数且 current_main_head 缺失，本函数按
+    失败关闭返回 stale=True（见下三种情形之二），那是函数层
+    面的判定结果，不是端点行为。
 
     返回 dict（调用方凭 stale / usable_as_pass_basis 判断，绝不能
     只看旧卡的 overall / overall_pass）：
@@ -1068,7 +1081,8 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None,
     （user_side_items 可选占位），product=build_product_scorecard
     （无真实产品轮证据时恒为 not_pass）。新增键不改变旧字段的值。
     另新增 rubric_version（见 cloud/scorecard_rubric.py）、
-    baseline_head（机器字段，见 check_baseline_stale）、
+    baseline_head（机器字段，见 check_baseline_stale；接受任意
+    字符串原样记录、不做格式校验，空串按未传入处理）、
     evidence_ledger_head 与 evidence_row_ids（见
     compute_evidence_ledger）三个机制字段；无证据行时头为 None。
     """

cloud/automations.py

@@ -56,6 +56,7 @@ import time
 import uuid
 from datetime import datetime, timezone
 
+from cloud.peer_messages import row_get
 from cloud.quota import decide as _quota_decide
 from cloud.quota import default_quota_usd_micros as _quota_default
 from cloud.quota import monthly_usage as _quota_monthly_usage
@@ -73,14 +74,9 @@ ACTIVE = "active"
 CANCELLED = "cancelled"
 
 
-def row_get(row, name):
+# 行取值走 cloud.peer_messages.row_get（A7 全仓统一）：旧私有版
-    """行取值容错：sqlite3.Row 缺列报 IndexError、dict 缺键报
+# 与统一版在真实行类型（None / dict / sqlite3.Row）下语义完全
-    KeyError、空行报 TypeError，一律回 None(老库行缺列时
+# 相同（缺键/缺列/空行一律回 None），故直接复用，不再自带一份。
-    读自动化列表不能因此炸掉整页)。"""
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return None
 
 
 AUTOMATION_COLUMNS = (

cloud/evidence_binding.py

@@ -32,7 +32,6 @@
 与 full_auto.audit_record 的 ts 同口径。
 """
 
-import os
 import time
 import uuid
 

cloud/marker_tools.py

@@ -131,7 +131,6 @@ import。DB 助手函数只用 ? 占位符，与 peer_messages 同款写法。
 """
 
 import json
-import os
 import re
 import time
 import uuid
@@ -140,6 +139,7 @@ from typing import Any
 
 from cloud.gates import RETRYABLE_STATUSES as _GATE_RETRYABLE
 from cloud.gates import TERMINAL_STATUSES as _GATE_TERMINAL
+from cloud.peer_messages import row_get
 from cloud.gates import decide_control as _gate_decide_control
 from cloud.gates import is_goal_task as _gate_is_goal_task
 from cloud.gates import resolve_inside as _gate_resolve_inside
@@ -286,14 +286,6 @@ def parse_first_run_at(value):
     return epoch
 
 
-def row_get(row, name):
-    """与 peer_messages.row_get 同款容错取值。"""
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return None
-
-
 def _as_int(value, default):
     if isinstance(value, bool):
         return default

cloud/resident.py

@@ -89,7 +89,7 @@ from urllib.parse import urlsplit
 
 from cloud.automations import _audit_row as _auto_audit_row
 from cloud.automations import _automation_quota_exceeded
-from cloud.automations import row_get
+from cloud.peer_messages import row_get
 from cloud.db import PostgresIntegrityError as _PgIntegrityError
 
 MIN_INTERVAL_SECONDS = 60

cloud/scorecard_baseline_store.py

@@ -13,7 +13,9 @@
 行字段：
 - rubric_version：评分口径版本（调用方传入的字符串，如 "v1"；
   本模块不规定口径内容、不替用户拍板阈值）。
-- baseline_head：冻结时调用方传入的 main 头（字符串，非空）。
+- baseline_head：冻结时调用方传入的 main 头（字符串，非空；
+  接受任意字符串原样存、不做格式校验，本模块不验证其真实性，
+  可信来源仍是调用方责任）。
 - frozen_at：冻结时间戳（秒，调用方不传时取当前时间）。
 - scorecard_snapshot：build_scorecard 输出的完整 dict，以 JSON
   文本存入，读回时还原为 dict。

cloud/server.py

@@ -68,6 +68,7 @@ from cloud.quota import quota_for_owner as _quota_for_owner
 from cloud.timeouts import STALE_SWEEP_SECONDS as _STALE_SWEEP
 from cloud.timeouts import approval_default_seconds as _approval_default
 from cloud.timeouts import validate_server_timeout as _validate_timeout
+from cloud.auth import audit_on_conn as _audit_on_conn
 from cloud.db import connect as _db_connect
 from cloud.db import create_database as _create_database
 from cloud.db import init_postgres_schema as _init_postgres_schema
@@ -2730,7 +2731,7 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
         elif path == "/metrics-report" and method == "GET":
             self._metrics_report(user_id)
         elif path == "/audit-scorecard" and method == "GET":
-            self._audit_scorecard(user_id)
+            self._audit_scorecard(user_id, query)
         elif path == "/models" and method == "GET":
             self._models_overview(user_id)
         elif path == "/relay/models" and method == "GET":
@@ -6578,15 +6579,47 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
                 "merge_gate": gate,
             })
             return
+        # G7（A4 时序残留）:状态提交与成功审计必须同一事务。
+        # 旧写法先 commit 状态、再另开 AuthStore 连接写审计：
+        # 审计失败时状态已是 merged、响应却是 500——“合并了却
+        # 报失败”，调用方无法靠状态判断，也无法安全重试
+        # （重复合并 409）。新写法用 audit_on_conn 把审计对
+        # 写进同一连接、一次 commit：审计失败则整段回滚，状态
+        # 仍是 completed、响应 500（_dispatch_safe 兜底），调用
+        # 方可原样重试合并。链头唯一冲突的重试口径与
+        # AuthStore.audit 完全一致（5 次、只重试
+        # IntegrityError，其它错误直接抛）。成功语义不变：
+        # 200 体、审计内容（merge_task/task_id）、门禁响应全同。
+        # 前提：tasks 与 audit_log/audit_ledger 在同一库
+        # （_db 与 _auth_store 都连 server.db_path；PG 模式下
+        # _connect 同样分发到同一 PG 库，? 占位由适配层转 %s）。
         now = time.time()
         conn = self._db()
         try:
-            conn.execute(
+            last_exc = None
-                "UPDATE tasks SET status = 'merged', updated_at = ?"
+            for _attempt in range(5):
-                " WHERE id = ?",
+                try:
-                (now, task_id),
+                    conn.execute(
-            )
+                        "UPDATE tasks SET status = 'merged',"
-            conn.commit()
+                        " updated_at = ? WHERE id = ?",
+                        (now, task_id),
+                    )
+                    _audit_on_conn(conn, user_id, task["project_id"],
+                                   "merge_task", task_id)
+                    conn.commit()
+                    break
+                except Exception as exc:
+                    last_exc = exc
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                    if not isinstance(
+                            exc, (sqlite3.IntegrityError,
+                                  _PgIntegrityError)):
+                        raise
+            else:
+                raise last_exc
             task = _task_to_dict(
                 conn.execute(
                     "SELECT * FROM tasks WHERE id = ?", (task_id,)
@@ -6594,8 +6627,6 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
             )
         finally:
             conn.close()
-        with _auth_store(self.server.db_path) as store:
-            store.audit(user_id, task["project_id"], "merge_task", task_id)
         task["merge_gate"] = gate
         self._send_json(200, task)
 
@@ -8697,9 +8728,31 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的
             conn.close()
         metrics = _compute_metrics(records, costs)
         self._send_json(200, _build_metrics_report(metrics))
-    def _audit_scorecard(self, user_id):
+    def _audit_scorecard(self, user_id, query=None):
         """只读 7 维审计评分卡报告端点。
 
+        可选漂移判定（gapfill G5，机制 item 5 端点快照接线）：
+        调用方可在查询串里传 ``?current_main_head=<头>``，端点
+        即用 cloud/audit_scorecard.py 的 check_baseline_stale()
+        对返回的卡做一次漂移判定，结果放在响应顶层的
+        ``baseline_staleness`` 里（stale / usable_as_pass_basis
+        等字段原样返回）。该判定绝不改写卡本身的 overall /
+        overall_pass（旧值保留只为追溯），也绝不挡住查询：
+        未传参数时默认不做判定，响应与以前完全一致；传了
+        参数也永远返回 200 + 报告，只是多一个判定字段。
+        生产代码绝不读 git、不起 subprocess 查版本：main 头
+        只能由调用方传入，端点本身不探测版本控制状态。
+        调用方契约（gapfill2 H4，只写文档、不改行为）：
+        current_main_head 由调用方提供，接受任意字符串、原样
+        使用，不做任何格式校验（不要求形如 commit hash，不
+        校验长度、字符集、前缀）；端点不验证该值是否真实、
+        是否可信、是否确为 main 头——可信 main 头的来源仍是
+        调用方责任。缺失（未传该查询参数）与空串一律按未传入
+        处理：端点不调用判定、不附加 baseline_staleness 字段，
+        响应与以前完全一致。传入非空值后，端点永远返回 200 +
+        报告、只附加判定字段：判定为 stale 也不挡查询，绝不改
+        写报告的 overall / overall_pass。
+
         权限：与 /metrics 同款，任意项目成员即可读（不要求 owner），
         但本端点不统计任何项目数据、不写库、不写账本、不改链头、
         不触发推回或重审——它只返回评分卡模块的一张历史示例卡。
@@ -8751,6 +8804,26 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的
                             "untested_items": [],
                             "product_gate_passed": False},
             }
+        # gapfill G5：可选漂移判定。未传参数时不做判定、不标
+        # stale，响应与以前一致；传了参数才附加判定字段，且
+        # 判定结果绝不改写报告的 overall / overall_pass。
+        heads = (query or {}).get("current_main_head") or []
+        current = heads[0] if heads else None
+        if current:
+            try:
+                from cloud.audit_scorecard import (
+                    check_baseline_stale)
+                report["baseline_staleness"] = check_baseline_stale(
+                    report, current)
+            except ImportError:  # pragma: no cover - 模块不可用时诚实说明
+                report["baseline_staleness"] = {
+                    "checked": False,
+                    "reason": "audit_scorecard_module_unavailable",
+                    "stale": True,
+                    "usable_as_pass_basis": False,
+                    "overall_usable_as_pass_basis": False,
+                    "overall_must_not_be_used_as_pass_basis": True,
+                }
         self._send_json(200, report)
 
 

cloud/tests/test_code_search_ext.py

@@ -13,14 +13,12 @@ import unittest
 from pathlib import Path
 
 try:
-    from cloud import websearch as _ws
     from cloud.websearch import (
         CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain,
         _normalize_allowed_hosts, search)
     from cloud.websearch_mcp import handle_request
 except ImportError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
-    from cloud import websearch as _ws
     from cloud.websearch import (
         CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain,
         _normalize_allowed_hosts, search)

cloud/tests/test_gapfill2_g7_rollback.py

@@ -0,0 +1,163 @@
+"""gapfill2 H3：_merge_task 成功审计段里 rollback 自身的 except: pass。
+
+上一批唯一未覆盖分支：G7 新段里 ``conn.rollback()`` 本身抛错时的
+防御分支（基底行号 server.py:6615-6616）。rollback 从不失败，故
+全量覆盖率里该分支一直缺失。本文件只用注入式测试覆盖它，绝不
+改动 rollback 语义、绝不把该分支改成可观测行为：
+
+- 注入一个 rollback 会抛错的连接代理（除 rollback 外全部透传
+  真实连接），同时让审计失败（沿用 G7 的 RuntimeError 注入），
+  使执行流必然走进 ``except Exception: pass``；
+- 断言：rollback 的异常被吞掉（原审计错误原样上抛，不被掩盖）、
+  任务状态仍是 completed、审计 0 行、原样重试后成功 200。
+
+真实模型 0 轮，不联网；不跑真实 HTTP（handler 直调 +
+真实 AuthStore/DB，审计行直接查库验证）。
+"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+import cloud.server as server_mod  # noqa: E402
+from cloud.auth import AuthStore  # noqa: E402
+from cloud.server import CloudControlPlaneHandler, init_db  # noqa: E402
+
+NO_BLOCK = {"merge_blocked": False, "release_blocked": False,
+            "blocked_by": None, "round": None, "failed_checks": []}
+
+HARD_BODY = {
+    "files_changed": ["apps/web/a.py"],
+    "owned_dir": "apps/web",
+    "branch_protection": "protected",
+    "ci_status": "success",
+    "ci_checks": [{"name": "ci", "status": "success"}],
+    "required_checks": ["ci"],
+    "contract_version": "v2",
+    "contract_frozen": True,
+    "coordinator_approved": True,
+    "unresolved_review_comments": 0,
+}
+
+
+class _RollbackFailingConn:
+    """代理真实连接：除 rollback 外全部透传，rollback 计数后抛错。
+
+    execute / commit / close / row_factory 等一律委托给真实连接，
+    只有 rollback 被替换为“计数 + 抛错”，精确模拟“rollback 自身
+    失败”这一防御分支的前提，不改动其它任何连接语义。
+    """
+
+    def __init__(self, real, calls):
+        self._real = real
+        self._calls = calls
+
+    def rollback(self):
+        self._calls["n"] += 1
+        raise RuntimeError("rollback down")
+
+    def __getattr__(self, name):
+        return getattr(self._real, name)
+
+
+class RollbackFailureSwallowedTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, _key = self.auth.create_user("owner")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO projects(id, name, owner_id, created_at,"
+                " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)",
+                ("p1", "P", self.owner_id, time.time(), None))
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " created_by, created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, ?, ?, ?)",
+                ("t1", "p1", "do work", "completed", self.owner_id,
+                 time.time(), time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+        self.auth.create_project("p1", self.owner_id)
+
+    def tearDown(self):
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _handler(self):
+        h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
+        h.server = SimpleNamespace(db_path=self.db_path)
+        sent = []
+        h._send_json = lambda status, obj, extra_headers=None: sent.append(
+            (status, obj))
+        h._read_json = lambda: (dict(HARD_BODY), None)
+        h.sent = sent
+        return h
+
+    def _merge(self, h):
+        with mock.patch.object(server_mod, "_pushback_block",
+                               return_value=dict(NO_BLOCK)):
+            h._merge_task(self.owner_id, "t1")
+
+    def _status(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            return conn.execute(
+                "SELECT status FROM tasks WHERE id = 't1'").fetchone()[0]
+        finally:
+            conn.close()
+
+    def _merge_audits(self):
+        return [a for a in self.auth.list_audit("p1")
+                if a["action"] == "merge_task"]
+
+    def test_rollback_failure_is_swallowed(self):
+        # 审计失败 + rollback 自身失败：rollback 的异常必须被吞掉，
+        # 原审计错误原样上抛（不被 rollback 错误掩盖），状态回滚
+        # 语义不变（completed、审计 0 行），原样重试后成功。
+        h = self._handler()
+        real_db = CloudControlPlaneHandler._db
+        calls = {"n": 0}
+
+        def failing_db(handler_self):
+            return _RollbackFailingConn(real_db(handler_self), calls)
+
+        with mock.patch.object(CloudControlPlaneHandler, "_db",
+                               failing_db):
+            with mock.patch.object(
+                    server_mod, "_audit_on_conn",
+                    side_effect=RuntimeError("audit down")):
+                with self.assertRaises(RuntimeError) as ctx:
+                    self._merge(h)
+        self.assertEqual(str(ctx.exception), "audit down")
+        self.assertEqual(calls["n"], 1)
+        self.assertEqual(h.sent, [])
+        self.assertEqual(self._status(), "completed")
+        self.assertEqual(self._merge_audits(), [])
+        # 原样重试（审计恢复、连接恢复后）必须成功：状态仍是
+        # completed，不是卡在 merged 导致重复合并 409。
+        h2 = self._handler()
+        self._merge(h2)
+        self.assertEqual(h2.sent[-1][0], 200)
+        self.assertEqual(self._status(), "merged")
+        self.assertEqual(len(self._merge_audits()), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_gapfill2_h4_docs.py

@@ -0,0 +1,61 @@
+"""gapfill2 H4：G5 调用方契约的文档钉死（只改文档、不改行为）。
+
+本文件断言的全部是 docstring 文本，不是行为：上一批残留
+“可信 main 头的来源仍是调用方责任”在本批补进 scorecard
+相关模块与端点的 docstring 后，必须写清四点——
+current_main_head 由调用方提供、接受什么形式的值、缺失与
+空串按未传入处理且不做 stale 判定、传入后永远 200 且不改写
+overall；且绝不新增任何校验或拒绝行为（行为断言见既有
+test_gapfill_g5_baseline_snapshot，本文件不重复测行为）。
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+import cloud.audit_scorecard as scorecard_mod  # noqa: E402
+import cloud.scorecard_baseline_store as store_mod  # noqa: E402
+from cloud.server import CloudControlPlaneHandler  # noqa: E402
+
+
+def _flat(doc):
+    """docstring 空白归一化：去掉全部空白后按语义短语查，
+    不依赖换行位置（中文短语常被换行切开）。"""
+    return "".join((doc or "").split())
+
+
+class CallerContractDocsTestCase(unittest.TestCase):
+    def test_endpoint_docstring_states_caller_contract(self):
+        doc = _flat(CloudControlPlaneHandler._audit_scorecard.__doc__)
+        self.assertIn("由调用方提供", doc)
+        self.assertIn("接受任意字符串", doc)
+        self.assertIn("不做任何格式校验", doc)
+        self.assertIn("不要求形如commithash", doc)
+        self.assertIn("缺失", doc)
+        self.assertIn("空串一律按未传入处理", doc)
+        self.assertIn("不附加baseline_staleness字段", doc)
+        self.assertIn("永远返回200", doc)
+        self.assertIn("绝不改写报告的overall/overall_pass", doc)
+        self.assertIn("可信main头的来源仍是调用方责任", doc)
+
+    def test_check_baseline_stale_docstring_states_contract(self):
+        doc = _flat(scorecard_mod.check_baseline_stale.__doc__)
+        self.assertIn("接受任意字符串", doc)
+        self.assertIn("不做任何格式校验", doc)
+        self.assertIn("空串先归一化为None", doc)
+        self.assertIn("根本不调用本函数", doc)
+        self.assertIn("可信", doc)
+
+    def test_module_and_store_docstrings_state_contract(self):
+        self.assertIn("接受任意字符串",
+                      _flat(scorecard_mod.__doc__))
+        self.assertIn("不做格式校验", _flat(store_mod.__doc__))
+        self.assertIn("可信来源仍是调用方责任",
+                      _flat(store_mod.__doc__))
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_gapfill2_sub90.py

@@ -0,0 +1,200 @@
+"""gapfill2 H2：全仓低于 90% 文件的扫尾（生产代码 0 改动）。
+
+基线（本批自己跑的 CI 同款覆盖率，基底 5d0651e）低于 90% 的
+文件共 37 个。本文件补其中三个缺口最小、可纯函数 / 注入式
+真实驱动的文件，缺口大的文件（resident、sandbox、ledger、
+playwright_browser 等，缺口均在 50 行以上且多需 Docker / 真实
+外部服务）不在本批硬凑，见总结残留清单：
+
+- cloud/approval_timeout.py：57 语句、缺失 8 行（41-42、46、
+  48、62-63、65、109）→ 本文件全覆盖。
+- cloud/merge_gate_hard.py：89 语句、缺失 12 行（103、108、
+  142-143、147、157-158、201-202、212-215）→ 本文件全覆盖。
+- cloud/worker_sweep.py：60 语句、缺失 8 行（54、82-83、
+  117-118、131-132、174）→ 本文件覆盖除 174（``__main__``
+  守卫行，import 无法覆盖）外的 7 行。
+
+真实模型 0 轮，不联网；approval_timeout 与 merge_gate_hard
+均为纯函数直调，worker_sweep 用假 AuthStore / 假 reap 注入。
+"""
+
+import sqlite3
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+import cloud.worker_sweep as sweep_mod  # noqa: E402
+from cloud.approval_timeout import (  # noqa: E402
+    approval_timeout_stats,
+    recommend_approval_timeout,
+    timeout_rate,
+)
+from cloud.merge_gate_hard import evaluate_hard_merge_gate  # noqa: E402
+
+
+def _gate(**overrides):
+    body = {
+        "files_changed": ["apps/web/a.py"],
+        "owned_dir": "apps/web",
+        "protection": "protected",
+        "ci_status": "success",
+        "ci_checks": [{"name": "ci", "status": "success"}],
+        "required_checks": ["ci"],
+        "contract_version": "v2",
+        "contract_frozen": True,
+        "coordinator_approved": True,
+        "unresolved_review_comments": 0,
+    }
+    body.update(overrides)
+    return evaluate_hard_merge_gate(**body)
+
+
+class ApprovalTimeoutBranchTestCase(unittest.TestCase):
+    def test_timeout_rate_non_numeric_none(self):
+        # 41-42：计数无法转 int 时回 None，不是 0%。
+        self.assertIsNone(timeout_rate("abc", 10))
+        self.assertIsNone(timeout_rate(1, "xyz"))
+
+    def test_timeout_rate_clamps(self):
+        # 46：expired 为负时按 0 计；48：expired 超过 decided
+        # 时按 decided 计（比率封顶 1.0）。
+        self.assertEqual(timeout_rate(-3, 10), 0.0)
+        self.assertEqual(timeout_rate(15, 10), 1.0)
+
+    def test_recommend_current_fallbacks(self):
+        # 62-63：current 非法时按默认值 600 计；65：current
+        # <= 0 时同样按默认值计。
+        self.assertEqual(
+            recommend_approval_timeout(0, 10,
+                                       current_seconds="bad")
+            ["current_seconds"], 600)
+        self.assertEqual(
+            recommend_approval_timeout(0, 10, current_seconds=0)
+            ["current_seconds"], 600)
+        self.assertEqual(
+            recommend_approval_timeout(0, 10, current_seconds=-5)
+            ["current_seconds"], 600)
+
+    def test_stats_empty_project_list(self):
+        # 109：project_ids 为空列表时直接回零统计，不拼 IN 查询。
+        conn = sqlite3.connect(":memory:")
+        try:
+            self.assertEqual(
+                approval_timeout_stats(conn, []),
+                {"decided_count": 0, "expired_count": 0,
+                 "pending_count": 0})
+        finally:
+            conn.close()
+
+
+class MergeGateHardBranchTestCase(unittest.TestCase):
+    def test_ci_check_entry_not_dict(self):
+        # 103 + 147：ci_checks 条目不是 dict 时索引为 None、
+        # 记 ci_check_failed 块。
+        result = _gate(ci_checks=["not-a-dict"])
+        self.assertFalse(result["allowed"])
+        self.assertIn("ci_check_failed", result["blocks"])
+
+    def test_ci_check_entry_bad_name_status(self):
+        # 108 + 147：条目 name 为空时索引为 None、记
+        # ci_check_failed 块。
+        result = _gate(ci_checks=[{"name": "", "status": "success"}])
+        self.assertFalse(result["allowed"])
+        self.assertIn("ci_check_failed", result["blocks"])
+
+    def test_ci_checks_not_list(self):
+        # 142-143：ci_checks 非列表时记 ci_evidence_missing。
+        result = _gate(ci_checks="ci")
+        self.assertFalse(result["allowed"])
+        self.assertIn("ci_evidence_missing", result["blocks"])
+
+    def test_required_checks_not_list(self):
+        # 157-158：required_checks 非列表时记 ci_evidence_missing。
+        result = _gate(required_checks="ci")
+        self.assertFalse(result["allowed"])
+        self.assertIn("ci_evidence_missing", result["blocks"])
+
+    def test_reviews_bool_is_missing_evidence(self):
+        # 201-202：评审意见为 bool 时不是“0 条”，而是无证据。
+        result = _gate(unresolved_review_comments=True)
+        self.assertFalse(result["allowed"])
+        self.assertIn("review_evidence_missing", result["blocks"])
+
+    def test_reviews_negative_and_other_type(self):
+        # 212：负数评审意见记 review_evidence_missing；
+        # 214-215：其它类型（字符串）同样记无证据。
+        result = _gate(unresolved_review_comments=-1)
+        self.assertIn("review_evidence_missing", result["blocks"])
+        result = _gate(unresolved_review_comments="many")
+        self.assertIn("review_evidence_missing", result["blocks"])
+
+
+class WorkerSweepBranchTestCase(unittest.TestCase):
+    def _fake_store_cls(self, user_id, audit_raises=False,
+                        close_raises=False):
+        class _FakeConn:
+            def close(self):
+                if close_raises:
+                    raise RuntimeError("close failed")
+
+        class _FakeStore:
+            def __init__(self, db_path):
+                self.conn = _FakeConn()
+
+            def authenticate(self, api_key):
+                return user_id if api_key else None
+
+            def audit(self, *args, **kwargs):
+                if audit_raises:
+                    raise RuntimeError("audit down")
+
+        return _FakeStore
+
+    def test_global_admin_ids_explicit(self):
+        # 54：显式传入 admin_ids 时去空白、去空项。
+        self.assertEqual(
+            sweep_mod.global_admin_ids([" a ", "", "b", " a "]),
+            {"a", "b"})
+
+    def test_denied_audit_failure_swallowed(self):
+        # 82-83：非管理员被拒时，拒绝审计本身抛错也被吞掉，
+        # PermissionError 照常抛出，绝不调 reap。
+        fake = self._fake_store_cls("u1", audit_raises=True)
+        with mock.patch.object(sweep_mod, "AuthStore", fake):
+            with self.assertRaises(PermissionError):
+                sweep_mod.sweep_stale_workers_admin(
+                    "/tmp/x.db", "k", admin_ids=["admin1"])
+
+    def test_sweep_audit_failure_marks_summary_false(self):
+        # 117-118：汇总审计抛错时 audit_summary 置 False，
+        # 调用仍正常返回（reap 结果不受影响）。
+        fake = self._fake_store_cls("admin1", audit_raises=True)
+        with mock.patch.object(sweep_mod, "AuthStore", fake):
+            with mock.patch.object(
+                    sweep_mod, "reap_stale_workers",
+                    return_value=[]):
+                summary = sweep_mod.sweep_stale_workers_admin(
+                    "/tmp/x.db", "k", admin_ids=["admin1"])
+        self.assertFalse(summary["audit_summary"])
+        self.assertEqual(summary["killed"], [])
+
+    def test_store_close_failure_swallowed(self):
+        # 131-132：finally 里 store.conn.close() 抛错时被吞掉，
+        # 汇总照常返回。
+        fake = self._fake_store_cls("admin1", close_raises=True)
+        with mock.patch.object(sweep_mod, "AuthStore", fake):
+            with mock.patch.object(
+                    sweep_mod, "reap_stale_workers",
+                    return_value=[]):
+                summary = sweep_mod.sweep_stale_workers_admin(
+                    "/tmp/x.db", "k", admin_ids=["admin1"])
+        self.assertTrue(summary["audit_summary"])
+        self.assertEqual(summary["user_id"], "admin1")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_gapfill2_worker_gaps.py

@@ -0,0 +1,610 @@
+"""gapfill2 H1：worker.py 覆盖缺口补测试（生产代码 0 改动）。
+
+基线（本批自己跑的 CI 同款覆盖率，基底 5d0651e）：
+cloud/worker.py 3493 语句、277 缺失、92%。本文件只加真实断言
+测试，瞄准基线缺失行里可真实驱动的分支；每组测试头上注明
+它覆盖的基底行号。需要 Docker / 协调员 / run_one 主循环等重
+型 harness 的缺失行（run_one、_run_attempt、_docker_task_context、
+run_loop、run_all 等）不在本文件硬凑，见总结残留清单。
+
+真实模型 0 轮，不联网；不启动真实 Docker / 子进程，凡需注入
+处均用假连接、假 runner、假 proc 或方法补丁，且断言均为行为
+断言（返回值、落盘行、调用记录），不是“跑过即过”。
+"""
+
+import json
+import logging
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+import cloud.worker as worker_mod  # noqa: E402
+from cloud.server import init_db  # noqa: E402
+from cloud.worker import DockerExecutor, Worker  # noqa: E402
+
+
+def _worker(db_path, **attrs):
+    w = Worker.__new__(Worker)
+    w.db_path = db_path
+    w.tenant_isolation = False
+    w.executor = SimpleNamespace()
+    w.logger = logging.getLogger("test_gapfill2_worker")
+    w._cancel_requested = set()
+    w._running = {}
+    w._running_lock = threading.Lock()
+    w._loop_stop = False
+    w.quota_poll_interval_seconds = 30
+    w.process_factory = None
+    w.repos_runner = None
+    for key, value in attrs.items():
+        setattr(w, key, value)
+    return w
+
+
+def _insert_task(db_path, task_id="T1", status="running", goal="old goal",
+                 goal_version=0, applied=0):
+    now = time.time()
+    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, goal, goal_mode,"
+            " verify_command, goal_max_rounds, goal_version,"
+            " applied_goal_version) VALUES(?, 'p1', ?, ?, NULL, '[]',"
+            " 0, 'tester', ?, ?, NULL, ?, 1, NULL, 0, ?, ?)",
+            (task_id, goal, status, now, now, goal, goal_version,
+             applied))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _task_row(db_path, task_id="T1"):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.row_factory = sqlite3.Row
+        return dict(conn.execute(
+            "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone())
+    finally:
+        conn.close()
+
+
+def _write_state(state_path, tasks):
+    with open(state_path, "w", encoding="utf-8") as fh:
+        json.dump({"tasks": tasks}, fh)
+
+
+class GoalSignalBranchTestCase(unittest.TestCase):
+    """goal 信号写侧 / 镜像分支（基底缺失 1294 / 1300 / 1306 /
+    1313-1314 / 1339-1340 / 1381-1382 / 1397-1398）。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "test.db")
+        init_db(self.db_path)
+        self.state_path = os.path.join(self.tmp.name, "host-state.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _signal_worker(self):
+        w = _worker(self.db_path)
+        w._host_state_path_for_task = lambda task: self.state_path
+        return w
+
+    def test_bad_goal_rejected(self):
+        # 1294：db_goal 为空或非字符串时返回 False，不写信号。
+        w = self._signal_worker()
+        task = {"id": "T1", "project_id": "p1"}
+        self.assertFalse(w._write_goal_signal_if_newer(task, None, 2))
+        self.assertFalse(w._write_goal_signal_if_newer(task, "", 2))
+        self.assertFalse(w._write_goal_signal_if_newer(task, 123, 2))
+        self.assertFalse(os.path.exists(
+            os.path.join(self.tmp.name, "goal-updates")))
+
+    def test_no_state_path_returns_false(self):
+        # 1300：解析不出 state 路径时返回 False，不猜路径。
+        w = _worker(self.db_path)
+        w._host_state_path_for_task = lambda task: None
+        self.assertFalse(w._write_goal_signal_if_newer(
+            {"id": "T1", "project_id": "p1"}, "new goal", 2))
+
+    def test_entry_not_dict_returns_false(self):
+        # 1306：state 里本任务条目不是 dict 时返回 False。
+        _write_state(self.state_path, {"T1": "not-a-dict"})
+        w = self._signal_worker()
+        self.assertFalse(w._write_goal_signal_if_newer(
+            {"id": "T1", "project_id": "p1"}, "new goal", 2))
+
+    def test_corrupt_state_swallowed(self):
+        # 1313-1314：state 文件损坏（json 解析失败）时异常被吞掉，
+        # 返回 False，绝不炸掉调用方。
+        with open(self.state_path, "w", encoding="utf-8") as fh:
+            fh.write("{not json")
+        w = self._signal_worker()
+        self.assertFalse(w._write_goal_signal_if_newer(
+            {"id": "T1", "project_id": "p1"}, "new goal", 2))
+
+    def test_write_signal_success_control(self):
+        # 对照项（非缺失行）：版本大于已应用版本时写信号成功，
+        # 证明上面几组失败测试的 harness 本身能走通成功路径。
+        _write_state(self.state_path, {
+            "T1": {"applied_goal_version": 0, "goal_rounds_done": 1}})
+        w = self._signal_worker()
+        self.assertTrue(w._write_goal_signal_if_newer(
+            {"id": "T1", "project_id": "p1"}, "new goal", 2))
+        signal_dir = os.path.join(self.tmp.name, "goal-updates")
+        self.assertTrue(os.path.isdir(signal_dir))
+        self.assertEqual(len(os.listdir(signal_dir)), 1)
+
+    def test_sync_resume_missing_id_swallowed(self):
+        # 1339-1340：续跑认领钩子读库时任务缺 id（KeyError）时
+        # 异常被吞掉，返回 False，续跑本身不受影响。
+        w = _worker(self.db_path)
+        self.assertFalse(w._sync_resume_goal_signal({}))
+
+    def test_mirror_bad_applied_ignored(self):
+        # 1381-1382：state 里 applied_goal_version 不是数字时，
+        # 转换失败置 None、跳过回写，不改任务行、不炸镜像。
+        _insert_task(self.db_path, applied=0)
+        _write_state(self.state_path, {"T1": {
+            "goal_rounds_done": 3, "status": "goal-running",
+            "applied_goal_version": "not-a-number",
+            "thread_id": "th-1"}})
+        w = self._signal_worker()
+        self.assertEqual(
+            w._mirror_checkpoint_once({"id": "T1", "project_id": "p1"}), 3)
+        self.assertEqual(_task_row(self.db_path)["applied_goal_version"],
+                         0)
+
+    def test_mirror_db_read_failure_skips_signal_only(self):
+        # 1397-1398：镜像内读库（goal/goal_version）失败时只跳过
+        # 反向信号，检查点镜像本身继续、返回轮次，不炸、不回 -1。
+        _insert_task(self.db_path, applied=0)
+        _write_state(self.state_path, {"T1": {
+            "goal_rounds_done": 2, "status": "goal-running",
+            "applied_goal_version": 0, "thread_id": "th-1"}})
+        w = self._signal_worker()
+        real_connect = worker_mod._db_connect
+
+        class _FailingGoalReadConn:
+            def __init__(self, real):
+                self._real = real
+
+            def execute(self, sql, params=()):
+                if "SELECT goal, goal_version" in sql:
+                    raise sqlite3.OperationalError(
+                        "injected goal read failure")
+                return self._real.execute(sql, params)
+
+            def __getattr__(self, name):
+                return getattr(self._real, name)
+
+        def failing_connect(db_path):
+            return _FailingGoalReadConn(real_connect(db_path))
+
+        with mock.patch.object(worker_mod, "_db_connect",
+                               failing_connect):
+            self.assertEqual(
+                w._mirror_checkpoint_once(
+                    {"id": "T1", "project_id": "p1"}), 2)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT round_no FROM task_checkpoints"
+                " WHERE task_id = 'T1'").fetchone()
+        finally:
+            conn.close()
+        self.assertEqual(row, (2,))
+
+
+class TenantEnvBranchTestCase(unittest.TestCase):
+    """租户目录 / 环境分支（基底缺失 3062 / 3091 / 3121-3122 /
+    3148-3149 / 3156-3157 / 3181-3182 / 3196）。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _tenant_worker(self, **extra):
+        w = _worker(os.path.join(self.tmp.name, "test.db"))
+        w.tenant_isolation = True
+        w.tenants_root = os.path.join(self.tmp.name, "tenants")
+        w.tenant_template_home = os.path.join(self.tmp.name, "tmpl-home")
+        w.tenant_template_repo = os.path.join(self.tmp.name, "tmpl-repo")
+        w.seed_skills = False
+        w.websearch_mcp = False
+        w.imagegen_mcp = False
+        w.playwright_mcp = False
+        for key, value in extra.items():
+            setattr(w, key, value)
+        return w
+
+    def test_tenant_dirs_missing_test_repo_template(self):
+        # 3062：test-repo 为空目录、模板不存在时直接抛错，
+        # 不静默播种一个空 test-repo。
+        os.makedirs(self._tenant_worker().tenant_template_home)
+        with open(os.path.join(self.tmp.name, "tmpl-home",
+                               "config.toml"), "w") as fh:
+            fh.write("[x]\n")
+        w = self._tenant_worker()
+        with self.assertRaises(RuntimeError) as ctx:
+            w._tenant_dirs("p1")
+        self.assertIn("tmpl-repo", str(ctx.exception))
+
+    def test_tenant_mounts_includes_repo_checkout(self):
+        # 3091：repo_checkout 目录已存在时，挂载列表里追加
+        # repos/<安全项目名> 挂载；不存在时不硬挂载。
+        tmpl_home = os.path.join(self.tmp.name, "tmpl-home")
+        tmpl_repo = os.path.join(self.tmp.name, "tmpl-repo")
+        os.makedirs(tmpl_home)
+        os.makedirs(tmpl_repo)
+        with open(os.path.join(tmpl_home, "config.toml"), "w") as fh:
+            fh.write("[x]\n")
+        with open(os.path.join(tmpl_repo, "sample.py"), "w") as fh:
+            fh.write("# sample\n")
+        w = self._tenant_worker()
+        dirs = w._tenant_dirs("p1")
+        os.makedirs(dirs["repo_checkout"], exist_ok=True)
+        mounts = w._tenant_mounts("p1")
+        self.assertIn(
+            "%s:/var/lib/codex-app-server/repos/p1:rw"
+            % dirs["repo_checkout"], mounts)
+
+    def test_env_host_dir_tenant_docker(self):
+        # 3121-3122：租户 + Docker 时环境宿主目录走
+        # tenants_root/<项目>/envs/<env_id>。
+        w = self._tenant_worker()
+        w.executor = DockerExecutor.__new__(DockerExecutor)
+        result = w._env_host_dir({"project_id": "p1", "id": "env1"})
+        self.assertTrue(result.startswith(w.tenants_root))
+        self.assertTrue(result.endswith(
+            os.path.join("envs", "env1")))
+
+    def test_fetch_environment_db_failure_none(self):
+        # 3148-3149：库连接失败时返回 None，不炸 worker。
+        w = _worker(os.path.join(self.tmp.name, "no-such-dir",
+                                 "test.db"))
+        self.assertIsNone(w._fetch_environment("e1"))
+
+    def test_fetch_environment_narrow_schema_none_fields(self):
+        # 3156-3157：老库 environments 表缺可选列时，_get 兜底
+        # 回 None，不炸整行；必备列仍正常读出。
+        db_path = os.path.join(self.tmp.name, "test.db")
+        conn = sqlite3.connect(db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE environments(id TEXT, project_id TEXT,"
+                " name TEXT, status TEXT)")
+            conn.execute(
+                "INSERT INTO environments VALUES('e1', 'p1', 'E',"
+                " 'ready')")
+            conn.commit()
+        finally:
+            conn.close()
+        w = _worker(db_path)
+        env = w._fetch_environment("e1")
+        self.assertEqual(env["id"], "e1")
+        self.assertEqual(env["project_id"], "p1")
+        self.assertEqual(env["status"], "ready")
+        self.assertIsNone(env["branch"])
+        self.assertIsNone(env["repo_url"])
+
+    def test_fetch_environment_missing_required_column_none(self):
+        # 3170-3171：environments 行缺必备列（project_id）时，
+        # 外层兜底回 None，不炸 worker。
+        db_path = os.path.join(self.tmp.name, "test.db")
+        conn = sqlite3.connect(db_path)
+        try:
+            conn.execute("CREATE TABLE environments(id TEXT)")
+            conn.execute("INSERT INTO environments VALUES('e1')")
+            conn.commit()
+        finally:
+            conn.close()
+        w = _worker(db_path)
+        self.assertIsNone(w._fetch_environment("e1"))
+
+    def test_run_git_typeerror_fallback(self):
+        # 3181-3182：repos_runner 单参调用抛 TypeError 时，
+        # 回落到 (cmd, timeout=None) 双参调用。
+        calls = []
+        _unset = object()
+
+        def runner(*args, timeout=_unset):
+            calls.append((args, timeout))
+            if timeout is _unset:
+                raise TypeError("single-arg not supported")
+            return SimpleNamespace(returncode=0, stdout="ok",
+                                   stderr="")
+
+        w = _worker(os.path.join(self.tmp.name, "test.db"))
+        w.repos_runner = runner
+        result = w._run_git(["git", "status"])
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "ok")
+        self.assertEqual(len(calls), 2)
+        self.assertIs(calls[1][1], None)
+
+    def test_prepare_environment_no_repo_url(self):
+        # 3196：环境无 repo_url 时直接返回错误文本，不建目录。
+        w = _worker(os.path.join(self.tmp.name, "test.db"))
+        self.assertEqual(w._prepare_environment({}),
+                         "environment has no repo_url")
+
+
+class DbFailureBranchTestCase(unittest.TestCase):
+    """查库失败即回落的分支（基底缺失 3333-3334 / 3339-3340 /
+    3583-3584 / 3604-3605 / 3620-3621 / 3650-3651 / 3665-3666 /
+    3694-3695 / 5343-5344 / 5798-5799）。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "test.db")
+        init_db(self.db_path)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _bad_db_worker(self):
+        return _worker(os.path.join(self.tmp.name, "no-such-dir",
+                                    "test.db"))
+
+    def test_project_repo_url_db_failure_none(self):
+        # 3333-3334：projects 表查库失败时回 None，不炸派发。
+        self.assertIsNone(self._bad_db_worker()._project_repo_url("p1"))
+
+    def test_project_repo_url_row_missing_key_none(self):
+        # 3339-3340：行里缺 repo_url 键（非 Row 行）时回 None。
+        class _Cursor:
+            def fetchone(self):
+                return {}
+
+        class _Conn:
+            def execute(self, sql, params=()):
+                return _Cursor()
+
+            def close(self):
+                pass
+
+        w = _worker(self.db_path)
+        with mock.patch.object(worker_mod, "_db_connect",
+                               lambda path: _Conn()):
+            self.assertIsNone(w._project_repo_url("p1"))
+
+    def test_claim_followup_db_connect_failure_none(self):
+        # 3583-3584：followup 认领时库连接失败回 None。
+        self.assertIsNone(
+            self._bad_db_worker()._claim_next_followup("t1"))
+
+    def test_claim_followup_missing_table_none(self):
+        # 3604-3605：task_followups 表不存在时回 None，不炸 run_one。
+        fresh = os.path.join(self.tmp.name, "fresh.db")
+        sqlite3.connect(fresh).close()
+        w = _worker(fresh)
+        self.assertIsNone(w._claim_next_followup("t1"))
+
+    def test_finish_followup_db_failure_silent(self):
+        # 3620-3621：结束 followup 时库失败静默，不抛。
+        self._bad_db_worker()._finish_followup("f1", "done")
+
+    def test_convert_to_goal_db_failure_false(self):
+        # 3650-3651：转 goal 时库失败回 False，不悄悄改状态。
+        self.assertFalse(
+            self._bad_db_worker()._convert_running_task_to_goal(
+                "t1", "new goal"))
+
+    def test_restore_followup_db_failure_silent(self):
+        # 3665-3666：放回 followup 时库失败静默，不抛。
+        self._bad_db_worker()._restore_followup("f1")
+
+    def test_queue_self_followup_db_failure_false(self):
+        # 3694-3695：排自有 followup 时库失败回 False。
+        w = self._bad_db_worker()
+        self.assertFalse(w._queue_self_followup(
+            {"id": "t1", "project_id": "p1"}, "body"))
+
+    def test_coord_child_rows_missing_key_skipped(self):
+        # 5343-5344：子任务行缺列（非 Row 行）时跳过该行，
+        # 不炸等齐判断。
+        class _Cursor:
+            def fetchone(self):
+                return {"id": "c1"}
+
+        class _Conn:
+            def execute(self, sql, params=()):
+                return _Cursor()
+
+            def close(self):
+                pass
+
+        w = _worker(self.db_path)
+        with mock.patch.object(worker_mod, "_db_connect",
+                               lambda path: _Conn()):
+            self.assertEqual(w._coord_child_rows(["c1"]), {})
+
+    def test_record_ledger_attempt_outbox_failure_swallowed(self):
+        # 5798-5799：台账写失败后 outbox 追加本身也失败时，
+        # 仍只吞掉、返回 None，不影响任务状态与返回值。
+        w = self._bad_db_worker()
+        w.attempt_ledger_hook = True
+        import cloud.ledger_outbox as outbox_mod  # noqa: E402
+        with mock.patch.object(
+                outbox_mod, "append_failure",
+                side_effect=RuntimeError("outbox down")):
+            self.assertIsNone(
+                w._record_ledger_attempt("t1", "fail", True))
+
+
+class ProcessWatchdogBranchTestCase(unittest.TestCase):
+    """进程 / 看门狗分支（基底缺失 2505-2506 / 2892-2893 /
+    3853-3854 / 3874-3875 / 3881-3882）。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "test.db")
+        init_db(self.db_path)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_docker_cancel_watchdog_db_interrupted(self):
+        # 2505-2506：看门狗查库发现任务已 interrupted 时杀容器。
+        killed = []
+        w = _worker(self.db_path)
+        w.executor = SimpleNamespace(
+            kill_container=lambda name: killed.append(name))
+        w.get_task = lambda task_id: {"status": "interrupted"}
+        stop_event = threading.Event()
+        w._docker_cancel_watchdog("t1", "c1", stop_event)
+        self.assertEqual(killed, ["c1"])
+
+    def test_execute_via_process_kill_failure_swallowed(self):
+        # 2892-2893：communicate 超时后 proc.kill() 本身抛错时
+        # 被吞掉，第二次 communicate 的结果仍正常返回 completed。
+        calls = {"kill": 0}
+
+        class _FakeProc:
+            pid = 999
+            returncode = 0
+
+            def poll(self):
+                return 0
+
+            def communicate(self, timeout=None):
+                if timeout is not None:
+                    raise RuntimeError("communicate timeout")
+                return ("out", "err")
+
+            def kill(self):
+                calls["kill"] += 1
+                raise RuntimeError("kill failed")
+
+        w = _worker(self.db_path)
+        w.process_factory = lambda task: _FakeProc()
+        with mock.patch.object(Worker, "_ensure_gateway",
+                               return_value=(None, None)):
+            result = w._execute_via_process({"id": "t1"})
+        self.assertEqual(calls["kill"], 1)
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["stdout"], "out")
+        self.assertEqual(result["stderr"], "err")
+        self.assertEqual(result["returncode"], 0)
+
+    def test_marker_auth_close_failure_swallowed(self):
+        # 3853-3854：权限查完后 store.conn.close() 抛错时被吞掉，
+        # 权限结论（True）不受影响。
+        class _FakeConn:
+            def close(self):
+                raise RuntimeError("close failed")
+
+        class _FakeStore:
+            def __init__(self, db_path):
+                self.conn = _FakeConn()
+
+            def check_permission(self, creator, project_id, action):
+                return True
+
+        w = _worker(self.db_path)
+        with mock.patch.object(worker_mod, "AuthStore", _FakeStore):
+            self.assertEqual(
+                w._marker_auth({"created_by": "u1",
+                                "project_id": "p1"}, "read"),
+                (True, ""))
+
+    def test_task_files_worktree_remote_host(self):
+        # 3874-3875：worktree 在别机时直接拒绝，不指错目录。
+        w = _worker(self.db_path)
+        path, reason = w._task_files_worktree(
+            {"executor": "local", "worktree_host": "other-host-xyz",
+             "id": "t1"})
+        self.assertIsNone(path)
+        self.assertIn("other-host-xyz", reason)
+
+    def test_task_files_worktree_bad_project_falls_through(self):
+        # 3881-3882：project_id 缺失致安全名解析失败时，
+        # safe_project 置 None、走本地 worktree 分支，不炸。
+        w = _worker(self.db_path)
+        w.host_py_path = os.path.join(self.tmp.name, "host.py")
+        path, reason = w._task_files_worktree(
+            {"executor": "local", "project_id": None, "id": "t1"})
+        self.assertIsNone(path)
+        self.assertIn("没有本地 worktree", reason)
+
+
+class PollWaitBranchTestCase(unittest.TestCase):
+    """轮询 / 等待分支（基底缺失 3711-3712 / 3720-3721 /
+    3728-3729 / 3747 / 3768-3769 / 3774-3775 / 3804-3805 / 3825）。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "test.db")
+        init_db(self.db_path)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _poll(self, timeout=10):
+        return {"task_id": "T2", "interval_seconds": 1,
+                "timeout_seconds": timeout, "chars": 100}
+
+    def test_poll_cancelled_before_snapshot(self):
+        # 3711-3712 + 3747：父任务已取消时不等快照、直接回
+        # “读取失败”，不挂起。
+        w = _worker(self.db_path)
+        w._cancel_requested = {"P1"}
+        result = w._poll_task_snapshot(
+            {"id": "P1", "project_id": "p1"}, self._poll())
+        self.assertEqual(result, "轮询任务 T2：读取失败。")
+
+    def test_poll_db_failure_returns_missing(self):
+        # 3720-3721 + 3728-3729：心跳与快照查库都失败时，
+        # 异常被吞掉、回“不存在”，不炸等待循环。
+        w = _worker(os.path.join(self.tmp.name, "no-such-dir",
+                                 "test.db"))
+        w.get_task = lambda task_id: None
+        result = w._poll_task_snapshot(
+            {"id": "P1", "project_id": "p1"}, self._poll())
+        self.assertEqual(result, "轮询任务 T2：不存在（同项目内也查不到）。")
+
+    def test_wait_cancelled(self):
+        # 3768-3769 + 3825 + 3831：父任务已取消时等待中止，
+        # 未读到的目标记“读取失败”，末尾附取消说明。
+        w = _worker(self.db_path)
+        w._cancel_requested = {"P1"}
+        lines = w._wait_task_targets(
+            {"id": "P1", "project_id": "p1"},
+            {"task_ids": ["T2"], "interval_seconds": 1,
+             "timeout_seconds": 10, "chars": 100})
+        self.assertEqual(lines[0], "[等待 1 个任务]")
+        self.assertIn("等待任务 T2：读取失败。", lines)
+        self.assertIn("说明：等待期间父任务被取消，等待已中止。", lines)
+
+    def test_wait_db_failure_silent(self):
+        # 3774-3775 + 3804-3805：心跳与批量快照查库都失败时，
+        # 异常被吞掉、超时 0 立即收尾，不炸、不挂起。
+        w = _worker(os.path.join(self.tmp.name, "no-such-dir",
+                                 "test.db"))
+        w.get_task = lambda task_id: None
+        lines = w._wait_task_targets(
+            {"id": "P1", "project_id": "p1"},
+            {"task_ids": ["T2"], "interval_seconds": 1,
+             "timeout_seconds": 0, "chars": 100})
+        self.assertEqual(lines[0], "[等待 1 个任务]")
+        self.assertIn("等待任务 T2：读取失败。", lines)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_gapfill_g2_goal_edges.py

@@ -0,0 +1,194 @@
+"""gapfill G2：goal_live_update.py 边缘分支补测。
+
+补测基线（055da3c 线定向子集对本模块的覆盖）：
+80 语句缺 15 行：57、59（写信号校验抛错）、79-80（读信号
+applied_version 非整数回落）、88（读信号非 dict）、109-110
+（删信号 applied_version 非整数回落）、115-116（删信号
+文件缺失/坏 JSON）、118（删信号非 dict）、121（删信号
+version 非整数）、126-127（删信号 os.remove 失败）、181
+（apply_goal_update old_version 为 0 分支）、187（apply
+CAS 未命中回 None）。
+
+只加测试，不改生产代码；信号只写临时目录，不联网。
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.goal_live_update import (
+        apply_goal_update,
+        consume_goal_update_signal,
+        ensure_goal_history_table,
+        read_goal_update_signal,
+        signal_path,
+        write_goal_update_signal,
+    )
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.goal_live_update import (
+        apply_goal_update,
+        consume_goal_update_signal,
+        ensure_goal_history_table,
+        read_goal_update_signal,
+        signal_path,
+        write_goal_update_signal,
+    )
+
+
+class WriteValidationTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.state = os.path.join(self.tmp.name, "host-state.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_bad_version_raises(self):
+        # 57：version 非正整数（0、负数、bool、字符串）一律
+        # ValueError，绝不写出信号文件。
+        for bad in (0, -1, True, "2", 2.5):
+            with self.assertRaises(ValueError):
+                write_goal_update_signal(
+                    self.state, "t1", bad, "new goal")
+        self.assertFalse(os.path.exists(
+            signal_path(self.state, "t1")))
+
+    def test_bad_goal_raises(self):
+        # 59：goal 非空字符串校验（空串、空白串、非字符串）。
+        for bad in ("", "   ", None, 123):
+            with self.assertRaises(ValueError):
+                write_goal_update_signal(self.state, "t1", 1, bad)
+
+
+class ReadEdgeTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.state = os.path.join(self.tmp.name, "host-state.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_bad_applied_version_falls_back_to_zero(self):
+        # 79-80：applied_version 非整数（int() 抛 ValueError /
+        # TypeError）时按 0 计，version=2 的信号仍可应用。
+        write_goal_update_signal(self.state, "t1", 2, "goal v2")
+        self.assertEqual(
+            read_goal_update_signal(self.state, "t1", "not-a-number"),
+            (2, "goal v2"))
+
+    def test_non_dict_signal_returns_none(self):
+        # 88：信号文件是合法 JSON 但不是 dict（列表）时回 None。
+        path = signal_path(self.state, "t1")
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump([1, 2], fh)
+        self.assertIsNone(read_goal_update_signal(self.state, "t1", 0))
+
+
+class ConsumeEdgeTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.state = os.path.join(self.tmp.name, "host-state.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _write_raw(self, payload):
+        path = signal_path(self.state, "t1")
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(payload, fh)
+        return path
+
+    def test_missing_file_returns_false(self):
+        # 115-116：信号文件不存在时回 False，不抛。
+        self.assertFalse(
+            consume_goal_update_signal(self.state, "t1", 5))
+
+    def test_bad_applied_version_falls_back_to_zero(self):
+        # 109-110：applied_version 非整数时按 0 计；version=1
+        # 的信号大于 0，故不删、回 False（下一条才是删的路）。
+        write_goal_update_signal(self.state, "t1", 1, "goal v1")
+        self.assertFalse(consume_goal_update_signal(
+            self.state, "t1", "not-a-number"))
+        self.assertTrue(os.path.exists(signal_path(self.state, "t1")))
+
+    def test_non_dict_signal_returns_false(self):
+        # 118：信号文件非 dict 时回 False，文件原样保留。
+        path = self._write_raw([1, 2])
+        self.assertFalse(
+            consume_goal_update_signal(self.state, "t1", 9))
+        self.assertTrue(os.path.exists(path))
+
+    def test_non_int_version_returns_false(self):
+        # 121：version 非整数（字符串）时回 False，文件保留。
+        path = self._write_raw({"version": "3", "goal": "g"})
+        self.assertFalse(
+            consume_goal_update_signal(self.state, "t1", 9))
+        self.assertTrue(os.path.exists(path))
+
+    def test_remove_failure_returns_false(self):
+        # 126-127：os.remove 失败时回 False，不抛。
+        write_goal_update_signal(self.state, "t1", 1, "goal v1")
+        with mock.patch("cloud.goal_live_update.os.remove",
+                        side_effect=OSError("disk gone")):
+            self.assertFalse(
+                consume_goal_update_signal(self.state, "t1", 1))
+        self.assertTrue(os.path.exists(signal_path(self.state, "t1")))
+
+
+class ApplyGoalUpdateTestCase(unittest.TestCase):
+    def setUp(self):
+        self.conn = sqlite3.connect(":memory:")
+        self.conn.execute(
+            "CREATE TABLE tasks(id TEXT PRIMARY KEY, goal TEXT, "
+            "goal_version INTEGER, updated_at REAL)")
+        ensure_goal_history_table(self.conn)
+        self.conn.execute(
+            "INSERT INTO tasks VALUES('t1', 'old', NULL, 0)")
+        self.conn.execute(
+            "INSERT INTO tasks VALUES('t2', 'old', 5, 0)")
+        self.conn.commit()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_null_version_branch(self):
+        # 181：old_version 为 NULL（按 0 计）时走 NULL/0 分支，
+        # 新 version 为 1，history 写一行。
+        self.assertEqual(
+            apply_goal_update(self.conn, "t1", "old", None,
+                              "new", "u1", 123.0), 1)
+        row = self.conn.execute(
+            "SELECT goal, goal_version FROM tasks WHERE id='t1'"
+        ).fetchone()
+        self.assertEqual(row, ("new", 1))
+        hist = self.conn.execute(
+            "SELECT old_goal, new_goal, version FROM task_goal_history "
+            "WHERE task_id='t1'").fetchall()
+        self.assertEqual(hist, [("old", "new", 1)])
+
+    def test_cas_miss_returns_none_without_history(self):
+        # 187：CAS 未命中（库里 version=5，调用方按 4 来）时
+        # 回 None，任务行与 history 一律不动。
+        self.assertIsNone(
+            apply_goal_update(self.conn, "t2", "old", 4,
+                              "new", "u1", 123.0))
+        row = self.conn.execute(
+            "SELECT goal, goal_version FROM tasks WHERE id='t2'"
+        ).fetchone()
+        self.assertEqual(row, ("old", 5))
+        self.assertEqual(self.conn.execute(
+            "SELECT COUNT(*) FROM task_goal_history "
+            "WHERE task_id='t2'").fetchone()[0], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_gapfill_g5_baseline_snapshot.py

@@ -0,0 +1,102 @@
+"""gapfill G5：机制 item 5 端点快照接线定向测试。
+
+接线方式：GET /audit-scorecard 可选查询串
+``?current_main_head=<头>``，端点用 check_baseline_stale()
+对返回的卡做一次漂移判定，结果放在响应顶层
+``baseline_staleness`` 里。锁定点：
+- 未传参数时默认不做判定：响应里没有 baseline_staleness，
+  其余字段与以前完全一致，查询绝不被挡住（永远 200）。
+- 传了参数时附加判定字段，但判定绝不改写卡本身的
+  overall / overall_pass（示例卡 baseline_head 为 None，
+  故任何传入的头都判 stale=True、原因 baseline_head_missing，
+  这是失败关闭，不是把旧分改写）。
+- 空串参数按未传入处理：不做判定、不标 stale。
+- 端点本身不读 git、不起 subprocess 查版本：main 头只能
+  由调用方传入（本文件只测 HTTP 入参一路，不测任何
+  版本控制探测——check_baseline_stale 本身已在机制基线
+  测试里钉住“只认调用方传入参数”）。
+
+只加测试；生产改动在 cloud/server.py 的 _audit_scorecard。
+"""
+
+import http.client
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+
+class BaselineSnapshotEndpointTestCase(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.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()
+
+    def _req(self, path):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        try:
+            conn.request("GET", path,
+                         headers={"X-API-Key": self.owner_key})
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        return resp.status, json.loads(data.decode("utf-8"))
+
+    def test_no_param_means_no_staleness_field(self):
+        status, report = self._req("/audit-scorecard")
+        self.assertEqual(status, 200)
+        self.assertEqual(report["overall"], "not_pass")
+        self.assertNotIn("baseline_staleness", report)
+
+    def test_param_attaches_staleness_without_rewriting_overall(self):
+        status, report = self._req(
+            "/audit-scorecard?current_main_head=82d0e84")
+        self.assertEqual(status, 200)
+        # 示例卡无冻结基线：显式要求判定时失败关闭为 stale，
+        # 但卡本身的 overall / overall_pass 原样不动。
+        self.assertEqual(report["overall"], "not_pass")
+        self.assertFalse(report["overall_pass"])
+        stale = report["baseline_staleness"]
+        self.assertTrue(stale["stale"])
+        self.assertFalse(stale["usable_as_pass_basis"])
+        self.assertTrue(stale["overall_must_not_be_used_as_pass_basis"])
+        self.assertIn("baseline_head_missing", stale["reason"])
+        self.assertEqual(stale["current_main_head"], "82d0e84")
+
+    def test_empty_param_means_no_staleness_field(self):
+        status, report = self._req("/audit-scorecard?current_main_head=")
+        self.assertEqual(status, 200)
+        self.assertNotIn("baseline_staleness", report)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_gapfill_g6_visibility.py

@@ -0,0 +1,258 @@
+"""gapfill G6：两个可见性缺口的测试钉死，不发明通道。
+
+只加测试，不改生产代码；真实模型 0 轮，不联网。
+
+钉住的两条“提交成功、永不应用”必须是可观测的，
+而不是悄无声息：
+
+1. CloudExecutor 无 host-state：
+   - CloudExecutor 本身没有 extra_mounts / config_path 属性，
+     故解析不出租户/Docker 的 state 路径；信号写侧在 state
+     文件不存在时返回 False，绝不猜路径、绝不写信号。
+   - 端点层面：running 任务改 goal 成功后返回体永远是
+     pending_mirror_to_host_loop，库里 goal_version 已 bump、
+     applied_goal_version 原地不动；反复读任务行，applied
+     永远不前进；task_goal_updated 审计与 history 行是这次
+     提交留下的唯一可查痕迹。
+   - 注意（既有行为，不在本批改）：_host_state_path_for_task
+     在解析不出路径时回落到 /opt/codex-app-server/work/
+     host-state.json；本测试只在该回落文件不存在时断言
+     写侧返回 False——回落文件若存在，写侧行为与此不同，
+     那是既有回落口径，不在本批悄悄改。
+
+2. argv-only 形态（任务表 verify_command 为空、host-state
+   里有 verify_argv）：
+   - 服务端放行：_goal_update_reject 只以 verify_command
+     把关，argv 只活在 host-state 里、不是任务表列，故
+     端点对这类任务返回 200 updated（不是 409），版本
+     bump、history 与审计都有行——“提交成功”本身可查。
+   - host 循环第二道门：有验收的循环一律忽略信号
+     （_apply_goal_update 内 has_verify 即返回 False），
+     故循环按旧 goal 正常结束（验收通过即 ok），但 saved
+     state 里 goal 仍是旧 goal、applied_goal_version 仍是 0、
+     信号文件仍在（未被误删，因为 version > 已应用版本，
+     清理函数不会删它）——版本差（goal_version >
+     applied_goal_version）+ 残留信号文件就是“永不应用”
+     的可观测证据，不是悄无声息。
+   - 本批不为 CloudExecutor 发明信号通道，不在未查清
+     argv 任务合法来源前改动任务提交的接受语义：服务端
+     拒绝 argv-only 若是安全的干净校验，需另行论证，本批
+     只钉死现状，不改接受语义。
+"""
+
+import contextlib
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+import host as host_mod  # noqa: E402
+from cloud import goal_live_update as glu  # noqa: E402
+from cloud.sandbox import CloudExecutor  # noqa: E402
+from cloud.server import CloudControlPlaneHandler, init_db  # noqa: E402
+from cloud.worker import Worker  # noqa: E402
+
+
+def _handler(db_path, body, permitted=True):
+    h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
+    h.server = SimpleNamespace(db_path=db_path)
+    sent = []
+    h._send_json = lambda status, obj, extra_headers=None: sent.append(
+        (status, obj))
+    h._read_json = lambda: (body, None)
+    h.sent = sent
+    audits = []
+
+    class FakeStore:
+        def check_permission(self, user_id, project_id, action):
+            return permitted and action == "approve_merge"
+
+        def audit(self, user_id, project_id, action, detail=""):
+            audits.append((user_id, project_id, action, detail))
+
+    @contextlib.contextmanager
+    def fake_auth_store(_db_path):
+        yield FakeStore()
+
+    h._audits = audits
+    return h, fake_auth_store
+
+
+def _insert_task(db_path, task_id="T1", status="running",
+                 goal="old goal", goal_mode=1, verify_command=None,
+                 goal_version=0, applied=0):
+    now = time.time()
+    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, goal, goal_mode,"
+            " verify_command, goal_max_rounds, goal_version,"
+            " applied_goal_version) VALUES(?, 'p1', ?, ?, NULL, '[]',"
+            " 0, 'tester', ?, ?, NULL, ?, ?, ?, 0, ?, ?)",
+            (task_id, goal, status, now, now, goal, goal_mode,
+             verify_command, goal_version, applied))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _task_row(db_path, task_id="T1"):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.row_factory = sqlite3.Row
+        return dict(conn.execute(
+            "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone())
+    finally:
+        conn.close()
+
+
+def _history(db_path, task_id="T1"):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(
+            "SELECT old_goal, new_goal, version, created_by FROM"
+            " task_goal_history WHERE task_id = ? ORDER BY version",
+            (task_id,)).fetchall()
+    finally:
+        conn.close()
+
+
+class CloudExecutorVisibilityTestCase(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)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_cloud_executor_has_no_state_path_attrs(self):
+        # CloudExecutor 不带 extra_mounts / config_path，故解析
+        # 不出任何 host-state 路径——这是“无 host-state”的根因。
+        ex = CloudExecutor()
+        self.assertIsNone(getattr(ex, "extra_mounts", None))
+        self.assertIsNone(getattr(ex, "config_path", None))
+
+    def test_write_signal_returns_false_without_state_file(self):
+        # 信号写侧唯一写点：state 文件不存在时回 False。
+        # 回落路径若在本机存在则跳过（既有回落口径，不在本批改）。
+        w = Worker.__new__(Worker)
+        w.db_path = self.db_path
+        w.tenant_isolation = False
+        w.executor = CloudExecutor()
+        task = {"id": "T1", "project_id": "p1"}
+        resolved = w._host_state_path_for_task(task)
+        if resolved and os.path.exists(resolved):
+            self.skipTest(
+                "fallback state file exists on this machine; "
+                "no-state branch not exercisable here")
+        self.assertFalse(w._write_goal_signal_if_newer(
+            task, "new goal", 2))
+
+    def test_running_update_stays_pending_and_unapplied(self):
+        # 端点层面：running 任务改 goal 成功，但 applied 永不
+        # 前进；反复读行，applied 仍是 0；审计与 history 是痕迹。
+        _insert_task(self.db_path, status="running")
+        h, fake_auth = _handler(self.db_path, {"goal": "new goal"})
+        with mock.patch.object(
+                __import__("cloud.server", fromlist=["_auth_store"]),
+                "_auth_store", fake_auth):
+            h._update_goal_task("tester", "T1")
+        status, obj = h.sent[-1]
+        self.assertEqual(status, 200)
+        self.assertEqual(obj["goal_update"], "updated")
+        self.assertEqual(obj["goal_signal_status"],
+                         "pending_mirror_to_host_loop")
+        row = _task_row(self.db_path)
+        self.assertEqual(row["goal"], "new goal")
+        self.assertEqual(row["goal_version"], 1)
+        self.assertEqual(row["applied_goal_version"], 0)
+        # 再读一次：没有任何 worker 把 applied 推上去。
+        row2 = _task_row(self.db_path)
+        self.assertEqual(row2["applied_goal_version"], 0)
+        self.assertEqual(_history(self.db_path),
+                         [("old goal", "new goal", 1, "tester")])
+        self.assertIn(("tester", "p1", "task_goal_updated",
+                       "T1:goal_version=1"), h._audits)
+
+
+class ArgvOnlyVisibilityTestCase(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.state_path = os.path.join(self.tmp.name, "host-state.json")
+        self.state = {"tasks": {}}
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_server_passes_argv_only_task(self):
+        # 服务端只以 verify_command 把关：任务表 verify_command
+        # 为空时放行（200 updated），哪怕 host-state 里有 argv。
+        _insert_task(self.db_path, status="running",
+                     verify_command=None)
+        h, fake_auth = _handler(self.db_path, {"goal": "new goal"})
+        with mock.patch.object(
+                __import__("cloud.server", fromlist=["_auth_store"]),
+                "_auth_store", fake_auth):
+            h._update_goal_task("tester", "T1")
+        status, obj = h.sent[-1]
+        self.assertEqual(status, 200)
+        self.assertEqual(obj["goal_update"], "updated")
+        row = _task_row(self.db_path)
+        self.assertEqual(row["goal_version"], 1)
+        self.assertEqual(row["applied_goal_version"], 0)
+        self.assertEqual(_history(self.db_path),
+                         [("old goal", "new goal", 1, "tester")])
+
+    def test_host_loop_ignores_signal_when_argv_verify_present(self):
+        # host 第二道门：argv 验收循环忽略信号，循环按旧 goal
+        # 验收通过正常结束，但 goal/applied 一律不动，信号文件
+        # 仍在——版本差 + 残留信号就是“永不应用”的可观测证据。
+        task = {"name": "codex-abc123", "cwd": self.tmp.name,
+                "thread_id": "th1", "turns": []}
+        self.state["tasks"]["codex-abc123"] = task
+        glu.write_goal_update_signal(
+            self.state_path, "codex-abc123", 1, "new goal", "T1")
+
+        def fake_turn(server, st, t, prompt, model, effort, sandbox,
+                      timeout=None, interrupt_file=None):
+            return {"status": "completed", "text": "done",
+                    "usage": None}
+
+        with mock.patch.object(host_mod, "run_task_turn", fake_turn), \
+                mock.patch.object(
+                    host_mod, "run_verify_argv",
+                    lambda *a, **k: (0, "ok")):
+            result = host_mod.run_goal_loop(
+                SimpleNamespace(), self.state, self.state_path, task,
+                "old goal", None, 3, None, None, None, cfg=None,
+                verify_argv=["echo", "ok"])
+        self.assertTrue(result["ok"])
+        with open(self.state_path, encoding="utf-8") as fh:
+            saved = json.load(fh)
+        entry = saved["tasks"]["codex-abc123"]
+        self.assertEqual(entry["goal"], "old goal")
+        self.assertEqual(entry["applied_goal_version"], 0)
+        # 信号未被应用、也未被误删（version > 已应用版本）。
+        self.assertTrue(os.path.exists(glu.signal_path(
+            self.state_path, "codex-abc123")))
+        self.assertEqual(
+            glu.read_goal_update_signal(
+                self.state_path, "codex-abc123", 0), (1, "new goal"))
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_gapfill_g7_merge_atomic.py

@@ -0,0 +1,184 @@
+"""gapfill G7：_merge_task 状态提交与成功审计同一事务。
+
+旧写法（A4 时序残留）：先 commit 状态为 merged、再另开
+AuthStore 连接写 merge_task 审计；审计失败时状态已是
+merged、响应却是 500——“合并了却报失败”，调用方无法靠
+状态判断，也无法安全重试（重复合并 409）。
+
+新写法：audit_on_conn 把审计对（audit_log 行 + 链条目）
+写进同一连接、一次 commit；审计失败则整段回滚，状态仍是
+completed、异常上抛后由 _dispatch_safe 转 500 JSON，调用方
+可原样重试合并。链头唯一冲突的重试口径与 AuthStore.audit
+完全一致（5 次、只重试 IntegrityError，其它错误直接抛）。
+成功语义不变：200 体、审计内容、门禁响应全同；本文件第
+一项钉住成功路径与旧行为一致，后三项钉住失败路径的新行为。
+
+真实模型 0 轮，不联网；不跑真实 HTTP（handler 直调 +
+真实 AuthStore/DB，审计行直接查库验证）。
+"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+import cloud.server as server_mod  # noqa: E402
+from cloud.auth import AuthStore, audit_on_conn  # noqa: E402
+from cloud.server import CloudControlPlaneHandler, init_db  # noqa: E402
+
+NO_BLOCK = {"merge_blocked": False, "release_blocked": False,
+            "blocked_by": None, "round": None, "failed_checks": []}
+
+HARD_BODY = {
+    "files_changed": ["apps/web/a.py"],
+    "owned_dir": "apps/web",
+    "branch_protection": "protected",
+    "ci_status": "success",
+    "ci_checks": [{"name": "ci", "status": "success"}],
+    "required_checks": ["ci"],
+    "contract_version": "v2",
+    "contract_frozen": True,
+    "coordinator_approved": True,
+    "unresolved_review_comments": 0,
+}
+
+
+class MergeAtomicTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, _key = self.auth.create_user("owner")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO projects(id, name, owner_id, created_at,"
+                " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)",
+                ("p1", "P", self.owner_id, time.time(), None))
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " created_by, created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, ?, ?, ?)",
+                ("t1", "p1", "do work", "completed", self.owner_id,
+                 time.time(), time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+        self.auth.create_project("p1", self.owner_id)
+
+    def tearDown(self):
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _handler(self):
+        h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
+        h.server = SimpleNamespace(db_path=self.db_path)
+        sent = []
+        h._send_json = lambda status, obj, extra_headers=None: sent.append(
+            (status, obj))
+        h._read_json = lambda: (dict(HARD_BODY), None)
+        h.sent = sent
+        return h
+
+    def _merge(self, h):
+        with mock.patch.object(server_mod, "_pushback_block",
+                               return_value=dict(NO_BLOCK)):
+            h._merge_task(self.owner_id, "t1")
+
+    def _status(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            return conn.execute(
+                "SELECT status FROM tasks WHERE id = 't1'").fetchone()[0]
+        finally:
+            conn.close()
+
+    def _merge_audits(self):
+        return [a for a in self.auth.list_audit("p1")
+                if a["action"] == "merge_task"]
+
+    def test_success_unchanged(self):
+        # 成功语义与旧写法一致：200、merged、审计有行、门禁在体里。
+        h = self._handler()
+        self._merge(h)
+        status, obj = h.sent[-1]
+        self.assertEqual(status, 200)
+        self.assertEqual(obj["status"], "merged")
+        self.assertIn("merge_gate", obj)
+        self.assertEqual(self._status(), "merged")
+        self.assertEqual(len(self._merge_audits()), 1)
+        self.assertEqual(self._merge_audits()[0]["detail"], "t1")
+
+    def test_audit_failure_rolls_back_status(self):
+        # 旧行为是 merged + 500；新行为是 completed + 抛错
+        # （_dispatch_safe 转 500 JSON），调用方可原样重试。
+        h = self._handler()
+        with mock.patch.object(
+                server_mod, "_audit_on_conn",
+                side_effect=RuntimeError("audit down")):
+            with self.assertRaises(RuntimeError):
+                self._merge(h)
+        self.assertEqual(h.sent, [])
+        self.assertEqual(self._status(), "completed")
+        self.assertEqual(self._merge_audits(), [])
+        # 原样重试（审计恢复后）必须成功：状态仍是 completed，
+        # 不是卡在 merged 导致重复合并 409。
+        h2 = self._handler()
+        self._merge(h2)
+        self.assertEqual(h2.sent[-1][0], 200)
+        self.assertEqual(self._status(), "merged")
+
+    def test_ledger_conflict_retries_then_succeeds(self):
+        # 链头唯一冲突重试整段（与 AuthStore.audit 同口径）：
+        # 前两次冲突、第三次成功，整体仍是一次合并、一条审计。
+        calls = {"n": 0}
+
+        def flaky(conn, user_id, project_id, action, detail="",
+                  created_at=None):
+            calls["n"] += 1
+            if calls["n"] <= 2:
+                raise sqlite3.IntegrityError("chain head race")
+            return audit_on_conn(conn, user_id, project_id, action,
+                                 detail=detail, created_at=created_at)
+
+        h = self._handler()
+        with mock.patch.object(server_mod, "_audit_on_conn",
+                               side_effect=flaky):
+            self._merge(h)
+        self.assertEqual(h.sent[-1][0], 200)
+        self.assertEqual(calls["n"], 3)
+        self.assertEqual(self._status(), "merged")
+        self.assertEqual(len(self._merge_audits()), 1)
+
+    def test_ledger_conflict_exhausted_rolls_back(self):
+        # 5 次冲突耗尽：抛错、状态回滚、审计无行、不留半截合并。
+        calls = {"n": 0}
+
+        def always_conflict(*args, **kwargs):
+            calls["n"] += 1
+            raise sqlite3.IntegrityError("chain head race")
+
+        h = self._handler()
+        with mock.patch.object(server_mod, "_audit_on_conn",
+                               side_effect=always_conflict):
+            with self.assertRaises(sqlite3.IntegrityError):
+                self._merge(h)
+        self.assertEqual(calls["n"], 5)
+        self.assertEqual(self._status(), "completed")
+        self.assertEqual(self._merge_audits(), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_handler_split_conformance.py

@@ -9,10 +9,14 @@ Pins the split contract so a future violation fails CI:
   as #238 arch_conformance.scan_imports);
 - server.py re-exports the moved shared helpers as identical objects,
   so `from cloud.server import _task_to_dict` keeps working;
-- handler method accounting: 243 total, 149 left in server.py itself,
+- handler method accounting: 244 total, 150 left in server.py itself,
   94 in mixins, no name defined twice. (v5 rebase: main #275 added
   5 resident methods that legitimately stay in server.py, so the
-  stack's 238/144 becomes 243/149; moved count stays 94.)
+  stack's 238/144 becomes 243/149; moved count stays 94.
+  Reconcile: goal live-update adds _update_goal_task in server.py,
+  so 243/149 becomes 244/150; moved count still 94. The source
+  line's 239/223/16 counts its 12-mixin split and do not apply on
+  this line, which keeps the target's 6-mixin split.)
 
 Only the standard library; no real model, no network, no secrets.
 """
@@ -140,7 +144,7 @@ class TestReexports(unittest.TestCase):
 
 
 class TestMethodAccounting(unittest.TestCase):
-    def test_total_243_no_duplicates(self):
+    def test_total_244_no_duplicates(self):
         seen = {}
         dups = []
         classes = [CloudControlPlaneHandler]
@@ -155,9 +159,9 @@ class TestMethodAccounting(unittest.TestCase):
                     dups.append((seen[name], cls.__name__, name))
                 seen[name] = cls.__name__
         self.assertEqual(dups, [])
-        self.assertEqual(total, 243)
+        self.assertEqual(total, 244)
         self.assertEqual(len(_own_methods(CloudControlPlaneHandler)),
-                         243 - 94)
+                         244 - 94)
 
 
 if __name__ == "__main__":

cloud/tests/test_longfunc_inventory.py

@@ -7,9 +7,19 @@ pins the full 17-method list by AST over ``cloud/handler_*.py``
 plus ``cloud/server.py`` -- no external summary document is the
 source of truth, the code itself is.
 
+Reconcile re-pin (code-to10-reconcile): the unified line keeps
+the target's 6-mixin split, so 16 of the 17 methods live in
+``cloud/server.py`` (``CloudControlPlaneHandler``) rather than in
+the source line's lifecycle/core/admin/browser/webhook mixins;
+the method NAMES are the same 17, only their homes differ, and
+no method was split or dropped to reach this list. The
+module-level >100-line definition likewise lives in
+``cloud/server.py`` (``migrate_tasks_table``, 114 lines) on this
+line, not in ``handler_support.py``.
+
 Counting rule, stated because a naive AST count gives 18:
 the 17 are class methods. The 18th >100-line definition,
-``handler_support.migrate_tasks_table`` (module-level function,
+``server.migrate_tasks_table`` (module-level function,
 not a mixin method), is asserted separately as the ONLY
 module-level >100-line definition, so the 17/18 difference is
 explained, not hidden. No method is split by this batch:
@@ -21,20 +31,20 @@ import unittest
 from pathlib import Path
 
 EXPECTED_METHODS = {
-    ("handler_admin.py", "AdminMixin._task_pr"),
+    ("server.py", "CloudControlPlaneHandler._task_pr"),
-    ("handler_admin.py", "AdminMixin._metrics"),
+    ("server.py", "CloudControlPlaneHandler._metrics"),
-    ("handler_browser.py", "BrowserMixin._browser_stream"),
+    ("server.py", "CloudControlPlaneHandler._browser_stream"),
-    ("handler_core.py", "TaskCoreMixin._create_task"),
+    ("server.py", "CloudControlPlaneHandler._create_task"),
-    ("handler_core.py", "TaskCoreMixin._create_followup"),
+    ("server.py", "CloudControlPlaneHandler._create_followup"),
     ("handler_fullauto.py", "FullAutoMixin._full_auto_execute"),
-    ("handler_lifecycle.py", "TaskLifecycleMixin._task_thread_read"),
+    ("server.py", "CloudControlPlaneHandler._task_thread_read"),
-    ("handler_lifecycle.py", "TaskLifecycleMixin._deliver_to_thread"),
+    ("server.py", "CloudControlPlaneHandler._deliver_to_thread"),
-    ("handler_lifecycle.py", "TaskLifecycleMixin._task_wait"),
+    ("server.py", "CloudControlPlaneHandler._task_wait"),
-    ("handler_lifecycle.py", "TaskLifecycleMixin._merge_task"),
+    ("server.py", "CloudControlPlaneHandler._merge_task"),
-    ("handler_lifecycle.py", "TaskLifecycleMixin._loop_run"),
+    ("server.py", "CloudControlPlaneHandler._loop_run"),
-    ("handler_lifecycle.py", "TaskLifecycleMixin._release_task"),
+    ("server.py", "CloudControlPlaneHandler._release_task"),
-    ("handler_lifecycle.py", "TaskLifecycleMixin._continue_task"),
+    ("server.py", "CloudControlPlaneHandler._continue_task"),
-    ("handler_webhook.py", "WebhookMixin._handle_github_webhook"),
+    ("server.py", "CloudControlPlaneHandler._handle_github_webhook"),
     ("server.py", "CloudControlPlaneHandler._dispatch_task_route"),
     ("server.py", "CloudControlPlaneHandler._dispatch"),
     ("server.py", "CloudControlPlaneHandler._dispatch_project"),
@@ -74,7 +84,7 @@ class LongFuncInventoryTestCase(unittest.TestCase):
         # rather than silently dropped from the count.
         self.assertEqual(
             module_funcs,
-            {("handler_support.py", "migrate_tasks_table")})
+            {("server.py", "migrate_tasks_table")})
 
 
 if __name__ == "__main__":

cloud/tests/test_row_get_unified.py

@@ -2,16 +2,24 @@
 
 只用标准库 + 内存 SQLite，不联网。锁定点：
 - 统一身份：auto_memory._row_get 与 lifecycle._row_get 就是
-  peer_messages.row_get 本体（不是又一份复写）。
+  peer_messages.row_get 本体（不是又一份复写）；gapfill G3
+  之后 approvals、automations、marker_tools、todos、wait_ops
+  的 row_get 以及 resident.py 导入的 row_get 也是同一本体。
 - 真实行类型矩阵（None / dict / sqlite3.Row）下取值与回落
-  与三处旧版一致：现值原样回（含 0 / 空串等假值），缺列 /
+  与旧版一致：现值原样回（含 0 / 空串等假值），缺列 /
   缺键 / 空行回 default（默认 None）。
-- default 回落差异钉住：三处旧版的默认回落都是 None——
+- default 回落差异钉住：旧版的默认回落都是 None——
   auto_memory 旧版 default 参数默认 None，lifecycle 旧版无
-  default 参数一律回 None，peer 旧版回 None；统一后两模块
+  default 参数一律回 None，peer 旧版回 None；统一后各模块
   调用处仍不传 default，缺失时仍为 None，不算改语义。
 - 统一路径实测：lifecycle.snapshot_to_dict 在 dict 缺键行与
   sqlite3.Row 行上都经统一口出正确字段。
+- gapfill G3 全仓收尾：五处旧版与统一版在真实行类型下
+  语义完全相同（缺键/缺列/空行一律回 None，调用处均不传
+  default），故统一不算改语义；todos / wait_ops 模块
+  docstring 里“不 import cloud 其它模块”的表述同步修正为
+  “除行取值复用 peer_messages（本身只依赖标准库）外不
+  import cloud 其它模块”。
 """
 
 import sqlite3
@@ -20,13 +28,23 @@ import unittest
 from pathlib import Path
 
 try:
+    import cloud.approvals as approvals
     import cloud.auto_memory as auto_memory
+    import cloud.automations as automations
     import cloud.lifecycle as lifecycle
+    import cloud.marker_tools as marker_tools
+    import cloud.todos as todos
+    import cloud.wait_ops as wait_ops
     from cloud.peer_messages import row_get
 except ImportError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    import cloud.approvals as approvals
     import cloud.auto_memory as auto_memory
+    import cloud.automations as automations
     import cloud.lifecycle as lifecycle
+    import cloud.marker_tools as marker_tools
+    import cloud.todos as todos
+    import cloud.wait_ops as wait_ops
     from cloud.peer_messages import row_get
 
 
@@ -48,6 +66,25 @@ class RowGetUnifiedTestCase(unittest.TestCase):
         self.assertIs(auto_memory._row_get, row_get)
         self.assertIs(lifecycle._row_get, row_get)
 
+    def test_gapfill_five_modules_are_the_same_function(self):
+        # gapfill G3：五处私有 row_get 已删，模块属性就是统一本体；
+        # resident.py 同样从 peer_messages 导入，故一并钉住。
+        import cloud.resident as resident
+
+        self.assertIs(approvals.row_get, row_get)
+        self.assertIs(automations.row_get, row_get)
+        self.assertIs(marker_tools.row_get, row_get)
+        self.assertIs(todos.row_get, row_get)
+        self.assertIs(wait_ops.row_get, row_get)
+        self.assertIs(resident.row_get, row_get)
+        # 五处调用处均不传 default，缺失回落仍是 None。
+        for fn in (approvals.row_get, automations.row_get,
+                   marker_tools.row_get, todos.row_get,
+                   wait_ops.row_get):
+            self.assertIsNone(fn({}, "nope"))
+            self.assertIsNone(fn(None, "nope"))
+            self.assertEqual(fn({"a": 0}, "a"), 0)
+
     def test_real_row_type_matrix(self):
         full, partial = _sqlite_row()
         # 现值原样回（含假值 0，不被当成缺失）。

cloud/tests/test_v6_a9_caps_quota.py

@@ -17,18 +17,30 @@
 只加测试，不改生产代码。
 """
 
+import os
 import sys
 import unittest
 from datetime import datetime, timezone
 from pathlib import Path
+from unittest import mock
 
 try:
-    from cloud.caps import running_count
+    from cloud.caps import DEFAULT_CAPACITY, resolve_capacity, running_count
-    from cloud.quota import monthly_usage, utc_month
+    from cloud.quota import (
+        DEFAULT_QUOTA_USD_MICROS,
+        default_quota_usd_micros,
+        monthly_usage,
+        utc_month,
+    )
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
-    from cloud.caps import running_count
+    from cloud.caps import DEFAULT_CAPACITY, resolve_capacity, running_count
-    from cloud.quota import monthly_usage, utc_month
+    from cloud.quota import (
+        DEFAULT_QUOTA_USD_MICROS,
+        default_quota_usd_micros,
+        monthly_usage,
+        utc_month,
+    )
 
 
 class CapsRunningCountFallbackTestCase(unittest.TestCase):
@@ -46,6 +58,45 @@ class CapsRunningCountFallbackTestCase(unittest.TestCase):
         self.assertEqual(running_count({"a": {}, "b": {}}), 2)
 
 
+class _RaisingEnv:
+    """get 一律抛异常的环境替身：模拟 env 映射本身坏掉时
+    resolve_capacity 的 except 分支（caps.py:54-55）。"""
+
+    def get(self, key):
+        raise RuntimeError("env broken")
+
+
+class CapsResolveCapacityEnvErrorTestCase(unittest.TestCase):
+    def test_raising_env_falls_back_to_default(self):
+        # caps.py:54-55：source.get 抛异常 -> raw=None -> 默认上限，
+        # 异常不得漏到调用方。
+        self.assertEqual(resolve_capacity(None, env=_RaisingEnv()),
+                         DEFAULT_CAPACITY)
+        self.assertEqual(DEFAULT_CAPACITY, 1)
+
+    def test_normal_env_still_parses(self):
+        # 回落分支不得改变正常解析语义。
+        self.assertEqual(
+            resolve_capacity(None, env={"CLOUD_WORKER_CAPACITY": "4"}), 4)
+
+
+class QuotaDefaultIllegalEnvTestCase(unittest.TestCase):
+    def test_illegal_env_value_falls_back_to_default(self):
+        # quota.py:73-74：int(非法串) 抛 ValueError -> 回落 $10
+        #（DEFAULT_QUOTA_USD_MICROS 微美元），不得把异常漏出。
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_DEFAULT_QUOTA_USD_MICROS": "not-a-number"}):
+            self.assertEqual(default_quota_usd_micros(),
+                             DEFAULT_QUOTA_USD_MICROS)
+        self.assertEqual(DEFAULT_QUOTA_USD_MICROS, 10000000)
+
+    def test_valid_env_value_still_parses(self):
+        # 回落分支不得改变正常解析语义。
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_DEFAULT_QUOTA_USD_MICROS": "250"}):
+            self.assertEqual(default_quota_usd_micros(), 250)
+
+
 class QuotaUtcMonthTestCase(unittest.TestCase):
     def test_utc_month_direct_call(self):
         # quota.py:79-80：显式 now 直接格式化，不走 now=None 分支。

cloud/tests/test_websearch_paid_freeze.py

@@ -7,7 +7,6 @@ an error string or a tool result would be caught by assertion.
 """
 
 import json
-import os
 import socket
 import sys
 import unittest

cloud/tests/test_worker_playwright_seed_f821.py

@@ -71,6 +71,22 @@ class WorkerPlaywrightSeedF821TestCase(unittest.TestCase):
         self.assertTrue(any("projA" in line and "playwright_mcp.py" in line
                             for line in logs.output))
 
+    def test_skipped_modified_script_logs_project_id_without_name_error(self):
+        # Sibling of the refreshed branch: the skipped-modified log
+        # line still used task.get("project_id") with no task in scope
+        # (ruff F821), raising NameError whenever seeding reported a
+        # tenant-modified script. Fixed to project_id, same pattern
+        # as the refreshed branch above.
+        worker = self._worker()
+        with mock.patch("cloud.worker._ensure_playwright_mcp",
+                        return_value={"script_skipped_modified":
+                                      ["playwright_mcp.py"]}):
+            with self.assertLogs(worker.logger, level="WARNING") as logs:
+                dirs = worker._tenant_dirs("projA")
+        self.assertIn("codex_home", dirs)
+        self.assertTrue(any("projA" in line and "playwright_mcp.py" in line
+                            for line in logs.output))
+
     def test_no_refresh_still_seeds(self):
         worker = self._worker()
         with mock.patch("cloud.worker._ensure_playwright_mcp",

cloud/todos.py

@@ -27,8 +27,9 @@ pending / in_progress / blocked / completed / cancelled，
   一个项目里多人并行，多项 in_progress 是正常态，强制
   单一会把板子变成单人队列，所以这里不做互斥。
 
-本模块只依赖标准库，不 import cloud 其它模块，避免循环
+本模块除行取值复用 cloud.peer_messages（该模块本身只依赖
-import；表迁移写法仿 automations（CREATE TABLE IF NOT
+标准库，故不产生循环 import）外不 import cloud 其它模块；
+表迁移写法仿 automations（CREATE TABLE IF NOT
 EXISTS + 查列补齐，SQLite 与 Postgres 适配层同款：调用方
 仍写 ? 占位符，db.py 适配层转 PG 方言）。
 """
@@ -36,6 +37,8 @@ EXISTS + 查列补齐，SQLite 与 Postgres 适配层同款：调用方
 import time
 import uuid
 
+from cloud.peer_messages import row_get
+
 STATUSES = ("pending", "in_progress", "blocked", "completed", "cancelled")
 PRIORITIES = ("low", "medium", "high")
 MAX_TITLE_LEN = 500
@@ -47,13 +50,6 @@ TODO_COLUMNS = (
     "updated_at", "completed_at")
 
 
-def row_get(row, name):
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return None
-
-
 def _row_dict(row, columns):
     if row is None or isinstance(row, dict):
         return row

cloud/wait_ops.py

@@ -1,7 +1,8 @@
 """线程等待与移交操作（对照 codex app 的 8 项工具面）。
 
-本模块只做控制面纯逻辑，不跑模型、不联网、不 import cloud 其它
+本模块只做控制面纯逻辑，不跑模型、不联网；行取值复用
-模块（与 automations / peer_messages 同款，避免循环 import）；
+cloud.peer_messages（该模块本身只依赖标准库，故不产生循环
+import），其余不 import cloud 其它模块；
 server.py 只负责鉴权、路由与错误码映射。
 
 8 项在控制面的落点（详见 docs/thread-wait-ops.md 对照表）：
@@ -40,6 +41,8 @@ import re
 import time
 import uuid
 
+from cloud.peer_messages import row_get
+
 HANDOFF_STATUSES = ("pending", "accepted", "running",
                     "completed", "failed")
 # report 允许的状态转移：终态（completed / failed）不可再动；
@@ -74,16 +77,6 @@ SIGNIFICANT_KINDS = ("terminal", "needs_approval", "needs_input",
                      "needs_continue")
 
 
-def row_get(row, name):
-    """行取值容错：sqlite3.Row 缺列报 IndexError、dict 缺键报
-    KeyError、空行报 TypeError，一律回 None（与 automations
-    的 row_get 同款，老库行缺列时读移交列表不能因此炸掉）。"""
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return None
-
-
 def _row_dict(row, columns):
     if row is None or isinstance(row, dict):
         return row

cloud/worker.py

@@ -3026,10 +3026,14 @@ class Worker:
                     project_id,
                     ",".join(_seeded["script_refreshed"]))
             if _seeded.get("script_skipped_modified"):
+                # Sibling of the refreshed branch above: _tenant_dirs
+                # only receives project_id, there is no task in scope;
+                # the pre-existing task.get() here raised NameError
+                # whenever seeding reported a tenant-modified script.
                 self.logger.warning(
                     "playwright MCP scripts skipped (tenant-modified, "
                     "left untouched, see audit) for project %s: %s",
-                    task.get("project_id"),
+                    project_id,
                     ",".join(_seeded["script_skipped_modified"]))
         if self.seed_skills:
             # 技能播种失败绝不阻塞任务：技能是可选的流程