from __future__ import annotations import asyncio from dataclasses import replace from datetime import UTC, datetime, timedelta from hashlib import sha256 from pathlib import Path from types import SimpleNamespace from typing import Any, cast import pytest from agent import AgentRunner, FileSystemArtifactStore, FileSystemTaskStore, FileSystemTraceStore from agent.orchestration import ( AgentRole, ArtifactRef, AttemptSubmission, CriterionResult, DecisionAction, OperationStatus, OrchestrationConfig, TaskCoordinator, TaskStatus, ValidationVerdict, ) from agent.orchestration.protocols import ValidatorRunResult, WorkerRunResult from script_build_host.application.mission_factory import ScriptMissionFactory from script_build_host.application.mission_service import ScriptMissionService from script_build_host.application.phase_two_planning import ( PhaseTwoPlanningService, _execution_seconds, ) from script_build_host.domain.errors import ProtocolViolation from script_build_host.domain.records import BuildStatus, MissionBinding, Principal from script_build_host.domain.task_contracts import ( PhaseTwoLimits, ScriptTaskContractV1, TaskContractError, ) from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore def test_execution_budget_counts_stage_intervals_not_planner_idle_time() -> None: now = datetime.now(UTC) stages = ( SimpleNamespace( started_at=(now - timedelta(hours=2)).isoformat(), completed_at=(now - timedelta(hours=2) + timedelta(seconds=12)).isoformat(), ), SimpleNamespace( started_at=(now - timedelta(minutes=1)).isoformat(), completed_at=(now - timedelta(minutes=1) + timedelta(seconds=8)).isoformat(), ), ) assert _execution_seconds(stages) == pytest.approx(20.0) def _contract( kind: str, *, scope: str = "script-build://scopes/full", execution_ready: bool = True, ) -> dict[str, object]: output = { "direction": "script-direction/v1", "candidate-portfolio": "candidate-portfolio/v1", "compose": "structured-script/v1", "paragraph": "paragraph-artifact/v1", "element-set": "element-set-artifact/v1", "structure": "structure-artifact/v1", "compare": "comparison-artifact/v1", "decode-retrieval": "evidence-record/v1", }[kind] payload: dict[str, object] = { "schema_version": "script-task-contract/v1", "task_kind": kind, "scope_ref": scope, "intent_class": ( "portfolio" if kind == "candidate-portfolio" else "compose" if kind == "compose" else "explore" ), "objective": f"produce one bounded {kind} increment", "input_decision_refs": [], "base_artifact_ref": None, "write_scope": [scope], "gap_ref": None, "output_schema": output, "criteria": [ { "criterion_id": "closed", "description": "the bounded increment is independently verifiable", "hard": True, } ], "budget": { "max_attempts": 4, "max_tokens": 32000, "max_seconds": 900, "max_external_queries": 40, "max_no_improvement": 3, }, "supersedes_decision_ids": [], "candidate_closure_decision_refs": [], "adopted_decision_ids": [], "held_or_rejected_decision_ids": [], "compose_order": [], "comparison_decision_refs": [], } if execution_ready and kind in {"compose", "candidate-portfolio"}: ref = { "decision_id": "decision-1", "artifact_ref": { "uri": "script-build://artifact-versions/1", "kind": "structured_script" if kind == "candidate-portfolio" else "paragraph", "version": "1", "digest": "sha256:" + "a" * 64, }, "scope_ref": scope, "expected_task_kind": "compose" if kind == "candidate-portfolio" else "paragraph", } payload["candidate_closure_decision_refs"] = [ref] payload["adopted_decision_ids"] = ["decision-1"] payload["compose_order"] = ["decision-1"] return payload class _Bindings: def __init__(self, root: str) -> None: self.binding = MissionBinding( binding_id=1, script_build_id=7, root_trace_id=root, input_snapshot_id=11, active_direction_artifact_version_id=None, accepted_root_artifact_version_id=None, engine_version="test", schema_version="v1", created_at=datetime.now(UTC), updated_at=datetime.now(UTC), ) async def get_by_root(self, root: str) -> MissionBinding: assert root == self.binding.root_trace_id return self.binding async def get_by_build(self, script_build_id: int) -> MissionBinding: assert script_build_id == self.binding.script_build_id return self.binding class _PlaceholderThenReplacementExecutor: """Exercise Coordinator state transitions without bypassing durable operations.""" def __init__(self, coordinator: TaskCoordinator) -> None: self.coordinator = coordinator self.artifact_version = 100 async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult: refs = cast(dict[str, Any], context["task_spec"])["context_refs"] kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item) artifact_kind = { "compose": "structured_script", "structure": "structure", "paragraph": "paragraph", "element-set": "element_set", }[kind] digest = "sha256:" + sha256(context["attempt_id"].encode()).hexdigest() version = str(self.artifact_version) self.artifact_version += 1 ref = ArtifactRef( f"script-build://artifact-versions/{version}", artifact_kind, version, digest, ) await self.coordinator.submit_attempt( { "role": AgentRole.WORKER.value, "root_trace_id": context["root_trace_id"], "task_id": context["task_id"], "attempt_id": context["attempt_id"], "spec_version": context["spec_version"], "trace_id": context["worker_trace_id"], "tool_call_id": f"submit:{context['attempt_id']}", "operation_id": context["operation_id"], "execution_epoch": context["execution_epoch"], }, AttemptSubmission( summary=f"kind={artifact_kind};status=frozen", artifact_refs=[ref], ), ) return WorkerRunResult(context["worker_trace_id"], "completed") async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult: refs = cast(dict[str, Any], context["task_spec"])["context_refs"] kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item) placeholder = kind == "compose" and context["spec_version"] == 1 verdict = ValidationVerdict.FAILED if placeholder else ValidationVerdict.PASSED criterion = cast(dict[str, Any], context["task_spec"])["acceptance_criteria"][0] await self.coordinator.submit_validation( { "role": AgentRole.VALIDATOR.value, "root_trace_id": context["root_trace_id"], "task_id": context["task_id"], "attempt_id": context["attempt_id"], "validation_id": context["validation_id"], "snapshot_id": context["snapshot_id"], "trace_id": context["validator_trace_id"], "tool_call_id": f"validate:{context['validation_id']}", "operation_id": context["operation_id"], "execution_epoch": context["execution_epoch"], }, verdict, [ CriterionResult( criterion_id=criterion["criterion_id"], verdict=verdict, reason=( "REALIZATION_PLACEHOLDER at artifact.paragraphs[0].description" if placeholder else "the replacement is concretely rendered" ), ) ], ( "hard_defects=1;defect_codes=REALIZATION_PLACEHOLDER" if placeholder else "hard_defects=0" ), (), ("artifact.paragraphs[0].description",) if placeholder else (), ("REALIZATION_PLACEHOLDER",) if placeholder else (), "split" if placeholder else "accept", ) return ValidatorRunResult(context["validator_trace_id"], "completed") async def stop(self, trace_id: str) -> bool: del trace_id return True async def _service(tmp_path: Path) -> tuple[PhaseTwoPlanningService, TaskCoordinator, str]: root = "root-contracts" coordinator = TaskCoordinator( FileSystemTaskStore(str(tmp_path / "ledger")), FileSystemArtifactStore(str(tmp_path / "snapshots")), FileSystemTraceStore(str(tmp_path / "traces")), ) await coordinator.ensure_ledger( root, { "objective": "build a script", "acceptance_criteria": [ {"criterion_id": "ready", "description": "ready", "hard": True} ], "context_refs": ["script-build://inputs/11"], }, ) service = PhaseTwoPlanningService( coordinator=coordinator, bindings=_Bindings(root), contracts=FileScriptTaskContractStore(tmp_path / "data"), ) return service, coordinator, root @pytest.mark.asyncio async def test_dispatch_explains_that_blocked_children_do_not_close_parent( tmp_path: Path, ) -> None: service, coordinator, root = await _service(tmp_path) direction = await service.plan_script_tasks( contract_payloads=[_contract("direction")], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "direction"}, ) direction_id = direction["task_ids"][0] retrieval = await service.plan_script_tasks( contract_payloads=[_contract("decode-retrieval")], parent_task_id=direction_id, context={"root_trace_id": root, "tool_call_id": "retrieval"}, ) child_id = retrieval["task_ids"][0] await coordinator.decide_task( root, child_id, None, DecisionAction.BLOCK, {"reason": "unusable evidence"}, "block-child", ) with pytest.raises(TaskContractError, match="BLOCKED is resumable"): await service.dispatch_script_tasks( task_ids=[direction_id], context={"root_trace_id": root, "tool_call_id": "dispatch-parent"}, ) @pytest.mark.asyncio async def test_unique_portfolio_container_cannot_be_cancelled(tmp_path: Path) -> None: service, _coordinator, root = await _service(tmp_path) portfolio = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "portfolio"}, ) with pytest.raises(TaskContractError, match="cannot be cancelled"): await service.decide_script_task( task_id=portfolio["task_ids"][0], action="cancel", reason="wrong structural recovery", validation_id=None, replacement_contract=None, child_contracts=(), context={"root_trace_id": root, "tool_call_id": "cancel-portfolio"}, ) @pytest.mark.asyncio async def test_accepted_input_scope_is_rejected_before_worker_dispatch(tmp_path: Path) -> None: service, _coordinator, root = await _service(tmp_path) structure_scope = "script-build://direction/main/compose/candidate-1/structure/main" structure = ScriptTaskContractV1.from_payload(_contract("structure", scope=structure_scope)) paragraph_payload = _contract( "paragraph", scope="script-build://direction/main/compose/candidate-1/paragraph/p1", ) accepted_ref = { "decision_id": "accepted-structure", "artifact_ref": { "uri": "script-build://artifact-versions/1", "kind": "structure", "version": "1", "digest": "sha256:" + "a" * 64, }, "scope_ref": structure_scope, "expected_task_kind": "structure", } paragraph_payload["input_decision_refs"] = [accepted_ref] paragraph_payload["base_artifact_ref"] = accepted_ref["artifact_ref"] paragraph = ScriptTaskContractV1.from_payload(paragraph_payload) producer_task = SimpleNamespace(task_id="structure-task") ledger = SimpleNamespace( decisions={ "accepted-structure": SimpleNamespace( action=DecisionAction.ACCEPT, attempt_id="structure-attempt", task_id=producer_task.task_id, ) }, tasks={producer_task.task_id: producer_task}, ) async def frozen_contract(_root_trace_id: str, task: Any) -> Any: assert task is producer_task return SimpleNamespace(contract=structure) service.contract_for_task = frozen_contract # type: ignore[method-assign] with pytest.raises(TaskContractError, match="paragraph scope must equal or nest within"): await service._guard_accepted_input_scopes(root, ledger, (paragraph,)) class _BlockingExecutor: def __init__(self) -> None: self.started = asyncio.Event() self.stopped_trace_ids: list[str] = [] async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult: self.started.set() await asyncio.Event().wait() raise AssertionError(f"worker unexpectedly resumed: {context['attempt_id']}") async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult: raise AssertionError(f"validator must not start: {context['validation_id']}") async def stop(self, trace_id: str) -> bool: self.stopped_trace_ids.append(trace_id) return True class _StopState: def __init__(self) -> None: self.status = BuildStatus.RUNNING self.error_summary: str | None = None async def get_status(self, script_build_id: int) -> BuildStatus: assert script_build_id == 7 return self.status async def set_status( self, script_build_id: int, status: BuildStatus, *, error_summary: str | None = None, ) -> None: assert script_build_id == 7 self.status = status self.error_summary = error_summary @pytest.mark.asyncio async def test_real_operation_stop_converges_and_forbids_new_dispatch(tmp_path: Path) -> None: root = "root-stop-operation" trace_store = FileSystemTraceStore(str(tmp_path / "stop-traces")) coordinator = TaskCoordinator( FileSystemTaskStore(str(tmp_path / "stop-ledger")), FileSystemArtifactStore(str(tmp_path / "stop-artifacts")), trace_store, config=OrchestrationConfig(stop_grace_seconds=0.01), ) await coordinator.ensure_ledger( root, { "objective": "build a script", "acceptance_criteria": [ {"criterion_id": "ready", "description": "ready", "hard": True} ], "context_refs": ["script-build://inputs/11"], }, ) bindings = _Bindings(root) planning = PhaseTwoPlanningService( coordinator=coordinator, bindings=bindings, contracts=FileScriptTaskContractStore(tmp_path / "stop-contracts"), ) portfolio = await planning.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "stop:portfolio"}, ) compose = await planning.plan_script_tasks( contract_payloads=[_contract("compose", execution_ready=False)], parent_task_id=portfolio["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "stop:compose"}, ) paragraph = await planning.plan_script_tasks( contract_payloads=[_contract("paragraph")], parent_task_id=compose["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "stop:paragraph"}, ) executor = _BlockingExecutor() coordinator.set_executor(executor) async def unused_llm(**_kwargs: Any) -> Any: raise AssertionError("Planner LLM must not run during operation stop") runner = AgentRunner(trace_store=trace_store, llm_call=unused_llm) state = _StopState() class Authorizer: async def require_access(self, principal: Principal, script_build_id: int) -> None: assert principal.subject == "owner" and script_build_id == 7 mission = ScriptMissionService( runner=runner, coordinator=coordinator, factory=ScriptMissionFactory(), input_snapshots=SimpleNamespace(), bindings=bindings, legacy_state=state, authorizer=Authorizer(), direction_reconciler=SimpleNamespace(), stop_timeout_seconds=1, ) planning.dispatch_guard = mission dispatch = asyncio.create_task( planning.dispatch_script_tasks( task_ids=paragraph["task_ids"], context={"root_trace_id": root, "tool_call_id": "stop:dispatch"}, ) ) await asyncio.wait_for(executor.started.wait(), timeout=1) stopped = await mission.stop(7, Principal("owner")) dispatch_result = await asyncio.gather(dispatch, return_exceptions=True) assert stopped.status is BuildStatus.STOPPED assert len(stopped.stopped_operation_ids) == 1 assert executor.stopped_trace_ids assert not isinstance(dispatch_result[0], asyncio.CancelledError) ledger = await coordinator.task_store.load(root) operation = ledger.operations[stopped.stopped_operation_ids[0]] assert operation.status is OperationStatus.STOPPED assert ledger.tasks[paragraph["task_ids"][0]].status is TaskStatus.NEEDS_REPLAN assert (await mission.stop(7, Principal("owner"))).status is BuildStatus.STOPPED with pytest.raises(ProtocolViolation, match="forbidden after stop intent"): await planning.dispatch_script_tasks( task_ids=paragraph["task_ids"], context={"root_trace_id": root, "tool_call_id": "stop:late-dispatch"}, ) @pytest.mark.asyncio async def test_contract_store_is_content_addressed_and_detects_tampering(tmp_path: Path) -> None: store = FileScriptTaskContractStore(tmp_path) from script_build_host.domain.task_contracts import ScriptTaskContractV1 contract = ScriptTaskContractV1.from_payload(_contract("paragraph")) first = await store.freeze("root-a", contract) second = await store.freeze("root-a", contract) assert first == second path = tmp_path / "script-task-contracts" / "root-a" / "sha256" / first.uri.rsplit("/", 1)[-1] path.write_bytes(b"{}") with pytest.raises(TaskContractError, match="TASK_CONTRACT_DIGEST_MISMATCH"): await store.read("root-a", first.uri) assert not await store.verify("root-a", first.uri, first.digest) @pytest.mark.asyncio async def test_orphan_contract_reclamation_uses_only_ledger_references(tmp_path: Path) -> None: service, _, root = await _service(tmp_path) referenced = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "portfolio"}, ) orphan = await service.contracts.freeze( root, ScriptTaskContractV1.from_payload(_contract("paragraph")) ) removed = await service.reclaim_orphan_contracts(root_trace_id=root) assert removed == (orphan.uri,) assert await service.contracts.verify( root, referenced["contracts"][0]["contract_ref"], referenced["contracts"][0]["contract_digest"], ) assert not await service.contracts.verify(root, orphan.uri, orphan.digest) @pytest.mark.asyncio async def test_failed_operation_discards_workspace_but_recovery_required_preserves_it( tmp_path: Path, ) -> None: service, _, root = await _service(tmp_path) class Lifecycle: def __init__(self) -> None: self.calls: list[dict[str, str]] = [] async def discard_attempt(self, **values: str) -> None: self.calls.append(values) lifecycle = Lifecycle() service.workspace_lifecycle = lifecycle attempts = { "failed-attempt": SimpleNamespace(attempt_id="failed-attempt", task_id="paragraph-task"), "recovery-attempt": SimpleNamespace( attempt_id="recovery-attempt", task_id="paragraph-task" ), } ledger = SimpleNamespace( attempts=attempts, operations={ "failed": SimpleNamespace( status=OperationStatus.FAILED, error="worker failed before submit", attempt_ids=["failed-attempt"], ), "recovery": SimpleNamespace( status=OperationStatus.FAILED, error="MISSION_RECOVERY_REQUIRED", attempt_ids=["recovery-attempt"], ), }, ) await service._reconcile_terminal_workspaces(root, ledger) assert lifecycle.calls == [ { "root_trace_id": root, "task_id": "paragraph-task", "attempt_id": "failed-attempt", "reason": "worker failed before submit", } ] def test_comparison_refs_are_kind_specific_and_limits_cannot_be_raised() -> None: candidate_refs = [] for index, kind in ((1, "paragraph"), (2, "structure")): candidate_refs.append( { "decision_id": f"decision-{index}", "artifact_ref": { "uri": f"script-build://artifact-versions/{index}", "kind": kind, "version": str(index), "digest": "sha256:" + str(index) * 64, }, "scope_ref": "script-build://scopes/full", "expected_task_kind": kind, } ) compare = _contract("compare") compare["comparison_decision_refs"] = candidate_refs assert ScriptTaskContractV1.from_payload(compare).task_kind.value == "compare" comparison_ref = { "decision_id": "decision-compare", "artifact_ref": { "uri": "script-build://artifact-versions/3", "kind": "comparison", "version": "3", "digest": "sha256:" + "3" * 64, }, "scope_ref": "script-build://scopes/full", "expected_task_kind": "compare", } compose = _contract("compose") compose["comparison_decision_refs"] = [comparison_ref] ScriptTaskContractV1.from_payload(compose) portfolio = _contract("candidate-portfolio") portfolio["comparison_decision_refs"] = [comparison_ref] with pytest.raises(TaskContractError, match="comparison_decision_refs"): ScriptTaskContractV1.from_payload(portfolio) with pytest.raises(TaskContractError, match="TASK_BUDGET_EXCEEDED"): PhaseTwoLimits(max_tasks=65) @pytest.mark.asyncio async def test_planning_freezes_contract_and_creates_only_one_portfolio(tmp_path: Path) -> None: service, coordinator, root = await _service(tmp_path) result = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "plan-portfolio"}, ) task_id = result["task_ids"][0] ledger = await coordinator.task_store.load(root) task = ledger.tasks[task_id] assert task.current_spec.context_refs[0].endswith("candidate-portfolio") assert task.current_spec.context_refs[1].startswith("script-build://task-contracts/sha256/") assert task.current_spec.context_refs[2] == "script-build://inputs/11" frozen = await service.contract_for_task(root, task) assert frozen.contract.scope_ref == "script-build://scopes/full" with pytest.raises(TaskContractError, match="one Portfolio"): await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "plan-portfolio-2"}, ) @pytest.mark.asyncio async def test_invalid_contract_batch_creates_zero_tasks(tmp_path: Path) -> None: service, coordinator, root = await _service(tmp_path) invalid = _contract("paragraph") invalid["output_schema"] = "structured-script/v1" with pytest.raises(TaskContractError, match="output_schema"): await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio"), invalid], parent_task_id=None, context={"root_trace_id": root}, ) ledger = await coordinator.task_store.load(root) assert list(ledger.tasks) == [ledger.root_task_id] @pytest.mark.asyncio async def test_planning_rejects_unresolved_supersession_before_freeze(tmp_path: Path) -> None: service, coordinator, root = await _service(tmp_path) portfolio = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "portfolio"}, ) compose = await service.plan_script_tasks( contract_payloads=[_contract("compose", execution_ready=False)], parent_task_id=portfolio["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "compose"}, ) replacement = _contract("paragraph") replacement["supersedes_decision_ids"] = ["missing-accept"] with pytest.raises(TaskContractError, match="import every superseded"): await service.plan_script_tasks( contract_payloads=[replacement], parent_task_id=compose["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "invalid-replacement"}, ) ledger = await coordinator.task_store.load(root) assert set(ledger.tasks) == { ledger.root_task_id, portfolio["task_ids"][0], compose["task_ids"][0], } @pytest.mark.asyncio async def test_open_compose_container_cannot_dispatch(tmp_path: Path) -> None: service, coordinator, root = await _service(tmp_path) portfolio = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "p"}, ) compose = await service.plan_script_tasks( contract_payloads=[_contract("compose", execution_ready=False)], parent_task_id=portfolio["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "c"}, ) with pytest.raises(TaskContractError, match="first accept child candidates"): await service.dispatch_script_tasks( task_ids=compose["task_ids"], context={"root_trace_id": root} ) ledger = await coordinator.task_store.load(root) assert not ledger.operations @pytest.mark.asyncio async def test_compose_placeholder_split_replacement_then_revised_v2_passes( tmp_path: Path, ) -> None: service, coordinator, root = await _service(tmp_path) coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator)) portfolio = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "portfolio"}, ) compose = await service.plan_script_tasks( contract_payloads=[_contract("compose")], parent_task_id=portfolio["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "compose-v1"}, ) compose_id = compose["task_ids"][0] first_cycle = ( await service.dispatch_script_tasks( task_ids=(compose_id,), context={"root_trace_id": root, "tool_call_id": "dispatch-compose-v1"}, ) )[0] assert first_cycle["error"] is None ledger = await coordinator.task_store.load(root) first_validation = ledger.validations[first_cycle["validation_id"]] assert first_validation.verdict is ValidationVerdict.FAILED assert first_validation.unverified_claims == ["artifact.paragraphs[0].description"] assert first_validation.risks == ["REALIZATION_PLACEHOLDER"] split = await service.decide_script_task( task_id=compose_id, action=DecisionAction.SPLIT.value, reason="replace only the unrealized opening scope", validation_id=first_validation.validation_id, replacement_contract=None, child_contracts=(_contract("paragraph"),), context={"root_trace_id": root, "tool_call_id": "split-placeholder"}, ) replacement_id = split["payload"]["child_task_ids"][0] replacement_cycle = ( await service.dispatch_script_tasks( task_ids=(replacement_id,), context={"root_trace_id": root, "tool_call_id": "dispatch-replacement"}, ) )[0] replacement_accept = await service.decide_script_task( task_id=replacement_id, action=DecisionAction.ACCEPT.value, reason="the local replacement passed independent validation", validation_id=replacement_cycle["validation_id"], replacement_contract=None, child_contracts=(), context={"root_trace_id": root, "tool_call_id": "accept-replacement"}, ) ledger = await coordinator.task_store.load(root) assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN replacement_decision_id = replacement_accept["decision_id"] replacement_attempt = ledger.attempts[ cast(str, ledger.decisions[replacement_decision_id].attempt_id) ] assert replacement_attempt.submission is not None replacement_ref = replacement_attempt.submission.artifact_refs[0] frozen_replacement = { "decision_id": replacement_decision_id, "artifact_ref": { "uri": replacement_ref.uri, "kind": replacement_ref.kind, "version": replacement_ref.version, "digest": replacement_ref.digest, }, "scope_ref": "script-build://scopes/full", "expected_task_kind": "paragraph", } compose_v2 = _contract("compose", execution_ready=False) compose_v2["candidate_closure_decision_refs"] = [frozen_replacement] compose_v2["adopted_decision_ids"] = [replacement_decision_id] compose_v2["compose_order"] = [replacement_decision_id] await service.decide_script_task( task_id=compose_id, action=DecisionAction.REVISE.value, reason="pin the accepted replacement and retry the complete render", validation_id=None, replacement_contract=compose_v2, child_contracts=(), context={"root_trace_id": root, "tool_call_id": "revise-compose-v2"}, ) second_cycle = ( await service.dispatch_script_tasks( task_ids=(compose_id,), context={"root_trace_id": root, "tool_call_id": "dispatch-compose-v2"}, ) )[0] compose_accept = await service.decide_script_task( task_id=compose_id, action=DecisionAction.ACCEPT.value, reason="the revised full render passed independent validation", validation_id=second_cycle["validation_id"], replacement_contract=None, child_contracts=(), context={"root_trace_id": root, "tool_call_id": "accept-compose-v2"}, ) reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root) compose_task = reloaded.tasks[compose_id] assert compose_task.status is TaskStatus.COMPLETED assert compose_task.current_spec_version == 2 assert len(compose_task.attempt_ids) == 2 assert [ reloaded.validations[validation_id].verdict for validation_id in compose_task.validation_ids ] == [ValidationVerdict.FAILED, ValidationVerdict.PASSED] assert [reloaded.decisions[item].action for item in compose_task.decision_ids] == [ DecisionAction.SPLIT, DecisionAction.REVISE, DecisionAction.ACCEPT, ] assert compose_accept["status"] == TaskStatus.COMPLETED.value assert reloaded.tasks[replacement_id].status is TaskStatus.COMPLETED @pytest.mark.asyncio @pytest.mark.parametrize( "creation_order", [ ("element-set", "paragraph", "structure"), ("paragraph", "element-set", "structure"), ("structure", "paragraph", "element-set"), ], ) async def test_exploration_order_and_completion_order_do_not_choose_compose_order( tmp_path: Path, creation_order: tuple[str, str, str], ) -> None: service, coordinator, root = await _service(tmp_path) coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator)) portfolio = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "portfolio"}, ) compose = await service.plan_script_tasks( contract_payloads=[_contract("compose", execution_ready=False)], parent_task_id=portfolio["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "compose"}, ) compose_id = compose["task_ids"][0] task_by_kind: dict[str, str] = {} for kind in creation_order: planned = await service.plan_script_tasks( contract_payloads=[_contract(kind)], parent_task_id=compose_id, context={"root_trace_id": root, "tool_call_id": f"plan:{kind}"}, ) task_by_kind[kind] = planned["task_ids"][0] decision_by_kind: dict[str, str] = {} ref_by_kind: dict[str, ArtifactRef] = {} completion_order = tuple(reversed(creation_order)) for kind in completion_order: task_id = task_by_kind[kind] cycle = ( await service.dispatch_script_tasks( task_ids=(task_id,), context={"root_trace_id": root, "tool_call_id": f"dispatch:{kind}"}, ) )[0] accepted = await service.decide_script_task( task_id=task_id, action=DecisionAction.ACCEPT.value, reason=f"the bounded {kind} increment passed validation", validation_id=cycle["validation_id"], replacement_contract=None, child_contracts=(), context={"root_trace_id": root, "tool_call_id": f"accept:{kind}"}, ) decision_id = accepted["decision_id"] decision_by_kind[kind] = decision_id ledger = await coordinator.task_store.load(root) attempt = ledger.attempts[cast(str, ledger.decisions[decision_id].attempt_id)] assert attempt.submission is not None ref_by_kind[kind] = attempt.submission.artifact_refs[0] ledger = await coordinator.task_store.load(root) assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN adoption_kinds = ("paragraph", "structure", "element-set") closure = [] for kind in adoption_kinds: ref = ref_by_kind[kind] closure.append( { "decision_id": decision_by_kind[kind], "artifact_ref": { "uri": ref.uri, "kind": ref.kind, "version": ref.version, "digest": ref.digest, }, "scope_ref": "script-build://scopes/full", "expected_task_kind": kind, } ) execution_ready = _contract("compose", execution_ready=False) execution_ready["candidate_closure_decision_refs"] = closure execution_ready["adopted_decision_ids"] = [decision_by_kind[kind] for kind in adoption_kinds] execution_ready["compose_order"] = [decision_by_kind[kind] for kind in adoption_kinds] await service.decide_script_task( task_id=compose_id, action=DecisionAction.REVISE.value, reason="formal adoption is based on scope and intent, not timing", validation_id=None, replacement_contract=execution_ready, child_contracts=(), context={"root_trace_id": root, "tool_call_id": "close-compose"}, ) reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root) frozen = await service.contract_for_task(root, reloaded.tasks[compose_id]) assert frozen.contract.compose_order == tuple(decision_by_kind[kind] for kind in adoption_kinds) assert [ next(kind for kind, task_id in task_by_kind.items() if task_id == child_id) for child_id in reloaded.tasks[compose_id].child_task_ids ] == list(creation_order) assert completion_order == tuple(reversed(creation_order)) @pytest.mark.asyncio async def test_phase_two_retrieval_counts_toward_task_and_depth_limits(tmp_path: Path) -> None: service, _, root = await _service(tmp_path) service.limits = PhaseTwoLimits(max_tasks=2) portfolio = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root, "tool_call_id": "portfolio"}, ) compose = await service.plan_script_tasks( contract_payloads=[_contract("compose", execution_ready=False)], parent_task_id=portfolio["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "compose"}, ) with pytest.raises(TaskContractError, match="Task budget"): await service.plan_script_tasks( contract_payloads=[_contract("decode-retrieval")], parent_task_id=compose["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "retrieval"}, ) service.limits = PhaseTwoLimits(max_depth=1) with pytest.raises(TaskContractError, match="depth"): await service.plan_script_tasks( contract_payloads=[_contract("structure", scope="script-build://scopes/full/local")], parent_task_id=compose["task_ids"][0], context={"root_trace_id": root, "tool_call_id": "too-deep"}, ) @pytest.mark.asyncio async def test_task_contract_cannot_be_silently_changed_in_task_spec(tmp_path: Path) -> None: service, coordinator, root = await _service(tmp_path) result = await service.plan_script_tasks( contract_payloads=[_contract("candidate-portfolio", execution_ready=False)], parent_task_id=None, context={"root_trace_id": root}, ) ledger = await coordinator.task_store.load(root) task = ledger.tasks[result["task_ids"][0]] task.specs[-1] = replace(task.specs[-1], objective="tampered") await coordinator.task_store.commit(ledger, expected_revision=ledger.revision) with pytest.raises(TaskContractError, match="DIGEST_MISMATCH"): await service.contract_for_task(root, task)