Codexマルチエージェント実装ガイド

依存関係なし → 完全並列実行
直列依存     → パイプライン（前タスク完了待ち）
部分依存     → ファンアウト/ファンイン（依存部だけシリアル）

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional

class TaskType(Enum):
    INDEPENDENT = "independent"   # 完全並列化可能
    SEQUENTIAL = "sequential"     # 直列実行
    DEPENDENT = "dependent"       # 他タスクの完了を待つ

@dataclass
class CodexTask:
    id: str
    prompt: str
    target_files: List[str]
    task_type: TaskType
    depends_on: List[str] = field(default_factory=list)
    max_retries: int = 3

def classify_tasks(tasks: List[CodexTask]) -> dict:
    """依存グラフを構築してタスクを分類する"""
    dependency_map = {t.id: t.depends_on for t in tasks}
    independent = [t for t in tasks if not t.depends_on]
    dependent = [t for t in tasks if t.depends_on]
    return {"independent": independent, "dependent": dependent}

import math
from pathlib import Path
from typing import List

def split_files_into_chunks(
    file_paths: List[Path],
    worker_count: int,
    max_tokens_per_chunk: int = 8000
) -> List[List[Path]]:
    """
    ファイルをトークン数を考慮してチャンクに分割する。
    大きなファイルは単独チャンクに。
    """
    chunks: List[List[Path]] = [[] for _ in range(worker_count)]
    chunk_sizes = [0] * worker_count

    # ファイルサイズでソートし、大きいものを先に割り当て（bin packing）
    sorted_files = sorted(file_paths, key=lambda p: p.stat().st_size, reverse=True)

    for file_path in sorted_files:
        # 最も空きのあるチャンクに割り当て
        min_idx = chunk_sizes.index(min(chunk_sizes))
        chunks[min_idx].append(file_path)
        chunk_sizes[min_idx] += file_path.stat().st_size

    return [chunk for chunk in chunks if chunk]

// orchestrator.js
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const { spawn } = require('child_process');
const path = require('path');

const MAX_WORKERS = 4;
const RATE_LIMIT_RPM = 60; // requests per minute (公式値: Codex API tier による)

class TokenBucket {
  constructor(capacity, refillRate) {
    this.capacity = capacity;
    this.tokens = capacity;
    this.refillRate = refillRate; // tokens per second
    this.lastRefill = Date.now();
  }

  async consume(tokens = 1) {
    this.refill();
    if (this.tokens < tokens) {
      const waitMs = ((tokens - this.tokens) / this.refillRate) * 1000;
      await new Promise(resolve => setTimeout(resolve, waitMs));
      this.refill();
    }
    this.tokens -= tokens;
  }

  refill() {
    const now = Date.now();
    const elapsed = (now - this.lastRefill) / 1000;
    this.tokens = Math.min(this.capacity, this.tokens + elapsed * this.refillRate);
    this.lastRefill = now;
  }
}

class CodexOrchestrator {
  constructor(maxWorkers = MAX_WORKERS) {
    this.maxWorkers = maxWorkers;
    this.activeWorkers = 0;
    this.queue = [];
    this.results = new Map();
    this.errors = new Map();
    // RATE_LIMIT_RPM を秒換算してバケット設定
    this.rateLimiter = new TokenBucket(RATE_LIMIT_RPM, RATE_LIMIT_RPM / 60);
  }

  async executeTask(task) {
    await this.rateLimiter.consume();
    return new Promise((resolve, reject) => {
      const proc = spawn('codex', [
        '--model', 'codex-1',
        '--quiet',
        task.prompt
      ], {
        cwd: task.workDir,
        env: { ...process.env }
      });

      let stdout = '';
      let stderr = '';

      proc.stdout.on('data', data => { stdout += data.toString(); });
      proc.stderr.on('data', data => { stderr += data.toString(); });

      proc.on('close', code => {
        if (code === 0) {
          resolve({ taskId: task.id, output: stdout });
        } else {
          reject(new Error(`Task ${task.id} failed (exit ${code}): ${stderr}`));
        }
      });
    });
  }

  async runAll(tasks) {
    const results = await Promise.allSettled(
      tasks.map(task => this.executeWithRetry(task))
    );
    return this.mergeResults(results, tasks);
  }

  async executeWithRetry(task, attempt = 0) {
    try {
      return await this.executeTask(task);
    } catch (err) {
      if (attempt < task.maxRetries && this.isRetryable(err)) {
        const backoffMs = Math.pow(2, attempt) * 1000;
        await new Promise(r => setTimeout(r, backoffMs));
        return this.executeWithRetry(task, attempt + 1);
      }
      throw err;
    }
  }

  isRetryable(error) {
    // rate limit エラーと一時的なネットワークエラーのみリトライ
    return error.message.includes('rate_limit') ||
           error.message.includes('ECONNRESET') ||
           error.message.includes('529');
  }

  mergeResults(settledResults, tasks) {
    const succeeded = [];
    const failed = [];

    settledResults.forEach((result, idx) => {
      if (result.status === 'fulfilled') {
        succeeded.push(result.value);
      } else {
        failed.push({ taskId: tasks[idx].id, error: result.reason.message });
      }
    });

    return { succeeded, failed, total: tasks.length };
  }
}

module.exports = { CodexOrchestrator };

import asyncio
import time
import random
from typing import Callable, TypeVar, Awaitable

T = TypeVar('T')

async def retry_with_backoff(
    func: Callable[[], Awaitable[T]],
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True
) -> T:
    """
    指数バックオフ + jitter でリトライする汎用ラッパー。
    429 (rate limit) と 529 (overloaded) のみリトライ対象。
    """
    for attempt in range(max_retries + 1):
        try:
            return await func()
        except Exception as e:
            error_code = getattr(e, 'status_code', None)
            if attempt == max_retries or error_code not in (429, 529):
                raise

            delay = min(base_delay * (2 ** attempt), max_delay)
            if jitter:
                delay *= (0.5 + random.random() * 0.5)

            print(f"Attempt {attempt + 1} failed (code={error_code}). "
                  f"Retrying in {delay:.1f}s...")
            await asyncio.sleep(delay)

import asyncio
from collections import deque

class SlidingWindowRateLimiter:
    """1分間のスライディングウィンドウでリクエスト数を制限する"""

    def __init__(self, max_requests: int, window_seconds: float = 60.0):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.request_times: deque = deque()
        self.lock = asyncio.Lock()

    async def acquire(self):
        async with self.lock:
            now = time.monotonic()
            # 古いリクエスト記録を削除
            while self.request_times and \
                  now - self.request_times[0] > self.window_seconds:
                self.request_times.popleft()

            if len(self.request_times) >= self.max_requests:
                # 最も古いリクエストがウィンドウ外になるまで待機
                wait_time = self.window_seconds - (now - self.request_times[0])
                await asyncio.sleep(wait_time + 0.01)

            self.request_times.append(time.monotonic())

import json
import fcntl
from pathlib import Path

class SharedStateManager:
    """
    ファイルロックを使ってワーカー間で状態を安全に共有する。
    大規模並列（>10ワーカー）では Redis や SQLite を推奨。
    """

    def __init__(self, state_file: Path):
        self.state_file = state_file
        self.lock_file = state_file.with_suffix('.lock')
        self.state_file.parent.mkdir(parents=True, exist_ok=True)
        if not self.state_file.exists():
            self.state_file.write_text('{}')

    def update(self, task_id: str, result: dict):
        with open(self.lock_file, 'w') as lock:
            fcntl.flock(lock, fcntl.LOCK_EX)
            try:
                state = json.loads(self.state_file.read_text())
                state[task_id] = result
                self.state_file.write_text(json.dumps(state, indent=2))
            finally:
                fcntl.flock(lock, fcntl.LOCK_UN)

    def read(self) -> dict:
        return json.loads(self.state_file.read_text())

from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"

class ProgressTracker:
    def __init__(self, state_manager: SharedStateManager):
        self.state = state_manager

    def is_completed(self, task_id: str) -> bool:
        """冪等実行のため、完了済みタスクをスキップできる"""
        state = self.state.read()
        return state.get(task_id, {}).get('status') == TaskStatus.COMPLETED.value

    def mark_running(self, task_id: str):
        self.state.update(task_id, {'status': TaskStatus.RUNNING.value,
                                     'started_at': time.time()})

    def mark_completed(self, task_id: str, output: str):
        self.state.update(task_id, {
            'status': TaskStatus.COMPLETED.value,
            'output': output,
            'completed_at': time.time()
        })

    def mark_failed(self, task_id: str, error: str):
        self.state.update(task_id, {
            'status': TaskStatus.FAILED.value,
            'error': error,
            'failed_at': time.time()
        })

from typing import List

class TaskQueue:
    def __init__(self, state_manager: SharedStateManager):
        self.state = state_manager
        self.dlq: List[dict] = []

    async def process_with_dlq(self, tasks: List[CodexTask], runner):
        results = []
        for task in tasks:
            try:
                result = await retry_with_backoff(
                    lambda: runner.execute(task),
                    max_retries=task.max_retries
                )
                results.append(result)
            except Exception as e:
                # リトライ上限を超えたら DLQ へ
                self.dlq.append({
                    'task_id': task.id,
                    'error': str(e),
                    'timestamp': time.time(),
                    'task': task
                })
        return results, self.dlq

import subprocess
from pathlib import Path

class ResultMerger:
    """
    ワーカーが生成したファイル変更を安全にマージする。
    競合が発生した場合はマーカーを残してレビューを促す。
    """

    def merge_file_changes(
        self,
        base_file: Path,
        changes: list[dict]
    ) -> tuple[str, list[str]]:
        """
        複数ワーカーの変更を順にパッチ適用する。
        競合検出時は conflict_files に追記。
        """
        current_content = base_file.read_text()
        conflict_files = []

        for change in changes:
            try:
                result = subprocess.run(
                    ['patch', '--merge', str(base_file)],
                    input=change['patch'],
                    capture_output=True,
                    text=True
                )
                if result.returncode != 0:
                    conflict_files.append(base_file.name)
            except Exception as e:
                conflict_files.append(f"{base_file.name}: {e}")

        return base_file.read_text(), conflict_files

    def generate_summary(self, results: list[dict]) -> str:
        succeeded = [r for r in results if r.get('status') == 'completed']
        failed = [r for r in results if r.get('status') == 'failed']
        return (
            f"処理完了: {len(succeeded)}/{len(results)} タスク成功, "
            f"{len(failed)} タスク失敗"
        )