# [DEF:backend.src.services.clean_release.manifest_service:Module] # @TIER: CRITICAL # @SEMANTICS: clean-release, manifest, versioning, immutability, lifecycle # @PURPOSE: Build immutable distribution manifests with deterministic digest and version increment. # @LAYER: Domain # @RELATION: DEPENDS_ON -> backend.src.services.clean_release.repository # @RELATION: DEPENDS_ON -> backend.src.services.clean_release.manifest_builder # @RELATION: DEPENDS_ON -> backend.src.models.clean_release # @PRE: Candidate exists and is PREPARED or MANIFEST_BUILT; artifacts are present. # @POST: New immutable manifest is persisted with incremented version and deterministic digest. # @INVARIANT: Existing manifests are never mutated. from __future__ import annotations from typing import Any, Dict, List from ...models.clean_release import DistributionManifest from .enums import CandidateStatus from .manifest_builder import build_distribution_manifest from .repository import CleanReleaseRepository # [DEF:build_manifest_snapshot:Function] # @PURPOSE: Create a new immutable manifest version for a candidate. # @PRE: Candidate is prepared, artifacts are available, candidate_id is valid. # @POST: Returns persisted DistributionManifest with monotonically incremented version. def build_manifest_snapshot( repository: CleanReleaseRepository, candidate_id: str, created_by: str, policy_id: str = "policy-default", ) -> DistributionManifest: if not candidate_id or not candidate_id.strip(): raise ValueError("candidate_id must be non-empty") if not created_by or not created_by.strip(): raise ValueError("created_by must be non-empty") candidate = repository.get_candidate(candidate_id) if candidate is None: raise ValueError(f"candidate '{candidate_id}' not found") if candidate.status not in {CandidateStatus.PREPARED.value, CandidateStatus.MANIFEST_BUILT.value}: raise ValueError("candidate must be PREPARED or MANIFEST_BUILT to build manifest") artifacts = repository.get_artifacts_by_candidate(candidate_id) if not artifacts: raise ValueError("candidate artifacts are required to build manifest") existing = repository.get_manifests_by_candidate(candidate_id) for manifest in existing: if not manifest.immutable: raise ValueError("existing manifest immutability invariant violated") next_version = max((m.manifest_version for m in existing), default=0) + 1 manifest_id = f"manifest-{candidate_id}-v{next_version}" classified_artifacts: List[Dict[str, Any]] = [ { "path": artifact.path, "category": artifact.detected_category or "generic", "classification": "allowed", "reason": "artifact import", "checksum": artifact.sha256, } for artifact in artifacts ] manifest = build_distribution_manifest( manifest_id=manifest_id, candidate_id=candidate_id, policy_id=policy_id, generated_by=created_by, artifacts=classified_artifacts, ) manifest.manifest_version = next_version manifest.source_snapshot_ref = candidate.source_snapshot_ref manifest.artifacts_digest = manifest.manifest_digest manifest.immutable = True repository.save_manifest(manifest) if candidate.status == CandidateStatus.PREPARED.value: candidate.transition_to(CandidateStatus.MANIFEST_BUILT) repository.save_candidate(candidate) return manifest # [/DEF:build_manifest_snapshot:Function] # [/DEF:backend.src.services.clean_release.manifest_service:Module]