symfonic.memory.backends.procedure_review_transition

procedure_review_transition

Data integrity for a dedicated review CAS; caller authorization is external.

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_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")
    current = node.properties
    if (procedure_lifecycle(current)[0] != "draft"
        or is_pending(current) or is_retracted(current)):
        return False
    return snapshot(current) == snapshot(expected)

owns

owns(scope: Any, node: Any) -> bool

Review writes require exact effective ownership, not ancestor visibility.

Native scopes never infer an owner. Legacy absent/NULL property owners resolve to the tenant root, matching the established dual-read contract.

Source code in src/symfonic/memory/backends/procedure_review_transition.py

def owns(scope: Any, node: Any) -> bool:
    """Review writes require exact effective ownership, not ancestor visibility.

    Native scopes never infer an owner. Legacy absent/NULL property owners
    resolve to the tenant root, matching the established dual-read contract.
    """
    return stored_scope_path(node.properties, node.tenant_id, scope=scope) == (
        materialise_scope_path(scope)
    )

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_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)
    status = following.get("status")
    if not isinstance(status, str) or status not in {"approved", "rejected"}:
        raise ValueError("procedure review must approve or reject a draft")
    if following.get("active") is not (status == "approved"):
        raise ValueError("procedure review active flag must match its status")
    authority = following.get("procedure_review_authority")
    if not isinstance(authority, str) or authority not in {"human", "policy"}:
        raise ValueError("procedure review requires a recognized authority")
    actor = following.get("procedure_review_actor")
    if not isinstance(actor, str) or not actor.strip():
        raise ValueError("procedure review requires a nonblank actor")
    return preserved(expected, following, status)

preserved

preserved(expected: dict[str, Any], following: dict[str, Any], status: str) -> dict[str, Any]

Shared artifact, binding and revision rules for every review rule.

Source code in src/symfonic/memory/backends/procedure_review_transition.py

def preserved(expected: dict[str, Any], following: dict[str, Any], status: str) -> dict[str, Any]:
    """Shared artifact, binding and revision rules for every review rule."""
    next_revision = revision(expected) + 1
    excluded = _REVIEW_FIELDS | _BINDINGS
    original_artifact = {key: value for key, value in expected.items() if key not in excluded}
    reviewed_artifact = {key: value for key, value in following.items() if key not in excluded}
    if snapshot(original_artifact) != snapshot(reviewed_artifact):
        raise ValueError("procedure review cannot amend or erase artifact properties")
    original_bindings = {key: value for key, value in expected.items() if key in _BINDINGS}
    reviewed_bindings = {key: value for key, value in following.items() if key in _BINDINGS}
    if status == "rejected" and snapshot(original_bindings) != snapshot(reviewed_bindings):
        raise ValueError("only an approved procedure may change bindings")
    for key in _BINDINGS:
        if key in expected and key not in following:
            raise ValueError("procedure review cannot remove an existing binding")
        if key in following and following[key] is None and (
            key not in expected or expected[key] is not None
        ):
            raise ValueError("procedure review binding changes require non-None values")
    if REVISION_KEY in following and (
        type(following[REVISION_KEY]) is not int or following[REVISION_KEY] != next_revision
    ):
        raise ValueError("procedure review revision must advance exactly once")
    snapshot(expected)
    result = json.loads(snapshot(following))
    result[REVISION_KEY] = next_revision
    return result

snapshot

snapshot(properties: dict[str, Any]) -> str

JSON equality preserves absent/null and boolean/number distinctions.

Source code in src/symfonic/memory/backends/procedure_review_transition.py

def snapshot(properties: dict[str, Any]) -> str:
    """JSON equality preserves absent/null and boolean/number distinctions."""
    try:
        return json.dumps(properties, sort_keys=True, allow_nan=False)
    except (TypeError, ValueError) as exc:
        raise ValueError("procedure review properties must be JSON objects") from exc