gapfill G6: nail CloudExecutor and argv-only visibility gaps with tests only (pending_mirror_to_host_loop never claims applied; server passes argv-only, host verify gate blocks it, version gap plus leftover signal file are the observable trace); no production change, no invented channel

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()