Add Telegram Codex usage command

README.md

@@ -14,6 +14,7 @@ only keeps connection-management commands.
 /repo <path-under-workspace-root>
 /ask <prompt>
 /work <prompt>
+/usage
 /new
 /status
 /settings
@@ -51,6 +52,8 @@ explicitly.
 `/settings`, `/models`, `/schedules`, `/records`, and `/sessions` return inline
 buttons, so you can change speed/model/reasoning settings, delete schedules,
 manage personal records, and switch or archive sessions without copying ids.
+`/usage` asks the local Codex app-server for account, plan, rate-limit, and
+credit information when the installed Codex version exposes it.
 
 ## Runtime settings
 

app/bot.py

@@ -35,6 +35,7 @@ HELP_TEXT = """Telegram-Codex 연결 브리지
 /ask <내용> - 내용을 Codex로 전달
 /work <내용> - 파일 수정, 커밋, 푸시 같은 쓰기 작업 실행
 /cancel - 현재 진행 중인 작업 취소
+/usage - Codex 사용량/제한 확인
 /new - 이 채팅의 Codex 세션 초기화
 /status - 브리지 상태 확인
 /settings - 모델, 추론강도, 속도 설정 확인
@@ -130,6 +131,9 @@ class TelegramCodexBot:
         app.add_handler(CommandHandler("work", self.work))
         app.add_handler(CommandHandler("write", self.work))
         app.add_handler(CommandHandler("cancel", self.cancel))
+        app.add_handler(CommandHandler("usage", self.usage))
+        app.add_handler(CommandHandler("limits", self.usage))
+        app.add_handler(CommandHandler("quota", self.usage))
         app.add_handler(CommandHandler("new", self.new))
         app.add_handler(CommandHandler("status", self.status))
         app.add_handler(CommandHandler("settings", self.settings))
@@ -289,6 +293,24 @@ class TelegramCodexBot:
         await self._close_client_for_chat(chat_id)
         await self._reply(update, f"작업 취소를 요청했습니다.\n{self._format_job_status(chat_id)}")
 
+    async def usage(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        if not await self._guard(update):
+            return
+        await self._reply_usage(update)
+
+    async def _reply_usage(self, update: Update) -> None:
+        try:
+            account = await self.management_runner.call("account/read", {})
+        except Exception as exc:
+            await self._reply(
+                update,
+                "Codex 사용량 정보를 가져오지 못했습니다.\n"
+                f"오류: {exc}\n\n"
+                "NAS의 Codex app-server가 구버전이거나 계정 정보 조회 표면을 제공하지 않을 수 있습니다.",
+            )
+            return
+        await self._reply(update, _format_codex_usage(account))
+
     async def status(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         if not await self._guard(update):
             return
@@ -925,6 +947,9 @@ class TelegramCodexBot:
         text = text.strip()
         if not text:
             return
+        if _looks_like_usage_question(text):
+            await self._reply_usage(update)
+            return
         await self._run_codex_from_update(update, text)
 
     async def _run_codex_from_update(
@@ -1432,6 +1457,139 @@ def _repo_name_tokens(prompt: str) -> list[str]:
     return _dedupe(tokens)
 
 
+def _looks_like_usage_question(text: str) -> bool:
+    normalized = text.casefold()
+    usage_terms = ("사용량", "남은량", "남은 사용량", "한도", "쿼터", "quota", "limit", "usage", "credits", "크레딧")
+    if not any(term in normalized for term in usage_terms):
+        return False
+
+    explicit_codex_terms = ("codex", "코덱스", "openai", "gpt", "모델")
+    if any(term in normalized for term in explicit_codex_terms):
+        return True
+
+    non_codex_terms = ("디스크", "용량", "저장소", "나스", "nas", "gitea", "repo", "cpu", "메모리")
+    if any(term in normalized for term in non_codex_terms):
+        return False
+
+    implicit_codex_terms = ("주간 사용량", "월간 사용량", "남은 사용량", "남은 한도", "쿼터", "quota", "credits", "크레딧")
+    return any(term in normalized for term in implicit_codex_terms)
+
+
+def _format_codex_usage(payload: Any) -> str:
+    if not isinstance(payload, dict):
+        return "Codex 사용량 응답 형식을 해석하지 못했습니다."
+
+    lines = ["Codex 사용량/제한:"]
+    account = payload.get("account") if isinstance(payload.get("account"), dict) else payload
+    plan = _first_present(account, "planType", "plan_type", "plan", "tier")
+    email = _first_present(account, "email", "userEmail", "user_email")
+    auth_mode = _first_present(account, "authMode", "auth_mode")
+    if email:
+        lines.append(f"계정={email}")
+    if plan:
+        lines.append(f"플랜={plan}")
+    if auth_mode:
+        lines.append(f"인증={auth_mode}")
+
+    rate_limits = payload.get("rateLimitsByLimitId")
+    if isinstance(rate_limits, dict) and rate_limits:
+        lines.append("제한:")
+        for limit_id, bucket in rate_limits.items():
+            if isinstance(bucket, dict):
+                lines.append(_format_rate_limit_bucket(str(limit_id), bucket))
+    else:
+        legacy = payload.get("rateLimits")
+        if isinstance(legacy, dict) and legacy:
+            lines.append("제한:")
+            lines.append(_format_rate_limit_bucket("codex", legacy))
+        elif isinstance(legacy, list) and legacy:
+            lines.append("제한:")
+            for index, bucket in enumerate(legacy, start=1):
+                if isinstance(bucket, dict):
+                    lines.append(_format_rate_limit_bucket(str(bucket.get("limitId") or index), bucket))
+
+    credits = payload.get("credits")
+    if isinstance(credits, dict) and credits:
+        lines.append("크레딧:")
+        for key, value in credits.items():
+            if _is_safe_scalar(value):
+                lines.append(f"- {key}={value}")
+
+    if len(lines) == 1:
+        lines.append("현재 Codex app-server가 사용량/제한 값을 반환하지 않았습니다.")
+        lines.append("Codex UI의 usage dashboard 또는 CLI 활성 세션의 /status에서도 확인해 보세요.")
+    return "\n".join(lines)
+
+
+def _format_rate_limit_bucket(limit_id: str, bucket: dict[str, Any]) -> str:
+    name = bucket.get("limitName") or bucket.get("name") or limit_id
+    used_percent = _number_or_none(_value_for_any_key(bucket, "usedPercent", "used_percent"))
+    remaining = None if used_percent is None else max(0.0, 100.0 - used_percent)
+    window = _value_for_any_key(bucket, "windowDurationMins", "window_duration_mins")
+    resets_at = _value_for_any_key(bucket, "resetsAt", "resets_at")
+    plan = _value_for_any_key(bucket, "planType", "plan_type")
+    reached = _value_for_any_key(bucket, "rateLimitReachedType", "rate_limit_reached_type")
+
+    parts = [f"- {name}"]
+    if remaining is not None:
+        parts.append(f"남음≈{remaining:.1f}%")
+        parts.append(f"사용={used_percent:.1f}%")
+    if window:
+        parts.append(f"창={window}분")
+    reset_text = _format_unix_time(resets_at)
+    if reset_text:
+        parts.append(f"리셋={reset_text}")
+    if plan:
+        parts.append(f"플랜={plan}")
+    if reached:
+        parts.append(f"상태={reached}")
+    return " / ".join(parts)
+
+
+def _first_present(mapping: dict[str, Any], *keys: str) -> Any:
+    for key in keys:
+        value = mapping.get(key)
+        if value:
+            return value
+    return None
+
+
+def _value_for_any_key(mapping: dict[str, Any], *keys: str) -> Any:
+    for key in keys:
+        if key in mapping and mapping[key] is not None:
+            return mapping[key]
+    return None
+
+
+def _number_or_none(value: Any) -> float | None:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _format_unix_time(value: Any) -> str | None:
+    if isinstance(value, str) and value.strip():
+        return value.strip()
+    number = _number_or_none(value)
+    if number is None:
+        return None
+    if number > 10_000_000_000:
+        number = number / 1000
+    try:
+        from datetime import datetime
+
+        return datetime.fromtimestamp(number).astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
+    except Exception:
+        return str(value)
+
+
+def _is_safe_scalar(value: Any) -> bool:
+    return isinstance(value, (str, int, float, bool)) or value is None
+
+
 def _checked_label(label: str, checked: bool) -> str:
     return f"✓ {label}" if checked else label