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
This commit is contained in:
host committed
1 parent
29af9c7bbf
commit
7524a779f2
1 file changed
+258
@@ -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()
|
||||
Reference in new issue
Block a user