vignette/apps/api/app/services/continuous_improvement_store.py

"""Append-only persistence and automation boundary for G8 Continuous Improvement OS."""

from __future__ import annotations

import hashlib
import json
from collections.abc import Mapping, Sequence
from typing import Any, Literal, Protocol
from uuid import UUID, uuid5

import asyncpg
from pydantic import BaseModel, ConfigDict, Field, field_validator

from ..contracts.continuous_improvement import (
    AgenticReleaseManifest,
    ContentBenchmarkQualification,
    ContentSourceArtifact,
    GeneratedContentDraft,
    IndependentRedTeamReview,
    ModelCalibrationSnapshot,
    OperationalIncident,
)
from .continuous_improvement import (
    build_incident_regression_dag,
    decide_model_change,
    promote_content_to_catalog,
    release_allowed,
)


DATA_CLASSIFICATION = "synthetic_replay_red_team_coverage_drift"
_UUID_NAMESPACE = UUID("ea8df01e-46f7-4c59-9505-4b267943ac52")


class ContinuousImprovementError(ValueError):
    """Base persistence boundary error."""


class ContinuousImprovementConflictError(ContinuousImprovementError):
    """Stable submission identifier was reused with changed content."""


class ContinuousImprovementNotFoundError(ContinuousImprovementError):
    """A requested gate or qualification is absent or invisible."""


class RollbackExecutionRequest(BaseModel):
    """Pinned, idempotent command sent to the model/runtime control plane."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    schema_version: Literal["oas.rollback-executor.v1"] = "oas.rollback-executor.v1"
    idempotency_key: UUID
    approval_event_id: UUID
    rollback_scope: Literal["model", "runtime"]
    target_kind: Literal["model_change_gate", "release_gate"]
    target_id: UUID
    subject_id: str = Field(min_length=1, max_length=180)
    rollback_target_id: str = Field(min_length=1, max_length=180)
    artifact_record_id: UUID
    artifact_id: str = Field(min_length=1, max_length=240)
    artifact_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
    artifact_provenance_uri: str = Field(min_length=1, max_length=500)
    authorization_evidence_refs: tuple[str, ...] = Field(min_length=1, max_length=100)


class RollbackExecutionReceipt(BaseModel):
    """Strict success receipt. Any missing or mismatched binding means failure."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    schema_version: Literal["oas.rollback-executor.v1"] = "oas.rollback-executor.v1"
    status: Literal["executed"] = "executed"
    execution_id: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
    idempotency_key: UUID
    rollback_scope: Literal["model", "runtime"]
    target_kind: Literal["model_change_gate", "release_gate"]
    target_id: UUID
    artifact_record_id: UUID
    artifact_sha256: str = Field(pattern=r"^[a-f0-9]{64}$")
    evidence_refs: tuple[str, ...] = Field(min_length=1, max_length=100)

    @field_validator("evidence_refs")
    @classmethod
    def evidence_refs_are_unique_and_absolute(
        cls, value: tuple[str, ...]
    ) -> tuple[str, ...]:
        if len(value) != len(set(value)):
            raise ValueError("rollback execution evidence refs must be unique")
        if any(
            not item.startswith(("https://", "audit://", "db://", "repo://"))
            for item in value
        ):
            raise ValueError("rollback execution evidence refs must be durable secure URIs")
        return value


class RollbackExecutor(Protocol):
    async def execute(
        self, request: RollbackExecutionRequest
    ) -> RollbackExecutionReceipt: ...


def _value(row: Mapping[str, Any], key: str, default: Any = None) -> Any:
    try:
        return row[key]
    except (KeyError, TypeError):
        return default


def _canonical_hash(payload: Any) -> str:
    serialized = json.dumps(
        payload,
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
        default=str,
    )
    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()


async def _begin_submission(
    conn: asyncpg.Connection,
    *,
    submission_id: UUID,
    operation_kind: str,
    result_id: UUID,
    payload: Any,
) -> bool:
    content_hash = _canonical_hash(payload)
    await conn.execute(
        "SELECT pg_advisory_xact_lock(hashtextextended($1::text, 81))",
        str(submission_id),
    )
    row = await conn.fetchrow(
        """
        SELECT content_hash, operation_kind, result_id
        FROM app.ci_ingestion_submission WHERE submission_id = $1
        """,
        submission_id,
    )
    if row is not None:
        if (
            str(_value(row, "content_hash")) != content_hash
            or str(_value(row, "operation_kind")) != operation_kind
            or UUID(str(_value(row, "result_id"))) != result_id
        ):
            raise ContinuousImprovementConflictError(
                "submission id was already used with different content"
            )
        return True
    await conn.execute(
        """
        INSERT INTO app.ci_ingestion_submission (
          submission_id, content_hash, operation_kind, result_id, data_classification
        ) VALUES ($1,$2,$3,$4,$5)
        """,
        submission_id,
        content_hash,
        operation_kind,
        result_id,
        DATA_CLASSIFICATION,
    )
    return False


async def _ensure_source(
    conn: asyncpg.Connection, source: ContentSourceArtifact
) -> UUID:
    source_record_id = uuid5(_UUID_NAMESPACE, f"{source.source_id}|{source.version}")
    content_hash = _canonical_hash(source.model_dump(mode="json"))
    await conn.execute(
        """
        INSERT INTO app.ci_source_artifact (
          source_record_id, source_id, source_version, content_sha256,
          provenance_uri, usage_status, citation_label, content_hash
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
        ON CONFLICT (source_record_id) DO NOTHING
        """,
        source_record_id,
        source.source_id,
        source.version,
        source.content_sha256,
        source.provenance_uri,
        source.usage_status,
        source.citation_label,
        content_hash,
    )
    row = await conn.fetchrow(
        "SELECT content_hash FROM app.ci_source_artifact WHERE source_record_id = $1",
        source_record_id,
    )
    if row is None:
        raise ContinuousImprovementNotFoundError("source artifact is not visible")
    if str(_value(row, "content_hash")) != content_hash:
        raise ContinuousImprovementConflictError(
            "source id/version resolved to changed provenance"
        )
    return source_record_id


async def find_content_pipeline_submission(
    conn: asyncpg.Connection, *, submission_id: UUID
) -> dict[str, Any] | None:
    """Return an existing agentic pipeline replay without invoking models again."""

    row = await conn.fetchrow(
        """
        SELECT s.operation_kind, s.result_id,
               p.pipeline_id, p.draft_id, p.prompt_sha256,
               q.qualification_id, q.catalog_entry_id AS candidate_catalog_entry_id,
               b.benchmark_record_id, b.benchmark_id,
               b.variant_count AS benchmark_variant_count,
               (SELECT count(*) FROM app.ci_red_team_review r
                WHERE r.pipeline_id = p.pipeline_id) AS red_team_review_count
        FROM app.ci_ingestion_submission s
        LEFT JOIN app.ci_content_pipeline p ON p.submission_id = s.submission_id
        LEFT JOIN app.ci_content_qualification q ON q.pipeline_id = p.pipeline_id
        LEFT JOIN app.ci_content_benchmark b ON b.pipeline_id = p.pipeline_id
        WHERE s.submission_id = $1
        """,
        submission_id,
    )
    if row is None:
        return None
    if str(_value(row, "operation_kind")) != "content_pipeline":
        raise ContinuousImprovementConflictError(
            "submission id belongs to another continuous-improvement operation"
        )
    required = (
        "pipeline_id",
        "draft_id",
        "prompt_sha256",
        "qualification_id",
        "candidate_catalog_entry_id",
        "benchmark_record_id",
        "benchmark_id",
        "benchmark_variant_count",
        "red_team_review_count",
    )
    if any(_value(row, key) is None for key in required):
        raise ContinuousImprovementNotFoundError(
            "content pipeline submission is incomplete"
        )
    return {key: _value(row, key) for key in required}


async def read_operational_incident(
    conn: asyncpg.Connection, *, incident_record_id: UUID
) -> OperationalIncident:
    row = await conn.fetchrow(
        """
        SELECT incident_id, error_fingerprint, affected_contract,
               evidence_refs, pii_included
        FROM app.ci_operational_incident
        WHERE incident_record_id = $1
        """,
        incident_record_id,
    )
    if row is None:
        raise ContinuousImprovementNotFoundError("operational incident not found")
    return OperationalIncident(
        incident_id=str(_value(row, "incident_id")),
        error_fingerprint=str(_value(row, "error_fingerprint")),
        affected_contract=str(_value(row, "affected_contract")),
        evidence_refs=tuple(_value(row, "evidence_refs") or ()),
        pii_included=bool(_value(row, "pii_included")),
    )


async def submit_content_pipeline(
    conn: asyncpg.Connection,
    *,
    submission_id: UUID,
    pipeline_id: UUID,
    benchmark_record_id: UUID,
    qualification_id: UUID,
    draft: GeneratedContentDraft,
    sources: Sequence[ContentSourceArtifact],
    reviews: Sequence[IndependentRedTeamReview],
    benchmark: ContentBenchmarkQualification,
    draft_payload: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    if len(draft.source_refs) != len(set(draft.source_refs)):
        raise ContinuousImprovementError("content source refs must be unique")
    review_ids = [item.review_id for item in reviews]
    if len(review_ids) != len(set(review_ids)):
        raise ContinuousImprovementError("red-team review ids must be unique")
    candidate = promote_content_to_catalog(
        draft=draft,
        sources=sources,
        reviews=reviews,
        benchmark=benchmark,
    )
    payload = {
        "pipeline_id": str(pipeline_id),
        "benchmark_record_id": str(benchmark_record_id),
        "qualification_id": str(qualification_id),
        "draft": draft.model_dump(mode="json"),
        "sources": [item.model_dump(mode="json") for item in sources],
        "reviews": [item.model_dump(mode="json") for item in reviews],
        "benchmark": benchmark.model_dump(mode="json"),
        "data_classification": DATA_CLASSIFICATION,
    }
    if draft_payload is not None:
        payload["draft_payload"] = dict(draft_payload)
    if await _begin_submission(
        conn,
        submission_id=submission_id,
        operation_kind="content_pipeline",
        result_id=qualification_id,
        payload=payload,
    ):
        return {
            "submission_id": submission_id,
            "pipeline_id": pipeline_id,
            "qualification_id": qualification_id,
            "candidate_catalog_entry_id": candidate.catalog_entry_id,
            "state": "pending_human_approval",
            "human_approval_required": True,
            "catalog_promoted": False,
            "idempotent_replay": True,
            "clinical_claim_allowed": False,
        }

    source_records = {
        source.source_id: await _ensure_source(conn, source) for source in sources
    }
    source_record_ids = [source_records[source_id] for source_id in draft.source_refs]
    await conn.execute(
        """
        INSERT INTO app.ci_content_pipeline (
          pipeline_id, submission_id, draft_id, content_kind, source_record_ids,
          generation_model, prompt_version, prompt_sha256, payload_sha256, draft_payload,
          synthetic_identity_id, difficulty_level, hidden_answer_fingerprint,
          visible_answer_overlap_tokens, pii_findings, unsupported_clinical_claims
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)
        """,
        pipeline_id,
        submission_id,
        draft.draft_id,
        draft.content_kind,
        source_record_ids,
        draft.generation_model,
        draft.prompt_version,
        draft.prompt_sha256,
        draft.payload_sha256,
        dict(draft_payload) if draft_payload is not None else None,
        draft.synthetic_identity_id,
        draft.difficulty_level,
        draft.hidden_answer_fingerprint,
        draft.visible_answer_overlap_tokens,
        draft.pii_findings,
        draft.unsupported_clinical_claims,
    )
    review_record_ids: list[UUID] = []
    for review in reviews:
        review_record_id = uuid5(pipeline_id, review.review_id)
        review_record_ids.append(review_record_id)
        await conn.execute(
            """
            INSERT INTO app.ci_red_team_review (
              review_record_id, pipeline_id, review_id, reviewer_agent_id,
              dimensions, reviewed_payload_sha256
            ) VALUES ($1,$2,$3,$4,$5,$6)
            """,
            review_record_id,
            pipeline_id,
            review.review_id,
            review.reviewer_agent_id,
            list(review.dimensions),
            review.reviewed_payload_sha256,
        )
        for finding in review.findings:
            await conn.execute(
                """
                INSERT INTO app.ci_red_team_finding (
                  finding_record_id, review_record_id, finding_id, dimension,
                  severity, finding_state, evidence_ref, remediation_ref
                ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
                """,
                uuid5(review_record_id, finding.finding_id),
                review_record_id,
                finding.finding_id,
                finding.dimension,
                finding.severity,
                finding.state,
                finding.evidence_ref,
                finding.remediation_ref,
            )
    await conn.execute(
        """
        INSERT INTO app.ci_content_benchmark (
          benchmark_record_id, pipeline_id, benchmark_id, variant_count,
          variant_pass_rate, answer_leakage_count, pii_finding_count,
          unsupported_claim_count, safety_failure_count, reward_hacking_count,
          evidence_refs, qualified
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
        """,
        benchmark_record_id,
        pipeline_id,
        benchmark.benchmark_id,
        benchmark.variant_count,
        benchmark.variant_pass_rate,
        benchmark.answer_leakage_count,
        benchmark.pii_finding_count,
        benchmark.unsupported_claim_count,
        benchmark.safety_failure_count,
        benchmark.reward_hacking_count,
        list(benchmark.evidence_refs),
        benchmark.qualified,
    )
    await conn.execute(
        """
        INSERT INTO app.ci_content_qualification (
          qualification_id, pipeline_id, benchmark_record_id, catalog_entry_id,
          payload_sha256, source_record_ids, review_record_ids
        ) VALUES ($1,$2,$3,$4,$5,$6,$7)
        """,
        qualification_id,
        pipeline_id,
        benchmark_record_id,
        candidate.catalog_entry_id,
        candidate.payload_sha256,
        source_record_ids,
        review_record_ids,
    )
    return {
        "submission_id": submission_id,
        "pipeline_id": pipeline_id,
        "qualification_id": qualification_id,
        "candidate_catalog_entry_id": candidate.catalog_entry_id,
        "state": "pending_human_approval",
        "human_approval_required": True,
        "catalog_promoted": False,
        "idempotent_replay": False,
        "clinical_claim_allowed": False,
    }


def _artifact_payload(artifact: Mapping[str, Any]) -> dict[str, Any]:
    return {
        "artifact_record_id": str(artifact["artifact_record_id"]),
        "artifact_id": str(artifact["artifact_id"]),
        "content_sha256": str(artifact["content_sha256"]),
        "provenance_uri": str(artifact["provenance_uri"]),
    }


async def _insert_gate_artifacts(
    conn: asyncpg.Connection,
    *,
    owner_kind: str,
    owner_id: UUID,
    baseline: Mapping[str, Any],
    threshold: Mapping[str, Any],
    provenance: Sequence[Mapping[str, Any]],
    rollback: Mapping[str, Any],
) -> tuple[UUID, UUID, list[UUID], UUID]:
    if not provenance:
        raise ContinuousImprovementError(
            "release gate requires at least one provenance artifact"
        )
    groups = (
        ("baseline", (baseline,)),
        ("threshold", (threshold,)),
        ("provenance", provenance),
        ("rollback", (rollback,)),
    )
    ids: dict[str, list[UUID]] = {}
    for artifact_kind, artifacts in groups:
        ids[artifact_kind] = []
        for artifact in artifacts:
            artifact_id = UUID(str(artifact["artifact_record_id"]))
            ids[artifact_kind].append(artifact_id)
            await conn.execute(
                """
                INSERT INTO app.ci_gate_artifact (
                  artifact_record_id, owner_kind, owner_id, artifact_kind,
                  artifact_id, content_sha256, provenance_uri
                ) VALUES ($1,$2,$3,$4,$5,$6,$7)
                """,
                artifact_id,
                owner_kind,
                owner_id,
                artifact_kind,
                str(artifact["artifact_id"]),
                str(artifact["content_sha256"]),
                str(artifact["provenance_uri"]),
            )
    return (
        ids["baseline"][0],
        ids["threshold"][0],
        ids["provenance"],
        ids["rollback"][0],
    )


async def submit_model_change_gate(
    conn: asyncpg.Connection,
    *,
    submission_id: UUID,
    gate_id: UUID,
    baseline_snapshot_record_id: UUID,
    candidate_snapshot_record_id: UUID,
    baseline: ModelCalibrationSnapshot,
    candidate: ModelCalibrationSnapshot,
    baseline_artifact: Mapping[str, Any],
    threshold_artifact: Mapping[str, Any],
    provenance_artifacts: Sequence[Mapping[str, Any]],
    rollback_artifact: Mapping[str, Any],
) -> dict[str, Any]:
    gate = decide_model_change(baseline=baseline, candidate=candidate)
    payload = {
        "gate_id": str(gate_id),
        "baseline_snapshot_record_id": str(baseline_snapshot_record_id),
        "candidate_snapshot_record_id": str(candidate_snapshot_record_id),
        "baseline": baseline.model_dump(mode="json"),
        "candidate": candidate.model_dump(mode="json"),
        "artifacts": {
            "baseline": _artifact_payload(baseline_artifact),
            "threshold": _artifact_payload(threshold_artifact),
            "provenance": [_artifact_payload(item) for item in provenance_artifacts],
            "rollback": _artifact_payload(rollback_artifact),
        },
        "data_classification": DATA_CLASSIFICATION,
    }
    if await _begin_submission(
        conn,
        submission_id=submission_id,
        operation_kind="model_change_gate",
        result_id=gate_id,
        payload=payload,
    ):
        return {
            "submission_id": submission_id,
            "gate_id": gate_id,
            "gate_decision": gate.decision,
            "state": "pending_human_approval",
            "human_approval_required": True,
            "promotion_executed": False,
            "idempotent_replay": True,
        }
    artifact_ids = await _insert_gate_artifacts(
        conn,
        owner_kind="model_change_gate",
        owner_id=gate_id,
        baseline=baseline_artifact,
        threshold=threshold_artifact,
        provenance=provenance_artifacts,
        rollback=rollback_artifact,
    )
    for record_id, role, snapshot in (
        (baseline_snapshot_record_id, "baseline", baseline),
        (candidate_snapshot_record_id, "candidate", candidate),
    ):
        await conn.execute(
            """
            INSERT INTO app.ci_model_calibration_snapshot (
              snapshot_record_id, gate_id, snapshot_role, snapshot_id, model_name,
              prompt_version, benchmark_version, task_accuracy, critical_miss_count,
              leakage_count, pii_count, calibration_error, subgroup_max_gap
            ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
            """,
            record_id,
            gate_id,
            role,
            snapshot.snapshot_id,
            snapshot.model,
            snapshot.prompt_version,
            snapshot.benchmark_version,
            snapshot.task_accuracy,
            snapshot.critical_miss_count,
            snapshot.leakage_count,
            snapshot.pii_count,
            snapshot.calibration_error,
            snapshot.subgroup_max_gap,
        )
    await conn.execute(
        """
        INSERT INTO app.ci_model_change_gate (
          gate_id, submission_id, baseline_snapshot_record_id,
          candidate_snapshot_record_id, baseline_artifact_id,
          threshold_artifact_id, provenance_artifact_ids, rollback_artifact_id,
          gate_decision, reasons, rollback_target_snapshot_id
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
        """,
        gate_id,
        submission_id,
        baseline_snapshot_record_id,
        candidate_snapshot_record_id,
        artifact_ids[0],
        artifact_ids[1],
        artifact_ids[2],
        artifact_ids[3],
        gate.decision,
        list(gate.reasons),
        gate.rollback_target_snapshot_id,
    )
    return {
        "submission_id": submission_id,
        "gate_id": gate_id,
        "gate_decision": gate.decision,
        "state": "pending_human_approval",
        "human_approval_required": True,
        "promotion_executed": False,
        "idempotent_replay": False,
    }


async def submit_release_gate(
    conn: asyncpg.Connection,
    *,
    submission_id: UUID,
    gate_id: UUID,
    manifest: AgenticReleaseManifest,
    baseline_artifact: Mapping[str, Any],
    threshold_artifact: Mapping[str, Any],
    provenance_artifacts: Sequence[Mapping[str, Any]],
    rollback_artifact: Mapping[str, Any],
) -> dict[str, Any]:
    qualified = release_allowed(manifest)
    payload = {
        "gate_id": str(gate_id),
        "manifest": manifest.model_dump(mode="json"),
        "artifacts": {
            "baseline": _artifact_payload(baseline_artifact),
            "threshold": _artifact_payload(threshold_artifact),
            "provenance": [_artifact_payload(item) for item in provenance_artifacts],
            "rollback": _artifact_payload(rollback_artifact),
        },
        "data_classification": DATA_CLASSIFICATION,
    }
    if await _begin_submission(
        conn,
        submission_id=submission_id,
        operation_kind="release_gate",
        result_id=gate_id,
        payload=payload,
    ):
        return {
            "submission_id": submission_id,
            "gate_id": gate_id,
            "qualified": qualified,
            "state": "pending_human_approval",
            "human_approval_required": True,
            "promotion_executed": False,
            "idempotent_replay": True,
        }
    artifact_ids = await _insert_gate_artifacts(
        conn,
        owner_kind="release_gate",
        owner_id=gate_id,
        baseline=baseline_artifact,
        threshold=threshold_artifact,
        provenance=provenance_artifacts,
        rollback=rollback_artifact,
    )
    await conn.execute(
        """
        INSERT INTO app.ci_release_gate (
          gate_id, submission_id, release_id, red_green_passed, contract_passed,
          e2e_passed, runtime_proof_passed, public_proof_passed, ssot_synced,
          evidence_refs, baseline_artifact_id, threshold_artifact_id,
          provenance_artifact_ids, rollback_artifact_id, qualified
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
        """,
        gate_id,
        submission_id,
        manifest.release_id,
        manifest.red_green_passed,
        manifest.contract_passed,
        manifest.e2e_passed,
        manifest.runtime_proof_passed,
        manifest.public_proof_passed,
        manifest.ssot_synced,
        list(manifest.evidence_refs),
        artifact_ids[0],
        artifact_ids[1],
        artifact_ids[2],
        artifact_ids[3],
        qualified,
    )
    return {
        "submission_id": submission_id,
        "gate_id": gate_id,
        "qualified": qualified,
        "state": "pending_human_approval",
        "human_approval_required": True,
        "promotion_executed": False,
        "idempotent_replay": False,
    }


async def submit_incident_dag(
    conn: asyncpg.Connection,
    *,
    submission_id: UUID,
    incident_record_id: UUID,
    incident: OperationalIncident,
) -> dict[str, Any]:
    dag = build_incident_regression_dag(incident)
    payload = {
        "incident_record_id": str(incident_record_id),
        "incident": incident.model_dump(mode="json"),
        "data_classification": DATA_CLASSIFICATION,
    }
    if await _begin_submission(
        conn,
        submission_id=submission_id,
        operation_kind="incident_dag",
        result_id=incident_record_id,
        payload=payload,
    ):
        return {
            "submission_id": submission_id,
            "incident_record_id": incident_record_id,
            "node_count": 4,
            "idempotent_replay": True,
            "pii_included": False,
        }
    await conn.execute(
        """
        INSERT INTO app.ci_operational_incident (
          incident_record_id, submission_id, incident_id, error_fingerprint,
          affected_contract, evidence_refs, pii_included
        ) VALUES ($1,$2,$3,$4,$5,$6,FALSE)
        """,
        incident_record_id,
        submission_id,
        incident.incident_id,
        incident.error_fingerprint,
        incident.affected_contract,
        list(incident.evidence_refs),
    )
    node_ids = {
        node.node_id: uuid5(incident_record_id, node.node_id) for node in dag.nodes
    }
    for node in dag.nodes:
        await conn.execute(
            """
            INSERT INTO app.ci_regression_dag_node (
              node_record_id, incident_record_id, node_id, node_type,
              depends_on_record_ids, evidence_ref, node_status
            ) VALUES ($1,$2,$3,$4,$5,$6,$7)
            """,
            node_ids[node.node_id],
            incident_record_id,
            node.node_id,
            node.node_type,
            [node_ids[parent] for parent in node.depends_on],
            node.evidence_ref,
            node.status,
        )
    return {
        "submission_id": submission_id,
        "incident_record_id": incident_record_id,
        "node_count": 4,
        "idempotent_replay": False,
        "pii_included": False,
    }


async def _build_rollback_execution_request(
    conn: asyncpg.Connection,
    *,
    effect_record_id: UUID,
    approval_event_id: UUID,
    target_kind: str,
    target_id: UUID,
    evidence_refs: Sequence[str],
) -> tuple[RollbackExecutionRequest, UUID]:
    if target_kind == "model_change_gate":
        row = await conn.fetchrow(
            """
            SELECT g.rollback_artifact_id, a.artifact_id, a.content_sha256,
                   a.provenance_uri,
                   candidate.snapshot_id AS subject_id,
                   COALESCE(g.rollback_target_snapshot_id, baseline.snapshot_id)
                     AS rollback_target_id
            FROM app.ci_model_change_gate g
            JOIN app.ci_gate_artifact a
              ON a.artifact_record_id = g.rollback_artifact_id
             AND a.owner_kind = 'model_change_gate'
             AND a.owner_id = g.gate_id
             AND a.artifact_kind = 'rollback'
            JOIN app.ci_model_calibration_snapshot candidate
              ON candidate.snapshot_record_id = g.candidate_snapshot_record_id
             AND candidate.snapshot_role = 'candidate'
            JOIN app.ci_model_calibration_snapshot baseline
              ON baseline.snapshot_record_id = g.baseline_snapshot_record_id
             AND baseline.snapshot_role = 'baseline'
            WHERE g.gate_id = $1
            """,
            target_id,
        )
        rollback_scope: Literal["model", "runtime"] = "model"
    elif target_kind == "release_gate":
        row = await conn.fetchrow(
            """
            SELECT g.rollback_artifact_id, a.artifact_id, a.content_sha256,
                   a.provenance_uri, g.release_id AS subject_id,
                   a.artifact_id AS rollback_target_id
            FROM app.ci_release_gate g
            JOIN app.ci_gate_artifact a
              ON a.artifact_record_id = g.rollback_artifact_id
             AND a.owner_kind = 'release_gate'
             AND a.owner_id = g.gate_id
             AND a.artifact_kind = 'rollback'
            WHERE g.gate_id = $1
            """,
            target_id,
        )
        rollback_scope = "runtime"
    else:
        raise ContinuousImprovementError("rollback target kind is invalid")
    if row is None:
        raise ContinuousImprovementNotFoundError(
            "rollback gate or pinned artifact not found"
        )
    artifact_record_id = UUID(str(_value(row, "rollback_artifact_id")))
    return (
        RollbackExecutionRequest(
            idempotency_key=effect_record_id,
            approval_event_id=approval_event_id,
            rollback_scope=rollback_scope,
            target_kind=target_kind,
            target_id=target_id,
            subject_id=str(_value(row, "subject_id")),
            rollback_target_id=str(_value(row, "rollback_target_id")),
            artifact_record_id=artifact_record_id,
            artifact_id=str(_value(row, "artifact_id")),
            artifact_sha256=str(_value(row, "content_sha256")),
            artifact_provenance_uri=str(_value(row, "provenance_uri")),
            authorization_evidence_refs=tuple(evidence_refs),
        ),
        artifact_record_id,
    )


def _receipt_matches_request(
    receipt: RollbackExecutionReceipt, request: RollbackExecutionRequest
) -> bool:
    return (
        receipt.idempotency_key == request.idempotency_key
        and receipt.rollback_scope == request.rollback_scope
        and receipt.target_kind == request.target_kind
        and receipt.target_id == request.target_id
        and receipt.artifact_record_id == request.artifact_record_id
        and receipt.artifact_sha256 == request.artifact_sha256
    )


async def append_human_approval(
    conn: asyncpg.Connection,
    *,
    submission_id: UUID,
    approval_event_id: UUID,
    effect_record_id: UUID,
    target_kind: str,
    target_id: UUID,
    decision: str,
    actor_uid: UUID,
    reason_code: str,
    evidence_refs: Sequence[str],
    rollback_executor: RollbackExecutor | None = None,
) -> dict[str, Any]:
    payload = {
        "approval_event_id": str(approval_event_id),
        "effect_record_id": str(effect_record_id),
        "target_kind": target_kind,
        "target_id": str(target_id),
        "decision": decision,
        "actor_uid": str(actor_uid),
        "reason_code": reason_code,
        "evidence_refs": list(evidence_refs),
    }
    content_hash = _canonical_hash(payload)
    if await _begin_submission(
        conn,
        submission_id=submission_id,
        operation_kind="human_approval",
        result_id=approval_event_id,
        payload=payload,
    ):
        lifecycle_status = None
        if decision in {"approve_promotion", "authorize_rollback"}:
            lifecycle_row = await conn.fetchrow(
                """
                SELECT event_status FROM audit.ci_lifecycle_event
                WHERE submission_id = $1
                """,
                submission_id,
            )
            lifecycle_status = _value(lifecycle_row, "event_status")
        return {
            "submission_id": submission_id,
            "approval_event_id": approval_event_id,
            "target_kind": target_kind,
            "target_id": target_id,
            "decision": decision,
            "effect_record_id": effect_record_id,
            "lifecycle_status": lifecycle_status,
            "idempotent_replay": True,
        }
    await conn.execute(
        """
        INSERT INTO audit.ci_human_approval_event (
          approval_event_id, submission_id, target_kind, target_id, decision,
          actor_uid, reason_code, evidence_refs, content_hash
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
        """,
        approval_event_id,
        submission_id,
        target_kind,
        target_id,
        decision,
        actor_uid,
        reason_code,
        list(evidence_refs),
        content_hash,
    )
    lifecycle_status: str | None = None
    if decision == "approve_content":
        row = await conn.fetchrow(
            "SELECT * FROM app.ci_content_qualification WHERE qualification_id = $1",
            target_id,
        )
        if row is None:
            raise ContinuousImprovementNotFoundError("content qualification not found")
        await conn.execute(
            """
            INSERT INTO app.ci_catalog_entry (
              catalog_record_id, qualification_id, approval_event_id,
              catalog_entry_id, payload_sha256, source_record_ids,
              review_record_ids, benchmark_record_id
            ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
            """,
            effect_record_id,
            target_id,
            approval_event_id,
            _value(row, "catalog_entry_id"),
            _value(row, "payload_sha256"),
            _value(row, "source_record_ids"),
            _value(row, "review_record_ids"),
            _value(row, "benchmark_record_id"),
        )
    elif decision in {"approve_promotion", "authorize_rollback"}:
        is_rollback = decision == "authorize_rollback"
        executor_receipt_id: str | None = None
        executor_evidence_refs: list[str] | None = None
        lifecycle_evidence_refs = list(evidence_refs)
        artifact_record_id: UUID | None = None
        execution_payload: dict[str, Any] | None = None
        if is_rollback:
            request, artifact_record_id = await _build_rollback_execution_request(
                conn,
                effect_record_id=effect_record_id,
                approval_event_id=approval_event_id,
                target_kind=target_kind,
                target_id=target_id,
                evidence_refs=evidence_refs,
            )
            lifecycle_status = "requested"
            if rollback_executor is not None:
                try:
                    raw_receipt = await rollback_executor.execute(request)
                    receipt = RollbackExecutionReceipt.model_validate(raw_receipt)
                    if not _receipt_matches_request(receipt, request):
                        raise ValueError("rollback receipt does not match request")
                except Exception:
                    lifecycle_status = "failed"
                else:
                    lifecycle_status = "executed"
                    executor_receipt_id = receipt.execution_id
                    executor_evidence_refs = list(receipt.evidence_refs)
                    lifecycle_evidence_refs = list(
                        dict.fromkeys([*evidence_refs, *receipt.evidence_refs])
                    )
                    execution_payload = receipt.model_dump(mode="json")
        else:
            table = (
                "app.ci_model_change_gate"
                if target_kind == "model_change_gate"
                else "app.ci_release_gate"
            )
            row = await conn.fetchrow(
                f"SELECT rollback_artifact_id FROM {table} WHERE gate_id = $1",
                target_id,
            )
            if row is None:
                raise ContinuousImprovementNotFoundError("release gate not found")
            lifecycle_status = "approved"
        lifecycle_content_hash = _canonical_hash(
            {
                **payload,
                "lifecycle_status": lifecycle_status,
                "executor_receipt": execution_payload,
            }
        )
        await conn.execute(
            """
            INSERT INTO audit.ci_lifecycle_event (
              lifecycle_event_id, submission_id, target_kind, target_id,
              event_type, event_status, approval_event_id, artifact_record_id,
              evidence_refs, content_hash, executor_receipt_id,
              executor_evidence_refs
            ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
            """,
            effect_record_id,
            submission_id,
            target_kind,
            target_id,
            "rollback" if is_rollback else "promotion",
            lifecycle_status,
            approval_event_id,
            artifact_record_id,
            lifecycle_evidence_refs,
            lifecycle_content_hash,
            executor_receipt_id,
            executor_evidence_refs,
        )
    return {
        "submission_id": submission_id,
        "approval_event_id": approval_event_id,
        "target_kind": target_kind,
        "target_id": target_id,
        "decision": decision,
        "effect_record_id": effect_record_id,
        "lifecycle_status": lifecycle_status,
        "idempotent_replay": False,
    }


async def append_monitor_event(
    conn: asyncpg.Connection,
    *,
    submission_id: UUID,
    lifecycle_event_id: UUID,
    target_kind: str,
    target_id: UUID,
    event_status: str,
    evidence_refs: Sequence[str],
) -> dict[str, Any]:
    payload = {
        "lifecycle_event_id": str(lifecycle_event_id),
        "target_kind": target_kind,
        "target_id": str(target_id),
        "event_status": event_status,
        "evidence_refs": list(evidence_refs),
        "data_classification": DATA_CLASSIFICATION,
    }
    content_hash = _canonical_hash(payload)
    if await _begin_submission(
        conn,
        submission_id=submission_id,
        operation_kind="monitor_event",
        result_id=lifecycle_event_id,
        payload=payload,
    ):
        return {
            "submission_id": submission_id,
            "lifecycle_event_id": lifecycle_event_id,
            "event_status": event_status,
            "idempotent_replay": True,
        }
    await conn.execute(
        """
        INSERT INTO audit.ci_lifecycle_event (
          lifecycle_event_id, submission_id, target_kind, target_id,
          event_type, event_status, evidence_refs, content_hash
        ) VALUES ($1,$2,$3,$4,'monitor',$5,$6,$7)
        """,
        lifecycle_event_id,
        submission_id,
        target_kind,
        target_id,
        event_status,
        list(evidence_refs),
        content_hash,
    )
    return {
        "submission_id": submission_id,
        "lifecycle_event_id": lifecycle_event_id,
        "event_status": event_status,
        "idempotent_replay": False,
    }


async def read_continuous_improvement_view(
    conn: asyncpg.Connection,
) -> dict[str, Any]:
    qualifications = await conn.fetch(
        """
        SELECT q.qualification_id, q.pipeline_id, q.catalog_entry_id,
               q.payload_sha256, q.gate_state, q.created_at,
               p.content_kind, p.difficulty_level, p.synthetic_identity_id,
               cardinality(q.source_record_ids) AS source_count,
               (SELECT count(*)::int FROM app.ci_red_team_review r
                WHERE r.pipeline_id = q.pipeline_id) AS red_team_review_count,
               b.variant_count AS benchmark_variant_count,
               b.variant_pass_rate AS benchmark_pass_rate,
               ARRAY(
                 SELECT s.provenance_uri
                 FROM app.ci_source_artifact s
                 WHERE s.source_record_id = ANY(q.source_record_ids)
                 ORDER BY s.provenance_uri
               ) AS source_provenance_uris,
               CASE WHEN p.draft_payload IS NULL THEN NULL ELSE jsonb_build_object(
                 'title', p.draft_payload->'title',
                 'synthetic_profile', p.draft_payload->'synthetic_profile',
                 'scenario', p.draft_payload->'scenario',
                 'rupture_or_challenge', p.draft_payload->'rupture_or_challenge',
                 'learner_task', p.draft_payload->'learner_task',
                 'success_criteria', p.draft_payload->'success_criteria',
                 'source_refs', p.draft_payload->'source_refs',
                 'grounded_claims', p.draft_payload->'grounded_claims'
               ) END AS draft_payload
        FROM app.ci_content_qualification q
        JOIN app.ci_content_pipeline p ON p.pipeline_id = q.pipeline_id
        JOIN app.ci_content_benchmark b ON b.benchmark_record_id = q.benchmark_record_id
        ORDER BY q.created_at DESC
        """
    )
    model_gates = await conn.fetch(
        """
        SELECT gate_id, gate_decision, reasons, state, created_at
        FROM app.ci_model_change_gate ORDER BY created_at DESC
        """
    )
    release_gates = await conn.fetch(
        """
        SELECT gate_id, release_id, qualified, state, created_at
        FROM app.ci_release_gate ORDER BY created_at DESC
        """
    )
    gate_artifacts = await conn.fetch(
        """
        SELECT artifact_record_id, owner_kind, owner_id, artifact_kind,
               artifact_id, content_sha256, provenance_uri, created_at
        FROM app.ci_gate_artifact ORDER BY created_at DESC
        """
    )
    approvals = await conn.fetch(
        """
        SELECT approval_event_id, target_kind, target_id, decision, reason_code,
               evidence_refs, created_at
        FROM audit.ci_human_approval_event ORDER BY created_at DESC
        """
    )
    catalog_entries = await conn.fetch(
        """
        SELECT catalog_record_id, qualification_id, catalog_entry_id, status,
               clinical_claim_allowed, created_at
        FROM app.ci_catalog_entry ORDER BY created_at DESC
        """
    )
    lifecycle = await conn.fetch(
        """
        SELECT lifecycle_event_id, target_kind, target_id, event_type,
               event_status, approval_event_id, artifact_record_id,
               evidence_refs, executor_receipt_id, executor_evidence_refs,
               created_at
        FROM audit.ci_lifecycle_event ORDER BY created_at DESC
        """
    )
    incidents = await conn.fetch(
        """
        SELECT incident_record_id, incident_id, error_fingerprint,
               affected_contract, evidence_refs, pii_included, created_at
        FROM app.ci_operational_incident ORDER BY created_at DESC
        """
    )
    regression_nodes = await conn.fetch(
        """
        SELECT node_record_id, incident_record_id, node_id, node_type,
               depends_on_record_ids, evidence_ref, node_status, created_at
        FROM app.ci_regression_dag_node ORDER BY created_at ASC
        """
    )
    return {
        "content_qualifications": [dict(row) for row in qualifications],
        "model_change_gates": [dict(row) for row in model_gates],
        "release_gates": [dict(row) for row in release_gates],
        "gate_artifacts": [dict(row) for row in gate_artifacts],
        "approvals": [dict(row) for row in approvals],
        "catalog_entries": [dict(row) for row in catalog_entries],
        "lifecycle_events": [dict(row) for row in lifecycle],
        "incidents": [dict(row) for row in incidents],
        "regression_dag_nodes": [dict(row) for row in regression_nodes],
        "data_classification": DATA_CLASSIFICATION,
        "silent_auto_promotion_allowed": False,
        "raw_transcript_included": False,
        "pii_included": False,
        "clinical_claim_allowed": False,
    }


async def read_approved_catalog_entries(
    conn: asyncpg.Connection,
) -> list[dict[str, Any]]:
    """Project only human-approved, learner-visible fields from durable catalog rows."""

    rows = await conn.fetch(
        """
        SELECT c.catalog_record_id, c.qualification_id, c.catalog_entry_id,
               c.payload_sha256, c.status, c.clinical_claim_allowed,
               c.created_at AS approved_at,
               p.content_kind, p.difficulty_level, p.synthetic_identity_id,
               ARRAY(
                 SELECT s.provenance_uri
                 FROM app.ci_source_artifact s
                 WHERE s.source_record_id = ANY(c.source_record_ids)
                 ORDER BY s.provenance_uri
               ) AS source_provenance_uris,
               jsonb_build_object(
                 'title', p.draft_payload->'title',
                 'synthetic_profile', p.draft_payload->'synthetic_profile',
                 'scenario', p.draft_payload->'scenario',
                 'rupture_or_challenge', p.draft_payload->'rupture_or_challenge',
                 'learner_task', p.draft_payload->'learner_task',
                 'success_criteria', p.draft_payload->'success_criteria',
                 'source_refs', p.draft_payload->'source_refs',
                 'grounded_claims', p.draft_payload->'grounded_claims'
               ) AS payload
        FROM app.ci_catalog_entry c
        JOIN app.ci_content_qualification q
          ON q.qualification_id = c.qualification_id
        JOIN app.ci_content_pipeline p ON p.pipeline_id = q.pipeline_id
        WHERE c.status = 'approved' AND p.draft_payload IS NOT NULL
        ORDER BY c.created_at DESC
        """
    )
    return [dict(row) for row in rows]


__all__ = [
    "ContinuousImprovementConflictError",
    "ContinuousImprovementError",
    "ContinuousImprovementNotFoundError",
    "DATA_CLASSIFICATION",
    "RollbackExecutionReceipt",
    "RollbackExecutionRequest",
    "RollbackExecutor",
    "append_human_approval",
    "append_monitor_event",
    "find_content_pipeline_submission",
    "read_approved_catalog_entries",
    "read_continuous_improvement_view",
    "read_operational_incident",
    "submit_content_pipeline",
    "submit_incident_dag",
    "submit_model_change_gate",
    "submit_release_gate",
]