from __future__ import annotations import asyncio from dataclasses import replace import pytest from sqlalchemy import func, select from script_build_host.domain.artifacts import ArtifactKind, ArtifactState from script_build_host.domain.phase_two_artifacts import CandidateLineageV1, ParagraphArtifactV1 from script_build_host.domain.workspaces import CandidateWriteContext, WorkspaceError from script_build_host.infrastructure.legacy_tables import ( script_build_element, script_build_paragraph, script_build_paragraph_element, script_build_task_plan_step, ) from script_build_host.infrastructure.tables import artifact_version_table from script_build_host.repositories.sqlalchemy import ( SqlAlchemyScriptBusinessArtifactRepository, ) from script_build_host.repositories.workspace import ( SqlAlchemyCandidateWorkspaceRepository, ) def _context(attempt: str, *, task: str = "task-1", build: int = 1) -> CandidateWriteContext: return CandidateWriteContext( script_build_id=build, task_id=task, attempt_id=attempt, spec_version=1, objective="produce the opening increment", input_refs=("script-build://artifact-versions/90",), write_scope=("script-build://scopes/opening",), ) def _lineage( *, base_ref: str | None = None, base_digest: str | None = None, base_revision: int | None = None, ) -> CandidateLineageV1: return CandidateLineageV1( scope_ref="script-build://scopes/opening", input_snapshot_ref="script-build://inputs/11", input_closure_digest="sha256:" + "a" * 64, write_scope=("script-build://scopes/opening",), base_artifact_ref=base_ref, base_artifact_digest=base_digest, base_revision=base_revision, ) @pytest.mark.asyncio async def test_workspace_allocates_positive_branch_and_freezes_full_paragraph(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) context = _context("attempt-paragraph") workspace = await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH) assert workspace.branch_id == workspace.artifact_version_id > 0 assert ( await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH) == workspace ) parent = await repository.create_paragraph( context, paragraph_index=1, name="opening", content_range={"topics": [1, 2]}, theme_elements=({"维度": "主题", "原子点": "冲突", "维度类型": "主维度"},), form_elements=({"维度": "形式", "原子点": "对照", "维度类型": "主维度"},), ) child = await repository.create_paragraph( context, paragraph_index=2, name="opening detail", content_range={"topics": [1]}, level=2, parent_paragraph_id=parent, ) await repository.batch_update_paragraphs( context, [ { "paragraph_id": parent, "theme": "a concrete conflict", "form": "contrast", "function": "hook", "feeling": "curiosity", "description": "bounded description", "full_description": "fully realized opening", } ], ) added = await repository.append_paragraph_atoms( context, paragraph_id=child, column="feeling_elements", atoms=({"维度": "情绪", "原子点": "期待"},), ) assert added == 1 version, reference = await repository.freeze(context, lineage=_lineage()) assert version.state is ArtifactState.FROZEN assert reference.kind == ArtifactKind.PARAGRAPH.value assert isinstance(version.artifact, ParagraphArtifactV1) assert len(version.artifact.paragraphs) == 2 assert version.artifact.paragraphs[0].full_description == "fully realized opening" with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"): await repository.create_paragraph( context, paragraph_index=3, name="late mutation", content_range={"topics": [2]}, ) replayed, replay_ref = await repository.freeze(context, lineage=_lineage()) assert (replayed.artifact_version_id, replay_ref) == ( version.artifact_version_id, reference, ) with pytest.raises(WorkspaceError, match="STALE_BASE_REVISION"): await repository.freeze( context, lineage=CandidateLineageV1( scope_ref="script-build://scopes/opening", input_snapshot_ref="script-build://inputs/12", input_closure_digest="sha256:" + "b" * 64, write_scope=("script-build://scopes/opening",), ), ) async with sessions() as session: plan = ( ( await session.execute( select(script_build_task_plan_step).where( script_build_task_plan_step.c.script_build_id == 1, script_build_task_plan_step.c.branch_id == workspace.branch_id, ) ) ) .mappings() .one() ) assert plan["step_key"] == "1" assert plan["round_index"] is None assert plan["status"] == "完成" @pytest.mark.asyncio async def test_semantic_write_scope_is_enforced_before_atomic_batch(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) full_context = _context("write-scope") await repository.get_or_create(full_context, artifact_kind=ArtifactKind.PARAGRAPH) first = await repository.create_paragraph( full_context, paragraph_index=1, name="one", content_range={"topics": [1]} ) second = await repository.create_paragraph( full_context, paragraph_index=2, name="two", content_range={"topics": [2]} ) bounded = replace(full_context, write_scope=(f"script-build://writes/paragraphs/{first}",)) with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"): await repository.batch_update_paragraphs( bounded, ( {"paragraph_id": first, "description": "authorized but must roll back"}, {"paragraph_id": second, "description": "outside scope"}, ), ) snapshot = await repository.snapshot(full_context) assert [item.description for item in snapshot.paragraphs] == [None, None] with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"): await repository.create_element( bounded, name="not allowed", dimension_primary="实质", dimension_secondary="detail", ) @pytest.mark.asyncio async def test_content_range_scope_must_cover_every_key_and_value(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) context = replace( _context("content-range-scope"), write_scope=("script-build://writes/content-ranges/topics/1",), ) await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH) with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"): await repository.create_paragraph( context, paragraph_index=1, name="partially authorized", content_range={"topics": [1, 2], "sections": ["ending"]}, ) snapshot = await repository.snapshot(context) assert snapshot.paragraphs == () @pytest.mark.asyncio async def test_discard_marks_workspace_and_legacy_plan_blocked(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) context = _context("discarded-attempt") workspace = await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH) await repository.create_paragraph( context, paragraph_index=1, name="draft", content_range={"topics": [1]} ) discarded = await repository.discard(context, reason="validator hard defect") assert discarded.state is ArtifactState.DISCARDED assert ( await repository.discard(context, reason="idempotent replay") ).state is ArtifactState.DISCARDED async with sessions() as session: plan = ( ( await session.execute( select(script_build_task_plan_step).where( script_build_task_plan_step.c.script_build_id == context.script_build_id, script_build_task_plan_step.c.branch_id == workspace.branch_id, ) ) ) .mappings() .one() ) assert plan["status"] == "受阻" assert plan["result_note"] == "validator hard defect" with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"): await repository.create_paragraph( context, paragraph_index=2, name="late", content_range={"topics": [2]} ) @pytest.mark.asyncio async def test_element_and_link_batch_is_atomic_and_workspace_scoped(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) paragraph_context = _context("base-paragraph") await repository.get_or_create(paragraph_context, artifact_kind=ArtifactKind.PARAGRAPH) paragraph_id = await repository.create_paragraph( paragraph_context, paragraph_index=1, name="body", content_range={"topics": [1]}, ) base, base_ref = await repository.freeze(paragraph_context, lineage=_lineage()) element_context = _context("element-attempt", task="element-task") workspace = await repository.get_or_create( element_context, artifact_kind=ArtifactKind.ELEMENT_SET, base_artifact_ref=base_ref, ) snapshot = await repository.snapshot(element_context) copied_paragraph = snapshot.paragraphs[0].paragraph_id assert copied_paragraph != paragraph_id assert snapshot.source_identity_map == ( { "entity": "paragraph", "source_local_id": paragraph_id, "local_id": copied_paragraph, }, ) element_id = await repository.create_element( element_context, name="specific observation", dimension_primary="实质", dimension_secondary="detail", commonality_analysis={"shared": "grounded"}, topic_support={"topic": 1}, weight_score={"score": 0.8}, support_elements=({"ref": "source-1"},), ) with pytest.raises(WorkspaceError, match="LEGACY_REFERENCE_INVALID"): await repository.batch_link( element_context, [(copied_paragraph, [element_id]), (999999, [element_id])], ) async with sessions() as session: assert ( await session.scalar(select(func.count()).select_from(script_build_paragraph_element)) ) == 0 assert await repository.batch_link(element_context, [(copied_paragraph, [element_id])]) == 1 with pytest.raises(WorkspaceError, match="STALE_BASE_REVISION"): await repository.freeze( element_context, lineage=_lineage( base_ref=base_ref.uri, base_digest="sha256:" + "f" * 64, base_revision=base.artifact_version_id, ), ) version, _ = await repository.freeze( element_context, lineage=_lineage( base_ref=base_ref.uri, base_digest=base_ref.digest, base_revision=base.artifact_version_id, ), ) assert version.artifact.lineage.source_lineage[0]["source_local_id"] == paragraph_id assert workspace.branch_id == version.artifact_version_id @pytest.mark.asyncio async def test_wrong_adapter_cross_context_and_stale_base_are_rejected(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) context = _context("element-only") await repository.get_or_create(context, artifact_kind=ArtifactKind.ELEMENT_SET) with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"): await repository.create_paragraph( context, paragraph_index=1, name="forbidden", content_range={"topics": [1]}, ) with pytest.raises(WorkspaceError, match="LEGACY_REFERENCE_INVALID"): await repository.require(_context("element-only", task="other-task")) with pytest.raises(WorkspaceError, match="STALE_BASE_REVISION"): await repository.freeze( context, lineage=CandidateLineageV1( scope_ref="script-build://scopes/opening", input_snapshot_ref="script-build://inputs/11", input_closure_digest="sha256:" + "a" * 64, write_scope=("script-build://scopes/opening",), base_artifact_ref="script-build://artifact-versions/999", base_artifact_digest="sha256:" + "b" * 64, base_revision=999, ), ) async with sessions() as session: counts = { "paragraph": await session.scalar( select(func.count()).select_from(script_build_paragraph) ), "element": await session.scalar(select(func.count()).select_from(script_build_element)), } assert counts == {"paragraph": 0, "element": 0} @pytest.mark.asyncio async def test_concurrent_workspace_replay_converges_on_one_attempt_and_branch(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) context = _context("attempt-concurrent") first, second = await asyncio.gather( repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH), repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH), ) assert first == second assert first.branch_id == first.artifact_version_id > 0 async with sessions() as session: rows = ( ( await session.execute( select(artifact_version_table).where( artifact_version_table.c.attempt_id == context.attempt_id ) ) ) .mappings() .all() ) assert len(rows) == 1 assert rows[0]["legacy_branch_id"] == rows[0]["id"] @pytest.mark.asyncio async def test_batch_update_validates_every_paragraph_before_mutating(database) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) context = _context("attempt-atomic-update") await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH) first_id = await repository.create_paragraph( context, paragraph_index=1, name="original", content_range={"topics": [1]}, ) second_id = await repository.create_paragraph( context, paragraph_index=2, name="second", content_range={"topics": [2]}, ) with pytest.raises(WorkspaceError, match="LEGACY_REFERENCE_INVALID"): await repository.batch_update_paragraphs( context, [ {"paragraph_id": first_id, "name": "must roll back"}, {"paragraph_id": 999_999, "name": "outside workspace"}, ], ) snapshot = await repository.snapshot(context) assert snapshot.paragraphs[0].name == "original" with pytest.raises(WorkspaceError, match="paragraph name"): await repository.batch_update_paragraphs( context, [ {"paragraph_id": first_id, "description": "must still roll back"}, {"paragraph_id": second_id, "name": ""}, ], ) snapshot = await repository.snapshot(context) assert snapshot.paragraphs[0].description is None assert snapshot.paragraphs[1].name == "second" @pytest.mark.asyncio async def test_element_patch_copies_complete_projection_and_freezes_change_manifest( database, ) -> None: _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) paragraph_context = _context("paragraph-base") await repository.get_or_create(paragraph_context, artifact_kind=ArtifactKind.PARAGRAPH) source_paragraph_id = await repository.create_paragraph( paragraph_context, paragraph_index=1, name="body", content_range={"topics": [1]}, ) paragraph_version, paragraph_ref = await repository.freeze( paragraph_context, lineage=_lineage() ) element_context = _context("element-base", task="element-base-task") await repository.get_or_create( element_context, artifact_kind=ArtifactKind.ELEMENT_SET, base_artifact_ref=paragraph_ref, ) copied_paragraph_id = (await repository.snapshot(element_context)).paragraphs[0].paragraph_id source_element_id = await repository.create_element( element_context, name="old detail", dimension_primary="实质", dimension_secondary="detail", ) await repository.batch_link(element_context, [(copied_paragraph_id, [source_element_id])]) element_version, element_ref = await repository.freeze( element_context, lineage=_lineage( base_ref=paragraph_ref.uri, base_digest=paragraph_ref.digest, base_revision=paragraph_version.artifact_version_id, ), ) assert len(element_version.artifact.paragraphs) == 1 assert len(element_version.artifact.paragraph_element_links) == 1 replacement_context = _context("element-replacement", task="element-replacement-task") replacement = await repository.get_or_create( replacement_context, artifact_kind=ArtifactKind.ELEMENT_SET, base_artifact_ref=element_ref, ) copied = await repository.snapshot(replacement_context) replacement_paragraph_id = copied.paragraphs[0].paragraph_id replacement_element_id = copied.elements[0].element_id assert replacement_paragraph_id not in {source_paragraph_id, copied_paragraph_id} assert replacement_element_id != source_element_id assert copied.links[0].paragraph_id == replacement_paragraph_id assert copied.links[0].element_id == replacement_element_id await repository.update_element( replacement_context, element_id=replacement_element_id, values={"name": "replaced detail", "is_active": False}, ) new_element_id = await repository.create_element( replacement_context, name="new detail", dimension_primary="实质", dimension_secondary="detail", ) await repository.batch_link(replacement_context, [(replacement_paragraph_id, [new_element_id])]) replacement_version, _ = await repository.freeze( replacement_context, lineage=_lineage( base_ref=element_ref.uri, base_digest=element_ref.digest, base_revision=element_version.artifact_version_id, ), ) manifest = replacement_version.artifact.change_manifest assert manifest["created"]["element_ids"] == [new_element_id] assert manifest["created"]["links"] == [ {"paragraph_id": replacement_paragraph_id, "element_id": new_element_id} ] assert manifest["updated"]["element_ids"] == [replacement_element_id] assert manifest["deactivated"]["element_ids"] == [replacement_element_id] assert manifest["deleted_links"] == [ {"paragraph_id": copied_paragraph_id, "element_id": source_element_id} ] assert replacement.branch_id == replacement_version.artifact_version_id > 0 for operation in ( repository.create_element( replacement_context, name="late", dimension_primary="实质", dimension_secondary="detail", ), repository.update_element( replacement_context, element_id=new_element_id, values={"name": "late"}, ), repository.batch_link(replacement_context, [(replacement_paragraph_id, [new_element_id])]), repository.delete_links( replacement_context, pairs=((replacement_paragraph_id, new_element_id),), ), ): with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"): await operation @pytest.mark.asyncio async def test_run_454_shape_roundtrips_without_branch_zero(database) -> None: """The legacy 5/11/13 detail shape remains expressible without old Branch/Round rows.""" _, sessions = database artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions) repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts) context = _context("run-454-shape", task="compose-454") workspace = await repository.get_or_create( context, artifact_kind=ArtifactKind.STRUCTURED_SCRIPT ) paragraph_ids = [ await repository.create_paragraph( context, paragraph_index=index, name=f"paragraph-{index}", content_range={"topics": [index]}, ) for index in range(1, 6) ] element_ids = [ await repository.create_element( context, name=f"element-{index}", dimension_primary="实质" if index <= 6 else "形式", dimension_secondary=f"dimension-{index}", ) for index in range(1, 12) ] link_pairs = [(paragraph_ids[index % 5], [element_ids[index % 11]]) for index in range(11)] link_pairs.extend( [ (paragraph_ids[0], [element_ids[1]]), (paragraph_ids[1], [element_ids[2]]), ] ) assert await repository.batch_link(context, link_pairs) == 13 version, reference = await repository.freeze( context, lineage=_lineage(), structured_script={ "direction_ref": "script-build://artifact-versions/999", "source_artifact_refs": ["script-build://artifact-versions/888"], "evidence_refs": ["script-build://artifact-versions/777"], "acceptance_notes": ["legacy detail shape preserved"], }, ) loaded = await artifacts.read_by_ref(reference, script_build_id=1) assert loaded == version assert len(loaded.artifact.paragraphs) == 5 assert len(loaded.artifact.elements) == 11 assert len(loaded.artifact.paragraph_element_links) == 13 assert workspace.branch_id == version.artifact_version_id > 0 async with sessions() as session: paragraph_branches = set( ( await session.execute( select(script_build_paragraph.c.branch_id).where( script_build_paragraph.c.script_build_id == 1 ) ) ).scalars() ) element_branches = set( ( await session.execute( select(script_build_element.c.branch_id).where( script_build_element.c.script_build_id == 1 ) ) ).scalars() ) link_branches = set( ( await session.execute( select(script_build_paragraph_element.c.branch_id).where( script_build_paragraph_element.c.script_build_id == 1 ) ) ).scalars() ) assert paragraph_branches == element_branches == link_branches == {workspace.branch_id}