259 lines
11 KiB
Python
259 lines
11 KiB
Python
"""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()
|