learn-claude-code Documentation

Technical Documentation: shareAI-lab/learn-claude-code

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (`README.md`) Copy Section

English | 中文 | 日本語

Learn Claude Code -- Harness Engineering for Real Agents

Agency Comes from the Model. An Agent Product = Model + Harness. Copy Chapter

Before we write any code, one thing needs to be clear.

Agency -- the capacity to perceive, reason, and act -- comes from model training, not from external code orchestration. But a working agent product needs both the model and the harness. The model is the driver. The harness is the vehicle. This repository teaches you how to build the vehicle.

Where Agency Comes From Copy Section

At the core of every agent is a neural network -- a Transformer, an RNN, a trained function -- shaped by billions of gradient updates on sequences of perception, reasoning, and action. Agency was never bestowed by the surrounding code. It was learned during training.

Humans are the original proof. A biological neural network, refined by millions of years of evolutionary pressure, perceives the world through senses, reasons through a brain, and acts through a body. When DeepMind, OpenAI, or Anthropic say "agent," they all mean the same core thing: a model that learned to act through training, plus the infrastructure that lets it operate in a specific environment.

The historical record is unambiguous:

• 2013 -- DeepMind DQN plays Atari. A single neural network, receiving only raw pixels and game scores, learned 7 Atari 2600 games -- surpassing prior algorithms and beating human experts in 3 of them. By 2015, scaled to 49 games at professional tester level, published in Nature. No game-specific rules. One model, learning from experience.

• 2019 -- OpenAI Five conquers Dota 2. Five neural networks played 45,000 years of Dota 2 against themselves over 10 months, then defeated OG -- the TI8 world champions -- 2-0 in a live match. In the public arena, the AI won 99.4% of 42,729 games. No scripted strategies. Models learned teamwork through self-play.

• 2019 -- DeepMind AlphaStar masters StarCraft II. AlphaStar beat a professional player 10-1 in closed matches, then reached Grandmaster rank on the European server -- top 0.15% of 90,000 players. An incomplete-information, real-time game with a combinatorial action space far exceeding chess or Go.

• 2019 -- Tencent Jueyu dominates Honor of Kings. Tencent AI Lab's "Jueyu" system defeated KPL professional players in full 5v5 at the World Champion Cup semifinal. In 1v1 mode, pros won just 1 out of 15 matches, lasting under 8 minutes at best. Training intensity: one day equaled 440 human years. A model that learned the entire game from scratch through self-play.

• 2024-2025 -- LLM agents reshape software engineering. Claude, GPT, Gemini -- large language models trained on the full breadth of human code and reasoning -- are deployed as coding agents. They read codebases, write implementations, debug failures, and coordinate as teams. The architecture is identical to every previous agent: a trained model, placed in an environment, given tools for perception and action.

Every milestone points to the same fact: Agency -- the ability to perceive, reason, and act -- is trained, not coded. But every agent also needs an environment to operate in: an Atari emulator, the Dota 2 client, the StarCraft II engine, an IDE and a terminal. The model supplies the intelligence. The environment supplies the action space. Together they form a complete agent.

What an Agent Is NOT Copy Section

The word "agent" has been hijacked by an entire prompt-plumbing industry.

Drag-and-drop workflow builders. No-code "AI Agent" platforms. Prompt-chain orchestration libraries. They share a single delusion: that stringing LLM API calls together with if-else branches, node graphs, and hardcoded routing logic constitutes "building an agent."

It does not. What they produce are Rube Goldberg machines -- over-engineered, brittle, procedural rule pipelines with an LLM wedged in as a glorified text-completion node. That is not an agent. That is a shell script with grandiose pretensions.

You cannot brute-force intelligence by stacking procedural logic -- sprawling rule trees, node graphs, chained prompt waterfalls -- and praying that enough glue code will spontaneously produce autonomous behavior. It will not. You cannot engineer agency into existence. Agency is learned, not coded.

The Mindshift: From "Building Agents" to Building Harnesses Copy Section

When someone says "I am building an agent," they can only mean one of two things:

1. Training a model. Adjusting weights through reinforcement learning, fine-tuning, RLHF, or another gradient-based method. Collecting trajectory data -- real-world sequences of perception, reasoning, and action in a target domain -- and using it to shape the model's behavior. This is what DeepMind, OpenAI, Tencent AI Lab, and Anthropic do.

2. Building a harness. Writing the code that gives a model an operational environment. This is what most of us do, and it is the core of this repository.

A harness is everything an agent needs to work in a specific domain:

Harness = Tools + Knowledge + Observation + Action Interfaces + Permissions

    Tools:          file I/O, shell, network, database, browser
    Knowledge:      product docs, domain references, API specs, style guides
    Observation:    git diff, error logs, browser state, sensor data
    Action:         CLI commands, API calls, UI interactions
    Permissions:    sandbox isolation, approval workflows, trust boundaries

The model decides. The harness executes. The model reasons. The harness provides context. The model is the driver. The harness is the vehicle.

This repository teaches you to build the vehicle. A vehicle for coding. But the design patterns generalize to any domain.

What Harness Engineers Actually Do Copy Section

If you are reading this repository, you are most likely a harness engineer. Here is what the job actually entails:

• Implement tools. Give the agent hands. File read/write, shell execution, API calls, browser control, database queries. Each tool is one action the agent can take in its environment. Design them atomic, composable, and clearly described.

• Curate knowledge. Give the agent domain expertise. Product documentation, architecture decision records, style guides, compliance requirements. Load on demand, not upfront.

• Manage context. Subagents keep focused work in a separate message list. Context compaction shortens older history. Task systems let goals persist beyond a single conversation.

• Control permissions. Give the agent boundaries. Sandbox file access. Require approval for destructive operations. Enforce trust boundaries between the agent and external systems.

• Collect trajectory data. Every action sequence the agent executes in your harness is training signal. Real deployment trajectories are the raw material for fine-tuning the next generation of agent models.

You are not writing intelligence. You are building the world that intelligence inhabits. The quality of that world directly determines how effectively the intelligence can express itself.

Build the harness well. The model will do the rest.

Why Claude Code Copy Section

Because Claude Code is the most elegant, most complete agent harness implementation we have seen. Not because of any clever trick, but because of what it does not do: it does not try to be the agent. It does not impose rigid workflows. It does not substitute hand-crafted decision trees for the model's own judgment. It gives the model tools, knowledge, context management, and permission boundaries -- then gets out of the way.

Strip Claude Code down to its essence:

Claude Code = one agent loop
            + tools (bash, read, write, edit, glob, grep, browser...)
            + on-demand skill loading
            + context compaction
            + subagent spawning
            + task system with dependency graphs
            + async mailbox team coordination
            + task-bound worktrees for parallel edits
            + permission governance
            + hooks extension system
            + memory persistence
            + MCP external capability routing

That is it. The agent itself? Claude. A model. Trained by Anthropic on the full breadth of human reasoning and code. The harness did not make Claude smart. Claude was already smart. The harness gave Claude hands, eyes, and a workspace.

The takeaway is not "copy Claude Code." The takeaway is: the best agent products come from engineers who understand that their job is the harness, not the intelligence.

THE AGENT PATTERN
                    =================

    User --> messages[] --> LLM --> response
                                      |
                              contains tool_use block?
                           /                          \
                         yes                           no
                          |                             |
                    execute tools                    return text
                    append results
                    loop back -----------------> messages[]


    The model decides when to call tools and when to stop.
    The code just executes what the model asks for.
    This repo teaches you to build everything around this loop --
    the harness that makes the agent effective in a specific domain.

Core Pattern Copy Chapter

def agent_loop(messages):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            messages=messages, tools=TOOLS,
        )
        messages.append({"role": "assistant",
                         "content": response.content})

        tool_calls = [
            block for block in response.content if block.type == "tool_use"
        ]
        if not tool_calls:
            return

        results = []
        for block in tool_calls:
            output = TOOL_HANDLERS[block.name](**block.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

Each lesson isolates one harness mechanism around this loop. s15 reconnects the cumulative runtime; s16 and s17 then study workflow orchestration and goal closure as focused examples. The loop belongs to the agent. The mechanisms belong to the harness.

The loop is constant. Tools, knowledge, and permissions change. Agent = Model (LLM) + a generalized operational environment (Harness).

Version Status Copy Chapter

This repository currently contains two tutorial tracks:

• Current track: root-level s01-s17
  The root-level s01_* ... s17_* folders are the canonical version. Each chapter contains an English default README, Chinese/Japanese translations, runnable code.py, and diagrams where needed.
• Legacy transition track: docs/ and agents/
  These preserve the older 12-lesson version for existing readers and old links during migration.

If you are starting now, read the root-level s01_agent_loop/ through s17_goal_loop/ chapters. The legacy and current chapter numbers do not always match, so avoid mixing chapter numbers across tracks.

Legacy-to-Current Mapping Copy Section

Course Boundary Copy Chapter

This is a 0-to-1 harness engineering course. Each chapter isolates one mechanism, then s15 reconnects the cumulative runtime in a complete agent loop. s16 extends that loop with workflow orchestration. s17 uses a smaller tool pool to focus on goal-controlled continuation; it is a mechanism example, not another cumulative runtime.

17 Progressive Lessons Copy Chapter

Each lesson adds one harness mechanism. Each mechanism has a motto.

s01   "One loop & Bash is all you need" — one tool + one loop = one agent

s02   "Adding a tool means adding one handler" — the loop stays untouched; new tools register into the dispatch map

s03   "Set boundaries first, then grant freedom" — check what can run, what must stop, and what needs approval

s04   "Hook around the loop, never rewrite the loop" — add extension points without changing the main loop

s05   "An agent without a plan drifts" — list the steps before starting; completion rate doubles

s06   Give a subtask fresh messages[]; its final text returns as one tool result

s07   "Load knowledge on demand, not upfront" — list skills first, expand them only when needed

s08   "Context always fills up -- have a way to make room" — four compaction steps reduce tool results first, then summarize history when it remains over the limit

s09   "Remember what matters, forget what doesn't" — three subsystems: selection, extraction, consolidation

s10   "Big goals break into small tasks, ordered, persisted to disk" — a file-backed task graph that lays the groundwork for multi-agent coordination

s11   "Slow ops go background, agent keeps thinking" — background threads run commands; notifications inject on completion

s12   "Fire on schedule, no human kick needed" — trigger tasks automatically by time

s13   "Too big for one agent -- let teammates divide the work" — persistent teammates coordinate, claim ready tasks, and use task-bound working directories

s14   "Not enough capability? Plug in more via MCP" — connect external tools into the same tool pool

s15   "Many mechanisms, one loop" — the mechanisms used by the integrated example share one harness

s16   "When the orchestration shape is fixed, put it in code" — saved workflows with resumable journals

s17   "A goal decides when the loop may stop" — an independent evaluator reviews each proposed stop; impossible, failed, or over-limit goals return control to the user

Learning Path Copy Chapter

Main line: act → handle complex work → remember across sessions → run long tasks → collaborate → extend and assemble → orchestrate and close goals.

/* Detailed source-code truncated for AI context efficiency. */

All Chapters Copy Chapter

How to Read Copy Chapter

Each chapter is a folder. Open one and you will find:

s08_context_compact/
  README.md              # English, the default chapter README
  README.zh.md           # Chinese translation
  README.ja.md           # Japanese translation
  code.py                # standalone runnable implementation
  images/                # SVG diagrams (where needed)

Read the README.md for the core idea and work through the code. Complex chapters have `` folds for deep dives -- open them when you want to go deeper. Simple chapters have 0-1 diagrams, complex chapters have more.

Read from s01 through s17 in order. Some mechanisms build directly on the previous runtime; independent mechanism chapters state which earlier kernel they use.

Quick Start Copy Chapter

Current 17-Lesson Track Copy Section

git clone https://github.com/shareAI-lab/learn-claude-code
cd learn-claude-code
pip install -r requirements.txt
cp .env.example .env   # configure ANTHROPIC_API_KEY

python s01_agent_loop/code.py        # Start here -- one loop + bash
python s08_context_compact/code.py   # Context compaction (complex)
python s17_goal_loop/code.py         # Endpoint: continue until a checkable goal is met

Legacy 12-Lesson Track Copy Section

python agents/s01_agent_loop.py
python agents/s12_worktree_task_isolation.py
python agents/s_full.py

Web Platform Copy Section

The web app extracts the root-level course. Lessons s16 and s17 include reading, source, simulator, and architecture views; only their dedicated hero visualizations remain intentionally minimal.

cd web && npm install && npm run dev   # http://localhost:3000

Project Structure Copy Chapter

learn-claude-code/
  s01_agent_loop/          # one folder per chapter
    README.md              #   English default (complete narrative)
    README.zh.md           #   Chinese translation
    README.ja.md           #   Japanese translation
    code.py                #   standalone runnable code
    images/                #   SVG diagrams
  s02_tool_use/
  ...
  s14_mcp_plugin/
  s15_integrated_harness/
  s16_workflow_runtime/
  s17_goal_loop/           # endpoint chapter
  agents/                  # legacy 12 runnable copies + s_full.py
  skills/                  # skill files used by s07
  docs/                    # legacy 12-lesson docs, kept during transition
  web/                     # generated from the root-level course
  tests/

What's Next Copy Chapter

After 17 lessons, you understand harness engineering from the inside out. Two paths to turn that knowledge into product:

Kode Agent CLI -- Open-Source Coding Agent CLI Copy Section

npm i @shareai-lab/kode

Skill and LSP support, Windows compatible, works with GLM / MiniMax / DeepSeek and other open models. Install and go.

GitHub: shareAI-lab/Kode-CLI

Kode Agent SDK -- Embed Agent Capabilities in Your Application Copy Section

A standalone library with no per-user process overhead. Embed it in backends, browser extensions, embedded devices, or any runtime.

GitHub: shareAI-lab/kode-agent-sdk

Sister Tutorial: From Passive Sessions to Always-On Assistants Copy Chapter

The harness taught in this repository is the use-and-discard kind -- open a terminal, give the agent a task, close when done, next session starts fresh. Claude Code works this way.

But OpenClaw proves another possibility: on the same agent core, two additional harness mechanisms turn an agent from "poke it and it moves" into "wakes itself every 30 seconds to look for work":

• Heartbeat -- every 30 seconds the harness sends the agent a message, letting it check for pending work. Nothing to do? Keep sleeping. Something appeared? Act immediately.
• Cron -- the agent can schedule its own future tasks, which fire automatically when the time arrives.

Add IM multi-channel routing (WhatsApp / Telegram / Slack / Discord and 13+ other platforms), persistent context memory, and a Soul personality system, and the agent transforms from a disposable tool into an always-on personal AI assistant.

claw0 is our sister teaching repository, breaking down these harness mechanisms from scratch:

claw agent = agent core + heartbeat + cron + IM chat + memory + soul

learn-claude-code                   claw0
(agent harness internals:            (always-on harness:
 loop, tools, planning,               heartbeat, cron, IM channels,
 teams, task-bound worktrees)          memory, Soul personality)

License Copy Chapter

MIT

Agency comes from the model. The harness gives agency a place to land. Build the harness well, and the model will do the rest.

Bash is all you need. Real agents are all the universe needs.

This is not "copy the source code." This is "grasp the key designs and build it yourself."

Overview (`s01_agent_loop/README.md`) Copy Section

s01: The Agent Loop — One Loop Is All You Need

English · 中文 · 日本語

s01 → s02 → s03 → s04 → ... → s16 → s17

"One loop & Bash is all you need" — One tool + one loop = one Agent.

Harness Layer: The Loop — the first bridge between the model and the real world.

The Problem Copy Chapter

You ask the model: "List the files in my directory and run XXX.py."

The model can output a bash command, but once it's done outputting, it stops — it won't execute the command on its own, and it won't keep reasoning based on the result.

You could run it manually, paste the output back into the chat, and let it continue. Next command comes out, you run it again, paste it back.

Every round-trip, you're the middle layer. Automating that is what this chapter is about.

The Solution Copy Chapter

A while True loop: keep going when the model calls a tool, stop when it doesn't. The loop checks the response content blocks directly:

How It Works Copy Chapter

Let's translate this process into code. Step by step:

Step 1: Start with the user's question as the first message.

messages = [{"role": "user", "content": query}]

Step 2: Send the messages and tool definitions to the LLM.

response = client.messages.create(
    model=MODEL, system=SYSTEM, messages=messages,
    tools=TOOLS, max_tokens=8000,
)

Step 3: Append the model's response and check whether it called a tool. No tool call → done.

messages.append({"role": "assistant", "content": response.content})
tool_calls = [
    block for block in response.content if block.type == "tool_use"
]
if not tool_calls:
    return

Only concrete tool_use blocks enter the execution stage, so the loop never appends an empty tool-result message.

Step 4: Execute the tool the model requested and collect the results.

results = []
for block in tool_calls:
    output = run_bash(block.input["command"])
    results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": output,
    })

Step 5: Append the tool results as a new message and go back to Step 2.

messages.append({"role": "user", "content": results})

Assembled into a complete function:

def agent_loop(messages):
    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:
            return

        results = []
        for block in tool_calls:
            output = run_bash(block.input["command"])
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

Just over 30 lines — that's the minimal runnable agent harness kernel. It's not intelligence itself, but the smallest runtime framework that lets the model keep acting. The model decides (whether to call a tool, which one), the harness executes (calls the tool and appends the result as a new message). The next 16 chapters all add mechanisms on top of this loop. The loop itself never changes.

Try It Copy Chapter

Safety notice: The code executes shell commands generated by the model. Run it in a temporary test directory to avoid affecting your project files. s03 adds permission controls.

Setup (first run):

pip install -r requirements.txt
cp .env.example .env
# Edit .env, fill in ANTHROPIC_API_KEY and MODEL_ID

Run:

python s01_agent_loop/code.py

Try these prompts:

1. Create a file called hello.py that prints "Hello, World!"
2. List all Python files in this directory
3. What is the current git branch?

What to watch for: When does the model call a tool (loop continues), and when does it not (loop ends)?

What's Next Copy Chapter

Right now the model only has bash — reading files requires cat, writing files requires echo ... >, finding files requires find. Ugly and error-prone.

→ s02 Tool Use: What happens when we give it 5 proper tools? Will the model call multiple tools at once? Will parallel tool executions step on each other?

Overview (`s02_tool_use/README.md`) Copy Section

s02: Tool Use — Add a Tool, Add Just One Line

English · 中文 · 日本語

s01 → s02 → s03 → s04 → ... → s16 → s17

"Add a tool, add just one handler" — The loop stays the same. Register the new tool in the dispatch map and you're done.

Harness Layer: Tool Dispatch — Expanding the model's reach.

Only One Tool: Bash Copy Chapter

The s01 Agent has only one tool: bash. To read a file, cat; to write, echo "..." > file.py; to edit, sed.

The model thinks "read this file" but has to spell out cat path/to/file. An extra layer of translation that wastes tokens and invites errors.

Overview: Tool Dispatch Copy Chapter

The s01 loop is fully preserved (LLM call, tool_use block check, message append — not a single word changed). The only change is in that one line of tool execution: run_bash() is replaced with TOOL_HANDLERS[block.name]() dispatch lookup.

Adding a tool to the Agent requires just two things:

1. Define the tool: Add one entry to the TOOLS array
2. Register the handler: Add one mapping in the TOOL_HANDLERS dict

From 1 Tool to 5 Tools Copy Chapter

s01 had only bash:

TOOLS = [{"name": "bash", ...}]

def run_bash(command): ...

s02 expands to 5 tools, each independently defined:

TOOLS = [
    {"name": "bash",       "description": "Run a shell command.", ...},
    {"name": "read_file",  "description": "Read file contents.",  ...},
    {"name": "write_file", "description": "Write content to file.", ...},
    {"name": "edit_file",  "description": "Replace text in file once.", ...},
    {"name": "glob",       "description": "Find files by pattern.", ...},
]

Each tool has its own implementation function:

def run_read(path, limit=None):
    lines = safe_path(path).read_text(encoding="utf-8").splitlines()
    if limit:
        lines = lines[:limit]
    return "\n".join(lines)

def run_write(path, content):
    safe_path(path).write_text(content, encoding="utf-8")
    return f"Wrote {len(content)} bytes to {path}"

def run_edit(path, old_text, new_text):
    text = safe_path(path).read_text(encoding="utf-8")
    if old_text not in text:
        return "Error: text not found"
    safe_path(path).write_text(text.replace(old_text, new_text, 1), encoding="utf-8")
    return f"Edited {path}"

def run_glob(pattern):
    import glob as g
    matches = sorted(set(g.glob(
        pattern, root_dir=WORKDIR, recursive=True)))
    shown = matches[:200]
    if len(matches) > 200:
        shown.append("... (more matches omitted; narrow the pattern)")
    return "\n".join(shown)

Tool Dispatch Copy Chapter

TOOL_HANDLERS = {
    "bash":       run_bash,
    "read_file":  run_read,
    "write_file": run_write,
    "edit_file":  run_edit,
    "glob":       run_glob,
}

# Only one line changed in the loop — from hardcoded run_bash to dispatch lookup:
for block in tool_calls:
    handler = TOOL_HANDLERS[block.name]    # lookup
    output = handler(**block.input)         # call
    results.append(...)

Adding a tool = one entry in TOOLS array + one line in TOOL_HANDLERS dict. The loop stays the same.

Multiple Tool Calls Copy Chapter

The model often returns multiple tool_use calls at once — "read a.py and b.py, then list all .py files".

Calls are executed one by one in their original response.content order.

Quick Reference Copy Chapter

Changes from s01 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python s02_tool_use/code.py

Try these prompts:

1. Read the file README.md and tell me what this project is about
2. Create a file called test.py that prints "hello", then read it back
3. Find all Python files in this directory
4. Read both README.md and requirements.txt, then create a summary file

What to watch for: When does the model call just one tool, and when does it call multiple at once? Are multiple tool calls executed in the correct order?

What's Next Copy Chapter

The Agent now has 5 specialized tools. File tools are protected by safe_path, but bash is unrestricted — rm -rf / still runs.

→ s03 Permission: Add a gate before tool execution — is this operation safe? Does it need user approval?

Overview (`s03_permission/README.md`) Copy Section

s03: Permission — Check Permissions Before Execution

English · 中文 · 日本語

s01 → s02 → s03 → s04 → s05 → ... → s16 → s17

"Check permissions before executing" — The permission pipeline decides which operations need approval.

Harness Layer: Permission — a gate before tool execution.

The Problem Copy Chapter

s02's Agent has 5 tools. File tools are protected by safe_path, but bash is unrestricted. Ask it to "clean up the project," and it might run rm -rf /.

Safety can't rely on trusting the model — it needs code: a check before every tool execution.

The Solution Copy Chapter

s02's loop is fully preserved. The only change is inserting check_permission() before tool execution — each tool call passes through three gates in a fixed order: hard deny first, then soft ask, and if neither matches, allow.

The three gates correspond to three decisions:

None of the three gates match → execute directly. Most routine operations take this path.

How It Works Copy Chapter

Gate 1: A hard deny list. Check first; if matched, return a block message. This list uses simple string matching to show where the permission gate sits; it is not a complete security boundary.

DENY_LIST = [
    "rm -rf /", "sudo", "shutdown", "reboot",
    "mkfs", "dd if=", "> /dev/sda",
]

def check_deny_list(command: str) -> str | None:
    for pattern in DENY_LIST:
        if pattern in command:
            return f"Blocked: '{pattern}' is on the deny list"
    return None

Gate 2: Rule matching — describes "when to ask the user." Each rule specifies a tool and a check condition.

import re

DESTRUCTIVE_COMMAND_WORD = re.compile(
    r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])"
)

def contains_destructive_command(command: str) -> bool:
    return bool(DESTRUCTIVE_COMMAND_WORD.search(command))

PERMISSION_RULES = [
    {
        "tools": ["read_file", "write_file", "edit_file"],
        "check": lambda args: not (WORKDIR / args.get("path", "")).resolve().is_relative_to(WORKDIR),
        "message": "Access outside workspace",
    },
    {
        "tools": ["bash"],
        "check": lambda args: contains_destructive_command(args.get("command", "")) or any(
            kw in args.get("command", "") for kw in ["rm ", "> /etc/", "chmod 777"]
        ),
        "message": "Potentially destructive command",
    },
]

def check_rules(tool_name: str, args: dict) -> str | None:
    for rule in PERMISSION_RULES:
        if tool_name in rule["tools"] and rule["check"](args):
            return rule["message"]
    return None

Gate 3: After a rule matches, pause for user input.

def ask_user(tool_name: str, args: dict, reason: str) -> str:
    print(f"\n⚠  {reason}")
    print(f"   Tool: {tool_name}({args})")
    choice = input("   Allow? [y/N] ").strip().lower()
    return "allow" if choice in ("y", "yes") else "deny"

All three gates chained together, inserted before tool execution:

def check_permission(block) -> bool:
    # Gate 1: Hard deny
    if block.name == "bash":
        reason = check_deny_list(block.input.get("command", ""))
        if reason:
            print(f"\n⛔ {reason}")
            return False

    # Gate 2 + 3: Rule matching → User approval
    reason = check_rules(block.name, block.input)
    if reason:
        decision = ask_user(block.name, block.input, reason)
        if decision == "deny":
            return False

    return True

# In agent_loop — s02's loop with just one line added:
for block in tool_calls:
    if not check_permission(block):           # ← NEW
        results.append({... "content": "Permission denied."})
        continue
    output = TOOL_HANDLERS[block.name](**block.input)  # s02 original
    results.append(...)

Changes from s02 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python s03_permission/code.py

Try these prompts:

1. Create a file called test.txt in the current directory (should pass through)
2. Delete the file test.txt (bash + rm triggers Gate 2)
3. What files are in the current directory? (read-only, all pass)
4. Try to write a file to /etc/something (writing outside workspace triggers Gate 2)
5. On Windows, del test.txt and DEL test.txt trigger Gate 2, while model, delimiter, and echo del test.txt do not.

What to watch for: Which operations pass through? Which need your confirmation? Which are denied outright?

What's Next Copy Chapter

Permission checks are in place — but every check is hardcoded as check_permission() inside the loop. What if you want to add logging before and after each tool execution? What if you want to auto-trigger a git commit after certain operations? Scattering this extension logic throughout the loop makes it bloat.

→ s04 Hooks: Add hooks to the loop. Extension logic hangs on hooks; the loop stays clean.

Overview (`s04_hooks/README.md`) Copy Section

s04: Hooks — Hang on the Loop, Don't Write into It

English · 中文 · 日本語

s01 → s02 → s03 → s04 → s05 → s06 → ... → s16 → s17

"Hang on the loop, don't write into it" — Hooks inject extension logic before and after tool execution.

Harness Layer: Hooks — Extension points that don't invade the loop.

The Problem Copy Chapter

The s03 Agent has permission checks. But every new check, "log every bash call", "auto git add after writes", requires modifying the agent_loop function.

The loop quickly becomes this:

def agent_loop(messages):
    while True:
        # ... LLM call ...
        for block in response.content:
            if block.type != "tool_use":
                continue
            log_to_file(block)          # added a line
            check_permission(block)     # added a line
            notify_slack(block)         # added another line
            output = execute(block)
            auto_git_add(block)         # yet another line
            # ... the loop is unrecognizable

What you want to extend is the Agent's behavior, but what you're modifying is the loop itself. The loop should be a stable core; extensions should hang on the outside.

The Solution Copy Chapter

The s03 loop and permission logic are fully preserved. The only change is moving check_permission() from inside the loop body onto a hook. The loop no longer directly calls any check function. Instead it calls trigger_hooks("PreToolUse", block), and the registry decides what to run.

Four events, covering a complete agent cycle:

Extensions are added via register_hook(). The loop only calls trigger_hooks().

How It Works Copy Chapter

Hook registry: a dict mapping event names to callback lists.

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 value ≠ None → hook says "stop"
            return result
    return None

When PreToolUse returns non-None, the current tool execution is blocked. When Stop returns non-None, the loop continues. Return values from UserPromptSubmit and PostToolUse do not affect control flow.

UserPromptSubmit triggers after user input and before entering the LLM. The following hook records the current working directory:

def context_inject_hook(query: str) -> str | None:
    """Inject current working directory info into every prompt."""
    print(f"\033[90m[HOOK] UserPromptSubmit: working in {WORKDIR}\033[0m")
    return None   # return None = no modification, let prompt through

register_hook("UserPromptSubmit", context_inject_hook)

In the main loop, triggered right after user input:

query = input("s04 >> ")
trigger_hooks("UserPromptSubmit", query)   # ← before entering LLM
history.append({"role": "user", "content": query})
agent_loop(history)

PreToolUse / PostToolUse, hooks before and after tool execution. s03's permission check logic is now wrapped as a PreToolUse hook, plus a logging hook and a large-output reminder:

# PreToolUse: permission check (s03 logic, moved from loop to hook)
def permission_hook(block):
    if block.name == "bash":
        for pattern in DENY_LIST:
            if pattern in block.input.get("command", ""):
                return "Permission denied by deny list"
    if block.name in ("read_file", "write_file", "edit_file"):
        path = block.input.get("path", "")
        if not (WORKDIR / path).resolve().is_relative_to(WORKDIR):
            choice = input("   Allow? [y/N] ").strip().lower()
            if choice not in ("y", "yes"):
                return "Permission denied by user"
    return None

# PreToolUse: logging
def log_hook(block):
    print(f"[HOOK] {block.name}(...)")

# PostToolUse: large output reminder
def large_output_hook(block, output):
    if len(str(output)) > 100000:
        print(f"[HOOK] ⚠ Large output from {block.name}")

register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)

Stop triggers when the loop is about to exit. The following hook prints a cleanup summary:

def summary_hook(messages: list) -> str | None:
    """Print a summary when the loop is about to stop."""
    tool_count = sum(1 for m in messages
                     for b in (m.get("content") if isinstance(m.get("content"), list) else [])
                     if isinstance(b, dict) and b.get("type") == "tool_result")
    print(f"\033[90m[HOOK] Stop: session used {tool_count} tool calls\033[0m")
    return None   # return None = allow stop, return string = force continuation

register_hook("Stop", summary_hook)

In agent_loop, triggered before exit:

tool_calls = [
    block for block in response.content if block.type == "tool_use"
]
if not tool_calls:
    force = trigger_hooks("Stop", messages)   # ← before exiting
    if force:
        # hook returned a message → inject it and continue
        messages.append({"role": "user", "content": force})
        continue
    return

Only one change in the loop: s03 directly called check_permission(block), s04 replaces it with trigger_hooks("PreToolUse", block):

for block in tool_calls:
    # s03: if not check_permission(block): ...
    # s04: hooks replace hardcoding
    blocked = trigger_hooks("PreToolUse", block)
    if blocked:
        results.append({"type": "tool_result", "tool_use_id": block.id,
                        "content": str(blocked)})
        continue

    handler = TOOL_HANDLERS.get(block.name)
    output = handler(**block.input) if handler else f"Unknown: {block.name}"

    trigger_hooks("PostToolUse", block, output)

    results.append({"type": "tool_result", "tool_use_id": block.id,
                    "content": output})

Four hooks cover the critical nodes of the agent cycle: input → before execution → after execution → exit. The loop only calls trigger_hooks(); all logic lives in hook callbacks.

Changes from s03 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python s04_hooks/code.py

Try these prompts:

1. Read the file README.md (should pass directly, observe hook logs)
2. Create a file called test.txt (after creation, observe if PostToolUse fires)
3. Delete all temporary files in /tmp (bash + rm triggers permission hook)

What to watch for: Before each tool execution, does the [HOOK] log appear? When permission is denied, was it intercepted by a hook or hardcoded in the loop?

What's Next Copy Chapter

The Agent can now safely execute operations. But does it ever stop to think "what should I do first, and what next?" Given a complex task, does it jump straight in, or plan first?

→ s05 TodoWrite: Give the Agent a planning tool. Make a list first, then execute.

Overview (`s05_todo_write/README.md`) Copy Section

s05: TodoWrite — An Agent Without a Plan Drifts Off Course

English · 中文 · 日本語

s01 → s02 → s03 → s04 → s05 → s06 → s07 → ... → s16 → s17

"An agent without a plan goes wherever the wind blows" — List the steps first, then execute. Complex tasks are less likely to miss steps.

Harness Layer: Planning — Let the Agent think before it acts.

The Problem Copy Chapter

Give the Agent a complex task: "Rename all Python files to snake_case, run tests, and fix failures."

The Agent starts working, renames 3 files, runs a test, finds 2 failures, starts fixing. While fixing, it forgets the original goal was "rename to snake_case", the test failures have consumed all its attention.

The longer the conversation, the worse it gets: tool results keep filling the context, diluting the system prompt's influence. A 10-step refactoring: after steps 1-3, the Agent starts improvising because steps 4-10 have been pushed out of its attention.

The Solution Copy Chapter

S05 keeps the tool dispatch, permissions, and hooks from S04, then adds todo_write and a reminder counter. todo_write only updates planning state; the existing tools still perform the work.

The new tool uses the same TOOL_HANDLERS[block.name] dispatch path. After three consecutive tool-use rounds without todo_write, the harness adds a reminder to that round's tool results.

How It Works Copy Chapter

TodoManager owns the in-memory list, validates updates, and renders the state returned to the model. run_todo_write also prints that state in the terminal:

class TodoManager:
    def __init__(self):
        self.items = []

    def update(self, todos: list | str) -> str:
        # Parse and validate before replacing the current list.
        validated = []
        ...
        self.items = validated
        return self.render()

    def render(self) -> str:
        # [ ] pending, [>] in progress, [x] completed
        ...


TODO = TodoManager()

def run_todo_write(todos: list | str) -> str:
    output = TODO.update(todos)
    print(output)
    return output

An update may contain at most 20 items, each item needs non-empty content, and only one item may be in_progress. The string input path accepts JSON or a Python list representation without using eval.

The tool definition joins the other 5 in the dispatch map:

TOOLS = [
    {"name": "bash",       ...},
    {"name": "read_file",  ...},
    {"name": "write_file", ...},
    {"name": "edit_file",  ...},
    {"name": "glob",       ...},
    # s05: new entry
    {"name": "todo_write", "description": "Create and manage a task list ...",
     "input_schema": {
         "type": "object",
         "properties": {
             "todos": {
                 "type": "array",
                 "items": {
                     "type": "object",
                     "properties": {
                         "content": {"type": "string"},
                         "status": {"type": "string", "enum": ["pending", "in_progress", "completed"]},
                     },
                 },
             },
         },
     },
    },
]

TOOL_HANDLERS["todo_write"] = run_todo_write

Reminder: after three tool-use rounds without todo_write, the reminder is appended to the third round's results and the counter resets:

rounds_since_todo = 0 if used_todo else rounds_since_todo + 1
if rounds_since_todo >= 3:
    results.append({
        "type": "text",
        "text": "<reminder>Update your todos.</reminder>",
    })
    rounds_since_todo = 0

Typical flow when the Agent receives a task: first call todo_write to list all steps (all pending) → pick one step, set it to in_progress → complete it, set to completed → look at the next pending → continue.

Key insight: todo_write doesn't give the Agent any additional execution capability. What it adds is planning capability.

Changes from s04 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python s05_todo_write/code.py

Try these prompts:

1. Refactor s05_todo_write/example/hello.py: add type hints, docstrings, and a main guard (should list 3 steps first, then execute)
2. Create a Python package under s05_todo_write/example/demo_pkg with __init__.py, utils.py, and tests/test_utils.py
3. Review Python files under s05_todo_write/example and fix any style issues

What to watch for: Was the first tool call todo_write? How many TODO steps were listed? Did statuses move from pending to in_progress / completed during execution?

What's Next Copy Chapter

The Agent can plan now. But if a task is too large, say "refactor the entire auth module", a TODO list alone isn't enough. That task is itself a collection of dozens of subtasks that would drown in a single conversation's context.

→ s06 Subagent: Break large tasks into subtasks, each handled by an independent Agent with its own clean context, no cross-contamination.

Overview (`s06_subagent/README.md`) Copy Section

s06: Subagent — Give a Subtask Its Own Context

English · 中文 · 日本語

s01 → s02 → s03 → s04 → s05 → s06 → s07 → s08 → ... → s16 → s17

A subagent starts with a fresh messages[]. Its final text returns to the parent; its intermediate conversation does not.

Harness Layer: Delegation — Run a focused task in a separate conversation context.

The Problem Copy Chapter

The Agent is fixing a bug. It reads many files to trace the call chain, and every tool call and result stays in the parent's messages[]. Once the call chain is understood, most of those intermediate details are no longer needed, but they still occupy context.

The Solution Copy Chapter

Calling task synchronously runs a nested agent loop with a fresh messages[]. When that loop finishes, its final text becomes the tool result in the parent conversation.

This is message isolation, not process or filesystem isolation. Parent and subagent run in the same Python process and share WORKDIR, so writes and commands still affect the same workspace. The subagent has the five base tools but no task, and its tool calls use the same permission and lifecycle hooks as the parent.

How It Works Copy Chapter

run_subagent creates the fresh message list, runs the nested loop, and returns the final text:

SUB_TOOLS = list(BASE_TOOLS)  # no task tool

def run_subagent(prompt: str) -> str:
    messages = [{"role": "user", "content": prompt}]

    for _ in range(30):
        response = client.messages.create(
            model=MODEL, system=SUB_SYSTEM,
            messages=messages, tools=SUB_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:
            return extract_text(response.content) or "(no summary)"

        results = []
        for block in tool_calls:
            output = execute_tool(block, SUB_HANDLERS)
            results.append({... "content": output})
        messages.append({"role": "user", "content": results})

    return "Subagent stopped after 30 turns without a final answer."

The main Agent calls it just like any other tool:

TASK_TOOL = {
    "name": "task",
    "description": "Run a subagent with fresh conversation context and return its final text.",
    "input_schema": {
        "type": "object",
        "properties": {"prompt": {"type": "string"}},
        "required": ["prompt"],
    },
}

TOOLS = [*BASE_TOOLS, TASK_TOOL]
TOOL_HANDLERS = {**BASE_HANDLERS, "task": run_subagent}

The boundary is:

The parent dispatches task through the same handler map as its other tools. The subagent uses SUB_SYSTEM, SUB_TOOLS, and its own local messages list.

Try It Copy Chapter

cd learn-claude-code
python s06_subagent/code.py

Try these prompts:

1. Use a subtask to find what testing framework this project uses (sub-Agent reads files, main Agent receives only the conclusion)
2. Delegate: read all .py files in agents/ and summarize what each one does
3. Use a task to create s06_subagent/example/string_tools.py with a slugify(text: str) function, then verify it from the parent agent

What to watch for: Do [Subagent started] / [Subagent done] appear? Do subagent tool calls print as [sub] ...? Does the parent continue with only the final text returned by task?

What's Next Copy Chapter

The Agent can now break tasks apart. But different tasks require different knowledge: editing frontend components needs React conventions, writing SQL needs table schemas. Stuffing all this knowledge into the system prompt would blow up the context.

→ s07 Skill Loading: Inject skills on demand instead of piling documents into the system prompt. Load only when needed, as natural as reading a file.

Overview (`s07_skill_loading/README.md`) Copy Section

s07: Skill Loading — Load Skills When Needed

English · 中文 · 日本語

s01 → s02 → s03 → s04 → s05 → s06 → s07 → s08 → s09 → ... → s16 → s17

The system prompt contains the skill catalog; load_skill returns the full SKILL.md.

Harness Layer: Knowledge loading — show the model which skills exist, then load one by name.

The Problem Copy Chapter

Suppose a project has a React component specification, a SQL style guide, and an API design document. We want the Agent to follow these rules during development, so the most direct approach is to put all of them into the system prompt:

SYSTEM = (
    f"You are a coding agent. "
    + open("docs/react-style.md").read()
    + open("docs/sql-style.md").read()
    + open("docs/api-design.md").read()
)

This approach lets the Agent read every specification, but it fixes all three documents in the system prompt instead of selecting only the one needed for the current task. Every LLM call sends the full text of all three documents to the model. When the task only changes React components, only the React specification is relevant; the SQL style guide and API design document still consume input tokens and context-window space that could hold code, conversation, and tool results.

The Solution Copy Chapter

At startup, SkillLoader scans skills/*/SKILL.md, reads name and description from YAML frontmatter, and adds that catalog to the system prompt. When the model needs the full instructions, it calls load_skill(name); the returned SKILL.md is appended to the message list as a tool_result.

How It Works Copy Chapter

Each skill is a directory containing SKILL.md:

skills/
  agent-builder/SKILL.md
  code-review/SKILL.md
  mcp-builder/SKILL.md
  pdf/SKILL.md

Scan Skills Copy Section

class SkillLoader:
    def scan(self):
        self.skills.clear()
        skills_root = self.skills_dir.resolve()
        for manifest in sorted(self.skills_dir.glob("*/SKILL.md")):
            if (not manifest.is_file()
                    or not manifest.resolve().is_relative_to(skills_root)):
                continue
            content = manifest.read_text(encoding="utf-8")
            metadata, body = self.parse_frontmatter(content)
            raw_name = metadata.get("name")
            name = raw_name.strip() if isinstance(raw_name, str) else ""
            name = name or manifest.parent.name
            raw_description = metadata.get("description")
            description = (raw_description.strip()
                           if isinstance(raw_description, str) else "")
            description = description or body.split("\n", 1)[0]
            description = " ".join(str(description).lstrip("# ").split())
            self.skills[name] = {
                "name": name,
                "description": description,
                "content": content,
            }

catalog() returns only names and descriptions:

- code-review: Perform thorough code reviews...
- pdf: Process PDF files...

Build the System Prompt Copy Section

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"
)

def build_system_prompt() -> str:
    return (
        f"You are a coding agent at {WORKDIR}. Environment: {ENVIRONMENT_PROMPT}. "
        "Use tools to solve tasks. "
        "Act, don't explain.\n\n"
        f"Skills available:\n{SKILL_LOADER.catalog()}\n\n"
        "Use load_skill to read the full instructions when a skill applies."
    )

This function combines the fixed Agent instructions with the catalog found at startup.

Load Full Content Copy Section

def load(self, name: str) -> str:
    skill = self.skills.get(name)
    if skill:
        return skill["content"]
    available = ", ".join(self.skills) or "none"
    return f"Error: Unknown skill '{name}'. Available: {available}"

name looks up the startup registry; it is not interpreted as a file path. After the tool returns, the existing Agent Loop appends its content as a new tool_result message.

Try It Copy Chapter

cd learn-claude-code
python s07_skill_loading/code.py

Try these prompts:

1. What skills are available?
2. Load the code-review skill and follow its instructions
3. Review README.md and load the relevant skill first

Check that the system prompt contains only the catalog and that the full SKILL.md appears after load_skill is called.

What's Next Copy Chapter

As tool calls accumulate, messages[] retains earlier file contents and tool results.

→ s08 Context Compact: shorten earlier messages and keep context available for later calls.

Overview (`s08_context_compact/README.md`) Copy Section

s08: Context Compact: Make Room Before the Context Fills Up

English · 中文 · 日本語

s01 → s02 → s03 → s04 → s05 → s06 → s07 → s08 → s09 → s10 → ... → s16 → s17

"Context will fill up, so the Harness needs a way to make room." Four steps run from lower cost to higher cost.

Harness layer: Compaction keeps a limited context useful throughout a long task.

As the Agent works, every file read, command result, and model response remains in messages. The history eventually exceeds the model's context window.

This lesson adds a four-step compaction pipeline. It first reduces recoverable tool output and summarizes history only when those reductions are not enough.

Understanding Context Copy Chapter

Think of the context window as the model's current scratchpad. User messages, model responses, tool_use, and tool_result blocks are written onto it in order. The model reads that material again whenever it continues the task.

The scratchpad has a fixed size. When a request exceeds it, the API rejects the call with prompt_too_long. Tool results usually consume most of the space in coding tasks:

• Reading a long file puts its contents into the context.
• Test and build logs can add tens of kilobytes at once.
• Searching many files keeps appending more results.

As a task continues, messages keeps growing. Compaction controls that growth while preserving the current goal, user constraints, and active work.

Why Tool Results Come First Copy Chapter

Summarizing the whole history can shrink it quickly, but every summary loses some detail and requires another model call.

Tool results are better first targets:

1. A large file result can be stored on disk and read again later.
2. An old command can be run again.
3. The latest results are usually more relevant to the current step.
4. Text trimming and structural edits do not call the model.

The pipeline therefore follows increasing information loss and cost: persist, trim, replace old results, and summarize last.

Step 1: tool_result_budget Copy Chapter

A model response may request several tools at once. Their completed tool_result blocks are written into the final user message together. When their combined content exceeds 200_000 characters, tool_result_budget processes the largest results first.

Each result above LARGE_RESULT_CHAR_LIMIT = 30000 is written in full to:

.task_outputs/tool-results/<tool_use_id>.txt

The context keeps the file path and a 2,000-character preview:

The core loop persists results in descending size order:

blocks = [block for block in content
          if isinstance(block, dict)
          and block.get("type") == "tool_result"]
total = sum(len(str(block.get("content", ""))) for block in blocks)

ranked = sorted(
    blocks,
    key=lambda block: len(str(block.get("content", ""))),
    reverse=True,
)
for block in ranked:
    if total <= max_chars:
        break
    content = str(block.get("content", ""))
    if len(content) <= self.LARGE_RESULT_CHAR_LIMIT:
        continue
    block["content"] = self.persist_large_output(
        block.get("tool_use_id", "unknown"), content)
    total = sum(len(str(item.get("content", ""))) for item in blocks)

This step examines only the latest batch of tool results. The complete output remains available at the saved path, so persistence is the safest operation to run first.

Step 2: snip_compact Copy Chapter

Once the history exceeds 50 messages, snip_compact writes the complete history to .transcripts/, then keeps the first 3 and latest 46 messages. The archive marker occupies the remaining slot, records how many messages were removed, and points to the complete transcript.

head_end = 3
tail_start = len(messages) - (max_messages - head_end - 1)

if self.has_tool_use(messages[head_end - 1]):
    while (head_end < tail_start
           and self.is_tool_result(messages[head_end])):
        head_end += 1

if (tail_start > 0
        and self.is_tool_result(messages[tail_start])
        and self.has_tool_use(messages[tail_start - 1])):
    tail_start -= 1

transcript = self.write_transcript(messages)
marker = {"role": "user", "content":
          f"[{tail_start - head_end} messages archived at {transcript}]"}
messages = [*messages[:head_end], marker, *messages[tail_start:]]

The cut points protect every assistant(tool_use) and user(tool_result) pair. An orphaned result has no matching tool call, so the next API request would be invalid.

This step controls the number of messages. Tool results inside the retained messages may still be long.

Step 3: micro_compact Copy Chapter

After the first two steps, prepare estimates the remaining context size and runs micro_compact only when it is above CONTEXT_CHAR_LIMIT. Among results the model has already consumed, micro_compact keeps the latest 3 and shortens older results longer than 120 characters until the context approaches 80% of the limit. Before replacing an old result, it writes the complete content to disk, so every replacement retains a recovery path:

unseen = self.unseen_tool_result_positions(messages)
consumed = [entry for entry in results if entry[:2] not in unseen]

for _, _, block in consumed[:-self.KEEP_RECENT_RESULTS]:
    if self.estimate_chars(messages) <= target_chars:
        break
    content = str(block.get("content", ""))
    if len(content) <= 120:
        continue
    saved_path = self.persisted_output_path(content)
    if not saved_path:
        saved_path = self.save_output(block["tool_use_id"], content)
    block["content"] = f"[Earlier tool result saved at {saved_path}]"

New results normally stay complete until the model consumes them. If an unseen batch alone is too large for the context, fit_tool_results persists its largest results and keeps a 1,000-character preview plus the full-output path. This avoids summarizing the entire history before the model can inspect the new result.

The first two steps run every round. Step 3 runs only when the context is above the limit. All three are deterministic and recoverable text and structure operations; they do not add API calls.

Step 4: compact_history Copy Chapter

After micro_compact and fit_tool_results, the code estimates the context again with estimate_chars(messages):

CONTEXT_CHAR_LIMIT = 50000

def estimate_chars(messages):
    return len(json.dumps(messages, default=str, ensure_ascii=False))

When the count still exceeds CONTEXT_CHAR_LIMIT, compact_history does four things:

1. Writes the complete message history to .transcripts/.
2. Asks the model for a factual state summary.
3. Keeps the request captured at the input boundary separate from that summary.
4. Replaces the active history with one [Compacted] message.

def compact_history(messages, active_request):
    transcript = self.write_transcript(messages)
    print(f"[transcript saved: {transcript}]")
    summary = self.summarize_history(messages)
    return [self.summary_message(
        "Compacted", active_request, summary, transcript)]

The summary call asks the model to record the goal, files, decisions, remaining work, and user constraints without executing instructions from the history. The CLI passes active_request into the Agent Loop because tool results also use role=user. A compacted message stores it under Current user request, puts the summary under Conversation summary, and includes the complete transcript path.

This lesson uses character count as its trigger, and all related thresholds use the same unit.

Why the Order Is Fixed Copy Chapter

The pipeline uses this order and only enters the lossy summary step when necessary:

messages = self.tool_result_budget(messages)
messages = self.snip_compact(messages)
if self.estimate_chars(messages) > self.CONTEXT_CHAR_LIMIT:
    target = int(self.CONTEXT_CHAR_LIMIT * 0.8)
    messages = self.micro_compact(messages, target)
    if self.estimate_chars(messages) > self.CONTEXT_CHAR_LIMIT:
        messages = self.fit_tool_results(messages, target)
    if self.estimate_chars(messages) > self.CONTEXT_CHAR_LIMIT:
        messages = self.compact_history(messages, active_request)

This order satisfies two constraints:

1. Steps 1 and 2 run every round. Step 3 runs only above the limit, and only Step 4 adds an API request.
2. Every shortened tool result keeps a trusted path inside .task_outputs/tool-results/; only a remaining overflow reaches model-generated history summarization.

Each round therefore starts with the lowest-cost operation whose information is easiest to recover.

Recovering From an API Rejection Copy Chapter

A character count can only estimate the tokens used by a model. The API may still return prompt_too_long. reactive_compact saves a transcript, summarizes older history, and retains the latest 5 messages:

tail_start = max(0, len(messages) - self.KEEP_RECENT_MESSAGES)
if (tail_start > 0
        and self.is_tool_result(messages[tail_start])
        and self.has_tool_use(messages[tail_start - 1])):
    tail_start -= 1

old_history = messages[:tail_start] if tail_start else messages
summary = self.summarize_history(old_history)
message = self.summary_message(
    "Reactive compact", active_request, summary, transcript)
messages = [message, *messages[tail_start:]] if tail_start else [message]

The cut point also avoids splitting a tool call from its result, while active_request carries the current user request explicitly. MAX_REACTIVE_RETRIES = 1 permits one recovery attempt. A second context-length error is raised to the caller.

Putting It Into the Agent Loop Copy Chapter

def agent_loop(messages, active_request):
    while True:
        messages[:] = COMPACTOR.prepare(messages, active_request)

        try:
            response = client.messages.create(
                model=MODEL, system=SYSTEM, messages=messages,
                tools=TOOLS, max_tokens=8000)
            reactive_retries = 0
        except Exception as error:
            message = str(error).lower()
            too_long = ("prompt_too_long" in message
                        or "too many tokens" in message)
            if too_long and reactive_retries < MAX_REACTIVE_RETRIES:
                messages[:] = COMPACTOR.reactive_compact(
                    messages, active_request)
                reactive_retries += 1
                continue
            raise

Every model call enters through the same pipeline. After appending query, the CLI calls agent_loop(history, query), so repeated compaction cannot lose the current request. The code asks for a summary only when micro_compact still leaves the context above the limit or when the API rejects it.

The compact Tool Copy Chapter

An automatic threshold knows only how large the context is. The model can also call compact after completing a stage when the next stage needs only a summary:

{"name": "compact",
 "description": "Summarize earlier conversation to free context space."}

A response may request several tools at once, such as writing a file and then compacting. The Harness first executes the complete batch and appends one tool_result for every tool_use. It summarizes only after that turn is complete:

tool_calls = [
    block for block in response.content if block.type == "tool_use"
]
results = []
compact_requested = False

for block in tool_calls:
    if block.name == "compact":
        output = "Compaction requested after this tool batch."
        compact_requested = True
    else:
        output = execute_tool(block)
    results.append({"type": "tool_result", "tool_use_id": block.id,
                    "content": output})

messages.append({"role": "user", "content": results})

if compact_requested:
    messages[:] = COMPACTOR.compact_history(messages, active_request)

This leaves no orphaned tool result. It also preserves the record of a file write or another side effect before compaction, so the model does not repeat it.

What This Lesson Adds Copy Chapter

Boundary with s09: s08 manages the limited context of the current session and may discard recoverable details. s09 stores information that must survive compaction and future sessions.

Try It Copy Chapter

cd learn-claude-code
python s08_context_compact/code.py

Experiment 1: Replace Earlier Results Copy Section

Read the README.md files from s01_agent_loop through s05_todo_write.
Compare their top-level headings and summarize the naming pattern.

This task produces at least 5 file results. New results normally remain complete until the model sees them once; an oversized unseen result keeps a preview and recovery path instead. On later turns, the latest 3 consumed results remain complete while older long results become [Earlier tool result saved at ...] references.

Experiment 2: Persist a Large Result Copy Section

Analyze the structure of web/src/data/generated/docs.json
and explain the main fields in one lesson record.

When the file exceeds the per-turn budget, the task can still finish and the complete result appears under .task_outputs/tool-results/.

Experiment 3: Trigger an Automatic Summary Copy Section

Compare s08_context_compact/code.py with s09_memory/code.py.
Explain how they manage current context and persistent memory.

When the file results push estimate_chars(messages) above 50000, the terminal prints [auto compact] and a transcript path. The next call continues from the [Compacted] summary.

Inspect .transcripts/ and .task_outputs/tool-results/ to see history archives and persisted large outputs.

What's Next Copy Chapter

Context compaction lets an Agent continue a long task within a limited window. Information that must survive compaction and future sessions needs a separate persistent memory system.

s09 Memory adds memory writing, retrieval, and consolidation.

Overview (`s09_memory/README.md`) Copy Section

s09: Memory — Keep Useful Knowledge Across Sessions

English · 中文 · 日本語

s01 → ... → s07 → s08 → s09 → s10 → s11 → ... → s16 → s17

"Keep information that later tasks will need." File storage + an index + relevance selection + on-demand recall.

Harness layer: Memory stores reusable knowledge outside the conversation and recalls it for related tasks.

The Problem Copy Chapter

An Agent starts a new session without the previous conversation in messages. A coding preference, project fact, or debugging clue from an earlier session may still matter. Without persistent storage, the user has to provide it again.

A complete transcript works as an archive, but sending it with every request does not scale. The conversation keeps growing, useful information becomes hard to locate, and old facts may no longer be true. Memory must decide what is worth keeping across sessions and which records belong in the current task.

Why Not Put Everything in the System Prompt? Copy Chapter

The direct approach is to write preferences and project facts into one file, then put the entire file in the system prompt. It remembers the information, but every LLM call must resend all of it. As the store grows, more unrelated material consumes input tokens and context space.

s07 showed a better reading pattern: keep a short index available and load full content only when needed. Skills are human-authored and read-only. Memory lets the Agent extract information from conversation and reuse it in later work.

This chapter therefore needs four parts: storage, recall, extraction, and consolidation.

Storage: One File per Record Copy Chapter

Each memory is a Markdown file under .memory/. YAML frontmatter stores its name, description, and type:

---
name: user-preference-tabs
description: User prefers tabs for indentation
type: user
---

User prefers using tabs, not spaces, for indentation.

There are four memory types:

MEMORY.md is the index, with one line per memory file. After a write, rebuild_memory_index() regenerates it from the files:

def write_memory_file(name, mem_type, description, body):
    path = MEMORY_DIR / f"{memory_slug(name)}.md"
    path.write_text(
        memory_document(name, mem_type, description, body), encoding="utf-8"
    )
    rebuild_memory_index()
    return path

The index supports selection while full content stays in the individual files.

Recall: Select First, Then Load Full Records Copy Chapter

At the start of a user request, select_relevant_memories() sends the recent user text and memory catalog to a lightweight model call. It selects at most five relevant records:

prompt = (
    "Select memory records that are relevant to the current user request. "
    "Return only a JSON array of catalog indices, such as [0, 2]. "
    "Return [] when none are relevant."
)

If the model call or JSON parsing fails, the code falls back to keyword matching. Only after selection does load_memories() read the corresponding files, with a limit on the total recalled text.

relevant_memories = load_memories(messages)
system = build_system(relevant_memories)

build_system() states that recalled content is background knowledge, not a new user command. The current request wins when it conflicts with memory. This lets the Agent use old information without letting old records issue instructions on the user's behalf.

Extraction: Save Reusable Information After the Turn Copy Chapter

Users do not always say "remember this." After the Agent finishes the current response, extract_memories() inspects the conversation and keeps only information likely to help later:

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
    if extract_memories(messages):
        consolidate_memories()
    return

The model returns candidates, not records that are automatically allowed onto disk. Each candidate carries a scope: only persistent means that the information should survive into later sessions. current_task covers one-off commands, temporary paths, and temporary restrictions.

should_store_memory() performs the final admission check. It rejects incomplete candidates, phrases that refer to the current session or task, and duplicates of existing records. For example, "do not create files in this session" constrains the current work; it must not remain active in the next session.

Consolidation: Merge Duplicate and Stale Records Copy Chapter

As memory files accumulate, some become duplicate, contradictory, or stale. The teaching implementation calls consolidate_memories() after the store reaches ten records and asks the model for a cleaned list.

The code parses and validates the new list before replacing old files. It snapshots the current records first; if deletion or writing fails, it restores the originals and rebuilds the index:

snapshot = {
    path.name: path.read_text(encoding="utf-8")
    for path in MEMORY_DIR.glob("*.md")
    if path.name != MEMORY_INDEX.name
}

try:
    for path in MEMORY_DIR.glob("*.md"):
        if path.name != MEMORY_INDEX.name:
            path.unlink()
    for record in consolidated:
        path = MEMORY_DIR / f"{memory_slug(record['name'])}.md"
        path.write_text(memory_document(
            record["name"], record["type"],
            record["description"], record["body"],
        ), encoding="utf-8")
    rebuild_memory_index()
except Exception:
    for path in MEMORY_DIR.glob("*.md"):
        if path.name != MEMORY_INDEX.name:
            path.unlink()
    for filename, content in snapshot.items():
        (MEMORY_DIR / filename).write_text(content, encoding="utf-8")
    rebuild_memory_index()
    raise

The course uses a simple count threshold. A real application must also choose a schedule that fits its data volume and prevent concurrent processes from rewriting the same store.

This Lesson's Code Copy Chapter

Boundary with s08: s08 manages the active session's context budget. s09 manages reusable knowledge outside the conversation. Memory is selective storage, not a lossless transcript backup, and it does not replace context compaction.

Try It Copy Chapter

cd learn-claude-code
python s09_memory/code.py

1. Enter I prefer using tabs for indentation. Remember that. After the turn, check that .memory/ contains a new record and MEMORY.md contains its index entry.
2. Enter q, restart the program, and ask What indentation style do I prefer? Confirm that a new session can recall the preference.
3. Store another preference unrelated to code formatting, then ask about indentation. Observe that the current request loads only relevant records.
4. Enter Do not create files in this session. Confirm that this temporary requirement does not become a persistent rule for the next session.

Exact wording and extraction counts can vary by model. Check what was written to .memory/ and whether a later session recalls only relevant information.

What's Next Copy Chapter

Memory preserves information across sessions, but a complex task also needs durable status and dependency tracking. A TODO kept only in the conversation cannot carry progress across process restarts.

s10 Task System → Persist tasks, statuses, and dependencies to disk.

Overview (`s10_task_system/README.md`) Copy Section

s10: Task System — From an Execution Checklist to Coordinated Task State

English · 中文 · 日本語

s01 → ... → s08 → s09 → s10 → s11 → s12 → ... → s16 → s17

"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 Copy Chapter

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 Copy Chapter

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:

How It Works Copy Chapter

Task: Data Structure Copy Section

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 IDs

IDs 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 Copy Section

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 Copy Section

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 Copy Section

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 Copy Section

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 Copy Section

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 msg

After completing "schema", can_start returns True for "endpoints" and "docs"; they can begin.

get_task: View Full Details Copy Section

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 Copy Section

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 Copy Section

# 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)     # ✓ Completed

Each 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 Copy Chapter

cd learn-claude-code
python s10_task_system/code.py

Try these prompts:

1. Create tasks: setup database schema, create API endpoints (depends on schema), write tests (depends on endpoints), write docs (depends on schema)
2. List all tasks and their statuses
3. Claim the first unblocked task and complete it
4. List 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?

What's Next Copy Chapter

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.

Overview (`s11_background_tasks/README.md`) Copy Section

s11: Background Tasks — Slow Operations Go to the Background

English · 中文 · 日本語

s01 → ... → s09 → s10 → s11 → s12 → s13 → ... → s16 → s17

"Slow operations go to the background, the Agent Loop continues" — Background threads run commands, and later turns collect completed results.

Harness Layer: Background — Async execution, doesn't block the main loop.

The Problem Copy Chapter

Reading a file or running git status usually returns quickly, so synchronous execution causes little noticeable delay. Installing dependencies, running a full test suite, or building a project can take several minutes. Until the command returns, the Harness cannot process the next tool call in the current response or start the next model turn.

If later work does not depend on that command, there is no need to block it. For example, after starting a full test suite, the Agent could inspect documentation or organize other files while the tests run.

S11 addresses this by running slow Bash commands in the background, allowing the Agent Loop to continue and collect completed results on a later turn.

The Solution Copy Chapter

This chapter sends slow operations to background threads. The current tool call first returns a placeholder tool_result, allowing the Agent Loop to continue. At the start of a later turn, completed results are collected and added to the conversation as notifications.

Sync vs Background:

How It Works Copy Chapter

should_run_background: Explicit Request Copy Section

The model requests background execution through the bash tool's run_in_background parameter. Only bash calls with the parameter explicitly set to true enter this path. Other calls still run synchronously.

def should_run_background(tool_name: str, tool_input: dict) -> bool:
    return (
        tool_name == "bash"
        and tool_input.get("run_in_background") is True
    )

The Harness no longer guesses from keywords such as install, build, or test. The tool call chooses the execution mode explicitly.

BackgroundManager: Background Execution and Lifecycle Copy Section

BackgroundManager owns task state and the completion queue. start() registers a task, starts a daemon thread, and returns bg_id immediately:

class BackgroundManager:
    def __init__(self):
        self.tasks = {}
        self.results = {}
        self._ready = []
        self._lock = threading.Lock()

    def start(self, block) -> str:
        # Register task, then run _run() in a daemon thread.
        ...

    def _run(self, task_id: str, command: str):
        output, exit_code = _run_bash_process(command)
        status = "completed" if exit_code == 0 else "failed"
        with self._lock:
            self.tasks[task_id]["status"] = status
            self.results[task_id] = _format_bash_result(output, exit_code)
            self._ready.append(task_id)

A non-zero exit code or worker exception becomes failed. The shell starts in its own process group. When the command finishes, times out, or the Agent exits through the normal or SIGTERM path, the runtime stops that original group. This is lifecycle cleanup, not a sandbox: a process that creates another session can leave the group.

collect_background_results: Notification Collection Copy Section

At the start of a later turn, collect() removes completed results from the queue and formats them as <task_notification> messages:

def collect_background_results() -> list[str]:
    return BACKGROUND.collect()

Notifications don't reuse the original tool_use_id. The original tool call was already answered with a placeholder tool_result; when the completed result is collected, it is added as an independent event in task_notification format. One tool_use still gets exactly one tool_result.

Loop Integration Copy Section

Before each LLM call, the Agent Loop collects completed background results. execute_tool() still runs PreToolUse on the main thread before choosing synchronous or background execution:

while True:
    inject_background_results(messages)
    response = client.messages.create(...)

def execute_tool(block) -> str:
    blocked = trigger_hooks("PreToolUse", block)
    if blocked is not None:
        return str(blocked)
    if should_run_background(block.name, block.input):
        task_id = start_background_task(block)
        output = f"[Background task {task_id} started]"
    else:
        output = call_tool(block)
    trigger_hooks("PostToolUse", block, output)
    return output

Slow operations first return a placeholder tool_result with bg_id. A completed task does not wake the Agent by itself; inject_background_results() collects it the next time the Agent Loop runs.

Putting It Together Copy Section

Turn 1:
  LLM → bash "npm install" (run_in_background=true)
  → start_background_task → bg_0001
  → tool_result: "[Background task bg_0001 started]..."
  → LLM: "OK, I'll check later. Let me also read the config."

Turn 2:
  LLM → read_file "package.json" (fast, sync)
  → tool_result: file content

Turn 3:
  → collect bg_0001 as <task_notification>
  → LLM sees: config file + install notification in one message

While npm install ran in the background, the Agent Loop continued with read_file.

What s11 Adds Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python s11_background_tasks/code.py

Try these prompts:

1. Run pip list in the background and find all Python files in this directory
2. Run npm install (use run_in_background) and while waiting, read package.json
3. Run a short sleep in the background, then list all Markdown files

What to observe: After explicitly setting run_in_background, is the command dispatched to the background? Is a bg_id returned? Are completed results collected in <task_notification> format on a later turn?

What's Next Copy Chapter

Background tasks solved "slow operations don't block." But what if you want to do something on a schedule? Like "run tests every morning at 9am" or "check server status every 5 minutes."

s12 Cron Scheduler → Give the agent an alarm clock.

Overview (`s12_cron_scheduler/README.md`) Copy Section

s12: Cron Scheduler — Start Work on a Schedule

English · 中文 · 日本語

s01 → ... → s10 → s11 → s12 → s13 → ... → s17

The Problem Copy Chapter

S11 changes how a command runs after it starts: a long Bash command can run in the background. It does not record when future work should start, and no component keeps checking the current time.

For requests such as "run tests every morning at 9am" or "check CI status every 30 minutes," the user would still have to submit the prompt again at each scheduled time. The Harness needs to store the schedule, put the corresponding prompt into a pending queue when it becomes due, and deliver it to the Agent Loop when the Agent is idle.

The Solution Copy Chapter

Suppose the Agent registers this job:

cron:   0 9 * * *
prompt: run tests

At 09:00 local time, the scheduler thread matches the job and puts [Scheduled] run tests into cron_queue. The queue processor waits until the Agent is idle, then starts an Agent Loop turn. The model can then call Bash to run the tests.

The S12 code keeps the five base tools and Hooks from S04, then adds schedule_cron, list_crons, and cancel_cron. It does not include S11 background commands because this chapter delivers a prompt to start work, not the result of a command that is already running.

How It Works Copy Chapter

What CronJob stores Copy Section

@dataclass
class CronJob:
    id: str
    cron: str
    prompt: str
    recurring: bool
    durable: bool
    pending_delivery: bool = False
    last_fired: str | None = None

cron controls when the job becomes due. prompt is the task sent to the Agent. pending_delivery marks a due job that the model has not accepted, while last_fired prevents another enqueue in the same minute.

Five-field cron expressions Copy Section

minute hour day month weekday
   *     *    *    *      *    every minute
   0     9    *    *      *    every day at 09:00
  */5    *    *    *      *    every 5 minutes
   0     9    *    *     1-5   weekdays at 09:00

This chapter supports *, */N, N, N-M, and N,M,.... Before saving a job, schedule_job() calls validate_cron() and rejects expressions with the wrong number of fields or out-of-range values.

Enqueue when due Copy Section

The scheduler thread reads local time once per second. When an expression matches and the job has not fired in the current minute, _enqueue_due_job() saves pending_delivery and last_fired before adding the job to the in-memory queue:

def poll_due_jobs(moment: datetime):
    minute_marker = moment.strftime("%Y-%m-%d %H:%M")
    with cron_lock:
        for job in list(scheduled_jobs.values()):
            if job.pending_delivery or job.last_fired == minute_marker:
                continue
            if cron_matches(job.cron, moment):
                _enqueue_due_job(job, minute_marker)

If persistence fails, _enqueue_due_job() restores the previous state and does not expose a memory-only delivery to the queue processor.

Deliver when the Agent is idle Copy Section

queue_processor_loop() does not check the time. It checks the queue, and agent_lock prevents a scheduled turn from changing the session while a user turn is running:

def queue_processor_loop(stop_event=RUNTIME_STOP):
    while not stop_event.wait(0.2):
        if not has_cron_queue() or not agent_lock.acquire(blocking=False):
            continue
        try:
            if has_cron_queue():
                run_agent_turn_locked()
        finally:
            agent_lock.release()

The Agent Loop takes due jobs from the queue and appends each one as a new user message:

fired = consume_cron_queue()
for job in fired:
    messages.append({"role": "user", "content": f"[Scheduled] {job.prompt}"})

If the model call fails, those messages are removed from the current session and the jobs return to the queue. Once the model accepts the call, one-shot jobs are removed and recurring jobs clear pending_delivery until the next match.

Persistence boundary Copy Section

The code updates .scheduled_tasks.json through a temporary file and os.replace(). If the file is corrupt, startup reports the error instead of ignoring it.

Delivery is at least once. If the process exits after the model accepts a prompt but before the acknowledgement reaches disk, the same job may be delivered again after restart.

Runtime boundary Copy Section

• The scheduler uses the Agent process's local time.
• The scheduler stops when the Agent process exits. durable preserves the job definition only.
• Restart loads saved jobs but does not replay schedule times missed while the process was down.
• Scheduled turns run in the queue processor thread. A tool call that needs interactive approval is denied instead of competing with the main terminal for input.
• Scheduler and queue processor threads start only in the CLI. Importing code.py starts no background thread.

Use crontab, a systemd timer, or an external scheduler when jobs must run while the Agent is closed.

Try It Copy Chapter

cd learn-claude-code
python s12_cron_scheduler/code.py

Enter these prompts in order:

1. Schedule "run date" every 2 minutes and keep it after restart.
2. List all cron jobs.
3. Cancel the cron job you just created.

You can inspect .scheduled_tasks.json and watch for the [Scheduled] run date message when the job becomes due. Keep the Agent process running while testing a minute-level schedule.

What's Next Copy Chapter

The scheduler can start an Agent Loop turn at a specified time, but one Agent still handles that turn. When a task requires parallel investigation, changes across multiple modules, and a combined result, the Harness also needs to assign work to multiple Agents and collect what each one produces.

s13 Agent Teams → A Lead assigns tasks, teammates run independently, and results return through inboxes.

Overview (`s13_agent_teams/README.md`) Copy Section

s13: Agent Teams — Runtime and Coordination Protocols

English · 中文 · 日本語

s01 → ... → s10 → s13 → s14 → s15 → s16 → s17

"When one agent cannot hold the whole job, let teammates divide the work." — Persistent teammates, shared task selection, optional worktrees, and coordination protocols.

Harness layer: Team — how multiple agents divide work, share state, and stay under Lead's control.

The Problem Copy Chapter

Suppose we ask an agent to refactor an entire backend. The work may cover configuration loading, authentication, and tests. One agent can process those areas sequentially, but it takes longer and earlier details gradually leave its context.

This is a good candidate for parallel work, yet users normally describe the goal rather than design the team:

Refactor this sample backend. Clean up configuration loading,
authentication, and tests, preserve the existing interfaces,
and make sure the tests pass.

The harness has to answer a connected set of questions:

1. Who decides that parallel work is useful, and who confirms the extra agents?
2. How does each teammate keep its identity and context across assignments?
3. How do results return to Lead without asking the model to poll an inbox?
4. Can an idle teammate pick up ready work without waiting for another assignment?
5. Which directory should a task use when parallel edits may conflict?
6. How do shutdown and plan approval become traceable, enforceable protocols?

The Solution Copy Chapter

s13 reuses s10's base tools, hooks, permission checks, and Task System, then adds a Lead-managed team runtime:

• Lead owns the user conversation, proposes a division of work, and waits for confirmation.
• Teammates run independent agent loops and alternate between WORK and IDLE.
• MessageBus carries ordinary messages, results, and control events through file-backed mailboxes.
• Runtime delivery consumes Lead's mailbox and injects team events into the next turn.
• The shared task board lets idle teammates find ready work and claim it under a lock.
• Optional worktrees bind a task to another working directory when the work needs it. Unbound tasks use the normal repository directory.
• Typed protocols and a plan gate make shutdown and approval state explicit and block mutating tools until a required plan is approved.

Task graph authoring keeps s10's two-phase contract. The Lead first calls create_task for every node, then uses the returned runtime IDs with update_task(addBlockedBy=...) before assigning ready work. Only the Lead receives update_task; teammates can list, claim, and complete tasks but cannot rewrite graph structure while the team is running.

s11 background tasks and s12 scheduled tasks are not carried into this chapter. Neither mechanism is required for teammate communication, task claiming, or plan approval.

These are all parts of the Team harness layer. Teammates do not need a separate loop for task discovery, and a worktree does not create a new kind of agent.

How It Works Copy Chapter

1. Lead proposes a team and waits for user confirmation Copy Section

Starting teammates changes cost, concurrency, and the set of actors that may edit the workspace. Lead's system prompt keeps that boundary visible:

"When parallel work would help, first propose a small team with clear "
"responsibilities and wait for the user's confirmation. Do not call "
"spawn_teammate before the user confirms."

For the first request, Lead only proposes a split:

I suggest three parallel areas:
- config: clean up configuration loading
- auth: refactor authentication
- tests: add regression coverage

I will start the teammates after you confirm.

After the user says "Go ahead," Lead can call spawn_teammate. Lead creates the Task first and passes its initial task_id to the teammate. The user states the goal, Lead designs the team, and the user confirms the execution boundary.

2. Every teammate owns an independent loop Copy Section

An s06 subagent is a one-shot call. A teammate is a persistent execution unit:

TeammateRuntime gives each teammate its own system prompt, messages, tools, and current Task, then runs its WORK / IDLE loop in a daemon thread. Lead can keep coordinating while teammates work. The names lead and agent are reserved for runtime identities, while MessageBus still accepts lead as the coordinator mailbox.

spawn_teammate claims the initial Task before the thread starts. A failed claim prevents the teammate from starting. Without a Task, workspace and Shell tools ask the teammate to claim one instead of falling back to the repository directory.

3. MessageBus keeps communication outside model context Copy Section

Lead and teammates cannot share one messages array. Otherwise one teammate's tool results would leak into another teammate's reasoning. MessageBus gives each agent a .mailboxes/<name>.jsonl inbox:

class MessageBus:
    def send(self, from_agent, to_agent, content,
             msg_type="message", metadata=None):
        msg = {
            "from": from_agent,
            "to": to_agent,
            "content": content,
            "type": msg_type,
            "metadata": metadata or {},
        }
        with self._changed:
            MAILBOX_DIR.mkdir(parents=True, exist_ok=True)
            with self._path(to_agent).open("a", encoding="utf-8") as handle:
                handle.write(json.dumps(msg, ensure_ascii=True) + "\n")
            self._changed.notify_all()

    def wait_for_messages(self, agent, timeout=None):
        deadline = None if timeout is None else time.monotonic() + timeout
        with self._changed:
            while not self.peek(agent):
                remaining = (None if deadline is None
                             else deadline - time.monotonic())
                if remaining is not None and remaining <= 0:
                    return []
                self._changed.wait(remaining)
            return self._read_unlocked(agent)

A lock protects mailbox files from concurrent access. A Condition lets the runtime wake a teammate for a message and also supports the short timeout used while IDLE.

4. The runtime delivers inbox events Copy Section

read_inbox() consumes messages by reading and deleting the mailbox file, so Lead keeps a single consumer, consume_lead_inbox():

def consume_lead_inbox():
    messages = BUS.read_inbox("lead")
    for message in messages:
        if message["type"].endswith("_response"):
            match_response(...)
    return messages

The CLI loop waits for terminal input and Lead's mailbox at the same time. When a message arrives, it consumes the mailbox before starting another Lead turn:

MessageBus → consume_lead_inbox
           → update protocol state
           → inject [Team events] into history
           → start another Lead turn

After spawning a teammate, Lead ends the current turn instead of repeatedly calling list_teammates or get_task. The runtime starts the next turn when a team event arrives.

check_inbox is not a model tool. Message arrival belongs to the runtime; the model handles events after the runtime has delivered them into its context.

5. Result and IDLE are separate events Copy Section

When a teammate finishes one assignment, the runtime sends two events in order:

result:            "Authentication refactored; related tests pass."
idle_notification: "Waiting for more work."

result answers "What did this assignment produce?" idle_notification answers "Can this teammate accept more work?" One vague "done" cannot represent both facts.

An idle teammate does not exit. A direct message or a ready task returns it to WORK; a shutdown_request starts a graceful shutdown handshake.

6. IDLE checks the mailbox before looking for ready tasks Copy Section

IDLE gives messages priority, then checks the shared task board:

while True:
    inbox = BUS.wait_for_messages(name, IDLE_SCAN_INTERVAL)
    if inbox:
        should_stop = handle_messages(inbox)
        if should_stop or messages[-1]["role"] == "user":
            break
        continue

    task = claim_next_task(name)
    if task:
        messages.append({
            "role": "user",
            "content": f"[Auto-claimed task {task.id}] {task.subject}",
        })
        break

Shutdown, plan approval, and direct instructions from Lead should arrive before opportunistic work. If there is no message and no ready task, the teammate remains IDLE. A blocked task may become ready after another teammate completes its prerequisite.

7. Discovery and claim are separate, and claim is atomic Copy Section

Scanning only finds candidates:

def scan_unclaimed_tasks() -> list[Task]:
    return [
        task for task in list_tasks()
        if task.status == "pending"
        and task.owner is None
        and can_start(task.id)
    ]

The list is a snapshot. Another teammate, or another harness process using the same task directory, may see the same task. Ownership changes therefore happen inside claim_task() under task_store_lock(), which combines the in-process lock with a file lock:

def claim_task(task_id: str, owner: str) -> str:
    with task_store_lock():
        task = load_task(task_id)
        if task.status != "pending" or task.owner is not None:
            return "Task is no longer available"
        if _owner_in_progress(owner):
            return "Owner must complete its current task first"
        if not can_start(task_id):
            return "Task is blocked"
        cwd, error = task_worktree_cwd(task)
        if error:
            return f"Cannot claim {task_id}: {error}"
        task.owner = owner
        task.status = "in_progress"
        save_task(task)
        teammate_assignments[owner] = {"task_id": task.id, "cwd": cwd}
        return f"Claimed {task.id}"

Many teammates may discover the same candidate, but only one claim can move it to in_progress. Task files are written through a temporary file and atomically replaced while the same store lock is held. A teammate must also finish its current task before claiming another, and a broken worktree binding fails closed rather than falling back to the repository directory.

8. Claimed work reuses the same WORK loop Copy Section

After a successful claim, the runtime injects the task ID, subject, and description into the teammate's messages:

ready task appears
  → IDLE teammate discovers it
  → claim_task writes owner and in_progress
  → task enters teammate messages
  → WORK
  → complete_task
  → result + idle_notification
  → IDLE

The teammate uses the same model call, file tools, Shell, plan gate, result reporting, and shutdown protocol as a direct Lead assignment. Task discovery is another entry into the existing WORK loop.

9. The task selects the tools' working directory Copy Section

Task.worktree is optional:

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str
    owner: str | None
    blockedBy: list[str]
    worktree: str | None = None

Lead can create and bind a worktree when separate directories will help:

create_worktree(name="auth-refactor", task_id="task_1a2b3c4d")

create_worktree is a Lead-only tool. It accepts a pending, unowned, unbound task, validates the name, path, branch, and Git registry, creates the checkout, then writes the task binding. If Git reports failure after leaving a branch or registered checkout, the runtime reports a partial operation, leaves the task unbound, and preserves those artifacts for manual recovery. Teammates only see task and file tools.

Claiming the task stores its resolved directory in teammate_assignments; that teammate's bash, read_file, write_file, edit_file, and glob wrappers read the directory from the assignment. A task with no worktree resolves to WORKDIR; a teammate without a claimed Task cannot use those workspace tools:

cwd, error = task_worktree_cwd(task)
if not error:
    teammate_assignments[owner] = {
        "task_id": task.id,
        "cwd": cwd,
    }

complete_task(task_id, owner) checks that the caller owns the in-progress task. Successful completion records the result but keeps the assignment directory selected until that model turn ends. This lets later tool calls in the same response stay in the task's worktree. The runtime releases the assignment when the teammate returns to IDLE; a failed completion keeps it so the teammate can fix the task and try again.

After a restart, assignment_cwd() can rebuild an in-progress assignment from the durable task owner and worktree binding. It also replaces a stale local lease when the same owner has moved to another task. A missing or invalid binding fails closed instead of silently routing work to the repository directory.

A worktree separates Git working directories and branches. It is not a sandbox: Shell commands can still access paths and resources allowed to the parent process.

10. Worktree removal belongs to the host Copy Section

The model can create a task-bound worktree, but it cannot remove one. Cleanup remains a host helper so the user or host can first inspect task ownership, the assignment lease, and Git status. The helper refuses pending or in-progress task bindings and current-turn leases. Without an explicit destructive choice, tracked, untracked, and ignored files all block removal.

remove_worktree(name, discard_changes=True) is reserved for host code that has already obtained explicit user confirmation. Either removal path retains the wt/<name> branch, including clean local commits with no upstream. A successful removal clears the task binding because the checkout no longer exists.

clean worktree   → host may remove directory and retain wt/<name> branch
changed worktree → user decides how to preserve or discard it
pending/running task → refuse removal

Task completion also stays separate from worktree cleanup. complete_task records the task result; after the teammate reaches IDLE, the user or host can inspect, merge, keep, or remove the worktree.

11. Control messages use types and request IDs Copy Section

Free-form text works for ordinary collaboration, but shutdown and approval should not depend on guessing intent. They use structured messages:

@dataclass
class ProtocolState:
    request_id: str
    type: str
    sender: str
    target: str
    status: str
    payload: str
    work_version: int | None = None
    task_id: str | None = None


pending_requests: dict[str, ProtocolState] = {}

The shutdown path is:

Lead creates a pending shutdown request
  → shutdown_request(request_id) enters the teammate inbox
  → the teammate finishes its current step
  → shutdown_response(request_id) returns to Lead
  → request_id locates the original request
  → pending becomes approved and the teammate loop exits

The ID correlates one reply with one request, the type prevents a mismatched reply from changing state, and the status prevents duplicate responses from being applied twice.

12. Plan approval constrains execution Copy Section

The plan protocol runs in the opposite direction:

Lead → plan_request
teammate → plan_approval_request(request_id, plan)
Lead → plan_approval_response(request_id, approve, feedback)

When Lead already knows that a teammate must plan first, spawn_teammate(..., task_id=task.id, require_plan=True) claims the Task and activates the gate before the teammate thread starts. request_plan can also require a plan from a teammate that is already running.

Tool dispatch enforces the gate:

def _run_teammate_tool(name, block, handlers):
    gate = plan_gates.get(name, "not_required")
    if block.name in {"bash", "write_file", "edit_file"} and gate not in {
        "not_required", "approved"
    }:
        return f"Blocked: plan status is {gate}."
    try:
        return handlers[block.name](**block.input)
    except Exception as error:
        return f"Error: {type(error).__name__}: {error}"

While the state is required, pending, or rejected, the teammate can read files and submit or revise a plan, but it cannot run Shell commands, write files, or edit files. A submitted plan records the teammate's current task and work version. Claiming or releasing a Task changes that version and invalidates the old approval; an ordinary message changes neither the task identity nor the approval state.

Teammates do not read user input from their background threads. A dangerous command or path outside the workspace returns a permission error so Lead can handle the decision with the user.

One Complete Run Copy Chapter

s13 >> Put the backend refactor on a shared task board. Clean up
       configuration, authentication, and tests in parallel where possible.
       Use a worktree for authentication, preserve existing interfaces,
       and make sure the tests pass.

Lead: I suggest config, auth, and tests as three areas.
      Shall I start the team?

s13 >> Go ahead.

[task] config created
[task] auth created → worktree auth-refactor
[task] tests created
[claim] alice → config (cwd: repository)
[claim] bob → auth (cwd: .worktrees/auth-refactor)
[teammate] alice spawned
[teammate] bob spawned
[complete] auth
[bus] bob → lead (result) ...
[bus] bob → lead (idle_notification) ...
[wake: 2 team events → new turn]
Lead: I received the authentication result and will coordinate the rest.

The terminal exposes the user request, Lead's proposal, task state, claims, selected directories, results, IDLE transitions, and control events. The user does not have to name a Lead or ask it to check an inbox.

What Changed from s10 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python s13_agent_teams/code.py

Start with an ordinary request:

Put the backend refactor on a shared task board. Complete configuration,
authentication, and tests in parallel where dependencies allow. Use a
worktree for authentication, preserve existing interfaces, and summarize
the result.

After Lead proposes the team, reply:

Go ahead.

Watch .tasks/ move from pending to in_progress and completed, .mailboxes/ deliver result and idle_notification, and .worktrees/ appear only for the bound task. Also check that direct messages beat task-board scans and that a failed complete_task does not reset the teammate's working directory.

What's Next Copy Chapter

The Lead and its teammates can only call tools defined directly in code.py. Connecting Jira, a deployment platform, or a knowledge base still requires separate tool schemas and handlers for each external system. Changes to those external tools also require changes to the course code.

s14 MCP Tools → Connect external services at runtime through one discovery and invocation protocol, then add their tools to the tool pool.

Overview (`s14_mcp_plugin/README.md`) Copy Section

s14: MCP Tools — Discover and Invoke External Tools

English · 中文 · 日本語

s04 → s14 → s15 → s16 → s17

Harness layer: MCP Tools — connect to services, discover tools, and add them to the agent loop.

The Problem Copy Chapter

The base tools in earlier chapters are written directly in code.py. We could integrate a documentation system and deployment platform by adding search_docs, deploy_status, and trigger_deploy, but every service would require another set of tool definitions, parameter schemas, and call handlers.

MCP separates those responsibilities. A server provides a tool list and invocation endpoint. The harness connects to it, assigns model-facing names, applies permission checks, and gives the discovered tools to the model.

The Solution Copy Chapter

This chapter starts from s04's five base tools and hooks, then adds three parts:

• MCPClient stores the tool definitions and call handlers returned by a server.
• connect_mcp connects to one server and obtains its tool list.
• assemble_tool_pool combines the base tools with tools from every connected server.

The docs and deploy servers are in-process stand-ins for tools/list, tools/call, and a dynamic tool pool. This chapter does not implement a real MCP transport.

How It Works Copy Chapter

1. The base agent loop stays the same Copy Section

Before each model call, the harness assembles the current tool pool:

def agent_loop(messages: list):
    while True:
        tools, handlers = assemble_tool_pool()
        response = client.messages.create(
            model=MODEL,
            system=assemble_system_prompt(),
            messages=messages,
            tools=tools,
            max_tokens=8000,
        )
        ...

After a new server connects, the next assemble_tool_pool() call adds its tools to the model input. Tool results are still appended to messages as tool_result blocks.

2. MCPClient stores discovery results and call handlers Copy Section

class MCPClient:
    def register(self, tool_defs, handlers):
        self.tools = list(tool_defs)
        self._handlers = dict(handlers)

    def call_tool(self, tool_name, args):
        handler = self._handlers.get(tool_name)
        if not handler:
            return f"MCP error: unknown tool '{tool_name}'"
        try:
            return str(handler(**args))
        except Exception as error:
            return f"MCP error: {type(error).__name__}: {error}"

register() represents the discovered tool list. call_tool() represents the invocation boundary. Errors return to the model instead of terminating the agent loop.

3. connect_mcp only connects and discovers Copy Section

def connect_mcp(name: str) -> str:
    if name in mcp_clients:
        return f"MCP server '{name}' already connected"
    factory = MOCK_SERVERS.get(name)
    if not factory:
        return f"Unknown server '{name}'"
    server = factory()
    mcp_clients[name] = server
    ...

Initially, the model sees the five base tools and connect_mcp. After connect_mcp(name="docs"), the harness stores the docs client. The next model call also sees:

mcp__docs__search
mcp__docs__get_version

4. Prefixes separate tools from different servers Copy Section

Several servers may expose search or status. The harness uses:

mcp__{server}__{tool}

normalize_mcp_name() replaces characters outside the model tool-name alphabet with underscores. Tool-pool assembly also checks normalized-name collisions and the 64-character limit:

prefixed = f"mcp__{safe_server}__{safe_tool}"
if prefixed in origins:
    raise ValueError("MCP tool name collision after normalization")

As a result, docs.one/get.version and docs_one/get_version cannot silently map to the same name.

5. Tool definitions and handlers enter the pool together Copy Section

tools.append({
    "name": prefixed,
    "description": tool_def.get("description", ""),
    "input_schema": schema,
})
handlers[prefixed] = (
    lambda *, client=server, tool=raw_name, **kwargs:
    client.call_tool(tool, kwargs)
)

The model sees the prefixed name. The handler calls MCPClient with the server's original tool name. Default arguments capture the current client and tool so every lambda does not point to the last item in the loop.

6. The host decides permissions Copy Section

An MCP server may provide readOnlyHint or destructiveHint, but those hints come from the server and are not authorization. This chapter uses a host-side policy:

MCP_HOST_POLICY = {
    ("docs", "search"): "allow",
    ("docs", "get_version"): "allow",
    ("deploy", "status"): "allow",
    ("deploy", "trigger"): "confirm",
}

permission_hook() looks up this policy using the normalized tool name. An unconfigured external tool requires confirmation by default. A description containing readOnly does not make a tool trusted.

7. Input errors stay at the tool boundary Copy Section

The model may omit a required argument or send a field the server does not accept. Both execute_tool() and MCPClient.call_tool() catch those errors and return an error tool_result:

MCP error: TypeError: <lambda>() missing 1 required argument: 'query'

The model can correct its arguments on the next turn without terminating the lesson script.

What Changed from s04 Copy Chapter

This chapter does not carry Task, Background, Cron, Team, or Worktree. They join MCP in the s15 Integrated Harness.

Try It Out Copy Chapter

cd learn-claude-code
python s14_mcp_plugin/code.py

Enter:

Connect to the docs server, search for agent hooks, and tell me the current documentation API version.

A typical tool trace is:

connect_mcp(name="docs")
mcp__docs__search(query="agent hooks")
mcp__docs__get_version()

Then enter:

Connect to the deploy server and check the web service status. Do not trigger a deployment.

status runs under the host policy. trigger requires user confirmation.

What's Next Copy Chapter

MCP is still an independent course branch here. s15 Integrated Harness combines the base tools, hooks, skills, context, memory, tasks, background work, cron, teams, and MCP in one runtime.

Overview (`s15_integrated_harness/README.md`) Copy Section

s15: Integrated Harness — Many Mechanisms, One Loop

English · 中文 · 日本語

s01 → ... → s13 → s14 → s15 → s16 → s17

"Many mechanisms, one loop" — tools, permissions, memory, tasks, teams, and plugins all hang off the same while True.

Harness layer: Integration — put the mechanisms used by this example into one runnable system.

Problem Copy Chapter

The earlier chapters keep separate mechanisms in separate runnable examples. This chapter connects the mechanisms needed by the integrated runtime.

A long-running coding agent needs all of these at once:

• tool dispatch and permission boundaries
• hook extension points
• todo planning and task graphs
• skills, memory, and runtime system prompt assembly
• compaction and error recovery
• background tasks and cron scheduling
• teams, protocols, and IDLE task claiming
• task-bound worktrees
• MCP external tool integration

S15 does not introduce another isolated mechanism. It shows where the existing mechanisms enter the model loop and how their events return to the same conversation.

Solution Copy Chapter

S15 does not introduce a new mechanism. It connects the components from the earlier chapters in one integrated harness:

user input
  → UserPromptSubmit hooks
  → cron/background notification injection
  → context compact
  → memory + skills + MCP state assemble the system prompt
  → LLM
  → has tool_use block?
      no  → Stop hooks → return
      yes → PreToolUse hooks + permission
          → TOOL_HANDLERS / MCP handlers / background dispatch
          → PostToolUse hooks
          → tool_result / task_notification back to messages
          → next round

The loop keeps the same structure: call the model, check whether the response contains a tool_use block, execute tools, and append results to messages. The presence of a tool_use block decides whether tool execution continues.

Where Each Component Sits Copy Chapter

What code.py Contains Copy Chapter

Tools and Dispatch Copy Section

The built-in tool pool contains 26 tools:

bash, read_file, write_file, edit_file, glob
todo_write, task, load_skill, compact
create_task, update_task, list_tasks, get_task, claim_task, complete_task
schedule_cron, list_crons, cancel_cron
spawn_teammate, list_teammates, send_message
request_shutdown, request_plan, review_plan
create_worktree
connect_mcp

assemble_tool_pool() assembles these every round:

BUILTIN_TOOLS + connected MCP tools
BUILTIN_HANDLERS + mcp__server__tool handlers

After connect_mcp("docs"), the next round exposes tools like mcp__docs__search.

Permissions and Hooks Copy Section

Permission is not hardcoded into the tool execution line. It is a PreToolUse hook:

blocked = trigger_hooks("PreToolUse", block)
if blocked:
    results.append(tool_result(block.id, blocked))
    continue

That means permission, logging, and audit logic all attach to the same hook point. Lead tools, one-shot subagent tools, and teammate tools all pass through PreToolUse; an allowed call then runs PostToolUse after its handler.

The policy does not trust an MCP server's own description as authorization. The host owns a small exact allowlist for known read-only calls; every other MCP tool asks the user. File tools are denied outside WORKDIR, and every bash command asks before execution. Only the foreground user turn may open an interactive approval prompt; asynchronous turns fail closed instead of competing with the main CLI for stdin.

Planning and Tasks Copy Section

S15 keeps two planning layers:

• todo_write: lightweight plan for the current session, kept in memory
• task graph: cross-session, dependency-aware, claimable task files under .tasks/task_*.json

The first keeps a single agent from drifting. The second supports team coordination.

They share an intent, not an implementation: todo_write replaces one session checklist, while task records have stable IDs and individual lifecycle updates. The separate task tool below means "dispatch one isolated subagent"; it is not the Task System.

Task graph construction remains two-phase in the integrated host: the Lead creates all task nodes first, then calls update_task with the runtime IDs returned by create_task. Teammates receive only list, claim, and complete operations, so dependency structure is fixed by the Lead before work is distributed.

Subagents and Teams Copy Section

S15 has two kinds of delegation:

• task: one-shot subagent. It uses an isolated messages[], discards intermediate context, and returns only a final summary.
• spawn_teammate: persistent teammate thread. When given a ready task_id, the runtime claims it before the thread starts; without one, the teammate can wait in IDLE for later work. A teammate without an assignment cannot use file or Shell tools. It follows WORK → result → IDLE without a fixed tool-round cap; model or dispatch failures emit an error, and thread cleanup releases an unfinished assignment back to the task board. It drains its inbox before every model call, so direct messages and shutdown requests cannot wait behind an unbroken tool-use sequence. While idle it waits for MessageBus delivery first, then scans ready tasks only after the wait times out and atomically claims at most one.

After spawning a teammate, Lead ends the current turn instead of repeatedly querying its status inside the model loop. A team event in Lead's mailbox makes the runtime start the next turn.

One-shot subagents solve context isolation. Persistent teammates solve long-running parallel collaboration.

Memory, Skills, and Prompt Copy Section

S15 reuses the s09 memory runtime directly. Before each model call, it reads the .memory/MEMORY.md catalog, selects records relevant to the current request, and passes their contents to assemble_system_prompt(context). At the end of the turn, extract_memories() keeps information that can help in later sessions; when new records are stored, consolidate_memories() runs next.

The same system prompt also includes identity, tool guidance, the workspace, the skills catalog, and connected MCP servers. Skills contribute only their catalog; load_skill(name) loads full content on demand.

Compaction and Recovery Copy Section

Before the LLM call, S15 runs the compaction pipeline:

tool_result_budget → snip_compact → micro_compact → compact_history

snip_compact archives the complete history before trimming its middle. micro_compact runs only above the context limit: it saves older consumed results before replacing them with recovery paths, keeps the latest 3 complete, and stops near 80% of the limit. If a new unseen result is itself too large, S15 keeps a preview and the full-output path before considering history summarization.

The model call is wrapped with recovery:

• 429: exponential backoff retry
• 529: exponential backoff, optionally switch to fallback model after repeated failures
• max_tokens: raise max tokens, then request continuation
• prompt too long: reactive compact and retry

Background and Cron Copy Section

When a bash call sets run_in_background=true, the main loop returns a placeholder without waiting for the command:

should_run_background → start_background_task → placeholder tool_result
background done → task_notification → next round injects messages

Only explicitly marked bash calls enter the background path. A non-zero exit or worker exception produces a failed notification. Each shell runs in its own process group, which the runtime stops when the command or Agent process ends through the normal or SIGTERM path. A process that creates another session can leave that group.

The cron scheduler runs as a daemon thread and checks once per second. A durable one-shot job is persisted as pending_delivery before entering the queue and remains there until the model call containing its prompt succeeds; a failed call restores it to the queue, and a restart queues it again. Delivery is therefore at-least-once. The CLI watches cron_queue, Lead's inbox, and terminal background work; any of them can wake one automatic agent turn.

Worktree and MCP Copy Section

The task-scoped worktree behavior inherited from s13 manages working directories:

• a pending, unowned task may remain in the main workspace or be bound by create_worktree(name, task_id) to a separate branch and directory
• creation prevalidates the task, name, path, branch, and Git registry; a failed Git command is reconciled against the registry and branch state, and any partial checkout remains unbound and preserved for manual recovery
• an idle teammate atomically claims one ready task; the assignment records both task_id and its effective cwd
• Lead can also pass a ready task_id to spawn_teammate; the thread starts only after the claim succeeds
• all teammate file tools use that cwd; only the owning teammate can complete the task, and the assignment stays selected until that model turn ends
• removal stays in the host-side remove_worktree() helper. The model cannot call it. The user or host first checks task ownership, assignment leases, background work, and Git state; destructive removal requires separate user confirmation

The worktree changes tool default directories. It separates working copies; it is not a sandbox, and process-group cleanup does not contain a process that starts another session. This is why deletion remains host-owned.

Claiming or releasing a Task changes the assignment version and invalidates an old plan approval. An ordinary send_message only delivers text; it changes neither the Task identity nor the plan state.

MCP owns external capability:

• connect_mcp(name) connects a mock server
• assemble_tool_pool() assembles MCP tools and rejects normalized name collisions
• tool names use mcp__server__tool

Changes from s14 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python s15_integrated_harness/code.py

Try:

1. Inspect this repository and tell me which Python files matter most.
2. Search the connected documentation for agent loop guidance.
3. Refactor the authentication module and login page in parallel in separate worktrees. Show me each plan before editing.
4. Remind me about the meeting in 3 minutes.
5. Install the dependencies in the background while you read README.md.

Watch for:

• whether each tool call passes through hooks/permission
• whether MCP tools appear on the next round after connect_mcp
• whether a bash call with run_in_background=true returns a background placeholder
• whether cron automatically reminds you when the time arrives
• whether teammates submit plans and pause before approval
• whether an idle teammate atomically claims only one ready task
• whether every teammate file tool switches to the claimed task's cwd
• whether completion keeps the task cwd through the rest of the turn and releases it at IDLE

Next Copy Chapter

s16 Workflow Runtime adds a Workflow tool to this host. A workflow keeps a fixed orchestration path in code and records progress so the same run can resume.

Overview (`s16_workflow_runtime/README.md`) Copy Section

s16: Workflow Runtime — The Model Decides Each Step; a Script Decides the Orchestration

English · 中文 · 日本語

s01 → ... → s14 → s15 → s16 → s17

"One tool_use runs an entire orchestration" — The Workflow tool starts a recoverable script runtime that coordinates many agent calls.

Harness layer: Orchestration — run saved multi-agent scripts above the single-agent loop.

From s01 through s15, the model decides which tools to call in each round. Their results enter messages[], and the model decides the next step from the updated context. This works well when the path depends on what the previous step discovers.

Some tasks repeat a fixed sequence. A code review may inspect several dimensions concurrently, verify each finding, combine duplicates, and sort the result. The sequence and dependencies are known before execution. Here the host needs three things:

• Parallelism, rather than waiting for one item at a time;
• A stable result structure, even when individual agent answers vary;
• Recoverability, so an interruption does not rerun work that is already complete.

If this orchestration exists only in conversation history, its ordering and checkpoints also exist only in that history. A saved workflow puts the fixed sequence in code and records completed calls in a journal.

Put the Plan in Code, Not in a Sequence of Chat Turns Copy Chapter

Add a Workflow tool to the harness tool pool. The host registers trusted scripts built from agent(), parallel(), pipeline(), and phase(). The model supplies only a saved workflow name, arguments, and an optional run ID to resume; it does not send executable code or metadata.

The workflow enters the main loop as one tool_use. As the script runs, the runtime emits lifecycle and progress events and records every step in a journal on disk. When the script finishes, the call returns the launch envelope, result, and task state. Intermediate script results live in variables instead of taking space in conversation history. When restarted with resume_from_run_id, unchanged agent() calls hit the journal cache and reuse previous results.

SAMPLE_META = {"name": "review-changes", "description": "Review code changes", "phases": ["Review", "Verify"]}

async def sample_workflow(ctx, args):
    ctx.phase("Review")
    results = await ctx.pipeline(DIMENSIONS, audit, verify)   # Each dimension independently runs audit → verify
    confirmed = [f for r in results if r for f in r["confirmed"]]
    ctx.log(f"Confirmed {len(confirmed)} real issues")
    return {"confirmed": confirmed}

The Workflow Tool: One Call, One Complete Run Copy Chapter

Workflow is added to the s15 host's existing tool pool. The user can request a saved workflow, or the model can select it when a task matches a known orchestration. The adapter resolves the name through the host-owned WORKFLOWS registry, then passes its trusted metadata and function to the runtime. The other s15 tools remain available in the same loop.

The model-facing schema accepts name, args, and resume_from_run_id. Unknown names and malformed arguments become an error tool result instead of ending the host loop. The runtime then validates the registered metadata, checks permissions, registers a local workflow task, and emits async_launched before running the script. Progress events follow, then the final task_notification; the call returns JSON-safe launch information, result, and task state.

WORKFLOW_TOOL = {
    "name": "Workflow",
    "input_schema": {
        "type": "object",
        "properties": {
            "name": {"type": "string"},
            "args": {"type": "object"},
            "resume_from_run_id": {"type": "string"},
        },
        "required": ["name"],
        "additionalProperties": False,
    },
}

async def run_workflow(name, args=None, resume_from_run_id=None):
    meta, script_fn = WORKFLOWS[name]
    out = await WorkflowTool().call(
        meta, script_fn,
        args=args,
        resume_from_run_id=resume_from_run_id,
    )
    return {"launched": out["launched"], "result": out["result"],
            "task": serialize_task(out["task"])}

Workflow Metadata: Validate Before Launch Copy Chapter

Each saved workflow registers trusted metadata with name, description, and optional phases. The runtime validates it before executing workflow code. name and description identify the task in the UI, while phases names groups in the progress display. These fields belong to the host registry, not to model input.

Invalid registration raises WorkflowInputError before launch. This is the same idea as validating cron expressions in s12: do not wait until execution to discover a bad saved workflow.

Because the runtime uses meta.name in local artifact filenames, it also requires a 1-64 character safe slug containing letters, numbers, ., _, or -.

def validate_meta(meta):
    if not isinstance(meta, dict):
        raise WorkflowInputError("meta must be an object literal")
    if not meta.get("name") or not meta.get("description"):
        raise WorkflowInputError("meta requires name and description")
    if not isinstance(meta["name"], str) or not WORKFLOW_NAME_RE.fullmatch(meta["name"]):
        raise WorkflowInputError("meta.name must be a safe 1-64 character slug")
    if "phases" in meta and (
        not isinstance(meta["phases"], list)
        or not all(isinstance(p, str) and p for p in meta["phases"])
    ):
        raise WorkflowInputError("meta.phases must contain non-empty strings")
    return meta

Orchestration Primitives Copy Chapter

A script receives an ExecutionState exposing a small set of orchestration primitives. It does not read files or run shell commands directly. The default interactive mode connects agent() to the same real API client as the host, and each workflow agent reads only the content supplied through workflow arguments. demo and unit tests use MockAgentRunner so events and journal replay are repeatable.

Use pipeline when each item independently crosses the same stages. Item A may reach stage three while item B is still in stage one. Use parallel when the next step needs every result from the preceding group.

async def pipeline(self, items, *stages):
    async def run_item(item, idx):
        value = item
        for stage in stages:                       # Each item independently completes every stage
            value = await stage(value, item, idx)
        return value
    return await asyncio.gather(*[run_item(it, i) for i, it in enumerate(items)])

Structured Output: Do Not Let Subagents Return Essays Copy Chapter

agent({schema}) asks a workflow agent to return only a JSON object matching the schema. The runtime parses and validates the result, then retries once if it does not match. Downstream code receives an object instead of extracting fields from prose.

s05 warned that tool arguments cannot be trusted completely. This is the same lesson in reverse: subagent output cannot be trusted completely either. Validate at the orchestration boundary, give one retry, and keep uncertainty out of the rest of the flow.

run = await asyncio.to_thread(self.runner.run, prompt, schema, label)
result = run.value
if schema is not None:
    ok, err = SimpleJsonSchema(schema).validate(result)
    if not ok:                                       # Retry once with a reminder, then fail
        retry = await asyncio.to_thread(
            self.runner.run, prompt + "\n\nReturn valid JSON.", schema, label
        )
        result = retry.value
        ok, err = SimpleJsonSchema(schema).validate(result)
        if not ok:
            raise WorkflowInputError(f"agent({{schema}}) returned invalid output: {err}")

Task State and Progress Events Copy Chapter

LocalWorkflowTask maintains status and token usage and emits an SDK-style event stream: task_started → a sequence of task_progress events containing phase changes, subagent starts, and log batches → one final task_notification reporting completion or failure, plus the output file and agent and token counts.

The demo prints these events in order and returns the task state after the final notification.

class LocalWorkflowTask:
    def progress_event(self, ptype, **data):         # Phase/subagent/log
        self.progress.append({"type": ptype, **data})
        print(f"  progress   {ptype} ...")

Storage: Snapshot + Journal for Resuming after Interruptions Copy Chapter

The runtime stores each run under s16_workflow_runtime/.runtime/: a <runId>.json snapshot, <runId>.output.json output, <runId>.journal.jsonl journal, and <runId>.lock coordination file. Every fresh run reserves a new runId with exclusive file creation before opening its journal. The run lock stays held through execution and final persistence, so another process cannot resume the same run at the same time. Its snapshot records the workflow name, arguments, and task state; resume validates the saved snapshot and journal before changing either successful artifact.

The journal is the core of checkpointed resume. It records every agent() result one line at a time:

class WorkflowJournal:
    def record(self, key, value):
        self._f.write(json.dumps({"key": key, "value": value}) + "\n")
        self._f.flush()
        self.cache[key] = value

Resume: Continue by runId and Reuse Everything Unchanged Copy Chapter

Calling the workflow again with resume_from_run_id reruns the script, but every agent() computes a deterministic semantic key. If that key is present in the journal, it returns the cached result without executing again. Every unchanged call hits the cache; only a changed call and the downstream steps that depend on it actually rerun.

The key detail is that keys cannot depend on concurrency order. Agents in parallel and pipeline finish in nondeterministic order. If "the nth completion" became the key, cache entries would map to the wrong calls on the next run. A key therefore uses a stable hash of call content, including type, label, prompt, and schema, rather than a shared counter:

def key(self, kind, label, prompt, schema):
    basis = f"{kind}|{label}|{prompt}|{json.dumps(schema, sort_keys=True)}"
    return f"{kind}-{_stable_hash(basis) % 10**10:010d}"

# Inside agent():
cached = self.journal.cached(key)
if cached is not MISS:
    self.task.progress_event("workflow_agent", label=label, status="cached")
    return cached

Stable Call Keys Copy Chapter

On resume, the runtime must match each current agent() call with its earlier journal record. A stable hash gives unchanged workflow code and arguments the same call key. Real model output may vary; when the call content has not changed, resume uses the result already saved in the journal.

See It Run Copy Chapter

The sample review-changes workflow uses pipeline to send each review dimension independently through audit → verify. Interactive mode uses the real API and reads the material to review from args.changes. demo uses fixed runner data to show pipeline, validation, journal, and resume behavior.

async def sample_workflow(ctx, args):
    ctx.phase("Review")
    changes = args.get("changes", "")

    async def audit(_v, dimension, _i):
        out = await ctx.agent(f"Inspect this change for {dimension} issues:\n{changes}",
                              schema=FINDINGS_SCHEMA, label=f"audit:{dimension}", phase="Review")
        return {"dimension": dimension, "findings": out["findings"]}

    async def verify(audited, dimension, _i):
        ctx.phase("Verify")
        verdicts = await ctx.parallel([                       # Verify every finding independently
            (lambda f=f: ctx.agent(f"Verify this finding against the change:\n{changes}\n\n{f}",
                                   schema=VERDICT_SCHEMA, label=f"verify:{dimension}:{f['title']}"))
            for f in audited["findings"]])
        return {"dimension": dimension,
                "confirmed": [f for f, v in zip(audited["findings"], verdicts) if v and v["isReal"]]}

    results = await ctx.pipeline(DIMENSIONS, audit, verify)
    ...

Changes from s15 Copy Chapter

s16 does not replace the main loop. It exposes Workflow at the tool layer and starts a local workflow runtime behind it: one saved script coordinates N calls through an agent-runner boundary. An s06 subagent is dispatched once at the model's discretion; s16 turns the orchestration into resumable host code.

Try It Copy Chapter

python s16_workflow_runtime/code.py          # Both the main model and Workflow agents use the real API
python s16_workflow_runtime/code.py demo     # Deterministic review-changes fixture and event stream
python s16_workflow_runtime/code.py resume   # Resume by the last runId; every agent() hits the journal cache

In the default command, ask the model to read the changes, place that text in args.changes, and run the saved review-changes workflow. Both the main model and workflow agents use the real API. The demo command uses fixed runner data so lifecycle and resume behavior can be observed repeatedly. A resumed demo reports agents=0 tokens=0 when every call hits the cache.

Next Copy Chapter

s17 Goal Loop uses a smaller, independent loop to check whether a stated goal has been reached and decide whether another turn is needed.

Overview (`s17_goal_loop/README.md`) Copy Section

s17: Goal Loop: The Model Proposes a Stop; an Independent Evaluator Decides Whether to Continue

English · 中文 · 日本語

s01 → ... → s15 → s16 → s17

"The model making no more tool calls means that one turn wants to stop. A separate evaluator decides whether the whole goal is complete."

Harness layer: continued execution. Check a completion condition at the end of every turn, and start another turn when work remains.

Since s01, the agent loop has had one simple exit condition: when the model stops calling tools, the program returns.

That is enough for ordinary conversations, but not always for tasks such as "keep fixing until every test passes" or "finish every acceptance criterion." The model may believe the work is done after only part of it. No new tool_use means only that the current turn ended; it does not prove that the whole goal was achieved.

/goal adds one independent decision before the real return.

/goal is a session-scoped Stop hook Copy Chapter

Enter:

/goal pytest tests/auth exits with code 0 and lint reports no errors

The program stores the completion condition and immediately gives it to the main model as the current task. You do not need to send a second "start working" prompt.

When the main model stops calling tools, the loop runs the Goal Stop hook before returning:

if tool_results:
    messages.append({"role": "user", "content": tool_results})
    continue

decision = await self.goal.evaluate_after_turn(self.messages)
if decision.action == "block":
    self.messages.append({
        "role": "user",
        "content": decision.reason,
    })
    continue

return SessionResult(text=text, status=decision.action)

With no active goal, the hook allows the stop immediately, so the return condition is the same as in s01.

The evaluator is separate from the worker Copy Chapter

The main model edits code, runs commands, and solves the task. The Goal evaluator is a separate model call with one job: judge the completion condition.

GoalController owns the evaluator as an internal dependency of the Goal gate. It is not a second return path beside the main loop.

This lesson has no separate CommandQueue: when evaluation blocks the stop, the controller appends the reason to the same messages[] and starts the next turn. A larger host may use a shared queue to carry user input, background results, and continuation commands back into the session, but that queue is transport for the whole host, not a component owned by the Goal gate. Putting it inside the gate would blur the decision with the path used to deliver that decision.

The evaluator sees:

• the active Goal condition;
• the conversation so far;
• tool results that the worker placed in that conversation.

It has no tools. It cannot read a file or rerun a test on its own. It can only judge what is already present in the conversation:

{
  "ok": false,
  "reason": "The conversation does not contain pytest's exit code yet.",
  "impossible": false
}

ok=true means the condition is satisfied. ok=false means another turn is needed. If the task can no longer be completed, the evaluator can return impossible=true.

The conversation is the evaluator's input Copy Chapter

The evaluator reads the current conversation. Tool results, worker explanations, and background-task notifications all enter it as messages, and the decision depends on what those messages actually say.

The evaluator input keeps the most recent complete messages. If the newest message alone is too large, it keeps that message's beginning and end so one tool result cannot fill the whole evaluator request.

That does not mean a bare "tests passed" claim must be accepted. The evaluator prompt explicitly requires concrete results from the conversation and tells the model not to assume an unreported command succeeded.

It is still a model reading text, so reliability depends on whether important results were surfaced clearly. The worker's system prompt therefore says:

After running a verification command, report the command and its result clearly enough for an independent evaluator to inspect.

Goal Loop is not a test framework. Tools still perform the real verification. The Goal evaluator only decides whether those verification results are present in the current work record.

A good completion condition is checkable Copy Chapter

"Make the code good" is too vague. The evaluator cannot know what "good" means.

A useful condition states three things:

1. End state: what must be true when work is done;
2. Check: which command or output proves it;
3. Constraints: what must not be broken along the way.

For example:

/goal finish the authentication migration until pytest tests/auth exits 0,
without modifying test files outside tests/auth

If you need to bound unattended work, use the main loop's global turn limit instead of hiding a fixed budget inside Goal:

MAX_TURNS=20 python s17_goal_loop/code.py \
  "/goal fix the type errors until npm run typecheck exits 0"

Unfinished work returns to the same loop Copy Chapter

When the evaluator says the condition is not met, it returns a short reason:

The conversation has no complete test result. Run pytest tests/auth and report its exit code.

The program appends that reason to messages[] and executes continue in the current while loop. The main model starts another turn without waiting for the user to type "continue."

There is no separate continuation queue. Goal evaluation happens at the loop's return boundary, and unfinished work returns through that same boundary.

Wait before judging unfinished background work Copy Chapter

A Workflow, background command, or other asynchronous task may still be running when the main model ends its current turn.

Evaluating immediately would be premature because the important result has not returned to the conversation. The Goal Stop hook returns defer, keeps the Goal active, and skips the evaluator. When the task finishes, the host passes its completion message to submit_background_result(); that message enters the same messages[], and the loop resumes.

A Workflow notification has no mechanical privilege. It enters the conversation like other messages, and the evaluator judges the actual result it contains.

Automatic continuation still needs an exit Copy Chapter

Goal has no hidden default budget of twenty turns. The evaluator judges the condition again after each completed turn.

No automatic mechanism should monopolize one request forever, however. This lesson keeps two general exits outside the goal itself:

• the main loop's global max_turns;
• a cap on consecutive Stop-hook blocks.

When a limit is reached, the program returns control to the user. It does not mark the goal complete and does not silently clear it. The user can inspect status, provide more information, continue, or clear the goal.

An evaluator error follows the same rule: stop automatic continuation, leave the goal active, and surface the error instead of claiming success when completion could not be judged.

Inspect, replace, and clear Copy Chapter

One session has at most one active Goal.

/goal

Shows the condition, elapsed time, evaluation count, main Agent token spend, and the latest evaluator reason.

/goal a new completion condition

Replaces the previous Goal and begins work under the new condition immediately.

/goal clear

Clears the active Goal. stop, off, reset, none, and cancel are accepted aliases.

GoalController.restore() can restore a still-active Goal from goal_status events persisted by the host; this lesson's CLI does not persist a whole session. A completed, failed, or cleared Goal does not restart. The condition carries over, while turn count, elapsed time, and token baseline start fresh.

What the code adds Copy Chapter

This is an independent mechanism example built on the S04 kernel. It keeps the five base tools and the four hook points, then adds four Goal-specific pieces:

The integration point is only a few lines:

decision = await self.goal.evaluate_after_turn(self.messages)
if decision.action == "block":
    continue
return SessionResult(text=text, status=decision.action)

Try it Copy Chapter

Install dependencies and prepare .env:

pip install -r requirements.txt

# .env
ANTHROPIC_API_KEY=...
MODEL_ID=...

# Optional: use a smaller model for Goal evaluation
GOAL_EVALUATOR_MODEL_ID=...

Start the interactive session:

python s17_goal_loop/code.py

Then enter:

/goal python -m pytest exits with code 0

You can also set a Goal directly from the command line:

python s17_goal_loop/code.py "/goal python -m pytest exits with code 0"

Relationship to s16 Copy Chapter

s16 answers how a batch of work should run: which steps are concurrent, how results are verified, and how an interrupted run resumes.

s17 answers whether the entire task is complete. A Workflow may finish successfully while the user's final requirements are still unmet. Once the Workflow result enters the conversation, the Goal evaluator decides whether the session should stop or continue.

You can use either mechanism on its own. When one host connects them, the Workflow completion message enters the conversation and Goal Loop decides whether the overall task needs another turn.

Overview (`web/README.md`) Copy Section

This is a Next.js project bootstrapped with create-next-app.

Getting Started Copy Chapter

First, run the development server:

npm run dev
# or
yarn dev
# or
pnpm dev
# or
bun dev

Open http://localhost:3000 with your browser to see the result.

You can start editing the page by modifying app/page.tsx. The page auto-updates as you edit the file.

This project uses next/font to automatically optimize and load Geist, a new font family for Vercel.

Learn More Copy Chapter

To learn more about Next.js, take a look at the following resources:

• Next.js Documentation - learn about Next.js features and API.
• Learn Next.js - an interactive Next.js tutorial.

You can check out the Next.js GitHub repository - your feedback and contributions are welcome!

Deploy on Vercel Copy Chapter

The easiest way to deploy your Next.js app is to use the Vercel Platform from the creators of Next.js.

Check out our Next.js deployment documentation for more details.

En S01 The Agent Loop (`docs/en/s01-the-agent-loop.md`) Copy Section

s01: The Agent Loop

[ s01 ] > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"One loop & Bash is all you need" -- one tool + one loop = an agent.

Harness layer: The loop -- the model's first connection to the real world.

Problem Copy Chapter

A language model can reason about code, but it can't touch the real world -- can't read files, run tests, or check errors. Without a loop, every tool call requires you to manually copy-paste results back. You become the loop.

Solution Copy Chapter

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> |  Tool   |
| prompt |      |       |      | execute |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                    (loop until stop_reason != "tool_use")

One exit condition controls the entire flow. The loop runs until the model stops calling tools.

How It Works Copy Chapter

1. User prompt becomes the first message.

messages.append({"role": "user", "content": query})

2. Send messages + tool definitions to the LLM.

response = client.messages.create(
    model=MODEL, system=SYSTEM, messages=messages,
    tools=TOOLS, max_tokens=8000,
)

3. Append the assistant response. Check stop_reason -- if the model didn't call a tool, we're done.

messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
    return

4. Execute each tool call, collect results, append as a user message. Loop back to step 2.

results = []
for block in response.content:
    if block.type == "tool_use":
        output = run_bash(block.input["command"])
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })
messages.append({"role": "user", "content": results})

Assembled into one function:

def agent_loop(query):
    messages = [{"role": "user", "content": query}]
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return

        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = run_bash(block.input["command"])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

That's the entire agent in under 30 lines. Everything else in this course layers on top -- without changing the loop.

What Changed Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s01_agent_loop.py

1. Create a file called hello.py that prints "Hello, World!"
2. List all Python files in this directory
3. What is the current git branch?
4. Create a directory called test_output and write 3 files in it

En S02 Tool Use (`docs/en/s02-tool-use.md`) Copy Section

s02: Tool Use

s01 > [ s02 ] > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"Adding a tool means adding one handler" -- the loop stays the same; new tools register into the dispatch map.

Harness layer: Tool dispatch -- expanding what the model can reach.

Problem Copy Chapter

With only bash, the agent shells out for everything. cat truncates unpredictably, sed fails on special characters, and every bash call is an unconstrained security surface. Dedicated tools like read_file and write_file let you enforce path sandboxing at the tool level.

The key insight: adding tools does not require changing the loop.

Solution Copy Chapter

+--------+      +-------+      +------------------+
|  User  | ---> |  LLM  | ---> | Tool Dispatch    |
| prompt |      |       |      | {                |
+--------+      +---+---+      |   bash: run_bash |
                    ^           |   read: run_read |
                    |           |   write: run_wr  |
                    +-----------+   edit: run_edit |
                    tool_result | }                |
                                +------------------+

The dispatch map is a dict: {tool_name: handler_function}.
One lookup replaces any if/elif chain.

How It Works Copy Chapter

1. Each tool gets a handler function. Path sandboxing prevents workspace escape.

def safe_path(p: str) -> Path:
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

def run_read(path: str, limit: int = None) -> str:
    text = safe_path(path).read_text(encoding="utf-8")
    lines = text.splitlines()
    if limit and limit < len(lines):
        lines = lines[:limit]
    return "\n".join(lines)[:50000]

2. The dispatch map links tool names to handlers.

TOOL_HANDLERS = {
    "bash":       lambda **kw: run_bash(kw["command"]),
    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),
    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"],
                                        kw["new_text"]),
}

3. In the loop, look up the handler by name. The loop body itself is unchanged from s01.

for block in response.content:
    if block.type == "tool_use":
        handler = TOOL_HANDLERS.get(block.name)
        output = handler(**block.input) if handler \
            else f"Unknown tool: {block.name}"
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })

Add a tool = add a handler + add a schema entry. The loop never changes.

What Changed From s01 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s02_tool_use.py

1. Read the file requirements.txt
2. Create a file called greet.py with a greet(name) function
3. Edit greet.py to add a docstring to the function
4. Read greet.py to verify the edit worked

En S03 Todo Write (`docs/en/s03-todo-write.md`) Copy Section

s03: TodoWrite

s01 > s02 > [ s03 ] > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"An agent without a plan drifts" -- list the steps first, then execute.

Harness layer: Planning -- keeping the model on course without scripting the route.

Problem Copy Chapter

On multi-step tasks, the model loses track. It repeats work, skips steps, or wanders off. Long conversations make this worse -- the system prompt fades as tool results fill the context. A 10-step refactoring might complete steps 1-3, then the model starts improvising because it forgot steps 4-10.

Solution Copy Chapter

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> | Tools   |
| prompt |      |       |      | + todo  |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                          |
              +-----------+-----------+
              | TodoManager state     |
              | [ ] task A            |
              | [>] task B  <- doing  |
              | [x] task C            |
              +-----------------------+
                          |
              if rounds_since_todo >= 3:
                inject <reminder> into tool_result

How It Works Copy Chapter

1. TodoManager stores items with statuses. Only one item can be in_progress at a time.

class TodoManager:
    def update(self, items: list) -> str:
        validated, in_progress_count = [], 0
        for item in items:
            status = item.get("status", "pending")
            if status == "in_progress":
                in_progress_count += 1
            validated.append({"id": item["id"], "text": item["text"],
                              "status": status})
        if in_progress_count > 1:
            raise ValueError("Only one task can be in_progress")
        self.items = validated
        return self.render()

2. The todo tool goes into the dispatch map like any other tool.

TOOL_HANDLERS = {
    # ...base tools...
    "todo": lambda **kw: TODO.update(kw["items"]),
}

3. A nag reminder injects a nudge if the model goes 3+ rounds without calling todo.

if rounds_since_todo >= 3 and messages:
    last = messages[-1]
    if last["role"] == "user" and isinstance(last.get("content"), list):
        last["content"].insert(0, {
            "type": "text",
            "text": "<reminder>Update your todos.</reminder>",
        })

The "one in_progress at a time" constraint forces sequential focus. The nag reminder creates accountability.

What Changed From s02 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s03_todo_write.py

1. Refactor the file hello.py: add type hints, docstrings, and a main guard
2. Create a Python package with __init__.py, utils.py, and tests/test_utils.py
3. Review all Python files and fix any style issues

En S04 Subagent (`docs/en/s04-subagent.md`) Copy Section

s04: Subagents

s01 > s02 > s03 > [ s04 ] > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"Break big tasks down; each subtask gets a clean context" -- subagents use independent messages[], keeping the main conversation clean.

Harness layer: Context isolation -- protecting the model's clarity of thought.

Problem Copy Chapter

As the agent works, its messages array grows. Every file read, every bash output stays in context permanently. "What testing framework does this project use?" might require reading 5 files, but the parent only needs the answer: "pytest."

Solution Copy Chapter

Parent agent                     Subagent
+------------------+             +------------------+
| messages=[...]   |             | messages=[]      | <-- fresh
|                  |  dispatch   |                  |
| tool: task       | ----------> | while tool_use:  |
|   prompt="..."   |             |   call tools     |
|                  |  summary    |   append results |
|   result = "..." | <---------- | return last text |
+------------------+             +------------------+

Parent context stays clean. Subagent context is discarded.

How It Works Copy Chapter

1. The parent gets a task tool. The child gets all base tools except task (no recursive spawning).

PARENT_TOOLS = CHILD_TOOLS + [
    {"name": "task",
     "description": "Spawn a subagent with fresh context.",
     "input_schema": {
         "type": "object",
         "properties": {"prompt": {"type": "string"}},
         "required": ["prompt"],
     }},
]

2. The subagent starts with messages=[] and runs its own loop. Only the final text returns to the parent.

def run_subagent(prompt: str) -> str:
    sub_messages = [{"role": "user", "content": prompt}]
    for _ in range(30):  # safety limit
        response = client.messages.create(
            model=MODEL, system=SUBAGENT_SYSTEM,
            messages=sub_messages,
            tools=CHILD_TOOLS, max_tokens=8000,
        )
        sub_messages.append({"role": "assistant",
                             "content": response.content})
        if response.stop_reason != "tool_use":
            break
        results = []
        for block in response.content:
            if block.type == "tool_use":
                handler = TOOL_HANDLERS.get(block.name)
                output = handler(**block.input)
                results.append({"type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output)[:50000]})
        sub_messages.append({"role": "user", "content": results})
    return "".join(
        b.text for b in response.content if hasattr(b, "text")
    ) or "(no summary)"

The child's entire message history (possibly 30+ tool calls) is discarded. The parent receives a one-paragraph summary as a normal tool_result.

What Changed From s03 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s04_subagent.py

1. Use a subtask to find what testing framework this project uses
2. Delegate: read all .py files and summarize what each one does
3. Use a task to create a new module, then verify it from here

En S05 Skill Loading (`docs/en/s05-skill-loading.md`) Copy Section

s05: Skills

s01 > s02 > s03 > s04 > [ s05 ] > s06 | s07 > s08 > s09 > s10 > s11 > s12

"Load knowledge when you need it, not upfront" -- inject via tool_result, not the system prompt.

Harness layer: On-demand knowledge -- domain expertise, loaded when the model asks.

Problem Copy Chapter

You want the agent to follow domain-specific workflows: git conventions, testing patterns, code review checklists. Putting everything in the system prompt wastes tokens on unused skills. 10 skills at 2000 tokens each = 20,000 tokens, most of which are irrelevant to any given task.

Solution Copy Chapter

System prompt (Layer 1 -- always present):
+--------------------------------------+
| You are a coding agent.              |
| Skills available:                    |
|   - git: Git workflow helpers        |  ~100 tokens/skill
|   - test: Testing best practices     |
+--------------------------------------+

When model calls load_skill("git"):
+--------------------------------------+
| tool_result (Layer 2 -- on demand):  |
| <skill name="git">                   |
|   Full git workflow instructions...  |  ~2000 tokens
|   Step 1: ...                        |
| </skill>                             |
+--------------------------------------+

Layer 1: skill names in system prompt (cheap). Layer 2: full body via tool_result (on demand).

How It Works Copy Chapter

1. Each skill is a directory containing a SKILL.md with YAML frontmatter.

skills/
  pdf/
    SKILL.md       # ---\n name: pdf\n description: Process PDF files\n ---\n ...
  code-review/
    SKILL.md       # ---\n name: code-review\n description: Review code\n ---\n ...

2. SkillLoader scans for SKILL.md files, uses the directory name as the skill identifier.

class SkillLoader:
    def __init__(self, skills_dir: Path):
        self.skills = {}
        for f in sorted(skills_dir.rglob("SKILL.md")):
            text = f.read_text(encoding="utf-8")
            meta, body = self._parse_frontmatter(text)
            name = meta.get("name", f.parent.name)
            self.skills[name] = {"meta": meta, "body": body}

    def get_descriptions(self) -> str:
        lines = []
        for name, skill in self.skills.items():
            desc = skill["meta"].get("description", "")
            lines.append(f"  - {name}: {desc}")
        return "\n".join(lines)

    def get_content(self, name: str) -> str:
        skill = self.skills.get(name)
        if not skill:
            return f"Error: Unknown skill '{name}'."
        return f"<skill name=\"{name}\">\n{skill['body']}\n</skill>"

3. Layer 1 goes into the system prompt. Layer 2 is just another tool handler.

SYSTEM = f"""You are a coding agent at {WORKDIR}.
Skills available:
{SKILL_LOADER.get_descriptions()}"""

TOOL_HANDLERS = {
    # ...base tools...
    "load_skill": lambda **kw: SKILL_LOADER.get_content(kw["name"]),
}

The model learns what skills exist (cheap) and loads them when relevant (expensive).

What Changed From s04 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s05_skill_loading.py

1. What skills are available?
2. Load the agent-builder skill and follow its instructions
3. I need to do a code review -- load the relevant skill first
4. Build an MCP server using the mcp-builder skill

En S06 Context Compact (`docs/en/s06-context-compact.md`) Copy Section

s06: Context Compact

s01 > s02 > s03 > s04 > s05 > [ s06 ] | s07 > s08 > s09 > s10 > s11 > s12

"Context will fill up; you need a way to make room" -- three-layer compression strategy for infinite sessions.

Harness layer: Compression -- clean memory for infinite sessions.

Problem Copy Chapter

The context window is finite. A single read_file on a 1000-line file costs ~4000 tokens. After reading 30 files and running 20 bash commands, you hit 100,000+ tokens. The agent cannot work on large codebases without compression.

Solution Copy Chapter

Three layers, increasing in aggressiveness:

Every turn:
+------------------+
| Tool call result |
+------------------+
        |
        v
[Layer 1: micro_compact]        (silent, every turn)
  Replace tool_result > 3 turns old
  with "[Previous: used {tool_name}]"
        |
        v
[Check: tokens > 50000?]
   |               |
   no              yes
   |               |
   v               v
continue    [Layer 2: auto_compact]
              Save transcript to .transcripts/
              LLM summarizes conversation.
              Replace all messages with [summary].
                    |
                    v
            [Layer 3: compact tool]
              Model calls compact explicitly.
              Same summarization as auto_compact.

How It Works Copy Chapter

1. Layer 1 -- micro_compact: Before each LLM call, replace old tool results with placeholders.

def micro_compact(messages: list) -> list:
    tool_results = []
    for i, msg in enumerate(messages):
        if msg["role"] == "user" and isinstance(msg.get("content"), list):
            for j, part in enumerate(msg["content"]):
                if isinstance(part, dict) and part.get("type") == "tool_result":
                    tool_results.append((i, j, part))
    if len(tool_results) <= KEEP_RECENT:
        return messages
    for _, _, part in tool_results[:-KEEP_RECENT]:
        if len(part.get("content", "")) > 100:
            part["content"] = f"[Previous: used {tool_name}]"
    return messages

2. Layer 2 -- auto_compact: When tokens exceed threshold, save full transcript to disk, then ask the LLM to summarize.

def auto_compact(messages: list) -> list:
    # Save transcript for recovery
    transcript_path = TRANSCRIPT_DIR / f"transcript_{int(time.time())}.jsonl"
    with open(transcript_path, "w") as f:
        for msg in messages:
            f.write(json.dumps(msg, default=str) + "\n")
    # LLM summarizes
    response = client.messages.create(
        model=MODEL,
        messages=[{"role": "user", "content":
            "Summarize this conversation for continuity..."
            + json.dumps(messages, default=str)[:80000]}],
        max_tokens=2000,
    )
    return [
        {"role": "user", "content": f"[Compressed]\n\n{response.content[0].text}"},
    ]

3. Layer 3 -- manual compact: The compact tool triggers the same summarization on demand.

4. The loop integrates all three:

def agent_loop(messages: list):
    while True:
        micro_compact(messages)                        # Layer 1
        if estimate_tokens(messages) > THRESHOLD:
            messages[:] = auto_compact(messages)       # Layer 2
        response = client.messages.create(...)
        # ... tool execution ...
        if manual_compact:
            messages[:] = auto_compact(messages)       # Layer 3

Transcripts preserve full history on disk. Nothing is truly lost -- just moved out of active context.

What Changed From s05 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s06_context_compact.py

1. Read every Python file in the agents/ directory one by one (watch micro-compact replace old results)
2. Keep reading files until compression triggers automatically
3. Use the compact tool to manually compress the conversation

En S07 Task System (`docs/en/s07-task-system.md`) Copy Section

s07: Task System

s01 > s02 > s03 > s04 > s05 > s06 | [ s07 ] > s08 > s09 > s10 > s11 > s12

"Break big goals into small tasks, order them, persist to disk" -- a file-based task graph with dependencies, laying the foundation for multi-agent collaboration.

Harness layer: Persistent tasks -- goals that outlive any single conversation.

Problem Copy Chapter

s03's TodoManager is a flat checklist in memory: no ordering, no dependencies, no status beyond done-or-not. Real goals have structure -- task B depends on task A, tasks C and D can run in parallel, task E waits for both C and D.

Without explicit relationships, the agent can't tell what's ready, what's blocked, or what can run concurrently. And because the list lives only in memory, context compression (s06) wipes it clean.

Solution Copy Chapter

Promote the checklist into a task graph persisted to disk. Each task is a JSON file with status, dependencies (blockedBy). The graph answers three questions at any moment:

• What's ready? -- tasks with pending status and empty blockedBy.
• What's blocked? -- tasks waiting on unfinished dependencies.
• What's done? -- completed tasks, whose completion automatically unblocks dependents.

.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):
                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

Ordering:     task 1 must finish before 2 and 3
Parallelism:  tasks 2 and 3 can run at the same time
Dependencies: task 4 waits for both 2 and 3
Status:       pending -> in_progress -> completed

This task graph becomes the coordination backbone for everything after s07: background execution (s08), multi-agent teams (s09+), and worktree isolation (s12) all read from and write to this same structure.

How It Works Copy Chapter

1. TaskManager: one JSON file per task, CRUD with dependency graph.

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": [],
                "owner": ""}
        self._save(task)
        self._next_id += 1
        return json.dumps(task, indent=2)

2. Dependency resolution: completing a task clears its ID from every other task's blockedBy list, automatically unblocking dependents.

def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text(encoding="utf-8"))
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)

3. Status + dependency wiring: update handles transitions and dependency edges.

def update(self, task_id, status=None,
           add_blocked_by=None, remove_blocked_by=None):
    task = self._load(task_id)
    if status:
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)
    if add_blocked_by:
        task["blockedBy"] = list(set(task["blockedBy"] + add_blocked_by))
    if remove_blocked_by:
        task["blockedBy"] = [x for x in task["blockedBy"] if x not in remove_blocked_by]
    self._save(task)

4. Four task tools go into the dispatch map.

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"]),
}

From s07 onward, the task graph is the default for multi-step work. s03's Todo remains for quick single-session checklists.

What Changed From s06 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s07_task_system.py

1. Create 3 tasks: "Setup project", "Write code", "Write tests". Make them depend on each other in order.
2. List all tasks and show the dependency graph
3. Complete task 1 and then list tasks to see task 2 unblocked
4. Create a task board for refactoring: parse -> transform -> emit -> test, where transform and emit can run in parallel after parse

En S08 Background Tasks (`docs/en/s08-background-tasks.md`) Copy Section

s08: Background Tasks

s01 > s02 > s03 > s04 > s05 > s06 | s07 > [ s08 ] > s09 > s10 > s11 > s12

"Run slow operations in the background; the agent keeps thinking" -- daemon threads run commands, inject notifications on completion.

Harness layer: Background execution -- the model thinks while the harness waits.

Problem Copy Chapter

Some commands take minutes: npm install, pytest, docker build. With a blocking loop, the model sits idle waiting. If the user asks "install dependencies and while that runs, create the config file," the agent does them sequentially, not in parallel.

Solution Copy Chapter

Main thread                Background thread
+-----------------+        +-----------------+
| agent loop      |        | subprocess runs |
| ...             |        | ...             |
| [LLM call] <---+------- | enqueue(result) |
|  ^drain queue   |        +-----------------+
+-----------------+

Timeline:
Agent --[spawn A]--[spawn B]--[other work]----
             |          |
             v          v
          [A runs]   [B runs]      (parallel)
             |          |
             +-- results injected before next LLM call --+

How It Works Copy Chapter

1. BackgroundManager tracks tasks with a thread-safe notification queue.

class BackgroundManager:
    def __init__(self):
        self.tasks = {}
        self._notification_queue = []
        self._lock = threading.Lock()

2. run() starts a daemon thread and returns immediately.

def run(self, command: str) -> str:
    task_id = str(uuid.uuid4())[:8]
    self.tasks[task_id] = {"status": "running", "command": command}
    thread = threading.Thread(
        target=self._execute, args=(task_id, command), daemon=True)
    thread.start()
    return f"Background task {task_id} started"

3. When the subprocess finishes, its result goes into the notification queue.

def _execute(self, task_id, command):
    try:
        r = subprocess.run(command, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=300)
        output = (r.stdout + r.stderr).strip()[:50000]
    except subprocess.TimeoutExpired:
        output = "Error: Timeout (300s)"
    with self._lock:
        self._notification_queue.append({
            "task_id": task_id, "result": output[:500]})

4. The agent loop drains notifications before each LLM call.

def agent_loop(messages: list):
    while True:
        notifs = BG.drain_notifications()
        if notifs:
            notif_text = "\n".join(
                f"[bg:{n['task_id']}] {n['result']}" for n in notifs)
            messages.append({"role": "user",
                "content": f"<background-results>\n{notif_text}\n"
                           f"</background-results>"})
        response = client.messages.create(...)

The loop stays single-threaded. Only subprocess I/O is parallelized.

What Changed From s07 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s08_background_tasks.py

1. Run "sleep 5 && echo done" in the background, then create a file while it runs
2. Start 3 background tasks: "sleep 2", "sleep 4", "sleep 6". Check their status.
3. Run pytest in the background and keep working on other things

En S09 Agent Teams (`docs/en/s09-agent-teams.md`) Copy Section

s09: Agent Teams

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > [ s09 ] > s10 > s11 > s12

"When the task is too big for one, delegate to teammates" -- persistent teammates + async mailboxes.

Harness layer: Team mailboxes -- multiple models, coordinated through files.

Problem Copy Chapter

Subagents (s04) are disposable: spawn, work, return summary, die. No identity, no memory between invocations. Background tasks (s08) run shell commands but can't make LLM-guided decisions.

Real teamwork needs: (1) persistent agents that outlive a single prompt, (2) identity and lifecycle management, (3) a communication channel between agents.

Solution Copy Chapter

Teammate lifecycle:
  spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN

Communication:
  .team/
    config.json           <- team roster + statuses
    inbox/
      alice.jsonl         <- append-only, drain-on-read
      bob.jsonl
      lead.jsonl

              +--------+    send("alice","bob","...")    +--------+
              | alice  | -----------------------------> |  bob   |
              | loop   |    bob.jsonl << {json_line}    |  loop  |
              +--------+                                +--------+
                   ^                                         |
                   |        BUS.read_inbox("alice")          |
                   +---- alice.jsonl -> read + drain ---------+

How It Works Copy Chapter

1. TeammateManager maintains config.json with the team roster.

class TeammateManager:
    def __init__(self, team_dir: Path):
        self.dir = team_dir
        self.dir.mkdir(exist_ok=True)
        self.config_path = self.dir / "config.json"
        self.config = self._load_config()
        self.threads = {}

2. spawn() creates a teammate and starts its agent loop in a thread.

def spawn(self, name: str, role: str, prompt: str) -> str:
    member = {"name": name, "role": role, "status": "working"}
    self.config["members"].append(member)
    self._save_config()
    thread = threading.Thread(
        target=self._teammate_loop,
        args=(name, role, prompt), daemon=True)
    thread.start()
    return f"Spawned teammate '{name}' (role: {role})"

3. MessageBus: append-only JSONL inboxes. send() appends a JSON line; read_inbox() reads all and drains.

class MessageBus:
    def send(self, sender, to, content, msg_type="message", extra=None):
        msg = {"type": msg_type, "from": sender,
               "content": content, "timestamp": time.time()}
        if extra:
            msg.update(extra)
        with open(self.dir / f"{to}.jsonl", "a") as f:
            f.write(json.dumps(msg) + "\n")

    def read_inbox(self, name):
        path = self.dir / f"{name}.jsonl"
        if not path.exists(): return "[]"
        msgs = [json.loads(l) for l in path.read_text(encoding="utf-8").strip().splitlines() if l]
        path.write_text("", encoding="utf-8")  # drain
        return json.dumps(msgs, indent=2)

4. Each teammate checks its inbox before every LLM call, injecting received messages into context.

def _teammate_loop(self, name, role, prompt):
    messages = [{"role": "user", "content": prompt}]
    for _ in range(50):
        inbox = BUS.read_inbox(name)
        if inbox != "[]":
            messages.append({"role": "user",
                "content": f"<inbox>{inbox}</inbox>"})
        response = client.messages.create(...)
        if response.stop_reason != "tool_use":
            break
        # execute tools, append results...
    self._find_member(name)["status"] = "idle"

What Changed From s08 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s09_agent_teams.py

1. Spawn alice (coder) and bob (tester). Have alice send bob a message.
2. Broadcast "status update: phase 1 complete" to all teammates
3. Check the lead inbox for any messages
4. Type /team to see the team roster with statuses
5. Type /inbox to manually check the lead's inbox

En S10 Team Protocols (`docs/en/s10-team-protocols.md`) Copy Section

s10: Team Protocols

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > [ s10 ] > s11 > s12

"Teammates need shared communication rules" -- one request-response pattern drives all negotiation.

Harness layer: Protocols -- structured handshakes between models.

Problem Copy Chapter

In s09, teammates work and communicate but lack structured coordination:

Shutdown: Killing a thread leaves files half-written and config.json stale. You need a handshake: the lead requests, the teammate approves (finish and exit) or rejects (keep working).

Plan approval: When the lead says "refactor the auth module," the teammate starts immediately. For high-risk changes, the lead should review the plan first.

Both share the same structure: one side sends a request with a unique ID, the other responds referencing that ID.

Solution Copy Chapter

Shutdown Protocol            Plan Approval Protocol
==================           ======================

Lead             Teammate    Teammate           Lead
  |                 |           |                 |
  |--shutdown_req-->|           |--plan_req------>|
  | {req_id:"abc"}  |           | {req_id:"xyz"}  |
  |                 |           |                 |
  |<--shutdown_resp-|           |<--plan_resp-----|
  | {req_id:"abc",  |           | {req_id:"xyz",  |
  |  approve:true}  |           |  approve:true}  |

Shared FSM:
  [pending] --approve--> [approved]
  [pending] --reject---> [rejected]

Trackers:
  shutdown_requests = {req_id: {target, status}}
  plan_requests     = {req_id: {from, plan, status}}

How It Works Copy Chapter

1. The lead initiates shutdown by generating a request_id and sending through the inbox.

shutdown_requests = {}

def handle_shutdown_request(teammate: str) -> str:
    req_id = str(uuid.uuid4())[:8]
    shutdown_requests[req_id] = {"target": teammate, "status": "pending"}
    BUS.send("lead", teammate, "Please shut down gracefully.",
             "shutdown_request", {"request_id": req_id})
    return f"Shutdown request {req_id} sent (status: pending)"

2. The teammate receives the request and responds with approve/reject.

if tool_name == "shutdown_response":
    req_id = args["request_id"]
    approve = args["approve"]
    shutdown_requests[req_id]["status"] = "approved" if approve else "rejected"
    BUS.send(sender, "lead", args.get("reason", ""),
             "shutdown_response",
             {"request_id": req_id, "approve": approve})

3. Plan approval follows the identical pattern. The teammate submits a plan (generating a request_id), the lead reviews (referencing the same request_id).

plan_requests = {}

def handle_plan_review(request_id, approve, feedback=""):
    req = plan_requests[request_id]
    req["status"] = "approved" if approve else "rejected"
    BUS.send("lead", req["from"], feedback,
             "plan_approval_response",
             {"request_id": request_id, "approve": approve})

One FSM, two applications. The same pending -> approved | rejected state machine handles any request-response protocol.

What Changed From s09 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s10_team_protocols.py

1. Spawn alice as a coder. Then request her shutdown.
2. List teammates to see alice's status after shutdown approval
3. Spawn bob with a risky refactoring task. Review and reject his plan.
4. Spawn charlie, have him submit a plan, then approve it.
5. Type /team to monitor statuses

En S11 Autonomous Agents (`docs/en/s11-autonomous-agents.md`) Copy Section

s11: Autonomous Agents

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.

Harness layer: Autonomy -- models that find work without being told.

Problem Copy Chapter

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 Copy Chapter

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 Copy Chapter

1. 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")

2. 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

3. 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(encoding="utf-8"))
        if (task.get("status") == "pending"
                and not task.get("owner")
                and not task.get("blockedBy")):
            unclaimed.append(task)
    return unclaimed

4. 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 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s11_autonomous_agents.py

1. Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.
2. Spawn a coder teammate and let it find work from the task board itself
3. Create tasks with dependencies. Watch teammates respect the blocked order.
4. Type /tasks to see the task board with owners
5. Type /team to monitor who is working vs idle

En S12 Worktree Task Isolation (`docs/en/s12-worktree-task-isolation.md`) Copy Section

s12: Worktree + Task Isolation

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > [ s12 ]

"Each works in its own directory, no interference" -- tasks manage goals, worktrees manage directories, bound by ID.

Harness layer: Directory isolation -- parallel execution lanes that never collide.

Problem Copy Chapter

By s11, agents can claim and complete tasks autonomously. But every task runs in one shared directory. Two agents refactoring different modules at the same time will collide: agent A edits config.py, agent B edits config.py, unstaged changes mix, and neither can roll back cleanly.

The task board tracks what to do but has no opinion about where to do it. The fix: give each task its own git worktree directory. Tasks manage goals, worktrees manage execution context. Bind them by task ID.

Solution Copy Chapter

Control plane (.tasks/)             Execution plane (.worktrees/)
+------------------+                +------------------------+
| task_1.json      |                | auth-refactor/         |
|   status: in_progress  <------>   branch: wt/auth-refactor
|   worktree: "auth-refactor"   |   task_id: 1             |
+------------------+                +------------------------+
| task_2.json      |                | ui-login/              |
|   status: pending    <------>     branch: wt/ui-login
|   worktree: "ui-login"       |   task_id: 2             |
+------------------+                +------------------------+
                                    |
                          index.json (worktree registry)
                          events.jsonl (lifecycle log)

State machines:
  Task:     pending -> in_progress -> completed
  Worktree: absent  -> active      -> removed | kept

How It Works Copy Chapter

1. Create a task. Persist the goal first.

TASKS.create("Implement auth refactor")
# -> .tasks/task_1.json  status=pending  worktree=""

2. Create a worktree and bind to the task. Passing task_id auto-advances the task to in_progress.

WORKTREES.create("auth-refactor", task_id=1)
# -> git worktree add -b wt/auth-refactor .worktrees/auth-refactor HEAD
# -> index.json gets new entry, task_1.json gets worktree="auth-refactor"

The binding writes state to both sides:

def bind_worktree(self, task_id, worktree):
    task = self._load(task_id)
    task["worktree"] = worktree
    if task["status"] == "pending":
        task["status"] = "in_progress"
    self._save(task)

3. Run commands in the worktree. cwd points to the isolated directory.

subprocess.run(command, shell=True, cwd=worktree_path,
               capture_output=True, text=True, timeout=300)

4. Close out. Two choices:
   • worktree_keep(name) -- preserve the directory for later.
   • worktree_remove(name, complete_task=True) -- remove directory, complete the bound task, emit event. One call handles teardown + completion.

def remove(self, name, force=False, complete_task=False):
    self._run_git(["worktree", "remove", wt["path"]])
    if complete_task and wt.get("task_id") is not None:
        self.tasks.update(wt["task_id"], status="completed")
        self.tasks.unbind_worktree(wt["task_id"])
        self.events.emit("task.completed", ...)

5. Event stream. Every lifecycle step emits to .worktrees/events.jsonl:

{
  "event": "worktree.remove.after",
  "task": {"id": 1, "status": "completed"},
  "worktree": {"name": "auth-refactor", "status": "removed"},
  "ts": 1730000000
}

Events emitted: worktree.create.before/after/failed, worktree.remove.before/after/failed, worktree.keep, task.completed.

After a crash, state reconstructs from .tasks/ + .worktrees/index.json on disk. Conversation memory is volatile; file state is durable.

What Changed From s11 Copy Chapter

Try It Copy Chapter

cd learn-claude-code
python agents/s12_worktree_task_isolation.py

1. Create tasks for backend auth and frontend login page, then list tasks.
2. Create worktree "auth-refactor" for task 1, then bind task 2 to a new worktree "ui-login".
3. Run "git status --short" in worktree "auth-refactor".
4. Keep worktree "ui-login", then list worktrees and inspect events.
5. Remove worktree "auth-refactor" with complete_task=true, then list tasks/worktrees/events.

Ja S01 The Agent Loop (`docs/ja/s01-the-agent-loop.md`) Copy Section

s01: The Agent Loop

[ s01 ] s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"One loop & Bash is all you need" -- 1つのツール + 1つのループ = エージェント。

Harness 層: ループ -- モデルと現実世界を繋ぐ最初の接点。

問題 Copy Chapter

言語モデルはコードについて推論できるが、現実世界に触れられない。ファイルを読めず、テストを実行できず、エラーを確認できない。ループがなければ、ツール呼び出しのたびにユーザーが手動で結果をコピーペーストする必要がある。つまりユーザー自身がループになる。

解決策 Copy Chapter

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> |  Tool   |
| prompt |      |       |      | execute |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                    (loop until stop_reason != "tool_use")

1つの終了条件がフロー全体を制御する。モデルがツール呼び出しを止めるまでループが回り続ける。

仕組み Copy Chapter

1. ユーザーのプロンプトが最初のメッセージになる。

messages.append({"role": "user", "content": query})

2. メッセージとツール定義をLLMに送信する。

response = client.messages.create(
    model=MODEL, system=SYSTEM, messages=messages,
    tools=TOOLS, max_tokens=8000,
)

3. アシスタントのレスポンスを追加し、stop_reasonを確認する。ツールが呼ばれなければ終了。

messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
    return

4. 各ツール呼び出しを実行し、結果を収集してuserメッセージとして追加。ステップ2に戻る。

results = []
for block in response.content:
    if block.type == "tool_use":
        output = run_bash(block.input["command"])
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })
messages.append({"role": "user", "content": results})

1つの関数にまとめると:

def agent_loop(query):
    messages = [{"role": "user", "content": query}]
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return

        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = run_bash(block.input["command"])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

これでエージェント全体が30行未満に収まる。本コースの残りはすべてこのループの上に積み重なる -- ループ自体は変わらない。

変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s01_agent_loop.py

1. Create a file called hello.py that prints "Hello, World!"
2. List all Python files in this directory
3. What is the current git branch?
4. Create a directory called test_output and write 3 files in it

Ja S02 Tool Use (`docs/ja/s02-tool-use.md`) Copy Section

s02: Tool Use

s01 > [ s02 ] s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"ツールを足すなら、ハンドラーを1つ足すだけ" -- ループは変わらない。新ツールは dispatch map に登録するだけ。

Harness 層: ツール分配 -- モデルが届く範囲を広げる。

問題 Copy Chapter

bashだけでは、エージェントは何でもシェル経由で行う。catは予測不能に切り詰め、sedは特殊文字で壊れ、すべてのbash呼び出しが制約のないセキュリティ面になる。read_fileやwrite_fileのような専用ツールなら、ツールレベルでパスのサンドボックス化を強制できる。

重要な点: ツールを追加してもループの変更は不要。

解決策 Copy Chapter

+--------+      +-------+      +------------------+
|  User  | ---> |  LLM  | ---> | Tool Dispatch    |
| prompt |      |       |      | {                |
+--------+      +---+---+      |   bash: run_bash |
                    ^           |   read: run_read |
                    |           |   write: run_wr  |
                    +-----------+   edit: run_edit |
                    tool_result | }                |
                                +------------------+

The dispatch map is a dict: {tool_name: handler_function}.
One lookup replaces any if/elif chain.

仕組み Copy Chapter

1. 各ツールにハンドラ関数を定義する。パスのサンドボックス化でワークスペース外への脱出を防ぐ。

def safe_path(p: str) -> Path:
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

def run_read(path: str, limit: int = None) -> str:
    text = safe_path(path).read_text(encoding="utf-8")
    lines = text.splitlines()
    if limit and limit < len(lines):
        lines = lines[:limit]
    return "\n".join(lines)[:50000]

2. ディスパッチマップがツール名とハンドラを結びつける。

TOOL_HANDLERS = {
    "bash":       lambda **kw: run_bash(kw["command"]),
    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),
    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"],
                                        kw["new_text"]),
}

3. ループ内で名前によりハンドラをルックアップする。ループ本体はs01から不変。

for block in response.content:
    if block.type == "tool_use":
        handler = TOOL_HANDLERS.get(block.name)
        output = handler(**block.input) if handler \
            else f"Unknown tool: {block.name}"
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })

ツール追加 = ハンドラ追加 + スキーマ追加。ループは決して変わらない。

s01からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s02_tool_use.py

1. Read the file requirements.txt
2. Create a file called greet.py with a greet(name) function
3. Edit greet.py to add a docstring to the function
4. Read greet.py to verify the edit worked

Ja S03 Todo Write (`docs/ja/s03-todo-write.md`) Copy Section

s03: TodoWrite

s01 > s02 > [ s03 ] s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"計画のないエージェントは行き当たりばったり" -- まずステップを書き出し、それから実行。

Harness 層: 計画 -- 航路を描かずにモデルを軌道に乗せる。

問題 Copy Chapter

マルチステップのタスクで、モデルは途中で迷子になる。作業を繰り返したり、ステップを飛ばしたり、脱線したりする。長い会話になるほど悪化する -- ツール結果がコンテキストを埋めるにつれ、システムプロンプトの影響力が薄れる。10ステップのリファクタリングでステップ1-3を完了した後、残りを忘れて即興を始めてしまう。

解決策 Copy Chapter

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> | Tools   |
| prompt |      |       |      | + todo  |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                          |
              +-----------+-----------+
              | TodoManager state     |
              | [ ] task A            |
              | [>] task B  <- doing  |
              | [x] task C            |
              +-----------------------+
                          |
              if rounds_since_todo >= 3:
                inject <reminder> into tool_result

仕組み Copy Chapter

1. TodoManagerはアイテムのリストをステータス付きで保持する。in_progressにできるのは同時に1つだけ。

class TodoManager:
    def update(self, items: list) -> str:
        validated, in_progress_count = [], 0
        for item in items:
            status = item.get("status", "pending")
            if status == "in_progress":
                in_progress_count += 1
            validated.append({"id": item["id"], "text": item["text"],
                              "status": status})
        if in_progress_count > 1:
            raise ValueError("Only one task can be in_progress")
        self.items = validated
        return self.render()

2. todoツールは他のツールと同様にディスパッチマップに追加される。

TOOL_HANDLERS = {
    # ...base tools...
    "todo": lambda **kw: TODO.update(kw["items"]),
}

3. nagリマインダーが、モデルが3ラウンド以上todoを呼ばなかった場合にナッジを注入する。

if rounds_since_todo >= 3 and messages:
    last = messages[-1]
    if last["role"] == "user" and isinstance(last.get("content"), list):
        last["content"].insert(0, {
            "type": "text",
            "text": "<reminder>Update your todos.</reminder>",
        })

「一度にin_progressは1つだけ」の制約が逐次的な集中を強制し、nagリマインダーが説明責任を生む。

s02からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s03_todo_write.py

1. Refactor the file hello.py: add type hints, docstrings, and a main guard
2. Create a Python package with __init__.py, utils.py, and tests/test_utils.py
3. Review all Python files and fix any style issues

Ja S04 Subagent (`docs/ja/s04-subagent.md`) Copy Section

s04: Subagents

s01 > s02 > s03 > [ s04 ] s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"大きなタスクを分割し、各サブタスクにクリーンなコンテキストを" -- サブエージェントは独立した messages[] を使い、メイン会話を汚さない。

Harness 層: コンテキスト隔離 -- モデルの思考の明晰さを守る。

問題 Copy Chapter

エージェントが作業するにつれ、messages配列は膨張し続ける。すべてのファイル読み取り、すべてのbash出力がコンテキストに永久に残る。「このプロジェクトはどのテストフレームワークを使っているか」という質問は5つのファイルを読む必要があるかもしれないが、親に必要なのは「pytest」という答えだけだ。

解決策 Copy Chapter

Parent agent                     Subagent
+------------------+             +------------------+
| messages=[...]   |             | messages=[]      | <-- fresh
|                  |  dispatch   |                  |
| tool: task       | ----------> | while tool_use:  |
|   prompt="..."   |             |   call tools     |
|                  |  summary    |   append results |
|   result = "..." | <---------- | return last text |
+------------------+             +------------------+

Parent context stays clean. Subagent context is discarded.

仕組み Copy Chapter

1. 親にtaskツールを追加する。子はtaskを除くすべての基本ツールを取得する(再帰的な生成は不可)。

PARENT_TOOLS = CHILD_TOOLS + [
    {"name": "task",
     "description": "Spawn a subagent with fresh context.",
     "input_schema": {
         "type": "object",
         "properties": {"prompt": {"type": "string"}},
         "required": ["prompt"],
     }},
]

2. サブエージェントはmessages=[]で開始し、自身のループを実行する。最終テキストだけが親に返る。

def run_subagent(prompt: str) -> str:
    sub_messages = [{"role": "user", "content": prompt}]
    for _ in range(30):  # safety limit
        response = client.messages.create(
            model=MODEL, system=SUBAGENT_SYSTEM,
            messages=sub_messages,
            tools=CHILD_TOOLS, max_tokens=8000,
        )
        sub_messages.append({"role": "assistant",
                             "content": response.content})
        if response.stop_reason != "tool_use":
            break
        results = []
        for block in response.content:
            if block.type == "tool_use":
                handler = TOOL_HANDLERS.get(block.name)
                output = handler(**block.input)
                results.append({"type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output)[:50000]})
        sub_messages.append({"role": "user", "content": results})
    return "".join(
        b.text for b in response.content if hasattr(b, "text")
    ) or "(no summary)"

子のメッセージ履歴全体(30回以上のツール呼び出し)は破棄される。親は1段落の要約を通常のtool_resultとして受け取る。

s03からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s04_subagent.py

1. Use a subtask to find what testing framework this project uses
2. Delegate: read all .py files and summarize what each one does
3. Use a task to create a new module, then verify it from here

Ja S05 Skill Loading (`docs/ja/s05-skill-loading.md`) Copy Section

s05: Skills

s01 > s02 > s03 > s04 > [ s05 ] s06 | s07 > s08 > s09 > s10 > s11 > s12

"必要な知識を、必要な時に読み込む" -- system prompt ではなく tool_result で注入。

Harness 層: オンデマンド知識 -- モデルが求めた時だけ渡すドメイン専門性。

問題 Copy Chapter

エージェントにドメイン固有のワークフローを遵守させたい: gitの規約、テストパターン、コードレビューチェックリスト。すべてをシステムプロンプトに入れると、使われないスキルにトークンを浪費する。10スキル x 2000トークン = 20,000トークン、ほとんどが任意のタスクに無関係だ。

解決策 Copy Chapter

System prompt (Layer 1 -- always present):
+--------------------------------------+
| You are a coding agent.              |
| Skills available:                    |
|   - git: Git workflow helpers        |  ~100 tokens/skill
|   - test: Testing best practices     |
+--------------------------------------+

When model calls load_skill("git"):
+--------------------------------------+
| tool_result (Layer 2 -- on demand):  |
| <skill name="git">                   |
|   Full git workflow instructions...  |  ~2000 tokens
|   Step 1: ...                        |
| </skill>                             |
+--------------------------------------+

第1層: スキル名をシステムプロンプトに(低コスト)。第2層: スキル本体をtool_resultに(オンデマンド)。

仕組み Copy Chapter

1. 各スキルは SKILL.md ファイルを含むディレクトリとして配置される。

skills/
  pdf/
    SKILL.md       # ---\n name: pdf\n description: Process PDF files\n ---\n ...
  code-review/
    SKILL.md       # ---\n name: code-review\n description: Review code\n ---\n ...

2. SkillLoaderが SKILL.md を再帰的に探索し、ディレクトリ名をスキル識別子として使用する。

class SkillLoader:
    def __init__(self, skills_dir: Path):
        self.skills = {}
        for f in sorted(skills_dir.rglob("SKILL.md")):
            text = f.read_text(encoding="utf-8")
            meta, body = self._parse_frontmatter(text)
            name = meta.get("name", f.parent.name)
            self.skills[name] = {"meta": meta, "body": body}

    def get_descriptions(self) -> str:
        lines = []
        for name, skill in self.skills.items():
            desc = skill["meta"].get("description", "")
            lines.append(f"  - {name}: {desc}")
        return "\n".join(lines)

    def get_content(self, name: str) -> str:
        skill = self.skills.get(name)
        if not skill:
            return f"Error: Unknown skill '{name}'."
        return f"<skill name=\"{name}\">\n{skill['body']}\n</skill>"

3. 第1層はシステムプロンプトに配置。第2層は通常のツールハンドラ。

SYSTEM = f"""You are a coding agent at {WORKDIR}.
Skills available:
{SKILL_LOADER.get_descriptions()}"""

TOOL_HANDLERS = {
    # ...base tools...
    "load_skill": lambda **kw: SKILL_LOADER.get_content(kw["name"]),
}

モデルはどのスキルが存在するかを知り(低コスト)、関連する時にだけ読み込む(高コスト)。

s04からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s05_skill_loading.py

1. What skills are available?
2. Load the agent-builder skill and follow its instructions
3. I need to do a code review -- load the relevant skill first
4. Build an MCP server using the mcp-builder skill

Ja S06 Context Compact (`docs/ja/s06-context-compact.md`) Copy Section

s06: Context Compact

s01 > s02 > s03 > s04 > s05 > [ s06 ] | s07 > s08 > s09 > s10 > s11 > s12

"コンテキストはいつか溢れる、空ける手段が要る" -- 3層圧縮で無限セッションを実現。

Harness 層: 圧縮 -- クリーンな記憶、無限のセッション。

問題 Copy Chapter

コンテキストウィンドウは有限だ。1000行のファイルに対するread_file1回で約4000トークンを消費する。30ファイルを読み20回のbashコマンドを実行すると、100,000トークン超。圧縮なしでは、エージェントは大規模コードベースで作業できない。

解決策 Copy Chapter

積極性を段階的に上げる3層構成:

Every turn:
+------------------+
| Tool call result |
+------------------+
        |
        v
[Layer 1: micro_compact]        (silent, every turn)
  Replace tool_result > 3 turns old
  with "[Previous: used {tool_name}]"
        |
        v
[Check: tokens > 50000?]
   |               |
   no              yes
   |               |
   v               v
continue    [Layer 2: auto_compact]
              Save transcript to .transcripts/
              LLM summarizes conversation.
              Replace all messages with [summary].
                    |
                    v
            [Layer 3: compact tool]
              Model calls compact explicitly.
              Same summarization as auto_compact.

仕組み Copy Chapter

1. 第1層 -- micro_compact: 各LLM呼び出しの前に、古いツール結果をプレースホルダーに置換する。

def micro_compact(messages: list) -> list:
    tool_results = []
    for i, msg in enumerate(messages):
        if msg["role"] == "user" and isinstance(msg.get("content"), list):
            for j, part in enumerate(msg["content"]):
                if isinstance(part, dict) and part.get("type") == "tool_result":
                    tool_results.append((i, j, part))
    if len(tool_results) <= KEEP_RECENT:
        return messages
    for _, _, part in tool_results[:-KEEP_RECENT]:
        if len(part.get("content", "")) > 100:
            part["content"] = f"[Previous: used {tool_name}]"
    return messages

2. 第2層 -- auto_compact: トークンが閾値を超えたら、完全なトランスクリプトをディスクに保存し、LLMに要約を依頼する。

def auto_compact(messages: list) -> list:
    # Save transcript for recovery
    transcript_path = TRANSCRIPT_DIR / f"transcript_{int(time.time())}.jsonl"
    with open(transcript_path, "w") as f:
        for msg in messages:
            f.write(json.dumps(msg, default=str) + "\n")
    # LLM summarizes
    response = client.messages.create(
        model=MODEL,
        messages=[{"role": "user", "content":
            "Summarize this conversation for continuity..."
            + json.dumps(messages, default=str)[:80000]}],
        max_tokens=2000,
    )
    return [
        {"role": "user", "content": f"[Compressed]\n\n{response.content[0].text}"},
    ]

3. 第3層 -- manual compact: compactツールが同じ要約処理をオンデマンドでトリガーする。

4. ループが3層すべてを統合する:

def agent_loop(messages: list):
    while True:
        micro_compact(messages)                        # Layer 1
        if estimate_tokens(messages) > THRESHOLD:
            messages[:] = auto_compact(messages)       # Layer 2
        response = client.messages.create(...)
        # ... tool execution ...
        if manual_compact:
            messages[:] = auto_compact(messages)       # Layer 3

トランスクリプトがディスク上に完全な履歴を保持する。何も真に失われず、アクティブなコンテキストの外に移動されるだけ。

s05からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s06_context_compact.py

1. Read every Python file in the agents/ directory one by one (micro-compactが古い結果を置換するのを観察する)
2. Keep reading files until compression triggers automatically
3. Use the compact tool to manually compress the conversation

Ja S07 Task System (`docs/ja/s07-task-system.md`) Copy Section

s07: Task System

s01 > s02 > s03 > s04 > s05 > s06 | [ s07 ] s08 > s09 > s10 > s11 > s12

"大きな目標を小タスクに分解し、順序付けし、ディスクに記録する" -- ファイルベースのタスクグラフ、マルチエージェント協調の基盤。

Harness 層: 永続タスク -- どの会話よりも長く生きる目標。

問題 Copy Chapter

s03のTodoManagerはメモリ上のフラットなチェックリストに過ぎない: 順序なし、依存関係なし、ステータスは完了か未完了のみ。実際の目標には構造がある -- タスクBはタスクAに依存し、タスクCとDは並行実行でき、タスクEはCとDの両方を待つ。

明示的な関係がなければ、エージェントは何が実行可能で、何がブロックされ、何が同時に走れるかを判断できない。しかもリストはメモリ上にしかないため、コンテキスト圧縮(s06)で消える。

解決策 Copy Chapter

フラットなチェックリストをディスクに永続化するタスクグラフに昇格させる。各タスクは1つのJSONファイルで、ステータス・前方依存(blockedBy)を持つ。タスクグラフは常に3つの問いに答える:

• 何が実行可能か? -- 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"}

タスクグラフ (DAG):
                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

順序:       task 1 は 2 と 3 より先に完了する必要がある
並行:       task 2 と 3 は同時に実行できる
依存:       task 4 は 2 と 3 の両方を待つ
ステータス: pending -> in_progress -> completed

このタスクグラフは s07 以降の全メカニズムの協調バックボーンとなる: バックグラウンド実行(s08)、マルチエージェントチーム(s09+)、worktree分離(s12)はすべてこの同じ構造を読み書きする。

仕組み Copy Chapter

1. TaskManager: タスクごとに1つのJSONファイル、依存グラフ付きCRUD。

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": [],
                "owner": ""}
        self._save(task)
        self._next_id += 1
        return json.dumps(task, indent=2)

2. 依存解除: タスク完了時に、他タスクのblockedByリストから完了IDを除去し、後続タスクをアンブロックする。

def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text(encoding="utf-8"))
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)

3. ステータス遷移 + 依存配線: updateがステータス変更と依存エッジを担う。

def update(self, task_id, status=None,
           add_blocked_by=None, remove_blocked_by=None):
    task = self._load(task_id)
    if status:
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)
    if add_blocked_by:
        task["blockedBy"] = list(set(task["blockedBy"] + add_blocked_by))
    if remove_blocked_by:
        task["blockedBy"] = [x for x in task["blockedBy"] if x not in remove_blocked_by]
    self._save(task)

4. 4つのタスクツールをディスパッチマップに追加する。

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"]),
}

s07以降、タスクグラフがマルチステップ作業のデフォルト。s03のTodoは軽量な単一セッション用チェックリストとして残る。

s06からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s07_task_system.py

1. Create 3 tasks: "Setup project", "Write code", "Write tests". Make them depend on each other in order.
2. List all tasks and show the dependency graph
3. Complete task 1 and then list tasks to see task 2 unblocked
4. Create a task board for refactoring: parse -> transform -> emit -> test, where transform and emit can run in parallel after parse

Ja S08 Background Tasks (`docs/ja/s08-background-tasks.md`) Copy Section

s08: Background Tasks

s01 > s02 > s03 > s04 > s05 > s06 | s07 > [ s08 ] s09 > s10 > s11 > s12

"遅い操作はバックグラウンドへ、エージェントは次を考え続ける" -- デーモンスレッドがコマンド実行、完了後に通知を注入。

Harness 層: バックグラウンド実行 -- モデルが考え続ける間、Harness が待つ。

問題 Copy Chapter

一部のコマンドは数分かかる: npm install、pytest、docker build。ブロッキングループでは、モデルはサブプロセスの完了を待って座っている。ユーザーが「依存関係をインストールして、その間にconfigファイルを作って」と言っても、エージェントは並列ではなく逐次的に処理する。

解決策 Copy Chapter

Main thread                Background thread
+-----------------+        +-----------------+
| agent loop      |        | subprocess runs |
| ...             |        | ...             |
| [LLM call] <---+------- | enqueue(result) |
|  ^drain queue   |        +-----------------+
+-----------------+

Timeline:
Agent --[spawn A]--[spawn B]--[other work]----
             |          |
             v          v
          [A runs]   [B runs]      (parallel)
             |          |
             +-- results injected before next LLM call --+

仕組み Copy Chapter

1. BackgroundManagerがスレッドセーフな通知キューでタスクを追跡する。

class BackgroundManager:
    def __init__(self):
        self.tasks = {}
        self._notification_queue = []
        self._lock = threading.Lock()

2. run()がデーモンスレッドを開始し、即座にリターンする。

def run(self, command: str) -> str:
    task_id = str(uuid.uuid4())[:8]
    self.tasks[task_id] = {"status": "running", "command": command}
    thread = threading.Thread(
        target=self._execute, args=(task_id, command), daemon=True)
    thread.start()
    return f"Background task {task_id} started"

3. サブプロセス完了時に、結果を通知キューへ。

def _execute(self, task_id, command):
    try:
        r = subprocess.run(command, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=300)
        output = (r.stdout + r.stderr).strip()[:50000]
    except subprocess.TimeoutExpired:
        output = "Error: Timeout (300s)"
    with self._lock:
        self._notification_queue.append({
            "task_id": task_id, "result": output[:500]})

4. エージェントループが各LLM呼び出しの前に通知をドレインする。

def agent_loop(messages: list):
    while True:
        notifs = BG.drain_notifications()
        if notifs:
            notif_text = "\n".join(
                f"[bg:{n['task_id']}] {n['result']}" for n in notifs)
            messages.append({"role": "user",
                "content": f"<background-results>\n{notif_text}\n"
                           f"</background-results>"})
        response = client.messages.create(...)

ループはシングルスレッドのまま。サブプロセスI/Oだけが並列化される。

s07からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s08_background_tasks.py

1. Run "sleep 5 && echo done" in the background, then create a file while it runs
2. Start 3 background tasks: "sleep 2", "sleep 4", "sleep 6". Check their status.
3. Run pytest in the background and keep working on other things

Ja S09 Agent Teams (`docs/ja/s09-agent-teams.md`) Copy Section

s09: Agent Teams

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > [ s09 ] s10 > s11 > s12

"一人で終わらないなら、チームメイトに任せる" -- 永続チームメイト + 非同期メールボックス。

Harness 層: チームメールボックス -- 複数モデルをファイルで協調。

問題 Copy Chapter

サブエージェント(s04)は使い捨てだ: 生成し、作業し、要約を返し、消滅する。アイデンティティもなく、呼び出し間の記憶もない。バックグラウンドタスク(s08)はシェルコマンドを実行するが、LLM誘導の意思決定はできない。

本物のチームワークには: (1)単一プロンプトを超えて存続する永続エージェント、(2)アイデンティティとライフサイクル管理、(3)エージェント間の通信チャネルが必要だ。

解決策 Copy Chapter

Teammate lifecycle:
  spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN

Communication:
  .team/
    config.json           <- team roster + statuses
    inbox/
      alice.jsonl         <- append-only, drain-on-read
      bob.jsonl
      lead.jsonl

              +--------+    send("alice","bob","...")    +--------+
              | alice  | -----------------------------> |  bob   |
              | loop   |    bob.jsonl << {json_line}    |  loop  |
              +--------+                                +--------+
                   ^                                         |
                   |        BUS.read_inbox("alice")          |
                   +---- alice.jsonl -> read + drain ---------+

仕組み Copy Chapter

1. TeammateManagerがconfig.jsonでチーム名簿を管理する。

class TeammateManager:
    def __init__(self, team_dir: Path):
        self.dir = team_dir
        self.dir.mkdir(exist_ok=True)
        self.config_path = self.dir / "config.json"
        self.config = self._load_config()
        self.threads = {}

2. spawn()がチームメイトを作成し、そのエージェントループをスレッドで開始する。

def spawn(self, name: str, role: str, prompt: str) -> str:
    member = {"name": name, "role": role, "status": "working"}
    self.config["members"].append(member)
    self._save_config()
    thread = threading.Thread(
        target=self._teammate_loop,
        args=(name, role, prompt), daemon=True)
    thread.start()
    return f"Spawned teammate '{name}' (role: {role})"

3. MessageBus: 追記専用のJSONLインボックス。send()がJSON行を追記し、read_inbox()がすべて読み取ってドレインする。

class MessageBus:
    def send(self, sender, to, content, msg_type="message", extra=None):
        msg = {"type": msg_type, "from": sender,
               "content": content, "timestamp": time.time()}
        if extra:
            msg.update(extra)
        with open(self.dir / f"{to}.jsonl", "a") as f:
            f.write(json.dumps(msg) + "\n")

    def read_inbox(self, name):
        path = self.dir / f"{name}.jsonl"
        if not path.exists(): return "[]"
        msgs = [json.loads(l) for l in path.read_text(encoding="utf-8").strip().splitlines() if l]
        path.write_text("", encoding="utf-8")  # drain
        return json.dumps(msgs, indent=2)

4. 各チームメイトは各LLM呼び出しの前にインボックスを確認し、受信メッセージをコンテキストに注入する。

def _teammate_loop(self, name, role, prompt):
    messages = [{"role": "user", "content": prompt}]
    for _ in range(50):
        inbox = BUS.read_inbox(name)
        if inbox != "[]":
            messages.append({"role": "user",
                "content": f"<inbox>{inbox}</inbox>"})
        response = client.messages.create(...)
        if response.stop_reason != "tool_use":
            break
        # execute tools, append results...
    self._find_member(name)["status"] = "idle"

s08からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s09_agent_teams.py

1. Spawn alice (coder) and bob (tester). Have alice send bob a message.
2. Broadcast "status update: phase 1 complete" to all teammates
3. Check the lead inbox for any messages
4. /teamと入力してステータス付きのチーム名簿を確認する
5. /inboxと入力してリーダーのインボックスを手動確認する

Ja S10 Team Protocols (`docs/ja/s10-team-protocols.md`) Copy Section

s10: Team Protocols

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > [ s10 ] s11 > s12

"チームメイト間には統一の通信ルールが必要" -- 1つの request-response パターンが全交渉を駆動。

Harness 層: プロトコル -- モデル間の構造化されたハンドシェイク。

問題 Copy Chapter

s09ではチームメイトが作業し通信するが、構造化された協調がない:

シャットダウン: スレッドを強制終了するとファイルが中途半端に書かれ、config.jsonが不正な状態になる。ハンドシェイクが必要 -- リーダーが要求し、チームメイトが承認(完了して退出)か拒否(作業継続)する。

プラン承認: リーダーが「認証モジュールをリファクタリングして」と言うと、チームメイトは即座に開始する。リスクの高い変更では、実行前にリーダーが計画をレビューすべきだ。

両方とも同じ構造: 一方がユニークIDを持つリクエストを送り、他方がそのIDで応答する。

解決策 Copy Chapter

Shutdown Protocol            Plan Approval Protocol
==================           ======================

Lead             Teammate    Teammate           Lead
  |                 |           |                 |
  |--shutdown_req-->|           |--plan_req------>|
  | {req_id:"abc"}  |           | {req_id:"xyz"}  |
  |                 |           |                 |
  |<--shutdown_resp-|           |<--plan_resp-----|
  | {req_id:"abc",  |           | {req_id:"xyz",  |
  |  approve:true}  |           |  approve:true}  |

Shared FSM:
  [pending] --approve--> [approved]
  [pending] --reject---> [rejected]

Trackers:
  shutdown_requests = {req_id: {target, status}}
  plan_requests     = {req_id: {from, plan, status}}

仕組み Copy Chapter

1. リーダーがrequest_idを生成し、インボックス経由でシャットダウンを開始する。

shutdown_requests = {}

def handle_shutdown_request(teammate: str) -> str:
    req_id = str(uuid.uuid4())[:8]
    shutdown_requests[req_id] = {"target": teammate, "status": "pending"}
    BUS.send("lead", teammate, "Please shut down gracefully.",
             "shutdown_request", {"request_id": req_id})
    return f"Shutdown request {req_id} sent (status: pending)"

2. チームメイトがリクエストを受信し、承認または拒否で応答する。

if tool_name == "shutdown_response":
    req_id = args["request_id"]
    approve = args["approve"]
    shutdown_requests[req_id]["status"] = "approved" if approve else "rejected"
    BUS.send(sender, "lead", args.get("reason", ""),
             "shutdown_response",
             {"request_id": req_id, "approve": approve})

3. プラン承認も同一パターン。チームメイトがプランを提出(request_idを生成)、リーダーがレビュー(同じrequest_idを参照)。

plan_requests = {}

def handle_plan_review(request_id, approve, feedback=""):
    req = plan_requests[request_id]
    req["status"] = "approved" if approve else "rejected"
    BUS.send("lead", req["from"], feedback,
             "plan_approval_response",
             {"request_id": request_id, "approve": approve})

1つのFSM、2つの応用。同じpending -> approved | rejected状態機械が、あらゆるリクエスト-レスポンスプロトコルに適用できる。

s09からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s10_team_protocols.py

1. Spawn alice as a coder. Then request her shutdown.
2. List teammates to see alice's status after shutdown approval
3. Spawn bob with a risky refactoring task. Review and reject his plan.
4. Spawn charlie, have him submit a plan, then approve it.
5. /teamと入力してステータスを監視する

Ja S11 Autonomous Agents (`docs/ja/s11-autonomous-agents.md`) Copy Section

s11: Autonomous Agents

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > [ s11 ] s12

"チームメイトが自らボードを見て、仕事を取る" -- リーダーが逐一割り振る必要はない。

Harness 層: 自律 -- 指示なしで仕事を見つけるモデル。

問題 Copy Chapter

s09-s10では、チームメイトは明示的に指示された時のみ作業する。リーダーは各チームメイトを特定のプロンプトでspawnしなければならない。タスクボードに未割り当てのタスクが10個あっても、リーダーが手動で各タスクを割り当てる。これはスケールしない。

真の自律性とは、チームメイトが自分で作業を見つけること: タスクボードをスキャンし、未確保のタスクを確保し、作業し、完了したら次を探す。

もう1つの問題: コンテキスト圧縮(s06)後にエージェントが自分の正体を忘れる可能性がある。アイデンティティ再注入がこれを解決する。

解決策 Copy Chapter

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)

仕組み Copy Chapter

1. チームメイトのループはWORKとIDLEの2フェーズ。LLMがツール呼び出しを止めた時(またはidleツールを呼んだ時)、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")

2. IDLEフェーズがインボックスとタスクボードをポーリングする。

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

3. タスクボードスキャン: pendingかつ未割り当てかつブロックされていないタスクを探す。

def scan_unclaimed_tasks() -> list:
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text(encoding="utf-8"))
        if (task.get("status") == "pending"
                and not task.get("owner")
                and not task.get("blockedBy")):
            unclaimed.append(task)
    return unclaimed

4. アイデンティティ再注入: コンテキストが短すぎる(圧縮が起きた)場合にアイデンティティブロックを挿入する。

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

s10からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s11_autonomous_agents.py

1. Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.
2. Spawn a coder teammate and let it find work from the task board itself
3. Create tasks with dependencies. Watch teammates respect the blocked order.
4. /tasksと入力してオーナー付きのタスクボードを確認する
5. /teamと入力して誰が作業中でアイドルかを監視する

Ja S12 Worktree Task Isolation (`docs/ja/s12-worktree-task-isolation.md`) Copy Section

s12: Worktree + Task Isolation

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > [ s12 ]

"各自のディレクトリで作業し、互いに干渉しない" -- タスクは目標を管理、worktree はディレクトリを管理、IDで紐付け。

Harness 層: ディレクトリ隔離 -- 決して衝突しない並列実行レーン。

問題 Copy Chapter

s11までにエージェントはタスクを自律的に確保して完了できるようになった。しかし全タスクが1つの共有ディレクトリで走る。2つのエージェントが同時に異なるモジュールをリファクタリングすると衝突する: 片方がconfig.pyを編集し、もう片方もconfig.pyを編集し、未コミットの変更が混ざり合い、どちらもクリーンにロールバックできない。

タスクボードは何をやるかを追跡するが、どこでやるかには関知しない。解決策: 各タスクに専用のgit worktreeディレクトリを与える。タスクが目標を管理し、worktreeが実行コンテキストを管理する。タスクIDで紐付ける。

解決策 Copy Chapter

Control plane (.tasks/)             Execution plane (.worktrees/)
+------------------+                +------------------------+
| task_1.json      |                | auth-refactor/         |
|   status: in_progress  <------>   branch: wt/auth-refactor
|   worktree: "auth-refactor"   |   task_id: 1             |
+------------------+                +------------------------+
| task_2.json      |                | ui-login/              |
|   status: pending    <------>     branch: wt/ui-login
|   worktree: "ui-login"       |   task_id: 2             |
+------------------+                +------------------------+
                                    |
                          index.json (worktree registry)
                          events.jsonl (lifecycle log)

State machines:
  Task:     pending -> in_progress -> completed
  Worktree: absent  -> active      -> removed | kept

仕組み Copy Chapter

1. タスクを作成する。 まず目標を永続化する。

TASKS.create("Implement auth refactor")
# -> .tasks/task_1.json  status=pending  worktree=""

2. worktreeを作成してタスクに紐付ける。 task_idを渡すと、タスクが自動的にin_progressに遷移する。

WORKTREES.create("auth-refactor", task_id=1)
# -> git worktree add -b wt/auth-refactor .worktrees/auth-refactor HEAD
# -> index.json gets new entry, task_1.json gets worktree="auth-refactor"

紐付けは両側に状態を書き込む:

def bind_worktree(self, task_id, worktree):
    task = self._load(task_id)
    task["worktree"] = worktree
    if task["status"] == "pending":
        task["status"] = "in_progress"
    self._save(task)

3. worktree内でコマンドを実行する。 cwdが分離ディレクトリを指す。

subprocess.run(command, shell=True, cwd=worktree_path,
               capture_output=True, text=True, timeout=300)

4. 終了処理。 2つの選択肢:
   • worktree_keep(name) -- ディレクトリを保持する。
   • worktree_remove(name, complete_task=True) -- ディレクトリを削除し、紐付けられたタスクを完了し、イベントを発行する。1回の呼び出しで後片付けと完了を処理する。

def remove(self, name, force=False, complete_task=False):
    self._run_git(["worktree", "remove", wt["path"]])
    if complete_task and wt.get("task_id") is not None:
        self.tasks.update(wt["task_id"], status="completed")
        self.tasks.unbind_worktree(wt["task_id"])
        self.events.emit("task.completed", ...)

5. イベントストリーム。 ライフサイクルの各ステップが.worktrees/events.jsonlに記録される:

{
  "event": "worktree.remove.after",
  "task": {"id": 1, "status": "completed"},
  "worktree": {"name": "auth-refactor", "status": "removed"},
  "ts": 1730000000
}

発行されるイベント: worktree.create.before/after/failed, worktree.remove.before/after/failed, worktree.keep, task.completed。

クラッシュ後も.tasks/ + .worktrees/index.jsonから状態を再構築できる。会話メモリは揮発性だが、ファイル状態は永続的だ。

s11からの変更点 Copy Chapter

試してみる Copy Chapter

cd learn-claude-code
python agents/s12_worktree_task_isolation.py

1. Create tasks for backend auth and frontend login page, then list tasks.
2. Create worktree "auth-refactor" for task 1, then bind task 2 to a new worktree "ui-login".
3. Run "git status --short" in worktree "auth-refactor".
4. Keep worktree "ui-login", then list worktrees and inspect events.
5. Remove worktree "auth-refactor" with complete_task=true, then list tasks/worktrees/events.

Zh S01 The Agent Loop (`docs/zh/s01-the-agent-loop.md`) Copy Section

s01: The Agent Loop (Agent 循环)

[ s01 ] s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"One loop & Bash is all you need" -- 一个工具 + 一个循环 = 一个 Agent。

Harness 层: 循环 -- 模型与真实世界的第一道连接。

问题 Copy Chapter

语言模型能推理代码, 但碰不到真实世界 -- 不能读文件、跑测试、看报错。没有循环, 每次工具调用你都得手动把结果粘回去。你自己就是那个循环。

解决方案 Copy Chapter

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> |  Tool   |
| prompt |      |       |      | execute |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                    (loop until stop_reason != "tool_use")

一个退出条件控制整个流程。循环持续运行, 直到模型不再调用工具。

工作原理 Copy Chapter

1. 用户 prompt 作为第一条消息。

messages.append({"role": "user", "content": query})

2. 将消息和工具定义一起发给 LLM。

response = client.messages.create(
    model=MODEL, system=SYSTEM, messages=messages,
    tools=TOOLS, max_tokens=8000,
)

3. 追加助手响应。检查 stop_reason -- 如果模型没有调用工具, 结束。

messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
    return

4. 执行每个工具调用, 收集结果, 作为 user 消息追加。回到第 2 步。

results = []
for block in response.content:
    if block.type == "tool_use":
        output = run_bash(block.input["command"])
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })
messages.append({"role": "user", "content": results})

组装为一个完整函数:

def agent_loop(query):
    messages = [{"role": "user", "content": query}]
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return

        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = run_bash(block.input["command"])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

不到 30 行, 这就是整个 Agent。后面 11 个章节都在这个循环上叠加机制 -- 循环本身始终不变。

变更内容 Copy Chapter

试一试 Copy Chapter

cd learn-claude-code
python agents/s01_agent_loop.py

试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):

1. Create a file called hello.py that prints "Hello, World!"
2. List all Python files in this directory
3. What is the current git branch?
4. Create a directory called test_output and write 3 files in it

Zh S02 Tool Use (`docs/zh/s02-tool-use.md`) Copy Section

s02: Tool Use (工具使用)

s01 > [ s02 ] s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"加一个工具, 只加一个 handler" -- 循环不用动, 新工具注册进 dispatch map 就行。

Harness 层: 工具分发 -- 扩展模型能触达的边界。

问题 Copy Chapter

只有 bash 时, 所有操作都走 shell。cat 截断不可预测, sed 遇到特殊字符就崩, 每次 bash 调用都是不受约束的安全面。专用工具 (read_file, write_file) 可以在工具层面做路径沙箱。

关键洞察: 加工具不需要改循环。

解决方案 Copy Chapter

+--------+      +-------+      +------------------+
|  User  | ---> |  LLM  | ---> | Tool Dispatch    |
| prompt |      |       |      | {                |
+--------+      +---+---+      |   bash: run_bash |
                    ^           |   read: run_read |
                    |           |   write: run_wr  |
                    +-----------+   edit: run_edit |
                    tool_result | }                |
                                +------------------+

The dispatch map is a dict: {tool_name: handler_function}.
One lookup replaces any if/elif chain.

工作原理 Copy Chapter

1. 每个工具有一个处理函数。路径沙箱防止逃逸工作区。

def safe_path(p: str) -> Path:
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

def run_read(path: str, limit: int = None) -> str:
    text = safe_path(path).read_text(encoding="utf-8")
    lines = text.splitlines()
    if limit and limit < len(lines):
        lines = lines[:limit]
    return "\n".join(lines)[:50000]

2. dispatch map 将工具名映射到处理函数。

TOOL_HANDLERS = {
    "bash":       lambda **kw: run_bash(kw["command"]),
    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),
    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"],
                                        kw["new_text"]),
}

3. 循环中按名称查找处理函数。循环体本身与 s01 完全一致。

for block in response.content:
    if block.type == "tool_use":
        handler = TOOL_HANDLERS.get(block.name)
        output = handler(**block.input) if handler \
            else f"Unknown tool: {block.name}"
        results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })

加工具 = 加 handler + 加 schema。循环永远不变。

相对 s01 的变更 Copy Chapter

试一试 Copy Chapter

cd learn-claude-code
python agents/s02_tool_use.py

试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):

1. Read the file requirements.txt
2. Create a file called greet.py with a greet(name) function
3. Edit greet.py to add a docstring to the function
4. Read greet.py to verify the edit worked

Zh S03 Todo Write (`docs/zh/s03-todo-write.md`) Copy Section

s03: TodoWrite (待办写入)

s01 > s02 > [ s03 ] s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

"没有计划的 agent 走哪算哪" -- 先列步骤再动手, 完成率翻倍。

Harness 层: 规划 -- 让模型不偏航, 但不替它画航线。

问题 Copy Chapter

多步任务中, 模型会丢失进度 -- 重复做过的事、跳步、跑偏。对话越长越严重: 工具结果不断填满上下文, 系统提示的影响力逐渐被稀释。一个 10 步重构可能做完 1-3 步就开始即兴发挥, 因为 4-10 步已经被挤出注意力了。

解决方案 Copy Chapter

+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> | Tools   |
| prompt |      |       |      | + todo  |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                          |
              +-----------+-----------+
              | TodoManager state     |
              | [ ] task A            |
              | [>] task B  <- doing  |
              | [x] task C            |
              +-----------	------------+
                          |
              if rounds_since_todo >= 3:
                inject <reminder> into tool_result

工作原理 Copy Chapter

1. TodoManager 存储带状态的项目。同一时间只允许一个 in_progress。

class TodoManager:
    def update(self, items: list) -> str:
        validated, in_progress_count = [], 0
        for item in items:
            status = item.get("status", "pending")
            if status == "in_progress":
                in_progress_count += 1
            validated.append({"id": item["id"], "text": item["text"],
                              "status": status})
        if in_progress_count > 1:
            raise ValueError("Only one task can be in_progress")
        self.items = validated
        return self.render()

2. todo 工具和其他工具一样加入 dispatch map。

TOOL_HANDLERS = {
    # ...base tools...
    "todo": lambda **kw: TODO.update(kw["items"]),
}

3. nag reminder: 模型连续 3 轮以上不调用 todo 时注入提醒。

if rounds_since_todo >= 3 and messages:
    last = messages[-1]
    if last["role"] == "user" and isinstance(last.get("content"), list):
        last["content"].insert(0, {
            "type": "text",
            "text": "<reminder>Update your todos.</reminder>",
        })

"同时只能有一个 in_progress" 强制顺序聚焦。nag reminder 制造问责压力 -- 你不更新计划, 系统就追着你问。

相对 s02 的变更 Copy Chapter

试一试 Copy Chapter