Skip to main content

​
Purpose

The Approval Service implements step-up authorization for actions that require human judgment before execution. It handles two authorization decisions from the Policy Engine:
  • STEP_UP — Action is high-risk and requires explicit human approval before proceeding
  • DEFER → escalation — Action was deferred due to insufficient context, and automated resolution was unable to collect enough information; the deferral is escalated to a human approver
In both cases, the action is blocked from executing until a human decision is received or a timeout fires. To prevent approval fatigue, the system utilizes risk levels to ensure only high-impact or genuinely ambiguous actions are routed for human review.
The Approval Service must present full action context to approvers — not just the action itself. Approvers who see only “Agent wants to send an email” cannot make informed decisions. Approvers who see the original user request, prior actions, data classifications accessed, and semantic distance score can.

​
Flow

Action triggers STEP_UP policy (or DEFER escalates)
         ↓
Approval Service receives request with full context
         ↓
┌────────────────────────────────────────────────┐
│  Approver sees:                                │
│  • Original user request                       │
│  • Action: tool, operation, parameters         │
│  • Prior actions in session                    │
│  • Data classifications accessed               │
│  • Semantic distance from original intent      │
│  • Risk level identified                       │
│  • Identity: human principal + service + agent │
│  • Why approval is needed (policy matched)     │
└────────────────────────────────────────────────┘
         ↓
Notification sent to approvers (Slack, email, etc.)
         ↓
Approver reviews action + context
         ↓
    ┌────┴────┐
 Approve    Deny
    ↓         ↓
 Execute    Block
    ↓         ↓
Receipt generated with approver identity and decision

​
Interface

@dataclass
class ApprovalRequest:
    id: str
    action: Action
    context: SessionContext         # Full accumulated context
    identity: ActionIdentity       # Human + service + agent + role/privilege scope
    approvers: list[str]
    risk_level: Literal["LOW", "MEDIUM", "HIGH", "CRITICAL"]
    confidence: float              # Policy confidence 0.0 to 1.0
    timeout: int                   # seconds
    reason: str                    # Why approval is needed (policy matched)
    source: Literal["step_up", "defer_escalation"]
    semantic_distance: float | None  # Distance from original intent

@dataclass
class ApprovalResult:
    granted: bool
    approver: str | None
    reason: str | None
    timestamp: datetime
    identity_verified: bool        # Was approver identity validated?

class ApprovalService:
    async def request(
        self,
        action: Action,
        context: SessionContext,
        identity: ActionIdentity,
        approvers: list[str],
        timeout: int = 3600,
        source: str = "step_up"
    ) -> ApprovalResult:
        """Submit action for human approval. Blocks until decision or timeout."""
        ...

    async def approve(self, request_id: str, approver: str):
        """Approver grants execution. Identity is validated before acceptance."""
        ...

    async def deny(self, request_id: str, approver: str, reason: str):
        """Approver blocks execution with documented reason."""
        ...

​
Implementation

class ApprovalService:
    def __init__(self, notifier: Notifier, store: ApprovalStore):
        self.notifier = notifier
        self.store = store

    async def request(
        self,
        action: Action,
        context: SessionContext,
        identity: ActionIdentity,
        approvers: list[str],
        timeout: int = 3600,
        source: str = "step_up"
    ) -> ApprovalResult:
        # Create request with full context
        request = ApprovalRequest(
            id=generate_id(),
            action=action,
            context=context,
            identity=identity,
            approvers=approvers,
            timeout=timeout,
            reason=self._build_reason(action, context),
            source=source,
            semantic_distance=context.semantic_distance
        )
        await self.store.save(request)

        # Notify approvers with context summary
        await self.notifier.send(approvers, request)

        # Block until decision or timeout
        try:
            result = await asyncio.wait_for(
                self.store.wait_for_decision(request.id),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            result = ApprovalResult(
                granted=False,
                approver=None,
                reason="Timeout — no response within configured window",
                timestamp=datetime.utcnow(),
                identity_verified=False
            )

        # Record in receipt regardless of outcome
        await self.store.record_outcome(request.id, result)
        return result

​
Context Presentation

Effective approval decisions require presenting the right information. The Approval Service should render context appropriate to the notification channel.

​
What Approvers Must See

FieldWhy It Matters
Original user requestWas the user actually asking for this?
Action detailsTool, operation, parameters — what will happen
Prior actionsWhat the agent already did in this session
Data classificationsDid the agent access PII, CONFIDENTIAL, or other sensitive data?
Semantic distanceHow far has the agent drifted from the original request?
Risk levelThe potential impact or blast radius of the action (e.g., CRITICAL for production DB deletes).
Policy ConfidenceHow certain the Policy Engine is that this action matches a specific rule or intent.
Identity chainHuman principal → service account → agent session → role/privilege scope
Policy matchedWhich rule triggered this approval and why
SourceIs this a direct STEP_UP or an escalated DEFER?
Approval fatigue is a real threat. If every other action requires approval, approvers rubber-stamp without reading. The Policy Engine and Deferral Service should resolve what they can autonomously, routing only genuinely ambiguous or high-risk actions to humans. See Research Directions for work on dynamic approval thresholds.

​
Notifiers

​
Slack

class SlackNotifier:
    async def send(self, approvers: list[str], request: ApprovalRequest):
        source_label = (
            "⏸️ Escalated from DEFER" 
            if request.source == "defer_escalation" 
            else "🔐 Approval Required"
        )
        drift_warning = ""
        if request.semantic_distance and request.semantic_distance > 0.5:
            drift_warning = f"\n⚠️ *Semantic distance:* {request.semantic_distance:.2f} — action may have drifted from original intent"

        blocks = [
            {"type": "header", "text": {
                "type": "plain_text", "text": source_label
            }},
            {"type": "section", "text": {"type": "mrkdwn", "text": "\n".join([
                f"*Tool:* `{request.action.tool}.{request.action.operation}`",
                f"*User:* {request.identity.human_principal}",
                f"*Original request:* {request.context.original_request[:200]}",
                f"*Prior actions:* {len(request.context.prior_actions)} in session",
                f"*Data accessed:* {', '.join(request.context.data_classifications) or 'None flagged'}",
                f"*Risk Level:* `{request.risk_level}`",
                f"*Policy Confidence:* {conf_color} {request.policy_confidence * 100:.0f}%",
                f"*Reason:* {request.reason}",
                drift_warning
            ])}},
            {"type": "section", "text": {"type": "mrkdwn", "text":
                f"*Parameters:*\n```{json.dumps(request.action.parameters, indent=2)[:500]}```"
            }},
            {"type": "actions", "elements": [
                {"type": "button", "text": {"type": "plain_text", "text": "✅ Approve"},
                 "style": "primary", "action_id": f"approve_{request.id}"},
                {"type": "button", "text": {"type": "plain_text", "text": "❌ Deny"},
                 "style": "danger", "action_id": f"deny_{request.id}"}
            ]}
        ]
        for approver in approvers:
            await self.client.chat_postMessage(channel=approver, blocks=blocks)

​
Email

class EmailNotifier:
    async def send(self, approvers: list[str], request: ApprovalRequest):
        subject_prefix = (
            "[DEFER ESCALATION]" 
            if request.source == "defer_escalation" 
            else "[APPROVAL REQUIRED]"
        )
        for approver in approvers:
            await self.send_email(
                to=approver,
                subject=f"{subject_prefix} {request.action.tool}.{request.action.operation}",
                body=self.render_template(request)  # Template includes full context
            )

​
Identity Preservation

Actions may be deferred or queued for approval for extended periods. The Approval Service must preserve the identity context captured at the time of the original action submission:
  • Human principal at action time — not at approval time
  • Service identity and agent session from the original request
  • Role and privilege scope as they existed when the action was submitted
If identity claims (tokens, sessions, role assignments) expire before approval is granted, the system must re-validate them before executing. Stale identity must never be used to authorize execution. See R5: Identity Binding.

​
Configuration

approval:
  default_timeout: 3600       # 1 hour
  timeout_action: DENY        # DENY on timeout (recommended)
  
  # Identity validation
  revalidate_identity: true   # Re-check identity claims before executing approved actions
  
  notifiers:
    - type: slack
      channel_mapping:
        security-team: "#security-approvals"
        database-owner: "@db-admin"
    
    - type: email
      from: approvals@company.com
  
  escalation:
    enabled: true
    after: 1800                # 30 minutes
    to: [security-manager]
  
  # Defer escalation handling
  defer_escalation:
    enabled: true
    label_prefix: "[DEFER]"   # Visual distinction from direct STEP_UP
    priority: high             # Deferred actions have been waiting longer

​
Interaction with Deferral Service

The Approval Service and Deferral Service are complementary:
DecisionFirst handled byEscalates to Approval Service when…
STEP_UPApproval Service directly— (starts here)
DEFERDeferral ServiceAutomated resolution fails or times out
When a deferred action escalates, the Approval Service receives the accumulated context from the Deferral Service’s resolution attempts, giving the human approver more information than was available at the original deferral.

​
Requirements

RequirementLevelNotes
Block execution until decision receivedMUSTNo effects may occur before approval (R1, R4)
Route to configured approversMUST
Enforce configurable timeoutMUSTDENY on timeout recommended (R4)
Present full action context to approversMUSTOriginal request, prior actions, data classifications, identity chain
Record approver identity and decision in receiptMUSTIncluding timestamp and reason (R5)
Preserve original identity context across deferral/approvalMUSTRe-validate before executing if claims may have expired (R6)
Accept escalations from Deferral ServiceMUSTDEFER → STEP_UP pathway
Expose Risk LevelMUSTHigh-risk actions should be visually prioritized in UI/Slack.
Expose Policy ConfidenceSHOULDHelps approvers understand if the system is “unsure” vs “highly certain.”
Support multiple notification channelsSHOULDSlack, email, webhook, etc.
Escalation on non-responseSHOULDRoute to backup approver after configurable window
Visual distinction between STEP_UP and DEFER escalationsSHOULDApprovers should know why they’re being asked