"""ORM models for the auditor domain.

See ``server/alembic/versions/092_security_auditor.py`` for the
canonical schema and ``constants.py`` for the value enumerations.
"""
from __future__ import annotations

import uuid
from datetime import datetime
from typing import Any

from sqlalchemy import (
    JSON,
    BigInteger,
    Boolean,
    Date,
    DateTime,
    ForeignKey,
    Index,
    Integer,
    String,
    Text,
    UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.sql import func

from app.core.database import Base
from app.core.sql_types import GUID as Uuid
from app.core.sql_types import PreciseDateTime

# ════════════════════════════════════════════════════════════════════
# 1. MFA compliance audit
# ════════════════════════════════════════════════════════════════════


class AuditorMfaStatus(Base):
    """One row per (person, provider). Snapshot is overwritten on each
    sync — historic state is recoverable from the audit log if needed.
    """

    __tablename__ = "auditor_mfa_status"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False,
    )
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    provider_user_id: Mapped[str | None] = mapped_column(String(255))
    provider_email: Mapped[str | None] = mapped_column(String(255))
    is_enrolled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    enrollment_methods: Mapped[list[str] | None] = mapped_column(JSON)
    is_enforced: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    # Was the fact above ACTUALLY reported by the provider, or is the
    # ``False`` just "we could not see"? A provider that cannot answer
    # must never be read as answering "no" (2026-08-18): all 528
    # Atlassian rows carry ``policy_status='policy_absent'`` (the org
    # 2FA policy is unreadable), and that unknown-as-failure produced
    # 412 unactionable "MFA not enrolled" findings, an
    # ``mfa_missing_provider`` risk penalty on 528 of 570 people, a
    # ``mfa_coverage`` ISMS control reading ``0/412 enrolled (0.0%)``,
    # and a 30/100 health metric — while Google Workspace was in fact
    # enforcing 2SV on 98% of the workforce.
    #
    # Written by ``MfaSyncService``, the only code that knows what the
    # upstream API actually answered:
    #   * google    → enrollment ✓ (``isEnrolledIn2Sv``),
    #                 enforcement ✓ (``isEnforcedIn2Sv``)
    #   * atlassian → both ✓ only when ``policy_status != policy_absent``
    #                 (its ``is_enrolled`` IS the policy state)
    #   * feishu    → enrollment ✓ only when ``security_strategy`` came
    #                 back (needs enterprise + audit scope); enforcement
    #                 is not exposed at all
    #   * github    → enrollment ✓; enforcement ✗ (org SAML/2FA
    #                 enforcement is a separate unreadable toggle)
    # NULL = pre-fix row not yet restated by a sync. Every consumer reads
    # NULL as "not observed", which is the safe direction: the row leaves
    # the coverage denominator instead of becoming a false accusation.
    enrollment_observed: Mapped[bool | None] = mapped_column(Boolean)
    enforcement_observed: Mapped[bool | None] = mapped_column(Boolean)
    last_verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_synced_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    raw_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "person_id", "provider",
            name="uq_auditor_mfa_status_person_provider",
        ),
        Index("idx_auditor_mfa_provider_enrolled", "provider", "is_enrolled"),
        Index("idx_auditor_mfa_synced", "last_synced_at"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorMfaStatus person={self.person_id} "
            f"provider={self.provider} enrolled={self.is_enrolled}>"
        )


# ════════════════════════════════════════════════════════════════════
# 3. Public exposure scan
# ════════════════════════════════════════════════════════════════════


class AuditorPublicExposureFinding(Base):
    """One open or resolved exposure finding.

    Scan run id groups findings by scan invocation. ``UNIQUE(source,
    resource_id, exposure_type)`` enforces upsert semantics — the same
    SG/port re-detected updates ``last_seen_at`` instead of inserting.
    """

    __tablename__ = "auditor_public_exposure_findings"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    scan_run_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    source: Mapped[str] = mapped_column(String(64), nullable=False)
    provider_account_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    region: Mapped[str | None] = mapped_column(String(64))
    resource_id: Mapped[str] = mapped_column(String(255), nullable=False)
    resource_arn: Mapped[str | None] = mapped_column(String(512))
    resource_name: Mapped[str | None] = mapped_column(String(255))
    exposure_type: Mapped[str] = mapped_column(String(64), nullable=False)
    exposure_detail: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    severity: Mapped[str] = mapped_column(
        String(16), nullable=False, default="medium",
    )
    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    last_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="open",
    )
    resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    resolved_by_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    resolution_note: Mapped[str | None] = mapped_column(Text)
    raw_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "source", "resource_id", "exposure_type",
            name="uq_auditor_pubex_resource",
        ),
        Index("idx_auditor_pubex_status_severity", "status", "severity"),
        Index("idx_auditor_pubex_source", "source"),
        Index("idx_auditor_pubex_last_seen", "last_seen_at"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorPublicExposureFinding {self.source}/{self.resource_id} "
            f"type={self.exposure_type} status={self.status} "
            f"severity={self.severity}>"
        )


# ════════════════════════════════════════════════════════════════════
# 4. Secret rotation ledger
# ════════════════════════════════════════════════════════════════════


class AuditorSecretRotationRecord(Base):
    """One row per tracked secret. Status & due date refreshed daily."""

    __tablename__ = "auditor_secret_rotation_records"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    secret_kind: Mapped[str] = mapped_column(String(64), nullable=False)
    source: Mapped[str] = mapped_column(String(64), nullable=False)
    resource_id: Mapped[str] = mapped_column(String(255), nullable=False)
    resource_name: Mapped[str | None] = mapped_column(String(255))
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    owner_email: Mapped[str | None] = mapped_column(String(255))
    created_at_source: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_rotated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    rotation_period_days: Mapped[int] = mapped_column(
        Integer, nullable=False, default=90,
    )
    next_rotation_due_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="current",
    )
    days_until_due: Mapped[int | None] = mapped_column(Integer)
    raw_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    last_synced_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "source", "secret_kind", "resource_id",
            name="uq_auditor_secret_resource",
        ),
        Index("idx_auditor_secret_status_due", "status", "next_rotation_due_at"),
        Index("idx_auditor_secret_kind", "secret_kind"),
        Index("idx_auditor_secret_owner", "owner_person_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorSecretRotationRecord {self.source}/{self.secret_kind}/"
            f"{self.resource_id} status={self.status}>"
        )


# ════════════════════════════════════════════════════════════════════
# 5. Signin events
# ════════════════════════════════════════════════════════════════════


class AuditorSigninEvent(Base):
    """One signin event from any provider, with anomaly score attached.

    ``person_id`` may be NULL if the source identity does not match a
    Helix person (e.g. ex-employee, malicious external login attempt).
    The auditor still cares about those — they are precisely the
    "post-offboard signin" anomaly we score highest.
    """

    __tablename__ = "auditor_signin_events"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    provider_event_id: Mapped[str | None] = mapped_column(String(255))
    provider_user_id: Mapped[str | None] = mapped_column(String(255))
    provider_email: Mapped[str | None] = mapped_column(String(255))
    event_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    event_kind: Mapped[str] = mapped_column(String(64), nullable=False)
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="success",
    )
    source_ip: Mapped[str | None] = mapped_column(String(64))
    source_country: Mapped[str | None] = mapped_column(String(8))
    source_city: Mapped[str | None] = mapped_column(String(128))
    user_agent: Mapped[str | None] = mapped_column(String(512))
    device_id: Mapped[str | None] = mapped_column(String(255))

    anomaly_score: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0,
    )
    anomaly_level: Mapped[str] = mapped_column(
        String(16), nullable=False, default="normal",
    )
    anomaly_reasons: Mapped[list[str] | None] = mapped_column(JSON)

    acknowledged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    acknowledged_by_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    acknowledged_note: Mapped[str | None] = mapped_column(Text)

    raw_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "provider", "provider_event_id",
            name="uq_auditor_signin_provider_event",
        ),
        Index("idx_auditor_signin_person_at", "person_id", "event_at"),
        Index("idx_auditor_signin_anomaly", "anomaly_level", "event_at"),
        Index("idx_auditor_signin_provider_at", "provider", "event_at"),
        Index("idx_auditor_signin_event_at", "event_at"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorSigninEvent {self.provider} person={self.person_id} "
            f"at={self.event_at} score={self.anomaly_score}>"
        )


# ════════════════════════════════════════════════════════════════════
# 6. PAT / MCP / OAuth audit
# ════════════════════════════════════════════════════════════════════


class AuditorPatAudit(Base):
    """One row per token. Status / risk_flags refreshed daily.

    ``token_external_id`` is the upstream id (GitHub PAT id, Helix
    api-key id, OAuth grant id). Within a single ``source+kind`` it
    uniquely identifies the token.
    """

    __tablename__ = "auditor_pat_audits"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    kind: Mapped[str] = mapped_column(String(32), nullable=False)
    source: Mapped[str] = mapped_column(String(32), nullable=False)
    token_external_id: Mapped[str] = mapped_column(String(255), nullable=False)
    person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    owner_email: Mapped[str | None] = mapped_column(String(255))
    token_name: Mapped[str | None] = mapped_column(String(255))
    scopes: Mapped[list[str] | None] = mapped_column(JSON)
    created_at_source: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="active",
    )
    risk_flags: Mapped[list[str] | None] = mapped_column(JSON)
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    revoked_by_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    raw_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    last_synced_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "source", "kind", "token_external_id",
            name="uq_auditor_pat_token",
        ),
        Index("idx_auditor_pat_status_kind", "status", "kind"),
        Index("idx_auditor_pat_person", "person_id"),
        Index("idx_auditor_pat_expires", "expires_at"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorPatAudit {self.kind}/{self.token_external_id} "
            f"owner={self.owner_email} status={self.status}>"
        )


# ════════════════════════════════════════════════════════════════════
# 7. Cert + ICP expiry
# ════════════════════════════════════════════════════════════════════


class AuditorCertExpiry(Base):
    """One row per tracked certificate / domain / ICP filing."""

    __tablename__ = "auditor_cert_expiries"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    asset_kind: Mapped[str] = mapped_column(String(32), nullable=False)
    source: Mapped[str] = mapped_column(String(32), nullable=False)
    external_id: Mapped[str] = mapped_column(String(512), nullable=False)
    name: Mapped[str] = mapped_column(String(512), nullable=False)
    subject: Mapped[str | None] = mapped_column(String(512))
    issuer: Mapped[str | None] = mapped_column(String(255))
    issued_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    expires_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    days_until_expiry: Mapped[int | None] = mapped_column(Integer)
    alert_level: Mapped[str] = mapped_column(
        String(16), nullable=False, default="none",
    )
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="active",
    )
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    owner_email: Mapped[str | None] = mapped_column(String(255))
    environment: Mapped[str | None] = mapped_column(String(32))
    notes: Mapped[str | None] = mapped_column(Text)
    raw_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    last_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "source", "asset_kind", "external_id",
            name="uq_auditor_cert_external",
        ),
        Index("idx_auditor_cert_alert_level", "alert_level"),
        Index("idx_auditor_cert_expires", "expires_at"),
        Index("idx_auditor_cert_kind", "asset_kind"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorCertExpiry {self.asset_kind}/{self.name} "
            f"expires={self.expires_at} alert={self.alert_level}>"
        )


# ════════════════════════════════════════════════════════════════════
# 8a. Risk scores
# ════════════════════════════════════════════════════════════════════


class AuditorRiskScore(Base):
    """One row per active person — overwritten on each recompute.

    ``factors`` is a flat ``{factor_name: int}`` map of accumulated
    weight from each contributing signal. ``breakdown`` carries the
    underlying evidence (which MFA missing, which signin scored high,
    which secret overdue) so the auditor can drill in without joining
    half a dozen tables.
    """

    __tablename__ = "auditor_risk_scores"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False,
    )
    score: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    level: Mapped[str] = mapped_column(
        String(16), nullable=False, default="low",
    )
    factors: Mapped[dict[str, int]] = mapped_column(JSON, nullable=False)
    breakdown: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
    computed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("person_id", name="uq_auditor_risk_person"),
        Index("idx_auditor_risk_level", "level", "score"),
        Index("idx_auditor_risk_computed", "computed_at"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorRiskScore person={self.person_id} "
            f"score={self.score} level={self.level}>"
        )


# ════════════════════════════════════════════════════════════════════
# 8b. Evidence packs
# ════════════════════════════════════════════════════════════════════


class AuditorEvidencePack(Base):
    """Generated PDF + ZIP evidence package for an audit period.

    Run asynchronously through the worker queue — auditor clicks
    "build", we write a row in ``pending`` status, the worker picks it
    up, generates the PDF/ZIP, writes to disk, updates ``status`` to
    ``completed`` (or ``failed`` with ``error_message``).
    """

    __tablename__ = "auditor_evidence_packs"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    requested_by_person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False,
    )
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    period_start: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    period_end: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    modules: Mapped[list[str]] = mapped_column(JSON, nullable=False)
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="pending",
    )
    file_path: Mapped[str | None] = mapped_column(String(512))
    file_size: Mapped[int | None] = mapped_column(BigInteger)
    checksum_sha256: Mapped[str | None] = mapped_column(String(64))
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    error_message: Mapped[str | None] = mapped_column(Text)
    download_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0,
    )
    last_downloaded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    retention_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_evidence_status", "status"),
        Index("idx_auditor_evidence_requester", "requested_by_person_id"),
        Index("idx_auditor_evidence_period", "period_start", "period_end"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorEvidencePack {self.title} status={self.status} "
            f"period={self.period_start}..{self.period_end}>"
        )


# ════════════════════════════════════════════════════════════════════
# 9. Unified findings (cross-control actionable worklist)
# ════════════════════════════════════════════════════════════════════
#
# Every detective control produces *findings* — one row per violating
# item (a person without MFA, an overdue secret, an open exposure, an
# unacknowledged anomaly). Before this table each control kept its own
# bespoke notion of "open item" (or none at all), so there was no single
# inbox an operator could work. ``AuditorFinding`` is that inbox: one
# uniform lifecycle (open → acknowledged → accepted_risk / resolved)
# across all controls, deduplicated per ``(control_key, dedup_key)`` and
# reconciled on every sync (re-seen → bump ``last_seen_at``; disappeared
# → auto-resolved).
#
# The raw per-control tables above remain the full inventory (the MFA
# roster, the secret ledger); findings are the actionable subset layered
# on top. ``compliance_control_status`` "fail" is derived from open
# findings, so posture, the worklist, and the control catalog all read
# from one source of truth.


class AuditorFinding(Base):
    """One actionable security finding emitted by a detective control."""

    __tablename__ = "auditor_findings"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # ComplianceControl.key that produced it ('mfa_coverage', …). Not a FK
    # so a finding survives a control being disabled/removed.
    control_key: Mapped[str] = mapped_column(String(64), nullable=False)
    # Sub-type within the control ('mfa_not_enrolled', 'secret_overdue', …).
    finding_type: Mapped[str] = mapped_column(String(64), nullable=False)
    # Stable identity for upsert within a control. Producers keep it
    # deterministic; long natural keys are hashed (see findings_service).
    dedup_key: Mapped[str] = mapped_column(String(255), nullable=False)

    # What the finding is about.
    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
    subject_ref: Mapped[str | None] = mapped_column(String(512))
    subject_display: Mapped[str | None] = mapped_column(String(255))
    # Set when the finding is about / owned by a person — drives scoping
    # and the risk rollup. NULL for resource-only findings.
    person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )

    severity: Mapped[str] = mapped_column(
        String(16), nullable=False, default="medium",
    )
    title: Mapped[str] = mapped_column(String(512), nullable=False)
    detail: Mapped[dict[str, Any] | None] = mapped_column(JSON)

    # open | acknowledged | accepted_risk | resolved
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="open",
    )

    # For controls whose lifecycle is mirrored onto a legacy row
    # (public exposure, signin events) so a decision can be written back.
    source_table: Mapped[str | None] = mapped_column(String(64))
    source_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    last_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )

    decided_by_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    decision_note: Mapped[str | None] = mapped_column(Text)
    # When accept_risk has an expiry — re-opened by reconcile once passed.
    accepted_until: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # True when resolved by reconcile (item disappeared) vs a manual decision.
    auto_resolved: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False,
    )
    resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "control_key", "dedup_key", name="uq_auditor_finding_dedup",
        ),
        Index("idx_auditor_finding_status_sev", "status", "severity"),
        Index("idx_auditor_finding_control_status", "control_key", "status"),
        Index("idx_auditor_finding_person", "person_id"),
        Index("idx_auditor_finding_last_seen", "last_seen_at"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorFinding {self.control_key}/{self.finding_type} "
            f"{self.dedup_key} status={self.status} sev={self.severity}>"
        )


# ════════════════════════════════════════════════════════════════════
# 11. GRC spine — control catalog + multi-framework crosswalk
# ════════════════════════════════════════════════════════════════════
#
# The 10 tables above each implement ONE detective control in isolation.
# This block promotes "Control" to a first-class entity so the security
# surface can answer the question every GRC platform (Vanta / Drata /
# SailPoint) is built around: "are we compliant — and against which
# framework clause?".
#
# Model: a Control has a status (pass/fail/stale/…), an owner, evidence,
# and is mapped to N framework requirements. A framework's readiness is
# a projection of its requirements' controls' statuses — it stores NO
# duplicate data. SOC2 / ISO27001 / PCI / 等保 reuse the SAME underlying
# controls via the crosswalk, so a control built once is cited everywhere.


class ComplianceFramework(Base):
    """A compliance standard Helix evidences against (SOC2, ISO27001, …)."""

    __tablename__ = "compliance_frameworks"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Stable machine key: 'soc2' / 'iso27001' / 'pci_dss' / 'mlps'.
    key: Mapped[str] = mapped_column(String(32), nullable=False)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    version: Mapped[str | None] = mapped_column(String(32))
    description: Mapped[str | None] = mapped_column(Text)
    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("key", name="uq_compliance_framework_key"),
        Index("idx_compliance_framework_enabled", "enabled"),
    )

    def __repr__(self) -> str:
        return f"<ComplianceFramework {self.key} v={self.version}>"


class ComplianceRequirement(Base):
    """One clause / control objective within a framework (e.g. SOC2 CC6.1)."""

    __tablename__ = "compliance_requirements"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    framework_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("compliance_frameworks.id", ondelete="CASCADE"),
        nullable=False,
    )
    # Official clause reference as auditors cite it: 'CC6.1', 'A.5.15',
    # 'Req-8.3.6', '8.1.4.2'. Unique per framework.
    ref: Mapped[str] = mapped_column(String(64), nullable=False)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    description: Mapped[str | None] = mapped_column(Text)
    # Grouping within the framework ('Logical & Physical Access Controls',
    # 'Organizational controls', …) for rendering the framework tree.
    category: Mapped[str | None] = mapped_column(String(128))
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "framework_id", "ref", name="uq_compliance_req_framework_ref",
        ),
        Index("idx_compliance_req_framework", "framework_id"),
    )

    def __repr__(self) -> str:
        return f"<ComplianceRequirement {self.ref} {self.title!r}>"


class ComplianceControl(Base):
    """A first-class control. Its status is computed from a backing signal
    (a scheduled-job-fed detective control, an access-certification
    campaign, a SoD sweep, or a manual attestation) and projected onto
    every framework requirement it's mapped to.
    """

    __tablename__ = "compliance_controls"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Stable machine key, e.g. 'mfa_coverage', 'privileged_access_review',
    # 'secret_rotation'. Also the evaluator-registry key.
    key: Mapped[str] = mapped_column(String(64), nullable=False)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    # 'identity' | 'access' | 'posture' | 'data_protection' | 'logging'
    # | 'crypto' | 'vuln_mgmt' — used to group the catalog page.
    category: Mapped[str] = mapped_column(String(32), nullable=False)
    description: Mapped[str | None] = mapped_column(Text)
    # 'scheduled_job' | 'cert_campaign' | 'sod' | 'manual'.
    signal_type: Mapped[str] = mapped_column(
        String(32), nullable=False, default="scheduled_job",
    )
    # For signal_type='scheduled_job' this is the scheduled_jobs.handler_key
    # whose freshness feeds the control's stale/fresh state.
    signal_ref: Mapped[str | None] = mapped_column(String(128))
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    # Optional evaluator tuning, e.g. {"min_coverage_pct": 95}.
    config: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("key", name="uq_compliance_control_key"),
        Index("idx_compliance_control_category", "category"),
        Index("idx_compliance_control_enabled", "enabled"),
    )

    def __repr__(self) -> str:
        return f"<ComplianceControl {self.key} ({self.category})>"


class ComplianceControlRequirement(Base):
    """Crosswalk: which framework requirement(s) a control satisfies (M:N)."""

    __tablename__ = "compliance_control_requirements"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    control_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("compliance_controls.id", ondelete="CASCADE"),
        nullable=False,
    )
    requirement_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("compliance_requirements.id", ondelete="CASCADE"),
        nullable=False,
    )
    note: Mapped[str | None] = mapped_column(String(255))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "control_id", "requirement_id",
            name="uq_compliance_ctrl_req",
        ),
        Index("idx_compliance_ctrl_req_control", "control_id"),
        Index("idx_compliance_ctrl_req_requirement", "requirement_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<ComplianceControlRequirement c={self.control_id} "
            f"r={self.requirement_id}>"
        )


class ComplianceControlStatus(Base):
    """Latest evaluated status per control (continuous control monitoring).

    Upserted by the ``compliance.control_status_recompute`` cron. One row
    per control (``UNIQUE(control_id)``). Keeps the posture dashboard fast
    and gives the recompute job an idempotent write target. History/trend
    is a deliberate later add — this is the current-state snapshot.
    """

    __tablename__ = "compliance_control_status"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    control_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("compliance_controls.id", ondelete="CASCADE"),
        nullable=False,
    )
    # 'pass' | 'fail' | 'stale' | 'not_configured' | 'unknown'.
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="unknown",
    )
    # 0-100 where meaningful (e.g. MFA coverage %); NULL for binary controls.
    score: Mapped[int | None] = mapped_column(Integer)
    # Human-readable one-liner: "142/150 enrolled (94.7%)".
    summary: Mapped[str | None] = mapped_column(String(512))
    detail: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    evaluated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    # Freshness of the backing signal (e.g. feeding job's last_run_at).
    source_fresh_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "control_id", name="uq_compliance_control_status_control",
        ),
        Index("idx_compliance_control_status_status", "status"),
    )

    def __repr__(self) -> str:
        return (
            f"<ComplianceControlStatus control={self.control_id} "
            f"status={self.status} score={self.score}>"
        )


# ════════════════════════════════════════════════════════════════════
# 12. Access certification (generalized periodic access review)
# ════════════════════════════════════════════════════════════════════
#
# The general engine for periodic access reviews — the IGA primitive
# (SailPoint/Saviynt) that SOC2 CC6.3 / ISO A.5.15 & A.5.18 require:
# designated reviewers attest, on a recurring cycle, that each user's
# access is still appropriate, and unneeded access is revoked.
#
# A campaign snapshots a set of (person, entitlement) rows scoped by
# ``campaign_type`` + ``scope_config`` (e.g. "every active software seat
# held by someone in department X"), assigns each row a reviewer (the
# holder's manager), and tracks the keep/revoke decision + revocation.
# The privileged-role review folds into this as ``campaign_type=
# 'privileged_role'`` in a follow-up PR; until then it runs alongside.


class AccessCertification(Base):
    """One access-review campaign."""

    __tablename__ = "access_certifications"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    label: Mapped[str] = mapped_column(String(128), nullable=False)
    # 'software_seat' | 'cloud_grant' | 'application' | 'privileged_role'
    # | 'custom' — selects the snapshot builder.
    campaign_type: Mapped[str] = mapped_column(String(32), nullable=False)
    # Builder input, e.g. {"department": "Security & Compliance"}.
    scope_config: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="in_progress",
    )  # in_progress | completed | cancelled
    started_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    due_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_by_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    snapshot_item_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0,
    )
    decided_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0,
    )
    revoked_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0,
    )
    notes: Mapped[str | None] = mapped_column(Text)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_access_cert_status", "status"),
        Index("idx_access_cert_type", "campaign_type"),
        Index("idx_access_cert_due", "due_at"),
    )

    def __repr__(self) -> str:
        return (
            f"<AccessCertification {self.label} {self.campaign_type} "
            f"status={self.status}>"
        )


class AccessCertificationItem(Base):
    """One (person, entitlement) row inside a campaign, with its decision."""

    __tablename__ = "access_certification_items"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    certification_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("access_certifications.id", ondelete="CASCADE"),
        nullable=False,
    )
    person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    holder_email: Mapped[str | None] = mapped_column(String(255))
    holder_display_name: Mapped[str | None] = mapped_column(String(255))
    # 'software_seat' | 'cloud_grant' | 'privileged_role' | …
    access_type: Mapped[str] = mapped_column(String(32), nullable=False)
    # The underlying entitlement id (seat allocation id, grant id, …).
    access_ref: Mapped[str | None] = mapped_column(String(128))
    # Human-readable entitlement, e.g. "Jira / Standard plan".
    access_label: Mapped[str] = mapped_column(String(255), nullable=False)
    resource_name: Mapped[str | None] = mapped_column(String(255))
    assigned_reviewer_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    decided_at_deadline: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    decision: Mapped[str] = mapped_column(
        String(16), nullable=False, default="pending",
    )  # pending | keep | revoke
    decided_by_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    decision_note: Mapped[str | None] = mapped_column(Text)
    # none | pending | completed | failed | manual_required
    # (revoke-enforcement state; manual_required = no automated path, a
    #  human must remove the access — e.g. an external privileged_role)
    revocation_status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="none",
    )
    revocation_completed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    revocation_trace_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    escalated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    snapshot_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    raw_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_access_cert_item_cert_decision",
              "certification_id", "decision"),
        Index("idx_access_cert_item_person", "person_id"),
        Index("idx_access_cert_item_reviewer", "assigned_reviewer_person_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AccessCertificationItem {self.access_label} "
            f"holder={self.holder_email} decision={self.decision}>"
        )


# ════════════════════════════════════════════════════════════════════
# 13. Periodic self-check (周期性自我检查)
# ════════════════════════════════════════════════════════════════════
#
# The "等保、认证、审计 → 周期性自我检查" workbook, modelled as a first-class
# domain instead of a wiki table. A *category* is a recurring check line
# (bastion-host log audit, anti-virus sweep, full-disk-encryption check,
# …), each on its own cadence (monthly / quarterly / semiannual / annual).
# A scheduled job opens one *record* per category per period; the assigned
# checker performs the check, attaches evidence (screenshots / log exports)
# and signs off. The records are the auditable台账 — exportable to xlsx and
# fed into the GRC control catalog (control key ``periodic_self_check``) so
# on-time completion shows up as continuous control monitoring.
#
# person_id-shaped columns are plain id columns (NO FK to identity_persons)
# — the same TiDB-FK-safe convention the recent gamification / wrapped
# tables follow. Intra-feature parent refs carry real FKs (same charset).


class AuditorSelfCheckCategory(Base):
    """One recurring self-check line (a "类目")."""

    __tablename__ = "auditor_self_check_categories"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Stable machine key, e.g. 'bastion_log_audit'. Drives idempotent seed
    # and lets a record survive a display-name rename.
    key: Mapped[str] = mapped_column(String(64), nullable=False)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    # Optional grouping for the catalog tree, e.g. '日志审计'.
    group_label: Mapped[str | None] = mapped_column(String(64))
    # 'monthly' | 'quarterly' | 'semiannual' | 'annual'.
    periodicity: Mapped[str] = mapped_column(
        String(16), nullable=False, default="quarterly",
    )
    # The check item / what to verify, e.g. '日志是否有异常操作'.
    check_item: Mapped[str] = mapped_column(Text, nullable=False)
    # Longer how-to guidance (what counts as pass, where to look).
    guidance: Mapped[str | None] = mapped_column(Text)
    # Default assignee for newly-opened records (NULL → falls to admins).
    default_checker_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    requires_attachment: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True,
    )
    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("key", name="uq_auditor_self_check_category_key"),
        Index("idx_auditor_self_check_cat_enabled", "enabled", "sort_order"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorSelfCheckCategory {self.key} "
            f"periodicity={self.periodicity} enabled={self.enabled}>"
        )


class AuditorSelfCheckRecord(Base):
    """One check occurrence for a (category, period). The auditable台账 row."""

    __tablename__ = "auditor_self_check_records"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    category_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_self_check_categories.id", ondelete="CASCADE"),
        nullable=False,
    )
    period_year: Mapped[int] = mapped_column(Integer, nullable=False)
    # 'Q1'..'Q4' | 'H1'..'H2' | 'M01'..'M12' | 'FY'.
    period_label: Mapped[str] = mapped_column(String(16), nullable=False)
    period_start: Mapped[datetime] = mapped_column(Date, nullable=False)
    period_end: Mapped[datetime] = mapped_column(Date, nullable=False)
    # When the check should be completed by (= end of period).
    due_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    # pending | completed | skipped.
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="pending",
    )
    # Snapshot of the category's check_item at open time, so editing the
    # category later does not rewrite历史 records.
    check_item_snapshot: Mapped[str] = mapped_column(Text, nullable=False)
    # normal | issue_found | not_applicable (filled on completion).
    outcome: Mapped[str | None] = mapped_column(String(16))
    assigned_checker_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    checker_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    remark: Mapped[str | None] = mapped_column(Text)
    # Denormalised count for fast台账 list rendering.
    attachment_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0,
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "category_id", "period_year", "period_label",
            name="uq_auditor_self_check_record_period",
        ),
        Index("idx_auditor_self_check_rec_status_due", "status", "due_at"),
        Index("idx_auditor_self_check_rec_category", "category_id"),
        Index("idx_auditor_self_check_rec_checker", "checker_person_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorSelfCheckRecord cat={self.category_id} "
            f"{self.period_year}/{self.period_label} status={self.status}>"
        )


class AuditorSelfCheckAttachment(Base):
    """An evidence file (screenshot / log export) attached to a record."""

    __tablename__ = "auditor_self_check_attachments"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    record_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_self_check_records.id", ondelete="CASCADE"),
        nullable=False,
    )
    # FULL S3 object key (with prefix) as returned by presign_put. Never
    # store the URL — mint a fresh presigned GET on download.
    object_key: Mapped[str] = mapped_column(String(512), nullable=False)
    filename: Mapped[str] = mapped_column(String(255), nullable=False)
    content_type: Mapped[str] = mapped_column(String(128), nullable=False)
    size_bytes: Mapped[int | None] = mapped_column(BigInteger)
    uploaded_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    uploaded_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_self_check_att_record", "record_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorSelfCheckAttachment record={self.record_id} "
            f"file={self.filename!r}>"
        )


# ════════════════════════════════════════════════════════════════════
# 14. 信息安全自评估 (semiannual ISMS self-assessment report)
# ════════════════════════════════════════════════════════════════════
#
# The 「等保、认证、审计」second pillar: the formal semiannual ISMS report.
# A *cycle* covers one half-year; instantiating it materialises one
# *section* per the code template (``assessment_catalog``), auto-filling
# the data tables from live Helix tables (``assessment_data``). The
# auditor confirms/edits the narrative + 评估结论 (AI-drafted in PR3),
# then finalizes → export (PR3). The *measures* tracker is the evergreen
# 整改/措施 worklist that feeds 管理层承诺. person columns carry NO FK to
# identity_persons (TiDB-FK-safe); cycle→section is a real intra-feature FK.


class AuditorAssessmentCycle(Base):
    """One semiannual self-assessment report instance."""

    __tablename__ = "auditor_assessment_cycles"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    template_key: Mapped[str] = mapped_column(
        String(64), nullable=False, default="isms_default",
    )
    # e.g. '2026-H1' — one assessment per half-year (unique).
    period_key: Mapped[str] = mapped_column(String(16), nullable=False)
    period_start: Mapped[datetime] = mapped_column(Date, nullable=False)
    period_end: Mapped[datetime] = mapped_column(Date, nullable=False)
    report_date: Mapped[datetime | None] = mapped_column(Date)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    # draft | finalized.
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="draft",
    )
    intro_narrative: Mapped[str | None] = mapped_column(Text)
    conclusion_narrative: Mapped[str | None] = mapped_column(Text)
    # compliant | partial | non_compliant | not_assessed (overall verdict).
    overall_conclusion: Mapped[str] = mapped_column(
        String(16), nullable=False, default="not_assessed",
    )
    created_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    finalized_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    finalized_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("period_key", name="uq_auditor_assessment_period"),
        Index("idx_auditor_assessment_status", "status"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorAssessmentCycle {self.period_key} status={self.status}>"
        )


class AuditorAssessmentSection(Base):
    """One materialised report section within a cycle."""

    __tablename__ = "auditor_assessment_sections"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    cycle_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_assessment_cycles.id", ondelete="CASCADE"),
        nullable=False,
    )
    section_key: Mapped[str] = mapped_column(String(64), nullable=False)
    parent_key: Mapped[str | None] = mapped_column(String(64))
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    has_table: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    narrative: Mapped[str | None] = mapped_column(Text)
    # compliant | partial | non_compliant | not_assessed.
    conclusion: Mapped[str] = mapped_column(
        String(16), nullable=False, default="not_assessed",
    )
    conclusion_applicable: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True,
    )
    # Display columns [(key,label)] for the table header (from catalog).
    table_columns: Mapped[list[Any] | None] = mapped_column(JSON)
    # Auto-pulled data rows (list of dict) — the editable starting point.
    data_table: Mapped[list[Any] | None] = mapped_column(JSON)
    auto_source: Mapped[str | None] = mapped_column(String(64))
    # Last raw aggregator snapshot (summary + numbers) for provenance.
    auto_data: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    auto_refreshed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # Optional crosswalk to a GRC control whose status informs this section.
    control_key: Mapped[str | None] = mapped_column(String(64))
    ai_drafted: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "cycle_id", "section_key",
            name="uq_auditor_assessment_section",
        ),
        Index("idx_auditor_assessment_section_cycle", "cycle_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorAssessmentSection cycle={self.cycle_id} "
            f"key={self.section_key}>"
        )


class AuditorAssessmentMeasure(Base):
    """Evergreen 整改/措施 worklist item (the 安全自评估 measures tracker)."""

    __tablename__ = "auditor_assessment_measures"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    item: Mapped[str] = mapped_column(String(255), nullable=False)
    content: Mapped[str | None] = mapped_column(Text)
    measures: Mapped[str | None] = mapped_column(Text)
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # pending 待开展 | in_progress 进行中 | done 已完成.
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="pending",
    )
    note: Mapped[str | None] = mapped_column(Text)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_assessment_measure_status", "status", "archived"),
    )

    def __repr__(self) -> str:
        return f"<AuditorAssessmentMeasure {self.item!r} status={self.status}>"


# ════════════════════════════════════════════════════════════════════
# 15. 安全问卷 — 甲方侧供应商安全评估 (vendor security assessment / TPRM)
# ════════════════════════════════════════════════════════════════════
#
# The 甲方 half of the two-sided 安全问卷 feature: Helix (as the customer)
# assesses a cloud provider / vendor's security posture. A reusable
# bilingual *template* (seeded from ISO/IEC 27001:2022 Annex A + ISO/IEC
# 27701) holds *template questions*. An *assessment* instantiates one
# template for one vendor: creating it materialises one *answer* row per
# template question, SNAPSHOTTING the question text/type/weight so a later
# template edit never rewrites a historical assessment (same discipline as
# self-check's ``check_item_snapshot``). Evidence files (the vendor's
# ISO/SOC 2 certs, pentest reports, or the returned filled workbook)
# attach via S3 presign. On import of the vendor's filled xlsx the answers
# are weighted-auto-scored → a risk rating; a low-scoring / concerning
# answer can be promoted to an ``auditor_findings`` row for remediation
# tracking. A daily cron reminds the owner before ``next_review_date``.
#
# person/supplier columns carry NO FK (TiDB-FK-safe, and the supplier
# lives in the procurement domain — cross-domain refs are soft ids).
# Intra-feature parent refs (template→question, assessment→answer /
# attachment) carry real same-charset FKs.


class AuditorSqTemplate(Base):
    """A reusable bilingual questionnaire template (甲方侧下发给供应商的问卷)."""

    __tablename__ = "auditor_sq_templates"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Stable machine key, e.g. 'iso27001_full'. Drives idempotent seed.
    key: Mapped[str] = mapped_column(String(64), nullable=False)
    name_zh: Mapped[str] = mapped_column(String(255), nullable=False)
    name_en: Mapped[str] = mapped_column(String(255), nullable=False)
    description_zh: Mapped[str | None] = mapped_column(Text)
    description_en: Mapped[str | None] = mapped_column(Text)
    # 'vendor' today; kept for a future inbound/other kind without a
    # migration. All PR1 templates are 'vendor'.
    kind: Mapped[str] = mapped_column(String(16), nullable=False, default="vendor")
    # Depth of the template: 'full' | 'lite' (or NULL for a custom one).
    tier: Mapped[str | None] = mapped_column(String(16))
    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    # System (seeded) templates cannot be deleted, only cloned/disabled.
    is_system: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("key", name="uq_auditor_sq_template_key"),
        Index("idx_auditor_sq_template_enabled", "enabled", "sort_order"),
    )

    def __repr__(self) -> str:
        return f"<AuditorSqTemplate {self.key} tier={self.tier} v{self.version}>"


class AuditorSqTemplateQuestion(Base):
    """One question inside a template. ``code`` is the round-trip key that
    survives the xlsx export→vendor-fills→import cycle."""

    __tablename__ = "auditor_sq_template_questions"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    template_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_sq_templates.id", ondelete="CASCADE"),
        nullable=False,
    )
    # Stable per-template question code, e.g. 'A.5.1'. The hidden machine
    # column in the exported workbook so a filled row re-binds on import.
    code: Mapped[str] = mapped_column(String(32), nullable=False)
    section_key: Mapped[str] = mapped_column(String(64), nullable=False)
    section_zh: Mapped[str] = mapped_column(String(255), nullable=False)
    section_en: Mapped[str] = mapped_column(String(255), nullable=False)
    text_zh: Mapped[str] = mapped_column(Text, nullable=False)
    text_en: Mapped[str] = mapped_column(Text, nullable=False)
    guidance_zh: Mapped[str | None] = mapped_column(Text)
    guidance_en: Mapped[str | None] = mapped_column(Text)
    # yes_no | yes_no_na | single_choice | multi_choice | scale_1_5 | text.
    answer_type: Mapped[str] = mapped_column(String(24), nullable=False)
    # For choice types: [{"value","label_zh","label_en","score"}]. score is
    # an integer 0..weight contribution for single_choice; NULL otherwise.
    options: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON)
    weight: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    # For yes_no/yes_no_na auto-score: which answer earns full weight.
    expected_answer: Mapped[str | None] = mapped_column(String(64))
    requires_evidence: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False,
    )
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "template_id", "code", name="uq_auditor_sq_template_question_code",
        ),
        Index("idx_auditor_sq_tmpl_question_tmpl", "template_id", "sort_order"),
    )

    def __repr__(self) -> str:
        return f"<AuditorSqTemplateQuestion {self.code} type={self.answer_type}>"


class AuditorVendorAssessment(Base):
    """One vendor security assessment instance (供应商安全评估单)."""

    __tablename__ = "auditor_vendor_assessments"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Soft ref to the source template (no FK — an assessment must survive a
    # template deletion; the answers carry snapshots).
    template_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    template_key_snapshot: Mapped[str | None] = mapped_column(String(64))
    template_name_snapshot: Mapped[str | None] = mapped_column(String(255))
    # Cross-domain soft ref to procurement_suppliers.id (no FK); vendor_name
    # is the always-present display fallback (cloud providers may not be in
    # the procurement supplier registry).
    supplier_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    vendor_name: Mapped[str] = mapped_column(String(255), nullable=False)
    vendor_contact_name: Mapped[str | None] = mapped_column(String(128))
    vendor_contact_email: Mapped[str | None] = mapped_column(String(255))
    # critical | high | medium | low — data-sensitivity tier; drives the
    # default review cadence and the recommended template depth.
    tier: Mapped[str] = mapped_column(String(16), nullable=False, default="medium")
    # draft | sent | answered | completed | archived | cancelled.
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="draft",
    )
    score_earned: Mapped[int | None] = mapped_column(Integer)
    score_max: Mapped[int | None] = mapped_column(Integer)
    score_pct: Mapped[int | None] = mapped_column(Integer)
    # low | medium | high | critical (residual risk verdict).
    risk_rating: Mapped[str | None] = mapped_column(String(16))
    rating_overridden: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False,
    )
    summary: Mapped[str | None] = mapped_column(Text)
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    reviewer_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    answered_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    next_review_date: Mapped[datetime | None] = mapped_column(Date)
    review_period_months: Mapped[int | None] = mapped_column(Integer)
    # Which staged review reminder was last sent ('30'|'14'|'due'|'overdue')
    # — makes the daily reminder cron idempotent per stage.
    review_reminder_stage: Mapped[str | None] = mapped_column(String(16))
    review_reminder_sent_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    created_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_vendor_assess_status", "status"),
        Index("idx_auditor_vendor_assess_review", "next_review_date"),
        Index("idx_auditor_vendor_assess_supplier", "supplier_id"),
        Index("idx_auditor_vendor_assess_owner", "owner_person_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorVendorAssessment {self.vendor_name!r} "
            f"status={self.status} rating={self.risk_rating}>"
        )


class AuditorVendorAssessmentAnswer(Base):
    """One question+answer row inside an assessment (snapshotted from the
    template at creation time)."""

    __tablename__ = "auditor_vendor_assessment_answers"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    assessment_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_vendor_assessments.id", ondelete="CASCADE"),
        nullable=False,
    )
    question_code: Mapped[str] = mapped_column(String(32), nullable=False)
    section_key: Mapped[str] = mapped_column(String(64), nullable=False)
    section_zh: Mapped[str] = mapped_column(String(255), nullable=False)
    section_en: Mapped[str] = mapped_column(String(255), nullable=False)
    text_zh: Mapped[str] = mapped_column(Text, nullable=False)
    text_en: Mapped[str] = mapped_column(Text, nullable=False)
    guidance_zh: Mapped[str | None] = mapped_column(Text)
    guidance_en: Mapped[str | None] = mapped_column(Text)
    answer_type: Mapped[str] = mapped_column(String(24), nullable=False)
    options: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON)
    weight: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    expected_answer: Mapped[str | None] = mapped_column(String(64))
    requires_evidence: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False,
    )
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    # The vendor's answer: a scalar ('yes'/'no'/'na'/option value/'1'..'5')
    # or a JSON array (multi_choice) or free text.
    answer_value: Mapped[str | None] = mapped_column(Text)
    answer_comment: Mapped[str | None] = mapped_column(Text)
    # Weighted score for this answer: auto_score / max_score. max_score is
    # NULL when the question does not participate in scoring (text/multi) or
    # was answered N/A (excluded from the denominator).
    auto_score: Mapped[int | None] = mapped_column(Integer)
    max_score: Mapped[int | None] = mapped_column(Integer)
    is_scorable: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False,
    )
    # ok | concern | finding — the auditor's per-answer review verdict.
    review_flag: Mapped[str | None] = mapped_column(String(16))
    reviewer_note: Mapped[str | None] = mapped_column(Text)
    # Soft ref to the auditor_findings row this answer was promoted to.
    finding_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    attachment_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0,
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "assessment_id", "question_code",
            name="uq_auditor_vendor_answer_code",
        ),
        Index("idx_auditor_vendor_answer_assess", "assessment_id", "sort_order"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorVendorAssessmentAnswer {self.question_code} "
            f"score={self.auto_score}/{self.max_score}>"
        )


class AuditorVendorAssessmentAttachment(Base):
    """An evidence file attached to an assessment or a single answer."""

    __tablename__ = "auditor_vendor_assessment_attachments"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    assessment_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_vendor_assessments.id", ondelete="CASCADE"),
        nullable=False,
    )
    # NULL = assessment-level evidence (a cert / the returned workbook);
    # set = evidence for one specific answer. Soft ref (no FK) so deleting
    # an answer never blocks; orphan cleanup is by assessment cascade.
    answer_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # evidence | response_file | cert.
    kind: Mapped[str] = mapped_column(String(24), nullable=False, default="evidence")
    object_key: Mapped[str] = mapped_column(String(512), nullable=False)
    filename: Mapped[str] = mapped_column(String(255), nullable=False)
    content_type: Mapped[str] = mapped_column(String(128), nullable=False)
    size_bytes: Mapped[int | None] = mapped_column(BigInteger)
    uploaded_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    uploaded_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_vendor_att_assess", "assessment_id"),
        Index("idx_auditor_vendor_att_answer", "answer_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorVendorAssessmentAttachment assess={self.assessment_id} "
            f"kind={self.kind} file={self.filename!r}>"
        )


# ════════════════════════════════════════════════════════════════════
# 16. 安全问卷 — 乙方侧客户问卷 + 答案库 + 证据文档库
# ════════════════════════════════════════════════════════════════════
#
# The 乙方 half of the 安全问卷 feature: when a CUSTOMER sends US a security
# questionnaire to fill. Three co-operating pieces:
#
#   * Answer library (``auditor_answer_library_entries``) — the reusable,
#     bilingual, owner-curated Q&A knowledge base (the Conveyor / Whistic
#     "answer library" pattern). Every answered inbound question can be
#     BACKFILLED here, so the library compounds and the next questionnaire
#     is faster. Each entry links the evidence documents that support it.
#   * Evidence document register (``auditor_evidence_documents``) — OUR
#     own security artifacts customers ask for (DR plan, ISO 27701 / SOC 2
#     certs, pentest reports). S3-backed, shareable-flag + validity date.
#   * Inbound questionnaires (``auditor_inbound_questionnaires`` +
#     ``_questions`` + ``_versions``) — a customer's questionnaire: intake
#     (customer / due date / owner / original file), per-question rows the
#     Bailian AI drafts by matching the answer library, human review, then
#     export of the filled workbook with each sent version archived for
#     accountability.
#
# person columns carry NO FK (TiDB-FK-safe). Intra-feature parent refs
# (questionnaire→question / →version) carry real same-charset FKs. Soft
# refs (matched_library_entry_id, source_doc_ids) are plain ids.


class AuditorAnswerLibraryEntry(Base):
    """One reusable bilingual Q&A entry in the answer library."""

    __tablename__ = "auditor_answer_library_entries"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Grouping, e.g. 'access_control' / 'encryption' / 'bcp' / 'privacy'.
    category: Mapped[str] = mapped_column(String(64), nullable=False, default="general")
    question_zh: Mapped[str] = mapped_column(Text, nullable=False)
    question_en: Mapped[str] = mapped_column(Text, nullable=False)
    answer_zh: Mapped[str] = mapped_column(Text, nullable=False)
    answer_en: Mapped[str] = mapped_column(Text, nullable=False)
    tags: Mapped[list[str] | None] = mapped_column(JSON)
    #: How employees actually ask for this, newline-separated (migration 594).
    #:
    #: Every entry here is written in the register of a CUSTOMER SECURITY
    #: QUESTIONNAIRE — 「贵公司是否部署数据防泄漏(DLP)方案?」 — while an
    #: employee asks 「我能用个人电脑写公司代码吗」. Semantic retrieval bridges
    #: part of that gap and cannot bridge vocabulary: 「个人电脑」 and 「BYOD」
    #: are not especially close in vector space, and only the second is in the
    #: text. This column is where the employee's own words go.
    #:
    #: Deliberately NOT ``tags``: those are the security team's classification
    #: keywords, and mixing employee phrasing into them would pollute both the
    #: category filter and their meaning inside the embedding text.
    #:
    #: The only writer is the knowledge loop's proposal queue, and the only
    #: source is the gap ledger — so every word here was typed by a real
    #: employee who did not get an answer.
    employee_phrasings: Mapped[str | None] = mapped_column(Text, nullable=True)
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # active | archived.
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
    usage_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    # Whether an ordinary employee may read this entry in 安全 FAQ
    # (``/me/security-faq``). Defaults to visible — the library holds our
    # own posture, not customer material — but the switch exists so the
    # security team can retract one entry without archiving it, which
    # would also pull it out of the questionnaire flow.
    employee_visible: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="1",
    )
    # Pinned above the search box on the employee page.
    is_featured: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="0",
    )
    # Employee-side counters. Deliberately NOT folded into
    # ``usage_count``: that one means "reused when answering a customer"
    # and drives the review cadence the security team reads.
    employee_view_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    employee_helpful_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    employee_unhelpful_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    last_reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    review_interval_months: Mapped[int] = mapped_column(
        Integer, nullable=False, default=12,
    )
    # Soft refs to auditor_evidence_documents.id backing this answer.
    source_doc_ids: Mapped[list[str] | None] = mapped_column(JSON)
    # Which controls this answer's claim rests on. The library already
    # tracked "when is it due for review"; it had no way to know that the
    # FACT changed. An answer saying "MFA is enforced on all privileged
    # accounts" would otherwise sit quietly until its next review cycle
    # after the MFA control went red — and be sent to a customer as-is.
    backing_control_keys: Mapped[list[str] | None] = mapped_column(JSON)
    stale_reason: Mapped[str | None] = mapped_column(String(64))
    stale_since: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # Semantic-retrieval vector for 安全 FAQ (migration 561). Portable JSON
    # float array — NOT TiDB VECTOR — so local MySQL tests run unchanged and
    # ranking is Python cosine over a few hundred curated rows.
    # ``embedding_model`` records the producing model so a model switch can
    # re-embed selectively; NULL embedding simply means "semantic retrieval
    # unavailable for this row", and lexical search still answers.
    # none_as_null: store Python None as SQL NULL (not JSON 'null'), so
    # ``embedding IS NULL`` reliably means "not embedded yet".
    embedding: Mapped[list[float] | None] = mapped_column(
        JSON(none_as_null=True), nullable=True,
    )
    embedding_model: Mapped[str | None] = mapped_column(String(64), nullable=True)
    embedded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_answer_lib_category", "category", "status"),
        Index("idx_auditor_answer_lib_owner", "owner_person_id"),
        Index(
            "idx_auditor_answer_lib_employee",
            "employee_visible", "status", "category",
        ),
    )

    def __repr__(self) -> str:
        return f"<AuditorAnswerLibraryEntry {self.category} {self.question_en[:40]!r}>"


class AuditorSecurityFaqGap(Base):
    """A question employees asked that the answer library could not answer.

    One row per normalised query, not per ask — thirty people asking the
    same thing is one gap worth ``ask_count = 30``, which is the number
    that decides what gets written next. An admin's resolution is never
    overwritten by a later ask (see the 559 migration docstring), so a
    dismissed gap stays dismissed; ``last_asked_at`` moving past
    ``resolved_at`` is how the console surfaces "we answered this and
    they are still asking".
    """

    __tablename__ = "auditor_security_faq_gaps"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    query_norm: Mapped[str] = mapped_column(String(255), nullable=False)
    query_sample: Mapped[str] = mapped_column(Text, nullable=False)
    lang: Mapped[str] = mapped_column(String(8), nullable=False, default="zh")
    ask_count: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    first_asked_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    last_asked_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    last_asker_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # open | answered | dismissed
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="open")
    resolved_entry_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    resolved_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    note: Mapped[str | None] = mapped_column(String(500))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("query_norm", name="uq_auditor_security_faq_gaps_query"),
        Index("ix_auditor_security_faq_gaps_status", "status", "ask_count"),
    )

    def __repr__(self) -> str:
        return f"<AuditorSecurityFaqGap {self.query_norm[:40]!r} x{self.ask_count}>"


class AuditorSecurityFaqVote(Base):
    """One employee's 有帮助 / 没帮助 verdict on one FAQ entry.

    The entry carries denormalised counters for list rendering; this table
    is what keeps them honest. Unique on (entry, person), so a vote can be
    changed but never stacked.
    """

    __tablename__ = "auditor_security_faq_votes"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    entry_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    person_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    helpful: Mapped[bool] = mapped_column(Boolean, nullable=False)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("entry_id", "person_id", name="uq_auditor_security_faq_votes"),
    )

    def __repr__(self) -> str:
        return f"<AuditorSecurityFaqVote {self.entry_id} {self.person_id} {self.helpful}>"


class AuditorEvidenceDocument(Base):
    """One of OUR security artifacts customers ask for (cert / policy / report)."""

    __tablename__ = "auditor_evidence_documents"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    # certificate | policy | report | dr_plan | other.
    category: Mapped[str] = mapped_column(String(32), nullable=False, default="other")
    description: Mapped[str | None] = mapped_column(Text)
    object_key: Mapped[str] = mapped_column(String(512), nullable=False)
    filename: Mapped[str] = mapped_column(String(255), nullable=False)
    content_type: Mapped[str] = mapped_column(String(128), nullable=False)
    size_bytes: Mapped[int | None] = mapped_column(BigInteger)
    version: Mapped[str | None] = mapped_column(String(32))
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    valid_until: Mapped[datetime | None] = mapped_column(Date)
    # Whether this document may be shared with a customer.
    is_shareable: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    # active | archived.
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
    uploaded_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_evidence_doc_category", "category", "status"),
        Index("idx_auditor_evidence_doc_valid", "valid_until"),
    )

    def __repr__(self) -> str:
        return f"<AuditorEvidenceDocument {self.category} {self.title!r}>"


class AuditorInboundQuestionnaire(Base):
    """A customer's security questionnaire that WE must fill in."""

    __tablename__ = "auditor_inbound_questionnaires"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    customer_name: Mapped[str] = mapped_column(String(255), nullable=False)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    due_date: Mapped[datetime | None] = mapped_column(Date)
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # received | answering | review | completed | archived | cancelled.
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="received")
    # The original customer file, if uploaded (S3 object key).
    source_object_key: Mapped[str | None] = mapped_column(String(512))
    source_filename: Mapped[str | None] = mapped_column(String(255))
    notes: Mapped[str | None] = mapped_column(Text)
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    exported_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Staged due reminder bookkeeping (idempotent per stage).
    reminder_stage: Mapped[str | None] = mapped_column(String(16))
    reminder_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_inbound_status", "status"),
        Index("idx_auditor_inbound_due", "due_date"),
        Index("idx_auditor_inbound_owner", "owner_person_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorInboundQuestionnaire {self.customer_name!r} "
            f"status={self.status}>"
        )


class AuditorInboundQuestion(Base):
    """One question inside a customer's inbound questionnaire."""

    __tablename__ = "auditor_inbound_questions"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    questionnaire_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_inbound_questionnaires.id", ondelete="CASCADE"),
        nullable=False,
    )
    seq: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    section: Mapped[str | None] = mapped_column(String(255))
    # The customer's question, as received (single language).
    question_text: Mapped[str] = mapped_column(Text, nullable=False)
    # Soft ref to the answer-library entry matched by AI / human.
    matched_library_entry_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    match_confidence: Mapped[int | None] = mapped_column(Integer)
    # The drafted / final answer we will send.
    answer_text: Mapped[str | None] = mapped_column(Text)
    # pending | drafted | reviewed.
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")
    reviewed_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_inbound_q_questionnaire", "questionnaire_id", "seq"),
    )

    def __repr__(self) -> str:
        return f"<AuditorInboundQuestion seq={self.seq} status={self.status}>"


class AuditorInboundVersion(Base):
    """An archived exported version of a filled inbound questionnaire.

    The accountability trail: what we sent to which customer, when — so a
    future dispute can reconstruct exactly what was promised.
    """

    __tablename__ = "auditor_inbound_versions"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    questionnaire_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("auditor_inbound_questionnaires.id", ondelete="CASCADE"),
        nullable=False,
    )
    version_no: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    object_key: Mapped[str] = mapped_column(String(512), nullable=False)
    filename: Mapped[str] = mapped_column(String(255), nullable=False)
    note: Mapped[str | None] = mapped_column(Text)
    exported_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    exported_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    __table_args__ = (
        UniqueConstraint(
            "questionnaire_id", "version_no",
            name="uq_auditor_inbound_version_no",
        ),
        Index("idx_auditor_inbound_version_q", "questionnaire_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorInboundVersion q={self.questionnaire_id} "
            f"v{self.version_no}>"
        )


# ════════════════════════════════════════════════════════════════════
# 13. Evidence queries — frozen, reproducible audit evidence
# ════════════════════════════════════════════════════════════════════


class AuditorEvidenceBundle(Base):
    """One frozen execution of one registered evidence query.

    An auditor asking "who held X access in May, who approved it, when
    was it revoked" needs an answer that is (a) deterministic, (b)
    re-checkable against the underlying records, and (c) *the same
    answer* when the external auditor re-opens it in November. A live
    query satisfies none of those: the underlying rows keep moving.

    So a run is frozen. The header row records exactly what was asked
    (``query_key`` + ``query_version`` + canonicalised ``params``) and
    a ``content_hash`` over the emitted rows. Re-running the same query
    with the same params later produces a NEW bundle; if the hash
    differs, the underlying data moved and that difference is itself
    evidence — which is why bundles are never updated in place.

    This is also what makes the output usable as SOC2 "Information
    Produced by Entity": the parameters, the query version and the row
    hash together let a reviewer establish completeness rather than
    take a screenshot on faith.
    """

    __tablename__ = "auditor_evidence_bundles"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    query_key: Mapped[str] = mapped_column(String(64), nullable=False)
    # Bumped by the handler whenever its logic changes shape. Two
    # bundles with the same params but different versions are NOT
    # comparable, and the UI must not present them as a diff.
    query_version: Mapped[str] = mapped_column(String(16), nullable=False)
    # Canonicalised (sorted-key) parameter JSON — the other half of the
    # reproducibility contract.
    params: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
    # Window the evidence covers, lifted out of params so the evidence
    # pack and the control-mapping cron can filter without parsing.
    period_start: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    period_end: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    row_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    # sha256 over the canonical serialisation of the emitted rows.
    content_hash: Mapped[str] = mapped_column(String(64), nullable=False)
    # Handler-declared caveats: sources that could not be read, ranges
    # predating a ledger's start, vendor gaps. Surfaced verbatim —
    # silent incompleteness is the failure mode that makes an evidence
    # tool worse than no tool.
    data_gaps: Mapped[list[Any] | None] = mapped_column(JSON)

    generated_by_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # ``manual`` (an auditor ran it) or ``scheduled`` (the monthly
    # pre-compile cron). Kept so a reviewer can tell "someone went
    # looking" apart from routine collection.
    generated_via: Mapped[str] = mapped_column(
        String(16), nullable=False, default="manual",
    )
    trace_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    duration_ms: Mapped[int | None] = mapped_column(Integer)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    __table_args__ = (
        Index("idx_auditor_evidence_bundle_query", "query_key", "created_at"),
        Index("idx_auditor_evidence_bundle_actor", "generated_by_person_id"),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorEvidenceBundle {self.query_key}@{self.query_version} "
            f"rows={self.row_count}>"
        )


class AuditorEvidenceBundleRow(Base):
    """One evidence row inside a frozen bundle, with its provenance.

    ``source_table`` / ``source_pk`` (or ``audit_log_id`` / ``trace_id``
    when the fact came from the audit trail) are what make the row
    checkable rather than assertable: the UI drills straight through to
    the originating record, and an LLM narrating the bundle later can
    only reference rows that exist here.

    Rows are stored rather than recomputed on read because the point of
    a bundle is that it does not change — see the header docstring.
    """

    __tablename__ = "auditor_evidence_bundle_rows"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    bundle_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    # Stable ordering within the bundle. The hash is computed over rows
    # in this order, so a reader can reproduce it.
    seq: Mapped[int] = mapped_column(Integer, nullable=False)

    # Who/what the row is about, denormalised for filtering and for the
    # CSV rendering in the evidence pack.
    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
    subject_id: Mapped[str | None] = mapped_column(String(64))
    subject_label: Mapped[str | None] = mapped_column(String(255))

    source_table: Mapped[str | None] = mapped_column(String(64))
    source_pk: Mapped[str | None] = mapped_column(String(64))
    audit_log_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    source_trace_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    # Microsecond precision: ``content_hash`` digests this value, and a
    # whole-second column would round it up on insert so the digest
    # could no longer be reproduced from the stored rows.
    occurred_at: Mapped[datetime | None] = mapped_column(PreciseDateTime)
    payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

    __table_args__ = (
        UniqueConstraint("bundle_id", "seq", name="uq_auditor_evidence_row_seq"),
        Index("idx_auditor_evidence_row_bundle", "bundle_id", "seq"),
    )

    def __repr__(self) -> str:
        return f"<AuditorEvidenceBundleRow {self.bundle_id}#{self.seq}>"


class AuditorSelfAttestationAdvice(Base):
    """The model's suggestion for ONE holder's self-attestation footprint.

    One row per ``(certification, person)``. Written by
    ``auditor.self_attestation_advice``; read by the holder's review page
    and by the manager follow-up campaign.

    **The suggestion lives here, beside the holder's decision, never in
    it.** ``access_certification_items.decision`` stays the only place a
    choice is recorded, and nothing is executed from this table — the
    same separation ``HrRoleChangeReview.ai_recommendation`` keeps for
    the mover packet.

    ``facts_snapshot`` is the fact package the narrative was generated
    from. Kept because a suggestion an operator disagrees with is only
    arguable if you can see what it was told — and because it is the
    only way to tell "the model was wrong" from "the model was right
    about stale inputs".

    ``failure_code`` records why there is no advice (no API key, vendor
    error, ungrounded reply). The surfaces render that reason rather
    than a blank space, which would read as "nothing to worry about".
    """

    __tablename__ = "auditor_self_attestation_advice"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        UniqueConstraint(
            "certification_id", "person_id",
            name="uq_self_att_advice_cert_person",
        ),
        Index("ix_self_att_advice_cert", "certification_id"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    certification_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("access_certifications.id", ondelete="CASCADE"),
        nullable=False,
    )
    #: Plain id column, no FK — the TiDB-FK-safe convention the rest of
    #: this domain's person_id columns follow.
    person_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    summary_zh: Mapped[str | None] = mapped_column(Text)
    summary_en: Mapped[str | None] = mapped_column(Text)
    #: ``{access_type: {"recommendation": "keep"|"release", "reason": str}}``
    type_recommendations: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    facts_snapshot: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    failure_code: Mapped[str | None] = mapped_column(String(64))
    generated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
        onupdate=func.now(),
    )

    def __repr__(self) -> str:
        return (
            f"<AuditorSelfAttestationAdvice cert={self.certification_id} "
            f"person={self.person_id} failure={self.failure_code}>"
        )