src/endopath/storage.py

"""Storage layer for cases and checklist fields, dual-backend (issue #23).

SQLite by default (local dev and the test suite: zero setup, fast) and Postgres
when DATABASE_URL is set (the deployed Space, e.g. Supabase). Both go through
SQLAlchemy, and the hand-written SQL is kept portable so one code path serves
either backend: the DDL uses only TEXT/REAL/INTEGER, and the upserts use
`ON CONFLICT (...) DO UPDATE SET x = excluded.x`, which SQLite and Postgres both
accept. Which backend a call uses is decided by _resolve_url: an explicit
db_path (what the tests pass) is always SQLite; the default path plus a
DATABASE_URL is Postgres. So tests and local dev never touch Postgres by
accident.

A normalized fields table (rather than one JSON blob per case) so the worklist
can answer queries like "everything with a low-confidence FIGO grade" (docs/prd.md
section 8.1's own named example) directly in SQL.
"""

from __future__ import annotations

import json
import logging
import os
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

from sqlalchemy import Connection, Engine, create_engine, event, text
from sqlalchemy.exc import OperationalError, ProgrammingError

from endopath import mapping
from endopath.schema import (
    Case,
    CaseStatus,
    ChecklistField,
    EndometrialChecklist,
    EvidenceSpan,
    FieldStatus,
    FigoStagingEdition,
    Report,
    ReportType,
    Specimen,
)

logger = logging.getLogger(__name__)

# How an observation reached the store, so the checklist-field view and the export read only the
# atoms the extraction pass produced, and the induction pass's atoms (#107) sit beside them in the
# same table without being mistaken for the confirmed checklist value.
EXTRACTION_METHOD_CHECKLIST = "checklist_extraction"
EXTRACTION_METHOD_INDUCTION = "report_induction"

# Distinguishes "no derived value passed" from "the derived value is None (an absent field)".
_UNSET = object()

DEFAULT_DB_PATH = Path("data") / "app.db"

# Portable across SQLite and Postgres: only TEXT/REAL/INTEGER, IF NOT EXISTS,
# a composite PK, a FK, and a plain index. Executed statement by statement
# (SQLite's executescript-style multi-statement blob isn't portable).
_SCHEMA_STATEMENTS = [
    """
    CREATE TABLE IF NOT EXISTS cases (
        case_barcode TEXT PRIMARY KEY,
        patient_filename TEXT,
        report_date TEXT,
        staging_edition TEXT NOT NULL,
        status TEXT NOT NULL
    )
    """,
    # A project ("repo"): the dataset a team labels. One row per project. Cases and accounts scope
    # to a project through their migrated `project_id` column; the corpus that predates this feature belongs
    # to the seeded default project (`DEFAULT_PROJECT_ID`), backfilled at schema init so the existing 548
    # cases keep working. `created_by` is the account that created the project, or 'system' for the seeded
    # default.
    """
    CREATE TABLE IF NOT EXISTS projects (
        project_id TEXT PRIMARY KEY,
        name TEXT NOT NULL,
        created_by TEXT,
        created_at TEXT NOT NULL
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS fields (
        case_barcode TEXT NOT NULL REFERENCES cases(case_barcode),
        field_name TEXT NOT NULL,
        value_json TEXT,
        confidence REAL,
        evidence_json TEXT,
        status TEXT NOT NULL,
        applicable INTEGER NOT NULL,
        not_applicable_reason TEXT,
        search_log_json TEXT NOT NULL,
        concept_id TEXT,
        PRIMARY KEY (case_barcode, field_name)
    )
    """,
    "CREATE INDEX IF NOT EXISTS idx_fields_lookup ON fields (field_name, status, confidence)",
    # The report/specimen grain (#110): a case assembles one or more reports (biopsy, hysterectomy,
    # amendment, external consult, ancillary), each with one or more specimens. Additive alongside
    # `cases`/`fields`: the `fields` table above still holds the resolved report's checklist values,
    # addressed by case_barcode alone; these tables persist which reports and specimens exist, their
    # type, institution, date, amendment lineage, and reviewer-designated authoritative report.
    """
    CREATE TABLE IF NOT EXISTS reports (
        report_id TEXT PRIMARY KEY,
        case_barcode TEXT NOT NULL REFERENCES cases(case_barcode),
        report_type TEXT NOT NULL,
        institution TEXT,
        report_date TEXT,
        staging_edition TEXT NOT NULL,
        supersedes_report_id TEXT,
        is_authoritative INTEGER,
        report_text TEXT
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS specimens (
        specimen_id TEXT PRIMARY KEY,
        report_id TEXT NOT NULL REFERENCES reports(report_id),
        procedure TEXT
    )
    """,
    "CREATE INDEX IF NOT EXISTS idx_reports_case ON reports (case_barcode)",
    "CREATE INDEX IF NOT EXISTS idx_specimens_report ON specimens (report_id)",
    # The observation atom (#111): the durable, queryable value a source stated, keyed to the grain
    # (#110). One row per InducedField / ChecklistField, so the primitive the report stated survives as
    # its own row and the checklist field becomes a view derived from it (CLAUDE.md invariant 1). The
    # row shape is reconciled with induction.InducedField: `raw_wording` is its `value_or_absent`,
    # `absence_reason` its `reason_if_absent`, `page`/`bbox` its `span`, `confidence` its `confidence`.
    # `normalized_value` records the controlled representation the extraction pass coerced the wording
    # into (invariant 1: a category is derived from the primitive, never stored in its place), with
    # `value_kind` naming the transformation (`direct` surface wording, `normalized` controlled value;
    # the column is open for `derived` and later kinds #113 adds). `absence_reason` draws from the
    # closed induction.AbsenceReason vocabulary; expanding it (`constrained`, `historically_unrecoverable`)
    # is #112, not here. `absence_detail` carries the human-readable gate condition a coded absence
    # names. Evidence rides the row itself (`page`, `bbox`, `quote`, `source_adapter`), so no value is
    # stored without it (invariant 2). report_id references the grain's report; specimen_key names the
    # specimen within it and is not FK-constrained, matching EvidenceSpan's optional specimen ref.
    """
    CREATE TABLE IF NOT EXISTS observations (
        observation_id TEXT PRIMARY KEY,
        case_barcode TEXT NOT NULL REFERENCES cases(case_barcode),
        report_id TEXT NOT NULL REFERENCES reports(report_id),
        specimen_key TEXT,
        concept_id TEXT,
        field_name TEXT NOT NULL,
        raw_wording TEXT,
        normalized_value TEXT,
        units TEXT,
        value_kind TEXT,
        absence_reason TEXT,
        absence_detail TEXT,
        page INTEGER,
        bbox TEXT,
        quote TEXT,
        source_adapter TEXT,
        extraction_method TEXT,
        confidence REAL,
        review_status TEXT,
        rule_id TEXT,
        rule_version TEXT,
        created_at TEXT NOT NULL
    )
    """,
    "CREATE INDEX IF NOT EXISTS idx_observations_case_field ON observations (case_barcode, field_name)",
    "CREATE INDEX IF NOT EXISTS idx_observations_report ON observations (report_id)",
    "CREATE INDEX IF NOT EXISTS idx_observations_method ON observations (extraction_method)",
    # Where a confirmed crosswalk edge's provenance lives is an open decision recorded on #93 (grow
    # #35's crosswalk.csv schema, or a dedicated table). This table is the provisional home the
    # mapping-review surface (#95, #103) writes to; it does not resolve that decision. Keyed by
    # mapping_id, one decision record per edge, carrying the reviewer, their role, the resolved concept
    # and relation, the per-variable licence-required flag (docs/prd.md section 4.4), and the time.
    #
    # It doubles as the decision ledger (#103): each row records the disposition (a confirmed
    # resolution, or a flag or a deferral that parks the edge without resolving it), how the answer was
    # reached (decided_via: an offered candidate, a written-in resolution, or a resolved chat), the
    # question the decision answered, and the options considered, so a second reviewer reconstructs how
    # every edge reached its state. Amending a decision upserts the same row.
    """
    CREATE TABLE IF NOT EXISTS mapping_confirmations (
        mapping_id TEXT PRIMARY KEY,
        source_std TEXT NOT NULL,
        source_field TEXT NOT NULL,
        concept_id TEXT,
        relation TEXT NOT NULL,
        predicate TEXT NOT NULL,
        confirmed_by TEXT NOT NULL,
        role TEXT NOT NULL,
        note TEXT,
        licence_required INTEGER NOT NULL,
        confirmed_at TEXT NOT NULL,
        disposition TEXT NOT NULL DEFAULT 'confirmed',
        decided_via TEXT,
        question TEXT,
        options_json TEXT
    )
    """,
    # A confirmed split's component edges (#103 follow-on). One source field can resolve to several
    # canonical concepts at once: molecular_classification, for example, fans out to the POLE, MMR, and
    # p53 primitives, with the ProMisE class kept as the derived headline. The field's headline decision
    # stays in mapping_confirmations; each component concept it also resolves to is one row here, keyed by
    # the parent mapping_id and the component concept. The rows for a mapping_id are replaced wholesale
    # each time its split is re-applied, so a later plain confirmation clears them.
    """
    CREATE TABLE IF NOT EXISTS mapping_confirmation_components (
        mapping_id TEXT NOT NULL,
        source_std TEXT NOT NULL,
        concept_id TEXT NOT NULL,
        relation TEXT NOT NULL,
        predicate TEXT NOT NULL,
        note TEXT,
        PRIMARY KEY (mapping_id, concept_id)
    )
    """,
    # The decision ledger for the case-review confirmation surface (#98), the same
    # source-to-decision trail the crosswalk keeps in mapping_confirmations (#103): each row records one
    # reviewer's decision on one checklist field. Keyed by (case_barcode, field_name), one row per
    # decided field, so amending a decision upserts the same row. It carries the value the reviewer
    # confirmed or edited into, the resulting field status, how the answer was reached (decided_via:
    # accepted the drafted value, edited it, or resolved it through the value chat), the reviewer and
    # their role from the session (docs/prd.md section 4.4), and the question and options the decision
    # weighed. A value counts as data only through a row here (CLAUDE.md invariant).
    """
    CREATE TABLE IF NOT EXISTS field_confirmations (
        case_barcode TEXT NOT NULL,
        field_name TEXT NOT NULL,
        value_json TEXT,
        status TEXT NOT NULL,
        disposition TEXT NOT NULL DEFAULT 'confirmed',
        decided_via TEXT,
        confirmed_by TEXT NOT NULL,
        role TEXT NOT NULL,
        note TEXT,
        question TEXT,
        options_json TEXT,
        confirmed_at TEXT NOT NULL,
        PRIMARY KEY (case_barcode, field_name)
    )
    """,
    # The target dictionary an Intake submission points at, stored so the compilation worker reads it
    # back by id after the request that submitted it has returned (#91). variables_json holds the
    # DictionaryVariable list. Replaces dictionary.py's process-level registry.
    """
    CREATE TABLE IF NOT EXISTS dictionaries (
        id TEXT PRIMARY KEY,
        title TEXT NOT NULL,
        variables_json TEXT NOT NULL,
        created_at TEXT NOT NULL
    )
    """,
    # Reviewer accounts (#128): the write endpoints authenticate a per-user login rather than a single
    # shared token, so a confirmation's recorded confirmer and role are the authenticated account's, not a
    # self-declared field (PRD section 4.4). `role` and `holds_licence` are account attributes set at
    # provisioning: a licence_required edge or field can only be confirmed by an account whose
    # holds_licence is set. password_hash is a self-describing scrypt string (endopath.auth); the plaintext
    # is never stored. Seeded from the ENDOPATH_SEED_USERS secret at startup.
    """
    CREATE TABLE IF NOT EXISTS users (
        username TEXT PRIMARY KEY,
        password_hash TEXT NOT NULL,
        role TEXT NOT NULL,
        holds_licence INTEGER NOT NULL,
        display_name TEXT,
        created_at TEXT NOT NULL
    )
    """,
    # The Intake follow-up queue, durable so a submission's jobs survive the request and a restart
    # (#92, #91). One row per enqueued job, claimed and drained by the in-process worker. status moves
    # enqueued -> running -> done or failed; error carries the failure text. Replaces jobs.py's
    # process-level queue. id is a uuid string so the insert stays portable across SQLite and Postgres
    # without an autoincrement column.
    """
    CREATE TABLE IF NOT EXISTS jobs (
        id TEXT PRIMARY KEY,
        kind TEXT NOT NULL,
        ticket TEXT NOT NULL,
        status TEXT NOT NULL,
        dictionary_id TEXT NOT NULL,
        source_kind TEXT NOT NULL,
        case_count INTEGER NOT NULL,
        variable_count INTEGER NOT NULL,
        error TEXT,
        created_at TEXT NOT NULL,
        started_at TEXT,
        finished_at TEXT
    )
    """,
    "CREATE INDEX IF NOT EXISTS idx_jobs_claim ON jobs (status, created_at)",
    # What a follow-up job produced for one dictionary submission: the induced report schema (#92) and
    # the drafted crosswalk (#91, #93), each a strict-JSON payload the mapping surface reads back. Keyed
    # by (dictionary_id, kind), one row per artifact kind, so re-running a submission upserts in place.
    """
    CREATE TABLE IF NOT EXISTS submission_artifacts (
        dictionary_id TEXT NOT NULL,
        kind TEXT NOT NULL,
        payload_json TEXT NOT NULL,
        created_at TEXT NOT NULL,
        PRIMARY KEY (dictionary_id, kind)
    )
    """,
    # Longitudinal records (#177): GDC clinical events (follow-ups, treatments, recurrence, vital status)
    # per participant, one layer of the fuller patient record beside the pathology checklist. Kept in its
    # own tables so it never touches the extraction path or the two invariants; a case-review panel reads it
    # as completeness-and-provenance framing (what the record contains and where it came from), not a value
    # in the checklist. Empty on a deployment that has not ingested it, so the layer degrades to nothing.
    # structured_data is the raw GDC JSON, kept for provenance.
    """
    CREATE TABLE IF NOT EXISTS longitudinal_records (
        record_id TEXT PRIMARY KEY,
        participant_id TEXT NOT NULL,
        case_barcode TEXT,
        record_type TEXT NOT NULL,
        record_index INTEGER NOT NULL,
        days_to_followup INTEGER,
        days_to_treatment INTEGER,
        days_to_recurrence INTEGER,
        days_to_lost_to_followup INTEGER,
        biopsy_result TEXT,
        followup_treatment_success TEXT,
        recurrence_type TEXT,
        therapy_type TEXT,
        lost_to_followup TEXT,
        structured_data TEXT,
        ingested_at TEXT NOT NULL
    )
    """,
    "CREATE INDEX IF NOT EXISTS idx_longitudinal_case ON longitudinal_records (case_barcode)",
    "CREATE INDEX IF NOT EXISTS idx_longitudinal_participant ON longitudinal_records (participant_id)",
    "CREATE INDEX IF NOT EXISTS idx_longitudinal_type ON longitudinal_records (record_type)",
    # A reviewer's decision that a longitudinal record belongs with a case (#176). disposition is confirmed,
    # rejected, pending, or uncertain; confirmed_by/role/confirmed_at attribute it the same way
    # field_confirmations does. Empty until a reviewer joins one.
    """
    CREATE TABLE IF NOT EXISTS longitudinal_joins (
        join_id TEXT PRIMARY KEY,
        case_barcode TEXT NOT NULL,
        record_id TEXT NOT NULL,
        disposition TEXT NOT NULL,
        confirmed_by TEXT,
        role TEXT,
        confirmed_at TEXT,
        note TEXT,
        question TEXT
    )
    """,
    # The file record types a project draws its values from (#237): each is a source of extracted data,
    # with a source_class for grouping/colour and a blurb. The coverage view maps each target-dictionary
    # field to the record type that fulfils it, so the hardcoded source strings become switchable DB rows.
    """
    CREATE TABLE IF NOT EXISTS record_types (
        project_id TEXT NOT NULL,
        record_type_id TEXT NOT NULL,
        name TEXT NOT NULL,
        source_class TEXT NOT NULL,
        blurb TEXT,
        sort_order INTEGER NOT NULL DEFAULT 0,
        PRIMARY KEY (project_id, record_type_id)
    )
    """,
    # A project's target dictionary, one row per field (#237): the field, its section, the record type that
    # fulfils it, and how its confirmed fill is measured (fill_via 'checklist'|'outcome' + fill_key). This is
    # the DB-backed field coverage that replaces the generated frontend data file; a different project or
    # dictionary_id is a different format to review.
    """
    CREATE TABLE IF NOT EXISTS coverage_fields (
        project_id TEXT NOT NULL,
        dictionary_id TEXT NOT NULL,
        name TEXT NOT NULL,
        label TEXT NOT NULL,
        section TEXT,
        record_type_id TEXT,
        fill_via TEXT,
        fill_key TEXT,
        sort_order INTEGER NOT NULL DEFAULT 0,
        PRIMARY KEY (project_id, dictionary_id, name)
    )
    """,
    # A record type's extracted dictionary, populated (#237, Phase B): one row per (case, field) read from
    # that source's documents, with the verbatim evidence snippet. Seeded in-conversation from committed,
    # de-identified samples, not through the extraction API. evidence is NULL for a coded absence.
    """
    CREATE TABLE IF NOT EXISTS record_type_extractions (
        project_id TEXT NOT NULL,
        record_type_id TEXT NOT NULL,
        case_barcode TEXT NOT NULL,
        field TEXT NOT NULL,
        value TEXT NOT NULL,
        evidence TEXT,
        sort_order INTEGER NOT NULL DEFAULT 0,
        PRIMARY KEY (project_id, record_type_id, case_barcode, field)
    )
    """,
]

# Ledger columns added to mapping_confirmations after its first shape (#103). Run once per engine,
# each guarded on its own, so a database created under the earlier schema gains them and a database
# already carrying them is left unchanged. SQLite has no ADD COLUMN IF NOT EXISTS, so a duplicate-column
# error is the expected no-op here, not a failure.
_MIGRATION_STATEMENTS = [
    "ALTER TABLE mapping_confirmations ADD COLUMN disposition TEXT NOT NULL DEFAULT 'confirmed'",
    "ALTER TABLE mapping_confirmations ADD COLUMN decided_via TEXT",
    "ALTER TABLE mapping_confirmations ADD COLUMN question TEXT",
    "ALTER TABLE mapping_confirmations ADD COLUMN options_json TEXT",
    # #39: the OCR-text and concept-linkage columns added to the reports/fields tables #110 landed.
    "ALTER TABLE reports ADD COLUMN report_text TEXT",
    "ALTER TABLE fields ADD COLUMN concept_id TEXT",
    # The project ("repo") scope: cases and accounts each belong to one project. A database created before
    # this feature gains the column here; `_seed_default_project` then backfills every existing row to the
    # default project, so the pre-existing corpus keeps working.
    "ALTER TABLE cases ADD COLUMN project_id TEXT",
    "ALTER TABLE users ADD COLUMN project_id TEXT",
]

# The default project every pre-existing case and account belongs to, seeded at schema init so the corpus
# that predates the project scope keeps working.
DEFAULT_PROJECT_ID = "endometrial"
DEFAULT_PROJECT_NAME = "TCGA-UCEC"

_engines: dict[str, Engine] = {}
_initialized: set[str] = set()


def _postgres_url(raw: str) -> str:
    """Normalize a Supabase/Postgres connection string to the psycopg (v3)
    SQLAlchemy driver, accepting either postgres:// or postgresql://."""
    if raw.startswith("postgres://"):
        raw = "postgresql://" + raw[len("postgres://") :]
    if raw.startswith("postgresql://"):
        raw = "postgresql+psycopg://" + raw[len("postgresql://") :]
    return raw


def _resolve_url(db_path: Path | str) -> str:
    database_url = os.environ.get("DATABASE_URL")
    if database_url and Path(db_path) == DEFAULT_DB_PATH:
        return _postgres_url(database_url)
    return f"sqlite:///{db_path}"


def _init_schema(engine: Engine, *, resilient: bool) -> None:
    """Run the schema DDL. Every statement is `CREATE ... IF NOT EXISTS`, so idempotent.

    On SQLite the statements run in one transaction and any error is fatal: a fresh local database must
    initialize fully or fail loudly. On Postgres (`resilient`) each runs in its own transaction under the
    connection's short lock_timeout, and a lock or statement timeout is logged and skipped rather than
    crashing the boot. A timeout means the object already exists and a leaked idle-in-transaction write
    held its lock (#195), so continuing is safe; a genuinely fresh Postgres database has no such contention,
    so every statement runs.
    """
    if not resilient:
        with engine.begin() as conn:
            for statement in _SCHEMA_STATEMENTS:
                conn.execute(text(statement))
        return
    for statement in _SCHEMA_STATEMENTS:
        try:
            with engine.begin() as conn:
                conn.execute(text(statement))
        except OperationalError as exc:
            logger.warning(
                "schema-init statement blocked or timed out; the object is idempotent, continuing "
                "without it: %s",
                str(exc).splitlines()[0],
            )


def _get_engine(url: str) -> Engine:
    if url not in _engines:
        if url.startswith("sqlite"):
            # check_same_thread=False: the API's threadpool creates and closes a
            # connection across worker threads; each request still gets its own.
            engine = create_engine(url, connect_args={"check_same_thread": False})

            @event.listens_for(engine, "connect")
            def _enable_sqlite_fk(dbapi_conn, _record):  # noqa: ANN001
                # SQLite does not enforce foreign keys unless asked per-connection
                # (Postgres enforces them by default, so this is SQLite-only).
                cur = dbapi_conn.cursor()
                cur.execute("PRAGMA foreign_keys=ON")
                cur.close()

        else:
            # pool_pre_ping guards against a pooled Postgres connection that the
            # server (or Supabase's pooler) has since dropped. lock_timeout makes a
            # blocked statement fail fast instead of waiting past the server statement
            # timeout, and idle_in_transaction_session_timeout self-terminates one of
            # our own connections left idle mid-transaction, so a leaked write lock
            # frees itself instead of blocking the next boot's schema-init (#195).
            engine = create_engine(
                url,
                pool_pre_ping=True,
                connect_args={
                    "options": "-c lock_timeout=3000 -c idle_in_transaction_session_timeout=300000"
                },
            )
        _engines[url] = engine

    engine = _engines[url]
    if url not in _initialized:
        _init_schema(engine, resilient=not url.startswith("sqlite"))
        for statement in _MIGRATION_STATEMENTS:
            # Each in its own transaction: a duplicate-column error on one aborts only that statement,
            # not the columns that migrated cleanly before it.
            try:
                with engine.begin() as conn:
                    conn.execute(text(statement))
            except (OperationalError, ProgrammingError):
                pass
        # After the project_id columns are migrated on: seed the default project and backfill every
        # pre-existing case and account into it. Idempotent, so this runs safely on every startup.
        with engine.begin() as conn:
            _seed_default_project(conn)
        _initialized.add(url)
    return engine


def _seed_default_project(conn: Connection) -> None:
    """Ensure the default project row exists and every case and account belongs to it. Idempotent: the
    insert is a no-op when the row is already present, and the backfills only touch rows whose project_id
    is still null. Portable across SQLite and Postgres."""
    conn.execute(
        text(
            "INSERT INTO projects (project_id, name, created_by, created_at) "
            "VALUES (:project_id, :name, 'system', :created_at) "
            "ON CONFLICT (project_id) DO NOTHING"
        ),
        {"project_id": DEFAULT_PROJECT_ID, "name": DEFAULT_PROJECT_NAME, "created_at": _utc_now_iso()},
    )
    conn.execute(
        text("UPDATE cases SET project_id = :project_id WHERE project_id IS NULL"),
        {"project_id": DEFAULT_PROJECT_ID},
    )
    conn.execute(
        text("UPDATE users SET project_id = :project_id WHERE project_id IS NULL"),
        {"project_id": DEFAULT_PROJECT_ID},
    )


def connect(db_path: Path | str = DEFAULT_DB_PATH) -> Connection:
    url = _resolve_url(db_path)
    if url.startswith("sqlite"):
        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
    return _get_engine(url).connect()


# --- the observation atom (#111) ------------------------------------------------------------------
# One durable, queryable row per stated value. The checklist field is a view over these (invariant 1),
# and the export reads them rather than the collapsed `fields` cache. The shape is reconciled with
# induction.InducedField: `from_checklist_field` maps an extracted ChecklistField and `from_induced_field`
# maps an InducedField into the one row, so the two passes share a single observation shape.

# The absence codes an extracted field maps onto, a subset of induction.AbsenceReason. A gated field is
# `not_applicable`; a flagged field was searched for and confirmed absent (`asked-unknown`). An applicable
# field with no value yet drafted carries no code: it is unsettled, not a confirmed absence, and coding it
# would over-claim. Widening this mapping (`indeterminate`, `constrained`) is #112.
_ABSENCE_NOT_APPLICABLE = "not_applicable"
_ABSENCE_ASKED_UNKNOWN = "asked-unknown"


def _observation_id(
    case_barcode: str,
    report_id: str,
    specimen_key: Optional[str],
    field_name: str,
    value_kind: Optional[str],
    extraction_method: str,
) -> str:
    """A deterministic id, so re-persisting the same case replaces its observation rows in place rather
    than duplicating them. The value_kind and extraction_method are part of the key: a field's `direct`
    and `normalized` observations, and its extracted and induced observations, are distinct rows."""
    parts = [case_barcode, report_id, specimen_key or "", field_name, value_kind or "absent", extraction_method]
    return "::".join(parts)


def _absence_reason_for_field(field: ChecklistField) -> Optional[str]:
    if not field.applicable:
        return _ABSENCE_NOT_APPLICABLE
    if field.status is FieldStatus.FLAGGED:
        return _ABSENCE_ASKED_UNKNOWN
    return None


def _primitive(value: object) -> object:
    """An enum to its stored value, everything else unchanged, so a row carries JSON primitives."""
    return value.value if hasattr(value, "value") else value


@dataclass(frozen=True)
class Observation:
    """One stored observation: the primitive a source stated, its controlled representation, the evidence
    it rode in on, and the grain it attaches to. The columns of the `observations` table, in one shape both
    the extraction and induction passes build."""

    observation_id: str
    case_barcode: str
    report_id: str
    specimen_key: Optional[str]
    concept_id: Optional[str]
    field_name: str
    raw_wording: Optional[str]
    normalized_value: Optional[str]
    units: Optional[str]
    value_kind: Optional[str]
    absence_reason: Optional[str]
    absence_detail: Optional[str]
    page: Optional[int]
    bbox: Optional[str]
    quote: Optional[str]
    source_adapter: Optional[str]
    extraction_method: str
    confidence: Optional[float]
    review_status: Optional[str]
    rule_id: Optional[str]
    rule_version: Optional[str]
    created_at: str

    @classmethod
    def from_checklist_field(
        cls,
        field: ChecklistField,
        *,
        case_barcode: str,
        report_id: str,
        specimen_key: Optional[str],
        field_name: str,
        concept_id: Optional[str],
        created_at: str,
    ) -> "Observation":
        """The observation an extracted checklist field states. The controlled value the pass coerced to
        lands in `normalized_value`; the verbatim evidence snippet is `raw_wording` (invariant 1), and the
        evidence span rides `page`/`quote`/`source_adapter` (invariant 2). An absent field carries its coded
        reason and the gate's human-readable detail instead of a value."""
        evidence = field.evidence
        primitive = _primitive(field.value)
        quote = evidence.quote if evidence else None
        page = evidence.page_number if evidence else None
        source_adapter = evidence.source if evidence else None
        if primitive is not None:
            normalized = json.dumps(primitive)
            raw_wording = quote if quote is not None else str(primitive)
            # `normalized` when the surface wording and the controlled value differ (the usual case: a
            # sentence coerced to an enum or a number); `direct` when the stored value is the wording.
            value_kind = "normalized" if quote is not None and quote != str(primitive) else "direct"
            absence_reason: Optional[str] = None
            absence_detail: Optional[str] = None
        else:
            normalized = None
            raw_wording = None
            value_kind = None
            absence_reason = _absence_reason_for_field(field)
            absence_detail = field.not_applicable_reason
        return cls(
            observation_id=_observation_id(
                case_barcode, report_id, specimen_key, field_name, value_kind, EXTRACTION_METHOD_CHECKLIST
            ),
            case_barcode=case_barcode,
            report_id=report_id,
            specimen_key=specimen_key,
            concept_id=concept_id,
            field_name=field_name,
            raw_wording=raw_wording,
            normalized_value=normalized,
            units=None,
            value_kind=value_kind,
            absence_reason=absence_reason,
            absence_detail=absence_detail,
            page=page,
            bbox=None,
            quote=quote,
            source_adapter=source_adapter,
            extraction_method=EXTRACTION_METHOD_CHECKLIST,
            confidence=field.confidence,
            review_status=field.status.value,
            rule_id=None,
            rule_version=None,
            created_at=created_at,
        )

    @classmethod
    def from_induced_field(
        cls,
        induced: Any,
        *,
        case_barcode: str,
        report_id: str,
        concept_id: Optional[str] = None,
        created_at: str,
    ) -> "Observation":
        """The observation an induced field (induction.InducedField) states. Its `value_or_absent` is the
        surface wording (a direct observation, `raw_wording`), its `span` the evidence page and normalized
        bbox, its `reason_if_absent` the coded absence. Typed loosely so this module does not import the
        induction stack (and its anthropic dependency)."""
        span = induced.span
        specimen_key = span.specimen_id if span is not None else None
        page = span.page if span is not None else None
        bbox = json.dumps(list(span.bbox)) if span is not None else None
        present = induced.value_or_absent is not None
        value_kind = "direct" if present else None
        reason = induced.reason_if_absent
        absence_reason = reason.value if reason is not None else None
        return cls(
            observation_id=_observation_id(
                case_barcode, report_id, specimen_key, induced.name, value_kind, EXTRACTION_METHOD_INDUCTION
            ),
            case_barcode=case_barcode,
            report_id=report_id,
            specimen_key=specimen_key,
            concept_id=concept_id,
            field_name=induced.name,
            raw_wording=induced.value_or_absent,
            normalized_value=None,
            units=None,
            value_kind=value_kind,
            absence_reason=absence_reason,
            absence_detail=None,
            page=page,
            bbox=bbox,
            quote=induced.value_or_absent if present else None,
            source_adapter="induction",
            extraction_method=EXTRACTION_METHOD_INDUCTION,
            confidence=induced.confidence,
            review_status=FieldStatus.NEEDS_REVIEW.value,
            rule_id=None,
            rule_version=None,
            created_at=created_at,
        )


_OBSERVATION_COLUMNS = (
    "observation_id",
    "case_barcode",
    "report_id",
    "specimen_key",
    "concept_id",
    "field_name",
    "raw_wording",
    "normalized_value",
    "units",
    "value_kind",
    "absence_reason",
    "absence_detail",
    "page",
    "bbox",
    "quote",
    "source_adapter",
    "extraction_method",
    "confidence",
    "review_status",
    "rule_id",
    "rule_version",
    "created_at",
)


def insert_observation(conn: Connection, observation: Observation) -> None:
    """Upsert one observation by its deterministic id. Portable across SQLite and Postgres."""
    non_pk = [c for c in _OBSERVATION_COLUMNS if c != "observation_id"]
    columns = ", ".join(_OBSERVATION_COLUMNS)
    placeholders = ", ".join(f":{c}" for c in _OBSERVATION_COLUMNS)
    updates = ", ".join(f"{c} = excluded.{c}" for c in non_pk)
    conn.execute(
        text(
            f"INSERT INTO observations ({columns}) VALUES ({placeholders}) "
            f"ON CONFLICT (observation_id) DO UPDATE SET {updates}"
        ),
        {c: getattr(observation, c) for c in _OBSERVATION_COLUMNS},
    )


def _delete_checklist_observations(conn: Connection, case_barcode: str) -> None:
    """Clear a case's extraction-pass observations before re-writing them, so a field whose value or
    value_kind changed does not leave a stale row behind. Scoped to the checklist pass, so an induced
    observation (#107) beside it is left untouched."""
    conn.execute(
        text(
            "DELETE FROM observations WHERE case_barcode = :cb AND extraction_method = :method"
        ),
        {"cb": case_barcode, "method": EXTRACTION_METHOD_CHECKLIST},
    )


def resolved_report_id(conn: Connection, case_barcode: str) -> Optional[str]:
    """The report id an induced observation attaches to: the sole report, or the one a reviewer
    designated authoritative (issue #110). Returns None on a genuinely ambiguous multi-report case
    rather than raising, so the induction pass records the contract and skips the observation write
    instead of failing the family."""
    rows = (
        conn.execute(
            text(
                "SELECT report_id, is_authoritative FROM reports WHERE case_barcode = :cb ORDER BY report_id"
            ),
            {"cb": case_barcode},
        )
        .mappings()
        .fetchall()
    )
    if not rows:
        return None
    if len(rows) == 1:
        return rows[0]["report_id"]
    authoritative = [row["report_id"] for row in rows if row["is_authoritative"]]
    return authoritative[0] if len(authoritative) == 1 else None


def delete_induced_observations(conn: Connection, case_barcode: str) -> None:
    """Clear a case's induction-pass observations before re-writing them, so re-running the induction
    job replaces them in place. Scoped to the induction pass, so the checklist observations are untouched."""
    conn.execute(
        text("DELETE FROM observations WHERE case_barcode = :cb AND extraction_method = :method"),
        {"cb": case_barcode, "method": EXTRACTION_METHOD_INDUCTION},
    )


def observations_for_case(
    conn: Connection, case_barcode: str, *, extraction_method: Optional[str] = None
) -> list[dict]:
    """Every stored observation for one case, newest grain first, optionally filtered to one pass."""
    query = "SELECT * FROM observations WHERE case_barcode = :cb"
    params: dict = {"cb": case_barcode}
    if extraction_method is not None:
        query += " AND extraction_method = :method"
        params["method"] = extraction_method
    query += " ORDER BY field_name"
    return [dict(row) for row in conn.execute(text(query), params).mappings().fetchall()]


def derive_field_value(observation_rows: list[dict]) -> Optional[Any]:
    """The checklist field's value as a view over its observations (#111). The controlled
    `normalized_value` when the pass produced one, else the `raw_wording`, else absent. The field cache
    and the export both read the value through here, so neither re-stores a value the observations do not
    carry."""
    for row in observation_rows:
        if row.get("normalized_value") is not None:
            return json.loads(row["normalized_value"])
    for row in observation_rows:
        if row.get("raw_wording") is not None:
            return row["raw_wording"]
    return None


def upsert_case(
    conn: Connection,
    case: Case,
    patient_filename: Optional[str] = None,
    *,
    project_id: str = DEFAULT_PROJECT_ID,
) -> None:
    # A new case lands in `project_id` (the default project unless a caller names another), so a
    # freshly-ingested case is scoped the same way the backfilled corpus is. project_id is intentionally
    # left out of the ON CONFLICT update: re-upserting a case (every field confirm and edit does) preserves
    # the project it already belongs to rather than resetting it to the default.
    conn.execute(
        text(
            """
            INSERT INTO cases (
                case_barcode, patient_filename, report_date, staging_edition, status, project_id
            )
            VALUES (
                :case_barcode, :patient_filename, :report_date, :staging_edition, :status, :project_id
            )
            ON CONFLICT (case_barcode) DO UPDATE SET
                patient_filename = excluded.patient_filename,
                report_date = excluded.report_date,
                staging_edition = excluded.staging_edition,
                status = excluded.status
            """
        ),
        {
            "case_barcode": case.case_barcode,
            "patient_filename": patient_filename,
            "report_date": case.report_date.isoformat() if case.report_date else None,
            "staging_edition": case.staging_edition.value,
            "status": case.status.value,
            "project_id": project_id,
        },
    )

    # Persist the reports/specimens first: an observation references its report (FK), and the resolved
    # report is the grain each checklist observation attaches to. case.checklist below already
    # auto-vivifies a default report/specimen if case.reports was empty (issue #110), so this always
    # persists at least one report.
    for report in case.reports:
        _upsert_report(conn, case.case_barcode, report)

    # The grain the checklist observations attach to: the resolved report and its specimen. Reading
    # case.checklist below already resolves it (and raises the same UndesignatedReportError on an
    # ambiguous multi-report case), so resolving here changes no behaviour.
    resolved = case._resolved_report()
    report_id = resolved.report_id
    specimen_key = resolved.specimens[0].specimen_id if resolved.specimens else None

    now = _utc_now_iso()
    coverage = mapping.cap_checklist_coverage()
    _delete_checklist_observations(conn, case.case_barcode)
    for field_name, field in case.checklist.all_fields().items():
        concept_id = coverage.get(field_name)
        observation = Observation.from_checklist_field(
            field,
            case_barcode=case.case_barcode,
            report_id=report_id,
            specimen_key=specimen_key,
            field_name=field_name,
            concept_id=concept_id,
            created_at=now,
        )
        insert_observation(conn, observation)
        # The field cache stores the value as a view over the observation just written, not the field's
        # own value directly (#111): the observation is the stored atom, the cache serves it.
        derived_value = derive_field_value([_observation_as_row(observation)])
        _upsert_field(
            conn, case.case_barcode, field_name, field, concept_id=concept_id, value=derived_value
        )

    conn.commit()


def _observation_as_row(observation: Observation) -> dict:
    """An Observation as the dict shape `derive_field_value` reads, so the write path derives the cached
    value the same way the export reads it back from the table."""
    return {c: getattr(observation, c) for c in _OBSERVATION_COLUMNS}


def _upsert_report(conn: Connection, case_barcode: str, report: Report) -> None:
    conn.execute(
        text(
            """
            INSERT INTO reports (
                report_id, case_barcode, report_type, institution, report_date,
                staging_edition, supersedes_report_id, is_authoritative, report_text
            )
            VALUES (
                :report_id, :case_barcode, :report_type, :institution, :report_date,
                :staging_edition, :supersedes_report_id, :is_authoritative, :report_text
            )
            ON CONFLICT (report_id) DO UPDATE SET
                report_type = excluded.report_type,
                institution = excluded.institution,
                report_date = excluded.report_date,
                staging_edition = excluded.staging_edition,
                supersedes_report_id = excluded.supersedes_report_id,
                is_authoritative = excluded.is_authoritative,
                report_text = excluded.report_text
            """
        ),
        {
            "report_id": report.report_id,
            "case_barcode": case_barcode,
            "report_type": report.report_type.value,
            "institution": report.institution,
            "report_date": report.report_date.isoformat() if report.report_date else None,
            "staging_edition": report.staging_edition.value,
            "supersedes_report_id": report.supersedes_report_id,
            "is_authoritative": (
                None if report.is_authoritative is None else int(report.is_authoritative)
            ),
            "report_text": report.report_text,
        },
    )
    for specimen in report.specimens:
        _upsert_specimen(conn, report.report_id, specimen)


def _upsert_specimen(conn: Connection, report_id: str, specimen: Specimen) -> None:
    conn.execute(
        text(
            """
            INSERT INTO specimens (specimen_id, report_id, procedure)
            VALUES (:specimen_id, :report_id, :procedure)
            ON CONFLICT (specimen_id) DO UPDATE SET
                report_id = excluded.report_id,
                procedure = excluded.procedure
            """
        ),
        {"specimen_id": specimen.specimen_id, "report_id": report_id, "procedure": specimen.procedure},
    )


def _upsert_field(
    conn: Connection,
    case_barcode: str,
    field_name: str,
    field: ChecklistField,
    *,
    concept_id: Optional[str] = None,
    value: Any = _UNSET,
) -> None:
    # The cache value is the one the observation view derived (#111), passed in by upsert_case. It falls
    # back to the field's own value only when a caller writes a field cache without an observation.
    resolved_value = _primitive(field.value) if value is _UNSET else value
    value_json = json.dumps(resolved_value)
    conn.execute(
        text(
            """
            INSERT INTO fields (
                case_barcode, field_name, value_json, confidence, evidence_json,
                status, applicable, not_applicable_reason, search_log_json, concept_id
            )
            VALUES (
                :case_barcode, :field_name, :value_json, :confidence, :evidence_json,
                :status, :applicable, :not_applicable_reason, :search_log_json, :concept_id
            )
            ON CONFLICT (case_barcode, field_name) DO UPDATE SET
                value_json = excluded.value_json,
                confidence = excluded.confidence,
                evidence_json = excluded.evidence_json,
                status = excluded.status,
                applicable = excluded.applicable,
                not_applicable_reason = excluded.not_applicable_reason,
                search_log_json = excluded.search_log_json,
                concept_id = excluded.concept_id
            """
        ),
        {
            "case_barcode": case_barcode,
            "field_name": field_name,
            "value_json": value_json,
            "confidence": field.confidence,
            "evidence_json": field.evidence.model_dump_json() if field.evidence else None,
            "status": field.status.value,
            "applicable": int(field.applicable),
            "not_applicable_reason": field.not_applicable_reason,
            "search_log_json": json.dumps(field.search_log),
            "concept_id": concept_id,
        },
    )


def get_case(conn: Connection, case_barcode: str) -> Optional[tuple[Case, Optional[str]]]:
    case_row = (
        conn.execute(text("SELECT * FROM cases WHERE case_barcode = :cb"), {"cb": case_barcode})
        .mappings()
        .fetchone()
    )
    if case_row is None:
        return None
    field_rows = list(
        conn.execute(text("SELECT * FROM fields WHERE case_barcode = :cb"), {"cb": case_barcode})
        .mappings()
        .fetchall()
    )
    report_rows = list(
        conn.execute(
            text("SELECT * FROM reports WHERE case_barcode = :cb ORDER BY report_id"),
            {"cb": case_barcode},
        )
        .mappings()
        .fetchall()
    )
    specimens_by_report: dict[str, list[dict]] = {}
    for row in (
        conn.execute(
            text(
                "SELECT s.* FROM specimens s JOIN reports r ON r.report_id = s.report_id "
                "WHERE r.case_barcode = :cb"
            ),
            {"cb": case_barcode},
        )
        .mappings()
        .fetchall()
    ):
        specimens_by_report.setdefault(row["report_id"], []).append(dict(row))
    return _assemble_case(case_row, field_rows, report_rows, specimens_by_report)


def _build_checklist(field_rows: list) -> EndometrialChecklist:
    """Reconstruct a case's checklist from its `fields` rows."""
    checklist_kwargs = {}
    for row in field_rows:
        evidence = EvidenceSpan.model_validate_json(row["evidence_json"]) if row["evidence_json"] else None
        checklist_kwargs[row["field_name"]] = ChecklistField(
            value=json.loads(row["value_json"]) if row["value_json"] is not None else None,
            confidence=row["confidence"],
            evidence=evidence,
            status=FieldStatus(row["status"]),
            applicable=bool(row["applicable"]),
            not_applicable_reason=row["not_applicable_reason"],
            search_log=json.loads(row["search_log_json"]),
        )
    return EndometrialChecklist(**checklist_kwargs)


def _build_reports(
    report_rows: list, specimens_by_report: dict[str, list[dict]]
) -> Optional[list[Report]]:
    """Reconstruct a case's reports/specimens from its `reports` rows and specimens keyed by report_id
    (issue #110), each specimen's checklist left empty -- the assembler attaches the real one, read from
    the `fields` table, to whichever report resolves. Returns `None` when no grain has been persisted."""
    if not report_rows:
        return None
    return [
        Report(
            report_id=row["report_id"],
            report_type=ReportType(row["report_type"]),
            institution=row["institution"],
            report_date=row["report_date"],
            staging_edition=FigoStagingEdition(row["staging_edition"]),
            supersedes_report_id=row["supersedes_report_id"],
            is_authoritative=(
                None if row["is_authoritative"] is None else bool(row["is_authoritative"])
            ),
            report_text=row["report_text"],
            specimens=[
                Specimen(specimen_id=meta["specimen_id"], procedure=meta["procedure"])
                for meta in specimens_by_report.get(row["report_id"], [])
            ],
        )
        for row in report_rows
    ]


def _assemble_case(
    case_row, field_rows: list, report_rows: list, specimens_by_report: dict[str, list[dict]]
) -> tuple[Case, Optional[str]]:
    """Assemble one Case from its already-fetched rows. Shared by `get_case` (one case's rows) and
    `load_all_cases` (every case's rows in bulk), so the two reconstruct a case identically."""
    real_checklist = _build_checklist(field_rows)
    reports = _build_reports(report_rows, specimens_by_report)
    if reports is None:
        # No grain persisted yet -- a case upserted before issue #110, or by anything that predates it.
        # Behave exactly as before: a single auto-vivified report/specimen.
        case = Case(
            case_barcode=case_row["case_barcode"],
            report_date=case_row["report_date"],
            status=CaseStatus(case_row["status"]),
            checklist=real_checklist,
        )
        # staging_edition is normally re-derived from report_date by the model validator; override with the
        # stored value in case it was set without a report_date (derivation defaults to FIGO_2009 then).
        case.staging_edition = FigoStagingEdition(case_row["staging_edition"])
        return case, case_row["patient_filename"]

    case = Case(case_barcode=case_row["case_barcode"], status=CaseStatus(case_row["status"]), reports=reports)
    # The fields table stays case_barcode-scoped, not report-scoped (issue #110): it holds whichever
    # report was resolved when it was last written. Attach it the same way.
    resolved = case._resolved_report()
    if resolved.specimens:
        resolved.specimens[0].checklist = real_checklist
    else:
        resolved.specimens.append(
            Specimen(specimen_id=f"{resolved.report_id}-specimen-1", checklist=real_checklist)
        )
    return case, case_row["patient_filename"]


def load_all_cases(
    conn: Connection, project: Optional[str] = None
) -> list[tuple[Case, Optional[str]]]:
    """Every case reconstructed with a fixed number of queries, for the cohort readers (export #101,
    dashboard #100). `get_case` issues four queries per case, fine for one case but a per-case N+1 across
    the whole cohort on a remote database (issue #131): reading 548 cases one at a time takes ~100s over
    Postgres and tips `/export` past a timeout. This loads the cases, fields, reports, and specimens in
    four bulk queries and assembles each Case in memory. Ordered by case_barcode, matching the per-case
    loop it replaces.

    `project` scopes the cohort to one project's cases (issue #149), joining the fields, reports, and
    specimens reads to the same project so the whole cohort table is not read for one project. Omitted,
    every case loads, so the pre-project call sites keep working unchanged.
    """
    params: dict = {}
    if project is not None:
        params["project"] = project
        case_query = "SELECT * FROM cases WHERE project_id = :project ORDER BY case_barcode"
        fields_query = (
            "SELECT f.* FROM fields f JOIN cases c ON c.case_barcode = f.case_barcode "
            "WHERE c.project_id = :project"
        )
        reports_query = (
            "SELECT r.* FROM reports r JOIN cases c ON c.case_barcode = r.case_barcode "
            "WHERE c.project_id = :project ORDER BY r.report_id"
        )
        specimens_query = (
            "SELECT s.* FROM specimens s JOIN reports r ON r.report_id = s.report_id "
            "JOIN cases c ON c.case_barcode = r.case_barcode WHERE c.project_id = :project"
        )
    else:
        case_query = "SELECT * FROM cases ORDER BY case_barcode"
        fields_query = "SELECT * FROM fields"
        reports_query = "SELECT * FROM reports ORDER BY report_id"
        specimens_query = "SELECT * FROM specimens"

    case_rows = conn.execute(text(case_query), params).mappings().fetchall()

    fields_by_case: dict[str, list] = {}
    for row in conn.execute(text(fields_query), params).mappings().fetchall():
        fields_by_case.setdefault(row["case_barcode"], []).append(row)

    reports_by_case: dict[str, list] = {}
    for row in conn.execute(text(reports_query), params).mappings().fetchall():
        reports_by_case.setdefault(row["case_barcode"], []).append(row)

    specimens_by_report: dict[str, list[dict]] = {}
    for row in conn.execute(text(specimens_query), params).mappings().fetchall():
        specimens_by_report.setdefault(row["report_id"], []).append(dict(row))

    return [
        _assemble_case(
            case_row,
            fields_by_case.get(case_row["case_barcode"], []),
            reports_by_case.get(case_row["case_barcode"], []),
            specimens_by_report,
        )
        for case_row in case_rows
    ]


def resolve_evidence_snippet(
    conn: Connection, report_id: str, char_start: Optional[int], char_end: Optional[int]
) -> Optional[str]:
    """The verbatim snippet a stored `char_start`/`char_end` names, read from the report's own
    persisted `report_text` (issue #39) rather than re-reading the source corpus file. Returns
    `None` when the report carries no stored text, or the span is unset (a visual-pass span
    carries a page and no character offsets)."""
    if char_start is None or char_end is None:
        return None
    row = (
        conn.execute(text("SELECT report_text FROM reports WHERE report_id = :rid"), {"rid": report_id})
        .mappings()
        .fetchone()
    )
    if row is None or row["report_text"] is None:
        return None
    return row["report_text"][char_start:char_end]


def set_report_text(
    conn: Connection, case_barcode: str, report_text: str, *, commit: bool = True
) -> int:
    """Replace the persisted `report_text` for every report of a case, returning the number of report
    rows updated. This is the write side of the re-OCR path (#212): a cleaner transcription of the page
    images supersedes the damaged corpus OCR as the report's stored text, so the extraction pass reads
    it and `resolve_evidence_snippet` serves evidence from it. A case with no report row updates zero
    rows; the caller decides whether that is a miss to log."""
    result = conn.execute(
        text("UPDATE reports SET report_text = :t WHERE case_barcode = :cb"),
        {"t": report_text, "cb": case_barcode},
    )
    if commit:
        conn.commit()
    return result.rowcount


def list_cases(
    conn: Connection,
    status: Optional[str] = None,
    field_name: Optional[str] = None,
    field_status: Optional[str] = None,
    max_confidence: Optional[float] = None,
    project: Optional[str] = None,
) -> list[dict]:
    """Worklist query. field_name + max_confidence together implement the
    PRD's own named filter example: "everything with a low-confidence FIGO
    grade".

    `project` scopes the worklist to one project's cases. Omitted, every
    case is listed, so the pre-project call sites keep working unchanged.
    """
    params: dict = {}
    if field_name is not None:
        query = """
            SELECT c.case_barcode, c.patient_filename, c.status
            FROM cases c
            JOIN fields f ON f.case_barcode = c.case_barcode
            WHERE f.field_name = :field_name
        """
        params["field_name"] = field_name
        if field_status is not None:
            query += " AND f.status = :field_status"
            params["field_status"] = field_status
        if max_confidence is not None:
            query += " AND (f.confidence IS NOT NULL AND f.confidence <= :max_confidence)"
            params["max_confidence"] = max_confidence
        if status is not None:
            query += " AND c.status = :status"
            params["status"] = status
        if project is not None:
            query += " AND c.project_id = :project"
            params["project"] = project
    else:
        query = "SELECT case_barcode, patient_filename, status FROM cases WHERE 1=1"
        if status is not None:
            query += " AND status = :status"
            params["status"] = status
        if project is not None:
            query += " AND project_id = :project"
            params["project"] = project

    rows = conn.execute(text(query), params).mappings().fetchall()
    return [dict(row) for row in rows]


# Cases in these statuses count as "done" for completion-rate purposes: the reviewer has settled them.
# A flagged case carries an unresolved problem, so it is not done and is excluded here (#231): counting it
# as done inflated completion and disagreed with the Worklist, which keeps flagged in its own bucket.
_DONE_CASE_STATUSES = ("confirmed", "exported")


def dashboard_summary(conn: Connection, project: Optional[str] = None) -> dict:
    """Cohort-wide aggregate (docs/prd.md section 8.1's Cohort dashboard):
    completion rate, and which fields are most often low-confidence or
    flagged across the whole cohort -- the mechanism for catching a
    systematic per-field problem centrally instead of case by case.

    `project` scopes every count to one project's cases (issue #149): the
    total, the done count, and the per-field aggregates all join to the same
    project. Omitted, the whole cohort is summarized, so the pre-project call
    sites keep working unchanged.
    """
    params: dict = {}
    case_filter = ""
    done_filter = ""
    fields_join = ""
    if project is not None:
        params["project"] = project
        case_filter = " WHERE project_id = :project"
        done_filter = " AND project_id = :project"
        fields_join = " JOIN cases c ON c.case_barcode = f.case_barcode WHERE c.project_id = :project"

    total_cases = conn.execute(text(f"SELECT COUNT(*) FROM cases{case_filter}"), params).scalar() or 0

    done_placeholders = ",".join(f":s{i}" for i in range(len(_DONE_CASE_STATUSES)))
    done_params = {f"s{i}": value for i, value in enumerate(_DONE_CASE_STATUSES)}
    done_params.update(params)
    done_cases = (
        conn.execute(
            text(f"SELECT COUNT(*) FROM cases WHERE status IN ({done_placeholders}){done_filter}"),
            done_params,
        ).scalar()
        or 0
    )

    # Columns are qualified `f.` so the optional cases join stays unambiguous: `cases` carries its own
    # `status` column, which would otherwise collide with the fields `status` referenced here.
    field_rows = conn.execute(
        text(
            f"""
            SELECT
                f.field_name AS field_name,
                COUNT(*) AS total,
                SUM(CASE WHEN f.status = 'confirmed' THEN 1 ELSE 0 END) AS confirmed,
                SUM(CASE WHEN f.status = 'needs_review' THEN 1 ELSE 0 END) AS needs_review,
                SUM(CASE WHEN f.status = 'flagged' THEN 1 ELSE 0 END) AS flagged,
                AVG(CASE WHEN f.applicable = 1 THEN f.confidence END) AS avg_confidence
            FROM fields f{fields_join}
            GROUP BY f.field_name
            """
        ),
        params,
    ).mappings().fetchall()

    # Coerce explicitly: Postgres returns SUM as bigint and AVG as numeric
    # (Decimal), which would otherwise leak Decimal into the JSON response.
    fields = [
        {
            "field_name": row["field_name"],
            "total": int(row["total"]),
            "confirmed": int(row["confirmed"] or 0),
            "needs_review": int(row["needs_review"] or 0),
            "flagged": int(row["flagged"] or 0),
            "avg_confidence": float(row["avg_confidence"]) if row["avg_confidence"] is not None else None,
        }
        for row in field_rows
    ]
    # Worst-first: the field most often needing review or flagged is the
    # one worth investigating as a systematic extraction problem.
    fields.sort(key=lambda f: f["needs_review"] + f["flagged"], reverse=True)

    return {
        "total_cases": int(total_cases),
        "done_cases": int(done_cases),
        "completion_rate": (done_cases / total_cases) if total_cases else 0.0,
        "fields": fields,
    }


def record_mapping_confirmation(
    conn: Connection,
    *,
    mapping_id: str,
    source_std: str,
    source_field: str,
    concept_id: Optional[str],
    relation: str,
    predicate: str,
    confirmed_by: str,
    role: str,
    note: Optional[str],
    licence_required: bool,
    confirmed_at: str,
    disposition: str = "confirmed",
    decided_via: Optional[str] = None,
    question: Optional[str] = None,
    options_json: Optional[str] = None,
) -> None:
    """Record one reviewer's decision on a crosswalk edge (#95, #103). Upserts by mapping_id, so
    amending a decision replaces the prior row. Beyond the resolved concept and relation, it records the
    disposition (confirmed, flagged, or deferred), how the answer was reached (decided_via), and the
    question and options the decision weighed, so the row reads as a decision-ledger entry. See the
    mapping_confirmations DDL for the open decision on where this provenance ultimately lives (#93)."""
    conn.execute(
        text(
            """
            INSERT INTO mapping_confirmations (
                mapping_id, source_std, source_field, concept_id, relation, predicate,
                confirmed_by, role, note, licence_required, confirmed_at,
                disposition, decided_via, question, options_json
            )
            VALUES (
                :mapping_id, :source_std, :source_field, :concept_id, :relation, :predicate,
                :confirmed_by, :role, :note, :licence_required, :confirmed_at,
                :disposition, :decided_via, :question, :options_json
            )
            ON CONFLICT (mapping_id) DO UPDATE SET
                concept_id = excluded.concept_id,
                relation = excluded.relation,
                predicate = excluded.predicate,
                confirmed_by = excluded.confirmed_by,
                role = excluded.role,
                note = excluded.note,
                licence_required = excluded.licence_required,
                confirmed_at = excluded.confirmed_at,
                disposition = excluded.disposition,
                decided_via = excluded.decided_via,
                question = excluded.question,
                options_json = excluded.options_json
            """
        ),
        {
            "mapping_id": mapping_id,
            "source_std": source_std,
            "source_field": source_field,
            "concept_id": concept_id,
            "relation": relation,
            "predicate": predicate,
            "confirmed_by": confirmed_by,
            "role": role,
            "note": note,
            "licence_required": int(licence_required),
            "confirmed_at": confirmed_at,
            "disposition": disposition,
            "decided_via": decided_via,
            "question": question,
            "options_json": options_json,
        },
    )
    conn.commit()


def list_mapping_confirmations(conn: Connection, source_std: str) -> dict[str, dict]:
    """The recorded edge decisions for one source standard, keyed by mapping_id (#95, #103). Each value
    carries the resolution and its ledger fields (disposition, decided_via, question, options)."""
    rows = (
        conn.execute(
            text("SELECT * FROM mapping_confirmations WHERE source_std = :s"),
            {"s": source_std},
        )
        .mappings()
        .fetchall()
    )
    return {
        row["mapping_id"]: {
            "mapping_id": row["mapping_id"],
            "source_field": row["source_field"],
            "concept_id": row["concept_id"],
            "relation": row["relation"],
            "predicate": row["predicate"],
            "confirmed_by": row["confirmed_by"],
            "role": row["role"],
            "note": row["note"],
            "licence_required": bool(row["licence_required"]),
            "confirmed_at": row["confirmed_at"],
            "disposition": row["disposition"] or "confirmed",
            "decided_via": row["decided_via"],
            "question": row["question"],
            "options_json": row["options_json"],
        }
        for row in rows
    }


def record_mapping_components(
    conn: Connection, *, mapping_id: str, source_std: str, components: list[dict]
) -> None:
    """Replace the component edges recorded for one source field's split (#103 follow-on). Deletes every
    component row for the mapping_id, then inserts the given set, so applying a split is idempotent and a
    later plain confirmation (empty `components`) clears the split. Each component is a dict of
    concept_id, relation, predicate, and an optional note. The parent headline lives in
    mapping_confirmations and is written separately."""
    conn.execute(
        text("DELETE FROM mapping_confirmation_components WHERE mapping_id = :m"),
        {"m": mapping_id},
    )
    for component in components:
        conn.execute(
            text(
                """
                INSERT INTO mapping_confirmation_components (
                    mapping_id, source_std, concept_id, relation, predicate, note
                )
                VALUES (:mapping_id, :source_std, :concept_id, :relation, :predicate, :note)
                """
            ),
            {
                "mapping_id": mapping_id,
                "source_std": source_std,
                "concept_id": component["concept_id"],
                "relation": component["relation"],
                "predicate": component["predicate"],
                "note": component.get("note"),
            },
        )
    conn.commit()


def list_mapping_components(conn: Connection, source_std: str) -> dict[str, list[dict]]:
    """The component edges of every split recorded for one source standard, keyed by parent mapping_id
    (#103 follow-on). A source field with no split has no entry."""
    rows = (
        conn.execute(
            text("SELECT * FROM mapping_confirmation_components WHERE source_std = :s"),
            {"s": source_std},
        )
        .mappings()
        .fetchall()
    )
    by_mapping: dict[str, list[dict]] = {}
    for row in rows:
        by_mapping.setdefault(row["mapping_id"], []).append(
            {
                "concept_id": row["concept_id"],
                "relation": row["relation"],
                "predicate": row["predicate"],
                "note": row["note"],
            }
        )
    return by_mapping


def record_field_confirmation(
    conn: Connection,
    *,
    case_barcode: str,
    field_name: str,
    value_json: Optional[str],
    status: str,
    confirmed_by: str,
    role: str,
    note: Optional[str] = None,
    disposition: str = "confirmed",
    decided_via: Optional[str] = None,
    question: Optional[str] = None,
    options_json: Optional[str] = None,
    confirmed_at: str,
) -> None:
    """Record one reviewer's decision on one checklist field, the case-review decision ledger (#98).
    Upserts by (case_barcode, field_name), so amending a decision replaces the prior row. Beyond the
    confirmed value and status, it records the disposition, how the answer was reached (decided_via), the
    confirmer and their role from the session, and the question and options the decision weighed, so the
    row reads as a source-to-decision ledger entry (#103's pattern)."""
    conn.execute(
        text(
            """
            INSERT INTO field_confirmations (
                case_barcode, field_name, value_json, status, disposition, decided_via,
                confirmed_by, role, note, question, options_json, confirmed_at
            )
            VALUES (
                :case_barcode, :field_name, :value_json, :status, :disposition, :decided_via,
                :confirmed_by, :role, :note, :question, :options_json, :confirmed_at
            )
            ON CONFLICT (case_barcode, field_name) DO UPDATE SET
                value_json = excluded.value_json,
                status = excluded.status,
                disposition = excluded.disposition,
                decided_via = excluded.decided_via,
                confirmed_by = excluded.confirmed_by,
                role = excluded.role,
                note = excluded.note,
                question = excluded.question,
                options_json = excluded.options_json,
                confirmed_at = excluded.confirmed_at
            """
        ),
        {
            "case_barcode": case_barcode,
            "field_name": field_name,
            "value_json": value_json,
            "status": status,
            "disposition": disposition,
            "decided_via": decided_via,
            "confirmed_by": confirmed_by,
            "role": role,
            "note": note,
            "question": question,
            "options_json": options_json,
            "confirmed_at": confirmed_at,
        },
    )
    conn.commit()


def list_field_confirmations(conn: Connection, case_barcode: str) -> dict[str, dict]:
    """The recorded field decisions for one case, keyed by field_name (#98). Each value carries the
    confirmed value and status plus its ledger fields (disposition, decided_via, confirmer, role, note,
    question, options, time)."""
    rows = (
        conn.execute(
            text("SELECT * FROM field_confirmations WHERE case_barcode = :cb"),
            {"cb": case_barcode},
        )
        .mappings()
        .fetchall()
    )
    return {
        row["field_name"]: {
            "case_barcode": row["case_barcode"],
            "field_name": row["field_name"],
            "value_json": row["value_json"],
            "status": row["status"],
            "disposition": row["disposition"] or "confirmed",
            "decided_via": row["decided_via"],
            "confirmed_by": row["confirmed_by"],
            "role": row["role"],
            "note": row["note"],
            "question": row["question"],
            "options_json": row["options_json"],
            "confirmed_at": row["confirmed_at"],
        }
        for row in rows
    }


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


# --- reviewer accounts (#128) --------------------------------------------------


def upsert_user(
    conn: Connection,
    *,
    username: str,
    password_hash: str,
    role: str,
    holds_licence: bool,
    display_name: Optional[str] = None,
    created_at: Optional[str] = None,
    project_id: str = DEFAULT_PROJECT_ID,
) -> None:
    """Create or update one reviewer account, keyed by username. Upsert so re-running the seed rotates a
    password or a role in place rather than failing on a duplicate. `password_hash` is an
    endopath.auth scrypt string; the plaintext is never passed here. `project_id` scopes the account to a
    project; the seeded users belong to the default project."""
    conn.execute(
        text(
            """
            INSERT INTO users (
                username, password_hash, role, holds_licence, display_name, created_at, project_id
            )
            VALUES (
                :username, :password_hash, :role, :holds_licence, :display_name, :created_at, :project_id
            )
            ON CONFLICT (username) DO UPDATE SET
                password_hash = excluded.password_hash,
                role = excluded.role,
                holds_licence = excluded.holds_licence,
                display_name = excluded.display_name,
                project_id = excluded.project_id
            """
        ),
        {
            "username": username,
            "password_hash": password_hash,
            "role": role,
            "holds_licence": int(holds_licence),
            "display_name": display_name,
            "created_at": created_at or _utc_now_iso(),
            "project_id": project_id,
        },
    )
    conn.commit()


def get_user(conn: Connection, username: str) -> Optional[dict]:
    """One account by username, or None. Carries the stored password hash, so callers verify against it
    with endopath.auth.verify_password rather than this module knowing about passwords."""
    row = (
        conn.execute(text("SELECT * FROM users WHERE username = :u"), {"u": username})
        .mappings()
        .fetchone()
    )
    if row is None:
        return None
    return {
        "username": row["username"],
        "password_hash": row["password_hash"],
        "role": row["role"],
        "holds_licence": bool(row["holds_licence"]),
        "display_name": row["display_name"] or "",
        "created_at": row["created_at"],
    }


def count_users(conn: Connection) -> int:
    """How many reviewer accounts exist. Startup seeding checks this so a deployment that already has
    accounts is left alone, and it distinguishes an auth-enabled Space with no way to log in."""
    return int(conn.execute(text("SELECT COUNT(*) FROM users")).scalar_one())


# --- projects ("repos"): the dataset a team labels --------------------------------


def list_projects(conn: Connection) -> list[dict]:
    """Every project with its case count, the default project first. `case_count` is a left join on
    `cases.project_id`, so a project with no cases yet reports zero rather than dropping out."""
    rows = (
        conn.execute(
            text(
                """
                SELECT p.project_id, p.name, COUNT(c.case_barcode) AS case_count
                FROM projects p
                LEFT JOIN cases c ON c.project_id = p.project_id
                GROUP BY p.project_id, p.name
                """
            )
        )
        .mappings()
        .fetchall()
    )
    projects = [
        {
            "project_id": row["project_id"],
            "name": row["name"],
            "case_count": int(row["case_count"]),
        }
        for row in rows
    ]
    # The default project first, then the rest by name, so the seeded project heads a stable list.
    projects.sort(key=lambda p: (p["project_id"] != DEFAULT_PROJECT_ID, p["name"]))
    return projects


def create_project(conn: Connection, *, project_id: str, name: str, created_by: Optional[str]) -> dict:
    """Insert one project and return it (`project_id`, `name`, `case_count` of zero). Raises ValueError
    when the project_id already exists, so the caller renders a clear conflict rather than an opaque
    integrity error."""
    existing = conn.execute(
        text("SELECT project_id FROM projects WHERE project_id = :project_id"),
        {"project_id": project_id},
    ).fetchone()
    if existing is not None:
        raise ValueError(f"project_id {project_id!r} already exists")
    conn.execute(
        text(
            "INSERT INTO projects (project_id, name, created_by, created_at) "
            "VALUES (:project_id, :name, :created_by, :created_at)"
        ),
        {
            "project_id": project_id,
            "name": name,
            "created_by": created_by,
            "created_at": _utc_now_iso(),
        },
    )
    conn.commit()
    return {"project_id": project_id, "name": name, "case_count": 0}


# --- Intake follow-up state: submitted dictionaries and the job queue ----------
# The durable backing for dictionary.py and jobs.py. A process-level registry would not survive the
# request that submitted the dictionary, let alone a restart, so a submission's dictionary and its
# follow-up jobs live here in the same store as cases and confirmations.


def upsert_dictionary(conn: Connection, *, dictionary_id: str, title: str, variables_json: str) -> None:
    """Store a submitted target dictionary so the compilation worker reads it back by id (#91). Upserts
    by id, so re-submitting the same dictionary replaces its row."""
    conn.execute(
        text(
            """
            INSERT INTO dictionaries (id, title, variables_json, created_at)
            VALUES (:id, :title, :variables_json, :created_at)
            ON CONFLICT (id) DO UPDATE SET
                title = excluded.title,
                variables_json = excluded.variables_json,
                created_at = excluded.created_at
            """
        ),
        {
            "id": dictionary_id,
            "title": title,
            "variables_json": variables_json,
            "created_at": _utc_now_iso(),
        },
    )
    conn.commit()


def get_dictionary(conn: Connection, dictionary_id: str) -> Optional[dict]:
    """The stored dictionary row (id, title, variables_json), or None if it was never submitted."""
    row = (
        conn.execute(
            text("SELECT id, title, variables_json FROM dictionaries WHERE id = :id"),
            {"id": dictionary_id},
        )
        .mappings()
        .fetchone()
    )
    return dict(row) if row is not None else None


def list_dictionaries(conn: Connection) -> list[dict]:
    """Every submitted dictionary (id, title), newest submission first."""
    rows = (
        conn.execute(text("SELECT id, title FROM dictionaries ORDER BY created_at DESC, id"))
        .mappings()
        .fetchall()
    )
    return [dict(row) for row in rows]


# --- Field coverage: a project's target dictionary and the record types that fulfil it (#237) ------
# The upserts do not commit, so a seed batches many rows and commits once.


def list_record_types(conn: Connection, project: str) -> list[dict]:
    """The file record types a project draws its values from, in display order."""
    rows = (
        conn.execute(
            text(
                "SELECT record_type_id, name, source_class, blurb FROM record_types "
                "WHERE project_id = :p ORDER BY sort_order, record_type_id"
            ),
            {"p": project},
        )
        .mappings()
        .fetchall()
    )
    return [dict(row) for row in rows]


def list_coverage_fields(
    conn: Connection, project: str, dictionary_id: Optional[str] = None
) -> list[dict]:
    """A project's target-dictionary fields with the record type that fulfils each and its fill config, in
    dictionary order. Omit dictionary_id for every coverage field the project carries."""
    params: dict = {"p": project}
    where = "project_id = :p"
    if dictionary_id is not None:
        where += " AND dictionary_id = :d"
        params["d"] = dictionary_id
    rows = (
        conn.execute(
            text(
                "SELECT dictionary_id, name, label, section, record_type_id, fill_via, fill_key "
                f"FROM coverage_fields WHERE {where} ORDER BY sort_order, name"
            ),
            params,
        )
        .mappings()
        .fetchall()
    )
    return [dict(row) for row in rows]


def upsert_record_type(
    conn: Connection,
    *,
    project: str,
    record_type_id: str,
    name: str,
    source_class: str,
    blurb: Optional[str],
    sort_order: int,
) -> None:
    conn.execute(
        text(
            """
            INSERT INTO record_types (project_id, record_type_id, name, source_class, blurb, sort_order)
            VALUES (:p, :rt, :name, :sc, :blurb, :so)
            ON CONFLICT (project_id, record_type_id) DO UPDATE SET
                name = excluded.name, source_class = excluded.source_class,
                blurb = excluded.blurb, sort_order = excluded.sort_order
            """
        ),
        {"p": project, "rt": record_type_id, "name": name, "sc": source_class, "blurb": blurb, "so": sort_order},
    )


def upsert_coverage_field(
    conn: Connection,
    *,
    project: str,
    dictionary_id: str,
    name: str,
    label: str,
    section: Optional[str],
    record_type_id: Optional[str],
    fill_via: Optional[str],
    fill_key: Optional[str],
    sort_order: int,
) -> None:
    conn.execute(
        text(
            """
            INSERT INTO coverage_fields
                (project_id, dictionary_id, name, label, section, record_type_id, fill_via, fill_key, sort_order)
            VALUES (:p, :d, :name, :label, :section, :rt, :via, :key, :so)
            ON CONFLICT (project_id, dictionary_id, name) DO UPDATE SET
                label = excluded.label, section = excluded.section, record_type_id = excluded.record_type_id,
                fill_via = excluded.fill_via, fill_key = excluded.fill_key, sort_order = excluded.sort_order
            """
        ),
        {
            "p": project,
            "d": dictionary_id,
            "name": name,
            "label": label,
            "section": section,
            "rt": record_type_id,
            "via": fill_via,
            "key": fill_key,
            "so": sort_order,
        },
    )


def has_coverage(conn: Connection, project: str) -> bool:
    """Whether the project's field coverage has been seeded."""
    return bool(
        conn.execute(
            text("SELECT 1 FROM coverage_fields WHERE project_id = :p LIMIT 1"), {"p": project}
        ).scalar()
    )


def upsert_record_type_extraction(
    conn: Connection,
    *,
    project: str,
    record_type_id: str,
    case_barcode: str,
    field: str,
    value: str,
    evidence: Optional[str],
    sort_order: int,
) -> None:
    conn.execute(
        text(
            """
            INSERT INTO record_type_extractions
                (project_id, record_type_id, case_barcode, field, value, evidence, sort_order)
            VALUES (:p, :rt, :bc, :field, :value, :ev, :so)
            ON CONFLICT (project_id, record_type_id, case_barcode, field) DO UPDATE SET
                value = excluded.value, evidence = excluded.evidence, sort_order = excluded.sort_order
            """
        ),
        {
            "p": project,
            "rt": record_type_id,
            "bc": case_barcode,
            "field": field,
            "value": value,
            "ev": evidence,
            "so": sort_order,
        },
    )


def list_record_type_extractions(conn: Connection, project: str, record_type_id: str) -> list[dict]:
    """The populated extractions for a record type's extracted dictionary: one row per (case, field) with
    the value and its verbatim evidence, in case then field order."""
    rows = (
        conn.execute(
            text(
                "SELECT case_barcode, field, value, evidence FROM record_type_extractions "
                "WHERE project_id = :p AND record_type_id = :rt ORDER BY case_barcode, sort_order, field"
            ),
            {"p": project, "rt": record_type_id},
        )
        .mappings()
        .fetchall()
    )
    return [dict(row) for row in rows]


def has_record_type_extractions(conn: Connection, project: str, record_type_id: str) -> bool:
    """Whether the record type's extracted dictionary has been populated."""
    return bool(
        conn.execute(
            text(
                "SELECT 1 FROM record_type_extractions WHERE project_id = :p AND record_type_id = :rt LIMIT 1"
            ),
            {"p": project, "rt": record_type_id},
        ).scalar()
    )


def insert_jobs(conn: Connection, rows: list[dict]) -> None:
    """Enqueue follow-up jobs (#92, #91). Each row carries id, kind, ticket, dictionary_id, source_kind,
    case_count, variable_count, and created_at; status starts at 'enqueued'."""
    for row in rows:
        conn.execute(
            text(
                """
                INSERT INTO jobs (
                    id, kind, ticket, status, dictionary_id, source_kind,
                    case_count, variable_count, created_at
                )
                VALUES (
                    :id, :kind, :ticket, 'enqueued', :dictionary_id, :source_kind,
                    :case_count, :variable_count, :created_at
                )
                """
            ),
            row,
        )
    conn.commit()


def list_jobs(conn: Connection) -> list[dict]:
    """Every enqueued or drained follow-up job, oldest submission first."""
    rows = conn.execute(text("SELECT * FROM jobs ORDER BY created_at, id")).mappings().fetchall()
    return [dict(row) for row in rows]


def claim_next_job(conn: Connection) -> Optional[dict]:
    """Claim the oldest enqueued job for the worker, or return None when the queue is drained. The claim
    is a conditional update guarded by rowcount, so a second worker that read the same row before the
    update loses the race and sees no claim rather than running the job twice."""
    row = (
        conn.execute(
            text("SELECT * FROM jobs WHERE status = 'enqueued' ORDER BY created_at, id LIMIT 1")
        )
        .mappings()
        .fetchone()
    )
    if row is None:
        return None
    started_at = _utc_now_iso()
    result = conn.execute(
        text(
            "UPDATE jobs SET status = 'running', started_at = :t "
            "WHERE id = :id AND status = 'enqueued'"
        ),
        {"t": started_at, "id": row["id"]},
    )
    conn.commit()
    if result.rowcount != 1:
        return None
    claimed = dict(row)
    claimed["status"] = "running"
    claimed["started_at"] = started_at
    return claimed


def finish_job(conn: Connection, *, job_id: str, status: str, error: Optional[str] = None) -> None:
    """Mark a claimed job done or failed, stamping finished_at and any failure text."""
    conn.execute(
        text("UPDATE jobs SET status = :status, error = :error, finished_at = :t WHERE id = :id"),
        {"status": status, "error": error, "t": _utc_now_iso(), "id": job_id},
    )
    conn.commit()


def clear_intake_state(conn: Connection) -> None:
    """Drop every submitted dictionary, enqueued job, and job artifact. For tests and a fresh process."""
    conn.execute(text("DELETE FROM jobs"))
    conn.execute(text("DELETE FROM dictionaries"))
    conn.execute(text("DELETE FROM submission_artifacts"))
    conn.commit()


def upsert_submission_artifact(
    conn: Connection, *, dictionary_id: str, kind: str, payload_json: str
) -> None:
    """Store what a follow-up job produced for one dictionary submission (#107 induction, #108 crosswalk
    draft). Upserts by (dictionary_id, kind), so re-running a submission replaces the artifact."""
    conn.execute(
        text(
            """
            INSERT INTO submission_artifacts (dictionary_id, kind, payload_json, created_at)
            VALUES (:dictionary_id, :kind, :payload_json, :created_at)
            ON CONFLICT (dictionary_id, kind) DO UPDATE SET
                payload_json = excluded.payload_json,
                created_at = excluded.created_at
            """
        ),
        {
            "dictionary_id": dictionary_id,
            "kind": kind,
            "payload_json": payload_json,
            "created_at": _utc_now_iso(),
        },
    )
    conn.commit()


def get_submission_artifact(conn: Connection, dictionary_id: str, kind: str) -> Optional[dict]:
    """The stored artifact row for one dictionary and kind, or None if the job has not produced it yet."""
    row = (
        conn.execute(
            text(
                "SELECT dictionary_id, kind, payload_json, created_at FROM submission_artifacts "
                "WHERE dictionary_id = :d AND kind = :k"
            ),
            {"d": dictionary_id, "k": kind},
        )
        .mappings()
        .fetchone()
    )
    return dict(row) if row is not None else None


def export_data(conn: Connection) -> dict:
    """Dataset + provenance table together, one export action (section
    7.1: "not as separate afterthoughts").

    Both are generated from the observation atom (#111): each field's value is the view its
    observation derives (`derive_field_value`), and the provenance carries that observation's evidence
    and coded state. The `fields` cache is not read here, so the export speaks the same atoms the
    invariants are stated over.
    """
    case_rows = conn.execute(
        text("SELECT case_barcode, report_date, staging_edition, status FROM cases ORDER BY case_barcode")
    ).mappings().fetchall()
    observation_rows = conn.execute(
        text(
            "SELECT * FROM observations WHERE extraction_method = :method "
            "ORDER BY case_barcode, field_name"
        ),
        {"method": EXTRACTION_METHOD_CHECKLIST},
    ).mappings().fetchall()

    # Group each field's observations, so a field the extraction produced more than one row for (a direct
    # and a normalized reading) resolves through one derivation rather than the last row written.
    observations_by_field: dict[tuple[str, str], list[dict]] = {}
    order: list[tuple[str, str]] = []
    for row in observation_rows:
        key = (row["case_barcode"], row["field_name"])
        if key not in observations_by_field:
            observations_by_field[key] = []
            order.append(key)
        observations_by_field[key].append(dict(row))

    fields_by_case: dict[str, dict] = {}
    provenance: list[dict] = []
    for case_barcode, field_name in order:
        rows = observations_by_field[(case_barcode, field_name)]
        value = derive_field_value(rows)
        fields_by_case.setdefault(case_barcode, {})[field_name] = value

        primary = rows[0]
        # A gate (`not_applicable`) is the only absence code that removes the field's meaning; every other
        # state leaves it applicable. The gate's human-readable condition rides `absence_detail`.
        applicable = primary["absence_reason"] != _ABSENCE_NOT_APPLICABLE
        provenance.append(
            {
                "case_barcode": case_barcode,
                "field_name": field_name,
                "value": value,
                "confidence": primary["confidence"],
                "status": primary["review_status"],
                "applicable": applicable,
                "not_applicable_reason": primary["absence_detail"],
                "evidence_quote": primary["quote"],
                "evidence_source": primary["source_adapter"],
            }
        )

    dataset = [
        {
            "case_barcode": row["case_barcode"],
            "report_date": row["report_date"],
            "staging_edition": row["staging_edition"],
            "status": row["status"],
            **fields_by_case.get(row["case_barcode"], {}),
        }
        for row in case_rows
    ]

    return {"dataset": dataset, "provenance": provenance}