GovOPlaN/govoplan-forms-runtime
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add governed form handoffs

Browse Source
This commit is contained in:
zemion
2026-08-01 20:57:26 +02:00
parent 9dc49fe27b
commit b4388b1e1e
15 changed files with 2236 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_forms_runtime/backend/db/models.py
+66 -1
View File
@@ -4,7 +4,15 @@ from datetime import datetime
from typing import Any
import uuid
from sqlalchemy import DateTime, ForeignKey, Index, Integer, JSON, String, UniqueConstraint
from sqlalchemy import (
DateTime,
ForeignKey,
Index,
Integer,
JSON,
String,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column
from govoplan_core.db.base import Base, TimestampMixin
@@ -125,8 +133,65 @@ class FormInstanceEvent(Base, TimestampMixin):
payload: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
class FormHandoffEffect(Base, TimestampMixin):
__tablename__ = "form_handoff_effects"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"effect_id",
name="uq_form_handoff_effect",
),
UniqueConstraint(
"tenant_id",
"idempotency_key",
name="uq_form_handoff_idempotency",
),
UniqueConstraint(
"tenant_id",
"provider_key",
name="uq_form_handoff_provider_key",
),
Index(
"ix_form_handoff_instance_state",
"tenant_id",
"instance_id",
"state",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
instance_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
effect_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
instance_revision: Mapped[int] = mapped_column(Integer, nullable=False)
idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
provider_key: Mapped[str] = mapped_column(String(255), nullable=False)
request_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
binding_kind: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
binding_reference: Mapped[str] = mapped_column(String(500), nullable=False)
provider_capability: Mapped[str] = mapped_column(String(120), nullable=False)
state: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
requested_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, index=True
)
resolved_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, index=True
)
target_ref: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
href: Mapped[str | None] = mapped_column(String(2000), nullable=True)
evidence: Mapped[list[dict[str, Any]]] = mapped_column(
JSON, default=list, nullable=False
)
last_error: Mapped[str | None] = mapped_column(String(2000), nullable=True)
details: Mapped[dict[str, Any]] = mapped_column(
"metadata", JSON, default=dict, nullable=False
)
__all__ = [
"FormInstanceEvent",
"FormHandoffEffect",
"FormInstanceIdentity",
"FormInstanceRevision",
]
src/govoplan_forms_runtime/backend/domain.py
+86 -2
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from dataclasses import dataclass, field, replace
from datetime import datetime
from typing import Mapping
from typing import Literal, Mapping
from govoplan_core.core.institutional import (
EvidenceReference,
@@ -25,6 +25,84 @@ FORM_INSTANCE_STATUSES = frozenset(
"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)
@@ -139,4 +217,10 @@ class FormInstance:
}
__all__ = ["FORM_INSTANCE_STATUSES", "FormInstance"]
__all__ = [
"FORM_HANDOFF_STATES",
"FORM_INSTANCE_STATUSES",
"FormHandoff",
"FormHandoffState",
"FormInstance",
]
src/govoplan_forms_runtime/backend/handoffs.py
+782
View File
@@ -0,0 +1,782 @@
from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime
import hashlib
import uuid
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_SERVICE_DEFINITIONS,
InstitutionalContextError,
InstitutionalReference,
ServiceBinding,
ServiceDefinition,
ServiceDefinitionProvider,
ServiceLaunchRequest,
ServiceLaunchResult,
ServiceLauncher,
service_launch_capability,
)
from govoplan_forms_runtime.backend.db.models import (
FormHandoffEffect,
FormInstanceEvent,
)
from govoplan_forms_runtime.backend.domain import FormHandoff, FormHandoffState
from govoplan_forms_runtime.backend.service import (
FormRuntimeError,
FormRuntimeService,
_aware,
_capability,
_current_instance,
_principal_actor,
_principal_tenant,
_request_hash,
_text,
)
_HANDOFF_NAMESPACE = uuid.uuid5(
uuid.NAMESPACE_URL,
"https://govoplan.add-ideas.de/contracts/forms-runtime/handoff/v1",
)
_NATIVE_HANDOFF_KINDS = frozenset({"case", "workflow"})
_TERMINAL_STATES = frozenset({"accepted", "reconciled", "compensated"})
class FormHandoffService:
def __init__(self, registry: object | None) -> None:
self._registry = registry
self._runtime = FormRuntimeService(registry)
def prepare(
self,
session: Session,
principal: object,
*,
instance_id: str,
expected_revision: int,
binding_kind: str,
binding_reference: str | None,
idempotency_key: str,
requested_at: datetime,
allow_all: bool,
) -> FormHandoff:
tenant_id = _principal_tenant(principal)
actor_id = _principal_actor(principal)
clean_kind = _text(binding_kind, "Form handoff kind", 30)
if clean_kind not in _NATIVE_HANDOFF_KINDS:
raise FormRuntimeError(
"Native Form handoffs currently support Case and Workflow targets."
)
clean_key = _text(idempotency_key, "Form handoff idempotency key", 255)
if requested_at.tzinfo is None or requested_at.utcoffset() is None:
raise FormRuntimeError("Form handoff requested_at must include a timezone.")
replay = (
session.query(FormHandoffEffect)
.filter(
FormHandoffEffect.tenant_id == tenant_id,
FormHandoffEffect.idempotency_key == clean_key,
)
.one_or_none()
)
if replay is not None:
if (
replay.instance_id != instance_id
or replay.instance_revision != expected_revision
or replay.binding_kind != clean_kind
or _aware(replay.requested_at) != requested_at
or (
binding_reference is not None
and replay.binding_reference != binding_reference
)
or str(replay.details.get("requested_by") or "") != actor_id
):
raise FormRuntimeError(
"Form handoff idempotency conflict: this key was used for another request."
)
return _handoff_from_row(replay)
current, _ = _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."
)
if current.status not in {"submitted", "validated", "needs_review", "accepted"}:
raise FormRuntimeError(
f"Form status {current.status!r} does not permit a native handoff."
)
definition = self._runtime._definition(
session,
principal,
reference=current.definition_ref,
effective_at=requested_at,
)
if clean_kind not in definition.handoff_kinds:
raise FormRuntimeError(
f"Form definition does not permit a {clean_kind!r} handoff."
)
service, binding, provider_capability = self._resolve_target(
session,
principal,
current=current,
binding_kind=clean_kind,
binding_reference=binding_reference,
effective_at=requested_at,
)
self._runtime._evaluate_policy(
session,
principal,
definition=definition,
action=f"handoff:{clean_kind}:request",
instance=current,
)
self._launcher(provider_capability)
request = {
"instance_id": current.instance_id,
"instance_revision": current.revision,
"definition_ref": current.definition_ref.to_dict(),
"service_ref": service.reference.to_dict(),
"binding": binding.to_dict(),
"actor_id": actor_id,
"requested_at": requested_at.isoformat(),
}
request_sha256 = _request_hash(request)
effect_id = str(uuid.uuid4())
provider_key = str(
uuid.uuid5(
_HANDOFF_NAMESPACE,
":".join((tenant_id, current.instance_id, clean_key)),
)
)
row = FormHandoffEffect(
tenant_id=tenant_id,
instance_id=current.instance_id,
effect_id=effect_id,
instance_revision=current.revision,
idempotency_key=clean_key,
provider_key=provider_key,
request_sha256=request_sha256,
binding_kind=clean_kind,
binding_reference=binding.reference,
provider_capability=provider_capability,
state="requested",
attempt_count=0,
requested_at=requested_at,
evidence=[],
details={
"requested_by": actor_id,
"definition_ref": current.definition_ref.to_dict(),
"service_ref": service.reference.to_dict(),
"binding": binding.to_dict(),
},
)
session.add(row)
session.flush()
_record_transition(
session,
principal,
row=row,
instance_status=current.status,
state="requested",
occurred_at=requested_at,
)
return _handoff_from_row(row)
def execute(
self,
session: Session,
principal: object,
*,
effect_id: str,
executed_at: datetime,
allow_all: bool,
reconcile: bool = False,
) -> tuple[FormHandoff, object | None]:
row = self._effect_row(
session,
principal,
effect_id=effect_id,
lock=True,
allow_all=allow_all,
)
if row.state in _TERMINAL_STATES:
return _handoff_from_row(row), None
if reconcile:
if row.state != "outcome_unknown":
raise FormRuntimeError(
"Only an outcome-unknown Form handoff can be reconciled."
)
elif row.state != "requested":
if row.state == "outcome_unknown":
raise FormRuntimeError(
"Outcome-unknown Form handoffs must be reconciled with the same provider key."
)
raise FormRuntimeError(
f"Form handoff state {row.state!r} cannot be executed."
)
current, _ = _current_instance(
session,
principal,
instance_id=row.instance_id,
lock=True,
allow_all=allow_all,
)
if current.revision != row.instance_revision:
return self._fail_before_effect(
session,
principal,
row=row,
instance_status=current.status,
state="rejected",
occurred_at=executed_at,
message="The Form instance changed after this handoff was requested.",
), None
definition = self._runtime._definition(
session,
principal,
reference=current.definition_ref,
effective_at=executed_at,
)
try:
self._runtime._evaluate_policy(
session,
principal,
definition=definition,
action=f"handoff:{row.binding_kind}:confirm",
instance=current,
)
service, binding, provider_capability = self._resolve_target(
session,
principal,
current=current,
binding_kind=row.binding_kind,
binding_reference=row.binding_reference,
effective_at=executed_at,
)
if provider_capability != row.provider_capability:
raise FormRuntimeError(
"The configured Form handoff provider changed after request."
)
launcher = self._launcher(row.provider_capability)
except (
FormRuntimeError,
InstitutionalContextError,
PermissionError,
LookupError,
) as exc:
return self._fail_before_effect(
session,
principal,
row=row,
instance_status=current.status,
state="rejected",
occurred_at=executed_at,
message=str(exc),
), None
row.attempt_count += 1
launch_request = ServiceLaunchRequest(
service_ref=service.reference,
binding=binding,
idempotency_key=row.provider_key,
requested_at=_aware(row.requested_at),
parameters=_handoff_parameters(current),
)
try:
with session.begin_nested():
result = launcher.launch_service(
session,
principal,
definition=service,
request=launch_request,
)
_validate_result(
result,
tenant_id=row.tenant_id,
service=service,
binding=binding,
)
except (
InstitutionalContextError,
FormRuntimeError,
PermissionError,
LookupError,
ValueError,
) as exc:
row.state = "rejected"
row.last_error = _bounded_error(exc)
row.resolved_at = executed_at
_record_transition(
session,
principal,
row=row,
instance_status=current.status,
state="rejected",
occurred_at=executed_at,
)
return _handoff_from_row(row), None
except Exception as exc:
# The provider may have accepted the effect before connectivity was
# lost. Retain the exact provider key and require reconciliation.
row.state = "outcome_unknown"
row.last_error = _bounded_error(exc)
row.resolved_at = executed_at
_record_transition(
session,
principal,
row=row,
instance_status=current.status,
state="outcome_unknown",
occurred_at=executed_at,
)
return _handoff_from_row(row), None
target_refs = current.handoff_refs
if result.target_ref is not None:
target_refs = tuple(dict.fromkeys((*target_refs, result.target_ref)))
final_state: FormHandoffState = "reconciled" if reconcile else "accepted"
revised = self._runtime._revise(
session,
principal,
current=current,
identity=_current_instance(
session,
principal,
instance_id=row.instance_id,
lock=True,
allow_all=allow_all,
)[1],
expected_revision=current.revision,
status="handed_off",
values=current.values,
attachment_refs=current.attachment_refs,
signature_refs=current.signature_refs,
handoff_refs=target_refs,
idempotency_key=f"handoff:{row.effect_id}:{final_state}",
recorded_at=executed_at,
change_reason=(
f"Reconciled {row.binding_kind} handoff."
if reconcile
else f"Completed {row.binding_kind} handoff."
),
operation=f"handoff_{final_state}",
final_validation=True,
definition=definition,
allowed_current_statuses=(
"submitted",
"validated",
"needs_review",
"accepted",
),
)
row.state = final_state
row.resolved_at = executed_at
row.target_ref = result.target_ref.to_dict() if result.target_ref else None
row.href = result.href
row.evidence = [
item.to_dict(include_inspection=False) for item in result.evidence
]
row.last_error = None
row.details = {
**dict(row.details),
"provider_result": dict(result.metadata),
"provider_replayed": result.replayed,
}
_record_transition(
session,
principal,
row=row,
instance_status=revised.status,
state=final_state,
occurred_at=executed_at,
)
return _handoff_from_row(row), revised
def retry_rejected(
self,
session: Session,
principal: object,
*,
effect_id: str,
requested_at: datetime,
allow_all: bool,
) -> FormHandoff:
row = self._effect_row(
session,
principal,
effect_id=effect_id,
lock=True,
allow_all=allow_all,
)
if row.state != "rejected":
raise FormRuntimeError(
"Only a rejected Form handoff can be retried. Outcome-unknown effects require reconciliation."
)
current, _ = _current_instance(
session,
principal,
instance_id=row.instance_id,
lock=True,
allow_all=allow_all,
)
if current.revision != row.instance_revision:
raise FormRuntimeError(
"The Form instance changed after this handoff failed; create a new handoff request."
)
row.state = "requested"
row.last_error = None
row.resolved_at = None
row.requested_at = requested_at
_record_transition(
session,
principal,
row=row,
instance_status=current.status,
state="requested",
occurred_at=requested_at,
)
return _handoff_from_row(row)
def compensate(
self,
session: Session,
principal: object,
*,
effect_id: str,
compensated_at: datetime,
confirmed_absent: bool,
change_reason: str,
allow_all: bool,
) -> FormHandoff:
if not confirmed_absent:
raise FormRuntimeError(
"Compensation requires confirmation that no target effect exists."
)
row = self._effect_row(
session,
principal,
effect_id=effect_id,
lock=True,
allow_all=allow_all,
)
if row.state not in {"rejected", "outcome_unknown"}:
raise FormRuntimeError(
"Only rejected or outcome-unknown Form handoffs can be compensated."
)
current, _ = _current_instance(
session,
principal,
instance_id=row.instance_id,
lock=True,
allow_all=allow_all,
)
definition = self._runtime._definition(
session,
principal,
reference=current.definition_ref,
effective_at=compensated_at,
)
self._runtime._evaluate_policy(
session,
principal,
definition=definition,
action=f"handoff:{row.binding_kind}:compensate",
instance=current,
)
row.state = "compensated"
row.resolved_at = compensated_at
row.last_error = None
row.details = {
**dict(row.details),
"compensation": {
"confirmed_absent": True,
"change_reason": _text(change_reason, "Compensation reason", 1000),
"actor_id": _principal_actor(principal),
},
}
_record_transition(
session,
principal,
row=row,
instance_status=current.status,
state="compensated",
occurred_at=compensated_at,
)
return _handoff_from_row(row)
def list(
self,
session: Session,
principal: object,
*,
instance_id: str,
allow_all: bool,
) -> tuple[FormHandoff, ...]:
self._runtime.get_instance(
session,
principal,
instance_id=instance_id,
allow_all=allow_all,
)
rows = (
session.query(FormHandoffEffect)
.filter(
FormHandoffEffect.tenant_id == _principal_tenant(principal),
FormHandoffEffect.instance_id == instance_id,
)
.order_by(FormHandoffEffect.requested_at.asc())
.all()
)
return tuple(_handoff_from_row(row) for row in rows)
def _resolve_target(
self,
session: Session,
principal: object,
*,
current: object,
binding_kind: str,
binding_reference: str | None,
effective_at: datetime,
) -> tuple[ServiceDefinition, ServiceBinding, str]:
service_ref = getattr(current, "service_ref", None)
if not isinstance(service_ref, InstitutionalReference):
raise FormRuntimeError(
"A native Form handoff requires an exact Service launch provenance."
)
provider = _capability(self._registry, CAPABILITY_SERVICE_DEFINITIONS)
if not isinstance(provider, ServiceDefinitionProvider):
raise FormRuntimeError(
"The configured Service definition provider is unavailable or invalid."
)
service = provider.get_service_definition(
session,
principal,
reference=service_ref,
effective_at=effective_at,
)
if service is None or service.reference != service_ref:
raise FormRuntimeError(
"The exact Service definition for this Form handoff is unavailable."
)
candidates = tuple(
item
for item in service.bindings
if item.kind == binding_kind
and (binding_reference is None or item.reference == binding_reference)
)
if len(candidates) != 1:
raise FormRuntimeError(
"Select exactly one matching Service handoff binding."
)
return service, candidates[0], service_launch_capability(binding_kind)
def _launcher(self, capability: str) -> ServiceLauncher:
launcher = _capability(self._registry, capability)
if not isinstance(launcher, ServiceLauncher):
raise FormRuntimeError(
f"The configured handoff provider is unavailable or invalid: {capability}."
)
return launcher
def _effect_row(
self,
session: Session,
principal: object,
*,
effect_id: str,
lock: bool,
allow_all: bool,
) -> FormHandoffEffect:
query = session.query(FormHandoffEffect).filter(
FormHandoffEffect.tenant_id == _principal_tenant(principal),
FormHandoffEffect.effect_id == effect_id,
)
if lock:
query = query.with_for_update()
row = query.one_or_none()
if row is None:
raise LookupError("Form handoff not found.")
instance = self._runtime.get_instance(
session,
principal,
instance_id=row.instance_id,
allow_all=allow_all,
)
if instance is None:
raise LookupError("Form instance not found.")
return row
def _fail_before_effect(
self,
session: Session,
principal: object,
*,
row: FormHandoffEffect,
instance_status: str,
state: FormHandoffState,
occurred_at: datetime,
message: str,
) -> FormHandoff:
row.state = state
row.last_error = message[:2000]
row.resolved_at = occurred_at
_record_transition(
session,
principal,
row=row,
instance_status=instance_status,
state=state,
occurred_at=occurred_at,
)
return _handoff_from_row(row)
def _handoff_parameters(instance: object) -> Mapping[str, object]:
return {
"title": f"Form submission {getattr(instance, 'instance_id')}",
"form_submission_id": getattr(instance, "instance_id"),
"form_submission_revision": getattr(instance, "revision"),
"form_definition_ref": getattr(instance, "definition_ref").to_dict(),
"form_values": dict(getattr(instance, "values")),
"attachment_refs": [
item.to_dict(include_inspection=False)
for item in getattr(instance, "attachment_refs")
],
"signature_refs": [
item.to_dict(include_inspection=False)
for item in getattr(instance, "signature_refs")
],
"actor_id": getattr(instance, "changed_by"),
"institutional_context": dict(getattr(instance, "metadata")),
}
def _validate_result(
result: ServiceLaunchResult,
*,
tenant_id: str,
service: ServiceDefinition,
binding: ServiceBinding,
) -> None:
if result.service_ref != service.reference or result.binding != binding:
raise FormRuntimeError(
"The handoff provider returned evidence for another Service binding."
)
if result.target_ref is not None and result.target_ref.tenant_id != tenant_id:
raise FormRuntimeError("The handoff provider returned a cross-tenant target.")
if any(item.tenant_id != tenant_id for item in result.evidence):
raise FormRuntimeError("The handoff provider returned cross-tenant evidence.")
def _record_transition(
session: Session,
principal: object,
*,
row: FormHandoffEffect,
instance_status: str,
state: FormHandoffState,
occurred_at: datetime,
) -> None:
event_id = str(uuid.uuid4())
event = FormInstanceEvent(
tenant_id=row.tenant_id,
instance_id=row.instance_id,
instance_revision=row.instance_revision,
event_id=event_id,
event_type=f"forms_runtime.handoff.{state}",
status=instance_status,
occurred_at=occurred_at,
actor_id=_principal_actor(principal),
idempotency_key=f"handoff:{row.effect_id}:{state}:{row.attempt_count}",
request_sha256=row.request_sha256,
payload={
"effect_id": row.effect_id,
"binding_kind": row.binding_kind,
"binding_reference": row.binding_reference,
"provider_capability": row.provider_capability,
"provider_key_sha256": hashlib.sha256(
row.provider_key.encode("utf-8")
).hexdigest(),
"state": state,
"attempt_count": row.attempt_count,
"target_ref": row.target_ref,
"href": row.href,
"last_error": row.last_error,
},
)
session.add(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=occurred_at,
actor=EventActorRef(type="account", id=_principal_actor(principal)),
tenant=EventTenantRef(id=row.tenant_id),
resource=EventObjectRef(
type="form_submission",
id=row.instance_id,
),
classification="confidential",
),
)
def _handoff_from_row(row: FormHandoffEffect) -> FormHandoff:
return FormHandoff(
tenant_id=row.tenant_id,
instance_id=row.instance_id,
effect_id=row.effect_id,
instance_revision=row.instance_revision,
idempotency_key=row.idempotency_key,
provider_key=row.provider_key,
request_sha256=row.request_sha256,
binding_kind=row.binding_kind,
binding_reference=row.binding_reference,
provider_capability=row.provider_capability,
state=row.state, # type: ignore[arg-type]
attempt_count=row.attempt_count,
requested_at=_aware(row.requested_at),
resolved_at=_aware(row.resolved_at) if row.resolved_at else None,
target_ref=(
InstitutionalReference.from_mapping(row.target_ref)
if isinstance(row.target_ref, Mapping)
else None
),
href=row.href,
evidence=tuple(_evidence_from_mapping(item) for item in (row.evidence or [])),
last_error=row.last_error,
metadata=dict(row.details),
)
def _evidence_from_mapping(value: Mapping[str, object]):
from govoplan_core.core.institutional import EvidenceReference
return EvidenceReference.from_mapping(value)
def _bounded_error(exc: Exception) -> str:
message = str(exc).strip() or type(exc).__name__
return message[:2000]
__all__ = ["FormHandoffService"]
src/govoplan_forms_runtime/backend/manifest.py
+32 -4
View File
@@ -6,7 +6,10 @@ from govoplan_core.core.access import (
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
)
from govoplan_core.core.institutional import CAPABILITY_FORM_DEFINITIONS
from govoplan_core.core.institutional import (
CAPABILITY_FORM_DEFINITIONS,
CAPABILITY_SERVICE_DEFINITIONS,
)
from govoplan_core.core.module_guards import (
drop_table_retirement_provider,
persistent_table_uninstall_guard,
@@ -157,7 +160,12 @@ manifest = ModuleManifest(
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
CAPABILITY_FORM_DEFINITIONS,
),
optional_capabilities=(CAPABILITY_FORMS_RUNTIME_POLICY_EVALUATOR,),
optional_capabilities=(
CAPABILITY_FORMS_RUNTIME_POLICY_EVALUATOR,
CAPABILITY_SERVICE_DEFINITIONS,
"cases.service_launcher",
"workflow_engine.service_launcher",
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
route_factory=_router,
@@ -230,6 +238,24 @@ manifest = ModuleManifest(
version_min="0.1.0",
version_max_exclusive="0.2.0",
),
ModuleInterfaceRequirement(
name="services.definitions",
version_min="0.1.0",
version_max_exclusive="0.2.0",
optional=True,
),
ModuleInterfaceRequirement(
name="cases.service_launcher",
version_min="0.1.0",
version_max_exclusive="0.2.0",
optional=True,
),
ModuleInterfaceRequirement(
name="workflow_engine.service_launcher",
version_min="0.1.0",
version_max_exclusive="0.2.0",
optional=True,
),
),
capability_factories={
CAPABILITY_FORMS_RUNTIME_REGISTRY: _registry,
@@ -254,6 +280,7 @@ manifest = ModuleManifest(
migration_after=("forms",),
retirement_supported=True,
retirement_provider=drop_table_retirement_provider(
runtime_models.FormHandoffEffect,
runtime_models.FormInstanceEvent,
runtime_models.FormInstanceRevision,
runtime_models.FormInstanceIdentity,
@@ -263,6 +290,7 @@ manifest = ModuleManifest(
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
runtime_models.FormHandoffEffect,
runtime_models.FormInstanceIdentity,
runtime_models.FormInstanceRevision,
runtime_models.FormInstanceEvent,
@@ -277,7 +305,7 @@ manifest = ModuleManifest(
summary="Save permitted drafts, submit validated values, and retain exact definition and handoff evidence.",
body=(
"Every instance resolves one immutable published Form revision. Draft and final values are validated on the server; final submission also enforces attachment, signature, and policy requirements. "
"Service launches retain the exact Service and binding. History, receipts, and handoffs are append-only, replay-safe, and optimistic-concurrency guarded."
"Service launches retain the exact Service and binding. Native Case and Workflow handoffs persist intent before execution, use owner capabilities with stable provider keys, and require reconciliation after unknown outcomes. History, receipts, and handoffs are replay-safe and optimistic-concurrency guarded."
),
layer="configured",
documentation_types=("admin", "user"),
@@ -298,7 +326,7 @@ manifest = ModuleManifest(
documentation_ref="docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md",
test_ref="tests/test_forms_runtime.py",
known_limits=(
"Anonymous public intake, concrete Files/signature adapters, and automatic target handoff execution remain adapter depth; authenticated Portal entry is supported.",
"Anonymous public intake, concrete Files/signature adapters, and target kinds beyond the native Case/Workflow handoffs remain adapter depth; authenticated Portal entry is supported.",
),
supported_authority_modes=("native_authoritative",),
owned_concepts=(
src/govoplan_forms_runtime/backend/migrations/versions/a3d5f7b9c1e2_governed_form_handoffs.py
+77
View File
@@ -0,0 +1,77 @@
"""Persist governed Forms Runtime handoff effects.
Revision ID: a3d5f7b9c1e2
Revises: f2a3b4c5d6e7
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "a3d5f7b9c1e2"
down_revision = "f2a3b4c5d6e7"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"form_handoff_effects",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("instance_id", sa.String(length=255), nullable=False),
sa.Column("effect_id", sa.String(length=36), nullable=False),
sa.Column("instance_revision", sa.Integer(), nullable=False),
sa.Column("idempotency_key", sa.String(length=255), nullable=False),
sa.Column("provider_key", sa.String(length=255), nullable=False),
sa.Column("request_sha256", sa.String(length=64), nullable=False),
sa.Column("binding_kind", sa.String(length=30), nullable=False),
sa.Column("binding_reference", sa.String(length=500), nullable=False),
sa.Column("provider_capability", sa.String(length=120), nullable=False),
sa.Column("state", sa.String(length=30), nullable=False),
sa.Column("attempt_count", sa.Integer(), nullable=False),
sa.Column("requested_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("target_ref", sa.JSON(), nullable=True),
sa.Column("href", sa.String(length=2000), nullable=True),
sa.Column("evidence", sa.JSON(), nullable=False),
sa.Column("last_error", sa.String(length=2000), nullable=True),
sa.Column("metadata", sa.JSON(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint("id", name=op.f("pk_form_handoff_effects")),
sa.UniqueConstraint("tenant_id", "effect_id", name="uq_form_handoff_effect"),
sa.UniqueConstraint(
"tenant_id", "idempotency_key", name="uq_form_handoff_idempotency"
),
sa.UniqueConstraint(
"tenant_id", "provider_key", name="uq_form_handoff_provider_key"
),
)
for column in (
"tenant_id",
"instance_id",
"effect_id",
"binding_kind",
"state",
"requested_at",
"resolved_at",
):
op.create_index(
op.f(f"ix_form_handoff_effects_{column}"),
"form_handoff_effects",
[column],
unique=False,
)
op.create_index(
"ix_form_handoff_instance_state",
"form_handoff_effects",
["tenant_id", "instance_id", "state"],
unique=False,
)
def downgrade() -> None:
op.drop_table("form_handoff_effects")
src/govoplan_forms_runtime/backend/router.py
+213 -7
View File
@@ -19,19 +19,24 @@ from govoplan_forms_runtime.backend.manifest import (
from govoplan_forms_runtime.backend.schemas import (
FormDraftUpdateRequest,
FormHandoffRequest,
FormHandoffActionRequest,
FormHandoffCompensateRequest,
FormInstanceCreateRequest,
FormInstanceEventsResponse,
FormInstanceHistoryResponse,
FormInstanceListResponse,
FormSubmitRequest,
FormTransitionRequest,
FormNativeHandoffRequest,
)
from govoplan_forms_runtime.backend.handoffs import FormHandoffService
from govoplan_forms_runtime.backend.service import FormRuntimeError, FormRuntimeService
def create_router(registry: object | None) -> APIRouter:
router = APIRouter(prefix="/forms-runtime", tags=["forms-runtime"])
runtime = FormRuntimeService(registry)
handoffs = FormHandoffService(registry)
@router.get("/instances", response_model=FormInstanceListResponse)
def api_list_instances(
@@ -162,7 +167,12 @@ def create_router(registry: object | None) -> APIRouter:
allow_all=has_scope(principal, WRITE_SCOPE),
)
session.commit()
except (FormRuntimeError, InstitutionalContextError, LookupError, PermissionError) as exc:
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return item.to_dict()
@@ -192,7 +202,12 @@ def create_router(registry: object | None) -> APIRouter:
allow_all=has_scope(principal, WRITE_SCOPE),
)
session.commit()
except (FormRuntimeError, InstitutionalContextError, LookupError, PermissionError) as exc:
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return item.to_dict()
@@ -221,7 +236,12 @@ def create_router(registry: object | None) -> APIRouter:
allow_all=True,
)
session.commit()
except (FormRuntimeError, InstitutionalContextError, LookupError, PermissionError) as exc:
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return item.to_dict()
@@ -250,11 +270,199 @@ def create_router(registry: object | None) -> APIRouter:
allow_all=True,
)
session.commit()
except (FormRuntimeError, InstitutionalContextError, LookupError, PermissionError) as exc:
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return item.to_dict()
@router.get(
"/instances/{instance_id}/handoffs",
response_model=dict[str, object],
)
def api_list_handoffs(
instance_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require_any(principal, PARTICIPATE_SCOPE, READ_SCOPE)
try:
items = handoffs.list(
session,
principal,
instance_id=instance_id,
allow_all=has_scope(principal, READ_SCOPE),
)
except (FormRuntimeError, LookupError, PermissionError) as exc:
raise _error(exc) from exc
return {"handoffs": [item.to_dict() for item in items]}
@router.post(
"/instances/{instance_id}/handoffs/native",
response_model=dict[str, object],
)
def api_native_handoff(
instance_id: str,
payload: FormNativeHandoffRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require_any(principal, WRITE_SCOPE, ADMIN_SCOPE)
try:
prepared = handoffs.prepare(
session,
principal,
instance_id=instance_id,
expected_revision=payload.expected_revision,
binding_kind=payload.binding_kind,
binding_reference=payload.binding_reference,
idempotency_key=payload.idempotency_key,
requested_at=payload.requested_at,
allow_all=True,
)
# The intent is durable before owner capability execution starts.
session.commit()
item, instance = handoffs.execute(
session,
principal,
effect_id=prepared.effect_id,
executed_at=payload.requested_at,
allow_all=True,
)
session.commit()
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return {
"handoff": item.to_dict(),
"instance": instance.to_dict() if instance is not None else None,
}
@router.post(
"/instances/{instance_id}/handoffs/{effect_id}/retry",
response_model=dict[str, object],
)
def api_retry_handoff(
instance_id: str,
effect_id: str,
payload: FormHandoffActionRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require_any(principal, WRITE_SCOPE, ADMIN_SCOPE)
try:
prepared = handoffs.retry_rejected(
session,
principal,
effect_id=effect_id,
requested_at=payload.recorded_at,
allow_all=True,
)
if prepared.instance_id != instance_id:
raise LookupError("Form handoff does not belong to this instance.")
session.commit()
item, instance = handoffs.execute(
session,
principal,
effect_id=effect_id,
executed_at=payload.recorded_at,
allow_all=True,
)
session.commit()
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return {
"handoff": item.to_dict(),
"instance": instance.to_dict() if instance is not None else None,
}
@router.post(
"/instances/{instance_id}/handoffs/{effect_id}/reconcile",
response_model=dict[str, object],
)
def api_reconcile_handoff(
instance_id: str,
effect_id: str,
payload: FormHandoffActionRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require_any(principal, WRITE_SCOPE, ADMIN_SCOPE)
try:
item, instance = handoffs.execute(
session,
principal,
effect_id=effect_id,
executed_at=payload.recorded_at,
allow_all=True,
reconcile=True,
)
if item.instance_id != instance_id:
raise LookupError("Form handoff does not belong to this instance.")
session.commit()
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return {
"handoff": item.to_dict(),
"instance": instance.to_dict() if instance is not None else None,
}
@router.post(
"/instances/{instance_id}/handoffs/{effect_id}/compensate",
response_model=dict[str, object],
)
def api_compensate_handoff(
instance_id: str,
effect_id: str,
payload: FormHandoffCompensateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require_any(principal, ADMIN_SCOPE)
try:
item = handoffs.compensate(
session,
principal,
effect_id=effect_id,
compensated_at=payload.recorded_at,
confirmed_absent=payload.confirmed_absent,
change_reason=payload.change_reason,
allow_all=True,
)
if item.instance_id != instance_id:
raise LookupError("Form handoff does not belong to this instance.")
session.commit()
except (
FormRuntimeError,
InstitutionalContextError,
LookupError,
PermissionError,
) as exc:
session.rollback()
raise _error(exc) from exc
return {"handoff": item.to_dict()}
@router.get(
"/instances/{instance_id}/history",
response_model=FormInstanceHistoryResponse,
@@ -278,9 +486,7 @@ def create_router(registry: object | None) -> APIRouter:
raise _error(exc) from exc
if not items:
raise HTTPException(status_code=404, detail="Form instance not found")
return FormInstanceHistoryResponse(
revisions=[item.to_dict() for item in items]
)
return FormInstanceHistoryResponse(revisions=[item.to_dict() for item in items])
@router.get(
"/instances/{instance_id}/events",
src/govoplan_forms_runtime/backend/schemas.py
+24
View File
@@ -68,6 +68,27 @@ class FormHandoffRequest(BaseModel):
change_reason: str = Field(min_length=1, max_length=1000)
class FormNativeHandoffRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
expected_revision: int = Field(ge=1)
binding_kind: Literal["case", "workflow"]
binding_reference: str | None = Field(default=None, min_length=1, max_length=500)
idempotency_key: str = Field(min_length=1, max_length=255)
requested_at: datetime
class FormHandoffActionRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
recorded_at: datetime
class FormHandoffCompensateRequest(FormHandoffActionRequest):
confirmed_absent: bool
change_reason: str = Field(min_length=1, max_length=1000)
class FormInstanceListResponse(BaseModel):
instances: list[dict[str, Any]]
total: int
@@ -86,6 +107,9 @@ class FormInstanceEventsResponse(BaseModel):
__all__ = [
"FormDraftUpdateRequest",
"FormHandoffRequest",
"FormHandoffActionRequest",
"FormHandoffCompensateRequest",
"FormNativeHandoffRequest",
"FormInstanceCreateRequest",
"FormInstanceEventsResponse",
"FormInstanceHistoryResponse",
src/govoplan_forms_runtime/backend/service.py
+214 -29
View File
@@ -21,6 +21,7 @@ from govoplan_core.core.events import (
from govoplan_core.core.institutional import (
CAPABILITY_FORM_DEFINITIONS,
EvidenceReference,
FormConditionExpression,
FormDefinition,
FormDefinitionProvider,
FormFieldDefinition,
@@ -107,14 +108,17 @@ class FormRuntimeService:
effective_at=recorded_at,
)
supplied_values = _mapping_copy(values, "Form values")
clean_values = {
**{
field.key: field.default_value
for field in definition.fields
if field.default_value is not None
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,
},
**supplied_values,
}
)
attachments = tuple(attachment_refs)
signatures = tuple(signature_refs)
diagnostics = validate_form_values(
@@ -402,6 +406,10 @@ class FormRuntimeService:
)
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."
@@ -413,9 +421,7 @@ class FormRuntimeService:
action="handoff",
instance=current,
)
handoffs = tuple(
dict.fromkeys((*current.handoff_refs, target_ref))
)
handoffs = tuple(dict.fromkeys((*current.handoff_refs, target_ref)))
return self._revise(
session,
principal,
@@ -528,7 +534,9 @@ class FormRuntimeService:
FormInstanceIdentity.created_by == _principal_actor(principal)
)
if statuses:
statement = statement.filter(FormInstanceRevision.status.in_(tuple(statuses)))
statement = statement.filter(
FormInstanceRevision.status.in_(tuple(statuses))
)
if definition_id:
statement = statement.filter(
FormInstanceIdentity.definition_id == definition_id
@@ -635,7 +643,10 @@ class FormRuntimeService:
) -> FormInstance:
_require_aware(recorded_at, "Form instance recorded_at")
clean_reason = _text(change_reason, "Form instance change reason", 1000)
clean_values = _mapping_copy(values, "Form values")
clean_values = normalize_form_values(
definition,
_mapping_copy(values, "Form values"),
)
attachments = tuple(attachment_refs)
signatures = tuple(signature_refs)
handoffs = tuple(handoff_refs)
@@ -753,7 +764,9 @@ class FormRuntimeService:
effective_at=effective_at,
)
if definition is None:
raise FormRuntimeError("The exact Form definition was not found or effective.")
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."
@@ -818,9 +831,7 @@ class FormsServiceLauncher:
raise FormRuntimeError(
"Form Service launch requires the exact published Service and binding."
)
form_id, form_revision = parse_form_binding_reference(
request.binding.reference
)
form_id, form_revision = parse_form_binding_reference(request.binding.reference)
definition_ref = InstitutionalReference(
kind="form",
owner_module="forms",
@@ -883,9 +894,17 @@ def validate_form_values(
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.")
_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(
@@ -923,6 +942,117 @@ def validate_form_values(
return tuple(diagnostics)
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,
@@ -956,17 +1086,30 @@ def _validate_field_value(
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.")
_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.")
_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]
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.")
_diagnostic(
field.key,
"error",
"field.option",
"Every selected value must be a declared option.",
)
)
constraints = field.constraints
if isinstance(value, str):
@@ -974,9 +1117,23 @@ def _validate_field_value(
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."))
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."))
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
@@ -985,14 +1142,35 @@ def _validate_field_value(
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."))
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}."))
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}."))
diagnostics.append(
_diagnostic(
field.key,
"error",
"field.maximum",
f"Enter a value no greater than {maximum}.",
)
)
return tuple(diagnostics)
@@ -1332,8 +1510,10 @@ def _required_mapping(value: Mapping[str, object], key: str) -> Mapping[str, obj
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
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
@@ -1342,7 +1522,9 @@ def _identifier(value: str, label: str) -> str:
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.")
raise FormRuntimeError(
f"{label} is required and limited to {maximum} characters."
)
return clean
@@ -1381,6 +1563,9 @@ __all__ = [
"FormRuntimePolicyEvaluator",
"FormRuntimeService",
"FormsServiceLauncher",
"evaluate_form_condition",
"normalize_form_values",
"parse_form_binding_reference",
"validate_form_values",
"visible_form_field_keys",
]