script-build-and-framework-tmp/script_build_host/tests/test_phase_two_workspace.py

from __future__ import annotations

import asyncio
import os
from dataclasses import replace

import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

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.mysql
@pytest.mark.asyncio
async def test_mysql_workspace_writes_only_through_candidate_view() -> None:
    dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_DSN")
    if not dsn:
        pytest.skip("SCRIPT_BUILD_TEST_MYSQL_DSN is not configured")
    engine = create_async_engine(dsn, isolation_level="READ COMMITTED", pool_pre_ping=True)
    sessions = async_sessionmaker(engine, expire_on_commit=False)
    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
    context = _context("mysql-candidate-view", build=920000)
    try:
        workspace = await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH)
        paragraph_id = await repository.create_paragraph(
            context,
            paragraph_index=1,
            name="candidate-only",
            content_range={},
        )
        async with sessions() as session:
            branch = await session.scalar(
                select(script_build_paragraph.c.branch_id).where(
                    script_build_paragraph.c.id == paragraph_id
                )
            )
        assert branch == workspace.branch_id > 0
        assert repository._paragraphs.name == "script_build_candidate_paragraph_v"
    finally:
        await engine.dispose()


@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}