Code Agent Dissection (08): When a Task Is Too Complex, How Do You Delegate to a Sub-Agent?

Main agent calls Task tool
    ↓
TaskTool.run()                     validates params, constructs TaskRequest
    ↓
_DeferredSubagentLauncher.launch() proxy layer, lazily initializes the real launcher
    ↓
SubagentLauncher.launch()          the real entry point (subagents.py)
    ├─ 1. look up RuntimeProfile (tool allowlist, step limit, token budget, model)
    ├─ 2. _select_llm(): prefer LIGHT_LLM, fall back to main model if not configured
    ├─ 3. _create_child_trace(): create independent trace log
    ├─ 4. emit subagent_requested/started events to parent trace
    ├─ 5. _SubagentRuntimeHost: create isolated sandbox (fresh history/context/tools)
    ├─ 6. _render_request(): combine task description + structured context into prompt
    ├─ 7. RuntimeRunner(host).run(prompt) ← full ReAct loop
    ├─ 8. _child_metrics(): extract terminal_reason/tool_usage/token_usage from trace events
    └─ 9. ExploreResult.from_json(): parse and validate sub-agent JSON output
    ↓
SubagentLaunchResult → TaskTool packages into standard envelope; main agent receives summary and continues

# task.py                          # subagents.py
class TaskRequest:                  class SubagentRequest:
    profile_name: str                   profile_name: str
    task: str                           task: str
    model_choice: str | None            model_choice: str | None
    structured_context: dict            structured_context: dict
    parent_session_id: str | None       parent_session_id: str | None
    parent_run_id: str | None           parent_run_id: str | None

# tools/builtin/task.py — TaskTool.run() (simplified)
def run(self, parameters):
    description = parameters.get("description")  # short description for main agent (a label)
    prompt = parameters.get("prompt")            # self-contained exploration instruction (full task for sub-agent)
    profile = parameters.get("subagent_type")    # currently only "explore" is supported
    model = parameters.get("model", "light")     # "light" (lightweight model) or "main"
 
    # After parameter validation, construct TaskRequest and delegate
    launched = self._launcher.launch(
        TaskRequest(
            profile_name="explore",   # ← this string determines the path taken
            task=f"{description}\n\n{prompt}",
            model_choice=model,
        )
    )

# runtime/host.py — _initialize_runtime_components()
self.tool_registry.register_tool(
    TaskTool(
        project_root=...,
        launcher=self._DeferredSubagentLauncher(self._get_subagent_launcher),
        #                ↑ this is the real identity of self._launcher
    )
)

TaskTool.run()
  self._launcher.launch(TaskRequest(profile_name="explore", ...))
    ↓  [host.py _DeferredSubagentLauncher.launch]
    self._get_launcher()          ← lazy init, only creates SubagentLauncher on first call
      ↓  [host.py _get_subagent_launcher]
      create_subagent_launcher(host)   ← factory.py, passes main agent's llm/registry in
        → SubagentLauncher(main_llm, tool_registry, ...)
    SubagentLauncher.launch(request)   ← subagents.py, the real implementation
      ↓
      profile = RUNTIME_PROFILES.get(request.profile_name)
      #         ↑ "explore" → EXPLORE_PROFILE (tool allowlist, step limit, token budget all here)
      ↓
      create sandbox, run ReAct loop, parse results

# runtime/subagents.py
EXPLORE_PROFILE = RuntimeProfile(
    name="explore",
    system_prompt=EXPLORE_SYSTEM_PROMPT,         # fixed system prompt requiring JSON return
    tool_allowlist={"Read", "Grep", "Glob"},     # read-only tools; no file writes, no command execution
    max_steps=12,                                # max 12 steps to prevent infinite loops
    context_token_budget=16_000,                 # single context window limit (main agent is 128k)
    total_token_budget=32_000,                   # cumulative token limit; forced termination if exceeded
    model_choice="light",                        # default to lightweight model for cost savings
    result_contract="ExploreResult",             # must return JSON matching this contract
)

if self.recursive_subagents or "Task" in self.tool_allowlist:
    raise ValueError("formal subagent profiles cannot recurse")
forbidden = {"Edit", "Bash"}
if forbidden & self.tool_allowlist:
    raise ValueError("formal subagent profiles must be strictly read-only")

# subagents.py — SubagentLauncher.launch()
llm, model_choice = self._select_llm(requested_model)
# _select_llm: prefer light_llm (created from LIGHT_LLM_* env vars)
# Falls back to main agent's main_llm if LIGHT_LLM_MODEL_ID is not set
 
child_trace = self._create_child_trace()
# Independent trace JSONL file; session_id starts with "child-"
# Separated from main agent trace, but launch() emits parent-child events
# linked through parent_session_id

# 1. config: override key fields with profile budgets
self.config = Config.from_env().model_copy(update={
    "context_window": profile.context_token_budget,  # explore=16000, far less than main agent's 128k
})
self.max_steps = profile.max_steps          # explore=12, hard step limit
self.max_total_tokens = profile.total_token_budget  # explore=32000, forced termination if exceeded
 
# 2. registry: only contains allowlisted tools (build_registry filters)
# Of the main agent's tools, only Read/Grep/Glob enter the sub-agent's registry
 
# 3. ContextBuilder: system prompt fixed to profile.system_prompt (requires JSON return)
#    No MCP tool prompts or Skills loaded; non-allowlisted tool schemas not exposed
 
# 4. Context compression: uses _summarize_child_messages (pure truncation, no LLM)
#    Reason: sub-agent budget is small; LLM compression is too expensive; simple truncation suffices
 
# 5. Double-layer permission protection:
permission_context = PermissionContext(runtime_mode="readonly_subagent")
# RiskClassifier directly DENYs Edit/Bash/Task in this mode
# Even if these tools accidentally appear in the registry, they can't execute
# — allowlist + permission gate provide double protection
 
# 6. completion_verifier: _StructuredResultCompletionVerifier (sub-agent specific)
#    Checks if output matches result_contract JSON format; triggers FAIL feedback if not,
#    prompting sub-agent to re-output

# _render_request: combines task text and structured context into a prompt
# Output format: task text + "\n\nStructured context:\n" + JSON
# This is the only context the sub-agent sees — it can't see the main agent's history
prompt = _render_request(request)
 
# Same RuntimeRunner as main agent, runs full ReAct loop
# Sub-agent's system prompt requires it to ultimately output a JSON object
raw_result = RuntimeRunner(host).run(prompt)

# _child_metrics: iterates child_trace.events, aggregates three types of info:
# terminal_reason: how the sub-agent ended (completed/max_steps/token_budget, etc.)
# tool_usage: how many times each tool was called
# token_usage: cumulative token count
terminal_reason, tool_usage, token_usage = _child_metrics(child_trace.events)
 
# Sub-agent must end with completed/completed_unverified; any other reason is failure
if terminal_reason not in {"completed", "completed_unverified"}:
    raise ValueError(f"child terminal reason: {terminal_reason}")
 
# ExploreResult.from_json strict validation:
# - status can only be completed/partial
# - summary cannot be empty
# - if Markdown fence (```) is present, raises exception immediately
# Invalid → launch() catches exception, returns status=FAILED
structured = ExploreResult.from_json(raw_result, tool_usage=tool_usage, ...)

You are an Explore Agent.
Inspect the repository with read-only tools and return exactly one JSON object:
{"status":"completed|partial","summary":"...","findings":["..."],
"evidence":["relative/path.py:line"],"unresolved_questions":["..."]}.
Do not use markdown fences.

raw_result = RuntimeRunner(host).run(prompt)
 
structured = ExploreResult.from_json(
    raw_result,
    tool_usage=tool_usage,
    terminal_reason=terminal_reason,
)

def _select_llm(self, requested_model):
    if requested_model == "light":
        if self.light_llm is None:
            self.light_llm = _create_light_llm()   # created from LIGHT_LLM_* env vars
        if self.light_llm is not None:
            return self.light_llm, "light"
    return self.main_llm, "main"   # fall back to main model if no lightweight configured

return self.success_result(
    data={
        "status": "completed",
        "profile": "explore",
        "result": {
            "summary": "Found 3 authentication-related code locations...",
            "findings": ["auth/login.py:45 — JWT verification", ...],
            "evidence": ["auth/login.py:45", ...],
        },
    },
    text=result.summary,   # ← summary for the model, goes directly into observation
    extra_stats={
        "tool_calls": 8,
        "token_usage": 12000,
        "model": "light",
    },
)

git clone https://github.com/chendongqi/MyCodeAgent
cd MyCodeAgent
cp .env.example .env   # fill in your LLM API key
uv sync
uv run python main.py