Skip to content

symfonic.platform.procedure_review

procedure_review

An explicit, audited policy seam for learned-procedure review.

The native default is deliberately not a policy: phase 12 writes a draft and the public review service leaves it there until a person calls MemoryAdminService.approve_procedure or reject_procedure. A deployment that wants automation composes this service itself and supplies both a policy and a durable audit sink. This keeps automatic review out of HTTP handlers and out of the legacy engine configuration surface.

ProcedureApprovalDecision dataclass 

ProcedureApprovalDecision(disposition: ProcedureDisposition, actor: str, reason: str, action_tool: str | None = None, precondition: Any = None)

One policy result, including the accountable actor and explanation.

ProcedureApprovalPolicy

Bases: Protocol

Deployment policy deciding one draft without mutating it directly.

decide async 

decide(candidate: ProcedureCandidate) -> ProcedureApprovalDecision

Return an accountable approve, reject, or defer decision.

Source code in src/symfonic/platform/procedure_review.py 
67
68
async def decide(self, candidate: ProcedureCandidate) -> ProcedureApprovalDecision:
    """Return an accountable approve, reject, or defer decision."""

ProcedureCandidate dataclass 

ProcedureCandidate(scope_path: str, procedure_id: str, source: object | None, source_present: bool, action_tool: str | None, action_tool_present: bool)

The bounded, non-content context available to a review policy.

Policies see stable provenance and the proposed tool binding, never the procedure's free-text body or a graph handle. A policy that needs richer evidence must fetch it through an authorised review workflow rather than receive tenant memory implicitly.

ProcedureReviewService 

ProcedureReviewService(admin: MemoryAdminService, policy: ProcedureApprovalPolicy, *, sink: AuditSink | None)

Apply an opt-in policy through the public memory-admin review door.

sink is mandatory because an automated decision that cannot be attributed must not change a procedure's governing status. The service records intent before mutation. That intent record is the fail-closed boundary: if it cannot land, no mutation starts. Completion and failure records are best-effort observations after that boundary; a store cannot safely roll back a completed review merely because its later receipt was unavailable. Backend exception text is never serialised because it may contain stored content.

Source code in src/symfonic/platform/procedure_review.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def __init__(
    self,
    admin: MemoryAdminService,
    policy: ProcedureApprovalPolicy,
    *,
    sink: AuditSink | None,
) -> None:
    if sink is None or isinstance(sink, NullAuditSink):
        raise AuditSinkError(
            "automatic procedure review requires an explicitly configured audit sink"
        )
    self._admin = admin
    self._policy = policy
    self._audit = AuditSeam(sink)

review async 

review(scope: MemoryScope, procedure: Any) -> Any | None

Audit and apply a policy decision to one draft.

defer is a terminal result for this call, not a hidden retry: the draft remains queued for a person or an explicitly scheduled policy.

Source code in src/symfonic/platform/procedure_review.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def review(self, scope: MemoryScope, procedure: Any) -> Any | None:
    """Audit and apply a policy decision to one draft.

    ``defer`` is a terminal result for this call, not a hidden retry: the
    draft remains queued for a person or an explicitly scheduled policy.
    """
    candidate = _candidate(scope, procedure)
    decision = await self._policy.decide(candidate)
    _validate(decision)
    await self._audit.record(_record(candidate, decision, "initiated"), destructive=True)

    if decision.disposition == "defer":
        await self._audit.record(_record(candidate, decision, "deferred"))
        return None
    try:
        result = await self._admin.decide_draft_procedure(
            scope,
            procedure,
            status="approved" if decision.disposition == "approve" else "rejected",
            action_tool=decision.action_tool,
            precondition=decision.precondition,
            expected_source=candidate.source,
            expected_source_present=candidate.source_present,
            expected_action_tool=candidate.action_tool,
            expected_action_tool_present=candidate.action_tool_present,
            actor=decision.actor,
        )
    except Exception:
        await self._audit.record(_record(candidate, decision, "failed"))
        raise
    if result is None:
        await self._audit.record(_record(candidate, decision, "superseded"))
        return None
    await self._audit.record(_record(candidate, decision, "completed"))
    return result