diff --git a/files/claude_agents.py b/files/claude_agents.py index 6b1e86c..f47a22e 100644 --- a/files/claude_agents.py +++ b/files/claude_agents.py @@ -25,7 +25,9 @@ import json import os import re import shutil +import socket import subprocess +import sys import time import urllib.error import urllib.parse @@ -1383,21 +1385,66 @@ async def handle(reader, writer): writer.close() -async def main(connection): +# ---------- запуск ---------- +# Два режима одного файла: служба launchd (--headless) держит порт и всё считает, +# скрипт AutoLaunch в iTerm2 только показывает её страницу в Toolbelt. + +SERVICE_LABEL = "com.andrey.claude-agents" # метка LaunchAgent и идентификатор web view + + +async def serve(): + """Сервер панели без iTerm2: страница, дерево сессий, лимиты и баланс. Работает до отмены.""" global _server load_config() _server = await asyncio.start_server(handle, "127.0.0.1", PORT) - asyncio.create_task(refresher()) - asyncio.create_task(usage_poller()) - asyncio.create_task(balance_poller()) + tasks = [asyncio.create_task(c) for c in (refresher(), usage_poller(), balance_poller())] + try: + async with _server: + await _server.serve_forever() + finally: + for t in tasks: + t.cancel() + + +def port_alive(port, host="127.0.0.1", timeout=0.5) -> bool: + try: + with socket.create_connection((host, port), timeout=timeout): + return True + except OSError: + return False + + +def kickstart_service(): + """Просит launchd запустить службу панели, если она не запущена.""" + try: + subprocess.run(["launchctl", "kickstart", f"gui/{os.getuid()}/{SERVICE_LABEL}"], + capture_output=True, timeout=10) + except Exception: + pass + + +async def register_toolbelt(connection): + """Скрипт AutoLaunch: регистрирует web view со страницей службы; свой сервер не поднимает.""" + load_config() + if not port_alive(PORT): + kickstart_service() await iterm2.tool.async_register_web_view_tool( connection, display_name="Claude", - identifier="com.andrey.claude-agents", + identifier=SERVICE_LABEL, reveal_if_already_registered=False, url=f"http://127.0.0.1:{PORT}/", ) -if __name__ == "__main__" and iterm2 is not None: - iterm2.run_forever(main) +def run(argv): + if "--headless" in argv: + asyncio.run(serve()) + elif iterm2 is not None: + iterm2.run_forever(register_toolbelt) + else: + raise SystemExit("нужен iTerm2 или флаг --headless") + + +if __name__ == "__main__": + run(sys.argv[1:]) diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..70a5127 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,167 @@ +import asyncio +import json +import os +import socket +import sys +import types +import unittest +import urllib.request +from unittest import mock + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "files")) +import claude_agents as ca # noqa: E402 + + +def free_port(): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def http_get(port, path): + with urllib.request.urlopen(f"http://127.0.0.1:{port}{path}", timeout=3) as r: + return r.status, r.read().decode("utf-8") + + +class PortAlive(unittest.TestCase): + def test_port_alive(self): + s = socket.socket() + s.bind(("127.0.0.1", 0)) + s.listen() + port = s.getsockname()[1] + self.assertTrue(ca.port_alive(port)) + s.close() + self.assertFalse(ca.port_alive(port)) + + +class Serve(unittest.TestCase): + def setUp(self): + self.port = free_port() + patches = [ + mock.patch.object(ca, "PORT", self.port), + mock.patch.object(ca, "CONFIG_PATH", "/nonexistent/claude-agents/config.json"), + mock.patch.object(ca, "load_sessions", lambda: []), + mock.patch.object(ca, "fetch_usage", lambda: {}), + mock.patch.object(ca, "fetch_balance", lambda: ({}, "test")), + ] + for p in patches: + p.start() + self.addCleanup(p.stop) + + async def _with_server(self, check): + task = asyncio.create_task(ca.serve()) + loop = asyncio.get_running_loop() + try: + for _ in range(60): + if ca.port_alive(self.port): + break + await asyncio.sleep(0.05) + return await check(loop) + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + def test_serve_answers(self): + async def check(loop): + status, body = await loop.run_in_executor(None, http_get, self.port, "/") + self.assertEqual(status, 200) + self.assertIn('