feat: add governance backend APIs

backend/app/api/governance.py

@@ -0,0 +1,280 @@
+from typing import Optional
+
+from fastapi import APIRouter, Depends, Form, Header
+from sqlalchemy.orm import Session
+
+from app.core.response import success_response
+from app.db import get_db_session
+from app.services.audit_log_service import AuditLogService
+from app.services.auth_service import AuthService
+from app.services.governance_service import GovernanceService
+
+router = APIRouter(prefix="/governance", tags=["governance"])
+ROUTE_PERMISSION_SCOPE = "governance_management"
+ROUTE_PERMISSION_ACTIONS = {
+    "assets_list": "governance:assets:list",
+    "asset_detail": "governance:assets:detail",
+    "assignments_list": "governance:assignments:list",
+    "assignment_save": "governance:assignments:save",
+    "decision_submit": "governance:decisions:submit",
+    "evidence_save": "governance:evidences:save",
+    "release_summary": "governance:releases:summary",
+    "release_detail": "governance:releases:detail",
+    "formal_launch_import": "governance:imports:formal_launch",
+}
+
+
+def get_governance_service(db: Session = Depends(get_db_session)) -> GovernanceService:
+    return GovernanceService(db)
+
+
+def get_audit_log_service(db: Session = Depends(get_db_session)) -> AuditLogService:
+    return AuditLogService(db)
+
+
+def require_current_user(
+    authorization: Optional[str] = Header(default=None),
+    db: Session = Depends(get_db_session),
+):
+    return AuthService(db).require_current_user(authorization)
+
+
+@router.get("/assets")
+def list_governance_assets(
+    page: int = 1,
+    page_size: int = 20,
+    asset_type: Optional[str] = None,
+    governance_status: Optional[str] = None,
+    review_status: Optional[str] = None,
+    production_status: Optional[str] = None,
+    assignment_role: Optional[str] = None,
+    governance_service: GovernanceService = Depends(get_governance_service),
+    current_user: dict = Depends(require_current_user),
+):
+    return success_response(
+        governance_service.list_assets(
+            current_user=current_user,
+            page=page,
+            page_size=page_size,
+            asset_type=asset_type,
+            governance_status=governance_status,
+            review_status=review_status,
+            production_status=production_status,
+            assignment_role=assignment_role,
+        )
+    )
+
+
+@router.get("/assets/{governance_asset_id}")
+def get_governance_asset_detail(
+    governance_asset_id: str,
+    governance_service: GovernanceService = Depends(get_governance_service),
+    current_user: dict = Depends(require_current_user),
+):
+    return success_response(governance_service.get_asset_detail(current_user, governance_asset_id))
+
+
+@router.get("/assignments")
+def list_governance_assignments(
+    page: int = 1,
+    page_size: int = 20,
+    assignment_role: Optional[str] = None,
+    status_filter: Optional[str] = None,
+    assignee_user_id: Optional[str] = None,
+    governance_asset_id: Optional[str] = None,
+    governance_service: GovernanceService = Depends(get_governance_service),
+    current_user: dict = Depends(require_current_user),
+):
+    return success_response(
+        governance_service.list_assignments(
+            current_user=current_user,
+            page=page,
+            page_size=page_size,
+            assignment_role=assignment_role,
+            status_filter=status_filter,
+            assignee_user_id=assignee_user_id,
+            governance_asset_id=governance_asset_id,
+        )
+    )
+
+
+@router.post("/assignments")
+def save_governance_assignment(
+    governance_asset_id: str = Form(...),
+    assignment_role: str = Form(...),
+    assignee_user_id: Optional[str] = Form(default=None),
+    assignee_name: Optional[str] = Form(default=None),
+    due_at: Optional[str] = Form(default=None),
+    status_value: Optional[str] = Form(default=None),
+    priority: Optional[str] = Form(default=None),
+    notes: Optional[str] = Form(default=None),
+    assignment_id: Optional[str] = Form(default=None),
+    is_active: bool = Form(default=True),
+    governance_service: GovernanceService = Depends(get_governance_service),
+    audit_log_service: AuditLogService = Depends(get_audit_log_service),
+    current_user: dict = Depends(require_current_user),
+):
+    result = governance_service.save_assignment(
+        current_user=current_user,
+        governance_asset_id=governance_asset_id,
+        assignment_role=assignment_role,
+        assignee_user_id=assignee_user_id,
+        assignee_name=assignee_name,
+        due_at=due_at,
+        status_value=status_value,
+        priority=priority,
+        notes=notes,
+        assignment_id=assignment_id,
+        is_active=is_active,
+    )
+    audit_log_service.record_event(
+        tenant_id=current_user.get("tenant_id"),
+        user_id=current_user.get("id"),
+        action="save_governance_assignment",
+        resource_type="governance_assignment",
+        resource_id=result["id"],
+        detail_json={
+            "governance_asset_id": result["governance_asset_id"],
+            "assignment_role": result["assignment_role"],
+            "assignee_user_id": result["assignee_user_id"],
+            "assignee_name": result["assignee_name"],
+            "status": result["status"],
+            "priority": result["priority"],
+            "is_active": result["is_active"],
+        },
+    )
+    return success_response(result)
+
+
+@router.post("/decisions")
+def submit_governance_decision(
+    governance_asset_id: str = Form(...),
+    assignment_id: Optional[str] = Form(default=None),
+    decision_role: str = Form(...),
+    decision_result: str = Form(...),
+    decision_summary: Optional[str] = Form(default=None),
+    required_changes: Optional[str] = Form(default=None),
+    decided_by_name: Optional[str] = Form(default=None),
+    decision_at: Optional[str] = Form(default=None),
+    metadata_json: Optional[str] = Form(default=None),
+    governance_service: GovernanceService = Depends(get_governance_service),
+    audit_log_service: AuditLogService = Depends(get_audit_log_service),
+    current_user: dict = Depends(require_current_user),
+):
+    result = governance_service.submit_decision(
+        current_user=current_user,
+        governance_asset_id=governance_asset_id,
+        assignment_id=assignment_id,
+        decision_role=decision_role,
+        decision_result=decision_result,
+        decision_summary=decision_summary,
+        required_changes=required_changes,
+        decided_by_name=decided_by_name,
+        decision_at=decision_at,
+        metadata_json=metadata_json,
+    )
+    audit_log_service.record_event(
+        tenant_id=current_user.get("tenant_id"),
+        user_id=current_user.get("id"),
+        action="submit_governance_decision",
+        resource_type="governance_decision",
+        resource_id=result["decision"]["id"],
+        detail_json={
+            "governance_asset_id": governance_asset_id,
+            "assignment_id": assignment_id,
+            "decision_role": result["decision"]["decision_role"],
+            "decision_result": result["decision"]["decision_result"],
+        },
+    )
+    return success_response(result)
+
+
+@router.post("/evidences")
+def save_governance_evidence(
+    governance_asset_id: str = Form(...),
+    decision_id: Optional[str] = Form(default=None),
+    file_id: Optional[str] = Form(default=None),
+    evidence_type: Optional[str] = Form(default=None),
+    storage_type: Optional[str] = Form(default=None),
+    external_url: Optional[str] = Form(default=None),
+    file_name: Optional[str] = Form(default=None),
+    checksum: Optional[str] = Form(default=None),
+    status_value: Optional[str] = Form(default=None),
+    reviewer_name: Optional[str] = Form(default=None),
+    collected_at: Optional[str] = Form(default=None),
+    metadata_json: Optional[str] = Form(default=None),
+    governance_service: GovernanceService = Depends(get_governance_service),
+    audit_log_service: AuditLogService = Depends(get_audit_log_service),
+    current_user: dict = Depends(require_current_user),
+):
+    result = governance_service.save_evidence(
+        current_user=current_user,
+        governance_asset_id=governance_asset_id,
+        decision_id=decision_id,
+        file_id=file_id,
+        evidence_type=evidence_type,
+        storage_type=storage_type,
+        external_url=external_url,
+        file_name=file_name,
+        checksum=checksum,
+        status_value=status_value,
+        reviewer_name=reviewer_name,
+        collected_at=collected_at,
+        metadata_json=metadata_json,
+    )
+    audit_log_service.record_event(
+        tenant_id=current_user.get("tenant_id"),
+        user_id=current_user.get("id"),
+        action="save_governance_evidence",
+        resource_type="governance_evidence",
+        resource_id=result["evidence"]["id"],
+        detail_json={
+            "governance_asset_id": governance_asset_id,
+            "decision_id": decision_id,
+            "status": result["evidence"]["status"],
+            "storage_type": result["evidence"]["storage_type"],
+            "external_url": result["evidence"]["external_url"],
+            "file_id": result["evidence"]["file_id"],
+        },
+    )
+    return success_response(result)
+
+
+@router.get("/releases/summary")
+def get_governance_release_summary(
+    governance_service: GovernanceService = Depends(get_governance_service),
+    current_user: dict = Depends(require_current_user),
+):
+    return success_response(governance_service.get_release_summary(current_user))
+
+
+@router.get("/releases/{release_batch_id}")
+def get_governance_release_detail(
+    release_batch_id: str,
+    governance_service: GovernanceService = Depends(get_governance_service),
+    current_user: dict = Depends(require_current_user),
+):
+    return success_response(governance_service.get_release_detail(current_user, release_batch_id))
+
+
+@router.post("/imports/formal-launch")
+def import_formal_launch_governance(
+    batch_name: Optional[str] = Form(default=None),
+    governance_service: GovernanceService = Depends(get_governance_service),
+    audit_log_service: AuditLogService = Depends(get_audit_log_service),
+    current_user: dict = Depends(require_current_user),
+):
+    result = governance_service.import_formal_launch(current_user, batch_name=batch_name)
+    audit_log_service.record_event(
+        tenant_id=current_user.get("tenant_id"),
+        user_id=current_user.get("id"),
+        action="import_formal_launch_governance",
+        resource_type="governance_import",
+        resource_id=result["release_batch"]["id"],
+        detail_json={
+            "batch_name": result["release_batch"]["batch_name"],
+            "status": result["release_batch"]["status"],
+            "imported": result["imported"],
+        },
+    )
+    return success_response(result)

backend/app/main.py

@@ -8,6 +8,7 @@ from app.api.chat import router as chat_router
 from app.api.contracts import router as contracts_router
 from app.api.conversations import router as conversations_router
 from app.api.files import router as files_router
+from app.api.governance import router as governance_router
 from app.api.knowledge_bases import router as knowledge_bases_router
 from app.api.review_rules import router as review_rules_router
 from app.api.reviews import router as reviews_router
@@ -34,6 +35,7 @@ app.include_router(chat_router, prefix=settings.api_v1_prefix)
 app.include_router(contracts_router, prefix=settings.api_v1_prefix)
 app.include_router(conversations_router, prefix=settings.api_v1_prefix)
 app.include_router(files_router, prefix=settings.api_v1_prefix)
+app.include_router(governance_router, prefix=settings.api_v1_prefix)
 app.include_router(knowledge_bases_router, prefix=settings.api_v1_prefix)
 app.include_router(review_rules_router, prefix=settings.api_v1_prefix)
 app.include_router(templates_router, prefix=settings.api_v1_prefix)

backend/app/repositories/governance_asset_repository.py

@@ -0,0 +1,117 @@
+from typing import Iterable, List, Optional
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.models.governance_asset import GovernanceAsset
+
+
+class GovernanceAssetRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, asset: GovernanceAsset) -> GovernanceAsset:
+        self.db.add(asset)
+        self.db.commit()
+        self.db.refresh(asset)
+        return asset
+
+    def update(self, asset: GovernanceAsset) -> GovernanceAsset:
+        self.db.add(asset)
+        self.db.commit()
+        self.db.refresh(asset)
+        return asset
+
+    def get_by_id(self, asset_id: str, tenant_id: Optional[str] = None) -> Optional[GovernanceAsset]:
+        query = self.db.query(GovernanceAsset).filter(GovernanceAsset.id == asset_id)
+        if tenant_id is not None:
+            query = query.filter(GovernanceAsset.tenant_id == tenant_id)
+        return query.first()
+
+    def get_by_identity(self, tenant_id: str, asset_type: str, source_asset_id: str) -> Optional[GovernanceAsset]:
+        return (
+            self.db.query(GovernanceAsset)
+            .filter(
+                GovernanceAsset.tenant_id == tenant_id,
+                GovernanceAsset.asset_type == asset_type,
+                GovernanceAsset.asset_id == source_asset_id,
+            )
+            .first()
+        )
+
+    def list_paginated(
+        self,
+        tenant_id: str,
+        page: int,
+        page_size: int,
+        asset_ids: Optional[Iterable[str]] = None,
+        asset_type: Optional[str] = None,
+        governance_status: Optional[str] = None,
+        review_status: Optional[str] = None,
+        production_status: Optional[str] = None,
+    ) -> List[GovernanceAsset]:
+        query = self.db.query(GovernanceAsset).filter(GovernanceAsset.tenant_id == tenant_id)
+        if asset_ids is not None:
+            normalized_asset_ids = [item for item in asset_ids if item]
+            if not normalized_asset_ids:
+                return []
+            query = query.filter(GovernanceAsset.id.in_(normalized_asset_ids))
+        if asset_type:
+            query = query.filter(GovernanceAsset.asset_type == asset_type)
+        if governance_status:
+            query = query.filter(GovernanceAsset.governance_status == governance_status)
+        if review_status:
+            query = query.filter(GovernanceAsset.review_status == review_status)
+        if production_status:
+            query = query.filter(GovernanceAsset.production_status == production_status)
+        return (
+            query.order_by(GovernanceAsset.updated_at.desc(), GovernanceAsset.created_at.desc())
+            .offset((page - 1) * page_size)
+            .limit(page_size)
+            .all()
+        )
+
+    def count(
+        self,
+        tenant_id: str,
+        asset_ids: Optional[Iterable[str]] = None,
+        asset_type: Optional[str] = None,
+        governance_status: Optional[str] = None,
+        review_status: Optional[str] = None,
+        production_status: Optional[str] = None,
+    ) -> int:
+        query = self.db.query(GovernanceAsset).filter(GovernanceAsset.tenant_id == tenant_id)
+        if asset_ids is not None:
+            normalized_asset_ids = [item for item in asset_ids if item]
+            if not normalized_asset_ids:
+                return 0
+            query = query.filter(GovernanceAsset.id.in_(normalized_asset_ids))
+        if asset_type:
+            query = query.filter(GovernanceAsset.asset_type == asset_type)
+        if governance_status:
+            query = query.filter(GovernanceAsset.governance_status == governance_status)
+        if review_status:
+            query = query.filter(GovernanceAsset.review_status == review_status)
+        if production_status:
+            query = query.filter(GovernanceAsset.production_status == production_status)
+        return query.count()
+
+    def count_grouped_by_status(self, tenant_id: str, column_name: str) -> dict[str, int]:
+        column = getattr(GovernanceAsset, column_name)
+        rows = (
+            self.db.query(column, func.count(GovernanceAsset.id))
+            .filter(GovernanceAsset.tenant_id == tenant_id)
+            .group_by(column)
+            .all()
+        )
+        return {status or "unknown": count for status, count in rows}
+
+    def list_by_ids(self, tenant_id: str, governance_asset_ids: Iterable[str]) -> List[GovernanceAsset]:
+        asset_ids = [item for item in governance_asset_ids if item]
+        if not asset_ids:
+            return []
+        return (
+            self.db.query(GovernanceAsset)
+            .filter(GovernanceAsset.tenant_id == tenant_id, GovernanceAsset.id.in_(asset_ids))
+            .all()
+        )

backend/app/repositories/governance_evidence_repository.py

@@ -0,0 +1,67 @@
+from typing import Iterable, List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.governance_evidence import GovernanceEvidence
+
+
+class GovernanceEvidenceRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, evidence: GovernanceEvidence) -> GovernanceEvidence:
+        self.db.add(evidence)
+        self.db.commit()
+        self.db.refresh(evidence)
+        return evidence
+
+    def update(self, evidence: GovernanceEvidence) -> GovernanceEvidence:
+        self.db.add(evidence)
+        self.db.commit()
+        self.db.refresh(evidence)
+        return evidence
+
+    def get_by_id(self, evidence_id: str, tenant_id: Optional[str] = None) -> Optional[GovernanceEvidence]:
+        query = self.db.query(GovernanceEvidence).filter(GovernanceEvidence.id == evidence_id)
+        if tenant_id is not None:
+            query = query.filter(GovernanceEvidence.tenant_id == tenant_id)
+        return query.first()
+
+    def get_latest_by_asset(self, tenant_id: str, governance_asset_id: str) -> Optional[GovernanceEvidence]:
+        return (
+            self.db.query(GovernanceEvidence)
+            .filter(
+                GovernanceEvidence.tenant_id == tenant_id,
+                GovernanceEvidence.governance_asset_id == governance_asset_id,
+            )
+            .order_by(GovernanceEvidence.collected_at.desc().nullslast(), GovernanceEvidence.created_at.desc())
+            .first()
+        )
+
+    def list_by_asset_id(self, tenant_id: str, governance_asset_id: str) -> List[GovernanceEvidence]:
+        return (
+            self.db.query(GovernanceEvidence)
+            .filter(
+                GovernanceEvidence.tenant_id == tenant_id,
+                GovernanceEvidence.governance_asset_id == governance_asset_id,
+            )
+            .order_by(GovernanceEvidence.collected_at.desc().nullslast(), GovernanceEvidence.created_at.desc())
+            .all()
+        )
+
+    def count_by_asset_ids(self, tenant_id: str, governance_asset_ids: Iterable[str]) -> dict[str, int]:
+        asset_ids = [item for item in governance_asset_ids if item]
+        if not asset_ids:
+            return {}
+        rows = (
+            self.db.query(GovernanceEvidence.governance_asset_id, GovernanceEvidence.id)
+            .filter(
+                GovernanceEvidence.tenant_id == tenant_id,
+                GovernanceEvidence.governance_asset_id.in_(asset_ids),
+            )
+            .all()
+        )
+        counts: dict[str, int] = {}
+        for governance_asset_id, _ in rows:
+            counts[governance_asset_id] = counts.get(governance_asset_id, 0) + 1
+        return counts

backend/app/repositories/governance_gate_run_repository.py

@@ -0,0 +1,27 @@
+from typing import List
+
+from sqlalchemy.orm import Session
+
+from app.models.governance_gate_run import GovernanceGateRun
+
+
+class GovernanceGateRunRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, gate_run: GovernanceGateRun) -> GovernanceGateRun:
+        self.db.add(gate_run)
+        self.db.commit()
+        self.db.refresh(gate_run)
+        return gate_run
+
+    def list_by_release_batch_id(self, tenant_id: str, release_batch_id: str) -> List[GovernanceGateRun]:
+        return (
+            self.db.query(GovernanceGateRun)
+            .filter(
+                GovernanceGateRun.tenant_id == tenant_id,
+                GovernanceGateRun.release_batch_id == release_batch_id,
+            )
+            .order_by(GovernanceGateRun.executed_at.desc(), GovernanceGateRun.created_at.desc())
+            .all()
+        )

backend/app/repositories/governance_release_batch_repository.py

@@ -0,0 +1,37 @@
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.governance_release_batch import GovernanceReleaseBatch
+
+
+class GovernanceReleaseBatchRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, release_batch: GovernanceReleaseBatch) -> GovernanceReleaseBatch:
+        self.db.add(release_batch)
+        self.db.commit()
+        self.db.refresh(release_batch)
+        return release_batch
+
+    def update(self, release_batch: GovernanceReleaseBatch) -> GovernanceReleaseBatch:
+        self.db.add(release_batch)
+        self.db.commit()
+        self.db.refresh(release_batch)
+        return release_batch
+
+    def get_by_id(self, release_batch_id: str, tenant_id: Optional[str] = None) -> Optional[GovernanceReleaseBatch]:
+        query = self.db.query(GovernanceReleaseBatch).filter(GovernanceReleaseBatch.id == release_batch_id)
+        if tenant_id is not None:
+            query = query.filter(GovernanceReleaseBatch.tenant_id == tenant_id)
+        return query.first()
+
+    def list_recent(self, tenant_id: str, limit: int = 10) -> List[GovernanceReleaseBatch]:
+        return (
+            self.db.query(GovernanceReleaseBatch)
+            .filter(GovernanceReleaseBatch.tenant_id == tenant_id)
+            .order_by(GovernanceReleaseBatch.started_at.desc(), GovernanceReleaseBatch.created_at.desc())
+            .limit(limit)
+            .all()
+        )

backend/app/repositories/governance_review_assignment_repository.py

@@ -0,0 +1,122 @@
+from typing import Iterable, List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.governance_review_assignment import GovernanceReviewAssignment
+
+
+class GovernanceReviewAssignmentRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, assignment: GovernanceReviewAssignment) -> GovernanceReviewAssignment:
+        self.db.add(assignment)
+        self.db.commit()
+        self.db.refresh(assignment)
+        return assignment
+
+    def update(self, assignment: GovernanceReviewAssignment) -> GovernanceReviewAssignment:
+        self.db.add(assignment)
+        self.db.commit()
+        self.db.refresh(assignment)
+        return assignment
+
+    def get_by_id(self, assignment_id: str, tenant_id: Optional[str] = None) -> Optional[GovernanceReviewAssignment]:
+        query = self.db.query(GovernanceReviewAssignment).filter(GovernanceReviewAssignment.id == assignment_id)
+        if tenant_id is not None:
+            query = query.filter(GovernanceReviewAssignment.tenant_id == tenant_id)
+        return query.first()
+
+    def get_active_by_asset_role(self, tenant_id: str, governance_asset_id: str, assignment_role: str) -> Optional[GovernanceReviewAssignment]:
+        return (
+            self.db.query(GovernanceReviewAssignment)
+            .filter(
+                GovernanceReviewAssignment.tenant_id == tenant_id,
+                GovernanceReviewAssignment.governance_asset_id == governance_asset_id,
+                GovernanceReviewAssignment.assignment_role == assignment_role,
+                GovernanceReviewAssignment.is_active.is_(True),
+            )
+            .order_by(GovernanceReviewAssignment.updated_at.desc(), GovernanceReviewAssignment.created_at.desc())
+            .first()
+        )
+
+    def list_paginated(
+        self,
+        tenant_id: str,
+        page: int,
+        page_size: int,
+        assignment_role: Optional[str] = None,
+        status: Optional[str] = None,
+        assignee_user_id: Optional[str] = None,
+        governance_asset_id: Optional[str] = None,
+    ) -> List[GovernanceReviewAssignment]:
+        query = self.db.query(GovernanceReviewAssignment).filter(GovernanceReviewAssignment.tenant_id == tenant_id)
+        if assignment_role:
+            query = query.filter(GovernanceReviewAssignment.assignment_role == assignment_role)
+        if status:
+            query = query.filter(GovernanceReviewAssignment.status == status)
+        if assignee_user_id:
+            query = query.filter(GovernanceReviewAssignment.assignee_user_id == assignee_user_id)
+        if governance_asset_id:
+            query = query.filter(GovernanceReviewAssignment.governance_asset_id == governance_asset_id)
+        return (
+            query.order_by(GovernanceReviewAssignment.updated_at.desc(), GovernanceReviewAssignment.created_at.desc())
+            .offset((page - 1) * page_size)
+            .limit(page_size)
+            .all()
+        )
+
+    def count(
+        self,
+        tenant_id: str,
+        assignment_role: Optional[str] = None,
+        status: Optional[str] = None,
+        assignee_user_id: Optional[str] = None,
+        governance_asset_id: Optional[str] = None,
+    ) -> int:
+        query = self.db.query(GovernanceReviewAssignment).filter(GovernanceReviewAssignment.tenant_id == tenant_id)
+        if assignment_role:
+            query = query.filter(GovernanceReviewAssignment.assignment_role == assignment_role)
+        if status:
+            query = query.filter(GovernanceReviewAssignment.status == status)
+        if assignee_user_id:
+            query = query.filter(GovernanceReviewAssignment.assignee_user_id == assignee_user_id)
+        if governance_asset_id:
+            query = query.filter(GovernanceReviewAssignment.governance_asset_id == governance_asset_id)
+        return query.count()
+
+    def list_active_by_asset_ids(self, tenant_id: str, governance_asset_ids: Iterable[str]) -> List[GovernanceReviewAssignment]:
+        asset_ids = [item for item in governance_asset_ids if item]
+        if not asset_ids:
+            return []
+        return (
+            self.db.query(GovernanceReviewAssignment)
+            .filter(
+                GovernanceReviewAssignment.tenant_id == tenant_id,
+                GovernanceReviewAssignment.governance_asset_id.in_(asset_ids),
+                GovernanceReviewAssignment.is_active.is_(True),
+            )
+            .all()
+        )
+
+    def list_active_by_role(self, tenant_id: str, assignment_role: str) -> List[GovernanceReviewAssignment]:
+        return (
+            self.db.query(GovernanceReviewAssignment)
+            .filter(
+                GovernanceReviewAssignment.tenant_id == tenant_id,
+                GovernanceReviewAssignment.assignment_role == assignment_role,
+                GovernanceReviewAssignment.is_active.is_(True),
+            )
+            .all()
+        )
+
+    def list_by_asset_id(self, tenant_id: str, governance_asset_id: str) -> List[GovernanceReviewAssignment]:
+        return (
+            self.db.query(GovernanceReviewAssignment)
+            .filter(
+                GovernanceReviewAssignment.tenant_id == tenant_id,
+                GovernanceReviewAssignment.governance_asset_id == governance_asset_id,
+            )
+            .order_by(GovernanceReviewAssignment.updated_at.desc(), GovernanceReviewAssignment.created_at.desc())
+            .all()
+        )

backend/app/repositories/governance_review_decision_repository.py

@@ -0,0 +1,56 @@
+from typing import Iterable, List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.governance_review_decision import GovernanceReviewDecision
+
+
+class GovernanceReviewDecisionRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, decision: GovernanceReviewDecision) -> GovernanceReviewDecision:
+        self.db.add(decision)
+        self.db.commit()
+        self.db.refresh(decision)
+        return decision
+
+    def get_latest_by_asset_role(self, tenant_id: str, governance_asset_id: str, decision_role: str) -> Optional[GovernanceReviewDecision]:
+        return (
+            self.db.query(GovernanceReviewDecision)
+            .filter(
+                GovernanceReviewDecision.tenant_id == tenant_id,
+                GovernanceReviewDecision.governance_asset_id == governance_asset_id,
+                GovernanceReviewDecision.decision_role == decision_role,
+            )
+            .order_by(GovernanceReviewDecision.decision_at.desc().nullslast(), GovernanceReviewDecision.created_at.desc())
+            .first()
+        )
+
+    def list_by_asset_id(self, tenant_id: str, governance_asset_id: str) -> List[GovernanceReviewDecision]:
+        return (
+            self.db.query(GovernanceReviewDecision)
+            .filter(
+                GovernanceReviewDecision.tenant_id == tenant_id,
+                GovernanceReviewDecision.governance_asset_id == governance_asset_id,
+            )
+            .order_by(GovernanceReviewDecision.decision_at.desc().nullslast(), GovernanceReviewDecision.created_at.desc())
+            .all()
+        )
+
+    def count_by_asset_ids(self, tenant_id: str, governance_asset_ids: Iterable[str]) -> dict[str, int]:
+        asset_ids = [item for item in governance_asset_ids if item]
+        if not asset_ids:
+            return {}
+        rows = (
+            self.db.query(GovernanceReviewDecision.governance_asset_id, GovernanceReviewDecision.id)
+            .filter(
+                GovernanceReviewDecision.tenant_id == tenant_id,
+                GovernanceReviewDecision.governance_asset_id.in_(asset_ids),
+            )
+            .all()
+        )
+        counts: dict[str, int] = {}
+        for governance_asset_id, _ in rows:
+            counts[governance_asset_id] = counts.get(governance_asset_id, 0) + 1
+        return counts

backend/tests/test_governance_api.py

@@ -0,0 +1,303 @@
+import unittest
+
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+from app.api.governance import get_audit_log_service, get_governance_service, require_current_user
+from app.main import app
+
+
+class StubAuditLogService:
+    def __init__(self):
+        self.events = []
+
+    def record_event(self, **kwargs):
+        self.events.append(kwargs)
+        return kwargs
+
+
+class StubGovernanceService:
+    def list_assets(self, current_user, page, page_size, asset_type, governance_status, review_status, production_status, assignment_role):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "items": [
+                {
+                    "id": "govast-123",
+                    "tenant_id": current_user["tenant_id"],
+                    "asset_type": asset_type or "contract_template",
+                    "asset_id": "TPL-SALES-GOODS-001",
+                    "asset_name": "销售合同模板",
+                    "source_table": "formal_launch_template_signoff",
+                    "source_path": "project-docs/formal-launch-ops/signoff/production-template-signoff-register.csv",
+                    "governance_status": governance_status or "reviewing",
+                    "review_status": review_status or "pending",
+                    "production_status": production_status or "blocked",
+                    "latest_decision_result": None,
+                    "latest_evidence_status": "template",
+                    "metadata": {"contract_type": "sales"},
+                    "notes": None,
+                    "active_assignment_roles": [assignment_role or "legal_reviewer"],
+                    "decision_count": 0,
+                    "evidence_count": 1,
+                    "created_at": "2026-05-02T00:00:00+00:00",
+                    "updated_at": "2026-05-02T00:00:00+00:00",
+                }
+            ],
+            "page": page,
+            "page_size": page_size,
+            "total": 1,
+            "summary": {
+                "total_assets": 1,
+                "governance_status": {"reviewing": 1},
+                "review_status": {"pending": 1},
+                "production_status": {"blocked": 1},
+                "pending_assignments": 1,
+                "in_review_assignments": 0,
+                "blocking_items": [{"type": "assignment", "status": "pending", "count": 1}],
+            },
+        }
+
+    def get_asset_detail(self, current_user, governance_asset_id):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "asset": {
+                "id": governance_asset_id,
+                "asset_name": "销售合同模板",
+            },
+            "assignments": [{"id": "govasg-123"}],
+            "decisions": [],
+            "evidences": [{"id": "govev-123"}],
+        }
+
+    def list_assignments(self, current_user, page, page_size, assignment_role, status_filter, assignee_user_id, governance_asset_id):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "items": [
+                {
+                    "id": "govasg-123",
+                    "tenant_id": current_user["tenant_id"],
+                    "governance_asset_id": governance_asset_id or "govast-123",
+                    "assignment_role": assignment_role or "legal_reviewer",
+                    "assignee_user_id": assignee_user_id,
+                    "assignee_name": "法务负责人A",
+                    "due_at": "2026-05-03T00:00:00+00:00",
+                    "status": status_filter or "pending",
+                    "priority": "high",
+                    "is_active": True,
+                    "notes": None,
+                    "created_at": "2026-05-02T00:00:00+00:00",
+                    "updated_at": "2026-05-02T00:00:00+00:00",
+                    "asset_name": "销售合同模板",
+                    "asset_type": "contract_template",
+                }
+            ],
+            "page": page,
+            "page_size": page_size,
+            "total": 1,
+        }
+
+    def save_assignment(self, current_user, governance_asset_id, assignment_role, assignee_user_id, assignee_name, due_at, status_value, priority, notes, assignment_id=None, is_active=True):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "id": assignment_id or "govasg-123",
+            "tenant_id": current_user["tenant_id"],
+            "governance_asset_id": governance_asset_id,
+            "assignment_role": assignment_role,
+            "assignee_user_id": assignee_user_id,
+            "assignee_name": assignee_name,
+            "due_at": due_at,
+            "status": status_value or "pending",
+            "priority": priority or "normal",
+            "is_active": is_active,
+            "notes": notes,
+            "created_at": "2026-05-02T00:00:00+00:00",
+            "updated_at": "2026-05-02T00:00:00+00:00",
+            "asset_name": "销售合同模板",
+            "asset_type": "contract_template",
+        }
+
+    def submit_decision(self, current_user, governance_asset_id, assignment_id, decision_role, decision_result, decision_summary, required_changes, decided_by_name, decision_at, metadata_json):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "decision": {
+                "id": "govdec-123",
+                "governance_asset_id": governance_asset_id,
+                "assignment_id": assignment_id,
+                "decision_role": decision_role,
+                "decision_result": decision_result,
+            },
+            "asset": {
+                "id": governance_asset_id,
+                "review_status": "approved" if decision_result == "approved" else "blocked",
+            },
+        }
+
+    def save_evidence(self, current_user, governance_asset_id, decision_id, file_id, evidence_type, storage_type, external_url, file_name, checksum, status_value, reviewer_name, collected_at, metadata_json):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "evidence": {
+                "id": "govev-123",
+                "governance_asset_id": governance_asset_id,
+                "decision_id": decision_id,
+                "file_id": file_id,
+                "storage_type": storage_type or "external",
+                "external_url": external_url,
+                "status": status_value or "ready",
+            },
+            "asset": {
+                "id": governance_asset_id,
+                "latest_evidence_status": status_value or "ready",
+            },
+        }
+
+    def get_release_summary(self, current_user):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "summary": {
+                "total_assets": 10,
+                "blocking_items": [{"type": "assignment", "status": "pending", "count": 4}],
+            },
+            "recent_batches": [
+                {
+                    "id": "govrel-123",
+                    "batch_name": "formal-launch-import",
+                    "status": "blocked",
+                }
+            ],
+        }
+
+    def get_release_detail(self, current_user, release_batch_id):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "release_batch": {"id": release_batch_id, "status": "blocked"},
+            "gate_runs": [{"id": "govgate-123", "gate_name": "formal_launch_import"}],
+        }
+
+    def import_formal_launch(self, current_user, batch_name=None):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "imported": {
+                "assets": 10,
+                "assignments": 12,
+                "decisions": 0,
+                "evidences": 10,
+            },
+            "summary": {"total_assets": 10},
+            "release_batch": {"id": "govrel-123", "batch_name": batch_name or "formal-launch-import", "status": "blocked"},
+            "gate_run": {"id": "govgate-123", "status": "blocked"},
+        }
+
+
+class GovernanceApiTestCase(unittest.TestCase):
+    def setUp(self):
+        self.audit_log_service = StubAuditLogService()
+        app.dependency_overrides[get_governance_service] = lambda: StubGovernanceService()
+        app.dependency_overrides[get_audit_log_service] = lambda: self.audit_log_service
+        self.client = TestClient(app)
+
+    def tearDown(self):
+        app.dependency_overrides.clear()
+
+    def _set_admin(self):
+        app.dependency_overrides[require_current_user] = lambda: {
+            "id": "user-admin",
+            "tenant_id": "tenant-demo-001",
+            "role": "admin",
+            "display_name": "管理员",
+        }
+
+    def test_admin_can_list_governance_assets(self):
+        self._set_admin()
+
+        response = self.client.get("/api/v1/governance/assets", params={"asset_type": "contract_template"})
+
+        self.assertEqual(response.status_code, 200)
+        body = response.json()["data"]
+        self.assertEqual(body["items"][0]["asset_name"], "销售合同模板")
+        self.assertEqual(body["summary"]["total_assets"], 1)
+
+    def test_admin_can_list_governance_assignments(self):
+        self._set_admin()
+
+        response = self.client.get("/api/v1/governance/assignments", params={"assignment_role": "legal_reviewer"})
+
+        self.assertEqual(response.status_code, 200)
+        body = response.json()["data"]
+        self.assertEqual(body["items"][0]["assignment_role"], "legal_reviewer")
+        self.assertEqual(body["items"][0]["asset_name"], "销售合同模板")
+
+    def test_admin_can_save_assignment(self):
+        self._set_admin()
+
+        response = self.client.post(
+            "/api/v1/governance/assignments",
+            data={
+                "governance_asset_id": "govast-123",
+                "assignment_role": "legal_reviewer",
+                "assignee_name": "法务负责人A",
+                "status_value": "pending",
+                "priority": "high",
+            },
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["data"]["assignment_role"], "legal_reviewer")
+        self.assertEqual(self.audit_log_service.events[0]["action"], "save_governance_assignment")
+
+    def test_admin_can_submit_decision(self):
+        self._set_admin()
+
+        response = self.client.post(
+            "/api/v1/governance/decisions",
+            data={
+                "governance_asset_id": "govast-123",
+                "assignment_id": "govasg-123",
+                "decision_role": "legal_reviewer",
+                "decision_result": "approved",
+            },
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["data"]["decision"]["decision_result"], "approved")
+        self.assertEqual(self.audit_log_service.events[0]["action"], "submit_governance_decision")
+
+    def test_admin_can_save_evidence(self):
+        self._set_admin()
+
+        response = self.client.post(
+            "/api/v1/governance/evidences",
+            data={
+                "governance_asset_id": "govast-123",
+                "external_url": "https://example.com/evidence/1",
+                "status_value": "ready",
+            },
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["data"]["evidence"]["status"], "ready")
+        self.assertEqual(self.audit_log_service.events[0]["action"], "save_governance_evidence")
+
+    def test_admin_can_get_release_summary_and_import(self):
+        self._set_admin()
+
+        summary_response = self.client.get("/api/v1/governance/releases/summary")
+        import_response = self.client.post("/api/v1/governance/imports/formal-launch", data={"batch_name": "formal-launch-import"})
+
+        self.assertEqual(summary_response.status_code, 200)
+        self.assertEqual(summary_response.json()["data"]["recent_batches"][0]["status"], "blocked")
+        self.assertEqual(import_response.status_code, 200)
+        self.assertEqual(import_response.json()["data"]["release_batch"]["batch_name"], "formal-launch-import")
+        self.assertEqual(self.audit_log_service.events[-1]["action"], "import_formal_launch_governance")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_governance_service.py

@@ -0,0 +1,96 @@
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.db import Base
+from app.services.governance_service import GovernanceService
+import app.models  # noqa: F401
+
+
+class GovernanceServiceTestCase(unittest.TestCase):
+    def setUp(self):
+        self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+        TestingSessionLocal = sessionmaker(bind=self.engine, autoflush=False, autocommit=False, future=True)
+        Base.metadata.create_all(self.engine)
+        self.db = TestingSessionLocal()
+        self.service = GovernanceService(self.db)
+        self.admin_user = {
+            "id": "user-admin",
+            "tenant_id": "tenant-demo-001",
+            "role": "admin",
+            "display_name": "管理员",
+            "username": "admin",
+        }
+
+    def tearDown(self):
+        self.db.close()
+        self.engine.dispose()
+
+    def test_assignment_decision_and_evidence_refresh_asset_status(self):
+        asset, _ = self.service._upsert_governance_asset(
+            tenant_id=self.admin_user["tenant_id"],
+            asset_type="contract_template",
+            asset_id="TPL-DEMO-001",
+            asset_name="演示模板",
+            source_path=None,
+            source_table="manual",
+            metadata={},
+        )
+
+        assignment = self.service.save_assignment(
+            current_user=self.admin_user,
+            governance_asset_id=asset.id,
+            assignment_role="legal_reviewer",
+            assignee_user_id=None,
+            assignee_name="法务A",
+            due_at="2026-05-03",
+            status_value="pending",
+            priority="high",
+            notes=None,
+        )
+        self.assertEqual(assignment["status"], "pending")
+
+        self.service.save_evidence(
+            current_user=self.admin_user,
+            governance_asset_id=asset.id,
+            decision_id=None,
+            file_id=None,
+            evidence_type="document",
+            storage_type="external",
+            external_url="https://example.com/evidence/ready",
+            file_name="evidence.md",
+            checksum=None,
+            status_value="ready",
+            reviewer_name="法务A",
+            collected_at="2026-05-02",
+            metadata_json=None,
+        )
+
+        decision_result = self.service.submit_decision(
+            current_user=self.admin_user,
+            governance_asset_id=asset.id,
+            assignment_id=assignment["id"],
+            decision_role="legal_reviewer",
+            decision_result="approved",
+            decision_summary="通过",
+            required_changes=None,
+            decided_by_name="法务A",
+            decision_at="2026-05-02",
+            metadata_json=None,
+        )
+
+        self.assertEqual(decision_result["asset"]["review_status"], "approved")
+        self.assertEqual(decision_result["asset"]["production_status"], "ready")
+
+    def test_import_formal_launch_creates_assets_and_release_batch(self):
+        result = self.service.import_formal_launch(self.admin_user, batch_name="formal-launch-import")
+
+        self.assertGreater(result["imported"]["assets"], 0)
+        self.assertGreater(result["summary"]["total_assets"], 0)
+        self.assertEqual(result["release_batch"]["batch_name"], "formal-launch-import")
+        self.assertIn(result["release_batch"]["status"], {"blocked", "in_review", "approved"})
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_permission_metadata.py

@@ -1,6 +1,6 @@
 import unittest
 
-from app.api import audit_logs, auth, chat, contracts, conversations, files, knowledge_bases, review_rules, reviews, system, templates, tenant_settings, users
+from app.api import audit_logs, auth, chat, contracts, conversations, files, governance, knowledge_bases, review_rules, reviews, system, templates, tenant_settings, users
 
 
 class PermissionMetadataTestCase(unittest.TestCase):
@@ -12,6 +12,7 @@ class PermissionMetadataTestCase(unittest.TestCase):
             contracts,
             conversations,
             files,
+            governance,
             knowledge_bases,
             review_rules,
             reviews,
@@ -35,6 +36,7 @@ class PermissionMetadataTestCase(unittest.TestCase):
             contracts,
             conversations,
             files,
+            governance,
             knowledge_bases,
             review_rules,
             reviews,

project-docs/formal-launch-ops/governance-productization-backlog.csv

@@ -6,12 +6,12 @@ batch,priority,area,task_id,title,current_gap,target_deliverable,depends_on,acce
 5A,P0,database,GOV-DB-005,新增发布批次与门禁结果表,当前上线批准状态只存在 PowerShell 产物中,新增 governance_release_batches 和 governance_gate_runs 表,GOV-DB-001;GOV-DB-003;GOV-DB-004,系统可追踪一次发布批次的门禁快照和结果,COMPLETED
 5A,P1,database,GOV-DB-006,模板表补治理字段,contract_templates 只有内容和启停没有版本与发布元数据,为 contract_templates 增加 template_code version published_at published_by latest_governance_asset_id,GOV-DB-001,模板可区分草稿版和生产发布版,PLANNED
 5A,P1,database,GOV-DB-007,知识库文档表补法源元数据,知识库文档缺少效力级别生效日期发文机关和法源类型,为知识库相关表增加 source_type authority effective_date region tags,none,知识库可按法源维度治理和过滤,PLANNED
-5B,P0,backend_api,GOV-API-001,新增治理资产列表接口,当前没有统一治理视图接口,新增 /governance/assets 列表筛选汇总接口,GOV-DB-001,前端可按资产类型状态角色查看治理资产,PLANNED
-5B,P0,backend_api,GOV-API-002,新增复核分配接口,当前无法在系统内分配真实复核人,新增 /governance/assignments 创建更新撤销接口,GOV-DB-002,管理员可在系统内完成复核任务分配,PLANNED
-5B,P0,backend_api,GOV-API-003,新增复核回填接口,当前复核结论只能通过CSV和脚本回填,新增 /governance/decisions 提交接口和批量导入接口,GOV-DB-003,真实复核人或管理员可提交复核结果,PLANNED
-5B,P0,backend_api,GOV-API-004,新增证据上传绑定接口,当前系统没有专门的治理证据绑定接口,新增 /governance/evidences 上传绑定替换校验接口,GOV-DB-004,系统可阻止模板证据进入已批准状态,PLANNED
-5B,P0,backend_api,GOV-API-005,新增上线批准汇总接口,当前没有系统内 release approval summary,新增 /governance/releases/summary 和 /governance/releases/{id} 接口,GOV-DB-005,前端可直接展示当前上线阻断项与通过项,PLANNED
-5B,P0,backend_api,GOV-API-006,新增CSV台账导入接口,当前正式上线台账只靠仓库脚本导入,新增 /governance/imports/formal-launch 接口读取既有 CSV 映射到治理表,GOV-DB-001;GOV-DB-002;GOV-DB-003;GOV-DB-004,旧台账可一次性导入数据库且保留来源信息,PLANNED
+5B,P0,backend_api,GOV-API-001,新增治理资产列表接口,当前没有统一治理视图接口,新增 /governance/assets 列表筛选汇总接口,GOV-DB-001,前端可按资产类型状态角色查看治理资产,COMPLETED
+5B,P0,backend_api,GOV-API-002,新增复核分配接口,当前无法在系统内分配真实复核人,新增 /governance/assignments 创建更新撤销接口,GOV-DB-002,管理员可在系统内完成复核任务分配,COMPLETED
+5B,P0,backend_api,GOV-API-003,新增复核回填接口,当前复核结论只能通过CSV和脚本回填,新增 /governance/decisions 提交接口和批量导入接口,GOV-DB-003,真实复核人或管理员可提交复核结果,COMPLETED
+5B,P0,backend_api,GOV-API-004,新增证据上传绑定接口,当前系统没有专门的治理证据绑定接口,新增 /governance/evidences 上传绑定替换校验接口,GOV-DB-004,系统可阻止模板证据进入已批准状态,COMPLETED
+5B,P0,backend_api,GOV-API-005,新增上线批准汇总接口,当前没有系统内 release approval summary,新增 /governance/releases/summary 和 /governance/releases/{id} 接口,GOV-DB-005,前端可直接展示当前上线阻断项与通过项,COMPLETED
+5B,P0,backend_api,GOV-API-006,新增CSV台账导入接口,当前正式上线台账只靠仓库脚本导入,新增 /governance/imports/formal-launch 接口读取既有 CSV 映射到治理表,GOV-DB-001;GOV-DB-002;GOV-DB-003;GOV-DB-004,旧台账可一次性导入数据库且保留来源信息,COMPLETED
 5B,P1,backend_api,GOV-API-007,模板规则知识库发布接口,当前内容模块没有生产发布动作,为 templates review-rules knowledge-bases 增加 publish/unpublish/rollback 接口,GOV-DB-005;GOV-DB-006,已批准资产才能被发布到生产可用集合,PLANNED
 5C,P0,frontend,GOV-WEB-001,新增复核总控中心页面,当前没有统一治理控制台,新增 /governance 页面展示资产统计阻断项角色视图和发布批次,GOV-API-001;GOV-API-005,管理员可在单页看到当前正式上线治理全貌,PLANNED
 5C,P0,frontend,GOV-WEB-002,新增复核人工作台页面,当前真实法务业务运营没有系统内待办页,新增 /governance/workbench 页面支持按角色查看待处理任务,GOV-API-001;GOV-API-002;GOV-API-003,复核人可只看到与自己相关的任务包,PLANNED

project-docs/正式上线运营准备阶段5B后端治理接口执行记录-2026-05-02.md

@@ -0,0 +1,120 @@
+# 正式上线运营准备阶段 5B 后端治理接口执行记录（2026-05-02）
+
+## 本批目标
+
+本批执行 `5B / GOV-API-001 ~ GOV-API-006`，目标是在已完成的治理域数据库骨架上，补齐系统内的治理接口，让“治理资产、复核任务、复核结论、证据、上线批准汇总、CSV 导入”不再只存在于仓库脚本中。
+
+## 已完成接口
+
+新增路由文件：
+
+- `backend/app/api/governance.py`
+
+并已注册到：
+
+- `backend/app/main.py`
+
+本批已完成接口：
+
+1. `GET /api/v1/governance/assets`
+   支持治理资产列表、状态筛选和汇总摘要。
+
+2. `GET /api/v1/governance/assets/{governance_asset_id}`
+   返回单资产详情、分配、决议和证据。
+
+3. `GET /api/v1/governance/assignments`
+   返回复核分配列表。
+
+4. `POST /api/v1/governance/assignments`
+   创建或更新复核分配。
+
+5. `POST /api/v1/governance/decisions`
+   提交复核结论，并回写资产治理状态。
+
+6. `POST /api/v1/governance/evidences`
+   绑定证据，并回写资产证据状态。
+
+7. `GET /api/v1/governance/releases/summary`
+   返回当前治理汇总和最近发布批次。
+
+8. `GET /api/v1/governance/releases/{release_batch_id}`
+   返回指定发布批次和门禁执行记录。
+
+9. `POST /api/v1/governance/imports/formal-launch`
+   读取现有 formal-launch CSV 台账并导入数据库。
+
+## 已完成仓储与服务
+
+新增仓储：
+
+- `governance_asset_repository.py`
+- `governance_review_assignment_repository.py`
+- `governance_review_decision_repository.py`
+- `governance_evidence_repository.py`
+- `governance_release_batch_repository.py`
+- `governance_gate_run_repository.py`
+
+新增服务：
+
+- `backend/app/services/governance_service.py`
+
+### 服务能力
+
+`GovernanceService` 已实现：
+
+- 治理资产分页查询。
+- 单资产详情聚合。
+- 复核分配查询与保存。
+- 复核结论提交。
+- 证据登记。
+- 发布汇总查询。
+- formal-launch 台账导入。
+
+## formal-launch 导入能力
+
+导入接口会读取以下已有台账：
+
+- `real-review-task-package.csv`
+- `production-material-signoff-register.csv`
+- `production-template-signoff-register.csv`
+- `production-knowledge-signoff-register.csv`
+
+导入后会自动：
+
+1. 建立治理资产。
+2. 建立复核分配。
+3. 导入已存在的决议状态。
+4. 读取证据路径并判定 `template / ready / missing`。
+5. 生成一条发布批次。
+6. 生成一条门禁快照记录。
+
+## 已完成验证
+
+执行测试：
+
+`backend/.venv/Scripts/python.exe -m unittest tests.test_governance_models tests.test_governance_api tests.test_governance_service tests.test_permission_metadata`
+
+结果：
+
+- 13 个测试全部通过。
+
+覆盖内容：
+
+- 治理模型注册与外键。
+- 治理 API 路由与审计事件。
+- formal-launch 导入和发布批次生成。
+- 权限元数据暴露。
+
+## 当前状态
+
+`GOV-API-001 ~ GOV-API-006` 已完成，backlog 已更新为 `COMPLETED`。
+
+尚未进入：
+
+- `GOV-API-007` 模板、规则、知识库的发布/回滚接口。
+- `5C` 前端复核总控中心、复核工作台、资产详情页、上线批准页。
+- `5D` 门禁切换为数据库读模式。
+
+## 下一步建议
+
+最自然的下一步是进入 `5C / GOV-WEB-001 ~ GOV-WEB-004`，先把复核总控中心、复核人工作台、资产详情页和上线批准页做起来，让刚完成的治理接口真正可用。