diff --git a/backend/services/review_plan.py b/backend/services/review_plan.py index 6bbed0a..8ffe64c 100644 --- a/backend/services/review_plan.py +++ b/backend/services/review_plan.py @@ -181,3 +181,50 @@ def assemble(db, plan_id: str, legacy_catalog=None) -> dict: frozen["dimensions"] = dimensions frozen["presentation"] = result return frozen + + +# ---------------------------------------------------------------- 方案治理（T06-c） +# ADR-11：Tools 由后台批准，**不开放业务 Tools 卡**。方案只能引用已登记工具， +# 不得自带工具配置面——否则业务侧就能绕过工具白名单。 +FORBIDDEN_PLAN_KEYS = ("tools", "tool_cards", "tool_config", "allowed_tools", "permissions") + + +def _plan_codes(frozen: dict) -> set[str]: + """方案内产生的核查编码：Skill 的 code 与规则条件的 code。""" + codes = {skill.get("code") for skill in frozen.get("skills", []) if skill.get("code")} + for rule in frozen.get("rules", []): + codes |= {item.get("code") for item in rule.get("conditions", []) if item.get("code")} + return codes + + +def validate_critical_checks(frozen: dict, critical_check_ids) -> None: + """关键核查项必须真由本方案的 Skill/规则产生，不能凭空声明。 + + 否则"关键项"就成了无法追溯的标签，会绕过 BR-06 的硬性阻断语义。 + """ + declared = list(critical_check_ids or ()) + if not declared: + return + available = _plan_codes(frozen) + unknown = sorted(set(declared) - available) + if unknown: + raise ValueError(f"关键核查项不属于本方案的 Skill 或规则：{unknown}") + + +def propose_plan(db, body: dict, actor: str, project_id: str): + """提议一份方案草稿。复用既有组件治理，不新建发布路径。""" + forbidden = sorted(key for key in FORBIDDEN_PLAN_KEYS if key in body) + if forbidden: + raise ValueError( + "方案不得自带工具配置面（不开放业务 Tools 卡）；" + f"工具由后台治理，请移除：{forbidden}" + ) + if body.get("kind") != "plan": + raise ValueError("提案类型必须是 plan") + + spec_inputs = body.get("dimensions", []) + resolved = resolve_dimensions(spec_inputs) # 去重与刻度冲突在此拦截 + result = presentation(resolved) + assert_nothing_dropped(result, resolved) + + return registry.draft(db, dict(body, project_id=project_id, dimensions=resolved), actor) diff --git a/backend/test_review_plan.py b/backend/test_review_plan.py index fcfab48..b0cb6a5 100644 --- a/backend/test_review_plan.py +++ b/backend/test_review_plan.py @@ -251,5 +251,97 @@ class PlanAssemblyTests(unittest.TestCase): self.assertIn("不一致定义", str(ctx.exception)) + + +class PlanGovernanceTests(unittest.TestCase): + """T06-c：方案治理复用既有入口；不得出现业务 Tools 卡（ADR-11）。""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db_path = Path(self._tmp.name) / "governance.sqlite3" + store.initialize(self.db_path) + with store.transaction(self.db_path) as db: + registry.seed(db) + + def tearDown(self): + self._tmp.cleanup() + + def plan_body(self, **over): + body = {"id": "corporate.custom", "kind": "plan", "name": "自定义方案", + "version": "1.0.0", "scope": "demo", "skills": ["corporate.cashflow"], + "rules": ["demo.coverage"], "template": "template.ai-pre-review", + "dimensions": [spec("d1", order=0), spec("d2", "负债率", 1), spec("d3", "设备", 2)]} + body.update(over) + return body + + def propose(self, body=None, actor="demo-manager"): + with store.transaction(self.db_path) as db: + return plan.propose_plan(db, body or self.plan_body(), actor, "DEMO-A") + + def test_plan_proposal_creates_a_draft_only(self): + draft = self.propose() + self.assertEqual(draft["status"], "draft") + with store.transaction(self.db_path) as db: + published = [r for r in store.listing(db, "component_version") + if r.get("component_id") == "corporate.custom"] + self.assertEqual(published, [], "提议方案不得直接生效") + + def test_business_tools_card_is_rejected(self): + # ADR-11：不开放业务 Tools 卡，方案不得自带工具配置面 + for key in ("tools", "tool_cards", "allowed_tools", "permissions"): + with self.subTest(key=key): + with self.assertRaises(ValueError) as ctx: + self.propose(self.plan_body(**{key: ["terminal"]})) + self.assertIn("不开放业务 Tools 卡", str(ctx.exception)) + + def test_dimensions_are_deduplicated_at_proposal(self): + draft = self.propose(self.plan_body(dimensions=[spec("d1"), spec("d1")])) + with store.transaction(self.db_path) as db: + row = store.get(db, "component_draft", draft["id"]) + self.assertEqual(len(row["dimensions"]), 1, "相同定义只应保留一份") + + def test_conflicting_dimensions_are_rejected_at_proposal(self): + with self.assertRaises(ValueError) as ctx: + self.propose(self.plan_body(dimensions=[spec("d1"), spec("d1", scale_version="v2")])) + self.assertIn("不一致定义", str(ctx.exception)) + + def test_self_approval_is_rejected_by_existing_governance(self): + draft = self.propose() + with store.transaction(self.db_path) as db: + with self.assertRaises(PermissionError) as ctx: + registry.release(db, draft["id"], "demo-manager") + self.assertIn("独立审核人", str(ctx.exception)) + + def test_only_plans_can_be_proposed_through_this_entry(self): + with self.assertRaises(ValueError): + self.propose(self.plan_body(kind="skill")) + + +class CriticalCheckTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db_path = Path(self._tmp.name) / "checks.sqlite3" + store.initialize(self.db_path) + with store.transaction(self.db_path) as db: + registry.seed(db) + self.frozen = plan.assemble(db, "corporate.default") + + def tearDown(self): + self._tmp.cleanup() + + def test_real_codes_are_accepted(self): + codes = sorted(plan._plan_codes(self.frozen)) + self.assertTrue(codes) + plan.validate_critical_checks(self.frozen, codes[:1]) + + def test_invented_critical_check_is_rejected(self): + with self.assertRaises(ValueError) as ctx: + plan.validate_critical_checks(self.frozen, ["not.a.real.code"]) + self.assertIn("不属于本方案", str(ctx.exception)) + + def test_empty_critical_list_is_allowed(self): + plan.validate_critical_checks(self.frozen, []) + + if __name__ == "__main__": unittest.main()