1239 lines
44 KiB
Python
1239 lines
44 KiB
Python
"""Append-only persistence and automation boundary for G8 Continuous Improvement OS."""
|
|
|
|
from __future__ import annotations
|
|
|
|
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,
|
|
)
|
|
from .outcome_repository_values import (
|
|
canonical_hash as _canonical_hash,
|
|
value as _value,
|
|
)
|
|
|
|
|
|
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: ...
|
|
|
|
|
|
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",
|
|
]
|