Lesson 7 of 20

Long-Running Agents

Progress Tracking

4 min read

Long-running agents need to know where they are, what's done, and what's next. Effective progress tracking isn't just about state—it's about recoverable state.

Three Levels of Tracking

1. Task-Level Progress

Track high-level subtask completion:

class TaskTracker:
    def __init__(self):
        self.tasks = []

    def add_task(self, task_id: str, description: str, dependencies: list = None):
        self.tasks.append({
            "id": task_id,
            "description": description,
            "dependencies": dependencies or [],
            "status": "pending",  # pending, in_progress, completed, failed
            "started_at": None,
            "completed_at": None,
            "result": None
        })

    def start_task(self, task_id: str):
        task = self.get_task(task_id)
        task["status"] = "in_progress"
        task["started_at"] = datetime.now().isoformat()

    def complete_task(self, task_id: str, result: dict):
        task = self.get_task(task_id)
        task["status"] = "completed"
        task["completed_at"] = datetime.now().isoformat()
        task["result"] = result

    def get_next_task(self) -> dict | None:
        """Get next task with satisfied dependencies."""
        for task in self.tasks:
            if task["status"] != "pending":
                continue
            deps_satisfied = all(
                self.get_task(dep)["status"] == "completed"
                for dep in task["dependencies"]
            )
            if deps_satisfied:
                return task
        return None

2. Feature-Level Progress

For development tasks, track features not just tasks:

# features.json - Human-readable progress
{
  "feature": "User Authentication",
  "subtasks": [
    {"name": "Create User model", "status": "completed", "files": ["models/user.py"]},
    {"name": "Add login endpoint", "status": "completed", "files": ["api/auth.py"]},
    {"name": "Add JWT middleware", "status": "in_progress", "files": ["middleware/auth.py"]},
    {"name": "Write tests", "status": "pending", "files": []}
  ],
  "overall_progress": "60%",
  "blockers": [],
  "last_updated": "2025-12-16T10:30:00Z"
}

3. Git-Level Progress

Use version control as a checkpoint system:

import subprocess

class GitCheckpointer:
    def __init__(self, repo_path: str):
        self.repo_path = repo_path

    def checkpoint(self, message: str, task_id: str):
        """Create a checkpoint commit."""
        subprocess.run(["git", "add", "."], cwd=self.repo_path)
        subprocess.run([
            "git", "commit", "-m",
            f"[AGENT] {message}\n\nTask: {task_id}\nTimestamp: {datetime.now().isoformat()}"
        ], cwd=self.repo_path)

    def rollback_to_checkpoint(self, task_id: str):
        """Rollback to last checkpoint for a task."""
        # Find commit by task_id in message
        result = subprocess.run(
            ["git", "log", "--oneline", "--grep", f"Task: {task_id}"],
            capture_output=True, text=True, cwd=self.repo_path
        )
        if result.stdout:
            commit_hash = result.stdout.split()[0]
            subprocess.run(["git", "reset", "--hard", commit_hash], cwd=self.repo_path)
            return True
        return False

    def get_changes_since(self, task_id: str) -> list:
        """Get all files changed since task started."""
        result = subprocess.run(
            ["git", "diff", "--name-only", f"HEAD~10"],
            capture_output=True, text=True, cwd=self.repo_path
        )
        return result.stdout.strip().split("\n")

Progress Reporting

Make progress visible to humans:

def generate_progress_report(tracker: TaskTracker) -> str:
    completed = len([t for t in tracker.tasks if t["status"] == "completed"])
    total = len(tracker.tasks)
    current = next((t for t in tracker.tasks if t["status"] == "in_progress"), None)

    report = f"""
## Progress Report
**Overall**: {completed}/{total} tasks ({completed/total*100:.0f}%)

### Completed
{chr(10).join(f"✓ {t['description']}" for t in tracker.tasks if t["status"] == "completed")}

### In Progress
{"→ " + current["description"] if current else "None"}

### Remaining
{chr(10).join(f"○ {t['description']}" for t in tracker.tasks if t["status"] == "pending")}

**Last updated**: {datetime.now().strftime("%Y-%m-%d %H:%M")}
"""
    return report

Best Practices

Practice Why
Checkpoint after every subtask Minimize lost work on failure
Use git commits as checkpoints Easy rollback, built-in history
Keep progress files human-readable Humans might need to intervene
Track time spent per task Helps estimate remaining work
Log decisions, not just actions Understand why changes were made

Nerd Note: A progress.json file that a human can read and understand is worth more than clever object serialization. When things go wrong at 3 AM, you'll thank yourself.

Next: What happens when things fail—recovery and resumption. :::

Quiz

Module 2: Long-Running Agents

Take Quiz