from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime
import hashlib
import json
from typing import Any, Literal
import uuid

from sqlalchemy import func
from sqlalchemy.orm import Session

from govoplan_core.core.events import (
    EventActorRef,
    EventObjectRef,
    EventTenantRef,
    PlatformEvent,
    emit_platform_event,
)
from govoplan_core.core.institutional import (
    EvidenceReference,
    FormalDecision,
    GovernedContextEnvelope,
    InstitutionalContextError,
    InstitutionalReference,
)
from govoplan_committee.backend.db.models import (
    CommitteeDecisionProjection,
    CommitteeWorkspaceEvent,
    CommitteeWorkspaceRevision,
)


CAPABILITY_COMMITTEE_WORKSPACE = "committee.workspace"
CommitteeObjectKind = Literal["body", "meeting", "agenda_item", "vote", "minute"]

_PARENT_KIND: dict[str, str | None] = {
    "body": None,
    "meeting": "body",
    "agenda_item": "meeting",
    "vote": "agenda_item",
    "minute": "meeting",
}
_STATES: dict[str, frozenset[str]] = {
    "body": frozenset({"draft", "active", "suspended", "retired"}),
    "meeting": frozenset({"draft", "scheduled", "open", "closed", "cancelled"}),
    "agenda_item": frozenset(
        {"draft", "scheduled", "deliberating", "decided", "withdrawn"}
    ),
    "vote": frozenset({"draft", "open", "closed", "cancelled"}),
    "minute": frozenset({"draft", "proposed", "accepted", "corrected"}),
}
_TRANSITIONS: dict[str, dict[str, frozenset[str]]] = {
    "body": {
        "draft": frozenset({"draft", "active", "retired"}),
        "active": frozenset({"active", "suspended", "retired"}),
        "suspended": frozenset({"active", "suspended", "retired"}),
        "retired": frozenset(),
    },
    "meeting": {
        "draft": frozenset({"draft", "scheduled", "cancelled"}),
        "scheduled": frozenset({"scheduled", "open", "cancelled"}),
        "open": frozenset({"open", "closed", "cancelled"}),
        "closed": frozenset(),
        "cancelled": frozenset(),
    },
    "agenda_item": {
        "draft": frozenset({"draft", "scheduled", "withdrawn"}),
        "scheduled": frozenset({"scheduled", "deliberating", "withdrawn"}),
        "deliberating": frozenset({"deliberating", "decided", "withdrawn"}),
        "decided": frozenset(),
        "withdrawn": frozenset(),
    },
    "vote": {
        "draft": frozenset({"draft", "open", "cancelled"}),
        "open": frozenset({"open", "closed", "cancelled"}),
        "closed": frozenset(),
        "cancelled": frozenset(),
    },
    "minute": {
        "draft": frozenset({"draft", "proposed"}),
        "proposed": frozenset({"proposed", "accepted"}),
        "accepted": frozenset({"corrected"}),
        "corrected": frozenset({"corrected"}),
    },
}


class CommitteeWorkspaceError(ValueError):
    pass


@dataclass(frozen=True, slots=True)
class CommitteeWorkspaceRecord:
    tenant_id: str
    object_kind: CommitteeObjectKind
    object_id: str
    revision: int
    state: str
    title: str
    recorded_at: datetime
    change_reason: str
    parent_id: str | None = None
    attributes: Mapping[str, Any] = field(default_factory=dict)
    context: GovernedContextEnvelope | None = None
    evidence: tuple[EvidenceReference, ...] = ()
    record_refs: tuple[InstitutionalReference, ...] = ()

    def __post_init__(self) -> None:
        if self.object_kind not in _PARENT_KIND:
            raise CommitteeWorkspaceError("Unsupported Committee object kind.")
        _identifier(self.tenant_id, "Committee tenant id", maximum=36)
        _identifier(self.object_id, "Committee object id", maximum=255)
        if self.revision < 1:
            raise CommitteeWorkspaceError("Committee object revision must be positive.")
        if self.state not in _STATES[self.object_kind]:
            raise CommitteeWorkspaceError(
                f"Unsupported {self.object_kind} state: {self.state!r}."
            )
        _required_text(self.title, "Committee object title", maximum=500)
        _required_text(self.change_reason, "Committee change reason", maximum=1_000)
        _require_aware(self.recorded_at, "Committee recorded_at")
        expected_parent = _PARENT_KIND[self.object_kind]
        if expected_parent is None and self.parent_id is not None:
            raise CommitteeWorkspaceError(
                "Committee bodies cannot have a workspace parent."
            )
        if expected_parent is not None:
            _identifier(
                self.parent_id or "",
                f"Committee {expected_parent} parent id",
                maximum=255,
            )
        if self.context is not None and self.context.tenant_id != self.tenant_id:
            raise CommitteeWorkspaceError(
                "Committee institutional context belongs to another tenant."
            )
        if any(item.tenant_id != self.tenant_id for item in self.evidence):
            raise CommitteeWorkspaceError("Committee evidence cannot cross tenants.")
        if any(
            item.tenant_id != self.tenant_id or item.kind != "record"
            for item in self.record_refs
        ):
            raise CommitteeWorkspaceError(
                "Committee record references must be same-tenant record references."
            )
        _validate_attributes(self)

    def to_dict(self) -> dict[str, Any]:
        return {
            "tenant_id": self.tenant_id,
            "object_kind": self.object_kind,
            "object_id": self.object_id,
            "revision": self.revision,
            "state": self.state,
            "title": self.title,
            "parent_id": self.parent_id,
            "recorded_at": self.recorded_at.isoformat(),
            "change_reason": self.change_reason,
            "attributes": _json_value(self.attributes),
            "context": self.context.to_dict() if self.context else None,
            "evidence": [item.to_dict() for item in self.evidence],
            "record_refs": [item.to_dict() for item in self.record_refs],
        }

    @classmethod
    def from_mapping(cls, value: Mapping[str, object]) -> "CommitteeWorkspaceRecord":
        context = value.get("context")
        attributes = value.get("attributes") or {}
        if not isinstance(attributes, Mapping):
            raise CommitteeWorkspaceError("Committee attributes must be an object.")
        return cls(
            tenant_id=_required_text(
                value.get("tenant_id"), "Committee tenant id", maximum=36
            ),
            object_kind=_object_kind(value.get("object_kind")),
            object_id=_required_text(
                value.get("object_id"), "Committee object id", maximum=255
            ),
            revision=_positive_int(value.get("revision"), "Committee revision"),
            state=_required_text(value.get("state"), "Committee state", maximum=30),
            title=_required_text(value.get("title"), "Committee title", maximum=500),
            parent_id=_optional_text(value.get("parent_id"), maximum=255),
            recorded_at=_datetime(value.get("recorded_at"), "Committee recorded_at"),
            change_reason=_required_text(
                value.get("change_reason"), "Committee change reason", maximum=1_000
            ),
            attributes=dict(attributes),
            context=(
                GovernedContextEnvelope.from_mapping(context)
                if isinstance(context, Mapping)
                else None
            ),
            evidence=tuple(
                EvidenceReference.from_mapping(item)
                for item in _mapping_items(value.get("evidence"), "Committee evidence")
            ),
            record_refs=tuple(
                InstitutionalReference.from_mapping(item)
                for item in _mapping_items(
                    value.get("record_refs"), "Committee record references"
                )
            ),
        )


def record_workspace_object(
    session: Session,
    principal: object,
    *,
    record: CommitteeWorkspaceRecord,
    idempotency_key: str,
    expected_revision: int | None = None,
    _provider_finalization: bool = False,
) -> CommitteeWorkspaceRecord:
    tenant_id = _principal_tenant(principal)
    if record.tenant_id != tenant_id:
        raise CommitteeWorkspaceError(
            "Committee workspace records cannot cross tenants."
        )
    clean_key = _required_text(
        idempotency_key,
        "Committee idempotency key",
        maximum=255,
    )
    request_sha256 = _sha256(
        {
            "record": record.to_dict(),
            "expected_revision": expected_revision,
        }
    )
    replay = _replay(
        session,
        tenant_id=tenant_id,
        idempotency_key=clean_key,
        request_sha256=request_sha256,
    )
    if replay is not None:
        return replay
    current = _current_row(
        session,
        tenant_id=tenant_id,
        object_kind=record.object_kind,
        object_id=record.object_id,
        lock=True,
    )
    if current is None:
        if expected_revision is not None or record.revision != 1:
            raise CommitteeWorkspaceError(
                "A new Committee object must start at revision 1 without an expected revision."
            )
    else:
        if expected_revision != current.revision:
            raise CommitteeWorkspaceError(
                "Committee revision conflict: the expected revision is stale."
            )
        if record.revision != current.revision + 1:
            raise CommitteeWorkspaceError(
                "Committee object revisions must be consecutive."
            )
        if record.parent_id != current.parent_id:
            raise CommitteeWorkspaceError(
                "A Committee object's parent cannot change across revisions."
            )
        if record.state not in _TRANSITIONS[record.object_kind][current.state]:
            raise CommitteeWorkspaceError(
                f"Committee {record.object_kind} transition {current.state!r} to "
                f"{record.state!r} is not allowed."
            )
        current_payload = (
            current.payload if isinstance(current.payload, Mapping) else {}
        )
        current_attributes = current_payload.get("attributes")
        provider_id = (
            str(current_attributes.get("provider_id") or "").strip()
            if isinstance(current_attributes, Mapping)
            else ""
        )
        voting_ballot_id = (
            str(current_attributes.get("voting_ballot_id") or "").strip()
            if isinstance(current_attributes, Mapping)
            else ""
        )
        if (
            record.object_kind == "vote"
            and current.state == "open"
            and record.state == "closed"
            and (provider_id or voting_ballot_id)
            and not _provider_finalization
        ):
            raise CommitteeWorkspaceError(
                "A managed Committee vote must be finalized through Voting or its compatibility ballot adapter."
            )
        current.superseded_at = record.recorded_at
    _validate_parent(session, record)
    _validate_related_state(session, record)
    row = CommitteeWorkspaceRevision(
        tenant_id=tenant_id,
        object_kind=record.object_kind,
        object_id=record.object_id,
        revision=record.revision,
        previous_revision_id=current.id if current is not None else None,
        parent_kind=_PARENT_KIND[record.object_kind],
        parent_id=record.parent_id,
        state=record.state,
        title=record.title,
        search_text=f"{record.title} {record.object_id} {record.state}".casefold(),
        recorded_at=record.recorded_at,
        payload=record.to_dict(),
        changed_by=_principal_actor(principal),
    )
    event_id = str(uuid.uuid4())
    operation = "created" if current is None else "updated"
    event = CommitteeWorkspaceEvent(
        tenant_id=tenant_id,
        object_kind=record.object_kind,
        object_id=record.object_id,
        object_revision=record.revision,
        event_id=event_id,
        event_type=f"committee.{record.object_kind}.{operation}",
        occurred_at=record.recorded_at,
        actor_id=_principal_actor(principal),
        idempotency_key=clean_key,
        request_sha256=request_sha256,
        payload={
            "state": record.state,
            "revision": record.revision,
            "parent_id": record.parent_id,
            "change_reason": record.change_reason,
        },
    )
    session.add_all((row, event))
    session.flush()
    emit_platform_event(
        session,
        PlatformEvent(
            event_id=event_id,
            type=event.event_type,
            module_id="committee",
            payload=dict(event.payload),
            occurred_at=record.recorded_at,
            actor=EventActorRef(type="account", id=_principal_actor(principal)),
            tenant=EventTenantRef(id=tenant_id),
            resource=EventObjectRef(
                type=f"committee_{record.object_kind}",
                id=record.object_id,
                label=record.title,
            ),
            classification="internal",
            institutional_context=record.context,
        ),
    )
    return _workspace_from_row(row)


def get_workspace_object(
    session: Session,
    principal: object,
    *,
    object_kind: str,
    object_id: str,
    revision: int | None = None,
) -> CommitteeWorkspaceRecord | None:
    tenant_id = _principal_tenant(principal)
    kind = _object_kind(object_kind)
    query = session.query(CommitteeWorkspaceRevision).filter(
        CommitteeWorkspaceRevision.tenant_id == tenant_id,
        CommitteeWorkspaceRevision.object_kind == kind,
        CommitteeWorkspaceRevision.object_id == object_id,
    )
    if revision is None:
        query = query.filter(CommitteeWorkspaceRevision.superseded_at.is_(None))
    else:
        query = query.filter(CommitteeWorkspaceRevision.revision == revision)
    row = query.order_by(CommitteeWorkspaceRevision.revision.desc()).first()
    return _workspace_from_row(row) if row is not None else None


def list_workspace_objects(
    session: Session,
    principal: object,
    *,
    object_kind: str,
    parent_id: str | None = None,
    states: Sequence[str] | None = None,
    query: str = "",
    offset: int = 0,
    limit: int = 100,
) -> tuple[tuple[CommitteeWorkspaceRecord, ...], int]:
    tenant_id = _principal_tenant(principal)
    kind = _object_kind(object_kind)
    if offset < 0 or not 1 <= limit <= 200:
        raise CommitteeWorkspaceError(
            "Committee list offset must be non-negative and limit between 1 and 200."
        )
    statement = session.query(CommitteeWorkspaceRevision).filter(
        CommitteeWorkspaceRevision.tenant_id == tenant_id,
        CommitteeWorkspaceRevision.object_kind == kind,
        CommitteeWorkspaceRevision.superseded_at.is_(None),
    )
    if parent_id is not None:
        statement = statement.filter(CommitteeWorkspaceRevision.parent_id == parent_id)
    if states:
        statement = statement.filter(
            CommitteeWorkspaceRevision.state.in_(tuple(states))
        )
    clean_query = query.strip().casefold()
    if clean_query:
        statement = statement.filter(
            CommitteeWorkspaceRevision.search_text.contains(clean_query)
        )
    total = int(statement.with_entities(func.count()).scalar() or 0)
    rows = (
        statement.order_by(
            CommitteeWorkspaceRevision.recorded_at.desc(),
            CommitteeWorkspaceRevision.object_id.asc(),
        )
        .offset(offset)
        .limit(limit)
        .all()
    )
    return tuple(_workspace_from_row(row) for row in rows), total


def workspace_history(
    session: Session,
    principal: object,
    *,
    object_kind: str,
    object_id: str,
    limit: int = 100,
) -> tuple[CommitteeWorkspaceRecord, ...]:
    if not 1 <= limit <= 200:
        raise CommitteeWorkspaceError(
            "Committee history limit must be between 1 and 200."
        )
    rows = (
        session.query(CommitteeWorkspaceRevision)
        .filter(
            CommitteeWorkspaceRevision.tenant_id == _principal_tenant(principal),
            CommitteeWorkspaceRevision.object_kind == _object_kind(object_kind),
            CommitteeWorkspaceRevision.object_id == object_id,
        )
        .order_by(CommitteeWorkspaceRevision.revision.desc())
        .limit(limit)
        .all()
    )
    return tuple(_workspace_from_row(row) for row in rows)


class SqlCommitteeWorkspace:
    def record(
        self,
        session: object,
        principal: object,
        *,
        record: CommitteeWorkspaceRecord,
        idempotency_key: str,
        expected_revision: int | None = None,
    ) -> CommitteeWorkspaceRecord:
        return record_workspace_object(
            _session(session),
            principal,
            record=record,
            idempotency_key=idempotency_key,
            expected_revision=expected_revision,
        )

    def get(
        self,
        session: object,
        principal: object,
        *,
        object_kind: str,
        object_id: str,
        revision: int | None = None,
    ) -> CommitteeWorkspaceRecord | None:
        return get_workspace_object(
            _session(session),
            principal,
            object_kind=object_kind,
            object_id=object_id,
            revision=revision,
        )

    def record_local_decision(
        self,
        session: object,
        principal: object,
        *,
        decision: FormalDecision,
        meeting_id: str,
        agenda_item_id: str,
        expected_revision: str | None = None,
    ) -> FormalDecision:
        return record_local_decision(
            _session(session),
            principal,
            decision=decision,
            meeting_id=meeting_id,
            agenda_item_id=agenda_item_id,
            expected_revision=expected_revision,
        )

    def get_local_decision(
        self,
        session: object,
        principal: object,
        *,
        decision_id: str,
        revision: str | None = None,
    ) -> FormalDecision | None:
        return get_local_decision(
            _session(session), principal, decision_id=decision_id, revision=revision
        )


def record_local_decision(
    session: Session,
    principal: object,
    *,
    decision: FormalDecision,
    meeting_id: str,
    agenda_item_id: str,
    expected_revision: str | None = None,
) -> FormalDecision:
    tenant_id = _principal_tenant(principal)
    if decision.reference.owner_module != "committee":
        raise CommitteeWorkspaceError(
            "Only a Committee-owned fallback Decision may use the local projection."
        )
    if decision.reference.tenant_id != tenant_id:
        raise CommitteeWorkspaceError(
            "Committee Decision projections cannot cross tenants."
        )
    for kind, object_id in (("meeting", meeting_id), ("agenda_item", agenda_item_id)):
        if (
            get_workspace_object(
                session,
                principal,
                object_kind=kind,
                object_id=object_id,
            )
            is None
        ):
            raise CommitteeWorkspaceError(
                f"Committee Decision projection requires an existing {kind.replace('_', ' ')}."
            )
    payload = decision.to_dict(include_protected=True)
    replay = (
        session.query(CommitteeDecisionProjection)
        .filter(
            CommitteeDecisionProjection.tenant_id == tenant_id,
            CommitteeDecisionProjection.decision_id == decision.reference.object_id,
            CommitteeDecisionProjection.revision == decision.temporal.revision,
        )
        .one_or_none()
    )
    if replay is not None:
        if replay.payload != payload:
            raise CommitteeWorkspaceError(
                "A different Committee Decision already uses this revision."
            )
        return FormalDecision.from_mapping(replay.payload)
    current = (
        session.query(CommitteeDecisionProjection)
        .filter(
            CommitteeDecisionProjection.tenant_id == tenant_id,
            CommitteeDecisionProjection.decision_id == decision.reference.object_id,
            CommitteeDecisionProjection.superseded_at.is_(None),
        )
        .with_for_update()
        .one_or_none()
    )
    if current is None:
        if expected_revision is not None:
            raise CommitteeWorkspaceError(
                "Committee Decision revision conflict: no current projection exists."
            )
    else:
        if expected_revision != current.revision:
            raise CommitteeWorkspaceError(
                "Committee Decision revision conflict: the expected revision is stale."
            )
        current.superseded_at = decision.temporal.recorded_at
    if decision.temporal.recorded_at is None:
        raise CommitteeWorkspaceError(
            "Committee Decision projection requires recorded_at."
        )
    row = CommitteeDecisionProjection(
        tenant_id=tenant_id,
        decision_id=decision.reference.object_id,
        revision=decision.temporal.revision,
        previous_revision_id=current.id if current is not None else None,
        meeting_id=meeting_id,
        agenda_item_id=agenda_item_id,
        state=decision.state,
        recorded_at=decision.temporal.recorded_at,
        payload=payload,
        changed_by=_principal_actor(principal),
    )
    session.add(row)
    session.flush()
    emit_platform_event(
        session,
        PlatformEvent(
            type="committee.decision.projected",
            module_id="committee",
            payload={
                "revision": decision.temporal.revision,
                "state": decision.state,
                "meeting_id": meeting_id,
                "agenda_item_id": agenda_item_id,
            },
            occurred_at=decision.temporal.recorded_at,
            actor=EventActorRef(type="account", id=_principal_actor(principal)),
            tenant=EventTenantRef(id=tenant_id),
            resource=EventObjectRef(type="decision", id=decision.reference.object_id),
            classification="restricted",
            institutional_context=decision.authority_context,
        ),
    )
    return FormalDecision.from_mapping(row.payload)


def get_local_decision(
    session: Session,
    principal: object,
    *,
    decision_id: str,
    revision: str | None = None,
) -> FormalDecision | None:
    query = session.query(CommitteeDecisionProjection).filter(
        CommitteeDecisionProjection.tenant_id == _principal_tenant(principal),
        CommitteeDecisionProjection.decision_id == decision_id,
    )
    if revision is None:
        query = query.filter(CommitteeDecisionProjection.superseded_at.is_(None))
    else:
        query = query.filter(CommitteeDecisionProjection.revision == revision)
    row = query.order_by(CommitteeDecisionProjection.recorded_at.desc()).first()
    return FormalDecision.from_mapping(row.payload) if row is not None else None


def _validate_parent(session: Session, record: CommitteeWorkspaceRecord) -> None:
    parent_kind = _PARENT_KIND[record.object_kind]
    if parent_kind is None:
        return
    parent = _current_row(
        session,
        tenant_id=record.tenant_id,
        object_kind=parent_kind,
        object_id=record.parent_id or "",
        lock=False,
    )
    if parent is None:
        raise CommitteeWorkspaceError(
            f"Committee {record.object_kind} requires an existing {parent_kind.replace('_', ' ')}."
        )
    if parent.state in {"retired", "cancelled", "withdrawn"}:
        raise CommitteeWorkspaceError(
            "Committee objects cannot be added below a terminal parent."
        )


def _validate_related_state(session: Session, record: CommitteeWorkspaceRecord) -> None:
    if record.object_kind == "meeting" and record.state == "closed":
        unfinished = (
            session.query(CommitteeWorkspaceRevision.id)
            .filter(
                CommitteeWorkspaceRevision.tenant_id == record.tenant_id,
                CommitteeWorkspaceRevision.object_kind == "agenda_item",
                CommitteeWorkspaceRevision.parent_id == record.object_id,
                CommitteeWorkspaceRevision.superseded_at.is_(None),
                ~CommitteeWorkspaceRevision.state.in_(("decided", "withdrawn")),
            )
            .first()
        )
        if unfinished is not None:
            raise CommitteeWorkspaceError(
                "A meeting cannot close while agenda items remain unfinished."
            )
    if record.object_kind == "agenda_item" and record.state == "decided":
        decision_ref = _reference(
            record.attributes.get("decision_ref"),
            "decision",
            record.tenant_id,
            required=True,
        )
        if decision_ref is None:
            raise CommitteeWorkspaceError(
                "A decided agenda item requires a formal Decision reference."
            )


def _validate_attributes(record: CommitteeWorkspaceRecord) -> None:
    attributes = record.attributes
    if len(attributes) > 100:
        raise CommitteeWorkspaceError(
            "Committee attributes are limited to 100 entries."
        )
    if record.object_kind == "body":
        _reference(
            attributes.get("organization_unit_ref"),
            "organization_unit",
            record.tenant_id,
            required=True,
        )
        _references(
            attributes.get("function_refs"),
            "function",
            record.tenant_id,
        )
        quorum = attributes.get("quorum")
        if quorum is not None and not isinstance(quorum, Mapping):
            raise CommitteeWorkspaceError("Committee body quorum must be an object.")
    elif record.object_kind == "meeting":
        starts_at = _datetime(attributes.get("starts_at"), "Meeting starts_at")
        ends_at = _datetime(attributes.get("ends_at"), "Meeting ends_at")
        if ends_at <= starts_at:
            raise CommitteeWorkspaceError("Meeting ends_at must follow starts_at.")
    elif record.object_kind == "agenda_item":
        if _positive_int(attributes.get("position"), "Agenda position") < 1:
            raise CommitteeWorkspaceError("Agenda position must be positive.")
        subjects = _references(
            attributes.get("subject_refs"),
            None,
            record.tenant_id,
        )
        if not subjects:
            raise CommitteeWorkspaceError(
                "Committee agenda items require at least one subject reference."
            )
        if record.state == "decided":
            _reference(
                attributes.get("decision_ref"),
                "decision",
                record.tenant_id,
                required=True,
            )
    elif record.object_kind == "vote":
        method = str(attributes.get("method") or "recorded")
        if method not in {"recorded", "public", "secret"}:
            raise CommitteeWorkspaceError("Unsupported Committee vote method.")
        choices = tuple(
            str(item).strip()
            for item in _sequence(attributes.get("choices"), "Vote choices")
            if str(item).strip()
        )
        if len(choices) < 2 or len(choices) != len(set(choices)):
            raise CommitteeWorkspaceError(
                "Committee votes require at least two unique choices."
            )
        eligible = _non_negative_int(
            attributes.get("eligible_count"),
            "Vote eligible_count",
        )
        cast = _non_negative_int(attributes.get("cast_count"), "Vote cast_count")
        if cast > eligible:
            raise CommitteeWorkspaceError(
                "Vote cast_count cannot exceed eligible_count."
            )
        if record.state == "closed":
            counts = attributes.get("counts")
            if not isinstance(counts, Mapping):
                raise CommitteeWorkspaceError("A closed vote requires result counts.")
            clean_counts = {
                str(key): _non_negative_int(value, "Vote count")
                for key, value in counts.items()
            }
            if set(clean_counts) - set(choices) or sum(clean_counts.values()) != cast:
                raise CommitteeWorkspaceError(
                    "Closed vote counts must match the configured choices and cast_count."
                )
            if not isinstance(attributes.get("quorum_met"), bool):
                raise CommitteeWorkspaceError(
                    "A closed vote requires an explicit quorum_met result."
                )
            _reference(
                attributes.get("approval_ref"),
                "approval",
                record.tenant_id,
                required=True,
            )
            if not record.evidence:
                raise CommitteeWorkspaceError("A closed vote requires result evidence.")
    elif record.object_kind == "minute":
        _reference(
            attributes.get("content_ref"),
            "record",
            record.tenant_id,
            required=True,
        )
        if record.state in {"accepted", "corrected"}:
            _reference(
                attributes.get("approval_ref"),
                "approval",
                record.tenant_id,
                required=True,
            )
            if not record.evidence:
                raise CommitteeWorkspaceError(
                    "Accepted or corrected minutes require evidence."
                )


def _replay(
    session: Session, *, tenant_id: str, idempotency_key: str, request_sha256: str
) -> CommitteeWorkspaceRecord | None:
    row = (
        session.query(CommitteeWorkspaceEvent)
        .filter(
            CommitteeWorkspaceEvent.tenant_id == tenant_id,
            CommitteeWorkspaceEvent.idempotency_key == idempotency_key,
        )
        .one_or_none()
    )
    if row is None:
        return None
    if row.request_sha256 != request_sha256:
        raise CommitteeWorkspaceError(
            "Committee idempotency conflict: the key was used for another request."
        )
    revision = (
        session.query(CommitteeWorkspaceRevision)
        .filter(
            CommitteeWorkspaceRevision.tenant_id == tenant_id,
            CommitteeWorkspaceRevision.object_kind == row.object_kind,
            CommitteeWorkspaceRevision.object_id == row.object_id,
            CommitteeWorkspaceRevision.revision == row.object_revision,
        )
        .one()
    )
    return _workspace_from_row(revision)


def _current_row(
    session: Session, *, tenant_id: str, object_kind: str, object_id: str, lock: bool
) -> CommitteeWorkspaceRevision | None:
    query = session.query(CommitteeWorkspaceRevision).filter(
        CommitteeWorkspaceRevision.tenant_id == tenant_id,
        CommitteeWorkspaceRevision.object_kind == object_kind,
        CommitteeWorkspaceRevision.object_id == object_id,
        CommitteeWorkspaceRevision.superseded_at.is_(None),
    )
    if lock:
        query = query.with_for_update()
    return query.one_or_none()


def _workspace_from_row(row: CommitteeWorkspaceRevision) -> CommitteeWorkspaceRecord:
    payload = dict(row.payload)
    payload["revision"] = row.revision
    payload["state"] = row.state
    payload["recorded_at"] = _datetime_text(row.recorded_at)
    return CommitteeWorkspaceRecord.from_mapping(payload)


def _object_kind(value: object) -> CommitteeObjectKind:
    result = str(value or "").strip()
    if result not in _PARENT_KIND:
        raise CommitteeWorkspaceError("Unsupported Committee object kind.")
    return result  # type: ignore[return-value]


def _reference(
    value: object, kind: str | None, tenant_id: str, *, required: bool
) -> InstitutionalReference | None:
    if value is None:
        if required:
            raise CommitteeWorkspaceError(
                f"Committee {kind or 'institutional'} reference is required."
            )
        return None
    if not isinstance(value, Mapping):
        raise CommitteeWorkspaceError("Committee reference must be an object.")
    result = InstitutionalReference.from_mapping(value)
    if result.tenant_id != tenant_id or (kind is not None and result.kind != kind):
        raise CommitteeWorkspaceError(
            "Committee reference has the wrong tenant or kind."
        )
    return result


def _references(
    value: object, kind: str | None, tenant_id: str
) -> tuple[InstitutionalReference, ...]:
    return tuple(
        item
        for item in (
            _reference(candidate, kind, tenant_id, required=True)
            for candidate in _sequence(value, "Committee references")
        )
        if item is not None
    )


def _mapping_items(value: object, label: str) -> tuple[Mapping[str, object], ...]:
    items = _sequence(value, label)
    if any(not isinstance(item, Mapping) for item in items):
        raise CommitteeWorkspaceError(f"{label} must contain objects.")
    return tuple(items)  # type: ignore[return-value]


def _sequence(value: object, label: str) -> tuple[object, ...]:
    if value is None:
        return ()
    if not isinstance(value, (list, tuple)):
        raise CommitteeWorkspaceError(f"{label} must be a list.")
    return tuple(value)


def _datetime(value: object, label: str) -> datetime:
    if isinstance(value, datetime):
        result = value
    else:
        try:
            result = datetime.fromisoformat(str(value or "").replace("Z", "+00:00"))
        except ValueError as exc:
            raise CommitteeWorkspaceError(f"{label} is invalid.") from exc
    _require_aware(result, label)
    return result


def _require_aware(value: datetime, label: str) -> None:
    if value.tzinfo is None or value.utcoffset() is None:
        raise CommitteeWorkspaceError(f"{label} must include a timezone.")


def _positive_int(value: object, label: str) -> int:
    result = _non_negative_int(value, label)
    if result < 1:
        raise CommitteeWorkspaceError(f"{label} must be positive.")
    return result


def _non_negative_int(value: object, label: str) -> int:
    if isinstance(value, bool):
        raise CommitteeWorkspaceError(f"{label} must be an integer.")
    try:
        result = int(value)  # type: ignore[arg-type]
    except (TypeError, ValueError) as exc:
        raise CommitteeWorkspaceError(f"{label} must be an integer.") from exc
    if result < 0:
        raise CommitteeWorkspaceError(f"{label} cannot be negative.")
    return result


def _identifier(value: object, label: str, *, maximum: int) -> str:
    result = _required_text(value, label, maximum=maximum)
    if any(character.isspace() for character in result):
        raise CommitteeWorkspaceError(f"{label} cannot contain whitespace.")
    return result


def _required_text(value: object, label: str, *, maximum: int) -> str:
    result = str(value or "").strip()
    if not result:
        raise CommitteeWorkspaceError(f"{label} is required.")
    if len(result) > maximum:
        raise CommitteeWorkspaceError(f"{label} is limited to {maximum} characters.")
    return result


def _optional_text(value: object, *, maximum: int) -> str | None:
    result = str(value or "").strip()
    if not result:
        return None
    if len(result) > maximum:
        raise CommitteeWorkspaceError(f"Text is limited to {maximum} characters.")
    return result


def _sha256(value: object) -> str:
    return hashlib.sha256(
        json.dumps(
            _json_value(value), sort_keys=True, separators=(",", ":"), ensure_ascii=True
        ).encode("utf-8")
    ).hexdigest()


def _json_value(value: object) -> Any:
    if isinstance(value, datetime):
        return value.isoformat()
    if hasattr(value, "to_dict"):
        return _json_value(value.to_dict())
    if isinstance(value, Mapping):
        return {str(key): _json_value(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_json_value(item) for item in value]
    return value


def _principal_tenant(principal: object) -> str:
    tenant_id = str(getattr(principal, "tenant_id", "") or "").strip()
    if not tenant_id:
        raise InstitutionalContextError(
            "Committee operations require a tenant-bound principal."
        )
    return tenant_id


def _principal_actor(principal: object) -> str | None:
    user = getattr(principal, "user", None)
    for value in (
        getattr(user, "id", None),
        getattr(principal, "account_id", None),
        getattr(principal, "identity_id", None),
        getattr(principal, "membership_id", None),
    ):
        candidate = str(value or "").strip()
        if candidate:
            return candidate
    return None


def _session(value: object) -> Session:
    if not hasattr(value, "query"):
        raise InstitutionalContextError(
            "Committee workspace requires a database session."
        )
    return value  # type: ignore[return-value]


def _datetime_text(value: datetime | None) -> str | None:
    if value is None:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=UTC)
    return value.isoformat()


__all__ = [
    "CAPABILITY_COMMITTEE_WORKSPACE",
    "CommitteeObjectKind",
    "CommitteeWorkspaceError",
    "CommitteeWorkspaceRecord",
    "SqlCommitteeWorkspace",
    "get_local_decision",
    "get_workspace_object",
    "list_workspace_objects",
    "record_local_decision",
    "record_workspace_object",
    "workspace_history",
]