7. מערכת משימות

הבעיה

ה-TodoManager של סשן TodoWrite הוא רשימת תיוגים שטוחה בזיכרון: ללא סדר, ללא תלויות, ללא סטטוס מעבר ל-done-or-not. למטרות אמיתיות יש מבנה — משימה B תלויה במשימה A, משימות C ו-D יכולות לרוץ במקביל, משימה E מחכה לשתי C ו-D.

ללא קשרים מפורשים, הסוכן לא יכול לדעת מה מוכן, מה חסום, או מה יכול לרוץ בו-זמנית. ומכיוון שהרשימה חיה רק בזיכרון, דחיסת ה-context (סשן דחיסת הקשר) מוחקת אותה לחלוטין.

הפתרון

קדמו את רשימת התיוגים לגרף משימות השמור לדיסק. כל משימה היא קובץ JSON עם סטטוס ותלויות (blockedBy). הגרף עונה על שלוש שאלות בכל רגע:

• מה מוכן? — משימות עם סטטוס pending ו-blockedBy ריק
• מה חסום? — משימות שמחכות לתלויות שלא הסתיימו
• מה נסגר? — משימות completed, שסיומן מבטל אוטומטית חסימת תלויות

.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"}

Task graph (DAG):
  1 --> 2 --> 4
  1 --> 3 --> 4

When task 1 completes: 2 and 3 become ready.
When 2 and 3 complete: 4 becomes ready.

איך זה עובד

1. כל משימה היא קובץ JSON. ה-TaskManager קורא את כל הקבצים ובונה את הגרף בזיכרון.

import json
from pathlib import Path

class TaskManager:
    def __init__(self, tasks_dir: str = ".tasks"):
        self.dir = Path(tasks_dir)
        self.dir.mkdir(exist_ok=True)

    def _load_all(self) -> list:
        tasks = []
        for f in sorted(self.dir.glob("task_*.json")):
            tasks.append(json.loads(f.read_text()))
        return tasks

    def create(self, title: str, blocked_by: list = None) -> dict:
        tasks = self._load_all()
        new_id = max((t["id"] for t in tasks), default=0) + 1
        task = {"id": new_id, "title": title,
                "status": "pending", "blockedBy": blocked_by or []}
        path = self.dir / f"task_{new_id}.json"
        path.write_text(json.dumps(task, indent=2))
        return task

    def complete(self, task_id: int) -> str:
        path = self.dir / f"task_{task_id}.json"
        task = json.loads(path.read_text())
        task["status"] = "completed"
        path.write_text(json.dumps(task, indent=2))
        return f"Task {task_id} completed."

    def ready(self) -> list:
        tasks = self._load_all()
        done_ids = {t["id"] for t in tasks if t["status"] == "completed"}
        return [
            t for t in tasks
            if t["status"] == "pending"
            and all(dep in done_ids for dep in t.get("blockedBy", []))
        ]

2. הסוכן מקבל שלושה כלים: create_task, complete_task ו-list_ready_tasks.

TASK_MANAGER = TaskManager()

TOOL_HANDLERS = {
    # ...base tools...
    "create_task": lambda **kw: str(TASK_MANAGER.create(
        kw["title"], kw.get("blocked_by", []))),
    "complete_task": lambda **kw: TASK_MANAGER.complete(kw["task_id"]),
    "list_ready_tasks": lambda **kw: json.dumps(TASK_MANAGER.ready(), indent=2),
}

3. תהליך העבודה של הסוכן הופך ל: יצירת גרף המשימות, ואז עבודה דרך משימות מוכנות אחת אחת.

Agent: "I need to refactor auth module."

1. create_task("Read current auth code")            -> id=1
2. create_task("Write new AuthManager",  blocked_by=[1]) -> id=2
3. create_task("Update tests", blocked_by=[2])      -> id=3
4. create_task("Update docs",  blocked_by=[2])      -> id=4

list_ready_tasks() -> [task 1]

[works on task 1]
complete_task(1)

list_ready_tasks() -> [task 2]
...

מה השתנה מ-דחיסת הקשר

מסקנה מרכזית

שמירת משימות לדיסק היא מה שהופך את תוכניות הסוכן לעמידות. גרף המשימות מקודד לא רק מה לעשות אלא באיזה סדר ומה יכול לרוץ במקביל. בשילוב עם דחיסת הקשר, הסוכן יכול לעבוד על מטרות גדולות באמת לאורך סיבובים רבים מבלי לאבד עקב.