from __future__ import annotations from dataclasses import replace from datetime import UTC, datetime from hashlib import sha256 from types import SimpleNamespace from typing import Any, cast import pytest from agent.orchestration import ArtifactRef from script_build_host.application.phase_two_candidates import ( PhaseTwoCandidateError, PhaseTwoCandidateService, ) from script_build_host.application.phase_two_inputs import ( ActiveFrontierResolver, PhaseTwoInputError, ) from script_build_host.domain.artifacts import ( ArtifactKind, ArtifactState, ArtifactVersion, DirectionArtifact, DirectionGoal, EvidenceRecordV1, ) from script_build_host.domain.errors import ProtocolViolation from script_build_host.domain.phase_two_artifacts import ( CandidateLineageV1, CandidatePortfolioArtifactV1, ComparisonArtifactV1, ElementSetArtifactV1, GoalCoverage, ParagraphArtifactV1, ScriptElementV1, ScriptParagraphElementLinkV1, ScriptParagraphV1, StructureArtifactV1, StructuredScriptArtifactV1, ) from script_build_host.domain.task_contracts import ( AcceptedDecisionRef, AcceptedInput, AcceptedInputBundleV1, ScriptCriterion, ScriptIntentClass, ScriptTaskBudget, ScriptTaskContractV1, ScriptTaskKind, ) from script_build_host.domain.workspaces import CandidateWriteContext, WorkspaceError from script_build_host.infrastructure.canonical_json import canonical_sha256 from script_build_host.repositories.sqlalchemy import ( SqlAlchemyScriptBusinessArtifactRepository, ) from script_build_host.repositories.workspace import ( SqlAlchemyCandidateWorkspaceRepository, ) DIGEST = "sha256:" + "a" * 64 SCOPE = "script-build://scopes/opening" WRITE = "script-build://writes/paragraphs/opening" def _complete_paragraph() -> ScriptParagraphV1: atom = ({"原子点": "内容", "维度": "开场"},) return ScriptParagraphV1( 1, 1, 1, None, "opening", {"beats": ["抛出问题", "给出答案"]}, theme="为什么要主动改变", form="问题与回答", function="建立紧迫感", feeling="坚定且真诚", theme_elements=atom, form_elements=atom, function_elements=atom, feeling_elements=atom, description="先指出停滞的代价。再给出可执行的转折。", full_description="你可能一直在等待更好的时机。但真正的改变来自今天的第一步。", ) def _complete_element() -> ScriptElementV1: return ScriptElementV1(1, "停滞代价", "实质", "核心论点") def _complete_link() -> ScriptParagraphElementLinkV1: return ScriptParagraphElementLinkV1(1, 1) def _lineage() -> CandidateLineageV1: return CandidateLineageV1( scope_ref=SCOPE, input_snapshot_ref="script-build://input-snapshots/11", input_closure_digest=DIGEST, write_scope=(WRITE,), goal_ids=("goal-1",), ) def _source_artifacts() -> tuple[ StructureArtifactV1, ParagraphArtifactV1, ElementSetArtifactV1, ]: paragraph = _complete_paragraph() element = _complete_element() link = _complete_link() lineage = _lineage() return ( StructureArtifactV1(lineage, (paragraph,)), ParagraphArtifactV1( lineage, (paragraph,), (element,), (link,), {"created": {"paragraph_ids": [1]}}, ), ElementSetArtifactV1( lineage, (element,), (link,), (paragraph,), {"created": {"element_ids": [1]}}, ), ) def _contract( kind: ScriptTaskKind, *, refs: tuple[AcceptedDecisionRef, ...] = (), comparison_refs: tuple[AcceptedDecisionRef, ...] = (), adopted: tuple[str, ...] = (), held: tuple[str, ...] = (), order: tuple[str, ...] = (), ) -> ScriptTaskContractV1: schemas = { ScriptTaskKind.PARAGRAPH: "paragraph-artifact/v1", ScriptTaskKind.DECODE_RETRIEVAL: "evidence-record/v1", ScriptTaskKind.COMPARE: "comparison-artifact/v1", ScriptTaskKind.COMPOSE: "structured-script/v1", ScriptTaskKind.CANDIDATE_PORTFOLIO: "candidate-portfolio/v1", } return ScriptTaskContractV1( task_kind=kind, scope_ref=SCOPE, intent_class=( ScriptIntentClass.COMPOSE if kind is ScriptTaskKind.COMPOSE else ScriptIntentClass.COMPARE if kind is ScriptTaskKind.COMPARE else ScriptIntentClass.EXPLORE ), objective="produce one bounded candidate increment", input_decision_refs=refs if kind is ScriptTaskKind.PARAGRAPH else (), base_artifact_ref=None, write_scope=(WRITE,), gap_ref=None, output_schema=schemas[kind], criteria=(ScriptCriterion("closed", "output is concretely complete"),), budget=ScriptTaskBudget(), goal_ids=("goal-1",), candidate_closure_decision_refs=( refs if kind in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO} else () ), adopted_decision_ids=adopted, held_or_rejected_decision_ids=held, compose_order=order, comparison_decision_refs=comparison_refs, ) class _Bindings: def __init__(self, active_direction_artifact_version_id: int = 10) -> None: self.active_direction_artifact_version_id = active_direction_artifact_version_id async def get_by_root(self, root_trace_id: str) -> Any: return SimpleNamespace( root_trace_id=root_trace_id, script_build_id=7, input_snapshot_id=11, active_direction_artifact_version_id=self.active_direction_artifact_version_id, ) class _TaskStore: def __init__(self, ledger: Any) -> None: self.ledger = ledger async def load(self, root_trace_id: str) -> Any: assert root_trace_id == "root" return self.ledger class _AcceptedInputs: def __init__( self, contracts: dict[str, ScriptTaskContractV1], bundle: AcceptedInputBundleV1 ) -> None: self.contracts = contracts self.bundle = bundle async def contract_for_task(self, *, task: Any, **_: Any) -> ScriptTaskContractV1: return self.contracts[task.task_id] async def resolve(self, **_: Any) -> AcceptedInputBundleV1: return self.bundle class _FrameworkArtifacts: async def get(self, *_: Any) -> Any: raise AssertionError("framework snapshot should not be read") class _SnapshotArtifacts: def __init__(self, ref: ArtifactRef) -> None: self.ref = ref async def get(self, root_trace_id: str, snapshot_id: str) -> Any: assert (root_trace_id, snapshot_id) == ("root", "validation-snapshot") return SimpleNamespace(artifact_refs=[self.ref], evidence_refs=[]) class _RawStore: def __init__(self, content: bytes) -> None: self.content = content async def read_bytes(self, ref: str) -> bytes: assert ref.endswith(sha256(self.content).hexdigest()) return self.content class _Artifacts: def __init__(self, versions: dict[str, ArtifactVersion]) -> None: self.versions = versions self.frozen: ArtifactVersion | None = None async def read_by_ref(self, ref: ArtifactRef, **_: Any) -> ArtifactVersion: return self.versions[ref.uri] async def get_by_attempt(self, **owners: Any) -> ArtifactVersion: for value in self.versions.values(): if value.attempt_id == owners["attempt_id"]: return value raise AssertionError("Attempt artifact not found") async def get_by_id(self, identifier: int, **_: Any) -> ArtifactVersion: for value in self.versions.values(): if value.artifact_version_id == identifier: return value if identifier != 10: raise AssertionError("Artifact version not found") direction = DirectionArtifact( goals=( DirectionGoal( "goal-1", "complete the script", "the delivery needs one complete script", success_criteria=("the script is complete",), ), ), evidence_refs=("script-build://artifact-versions/90",), ) return ArtifactVersion( 10, 7, "direction", "direction-attempt", 1, ArtifactKind.DIRECTION, DIGEST, ArtifactState.FROZEN, direction, datetime.now(UTC), datetime.now(UTC), ) async def freeze(self, *, artifact: Any, **owners: Any) -> tuple[ArtifactVersion, ArtifactRef]: digest = canonical_sha256(artifact.content_payload()).wire artifact = replace(artifact, canonical_sha256=digest) artifact_kind = ( ArtifactKind.COMPARISON if isinstance(artifact, ComparisonArtifactV1) else ArtifactKind.CANDIDATE_PORTFOLIO ) version = ArtifactVersion( artifact_version_id=99, script_build_id=owners["script_build_id"], task_id=owners["task_id"], attempt_id=owners["attempt_id"], spec_version=owners["spec_version"], artifact_type=artifact_kind, canonical_sha256=digest, state=ArtifactState.FROZEN, artifact=artifact, created_at=datetime.now(UTC), frozen_at=datetime.now(UTC), ) self.frozen = version ref = ArtifactRef( "script-build://artifact-versions/99", artifact_kind.value, "99", digest, ) self.versions[ref.uri] = version return version, ref def _scope_fixture( contract: ScriptTaskContractV1, *, bundle: AcceptedInputBundleV1 | None = None, task_id: str = "task", attempt_id: str = "attempt", ) -> tuple[Any, dict[str, Any]]: task = SimpleNamespace(task_id=task_id, attempt_ids=[attempt_id]) attempt = SimpleNamespace( attempt_id=attempt_id, task_id=task_id, spec_version=1, accepted_child_decision_ids=(), ) ledger = SimpleNamespace( tasks={task_id: task}, attempts={attempt_id: attempt}, decisions={}, ) accepted = _AcceptedInputs( {task_id: contract}, bundle or AcceptedInputBundleV1((), DIGEST), ) return (ledger, accepted), { "root_trace_id": "root", "task_id": task_id, "attempt_id": attempt_id, "spec_version": 1, } @pytest.mark.asyncio async def test_phase_one_manifest_does_not_require_candidate_workspace() -> None: contract = _contract(ScriptTaskKind.DECODE_RETRIEVAL) (ledger, accepted), context = _scope_fixture(contract) evidence = EvidenceRecordV1( evidence_id="evidence", source_type="decode", tool_name="retrieve_decode", query={}, source_refs=("source",), raw_artifact_ref=None, summary="bounded evidence", supports=("goal",), confidence="high", limitations=(), content_sha256=DIGEST, created_at=datetime.now(UTC), ) version = ArtifactVersion( 1, 7, "task", "attempt", 1, ArtifactKind.EVIDENCE, DIGEST, ArtifactState.FROZEN, evidence, datetime.now(UTC), datetime.now(UTC), ) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=cast(Any, _Artifacts({"script-build://artifact-versions/1": version})), accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=None, ) manifest = await service.resolve_attempt_manifest(context=context) assert manifest.artifact_ref.kind == "evidence" assert manifest.scope_ref == SCOPE @pytest.mark.asyncio async def test_paragraph_workspace_uses_positive_branch_and_freezes_on_manifest( database: Any, ) -> None: _, sessions = database repository = SqlAlchemyScriptBusinessArtifactRepository(sessions) workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository) contract = _contract(ScriptTaskKind.PARAGRAPH) (ledger, accepted), context = _scope_fixture(contract) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=repository, accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=workspaces, ) created = await service.create_script_paragraph( payload={"paragraph_index": 1, "name": "opening", "content_range": {}}, context=context, ) manifest = await service.resolve_attempt_manifest(context=context) version = await repository.get_by_attempt( script_build_id=7, task_id="task", attempt_id="attempt" ) assert created["paragraph_id"] > 0 assert version.artifact_version_id > 0 assert manifest.artifact_ref.digest == version.canonical_sha256 assert manifest.artifact_ref.kind == "paragraph" with pytest.raises(PhaseTwoCandidateError, match="WRITE_SCOPE_VIOLATION"): await service.create_script_paragraph( payload={ "paragraph_index": 2, "name": "forged", "content_range": {}, "branch_id": 0, }, context=context, ) @pytest.mark.asyncio async def test_paragraph_batch_maps_parent_keys_and_rolls_back_atomically(database: Any) -> None: _, sessions = database repository = SqlAlchemyScriptBusinessArtifactRepository(sessions) workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository) contract = _contract(ScriptTaskKind.PARAGRAPH) (ledger, accepted), context = _scope_fixture(contract) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=repository, accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=workspaces, ) created = await service.create_script_paragraphs( paragraphs=( { "client_key": "root", "paragraph_index": 1, "name": "opening", "content_range": {"beats": ["setup", "turn"]}, }, { "client_key": "child", "parent_client_key": "root", "paragraph_index": 2, "level": 2, "name": "turn", "content_range": {"beats": ["turn"]}, }, ), context=context, ) assert created["created"] == 2 assert set(created["paragraph_ids_by_client_key"]) == {"root", "child"} contract2 = replace(contract, objective="second isolated workspace") (ledger2, accepted2), context2 = _scope_fixture( contract2, task_id="task-2", attempt_id="attempt-2" ) service2 = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger2), framework_artifact_store=_FrameworkArtifacts(), artifacts=repository, accepted_inputs=accepted2, active_frontier=ActiveFrontierResolver(), workspaces=workspaces, ) with pytest.raises(WorkspaceError, match="child content range"): await service2.create_script_paragraphs( paragraphs=( { "client_key": "root", "paragraph_index": 1, "name": "root", "content_range": {"beats": ["setup"]}, }, { "client_key": "child", "parent_client_key": "root", "paragraph_index": 2, "level": 2, "name": "child", "content_range": {"beats": ["missing"]}, }, ), context=context2, ) write_context = CandidateWriteContext( script_build_id=7, task_id="task-2", attempt_id="attempt-2", spec_version=1, objective=contract2.objective, input_refs=(), write_scope=contract2.write_scope, ) assert not (await workspaces.snapshot(write_context)).paragraphs @pytest.mark.asyncio async def test_candidate_service_discards_existing_draft_workspace_on_abandoned_attempt( database: Any, ) -> None: _, sessions = database repository = SqlAlchemyScriptBusinessArtifactRepository(sessions) workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository) contract = _contract(ScriptTaskKind.PARAGRAPH) (ledger, accepted), context = _scope_fixture(contract) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=repository, accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=workspaces, ) await service.create_script_paragraph( payload={"paragraph_index": 1, "name": "draft", "content_range": {}}, context=context, ) await service.discard_attempt( root_trace_id="root", task_id="task", attempt_id="attempt", reason="validator blocked the bounded increment", ) write_context = CandidateWriteContext( script_build_id=7, task_id="task", attempt_id="attempt", spec_version=1, objective=contract.objective, input_refs=(), write_scope=contract.write_scope, ) assert (await workspaces.require(write_context)).state is ArtifactState.DISCARDED with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"): await service.create_script_paragraph( payload={"paragraph_index": 2, "name": "late", "content_range": {}}, context=context, ) @pytest.mark.asyncio async def test_frozen_image_must_be_accepted_and_digest_and_mime_are_rechecked() -> None: content = b"\x89PNG\r\n\x1a\n" + b"safe" raw_ref = f"script-build://raw-artifacts/sha256/{sha256(content).hexdigest()}" ref = ArtifactRef("script-build://artifact-versions/2", "evidence", "2", DIGEST) item = AcceptedInput("d1", ref, ScriptTaskKind.DECODE_RETRIEVAL, SCOPE, "explicit") bundle = AcceptedInputBundleV1((item,), DIGEST) contract = _contract(ScriptTaskKind.PARAGRAPH) (ledger, accepted), context = _scope_fixture(contract, bundle=bundle) evidence = EvidenceRecordV1( "e", "decode", "retrieve_decode", {}, ("source",), raw_ref, "summary", ("goal",), "high", (), DIGEST, datetime.now(UTC), ) version = ArtifactVersion( 2, 7, "child", "child-attempt", 1, ArtifactKind.EVIDENCE, DIGEST, ArtifactState.FROZEN, evidence, datetime.now(UTC), datetime.now(UTC), ) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=cast(Any, _Artifacts({ref.uri: version})), accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=None, raw_artifacts=_RawStore(content), ) images = await service.view_frozen_images(raw_artifact_refs=[raw_ref], context=context) assert images[0]["media_type"] == "image/png" assert "url" not in images[0] with pytest.raises(PhaseTwoCandidateError, match="outside the accepted"): await service.view_frozen_images(raw_artifact_refs=[_RAW_OTHER], context=context) @pytest.mark.asyncio async def test_active_frontier_write_conflict_is_not_resolved_by_completion_order() -> None: ref1 = ArtifactRef("script-build://artifact-versions/1", "paragraph", "1", DIGEST) ref2 = ArtifactRef("script-build://artifact-versions/2", "paragraph", "2", DIGEST) refs = ( AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.PARAGRAPH), AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.PARAGRAPH), ) contract = _contract( ScriptTaskKind.COMPOSE, refs=refs, adopted=("d1", "d2"), order=("d1", "d2") ) bundle = AcceptedInputBundleV1( ( AcceptedInput("d1", ref1, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"), AcceptedInput("d2", ref2, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"), ), DIGEST, ) (ledger, accepted), context = _scope_fixture(contract, bundle=bundle) child_contract = _contract(ScriptTaskKind.PARAGRAPH) for decision_id in ("d1", "d2"): task_id = f"task-{decision_id}" ledger.tasks[task_id] = SimpleNamespace(task_id=task_id, attempt_ids=[f"a-{decision_id}"]) ledger.attempts[f"a-{decision_id}"] = SimpleNamespace( attempt_id=f"a-{decision_id}", task_id=task_id, spec_version=1 ) ledger.decisions[decision_id] = SimpleNamespace( task_id=task_id, attempt_id=f"a-{decision_id}" ) accepted.contracts[task_id] = child_contract service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=cast(Any, _Artifacts({})), accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=None, ) with pytest.raises(PhaseTwoInputError, match="WRITE_SCOPE_CONFLICT"): await service.read_active_frontier(context=context) @pytest.mark.asyncio async def test_comparison_candidates_are_contract_derived_and_pass_fairness_precheck() -> None: candidate_refs = ( ArtifactRef("script-build://artifact-versions/1", "paragraph", "1", DIGEST), ArtifactRef("script-build://artifact-versions/2", "paragraph", "2", DIGEST), ) decision_refs = tuple( AcceptedDecisionRef( f"candidate-{index}", ref, SCOPE, ScriptTaskKind.PARAGRAPH, ) for index, ref in enumerate(candidate_refs, start=1) ) bundle = AcceptedInputBundleV1( tuple( AcceptedInput( decision_ref.decision_id, decision_ref.artifact_ref, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit", ) for decision_ref in decision_refs ), DIGEST, ) contract = _contract(ScriptTaskKind.COMPARE, comparison_refs=decision_refs) (ledger, accepted), context = _scope_fixture(contract, bundle=bundle) artifacts = _Artifacts({}) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=cast(Any, artifacts), accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=None, ) criterion_results = [ { "criterion_id": "closed", "candidate_results": [ {"artifact_ref": candidate_refs[0].uri, "reason": "uses one visible detail"}, {"artifact_ref": candidate_refs[1].uri, "reason": "remains more abstract"}, ], } ] frozen = await service.save_comparison_candidate( payload={ "criterion_results": criterion_results, "conflicts": ["both candidates write the opening scope"], "recommendation": candidate_refs[0].uri, }, context=context, ) assert artifacts.frozen is not None comparison = artifacts.frozen.artifact assert isinstance(comparison, ComparisonArtifactV1) assert comparison.candidate_artifact_refs == tuple(ref.uri for ref in candidate_refs) assert comparison.criterion_results == tuple(criterion_results) frozen_ref = ArtifactRef.from_dict(frozen["artifact_ref"]) service._framework_artifact_store = _SnapshotArtifacts(frozen_ref) rules = await service.deterministic_precheck( context={**context, "snapshot_id": "validation-snapshot"} ) assert {str(item["rule_id"]): item["verdict"] for item in rules} == { "business-artifact-owner": "passed", "realized-content": "passed", "candidate-lineage": "passed", "comparison-fairness": "passed", } with pytest.raises(PhaseTwoCandidateError, match="derived from the frozen contract"): await service.save_comparison_candidate( payload={ "candidate_artifact_refs": [candidate_refs[1].uri], "criterion_results": criterion_results, "recommendation": candidate_refs[1].uri, }, context=context, ) @pytest.mark.asyncio async def test_portfolio_adoption_and_canonical_digest_come_from_frozen_contract() -> None: ref1 = ArtifactRef("script-build://artifact-versions/1", "structured_script", "1", DIGEST) ref2 = ArtifactRef("script-build://artifact-versions/2", "structured_script", "2", DIGEST) refs = ( AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.COMPOSE), AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.COMPOSE), ) contract = _contract( ScriptTaskKind.CANDIDATE_PORTFOLIO, refs=refs, adopted=("d1",), held=("d2",), order=("d1",), ) bundle = AcceptedInputBundleV1( ( AcceptedInput("d1", ref1, ScriptTaskKind.COMPOSE, SCOPE, "explicit"), AcceptedInput("d2", ref2, ScriptTaskKind.COMPOSE, SCOPE, "explicit"), ), DIGEST, ) (ledger, accepted), context = _scope_fixture(contract, bundle=bundle) paragraph = _complete_paragraph() element = _complete_element() link = _complete_link() def source_versions(identifier: int) -> dict[str, ArtifactVersion]: base = 20 + identifier * 10 versions: dict[str, ArtifactVersion] = {} for offset, (kind, artifact) in enumerate( zip( ( ArtifactKind.STRUCTURE, ArtifactKind.PARAGRAPH, ArtifactKind.ELEMENT_SET, ), _source_artifacts(), strict=True, ), start=1, ): version_id = base + offset uri = f"script-build://artifact-versions/{version_id}" versions[uri] = ArtifactVersion( version_id, 7, f"source-{version_id}", f"source-attempt-{version_id}", 1, kind, DIGEST, ArtifactState.FROZEN, artifact, datetime.now(UTC), datetime.now(UTC), ) return versions def structured(identifier: int) -> ArtifactVersion: source_refs = tuple(source_versions(identifier)) artifact = StructuredScriptArtifactV1( direction_ref="script-build://artifact-versions/10", input_closure_digest=DIGEST, paragraphs=(paragraph,), elements=(element,), paragraph_element_links=(link,), source_artifact_refs=source_refs, goal_coverage=( GoalCoverage("goal-1", (source_refs[1], source_refs[2])), ), evidence_refs=(), acceptance_notes=("accepted",), canonical_sha256=DIGEST, ) return ArtifactVersion( identifier, 7, f"compose-{identifier}", f"compose-attempt-{identifier}", 1, ArtifactKind.STRUCTURED_SCRIPT, DIGEST, ArtifactState.FROZEN, artifact, datetime.now(UTC), datetime.now(UTC), ) versions = { **source_versions(1), **source_versions(2), ref1.uri: structured(1), ref2.uri: structured(2), } artifacts = _Artifacts(versions) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=cast(Any, artifacts), accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=None, ) result = await service.save_candidate_portfolio( payload={ "unresolved_defects": [ { "severity": "warning", "defect_code": "MINOR_STYLE", "criterion_id": "closed", "scope_ref": SCOPE, "observed_excerpt": "opening cadence is slightly repetitive", "evidence_refs": [], "invalidated_inputs": [], "recommended_action_class": "revise", } ], }, context=context, ) assert artifacts.frozen is not None portfolio = artifacts.frozen.artifact assert isinstance(portfolio, CandidatePortfolioArtifactV1) assert portfolio.accepted_decision_ids == ("d1",) assert portfolio.rejected_or_held_decision_ids == ("d2",) assert result["artifact_ref"]["digest"] == canonical_sha256(portfolio.content_payload()).wire with pytest.raises(PhaseTwoCandidateError, match="outside the accepted closure"): await service.save_candidate_portfolio( payload={ "unresolved_defects": [ { "severity": "warning", "defect_code": "UNTRUSTED_EVIDENCE", "criterion_id": "closed", "scope_ref": SCOPE, "observed_excerpt": "unsupported warning", "evidence_refs": ["script-build://artifact-versions/999"], "invalidated_inputs": [], "recommended_action_class": "revise", } ], }, context=context, ) with pytest.raises(ProtocolViolation, match=r"unsupported enum|must not be blank"): await service.save_candidate_portfolio( payload={ "unresolved_defects": [ {"severity": "warning", "defect_code": "INCOMPLETE_WARNING"} ], }, context=context, ) with pytest.raises(PhaseTwoCandidateError, match="derived from the frozen contract"): await service.save_candidate_portfolio( payload={"adopted_structured_script_ref": ref2.uri}, context=context ) @pytest.mark.asyncio async def test_candidate_portfolio_command_replay_reuses_one_frozen_version(database: Any) -> None: _, sessions = database repository = SqlAlchemyScriptBusinessArtifactRepository(sessions) paragraph = _complete_paragraph() element = _complete_element() link = _complete_link() direction_version, direction_ref = await repository.freeze( script_build_id=7, task_id="direction", attempt_id="direction-attempt", spec_version=1, artifact=DirectionArtifact( goals=( DirectionGoal( "goal-1", "complete the script", "the delivery needs one complete script", success_criteria=("the script is complete",), ), ), evidence_refs=("script-build://artifact-versions/90",), ), ) async def freeze_structured(identifier: int) -> tuple[ArtifactVersion, ArtifactRef]: structure, paragraph_source, element_source = _source_artifacts() source_refs: list[str] = [] for suffix, source in ( ("structure", structure), ("paragraph", paragraph_source), ("elements", element_source), ): _, source_ref = await repository.freeze( script_build_id=7, task_id=f"{suffix}-{identifier}", attempt_id=f"{suffix}-attempt-{identifier}", spec_version=1, artifact=source, ) source_refs.append(source_ref.uri) return await repository.freeze( script_build_id=7, task_id=f"compose-{identifier}", attempt_id=f"compose-attempt-{identifier}", spec_version=1, artifact=StructuredScriptArtifactV1( direction_ref=direction_ref.uri, input_closure_digest=DIGEST, paragraphs=(paragraph,), elements=(element,), paragraph_element_links=(link,), source_artifact_refs=tuple(source_refs), goal_coverage=( GoalCoverage("goal-1", (source_refs[1], source_refs[2])), ), evidence_refs=(), acceptance_notes=("accepted",), ), ) (_, stored_ref1), (_, stored_ref2) = await freeze_structured(1), await freeze_structured(2) ref1 = ArtifactRef( stored_ref1.uri, stored_ref1.kind, stored_ref1.version, stored_ref1.digest, ) ref2 = ArtifactRef( stored_ref2.uri, stored_ref2.kind, stored_ref2.version, stored_ref2.digest, ) refs = ( AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.COMPOSE), AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.COMPOSE), ) contract = _contract( ScriptTaskKind.CANDIDATE_PORTFOLIO, refs=refs, adopted=("d1",), held=("d2",), order=("d1",), ) bundle = AcceptedInputBundleV1( ( AcceptedInput("d1", ref1, ScriptTaskKind.COMPOSE, SCOPE, "explicit"), AcceptedInput("d2", ref2, ScriptTaskKind.COMPOSE, SCOPE, "explicit"), ), DIGEST, ) (ledger, accepted), context = _scope_fixture( contract, bundle=bundle, task_id="portfolio", attempt_id="portfolio-attempt", ) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings(direction_version.artifact_version_id)), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=repository, accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=None, ) first = await service.save_candidate_portfolio( payload={"unresolved_defects": []}, context=context ) replay = await service.save_candidate_portfolio( payload={"unresolved_defects": []}, context=context ) frozen = await repository.get_by_attempt( script_build_id=7, task_id="portfolio", attempt_id="portfolio-attempt", ) assert first == replay assert first["artifact_ref"]["version"] == str(frozen.artifact_version_id) assert frozen.artifact_type is ArtifactKind.CANDIDATE_PORTFOLIO _RAW_OTHER = "script-build://raw-artifacts/sha256/" + "b" * 64