fix: code review (#16)

src/takopi/backends.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from .runner import Runner

src/takopi/bridge.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import logging
 import time
-import inspect
 from collections import deque
 from collections.abc import AsyncIterator, Awaitable, Callable
 from dataclasses import dataclass, field
@@ -28,12 +27,6 @@ def _resolve_resume(
     return runner.extract_resume(text) or runner.extract_resume(reply_text)
 
 
-def _summarize_error(error: str | None) -> str:
-    if not error:
-        return "error"
-    return error
-
-
 def _log_runner_event(evt: TakopiEvent) -> None:
     for line in render_event_cli(evt):
         logger.info("[runner] %s", line)
@@ -41,7 +34,7 @@ def _log_runner_event(evt: TakopiEvent) -> None:
         if evt.ok:
             logger.info("[runner] done")
         else:
-            logger.info("[runner] error: %s", _summarize_error(evt.error))
+            logger.info("[runner] error: %s", evt.error or "error")
 
 
 def _is_cancel_command(text: str) -> bool:
@@ -516,7 +509,7 @@ async def handle_message(
         await cfg.bot.delete_message(chat_id=chat_id, message_id=progress_id)
 
 
-async def poll_updates(cfg: BridgeConfig):
+async def poll_updates(cfg: BridgeConfig) -> AsyncIterator[dict[str, Any]]:
     offset: int | None = None
     offset = await _drain_backlog(cfg, offset)
     await _send_startup(cfg)
@@ -605,7 +598,7 @@ async def _wait_for_resume(running_task: RunningTask) -> ResumeToken | None:
 
 async def _send_with_resume(
     bot: BotClient,
-    enqueue: Callable[[int, int, str, ResumeToken], Awaitable[None] | None],
+    enqueue: Callable[[int, int, str, ResumeToken], Awaitable[None]],
     running_task: RunningTask,
     chat_id: int,
     user_msg_id: int,
@@ -620,9 +613,7 @@ async def _send_with_resume(
             disable_notification=True,
         )
         return
-    result = enqueue(chat_id, user_msg_id, text, resume)
+    await enqueue(chat_id, user_msg_id, text, resume)
-    if inspect.isawaitable(result):
-        await result
 
 
 async def _run_main_loop(

src/takopi/cli.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-from typing import Callable
+from collections.abc import Callable
 
 import anyio
 import typer

src/takopi/config.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import shutil
 import tomllib
 from pathlib import Path
 
@@ -38,8 +39,7 @@ def _maybe_migrate_legacy(legacy_path: Path, target_path: Path) -> None:
         return
     try:
         target_path.parent.mkdir(parents=True, exist_ok=True)
-        raw = legacy_path.read_text(encoding="utf-8")
+        shutil.move(legacy_path, target_path)
-        target_path.write_text(raw, encoding="utf-8")
     except OSError as e:
         raise ConfigError(
             f"Failed to migrate legacy config {legacy_path} to {target_path}: {e}"
@@ -64,19 +64,23 @@ def load_telegram_config(path: str | Path | None = None) -> tuple[dict, Path]:
         cfg_path = Path(path).expanduser()
         return _read_config(cfg_path), cfg_path
 
-    for legacy, target in zip(_legacy_candidates(), _config_candidates(), strict=True):
+    config_candidates = _config_candidates()
+    legacy_candidates = _legacy_candidates()
+    for legacy, target in zip(legacy_candidates, config_candidates, strict=True):
         _maybe_migrate_legacy(legacy, target)
 
-    candidates = _config_candidates()
+    for candidate in config_candidates:
-    for candidate in candidates:
         if candidate.is_file():
             return _read_config(candidate), candidate
 
-    legacy_candidates = _legacy_candidates()
     for candidate in legacy_candidates:
         if candidate.is_file():
             return _read_config(candidate), candidate
 
-    if len(candidates) == 1:
+    checked: list[Path] = []
-        raise ConfigError("Missing takopi config.")
+    for candidate in [*config_candidates, *legacy_candidates]:
-    raise ConfigError("Missing takopi config.")
+        if candidate in checked:
+            continue
+        checked.append(candidate)
+    checked_display = ", ".join(str(candidate) for candidate in checked)
+    raise ConfigError(f"Missing takopi config. Checked: {checked_display}")

src/takopi/engines.py

@@ -2,14 +2,15 @@ from __future__ import annotations
 
 import importlib
 import pkgutil
+from collections.abc import Mapping
+from functools import cache
 from pathlib import Path
+from types import MappingProxyType
 from typing import Any
 
 from .backends import EngineBackend, EngineConfig
 from .config import ConfigError
 
-_BACKENDS: dict[str, EngineBackend] | None = None
-
 
 def _discover_backends() -> dict[str, EngineBackend]:
     import takopi.runners as runners_pkg
@@ -33,34 +34,30 @@ def _discover_backends() -> dict[str, EngineBackend]:
     return backends
 
 
-def _ensure_loaded() -> None:
+@cache
-    global _BACKENDS
+def _backends() -> Mapping[str, EngineBackend]:
-    if _BACKENDS is None:
+    backends = _discover_backends()
-        _BACKENDS = _discover_backends()
+    return MappingProxyType(backends)
 
 
 def get_backend(engine_id: str) -> EngineBackend:
-    _ensure_loaded()
+    backends = _backends()
-    assert _BACKENDS is not None
     try:
-        return _BACKENDS[engine_id]
+        return backends[engine_id]
     except KeyError as exc:
-        available = ", ".join(sorted(_BACKENDS))
+        available = ", ".join(sorted(backends))
         raise ConfigError(
             f"Unknown engine {engine_id!r}. Available: {available}."
         ) from exc
 
 
 def list_backends() -> list[EngineBackend]:
-    _ensure_loaded()
+    backends = _backends()
-    assert _BACKENDS is not None
+    return [backends[key] for key in sorted(backends)]
-    return [_BACKENDS[key] for key in sorted(_BACKENDS)]
 
 
 def list_backend_ids() -> list[str]:
-    _ensure_loaded()
+    return sorted(_backends())
-    assert _BACKENDS is not None
-    return sorted(_BACKENDS)
 
 
 def get_engine_config(

src/takopi/markdown.py

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import re
-from typing import Any, Callable
+from collections.abc import Callable
+from typing import Any
 
 from markdown_it import MarkdownIt
 from sulguk import transform_html

src/takopi/render.py

@@ -4,8 +4,8 @@ from __future__ import annotations
 
 import textwrap
 from collections import deque
+from collections.abc import Callable
 from pathlib import Path
-from typing import Callable
 
 from .model import Action, ActionEvent, ResumeToken, StartedEvent, TakopiEvent
 from .utils.paths import relativize_path

src/takopi/runner.py

@@ -126,7 +126,7 @@ class BaseRunner(SessionLockMixin):
         raise NotImplementedError
 
 
-@dataclass
+@dataclass(slots=True)
 class JsonlRunState:
     note_seq: int = 0
 
@@ -302,12 +302,12 @@ class JsonlSubprocessRunner(BaseRunner):
 
         tag = self.tag()
         logger = self.get_logger()
-        args = [self.command(), *self.build_args(prompt, resume, state=state)]
+        cmd = [self.command(), *self.build_args(prompt, resume, state=state)]
         payload = self.stdin_payload(prompt, resume, state=state)
         env = self.env(state=state)
 
         async with manage_subprocess(
-            *args,
+            cmd,
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
@@ -318,7 +318,7 @@ class JsonlSubprocessRunner(BaseRunner):
             if payload is not None and proc.stdin is None:
                 raise RuntimeError(self.pipes_error_message())
 
-            logger.debug("[%s] spawn pid=%s args=%r", tag, proc.pid, args)
+            logger.debug("[%s] spawn pid=%s args=%r", tag, proc.pid, cmd)
 
             if payload is not None:
                 assert proc.stdin is not None

src/takopi/runners/claude.py

@@ -31,7 +31,7 @@ _RESUME_RE = re.compile(
 )
 
 
-@dataclass
+@dataclass(slots=True)
 class ClaudeStreamState:
     pending_actions: dict[str, Action] = field(default_factory=dict)
     last_assistant_text: str | None = None

src/takopi/runners/codex.py

@@ -381,7 +381,7 @@ def translate_codex_event(event: dict[str, Any], *, title: str) -> list[TakopiEv
     return []
 
 
-@dataclass
+@dataclass(slots=True)
 class CodexRunState:
     note_seq: int = 0
     final_answer: str | None = None

src/takopi/telegram.py

@@ -72,14 +72,30 @@ class TelegramClient:
             return None
 
         try:
-            payload = resp.json()
+            resp.raise_for_status()
-        except Exception as e:
+        except httpx.HTTPStatusError as e:
+            body = resp.text
             logger.error(
-                "[telegram] bad response method=%s status=%s url=%s: %s",
+                "[telegram] http error method=%s status=%s url=%s: %s body=%r",
                 method,
                 resp.status_code,
                 resp.request.url,
                 e,
+                body,
+            )
+            return None
+
+        try:
+            payload = resp.json()
+        except Exception as e:
+            body = resp.text
+            logger.error(
+                "[telegram] bad response method=%s status=%s url=%s: %s body=%r",
+                method,
+                resp.status_code,
+                resp.request.url,
+                e,
+                body,
             )
             return None
 

src/takopi/utils/subprocess.py

@@ -3,7 +3,9 @@ from __future__ import annotations
 import logging
 import os
 import signal
+from collections.abc import AsyncIterator, Sequence
 from contextlib import asynccontextmanager
+from typing import Any
 
 import anyio
 from anyio.abc import Process
@@ -52,11 +54,13 @@ def kill_process(proc: Process) -> None:
 
 
 @asynccontextmanager
-async def manage_subprocess(*args, **kwargs):
+async def manage_subprocess(
+    cmd: Sequence[str], **kwargs: Any
+) -> AsyncIterator[Process]:
     """Ensure subprocesses receive SIGTERM, then SIGKILL after a 2s timeout."""
     if os.name == "posix":
         kwargs.setdefault("start_new_session", True)
-    proc = await anyio.open_process(args, **kwargs)
+    proc = await anyio.open_process(cmd, **kwargs)
     try:
         yield proc
     finally:

tests/test_exec_bridge.py

@@ -715,7 +715,9 @@ async def test_send_with_resume_waits_for_token() -> None:
     bot = _FakeBot()
     sent: list[tuple[int, int, str, ResumeToken | None]] = []
 
-    def enqueue(chat_id: int, user_msg_id: int, text: str, resume: ResumeToken) -> None:
+    async def enqueue(
+        chat_id: int, user_msg_id: int, text: str, resume: ResumeToken
+    ) -> None:
         sent.append((chat_id, user_msg_id, text, resume))
 
     running_task = RunningTask()
@@ -748,7 +750,9 @@ async def test_send_with_resume_reports_when_missing() -> None:
     bot = _FakeBot()
     sent: list[tuple[int, int, str, ResumeToken | None]] = []
 
-    def enqueue(chat_id: int, user_msg_id: int, text: str, resume: ResumeToken) -> None:
+    async def enqueue(
+        chat_id: int, user_msg_id: int, text: str, resume: ResumeToken
+    ) -> None:
         sent.append((chat_id, user_msg_id, text, resume))
 
     running_task = RunningTask()

tests/test_subprocess.py

@@ -16,9 +16,11 @@ async def test_manage_subprocess_kills_when_terminate_times_out(
     monkeypatch.setattr(subprocess_utils, "wait_for_process", fake_wait_for_process)
 
     async with subprocess_utils.manage_subprocess(
-        sys.executable,
+        [
-        "-c",
+            sys.executable,
-        "import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(10)",
+            "-c",
+            "import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(10)",
+        ]
     ) as proc:
         assert proc.returncode is None