# Phase 5: CLI Agent State Integration & Subscription Tracking

**Date:** 2026-01-15
**Status:** Planning
**Priority:** HIGH
**Estimated Effort:** 40-50 hours
**Dependencies:** Phase 4 (Architecture Improvements)

---

## Executive Summary

This phase implements native CLI agent state file integration, subscription tier tracking, automatic user interaction detection, and a dashboard API. These enhancements address critical gaps identified in production readiness:

1. **CLI State File Reading** - Read native state files from Claude Code, Codex CLI, and Gemini CLI
2. **Subscription Tier Tracking** - Track account types (Claude Max, ChatGPT Plus, Gemini Pro)
3. **User Interaction Detection** - Detect when CLI agents are waiting for user input
4. **Auto-Response Handler** - Orchestrator can handle or escalate interaction requests
5. **Dashboard API** - REST endpoints for usage visualization and monitoring

**Key Insight:** The [Agent Sessions](https://github.com/jazzyalex/agent-sessions) project demonstrates that reading native CLI state files is the most reliable method for tracking rate limits and session states, rather than inferring from orchestrator-level tracking.

---

## Table of Contents

1. [Motivation](#motivation)
2. [Architecture Overview](#architecture-overview)
3. [Phase 5.1: CLI State File Reader](#phase-51-cli-state-file-reader)
4. [Phase 5.2: Subscription Tier Model](#phase-52-subscription-tier-model)
5. [Phase 5.3: User Interaction Detection](#phase-53-user-interaction-detection)
6. [Phase 5.4: Auto-Response Handler](#phase-54-auto-response-handler)
7. [Phase 5.5: Dashboard API](#phase-55-dashboard-api)
8. [Configuration Schema](#configuration-schema)
9. [Testing Strategy](#testing-strategy)
10. [Reference Projects](#reference-projects)
11. [Success Criteria](#success-criteria)

---

## Motivation

### Current Gaps

| Gap | Impact | Current Workaround |
|-----|--------|-------------------|
| No visibility into CLI agent rate limits | Agents hit limits unexpectedly | Hardcoded estimates |
| No subscription tier awareness | Can't optimize for account capabilities | Manual configuration |
| Can't detect "waiting for input" states | Orchestrator blind to CLI prompts | Manual monitoring |
| No way to auto-respond to CLI prompts | Requires constant human attention | N/A |
| No web dashboard or API | CLI-only status display | Terminal access required |

### Business Value

- **Reduced Manual Intervention:** Auto-respond to low-risk CLI prompts
- **Better Resource Utilization:** Avoid hitting rate limits unexpectedly
- **Improved Visibility:** Dashboard for usage monitoring
- **Cost Optimization:** Track actual usage vs. subscription limits

---

## Architecture Overview

```
┌─────────────────────────────────────────────────────────────────┐
│                    Orchestration Layer                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────────┐    ┌──────────────────┐    ┌───────────┐  │
│  │ CLIStateReader  │───▶│ InteractionRouter │───▶│ Dashboard │  │
│  │                 │    │                   │    │    API    │  │
│  └────────┬────────┘    └────────┬──────────┘    └───────────┘  │
│           │                      │                              │
│           ▼                      ▼                              │
│  ┌─────────────────┐    ┌──────────────────┐                   │
│  │ SubscriptionMgr │    │ AutoResponseHdlr │                   │
│  │                 │    │                  │                   │
│  └─────────────────┘    └──────────────────┘                   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘
         │                        │
         ▼                        ▼
┌─────────────────┐      ┌─────────────────┐
│  Native State   │      │    Existing     │
│     Files       │      │   Interrupt     │
│                 │      │   Handlers      │
│ ~/.claude/      │      │                 │
│ ~/.codex/       │      │ CLIInterrupt    │
│ ~/.gemini/      │      │ AsyncInterrupt  │
└─────────────────┘      └─────────────────┘
```

### New Modules

```
src/agent_orchestrator/
├── tracking/                    # NEW: CLI State Tracking
│   ├── __init__.py
│   ├── cli_state_reader.py      # Read native CLI state files
│   ├── state_models.py          # State dataclasses per CLI
│   └── rate_limit_monitor.py    # Real-time limit monitoring
│
├── subscriptions/               # NEW: Subscription Management
│   ├── __init__.py
│   ├── tiers.py                 # SubscriptionTier enum + limits
│   ├── manager.py               # SubscriptionManager class
│   └── detection.py             # Auto-detect subscription tier
│
├── interaction/                 # NEW: Interaction Handling
│   ├── __init__.py
│   ├── detector.py              # Detect CLI waiting states
│   ├── router.py                # Route interactions
│   └── auto_handler.py          # Auto-respond to low-risk
│
└── api/                         # NEW: Dashboard API
    ├── __init__.py
    ├── app.py                   # FastAPI application
    ├── routes/
    │   ├── agents.py            # Agent status endpoints
    │   ├── usage.py             # Usage/limits endpoints
    │   ├── costs.py             # Cost tracking endpoints
    │   └── health.py            # Health check endpoints
    └── models.py                # Pydantic response models
```

---

## Phase 5.1: CLI State File Reader

**Goal:** Read native CLI agent state files to get accurate rate limits and session states.

**Priority:** HIGH
**Effort:** 10-12 hours

### State File Locations

| CLI Agent | State Directory | Key Files |
|-----------|-----------------|-----------|
| Claude Code | `~/.claude/` | `sessions/`, `config.json` |
| Codex CLI | `~/.codex/` | `sessions/`, `state.json` |
| Gemini CLI | `~/.gemini/` | `tmp/`, `config.yaml` |
| Copilot CLI | `~/.copilot/` | `session-state/` |
| OpenCode | `~/.local/share/opencode/` | `storage/session` |

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 5.1.1 | Define `CLIStateReader` base class | `tracking/cli_state_reader.py` |
| 5.1.2 | Implement `ClaudeStateReader` | `tracking/cli_state_reader.py` |
| 5.1.3 | Implement `CodexStateReader` | `tracking/cli_state_reader.py` |
| 5.1.4 | Implement `GeminiStateReader` | `tracking/cli_state_reader.py` |
| 5.1.5 | Define state models per CLI | `tracking/state_models.py` |
| 5.1.6 | Implement `RateLimitMonitor` | `tracking/rate_limit_monitor.py` |
| 5.1.7 | Add file watcher for real-time updates | `tracking/cli_state_reader.py` |
| 5.1.8 | Write unit tests | `tests/unit/test_cli_state_reader.py` |

### Implementation

```python
# src/agent_orchestrator/tracking/cli_state_reader.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Optional
import json
import os

@dataclass
class RateLimitState:
    """Current rate limit state for a CLI agent."""
    requests_used: int
    requests_limit: int
    reset_at: datetime
    window_type: str  # "5_hour", "daily", "weekly"
    percentage_used: float

    @property
    def is_exhausted(self) -> bool:
        return self.requests_used >= self.requests_limit

    @property
    def remaining(self) -> int:
        return max(0, self.requests_limit - self.requests_used)

@dataclass
class SessionState:
    """Current session state for a CLI agent."""
    session_id: str
    started_at: datetime
    last_activity_at: datetime
    is_active: bool
    waiting_for_input: bool
    input_prompt: Optional[str] = None
    current_task: Optional[str] = None

class CLIStateReader(ABC):
    """Base class for reading CLI agent state files."""

    def __init__(self, state_dir: Path):
        self.state_dir = Path(state_dir).expanduser()

    @abstractmethod
    def get_rate_limit_state(self) -> Optional[RateLimitState]:
        """Get current rate limit state."""
        pass

    @abstractmethod
    def get_session_state(self) -> Optional[SessionState]:
        """Get current session state."""
        pass

    @abstractmethod
    def is_waiting_for_input(self) -> bool:
        """Check if CLI is waiting for user input."""
        pass

    def exists(self) -> bool:
        """Check if state directory exists."""
        return self.state_dir.exists()

class ClaudeStateReader(CLIStateReader):
    """Read Claude Code state files from ~/.claude/"""

    def __init__(self):
        super().__init__(Path("~/.claude"))
        self.sessions_dir = self.state_dir / "sessions"

    def get_rate_limit_state(self) -> Optional[RateLimitState]:
        """Parse Claude Code rate limits from session files."""
        if not self.sessions_dir.exists():
            return None

        # Find most recent session file
        session_files = list(self.sessions_dir.glob("*.json"))
        if not session_files:
            return None

        latest = max(session_files, key=lambda f: f.stat().st_mtime)

        try:
            with open(latest) as f:
                data = json.load(f)

            # Extract rate limit info (structure varies by version)
            usage = data.get("usage", {})
            limits = data.get("limits", {})

            return RateLimitState(
                requests_used=usage.get("messages_sent", 0),
                requests_limit=limits.get("messages_per_5h", 225),  # Opus default
                reset_at=datetime.fromisoformat(usage.get("reset_at", datetime.now().isoformat())),
                window_type="5_hour",
                percentage_used=usage.get("messages_sent", 0) / limits.get("messages_per_5h", 225) * 100
            )
        except (json.JSONDecodeError, KeyError, FileNotFoundError):
            return None

    def get_session_state(self) -> Optional[SessionState]:
        """Parse Claude Code session state."""
        if not self.sessions_dir.exists():
            return None

        session_files = list(self.sessions_dir.glob("*.json"))
        if not session_files:
            return None

        latest = max(session_files, key=lambda f: f.stat().st_mtime)

        try:
            with open(latest) as f:
                data = json.load(f)

            return SessionState(
                session_id=data.get("session_id", latest.stem),
                started_at=datetime.fromisoformat(data.get("started_at", datetime.now().isoformat())),
                last_activity_at=datetime.fromtimestamp(latest.stat().st_mtime),
                is_active=data.get("is_active", False),
                waiting_for_input=data.get("waiting_for_input", False),
                input_prompt=data.get("pending_prompt"),
                current_task=data.get("current_task")
            )
        except (json.JSONDecodeError, KeyError, FileNotFoundError):
            return None

    def is_waiting_for_input(self) -> bool:
        """Check if Claude Code is waiting for user input."""
        state = self.get_session_state()
        return state.waiting_for_input if state else False


class CodexStateReader(CLIStateReader):
    """Read Codex CLI state files from ~/.codex/"""

    def __init__(self):
        super().__init__(Path("~/.codex"))
        self.sessions_dir = self.state_dir / "sessions"

    def get_rate_limit_state(self) -> Optional[RateLimitState]:
        """Parse Codex CLI rate limits."""
        # Similar implementation for Codex
        pass

    def get_session_state(self) -> Optional[SessionState]:
        """Parse Codex session state."""
        pass

    def is_waiting_for_input(self) -> bool:
        """Check if Codex is waiting for user input."""
        pass


class GeminiStateReader(CLIStateReader):
    """Read Gemini CLI state files from ~/.gemini/"""

    def __init__(self):
        super().__init__(Path("~/.gemini"))
        self.tmp_dir = self.state_dir / "tmp"

    def get_rate_limit_state(self) -> Optional[RateLimitState]:
        """Parse Gemini CLI rate limits."""
        pass

    def get_session_state(self) -> Optional[SessionState]:
        """Parse Gemini session state."""
        pass

    def is_waiting_for_input(self) -> bool:
        """Check if Gemini is waiting for user input."""
        pass
```

### Rate Limit Monitor

```python
# src/agent_orchestrator/tracking/rate_limit_monitor.py
import asyncio
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, Optional, Callable
from .cli_state_reader import CLIStateReader, RateLimitState, ClaudeStateReader, CodexStateReader, GeminiStateReader

@dataclass
class RateLimitAlert:
    """Alert when rate limits are approaching."""
    agent_id: str
    current_percentage: float
    threshold: str  # "warning", "critical", "exhausted"
    reset_at: datetime
    message: str

class RateLimitMonitor:
    """Monitor rate limits across all CLI agents."""

    WARNING_THRESHOLD = 75.0
    CRITICAL_THRESHOLD = 90.0

    def __init__(self):
        self.readers: Dict[str, CLIStateReader] = {
            "claude_code": ClaudeStateReader(),
            "codex_cli": CodexStateReader(),
            "gemini_cli": GeminiStateReader(),
        }
        self._callbacks: list[Callable[[RateLimitAlert], None]] = []
        self._running = False

    def register_alert_callback(self, callback: Callable[[RateLimitAlert], None]):
        """Register callback for rate limit alerts."""
        self._callbacks.append(callback)

    def get_all_states(self) -> Dict[str, Optional[RateLimitState]]:
        """Get rate limit states for all agents."""
        return {
            agent_id: reader.get_rate_limit_state()
            for agent_id, reader in self.readers.items()
            if reader.exists()
        }

    def get_usage_summary(self) -> Dict[str, dict]:
        """Get usage summary for dashboard display."""
        summary = {}
        for agent_id, state in self.get_all_states().items():
            if state:
                summary[agent_id] = {
                    "used": state.requests_used,
                    "limit": state.requests_limit,
                    "remaining": state.remaining,
                    "percentage": state.percentage_used,
                    "reset_at": state.reset_at.isoformat(),
                    "window": state.window_type,
                    "status": self._get_status(state.percentage_used),
                }
        return summary

    def _get_status(self, percentage: float) -> str:
        """Get status string from percentage."""
        if percentage >= 100:
            return "exhausted"
        elif percentage >= self.CRITICAL_THRESHOLD:
            return "critical"
        elif percentage >= self.WARNING_THRESHOLD:
            return "warning"
        return "healthy"

    async def start_monitoring(self, interval_seconds: int = 60):
        """Start background monitoring loop."""
        self._running = True
        while self._running:
            await self._check_all_limits()
            await asyncio.sleep(interval_seconds)

    def stop_monitoring(self):
        """Stop the monitoring loop."""
        self._running = False

    async def _check_all_limits(self):
        """Check all limits and fire alerts if needed."""
        for agent_id, state in self.get_all_states().items():
            if not state:
                continue

            alert = None
            if state.is_exhausted:
                alert = RateLimitAlert(
                    agent_id=agent_id,
                    current_percentage=state.percentage_used,
                    threshold="exhausted",
                    reset_at=state.reset_at,
                    message=f"{agent_id} rate limit exhausted. Resets at {state.reset_at}"
                )
            elif state.percentage_used >= self.CRITICAL_THRESHOLD:
                alert = RateLimitAlert(
                    agent_id=agent_id,
                    current_percentage=state.percentage_used,
                    threshold="critical",
                    reset_at=state.reset_at,
                    message=f"{agent_id} at {state.percentage_used:.1f}% of rate limit"
                )
            elif state.percentage_used >= self.WARNING_THRESHOLD:
                alert = RateLimitAlert(
                    agent_id=agent_id,
                    current_percentage=state.percentage_used,
                    threshold="warning",
                    reset_at=state.reset_at,
                    message=f"{agent_id} at {state.percentage_used:.1f}% of rate limit"
                )

            if alert:
                for callback in self._callbacks:
                    callback(alert)
```

---

## Phase 5.2: Subscription Tier Model

**Goal:** Track and manage subscription tiers for each CLI agent.

**Priority:** HIGH
**Effort:** 6-8 hours

### Subscription Tiers

| Provider | Tier | Rate Limits | Features |
|----------|------|-------------|----------|
| Claude | Free | 5 msg/5h | Basic |
| Claude | Pro | 45 msg/5h (Opus) | Extended context |
| Claude | Max | 225 msg/5h (Opus) | Highest limits |
| ChatGPT | Free | 10 msg/3h | GPT-4o mini |
| ChatGPT | Plus | 80 msg/3h | GPT-4o, o1 |
| Gemini | Free | 50 msg/day | Basic |
| Gemini | Pro | Unlimited | 1M context |

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 5.2.1 | Define `SubscriptionTier` enum | `subscriptions/tiers.py` |
| 5.2.2 | Define `TierLimits` dataclass | `subscriptions/tiers.py` |
| 5.2.3 | Define `TIER_CONFIGURATIONS` | `subscriptions/tiers.py` |
| 5.2.4 | Implement `SubscriptionManager` | `subscriptions/manager.py` |
| 5.2.5 | Implement tier auto-detection | `subscriptions/detection.py` |
| 5.2.6 | Add database persistence | `persistence/models.py` |
| 5.2.7 | Write unit tests | `tests/unit/test_subscriptions.py` |

### Implementation

```python
# src/agent_orchestrator/subscriptions/tiers.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class Provider(Enum):
    """Supported AI providers."""
    ANTHROPIC = "anthropic"
    OPENAI = "openai"
    GOOGLE = "google"

class SubscriptionTier(Enum):
    """Subscription tier levels."""
    # Claude tiers
    CLAUDE_FREE = "claude_free"
    CLAUDE_PRO = "claude_pro"
    CLAUDE_MAX = "claude_max"

    # ChatGPT tiers
    CHATGPT_FREE = "chatgpt_free"
    CHATGPT_PLUS = "chatgpt_plus"
    CHATGPT_PRO = "chatgpt_pro"

    # Gemini tiers
    GEMINI_FREE = "gemini_free"
    GEMINI_PRO = "gemini_pro"
    GEMINI_ULTRA = "gemini_ultra"

@dataclass
class TierLimits:
    """Rate limits for a subscription tier."""
    messages_per_window: int
    window_hours: float
    daily_messages: Optional[int] = None
    weekly_messages: Optional[int] = None
    context_window: int = 128000
    supports_streaming: bool = True
    supports_vision: bool = True
    supports_tools: bool = True
    max_output_tokens: int = 4096

@dataclass
class SubscriptionConfig:
    """Full subscription configuration."""
    tier: SubscriptionTier
    provider: Provider
    limits: TierLimits
    autonomy_level: str  # "auto", "assisted", "manual"
    auto_response_enabled: bool = False

# Provider-specific tier configurations
TIER_CONFIGURATIONS = {
    # Claude tiers
    SubscriptionTier.CLAUDE_FREE: TierLimits(
        messages_per_window=5,
        window_hours=5,
        context_window=128000,
    ),
    SubscriptionTier.CLAUDE_PRO: TierLimits(
        messages_per_window=45,
        window_hours=5,
        context_window=200000,
    ),
    SubscriptionTier.CLAUDE_MAX: TierLimits(
        messages_per_window=225,
        window_hours=5,
        context_window=200000,
        max_output_tokens=16384,
    ),

    # ChatGPT tiers
    SubscriptionTier.CHATGPT_FREE: TierLimits(
        messages_per_window=10,
        window_hours=3,
        context_window=8192,
    ),
    SubscriptionTier.CHATGPT_PLUS: TierLimits(
        messages_per_window=80,
        window_hours=3,
        context_window=128000,
    ),
    SubscriptionTier.CHATGPT_PRO: TierLimits(
        messages_per_window=200,
        window_hours=3,
        context_window=128000,
        max_output_tokens=16384,
    ),

    # Gemini tiers
    SubscriptionTier.GEMINI_FREE: TierLimits(
        messages_per_window=50,
        window_hours=24,
        daily_messages=50,
        context_window=128000,
    ),
    SubscriptionTier.GEMINI_PRO: TierLimits(
        messages_per_window=1000,
        window_hours=24,
        daily_messages=1000,
        context_window=1000000,  # 1M context
    ),
}
```

### Subscription Manager

```python
# src/agent_orchestrator/subscriptions/manager.py
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, Optional
from .tiers import SubscriptionTier, TierLimits, SubscriptionConfig, Provider, TIER_CONFIGURATIONS
from ..persistence.database import OrchestratorDB

@dataclass
class AgentSubscription:
    """Subscription info for an agent."""
    agent_id: str
    tier: SubscriptionTier
    provider: Provider
    limits: TierLimits
    autonomy_level: str
    detected_at: datetime
    manually_set: bool = False

class SubscriptionManager:
    """Manage subscription tiers for all agents."""

    def __init__(self, db: OrchestratorDB):
        self.db = db
        self._subscriptions: Dict[str, AgentSubscription] = {}

    def set_subscription(
        self,
        agent_id: str,
        tier: SubscriptionTier,
        autonomy_level: str = "assisted"
    ):
        """Manually set subscription tier for an agent."""
        provider = self._get_provider(tier)
        limits = TIER_CONFIGURATIONS.get(tier)

        if not limits:
            raise ValueError(f"Unknown tier: {tier}")

        subscription = AgentSubscription(
            agent_id=agent_id,
            tier=tier,
            provider=provider,
            limits=limits,
            autonomy_level=autonomy_level,
            detected_at=datetime.now(),
            manually_set=True,
        )

        self._subscriptions[agent_id] = subscription
        self._persist_subscription(subscription)

    def get_subscription(self, agent_id: str) -> Optional[AgentSubscription]:
        """Get subscription for an agent."""
        return self._subscriptions.get(agent_id)

    def get_limits(self, agent_id: str) -> Optional[TierLimits]:
        """Get rate limits for an agent."""
        sub = self.get_subscription(agent_id)
        return sub.limits if sub else None

    def get_autonomy_level(self, agent_id: str) -> str:
        """Get autonomy level for an agent."""
        sub = self.get_subscription(agent_id)
        return sub.autonomy_level if sub else "manual"

    def _get_provider(self, tier: SubscriptionTier) -> Provider:
        """Determine provider from tier."""
        tier_name = tier.value
        if tier_name.startswith("claude"):
            return Provider.ANTHROPIC
        elif tier_name.startswith("chatgpt"):
            return Provider.OPENAI
        elif tier_name.startswith("gemini"):
            return Provider.GOOGLE
        raise ValueError(f"Unknown provider for tier: {tier}")

    def _persist_subscription(self, subscription: AgentSubscription):
        """Save subscription to database."""
        # Implementation: store in subscriptions table
        pass

    def load_subscriptions(self):
        """Load subscriptions from database."""
        # Implementation: load from subscriptions table
        pass
```

---

## Phase 5.3: User Interaction Detection

**Goal:** Detect when CLI agents are waiting for user input and categorize the interaction type.

**Priority:** HIGH
**Effort:** 8-10 hours

### Interaction Types

| Type | Risk Level | Auto-Response Eligible |
|------|------------|------------------------|
| `confirm_file_read` | LOW | Yes |
| `confirm_directory_list` | LOW | Yes |
| `confirm_file_write` | MEDIUM | Configurable |
| `confirm_command_exec` | MEDIUM | Configurable |
| `confirm_git_operation` | HIGH | No |
| `confirm_file_delete` | HIGH | No |
| `confirm_external_api` | HIGH | No |
| `request_clarification` | MEDIUM | No |
| `authentication_required` | CRITICAL | No |

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 5.3.1 | Define `InteractionType` enum | `interaction/detector.py` |
| 5.3.2 | Define `InteractionRequest` dataclass | `interaction/detector.py` |
| 5.3.3 | Implement `InteractionDetector` class | `interaction/detector.py` |
| 5.3.4 | Add pattern matching for each CLI | `interaction/detector.py` |
| 5.3.5 | Implement `InteractionRouter` | `interaction/router.py` |
| 5.3.6 | Integrate with health monitoring | `control/health.py` |
| 5.3.7 | Write unit tests | `tests/unit/test_interaction_detector.py` |

### Implementation

```python
# src/agent_orchestrator/interaction/detector.py
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional, List
import re

class InteractionType(Enum):
    """Types of user interactions CLI agents may request."""
    # Low risk - auto-response eligible
    CONFIRM_FILE_READ = "confirm_file_read"
    CONFIRM_DIRECTORY_LIST = "confirm_directory_list"
    CONFIRM_GLOB_SEARCH = "confirm_glob_search"

    # Medium risk - configurable
    CONFIRM_FILE_WRITE = "confirm_file_write"
    CONFIRM_FILE_EDIT = "confirm_file_edit"
    CONFIRM_COMMAND_EXEC = "confirm_command_exec"
    CONFIRM_PACKAGE_INSTALL = "confirm_package_install"

    # High risk - always escalate
    CONFIRM_GIT_PUSH = "confirm_git_push"
    CONFIRM_GIT_FORCE = "confirm_git_force"
    CONFIRM_FILE_DELETE = "confirm_file_delete"
    CONFIRM_EXTERNAL_API = "confirm_external_api"
    CONFIRM_SENSITIVE_DATA = "confirm_sensitive_data"

    # Special cases
    REQUEST_CLARIFICATION = "request_clarification"
    AUTHENTICATION_REQUIRED = "authentication_required"
    UNKNOWN = "unknown"

@dataclass
class InteractionRequest:
    """A detected interaction request from a CLI agent."""
    agent_id: str
    interaction_type: InteractionType
    prompt_text: str
    target: Optional[str] = None  # file path, command, etc.
    options: List[str] = None  # available response options
    detected_at: datetime = None
    context: dict = None

    def __post_init__(self):
        if self.detected_at is None:
            self.detected_at = datetime.now()
        if self.options is None:
            self.options = []
        if self.context is None:
            self.context = {}

    @property
    def risk_level(self) -> str:
        """Get risk level for this interaction."""
        low_risk = {
            InteractionType.CONFIRM_FILE_READ,
            InteractionType.CONFIRM_DIRECTORY_LIST,
            InteractionType.CONFIRM_GLOB_SEARCH,
        }
        high_risk = {
            InteractionType.CONFIRM_GIT_PUSH,
            InteractionType.CONFIRM_GIT_FORCE,
            InteractionType.CONFIRM_FILE_DELETE,
            InteractionType.CONFIRM_EXTERNAL_API,
            InteractionType.CONFIRM_SENSITIVE_DATA,
            InteractionType.AUTHENTICATION_REQUIRED,
        }

        if self.interaction_type in low_risk:
            return "low"
        elif self.interaction_type in high_risk:
            return "high"
        return "medium"

class InteractionDetector:
    """Detect and classify user interaction requests from CLI agents."""

    # Claude Code patterns
    CLAUDE_PATTERNS = {
        r"(?:Allow|Approve)\s+reading?\s+(?:file|from)\s+['\"]?([^'\"]+)['\"]?": InteractionType.CONFIRM_FILE_READ,
        r"(?:Allow|Approve)\s+(?:writing?|editing?)\s+(?:to|file)\s+['\"]?([^'\"]+)['\"]?": InteractionType.CONFIRM_FILE_WRITE,
        r"(?:Allow|Approve)\s+(?:running?|executing?)\s+(?:command:?)?\s*['\"]?(.+)['\"]?": InteractionType.CONFIRM_COMMAND_EXEC,
        r"(?:Allow|Approve)\s+(?:deleting?|removing?)\s+['\"]?([^'\"]+)['\"]?": InteractionType.CONFIRM_FILE_DELETE,
        r"(?:Allow|Approve)\s+git\s+push": InteractionType.CONFIRM_GIT_PUSH,
        r"(?:Allow|Approve)\s+git\s+.*--force": InteractionType.CONFIRM_GIT_FORCE,
    }

    # Codex patterns
    CODEX_PATTERNS = {
        r"Continue\?.*\[y/N\]": InteractionType.CONFIRM_COMMAND_EXEC,
        r"This will (?:create|modify|delete)": InteractionType.CONFIRM_FILE_WRITE,
    }

    # Gemini patterns
    GEMINI_PATTERNS = {
        r"Do you want to proceed\?": InteractionType.CONFIRM_COMMAND_EXEC,
        r"Please confirm": InteractionType.REQUEST_CLARIFICATION,
    }

    def __init__(self):
        self._agent_patterns = {
            "claude_code": self.CLAUDE_PATTERNS,
            "codex_cli": self.CODEX_PATTERNS,
            "gemini_cli": self.GEMINI_PATTERNS,
        }

    def detect(self, agent_id: str, output: str) -> Optional[InteractionRequest]:
        """
        Detect if output contains an interaction request.

        Args:
            agent_id: The agent producing the output
            output: The CLI output text to analyze

        Returns:
            InteractionRequest if detected, None otherwise
        """
        patterns = self._agent_patterns.get(agent_id, {})

        for pattern, interaction_type in patterns.items():
            match = re.search(pattern, output, re.IGNORECASE)
            if match:
                target = match.group(1) if match.groups() else None
                return InteractionRequest(
                    agent_id=agent_id,
                    interaction_type=interaction_type,
                    prompt_text=output,
                    target=target,
                    options=self._extract_options(output),
                )

        # Check for generic waiting patterns
        if self._is_waiting_for_input(output):
            return InteractionRequest(
                agent_id=agent_id,
                interaction_type=InteractionType.UNKNOWN,
                prompt_text=output,
            )

        return None

    def _extract_options(self, output: str) -> List[str]:
        """Extract available response options from prompt."""
        # Common patterns: [y/n], [Y/n], (yes/no), etc.
        patterns = [
            r'\[([yYnN])/([yYnN])\]',
            r'\(([a-z]+)/([a-z]+)\)',
            r'\[([a-zA-Z]+)\]',
        ]

        options = []
        for pattern in patterns:
            match = re.search(pattern, output)
            if match:
                options.extend(match.groups())

        return list(set(options))

    def _is_waiting_for_input(self, output: str) -> bool:
        """Check if output indicates waiting for input."""
        waiting_indicators = [
            r'\?\s*$',  # Ends with question mark
            r'\[y/n\]',
            r'\(yes/no\)',
            r'press enter',
            r'continue\?',
            r'proceed\?',
        ]

        for indicator in waiting_indicators:
            if re.search(indicator, output, re.IGNORECASE):
                return True
        return False
```

---

## Phase 5.4: Auto-Response Handler

**Goal:** Automatically respond to low-risk CLI prompts based on configuration.

**Priority:** MEDIUM
**Effort:** 8-10 hours

### Auto-Response Policies

```yaml
# Example configuration
auto_response:
  enabled: true
  default_policy: escalate  # auto, escalate

  rules:
    - interaction_type: confirm_file_read
      policy: auto
      response: "y"

    - interaction_type: confirm_directory_list
      policy: auto
      response: "y"

    - interaction_type: confirm_file_write
      policy: auto
      conditions:
        - path_pattern: "src/**/*.py"
        - path_pattern: "tests/**/*.py"
      response: "y"

    - interaction_type: confirm_command_exec
      policy: escalate  # Always ask human

    - interaction_type: confirm_git_push
      policy: escalate

    - interaction_type: confirm_file_delete
      policy: deny  # Never auto-approve
      response: "n"
```

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 5.4.1 | Define `AutoResponsePolicy` | `interaction/auto_handler.py` |
| 5.4.2 | Define `ResponseRule` dataclass | `interaction/auto_handler.py` |
| 5.4.3 | Implement `AutoResponseHandler` | `interaction/auto_handler.py` |
| 5.4.4 | Implement rule matching logic | `interaction/auto_handler.py` |
| 5.4.5 | Implement response injection | `interaction/auto_handler.py` |
| 5.4.6 | Integrate with InteractionRouter | `interaction/router.py` |
| 5.4.7 | Add configuration loading | `config.py` |
| 5.4.8 | Write unit tests | `tests/unit/test_auto_handler.py` |
| 5.4.9 | Write integration tests | `tests/integration/test_auto_response.py` |

### Implementation

```python
# src/agent_orchestrator/interaction/auto_handler.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional, List
import fnmatch
import re

from .detector import InteractionRequest, InteractionType
from ..subscriptions.manager import SubscriptionManager

class ResponsePolicy(Enum):
    """How to handle an interaction."""
    AUTO = "auto"          # Automatically respond
    ESCALATE = "escalate"  # Ask human
    DENY = "deny"          # Automatically deny

@dataclass
class ResponseRule:
    """Rule for handling an interaction type."""
    interaction_type: InteractionType
    policy: ResponsePolicy
    response: Optional[str] = None  # Response to send if AUTO
    conditions: Optional[List[dict]] = None  # Additional conditions

    def matches(self, request: InteractionRequest) -> bool:
        """Check if this rule matches the request."""
        if request.interaction_type != self.interaction_type:
            return False

        if not self.conditions:
            return True

        # Check all conditions
        for condition in self.conditions:
            if "path_pattern" in condition and request.target:
                if not fnmatch.fnmatch(request.target, condition["path_pattern"]):
                    return False
            if "command_pattern" in condition and request.target:
                if not re.match(condition["command_pattern"], request.target):
                    return False

        return True

@dataclass
class AutoResponseResult:
    """Result of auto-response evaluation."""
    should_respond: bool
    response: Optional[str] = None
    policy: ResponsePolicy = ResponsePolicy.ESCALATE
    matched_rule: Optional[ResponseRule] = None
    reason: str = ""

class AutoResponseHandler:
    """Handle automatic responses to CLI agent interactions."""

    # Default rules for common interactions
    DEFAULT_RULES = [
        # Low risk - auto-approve
        ResponseRule(
            interaction_type=InteractionType.CONFIRM_FILE_READ,
            policy=ResponsePolicy.AUTO,
            response="y"
        ),
        ResponseRule(
            interaction_type=InteractionType.CONFIRM_DIRECTORY_LIST,
            policy=ResponsePolicy.AUTO,
            response="y"
        ),
        ResponseRule(
            interaction_type=InteractionType.CONFIRM_GLOB_SEARCH,
            policy=ResponsePolicy.AUTO,
            response="y"
        ),

        # High risk - auto-deny
        ResponseRule(
            interaction_type=InteractionType.CONFIRM_GIT_FORCE,
            policy=ResponsePolicy.DENY,
            response="n"
        ),

        # Everything else - escalate
        ResponseRule(
            interaction_type=InteractionType.CONFIRM_FILE_DELETE,
            policy=ResponsePolicy.ESCALATE,
        ),
        ResponseRule(
            interaction_type=InteractionType.CONFIRM_GIT_PUSH,
            policy=ResponsePolicy.ESCALATE,
        ),
    ]

    def __init__(
        self,
        subscription_manager: SubscriptionManager,
        rules: Optional[List[ResponseRule]] = None,
    ):
        self.subscription_manager = subscription_manager
        self.rules = rules or self.DEFAULT_RULES

    def evaluate(self, request: InteractionRequest) -> AutoResponseResult:
        """
        Evaluate an interaction request and determine response.

        Args:
            request: The interaction request to evaluate

        Returns:
            AutoResponseResult with decision and response
        """
        # Check agent's autonomy level
        autonomy = self.subscription_manager.get_autonomy_level(request.agent_id)

        if autonomy == "manual":
            return AutoResponseResult(
                should_respond=False,
                policy=ResponsePolicy.ESCALATE,
                reason="Agent configured for manual mode"
            )

        # Find matching rule
        for rule in self.rules:
            if rule.matches(request):
                if rule.policy == ResponsePolicy.AUTO:
                    return AutoResponseResult(
                        should_respond=True,
                        response=rule.response,
                        policy=rule.policy,
                        matched_rule=rule,
                        reason=f"Auto-approved: {request.interaction_type.value}"
                    )
                elif rule.policy == ResponsePolicy.DENY:
                    return AutoResponseResult(
                        should_respond=True,
                        response=rule.response,
                        policy=rule.policy,
                        matched_rule=rule,
                        reason=f"Auto-denied: {request.interaction_type.value}"
                    )
                else:  # ESCALATE
                    return AutoResponseResult(
                        should_respond=False,
                        policy=rule.policy,
                        matched_rule=rule,
                        reason=f"Escalating: {request.interaction_type.value}"
                    )

        # No rule matched - default to escalate
        return AutoResponseResult(
            should_respond=False,
            policy=ResponsePolicy.ESCALATE,
            reason="No matching rule - escalating to human"
        )

    def add_rule(self, rule: ResponseRule):
        """Add a response rule."""
        self.rules.insert(0, rule)  # Higher priority

    def remove_rule(self, interaction_type: InteractionType):
        """Remove rules for an interaction type."""
        self.rules = [r for r in self.rules if r.interaction_type != interaction_type]
```

---

## Phase 5.5: Dashboard API

**Goal:** Provide REST API endpoints for monitoring and visualization.

**Priority:** MEDIUM
**Effort:** 10-12 hours

### API Endpoints

| Endpoint | Method | Description |
|----------|--------|-------------|
| `/api/health` | GET | System health check |
| `/api/agents` | GET | List all agents with status |
| `/api/agents/{id}` | GET | Get agent details |
| `/api/agents/{id}/usage` | GET | Get agent usage stats |
| `/api/usage/summary` | GET | Usage summary across all agents |
| `/api/usage/limits` | GET | Rate limit status per agent |
| `/api/costs` | GET | Cost tracking summary |
| `/api/costs/daily` | GET | Daily cost breakdown |
| `/api/tasks` | GET | List recent tasks |
| `/api/tasks/{id}` | GET | Get task details |
| `/api/interactions` | GET | Pending interactions |
| `/api/interactions/{id}/respond` | POST | Respond to interaction |

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 5.5.1 | Create FastAPI application | `api/app.py` |
| 5.5.2 | Define Pydantic response models | `api/models.py` |
| 5.5.3 | Implement health endpoints | `api/routes/health.py` |
| 5.5.4 | Implement agents endpoints | `api/routes/agents.py` |
| 5.5.5 | Implement usage endpoints | `api/routes/usage.py` |
| 5.5.6 | Implement costs endpoints | `api/routes/costs.py` |
| 5.5.7 | Implement tasks endpoints | `api/routes/tasks.py` |
| 5.5.8 | Implement interactions endpoints | `api/routes/interactions.py` |
| 5.5.9 | Add WebSocket for real-time updates | `api/websocket.py` |
| 5.5.10 | Write API tests | `tests/api/test_endpoints.py` |

### Implementation

```python
# src/agent_orchestrator/api/app.py
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Optional
from datetime import datetime

from .routes import agents, usage, costs, health, tasks, interactions
from ..orchestrator.brain import OrchestrationBrain

def create_app(brain: OrchestrationBrain) -> FastAPI:
    """Create FastAPI application with all routes."""

    app = FastAPI(
        title="Agent Orchestrator API",
        description="API for monitoring and controlling the agent orchestration system",
        version="1.0.0",
    )

    # CORS middleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],  # Configure for production
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Store brain reference
    app.state.brain = brain

    # Include routers
    app.include_router(health.router, prefix="/api", tags=["Health"])
    app.include_router(agents.router, prefix="/api", tags=["Agents"])
    app.include_router(usage.router, prefix="/api", tags=["Usage"])
    app.include_router(costs.router, prefix="/api", tags=["Costs"])
    app.include_router(tasks.router, prefix="/api", tags=["Tasks"])
    app.include_router(interactions.router, prefix="/api", tags=["Interactions"])

    return app
```

```python
# src/agent_orchestrator/api/routes/usage.py
from fastapi import APIRouter, Depends, HTTPException
from typing import Dict, List
from datetime import datetime, timedelta

from ..models import UsageSummary, AgentUsage, RateLimitStatus
from ...tracking.rate_limit_monitor import RateLimitMonitor

router = APIRouter()

@router.get("/usage/summary", response_model=UsageSummary)
async def get_usage_summary():
    """Get usage summary across all agents."""
    monitor = RateLimitMonitor()
    states = monitor.get_usage_summary()

    return UsageSummary(
        timestamp=datetime.now(),
        agents=states,
        total_requests_today=sum(s.get("used", 0) for s in states.values()),
    )

@router.get("/usage/limits", response_model=Dict[str, RateLimitStatus])
async def get_rate_limits():
    """Get rate limit status for all agents."""
    monitor = RateLimitMonitor()
    return monitor.get_usage_summary()

@router.get("/agents/{agent_id}/usage", response_model=AgentUsage)
async def get_agent_usage(agent_id: str):
    """Get detailed usage for a specific agent."""
    monitor = RateLimitMonitor()
    states = monitor.get_all_states()

    if agent_id not in states:
        raise HTTPException(status_code=404, detail=f"Agent {agent_id} not found")

    state = states[agent_id]
    if not state:
        raise HTTPException(status_code=404, detail=f"No usage data for {agent_id}")

    return AgentUsage(
        agent_id=agent_id,
        requests_used=state.requests_used,
        requests_limit=state.requests_limit,
        percentage_used=state.percentage_used,
        reset_at=state.reset_at,
        window_type=state.window_type,
    )
```

```python
# src/agent_orchestrator/api/models.py
from pydantic import BaseModel
from typing import Dict, List, Optional
from datetime import datetime
from enum import Enum

class AgentStatus(str, Enum):
    AVAILABLE = "available"
    BUSY = "busy"
    EXHAUSTED = "exhausted"
    ERROR = "error"
    OFFLINE = "offline"

class AgentInfo(BaseModel):
    """Agent information."""
    agent_id: str
    agent_type: str
    status: AgentStatus
    subscription_tier: Optional[str] = None
    current_task: Optional[str] = None
    last_activity: Optional[datetime] = None

class RateLimitStatus(BaseModel):
    """Rate limit status for an agent."""
    used: int
    limit: int
    remaining: int
    percentage: float
    reset_at: datetime
    window: str
    status: str  # "healthy", "warning", "critical", "exhausted"

class UsageSummary(BaseModel):
    """Usage summary across agents."""
    timestamp: datetime
    agents: Dict[str, RateLimitStatus]
    total_requests_today: int

class AgentUsage(BaseModel):
    """Detailed usage for an agent."""
    agent_id: str
    requests_used: int
    requests_limit: int
    percentage_used: float
    reset_at: datetime
    window_type: str

class CostSummary(BaseModel):
    """Cost tracking summary."""
    total_cost_today: float
    total_cost_month: float
    by_agent: Dict[str, float]
    by_model: Dict[str, float]

class InteractionPending(BaseModel):
    """Pending interaction request."""
    id: str
    agent_id: str
    interaction_type: str
    prompt_text: str
    target: Optional[str]
    risk_level: str
    detected_at: datetime
    options: List[str]
```

---

## Configuration Schema

```yaml
# config/orchestrator.yaml

# Phase 5 Configuration
cli_state_tracking:
  enabled: true
  poll_interval_seconds: 30

  # State file locations (defaults)
  state_directories:
    claude_code: "~/.claude"
    codex_cli: "~/.codex"
    gemini_cli: "~/.gemini"

subscriptions:
  # Per-agent subscription configuration
  agents:
    claude_code:
      tier: claude_max
      autonomy_level: assisted

    codex_cli:
      tier: chatgpt_plus
      autonomy_level: manual

    gemini_cli:
      tier: gemini_pro
      autonomy_level: auto

auto_response:
  enabled: true
  default_policy: escalate

  # Custom rules (override defaults)
  rules:
    - interaction_type: confirm_file_write
      policy: auto
      conditions:
        - path_pattern: "src/**/*.py"
        - path_pattern: "tests/**/*.py"

    - interaction_type: confirm_command_exec
      policy: auto
      conditions:
        - command_pattern: "^pytest.*"
        - command_pattern: "^npm test.*"

rate_limit_alerts:
  enabled: true
  warning_threshold: 75
  critical_threshold: 90

  notifications:
    - type: slack
      webhook_url: "${SLACK_WEBHOOK_URL}"
    - type: email
      recipients: ["admin@example.com"]

dashboard_api:
  enabled: true
  host: "0.0.0.0"
  port: 8080
  cors_origins: ["http://localhost:3000"]
```

---

## Testing Strategy

### Unit Tests

| Test File | Coverage |
|-----------|----------|
| `test_cli_state_reader.py` | State file parsing |
| `test_subscriptions.py` | Tier management |
| `test_interaction_detector.py` | Pattern matching |
| `test_auto_handler.py` | Response rules |
| `test_api_models.py` | Pydantic models |

### Integration Tests

| Test File | Coverage |
|-----------|----------|
| `test_rate_limit_monitor.py` | End-to-end monitoring |
| `test_auto_response_flow.py` | Detection → Response flow |
| `test_api_endpoints.py` | All API routes |

### Mock Data

```python
# tests/fixtures/cli_state_fixtures.py

MOCK_CLAUDE_SESSION = {
    "session_id": "test-session-123",
    "started_at": "2026-01-15T10:00:00Z",
    "is_active": True,
    "waiting_for_input": True,
    "pending_prompt": "Allow reading file src/main.py?",
    "usage": {
        "messages_sent": 150,
        "reset_at": "2026-01-15T15:00:00Z"
    },
    "limits": {
        "messages_per_5h": 225
    }
}

MOCK_CODEX_SESSION = {
    "session_id": "codex-456",
    "active": True,
    "requests_this_hour": 45,
    "limit_per_hour": 80
}
```

---

## Reference Projects

| Project | Key Learning | Relevance |
|---------|--------------|-----------|
| [Agent Sessions](https://github.com/jazzyalex/agent-sessions) | Native state file reading | HIGH - Primary reference |
| [AgentOps](https://github.com/AgentOps-AI/agentops) | Usage monitoring dashboard | MEDIUM - Dashboard patterns |
| [AgentOS](https://github.com/saadnvd1/agent-os) | Web UI for multi-agent | MEDIUM - API patterns |
| [Claude Code Bridge](https://github.com/bfly123/claude_code_bridge) | Cross-AI coordination | LOW - Daemon patterns |

---

## Success Criteria

| Metric | Target | Verification |
|--------|--------|--------------|
| CLI state reading accuracy | >95% | Unit tests against mock files |
| Rate limit detection latency | <5 seconds | Integration tests |
| Auto-response accuracy | 100% for low-risk | E2E tests |
| API response time | <100ms p95 | Load tests |
| Dashboard availability | 99.9% | Health checks |
| Test coverage | >80% | Coverage reports |

---

## Implementation Checklist

### Phase 5.1: CLI State File Reader
- [ ] Define `CLIStateReader` base class
- [ ] Implement `ClaudeStateReader`
- [ ] Implement `CodexStateReader`
- [ ] Implement `GeminiStateReader`
- [ ] Define state models
- [ ] Implement `RateLimitMonitor`
- [ ] Add file watcher
- [ ] Write unit tests

### Phase 5.2: Subscription Tier Model
- [ ] Define `SubscriptionTier` enum
- [ ] Define `TierLimits` dataclass
- [ ] Define tier configurations
- [ ] Implement `SubscriptionManager`
- [ ] Implement tier auto-detection
- [ ] Add database persistence
- [ ] Write unit tests

### Phase 5.3: User Interaction Detection
- [ ] Define `InteractionType` enum
- [ ] Define `InteractionRequest` dataclass
- [ ] Implement `InteractionDetector`
- [ ] Add pattern matching per CLI
- [ ] Implement `InteractionRouter`
- [ ] Integrate with health monitoring
- [ ] Write unit tests

### Phase 5.4: Auto-Response Handler
- [ ] Define `AutoResponsePolicy`
- [ ] Define `ResponseRule` dataclass
- [ ] Implement `AutoResponseHandler`
- [ ] Implement rule matching
- [ ] Implement response injection
- [ ] Integrate with router
- [ ] Add configuration loading
- [ ] Write unit and integration tests

### Phase 5.5: Dashboard API
- [ ] Create FastAPI application
- [ ] Define Pydantic models
- [ ] Implement health endpoints
- [ ] Implement agents endpoints
- [ ] Implement usage endpoints
- [ ] Implement costs endpoints
- [ ] Implement tasks endpoints
- [ ] Implement interactions endpoints
- [ ] Add WebSocket support
- [ ] Write API tests

---

## Notes

- This phase can run parallel to Phase 4 (Architecture Improvements)
- All implementations should be feature-flagged initially
- Native state file reading is read-only - no modifications
- API should support authentication for production use
- Consider rate limiting the API itself

---

*Document created: January 15, 2026*
*Status: Planning - Ready for implementation*
