Files
govoplan-forms-runtime/src/govoplan_forms_runtime/backend/service.py
T

1826 lines
61 KiB
Python

from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import replace
from datetime import UTC, date, datetime
import hashlib
import json
import re
from typing import Protocol, runtime_checkable
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 (
CAPABILITY_FORM_DEFINITIONS,
EvidenceReference,
FormConditionExpression,
FormDefinition,
FormDefinitionProvider,
FormFieldDefinition,
InstitutionalContextError,
InstitutionalReference,
ServiceBinding,
ServiceDefinition,
ServiceLaunchRequest,
ServiceLaunchResult,
)
from govoplan_forms_runtime.backend.db.models import (
FormAssistedConfirmation,
FormInstanceEvent,
FormInstanceIdentity,
FormInstanceRevision,
FormIntakeSession,
)
from govoplan_forms_runtime.backend.assisted import assisted_submission_payload_sha256
from govoplan_forms_runtime.backend.domain import FormInstance
from govoplan_forms_runtime.backend.evidence import FormEvidenceCoordinator
CAPABILITY_FORMS_RUNTIME_REGISTRY = "forms_runtime.registry"
CAPABILITY_FORMS_RUNTIME_SERVICE_LAUNCHER = "forms_runtime.service_launcher"
CAPABILITY_FORMS_RUNTIME_POLICY_EVALUATOR = "forms_runtime.policy_evaluator"
_EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$")
_STATUS_TRANSITIONS: dict[str, frozenset[str]] = {
"started": frozenset({"submitted", "archived"}),
"draft": frozenset({"submitted", "archived"}),
"submitted": frozenset(
{"validated", "needs_review", "accepted", "rejected", "handed_off", "archived"}
),
"validated": frozenset(
{"needs_review", "accepted", "rejected", "handed_off", "archived"}
),
"needs_review": frozenset({"accepted", "rejected", "handed_off", "archived"}),
"accepted": frozenset({"handed_off", "archived"}),
"rejected": frozenset({"archived"}),
"handed_off": frozenset({"archived"}),
"archived": frozenset(),
}
class FormRuntimeError(ValueError):
pass
@runtime_checkable
class FormRuntimePolicyEvaluator(Protocol):
def evaluate_form_access(
self,
session: object,
principal: object,
*,
definition: FormDefinition,
action: str,
instance: FormInstance | None,
) -> bool: ...
class FormRuntimeService:
def __init__(self, registry: object | None) -> None:
self._registry = registry
self._evidence = FormEvidenceCoordinator(registry)
def create_instance(
self,
session: Session,
principal: object,
*,
definition_ref: InstitutionalReference,
values: Mapping[str, object],
attachment_refs: Sequence[EvidenceReference] = (),
signature_refs: Sequence[EvidenceReference] = (),
idempotency_key: str,
recorded_at: datetime,
instance_id: str | None = None,
service_ref: InstitutionalReference | None = None,
service_binding: ServiceBinding | None = None,
metadata: Mapping[str, object] | None = None,
) -> FormInstance:
tenant_id = _principal_tenant(principal)
_require_aware(recorded_at, "Form instance recorded_at")
definition = self._definition(
session,
principal,
reference=definition_ref,
effective_at=recorded_at,
)
supplied_values = _mapping_copy(values, "Form values")
clean_values = normalize_form_values(
definition,
{
**{
field.key: field.default_value
for field in definition.fields
if field.default_value is not None
},
**supplied_values,
},
)
attachments = tuple(attachment_refs)
signatures = tuple(signature_refs)
diagnostics = validate_form_values(
definition,
clean_values,
attachment_refs=attachments,
signature_refs=signatures,
final=False,
)
request = {
"operation": "create",
"definition_ref": definition_ref.to_dict(),
"values": clean_values,
"attachment_refs": [item.to_dict() for item in attachments],
"signature_refs": [item.to_dict() for item in signatures],
"service_ref": service_ref.to_dict() if service_ref else None,
"service_binding": service_binding.to_dict() if service_binding else None,
"instance_id": instance_id,
"recorded_at": recorded_at.isoformat(),
"metadata": dict(metadata or {}),
}
request_sha256 = _request_hash(request)
replay = _replay(
session,
principal,
tenant_id=tenant_id,
idempotency_key=idempotency_key,
request_sha256=request_sha256,
)
if replay is not None:
return replay
_require_startable(definition)
self._evaluate_policy(
session,
principal,
definition=definition,
action="start",
instance=None,
)
actor_id = _principal_actor(principal)
resolved_id = _identifier(instance_id or str(uuid.uuid4()), "Form instance id")
existing_identity = (
session.query(FormInstanceIdentity.id)
.filter(
FormInstanceIdentity.tenant_id == tenant_id,
FormInstanceIdentity.instance_id == resolved_id,
)
.first()
)
if existing_identity is not None:
raise FormRuntimeError(
"Form instance conflict: this instance id is already in use."
)
identity = FormInstanceIdentity(
tenant_id=tenant_id,
instance_id=resolved_id,
definition_id=definition_ref.object_id,
definition_revision=str(definition_ref.version),
created_by=actor_id,
)
session.add(identity)
session.flush()
instance = FormInstance(
tenant_id=tenant_id,
instance_id=resolved_id,
revision=1,
status="draft" if definition.allow_drafts else "started",
definition_ref=definition_ref,
values=clean_values,
validation_results=diagnostics,
attachment_refs=attachments,
signature_refs=signatures,
service_ref=service_ref,
service_binding=service_binding,
recorded_at=recorded_at,
change_reason="Form instance started.",
created_by=actor_id,
changed_by=actor_id,
metadata=dict(metadata or {}),
)
evidence_diagnostics, evidence_snapshots = self._inspect_evidence(
session,
principal,
instance=instance,
definition=definition,
values=clean_values,
attachment_refs=attachments,
signature_refs=signatures,
purpose="start Form instance",
final=False,
observed_at=recorded_at,
)
if evidence_diagnostics or evidence_snapshots:
instance = replace(
instance,
validation_results=(*diagnostics, *evidence_diagnostics),
metadata={
**dict(instance.metadata),
"evidence_verification": list(evidence_snapshots),
},
)
return _record_instance(
session,
principal,
identity=identity,
current=None,
instance=instance,
idempotency_key=idempotency_key,
request_sha256=request_sha256,
operation="started",
)
def update_draft(
self,
session: Session,
principal: object,
*,
instance_id: str,
expected_revision: int,
values: Mapping[str, object],
attachment_refs: Sequence[EvidenceReference],
signature_refs: Sequence[EvidenceReference],
idempotency_key: str,
recorded_at: datetime,
change_reason: str,
allow_all: bool = False,
) -> FormInstance:
current, identity = _current_instance(
session,
principal,
instance_id=instance_id,
lock=True,
allow_all=allow_all,
)
definition = self._definition(
session,
principal,
reference=current.definition_ref,
effective_at=recorded_at,
)
_require_published(definition)
if current.status != "draft" or not definition.allow_drafts:
raise FormRuntimeError(
"This Form instance does not allow intermediate draft updates."
)
self._evaluate_policy(
session,
principal,
definition=definition,
action="save_draft",
instance=current,
)
return self._revise(
session,
principal,
current=current,
identity=identity,
expected_revision=expected_revision,
status="draft",
values=values,
attachment_refs=attachment_refs,
signature_refs=signature_refs,
handoff_refs=current.handoff_refs,
idempotency_key=idempotency_key,
recorded_at=recorded_at,
change_reason=change_reason,
operation="draft_saved",
final_validation=False,
definition=definition,
)
def submit_instance(
self,
session: Session,
principal: object,
*,
instance_id: str,
expected_revision: int,
values: Mapping[str, object],
attachment_refs: Sequence[EvidenceReference],
signature_refs: Sequence[EvidenceReference],
idempotency_key: str,
recorded_at: datetime,
allow_all: bool = False,
) -> FormInstance:
current, identity = _current_instance(
session,
principal,
instance_id=instance_id,
lock=True,
allow_all=allow_all,
)
definition = self._definition(
session,
principal,
reference=current.definition_ref,
effective_at=recorded_at,
)
_require_published(definition)
self._evaluate_policy(
session,
principal,
definition=definition,
action="submit",
instance=current,
)
submitted = self._revise(
session,
principal,
current=current,
identity=identity,
expected_revision=expected_revision,
status="submitted",
values=values,
attachment_refs=attachment_refs,
signature_refs=signature_refs,
handoff_refs=current.handoff_refs,
idempotency_key=idempotency_key,
recorded_at=recorded_at,
change_reason="Form submitted.",
operation="submitted",
final_validation=True,
definition=definition,
allowed_current_statuses=("started", "draft"),
)
if not submitted.replayed:
from govoplan_forms_runtime.backend.status_access import (
FormStatusAccessService,
)
FormStatusAccessService(self._registry).ensure_for_submission(
session,
instance=submitted,
issued_at=recorded_at,
)
assisted_session = (
session.query(FormIntakeSession)
.filter(
FormIntakeSession.tenant_id == submitted.tenant_id,
FormIntakeSession.instance_id == submitted.instance_id,
FormIntakeSession.mode == "assisted",
)
.one_or_none()
)
if assisted_session is not None and assisted_session.status != "submitted":
assisted_session.status = "submitted"
assisted_session.submitted_at = recorded_at
session.add(assisted_session)
session.flush()
return submitted
def transition_instance(
self,
session: Session,
principal: object,
*,
instance_id: str,
expected_revision: int,
status: str,
idempotency_key: str,
recorded_at: datetime,
change_reason: str,
allow_all: bool = False,
) -> FormInstance:
current, identity = _current_instance(
session,
principal,
instance_id=instance_id,
lock=True,
allow_all=allow_all,
)
definition = self._definition(
session,
principal,
reference=current.definition_ref,
effective_at=recorded_at,
)
self._evaluate_policy(
session,
principal,
definition=definition,
action=f"transition:{status}",
instance=current,
)
return self._revise(
session,
principal,
current=current,
identity=identity,
expected_revision=expected_revision,
status=status,
values=current.values,
attachment_refs=current.attachment_refs,
signature_refs=current.signature_refs,
handoff_refs=current.handoff_refs,
idempotency_key=idempotency_key,
recorded_at=recorded_at,
change_reason=change_reason,
operation=f"status_{status}",
final_validation=status not in {"archived"},
definition=definition,
enforce_status_transition=True,
)
def handoff_instance(
self,
session: Session,
principal: object,
*,
instance_id: str,
expected_revision: int,
target_ref: InstitutionalReference,
idempotency_key: str,
recorded_at: datetime,
change_reason: str,
allow_all: bool = False,
) -> FormInstance:
current, identity = _current_instance(
session,
principal,
instance_id=instance_id,
lock=True,
allow_all=allow_all,
)
definition = self._definition(
session,
principal,
reference=current.definition_ref,
effective_at=recorded_at,
)
if target_ref.tenant_id != current.tenant_id:
raise FormRuntimeError("Form handoff cannot cross tenants.")
if target_ref.kind in {"case", "workflow"}:
raise FormRuntimeError(
"Case and Workflow handoffs must use the governed native handoff endpoint."
)
if target_ref.kind not in definition.handoff_kinds:
raise FormRuntimeError(
f"Form definition does not permit a {target_ref.kind!r} handoff."
)
self._evaluate_policy(
session,
principal,
definition=definition,
action="handoff",
instance=current,
)
handoffs = tuple(dict.fromkeys((*current.handoff_refs, target_ref)))
return self._revise(
session,
principal,
current=current,
identity=identity,
expected_revision=expected_revision,
status="handed_off",
values=current.values,
attachment_refs=current.attachment_refs,
signature_refs=current.signature_refs,
handoff_refs=handoffs,
idempotency_key=idempotency_key,
recorded_at=recorded_at,
change_reason=change_reason,
operation="handed_off",
final_validation=True,
definition=definition,
allowed_current_statuses=(
"submitted",
"validated",
"needs_review",
"accepted",
),
)
def get_instance(
self,
session: Session,
principal: object,
*,
instance_id: str,
revision: int | None = None,
allow_all: bool = False,
) -> FormInstance | None:
tenant_id = _principal_tenant(principal)
identity = (
session.query(FormInstanceIdentity)
.filter(
FormInstanceIdentity.tenant_id == tenant_id,
FormInstanceIdentity.instance_id == instance_id,
)
.one_or_none()
)
if identity is None:
return None
_assert_access(identity, principal, allow_all=allow_all)
query = session.query(FormInstanceRevision).filter(
FormInstanceRevision.tenant_id == tenant_id,
FormInstanceRevision.instance_id == instance_id,
)
if revision is None:
query = query.filter(FormInstanceRevision.superseded_at.is_(None))
else:
query = query.filter(FormInstanceRevision.revision == revision)
row = query.order_by(FormInstanceRevision.revision.desc()).first()
return _instance_from_row(row) if row is not None else None
def get_instance_definition(
self,
session: Session,
principal: object,
*,
instance_id: str,
allow_all: bool = False,
) -> FormDefinition | None:
instance = self.get_instance(
session,
principal,
instance_id=instance_id,
allow_all=allow_all,
)
if instance is None:
return None
return self._definition(
session,
principal,
reference=instance.definition_ref,
effective_at=instance.recorded_at,
)
def list_instances(
self,
session: Session,
principal: object,
*,
statuses: Sequence[str] | None = None,
definition_id: str | None = None,
offset: int = 0,
limit: int = 100,
allow_all: bool = False,
) -> tuple[tuple[FormInstance, ...], int]:
tenant_id = _principal_tenant(principal)
if offset < 0 or not 1 <= limit <= 200:
raise FormRuntimeError(
"Form instance offset must be non-negative and limit between 1 and 200."
)
statement = (
session.query(FormInstanceRevision)
.join(
FormInstanceIdentity,
FormInstanceIdentity.id == FormInstanceRevision.identity_id,
)
.filter(
FormInstanceRevision.tenant_id == tenant_id,
FormInstanceRevision.superseded_at.is_(None),
)
)
if not allow_all:
statement = statement.filter(
FormInstanceIdentity.created_by == _principal_actor(principal)
)
if statuses:
statement = statement.filter(
FormInstanceRevision.status.in_(tuple(statuses))
)
if definition_id:
statement = statement.filter(
FormInstanceIdentity.definition_id == definition_id
)
total = int(statement.with_entities(func.count()).scalar() or 0)
rows = (
statement.order_by(FormInstanceRevision.recorded_at.desc())
.offset(offset)
.limit(limit)
.all()
)
return tuple(_instance_from_row(row) for row in rows), total
def history(
self,
session: Session,
principal: object,
*,
instance_id: str,
limit: int = 100,
allow_all: bool = False,
) -> tuple[FormInstance, ...]:
current = self.get_instance(
session,
principal,
instance_id=instance_id,
allow_all=allow_all,
)
if current is None:
return ()
rows = (
session.query(FormInstanceRevision)
.filter(
FormInstanceRevision.tenant_id == current.tenant_id,
FormInstanceRevision.instance_id == instance_id,
)
.order_by(FormInstanceRevision.revision.desc())
.limit(max(1, min(limit, 200)))
.all()
)
return tuple(_instance_from_row(row) for row in rows)
def events(
self,
session: Session,
principal: object,
*,
instance_id: str,
limit: int = 200,
allow_all: bool = False,
) -> tuple[Mapping[str, object], ...]:
current = self.get_instance(
session,
principal,
instance_id=instance_id,
allow_all=allow_all,
)
if current is None:
return ()
rows = (
session.query(FormInstanceEvent)
.filter(
FormInstanceEvent.tenant_id == current.tenant_id,
FormInstanceEvent.instance_id == instance_id,
)
.order_by(FormInstanceEvent.occurred_at.asc())
.limit(max(1, min(limit, 500)))
.all()
)
return tuple(
{
"event_id": row.event_id,
"event_type": row.event_type,
"instance_revision": row.instance_revision,
"status": row.status,
"occurred_at": _aware(row.occurred_at).isoformat(),
"actor_id": row.actor_id,
"payload": dict(row.payload),
}
for row in rows
)
def create_evidence_grant(
self,
session: Session,
principal: object,
*,
instance_id: str,
expected_revision: int,
provider_id: str,
custodian_ref: str,
purpose: str,
idempotency_key: str,
expires_at: datetime,
max_size_bytes: int | None = None,
allowed_content_types: Sequence[str] = (),
attachment_refs: Sequence[EvidenceReference] = (),
allow_all: bool = False,
):
current, _identity = _current_instance(
session,
principal,
instance_id=instance_id,
lock=True,
allow_all=allow_all,
)
if current.revision != expected_revision:
raise FormRuntimeError(
"Form instance revision conflict: the expected revision is stale."
)
definition = self._definition(
session,
principal,
reference=current.definition_ref,
effective_at=current.recorded_at,
)
try:
return self._evidence.create_upload_grant(
session,
principal,
instance=current,
definition=definition,
provider_id=provider_id,
custodian_ref=custodian_ref,
purpose=purpose,
idempotency_key=idempotency_key,
expires_at=expires_at,
max_size_bytes=max_size_bytes,
allowed_content_types=allowed_content_types,
attachment_refs=attachment_refs,
)
except ValueError as exc:
raise FormRuntimeError(str(exc)) from exc
def acknowledge(
self,
session: Session,
principal: object,
*,
instance_id: str,
expected_revision: int,
statement_id: str,
statement_version: str,
values: Mapping[str, object],
attachment_refs: Sequence[EvidenceReference],
accepted_at: datetime,
idempotency_key: str,
allow_all: bool = False,
) -> EvidenceReference:
current, _identity = _current_instance(
session,
principal,
instance_id=instance_id,
lock=True,
allow_all=allow_all,
)
definition = self._definition(
session,
principal,
reference=current.definition_ref,
effective_at=accepted_at,
)
clean_values = normalize_form_values(
definition,
_mapping_copy(values, "Form acknowledgement values"),
)
attachments = tuple(attachment_refs)
if any(item.tenant_id != current.tenant_id for item in attachments):
raise FormRuntimeError(
"Form acknowledgement attachments cannot cross tenants."
)
try:
return self._evidence.create_acknowledgement(
session,
principal,
instance=current,
expected_revision=expected_revision,
statement_id=statement_id,
statement_version=statement_version,
values=clean_values,
attachment_refs=attachments,
accepted_at=accepted_at,
idempotency_key=idempotency_key,
)
except ValueError as exc:
raise FormRuntimeError(str(exc)) from exc
def _revise(
self,
session: Session,
principal: object,
*,
current: FormInstance,
identity: FormInstanceIdentity,
expected_revision: int,
status: str,
values: Mapping[str, object],
attachment_refs: Sequence[EvidenceReference],
signature_refs: Sequence[EvidenceReference],
handoff_refs: Sequence[InstitutionalReference],
idempotency_key: str,
recorded_at: datetime,
change_reason: str,
operation: str,
final_validation: bool,
definition: FormDefinition,
allowed_current_statuses: Sequence[str] | None = None,
enforce_status_transition: bool = False,
) -> FormInstance:
_require_aware(recorded_at, "Form instance recorded_at")
clean_reason = _text(change_reason, "Form instance change reason", 1000)
clean_values = normalize_form_values(
definition,
_mapping_copy(values, "Form values"),
)
attachments = tuple(attachment_refs)
signatures = tuple(signature_refs)
handoffs = tuple(handoff_refs)
diagnostics = validate_form_values(
definition,
clean_values,
attachment_refs=attachments,
signature_refs=signatures,
final=final_validation,
)
evidence_diagnostics, evidence_snapshots = self._inspect_evidence(
session,
principal,
instance=current,
definition=definition,
values=clean_values,
attachment_refs=attachments,
signature_refs=signatures,
purpose=operation.replace("_", " "),
final=final_validation,
observed_at=recorded_at,
)
diagnostics = (*diagnostics, *evidence_diagnostics)
request = {
"operation": operation,
"instance_id": current.instance_id,
"expected_revision": expected_revision,
"status": status,
"values": clean_values,
"attachment_refs": [item.to_dict() for item in attachments],
"signature_refs": [item.to_dict() for item in signatures],
"handoff_refs": [item.to_dict() for item in handoffs],
"recorded_at": recorded_at.isoformat(),
"change_reason": clean_reason,
}
request_sha256 = _request_hash(request)
replay = _replay(
session,
principal,
tenant_id=current.tenant_id,
idempotency_key=idempotency_key,
request_sha256=request_sha256,
)
if replay is not None:
return replay
if operation == "submitted" and _is_assisted_instance(current):
assisted_session = (
session.query(FormIntakeSession)
.filter(
FormIntakeSession.tenant_id == current.tenant_id,
FormIntakeSession.instance_id == current.instance_id,
FormIntakeSession.mode == "assisted",
)
.one_or_none()
)
if (
assisted_session is None
or assisted_session.status != "active"
or _aware(assisted_session.expires_at) <= recorded_at
):
raise FormRuntimeError(
"Assisted intake submission requires an active, unexpired session."
)
expected_confirmation_digest = assisted_submission_payload_sha256(
current,
values=clean_values,
attachment_refs=attachments,
signature_refs=signatures,
)
confirmation = (
session.query(FormAssistedConfirmation)
.filter(
FormAssistedConfirmation.tenant_id == current.tenant_id,
FormAssistedConfirmation.instance_id == current.instance_id,
FormAssistedConfirmation.instance_revision == current.revision,
FormAssistedConfirmation.payload_sha256
== expected_confirmation_digest,
)
.order_by(FormAssistedConfirmation.confirmed_at.desc())
.first()
)
if confirmation is None:
raise FormRuntimeError(
"Assisted intake submission requires a current read-back confirmation for the exact values and evidence."
)
if allowed_current_statuses is not None and current.status not in set(
allowed_current_statuses
):
raise FormRuntimeError(
f"Form status {current.status!r} does not permit this operation."
)
if enforce_status_transition and status not in _STATUS_TRANSITIONS.get(
current.status, frozenset()
):
raise FormRuntimeError(
f"Form status transition {current.status!r} to {status!r} is not allowed."
)
if current.revision != expected_revision:
raise FormRuntimeError(
"Form instance revision conflict: the expected revision is stale."
)
actor_id = _principal_actor(principal)
instance = FormInstance(
tenant_id=current.tenant_id,
instance_id=current.instance_id,
revision=current.revision + 1,
status=status,
definition_ref=current.definition_ref,
values=clean_values,
validation_results=diagnostics,
attachment_refs=attachments,
signature_refs=signatures,
handoff_refs=handoffs,
service_ref=current.service_ref,
service_binding=current.service_binding,
receipt_id=(
str(uuid.uuid4()) if status == "submitted" else current.receipt_id
),
recorded_at=recorded_at,
change_reason=clean_reason,
created_by=current.created_by,
changed_by=actor_id,
metadata=(
{
**dict(current.metadata),
"evidence_verification": list(evidence_snapshots),
}
if evidence_snapshots
else current.metadata
),
)
current_row = (
session.query(FormInstanceRevision)
.filter(
FormInstanceRevision.tenant_id == current.tenant_id,
FormInstanceRevision.instance_id == current.instance_id,
FormInstanceRevision.revision == current.revision,
FormInstanceRevision.superseded_at.is_(None),
)
.with_for_update()
.one_or_none()
)
if current_row is None:
raise FormRuntimeError(
"Form instance revision conflict: the current revision changed."
)
return _record_instance(
session,
principal,
identity=identity,
current=current_row,
instance=instance,
idempotency_key=idempotency_key,
request_sha256=request_sha256,
operation=operation,
)
def _inspect_evidence(
self,
session: Session,
principal: object,
*,
instance: FormInstance,
definition: FormDefinition,
values: Mapping[str, object],
attachment_refs: Sequence[EvidenceReference],
signature_refs: Sequence[EvidenceReference],
purpose: str,
final: bool,
observed_at: datetime,
) -> tuple[tuple[Mapping[str, object], ...], tuple[Mapping[str, object], ...]]:
if not attachment_refs and not signature_refs:
return (), ()
try:
return self._evidence.inspect(
session,
principal,
instance=instance,
definition=definition,
values=values,
attachment_refs=attachment_refs,
signature_refs=signature_refs,
purpose=purpose,
final=final,
observed_at=observed_at,
)
except ValueError as exc:
raise FormRuntimeError(str(exc)) from exc
def _definition(
self,
session: Session,
principal: object,
*,
reference: InstitutionalReference,
effective_at: datetime,
) -> FormDefinition:
provider = _capability(self._registry, CAPABILITY_FORM_DEFINITIONS)
if not isinstance(provider, FormDefinitionProvider):
raise FormRuntimeError(
"The configured Forms definition provider is unavailable or invalid."
)
definition = provider.get_form_definition(
session,
principal,
reference=reference,
effective_at=effective_at,
)
if definition is None:
raise FormRuntimeError(
"The exact Form definition was not found or effective."
)
if not _same_exact_form_reference(definition.reference, reference):
raise FormRuntimeError(
"The Forms provider returned a different definition or revision."
)
if not definition.temporal.effective_at(effective_at):
raise FormRuntimeError(
"The exact Form definition is not effective at the requested time."
)
return definition
def _evaluate_policy(
self,
session: Session,
principal: object,
*,
definition: FormDefinition,
action: str,
instance: FormInstance | None,
) -> None:
if not definition.policy_refs:
return
evaluator = _capability(
self._registry,
CAPABILITY_FORMS_RUNTIME_POLICY_EVALUATOR,
required=False,
)
if not isinstance(evaluator, FormRuntimePolicyEvaluator):
raise PermissionError(
"This Form requires a policy evaluator that is not available."
)
if not evaluator.evaluate_form_access(
session,
principal,
definition=definition,
action=action,
instance=instance,
):
raise PermissionError("Form policy denied this operation.")
class FormsServiceLauncher:
def __init__(self, registry: object | None) -> None:
self._runtime = FormRuntimeService(registry)
def launch_service(
self,
session: object,
principal: object,
*,
definition: ServiceDefinition,
request: ServiceLaunchRequest,
) -> ServiceLaunchResult:
db = _session(session)
tenant_id = _principal_tenant(principal)
if (
definition.reference != request.service_ref
or request.binding.kind != "form"
or request.binding not in definition.bindings
or definition.reference.tenant_id != tenant_id
or definition.publication_state != "published"
):
raise FormRuntimeError(
"Form Service launch requires the exact published Service and binding."
)
form_id, form_revision = parse_form_binding_reference(request.binding.reference)
definition_ref = InstitutionalReference(
kind="form",
owner_module="forms",
object_id=form_id,
tenant_id=tenant_id,
version=form_revision,
valid_at=request.requested_at,
)
instance = self._runtime.create_instance(
db,
principal,
definition_ref=definition_ref,
values=request.parameters,
idempotency_key=f"service-launch:{request.idempotency_key}",
recorded_at=request.requested_at,
service_ref=request.service_ref,
service_binding=request.binding,
metadata={"launch_source": "portal"},
)
return ServiceLaunchResult(
service_ref=request.service_ref,
binding=request.binding,
state="started",
target_ref=instance.reference,
href=f"/forms-runtime/{instance.instance_id}",
replayed=instance.replayed,
metadata={
"form_instance_id": instance.instance_id,
"form_instance_revision": instance.revision,
"form_definition_id": form_id,
"form_definition_revision": form_revision,
"status": instance.status,
},
)
def parse_form_binding_reference(value: str) -> tuple[str, str]:
clean = str(value or "").strip()
if "/" not in clean:
raise FormRuntimeError(
"Form Service bindings must use the exact '<form-id>/<revision>' format."
)
form_id, revision = clean.rsplit("/", 1)
return (
_identifier(form_id, "Form binding form id"),
_identifier(revision, "Form binding revision"),
)
def validate_form_values(
definition: FormDefinition,
values: Mapping[str, object],
*,
attachment_refs: Sequence[EvidenceReference],
signature_refs: Sequence[EvidenceReference],
final: bool,
) -> tuple[Mapping[str, object], ...]:
fields = {item.key: item for item in definition.fields}
diagnostics: list[Mapping[str, object]] = []
unknown = sorted(set(values) - set(fields))
for key in unknown:
diagnostics.append(
_diagnostic(
key,
"error",
"field.unknown",
"This field is not part of the exact Form revision.",
)
)
visible_fields = visible_form_field_keys(definition, values)
for field in definition.fields:
if field.key not in visible_fields:
continue
present = field.key in values and values[field.key] not in (None, "")
if field.required and not present:
diagnostics.append(
_diagnostic(
field.key,
"error" if final else "warning",
"field.required",
"A value is required before submission.",
)
)
if present:
diagnostics.extend(_validate_field_value(field, values[field.key]))
if len(attachment_refs) > definition.max_attachments:
diagnostics.append(
_diagnostic(
None,
"error",
"attachments.limit",
f"This Form permits at most {definition.max_attachments} attachments.",
)
)
if final and definition.signature_requirement == "required" and not signature_refs:
diagnostics.append(
_diagnostic(
None,
"error",
"signature.required",
"A signature is required before submission.",
)
)
errors = [item for item in diagnostics if item["severity"] == "error"]
if errors:
summary = "; ".join(str(item["message"]) for item in errors[:5])
raise FormRuntimeError(f"Form values failed validation: {summary}")
return tuple(diagnostics)
def _is_assisted_instance(instance: FormInstance) -> bool:
intake = instance.metadata.get("intake")
return isinstance(intake, Mapping) and intake.get("mode") == "assisted"
def normalize_form_values(
definition: FormDefinition,
values: Mapping[str, object],
) -> dict[str, object]:
"""Drop declared fields hidden by authoritative conditions.
Unknown fields remain in the payload so regular validation reports them
instead of silently accepting misspelled or retired keys.
"""
result = _mapping_copy(values, "Form values")
declared = {item.key for item in definition.fields}
for _ in range(len(definition.fields) + 1):
visible = visible_form_field_keys(definition, result)
next_result = {
key: value
for key, value in result.items()
if key not in declared or key in visible
}
if next_result == result:
return next_result
result = next_result
raise FormRuntimeError(
"Form visibility conditions did not converge; verify the published definition."
)
def visible_form_field_keys(
definition: FormDefinition,
values: Mapping[str, object],
) -> frozenset[str]:
visible = {
item.key
for item in definition.fields
if item.visibility_condition is None
or evaluate_form_condition(item.visibility_condition, values)
}
if not definition.pages:
return frozenset(visible)
placed_visible: set[str] = set()
for page in definition.pages:
if page.visibility_condition is not None and not evaluate_form_condition(
page.visibility_condition,
values,
):
continue
for section in page.sections:
if section.visibility_condition is not None and not evaluate_form_condition(
section.visibility_condition, values
):
continue
placed_visible.update(section.field_keys)
return frozenset(visible & placed_visible)
def evaluate_form_condition(
condition: FormConditionExpression,
values: Mapping[str, object],
) -> bool:
if condition.kind == "all":
return all(
evaluate_form_condition(item, values) for item in condition.conditions
)
if condition.kind == "any":
return any(
evaluate_form_condition(item, values) for item in condition.conditions
)
if condition.kind == "not":
return not evaluate_form_condition(condition.conditions[0], values)
actual = values.get(str(condition.field_key))
expected = condition.value
operator = condition.operator
if operator == "eq":
return actual == expected
if operator == "neq":
return actual != expected
if operator == "is_empty":
return _empty_value(actual)
if operator == "is_not_empty":
return not _empty_value(actual)
if operator == "in":
return actual in expected # type: ignore[operator]
if operator == "not_in":
return actual not in expected # type: ignore[operator]
if operator == "contains":
try:
return expected in actual # type: ignore[operator]
except TypeError:
return False
try:
if operator == "lt":
return actual < expected # type: ignore[operator]
if operator == "lte":
return actual <= expected # type: ignore[operator]
if operator == "gt":
return actual > expected # type: ignore[operator]
if operator == "gte":
return actual >= expected # type: ignore[operator]
except TypeError:
return False
raise FormRuntimeError(f"Unsupported Form condition operator: {operator!r}.")
def _empty_value(value: object) -> bool:
if value is None or value == "":
return True
if isinstance(value, (Mapping, Sequence)) and not isinstance(value, (str, bytes)):
return len(value) == 0
return False
def _validate_field_value(
field: FormFieldDefinition,
value: object,
) -> tuple[Mapping[str, object], ...]:
valid = True
if field.value_type in {"text", "multiline_text", "email", "choice"}:
valid = isinstance(value, str)
elif field.value_type == "integer":
valid = isinstance(value, int) and not isinstance(value, bool)
elif field.value_type == "number":
valid = isinstance(value, (int, float)) and not isinstance(value, bool)
elif field.value_type == "boolean":
valid = isinstance(value, bool)
elif field.value_type == "object":
valid = isinstance(value, Mapping)
elif field.value_type in {"list", "multi_choice"}:
valid = isinstance(value, Sequence) and not isinstance(value, (str, bytes))
elif field.value_type == "date":
valid = _is_iso_date(value, include_time=False)
elif field.value_type == "datetime":
valid = _is_iso_date(value, include_time=True)
if not valid:
return (
_diagnostic(
field.key,
"error",
"field.type",
f"The value must use the {field.value_type} type.",
),
)
diagnostics: list[Mapping[str, object]] = []
if field.value_type == "email" and not _EMAIL_RE.fullmatch(str(value)):
diagnostics.append(
_diagnostic(
field.key, "error", "field.email", "Enter a valid email address."
)
)
if field.value_type == "choice" and value not in field.options:
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.option",
"Select one of the declared options.",
)
)
if field.value_type == "multi_choice" and any(
item not in field.options
for item in value # type: ignore[union-attr]
):
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.option",
"Every selected value must be a declared option.",
)
)
constraints = field.constraints
if isinstance(value, str):
minimum = constraints.get("min_length")
maximum = constraints.get("max_length")
pattern = constraints.get("pattern")
if isinstance(minimum, int) and len(value) < minimum:
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.min_length",
f"Enter at least {minimum} characters.",
)
)
if isinstance(maximum, int) and len(value) > maximum:
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.max_length",
f"Enter at most {maximum} characters.",
)
)
if isinstance(pattern, str):
try:
matches = re.fullmatch(pattern, value) is not None
except re.error as exc:
raise FormRuntimeError(
f"Form definition field {field.key!r} has an invalid pattern."
) from exc
if not matches:
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.pattern",
"The value does not match the required format.",
)
)
if isinstance(value, (int, float)) and not isinstance(value, bool):
minimum = constraints.get("minimum")
maximum = constraints.get("maximum")
if isinstance(minimum, (int, float)) and value < minimum:
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.minimum",
f"Enter a value of at least {minimum}.",
)
)
if isinstance(maximum, (int, float)) and value > maximum:
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.maximum",
f"Enter a value no greater than {maximum}.",
)
)
return tuple(diagnostics)
def _diagnostic(
field: str | None,
severity: str,
code: str,
message: str,
) -> Mapping[str, object]:
return {"field": field, "severity": severity, "code": code, "message": message}
def _record_instance(
session: Session,
principal: object,
*,
identity: FormInstanceIdentity,
current: FormInstanceRevision | None,
instance: FormInstance,
idempotency_key: str,
request_sha256: str,
operation: str,
) -> FormInstance:
clean_key = _text(idempotency_key, "Form idempotency key", 255)
if current is not None:
current.superseded_at = instance.recorded_at
row = FormInstanceRevision(
tenant_id=instance.tenant_id,
instance_id=instance.instance_id,
identity_id=identity.id,
revision=instance.revision,
previous_revision_id=current.id if current is not None else None,
status=instance.status,
recorded_at=instance.recorded_at,
snapshot=instance.to_dict(),
changed_by=instance.changed_by,
)
event_id = str(uuid.uuid4())
event = FormInstanceEvent(
tenant_id=instance.tenant_id,
instance_id=instance.instance_id,
instance_revision=instance.revision,
event_id=event_id,
event_type=f"forms_runtime.instance.{operation}",
status=instance.status,
occurred_at=instance.recorded_at,
actor_id=_principal_actor(principal),
idempotency_key=clean_key,
request_sha256=request_sha256,
payload={
"definition_id": instance.definition_ref.object_id,
"definition_revision": instance.definition_ref.version,
"status": instance.status,
"revision": instance.revision,
"validation_warning_count": sum(
item.get("severity") == "warning"
for item in instance.validation_results
),
"attachment_count": len(instance.attachment_refs),
"signature_count": len(instance.signature_refs),
"handoff_count": len(instance.handoff_refs),
"receipt_id": instance.receipt_id,
},
)
session.add_all((row, event))
session.flush()
emit_platform_event(
session,
PlatformEvent(
event_id=event_id,
type=event.event_type,
module_id="forms_runtime",
payload=dict(event.payload),
occurred_at=instance.recorded_at,
actor=EventActorRef(type="account", id=_principal_actor(principal)),
tenant=EventTenantRef(id=instance.tenant_id),
resource=EventObjectRef(
type="form_submission",
id=instance.instance_id,
label=instance.definition_ref.label,
),
classification="confidential",
),
)
return _instance_from_row(row)
def _current_instance(
session: Session,
principal: object,
*,
instance_id: str,
lock: bool,
allow_all: bool,
) -> tuple[FormInstance, FormInstanceIdentity]:
tenant_id = _principal_tenant(principal)
identity = (
session.query(FormInstanceIdentity)
.filter(
FormInstanceIdentity.tenant_id == tenant_id,
FormInstanceIdentity.instance_id == instance_id,
)
.one_or_none()
)
if identity is None:
raise LookupError("Form instance not found.")
_assert_access(identity, principal, allow_all=allow_all)
query = session.query(FormInstanceRevision).filter(
FormInstanceRevision.tenant_id == tenant_id,
FormInstanceRevision.instance_id == instance_id,
FormInstanceRevision.superseded_at.is_(None),
)
if lock:
query = query.with_for_update()
row = query.one_or_none()
if row is None:
raise LookupError("Form instance has no current revision.")
return _instance_from_row(row), identity
def _replay(
session: Session,
principal: object,
*,
tenant_id: str,
idempotency_key: str,
request_sha256: str,
) -> FormInstance | None:
clean_key = _text(idempotency_key, "Form idempotency key", 255)
event = (
session.query(FormInstanceEvent)
.filter(
FormInstanceEvent.tenant_id == tenant_id,
FormInstanceEvent.idempotency_key == clean_key,
)
.one_or_none()
)
if event is None:
return None
if event.actor_id != _principal_actor(principal):
raise FormRuntimeError(
"Form idempotency conflict: this key belongs to another actor."
)
if event.request_sha256 != request_sha256:
raise FormRuntimeError(
"Form idempotency conflict: this key was used for another request."
)
row = (
session.query(FormInstanceRevision)
.filter(
FormInstanceRevision.tenant_id == tenant_id,
FormInstanceRevision.instance_id == event.instance_id,
FormInstanceRevision.revision == event.instance_revision,
)
.one()
)
return _instance_from_row(row).with_replay()
def _same_exact_form_reference(
actual: InstitutionalReference,
requested: InstitutionalReference,
) -> bool:
return (
actual.kind == requested.kind == "form"
and actual.owner_module == requested.owner_module == "forms"
and actual.object_id == requested.object_id
and actual.tenant_id == requested.tenant_id
and actual.version == requested.version
)
def _instance_from_row(row: FormInstanceRevision) -> FormInstance:
value = dict(row.snapshot)
definition_ref = InstitutionalReference.from_mapping(
_required_mapping(value, "definition_ref")
)
service_ref_payload = value.get("service_ref")
binding_payload = value.get("service_binding")
return FormInstance(
tenant_id=str(value["tenant_id"]),
instance_id=str(value["instance_id"]),
revision=int(value["revision"]),
status=str(value["status"]),
definition_ref=definition_ref,
values=_required_mapping(value, "values"),
validation_results=tuple(
dict(item) for item in _mapping_list(value.get("validation_results"))
),
attachment_refs=tuple(
EvidenceReference.from_mapping(item)
for item in _mapping_list(value.get("attachment_refs"))
),
signature_refs=tuple(
EvidenceReference.from_mapping(item)
for item in _mapping_list(value.get("signature_refs"))
),
handoff_refs=tuple(
InstitutionalReference.from_mapping(item)
for item in _mapping_list(value.get("handoff_refs"))
),
service_ref=(
InstitutionalReference.from_mapping(service_ref_payload)
if isinstance(service_ref_payload, Mapping)
else None
),
service_binding=(
ServiceBinding.from_mapping(binding_payload)
if isinstance(binding_payload, Mapping)
else None
),
receipt_id=_optional_text(value.get("receipt_id")),
recorded_at=_aware(datetime.fromisoformat(str(value["recorded_at"]))),
change_reason=str(value["change_reason"]),
created_by=str(value["created_by"]),
changed_by=str(value["changed_by"]),
replayed=bool(value.get("replayed", False)),
metadata=dict(value.get("metadata") or {}),
)
def _require_published(definition: FormDefinition) -> None:
if definition.publication_state != "published":
raise FormRuntimeError("Only a published Form definition can be used.")
def _require_startable(definition: FormDefinition) -> None:
_require_published(definition)
if definition.temporal.superseded_at is not None:
raise FormRuntimeError(
"A superseded Form definition cannot start a new instance."
)
def _assert_access(
identity: FormInstanceIdentity,
principal: object,
*,
allow_all: bool,
) -> None:
if not allow_all and identity.created_by != _principal_actor(principal):
raise PermissionError("Form instance access is denied.")
def _principal_tenant(principal: object) -> str:
tenant_id = str(getattr(principal, "tenant_id", "") or "").strip()
if not tenant_id:
raise InstitutionalContextError(
"Forms Runtime operations require a tenant-bound principal."
)
return tenant_id
def _principal_actor(principal: object) -> str:
for name in ("account_id", "identity_id", "membership_id"):
value = str(getattr(principal, name, "") or "").strip()
if value:
return value
raise InstitutionalContextError(
"Forms Runtime operations require an acting identity."
)
def _capability(
registry: object | None,
name: str,
*,
required: bool = True,
) -> object | None:
if registry is None or not hasattr(registry, "has_capability"):
if required:
raise FormRuntimeError(f"Required capability is unavailable: {name}")
return None
if not registry.has_capability(name):
if required:
raise FormRuntimeError(f"Required capability is unavailable: {name}")
return None
if hasattr(registry, "require_capability"):
return registry.require_capability(name)
if hasattr(registry, "capability"):
return registry.capability(name)
if required:
raise FormRuntimeError(f"Required capability cannot be resolved: {name}")
return None
def _session(value: object) -> Session:
if not hasattr(value, "query"):
raise FormRuntimeError("Forms Runtime requires a database session.")
return value # type: ignore[return-value]
def _request_hash(value: Mapping[str, object]) -> str:
try:
payload = json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
default=_json_default,
)
except (TypeError, ValueError) as exc:
raise FormRuntimeError("Form values must be JSON serializable.") from exc
if len(payload.encode("utf-8")) > 2_000_000:
raise FormRuntimeError("Form instance payload exceeds the 2 MB limit.")
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _json_default(value: object) -> object:
if isinstance(value, (date, datetime)):
return value.isoformat()
raise TypeError(f"Unsupported JSON value: {type(value).__name__}")
def _mapping_copy(value: Mapping[str, object], label: str) -> dict[str, object]:
if not isinstance(value, Mapping):
raise FormRuntimeError(f"{label} must be an object.")
return {str(key): item for key, item in value.items()}
def _mapping_list(value: object | None) -> tuple[Mapping[str, object], ...]:
if value is None:
return ()
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
raise FormRuntimeError("Expected a list of object payloads.")
if any(not isinstance(item, Mapping) for item in value):
raise FormRuntimeError("Expected object payloads in list.")
return tuple(value) # type: ignore[return-value]
def _required_mapping(value: Mapping[str, object], key: str) -> Mapping[str, object]:
item = value.get(key)
if not isinstance(item, Mapping):
raise FormRuntimeError(f"Form snapshot {key} must be an object.")
return item
def _identifier(value: str, label: str) -> str:
clean = str(value or "").strip()
if (
not clean
or len(clean) > 255
or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:/-]*", clean)
):
raise FormRuntimeError(f"{label} is invalid.")
return clean
def _text(value: str, label: str, maximum: int) -> str:
clean = str(value or "").strip()
if not clean or len(clean) > maximum:
raise FormRuntimeError(
f"{label} is required and limited to {maximum} characters."
)
return clean
def _optional_text(value: object | None) -> str | None:
clean = str(value or "").strip()
return clean or None
def _require_aware(value: datetime, label: str) -> None:
if value.tzinfo is None or value.utcoffset() is None:
raise FormRuntimeError(f"{label} must include a timezone.")
def _aware(value: datetime) -> datetime:
return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
def _is_iso_date(value: object, *, include_time: bool) -> bool:
if not isinstance(value, str):
return False
try:
if include_time:
parsed = datetime.fromisoformat(value)
return parsed.tzinfo is not None and parsed.utcoffset() is not None
date.fromisoformat(value)
return True
except ValueError:
return False
__all__ = [
"CAPABILITY_FORMS_RUNTIME_POLICY_EVALUATOR",
"CAPABILITY_FORMS_RUNTIME_REGISTRY",
"CAPABILITY_FORMS_RUNTIME_SERVICE_LAUNCHER",
"FormRuntimeError",
"FormRuntimePolicyEvaluator",
"FormRuntimeService",
"FormsServiceLauncher",
"evaluate_form_condition",
"normalize_form_values",
"parse_form_binding_reference",
"validate_form_values",
"visible_form_field_keys",
]