Task
Pianificazione & CoordinamentoTask Graph + Dependencies
A file-based task graph with ordering, parallelism, and dependencies -- the coordination backbone for multi-agent work
s01 > s02 > s03 > s04 > s05 > s06 | [ s07 ] s08 > s09 > s10 > s11 > s12
"Dividi obiettivi grandi in task piccoli, ordinali, persisti su disco" -- un grafo di task basato su file con dipendenze, gettando le basi per la collaborazione multi-agente.
Problema
Il TodoManager di s03 è una checklist piatta in memoria: nessun ordinamento, nessuna dipendenza, nessuno stato oltre fatto-o-no. Gli obiettivi reali hanno struttura -- il task B dipende dal task A, i task C e D possono eseguire in parallelo, il task E aspetta sia C che D.
Senza relazioni esplicite, l'agente non può dire cosa è pronto, cosa è bloccato, o cosa può eseguire concurrentemente. E poiché la lista vive solo in memoria, la compressione del contesto (s06) la cancella completamente.
Soluzione
Promuovi la checklist in un grafo di task persistito su disco. Ogni task è un file JSON con stato, dipendenze (blockedBy) e dipendenti (blocks). Il grafo risponde a tre domande in qualsiasi momento:
- Cosa è pronto? -- task con stato
pendingeblockedByvuoto. - Cosa è bloccato? -- task in attesa di dipendenze non completate.
- Cosa è fatto? -- task
completed, il cui completamento sblocca automaticamente i dipendenti.
.tasks/
task_1.json {"id":1, "status":"completed"}
task_2.json {"id":2, "blockedBy":[1], "status":"pending"}
task_3.json {"id":3, "blockedBy":[1], "status":"pending"}
task_4.json {"id":4, "blockedBy":[2,3], "status":"pending"}
Grafo dei task (DAG):
+----------+
+--> | task 2 | --+
| | pending | |
+----------+ +----------+ +--> +----------+
| task 1 | | task 4 |
| completed| --> +----------+ +--> | blocked |
+----------+ | task 3 | --+ +----------+
| pending |
+----------+
Ordinamento: task 1 deve finire prima di 2 e 3
Parallelismo: task 2 e 3 possono eseguire contemporaneamente
Dipendenze: task 4 aspetta sia 2 che 3
Stato: pending -> in_progress -> completed
Questo grafo di task diventa la spina dorsale di coordinamento per tutto ciò che viene dopo s07: esecuzione in background (s08), team multi-agente (s09+), e isolamento worktree (s12) leggono e scrivono tutti nella stessa struttura.
Come Funziona
- TaskManager: un file JSON per task, CRUD con grafo delle dipendenze.
class TaskManager:
def __init__(self, tasks_dir: Path):
self.dir = tasks_dir
self.dir.mkdir(exist_ok=True)
self._next_id = self._max_id() + 1
def create(self, subject, description=""):
task = {"id": self._next_id, "subject": subject,
"status": "pending", "blockedBy": [],
"blocks": [], "owner": ""}
self._save(task)
self._next_id += 1
return json.dumps(task, indent=2)
- Risoluzione dipendenze: completare un task rimuove il suo ID dalla lista
blockedBydi ogni altro task, sbloccando automaticamente i dipendenti.
def _clear_dependency(self, completed_id):
for f in self.dir.glob("task_*.json"):
task = json.loads(f.read_text())
if completed_id in task.get("blockedBy", []):
task["blockedBy"].remove(completed_id)
self._save(task)
- Stato + cablaggio dipendenze:
updategestisce transizioni e archi di dipendenza.
def update(self, task_id, status=None,
add_blocked_by=None, add_blocks=None):
task = self._load(task_id)
if status:
task["status"] = status
if status == "completed":
self._clear_dependency(task_id)
self._save(task)
- Quattro strumenti task vanno nella mappa di dispatch.
TOOL_HANDLERS = {
# ...base tools...
"task_create": lambda **kw: TASKS.create(kw["subject"]),
"task_update": lambda **kw: TASKS.update(kw["task_id"], kw.get("status")),
"task_list": lambda **kw: TASKS.list_all(),
"task_get": lambda **kw: TASKS.get(kw["task_id"]),
}
Da s07 in poi, il grafo di task è il default per il lavoro multi-passo. Il Todo di s03 rimane per checklist rapide di una singola sessione.
Cosa è Cambiato da s06
| Componente | Prima (s06) | Dopo (s07) |
|---|---|---|
| Strumenti | 5 | 8 (task_create/update/list/get) |
| Modello pianificazione | Checklist piatta (in memoria) | Grafo di task con dipendenze (su disco) |
| Relazioni | Nessuno | Archi blockedBy + blocks |
| Tracciamento stato | Fatto o no | pending -> in_progress -> completed |
| Persistenza | Perso su compressione | Sopravvive a compressione e riavvii |
Provalo
python agents/s07_task_system.py
Crea 3 task: "Setup project", "Write code", "Write tests". Falli dipendere l'uno dall'altro in ordine.Elenca tutti i task e mostra il grafo delle dipendenzeCompleta il task 1 e poi elenca i task per vedere il task 2 sbloccatoCrea una bacheca task per refactoring: parse -> transform -> emit -> test, dove transform e emit possono eseguire in parallelo dopo parse
