Data integrity for an explicit human re-review CAS; authorization is external.
A human may review a draft or replace a policy decision. The result carries
human authority and no actor: no verified human identity exists at this
seam, so retaining the policy actor would misattribute the decision and
minting one would invent it. A human-reviewed row is not eligible here.
eligible
eligible(properties: dict[str, Any]) -> bool
A published draft, or a terminal decision sealed by an attributed policy.
Source code in src/symfonic/memory/backends/procedure_human_review_transition.py
| def eligible(properties: dict[str, Any]) -> bool:
"""A published draft, or a terminal decision sealed by an attributed policy."""
if is_pending(properties) or is_retracted(properties):
return False
# The read projection folds a malformed status into draft; a seal must not.
# Only an absent status is a legacy draft.
status = _declared(properties, "status")
active = _declared(properties, "active")
if status is _ABSENT or status == "draft":
# Only a review mints these fields; a draft carrying either is malformed.
return (
(active is _ABSENT or active is False)
and _AUTHORITY not in properties and _ACTOR not in properties
)
actor = properties.get(_ACTOR)
return (
(status == "approved" and isinstance(active, bool)
or status == "rejected" and active is False)
and properties.get(_AUTHORITY) == "policy"
and isinstance(actor, str) and bool(actor.strip())
)
|
matches
matches(node: Any, expected: dict[str, Any]) -> bool
Evaluate the committed row only while holding its backend write lock.
Source code in src/symfonic/memory/backends/procedure_human_review_transition.py
| def matches(node: Any, expected: dict[str, Any]) -> bool:
"""Evaluate the committed row only while holding its backend write lock."""
if node.layer != MemoryLayer.PROCEDURAL:
raise ValueError("procedure review requires a procedural node")
# prepare() proved ``expected`` eligible; exact equality carries that over.
return snapshot(node.properties) == snapshot(expected)
|
prepare
prepare(expected: dict[str, Any], following: dict[str, Any]) -> dict[str, Any]
Validate and detach the proposal before waiting for a backend lock.
Source code in src/symfonic/memory/backends/procedure_human_review_transition.py
| def prepare(expected: dict[str, Any], following: dict[str, Any]) -> dict[str, Any]:
"""Validate and detach the proposal before waiting for a backend lock."""
if not isinstance(expected, dict) or not isinstance(following, dict):
raise ValueError("procedure review properties must be JSON objects")
revision(expected)
snapshot(expected)
if not eligible(expected):
raise ValueError("human re-review requires a draft or policy-reviewed preimage")
status = following.get("status")
if not isinstance(status, str) or status not in {"approved", "rejected"}:
raise ValueError("human review must approve or reject")
if following.get("active") is not (status == "approved"):
raise ValueError("procedure review active flag must match its status")
if following.get(_AUTHORITY) != "human":
raise ValueError("human review requires human authority")
if _ACTOR in following:
raise ValueError("human review has no verified actor to record")
return preserved(expected, following, status)
|