# 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 1. **Adopt checkpoint-based approval pattern** aligned with Syndarix's three autonomy levels 2. **Use confidence-based routing** to automatically escalate uncertain AI decisions 3. **Implement batch approval UI** for FULL_CONTROL mode efficiency 4. **Leverage existing SSE infrastructure** for real-time approval notifications 5. **Support mobile-responsive approval interface** with push notification integration 6. **Design flexible timeout and escalation policies** per project configuration --- ## Table of Contents 1. [Research Questions](#1-research-questions) 2. [Syndarix Autonomy Levels](#2-syndarix-autonomy-levels) 3. [Approval UX Patterns](#3-approval-ux-patterns) 4. [Presenting AI Decisions for Review](#4-presenting-ai-decisions-for-review) 5. [Timeout and Escalation Handling](#5-timeout-and-escalation-handling) 6. [Delegation Patterns](#6-delegation-patterns) 7. [Batch vs Individual Approval](#7-batch-vs-individual-approval) 8. [Mobile-Friendly Interface](#8-mobile-friendly-interface) 9. [Audit Trail Design](#9-audit-trail-design) 10. [Notification System Integration](#10-notification-system-integration) 11. [Database Schema](#11-database-schema) 12. [Code Examples](#12-code-examples) 13. [UI Mockup Descriptions](#13-ui-mockup-descriptions) 14. [Implementation Roadmap](#14-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 ```python 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 1. **Status Visibility**: Clear indicators showing approved, pending, rejected states with color coding 2. **Context Preservation**: Full decision context without requiring navigation away 3. **One-Touch Actions**: Approve/Reject buttons immediately visible 4. **Batch Operations**: Select multiple items for bulk approval 5. **SLA Tracking**: Visual indicators for time-sensitive decisions 6. **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 ```python 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 ```python @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 ```python 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 ```python @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 ```python 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 ```python 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 1. **Progressive Disclosure**: Show minimal info, expand on tap 2. **Swipe Actions**: Swipe right to approve, left to reject 3. **Push Notifications**: Actionable notifications with quick responses 4. **Offline Queue**: Cache pending approvals for offline review 5. **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 ```python @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 ```python 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 | | Email | 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 ```python 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 ```python # 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 ```python # 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 ```sql -- 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 ```python # 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 ```python # 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 ```python # 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 ```python # 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](https://www.permit.io/blog/human-in-the-loop-for-ai-agents-best-practices-frameworks-use-cases-and-demo) - [Human-in-the-Loop AI Review Queues: Workflow Patterns That Scale](https://alldaystech.com/guides/artificial-intelligence/human-in-the-loop-ai-review-queue-workflows) - [Human-in-the-Loop AI in 2025: Proven Design Patterns](https://blog.ideafloats.com/human-in-the-loop-ai-in-2025/) - [LangGraph: Human-in-the-Loop for Reliable AI Workflows](https://medium.com/@sitabjapal03/langgraph-part-4-human-in-the-loop-for-reliable-ai-workflows-aa4cc175bce4) - [Improving Approval Request Process - UX Case Study](https://bootcamp.uxdesign.cc/improving-the-approval-request-process-on-an-enterprise-application-a-ux-case-study-12d2756af876) - [Complex Approvals - How to Design an App to Streamline Approvals](https://www.uxpin.com/studio/blog/complex-approvals-app-design/) - [Bulk Approval - Achieving 83% Efficiency](https://sreeja-ux-ui-portfolio.webflow.io/projects/bulk-approval) - [Enterprise Manager's Guide to Mobile Approvals](https://propelapps.com/blog/mobile-approvals-solutions/enterprise-manager-guide-to-mobile-approvals/) - [Notification System Design: Architecture & Best Practices](https://www.magicbell.com/blog/notification-system-design) - [Explainable AI: Transparent Decisions for AI Agents](https://www.rapidinnovation.io/post/for-developers-implementing-explainable-ai-for-transparent-agent-decisions) - [Human-Centered Explainable AI Interface Design](https://arxiv.org/html/2403.14496v1) ### Internal References - [SPIKE-003: Real-time Updates Architecture](./SPIKE-003-realtime-updates.md) - [SPIKE-001: MCP Integration Pattern](./SPIKE-001-mcp-integration-pattern.md) - [ADR-002: Real-time Communication](../adrs/ADR-002-realtime-communication.md) - [ADR-006: Agent Orchestration](../adrs/ADR-006-agent-orchestration.md) - [Requirements: FR-203 Autonomy Level Configuration](../requirements/SYNDARIX_REQUIREMENTS.md) --- ## Decision **Adopt a checkpoint-based approval system** with: 1. **Queue-based UI pattern** with priority lanes and batch approval support 2. **Confidence-aware routing** that escalates low-confidence AI decisions 3. **Configurable escalation policies** per project with timeout handling 4. **Multi-channel notifications** leveraging existing SSE infrastructure 5. **Full audit trail** with immutable event logging 6. **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.*