Parallel Subagent Orchestration

Parallel Subagent Orchestration | CodeSignal Learn

Introduction

Welcome back to Codex Subagents & Multi-Agent Orchestration! You have completed the first lesson where we explored subagent isolation and output contracts. Those foundational concepts enable us to run reliable, contained AI operations with predictable results. In this second lesson, we are scaling up: Parallel Subagents + Deterministic Aggregation. Instead of running a single isolated subagent, we will orchestrate multiple subagents working concurrently on different parts of a codebase. Each subagent will analyze a specific scope, and we will combine their results into a unified report. By the end of this lesson, you will understand how to build supervisor scripts that coordinate parallel AI work while maintaining full observability and control.

Why Parallel Execution Matters

When analyzing a large codebase, running subagents sequentially can be time-consuming. If each subagent takes 30 seconds, analyzing three different packages would require 90 seconds of sequential execution. With parallel execution, all three run simultaneously, reducing total time to roughly 30 seconds. Beyond speed, parallelism offers other benefits: Scope isolation : Each subagent stays focused on its assigned area without contamination from other scopes. Independent failures : If one subagent fails, others can still succeed and return useful results. Resource utilization : Modern systems have multiple cores; parallel execution leverages this capacity. The challenge lies in coordination: launching multiple processes, collecting their outputs, and merging results into a coherent whole. This is where deterministic aggregation becomes essential.

The Architecture

Our approach involves three key components working together. First, a runner function executes individual subagents, capturing their output and metadata. Second, a parallel executor launches multiple runners concurrently and collects results as they complete. Third, an aggregator function combines these individual results into a unified report. This architecture maintains the isolation principles from our previous lesson while adding coordination. Each subagent still operates within strict boundaries with a JSON contract, but now we are managing multiple subagents as a cohesive workflow. The supervisor script becomes the orchestrator, deciding what tasks to run, monitoring their execution, and synthesizing their findings.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Running a Single Subagent

Let us start by examining how we execute a single subagent programmatically: Pythondef run_agent(scope: str, task: str) -> Dict[str, Any]: prompt = f""" Analyze ONLY files under {scope}. Task: {task} Return STRICT JSON: {{ "status": "success|partial|failed", "summary": "string", "findings": ["string"], "files_read": ["string"], "files_modified": ["string"] }} """.strip()def run_agent(scope: str, task: str) -> Dict[str, Any]: prompt = f""" Analyze ONLY files under {scope}. Task: {task} Return STRICT JSON: {{ "status": "success|partial|failed", "summary": "string", "findings": ["string"], "files_read": ["string"], "files_modified": ["string"] }} """.strip() The run_agent function takes two parameters: a scope (directory path) and a task (what to analyze). It constructs a prompt dynamically, embedding these values into the instruction template. Notice the familiar contract from our previous lesson; we are enforcing the same strict JSON output format. The .strip() method removes leading and trailing whitespace, ensuring clean input to the codex command.

Executing the Subagent Process

With our prompt ready, we launch the subagent using Python's subprocess module: Python proc = subprocess.run(["codex", "-a", "never", "-s", "workspace-write", "exec", "--json", prompt], capture_output=True, text=True,) proc = subprocess.run(["codex", "-a", "never", "-s", "workspace-write", "exec", "--json", prompt], capture_output=True, text=True,) This runs the subagent in non-interactive mode and emits structured CLI events as JSONL on stdout. The capture_output=True argument captures both stdout and stderr, while text=True returns string output. The completed process object proc includes returncode, stdout, and stderr for supervisor logic.

Capturing Timing Metrics

Before launching the subagent, we record the start time; after completion, we calculate runtime_ms : Python def run_agent(scope: str, task: str) -> Dict[str, Any]: start = time.time() # ... prompt construction and subprocess execution ... runtime_ms = int((time.time() - start) * 1000) def run_agent(scope: str, task: str) -> Dict[str, Any]: start = time.time() # ... prompt construction and subprocess execution ... runtime_ms = int((time.time() - start) * 1000) This simple pattern provides crucial observability. By measuring how long each subagent takes, we can identify performance bottlenecks, compare different task types, and track improvements over time. The conversion to milliseconds (* 1000) and integer casting give us precise, easy-to-compare metrics. We will attach this runtime_ms value to each subagent's result for later analysis.

Extracting the JSON Contract from CLI Events

Because --json returns event lines (not a single JSON object), the supervisor should extract the contract payload from the completed agent message event. A reusable helper keeps this logic consistent across tasks: Pythondef parse_contract_from_jsonl(stdout: str) -> Dict[str, Any]: for raw_line in stdout.splitlines(): line = raw_line.strip() if not line: continue event = json.loads(line) if event.get("type") != "item.completed": continue item = event.get("item") or {} if item.get("type") != "agent_message": continue text = item.get("text") if not isinstance(text, str) or not text.strip(): continue data = json.loads(text) if not isinstance(data, dict): raise ValueError("Contract JSON is not an object") return data raise ValueError("No agent_message JSON contract found in JSONL stream")def parse_contract_from_jsonl(stdout: str) -> Dict[str, Any]: for raw_line in stdout.splitlines(): line = raw_line.strip() if not line: continue event = json.loads(line) if event.get("type") != "item.completed": continue item = event.get("item") or {} if item.get("type") != "agent_message": continue text = item.get("text") if not isinstance(text, str) or not text.strip(): continue data = json.loads(text) if not isinstance(data, dict): raise ValueError("Contract JSON is not an object") return data raise ValueError("No agent_message JSON contract found in JSONL stream") This helper parses each JSONL event, selects the agent_message, and then parses item.text into the final contract object. If the expected payload is missing, it raises a clear error that the caller can handle.

Handling Invalid Output

After the subprocess completes, capture process metadata first, then parse with the helper: Python runtime_ms = int((time.time() - start) * 1000) exit_code = proc.returncode stderr = (proc.stderr or "").strip() stdout = (proc.stdout or "").strip() runtime_ms = int((time.time() - start) * 1000) exit_code = proc.returncode stderr = (proc.stderr or "").strip() stdout = (proc.stdout or "").strip() Then parse the contract. If parsing fails, return a standardized failure payload with diagnostics: Python try: data = parse_contract_from_jsonl(stdout) except Exception as e: return { "status": "failed", "summary": f"Invalid JSON output (exit={exit_code})", "findings": [], "files_read": [], "files_modified": [], "runtime_ms": runtime_ms, "scope": scope, # diagnostics captured by the supervisor: "reason": "invalid_json", "error": str(e), "exit_code": exit_code, "stderr": stderr, "raw_stdout": stdout, } try: data = parse_contract_from_jsonl(stdout) except Exception as e: return { "status": "failed", "summary": f"Invalid JSON output (exit={exit_code})", "findings": [], "files_read": [], "files_modified": [], "runtime_ms": runtime_ms, "scope": scope, # diagnostics captured by the supervisor: "reason": "invalid_json", "error": str(e), "exit_code": exit_code, "stderr": stderr, "raw_stdout": stdout, } This pattern ensures one failing subagent does not crash the entire orchestration, and it produces a failure record that can be inspected later without re-running.

Enriching the Result

When contract parsing succeeds, enrich the subagent's output with supervisor-level metadata: Python data["runtime_ms"] = runtime_ms data["scope"] = scope data["exit_code"] = exit_code data["runtime_ms"] = runtime_ms data["scope"] = scope data["exit_code"] = exit_code Sometimes a process can exit with a non-zero code while still producing parseable contract JSON. Recording that as a reason makes those cases visible in the final artifact: Python if exit_code != 0 and not data.get("reason"): data["reason"] = "nonzero_exit" data["error"] = stderr return data if exit_code != 0 and not data.get("reason"): data["reason"] = "nonzero_exit" data["error"] = stderr return data

Launching Parallel Agents

Now we reach the core of parallel execution. We define our jobs and use a thread pool to run them concurrently: Python if __name__ == "__main__": jobs = [("packages/math/", "Identify areas with weak unit test coverage; list 3 high-value missing tests."), ("packages/element/", "Identify performance-sensitive modules/functions; list 3 concrete hotspots."), ("excalidraw-app/", "Audit error handling patterns; list 3 improvements for robustness."),] results: List[Dict[str, Any]] = [] with ThreadPoolExecutor(max_workers=3) as pool: futures = [pool.submit(run_agent, scope, task) for scope, task in jobs] if __name__ == "__main__": jobs = [("packages/math/", "Identify areas with weak unit test coverage; list 3 high-value missing tests."), ("packages/element/", "Identify performance-sensitive modules/functions; list 3 concrete hotspots."), ("excalidraw-app/", "Audit error handling patterns; list 3 improvements for robustness."),] results: List[Dict[str, Any]] = [] with ThreadPoolExecutor(max_workers=3) as pool: futures = [pool.submit(run_agent, scope, task) for scope, task in jobs] The jobs list defines three distinct tasks, each analyzing a different directory with a specific objective. The ThreadPoolExecutor with max_workers=3 creates a pool capable of running three subagents simultaneously. For each job, pool.submit schedules the run_agent function to execute with the given scope and task, returning a Future object that will eventually contain the result.

Collecting Results

As subagents finish, collect results in completion order: Python for fut in as_completed(futures): results.append(fut.result()) for fut in as_completed(futures): results.append(fut.result()) The as_completed iterator yields futures as they finish, regardless of submission order. This means faster subagents contribute their results immediately rather than waiting for slower ones. The fut.result() call retrieves the actual return value from run_agent, which is our enriched dictionary containing status, findings, and metadata. We accumulate these in the results list for aggregation.

Aggregation Strategy

With all results collected, aggregation builds a deterministic, traceable report. Start with a consistent skeleton: Pythondef aggregate(results: List[Dict[str, Any]]) -> Dict[str, Any]: combined = { "status": "success", "summaries": [], "findings": [], "failures": [], } saw_partial = False failed_count = 0def aggregate(results: List[Dict[str, Any]]) -> Dict[str, Any]: combined = { "status": "success", "summaries": [], "findings": [], "failures": [], } saw_partial = False failed_count = 0 Process each result using a simple status lattice. Failed results are recorded in the failures list, while partial results still contribute summaries and findings but downgrade the overall status: Python for r in results: status = r.get("status") if status == "failed": failed_count += 1 combined["failures"].append({ "scope": r.get("scope"), "summary": r.get("summary"), "runtime_ms": r.get("runtime_ms"), "reason": r.get("reason"), "error": r.get("error"), "exit_code": r.get("exit_code"), }) else: if status == "partial": saw_partial = True combined["summaries"].append({ "scope": r.get("scope"), "summary": r.get("summary"), }) combined["findings"].extend([ {"scope": r.get("scope"), "finding": f} for f in r.get("findings", []) ]) for r in results: status = r.get("status") if status == "failed": failed_count += 1 combined["failures"].append({ "scope": r.get("scope"), "summary": r.get("summary"), "runtime_ms": r.get("runtime_ms"), "reason": r.get("reason"), "error": r.get("error"), "exit_code": r.get("exit_code"), }) else: if status == "partial": saw_partial = True combined["summaries"].append({ "scope": r.get("scope"), "summary": r.get("summary"), }) combined["findings"].extend([ {"scope": r.get("scope"), "finding": f} for f in r.get("findings", []) ]) Finally, compute the overall status deterministically: if everything failed, mark the run as "failed"; otherwise, any failed or partial result downgrades the run to "partial"; only all-success results keep the overall status at "success": Python if failed_count == len(results): combined["status"] = "failed" elif failed_count > 0 or saw_partial: combined["status"] = "partial" return combined if failed_count == len(results): combined["status"] = "failed" elif failed_count > 0 or saw_partial: combined["status"] = "partial" return combined This preserves useful output from partial runs while ensuring the top-level status accurately reflects degraded subagent outcomes.

Writing the Output

Finally, we serialize the aggregated results to a JSON file: Python out = aggregate(results) Path("artifacts").mkdir(exist_ok=True) Path("artifacts/parallel_summary.json").write_text(json.dumps(out, indent=2), encoding="utf-8") print("Wrote artifacts/parallel_summary.json") out = aggregate(results) Path("artifacts").mkdir(exist_ok=True) Path("artifacts/parallel_summary.json").write_text(json.dumps(out, indent=2), encoding="utf-8") print("Wrote artifacts/parallel_summary.json") The Path("artifacts").mkdir(exist_ok=True) ensures our output directory exists without failing if it is already there. We then write the aggregated data with indent=2 for human readability and specify utf-8 encoding for consistent text handling. This creates a persistent artifact that can be consumed by other tools, archived for auditing, or inspected manually.

Introduction

Why Parallel Execution Matters

The Architecture

Join the 1M+ learners on CodeSignal

Running a Single Subagent

Executing the Subagent Process

Capturing Timing Metrics

Extracting the JSON Contract from CLI Events

Handling Invalid Output

Enriching the Result

Launching Parallel Agents

Collecting Results

Aggregation Strategy

Writing the Output

Putting It All Together

Conclusion and Next Steps