# [DEF:backend.src.api.routes.clean_release:Module] # @TIER: STANDARD # @SEMANTICS: api, clean-release, candidate-preparation, compliance # @PURPOSE: Expose clean release endpoints for candidate preparation and subsequent compliance flow. # @LAYER: API # @RELATION: DEPENDS_ON -> backend.src.dependencies.get_clean_release_repository # @RELATION: DEPENDS_ON -> backend.src.services.clean_release.preparation_service # @INVARIANT: API never reports prepared status if preparation errors are present. from __future__ import annotations from datetime import datetime, timezone from typing import Any, Dict, List from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel, Field from ...core.logger import belief_scope, logger from ...dependencies import get_clean_release_repository from ...services.clean_release.preparation_service import prepare_candidate from ...services.clean_release.repository import CleanReleaseRepository from ...services.clean_release.compliance_orchestrator import CleanComplianceOrchestrator from ...services.clean_release.report_builder import ComplianceReportBuilder from ...models.clean_release import ( CheckFinalStatus, CheckStageName, CheckStageResult, CheckStageStatus, ComplianceViolation, ViolationCategory, ViolationSeverity, ) router = APIRouter(prefix="/api/clean-release", tags=["Clean Release"]) # [DEF:PrepareCandidateRequest:Class] # @PURPOSE: Request schema for candidate preparation endpoint. class PrepareCandidateRequest(BaseModel): candidate_id: str = Field(min_length=1) artifacts: List[Dict[str, Any]] = Field(default_factory=list) sources: List[str] = Field(default_factory=list) operator_id: str = Field(min_length=1) # [/DEF:PrepareCandidateRequest:Class] # [DEF:StartCheckRequest:Class] # @PURPOSE: Request schema for clean compliance check run startup. class StartCheckRequest(BaseModel): candidate_id: str = Field(min_length=1) profile: str = Field(default="enterprise-clean") execution_mode: str = Field(default="tui") triggered_by: str = Field(default="system") # [/DEF:StartCheckRequest:Class] # [DEF:prepare_candidate_endpoint:Function] # @PURPOSE: Prepare candidate with policy evaluation and deterministic manifest generation. # @PRE: Candidate and active policy exist in repository. # @POST: Returns preparation result including manifest reference and violations. @router.post("/candidates/prepare") async def prepare_candidate_endpoint( payload: PrepareCandidateRequest, repository: CleanReleaseRepository = Depends(get_clean_release_repository), ): try: result = prepare_candidate( repository=repository, candidate_id=payload.candidate_id, artifacts=payload.artifacts, sources=payload.sources, operator_id=payload.operator_id, ) return result except ValueError as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={"message": str(exc), "code": "CLEAN_PREPARATION_ERROR"}, ) # [/DEF:prepare_candidate_endpoint:Function] # [DEF:start_check:Function] # @PURPOSE: Start and finalize a clean compliance check run and persist report artifacts. # @PRE: Active policy and candidate exist. # @POST: Returns accepted payload with check_run_id and started_at. @router.post("/checks", status_code=status.HTTP_202_ACCEPTED) async def start_check( payload: StartCheckRequest, repository: CleanReleaseRepository = Depends(get_clean_release_repository), ): with belief_scope("clean_release.start_check"): logger.reason("Starting clean-release compliance check run") policy = repository.get_active_policy() if policy is None: raise HTTPException(status_code=409, detail={"message": "Active policy not found", "code": "POLICY_NOT_FOUND"}) candidate = repository.get_candidate(payload.candidate_id) if candidate is None: raise HTTPException(status_code=409, detail={"message": "Candidate not found", "code": "CANDIDATE_NOT_FOUND"}) orchestrator = CleanComplianceOrchestrator(repository) run = orchestrator.start_check_run( candidate_id=payload.candidate_id, policy_id=policy.policy_id, triggered_by=payload.triggered_by, execution_mode=payload.execution_mode, ) forced = [ CheckStageResult(stage=CheckStageName.DATA_PURITY, status=CheckStageStatus.PASS, details="ok"), CheckStageResult(stage=CheckStageName.INTERNAL_SOURCES_ONLY, status=CheckStageStatus.PASS, details="ok"), CheckStageResult(stage=CheckStageName.NO_EXTERNAL_ENDPOINTS, status=CheckStageStatus.PASS, details="ok"), CheckStageResult(stage=CheckStageName.MANIFEST_CONSISTENCY, status=CheckStageStatus.PASS, details="ok"), ] run = orchestrator.execute_stages(run, forced_results=forced) run = orchestrator.finalize_run(run) if run.final_status == CheckFinalStatus.BLOCKED: logger.explore("Run ended as BLOCKED, persisting synthetic external-source violation") violation = ComplianceViolation( violation_id=f"viol-{run.check_run_id}", check_run_id=run.check_run_id, category=ViolationCategory.EXTERNAL_SOURCE, severity=ViolationSeverity.CRITICAL, location="external.example.com", remediation="Replace with approved internal server", blocked_release=True, detected_at=datetime.now(timezone.utc), ) repository.save_violation(violation) builder = ComplianceReportBuilder(repository) report = builder.build_report_payload(run, repository.get_violations_by_check_run(run.check_run_id)) builder.persist_report(report) logger.reflect(f"Compliance report persisted for check_run_id={run.check_run_id}") return { "check_run_id": run.check_run_id, "candidate_id": run.candidate_id, "status": "running", "started_at": run.started_at.isoformat(), } # [/DEF:start_check:Function] # [DEF:get_check_status:Function] # @PURPOSE: Return terminal/intermediate status payload for a check run. # @PRE: check_run_id references an existing run. # @POST: Deterministic payload shape includes checks and violations arrays. @router.get("/checks/{check_run_id}") async def get_check_status(check_run_id: str, repository: CleanReleaseRepository = Depends(get_clean_release_repository)): with belief_scope("clean_release.get_check_status"): run = repository.get_check_run(check_run_id) if run is None: raise HTTPException(status_code=404, detail={"message": "Check run not found", "code": "CHECK_NOT_FOUND"}) logger.reflect(f"Returning check status for check_run_id={check_run_id}") return { "check_run_id": run.check_run_id, "candidate_id": run.candidate_id, "final_status": run.final_status.value, "started_at": run.started_at.isoformat(), "finished_at": run.finished_at.isoformat() if run.finished_at else None, "checks": [c.model_dump() for c in run.checks], "violations": [v.model_dump() for v in repository.get_violations_by_check_run(check_run_id)], } # [/DEF:get_check_status:Function] # [DEF:get_report:Function] # @PURPOSE: Return persisted compliance report by report_id. # @PRE: report_id references an existing report. # @POST: Returns serialized report object. @router.get("/reports/{report_id}") async def get_report(report_id: str, repository: CleanReleaseRepository = Depends(get_clean_release_repository)): with belief_scope("clean_release.get_report"): report = repository.get_report(report_id) if report is None: raise HTTPException(status_code=404, detail={"message": "Report not found", "code": "REPORT_NOT_FOUND"}) logger.reflect(f"Returning compliance report report_id={report_id}") return report.model_dump() # [/DEF:get_report:Function] # [/DEF:backend.src.api.routes.clean_release:Module]