Reliable Multiagent Orchestration

Reliable Multiagent Orchestration | CodeSignal Learn

Introduction

Welcome to the final lesson of Codex Subagents & Multi-Agent Orchestration! Over the past three lessons, we have established the foundations of reliable subagent automation: strict contracts for predictable output, parallel orchestration for concurrent work, and gated pipelines for safe code modifications. Each pattern addressed specific challenges in building production-ready automation. This capstone lesson brings these concepts together into a comprehensive, production-grade orchestration script. We will build a multi-agent quality audit system that runs parallel analyses across multiple package directories, implements robust retry logic, enforces failure thresholds, and generates both human-readable reports and machine-parsable summaries. By the end, you will understand how to combine reliability patterns with observability practices to create maintainable, debuggable automation workflows.

Principles of Reliable Multi-Agent Systems

When operating multiple subagents in production, two concerns become paramount: reliability and observability . Reliability means the system degrades gracefully when individual agents fail, rather than cascading into total failure. Observability means we can understand what happened during execution, debug failures, and audit agent behavior after the fact. Achieving reliability requires several strategies: timing operations to detect performance issues, retrying failed agents to handle transient errors, and enforcing thresholds so that widespread failures halt the pipeline rather than producing unreliable output. Observability demands comprehensive logging, structured artifacts for both humans and machines, and clear failure diagnostics. These principles shape every design decision in our capstone script. We will instrument agent execution with timing data, persist detailed logs for each agent, aggregate results into multiple report formats, and validate overall quality before declaring success.

Architecture of the Capstone Script

Our capstone orchestrator follows a specific workflow: Discover work scopes by listing package directories in a monorepo structure. Execute agents (one per scope), capturing timing, output, and errors. Retry once when an agent fails to produce valid output (to handle transient failures). Aggregate results into: a Markdown report for human review a JSON summary for programmatic processing Enforce quality gates by calculating the failure rate across all agents and halting when it exceeds a threshold. This architecture balances concurrency and speed (easy to parallelize) with observability (logs + artifacts) and reliability (retries + thresholds).

Join the 1M+ learners on CodeSignal

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

Discovering Package Directories

We begin by identifying which directories contain code that needs auditing. Sort the scopes for deterministic runs (important for CI reproducibility and stable artifact diffs): Python def list_package_dirs() -> List[str]: p = Path("packages") return sorted(str(x) + "/" for x in p.iterdir() if x.is_dir()) def list_package_dirs() -> List[str]: p = Path("packages") return sorted(str(x) + "/" for x in p.iterdir() if x.is_dir()) This function scans the packages/ directory and returns a sorted list of subdirectory paths, each with a trailing slash. The trailing slash convention creates unambiguous scope boundaries for agents: packages/common/ means the agent should analyze only files under that path.

Executing and Instrumenting Agents

The core execution function combines subprocess invocation with timing instrumentation: Python def run_agent(scope: str, prompt: str, log_path: Path) -> Dict[str, Any]: start = time.time() proc = subprocess.run(["codex", "-a", "never", "-s", "read-only", "exec", "--json", prompt], capture_output=True, text=True,) runtime_ms = int((time.time() - start) * 1000) def run_agent(scope: str, prompt: str, log_path: Path) -> Dict[str, Any]: start = time.time() proc = subprocess.run(["codex", "-a", "never", "-s", "read-only", "exec", "--json", prompt], capture_output=True, text=True,) runtime_ms = int((time.time() - start) * 1000) We capture the start time before launching the subagent, then measure the elapsed time in milliseconds after completion. This metric reveals performance outliers: if one agent takes significantly longer than others, it might indicate code complexity or an agent struggling with its task. We use -a never for non-interactive execution and -s read-only for a read-only sandbox. As in earlier units, -a never is an approval policy, not what makes the run read-only; the read-only guarantee here comes from the sandbox mode.

Capturing Agent Output and Errors

After execution completes, we persist comprehensive logs: Python log_path.write_text( f"scope={scope}\nexit={proc.returncode}\n--- stdout ---\n{proc.stdout}\n--- stderr ---\n{proc.stderr}\n", encoding="utf-8", ) log_path.write_text( f"scope={scope}\nexit={proc.returncode}\n--- stdout ---\n{proc.stdout}\n--- stderr ---\n{proc.stderr}\n", encoding="utf-8", ) Each log file captures the agent’s scope, exit code, standard output, and standard error. These logs serve multiple purposes: they help debug failures, provide audit trails for what each agent observed, and preserve error messages when contract extraction fails. Having persistent logs means that even if the orchestrator crashes, you can inspect partial results and understand which agents completed successfully.

Handling JSON Failures Gracefully

As in the previous lessons, exec --json returns a JSONL event stream on stdout, so we reuse the same helper to extract the final contract object from the completed agent_message event. We also keep the same defensive pattern: if extraction fails, return a well-formed failure result instead of crashing the whole orchestration. Production note: Another Codex option you may encounter is --output-schema. This flag lets you define the expected shape of the agent’s final response, which can make downstream parsing more reliable. It solves a different problem than --json: --json makes the CLI output machine-readable as structured JSONL events, while --output-schema constrains the final model payload itself. In production, you may use both together: --json for structured event transport and --output-schema for stronger response-shape guarantees. We still keep defensive parsing in this capstone because a reliable orchestrator should handle malformed, missing, or incomplete outputs gracefully even when stronger schema controls are available. If you choose to use it, --output-schema would be added to the codex exec invocation alongside --json, before the prompt argument. In this capstone, the new emphasis is on making every result carry orchestration metadata (runtime_ms, exit_code, and scope) so retries, reporting, and failure-threshold checks can work from a uniform record shape. Python try: data = parse_contract_from_jsonl(proc.stdout) except Exception: return { "status": "failed", "summary": "Invalid JSON", "findings": [], "files_read": [], "files_modified": [], "runtime_ms": runtime_ms, "exit_code": proc.returncode, "scope": scope, } try: data = parse_contract_from_jsonl(proc.stdout) except Exception: return { "status": "failed", "summary": "Invalid JSON", "findings": [], "files_read": [], "files_modified": [], "runtime_ms": runtime_ms, "exit_code": proc.returncode, "scope": scope, } When extraction succeeds, we keep the result shape consistent and attach the orchestrator-owned fields needed later by the report and summary writers: Python if isinstance(data, dict): data.setdefault("status", "partial") data.setdefault("summary", "") data.setdefault("findings", []) data.setdefault("files_read", []) data.setdefault("files_modified", []) data["runtime_ms"] = runtime_ms data["exit_code"] = proc.returncode data["scope"] = scope return data return { "status": "failed", "summary": "Non-object JSON", "findings": [], "files_read": [], "files_modified": [], "runtime_ms": runtime_ms, "exit_code": proc.returncode, "scope": scope, } if isinstance(data, dict): data.setdefault("status", "partial") data.setdefault("summary", "") data.setdefault("findings", []) data.setdefault("files_read", []) data.setdefault("files_modified", []) data["runtime_ms"] = runtime_ms data["exit_code"] = proc.returncode data["scope"] = scope return data return { "status": "failed", "summary": "Non-object JSON", "findings": [], "files_read": [], "files_modified": [], "runtime_ms": runtime_ms, "exit_code": proc.returncode, "scope": scope, }

Implementing Audit Tasks with Retries

The audit_scope function combines task definition with retry logic. First, we define a strict contract that matches earlier lessons, including the consistent shape for files_modified (["string"], not a literal []). For read-only audits, the value must be []: Pythondef audit_scope(scope: str, out_dir: Path) -> Dict[str, Any]: task = f""" Analyze ONLY files under {scope}. Task: Produce a brief quality audit: - 3 maintainability risks - 3 quick wins Return STRICT JSON: {{ "status": "success|partial|failed", "summary": "string", "findings": ["string"], "files_read": ["string"], "files_modified": ["string"] }} For this read-only audit, files_modified must be []. """.strip() stem = scope.strip('/').replace('/', '_') log_path = out_dir / f"agent_{stem}.log" result = run_agent(scope, task, log_path)def audit_scope(scope: str, out_dir: Path) -> Dict[str, Any]: task = f""" Analyze ONLY files under {scope}. Task: Produce a brief quality audit: - 3 maintainability risks - 3 quick wins Return STRICT JSON: {{ "status": "success|partial|failed", "summary": "string", "findings": ["string"], "files_read": ["string"], "files_modified": ["string"] }} For this read-only audit, files_modified must be []. """.strip() stem = scope.strip('/').replace('/', '_') log_path = out_dir / f"agent_{stem}.log" result = run_agent(scope, task, log_path) We generate a deterministic log filename by transforming the scope path into a safe filename format. That naming convention becomes important later when you correlate summary.json entries with individual log files. After the initial execution, we implement simple retry logic: Python if result["status"] == "failed": for i in range(RETRY_ON_FAIL): retry_log = out_dir / f"agent_{stem}_retry{i+1}.log" result = run_agent(scope, task, retry_log) if result["status"] != "failed": break return result if result["status"] == "failed": for i in range(RETRY_ON_FAIL): retry_log = out_dir / f"agent_{stem}_retry{i+1}.log" result = run_agent(scope, task, retry_log) if result["status"] != "failed": break return result If the first attempt fails, we retry up to RETRY_ON_FAIL times (set to 1 at the module level). The retry uses a different log filename so that we preserve both attempts, and numbered retry suffixes (_retry1.log, _retry2.log, ...) scale cleanly if you later increase the retry count. This is the key capstone reliability pattern: tolerate transient failures without losing evidence from the original run.

Generating Human-Readable Reports

After collecting results from all agents, we synthesize them into a Markdown report: Pythondef write_report(results: List[Dict[str, Any]], out_dir: Path) -> None: failures = [r for r in results if r["status"] == "failed"] lines = [] lines.append(f"# Multi-Agent Quality Report\n") lines.append(f"- Total agents: {len(results)}\n") lines.append(f"- Failed agents: {len(failures)}\n") lines.append("\n## Findings\n") for r in results: if r["status"] != "failed": lines.append(f"### {r['scope']}\n") lines.append(f"{r.get('summary','')}\n") for f in r.get("findings", []): lines.append(f"- {f}\n") lines.append("\n")def write_report(results: List[Dict[str, Any]], out_dir: Path) -> None: failures = [r for r in results if r["status"] == "failed"] lines = [] lines.append(f"# Multi-Agent Quality Report\n") lines.append(f"- Total agents: {len(results)}\n") lines.append(f"- Failed agents: {len(failures)}\n") lines.append("\n## Findings\n") for r in results: if r["status"] != "failed": lines.append(f"### {r['scope']}\n") lines.append(f"{r.get('summary','')}\n") for f in r.get("findings", []): lines.append(f"- {f}\n") lines.append("\n") We start by calculating summary statistics: the total agent count and the failure count. The main findings section iterates through successful results, creating a subsection for each scope. Each subsection includes the agent’s summary and a bulleted list of findings. The report concludes with a failures section for quick triage: Python if failures: lines.append("\n## Failures\n") for r in failures: lines.append(f"- {r['scope']}: {r.get('summary')}\n") (out_dir / "quality_report.md").write_text("".join(lines), encoding="utf-8") if failures: lines.append("\n## Failures\n") for r in failures: lines.append(f"- {r['scope']}: {r.get('summary')}\n") (out_dir / "quality_report.md").write_text("".join(lines), encoding="utf-8") By separating successful findings from failures, you make it easy to prioritize: fix the failures first to get complete coverage, then address the quality issues identified by successful agents.

Persisting Machine-Readable Summaries

Alongside the human-readable Markdown, we generate structured JSON for programmatic consumption: Python (out_dir / "summary.json").write_text( json.dumps({"results": results}, indent=2), encoding="utf-8", ) (out_dir / "summary.json").write_text( json.dumps({"results": results}, indent=2), encoding="utf-8", ) This JSON file contains the complete results array, preserving all fields from each agent: status, summary, findings, files_read, files_modified, timing metrics, exit_code, and scope. Downstream tools can parse this file to generate dashboards, track quality trends over time, or integrate with CI/CD systems.

Orchestrating the Complete Workflow

The main execution block ties everything together with proper artifact management: Python if __name__ == "__main__": ts = time.strftime("%Y%m%d-%H%M%S") out_dir = Path("artifacts") / ts out_dir.mkdir(parents=True, exist_ok=True) scopes = list_package_dirs() results = [audit_scope(s, out_dir) for s in scopes] write_report(results, out_dir) if __name__ == "__main__": ts = time.strftime("%Y%m%d-%H%M%S") out_dir = Path("artifacts") / ts out_dir.mkdir(parents=True, exist_ok=True) scopes = list_package_dirs() results = [audit_scope(s, out_dir) for s in scopes] write_report(results, out_dir) We create a timestamped output directory, ensuring each run produces independent artifacts that do not overwrite previous results. This design supports historical analysis: you can compare quality reports across time to track improvements or regressions. The list comprehension executes agents sequentially; you can replace it with a parallel pattern (as in Lesson 2) without changing the artifact or gating model.

Enforcing Failure Thresholds

After orchestration completes, we validate overall quality: Python failed = sum(1 for r in results if r["status"] == "failed") total = len(scopes) failure_rate = (failed / total) if total else 0.0 if scopes and failure_rate > FAILURE_THRESHOLD: raise SystemExit( f"Too many failed agents: {failed}/{total} (={failure_rate:.0%}) > {FAILURE_THRESHOLD:.0%}" ) print(f"Artifacts written to {out_dir}/") failed = sum(1 for r in results if r["status"] == "failed") total = len(scopes) failure_rate = (failed / total) if total else 0.0 if scopes and failure_rate > FAILURE_THRESHOLD: raise SystemExit( f"Too many failed agents: {failed}/{total} (={failure_rate:.0%}) > {FAILURE_THRESHOLD:.0%}" ) print(f"Artifacts written to {out_dir}/") We count how many agents failed, compute the failure rate, and halt if that rate exceeds our threshold (30% by default). The error message includes both the fraction and the computed percentage to make it obvious that the gate is ratio-based.

Introduction

Principles of Reliable Multi-Agent Systems

Architecture of the Capstone Script

Join the 1M+ learners on CodeSignal

Discovering Package Directories

Executing and Instrumenting Agents

Capturing Agent Output and Errors

Handling JSON Failures Gracefully

Implementing Audit Tasks with Retries

Generating Human-Readable Reports

Persisting Machine-Readable Summaries

Debug Playbook (How to Use summary.json + Logs Together)

Orchestrating the Complete Workflow

Enforcing Failure Thresholds

Conclusion and Next Steps