"""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" SQLITE_JOURNAL_MODE_ENV = "COLLECTION_SQLITE_JOURNAL_MODE" SUPPORTED_SQLITE_JOURNAL_MODES = frozenset({"WAL", "DELETE"}) 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"(? 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 _configured_sqlite_journal_mode() -> str: mode = os.environ.get(SQLITE_JOURNAL_MODE_ENV, "WAL").strip().upper() or "WAL" if mode not in SUPPORTED_SQLITE_JOURNAL_MODES: raise StorageConfigurationError( f"{SQLITE_JOURNAL_MODE_ENV} must be WAL for local POSIX storage or " "DELETE for a tested single-writer network volume" ) return mode 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") requested_journal_mode = _configured_sqlite_journal_mode() actual_journal_mode = connection.execute( f"PRAGMA journal_mode = {requested_journal_mode}" ).fetchone()[0] if str(actual_journal_mode).upper() != requested_journal_mode: connection.close() raise StorageConfigurationError( "The configured storage does not support the requested SQLite " f"journal mode {requested_journal_mode}" ) 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] configured_path = _configured_db_path() is not None path_ready = self.path.is_file() and check == "ok" 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": ( "configured_file_path" if configured_path else "ephemeral_explicit_opt_in" ), "storage_path_configured": configured_path, "storage_path_ready": path_ready, # A live request can verify only that this process can open and # check the configured file. Persistence across a Space restart # must be verified externally and is deliberately not claimed here. "persistence_verification": "external_restart_test_required", } 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)