並行サブエージェントが同じAPIの上限を奪い合うとき — 共有トークンバケットで集約レートを抑える

// 一見正しいが、並行時に破綻する「各自バックオフ」方式
async function callWithBackoff(fn: () => Promise<Response>): Promise<Response> {
  let delay = 500;
  for (let attempt = 0; attempt < 6; attempt++) {
    const res = await fn();
    if (res.status !== 429 && res.status !== 403) return res;
    await sleep(delay);     // 全エージェントがほぼ同時に同じ delay を待つ
    delay *= 2;
  }
  throw new Error("rate limit: gave up");
}

// shared-limiter.ts — FIFO 公平な非同期トークンバケット
type Waiter = { cost: number; resolve: () => void };
 
export class TokenBucket {
  private tokens: number;
  private last: number;
  private waiters: Waiter[] = [];
  private timer: ReturnType<typeof setInterval> | null = null;
 
  constructor(
    private readonly capacity: number,    // バースト許容量
    private readonly refillPerSec: number // 定常レート（毎秒の発行数）
  ) {
    this.tokens = capacity;
    this.last = Date.now();
  }
 
  private refill(): void {
    const now = Date.now();
    const elapsed = (now - this.last) / 1000;
    if (elapsed <= 0) return;
    this.tokens = Math.min(this.capacity, this.tokens + elapsed * this.refillPerSec);
    this.last = now;
  }
 
  // 呼び出し前に必ず await する。待たされた分だけ送信が遅延し、集約レートが上限内に収まる
  async acquire(cost = 1): Promise<void> {
    if (cost > this.capacity) {
      throw new Error("cost が capacity を超えています: 永久に取得できません");
    }
    this.refill();
    // 先客がいなければ即時取得（待ち行列を追い越さない＝公平性）
    if (this.waiters.length === 0 && this.tokens >= cost) {
      this.tokens -= cost;
      return;
    }
    return new Promise<void>((resolve) => {
      this.waiters.push({ cost, resolve });
      this.startDraining();
    });
  }
 
  private startDraining(): void {
    if (this.timer) return;
    this.timer = setInterval(() => {
      this.refill();
      // 先頭から順に、トークンが足りる限り解放（FIFO）
      while (this.waiters.length > 0 && this.tokens >= this.waiters[0].cost) {
        const w = this.waiters.shift()!;
        this.tokens -= w.cost;
        w.resolve();
      }
      if (this.waiters.length === 0 && this.timer) {
        clearInterval(this.timer);
        this.timer = null;
      }
    }, 50); // 50ms ごとに補充と解放を判定
  }
}

// GitHub のコンテンツ作成系は控えめに「毎秒1.0・バースト5」に絞る
const github = new TokenBucket(5, 1.0);
 
async function commitViaSubAgent(agentId: string, change: FileChange): Promise<void> {
  await github.acquire(1);          // ここで順番待ちが起きる
  await githubApi.createCommit(change);
}
 
// 6個のサブエージェントが同じ github バケツを共有して並行実行
await Promise.all(
  subAgents.map((a) => commitViaSubAgent(a.id, a.pendingChange))
);