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, _candidate_state_revision, ) 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.workbench import paragraph_target_key, semantic_handle 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={}, revision=1, ) 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_semantic_paragraph_save_is_atomic_and_exact_replay_is_idempotent( 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, ) scope = await service._workspace_scope(context, paragraph=True) before = await workspaces.snapshot(scope.write_context) revision = _candidate_state_revision(scope, before, ledger.revision) atom = [{"原子点": "observable reversal", "维度": "opening", "维度类型": "主维度"}] payload = ( { "client_key": "opening", "paragraph_index": 1, "name": "opening", "content_range": {"scope": "opening"}, "level": 1, "theme_elements": atom, "form_elements": atom, "function_elements": atom, "feeling_elements": [{"原子点": "curiosity", "维度": "tone"}], "theme": "A visible detail changes the original assumption.", "form": "Concrete setup followed by a concise reversal.", "function": "Give the audience a reason to continue.", "feeling": "Measured surprise grounded in observation.", "description": "The opening overturns one assumption with evidence.", "full_description": ( "Start from a familiar expectation and show exactly how the observed " "detail changes it." ), }, ) first = await service.save_script_paragraphs( paragraphs=payload, expected_state_revision=revision, context=context, ) replay = await service.save_script_paragraphs( paragraphs=payload, expected_state_revision=revision, context=context, ) assert first["status"] == "saved" assert first["created"] == 1 assert replay["status"] == "not_modified" assert replay["created"] == replay["updated"] == 0 assert replay["changed_target_keys"] == first["changed_target_keys"] assert len((await workspaces.snapshot(scope.write_context)).paragraphs) == 1 changed_payload = (dict(payload[0], name="changed after stale projection"),) with pytest.raises(PhaseTwoCandidateError, match="STALE_WORKBENCH_STATE"): await service.save_script_paragraphs( paragraphs=changed_payload, expected_state_revision=revision, 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"]}, "theme": "具体主题正文", "form": "对话漫画", "function": "建立冲突", "feeling": "好奇", "description": "角色在器械旁发生可见冲突。", "full_description": "角色先指责别人占用器械,随后被发现自己也在刷手机。", }, { "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"} workspace = await service.read_attempt_workspace(context=context) root_row = next(item for item in workspace["paragraphs"] if item["name"] == "opening") assert root_row["theme"] == "具体主题正文" assert root_row["full_description"].endswith("自己也在刷手机。") 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_element_set_materializes_all_accepted_paragraphs_before_linking( database: Any, ) -> None: _, sessions = database repository = SqlAlchemyScriptBusinessArtifactRepository(sessions) workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository) paragraph = replace( _complete_paragraph(), theme_elements=(({"原子点": "内容", "维度": "主题", "维度类型": "主维度"}),), form_elements=(({"原子点": "对话", "维度": "形式", "维度类型": "主维度"}),), function_elements=(({"原子点": "冲突", "维度": "作用", "维度类型": "主维度"}),), feeling_elements=(({"原子点": "好奇", "维度": "感受"}),), ) _, paragraph_ref = await repository.freeze( script_build_id=7, task_id="paragraph-source", attempt_id="paragraph-source-attempt", spec_version=1, artifact=ParagraphArtifactV1( _lineage(), (paragraph,), (), (), {"created": {"paragraph_ids": [1]}}, ), ) paragraph_ref = ArtifactRef( paragraph_ref.uri, paragraph_ref.kind, paragraph_ref.version, paragraph_ref.digest, ) accepted_ref = AcceptedDecisionRef( "paragraph-decision", paragraph_ref, SCOPE, ScriptTaskKind.PARAGRAPH, ) contract = ScriptTaskContractV1( task_kind=ScriptTaskKind.ELEMENT_SET, scope_ref=SCOPE, intent_class=ScriptIntentClass.EXPLORE, objective="attach concrete elements to every accepted paragraph", input_decision_refs=(accepted_ref,), base_artifact_ref=None, write_scope=( "script-build://writes/paragraphs", "script-build://writes/elements", ), gap_ref=None, output_schema="element-set-artifact/v1", criteria=(ScriptCriterion("closed", "all elements are linked"),), budget=ScriptTaskBudget(), goal_ids=("goal-1",), ) bundle = AcceptedInputBundleV1( ( AcceptedInput( "paragraph-decision", paragraph_ref, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit", ), ), DIGEST, ) (ledger, accepted), context = _scope_fixture(contract, bundle=bundle) service = PhaseTwoCandidateService( bindings=cast(Any, _Bindings()), task_store=_TaskStore(ledger), framework_artifact_store=_FrameworkArtifacts(), artifacts=repository, accepted_inputs=accepted, active_frontier=ActiveFrontierResolver(), workspaces=workspaces, ) workspace = await service.read_attempt_workspace(context=context) assert len(workspace["paragraphs"]) == 1 paragraph_id = workspace["paragraphs"][0]["paragraph_id"] assert paragraph_id > 0 assert workspace["paragraphs"][0]["full_description"] with pytest.raises(WorkspaceError, match="link paragraph"): await service.create_script_elements( elements=( { "client_key": "invalid", "name": "不应落库的元素", "dimension_primary": "实质", "dimension_secondary": "回滚验证", }, ), links=( { "paragraph_id": paragraph_id + 999, "element_client_keys": ["invalid"], }, ), context=context, ) rolled_back = await service.read_attempt_workspace(context=context) assert rolled_back["elements"] == [] assert rolled_back["paragraph_element_links"] == [] scope = await service._workspace_scope(context, element=True) before = await workspaces.snapshot(scope.write_context) revision = _candidate_state_revision(scope, before, ledger.revision) target = paragraph_target_key("root", "task", "attempt", paragraph_id) elements = ( { "client_key": "machine", "name": "被占用的高位下拉机", "dimension_primary": "实质", "dimension_secondary": "冲突道具", }, { "client_key": "bubble", "name": "占着不练的对话气泡", "dimension_primary": "形式", "dimension_secondary": "视觉文字", }, ) links = ( { "paragraph_target_key": target, "element_client_keys": ["machine", "bubble"], }, ) created = await service.save_script_elements( elements=elements, links=links, expected_state_revision=revision, context=context, ) replay = await service.save_script_elements( elements=elements, links=links, expected_state_revision=revision, context=context, ) assert created["status"] == "saved" assert created["created"] == 2 assert created["linked"] == 2 assert replay["status"] == "not_modified" final = await workspaces.snapshot(scope.write_context) assert len(final.elements) == 2 assert len(final.links) == 2 changed_elements = (dict(elements[0], name="changed after stale projection"), elements[1]) with pytest.raises(PhaseTwoCandidateError, match="STALE_WORKBENCH_STATE"): await service.save_script_elements( elements=changed_elements, links=links, expected_state_revision=revision, context=context, ) manifest = await service.resolve_attempt_manifest(context=context) assert manifest.artifact_ref.kind == ArtifactKind.ELEMENT_SET.value @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( image_handles=[semantic_handle("image", 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( image_handles=[semantic_handle("image", _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": [ {"decision_id": "candidate-1", "reason": "uses one visible detail"}, {"decision_id": "candidate-2", "reason": "remains more abstract"}, ], } ] frozen = await service.save_comparison_candidate( payload={ "criterion_results": criterion_results, "conflicts": ["both candidates write the opening scope"], "recommended_decision_id": "candidate-1", }, 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 [ item["artifact_ref"] for item in comparison.criterion_results[0]["candidate_results"] ] == [ref.uri for ref in candidate_refs] 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, "recommended_decision_id": "candidate-2", }, 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