11. Autonomous Agents

"Teammates scan the board and claim tasks themselves"

30 min read
πŸ’‘New to this?

What is task claiming?

When a teammate atomically changes a task's status from 'pending' to 'in_progress' and writes its own name as the assignee. Because each task is a separate JSON file, two agents can't claim the same task simultaneously β€” one will win, the other retries.

What is identity re-injection?

After context compression (Context Compact session) wipes the conversation history, the agent might forget who it is. The harness injects an <identity> block at the start of the compressed context: 'You are Alice, a backend specialist, working on task 3.' This restores the agent's sense of self.

What is an idle cycle?

When a teammate finds no ready tasks, it waits briefly (sleeps 2 seconds) and then checks again. This polling loop is the mechanism that lets teammates work indefinitely without the lead assigning each task individually.

The Problem

In the Agent Teams through Team Protocols sessions, teammates only work when explicitly told to. The lead must spawn each one with a specific prompt. 10 unclaimed tasks on the board? The lead assigns each one manually. Doesn’t scale.

True autonomy: teammates scan the task board themselves, claim unclaimed tasks, work on them, then look for more.

One subtlety: after context compression (Context Compact session), the agent might forget who it is. Identity re-injection fixes this.

The Solution

Teammate lifecycle with idle cycle:

+-------+
| spawn |
+---+---+
    |
    v
+-------+   tool_use     +-------+
| WORK  | <------------- |  LLM  |
+---+---+                +-------+
    |                       ^
    | done or no tasks      |
    v                       |
+-------+   poll tasks      |
| IDLE  | --check_ready()---+
+-------+   (wait 2s)
    |
    | shutdown_req received
    v
+----------+
| SHUTDOWN |
+----------+

How It Works

  1. Claiming a task is atomic β€” write owner + status together.
def claim_task(name: str) -> dict | None:
    """Find and claim the first ready task. Returns the task or None."""
    tasks_dir = Path(".tasks")
    ready = get_ready_tasks()  # reads all task files
    if not ready:
        return None

    task = ready[0]
    task_path = tasks_dir / f"task_{task['id']}.json"

    # Atomic claim: read current state, update only if still pending
    current = json.loads(task_path.read_text())
    if current["status"] != "pending":
        return None  # someone else claimed it

    current["status"] = "in_progress"
    current["assignee"] = name
    task_path.write_text(json.dumps(current, indent=2))
    return current
  1. The teammate loop polls for tasks and works through them.
def autonomous_teammate(name: str, role: str) -> None:
    system = build_system_with_identity(name, role)

    while True:
        # Check inbox for shutdown requests
        process_inbox(name)
        if get_status(name) == "SHUTDOWN":
            break

        task = claim_task(name)
        if task is None:
            # No ready tasks β€” idle cycle
            update_status(name, "IDLE")
            time.sleep(2)
            continue

        update_status(name, "WORKING")
        history = [{
            "role": "user",
            "content": f"Work on task {task['id']}: {task['title']}\n{task.get('description','')}"
        }]
        run_agent_with_identity(history, system, name, task["id"])
        complete_task(task["id"])
  1. Identity re-injection wraps the hard_compact function from the Context Compact session.
def build_identity_block(name: str, role: str, task_id: int) -> str:
    return (
        f"<identity>\n"
        f"You are {name}, a {role}.\n"
        f"You are currently working on task {task_id}.\n"
        f"After completing it, claim another ready task from .tasks/.\n"
        f"</identity>"
    )

def hard_compact_with_identity(messages: list, name: str, role: str, task_id: int) -> list:
    # Regular hard compact...
    compacted = hard_compact(messages)
    # Prepend identity to the summary
    identity = build_identity_block(name, role, task_id)
    first_msg = compacted[0]
    first_msg["content"] = identity + "\n\n" + first_msg["content"]
    return compacted

What Changed From Team Protocols

ComponentBefore (Team Protocols)After (Autonomous Agents)
Task assignmentLead sends explicitlyTeammates claim autonomously
Idle stateWait for messagePoll ready tasks every 2s
IdentityNo special handlingRe-injected after compression
Lead roleCoordinator + assignerCoordinator only (board-level view)

Key Takeaway

Autonomous task claiming is what separates a team from a relay race. Each teammate is a self-directed agent that scans the board, claims what it can do, works, and loops. The lead’s job shrinks to creating tasks and handling protocols. Identity re-injection is the glue that keeps the agent coherent across compression boundaries.

Interactive Code Walkthrough

Idle Cycle and Autonomous Task Claiming
1def claim_task(name: str) -> dict | None:
2 ready = get_ready_tasks()
3 if not ready:
4 return None
5 task = ready[0]
6 task_path = Path(".tasks") / f"task_{task['id']}.json"
7 current = json.loads(task_path.read_text())
8 if current["status"] != "pending":
9 return None # someone else claimed it
10 current["status"] = "in_progress"
11 current["assignee"] = name
12 task_path.write_text(json.dumps(current, indent=2))
13 return current
14 
15def autonomous_teammate(name: str, role: str) -> None:
16 system = build_system_with_identity(name, role)
17 while True:
18 process_inbox(name)
19 if get_status(name) == "SHUTDOWN":
20 break
21 task = claim_task(name)
22 if task is None:
23 update_status(name, "IDLE")
24 time.sleep(2)
25 continue
26 update_status(name, "WORKING")
27 history = [{"role": "user", "content":
28 f"Work on task {task['id']}: {task['title']}"}]
29 run_agent_with_identity(history, system, name, task["id"])
30 complete_task(task["id"])
31 
claim_task() first finds all ready tasks. If none exist, it returns None immediately β€” the caller will put the agent into idle mode.
Step 1 of 5
πŸ§ͺ Try it yourself
πŸ”₯ Warm-up ~5 min

What happens if two autonomous agents try to claim the same task simultaneously? How would you prevent race conditions?

Hint

File-based locking: the first agent to write its name to the task file 'owns' it. Use atomic file operations.

πŸ”¨ Build ~20 min

Create 5 tasks on a board, spawn 2 autonomous agents, and watch them claim and complete tasks independently.

Hint

Each agent polls the board, filters for ready + unclaimed tasks, and claims one.

πŸš€ Stretch ~45 min

Implement work stealing: when an agent finishes all its tasks and the board is empty but other agents are still working, it offers to help by splitting a large in-progress task.

Hint

Add a 'splittable' flag to tasks and a 'split_task' function that creates subtasks.

Found a mistake? Report it β†’