# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0

"""Ontology proposals API.

Proposals are generated by ``/induce`` and stored in DynamoDB. Users can
list, view, accept, reject, cancel, validate, or update them.

Accept is **asynchronous**: ``POST /accept`` flips the proposal's status
to ``accepting``, kicks off the shared ingest pipeline in
:mod:`coa_ontology.catalog.ingest` on a background thread,
and returns 202 immediately. Clients poll the existing
``GET /proposals/{id}`` endpoint and watch the ``status`` field flip to
``accepted`` (success) or ``accept_failed`` with ``accept_error`` naming
the pipeline step that failed. An ``accept_failed`` proposal is
re-acceptable: re-running converges on a correct graph (triples dedup,
counts are reconciled from Neptune), with the known non-idempotent
residual documented at the ``ingest`` step. It converges but does NOT
resume — every re-accept restarts at ``ingest`` (#851). The work itself
can take 30–60 seconds (catalog projection,
Turtle bulk load, embedding accumulation, S3 persist), which exceeds
API Gateway's 29-second integration timeout, so we must run it
asynchronously.
"""

import json
import logging
import os
import random
import time as _t
import uuid
from collections.abc import Callable
from datetime import UTC, datetime
from threading import Thread
from typing import Any

from botocore.exceptions import ClientError
from coa_common import sql_ident as _sql_ident
from coa_common.constants import EVENT_SOURCE_PREFIX, VOCAB_URI
from coa_control_plane_server.models.proposal_status import ProposalStatus
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from rdflib import Graph

from coa_ontology import dynamo_store
from coa_ontology.catalog.ingest import (
    IngestMetadata,
    IngestParseError,
    IngestValidationError,
    ingest_ontology,
    wait_for_embeddings_searchable,
)
from coa_ontology.stores import build_stores as _build_stores  # noqa: F401 (re-exported for tests)

_log = logging.getLogger("proposals.accept")

router = APIRouter()

# Proposal statuses come from the Smithy-generated ProposalStatus enum — the API
# contract in models/ontology-induction.smithy is the single source of truth, so
# the wire value and the stored value can never drift. Aliased here because the
# accept pipeline references a few of them constantly.
#
# ``embeddings_sync``: set after ingest while the worker waits for the embeddings
# it wrote to become searchable in the eventually-consistent vector index. Sits
# between ``accepting`` and ``accepted`` so the UI shows a distinct
# "Syncing embeddings…" phase rather than a stalled accept.
PROPOSAL_STATUS_EMBEDDINGS_SYNC = ProposalStatus.EMBEDDINGS_SYNC.value

# Terminal state for an accept that ran, hit an error on some step, and
# exhausted its per-step retries (#456/#466/#467). Distinct from ``pending``
# (never attempted) so the UI can say "accept failed, fix and retry" and the
# stale-sweep / accept guard can tell a broken attempt from a fresh proposal.
# Re-accepting from this state re-runs the pipeline from the START — it converges
# but does not resume (#851); see the ingest step for the non-idempotent residual.
PROPOSAL_STATUS_ACCEPT_FAILED = ProposalStatus.ACCEPT_FAILED.value

# In-progress accept states — a worker is (or was) running, so Accept is
# disabled and a new induction is blocked. Used by the accept guard, the
# stale-sweep, and the induction in-flight lock.
_ACCEPT_IN_PROGRESS_STATUSES = (
    ProposalStatus.ACCEPTING.value,
    PROPOSAL_STATUS_EMBEDDINGS_SYNC,
)

# Statuses that block starting a NEW induction in the namespace (see the
# in-flight guard in induce_catalog). Two kinds of "not finished with this one":
#
#   - unreviewed work: ``pending`` / ``updated`` / ``accept_failed``. The first
#     two preserve the original behaviour; ``accept_failed`` is included because
#     before that state existed a failed accept rolled back to ``pending``, which
#     blocked. Dropping it would remove the guard for exactly the case a user is
#     most likely to hit — accept broke, so they trigger another induction.
#   - an accept actively merging: ``accepting`` / ``embeddings_sync``. The
#     induction lock (acquire_induction_lock) is only held for the duration of an
#     INDUCTION, so without these a new induction can start while a worker is
#     mid-merge, leaving two competing structured proposals in the namespace.
PROPOSAL_STATUSES_BLOCKING_NEW_INDUCTION = (
    ProposalStatus.PENDING.value,
    ProposalStatus.UPDATED.value,
    PROPOSAL_STATUS_ACCEPT_FAILED,
    *_ACCEPT_IN_PROGRESS_STATUSES,
)

# Per-step auto-retry policy for the accept pipeline. A transient step error
# (AOSS 5xx — see #582, Neptune/Bedrock blip, timeout) is retried up to
# _ACCEPT_STEP_MAX_ATTEMPTS with exponential backoff before the accept is
# marked ``accept_failed``. Structural errors (bad Turtle, validation) are
# never retried — they need the user to fix the proposal first.
_ACCEPT_STEP_MAX_ATTEMPTS = int(os.getenv("ACCEPT_STEP_MAX_ATTEMPTS", "3"))
_ACCEPT_STEP_BACKOFF_SECONDS = float(os.getenv("ACCEPT_STEP_BACKOFF_SECONDS", "2"))


# ── Schemas ─────────────────────────────────────────────────────────────


class RejectRequest(BaseModel):
    """Request body for rejecting one or more proposals with a reason."""

    proposal_ids: list[str]
    reason: str


class UpdateProposalRequest(BaseModel):
    """Request body for editing a proposal's ontology and/or R2RML Turtle."""

    proposal_id: str
    ontology_turtle: str | None = None
    r2rml_turtle: str | None = None
    # When True, the corresponding artifact was uploaded straight to its S3
    # key via a presigned PUT (see ``/{proposal_id}/upload-url``) instead of
    # being sent inline — used for ontologies too large for the 6 MB request
    # cap. The server reads the uploaded artifact back from S3 to validate it.
    ontology_uploaded: bool = False
    r2rml_uploaded: bool = False
    grounding_overrides: dict[str, str | None] | None = None


class ProposalUploadUrlResponse(BaseModel):
    """Presigned S3 PUT URLs for uploading edited proposal Turtle directly."""

    ontology_put_url: str
    r2rml_put_url: str


class AcceptProposalRequest(BaseModel):
    """Optional body for ``POST /ontology/proposals/{id}/accept``.

    ``ontology_id`` — the target ontology URI under which the proposal's
    Turtle should be merged. If omitted, the proposal's own embedded
    ``ontology_id`` is used (each accept produces its own ontology).
    Supplying a shared target across multiple proposal-accept calls
    merges them under one per-ontology named graph and one registry row.
    """

    ontology_id: str | None = None


class AcceptProposalResponse(BaseModel):
    """202 payload returned by ``POST /proposals/{id}/accept``.

    The only user-visible signal that the request was accepted is that
    the proposal's status moved to ``accepting``. Clients poll
    ``GET /proposals/{id}`` from there.
    """

    proposal_id: str
    status: str


# ── S3 persistence for induced ontologies ───────────────────────────────

_S3_BUCKET = dynamo_store.resolve_ontology_bucket()
_S3_PREFIX = "ontologies"


class SchemaSqlGenerationError(Exception):
    """schema.sql generation failed (catalog fetch or DDL emission threw).

    Distinct from the "nothing to describe" empty states (no datasources / no
    tables), which return ``None``.
    """


def _ontology_short_id(ontology_id: str) -> str:
    """Return the last meaningful segment of an ontology IRI.

    Used as the S3 filename stem and as the ``version`` in the
    ontology.published event. Shared by ``_persist_induced_to_s3`` and the
    ``notify_vkg`` accept step so the event's version always matches the object
    actually written.

    Args:
        ontology_id: The ontology IRI to shorten.

    Returns:
        The trailing segment, with IRI punctuation replaced by underscores.
    """
    safe_id = ontology_id.replace("/", "_").replace("#", "").replace(":", "_").rstrip("_")
    return safe_id.split("_")[-1] if "_" in safe_id else safe_id


def _generate_schema_sql(namespace: str, ontology_id: str) -> str | None:
    """Generate H2 DDL from datasource catalog metadata.

    Returns the schema.sql text, or ``None`` when there is nothing to describe
    (no accepted datasources, or datasources that yield no catalog tables — e.g.
    an unstructured ontology). Identifiers are emitted verbatim (SQL-delimited)
    so they match the source datasource; no canonical→original column map is
    needed.

    Raises :class:`SchemaSqlGenerationError` on a genuine failure so the caller
    can tell it apart from the ``None`` empty state.
    """
    import os

    try:
        from coa_ontology.induce_catalog import _catalog_to_tables, _fetch_catalog_from_smus
        from coa_ontology.inducer.services.data_catalog import CatalogTable

        # Collect datasource IDs from accepted proposals for this ontology
        accepted = dynamo_store.list_proposals(
            namespace=namespace, ontology_id=ontology_id, status=ProposalStatus.ACCEPTED.value
        )
        ds_ids: set[str] = set()
        for p in accepted:
            meta = p.get("metadata") or {}
            for ds_id in meta.get("datasource_ids") or []:
                ds_ids.add(ds_id)

        # Legitimate empty state: an ontology with no structured datasources
        # (e.g. unstructured) has no tables to describe. Not an error.
        if not ds_ids:
            return None

        # Fetch catalog tables for each datasource
        catalog_source = os.environ.get("CATALOG_SOURCE", "http")
        config = {
            "catalog_source": catalog_source,
            "data_catalog_url": os.environ.get("DATA_CATALOG_URL", ""),
            "smus_domain_id": os.environ.get("SMUS_DOMAIN_ID", ""),
            "namespaces_table": os.environ.get("NAMESPACES_TABLE", ""),
            "smus_datasources_table": os.environ.get("DATASOURCES_TABLE", ""),
            "sources_table": os.environ.get("SOURCES_TABLE", ""),
            "smus_region": os.environ.get("AWS_REGION", "us-east-1"),
        }

        all_tables: list[CatalogTable] = []
        for ds_id in ds_ids:
            if catalog_source == "smus":
                catalog = _fetch_catalog_from_smus(config, ds_id, namespace=namespace)
            else:
                from coa_ontology.induce_catalog import _fetch_catalog

                catalog = _fetch_catalog(config["data_catalog_url"], ds_id)
            for tbl_dict in _catalog_to_tables(catalog):
                all_tables.append(CatalogTable(**tbl_dict))

        # Legitimate empty state: datasources resolved but exposed no tables.
        if not all_tables:
            return None

        return _tables_to_h2_ddl(all_tables)

    except Exception as e:
        # Genuine failure — do NOT mask as the ``None`` empty state (#457).
        raise SchemaSqlGenerationError(f"{type(e).__name__}: {e}") from e


# H2 type mapping from catalog types
_CATALOG_TO_H2 = {
    "INT": "INTEGER",
    "INTEGER": "INTEGER",
    "BIGINT": "INTEGER",
    "SMALLINT": "SMALLINT",
    "TINYINT": "TINYINT",
    "FLOAT": "FLOAT",
    "DOUBLE": "DOUBLE",
    "DECIMAL": "DECIMAL",
    "NUMERIC": "DECIMAL",
    "VARCHAR": "VARCHAR(255)",
    "TEXT": "VARCHAR(65535)",
    "CHAR": "CHAR(255)",
    "STRING": "VARCHAR(255)",
    "BOOLEAN": "BOOLEAN",
    # Temporal types map faithfully. These were previously VARCHAR(255), which
    # forced the R2RML rr:datatype and the ontology rdfs:range to xsd:string to
    # stay consistent with this schema. The consequence was that Ontop's type
    # reasoner saw every temporal column as a string, found comparisons against
    # xsd:date/xsd:dateTime literals disjoint, proved the query unsatisfiable,
    # and emitted its no-mapping placeholder ("SELECT 1 AS uselessVariable") —
    # so EVERY date-filtered SPARQL query returned nothing. Verified against
    # Ontop 5.5.0: faithful temporal types load cleanly (no
    # MappingOntologyMismatchException) and produce real SQL.
    "DATE": "DATE",
    "TIMESTAMP": "TIMESTAMP",
    "DATETIME": "TIMESTAMP",
    "TIME": "TIME",
    # UUID is textual; declared explicitly so it does not rely on the
    # VARCHAR fallback (which would hide a genuinely unmapped type).
    "UUID": "VARCHAR(36)",
    "BINARY": "VARBINARY",
    "BLOB": "BLOB",
}


def _tables_to_h2_ddl(tables: list) -> str:
    """Convert CatalogTable list to H2-compatible CREATE TABLE statements.

    Table and column names are emitted verbatim as SQL-delimited (double-quoted)
    identifiers via :func:`sql_ident`. Double-quoting lets H2 accept any source
    name — spaces, parens, percent signs, hyphens, mixed case, reserved words —
    so the identifiers exactly match the source datasource (and therefore the
    R2RML ``rr:tableName`` / ``rr:column`` values that Ontop validates against
    this schema). No canonicalization or uppercasing is applied.
    """
    lines = ["-- Auto-generated schema for Ontop H2 validation (from datasource catalog)"]
    for table in sorted(tables, key=lambda t: t.name):
        col_defs = []
        pk_cols = []
        if table.tableConstraints:
            for tc in table.tableConstraints:
                if tc.constraintType == "PRIMARY_KEY":
                    pk_cols = tc.columns
        for col in table.columns:
            h2_type = _CATALOG_TO_H2.get(col.dataType.split("(")[0].upper(), "VARCHAR(255)")
            col_def = f"{_sql_ident(col.name)} {h2_type}"
            if col.name in pk_cols:
                col_def += " NOT NULL"
            col_defs.append(col_def)
        if pk_cols:
            pk_quoted = ", ".join(_sql_ident(c) for c in pk_cols)
            col_defs.append(f"PRIMARY KEY ({pk_quoted})")
        lines.append(f"CREATE TABLE IF NOT EXISTS {_sql_ident(table.name)} ({', '.join(col_defs)});")
    return "\n".join(lines)


def _clean_ontology_for_vkg(turtle: str, r2rml_turtle: str | None = None) -> str:
    """Strip OWL constructs incompatible with Ontop's OWL 2 QL profile.

    Removes:
    - rdf:Property type assertions (redundant; Ontop needs only owl:ObjectProperty/DatatypeProperty)
    - owl:hasKey axioms (not in OWL 2 QL, and blank-node lists confuse Ontop)
    - Internal workbench metadata triples (ontology-workbench.local/vocab#)
    - coa:suggestedSubClassOf and coa:suggestionReason (proposal-internal; never meaningful to Ontop)
    - Preserves temporal ranges (xsd:date/dateTime/time) — they match R2RML
    - Aligns property ranges with R2RML rr:datatype declarations (prevents MappingOntologyMismatchException)
    """
    from rdflib import OWL, RDF, RDFS, XSD, Graph, Namespace, URIRef

    WB = Namespace("https://ontology-workbench.local/vocab#")
    RR = Namespace("http://www.w3.org/ns/r2rml#")

    g = Graph()
    try:
        g.parse(data=turtle, format="turtle")
    except Exception:
        # If we can't parse, return as-is (best effort)
        return turtle

    # Canonicalize malformed / mis-cased / aliased XSD datatype tokens FIRST, so
    # the exact-cased date/time normalization below actually matches them. Without
    # this, a lowercase ``xsd:datetime`` (or ``xsd:varchar`` etc.) reaches Ontop as
    # an undefined datatype — it slips past the ``(XSD.dateTime, XSD.date,
    # XSD.time)`` match, which is case-sensitive. This is the accept-time SAFETY
    # NET for the served VKG copy: the interactive edit path no longer canonicalizes
    # (user data is persisted verbatim + surfaced for consent-gated repair), so this
    # is what guarantees a *skipped* repair can't push a malformed token to Ontop.
    from coa_ontology.datatype_canonicalizer import canonicalize_datatypes

    changes = canonicalize_datatypes(g)
    if changes:
        _log.info("VKG datatype canonicalization rewrote %d token(s): %s", len(changes), changes)

    # Strip owl:imports (boundary defense for uploaded ontologies): Ontop/OWLAPI
    # network-resolves each import at VKG load and fails every query on a
    # non-dereferenceable foundational URI. Grounding survives via the
    # rdfs:subClassOf + skos:* / coa:groundedTo axioms already in the graph.
    for s, p, o in list(g.triples((None, OWL.imports, None))):
        g.remove((s, p, o))

    # Remove rdf:Property declarations (keep only owl:ObjectProperty / owl:DatatypeProperty)
    for s in list(g.subjects(RDF.type, RDF.Property)):
        g.remove((s, RDF.type, RDF.Property))

    # Remove owl:hasKey axioms and their associated blank-node lists
    for s, p, o in list(g.triples((None, OWL.hasKey, None))):
        _remove_rdf_list(g, o)
        g.remove((s, p, o))

    # Remove workbench-internal metadata
    for s, p, o in list(g.triples((None, None, None))):
        if str(p).startswith(str(WB)):
            g.remove((s, p, o))

    # Strip proposal-internal suggestion triples — these are never meaningful
    # to Ontop/VKG and should not survive into the accepted ontology. Any
    # suggestion the user confirmed was already promoted to rdfs:subClassOf
    # via the UI's save flow before accept was called.
    SCL = Namespace(VOCAB_URI)
    for s, p, o in list(g.triples((None, SCL.suggestedSubClassOf, None))):
        g.remove((s, p, o))
    for s, p, o in list(g.triples((None, SCL.suggestionReason, None))):
        g.remove((s, p, o))

    # Resolve dual-typed properties: if both ObjectProperty and DatatypeProperty,
    # keep only ObjectProperty (OWL 2 QL requires exclusive typing).
    obj_props = set(g.subjects(RDF.type, OWL.ObjectProperty))
    data_props = set(g.subjects(RDF.type, OWL.DatatypeProperty))
    for prop in obj_props & data_props:
        g.remove((prop, RDF.type, OWL.DatatypeProperty))
        for _, _, o in list(g.triples((prop, RDFS.range, None))):
            if str(o).startswith(str(XSD)):
                g.remove((prop, RDFS.range, o))

    # Temporal ranges are left INTACT. They used to be rewritten to xsd:string
    # to match the R2RML rr:datatype, which itself only said xsd:string because
    # the H2 validation schema declared temporal columns as VARCHAR. All three
    # layers now carry the real type, so the alignment below is satisfied
    # without discarding it. Rewriting them made every date FILTER
    # unsatisfiable — see the _CATALOG_TO_H2 comment above.

    # Align ontology property ranges with R2RML rr:datatype declarations.
    # Ontop is strict: if the R2RML says xsd:decimal but the ontology says
    # xsd:string, translation fails with MappingOntologyMismatchException.
    if r2rml_turtle:
        try:
            from coa_ontology.datatype_canonicalizer import _canonical_for

            r2rml_g = Graph()
            r2rml_g.parse(data=r2rml_turtle, format="turtle")
            # Build map: property IRI → rr:datatype from R2RML
            r2rml_datatypes: dict[str, URIRef] = {}
            for pom in r2rml_g.subjects(RR.predicate, None):
                predicate = r2rml_g.value(pom, RR.predicate)
                obj_map = r2rml_g.value(pom, RR.objectMap)
                if predicate and obj_map:
                    datatype = r2rml_g.value(obj_map, RR.datatype)
                    if datatype:
                        # Canonicalize the R2RML token here too: this alignment step
                        # copies rr:datatype into the served ontology range, so a
                        # malformed rr:datatype (e.g. xsd:datetime) would re-inject an
                        # undefined datatype into the VKG copy AFTER the graph-level
                        # canonicalize above already cleaned it. _canonical_for returns
                        # None for already-valid/unknown tokens → keep the original.
                        dt_ref = URIRef(str(datatype))
                        r2rml_datatypes[str(predicate)] = _canonical_for(dt_ref) or dt_ref

            # Update ontology ranges to match R2RML datatypes
            for prop_iri, expected_dt in r2rml_datatypes.items():
                prop_ref = URIRef(prop_iri)
                current_ranges = list(g.objects(prop_ref, RDFS.range))
                if not current_ranges:
                    continue
                for current_range in current_ranges:
                    if str(current_range).startswith(str(XSD)) and current_range != expected_dt:
                        g.remove((prop_ref, RDFS.range, current_range))
                        g.add((prop_ref, RDFS.range, expected_dt))
        except Exception as e:
            # Best effort — don't break publish if R2RML parsing/alignment fails.
            # Bind the exception type + message (in addition to exc_info) so a code
            # bug in the canonicalization above is greppable and distinguishable from
            # a user-fixable parse error, not just a bare "skipped" line.
            _log.warning("r2rml_range_alignment_skipped: %s: %s", type(e).__name__, e, exc_info=True)

    return g.serialize(format="turtle")


def _remove_rdf_list(g, node) -> None:
    """Recursively remove an RDF list (rdf:first/rdf:rest chain)."""
    from rdflib import RDF, URIRef

    if node == RDF.nil or isinstance(node, URIRef):
        return
    first_val = g.value(node, RDF.first)
    rest_val = g.value(node, RDF.rest)
    g.remove((node, RDF.first, first_val))
    g.remove((node, RDF.rest, rest_val))
    if rest_val and rest_val != RDF.nil:
        _remove_rdf_list(g, rest_val)


def _persist_induced_to_s3(ontology_id: str, namespace: str, graph_uri: str) -> None:
    """Persist the full induced ontology + accumulated R2RML to S3.

    Called after each proposal accept. Fetches the current full state:
    - Ontology Turtle from Neptune (the named graph is the source of truth,
      accumulates across all accepted proposals for this ontology)
    - R2RML from all accepted proposals targeting this ontology (concatenated)

    Writes to:
      s3://{bucket}/{prefix}/{namespace}/induced/{safe_id}.ttl
      s3://{bucket}/{prefix}/{namespace}/induced/{safe_id}.r2rml.ttl
      s3://{bucket}/{prefix}/{namespace}/latest/ontology.ttl       (VKG consumption)
      s3://{bucket}/{prefix}/{namespace}/latest/mappings.r2rml     (VKG consumption)
    """
    import boto3

    from coa_ontology.stores.neptune_db_graph import _gsp_get_turtle

    try:
        # 1. Get full ontology turtle from Neptune named graph
        ontology_turtle = _gsp_get_turtle(graph_uri) if graph_uri else None
        if not ontology_turtle:
            _log.warning("s3 persist: no turtle from graph %s, skipping", graph_uri)
            return

        # 2. Accumulate R2RML from all accepted proposals for this ontology
        accepted = dynamo_store.list_proposals(
            namespace=namespace, ontology_id=ontology_id, status=ProposalStatus.ACCEPTED.value
        )
        r2rml_parts = []
        prefixes_seen: set[str] = set()
        for p in accepted:
            r2rml = p.get("r2rml_turtle", "")
            if not r2rml and p.get("r2rml_s3_key"):
                r2rml = dynamo_store._get_artifact_s3(namespace, p["proposal_id"], "r2rml") or ""
            if not r2rml:
                continue
            # Deduplicate @prefix lines across proposals
            lines = []
            for line in r2rml.split("\n"):
                if line.strip().startswith("@prefix"):
                    if line.strip() not in prefixes_seen:
                        prefixes_seen.add(line.strip())
                        lines.append(line)
                else:
                    lines.append(line)
            r2rml_parts.append("\n".join(lines))
        r2rml_turtle = "\n\n".join(r2rml_parts) if r2rml_parts else None

        # Deduplicate R2RML TriplesMaps: multiple proposals for the same ontology
        # can contribute identical TriplesMap definitions. Ontop rejects R2RML with
        # duplicate logicalTable nodes on the same TriplesMap subject.
        if r2rml_turtle:
            try:
                from rdflib import Graph
                from rdflib import Namespace as RdfNamespace

                RR = RdfNamespace("http://www.w3.org/ns/r2rml#")
                r2rml_g = Graph()
                r2rml_g.parse(data=r2rml_turtle, format="turtle")
                for tmap in r2rml_g.subjects(predicate=None, object=RR.TriplesMap):
                    lts = list(r2rml_g.objects(tmap, RR.logicalTable))
                    if len(lts) > 1:
                        for lt in lts[1:]:
                            r2rml_g.remove((tmap, RR.logicalTable, lt))
                            for tn in r2rml_g.objects(lt, RR.tableName):
                                r2rml_g.remove((lt, RR.tableName, tn))
                r2rml_turtle = r2rml_g.serialize(format="turtle")
            except Exception:
                # Best effort — don't break persist if dedup fails.
                _log.warning("r2rml_triplesmap_dedup_skipped", exc_info=True)

        # 3. Write to S3
        short_id = _ontology_short_id(ontology_id)

        s3 = boto3.client("s3", region_name=os.environ.get("AWS_REGION", "us-east-1"))
        base_key = f"{_S3_PREFIX}/{namespace}/induced/{short_id}"

        s3.put_object(
            Bucket=_S3_BUCKET,
            Key=f"{base_key}.ttl",
            Body=ontology_turtle.encode("utf-8"),
            ContentType="text/turtle",
        )
        _log.info("s3 persist: wrote %s.ttl (%d bytes)", base_key, len(ontology_turtle))

        if r2rml_turtle:
            s3.put_object(
                Bucket=_S3_BUCKET,
                Key=f"{base_key}.r2rml.ttl",
                Body=r2rml_turtle.encode("utf-8"),
                ContentType="text/turtle",
            )
            _log.info("s3 persist: wrote %s.r2rml.ttl (%d bytes)", base_key, len(r2rml_turtle))

        # 4. Write to the VKG-expected "latest/" path so the VKG service
        #    can discover and load the ontology + mappings at startup.
        #    The VKG ontology is cleaned for OWL 2 QL compatibility (Ontop).
        latest_key = f"{_S3_PREFIX}/{namespace}/latest"
        vkg_turtle = _clean_ontology_for_vkg(ontology_turtle, r2rml_turtle=r2rml_turtle)
        s3.put_object(
            Bucket=_S3_BUCKET,
            Key=f"{latest_key}/ontology.ttl",
            Body=vkg_turtle.encode("utf-8"),
            ContentType="text/turtle",
        )
        _log.info("s3 persist: wrote %s/ontology.ttl (%d bytes)", latest_key, len(vkg_turtle))

        if r2rml_turtle:
            s3.put_object(
                Bucket=_S3_BUCKET,
                Key=f"{latest_key}/mappings.r2rml",
                Body=r2rml_turtle.encode("utf-8"),
                ContentType="text/turtle",
            )
            _log.info("s3 persist: wrote %s/mappings.r2rml (%d bytes)", latest_key, len(r2rml_turtle))

        # 5. Generate and upload schema.sql from datasource catalog.
        #    Ontop needs schema.sql to validate R2RML column references at startup.
        #    Identifiers are emitted verbatim (SQL-delimited), so they match the
        #    source datasource directly — no canonical→original column map needed.
        #
        #    A generation failure is logged at ERROR but kept best-effort: the
        #    rest of the VKG payload is already written. (Failing the whole accept
        #    on this is coupled to the accept-flow ordering rework — #467.)
        try:
            schema_sql = _generate_schema_sql(namespace, ontology_id)
        except SchemaSqlGenerationError as e:
            schema_sql = None
            _log.error(
                "s3 persist: schema.sql generation FAILED for %s (ns=%s) — Ontop R2RML "
                "column validation will be degraded until re-accept: %s",
                ontology_id,
                namespace,
                e,
            )
        if schema_sql:
            s3.put_object(
                Bucket=_S3_BUCKET,
                Key=f"{latest_key}/schema.sql",
                Body=schema_sql.encode("utf-8"),
                ContentType="text/sql",
            )
            _log.info("s3 persist: wrote %s/schema.sql (%d bytes)", latest_key, len(schema_sql))
    except Exception:
        # RAISE — do NOT swallow (#467). Warning-and-returning here hid a
        # Neptune<->S3 split-brain: the graph is committed and DDB says
        # ``accepted``, but VKG keeps serving the stale ``latest/`` copy with no
        # signal. The caller retries this as the ``s3_persist`` step and logs
        # each attempt, so no log line is needed here.
        raise


def _emit_ontology_published(namespace: str, version: str) -> bool:
    """Emit ontology.published event to EventBridge so VKG reloads.

    RAISES on failure, including a partial ``FailedEntryCount`` (which is a
    per-entry failure EventBridge reports in a 200 response, so it must be
    checked explicitly). As with :func:`_reconcile_counts`, non-fatality is the
    CALLER's declaration via ``_run_accept_step(..., fatal=False)`` — swallowing
    here would mean a single transient ``PutEvents`` blip became a permanent
    skip instead of being retried 3×.

    Why the caller declares it non-fatal (#467): losing this event means S3 holds
    the correct new ontology while the VKG service keeps serving the previous
    payload — real user-visible staleness (Tier-2 answers from the old mappings,
    with no error raised), but the ontology itself is complete and correct, so it
    is not worth failing an otherwise-perfect accept. It is its OWN pipeline step
    (was a silent tail-call inside :func:`_persist_induced_to_s3`), so an
    exhausted retry is recorded as
    ``accept_progress {step: notify_vkg, state: skipped}`` and is greppable.

    GAP (#851): a skipped notify_vkg has no in-product recovery — re-accept
    returns early on ``accepted`` and cannot re-emit. Today an operator forces a
    new VKG ECS deployment. The EventBridge retry/alarm only cover delivery of an
    event that reached EventBridge, not a ``put_events`` that never succeeded.
    """
    import boto3

    events_client = boto3.client("events", region_name=os.environ.get("AWS_REGION", "us-east-1"))
    resp = events_client.put_events(
        Entries=[
            {
                "Source": f"{EVENT_SOURCE_PREFIX}.ontology",
                "DetailType": "ontology.published",
                "Detail": json.dumps(
                    {
                        "namespace": namespace,
                        "version": version,
                        "s3Bucket": _S3_BUCKET,
                        "s3Prefix": f"{_S3_PREFIX}/{namespace}/latest/",
                    }
                ),
            }
        ]
    )
    failed = resp.get("FailedEntryCount", 0)
    if failed:
        raise RuntimeError(f"put_events reported FailedEntryCount={failed}: {resp.get('Entries')}")
    _log.info("eventbridge: emitted ontology.published for namespace=%s version=%s", namespace, version)
    return True


# ── Routes ──────────────────────────────────────────────────────────────


@router.get("/")
def list_proposals(
    namespace: str | None = None,
    ontology_id: str | None = None,
    status: str | None = None,
    type: str | None = None,
    source_type: str | None = None,
    limit: int = 50,
):
    """List proposals with optional filters.

    The ``source_type`` query parameter is forwarded to
    :func:`dynamo_store.list_proposals` unchanged — see that function's
    docstring for the filter semantics, including the backfill rule
    for pre-schema-delta items that lack the attribute. The route does
    not validate the value (no ``Literal`` restriction): unknown values
    simply produce no matches, which lets the route degrade gracefully
    if a client sends a stale enum member.
    """
    items = dynamo_store.list_proposals(
        namespace=namespace,
        ontology_id=ontology_id,
        status=status,
        proposal_type=type,
        source_type=source_type,
        limit=limit,
    )
    # Strip large content fields from list view
    for item in items:
        item.pop("ontology_turtle", None)
        item.pop("r2rml_turtle", None)
    return items


@router.get("/{proposal_id}")
def get_proposal(proposal_id: str, namespace: str = "default"):
    """Get a single proposal.

    Large offloaded artifacts are NOT inlined — each would risk the 6 MB API
    Gateway / Lambda response limit and 500 the call. Instead the response
    carries presigned S3 GET URLs the client fetches directly from S3, keeping
    this response small regardless of proposal size:

    - Turtle (``ontology_url``/``r2rml_url``): a large induced ontology is 6+ MB.
    - Grounding matches (``matches_url``): the match list is one entry per
      grounded table, each with up to ~30 candidate classes (with definitions),
      so a wide schema (thousands of tables) can exceed the cap on its own.

    Each URL is ``null`` when the artifact is absent. ``hydrate_matches=False``
    also spares an S3 read on every poll of this endpoint.

    Legacy proposals that predate the matches offload carry a small (formerly
    ``[:50]``-capped) match list inline in ``metadata.matches`` and have no S3
    object; for those ``matches_url`` is ``null`` and the inline list is left in
    place so the client's inline-else-URL fallback still renders their grounding.
    """
    item = dynamo_store.get_proposal_by_id(
        proposal_id, namespace=namespace, hydrate_turtle=False, hydrate_matches=False
    )
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")
    # Serve Turtle out-of-band via presigned S3 URLs rather than inlining it.
    item.pop("ontology_turtle", None)
    item.pop("r2rml_turtle", None)
    item["ontology_url"] = dynamo_store.presign_proposal_artifact(namespace, proposal_id, "ontology")
    item["r2rml_url"] = dynamo_store.presign_proposal_artifact(namespace, proposal_id, "r2rml")
    # Serve grounding matches out-of-band the same way. Only strip an inline copy
    # when an S3 object exists (modern offloaded proposals): the inline copy is
    # already absent there, so this is defensive. Legacy inline-only proposals
    # (no S3 key -> matches_url is None) keep their small inline list.
    matches_url = dynamo_store.presign_proposal_matches(namespace, proposal_id)
    item["matches_url"] = matches_url
    if matches_url:
        (item.get("metadata") or {}).pop("matches", None)
    return item


# ── Accept (async) ──────────────────────────────────────────────────────
# The accept flow runs Neptune SPARQL projections, GSP bulk Turtle load,
# Bedrock embedding for every class+property, and S3 persistence — easily
# 30-60 seconds for a real proposal, well past API Gateway's 29-second
# integration timeout. The handler returns 202 after flipping the
# proposal status to ``accepting``; a daemon thread completes the work
# and writes the terminal status (``accepted``, or ``accept_failed`` with
# ``accept_error`` naming the failed step) to DDB. Clients observe
# progress via ``GET /proposals/{id}``.


def _reconcile_counts(ontology_id: str, namespace: str, graph_uri: str) -> bool:
    """Overwrite DDB registry counts with actual Neptune graph counts (deduped).

    RAISES on failure. Non-fatality is NOT decided here — it is declared by the
    caller via ``_run_accept_step(..., fatal=False)``, which retries this 3× and
    only then checkpoints ``skipped`` and lets the accept finish. Swallowing here
    instead would defeat that retry (the wrapper would never see a failure), and
    would put the blast-radius decision in the step rather than at the pipeline
    call site where it is reviewable.

    Why the caller declares it non-fatal (#467): a failure leaves the registry
    holding the ingest-derived (additive) counts, so the Catalog tab can show
    inflated class/property numbers for an otherwise-correct ontology. Misleading
    but not corrupting, and the next accept recomputes it — not worth blocking an
    accept whose graph, embeddings and S3 payload are all good.

    Returns ``False`` when there is genuinely nothing to reconcile (no
    ``graph_uri``, or the backend reports no counts — see
    :meth:`NAGraphStore.count_classes_and_properties`), which the caller
    checkpoints as ``skipped`` rather than a false ``done``.
    """
    if not graph_uri:
        return False
    from coa_ontology.stores import build_stores

    graph_store, _ = build_stores(namespace=namespace)
    counts = graph_store.count_classes_and_properties(ontology_id)
    if not counts:
        return False
    dynamo_store.update_ontology_registry(
        namespace, ontology_id, class_count=counts["classes"], property_count=counts["properties"]
    )
    return True


class _AcceptStepError(Exception):
    """A step in the accept pipeline failed after exhausting retries.

    Carries the step name so the terminal ``accept_failed`` status can tell the
    user exactly which stage broke (Neptune load, embeddings, S3 persist, …).
    """

    def __init__(self, step: str, message: str):
        self.step = step
        super().__init__(message)


# Errors that mean the PROPOSAL itself is bad — retrying can't help, the user
# must fix the Turtle/validation and re-accept. Everything else (store/network
# /timeout/5xx) is treated as transient and retried.
_ACCEPT_STRUCTURAL_ERRORS = (IngestParseError, IngestValidationError)


def _checkpoint_accept_step(proposal_id: str, namespace: str, step: str, state: str) -> None:
    """Record ``{step, state}`` in ``accept_progress`` — and heartbeat ``updated_at``.

    NOT resume state: nothing reads ``accept_progress``, so a re-accept restarts at
    ``ingest`` rather than continuing from here (#851). It is diagnostics plus the
    heartbeat.

    Best-effort by design: a progress-write hiccup must never fail a step whose
    actual work succeeded. The heartbeat is the load-bearing side effect (the
    stale-sweep in :mod:`dynamo_store` keys off ``updated_at``), so this is
    called on both the success and the exhausted-non-fatal path.
    """
    try:
        dynamo_store.update_proposal(proposal_id, namespace=namespace, accept_progress={"step": step, "state": state})
    except Exception:  # noqa: BLE001
        _log.warning("accept[%s]: progress checkpoint for step %s failed (non-fatal)", proposal_id, step)


def _run_accept_step(
    step: str,
    proposal_id: str,
    namespace: str,
    fn: Callable[[], Any],
    *,
    fatal: bool = True,
) -> Any:
    """Run one accept-pipeline step with bounded retry + progress checkpoint.

    ``fn`` is a zero-arg callable performing the step. On success the step is
    recorded in the proposal's ``accept_progress`` map (which also refreshes
    ``updated_at`` — the heartbeat the stale-sweep keys off, so a long but live
    accept isn't mistaken for a dead worker). Transient errors are retried up to
    ``_ACCEPT_STEP_MAX_ATTEMPTS`` with exponential backoff; a structural error
    (:data:`_ACCEPT_STRUCTURAL_ERRORS`) fails immediately.

    ``fatal`` declares the blast radius of this step AT THE CALL SITE, which is
    where a reviewer looks — the pipeline in :func:`_run_accept_proposal` reads
    top-to-bottom as which stages can sink an accept and which cannot:

    * ``fatal=True`` (default) — retries exhausted raises :class:`_AcceptStepError`
      naming the step, so the accept lands ``accept_failed`` / ``<step>``.
    * ``fatal=False`` — retries exhausted is swallowed and checkpointed
      ``{step, state: skipped}``; the accept continues. Use ONLY for steps whose
      failure leaves the ontology itself correct (see :func:`_reconcile_counts`,
      :func:`_emit_ontology_published`).

    Either way the step gets the SAME retry. That is the point of routing
    non-fatal steps through here too: previously they caught their own exception
    and returned ``False``, so this loop never saw a failure and the documented
    3-attempt retry silently never happened — one EventBridge blip was a
    permanent skip. Steps must therefore RAISE on failure, not return ``False``.

    A step returning ``False`` is still checkpointed ``skipped`` rather than
    ``done``, for steps that legitimately report "did not complete" without
    failing (:func:`wait_for_embeddings_searchable` returns ``False`` on a stalled
    index and never raises). Any other return value — including ``None`` —
    counts as ``done``.
    """
    attempt = 0
    while True:
        attempt += 1
        try:
            result = fn()
            # Checkpoint + heartbeat. Best-effort: a progress-write hiccup must
            # not fail an otherwise-successful step.
            state = "skipped" if result is False else "done"
            _checkpoint_accept_step(proposal_id, namespace, step, state)
            if state == "skipped":
                _log.warning("accept[%s]: step %s did not complete (non-fatal, accept continues)", proposal_id, step)
            return result
        except _ACCEPT_STRUCTURAL_ERRORS:
            raise  # bad proposal — surfaced by the caller, never retried
        except Exception as e:  # noqa: BLE001 — transient step failure
            if attempt >= _ACCEPT_STEP_MAX_ATTEMPTS:
                _log.error(
                    "accept[%s]: step %s failed after %d attempts: %s: %s",
                    proposal_id,
                    step,
                    attempt,
                    type(e).__name__,
                    e,
                )
                if not fatal:
                    # Declared non-fatal: record the truth and let the accept
                    # finish. accept_progress must never claim ``done`` here.
                    _checkpoint_accept_step(proposal_id, namespace, step, "skipped")
                    _log.warning(
                        "accept[%s]: step %s exhausted retries but is non-fatal; accept continues", proposal_id, step
                    )
                    return False
                raise _AcceptStepError(step, f"{type(e).__name__}: {e}") from e
            # Jitter matches the idiom used by every other backoff in the repo
            # (e.g. libs/common opensearch/retry.py, embeddings.py): without it,
            # two accept workers that hit the same Neptune/AOSS blip retry in
            # lockstep at exactly the same instants and re-collide. Scaled by the
            # configured base so setting the base to 0 (tests) stays instant.
            base = _ACCEPT_STEP_BACKOFF_SECONDS * (2 ** (attempt - 1))
            backoff = base + random.uniform(0, _ACCEPT_STEP_BACKOFF_SECONDS / 2)
            _log.warning(
                "accept[%s]: step %s attempt %d/%d failed (%s: %s); retrying in %.1fs",
                proposal_id,
                step,
                attempt,
                _ACCEPT_STEP_MAX_ATTEMPTS,
                type(e).__name__,
                e,
                backoff,
            )
            _t.sleep(backoff)


def _run_accept_proposal(
    proposal_id: str,
    namespace: str,
    target_ontology_id: str,
    label: str,
    turtle: str,
    proposal_job_id: str,
    r2rml_turtle: str = "",
) -> None:
    """Background worker for proposal accept.

    Runs the accept pipeline as discrete, individually-retried steps
    (Neptune load + embeddings via ``ingest_ontology`` → embeddings-searchable
    wait → local cache → DDB flip → count reconcile → S3 persist), checkpointing
    each into ``accept_progress`` (which also heartbeats ``updated_at``). A step
    that exhausts its retries moves the proposal to the terminal
    ``accept_failed`` state with an ``accept_error`` naming the step, so the UI
    can surface it and the user can fix the root cause and re-accept. Re-running
    converges rather than corrupting: Neptune ``INSERT DATA`` dedups identical
    triples (ingest append semantics), and the count-reconcile step SETs the
    deduped Neptune cardinality (not an additive ADD), so class/property counts
    self-heal on every accept, and the S3 persist that #467 used to swallow is
    now a retried, failure-propagating step.

    Converges, does NOT resume (#851): every re-accept restarts at ``ingest``, so
    a failure in the last step still re-runs the Neptune load and the full
    embedding pass. Costly, and the AOSS duplication below is the price.

    KNOWN LIMITATION (tracked by #594, not fixed here): AOSS has no
    client-supplied ``_id`` and no per-entity delete — only an ontology-wide
    ``delete_embeddings_for_ontology`` — so re-accepting can leave duplicate
    embedding docs. We deliberately do NOT pre-clean embeddings in the accept
    path: on the append-merge flow that would wipe co-merged proposals'
    embeddings. AOSS dedupe belongs to the cancel/reject cleanup (#594).
    """
    from coa_ontology.stores import build_stores

    try:
        graph_store, vector_store = build_stores(namespace=namespace)
        md = IngestMetadata(
            ontology_id=target_ontology_id,
            title=label,
            description=f"Accepted from proposal {proposal_id}",
            ontology_type="induced",
            format="turtle",
            domain_tags=["induced"],
            source="proposal-accept",
        )

        # ── Step: ingest (Neptune load + embedding accumulation) ──────────
        # ingest_ontology bundles parse → Neptune load_turtle → embedding
        # accumulation, retried as one unit (splitting it would mean
        # re-implementing the pipeline here).
        #
        # RETRY IS NOT FULLY IDEMPOTENT — known, bounded, not papered over:
        #   - Neptune triples dedup, so the graph is always correct.
        #   - class_count / property_count inflate per attempt, but reconcile_counts
        #     below overwrites them with deduped Neptune counts — self-healing.
        #   - axiom_count / embedding_count are ADD-only, so they stay high.
        #     Display-only.
        #   - AOSS embeddings DUPLICATE (no client-supplied _id — see
        #     opensearch_vector), degrading recall rather than breaking it. #824.
        # No local fix: append mode shares one ontology across proposals, so a
        # pre-retry delete_embeddings_for_ontology would wipe co-tenant embeddings.
        # This is also why the no-resume gap (#851) is expensive: a failure in the
        # LAST step re-runs this whole step, duplicating embeddings for nothing.
        result = _run_accept_step(
            "ingest",
            proposal_id,
            namespace,
            lambda: ingest_ontology(
                graph_store=graph_store,
                vector_store=vector_store,
                content=turtle,
                namespace=namespace,
                validate=False,
                allow_append=True,
                source_proposal_id=proposal_id,
                r2rml_content=r2rml_turtle,
                metadata=md,
            ),
        )

        # ── Step: wait until embeddings are searchable — NON-FATAL ─────────
        # AOSS is eventually consistent; a later induction grounding on this
        # ontology (or a serve request) could miss just-written embeddings.
        # Surface as ``embeddings_sync`` so a poller shows "Syncing embeddings…".
        #
        # fatal=False is explicit rather than incidental: wait_for_embeddings_
        # searchable NEVER raises — it returns False on a stalled index — so this
        # step could not fail an accept no matter what was declared here. Saying
        # fatal=False makes the pipeline honest about that. It is also the right
        # policy: the embeddings are durably written and only their search
        # visibility lagged, so failing the accept would strand a correct
        # ontology. A stall is checkpointed ``skipped`` and logged.
        embedded_uris = (result.get("embeddings") or {}).get("entity_uris", [])
        dynamo_store.update_proposal(proposal_id, namespace=namespace, status=PROPOSAL_STATUS_EMBEDDINGS_SYNC)
        _run_accept_step(
            "embeddings_searchable",
            proposal_id,
            namespace,
            lambda: wait_for_embeddings_searchable(
                vector_store=vector_store,
                ontology_id=result.get("ontology_id", target_ontology_id),
                namespace=namespace,
                expected_uris=embedded_uris,
                # Heartbeat from INSIDE the wait. This is the one step with no
                # total wall-clock cap (its timeout is a no-progress deadline
                # that resets on every advance), so a large-but-healthy sync can
                # run past PROPOSAL_ACCEPT_STALE_SECONDS. Without a heartbeat here
                # the stale-sweep would flip a LIVE accept to accept_failed, and a
                # re-accept would then run a second worker concurrently with the
                # first. Checkpointing per advance keeps updated_at fresh.
                on_progress=lambda done, total: _checkpoint_accept_step(
                    proposal_id, namespace, "embeddings_searchable", f"syncing {done}/{total}"
                ),
            ),
            fatal=False,
        )

        # ── Cache Turtle to local disk (non-fatal, not a pipeline step) ───
        # /download serves this; if it fails the ontology is still registered +
        # loaded and /download 404s until the next re-accept. Best-effort, so
        # it's intentionally NOT wrapped in _run_accept_step (never fails accept).
        storage_path = os.getenv("ONTOLOGY_STORAGE_PATH", "./data/ontologies")
        safe_name = target_ontology_id.replace("/", "_").replace("#", "_").replace(":", "_").replace("?", "_")
        file_dest = os.path.join(storage_path, f"{safe_name}.ttl")
        try:
            os.makedirs(storage_path, exist_ok=True)
            with open(file_dest, "w", encoding="utf-8") as f:
                f.write(turtle)
            graph_store.update_ontology(target_ontology_id, {"file_path": file_dest})
            _log.info("accept[%s]: cached turtle to %s (%d bytes)", proposal_id, file_dest, len(turtle))
        except Exception as e:  # noqa: BLE001 — file-cache failures are non-fatal
            _log.warning(
                "accept[%s]: local cache failed (non-fatal): %s: %s",
                proposal_id,
                type(e).__name__,
                e,
            )

        # ── Step: flip DDB to accepted ────────────────────────────────────
        # Only after ingest + searchable succeeded. ``accept_error`` is cleared so
        # a previously-failed proposal that now succeeds carries no stale error.
        # ``accept_progress`` is NOT cleared here: _run_accept_step checkpoints
        # this very step immediately after, and the later reconcile_counts /
        # s3_persist / notify_vkg steps checkpoint after that, so it always
        # reflects the last completed step rather than being empty.
        _run_accept_step(
            "flip_accepted",
            proposal_id,
            namespace,
            lambda: dynamo_store.update_proposal(
                proposal_id, namespace=namespace, status=ProposalStatus.ACCEPTED.value, accept_error=""
            ),
        )

        # ── Step: mirror job status — NON-FATAL by design (#467) ───────────
        # A pipeline step rather than a bare try/except, so it gets the same
        # 3-attempt retry and a real checkpoint. A failure only leaves the job row
        # showing its pre-accept status while the proposal itself is correctly
        # ``accepted`` — cosmetic, and the proposal row is what the UI reads — so
        # it must not fail the accept. It is recorded as
        # ``{step: job_status_mirror, state: skipped}`` instead of vanishing into
        # a log warning.
        _run_accept_step(
            "job_status_mirror",
            proposal_id,
            namespace,
            lambda: dynamo_store.update_job_status(proposal_job_id, "ingested", namespace=namespace),
            fatal=False,
        )

        # ── Step: reconcile counts — NON-FATAL by design (#467) ───────────
        # fatal=False: retried 3× like every other step, but an exhausted retry
        # is checkpointed ``skipped`` and the accept still succeeds. A failure
        # only leaves inflated Catalog-tab counts (the next accept recomputes
        # them), so it must NOT block an accept whose graph/embeddings/S3 are
        # good — but it is never recorded as a false ``done``.
        _run_accept_step(
            "reconcile_counts",
            proposal_id,
            namespace,
            lambda: _reconcile_counts(target_ontology_id, namespace, result.get("graph_uri", "")),
            fatal=False,
        )

        # ── Step: persist induced ontology + R2RML to S3 for VKG — FATAL (#467)
        # Previously swallowed on failure, silently stranding VKG on a stale
        # copy. Now a retried, failure-propagating step: on exhaustion the accept
        # lands ``accept_failed`` / ``s3_persist``. This also covers schema.sql
        # generation, which raises SchemaSqlGenerationError (#457) rather than
        # masking a real failure as the "nothing to describe" None — without
        # schema.sql Ontop cannot validate the R2RML and Tier-2 SQL breaks at
        # query time, so a structured ontology must not be accepted without one.
        _run_accept_step(
            "s3_persist",
            proposal_id,
            namespace,
            lambda: _persist_induced_to_s3(target_ontology_id, namespace, result.get("graph_uri", "")),
        )

        # ── Step: notify VKG to reload — NON-FATAL by design (#467) ────────
        # Own step (was a silent tail-call inside _persist_induced_to_s3) so it
        # gets the same retry as everything else. fatal=False: if it still fails,
        # S3 is correct but VKG serves the previous payload until something else
        # triggers a reload: recoverable, so the accept succeeds and the lost
        # signal is recorded as ``{step: notify_vkg, state: skipped}`` instead of
        # disappearing into a log warning.
        _run_accept_step(
            "notify_vkg",
            proposal_id,
            namespace,
            lambda: _emit_ontology_published(namespace, _ontology_short_id(target_ontology_id)),
            fatal=False,
        )
    # The two structural handlers below must stay in sync with
    # ``_ACCEPT_STRUCTURAL_ERRORS``: a type added there but not here still fails
    # the accept, but lands in the generic handler as step="unknown" with a full
    # traceback instead of a user-actionable message. They are spelled out rather
    # than collapsed to ``except _ACCEPT_STRUCTURAL_ERRORS`` so each keeps its own
    # wording (a parse error and a validation error read very differently).
    except IngestParseError as e:
        _accept_fail(proposal_id, namespace, f"Invalid ontology Turtle: {e}", step="ingest")
    except IngestValidationError as e:
        msg = str(e) or "Validation failed"
        _accept_fail(proposal_id, namespace, msg, step="ingest")
    except _AcceptStepError as e:
        _accept_fail(proposal_id, namespace, f"Step '{e.step}' failed: {e}", step=e.step)
    except Exception as e:  # noqa: BLE001 — surface any other failure
        # Kept alongside _accept_fail's ERROR line: this one carries the
        # traceback, which an unexpected (unclassified) failure needs.
        _log.exception("accept[%s]: unexpected error", proposal_id)
        _accept_fail(proposal_id, namespace, f"{type(e).__name__}: {e}", step="unknown")


def _accept_fail(proposal_id: str, namespace: str, error: str, step: str = "unknown") -> None:
    """Move a failed accept to the terminal ``accept_failed`` state.

    The user observes this via the next ``GET /proposals/{id}`` poll: the status
    is ``accept_failed`` (distinct from ``pending``, so the UI can say "accept
    failed on step X — fix and retry" rather than showing a fresh proposal) and
    ``accept_error`` carries the reason. ``accept_failed`` is re-acceptable: the
    accept guard treats it like ``pending``/``updated``. Re-running CONVERGES but
    does NOT resume — it restarts at ``ingest`` no matter how far the failed
    attempt got, because nothing reads ``accept_progress`` (#851).

    ``step`` names the failing pipeline stage (ingest, embeddings_searchable,
    reconcile_counts, s3_persist, …) and is recorded on ``accept_progress`` for
    diagnosis. We swallow the rollback write error so the worker doesn't crash on
    a transient DDB hiccup; the stale-sweep is the backstop if this write is lost.

    Error messages are truncated to 1000 chars to avoid DynamoDB
    ValidationException on large error payloads (e.g. BulkIndexError with
    embedded vectors).

    Logs at ERROR before writing. This is the single choke point for every
    terminal-failure path (parse, validation, exhausted step retries, and the
    catch-all), so logging here means a failed accept is always visible in
    CloudWatch — without it an operator only sees the route's 202 and no trace of
    why the accept never completed.

    NEVER DOWNGRADES AN ALREADY-``accepted`` PROPOSAL. ``flip_accepted`` runs
    BEFORE ``reconcile_counts``/``s3_persist``/``notify_vkg`` (it has to: those
    steps read ``list_proposals(status="accepted")`` to accumulate this
    proposal's own R2RML and datasource ids). So a post-flip step failure hits a
    proposal whose graph, embeddings and registry row are already live and
    serving. Reverting it to ``accept_failed`` would be actively harmful: four
    call sites select on ``status="accepted"`` — ``_generate_schema_sql``,
    ``_persist_induced_to_s3``, ``list_induced_datasources`` and the
    duplicate-detection scan — so the reverted proposal silently drops out of
    them, and the NEXT accept into the same ontology rebuilds
    ``latest/mappings.r2rml`` WITHOUT its mappings. That is silent mapping loss,
    the opposite of what #467 set out to fix.

    The write is therefore conditional on the status still being non-terminal.
    When it is already ``accepted`` the status is left alone and only the
    diagnostic fields are recorded, so the failed step is still greppable and
    visible on the proposal while the live ontology keeps its ``accepted``
    identity. A re-accept re-runs the tail steps (they are overwrites).
    """
    _log.error("accept[%s]: FAILED on step '%s': %s", proposal_id, step, str(error)[:500])
    error_msg = str(error)[:1000]
    try:
        dynamo_store.update_proposal(
            proposal_id,
            namespace=namespace,
            status=PROPOSAL_STATUS_ACCEPT_FAILED,
            accept_error=error_msg,
            accept_progress={"step": step, "state": "failed"},
            expected_status_not="accepted",
        )
    except ClientError as e:
        if e.response.get("Error", {}).get("Code") != "ConditionalCheckFailedException":
            _log.exception("accept[%s]: failed to record accept_failed status", proposal_id)
            return
        # Already ``accepted``: the ontology is live. Keep the status, record the
        # failed tail step so it is still diagnosable and re-runnable.
        _log.error(
            "accept[%s]: step '%s' failed AFTER the proposal was accepted — the ontology is "
            "live but this step did not complete; re-accept to re-run it",
            proposal_id,
            step,
        )
        try:
            dynamo_store.update_proposal(
                proposal_id,
                namespace=namespace,
                accept_error=error_msg,
                accept_progress={"step": step, "state": "failed"},
            )
        except Exception:  # noqa: BLE001 — diagnostics only; the log line above stands
            _log.exception("accept[%s]: failed to record post-accept step failure", proposal_id)
    except Exception:  # noqa: BLE001
        _log.exception("accept[%s]: failed to record accept_failed status", proposal_id)


@router.post("/{proposal_id}/accept", status_code=202, response_model=AcceptProposalResponse)
def accept_proposal(
    proposal_id: str,
    body: AcceptProposalRequest | None = None,
    namespace: str = "default",
) -> AcceptProposalResponse:
    """Kick off proposal acceptance asynchronously.

    Returns 202 immediately with the proposal flipped to ``accepting``.
    The background worker merges the proposal's Turtle into the target
    ontology, then writes the terminal status to DDB. Clients poll
    ``GET /proposals/{id}`` and watch ``status`` transition to
    ``accepted`` (success) or ``accept_failed`` with ``accept_error``
    naming the failed step. ``accept_failed`` is re-acceptable.

    ``body.ontology_id`` selects the target. When omitted the proposal's
    own embedded ``ontology_id`` is used (one-to-one mapping, default
    behaviour). Multiple proposals can name the same target, in which
    case their Turtle content is merged into a single per-ontology named
    graph and their proposal IDs accumulate on a single Dynamo registry
    row's ``source_proposals`` list.

    Idempotent for already-accepted proposals: returns ``status:
    accepted`` instead of starting a new job, so the UI can treat the
    response identically to a successful poll.
    """
    item = dynamo_store.get_proposal_by_id(proposal_id, namespace=namespace)
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")
    status = item.get("status")
    if status == ProposalStatus.ACCEPTED.value:
        return AcceptProposalResponse(proposal_id=proposal_id, status=ProposalStatus.ACCEPTED.value)
    if status == ProposalStatus.REJECTED.value:
        raise HTTPException(409, "Cannot accept a rejected proposal")
    if status in _ACCEPT_IN_PROGRESS_STATUSES:
        raise HTTPException(409, "Accept already in progress for this proposal")
    # ``accept_failed`` falls through to the accept path below — it is
    # re-acceptable, same as pending/updated. Re-running converges on a correct end
    # state but re-executes every step from ``ingest`` (no resume — see #851).

    # Prefer user-reviewed Turtle if the user revised the proposal.
    reviewed = (
        dynamo_store.get_reviewed_proposal(item["job_id"], kind="ontology", namespace=namespace)
        if hasattr(dynamo_store, "get_reviewed_proposal")
        else None
    )
    if reviewed and reviewed.get("ontology_turtle"):
        turtle = reviewed["ontology_turtle"]
    else:
        turtle = item.get("ontology_turtle", "")
    if not turtle:
        raise HTTPException(409, "Proposal has no ontology_turtle")

    # R2RML for this proposal (hydrated from S3 by get_proposal_by_id). Passed
    # into ingest so it can derive the per-class Tier-2 answerability marker
    # (coa:isMapped in Neptune + data_source_id in AOSS). Prefer the reviewed
    # copy if the steward revised it; unstructured proposals have no R2RML ("").
    if reviewed and reviewed.get("r2rml_turtle"):
        r2rml_turtle = reviewed["r2rml_turtle"]
    else:
        r2rml_turtle = item.get("r2rml_turtle", "")

    explicit_target = body.ontology_id if body else None
    target_ontology_id = explicit_target or item.get("ontology_id", "")
    if not target_ontology_id:
        raise HTTPException(
            422,
            "Cannot determine target ontology_id: neither the request body nor the proposal carries one.",
        )

    # Guard: enforce one ontology per namespace (the product invariant — the UI
    # induces every source under a namespace-derived prefix, so all sources in a
    # namespace share one ontology_id and merge into a single VKG payload).
    #
    # A caller (direct API) *can* induce sources under DISTINCT
    # ``ontology_uri_prefix`` values, producing distinct ontology_ids in one
    # namespace. Accepting the second then used to OVERWRITE the first source's
    # per-namespace ``latest/`` VKG payload (single Ontop instance serves one
    # payload per namespace), silently breaking Tier-2 for the first source.
    # Rather than tolerate that by merging mismatched ontologies, reject it
    # loudly: a namespace holds exactly one induced ontology.
    #
    # The explicit ``body.ontology_id`` override is the deliberate
    # "merge this proposal into target ontology X" path — honoured as-is,
    # since the caller is naming the target on purpose.
    if explicit_target is None:
        # Scope to INDUCED ontologies only. Foundational/reference ontologies
        # (loaded for grounding) are also ONTOLOGY# registry rows, but they are
        # MEANT to coexist with the induced one — the invariant is "one *induced*
        # ontology per namespace". Without this filter, loading FIBO before the
        # first induction would make existing_ids non-empty and wrongly 409-reject
        # that first legitimate induced accept.
        existing_ids: set[str] = set()
        for o in dynamo_store.list_ontologies_registry(namespace, ontology_type="induced"):
            oid = o.get("ontology_id")
            if oid:
                existing_ids.add(str(oid))
        if existing_ids and target_ontology_id not in existing_ids:
            raise HTTPException(
                409,
                (
                    f"Namespace '{namespace}' already contains ontology "
                    f"'{sorted(existing_ids)[0]}'; a namespace holds exactly one induced "
                    f"ontology. This proposal was induced under a different ontology URI "
                    f"('{target_ontology_id}') and would overwrite the existing one's "
                    f"served mappings. Re-induce under the namespace's ontology prefix, "
                    f"or pass ontology_id explicitly to merge into the existing ontology."
                ),
            )

    # Fall back to a source-appropriate default only if the proposal never
    # stored a label (rare — induction always sets one).
    _default_label = (
        "Induced Unstructured Ontology" if item.get("source_type") == "UNSTRUCTURED" else "Induced Structured Ontology"
    )
    label = (item.get("metadata") or {}).get("label", _default_label)

    # Mark the proposal as accept-in-progress so concurrent UI polls + the
    # in-progress check above see the right state. Clear any stale
    # ``accept_error`` from a prior failed attempt at the same time. If
    # the worker fails, ``_accept_fail`` moves this to ``accept_failed``
    # with the new error (still re-acceptable).
    #
    # ``expected_status`` makes the transition atomic in DDB. The earlier
    # ``get → check → update`` sequence had a small race window: two
    # concurrent accepts both reading ``pending``, both writing
    # ``accepting``, both spawning workers. The conditional update
    # collapses that window — only one writer can transition the row,
    # the other gets ``ConditionalCheckFailedException`` which we
    # surface as 409 (same response the up-front check would have
    # returned). Cost: one extra ``ConditionExpression`` in the same
    # request, no extra round-trip.
    try:
        dynamo_store.update_proposal(
            proposal_id,
            namespace=namespace,
            status="accepting",
            accept_error="",
            expected_status=status,
        )
    except ClientError as e:
        if e.response.get("Error", {}).get("Code") == "ConditionalCheckFailedException":
            raise HTTPException(
                409,
                "Proposal status changed concurrently — refresh and retry",
            ) from e
        raise

    Thread(
        target=_run_accept_proposal,
        args=(
            proposal_id,
            namespace,
            target_ontology_id,
            label,
            turtle,
            item["job_id"],
            r2rml_turtle,
        ),
        daemon=True,
    ).start()
    return AcceptProposalResponse(proposal_id=proposal_id, status="accepting")


@router.post("/{proposal_id}/cancel")
def cancel_proposal(proposal_id: str, namespace: str = "default"):
    """Cancel a pending/updated proposal so a new induction can proceed."""
    item = dynamo_store.get_proposal_by_id(proposal_id, namespace=namespace)
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")
    # ``accept_failed`` is cancellable: before this state existed a failed accept
    # rolled back to ``pending``, so it was. The UI enables Cancel for it too
    # (isTerminal is only accepted/rejected/cancelled), so omitting it here would
    # 409 a button the user is invited to press. It is a reviewable, discardable
    # proposal like any other non-terminal one.
    if item.get("status") not in (
        ProposalStatus.PENDING.value,
        ProposalStatus.UPDATED.value,
        PROPOSAL_STATUS_ACCEPT_FAILED,
    ):
        raise HTTPException(409, f"Cannot cancel a {item.get('status')} proposal")
    dynamo_store.update_proposal(proposal_id, namespace=namespace, status=ProposalStatus.CANCELLED.value)
    return {"proposal_id": proposal_id, "status": "cancelled"}


@router.post("/reject")
def reject_proposals(body: RejectRequest, namespace: str = "default"):
    """Reject one or many proposals with a reason."""
    results = []
    for pid in body.proposal_ids:
        item = dynamo_store.get_proposal_by_id(pid, namespace=namespace)
        if not item:
            results.append({"proposal_id": pid, "error": "not found"})
            continue
        if item.get("status") == ProposalStatus.ACCEPTED.value:
            results.append({"proposal_id": pid, "error": "already accepted"})
            continue
        dynamo_store.update_proposal(
            pid, namespace=namespace, status=ProposalStatus.REJECTED.value, reject_reason=body.reason
        )
        results.append({"proposal_id": pid, "status": "rejected"})
    return results


@router.post("/{proposal_id}/upload-url", response_model=ProposalUploadUrlResponse)
def get_proposal_upload_urls(proposal_id: str, namespace: str = "default"):
    """Presigned S3 PUT URLs for uploading edited proposal Turtle directly to S3.

    The mirror of the presigned-GET read path: lets the browser upload a large
    edited ontology / R2RML straight to S3, bypassing the 6 MB API Gateway /
    Lambda *request* cap. After uploading, the client calls ``POST /update``
    with ``ontology_uploaded`` / ``r2rml_uploaded`` set (instead of sending the
    Turtle inline).
    """
    item = dynamo_store.get_proposal_by_id(proposal_id, namespace=namespace, hydrate_turtle=False)
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")
    if item.get("status") in (ProposalStatus.ACCEPTED.value, ProposalStatus.REJECTED.value):
        raise HTTPException(409, f"Cannot update a {item['status']} proposal")
    return ProposalUploadUrlResponse(
        ontology_put_url=dynamo_store.presign_proposal_artifact_put(namespace, proposal_id, "ontology"),
        r2rml_put_url=dynamo_store.presign_proposal_artifact_put(namespace, proposal_id, "r2rml"),
    )


@router.post("/update")
def update_proposal(body: UpdateProposalRequest, namespace: str = "default"):
    """Upload a modified proposal (user revision) or apply grounding overrides.

    When ``grounding_overrides`` is present, the backend patches the stored
    matches, re-fetches catalog tables, and rebuilds the Turtle + R2RML.
    This takes priority over raw ``ontology_turtle`` if both are sent.
    """
    item = dynamo_store.get_proposal_by_id(body.proposal_id, namespace=namespace)
    if not item:
        raise HTTPException(404, f"Proposal '{body.proposal_id}' not found")
    if item.get("status") in (ProposalStatus.ACCEPTED.value, ProposalStatus.REJECTED.value):
        raise HTTPException(409, f"Cannot update a {item['status']} proposal")

    if body.grounding_overrides:
        return _apply_grounding_overrides(body.proposal_id, body.grounding_overrides, item, namespace)

    # Turtle bodies are offloaded to S3 (only the S3 key is stored on the
    # DynamoDB item) — a large edited ontology written inline would trip the
    # 400 KB item limit, same failure mode as the grounding-override path.
    #
    # The edited Turtle arrives one of two ways:
    #   * inline ``*_turtle`` — fine for small ontologies, but the request
    #     itself is capped at the 6 MB API Gateway / Lambda limit; or
    #   * ``*_uploaded`` — the client PUT the artifact straight to its S3 key
    #     via a presigned URL (see ``get_proposal_upload_urls``), bypassing
    #     that request cap (the mirror of the presigned-GET read path). The
    #     server reads it back here to validate before committing the pointer.
    def _validate_turtle(ttl: str, what: str) -> None:
        from rdflib import Graph as RdfGraph

        try:
            RdfGraph().parse(data=ttl, format="turtle")
        except Exception as e:
            raise HTTPException(422, f"Invalid {what} Turtle: {e}") from e

    # NOTE: we deliberately do NOT canonicalize datatype tokens here. Editing a
    # proposal persists the user's Turtle verbatim — silently rewriting their
    # data at the write boundary is the anti-pattern this feature removes. Any
    # malformed/mis-cased/aliased token (xsd:datetime, xsd:varchar, …) is instead
    # surfaced by DatatypeTokenValidator at validation time and fixed only via the
    # consent-gated /repair-datatypes action. (The served VKG copy is still
    # sanitized at accept time in _clean_ontology_for_vkg, so a skipped repair
    # can't break Ontop queries.)
    updates: dict = {}

    if body.ontology_turtle is not None:
        _validate_turtle(body.ontology_turtle, "ontology")
        updates["ontology_s3_key"] = dynamo_store._put_artifact_s3(
            namespace, body.proposal_id, "ontology", body.ontology_turtle
        )
    elif body.ontology_uploaded:
        ttl = dynamo_store._get_artifact_s3(namespace, body.proposal_id, "ontology")
        if ttl is None:
            raise HTTPException(400, "ontology_uploaded was set but no uploaded artifact was found in S3")
        _validate_turtle(ttl, "ontology")
        # Already stored at its key by the presigned upload; nothing to re-store.
        updates["ontology_s3_key"] = dynamo_store._proposal_s3_key(namespace, body.proposal_id, "ontology")

    if body.r2rml_turtle is not None:
        _validate_turtle(body.r2rml_turtle, "R2RML")
        updates["r2rml_s3_key"] = dynamo_store._put_artifact_s3(namespace, body.proposal_id, "r2rml", body.r2rml_turtle)
    elif body.r2rml_uploaded:
        ttl = dynamo_store._get_artifact_s3(namespace, body.proposal_id, "r2rml")
        if ttl is None:
            raise HTTPException(400, "r2rml_uploaded was set but no uploaded artifact was found in S3")
        _validate_turtle(ttl, "R2RML")
        updates["r2rml_s3_key"] = dynamo_store._proposal_s3_key(namespace, body.proposal_id, "r2rml")

    if not updates:
        raise HTTPException(
            400, "Provide ontology_turtle/r2rml_turtle, ontology_uploaded/r2rml_uploaded, or grounding_overrides"
        )

    updates["status"] = "updated"
    dynamo_store.update_proposal(body.proposal_id, namespace=namespace, **updates)

    return {"proposal_id": body.proposal_id, "status": "updated"}


class RepairDatatypesRequest(BaseModel):
    """Request body for ``POST /proposals/{id}/repair-datatypes``."""

    # When True, compute + return the changes WITHOUT persisting them. Used by the
    # accept-time guard to check "does this proposal have unrepaired datatype
    # issues?" cheaply, without a reasoner run and without mutating anything.
    dry_run: bool = False


class RepairDatatypesResponse(BaseModel):
    """Result of a datatype-repair run (or preview, when ``dry_run``)."""

    proposal_id: str
    repaired_count: int
    changes: list[dict[str, str]]
    # Echoes the request mode so a client can distinguish "would repair N" (preview)
    # from "repaired N" (applied) off a single shape.
    dry_run: bool = False


@router.post("/{proposal_id}/repair-datatypes", response_model=RepairDatatypesResponse)
def repair_proposal_datatypes(
    proposal_id: str,
    body: RepairDatatypesRequest | None = None,
    namespace: str = "default",
) -> RepairDatatypesResponse:
    """Repair (or, with ``dry_run``, preview) malformed XSD datatype tokens.

    User-initiated, consent-gated counterpart to the DatatypeTokenValidator
    finding: the validator surfaces the offending tokens at validation time and
    the user invokes this to apply the fix (replacing the previous silent
    write-boundary canonicalization). Rewrites the stored ontology Turtle AND the
    R2RML mappings, since a malformed ``rr:datatype`` is copied into the served
    ontology range at accept. Server-authoritative: the canonical XSD map lives
    here, not in the client. Idempotent — a clean proposal returns
    ``repaired_count: 0`` and persists nothing.

    ``dry_run=true`` computes the same change list but persists NOTHING (and does
    not flip status) — the accept-time guard uses it to detect unrepaired tokens
    before an accept without mutating the proposal.
    """
    from coa_ontology.datatype_canonicalizer import canonicalize_turtle

    dry_run = bool(body and body.dry_run)

    item = dynamo_store.get_proposal_by_id(proposal_id, namespace=namespace)
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")
    status = item.get("status")
    # Only the mutating path is state-gated; a dry-run preview is read-only and
    # safe on any status.
    if not dry_run and status in (ProposalStatus.ACCEPTED.value, ProposalStatus.REJECTED.value):
        raise HTTPException(409, f"Cannot repair a {status} proposal")

    changes: list[tuple[str, str]] = []
    updates: dict = {}

    def _canonicalize_or_422(ttl: str, what: str) -> tuple[str, list[tuple[str, str]]]:
        # canonicalize_turtle is a pure util that lets an rdflib ParserError
        # propagate. Stored Turtle is normally validated on write (update_proposal
        # -> _validate_turtle 422s on parse failure), so this only fires on
        # out-of-band corruption (manual DDB/S3 edit, truncated object). Convert it
        # to an actionable 422 — matching _validate_turtle — instead of a 500 +
        # traceback, and so the accept-time guard's dry_run call doesn't surface a 500.
        try:
            return canonicalize_turtle(ttl)
        except Exception as e:
            raise HTTPException(422, f"Stored {what} Turtle is not parseable: {e}") from e

    ontology_ttl = item.get("ontology_turtle") or ""
    if ontology_ttl:
        canonical, onto_changes = _canonicalize_or_422(ontology_ttl, "ontology")
        if onto_changes:
            if not dry_run:
                updates["ontology_s3_key"] = dynamo_store._put_artifact_s3(
                    namespace, proposal_id, "ontology", canonical
                )
            changes.extend(onto_changes)

    r2rml_ttl = item.get("r2rml_turtle") or ""
    if r2rml_ttl:
        canonical_r2rml, r2rml_changes = _canonicalize_or_422(r2rml_ttl, "R2RML")
        if r2rml_changes:
            if not dry_run:
                updates["r2rml_s3_key"] = dynamo_store._put_artifact_s3(
                    namespace, proposal_id, "r2rml", canonical_r2rml
                )
            changes.extend(r2rml_changes)

    if updates:  # only populated on the non-dry-run path
        updates["status"] = "updated"
        dynamo_store.update_proposal(proposal_id, namespace=namespace, **updates)
        _log.info("Repaired %d datatype token(s) on proposal %s: %s", len(changes), proposal_id, changes)

    return RepairDatatypesResponse(
        proposal_id=proposal_id,
        repaired_count=len(changes),
        changes=[{"found": before, "canonical": after} for before, after in changes],
        dry_run=dry_run,
    )


def _apply_grounding_overrides(proposal_id: str, overrides: dict[str, str | None], item: dict, namespace: str) -> dict:
    """Patch grounding matches and rebuild Turtle + R2RML."""
    import os

    from coa_ontology.induce_catalog import _catalog_to_tables, _fetch_catalog_from_smus
    from coa_ontology.inducer.schemas import ConceptMatch
    from coa_ontology.inducer.services.data_catalog import CatalogTable
    from coa_ontology.inducer.services.grounding import classify_score_tier
    from coa_ontology.inducer.strategies.table_to_ontology import TableToOntologyStrategy

    meta = item.get("metadata") or {}
    raw_matches = meta.get("matches") or []
    if not raw_matches:
        raise HTTPException(409, "Proposal has no grounding matches to update")

    matches = [ConceptMatch(**m) if isinstance(m, dict) else m for m in raw_matches]

    for m in matches:
        if m.source_column:
            continue
        override_uri = overrides.get(m.source_table)
        if override_uri is None and m.source_table in overrides:
            m.matched_class_uri = None
            m.matched_ontology_id = None
            m.match_type = "novel"
            m.similarity = None
        elif override_uri is not None:
            m.matched_class_uri = override_uri
            # Derive the tier from the chosen candidate's own score rather than
            # hardcoding "high_confidence". Hardcoding was lossy and wrong: an
            # original *exact* match (e.g. publication → Publication @ 0.95)
            # demoted to high_confidence on the first override and never
            # recovered — even when the user reverted to that same candidate.
            # Each stored candidate keeps its score, so re-classifying with the
            # grounding service's own thresholds makes overrides round-trip:
            # reverting to the original grounding restores its original tier.
            cand = None
            if m.candidates:
                cand = next((c for c in m.candidates if c.entity_uri == override_uri), None)
            if cand:
                m.matched_ontology_id = cand.ontology_id
                score = cand.fused_score if cand.fused_score is not None else cand.lexical_sim
                m.similarity = score
                # Candidate scores are rerank/fused scores → classify on the
                # rerank ladder (≥0.85 exact, ≥0.65 high_confidence, …). That
                # ladder is margin-independent, so the public score-only tier
                # helper is the right entry point here.
                m.match_type = classify_score_tier(score, has_rerank=True)
            else:
                # Override URI isn't among the stored candidates (e.g. a class
                # the user picked manually that wasn't a recall candidate); we
                # have no score to classify from, so keep the prior conservative
                # default rather than asserting an unjustified tier.
                m.match_type = "high_confidence"

    uri_prefix = item.get("ontology_id") or meta.get("ontology_uri_prefix", "http://example.org/ind#")

    # ── Unstructured branch ──────────────────────────────────────────────
    # Unstructured proposals have no DB catalog tables to rebuild from — their
    # ontology is the induced lexical-graph Turtle. Re-apply the (already
    # patched) grounding matches by surgically rewriting the grounding axioms
    # on the stored Turtle, rather than running the structured table rebuild
    # (which would 409 on 'no catalog tables'). No R2RML for unstructured.
    if item.get("source_type") == "UNSTRUCTURED":
        from coa_ontology.inducer.unstructured.services.class_grounding import (
            apply_overrides_to_turtle,
        )

        stored_turtle = dynamo_store._get_artifact_s3(namespace, proposal_id, "ontology") or ""
        if not stored_turtle:
            raise HTTPException(409, "Proposal has no stored ontology Turtle to patch")

        new_turtle, grounded_count = apply_overrides_to_turtle(stored_turtle, matches, uri_prefix=uri_prefix)

        from coa_ontology.induce_catalog import _floats_to_decimal

        updated_matches = [m.model_dump() for m in matches if not m.source_column]
        matches_s3_key = dynamo_store.put_proposal_matches_s3(proposal_id, updated_matches, namespace=namespace)
        meta_stripped = {k: v for k, v in meta.items() if k not in ("matches", "constraint_config")}
        meta_stripped["foundational_grounded_count"] = grounded_count
        onto_key = dynamo_store._put_artifact_s3(namespace, proposal_id, "ontology", new_turtle)
        dynamo_store.update_proposal(
            proposal_id,
            namespace=namespace,
            ontology_s3_key=onto_key,
            matches_s3_key=matches_s3_key,
            status=ProposalStatus.UPDATED.value,
            metadata=_floats_to_decimal(meta_stripped),
        )
        return {"proposal_id": proposal_id, "status": "updated"}

    # ── Structured branch (rebuild from DB catalog tables) ───────────────
    ds_ids = meta.get("datasource_ids") or []
    catalog_source = os.environ.get("CATALOG_SOURCE", "http")
    config = {
        "catalog_source": catalog_source,
        "data_catalog_url": os.environ.get("DATA_CATALOG_URL", ""),
        "smus_domain_id": os.environ.get("SMUS_DOMAIN_ID", ""),
        "namespaces_table": os.environ.get("NAMESPACES_TABLE", ""),
        "smus_datasources_table": os.environ.get("DATASOURCES_TABLE", ""),
        "sources_table": os.environ.get("SOURCES_TABLE", ""),
        "smus_region": os.environ.get("AWS_REGION", "us-east-1"),
    }

    all_tables: list[CatalogTable] = []
    for ds_id in ds_ids:
        try:
            if catalog_source == "smus":
                catalog = _fetch_catalog_from_smus(config, ds_id, namespace=namespace)
            else:
                from coa_ontology.induce_catalog import _fetch_catalog

                catalog = _fetch_catalog(config["data_catalog_url"], ds_id)
            for tbl_dict in _catalog_to_tables(catalog):
                tbl_dict["datasourceId"] = ds_id
                fqn = tbl_dict.get("fullyQualifiedName", "")
                tbl_dict["sourceSchema"] = fqn.split(".")[0] if "." in fqn else None
                all_tables.append(CatalogTable(**tbl_dict))
        except Exception as e:
            _log.warning("Failed to fetch catalog for %s: %s", ds_id, e)

    if not all_tables:
        raise HTTPException(409, "Could not fetch catalog tables for this proposal's datasources")

    strategy = TableToOntologyStrategy()
    proposal_graph, novel_tables = strategy._build_proposal_ontology(uri_prefix, all_tables, matches)
    r2rml_graph = strategy.build_r2rml(
        ontology_uri_prefix=uri_prefix,
        tables=all_tables,
        novel_tables=novel_tables,
        proposal_graph=proposal_graph,
    )

    ontology_ttl = proposal_graph.serialize(format="turtle")
    r2rml_ttl = r2rml_graph.serialize(format="turtle")

    from coa_ontology.induce_catalog import _floats_to_decimal

    # Matches are serialized to S3 as JSON, so they keep native floats — NOT
    # ``_floats_to_decimal`` (which is only for inline DynamoDB attributes;
    # ``Decimal`` is not JSON-serializable and would break ``json.dumps``).
    updated_matches = [m.model_dump() for m in matches if not m.source_column]
    # Offload matches to S3 (they exceed the DynamoDB 400 KB item limit once
    # candidates + definitions are included) and persist only the S3 key on the
    # item — mirroring create_proposal. Writing them inline into ``metadata``
    # fails with "Item size has exceeded the maximum allowed size for :metadata".
    matches_s3_key = dynamo_store.put_proposal_matches_s3(proposal_id, updated_matches, namespace=namespace)
    # Strip blobs that live in S3 from the inline metadata before the write:
    #   - ``matches``: re-persisted to S3 above (matches_s3_key).
    #   - ``constraint_config``: get_proposal_by_id hydrates it INTO metadata
    #     from S3 (flagged by ``has_constraint_config``); writing it back inline
    #     would re-bloat the item. The S3 copy is untouched, so dropping the
    #     inline copy keeps the flag-driven hydration working.
    meta_stripped = {k: v for k, v in meta.items() if k not in ("matches", "constraint_config")}
    # Turtle is offloaded to S3 too; pass the keys, not the bodies.
    onto_key = dynamo_store._put_artifact_s3(namespace, proposal_id, "ontology", ontology_ttl)
    r2rml_key = dynamo_store._put_artifact_s3(namespace, proposal_id, "r2rml", r2rml_ttl)
    dynamo_store.update_proposal(
        proposal_id,
        namespace=namespace,
        ontology_s3_key=onto_key,
        r2rml_s3_key=r2rml_key,
        matches_s3_key=matches_s3_key,
        status=ProposalStatus.UPDATED.value,
        metadata=_floats_to_decimal(meta_stripped),
    )

    return {"proposal_id": proposal_id, "status": "updated"}


# ── Refresh Grounding ──────────────────────────────────────────────────


# ── Validate ───────────────────────────────────────────────────────────


class ValidateProposalRequest(BaseModel):
    """Request body for validating a proposal across the selected tiers."""

    tiers: list[str] = ["tier1_blocking", "tier2_scoring", "tier3_review"]
    shacl_shapes_turtle: str | None = None
    competency_questions: list[str] | None = None


_MAX_VALIDATION_JOBS = 200
_validation_jobs: dict[str, dict] = {}

# Statuses past which a job never changes (mirrors dynamo_store).
_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled"})


def _persist_async_job(kind: str, job: dict, namespace: str) -> None:
    """Best-effort write of a job's current state to DynamoDB.

    The in-memory dict is a fast-path cache; DynamoDB is the durable copy that
    survives ECS task restarts and cross-instance polls. Uses an unconditional
    put (upsert) — ``job`` always holds the full current state, so we don't need
    a read-before-write to choose insert-vs-update (that was an extra round-trip
    on every status transition). Persisted under the proposal's ``namespace`` so
    the async-job partition stays tenant-isolated, matching the proposal row.
    Persistence failures are logged and swallowed — the poll can still read the
    in-memory copy on the same task.
    """
    try:
        dynamo_store.put_async_job(kind, job["job_id"], job, namespace=namespace)
    except Exception:  # noqa: BLE001 — durability is best-effort
        _log.warning("Failed to persist %s job %s to DynamoDB", kind, job.get("job_id"), exc_info=True)


def _safe_get_async_job(kind: str, job_id: str, namespace: str) -> dict | None:
    """Read the durable job copy, tolerating an unreachable DynamoDB.

    The poll's in-memory miss falls back here; if DynamoDB is unavailable we
    return None (→ caller raises a clean 404) rather than surfacing a 500. This
    keeps the not-found path working even when the durable store can't be read.
    """
    try:
        return dynamo_store.get_async_job(kind, job_id, namespace=namespace)
    except Exception:  # noqa: BLE001 — fall through to 404 on any store error
        _log.warning("Failed to read %s job %s from DynamoDB", kind, job_id, exc_info=True)
        return None


def _evict_terminal_jobs(jobs: dict[str, dict], cap: int) -> None:
    """Bound the in-memory job dict, preferring to evict TERMINAL jobs.

    The in-memory dict is only a fast-path cache — every job is also persisted
    to DynamoDB, and the poll falls back to that durable copy on a cache miss.
    So we evict oldest TERMINAL jobs first (their poll reads the finished DDB
    row); if that alone can't get under the cap because everything is still
    running, we force-evict the oldest entries anyway at a hard ceiling. The
    running job's poll still resolves via the DynamoDB fallback, so eviction
    never loses a result — it just falls back to the durable store. Without the
    hard ceiling the dict could grow unbounded when all jobs are non-terminal.
    Insertion order = age (dict preserves it).
    """
    if len(jobs) < cap:
        return
    # First pass: drop oldest terminal jobs.
    for jid in list(jobs.keys()):
        if len(jobs) < cap:
            return
        if str(jobs[jid].get("status")) in _TERMINAL_STATUSES:
            del jobs[jid]
    # Hard ceiling: if still over cap (all non-terminal), force-evict oldest.
    # Their status lives in DynamoDB, so the poll still resolves via fallback.
    for jid in list(jobs.keys()):
        if len(jobs) < cap:
            break
        del jobs[jid]


def _run_proposal_validation(
    job_id: str,
    turtle: str,
    body: ValidateProposalRequest,
    proposal_id: str,
    namespace: str,
    r2rml_turtle: str | None = None,
):
    from coa_ontology.validation.schemas import (
        ValidationTier,
    )
    from coa_ontology.validation.validators import OntologyValidator
    from coa_ontology.validation.validators.tier1 import (
        ConnectivityValidator,
        ConsistencyValidator,
        DatatypeTokenValidator,
        SHACLValidator,
        TaxonomyCycleValidator,
    )
    from coa_ontology.validation.validators.tier2 import StructuralMetricsValidator
    from coa_ontology.validation.validators.tier3 import (
        AmbiguousMatchValidator,
        CompetencyQuestionValidator,
        LabelCompletenessValidator,
        OoPSValidator,
    )

    validators_by_tier: dict[ValidationTier, list[OntologyValidator]] = {
        ValidationTier.tier1_blocking: [
            ConsistencyValidator(),
            TaxonomyCycleValidator(),
            ConnectivityValidator(),
            SHACLValidator(),
            DatatypeTokenValidator(),
        ],
        ValidationTier.tier2_scoring: [
            StructuralMetricsValidator(),
        ],
        ValidationTier.tier3_review: [
            OoPSValidator(),
            LabelCompletenessValidator(),
            AmbiguousMatchValidator(),
            CompetencyQuestionValidator(),
        ],
    }

    job = _validation_jobs.get(job_id)
    if not job:
        return
    job["status"] = "running"
    _persist_async_job("VALIDATE", job, namespace)
    try:
        g = Graph()
        g.parse(data=turtle, format="turtle")

        all_findings = []
        metrics_out: dict = {}
        tiers = [ValidationTier(t) for t in body.tiers]

        for tier in tiers:
            for v in validators_by_tier.get(tier) or []:
                all_findings.extend(
                    v.validate(
                        g,
                        shacl_shapes_turtle=body.shacl_shapes_turtle,
                        competency_questions=body.competency_questions or [],
                        r2rml_turtle=r2rml_turtle,
                        _metrics_out=metrics_out,
                    )
                )

        passed = not any(f.severity == "error" for f in all_findings)

        job["status"] = "completed"
        job["report"] = {
            "job_id": job_id,
            "proposal_id": proposal_id,
            "passed": passed,
            "findings": [f.model_dump() for f in all_findings],
            "metrics": metrics_out if metrics_out else None,
            "tiers_run": body.tiers,
            "created_at": job["created_at"],
        }
        _persist_async_job("VALIDATE", job, namespace)
    except Exception as e:
        _log.error("Proposal validation failed for job %s: %s", job_id, e, exc_info=True)
        job["status"] = "failed"
        job["error"] = f"Validation failed: {type(e).__name__}"
        _persist_async_job("VALIDATE", job, namespace)


@router.post("/{proposal_id}/validate")
def validate_proposal(
    proposal_id: str,
    body: ValidateProposalRequest | None = None,
    namespace: str = "default",
):
    """Run multi-tier validation on a proposal's Turtle without accepting it.

    Returns a validation job that can be polled for results. This lets
    the UI show reasoner findings, structural metrics, and OoPS! pitfalls
    before the user decides to accept or reject.
    """
    item = dynamo_store.get_proposal_by_id(proposal_id, namespace=namespace)
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")

    turtle = item.get("ontology_turtle", "")
    if not turtle:
        raise HTTPException(409, "Proposal has no ontology_turtle to validate")

    # R2RML too (inline or from S3) so the datatype validator can flag malformed
    # rr:datatype tokens — they reach Ontop via the accept-time range-alignment
    # step even when the ontology turtle itself is clean.
    r2rml_turtle = item.get("r2rml_turtle") or ""
    if not r2rml_turtle and item.get("r2rml_s3_key"):
        # Best-effort: validation still runs against the ontology turtle if this
        # fails. But log it — a silent failure here means the datatype validator
        # skips rr:datatype tokens (which reach Ontop via accept-time range
        # alignment), so a masked infra issue would silently defeat the feature.
        try:
            r2rml_turtle = dynamo_store._get_artifact_s3(namespace, proposal_id, "r2rml") or ""
        except Exception as e:
            _log.warning(
                "validate[%s]: failed to fetch R2RML from S3; datatype validator will skip rr:datatype (%s: %s)",
                proposal_id,
                type(e).__name__,
                e,
            )

    if body is None:
        body = ValidateProposalRequest()

    job_id = str(uuid.uuid4())

    # Evict oldest TERMINAL jobs only — never a running/pending one.
    _evict_terminal_jobs(_validation_jobs, _MAX_VALIDATION_JOBS)

    _validation_jobs[job_id] = {
        "job_id": job_id,
        "proposal_id": proposal_id,
        "status": "pending",
        "created_at": datetime.now(UTC).isoformat(),
        "report": None,
        "error": None,
    }
    _persist_async_job("VALIDATE", _validation_jobs[job_id], namespace)

    Thread(
        target=_run_proposal_validation,
        args=(job_id, turtle, body, proposal_id, namespace),
        kwargs={"r2rml_turtle": r2rml_turtle or None},
        daemon=True,
    ).start()
    return _validation_jobs[job_id]


@router.get("/{proposal_id}/validate/jobs/{job_id}")
def get_proposal_validation_job(proposal_id: str, job_id: str, namespace: str = "default"):
    """Poll a proposal validation job for results.

    Reads the in-memory copy first (fast path, same task) and falls back to the
    DynamoDB copy so the poll still resolves after a task restart / on a
    different instance (a stale non-terminal row is swept to ``failed``).
    """
    job = _validation_jobs.get(job_id)
    if not job:
        job = _safe_get_async_job("VALIDATE", job_id, namespace)
    if not job or job.get("proposal_id") != proposal_id:
        raise HTTPException(404, "Validation job not found")
    return job


# ── Constraint config endpoints ────────────────────────────────────────


_MAX_INFER_JOBS = 200
_infer_jobs: dict[str, dict] = {}


def _run_infer_constraints(
    job_id: str, turtle: str, existing_config_dict: dict | None, proposal_id: str, namespace: str
):
    """Background worker for LLM constraint inference.

    Runs the Bedrock ``infer_constraints_from_ontology`` call (up to 4096
    output tokens) off the request thread so the API-Gateway→ECS proxy
    round trip returns 202 immediately, well inside the 29s integration
    timeout. Status is kept in the in-process ``_infer_jobs`` dict — safe
    because the ontology-engine API is a single long-running Fargate task,
    so the job survives across the POST/poll round trip (same pattern as
    ``_run_proposal_validation``).
    """
    from coa_ontology.validation.shapes.config import ConstraintConfig
    from coa_ontology.validation.shapes.nl_generator import infer_constraints_from_ontology

    job = _infer_jobs.get(job_id)
    if not job:
        return
    job["status"] = "running"
    _persist_async_job("INFER", job, namespace)
    try:
        existing_config = ConstraintConfig(**existing_config_dict) if existing_config_dict else None
        inferred = infer_constraints_from_ontology(turtle, existing_config=existing_config)
        job["status"] = "completed"
        job["result"] = {"inferred_constraints": inferred.model_dump()}
    except RuntimeError as e:
        # The LLM/parse failure path raises RuntimeError with a scrubbed
        # message (``Constraint inference failed: …``); surface it verbatim
        # so the UI can show what went wrong, matching the synchronous
        # 500-with-str(e) behaviour it replaced.
        _log.error("Constraint inference failed for job %s: %s", job_id, e, exc_info=True)
        job["status"] = "failed"
        job["error"] = str(e)
    except Exception as e:
        _log.error("Constraint inference failed for job %s: %s", job_id, e, exc_info=True)
        job["status"] = "failed"
        job["error"] = f"Constraint inference failed: {type(e).__name__}"
    _persist_async_job("INFER", job, namespace)


@router.post("/{proposal_id}/infer-constraints", status_code=202)
def infer_constraints(proposal_id: str, namespace: str = "default"):
    """LLM infers semantic constraints from the proposal's ontology (async).

    Returns 202 + a job handle immediately and runs the Bedrock inference
    on a background thread; poll ``GET /{id}/infer-constraints/jobs/{job}``
    for the resulting ConstraintConfig. Async because the LLM call (up to
    4096 output tokens) can exceed API Gateway's 29s integration timeout.
    The eventual result carries ``inferred_constraints`` — a ConstraintConfig
    with llm_inferred entries the user reviews (enable/disable) before
    compiling to SHACL.
    """
    from coa_ontology.validation.shapes.config import ConstraintConfig

    item = dynamo_store.get_proposal_by_id(proposal_id, namespace=namespace)
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")

    turtle = item.get("ontology_turtle", "")
    if not turtle:
        raise HTTPException(409, "Proposal has no ontology_turtle for context")

    # Validate/normalise the existing config on the request thread so a
    # malformed stored config surfaces as a 422 here rather than a silent
    # job failure the client has to poll to discover.
    existing_config_dict: dict | None = None
    meta = item.get("metadata") or {}
    if meta.get("constraint_config"):
        # Round-trip through the model to normalise + reject bad shapes early.
        existing_config_dict = ConstraintConfig(**meta["constraint_config"]).model_dump()

    job_id = str(uuid.uuid4())

    # Evict oldest TERMINAL jobs only — never a running/pending one.
    _evict_terminal_jobs(_infer_jobs, _MAX_INFER_JOBS)

    _infer_jobs[job_id] = {
        "job_id": job_id,
        "proposal_id": proposal_id,
        "status": "pending",
        "created_at": datetime.now(UTC).isoformat(),
        "result": None,
        "error": None,
    }
    _persist_async_job("INFER", _infer_jobs[job_id], namespace)

    Thread(
        target=_run_infer_constraints,
        args=(job_id, turtle, existing_config_dict, proposal_id, namespace),
        daemon=True,
    ).start()
    return _infer_jobs[job_id]


@router.get("/{proposal_id}/infer-constraints/jobs/{job_id}")
def get_infer_constraints_job(proposal_id: str, job_id: str, namespace: str = "default"):
    """Poll a constraint-inference job for its result.

    In-memory fast path, with a DynamoDB fallback so the poll resolves after a
    task restart / on a different instance (stale rows are swept to ``failed``).
    """
    job = _infer_jobs.get(job_id)
    if not job:
        job = _safe_get_async_job("INFER", job_id, namespace)
    if not job or job.get("proposal_id") != proposal_id:
        raise HTTPException(404, "Infer-constraints job not found")
    return job


class CompileConstraintsRequest(BaseModel):
    """Request body carrying a reviewed constraint config to compile to SHACL."""

    constraint_config: dict
    custom_turtle: str | None = None


@router.post("/{proposal_id}/compile-constraints")
def compile_constraints(proposal_id: str, body: CompileConstraintsRequest, namespace: str = "default"):
    """Compile a reviewed ConstraintConfig into SHACL Turtle.

    The frontend sends back the config after user review (toggles, edits)
    and receives compiled SHACL ready for validation.
    """
    from coa_ontology.validation.shapes.config import ConstraintConfig, compile_to_shacl

    item = dynamo_store.get_proposal_by_id(proposal_id, namespace=namespace)
    if not item:
        raise HTTPException(404, f"Proposal '{proposal_id}' not found")

    uri_prefix = item.get("ontology_id") or (item.get("metadata") or {}).get(
        "ontology_uri_prefix", "http://example.org/ind#"
    )

    config = ConstraintConfig(**body.constraint_config)
    shapes_turtle = compile_to_shacl(config, uri_prefix, custom_turtle=body.custom_turtle)

    return {"proposal_id": proposal_id, "shapes_turtle": shapes_turtle}