s11
Autonomous Agents
CollaborationScan Board, Claim Tasks
499 LOC14 toolsTask board polling + timeout-based self-governance
Teammates scan the board and claim tasks themselves; no need for the lead to assign each one
s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > [ s11 ] s12
"Teammates scan the board and claim tasks themselves" -- no need for the lead to assign each one.
Problem
In s09-s10, 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 (s06), the agent might forget who it is. Identity re-injection fixes this.
Solution
Teammate lifecycle with idle cycle:
+-------+
| spawn |
+---+---+
|
v
+-------+ tool_use +-------+
| WORK | <------------- | LLM |
+---+---+ +-------+
|
| stop_reason != tool_use (or idle tool called)
v
+--------+
| IDLE | poll every 5s for up to 60s
+---+----+
|
+---> check inbox --> message? ----------> WORK
|
+---> scan .tasks/ --> unclaimed? -------> claim -> WORK
|
+---> 60s timeout ----------------------> SHUTDOWN
Identity re-injection after compression:
if len(messages) <= 3:
messages.insert(0, identity_block)
How It Works
- The teammate loop has two phases: WORK and IDLE. When the LLM stops calling tools (or calls
idle), the teammate enters IDLE.
def _loop(self, name, role, prompt):
while True:
# -- WORK PHASE --
messages = [{"role": "user", "content": prompt}]
for _ in range(50):
response = client.messages.create(...)
if response.stop_reason != "tool_use":
break
# execute tools...
if idle_requested:
break
# -- IDLE PHASE --
self._set_status(name, "idle")
resume = self._idle_poll(name, messages)
if not resume:
self._set_status(name, "shutdown")
return
self._set_status(name, "working")
- The idle phase polls inbox and task board in a loop.
def _idle_poll(self, name, messages):
for _ in range(IDLE_TIMEOUT // POLL_INTERVAL): # 60s / 5s = 12
time.sleep(POLL_INTERVAL)
inbox = BUS.read_inbox(name)
if inbox:
messages.append({"role": "user",
"content": f"<inbox>{inbox}</inbox>"})
return True
unclaimed = scan_unclaimed_tasks()
if unclaimed:
claim_task(unclaimed[0]["id"], name)
messages.append({"role": "user",
"content": f"<auto-claimed>Task #{unclaimed[0]['id']}: "
f"{unclaimed[0]['subject']}</auto-claimed>"})
return True
return False # timeout -> shutdown
- Task board scanning: find pending, unowned, unblocked tasks.
def scan_unclaimed_tasks() -> list:
unclaimed = []
for f in sorted(TASKS_DIR.glob("task_*.json")):
task = json.loads(f.read_text())
if (task.get("status") == "pending"
and not task.get("owner")
and not task.get("blockedBy")):
unclaimed.append(task)
return unclaimed
- Identity re-injection: when context is too short (compression happened), insert an identity block.
if len(messages) <= 3:
messages.insert(0, {"role": "user",
"content": f"<identity>You are '{name}', role: {role}, "
f"team: {team_name}. Continue your work.</identity>"})
messages.insert(1, {"role": "assistant",
"content": f"I am {name}. Continuing."})
What Changed From s10
| Component | Before (s10) | After (s11) |
|---|---|---|
| Tools | 12 | 14 (+idle, +claim_task) |
| Autonomy | Lead-directed | Self-organizing |
| Idle phase | None | Poll inbox + task board |
| Task claiming | Manual only | Auto-claim unclaimed tasks |
| Identity | System prompt | + re-injection after compress |
| Timeout | None | 60s idle -> auto shutdown |
Try It
python agents/s11_autonomous_agents.py
Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.Spawn a coder teammate and let it find work from the task board itselfCreate tasks with dependencies. Watch teammates respect the blocked order.- Type
/tasksto see the task board with owners - Type
/teamto monitor who is working vs idle
