ss-tools/backend/src/models/clean_release.py

# [DEF:backend.src.models.clean_release:Module]
# @TIER: CRITICAL
# @SEMANTICS: clean-release, models, lifecycle, compliance, evidence, immutability
# @PURPOSE: Define canonical clean release domain entities and lifecycle guards.
# @LAYER: Domain
# @INVARIANT: Immutable snapshots are never mutated; forbidden lifecycle transitions are rejected.

from datetime import datetime
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Dict, Any
from sqlalchemy import Column, String, DateTime, JSON, ForeignKey, Integer, Boolean
from sqlalchemy.orm import relationship
from .mapping import Base
from ..services.clean_release.enums import (
    CandidateStatus, RunStatus, ComplianceDecision,
    ApprovalDecisionType, PublicationStatus, ClassificationType
)
from ..services.clean_release.exceptions import IllegalTransitionError

# [DEF:CheckFinalStatus:Class]
# @PURPOSE: Backward-compatible final status enum for legacy TUI/orchestrator tests.
class CheckFinalStatus(str, Enum):
    COMPLIANT = "COMPLIANT"
    BLOCKED = "BLOCKED"
    FAILED = "FAILED"
# [/DEF:CheckFinalStatus:Class]

# [DEF:CheckStageName:Class]
# @PURPOSE: Backward-compatible stage name enum for legacy TUI/orchestrator tests.
class CheckStageName(str, Enum):
    DATA_PURITY = "DATA_PURITY"
    INTERNAL_SOURCES_ONLY = "INTERNAL_SOURCES_ONLY"
    NO_EXTERNAL_ENDPOINTS = "NO_EXTERNAL_ENDPOINTS"
    MANIFEST_CONSISTENCY = "MANIFEST_CONSISTENCY"
# [/DEF:CheckStageName:Class]

# [DEF:CheckStageStatus:Class]
# @PURPOSE: Backward-compatible stage status enum for legacy TUI/orchestrator tests.
class CheckStageStatus(str, Enum):
    PASS = "PASS"
    FAIL = "FAIL"
    SKIPPED = "SKIPPED"
    RUNNING = "RUNNING"
# [/DEF:CheckStageStatus:Class]

# [DEF:CheckStageResult:Class]
# @PURPOSE: Backward-compatible stage result container for legacy TUI/orchestrator tests.
@dataclass
class CheckStageResult:
    stage: CheckStageName
    status: CheckStageStatus
    details: str = ""
# [/DEF:CheckStageResult:Class]

# [DEF:ProfileType:Class]
# @PURPOSE: Backward-compatible profile enum for legacy TUI bootstrap logic.
class ProfileType(str, Enum):
    ENTERPRISE_CLEAN = "enterprise-clean"
# [/DEF:ProfileType:Class]

# [DEF:RegistryStatus:Class]
# @PURPOSE: Backward-compatible registry status enum for legacy TUI bootstrap logic.
class RegistryStatus(str, Enum):
    ACTIVE = "ACTIVE"
    INACTIVE = "INACTIVE"
# [/DEF:RegistryStatus:Class]

# [DEF:ReleaseCandidateStatus:Class]
# @PURPOSE: Backward-compatible release candidate status enum for legacy TUI.
class ReleaseCandidateStatus(str, Enum):
    DRAFT = CandidateStatus.DRAFT.value
    PREPARED = CandidateStatus.PREPARED.value
    MANIFEST_BUILT = CandidateStatus.MANIFEST_BUILT.value
    CHECK_PENDING = CandidateStatus.CHECK_PENDING.value
    CHECK_RUNNING = CandidateStatus.CHECK_RUNNING.value
    CHECK_PASSED = CandidateStatus.CHECK_PASSED.value
    CHECK_BLOCKED = CandidateStatus.CHECK_BLOCKED.value
    CHECK_ERROR = CandidateStatus.CHECK_ERROR.value
    APPROVED = CandidateStatus.APPROVED.value
    PUBLISHED = CandidateStatus.PUBLISHED.value
    REVOKED = CandidateStatus.REVOKED.value
# [/DEF:ReleaseCandidateStatus:Class]

# [DEF:ResourceSourceEntry:Class]
# @PURPOSE: Backward-compatible source entry model for legacy TUI bootstrap logic.
@dataclass
class ResourceSourceEntry:
    source_id: str
    host: str
    protocol: str
    purpose: str
    enabled: bool = True
# [/DEF:ResourceSourceEntry:Class]

# [DEF:ResourceSourceRegistry:Class]
# @PURPOSE: Backward-compatible source registry model for legacy TUI bootstrap logic.
@dataclass
class ResourceSourceRegistry:
    registry_id: str
    name: str
    entries: List[ResourceSourceEntry]
    updated_at: datetime
    updated_by: str
    status: str = "ACTIVE"

    @property
    def id(self) -> str:
        return self.registry_id
# [/DEF:ResourceSourceRegistry:Class]

# [DEF:CleanProfilePolicy:Class]
# @PURPOSE: Backward-compatible policy model for legacy TUI bootstrap logic.
@dataclass
class CleanProfilePolicy:
    policy_id: str
    policy_version: str
    profile: str
    active: bool
    internal_source_registry_ref: str
    prohibited_artifact_categories: List[str]
    effective_from: datetime
    required_system_categories: Optional[List[str]] = None

    @property
    def id(self) -> str:
        return self.policy_id

    @property
    def registry_snapshot_id(self) -> str:
        return self.internal_source_registry_ref
# [/DEF:CleanProfilePolicy:Class]

# [DEF:ComplianceCheckRun:Class]
# @PURPOSE: Backward-compatible run model for legacy TUI typing/import compatibility.
@dataclass
class ComplianceCheckRun:
    check_run_id: str
    candidate_id: str
    policy_id: str
    requested_by: str
    execution_mode: str
    checks: List[CheckStageResult]
    final_status: CheckFinalStatus
# [/DEF:ComplianceCheckRun:Class]

# [DEF:ReleaseCandidate:Class]
# @PURPOSE: Represents the release unit being prepared and governed.
# @PRE: id, version, source_snapshot_ref are non-empty.
# @POST: status advances only through legal transitions.
class ReleaseCandidate(Base):
    __tablename__ = "clean_release_candidates"
    
    id = Column(String, primary_key=True)
    name = Column(String, nullable=True) # Added back for backward compatibility with some legacy DTOs
    version = Column(String, nullable=False)
    source_snapshot_ref = Column(String, nullable=False)
    build_id = Column(String, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    created_by = Column(String, nullable=False)
    status = Column(String, default=CandidateStatus.DRAFT)

    @property
    def candidate_id(self) -> str:
        return self.id

    def transition_to(self, new_status: CandidateStatus):
        """
        @PURPOSE: Enforce legal state transitions.
        @PRE: Transition must be allowed by lifecycle rules.
        """
        allowed = {
            CandidateStatus.DRAFT: [CandidateStatus.PREPARED],
            CandidateStatus.PREPARED: [CandidateStatus.MANIFEST_BUILT],
            CandidateStatus.MANIFEST_BUILT: [CandidateStatus.CHECK_PENDING],
            CandidateStatus.CHECK_PENDING: [CandidateStatus.CHECK_RUNNING],
            CandidateStatus.CHECK_RUNNING: [
                CandidateStatus.CHECK_PASSED, 
                CandidateStatus.CHECK_BLOCKED, 
                CandidateStatus.CHECK_ERROR
            ],
            CandidateStatus.CHECK_PASSED: [CandidateStatus.APPROVED, CandidateStatus.CHECK_PENDING],
            CandidateStatus.CHECK_BLOCKED: [CandidateStatus.CHECK_PENDING],
            CandidateStatus.CHECK_ERROR: [CandidateStatus.CHECK_PENDING],
            CandidateStatus.APPROVED: [CandidateStatus.PUBLISHED],
            CandidateStatus.PUBLISHED: [CandidateStatus.REVOKED],
            CandidateStatus.REVOKED: []
        }
        current_status = CandidateStatus(self.status)
        if new_status not in allowed.get(current_status, []):
            raise IllegalTransitionError(f"Forbidden transition from {current_status} to {new_status}")
        self.status = new_status.value
# [/DEF:ReleaseCandidate:Class]

# [DEF:CandidateArtifact:Class]
# @PURPOSE: Represents one artifact associated with a release candidate.
class CandidateArtifact(Base):
    __tablename__ = "clean_release_artifacts"
    
    id = Column(String, primary_key=True)
    candidate_id = Column(String, ForeignKey("clean_release_candidates.id"), nullable=False)
    path = Column(String, nullable=False)
    sha256 = Column(String, nullable=False)
    size = Column(Integer, nullable=False)
    detected_category = Column(String, nullable=True)
    declared_category = Column(String, nullable=True)
    source_uri = Column(String, nullable=True)
    source_host = Column(String, nullable=True)
    metadata_json = Column(JSON, default=dict)
# [/DEF:CandidateArtifact:Class]

# [DEF:ManifestItem:Class]
@dataclass
class ManifestItem:
    path: str
    category: str
    classification: ClassificationType
    reason: str
    checksum: Optional[str] = None
# [/DEF:ManifestItem:Class]

# [DEF:ManifestSummary:Class]
@dataclass
class ManifestSummary:
    included_count: int
    excluded_count: int
    prohibited_detected_count: int
# [/DEF:ManifestSummary:Class]

# [DEF:DistributionManifest:Class]
# @PURPOSE: Immutable snapshot of the candidate payload.
# @INVARIANT: Immutable after creation.
class DistributionManifest(Base):
    __tablename__ = "clean_release_manifests"
    
    id = Column(String, primary_key=True)
    candidate_id = Column(String, ForeignKey("clean_release_candidates.id"), nullable=False)
    manifest_version = Column(Integer, nullable=False)
    manifest_digest = Column(String, nullable=False)
    artifacts_digest = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    created_by = Column(String, nullable=False)
    source_snapshot_ref = Column(String, nullable=False)
    content_json = Column(JSON, nullable=False)
    immutable = Column(Boolean, default=True)

    # Redesign compatibility fields (not persisted directly but used by builder/facade)
    def __init__(self, **kwargs):
        # Handle fields from manifest_builder.py
        if "manifest_id" in kwargs:
            kwargs["id"] = kwargs.pop("manifest_id")
        if "generated_at" in kwargs:
            kwargs["created_at"] = kwargs.pop("generated_at")
        if "generated_by" in kwargs:
            kwargs["created_by"] = kwargs.pop("generated_by")
        if "deterministic_hash" in kwargs:
            kwargs["manifest_digest"] = kwargs.pop("deterministic_hash")
        
        # Ensure required DB fields have defaults if missing
        if "manifest_version" not in kwargs:
            kwargs["manifest_version"] = 1
        if "artifacts_digest" not in kwargs:
            kwargs["artifacts_digest"] = kwargs.get("manifest_digest", "pending")
        if "source_snapshot_ref" not in kwargs:
            kwargs["source_snapshot_ref"] = "pending"
        
        # Pack items and summary into content_json if provided
        if "items" in kwargs or "summary" in kwargs:
            content = kwargs.get("content_json", {})
            if "items" in kwargs:
                items = kwargs.pop("items")
                content["items"] = [
                    {
                        "path": i.path,
                        "category": i.category,
                        "classification": i.classification.value,
                        "reason": i.reason,
                        "checksum": i.checksum
                    } for i in items
                ]
            if "summary" in kwargs:
                summary = kwargs.pop("summary")
                content["summary"] = {
                    "included_count": summary.included_count,
                    "excluded_count": summary.excluded_count,
                    "prohibited_detected_count": summary.prohibited_detected_count
                }
            kwargs["content_json"] = content

        super().__init__(**kwargs)
# [/DEF:DistributionManifest:Class]

# [DEF:SourceRegistrySnapshot:Class]
# @PURPOSE: Immutable registry snapshot for allowed sources.
class SourceRegistrySnapshot(Base):
    __tablename__ = "clean_release_registry_snapshots"
    
    id = Column(String, primary_key=True)
    registry_id = Column(String, nullable=False)
    registry_version = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    allowed_hosts = Column(JSON, nullable=False) # List[str]
    allowed_schemes = Column(JSON, nullable=False) # List[str]
    allowed_source_types = Column(JSON, nullable=False) # List[str]
    immutable = Column(Boolean, default=True)
# [/DEF:SourceRegistrySnapshot:Class]

# [DEF:CleanPolicySnapshot:Class]
# @PURPOSE: Immutable policy snapshot used to evaluate a run.
class CleanPolicySnapshot(Base):
    __tablename__ = "clean_release_policy_snapshots"
    
    id = Column(String, primary_key=True)
    policy_id = Column(String, nullable=False)
    policy_version = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    content_json = Column(JSON, nullable=False)
    registry_snapshot_id = Column(String, ForeignKey("clean_release_registry_snapshots.id"), nullable=False)
    immutable = Column(Boolean, default=True)
# [/DEF:CleanPolicySnapshot:Class]

# [DEF:ComplianceRun:Class]
# @PURPOSE: Operational record for one compliance execution.
class ComplianceRun(Base):
    __tablename__ = "clean_release_compliance_runs"
    
    id = Column(String, primary_key=True)
    candidate_id = Column(String, ForeignKey("clean_release_candidates.id"), nullable=False)
    manifest_id = Column(String, ForeignKey("clean_release_manifests.id"), nullable=False)
    manifest_digest = Column(String, nullable=False)
    policy_snapshot_id = Column(String, ForeignKey("clean_release_policy_snapshots.id"), nullable=False)
    registry_snapshot_id = Column(String, ForeignKey("clean_release_registry_snapshots.id"), nullable=False)
    requested_by = Column(String, nullable=False)
    requested_at = Column(DateTime, default=datetime.utcnow)
    started_at = Column(DateTime, nullable=True)
    finished_at = Column(DateTime, nullable=True)
    status = Column(String, default=RunStatus.PENDING)
    final_status = Column(String, nullable=True) # ComplianceDecision
    failure_reason = Column(String, nullable=True)
    task_id = Column(String, nullable=True)

    @property
    def check_run_id(self) -> str:
        return self.id
# [/DEF:ComplianceRun:Class]

# [DEF:ComplianceStageRun:Class]
# @PURPOSE: Stage-level execution record inside a run.
class ComplianceStageRun(Base):
    __tablename__ = "clean_release_compliance_stage_runs"
    
    id = Column(String, primary_key=True)
    run_id = Column(String, ForeignKey("clean_release_compliance_runs.id"), nullable=False)
    stage_name = Column(String, nullable=False)
    status = Column(String, nullable=False)
    started_at = Column(DateTime, nullable=True)
    finished_at = Column(DateTime, nullable=True)
    decision = Column(String, nullable=True) # ComplianceDecision
    details_json = Column(JSON, default=dict)
# [/DEF:ComplianceStageRun:Class]

# [DEF:ComplianceViolation:Class]
# @PURPOSE: Violation produced by a stage.
class ComplianceViolation(Base):
    __tablename__ = "clean_release_compliance_violations"
    
    id = Column(String, primary_key=True)
    run_id = Column(String, ForeignKey("clean_release_compliance_runs.id"), nullable=False)
    stage_name = Column(String, nullable=False)
    code = Column(String, nullable=False)
    severity = Column(String, nullable=False)
    artifact_path = Column(String, nullable=True)
    artifact_sha256 = Column(String, nullable=True)
    message = Column(String, nullable=False)
    evidence_json = Column(JSON, default=dict)
# [/DEF:ComplianceViolation:Class]

# [DEF:ComplianceReport:Class]
# @PURPOSE: Immutable result derived from a completed run.
# @INVARIANT: Immutable after creation.
class ComplianceReport(Base):
    __tablename__ = "clean_release_compliance_reports"
    
    id = Column(String, primary_key=True)
    run_id = Column(String, ForeignKey("clean_release_compliance_runs.id"), nullable=False)
    candidate_id = Column(String, ForeignKey("clean_release_candidates.id"), nullable=False)
    final_status = Column(String, nullable=False) # ComplianceDecision
    summary_json = Column(JSON, nullable=False)
    generated_at = Column(DateTime, default=datetime.utcnow)
    immutable = Column(Boolean, default=True)
# [/DEF:ComplianceReport:Class]

# [DEF:ApprovalDecision:Class]
# @PURPOSE: Approval or rejection bound to a candidate and report.
class ApprovalDecision(Base):
    __tablename__ = "clean_release_approval_decisions"
    
    id = Column(String, primary_key=True)
    candidate_id = Column(String, ForeignKey("clean_release_candidates.id"), nullable=False)
    report_id = Column(String, ForeignKey("clean_release_compliance_reports.id"), nullable=False)
    decision = Column(String, nullable=False) # ApprovalDecisionType
    decided_by = Column(String, nullable=False)
    decided_at = Column(DateTime, default=datetime.utcnow)
    comment = Column(String, nullable=True)
# [/DEF:ApprovalDecision:Class]

# [DEF:PublicationRecord:Class]
# @PURPOSE: Publication or revocation record.
class PublicationRecord(Base):
    __tablename__ = "clean_release_publication_records"
    
    id = Column(String, primary_key=True)
    candidate_id = Column(String, ForeignKey("clean_release_candidates.id"), nullable=False)
    report_id = Column(String, ForeignKey("clean_release_compliance_reports.id"), nullable=False)
    published_by = Column(String, nullable=False)
    published_at = Column(DateTime, default=datetime.utcnow)
    target_channel = Column(String, nullable=False)
    publication_ref = Column(String, nullable=True)
    status = Column(String, default=PublicationStatus.ACTIVE)
# [/DEF:PublicationRecord:Class]

# [DEF:CleanReleaseAuditLog:Class]
# @PURPOSE: Represents a persistent audit log entry for clean release actions.
import uuid
class CleanReleaseAuditLog(Base):
    __tablename__ = "clean_release_audit_logs"
    
    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    candidate_id = Column(String, index=True, nullable=True)
    action = Column(String, nullable=False) # e.g. "TRANSITION", "APPROVE", "PUBLISH"
    actor = Column(String, nullable=False)
    timestamp = Column(DateTime, default=datetime.utcnow)
    details_json = Column(JSON, default=dict)
# [/DEF:CleanReleaseAuditLog:Class]

# [/DEF:backend.src.models.clean_release:Module]