codex-app-server/cloud/tests/test_audit_loop.py

"""有界自动审计循环测试：纯状态机 + worker 显式入口。

只测本轨：确定性门循环、推回包完整性、标准不弱化、上限升级、
全量回归、主观不循环、计数损坏 fail-closed、默认关闭。
"""

import json
import os
import sqlite3
import sys
import tempfile
import unittest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))

from cloud import audit_loop
from cloud.audit_loop import Finding, LoopState, evaluate
from cloud.worker import Worker


def det(fid, gate=None):
    return Finding(fid, "high", "evidence for %s" % fid,
                   "re-run %s" % (gate or fid),
                   "%s passes" % (gate or fid), gate=gate)


class PureEngineTests(unittest.TestCase):
    def test_pass_all_gates(self):
        out = evaluate(None, "t1", {"tests": True, "capability": True})
        self.assertEqual(out["decision"], "pass")
        self.assertEqual(out["round"], 0)
        self.assertIsNone(out["packet"])

    def test_pushback_packet_complete(self):
        out = evaluate(None, "t1", {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "pushback")
        self.assertEqual(out["round"], 1)
        packet = out["packet"]
        self.assertEqual(packet.findings[0].evidence, "evidence for f1")
        self.assertEqual(packet.findings[0].reproduction, "re-run tests")
        self.assertEqual(packet.findings[0].acceptance, "tests passes")
        self.assertIn("f1", packet.criteria)

    def test_invalid_finding_is_caller_bug(self):
        with self.assertRaises(ValueError):
            Finding("f1", "high", "", "repro", "accept")
        with self.assertRaises(ValueError):
            evaluate(None, "t1", {}, [])

    def test_escalate_at_third_round(self):
        state = LoopState("t1", round=3, passed_gates=[],
                          criteria={"f1": "tests"})
        out = evaluate(state, "t1", {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "escalate")
        self.assertEqual(out["reason"], "max_rounds_exceeded")
        self.assertEqual(out["round"], 3)

    def test_weakening_is_escalate(self):
        # 上一轮 finding f1（无关联门）本轮既不重报、又无 fixed 证据。
        state = LoopState("t1", round=1, passed_gates=[],
                          criteria={"f1": None})
        out = evaluate(state, "t1", {"other": True}, [])
        self.assertEqual(out["decision"], "escalate")
        self.assertEqual(out["reason"], "standards_weakened")
        self.assertIn("f1", out["weakened"])

    def test_fixed_via_gate_pass_is_not_weakening(self):
        state = LoopState("t1", round=1, passed_gates=[],
                          criteria={"f1": "tests"})
        out = evaluate(state, "t1", {"tests": True}, [])
        self.assertEqual(out["decision"], "pass")

    def test_regression_blocks_pass(self):
        state = LoopState("t1", round=1, passed_gates=["tests"],
                          criteria={})
        out = evaluate(state, "t1", {"tests": False}, [])
        self.assertEqual(out["decision"], "pushback")
        self.assertTrue(any(f.id == "regression:tests"
                            for f in out["packet"].findings))

    def test_missing_gate_escalates(self):
        state = LoopState("t1", round=1, passed_gates=["tests"],
                          criteria={})
        out = evaluate(state, "t1", {"other": True}, [])
        self.assertEqual(out["decision"], "escalate")
        self.assertEqual(out["reason"], "gate_missing")
        self.assertIn("tests", out["missing_gates"])

    def test_subjective_never_loops(self):
        sub = Finding("design", "medium", "looks off", "read code",
                      "reviewer approves", kind="subjective")
        out = evaluate(None, "t1", {"tests": True}, [sub])
        self.assertEqual(out["decision"], "needs_review")
        self.assertIsNone(out["new_state"])
        self.assertEqual(out["round"], 0)

    def test_corrupted_state_rejected(self):
        with self.assertRaises(ValueError):
            LoopState.from_dict({"task_id": "t1", "round": -1})
        with self.assertRaises(ValueError):
            LoopState.from_dict({"task_id": "t1", "round": "x"})


class WorkerEntryTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = os.path.join(self.tmp.name, "test.db")
        conn = sqlite3.connect(self.db)
        conn.execute(
            "CREATE TABLE IF NOT EXISTS task_attempts("
            "task_id TEXT, attempt_no INTEGER, kind TEXT,"
            " interrupted_reason TEXT, checkpoint_round INTEGER,"
            " result_text TEXT, error_text TEXT,"
            " created_at DOUBLE PRECISION,"
            " PRIMARY KEY(task_id, attempt_no))")
        conn.commit()
        conn.close()
        self.worker = Worker.__new__(Worker)
        self.worker.db_path = self.db
        self.worker.audits = []
        self.worker._audit = lambda p, a, d="": self.worker.audits.append(
            (p, a, d))

    def tearDown(self):
        self.tmp.cleanup()

    def task(self, loop=True):
        task = {"id": "task-1", "project_id": "proj-1"}
        if loop:
            task["audit_loop"] = True
        return task

    def test_off_by_default_touches_nothing(self):
        out = self.worker.audit_loop_evaluate(
            self.task(loop=False), {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "off")
        conn = sqlite3.connect(self.db)
        rows = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND"
            " name='task_audit_loop_state'").fetchall()
        conn.close()
        self.assertEqual(rows, [])

    def test_fail_fix_fail_fix_pass_path(self):
        # 轮 1：fail -> 推回；轮 2：fail -> 推回；轮 3：pass。
        out = self.worker.audit_loop_evaluate(
            self.task(), {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "pushback")
        self.assertEqual(out["round"], 1)
        self.assertEqual(out["packet"]["findings"][0]["id"], "f1")
        out = self.worker.audit_loop_evaluate(
            self.task(), {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "pushback")
        self.assertEqual(out["round"], 2)
        out = self.worker.audit_loop_evaluate(
            self.task(), {"tests": True}, [])
        self.assertEqual(out["decision"], "pass")

    def test_three_rounds_then_escalate(self):
        for expected in (1, 2, 3):
            out = self.worker.audit_loop_evaluate(
                self.task(), {"tests": False}, [det("f1", "tests")])
            self.assertEqual(out["decision"], "pushback")
            self.assertEqual(out["round"], expected)
        out = self.worker.audit_loop_evaluate(
            self.task(), {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "escalate")
        self.assertEqual(out["reason"], "max_rounds_exceeded")
        self.assertTrue(any(a == "worker_audit_loop_escalated"
                            for _, a, _ in self.worker.audits))

    def test_corrupted_count_never_loops(self):
        self.worker.audit_loop_evaluate(
            self.task(), {"tests": False}, [det("f1", "tests")])
        conn = sqlite3.connect(self.db)
        conn.execute(
            "UPDATE task_audit_loop_state SET round='not-a-number'"
            " WHERE task_id='task-1'")
        conn.commit()
        conn.close()
        out = self.worker.audit_loop_evaluate(
            self.task(), {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "escalate")
        self.assertEqual(out["reason"], "count_corrupted")
        self.assertTrue(any("count_corrupted" in d
                            for _, _, d in self.worker.audits))

    def test_state_survives_worker_restart(self):
        self.worker.audit_loop_evaluate(
            self.task(), {"tests": False}, [det("f1", "tests")])
        fresh = Worker.__new__(Worker)
        fresh.db_path = self.db
        fresh.audits = []
        fresh._audit = lambda p, a, d="": fresh.audits.append((p, a, d))
        out = fresh.audit_loop_evaluate(
            self.task(), {"tests": False}, [det("f1", "tests")])
        self.assertEqual(out["decision"], "pushback")
        self.assertEqual(out["round"], 2)

    def test_subjective_via_worker_needs_review(self):
        sub = Finding("design", "medium", "looks off", "read code",
                      "reviewer approves", kind="subjective")
        out = self.worker.audit_loop_evaluate(
            self.task(), {"tests": True}, [sub])
        self.assertEqual(out["decision"], "needs_review")


if __name__ == "__main__":
    unittest.main()