toki
0e12a540c8
로컬 모델의 작업 비중을 유지하면서 복합 위험과 반복 실패가 누적될 때만 cloud로 승격하고, 라우팅 자체의 재분석 비용과 반복 루프를 줄이기 위함이다.
325 lines
12 KiB
Python
Executable file
325 lines
12 KiB
Python
Executable file
#!/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"
|
|
|
|
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_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_three_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 >= 3 else "local-fit"
|
|
expected_lane = "cloud" if risk_count >= 3 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()
|