Selaa lähdekoodia

refactor(输入解析): 复用 accepted input 兼容规则

让运行时 AcceptedInputResolver 调用领域层统一规则,不再维护第二套任务类型和作用域判断。将兼容错误稳定映射为 INPUT_SCOPE_MISMATCH,避免规划期与执行期校验漂移。
SamLee 1 päivä sitten
vanhempi
commit
3a4a9ba61a

+ 19 - 49
script_build_host/src/script_build_host/application/phase_two_inputs.py

@@ -20,6 +20,12 @@ from agent.orchestration import (
 from script_build_host.domain.artifacts import ArtifactState, ArtifactVersion
 from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import ScriptBuildError
+from script_build_host.domain.input_compatibility import (
+    AcceptedInputCompatibilityError,
+    AcceptedInputSource,
+    scopes_compatible,
+    validate_accepted_input,
+)
 from script_build_host.domain.phase_two_artifacts import CandidateLineageV1
 from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
 from script_build_host.domain.ports import ScriptBusinessArtifactRepository
@@ -538,7 +544,7 @@ def _input_snapshot_ref(value: str) -> str:
 
 
 def _scope_compatible(left: str, right: str) -> bool:
-    return left == right or left.startswith(f"{right}/") or right.startswith(f"{left}/")
+    return scopes_compatible(left, right)
 
 
 def _validate_consumer_input(
@@ -547,57 +553,21 @@ def _validate_consumer_input(
     *,
     direct_child: bool,
 ) -> None:
-    cross_scope_element_placement = (
-        not direct_child
-        and consumer.task_kind is ScriptTaskKind.ELEMENT_SET
-        and produced.task_kind is ScriptTaskKind.PARAGRAPH
-    )
-    if not _scope_compatible(consumer.scope_ref, produced.scope_ref) and not (
-        cross_scope_element_placement
-    ):
-        raise PhaseTwoInputError(
-            "INPUT_SCOPE_MISMATCH",
-            "accepted input is outside the consumer scope",
+    try:
+        validate_accepted_input(
+            consumer_kind=consumer.task_kind,
+            consumer_scope=consumer.scope_ref,
+            producer_kind=produced.task_kind,
+            producer_scope=produced.scope_ref,
+            source=(
+                AcceptedInputSource.DIRECT_CHILD if direct_child else AcceptedInputSource.EXPLICIT
+            ),
         )
-    retrieval = {
-        ScriptTaskKind.PATTERN_RETRIEVAL,
-        ScriptTaskKind.DECODE_RETRIEVAL,
-        ScriptTaskKind.EXTERNAL_RETRIEVAL,
-        ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
-    }
-    creative = {
-        ScriptTaskKind.STRUCTURE,
-        ScriptTaskKind.PARAGRAPH,
-        ScriptTaskKind.ELEMENT_SET,
-        ScriptTaskKind.COMPARE,
-    }
-    direct_allowed = {
-        ScriptTaskKind.DIRECTION: retrieval,
-        ScriptTaskKind.CANDIDATE_PORTFOLIO: {ScriptTaskKind.COMPOSE},
-        ScriptTaskKind.COMPOSE: creative | retrieval,
-        ScriptTaskKind.STRUCTURE: creative | retrieval,
-        ScriptTaskKind.PARAGRAPH: creative | retrieval,
-        ScriptTaskKind.ELEMENT_SET: creative | retrieval,
-        ScriptTaskKind.COMPARE: creative | retrieval,
-    }.get(consumer.task_kind, set())
-    explicit_allowed = {
-        ScriptTaskKind.DIRECTION: retrieval,
-        ScriptTaskKind.CANDIDATE_PORTFOLIO: {
-            ScriptTaskKind.COMPOSE,
-            ScriptTaskKind.COMPARE,
-        },
-        ScriptTaskKind.COMPOSE: creative | retrieval | {ScriptTaskKind.DIRECTION},
-        ScriptTaskKind.STRUCTURE: creative | retrieval | {ScriptTaskKind.DIRECTION},
-        ScriptTaskKind.PARAGRAPH: creative | retrieval | {ScriptTaskKind.DIRECTION},
-        ScriptTaskKind.ELEMENT_SET: creative | retrieval | {ScriptTaskKind.DIRECTION},
-        ScriptTaskKind.COMPARE: creative | retrieval | {ScriptTaskKind.DIRECTION},
-    }.get(consumer.task_kind, set())
-    allowed = direct_allowed if direct_child else explicit_allowed
-    if produced.task_kind not in allowed:
+    except AcceptedInputCompatibilityError as exc:
         raise PhaseTwoInputError(
             "INPUT_SCOPE_MISMATCH",
-            "accepted Task kind is not valid for the consumer contract and input source",
-        )
+            str(exc),
+        ) from exc
 
 
 def _writes_overlap(left: Iterable[str], right: Iterable[str]) -> bool: