task-decomposer

Task Decomposer Skill

Zerlegt uebergeordnete Ziele in atomare, testbare Teilaufgaben mit Dependency-Tracking und Status-Management. Single Source of Truth: tasks.json.

Trigger-Phrases

• "aufgabe zerlegen", "task breakdown", "plan erstellen"
• "decompose task", "split into subtasks", "was sind die schritte"
• "naechste aufgabe", "next task", "was steht an"
• "task status", "fortschritt", "progress"

Assumptions

• tasks.json liegt im Projekt-Root (wird bei Bedarf initialisiert)
• Jede Aufgabe ist atomar genug fuer eine einzelne Claude Code Session
• Abhaengigkeiten bilden einen DAG (keine Zyklen)
• Aufwandsschaetzung in T-Shirt-Sizes: XS (<15min), S (15-30min), M (30-60min), L (1-2h), XL (>2h)

tasks.json Schema

{
  "$schema": "self-improving-agent/tasks-schema",
  "version": "1.0",
  "goal": "Uebergeordnetes Ziel in einem Satz",
  "created_at": "ISO-8601",
  "updated_at": "ISO-8601",
  "tasks": [
    {
      "id": "T001",
      "title": "Kurztitel der Aufgabe",
      "description": "Detaillierte Beschreibung was genau getan werden muss",
      "acceptance_criteria": [
        "Kriterium 1: Pruefbar formuliert",
        "Kriterium 2: ..."
      ],
      "dependencies": ["T000"],
      "status": "pending",
      "effort": "M",
      "priority": 1,
      "assigned_skills": ["tdd", "test-validator"],
      "notes": "",
      "started_at": null,
      "completed_at": null,
      "blocked_reason": null
    }
  ]
}

Status-Uebergaenge

pending --> in_progress --> done
  |              |
  v              v
blocked       blocked
  |              |
  v              v
pending      in_progress

• pending: Noch nicht begonnen
• in_progress: Wird gerade bearbeitet (maximal 1 Task gleichzeitig)
• done: Alle Akzeptanzkriterien erfuellt
• blocked: Kann nicht fortgesetzt werden (Grund in blocked_reason)

Prozedur

Modus 1: Neues Ziel zerlegen

Eingabe: Ein uebergeordnetes Ziel als Freitext.

Schritt 1: Ziel analysieren

• Was ist der gewuenschte Endzustand?
• Welche Teilbereiche sind betroffen (Code, Tests, Doku, Config)?
• Gibt es harte Constraints oder Abhaengigkeiten?

Schritt 2: Top-Down-Zerlegung

1. Zerlege das Ziel in 3-7 Hauptbloecke (epics)
2. Zerlege jeden Block in atomare Aufgaben
3. Atomar = in einer Claude Code Session abschliessbar, mit klarem Done-Kriterium
4. Reihenfolge: Fundament zuerst (Datenmodell, Interfaces), dann Implementation, dann Tests, dann Integration

Schritt 3: Abhaengigkeiten bestimmen

Fuer jede Aufgabe: Welche anderen Aufgaben muessen VORHER erledigt sein?

Validiere den Dependency-Graph:

#!/usr/bin/env python3
"""Validate task dependency graph is a DAG (no cycles)."""
import json

def validate_dag(tasks_file="tasks.json"):
    with open(tasks_file) as f:
        data = json.load(f)

    task_ids = {t["id"] for t in data["tasks"]}
    deps = {t["id"]: set(t.get("dependencies", [])) for t in data["tasks"]}

    # Check for references to non-existent tasks
    for tid, task_deps in deps.items():
        missing = task_deps - task_ids
        if missing:
            print(f"ERROR: Task {tid} depends on non-existent: {missing}")
            return False

    # Topological sort to detect cycles
    visited = set()
    in_stack = set()

    def has_cycle(node):
        visited.add(node)
        in_stack.add(node)
        for dep in deps.get(node, []):
            if dep in in_stack:
                print(f"ERROR: Cycle detected involving {node} -> {dep}")
                return True
            if dep not in visited and has_cycle(dep):
                return True
        in_stack.discard(node)
        return False

    for tid in task_ids:
        if tid not in visited:
            if has_cycle(tid):
                return False

    print("OK: Dependency graph is a valid DAG")
    return True

if __name__ == "__main__":
    validate_dag()

Schritt 4: Priorisierung

Berechne Prioritaet basierend auf:

priority_score = dependency_depth * 10 + effort_weight + critical_path_bonus

effort_weight:
  XS: 1, S: 2, M: 3, L: 5, XL: 8

critical_path_bonus:
  +10 wenn die Aufgabe auf dem kritischen Pfad liegt
  (= andere Aufgaben sind davon abhaengig)

Niedrigerer Score = hoehere Prioritaet (wird frueher bearbeitet).

Schritt 5: tasks.json schreiben

Speichere das Ergebnis als tasks.json im Projekt-Root.

Modus 2: Naechste Aufgabe identifizieren

Wird vom Orchestrator aufgerufen um die naechste auszufuehrende Aufgabe zu bestimmen.

Algorithmus:

def get_next_task(tasks):
    """Return the highest-priority task that is ready to execute."""
    done_ids = {t["id"] for t in tasks if t["status"] == "done"}
    in_progress = [t for t in tasks if t["status"] == "in_progress"]

    # Maximal 1 Task gleichzeitig
    if in_progress:
        return in_progress[0]  # Aktuellen Task fortsetzen

    candidates = []
    for task in tasks:
        if task["status"] != "pending":
            continue
        # Alle Abhaengigkeiten muessen done sein
        deps_met = all(d in done_ids for d in task.get("dependencies", []))
        if deps_met:
            candidates.append(task)

    if not candidates:
        blocked = [t for t in tasks if t["status"] == "blocked"]
        if blocked:
            return None  # Alle verbleibenden Tasks sind blockiert
        return None  # Alle Tasks erledigt

    # Sortiere nach Prioritaet (niedrigster Score zuerst)
    candidates.sort(key=lambda t: t.get("priority", 999))
    return candidates[0]

Output:

{
  "next_task": { "...task object..." },
  "progress": {
    "total": 12,
    "done": 5,
    "in_progress": 0,
    "pending": 6,
    "blocked": 1
  },
  "estimated_remaining": "3h 30min"
}

Modus 3: Task-Status aktualisieren

# Task starten
# Setze status: "in_progress", started_at: now()

# Task abschliessen
# Setze status: "done", completed_at: now()
# Pruefe: Werden dadurch blockierte Tasks freigegeben?

# Task blockieren
# Setze status: "blocked", blocked_reason: "..."

Aktualisiere tasks.json nach jeder Statusaenderung. Setze updated_at auf den aktuellen Zeitstempel.

Modus 4: Fortschrittsbericht

Generiere einen kompakten Fortschrittsbericht:

## Fortschritt: {GOAL}

{PROGRESS_BAR} {DONE}/{TOTAL} ({PERCENT}%)

### Erledigt
- [x] T001: {TITLE} (XS, 12min)
- [x] T002: {TITLE} (M, 45min)

### In Arbeit
- [ ] T003: {TITLE} (S, seit 10min)

### Naechste
- [ ] T004: {TITLE} (M, wartet auf T003)

### Blockiert
- [!] T007: {TITLE} — {BLOCKED_REASON}

### Schaetzung
Verbleibend: ~{REMAINING_EFFORT} | Bisherige Dauer: {ELAPSED}

Error-Handling

Fehlerfall	Reaktion
tasks.json nicht gefunden	Initialisiere leere tasks.json mit Schema
Korrupte tasks.json	Versuche Backup .agent-memory/tasks.json.backup zu laden
Zyklische Abhaengigkeiten	Fehler melden, betroffene Tasks auflisten
Alle Tasks blockiert	Blockierte Tasks auflisten, Loesungsvorschlaege machen
Kein Ziel angegeben	Nachfragen: "Was ist das uebergeordnete Ziel?"
Task-ID existiert bereits	Naechste freie ID vergeben (T{MAX+1})

Backup-Strategie

Vor jeder Aenderung an tasks.json:

mkdir -p .agent-memory
cp tasks.json .agent-memory/tasks.json.backup

Falls cp fehlschlaegt (z.B. Berechtigungsproblem): Warnung ausgeben, aber Hauptoperation trotzdem fortsetzen — Backup ist nice-to-have, nicht blocking.

Abgrenzung

• iteration-logger: Dokumentiert WAS bei einem Task passiert ist. Task-Decomposer verwaltet WELCHE Tasks existieren und ihren Status.
• context-keeper: Speichert Architektur-Entscheidungen. Task-Decomposer speichert Aufgaben-Planung.
• self-improve Skill: Orchestriert den Gesamtloop. Task-Decomposer ist ein Baustein davon.