Много сервисов, один конвейер: контракты, порядок выкладки и фабрика агентов

interface Subtask { repo: string; goal: string; contract: string }
interface Result { repo: string; ok: boolean; log: string }

async function orchestrate(feature: string): Promise {
  const plan: Subtask[] = await planSubtasks(feature);       // оркестратор делит задачу

  // fan-out: все исполнители работают параллельно, каждый в своём worktree
  let results = await Promise.all(plan.map(t => runWorker(t)));

  // проверка контрактов и повтор с текстом ошибки, не больше 2 раз
  for (let attempt = 0; attempt < 2; attempt++) {
    const failed = await contractTests(results);
    if (failed.length === 0) break;
    const retried = await Promise.all(
      failed.map(f => runWorker({ ...plan.find(p => p.repo === f.repo)!, goal: `Исправь: ${f.log}` })),
    );
    results = results.map(r => retried.find(x => x.repo === r.repo) ?? r);
  }
  return results;                                             // fan-in: связанные PR
}

import asyncio

async def orchestrate(feature: str) -> list[dict]:
    plan = await plan_subtasks(feature)            # [{"repo": "orders", "goal": ...}, ...]

    # fan-out: исполнители работают одновременно
    results = await asyncio.gather(*(run_worker(t) for t in plan))

    for _ in range(2):                             # не больше двух повторов
        failed = await contract_tests(results)
        if not failed:
            break
        retried = await asyncio.gather(*(
            run_worker({**t, "goal": f"Исправь: {f['log']}"})
            for f in failed for t in plan if t["repo"] == f["repo"]
        ))
        by_repo = {r["repo"]: r for r in retried}
        results = [by_repo.get(r["repo"], r) for r in results]

    return results                                 # fan-in

<?php
// Идея на Symfony Messenger (псевдокод): каждая подзадача — сообщение в очередь,
// отдельный воркер на репозиторий. Оркестратор ждёт все ответы и запускает проверки.

final class Orchestrator
{
    public function __construct(private MessageBusInterface $bus) {}

    public function start(Feature $feature): void
    {
        foreach ($this->planner->split($feature) as $subtask) {
            // fan-out: воркеры разбирают очередь параллельно
            $this->bus->dispatch(new RunWorker($feature->id, $subtask->repo, $subtask->goal));
        }
    }

    // вызывается, когда пришёл результат последнего исполнителя (fan-in)
    public function onAllDone(string $featureId): void
    {
        $failed = $this->contracts->check($featureId);
        foreach ($failed as $f) {
            $this->bus->dispatch(new RunWorker($featureId, $f->repo, 'Исправь: ' . $f->log));
        }
    }
}