forked from cardosofelipe/fast-next-template
Felipe Cardoso
5594655fba
Add comprehensive spike research documents: - SPIKE-002: Agent Orchestration Pattern (LangGraph + Temporal hybrid) - SPIKE-006: Knowledge Base pgvector (RAG with hybrid search) - SPIKE-007: Agent Communication Protocol (JSON-RPC + Redis Streams) - SPIKE-008: Workflow State Machine (transitions lib + event sourcing) - SPIKE-009: Issue Synchronization (bi-directional sync with conflict resolution) - SPIKE-010: Cost Tracking (LiteLLM callbacks + budget enforcement) - SPIKE-011: Audit Logging (structured event sourcing) - SPIKE-012: Client Approval Flow (checkpoint-based approvals) Add architecture documentation: - ARCHITECTURE_DEEP_ANALYSIS.md: Memory management, security, testing strategy - IMPLEMENTATION_ROADMAP.md: 6-phase, 24-week implementation plan Closes #2, #6, #7, #8, #9, #10, #11, #12 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
59 KiB
59 KiB
SPIKE-012: Client Approval Flow
Status: Completed Date: 2025-12-29 Author: Architecture Team Related Issue: #12
Executive Summary
This spike researches the optimal patterns for implementing client approval workflows in Syndarix, enabling human oversight at configurable checkpoints within autonomous agent operations. Based on industry research and Syndarix's existing architecture, we recommend a checkpoint-based approval system with confidence-aware routing, multi-channel notifications, and timeout escalation.
Key Recommendations
- Adopt checkpoint-based approval pattern aligned with Syndarix's three autonomy levels
- Use confidence-based routing to automatically escalate uncertain AI decisions
- Implement batch approval UI for FULL_CONTROL mode efficiency
- Leverage existing SSE infrastructure for real-time approval notifications
- Support mobile-responsive approval interface with push notification integration
- Design flexible timeout and escalation policies per project configuration
Table of Contents
- Research Questions
- Syndarix Autonomy Levels
- Approval UX Patterns
- Presenting AI Decisions for Review
- Timeout and Escalation Handling
- Delegation Patterns
- Batch vs Individual Approval
- Mobile-Friendly Interface
- Audit Trail Design
- Notification System Integration
- Database Schema
- Code Examples
- UI Mockup Descriptions
- Implementation Roadmap
1. Research Questions
| Question | Summary Answer |
|---|---|
| Best UX patterns for approval workflows? | Checkpoint-based with queue management and clear status visibility |
| How to present AI decisions for human review? | Explainable AI with confidence scores, reasoning chains, and impact summaries |
| Timeout handling for pending approvals? | Configurable escalation paths with auto-approve/block policies |
| Delegation and escalation patterns? | Role-based delegation with backup approvers and team escalation |
| Batch approval vs individual approval? | Context-dependent: batch for routine decisions, individual for high-stakes |
| Mobile-friendly approval interfaces? | One-touch approve/reject with push notifications and minimal context display |
| Audit trail for approval decisions? | Immutable event log with decision rationale and full context snapshot |
2. Syndarix Autonomy Levels
Syndarix supports three autonomy levels that determine when client approval is required:
2.1 Autonomy Level Matrix
| Action | FULL_CONTROL | MILESTONE | AUTONOMOUS |
|---|---|---|---|
| Requirements approval | Required | Required | Required |
| Architecture approval | Required | Required | Required |
| Sprint start | Required | Required | Auto |
| Story implementation | Required | Auto | Auto |
| PR merge | Required | Auto | Auto |
| Sprint completion | Required | Required | Auto |
| Bug fixes | Required | Auto | Auto |
| Documentation updates | Required | Auto | Auto |
| Budget threshold exceeded | Required | Required | Required |
| Production deployment | Required | Required | Required |
| Agent conflict resolution | Required | Required | Required |
2.2 Approval Checkpoint Categories
class ApprovalCategory(str, Enum):
"""Categories of approval checkpoints."""
# Always require approval regardless of autonomy level
CRITICAL = "critical" # Budget, production, architecture
# Require approval at MILESTONE and FULL_CONTROL
MILESTONE = "milestone" # Sprint boundaries, major features
# Only require approval at FULL_CONTROL
ROUTINE = "routine" # Individual stories, bug fixes
# Agent-initiated (confidence below threshold)
UNCERTAINTY = "uncertainty" # Low confidence decisions
# Human expertise explicitly requested
EXPERTISE = "expertise" # Agent needs human input
2.3 Approval Triggers
| Scenario | Category | Description |
|---|---|---|
| Before starting a new sprint | MILESTONE | Sprint scope and goal confirmation |
| Before merging PR to main branch | ROUTINE | Code review approval |
| Before deploying to production | CRITICAL | Release sign-off |
| When budget threshold exceeded | CRITICAL | Cost authorization |
| When agent requests human expertise | EXPERTISE | Knowledge gap |
| When conflicting decisions between agents | UNCERTAINTY | Conflict resolution |
| Low confidence AI decision | UNCERTAINTY | Automatic escalation |
3. Approval UX Patterns
3.1 Recommended Pattern: Queue-Based Checkpoint System
Based on industry research, we recommend a queue-based approval system with the following characteristics:
+------------------+ +------------------+ +------------------+
| Approval Queue | --> | Review Panel | --> | Decision Log |
+------------------+ +------------------+ +------------------+
| - Priority lanes | | - Context view | | - Audit trail |
| - Grouping | | - Action buttons | | - Metrics |
| - SLA indicators | | - History | | - Analytics |
+------------------+ +------------------+ +------------------+
3.2 Queue Pattern Components
Single Queue with Priority Lanes:
+---------------------------------------------------------------+
| APPROVAL QUEUE |
+---------------------------------------------------------------+
| [CRITICAL - 2] | [MILESTONE - 5] | [ROUTINE - 12] |
+---------------------------------------------------------------+
| > Budget overrun (2h ago) | SLA: 4h | URGENT |
| > Sprint 3 start request | SLA: 24h | PENDING |
| > PR #45: Add user authentication | SLA: 48h | PENDING |
| > PR #46: Fix login validation | SLA: 48h | PENDING |
| > [Batch Select: 8 routine items] | | BULK |
+---------------------------------------------------------------+
3.3 Key UX Principles
- Status Visibility: Clear indicators showing approved, pending, rejected states with color coding
- Context Preservation: Full decision context without requiring navigation away
- One-Touch Actions: Approve/Reject buttons immediately visible
- Batch Operations: Select multiple items for bulk approval
- SLA Tracking: Visual indicators for time-sensitive decisions
- Undo Capability: Brief window to reverse accidental approvals
4. Presenting AI Decisions for Review
4.1 Explainable AI Interface
AI decisions must be presented with sufficient context for informed human review:
+---------------------------------------------------------------+
| APPROVAL REQUEST: Architecture Decision |
+---------------------------------------------------------------+
| |
| DECISION: Use PostgreSQL with pgvector for knowledge base |
| |
| CONFIDENCE: 87% [=========> ] |
| |
| REASONING: |
| 1. Requirement: Vector similarity search for RAG |
| 2. Constraint: Must integrate with existing PostgreSQL |
| 3. Alternative considered: Pinecone (rejected: external dep) |
| 4. Alternative considered: Milvus (rejected: operational cost)|
| |
| IMPACT ASSESSMENT: |
| - Cost: $0 additional (existing infrastructure) |
| - Complexity: Low (native PostgreSQL extension) |
| - Risk: Low (mature, well-documented) |
| |
| RECOMMENDED BY: Architect Agent (Sofia) |
| SUPPORTED BY: DevOps Agent (Marcus) - "Easy to maintain" |
| |
| [APPROVE] [REJECT] [REQUEST MORE INFO] [DELEGATE] |
+---------------------------------------------------------------+
4.2 Information Hierarchy
| Level | Content | Always Shown |
|---|---|---|
| 1. Decision Summary | One-line description | Yes |
| 2. Confidence Score | Visual indicator (0-100%) | Yes |
| 3. Key Reasoning | Top 3 factors | Yes |
| 4. Impact Assessment | Cost, risk, complexity | Yes |
| 5. Alternatives Considered | Other options evaluated | Expandable |
| 6. Full Agent Conversation | Complete discussion thread | Expandable |
| 7. Related Artifacts | Code, documents, diagrams | Linked |
4.3 Confidence-Based Presentation
def get_presentation_level(confidence: float, category: ApprovalCategory) -> str:
"""Determine how much detail to present based on confidence and category."""
if category == ApprovalCategory.CRITICAL:
return "full" # Always show full context for critical decisions
if confidence >= 0.9:
return "summary" # High confidence: summary with expand option
elif confidence >= 0.7:
return "detailed" # Medium confidence: show reasoning
else:
return "full" # Low confidence: full context with alternatives
5. Timeout and Escalation Handling
5.1 Escalation Policy Configuration
@dataclass
class EscalationPolicy:
"""Configuration for approval timeout and escalation."""
# Initial timeout before first escalation
initial_timeout_hours: int = 24
# Maximum escalation levels
max_escalation_levels: int = 3
# Escalation path
escalation_path: list[str] = field(default_factory=lambda: [
"project_owner",
"team_lead",
"admin"
])
# Reminder intervals before timeout
reminder_intervals: list[int] = field(default_factory=lambda: [
4, # 4 hours before timeout
1, # 1 hour before timeout
])
# Action on final timeout
final_action: str = "block" # "block", "auto_approve", "auto_reject"
# Category-specific overrides
category_overrides: dict[ApprovalCategory, dict] = field(default_factory=dict)
5.2 Timeout Behavior by Category
| Category | Default Timeout | Escalation Path | Final Action |
|---|---|---|---|
| CRITICAL | 4 hours | Owner -> Admin -> Block | Block (halt work) |
| MILESTONE | 24 hours | Owner -> Team Lead | Block |
| ROUTINE | 48 hours | Owner only | Auto-approve (with flag) |
| UNCERTAINTY | 12 hours | Owner -> Architect | Request more info |
| EXPERTISE | 24 hours | Owner -> External | Block |
5.3 Escalation Flow
+-------------+ +-------------+ +-------------+
| Created | --> | Pending | --> | Escalated |
+-------------+ +-------------+ +-------------+
| |
[Timeout] [Timeout]
| |
v v
+-------------+ +-------------+
| Reminder | | Final |
| Sent | | Action |
+-------------+ +-------------+
5.4 Notification Sequence
async def handle_approval_timeout(approval: ApprovalRequest):
"""Process approval timeout with escalation."""
current_level = approval.escalation_level
policy = get_escalation_policy(approval.project_id, approval.category)
if current_level < policy.max_escalation_levels:
# Escalate to next level
next_approver = policy.escalation_path[current_level]
await notify_escalation(
approval=approval,
new_approver=next_approver,
reason="timeout",
previous_approver=approval.current_approver
)
approval.current_approver = next_approver
approval.escalation_level += 1
approval.escalated_at = datetime.utcnow()
else:
# Final action
if policy.final_action == "block":
await block_workflow(approval)
await notify_workflow_blocked(approval)
elif policy.final_action == "auto_approve":
await auto_approve(approval, reason="timeout_policy")
await notify_auto_approved(approval)
elif policy.final_action == "auto_reject":
await auto_reject(approval, reason="timeout_policy")
await notify_auto_rejected(approval)
6. Delegation Patterns
6.1 Delegation Types
| Type | Description | Use Case |
|---|---|---|
| Temporary | Delegate for specific time period | Vacation, OOO |
| Permanent | Assign backup approver | Team redundancy |
| Categorical | Delegate by approval category | Expertise-based routing |
| Threshold | Delegate below certain impact level | Routine automation |
6.2 Delegation Configuration
@dataclass
class DelegationRule:
"""Rule for delegating approval authority."""
delegator_id: UUID
delegate_id: UUID
# Scope
project_ids: list[UUID] | None = None # None = all projects
categories: list[ApprovalCategory] | None = None
# Constraints
max_impact_level: str | None = None # "low", "medium", "high"
requires_notification: bool = True
# Time bounds
start_date: datetime | None = None
end_date: datetime | None = None
# Status
is_active: bool = True
6.3 Delegation Resolution
async def resolve_approver(
approval: ApprovalRequest,
original_approver_id: UUID
) -> UUID:
"""Resolve the actual approver considering delegation rules."""
# Check for active delegation rules
delegation = await get_active_delegation(
delegator_id=original_approver_id,
project_id=approval.project_id,
category=approval.category,
impact_level=approval.impact_level
)
if delegation:
# Verify delegate is available
if await is_user_available(delegation.delegate_id):
if delegation.requires_notification:
await notify_delegation_used(delegation, approval)
return delegation.delegate_id
return original_approver_id
7. Batch vs Individual Approval
7.1 Decision Framework
| Factor | Batch Approval | Individual Approval |
|---|---|---|
| Category | ROUTINE | CRITICAL, UNCERTAINTY |
| Confidence | > 85% | < 85% |
| Similar context | Same sprint/feature | Different contexts |
| Risk level | Low | Medium/High |
| Time pressure | High volume | Single items |
7.2 Batch Approval UI
+---------------------------------------------------------------+
| BATCH APPROVAL: 8 Routine Code Reviews |
+---------------------------------------------------------------+
| |
| SUMMARY: |
| - Sprint: Sprint 3 - Authentication |
| - Agent: Dave (Software Engineer) |
| - Average Confidence: 91% |
| - Test Coverage: All passing, avg 87% coverage |
| |
| ITEMS: |
| [x] PR #45: Add login form validation | 94% | Low risk |
| [x] PR #46: Add password reset endpoint | 92% | Low risk |
| [x] PR #47: Add email verification | 89% | Low risk |
| [x] PR #48: Add session management | 91% | Low risk |
| [ ] PR #49: Add OAuth integration | 76% | Med risk | <- Excluded
| [x] PR #50: Add logout functionality | 95% | Low risk |
| [x] PR #51: Add remember me feature | 88% | Low risk |
| [x] PR #52: Add login rate limiting | 93% | Low risk |
| |
| [APPROVE ALL (7)] [REJECT ALL] [REVIEW INDIVIDUALLY] |
+---------------------------------------------------------------+
7.3 Batch Approval Rules
class BatchApprovalPolicy:
"""Policy for batch approval eligibility."""
# Minimum items for batch
min_items: int = 3
# Maximum items per batch
max_items: int = 20
# Grouping criteria
group_by: list[str] = ["sprint_id", "agent_type", "category"]
# Eligibility criteria
min_confidence: float = 0.85
max_risk_level: str = "low"
excluded_categories: list[ApprovalCategory] = [
ApprovalCategory.CRITICAL,
ApprovalCategory.UNCERTAINTY
]
# Require same context
require_same_sprint: bool = True
require_same_feature: bool = False
8. Mobile-Friendly Interface
8.1 Mobile Design Principles
- Progressive Disclosure: Show minimal info, expand on tap
- Swipe Actions: Swipe right to approve, left to reject
- Push Notifications: Actionable notifications with quick responses
- Offline Queue: Cache pending approvals for offline review
- Biometric Auth: Face ID / fingerprint for high-stakes approvals
8.2 Mobile Notification Design
+----------------------------------+
| SYNDARIX now |
| Approval Required |
| |
| Sprint 3 Ready to Start |
| Project: E-Commerce Platform |
| Confidence: 92% |
| |
| [APPROVE] [REJECT] [VIEW] |
+----------------------------------+
8.3 Mobile Approval Card
+----------------------------------+
| < Back Approval #127 |
+----------------------------------+
| |
| Sprint 3 Start Request |
| ============================ |
| |
| Requested by: PM Agent (Alex) |
| 2 hours ago |
| |
| [=========> ] 92% Confidence |
| |
| Sprint Goal: |
| Implement user authentication |
| and authorization system |
| |
| Stories: 8 | Points: 21 |
| Estimated: 2 weeks |
| |
| [v] Show Details |
| |
+----------------------------------+
| |
| [ REJECT ] [ APPROVE ] |
| |
+----------------------------------+
8.4 Responsive Breakpoints
| Breakpoint | Layout | Features |
|---|---|---|
| Mobile (< 640px) | Single column, cards | Swipe actions, minimal info |
| Tablet (640-1024px) | Two columns | Side-by-side compare |
| Desktop (> 1024px) | Full queue + detail panel | Batch operations, keyboard shortcuts |
9. Audit Trail Design
9.1 Audit Event Structure
@dataclass
class ApprovalAuditEvent:
"""Immutable audit record for approval decisions."""
id: UUID
approval_id: UUID
# Event info
event_type: str # "created", "viewed", "approved", "rejected", "escalated", "delegated"
timestamp: datetime
# Actor info
actor_id: UUID
actor_type: str # "user", "system", "timeout"
actor_ip: str | None
actor_device: str | None
# Decision info (for approve/reject)
decision: str | None
rationale: str | None
# Context snapshot (immutable copy at decision time)
context_snapshot: dict
# Metadata
duration_seconds: int | None # Time spent reviewing
confidence_at_decision: float | None
9.2 Audit Trail Query API
class ApprovalAuditService:
"""Service for querying and analyzing approval audit trails."""
async def get_audit_trail(
self,
approval_id: UUID
) -> list[ApprovalAuditEvent]:
"""Get complete audit trail for an approval."""
pass
async def get_user_decisions(
self,
user_id: UUID,
start_date: datetime,
end_date: datetime
) -> list[ApprovalAuditEvent]:
"""Get all decisions made by a user in a time range."""
pass
async def get_approval_metrics(
self,
project_id: UUID | None = None,
start_date: datetime | None = None
) -> ApprovalMetrics:
"""Get approval workflow metrics."""
pass
9.3 Audit Metrics Dashboard
| Metric | Description | Target |
|---|---|---|
| Average Decision Time | Time from request to decision | < 4 hours (CRITICAL) |
| Approval Rate | % of approvals vs rejections | > 80% |
| Escalation Rate | % requiring escalation | < 10% |
| Timeout Rate | % hitting timeout | < 5% |
| Override Rate | Auto-decisions overridden | Track only |
| Batch Efficiency | Items approved in batch vs individual | > 60% for ROUTINE |
10. Notification System Integration
10.1 Notification Channels
Leveraging Syndarix's existing SSE infrastructure (SPIKE-003):
| Channel | Use Case | Priority |
|---|---|---|
| In-App (SSE) | Real-time dashboard updates | All |
| Digest and escalation alerts | MILESTONE, CRITICAL | |
| Push (Mobile) | Urgent approvals | CRITICAL, ESCALATED |
| Slack/Teams | Team notifications | Configurable |
| SMS | Critical timeout warnings | CRITICAL (final escalation) |
10.2 Notification Events
class ApprovalNotificationEvent(str, Enum):
"""Events that trigger notifications."""
# New approval
APPROVAL_REQUESTED = "approval_requested"
# Reminders
APPROVAL_REMINDER = "approval_reminder"
# Escalation
APPROVAL_ESCALATED = "approval_escalated"
APPROVAL_ESCALATED_TO_YOU = "approval_escalated_to_you"
# Resolution
APPROVAL_APPROVED = "approval_approved"
APPROVAL_REJECTED = "approval_rejected"
APPROVAL_AUTO_APPROVED = "approval_auto_approved"
# Delegation
APPROVAL_DELEGATED = "approval_delegated"
# Workflow impact
WORKFLOW_BLOCKED = "workflow_blocked"
WORKFLOW_RESUMED = "workflow_resumed"
10.3 SSE Integration
# Extend existing SSE event types from SPIKE-003
class EventType(str, Enum):
# ... existing events ...
# Approval Events
APPROVAL_REQUESTED = "approval_requested"
APPROVAL_UPDATED = "approval_updated"
APPROVAL_RESOLVED = "approval_resolved"
APPROVAL_ESCALATED = "approval_escalated"
APPROVAL_BATCH_READY = "approval_batch_ready"
# Publishing approval events
async def publish_approval_event(approval: ApprovalRequest, event_type: str):
"""Publish approval event via existing EventBus."""
await event_bus.publish(
channel=f"project:{approval.project_id}",
event=Event(
type=event_type,
data={
"approval_id": str(approval.id),
"category": approval.category.value,
"summary": approval.summary,
"confidence": approval.confidence,
"deadline": approval.deadline.isoformat(),
"approver_id": str(approval.current_approver_id),
},
project_id=str(approval.project_id),
agent_id=str(approval.requesting_agent_id) if approval.requesting_agent_id else None
)
)
11. Database Schema
11.1 Core Tables
# app/models/approval.py
from sqlalchemy import Column, String, Text, Float, ForeignKey, Enum, Boolean
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import relationship
from app.models.base import Base, TimestampMixin, UUIDMixin
class ApprovalCategory(str, Enum):
CRITICAL = "critical"
MILESTONE = "milestone"
ROUTINE = "routine"
UNCERTAINTY = "uncertainty"
EXPERTISE = "expertise"
class ApprovalStatus(str, Enum):
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
ESCALATED = "escalated"
EXPIRED = "expired"
DELEGATED = "delegated"
class ApprovalRequest(Base, UUIDMixin, TimestampMixin):
"""Approval request entity."""
__tablename__ = "approval_requests"
# Foreign keys
project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id"), nullable=False, index=True)
sprint_id = Column(UUID(as_uuid=True), ForeignKey("sprints.id"), nullable=True)
requesting_agent_id = Column(UUID(as_uuid=True), ForeignKey("agent_instances.id"), nullable=True)
current_approver_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False, index=True)
original_approver_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
# Approval details
category = Column(Enum(ApprovalCategory), nullable=False, index=True)
status = Column(Enum(ApprovalStatus), default=ApprovalStatus.PENDING, index=True)
# Content
title = Column(String(255), nullable=False)
summary = Column(Text, nullable=False)
detailed_context = Column(JSONB, nullable=False, default=dict)
# AI decision metadata
confidence = Column(Float, nullable=True)
reasoning = Column(JSONB, nullable=True) # Structured reasoning chain
alternatives_considered = Column(JSONB, nullable=True)
impact_assessment = Column(JSONB, nullable=True)
# Workflow reference
workflow_type = Column(String(50), nullable=False) # "sprint_start", "pr_merge", etc.
workflow_reference_id = Column(UUID(as_uuid=True), nullable=True) # PR ID, Sprint ID, etc.
workflow_reference_url = Column(String(500), nullable=True) # External link
# Escalation tracking
escalation_level = Column(Integer, default=0)
escalated_at = Column(DateTime(timezone=True), nullable=True)
escalation_reason = Column(String(255), nullable=True)
# Timing
deadline = Column(DateTime(timezone=True), nullable=False)
reminded_at = Column(DateTime(timezone=True), nullable=True)
resolved_at = Column(DateTime(timezone=True), nullable=True)
# Resolution
resolution_decision = Column(String(50), nullable=True) # "approved", "rejected"
resolution_rationale = Column(Text, nullable=True)
resolved_by_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
resolution_type = Column(String(50), nullable=True) # "manual", "auto", "timeout"
# Batch tracking
batch_id = Column(UUID(as_uuid=True), nullable=True, index=True)
is_batch_eligible = Column(Boolean, default=True)
# Relationships
project = relationship("Project", back_populates="approval_requests")
current_approver = relationship("User", foreign_keys=[current_approver_id])
resolved_by = relationship("User", foreign_keys=[resolved_by_id])
audit_events = relationship("ApprovalAuditEvent", back_populates="approval_request")
class ApprovalAuditEvent(Base, UUIDMixin):
"""Immutable audit trail for approval decisions."""
__tablename__ = "approval_audit_events"
approval_id = Column(UUID(as_uuid=True), ForeignKey("approval_requests.id"), nullable=False, index=True)
# Event details
event_type = Column(String(50), nullable=False) # "created", "viewed", "approved", etc.
timestamp = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
# Actor
actor_id = Column(UUID(as_uuid=True), nullable=False)
actor_type = Column(String(20), nullable=False) # "user", "system", "timeout"
actor_ip = Column(String(45), nullable=True)
actor_user_agent = Column(String(500), nullable=True)
# Decision details (for resolution events)
decision = Column(String(50), nullable=True)
rationale = Column(Text, nullable=True)
# Context snapshot (immutable)
context_snapshot = Column(JSONB, nullable=False)
# Metrics
review_duration_seconds = Column(Integer, nullable=True)
# Relationships
approval_request = relationship("ApprovalRequest", back_populates="audit_events")
class DelegationRule(Base, UUIDMixin, TimestampMixin):
"""Delegation rules for approval authority."""
__tablename__ = "delegation_rules"
delegator_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False, index=True)
delegate_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
# Scope
project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id"), nullable=True) # None = all projects
categories = Column(JSONB, nullable=True) # None = all categories
# Constraints
max_impact_level = Column(String(20), nullable=True)
requires_notification = Column(Boolean, default=True)
# Time bounds
start_date = Column(DateTime(timezone=True), nullable=True)
end_date = Column(DateTime(timezone=True), nullable=True)
# Status
is_active = Column(Boolean, default=True)
# Relationships
delegator = relationship("User", foreign_keys=[delegator_id])
delegate = relationship("User", foreign_keys=[delegate_id])
class EscalationPolicy(Base, UUIDMixin, TimestampMixin):
"""Project-level escalation policy configuration."""
__tablename__ = "escalation_policies"
project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id"), nullable=False, unique=True)
# Default settings
default_timeout_hours = Column(Integer, default=24)
max_escalation_levels = Column(Integer, default=3)
# Escalation path (list of role names)
escalation_path = Column(JSONB, default=["project_owner", "team_lead", "admin"])
# Reminder intervals (hours before timeout)
reminder_intervals = Column(JSONB, default=[4, 1])
# Final action
final_action = Column(String(20), default="block") # "block", "auto_approve", "auto_reject"
# Category-specific overrides
category_overrides = Column(JSONB, default=dict)
# Relationships
project = relationship("Project", back_populates="escalation_policy")
11.2 Database Indexes
-- Performance indexes for approval queries
CREATE INDEX idx_approval_requests_pending ON approval_requests(project_id, status)
WHERE status = 'pending';
CREATE INDEX idx_approval_requests_approver ON approval_requests(current_approver_id, status)
WHERE status = 'pending';
CREATE INDEX idx_approval_requests_deadline ON approval_requests(deadline)
WHERE status = 'pending';
CREATE INDEX idx_approval_requests_batch ON approval_requests(batch_id)
WHERE batch_id IS NOT NULL;
CREATE INDEX idx_audit_events_approval ON approval_audit_events(approval_id, timestamp);
CREATE INDEX idx_delegation_active ON delegation_rules(delegator_id, is_active)
WHERE is_active = true;
12. Code Examples
12.1 Approval Gate Decorator
# app/services/approval/decorators.py
from functools import wraps
from typing import Callable, Any
from app.models.approval import ApprovalCategory, ApprovalRequest
from app.services.approval.service import ApprovalService
def requires_approval(
category: ApprovalCategory,
title_template: str,
summary_template: str,
confidence_threshold: float = 0.85
):
"""
Decorator to create an approval checkpoint before executing an action.
Usage:
@requires_approval(
category=ApprovalCategory.MILESTONE,
title_template="Start Sprint {sprint_name}",
summary_template="Sprint goal: {sprint_goal}\nStories: {story_count}"
)
async def start_sprint(self, sprint_id: UUID, **kwargs) -> Sprint:
# This only executes after approval
...
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(self, *args, **kwargs) -> Any:
# Get context from kwargs
project_id = kwargs.get('project_id')
agent_id = kwargs.get('agent_id')
confidence = kwargs.get('confidence', 1.0)
# Check if approval is needed based on autonomy level
project = await get_project(project_id)
if should_require_approval(
autonomy_level=project.autonomy_level,
category=category,
confidence=confidence,
threshold=confidence_threshold
):
# Create approval request
approval_service = ApprovalService()
approval = await approval_service.create_approval(
project_id=project_id,
requesting_agent_id=agent_id,
category=category,
title=title_template.format(**kwargs),
summary=summary_template.format(**kwargs),
confidence=confidence,
workflow_type=func.__name__,
workflow_context=kwargs
)
# Wait for approval (blocking)
result = await approval_service.wait_for_resolution(approval.id)
if result.status == ApprovalStatus.REJECTED:
raise ApprovalRejectedError(
f"Approval rejected: {result.resolution_rationale}"
)
# Execute the action
return await func(self, *args, **kwargs)
return wrapper
return decorator
def should_require_approval(
autonomy_level: str,
category: ApprovalCategory,
confidence: float,
threshold: float
) -> bool:
"""Determine if approval is required based on autonomy and confidence."""
# Critical always requires approval
if category == ApprovalCategory.CRITICAL:
return True
# Low confidence triggers approval regardless of autonomy
if confidence < threshold:
return True
# Check autonomy level
if autonomy_level == "FULL_CONTROL":
return True
elif autonomy_level == "MILESTONE":
return category in [ApprovalCategory.MILESTONE, ApprovalCategory.CRITICAL]
else: # AUTONOMOUS
return category == ApprovalCategory.CRITICAL
12.2 Approval Service
# app/services/approval/service.py
from datetime import datetime, timedelta
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.approval import ApprovalRequest, ApprovalStatus, ApprovalAuditEvent
from app.services.events import EventBus
class ApprovalService:
"""Service for managing approval workflows."""
def __init__(self, db: AsyncSession, event_bus: EventBus):
self.db = db
self.event_bus = event_bus
async def create_approval(
self,
project_id: UUID,
requesting_agent_id: UUID | None,
category: ApprovalCategory,
title: str,
summary: str,
confidence: float | None = None,
workflow_type: str = "generic",
workflow_context: dict | None = None,
reasoning: dict | None = None,
alternatives: list[dict] | None = None,
impact_assessment: dict | None = None,
) -> ApprovalRequest:
"""Create a new approval request."""
# Get project and escalation policy
project = await self._get_project(project_id)
policy = await self._get_escalation_policy(project_id)
# Determine approver (with delegation check)
approver_id = await self._resolve_approver(
project.owner_id, project_id, category
)
# Calculate deadline
timeout_hours = self._get_timeout_hours(policy, category)
deadline = datetime.utcnow() + timedelta(hours=timeout_hours)
# Create approval request
approval = ApprovalRequest(
project_id=project_id,
sprint_id=workflow_context.get('sprint_id') if workflow_context else None,
requesting_agent_id=requesting_agent_id,
current_approver_id=approver_id,
original_approver_id=project.owner_id,
category=category,
title=title,
summary=summary,
detailed_context=workflow_context or {},
confidence=confidence,
reasoning=reasoning,
alternatives_considered=alternatives,
impact_assessment=impact_assessment,
workflow_type=workflow_type,
deadline=deadline,
is_batch_eligible=self._is_batch_eligible(category, confidence),
)
self.db.add(approval)
await self.db.commit()
await self.db.refresh(approval)
# Create audit event
await self._create_audit_event(
approval_id=approval.id,
event_type="created",
actor_id=requesting_agent_id or UUID('00000000-0000-0000-0000-000000000000'),
actor_type="agent" if requesting_agent_id else "system",
context_snapshot=workflow_context or {}
)
# Publish event
await self.event_bus.publish(
channel=f"project:{project_id}",
event={
"type": "approval_requested",
"approval_id": str(approval.id),
"category": category.value,
"title": title,
"approver_id": str(approver_id),
}
)
# Send notification
await self._notify_approver(approval, approver_id)
return approval
async def approve(
self,
approval_id: UUID,
user_id: UUID,
rationale: str | None = None,
ip_address: str | None = None,
user_agent: str | None = None,
) -> ApprovalRequest:
"""Approve a pending request."""
approval = await self._get_pending_approval(approval_id)
# Verify user can approve
await self._verify_approver(approval, user_id)
# Update approval
approval.status = ApprovalStatus.APPROVED
approval.resolution_decision = "approved"
approval.resolution_rationale = rationale
approval.resolved_by_id = user_id
approval.resolved_at = datetime.utcnow()
approval.resolution_type = "manual"
await self.db.commit()
# Create audit event
await self._create_audit_event(
approval_id=approval.id,
event_type="approved",
actor_id=user_id,
actor_type="user",
context_snapshot={"rationale": rationale},
actor_ip=ip_address,
actor_user_agent=user_agent,
decision="approved",
rationale=rationale,
)
# Publish resolution event
await self.event_bus.publish(
channel=f"project:{approval.project_id}",
event={
"type": "approval_resolved",
"approval_id": str(approval.id),
"decision": "approved",
}
)
# Resume workflow
await self._resume_workflow(approval)
return approval
async def reject(
self,
approval_id: UUID,
user_id: UUID,
rationale: str, # Required for rejection
ip_address: str | None = None,
user_agent: str | None = None,
) -> ApprovalRequest:
"""Reject a pending request."""
approval = await self._get_pending_approval(approval_id)
# Verify user can approve
await self._verify_approver(approval, user_id)
# Update approval
approval.status = ApprovalStatus.REJECTED
approval.resolution_decision = "rejected"
approval.resolution_rationale = rationale
approval.resolved_by_id = user_id
approval.resolved_at = datetime.utcnow()
approval.resolution_type = "manual"
await self.db.commit()
# Create audit event
await self._create_audit_event(
approval_id=approval.id,
event_type="rejected",
actor_id=user_id,
actor_type="user",
context_snapshot={"rationale": rationale},
actor_ip=ip_address,
actor_user_agent=user_agent,
decision="rejected",
rationale=rationale,
)
# Notify requesting agent
await self._notify_rejection(approval, rationale)
return approval
async def batch_approve(
self,
approval_ids: list[UUID],
user_id: UUID,
rationale: str | None = None,
) -> list[ApprovalRequest]:
"""Approve multiple requests in batch."""
results = []
for approval_id in approval_ids:
try:
approval = await self.approve(
approval_id=approval_id,
user_id=user_id,
rationale=rationale or "Batch approved",
)
results.append(approval)
except Exception as e:
# Log error but continue with other approvals
logger.error(f"Failed to approve {approval_id}: {e}")
return results
async def wait_for_resolution(
self,
approval_id: UUID,
timeout_seconds: int = 86400 # 24 hours
) -> ApprovalRequest:
"""Wait for an approval to be resolved (blocking)."""
import asyncio
start_time = datetime.utcnow()
poll_interval = 5 # seconds
while True:
approval = await self._get_approval(approval_id)
if approval.status != ApprovalStatus.PENDING:
return approval
# Check timeout
elapsed = (datetime.utcnow() - start_time).total_seconds()
if elapsed > timeout_seconds:
raise TimeoutError(f"Approval {approval_id} not resolved within timeout")
await asyncio.sleep(poll_interval)
12.3 Celery Task for Timeout Processing
# app/tasks/approval_tasks.py
from celery import shared_task
from datetime import datetime, timedelta
from app.services.approval.service import ApprovalService
from app.core.database import get_db
@shared_task
def process_approval_timeouts():
"""
Periodic task to process approval timeouts and escalations.
Run every 15 minutes.
"""
with get_db() as db:
approval_service = ApprovalService(db)
# Find approvals approaching deadline
approaching_deadline = approval_service.get_approvals_approaching_deadline(
hours=4 # Reminder threshold
)
for approval in approaching_deadline:
if not approval.reminded_at:
send_reminder.delay(str(approval.id))
# Find expired approvals
expired = approval_service.get_expired_approvals()
for approval in expired:
process_escalation.delay(str(approval.id))
@shared_task
def send_reminder(approval_id: str):
"""Send reminder notification for pending approval."""
with get_db() as db:
approval_service = ApprovalService(db)
await approval_service.send_reminder(UUID(approval_id))
@shared_task
def process_escalation(approval_id: str):
"""Process escalation for expired approval."""
with get_db() as db:
approval_service = ApprovalService(db)
await approval_service.escalate_or_timeout(UUID(approval_id))
12.4 API Endpoints
# app/api/v1/approvals.py
from fastapi import APIRouter, Depends, Query
from typing import Optional
from uuid import UUID
from app.schemas.approval import (
ApprovalRequestCreate,
ApprovalRequestResponse,
ApprovalListResponse,
ApprovalDecision,
BatchApprovalRequest,
)
from app.services.approval.service import ApprovalService
from app.api.deps import get_current_user, get_db
router = APIRouter(prefix="/approvals", tags=["approvals"])
@router.get("/pending", response_model=ApprovalListResponse)
async def list_pending_approvals(
project_id: Optional[UUID] = None,
category: Optional[str] = None,
batch_eligible: bool = False,
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
current_user = Depends(get_current_user),
db = Depends(get_db),
):
"""List pending approvals for the current user."""
service = ApprovalService(db)
return await service.list_pending(
approver_id=current_user.id,
project_id=project_id,
category=category,
batch_eligible=batch_eligible,
page=page,
page_size=page_size,
)
@router.get("/{approval_id}", response_model=ApprovalRequestResponse)
async def get_approval(
approval_id: UUID,
current_user = Depends(get_current_user),
db = Depends(get_db),
):
"""Get approval details with full context."""
service = ApprovalService(db)
approval = await service.get_with_context(approval_id)
# Record view event
await service.record_view(approval_id, current_user.id)
return approval
@router.post("/{approval_id}/approve", response_model=ApprovalRequestResponse)
async def approve_request(
approval_id: UUID,
decision: ApprovalDecision,
current_user = Depends(get_current_user),
db = Depends(get_db),
request: Request = None,
):
"""Approve a pending request."""
service = ApprovalService(db)
return await service.approve(
approval_id=approval_id,
user_id=current_user.id,
rationale=decision.rationale,
ip_address=request.client.host if request else None,
user_agent=request.headers.get("user-agent") if request else None,
)
@router.post("/{approval_id}/reject", response_model=ApprovalRequestResponse)
async def reject_request(
approval_id: UUID,
decision: ApprovalDecision,
current_user = Depends(get_current_user),
db = Depends(get_db),
request: Request = None,
):
"""Reject a pending request."""
service = ApprovalService(db)
if not decision.rationale:
raise HTTPException(400, "Rationale is required for rejection")
return await service.reject(
approval_id=approval_id,
user_id=current_user.id,
rationale=decision.rationale,
ip_address=request.client.host if request else None,
user_agent=request.headers.get("user-agent") if request else None,
)
@router.post("/batch/approve", response_model=list[ApprovalRequestResponse])
async def batch_approve(
batch_request: BatchApprovalRequest,
current_user = Depends(get_current_user),
db = Depends(get_db),
):
"""Approve multiple requests in batch."""
service = ApprovalService(db)
return await service.batch_approve(
approval_ids=batch_request.approval_ids,
user_id=current_user.id,
rationale=batch_request.rationale,
)
@router.post("/{approval_id}/delegate")
async def delegate_approval(
approval_id: UUID,
delegate_to: UUID,
current_user = Depends(get_current_user),
db = Depends(get_db),
):
"""Delegate an approval to another user."""
service = ApprovalService(db)
return await service.delegate(
approval_id=approval_id,
from_user_id=current_user.id,
to_user_id=delegate_to,
)
13. UI Mockup Descriptions
13.1 Approval Dashboard (Desktop)
+-----------------------------------------------------------------------+
| SYNDARIX [User Menu] [Notifications]|
+-----------------------------------------------------------------------+
| |
| +---------------+ +------------------------------------------------+ |
| | SIDEBAR | | APPROVAL QUEUE | |
| | | +------------------------------------------------+ |
| | Dashboard | | | |
| | Projects | | Filter: [All Categories v] [All Projects v] | |
| | Agents | | | |
| | > Approvals | | +----------------------------------------------+ | |
| | - Pending | | | CRITICAL (2) SLA: 4h remaining | | |
| | - History | | +----------------------------------------------+ | |
| | Settings | | | [!] Budget Overrun - Project Alpha | | |
| | | | | $12,500 over threshold ($10,000) | | |
| | | | | Requested by: PM Agent (Alex) | | |
| | | | | [APPROVE] [REJECT] [VIEW] | | |
| | | | +----------------------------------------------+ | |
| | | | | [!] Production Deploy - Project Beta | | |
| | | | | v2.3.1 ready for production | | |
| | | | | CI: All green | Tests: 98% pass | | |
| | | | | [APPROVE] [REJECT] [VIEW] | | |
| | | | +----------------------------------------------+ | |
| | | | | |
| | | | +----------------------------------------------+ | |
| | | | | MILESTONE (3) SLA: 24h remaining | | |
| | | | +----------------------------------------------+ | |
| | | | | Sprint 4 Start - Project Alpha | | |
| | | | | Sprint 2 Complete - Project Gamma | | |
| | | | | Architecture Review - Project Delta | | |
| | | | +----------------------------------------------+ | |
| | | | | |
| | | | +----------------------------------------------+ | |
| | | | | ROUTINE (12) [Batch Approve] | | |
| | | | +----------------------------------------------+ | |
| | | | | [x] PR #45: Login validation | | |
| | | | | [x] PR #46: Password reset | | |
| | | | | [x] PR #47: Email verification | | |
| | | | | [ ] ... 9 more items | | |
| | | | +----------------------------------------------+ | |
| +---------------+ +------------------------------------------------+ |
| |
+-----------------------------------------------------------------------+
13.2 Approval Detail Panel
+-----------------------------------------------------------------------+
| < Back to Queue Approval #APR-127 |
+-----------------------------------------------------------------------+
| |
| +-------------------------------------------------------------------+ |
| | SPRINT 4 START REQUEST PENDING | |
| +-------------------------------------------------------------------+ |
| | | |
| | Project: E-Commerce Platform | |
| | Requested: 2 hours ago by PM Agent (Alex) | |
| | Deadline: 22 hours remaining | |
| | | |
| | +-----------------------------------------------------------------+ | |
| | | CONFIDENCE | | |
| | | [================> ] 87% | | |
| | +-----------------------------------------------------------------+ | |
| | | |
| | +-----------------------------------------------------------------+ | |
| | | SPRINT GOAL | | |
| | | Implement complete user authentication and authorization system | | |
| | | including OAuth2 integration and role-based access control. | | |
| | +-----------------------------------------------------------------+ | |
| | | |
| | +-----------------------------------------------------------------+ | |
| | | SPRINT DETAILS | | |
| | | Stories: 8 | Points: 21 | Duration: 2 weeks | | |
| | +-----------------------------------------------------------------+ | |
| | | |
| | [v] View Stories | |
| | [v] View Agent Reasoning | |
| | [v] View Risk Assessment | |
| | | |
| | +-----------------------------------------------------------------+ | |
| | | AGENT RECOMMENDATION | | |
| | | "This sprint has well-defined stories with clear acceptance | | |
| | | criteria. The team composition includes 3 engineers with | | |
| | | relevant auth experience. Risk is low." | | |
| | | - PM Agent (Alex) | | |
| | +-----------------------------------------------------------------+ | |
| | | |
| +-------------------------------------------------------------------+ |
| |
| +-------------------------------------------------------------------+ |
| | | |
| | Optional feedback: | |
| | +---------------------------------------------------------------+ | |
| | | | | |
| | +---------------------------------------------------------------+ | |
| | | |
| | [ REJECT ] [ APPROVE ] | |
| | | |
| +-------------------------------------------------------------------+ |
| |
+-----------------------------------------------------------------------+
13.3 Mobile Approval List
+---------------------------+
| SYNDARIX [=] [Bell] |
+---------------------------+
| |
| Pending Approvals (5) |
| |
+---------------------------+
| [!] CRITICAL |
+---------------------------+
| |
| Budget Overrun |
| Project Alpha |
| $12,500 over threshold |
| |
| 4h remaining |
| |
| [APPROVE] [REJECT] |
| |
+---------------------------+
| [!] CRITICAL |
+---------------------------+
| |
| Production Deploy |
| Project Beta - v2.3.1 |
| |
| 6h remaining |
| |
| [APPROVE] [REJECT] |
| |
+---------------------------+
| MILESTONE |
+---------------------------+
| |
| Sprint 4 Start |
| E-Commerce Platform |
| 87% confidence |
| |
| 22h remaining |
| |
| Tap for details > |
| |
+---------------------------+
14. Implementation Roadmap
Phase 1: Foundation (Week 1-2)
| Task | Priority | Effort |
|---|---|---|
| Database schema implementation | High | 2 days |
| ApprovalService core CRUD | High | 2 days |
| API endpoints | High | 2 days |
| SSE event integration | High | 1 day |
| Basic approval decorator | High | 1 day |
| Unit tests | High | 2 days |
Phase 2: Workflow Integration (Week 3-4)
| Task | Priority | Effort |
|---|---|---|
| Sprint workflow integration | High | 2 days |
| PR merge workflow integration | High | 2 days |
| Production deploy gate | High | 1 day |
| Budget threshold triggers | Medium | 1 day |
| Agent conflict resolution | Medium | 2 days |
| Confidence-based routing | Medium | 2 days |
Phase 3: UX & Notifications (Week 5-6)
| Task | Priority | Effort |
|---|---|---|
| Desktop approval dashboard | High | 3 days |
| Approval detail panel | High | 2 days |
| Batch approval UI | Medium | 2 days |
| Email notifications | High | 1 day |
| Mobile responsive design | Medium | 2 days |
| Push notification integration | Low | 2 days |
Phase 4: Advanced Features (Week 7-8)
| Task | Priority | Effort |
|---|---|---|
| Timeout/escalation processing | High | 2 days |
| Delegation rules | Medium | 2 days |
| Audit trail & reporting | Medium | 2 days |
| Metrics dashboard | Low | 2 days |
| Slack/Teams integration | Low | 2 days |
| E2E testing | High | 2 days |
References
Research Sources
- Human-in-the-Loop for AI Agents: Best Practices
- Human-in-the-Loop AI Review Queues: Workflow Patterns That Scale
- Human-in-the-Loop AI in 2025: Proven Design Patterns
- LangGraph: Human-in-the-Loop for Reliable AI Workflows
- Improving Approval Request Process - UX Case Study
- Complex Approvals - How to Design an App to Streamline Approvals
- Bulk Approval - Achieving 83% Efficiency
- Enterprise Manager's Guide to Mobile Approvals
- Notification System Design: Architecture & Best Practices
- Explainable AI: Transparent Decisions for AI Agents
- Human-Centered Explainable AI Interface Design
Internal References
- SPIKE-003: Real-time Updates Architecture
- SPIKE-001: MCP Integration Pattern
- ADR-002: Real-time Communication
- ADR-006: Agent Orchestration
- Requirements: FR-203 Autonomy Level Configuration
Decision
Adopt a checkpoint-based approval system with:
- Queue-based UI pattern with priority lanes and batch approval support
- Confidence-aware routing that escalates low-confidence AI decisions
- Configurable escalation policies per project with timeout handling
- Multi-channel notifications leveraging existing SSE infrastructure
- Full audit trail with immutable event logging
- Mobile-responsive design with push notification support
This design integrates seamlessly with Syndarix's existing autonomy level configuration (FR-203) and real-time event infrastructure (SPIKE-003).
Spike completed. Findings will inform implementation of client approval workflows.