Human-in-the-Loop

Human-in-the-Loop Escalation Gating

When a policy decision is escalate, the agent can now block and wait until a human reviews the action on the Authorisation Layer dashboard. This is controlled by the wait_on_escalate parameter on the @guard decorator (default: True).

Full Flow

1. Agent calls the guarded function → Authorisation Layer intercepts and returns decision: "escalate" with an escalation_id.
2. SDK polls GET /v1/enforce/escalations/{id}/status every escalation_poll_interval seconds.
3. Human opens Authorisation Layer → Control Plane, sees the pending escalation card with Approve and Reject buttons.
4. If approved: agent unblocks and the wrapped function executes normally.
5. If rejected (or timeout exceeded): AgentBlockedError is raised and the function never runs.

Code Example

from xybern import Xybern, AgentCredential, AgentBlockedError

client = Xybern(api_key="xb_your_key")
cred   = AgentCredential.load("./bot.cred")

@client.agents.guard(
    credential=cred,
    wait_on_escalate=True,      # Block until human decides (default)
    escalation_timeout=300,     # Wait up to 5 minutes
    escalation_poll_interval=5, # Check every 5 seconds
)
def wire_transfer(amount_usd: float, recipient: str):
    # Only executes if a human approves on the Authorisation Layer dashboard
    execute_wire(amount_usd, recipient)

try:
    wire_transfer(50000, "vendor@example.com")
except AgentBlockedError as e:
    print(f"Transfer rejected or timed out: {e}")

InterceptResult Attributes

Attribute	Type	Description
.decision	str	"allow", "block", or "escalate"
.trust_score	float	Agent trust score at time of decision (0-100)
.decision_id	str	Immutable decision record ID
.escalation_id	str | None	Set when decision == "escalate"; used to poll for human resolution
.policy_name	str | None	Name of the policy that triggered this decision
.latency_ms	int	Decision latency in milliseconds

wait_for_escalation() Method

The @guard decorator calls this automatically when wait_on_escalate=True. You can also call it directly when handling escalations manually.

# Manual escalation polling (when wait_on_escalate=False)
result = client.agents.intercept(
    action_type="wire_transfer",
    action_content="Transfer $50,000 to vendor",
    agent_id=cred.agent_id,
)

if result.decision == "escalate":
    resolution = client.agents.wait_for_escalation(
        escalation_id=result.escalation_id,
        timeout=300,        # seconds
        poll_interval=5,    # seconds
    )
    # resolution == "approved" or "rejected"
    if resolution == "approved":
        execute_wire(50000, "vendor@example.com")

Escalation Status Endpoint

GET /v1/enforce/escalations/{escalation_id}/status Poll the resolution status of a pending escalation. Returns {"ok": true, "status": "pending"|"approved"|"rejected"}. Used internally by the SDK's wait_for_escalation() method.

Dashboard: Approve / Reject

Pending escalations appear as cards in Authorisation Layer → Control Plane. Each card shows the action details, agent identity, and two buttons: Approve and Reject. Resolving a card immediately unblocks any waiting SDK call.

The underlying endpoint is: POST /api/sentinel/enforcement/escalations/{escalation_id}/resolve with body {"resolution": "approved"|"rejected"}.

Info

Set wait_on_escalate=False if you want fire-and-forget escalation behaviour, the SDK returns immediately with decision="escalate" and you handle the escalation_id asynchronously.