Explorar el Código

候选产物:记录 Goal 血缘并要求最终覆盖

CandidateLineage 固化来自任务合同的 goal_ids,模型无法单独覆盖;StructuredScript 增加唯一持久化的 goal_coverage。

水合旧候选或旧 StructuredScript 时严格拒绝缺失字段,Portfolio 和 Root Manifest 不重复存储覆盖关系。
SamLee hace 1 día
padre
commit
2dbde0246b

+ 27 - 0
script_build_host/src/script_build_host/domain/phase_two_artifacts.py

@@ -9,6 +9,7 @@ from typing import Any, Self
 from .defects import MAX_DEFECTS_PER_VALIDATION, DefectSeverity, ScriptDefectV1
 from .digests import Sha256Digest
 from .errors import ProtocolViolation
+from .goal_coverage import GoalCoverage, GoalPolicyError
 
 _CONTROL_PREFIX = "script-build://"
 _ARTIFACT_PREFIX = "script-build://artifact-versions/"
@@ -20,6 +21,7 @@ class CandidateLineageV1:
     input_snapshot_ref: str
     input_closure_digest: str
     write_scope: tuple[str, ...]
+    goal_ids: tuple[str, ...]
     base_artifact_ref: str | None = None
     base_artifact_digest: str | None = None
     base_revision: int | None = None
@@ -31,6 +33,9 @@ class CandidateLineageV1:
         _control_uri(self.input_snapshot_ref, "input_snapshot_ref")
         Sha256Digest.from_wire(self.input_closure_digest)
         _unique_nonblank(self.write_scope, "write_scope")
+        if not self.goal_ids:
+            raise GoalPolicyError("GOAL_SCOPE_INVALID", "candidate lineage requires goal_ids")
+        _unique_nonblank(self.goal_ids, "goal_ids")
         for item in self.write_scope:
             _control_uri(item, "write_scope")
         _unique_nonblank(self.supersedes_decision_ids, "supersedes_decision_ids")
@@ -53,12 +58,15 @@ class CandidateLineageV1:
             "base_artifact_digest": self.base_artifact_digest,
             "base_revision": self.base_revision,
             "write_scope": list(self.write_scope),
+            "goal_ids": list(self.goal_ids),
             "supersedes_decision_ids": list(self.supersedes_decision_ids),
             "source_lineage": list(self.source_lineage),
         }
 
     @classmethod
     def from_payload(cls, value: Mapping[str, Any]) -> Self:
+        if "goal_ids" not in value:
+            raise GoalPolicyError("GOAL_SCOPE_INVALID", "candidate lineage requires goal_ids")
         return cls(
             scope_ref=str(value.get("scope_ref", "")),
             input_snapshot_ref=str(value.get("input_snapshot_ref", "")),
@@ -69,6 +77,7 @@ class CandidateLineageV1:
                 int(value["base_revision"]) if value.get("base_revision") is not None else None
             ),
             write_scope=_text_tuple(value.get("write_scope")),
+            goal_ids=_text_tuple(value.get("goal_ids")),
             supersedes_decision_ids=_text_tuple(value.get("supersedes_decision_ids")),
             source_lineage=_mapping_tuple(value.get("source_lineage")),
         )
@@ -332,6 +341,7 @@ class StructuredScriptArtifactV1:
     elements: tuple[ScriptElementV1, ...]
     paragraph_element_links: tuple[ScriptParagraphElementLinkV1, ...]
     source_artifact_refs: tuple[str, ...]
+    goal_coverage: tuple[GoalCoverage, ...]
     evidence_refs: tuple[str, ...]
     acceptance_notes: tuple[str, ...]
     canonical_sha256: str = ""
@@ -348,6 +358,10 @@ class StructuredScriptArtifactV1:
             _unique_nonblank(refs, label)
             for item in refs:
                 _artifact_uri(item, label)
+        if not self.goal_coverage:
+            raise GoalPolicyError(
+                "GOAL_COVERAGE_INCOMPLETE", "StructuredScript requires Goal coverage"
+            )
 
     def content_payload(self) -> dict[str, Any]:
         return {
@@ -358,6 +372,7 @@ class StructuredScriptArtifactV1:
             "elements": [item.to_payload() for item in self.elements],
             "paragraph_element_links": [item.to_payload() for item in self.paragraph_element_links],
             "source_artifact_refs": list(self.source_artifact_refs),
+            "goal_coverage": [item.to_payload() for item in self.goal_coverage],
             "evidence_refs": list(self.evidence_refs),
             "acceptance_notes": list(self.acceptance_notes),
         }
@@ -451,6 +466,10 @@ def hydrate_phase_two_artifact(payload: Mapping[str, Any]) -> PhaseTwoBusinessAr
             recommendation=str(payload.get("recommendation", "")),
         )
     if schema == "structured-script/v1":
+        if "goal_coverage" not in payload:
+            raise GoalPolicyError(
+                "GOAL_COVERAGE_INCOMPLETE", "StructuredScript requires goal_coverage"
+            )
         return StructuredScriptArtifactV1(
             direction_ref=str(payload.get("direction_ref", "")),
             input_closure_digest=str(payload.get("input_closure_digest", "")),
@@ -458,6 +477,7 @@ def hydrate_phase_two_artifact(payload: Mapping[str, Any]) -> PhaseTwoBusinessAr
             elements=_elements(payload.get("elements")),
             paragraph_element_links=_links(payload.get("paragraph_element_links")),
             source_artifact_refs=_text_tuple(payload.get("source_artifact_refs")),
+            goal_coverage=_goal_coverage(payload.get("goal_coverage")),
             evidence_refs=_text_tuple(payload.get("evidence_refs")),
             acceptance_notes=_text_tuple(payload.get("acceptance_notes")),
         )
@@ -570,11 +590,18 @@ def _links(value: Any) -> tuple[ScriptParagraphElementLinkV1, ...]:
     return tuple(ScriptParagraphElementLinkV1.from_payload(item) for item in _mapping_tuple(value))
 
 
+def _goal_coverage(value: Any) -> tuple[GoalCoverage, ...]:
+    if not isinstance(value, list):
+        raise ProtocolViolation("goal_coverage must be an object array")
+    return tuple(GoalCoverage.from_payload(item) for item in value)
+
+
 __all__ = [
     "CandidateLineageV1",
     "CandidatePortfolioArtifactV1",
     "ComparisonArtifactV1",
     "ElementSetArtifactV1",
+    "GoalCoverage",
     "ParagraphArtifactV1",
     "PhaseTwoBusinessArtifact",
     "ScriptElementV1",