From 4f0ae2cc8861657bdcc3c4487863805b643dbe9c Mon Sep 17 00:00:00 2001 From: y2keui Date: Wed, 10 Jun 2026 02:24:01 +0900 Subject: [PATCH] Run Telegram Codex jobs in background --- app/bot.py | 151 ++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 143 insertions(+), 8 deletions(-) diff --git a/app/bot.py b/app/bot.py index b7f09ad..a76604e 100644 --- a/app/bot.py +++ b/app/bot.py @@ -3,6 +3,8 @@ from __future__ import annotations import asyncio import json import re +import time +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -32,6 +34,7 @@ HELP_TEXT = """Telegram-Codex 연결 브리지 /repo <경로> - /workspaces 아래 작업 폴더 설정 /ask <내용> - 내용을 Codex로 전달 /work <내용> - 파일 수정, 커밋, 푸시 같은 쓰기 작업 실행 +/cancel - 현재 진행 중인 작업 취소 /new - 이 채팅의 Codex 세션 초기화 /status - 브리지 상태 확인 /settings - 모델, 추론강도, 속도 설정 확인 @@ -82,6 +85,14 @@ RECORD_CATEGORY_LABELS = { } +@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 @@ -100,6 +111,7 @@ class TelegramCodexBot: ) self.chat_clients: dict[int, PersistentAppServerRunner] = {} self.model_menus: dict[int, list[str]] = {} + self.running_jobs: dict[int, RunningJob] = {} self.scheduler_task: asyncio.Task | None = None def build_app(self) -> Application: @@ -117,6 +129,7 @@ class TelegramCodexBot: 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("new", self.new)) app.add_handler(CommandHandler("status", self.status)) app.add_handler(CommandHandler("settings", self.settings)) @@ -145,8 +158,11 @@ class TelegramCodexBot: 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, ) @@ -253,10 +269,26 @@ class TelegramCodexBot: 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 + 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 + job.task.cancel() + await self._close_client_for_chat(chat_id) + await self._reply(update, f"작업 취소를 요청했습니다.\n{self._format_job_status(chat_id)}") + async def status(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await self._guard(update): return @@ -267,6 +299,7 @@ class TelegramCodexBot: 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( @@ -278,6 +311,7 @@ class TelegramCodexBot: f"persistent_ws={persistent_status}", f"persistent_ws_유휴제한={self.config.persistent_ws_idle_seconds}초", f"스케줄={len(schedules)}개", + job_status, self._format_runtime_settings(), ] ), @@ -900,15 +934,73 @@ class TelegramCodexBot: sandbox: str | None = None, ) -> None: chat_id = update.effective_chat.id - if update.effective_chat: - await update.effective_chat.send_action(ChatAction.TYPING) - result, fallback_note = await self._run_codex_for_chat( - chat_id=chat_id, - prompt=prompt, - sandbox=sandbox or self.config.readonly_sandbox, - update_chat_state=True, + active_job = self.running_jobs.get(chat_id) + if active_job and not active_job.task.done(): + await self._reply( + update, + "이미 작업이 진행 중입니다.\n" + f"{self._format_job_status(chat_id)}\n\n" + "끝난 뒤 새 요청을 보내거나 /status로 상태를 확인해 주세요.", + ) + return + + selected_sandbox = sandbox or self.config.readonly_sandbox + 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, + ) ) - await self._reply_result(update, result, fallback_note) + self.running_jobs[chat_id] = RunningJob( + task=task, + started_monotonic=time.monotonic(), + prompt_preview=preview, + sandbox=selected_sandbox, + ) + mode = "쓰기 작업" if selected_sandbox != self.config.readonly_sandbox else "조회 작업" + await self._reply( + update, + f"{mode}을 시작했습니다.\n" + f"요청: {preview}\n" + "완료되면 결과를 보내드릴게요. 진행 중에는 /status로 확인할 수 있습니다.", + ) + + 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) + + async def _job_heartbeat(self, bot: Any, chat_id: int) -> None: + await asyncio.sleep(60) + while True: + job_status = self._format_job_status(chat_id) + await self._send_chat(bot, chat_id, f"아직 작업 중입니다.\n{job_status}") + await asyncio.sleep(120) async def _reply_result(self, update: Update, result: CodexResult, fallback_note: str = "") -> None: if result.ok: @@ -924,6 +1016,28 @@ class TelegramCodexBot: 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, @@ -1143,6 +1257,20 @@ class TelegramCodexBot: ] ) + 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 = "write" if job.sandbox != self.config.readonly_sandbox else "read-only" + return ( + "현재작업=진행 중\n" + f"작업모드={mode}\n" + f"경과={minutes}분 {seconds}초\n" + f"요청={job.prompt_preview}" + ) + def _with_bridge_context( self, user_prompt: str, @@ -1253,6 +1381,13 @@ 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 _checked_label(label: str, checked: bool) -> str: return f"✓ {label}" if checked else label