nas_connection/app/bot.py

from __future__ import annotations

import asyncio
import json
import re
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ChatAction
from telegram.ext import (
    Application,
    CallbackQueryHandler,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)

from app.app_server_runner import AppServerRunner, PersistentAppServerRunner
from app.codex_runner import CodexResult, CodexRunner
from app.config import Config
from app.state import ChatState, StateStore, format_personal_record, format_schedule


HELP_TEXT = """Telegram-Codex 연결 브리지

일반 메시지는 그대로 Codex app-server로 전달됩니다.

명령어:
/whoami - 내 Telegram user_id/chat_id 확인
/repo <경로> - /workspaces 아래 작업 폴더 설정
/ask <내용> - 내용을 Codex로 전달
/work <내용> - 파일 수정, 커밋, 푸시 같은 쓰기 작업 실행
/cancel - 현재 진행 중인 작업 취소
/usage - Codex 사용량/제한 확인
/new - 이 채팅의 Codex 세션 초기화
/status - 브리지 상태 확인
/settings - 모델, 추론강도, 속도 설정 확인
/set_model <model|latest|default> - 모델 변경
/set_effort <minimal|low|medium|high|xhigh|default> - 추론강도 변경
/set_speed <standard|fast|보통|고속|default> - 속도 모드 변경
/fast <on|off|status> - 고속 모드 바로 변경/확인
/models - 모델 선택 버튼 열기
/schedules - 등록된 스케줄 목록/삭제 버튼 열기
/records - 개인 기록 목록/삭제 버튼 열기
/todos - 할 일 목록 열기
/purchases - 구매 후보 목록 열기
/birthdays - 생일 목록 열기
/sessions - app-server 세션 목록 확인
/sessions archived - 보관된 세션 목록 확인
/use_session <thread_id> - 기존 세션으로 전환
/archive_session [thread_id] - 세션 보관
/unarchive_session <thread_id> - 보관된 세션 복원
/help - 도움말 보기
"""

EFFORT_VALUES = {"minimal", "low", "medium", "high", "xhigh"}
RESET_WORDS = {"default", "reset", "unset", "기본", "초기화"}
MODEL_ALIASES = {
    "spark": "gpt-5.3-codex-spark",
    "스파크": "gpt-5.3-codex-spark",
    "mini": "gpt-5.4-mini",
    "미니": "gpt-5.4-mini",
}
SPEED_ALIASES = {
    "standard": "standard",
    "normal": "standard",
    "off": "standard",
    "보통": "standard",
    "기본": "standard",
    "fast": "fast",
    "on": "fast",
    "고속": "fast",
}
RECORD_CATEGORY_LABELS = {
    "work_log": "업무이력",
    "todo": "할 일",
    "purchase": "구매",
    "birthday": "생일",
    "contact": "연락처",
    "note": "메모",
    "preference": "취향/설정",
}


@dataclass
class RunningJob:
    task: asyncio.Task
    started_monotonic: float
    prompt_preview: str
    sandbox: str


class TelegramCodexBot:
    def __init__(self, config: Config) -> None:
        self.config = config
        self.state = StateStore(config.state_path)
        self.exec_runner = CodexRunner(
            codex_bin=config.codex_bin,
            timeout_seconds=config.codex_timeout_seconds,
            resume_json_flag_style=config.resume_json_flag_style,
        )
        self.management_runner = AppServerRunner(
            url=config.app_server_url,
            timeout_seconds=config.codex_timeout_seconds,
            model=config.app_server_model,
            effort=config.app_server_reasoning_effort,
            service_tier=_service_tier_for_speed(config.app_server_speed_mode),
        )
        self.chat_clients: dict[int, PersistentAppServerRunner] = {}
        self.model_menus: dict[int, list[str]] = {}
        self.running_jobs: dict[int, RunningJob] = {}
        self.pending_prompts: dict[int, list[str]] = {}
        self.scheduler_task: asyncio.Task | None = None

    def build_app(self) -> Application:
        app = (
            Application.builder()
            .token(self.config.telegram_bot_token)
            .post_init(self.post_init)
            .post_shutdown(self.post_shutdown)
            .build()
        )
        app.add_handler(CommandHandler("start", self.start))
        app.add_handler(CommandHandler("help", self.start))
        app.add_handler(CommandHandler("whoami", self.whoami))
        app.add_handler(CommandHandler("repo", self.repo))
        app.add_handler(CommandHandler("ask", self.ask))
        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))
        app.add_handler(CommandHandler("set_model", self.set_model))
        app.add_handler(CommandHandler("set_effort", self.set_effort))
        app.add_handler(CommandHandler("set_speed", self.set_speed))
        app.add_handler(CommandHandler("fast", self.fast))
        app.add_handler(CommandHandler("models", self.models))
        app.add_handler(CommandHandler("schedules", self.schedules))
        app.add_handler(CommandHandler("records", self.records))
        app.add_handler(CommandHandler("todos", self.todos))
        app.add_handler(CommandHandler("purchases", self.purchases))
        app.add_handler(CommandHandler("birthdays", self.birthdays))
        app.add_handler(CommandHandler("sessions", self.sessions))
        app.add_handler(CommandHandler("use_session", self.use_session))
        app.add_handler(CommandHandler("archive_session", self.archive_session))
        app.add_handler(CommandHandler("unarchive_session", self.unarchive_session))
        app.add_handler(CallbackQueryHandler(self.callback))
        if self.config.allow_plain_text:
            app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.plain_text))
        return app

    async def post_init(self, app: Application) -> None:
        self.scheduler_task = app.create_task(self.schedule_loop(app))

    async def post_shutdown(self, app: Application) -> None:
        if self.scheduler_task:
            self.scheduler_task.cancel()
        for job in self.running_jobs.values():
            job.task.cancel()
        await asyncio.gather(
            *(client.close() for client in self.chat_clients.values()),
            *(job.task for job in self.running_jobs.values()),
            return_exceptions=True,
        )

    def _is_allowed(self, update: Update) -> bool:
        if not self.config.allowed_user_ids:
            return False
        user = update.effective_user
        return bool(user and user.id in self.config.allowed_user_ids)

    async def _guard(self, update: Update) -> bool:
        if self._is_allowed(update):
            return True
        await self._reply(update, "허용되지 않은 Telegram 사용자입니다. /whoami로 user_id를 확인해 주세요.")
        return False

    async def _reply(
        self,
        update: Update,
        text: str,
        reply_markup: InlineKeyboardMarkup | None = None,
    ) -> None:
        if not update.effective_message:
            return
        chunks = _split_message(text)
        for index, chunk in enumerate(chunks):
            await update.effective_message.reply_text(
                chunk,
                reply_markup=reply_markup if index == len(chunks) - 1 else None,
            )

    async def _reply_or_edit(
        self,
        update: Update,
        text: str,
        reply_markup: InlineKeyboardMarkup | None = None,
    ) -> None:
        query = update.callback_query
        if query and query.message:
            try:
                await query.edit_message_text(text, reply_markup=reply_markup)
                return
            except Exception:
                await query.message.reply_text(text, reply_markup=reply_markup)
                return
        if update.effective_message:
            await update.effective_message.reply_text(text, reply_markup=reply_markup)

    async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        await self._reply_or_edit(update, HELP_TEXT, self._main_keyboard())

    async def whoami(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        user = update.effective_user
        chat = update.effective_chat
        await self._reply(
            update,
            f"user_id={user.id if user else 'unknown'}\nchat_id={chat.id if chat else 'unknown'}",
        )

    async def repo(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        if not context.args:
            await self._reply(update, "사용법: /repo <작업폴더 경로>")
            return

        requested = " ".join(context.args).strip()
        try:
            repo_path = self._resolve_workspace_path(requested)
        except ValueError as exc:
            await self._reply(update, str(exc))
            return

        if not repo_path.exists():
            await self._reply(update, f"경로가 존재하지 않습니다: {repo_path}")
            return

        self.state.set_repo(update.effective_chat.id, repo_path)
        await self._close_client_for_chat(update.effective_chat.id)
        await self._reply(update, f"작업 폴더를 설정했습니다:\n{repo_path}\nCodex 세션을 초기화했습니다.")

    async def ask(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        prompt = _args_text(context)
        if not prompt:
            await self._reply(update, "사용법: /ask <요청 내용>")
            return
        await self._run_codex_from_update(update, prompt)

    async def work(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        prompt = _args_text(context)
        if not prompt:
            await self._reply(update, "사용법: /work <파일 수정, 커밋, 푸시 등 작업 내용>")
            return
        await self._run_codex_from_update(
            update,
            prompt,
            sandbox=self.config.write_sandbox,
        )

    async def new(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        chat_id = update.effective_chat.id
        job = self.running_jobs.get(chat_id)
        if job and not job.task.done():
            await self._reply(update, "작업이 진행 중입니다. 먼저 /cancel로 취소하거나 작업 완료 후 /new를 사용해 주세요.")
            return
        self.state.clear_session(chat_id)
        await self._close_client_for_chat(chat_id)
        await self._reply(update, "이 Telegram 채팅의 Codex 세션을 초기화했습니다.")

    async def cancel(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        await self._cancel_current_job(update)

    async def _cancel_current_job(self, update: Update) -> None:
        chat_id = update.effective_chat.id
        job = self.running_jobs.get(chat_id)
        if not job or job.task.done():
            await self._reply(update, "현재 진행 중인 작업이 없습니다.")
            return
        queued_count = len(self.pending_prompts.pop(chat_id, []))
        job.task.cancel()
        await self._close_client_for_chat(chat_id)
        queue_note = f"\n대기 중이던 후속 지시 {queued_count}개도 비웠습니다." if queued_count else ""
        await self._reply(update, f"작업 취소를 요청했습니다.{queue_note}\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
        chat_id = update.effective_chat.id
        chat_state = self.state.get_chat(chat_id)
        client = self.chat_clients.get(chat_id)
        schedules = self.state.list_schedules(chat_id=chat_id)
        persistent_status = "연결 안 됨"
        if client and client.is_connected:
            persistent_status = f"연결됨, 유휴 {int(client.idle_seconds())}초"
        job_status = self._format_job_status(chat_id)
        await self._reply(
            update,
            "\n".join(
                [
                    f"채팅프로필={self._chat_mode(chat_id)}",
                    f"작업폴더={chat_state.repo_path or '(설정 안 됨)'}",
                    f"백엔드={chat_state.session_backend or '(새 세션)'}",
                    f"세션ID={chat_state.session_id or '(새 세션)'}",
                    f"선호백엔드={self.config.codex_backend}",
                    f"persistent_ws={persistent_status}",
                    f"persistent_ws_유휴제한={self.config.persistent_ws_idle_seconds}초",
                    f"스케줄={len(schedules)}개",
                    job_status,
                    self._format_runtime_settings(),
                ]
            ),
            self._main_keyboard(),
        )

    async def settings(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        await self._reply_or_edit(update, self._format_runtime_settings(), self._settings_keyboard())

    async def set_model(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        raw = _args_text(context)
        if not raw:
            await self._reply(update, "사용법: /set_model <model|latest|default>")
            return

        normalized = raw.strip()
        lowered = normalized.lower()
        if lowered in RESET_WORDS:
            self.state.update_runtime_settings({"model": None})
            action = "모델을 환경 기본값으로 되돌렸습니다."
        else:
            model = await self._resolve_model_input(lowered, normalized)
            self.state.update_runtime_settings({"model": model})
            action = f"모델을 {model}(으)로 설정했습니다."
        await self._close_all_clients()
        await self._reply(update, f"{action}\n\n{self._format_runtime_settings()}")

    async def set_effort(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        raw = _args_text(context).strip().lower()
        if not raw:
            await self._reply(update, "사용법: /set_effort <minimal|low|medium|high|xhigh|default>")
            return
        if raw in RESET_WORDS:
            self.state.update_runtime_settings({"effort": None})
            action = "추론강도를 환경 기본값으로 되돌렸습니다."
        elif raw in EFFORT_VALUES:
            self.state.update_runtime_settings({"effort": raw})
            action = f"추론강도를 {raw}(으)로 설정했습니다."
        else:
            await self._reply(update, "사용 가능한 값: minimal, low, medium, high, xhigh, default")
            return
        await self._close_all_clients()
        await self._reply(update, f"{action}\n\n{self._format_runtime_settings()}")

    async def set_speed(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        raw = _args_text(context).strip().lower()
        if not raw:
            await self._reply(update, "사용법: /set_speed <standard|fast|보통|고속|default>")
            return
        if raw in RESET_WORDS:
            self.state.update_runtime_settings({"speed_mode": None})
            action = "속도 모드를 환경 기본값으로 되돌렸습니다."
        else:
            mode = SPEED_ALIASES.get(raw)
            if not mode:
                await self._reply(update, "사용 가능한 값: standard, fast, 보통, 고속, default")
                return
            self.state.update_runtime_settings({"speed_mode": mode})
            action = f"속도 모드를 {_display_speed(mode)}(으)로 설정했습니다."
        await self._close_all_clients()
        await self._reply(update, f"{action}\n\n{self._format_runtime_settings()}")

    async def fast(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        raw = _args_text(context).strip().lower()
        if not raw or raw == "status":
            await self._reply(update, self._format_runtime_settings())
            return
        if raw not in {"on", "off"}:
            await self._reply(update, "사용법: /fast <on|off|status>")
            return
        mode = "fast" if raw == "on" else "standard"
        self.state.update_runtime_settings({"speed_mode": mode})
        await self._close_all_clients()
        await self._reply(update, f"속도 모드를 {_display_speed(mode)}(으)로 설정했습니다.\n\n{self._format_runtime_settings()}")

    async def models(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        await self._show_model_menu(update)

    async def schedules(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        await self._show_schedule_list(update)

    async def records(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        arg = context.args[0] if context.args else None
        category = arg if arg in RECORD_CATEGORY_LABELS else None
        query = " ".join(context.args).strip() if arg and category is None else None
        await self._show_record_list(update, category=category, query=query)

    async def todos(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        await self._show_record_list(update, category="todo", status="active")

    async def purchases(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        await self._show_record_list(update, category="purchase", status="active")

    async def birthdays(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        await self._show_record_list(update, category="birthday")

    async def sessions(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        archived = bool(context.args and context.args[0].lower() in {"archived", "archive"})
        try:
            result = await self.management_runner.list_threads(limit=12, archived=archived)
        except Exception as exc:
            await self._reply(update, f"세션 목록을 가져오지 못했습니다: {exc}")
            return
        text, keyboard = _format_thread_list(result, archived=archived)
        await self._reply_or_edit(update, text, keyboard)

    async def use_session(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        thread_id = _args_text(context)
        if not thread_id:
            await self._reply(update, "사용법: /use_session <thread_id>")
            return
        try:
            resumed_id = await self._use_session_for_chat(update.effective_chat.id, thread_id)
        except Exception as exc:
            await self._reply(update, f"세션을 전환하지 못했습니다: {exc}")
            return
        await self._reply(update, f"Codex 세션을 전환했습니다.\n세션ID={resumed_id}")

    async def callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        query = update.callback_query
        if not query:
            return
        await query.answer()
        if not await self._guard(update):
            return
        data = query.data or ""
        chat_id = update.effective_chat.id
        try:
            if data == "sessions":
                result = await self.management_runner.list_threads(limit=12, archived=False)
                text, keyboard = _format_thread_list(result, archived=False)
                await self._reply_or_edit(update, text, keyboard)
                return
            if data == "sessions_archived":
                result = await self.management_runner.list_threads(limit=12, archived=True)
                text, keyboard = _format_thread_list(result, archived=True)
                await self._reply_or_edit(update, text, keyboard)
                return
            if data == "settings":
                await self._reply_or_edit(update, self._format_runtime_settings(), self._settings_keyboard())
                return
            if data == "settings_models":
                await self._show_model_menu(update)
                return
            if data == "settings_effort":
                await self._show_effort_menu(update)
                return
            if data == "schedules":
                await self._show_schedule_list(update)
                return
            if data == "records":
                await self._show_record_list(update)
                return
            if data.startswith("records_cat:"):
                await self._show_record_list(update, category=data.split(":", 1)[1])
                return
            if data.startswith("record_done:"):
                await self._mark_record_done_from_button(update, data.split(":", 1)[1])
                return
            if data.startswith("record_delete_confirm:"):
                await self._confirm_record_delete(update, data.split(":", 1)[1])
                return
            if data.startswith("record_delete:"):
                await self._delete_record_from_button(update, data.split(":", 1)[1])
                return
            if data.startswith("schedule_delete_confirm:"):
                await self._confirm_schedule_delete(update, data.split(":", 1)[1])
                return
            if data.startswith("schedule_delete:"):
                await self._delete_schedule_from_button(update, data.split(":", 1)[1])
                return
            if data.startswith("set_speed:"):
                await self._set_speed_mode_from_button(update, data.split(":", 1)[1])
                return
            if data.startswith("set_effort:"):
                await self._set_effort_from_button(update, data.split(":", 1)[1])
                return
            if data == "model_default":
                self.state.update_runtime_settings({"model": None})
                await self._close_all_clients()
                await self._reply_or_edit(update, "모델을 환경 기본값으로 되돌렸습니다.\n\n" + self._format_runtime_settings(), self._settings_keyboard())
                return
            if data == "model_latest":
                models = await self._model_candidates(chat_id)
                model = _latest_model(models)
                self.state.update_runtime_settings({"model": model})
                await self._close_all_clients()
                await self._reply_or_edit(update, f"모델을 최신 추천값 {model}(으)로 설정했습니다.\n\n{self._format_runtime_settings()}", self._settings_keyboard())
                return
            if data.startswith("model_pick:"):
                await self._set_model_from_button(update, data.split(":", 1)[1])
                return
            if data.startswith("use_session:"):
                resumed_id = await self._use_session_for_chat(chat_id, data.split(":", 1)[1])
                await self._reply_or_edit(update, f"Codex 세션을 전환했습니다.\n세션ID={resumed_id}")
                return
            if data.startswith("archive_session:"):
                thread_id = data.split(":", 1)[1]
                await self._archive_session_for_chat(chat_id, thread_id)
                await self._reply_or_edit(update, f"세션을 보관했습니다.\n세션ID={thread_id}")
                return
            if data.startswith("unarchive_session:"):
                thread_id = data.split(":", 1)[1]
                await self.management_runner.unarchive_thread(thread_id)
                await self._reply_or_edit(update, f"세션을 복원했습니다.\n세션ID={thread_id}")
                return
        except Exception as exc:
            await self._reply_or_edit(update, f"버튼 작업을 처리하지 못했습니다: {exc}")
            return
        await self._reply_or_edit(update, "알 수 없는 버튼 동작입니다.")

    def _main_keyboard(self) -> InlineKeyboardMarkup:
        return InlineKeyboardMarkup(
            [
                [
                    InlineKeyboardButton("설정", callback_data="settings"),
                    InlineKeyboardButton("세션", callback_data="sessions"),
                ],
                [
                    InlineKeyboardButton("스케줄", callback_data="schedules"),
                    InlineKeyboardButton("기록", callback_data="records"),
                ],
                [
                    InlineKeyboardButton("모델", callback_data="settings_models"),
                    InlineKeyboardButton("할 일", callback_data="records_cat:todo"),
                ],
            ]
        )

    def _settings_keyboard(self) -> InlineKeyboardMarkup:
        settings = self._runtime_settings()
        speed_mode = str(settings.get("speed_mode") or "standard")
        return InlineKeyboardMarkup(
            [
                [
                    InlineKeyboardButton(_checked_label("보통", speed_mode == "standard"), callback_data="set_speed:standard"),
                    InlineKeyboardButton(_checked_label("고속", speed_mode == "fast"), callback_data="set_speed:fast"),
                ],
                [
                    InlineKeyboardButton("모델 선택", callback_data="settings_models"),
                    InlineKeyboardButton("추론강도", callback_data="settings_effort"),
                ],
                [
                    InlineKeyboardButton("세션", callback_data="sessions"),
                    InlineKeyboardButton("스케줄", callback_data="schedules"),
                ],
                [
                    InlineKeyboardButton("개인 기록", callback_data="records"),
                ],
            ]
        )

    async def _set_speed_mode_from_button(self, update: Update, mode: str) -> None:
        if mode not in {"standard", "fast"}:
            await self._reply_or_edit(update, "알 수 없는 속도 모드입니다.", self._settings_keyboard())
            return
        self.state.update_runtime_settings({"speed_mode": mode})
        await self._close_all_clients()
        await self._reply_or_edit(
            update,
            f"속도 모드를 {_display_speed(mode)}(으)로 설정했습니다.\n\n{self._format_runtime_settings()}",
            self._settings_keyboard(),
        )

    async def _set_effort_from_button(self, update: Update, effort: str) -> None:
        if effort == "default":
            self.state.update_runtime_settings({"effort": None})
            action = "추론강도를 환경 기본값으로 되돌렸습니다."
        elif effort in EFFORT_VALUES:
            self.state.update_runtime_settings({"effort": effort})
            action = f"추론강도를 {effort}(으)로 설정했습니다."
        else:
            await self._reply_or_edit(update, "알 수 없는 추론강도입니다.", self._settings_keyboard())
            return
        await self._close_all_clients()
        await self._reply_or_edit(update, f"{action}\n\n{self._format_runtime_settings()}", self._settings_keyboard())

    async def _set_model_from_button(self, update: Update, index_text: str) -> None:
        chat_id = update.effective_chat.id
        try:
            index = int(index_text)
        except ValueError:
            await self._reply_or_edit(update, "알 수 없는 모델 선택입니다.", self._settings_keyboard())
            return
        models = self.model_menus.get(chat_id)
        if not models:
            models = await self._model_candidates(chat_id)
        if index < 0 or index >= len(models):
            await self._reply_or_edit(update, "모델 목록이 오래되었습니다. /models로 다시 열어주세요.", self._settings_keyboard())
            return
        model = models[index]
        self.state.update_runtime_settings({"model": model})
        await self._close_all_clients()
        await self._reply_or_edit(update, f"모델을 {model}(으)로 설정했습니다.\n\n{self._format_runtime_settings()}", self._settings_keyboard())

    async def _show_effort_menu(self, update: Update) -> None:
        settings = self._runtime_settings()
        current = settings.get("effort")
        rows: list[list[InlineKeyboardButton]] = []
        rows.append([InlineKeyboardButton(_checked_label("기본값", current is None), callback_data="set_effort:default")])
        for effort in ["minimal", "low", "medium", "high", "xhigh"]:
            rows.append([InlineKeyboardButton(_checked_label(effort, current == effort), callback_data=f"set_effort:{effort}")])
        rows.append([InlineKeyboardButton("설정으로 돌아가기", callback_data="settings")])
        await self._reply_or_edit(update, "추론강도를 선택해 주세요.", InlineKeyboardMarkup(rows))

    async def _show_model_menu(self, update: Update) -> None:
        chat_id = update.effective_chat.id
        models = await self._model_candidates(chat_id)
        current = self._runtime_settings().get("model")
        latest = _latest_model(models)
        rows: list[list[InlineKeyboardButton]] = [
            [InlineKeyboardButton(_checked_label("기본값", current is None), callback_data="model_default")],
            [InlineKeyboardButton(f"최신 추천: {latest}", callback_data="model_latest")],
        ]
        for index, model in enumerate(models[:12]):
            rows.append(
                [InlineKeyboardButton(_checked_label(model, current == model), callback_data=f"model_pick:{index}")]
            )
        rows.append([InlineKeyboardButton("목록 새로고침", callback_data="settings_models")])
        rows.append([InlineKeyboardButton("설정으로 돌아가기", callback_data="settings")])
        self.model_menus[chat_id] = models
        await self._reply_or_edit(
            update,
            "모델을 선택해 주세요.\n목록은 NAS의 Codex 모델 카탈로그를 우선 사용합니다.",
            InlineKeyboardMarkup(rows),
        )

    async def _model_candidates(self, chat_id: int) -> list[str]:
        models = await self._discover_codex_models()
        if not models:
            models = self._fallback_models()
        models = _sort_models(models, preferred=self.config.app_server_model)
        self.model_menus[chat_id] = models
        return models

    async def _resolve_model_input(self, lowered: str, original: str) -> str:
        if lowered in {"latest", "최신"}:
            return _latest_model(await self._model_candidates(chat_id=0))
        return MODEL_ALIASES.get(lowered, original)

    async def _discover_codex_models(self) -> list[str]:
        for args in (["debug", "models"], ["debug", "models", "--bundled"]):
            try:
                proc = await asyncio.create_subprocess_exec(
                    self.config.codex_bin,
                    *args,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                stdout, _stderr = await asyncio.wait_for(proc.communicate(), timeout=20)
            except Exception:
                continue
            if proc.returncode != 0:
                continue
            try:
                payload = json.loads(stdout.decode("utf-8", errors="replace"))
            except json.JSONDecodeError:
                continue
            models = _extract_model_ids(payload)
            if models:
                return models
        return []

    def _fallback_models(self) -> list[str]:
        return _dedupe(
            [
                self.config.app_server_model or "",
                "gpt-5.5",
                "gpt-5.4",
                "gpt-5.4-mini",
                "gpt-5.3-codex-spark",
            ]
        )

    async def _show_schedule_list(self, update: Update) -> None:
        chat_id = update.effective_chat.id
        schedules = self.state.list_schedules(chat_id=chat_id)
        if not schedules:
            await self._reply_or_edit(
                update,
                "등록된 스케줄이 없습니다.",
                InlineKeyboardMarkup([[InlineKeyboardButton("설정으로 돌아가기", callback_data="settings")]]),
            )
            return

        lines = ["등록된 스케줄:"]
        rows: list[list[InlineKeyboardButton]] = []
        for schedule in schedules[:20]:
            schedule_id = int(schedule.get("id") or 0)
            lines.append(_format_schedule_display(schedule))
            rows.append(
                [
                    InlineKeyboardButton(
                        f"삭제 #{schedule_id}",
                        callback_data=f"schedule_delete_confirm:{schedule_id}",
                    )
                ]
            )
        rows.append([InlineKeyboardButton("새로고침", callback_data="schedules")])
        rows.append([InlineKeyboardButton("설정으로 돌아가기", callback_data="settings")])
        await self._reply_or_edit(update, "\n".join(lines), InlineKeyboardMarkup(rows))

    async def _confirm_schedule_delete(self, update: Update, schedule_id_text: str) -> None:
        try:
            schedule_id = int(schedule_id_text)
        except ValueError:
            await self._reply_or_edit(update, "알 수 없는 스케줄입니다.")
            return
        schedule = self.state.get_schedule(schedule_id)
        if not schedule or int(schedule.get("chat_id", 0)) != update.effective_chat.id:
            await self._reply_or_edit(update, "해당 스케줄을 찾을 수 없습니다.")
            return
        rows = [
            [
                InlineKeyboardButton("삭제 확정", callback_data=f"schedule_delete:{schedule_id}"),
                InlineKeyboardButton("취소", callback_data="schedules"),
            ]
        ]
        await self._reply_or_edit(
            update,
            f"이 스케줄을 삭제할까요?\n{_format_schedule_display(schedule)}",
            InlineKeyboardMarkup(rows),
        )

    async def _delete_schedule_from_button(self, update: Update, schedule_id_text: str) -> None:
        try:
            schedule_id = int(schedule_id_text)
        except ValueError:
            await self._reply_or_edit(update, "알 수 없는 스케줄입니다.")
            return
        removed = self.state.remove_schedule(schedule_id, chat_id=update.effective_chat.id)
        if not removed:
            await self._reply_or_edit(update, "해당 스케줄을 찾을 수 없습니다.")
            return
        await self._reply_or_edit(
            update,
            f"스케줄을 삭제했습니다.\n{_format_schedule_display(removed)}",
            InlineKeyboardMarkup([[InlineKeyboardButton("스케줄 목록", callback_data="schedules")]]),
        )

    async def _show_record_list(
        self,
        update: Update,
        category: str | None = None,
        status: str | None = None,
        query: str | None = None,
    ) -> None:
        chat_id = update.effective_chat.id
        records = self.state.list_personal_records(
            chat_id=chat_id,
            category=category,
            status=status,
            query=query,
            limit=20,
        )
        if not records:
            rows = self._record_category_rows()
            rows.append([InlineKeyboardButton("설정으로 돌아가기", callback_data="settings")])
            title = _record_list_title(category, status, query)
            await self._reply_or_edit(update, f"{title}: 없음", InlineKeyboardMarkup(rows))
            return

        lines = [_record_list_title(category, status, query) + ":"]
        rows: list[list[InlineKeyboardButton]] = []
        for record in records:
            record_id = int(record.get("id") or 0)
            lines.append(_format_record_display(record))
            row: list[InlineKeyboardButton] = []
            if record.get("status", "active") == "active" and record.get("category") in {"todo", "purchase"}:
                row.append(InlineKeyboardButton(f"완료 #{record_id}", callback_data=f"record_done:{record_id}"))
            row.append(InlineKeyboardButton(f"삭제 #{record_id}", callback_data=f"record_delete_confirm:{record_id}"))
            rows.append(row)
        rows.extend(self._record_category_rows())
        rows.append([InlineKeyboardButton("새로고침", callback_data=f"records_cat:{category}" if category else "records")])
        rows.append([InlineKeyboardButton("설정으로 돌아가기", callback_data="settings")])
        await self._reply_or_edit(update, "\n".join(lines), InlineKeyboardMarkup(rows))

    def _record_category_rows(self) -> list[list[InlineKeyboardButton]]:
        return [
            [
                InlineKeyboardButton("업무이력", callback_data="records_cat:work_log"),
                InlineKeyboardButton("할 일", callback_data="records_cat:todo"),
            ],
            [
                InlineKeyboardButton("구매", callback_data="records_cat:purchase"),
                InlineKeyboardButton("생일", callback_data="records_cat:birthday"),
            ],
            [
                InlineKeyboardButton("메모", callback_data="records_cat:note"),
                InlineKeyboardButton("연락처", callback_data="records_cat:contact"),
            ],
        ]

    async def _mark_record_done_from_button(self, update: Update, record_id_text: str) -> None:
        try:
            record_id = int(record_id_text)
        except ValueError:
            await self._reply_or_edit(update, "알 수 없는 기록입니다.")
            return
        updated = self.state.update_personal_record(
            record_id,
            {"status": "done"},
            chat_id=update.effective_chat.id,
        )
        if not updated:
            await self._reply_or_edit(update, "해당 기록을 찾을 수 없습니다.")
            return
        await self._reply_or_edit(
            update,
            f"완료로 표시했습니다.\n{_format_record_display(updated)}",
            InlineKeyboardMarkup([[InlineKeyboardButton("기록 목록", callback_data="records")]]),
        )

    async def _confirm_record_delete(self, update: Update, record_id_text: str) -> None:
        try:
            record_id = int(record_id_text)
        except ValueError:
            await self._reply_or_edit(update, "알 수 없는 기록입니다.")
            return
        record = self.state.get_personal_record(record_id)
        if not record or int(record.get("chat_id", 0)) != update.effective_chat.id:
            await self._reply_or_edit(update, "해당 기록을 찾을 수 없습니다.")
            return
        rows = [
            [
                InlineKeyboardButton("삭제 확정", callback_data=f"record_delete:{record_id}"),
                InlineKeyboardButton("취소", callback_data="records"),
            ]
        ]
        await self._reply_or_edit(
            update,
            f"이 기록을 삭제할까요?\n{_format_record_display(record)}",
            InlineKeyboardMarkup(rows),
        )

    async def _delete_record_from_button(self, update: Update, record_id_text: str) -> None:
        try:
            record_id = int(record_id_text)
        except ValueError:
            await self._reply_or_edit(update, "알 수 없는 기록입니다.")
            return
        removed = self.state.remove_personal_record(record_id, chat_id=update.effective_chat.id)
        if not removed:
            await self._reply_or_edit(update, "해당 기록을 찾을 수 없습니다.")
            return
        await self._reply_or_edit(
            update,
            f"기록을 삭제했습니다.\n{_format_record_display(removed)}",
            InlineKeyboardMarkup([[InlineKeyboardButton("기록 목록", callback_data="records")]]),
        )

    async def archive_session(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        chat_state = self.state.get_chat(update.effective_chat.id)
        thread_id = _args_text(context) or chat_state.session_id
        if not thread_id:
            await self._reply(update, "사용법: /archive_session [thread_id]")
            return
        try:
            await self._archive_session_for_chat(update.effective_chat.id, thread_id)
        except Exception as exc:
            await self._reply(update, f"세션을 보관하지 못했습니다: {exc}")
            return
        await self._reply(update, f"세션을 보관했습니다.\n세션ID={thread_id}")

    async def unarchive_session(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        thread_id = _args_text(context)
        if not thread_id:
            await self._reply(update, "사용법: /unarchive_session <thread_id>")
            return
        try:
            await self.management_runner.unarchive_thread(thread_id)
        except Exception as exc:
            await self._reply(update, f"세션을 복원하지 못했습니다: {exc}")
            return
        await self._reply(update, f"세션을 복원했습니다.\n세션ID={thread_id}")

    async def plain_text(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        if not await self._guard(update):
            return
        text = update.effective_message.text if update.effective_message else ""
        text = text.strip()
        if not text:
            return
        if _looks_like_usage_question(text):
            await self._reply_usage(update)
            return
        if _looks_like_cancel_request(text):
            await self._cancel_current_job(update)
            return
        if _looks_like_status_question(text):
            await self._reply(update, self._format_conversational_status(update.effective_chat.id))
            return
        await self._run_codex_from_update(update, text)

    async def _run_codex_from_update(
        self,
        update: Update,
        prompt: str,
        sandbox: str | None = None,
    ) -> None:
        chat_id = update.effective_chat.id
        active_job = self.running_jobs.get(chat_id)
        if active_job and not active_job.task.done():
            self.pending_prompts.setdefault(chat_id, []).append(prompt)
            await self._reply(
                update,
                "지금 작업이 진행 중이라, 이 메시지는 다음 지시로 이어서 전달해둘게요.\n"
                f"{self._format_conversational_status(chat_id)}\n\n"
                f"대기 중인 후속 지시: {len(self.pending_prompts.get(chat_id, []))}개\n"
                f"추가한 지시: {_preview_prompt(prompt)}",
            )
            return

        await self._start_codex_job(
            bot=update.get_bot(),
            chat_id=chat_id,
            prompt=prompt,
            sandbox=sandbox,
        )

    async def _start_codex_job(
        self,
        bot: Any,
        chat_id: int,
        prompt: str,
        sandbox: str | None = None,
    ) -> None:
        chat_mode = self._chat_mode(chat_id)
        selected_sandbox = sandbox or self._default_sandbox_for_chat(chat_id)
        auto_repo_path = self._auto_select_repo_for_prompt(chat_id, prompt) if chat_mode == "work" else None
        if auto_repo_path is not None:
            await self._close_client_for_chat(chat_id)
        preview = _preview_prompt(prompt)
        task = asyncio.create_task(
            self._run_codex_job(
                bot=update.get_bot(),
                chat_id=chat_id,
                prompt=prompt,
                sandbox=selected_sandbox,
            )
        )
        self.running_jobs[chat_id] = RunningJob(
            task=task,
            started_monotonic=time.monotonic(),
            prompt_preview=preview,
            sandbox=selected_sandbox,
        )
        await self._send_chat(
            bot,
            chat_id,
            self._format_start_message(
                chat_mode=chat_mode,
                preview=preview,
                auto_repo_path=auto_repo_path,
            ),
        )

    async def _run_codex_job(
        self,
        bot: Any,
        chat_id: int,
        prompt: str,
        sandbox: str,
    ) -> None:
        heartbeat = asyncio.create_task(self._job_heartbeat(bot, chat_id))
        try:
            result, fallback_note = await self._run_codex_for_chat(
                chat_id=chat_id,
                prompt=prompt,
                sandbox=sandbox,
                update_chat_state=True,
            )
            await self._send_result(bot, chat_id, result, fallback_note)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            await self._send_chat(bot, chat_id, f"작업 중 오류가 발생했습니다:\n{exc}")
        finally:
            heartbeat.cancel()
            await asyncio.gather(heartbeat, return_exceptions=True)
            current = self.running_jobs.get(chat_id)
            if current and current.task is asyncio.current_task():
                self.running_jobs.pop(chat_id, None)
            await self._start_next_pending_job(bot, chat_id)

    async def _start_next_pending_job(self, bot: Any, chat_id: int) -> None:
        queue = self.pending_prompts.get(chat_id)
        if not queue:
            return
        next_prompt = queue.pop(0)
        if not queue:
            self.pending_prompts.pop(chat_id, None)
        await self._send_chat(bot, chat_id, "대기 중이던 다음 지시를 이어서 진행할게요.")
        await self._start_codex_job(
            bot=bot,
            chat_id=chat_id,
            prompt=next_prompt,
            sandbox=self._default_sandbox_for_chat(chat_id),
        )

    async def _job_heartbeat(self, bot: Any, chat_id: int) -> None:
        await asyncio.sleep(45 if self._chat_mode(chat_id) == "work" else 90)
        while True:
            await self._send_chat(bot, chat_id, self._format_progress_message(chat_id))
            await asyncio.sleep(90 if self._chat_mode(chat_id) == "work" else 180)

    async def _reply_result(self, update: Update, result: CodexResult, fallback_note: str = "") -> None:
        if result.ok:
            text = result.final_message
            if result.stderr and "runtime setting overrides" in result.stderr:
                text = f"브리지 참고: {result.stderr}\n\n{text}"
            if fallback_note:
                text = f"{fallback_note}\n\n{text}"
            await self._reply(update, text)
            return

        stderr_text = f"\n\nstderr:\n{result.stderr[-1000:]}" if result.stderr else ""
        note_text = f"{fallback_note}\n\n" if fallback_note else ""
        await self._reply(update, f"실패(returncode={result.returncode})\n\n{note_text}{result.final_message}{stderr_text}")

    async def _send_result(self, bot: Any, chat_id: int, result: CodexResult, fallback_note: str = "") -> None:
        if result.ok:
            text = result.final_message
            if result.stderr and "runtime setting overrides" in result.stderr:
                text = f"브리지 참고: {result.stderr}\n\n{text}"
            if fallback_note:
                text = f"{fallback_note}\n\n{text}"
            await self._send_chat(bot, chat_id, text)
            return

        stderr_text = f"\n\nstderr:\n{result.stderr[-1000:]}" if result.stderr else ""
        note_text = f"{fallback_note}\n\n" if fallback_note else ""
        await self._send_chat(
            bot,
            chat_id,
            f"실패(returncode={result.returncode})\n\n{note_text}{result.final_message}{stderr_text}",
        )

    async def _send_chat(self, bot: Any, chat_id: int, text: str) -> None:
        for chunk in _split_message(text):
            await bot.send_message(chat_id=chat_id, text=chunk)

    async def _run_codex_for_chat(
        self,
        chat_id: int,
        prompt: str,
        sandbox: str,
        update_chat_state: bool,
        session_id_override: str | None = None,
    ) -> tuple[CodexResult, str]:
        chat_state = self.state.get_chat(chat_id)
        cwd = Path(chat_state.repo_path) if chat_state.repo_path else self.config.workspace_root
        if not cwd.exists():
            return (
                CodexResult(
                    ok=False,
                    final_message=f"작업 폴더가 존재하지 않습니다: {cwd}",
                    session_id=chat_state.session_id,
                    returncode=1,
                ),
                "",
            )

        bridge_prompt = self._with_bridge_context(
            user_prompt=prompt,
            chat_id=chat_id,
            chat_state=chat_state,
            cwd=cwd,
            sandbox=sandbox,
        )
        temp_state = ChatState(
            repo_path=chat_state.repo_path,
            session_id=session_id_override or chat_state.session_id,
            session_backend="app-server" if session_id_override else chat_state.session_backend,
        )
        skip_git_repo_check = chat_state.repo_path is None
        result, backend_used, fallback_note = await self._run_with_backend(
            chat_id=chat_id,
            prompt=bridge_prompt,
            cwd=cwd,
            sandbox=sandbox,
            chat_state=temp_state,
            skip_git_repo_check=skip_git_repo_check,
        )

        if update_chat_state:
            chat_state.session_id = result.session_id or chat_state.session_id
            if result.session_id:
                chat_state.session_backend = backend_used
            self.state.update_chat(chat_id, chat_state)

        return result, fallback_note

    async def _run_with_backend(
        self,
        chat_id: int,
        prompt: str,
        cwd: Path,
        sandbox: str,
        chat_state: ChatState,
        skip_git_repo_check: bool,
    ) -> tuple[CodexResult, str, str]:
        if self._should_use_app_server(sandbox):
            app_session = _session_id_for_backend(chat_state, "app-server")
            app_result = await self._client_for_chat(chat_id).run(
                prompt=prompt,
                cwd=cwd,
                sandbox=sandbox,
                session_id=app_session,
                skip_git_repo_check=skip_git_repo_check,
            )
            if app_result.ok or self.config.codex_backend == "app-server":
                return app_result, "app-server", ""

            exec_session = _session_id_for_backend(chat_state, "exec")
            exec_result = await self.exec_runner.run(
                prompt=prompt,
                cwd=cwd,
                sandbox=sandbox,
                session_id=exec_session,
                skip_git_repo_check=skip_git_repo_check,
            )
            note = f"app-server 실행이 실패해서 이번 요청은 exec fallback으로 처리했습니다.\napp-server 오류: {app_result.final_message}"
            if not exec_result.ok:
                exec_result.final_message = (
                    "app-server와 exec fallback이 모두 실패했습니다.\n\n"
                    f"app-server: {app_result.final_message}\n\n"
                    f"exec: {exec_result.final_message}"
                )
            return exec_result, "exec", note

        exec_session = _session_id_for_backend(chat_state, "exec")
        exec_result = await self.exec_runner.run(
            prompt=prompt,
            cwd=cwd,
            sandbox=sandbox,
            session_id=exec_session,
            skip_git_repo_check=skip_git_repo_check,
        )
        return exec_result, "exec", ""

    async def schedule_loop(self, app: Application) -> None:
        await asyncio.sleep(5)
        while True:
            try:
                await self._close_idle_clients()
                for schedule in self.state.claim_due_schedules():
                    app.create_task(self._run_schedule(app, schedule))
            except Exception as exc:
                print(f"schedule loop error: {exc}", flush=True)
            await asyncio.sleep(max(5, self.config.schedule_poll_seconds))

    async def _close_idle_clients(self) -> None:
        closed_chat_ids: list[int] = []
        for chat_id, client in list(self.chat_clients.items()):
            if await client.close_if_idle(self.config.persistent_ws_idle_seconds):
                closed_chat_ids.append(chat_id)
        for chat_id in closed_chat_ids:
            print(f"closed idle app-server websocket for chat_id={chat_id}", flush=True)

    async def _close_client_for_chat(self, chat_id: int) -> None:
        client = self.chat_clients.pop(chat_id, None)
        if client is not None:
            await client.close()

    async def _close_all_clients(self) -> None:
        clients = list(self.chat_clients.values())
        self.chat_clients.clear()
        await asyncio.gather(*(client.close() for client in clients), return_exceptions=True)

    async def _run_schedule(self, app: Application, schedule: dict) -> None:
        chat_id = int(schedule["chat_id"])
        name = schedule.get("name") or f"스케줄 #{schedule.get('id')}"
        prompt = _scheduled_prompt(schedule)
        try:
            await app.bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING)
            result, fallback_note = await self._run_codex_for_chat(
                chat_id=chat_id,
                prompt=prompt,
                sandbox=self._default_sandbox_for_chat(chat_id),
                update_chat_state=False,
                session_id_override=schedule.get("thread_id") or None,
            )
            if result.session_id and not schedule.get("thread_id"):
                schedule["thread_id"] = result.session_id
                self.state.update_schedule(schedule)
            if result.ok:
                text = f"스케줄 실행 결과: {name}\n\n{result.final_message}"
                if fallback_note:
                    text = f"{fallback_note}\n\n{text}"
            else:
                text = f"스케줄 실행 실패: {name}\n\n{result.final_message}"
            for chunk in _split_message(text):
                await app.bot.send_message(chat_id=chat_id, text=chunk)
        except Exception as exc:
            await app.bot.send_message(chat_id=chat_id, text=f"스케줄 실행 실패: {name}\n\n{exc}")

    def _client_for_chat(self, chat_id: int) -> PersistentAppServerRunner:
        client = self.chat_clients.get(chat_id)
        if client is None:
            client = PersistentAppServerRunner(
                url=self.config.app_server_url,
                timeout_seconds=self.config.codex_timeout_seconds,
            )
            self.chat_clients[chat_id] = client
        self._apply_runtime_settings(client)
        return client

    async def _use_session_for_chat(self, chat_id: int, thread_id: str) -> str:
        self._apply_runtime_settings(self.management_runner)
        resumed_id = await self.management_runner.resume_thread(thread_id)
        chat_state = self.state.get_chat(chat_id)
        chat_state.session_id = resumed_id
        chat_state.session_backend = "app-server"
        self.state.update_chat(chat_id, chat_state)
        await self._close_client_for_chat(chat_id)
        return resumed_id

    async def _archive_session_for_chat(self, chat_id: int, thread_id: str) -> None:
        await self.management_runner.archive_thread(thread_id)
        chat_state = self.state.get_chat(chat_id)
        if chat_state.session_id == thread_id:
            chat_state.session_id = None
            chat_state.session_backend = None
            self.state.update_chat(chat_id, chat_state)
            await self._close_client_for_chat(chat_id)

    def _runtime_defaults(self) -> dict[str, str | None]:
        return {
            "model": self.config.app_server_model,
            "effort": self.config.app_server_reasoning_effort,
            "speed_mode": self.config.app_server_speed_mode,
        }

    def _runtime_settings(self) -> dict[str, str | None]:
        settings = self.state.get_runtime_settings(self._runtime_defaults())
        speed_mode = str(settings.get("speed_mode") or "standard").lower()
        if speed_mode not in {"standard", "fast"}:
            speed_mode = "standard"
        settings["speed_mode"] = speed_mode
        return settings

    def _apply_runtime_settings(self, runner: AppServerRunner) -> None:
        settings = self._runtime_settings()
        runner.model = settings.get("model")
        runner.effort = settings.get("effort")
        runner.service_tier = _service_tier_for_speed(str(settings.get("speed_mode") or "standard"))

    def _format_runtime_settings(self) -> str:
        settings = self._runtime_settings()
        speed_mode = str(settings.get("speed_mode") or "standard")
        service_tier = _service_tier_for_speed(speed_mode) or "standard"
        return "\n".join(
            [
                f"모델={settings.get('model') or '(기본값)'}",
                f"추론강도={settings.get('effort') or '(기본값)'}",
                f"속도={_display_speed(speed_mode)}",
                f"서비스티어={service_tier}",
            ]
        )

    def _format_job_status(self, chat_id: int) -> str:
        job = self.running_jobs.get(chat_id)
        if not job or job.task.done():
            return "현재작업=없음"
        elapsed = int(time.monotonic() - job.started_monotonic)
        minutes, seconds = divmod(elapsed, 60)
        mode = f"{self._chat_mode(chat_id)}/{_sandbox_mode_label(job.sandbox, self.config)}"
        return (
            "현재작업=진행 중\n"
            f"작업모드={mode}\n"
            f"경과={minutes}분 {seconds}초\n"
            f"요청={job.prompt_preview}"
        )

    def _format_conversational_status(self, chat_id: int) -> str:
        job = self.running_jobs.get(chat_id)
        if not job or job.task.done():
            if self._chat_mode(chat_id) == "work":
                return "지금 진행 중인 작업은 없습니다. 작업 그룹에 메시지를 보내면 그대로 Codex 작업 세션으로 전달합니다."
            return "지금 진행 중인 비서 작업은 없습니다."
        elapsed = int(time.monotonic() - job.started_monotonic)
        minutes, seconds = divmod(elapsed, 60)
        mode = f"{self._chat_mode(chat_id)}/{_sandbox_mode_label(job.sandbox, self.config)}"
        queued_count = len(self.pending_prompts.get(chat_id, []))
        queue_line = f"\n- 대기 중인 후속 지시: {queued_count}개" if queued_count else ""
        return (
            "아직 작업 중입니다.\n"
            f"- 모드: {mode}\n"
            f"- 경과: {minutes}분 {seconds}초\n"
            f"- 요청: {job.prompt_preview}"
            f"{queue_line}"
        )

    def _format_start_message(
        self,
        chat_mode: str,
        preview: str,
        auto_repo_path: Path | None,
    ) -> str:
        repo_note = f"\n작업 폴더는 `{auto_repo_path}`로 잡았습니다." if auto_repo_path else ""
        if chat_mode == "work":
            return (
                "좋아요. 작업 세션에 바로 넘겼습니다.\n"
                "먼저 필요한 파일과 현재 상태를 확인한 뒤 진행할게요."
                f"{repo_note}\n\n"
                f"받은 내용: {preview}"
            )
        return (
            "알겠습니다. 비서 세션에서 확인해볼게요.\n"
            f"받은 내용: {preview}"
        )

    def _format_progress_message(self, chat_id: int) -> str:
        if self._chat_mode(chat_id) == "work":
            return (
                "진행 중입니다. Codex 작업 세션이 아직 처리하고 있어요.\n"
                f"{self._format_conversational_status(chat_id)}"
            )
        return (
            "아직 확인 중입니다.\n"
            f"{self._format_conversational_status(chat_id)}"
        )

    def _with_bridge_context(
        self,
        user_prompt: str,
        chat_id: int,
        chat_state: ChatState,
        cwd: Path,
        sandbox: str,
    ) -> str:
        thread_id = chat_state.session_id or "(not assigned yet)"
        schedules = self.state.list_schedules(chat_id=chat_id)
        schedule_lines = "\n".join(f"- {format_schedule(schedule)}" for schedule in schedules[:12])
        if not schedule_lines:
            schedule_lines = "- none"
        records = self.state.list_personal_records(chat_id=chat_id, limit=8)
        record_lines = "\n".join(f"- {format_personal_record(record)}" for record in records)
        if not record_lines:
            record_lines = "- none"
        runtime_settings = self._format_runtime_settings()
        gitea_context = self._format_gitea_context()
        return (
            "[Telegram bridge context]\n"
            "You are connected to the user through a private Telegram bridge.\n"
            f"Chat profile: {self._chat_mode(chat_id)}\n"
            f"Default Telegram chat_id for replies, schedules, and notifications: {chat_id}\n"
            f"Current Codex app-server thread_id: {thread_id}\n"
            f"Current working directory: {cwd}\n"
            f"Current sandbox: {sandbox}\n"
            "Current runtime settings:\n"
            f"{runtime_settings}\n"
            "Configured Gitea integration:\n"
            f"{gitea_context}\n"
            "If the user asks for repository, commit, issue, branch, or pull-request metadata in Gitea, "
            "use the gitea_* MCP tools. Never reveal GITEA_TOKEN or print commands that embed it. "
            "For HTTPS git clone/fetch/push, the container may provide GIT_ASKPASS from the "
            "configured Gitea environment.\n"
            f"{self._chat_mode_instructions(chat_id)}\n"
            "If the user asks to schedule, remind, monitor, or notify later, use the "
            "telegram_create_schedule MCP tool. Use the default chat_id above and do not ask "
            "for a destination unless the user explicitly wants a different chat. The schedule "
            "tool's prompt field should describe only the task to perform at run time.\n"
            "If the user asks to list or delete schedules, use telegram_list_schedules or "
            "telegram_delete_schedule.\n"
            "If the user asks you to remember, save, log, manage, or later recall personal "
            "assistant information, use the personal record MCP tools. Categories are: "
            "work_log, todo, purchase, birthday, contact, note, preference. Store durable "
            "facts such as project history, todos, purchase candidates, birthdays, contacts, "
            "and personal preferences when the user explicitly provides them or asks to save them. "
            "Use telegram_list_personal_records before answering questions that depend on saved "
            "personal history or personal assistant records.\n"
            "Currently registered bridge schedules for this chat:\n"
            f"{schedule_lines}\n"
            "Recent personal assistant records for this chat:\n"
            f"{record_lines}\n"
            "[/Telegram bridge context]\n\n"
            "User message:\n"
            f"{user_prompt}"
        )

    def _format_gitea_context(self) -> str:
        if not self.config.gitea_base_url:
            return "- not configured"
        lines = [
            f"- base_url={self.config.gitea_base_url}",
            f"- username={self.config.gitea_username or '(not set)'}",
            f"- email={self.config.gitea_email or '(not set)'}",
            f"- default_owner={self.config.gitea_default_owner or '(not set)'}",
            f"- default_repo={self.config.gitea_default_repo or '(not set)'}",
            f"- token={'configured' if self.config.gitea_token_configured else 'missing'}",
        ]
        return "\n".join(lines)

    def _should_use_app_server(self, sandbox: str) -> bool:
        if self.config.codex_backend == "exec":
            return False
        return True

    def _chat_mode(self, chat_id: int) -> str:
        return "work" if chat_id in self.config.work_chat_ids else "assistant"

    def _default_sandbox_for_chat(self, chat_id: int) -> str:
        if self._chat_mode(chat_id) == "work":
            return self.config.write_sandbox
        return self.config.assistant_sandbox

    def _chat_mode_instructions(self, chat_id: int) -> str:
        if self._chat_mode(chat_id) == "work":
            return (
                "This is the user's work chat. Treat ordinary user messages as instructions "
                "or follow-up instructions for the current Codex work session, not as requests "
                "for the Telegram bridge server to interpret. You may perform repository work, "
                "edit files, run builds/tests, and prepare commits when the user asks. Commit "
                "or push only when the user explicitly asks for commit/push or when the request "
                "clearly includes publishing. When reporting progress or completion, be concise "
                "and include changed files, verification, commit/push status, and next action "
                "when relevant."
            )
        return (
            "This is the user's personal assistant chat. Focus on schedules, reminders, saved "
            "personal records, weather/news briefings, and lightweight Gitea/status lookup. "
            "Do not edit repository files, run coding tasks, commit, or push from this chat. "
            "When the user describes development work here, summarize it, save it as a todo or "
            "create a Gitea issue when useful, and direct execution to the work chat or Codex app."
        )

    def _auto_select_repo_for_prompt(self, chat_id: int, prompt: str) -> Path | None:
        chat_state = self.state.get_chat(chat_id)
        if chat_state.repo_path:
            return None
        repo_path = self._detect_workspace_repo(prompt)
        if repo_path is None:
            return None
        self.state.set_repo(chat_id, repo_path)
        return repo_path

    def _detect_workspace_repo(self, prompt: str) -> Path | None:
        root = self.config.workspace_root.resolve()
        if not root.exists():
            return None
        tokens = _repo_name_tokens(prompt)
        if not tokens:
            return None
        children: dict[str, Path] = {}
        try:
            for child in root.iterdir():
                if child.is_dir():
                    children[child.name.casefold()] = child.resolve()
        except OSError:
            return None
        for token in tokens:
            candidate = children.get(token.casefold())
            if candidate and (candidate / ".git").exists():
                return candidate
        for token in tokens:
            candidate = children.get(token.casefold())
            if candidate:
                return candidate
        return None

    def _resolve_workspace_path(self, requested: str) -> Path:
        root = self.config.workspace_root.resolve()
        path = Path(requested)
        if not path.is_absolute():
            path = root / requested
        resolved = path.resolve()
        if resolved != root and root not in resolved.parents:
            raise ValueError(f"작업 경로는 workspace root 안에 있어야 합니다: {root}")
        return resolved


def _session_id_for_backend(chat_state: ChatState, backend: str) -> str | None:
    if not chat_state.session_id:
        return None
    if chat_state.session_backend == backend:
        return chat_state.session_id
    if chat_state.session_backend is None and backend == "exec":
        return chat_state.session_id
    return None


def _service_tier_for_speed(speed_mode: str) -> str | None:
    return "fast" if speed_mode == "fast" else None


def _display_speed(speed_mode: str) -> str:
    return "fast(고속)" if speed_mode == "fast" else "standard(보통)"


def _sandbox_mode_label(sandbox: str, config: Config) -> str:
    if sandbox == "danger-full-access" or sandbox == config.write_sandbox:
        return "full-access"
    if sandbox == "workspace-write":
        return "workspace-write"
    if sandbox == "read-only" or sandbox == config.readonly_sandbox:
        return "read-only"
    return sandbox


def _args_text(context: ContextTypes.DEFAULT_TYPE) -> str:
    return " ".join(context.args).strip()


def _preview_prompt(prompt: str, limit: int = 120) -> str:
    preview = " ".join(prompt.split())
    if len(preview) <= limit:
        return preview
    return preview[: limit - 3] + "..."


def _repo_name_tokens(prompt: str) -> list[str]:
    tokens = re.findall(r"[A-Za-z0-9][A-Za-z0-9_.-]{1,79}", prompt)
    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 _looks_like_status_question(text: str) -> bool:
    normalized = " ".join(text.casefold().split())
    exact = {
        "지금 뭐해",
        "지금 뭐해?",
        "뭐하고 있어",
        "뭐하고 있어?",
        "뭐 하는 중",
        "뭐 하는 중?",
        "어디까지 했어",
        "어디까지 했어?",
        "진행상황",
        "진행 상황",
        "상태",
        "status",
    }
    if normalized in exact:
        return True
    if _looks_like_work_instruction(normalized):
        return False
    if len(normalized) > 32:
        return False
    status_terms = ("진행상황", "진행 상황", "현재 상태", "작업 상태", "어디까지", "뭐 하는 중", "뭐하고")
    return any(term in normalized for term in status_terms)


def _looks_like_cancel_request(text: str) -> bool:
    normalized = " ".join(text.casefold().split())
    exact = {
        "취소",
        "취소해",
        "멈춰",
        "중지",
        "중지해",
        "그만",
        "그만해",
        "stop",
        "cancel",
    }
    if normalized in exact:
        return True
    if _looks_like_work_instruction(normalized):
        return False
    if len(normalized) > 28:
        return False
    cancel_terms = ("작업 취소", "작업 중지", "멈춰줘", "취소해줘", "그만해줘", "stop it", "cancel it")
    return any(term in normalized for term in cancel_terms)


def _looks_like_work_instruction(normalized: str) -> bool:
    work_terms = (
        "고쳐",
        "수정",
        "구현",
        "만들",
        "추가",
        "확인하고",
        "빌드",
        "테스트",
        "생성",
        "작업해",
        "진행해",
        "기능",
        "파일",
        "코드",
        "커밋",
        "푸시",
    )
    return any(term in normalized for term in work_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


def _extract_model_ids(payload: Any) -> list[str]:
    candidates: list[str] = []

    def visit(value: Any, key: str = "") -> None:
        if isinstance(value, dict):
            for nested_key, nested_value in value.items():
                visit(nested_value, str(nested_key).lower())
            return
        if isinstance(value, list):
            for item in value:
                visit(item, key)
            return
        if not isinstance(value, str):
            return
        if key in {"id", "model", "model_id", "name", "slug"} or _looks_like_model_id(value):
            if _looks_like_model_id(value):
                candidates.append(value.strip())

    visit(payload)
    return _dedupe(candidates)


def _looks_like_model_id(value: str) -> bool:
    value = value.strip()
    if not value or len(value) > 80 or " " in value:
        return False
    lowered = value.lower()
    if lowered.startswith(("gpt-", "o1", "o3", "o4")):
        return True
    return "codex" in lowered and re.match(r"^[a-z0-9._:/-]+$", lowered) is not None


def _dedupe(values: list[str]) -> list[str]:
    seen: set[str] = set()
    result: list[str] = []
    for value in values:
        value = value.strip()
        if not value or value in seen:
            continue
        seen.add(value)
        result.append(value)
    return result


def _sort_models(models: list[str], preferred: str | None) -> list[str]:
    latest = _latest_model(models)

    def key(model: str) -> tuple[int, tuple[int, ...], str]:
        if preferred and model == preferred:
            return (0, (), model)
        if model == latest:
            return (1, (), model)
        if "mini" not in model and "spark" not in model and model.startswith("gpt-"):
            return (2, tuple(-part for part in _model_version_parts(model)), model)
        if "mini" in model:
            return (3, tuple(-part for part in _model_version_parts(model)), model)
        if "spark" in model or "codex" in model:
            return (4, tuple(-part for part in _model_version_parts(model)), model)
        return (5, tuple(-part for part in _model_version_parts(model)), model)

    return sorted(_dedupe(models), key=key)


def _latest_model(models: list[str]) -> str:
    candidates = [
        model
        for model in models
        if model.startswith("gpt-") and "mini" not in model and "spark" not in model
    ]
    if not candidates:
        candidates = [model for model in models if model.startswith("gpt-")]
    if not candidates:
        return models[0] if models else "gpt-5.5"
    return max(candidates, key=lambda model: (_model_version_parts(model), model))


def _model_version_parts(model: str) -> list[int]:
    return [int(part) for part in re.findall(r"\d+", model)]


def _format_schedule_display(schedule: dict) -> str:
    schedule_type = schedule.get("schedule_type", "daily")
    if schedule_type == "interval":
        cadence = f"{int(schedule.get('interval_minutes') or 60)}분마다"
    else:
        cadence = (
            f"매일 {int(schedule.get('hour') or 8):02}:"
            f"{int(schedule.get('minute') or 0):02} {schedule.get('timezone') or 'Asia/Seoul'}"
        )
    status_map = {
        "active": "활성",
        "paused": "일시정지",
        "deleted": "삭제됨",
    }
    status = status_map.get(str(schedule.get("status", "active")), str(schedule.get("status", "active")))
    name = schedule.get("name") or "스케줄 작업"
    return f"#{schedule.get('id')} {name} - {cadence} - {status}"


def _format_record_display(record: dict) -> str:
    category = str(record.get("category") or "note")
    category_label = RECORD_CATEGORY_LABELS.get(category, category)
    status_map = {
        "active": "진행중",
        "done": "완료",
        "cancelled": "취소",
        "archived": "보관",
    }
    status = status_map.get(str(record.get("status") or "active"), str(record.get("status") or "active"))
    title = record.get("title") or "(제목 없음)"
    date_bits = []
    for label, key in (("일자", "date"), ("시작", "start_date"), ("종료", "end_date"), ("기한", "due_date")):
        if record.get(key):
            date_bits.append(f"{label}:{record[key]}")
    tags = record.get("tags") or []
    tag_text = f" #{' #'.join(str(tag) for tag in tags)}" if tags else ""
    date_text = f" ({', '.join(date_bits)})" if date_bits else ""
    return f"#{record.get('id')} [{category_label}] {title} - {status}{date_text}{tag_text}"


def _record_list_title(category: str | None, status: str | None, query: str | None) -> str:
    if query:
        return f"개인 기록 검색: {query}"
    if category:
        label = RECORD_CATEGORY_LABELS.get(category, category)
        if status == "active":
            return f"{label} 목록"
        return f"{label} 기록"
    return "개인 기록"


def _format_thread_list(result: dict, archived: bool) -> tuple[str, InlineKeyboardMarkup | None]:
    threads = result.get("data") or result.get("threads") or result.get("items") or []
    if not isinstance(threads, list) or not threads:
        label = "보관된 세션" if archived else "세션"
        keyboard = InlineKeyboardMarkup(
            [[InlineKeyboardButton("새로고침", callback_data="sessions_archived" if archived else "sessions")]]
        )
        return f"{label}: 없음", keyboard

    lines = ["보관된 세션:" if archived else "세션:"]
    buttons: list[list[InlineKeyboardButton]] = []
    for index, thread in enumerate(threads, start=1):
        if not isinstance(thread, dict):
            continue
        thread_id = str(thread.get("id") or thread.get("threadId") or "(unknown)")
        name = _readable_thread_field(thread.get("name") or thread.get("title"), "(untitled)")
        status = _readable_thread_field(thread.get("status"), "unknown")
        cwd = thread.get("cwd") or thread.get("workingDirectory") or ""
        updated = thread.get("updatedAt") or thread.get("updated_at") or ""
        lines.append(f"{index}. {name} [{status}]")
        lines.append(f"   id={thread_id}")
        if cwd:
            lines.append(f"   cwd={cwd}")
        if updated:
            lines.append(f"   updated={updated}")
        if thread_id != "(unknown)":
            action = "복원" if archived else "보관"
            action_callback = "unarchive_session" if archived else "archive_session"
            buttons.append(
                [
                    InlineKeyboardButton(f"사용 {index}", callback_data=f"use_session:{thread_id}"),
                    InlineKeyboardButton(f"{action} {index}", callback_data=f"{action_callback}:{thread_id}"),
                ]
            )

    next_cursor = result.get("nextCursor") or result.get("next_cursor")
    if next_cursor:
        lines.append(f"\nnext_cursor={next_cursor}")
    buttons.append(
        [InlineKeyboardButton("새로고침", callback_data="sessions_archived" if archived else "sessions")]
    )
    if not archived:
        buttons.append([InlineKeyboardButton("보관된 세션", callback_data="sessions_archived")])
    else:
        buttons.append([InlineKeyboardButton("활성 세션", callback_data="sessions")])
    return "\n".join(lines), InlineKeyboardMarkup(buttons)


def _readable_thread_field(value: object, fallback: str) -> str:
    if isinstance(value, str) and value.strip():
        return value.strip()
    if isinstance(value, dict):
        for key in ("text", "name", "title", "status", "type"):
            nested = value.get(key)
            if isinstance(nested, str) and nested.strip():
                return nested.strip()
    return fallback


def _scheduled_prompt(schedule: dict) -> str:
    return (
        "This is a scheduled Telegram bridge task. Run the task now and produce a concise "
        "Telegram-ready response. Do not create another schedule unless the task explicitly "
        "asks you to change scheduling. If repository history is relevant, prefer the Gitea "
        "commit tools. Do not describe a missing tool or unavailable detail as a permission "
        "problem unless an API error explicitly says permission was denied. Only include a "
        "commit summary when the task asks for repository history or a repository can be "
        "identified from the task/context.\n\n"
        f"Schedule: {format_schedule(schedule)}\n\n"
        f"Task:\n{schedule.get('prompt') or ''}"
    )


def _split_message(text: str, limit: int = 3500) -> list[str]:
    if len(text) <= limit:
        return [text]
    chunks: list[str] = []
    remaining = text
    while remaining:
        chunks.append(remaining[:limit])
        remaining = remaining[limit:]
    return chunks