Spaces:
Sleeping
Sleeping
| """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 | |
| ) | |
| def normalize_modalities(cls, value: list[str]) -> list[str]: | |
| return sorted(set(value)) | |
| 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 | |
| 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 | |
| ) | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| ) | |
| 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 | |
| def _begin(connection: sqlite3.Connection) -> None: | |
| connection.execute("BEGIN IMMEDIATE") | |
| 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 | |
| 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, | |
| ), | |
| ) | |
| 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" | |
| ) | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| def collection_health() -> dict[str, Any]: | |
| try: | |
| return get_store().health() | |
| except Exception as exc: | |
| _raise_api_error(exc) | |
| def collection_stats() -> dict[str, Any]: | |
| try: | |
| return get_store().stats() | |
| except Exception as exc: | |
| _raise_api_error(exc) | |