Blame for src/collection_v2.py

"""Durable, append-only collection API for crowdsourced research events.

The module intentionally exposes only write endpoints plus aggregate health/stats.
Raw records may contain user-provided text and stay quarantined in SQLite until a
separate moderation/export process makes an explicit release decision.
"""

from __future__ import annotations

from src.raw_audio_schema import SCHEMA_SQL as RAW_AUDIO_SCHEMA_SQL, erase_participant_audio

import hashlib
import hmac
import json
import math
import os
import re
import sqlite3
import tempfile
import threading
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Annotated, Any, Literal

from fastapi import APIRouter, Body, Header, HTTPException, Response, status
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

from src.consent_lineage import (
    IEDID_DATASET_LOCK,
    VERIFIED_CONSENT_PROVENANCE,
    VERIFIED_CONSENT_STATUS,
    canonical_json_list,
    consent_lineage_is_verified,
)
from src.consent_policy import (
    get_current_consent_policy,
    validate_consent_policy_payload,
)


SCHEMA_VERSION = 3
# Existing append-only SQLite tables use storage schema 2 constraints. New
# records keep that physical row version while recording/exporting receipt
# schema 3 explicitly, avoiding an unsafe rewrite of historical event tables.
STORAGE_SCHEMA_VERSION = 2
DB_PATH_ENV = "COLLECTION_DB_PATH"
COMPAT_DB_PATH_ENVS = ("COLLECTION_V2_DB_PATH", "PUREPOLYGLOT_COLLECTION_DB")
ALLOW_EPHEMERAL_DB_ENV = "ALLOW_EPHEMERAL_COLLECTION"
IDEMPOTENCY_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$"
PARTICIPANT_PATTERN = re.compile(r"^participant_[0-9a-f]{32}$")
WALLET_PATTERN = re.compile(r"^0x[0-9a-fA-F]{40,64}$")
PARTICIPANT_TOKEN_PATTERN = r"^[A-Za-z0-9_-]{43,128}$"
SUBMISSION_METADATA_KEYS = frozenset(
    {
        "tone",
        "context",
        "pragmatics",
        "source_tag",
        "edit_ratios",
        "transcript_edit_ratio",
        "meaning_edit_ratio",
        "audio_sanitation",
        "voice_privacy_mode",
        "room_type",
        "title",
        "mission_subtext",
        "chat_mode",
        "output_semantics",
        "mode",
        "clarification_source",
    }
)
SENSITIVE_METADATA_VALUE_PATTERNS = (
    re.compile(r"0x[0-9a-fA-F]{40,64}"),
    re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"),
    re.compile(r"(?<!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d)"),
    re.compile(r"(?<![A-Za-z0-9_-])[A-Za-z0-9_-]{43,128}(?![A-Za-z0-9_-])"),
)

router = APIRouter(prefix="/api/v2", tags=["crowdsourcing-v2"])

# Shared with backend/api.py so old and v2 writers cannot race during the
# transition period while Pending Audit still consumes pending_approvals.csv.
PENDING_QUEUE_LOCK = threading.Lock()

# Withdrawal CSVs retain only random server locators and the minimum receipt/
# propagation evidence needed to prove that a release was revoked. Every other
# column is contribution or provenance data and is scrubbed fail-closed, which
# also protects future columns that are not yet known to this module.
_TOMBSTONE_AUDIT_COLUMNS = frozenset(
    {
        "Queue_ID",
        "Submission_ID",
        "Consent_ID",
        "Schema_Version",
        "Consent_Status",
        "Consent_Provenance",
        "Consent_Recorded_At",
        "Consent_Text_SHA256",
        "Consent_Record_SHA256",
        "Consent_Schema_Version",
        "Submission_Status",
        "Review_Status",
        "Release_Eligible",
        "Ethics_Review_Status",
        "Withdrawal_Request_ID",
        "Withdrawal_Requested_At",
        "Deletion_Propagation_Status",
        "Audio_Deletion_Status",
    }
)
_TOMBSTONE_TEXT_COLUMNS = frozenset(
    {
        "Utterance",
        "Clarification",
        "Tone",
        "Tone_Category",
        "Context",
        "Linguistic_Context",
        "Pragmatic_Analysis",
        "Syntax_Pattern",
        "Acoustic_Profile",
        "Prompt_Text",
        "Preceding_Context",
        "Source_Text",
        "Machine_Transcript_Initial",
        "User_Transcript_Final",
        "Machine_Translation_Initial",
        "Machine_Output_Initial",
        "User_Translation_Final",
        "User_Output_Final",
        "Metadata_JSON",
    }
)


def _scrub_tombstoned_csv_rows(frame: Any, mask: Any, marker: str) -> None:
    """Scrub every non-audit column for selected contribution rows."""
    for column in frame.columns:
        if column in _TOMBSTONE_AUDIT_COLUMNS:
            continue
        frame.loc[mask, column] = (
            marker if column in _TOMBSTONE_TEXT_COLUMNS else ""
        )

IdempotencyKey = Annotated[
    str,
    Header(
        alias="Idempotency-Key",
        min_length=8,
        max_length=128,
        pattern=IDEMPOTENCY_PATTERN,
        description="Stable client-generated key for safe retries.",
    ),
]

ParticipantToken = Annotated[
    str,
    Header(
        alias="X-Participant-Token",
        min_length=43,
        max_length=128,
        pattern=PARTICIPANT_TOKEN_PATTERN,
        description="Client-held secret that authenticates a pseudonymous participant.",
    ),
]


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace(
        "+00:00", "Z"
    )


def _new_id(prefix: str) -> str:
    return f"{prefix}_{uuid.uuid4().hex}"


def _canonical_json(value: Any) -> str:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )


def _request_hash(value: Any) -> str:
    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()


def _validate_participant_id(value: str) -> str:
    value = value.strip()
    if not PARTICIPANT_PATTERN.fullmatch(value):
        raise ValueError(
            "participant_id must be participant_ followed by 32 lowercase hex characters"
        )
    if WALLET_PATTERN.fullmatch(value) or "@" in value:
        raise ValueError(
            "participant_id must be a revocable pseudonym, not a wallet or email"
        )
    return value


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)


class ConsentCreate(StrictModel):
    participant_id: str = Field(min_length=8, max_length=128)
    withdrawal_token_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    app_id: str = Field(min_length=2, max_length=64)
    app_version: str = Field(min_length=1, max_length=64)
    consent_version: str = Field(min_length=1, max_length=128)
    privacy_notice_version: str = Field(min_length=1, max_length=128)
    dataset_license: str = Field(min_length=1, max_length=128)
    permitted_modalities: list[
        Literal["text", "speech", "image", "multimodal", "judgment", "telemetry"]
    ] = Field(min_length=1, max_length=6)
    consent_surface: str = Field(min_length=2, max_length=128)
    consent_locale: str = Field(
        min_length=2,
        max_length=35,
        pattern=r"^[A-Za-z]{2,3}(?:-[A-Za-z0-9]{1,8})*$",
    )
    consent_text_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    research_purposes: list[str] = Field(min_length=1, max_length=12)
    retention_policy_version: str = Field(min_length=1, max_length=128)
    withdrawal_policy_version: str = Field(min_length=1, max_length=128)
    eligibility_confirmed: Literal[True]
    public_release_confirmed: Literal[True]
    commercial_ai_reuse_confirmed: Literal[True]
    privacy_risks_acknowledged: Literal[True]
    operational_processing_acknowledged: Literal[True]
    voluntary_participation_confirmed: Literal[True]
    contribution_rights_confirmed: Literal[True]
    raw_audio_consent: bool = False
    data_controller_name: str = Field(min_length=2, max_length=200)
    data_controller_contact: str = Field(min_length=3, max_length=256)
    data_recipient_categories: list[str] = Field(min_length=1, max_length=12)
    consent_granted: Literal[True]

    _participant_is_pseudonymous = field_validator("participant_id")(
        _validate_participant_id
    )

    @field_validator("permitted_modalities")
    @classmethod
    def normalize_modalities(cls, value: list[str]) -> list[str]:
        return sorted(set(value))

    @field_validator("research_purposes", "data_recipient_categories")
    @classmethod
    def normalize_policy_lists(cls, value: list[str]) -> list[str]:
        normalized = sorted({item.strip() for item in value if item.strip()})
        if not normalized:
            raise ValueError("policy list must contain at least one non-empty value")
        if any(len(item) > 128 for item in normalized):
            raise ValueError("policy list values must not exceed 128 characters")
        return normalized

    @model_validator(mode="after")
    def policy_is_registered(self) -> "ConsentCreate":
        validate_consent_policy_payload(self.model_dump(mode="json"))
        return self


class SubmissionCreate(StrictModel):
    participant_id: str = Field(min_length=8, max_length=128)
    consent_id: str = Field(pattern=r"^cns_[0-9a-f]{32}$")
    app_id: str = Field(min_length=2, max_length=64)
    app_version: str = Field(min_length=1, max_length=64)
    game_id: str = Field(min_length=2, max_length=64)
    task_id: str = Field(min_length=1, max_length=128)
    prompt_text: str | None = Field(default=None, max_length=20_000)
    topic_id: str | None = Field(default=None, max_length=128)
    image_asset_id: str | None = Field(default=None, max_length=256)
    conversation_id: str | None = Field(default=None, max_length=128)
    turn_id: str | None = Field(default=None, max_length=128)
    preceding_context: str | None = Field(default=None, max_length=40_000)
    source_text: str | None = Field(default=None, max_length=20_000)
    source_language: str = Field(min_length=1, max_length=128)
    source_dialect: str = Field(min_length=1, max_length=256)
    target_language: str = Field(min_length=1, max_length=128)
    target_dialect: str = Field(min_length=1, max_length=256)
    input_modality: Literal["text", "speech", "image", "multimodal"]
    machine_transcript_initial: str | None = Field(default=None, max_length=20_000)
    user_transcript_final: str | None = Field(default=None, max_length=20_000)
    machine_output_initial: str | None = Field(default=None, max_length=20_000)
    user_output_final: str | None = Field(default=None, max_length=20_000)
    model_name: str | None = Field(default=None, max_length=128)
    model_version: str | None = Field(default=None, max_length=128)
    asr_model: str | None = Field(default=None, max_length=128)
    translation_route: str | None = Field(default=None, max_length=128)
    metadata: dict[str, Any] = Field(default_factory=dict)
    started_at: datetime | None = None
    duration_ms: int | None = Field(default=None, ge=0, le=86_400_000)

    _participant_is_pseudonymous = field_validator("participant_id")(
        _validate_participant_id
    )

    @field_validator("metadata")
    @classmethod
    def metadata_is_minimized(cls, value: dict[str, Any]) -> dict[str, Any]:
        unknown = sorted(set(value) - SUBMISSION_METADATA_KEYS)
        if unknown:
            raise ValueError(
                "metadata contains non-allowlisted keys: " + ", ".join(unknown)
            )
        normalized: dict[str, Any] = {}
        for key, item in value.items():
            if item is None or isinstance(item, bool):
                normalized[key] = item
                continue
            if isinstance(item, (int, float)) and not isinstance(item, bool):
                if not math.isfinite(float(item)):
                    raise ValueError(f"metadata.{key} must be finite")
                normalized[key] = item
                continue
            if not isinstance(item, str):
                raise ValueError(f"metadata.{key} must be a bounded scalar")
            if len(item) > 2048:
                raise ValueError(f"metadata.{key} exceeds 2048 characters")
            if any(pattern.search(item) for pattern in SENSITIVE_METADATA_VALUE_PATTERNS):
                raise ValueError(
                    f"metadata.{key} appears to contain a private identifier or token"
                )
            normalized[key] = item
        return normalized

    @field_validator("started_at")
    @classmethod
    def started_at_has_timezone(cls, value: datetime | None) -> datetime | None:
        if value is not None and value.tzinfo is None:
            raise ValueError("started_at must include a timezone")
        return value

    @field_validator("metadata")
    @classmethod
    def metadata_is_bounded(cls, value: dict[str, Any]) -> dict[str, Any]:
        try:
            serialized = _canonical_json(value).encode("utf-8")
        except (TypeError, ValueError) as exc:
            raise ValueError("metadata must be JSON serializable") from exc
        if len(serialized) > 16 * 1024:
            raise ValueError("metadata must not exceed 16 KiB when serialized")
        return value

    @model_validator(mode="after")
    def has_human_contribution(self) -> "SubmissionCreate":
        if not any(
            value and value.strip()
            for value in (
                self.source_text,
                self.user_transcript_final,
                self.user_output_final,
            )
        ):
            raise ValueError("at least one human-provided text field is required")
        if self.input_modality == "image" and not self.image_asset_id:
            raise ValueError("image_asset_id is required for image submissions")
        if self.turn_id and not self.conversation_id:
            raise ValueError("conversation_id is required when turn_id is provided")
        return self


class JudgmentCreate(StrictModel):
    participant_id: str = Field(min_length=8, max_length=128)
    consent_id: str = Field(pattern=r"^cns_[0-9a-f]{32}$")
    submission_id: str = Field(pattern=r"^sub_[0-9a-f]{32}$")
    task_id: str = Field(min_length=1, max_length=128)
    choice: Literal[
        "candidate_a",
        "candidate_b",
        "tie",
        "both_good",
        "both_bad",
        "accept",
        "edit",
        "challenge",
        "skip",
    ]
    error_span: str | None = Field(default=None, max_length=4_000)
    error_type: Literal[
        "meaning",
        "omission",
        "fluency",
        "tone",
        "register",
        "dialect",
        "safety",
        "other",
    ] | None = None
    correction: str | None = Field(default=None, max_length=20_000)
    confidence: float | None = Field(default=None, ge=0.0, le=1.0)
    latency_ms: int | None = Field(default=None, ge=0, le=86_400_000)

    _participant_is_pseudonymous = field_validator("participant_id")(
        _validate_participant_id
    )

    @model_validator(mode="after")
    def correction_matches_choice(self) -> "JudgmentCreate":
        if self.choice == "edit" and not self.correction:
            raise ValueError("correction is required when choice is edit")
        return self


class DeletionRequestCreate(StrictModel):
    reason: Literal[
        "participant_request", "consent_withdrawal", "privacy_concern", "other"
    ] = "participant_request"
    scope: Literal["all"] = "all"


class ConsentReceipt(StrictModel):
    consent_id: str
    recorded_at: str
    consent_record_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    consent_text_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    receipt_schema_version: int
    consent_version: str
    permitted_modalities: list[
        Literal["text", "speech", "image", "multimodal", "judgment", "telemetry"]
    ]
    consent_status: Literal["verified_v2"]
    consent_provenance: Literal["collection_v2_server_receipt"]
    idempotent_replay: bool


class SubmissionReceipt(StrictModel):
    submission_id: str
    submitted_at: str
    moderation_status: Literal["pending"]
    review_status: Literal["pending"]
    release_eligible: Literal[False]
    pending_queue_id: str
    pending_audit_status: Literal[
        "mirrored", "retry_pending", "withdrawn_not_mirrored"
    ]
    idempotent_replay: bool


class JudgmentReceipt(StrictModel):
    judgment_id: str
    recorded_at: str
    idempotent_replay: bool


class DeletionReceipt(StrictModel):
    deletion_request_id: str
    requested_at: str
    status: Literal["pending"]
    tombstoned: Literal[True]
    hard_deleted: Literal[False]
    raw_audio_active_payloads: Literal["erased"] = "erased"
    raw_audio_backup_erasure: Literal["not_attested"] = "not_attested"
    local_pending_audit_status: Literal["tombstoned", "retry_pending"]
    local_iedid_status: Literal["tombstoned", "retry_pending"]
    remote_export_status: Literal["propagation_pending"]
    private_pending_mirror_status: Literal["propagation_pending"] = (
        "propagation_pending"
    )
    propagation_pending: Literal[True]
    idempotent_replay: bool


class IdempotencyConflict(Exception):
    pass


class ParticipantTombstoned(Exception):
    pass


class ConsentError(Exception):
    pass


class StorageConfigurationError(Exception):
    pass


class MissingResource(Exception):
    pass


SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS schema_meta (
    schema_version INTEGER PRIMARY KEY,
    installed_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS consents (
    consent_id TEXT PRIMARY KEY,
    schema_version INTEGER NOT NULL CHECK (schema_version = 2),
    receipt_schema_version INTEGER NOT NULL CHECK (receipt_schema_version = 3),
    participant_id TEXT NOT NULL,
    app_id TEXT,
    app_version TEXT,
    consent_version TEXT NOT NULL,
    privacy_notice_version TEXT NOT NULL,
    dataset_license TEXT NOT NULL,
    permitted_modalities_json TEXT NOT NULL,
    consent_surface TEXT NOT NULL,
    consent_locale TEXT NOT NULL,
    consent_text_sha256 TEXT NOT NULL,
    research_purposes_json TEXT NOT NULL,
    retention_policy_version TEXT NOT NULL,
    withdrawal_policy_version TEXT NOT NULL,
    eligibility_confirmed INTEGER NOT NULL CHECK (eligibility_confirmed = 1),
    public_release_confirmed INTEGER NOT NULL CHECK (public_release_confirmed = 1),
    commercial_ai_reuse_confirmed INTEGER NOT NULL CHECK (commercial_ai_reuse_confirmed = 1),
    privacy_risks_acknowledged INTEGER NOT NULL CHECK (privacy_risks_acknowledged = 1),
    operational_processing_acknowledged INTEGER NOT NULL CHECK (operational_processing_acknowledged = 1),
    voluntary_participation_confirmed INTEGER NOT NULL CHECK (voluntary_participation_confirmed = 1),
    contribution_rights_confirmed INTEGER NOT NULL CHECK (contribution_rights_confirmed = 1),
    raw_audio_consent INTEGER NOT NULL CHECK (raw_audio_consent IN (0, 1)),
    data_controller_name TEXT NOT NULL,
    data_controller_contact TEXT NOT NULL,
    data_recipient_categories_json TEXT NOT NULL,
    withdrawal_token_sha256 TEXT NOT NULL,
    consent_status TEXT NOT NULL CHECK (consent_status = 'verified_v2'),
    consent_provenance TEXT NOT NULL CHECK (
        consent_provenance = 'collection_v2_server_receipt'
    ),
    consent_granted INTEGER NOT NULL CHECK (consent_granted = 1),
    recorded_at TEXT NOT NULL,
    request_sha256 TEXT NOT NULL,
    record_sha256 TEXT NOT NULL,
    record_json TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS submissions (
    submission_id TEXT PRIMARY KEY,
    schema_version INTEGER NOT NULL CHECK (schema_version = 2),
    participant_id TEXT NOT NULL,
    consent_id TEXT NOT NULL REFERENCES consents(consent_id),
    app_id TEXT NOT NULL,
    app_version TEXT NOT NULL,
    game_id TEXT NOT NULL,
    task_id TEXT NOT NULL,
    prompt_text TEXT,
    topic_id TEXT,
    image_asset_id TEXT,
    conversation_id TEXT,
    turn_id TEXT,
    preceding_context TEXT,
    source_text TEXT,
    source_language TEXT NOT NULL,
    source_dialect TEXT NOT NULL,
    target_language TEXT NOT NULL,
    target_dialect TEXT NOT NULL,
    input_modality TEXT NOT NULL,
    machine_transcript_initial TEXT,
    user_transcript_final TEXT,
    machine_output_initial TEXT,
    user_output_final TEXT,
    model_name TEXT,
    model_version TEXT,
    asr_model TEXT,
    translation_route TEXT,
    metadata_json TEXT NOT NULL,
    started_at TEXT,
    submitted_at TEXT NOT NULL,
    duration_ms INTEGER,
    moderation_status TEXT NOT NULL CHECK (moderation_status = 'pending'),
    review_status TEXT NOT NULL CHECK (review_status = 'pending'),
    release_eligible INTEGER NOT NULL CHECK (release_eligible = 0),
    request_sha256 TEXT NOT NULL,
    record_json TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS judgments (
    judgment_id TEXT PRIMARY KEY,
    schema_version INTEGER NOT NULL CHECK (schema_version = 2),
    participant_id TEXT NOT NULL,
    consent_id TEXT NOT NULL REFERENCES consents(consent_id),
    submission_id TEXT NOT NULL REFERENCES submissions(submission_id),
    task_id TEXT NOT NULL,
    choice TEXT NOT NULL,
    error_span TEXT,
    error_type TEXT,
    correction TEXT,
    confidence REAL,
    latency_ms INTEGER,
    gold_item INTEGER NOT NULL DEFAULT 0 CHECK (gold_item = 0),
    gold_result TEXT,
    recorded_at TEXT NOT NULL,
    release_eligible INTEGER NOT NULL CHECK (release_eligible = 0),
    request_sha256 TEXT NOT NULL,
    record_json TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS deletion_requests (
    deletion_request_id TEXT PRIMARY KEY,
    schema_version INTEGER NOT NULL CHECK (schema_version = 2),
    participant_id TEXT NOT NULL,
    reason TEXT NOT NULL,
    scope TEXT NOT NULL CHECK (scope = 'all'),
    requested_at TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status = 'pending'),
    tombstoned INTEGER NOT NULL CHECK (tombstoned = 1),
    hard_deleted INTEGER NOT NULL CHECK (hard_deleted = 0),
    request_sha256 TEXT NOT NULL,
    record_json TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS idempotency_keys (
    operation TEXT NOT NULL,
    idempotency_key TEXT NOT NULL,
    request_sha256 TEXT NOT NULL,
    resource_type TEXT NOT NULL,
    resource_id TEXT NOT NULL,
    response_json TEXT NOT NULL,
    created_at TEXT NOT NULL,
    PRIMARY KEY (operation, idempotency_key)
);

CREATE TABLE IF NOT EXISTS pending_audit_outbox (
    outbox_id TEXT PRIMARY KEY,
    submission_id TEXT NOT NULL UNIQUE REFERENCES submissions(submission_id),
    queue_id TEXT NOT NULL UNIQUE,
    pending_row_json TEXT NOT NULL,
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS pending_audit_attempts (
    attempt_id TEXT PRIMARY KEY,
    outbox_id TEXT NOT NULL REFERENCES pending_audit_outbox(outbox_id),
    attempted_at TEXT NOT NULL,
    succeeded INTEGER NOT NULL CHECK (succeeded IN (0, 1)),
    error_code TEXT
);

CREATE TABLE IF NOT EXISTS deletion_propagation_outbox (
    action_id TEXT PRIMARY KEY,
    deletion_request_id TEXT NOT NULL REFERENCES deletion_requests(deletion_request_id),
    participant_id TEXT NOT NULL,
    target TEXT NOT NULL CHECK (target IN ('pending_audit_csv', 'huggingface_export')),
    action TEXT NOT NULL CHECK (action = 'tombstone_participant_records'),
    payload_json TEXT NOT NULL,
    created_at TEXT NOT NULL,
    UNIQUE (deletion_request_id, target)
);

CREATE TABLE IF NOT EXISTS deletion_propagation_attempts (
    attempt_id TEXT PRIMARY KEY,
    action_id TEXT NOT NULL REFERENCES deletion_propagation_outbox(action_id),
    attempted_at TEXT NOT NULL,
    succeeded INTEGER NOT NULL CHECK (succeeded IN (0, 1)),
    error_code TEXT
);

CREATE TABLE IF NOT EXISTS iedid_deletion_outbox (
    action_id TEXT PRIMARY KEY,
    deletion_request_id TEXT NOT NULL UNIQUE REFERENCES deletion_requests(deletion_request_id),
    participant_id TEXT NOT NULL,
    consent_ids_json TEXT NOT NULL,
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS iedid_deletion_attempts (
    attempt_id TEXT PRIMARY KEY,
    action_id TEXT NOT NULL REFERENCES iedid_deletion_outbox(action_id),
    attempted_at TEXT NOT NULL,
    succeeded INTEGER NOT NULL CHECK (succeeded IN (0, 1)),
    error_code TEXT
);

CREATE INDEX IF NOT EXISTS idx_consents_participant
    ON consents(participant_id, recorded_at);
CREATE INDEX IF NOT EXISTS idx_submissions_participant
    ON submissions(participant_id, submitted_at);
CREATE INDEX IF NOT EXISTS idx_submissions_task
    ON submissions(app_id, game_id, task_id);
CREATE INDEX IF NOT EXISTS idx_judgments_submission
    ON judgments(submission_id, recorded_at);
CREATE INDEX IF NOT EXISTS idx_judgments_participant
    ON judgments(participant_id, recorded_at);
CREATE INDEX IF NOT EXISTS idx_deletions_participant
    ON deletion_requests(participant_id, requested_at);
"""

APPEND_ONLY_TABLES = (
    "consents",
    "submissions",
    "judgments",
    "deletion_requests",
    "idempotency_keys",
    "pending_audit_outbox",
    "pending_audit_attempts",
    "deletion_propagation_outbox",
    "deletion_propagation_attempts",
    "iedid_deletion_outbox",
    "iedid_deletion_attempts",
)

# Existing v2 databases predate the complete ethical-policy receipt. SQLite
# cannot add NOT NULL columns without inventing defaults, so migrations append
# nullable columns. Old receipts consequently remain explicitly unverified;
# only newly inserted receipts populate the complete policy snapshot.
CONSENT_COLUMN_MIGRATIONS = {
    "consent_surface": "TEXT",
    "receipt_schema_version": "INTEGER",
    "consent_locale": "TEXT",
    "consent_text_sha256": "TEXT",
    "research_purposes_json": "TEXT",
    "retention_policy_version": "TEXT",
    "withdrawal_policy_version": "TEXT",
    "eligibility_confirmed": "INTEGER",
    "public_release_confirmed": "INTEGER",
    "commercial_ai_reuse_confirmed": "INTEGER",
    "privacy_risks_acknowledged": "INTEGER",
    "operational_processing_acknowledged": "INTEGER",
    "voluntary_participation_confirmed": "INTEGER",
    "contribution_rights_confirmed": "INTEGER",
    "raw_audio_consent": "INTEGER",
    "data_controller_name": "TEXT",
    "data_controller_contact": "TEXT",
    "data_recipient_categories_json": "TEXT",
    "withdrawal_token_sha256": "TEXT",
    "consent_status": "TEXT",
    "consent_provenance": "TEXT",
    "record_sha256": "TEXT",
}


def _configured_db_path() -> Path | None:
    configured = os.environ.get(DB_PATH_ENV, "").strip()
    if not configured:
        configured = next(
            (
                os.environ.get(name, "").strip()
                for name in COMPAT_DB_PATH_ENVS
                if os.environ.get(name, "").strip()
            ),
            "",
        )
    if configured:
        if configured == ":memory:":
            raise StorageConfigurationError(
                f"{DB_PATH_ENV} must name a durable file, not an in-memory database"
            )
        return Path(configured).expanduser().resolve()
    return None


def _default_db_path() -> Path:
    configured = _configured_db_path()
    if configured is not None:
        return configured

    if os.environ.get(ALLOW_EPHEMERAL_DB_ENV, "").strip().casefold() != "true":
        raise StorageConfigurationError(
            "Crowdsourced collection is unavailable until COLLECTION_DB_PATH "
            "is explicitly configured on persistent storage. Set "
            "ALLOW_EPHEMERAL_COLLECTION=true only for disposable development "
            "or tests."
        )

    return (
        Path(tempfile.gettempdir())
        / "purepolyglot"
        / "collection_v2.sqlite3"
    ).resolve()


def _pending_queue_path() -> Path:
    configured = os.environ.get("PENDING_APPROVALS_FILE", "").strip()
    if configured:
        return Path(configured).expanduser().resolve()
    if Path("/app").exists():
        return Path("/app/pending_approvals.csv")
    return Path("pending_approvals.csv").resolve()


def _iedid_dataset_path() -> Path:
    configured = (
        os.environ.get("IEDID_DATASET_DIR", "").strip()
        or os.environ.get("DATASET_DIR", "").strip()
    )
    if configured:
        return Path(configured).expanduser().resolve()
    if Path("/app/IEDID").exists():
        return Path("/app/IEDID")
    return (Path(__file__).resolve().parents[1] / "IEDID").resolve()


def _consent_snapshot(consent: sqlite3.Row) -> dict[str, str]:
    """Build the immutable dataset lineage from an authoritative DB receipt."""
    keys = set(consent.keys())

    def value(column: str) -> Any:
        return consent[column] if column in keys else None

    def json_list(column: str) -> str:
        raw = value(column)
        if not raw:
            return ""
        try:
            parsed = json.loads(raw)
        except (TypeError, ValueError, json.JSONDecodeError):
            return ""
        if not isinstance(parsed, list) or not all(
            isinstance(item, str) and item for item in parsed
        ):
            return ""
        return canonical_json_list(parsed)

    candidate = {
        "Consent_ID": str(value("consent_id") or ""),
        "Consent_Status": str(value("consent_status") or "legacy_unverified"),
        "Consent_Provenance": str(
            value("consent_provenance") or "legacy_incomplete_v2_receipt"
        ),
        "Consent_Confirmed": "true" if value("consent_granted") == 1 else "",
        "Consent_Version": str(value("consent_version") or ""),
        "Consent_Recorded_At": str(value("recorded_at") or ""),
        "Privacy_Notice_Version": str(value("privacy_notice_version") or ""),
        "Dataset_License": str(value("dataset_license") or ""),
        "Permitted_Modalities": json_list("permitted_modalities_json"),
        "Consent_Surface": str(value("consent_surface") or ""),
        "Consent_Locale": str(value("consent_locale") or ""),
        "Consent_Text_SHA256": str(value("consent_text_sha256") or ""),
        "Consent_Record_SHA256": str(value("record_sha256") or ""),
        "Consent_Schema_Version": str(value("receipt_schema_version") or ""),
        "Research_Purposes": json_list("research_purposes_json"),
        "Retention_Policy_Version": str(
            value("retention_policy_version") or ""
        ),
        "Withdrawal_Policy_Version": str(
            value("withdrawal_policy_version") or ""
        ),
        "Eligibility_Confirmed": (
            "true" if value("eligibility_confirmed") == 1 else ""
        ),
        "Public_Release_Confirmed": (
            "true" if value("public_release_confirmed") == 1 else ""
        ),
        "Commercial_AI_Reuse_Confirmed": (
            "true" if value("commercial_ai_reuse_confirmed") == 1 else ""
        ),
        "Privacy_Risks_Acknowledged": (
            "true" if value("privacy_risks_acknowledged") == 1 else ""
        ),
        "Operational_Processing_Acknowledged": (
            "true" if value("operational_processing_acknowledged") == 1 else ""
        ),
        "Voluntary_Participation_Confirmed": (
            "true" if value("voluntary_participation_confirmed") == 1 else ""
        ),
        "Contribution_Rights_Confirmed": (
            "true" if value("contribution_rights_confirmed") == 1 else ""
        ),
        "Raw_Audio_Consent": (
            "true"
            if value("raw_audio_consent") == 1
            else "false"
            if value("raw_audio_consent") == 0
            else ""
        ),
        "Data_Controller_Name": str(value("data_controller_name") or ""),
        "Data_Controller_Contact": str(value("data_controller_contact") or ""),
        "Data_Recipient_Categories": json_list(
            "data_recipient_categories_json"
        ),
    }
    if not consent_lineage_is_verified(candidate):
        candidate["Consent_Status"] = "legacy_unverified"
        if candidate["Consent_Provenance"] == VERIFIED_CONSENT_PROVENANCE:
            candidate["Consent_Provenance"] = "legacy_incomplete_v2_receipt"
    return candidate


def _legacy_pending_row(
    payload: SubmissionCreate,
    submission_id: str,
    queue_id: str,
    submitted_at: str,
    consent_snapshot: dict[str, str],
) -> dict[str, Any]:
    metadata = payload.metadata
    final_utterance = payload.user_transcript_final or payload.source_text or ""
    final_output = payload.user_output_final or payload.machine_output_initial or ""
    context_parts = [
        payload.prompt_text or "",
        payload.preceding_context or "",
    ]
    context = str(metadata.get("context") or " | ".join(
        part for part in context_parts if part
    ))
    return {
        # Fields consumed by the current Pending Audit UI.
        "User": payload.participant_id,
        "Data_Origin": str(metadata.get("source_tag") or f"Game: {payload.game_id}"),
        "Utterance": final_utterance,
        "Dialect": payload.target_dialect,
        "Clarification": final_output,
        "Clarification_Source": "Schema v2 crowdsourcing submission",
        "Tone": str(metadata.get("tone") or ""),
        "Context": context,
        "Pragmatic_Analysis": str(metadata.get("pragmatics") or ""),
        "Audio": "",
        "Timestamp": submitted_at,
        "Chain_ID": "",
        "Approvers": "",
        "Language": payload.target_language,
        # Stable bridge fields make the mirror idempotent and auditable.
        "Queue_ID": queue_id,
        "Participant_ID": payload.participant_id,
        "Submission_ID": submission_id,
        "Interaction_ID": payload.task_id,
        "Schema_Version": str(SCHEMA_VERSION),
        "App_Source": payload.app_id,
        "App_Version": payload.app_version,
        "Game_ID": payload.game_id,
        "Task_ID": payload.task_id,
        "Source_App_ID": payload.app_id,
        "Source_App_Version": payload.app_version,
        "Source_Game_ID": payload.game_id,
        "Collected_At": submitted_at,
        "Submission_Status": "Pending Review",
        **consent_snapshot,
        "Source_Language": payload.source_language,
        "Source_Dialect": payload.source_dialect,
        "Source_Input_Mode": payload.input_modality,
        "Machine_Transcript_Initial": payload.machine_transcript_initial or "",
        "User_Transcript_Final": payload.user_transcript_final or "",
        "Target_Language": payload.target_language,
        "Target_Dialect": payload.target_dialect,
        "Machine_Translation_Initial": payload.machine_output_initial or "",
        "User_Translation_Final": payload.user_output_final or "",
        "AI_Model": payload.model_name or "",
        "AI_Model_Version": payload.model_version or "",
        "ASR_Model": payload.asr_model or "",
        "Translation_Route": payload.translation_route or "",
        "Metadata_JSON": _canonical_json(payload.metadata),
        "Prompt_Text": payload.prompt_text or "",
        "Topic_ID": payload.topic_id or "",
        "Image_Asset_ID": payload.image_asset_id or "",
        "Conversation_ID": payload.conversation_id or "",
        "Turn_ID": payload.turn_id or "",
        "Review_Submitted_At": submitted_at,
        "Release_Eligible": "false",
        "Ethics_Review_Status": "pending_moderation",
        "Withdrawal_Request_ID": "",
        "Withdrawal_Requested_At": "",
        "Deletion_Propagation_Status": "not_requested",
    }


class CollectionStore:
    """Small SQLite event store using one transaction per API mutation."""

    def __init__(self, path: Path):
        self.path = path
        self._initialize_lock = threading.Lock()
        self._initialized = False

    def _connect(self) -> sqlite3.Connection:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        connection = sqlite3.connect(
            str(self.path),
            timeout=30,
            isolation_level=None,
        )
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON")
        connection.execute("PRAGMA busy_timeout = 30000")
        connection.execute("PRAGMA journal_mode = WAL")
        connection.execute("PRAGMA synchronous = FULL")
        return connection

    def initialize(self) -> None:
        if self._initialized:
            return
        with self._initialize_lock:
            if self._initialized:
                return
            with self._connect() as connection:
                connection.execute("BEGIN IMMEDIATE")
                try:
                    connection.executescript(SCHEMA_SQL + "\n" + RAW_AUDIO_SCHEMA_SQL)
                    consent_columns = {
                        row["name"]
                        for row in connection.execute(
                            "PRAGMA table_info(consents)"
                        ).fetchall()
                    }
                    for column, sql_type in CONSENT_COLUMN_MIGRATIONS.items():
                        if column not in consent_columns:
                            connection.execute(
                                f"ALTER TABLE consents ADD COLUMN {column} {sql_type}"
                            )
                    connection.execute(
                        "INSERT OR IGNORE INTO schema_meta(schema_version, installed_at) "
                        "VALUES (?, ?)",
                        (SCHEMA_VERSION, _utc_now()),
                    )
                    for table in APPEND_ONLY_TABLES:
                        connection.executescript(
                            f"""
                            CREATE TRIGGER IF NOT EXISTS {table}_prevent_update
                            BEFORE UPDATE ON {table}
                            BEGIN
                                SELECT RAISE(ABORT, '{table} is append-only');
                            END;
                            CREATE TRIGGER IF NOT EXISTS {table}_prevent_delete
                            BEFORE DELETE ON {table}
                            BEGIN
                                SELECT RAISE(ABORT, '{table} is append-only');
                            END;
                            """
                        )
                    connection.commit()
                except Exception:
                    connection.rollback()
                    raise
            self._initialized = True

    @staticmethod
    def _begin(connection: sqlite3.Connection) -> None:
        connection.execute("BEGIN IMMEDIATE")

    @staticmethod
    def _idempotent_replay(
        connection: sqlite3.Connection,
        operation: str,
        idempotency_key: str,
        request_sha256: str,
    ) -> dict[str, Any] | None:
        row = connection.execute(
            "SELECT request_sha256, response_json FROM idempotency_keys "
            "WHERE operation = ? AND idempotency_key = ?",
            (operation, idempotency_key),
        ).fetchone()
        if row is None:
            return None
        if row["request_sha256"] != request_sha256:
            raise IdempotencyConflict(
                "Idempotency-Key was already used with a different request"
            )
        response = json.loads(row["response_json"])
        response["idempotent_replay"] = True
        return response

    @staticmethod
    def _record_idempotency(
        connection: sqlite3.Connection,
        operation: str,
        idempotency_key: str,
        request_sha256: str,
        resource_type: str,
        resource_id: str,
        response_body: dict[str, Any],
        created_at: str,
    ) -> None:
        connection.execute(
            "INSERT INTO idempotency_keys "
            "(operation, idempotency_key, request_sha256, resource_type, "
            "resource_id, response_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                operation,
                idempotency_key,
                request_sha256,
                resource_type,
                resource_id,
                _canonical_json(response_body),
                created_at,
            ),
        )

    @staticmethod
    def _ensure_active(connection: sqlite3.Connection, participant_id: str) -> None:
        tombstone = connection.execute(
            "SELECT 1 FROM deletion_requests WHERE participant_id = ? LIMIT 1",
            (participant_id,),
        ).fetchone()
        if tombstone:
            raise ParticipantTombstoned(
                "Participant has requested deletion; new collection is disabled"
            )

    @staticmethod
    def _authenticate_participant(
        connection: sqlite3.Connection,
        participant_id: str,
        participant_token: str,
        *,
        allow_first_binding: bool = False,
        supplied_token_sha256: str | None = None,
    ) -> str:
        token_sha256 = hashlib.sha256(participant_token.encode("utf-8")).hexdigest()
        if supplied_token_sha256 is not None and not hmac.compare_digest(
            token_sha256,
            supplied_token_sha256,
        ):
            raise ConsentError("participant token does not match its supplied SHA-256")
        rows = connection.execute(
            "SELECT withdrawal_token_sha256 FROM consents "
            "WHERE participant_id = ?",
            (participant_id,),
        ).fetchall()
        known_hashes = [
            str(row["withdrawal_token_sha256"] or "")
            for row in rows
            if str(row["withdrawal_token_sha256"] or "")
        ]
        if not rows and allow_first_binding:
            return token_sha256
        matched = False
        for known_hash in known_hashes:
            matched = hmac.compare_digest(token_sha256, known_hash) or matched
        if not matched:
            raise ConsentError("participant token authentication failed")
        return token_sha256

    @staticmethod
    def _require_consent(
        connection: sqlite3.Connection,
        consent_id: str,
        participant_id: str,
        modality: str,
        app_id: str | None = None,
    ) -> sqlite3.Row:
        consent = connection.execute(
            "SELECT * FROM consents WHERE consent_id = ?",
            (consent_id,),
        ).fetchone()
        if consent is None:
            raise ConsentError("consent_id does not exist")
        if consent["participant_id"] != participant_id:
            raise ConsentError("consent_id belongs to a different participant")
        permitted = json.loads(consent["permitted_modalities_json"])
        if modality not in permitted:
            raise ConsentError(f"consent does not permit {modality} contributions")
        receipt_app_id = str(consent["app_id"] or "").strip()
        if app_id and receipt_app_id and receipt_app_id.casefold() != app_id.casefold():
            raise ConsentError("consent_id was issued for a different app")
        return consent

    def get_consent_snapshot(
        self,
        consent_id: str,
        participant_id: str,
        modality: str,
        app_id: str | None = None,
    ) -> dict[str, str]:
        """Validate an active receipt and return its server-authoritative lineage."""
        self.initialize()
        with self._connect() as connection:
            self._ensure_active(connection, participant_id)
            consent = self._require_consent(
                connection,
                consent_id,
                participant_id,
                modality,
                app_id,
            )
            return _consent_snapshot(consent)

    def get_submission_consent_snapshot(
        self,
        submission_id: str,
        participant_id: str,
        consent_id: str,
        task_id: str,
    ) -> dict[str, str]:
        """Bind a release attempt to its canonical v2 submission and receipt."""
        return self.get_submission_release_snapshot(
            submission_id,
            participant_id,
            consent_id,
            task_id,
        )["consent_snapshot"]

    def get_submission_release_snapshot(
        self,
        submission_id: str,
        participant_id: str,
        consent_id: str,
        task_id: str,
    ) -> dict[str, Any]:
        """Return server-owned content/provenance for one active submission.

        The final writer uses this record while holding the shared IEDID lock,
        so a reviewer cannot release edited or CSV-forged content under an
        unrelated participant receipt.
        """
        self.initialize()
        with self._connect() as connection:
            submission = connection.execute(
                "SELECT * FROM submissions WHERE submission_id = ?",
                (submission_id,),
            ).fetchone()
            if submission is None:
                raise MissingResource("submission_id does not exist")
            expected = {
                "participant_id": participant_id,
                "consent_id": consent_id,
                "task_id": task_id,
            }
            if any(
                str(submission[field]) != str(value)
                for field, value in expected.items()
            ):
                raise ConsentError(
                    "submission identifiers do not match the canonical v2 record"
                )
            self._ensure_active(connection, participant_id)
            consent = self._require_consent(
                connection,
                consent_id,
                participant_id,
                submission["input_modality"],
                submission["app_id"],
            )
            try:
                metadata = json.loads(submission["metadata_json"] or "{}")
            except (TypeError, ValueError, json.JSONDecodeError) as exc:
                raise ConsentError(
                    "canonical submission metadata is not valid JSON"
                ) from exc
            if not isinstance(metadata, dict):
                raise ConsentError("canonical submission metadata is not an object")
            canonical_context = str(
                metadata.get("context")
                or " | ".join(
                    part
                    for part in (
                        str(submission["prompt_text"] or ""),
                        str(submission["preceding_context"] or ""),
                    )
                    if part
                )
            )
            return {
                "consent_snapshot": _consent_snapshot(consent),
                "submission_id": str(submission["submission_id"]),
                "participant_id": str(submission["participant_id"]),
                "consent_id": str(submission["consent_id"]),
                "task_id": str(submission["task_id"]),
                "app_id": str(submission["app_id"]),
                "app_version": str(submission["app_version"]),
                "game_id": str(submission["game_id"]),
                "input_modality": str(submission["input_modality"]),
                "utterance": str(
                    submission["user_transcript_final"]
                    or submission["source_text"]
                    or ""
                ),
                "clarification": str(
                    submission["user_output_final"]
                    or submission["machine_output_initial"]
                    or ""
                ),
                "language": str(submission["target_language"]),
                "dialect": str(submission["target_dialect"]),
                "source_language": str(submission["source_language"]),
                "source_dialect": str(submission["source_dialect"]),
                "tone": str(metadata.get("tone") or ""),
                "context": canonical_context,
                "pragmatics": str(metadata.get("pragmatics") or ""),
                "source_tag": str(
                    metadata.get("source_tag")
                    or f"Game: {submission['game_id']}"
                ),
                "machine_transcript_initial": str(
                    submission["machine_transcript_initial"] or ""
                ),
                "user_transcript_final": str(
                    submission["user_transcript_final"] or ""
                ),
                "machine_output_initial": str(
                    submission["machine_output_initial"] or ""
                ),
                "user_output_final": str(submission["user_output_final"] or ""),
                "model_name": str(submission["model_name"] or ""),
                "model_version": str(submission["model_version"] or ""),
                "asr_model": str(submission["asr_model"] or ""),
                "translation_route": str(submission["translation_route"] or ""),
                "submitted_at": str(submission["submitted_at"]),
            }

    def _record_pending_attempt(
        self,
        outbox_id: str,
        succeeded: bool,
        error_code: str | None = None,
    ) -> None:
        try:
            with self._connect() as connection:
                self._begin(connection)
                connection.execute(
                    "INSERT INTO pending_audit_attempts "
                    "(attempt_id, outbox_id, attempted_at, succeeded, error_code) "
                    "VALUES (?, ?, ?, ?, ?)",
                    (
                        _new_id("pat"),
                        outbox_id,
                        _utc_now(),
                        int(succeeded),
                        error_code,
                    ),
                )
                connection.commit()
        except sqlite3.Error:
            # The canonical submission and immutable outbox entry are already
            # durable. A later replay/drain can reconstruct this attempt.
            pass

    @staticmethod
    def _append_pending_csv(
        row: dict[str, Any],
        tombstone_check: Any = None,
    ) -> None:
        # pandas is imported only on delivery, keeping router imports lightweight.
        import pandas as pd

        pending_file = _pending_queue_path()
        pending_file.parent.mkdir(parents=True, exist_ok=True)
        with PENDING_QUEUE_LOCK:
            if callable(tombstone_check) and tombstone_check():
                raise ParticipantTombstoned(
                    "Participant withdrew before Pending Audit delivery"
                )
            if pending_file.exists():
                frame = pd.read_csv(pending_file, dtype=str).fillna("")
            else:
                frame = pd.DataFrame()

            for column in row:
                if column not in frame.columns:
                    frame[column] = ""

            if "Submission_ID" in frame.columns:
                same_submission = (
                    frame["Submission_ID"].astype(str) == row["Submission_ID"]
                )
                if same_submission.any():
                    return
            if "Queue_ID" in frame.columns:
                same_queue = frame["Queue_ID"].astype(str) == row["Queue_ID"]
                if same_queue.any():
                    return

            if frame.empty and not len(frame.columns):
                frame = pd.DataFrame(columns=list(row))
            frame.loc[len(frame)] = {
                column: row.get(column, "") for column in frame.columns
            }
            temp_file = pending_file.with_name(f"{pending_file.name}.v2.tmp")
            frame.to_csv(temp_file, index=False)
            os.replace(temp_file, pending_file)

    def _participant_has_tombstone(self, participant_id: str) -> bool:
        with self._connect() as connection:
            return connection.execute(
                "SELECT 1 FROM deletion_requests WHERE participant_id = ? LIMIT 1",
                (participant_id,),
            ).fetchone() is not None

    def mirror_submission(self, submission_id: str) -> str:
        """Deliver one immutable outbox item to the legacy Pending Audit CSV."""
        self.initialize()
        with self._connect() as connection:
            outbox = connection.execute(
                """
                SELECT outbox.outbox_id, outbox.pending_row_json,
                    submissions.participant_id,
                    EXISTS(
                        SELECT 1 FROM pending_audit_attempts attempts
                        WHERE attempts.outbox_id = outbox.outbox_id
                          AND attempts.succeeded = 1
                    ) AS delivered,
                    (
                        SELECT attempts.error_code
                        FROM pending_audit_attempts attempts
                        WHERE attempts.outbox_id = outbox.outbox_id
                          AND attempts.succeeded = 1
                        ORDER BY attempts.attempted_at DESC
                        LIMIT 1
                    ) AS delivery_code,
                    EXISTS(
                        SELECT 1 FROM deletion_requests requests
                        WHERE requests.participant_id = submissions.participant_id
                    ) AS participant_tombstoned
                FROM pending_audit_outbox AS outbox
                JOIN submissions
                  ON submissions.submission_id = outbox.submission_id
                WHERE outbox.submission_id = ?
                """,
                (submission_id,),
            ).fetchone()
        if outbox is None:
            return "retry_pending"
        if outbox["delivered"]:
            if outbox["delivery_code"] == "participant_tombstoned_no_delivery":
                return "withdrawn_not_mirrored"
            return "mirrored"
        if outbox["participant_tombstoned"]:
            self._record_pending_attempt(
                outbox["outbox_id"],
                True,
                error_code="participant_tombstoned_no_delivery",
            )
            return "withdrawn_not_mirrored"

        try:
            self._append_pending_csv(
                json.loads(outbox["pending_row_json"]),
                tombstone_check=lambda: self._participant_has_tombstone(
                    outbox["participant_id"]
                ),
            )
        except ParticipantTombstoned:
            self._record_pending_attempt(
                outbox["outbox_id"],
                True,
                error_code="participant_tombstoned_no_delivery",
            )
            return "withdrawn_not_mirrored"
        except Exception as exc:
            self._record_pending_attempt(
                outbox["outbox_id"],
                False,
                error_code=type(exc).__name__[:128],
            )
            return "retry_pending"

        self._record_pending_attempt(outbox["outbox_id"], True)
        return "mirrored"

    def drain_pending_audit(
        self,
        limit: int = 10,
        exclude_submission_id: str | None = None,
    ) -> int:
        """Best-effort retry of prior outbox items; failures remain recorded."""
        self.initialize()
        with self._connect() as connection:
            rows = connection.execute(
                """
                SELECT outbox.submission_id
                FROM pending_audit_outbox AS outbox
                WHERE outbox.submission_id != COALESCE(?, '')
                  AND NOT EXISTS (
                      SELECT 1 FROM pending_audit_attempts AS attempts
                      WHERE attempts.outbox_id = outbox.outbox_id
                        AND attempts.succeeded = 1
                  )
                ORDER BY outbox.created_at
                LIMIT ?
                """,
                (exclude_submission_id, max(1, min(limit, 100))),
            ).fetchall()
        return sum(
            self.mirror_submission(row["submission_id"]) == "mirrored"
            for row in rows
        )

    def _record_deletion_propagation_attempt(
        self,
        action_id: str,
        succeeded: bool,
        error_code: str | None = None,
    ) -> None:
        try:
            with self._connect() as connection:
                self._begin(connection)
                connection.execute(
                    "INSERT INTO deletion_propagation_attempts "
                    "(attempt_id, action_id, attempted_at, succeeded, error_code) "
                    "VALUES (?, ?, ?, ?, ?)",
                    (
                        _new_id("dpa"),
                        action_id,
                        _utc_now(),
                        int(succeeded),
                        error_code,
                    ),
                )
                connection.commit()
        except sqlite3.Error:
            pass

    @staticmethod
    def _tombstone_pending_csv(
        participant_id: str,
        deletion_request_id: str,
        requested_at: str,
        submission_ids: tuple[str, ...] = (),
        consent_ids: tuple[str, ...] = (),
    ) -> None:
        import pandas as pd

        pending_file = _pending_queue_path()
        with PENDING_QUEUE_LOCK:
            if not pending_file.exists():
                return
            frame = pd.read_csv(pending_file, dtype=str).fillna("")
            contribution_mask = pd.Series(False, index=frame.index, dtype=bool)
            if "User" in frame.columns:
                contribution_mask |= frame["User"].astype(str) == participant_id
            if "Participant_ID" in frame.columns:
                contribution_mask |= frame["Participant_ID"].astype(str) == participant_id
            if submission_ids and "Submission_ID" in frame.columns:
                contribution_mask |= frame["Submission_ID"].astype(str).isin(submission_ids)
            if consent_ids and "Consent_ID" in frame.columns:
                contribution_mask |= frame["Consent_ID"].astype(str).isin(consent_ids)
            approvers_changed = False
            if "Approvers" in frame.columns:
                cleaned_approvers = frame["Approvers"].astype(str).map(
                    lambda values: ",".join(
                        entry
                        for entry in values.split(",")
                        if entry and entry != participant_id
                    )
                )
                approvers_changed = not cleaned_approvers.equals(frame["Approvers"])
                frame["Approvers"] = cleaned_approvers

            if not contribution_mask.any() and not approvers_changed:
                return

            if contribution_mask.any():
                _scrub_tombstoned_csv_rows(
                    frame, contribution_mask, "[deleted]"
                )
                frame.loc[contribution_mask, "User"] = "deleted-participant"
                frame.loc[contribution_mask, "Submission_Status"] = (
                    "Deletion Requested"
                )
                if "Review_Status" in frame.columns:
                    frame.loc[contribution_mask, "Review_Status"] = (
                        "Withdrawal Requested"
                    )
                frame.loc[contribution_mask, "Release_Eligible"] = "false"
                if "Consent_Status" in frame.columns:
                    frame.loc[contribution_mask, "Consent_Status"] = "withdrawn"
                if "Consent_Confirmed" in frame.columns:
                    frame.loc[contribution_mask, "Consent_Confirmed"] = ""
                frame.loc[contribution_mask, "Ethics_Review_Status"] = "withdrawn"
                frame.loc[contribution_mask, "Withdrawal_Request_ID"] = deletion_request_id
                frame.loc[contribution_mask, "Withdrawal_Requested_At"] = requested_at
                frame.loc[
                    contribution_mask, "Deletion_Propagation_Status"
                ] = "local_tombstoned_remote_pending"

            temp_file = pending_file.with_name(f"{pending_file.name}.delete.tmp")
            frame.to_csv(temp_file, index=False)
            os.replace(temp_file, pending_file)

    def propagate_deletion_to_pending_audit(
        self,
        deletion_request_id: str,
    ) -> bool:
        """Tombstone a participant's local Pending Audit copies."""
        self.initialize()
        with self._connect() as connection:
            action = connection.execute(
                """
                SELECT outbox.action_id, outbox.participant_id, requests.requested_at,
                    EXISTS(
                        SELECT 1 FROM deletion_propagation_attempts AS attempts
                        WHERE attempts.action_id = outbox.action_id
                          AND attempts.succeeded = 1
                    ) AS delivered
                FROM deletion_propagation_outbox AS outbox
                JOIN deletion_requests AS requests
                  ON requests.deletion_request_id = outbox.deletion_request_id
                WHERE outbox.deletion_request_id = ?
                  AND outbox.target = 'pending_audit_csv'
                """,
                (deletion_request_id,),
            ).fetchone()
            submission_ids: tuple[str, ...] = ()
            consent_ids: tuple[str, ...] = ()
            if action is not None:
                submission_ids = tuple(sorted(
                    str(row["submission_id"])
                    for row in connection.execute(
                        "SELECT submission_id FROM submissions WHERE participant_id = ?",
                        (action["participant_id"],),
                    ).fetchall()
                ))
                consent_ids = tuple(sorted(
                    str(row["consent_id"])
                    for row in connection.execute(
                        "SELECT consent_id FROM consents WHERE participant_id = ?",
                        (action["participant_id"],),
                    ).fetchall()
                ))
        if action is None:
            return False
        if action["delivered"]:
            return True
        try:
            self._tombstone_pending_csv(
                action["participant_id"],
                deletion_request_id,
                action["requested_at"],
                submission_ids,
                consent_ids,
            )
        except Exception as exc:
            self._record_deletion_propagation_attempt(
                action["action_id"],
                False,
                type(exc).__name__[:128],
            )
            return False
        self._record_deletion_propagation_attempt(action["action_id"], True)
        return True

    def _record_iedid_deletion_attempt(
        self,
        action_id: str,
        succeeded: bool,
        error_code: str | None = None,
    ) -> None:
        try:
            with self._connect() as connection:
                self._begin(connection)
                connection.execute(
                    "INSERT INTO iedid_deletion_attempts "
                    "(attempt_id, action_id, attempted_at, succeeded, error_code) "
                    "VALUES (?, ?, ?, ?, ?)",
                    (
                        _new_id("ida"),
                        action_id,
                        _utc_now(),
                        int(succeeded),
                        error_code,
                    ),
                )
                connection.commit()
        except sqlite3.Error:
            pass

    @staticmethod
    def _tombstone_local_iedid(
        participant_id: str,
        consent_ids: list[str],
        deletion_request_id: str,
        requested_at: str,
    ) -> None:
        import pandas as pd

        dataset_root = _iedid_dataset_path()
        ignored = {
            "minted_history.csv",
            "system_feedback.csv",
            "eeqs_events.csv",
        }
        with IEDID_DATASET_LOCK:
            csv_paths = (
                set(dataset_root.rglob("*.csv"))
                if dataset_root.exists()
                else set()
            )
            rejected_audit = dataset_root.parent / "rejected_graveyard.csv"
            if rejected_audit.is_file():
                csv_paths.add(rejected_audit)
            for csv_path in sorted(csv_paths):
                if csv_path.name.casefold() in ignored:
                    continue
                frame = pd.read_csv(csv_path, dtype=str, encoding="utf-8-sig").fillna("")
                mask = pd.Series(False, index=frame.index)
                if "User" in frame.columns:
                    mask |= frame["User"].astype(str) == participant_id
                if "Participant_ID" in frame.columns:
                    mask |= frame["Participant_ID"].astype(str) == participant_id
                if consent_ids and "Consent_ID" in frame.columns:
                    mask |= frame["Consent_ID"].astype(str).isin(consent_ids)
                if not mask.any():
                    continue

                for column in (
                    "Consent_Status",
                    "Consent_Confirmed",
                    "Review_Status",
                    "Release_Eligible",
                    "Ethics_Review_Status",
                    "Withdrawal_Request_ID",
                    "Withdrawal_Requested_At",
                    "Deletion_Propagation_Status",
                    "Audio_Deletion_Status",
                ):
                    if column not in frame.columns:
                        frame[column] = ""
                _scrub_tombstoned_csv_rows(frame, mask, "[withdrawn]")
                if "User" in frame.columns:
                    frame.loc[mask, "User"] = "withdrawn-participant"
                if "Participant_ID" in frame.columns:
                    frame.loc[mask, "Participant_ID"] = ""
                frame.loc[mask, "Consent_Status"] = "withdrawn"
                frame.loc[mask, "Consent_Confirmed"] = ""
                frame.loc[mask, "Review_Status"] = "Withdrawal Requested"
                if "Submission_Status" in frame.columns:
                    frame.loc[mask, "Submission_Status"] = "Deletion Requested"
                frame.loc[mask, "Release_Eligible"] = "false"
                frame.loc[mask, "Ethics_Review_Status"] = "withdrawn"
                frame.loc[mask, "Withdrawal_Request_ID"] = deletion_request_id
                frame.loc[mask, "Withdrawal_Requested_At"] = requested_at
                frame.loc[
                    mask, "Deletion_Propagation_Status"
                ] = "local_tombstoned_remote_pending"
                frame.loc[mask, "Audio_Deletion_Status"] = "pending_propagation"

                temporary_path = csv_path.with_name(
                    f".{csv_path.name}.withdrawal.tmp"
                )
                frame.to_csv(temporary_path, index=False)
                os.replace(temporary_path, csv_path)

    def propagate_deletion_to_local_iedid(
        self,
        deletion_request_id: str,
    ) -> bool:
        """Tombstone controlled local IEDID rows; remote copies remain queued."""
        self.initialize()
        with self._connect() as connection:
            action = connection.execute(
                """
                SELECT outbox.action_id, outbox.participant_id,
                    outbox.consent_ids_json, requests.requested_at,
                    EXISTS(
                        SELECT 1 FROM iedid_deletion_attempts AS attempts
                        WHERE attempts.action_id = outbox.action_id
                          AND attempts.succeeded = 1
                    ) AS delivered
                FROM iedid_deletion_outbox AS outbox
                JOIN deletion_requests AS requests
                  ON requests.deletion_request_id = outbox.deletion_request_id
                WHERE outbox.deletion_request_id = ?
                """,
                (deletion_request_id,),
            ).fetchone()
        if action is None:
            return False
        if action["delivered"]:
            return True
        try:
            self._tombstone_local_iedid(
                action["participant_id"],
                json.loads(action["consent_ids_json"]),
                deletion_request_id,
                action["requested_at"],
            )
        except Exception as exc:
            self._record_iedid_deletion_attempt(
                action["action_id"],
                False,
                type(exc).__name__[:128],
            )
            return False
        self._record_iedid_deletion_attempt(action["action_id"], True)
        return True

    def create_consent(
        self,
        payload: ConsentCreate,
        idempotency_key: str,
        participant_token: str,
    ) -> tuple[dict[str, Any], bool]:
        self.initialize()
        request_data = payload.model_dump(mode="json")
        request_sha256 = _request_hash(request_data)
        receipt_data = dict(request_data)
        receipt_data.pop("withdrawal_token_sha256", None)
        receipt_data.pop("participant_id", None)
        operation = "create_consent"
        with self._connect() as connection:
            self._begin(connection)
            try:
                self._ensure_active(connection, payload.participant_id)
                token_sha256 = self._authenticate_participant(
                    connection,
                    payload.participant_id,
                    participant_token,
                    allow_first_binding=True,
                    supplied_token_sha256=payload.withdrawal_token_sha256,
                )
                replay = self._idempotent_replay(
                    connection, operation, idempotency_key, request_sha256
                )
                if replay is not None:
                    connection.commit()
                    return replay, True
                consent_id = _new_id("cns")
                recorded_at = _utc_now()
                record = {
                    **receipt_data,
                    "consent_id": consent_id,
                    "recorded_at": recorded_at,
                    "schema_version": SCHEMA_VERSION,
                    "consent_status": VERIFIED_CONSENT_STATUS,
                    "consent_provenance": VERIFIED_CONSENT_PROVENANCE,
                }
                record_sha256 = _request_hash(record)
                connection.execute(
                    """
                    INSERT INTO consents (
                        consent_id, schema_version, participant_id, app_id,
                        receipt_schema_version,
                        app_version, consent_version, privacy_notice_version, dataset_license,
                        permitted_modalities_json, consent_surface, consent_locale,
                        consent_text_sha256, research_purposes_json,
                        retention_policy_version, withdrawal_policy_version,
                        eligibility_confirmed, public_release_confirmed,
                        commercial_ai_reuse_confirmed, privacy_risks_acknowledged,
                        operational_processing_acknowledged,
                        voluntary_participation_confirmed,
                        contribution_rights_confirmed, raw_audio_consent,
                        data_controller_name, data_controller_contact,
                        data_recipient_categories_json, withdrawal_token_sha256,
                        consent_status,
                        consent_provenance, consent_granted, recorded_at,
                        request_sha256, record_sha256, record_json
                    ) VALUES (
                        :consent_id, :schema_version, :participant_id, :app_id,
                        :receipt_schema_version,
                        :app_version, :consent_version, :privacy_notice_version,
                        :dataset_license, :permitted_modalities_json,
                        :consent_surface, :consent_locale, :consent_text_sha256,
                        :research_purposes_json, :retention_policy_version,
                        :withdrawal_policy_version, 1, 1, 1, 1, 1, 1, 1,
                        :raw_audio_consent,
                        :data_controller_name, :data_controller_contact,
                        :data_recipient_categories_json, :withdrawal_token_sha256,
                        :consent_status,
                        :consent_provenance, 1, :recorded_at,
                        :request_sha256, :record_sha256, :record_json
                    )
                    """,
                    {
                        "consent_id": consent_id,
                        "schema_version": STORAGE_SCHEMA_VERSION,
                        "receipt_schema_version": SCHEMA_VERSION,
                        "participant_id": payload.participant_id,
                        "app_id": payload.app_id,
                        "app_version": payload.app_version,
                        "consent_version": payload.consent_version,
                        "privacy_notice_version": payload.privacy_notice_version,
                        "dataset_license": payload.dataset_license,
                        "permitted_modalities_json": _canonical_json(
                            payload.permitted_modalities
                        ),
                        "consent_surface": payload.consent_surface,
                        "consent_locale": payload.consent_locale,
                        "consent_text_sha256": payload.consent_text_sha256,
                        "research_purposes_json": _canonical_json(
                            payload.research_purposes
                        ),
                        "retention_policy_version": payload.retention_policy_version,
                        "withdrawal_policy_version": payload.withdrawal_policy_version,
                        "raw_audio_consent": int(payload.raw_audio_consent),
                        "data_controller_name": payload.data_controller_name,
                        "data_controller_contact": payload.data_controller_contact,
                        "data_recipient_categories_json": _canonical_json(
                            payload.data_recipient_categories
                        ),
                        "withdrawal_token_sha256": token_sha256,
                        "consent_status": VERIFIED_CONSENT_STATUS,
                        "consent_provenance": VERIFIED_CONSENT_PROVENANCE,
                        "recorded_at": recorded_at,
                        "request_sha256": request_sha256,
                        "record_sha256": record_sha256,
                        "record_json": _canonical_json(record),
                    },
                )
                response = {
                    "consent_id": consent_id,
                    "recorded_at": recorded_at,
                    "consent_record_sha256": record_sha256,
                    "consent_text_sha256": payload.consent_text_sha256,
                    "receipt_schema_version": SCHEMA_VERSION,
                    "consent_version": payload.consent_version,
                    "permitted_modalities": sorted(payload.permitted_modalities),
                    "consent_status": VERIFIED_CONSENT_STATUS,
                    "consent_provenance": VERIFIED_CONSENT_PROVENANCE,
                    "idempotent_replay": False,
                }
                self._record_idempotency(
                    connection,
                    operation,
                    idempotency_key,
                    request_sha256,
                    "consent",
                    consent_id,
                    response,
                    recorded_at,
                )
                connection.commit()
                return response, False
            except Exception:
                connection.rollback()
                raise

    def create_submission(
        self,
        payload: SubmissionCreate,
        idempotency_key: str,
        participant_token: str,
    ) -> tuple[dict[str, Any], bool]:
        self.initialize()
        request_data = payload.model_dump(mode="json")
        request_sha256 = _request_hash(request_data)
        operation = "create_submission"
        with self._connect() as connection:
            self._begin(connection)
            try:
                self._ensure_active(connection, payload.participant_id)
                self._authenticate_participant(
                    connection,
                    payload.participant_id,
                    participant_token,
                )
                replay = self._idempotent_replay(
                    connection, operation, idempotency_key, request_sha256
                )
                if replay is not None:
                    connection.commit()
                    return replay, True
                consent = self._require_consent(
                    connection,
                    payload.consent_id,
                    payload.participant_id,
                    payload.input_modality,
                    payload.app_id,
                )
                consent_snapshot = _consent_snapshot(consent)
                if not consent_lineage_is_verified(consent_snapshot):
                    raise ConsentError(
                        "consent receipt predates the complete ethical policy; "
                        "record a new consent receipt"
                    )
                submission_id = _new_id("sub")
                outbox_id = _new_id("out")
                pending_queue_id = f"v2-{uuid.uuid4()}"
                submitted_at = _utc_now()
                record = {
                    **request_data,
                    "submission_id": submission_id,
                    "submitted_at": submitted_at,
                    "schema_version": SCHEMA_VERSION,
                    "moderation_status": "pending",
                    "review_status": "pending",
                    "release_eligible": False,
                    "consent_snapshot": consent_snapshot,
                }
                connection.execute(
                    """
                    INSERT INTO submissions (
                        submission_id, schema_version, participant_id, consent_id,
                        app_id, app_version, game_id, task_id, prompt_text, topic_id,
                        image_asset_id, conversation_id, turn_id, preceding_context,
                        source_text, source_language, source_dialect, target_language,
                        target_dialect, input_modality, machine_transcript_initial,
                        user_transcript_final, machine_output_initial, user_output_final,
                        model_name, model_version, asr_model, translation_route,
                        metadata_json, started_at, submitted_at, duration_ms,
                        moderation_status, review_status, release_eligible,
                        request_sha256, record_json
                    ) VALUES (
                        ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                        ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', 'pending', 0, ?, ?
                    )
                    """,
                    (
                        submission_id,
                        STORAGE_SCHEMA_VERSION,
                        payload.participant_id,
                        payload.consent_id,
                        payload.app_id,
                        payload.app_version,
                        payload.game_id,
                        payload.task_id,
                        payload.prompt_text,
                        payload.topic_id,
                        payload.image_asset_id,
                        payload.conversation_id,
                        payload.turn_id,
                        payload.preceding_context,
                        payload.source_text,
                        payload.source_language,
                        payload.source_dialect,
                        payload.target_language,
                        payload.target_dialect,
                        payload.input_modality,
                        payload.machine_transcript_initial,
                        payload.user_transcript_final,
                        payload.machine_output_initial,
                        payload.user_output_final,
                        payload.model_name,
                        payload.model_version,
                        payload.asr_model,
                        payload.translation_route,
                        _canonical_json(payload.metadata),
                        request_data.get("started_at"),
                        submitted_at,
                        payload.duration_ms,
                        request_sha256,
                        _canonical_json(record),
                    ),
                )
                pending_row = _legacy_pending_row(
                    payload,
                    submission_id,
                    pending_queue_id,
                    submitted_at,
                    consent_snapshot,
                )
                connection.execute(
                    "INSERT INTO pending_audit_outbox "
                    "(outbox_id, submission_id, queue_id, pending_row_json, created_at) "
                    "VALUES (?, ?, ?, ?, ?)",
                    (
                        outbox_id,
                        submission_id,
                        pending_queue_id,
                        _canonical_json(pending_row),
                        submitted_at,
                    ),
                )
                response = {
                    "submission_id": submission_id,
                    "submitted_at": submitted_at,
                    "moderation_status": "pending",
                    "review_status": "pending",
                    "release_eligible": False,
                    "pending_queue_id": pending_queue_id,
                    "pending_audit_status": "retry_pending",
                    "idempotent_replay": False,
                }
                self._record_idempotency(
                    connection,
                    operation,
                    idempotency_key,
                    request_sha256,
                    "submission",
                    submission_id,
                    response,
                    submitted_at,
                )
                connection.commit()
                return response, False
            except Exception:
                connection.rollback()
                raise

    def create_judgment(
        self,
        payload: JudgmentCreate,
        idempotency_key: str,
        participant_token: str,
    ) -> tuple[dict[str, Any], bool]:
        self.initialize()
        request_data = payload.model_dump(mode="json")
        request_sha256 = _request_hash(request_data)
        operation = "create_judgment"
        with self._connect() as connection:
            self._begin(connection)
            try:
                self._ensure_active(connection, payload.participant_id)
                self._authenticate_participant(
                    connection,
                    payload.participant_id,
                    participant_token,
                )
                replay = self._idempotent_replay(
                    connection, operation, idempotency_key, request_sha256
                )
                if replay is not None:
                    connection.commit()
                    return replay, True
                submission = connection.execute(
                    "SELECT participant_id, task_id, app_id FROM submissions "
                    "WHERE submission_id = ?",
                    (payload.submission_id,),
                ).fetchone()
                if submission is None:
                    raise MissingResource("submission_id does not exist")
                if str(submission["task_id"]) != payload.task_id:
                    raise ConsentError(
                        "judgment task_id does not match the canonical submission"
                    )
                self._ensure_active(connection, str(submission["participant_id"]))
                consent = self._require_consent(
                    connection,
                    payload.consent_id,
                    payload.participant_id,
                    "judgment",
                    str(submission["app_id"]),
                )
                if not consent_lineage_is_verified(_consent_snapshot(consent)):
                    raise ConsentError(
                        "consent receipt predates the complete ethical policy; "
                        "record a new consent receipt"
                    )
                judgment_id = _new_id("jdg")
                recorded_at = _utc_now()
                record = {
                    **request_data,
                    "judgment_id": judgment_id,
                    "recorded_at": recorded_at,
                    "schema_version": SCHEMA_VERSION,
                    "gold_item": False,
                    "gold_result": None,
                    "release_eligible": False,
                }
                connection.execute(
                    """
                    INSERT INTO judgments (
                        judgment_id, schema_version, participant_id, consent_id,
                        submission_id, task_id, choice, error_span, error_type,
                        correction, confidence, latency_ms, gold_item, gold_result,
                        recorded_at, release_eligible, request_sha256, record_json
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, NULL, ?, 0, ?, ?)
                    """,
                    (
                        judgment_id,
                        STORAGE_SCHEMA_VERSION,
                        payload.participant_id,
                        payload.consent_id,
                        payload.submission_id,
                        payload.task_id,
                        payload.choice,
                        payload.error_span,
                        payload.error_type,
                        payload.correction,
                        payload.confidence,
                        payload.latency_ms,
                        recorded_at,
                        request_sha256,
                        _canonical_json(record),
                    ),
                )
                response = {
                    "judgment_id": judgment_id,
                    "recorded_at": recorded_at,
                    "idempotent_replay": False,
                }
                self._record_idempotency(
                    connection,
                    operation,
                    idempotency_key,
                    request_sha256,
                    "judgment",
                    judgment_id,
                    response,
                    recorded_at,
                )
                connection.commit()
                return response, False
            except Exception:
                connection.rollback()
                raise

    def request_deletion(
        self,
        participant_id: str,
        payload: DeletionRequestCreate,
        idempotency_key: str,
        participant_token: str,
    ) -> tuple[dict[str, Any], bool]:
        self.initialize()
        participant_id = _validate_participant_id(participant_id)
        request_data = {
            "participant_id": participant_id,
            **payload.model_dump(mode="json"),
        }
        request_sha256 = _request_hash(request_data)
        operation = "request_deletion"
        # Serialize the canonical tombstone with both release and Pending Audit
        # delivery. Either a writer finishes before withdrawal is recorded and
        # is subsequently tombstoned, or it observes the tombstone and stops.
        with IEDID_DATASET_LOCK, PENDING_QUEUE_LOCK, self._connect() as connection:
            self._begin(connection)
            try:
                self._authenticate_participant(
                    connection,
                    participant_id,
                    participant_token,
                )
                replay = self._idempotent_replay(
                    connection, operation, idempotency_key, request_sha256
                )
                if replay is not None:
                    connection.commit()
                    return replay, True
                deletion_request_id = _new_id("del")
                requested_at = _utc_now()
                connection.execute("PRAGMA secure_delete = ON")
                erase_participant_audio(connection, participant_id, requested_at)
                record = {
                    **request_data,
                    "deletion_request_id": deletion_request_id,
                    "requested_at": requested_at,
                    "schema_version": SCHEMA_VERSION,
                    "status": "pending",
                    "tombstoned": True,
                    "hard_deleted": False,
                }
                connection.execute(
                    """
                    INSERT INTO deletion_requests (
                        deletion_request_id, schema_version, participant_id, reason,
                        scope, requested_at, status, tombstoned, hard_deleted,
                        request_sha256, record_json
                    ) VALUES (?, ?, ?, ?, ?, ?, 'pending', 1, 0, ?, ?)
                    """,
                    (
                        deletion_request_id,
                        STORAGE_SCHEMA_VERSION,
                        participant_id,
                        payload.reason,
                        payload.scope,
                        requested_at,
                        request_sha256,
                        _canonical_json(record),
                    ),
                )
                for target in (
                    "pending_audit_csv",
                    "huggingface_export",
                ):
                    action_id = _new_id("dpo")
                    action_payload = {
                        "deletion_request_id": deletion_request_id,
                        "participant_id": participant_id,
                        "target": target,
                        "action": "tombstone_participant_records",
                    }
                    connection.execute(
                        """
                        INSERT INTO deletion_propagation_outbox (
                            action_id, deletion_request_id, participant_id, target,
                            action, payload_json, created_at
                        ) VALUES (?, ?, ?, ?, 'tombstone_participant_records', ?, ?)
                        """,
                        (
                            action_id,
                            deletion_request_id,
                            participant_id,
                            target,
                            _canonical_json(action_payload),
                            requested_at,
                        ),
                    )
                consent_ids = [
                    row["consent_id"]
                    for row in connection.execute(
                        "SELECT consent_id FROM consents WHERE participant_id = ?",
                        (participant_id,),
                    ).fetchall()
                ]
                connection.execute(
                    "INSERT INTO iedid_deletion_outbox "
                    "(action_id, deletion_request_id, participant_id, "
                    "consent_ids_json, created_at) VALUES (?, ?, ?, ?, ?)",
                    (
                        _new_id("ido"),
                        deletion_request_id,
                        participant_id,
                        _canonical_json(consent_ids),
                        requested_at,
                    ),
                )
                response = {
                    "deletion_request_id": deletion_request_id,
                    "requested_at": requested_at,
                    "status": "pending",
                    "tombstoned": True,
                    "hard_deleted": False,
                    "local_pending_audit_status": "retry_pending",
                    "local_iedid_status": "retry_pending",
                    "remote_export_status": "propagation_pending",
                    "private_pending_mirror_status": "propagation_pending",
                    "propagation_pending": True,
                    "idempotent_replay": False,
                }
                self._record_idempotency(
                    connection,
                    operation,
                    idempotency_key,
                    request_sha256,
                    "deletion_request",
                    deletion_request_id,
                    response,
                    requested_at,
                )
                connection.commit()
                return response, False
            except Exception:
                connection.rollback()
                raise

    def health(self) -> dict[str, Any]:
        self.initialize()
        with self._connect() as connection:
            check = connection.execute("PRAGMA quick_check").fetchone()[0]
            journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0]
            version = connection.execute(
                "SELECT MAX(schema_version) FROM schema_meta"
            ).fetchone()[0]
        return {
            "status": "ok" if check == "ok" else "degraded",
            "schema_version": version,
            "journal_mode": journal_mode.lower(),
            "append_only": True,
            "raw_records_exposed": False,
            "storage_mode": (
                "persistent_configured"
                if _configured_db_path() is not None
                else "ephemeral_explicit_opt_in"
            ),
            "storage_durable": _configured_db_path() is not None,
        }

    def stats(self) -> dict[str, Any]:
        self.initialize()
        with self._connect() as connection:
            counts = {
                table: connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
                for table in (
                    "consents",
                    "submissions",
                    "judgments",
                    "deletion_requests",
                )
            }
            tombstoned = connection.execute(
                "SELECT COUNT(DISTINCT participant_id) FROM deletion_requests"
            ).fetchone()[0]
            pending_audit = connection.execute(
                """
                SELECT COUNT(*)
                FROM pending_audit_outbox AS outbox
                WHERE NOT EXISTS (
                    SELECT 1 FROM pending_audit_attempts AS attempts
                    WHERE attempts.outbox_id = outbox.outbox_id
                      AND attempts.succeeded = 1
                )
                """
            ).fetchone()[0]
            deletion_propagation = {
                target: count
                for target, count in connection.execute(
                    """
                    SELECT outbox.target, COUNT(*)
                    FROM deletion_propagation_outbox AS outbox
                    WHERE NOT EXISTS (
                        SELECT 1 FROM deletion_propagation_attempts AS attempts
                        WHERE attempts.action_id = outbox.action_id
                          AND attempts.succeeded = 1
                    )
                    GROUP BY outbox.target
                    """
                ).fetchall()
            }
            local_iedid_deletion_pending = connection.execute(
                """
                SELECT COUNT(*)
                FROM iedid_deletion_outbox AS outbox
                WHERE NOT EXISTS (
                    SELECT 1 FROM iedid_deletion_attempts AS attempts
                    WHERE attempts.action_id = outbox.action_id
                      AND attempts.succeeded = 1
                )
                """
            ).fetchone()[0]
        return {
            "schema_version": SCHEMA_VERSION,
            "records": counts,
            "tombstoned_participants": tombstoned,
            "release_eligible_records": 0,
            "pending_audit_retry_count": pending_audit,
            "deletion_propagation_pending": {
                "pending_audit_csv": deletion_propagation.get(
                    "pending_audit_csv", 0
                ),
                "huggingface_export": deletion_propagation.get(
                    "huggingface_export", 0
                ),
                "private_pending_mirror": deletion_propagation.get(
                    "huggingface_export", 0
                ),
                "local_iedid_csv": local_iedid_deletion_pending,
            },
        }


_STORE_REGISTRY: dict[Path, CollectionStore] = {}
_STORE_REGISTRY_LOCK = threading.Lock()


def get_store() -> CollectionStore:
    path = _default_db_path()
    with _STORE_REGISTRY_LOCK:
        store = _STORE_REGISTRY.get(path)
        if store is None:
            store = CollectionStore(path)
            _STORE_REGISTRY[path] = store
        return store


def _raise_api_error(exc: Exception) -> None:
    if isinstance(exc, StorageConfigurationError):
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=str(exc),
        )
    if isinstance(exc, IdempotencyConflict):
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
    if isinstance(exc, ParticipantTombstoned):
        raise HTTPException(status_code=status.HTTP_410_GONE, detail=str(exc))
    if isinstance(exc, ConsentError):
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
    if isinstance(exc, MissingResource):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    if isinstance(exc, ValueError):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)
        )
    if isinstance(exc, sqlite3.Error):
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Collection store is temporarily unavailable",
        )
    raise exc


@router.get("/consent-policies/current")
def current_consent_policy(consent_locale: str = "en") -> dict[str, Any]:
    """Expose the immutable text/hash clients must verify before consenting."""
    try:
        return get_current_consent_policy(consent_locale)
    except ValueError as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
        ) from exc


@router.post(
    "/consents",
    response_model=ConsentReceipt,
    status_code=status.HTTP_201_CREATED,
)
def create_consent(
    payload: ConsentCreate,
    response: Response,
    idempotency_key: IdempotencyKey,
    participant_token: ParticipantToken,
) -> dict[str, Any]:
    try:
        result, replayed = get_store().create_consent(
            payload,
            idempotency_key,
            participant_token,
        )
    except Exception as exc:
        _raise_api_error(exc)
    if replayed:
        response.status_code = status.HTTP_200_OK
    return result


@router.post(
    "/submissions",
    response_model=SubmissionReceipt,
    status_code=status.HTTP_201_CREATED,
)
def create_submission(
    payload: SubmissionCreate,
    response: Response,
    idempotency_key: IdempotencyKey,
    participant_token: ParticipantToken,
) -> dict[str, Any]:
    try:
        store = get_store()
        result, replayed = store.create_submission(
            payload,
            idempotency_key,
            participant_token,
        )
        result["pending_audit_status"] = store.mirror_submission(
            result["submission_id"]
        )
        store.drain_pending_audit(
            limit=10,
            exclude_submission_id=result["submission_id"],
        )
    except Exception as exc:
        _raise_api_error(exc)
    if replayed:
        response.status_code = status.HTTP_200_OK
    return result


@router.post(
    "/judgments",
    response_model=JudgmentReceipt,
    status_code=status.HTTP_201_CREATED,
)
def create_judgment(
    payload: JudgmentCreate,
    response: Response,
    idempotency_key: IdempotencyKey,
    participant_token: ParticipantToken,
) -> dict[str, Any]:
    try:
        result, replayed = get_store().create_judgment(
            payload,
            idempotency_key,
            participant_token,
        )
    except Exception as exc:
        _raise_api_error(exc)
    if replayed:
        response.status_code = status.HTTP_200_OK
    return result


@router.delete(
    "/participants/{participant_id}/data",
    response_model=DeletionReceipt,
    status_code=status.HTTP_202_ACCEPTED,
)
def request_participant_deletion(
    participant_id: str,
    response: Response,
    idempotency_key: IdempotencyKey,
    participant_token: ParticipantToken,
    payload: Annotated[DeletionRequestCreate | None, Body()] = None,
) -> dict[str, Any]:
    try:
        store = get_store()
        result, replayed = store.request_deletion(
            participant_id,
            payload or DeletionRequestCreate(),
            idempotency_key,
            participant_token,
        )
        local_tombstoned = store.propagate_deletion_to_pending_audit(
            result["deletion_request_id"]
        )
        result["local_pending_audit_status"] = (
            "tombstoned" if local_tombstoned else "retry_pending"
        )
        local_iedid_tombstoned = store.propagate_deletion_to_local_iedid(
            result["deletion_request_id"]
        )
        result["local_iedid_status"] = (
            "tombstoned" if local_iedid_tombstoned else "retry_pending"
        )
        # Remote publication deletion is an explicit outbox action consumed by
        # the separate HF export job; this API does not overstate completion.
        result["remote_export_status"] = "propagation_pending"
        result["private_pending_mirror_status"] = "propagation_pending"
        result["propagation_pending"] = True
    except Exception as exc:
        _raise_api_error(exc)
    if replayed:
        response.status_code = status.HTTP_200_OK
    return result


@router.get("/health")
def collection_health() -> dict[str, Any]:
    try:
        return get_store().health()
    except Exception as exc:
        _raise_api_error(exc)


@router.get("/stats")
def collection_stats() -> dict[str, Any]:
    try:
        return get_store().stats()
    except Exception as exc:
        _raise_api_error(exc)