from __future__ import annotations from types import SimpleNamespace import pytest from agent.orchestration import ArtifactRef, ResourceClaim, TaskStatus from script_build_host.application.direction_evidence_gate import DirectionEvidenceGate from script_build_host.application.phase_two_planning import ( _claim_serialized_waves, ) from script_build_host.domain.errors import RequiredEvidenceUnsatisfied from script_build_host.domain.evidence_requirements import ( DirectionEvidenceRequirementFactory, EvidenceClosureEvaluator, EvidenceGate, EvidenceItem, EvidenceProvider, ) from script_build_host.domain.task_contract_compiler import ( PlanningContainer, PlanningDecision, PlanningSnapshot, PlanningTarget, TaskContractCompiler, TaskContractMutation, ) from script_build_host.domain.task_contracts import ( AcceptedDecisionRef, PlannerTaskIntentV2, ScriptCriterion, ScriptTaskBudget, ScriptTaskKind, TaskContractError, ) from script_build_host.infrastructure.task_contract_store import ( FileScriptTaskContractStore, ) def _decision( identity: str, kind: ScriptTaskKind, branch: str ) -> PlanningDecision: return PlanningDecision( AcceptedDecisionRef( identity, ArtifactRef( uri="script-build://artifact-versions/1", kind="artifact", version="1", digest="sha256:" + "1" * 64, ), f"script-build://scopes/{branch}/{kind.value}", kind, ), f"producer-{identity}", f"script-build://branches/{branch}", ("goal-a",), ) def test_same_workspace_lane_is_serialized_while_distinct_lanes_share_a_wave() -> None: lane_a = (ResourceClaim("script-build://resources/workspace/a"),) lane_b = (ResourceClaim("script-build://resources/workspace/b"),) waves = _claim_serialized_waves( ( ("paragraph-a", "worker", lane_a), ("element-a", "worker", lane_a), ("paragraph-b", "worker", lane_b), ) ) assert tuple(item[0] for item in waves[0]) == ( "paragraph-a", "paragraph-b", ) assert tuple(item[0] for item in waves[1]) == ("element-a",) def test_host_owned_root_delivery_uses_the_v2_catalog_contract() -> None: direction = _decision( "direction-1", ScriptTaskKind.DIRECTION, "phase-one" ).ref portfolio = _decision( "portfolio-1", ScriptTaskKind.CANDIDATE_PORTFOLIO, "portfolio" ).ref contract = TaskContractCompiler().compile_host_contract( task_id="root-task", parent_task_id="root-task", task_kind=ScriptTaskKind.ROOT_DELIVERY, objective="publish", scope_ref=direction.scope_ref, branch_ref="script-build://branches/root-delivery", input_decision_refs=(direction, portfolio), criteria=(ScriptCriterion("published", "publish", True),), budget=ScriptTaskBudget(), goal_ids=("goal-a",), ) assert contract.to_payload()["schema_version"] == "script-task-contract/v2" assert contract.execution_profile.worker_preset == "script_root_worker" assert contract.catalog_digest == TaskContractCompiler().capabilities.digest @pytest.mark.asyncio async def test_file_contract_reader_preserves_v2_frozen_execution_profile( tmp_path, ) -> None: direction = _decision( "direction-1", ScriptTaskKind.DIRECTION, "phase-one" ).ref portfolio = _decision( "portfolio-1", ScriptTaskKind.CANDIDATE_PORTFOLIO, "portfolio" ).ref contract = TaskContractCompiler().compile_host_contract( task_id="root-task", parent_task_id="root-task", task_kind=ScriptTaskKind.ROOT_DELIVERY, objective="publish", scope_ref=direction.scope_ref, branch_ref="script-build://branches/root-delivery", input_decision_refs=(direction, portfolio), criteria=(ScriptCriterion("published", "publish", True),), budget=ScriptTaskBudget(), goal_ids=("goal-a",), ) store = FileScriptTaskContractStore(tmp_path) frozen = await store.freeze("root", contract) replay = await store.read("root", frozen.uri) assert replay.contract == contract def test_intent_order_does_not_change_compiled_contracts() -> None: snapshot = PlanningSnapshot( root_trace_id="root", phase=2, containers=( PlanningContainer( "root-task", None, None, None, ), ), decisions=(), active_direction_ref=_decision( "direction-1", ScriptTaskKind.DIRECTION, "phase-one" ).ref, direction_goal_ids=("goal-a",), ) intents = ( PlannerTaskIntentV2( ScriptTaskKind.STRUCTURE, "approach a", target_goal_ids=("goal-a",) ), PlannerTaskIntentV2( ScriptTaskKind.STRUCTURE, "approach b", target_goal_ids=("goal-a",) ), ) compiler = TaskContractCompiler() first = compiler.compile_batch( intents, snapshot, command_id="command", budget=ScriptTaskBudget() ) second = compiler.compile_batch( tuple(reversed(intents)), snapshot, command_id="command", budget=ScriptTaskBudget(), ) assert [item.contract.to_payload() for item in first] == [ item.contract.to_payload() for item in second ] assert [item.contract.task_kind for item in first].count( ScriptTaskKind.CANDIDATE_PORTFOLIO ) == 1 assert [item.contract.task_kind for item in first].count( ScriptTaskKind.COMPOSE ) == 2 def test_phase_two_scaffolds_then_closes_without_planner_contract_fields() -> None: compiler = TaskContractCompiler() direction = _decision("direction-1", ScriptTaskKind.DIRECTION, "phase-one").ref initial = PlanningSnapshot( "root", 2, (PlanningContainer("root-task", None, None, None),), (), active_direction_ref=direction, direction_goal_ids=("goal-a",), ) opened = compiler.compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.STRUCTURE, "open one creative approach", target_goal_ids=("goal-a",), ), ), initial, command_id="open", budget=ScriptTaskBudget(), ) portfolio = next( item for item in opened if item.contract.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO ) compose = next( item for item in opened if item.contract.task_kind is ScriptTaskKind.COMPOSE ) structure = next( item for item in opened if item.contract.task_kind is ScriptTaskKind.STRUCTURE ) assert not portfolio.contract.execution_ready assert not compose.contract.execution_ready assert structure.parent_task_id == compose.task_id structure_decision = PlanningDecision( AcceptedDecisionRef( "structure-1", ArtifactRef( uri="script-build://artifact-versions/10", kind="artifact", version="10", digest="sha256:" + "a" * 64, ), structure.contract.scope_ref, ScriptTaskKind.STRUCTURE, ), structure.task_id, structure.contract.branch_ref, structure.contract.goal_ids, ) containers = ( PlanningContainer("root-task", None, None, None), PlanningContainer( portfolio.task_id, portfolio.contract.task_kind, portfolio.contract.scope_ref, portfolio.contract.branch_ref, portfolio.contract.goal_ids, portfolio.parent_task_id, True, ), PlanningContainer( compose.task_id, compose.contract.task_kind, compose.contract.scope_ref, compose.contract.branch_ref, compose.contract.goal_ids, compose.parent_task_id, True, ), ) paragraph = compiler.compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.PARAGRAPH, "write paragraph", decision_ids=("structure-1",), target_goal_ids=("goal-a",), ), ), PlanningSnapshot( "root", 2, containers, (structure_decision,), ( PlanningTarget( "script-build://targets/paragraph-1", "structure-1", ScriptTaskKind.STRUCTURE, structure.contract.branch_ref, ("goal-a",), ), ), direction, ("goal-a",), ), command_id="paragraph", budget=ScriptTaskBudget(), )[0] paragraph_decision = PlanningDecision( AcceptedDecisionRef( "paragraph-1", ArtifactRef( uri="script-build://artifact-versions/11", kind="artifact", version="11", digest="sha256:" + "b" * 64, ), paragraph.contract.scope_ref, ScriptTaskKind.PARAGRAPH, ), paragraph.task_id, paragraph.contract.branch_ref, paragraph.contract.goal_ids, ) element = compiler.compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.ELEMENT_SET, "add elements", decision_ids=("paragraph-1",), target_goal_ids=("goal-a",), ), ), PlanningSnapshot( "root", 2, containers, (paragraph_decision,), ( PlanningTarget( "script-build://targets/paragraph-1", "paragraph-1", ScriptTaskKind.PARAGRAPH, paragraph.contract.branch_ref, ("goal-a",), ), ), direction, ("goal-a",), ), command_id="element", budget=ScriptTaskBudget(), )[0] element_decision = PlanningDecision( AcceptedDecisionRef( "element-1", ArtifactRef( uri="script-build://artifact-versions/12", kind="artifact", version="12", digest="sha256:" + "c" * 64, ), element.contract.scope_ref, ScriptTaskKind.ELEMENT_SET, ), element.task_id, element.contract.branch_ref, element.contract.goal_ids, ) closed_compose = compiler.compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.COMPOSE, "compose accepted branch", decision_ids=("structure-1", "paragraph-1", "element-1"), ), ), PlanningSnapshot( "root", 2, containers, (structure_decision, paragraph_decision, element_decision), active_direction_ref=direction, direction_goal_ids=("goal-a",), ), command_id="compose", budget=ScriptTaskBudget(), )[0] assert closed_compose.task_id == compose.task_id assert closed_compose.mutation is TaskContractMutation.REVISE assert closed_compose.contract.execution_ready compose_decision = PlanningDecision( AcceptedDecisionRef( "compose-1", ArtifactRef( uri="script-build://artifact-versions/13", kind="artifact", version="13", digest="sha256:" + "d" * 64, ), closed_compose.contract.scope_ref, ScriptTaskKind.COMPOSE, ), compose.task_id, compose.contract.branch_ref, closed_compose.contract.goal_ids, ) closed_portfolio = compiler.compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.CANDIDATE_PORTFOLIO, "adopt accepted script", decision_ids=("compose-1",), ), ), PlanningSnapshot( "root", 2, containers, (compose_decision,), active_direction_ref=direction, direction_goal_ids=("goal-a",), ), command_id="portfolio", budget=ScriptTaskBudget(), )[0] assert closed_portfolio.task_id == portfolio.task_id assert closed_portfolio.mutation is TaskContractMutation.REVISE assert closed_portfolio.contract.execution_ready def test_compose_scaffold_cannot_close_before_child_frontier_is_terminal() -> None: decisions = tuple( _decision(f"{kind.value}-1", kind, "branch-a") for kind in ( ScriptTaskKind.STRUCTURE, ScriptTaskKind.PARAGRAPH, ScriptTaskKind.ELEMENT_SET, ) ) snapshot = PlanningSnapshot( "root", 2, ( PlanningContainer("root-task", None, None, None), PlanningContainer( "compose-parent", ScriptTaskKind.COMPOSE, "script-build://scopes/compose", "script-build://branches/branch-a", ("goal-a",), "portfolio-parent", False, ), ), decisions, active_direction_ref=_decision( "direction-1", ScriptTaskKind.DIRECTION, "phase-one" ).ref, direction_goal_ids=("goal-a",), ) with pytest.raises(TaskContractError, match="nonterminal child work"): TaskContractCompiler().compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.COMPOSE, "close too early", decision_ids=tuple( item.ref.decision_id for item in decisions ), ), ), snapshot, command_id="premature-compose", budget=ScriptTaskBudget(), ) def test_paragraph_target_is_host_semantic_target_not_objective_text() -> None: structure = _decision("structure-1", ScriptTaskKind.STRUCTURE, "branch-a") snapshot = PlanningSnapshot( root_trace_id="root", phase=2, containers=( PlanningContainer( "compose-parent", ScriptTaskKind.COMPOSE, None, "script-build://branches/branch-a", ("goal-a",), ), ), decisions=(structure,), targets=( PlanningTarget( "script-build://targets/paragraph-a", structure.ref.decision_id, ScriptTaskKind.STRUCTURE, "script-build://branches/branch-a", ("goal-a",), ), ), active_direction_ref=_decision( "direction-1", ScriptTaskKind.DIRECTION, "phase-one" ).ref, direction_goal_ids=("goal-a",), ) compiled = TaskContractCompiler().compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.PARAGRAPH, "wording can change freely", decision_ids=("structure-1",), target_goal_ids=("goal-a",), ), ), snapshot, command_id="command", budget=ScriptTaskBudget(), )[0] assert compiled.contract.workspace_seed is not None assert ( compiled.contract.workspace_seed.semantic_target_ref == "script-build://targets/paragraph-a" ) def test_ambiguous_semantic_target_fails_before_task_creation() -> None: structure = _decision("structure-1", ScriptTaskKind.STRUCTURE, "branch-a") targets = tuple( PlanningTarget( f"script-build://targets/{suffix}", structure.ref.decision_id, ScriptTaskKind.STRUCTURE, "script-build://branches/branch-a", ("goal-a",), ) for suffix in ("a", "b") ) snapshot = PlanningSnapshot( "root", 2, ( PlanningContainer( "compose-parent", ScriptTaskKind.COMPOSE, None, "script-build://branches/branch-a", ("goal-a",), ), ), (structure,), targets, _decision("direction-1", ScriptTaskKind.DIRECTION, "phase-one").ref, ("goal-a",), ) with pytest.raises(TaskContractError, match="semantic targets"): TaskContractCompiler().compile_batch( ( PlannerTaskIntentV2( ScriptTaskKind.PARAGRAPH, "paragraph", decision_ids=("structure-1",), target_goal_ids=("goal-a",), ), ), snapshot, command_id="command", budget=ScriptTaskBudget(), ) def test_pattern_evidence_cannot_satisfy_missing_persona() -> None: manifest = DirectionEvidenceRequirementFactory().compile( topic_input_ref="input:topic", requested_account_name="account", persona_points=(), persona_input_ref=None, section_patterns=(), section_pattern_input_ref=None, ) closure = EvidenceClosureEvaluator().evaluate( manifest, ( EvidenceItem( "direction.section_pattern", "pattern:1", EvidenceProvider.PATTERN, ), ), gate=EvidenceGate.DIRECTION_ACCEPT, ) assert "direction.persona" in closure.unsatisfied_hard_requirement_ids @pytest.mark.asyncio async def test_direction_gate_rejects_accept_from_stale_child_spec() -> None: manifest = DirectionEvidenceRequirementFactory().compile( topic_input_ref="input:topic", requested_account_name="account", persona_points=(), persona_input_ref=None, section_patterns=({"pattern": "hook"},), section_pattern_input_ref="input:patterns", ) direction = _decision( "direction-1", ScriptTaskKind.DIRECTION, "phase-one" ).ref portfolio = _decision( "portfolio-1", ScriptTaskKind.CANDIDATE_PORTFOLIO, "portfolio" ).ref contract = TaskContractCompiler().compile_host_contract( task_id="root-task", parent_task_id="root-task", task_kind=ScriptTaskKind.ROOT_DELIVERY, objective="publish", scope_ref=direction.scope_ref, branch_ref="script-build://branches/root-delivery", input_decision_refs=(direction, portfolio), criteria=(ScriptCriterion("published", "publish", True),), budget=ScriptTaskBudget(), goal_ids=("goal-a",), evidence_requirements=manifest, ) child = SimpleNamespace( task_id="evidence-child", status=TaskStatus.COMPLETED, current_spec_version=2, current_spec=SimpleNamespace( context_refs=("script-build://task-kinds/evidence-retrieval",) ), decision_ids=["accepted-old-spec"], ) ledger = SimpleNamespace( tasks={child.task_id: child}, decisions={ "accepted-old-spec": SimpleNamespace( action=SimpleNamespace(value="accept"), attempt_id="attempt-v1", ) }, attempts={ "attempt-v1": SimpleNamespace( attempt_id="attempt-v1", spec_version=1, submission=SimpleNamespace(evidence_refs=("evidence://persona",)), ) }, ) class UnreadArtifacts: async def read_by_ref(self, *_: object, **__: object) -> object: raise AssertionError("stale accepted evidence must not be read") with pytest.raises(RequiredEvidenceUnsatisfied): await DirectionEvidenceGate(UnreadArtifacts()).require( script_build_id=1, ledger=ledger, direction_task=SimpleNamespace(child_task_ids=(child.task_id,)), contract=contract, gate=EvidenceGate.DIRECTION_ACCEPT, )