Module 8: Project — Complete Multi-Agent System

4. Monitoring — SDK for Observability, Remote Control for Approvals

Description

The team of 5 agents executes the task board, the hooks validate each action, and the team lead coordinates everything. But there's a problem: you have no visibility into what's happening until it finishes. How long did each agent take? How many tasks did they complete? Were there failures you didn't see? What was the total cost?

In this capsule you build the observability layer. A Python script (headless SDK) that monitors the team's progress, generates execution reports, and produces per-agent metrics. In addition, you configure remote control to approve critical operations from your phone — you don't need to be in front of the computer for the system to advance.

When you finish you'll have a multi-agent execution dashboard: you'll know exactly what each agent did, how long it took, how much it cost, and where the bottlenecks were.

The Monitoring Script

Monitor architecture

┌─────────────────────────────────────────────────┐
│              team-monitor.py                     │
│                                                  │
│  ┌──────────┐  ┌──────────┐  ┌──────────────┐   │
│  │  Log     │  │  Report  │  │  Dashboard   │   │
│  │  Parser  │  │  Builder │  │  Generator   │   │
│  └──────────┘  └──────────┘  └──────────────┘   │
│       │              │              │            │
│  Reads agent   Builds report   Generates Markdown│
│  activity.log  with metrics    with tables       │
│       │              │              │            │
│  ┌──────────────────────────────────────────┐    │
│  │       logs/agents/agent-activity.log      │    │
│  └──────────────────────────────────────────┘    │
└─────────────────────────────────────────────────┘

The complete script

Create scripts/monitor/team-monitor.py:

#!/usr/bin/env python3
"""
Multi-Agent Team Monitor
Parses agent activity logs, generates execution reports,
and produces a markdown dashboard.

Usage:
  python scripts/monitor/team-monitor.py              # Full report
  python scripts/monitor/team-monitor.py --summary     # Summary only
  python scripts/monitor/team-monitor.py --watch        # Live monitoring
"""

import json
import os
import sys
import time
import argparse
from datetime import datetime
from collections import defaultdict


LOG_FILE = "logs/agents/agent-activity.log"
REPORT_DIR = "docs/reports"


def parse_log_file(log_path):
    """Parse the agent activity log into structured entries."""
    entries = []

    if not os.path.exists(log_path):
        return entries

    with open(log_path, "r") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue

            try:
                parts = {}
                timestamp_end = line.index("]")
                parts["timestamp"] = line[1:timestamp_end]

                remainder = line[timestamp_end + 2:]
                for segment in remainder.split(" | "):
                    key, value = segment.split(": ", 1)
                    parts[key.strip().lower()] = value.strip()

                if "duration" in parts:
                    duration_str = parts["duration"].replace("ms", "")
                    parts["duration_ms"] = int(duration_str)

                entries.append(parts)
            except (ValueError, IndexError):
                continue

    return entries


def calculate_metrics(entries):
    """Calculate per-agent and overall metrics."""
    metrics = {
        "agents": defaultdict(lambda: {
            "tasks_completed": 0,
            "total_duration_ms": 0,
            "durations": [],
            "first_active": None,
            "last_active": None,
        }),
        "total_tasks": 0,
        "total_duration_ms": 0,
        "first_event": None,
        "last_event": None,
    }

    for entry in entries:
        agent = entry.get("agent", "unknown")
        duration = entry.get("duration_ms", 0)
        timestamp = entry.get("timestamp", "")

        agent_data = metrics["agents"][agent]
        agent_data["tasks_completed"] += 1
        agent_data["total_duration_ms"] += duration
        agent_data["durations"].append(duration)

        if agent_data["first_active"] is None:
            agent_data["first_active"] = timestamp
        agent_data["last_active"] = timestamp

        metrics["total_tasks"] += 1
        metrics["total_duration_ms"] += duration

        if metrics["first_event"] is None:
            metrics["first_event"] = timestamp
        metrics["last_event"] = timestamp

    for agent, data in metrics["agents"].items():
        durations = data["durations"]
        if durations:
            data["avg_duration_ms"] = sum(durations) / len(durations)
            data["min_duration_ms"] = min(durations)
            data["max_duration_ms"] = max(durations)
        else:
            data["avg_duration_ms"] = 0
            data["min_duration_ms"] = 0
            data["max_duration_ms"] = 0

    return metrics


def format_duration(ms):
    """Format milliseconds to human-readable string."""
    if ms < 1000:
        return f"{ms}ms"
    elif ms < 60000:
        return f"{ms / 1000:.1f}s"
    else:
        minutes = ms / 60000
        return f"{minutes:.1f}min"


def print_summary(metrics):
    """Print a concise summary to stdout."""
    print("=" * 60)
    print("  MULTI-AGENT TEAM — EXECUTION SUMMARY")
    print("=" * 60)
    print()

    if metrics["first_event"]:
        print(f"  First event:  {metrics['first_event']}")
        print(f"  Last event:   {metrics['last_event']}")
    print(f"  Total tasks:  {metrics['total_tasks']}")
    print(f"  Total time:   {format_duration(metrics['total_duration_ms'])}")
    print()

    print("  AGENT BREAKDOWN")
    print("  " + "-" * 56)
    print(f"  {'Agent':<22} {'Tasks':>6} {'Total':>10} {'Avg':>10} {'Max':>10}")
    print("  " + "-" * 56)

    sorted_agents = sorted(
        metrics["agents"].items(),
        key=lambda x: x[1]["total_duration_ms"],
        reverse=True
    )

    for agent, data in sorted_agents:
        print(
            f"  {agent:<22} "
            f"{data['tasks_completed']:>6} "
            f"{format_duration(data['total_duration_ms']):>10} "
            f"{format_duration(data['avg_duration_ms']):>10} "
            f"{format_duration(data['max_duration_ms']):>10}"
        )

    print("  " + "-" * 56)
    print()


def generate_markdown_report(metrics, entries):
    """Generate a full markdown execution report."""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    lines = []
    lines.append(f"# Multi-Agent Execution Report")
    lines.append(f"")
    lines.append(f"**Generated:** {now}")
    lines.append(f"")

    lines.append(f"## Summary")
    lines.append(f"")
    lines.append(f"| Metric | Value |")
    lines.append(f"|--------|-------|")
    lines.append(f"| Total tasks completed | {metrics['total_tasks']} |")
    lines.append(
        f"| Total agent time | "
        f"{format_duration(metrics['total_duration_ms'])} |"
    )
    lines.append(f"| First event | {metrics.get('first_event', 'N/A')} |")
    lines.append(f"| Last event | {metrics.get('last_event', 'N/A')} |")
    lines.append(f"| Agents active | {len(metrics['agents'])} |")
    lines.append(f"")

    lines.append(f"## Agent Performance")
    lines.append(f"")
    lines.append(
        f"| Agent | Tasks | Total Time | Avg Time | "
        f"Min Time | Max Time |"
    )
    lines.append(
        f"|-------|-------|------------|----------|"
        f"----------|----------|"
    )

    sorted_agents = sorted(
        metrics["agents"].items(),
        key=lambda x: x[1]["total_duration_ms"],
        reverse=True
    )

    for agent, data in sorted_agents:
        lines.append(
            f"| {agent} "
            f"| {data['tasks_completed']} "
            f"| {format_duration(data['total_duration_ms'])} "
            f"| {format_duration(data['avg_duration_ms'])} "
            f"| {format_duration(data['min_duration_ms'])} "
            f"| {format_duration(data['max_duration_ms'])} |"
        )

    lines.append(f"")

    if sorted_agents:
        most_active = sorted_agents[0]
        lines.append(f"### Insights")
        lines.append(f"")
        lines.append(
            f"- **Most active agent:** {most_active[0]} "
            f"({most_active[1]['tasks_completed']} tasks, "
            f"{format_duration(most_active[1]['total_duration_ms'])})"
        )

        if len(sorted_agents) > 1:
            least_active = sorted_agents[-1]
            lines.append(
                f"- **Least active agent:** {least_active[0]} "
                f"({least_active[1]['tasks_completed']} tasks, "
                f"{format_duration(least_active[1]['total_duration_ms'])})"
            )

        all_durations = []
        for _, data in sorted_agents:
            all_durations.extend(data["durations"])
        if all_durations:
            avg_all = sum(all_durations) / len(all_durations)
            lines.append(
                f"- **Average task duration (all agents):** "
                f"{format_duration(avg_all)}"
            )

    lines.append(f"")
    lines.append(f"## Activity Timeline")
    lines.append(f"")
    lines.append(f"| # | Timestamp | Agent | Duration | Status |")
    lines.append(f"|---|-----------|-------|----------|--------|")

    for i, entry in enumerate(entries, 1):
        lines.append(
            f"| {i} "
            f"| {entry.get('timestamp', 'N/A')} "
            f"| {entry.get('agent', 'unknown')} "
            f"| {format_duration(entry.get('duration_ms', 0))} "
            f"| {entry.get('status', 'unknown')} |"
        )

    lines.append(f"")

    return "\n".join(lines)


def watch_mode(log_path, interval=5):
    """Live monitoring mode — polls log file for new entries."""
    print("Watching for agent activity... (Ctrl+C to stop)")
    print()

    seen_lines = 0

    while True:
        if os.path.exists(log_path):
            with open(log_path, "r") as f:
                all_lines = f.readlines()

            new_lines = all_lines[seen_lines:]

            for line in new_lines:
                line = line.strip()
                if line:
                    print(f"  >> {line}")

            seen_lines = len(all_lines)

        time.sleep(interval)


def main():
    parser = argparse.ArgumentParser(
        description="Multi-Agent Team Monitor"
    )
    parser.add_argument(
        "--summary", action="store_true",
        help="Print summary only"
    )
    parser.add_argument(
        "--watch", action="store_true",
        help="Live monitoring mode"
    )
    parser.add_argument(
        "--report", action="store_true",
        help="Generate markdown report"
    )
    parser.add_argument(
        "--log-file", default=LOG_FILE,
        help="Path to agent activity log"
    )

    args = parser.parse_args()

    if args.watch:
        try:
            watch_mode(args.log_file)
        except KeyboardInterrupt:
            print("\nStopped watching.")
            sys.exit(0)

    entries = parse_log_file(args.log_file)

    if not entries:
        print("No agent activity found.")
        print(f"Expected log at: {args.log_file}")
        print("Run the multi-agent team first to generate activity logs.")
        sys.exit(0)

    metrics = calculate_metrics(entries)

    print_summary(metrics)

    if args.report or not args.summary:
        os.makedirs(REPORT_DIR, exist_ok=True)

        report = generate_markdown_report(metrics, entries)
        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        report_path = os.path.join(
            REPORT_DIR, f"execution-report-{timestamp}.md"
        )

        with open(report_path, "w") as f:
            f.write(report)

        print(f"Report saved to: {report_path}")


if __name__ == "__main__":
    main()

Using the monitor

# While the team executes (another terminal)
python scripts/monitor/team-monitor.py --watch

# After the execution — quick summary
python scripts/monitor/team-monitor.py --summary

# Generate a full report in Markdown
python scripts/monitor/team-monitor.py --report

Expected output (--summary)

============================================================
  MULTI-AGENT TEAM — EXECUTION SUMMARY
============================================================

  First event:  2026-03-13T14:22:01Z
  Last event:   2026-03-13T14:28:30Z
  Total tasks:  8
  Total time:   2.3min

  AGENT BREAKDOWN
  --------------------------------------------------------
  Agent                   Tasks      Total        Avg        Max
  --------------------------------------------------------
  backend-agent               3     52.7s      17.6s      22.1s
  testing-agent               2     45.3s      22.7s      25.9s
  frontend-agent              2     28.0s      14.0s      15.7s
  docs-review-agent           1     16.8s      16.8s      16.8s
  --------------------------------------------------------

SDK: Orchestrate the Complete Execution from Python

The orchestrator

A more advanced script that not only monitors, but orchestrates the complete execution using the headless SDK:

Create scripts/monitor/orchestrator.py:

#!/usr/bin/env python3
"""
Multi-Agent Orchestrator
Launches the team lead via SDK, monitors progress, and generates
the execution report automatically.

Usage:
  python scripts/monitor/orchestrator.py "Implement a tasks feature"
"""

import subprocess
import json
import sys
import os
import time
from datetime import datetime


def run_claude(prompt, agent=None, tools=None, timeout=900):
    """Execute Claude Code in headless mode."""
    cmd = ["claude", "-p", prompt, "--output-format", "json"]

    if agent:
        cmd.extend(["--agent", agent])

    if tools:
        cmd.extend(["--allowedTools", ",".join(tools)])

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout
        )
    except subprocess.TimeoutExpired:
        return {
            "is_error": True,
            "error": f"Timeout after {timeout}s"
        }
    except FileNotFoundError:
        return {
            "is_error": True,
            "error": "claude CLI not found"
        }

    if result.returncode != 0:
        return {
            "is_error": True,
            "error": result.stderr or "Unknown error"
        }

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return {
            "is_error": True,
            "error": f"Invalid JSON: {result.stdout[:200]}"
        }


def generate_report(result, feature_description, start_time):
    """Generate execution report from Claude's output."""
    end_time = datetime.now()
    duration = (end_time - start_time).total_seconds()

    report_lines = [
        "# Orchestrated Execution Report",
        "",
        f"**Feature:** {feature_description}",
        f"**Start:** {start_time.strftime('%Y-%m-%d %H:%M:%S')}",
        f"**End:** {end_time.strftime('%Y-%m-%d %H:%M:%S')}",
        f"**Duration:** {duration:.1f}s",
        f"**Cost:** ${result.get('cost_usd', 0):.4f}",
        f"**Turns:** {result.get('num_turns', 0)}",
        "",
        "## Team Lead Output",
        "",
        result.get("result", "No output"),
        "",
    ]

    return "\n".join(report_lines)


def main():
    if len(sys.argv) < 2:
        print("Usage: python orchestrator.py \"<feature description>\"")
        sys.exit(1)

    feature = sys.argv[1]
    start_time = datetime.now()

    print(f"Launching multi-agent team for: {feature}")
    print(f"Start time: {start_time.strftime('%H:%M:%S')}")
    print("This may take several minutes...")
    print()

    prompt = f"""Execute this feature with the full team:

{feature}

Process:
1. Generate task board with 8-10 tasks
2. Execute immediately without waiting for approval
3. Run all agents: backend, frontend, testing, docs/review
4. Produce final consolidated report

Execute now."""

    result = run_claude(
        prompt=prompt,
        agent="team-lead",
        timeout=1200
    )

    if result.get("is_error"):
        print(f"ERROR: {result.get('error')}")
        sys.exit(1)

    print("Execution complete!")
    print(f"Cost: ${result.get('cost_usd', 0):.4f}")
    print(f"Turns: {result.get('num_turns', 0)}")
    print()

    report = generate_report(result, feature, start_time)

    os.makedirs("docs/reports", exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    report_path = f"docs/reports/orchestrated-{timestamp}.md"

    with open(report_path, "w") as f:
        f.write(report)

    print(f"Report saved to: {report_path}")
    print()
    print("Team Lead Output (last 500 chars):")
    print("-" * 40)
    output = result.get("result", "")
    print(output[-500:] if len(output) > 500 else output)


if __name__ == "__main__":
    main()

Using the orchestrator

python scripts/monitor/orchestrator.py \
  "Implement task CRUD with endpoints, components, tests, and docs"

This script:

Launches the team lead in headless mode
The team lead generates the task board and executes automatically
When it finishes, the script receives the JSON result
Generates a report in docs/reports/
Prints the summary in the terminal

Remote Control: Approvals from Your Phone

Configuring remote control

Remote control lets the team lead or the hooks ask you for approval for critical operations, and you approve from your phone or any device with a browser.

# Enable remote control (if not enabled)
claude config set remote_control true

Approval scenarios

1. The team lead needs to run a destructive command:

[Team Lead] Task T4 requires dropping and recreating the test database.
Command: "python manage.py flush --no-input"
Awaiting approval via remote control...

You receive a notification → you approve from your phone → the execution continues.

2. A hook blocks an operation that needs an override:

[Hook] PreToolUse BLOCKED: "rm tests/test_old_tasks.py"
Reason: Destructive command.
To proceed, approve via remote control or adjust the hook.

3. The testing-agent wants to run the full suite (costly):

[Testing Agent] Ready to run full test suite (estimated 3 minutes).
Awaiting approval...

Combining remote control with the orchestrator

The orchestrator can be configured to send notifications when it needs approval:

import subprocess

def check_remote_approval_needed(result_text):
    """Check if the execution is waiting for remote approval."""
    approval_keywords = [
        "awaiting approval",
        "waiting for confirmation",
        "remote control",
        "approve"
    ]
    return any(
        kw in result_text.lower()
        for kw in approval_keywords
    )

The complete flow with remote control

1. You launch the orchestrator from the terminal
   $ python scripts/monitor/orchestrator.py "Feature X"

2. The team executes automatically
   - Backend and frontend work in parallel
   - Hooks validate each action
   - Logs are recorded

3. An operation needs approval
   - The hook blocks with exit 2
   - Remote control sends a notification
   - You approve from your phone

4. The execution continues
   - More tasks are completed
   - Tests run
   - Quality review is generated

5. Final report is generated automatically
   $ cat docs/reports/orchestrated-*.md

Metrics and Dashboarding

Extended metrics script

Create scripts/monitor/metrics.py:

#!/usr/bin/env python3
"""
Generates extended metrics from agent activity logs.
Produces a markdown dashboard with charts (text-based).
"""

import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from team_monitor import parse_log_file, calculate_metrics, format_duration


LOG_FILE = "logs/agents/agent-activity.log"


def generate_bar(value, max_value, width=30):
    """Generate a text-based bar chart."""
    if max_value == 0:
        return ""
    filled = int((value / max_value) * width)
    return "█" * filled + "░" * (width - filled)


def generate_dashboard(metrics):
    """Generate a text-based dashboard."""
    lines = []

    lines.append("╔══════════════════════════════════════════════════════════╗")
    lines.append("║          MULTI-AGENT EXECUTION DASHBOARD                ║")
    lines.append("╚══════════════════════════════════════════════════════════╝")
    lines.append("")

    lines.append(f"  Total Tasks: {metrics['total_tasks']:<6}  "
                 f"Total Time: {format_duration(metrics['total_duration_ms']):<10}  "
                 f"Agents: {len(metrics['agents'])}")
    lines.append("")

    lines.append("  TIME PER AGENT")
    lines.append("  " + "─" * 55)

    max_duration = max(
        (d["total_duration_ms"] for d in metrics["agents"].values()),
        default=0
    )

    sorted_agents = sorted(
        metrics["agents"].items(),
        key=lambda x: x[1]["total_duration_ms"],
        reverse=True
    )

    for agent, data in sorted_agents:
        bar = generate_bar(data["total_duration_ms"], max_duration, 25)
        lines.append(
            f"  {agent:<22} {bar} "
            f"{format_duration(data['total_duration_ms'])}"
        )

    lines.append("")
    lines.append("  TASKS PER AGENT")
    lines.append("  " + "─" * 55)

    max_tasks = max(
        (d["tasks_completed"] for d in metrics["agents"].values()),
        default=0
    )

    for agent, data in sorted_agents:
        bar = generate_bar(data["tasks_completed"], max_tasks, 25)
        lines.append(
            f"  {agent:<22} {bar} "
            f"{data['tasks_completed']} tasks"
        )

    lines.append("")

    if len(sorted_agents) >= 2:
        lines.append("  EFFICIENCY ANALYSIS")
        lines.append("  " + "─" * 55)

        for agent, data in sorted_agents:
            if data["tasks_completed"] > 0:
                efficiency = data["avg_duration_ms"]
                lines.append(
                    f"  {agent:<22} "
                    f"avg: {format_duration(efficiency):<8} "
                    f"min: {format_duration(data['min_duration_ms']):<8} "
                    f"max: {format_duration(data['max_duration_ms'])}"
                )

    lines.append("")

    return "\n".join(lines)


def main():
    entries = parse_log_file(LOG_FILE)

    if not entries:
        print("No agent activity found.")
        sys.exit(0)

    metrics = calculate_metrics(entries)
    dashboard = generate_dashboard(metrics)
    print(dashboard)

    os.makedirs("docs/reports", exist_ok=True)
    with open("docs/reports/dashboard.txt", "w") as f:
        f.write(dashboard)

    print(f"Dashboard saved to: docs/reports/dashboard.txt")


if __name__ == "__main__":
    main()

Dashboard output

╔══════════════════════════════════════════════════════════╗
║          MULTI-AGENT EXECUTION DASHBOARD                ║
╚══════════════════════════════════════════════════════════╝

  Total Tasks: 8       Total Time: 2.3min      Agents: 4

  TIME PER AGENT
  ───────────────────────────────────────────────────────
  backend-agent          █████████████████████████ 52.7s
  testing-agent          █████████████████████░░░░ 45.3s
  frontend-agent         █████████████░░░░░░░░░░░░ 28.0s
  docs-review-agent      ████████░░░░░░░░░░░░░░░░░ 16.8s

  TASKS PER AGENT
  ───────────────────────────────────────────────────────
  backend-agent          █████████████████████████ 3 tasks
  testing-agent          ████████████████░░░░░░░░░ 2 tasks
  frontend-agent         ████████████████░░░░░░░░░ 2 tasks
  docs-review-agent      ████████░░░░░░░░░░░░░░░░░ 1 tasks

  EFFICIENCY ANALYSIS
  ───────────────────────────────────────────────────────
  backend-agent          avg: 17.6s    min: 12.3s   max: 22.1s
  testing-agent          avg: 22.7s    min: 19.5s   max: 25.9s
  frontend-agent         avg: 14.0s    min: 12.3s   max: 15.7s
  docs-review-agent      avg: 16.8s    min: 16.8s   max: 16.8s

Integrating Everything: The Complete Workflow

The end-to-end flow with monitoring

Terminal 1: Execute
─────────────────────────────────────────
$ python scripts/monitor/orchestrator.py \
    "Implement complete task CRUD"

Terminal 2: Monitor (simultaneous)
─────────────────────────────────────────
$ python scripts/monitor/team-monitor.py --watch

  >> [2026-03-13T14:22:01Z] Agent: backend-agent | Duration: 18230ms
  >> [2026-03-13T14:22:15Z] Agent: frontend-agent | Duration: 8450ms
  >> [2026-03-13T14:23:42Z] Agent: backend-agent | Duration: 22100ms
  ...

Terminal 3: Dashboard (afterwards)
─────────────────────────────────────────
$ python scripts/monitor/metrics.py

[Dashboard output]

Automate the complete workflow

Create scripts/run-team.sh:

#!/bin/bash
# Run the multi-agent team with full monitoring

FEATURE="$1"

if [[ -z "$FEATURE" ]]; then
  echo "Usage: ./scripts/run-team.sh \"feature description\""
  exit 1
fi

mkdir -p logs/agents docs/reports

echo "Starting multi-agent execution..."
echo "Feature: $FEATURE"
echo "Time: $(date)"
echo ""

# Clear previous logs
> logs/agents/agent-activity.log

# Launch monitor in background
python scripts/monitor/team-monitor.py --watch &
MONITOR_PID=$!

# Run the orchestrator
python scripts/monitor/orchestrator.py "$FEATURE"
EXIT_CODE=$?

# Stop the monitor
kill $MONITOR_PID 2>/dev/null

# Generate dashboard
echo ""
echo "Generating dashboard..."
python scripts/monitor/metrics.py

echo ""
echo "Done! Check docs/reports/ for full reports."

exit $EXIT_CODE

chmod +x scripts/run-team.sh

Usage:

./scripts/run-team.sh "Implement task CRUD with tests and docs"

Exercises

Exercise 1: Run the monitor (Easy)

Execute the multi-agent team (capsule 03), then run python scripts/monitor/team-monitor.py --summary. Verify that the logs were recorded and the metrics are correct.

Exercise 2: Live monitoring (Easy)

Open two terminals. In the first, run the team with claude --agent team-lead. In the second, run python scripts/monitor/team-monitor.py --watch. Observe the events in real time while the team works.

Exercise 3: Add cost metrics (Medium)

Modify team-monitor.py so it also records and reports the estimated cost per agent. Use the cost_usd field from the SDK JSON output. Update the dashboard with a cost column.

Exercise 4: Notifications (Medium)

Add a function to orchestrator.py that sends a notification when the execution finishes. Options: write to a notifications file, send a webhook to Slack, or simply play a system sound with os.system("afplay /System/Library/Sounds/Glass.aiff") (macOS).

Exercise 5: Compare executions (Hard)

Modify metrics.py to accept a --compare flag that reads two log files (from two different executions) and compares the metrics side by side. Useful for measuring the impact of changing system prompts or hooks.

Exercise 6: Orchestrator with retry (Hard)

Modify orchestrator.py to detect whether the team didn't complete all the tasks (PARTIAL status in the output), and automatically launch a second execution with the missing tasks. The retry must include the context of what was already completed.

Troubleshooting

"No agent activity found"

Cause: The logs weren't generated because the SubagentStop hook didn't fire or the log file path is incorrect.

Solution:

# Verify that the hook is executable
ls -la scripts/hooks/subagent-stop-log.sh

# Verify that the log directory exists
mkdir -p logs/agents

# Verify settings.json
cat .claude/settings.json | jq '.hooks.SubagentStop'

"The orchestrator times out"

Cause: The multi-agent execution takes longer than the subprocess timeout.

Solution: Increase the timeout in the orchestrator:

result = run_claude(
    prompt=prompt,
    agent="team-lead",
    timeout=1800  # 30 minutes
)

"Remote control doesn't send notifications"

Cause: Remote control isn't enabled or the session isn't linked.

Solution:

claude config get remote_control
# If it's false:
claude config set remote_control true

"The dashboard shows incorrect data"

Cause: The log file has an inconsistent format or there are corrupted lines.

Solution: The parser has try/except for malformed lines. If it persists, clear the log file and re-execute.

"The metrics script imports from team_monitor but fails"

Cause: Python can't find the module because the path isn't configured.

Solution: Run from the project root:

cd your-project
python scripts/monitor/metrics.py

Comparison: Without Monitoring vs With Monitoring

Aspect	Without Monitoring	With Monitoring
Visibility	Only at the end	In real time
Metrics	None	Per agent: time, tasks, efficiency
Reports	Manual (copy the output)	Automatic (Markdown)
Debugging	Hard (no logs)	Per-agent activity logs
Reproducibility	Low	High (logs + reports)
Remote ops	Not possible	Approvals from phone

Summary

The team-monitor.py parses agent logs and generates metrics: tasks completed, duration per agent, averages, maximums
Three modes: --watch (real time), --summary (summary), --report (full Markdown)
The orchestrator.py runs the team lead via the headless SDK, receives the JSON result, and generates reports automatically
Remote control lets you approve critical operations from your phone while the team executes
The dashboard visualizes metrics with text-based bar charts: time and tasks per agent
The run-team.sh script automates the complete workflow: clear logs → monitor → execute → generate dashboard
The key metrics are: total time, tasks per agent, average duration per task, and efficiency (avg/max ratio)
Monitoring transforms the multi-agent execution from a "black box" to an "observable system"

Additional Resources

Claude Code CLI Reference — Flags -p, --output-format json, --agent for the headless SDK
Claude Code Hooks — SubagentStop hook for activity logging
Python subprocess Module — subprocess reference for programmatic invocation
Python argparse Module — CLI argument parsing for monitor scripts
Claude Code Best Practices — Automation and monitoring
Claude Code Overview — Context of Claude Code as a platform

Next capsule: In capsule 05 you'll analyze the results of the complete execution. Which agent was the most efficient? Where were the bottlenecks? What would you change for the next execution? It's the retrospective of the multi-agent system and the complete closing of the guide.

← Previous3. Execution — Task Board, Parallel Delegation, Quality Gates Next →5. Retrospective — Analysis, Optimization, and Closing the Guide