from __future__ import annotations from dataclasses import dataclass, field, replace from datetime import datetime from typing import Literal, Mapping from govoplan_core.core.institutional import ( EvidenceReference, InstitutionalContextError, InstitutionalReference, ServiceBinding, ) FORM_INSTANCE_STATUSES = frozenset( { "started", "draft", "submitted", "validated", "needs_review", "accepted", "rejected", "handed_off", "archived", } ) FormHandoffState = Literal[ "requested", "accepted", "rejected", "outcome_unknown", "reconciled", "compensated", ] FORM_HANDOFF_STATES = frozenset( { "requested", "accepted", "rejected", "outcome_unknown", "reconciled", "compensated", } ) @dataclass(frozen=True, slots=True) class FormHandoff: tenant_id: str instance_id: str effect_id: str instance_revision: int idempotency_key: str provider_key: str request_sha256: str binding_kind: str binding_reference: str provider_capability: str state: FormHandoffState attempt_count: int requested_at: datetime resolved_at: datetime | None = None target_ref: InstitutionalReference | None = None href: str | None = None evidence: tuple[EvidenceReference, ...] = () last_error: str | None = None metadata: Mapping[str, object] = field(default_factory=dict) def __post_init__(self) -> None: if self.state not in FORM_HANDOFF_STATES: raise InstitutionalContextError( f"Unsupported Form handoff state: {self.state!r}." ) if self.attempt_count < 0: raise InstitutionalContextError( "Form handoff attempt count cannot be negative." ) if self.target_ref is not None and self.target_ref.tenant_id != self.tenant_id: raise InstitutionalContextError("Form handoff target cannot cross tenants.") if any(item.tenant_id != self.tenant_id for item in self.evidence): raise InstitutionalContextError( "Form handoff evidence cannot cross tenants." ) def to_dict(self) -> dict[str, object]: return { "effect_id": self.effect_id, "instance_id": self.instance_id, "instance_revision": self.instance_revision, "binding_kind": self.binding_kind, "binding_reference": self.binding_reference, "provider_capability": self.provider_capability, "state": self.state, "attempt_count": self.attempt_count, "requested_at": self.requested_at.isoformat(), "resolved_at": self.resolved_at.isoformat() if self.resolved_at else None, "target_ref": self.target_ref.to_dict() if self.target_ref else None, "href": self.href, "evidence": [ item.to_dict(include_inspection=False) for item in self.evidence ], "last_error": self.last_error, "metadata": dict(self.metadata), } @dataclass(frozen=True, slots=True) class FormInstance: tenant_id: str instance_id: str revision: int status: str definition_ref: InstitutionalReference values: Mapping[str, object] validation_results: tuple[Mapping[str, object], ...] recorded_at: datetime change_reason: str created_by: str changed_by: str attachment_refs: tuple[EvidenceReference, ...] = () signature_refs: tuple[EvidenceReference, ...] = () handoff_refs: tuple[InstitutionalReference, ...] = () service_ref: InstitutionalReference | None = None service_binding: ServiceBinding | None = None receipt_id: str | None = None replayed: bool = False metadata: Mapping[str, object] = field(default_factory=dict) def __post_init__(self) -> None: if not self.tenant_id or not self.instance_id: raise InstitutionalContextError( "A Form instance requires tenant and instance identities." ) if self.revision < 1: raise InstitutionalContextError( "A Form instance revision must be positive." ) if self.status not in FORM_INSTANCE_STATUSES: raise InstitutionalContextError( f"Unsupported Form instance status: {self.status!r}." ) if ( self.definition_ref.kind != "form" or self.definition_ref.owner_module != "forms" or self.definition_ref.tenant_id != self.tenant_id or not self.definition_ref.version ): raise InstitutionalContextError( "A Form instance requires an exact same-tenant Forms definition." ) if self.service_ref is not None and ( self.service_ref.kind != "service" or self.service_ref.tenant_id != self.tenant_id or not self.service_ref.version ): raise InstitutionalContextError( "Form instance Service provenance must be exact and same-tenant." ) for item in (*self.attachment_refs, *self.signature_refs): if item.tenant_id != self.tenant_id: raise InstitutionalContextError( "Form instance evidence cannot cross tenants." ) for item in self.handoff_refs: if item.tenant_id != self.tenant_id: raise InstitutionalContextError( "Form instance handoff references cannot cross tenants." ) @property def reference(self) -> InstitutionalReference: return InstitutionalReference( kind="form_submission", owner_module="forms_runtime", object_id=self.instance_id, tenant_id=self.tenant_id, version=str(self.revision), valid_at=self.recorded_at, ) def with_replay(self) -> "FormInstance": return replace(self, replayed=True) def to_dict(self, *, include_values: bool = True) -> dict[str, object]: return { "reference": self.reference.to_dict(), "tenant_id": self.tenant_id, "instance_id": self.instance_id, "revision": self.revision, "status": self.status, "definition_ref": self.definition_ref.to_dict(disclose_label=True), "values": dict(self.values) if include_values else {}, "validation_results": [dict(item) for item in self.validation_results], "attachment_refs": [ item.to_dict(include_inspection=False) for item in self.attachment_refs ], "signature_refs": [ item.to_dict(include_inspection=False) for item in self.signature_refs ], "handoff_refs": [item.to_dict() for item in self.handoff_refs], "service_ref": ( self.service_ref.to_dict() if self.service_ref is not None else None ), "service_binding": ( self.service_binding.to_dict() if self.service_binding is not None else None ), "receipt_id": self.receipt_id, "recorded_at": self.recorded_at.isoformat(), "change_reason": self.change_reason, "created_by": self.created_by, "changed_by": self.changed_by, "replayed": self.replayed, "metadata": dict(self.metadata), } __all__ = [ "FORM_HANDOFF_STATES", "FORM_INSTANCE_STATUSES", "FormHandoff", "FormHandoffState", "FormInstance", ]