Módulo 8: Proyecto — Sistema Multi-Agente Completo

4. Monitoring — SDK para Observabilidad, Remote Control para Aprobaciones

Descripción

El equipo de 5 agentes ejecuta el task board, los hooks validan cada acción, y el team lead coordina todo. Pero hay un problema: no tienes visibilidad de lo que está pasando hasta que termina. ¿Cuánto tardó cada agente? ¿Cuántas tareas completaron? ¿Hubo fallos que no viste? ¿Cuál fue el costo total?

En esta cápsula construyes la capa de observabilidad. Un script Python (SDK headless) que monitorea el progreso del equipo, genera reportes de ejecución, y produce métricas por agente. Además, configuras remote control para aprobar operaciones críticas desde el celular — no necesitas estar frente a la computadora para que el sistema avance.

Al terminar tendrás un dashboard de ejecución multi-agente: sabrás exactamente qué hizo cada agente, cuánto tardó, cuánto costó, y dónde estuvieron los cuellos de botella.

El Script de Monitoring

Arquitectura del monitor

┌─────────────────────────────────────────────────┐
│              team-monitor.py                     │
│                                                  │
│  ┌──────────┐  ┌──────────┐  ┌──────────────┐   │
│  │  Log     │  │  Report  │  │  Dashboard   │   │
│  │  Parser  │  │  Builder │  │  Generator   │   │
│  └──────────┘  └──────────┘  └──────────────┘   │
│       │              │              │            │
│  Lee agent    Genera report    Genera Markdown   │
│  activity.log  con métricas    con tablas        │
│       │              │              │            │
│  ┌──────────────────────────────────────────┐    │
│  │       logs/agents/agent-activity.log      │    │
│  └──────────────────────────────────────────┘    │
└─────────────────────────────────────────────────┘

El script completo

Crea 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()

Uso del monitor

# Mientras el equipo ejecuta (otra terminal)
python scripts/monitor/team-monitor.py --watch

# Después de la ejecución — resumen rápido
python scripts/monitor/team-monitor.py --summary

# Generar reporte completo en Markdown
python scripts/monitor/team-monitor.py --report

Output esperado (--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: Orquestar la Ejecución Completa desde Python

El orquestador

Un script más avanzado que no solo monitorea, sino que orquesta la ejecución completa usando el SDK headless:

Crea 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 "Implementa feature de tasks"
"""

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()

Uso del orquestador

python scripts/monitor/orchestrator.py \
  "Implementa CRUD de tareas con endpoints, componentes, tests, y docs"

Este script:

Lanza al team lead en modo headless
El team lead genera el task board y ejecuta automáticamente
Al terminar, el script recibe el resultado JSON
Genera un reporte en docs/reports/
Imprime el resumen en la terminal

Remote Control: Aprobaciones desde el Celular

Configurando remote control

Remote control permite que el team lead o los hooks te pidan aprobación para operaciones críticas, y tú apruebes desde tu celular o cualquier dispositivo con browser.

# Habilitar remote control (si no está habilitado)
claude config set remote_control true

Escenarios de aprobación

1. El team lead necesita ejecutar un comando destructivo:

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

Recibes una notificación → apruebas desde el celular → la ejecución continúa.

2. Un hook bloquea una operación que necesita override:

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

3. El testing-agent quiere ejecutar la suite completa (costoso):

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

Combinando remote control con el orquestador

El orquestador puede configurarse para enviar notificaciones cuando necesita aprobación:

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
    )

El flujo completo con remote control

1. Lanzas el orquestador desde la terminal
   $ python scripts/monitor/orchestrator.py "Feature X"

2. El equipo ejecuta automáticamente
   - Backend y frontend trabajan en paralelo
   - Hooks validan cada acción
   - Logs se registran

3. Una operación necesita aprobación
   - El hook bloquea con exit 2
   - Remote control envía notificación
   - Tú apruebas desde el celular

4. La ejecución continúa
   - Más tareas se completan
   - Tests se ejecutan
   - Quality review se genera

5. Reporte final se genera automáticamente
   $ cat docs/reports/orchestrated-*.md

Métricas y Dashboarding

Script de métricas extendidas

Crea 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()

Output del dashboard

╔══════════════════════════════════════════════════════════╗
║          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

Integrando Todo: El Workflow Completo

El flujo end-to-end con monitoring

Terminal 1: Ejecutar
─────────────────────────────────────────
$ python scripts/monitor/orchestrator.py \
    "Implementa CRUD de tareas completo"

Terminal 2: Monitorear (simultáneo)
─────────────────────────────────────────
$ 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 (después)
─────────────────────────────────────────
$ python scripts/monitor/metrics.py

[Dashboard output]

Automatizar el workflow completo

Crea 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

Uso:

./scripts/run-team.sh "Implementa CRUD de tareas con tests y docs"

Ejercicios

Ejercicio 1: Ejecutar el monitor (Fácil)

Ejecuta el equipo multi-agente (cápsula 03), luego corre python scripts/monitor/team-monitor.py --summary. Verifica que los logs se registraron y las métricas son correctas.

Ejercicio 2: Live monitoring (Fácil)

Abre dos terminales. En la primera, ejecuta el equipo con claude --agent team-lead. En la segunda, ejecuta python scripts/monitor/team-monitor.py --watch. Observa los eventos en tiempo real mientras el equipo trabaja.

Ejercicio 3: Agregar métricas de costo (Medio)

Modifica team-monitor.py para que también registre y reporte el costo estimado por agente. Usa el campo cost_usd del output JSON del SDK. Actualiza el dashboard con una columna de costo.

Ejercicio 4: Notificaciones (Medio)

Agrega al orchestrator.py una función que envíe una notificación al terminar la ejecución. Opciones: escribir a un archivo de notificaciones, enviar un webhook a Slack, o simplemente tocar un sonido del sistema con os.system("afplay /System/Library/Sounds/Glass.aiff") (macOS).

Ejercicio 5: Comparar ejecuciones (Difícil)

Modifica metrics.py para que acepte un flag --compare que lea dos archivos de log (de dos ejecuciones diferentes) y compare las métricas lado a lado. Útil para medir el impacto de cambiar system prompts o hooks.

Ejercicio 6: Orquestador con retry (Difícil)

Modifica orchestrator.py para que detecte si el equipo no completó todas las tareas (PARTIAL status en el output), y automáticamente lance una segunda ejecución con las tareas faltantes. El retry debe incluir el contexto de lo que ya se completó.

Troubleshooting

"No agent activity found"

Causa: Los logs no se generaron porque el SubagentStop hook no se disparó o el path del log file es incorrecto.

Solución:

# Verificar que el hook es ejecutable
ls -la scripts/hooks/subagent-stop-log.sh

# Verificar que el log directory existe
mkdir -p logs/agents

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

"El orquestador timeout"

Causa: La ejecución multi-agente tarda más que el timeout del subprocess.

Solución: Incrementa el timeout en el orquestador:

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

"Remote control no envía notificaciones"

Causa: Remote control no está habilitado o la sesión no está vinculada.

Solución:

claude config get remote_control
# Si es false:
claude config set remote_control true

"El dashboard muestra datos incorrectos"

Causa: El log file tiene formato inconsistente o hay líneas corruptas.

Solución: El parser tiene try/except para líneas malformadas. Si persiste, limpia el log file y re-ejecuta.

"El script de métricas importa desde team_monitor pero falla"

Causa: Python no encuentra el módulo porque el path no está configurado.

Solución: Ejecuta desde la raíz del proyecto:

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

Comparación: Sin Monitoring vs Con Monitoring

Aspecto	Sin Monitoring	Con Monitoring
Visibilidad	Solo al terminar	En tiempo real
Métricas	Ninguna	Por agente: tiempo, tareas, eficiencia
Reportes	Manual (copiar output)	Automático (Markdown)
Debugging	Difícil (sin logs)	Logs de actividad por agente
Reproducibilidad	Baja	Alta (logs + reportes)
Remote ops	No posible	Aprobaciones desde celular

Resumen

El team-monitor.py parsea logs de agentes y genera métricas: tareas completadas, duración por agente, promedios, máximos
Tres modos: --watch (tiempo real), --summary (resumen), --report (Markdown completo)
El orchestrator.py ejecuta al team lead via SDK headless, recibe el resultado JSON, y genera reportes automáticamente
Remote control permite aprobar operaciones críticas desde el celular mientras el equipo ejecuta
El dashboard visualiza métricas con gráficos de barras en texto: tiempo y tareas por agente
El script run-team.sh automatiza el workflow completo: limpiar logs → monitorear → ejecutar → generar dashboard
Las métricas clave son: tiempo total, tareas por agente, duración promedio por tarea, y eficiencia (avg/max ratio)
El monitoring transforma la ejecución multi-agente de "caja negra" a "sistema observable"

Recursos Adicionales

Claude Code CLI Reference — Flags -p, --output-format json, --agent para SDK headless
Claude Code Hooks — SubagentStop hook para logging de actividad
Python subprocess Module — Referencia de subprocess para invocación programática
Python argparse Module — CLI argument parsing para scripts de monitor
Claude Code Best Practices — Automatización y monitoring
Claude Code Overview — Contexto de Claude Code como plataforma

Siguiente cápsula: En la cápsula 05 analizarás los resultados de la ejecución completa. ¿Qué agente fue más eficiente? ¿Dónde hubo cuellos de botella? ¿Qué cambiarías para la próxima ejecución? Es la retrospectiva del sistema multi-agente y el cierre completo de la guía.

← Anterior3. Ejecución — Task Board, Delegación Paralela, Quality Gates Siguiente →5. Retrospectiva — Análisis, Optimización, y Cierre de la Guía