Progress Tracking & Status

In multi-agent pipelines, the lead needs visibility into worker progress to make informed coordination decisions. Should it wait for a slow worker or reassign the task? Has a worker stalled or is it handling a legitimately complex subtask? Is the pipeline on track to complete within the time budget? Progress tracking answers these questions by giving the lead structured status updates from each worker.

File-based progress tracking is the most reliable pattern for Codex workflows. Each worker updates a shared progress file (e.g., .codex/progress.json) with its current status: not_started, in_progress, blocked, completed, or failed. The lead polls this file between delegation rounds to assess pipeline state. This pattern survives agent restarts, timeout recoveries, and partial failures because the state is persisted to disk rather than held in agent memory.

import json
from pathlib import Path
from datetime import datetime

PROGRESS_FILE = Path(".codex/progress.json")

def update_progress(agent: str, task: str, status: str,
                    detail: str = ""):
    """Worker calls this to report status."""
    progress = json.loads(PROGRESS_FILE.read_text())
    progress[agent] = {
        "task": task,
        "status": status,  # not_started|in_progress|completed|failed
        "detail": detail,
        "updated_at": datetime.now().isoformat(),
    }
    PROGRESS_FILE.write_text(json.dumps(progress, indent=2))

def check_pipeline_status() -> dict:
    """Lead calls this to assess overall progress."""
    progress = json.loads(PROGRESS_FILE.read_text())
    summary = {
        "total": len(progress),
        "completed": sum(1 for v in progress.values()
                        if v["status"] == "completed"),
        "failed": sum(1 for v in progress.values()
                      if v["status"] == "failed"),
        "blocked": [k for k, v in progress.items()
                    if v["status"] == "blocked"],
    }
    return summary