Agent Orchestration

Agent Orchestration

This is the heart of Maestro: how a single prompt becomes a routed, planned, executed, reviewed, and synthesized answer — durably, so a crash mid-task resumes instead of hanging. The pipeline lives in backend/app/agents/ and is driven by the durable engine in backend/app/services/task_engine.py.

The pipeline

Orchestrator  →  Main Agent  →  Subagent(s)  →  (optional) Reviewer  →  Synthesis
   ROUTE            EXECUTE ..................................              FINALIZE

The durable engine (task_engine.run → _walk) advances the task through three checkpointed steps: ROUTE → EXECUTE → FINALIZE. Each step's result is written to the task_checkpoints table so a resumed run replays completed steps instantly instead of re-doing LLM work.

Step 1 — ROUTE (Orchestrator)

app/agents/orchestrator.py exposes route_decision() / route() returning a RouteResult(domain, complexity).

• The orchestrator makes a schema-validated structured_call that returns RouteDecision(domain, reason, complexity).
• If the user explicitly picked a domain, classification is skipped and complexity defaults to "complex".
• The caller's routable custom agents are merged into the routing catalog as custom:{id} selectors (capped at ROUTING_CUSTOM_AGENTS_MAX = 10), so user-defined agents can actually receive routed tasks.
• Result is checkpointed. Then resolve_domain_info(user_id, domain) loads the DomainInfo (a built-in domain or a sandboxed custom:{id} agent), and ctx.token_budget is set from quota_service.resolve_task_token_budget.

complexity drives effort scaling downstream: simple → 1 team member and the reviewer is skipped; standard → up to 3; complex → up to 6.

Step 2 — EXECUTE (Main Agent)

app/agents/main_agent.py run(ctx, domain, prompt, reviewer_enabled, complexity) orchestrates the whole execution.

Planning (_assign → _plan)

The Main Agent LLM makes a structured_call returning a PlanResult, which is either:

• a clarifying question (triggers HITL, below), or
• a list of PlanAssignment[] briefing its team.

Important constraints, enforced in code:

• It only briefs its fixed domain team (DomainInfo.team). Unknown member ids are dropped.
• An empty plan falls back to role-scoped _fallback_assignments so the task never dies planless.
• depends_on is sanitized to earlier members only, so the dependency graph is acyclic by construction.
• Assignment count is capped by min(max_iterations, MAX_SUBTASKS_BY_COMPLEXITY[complexity], MAX_SUBTASKS) where MAX_SUBTASKS = 6.

Dependency-wave execution (_run_assignments)

Assignments run in waves by dependency depth. Independent members in the same wave run concurrently, bounded by asyncio.Semaphore(subagent_max_parallel) (default 3). Each completed teammate's output is injected into its dependents — full text, or a summary when the output is large (context compaction). Once budget.budget_exceeded trips, remaining members are skipped.

Per-member run (_run_with_review → subagent.run_subtask)

app/agents/subagent.py builds the subagent system prompt from its SubagentSpec (name / role / instructions / output_format) plus upstream teammate output plus RAG memory context, then runs the tool loop:

• _chat_with_tools — a provider-agnostic JSON directive loop: the model emits a ToolDirective as JSON, the runtime parses it (tools.parse_directive), runs the tool, and feeds results back.
• _native_tool_loop — used instead when capabilities.native_tools is true (e.g. OpenAI tool_calls).
• Tools available: web_search, data_fetch, code_execution, plus the always-present view_original_request. Bounded by per-tool usage caps and subagent_max_tool_calls (default 6).
• _compact_transcript trims long histories deterministically (no extra LLM call).

Reviewer (optional)

Skipped entirely when complexity == "simple". Otherwise app/agents/reviewer.py review():

1. Runs deterministic validators.validate first (cheap, no LLM) — e.g. nonempty_min_length, json_parses, code_blocks_present.
2. Runs a structured_call returning a ReviewVerdict.
3. Approves via weighted criteria: _weighted_approved against REVIEW_APPROVAL_THRESHOLD = 0.7. Any hard_fail criterion scored 0 rejects outright.

On rejection the subagent re-runs with retry_hints, up to max_review_iterations (default 3). reviewer_fail_mode (warn | approve | reject) decides what happens when the reviewer itself errors.

Synthesis (_synthesize)

Successful (member_name, output) pairs are merged by the SYNTHESIS_SYSTEM prompt and streamed to the client as AGENT_DELTA events. Failed subtasks become explicitly acknowledged "Known gaps" — the synthesis never fabricates over a missing result.

Step 3 — FINALIZE

_finalize_success sets the terminal status:

• completed — everything succeeded.
• completed_with_warnings — some subtasks failed but there is a usable partial result.
• failed — all subtasks failed.

It then persists the result + metadata to Mongo task_sessions, records usage (see Billing-and-Quota), and writes conversation memory to Qdrant (see RAG-and-Memory).

Human-in-the-loop (HITL)

If the plan returns a question and allow_questions is set:

1. ctx.ask_user persists a task_questions row and flips status to awaiting_answer.
2. The engine awaits the answer — via an in-process future, or the Redis control channel maestro:ctrl:{task_id} for cross-worker delivery.
3. On answer it re-plans once. See Realtime-and-WebSockets for the wire protocol.

Durable execution engine

services/task_engine.py is what makes the above crash-safe.

• _walk — the step loop; skips checkpointed steps on resume (checkpoint_store.is_complete).
• Lease + heartbeat — a worker claims a task run with a lease (LEASE_TTL_SECONDS = 60), renewed every LEASE_RENEW_SECONDS = 20. If the worker dies, the lease expires.
• Reconciliation — services/reconcile.py: startup_reclaim on boot and a periodic_sweep every RECLAIM_SWEEP_SECONDS = 30 find dead runs (expired lease) and either _resume them (up to TASK_MAX_RESUME_ATTEMPTS = 2) or _finalize_dead. A task therefore never stays stuck in running.
• Stores: checkpoint_store.py (task_checkpoints, append-only, token_sum), task_run_store.py (authoritative task_runs: lease, cancel flag, orphan claim), question_store.py (task_questions).

The run payload (task_runs.payload) is a frozen TaskCreate with no API keys — a resumed run re-decrypts the user's BYOK key from Postgres, so no secret is ever persisted in the run record.

Domain catalog

app/agents/domains/ — one module per built-in domain, assembled into DOMAIN_CATALOG by domains/__init__.py. Each DomainInfo declares a fixed team of SubagentSpecs and ReviewCriterions.

Built-in domains (11): software, finance, marketing, seo, searching, research, data, content, legal, education, general.

Custom agents are resolved at execution time by registry.resolve_domain_info under the custom: prefix; the user's agent persona is sandboxed inside an <agent_persona> block. If a referenced custom agent is unavailable, CustomAgentUnavailable is raised and routing falls back.

Key budgets and constants (core/constants.py)

Constant	Value	Meaning
MAX_SUBTASKS	6	Hard cap on team size
MAX_SUBTASKS_BY_COMPLEXITY	simple 1 / standard 3 / complex 6	Effort scaling
TASK_TOKEN_BUDGET_DEFAULT	200,000	Per-task token budget
ROUTE_MAX_TOKENS	512	Routing call cap
PLAN_MAX_TOKENS	1,536	Planning call cap
REVIEW_MAX_TOKENS	768	Review call cap
REVIEW_APPROVAL_THRESHOLD	0.7	Weighted approval bar
SYNTHESIS_MAX_TOKENS	4,096	Synthesis cap
SUBAGENT_MAX_TOKENS	8,192	Per-subagent cap
subagent_max_parallel	3	Concurrent subagents
subagent_max_tool_calls	6	Tool calls per subagent
MAX_ITERATIONS	10	Loop guard
MAX_REVIEW_ITERATIONS	3	Reviewer bounce guard
TASK_TIMEOUT_SECONDS	1,800	Total task timeout
HITL_TIMEOUT_SECONDS	180	Human answer timeout

Key modules

File	Responsibility
agents/base.py	AgentContext, SubagentResult, ReviewResult, JSON extraction helpers, date/memory formatting
agents/orchestrator.py	Routing only (route_decision, RouteResult)
agents/main_agent.py	Plan / assign / dependency waves / review loop / synthesize
agents/subagent.py	run_subtask, directive + native tool loops, transcript compaction
agents/reviewer.py	Weighted-criteria review, fail-mode handling
agents/registry.py	Domain catalog + dynamic custom-agent resolution
agents/prompts.py	All system-prompt templates
agents/schemas.py	RouteDecision, PlanAssignment, PlanResult, ReviewVerdict
agents/structured.py	structured_call (schema-validated LLM output)
agents/tools.py	Tool specs, directive parsing, BuiltinToolProvider
agents/validators.py	Deterministic pre-review checks
agents/budget.py	Hierarchical token budget guard
services/task_engine.py	Durable step loop, lease/heartbeat, replay
services/task_service.py	Public task API, emit envelopes, HITL delivery, RAG gather
services/reconcile.py	Crash recovery sweeps