code-to10 v4 coverage thickening: 219 new assertion tests (batches A/B1/C/D1-D6)

Baseline merge head e5ff26a: 28144 stmts, 2807 missing, exact 90.0263%, margin +7.
After thickening (full run): 2350 missing, exact 91.6501%, margin +464.
Full suite 3516 items: 3509 pass, 1 failure (pre-existing November date
failure, untouched), 6 skipped. No production code changed.

cloud/tests/test_v2cov_gaps_a.py

@@ -0,0 +1,564 @@
+"""Coverage batch v4-A: evidence modules, handler close/502 branches,
+server defensive branches, transplant v05 newline branch.
+
+Targets are the baseline-missing lines of the v4 merge head, verified
+against the subset-coverage intersection (see the v4 summary). Stubbing
+discipline: the method under test is never stubbed. Collaborators that
+are stubbed are: CostStore / _validate_sub2api_key /
+_list_sub2api_models (handler_identity namespace -- the v2 batch patched
+cloud.server by mistake after the split, which is why these lines stayed
+uncovered), _check_project / _project_exists predicates for the two
+project 404 branches (unreachable through the real check, which 404s
+earlier when the project row is absent), parse_paths_field for the
+upload count-mismatch branch (the real parser raises first, so the
+handler's own duplicate check is defensive), os.environ.get for the
+_default_executor None branch (a real environ value can never be
+None), and the _connect/_init/_ensure helpers for the init_db
+Postgres branch (no live PG in this environment).
+"""
+
+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 cloud.server as server_mod
+from cloud import evidence_binding
+from cloud import gate_verification
+from cloud import ledger_onboarding
+from cloud import ledger_outbox
+from cloud.evidence_binding import (
+    EvidenceNonceStore,
+    bind_evidence,
+    binding_presence,
+    validate_bound_evidence,
+)
+from cloud.gate_verification import verify_gate_evidence
+from cloud.ledger_onboarding import _table_columns, verify_ledger_ready
+from cloud.ledger_outbox import (
+    append_failure,
+    outbox_path_for,
+    read_failures,
+    replay_failures,
+)
+from cloud.server import CloudControlPlaneHandler, init_db
+
+
+def _handler(db_path):
+    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.sent = sent
+    return h
+
+
+def _conn(db_path):
+    c = sqlite3.connect(db_path)
+    c.row_factory = sqlite3.Row
+    return c
+
+
+class _Base(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 _h(self, body=None, err=None):
+        h = _handler(self.db_path)
+        h._read_json = lambda: (body, err)
+        return h
+
+    def _project(self, pid="p1", owner="u1"):
+        from cloud.server import _auth_store
+        c = _conn(self.db_path)
+        c.execute(
+            "INSERT INTO projects(id, name, owner_id, created_at,"
+            " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)",
+            (pid, "P", owner, time.time(), None))
+        c.commit()
+        c.close()
+        with _auth_store(self.db_path) as store:
+            store.create_project(pid, owner)
+
+    def _task(self, tid="t1", pid="p1", status="completed"):
+        c = _conn(self.db_path)
+        c.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " created_by, created_at, updated_at)"
+            " VALUES(?, ?, ?, ?, ?, ?, ?)",
+            (tid, pid, "do work", status, "u1", time.time(),
+             time.time()))
+        c.commit()
+        c.close()
+
+    def _last(self, h):
+        self.assertTrue(h.sent, "handler sent no response")
+        return h.sent[-1]
+
+
+class EvidenceBindingGapTestCase(unittest.TestCase):
+    def test_bind_non_dict_raises(self):
+        with self.assertRaises(ValueError):
+            bind_evidence(["not", "a", "dict"])
+
+    def test_validate_non_dict(self):
+        self.assertEqual(validate_bound_evidence("x"),
+                         (False, "evidence_binding_missing"))
+
+    def test_validate_both_missing(self):
+        self.assertEqual(validate_bound_evidence({}),
+                         (False, "evidence_binding_missing"))
+
+    def test_validate_bool_ts(self):
+        self.assertEqual(
+            validate_bound_evidence(
+                {"evidence_ts": True, "evidence_nonce": "n1"}),
+            (False, "evidence_ts_not_a_number"))
+
+    def test_validate_blank_nonce(self):
+        self.assertEqual(
+            validate_bound_evidence(
+                {"evidence_ts": 1000.0, "evidence_nonce": "   "}),
+            (False, "evidence_nonce_empty"))
+
+    def test_binding_presence_non_dict(self):
+        self.assertEqual(binding_presence(None), "absent")
+
+    def test_has_exception_returns_false(self):
+        class _RaisingConn:
+            def execute(self, *args):
+                raise RuntimeError("db gone")
+        store = EvidenceNonceStore(_RaisingConn())
+        self.assertFalse(store.has("n1"))
+
+
+class GateVerificationGapTestCase(unittest.TestCase):
+    def _live(self, ci_checks):
+        return {"ci_checks": ci_checks, "branch_protection": "protected"}
+
+    def test_reader_non_dict_result(self):
+        out = verify_gate_evidence(
+            {}, reader=lambda repo, ref: ["not", "a", "dict"],
+            repo="r", ref="s")
+        self.assertFalse(out["checked"])
+        self.assertFalse(out["verified"])
+        self.assertIn("non-dict", out["reason"])
+
+    def test_live_checks_non_dict_treated_empty(self):
+        out = verify_gate_evidence(
+            {}, reader=lambda repo, ref: self._live("nope"),
+            repo="r", ref="s")
+        self.assertTrue(out["checked"])
+        self.assertTrue(out["verified"])
+
+    def test_ci_checks_non_dict_item_skipped(self):
+        out = verify_gate_evidence(
+            {"ci_checks": ["oops", {"name": "ci", "status": "success"}]},
+            reader=lambda repo, ref: self._live({"ci": "success"}),
+            repo="r", ref="s")
+        self.assertTrue(out["checked"])
+        self.assertTrue(out["verified"])
+        self.assertEqual(out["mismatches"], [])
+
+    def test_ci_checks_nameless_item_skipped(self):
+        out = verify_gate_evidence(
+            {"ci_checks": [{"status": "success"}]},
+            reader=lambda repo, ref: self._live({}),
+            repo="r", ref="s")
+        self.assertTrue(out["checked"])
+        self.assertTrue(out["verified"])
+
+
+class _FakeConn:
+    def __init__(self, rows=None, exc=None):
+        self._rows = rows
+        self._exc = exc
+
+    def execute(self, sql, params=()):
+        if self._exc is not None:
+            raise self._exc
+        result = mock.Mock()
+        result.fetchall.return_value = self._rows
+        return result
+
+
+class LedgerOnboardingGapTestCase(unittest.TestCase):
+    def test_table_columns_exception_returns_none(self):
+        self.assertIsNone(_table_columns(
+            _FakeConn(exc=RuntimeError("pg dialect")), "task_attempts"))
+
+    def test_table_columns_malformed_rows_return_none(self):
+        # Row tuples too short for row[1]: IndexError -> None.
+        self.assertIsNone(_table_columns(_FakeConn(rows=[(1,)]),
+                                         "task_attempts"))
+
+    def _sqlite(self, ledger_cols, marker_cols):
+        conn = sqlite3.connect(":memory:")
+        if ledger_cols is not None:
+            conn.execute("CREATE TABLE task_attempts(%s)"
+                         % ", ".join("%s TEXT" % c for c in ledger_cols))
+        if marker_cols is not None:
+            conn.execute("CREATE TABLE task_attempt_origin_markers(%s)"
+                         % ", ".join("%s TEXT" % c for c in marker_cols))
+        return conn
+
+    def test_verify_ledger_missing_column(self):
+        conn = self._sqlite(
+            ["task_id", "attempt_no", "outcome", "is_defect",
+             "defect_category", "created_at"],
+            ["task_id", "origin", "queued_at"])
+        try:
+            out = verify_ledger_ready(conn)
+        finally:
+            conn.close()
+        self.assertFalse(out["tables_ready"])
+        self.assertEqual(
+            out["missing"],
+            ["task_attempts missing columns: corrects_attempt_no"])
+
+    def test_verify_marker_missing_column(self):
+        conn = self._sqlite(
+            ["task_id", "attempt_no", "outcome", "is_defect",
+             "defect_category", "created_at", "corrects_attempt_no"],
+            ["task_id", "origin"])
+        try:
+            out = verify_ledger_ready(conn)
+        finally:
+            conn.close()
+        self.assertFalse(out["tables_ready"])
+        self.assertEqual(
+            out["missing"],
+            ["task_attempt_origin_markers missing columns: queued_at"])
+
+
+class LedgerOutboxGapTestCase(unittest.TestCase):
+    def test_outbox_path_explicit_env_wins(self):
+        self.assertEqual(
+            outbox_path_for("/x/db.sqlite",
+                            env={"CLOUD_LEDGER_FAILURE_OUTBOX": "/o/f.jsonl"}),
+            "/o/f.jsonl")
+
+    def test_outbox_path_memory_and_empty_none(self):
+        self.assertIsNone(outbox_path_for(":memory:", env={}))
+        self.assertIsNone(outbox_path_for("", env={}))
+
+    def test_append_failure_falsy_path(self):
+        self.assertFalse(append_failure("", "t1", "failure", "boom"))
+        self.assertFalse(append_failure(None, "t1", "failure", "boom"))
+
+    def test_read_failures_missing_file(self):
+        self.assertEqual(read_failures("/definitely/not/here.jsonl"),
+                         ([], 0))
+
+    def test_read_failures_bad_lines_counted(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "out.jsonl")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write('{"not json\n')
+                fh.write('\n')
+                fh.write(json.dumps({"task_id": "t1",
+                                     "outcome": "weird"}) + "\n")
+                fh.write(json.dumps([1, 2]) + "\n")
+                fh.write(json.dumps({"task_id": "t2",
+                                     "outcome": "success"}) + "\n")
+            records, bad = read_failures(path)
+        self.assertEqual([r["task_id"] for r in records], ["t2"])
+        self.assertEqual(bad, 3)
+
+    def test_replay_failed_record_kept_and_rewritten(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "out.jsonl")
+            self.assertTrue(append_failure(path, "t9", "failure", "db"))
+            def _boom(conn, task_id, outcome, is_defect=False):
+                raise RuntimeError("still down")
+            stats = replay_failures(path, None, record_fn=_boom)
+            self.assertEqual(stats["replayed"], 0)
+            self.assertEqual(stats["failed"], 1)
+            self.assertEqual(stats["remaining"], 1)
+            self.assertEqual(stats["bad_lines"], 0)
+            records, bad = read_failures(path)
+            self.assertEqual([r["task_id"] for r in records], ["t9"])
+            self.assertEqual(bad, 0)
+
+    def test_replay_directory_path_write_failure_swallowed(self):
+        with tempfile.TemporaryDirectory() as d:
+            stats = replay_failures(d, None,
+                                    record_fn=lambda *a, **k: None)
+        self.assertEqual(stats, {"replayed": 0, "failed": 0,
+                                 "remaining": 0, "bad_lines": 0})
+
+    def test_replay_bad_lines_not_rewritten(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "out.jsonl")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write('{"broken\n')
+                fh.write(json.dumps({"task_id": "t3",
+                                     "outcome": "success"}) + "\n")
+            seen = []
+            def _ok(conn, task_id, outcome, is_defect=False):
+                seen.append(task_id)
+            stats = replay_failures(path, None, record_fn=_ok)
+            self.assertEqual(stats["replayed"], 1)
+            self.assertEqual(stats["bad_lines"], 1)
+            self.assertEqual(stats["remaining"], 1)
+            self.assertEqual(seen, ["t3"])
+            with open(path, encoding="utf-8") as fh:
+                self.assertIn('{"broken', fh.read())
+
+
+class HandlerGapTestCase(_Base):
+    def test_project_cost_close_failure_swallowed(self):
+        class _Store:
+            def __init__(self, db_path):
+                self._conn = mock.Mock()
+                self._conn.close.side_effect = RuntimeError("down")
+
+            def dual_column_report_by_project(self, project_id):
+                return {}
+        h = self._h()
+        with mock.patch("cloud.handler_identity.CostStore", _Store):
+            h._project_cost("p1", {})
+        self.assertEqual(self._last(h)[0], 200)
+
+    def test_task_cost_close_failure_swallowed(self):
+        self._project()
+        self._task()
+        class _Store:
+            def __init__(self, db_path):
+                self._conn = mock.Mock()
+                self._conn.close.side_effect = RuntimeError("down")
+
+            def dual_column_report_by_task(self, task_id):
+                return {}
+        h = self._h()
+        with mock.patch("cloud.handler_identity.CostStore", _Store):
+            h._task_cost("u1", "t1")
+        self.assertEqual(self._last(h)[0], 200)
+
+    def test_relay_models_sub2api_generic_error_502(self):
+        h = self._h()
+        with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \
+                mock.patch("cloud.handler_identity._validate_sub2api_key",
+                           side_effect=RuntimeError("net down")):
+            h._relay_models("u1")
+        self.assertEqual(self._last(h)[0], 502)
+
+    def test_relay_models_sub2api_list_error_502(self):
+        h = self._h()
+        with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \
+                mock.patch("cloud.handler_identity._validate_sub2api_key",
+                           return_value=("ok", "key", "sub", "kid")), \
+                mock.patch("cloud.handler_identity._list_sub2api_models",
+                           side_effect=RuntimeError("gw down")):
+            h._relay_models("u1")
+        self.assertEqual(self._last(h)[0], 502)
+
+    def test_models_overview_close_failure_swallowed(self):
+        self._project()
+        conn = mock.Mock()
+        conn.execute.return_value.fetchall.return_value = []
+        conn.close.side_effect = RuntimeError("down")
+
+        class _Store:
+            def __init__(self, db_path):
+                self._conn = conn
+        h = self._h()
+        with mock.patch("cloud.handler_identity.CostStore", _Store):
+            h._models_overview("u1")
+        self.assertEqual(self._last(h)[0], 200)
+
+    def test_create_environment_project_row_gone_404(self):
+        h = self._h(body={"project_id": "pX", "name": "env1"})
+        h._check_project = lambda *args: None
+        h._create_environment("u1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 404)
+        self.assertEqual(payload["error"], "project not found")
+
+    def test_update_project_row_gone_404(self):
+        h = self._h(body={"repo_url": ""})
+        h._check_project = lambda *args: None
+        h._update_project("u1", "pX")
+        code, payload = self._last(h)
+        self.assertEqual(code, 404)
+        self.assertEqual(payload["error"], "project not found")
+
+
+class ServerGapTestCase(_Base):
+    def test_read_json_empty_body(self):
+        h = _handler(self.db_path)
+        h._read_body = lambda: b""
+        self.assertEqual(h._read_json(), ({}, None))
+
+    def test_default_executor_none_env_value(self):
+        real_get = os.environ.get
+
+        def _fake_get(key, default=None):
+            if key == "CLOUD_DEFAULT_EXECUTOR":
+                return None
+            return real_get(key, default)
+        with mock.patch.object(os.environ, "get", side_effect=_fake_get):
+            self.assertEqual(server_mod._default_executor(), "docker")
+
+    def _backfill_conn(self, mode, rollback_raises=False):
+        class _Fake:
+            def __init__(self):
+                self.rollback_calls = 0
+                self.rollback_raises = rollback_raises
+                self.mode = mode
+
+            def execute(self, sql, params=()):
+                text = sql.strip()
+                if text.startswith("ROLLBACK TO"):
+                    raise RuntimeError("rollback-to down")
+                if text.startswith("SAVEPOINT"):
+                    if self.mode == "savepoint":
+                        raise RuntimeError("savepoint down")
+                    return mock.Mock()
+                if "INSERT OR IGNORE INTO user_quotas" in text:
+                    raise RuntimeError("insert down")
+                return mock.Mock()
+
+            def rollback(self):
+                self.rollback_calls += 1
+                if self.rollback_raises:
+                    raise RuntimeError("rollback down")
+        return _Fake()
+
+    def test_backfill_savepoint_failure_rollback_ok(self):
+        conn = self._backfill_conn("savepoint")
+        self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
+        self.assertEqual(conn.rollback_calls, 1)
+
+    def test_backfill_savepoint_failure_rollback_raises(self):
+        conn = self._backfill_conn("savepoint", rollback_raises=True)
+        self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
+        self.assertEqual(conn.rollback_calls, 1)
+
+    def test_backfill_insert_failure_rollback_to_failure(self):
+        conn = self._backfill_conn("insert")
+        self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
+        self.assertEqual(conn.rollback_calls, 1)
+
+    def test_backfill_insert_failure_rollback_raises(self):
+        conn = self._backfill_conn("insert", rollback_raises=True)
+        self.assertEqual(server_mod._backfill_default_quotas(conn), 0)
+        self.assertEqual(conn.rollback_calls, 1)
+
+    def test_init_db_postgres_branch(self):
+        conn = mock.Mock()
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_DB_BACKEND": "postgres"}), \
+                mock.patch("cloud.server._connect",
+                           return_value=conn), \
+                mock.patch("cloud.server._init_postgres_schema"), \
+                mock.patch("cloud.server.migrate_tasks_table"), \
+                mock.patch("cloud.server.migrate_projects_table"), \
+                mock.patch("cloud.server._ensure_followups_table"), \
+                mock.patch("cloud.server._ensure_continue_messages_table"), \
+                mock.patch("cloud.server._ensure_environments_table"), \
+                mock.patch("cloud.server._ensure_quotas_table"), \
+                mock.patch("cloud.server._backfill_default_quotas"), \
+                mock.patch("cloud.server._ensure_gateway_keys"), \
+                mock.patch("cloud.server._ensure_uploads_tables"), \
+                mock.patch("cloud.server._ensure_task_checkpoints_table"), \
+                mock.patch("cloud.server._ensure_task_steps_table"):
+            init_db("/nonexistent/pg.db")
+        conn.commit.assert_called_once_with()
+
+    def _multipart(self, boundary, fields, files):
+        out = b""
+        for name, value in fields.items():
+            out += (("--%s\r\nContent-Disposition: form-data; "
+                     "name=\"%s\"\r\n\r\n%s\r\n")
+                    % (boundary, name, value)).encode("utf-8")
+        for part_name, filename, data in files:
+            out += (("--%s\r\nContent-Disposition: form-data; "
+                     "name=\"%s\"; filename=\"%s\"\r\n\r\n")
+                    % (boundary, part_name, filename)).encode("utf-8")
+            out += data + b"\r\n"
+        out += ("--%s--\r\n" % boundary).encode("utf-8")
+        return out
+
+    def test_create_upload_paths_files_mismatch(self):
+        boundary = "TESTBOUNDARY"
+        raw = self._multipart(
+            boundary, {"paths": '["a.txt", "b.txt"]'},
+            [("file0", "a.txt", b"hello"), ("file1", "b.txt", b"world")])
+        h = _handler(self.db_path)
+        h.headers = {"Content-Type":
+                     "multipart/form-data; boundary=" + boundary,
+                     "Content-Length": str(len(raw))}
+        h._read_body = lambda: raw
+        with mock.patch("cloud.server.parse_paths_field",
+                        return_value=["only-one.txt"]):
+            h._create_upload("u1")
+        code, payload = h.sent[-1]
+        self.assertEqual(code, 400)
+        self.assertIn("不一致", payload["error"])
+
+    def test_release_task_bound_bad_ts_400(self):
+        self._task()
+        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,
+            "evidence_ts": "not-a-number",
+            "evidence_nonce": "n1",
+        }
+        h = self._h(body=body)
+        with mock.patch.dict(os.environ,
+                             {"CODEX_GLOBAL_ADMINS": "admin1"}):
+            h._release_task("admin1", "t1")
+        code, payload = self._last(h)
+        self.assertEqual(code, 400)
+        self.assertEqual(payload["error"], "evidence_ts_not_a_number")
+
+
+class TransplantGapTestCase(unittest.TestCase):
+    def test_transform_v05_notify_with_trailing_newline(self):
+        from importlib import util as _util
+        script = (REPO_ROOT / "cloud" / "skills" / "software-engineering"
+                  / "tools" / "check_transplant.py")
+        spec = _util.spec_from_file_location("check_transplant_v4a",
+                                             str(script))
+        module = _util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        lines = [
+            "前言。" + module.AUDIT_WARNING,
+            module.TIGHTEN3_NOTIFY,
+            "filler line after notify",
+            module.QA_HEADER,
+            "qa body",
+        ]
+        out_lines, stopped = module.transform_main(lines, "v05")
+        text = "\n".join(out_lines)
+        self.assertNotIn(module.TIGHTEN3_NOTIFY, text)
+        self.assertTrue(any("QA 全链路" in s for s in stopped))
+        self.assertTrue(any("收紧三" in s for s in stopped))
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_v2cov_gaps_b1.py

@@ -0,0 +1,540 @@
+"""Coverage batch v4-B1: cloud/worker.py small pure/error branches.
+
+Stubbing discipline: the method under test is never stubbed. Patched
+collaborators are: register_worker / heartbeat_worker (registry sinks),
+AuthStore (audit sink), AutoMemoryStore / _auto_prefix / _todo_list_db /
+_db_connect (data collaborators), log_event (only to drive the
+log-failure swallow branches), os.killpg (to drive the kill-error
+branches without depending on a real process group dying mid-test),
+and instance-level _host_state_path_for_task / _rebuild... /
+_task_never_started / _fetch_environment predicates where the real
+path cannot produce the branch without a race or a missing row.
+Every assertion reads a real return value or a real DB row.
+"""
+
+import os
+import signal
+import sqlite3
+import subprocess
+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))
+
+from cloud.server import init_db
+from cloud.worker import Worker, _code_signature
+
+FAKE_ENV = {"XCM_API_KEY": "test-placeholder"}
+
+
+def _worker(db_path, **kwargs):
+    return Worker(db_path, "/fake/host.py", env=FAKE_ENV, **kwargs)
+
+
+class _Base(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)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _insert_task(self, task_id="t1", status="queued", **cols):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " created_by, created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, ?, ?, ?)",
+                (task_id, "p1", "do work", status, "u1", time.time(),
+                 time.time()))
+            for key, value in cols.items():
+                conn.execute(
+                    "UPDATE tasks SET %s = ? WHERE id = ?" % key,
+                    (value, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _row(self, task_id="t1"):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.row_factory = sqlite3.Row
+            return conn.execute(
+                "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
+        finally:
+            conn.close()
+
+
+class InitAndRegistryTestCase(_Base):
+    def test_code_signature_stat_failure_none(self):
+        with mock.patch.object(os, "stat", side_effect=OSError("gone")):
+            self.assertIsNone(_code_signature("/nonexistent/host.py"))
+
+    def test_init_router_type_error(self):
+        with self.assertRaises(ValueError):
+            _worker(self.db_path, router="not-a-router")
+
+    def test_init_fallback_models_normalized(self):
+        w = _worker(self.db_path, model="m0",
+                    fallback_models=[123, " m1 ", "m1", "m0", "m2"])
+        self.assertEqual(w.fallback_models, ["m1", "m2"])
+
+    def test_init_executor_budget_flag_failure_swallowed(self):
+        class _BadExecutor:
+            @property
+            def ignore_host_budget(self):
+                return False
+
+            @ignore_host_budget.setter
+            def ignore_host_budget(self, value):
+                raise RuntimeError("read-only")
+        w = _worker(self.db_path, executor=_BadExecutor())
+        self.assertIsInstance(w.executor, _BadExecutor)
+
+    def test_register_failure_swallowed(self):
+        w = _worker(self.db_path)
+        with mock.patch("cloud.worker.register_worker",
+                        side_effect=RuntimeError("db down")):
+            w.register()
+
+    def test_registry_heartbeat_failure_swallowed(self):
+        w = _worker(self.db_path)
+        with mock.patch("cloud.worker.heartbeat_worker",
+                        side_effect=RuntimeError("db down")):
+            w._registry_heartbeat()
+
+    def test_close_cost_store_failure_swallowed(self):
+        w = _worker(self.db_path)
+        store = mock.Mock()
+        store._conn.close.side_effect = RuntimeError("down")
+        w._cost_store = store
+        w.close()
+
+
+class StepAndCheckpointTestCase(_Base):
+    def test_mark_interrupted_reason_db_failure_swallowed(self):
+        w = _worker(self.tmp.name)
+        w._mark_interrupted_reason("t1", "quota")
+
+    def test_record_step_conn_failure_swallowed(self):
+        w = _worker(self.db_path)
+        conn = mock.Mock()
+        conn.execute.side_effect = RuntimeError("db down")
+        w._record_step(conn, "t1", "round", "hello")
+
+    def test_checkpoint_round_real_and_failure(self):
+        w = _worker(self.db_path)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS task_checkpoints("
+                "task_id TEXT, round_no INTEGER, thread_id TEXT,"
+                " goal_status TEXT, state_sha TEXT, verify_out TEXT,"
+                " created_at DOUBLE PRECISION,"
+                " PRIMARY KEY(task_id, round_no))")
+            conn.execute("INSERT INTO task_checkpoints VALUES"
+                         "('t1', 2, 'th', 'ok', 'sha', '', 0)")
+            conn.execute("INSERT INTO task_checkpoints VALUES"
+                         "('t1', 5, 'th', 'ok', 'sha', '', 0)")
+            conn.commit()
+            self.assertEqual(w._checkpoint_round(conn, "t1"), 5)
+            self.assertEqual(w._checkpoint_round(conn, "nobody"), -1)
+        finally:
+            conn.close()
+        bad = mock.Mock()
+        bad.execute.side_effect = RuntimeError("db down")
+        self.assertEqual(w._checkpoint_round(bad, "t1"), -1)
+
+    def test_checkpoint_mirror_loop_exception_swallowed(self):
+        w = _worker(self.db_path)
+        calls = []
+        w._mirror_checkpoint_once = lambda task: (_ for _ in ()).throw(
+            RuntimeError("mirror down"))
+        class _Stop:
+            def __init__(self):
+                self.n = 0
+
+            def wait(self, timeout):
+                self.n += 1
+                return self.n >= 2
+        w._checkpoint_mirror_loop({"id": "t1"}, _Stop())
+
+    def test_host_state_path_mounts_tuple(self):
+        w = _worker(self.db_path)
+        w.executor = SimpleNamespace(extra_mounts=[("/srv/work",
+                                                    "/app/work")],
+                                     config_path=None)
+        self.assertEqual(
+            w._host_state_path_for_task({"project_id": "p1", "id": "t1"}),
+            "/srv/work/host-state.json")
+
+    def test_host_state_path_non_tuple_mount_and_none_task(self):
+        w = _worker(self.db_path)
+        w.executor = SimpleNamespace(extra_mounts=[123], config_path=None)
+        self.assertEqual(
+            w._host_state_path_for_task({"project_id": "p1", "id": "t1"}),
+            "/opt/codex-app-server/work/host-state.json")
+        self.assertIsNone(w._host_state_path_for_task(None))
+
+    def test_mirror_checkpoint_once_bad_json(self):
+        w = _worker(self.db_path)
+        work = os.path.join(self.tmp.name, "work")
+        os.makedirs(work, exist_ok=True)
+        with open(os.path.join(work, "host-state.json"), "w") as fh:
+            fh.write("{not json")
+        w.executor = SimpleNamespace(extra_mounts=[(work, "/app/work")],
+                                     config_path=None)
+        self.assertEqual(
+            w._mirror_checkpoint_once({"project_id": "p1", "id": "t1"}), -1)
+
+    def test_import_host_steps_task_none_swallowed(self):
+        w = _worker(self.db_path)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            w._import_host_steps(conn, None, "/tmp/x/host-state.json")
+        finally:
+            conn.close()
+
+    def test_rebuild_trow_missing(self):
+        w = _worker(self.db_path)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS task_checkpoints("
+                "task_id TEXT, round_no INTEGER, thread_id TEXT,"
+                " goal_status TEXT, state_sha TEXT, verify_out TEXT,"
+                " created_at DOUBLE PRECISION,"
+                " PRIMARY KEY(task_id, round_no))")
+            conn.execute("INSERT INTO task_checkpoints VALUES"
+                         "('ghost', 3, 'th', 'ok', 'sha', '', 0)")
+            conn.commit()
+        finally:
+            conn.close()
+        self.assertFalse(w._rebuild_host_state_from_checkpoint(
+            {"id": "ghost", "project_id": "p1"}))
+
+    def test_rebuild_state_path_none(self):
+        self._insert_task()
+        w = _worker(self.db_path)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS task_checkpoints("
+                "task_id TEXT, round_no INTEGER, thread_id TEXT,"
+                " goal_status TEXT, state_sha TEXT, verify_out TEXT,"
+                " created_at DOUBLE PRECISION,"
+                " PRIMARY KEY(task_id, round_no))")
+            conn.execute("INSERT INTO task_checkpoints VALUES"
+                         "('t1', 3, 'th', 'ok', 'sha', '', 0)")
+            conn.commit()
+        finally:
+            conn.close()
+        w._host_state_path_for_task = lambda task: None
+        self.assertFalse(w._rebuild_host_state_from_checkpoint(
+            {"id": "t1", "project_id": "p1"}))
+
+    def test_rebuild_already_present_true(self):
+        self._insert_task()
+        w = _worker(self.db_path)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS task_checkpoints("
+                "task_id TEXT, round_no INTEGER, thread_id TEXT,"
+                " goal_status TEXT, state_sha TEXT, verify_out TEXT,"
+                " created_at DOUBLE PRECISION,"
+                " PRIMARY KEY(task_id, round_no))")
+            conn.execute("INSERT INTO task_checkpoints VALUES"
+                         "('t1', 3, 'th', 'ok', 'sha', '', 0)")
+            conn.commit()
+        finally:
+            conn.close()
+        work = os.path.join(self.tmp.name, "work")
+        os.makedirs(work, exist_ok=True)
+        with open(os.path.join(work, "host-state.json"), "w") as fh:
+            import json as _json
+            _json.dump({"version": 1, "tasks": {"t1": {"goal": "g"}}}, fh)
+        w.executor = SimpleNamespace(extra_mounts=[(work, "/app/work")],
+                                     config_path=None)
+        self.assertTrue(w._rebuild_host_state_from_checkpoint(
+            {"id": "t1", "project_id": "p1"}))
+
+    def test_rebuild_db_failure_false(self):
+        w = _worker(self.tmp.name)
+        self.assertFalse(w._rebuild_host_state_from_checkpoint(
+            {"id": "t1", "project_id": "p1"}))
+
+    def test_never_started_state_path_none(self):
+        w = _worker(self.db_path)
+        w._host_state_path_for_task = lambda task: None
+        self.assertFalse(w._task_never_started(
+            {"id": "t1", "project_id": "p1"}))
+
+    def test_resolve_breakpoint_log_failure_swallowed(self):
+        w = _worker(self.db_path)
+        w._rebuild_host_state_from_checkpoint = lambda task: (_ for _ in (
+            )).throw(RuntimeError("rebuild down"))
+        w._task_never_started = lambda task: True
+        with mock.patch("cloud.worker.log_event",
+                        side_effect=RuntimeError("log down")):
+            out = w._resolve_breakpoint_resume(
+                {"id": "t1", "project_id": "p1"})
+        self.assertFalse(out["_goal_resume_breakpoint"])
+
+
+class EnsureColumnsTestCase(_Base):
+    def _fake(self, mode):
+        all_cols = ["result_text", "result_text_full", "error_text",
+                    "actual_model", "started_at", "finished_at",
+                    "child_pid", "executor", "pr_url", "pr_number", "goal",
+                    "verify_command", "goal_max_rounds", "goal_mode",
+                    "worktree_host", "claimed_by_worker", "model", "effort",
+                    "child_model", "child_effort", "approval", "plan",
+                    "forked_from", "environment_id", "resume_message",
+                    "turn_id", "root_turn_id", "coordinator_parts",
+                    "parent_task_id", "coordinator_part",
+                    "goal_resume_pending", "paused_by_user",
+                    "interrupted_reason", "auto_resume_count",
+                    "no_time_limit", "resume_after",
+                    "auto_resume_stall_count",
+                    "checkpoint_round_at_last_resume",
+                    "last_error_signature"]
+
+        class _Fake:
+            def execute(self, sql, params=()):
+                if sql.strip().startswith("PRAGMA"):
+                    rows = [(i, name, "TEXT", 0, None, 0)
+                            for i, name in enumerate(
+                                all_cols if mode == "create" else ["id"])]
+                    result = mock.Mock()
+                    result.fetchall.return_value = rows
+                    return result
+                if sql.strip().startswith("ALTER"):
+                    raise RuntimeError("alter down")
+                if sql.strip().startswith("CREATE"):
+                    if mode == "create":
+                        raise RuntimeError("create down")
+                    return mock.Mock()
+                return mock.Mock()
+        return _Fake()
+
+    def test_ensure_columns_pragma_failure(self):
+        w = _worker(self.db_path)
+        bad = mock.Mock()
+        bad.execute.side_effect = RuntimeError("pragma down")
+        w._ensure_result_columns(bad)
+
+    def test_ensure_columns_alter_failure(self):
+        w = _worker(self.db_path)
+        w._ensure_result_columns(self._fake("alter"))
+
+    def test_ensure_columns_create_failure(self):
+        w = _worker(self.db_path)
+        w._ensure_result_columns(self._fake("create"))
+
+
+class WriteResultTestCase(_Base):
+    def test_write_result_child_pid_branch(self):
+        self._insert_task()
+        w = _worker(self.db_path)
+        self.assertTrue(w._write_execution_result(
+            "t1", result_text="done", child_pid=99))
+        self.assertEqual(self._row()["child_pid"], 99)
+
+    def test_write_result_old_column_fallback(self):
+        self._insert_task()
+        w = _worker(self.db_path)
+        w._ensure_result_columns = lambda conn: None
+        real = sqlite3.connect(self.db_path)
+
+        class _Flaky:
+            def __init__(self):
+                self.flipped = False
+
+            def execute(self, sql, params=()):
+                if "result_text_full" in sql and not self.flipped:
+                    self.flipped = True
+                    raise sqlite3.OperationalError(
+                        "no such column: result_text_full")
+                return real.execute(sql, params)
+
+            def rollback(self):
+                raise RuntimeError("rollback down")
+
+            def commit(self):
+                return real.commit()
+
+            def close(self):
+                return real.close()
+        with mock.patch("cloud.worker._db_connect",
+                        return_value=_Flaky()):
+            self.assertTrue(w._write_execution_result(
+                "t1", result_text="fallback-ok"))
+        self.assertEqual(self._row()["result_text"], "fallback-ok")
+
+
+class KillAuditNotifyTestCase(_Base):
+    def test_kill_proc_group_term_error(self):
+        w = _worker(self.db_path)
+        proc = subprocess.Popen(["sleep", "30"], start_new_session=True)
+        try:
+            with mock.patch("os.killpg",
+                            side_effect=ProcessLookupError("gone")):
+                w._kill_proc_group(proc)
+            self.assertIsNone(proc.poll())
+        finally:
+            proc.kill()
+            proc.wait()
+
+    def test_kill_proc_group_sigkill_error(self):
+        w = _worker(self.db_path)
+        proc = subprocess.Popen(["sleep", "30"], start_new_session=True)
+        try:
+            def _killpg(pgid, sig):
+                if sig == signal.SIGKILL:
+                    raise ProcessLookupError("gone")
+                return None
+            with mock.patch("os.killpg", side_effect=_killpg):
+                w._kill_proc_group(proc)
+            self.assertIsNone(proc.poll())
+        finally:
+            proc.kill()
+            proc.wait()
+
+    def test_audit_close_failure_swallowed(self):
+        w = _worker(self.db_path)
+
+        class _Store:
+            def __init__(self, db_path):
+                self.conn = mock.Mock()
+                self.conn.close.side_effect = RuntimeError("down")
+
+            def audit(self, *args):
+                return None
+        with mock.patch("cloud.worker.AuthStore", _Store):
+            w._audit("p1", "act")
+
+    def test_notify_errors_and_exception_swallowed(self):
+        w = _worker(self.db_path)
+        with mock.patch("cloud.push.dispatch_event",
+                        return_value={"errors": ["e1"]}):
+            w._notify_task_finished("t1", "p1", "completed")
+        with mock.patch("cloud.push.dispatch_event",
+                        side_effect=RuntimeError("push down")):
+            w._notify_task_finished("t1", "p1", "failed")
+
+
+class InjectionTestCase(_Base):
+    def test_auto_memory_no_owner(self):
+        w = _worker(self.db_path)
+        task = {"id": "t1", "project_id": "p1", "prompt": "p"}
+        self.assertIs(w._apply_auto_memory_injection(task), task)
+
+    def test_auto_memory_store_and_log_failure(self):
+        w = _worker(self.db_path)
+        w.logger = mock.Mock()
+        w.logger.warning.side_effect = RuntimeError("log down")
+        task = {"id": "t1", "project_id": "p1", "prompt": "p",
+                "created_by": "u1"}
+        with mock.patch("cloud.worker.AutoMemoryStore",
+                        side_effect=RuntimeError("store down")):
+            self.assertIs(w._apply_auto_memory_injection(task), task)
+
+    def test_auto_memory_goal_branch(self):
+        w = _worker(self.db_path)
+
+        class _Store:
+            def __init__(self, db_path):
+                pass
+
+            def get_task_flags(self, task_id):
+                return {}
+
+            def close(self):
+                pass
+        task = {"id": "t1", "project_id": "p1", "prompt": "P",
+                "goal": "G", "goal_mode": 1, "created_by": "u1"}
+        with mock.patch("cloud.worker.AutoMemoryStore", _Store), \
+                mock.patch("cloud.worker._auto_prefix",
+                           return_value="MEM"):
+            out = w._apply_auto_memory_injection(task)
+        self.assertEqual(out["goal"], "MEM\n\nG")
+        self.assertEqual(out["prompt"], "MEM\n\nP")
+        self.assertTrue(out["_memory_injected"])
+        self.assertEqual(task["goal"], "G")
+
+    def test_todo_injection_db_and_log_failure(self):
+        w = _worker(self.tmp.name)
+        w.logger = mock.Mock()
+        w.logger.warning.side_effect = RuntimeError("log down")
+        task = {"id": "t1", "project_id": "p1", "prompt": "p"}
+        self.assertIs(w._apply_todo_injection(task), task)
+
+    def test_todo_injection_over_twenty(self):
+        w = _worker(self.db_path)
+        todos = [{"id": "td%d" % i, "title": "T%d" % i,
+                  "status": "pending"} for i in range(21)]
+        task = {"id": "t1", "project_id": "p1", "prompt": "p"}
+        with mock.patch("cloud.worker._todo_list_db",
+                        return_value=todos):
+            out = w._apply_todo_injection(task)
+        self.assertTrue(out["_todo_injected"])
+        self.assertIn("还有 1 条未列出", out["prompt"])
+
+    def test_project_injection_db_and_log_failure(self):
+        w = _worker(self.tmp.name)
+        w.logger = mock.Mock()
+        w.logger.warning.side_effect = RuntimeError("log down")
+        task = {"id": "t1", "project_id": "p1", "prompt": "p"}
+        self.assertIs(w._apply_project_injection(task), task)
+
+    def test_project_injection_row_key_errors(self):
+        w = _worker(self.db_path)
+
+        class _BadRow:
+            def __getitem__(self, key):
+                raise KeyError(key)
+
+        class _FakeConn:
+            def execute(self, sql, params=()):
+                result = mock.Mock()
+                result.fetchone.return_value = _BadRow()
+                return result
+
+            def close(self):
+                pass
+        task = {"id": "t1", "project_id": "p1", "prompt": "p"}
+        with mock.patch("cloud.worker._db_connect",
+                        return_value=_FakeConn()):
+            out = w._apply_project_injection(task)
+        self.assertTrue(out["_project_injected"])
+        self.assertIn("项目：p1（id=p1）", out["prompt"])
+        self.assertIn("仓库：未绑定", out["prompt"])
+
+    def test_project_injection_env_missing(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO projects(id, name, owner_id, created_at,"
+                " repo_url, auto_pr) VALUES('p1', 'Proj', 'u1', 0, NULL, 0)")
+            conn.commit()
+        finally:
+            conn.close()
+        w = _worker(self.db_path)
+        w._fetch_environment = lambda env_id: None
+        task = {"id": "t1", "project_id": "p1", "prompt": "p",
+                "environment_id": "envX"}
+        out = w._apply_project_injection(task)
+        self.assertIn("envX（查不到该环境）", out["prompt"])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_v2cov_gaps_d1.py

@@ -0,0 +1,301 @@
+"""Coverage batch v4-D1: cloud/worker.py resume/checkpoint/notify branches.
+
+Stubbing discipline: the method under test is never stubbed. Patched
+collaborators are: _quota_exceeded / _attempt_mark / _db_connect /
+dispatch_event / AutoMemoryStore / the instance logger (data and sink
+collaborators), plus instance-level _mirror_checkpoint_once where the
+real path cannot raise without a race. Every assertion reads a real
+return value or a real DB row.
+"""
+
+import json
+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))
+
+from cloud.server import init_db
+from cloud.worker import Worker
+
+FAKE_ENV = {"XCM_API_KEY": "test-placeholder"}
+
+
+def _worker(db_path, **kwargs):
+    return Worker(db_path, "/fake/host.py", env=FAKE_ENV, **kwargs)
+
+
+class _Base(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)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _insert_task(self, task_id="t1", status="queued", **cols):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " created_by, created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, ?, ?, ?)",
+                (task_id, "p1", "do work", status, "u1", time.time(),
+                 time.time()))
+            for key, value in cols.items():
+                conn.execute(
+                    "UPDATE tasks SET %s = ? WHERE id = ?" % key,
+                    (value, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _row(self, task_id="t1"):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.row_factory = sqlite3.Row
+            return conn.execute(
+                "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
+        finally:
+            conn.close()
+
+    def _fake_executor(self, state_dir):
+        cfg = os.path.join(state_dir, "config.json")
+        with open(cfg, "w", encoding="utf-8") as fh:
+            json.dump({"state": "host-state.json"}, fh)
+        return SimpleNamespace(extra_mounts=[], config_path=cfg)
+
+
+class AutoResumeExceptionTestCase(_Base):
+    def test_quota_check_raises_stays_interrupted(self):
+        # 1093-1094: _quota_exceeded raising inside auto_resume_interrupted
+        # must be swallowed into "quota_check_failed" and the task skipped.
+        self._insert_task(status="interrupted", goal_mode=1,
+                          interrupted_reason="quota")
+        w = _worker(self.db_path)
+        with mock.patch.object(
+                Worker, "_quota_exceeded",
+                side_effect=RuntimeError("quota db down")):
+            resumed = w.auto_resume_interrupted()
+        self.assertNotIn("t1", resumed)
+        self.assertEqual(self._row()["status"], "interrupted")
+
+    def test_attempt_mark_failure_swallowed_on_resume(self):
+        # 1163-1164: _attempt_mark raising after a resume must not block
+        # the resume itself.
+        self._insert_task(status="interrupted", goal_mode=1,
+                          interrupted_reason="worker_shutdown")
+        w = _worker(self.db_path, attempt_ledger_hook=True)
+        with mock.patch("cloud.worker._attempt_mark",
+                        side_effect=RuntimeError("ledger down")):
+            resumed = w.auto_resume_interrupted()
+        self.assertIn("t1", resumed)
+        self.assertEqual(self._row()["status"], "queued")
+
+    def test_db_connect_failure_returns_empty(self):
+        # 1182-1183: outer except of auto_resume_interrupted.
+        w = _worker(self.db_path)
+        with mock.patch("cloud.worker._db_connect",
+                        side_effect=RuntimeError("db down")):
+            self.assertEqual(w.auto_resume_interrupted(), [])
+
+    def test_checkpoint_select_failure_returns_minus_one(self):
+        # 1102-1103: checkpoint SELECT raising inside auto_resume must
+        # fall back to cp_round=-1 and still resume a platform task.
+        # NOTE: dropping the table cannot reach this branch, because
+        # _ensure_result_columns recreates task_checkpoints before the
+        # SELECT; the branch is reached only when the SELECT itself
+        # raises, so the connection is faked for that statement only.
+        self._insert_task(status="interrupted", goal_mode=1,
+                          interrupted_reason="worker_shutdown")
+        w = _worker(self.db_path)
+        w._ensure_result_columns = lambda conn: None
+        row = ("t1", "p1", "u1", None, "m0", "worker_shutdown", None,
+               0, None, 1, None, None)
+
+        class _Cursor:
+            def __init__(self, rows=None, rowcount=1):
+                self._rows = rows or []
+                self.rowcount = rowcount
+
+            def fetchall(self):
+                return self._rows
+
+            def fetchone(self):
+                return self._rows[0] if self._rows else None
+
+        class _FakeConn:
+            def execute(self, sql, params=()):
+                if "task_checkpoints" in sql:
+                    raise RuntimeError("checkpoint read down")
+                if "FROM tasks" in sql and "SELECT" in sql:
+                    return _Cursor(rows=[row])
+                return _Cursor()
+
+            def commit(self):
+                pass
+
+            def close(self):
+                pass
+
+        with mock.patch("cloud.worker._db_connect",
+                        return_value=_FakeConn()):
+            resumed = w.auto_resume_interrupted()
+        self.assertIn("t1", resumed)
+
+    def test_checkpoint_round_and_mirror_loop_exceptions(self):
+        w = _worker(self.db_path)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            self.assertEqual(w._checkpoint_round(conn, "nope"), -1)
+        finally:
+            conn.close()
+        bad = mock.Mock()
+        bad.execute.side_effect = RuntimeError("boom")
+        self.assertEqual(w._checkpoint_round(bad, "nope"), -1)
+        # 1230-1231: first mirror call raising, stop already set.
+        w._mirror_checkpoint_once = mock.Mock(
+            side_effect=RuntimeError("mirror down"))
+        stop = threading.Event()
+        stop.set()
+        w._checkpoint_mirror_loop({"id": "t1"}, stop)
+        # 1233-1236: loop-body mirror call raising, then stop.
+        calls = {"n": 0}
+
+        def _wait(timeout=None):
+            calls["n"] += 1
+            return calls["n"] > 1
+
+        class _Stop:
+            def wait(self, timeout=None):
+                return _wait(timeout)
+
+        w._checkpoint_mirror_loop({"id": "t1"}, _Stop())
+
+
+class HostStatePathTestCase(_Base):
+    def test_non_tuple_mount_item_falls_through(self):
+        # 1256: a non-tuple mounts entry yields (None, None) and the
+        # loop falls through to the /opt fallback.
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            extra_mounts=["not-a-tuple"], config_path=None))
+        path = w._host_state_path_for_task({"project_id": "p1"})
+        self.assertEqual(path, "/opt/codex-app-server/work/host-state.json")
+
+    def test_mirror_missing_state_returns_minus_one(self):
+        # 1286: no state file at all.
+        state_dir = os.path.join(self.tmp.name, "cfg")
+        os.makedirs(state_dir)
+        w = _worker(self.db_path,
+                    executor=self._fake_executor(state_dir))
+        self.assertEqual(
+            w._mirror_checkpoint_once({"id": "t1", "project_id": "p1"}), -1)
+
+    def test_rebuild_from_checkpoint_missing_state_file(self):
+        # 1411: state file absent -> fresh {"version": 1, "tasks": {}}
+        # is built and the rebuilt entry is written back.
+        self._insert_task(status="interrupted", verify_command="true",
+                          goal_max_rounds=5)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO task_checkpoints(task_id, round_no,"
+                " thread_id, goal_status) VALUES(?, ?, ?, ?)",
+                ("t1", 2, "th1", "interrupted"))
+            conn.commit()
+        finally:
+            conn.close()
+        state_dir = os.path.join(self.tmp.name, "cfg2")
+        os.makedirs(state_dir)
+        w = _worker(self.db_path,
+                    executor=self._fake_executor(state_dir))
+        self.assertTrue(w._rebuild_host_state_from_checkpoint(
+            {"id": "t1", "project_id": "p1"}))
+        with open(os.path.join(state_dir, "host-state.json"),
+                  encoding="utf-8") as fh:
+            state = json.load(fh)
+        self.assertEqual(state["tasks"]["t1"]["goal_rounds_done"], 2)
+        self.assertEqual(state["tasks"]["t1"]["thread_id"], "th1")
+        self.assertTrue(state["tasks"]["t1"]["rebuilt_from_checkpoint"])
+
+    def test_record_step_failure_swallowed(self):
+        # 1208-1209: step INSERT raising is logged, never raised.
+        w = _worker(self.db_path)
+        bad = mock.Mock()
+        bad.execute.side_effect = RuntimeError("db down")
+        w._record_step(bad, "t1", "round", "summary")
+        bad.execute.assert_called()
+
+
+class WriteResultTestCase(_Base):
+    def test_child_pid_branch_writes_full(self):
+        # 1664-1670, 1672-1673, 1675: child_pid is not None branch.
+        self._insert_task(status="running")
+        w = _worker(self.db_path)
+        self.assertTrue(w._write_execution_result(
+            "t1", result_text="out", result_full="full-out",
+            child_pid=1234, turn_id="turn-1"))
+        row = self._row()
+        self.assertEqual(row["child_pid"], 1234)
+        self.assertEqual(row["result_text_full"], "full-out")
+        self.assertEqual(row["turn_id"], "turn-1")
+
+    def test_write_result_non_column_error_reraised(self):
+        # 1686: an execute error unrelated to result_text_full is raised.
+        self._insert_task(status="running")
+        w = _worker(self.db_path)
+        w._ensure_result_columns = lambda conn: None
+
+        class _BadConn:
+            def execute(self, *args, **kwargs):
+                raise ValueError("boom")
+
+            def rollback(self):
+                pass
+
+            def commit(self):
+                pass
+
+            def close(self):
+                pass
+
+        with mock.patch("cloud.worker._db_connect",
+                        return_value=_BadConn()):
+            with self.assertRaises(ValueError):
+                w._write_execution_result("t1", result_text="out")
+
+
+class NotifyAndMemoryTestCase(_Base):
+    def test_notify_errors_branch_logged(self):
+        # 1784-1786: dispatch_event returning errors is logged.
+        w = _worker(self.db_path)
+        with mock.patch("cloud.push.dispatch_event",
+                        return_value={"errors": ["sub down"]}):
+            w._notify_task_finished("t1", "p1", "completed")
+        # No exception, no state change to assert beyond survival.
+
+    def test_auto_memory_logger_failure_swallowed(self):
+        # 1823: logger.warning itself raising inside the injection
+        # failure path is swallowed and the task is returned unchanged.
+        w = _worker(self.db_path)
+        task = {"id": "t1", "project_id": "p1", "created_by": "u1",
+                "prompt": "do work"}
+        w.logger = mock.Mock()
+        w.logger.warning.side_effect = RuntimeError("logger down")
+        with mock.patch("cloud.worker.AutoMemoryStore",
+                        side_effect=RuntimeError("store down")):
+            out = w._apply_auto_memory_injection(task)
+        self.assertIs(out, task)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_v2cov_gaps_d2.py

@@ -0,0 +1,576 @@
+"""Coverage batch v4-D2: cloud/worker.py tree/quota/cancel/watchdog/
+cleanup/gateway/process branches.
+
+Stubbing discipline: the method under test is never stubbed. Patched
+collaborators are: _tenant_safe_name / _tenant_dirs / _project_repo_url
+/ _quota_limit_and_used / _list_gateway_models / _gateway_name /
+_gateway_origin / ready_tasks / orphaned_containers / CostStore /
+_db_connect / dispatch sinks / instance-level _kill_proc_group /
+_set_child_pid / get_task / heartbeat (data, sink and predicate
+collaborators). Every assertion reads a real return value or a real
+DB row, except watchdog/cleanup tests where the assertion is the
+returned list or the absence of a raise plus a real side-effect set.
+"""
+
+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))
+
+from cloud.server import init_db
+from cloud.worker import Worker
+
+FAKE_ENV = {"XCM_API_KEY": "test-placeholder"}
+
+
+def _worker(db_path, **kwargs):
+    return Worker(db_path, "/fake/host.py", env=FAKE_ENV, **kwargs)
+
+
+class _Base(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)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _insert_task(self, task_id="t1", status="queued", **cols):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " created_by, created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, ?, ?, ?)",
+                (task_id, "p1", "do work", status, "u1", time.time(),
+                 time.time()))
+            for key, value in cols.items():
+                conn.execute(
+                    "UPDATE tasks SET %s = ? WHERE id = ?" % key,
+                    (value, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _row(self, task_id="t1"):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.row_factory = sqlite3.Row
+            return conn.execute(
+                "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
+        finally:
+            conn.close()
+
+
+class _StopOnce:
+    """wait() returns False once, then True."""
+
+    def __init__(self):
+        self.n = 0
+
+    def wait(self, timeout=None):
+        self.n += 1
+        return self.n > 1
+
+
+class TreeSnapshotTestCase(_Base):
+    def _docker_worker(self):
+        # Subclass of the real DockerExecutor (no-op __init__) so the
+        # isinstance check in _tree_snapshot_source passes without
+        # patching the module-level class.
+        from cloud.worker import DockerExecutor as _RealDocker
+
+        class _FakeDocker(_RealDocker):
+            def __init__(self):
+                pass
+
+        w = _worker(self.db_path, executor=_FakeDocker())
+        w.tenants_root = self.tmp.name
+        w.tenant_isolation = True
+        return w
+
+    def test_safe_name_failure_returns_none(self):
+        # 2003-2004
+        w = self._docker_worker()
+        w._tenant_safe_name = mock.Mock(side_effect=RuntimeError("bad"))
+        self.assertIsNone(w._tree_snapshot_source({"project_id": "p1"}))
+
+    def test_checkout_exists_returns_checkout(self):
+        # 2007
+        w = self._docker_worker()
+        w._tenant_safe_name = mock.Mock(return_value="safe")
+        os.makedirs(os.path.join(self.tmp.name, "safe", "safe"))
+        self.assertEqual(
+            w._tree_snapshot_source({"project_id": "p1"}),
+            os.path.join(self.tmp.name, "safe", "safe"))
+
+    def test_tenant_dirs_failure_returns_none(self):
+        # 2017-2018
+        w = self._docker_worker()
+        w._tenant_safe_name = mock.Mock(return_value="safe")
+        w._project_repo_url = mock.Mock(return_value=None)
+        w._tenant_dirs = mock.Mock(side_effect=RuntimeError("dirs down"))
+        self.assertIsNone(w._tree_snapshot_source({"project_id": "p1"}))
+
+    def test_test_repo_missing_returns_none(self):
+        # 2023
+        w = self._docker_worker()
+        w._tenant_safe_name = mock.Mock(return_value="safe")
+        w._project_repo_url = mock.Mock(return_value=None)
+        w._tenant_dirs = mock.Mock(
+            return_value={"test_repo": "/nonexistent/repo-xyz"})
+        self.assertIsNone(w._tree_snapshot_source({"project_id": "p1"}))
+
+    def test_injection_source_failure_swallowed(self):
+        # 2051-2053, 2056
+        w = _worker(self.db_path)
+        w._tree_snapshot_source = mock.Mock(
+            side_effect=RuntimeError("snap down"))
+        task = {"id": "t1", "project_id": "p1", "prompt": "do work"}
+        self.assertIs(w._apply_tree_injection(task), task)
+
+    def test_injection_logger_failure_swallowed(self):
+        # 2054-2055
+        w = _worker(self.db_path)
+        w._tree_snapshot_source = mock.Mock(
+            side_effect=RuntimeError("snap down"))
+        w.logger = mock.Mock()
+        w.logger.warning.side_effect = RuntimeError("logger down")
+        task = {"id": "t1", "project_id": "p1", "prompt": "do work"}
+        self.assertIs(w._apply_tree_injection(task), task)
+
+
+class QuotaTestCase(_Base):
+    def test_limit_commit_failure_swallowed_unlimited(self):
+        # 2131-2132: commit raising, explicit-NULL quota row -> None.
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO user_quotas(user_id, quota_usd_micros,"
+                " updated_at) VALUES(?, NULL, ?)", ("u1", time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+        w = _worker(self.db_path)
+        real = sqlite3.connect(self.db_path)
+
+        class _Proxy:
+            def __init__(self, conn):
+                self._conn = conn
+
+            def __getattr__(self, name):
+                return getattr(self._conn, name)
+
+            def commit(self):
+                raise RuntimeError("commit down")
+
+            def close(self):
+                self._conn.close()
+
+        with mock.patch("cloud.worker._db_connect",
+                        return_value=_Proxy(real)):
+            quota, used, month, unpriced = w._quota_limit_and_used("u1")
+        self.assertIsNone(quota)
+        self.assertEqual((used, unpriced), (0, 0))
+
+    def test_limit_coststore_close_failure_swallowed(self):
+        # 2142-2143
+        from cloud.cost import CostStore
+        CostStore(self.db_path).close() if hasattr(
+            CostStore(self.db_path), "close") else None
+        store = CostStore(self.db_path)
+        try:
+            store._conn.close()
+        except Exception:
+            pass
+        w = _worker(self.db_path)
+
+        class _BadStore:
+            def __init__(self, db_path):
+                self._conn = mock.Mock()
+                self._conn.close.side_effect = RuntimeError("close down")
+
+        with mock.patch("cloud.worker.CostStore", _BadStore):
+            quota, used, month, unpriced = w._quota_limit_and_used("u9")
+        self.assertIsNotNone(quota)
+        self.assertEqual((used, unpriced), (0, 0))
+
+    def test_quota_parent_row_type_error(self):
+        # 2196-2197
+        w = _worker(self.db_path)
+
+        class _Cursor:
+            def fetchone(self):
+                return 5
+
+        class _Conn:
+            def execute(self, *a, **k):
+                return _Cursor()
+
+            def close(self):
+                pass
+
+        with mock.patch("cloud.worker._db_connect", return_value=_Conn()):
+            msg = w._quota_exceeded({"parent_task_id": "p0"}, None)
+        self.assertIn("缺少属主", msg)
+
+    def test_quota_parent_lookup_failure(self):
+        # 2198-2199
+        w = _worker(self.db_path)
+        with mock.patch("cloud.worker._db_connect",
+                        side_effect=RuntimeError("db down")):
+            msg = w._quota_exceeded({"parent_task_id": "p0"}, None)
+        self.assertIn("缺少属主", msg)
+
+    def test_quota_limit_failure_logger_failure(self):
+        # 2213-2214, both on_error flavours
+        w = _worker(self.db_path)
+        w._quota_limit_and_used = mock.Mock(
+            side_effect=RuntimeError("quota db down"))
+        w.logger = mock.Mock()
+        w.logger.warning.side_effect = RuntimeError("logger down")
+        self.assertIn("quota_check_unavailable",
+                      w._quota_exceeded({"created_by": "u1"}, None,
+                                        on_error="closed"))
+        self.assertIn("quota_check_failed",
+                      w._quota_exceeded({"created_by": "u1"}, None))
+
+    def test_ready_budget_exceeded_empty(self):
+        # 2252
+        w = _worker(self.db_path)
+        w.usage_turns = 5
+        w.max_turns = 1
+        self.assertEqual(w._ready_tasks(), [])
+
+    def test_ready_value_error_empty(self):
+        # 2258-2259
+        self._insert_task()
+        w = _worker(self.db_path)
+        with mock.patch("cloud.worker.ready_tasks",
+                        side_effect=ValueError("cycle")):
+            self.assertEqual(w._ready_tasks(), [])
+
+
+class HeartbeatCancelTestCase(_Base):
+    def test_heartbeat_task_failure_swallowed(self):
+        # 2330-2331
+        w = _worker(self.db_path)
+        w._running_set("t1")
+        w.heartbeat = mock.Mock(side_effect=RuntimeError("beat down"))
+        w._heartbeat_running_tasks()
+
+    def test_cancel_proc_wait_failure_swallowed(self):
+        # 2349-2350
+        self._insert_task(status="running")
+        w = _worker(self.db_path)
+        proc = mock.Mock()
+        proc.wait.side_effect = RuntimeError("wait down")
+        w._running_set("t1", proc=proc)
+        w._kill_proc_group = lambda p: None
+        self.assertTrue(w.cancel("t1"))
+        self.assertEqual(self._row()["status"], "interrupted")
+
+    def test_cancel_container_kill_failure_swallowed(self):
+        # 2359-2360
+        self._insert_task(status="running")
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            kill_container=mock.Mock(
+                side_effect=RuntimeError("kill down"))))
+        w._running_set("t1", container="c1")
+        self.assertTrue(w.cancel("t1"))
+        self.assertEqual(self._row()["status"], "interrupted")
+
+
+class WatchdogTestCase(_Base):
+    def test_cancel_watchdog_lookup_failure_continues(self):
+        # 2407-2408
+        w = _worker(self.db_path)
+        w.get_task = mock.Mock(side_effect=RuntimeError("db down"))
+        w._docker_cancel_watchdog("t1", "c1", _StopOnce())
+
+    def test_cancel_watchdog_kill_failure_swallowed(self):
+        # 2420-2421
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            kill_container=mock.Mock(
+                side_effect=RuntimeError("kill down"))))
+        w._cancel_requested.add("t1")
+        w._docker_cancel_watchdog("t1", "c1", _StopOnce())
+
+    def test_quota_watchdog_check_failure_continues(self):
+        # 2446-2447
+        w = _worker(self.db_path)
+        w._docker_quota_watchdog(
+            "t1", "c1", _StopOnce(),
+            lambda: (_ for _ in ()).throw(RuntimeError("check down")))
+
+    def test_quota_watchdog_kill_failure_swallowed(self):
+        # 2459-2460
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            kill_container=mock.Mock(
+                side_effect=RuntimeError("kill down"))))
+        w._docker_quota_watchdog("t1", "c1", _StopOnce(), lambda: True)
+        self.assertIn("t1", w._quota_hit_tasks)
+
+
+class CleanupTestCase(_Base):
+    def test_orphan_no_executor_caps_returns_empty(self):
+        # 2485
+        w = _worker(self.db_path)
+        self.assertEqual(w.cleanup_orphaned_containers(), [])
+
+    def test_orphan_list_failure_returns_empty(self):
+        # 2488-2491
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            list_containers=mock.Mock(
+                side_effect=RuntimeError("list down")),
+            kill_container=mock.Mock()))
+        self.assertEqual(w.cleanup_orphaned_containers(), [])
+
+    def test_orphan_identify_failure_returns_empty(self):
+        # 2496-2499
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            list_containers=mock.Mock(return_value=["c1"]),
+            kill_container=mock.Mock()))
+        with mock.patch("cloud.worker.orphaned_containers",
+                        side_effect=RuntimeError("identify down")):
+            self.assertEqual(w.cleanup_orphaned_containers(), [])
+
+    def test_orphan_kill_failure_skipped(self):
+        # 2509-2510
+        kill = mock.Mock(side_effect=RuntimeError("kill down"))
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            list_containers=mock.Mock(return_value=["c1"]),
+            kill_container=kill))
+        with mock.patch("cloud.worker.orphaned_containers",
+                        return_value=[{"container": "c1",
+                                       "pid_verified": True}]):
+            self.assertEqual(w.cleanup_orphaned_containers(), [])
+
+    def test_stale_cleanup_failure_returns_empty(self):
+        # 2534-2537
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            cleanup_stale_containers=mock.Mock(
+                side_effect=RuntimeError("cleanup down"))))
+        self.assertEqual(w.cleanup_stale_containers(), [])
+
+    def test_stale_cleanup_nonempty_returned(self):
+        # 2539
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            cleanup_stale_containers=mock.Mock(return_value=["c1"])))
+        self.assertEqual(w.cleanup_stale_containers(), ["c1"])
+
+
+class GatewayTestCase(_Base):
+    def test_key_available_env_failure_false(self):
+        # 2564-2565
+        class _BadEnv:
+            def get(self, *a, **k):
+                raise RuntimeError("env down")
+
+        w = Worker(self.db_path, "/fake/host.py", env=_BadEnv())
+        self.assertFalse(w.key_available())
+
+    def _sub2api_env(self):
+        return (mock.patch("cloud.worker._gateway_name",
+                           return_value="sub2api"),
+                mock.patch("cloud.worker._gateway_origin",
+                           return_value="http://gw.example"))
+
+    def test_gateway_parent_row_type_error(self):
+        # 2607-2608 -> owner missing -> RuntimeError
+        w = _worker(self.db_path)
+        name_p, origin_p = self._sub2api_env()
+
+        class _Cursor:
+            def fetchone(self):
+                return 5
+
+        class _Conn:
+            def execute(self, *a, **k):
+                return _Cursor()
+
+            def close(self):
+                pass
+
+        with name_p, origin_p, mock.patch(
+                "cloud.worker._db_connect", return_value=_Conn()):
+            with self.assertRaises(RuntimeError):
+                w._gateway_context({"parent_task_id": "p0"})
+
+    def test_gateway_parent_lookup_failure(self):
+        # 2609-2610 -> owner missing -> RuntimeError
+        w = _worker(self.db_path)
+        name_p, origin_p = self._sub2api_env()
+        with name_p, origin_p, mock.patch(
+                "cloud.worker._db_connect",
+                side_effect=RuntimeError("db down")):
+            with self.assertRaises(RuntimeError):
+                w._gateway_context({"parent_task_id": "p0"})
+
+    def test_sub2api_default_model_none_available(self):
+        # 2660-2662
+        w = _worker(self.db_path)
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_SUB2API_DEFAULT_MODEL": ""}):
+            with mock.patch("cloud.worker._gateway_origin",
+                            return_value="http://gw.example"):
+                with mock.patch("cloud.worker._list_gateway_models",
+                                return_value=[]):
+                    with self.assertRaises(RuntimeError):
+                        w._sub2api_default_model(
+                            {"SUB2API_API_KEY": "k"})
+
+    def test_sub2api_default_model_first(self):
+        # 2663
+        w = _worker(self.db_path)
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_SUB2API_DEFAULT_MODEL": ""}):
+            with mock.patch("cloud.worker._gateway_origin",
+                            return_value="http://gw.example"):
+                with mock.patch("cloud.worker._list_gateway_models",
+                                return_value=["m1", "m2"]):
+                    self.assertEqual(
+                        w._sub2api_default_model(
+                            {"SUB2API_API_KEY": "k"}), "m1")
+
+    def test_task_seed_dir_db_failure_none(self):
+        # 2680-2681
+        w = _worker(self.db_path)
+        with mock.patch("cloud.worker._db_connect",
+                        side_effect=RuntimeError("db down")):
+            self.assertIsNone(w._task_seed_dir({"id": "t1"}))
+
+    def test_cleanup_seed_failure_swallowed(self):
+        # 2701-2702
+        w = _worker(self.db_path)
+        w._task_seed_dir = mock.Mock(
+            side_effect=RuntimeError("seed down"))
+        w._cleanup_task_seed({"id": "t1"})
+
+
+class ProcessTestCase(_Base):
+    def _finished_proc(self, returncode=0):
+        proc = mock.Mock()
+        proc.poll.return_value = returncode
+        proc.communicate.return_value = ("out", "err")
+        proc.returncode = returncode
+        proc.pid = 4242
+        return proc
+
+    def _proc_worker(self, proc):
+        return _worker(self.db_path,
+                       executor=SimpleNamespace(
+                           popen_task=mock.Mock(return_value=proc)))
+
+    def test_executor_popen_path(self):
+        # 2730 + 2820 (no _result_from_runner on the executor)
+        self._insert_task(status="running")
+        proc = self._finished_proc()
+        w = self._proc_worker(proc)
+        out = w._execute_via_process({"id": "t1", "project_id": "p1"})
+        self.assertEqual(out["status"], "completed")
+        self.assertEqual(out["stdout"], "out")
+
+    def test_set_child_pid_failure_swallowed(self):
+        # 2736-2737
+        self._insert_task(status="running")
+        w = _worker(self.db_path,
+                    process_factory=lambda task: self._finished_proc())
+        w._set_child_pid = mock.Mock(
+            side_effect=RuntimeError("pid down"))
+        out = w._execute_via_process({"id": "t1", "project_id": "p1"})
+        self.assertEqual(out["status"], "completed")
+
+    def test_quota_interval_none_defaults(self):
+        # 2747
+        self._insert_task(status="running")
+        w = _worker(self.db_path,
+                    process_factory=lambda task: self._finished_proc())
+        w.quota_poll_interval_seconds = None
+        out = w._execute_via_process({"id": "t1", "project_id": "p1"})
+        self.assertEqual(out["status"], "completed")
+
+    def test_loop_db_lookup_failure_continues(self):
+        # 2763-2764
+        self._insert_task(status="running")
+        proc = mock.Mock()
+        proc.poll.side_effect = [None, 0]
+        proc.communicate.return_value = ("out", "err")
+        proc.returncode = 0
+        proc.pid = 4242
+        w = _worker(self.db_path,
+                    process_factory=lambda task: proc)
+        w.get_task = mock.Mock(side_effect=RuntimeError("db down"))
+        out = w._execute_via_process({"id": "t1", "project_id": "p1"})
+        self.assertEqual(out["status"], "completed")
+
+    def test_loop_db_interrupted_cancels(self):
+        # 2767-2769
+        self._insert_task(status="running")
+        proc = mock.Mock()
+        proc.poll.side_effect = [None, 0]
+        proc.communicate.return_value = ("out", "err")
+        proc.returncode = 0
+        proc.pid = 4242
+        w = _worker(self.db_path,
+                    process_factory=lambda task: proc)
+        w.get_task = mock.Mock(return_value={"status": "interrupted"})
+        w._kill_proc_group = lambda p: None
+        out = w._execute_via_process({"id": "t1", "project_id": "p1"})
+        self.assertEqual(out["status"], "cancelled")
+
+    def test_loop_quota_check_failure_continues(self):
+        # 2780-2781
+        self._insert_task(status="running")
+        proc = mock.Mock()
+        proc.poll.side_effect = [None, 0]
+        proc.communicate.return_value = ("out", "err")
+        proc.returncode = 0
+        proc.pid = 4242
+        w = _worker(self.db_path,
+                    process_factory=lambda task: proc)
+        w.quota_poll_interval_seconds = 0
+        out = w._execute_via_process(
+            {"id": "t1", "project_id": "p1"},
+            quota_check=mock.Mock(side_effect=RuntimeError("quota down")))
+        self.assertEqual(out["status"], "completed")
+
+    def test_communicate_failure_falls_back(self):
+        # 2789-2794
+        self._insert_task(status="running")
+        proc = mock.Mock()
+        proc.poll.return_value = 0
+        proc.communicate.side_effect = [RuntimeError("comm down"),
+                                        ("out2", "err2")]
+        proc.returncode = 0
+        proc.pid = 4242
+        w = _worker(self.db_path,
+                    process_factory=lambda task: proc)
+        out = w._execute_via_process({"id": "t1", "project_id": "p1"})
+        self.assertEqual(out["stdout"], "out2")
+        proc.kill.assert_called_once()
+
+    def test_goal_result_from_runner(self):
+        # 2817
+        self._insert_task(status="running")
+        proc = self._finished_proc()
+        marker = {"status": "goal-met", "task_id": "t1"}
+        w = _worker(self.db_path, executor=SimpleNamespace(
+            popen_task=mock.Mock(return_value=proc),
+            _goal_result_from_runner=mock.Mock(return_value=marker)))
+        out = w._execute_via_process({"id": "t1", "project_id": "p1"},
+                                     kind="goal")
+        self.assertEqual(out, marker)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_v2cov_gaps_d4.py

@@ -0,0 +1,85 @@
+"""Coverage batch v4-D4: cloud/server.py dispatch-safe/upload branches
+and cloud/ledger_outbox.py replay rewrite.
+
+Stubbing discipline: the method under test is never stubbed. Patched
+collaborators are: _dispatch / _send_json / _read_body / parse_multipart
+/ parse_paths_field (sinks and data providers). Every assertion reads a
+real return value, a real captured response, or a real file.
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+from cloud import server as server_mod
+from cloud.ledger_outbox import replay_failures
+
+
+class DispatchSafeTestCase(unittest.TestCase):
+    def test_dispatch_and_send_both_raise_swallowed(self):
+        # 2310-2311: _dispatch raising and the 500 fallback also raising.
+        handler = object.__new__(server_mod.CloudControlPlaneHandler)
+        handler._responded = False
+        handler.path = "/x"
+        handler._dispatch = mock.Mock(
+            side_effect=RuntimeError("dispatch down"))
+        handler._send_json = mock.Mock(
+            side_effect=RuntimeError("send down"))
+        handler._dispatch_safe("GET")
+        handler._send_json.assert_called_once_with(
+            500, {"error": "internal_error"})
+
+
+class UploadMismatchTestCase(unittest.TestCase):
+    def test_paths_files_count_mismatch(self):
+        # 3149
+        handler = object.__new__(server_mod.CloudControlPlaneHandler)
+        handler.headers = {"Content-Type": "multipart/form-data",
+                           "Content-Length": "10"}
+        handler._read_body = mock.Mock(return_value=b"body")
+        sent = []
+        handler._send_json = lambda code, payload: sent.append(
+            (code, payload))
+        with mock.patch.object(
+                server_mod, "parse_multipart",
+                return_value=({}, [("a", b"x"), ("b", b"y")])):
+            with mock.patch.object(
+                    server_mod, "parse_paths_field",
+                    return_value=["only-one"]):
+                handler._create_upload("u1")
+        self.assertEqual(sent, [(400, {"error": "附件路径与文件数量不一致"})])
+
+
+class OutboxReplayTestCase(unittest.TestCase):
+    def test_failed_record_rewritten_back(self):
+        # 170: a failed replay row is rewritten into the outbox file.
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "outbox.jsonl")
+            good = {"task_id": "t1", "outcome": "success"}
+            bad = {"task_id": "t2", "outcome": "failure"}
+            with open(path, "w", encoding="utf-8") as f:
+                f.write(json.dumps(good) + "\n")
+                f.write(json.dumps(bad) + "\n")
+
+            def _record(conn, task_id, outcome, is_defect=False):
+                if task_id == "t2":
+                    raise RuntimeError("ledger down")
+
+            stats = replay_failures(path, None, record_fn=_record)
+            self.assertEqual(stats["replayed"], 1)
+            self.assertEqual(stats["failed"], 1)
+            self.assertEqual(stats["remaining"], 1)
+            with open(path, encoding="utf-8") as f:
+                rows = [json.loads(line) for line in f if line.strip()]
+            self.assertEqual(rows, [bad])
+
+
+if __name__ == "__main__":
+    unittest.main()