#!/usr/bin/env python3 from __future__ import annotations import itertools import subprocess import unittest from pathlib import Path SKILL_DIR = Path(__file__).resolve().parents[1] FORMATTER = SKILL_DIR / "scripts" / "finalize-task-route.sh" POLICY_FINALIZER = SKILL_DIR / "scripts" / "finalize-task-policy.sh" PLAN_SKILL = SKILL_DIR.parent / "plan" / "SKILL.md" COMMON_SKILLS_DIR = SKILL_DIR.parent COMMON_RULES_DIR = SKILL_DIR.parents[2] / "rules" / "common" GRADE_VECTORS = { 1: (0, 0, 0, 0, 0), 2: (2, 0, 0, 0, 0), 3: (2, 1, 0, 0, 0), 4: (2, 2, 0, 0, 0), 5: (2, 2, 1, 0, 0), 6: (2, 2, 2, 0, 0), 7: (2, 2, 2, 1, 0), 8: (2, 2, 2, 2, 0), 9: (2, 2, 2, 2, 1), 10: (2, 2, 2, 2, 2), } def run( script: Path, *args: object, check: bool = True ) -> subprocess.CompletedProcess[str]: return subprocess.run( [str(script), *(str(arg) for arg in args)], check=check, capture_output=True, text=True, timeout=5, ) def fields(stdout: str) -> dict[str, str]: return dict(line.split("=", 1) for line in stdout.splitlines()) def expected_build_route(basis: str, grade: int) -> tuple[str, bool]: if basis == "local-fit": return "local", grade <= 8 if basis == "capability-gap": return "cloud", True return "cloud", grade >= 9 def pair_args( build_basis: str, build_grade: int, review_grade: int, large_context: str = "false", risk_count: int = 0, review_rework_count: int = 0, evidence_integrity: str = "false", review_basis: str = "official-review", ) -> tuple[object, ...]: return ( "pair", build_basis, large_context, risk_count, review_rework_count, evidence_integrity, *GRADE_VECTORS[build_grade], review_basis, *GRADE_VECTORS[review_grade], ) class FinalizeTaskRoutingTests(unittest.TestCase): def assert_route( self, result: dict[str, str], target: str, basis: str, lane: str, grade: int, ) -> None: prefix = "PLAN" if target == "build" else "CODE_REVIEW" self.assertEqual(result[f"{target}_route_basis"], basis) self.assertEqual(result[f"{target}_lane"], lane) self.assertEqual(result[f"{target}_grade"], f"G{grade:02d}") self.assertEqual( result[f"{target}_filename"], f"{prefix}-{lane}-G{grade:02d}.md" ) def test_common_workflows_do_not_depend_on_project_runtime(self) -> None: contract_roots = ( COMMON_RULES_DIR / "rules-roadmap.md", COMMON_SKILLS_DIR / "create-roadmap", COMMON_SKILLS_DIR / "update-roadmap", COMMON_SKILLS_DIR / "sync-milestone-workstate", COMMON_SKILLS_DIR / "complete-milestone", COMMON_SKILLS_DIR / "plan", COMMON_SKILLS_DIR / "code-review", COMMON_SKILLS_DIR / "refine-local-plans", COMMON_SKILLS_DIR / "finalize-task-routing", ) contract_files: list[Path] = [] for root in contract_roots: candidates = (root,) if root.is_file() else root.rglob("*") contract_files.extend( path for path in candidates if path.is_file() and path.suffix in {".md", ".sh", ".yaml"} and "tests" not in path.parts ) forbidden = ( "agent-ops/skills/project/", "orchestrate-agent-task-loop", "WORK_LOG.md", "work_log_", "dispatch.py", ) for path in sorted(contract_files): text = path.read_text(encoding="utf-8") for needle in forbidden: with self.subTest(path=path, needle=needle): self.assertNotIn(needle, text) def test_request_to_worker_contract_is_ordered_and_consistent(self) -> None: routing_skill = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8") plan_skill = PLAN_SKILL.read_text(encoding="utf-8") self.assertIn("`matched_loop_risk_signatures`", routing_skill) self.assertIn("`temporal_state`", routing_skill) self.assertNotIn("`ordered_transitions`", routing_skill) self.assertIn("routing 때문에 source/test/log를 다시 읽지", routing_skill) self.assertIn("첫 번째로 일치하는 규칙", routing_skill) self.assertIn("이미 cloud인 capability/grade basis를 덮어쓰지 않는다", routing_skill) self.assertIn("review gap만 기록; build basis에는 영향 없음", routing_skill) markers = ( "**Assess split boundaries once**", "**Derive routing signals once**", "## Step 3 - Finalize Task Routing", ) positions = [plan_skill.index(marker) for marker in markers] self.assertEqual(positions, sorted(positions)) self.assertNotIn("**Extract the request contract**", plan_skill) self.assertNotIn("`요청 계약`", plan_skill) self.assertIn("reuse the values already validated and appended", plan_skill) policy_script = POLICY_FINALIZER.read_text(encoding="utf-8") self.assertNotIn("build_risk_triggered", policy_script) self.assertNotIn("boundary_triggered", policy_script) def test_formatter_grade_and_filename_matrix(self) -> None: for target, lane, grade in itertools.product( ("build", "review"), ("local", "cloud"), range(1, 11) ): with self.subTest(target=target, lane=lane, grade=grade): result = fields(run(FORMATTER, target, lane, *GRADE_VECTORS[grade]).stdout) prefix = "PLAN" if target == "build" else "CODE_REVIEW" self.assertEqual(result["grade"], f"G{grade:02d}") self.assertEqual( result["filename"], f"{prefix}-{lane}-G{grade:02d}.md" ) def test_formatter_invalid_inputs_are_rejected(self) -> None: valid = list(GRADE_VECTORS[4]) invalid_cases: list[tuple[object, ...]] = [ ("deploy", "local", *valid), ("build", "hybrid", *valid), ("build", "local", *valid[:-1]), ("build", "local", *valid, 1), ] for index in range(5): scores = valid.copy() scores[index] = 3 invalid_cases.append(("build", "local", *scores)) for args in invalid_cases: with self.subTest(args=args): self.assertEqual(run(FORMATTER, *args, check=False).returncode, 2) def test_pair_policy_validates_each_build_basis_and_grade(self) -> None: for basis, grade in itertools.product( ("local-fit", "capability-gap", "grade-boundary"), range(1, 11) ): with self.subTest(basis=basis, grade=grade): completed = run( POLICY_FINALIZER, *pair_args(basis, grade, 1), check=False, ) lane, valid = expected_build_route(basis, grade) if not valid: self.assertEqual(completed.returncode, 2) continue self.assertEqual(completed.returncode, 0, completed.stderr) result = fields(completed.stdout) self.assertEqual(result["finalizer_mode"], "pair") self.assert_route(result, "build", basis, lane, grade) def test_official_review_keeps_grade_and_fixes_execution_target(self) -> None: for grade in range(1, 11): with self.subTest(grade=grade): result = fields( run(POLICY_FINALIZER, *pair_args("local-fit", 1, grade)).stdout ) self.assert_route( result, "review", "official-review", "cloud", grade ) self.assertEqual(result["review_adapter"], "codex") self.assertEqual(result["review_model"], "gpt-5.6-sol") self.assertEqual(result["review_reasoning_effort"], "xhigh") def test_low_grade_cloud_requires_capability_gap_basis(self) -> None: rejected = run( POLICY_FINALIZER, *pair_args("grade-boundary", 4, 4), check=False, ) self.assertEqual(rejected.returncode, 2) accepted = fields( run( POLICY_FINALIZER, *pair_args("capability-gap", 4, 4), ).stdout ) self.assert_route(accepted, "build", "capability-gap", "cloud", 4) def test_loop_risk_boundary_routes_four_or_more_signatures_to_cloud(self) -> None: for risk_count in range(6): with self.subTest(risk_count=risk_count): result = fields( run( POLICY_FINALIZER, *pair_args("local-fit", 7, 6, risk_count=risk_count), ).stdout ) self.assertEqual(result["build_loop_risk_count"], str(risk_count)) expected_basis = "risk-boundary" if risk_count >= 4 else "local-fit" expected_lane = "cloud" if risk_count >= 4 else "local" self.assertEqual(result["build_base_route_basis"], "local-fit") self.assert_route(result, "build", expected_basis, expected_lane, 7) def test_large_indivisible_context_routes_to_cloud(self) -> None: result = fields( run( POLICY_FINALIZER, *pair_args("local-fit", 4, 4, large_context="true"), ).stdout ) self.assertEqual(result["build_risk_boundary_matched"], "true") self.assert_route(result, "build", "risk-boundary", "cloud", 4) def test_review_rework_and_evidence_integrity_route_to_cloud(self) -> None: one_rework = fields( run( POLICY_FINALIZER, *pair_args("local-fit", 4, 4, review_rework_count=1), ).stdout ) self.assert_route(one_rework, "build", "local-fit", "local", 4) two_reworks = fields( run( POLICY_FINALIZER, *pair_args("local-fit", 4, 4, review_rework_count=2), ).stdout ) self.assert_route( two_reworks, "build", "recovery-boundary", "cloud", 4 ) evidence_failure = fields( run( POLICY_FINALIZER, *pair_args("local-fit", 4, 4, evidence_integrity="true"), ).stdout ) self.assert_route( evidence_failure, "build", "recovery-boundary", "cloud", 4 ) recovery_over_risk = fields( run( POLICY_FINALIZER, *pair_args( "local-fit", 4, 4, large_context="true", review_rework_count=2, ), ).stdout ) self.assertEqual(recovery_over_risk["build_base_route_basis"], "local-fit") self.assert_route( recovery_over_risk, "build", "recovery-boundary", "cloud", 4 ) def test_grade_boundary_is_not_relabelled_by_escalation_signals(self) -> None: result = fields( run( POLICY_FINALIZER, *pair_args( "grade-boundary", 9, 4, large_context="true", risk_count=5, review_rework_count=2, evidence_integrity="true", ), ).stdout ) self.assertEqual(result["build_base_route_basis"], "grade-boundary") self.assertEqual(result["build_risk_boundary_matched"], "true") self.assertEqual(result["build_recovery_boundary_matched"], "true") self.assert_route(result, "build", "grade-boundary", "cloud", 9) def test_capability_gap_has_precedence_over_other_cloud_triggers(self) -> None: result = fields( run( POLICY_FINALIZER, *pair_args( "capability-gap", 4, 4, large_context="true", risk_count=5, review_rework_count=2, evidence_integrity="true", ), ).stdout ) self.assertEqual(result["build_base_route_basis"], "capability-gap") self.assert_route(result, "build", "capability-gap", "cloud", 4) def test_policy_invalid_inputs_are_rejected(self) -> None: valid = GRADE_VECTORS[4] invalid_cases = ( ("pair", "local-fit", "unknown", 0, 0, "false", *valid, "official-review", *valid), ("pair", "local-fit", "false", 6, 0, "false", *valid, "official-review", *valid), ("pair", "local-fit", "false", 0, -1, "false", *valid, "official-review", *valid), ("pair", "local-fit", "false", 0, 0, "unknown", *valid, "official-review", *valid), ("pair", "local", "false", 0, 0, "false", *valid, "official-review", *valid), ("pair", "local-fit", "false", 0, 0, "false", *valid, "cloud", *valid), ("build", "local-fit", "false", 0, 0, "false", *valid), ("pair", "local-fit", "false", 0, 0, "false", *valid, "official-review", *valid[:-1]), ) for args in invalid_cases: with self.subTest(args=args): self.assertEqual( run(POLICY_FINALIZER, *args, check=False).returncode, 2, ) if __name__ == "__main__": unittest.main()