from pathlib import Path
from agent_harness import Config, SubAgentParams, SubAgentTool, ToolInvocation
from agent_harness.tools.task import BUILTIN_ROLES
parent_config = Config(cwd=Path.cwd(), max_turns=8)
task_tool = SubAgentTool(parent_config)
role_rows = [
{
"role": name,
"allowed_tools": allowed_tools,
"prompt_excerpt": prompt[:58] + "...",
}
for name, (prompt, allowed_tools) in BUILTIN_ROLES.items()
]
print("delegation schema:", task_tool.to_openai_schema()["parameters"])
for row in role_rows:
print(row)
assert set(BUILTIN_ROLES) == {
"codebase_investigator",
"code_reviewer",
"test_writer",
"code_fixer",
"doc_writer",
}
assert task_tool.to_openai_schema()["name"] == "run_sub_agent"
assert any(row["role"] == "code_reviewer" for row in role_rows)Sub-Agents and Patterns: Delegation Without Context Collapse
agents
orchestration
reliability
Chapter 09 gave one agent loop observable lifecycle hooks. This chapter adds a second loop without pretending that two conversations make one reliable system. The parent agent becomes an orchestrator: it writes a bounded task packet, selects a worker role, and decides how evidence returns to the parent. The worker starts with fresh conversation state, a role-specific prompt, and an optionally restricted tool set, exactly as projects/agent-harness/ implements SubAgentTool.
The experiments stay offline. A fake child model exercises the real delegation boundary, while deterministic fixtures make context contamination, fan-out, fan-in, and review behavior visible without an API key. Continue the agent loop, context management, and permissions chapters before treating delegation as an escape hatch.
Delegation boundary
A parent can share its entire transcript with a child, or it can send a deliberately small task packet. The first choice is convenient and fragile: irrelevant history, half-finished hypotheses, and untrusted tool output become part of the child’s operating context. The second choice makes the handoff explicit. It also creates a new specification surface that can be tested.
A useful packet names four things:
- the question and repository scope;
- the role the child is allowed to play;
- the evidence or artifact it must return; and
- the budget and stopping condition.
SubAgentParams carries that contract. SubAgentTool then constructs a fresh Session, replaces its system message with the role prompt, and applies the role’s tool restriction. The parent does not receive hidden child state; it receives a formatted result and metadata. That boundary is the mechanism, not an implementation detail.
The first inspection is intentionally schema-only. It asks what a parent can tell the model about delegation and what built-in roles permit. No client is constructed and no model request is made.
The schema exposes a single action, but the role changes the action’s authority. An investigator can inspect with read_file, glob, grep, and shell; a reviewer has no write tool in its default list. This is least privilege at the delegation boundary. The restriction is not a proof that a worker is safe, because a permitted shell command can still be risky, but it reduces the space of accidental actions and gives the parent a concrete invariant to check.
Child configuration
Build one packet for a codebase investigator and inspect the child configuration that the tool would derive. Private helper methods are used here because the course is examining the existing implementation’s contract; the public execution path is exercised in the next experiment.
from agent_harness import ApprovalPolicy
investigation = SubAgentParams(
task=(
"Inspect the package's MCP bridge. Return the files examined, the schema "
"translation invariant, and one limitation. Do not edit files."
),
role="codebase_investigator",
max_turns=4,
)
role_prompt, allowed_tools = task_tool._resolve_role(investigation)
child_config = task_tool._build_child_config(
investigation,
role_prompt,
allowed_tools,
)
print(
{
"child_cwd": str(child_config.cwd),
"child_turn_budget": child_config.max_turns,
"child_approval": child_config.approval.value,
"allowed_tools": child_config.allowed_tools,
"inherits_parent_messages": False,
}
)
assert child_config.cwd == parent_config.cwd
assert child_config.max_turns == 4
assert child_config.approval == ApprovalPolicy.YOLO
assert child_config.allowed_tools == ["read_file", "list_dir", "glob", "grep", "shell"]
assert child_config.developer_instructions == role_promptThe child keeps the working directory and model configuration, but not the parent’s messages, turn count, or accumulated usage. The YOLO setting is an internal child override in this implementation: it prevents a child from stalling on an approval prompt while the parent remains responsible for whether delegation itself is permitted. If a deployment needs approval inside the child, that policy must be restored deliberately rather than assumed from the parent.
Delegation tests
SubAgentTool.execute normally constructs Session and Agent, then listens for the child’s TEXT_COMPLETE event. Patch only those two construction points with an offline child. This keeps the production tool’s parameter validation, role resolution, child configuration, output formatting, and metadata intact while removing the network boundary.
import asyncio
from pathlib import Path
from unittest.mock import patch
from agent_harness.events import AgentEvent, TokenUsage
captured: dict[str, object] = {}
class OfflineChildSession:
def __init__(self, config):
captured["config"] = config
self.messages = [{"role": "system", "content": "placeholder"}]
self.turn_count = 2
self.total_usage = TokenUsage(prompt_tokens=11, completion_tokens=5, total_tokens=16)
class OfflineChildAgent:
def __init__(self, config, session):
captured["session"] = session
async def run(self, task):
captured["task"] = task
yield AgentEvent.text_complete(
"Inspected bridge.py: schema translation is pure; transport is deferred."
)
async def run_offline_delegation():
invocation = ToolInvocation(
params=investigation.model_dump(),
cwd=Path.cwd(),
)
with (
patch("agent_harness.agent.Agent", OfflineChildAgent),
patch("agent_harness.session.Session", OfflineChildSession),
):
return await task_tool.execute(invocation)
result = asyncio.run(run_offline_delegation())
print(result.output)
print("metadata:", result.metadata)
assert result.success
assert "schema translation is pure" in result.output
assert captured["task"] == investigation.task
assert captured["config"].allowed_tools == ["read_file", "list_dir", "glob", "grep", "shell"]The same delegation primitive supports several control-flow patterns. They are not interchangeable:
| Pattern | When it helps | New failure to measure |
|---|---|---|
| Sequential pipeline | each step depends on the previous artifact | an early mistake becomes shared context |
| Parallel fan-out and fan-in | independent questions can run concurrently | the join can drop, duplicate, or over-trust findings |
| Reviewer or critic | a second pass can challenge a proposed answer | agreement can be mistaken for evidence |
| Supervisor and workers | a coordinator can route heterogeneous tasks | the supervisor becomes a single point of specification failure |
Choose a pattern from the dependency graph, not from a preference for more agents. If two workers need the same mutable state, isolate their reads and make the write or merge explicit. If their questions are independent, parallelism can reduce latency without sharing transcripts.
Fan-out
This fixture has two independent questions. The workers receive only their own packets, and asyncio.gather joins results in packet order. The model is absent on purpose: the experiment isolates orchestration order and the shape of the handoff.
from dataclasses import dataclass
@dataclass(frozen=True)
class Finding:
worker: str
topic: str
position: str
evidence: str
async def research_worker(packet: dict[str, str]) -> Finding:
await asyncio.sleep(0)
return Finding(
worker=packet["worker"],
topic=packet["topic"],
position=packet["position"],
evidence=packet["evidence"],
)
packets = [
{
"worker": "policy",
"topic": "approval",
"position": "manual approval",
"evidence": "policy.md:10",
},
{
"worker": "tests",
"topic": "regression coverage",
"position": "path boundary is covered",
"evidence": "tests/test_tools.py:99",
},
]
async def fan_out(packets: list[dict[str, str]]) -> list[Finding]:
return list(await asyncio.gather(*(research_worker(packet) for packet in packets)))
parallel_findings = asyncio.run(fan_out(packets))
print(parallel_findings)
assert [finding.worker for finding in parallel_findings] == ["policy", "tests"]
assert all(set(packet) == {"worker", "topic", "position", "evidence"} for packet in packets)The join preserves input order even though the worker tasks are concurrent. That makes a result reproducible, but it does not make the findings correct. Production fan-out still needs a concurrency bound, cancellation behavior, per-worker timeouts, and a join rule that records missing or conflicting results. Those are reliability controls around the pattern, not properties supplied by gather.
Fan-in
A naive fan-in concatenates prose and asks the parent to infer whether two workers disagree. A safer fan-in groups findings by topic and makes multiple positions explicit. The small reconciler below does not decide which source is true; it makes the contradiction observable so a reviewer or a later tool can resolve it.
def reconcile(findings: list[Finding]) -> dict[str, object]:
grouped: dict[str, list[Finding]] = {}
for finding in findings:
grouped.setdefault(finding.topic, []).append(finding)
conflicts = {
topic: sorted({finding.position for finding in topic_findings})
for topic, topic_findings in grouped.items()
if len({finding.position for finding in topic_findings}) > 1
}
return {
"topics": {
topic: [finding.__dict__ for finding in topic_findings]
for topic, topic_findings in grouped.items()
},
"conflicts": conflicts,
"review_required": bool(conflicts),
}
conflicting_findings = [
Finding("policy", "approval", "manual approval", "policy.md:10"),
Finding("vendor", "approval", "automatic approval", "vendor.md:4"),
Finding("tests", "regression coverage", "path boundary is covered", "tests/test_tools.py:99"),
]
reconciled = reconcile(conflicting_findings)
print(reconciled)
assert reconciled["conflicts"] == {"approval": ["automatic approval", "manual approval"]}
assert reconciled["review_required"] is True
assert len(reconciled["topics"]["approval"]) == 2The output keeps both sources, labels the disputed topic, and raises the review requirement. This is a small instance of a broader rule: a sub-agent result is an observation with provenance, not an instruction that outranks the parent. The parent can now route only the disputed topic to a critic instead of replaying the entire investigation.
Handoff invariants
A useful worker response is short enough to fit the parent context but structured enough to audit. The project returns text plus metadata, so the application-level contract still needs to say what the text must contain. The validator below is deliberately boring: status, answer, evidence, and limitations are all required, and evidence cannot be replaced by confidence language.
def valid_handoff(handoff: dict[str, object]) -> bool:
required = {"status", "answer", "evidence", "limitations"}
if not required.issubset(handoff):
return False
if handoff["status"] not in {"complete", "blocked", "uncertain"}:
return False
if not isinstance(handoff["answer"], str) or not handoff["answer"].strip():
return False
if not isinstance(handoff["evidence"], list) or not handoff["evidence"]:
return False
return isinstance(handoff["limitations"], list)
handoff = {
"status": "complete",
"answer": "Manual approval is required.",
"evidence": ["policy.md:10"],
"limitations": ["The vendor documentation was not authoritative."],
}
missing_evidence = {**handoff, "evidence": []}
print({"valid": valid_handoff(handoff), "missing_evidence": valid_handoff(missing_evidence)})
assert valid_handoff(handoff)
assert not valid_handoff(missing_evidence)
review_params = SubAgentParams(
task="Review only the two approval findings and cite the disagreement.",
role="code_reviewer",
max_turns=3,
)
_, reviewer_tools = task_tool._resolve_role(review_params)
assert reviewer_tools == ["read_file", "list_dir", "glob", "grep"]The handoff validator is not a truth detector. It guards a weaker but valuable contract: the parent can distinguish an answer from its support and its known limits. That distinction prevents a fluent child response from silently becoming an authoritative parent instruction.
Test matrix
The final offline check collects the chapter’s claims. It checks the public alias retained for compatibility, the exposed schema, role least privilege, child isolation, deterministic fan-in, and the handoff boundary. Each assertion names a failure that would otherwise be easy to hide behind a successful-looking final answer.
from agent_harness import TaskTool
checks = {
"public_alias": TaskTool is SubAgentTool,
"delegation_schema": task_tool.to_openai_schema()["name"] == "run_sub_agent",
"investigator_scope": set(child_config.allowed_tools) == {
"read_file",
"list_dir",
"glob",
"grep",
"shell",
},
"reviewer_scope": set(reviewer_tools).isdisjoint({"write_file", "edit"}),
"isolated_context_fixture": experiment["isolated"] != experiment["shared"],
"fan_in_surfaces_conflict": reconciled["review_required"],
"handoff_requires_evidence": not valid_handoff(missing_evidence),
}
print(checks)
assert all(checks.values())The reliability connection is the point of the chapter. Delegation can improve coverage and latency, but it introduces specification error at the task boundary, proxy optimization at the join, and correction failure when the parent treats a worker’s fluent text as verified truth. The controls are concrete: least-privilege role tools, bounded child turns, independent packets, provenance-preserving fan-in, reviewer routing for disagreement, and a machine-checkable handoff.
For a broader discussion of agent patterns, compare Anthropic’s guide to building effective agents with the implementation here. The next chapter applies the same boundary thinking to the Model Context Protocol: an external server is another worker-like authority, so namespacing, schema validation, and error visibility matter before transport details do.