GEWU/tests/application/test_acceptance.py

import hashlib
import json
import subprocess
import sys
import threading
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
from pathlib import Path

import pyotp
import pytest
from cryptography.fernet import Fernet

from gewu.adapters.git_repository import GitCanonicalRepository
from gewu.adapters.sqlite_identity import SQLiteIdentityStore
from gewu.adapters.sqlite_search import SQLiteSearchIndex
from gewu.adapters.text_extractor import TextMarkdownExtractor
from gewu.application.acceptance import (
    AcceptanceError,
    AcceptanceService,
    ActorContext,
)
from gewu.application.context import ValidationContext
from gewu.application.identity import Principal
from gewu.application.reading import CanonicalNodeReader
from gewu.application.search import CanonicalSearchService
from gewu.application.serialization import safe_yaml_record
from gewu.application.uploads import LocalArtifactStore
from gewu.domain.contracts import (
    AcceptanceManifest,
    EdgeEndpoint,
    ExtractionRecord,
    KnowledgeNode,
    ProposalOperation,
    ProposalRevision,
    ReliabilityGrade,
    Scope,
    SearchRequest,
    SourceRevisionRecord,
    TypedEdge,
)
from gewu.domain.validation import assertion_digest, evidence_digest, proposal_payload_digest
from tests.domain.proposal_fixtures import (
    EXTRACTION,
    NODE,
    POLICIES,
    PROJECTS,
    SOURCE_REVISION,
    proposal,
)

ACTOR = ActorContext(principal_id="agent-pilot", credential_id="agent-key-1")
SOURCE_BYTES = b"The installation hosted 400 visitors during the 2026 season."


class SimulatedProcessCrash(BaseException):
    pass


def _service(
    data_root: Path,
    *,
    hook: Callable[[str], None] = lambda _stage: None,
    authorized: bool = True,
) -> tuple[
    AcceptanceService,
    GitCanonicalRepository,
    LocalArtifactStore,
    SourceRevisionRecord,
    ExtractionRecord,
]:
    artifact_store = LocalArtifactStore(data_root / "private-data")
    upload = artifact_store.begin(
        filename="pilot.md",
        media_type="text/markdown",
        byte_length=len(SOURCE_BYTES),
        expected_sha256=hashlib.sha256(SOURCE_BYTES).hexdigest(),
        now=100,
    )
    artifact_store.put_chunk(
        upload.upload_id,
        offset=0,
        content=SOURCE_BYTES,
        chunk_sha256=hashlib.sha256(SOURCE_BYTES).hexdigest(),
        now=101,
    )
    artifact_store.finalize(upload.upload_id, now=102)
    source = SOURCE_REVISION.model_copy(
        update={
            "artifact_sha256": hashlib.sha256(SOURCE_BYTES).hexdigest(),
            "byte_length": len(SOURCE_BYTES),
        }
    )
    extraction = EXTRACTION
    repository = GitCanonicalRepository(data_root / "private-data" / "knowledge.git")
    context = ValidationContext(
        projects=PROJECTS,
        policies=POLICIES,
        sources={},
        extractions={extraction.extraction_id: extraction},
        existing_nodes={},
    )
    service = AcceptanceService(
        repository=repository,
        artifact_store=artifact_store,
        workflow_root=data_root / "private-data" / "staging",
        context_provider=lambda: context,
        actor_is_current=lambda _actor: authorized,
        failure_hook=hook,
    )
    return service, repository, artifact_store, source, extraction


def _proposal(repository: GitCanonicalRepository, source: SourceRevisionRecord) -> ProposalRevision:
    candidate = proposal(source=source, base_commit=repository.head())
    return candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)})


def test_acceptance_commits_records_and_retry_returns_original_receipt(tmp_path: Path) -> None:
    service, repository, _artifact_store, source, _extraction = _service(tmp_path)
    candidate = _proposal(repository, source)
    result = service.submit(candidate, actor=ACTOR)

    assert result.validation.eligible_for_auto_accept
    assert result.receipt is not None
    assert result.receipt.state == "accepted"
    assert result.receipt.commit_id == repository.head()
    assert result.receipt.accepted_node_ids == (NODE.node_id,)
    before_retry = repository.head()

    retry = service.submit(candidate, actor=ACTOR)
    assert retry.receipt == result.receipt
    assert repository.head() == before_retry
    status = service.proposal_status(candidate.proposal_id, candidate.revision, actor=ACTOR)
    assert status.lifecycle == "accepted"
    assert status.canonical_revision == result.receipt.commit_id
    assert status.receipt == result.receipt
    with pytest.raises(AcceptanceError) as accepted_withdrawal:
        service.withdraw_proposal(candidate.proposal_id, candidate.revision, actor=ACTOR)
    assert accepted_withdrawal.value.code == "PROPOSAL_ALREADY_ACCEPTED"

    with pytest.raises(AcceptanceError) as unauthorized:
        service.proposal_status(
            candidate.proposal_id,
            candidate.revision,
            actor=ActorContext(principal_id="agent-other", credential_id="other-key"),
        )
    assert unauthorized.value.code == "PROPOSAL_NOT_FOUND"

    changed_node = NODE.model_copy(update={"title": "Changed after acceptance"})
    changed = proposal(node=changed_node, source=source, base_commit=candidate.base_commit)
    with pytest.raises(AcceptanceError) as error:
        service.submit(changed, actor=ACTOR)
    assert error.value.code == "IDEMPOTENCY_CONFLICT"


def test_second_source_enriches_stable_node_and_preserves_history(tmp_path: Path) -> None:
    service, repository, artifact_store, source, extraction = _service(tmp_path)
    first = service.submit(_proposal(repository, source), actor=ACTOR)
    assert first.receipt is not None

    source_bytes = b"An independent attendance log records 400 visitors over the 2026 season."
    upload = artifact_store.begin(
        filename="independent-attendance.md",
        media_type="text/markdown",
        byte_length=len(source_bytes),
        expected_sha256=hashlib.sha256(source_bytes).hexdigest(),
        now=103,
    )
    artifact_store.put_chunk(
        upload.upload_id,
        offset=0,
        content=source_bytes,
        chunk_sha256=hashlib.sha256(source_bytes).hexdigest(),
        now=104,
    )
    artifact_store.finalize(upload.upload_id, now=105)
    second_source = source.model_copy(
        update={
            "source_id": "shared:independent-attendance",
            "revision_id": "shared:independent-attendance:r1",
            "artifact_sha256": hashlib.sha256(source_bytes).hexdigest(),
            "byte_length": len(source_bytes),
            "title": "Independent seasonal attendance log",
            "authors": ("Independent observer",),
        }
    )
    second_extraction = TextMarkdownExtractor().extract(
        second_source.revision_id, source_bytes, "text/markdown"
    )
    block = second_extraction.blocks[0]
    quote = block.normalized_text
    old_evidence = NODE.claims[0].evidence[0]
    new_evidence = old_evidence.model_copy(
        update={
            "source_revision_id": second_source.revision_id,
            "extraction_id": second_extraction.extraction_id,
            "quote": quote,
            "locator": old_evidence.locator.model_copy(
                update={"block_id": block.block_id, "start": 0, "end": len(quote)}
            ),
        }
    )
    assessment = NODE.claims[0].assessment.model_copy(
        update={
            "assessment_id": "assessment-independent-attendance",
            "evidence_digests": (evidence_digest(old_evidence), evidence_digest(new_evidence)),
            "assessor_id": "agent-independent",
            "run_id": "run-independent",
            "method": "independent source cross-check",
            "rationale": "The independently recorded log reports the same count and season.",
        }
    )
    enriched_claim = NODE.claims[0].model_copy(
        update={"evidence": (old_evidence, new_evidence), "assessment": assessment}
    )
    from tests.domain.proposal_fixtures import NOW

    enriched = NODE.model_copy(
        update={
            "updated": NOW + timedelta(minutes=1),
            "agent_id": "agent-independent",
            "run_id": "run-independent",
            "model_id": "fixture-second-agent",
            "claims": (enriched_claim,),
            "tags": (*NODE.tags, "independent-source-supported"),
        }
    )
    context = ValidationContext(
        projects=PROJECTS,
        policies=POLICIES,
        sources={source.revision_id: source, second_source.revision_id: second_source},
        extractions={
            extraction.extraction_id: extraction,
            second_extraction.extraction_id: second_extraction,
        },
        existing_nodes={NODE.node_id: NODE},
    )
    update_service = AcceptanceService(
        repository=repository,
        artifact_store=artifact_store,
        workflow_root=tmp_path / "private-data" / "staging",
        context_provider=lambda: context,
        actor_is_current=lambda _actor: True,
    )
    proposal_draft = ProposalRevision(
        proposal_id="proposal-second-source-enrichment",
        revision=1,
        initiating_project_id=NODE.project_id,
        scope=Scope(projects=(NODE.project_id,), include_shared=True),
        base_commit=repository.head(),
        payload_digest="0" * 64,
        proposer_id="agent-independent",
        run_id="run-independent",
        operations=(
            ProposalOperation(operation="register_source", source=second_source),
            ProposalOperation(
                operation="update_node", target_id=NODE.node_id, node=enriched
            ),
        ),
    )
    proposal_update = proposal_draft.model_copy(
        update={"payload_digest": proposal_payload_digest(proposal_draft)}
    )

    result = update_service.submit(
        proposal_update,
        actor=ActorContext("agent-independent", "agent-key-2"),
    )

    assert result.receipt is not None, result.validation.findings
    assert result.receipt.accepted_node_ids == (NODE.node_id,)
    node_path = (
        f"projects/{hashlib.sha256(NODE.project_id.encode()).hexdigest()}/nodes/"
        f"{hashlib.sha256(NODE.node_id.encode()).hexdigest()}.yaml"
    )
    old_record = repository.read_file(first.receipt.commit_id, node_path)
    new_record = repository.read_file(result.receipt.commit_id, node_path)
    assert old_record is not None and new_record is not None
    assert safe_yaml_record(old_record, KnowledgeNode) == NODE
    accepted = safe_yaml_record(new_record, KnowledgeNode)
    assert accepted.node_id == NODE.node_id
    assert accepted.claims[0].claim_id == NODE.claims[0].claim_id
    assert len(accepted.claims[0].evidence) == 2
    assert {item.source_revision_id for item in accepted.claims[0].evidence} == {
        source.revision_id,
        second_source.revision_id,
    }
    before_retry = repository.head()
    retry = update_service.submit(
        proposal_update,
        actor=ActorContext("agent-independent", "agent-key-2"),
    )
    assert retry.receipt == result.receipt
    assert repository.head() == before_retry


def test_policy_exception_requires_exact_owner_decision_in_manifest(tmp_path: Path) -> None:
    service, repository, _artifact_store, source, extraction = _service(tmp_path)
    identities = SQLiteIdentityStore(
        tmp_path / "owner-identity.sqlite3", owner_encryption_key=Fernet.generate_key()
    )
    secret = pyotp.random_base32()
    identities.provision_owner(
        principal_id="project-owner",
        password="correct horse battery staple",
        totp_secret=secret,
        project_ids=("dome-performance", "shared"),
    )
    projects = {
        **PROJECTS,
        "dome-performance": PROJECTS["dome-performance"].model_copy(
            update={"owners": ("project-owner",)}
        ),
        "shared": PROJECTS["dome-performance"].model_copy(
            update={
                "project_id": "shared",
                "title": "Shared knowledge",
                "owners": ("project-owner",),
                "autonomy_policy_id": "shared",
            }
        ),
    }
    policies = {
        **POLICIES,
        "dome-policy": POLICIES["dome-policy"].model_copy(
            update={"automatic_node_types": ("finding",)}
        ),
    }
    context = ValidationContext(
        projects=projects,
        policies=policies,
        sources={},
        extractions={extraction.extraction_id: extraction},
        existing_nodes={},
    )
    service.context_provider = lambda: context
    service.owner_challenge_issuer = identities.issue_owner_challenge
    service.owner_decision_lookup = identities.authorized_owner_challenges
    candidate = _proposal(repository, source)

    held = service.submit(candidate, actor=ACTOR)
    assert held.validation.status == "needs-review"
    assert held.receipt is None
    assert len(held.owner_challenge_ids) == 1
    challenge = identities.get_owner_challenge(
        held.owner_challenge_ids[0], principal_id="project-owner"
    )
    assert challenge is not None
    assert challenge["diff_json"]
    owner_session = identities.authenticate_owner(
        principal_id="project-owner",
        password="correct horse battery staple",
        totp_code=pyotp.TOTP(secret).now(),
        challenge_id=held.owner_challenge_ids[0],
        audience="gewu-acceptance-test",
    )
    assert owner_session is not None
    owner_principal = identities.authenticate_owner_session(
        owner_session.session_token,
        challenge_id=held.owner_challenge_ids[0],
        audience="gewu-acceptance-test",
    )
    assert owner_principal is not None
    decision = identities.decide_owner_challenge(
        challenge_id=held.owner_challenge_ids[0],
        principal_id="project-owner",
        proposal_revision=candidate.revision,
        payload_digest=candidate.payload_digest,
        diff_digest=challenge["diff_digest"],
        decision="authorize",
        rationale="Reviewed the exact file diff and approve this policy exception.",
        owner_session_id=owner_principal.credential_id.rpartition(":")[2],
        audience="gewu-acceptance-test",
    )
    assert decision["status"] == "authorized"

    original_context = context
    changed_policies = {
        **policies,
        "dome-policy": policies["dome-policy"].model_copy(update={"policy_version": 2}),
    }
    context = ValidationContext(
        projects=projects,
        policies=changed_policies,
        sources={},
        extractions={extraction.extraction_id: extraction},
        existing_nodes={},
    )
    stale_approval = service.submit(candidate, actor=ACTOR)
    assert stale_approval.receipt is None
    assert stale_approval.validation.status == "needs-review"
    assert repository.head() == candidate.base_commit
    context = original_context

    accepted = service.submit(candidate, actor=ACTOR)
    assert accepted.receipt is not None
    assert accepted.validation.status == "needs-review"
    raw_manifest = repository.read_file(
        accepted.receipt.commit_id,
        service._manifest_path(candidate.proposal_id, candidate.revision),
    )
    assert raw_manifest is not None
    manifest = safe_yaml_record(raw_manifest, AcceptanceManifest)
    assert len(manifest.owner_decisions) == 1
    assert manifest.owner_decisions[0].challenge_id == held.owner_challenge_ids[0]


def test_contradiction_bundle_publishes_both_claims_and_their_typed_edge(
    tmp_path: Path,
) -> None:
    content = (
        b"Field report: The seasonal audience count was 400.\n\n"
        b"Vendor report: The seasonal audience count was 450.\n"
    )
    digest = hashlib.sha256(content).hexdigest()
    artifact_store = LocalArtifactStore(tmp_path / "private-data")
    upload = artifact_store.begin(
        filename="conflicting-reports.md",
        media_type="text/markdown",
        byte_length=len(content),
        expected_sha256=digest,
        now=100,
    )
    artifact_store.put_chunk(
        upload.upload_id,
        offset=0,
        content=content,
        chunk_sha256=digest,
        now=101,
    )
    artifact_store.finalize(upload.upload_id, now=102)
    source = SOURCE_REVISION.model_copy(
        update={"artifact_sha256": digest, "byte_length": len(content)}
    )
    extraction = TextMarkdownExtractor().extract(source.revision_id, content, "text/markdown")

    def supported_node(
        node_id: str, claim_id: str, quote: str, title: str, value: int
    ) -> KnowledgeNode:
        block = next(item for item in extraction.blocks if quote in item.normalized_text)
        start = block.normalized_text.index(quote)
        evidence = (
            NODE.claims[0]
            .evidence[0]
            .model_copy(
                update={
                    "source_revision_id": source.revision_id,
                    "extraction_id": extraction.extraction_id,
                    "assertion_digest": assertion_digest(quote),
                    "quote": quote,
                    "locator": NODE.claims[0]
                    .evidence[0]
                    .locator.model_copy(
                        update={
                            "block_id": block.block_id,
                            "start": start,
                            "end": start + len(quote),
                        }
                    ),
                }
            )
        )
        assessment = NODE.claims[0].assessment.model_copy(
            update={
                "assertion_digest": assertion_digest(quote),
                "evidence_digests": (evidence_digest(evidence),),
            }
        )
        claim = NODE.claims[0].model_copy(
            update={
                "claim_id": claim_id,
                "text": quote,
                "evidence": (evidence,),
                "assessment": assessment,
            }
        )
        return NODE.model_copy(
            update={
                "node_id": node_id,
                "title": title,
                "summary": quote,
                "claims": (claim,),
                "value": value,
            }
        )

    first = supported_node(
        "dome:count-field-report",
        "dome:claim-count-400",
        "The seasonal audience count was 400.",
        "Field count",
        400,
    )
    second = supported_node(
        "dome:count-vendor-report",
        "dome:claim-count-450",
        "The seasonal audience count was 450.",
        "Vendor count",
        450,
    )
    edge = TypedEdge(
        edge_id="dome:counts-conflict",
        edge_type="contradicts",
        source=EdgeEndpoint(kind="node", id=first.node_id),
        target=EdgeEndpoint(kind="node", id=second.node_id),
        conditions="Both reports refer to the same seasonal audience count.",
        interpretation_rationale="The source-backed counts differ for the same reporting period.",
    )
    policies = {
        **POLICIES,
        "dome-policy": POLICIES["dome-policy"].model_copy(
            update={"automatic_edge_types": ("contradicts",)}
        ),
    }
    repository = GitCanonicalRepository(tmp_path / "private-data" / "knowledge.git")
    context = ValidationContext(
        projects=PROJECTS,
        policies=policies,
        sources={},
        extractions={extraction.extraction_id: extraction},
        existing_nodes={},
    )
    service = AcceptanceService(
        repository=repository,
        artifact_store=artifact_store,
        workflow_root=tmp_path / "private-data" / "staging",
        context_provider=lambda: context,
        actor_is_current=lambda _actor: True,
    )
    base = proposal(node=None, source=source, base_commit=repository.head())
    candidate = base.model_copy(
        update={
            "operations": (
                *base.operations,
                ProposalOperation(operation="upsert_node", node=first),
                ProposalOperation(operation="upsert_node", node=second),
                ProposalOperation(operation="add_edge", edge=edge),
            ),
            "payload_digest": "0" * 64,
        }
    )
    candidate = candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)})

    result = service.submit(candidate, actor=ACTOR)

    assert result.validation.status == "eligible", result.validation.findings
    assert result.receipt is not None
    assert result.receipt.accepted_node_ids == (first.node_id, second.node_id)
    for node in (first, second):
        stored = repository.read_file(
            result.receipt.commit_id,
            f"projects/{hashlib.sha256(node.project_id.encode()).hexdigest()}/nodes/"
            f"{hashlib.sha256(node.node_id.encode()).hexdigest()}.yaml",
        )
        assert stored is not None
        assert safe_yaml_record(stored, KnowledgeNode).claims[0].text == node.claims[0].text
    stored_edge = repository.read_file(
        result.receipt.commit_id,
        f"edges/{hashlib.sha256(edge.edge_id.encode()).hexdigest()}.yaml",
    )
    assert stored_edge is not None
    assert safe_yaml_record(stored_edge, TypedEdge) == edge


def test_independent_acceptance_services_exclude_a_concurrent_writer(tmp_path: Path) -> None:
    entered_commit = threading.Event()
    release_commit = threading.Event()

    def hold_before_commit(stage: str) -> None:
        if stage == "before_commit":
            entered_commit.set()
            assert release_commit.wait(timeout=5)

    service, repository, artifact_store, source, _extraction = _service(
        tmp_path, hook=hold_before_commit
    )
    second_context = ValidationContext(
        projects=PROJECTS,
        policies=POLICIES,
        sources={},
        extractions={EXTRACTION.extraction_id: EXTRACTION},
        existing_nodes={},
    )
    second_service = AcceptanceService(
        repository=repository,
        artifact_store=artifact_store,
        workflow_root=tmp_path / "private-data" / "staging",
        context_provider=lambda: second_context,
        actor_is_current=lambda _actor: True,
    )
    first = _proposal(repository, source).model_copy(update={"proposal_id": "proposal-writer-a"})
    second = _proposal(repository, source).model_copy(update={"proposal_id": "proposal-writer-b"})

    with ThreadPoolExecutor(max_workers=1) as workers:
        pending = workers.submit(service.submit, first, actor=ACTOR)
        try:
            assert entered_commit.wait(timeout=5)
            with pytest.raises(AcceptanceError) as busy:
                second_service.submit(second, actor=ACTOR)
            assert busy.value.code == "BUSY"
        finally:
            release_commit.set()
        accepted = pending.result(timeout=5)

    assert accepted.receipt is not None
    assert repository.head() == accepted.receipt.commit_id
    assert second_service.receipt(second.proposal_id, second.revision) is None


def test_acceptance_publishes_and_reads_evidence_backed_assumptions(tmp_path: Path) -> None:
    service, repository, _artifact_store, source, extraction = _service(tmp_path)
    original_claim = NODE.claims[0]
    statement = "The installation's seasonal count is suitable for comparison."
    statement_digest = assertion_digest(statement)
    evidence = original_claim.evidence[0].model_copy(
        update={
            "assertion_path": "assumptions[0].text",
            "assertion_digest": statement_digest,
        }
    )
    claim = original_claim.model_copy(
        update={
            "claim_id": "dome:assumption-seasonal-comparison",
            "text": statement,
            "evidence": (evidence,),
            "assessment": original_claim.assessment.model_copy(
                update={
                    "assertion_digest": statement_digest,
                    "evidence_digests": (evidence_digest(evidence),),
                }
            ),
        }
    )
    node = NODE.model_copy(update={"assumptions": (claim,)})
    candidate = proposal(node=node, source=source, base_commit=repository.head())
    candidate = candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)})

    result = service.submit(
        candidate,
        actor=ACTOR,
        staged_extractions={extraction.extraction_id: extraction},
    )

    assert result.receipt is not None
    assert (
        repository.read_file(
            result.receipt.commit_id, f"extractions/{extraction.extraction_id}.json"
        )
        is not None
    )
    reader = CanonicalNodeReader(repository)
    principal = Principal(
        "reader", "reader-key", frozenset({"viewer"}), frozenset({"dome-performance"})
    )
    read = reader.get_node(node.project_id, node.node_id, principal=principal)
    stored = read.data[0]
    assert stored["assumptions"][0]["text"] == statement
    assert any(item.assertion_digest == statement_digest for item in read.citations)


def test_reference_only_source_registers_without_an_artifact(tmp_path: Path) -> None:
    service, repository, _artifacts, _source, _extraction = _service(tmp_path)
    reference = SourceRevisionRecord(
        source_id="dome:missing-paper",
        revision_id="dome:missing-paper:r1",
        media_type="application/octet-stream",
        byte_length=0,
        captured_at=SOURCE_REVISION.captured_at,
        original_url="https://research.example/missing-paper",
        title="Paper whose archive is unavailable",
        license="unspecified",
        source_type="paper",
        reliability_grade=ReliabilityGrade.MEDIUM,
        ingested_for=("dome-performance",),
        archive_status="capture-failed",
    )
    candidate = ProposalRevision(
        proposal_id="proposal-reference-only",
        revision=1,
        initiating_project_id="dome-performance",
        scope=proposal().scope,
        base_commit=repository.head(),
        payload_digest="0" * 64,
        proposer_id=ACTOR.principal_id,
        run_id="run-reference-only",
        operations=(ProposalOperation(operation="register_source", source=reference),),
    )
    candidate = candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)})

    result = service.submit(candidate, actor=ACTOR)

    assert result.receipt is not None
    assert result.receipt.state == "accepted"
    paths = repository.list_files(repository.head(), "sources")
    stored = next(repository.read_file(repository.head(), path) for path in paths)
    assert stored is not None
    assert b"capture-failed" in stored
    assert b"artifact_sha256: null" in stored


@pytest.mark.parametrize("crash_stage", ("after_ref_update", "before_receipt"))
def test_ref_update_before_receipt_is_reconciled_without_republishing(
    tmp_path: Path, crash_stage: str
) -> None:
    def fail_after_ref_update(stage: str) -> None:
        if stage == crash_stage:
            raise SimulatedProcessCrash

    service, repository, _store, source, _extraction = _service(
        tmp_path, hook=fail_after_ref_update
    )
    candidate = _proposal(repository, source)
    with pytest.raises(SimulatedProcessCrash):
        service.submit(candidate, actor=ACTOR)
    accepted_head = repository.head()

    recovered, _same_repo, _same_store, _same_source, _same_extraction = (
        _service_without_new_artifact(service, repository, tmp_path, source)
    )
    recovered.recover()
    receipt = recovered.receipt(candidate.proposal_id, candidate.revision)
    assert receipt is not None
    assert receipt.commit_id == accepted_head
    assert repository.head() == accepted_head
    status = recovered.proposal_status(candidate.proposal_id, candidate.revision, actor=ACTOR)
    assert status.lifecycle == "accepted"
    assert status.receipt == receipt
    retry = recovered.submit(candidate, actor=ACTOR)
    assert retry.receipt == receipt


@pytest.mark.parametrize("crash_stage", ("after_ref_update", "before_receipt"))
def test_os_process_termination_recovers_publication_after_restart(
    tmp_path: Path, crash_stage: str
) -> None:
    service, _repository, _store, source, _extraction = _service(tmp_path)
    candidate = _proposal(service.repository, source)
    proposal_path = tmp_path / "candidate.json"
    proposal_path.write_bytes(candidate.model_dump_json().encode("utf-8"))

    crashed = subprocess.run(
        [
            sys.executable,
            "-m",
            "tests.application.crash_acceptance_worker",
            str(tmp_path),
            str(proposal_path),
            crash_stage,
        ],
        cwd=Path(__file__).parents[2],
        capture_output=True,
        check=False,
        timeout=60,
    )
    assert crashed.returncode == 73, crashed.stderr.decode("utf-8", errors="replace")

    # Construct fresh adapters as a new service process would, then recover from durable state.
    recovered, repository, _store, _source, _extraction = _service(tmp_path)
    committed_head = repository.head()
    recovered.recover()
    receipt = recovered.receipt(candidate.proposal_id, candidate.revision)
    assert receipt is not None
    assert receipt.commit_id == committed_head
    status = recovered.proposal_status(candidate.proposal_id, candidate.revision, actor=ACTOR)
    assert status.lifecycle == "accepted"
    assert status.receipt == receipt

    # The derived index was absent at the crash. Its first read after recovery must
    # rebuild from the committed revision and retain the accepted claim's citation.
    index = SQLiteSearchIndex(tmp_path / "private-data" / "search.sqlite3", repository)
    assert index.diagnostic_status()[0] is None
    reader = CanonicalSearchService(
        index=index,
        project_ids_provider=lambda: PROJECTS.keys(),
        canonical_head=repository.head,
    )
    search_result = reader.search(
        SearchRequest(
            query="visitors",
            scope=Scope(projects=("dome-performance",), include_shared=False),
            budget_tokens=5000,
        ),
        principal=Principal(
            "recovery-reader",
            "recovery-reader-key",
            frozenset({"viewer"}),
            frozenset({"dome-performance"}),
        ),
    )
    assert search_result.data
    indexed_node = search_result.data[0]["node"]
    assert isinstance(indexed_node, dict)
    assert indexed_node["node_id"] == NODE.node_id
    assert search_result.citations

    retry = recovered.submit(candidate, actor=ACTOR)
    assert retry.receipt == receipt
    assert repository.head() == committed_head


def test_prepublication_crash_leaves_proposal_staged_and_retryable(tmp_path: Path) -> None:
    def fail_before_commit(stage: str) -> None:
        if stage == "before_commit":
            raise SimulatedProcessCrash

    service, repository, _store, source, _extraction = _service(tmp_path, hook=fail_before_commit)
    candidate = _proposal(repository, source)
    base = repository.head()
    with pytest.raises(SimulatedProcessCrash):
        service.submit(candidate, actor=ACTOR)
    assert repository.head() == base

    recovered, *_ = _service_without_new_artifact(service, repository, tmp_path, source)
    recovered.recover()
    status = recovered.proposal_status(candidate.proposal_id, candidate.revision, actor=ACTOR)
    assert status.lifecycle == "staged"
    retry = recovered.submit(candidate, actor=ACTOR)
    assert retry.receipt is not None
    assert repository.head() == retry.receipt.commit_id


def test_unrelated_submission_waits_until_pending_publication_is_reconciled(
    tmp_path: Path,
) -> None:
    crash_pending_writer = True

    def fail_after_ref_update(stage: str) -> None:
        nonlocal crash_pending_writer
        if stage == "after_ref_update" and crash_pending_writer:
            crash_pending_writer = False
            raise SimulatedProcessCrash

    service, repository, _store, source, _extraction = _service(
        tmp_path, hook=fail_after_ref_update
    )
    pending = _proposal(repository, source)
    with pytest.raises(SimulatedProcessCrash):
        service.submit(pending, actor=ACTOR)
    committed_head = repository.head()

    journal_path = sorted((tmp_path / "private-data/staging/journal").glob("*/*.json"))[-1]
    journal_entry = json.loads(journal_path.read_text(encoding="utf-8"))
    expected_manifest_hash = journal_entry["manifest_sha256"]
    journal_entry["manifest_sha256"] = "0" * 64
    journal_path.write_text(json.dumps(journal_entry), encoding="utf-8")

    followup_node = NODE.model_copy(
        update={"node_id": "dome:follow-up-finding", "title": "Follow-up visitor finding"}
    )
    followup = proposal(
        node=followup_node,
        source=source,
        base_commit=committed_head,
    ).model_copy(update={"proposal_id": "proposal-pilot-2"})
    followup = followup.model_copy(update={"payload_digest": proposal_payload_digest(followup)})

    with pytest.raises(AcceptanceError) as blocked:
        service.submit(followup, actor=ACTOR)
    assert blocked.value.code == "RECOVERY_REQUIRED"
    assert repository.head() == committed_head
    assert not service._proposal_path(followup.proposal_id, followup.revision).exists()

    journal_entry["manifest_sha256"] = expected_manifest_hash
    journal_path.write_text(json.dumps(journal_entry), encoding="utf-8")
    service.recover()
    pending_receipt = service.receipt(pending.proposal_id, pending.revision)
    assert pending_receipt is not None
    assert pending_receipt.commit_id == committed_head

    completed = service.submit(followup, actor=ACTOR)
    assert completed.receipt is not None
    assert completed.receipt.commit_id == repository.head()
    assert repository.head() != committed_head


def test_proposal_withdrawal_is_idempotent_and_prevents_later_submission(tmp_path: Path) -> None:
    service, repository, _store, source, _extraction = _service(tmp_path)
    initial_head = repository.head()
    candidate = _proposal(repository, source).model_copy(update={"base_commit": "f" * 40})
    candidate = candidate.model_copy(update={"payload_digest": "0" * 64})
    candidate = candidate.model_copy(update={"payload_digest": proposal_payload_digest(candidate)})
    evaluated = service.submit(candidate, actor=ACTOR)
    assert evaluated.validation.status == "conflict"
    assert evaluated.receipt is None

    withdrawn = service.withdraw_proposal(candidate.proposal_id, candidate.revision, actor=ACTOR)
    assert withdrawn.lifecycle == "withdrawn"
    assert (
        service.withdraw_proposal(candidate.proposal_id, candidate.revision, actor=ACTOR)
        == withdrawn
    )
    assert (
        service.proposal_status(candidate.proposal_id, candidate.revision, actor=ACTOR) == withdrawn
    )
    with pytest.raises(AcceptanceError) as error:
        service.submit(candidate, actor=ACTOR)
    assert error.value.code == "PROPOSAL_WITHDRAWN"
    assert repository.head() == initial_head


def _service_without_new_artifact(
    old_service: AcceptanceService,
    repository: GitCanonicalRepository,
    data_root: Path,
    source: SourceRevisionRecord,
) -> tuple[
    AcceptanceService,
    GitCanonicalRepository,
    LocalArtifactStore,
    SourceRevisionRecord,
    ExtractionRecord,
]:
    from tests.domain.proposal_fixtures import EXTRACTION

    store = old_service.artifact_store
    context = ValidationContext(
        projects=PROJECTS,
        policies=POLICIES,
        sources={},
        extractions={EXTRACTION.extraction_id: EXTRACTION},
        existing_nodes={},
    )
    service = AcceptanceService(
        repository=repository,
        artifact_store=store,
        workflow_root=data_root / "private-data" / "staging",
        context_provider=lambda: context,
        actor_is_current=lambda _actor: True,
    )
    return service, repository, store, source, EXTRACTION


def test_changed_authority_blocks_publication(tmp_path: Path) -> None:
    service, repository, _store, source, _extraction = _service(tmp_path, authorized=False)
    candidate = _proposal(repository, source)
    with pytest.raises(AcceptanceError) as error:
        service.submit(candidate, actor=ACTOR)
    assert error.value.code == "UNAUTHORIZED"
    assert repository.head() == candidate.base_commit


def test_revoked_key_during_validation_is_rechecked_before_publication(tmp_path: Path) -> None:
    service, repository, _store, source, _extraction = _service(tmp_path)
    identities = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
    token = identities.provision(
        credential_id="agent-key-1",
        principal_id="agent-pilot",
        roles=("agent",),
        project_ids=("dome-performance", "shared"),
    )
    principal = identities.authenticate(token)
    assert principal is not None
    actor = ActorContext(
        principal_id=principal.principal_id,
        credential_id=principal.credential_id,
        roles=principal.roles,
        project_ids=principal.project_ids,
    )
    context = ValidationContext(
        projects=PROJECTS,
        policies=POLICIES,
        sources={},
        extractions={_extraction.extraction_id: _extraction},
        existing_nodes={},
    )
    context_reads = 0

    def context_provider() -> ValidationContext:
        nonlocal context_reads
        context_reads += 1
        if context_reads == 2:
            assert identities.revoke("agent-key-1")
        return context

    service.context_provider = context_provider
    service.actor_is_current = lambda candidate: identities.is_current(
        candidate.principal_id,
        candidate.credential_id,
        roles=candidate.roles,
        project_ids=candidate.project_ids,
    )
    candidate = _proposal(repository, source)

    with pytest.raises(AcceptanceError) as error:
        service.submit(candidate, actor=actor)
    assert error.value.code == "POLICY_CHANGED"
    assert repository.head() == candidate.base_commit
    status = service.proposal_status(candidate.proposal_id, candidate.revision, actor=actor)
    assert status.lifecycle == "evaluated"
    assert status.receipt is None