from __future__ import annotations from dataclasses import replace from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from agent.orchestration import TaskStatus from script_build_host.application.phase_three import ( PHASE_THREE_EXECUTION_STARTED, PHASE_THREE_POLICY, PhaseThreeContinuationService, ) from script_build_host.domain.errors import ( LegacyCanonicalIdentityConflict, MissionRecoveryRequired, PhaseThreeBoundaryNotReady, ProtocolViolation, ) from script_build_host.domain.phase_three_artifacts import ( RootDeliveryManifestV1, canonicalize_legacy_projection, ) from script_build_host.domain.phase_two_artifacts import ( GoalCoverage, ScriptElementV1, ScriptParagraphElementLinkV1, ScriptParagraphV1, StructuredScriptArtifactV1, ) from script_build_host.domain.records import BuildStatus _DIGEST = "sha256:" + "a" * 64 def _script( *, root_id: int = 10, child_id: int = 20, element_id: int = 30, child_description: str = "child body", ) -> StructuredScriptArtifactV1: return StructuredScriptArtifactV1( direction_ref="script-build://artifact-versions/1", input_closure_digest=_DIGEST, paragraphs=( ScriptParagraphV1( paragraph_id=root_id, paragraph_index=1, level=1, parent_id=None, name="root", content_range={"start": 0, "end": 10}, description="root body", ), ScriptParagraphV1( paragraph_id=child_id, paragraph_index=2, level=2, parent_id=root_id, name="child", content_range={"start": 10, "end": 20}, description=child_description, ), ), elements=( ScriptElementV1( element_id=element_id, name="hook", dimension_primary="形式", dimension_secondary="opening", topic_support={"score": 1}, ), ), paragraph_element_links=( ScriptParagraphElementLinkV1( paragraph_id=child_id, element_id=element_id, ), ), source_artifact_refs=("script-build://artifact-versions/2",), goal_coverage=( GoalCoverage("goal-1", ("script-build://artifact-versions/2",)), ), evidence_refs=(), acceptance_notes=(), ) def test_legacy_canonical_digest_ignores_every_physical_identity() -> None: first = canonicalize_legacy_projection(_script(), direction="# D", summary="summary") second = canonicalize_legacy_projection( _script(root_id=101, child_id=202, element_id=303), direction="# D", summary="summary", ) assert first.content_payload() == second.content_payload() assert first.canonical_sha256 == second.canonical_sha256 def test_legacy_canonical_digest_changes_with_business_content() -> None: first = canonicalize_legacy_projection(_script(), direction="# D", summary="summary") changed_body = canonicalize_legacy_projection( _script(child_description="changed"), direction="# D", summary="summary" ) changed_direction = canonicalize_legacy_projection( _script(), direction="# other", summary="summary" ) assert first.canonical_sha256 != changed_body.canonical_sha256 assert first.canonical_sha256 != changed_direction.canonical_sha256 def test_legacy_canonical_rejects_parent_cycles_and_duplicate_natural_elements() -> None: cycle = replace( _script(), paragraphs=( ScriptParagraphV1(10, 1, 2, 20, "a", {}), ScriptParagraphV1(20, 2, 2, 10, "b", {}), ), ) with pytest.raises(LegacyCanonicalIdentityConflict, match="cycle"): canonicalize_legacy_projection(cycle) duplicate_elements = replace( _script(), elements=( ScriptElementV1(30, "hook", "形式", "opening"), ScriptElementV1(31, "hook", "形式", "opening"), ), paragraph_element_links=(), ) with pytest.raises(LegacyCanonicalIdentityConflict, match="duplicated"): canonicalize_legacy_projection(duplicate_elements) def test_root_delivery_manifest_is_strict_bounded_and_publishable() -> None: manifest = RootDeliveryManifestV1.from_payload( { "schema_version": "root-delivery-manifest/v1", "direction_ref": "script-build://artifact-versions/1", "candidate_portfolio_ref": "script-build://artifact-versions/2", "structured_script_ref": "script-build://artifact-versions/3", "input_closure_digest": _DIGEST, "legacy_projection_digest": _DIGEST, "build_summary": "ok", "blocking_defects": [], } ).with_digest() manifest.require_publishable() assert manifest.canonical_sha256.startswith("sha256:") with pytest.raises(ProtocolViolation, match="unknown fields"): RootDeliveryManifestV1.from_payload({**manifest.content_payload(), "unknown": True}) with pytest.raises(ProtocolViolation, match="blocking defects"): replace( manifest, blocking_defects=("not ready",), canonical_sha256="" ).require_publishable() with pytest.raises(ProtocolViolation, match="2,000"): replace(manifest, build_summary="x" * 2_001, canonical_sha256="") @pytest.mark.asyncio async def test_phase_three_model_execution_marker_is_single_use() -> None: class TraceStore: def __init__(self) -> None: self.events: list[dict[str, str]] = [] async def get_events(self, _trace_id: str, _since: int): return list(self.events) async def append_event(self, _trace_id: str, event: str, payload: dict[str, str]): self.events.append({"event": event, **payload}) return len(self.events) service = object.__new__(PhaseThreeContinuationService) trace_store = TraceStore() service.runner = SimpleNamespace(trace_store=trace_store) await service._claim_policy_execution("root", "snapshot-3") assert trace_store.events == [ { "event": PHASE_THREE_EXECUTION_STARTED, "policy_version": PHASE_THREE_POLICY, "input_snapshot_id": "snapshot-3", } ] with pytest.raises(MissionRecoveryRequired): await service._claim_policy_execution("root", "snapshot-3") @pytest.mark.asyncio async def test_phase_three_reentry_is_allowed_only_before_model_execution() -> None: class TraceStore: def __init__(self) -> None: self.events = [ { "event": "planner_policy_migrated", "policy_version": PHASE_THREE_POLICY, "input_snapshot_id": "snapshot-3", } ] async def get_events(self, _trace_id: str, _since: int): return list(self.events) async def append_event(self, _trace_id: str, event: str, payload: dict[str, str]): self.events.append({"event": event, **payload}) return len(self.events) service = object.__new__(PhaseThreeContinuationService) trace_store = TraceStore() service.runner = SimpleNamespace(trace_store=trace_store) assert await service._phase_three_reentry_ready("root") is True await service._claim_policy_execution("root", "snapshot-3") with pytest.raises(MissionRecoveryRequired): await service._phase_three_reentry_ready("root") @pytest.mark.asyncio async def test_completed_root_does_not_restore_a_failed_build() -> None: service = object.__new__(PhaseThreeContinuationService) service._verify_owner = AsyncMock() service._set_root_accepted_checkpoint = AsyncMock() service.bindings = SimpleNamespace( get_by_build=AsyncMock( return_value=SimpleNamespace(root_trace_id="root", input_snapshot_id=3) ) ) service.legacy_state = SimpleNamespace(get_status=AsyncMock(return_value=BuildStatus.FAILED)) service.input_snapshots = SimpleNamespace(get=AsyncMock(return_value=SimpleNamespace())) root = SimpleNamespace(status=TaskStatus.COMPLETED) service.coordinator = SimpleNamespace( task_store=SimpleNamespace( load=AsyncMock( return_value=SimpleNamespace(root_task_id="root-task", tasks={"root-task": root}) ) ) ) owner_token = SimpleNamespace(root_trace_id="root") with pytest.raises(PhaseThreeBoundaryNotReady, match="failed"): await service.run(7, owner_token=owner_token) service._set_root_accepted_checkpoint.assert_not_awaited()