# Phase 6: Advanced Orchestration & User Experience Enhancements

**Date:** 2026-01-15
**Status:** Research Complete - Planning
**Priority:** MEDIUM
**Estimated Effort:** 80-100 hours
**Dependencies:** Phase 5 (CLI State Integration)

---

## Executive Summary

This phase focuses on advanced orchestration patterns, improved user experience, and enhanced monitoring capabilities based on research into leading GitHub projects and industry trends. Key areas include:

1. **Swarm Intelligence Patterns** - Parallel execution, task decomposition, hierarchical coordination
2. **Terminal UI (TUI) Dashboard** - Rich terminal interface for monitoring and control
3. **Observability Integration** - LangSmith/AgentOps-style session tracing
4. **Agent Coordination Patterns** - Handoffs, voting, consensus mechanisms
5. **Cost Optimization & Alerts** - GitHub-style budget tracking with alerts

---

## Table of Contents

1. [Research Summary](#research-summary)
2. [Phase 6.1: Swarm Intelligence Patterns](#phase-61-swarm-intelligence-patterns)
3. [Phase 6.2: Terminal UI Dashboard](#phase-62-terminal-ui-dashboard)
4. [Phase 6.3: Session Tracing & Replay](#phase-63-session-tracing--replay)
5. [Phase 6.4: Advanced Agent Coordination](#phase-64-advanced-agent-coordination)
6. [Phase 6.5: Cost Optimization System](#phase-65-cost-optimization-system)
7. [Phase 6.6: Plugin Marketplace Architecture](#phase-66-plugin-marketplace-architecture)
8. [Reference Projects](#reference-projects)
9. [Implementation Priority](#implementation-priority)

---

## Research Summary

### Key GitHub Projects Analyzed

| Project | Stars | Key Features | Relevance |
|---------|-------|--------------|-----------|
| [Swarms](https://github.com/kyegomez/swarms) | 3k+ | Concurrent workflows, hierarchical swarms, GroupChat | HIGH |
| [Claude Squad](https://github.com/smtg-ai/claude-squad) | 2k+ | TUI interface, session management, git worktree isolation | HIGH |
| [wshobson/agents](https://github.com/wshobson/agents) | 1k+ | 100 agents, 110 skills, plugin architecture | HIGH |
| [Claude Flow](https://github.com/ruvnet/claude-flow) | 5k+ | Swarm intelligence, MCP integration, workflow orchestration | HIGH |
| [AgentOps](https://github.com/AgentOps-AI/agentops) | 3k+ | 2-line integration, session replay, cost tracking | MEDIUM |
| [Toad](https://willmcgugan.github.io/announcing-toad/) | New | Universal TUI for AI agents (by Rich/Textual creator) | HIGH |
| [DeepResearchAgent](https://github.com/SkyworkAI/DeepResearchAgent) | 1k+ | Hierarchical task decomposition, planning agent | MEDIUM |
| [Agent-MCP](https://github.com/rinadelph/Agent-MCP) | 500+ | MCP-based coordination, shared memory | MEDIUM |

### Industry Trends (2025-2026)

1. **GitHub Agent HQ** - Universal agent orchestration across GitHub, VS Code, CLI
2. **VS Code 1.107** - Multi-agent orchestration with background agents
3. **LangSmith** - Zero-latency tracing, nested span visualization
4. **Textual/Rich** - 5-10x faster TUI rendering, 60+ FPS dashboards
5. **GitHub Budget Tracking** - Product-specific and bundled budgets with alerts

---

## Phase 6.1: Swarm Intelligence Patterns

**Goal:** Implement advanced multi-agent coordination patterns for parallel execution and task decomposition.

**Priority:** HIGH
**Effort:** 20-25 hours
**Reference:** [Swarms Framework](https://github.com/kyegomez/swarms), [DeepResearchAgent](https://github.com/SkyworkAI/DeepResearchAgent)

### Features to Implement

#### 6.1.1 Concurrent Workflow Engine

```python
# src/agent_orchestrator/swarm/concurrent.py
from dataclasses import dataclass
from typing import List, Dict, Any, Callable
import asyncio

@dataclass
class ConcurrentTask:
    """Task for concurrent execution."""
    task_id: str
    agent_id: str
    prompt: str
    dependencies: List[str] = None  # Task IDs this depends on
    timeout_seconds: int = 300

class ConcurrentWorkflow:
    """
    Execute multiple agents simultaneously.

    Inspired by Swarms ConcurrentWorkflow:
    "Agents run tasks simultaneously for maximum efficiency"
    """

    def __init__(self, max_concurrent: int = 5):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results: Dict[str, Any] = {}

    async def execute(self, tasks: List[ConcurrentTask]) -> Dict[str, Any]:
        """Execute tasks concurrently respecting dependencies."""
        # Build dependency graph
        pending = {t.task_id: t for t in tasks}
        completed = set()

        while pending:
            # Find tasks with satisfied dependencies
            ready = [
                t for t in pending.values()
                if not t.dependencies or all(d in completed for d in t.dependencies)
            ]

            if not ready:
                raise ValueError("Circular dependency detected")

            # Execute ready tasks concurrently
            results = await asyncio.gather(*[
                self._execute_task(t) for t in ready
            ])

            # Update state
            for task, result in zip(ready, results):
                self.results[task.task_id] = result
                completed.add(task.task_id)
                del pending[task.task_id]

        return self.results

    async def _execute_task(self, task: ConcurrentTask) -> Any:
        """Execute single task with semaphore limiting."""
        async with self.semaphore:
            # Get agent and execute
            pass
```

#### 6.1.2 Hierarchical Swarm Coordinator

```python
# src/agent_orchestrator/swarm/hierarchical.py
class HierarchicalSwarm:
    """
    Director agent creates plans and distributes tasks to workers.

    Inspired by Swarms HierarchicalSwarm:
    "Director who creates plans and distributes specific tasks"
    """

    def __init__(
        self,
        director_agent_id: str,
        worker_agent_ids: List[str],
    ):
        self.director = director_agent_id
        self.workers = worker_agent_ids

    async def execute(self, goal: str) -> SwarmResult:
        """Execute goal using hierarchical coordination."""
        # 1. Director creates plan
        plan = await self._create_plan(goal)

        # 2. Decompose into subtasks
        subtasks = await self._decompose_plan(plan)

        # 3. Assign to workers
        assignments = self._assign_tasks(subtasks)

        # 4. Execute in parallel
        results = await self._execute_parallel(assignments)

        # 5. Director synthesizes results
        synthesis = await self._synthesize_results(results)

        return synthesis
```

#### 6.1.3 Auto Task Decomposition

```python
# src/agent_orchestrator/swarm/decomposer.py
class TaskDecomposer:
    """
    Automatically break complex tasks into subtasks.

    Inspired by DeepResearchAgent:
    "Planning agent breaks down tasks into manageable sub-tasks"
    """

    DECOMPOSITION_PROMPT = """
    Analyze this task and break it into atomic subtasks:

    Task: {task}

    Requirements:
    1. Each subtask should be completable by a single agent
    2. Identify dependencies between subtasks
    3. Estimate complexity (low/medium/high)
    4. Suggest best agent type for each subtask

    Output JSON:
    {
        "subtasks": [
            {
                "id": "task_1",
                "description": "...",
                "dependencies": [],
                "complexity": "low",
                "suggested_agent": "code_writer"
            }
        ]
    }
    """

    async def decompose(self, task: str) -> List[Subtask]:
        """Decompose complex task into subtasks."""
        pass
```

#### 6.1.4 GroupChat Pattern

```python
# src/agent_orchestrator/swarm/groupchat.py
class GroupChat:
    """
    Conversational environment for multi-agent collaboration.

    Inspired by Swarms GroupChat:
    "Agents can interact, discuss, and collaboratively solve"
    """

    def __init__(
        self,
        agents: List[str],
        speaking_order: str = "round_robin",  # or "dynamic", "priority"
        max_rounds: int = 10,
    ):
        self.agents = agents
        self.speaking_order = speaking_order
        self.max_rounds = max_rounds
        self.conversation_history: List[Message] = []

    async def discuss(self, topic: str) -> DiscussionResult:
        """Facilitate multi-agent discussion on topic."""
        self.conversation_history.append(Message(
            role="system",
            content=f"Discussion topic: {topic}"
        ))

        for round_num in range(self.max_rounds):
            speaker = self._get_next_speaker(round_num)

            # Get agent's response considering full history
            response = await self._get_agent_response(
                speaker,
                self.conversation_history
            )

            self.conversation_history.append(Message(
                role=speaker,
                content=response
            ))

            # Check for consensus
            if self._has_consensus():
                break

        return self._synthesize_discussion()
```

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 6.1.1 | Implement `ConcurrentWorkflow` | `swarm/concurrent.py` |
| 6.1.2 | Implement `HierarchicalSwarm` | `swarm/hierarchical.py` |
| 6.1.3 | Implement `TaskDecomposer` | `swarm/decomposer.py` |
| 6.1.4 | Implement `GroupChat` | `swarm/groupchat.py` |
| 6.1.5 | Add `AgentRearrange` for non-linear workflows | `swarm/rearrange.py` |
| 6.1.6 | Implement `MixtureOfAgents` pattern | `swarm/mixture.py` |
| 6.1.7 | Write comprehensive tests | `tests/unit/test_swarm.py` |

---

## Phase 6.2: Terminal UI Dashboard

**Goal:** Build a rich terminal user interface for monitoring and controlling agents.

**Priority:** HIGH
**Effort:** 25-30 hours
**Reference:** [Claude Squad](https://github.com/smtg-ai/claude-squad), [Textual](https://textual.textualize.io/), [Toad](https://willmcgugan.github.io/announcing-toad/)

### Features to Implement

#### 6.2.1 Main Dashboard Layout

```
┌─────────────────────────────────────────────────────────────────────────────┐
│  Agent Orchestrator Dashboard                              [H]elp [Q]uit    │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  AGENTS                          │  ACTIVE TASK                             │
│  ──────────────────────────────  │  ────────────────────────────────────── │
│  🟢 claude_code    [45%] ██░░░   │  Task: Implement user auth              │
│  🟢 gemini_cli     [12%] █░░░░   │  Agent: claude_code                     │
│  🟡 codex_cli      [78%] ████░   │  Status: Running (3m 24s)               │
│  🔴 openai_agents  [EXHAUST]     │  Tokens: 12,450 / 50,000                │
│                                  │                                          │
│  ──────────────────────────────  │  Output:                                 │
│  [n]ew [p]ause [r]esume [k]ill   │  > Creating auth middleware...          │
│                                  │  > Writing tests for login...           │
│                                  │  > ✓ 3 files modified                   │
│                                                                             │
├─────────────────────────────────────────────────────────────────────────────┤
│  TASK QUEUE (5 pending)                                                     │
│  ───────────────────────────────────────────────────────────────────────── │
│  │ # │ Task                              │ Agent      │ Status    │ Wait  │ │
│  │ 1 │ Add API rate limiting             │ claude_code│ Queued    │ 2m    │ │
│  │ 2 │ Write unit tests for auth         │ -          │ Unassigned│ 5m    │ │
│  │ 3 │ Update documentation              │ gemini_cli │ Queued    │ 1m    │ │
│  └───┴───────────────────────────────────┴────────────┴───────────┴───────┘ │
│                                                                             │
├─────────────────────────────────────────────────────────────────────────────┤
│  COST TODAY: $12.45 │ TOKENS: 245,000 │ ALERTS: 2 warnings                  │
└─────────────────────────────────────────────────────────────────────────────┘
```

#### 6.2.2 Textual Implementation

```python
# src/agent_orchestrator/tui/dashboard.py
from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical
from textual.widgets import Header, Footer, Static, DataTable, ProgressBar
from textual.reactive import reactive
from rich.panel import Panel
from rich.table import Table

class AgentStatusWidget(Static):
    """Display agent status with usage bar."""

    def __init__(self, agent_id: str, **kwargs):
        super().__init__(**kwargs)
        self.agent_id = agent_id

    def compose(self) -> ComposeResult:
        yield Static(id="status-emoji")
        yield Static(id="agent-name")
        yield ProgressBar(id="usage-bar", total=100)
        yield Static(id="usage-percent")

    def update_status(self, status: dict):
        """Update widget with new status."""
        emoji = {"available": "🟢", "busy": "🟡", "exhausted": "🔴"}.get(
            status["status"], "⚪"
        )
        self.query_one("#status-emoji").update(emoji)
        self.query_one("#agent-name").update(self.agent_id)
        self.query_one("#usage-bar").progress = status["percentage"]
        self.query_one("#usage-percent").update(f"[{status['percentage']:.0f}%]")

class TaskQueueWidget(Static):
    """Display task queue as data table."""

    def compose(self) -> ComposeResult:
        table = DataTable(id="task-table")
        table.add_columns("#", "Task", "Agent", "Status", "Wait")
        yield table

    def update_tasks(self, tasks: List[dict]):
        """Update task table."""
        table = self.query_one("#task-table")
        table.clear()
        for i, task in enumerate(tasks, 1):
            table.add_row(
                str(i),
                task["description"][:40],
                task.get("agent_id", "-"),
                task["status"],
                task.get("wait_time", "-"),
            )

class OrchestratorDashboard(App):
    """Main TUI dashboard application."""

    CSS = """
    AgentStatusWidget {
        height: 3;
        margin: 1;
    }

    #agents-panel {
        width: 40%;
        border: solid green;
    }

    #task-panel {
        width: 60%;
        border: solid blue;
    }

    #footer-stats {
        dock: bottom;
        height: 3;
        background: $surface;
    }
    """

    BINDINGS = [
        ("q", "quit", "Quit"),
        ("n", "new_task", "New Task"),
        ("p", "pause_agent", "Pause"),
        ("r", "resume_agent", "Resume"),
        ("h", "help", "Help"),
    ]

    def compose(self) -> ComposeResult:
        yield Header()
        with Horizontal():
            with Vertical(id="agents-panel"):
                yield Static("AGENTS", classes="panel-title")
                # Agent widgets added dynamically
            with Vertical(id="task-panel"):
                yield Static("ACTIVE TASK", classes="panel-title")
                yield TaskQueueWidget()
        yield Footer()

    def on_mount(self):
        """Start background refresh."""
        self.set_interval(1.0, self.refresh_status)

    async def refresh_status(self):
        """Refresh all status displays."""
        # Fetch from orchestrator and update widgets
        pass
```

#### 6.2.3 Key Bindings (Claude Squad-inspired)

| Key | Action | Description |
|-----|--------|-------------|
| `j`/`k` or `↑`/`↓` | Navigate | Move between agents/tasks |
| `n` | New Task | Create new task |
| `N` | New with Prompt | Create task with initial prompt |
| `p` | Pause | Pause selected agent |
| `r` | Resume | Resume paused agent |
| `D` | Delete | Kill selected session |
| `↵`/`o` | Attach | Attach to agent session |
| `c` | Commit | Commit agent's changes |
| `d` | Diff | Show changes diff |
| `l` | Logs | View agent logs |
| `?` | Help | Show help menu |
| `q` | Quit | Exit dashboard |

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 6.2.1 | Create base Textual app structure | `tui/app.py` |
| 6.2.2 | Implement `AgentStatusWidget` | `tui/widgets/agent_status.py` |
| 6.2.3 | Implement `TaskQueueWidget` | `tui/widgets/task_queue.py` |
| 6.2.4 | Implement `ActiveTaskWidget` | `tui/widgets/active_task.py` |
| 6.2.5 | Implement `CostSummaryWidget` | `tui/widgets/cost_summary.py` |
| 6.2.6 | Implement `LogViewerWidget` | `tui/widgets/log_viewer.py` |
| 6.2.7 | Implement `DiffViewerWidget` | `tui/widgets/diff_viewer.py` |
| 6.2.8 | Add vim-style key bindings | `tui/bindings.py` |
| 6.2.9 | Add real-time WebSocket updates | `tui/realtime.py` |
| 6.2.10 | Write TUI tests | `tests/tui/test_dashboard.py` |

---

## Phase 6.3: Session Tracing & Replay

**Goal:** Implement comprehensive session tracing with replay capabilities.

**Priority:** MEDIUM
**Effort:** 15-20 hours
**Reference:** [AgentOps](https://github.com/AgentOps-AI/agentops), [LangSmith](https://www.langchain.com/langsmith/observability)

### Features to Implement

#### 6.3.1 Trace Data Model

```python
# src/agent_orchestrator/tracing/models.py
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Dict, Any, Optional
from enum import Enum

class SpanKind(Enum):
    """Type of trace span."""
    TASK = "task"
    AGENT = "agent"
    LLM_CALL = "llm_call"
    TOOL_CALL = "tool_call"
    APPROVAL = "approval"
    MEMORY_READ = "memory_read"
    MEMORY_WRITE = "memory_write"

@dataclass
class Span:
    """A single span in a trace."""
    span_id: str
    parent_span_id: Optional[str]
    trace_id: str
    name: str
    kind: SpanKind
    start_time: datetime
    end_time: Optional[datetime] = None
    status: str = "running"  # running, success, error

    # Metrics
    input_tokens: int = 0
    output_tokens: int = 0
    cost_usd: float = 0.0
    latency_ms: float = 0.0

    # Context
    input_data: Dict[str, Any] = field(default_factory=dict)
    output_data: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)
    error: Optional[str] = None

    # Nested spans
    children: List["Span"] = field(default_factory=list)

@dataclass
class Trace:
    """Complete trace of a task execution."""
    trace_id: str
    task_id: str
    agent_id: str
    start_time: datetime
    end_time: Optional[datetime] = None
    status: str = "running"

    # Aggregated metrics
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    total_latency_ms: float = 0.0

    # Root span
    root_span: Optional[Span] = None

    # Tags for filtering
    tags: Dict[str, str] = field(default_factory=dict)
```

#### 6.3.2 Tracer Implementation

```python
# src/agent_orchestrator/tracing/tracer.py
from contextlib import asynccontextmanager
import uuid

class Tracer:
    """
    Trace agent execution with nested spans.

    Inspired by AgentOps:
    "Session replays in 2 lines of code"
    """

    def __init__(self, storage: TraceStorage):
        self.storage = storage
        self._current_trace: Optional[Trace] = None
        self._span_stack: List[Span] = []

    def start_trace(self, task_id: str, agent_id: str, **metadata) -> Trace:
        """Start a new trace for a task."""
        trace = Trace(
            trace_id=str(uuid.uuid4()),
            task_id=task_id,
            agent_id=agent_id,
            start_time=datetime.now(),
            tags=metadata,
        )
        self._current_trace = trace
        return trace

    def end_trace(self, status: str = "success"):
        """End the current trace."""
        if self._current_trace:
            self._current_trace.end_time = datetime.now()
            self._current_trace.status = status
            self._aggregate_metrics()
            self.storage.save_trace(self._current_trace)
            self._current_trace = None

    @asynccontextmanager
    async def span(self, name: str, kind: SpanKind, **metadata):
        """Create a span context manager."""
        span = Span(
            span_id=str(uuid.uuid4()),
            parent_span_id=self._span_stack[-1].span_id if self._span_stack else None,
            trace_id=self._current_trace.trace_id,
            name=name,
            kind=kind,
            start_time=datetime.now(),
            metadata=metadata,
        )

        # Add to parent or root
        if self._span_stack:
            self._span_stack[-1].children.append(span)
        else:
            self._current_trace.root_span = span

        self._span_stack.append(span)

        try:
            yield span
            span.status = "success"
        except Exception as e:
            span.status = "error"
            span.error = str(e)
            raise
        finally:
            span.end_time = datetime.now()
            span.latency_ms = (span.end_time - span.start_time).total_seconds() * 1000
            self._span_stack.pop()
```

#### 6.3.3 Session Replay API

```python
# src/agent_orchestrator/tracing/replay.py
class SessionReplay:
    """
    Replay recorded sessions step-by-step.

    Inspired by AgentOps:
    "Session replays showing step-by-step execution graphs"
    """

    def __init__(self, storage: TraceStorage):
        self.storage = storage

    def get_trace(self, trace_id: str) -> Trace:
        """Get a trace by ID."""
        return self.storage.get_trace(trace_id)

    def get_execution_timeline(self, trace_id: str) -> List[TimelineEvent]:
        """Get chronological timeline of events."""
        trace = self.get_trace(trace_id)
        events = []

        def flatten_spans(span: Span, depth: int = 0):
            events.append(TimelineEvent(
                timestamp=span.start_time,
                event_type="span_start",
                span=span,
                depth=depth,
            ))
            for child in span.children:
                flatten_spans(child, depth + 1)
            events.append(TimelineEvent(
                timestamp=span.end_time,
                event_type="span_end",
                span=span,
                depth=depth,
            ))

        if trace.root_span:
            flatten_spans(trace.root_span)

        return sorted(events, key=lambda e: e.timestamp)

    def get_decision_points(self, trace_id: str) -> List[DecisionPoint]:
        """Get all decision points where agent made choices."""
        trace = self.get_trace(trace_id)
        decisions = []

        def find_decisions(span: Span):
            if span.kind in {SpanKind.TOOL_CALL, SpanKind.APPROVAL}:
                decisions.append(DecisionPoint(
                    span=span,
                    input=span.input_data,
                    output=span.output_data,
                    rationale=span.metadata.get("rationale"),
                ))
            for child in span.children:
                find_decisions(child)

        if trace.root_span:
            find_decisions(trace.root_span)

        return decisions
```

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 6.3.1 | Define trace data models | `tracing/models.py` |
| 6.3.2 | Implement `Tracer` class | `tracing/tracer.py` |
| 6.3.3 | Implement `TraceStorage` (SQLite) | `tracing/storage.py` |
| 6.3.4 | Implement `SessionReplay` | `tracing/replay.py` |
| 6.3.5 | Add trace visualization API | `api/routes/traces.py` |
| 6.3.6 | Integrate tracer with control loop | `control/loop.py` |
| 6.3.7 | Write tests | `tests/unit/test_tracing.py` |

---

## Phase 6.4: Advanced Agent Coordination

**Goal:** Implement sophisticated coordination patterns for multi-agent workflows.

**Priority:** MEDIUM
**Effort:** 15-20 hours
**Reference:** [Agent-MCP](https://github.com/rinadelph/Agent-MCP), [Microsoft Multi-Agent Reference Architecture](https://microsoft.github.io/multi-agent-reference-architecture/)

### Features to Implement

#### 6.4.1 Agent Handoff Protocol

```python
# src/agent_orchestrator/coordination/handoff.py
@dataclass
class HandoffContext:
    """Context transferred during agent handoff."""
    from_agent: str
    to_agent: str
    task_summary: str
    completed_work: List[str]
    remaining_work: List[str]
    shared_state: Dict[str, Any]
    files_modified: List[str]
    important_decisions: List[str]
    warnings: List[str]

class HandoffManager:
    """
    Manage context handoffs between agents.

    Inspired by Microsoft Reference Architecture:
    "Handoff tools enable task handover and memory transfer"
    """

    async def prepare_handoff(
        self,
        from_agent: str,
        to_agent: str,
        reason: str,
    ) -> HandoffContext:
        """Prepare context for handoff."""
        # Gather state from source agent
        working_memory = await self.memory.get_working_memory(from_agent)

        # Generate summary of work done
        summary = await self._generate_handoff_summary(
            from_agent,
            working_memory,
        )

        return HandoffContext(
            from_agent=from_agent,
            to_agent=to_agent,
            task_summary=summary.task_summary,
            completed_work=summary.completed_items,
            remaining_work=summary.remaining_items,
            shared_state=working_memory.state,
            files_modified=working_memory.files_modified,
            important_decisions=summary.decisions,
            warnings=summary.warnings,
        )

    async def execute_handoff(self, context: HandoffContext):
        """Execute the handoff to target agent."""
        # Inject context into target agent's prompt
        handoff_prompt = self._format_handoff_prompt(context)

        # Start target agent with context
        await self.orchestrator.start_agent(
            context.to_agent,
            initial_prompt=handoff_prompt,
            inherited_state=context.shared_state,
        )
```

#### 6.4.2 Shared Memory Protocol

```python
# src/agent_orchestrator/coordination/shared_memory.py
class SharedMemory:
    """
    Shared state accessible by all agents in a workflow.

    Inspired by Agent-MCP:
    "Multiple agents work simultaneously, coordinated through shared memory"
    """

    def __init__(self, workflow_id: str):
        self.workflow_id = workflow_id
        self._state: Dict[str, Any] = {}
        self._locks: Dict[str, asyncio.Lock] = {}
        self._history: List[StateChange] = []

    async def read(self, key: str) -> Any:
        """Read value from shared memory."""
        return self._state.get(key)

    async def write(self, key: str, value: Any, agent_id: str):
        """Write value to shared memory with lock."""
        if key not in self._locks:
            self._locks[key] = asyncio.Lock()

        async with self._locks[key]:
            old_value = self._state.get(key)
            self._state[key] = value

            self._history.append(StateChange(
                key=key,
                old_value=old_value,
                new_value=value,
                agent_id=agent_id,
                timestamp=datetime.now(),
            ))

    async def wait_for(self, key: str, timeout: float = 60.0) -> Any:
        """Wait for a key to be set."""
        start = datetime.now()
        while (datetime.now() - start).total_seconds() < timeout:
            if key in self._state:
                return self._state[key]
            await asyncio.sleep(0.1)
        raise TimeoutError(f"Timeout waiting for key: {key}")
```

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 6.4.1 | Implement `HandoffManager` | `coordination/handoff.py` |
| 6.4.2 | Implement `SharedMemory` | `coordination/shared_memory.py` |
| 6.4.3 | Implement `FileReservationSystem` | `coordination/reservations.py` |
| 6.4.4 | Add checkpoint/resume support | `coordination/checkpoints.py` |
| 6.4.5 | Write coordination tests | `tests/unit/test_coordination.py` |

---

## Phase 6.5: Cost Optimization System

**Goal:** Implement comprehensive cost tracking with budget alerts and optimization.

**Priority:** MEDIUM
**Effort:** 10-15 hours
**Reference:** [GitHub Budget Tracking](https://github.blog/changelog/2025-11-03-control-ai-spending-with-budget-tracking-for-github-ai-tools/)

### Features to Implement

#### 6.5.1 Budget Alert System

```python
# src/agent_orchestrator/budget/alerts.py
@dataclass
class BudgetConfig:
    """Budget configuration with alerts."""
    daily_limit_usd: float
    monthly_limit_usd: float
    alert_thresholds: List[int] = field(default_factory=lambda: [50, 75, 90, 100])
    block_on_exhaustion: bool = True
    notification_channels: List[str] = field(default_factory=list)

class BudgetAlertSystem:
    """
    GitHub-style budget tracking with alerts.

    Features:
    - Alerts at 75%, 90%, 100% thresholds
    - Product-specific and bundled budgets
    - Block further use when exhausted
    """

    def __init__(self, config: BudgetConfig):
        self.config = config
        self._alerts_sent: Set[str] = set()

    async def check_and_alert(self, current_spend: float, budget_type: str):
        """Check spend against budget and send alerts."""
        limit = (
            self.config.daily_limit_usd
            if budget_type == "daily"
            else self.config.monthly_limit_usd
        )

        percentage = (current_spend / limit) * 100

        for threshold in self.config.alert_thresholds:
            if percentage >= threshold:
                alert_key = f"{budget_type}_{threshold}"
                if alert_key not in self._alerts_sent:
                    await self._send_alert(
                        f"{budget_type.title()} budget at {threshold}%",
                        f"Current spend: ${current_spend:.2f} / ${limit:.2f}",
                        severity="critical" if threshold >= 90 else "warning",
                    )
                    self._alerts_sent.add(alert_key)

        # Block if exhausted
        if percentage >= 100 and self.config.block_on_exhaustion:
            raise BudgetExhaustedError(
                f"{budget_type.title()} budget exhausted: ${current_spend:.2f}"
            )
```

#### 6.5.2 Cost Optimization Recommendations

```python
# src/agent_orchestrator/budget/optimizer.py
class CostOptimizer:
    """Analyze usage and recommend cost optimizations."""

    async def analyze_usage(self, days: int = 7) -> CostAnalysis:
        """Analyze recent usage patterns."""
        usage = await self.storage.get_usage_history(days)

        return CostAnalysis(
            total_cost=sum(u.cost for u in usage),
            by_agent=self._group_by_agent(usage),
            by_model=self._group_by_model(usage),
            by_task_type=self._group_by_task_type(usage),
            peak_hours=self._find_peak_hours(usage),
            inefficiencies=self._find_inefficiencies(usage),
        )

    def _find_inefficiencies(self, usage: List[UsageRecord]) -> List[Inefficiency]:
        """Find potential cost inefficiencies."""
        issues = []

        # High retry rates
        for agent_id, records in self._group_by_agent(usage).items():
            retry_rate = sum(1 for r in records if r.is_retry) / len(records)
            if retry_rate > 0.2:
                issues.append(Inefficiency(
                    type="high_retry_rate",
                    agent_id=agent_id,
                    metric=retry_rate,
                    recommendation=f"Consider debugging {agent_id} - {retry_rate:.0%} retry rate",
                ))

        # Expensive model for simple tasks
        for record in usage:
            if record.model == "opus" and record.task_complexity == "low":
                issues.append(Inefficiency(
                    type="overprovisioned_model",
                    task_id=record.task_id,
                    recommendation="Use Sonnet or Haiku for simple tasks",
                ))

        return issues
```

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 6.5.1 | Implement `BudgetAlertSystem` | `budget/alerts.py` |
| 6.5.2 | Implement `CostOptimizer` | `budget/optimizer.py` |
| 6.5.3 | Add budget API endpoints | `api/routes/budget.py` |
| 6.5.4 | Implement cost projections | `budget/projections.py` |
| 6.5.5 | Write budget tests | `tests/unit/test_budget_alerts.py` |

---

## Phase 6.6: Plugin Marketplace Architecture

**Goal:** Create extensible plugin system for agents, skills, and tools.

**Priority:** LOW
**Effort:** 15-20 hours
**Reference:** [wshobson/agents](https://github.com/wshobson/agents)

### Features to Implement

#### 6.6.1 Plugin Structure

```
plugins/
├── security-auditor/
│   ├── manifest.json
│   ├── agents/
│   │   └── security_auditor.py
│   ├── skills/
│   │   └── vulnerability_scan.py
│   └── tools/
│       └── owasp_checker.py
├── kubernetes-expert/
│   ├── manifest.json
│   ├── agents/
│   ├── skills/
│   └── tools/
└── registry.json
```

#### 6.6.2 Plugin Manifest

```json
{
    "name": "security-auditor",
    "version": "1.0.0",
    "description": "Security auditing agent with OWASP scanning",
    "author": "Agent Orchestration Team",
    "agents": [
        {
            "id": "security_auditor",
            "tier": 1,
            "model": "opus",
            "capabilities": ["code_review", "security_scan"]
        }
    ],
    "skills": [
        {
            "id": "vulnerability_scan",
            "description": "Scan code for vulnerabilities",
            "tier": 2
        }
    ],
    "tools": [
        {
            "id": "owasp_checker",
            "description": "Check against OWASP Top 10",
            "requires_approval": false
        }
    ],
    "dependencies": ["base-code-review"]
}
```

#### 6.6.3 Plugin Registry

```python
# src/agent_orchestrator/plugins/registry.py
class PluginRegistry:
    """
    Manage installed plugins.

    Inspired by wshobson/agents:
    "68 focused, single-purpose plugins"
    """

    def __init__(self, plugins_dir: Path):
        self.plugins_dir = plugins_dir
        self._plugins: Dict[str, Plugin] = {}

    async def load_all(self):
        """Load all plugins from directory."""
        for plugin_dir in self.plugins_dir.iterdir():
            if plugin_dir.is_dir() and (plugin_dir / "manifest.json").exists():
                plugin = await self._load_plugin(plugin_dir)
                self._plugins[plugin.name] = plugin

    async def install(self, source: str):
        """Install plugin from source (git URL or path)."""
        pass

    async def uninstall(self, name: str):
        """Uninstall a plugin."""
        pass

    def get_agents(self) -> List[AgentConfig]:
        """Get all agents from all plugins."""
        agents = []
        for plugin in self._plugins.values():
            agents.extend(plugin.agents)
        return agents

    def get_skills(self) -> List[SkillConfig]:
        """Get all skills from all plugins."""
        skills = []
        for plugin in self._plugins.values():
            skills.extend(plugin.skills)
        return skills
```

### Tasks

| Task | Description | Output Files |
|------|-------------|--------------|
| 6.6.1 | Define plugin manifest schema | `plugins/schema.py` |
| 6.6.2 | Implement `PluginRegistry` | `plugins/registry.py` |
| 6.6.3 | Implement `PluginLoader` | `plugins/loader.py` |
| 6.6.4 | Add plugin CLI commands | `cli/plugins.py` |
| 6.6.5 | Create example plugins | `plugins/examples/` |
| 6.6.6 | Write plugin tests | `tests/unit/test_plugins.py` |

---

## Reference Projects

### Multi-Agent Orchestration

| Project | Key Features | GitHub |
|---------|--------------|--------|
| Swarms | Concurrent/hierarchical workflows, GroupChat, MixtureOfAgents | [kyegomez/swarms](https://github.com/kyegomez/swarms) |
| Claude Flow | Swarm intelligence, MCP integration, workflow orchestration | [ruvnet/claude-flow](https://github.com/ruvnet/claude-flow) |
| Agent-MCP | MCP-based coordination, shared memory, parallel execution | [rinadelph/Agent-MCP](https://github.com/rinadelph/Agent-MCP) |
| DeepResearchAgent | Hierarchical planning, task decomposition | [SkyworkAI/DeepResearchAgent](https://github.com/SkyworkAI/DeepResearchAgent) |

### UI & Monitoring

| Project | Key Features | GitHub |
|---------|--------------|--------|
| Claude Squad | TUI interface, session management, git worktree isolation | [smtg-ai/claude-squad](https://github.com/smtg-ai/claude-squad) |
| Toad | Universal AI agent TUI, ACP protocol | [Will McGugan](https://willmcgugan.github.io/announcing-toad/) |
| AgentOps | 2-line integration, session replay, cost tracking | [AgentOps-AI/agentops](https://github.com/AgentOps-AI/agentops) |
| Textual | 5-10x faster TUI rendering, rich widgets | [Textualize/textual](https://github.com/Textualize/textual) |

### Plugin Architecture

| Project | Key Features | GitHub |
|---------|--------------|--------|
| wshobson/agents | 100 agents, 110 skills, 76 tools, plugin architecture | [wshobson/agents](https://github.com/wshobson/agents) |

---

## Implementation Priority

| Phase | Feature | Priority | Effort | Dependencies |
|-------|---------|----------|--------|--------------|
| 6.2 | Terminal UI Dashboard | HIGH | 25-30h | Phase 5 |
| 6.1 | Swarm Intelligence | HIGH | 20-25h | Phase 5 |
| 6.3 | Session Tracing | MEDIUM | 15-20h | Phase 6.2 |
| 6.5 | Cost Optimization | MEDIUM | 10-15h | Phase 5 |
| 6.4 | Agent Coordination | MEDIUM | 15-20h | Phase 6.1 |
| 6.6 | Plugin Architecture | LOW | 15-20h | Phase 6.1 |

**Recommended Order:** 6.2 → 6.1 → 6.5 → 6.3 → 6.4 → 6.6

---

## Success Criteria

| Metric | Target |
|--------|--------|
| TUI refresh rate | 60 FPS |
| Concurrent workflow efficiency | 3x speedup vs sequential |
| Trace query latency | <100ms p95 |
| Budget alert accuracy | 100% |
| Plugin load time | <2 seconds |
| Test coverage | >80% |

---

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

---

## Sources

- [Swarms Framework](https://github.com/kyegomez/swarms)
- [Claude Squad](https://github.com/smtg-ai/claude-squad)
- [wshobson/agents](https://github.com/wshobson/agents)
- [Claude Flow](https://github.com/ruvnet/claude-flow)
- [AgentOps](https://github.com/AgentOps-AI/agentops)
- [Toad TUI](https://willmcgugan.github.io/announcing-toad/)
- [Textual Framework](https://textual.textualize.io/)
- [GitHub Agent HQ](https://github.blog/news-insights/company-news/welcome-home-agents/)
- [LangSmith Observability](https://www.langchain.com/langsmith/observability)
- [Microsoft Multi-Agent Reference Architecture](https://microsoft.github.io/multi-agent-reference-architecture/)
- [GitHub Budget Tracking](https://github.blog/changelog/2025-11-03-control-ai-spending-with-budget-tracking-for-github-ai-tools/)
