Claude API の本番運用で予算上限を確実に守る — 日次トークン超過を検知して自動停止するサーキットブレーカー設計

// src/lib/budget-breaker.ts
import Anthropic from "@anthropic-ai/sdk";
 
const DAILY_TOKEN_CAP = 5_000_000; // 1日500万トークン上限（例）
 
interface Env {
  ANTHROPIC_API_KEY: string;
  BUDGET_KV: KVNamespace; // wrangler.toml で binding 済み
}
 
interface BudgetState {
  inputTokens: number;
  outputTokens: number;
  date: string; // YYYY-MM-DD
}
 
// UTC基準の日付キー（タイムゾーンを統一）
function todayKey(): string {
  return new Date().toISOString().slice(0, 10);
}
 
// 現在の消費量を取得
async function readBudget(env: Env): Promise<BudgetState> {
  const key = `budget:${todayKey()}`;
  const raw = await env.BUDGET_KV.get(key);
  if (!raw) return { inputTokens: 0, outputTokens: 0, date: todayKey() };
  return JSON.parse(raw) as BudgetState;
}
 
// 事前チェック: 残額があれば true、なければ false
export async function checkBudget(env: Env): Promise<{
  allowed: boolean;
  used: number;
  remaining: number;
}> {
  const state = await readBudget(env);
  const used = state.inputTokens + state.outputTokens;
  return {
    allowed: used < DAILY_TOKEN_CAP,
    used,
    remaining: Math.max(0, DAILY_TOKEN_CAP - used),
  };
}
 
// 事後更新: 実消費量を記録（KV の TTL で2日後に自動削除）
export async function recordUsage(
  env: Env,
  inputTokens: number,
  outputTokens: number
): Promise<void> {
  const key = `budget:${todayKey()}`;
  const state = await readBudget(env);
  state.inputTokens += inputTokens;
  state.outputTokens += outputTokens;
  await env.BUDGET_KV.put(key, JSON.stringify(state), {
    expirationTtl: 60 * 60 * 48, // 48時間後に自動削除
  });
}
 
// ラッパー: API 呼び出しの前後で予算を管理
export async function callClaudeWithBudget(
  env: Env,
  request: Anthropic.MessageCreateParams
): Promise<Anthropic.Message | { error: "BUDGET_EXCEEDED" }> {
  const budget = await checkBudget(env);
  if (!budget.allowed) {
    console.warn(`[BUDGET] Blocked. used=${budget.used} cap=${DAILY_TOKEN_CAP}`);
    return { error: "BUDGET_EXCEEDED" };
  }
 
  const client = new Anthropic({ apiKey: env.ANTHROPIC_API_KEY });
  const response = await client.messages.create(request);
 
  // 実消費量を記録（input/output 別々に）
  await recordUsage(
    env,
    response.usage.input_tokens,
    response.usage.output_tokens
  );
 
  return response;
}
 
// 期待する出力例（remaining が 0 になった瞬間から、後続のリクエストは BUDGET_EXCEEDED を返す）:
// { type: "message", id: "msg_01...", content: [{ type: "text", text: "..." }],
//   usage: { input_tokens: 120, output_tokens: 480 } }
// もしくは予算超過時:
// { error: "BUDGET_EXCEEDED" }

// src/durable-objects/budget-tracker.ts
export class BudgetTracker {
  private state: DurableObjectState;
  private cap: number = 5_000_000;
 
  constructor(state: DurableObjectState) {
    this.state = state;
  }
 
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
 
    if (url.pathname === "/check") {
      const used = (await this.state.storage.get<number>("used")) ?? 0;
      return Response.json({
        allowed: used < this.cap,
        used,
        remaining: Math.max(0, this.cap - used),
      });
    }
 
    if (url.pathname === "/record") {
      const { tokens } = await request.json<{ tokens: number }>();
      // ⚠️ 重要: storage.transaction() で原子的に更新する
      await this.state.storage.transaction(async (txn) => {
        const current = (await txn.get<number>("used")) ?? 0;
        await txn.put("used", current + tokens);
      });
      return Response.json({ ok: true });
    }
 
    if (url.pathname === "/reset") {
      // 日次リセット用エンドポイント（Cron Trigger から叩く）
      await this.state.storage.deleteAll();
      return Response.json({ ok: true });
    }
 
    return new Response("Not Found", { status: 404 });
  }
}
 
// Worker 側: 単一の DO インスタンスにすべてのリクエストを集約
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const todayKey = new Date().toISOString().slice(0, 10);
    const id = env.BUDGET_DO.idFromName(`budget-${todayKey}`);
    const stub = env.BUDGET_DO.get(id);
 
    // 事前チェック
    const checkRes = await stub.fetch("https://do/check");
    const { allowed, used, remaining } = await checkRes.json<{
      allowed: boolean;
      used: number;
      remaining: number;
    }>();
 
    if (!allowed) {
      return new Response(
        JSON.stringify({ error: "BUDGET_EXCEEDED", used, remaining }),
        { status: 429, headers: { "content-type": "application/json" } }
      );
    }
 
    // Claude API を呼び出す
    const client = new Anthropic({ apiKey: env.ANTHROPIC_API_KEY });
    const response = await client.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 1024,
      messages: [{ role: "user", content: "..." }],
    });
 
    // 実消費量を記録（原子的）
    const totalTokens = response.usage.input_tokens + response.usage.output_tokens;
    await stub.fetch("https://do/record", {
      method: "POST",
      body: JSON.stringify({ tokens: totalTokens }),
    });
 
    return Response.json(response);
  },
} satisfies ExportedHandler<Env>;
 
// 期待する挙動:
// - 100並列で同時に呼び出しても、合計トークンは正確にカウンタに加算される
// - 閾値超過した瞬間から、後続リクエストは即座に 429 BUDGET_EXCEEDED を返す
// - 閾値オーバーは最大1リクエスト分（並列で通過する瞬間のレース）で抑えられる

// src/lib/budget-breaker-degraded.ts
type DegradationStrategy = "halt" | "fallback-model" | "cached-response" | "alert-only";
 
interface BreakerConfig {
  cap: number;
  strategy: DegradationStrategy;
  fallbackModel?: string;
  cachedResponse?: string;
  webhookUrl?: string; // Slack Incoming Webhook
}
 
export async function callClaudeWithDegradation(
  env: Env,
  config: BreakerConfig,
  request: Anthropic.MessageCreateParams
): Promise<Anthropic.Message> {
  const stub = getBudgetStub(env);
  const checkRes = await stub.fetch("https://do/check");
  const { allowed, used } = await checkRes.json<{ allowed: boolean; used: number }>();
 
  // 閾値の80%を超えた時点で予防的アラート（戻り値は変更しない）
  if (used >= config.cap * 0.8 && config.webhookUrl) {
    fetch(config.webhookUrl, {
      method: "POST",
      body: JSON.stringify({
        text: `⚠️ Claude API budget at ${Math.round((used / config.cap) * 100)}% (used=${used} / cap=${config.cap})`,
      }),
    }).catch(() => {}); // 通知失敗は無視
  }
 
  if (allowed) {
    return await callClaudeAndRecord(env, request);
  }
 
  // 予算超過時の分岐
  switch (config.strategy) {
    case "halt":
      throw new BudgetExceededError("Daily budget exhausted");
 
    case "fallback-model":
      // 安価モデルで再試行（フォールバックモデルにも別予算を持たせると堅牢）
      return await callClaudeAndRecord(env, {
        ...request,
        model: config.fallbackModel ?? "claude-haiku-4-5-20251001",
      });
 
    case "cached-response":
      // 事前用意のテンプレート応答を返す
      return makeFakeMessage(
        config.cachedResponse ?? "現在、応答の生成に時間がかかっています。"
      );
 
    case "alert-only":
      // 通知のみして、通常通り呼び出す
      await notify(config.webhookUrl, `[BREAKER] cap exceeded but allowing through. used=${used}`);
      return await callClaudeAndRecord(env, request);
  }
}
 
class BudgetExceededError extends Error {}
 
function makeFakeMessage(text: string): Anthropic.Message {
  return {
    id: `msg_fallback_${Date.now()}`,
    type: "message",
    role: "assistant",
    content: [{ type: "text", text, citations: null }],
    model: "fallback",
    stop_reason: "end_turn",
    stop_sequence: null,
    usage: { input_tokens: 0, output_tokens: 0, cache_creation_input_tokens: null, cache_read_input_tokens: null },
  } as Anthropic.Message;
}
 
// 期待する出力例:
// 通常時: 通常の Anthropic.Message
// 予算超過 + fallback-model: 同じ Anthropic.Message (model フィールドが Haiku になっている)
// 予算超過 + cached-response: { id: "msg_fallback_...", model: "fallback", usage: { ..._tokens: 0 } }

// ストリーミング時のトークン記録
let inputTokens = 0;
let outputTokens = 0;
 
for await (const event of stream) {
  if (event.type === "message_start") {
    inputTokens = event.message.usage.input_tokens;
  }
  if (event.type === "message_delta") {
    outputTokens = event.usage.output_tokens;
  }
  // 各 chunk をクライアントに転送
  yield event;
}
 
// ストリーム完了後に必ず記録
await recordUsage(env, inputTokens, outputTokens);
 
// 期待する挙動: ストリーミングが正常終了しても、途中切断（クライアント切断・タイムアウト）でも、
//             記録処理が呼ばれることを finally ブロックで保証する

// モデル別の単価（USD per 1M tokens、2026-04 時点の Sonnet 4.6 例）
const PRICING = {
  inputPer1M: 3.0,
  outputPer1M: 15.0,
  cacheWritePer1M: 3.75,  // 書き込みは1.25倍
  cacheReadPer1M: 0.3,    // 読み込みは1/10
};
 
function estimateCostUsd(usage: Anthropic.Usage): number {
  const input = (usage.input_tokens * PRICING.inputPer1M) / 1_000_000;
  const output = (usage.output_tokens * PRICING.outputPer1M) / 1_000_000;
  const cacheW = ((usage.cache_creation_input_tokens ?? 0) * PRICING.cacheWritePer1M) / 1_000_000;
  const cacheR = ((usage.cache_read_input_tokens ?? 0) * PRICING.cacheReadPer1M) / 1_000_000;
  return input + output + cacheW + cacheR;
}
// 単価は変動するため、最新値は Anthropic 公式の Pricing ページを参照すること

const BREAKER_MODE: "monitor" | "enforce" = (env.BREAKER_MODE as any) ?? "monitor";
 
if (!allowed) {
  if (BREAKER_MODE === "enforce") {
    return { error: "BUDGET_EXCEEDED" };
  } else {
    console.warn(`[BREAKER] Would block (monitor mode). used=${used}`);
    // 通常通り通過させる
  }
}

// Workers Analytics Engine への記録例
env.ANALYTICS.writeDataPoint({
  blobs: [
    request.model ?? "unknown",      // blob1: モデル名
    config.strategy,                  // blob2: 戦略
    allowed ? "passed" : "blocked",   // blob3: 通過 / 遮断
  ],
  doubles: [
    response.usage?.input_tokens ?? 0,   // double1: 入力トークン
    response.usage?.output_tokens ?? 0,  // double2: 出力トークン
    estimateCostUsd(response.usage),     // double3: 推定コスト
  ],
  indexes: [todayKey()], // YYYY-MM-DD
});
 
// 集計クエリ例（Wrangler / GraphQL API 経由）:
//  SELECT blob1 AS model, SUM(double1+double2) AS tokens, SUM(double3) AS cost_usd
//  FROM claude_api_dataset
//  WHERE timestamp > NOW() - INTERVAL '24' HOUR
//  GROUP BY blob1 ORDER BY tokens DESC;

// scope ごとに独立した予算カウンタを持つ
function getBudgetStub(env: Env, scope: string): DurableObjectStub {
  const todayKey = new Date().toISOString().slice(0, 10);
  const id = env.BUDGET_DO.idFromName(`budget-${scope}-${todayKey}`);
  return env.BUDGET_DO.get(id);
}
 
// 使用例
await callClaudeWithBudget(env, "frontend-chat", request);
await callClaudeWithBudget(env, "batch-summary", request);
await callClaudeWithBudget(env, "internal-tools", request);