feat: absorb phase13 templates rules clauses

backend/app/api/standard_clauses.py

@@ -0,0 +1,136 @@
+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.standard_clause_service import StandardClauseService
+
+router = APIRouter(prefix="/standard-clauses", tags=["standard-clauses"])
+ROUTE_PERMISSION_SCOPE = "standard_clause_management"
+ROUTE_PERMISSION_ACTIONS = {
+    "list": "standard_clauses:list",
+    "versions": "standard_clauses:versions",
+    "publish": "standard_clauses:publish",
+    "unpublish": "standard_clauses:unpublish",
+    "rollback": "standard_clauses:rollback",
+}
+
+
+def get_standard_clause_service(db: Session = Depends(get_db_session)) -> StandardClauseService:
+    return StandardClauseService(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("")
+def list_standard_clauses(
+    page: int = 1,
+    page_size: int = 20,
+    contract_type: Optional[str] = None,
+    clause_category: Optional[str] = None,
+    is_active: Optional[bool] = None,
+    standard_clause_service: StandardClauseService = Depends(get_standard_clause_service),
+    current_user: dict = Depends(require_current_user),
+):
+    return success_response(
+        standard_clause_service.list_clauses(
+            current_user,
+            page=page,
+            page_size=page_size,
+            contract_type=contract_type,
+            clause_category=clause_category,
+            is_active=is_active,
+        )
+    )
+
+
+@router.get("/{clause_id}/versions")
+def get_standard_clause_versions(
+    clause_id: str,
+    standard_clause_service: StandardClauseService = Depends(get_standard_clause_service),
+    current_user: dict = Depends(require_current_user),
+):
+    return success_response(standard_clause_service.get_clause_versions(current_user, clause_id))
+
+
+@router.post("/{clause_id}/publish")
+def publish_standard_clause(
+    clause_id: str,
+    standard_clause_service: StandardClauseService = Depends(get_standard_clause_service),
+    audit_log_service: AuditLogService = Depends(get_audit_log_service),
+    current_user: dict = Depends(require_current_user),
+):
+    result = standard_clause_service.publish_clause(current_user, clause_id)
+    audit_log_service.record_event(
+        tenant_id=current_user.get("tenant_id"),
+        user_id=current_user.get("id"),
+        action="publish_standard_clause",
+        resource_type="standard_clause",
+        resource_id=result["id"],
+        detail_json={
+            "clause_code": result["clause_code"],
+            "name": result["name"],
+            "published_at": result.get("published_at"),
+            "published_by": result.get("published_by"),
+            "latest_governance_asset_id": result.get("latest_governance_asset_id"),
+        },
+    )
+    return success_response(result)
+
+
+@router.post("/{clause_id}/unpublish")
+def unpublish_standard_clause(
+    clause_id: str,
+    standard_clause_service: StandardClauseService = Depends(get_standard_clause_service),
+    audit_log_service: AuditLogService = Depends(get_audit_log_service),
+    current_user: dict = Depends(require_current_user),
+):
+    result = standard_clause_service.unpublish_clause(current_user, clause_id)
+    audit_log_service.record_event(
+        tenant_id=current_user.get("tenant_id"),
+        user_id=current_user.get("id"),
+        action="unpublish_standard_clause",
+        resource_type="standard_clause",
+        resource_id=result["id"],
+        detail_json={"clause_code": result["clause_code"], "name": result["name"]},
+    )
+    return success_response(result)
+
+
+@router.post("/{clause_id}/rollback")
+def rollback_standard_clause(
+    clause_id: str,
+    target_version_label: str = Form(...),
+    change_summary: Optional[str] = Form(default=None),
+    standard_clause_service: StandardClauseService = Depends(get_standard_clause_service),
+    audit_log_service: AuditLogService = Depends(get_audit_log_service),
+    current_user: dict = Depends(require_current_user),
+):
+    result = standard_clause_service.rollback_clause(current_user, clause_id, target_version_label, change_summary)
+    audit_log_service.record_event(
+        tenant_id=current_user.get("tenant_id"),
+        user_id=current_user.get("id"),
+        action="rollback_standard_clause",
+        resource_type="standard_clause",
+        resource_id=result["id"],
+        detail_json={
+            "clause_code": result["clause_code"],
+            "name": result["name"],
+            "target_version_label": target_version_label,
+            "current_version_label": result.get("current_version_label"),
+        },
+    )
+    return success_response(result)

backend/app/main.py

@@ -21,6 +21,7 @@ from app.api.prompt_assets import router as prompt_assets_router
 from app.api.review_rules import router as review_rules_router
 from app.api.reviews import router as reviews_router
 from app.api.sales_followups import router as sales_followups_router
+from app.api.standard_clauses import router as standard_clauses_router
 from app.api.system import router as system_router
 from app.api.templates import router as templates_router
 from app.api.tenant_settings import router as tenant_settings_router
@@ -57,6 +58,7 @@ app.include_router(prompt_assets_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)
 app.include_router(sales_followups_router, prefix=settings.api_v1_prefix)
+app.include_router(standard_clauses_router, prefix=settings.api_v1_prefix)
 app.include_router(tenant_settings_router, prefix=settings.api_v1_prefix)
 app.include_router(users_router, prefix=settings.api_v1_prefix)
 app.include_router(audit_logs_router, prefix=settings.api_v1_prefix)

backend/app/models/__init__.py

@@ -35,5 +35,7 @@ from app.models.review_rule import ReviewRule
 from app.models.review_rule_version import ReviewRuleVersion
 from app.models.review_task import ReviewTask
 from app.models.sales_followup_event import SalesFollowupEvent
+from app.models.standard_clause import StandardClause
+from app.models.standard_clause_version import StandardClauseVersion
 from app.models.tenant import Tenant
 from app.models.user import User

backend/app/models/standard_clause.py

@@ -0,0 +1,49 @@
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from sqlalchemy import JSON, Boolean, DateTime, Index, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+
+
+class StandardClause(Base):
+    __tablename__ = "standard_clauses"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "clause_code", name="uq_standard_clauses_tenant_code"),
+        Index("ix_standard_clauses_tenant_category", "tenant_id", "clause_category"),
+        Index("ix_standard_clauses_tenant_contract_type", "tenant_id", "contract_type"),
+        Index("ix_standard_clauses_tenant_active", "tenant_id", "is_active"),
+    )
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    tenant_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    clause_code: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    contract_type: Mapped[Optional[str]] = mapped_column(String(64), nullable=True, index=True)
+    clause_category: Mapped[Optional[str]] = mapped_column(String(100), nullable=True, index=True)
+    variant_group: Mapped[Optional[str]] = mapped_column(String(100), nullable=True, index=True)
+    source_ids_json: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
+    applicable_conditions_json: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
+    excluded_conditions_json: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
+    risk_notes_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    clause_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    governance_status: Mapped[str] = mapped_column(String(32), default="draft")
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    current_version_label: Mapped[str] = mapped_column(String(32), default="v1.0")
+    latest_change_summary: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    latest_version_created_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    version_count: Mapped[int] = mapped_column(default=1)
+    published_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    published_by: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    latest_governance_asset_id: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    metadata_json: Mapped[Optional[dict[str, Any]]] = mapped_column(JSON, nullable=True)
+    created_by: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    updated_by: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+    )

backend/app/models/standard_clause_version.py

@@ -0,0 +1,35 @@
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+
+
+class StandardClauseVersion(Base):
+    __tablename__ = "standard_clause_versions"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    clause_id: Mapped[str] = mapped_column(String(64), ForeignKey("standard_clauses.id"), nullable=False, index=True)
+    tenant_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    version_label: Mapped[str] = mapped_column(String(32), nullable=False)
+    change_summary: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    clause_code: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    contract_type: Mapped[Optional[str]] = mapped_column(String(64), nullable=True, index=True)
+    clause_category: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
+    variant_group: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
+    source_ids_json: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
+    applicable_conditions_json: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
+    excluded_conditions_json: Mapped[Optional[list[str]]] = mapped_column(JSON, nullable=True)
+    risk_notes_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    clause_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    governance_status: Mapped[str] = mapped_column(String(32), default="draft")
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    metadata_json: Mapped[Optional[dict[str, Any]]] = mapped_column(JSON, nullable=True)
+    is_current: Mapped[bool] = mapped_column(Boolean, default=False)
+    created_by: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    created_by_name: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

backend/app/repositories/standard_clause_repository.py

@@ -0,0 +1,78 @@
+from typing import List, Optional
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.models.standard_clause import StandardClause
+
+
+class StandardClauseRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, clause: StandardClause) -> StandardClause:
+        self.db.add(clause)
+        self.db.commit()
+        self.db.refresh(clause)
+        return clause
+
+    def update(self, clause: StandardClause) -> StandardClause:
+        self.db.add(clause)
+        self.db.commit()
+        self.db.refresh(clause)
+        return clause
+
+    def get_by_id(self, clause_id: str, tenant_id: Optional[str] = None) -> Optional[StandardClause]:
+        query = self.db.query(StandardClause).filter(StandardClause.id == clause_id)
+        if tenant_id is not None:
+            query = query.filter(StandardClause.tenant_id == tenant_id)
+        return query.first()
+
+    def get_by_code(self, tenant_id: str, clause_code: str) -> Optional[StandardClause]:
+        return (
+            self.db.query(StandardClause)
+            .filter(
+                StandardClause.tenant_id == tenant_id,
+                func.lower(StandardClause.clause_code) == clause_code.strip().lower(),
+            )
+            .first()
+        )
+
+    def list_paginated(
+        self,
+        tenant_id: str,
+        page: int,
+        page_size: int,
+        contract_type: Optional[str] = None,
+        clause_category: Optional[str] = None,
+        is_active: Optional[bool] = None,
+    ) -> List[StandardClause]:
+        query = self.db.query(StandardClause).filter(StandardClause.tenant_id == tenant_id)
+        if contract_type:
+            query = query.filter(StandardClause.contract_type == contract_type)
+        if clause_category:
+            query = query.filter(StandardClause.clause_category == clause_category)
+        if is_active is not None:
+            query = query.filter(StandardClause.is_active.is_(is_active))
+        return (
+            query.order_by(StandardClause.updated_at.desc(), StandardClause.created_at.desc())
+            .offset((page - 1) * page_size)
+            .limit(page_size)
+            .all()
+        )
+
+    def count(
+        self,
+        tenant_id: str,
+        contract_type: Optional[str] = None,
+        clause_category: Optional[str] = None,
+        is_active: Optional[bool] = None,
+    ) -> int:
+        query = self.db.query(StandardClause).filter(StandardClause.tenant_id == tenant_id)
+        if contract_type:
+            query = query.filter(StandardClause.contract_type == contract_type)
+        if clause_category:
+            query = query.filter(StandardClause.clause_category == clause_category)
+        if is_active is not None:
+            query = query.filter(StandardClause.is_active.is_(is_active))
+        return query.count()

backend/app/repositories/standard_clause_version_repository.py

@@ -0,0 +1,50 @@
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.standard_clause_version import StandardClauseVersion
+
+
+class StandardClauseVersionRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create(self, version: StandardClauseVersion) -> StandardClauseVersion:
+        self.db.add(version)
+        self.db.commit()
+        self.db.refresh(version)
+        return version
+
+    def list_by_clause_id(self, tenant_id: str, clause_id: str) -> List[StandardClauseVersion]:
+        return (
+            self.db.query(StandardClauseVersion)
+            .filter(
+                StandardClauseVersion.tenant_id == tenant_id,
+                StandardClauseVersion.clause_id == clause_id,
+            )
+            .order_by(StandardClauseVersion.created_at.desc())
+            .all()
+        )
+
+    def clear_current(self, tenant_id: str, clause_id: str) -> None:
+        (
+            self.db.query(StandardClauseVersion)
+            .filter(
+                StandardClauseVersion.tenant_id == tenant_id,
+                StandardClauseVersion.clause_id == clause_id,
+                StandardClauseVersion.is_current.is_(True),
+            )
+            .update({"is_current": False}, synchronize_session=False)
+        )
+        self.db.commit()
+
+    def get_by_version_label(self, tenant_id: str, clause_id: str, version_label: str) -> Optional[StandardClauseVersion]:
+        return (
+            self.db.query(StandardClauseVersion)
+            .filter(
+                StandardClauseVersion.tenant_id == tenant_id,
+                StandardClauseVersion.clause_id == clause_id,
+                StandardClauseVersion.version_label == version_label,
+            )
+            .first()
+        )

backend/app/services/standard_clause_service.py

@@ -0,0 +1,292 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+from uuid import uuid4
+
+from fastapi import HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.models.standard_clause import StandardClause
+from app.models.standard_clause_version import StandardClauseVersion
+from app.repositories.governance_asset_repository import GovernanceAssetRepository
+from app.repositories.standard_clause_repository import StandardClauseRepository
+from app.repositories.standard_clause_version_repository import StandardClauseVersionRepository
+
+
+class StandardClauseService:
+    def __init__(self, db: Session):
+        self.db = db
+        self.repository = StandardClauseRepository(db)
+        self.version_repository = StandardClauseVersionRepository(db)
+        self.governance_asset_repository = GovernanceAssetRepository(db)
+
+    def list_clauses(
+        self,
+        current_user: Dict[str, Any],
+        page: int = 1,
+        page_size: int = 20,
+        contract_type: Optional[str] = None,
+        clause_category: Optional[str] = None,
+        is_active: Optional[bool] = None,
+    ) -> dict:
+        self._require_admin(current_user)
+        tenant_id = current_user["tenant_id"]
+        normalized_page = max(page, 1)
+        normalized_page_size = max(min(page_size, 100), 1)
+        normalized_contract_type = contract_type.strip() if contract_type else None
+        normalized_clause_category = clause_category.strip() if clause_category else None
+        items = self.repository.list_paginated(
+            tenant_id,
+            normalized_page,
+            normalized_page_size,
+            contract_type=normalized_contract_type,
+            clause_category=normalized_clause_category,
+            is_active=is_active,
+        )
+        total = self.repository.count(
+            tenant_id,
+            contract_type=normalized_contract_type,
+            clause_category=normalized_clause_category,
+            is_active=is_active,
+        )
+        return {
+            "items": [self._to_item(item) for item in items],
+            "page": normalized_page,
+            "page_size": normalized_page_size,
+            "total": total,
+        }
+
+    def get_clause_versions(self, current_user: Dict[str, Any], clause_id: str) -> dict:
+        self._require_admin(current_user)
+        clause = self._require_clause(current_user, clause_id)
+        versions = self.version_repository.list_by_clause_id(current_user["tenant_id"], clause_id)
+        if not versions:
+            versions = [self._build_fallback_version(clause)]
+        versions.sort(key=lambda item: (item.is_current, item.created_at), reverse=True)
+        return {
+            "clause_id": clause_id,
+            "current_version_label": clause.current_version_label or "v1.0",
+            "version_count": max(clause.version_count or 0, len(versions), 1),
+            "items": [self._to_version_item(version) for version in versions],
+        }
+
+    def publish_clause(self, current_user: Dict[str, Any], clause_id: str) -> dict:
+        self._require_admin(current_user)
+        clause = self._require_clause(current_user, clause_id)
+        governance_asset = self._require_publishable_governance_asset(current_user, clause)
+        clause.is_active = True
+        clause.governance_status = "published"
+        clause.published_at = self._now()
+        clause.published_by = current_user.get("id")
+        clause.latest_governance_asset_id = governance_asset.id
+        clause.updated_by = current_user.get("id")
+        updated = self.repository.update(clause)
+        return self._to_item(updated)
+
+    def unpublish_clause(self, current_user: Dict[str, Any], clause_id: str) -> dict:
+        self._require_admin(current_user)
+        clause = self._require_clause(current_user, clause_id)
+        clause.is_active = False
+        clause.governance_status = "draft"
+        clause.published_at = None
+        clause.published_by = None
+        clause.latest_governance_asset_id = None
+        clause.updated_by = current_user.get("id")
+        updated = self.repository.update(clause)
+        return self._to_item(updated)
+
+    def rollback_clause(
+        self,
+        current_user: Dict[str, Any],
+        clause_id: str,
+        target_version_label: str,
+        change_summary: Optional[str] = None,
+    ) -> dict:
+        self._require_admin(current_user)
+        clause = self._require_clause(current_user, clause_id)
+        target_version = self.version_repository.get_by_version_label(current_user["tenant_id"], clause_id, target_version_label)
+        if target_version is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="standard clause version not found")
+
+        next_version_label = self._bump_version_label(clause.current_version_label)
+        clause.clause_code = target_version.clause_code
+        clause.name = target_version.name
+        clause.contract_type = target_version.contract_type
+        clause.clause_category = target_version.clause_category
+        clause.variant_group = target_version.variant_group
+        clause.source_ids_json = target_version.source_ids_json
+        clause.applicable_conditions_json = target_version.applicable_conditions_json
+        clause.excluded_conditions_json = target_version.excluded_conditions_json
+        clause.risk_notes_path = target_version.risk_notes_path
+        clause.clause_path = target_version.clause_path
+        clause.content = target_version.content
+        clause.governance_status = "draft"
+        clause.is_active = False
+        clause.metadata_json = target_version.metadata_json
+        clause.current_version_label = next_version_label
+        clause.latest_change_summary = (change_summary or f"Rollback to {target_version_label}").strip()
+        clause.latest_version_created_at = self._now()
+        clause.version_count = max(clause.version_count or 0, 1) + 1
+        clause.published_at = None
+        clause.published_by = None
+        clause.latest_governance_asset_id = None
+        clause.updated_by = current_user.get("id")
+        updated = self.repository.update(clause)
+        self._create_version_snapshot(updated, current_user, clause.latest_change_summary, is_current=True)
+        return self._to_item(updated)
+
+    def _require_clause(self, current_user: Dict[str, Any], clause_id: str) -> StandardClause:
+        clause = self.repository.get_by_id(clause_id)
+        if clause is None or clause.tenant_id != current_user["tenant_id"]:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="standard clause not found")
+        return clause
+
+    def _require_admin(self, current_user: Dict[str, Any]) -> None:
+        if current_user.get("role") != "admin":
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="forbidden")
+
+    def _require_publishable_governance_asset(self, current_user: Dict[str, Any], clause: StandardClause):
+        if not clause.clause_code:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="clause_code is required before publish")
+        governance_asset = self.governance_asset_repository.get_by_identity(current_user["tenant_id"], "standard_clause", clause.clause_code)
+        if governance_asset is None:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="governance asset not found for clause_code")
+        if governance_asset.production_status not in {"ready", "approved"}:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="governance asset is not approved for production publish")
+        return governance_asset
+
+    def _create_version_snapshot(
+        self,
+        clause: StandardClause,
+        current_user: Dict[str, Any],
+        change_summary: str,
+        *,
+        is_current: bool,
+    ) -> StandardClauseVersion:
+        self.version_repository.clear_current(clause.tenant_id, clause.id)
+        version = StandardClauseVersion(
+            id=f"clausever-{uuid4().hex[:12]}",
+            clause_id=clause.id,
+            tenant_id=clause.tenant_id,
+            version_label=clause.current_version_label or "v1.0",
+            change_summary=change_summary,
+            clause_code=clause.clause_code,
+            name=clause.name,
+            contract_type=clause.contract_type,
+            clause_category=clause.clause_category,
+            variant_group=clause.variant_group,
+            source_ids_json=clause.source_ids_json,
+            applicable_conditions_json=clause.applicable_conditions_json,
+            excluded_conditions_json=clause.excluded_conditions_json,
+            risk_notes_path=clause.risk_notes_path,
+            clause_path=clause.clause_path,
+            content=clause.content,
+            governance_status=clause.governance_status,
+            is_active=clause.is_active,
+            metadata_json=clause.metadata_json,
+            is_current=is_current,
+            created_by=current_user.get("id"),
+            created_by_name=current_user.get("display_name"),
+            created_at=clause.latest_version_created_at or self._now(),
+        )
+        return self.version_repository.create(version)
+
+    def _build_fallback_version(self, clause: StandardClause) -> StandardClauseVersion:
+        return StandardClauseVersion(
+            id=f"fallback-{clause.id}",
+            clause_id=clause.id,
+            tenant_id=clause.tenant_id,
+            version_label=clause.current_version_label or "v1.0",
+            change_summary=clause.latest_change_summary,
+            clause_code=clause.clause_code,
+            name=clause.name,
+            contract_type=clause.contract_type,
+            clause_category=clause.clause_category,
+            variant_group=clause.variant_group,
+            source_ids_json=clause.source_ids_json,
+            applicable_conditions_json=clause.applicable_conditions_json,
+            excluded_conditions_json=clause.excluded_conditions_json,
+            risk_notes_path=clause.risk_notes_path,
+            clause_path=clause.clause_path,
+            content=clause.content,
+            governance_status=clause.governance_status,
+            is_active=clause.is_active,
+            metadata_json=clause.metadata_json,
+            is_current=True,
+            created_by=clause.updated_by or clause.created_by,
+            created_by_name=None,
+            created_at=clause.latest_version_created_at or clause.updated_at or clause.created_at,
+        )
+
+    @staticmethod
+    def _bump_version_label(current_version_label: Optional[str]) -> str:
+        if not current_version_label:
+            return "v1.1"
+        match = re.match(r"^v?(?P<major>\d+)\.(?P<minor>\d+)$", current_version_label.strip(), re.IGNORECASE)
+        if not match:
+            return "v1.1"
+        return f"v{int(match.group('major'))}.{int(match.group('minor')) + 1}"
+
+    @staticmethod
+    def _now() -> datetime:
+        return datetime.now(timezone.utc)
+
+    @staticmethod
+    def _to_item(clause: StandardClause) -> dict:
+        return {
+            "id": clause.id,
+            "clause_code": clause.clause_code,
+            "name": clause.name,
+            "contract_type": clause.contract_type,
+            "clause_category": clause.clause_category,
+            "variant_group": clause.variant_group,
+            "source_ids": clause.source_ids_json or [],
+            "applicable_conditions": clause.applicable_conditions_json or [],
+            "excluded_conditions": clause.excluded_conditions_json or [],
+            "risk_notes_path": clause.risk_notes_path,
+            "clause_path": clause.clause_path,
+            "content": clause.content,
+            "governance_status": clause.governance_status,
+            "is_active": clause.is_active,
+            "published_at": clause.published_at.isoformat() if clause.published_at else None,
+            "published_by": clause.published_by,
+            "latest_governance_asset_id": clause.latest_governance_asset_id,
+            "metadata": clause.metadata_json or {},
+            "current_version_label": clause.current_version_label or "v1.0",
+            "latest_change_summary": clause.latest_change_summary,
+            "latest_version_created_at": clause.latest_version_created_at.isoformat() if clause.latest_version_created_at else None,
+            "version_count": max(clause.version_count or 0, 1),
+            "created_by": clause.created_by,
+            "updated_by": clause.updated_by,
+            "created_at": clause.created_at.isoformat(),
+            "updated_at": clause.updated_at.isoformat(),
+        }
+
+    @staticmethod
+    def _to_version_item(version: StandardClauseVersion) -> dict:
+        return {
+            "id": version.id,
+            "clause_id": version.clause_id,
+            "version_label": version.version_label,
+            "change_summary": version.change_summary,
+            "clause_code": version.clause_code,
+            "name": version.name,
+            "contract_type": version.contract_type,
+            "clause_category": version.clause_category,
+            "variant_group": version.variant_group,
+            "source_ids": version.source_ids_json or [],
+            "applicable_conditions": version.applicable_conditions_json or [],
+            "excluded_conditions": version.excluded_conditions_json or [],
+            "risk_notes_path": version.risk_notes_path,
+            "clause_path": version.clause_path,
+            "content": version.content,
+            "governance_status": version.governance_status,
+            "is_active": version.is_active,
+            "metadata": version.metadata_json or {},
+            "is_current": version.is_current,
+            "created_by": version.created_by,
+            "created_by_name": version.created_by_name,
+            "created_at": version.created_at.isoformat(),
+        }

backend/migrations/versions/20260522_0040_create_standard_clauses.py

@@ -0,0 +1,164 @@
+"""create standard clause runtime tables"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "20260522_0040"
+down_revision = "20260522_0039"
+branch_labels = None
+depends_on = None
+
+
+def _table_names(inspector: sa.Inspector) -> set[str]:
+    return set(inspector.get_table_names())
+
+
+def _index_names(inspector: sa.Inspector, table_name: str) -> set[str]:
+    return {index["name"] for index in inspector.get_indexes(table_name)}
+
+
+def _constraint_names(inspector: sa.Inspector, table_name: str) -> set[str]:
+    constraints = inspector.get_unique_constraints(table_name)
+    return {item["name"] for item in constraints if item.get("name")}
+
+
+def _create_index_if_missing(
+    inspector: sa.Inspector,
+    index_name: str,
+    table_name: str,
+    columns: list[str],
+) -> None:
+    if index_name not in _index_names(inspector, table_name):
+        op.create_index(index_name, table_name, columns)
+
+
+def _drop_index_if_present(inspector: sa.Inspector, index_name: str, table_name: str) -> None:
+    if index_name in _index_names(inspector, table_name):
+        op.drop_index(index_name, table_name=table_name)
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    tables = _table_names(inspector)
+
+    if "standard_clauses" not in tables:
+        op.create_table(
+            "standard_clauses",
+            sa.Column("id", sa.String(length=64), primary_key=True),
+            sa.Column("tenant_id", sa.String(length=64), nullable=False),
+            sa.Column("clause_code", sa.String(length=64), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("contract_type", sa.String(length=64), nullable=True),
+            sa.Column("clause_category", sa.String(length=100), nullable=True),
+            sa.Column("variant_group", sa.String(length=100), nullable=True),
+            sa.Column("source_ids_json", sa.JSON(), nullable=True),
+            sa.Column("applicable_conditions_json", sa.JSON(), nullable=True),
+            sa.Column("excluded_conditions_json", sa.JSON(), nullable=True),
+            sa.Column("risk_notes_path", sa.String(length=500), nullable=True),
+            sa.Column("clause_path", sa.String(length=500), nullable=True),
+            sa.Column("content", sa.Text(), nullable=False),
+            sa.Column("governance_status", sa.String(length=32), nullable=False, server_default="draft"),
+            sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.true()),
+            sa.Column("current_version_label", sa.String(length=32), nullable=False, server_default="v1.0"),
+            sa.Column("latest_change_summary", sa.String(length=500), nullable=True),
+            sa.Column("latest_version_created_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("version_count", sa.Integer(), nullable=False, server_default="1"),
+            sa.Column("published_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("published_by", sa.String(length=64), nullable=True),
+            sa.Column("latest_governance_asset_id", sa.String(length=64), nullable=True),
+            sa.Column("metadata_json", sa.JSON(), nullable=True),
+            sa.Column("created_by", sa.String(length=64), nullable=True),
+            sa.Column("updated_by", sa.String(length=64), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.UniqueConstraint("tenant_id", "clause_code", name="uq_standard_clauses_tenant_code"),
+        )
+        inspector = sa.inspect(bind)
+    elif "uq_standard_clauses_tenant_code" not in _constraint_names(inspector, "standard_clauses"):
+        op.create_unique_constraint(
+            "uq_standard_clauses_tenant_code",
+            "standard_clauses",
+            ["tenant_id", "clause_code"],
+        )
+        inspector = sa.inspect(bind)
+
+    _create_index_if_missing(inspector, "ix_standard_clauses_tenant_id", "standard_clauses", ["tenant_id"])
+    _create_index_if_missing(inspector, "ix_standard_clauses_clause_code", "standard_clauses", ["clause_code"])
+    _create_index_if_missing(inspector, "ix_standard_clauses_contract_type", "standard_clauses", ["contract_type"])
+    _create_index_if_missing(inspector, "ix_standard_clauses_clause_category", "standard_clauses", ["clause_category"])
+    _create_index_if_missing(inspector, "ix_standard_clauses_variant_group", "standard_clauses", ["variant_group"])
+    _create_index_if_missing(inspector, "ix_standard_clauses_tenant_category", "standard_clauses", ["tenant_id", "clause_category"])
+    _create_index_if_missing(inspector, "ix_standard_clauses_tenant_contract_type", "standard_clauses", ["tenant_id", "contract_type"])
+    _create_index_if_missing(inspector, "ix_standard_clauses_tenant_active", "standard_clauses", ["tenant_id", "is_active"])
+
+    inspector = sa.inspect(bind)
+    tables = _table_names(inspector)
+    if "standard_clause_versions" not in tables:
+        op.create_table(
+            "standard_clause_versions",
+            sa.Column("id", sa.String(length=64), primary_key=True),
+            sa.Column("clause_id", sa.String(length=64), sa.ForeignKey("standard_clauses.id"), nullable=False),
+            sa.Column("tenant_id", sa.String(length=64), nullable=False),
+            sa.Column("version_label", sa.String(length=32), nullable=False),
+            sa.Column("change_summary", sa.String(length=500), nullable=True),
+            sa.Column("clause_code", sa.String(length=64), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("contract_type", sa.String(length=64), nullable=True),
+            sa.Column("clause_category", sa.String(length=100), nullable=True),
+            sa.Column("variant_group", sa.String(length=100), nullable=True),
+            sa.Column("source_ids_json", sa.JSON(), nullable=True),
+            sa.Column("applicable_conditions_json", sa.JSON(), nullable=True),
+            sa.Column("excluded_conditions_json", sa.JSON(), nullable=True),
+            sa.Column("risk_notes_path", sa.String(length=500), nullable=True),
+            sa.Column("clause_path", sa.String(length=500), nullable=True),
+            sa.Column("content", sa.Text(), nullable=False),
+            sa.Column("governance_status", sa.String(length=32), nullable=False, server_default="draft"),
+            sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.true()),
+            sa.Column("metadata_json", sa.JSON(), nullable=True),
+            sa.Column("is_current", sa.Boolean(), nullable=False, server_default=sa.false()),
+            sa.Column("created_by", sa.String(length=64), nullable=True),
+            sa.Column("created_by_name", sa.String(length=100), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        )
+        inspector = sa.inspect(bind)
+
+    _create_index_if_missing(inspector, "ix_standard_clause_versions_clause_id", "standard_clause_versions", ["clause_id"])
+    _create_index_if_missing(inspector, "ix_standard_clause_versions_tenant_id", "standard_clause_versions", ["tenant_id"])
+    _create_index_if_missing(inspector, "ix_standard_clause_versions_clause_code", "standard_clause_versions", ["clause_code"])
+    _create_index_if_missing(inspector, "ix_standard_clause_versions_contract_type", "standard_clause_versions", ["contract_type"])
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    tables = _table_names(inspector)
+
+    if "standard_clause_versions" in tables:
+        for index_name in [
+            "ix_standard_clause_versions_contract_type",
+            "ix_standard_clause_versions_clause_code",
+            "ix_standard_clause_versions_tenant_id",
+            "ix_standard_clause_versions_clause_id",
+        ]:
+            _drop_index_if_present(inspector, index_name, "standard_clause_versions")
+        op.drop_table("standard_clause_versions")
+        inspector = sa.inspect(bind)
+        tables = _table_names(inspector)
+
+    if "standard_clauses" in tables:
+        for index_name in [
+            "ix_standard_clauses_tenant_active",
+            "ix_standard_clauses_tenant_contract_type",
+            "ix_standard_clauses_tenant_category",
+            "ix_standard_clauses_variant_group",
+            "ix_standard_clauses_clause_category",
+            "ix_standard_clauses_contract_type",
+            "ix_standard_clauses_clause_code",
+            "ix_standard_clauses_tenant_id",
+        ]:
+            _drop_index_if_present(inspector, index_name, "standard_clauses")
+        op.drop_table("standard_clauses")

backend/tests/test_permission_metadata.py

@@ -1,6 +1,6 @@
 import unittest
 
-from app.api import audit_logs, auth, chat, clauses, contracts, conversations, files, governance, knowledge_bases, legal_articles, legal_change_impacts, legal_document_drafts, legal_document_types, operation_kpis, prompt_assets, review_rules, reviews, sales_followups, system, templates, tenant_settings, users
+from app.api import audit_logs, auth, chat, clauses, contracts, conversations, files, governance, knowledge_bases, legal_articles, legal_change_impacts, legal_document_drafts, legal_document_types, operation_kpis, prompt_assets, review_rules, reviews, sales_followups, standard_clauses, system, templates, tenant_settings, users
 
 
 class PermissionMetadataTestCase(unittest.TestCase):
@@ -24,6 +24,7 @@ class PermissionMetadataTestCase(unittest.TestCase):
             review_rules,
             reviews,
             sales_followups,
+            standard_clauses,
             system,
             templates,
             tenant_settings,
@@ -56,6 +57,7 @@ class PermissionMetadataTestCase(unittest.TestCase):
             review_rules,
             reviews,
             sales_followups,
+            standard_clauses,
             system,
             templates,
             tenant_settings,

backend/tests/test_standard_clause_service.py

@@ -0,0 +1,270 @@
+import unittest
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+from fastapi import HTTPException
+
+from app.services.standard_clause_service import StandardClauseService
+
+
+def create_clause(clause_id="clause-123", tenant_id="tenant-demo-001", **kwargs):
+    now = datetime.now(timezone.utc)
+    defaults = {
+        "id": clause_id,
+        "tenant_id": tenant_id,
+        "clause_code": "CLAUSE-PAYMENT-TERM-001",
+        "name": "Payment Term",
+        "contract_type": "purchase",
+        "clause_category": "payment",
+        "variant_group": "standard",
+        "source_ids_json": ["SRC-CIVIL-CODE-001"],
+        "applicable_conditions_json": ["milestone_payment"],
+        "excluded_conditions_json": [],
+        "risk_notes_path": "phase13/risk-note.md",
+        "clause_path": "phase13/clause.md",
+        "content": "The buyer shall pay by milestone.",
+        "governance_status": "draft",
+        "is_active": False,
+        "current_version_label": "v1.0",
+        "latest_change_summary": "Initial version",
+        "latest_version_created_at": now,
+        "version_count": 1,
+        "published_at": None,
+        "published_by": None,
+        "latest_governance_asset_id": None,
+        "metadata_json": {"phase": "Phase13"},
+        "created_by": "user-admin",
+        "updated_by": "user-admin",
+        "created_at": now,
+        "updated_at": now,
+    }
+    defaults.update(kwargs)
+    return SimpleNamespace(**defaults)
+
+
+def create_version(version_id="clausever-001", clause_id="clause-123", **kwargs):
+    now = datetime.now(timezone.utc)
+    defaults = {
+        "id": version_id,
+        "clause_id": clause_id,
+        "tenant_id": "tenant-demo-001",
+        "version_label": "v1.0",
+        "change_summary": "Initial version",
+        "clause_code": "CLAUSE-PAYMENT-TERM-001",
+        "name": "Payment Term",
+        "contract_type": "purchase",
+        "clause_category": "payment",
+        "variant_group": "standard",
+        "source_ids_json": ["SRC-CIVIL-CODE-001"],
+        "applicable_conditions_json": ["milestone_payment"],
+        "excluded_conditions_json": [],
+        "risk_notes_path": "phase13/risk-note.md",
+        "clause_path": "phase13/clause.md",
+        "content": "The buyer shall pay by milestone.",
+        "governance_status": "draft",
+        "is_active": False,
+        "metadata_json": {"phase": "Phase13"},
+        "is_current": True,
+        "created_by": "user-admin",
+        "created_by_name": "Admin",
+        "created_at": now,
+    }
+    defaults.update(kwargs)
+    return SimpleNamespace(**defaults)
+
+
+class StubClauseRepository:
+    def __init__(self):
+        self.items = {}
+
+    def update(self, clause):
+        clause.updated_at = datetime.now(timezone.utc)
+        self.items[clause.id] = clause
+        return clause
+
+    def get_by_id(self, clause_id):
+        return self.items.get(clause_id)
+
+    def list_paginated(self, tenant_id, page, page_size, contract_type=None, clause_category=None, is_active=None):
+        items = [item for item in self.items.values() if item.tenant_id == tenant_id]
+        if contract_type:
+            items = [item for item in items if item.contract_type == contract_type]
+        if clause_category:
+            items = [item for item in items if item.clause_category == clause_category]
+        if is_active is not None:
+            items = [item for item in items if item.is_active == is_active]
+        items.sort(key=lambda item: (item.updated_at, item.created_at), reverse=True)
+        start = (page - 1) * page_size
+        return items[start : start + page_size]
+
+    def count(self, tenant_id, contract_type=None, clause_category=None, is_active=None):
+        return len(self.list_paginated(tenant_id, 1, 10000, contract_type, clause_category, is_active))
+
+
+class StubVersionRepository:
+    def __init__(self):
+        self.items = []
+
+    def list_by_clause_id(self, tenant_id, clause_id):
+        items = [
+            item
+            for item in self.items
+            if item.tenant_id == tenant_id and item.clause_id == clause_id
+        ]
+        items.sort(key=lambda item: item.created_at, reverse=True)
+        return items
+
+    def get_by_version_label(self, tenant_id, clause_id, version_label):
+        for item in self.items:
+            if item.tenant_id == tenant_id and item.clause_id == clause_id and item.version_label == version_label:
+                return item
+        return None
+
+    def clear_current(self, tenant_id, clause_id):
+        for item in self.items:
+            if item.tenant_id == tenant_id and item.clause_id == clause_id and item.is_current:
+                item.is_current = False
+
+    def create(self, version):
+        self.items.append(version)
+        return version
+
+
+class StubGovernanceAssetRepository:
+    def __init__(self):
+        self.items = {}
+
+    def add(self, tenant_id, asset_type, asset_id, production_status="ready"):
+        self.items[(tenant_id, asset_type, asset_id)] = SimpleNamespace(
+            id=f"govast-{asset_id.lower()}",
+            production_status=production_status,
+        )
+
+    def get_by_identity(self, tenant_id, asset_type, source_asset_id):
+        return self.items.get((tenant_id, asset_type, source_asset_id))
+
+
+class StandardClauseServiceTestCase(unittest.TestCase):
+    def setUp(self):
+        self.service = StandardClauseService(db=None)
+        self.service.repository = StubClauseRepository()
+        self.service.version_repository = StubVersionRepository()
+        self.service.governance_asset_repository = StubGovernanceAssetRepository()
+        self.current_user = {
+            "id": "user-admin",
+            "tenant_id": "tenant-demo-001",
+            "role": "admin",
+            "display_name": "Admin",
+        }
+        self.member_user = {
+            "id": "user-member",
+            "tenant_id": "tenant-demo-001",
+            "role": "member",
+        }
+        self.other_tenant_user = {
+            "id": "user-other",
+            "tenant_id": "tenant-other",
+            "role": "admin",
+        }
+        self.clause = create_clause()
+        self.service.repository.items[self.clause.id] = self.clause
+
+    def test_admin_can_list_clauses_with_filters(self):
+        result = self.service.list_clauses(
+            self.current_user,
+            page=0,
+            page_size=999,
+            contract_type=" purchase ",
+            clause_category=" payment ",
+            is_active=False,
+        )
+
+        self.assertEqual(result["page"], 1)
+        self.assertEqual(result["page_size"], 100)
+        self.assertEqual(result["total"], 1)
+        self.assertEqual(result["items"][0]["clause_code"], "CLAUSE-PAYMENT-TERM-001")
+
+    def test_member_cannot_access_clause_management(self):
+        with self.assertRaises(HTTPException) as context:
+            self.service.list_clauses(self.member_user)
+
+        self.assertEqual(context.exception.status_code, 403)
+        self.assertEqual(context.exception.detail, "forbidden")
+
+    def test_get_clause_versions_uses_fallback_when_version_table_empty(self):
+        result = self.service.get_clause_versions(self.current_user, self.clause.id)
+
+        self.assertEqual(result["clause_id"], self.clause.id)
+        self.assertEqual(result["version_count"], 1)
+        self.assertEqual(len(result["items"]), 1)
+        self.assertEqual(result["items"][0]["version_label"], "v1.0")
+
+    def test_publish_and_unpublish_clause_with_ready_governance_asset(self):
+        self.service.governance_asset_repository.add(
+            "tenant-demo-001",
+            "standard_clause",
+            "CLAUSE-PAYMENT-TERM-001",
+            production_status="approved",
+        )
+
+        published = self.service.publish_clause(self.current_user, self.clause.id)
+        self.assertEqual(published["published_by"], "user-admin")
+        self.assertIsNotNone(published["published_at"])
+        self.assertEqual(published["latest_governance_asset_id"], "govast-clause-payment-term-001")
+        self.assertTrue(published["is_active"])
+        self.assertEqual(published["governance_status"], "published")
+
+        unpublished = self.service.unpublish_clause(self.current_user, self.clause.id)
+        self.assertIsNone(unpublished["published_at"])
+        self.assertIsNone(unpublished["published_by"])
+        self.assertIsNone(unpublished["latest_governance_asset_id"])
+        self.assertFalse(unpublished["is_active"])
+        self.assertEqual(unpublished["governance_status"], "draft")
+
+    def test_publish_clause_requires_ready_governance_asset(self):
+        with self.assertRaises(HTTPException) as context:
+            self.service.publish_clause(self.current_user, self.clause.id)
+        self.assertEqual(context.exception.status_code, 409)
+
+    def test_rollback_clause_creates_new_version_and_unpublishes(self):
+        self.clause.current_version_label = "v1.2"
+        self.clause.version_count = 2
+        self.clause.is_active = True
+        self.clause.published_at = datetime.now(timezone.utc)
+        self.clause.published_by = "user-admin"
+        self.clause.latest_governance_asset_id = "govast-clause-payment-term-001"
+        self.service.version_repository.items.append(
+            create_version(
+                version_id="clausever-v11",
+                clause_id=self.clause.id,
+                version_label="v1.1",
+                content="Rollback target content",
+                governance_status="published",
+                is_active=True,
+            )
+        )
+
+        rolled_back = self.service.rollback_clause(self.current_user, self.clause.id, "v1.1")
+
+        self.assertEqual(rolled_back["current_version_label"], "v1.3")
+        self.assertEqual(rolled_back["content"], "Rollback target content")
+        self.assertFalse(rolled_back["is_active"])
+        self.assertEqual(rolled_back["governance_status"], "draft")
+        self.assertIsNone(rolled_back["published_at"])
+        self.assertEqual(rolled_back["version_count"], 3)
+        self.assertTrue(any(item.version_label == "v1.3" and item.is_current for item in self.service.version_repository.items))
+
+    def test_rollback_clause_raises_when_target_version_missing(self):
+        with self.assertRaises(HTTPException) as context:
+            self.service.rollback_clause(self.current_user, self.clause.id, "v9.9")
+        self.assertEqual(context.exception.status_code, 404)
+
+    def test_tenant_isolation_for_clause_access(self):
+        with self.assertRaises(HTTPException) as context:
+            self.service.get_clause_versions(self.other_tenant_user, self.clause.id)
+        self.assertEqual(context.exception.status_code, 404)
+        self.assertEqual(context.exception.detail, "standard clause not found")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_standard_clauses_api.py

@@ -0,0 +1,210 @@
+import unittest
+
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+from app.api.standard_clauses import get_audit_log_service, get_standard_clause_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 StubStandardClauseService:
+    def list_clauses(self, current_user, page, page_size, contract_type, clause_category, is_active):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "items": [
+                {
+                    "id": "clause-123",
+                    "clause_code": "CLAUSE-PAYMENT-TERM-001",
+                    "name": "Payment Term",
+                    "contract_type": contract_type or "purchase",
+                    "clause_category": clause_category or "payment",
+                    "variant_group": "standard",
+                    "source_ids": ["SRC-CIVIL-CODE-001"],
+                    "applicable_conditions": ["milestone_payment"],
+                    "excluded_conditions": [],
+                    "risk_notes_path": "phase13/risk-note.md",
+                    "clause_path": "phase13/clause.md",
+                    "content": "The buyer shall pay by milestone.",
+                    "governance_status": "published",
+                    "is_active": True if is_active is None else is_active,
+                    "published_at": "2026-05-12T00:00:00+00:00",
+                    "published_by": "user-admin",
+                    "latest_governance_asset_id": "govast-clause-001",
+                    "metadata": {"phase": "Phase13"},
+                    "current_version_label": "v1.2",
+                    "latest_change_summary": "Improve payment schedule",
+                    "latest_version_created_at": "2026-05-12T00:00:00+00:00",
+                    "version_count": 3,
+                    "created_by": "user-admin",
+                    "updated_by": "user-admin",
+                    "created_at": "2026-05-10T00:00:00+00:00",
+                    "updated_at": "2026-05-12T00:00:00+00:00",
+                }
+            ],
+            "page": page,
+            "page_size": page_size,
+            "total": 1,
+        }
+
+    def get_clause_versions(self, current_user, clause_id):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "clause_id": clause_id,
+            "current_version_label": "v1.2",
+            "version_count": 3,
+            "items": [
+                {
+                    "id": "clausever-003",
+                    "clause_id": clause_id,
+                    "version_label": "v1.2",
+                    "change_summary": "Improve payment schedule",
+                    "clause_code": "CLAUSE-PAYMENT-TERM-001",
+                    "name": "Payment Term",
+                    "contract_type": "purchase",
+                    "clause_category": "payment",
+                    "variant_group": "standard",
+                    "source_ids": ["SRC-CIVIL-CODE-001"],
+                    "applicable_conditions": ["milestone_payment"],
+                    "excluded_conditions": [],
+                    "risk_notes_path": "phase13/risk-note.md",
+                    "clause_path": "phase13/clause.md",
+                    "content": "The buyer shall pay by milestone.",
+                    "governance_status": "published",
+                    "is_active": True,
+                    "metadata": {"phase": "Phase13"},
+                    "is_current": True,
+                    "created_by": "user-admin",
+                    "created_by_name": "Admin",
+                    "created_at": "2026-05-12T00:00:00+00:00",
+                }
+            ],
+        }
+
+    def publish_clause(self, current_user, clause_id):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "id": clause_id,
+            "clause_code": "CLAUSE-PAYMENT-TERM-001",
+            "name": "Payment Term",
+            "published_at": "2026-05-12T00:00:00+00:00",
+            "published_by": current_user["id"],
+            "latest_governance_asset_id": "govast-clause-001",
+            "current_version_label": "v1.2",
+        }
+
+    def unpublish_clause(self, current_user, clause_id):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "id": clause_id,
+            "clause_code": "CLAUSE-PAYMENT-TERM-001",
+            "name": "Payment Term",
+            "published_at": None,
+            "published_by": None,
+            "latest_governance_asset_id": None,
+            "current_version_label": "v1.2",
+        }
+
+    def rollback_clause(self, current_user, clause_id, target_version_label, change_summary=None):
+        if current_user["role"] != "admin":
+            raise HTTPException(status_code=403, detail="forbidden")
+        return {
+            "id": clause_id,
+            "clause_code": "CLAUSE-PAYMENT-TERM-001",
+            "name": "Payment Term",
+            "published_at": None,
+            "published_by": None,
+            "latest_governance_asset_id": None,
+            "current_version_label": "v1.3",
+            "latest_change_summary": change_summary or f"Rollback to {target_version_label}",
+        }
+
+
+class StandardClausesApiTestCase(unittest.TestCase):
+    def setUp(self):
+        self.audit_log_service = StubAuditLogService()
+        app.dependency_overrides[get_standard_clause_service] = lambda: StubStandardClauseService()
+        app.dependency_overrides[get_audit_log_service] = lambda: self.audit_log_service
+        self.client = TestClient(app)
+
+    def tearDown(self):
+        app.dependency_overrides.clear()
+
+    def test_admin_can_list_standard_clauses(self):
+        app.dependency_overrides[require_current_user] = lambda: {
+            "id": "user-admin",
+            "tenant_id": "tenant-demo-001",
+            "role": "admin",
+        }
+        response = self.client.get("/api/v1/standard-clauses", params={"contract_type": "purchase"})
+
+        self.assertEqual(response.status_code, 200)
+        payload = response.json()["data"]["items"][0]
+        self.assertEqual(payload["clause_code"], "CLAUSE-PAYMENT-TERM-001")
+        self.assertEqual(payload["contract_type"], "purchase")
+        self.assertEqual(payload["current_version_label"], "v1.2")
+
+    def test_admin_can_list_clause_versions(self):
+        app.dependency_overrides[require_current_user] = lambda: {
+            "id": "user-admin",
+            "tenant_id": "tenant-demo-001",
+            "role": "admin",
+        }
+        response = self.client.get("/api/v1/standard-clauses/clause-123/versions")
+
+        self.assertEqual(response.status_code, 200)
+        body = response.json()["data"]
+        self.assertEqual(body["current_version_label"], "v1.2")
+        self.assertEqual(body["version_count"], 3)
+        self.assertEqual(body["items"][0]["version_label"], "v1.2")
+
+    def test_admin_can_publish_unpublish_and_rollback_clause(self):
+        app.dependency_overrides[require_current_user] = lambda: {
+            "id": "user-admin",
+            "tenant_id": "tenant-demo-001",
+            "role": "admin",
+        }
+        publish_response = self.client.post("/api/v1/standard-clauses/clause-123/publish")
+        unpublish_response = self.client.post("/api/v1/standard-clauses/clause-123/unpublish")
+        rollback_response = self.client.post(
+            "/api/v1/standard-clauses/clause-123/rollback",
+            data={"target_version_label": "v1.1"},
+        )
+
+        self.assertEqual(publish_response.status_code, 200)
+        self.assertEqual(unpublish_response.status_code, 200)
+        self.assertEqual(rollback_response.status_code, 200)
+        self.assertEqual(publish_response.json()["data"]["published_by"], "user-admin")
+        self.assertEqual(unpublish_response.json()["data"]["published_at"], None)
+        self.assertEqual(rollback_response.json()["data"]["current_version_label"], "v1.3")
+        self.assertEqual(self.audit_log_service.events[-3]["action"], "publish_standard_clause")
+        self.assertEqual(self.audit_log_service.events[-2]["action"], "unpublish_standard_clause")
+        self.assertEqual(self.audit_log_service.events[-1]["action"], "rollback_standard_clause")
+
+    def test_member_cannot_access_standard_clauses(self):
+        app.dependency_overrides[require_current_user] = lambda: {
+            "id": "user-member",
+            "tenant_id": "tenant-demo-001",
+            "role": "member",
+        }
+        response = self.client.get("/api/v1/standard-clauses")
+
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.json()["message"], "forbidden")
+        self.assertEqual(self.audit_log_service.events, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tools/asset_inventory_gap_report.py

@@ -300,6 +300,10 @@ def build_gap_summary(
         "rules_and_clauses": {
             "local_clause_inventory_count": local_assets["inventories"]["clause_library"]["count"],
             "local_review_rule_inventory_count": local_assets["inventories"]["review_rules"]["count"],
+            "remote_standard_clauses": int(remote_db.get("standard_clauses") or 0),
+            "remote_standard_clause_versions": int(remote_db.get("standard_clause_versions") or 0),
+            "remote_standard_clauses_published": int(remote_db.get("standard_clauses_published") or 0),
+            "remote_standard_clauses_active": int(remote_db.get("standard_clauses_active") or 0),
             "remote_review_rules": int(remote_db.get("review_rules") or 0),
             "remote_review_rule_versions": int(remote_db.get("review_rule_versions") or 0),
             "remote_review_rules_published": int(remote_db.get("review_rules_published") or 0),
@@ -385,6 +389,7 @@ def write_markdown_report(path: Path, summary: dict[str, Any]) -> None:
         f"本地 Phase13 模板清单有 {templates['local_inventory_count']} 个。",
         prompt_conclusion,
         f"- 远程审查规则为 {rules['remote_review_rules']}，本地 Phase13 规则清单有 {rules['local_review_rule_inventory_count']} 条。",
+        f"- 远程标准条款为 {rules['remote_standard_clauses']} 条，已发布 {rules['remote_standard_clauses_published']} 条，本地 Phase13 标准条款清单有 {rules['local_clause_inventory_count']} 条。",
         "",
         "## 二、法律库盘点",
         "",
@@ -425,7 +430,9 @@ def write_markdown_report(path: Path, summary: dict[str, Any]) -> None:
         markdown_table(
             [
                 ["项目", "本地清单", "远程运行表"],
-                ["标准条款", rules["local_clause_inventory_count"], "当前没有独立标准条款运行表，主要在文档/模板/推荐规则中使用"],
+                ["标准条款", rules["local_clause_inventory_count"], rules["remote_standard_clauses"]],
+                ["标准条款版本", "-", rules["remote_standard_clause_versions"]],
+                ["已发布标准条款", "-", rules["remote_standard_clauses_published"]],
                 ["审查规则", rules["local_review_rule_inventory_count"], rules["remote_review_rules"]],
                 ["审查规则版本", "-", rules["remote_review_rule_versions"]],
                 ["已发布审查规则", "-", rules["remote_review_rules_published"]],

backend/tools/seed_phase13_template_rule_clause_assets.py

@@ -0,0 +1,987 @@
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import os
+import re
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+TEMPLATE_DOCUMENT_TYPE_CODES = {
+    "TPL-PURCHASE-GOODS-001": "procurement_goods",
+    "TPL-SERVICE-GENERAL-001": "service_general",
+    "TPL-SOFTWARE-SAAS-001": "software_saas",
+    "TPL-TECH-DEVELOPMENT-001": "software_development",
+    "TPL-NDA-COMMERCIAL-001": "nda_commercial",
+    "TPL-SALES-GOODS-001": "sales_goods",
+    "TPL-LEASE-HOUSE-001": "lease_house",
+    "TPL-COMMISSION-SERVICE-001": "commission_service",
+    "TPL-COOPERATION-FRAMEWORK-001": "cooperation_framework",
+    "TPL-LABOR-CONTRACT-001": "labor_contract",
+}
+
+TEMPLATE_CATEGORIES = {
+    "purchase": "procurement-supply",
+    "service": "service-delivery",
+    "software-saas": "software-service",
+    "technology-development": "technology-service",
+    "nda": "commercial-compliance",
+    "sales": "sales-commercial",
+    "lease": "asset-lease",
+    "commission": "service-delivery",
+    "cooperation": "commercial-cooperation",
+    "labor": "employment-hr",
+}
+
+SPECIALIZED_RULE_CONTRACT_TYPES = {
+    "RULE-DATA-PI-PROCESSING-001": "software-saas",
+    "RULE-IP-SOFTWARE-OWNERSHIP-001": "technology-development",
+}
+
+
+@dataclass
+class Phase13Counters:
+    created: int = 0
+    updated: int = 0
+    reused: int = 0
+    version_backfilled: int = 0
+    published: int = 0
+    failed: int = 0
+
+
+@dataclass
+class Phase13Results:
+    templates: list[dict[str, Any]] = field(default_factory=list)
+    review_rules: list[dict[str, Any]] = field(default_factory=list)
+    standard_clauses: list[dict[str, Any]] = field(default_factory=list)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Seed and publish Phase13 templates, review rules, and standard clauses.")
+    parser.add_argument("--database-url", default=None)
+    parser.add_argument("--bootstrap-mode", choices=["none", "sqlite_phase13_assets"], default="none")
+    parser.add_argument("--repo-root", type=Path, default=None)
+    parser.add_argument("--library-root", type=Path, default=None)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--tenant-id", default="tenant-demo-001")
+    parser.add_argument("--user-id", default="codex-autonomous-approval")
+    parser.add_argument("--username", default="codex-autonomous@local")
+    parser.add_argument("--display-name", default="CodexAutonomousApprovalMode")
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--no-publish", action="store_true")
+    return parser.parse_args()
+
+
+def configure_env(database_url: str | None) -> None:
+    if database_url:
+        os.environ["DATABASE_URL"] = database_url
+
+
+def resolve_repo_root(args: argparse.Namespace) -> Path:
+    if args.repo_root:
+        return args.repo_root.resolve()
+    return Path(__file__).resolve().parents[2]
+
+
+def resolve_library_root(args: argparse.Namespace, repo_root: Path) -> Path:
+    return (args.library_root or repo_root / "project-docs/phase13-legal-knowledge-template-library").resolve()
+
+
+def add_import_roots(repo_root: Path) -> None:
+    for candidate in [Path(__file__).resolve().parents[1], repo_root / "backend", repo_root]:
+        if (candidate / "app").exists() and str(candidate) not in sys.path:
+            sys.path.insert(0, str(candidate))
+
+
+def bootstrap_sqlite_phase13_assets() -> None:
+    from sqlalchemy import create_engine
+
+    from app.db import Base
+    from app.models.contract_template import ContractTemplate
+    from app.models.contract_template_version import ContractTemplateVersion
+    from app.models.governance_asset import GovernanceAsset
+    from app.models.review_rule import ReviewRule
+    from app.models.review_rule_version import ReviewRuleVersion
+    from app.models.standard_clause import StandardClause
+    from app.models.standard_clause_version import StandardClauseVersion
+
+    engine = create_engine(os.environ["DATABASE_URL"], future=True)
+    try:
+        Base.metadata.create_all(
+            engine,
+            tables=[
+                GovernanceAsset.__table__,
+                ContractTemplate.__table__,
+                ContractTemplateVersion.__table__,
+                ReviewRule.__table__,
+                ReviewRuleVersion.__table__,
+                StandardClause.__table__,
+                StandardClauseVersion.__table__,
+            ],
+        )
+    finally:
+        engine.dispose()
+
+
+def read_csv_rows(path: Path) -> list[dict[str, str]]:
+    with path.open("r", encoding="utf-8-sig", newline="") as file_obj:
+        return list(csv.DictReader(file_obj))
+
+
+def read_text(path: Path) -> str:
+    return path.read_text(encoding="utf-8-sig")
+
+
+def read_json(path: Path) -> dict[str, Any]:
+    return json.loads(path.read_text(encoding="utf-8-sig"))
+
+
+def split_multi(value: str | None) -> list[str]:
+    if not value:
+        return []
+    parts = re.split(r"[;；]", str(value))
+    return [part.strip() for part in parts if part.strip()]
+
+
+def stable_id(prefix: str, key: str) -> str:
+    digest = hashlib.sha1(key.encode("utf-8")).hexdigest()[:12]
+    slug = re.sub(r"[^A-Za-z0-9]+", "-", key.upper()).strip("-")[:36]
+    return f"{prefix}-{slug.lower()}-{digest}"[:64]
+
+
+def json_same(left: Any, right: Any) -> bool:
+    return json.dumps(left, ensure_ascii=False, sort_keys=True, default=str) == json.dumps(right, ensure_ascii=False, sort_keys=True, default=str)
+
+
+def extract_template_body(markdown: str) -> str:
+    lines = markdown.splitlines()
+    start_index = None
+    for index, line in enumerate(lines):
+        normalized = line.strip()
+        if normalized.startswith("## ") and ("合同正文" in normalized or "姝ｆ枃" in normalized):
+            start_index = index + 1
+            break
+    if start_index is None:
+        return markdown.strip()
+    end_index = len(lines)
+    for index in range(start_index, len(lines)):
+        normalized = lines[index].strip()
+        if normalized.startswith("## ") and ("使用边界" in normalized or "浣跨敤" in normalized):
+            end_index = index
+            break
+    return "\n".join(lines[start_index:end_index]).strip() or markdown.strip()
+
+
+def extract_clause_text(markdown: str) -> str:
+    lines = markdown.splitlines()
+    start_index = None
+    for index, line in enumerate(lines):
+        normalized = line.strip()
+        if normalized.startswith("## ") and ("推荐条款" in normalized or "鎺ㄨ崘鏉℃" in normalized):
+            start_index = index + 1
+            break
+    if start_index is None:
+        return markdown.strip()
+    end_index = len(lines)
+    for index in range(start_index, len(lines)):
+        if lines[index].strip().startswith("## "):
+            end_index = index
+            break
+    return "\n".join(lines[start_index:end_index]).strip() or markdown.strip()
+
+
+def extract_markdown_section(markdown: str, terms: list[str]) -> str | None:
+    lines = markdown.splitlines()
+    start_index = None
+    for index, line in enumerate(lines):
+        normalized = line.strip()
+        if normalized.startswith("## ") and any(term in normalized for term in terms):
+            start_index = index + 1
+            break
+    if start_index is None:
+        return None
+    end_index = len(lines)
+    for index in range(start_index, len(lines)):
+        if lines[index].strip().startswith("## "):
+            end_index = index
+            break
+    text = "\n".join(lines[start_index:end_index]).strip()
+    return text or None
+
+
+def bump_version_label(current_version_label: str | None) -> str:
+    if not current_version_label:
+        return "v1.1"
+    match = re.match(r"^v?(?P<major>\d+)\.(?P<minor>\d+)$", current_version_label.strip(), re.IGNORECASE)
+    if not match:
+        return "v1.1"
+    return f"v{int(match.group('major'))}.{int(match.group('minor')) + 1}"
+
+
+def current_user(args: argparse.Namespace) -> dict[str, Any]:
+    return {
+        "id": args.user_id,
+        "tenant_id": args.tenant_id,
+        "role": "admin",
+        "username": args.username,
+        "display_name": args.display_name,
+    }
+
+
+def ensure_governance_asset(
+    session: Any,
+    *,
+    tenant_id: str,
+    asset_type: str,
+    asset_id: str,
+    asset_name: str,
+    source_table: str,
+    source_path: str | None,
+    metadata: dict[str, Any],
+    now: datetime,
+) -> Any:
+    from app.models.governance_asset import GovernanceAsset
+
+    asset = (
+        session.query(GovernanceAsset)
+        .filter(
+            GovernanceAsset.tenant_id == tenant_id,
+            GovernanceAsset.asset_type == asset_type,
+            GovernanceAsset.asset_id == asset_id,
+        )
+        .first()
+    )
+    if asset is None:
+        asset = GovernanceAsset(
+            id=stable_id("govast", f"{tenant_id}:{asset_type}:{asset_id}"),
+            tenant_id=tenant_id,
+            asset_type=asset_type,
+            asset_id=asset_id,
+            asset_name=asset_name,
+            created_at=now,
+        )
+        session.add(asset)
+    asset.asset_name = asset_name
+    asset.source_table = source_table
+    asset.source_path = source_path
+    asset.governance_status = "approved"
+    asset.review_status = "approved"
+    asset.production_status = "approved"
+    asset.latest_decision_result = "approved"
+    asset.latest_evidence_status = "ready"
+    asset.metadata_json = metadata
+    asset.notes = "Approved by Codex autonomous approval mode for Phase13 formal absorption."
+    asset.updated_at = now
+    session.flush()
+    return asset
+
+
+def create_template_version(session: Any, template: Any, user: dict[str, Any], change_summary: str, now: datetime) -> None:
+    from app.models.contract_template_version import ContractTemplateVersion
+
+    (
+        session.query(ContractTemplateVersion)
+        .filter(
+            ContractTemplateVersion.tenant_id == template.tenant_id,
+            ContractTemplateVersion.template_id == template.id,
+            ContractTemplateVersion.is_current.is_(True),
+        )
+        .update({"is_current": False}, synchronize_session=False)
+    )
+    session.add(
+        ContractTemplateVersion(
+            id=stable_id("tplver", f"{template.id}:{template.current_version_label}:{now.isoformat()}"),
+            template_id=template.id,
+            tenant_id=template.tenant_id,
+            version_label=template.current_version_label,
+            change_summary=change_summary,
+            name=template.name,
+            contract_type=template.contract_type,
+            document_type_code=template.document_type_code,
+            template_category=template.template_category,
+            industry_category=template.industry_category,
+            description=template.description,
+            content=template.content,
+            variables_json=template.variables_json,
+            is_current=True,
+            created_by=user["id"],
+            created_by_name=user.get("display_name"),
+            created_at=template.latest_version_created_at or now,
+        )
+    )
+
+
+def create_review_rule_version(session: Any, rule: Any, user: dict[str, Any], change_summary: str, now: datetime) -> None:
+    from app.models.review_rule_version import ReviewRuleVersion
+
+    (
+        session.query(ReviewRuleVersion)
+        .filter(
+            ReviewRuleVersion.tenant_id == rule.tenant_id,
+            ReviewRuleVersion.rule_id == rule.id,
+            ReviewRuleVersion.is_current.is_(True),
+        )
+        .update({"is_current": False}, synchronize_session=False)
+    )
+    session.add(
+        ReviewRuleVersion(
+            id=stable_id("rulever", f"{rule.id}:{rule.current_version_label}:{now.isoformat()}"),
+            rule_id=rule.id,
+            tenant_id=rule.tenant_id,
+            version_label=rule.current_version_label,
+            change_summary=change_summary,
+            name=rule.name,
+            rule_code=rule.rule_code,
+            keyword=rule.keyword,
+            contract_type=rule.contract_type,
+            risk_level=rule.risk_level,
+            description=rule.description,
+            suggestion=rule.suggestion,
+            legal_basis=rule.legal_basis,
+            is_active=rule.is_active,
+            is_current=True,
+            created_by=user["id"],
+            created_by_name=user.get("display_name"),
+            created_at=rule.latest_version_created_at or now,
+        )
+    )
+
+
+def create_standard_clause_version(session: Any, clause: Any, user: dict[str, Any], change_summary: str, now: datetime) -> None:
+    from app.models.standard_clause_version import StandardClauseVersion
+
+    (
+        session.query(StandardClauseVersion)
+        .filter(
+            StandardClauseVersion.tenant_id == clause.tenant_id,
+            StandardClauseVersion.clause_id == clause.id,
+            StandardClauseVersion.is_current.is_(True),
+        )
+        .update({"is_current": False}, synchronize_session=False)
+    )
+    session.add(
+        StandardClauseVersion(
+            id=stable_id("clausever", f"{clause.id}:{clause.current_version_label}:{now.isoformat()}"),
+            clause_id=clause.id,
+            tenant_id=clause.tenant_id,
+            version_label=clause.current_version_label,
+            change_summary=change_summary,
+            clause_code=clause.clause_code,
+            name=clause.name,
+            contract_type=clause.contract_type,
+            clause_category=clause.clause_category,
+            variant_group=clause.variant_group,
+            source_ids_json=clause.source_ids_json,
+            applicable_conditions_json=clause.applicable_conditions_json,
+            excluded_conditions_json=clause.excluded_conditions_json,
+            risk_notes_path=clause.risk_notes_path,
+            clause_path=clause.clause_path,
+            content=clause.content,
+            governance_status=clause.governance_status,
+            is_active=clause.is_active,
+            metadata_json=clause.metadata_json,
+            is_current=True,
+            created_by=user["id"],
+            created_by_name=user.get("display_name"),
+            created_at=clause.latest_version_created_at or now,
+        )
+    )
+
+
+def template_payload(row: dict[str, str], library_root: Path) -> dict[str, Any]:
+    template_code = row["template_id"].strip()
+    template_path = library_root / row["template_path"].strip()
+    variables_path = library_root / row["variable_schema_path"].strip()
+    variables_schema = read_json(variables_path)
+    variables = [
+        {"name": item, "label": item, "required": True, "source": "phase13_variable_schema"}
+        for item in variables_schema.get("required_variables", [])
+    ]
+    variables.extend(
+        {"name": item, "label": item, "required": False, "source": "phase13_variable_schema"}
+        for item in variables_schema.get("optional_variables", [])
+    )
+    metadata = {
+        "phase": "Phase13",
+        "clause_ids": split_multi(row.get("clause_ids")),
+        "review_rule_ids": split_multi(row.get("review_rule_ids")),
+        "generation_sample_ids": split_multi(row.get("generation_sample_ids")),
+        "validation_rules": variables_schema.get("validation_rules") or [],
+        "review_required": str(row.get("review_required") or "").lower() == "true",
+        "source_inventory_status": row.get("status"),
+    }
+    contract_type = (row.get("contract_type") or "").strip()
+    return {
+        "template_code": template_code,
+        "name": row["template_name"].strip(),
+        "contract_type": contract_type,
+        "document_type_code": TEMPLATE_DOCUMENT_TYPE_CODES.get(template_code),
+        "template_category": TEMPLATE_CATEGORIES.get(contract_type, "general-business"),
+        "industry_category": "general-enterprise",
+        "description": (row.get("notes") or "").strip()[:500] or f"Phase13 template asset {template_code}",
+        "content": extract_template_body(read_text(template_path)),
+        "variables_json": variables,
+        "source_path": row["template_path"].strip(),
+        "variables_source_path": row["variable_schema_path"].strip(),
+        "metadata": metadata,
+    }
+
+
+def review_rule_payload(row: dict[str, str], library_root: Path) -> dict[str, Any]:
+    rule_code = row["rule_id"].strip()
+    rule_path = library_root / row["modification_suggestion_path"].strip()
+    markdown = read_text(rule_path)
+    suggestion = extract_markdown_section(markdown, ["修改建议", "淇敼寤鸿"]) or (row.get("review_output_template") or "").strip()
+    trigger = (row.get("trigger_condition") or "").strip()
+    keyword = (split_multi(trigger)[0] if split_multi(trigger) else trigger)[:100] or rule_code
+    return {
+        "rule_code": rule_code,
+        "name": row["risk_name"].strip(),
+        "keyword": keyword,
+        "contract_type": SPECIALIZED_RULE_CONTRACT_TYPES.get(rule_code, "all"),
+        "risk_level": (row.get("severity") or "medium").strip().lower(),
+        "description": markdown.strip(),
+        "suggestion": suggestion,
+        "legal_basis": ";".join(split_multi(row.get("source_ids"))),
+        "source_path": row["modification_suggestion_path"].strip(),
+        "metadata": {
+            "phase": "Phase13",
+            "risk_category": row.get("risk_category"),
+            "source_ids": split_multi(row.get("source_ids")),
+            "clause_ids": split_multi(row.get("clause_ids")),
+            "review_sample_ids": split_multi(row.get("review_sample_ids")),
+            "trigger_condition": row.get("trigger_condition"),
+            "review_output_template": row.get("review_output_template"),
+            "human_review_required": str(row.get("human_review_required") or "").lower() == "true",
+            "source_inventory_status": row.get("status"),
+        },
+    }
+
+
+def standard_clause_payload(row: dict[str, str], library_root: Path) -> dict[str, Any]:
+    clause_code = row["clause_id"].strip()
+    clause_path_value = row["clause_path"].strip()
+    clause_path = library_root / clause_path_value
+    markdown = read_text(clause_path)
+    return {
+        "clause_code": clause_code,
+        "name": row["clause_name"].strip(),
+        "contract_type": (row.get("contract_type") or "common").strip(),
+        "clause_category": (row.get("clause_category") or "").strip() or None,
+        "variant_group": (row.get("variant_group") or "").strip() or None,
+        "source_ids_json": split_multi(row.get("source_ids")),
+        "applicable_conditions_json": split_multi(row.get("applicable_conditions")),
+        "excluded_conditions_json": split_multi(row.get("excluded_conditions")),
+        "risk_notes_path": (row.get("risk_notes_path") or "").strip() or None,
+        "clause_path": clause_path_value,
+        "content": extract_clause_text(markdown),
+        "metadata": {
+            "phase": "Phase13",
+            "full_markdown": markdown.strip(),
+            "review_required": str(row.get("review_required") or "").lower() == "true",
+            "source_inventory_status": row.get("status"),
+            "notes": row.get("notes"),
+        },
+    }
+
+
+def upsert_templates(session: Any, rows: list[dict[str, str]], library_root: Path, args: argparse.Namespace, results: Phase13Results, counters: Phase13Counters) -> None:
+    from app.models.contract_template import ContractTemplate
+    from app.models.contract_template_version import ContractTemplateVersion
+
+    user = current_user(args)
+    now = datetime.now(timezone.utc)
+    for row in rows:
+        item_result = {"code": row.get("template_id"), "type": "contract_template", "action": "reused", "publish_status": "skipped"}
+        try:
+            payload = template_payload(row, library_root)
+            existing = (
+                session.query(ContractTemplate)
+                .filter(ContractTemplate.tenant_id == args.tenant_id, ContractTemplate.template_code == payload["template_code"])
+                .first()
+            )
+            governance_asset = ensure_governance_asset(
+                session,
+                tenant_id=args.tenant_id,
+                asset_type="contract_template",
+                asset_id=payload["template_code"],
+                asset_name=payload["name"],
+                source_table="phase13_contract_template_inventory",
+                source_path=payload["source_path"],
+                metadata=payload["metadata"],
+                now=now,
+            )
+            if args.dry_run:
+                item_result["action"] = "dry_run_exists" if existing else "dry_run_create"
+                results.templates.append(item_result)
+                counters.reused += 1 if existing else 0
+                continue
+
+            if existing is None:
+                existing = ContractTemplate(
+                    id=stable_id("tpl", f"{args.tenant_id}:{payload['template_code']}"),
+                    tenant_id=args.tenant_id,
+                    created_at=now,
+                )
+                session.add(existing)
+                existing.current_version_label = "v1.0"
+                existing.version_count = 1
+                existing.created_by = user["id"]
+                item_result["action"] = "created"
+                counters.created += 1
+            else:
+                expected = {
+                    "name": payload["name"],
+                    "contract_type": payload["contract_type"],
+                    "document_type_code": payload["document_type_code"],
+                    "template_category": payload["template_category"],
+                    "industry_category": payload["industry_category"],
+                    "description": payload["description"],
+                    "content": payload["content"],
+                    "variables_json": payload["variables_json"],
+                }
+                current = {key: getattr(existing, key) for key in expected}
+                if not json_same(current, expected):
+                    existing.current_version_label = bump_version_label(existing.current_version_label)
+                    existing.version_count = max(existing.version_count or 0, 1) + 1
+                    item_result["action"] = "updated"
+                    counters.updated += 1
+                else:
+                    counters.reused += 1
+
+            existing.name = payload["name"]
+            existing.contract_type = payload["contract_type"]
+            existing.document_type_code = payload["document_type_code"]
+            existing.template_code = payload["template_code"]
+            existing.template_category = payload["template_category"]
+            existing.industry_category = payload["industry_category"]
+            existing.description = payload["description"]
+            existing.content = payload["content"]
+            existing.variables_json = payload["variables_json"]
+            existing.enabled = True
+            existing.latest_change_summary = "Phase13 formal template absorption"
+            existing.latest_version_created_at = now
+            existing.updated_by = user["id"]
+            if not args.no_publish:
+                existing.published_at = now
+                existing.published_by = user["id"]
+                existing.latest_governance_asset_id = governance_asset.id
+                item_result["publish_status"] = "published"
+                counters.published += 1
+
+            session.flush()
+            current_version_count = (
+                session.query(ContractTemplateVersion)
+                .filter(
+                    ContractTemplateVersion.tenant_id == args.tenant_id,
+                    ContractTemplateVersion.template_id == existing.id,
+                    ContractTemplateVersion.is_current.is_(True),
+                )
+                .count()
+            )
+            if item_result["action"] in {"created", "updated"} or current_version_count == 0:
+                if current_version_count == 0 and item_result["action"] == "reused":
+                    item_result["action"] = "version_backfilled"
+                    counters.version_backfilled += 1
+                create_template_version(session, existing, user, "Phase13 formal template absorption", now)
+            item_result["id"] = existing.id
+            item_result["governance_asset_id"] = governance_asset.id
+        except Exception as exc:  # noqa: BLE001
+            counters.failed += 1
+            item_result["action"] = "failed"
+            item_result["error"] = str(exc)
+        results.templates.append(item_result)
+
+
+def upsert_review_rules(session: Any, rows: list[dict[str, str]], library_root: Path, args: argparse.Namespace, results: Phase13Results, counters: Phase13Counters) -> None:
+    from app.models.review_rule import ReviewRule
+    from app.models.review_rule_version import ReviewRuleVersion
+
+    user = current_user(args)
+    now = datetime.now(timezone.utc)
+    for row in rows:
+        item_result = {"code": row.get("rule_id"), "type": "review_risk_rule", "action": "reused", "publish_status": "skipped"}
+        try:
+            payload = review_rule_payload(row, library_root)
+            existing = (
+                session.query(ReviewRule)
+                .filter(ReviewRule.tenant_id == args.tenant_id, ReviewRule.rule_code == payload["rule_code"])
+                .first()
+            )
+            governance_asset = ensure_governance_asset(
+                session,
+                tenant_id=args.tenant_id,
+                asset_type="review_risk_rule",
+                asset_id=payload["rule_code"],
+                asset_name=payload["name"],
+                source_table="phase13_review_rule_inventory",
+                source_path=payload["source_path"],
+                metadata=payload["metadata"],
+                now=now,
+            )
+            if args.dry_run:
+                item_result["action"] = "dry_run_exists" if existing else "dry_run_create"
+                results.review_rules.append(item_result)
+                counters.reused += 1 if existing else 0
+                continue
+
+            if existing is None:
+                existing = ReviewRule(
+                    id=stable_id("rule", f"{args.tenant_id}:{payload['rule_code']}"),
+                    tenant_id=args.tenant_id,
+                    created_at=now,
+                )
+                session.add(existing)
+                existing.current_version_label = "v1.0"
+                existing.version_count = 1
+                item_result["action"] = "created"
+                counters.created += 1
+            else:
+                expected = {
+                    "name": payload["name"],
+                    "keyword": payload["keyword"],
+                    "contract_type": payload["contract_type"],
+                    "risk_level": payload["risk_level"],
+                    "description": payload["description"],
+                    "suggestion": payload["suggestion"],
+                    "legal_basis": payload["legal_basis"],
+                }
+                current = {key: getattr(existing, key) for key in expected}
+                if not json_same(current, expected):
+                    existing.current_version_label = bump_version_label(existing.current_version_label)
+                    existing.version_count = max(existing.version_count or 0, 1) + 1
+                    item_result["action"] = "updated"
+                    counters.updated += 1
+                else:
+                    counters.reused += 1
+
+            existing.name = payload["name"]
+            existing.rule_code = payload["rule_code"]
+            existing.keyword = payload["keyword"]
+            existing.contract_type = payload["contract_type"]
+            existing.risk_level = payload["risk_level"]
+            existing.description = payload["description"]
+            existing.suggestion = payload["suggestion"]
+            existing.legal_basis = payload["legal_basis"]
+            existing.is_active = True
+            existing.latest_change_summary = "Phase13 formal review rule absorption"
+            existing.latest_version_created_at = now
+            if not args.no_publish:
+                existing.published_at = now
+                existing.published_by = user["id"]
+                existing.latest_governance_asset_id = governance_asset.id
+                item_result["publish_status"] = "published"
+                counters.published += 1
+
+            session.flush()
+            current_version_count = (
+                session.query(ReviewRuleVersion)
+                .filter(
+                    ReviewRuleVersion.tenant_id == args.tenant_id,
+                    ReviewRuleVersion.rule_id == existing.id,
+                    ReviewRuleVersion.is_current.is_(True),
+                )
+                .count()
+            )
+            if item_result["action"] in {"created", "updated"} or current_version_count == 0:
+                if current_version_count == 0 and item_result["action"] == "reused":
+                    item_result["action"] = "version_backfilled"
+                    counters.version_backfilled += 1
+                create_review_rule_version(session, existing, user, "Phase13 formal review rule absorption", now)
+            item_result["id"] = existing.id
+            item_result["governance_asset_id"] = governance_asset.id
+        except Exception as exc:  # noqa: BLE001
+            counters.failed += 1
+            item_result["action"] = "failed"
+            item_result["error"] = str(exc)
+        results.review_rules.append(item_result)
+
+
+def upsert_standard_clauses(session: Any, rows: list[dict[str, str]], library_root: Path, args: argparse.Namespace, results: Phase13Results, counters: Phase13Counters) -> None:
+    from app.models.standard_clause import StandardClause
+    from app.models.standard_clause_version import StandardClauseVersion
+
+    user = current_user(args)
+    now = datetime.now(timezone.utc)
+    for row in rows:
+        item_result = {"code": row.get("clause_id"), "type": "standard_clause", "action": "reused", "publish_status": "skipped"}
+        try:
+            payload = standard_clause_payload(row, library_root)
+            existing = (
+                session.query(StandardClause)
+                .filter(StandardClause.tenant_id == args.tenant_id, StandardClause.clause_code == payload["clause_code"])
+                .first()
+            )
+            governance_asset = ensure_governance_asset(
+                session,
+                tenant_id=args.tenant_id,
+                asset_type="standard_clause",
+                asset_id=payload["clause_code"],
+                asset_name=payload["name"],
+                source_table="phase13_clause_library_inventory",
+                source_path=payload["clause_path"],
+                metadata=payload["metadata"],
+                now=now,
+            )
+            if args.dry_run:
+                item_result["action"] = "dry_run_exists" if existing else "dry_run_create"
+                results.standard_clauses.append(item_result)
+                counters.reused += 1 if existing else 0
+                continue
+
+            if existing is None:
+                existing = StandardClause(
+                    id=stable_id("clause", f"{args.tenant_id}:{payload['clause_code']}"),
+                    tenant_id=args.tenant_id,
+                    created_at=now,
+                )
+                session.add(existing)
+                existing.current_version_label = "v1.0"
+                existing.version_count = 1
+                existing.created_by = user["id"]
+                item_result["action"] = "created"
+                counters.created += 1
+            else:
+                expected = {
+                    "name": payload["name"],
+                    "contract_type": payload["contract_type"],
+                    "clause_category": payload["clause_category"],
+                    "variant_group": payload["variant_group"],
+                    "source_ids_json": payload["source_ids_json"],
+                    "applicable_conditions_json": payload["applicable_conditions_json"],
+                    "excluded_conditions_json": payload["excluded_conditions_json"],
+                    "risk_notes_path": payload["risk_notes_path"],
+                    "clause_path": payload["clause_path"],
+                    "content": payload["content"],
+                    "metadata_json": payload["metadata"],
+                }
+                current = {key: getattr(existing, key) for key in expected}
+                if not json_same(current, expected):
+                    existing.current_version_label = bump_version_label(existing.current_version_label)
+                    existing.version_count = max(existing.version_count or 0, 1) + 1
+                    item_result["action"] = "updated"
+                    counters.updated += 1
+                else:
+                    counters.reused += 1
+
+            existing.clause_code = payload["clause_code"]
+            existing.name = payload["name"]
+            existing.contract_type = payload["contract_type"]
+            existing.clause_category = payload["clause_category"]
+            existing.variant_group = payload["variant_group"]
+            existing.source_ids_json = payload["source_ids_json"]
+            existing.applicable_conditions_json = payload["applicable_conditions_json"]
+            existing.excluded_conditions_json = payload["excluded_conditions_json"]
+            existing.risk_notes_path = payload["risk_notes_path"]
+            existing.clause_path = payload["clause_path"]
+            existing.content = payload["content"]
+            existing.metadata_json = payload["metadata"]
+            existing.governance_status = "published" if not args.no_publish else "approved"
+            existing.is_active = True
+            existing.latest_change_summary = "Phase13 formal standard clause absorption"
+            existing.latest_version_created_at = now
+            existing.updated_by = user["id"]
+            if not args.no_publish:
+                existing.published_at = now
+                existing.published_by = user["id"]
+                existing.latest_governance_asset_id = governance_asset.id
+                item_result["publish_status"] = "published"
+                counters.published += 1
+
+            session.flush()
+            current_version_count = (
+                session.query(StandardClauseVersion)
+                .filter(
+                    StandardClauseVersion.tenant_id == args.tenant_id,
+                    StandardClauseVersion.clause_id == existing.id,
+                    StandardClauseVersion.is_current.is_(True),
+                )
+                .count()
+            )
+            if item_result["action"] in {"created", "updated"} or current_version_count == 0:
+                if current_version_count == 0 and item_result["action"] == "reused":
+                    item_result["action"] = "version_backfilled"
+                    counters.version_backfilled += 1
+                create_standard_clause_version(session, existing, user, "Phase13 formal standard clause absorption", now)
+            item_result["id"] = existing.id
+            item_result["governance_asset_id"] = governance_asset.id
+        except Exception as exc:  # noqa: BLE001
+            counters.failed += 1
+            item_result["action"] = "failed"
+            item_result["error"] = str(exc)
+        results.standard_clauses.append(item_result)
+
+
+def count_table(session: Any, model: Any, tenant_id: str, **filters: Any) -> int:
+    query = session.query(model).filter(model.tenant_id == tenant_id)
+    for field_name, value in filters.items():
+        query = query.filter(getattr(model, field_name) == value)
+    return int(query.count())
+
+
+def build_snapshot(session: Any, tenant_id: str) -> dict[str, Any]:
+    from app.models.contract_template import ContractTemplate
+    from app.models.contract_template_version import ContractTemplateVersion
+    from app.models.governance_asset import GovernanceAsset
+    from app.models.review_rule import ReviewRule
+    from app.models.review_rule_version import ReviewRuleVersion
+    from app.models.standard_clause import StandardClause
+    from app.models.standard_clause_version import StandardClauseVersion
+
+    return {
+        "contract_templates": count_table(session, ContractTemplate, tenant_id),
+        "phase13_templates": session.query(ContractTemplate)
+        .filter(ContractTemplate.tenant_id == tenant_id, ContractTemplate.template_code.in_(list(TEMPLATE_DOCUMENT_TYPE_CODES)))
+        .count(),
+        "phase13_templates_published": session.query(ContractTemplate)
+        .filter(ContractTemplate.tenant_id == tenant_id, ContractTemplate.template_code.in_(list(TEMPLATE_DOCUMENT_TYPE_CODES)), ContractTemplate.published_at.isnot(None))
+        .count(),
+        "contract_template_versions": count_table(session, ContractTemplateVersion, tenant_id),
+        "review_rules": count_table(session, ReviewRule, tenant_id),
+        "review_rules_published": count_table(session, ReviewRule, tenant_id, is_active=True),
+        "review_rule_versions": count_table(session, ReviewRuleVersion, tenant_id),
+        "standard_clauses": count_table(session, StandardClause, tenant_id),
+        "standard_clauses_published": session.query(StandardClause)
+        .filter(StandardClause.tenant_id == tenant_id, StandardClause.published_at.isnot(None), StandardClause.is_active.is_(True))
+        .count(),
+        "standard_clause_versions": count_table(session, StandardClauseVersion, tenant_id),
+        "governance_assets_contract_template": count_table(session, GovernanceAsset, tenant_id, asset_type="contract_template"),
+        "governance_assets_review_risk_rule": count_table(session, GovernanceAsset, tenant_id, asset_type="review_risk_rule"),
+        "governance_assets_standard_clause": count_table(session, GovernanceAsset, tenant_id, asset_type="standard_clause"),
+    }
+
+
+def write_markdown_report(path: Path, payload: dict[str, Any]) -> None:
+    def table(rows: list[dict[str, Any]]) -> list[str]:
+        output = ["| 编号 | 类型 | 动作 | 发布状态 |", "| --- | --- | --- | --- |"]
+        for item in rows:
+            output.append(f"| {item.get('code')} | {item.get('type')} | {item.get('action')} | {item.get('publish_status')} |")
+        return output
+
+    lines = [
+        "# Phase13 模板、审查规则、标准条款正式吸收报告",
+        "",
+        f"生成时间：{payload['generated_at']}",
+        f"租户：`{payload['tenant_id']}`",
+        "",
+        "## 结论",
+        "",
+        f"- 本轮处理模板 `{payload['summary']['templates']}` 个、审查规则 `{payload['summary']['review_rules']}` 条、标准条款 `{payload['summary']['standard_clauses']}` 条。",
+        f"- 新建 `{payload['counters']['created']}`，更新 `{payload['counters']['updated']}`，复用 `{payload['counters']['reused']}`，补版本 `{payload['counters']['version_backfilled']}`，发布 `{payload['counters']['published']}`，失败 `{payload['counters']['failed']}`。",
+        f"- 执行状态：`{payload['status']}`。",
+        "",
+        "## 运行表快照",
+        "",
+        f"- Phase13 模板：{payload['after_snapshot']['phase13_templates']}，已发布：{payload['after_snapshot']['phase13_templates_published']}，模板版本：{payload['after_snapshot']['contract_template_versions']}",
+        f"- 审查规则：{payload['after_snapshot']['review_rules']}，启用/发布口径：{payload['after_snapshot']['review_rules_published']}，规则版本：{payload['after_snapshot']['review_rule_versions']}",
+        f"- 标准条款：{payload['after_snapshot']['standard_clauses']}，已发布：{payload['after_snapshot']['standard_clauses_published']}，条款版本：{payload['after_snapshot']['standard_clause_versions']}",
+        "",
+        "## 模板",
+        "",
+        *table(payload["results"]["templates"]),
+        "",
+        "## 审查规则",
+        "",
+        *table(payload["results"]["review_rules"]),
+        "",
+        "## 标准条款",
+        "",
+        *table(payload["results"]["standard_clauses"]),
+        "",
+    ]
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def seed_phase13_assets(args: argparse.Namespace) -> dict[str, Any]:
+    if args.bootstrap_mode == "sqlite_phase13_assets":
+        bootstrap_sqlite_phase13_assets()
+
+    from app.db import SessionLocal
+
+    repo_root = resolve_repo_root(args)
+    library_root = resolve_library_root(args, repo_root)
+    template_rows = read_csv_rows(library_root / "inventories/contract-template-inventory.csv")
+    review_rule_rows = read_csv_rows(library_root / "inventories/review-risk-rule-inventory.csv")
+    clause_rows = read_csv_rows(library_root / "inventories/clause-library-inventory.csv")
+    counters = Phase13Counters()
+    results = Phase13Results()
+
+    session = SessionLocal()
+    try:
+        before_snapshot = build_snapshot(session, args.tenant_id)
+        upsert_templates(session, template_rows, library_root, args, results, counters)
+        upsert_review_rules(session, review_rule_rows, library_root, args, results, counters)
+        upsert_standard_clauses(session, clause_rows, library_root, args, results, counters)
+        if args.dry_run:
+            session.rollback()
+        else:
+            session.commit()
+        after_snapshot = build_snapshot(session, args.tenant_id)
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+    status = "PASS" if counters.failed == 0 else "FAIL"
+    return {
+        "status": status,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "tenant_id": args.tenant_id,
+        "user_id": args.user_id,
+        "repo_root": str(repo_root),
+        "library_root": str(library_root),
+        "dry_run": args.dry_run,
+        "publish_enabled": not args.no_publish,
+        "summary": {
+            "templates": len(template_rows),
+            "review_rules": len(review_rule_rows),
+            "standard_clauses": len(clause_rows),
+        },
+        "counters": counters.__dict__,
+        "before_snapshot": before_snapshot,
+        "after_snapshot": after_snapshot,
+        "results": {
+            "templates": results.templates,
+            "review_rules": results.review_rules,
+            "standard_clauses": results.standard_clauses,
+        },
+    }
+
+
+def main() -> int:
+    args = parse_args()
+    configure_env(args.database_url)
+    repo_root = resolve_repo_root(args)
+    add_import_roots(repo_root)
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    try:
+        payload = seed_phase13_assets(args)
+    except Exception as exc:  # noqa: BLE001
+        payload = {
+            "status": "FAIL",
+            "generated_at": datetime.now(timezone.utc).isoformat(),
+            "tenant_id": args.tenant_id,
+            "error": str(exc),
+        }
+        print(json.dumps(payload, ensure_ascii=False))
+        return 1
+
+    summary_json = args.output_dir / "phase13-template-rule-clause-absorption-summary.json"
+    summary_md = args.output_dir / "phase13-template-rule-clause-absorption-summary.md"
+    summary_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
+    write_markdown_report(summary_md, payload)
+    print(json.dumps(payload, ensure_ascii=False, default=str))
+    marker = "PASS" if payload["status"] == "PASS" else "FAIL"
+    print(f"PHASE13_TEMPLATE_RULE_CLAUSE_ABSORPTION_{marker} {args.output_dir}")
+    return 0 if payload["status"] == "PASS" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

project-docs/AI合同起草与审查助手-当前工作进度保存-2026-05-11.md

@@ -4602,3 +4602,64 @@
 - 当前闭环结论：
   - Prompt 资产正式种入、发布、黄金样本评测和运行时解析已闭环。
   - 生成、审查、Chat、条款推荐、措辞优化、法规影响扫描 6 条运行链路已经可以解析到已发布 Prompt。
+
+## 2026-05-22 Phase13 模板、审查规则、标准条款正式吸收闭环
+
+- 本轮目标：
+  - 补齐 Phase13 模板、审查规则、标准条款从资料包到业务运行表的正式吸收链路。
+  - 让模板、规则、标准条款都具备“运行表 -> 版本表 -> 治理资产 -> 发布状态 -> 远程 PostgreSQL 真相源验证”的闭环。
+- 新增后端能力：
+  - 新增标准条款运行表：`standard_clauses`。
+  - 新增标准条款版本表：`standard_clause_versions`。
+  - 新增标准条款管理接口：
+    - `GET /api/v1/standard-clauses`
+    - `GET /api/v1/standard-clauses/{clause_id}/versions`
+    - `POST /api/v1/standard-clauses/{clause_id}/publish`
+    - `POST /api/v1/standard-clauses/{clause_id}/unpublish`
+    - `POST /api/v1/standard-clauses/{clause_id}/rollback`
+  - 发布前校验 `standard_clause` 治理资产必须处于 `ready/approved` 生产可发布状态。
+  - 回滚会生成新版本并回到草稿/未发布状态，不直接覆盖旧版本。
+- 新增自动化工具：
+  - `backend/tools/seed_phase13_template_rule_clause_assets.py`
+  - `scripts/run-phase13-template-rule-clause-absorption.ps1`
+  - 该脚本会把 Phase13 资料包复制到远程后端容器，直接写入远程 PostgreSQL 真相源并生成吸收报告。
+- 资产盘点工具同步修正：
+  - `backend/tools/asset_inventory_gap_report.py`
+  - `scripts/run-asset-inventory-gap-report.ps1`
+  - 盘点报告现在会统计 `standard_clauses`、`standard_clause_versions`、标准条款发布数和启用数。
+- 本地验证：
+  - `python -m compileall backend/app backend/tests backend/tools`
+  - `$env:PYTHONPATH='D:\ai-contract-platform\backend'; python -m unittest backend.tests.test_templates_api backend.tests.test_template_service backend.tests.test_review_rules_api backend.tests.test_review_rule_service backend.tests.test_standard_clauses_api backend.tests.test_standard_clause_service backend.tests.test_permission_metadata`
+  - 结果：`Ran 57 tests in 0.179s`，`OK`。
+  - 隔离 SQLite 写入烟测：`PHASE13_TEMPLATE_RULE_CLAUSE_ABSORPTION_PASS D:\ai-contract-platform\project-docs\qa-artifacts\phase13-template-rule-clause-absorption\local-write`
+  - 隔离写入后快照：模板 `10/10` 发布、审查规则 `8/8` 发布、标准条款 `8/8` 发布，三类治理资产共 `26` 条。
+- 测试服务器部署：
+  - 部署命令：`$env:SSHPASS='***'; powershell -NoProfile -ExecutionPolicy Bypass -File scripts\deploy-ai-contract-restore-docker.ps1 -UseSshpass -AllowDirtyWorktree`
+  - 部署结果：`RESTORE_DOCKER_DEPLOY_WARN`
+  - `WARN` 原因：允许未提交工作区部署到测试服；所有 required 检查均通过。
+  - 部署证据目录：`project-docs/qa-artifacts/restore-docker-deploy/2026-05-22T11-20-38`
+  - 远程后端健康检查：`http://192.168.3.90:8001/health` 返回 HTTP 200。
+  - 远程 readiness：`http://192.168.3.90:8001/api/v1/system/readiness` 返回 HTTP 200。
+  - 远程前端烟测：`http://192.168.3.90:3003/operation-kpis` 返回 HTTP 200。
+  - 远程 Alembic：`20260522_0040 (head)`。
+- 远程 PostgreSQL 正式吸收结果：
+  - 命令：`$env:SSHPASS='***'; powershell -NoProfile -ExecutionPolicy Bypass -File scripts\run-phase13-template-rule-clause-absorption.ps1 -UseSshpass`
+  - 结果：`PHASE13_TEMPLATE_RULE_CLAUSE_ABSORPTION_PASS D:\ai-contract-platform\project-docs\qa-artifacts\phase13-template-rule-clause-absorption\2026-05-22T11-28-00`
+  - 处理资产：模板 `10` 个、审查规则 `8` 条、标准条款 `8` 条。
+  - 本轮新建 `23` 条、更新 `3` 条、发布 `26` 条、失败 `0` 条。
+  - 远程吸收后快照：
+    - Phase13 模板：`10` 个，已发布 `10` 个，模板版本 `14` 条。
+    - 审查规则：`8` 条，已发布/启用 `8` 条，规则版本 `8` 条。
+    - 标准条款：`8` 条，已发布 `8` 条，条款版本 `8` 条。
+    - 治理资产：模板 `10` 条、审查规则 `8` 条、标准条款 `8` 条。
+- 复跑盘点结论：
+  - 证据：`project-docs/qa-artifacts/asset-inventory-gap-report/2026-05-22T11-28-29/asset-inventory-gap-report.json`
+  - 本地 Phase13 模板清单 `10` 个，远程 Phase13 模板缺口为 `0`。
+  - 远程 `remote_templates_published=10`。
+  - 远程 `remote_review_rules=8`、`remote_review_rule_versions=8`、`remote_review_rules_published=8`。
+  - 远程 `remote_standard_clauses=8`、`remote_standard_clause_versions=8`、`remote_standard_clauses_published=8`、`remote_standard_clauses_active=8`。
+  - 必备 Prompt 类型缺口仍为 `无`。
+- 当前闭环结论：
+  - Phase13 模板、审查规则、标准条款已完成正式运行表吸收、版本化、治理资产绑定、发布和远程真相源复核。
+  - 本轮没有 required 阻塞项。
+  - 后续若继续补资产能力，下一步应转向“法律库增量抓取/来源差异复核”或“全量法规分批向量化任务”，而不是继续补 Phase13 模板/规则/条款运行表。

scripts/run-asset-inventory-gap-report.ps1

@@ -184,6 +184,8 @@ with engine.connect() as conn:
                 "contract_template_versions",
                 "review_rules",
                 "review_rule_versions",
+                "standard_clauses",
+                "standard_clause_versions",
                 "prompt_assets",
                 "prompt_asset_versions",
                 "prompt_asset_run_links",
@@ -212,6 +214,14 @@ with engine.connect() as conn:
         ),
         "review_rules_published": published_count(conn, "review_rules"),
         "review_rule_versions": count_table(conn, "review_rule_versions"),
+        "standard_clauses": count_table(conn, "standard_clauses"),
+        "standard_clauses_active": (
+            count_table(conn, "standard_clauses", "is_active = true")
+            if has_column("standard_clauses", "is_active")
+            else 0
+        ),
+        "standard_clauses_published": published_count(conn, "standard_clauses"),
+        "standard_clause_versions": count_table(conn, "standard_clause_versions"),
         "prompt_assets": count_table(conn, "prompt_assets"),
         "prompts_published": published_count(conn, "prompt_assets"),
         "prompt_asset_versions": count_table(conn, "prompt_asset_versions"),

scripts/run-phase13-template-rule-clause-absorption.ps1

@@ -0,0 +1,167 @@
+param(
+  [string]$RemoteHost = "192.168.3.90",
+  [string]$RemoteUser = "root",
+  [string]$BackendContainer = "ai-contract-backend-restore",
+  [string]$SshKeyPath = "$HOME\.ssh\id_rsa",
+  [switch]$UseSshpass,
+  [string]$TenantId = "tenant-demo-001",
+  [string]$UserId = "codex-autonomous-approval",
+  [string]$DisplayName = "CodexAutonomousApprovalMode",
+  [ValidateSet("none", "sqlite_phase13_assets")]
+  [string]$BootstrapMode = "none",
+  [switch]$DryRun,
+  [switch]$NoPublish,
+  [string]$ArtifactRoot = ""
+)
+
+$ErrorActionPreference = "Stop"
+
+function Resolve-SshpassPath {
+  $command = Get-Command sshpass.exe -ErrorAction SilentlyContinue
+  if ($null -ne $command) {
+    return $command.Source
+  }
+
+  $candidates = @(
+    (Join-Path $env:LOCALAPPDATA "Microsoft\WinGet\Packages\xhcoding.sshpass-win32_Microsoft.Winget.Source_8wekyb3d8bbwe\sshpass.exe")
+  )
+  foreach ($candidate in $candidates) {
+    if (Test-Path $candidate) {
+      return $candidate
+    }
+  }
+
+  return $null
+}
+
+function ConvertTo-ShellSingleQuote {
+  param([string]$Value)
+  return "'" + ($Value -replace "'", "'`"`"`'") + "'"
+}
+
+function Invoke-LoggedCommand {
+  param(
+    [string]$Name,
+    [string]$FilePath,
+    [string[]]$Arguments,
+    [string]$LogDir
+  )
+
+  $logPath = Join-Path $LogDir "$Name.log"
+  $previousErrorActionPreference = $ErrorActionPreference
+  $ErrorActionPreference = "Continue"
+  try {
+    $output = & $FilePath @Arguments 2>&1
+    $exitCode = if ($LASTEXITCODE -is [int]) { $LASTEXITCODE } else { 0 }
+  } finally {
+    $ErrorActionPreference = $previousErrorActionPreference
+  }
+
+  $outputText = ($output -join [Environment]::NewLine)
+  $outputText | Set-Content -Encoding UTF8 -Path $logPath
+  if ($exitCode -ne 0) {
+    throw "$Name failed with exit code $exitCode. See $logPath"
+  }
+  return $outputText
+}
+
+$repoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path
+if (-not (Test-Path "$HOME\.Codex\skills\gstack\bin")) {
+  throw "GSTACK_MISSING: gstack is required for AI-assisted work in this repo."
+}
+
+if ([string]::IsNullOrWhiteSpace($ArtifactRoot)) {
+  $ArtifactRoot = Join-Path $repoRoot "project-docs\qa-artifacts\phase13-template-rule-clause-absorption"
+}
+$timestamp = Get-Date -Format "yyyy-MM-ddTHH-mm-ss"
+$artifactDir = Join-Path $ArtifactRoot $timestamp
+New-Item -ItemType Directory -Force -Path $artifactDir | Out-Null
+
+$seedToolPath = Join-Path $repoRoot "backend\tools\seed_phase13_template_rule_clause_assets.py"
+$phase13LibraryPath = Join-Path $repoRoot "project-docs\phase13-legal-knowledge-template-library"
+foreach ($requiredPath in @($seedToolPath, $phase13LibraryPath)) {
+  if (-not (Test-Path -LiteralPath $requiredPath -PathType Leaf)) {
+    if (-not (Test-Path -LiteralPath $requiredPath -PathType Container)) {
+      throw "Required input path does not exist: $requiredPath"
+    }
+  }
+}
+
+$authMode = ""
+$sshpassPath = $null
+if ($UseSshpass -or -not [string]::IsNullOrWhiteSpace($env:SSHPASS)) {
+  $sshpassPath = Resolve-SshpassPath
+  if ([string]::IsNullOrWhiteSpace($sshpassPath)) {
+    throw "sshpass.exe not found. Install sshpass or use SSH key auth."
+  }
+  if ([string]::IsNullOrWhiteSpace($env:SSHPASS)) {
+    throw "SSHPASS environment variable is empty."
+  }
+  $authMode = "sshpass"
+} elseif (Test-Path $SshKeyPath) {
+  $authMode = "key"
+} else {
+  throw "SSH key not found and SSHPASS is not configured."
+}
+
+$target = "${RemoteUser}@${RemoteHost}"
+$remoteWorkDir = "/tmp/ai-contract-phase13-absorption-$timestamp"
+$containerScript = "$remoteWorkDir/seed_phase13_template_rule_clause_assets.py"
+$containerRepoRoot = "$remoteWorkDir/repo"
+$containerLibraryRoot = "$containerRepoRoot/project-docs/phase13-legal-knowledge-template-library"
+$containerOutput = "$remoteWorkDir/output"
+
+if ($authMode -eq "sshpass") {
+  $sshFile = $sshpassPath
+  $sshPrefix = @("-e", "ssh")
+  $scpPrefix = @("-e", "scp")
+} else {
+  $sshFile = "ssh"
+  $sshPrefix = @("-i", $SshKeyPath)
+  $scpPrefix = @("-i", $SshKeyPath)
+}
+
+$sshCommon = @("-o", "StrictHostKeyChecking=accept-new", "-o", "ConnectTimeout=10")
+$scpCommon = @("-o", "StrictHostKeyChecking=accept-new", "-o", "ConnectTimeout=10")
+if ($authMode -eq "key") {
+  $sshCommon = @("-o", "BatchMode=yes") + $sshCommon
+  $scpCommon = @("-o", "BatchMode=yes") + $scpCommon
+}
+
+Invoke-LoggedCommand -Name "remote-mkdir" -FilePath $sshFile -Arguments ($sshPrefix + $sshCommon + @($target, "mkdir -p $(ConvertTo-ShellSingleQuote $remoteWorkDir) $(ConvertTo-ShellSingleQuote "$remoteWorkDir/project-docs")")) -LogDir $artifactDir | Out-Null
+
+Invoke-LoggedCommand -Name "scp-seed-tool" -FilePath $(if ($authMode -eq "sshpass") { $sshpassPath } else { "scp" }) -Arguments ($scpPrefix + $scpCommon + @($seedToolPath, "${target}:$remoteWorkDir/seed_phase13_template_rule_clause_assets.py")) -LogDir $artifactDir | Out-Null
+
+Invoke-LoggedCommand -Name "scp-phase13-library" -FilePath $(if ($authMode -eq "sshpass") { $sshpassPath } else { "scp" }) -Arguments ($scpPrefix + $scpCommon + @("-r", $phase13LibraryPath, "${target}:$remoteWorkDir/project-docs/")) -LogDir $artifactDir | Out-Null
+
+$remoteCommandParts = @(
+  "docker exec $BackendContainer sh -lc $(ConvertTo-ShellSingleQuote "rm -rf $remoteWorkDir && mkdir -p $remoteWorkDir $containerRepoRoot $containerRepoRoot/project-docs")",
+  "docker cp $remoteWorkDir/seed_phase13_template_rule_clause_assets.py ${BackendContainer}:$containerScript",
+  "docker cp $remoteWorkDir/project-docs/phase13-legal-knowledge-template-library ${BackendContainer}:$containerRepoRoot/project-docs/phase13-legal-knowledge-template-library"
+)
+$seedArgs = @(
+  "PYTHONPATH=/app python $(ConvertTo-ShellSingleQuote $containerScript)",
+  "--repo-root $(ConvertTo-ShellSingleQuote $containerRepoRoot)",
+  "--library-root $(ConvertTo-ShellSingleQuote $containerLibraryRoot)",
+  "--output-dir $(ConvertTo-ShellSingleQuote $containerOutput)",
+  "--tenant-id $(ConvertTo-ShellSingleQuote $TenantId)",
+  "--user-id $(ConvertTo-ShellSingleQuote $UserId)",
+  "--display-name $(ConvertTo-ShellSingleQuote $DisplayName)",
+  "--bootstrap-mode $(ConvertTo-ShellSingleQuote $BootstrapMode)"
+)
+if ($DryRun) {
+  $seedArgs += "--dry-run"
+}
+if ($NoPublish) {
+  $seedArgs += "--no-publish"
+}
+$seedCommand = "cd /app && " + ($seedArgs -join " ")
+$remoteCommandParts += "docker exec $BackendContainer sh -lc $(ConvertTo-ShellSingleQuote $seedCommand)"
+$remoteCommandParts += "docker cp ${BackendContainer}:$containerOutput $remoteWorkDir/output"
+
+$remoteCommand = $remoteCommandParts -join " && "
+Invoke-LoggedCommand -Name "remote-seed-phase13-assets" -FilePath $sshFile -Arguments ($sshPrefix + $sshCommon + @($target, $remoteCommand)) -LogDir $artifactDir | Out-Null
+
+Invoke-LoggedCommand -Name "scp-seed-output" -FilePath $(if ($authMode -eq "sshpass") { $sshpassPath } else { "scp" }) -Arguments ($scpPrefix + $scpCommon + @("-r", "${target}:$remoteWorkDir/output/.", $artifactDir)) -LogDir $artifactDir | Out-Null
+
+Write-Output "PHASE13_TEMPLATE_RULE_CLAUSE_ABSORPTION_PASS $artifactDir"