Преглед на файлове

策略:集中注入任务硬标准与方向约束

新增 TaskPolicyCatalog,按 Task kind 注入不可删除、不可降级的固定验收标准,并把 active Direction 的硬约束编译成受保护 criterion。Planner 只追加业务标准,重复或覆盖固定标准会在创建 Task 前被拒绝。
SamLee преди 18 часа
родител
ревизия
3b18731fb7
променени са 2 файла, в които са добавени 158 реда и са изтрити 0 реда
  1. 124 0
      script_build_host/src/script_build_host/domain/task_policy.py
  2. 34 0
      script_build_host/tests/test_task_policy.py

+ 124 - 0
script_build_host/src/script_build_host/domain/task_policy.py

@@ -0,0 +1,124 @@
+"""Host-owned acceptance policy for script task kinds."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from hashlib import sha256
+from typing import Any
+
+from .task_contracts import (
+    PlannerCriterionInput,
+    ScriptCriterion,
+    ScriptTaskKind,
+    stable_semantic_id,
+)
+
+_FIXED: dict[ScriptTaskKind, tuple[ScriptCriterion, ...]] = {
+    ScriptTaskKind.DIRECTION: (
+        ScriptCriterion(
+            "direction-coherent", "Direction goals and hard constraints are coherent", True
+        ),
+    ),
+    ScriptTaskKind.PATTERN_RETRIEVAL: (
+        ScriptCriterion(
+            "evidence-usable", "Retrieval returns usable and attributable evidence", True
+        ),
+    ),
+    ScriptTaskKind.DECODE_RETRIEVAL: (
+        ScriptCriterion(
+            "evidence-usable", "Retrieval returns usable and attributable evidence", True
+        ),
+    ),
+    ScriptTaskKind.EXTERNAL_RETRIEVAL: (
+        ScriptCriterion(
+            "evidence-usable", "Retrieval returns usable and attributable evidence", True
+        ),
+    ),
+    ScriptTaskKind.KNOWLEDGE_RETRIEVAL: (
+        ScriptCriterion(
+            "evidence-usable", "Retrieval returns usable and attributable evidence", True
+        ),
+    ),
+    ScriptTaskKind.STRUCTURE: (
+        ScriptCriterion("structure-complete", "Structure covers its declared content scope", True),
+    ),
+    ScriptTaskKind.PARAGRAPH: (
+        ScriptCriterion(
+            "paragraph-content-complete", "Paragraph contains complete publishable prose", True
+        ),
+    ),
+    ScriptTaskKind.ELEMENT_SET: (
+        ScriptCriterion(
+            "elements-linked", "Every Element is valid and linked to a Paragraph", True
+        ),
+    ),
+    ScriptTaskKind.COMPARE: (
+        ScriptCriterion(
+            "comparison-grounded", "Recommendation is grounded in the frozen candidates", True
+        ),
+    ),
+    ScriptTaskKind.COMPOSE: (
+        ScriptCriterion(
+            "script-content-complete",
+            "Composed script satisfies the complete content contract",
+            True,
+        ),
+        ScriptCriterion(
+            "direction-goals-satisfied", "All Direction goals have real creative coverage", True
+        ),
+    ),
+    ScriptTaskKind.CANDIDATE_PORTFOLIO: (
+        ScriptCriterion(
+            "portfolio-closed", "Portfolio closes and adopts exactly one valid script", True
+        ),
+        ScriptCriterion(
+            "direction-goals-satisfied", "All Direction goals have real creative coverage", True
+        ),
+    ),
+    ScriptTaskKind.ROOT_DELIVERY: (),
+}
+
+
+class TaskPolicyCatalog:
+    def criteria(
+        self,
+        *,
+        task_id: str,
+        task_kind: ScriptTaskKind,
+        business: tuple[PlannerCriterionInput, ...],
+        direction_constraints: Sequence[Any] = (),
+    ) -> tuple[ScriptCriterion, ...]:
+        fixed = _FIXED[task_kind]
+        fixed_ids = {item.criterion_id for item in fixed}
+        constraints = tuple(
+            ScriptCriterion(
+                f"direction-constraint-{item.constraint_id}",
+                f"Hard Direction constraint: {item.statement}",
+                True,
+            )
+            for item in direction_constraints
+        )
+        extra = tuple(
+            ScriptCriterion(
+                stable_semantic_id(task_id, "criterion", item.client_key),
+                item.description,
+                item.hard,
+            )
+            for item in business
+        )
+        protected_ids = fixed_ids | {item.criterion_id for item in constraints}
+        if protected_ids & {item.criterion_id for item in extra}:
+            raise ValueError("business criteria cannot override fixed criteria")
+        return (*fixed, *constraints, *extra)
+
+    @property
+    def manifest_digest(self) -> str:
+        encoded = "\n".join(
+            f"{kind.value}:{item.criterion_id}:{item.description}:{int(item.hard)}"
+            for kind in ScriptTaskKind
+            for item in _FIXED[kind]
+        )
+        return "sha256:" + sha256(encoded.encode()).hexdigest()
+
+
+__all__ = ["TaskPolicyCatalog"]

+ 34 - 0
script_build_host/tests/test_task_policy.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from script_build_host.domain.artifacts import DirectionConstraint
+from script_build_host.domain.task_contracts import PlannerCriterionInput, ScriptTaskKind
+from script_build_host.domain.task_policy import TaskPolicyCatalog
+
+
+def test_task_policy_injects_protected_kind_and_direction_constraint_criteria() -> None:
+    catalog = TaskPolicyCatalog()
+    criteria = catalog.criteria(
+        task_id="paragraph-task",
+        task_kind=ScriptTaskKind.PARAGRAPH,
+        business=(
+            PlannerCriterionInput(
+                client_key="business-quality",
+                description="the opening is memorable",
+                hard=False,
+            ),
+        ),
+        direction_constraints=(
+            DirectionConstraint(
+                constraint_id="facts-traceable",
+                statement="Every factual claim remains traceable",
+                rationale="hard publication rule",
+            ),
+        ),
+    )
+
+    assert criteria[0].criterion_id == "paragraph-content-complete"
+    assert criteria[0].hard is True
+    assert criteria[1].criterion_id == "direction-constraint-facts-traceable"
+    assert criteria[1].hard is True
+    assert criteria[2].description == "the opening is memorable"
+    assert criteria[2].hard is False