diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py
index 1cf48ec17051..41711acebc19 100644
--- a/agent/codex_runtime.py
+++ b/agent/codex_runtime.py
@@ -22,6 +22,8 @@
 from types import SimpleNamespace
 from typing import Any, Dict, List
 
+from agent.credential_usage import resolve_credential_label
+
 logger = logging.getLogger(__name__)
 
 
@@ -209,6 +211,7 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]:
                 billing_base_url=agent.base_url,
                 billing_mode="subscription_included"
                 if cost_result.status == "included" else None,
+                credential_label=resolve_credential_label(agent),
                 model=agent.model,
                 api_call_count=1,
             )
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index 761c8c0f79e6..1f7dc13f9456 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -28,6 +28,7 @@
 from typing import Any, Dict, List, Optional
 
 from agent.codex_responses_adapter import _summarize_user_message_for_log
+from agent.credential_usage import resolve_credential_label
 from agent.conversation_compression import conversation_history_after_compression
 from agent.display import KawaiiSpinner
 from agent.error_classifier import FailoverReason, classify_api_error
@@ -2210,6 +2211,7 @@ def _perform_api_call(next_api_kwargs):
                                 billing_base_url=agent.base_url,
                                 billing_mode="subscription_included"
                                 if cost_result.status == "included" else None,
+                                credential_label=resolve_credential_label(agent),
                                 model=agent.model,
                                 api_call_count=1,
                             )
diff --git a/agent/credential_usage.py b/agent/credential_usage.py
new file mode 100644
index 000000000000..504d85ecb249
--- /dev/null
+++ b/agent/credential_usage.py
@@ -0,0 +1,46 @@
+"""Helpers for attributing model-token usage to credential-pool labels."""
+from __future__ import annotations
+
+from typing import Any, Optional
+
+
+def resolve_credential_label(agent: Any) -> Optional[str]:
+    """Best-effort credential label for the current model call.
+
+    The router selects a runtime token from the provider credential pool before
+    the agent is initialized. We intentionally store only the non-secret label,
+    never the token. If matching fails, return None and skip credential-level
+    telemetry for that call.
+    """
+    provider = str(getattr(agent, "provider", "") or "").strip().lower()
+    if not provider:
+        return None
+    api_key = str(getattr(agent, "api_key", "") or "").strip()
+    pool = getattr(agent, "_credential_pool", None)
+    if pool is None:
+        try:
+            from agent.credential_pool import load_pool
+
+            pool = load_pool(provider)
+        except Exception:
+            return None
+    try:
+        entries = pool._available_entries(clear_expired=False, refresh=False)
+    except Exception:
+        try:
+            entries = getattr(pool, "_entries", []) or []
+        except Exception:
+            return None
+    for entry in entries:
+        try:
+            if api_key and str(getattr(entry, "runtime_api_key", "") or "").strip() == api_key:
+                return str(getattr(entry, "label", "") or "").strip() or None
+        except Exception:
+            continue
+    try:
+        current = pool.current() if callable(getattr(pool, "current", None)) else None
+        if current is not None:
+            return str(getattr(current, "label", "") or "").strip() or None
+    except Exception:
+        return None
+    return None
diff --git a/gateway/run.py b/gateway/run.py
index 51e33ceb4996..0d0c0e8508da 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -9640,6 +9640,9 @@ async def _do_undo():
         if canonical == "usage":
             return await self._handle_usage_command(event)
 
+        if canonical == "codex-usage":
+            return await self._handle_codex_usage_command(event)
+
         if canonical == "credits":
             return await self._handle_credits_command(event)
 
diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py
index 14687e07dde0..f3ac1b10d8f6 100644
--- a/gateway/slash_commands.py
+++ b/gateway/slash_commands.py
@@ -4054,6 +4054,43 @@ async def _handle_usage_command(self, event: MessageEvent) -> str:
             return "\n".join(parts)
         return t("gateway.usage.no_data")
 
+    async def _handle_codex_usage_command(self, event: MessageEvent) -> str:
+        """Handle /codex-usage — current Codex quota by Hermes credential."""
+        raw_args = event.get_command_args().strip() if event else ""
+        argv = shlex.split(raw_args) if raw_args else ["--compact"]
+        if not argv:
+            argv = ["--compact"]
+        try:
+            from hermes_cli.codex_usage import collect, render_alert, render_compact, render_text
+
+            payload = await asyncio.to_thread(collect)
+            primary = None
+            secondary = None
+            alert = None
+            verbose = False
+            i = 0
+            while i < len(argv):
+                arg = argv[i]
+                if arg in {"--verbose", "verbose"}:
+                    verbose = True
+                elif arg == "--alert-threshold" and i + 1 < len(argv):
+                    i += 1
+                    alert = float(argv[i])
+                elif arg == "--primary-threshold" and i + 1 < len(argv):
+                    i += 1
+                    primary = float(argv[i])
+                elif arg == "--secondary-threshold" and i + 1 < len(argv):
+                    i += 1
+                    secondary = float(argv[i])
+                i += 1
+            if alert is not None or primary is not None or secondary is not None:
+                out = render_alert(payload, alert, None, primary_threshold=primary, secondary_threshold=secondary)
+                return out or "Codex usage OK."
+            return render_text(payload) if verbose else render_compact(payload)
+        except Exception as exc:
+            logger.warning("/codex-usage failed: %s", exc)
+            return f"Codex usage check failed: {exc}"
+
     async def _handle_insights_command(self, event: MessageEvent) -> str:
         """Handle /insights command -- show usage insights and analytics."""
         args = event.get_command_args().strip()
diff --git a/hermes_cli/codex_usage.py b/hermes_cli/codex_usage.py
new file mode 100644
index 000000000000..0864eb4982c5
--- /dev/null
+++ b/hermes_cli/codex_usage.py
@@ -0,0 +1,455 @@
+"""OpenAI Codex quota usage helpers for Hermes.
+
+This module powers both the `hermes codex-usage` CLI command and local watchdog
+scripts. It reads Hermes' `openai-codex` credential pool and calls the Codex
+quota endpoint once per credential. Runtime OAuth tokens are never printed.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import urllib.error
+import urllib.request
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+from hermes_cli.config import get_hermes_home
+
+USAGE_URL = "https://chatgpt.com/backend-api/wham/usage"
+DEFAULT_WATERMARK = get_hermes_home() / "state" / "codex_usage_alerts.json"
+
+
+def local_dt(epoch: Any) -> Optional[datetime]:
+    if isinstance(epoch, str):
+        try:
+            return datetime.fromisoformat(epoch).astimezone()
+        except ValueError:
+            return None
+    if not isinstance(epoch, (int, float)):
+        return None
+    return datetime.fromtimestamp(epoch).astimezone()
+
+
+def human_delta(target: Optional[datetime], now: Optional[datetime] = None) -> str:
+    if target is None:
+        return "unknown"
+    now = now or datetime.now().astimezone()
+    seconds = int((target - now).total_seconds())
+    if seconds <= 0:
+        return "already reset"
+    days, rem = divmod(seconds, 86400)
+    hours, rem = divmod(rem, 3600)
+    minutes, _ = divmod(rem, 60)
+    parts: list[str] = []
+    if days:
+        parts.append(f"{days}d")
+    if hours:
+        parts.append(f"{hours}h")
+    if minutes or not parts:
+        parts.append(f"{minutes}m")
+    return " ".join(parts)
+
+
+def risk(used_percent: Any) -> Dict[str, str]:
+    try:
+        value = float(used_percent)
+    except (TypeError, ValueError):
+        return {"level": "unknown", "icon": "⚪", "label": "unknown"}
+    if value >= 95:
+        return {"level": "critical", "icon": "🔴", "label": "거의 소진"}
+    if value >= 80:
+        return {"level": "warning", "icon": "🟠", "label": "주의"}
+    return {"level": "ok", "icon": "🟢", "label": "정상"}
+
+
+def summarize_window(window: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+    if not isinstance(window, dict):
+        return None
+    reset_dt = local_dt(window.get("reset_at"))
+    used = window.get("used_percent")
+    return {
+        "used_percent": used,
+        "limit_window_seconds": window.get("limit_window_seconds"),
+        "reset_at": reset_dt.isoformat(timespec="seconds") if reset_dt else window.get("reset_at"),
+        "remaining": human_delta(reset_dt),
+        "risk": risk(used),
+    }
+
+
+# Backward-compatible alias used by the first local helper script.
+window_summary = summarize_window
+
+
+def fetch_usage(access_token: str, account_id: Optional[str] = None, timeout: int = 30) -> Dict[str, Any]:
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "User-Agent": "Hermes Codex usage check",
+        "Accept": "application/json",
+    }
+    if account_id:
+        headers["ChatGPT-Account-Id"] = account_id
+    req = urllib.request.Request(USAGE_URL, headers=headers, method="GET")
+    with urllib.request.urlopen(req, timeout=timeout) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def collect() -> Dict[str, Any]:
+    from agent.credential_pool import load_pool
+
+    pool = load_pool("openai-codex")
+    # clear_expired=True + refresh=True lets Hermes refresh OAuth tokens if needed.
+    entries = pool._available_entries(clear_expired=True, refresh=True)  # intentional internal API
+    now = datetime.now().astimezone()
+    rows: list[dict[str, Any]] = []
+    for entry in entries:
+        row: Dict[str, Any] = {
+            "label": entry.label,
+            "priority": entry.priority,
+            "source": entry.source,
+            "last_status": entry.last_status or "ok",
+        }
+        account_id = (
+            entry.extra.get("account_id")
+            or entry.extra.get("chatgpt_account_id")
+            or entry.extra.get("accountId")
+        )
+        try:
+            data = fetch_usage(entry.runtime_api_key, account_id=account_id)
+            rate_limit = data.get("rate_limit") or {}
+            credits = data.get("credits") or {}
+            row.update(
+                {
+                    "ok": True,
+                    "plan_type": data.get("plan_type"),
+                    "primary_window": summarize_window(rate_limit.get("primary_window")),
+                    "secondary_window": summarize_window(rate_limit.get("secondary_window")),
+                    "credits": {
+                        key: credits.get(key)
+                        for key in ("has_credits", "unlimited", "balance")
+                        if key in credits
+                    },
+                }
+            )
+        except urllib.error.HTTPError as exc:
+            body = exc.read().decode("utf-8", "replace")[:500]
+            row.update({"ok": False, "http": exc.code, "error": body})
+        except Exception as exc:  # noqa: BLE001 - CLI report should survive per-account failures
+            row.update({"ok": False, "error": f"{type(exc).__name__}: {exc}"})
+        rows.append(row)
+    payload = {"checked_at": now.isoformat(timespec="seconds"), "provider": "openai-codex", "accounts": rows}
+    payload["recommendation"] = compute_recommendation(rows)
+    return payload
+
+
+def _percent(row: Dict[str, Any], window_name: str) -> float:
+    try:
+        value = (row.get(window_name) or {}).get("used_percent")
+        if value is None:
+            return 999.0
+        return float(value)
+    except (TypeError, ValueError):
+        return 999.0
+
+
+def compute_recommendation(accounts: List[Dict[str, Any]]) -> Optional[Dict[str, str]]:
+    ok_accounts = [row for row in accounts if row.get("ok")]
+    if not ok_accounts:
+        return None
+    best = min(ok_accounts, key=lambda row: (_percent(row, "secondary_window"), _percent(row, "primary_window")))
+    sec = best.get("secondary_window") or {}
+    pri = best.get("primary_window") or {}
+    reason = f"7d {sec.get('used_percent', '?')}%, 5h {pri.get('used_percent', '?')}%"
+    return {"label": str(best.get("label")), "reason": reason}
+
+
+# Backward-compatible alias.
+recommend = compute_recommendation
+
+
+def iter_windows(row: Dict[str, Any]) -> Iterable[Tuple[str, str, Dict[str, Any]]]:
+    yield "5h", "primary_window", row.get("primary_window") or {}
+    yield "7d", "secondary_window", row.get("secondary_window") or {}
+
+
+def render_text(payload: Dict[str, Any]) -> str:
+    lines = [f"Codex usage ({payload['checked_at']})"]
+    accounts = payload.get("accounts") or []
+    if not accounts:
+        lines.append("No available openai-codex credentials found.")
+        return "\n".join(lines)
+    recommendation = payload.get("recommendation") or {}
+    if recommendation:
+        lines.append(f"추천: {recommendation.get('label')} ({recommendation.get('reason')})")
+    for row in accounts:
+        lines.append("")
+        lines.append(f"[{row.get('label')}] plan={row.get('plan_type', 'unknown')} status={row.get('last_status', 'unknown')}")
+        if not row.get("ok"):
+            lines.append(f"  ERROR: {row.get('http', '')} {row.get('error', '')}".rstrip())
+            continue
+        for display, _key, window in iter_windows(row):
+            r = window.get("risk") or risk(window.get("used_percent"))
+            lines.append(
+                f"  {display}: {r.get('icon')} "
+                f"{window.get('used_percent', '?')}% used ({r.get('label')}), "
+                f"reset {window.get('reset_at', '?')} ({window.get('remaining', '?')})"
+            )
+        credits = row.get("credits") or {}
+        if credits:
+            lines.append(
+                "  credits: "
+                f"has={credits.get('has_credits')}, "
+                f"unlimited={credits.get('unlimited')}, "
+                f"balance={credits.get('balance')}"
+            )
+    return "\n".join(lines)
+
+
+def short_reset(reset_at: Any) -> str:
+    if not isinstance(reset_at, str):
+        return "?"
+    try:
+        dt = datetime.fromisoformat(reset_at)
+        return dt.strftime("%m/%d %H:%M")
+    except ValueError:
+        return reset_at
+
+
+def short_checked_at(checked_at: Any) -> str:
+    if not isinstance(checked_at, str):
+        return "?"
+    try:
+        return datetime.fromisoformat(checked_at).strftime("%m/%d %H:%M")
+    except ValueError:
+        return checked_at
+
+
+def usage_bar(used_percent: Any, width: int = 10) -> str:
+    """Return a compact text progress bar for Telegram/CLI scanning."""
+    try:
+        value = max(0.0, min(100.0, float(used_percent)))
+    except (TypeError, ValueError):
+        return "?" * max(1, width)
+    filled = int(value / 100 * width)
+    if value > 0 and filled == 0:
+        filled = 1
+    if value >= 95:
+        filled = width
+    return "█" * filled + "░" * (width - filled)
+
+
+def _fmt_percent(value: Any) -> str:
+    try:
+        f = float(value)
+    except (TypeError, ValueError):
+        return " ?%"
+    if f.is_integer():
+        return f"{int(f):>2}%"
+    return f"{f:>4.1f}%"
+
+
+def _worst_risk(accounts: list[dict[str, Any]]) -> dict[str, str]:
+    order = {"critical": 3, "warning": 2, "ok": 1, "unknown": 0}
+    worst = {"level": "unknown", "icon": "⚪", "label": "unknown"}
+    for row in accounts:
+        if not row.get("ok"):
+            return {"level": "error", "icon": "❌", "label": "조회 실패"}
+        for _display, _key, window in iter_windows(row):
+            r = window.get("risk") or risk(window.get("used_percent"))
+            if order.get(r.get("level", "unknown"), 0) > order.get(worst.get("level", "unknown"), 0):
+                worst = r
+    return worst
+
+
+def render_compact(payload: Dict[str, Any]) -> str:
+    accounts = payload.get("accounts") or []
+    worst = _worst_risk(accounts)
+    lines = [f"🧭 Codex 사용량 · {short_checked_at(payload.get('checked_at'))} · {worst.get('icon')} {worst.get('label')}"]
+    recommendation = payload.get("recommendation") or {}
+    if recommendation:
+        lines.append(f"✅ 추천 {recommendation.get('label')} — {recommendation.get('reason')}")
+    if not accounts:
+        lines.append("계정 없음")
+        return "\n".join(lines)
+    for row in accounts:
+        label = row.get("label")
+        plan = row.get("plan_type") or "unknown"
+        if not row.get("ok"):
+            lines.append(f"\n❌ {label} · {plan}")
+            lines.append(f"└ ERROR {row.get('http', '')} {row.get('error', '')}".rstrip())
+            continue
+        lines.append(f"\n• {label} · {plan}")
+        for display, _key, window in iter_windows(row):
+            r = window.get("risk") or risk(window.get("used_percent"))
+            used = window.get("used_percent")
+            lines.append(
+                f"  {display:>2} {_fmt_percent(used)} {r.get('icon')} "
+                f"[{usage_bar(used)}] reset {short_reset(window.get('reset_at'))} · {window.get('remaining', '?')}"
+            )
+    return "\n".join(lines)
+
+
+def load_seen(path: Path) -> set[str]:
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        if isinstance(data, list):
+            return {str(item) for item in data}
+    except FileNotFoundError:
+        return set()
+    except Exception:
+        return set()
+    return set()
+
+
+def save_seen(path: Path, seen: set[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(sorted(seen), ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def apply_alert_policy(
+    payload: Dict[str, Any],
+    *,
+    threshold: Optional[float] = None,
+    primary_threshold: Optional[float] = None,
+    secondary_threshold: Optional[float] = None,
+) -> List[Dict[str, Any]]:
+    """Return alert events from a collected usage payload.
+
+    `threshold` is the legacy single threshold. `primary_threshold` controls the
+    5h window and `secondary_threshold` controls the 7d window. Missing specific
+    thresholds fall back to `threshold`, then 90%.
+    """
+    fallback = 90.0 if threshold is None else float(threshold)
+    primary = float(primary_threshold) if primary_threshold is not None else fallback
+    secondary = float(secondary_threshold) if secondary_threshold is not None else fallback
+    thresholds = {"primary_window": primary, "secondary_window": secondary}
+
+    events: list[dict[str, Any]] = []
+    for row in payload.get("accounts") or []:
+        if not row.get("ok"):
+            events.append({"key": f"error:{row.get('label')}:{row.get('http')}:{row.get('error')}", "row": row, "error": True})
+            continue
+        for display, key, window in iter_windows(row):
+            try:
+                raw_used = window.get("used_percent")
+                if raw_used is None:
+                    continue
+                used = float(raw_used)
+            except (TypeError, ValueError):
+                continue
+            if used >= thresholds[key]:
+                reset_key = str(window.get("reset_at"))[:16]
+                events.append(
+                    {
+                        "key": f"quota:{row.get('label')}:{key}:{reset_key}",
+                        "label": row.get("label"),
+                        "window": display,
+                        "window_key": key,
+                        "threshold": thresholds[key],
+                        "used": used if used % 1 else int(used),
+                        "reset_at": window.get("reset_at"),
+                        "remaining": window.get("remaining"),
+                        "risk": window.get("risk") or risk(used),
+                    }
+                )
+    return events
+
+
+def alert_events(payload: Dict[str, Any], threshold: float) -> List[Dict[str, Any]]:
+    return apply_alert_policy(payload, threshold=threshold)
+
+
+def render_alert(
+    payload: Dict[str, Any],
+    threshold: Optional[float] = None,
+    watermark: Optional[Path] = None,
+    *,
+    primary_threshold: Optional[float] = None,
+    secondary_threshold: Optional[float] = None,
+) -> str:
+    events = apply_alert_policy(
+        payload,
+        threshold=threshold,
+        primary_threshold=primary_threshold,
+        secondary_threshold=secondary_threshold,
+    )
+    if watermark:
+        seen = load_seen(watermark)
+        fresh = [event for event in events if event["key"] not in seen]
+        if fresh:
+            seen.update(event["key"] for event in fresh)
+            save_seen(watermark, seen)
+        events = fresh
+    if not events:
+        return ""
+    lines = [f"⚠️ Codex usage alert ({payload['checked_at']})"]
+    for event in events:
+        if event.get("error"):
+            row = event["row"]
+            lines.append(f"{row.get('label')}: ERROR {row.get('http', '')} {row.get('error', '')}".rstrip())
+            continue
+        r = event.get("risk") or {}
+        lines.append(
+            f"{r.get('icon', '⚠️')} {event.get('label')} {event.get('window')} "
+            f"{event.get('used'):g}% >= {event.get('threshold'):g}% — "
+            f"reset {event.get('reset_at')} ({event.get('remaining')})"
+        )
+    recommendation = payload.get("recommendation") or {}
+    if recommendation:
+        lines.append(f"추천: {recommendation.get('label')} — {recommendation.get('reason')}")
+    return "\n".join(lines)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Show Hermes OpenAI Codex quota usage by credential")
+    parser.add_argument("--json", action="store_true", help="print machine-readable JSON")
+    parser.add_argument("--compact", action="store_true", help="print Telegram-friendly compact output (default)")
+    parser.add_argument("--verbose", action="store_true", help="print detailed per-account text output")
+    parser.add_argument("--alert-threshold", type=float, help="legacy: print only accounts/windows at or above this percent")
+    parser.add_argument("--primary-threshold", type=float, help="5h window alert threshold percent")
+    parser.add_argument("--secondary-threshold", type=float, help="7d window alert threshold percent")
+    parser.add_argument("--quiet-ok", action="store_true", help="with alert thresholds, print nothing when below threshold")
+    parser.add_argument(
+        "--watermark",
+        type=Path,
+        nargs="?",
+        const=DEFAULT_WATERMARK,
+        help="with alert thresholds, suppress duplicate alerts for the same reset window",
+    )
+    return parser
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    payload = collect()
+    alert_mode = any(
+        value is not None
+        for value in (args.alert_threshold, args.primary_threshold, args.secondary_threshold)
+    )
+    if args.json:
+        print(json.dumps(payload, ensure_ascii=False, indent=2))
+    elif alert_mode:
+        output = render_alert(
+            payload,
+            args.alert_threshold,
+            args.watermark,
+            primary_threshold=args.primary_threshold,
+            secondary_threshold=args.secondary_threshold,
+        )
+        if output:
+            print(output)
+        elif not args.quiet_ok:
+            primary = args.primary_threshold if args.primary_threshold is not None else args.alert_threshold or 90
+            secondary = args.secondary_threshold if args.secondary_threshold is not None else args.alert_threshold or 90
+            print(f"Codex usage OK: 5h below {primary:g}%, 7d below {secondary:g}%")
+    elif args.verbose:
+        print(render_text(payload))
+    else:
+        print(render_compact(payload))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))
diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py
index 3e2d03dc3580..a462a86c6bc4 100644
--- a/hermes_cli/commands.py
+++ b/hermes_cli/commands.py
@@ -229,6 +229,8 @@ class CommandDef:
     CommandDef("restart", "Gracefully restart the gateway after draining active runs", "Session",
                gateway_only=True),
     CommandDef("usage", "Show token usage and rate limits for the current session", "Info"),
+    CommandDef("codex-usage", "Show current OpenAI Codex quota usage by credential", "Info",
+               aliases=("codex_usage",), args_hint="[--compact|--verbose]"),
     CommandDef("credits", "Show Nous credit balance and top up", "Info"),
     CommandDef("billing", "Manage Nous terminal billing — buy credits, auto-reload, limits", "Info",
                cli_only=True),
diff --git a/hermes_cli/main.py b/hermes_cli/main.py
index caca5e6a8a37..01f83f3ea197 100644
--- a/hermes_cli/main.py
+++ b/hermes_cli/main.py
@@ -12639,8 +12639,34 @@ def cmd_insights(args):
 
         db = SessionDB()
         engine = InsightsEngine(db)
-        report = engine.generate(days=args.days, source=args.source)
-        print(engine.format_terminal(report))
+        if getattr(args, "by_credential", False):
+            provider = getattr(args, "provider", None)
+            rows = db.get_credential_usage(days=args.days, provider=provider)
+            if rows:
+                title_provider = f" for {provider}" if provider else ""
+                print(f"Credential usage{title_provider} (last {args.days}d)")
+                for row in rows:
+                    label = row.get("credential_label") or "unknown"
+                    model = row.get("model") or "unknown"
+                    total = row.get("total_tokens") or 0
+                    in_tok = row.get("input_tokens") or 0
+                    out_tok = row.get("output_tokens") or 0
+                    calls = row.get("api_calls") or 0
+                    print(
+                        f"  {label} · {model}: {total:,} tokens "
+                        f"({in_tok:,} in, {out_tok:,} out), {calls:,} calls"
+                    )
+            else:
+                suffix = f" for {provider}" if provider else ""
+                print(
+                    f"No credential-level usage rows{suffix} in the last {args.days}d. "
+                    "New rows are recorded only after this Hermes version handles model turns."
+                )
+        else:
+            report = engine.generate(days=args.days, source=args.source)
+            if getattr(args, "provider", None):
+                report["provider_filter_note"] = args.provider
+            print(engine.format_terminal(report))
         db.close()
     except Exception as e:
         print(f"Error generating insights: {e}")
@@ -13936,6 +13962,37 @@ def cmd_sessions(args):
     # =========================================================================
     build_insights_parser(subparsers, cmd_insights=cmd_insights)
 
+    codex_usage_parser = subparsers.add_parser(
+        "codex-usage",
+        help="Show OpenAI Codex quota usage for Hermes OAuth credentials",
+        description="Show current OpenAI Codex quota, reset windows, alerts, and recommended credential.",
+    )
+    codex_usage_parser.add_argument("--json", action="store_true", help="print machine-readable JSON")
+    codex_usage_parser.add_argument("--compact", action="store_true", help="print Telegram-friendly compact output (default)")
+    codex_usage_parser.add_argument("--verbose", action="store_true", help="print detailed per-account text output")
+    codex_usage_parser.add_argument("--alert-threshold", type=float, help="legacy: alert when either window is at/above this percent")
+    codex_usage_parser.add_argument("--primary-threshold", type=float, help="5h window alert threshold percent")
+    codex_usage_parser.add_argument("--secondary-threshold", type=float, help="7d window alert threshold percent")
+    codex_usage_parser.add_argument("--quiet-ok", action="store_true", help="with alert thresholds, print nothing when below threshold")
+    codex_usage_parser.add_argument("--watermark", nargs="?", const="__DEFAULT__", help="suppress duplicate alerts for the same reset window")
+    def _cmd_codex_usage(_args):
+        from hermes_cli.codex_usage import main as _codex_usage_main
+        argv = []
+        for _flag in ("json", "compact", "verbose", "quiet_ok"):
+            if getattr(_args, _flag, False):
+                argv.append("--" + _flag.replace("_", "-"))
+        for _flag in ("alert_threshold", "primary_threshold", "secondary_threshold"):
+            _val = getattr(_args, _flag, None)
+            if _val is not None:
+                argv.extend(["--" + _flag.replace("_", "-"), str(_val)])
+        _wm = getattr(_args, "watermark", None)
+        if _wm is not None:
+            argv.append("--watermark")
+            if _wm != "__DEFAULT__":
+                argv.append(str(_wm))
+        return _codex_usage_main(argv)
+    codex_usage_parser.set_defaults(func=_cmd_codex_usage)
+
     # =========================================================================
     # claw command  (parser built in hermes_cli/subcommands/claw.py)
     # =========================================================================
diff --git a/hermes_cli/subcommands/insights.py b/hermes_cli/subcommands/insights.py
index 42746e8030b0..b577d884946c 100644
--- a/hermes_cli/subcommands/insights.py
+++ b/hermes_cli/subcommands/insights.py
@@ -22,4 +22,12 @@ def build_insights_parser(subparsers, *, cmd_insights: Callable) -> None:
     insights_parser.add_argument(
         "--source", help="Filter by platform (cli, telegram, discord, etc.)"
     )
+    insights_parser.add_argument(
+        "--provider", help="Filter usage by billing provider (e.g. openai-codex)"
+    )
+    insights_parser.add_argument(
+        "--by-credential",
+        action="store_true",
+        help="Show per-credential token usage when credential telemetry is available",
+    )
     insights_parser.set_defaults(func=cmd_insights)
diff --git a/hermes_state.py b/hermes_state.py
index a2895b09c7a2..720c06c330eb 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -767,6 +767,21 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
     compacted INTEGER NOT NULL DEFAULT 0
 );
 
+CREATE TABLE IF NOT EXISTS credential_usage (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    session_id TEXT NOT NULL REFERENCES sessions(id),
+    timestamp REAL NOT NULL,
+    provider TEXT,
+    credential_label TEXT,
+    model TEXT,
+    input_tokens INTEGER DEFAULT 0,
+    output_tokens INTEGER DEFAULT 0,
+    cache_read_tokens INTEGER DEFAULT 0,
+    cache_write_tokens INTEGER DEFAULT 0,
+    reasoning_tokens INTEGER DEFAULT 0,
+    api_call_count INTEGER DEFAULT 0
+);
+
 CREATE TABLE IF NOT EXISTS state_meta (
     key TEXT PRIMARY KEY,
     value TEXT
@@ -808,6 +823,8 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
     ON sessions(source, user_id, chat_id, chat_type, thread_id, started_at DESC);
 CREATE INDEX IF NOT EXISTS idx_sessions_handoff_state
     ON sessions(handoff_state, started_at);
+CREATE INDEX IF NOT EXISTS idx_credential_usage_time_provider
+    ON credential_usage(timestamp DESC, provider, credential_label);
 """
 
 FTS_SQL = """
@@ -2414,6 +2431,7 @@ def update_token_counts(
         billing_provider: Optional[str] = None,
         billing_base_url: Optional[str] = None,
         billing_mode: Optional[str] = None,
+        credential_label: Optional[str] = None,
         api_call_count: int = 0,
         absolute: bool = False,
     ) -> None:
@@ -2494,8 +2512,83 @@ def update_token_counts(
         )
         def _do(conn):
             conn.execute(sql, params)
+            if credential_label:
+                conn.execute(
+                    """INSERT INTO credential_usage (
+                       session_id, timestamp, provider, credential_label, model,
+                       input_tokens, output_tokens, cache_read_tokens,
+                       cache_write_tokens, reasoning_tokens, api_call_count
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                    (
+                        session_id,
+                        time.time(),
+                        billing_provider,
+                        credential_label,
+                        model,
+                        input_tokens,
+                        output_tokens,
+                        cache_read_tokens,
+                        cache_write_tokens,
+                        reasoning_tokens,
+                        api_call_count,
+                    ),
+                )
         self._execute_write(_do)
 
+    def get_credential_usage(
+        self,
+        days: int = 30,
+        provider: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """Aggregate per-credential token usage rows for recent model calls."""
+        cutoff = time.time() - (max(0, int(days)) * 86400)
+        where = ["timestamp >= ?"]
+        params: list[Any] = [cutoff]
+        if provider:
+            where.append("provider = ?")
+            params.append(provider)
+        query = f"""
+            SELECT
+                COALESCE(provider, 'unknown') AS provider,
+                COALESCE(credential_label, 'unknown') AS credential_label,
+                COALESCE(model, 'unknown') AS model,
+                COALESCE(SUM(api_call_count), 0) AS api_calls,
+                COALESCE(SUM(input_tokens), 0) AS input_tokens,
+                COALESCE(SUM(output_tokens), 0) AS output_tokens,
+                COALESCE(SUM(cache_read_tokens), 0) AS cache_read_tokens,
+                COALESCE(SUM(cache_write_tokens), 0) AS cache_write_tokens,
+                COALESCE(SUM(reasoning_tokens), 0) AS reasoning_tokens
+            FROM credential_usage
+            WHERE {' AND '.join(where)}
+            GROUP BY provider, credential_label, model
+            ORDER BY (COALESCE(SUM(input_tokens), 0) + COALESCE(SUM(output_tokens), 0)
+                      + COALESCE(SUM(cache_read_tokens), 0) + COALESCE(SUM(cache_write_tokens), 0)
+                      + COALESCE(SUM(reasoning_tokens), 0)) DESC,
+                     credential_label ASC,
+                     model ASC
+        """
+        with self._lock:
+            cursor = self._conn.execute(query, tuple(params))
+            rows = [dict(row) for row in cursor.fetchall()]
+        for row in rows:
+            row["api_calls"] = int(row.get("api_calls") or 0)
+            for key in (
+                "input_tokens",
+                "output_tokens",
+                "cache_read_tokens",
+                "cache_write_tokens",
+                "reasoning_tokens",
+            ):
+                row[key] = int(row.get(key) or 0)
+            row["total_tokens"] = (
+                row["input_tokens"]
+                + row["output_tokens"]
+                + row["cache_read_tokens"]
+                + row["cache_write_tokens"]
+                + row["reasoning_tokens"]
+            )
+        return rows
+
     def ensure_session(
         self,
         session_id: str,
diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 239b26de1712..19f671557851 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -154,6 +154,10 @@ async def _shutdown_abandoned_app(app) -> None:
 
 try:
     from telegram import Update, Bot, Message, InlineKeyboardButton, InlineKeyboardMarkup
+    try:
+        from telegram import CopyTextButton
+    except ImportError:
+        CopyTextButton = None
     try:
         from telegram import LinkPreviewOptions
     except ImportError:
@@ -176,6 +180,7 @@ async def _shutdown_abandoned_app(app) -> None:
     Message = Any
     InlineKeyboardButton = Any
     InlineKeyboardMarkup = Any
+    CopyTextButton = None
     LinkPreviewOptions = None
     Application = Any
     CommandHandler = Any
@@ -251,7 +256,7 @@ def check_telegram_requirements() -> bool:
     so the adapter's class-level type aliases get rebound.
     """
     global TELEGRAM_AVAILABLE, Update, Bot, Message, InlineKeyboardButton
-    global InlineKeyboardMarkup, LinkPreviewOptions, Application
+    global InlineKeyboardMarkup, CopyTextButton, LinkPreviewOptions, Application
     global CommandHandler, CallbackQueryHandler, TelegramMessageHandler
     global ContextTypes, filters, ParseMode, ChatType, HTTPXRequest
     if TELEGRAM_AVAILABLE:
@@ -264,6 +269,10 @@ def check_telegram_requirements() -> bool:
     try:
         from telegram import Update as _Update, Bot as _Bot, Message as _Message
         from telegram import InlineKeyboardButton as _IKB, InlineKeyboardMarkup as _IKM
+        try:
+            from telegram import CopyTextButton as _CTB
+        except ImportError:
+            _CTB = None
         try:
             from telegram import LinkPreviewOptions as _LPO
         except ImportError:
@@ -283,6 +292,7 @@ def check_telegram_requirements() -> bool:
     Message = _Message
     InlineKeyboardButton = _IKB
     InlineKeyboardMarkup = _IKM
+    CopyTextButton = _CTB
     LinkPreviewOptions = _LPO
     Application = _App
     CommandHandler = _CH
@@ -347,6 +357,48 @@ def _separate_chunk_indicator_from_fence(text: str) -> str:
     return _CHUNK_INDICATOR_ON_FENCE_RE.sub(r'```\n\g<indicator>', text)
 
 
+# ---------------------------------------------------------------------------
+# Telegram copy-to-clipboard buttons
+# ---------------------------------------------------------------------------
+
+_COPY_BUTTON_LINE_RE = re.compile(
+    r'^\s*COPY_BUTTON:\s*(?P<label>[^|\n]{1,64})\|(?P<text>[^\n]{1,256})\s*$',
+    re.MULTILINE,
+)
+
+
+def _extract_copy_buttons(content: str) -> tuple[str, list[tuple[str, str]]]:
+    """Strip ``COPY_BUTTON: label | text`` marker lines and return buttons.
+
+    Telegram's Bot API exposes ``InlineKeyboardButton.copy_text`` for true
+    clipboard-copy buttons, but Hermes final answers are plain markdown.  This
+    narrowly-scoped marker gives the agent a way to request copy buttons on
+    Telegram without adding a new model tool or changing other platforms.
+
+    Invalid/oversized marker lines are left untouched so no user-visible text is
+    silently dropped.  Valid button copy payloads are limited to Telegram's
+    documented 1-256 character range; labels are capped to avoid mobile
+    truncation-heavy keyboards.  The adapter caps the rendered keyboard to the
+    first 20 buttons as a conservative Bot API/client guardrail.
+    """
+    if not content or "COPY_BUTTON:" not in content:
+        return content, []
+
+    buttons: list[tuple[str, str]] = []
+
+    def _replace(match: re.Match[str]) -> str:
+        label = match.group("label").strip()
+        text = match.group("text").strip()
+        if not label or not text or len(text) > 256 or len(buttons) >= 20:
+            return match.group(0)
+        buttons.append((label, text))
+        return ""
+
+    stripped = _COPY_BUTTON_LINE_RE.sub(_replace, content)
+    stripped = re.sub(r'\n{3,}', '\n\n', stripped).strip()
+    return stripped, buttons
+
+
 # ---------------------------------------------------------------------------
 # Markdown table → Telegram-friendly row groups
 # ---------------------------------------------------------------------------
@@ -1262,6 +1314,23 @@ def _link_preview_kwargs(self) -> Dict[str, Any]:
             return {"link_preview_options": LinkPreviewOptions(is_disabled=True)}
         return {"disable_web_page_preview": True}
 
+    def _copy_buttons_markup(self, buttons: list[tuple[str, str]]) -> Optional[Any]:
+        """Build Telegram copy-to-clipboard inline keyboard markup."""
+        if not buttons:
+            return None
+        if CopyTextButton is None:
+            logger.warning("[%s] copy_text buttons requested but CopyTextButton is unavailable", self.name)
+            return None
+        rows = []
+        for label, text in buttons[:20]:
+            rows.append([
+                InlineKeyboardButton(
+                    label,
+                    copy_text=CopyTextButton(text=text),  # type: ignore[reportCallIssue]
+                )
+            ])
+        return InlineKeyboardMarkup(rows)  # type: ignore[reportCallIssue]
+
     # ------------------------------------------------------------------
     # Bot API 10.1 Rich Messages (sendRichMessage)
     #
@@ -3495,14 +3564,22 @@ async def send(
         # Skip whitespace-only text to prevent Telegram 400 empty-text errors.
         if not content or not content.strip():
             return SendResult(success=True, message_id=None)
+
+        copy_markup = None
+        if CopyTextButton is not None:
+            content, copy_buttons = _extract_copy_buttons(content)
+            if copy_buttons and not content.strip():
+                content = "Copy buttons"
+            copy_markup = self._copy_buttons_markup(copy_buttons)
+        elif "COPY_BUTTON:" in content:
+            logger.warning("[%s] COPY_BUTTON markers requested but CopyTextButton is unavailable", self.name)
         
         try:
-            # Bot API 10.1 rich fast-path: send the raw agent markdown via
-            # sendRichMessage so tables/task lists/etc. render natively. Falls
-            # through to the legacy MarkdownV2 path on permanent/capability
-            # errors or DM-topic routing skips; returns directly on success or
-            # on a transient failure (which must NOT be legacy-resent).
-            if self._should_attempt_rich(content, metadata=metadata):
+            # Bot API 10.1 rich messages currently use the raw sendRichMessage
+            # endpoint path, which does not attach reply_markup here.  If copy
+            # buttons were requested, stay on the legacy sendMessage path so the
+            # clipboard keyboard is delivered with the text.
+            if not copy_markup and self._should_attempt_rich(content, metadata=metadata):
                 rich_result = await self._try_send_rich(chat_id, content, reply_to, metadata)
                 if rich_result is not None:
                     if rich_result.success:
@@ -3611,6 +3688,7 @@ async def send(
                                 **thread_kwargs,
                                 **self._link_preview_kwargs(),
                                 **self._notification_kwargs(metadata),
+                                **({"reply_markup": copy_markup} if copy_markup is not None and i == len(chunks) - 1 else {}),
                             )
                         except Exception as md_error:
                             # Markdown parsing failed, try plain text
@@ -3625,6 +3703,7 @@ async def send(
                                     **thread_kwargs,
                                     **self._link_preview_kwargs(),
                                     **self._notification_kwargs(metadata),
+                                    **({"reply_markup": copy_markup} if copy_markup is not None and i == len(chunks) - 1 else {}),
                                 )
                             else:
                                 raise
@@ -3860,6 +3939,15 @@ async def edit_message(
         if not self._bot:
             return SendResult(success=False, error="Not connected")
 
+        copy_markup = None
+        if finalize and CopyTextButton is not None:
+            content, copy_buttons = _extract_copy_buttons(content)
+            if copy_buttons and not content.strip():
+                content = "Copy buttons"
+            copy_markup = self._copy_buttons_markup(copy_buttons)
+        elif finalize and "COPY_BUTTON:" in content:
+            logger.warning("[%s] COPY_BUTTON markers requested but CopyTextButton is unavailable", self.name)
+
         # Rich finalize (Bot API 10.1): when the completed content has
         # constructs the legacy MarkdownV2 edit degrades (tables → bullet
         # lists, task lists, <details>, block math) and rich is available,
@@ -3870,7 +3958,7 @@ async def edit_message(
         # table that exceeds the MarkdownV2 limit must not be split into legacy
         # chunks.  Falls back to the legacy edit path (overflow split included)
         # on capability/permanent rejection.
-        if finalize and self._rich_eligible(content):
+        if finalize and not copy_markup and self._rich_eligible(content):
             rich_result = await self._try_edit_rich(
                 chat_id, message_id, content, metadata=metadata,
             )
@@ -3929,6 +4017,7 @@ async def edit_message(
                     message_id=int(message_id),
                     text=formatted,
                     parse_mode=ParseMode.MARKDOWN_V2,
+                    **({"reply_markup": copy_markup} if copy_markup is not None else {}),
                 )
             except Exception as fmt_err:
                 # "Message is not modified" is a no-op, not an error
@@ -3946,6 +4035,7 @@ async def edit_message(
                     chat_id=normalize_telegram_chat_id(chat_id),
                     message_id=int(message_id),
                     text=_plain,
+                    **({"reply_markup": copy_markup} if copy_markup is not None else {}),
                 )
             return SendResult(success=True, message_id=message_id)
         except Exception as e:
diff --git a/tests/gateway/test_telegram_copy_buttons.py b/tests/gateway/test_telegram_copy_buttons.py
new file mode 100644
index 000000000000..feb557a2bcd9
--- /dev/null
+++ b/tests/gateway/test_telegram_copy_buttons.py
@@ -0,0 +1,202 @@
+"""Tests for Telegram copy-to-clipboard inline buttons."""
+
+import sys
+from pathlib import Path
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+_repo = str(Path(__file__).resolve().parents[2])
+if _repo not in sys.path:
+    sys.path.insert(0, _repo)
+
+
+def _ensure_telegram_mock():
+    try:
+        import telegram  # noqa: F401
+        if hasattr(telegram, "__file__"):
+            return
+    except ImportError:
+        pass
+
+    if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
+        return
+
+    mod = MagicMock()
+    mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
+    mod.constants.ParseMode.MARKDOWN = "Markdown"
+    mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
+    mod.constants.ParseMode.HTML = "HTML"
+    mod.constants.ChatType.PRIVATE = "private"
+    mod.constants.ChatType.GROUP = "group"
+    mod.constants.ChatType.SUPERGROUP = "supergroup"
+    mod.constants.ChatType.CHANNEL = "channel"
+    mod.error.NetworkError = type("NetworkError", (OSError,), {})
+    mod.error.TimedOut = type("TimedOut", (OSError,), {})
+    mod.error.BadRequest = type("BadRequest", (Exception,), {})
+
+    for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
+        sys.modules[name] = mod
+    sys.modules["telegram.error"] = mod.error
+
+
+_ensure_telegram_mock()
+
+from gateway.config import PlatformConfig
+from plugins.platforms.telegram import adapter as telegram_adapter
+from plugins.platforms.telegram.adapter import TelegramAdapter, _extract_copy_buttons
+
+
+@pytest.fixture(autouse=True)
+def _unload_adapter_after_test():
+    """Let later tests that inject fake telegram modules import a fresh adapter."""
+    yield
+    sys.modules.pop("plugins.platforms.telegram.adapter", None)
+    pkg = sys.modules.get("plugins.platforms.telegram")
+    if pkg is not None and getattr(pkg, "adapter", None) is telegram_adapter:
+        try:
+            delattr(pkg, "adapter")
+        except AttributeError:
+            pass
+
+
+class FakeCopyTextButton:
+    def __init__(self, text: str):
+        self.text = text
+
+
+class FakeInlineKeyboardButton:
+    def __init__(self, text: str, **kwargs):
+        self.text = text
+        self.kwargs = kwargs
+
+
+class FakeInlineKeyboardMarkup:
+    def __init__(self, rows):
+        self.rows = rows
+
+
+def _make_adapter(monkeypatch):
+    monkeypatch.setattr(telegram_adapter, "CopyTextButton", FakeCopyTextButton)
+    monkeypatch.setattr(telegram_adapter, "InlineKeyboardButton", FakeInlineKeyboardButton)
+    monkeypatch.setattr(telegram_adapter, "InlineKeyboardMarkup", FakeInlineKeyboardMarkup)
+
+    config = PlatformConfig(enabled=True, token="test-token", extra={})
+    adapter = TelegramAdapter(config)
+    adapter._bot = AsyncMock()
+    adapter._app = MagicMock()
+    return adapter
+
+
+def test_extract_copy_buttons_strips_valid_markers_only():
+    content = """VNC 주소
+COPY_BUTTON: Mac mini IP | 100.96.33.123
+body remains
+COPY_BUTTON: bad | 
+"""
+
+    stripped, buttons = _extract_copy_buttons(content)
+
+    assert buttons == [("Mac mini IP", "100.96.33.123")]
+    assert "COPY_BUTTON: Mac mini IP" not in stripped
+    assert "body remains" in stripped
+    assert "COPY_BUTTON: bad |" in stripped
+
+
+@pytest.mark.asyncio
+async def test_send_attaches_copy_text_keyboard_and_strips_markers(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    mock_msg = MagicMock()
+    mock_msg.message_id = 123
+    bot = cast(Any, adapter._bot)
+    bot.send_message = AsyncMock(return_value=mock_msg)
+
+    result = await adapter.send(
+        chat_id="12345",
+        content=(
+            "VNC 주소\n"
+            "COPY_BUTTON: Mac mini IP | 100.96.33.123\n"
+            "COPY_BUTTON: MacBook VNC | 100.99.47.75:5900"
+        ),
+    )
+
+    assert result.success is True
+    kwargs = bot.send_message.call_args.kwargs
+    assert "COPY_BUTTON:" not in kwargs["text"]
+    assert "VNC 주소" in kwargs["text"]
+
+    markup = kwargs["reply_markup"]
+    assert isinstance(markup, FakeInlineKeyboardMarkup)
+    assert len(markup.rows) == 2
+    first = markup.rows[0][0]
+    second = markup.rows[1][0]
+    assert first.text == "Mac mini IP"
+    assert first.kwargs["copy_text"].text == "100.96.33.123"
+    assert second.text == "MacBook VNC"
+    assert second.kwargs["copy_text"].text == "100.99.47.75:5900"
+
+
+@pytest.mark.asyncio
+async def test_copy_button_only_message_gets_safe_fallback_text(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    mock_msg = MagicMock()
+    mock_msg.message_id = 124
+    bot = cast(Any, adapter._bot)
+    bot.send_message = AsyncMock(return_value=mock_msg)
+
+    result = await adapter.send(
+        chat_id="12345",
+        content="COPY_BUTTON: MacBook IP | 100.99.47.75",
+    )
+
+    assert result.success is True
+    kwargs = bot.send_message.call_args.kwargs
+    assert kwargs["text"] == "Copy buttons"
+    assert kwargs["reply_markup"].rows[0][0].kwargs["copy_text"].text == "100.99.47.75"
+
+
+@pytest.mark.asyncio
+async def test_send_leaves_markers_visible_when_copy_text_button_unavailable(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    monkeypatch.setattr(telegram_adapter, "CopyTextButton", None)
+    mock_msg = MagicMock()
+    mock_msg.message_id = 125
+    bot = cast(Any, adapter._bot)
+    bot.send_message = AsyncMock(return_value=mock_msg)
+
+    result = await adapter.send(
+        chat_id="12345",
+        content="VNC\nCOPY_BUTTON: MacBook IP | 100.99.47.75",
+    )
+
+    assert result.success is True
+    kwargs = bot.send_message.call_args.kwargs
+    assert "COPY\\_BUTTON: MacBook IP \\| 100\\.99\\.47\\.75" in kwargs["text"]
+    assert "reply_markup" not in kwargs
+
+
+@pytest.mark.asyncio
+async def test_finalize_edit_attaches_copy_text_keyboard_and_strips_markers(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    bot = cast(Any, adapter._bot)
+    bot.edit_message_text = AsyncMock(return_value=None)
+
+    result = await adapter.edit_message(
+        chat_id="12345",
+        message_id="777",
+        content=(
+            "VNC 주소\n"
+            "COPY_BUTTON: Mac mini IP | 100.96.33.123\n"
+            "COPY_BUTTON: MacBook VNC | 100.99.47.75:5900"
+        ),
+        finalize=True,
+    )
+
+    assert result.success is True
+    kwargs = bot.edit_message_text.call_args.kwargs
+    assert kwargs["message_id"] == 777
+    assert "COPY_BUTTON:" not in kwargs["text"]
+    assert "VNC 주소" in kwargs["text"]
+    assert kwargs["reply_markup"].rows[0][0].kwargs["copy_text"].text == "100.96.33.123"
+    assert kwargs["reply_markup"].rows[1][0].kwargs["copy_text"].text == "100.99.47.75:5900"
diff --git a/tests/hermes_cli/test_codex_usage.py b/tests/hermes_cli/test_codex_usage.py
new file mode 100644
index 000000000000..ada20be4eb62
--- /dev/null
+++ b/tests/hermes_cli/test_codex_usage.py
@@ -0,0 +1,75 @@
+from hermes_cli.codex_usage import (
+    apply_alert_policy,
+    compute_recommendation,
+    render_compact,
+    summarize_window,
+    usage_bar,
+)
+
+
+def test_risk_policy_supports_per_window_thresholds():
+    payload = {
+        "checked_at": "2026-07-06T15:00:00+09:00",
+        "accounts": [
+            {
+                "label": "personal-backup",
+                "ok": True,
+                "primary_window": {"used_percent": 94, "reset_at": "2026-07-06T16:00:00+09:00"},
+                "secondary_window": {"used_percent": 86, "reset_at": "2026-07-07T10:00:00+09:00"},
+            },
+            {
+                "label": "company-plus-100",
+                "ok": True,
+                "primary_window": {"used_percent": 10, "reset_at": "2026-07-06T19:00:00+09:00"},
+                "secondary_window": {"used_percent": 54, "reset_at": "2026-07-09T18:00:00+09:00"},
+            },
+        ],
+    }
+
+    events = apply_alert_policy(payload, primary_threshold=95, secondary_threshold=85)
+
+    assert [event["label"] for event in events] == ["personal-backup"]
+    assert events[0]["window"] == "7d"
+    assert events[0]["used"] == 86
+
+
+def test_recommendation_prefers_lowest_weekly_then_five_hour_usage():
+    accounts = [
+        {"label": "a", "ok": True, "primary_window": {"used_percent": 1}, "secondary_window": {"used_percent": 70}},
+        {"label": "b", "ok": True, "primary_window": {"used_percent": 60}, "secondary_window": {"used_percent": 40}},
+    ]
+
+    rec = compute_recommendation(accounts)
+
+    assert rec == {"label": "b", "reason": "7d 40%, 5h 60%"}
+
+
+def test_render_compact_includes_risk_and_recommendation():
+    payload = {
+        "checked_at": "2026-07-06T15:00:00+09:00",
+        "accounts": [
+            {
+                "label": "company-plus-100",
+                "ok": True,
+                "primary_window": summarize_window({"used_percent": 1, "reset_at": 1783333044}),
+                "secondary_window": summarize_window({"used_percent": 54, "reset_at": 1783609614}),
+            }
+        ],
+        "recommendation": {"label": "company-plus-100", "reason": "7d 54%, 5h 1%"},
+    }
+
+    text = render_compact(payload)
+
+    assert "🧭 Codex 사용량" in text
+    assert "✅ 추천 company-plus-100" in text
+    assert "company-plus-100" in text
+    assert "5h  1% 🟢" in text
+    assert "7d 54% 🟢" in text
+    assert "[" in text and "]" in text
+
+
+def test_usage_bar_visualizes_percent_buckets():
+    assert usage_bar(0, width=10) == "░░░░░░░░░░"
+    assert usage_bar(54, width=10) == "█████░░░░░"
+    assert usage_bar(98, width=10) == "██████████"
+    assert usage_bar(None, width=10) == "??????????"
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index 707272f10045..6646dc6f05d2 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -254,6 +254,46 @@ def test_update_token_counts_tracks_api_call_count(self, db):
         session = db.get_session("s1")
         assert session["api_call_count"] == 3
 
+    def test_update_token_counts_records_credential_usage_rows(self, db):
+        db.create_session(session_id="s1", source="cli")
+        db.update_token_counts(
+            "s1",
+            input_tokens=100,
+            output_tokens=50,
+            cache_read_tokens=10,
+            reasoning_tokens=5,
+            api_call_count=1,
+            billing_provider="openai-codex",
+            credential_label="company-plus-100",
+            model="gpt-5.5",
+        )
+        db.update_token_counts(
+            "s1",
+            input_tokens=20,
+            output_tokens=5,
+            api_call_count=1,
+            billing_provider="openai-codex",
+            credential_label="company-plus-100",
+            model="gpt-5.5",
+        )
+
+        rows = db.get_credential_usage(days=30, provider="openai-codex")
+
+        assert rows == [
+            {
+                "provider": "openai-codex",
+                "credential_label": "company-plus-100",
+                "model": "gpt-5.5",
+                "api_calls": 2,
+                "input_tokens": 120,
+                "output_tokens": 55,
+                "cache_read_tokens": 10,
+                "cache_write_tokens": 0,
+                "reasoning_tokens": 5,
+                "total_tokens": 190,
+            }
+        ]
+
     def test_update_token_counts_api_call_count_absolute(self, db):
         """absolute mode sets api_call_count directly."""
         db.create_session(session_id="s1", source="cli")
diff --git a/website/docs/user-guide/messaging/telegram.md b/website/docs/user-guide/messaging/telegram.md
index e458f28e0318..1b234014cbe1 100644
--- a/website/docs/user-guide/messaging/telegram.md
+++ b/website/docs/user-guide/messaging/telegram.md
@@ -990,6 +990,30 @@ gateway:
 
 When enabled, Hermes attaches Telegram's `LinkPreviewOptions(is_disabled=True)` to every outgoing message and falls back to the legacy `disable_web_page_preview` parameter on older `python-telegram-bot` versions.
 
+**Copy-to-clipboard buttons.** Telegram supports inline keyboard buttons that copy a short string to the user's clipboard. To attach copy buttons to a Telegram reply, include one marker line per button in the assistant's final text:
+
+```text
+COPY_BUTTON: Button label | text copied to clipboard
+```
+
+Hermes removes valid marker lines from the visible message and renders them as [`CopyTextButton`](https://core.telegram.org/bots/api#copytextbutton) inline buttons. This is useful for IP addresses, one-line commands, invite codes, URLs, and other mobile handoff values that are awkward to select from a Telegram message.
+
+```text
+Server address: 100.96.33.123
+
+COPY_BUTTON: Copy IP | 100.96.33.123
+COPY_BUTTON: Copy VNC address | 100.96.33.123:5900
+```
+
+Marker constraints follow Telegram's copy button limits:
+
+- the label must be 1-64 characters and cannot contain `|`
+- the copied text must be 1-256 characters
+- Hermes renders at most 20 buttons per message
+- invalid marker lines are left as ordinary text instead of being silently dropped
+
+Copy buttons are attached on both normal sends and streaming final edits. When copy buttons are requested, Hermes uses the MarkdownV2 send/edit path instead of the rich-message path because the rich raw endpoint used here does not carry `reply_markup`.
+
 ## Group Allowlisting
 
 Telegram groups and forum chats have two orthogonal gates you can configure: