【前編】Claude Code・Codexのトークン消費量を確認するツールを作ってみた

"""TokenScope: Claude Code/CodexのOTLP Logsを安全に可視化する教材アプリ。"""
    
    import json
    import sqlite3
    import statistics
    import threading
    from html import escape
    
    from fastapi import FastAPI, Request
    from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
    from google.protobuf import json_format
    from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import (
        ExportLogsServiceRequest,
    )
    
    
    _LOCK = threading.Lock()
    _DB = sqlite3.connect(":memory:", check_same_thread=False)
    _DB.row_factory = sqlite3.Row
    _DB.execute(
        """
        CREATE TABLE IF NOT EXISTS llm_calls (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            provider TEXT NOT NULL,
            runtime TEXT NOT NULL,
            model TEXT,
            analysis_unit TEXT NOT NULL,
            unit_kind TEXT NOT NULL,
            correlation_quality TEXT NOT NULL,
            scenario TEXT,
            variant TEXT,
            input_tokens INTEGER,
            output_tokens INTEGER,
            cache_read_tokens INTEGER,
            cache_create_tokens INTEGER,
            reasoning_tokens INTEGER,
            accounted_total_tokens INTEGER,
            token_semantics TEXT NOT NULL
        )
        """
    )
    _DB.commit()
    
    
    def _any_value(value):
        if not isinstance(value, dict):
            return None
        for key in ("stringValue", "intValue", "doubleValue", "boolValue"):
            if key in value:
                return value[key]
        return None
    
    
    def _attributes(items):
        result = {}
        for item in items or []:
            key = item.get("key")
            if isinstance(key, str) and key in {
                "service.name", "provider", "model", "prompt.id", "launcher_run",
                "run.id", "input_tokens", "output_tokens", "cache_read_input_tokens",
                "cache_creation_input_tokens", "reasoning_output_tokens",
                "tokenscope.scenario", "tokenscope.variant",
            }:
                result[key] = _any_value(item.get("value"))
        return result
    
    
    def _token(value):
        if value is None or isinstance(value, bool):
            return None
        try:
            parsed = int(value)
        except (TypeError, ValueError):
            return None
        return parsed if parsed >= 0 else None
    
    
    def _source(service_name, scope_name):
        value = f"{service_name or ''} {scope_name or ''}".lower()
        if "claude" in value or "anthropic" in value:
            return "anthropic", "claude-code"
        if "codex" in value or "openai" in value:
            return "openai", "codex"
        return "unknown", "unknown"
    
    
    def _semantics(provider, attrs):
        input_tokens = _token(attrs.get("input_tokens"))
        output_tokens = _token(attrs.get("output_tokens"))
        cache_read = _token(attrs.get("cache_read_input_tokens"))
        cache_create = _token(attrs.get("cache_creation_input_tokens"))
        reasoning = _token(attrs.get("reasoning_output_tokens"))
        if provider == "anthropic":
            parts = [input_tokens, cache_read or 0, cache_create or 0]
            input_total = sum(parts) if input_tokens is not None else None
            total = input_total + output_tokens if None not in (input_total, output_tokens) else None
            return input_total, output_tokens, cache_read, cache_create, reasoning, total, "anthropic-v1"
        if provider == "openai":
            invalid_cache = cache_read is not None and (
                input_tokens is None or cache_read > input_tokens
            )
            invalid_reasoning = reasoning is not None and (
                output_tokens is None or reasoning > output_tokens
            )
            total = None if invalid_cache or invalid_reasoning or None in (
                input_tokens, output_tokens
            ) else input_tokens + output_tokens
            return input_tokens, output_tokens, cache_read, None, reasoning, total, "openai-v1"
        return input_tokens, output_tokens, cache_read, cache_create, reasoning, None, "unknown"
    
    
    def normalize_otlp(data):
        calls = []
        for resource_log in data.get("resourceLogs", []):
            resource_attrs = _attributes(
                resource_log.get("resource", {}).get("attributes", [])
            )
            for scope_log in resource_log.get("scopeLogs", []):
                scope_name = scope_log.get("scope", {}).get("name")
                provider, runtime = _source(resource_attrs.get("service.name"), scope_name)
                for record in scope_log.get("logRecords", []):
                    attrs = {**resource_attrs, **_attributes(record.get("attributes", []))}
                    if provider == "anthropic" and attrs.get("prompt.id"):
                        unit, kind, quality = str(attrs["prompt.id"]), "prompt", "exact_provider"
                    elif provider == "openai" and (
                        attrs.get("launcher_run") or attrs.get("run.id")
                    ):
                        unit = str(attrs.get("launcher_run") or attrs.get("run.id"))
                        kind, quality = "launcher_run", "exact_launcher"
                    else:
                        continue
                    values = _semantics(provider, attrs)
                    calls.append({
                        "provider": provider,
                        "runtime": runtime,
                        "model": attrs.get("model"),
                        "analysis_unit": unit,
                        "unit_kind": kind,
                        "correlation_quality": quality,
                        "scenario": attrs.get("tokenscope.scenario"),
                        "variant": attrs.get("tokenscope.variant"),
                        "input_tokens": values[0],
                        "output_tokens": values[1],
                        "cache_read_tokens": values[2],
                        "cache_create_tokens": values[3],
                        "reasoning_tokens": values[4],
                        "accounted_total_tokens": values[5],
                        "token_semantics": values[6],
                    })
        return calls
    
    
    def decode_otlp(body, content_type):
        request = ExportLogsServiceRequest()
        if "json" in content_type:
            json_format.Parse(body.decode("utf-8"), request)
        elif "protobuf" in content_type or "octet-stream" in content_type:
            request.ParseFromString(body)
        else:
            raise ValueError("unsupported_content_type")
        return json_format.MessageToDict(request)
    
    
    def ingest(body, content_type):
        try:
            calls = normalize_otlp(decode_otlp(body, content_type))
        except Exception:
            return {"ok": False, "ingested": 0, "error": "invalid_otlp"}
        columns = tuple(calls[0].keys()) if calls else ()
        if calls:
            placeholders = ",".join("?" for _ in columns)
            with _LOCK:
                _DB.executemany(
                    f"INSERT INTO llm_calls ({','.join(columns)}) VALUES ({placeholders})",
                    [[call[column] for column in columns] for call in calls],
                )
                _DB.commit()
        return {"ok": True, "ingested": len(calls)}
    
    
    def _fixture(service, records):
        return {"resourceLogs": [{
            "resource": {"attributes": [{
                "key": "service.name", "value": {"stringValue": service}
            }]},
            "scopeLogs": [{"scope": {"name": service}, "logRecords": records}],
        }]}
    
    
    def _record(unit_key, unit, model, input_tokens, output_tokens, variant=None, **extra):
        values = {
            unit_key: unit,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "tokenscope.scenario": "article-body",
            "tokenscope.variant": variant,
            **extra,
        }
        attributes = []
        for key, value in values.items():
            if value is None:
                continue
            field = "intValue" if isinstance(value, int) else "stringValue"
            attributes.append({"key": key, "value": {field: str(value)}})
        return {"attributes": attributes}
    
    
    CLAUDE_FIXTURE = _fixture("claude-code", [
        _record("prompt.id", "prompt-before-a", "claude-opus-4-8", 18000, 4000,
                "before", cache_read_input_tokens=2000),
        _record("prompt.id", "prompt-before-b", "claude-opus-4-8", 17600, 4000,
                "before", cache_read_input_tokens=1800),
        _record("prompt.id", "prompt-before-c", "claude-opus-4-8", 19000, 4000,
                "before", cache_read_input_tokens=2200),
        _record("prompt.id", "prompt-after-a", "claude-opus-4-8", 11000, 3000,
                "after", cache_read_input_tokens=1600),
        _record("prompt.id", "prompt-after-b", "claude-opus-4-8", 10900, 3000,
                "after", cache_read_input_tokens=1500),
        _record("prompt.id", "prompt-after-c", "claude-opus-4-8", 11500, 3000,
                "after", cache_read_input_tokens=1700),
    ])
    CODEX_FIXTURE = _fixture("codex", [
        _record("launcher_run", "run-codex-001", "gpt-5.6-sol", 7200, 1800,
                cache_read_input_tokens=1200, reasoning_output_tokens=600),
        _record("launcher_run", "run-codex-missing", "gpt-5.6-sol", None, None),
    ])
    
    
    def load_fixtures():
        with _LOCK:
            _DB.execute("DELETE FROM llm_calls")
            _DB.commit()
        claude_body = json.dumps(CLAUDE_FIXTURE).encode("utf-8")
        proto = ExportLogsServiceRequest()
        json_format.ParseDict(CODEX_FIXTURE, proto)
        first = ingest(claude_body, "application/json")
        second = ingest(proto.SerializeToString(), "application/x-protobuf")
        return first["ingested"] + second["ingested"]
    
    
    def overview():
        with _LOCK:
            rows = _DB.execute(
                "SELECT provider, runtime, model, analysis_unit, unit_kind, "
                "correlation_quality, scenario, variant, input_tokens, output_tokens, "
                "cache_read_tokens, reasoning_tokens, accounted_total_tokens, token_semantics "
                "FROM llm_calls ORDER BY id"
            ).fetchall()
        calls = [dict(row) for row in rows]
        known = [row["accounted_total_tokens"] for row in calls
                 if row["accounted_total_tokens"] is not None]
        coverage = len(known) / len(calls) if calls else 0
        return {"calls": calls, "known_total": sum(known), "known": len(known),
                "detected": len(calls), "coverage": coverage}
    
    
    def bottlenecks(calls):
        groups = {}
        for call in calls:
            if call["accounted_total_tokens"] is None:
                continue
            key = (call["provider"], call["runtime"], call["model"], call["token_semantics"])
            groups.setdefault(key, []).append(call)
        ranked = []
        for key, items in groups.items():
            for rank, call in enumerate(
                sorted(items, key=lambda row: row["accounted_total_tokens"], reverse=True), 1
            ):
                ranked.append({**call, "rank": rank, "stratum": " / ".join(key)})
        return ranked
    
    
    def comparisons(calls):
        groups = {}
        for call in calls:
            if call["accounted_total_tokens"] is None or call["variant"] not in {
                "before", "after"
            }:
                continue
            key = (call["scenario"], call["provider"], call["runtime"],
                   call["model"], call["token_semantics"])
            groups.setdefault(key, {"before": [], "after": []})[call["variant"]].append(
                call["accounted_total_tokens"]
            )
        result = []
        for key, arms in groups.items():
            if not arms["before"] or not arms["after"]:
                continue
            before = statistics.median(arms["before"])
            after = statistics.median(arms["after"])
            reduction = 100 * (before - after) / before if before else None
            result.append({"scenario": key[0], "stratum": " / ".join(key[1:]),
                           "before": before, "after": after, "reduction": reduction,
                           "before_n": len(arms["before"]), "after_n": len(arms["after"]),
                           "p95": None})
        return result
    
    
    def _value(value):
        return "—" if value is None else escape(f"{value:,}" if isinstance(value, int) else str(value))
    
    
    def render_page():
        data = overview()
        coverage = f"{data['known']} / {data['detected']} ({data['coverage']:.0%})"
        rank_rows = "".join(
            f"<tr><td>{row['rank']}</td><td>{escape(row['stratum'])}</td>"
            f"<td>{escape(row['analysis_unit'])}</td><td>{row['unit_kind']}</td>"
            f"<td>{row['correlation_quality']}</td><td class='num'>{row['accounted_total_tokens']:,}</td></tr>"
            for row in bottlenecks(data["calls"])
        )
        comparison_rows = "".join(
            f"<tr><td>{escape(row['scenario'])}</td><td>{escape(row['stratum'])}</td>"
            f"<td>{row['before']:,.0f} (n={row['before_n']})</td>"
            f"<td>{row['after']:,.0f} (n={row['after_n']})</td>"
            f"<td>{row['reduction']:.1f}%</td><td>N/A (n&lt;20)</td></tr>"
            for row in comparisons(data["calls"])
        )
        detail_rows = "".join(
            f"<tr><td>{row['runtime']}</td><td>{escape(row['analysis_unit'])}</td>"
            f"<td>{row['unit_kind']} / {row['correlation_quality']}</td>"
            f"<td class='num'>{_value(row['input_tokens'])}</td>"
            f"<td class='num'>{_value(row['output_tokens'])}</td>"
            f"<td class='num'>{_value(row['cache_read_tokens'])}</td>"
            f"<td class='num'>{_value(row['reasoning_tokens'])}</td>"
            f"<td class='num'>{_value(row['accounted_total_tokens'])}</td></tr>"
            for row in data["calls"]
        )
        empty = "<p class='empty'>fixtureを投入すると分析結果が表示されます。</p>"
        return PAGE.format(
            total=f"{data['known_total']:,}", coverage=coverage,
            rank_rows=rank_rows or "<tr><td colspan='6'>未集計</td></tr>",
            comparison_rows=comparison_rows or "<tr><td colspan='6'>比較データなし</td></tr>",
            detail_rows=detail_rows or "<tr><td colspan='8'>未集計</td></tr>",
            empty=empty if not data["calls"] else "",
        )
    
    
    PAGE = """<!doctype html><html lang="ja"><head><meta charset="utf-8">
    <title>TokenScope</title><style>
    body{{font-family:system-ui,sans-serif;margin:0;background:#f4f7fb;color:#14213d}}
    main{{max-width:1120px;margin:auto;padding:28px}}header{{display:flex;justify-content:space-between;align-items:center}}
    h1{{margin:0;font-size:28px}}h2{{margin-top:28px;font-size:20px}}button{{background:#1769e0;color:white;border:0;border-radius:6px;padding:10px 18px;font-weight:700}}
    .stats{{display:grid;grid-template-columns:repeat(2,1fr);gap:14px;margin-top:20px}}.card{{background:white;border:1px solid #d8e1ee;border-radius:10px;padding:18px}}
    .label{{color:#60708a;font-size:13px}}.metric{{font-size:25px;font-weight:800;margin-top:4px}}.note{{color:#60708a;font-size:13px}}table{{width:100%;border-collapse:collapse;background:white;font-size:13px}}
    th,td{{border-bottom:1px solid #e1e7f0;padding:9px;text-align:left}}th{{background:#eaf1fb}}td.num{{text-align:right}}.empty{{padding:12px;background:#fff7db;border-left:4px solid #d49a00}}
    </style></head><body><main><header><div><h1>TokenScope</h1><p>Claude Code / Codexのトークンを、取得確度とcoverage付きで確認します。</p></div>
    <form method="post" action="/fixtures"><button id="load-fixtures" type="submit">JSON＋Protobuf fixtureを投入</button></form></header>{empty}
    <div class="stats"><div class="card"><div class="label">観測できたトークン合計</div><div id="known-total" class="metric">{total}</div><div class="note">provider別の包含規則を反映</div></div>
    <div class="card"><div class="label">field coverage</div><div id="coverage" class="metric">{coverage}</div><div class="note">未着・欠損は0にせずNULL</div></div></div>
    <h2 id="bottlenecks">ボトルネック順位</h2><table id="bottleneck-table"><thead><tr><th>順位</th><th>同一stratum</th><th>分析単位</th><th>粒度</th><th>相関品質</th><th>tokens</th></tr></thead><tbody>{rank_rows}</tbody></table>
    <h2 id="comparison">改善前後の中央値比較</h2><table id="comparison-table"><thead><tr><th>scenario</th><th>同一stratum</th><th>before中央値</th><th>after中央値</th><th>削減率</th><th>p95</th></tr></thead><tbody>{comparison_rows}</tbody></table>
    <h2 id="details">取得明細</h2><table id="detail-table"><thead><tr><th>runtime</th><th>分析単位</th><th>粒度 / 確度</th><th>input</th><th>output</th><th>cache read</th><th>reasoning</th><th>accounted total</th></tr></thead><tbody>{detail_rows}</tbody></table>
    </main></body></html>"""
    
    
    app = FastAPI(title="TokenScope")
    ROUTES = ["/", "/fixtures", "/v1/logs", "/api/overview"]
    
    
    @app.get("/", response_class=HTMLResponse)
    def dashboard():
        return HTMLResponse(render_page())
    
    
    @app.post("/fixtures")
    def fixtures():
        load_fixtures()
        return RedirectResponse("/#bottlenecks", status_code=303)
    
    
    @app.post("/v1/logs")
    async def otlp_logs(request: Request):
        result = ingest(await request.body(), request.headers.get("content-type", ""))
        return JSONResponse(result, status_code=200 if result["ok"] else 400)
    
    
    @app.get("/api/overview")
    def api_overview():
        data = overview()
        data["bottlenecks"] = bottlenecks(data["calls"])
        data["comparisons"] = comparisons(data["calls"])
        return data
    

"""既存プロンプトをClaude CodeまたはCodexで実行し、usageを記録する。"""
    
    from __future__ import annotations
    
    import argparse
    import hashlib
    import json
    import os
    import shutil
    import subprocess
    import sys
    import time
    import urllib.request
    import uuid
    from datetime import datetime
    from pathlib import Path
    from typing import Any
    
    
    DEFAULT_ENDPOINT = "http://127.0.0.1:8877/v1/logs"
    DEFAULT_CODEX_PACKAGE = "@openai/codex@0.146.0"
    
    
    def _command(name: str) -> str:
        candidates = [f"{name}.cmd", name] if os.name == "nt" else [name]
        for candidate in candidates:
            found = shutil.which(candidate)
            if found:
                return found
        raise RuntimeError(f"コマンドが見つかりません: {name}")
    
    
    def _run(command: list[str], prompt: str, cwd: Path, timeout: int,
             env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]:
        completed = subprocess.run(
            command,
            input=prompt,
            cwd=cwd,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
            check=False,
            env=env,
        )
        if completed.returncode != 0:
            tail = completed.stderr[-1200:].strip()
            raise RuntimeError(f"LLMの実行に失敗しました（exit={completed.returncode}）\n{tail}")
        return completed
    
    
    def _claude(prompt: str, cwd: Path, model: str | None, effort: str | None,
                profile: Path | None) -> tuple[Any, str, dict[str, int]]:
        command = [
            _command("claude"),
            "--print",
            "--output-format",
            "json",
            "--tools",
            "",
            "--permission-mode",
            "dontAsk",
            "--no-session-persistence",
        ]
        if model:
            command.extend(["--model", model])
        if effort:
            command.extend(["--effort", effort])
        env = os.environ.copy()
        if profile is not None:
            for name in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CONFIG_DIR"):
                env.pop(name, None)
            env["CLAUDE_CONFIG_DIR"] = str(profile.resolve())
        completed = _run(command, prompt, cwd, timeout=300, env=env)
        payload = json.loads(completed.stdout)
        raw_usage = payload.get("usage") or {}
        input_tokens = int(raw_usage.get("input_tokens") or 0)
        cache_read = int(raw_usage.get("cache_read_input_tokens") or 0)
        cache_create = int(raw_usage.get("cache_creation_input_tokens") or 0)
        output_tokens = int(raw_usage.get("output_tokens") or 0)
        usage = {
            "input_tokens": input_tokens,
            "cached_input_tokens": cache_read,
            "cache_write_input_tokens": cache_create,
            "output_tokens": output_tokens,
            "reasoning_output_tokens": 0,
            "observed_total_tokens": input_tokens + cache_read + cache_create + output_tokens,
        }
        return payload, str(payload.get("result") or ""), usage
    
    
    def _codex_command() -> list[str]:
        # Windows版デスクトップアプリ内のcodex.exeを直接起動できない環境でも、
        # 検証済みのnpmパッケージへフォールバックできるようにします。
        codex_cmd = shutil.which("codex.cmd")
        if codex_cmd:
            return [codex_cmd]
        if os.name != "nt":
            codex = shutil.which("codex")
            if codex:
                return [codex]
        return [_command("npx"), "-y", DEFAULT_CODEX_PACKAGE]
    
    
    def _codex(prompt: str, cwd: Path, model: str | None, effort: str | None) -> tuple[Any, str, dict[str, int]]:
        command = [
            *_codex_command(),
            "exec",
            "-",
            "--json",
            "--ephemeral",
            "--sandbox",
            "read-only",
            "--skip-git-repo-check",
            "--cd",
            str(cwd),
        ]
        if model:
            command.extend(["--model", model])
        if effort:
            command.extend(["--config", f'model_reasoning_effort="{effort}"'])
        completed = _run(command, prompt, cwd, timeout=420)
        events = [json.loads(line) for line in completed.stdout.splitlines() if line.strip()]
        answer = ""
        raw_usage: dict[str, Any] = {}
        for event in events:
            item = event.get("item")
            if event.get("type") == "item.completed" and isinstance(item, dict):
                if item.get("type") == "agent_message":
                    answer = str(item.get("text") or "")
            if event.get("type") == "turn.completed":
                raw_usage = event.get("usage") or {}
        if not raw_usage:
            raise RuntimeError("Codexのturn.completed.usageが見つかりません")
        input_tokens = int(raw_usage.get("input_tokens") or 0)
        output_tokens = int(raw_usage.get("output_tokens") or 0)
        usage = {
            "input_tokens": input_tokens,
            "cached_input_tokens": int(raw_usage.get("cached_input_tokens") or 0),
            "cache_write_input_tokens": int(raw_usage.get("cache_write_input_tokens") or 0),
            "output_tokens": output_tokens,
            "reasoning_output_tokens": int(raw_usage.get("reasoning_output_tokens") or 0),
            "observed_total_tokens": input_tokens + output_tokens,
        }
        return events, answer, usage
    
    
    def _attribute(key: str, value: str | int) -> dict[str, Any]:
        field = "intValue" if isinstance(value, int) else "stringValue"
        return {"key": key, "value": {field: str(value)}}
    
    
    def _otlp(runtime: str, model: str, effort: str, run_id: str,
              scenario: str, variant: str, usage: dict[str, int]) -> dict[str, Any]:
        unit_key = "prompt.id" if runtime == "claude-code" else "launcher_run"
        attributes = [
            _attribute(unit_key, run_id),
            _attribute("model", model),
            _attribute("input_tokens", usage["input_tokens"]),
            _attribute("output_tokens", usage["output_tokens"]),
            _attribute("tokenscope.scenario", scenario),
            _attribute("tokenscope.variant", variant),
        ]
        if usage["cached_input_tokens"]:
            attributes.append(_attribute("cache_read_input_tokens", usage["cached_input_tokens"]))
        if runtime == "claude-code" and usage["cache_write_input_tokens"]:
            attributes.append(_attribute(
                "cache_creation_input_tokens", usage["cache_write_input_tokens"]
            ))
        if runtime == "codex" and usage["reasoning_output_tokens"]:
            attributes.append(_attribute(
                "reasoning_output_tokens", usage["reasoning_output_tokens"]
            ))
        return {
            "resourceLogs": [{
                "resource": {"attributes": [_attribute("service.name", runtime)]},
                "scopeLogs": [{
                    "scope": {"name": runtime},
                    "logRecords": [{"attributes": attributes}],
                }],
            }],
        }
    
    
    def _post(endpoint: str, payload: dict[str, Any]) -> dict[str, Any]:
        request = urllib.request.Request(
            endpoint,
            data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(request, timeout=15) as response:
            return json.loads(response.read().decode("utf-8"))
    
    
    def main() -> int:
        parser = argparse.ArgumentParser(
            description="prompt.txtをClaude CodeまたはCodexで実行し、usageを記録します"
        )
        parser.add_argument("runtime", choices=("claude-code", "codex"))
        parser.add_argument("prompt_file", type=Path)
        parser.add_argument("--model")
        parser.add_argument("--effort", choices=("low", "medium", "high", "xhigh", "max"))
        parser.add_argument("--claude-profile", type=Path)
        parser.add_argument("--scenario", default="existing-prompt")
        parser.add_argument("--variant", default="baseline")
        parser.add_argument("--endpoint", default=DEFAULT_ENDPOINT)
        parser.add_argument("--cwd", type=Path, default=Path.cwd())
        parser.add_argument("--output-dir", type=Path, default=Path("measurement-output"))
        parser.add_argument(
            "--save-response",
            action="store_true",
            help="回答本文と生のCLI出力も結果JSONへ保存します",
        )
        args = parser.parse_args()
    
        prompt_path = args.prompt_file.resolve()
        cwd = args.cwd.resolve()
        prompt = prompt_path.read_text(encoding="utf-8")
        if not prompt.strip():
            raise ValueError("プロンプトファイルが空です")
    
        runtime = args.runtime
        model = args.model or "default"
        effort = args.effort or "default"
        run_id = f"run-{runtime}-{uuid.uuid4().hex[:12]}"
        started = time.monotonic()
        print(f"[1/3] {prompt_path.name}を{runtime}で実行します", flush=True)
        if runtime == "claude-code":
            raw, answer, usage = _claude(
                prompt, cwd, args.model, args.effort, args.claude_profile
            )
        else:
            raw, answer, usage = _codex(prompt, cwd, args.model, args.effort)
        elapsed_ms = round((time.monotonic() - started) * 1000)
    
        print("[2/3] usageを取得しました", flush=True)
        print(json.dumps(usage, ensure_ascii=False, indent=2), flush=True)
        ingest = _post(
            args.endpoint,
            _otlp(runtime, model, effort, run_id, args.scenario, args.variant, usage),
        )
        print(f"[3/3] 管理画面へ記録しました: {ingest}", flush=True)
    
        args.output_dir.mkdir(parents=True, exist_ok=True)
        stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        output_path = args.output_dir / f"{stamp}-{runtime}.json"
        result = {
            "runtime": runtime,
            "model": model,
            "effort": effort,
            "run_id": run_id,
            "scenario": args.scenario,
            "variant": args.variant,
            "prompt_file_name": prompt_path.name,
            "prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
            "prompt_chars": len(prompt),
            "usage": usage,
            "elapsed_ms": elapsed_ms,
            "ingest": ingest,
        }
        if args.save_response:
            result["answer"] = answer
            result["raw"] = raw
        output_path.write_text(
            json.dumps(result, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        print(f"結果ファイル: {output_path.resolve()}")
        return 0
    
    
    if __name__ == "__main__":
        sys.exit(main())