Approvals

zeroth.governance.approvals

Governed approval workflows.

This package re-exports the key models, the database repository, and the high-level service so callers can simply from zeroth.governance.approvals import ....

ApprovalDecision

Bases: StrEnum

The possible choices a human reviewer can make on an approval request.

APPROVE — accept as-is. REJECT — deny the action. EDIT_AND_APPROVE — modify the proposed payload and then accept it.

ApprovalRecord

Bases: BaseModel

The main approval object that tracks a single approval request.

Created when an agent workflow hits a node that requires human sign-off. Contains everything a reviewer needs: a summary, rationale, the proposed payload, which actions are allowed, and (once answered) the resolution.

ApprovalResolution

Bases: BaseModel

Holds the details of how an approval was resolved.

Stores who made the decision, what they decided, any edits they made to the payload, and when the decision happened.

ApprovalStatus

Bases: StrEnum

Whether an approval request is still waiting or has been answered.

PENDING means no decision yet; RESOLVED means a human has responded.

HumanInteractionType

Bases: StrEnum

The kind of interaction the system is requesting from a human.

For example, "approval" means the system needs a yes/no decision, while "clarification" means it needs more information to proceed.

ApprovalRepository

ApprovalRepository(database: AsyncDatabase)

Saves and loads approval records from an async database.

Use this when you need to persist approval requests so they survive restarts, or when you need to look up pending approvals by run, thread, or deployment.

scoped_for_deployment classmethod

scoped_for_deployment(
    database: AsyncDatabase,
    scope_context: ScopeContext | NullWorkspaceScopeContext,
    deployment_ref: str,
) -> ApprovalRepository

Bind scheduled SLA reads to one deployment owner.

write async

write(record: ApprovalRecord) -> ApprovalRecord

Save an approval record to the database.

If a record with the same approval_id already exists, it will be updated. Returns the freshly-read record from the database.

get async

get(
    approval_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    deployment_ref: str | None = None,
    graph_version_ref: str | None = None,
) -> ApprovalRecord | None

Look up one approval with every supplied scope predicate in SQL.

resolve_pending async

resolve_pending(
    record: ApprovalRecord,
    *,
    require_sla_deadline: bool = False,
) -> ApprovalRecord | None

Atomically publish record only while its exact scoped row is pending.

require_sla_deadline adds sla_deadline IS NOT NULL to the compare-and-set. Callers that CHANGE the stored status are already fenced by the status = PENDING predicate below: the loser of a race sees the winner's new status and matches zero rows. The alert-escalation latch is the exception -- it deliberately leaves the row PENDING and moves only the deadline -- so for it the status predicate is satisfied by every racer and provides no fence at all. Keying on the column the latch actually clears restores the one-winner property for that caller without changing the predicate for anybody else.

list_pending async

list_pending(
    *,
    run_id: str | None = None,
    thread_id: str | None = None,
    deployment_ref: str | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    graph_version_ref: str | None = None,
) -> list[ApprovalRecord]

Return all approval records that are still waiting for a decision.

You can optionally filter by run_id, thread_id, or deployment_ref. Results are sorted by creation time.

list async

list(
    *,
    run_id: str | None = None,
    thread_id: str | None = None,
    deployment_ref: str | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    graph_version_ref: str | None = None,
) -> list[ApprovalRecord]

Return approval records, optionally filtered by run, thread, or deployment.

list_overdue async

list_overdue() -> list[ApprovalRecord]

Return overdue approvals for this repository's deployment owner.

ApprovalContinuation

Bases: Protocol

The slice of the run orchestrator the approval workflow drives.

Structural, so the runtime orchestrator satisfies it without the approvals domain importing the orchestrator implementation.

record_approval_resolution async

record_approval_resolution(
    *,
    graph: Graph,
    run: Run,
    node: HumanApprovalNode,
    output_payload: Mapping[str, Any],
    approval_record: ApprovalRecord,
) -> Run

Record the result of a human approval decision on the run.

resume_graph async

resume_graph(graph: Graph, run_id: str) -> Run

Resume the paused run from its current node onward.

ApprovalService

ApprovalService(
    *,
    repository: ApprovalRepository,
    run_repository: RunRepository,
    audit_repository: AuditRepository | None = None,
    payload_sanitizer: PayloadSanitizer | None = None,
)

The main entry point for working with approvals.

Handles the full lifecycle: creating a pending approval when a workflow pauses, letting a human resolve it (approve / reject / edit-and-approve), and then resuming the run with the decision applied.

create_pending async

create_pending(
    *,
    run: Run,
    node: HumanApprovalNode,
    input_payload: dict[str, Any],
) -> ApprovalRecord

Create a new approval request and save it to the database.

Called when a workflow reaches a node that needs human sign-off. Builds the approval record from the run and node details, sanitizes sensitive data out of the payload, and persists it as "pending".

get async

get(
    approval_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    deployment_ref: str | None = None,
    graph_version_ref: str | None = None,
) -> ApprovalRecord | None

Fetch a single approval record by its ID. Returns None if not found.

list_pending async

list_pending(
    *,
    run_id: str | None = None,
    thread_id: str | None = None,
    deployment_ref: str | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    graph_version_ref: str | None = None,
) -> list[ApprovalRecord]

Return all approvals that are still waiting for a human decision.

Optionally filter by run_id, thread_id, or deployment_ref.

list async

list(
    *,
    run_id: str | None = None,
    thread_id: str | None = None,
    deployment_ref: str | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    graph_version_ref: str | None = None,
) -> list[ApprovalRecord]

Return approval records for a run, thread, or deployment.

get_visible_to_deployment async

get_visible_to_deployment(
    approval_id: str,
    *,
    deployment_ref: str,
    graph_version_ref: str,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
) -> ApprovalRecord | None

Return an approval owned by this deployment or one of its child runs.

Child records keep their real child deployment/graph provenance. The deployment becomes an authorized view only when the scoped run chain reaches an ancestor with this exact deployment and graph version.

list_pending_visible_to_deployment async

list_pending_visible_to_deployment(
    *,
    deployment_ref: str,
    graph_version_ref: str,
    run_id: str | None = None,
    thread_id: str | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
) -> list[ApprovalRecord]

List pending approvals whose scoped ancestry reaches a deployment.

list_visible_to_deployment async

list_visible_to_deployment(
    *,
    deployment_ref: str,
    graph_version_ref: str,
    run_id: str | None = None,
    thread_id: str | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
) -> list[ApprovalRecord]

List all approval evidence visible through scoped run ancestry.

visible_ancestor_run async

visible_ancestor_run(
    record: ApprovalRecord,
    *,
    deployment_ref: str,
    graph_version_ref: str,
) -> Run | None

Find the exact in-scope ancestor served by a deployment.

The walk is bounded and cycle-detecting. Every row comes through the deployment-bound RunRepository, so a forged parent id cannot cross tenant/workspace scope even when the global run id is guessed.

schedule_ancestor_continuation async

schedule_ancestor_continuation(
    approval_id: str,
    *,
    deployment_ref: str,
    graph_version_ref: str,
) -> Run

Atomically notify the deployment worker about a resolved child gate.

The child stays paused under its own provenance. The root deployment's run is re-queued with a signed linkage record in the same database transaction; its subgraph executor later resumes only the exact child named by the durable pause metadata.

reconcile_ancestor_continuations async

reconcile_ancestor_continuations(
    *, deployment_ref: str, graph_version_ref: str
) -> list[Run]

Repair resolved-child notifications after a process interruption.

The resolved approval is itself the durable reconciliation source. A deployment worker scans only records whose scoped ancestry reaches its exact active graph, and only requeues ancestors still waiting on that child. The transactional CAS and deterministic audit id in schedule_ancestor_continuation make concurrent workers safe.

escalate async

escalate(
    approval_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    deployment_ref: str | None = None,
    graph_version_ref: str | None = None,
) -> ApprovalRecord

Escalate an overdue approval based on its configured escalation action.

Supports three actions: - delegate: marks original ESCALATED, creates new approval for delegate - auto_reject: resolves original as REJECTED with system actor, then continues the run so the rejection actually fails it - alert (default): keeps the original PENDING so a human can still resolve it, but latches it out of the SLA sweep (nulls sla_deadline and marks urgency_metadata['escalated']). Flipping it to ESCALATED would hide it from list/get and make the run unresolvable forever.

If the approval is no longer pending, this is a no-op. That covers both double-escalation and, critically, a RESOLVED approval: SLA enforcement must not re-open a decision a human already made.

resolve async

resolve(
    approval_id: str,
    *,
    decision: ApprovalDecision,
    actor: ActorIdentity,
    edited_payload: dict[str, Any] | None = None,
    reason: str | None = None,
    tenant_id: str | None = None,
    workspace_id: str | None | object = _UNSCOPED,
    deployment_ref: str | None = None,
    graph_version_ref: str | None = None,
) -> ApprovalRecord

Record a decision and return its durable approval record.

schedule_continuation async

schedule_continuation(approval_id: str) -> Run

Prepare a resolved approval for durable worker pick-up.

Instead of driving the orchestrator inline (which conflicts with the worker-ownership model), this method: 1. Prepares the run state (as continue_run would before calling the orchestrator). 2. Transitions the run to PENDING so the worker's poll loop will claim it. 3. Clears the lease so any worker can pick it up.

The worker will call resume_graph on the next poll tick. Only call this from the approval HTTP endpoint when the durable worker is active.

continue_run async

continue_run(
    approval_id: str,
    *,
    graph: Graph,
    orchestrator: ApprovalContinuation,
) -> Run

Resume a paused workflow run after an approval has been resolved.

If the decision was REJECT, the run is marked as failed immediately. If APPROVE or EDIT_AND_APPROVE, the run is handed back to the orchestrator to continue executing from the approval node onward.