|
|
@@ -0,0 +1,857 @@
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+from dataclasses import replace
|
|
|
+from datetime import UTC, datetime
|
|
|
+from typing import Any, cast
|
|
|
+
|
|
|
+from agent.orchestration import ArtifactRef
|
|
|
+from sqlalchemy import CursorResult, Result, RowMapping, insert, select, update
|
|
|
+from sqlalchemy.exc import IntegrityError
|
|
|
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
+
|
|
|
+from script_build_host.domain.artifacts import (
|
|
|
+ ArtifactKind,
|
|
|
+ ArtifactState,
|
|
|
+ ArtifactVersion,
|
|
|
+ BusinessArtifact,
|
|
|
+ Criterion,
|
|
|
+ DirectionGoal,
|
|
|
+ EvidenceRecordV1,
|
|
|
+ ScriptDirectionArtifactV1,
|
|
|
+)
|
|
|
+from script_build_host.domain.digests import Sha256Digest
|
|
|
+from script_build_host.domain.errors import (
|
|
|
+ ArtifactAlreadyFrozen,
|
|
|
+ ArtifactDigestMismatch,
|
|
|
+ ArtifactNotFound,
|
|
|
+ ArtifactOwnershipMismatch,
|
|
|
+ BuildNotFound,
|
|
|
+ InputHashConflict,
|
|
|
+ ProtocolViolation,
|
|
|
+)
|
|
|
+from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
|
|
|
+from script_build_host.domain.records import (
|
|
|
+ MissionBinding,
|
|
|
+ Publication,
|
|
|
+ PublicationState,
|
|
|
+ PublicationType,
|
|
|
+)
|
|
|
+from script_build_host.infrastructure.canonical_json import canonical_sha256, normalize_json
|
|
|
+from script_build_host.infrastructure.redaction import redact_text
|
|
|
+from script_build_host.infrastructure.tables import (
|
|
|
+ artifact_version_table,
|
|
|
+ input_snapshot_table,
|
|
|
+ mission_binding_table,
|
|
|
+ publication_table,
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+def _utc_now() -> datetime:
|
|
|
+ return datetime.now(UTC)
|
|
|
+
|
|
|
+
|
|
|
+def _inserted_id(result: Result[Any]) -> int:
|
|
|
+ primary_key = cast(CursorResult[Any], result).inserted_primary_key
|
|
|
+ if primary_key is None or primary_key[0] is None:
|
|
|
+ raise ProtocolViolation("database did not return an inserted primary key")
|
|
|
+ return int(primary_key[0])
|
|
|
+
|
|
|
+
|
|
|
+def _rowcount(result: Result[Any]) -> int:
|
|
|
+ return int(cast(CursorResult[Any], result).rowcount)
|
|
|
+
|
|
|
+
|
|
|
+def _snapshot_from_row(row: RowMapping) -> ScriptBuildInputSnapshotV1:
|
|
|
+ payload = dict(row["canonical_json"])
|
|
|
+ return ScriptBuildInputSnapshotV1(
|
|
|
+ snapshot_id=str(row["id"]),
|
|
|
+ script_build_id=int(row["script_build_id"]),
|
|
|
+ execution_id=int(payload["execution_id"]),
|
|
|
+ topic_build_id=int(payload["topic_build_id"]),
|
|
|
+ topic_id=int(payload["topic_id"]),
|
|
|
+ topic=dict(payload["topic"]),
|
|
|
+ account=dict(payload["account"]),
|
|
|
+ persona_points=tuple(payload.get("persona_points", [])),
|
|
|
+ section_patterns=tuple(payload.get("section_patterns", [])),
|
|
|
+ strategies=tuple(payload.get("strategies", [])),
|
|
|
+ prompt_manifest=tuple(payload.get("prompt_manifest", [])),
|
|
|
+ datasource_manifest=dict(payload.get("datasource_manifest", {})),
|
|
|
+ model_manifest=dict(payload.get("model_manifest", {})),
|
|
|
+ canonical_sha256=Sha256Digest.from_db(str(row["canonical_sha256"])).wire,
|
|
|
+ created_at=cast(datetime, row["created_at"]),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class SqlAlchemyInputSnapshotRepository:
|
|
|
+ def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
|
|
|
+ self._sessions = sessions
|
|
|
+
|
|
|
+ async def freeze(
|
|
|
+ self, assembled: ScriptBuildInput, *, version: int = 1
|
|
|
+ ) -> ScriptBuildInputSnapshotV1:
|
|
|
+ payload = cast(dict[str, Any], normalize_json(assembled.canonical_payload()))
|
|
|
+ digest = canonical_sha256(payload)
|
|
|
+ now = _utc_now()
|
|
|
+ try:
|
|
|
+ async with self._sessions() as session, session.begin():
|
|
|
+ same_hash = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(input_snapshot_table).where(
|
|
|
+ input_snapshot_table.c.script_build_id == assembled.script_build_id,
|
|
|
+ input_snapshot_table.c.canonical_sha256 == digest.hex_value,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if same_hash is not None:
|
|
|
+ return _snapshot_from_row(same_hash)
|
|
|
+ existing = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(input_snapshot_table).where(
|
|
|
+ input_snapshot_table.c.script_build_id == assembled.script_build_id,
|
|
|
+ input_snapshot_table.c.version == version,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if existing is not None:
|
|
|
+ if existing["canonical_sha256"] != digest.hex_value:
|
|
|
+ raise InputHashConflict()
|
|
|
+ return _snapshot_from_row(existing)
|
|
|
+ result = await session.execute(
|
|
|
+ insert(input_snapshot_table).values(
|
|
|
+ script_build_id=assembled.script_build_id,
|
|
|
+ version=version,
|
|
|
+ canonical_json=payload,
|
|
|
+ canonical_sha256=digest.hex_value,
|
|
|
+ prompt_manifest_json=list(assembled.prompt_manifest),
|
|
|
+ schema_version="script-build-input/v1",
|
|
|
+ created_at=now,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ snapshot_id = _inserted_id(result)
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(input_snapshot_table).where(
|
|
|
+ input_snapshot_table.c.id == snapshot_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one()
|
|
|
+ )
|
|
|
+ return _snapshot_from_row(row)
|
|
|
+ except IntegrityError as error:
|
|
|
+ same_hash = await self._find_hash(assembled.script_build_id, digest.hex_value)
|
|
|
+ if same_hash is not None:
|
|
|
+ return _snapshot_from_row(same_hash)
|
|
|
+ existing = await self._find_version(assembled.script_build_id, version)
|
|
|
+ if existing is not None and existing["canonical_sha256"] == digest.hex_value:
|
|
|
+ return _snapshot_from_row(existing)
|
|
|
+ raise InputHashConflict() from error
|
|
|
+
|
|
|
+ async def _find_hash(self, script_build_id: int, digest: str) -> RowMapping | None:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ return (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(input_snapshot_table).where(
|
|
|
+ input_snapshot_table.c.script_build_id == script_build_id,
|
|
|
+ input_snapshot_table.c.canonical_sha256 == digest,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+
|
|
|
+ async def _find_version(self, script_build_id: int, version: int) -> RowMapping | None:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ return (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(input_snapshot_table).where(
|
|
|
+ input_snapshot_table.c.script_build_id == script_build_id,
|
|
|
+ input_snapshot_table.c.version == version,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+
|
|
|
+ async def get(self, snapshot_id: str, *, script_build_id: int) -> ScriptBuildInputSnapshotV1:
|
|
|
+ try:
|
|
|
+ identifier = int(snapshot_id)
|
|
|
+ except ValueError as error:
|
|
|
+ raise BuildNotFound() from error
|
|
|
+ async with self._sessions() as session:
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(input_snapshot_table).where(
|
|
|
+ input_snapshot_table.c.id == identifier,
|
|
|
+ input_snapshot_table.c.script_build_id == script_build_id,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if row is None:
|
|
|
+ raise BuildNotFound()
|
|
|
+ return _snapshot_from_row(row)
|
|
|
+
|
|
|
+
|
|
|
+def _binding_from_row(row: RowMapping) -> MissionBinding:
|
|
|
+ return MissionBinding(
|
|
|
+ binding_id=int(row["id"]),
|
|
|
+ script_build_id=int(row["script_build_id"]),
|
|
|
+ root_trace_id=str(row["root_trace_id"]),
|
|
|
+ input_snapshot_id=int(row["input_snapshot_id"]),
|
|
|
+ active_direction_artifact_version_id=row["active_direction_artifact_version_id"],
|
|
|
+ accepted_root_artifact_version_id=row["accepted_root_artifact_version_id"],
|
|
|
+ engine_version=str(row["engine_version"]),
|
|
|
+ schema_version=str(row["schema_version"]),
|
|
|
+ created_at=cast(datetime, row["created_at"]),
|
|
|
+ updated_at=cast(datetime, row["updated_at"]),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class SqlAlchemyMissionBindingRepository:
|
|
|
+ def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
|
|
|
+ self._sessions = sessions
|
|
|
+
|
|
|
+ async def create(
|
|
|
+ self,
|
|
|
+ *,
|
|
|
+ script_build_id: int,
|
|
|
+ root_trace_id: str,
|
|
|
+ input_snapshot_id: int,
|
|
|
+ engine_version: str,
|
|
|
+ schema_version: str,
|
|
|
+ ) -> MissionBinding:
|
|
|
+ now = _utc_now()
|
|
|
+ try:
|
|
|
+ async with self._sessions() as session, session.begin():
|
|
|
+ existing = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(mission_binding_table).where(
|
|
|
+ mission_binding_table.c.script_build_id == script_build_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if existing is not None:
|
|
|
+ return self._validate_existing(
|
|
|
+ existing,
|
|
|
+ root_trace_id=root_trace_id,
|
|
|
+ input_snapshot_id=input_snapshot_id,
|
|
|
+ )
|
|
|
+ result = await session.execute(
|
|
|
+ insert(mission_binding_table).values(
|
|
|
+ script_build_id=script_build_id,
|
|
|
+ root_trace_id=root_trace_id,
|
|
|
+ input_snapshot_id=input_snapshot_id,
|
|
|
+ engine_version=engine_version,
|
|
|
+ schema_version=schema_version,
|
|
|
+ created_at=now,
|
|
|
+ updated_at=now,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ binding_id = _inserted_id(result)
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(mission_binding_table).where(
|
|
|
+ mission_binding_table.c.id == binding_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one()
|
|
|
+ )
|
|
|
+ return _binding_from_row(row)
|
|
|
+ except IntegrityError:
|
|
|
+ existing = await self._find_by_build(script_build_id)
|
|
|
+ if existing is None:
|
|
|
+ raise
|
|
|
+ return self._validate_existing(
|
|
|
+ existing,
|
|
|
+ root_trace_id=root_trace_id,
|
|
|
+ input_snapshot_id=input_snapshot_id,
|
|
|
+ )
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _validate_existing(
|
|
|
+ row: RowMapping, *, root_trace_id: str, input_snapshot_id: int
|
|
|
+ ) -> MissionBinding:
|
|
|
+ if (
|
|
|
+ row["root_trace_id"] != root_trace_id
|
|
|
+ or int(row["input_snapshot_id"]) != input_snapshot_id
|
|
|
+ ):
|
|
|
+ raise ProtocolViolation("script build is already bound to another immutable mission")
|
|
|
+ return _binding_from_row(row)
|
|
|
+
|
|
|
+ async def _find_by_build(self, script_build_id: int) -> RowMapping | None:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ return (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(mission_binding_table).where(
|
|
|
+ mission_binding_table.c.script_build_id == script_build_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+
|
|
|
+ async def get_by_build(self, script_build_id: int) -> MissionBinding:
|
|
|
+ return await self._get(mission_binding_table.c.script_build_id == script_build_id)
|
|
|
+
|
|
|
+ async def get_by_root(self, root_trace_id: str) -> MissionBinding:
|
|
|
+ return await self._get(mission_binding_table.c.root_trace_id == root_trace_id)
|
|
|
+
|
|
|
+ async def _get(self, predicate: Any) -> MissionBinding:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ row = (
|
|
|
+ (await session.execute(select(mission_binding_table).where(predicate)))
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if row is None:
|
|
|
+ raise BuildNotFound()
|
|
|
+ return _binding_from_row(row)
|
|
|
+
|
|
|
+ async def set_active_direction(
|
|
|
+ self, *, script_build_id: int, artifact_version_id: int
|
|
|
+ ) -> MissionBinding:
|
|
|
+ async with self._sessions() as session, session.begin():
|
|
|
+ result = await session.execute(
|
|
|
+ update(mission_binding_table)
|
|
|
+ .where(mission_binding_table.c.script_build_id == script_build_id)
|
|
|
+ .values(
|
|
|
+ active_direction_artifact_version_id=artifact_version_id,
|
|
|
+ updated_at=_utc_now(),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ if _rowcount(result) != 1:
|
|
|
+ raise BuildNotFound()
|
|
|
+ return await self.get_by_build(script_build_id)
|
|
|
+
|
|
|
+
|
|
|
+def _artifact_kind(value: BusinessArtifact) -> ArtifactKind:
|
|
|
+ if isinstance(value, EvidenceRecordV1):
|
|
|
+ return ArtifactKind.EVIDENCE
|
|
|
+ if isinstance(value, ScriptDirectionArtifactV1):
|
|
|
+ return ArtifactKind.DIRECTION
|
|
|
+ raise ProtocolViolation("phase one accepts only evidence and direction artifacts")
|
|
|
+
|
|
|
+
|
|
|
+def _artifact_payload(value: BusinessArtifact) -> dict[str, Any]:
|
|
|
+ return dict(value.content_payload())
|
|
|
+
|
|
|
+
|
|
|
+def _hydrate_artifact(kind: ArtifactKind, payload: dict[str, Any], digest: str) -> BusinessArtifact:
|
|
|
+ if kind is ArtifactKind.EVIDENCE:
|
|
|
+ return EvidenceRecordV1(
|
|
|
+ evidence_id=str(payload["evidence_id"]),
|
|
|
+ source_type=str(payload["source_type"]),
|
|
|
+ tool_name=str(payload["tool_name"]),
|
|
|
+ query=dict(payload["query"]),
|
|
|
+ source_refs=tuple(payload.get("source_refs", [])),
|
|
|
+ raw_artifact_ref=payload.get("raw_artifact_ref"),
|
|
|
+ summary=str(payload["summary"]),
|
|
|
+ supports=tuple(payload.get("supports", [])),
|
|
|
+ confidence=str(payload["confidence"]),
|
|
|
+ limitations=tuple(payload.get("limitations", [])),
|
|
|
+ content_sha256=digest,
|
|
|
+ created_at=datetime.fromisoformat(str(payload["created_at"]).replace("Z", "+00:00")),
|
|
|
+ )
|
|
|
+ return ScriptDirectionArtifactV1(
|
|
|
+ goals=tuple(DirectionGoal(**item) for item in payload["goals"]),
|
|
|
+ criteria=tuple(Criterion(**item) for item in payload.get("criteria", [])),
|
|
|
+ domain_criteria=tuple(Criterion(**item) for item in payload.get("domain_criteria", [])),
|
|
|
+ topic_refs=tuple(payload.get("topic_refs", [])),
|
|
|
+ persona_refs=tuple(payload.get("persona_refs", [])),
|
|
|
+ strategy_refs=tuple(payload.get("strategy_refs", [])),
|
|
|
+ evidence_refs=tuple(payload.get("evidence_refs", [])),
|
|
|
+ legacy_markdown=str(payload.get("legacy_markdown", "")),
|
|
|
+ canonical_sha256=digest,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _version_from_row(row: RowMapping) -> ArtifactVersion:
|
|
|
+ kind = ArtifactKind(str(row["artifact_type"]))
|
|
|
+ digest = Sha256Digest.from_db(str(row["canonical_sha256"])).wire
|
|
|
+ payload = dict(row["canonical_json"])
|
|
|
+ return ArtifactVersion(
|
|
|
+ artifact_version_id=int(row["id"]),
|
|
|
+ script_build_id=int(row["script_build_id"]),
|
|
|
+ task_id=str(row["task_id"]),
|
|
|
+ attempt_id=str(row["attempt_id"]),
|
|
|
+ spec_version=int(row["spec_version"]),
|
|
|
+ artifact_type=kind,
|
|
|
+ canonical_sha256=digest,
|
|
|
+ state=ArtifactState(str(row["state"])),
|
|
|
+ artifact=_hydrate_artifact(kind, payload, digest),
|
|
|
+ created_at=cast(datetime, row["created_at"]),
|
|
|
+ frozen_at=cast(datetime, row["frozen_at"]),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class SqlAlchemyScriptBusinessArtifactRepository:
|
|
|
+ _URI_PREFIX = "script-build://artifact-versions/"
|
|
|
+
|
|
|
+ def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
|
|
|
+ self._sessions = sessions
|
|
|
+
|
|
|
+ async def freeze(
|
|
|
+ self,
|
|
|
+ *,
|
|
|
+ script_build_id: int,
|
|
|
+ task_id: str,
|
|
|
+ attempt_id: str,
|
|
|
+ spec_version: int,
|
|
|
+ artifact: BusinessArtifact,
|
|
|
+ ) -> tuple[ArtifactVersion, ArtifactRef]:
|
|
|
+ kind = _artifact_kind(artifact)
|
|
|
+ payload = cast(dict[str, Any], normalize_json(_artifact_payload(artifact)))
|
|
|
+ digest = canonical_sha256(payload)
|
|
|
+ if isinstance(artifact, EvidenceRecordV1):
|
|
|
+ artifact = replace(artifact, content_sha256=digest.wire)
|
|
|
+ else:
|
|
|
+ artifact = replace(artifact, canonical_sha256=digest.wire)
|
|
|
+ now = _utc_now()
|
|
|
+ try:
|
|
|
+ async with self._sessions() as session, session.begin():
|
|
|
+ existing = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(artifact_version_table).where(
|
|
|
+ artifact_version_table.c.attempt_id == attempt_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if existing is not None:
|
|
|
+ version = self._validate_replay(
|
|
|
+ existing,
|
|
|
+ script_build_id=script_build_id,
|
|
|
+ task_id=task_id,
|
|
|
+ spec_version=spec_version,
|
|
|
+ kind=kind,
|
|
|
+ digest=digest,
|
|
|
+ )
|
|
|
+ return version, self._to_ref(version)
|
|
|
+ result = await session.execute(
|
|
|
+ insert(artifact_version_table).values(
|
|
|
+ script_build_id=script_build_id,
|
|
|
+ task_id=task_id,
|
|
|
+ attempt_id=attempt_id,
|
|
|
+ spec_version=spec_version,
|
|
|
+ artifact_type=kind.value,
|
|
|
+ canonical_json=payload,
|
|
|
+ canonical_sha256=digest.hex_value,
|
|
|
+ state=ArtifactState.FROZEN.value,
|
|
|
+ created_at=now,
|
|
|
+ frozen_at=now,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ artifact_id = _inserted_id(result)
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(artifact_version_table).where(
|
|
|
+ artifact_version_table.c.id == artifact_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one()
|
|
|
+ )
|
|
|
+ version = _version_from_row(row)
|
|
|
+ return version, self._to_ref(version)
|
|
|
+ except IntegrityError:
|
|
|
+ existing = await self._find_by_attempt(attempt_id)
|
|
|
+ if existing is None:
|
|
|
+ raise
|
|
|
+ version = self._validate_replay(
|
|
|
+ existing,
|
|
|
+ script_build_id=script_build_id,
|
|
|
+ task_id=task_id,
|
|
|
+ spec_version=spec_version,
|
|
|
+ kind=kind,
|
|
|
+ digest=digest,
|
|
|
+ )
|
|
|
+ return version, self._to_ref(version)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _validate_replay(
|
|
|
+ row: RowMapping,
|
|
|
+ *,
|
|
|
+ script_build_id: int,
|
|
|
+ task_id: str,
|
|
|
+ spec_version: int,
|
|
|
+ kind: ArtifactKind,
|
|
|
+ digest: Sha256Digest,
|
|
|
+ ) -> ArtifactVersion:
|
|
|
+ if (
|
|
|
+ int(row["script_build_id"]) != script_build_id
|
|
|
+ or row["task_id"] != task_id
|
|
|
+ or row["artifact_type"] != kind.value
|
|
|
+ or int(row["spec_version"]) != spec_version
|
|
|
+ or row["canonical_sha256"] != digest.hex_value
|
|
|
+ ):
|
|
|
+ raise ArtifactAlreadyFrozen()
|
|
|
+ return _version_from_row(row)
|
|
|
+
|
|
|
+ async def _find_by_attempt(self, attempt_id: str) -> RowMapping | None:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ return (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(artifact_version_table).where(
|
|
|
+ artifact_version_table.c.attempt_id == attempt_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+
|
|
|
+ def _to_ref(self, value: ArtifactVersion) -> ArtifactRef:
|
|
|
+ return ArtifactRef(
|
|
|
+ uri=f"{self._URI_PREFIX}{value.artifact_version_id}",
|
|
|
+ kind=value.artifact_type.value,
|
|
|
+ version=str(value.artifact_version_id),
|
|
|
+ digest=value.canonical_sha256,
|
|
|
+ metadata={
|
|
|
+ "script_build_id": value.script_build_id,
|
|
|
+ "task_id": value.task_id,
|
|
|
+ "attempt_id": value.attempt_id,
|
|
|
+ },
|
|
|
+ )
|
|
|
+
|
|
|
+ async def read_by_ref(
|
|
|
+ self,
|
|
|
+ artifact_ref: ArtifactRef,
|
|
|
+ *,
|
|
|
+ script_build_id: int,
|
|
|
+ task_id: str | None = None,
|
|
|
+ attempt_id: str | None = None,
|
|
|
+ ) -> ArtifactVersion:
|
|
|
+ if not artifact_ref.uri.startswith(self._URI_PREFIX):
|
|
|
+ raise ProtocolViolation("artifact URI is outside the script-build artifact namespace")
|
|
|
+ raw_identifier = artifact_ref.uri.removeprefix(self._URI_PREFIX)
|
|
|
+ if not raw_identifier.isdigit() or "/" in raw_identifier:
|
|
|
+ raise ProtocolViolation("artifact URI does not contain one immutable version ID")
|
|
|
+ identifier = int(raw_identifier)
|
|
|
+ async with self._sessions() as session:
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(artifact_version_table).where(
|
|
|
+ artifact_version_table.c.id == identifier
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if row is None:
|
|
|
+ raise ArtifactNotFound()
|
|
|
+ version = _version_from_row(row)
|
|
|
+ if (
|
|
|
+ version.script_build_id != script_build_id
|
|
|
+ or (task_id is not None and version.task_id != task_id)
|
|
|
+ or (attempt_id is not None and version.attempt_id != attempt_id)
|
|
|
+ ):
|
|
|
+ raise ArtifactOwnershipMismatch()
|
|
|
+ if artifact_ref.kind != version.artifact_type.value:
|
|
|
+ raise ArtifactOwnershipMismatch()
|
|
|
+ if artifact_ref.version != str(version.artifact_version_id):
|
|
|
+ raise ArtifactOwnershipMismatch()
|
|
|
+ supplied_digest = Sha256Digest.from_wire(artifact_ref.digest or "")
|
|
|
+ if supplied_digest.wire != version.canonical_sha256:
|
|
|
+ raise ArtifactDigestMismatch()
|
|
|
+ recalculated = canonical_sha256(_artifact_payload(version.artifact)).wire
|
|
|
+ if recalculated != version.canonical_sha256:
|
|
|
+ raise ArtifactDigestMismatch()
|
|
|
+ return version
|
|
|
+
|
|
|
+ async def verify_digest(self, artifact_ref: ArtifactRef, *, script_build_id: int) -> bool:
|
|
|
+ try:
|
|
|
+ await self.read_by_ref(artifact_ref, script_build_id=script_build_id)
|
|
|
+ except (ArtifactDigestMismatch, ArtifactNotFound, ArtifactOwnershipMismatch):
|
|
|
+ return False
|
|
|
+ return True
|
|
|
+
|
|
|
+ async def get_by_id(self, artifact_version_id: int, *, script_build_id: int) -> ArtifactVersion:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(artifact_version_table).where(
|
|
|
+ artifact_version_table.c.id == artifact_version_id,
|
|
|
+ artifact_version_table.c.script_build_id == script_build_id,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if row is None:
|
|
|
+ raise ArtifactNotFound()
|
|
|
+ version = _version_from_row(row)
|
|
|
+ if canonical_sha256(_artifact_payload(version.artifact)).wire != version.canonical_sha256:
|
|
|
+ raise ArtifactDigestMismatch()
|
|
|
+ return version
|
|
|
+
|
|
|
+
|
|
|
+def _publication_from_row(row: RowMapping) -> Publication:
|
|
|
+ return Publication(
|
|
|
+ publication_id=int(row["id"]),
|
|
|
+ script_build_id=int(row["script_build_id"]),
|
|
|
+ publication_type=PublicationType(str(row["publication_type"])),
|
|
|
+ accept_decision_id=str(row["accept_decision_id"]),
|
|
|
+ artifact_version_id=int(row["artifact_version_id"]),
|
|
|
+ expected_sha256=Sha256Digest.from_db(str(row["expected_sha256"])).wire,
|
|
|
+ state=PublicationState(str(row["state"])),
|
|
|
+ attempt_count=int(row["attempt_count"]),
|
|
|
+ publication_revision=int(row["publication_revision"]),
|
|
|
+ last_error_code=row["last_error_code"],
|
|
|
+ last_error_summary=row["last_error_summary"],
|
|
|
+ created_at=cast(datetime, row["created_at"]),
|
|
|
+ updated_at=cast(datetime, row["updated_at"]),
|
|
|
+ published_at=row["published_at"],
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+class SqlAlchemyPublicationRepository:
|
|
|
+ def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
|
|
|
+ self._sessions = sessions
|
|
|
+
|
|
|
+ async def prepare(
|
|
|
+ self,
|
|
|
+ *,
|
|
|
+ script_build_id: int,
|
|
|
+ publication_type: PublicationType,
|
|
|
+ accept_decision_id: str,
|
|
|
+ artifact_version_id: int,
|
|
|
+ expected_sha256: str,
|
|
|
+ ) -> Publication:
|
|
|
+ if publication_type is not PublicationType.DIRECTION:
|
|
|
+ raise ProtocolViolation("phase one can publish only an accepted direction")
|
|
|
+ digest = Sha256Digest.from_wire(expected_sha256)
|
|
|
+ now = _utc_now()
|
|
|
+ try:
|
|
|
+ async with self._sessions() as session, session.begin():
|
|
|
+ existing = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(publication_table).where(
|
|
|
+ publication_table.c.accept_decision_id == accept_decision_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if existing is not None:
|
|
|
+ return self._validate_existing(
|
|
|
+ existing,
|
|
|
+ script_build_id=script_build_id,
|
|
|
+ publication_type=publication_type,
|
|
|
+ artifact_version_id=artifact_version_id,
|
|
|
+ digest=digest,
|
|
|
+ )
|
|
|
+ result = await session.execute(
|
|
|
+ insert(publication_table).values(
|
|
|
+ script_build_id=script_build_id,
|
|
|
+ publication_type=publication_type.value,
|
|
|
+ accept_decision_id=accept_decision_id,
|
|
|
+ artifact_version_id=artifact_version_id,
|
|
|
+ expected_sha256=digest.hex_value,
|
|
|
+ state=PublicationState.PENDING.value,
|
|
|
+ attempt_count=0,
|
|
|
+ publication_revision=0,
|
|
|
+ created_at=now,
|
|
|
+ updated_at=now,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ publication_id = _inserted_id(result)
|
|
|
+ except IntegrityError:
|
|
|
+ existing = await self._find_by_decision(accept_decision_id)
|
|
|
+ if existing is None:
|
|
|
+ raise
|
|
|
+ return self._validate_existing(
|
|
|
+ existing,
|
|
|
+ script_build_id=script_build_id,
|
|
|
+ publication_type=publication_type,
|
|
|
+ artifact_version_id=artifact_version_id,
|
|
|
+ digest=digest,
|
|
|
+ )
|
|
|
+ return await self._get(publication_id)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _validate_existing(
|
|
|
+ row: RowMapping,
|
|
|
+ *,
|
|
|
+ script_build_id: int,
|
|
|
+ publication_type: PublicationType,
|
|
|
+ artifact_version_id: int,
|
|
|
+ digest: Sha256Digest,
|
|
|
+ ) -> Publication:
|
|
|
+ if (
|
|
|
+ int(row["script_build_id"]) != script_build_id
|
|
|
+ or row["publication_type"] != publication_type.value
|
|
|
+ or int(row["artifact_version_id"]) != artifact_version_id
|
|
|
+ or row["expected_sha256"] != digest.hex_value
|
|
|
+ ):
|
|
|
+ raise ProtocolViolation("accept decision is already bound to another publication")
|
|
|
+ return _publication_from_row(row)
|
|
|
+
|
|
|
+ async def _find_by_decision(self, accept_decision_id: str) -> RowMapping | None:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ return (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(publication_table).where(
|
|
|
+ publication_table.c.accept_decision_id == accept_decision_id
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+
|
|
|
+ async def mark_published(self, publication_id: int) -> Publication:
|
|
|
+ now = _utc_now()
|
|
|
+ async with self._sessions() as session, session.begin():
|
|
|
+ publication_row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(publication_table).where(publication_table.c.id == publication_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if publication_row is None:
|
|
|
+ raise BuildNotFound()
|
|
|
+ if publication_row["state"] == PublicationState.PUBLISHED.value:
|
|
|
+ return _publication_from_row(publication_row)
|
|
|
+ artifact_version_id = publication_row["artifact_version_id"]
|
|
|
+ if artifact_version_id is None:
|
|
|
+ raise BuildNotFound()
|
|
|
+ result = await session.execute(
|
|
|
+ update(publication_table)
|
|
|
+ .where(publication_table.c.id == publication_id)
|
|
|
+ .values(
|
|
|
+ state=PublicationState.PUBLISHED.value,
|
|
|
+ publication_revision=publication_table.c.publication_revision + 1,
|
|
|
+ attempt_count=publication_table.c.attempt_count + 1,
|
|
|
+ updated_at=now,
|
|
|
+ published_at=now,
|
|
|
+ last_error_code=None,
|
|
|
+ last_error_summary=None,
|
|
|
+ )
|
|
|
+ )
|
|
|
+ if _rowcount(result) != 1:
|
|
|
+ raise BuildNotFound()
|
|
|
+ artifact_result = await session.execute(
|
|
|
+ update(artifact_version_table)
|
|
|
+ .where(
|
|
|
+ artifact_version_table.c.id == artifact_version_id,
|
|
|
+ artifact_version_table.c.state.in_(
|
|
|
+ [ArtifactState.FROZEN.value, ArtifactState.PUBLISHED.value]
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ .values(state=ArtifactState.PUBLISHED.value, published_at=now)
|
|
|
+ )
|
|
|
+ if _rowcount(artifact_result) != 1:
|
|
|
+ raise ProtocolViolation("publication artifact is not in a publishable state")
|
|
|
+ return await self._get(publication_id)
|
|
|
+
|
|
|
+ async def mark_failed(
|
|
|
+ self, publication_id: int, *, error_code: str, error_summary: str
|
|
|
+ ) -> Publication:
|
|
|
+ safe_summary = redact_text(error_summary)[:1000]
|
|
|
+ async with self._sessions() as session, session.begin():
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(publication_table).where(publication_table.c.id == publication_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if row is None:
|
|
|
+ raise BuildNotFound()
|
|
|
+ if row["state"] == PublicationState.PUBLISHED.value:
|
|
|
+ return _publication_from_row(row)
|
|
|
+ result = await session.execute(
|
|
|
+ update(publication_table)
|
|
|
+ .where(publication_table.c.id == publication_id)
|
|
|
+ .values(
|
|
|
+ state=PublicationState.FAILED.value,
|
|
|
+ attempt_count=publication_table.c.attempt_count + 1,
|
|
|
+ last_error_code=error_code[:64],
|
|
|
+ last_error_summary=safe_summary,
|
|
|
+ updated_at=_utc_now(),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ if _rowcount(result) != 1:
|
|
|
+ raise BuildNotFound()
|
|
|
+ return await self._get(publication_id)
|
|
|
+
|
|
|
+ async def _get(self, publication_id: int) -> Publication:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(publication_table).where(publication_table.c.id == publication_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ if row is None:
|
|
|
+ raise BuildNotFound()
|
|
|
+ return _publication_from_row(row)
|
|
|
+
|
|
|
+ async def get_by_build(
|
|
|
+ self, script_build_id: int, *, publication_type: PublicationType
|
|
|
+ ) -> Publication | None:
|
|
|
+ async with self._sessions() as session:
|
|
|
+ row = (
|
|
|
+ (
|
|
|
+ await session.execute(
|
|
|
+ select(publication_table)
|
|
|
+ .where(
|
|
|
+ publication_table.c.script_build_id == script_build_id,
|
|
|
+ publication_table.c.publication_type == publication_type.value,
|
|
|
+ )
|
|
|
+ .order_by(publication_table.c.id.desc())
|
|
|
+ .limit(1)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .mappings()
|
|
|
+ .one_or_none()
|
|
|
+ )
|
|
|
+ return _publication_from_row(row) if row is not None else None
|
SamLee