Przeglądaj źródła

目标规则:新增统一的 Goal 作用域与覆盖模型

集中定义 Goal 数量上限、合同 Goal 校验、创作来源覆盖推导和闭包校验。

统一返回 GOAL_SCOPE_INVALID、GOAL_COVERAGE_INCOMPLETE 与 GOAL_COVERAGE_INVALID,避免规划、组合和最终交付各自维护一套规则。
SamLee 1 dzień temu
rodzic
commit
12e00ae0f3

+ 147 - 0
script_build_host/src/script_build_host/domain/goal_coverage.py

@@ -0,0 +1,147 @@
+"""Pure Goal scope and coverage rules shared by planning and delivery."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Sequence
+from dataclasses import dataclass
+
+from .errors import ScriptBuildError
+
+MAX_DIRECTION_GOALS = 30
+_ARTIFACT_PREFIX = "script-build://artifact-versions/"
+
+
+class GoalPolicyError(ScriptBuildError):
+    def __init__(self, code: str, summary: str) -> None:
+        super().__init__(code, summary)
+
+
+@dataclass(frozen=True, slots=True)
+class GoalCoverage:
+    goal_id: str
+    source_artifact_refs: tuple[str, ...]
+
+    def __post_init__(self) -> None:
+        if not self.goal_id.strip():
+            raise GoalPolicyError("GOAL_COVERAGE_INVALID", "goal_id must not be blank")
+        if not self.source_artifact_refs:
+            raise GoalPolicyError(
+                "GOAL_COVERAGE_INCOMPLETE",
+                f"Goal {self.goal_id} has no creative implementation source",
+            )
+        if len(set(self.source_artifact_refs)) != len(self.source_artifact_refs):
+            raise GoalPolicyError(
+                "GOAL_COVERAGE_INVALID",
+                f"Goal {self.goal_id} repeats a creative source",
+            )
+        if any(not item.startswith(_ARTIFACT_PREFIX) for item in self.source_artifact_refs):
+            raise GoalPolicyError(
+                "GOAL_COVERAGE_INVALID",
+                f"Goal {self.goal_id} uses a non-immutable creative source",
+            )
+
+    def to_payload(self) -> dict[str, object]:
+        return {
+            "goal_id": self.goal_id,
+            "source_artifact_refs": list(self.source_artifact_refs),
+        }
+
+    @classmethod
+    def from_payload(cls, value: object) -> GoalCoverage:
+        if not isinstance(value, dict):
+            raise GoalPolicyError("GOAL_COVERAGE_INVALID", "goal coverage must be an object")
+        refs = value.get("source_artifact_refs")
+        if not isinstance(refs, list) or any(not isinstance(item, str) for item in refs):
+            raise GoalPolicyError(
+                "GOAL_COVERAGE_INVALID", "source_artifact_refs must be a string array"
+            )
+        return cls(
+            goal_id=str(value.get("goal_id", "")),
+            source_artifact_refs=tuple(refs),
+        )
+
+
+def validate_goal_ids(
+    goal_ids: Sequence[str],
+    *,
+    allowed_goal_ids: Sequence[str],
+    require_all: bool = False,
+) -> tuple[str, ...]:
+    """Validate one contract Goal set without imposing exclusive ownership."""
+
+    values = tuple(goal_ids)
+    allowed = tuple(allowed_goal_ids)
+    if not values or len(values) > MAX_DIRECTION_GOALS:
+        raise GoalPolicyError(
+            "GOAL_SCOPE_INVALID",
+            f"goal_ids must contain between 1 and {MAX_DIRECTION_GOALS} Goal IDs",
+        )
+    if any(not item.strip() for item in values) or len(set(values)) != len(values):
+        raise GoalPolicyError("GOAL_SCOPE_INVALID", "goal_ids must be nonblank and unique")
+    unknown = set(values) - set(allowed)
+    if unknown:
+        raise GoalPolicyError(
+            "GOAL_SCOPE_INVALID", f"unknown Goal IDs: {sorted(unknown)}"
+        )
+    if require_all and set(values) != set(allowed):
+        raise GoalPolicyError(
+            "GOAL_SCOPE_INVALID", "this Task must cover every active Direction Goal"
+        )
+    return values
+
+
+def build_goal_coverage(
+    *,
+    direction_goal_ids: Sequence[str],
+    creative_sources: Iterable[tuple[str, Sequence[str]]],
+) -> tuple[GoalCoverage, ...]:
+    """Derive ordered coverage solely from adopted creative artifacts."""
+
+    ordered_goals = tuple(direction_goal_ids)
+    by_goal: dict[str, list[str]] = {goal_id: [] for goal_id in ordered_goals}
+    for artifact_ref, goal_ids in creative_sources:
+        validated = validate_goal_ids(goal_ids, allowed_goal_ids=ordered_goals)
+        for goal_id in validated:
+            if artifact_ref not in by_goal[goal_id]:
+                by_goal[goal_id].append(artifact_ref)
+    missing = [goal_id for goal_id, refs in by_goal.items() if not refs]
+    if missing:
+        raise GoalPolicyError(
+            "GOAL_COVERAGE_INCOMPLETE",
+            f"Goals without creative implementation sources: {missing}",
+        )
+    return tuple(
+        GoalCoverage(goal_id=goal_id, source_artifact_refs=tuple(by_goal[goal_id]))
+        for goal_id in ordered_goals
+    )
+
+
+def validate_goal_coverage(
+    *,
+    direction_goal_ids: Sequence[str],
+    coverage: Sequence[GoalCoverage],
+    adopted_source_refs: Sequence[str],
+) -> None:
+    """Prove exact all-node coverage and adopted-closure membership."""
+
+    expected = tuple(direction_goal_ids)
+    actual = tuple(item.goal_id for item in coverage)
+    if len(set(actual)) != len(actual) or actual != expected:
+        code = (
+            "GOAL_COVERAGE_INCOMPLETE"
+            if set(actual) < set(expected)
+            else "GOAL_COVERAGE_INVALID"
+        )
+        raise GoalPolicyError(code, "Goal coverage must match every Direction Goal in order")
+    adopted = set(adopted_source_refs)
+    dangling = {
+        ref
+        for item in coverage
+        for ref in item.source_artifact_refs
+        if ref not in adopted
+    }
+    if dangling:
+        raise GoalPolicyError(
+            "GOAL_COVERAGE_INVALID",
+            f"Goal coverage references non-adopted artifacts: {sorted(dangling)}",
+        )