"Break big goals into small tasks, order them, persist" — File-persisted task graph, the foundation for multi-agent collaboration.
Harness Layer: Tasks — Persisted goals, recoverable progress.
The Problem
s05's TodoWrite lets an agent record the steps of its current task. Each checklist item has content and a status, helping the agent keep track of what remains.
When a project is split into three tasks—creating database tables, writing an API, and adding tests—the Harness also needs to know how they relate: the API must wait for the database tables, and the tests must wait for a stable API. It also needs to record who is responsible for each task.
TodoWrite does not record these dependencies or assignments. It can show that "write the API" is unfinished, but the Harness cannot use that information to decide whether the task is ready to start.
This chapter adds a Task System. Each task has its own ID and status; blockedBy records prerequisites, and owner records the agent responsible for the task.
The Solution
The code keeps S04's five base tools, Permission, Hooks, and shared execute_tool, then adds 6 task tools, persistence in the .tasks/ directory, and blockedBy dependency checks.
TodoWrite vs Task System:
| TodoWrite (s05) | Task System (s10) | |
|---|---|---|
| Role | Execution checklist for the current task | Recoverable task system |
| Storage | In-process / session state | .tasks/{id}.json |
| Dependencies | None | blockedBy dependency graph |
| Lifecycle | Current session / current task | Cross-session |
| Coordination | No task claiming | owner / claim |
| Status | pending / in_progress / completed | pending / in_progress / completed |
| Granularity | The agent's own steps | Tasks that can be claimed, tracked, and unblocked |
| Update contract | Replace the whole checklist | Create/get/update/list individual records |
How It Works
Task: Data Structure
Each task is a JSON file, stored in the .tasks/ directory:
@dataclass
class Task:
id: str
subject: str
description: str
status: str # pending | in_progress | completed
owner: str | None # Agent responsible for this task
blockedBy: list[str] # List of dependency task IDsIDs use the task_ prefix followed by 8 random hexadecimal characters. Files are created exclusively; an existing ID is discarded and regenerated.
TaskStore validates task IDs and reads and writes the JSON files. TASKS = TaskStore(TASKS_DIR) is the store used by this chapter.
create_task: Create Tasks
def create_task(subject: str, description: str = "") -> Task:
return TASKS.create(subject, description)TaskStore.create checks the subject, allocates a random ID, and writes .tasks/{id}.json. A new task always starts with an empty blockedBy list. The tool result returns the runtime-generated ID to the model.
update_task: Add Dependencies with Returned IDs
def update_task(task_id: str, addBlockedBy: list[str]) -> Task:
return TASKS.update_dependencies(task_id, addBlockedBy)Task graph construction uses two phases: create every node first, then call update_task with the IDs returned by create_task to add edges. This matters when the model emits several tool calls in one response: sibling calls are formed before any tool result exists, so one create_task call cannot consume another call's newly generated ID.
update_task validates the entire change before saving it. The target and dependencies must exist, the target must still be pending and unowned, and the new edges must not introduce self-dependencies or cycles. Repeating an existing edge is safe and does not duplicate it.
can_start: Dependency Check
A task can only start after all its blockedBy dependencies are completed:
def can_start(task_id: str) -> bool:
return not incomplete_dependencies(load_task(task_id))incomplete_dependencies loads each prerequisite. A task cannot be claimed if any prerequisite is not completed or its file no longer exists.
claim_task: Claim a Task
When the agent starts working on a task, it calls claim_task: sets owner, changes status from pending → in_progress. The owner field records who claimed the task:
def claim_task(task_id: str, owner: str = "agent") -> str:
task = load_task(task_id)
if task.status != "pending":
return f"Task {task_id} is {task.status}, cannot claim"
dependencies = incomplete_dependencies(task)
if dependencies:
return f"Blocked by: {dependencies}"
task.owner = owner
task.status = "in_progress"
TASKS.save(task)
return f"Claimed {task_id} ({task.subject})"The claim is rejected if the task is not pending or its dependencies are incomplete. S10 only updates task state sequentially.
complete_task: Complete and Unblock
When a task is done, set it to completed. Simultaneously scan all other tasks to find downstream tasks that were just unblocked:
def complete_task(task_id: str, owner: str = "agent") -> str:
task = load_task(task_id)
if task.status != "in_progress":
return f"Task {task_id} is {task.status}, cannot complete"
if task.owner != owner:
return f"Task {task_id} is owned by {task.owner}, not {owner}"
ready_before = {t.id for t in list_tasks()
if t.status == "pending" and t.blockedBy
and can_start(t.id)}
task.status = "completed"
TASKS.save(task)
unblocked = [t.subject for t in list_tasks()
if t.status == "pending" and t.blockedBy
and t.id not in ready_before
and can_start(t.id)]
msg = f"Completed {task_id} ({task.subject})"
if unblocked:
msg += f"\nUnblocked: {', '.join(unblocked)}"
return msgAfter completing "schema", can_start returns True for "endpoints" and "docs"; they can begin.
get_task: View Full Details
list_tasks only shows a one-line summary. get_task returns the full task JSON, including description and dependency details. When recovering across sessions, the agent needs to read the full description to continue work:
def get_task(task_id: str) -> str:
task = load_task(task_id)
return json.dumps(asdict(task), indent=2)State Machine: Two Actions, Three States
pending ──claim──→ in_progress ──complete──→ completed
Here claim / complete are actions, while pending / in_progress / completed are states:
- claim_task:
pending→in_progress. Sets owner, begins work. - complete_task:
in_progress→completed. Marks the task done and unblocks downstream.
Putting It Together
# Phase 1: create every node and receive its runtime ID
schema = create_task("setup database schema")
endpoints = create_task("create API endpoints")
tests = create_task("write tests")
docs = create_task("write docs")
# Phase 2: add edges using those returned IDs
update_task(endpoints.id, addBlockedBy=[schema.id])
update_task(tests.id, addBlockedBy=[endpoints.id])
update_task(docs.id, addBlockedBy=[schema.id])
# Agent claims the first available task
claim_task(schema.id) # ✓ Claimed (no dependencies)
complete_task(schema.id) # ✓ Completed → unblocks endpoints, docs
claim_task(endpoints.id) # ✓ Claimed (schema completed)
complete_task(endpoints.id) # ✓ Completed → unblocks tests
claim_task(docs.id) # ✓ Claimed (schema completed)
complete_task(docs.id) # ✓ Completed
claim_task(tests.id) # ✓ Claimed (endpoints completed)
complete_task(tests.id) # ✓ CompletedEach create_task writes a JSON file; update_task, claim_task, and complete_task update it. Across sessions, the .tasks/ directory persists — the agent reads the files to recover progress.
Try It
cd learn-claude-code
python s10_task_system/code.pyTry these prompts:
Create tasks: setup database schema, create API endpoints (depends on schema), write tests (depends on endpoints), write docs (depends on schema)List all tasks and their statusesClaim the first unblocked task and complete itList tasks again — which ones are now unblocked?
What to observe: Are JSON files generated in the .tasks/ directory? After completing a task, are the blocked tasks unblocked?
#!/usr/bin/env python3
"""
s10_task_system.py - Task System
.tasks/
task_a1b2c3d4.json {status: completed, blockedBy: []}
task_e5f6a7b8.json {status: pending, blockedBy: [task_a1b2c3d4]}
task_11223344.json {status: pending, blockedBy: [task_e5f6a7b8]}
Dependency graph:
+-----------+ +-----------+ +-----------+
| schema | ---> | API | ---> | tests |
| completed | | pending | | pending |
+-----------+ +-----------+ +-----------+
can_start(API) is true because schema is completed.
Task lifecycle:
pending --claim_task--> in_progress --complete_task--> completed
"""
import glob
import json
import os
import re
import secrets
import subprocess
from dataclasses import asdict, dataclass
from pathlib import Path
try:
import readline
readline.parse_and_bind("set bind-tty-special-chars off")
readline.parse_and_bind("set input-meta on")
readline.parse_and_bind("set output-meta on")
readline.parse_and_bind("set convert-meta off")
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv(override=True)
if os.getenv("ANTHROPIC_BASE_URL"):
os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)
WORKDIR = Path.cwd()
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL = os.environ["MODEL_ID"]
ENVIRONMENT_PROMPT = (
"Windows: the bash tool runs through cmd.exe; use cmd.exe syntax, not Unix "
"Bash or PowerShell syntax, and prefer dedicated file tools for file operations"
if os.name == "nt"
else "Unix-like: the bash tool runs the system shell"
)
SYSTEM = (
f"You are a coding agent at {WORKDIR}. Environment: {ENVIRONMENT_PROMPT}. "
"Use task tools to track dependencies and progress. Create all task nodes "
"first. After create_task returns runtime-generated IDs, use update_task "
"with those exact IDs to add dependencies."
)
# -- New in s10: persistent task records --
TASKS_DIR = WORKDIR / ".tasks"
TASK_ID_PATTERN = re.compile(r"^task_[0-9a-f]{8}$")
@dataclass
class Task:
id: str
subject: str
description: str
status: str
owner: str | None
blockedBy: list[str]
class TaskStore:
def __init__(self, directory: Path):
self.directory = directory
def _root(self, create: bool = False) -> Path:
if create:
self.directory.mkdir(parents=True, exist_ok=True)
root = self.directory.resolve()
if not root.is_relative_to(WORKDIR.resolve()):
raise ValueError("Task store escapes the workspace")
return root
def _path(self, task_id: str, create_root: bool = False) -> Path:
if not isinstance(task_id, str) or not TASK_ID_PATTERN.fullmatch(task_id):
raise ValueError(f"Invalid task ID: {task_id!r}")
root = self._root(create=create_root)
path = (root / f"{task_id}.json").resolve()
if not path.is_relative_to(root):
raise ValueError(f"Invalid task ID: {task_id!r}")
return path
def exists(self, task_id: str) -> bool:
return self._path(task_id).is_file()
def create(self, subject: str, description: str = "") -> Task:
subject = subject.strip()
if not subject:
raise ValueError("Task subject cannot be empty")
self._root(create=True)
for _ in range(100):
task = Task(
id=f"task_{secrets.token_hex(4)}",
subject=subject,
description=description,
status="pending",
owner=None,
blockedBy=[],
)
try:
with self._path(task.id, create_root=True).open(
"x", encoding="utf-8"
) as handle:
json.dump(asdict(task), handle, indent=2)
return task
except FileExistsError:
continue
raise RuntimeError("Could not allocate a unique task ID")
def _depends_on(self, task_id: str, target_id: str) -> bool:
"""Return whether task_id transitively depends on target_id."""
pending = [task_id]
visited = set()
while pending:
current = pending.pop()
if current == target_id:
return True
if current in visited:
continue
visited.add(current)
pending.extend(self.load(current).blockedBy)
return False
def update_dependencies(self, task_id: str,
add_blocked_by: list[str]) -> Task:
if not isinstance(add_blocked_by, list):
raise ValueError("addBlockedBy must be a list of task IDs")
task = self.load(task_id)
if task.status != "pending" or task.owner is not None:
raise ValueError(
f"Task {task_id} dependencies can only be updated while "
"pending and unowned"
)
dependencies = list(dict.fromkeys(add_blocked_by))
for dependency in dependencies:
if dependency == task_id:
raise ValueError("Task cannot depend on itself")
if not self.exists(dependency):
raise ValueError(f"Dependency not found: {dependency}")
if dependency not in task.blockedBy and self._depends_on(
dependency, task_id
):
raise ValueError(
f"Dependency cycle detected: {task_id} -> {dependency}"
)
task.blockedBy.extend(
dependency for dependency in dependencies
if dependency not in task.blockedBy
)
self.save(task)
return task
def save(self, task: Task) -> None:
self._path(task.id, create_root=True).write_text(
json.dumps(asdict(task), indent=2),
encoding="utf-8",
)
def load(self, task_id: str) -> Task:
data = json.loads(self._path(task_id).read_text(encoding="utf-8"))
task = Task(**data)
if task.id != task_id:
raise ValueError(f"Task file ID does not match {task_id}")
if task.status not in ("pending", "in_progress", "completed"):
raise ValueError(f"Invalid task status: {task.status}")
return task
def list(self) -> list[Task]:
if not self.directory.exists():
return []
root = self._root()
return [self.load(path.stem)
for path in sorted(root.glob("task_*.json"))]
TASKS = TaskStore(TASKS_DIR)
def create_task(subject: str, description: str = "") -> Task:
return TASKS.create(subject, description)
def update_task(task_id: str, addBlockedBy: list[str]) -> Task:
return TASKS.update_dependencies(task_id, addBlockedBy)
def load_task(task_id: str) -> Task:
return TASKS.load(task_id)
def list_tasks() -> list[Task]:
return TASKS.list()
def get_task(task_id: str) -> str:
return json.dumps(asdict(load_task(task_id)), indent=2)
def incomplete_dependencies(task: Task) -> list[str]:
incomplete = []
for dependency in task.blockedBy:
try:
if load_task(dependency).status != "completed":
incomplete.append(dependency)
except (FileNotFoundError, ValueError):
incomplete.append(dependency)
return incomplete
def can_start(task_id: str) -> bool:
return not incomplete_dependencies(load_task(task_id))
def claim_task(task_id: str, owner: str = "agent") -> str:
task = load_task(task_id)
if task.status != "pending":
return f"Task {task_id} is {task.status}, cannot claim"
dependencies = incomplete_dependencies(task)
if dependencies:
return f"Blocked by: {dependencies}"
task.owner = owner
task.status = "in_progress"
TASKS.save(task)
print(f" [claim] {task.subject} -> in_progress (owner: {owner})")
return f"Claimed {task.id} ({task.subject})"
def complete_task(task_id: str, owner: str = "agent") -> str:
task = load_task(task_id)
if task.status != "in_progress":
return f"Task {task_id} is {task.status}, cannot complete"
if task.owner != owner:
return f"Task {task_id} is owned by {task.owner}, not {owner}"
ready_before = {
candidate.id
for candidate in list_tasks()
if candidate.status == "pending"
and candidate.blockedBy
and can_start(candidate.id)
}
task.status = "completed"
TASKS.save(task)
unblocked = [candidate.subject for candidate in list_tasks()
if candidate.status == "pending"
and candidate.blockedBy
and candidate.id not in ready_before
and can_start(candidate.id)]
print(f" [complete] {task.subject}")
message = f"Completed {task.id} ({task.subject})"
if unblocked:
message += f"\nUnblocked: {', '.join(unblocked)}"
print(f" [unblocked] {', '.join(unblocked)}")
return message
# -- From s04: tool implementations --
def run_bash(command: str) -> str:
try:
result = subprocess.run(
command,
shell=True,
cwd=WORKDIR,
capture_output=True,
text=True, errors="replace",
timeout=120,
)
output = (result.stdout + result.stderr).strip()
return output[:50000] if output else "(no output)"
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)"
def run_read(path: str, limit: int | None = None) -> str:
try:
lines = (WORKDIR / path).resolve().read_text(encoding="utf-8").splitlines()
if limit and limit < len(lines):
lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]
return "\n".join(lines)
except Exception as error:
return f"Error: {error}"
def run_write(path: str, content: str) -> str:
try:
file_path = (WORKDIR / path).resolve()
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(content, encoding="utf-8")
return f"Wrote {len(content)} bytes to {path}"
except Exception as error:
return f"Error: {error}"
def run_edit(path: str, old_text: str, new_text: str) -> str:
try:
file_path = (WORKDIR / path).resolve()
text = file_path.read_text(encoding="utf-8")
if old_text not in text:
return f"Error: text not found in {path}"
file_path.write_text(text.replace(old_text, new_text, 1), encoding="utf-8")
return f"Edited {path}"
except Exception as error:
return f"Error: {error}"
def run_glob(pattern: str) -> str:
try:
matches = sorted({
match
for match in glob.glob(pattern, root_dir=WORKDIR, recursive=True)
if (WORKDIR / match).resolve().is_relative_to(WORKDIR)
})
shown = matches[:200]
if len(matches) > 200:
shown.append("... (more matches omitted; narrow the pattern)")
return "\n".join(shown) if shown else "(no matches)"
except Exception as error:
return f"Error: {error}"
def run_create_task(subject: str, description: str = "") -> str:
task = create_task(subject, description)
print(f" [create] {task.subject}")
return f"Created {task.id}: {task.subject}"
def run_update_task(task_id: str, addBlockedBy: list[str]) -> str:
task = update_task(task_id, addBlockedBy)
dependencies = ", ".join(task.blockedBy) or "(none)"
print(f" [update] {task.subject} blockedBy: {dependencies}")
return f"Updated {task.id} blockedBy: {dependencies}"
def run_list_tasks() -> str:
tasks = list_tasks()
if not tasks:
return "No tasks. Use create_task to add some."
lines = []
for task in tasks:
marker = {
"pending": "[ ]",
"in_progress": "[>]",
"completed": "[x]",
}.get(task.status, "[?]")
dependencies = (
f" (blockedBy: {', '.join(task.blockedBy)})"
if task.blockedBy else ""
)
owner = f" [{task.owner}]" if task.owner else ""
lines.append(
f"{marker} {task.id}: {task.subject} "
f"[{task.status}]{owner}{dependencies}"
)
return "\n".join(lines)
def run_get_task(task_id: str) -> str:
return get_task(task_id)
def run_claim_task(task_id: str) -> str:
return claim_task(task_id, owner="agent")
def run_complete_task(task_id: str) -> str:
return complete_task(task_id, owner="agent")
TOOLS = [
{"name": "bash", "description": "Run a shell command.",
"input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
{"name": "read_file", "description": "Read file contents.",
"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
{"name": "write_file", "description": "Write content to a file.",
"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
{"name": "edit_file", "description": "Replace exact text in a file once.",
"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
{"name": "glob", "description": "Find files matching a glob pattern; ** matches recursively.",
"input_schema": {"type": "object", "properties": {"pattern": {"type": "string"}}, "required": ["pattern"]}},
{"name": "create_task", "description": "Create a task and return its runtime-generated ID.",
"input_schema": {"type": "object", "properties": {"subject": {"type": "string"}, "description": {"type": "string"}}, "required": ["subject"], "additionalProperties": False}},
{"name": "update_task", "description": "Add dependencies using IDs returned by create_task.",
"input_schema": {"type": "object", "properties": {"task_id": {"type": "string", "pattern": "^task_[0-9a-f]{8}$"}, "addBlockedBy": {"type": "array", "items": {"type": "string", "pattern": "^task_[0-9a-f]{8}$"}, "minItems": 1}}, "required": ["task_id", "addBlockedBy"], "additionalProperties": False}},
{"name": "list_tasks", "description": "List tasks with status, owner, and dependencies.",
"input_schema": {"type": "object", "properties": {}}},
{"name": "get_task", "description": "Get a task by ID.",
"input_schema": {"type": "object", "properties": {"task_id": {"type": "string"}}, "required": ["task_id"]}},
{"name": "claim_task", "description": "Claim a pending task whose dependencies are complete.",
"input_schema": {"type": "object", "properties": {"task_id": {"type": "string"}}, "required": ["task_id"]}},
{"name": "complete_task", "description": "Complete the task claimed by this agent.",
"input_schema": {"type": "object", "properties": {"task_id": {"type": "string"}}, "required": ["task_id"]}},
]
TOOL_HANDLERS = {
"bash": run_bash,
"read_file": run_read,
"write_file": run_write,
"edit_file": run_edit,
"glob": run_glob,
"create_task": run_create_task,
"update_task": run_update_task,
"list_tasks": run_list_tasks,
"get_task": run_get_task,
"claim_task": run_claim_task,
"complete_task": run_complete_task,
}
# -- From s04: hooks and permission checks --
HOOKS = {"UserPromptSubmit": [], "PreToolUse": [], "PostToolUse": [], "Stop": []}
def register_hook(event: str, callback):
HOOKS[event].append(callback)
def trigger_hooks(event: str, *args):
for callback in HOOKS[event]:
result = callback(*args)
if result is not None:
return result
return None
DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="]
DESTRUCTIVE_COMMAND_WORD = re.compile(
r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])"
)
DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"]
def contains_destructive_command(command: str) -> bool:
return bool(DESTRUCTIVE_COMMAND_WORD.search(command))
def permission_hook(block):
if block.name == "bash":
command = block.input.get("command", "")
for pattern in DENY_LIST:
if pattern in command:
print(f"\n\033[31m[blocked] '{pattern}'\033[0m")
return "Permission denied by deny list"
if contains_destructive_command(command) or any(
keyword in command for keyword in DESTRUCTIVE
):
print("\n\033[33m[permission] Potentially destructive command\033[0m")
print(f" Tool: {block.name}({block.input})")
choice = input(" Allow? [y/N] ").strip().lower()
if choice not in ("y", "yes"):
return "Permission denied by user"
if block.name in ("read_file", "write_file", "edit_file"):
path = block.input.get("path", "")
if not (WORKDIR / path).resolve().is_relative_to(WORKDIR):
print("\n\033[33m[permission] Access outside workspace\033[0m")
print(f" Tool: {block.name}({block.input})")
choice = input(" Allow? [y/N] ").strip().lower()
if choice not in ("y", "yes"):
return "Permission denied by user"
return None
def log_hook(block):
preview = str(list(block.input.values())[:2])[:60]
print(f"\033[90m[HOOK] {block.name}({preview})\033[0m")
return None
def large_output_hook(block, output):
if len(str(output)) > 100000:
print(
f"\033[33m[HOOK] Large output from {block.name}: "
f"{len(str(output))} chars\033[0m"
)
return None
def context_hook(query: str):
print(f"\033[90m[HOOK] UserPromptSubmit: working in {WORKDIR}\033[0m")
return None
def summary_hook(messages: list):
tool_count = sum(
1
for message in messages
for block in (
message.get("content")
if isinstance(message.get("content"), list)
else []
)
if isinstance(block, dict) and block.get("type") == "tool_result"
)
print(f"\033[90m[HOOK] Stop: session used {tool_count} tool calls\033[0m")
return None
register_hook("UserPromptSubmit", context_hook)
register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)
def execute_tool(block) -> str:
blocked = trigger_hooks("PreToolUse", block)
if blocked:
return str(blocked)
handler = TOOL_HANDLERS.get(block.name)
try:
output = handler(**block.input) if handler else f"Unknown: {block.name}"
except Exception as error:
output = f"Error: {error}"
trigger_hooks("PostToolUse", block, output)
return str(output)
# -- Agent loop --
def agent_loop(messages: list):
while True:
response = client.messages.create(
model=MODEL,
system=SYSTEM,
messages=messages,
tools=TOOLS,
max_tokens=8000,
)
messages.append({"role": "assistant", "content": response.content})
tool_calls = [
block for block in response.content if block.type == "tool_use"
]
if not tool_calls:
force = trigger_hooks("Stop", messages)
if force:
messages.append({"role": "user", "content": force})
continue
return
results = []
for block in tool_calls:
output = execute_tool(block)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
messages.append({"role": "user", "content": results})
if __name__ == "__main__":
print("s10: Task System - dependencies and task state")
print("Enter a question, press Enter to send. Type q to quit.\n")
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s10 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
trigger_hooks("UserPromptSubmit", query)
history.append({"role": "user", "content": query})
agent_loop(history)
for block in history[-1]["content"]:
if getattr(block, "type", None) == "text":
print(block.text)
print()
What's Next
The task graph is in place, but full test suites, dependency installation, and deployment commands can take a long time. When these commands run synchronously, the Agent Loop remains blocked in the current tool call and cannot continue until the command finishes.
s11 Background Tasks → Slow operations run in the background. The Agent Loop can continue processing other tasks and receives a notification when the background work finishes.
Fuente y licencia
- Adaptado de shareAI-lab/learn-claude-code / s10_task_system/README.md (shareAI-lab contributors, Licencia MIT, ce8f9f1)