371 lines
15 KiB
Python
371 lines
15 KiB
Python
"""审计循环调用方接线测试：run_one 终局块 + 包投递 + 升级投递。
|
|
|
|
只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、
|
|
不联网、不登录任何服务；env 里的值只是占位用的假名。
|
|
|
|
覆盖（按任务书）：开关关闭时完成路径不变；确定性门 fail →
|
|
推回（任务重排、packet 可读、下一轮 prompt 含 packet）；修好
|
|
后 pass（packet 清理、任务完成）；3 轮仍 fail → 升级（任务终态、
|
|
审计在、不再重排）；主观 needs_review → 任务结局不变、不重排、
|
|
只记审计；计数损坏 → 直接升级、不循环。
|
|
"""
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
try:
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server, init_db
|
|
from cloud.worker import Worker
|
|
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server, init_db
|
|
from cloud.worker import Worker
|
|
|
|
|
|
def _insert_task(db_path, task_id, prompt="do work", status="queued",
|
|
project_id="p1", audit_loop=0, goal_mode=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, audit_loop,"
|
|
" goal_mode)"
|
|
" VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', ?, ?, NULL,"
|
|
" ?, ?)",
|
|
(task_id, project_id, prompt, status, now, now,
|
|
audit_loop, goal_mode),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _row(db_path, task_id):
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT status, error_text, audit_loop, audit_loop_packet"
|
|
" FROM tasks WHERE id = ?", (task_id,)).fetchone()
|
|
finally:
|
|
conn.close()
|
|
return row
|
|
|
|
|
|
def _packet(db_path, task_id):
|
|
row = _row(db_path, task_id)
|
|
if row is None or row[3] is None:
|
|
return None
|
|
return json.loads(row[3])
|
|
|
|
|
|
def _failing_runner(cmd):
|
|
name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
|
|
return SimpleNamespace(
|
|
returncode=0,
|
|
stdout=json.dumps({"task": name, "status": "failed",
|
|
"text": "boom"}),
|
|
stderr="")
|
|
|
|
|
|
def _completed_runner(cmd):
|
|
name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
|
|
return SimpleNamespace(
|
|
returncode=0,
|
|
stdout=json.dumps({"task": name, "status": "completed",
|
|
"text": "done"}),
|
|
stderr="")
|
|
|
|
|
|
class WireTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self._old_push_allow = os.environ.get(
|
|
"CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
|
|
os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
init_db(self.db_path)
|
|
self.env = {"XCM_API_KEY": "test-placeholder"}
|
|
self.calls = []
|
|
self.audits = []
|
|
|
|
def tearDown(self):
|
|
if getattr(self, "_old_push_allow", None) is None:
|
|
os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
|
|
else:
|
|
os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = (
|
|
self._old_push_allow)
|
|
self.tmp.cleanup()
|
|
|
|
def _worker(self, runner=None, **kwargs):
|
|
kwargs.setdefault("env", self.env)
|
|
kwargs.setdefault("max_retries", 0)
|
|
kwargs.setdefault("retry_backoff_seconds", 0)
|
|
worker = Worker(self.db_path, "/fake/host.py",
|
|
runner=runner or _completed_runner, **kwargs)
|
|
worker._audit = lambda p, a, d="": self.audits.append((p, a, d))
|
|
return worker
|
|
|
|
def _recording(self, inner):
|
|
def fake(cmd):
|
|
self.calls.append(list(cmd))
|
|
return inner(cmd)
|
|
return fake
|
|
|
|
# -- 开关关闭：完成路径不变 --------------------------------------
|
|
def test_off_task_completes_unchanged(self):
|
|
_insert_task(self.db_path, "T1", audit_loop=0)
|
|
out = self._worker().run_one()
|
|
self.assertEqual(out["status"], "completed")
|
|
self.assertEqual(_row(self.db_path, "T1")[0], "completed")
|
|
self.assertIsNone(_packet(self.db_path, "T1"))
|
|
|
|
# -- fail -> 推回 -> 修好 -> pass（完整 run_one 路径） ------------
|
|
def test_fail_pushback_then_pass_via_run_one(self):
|
|
_insert_task(self.db_path, "T2", audit_loop=1)
|
|
failing = self._worker(runner=self._recording(_failing_runner))
|
|
out = failing.run_one()
|
|
self.assertEqual(out["status"], "queued")
|
|
self.assertEqual(_row(self.db_path, "T2")[0], "queued")
|
|
packet = _packet(self.db_path, "T2")
|
|
self.assertIsNotNone(packet)
|
|
self.assertEqual(packet["round"], 1)
|
|
self.assertEqual(packet["findings"][0]["id"], "execution_failed")
|
|
self.assertTrue(any(a == "worker_audit_loop_pushback"
|
|
for _, a, _ in self.audits))
|
|
# 下一轮：包以前言贴进 prompt（库 prompt 本身不动），修好后 pass。
|
|
self.calls.clear()
|
|
self.audits.clear()
|
|
passing = self._worker(runner=self._recording(_completed_runner))
|
|
out = passing.run_one()
|
|
self.assertEqual(out["status"], "completed")
|
|
self.assertEqual(_row(self.db_path, "T2")[0], "completed")
|
|
self.assertIsNone(_packet(self.db_path, "T2"))
|
|
prompted = " ".join(" ".join(c) for c in self.calls)
|
|
self.assertIn("execution_failed", prompted)
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
prompt = conn.execute(
|
|
"SELECT prompt FROM tasks WHERE id = 'T2'").fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
self.assertEqual(prompt, "do work")
|
|
|
|
# -- 3 轮仍 fail -> 升级 -------------------------------------------
|
|
def test_three_rounds_then_escalate_via_run_one(self):
|
|
_insert_task(self.db_path, "T3", audit_loop=1)
|
|
worker = self._worker(runner=_failing_runner)
|
|
for _ in range(3):
|
|
out = worker.run_one()
|
|
self.assertEqual(out["status"], "queued")
|
|
out = worker.run_one()
|
|
self.assertEqual(out["status"], "failed")
|
|
self.assertEqual(_row(self.db_path, "T3")[0], "failed")
|
|
self.assertIsNone(_packet(self.db_path, "T3"))
|
|
self.assertIn("审计循环升级", _row(self.db_path, "T3")[1])
|
|
self.assertTrue(any(a == "worker_audit_loop_escalated"
|
|
for _, a, _ in self.audits))
|
|
# 升级后不再自动重排：下一轮认领不到该任务。
|
|
self.assertEqual(worker.run_one()["dispatched"], False)
|
|
|
|
# -- 计数损坏 -> 直接升级、不循环 ----------------------------------
|
|
def test_corrupted_count_escalates_via_wire(self):
|
|
_insert_task(self.db_path, "T4", status="failed", audit_loop=1)
|
|
worker = self._worker()
|
|
worker.audit_loop_evaluate(
|
|
{"id": "T4", "project_id": "p1", "audit_loop": True},
|
|
{"execution": False},
|
|
findings=[])
|
|
# 上面那行若因空 finding 报错也不影响：直接写坏状态行。
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO task_audit_loop_state"
|
|
"(task_id, round, passed_gates, criteria, last_decision,"
|
|
" updated_at) VALUES('T4', 'not-a-number', '[]', '{}',"
|
|
" 'pushback', ?)", (time.time(),))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
task = worker.get_task("T4")
|
|
final, error, decision = worker._audit_loop_wire(
|
|
task, "failed", {"stdout": ""}, "boom")
|
|
self.assertEqual(decision, "escalate")
|
|
self.assertEqual(final, "failed")
|
|
self.assertIn("审计循环升级", error)
|
|
self.assertIsNone(_packet(self.db_path, "T4"))
|
|
|
|
# -- goal 门：goal_met False 推回，None 不列入门 -------------------
|
|
def test_goal_met_false_pushbacks_and_none_is_not_a_gate(self):
|
|
worker = self._worker()
|
|
# 终局已写回的前提：库状态须与传入的 final 一致（completed），
|
|
# 否则接线的 CAS 会正确地拒绝改状态——那是测试搭错，不是 bug。
|
|
_insert_task(self.db_path, "T5", status="completed", audit_loop=1,
|
|
goal_mode=1)
|
|
task = worker.get_task("T5")
|
|
gates, findings = worker._audit_loop_gate_results(
|
|
task, "completed", {"goal_met": False, "goal_error": "nope"})
|
|
self.assertFalse(gates["goal_verify"])
|
|
self.assertEqual(findings[0].id, "goal_unmet")
|
|
final, _, decision = worker._audit_loop_wire(
|
|
task, "completed", {"goal_met": False, "goal_error": "nope"},
|
|
None)
|
|
self.assertEqual(decision, "pushback")
|
|
self.assertEqual(final, "queued")
|
|
gates, _ = worker._audit_loop_gate_results(
|
|
task, "completed", {"goal_met": None})
|
|
self.assertNotIn("goal_verify", gates)
|
|
|
|
# -- 主观 needs_review：结局不变、不重排、只记审计 -----------------
|
|
def test_needs_review_leaves_outcome_untouched(self):
|
|
_insert_task(self.db_path, "T6", status="failed", audit_loop=1)
|
|
worker = self._worker()
|
|
# 审计批 5：接线改走 _audit_loop_evaluate_on_conn（同连接
|
|
# 单事务），桩点随之移到新内核；意图不变——防御分支仍在，
|
|
# needs_review 时终局、包列一律不动、只记审计。
|
|
worker._audit_loop_evaluate_on_conn = (
|
|
lambda conn, t, g, findings=None, fixed=None: {
|
|
"enabled": True, "decision": "needs_review", "packet": None,
|
|
"new_state": None, "reason": "subjective"})
|
|
task = worker.get_task("T6")
|
|
final, _, decision = worker._audit_loop_wire(
|
|
task, "failed", {"stdout": ""}, "boom")
|
|
self.assertEqual(decision, "needs_review")
|
|
self.assertEqual(final, "failed")
|
|
self.assertEqual(_row(self.db_path, "T6")[0], "failed")
|
|
self.assertTrue(any(a == "worker_audit_loop_needs_review"
|
|
for _, a, _ in self.audits))
|
|
|
|
# -- 包注入：有效包贴前言、无效包不动、重复调用不叠加 --------------
|
|
def test_packet_injection_and_invalid_packet(self):
|
|
worker = self._worker()
|
|
packet = json.dumps({
|
|
"task_id": "T7", "round": 1,
|
|
"findings": [{"id": "execution_failed", "severity": "high",
|
|
"evidence": "boom", "reproduction": "re-run",
|
|
"acceptance": "task completes",
|
|
"kind": "deterministic", "gate": "execution"}],
|
|
"criteria": {"execution_failed": "execution"}})
|
|
task = {"id": "T7", "project_id": "p1", "prompt": "do work",
|
|
"audit_loop_packet": packet}
|
|
out = worker._apply_audit_loop_packet(task)
|
|
self.assertIn("execution_failed", out["prompt"])
|
|
self.assertTrue(out["prompt"].startswith("[审计推回包"))
|
|
again = worker._apply_audit_loop_packet(out)
|
|
self.assertEqual(again["prompt"].count("execution_failed"), 1)
|
|
bad = worker._apply_audit_loop_packet(
|
|
{"id": "T7", "project_id": "p1", "prompt": "do work",
|
|
"audit_loop_packet": "{not json"})
|
|
self.assertEqual(bad["prompt"], "do work")
|
|
|
|
# -- 列与回读：init_db 建列、get_task 读到开关 ----------------------
|
|
def test_columns_and_get_task_flag(self):
|
|
_insert_task(self.db_path, "T8", audit_loop=1)
|
|
worker = self._worker()
|
|
task = worker.get_task("T8")
|
|
self.assertEqual(task["audit_loop"], 1)
|
|
self.assertIsNone(task["audit_loop_packet"])
|
|
|
|
|
|
class ServerCreateTests(unittest.TestCase):
|
|
"""POST /tasks 的 audit_loop 字段：接受、拒绝、协调员拒绝。
|
|
|
|
复用 test_server_goal_notify 的 create_server + 真实 HTTP 骨架。
|
|
"""
|
|
|
|
def setUp(self):
|
|
self._old_push_allow = os.environ.get(
|
|
"CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
|
|
os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
self.auth = AuthStore(self.db_path)
|
|
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
|
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
|
self.port = self.server.server_address[1]
|
|
self.thread = threading.Thread(
|
|
target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=5)
|
|
if getattr(self, "_old_push_allow", None) is None:
|
|
os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
|
|
else:
|
|
os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = (
|
|
self._old_push_allow)
|
|
try:
|
|
self.auth.conn.close()
|
|
except Exception:
|
|
pass
|
|
self.tmp.cleanup()
|
|
|
|
def _req(self, method, path, body=None, key=None):
|
|
conn = http.client.HTTPConnection(
|
|
"127.0.0.1", self.port, timeout=10)
|
|
hdrs = {}
|
|
payload = None
|
|
if body is not None:
|
|
payload = json.dumps(body).encode("utf-8")
|
|
hdrs["Content-Type"] = "application/json"
|
|
if key is not None:
|
|
hdrs["X-API-Key"] = key
|
|
try:
|
|
conn.request(method, path, body=payload, headers=hdrs)
|
|
resp = conn.getresponse()
|
|
data = resp.read()
|
|
finally:
|
|
conn.close()
|
|
try:
|
|
parsed = json.loads(data.decode("utf-8")) if data else None
|
|
except ValueError:
|
|
parsed = None
|
|
return resp.status, parsed
|
|
|
|
def test_create_task_audit_loop_flag(self):
|
|
_, project = self._req(
|
|
"POST", "/projects", {"name": "p"}, self.owner_key)
|
|
pid = project["id"]
|
|
status, task = self._req(
|
|
"POST", "/tasks",
|
|
{"project_id": pid, "prompt": "do it", "audit_loop": True},
|
|
self.owner_key)
|
|
self.assertEqual(status, 201, task)
|
|
self.assertEqual(task["audit_loop"], 1)
|
|
status, _ = self._req(
|
|
"POST", "/tasks",
|
|
{"project_id": pid, "prompt": "bad", "audit_loop": "yes"},
|
|
self.owner_key)
|
|
self.assertEqual(status, 400)
|
|
# 协调员接线轨：父任务带 audit_loop 不再 400——此前拒绝
|
|
# 的理由是“父任务接线覆盖不到、接受即静默丢弃”，合流
|
|
# 接线（worker._coord_audit_loop_merge）落地后该理由已
|
|
# 消除；子任务仍不直接接受外部开关（worker 内部 INSERT
|
|
# 不写该列）。非法值仍 400。
|
|
status, task = self._req(
|
|
"POST", "/tasks",
|
|
{"project_id": pid, "prompt": "coord",
|
|
"coordinator_parts": [{"name": "a", "prompt": "do a"}],
|
|
"audit_loop": True},
|
|
self.owner_key)
|
|
self.assertEqual(status, 201, task)
|
|
self.assertEqual(task["audit_loop"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|