d466f358ce
Implementeaza planul claude-master-plan-discord-bridge-20260830 (15 taskuri, 3 lane-uri paralele) — un bot subtire discord.py peste CLI-ul `claude`, cu proces persistent per fir alimentat pe stdin cu --input-format stream-json. Nucleu: runner (proces persistent + reaper 20min + respawn --resume), stream (parser tolerant), session_store (scriere atomica, lock per fir, detectare PID reuse, recovery), limits (max 4 procese, timeout tur, rate per user, plafon cost pe zi), render (un loop de editare per canal, interval adaptiv). Adaptor: allowlist guild/canal/user fail-closed cu respingerea webhook-urilor, comenzi !new/!cd/!model/!status/!stop/!cleanup, cost si model in subsolul fiecarui raspuns. Mesajul sosit in timpul unui tur devine steering, nu tur nou. Securitate: hook PreToolUse fail-closed care cere confirmare in Discord pentru operatiuni ireversibile, wrapper `infra` cu lista explicita de hosturi. Deny rules raman strat cosmetic, nu bariera (verificat: /usr/bin/ssh trece pe langa). Ops: alerte email pe conventia repo-ului, !cleanup pentru orfani, unit systemd user cu KillMode=control-group si limite de memorie, install.sh idempotent. Verificat: 275 teste fara retea/Discord/API (10.8s), identic cu si fara discord.py instalat; e2e pe CLI real confirma steering-ul mid-tur (mesaj la 6s intr-un tool call de 25s schimba raspunsul final). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B29CApsP1JkSdjYaGaHpE7
399 lines
14 KiB
Python
399 lines
14 KiB
Python
"""Proces `claude` PERSISTENT per fir, alimentat pe stdin cu --input-format stream-json.
|
|
|
|
De ce persistent (T4): asa se poate face steering mid-tur -- un mesaj trimis in timpul
|
|
unui tool call lung ajunge la model si schimba raspunsul final (verificat in plan).
|
|
Totul e asyncio: NICIODATA readline blocant. stderr are task propriu si buffer circular.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import collections
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, AsyncIterator, Awaitable, Callable
|
|
|
|
import config
|
|
import stream as stream_mod
|
|
|
|
log = logging.getLogger("discord-bridge.runner")
|
|
|
|
# linii mari (tool results) -> limita generoasa pe StreamReader
|
|
_READ_LIMIT = 16 * 1024 * 1024
|
|
STDERR_TAIL = 50
|
|
|
|
|
|
class TurnFailed(RuntimeError):
|
|
"""Turul nu s-a putut duce la capat (proces mort, EOF inainte de result)."""
|
|
|
|
|
|
class TurnTimeout(TurnFailed):
|
|
"""Turul a depasit timpul maxim."""
|
|
|
|
|
|
@dataclass
|
|
class SessionRestarted:
|
|
"""Eveniment sintetic: procesul murise, l-am repornit cu --resume."""
|
|
thread_id: str
|
|
sid: str | None
|
|
reason: str = "procesul claude murise"
|
|
|
|
@property
|
|
def text(self) -> str:
|
|
return f"sesiune repornita ({self.reason})"
|
|
|
|
|
|
@dataclass
|
|
class TurnOutcome:
|
|
result: stream_mod.Result | None
|
|
restarted: bool = False
|
|
events: list[Any] = field(default_factory=list)
|
|
|
|
|
|
def build_cmd(
|
|
claude_bin: str | list[str],
|
|
model: str,
|
|
sid: str | None = None,
|
|
settings: str | os.PathLike | None = None,
|
|
extra: tuple[str, ...] = (),
|
|
) -> list[str]:
|
|
"""Comanda exacta din plan; `--resume <sid>` doar la respawn."""
|
|
base = [claude_bin] if isinstance(claude_bin, str) else list(claude_bin)
|
|
cmd = base + [
|
|
"-p",
|
|
"--input-format", "stream-json",
|
|
"--output-format", "stream-json",
|
|
"--verbose",
|
|
"--permission-mode", "bypassPermissions",
|
|
]
|
|
if settings:
|
|
cmd += ["--settings", os.fspath(settings)]
|
|
cmd += ["--model", model, "--autocompact", "auto"]
|
|
if sid:
|
|
cmd += ["--resume", sid]
|
|
cmd += list(extra)
|
|
return cmd
|
|
|
|
|
|
def user_message(text: str) -> str:
|
|
return json.dumps(
|
|
{"type": "user", "message": {"role": "user", "content": [{"type": "text", "text": text}]}},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
|
|
class ClaudeProcess:
|
|
"""Un proces claude viu, legat de un fir Discord."""
|
|
|
|
def __init__(
|
|
self,
|
|
thread_id: str,
|
|
cwd: str,
|
|
model: str,
|
|
sid: str | None = None,
|
|
*,
|
|
claude_bin: str | list[str] | None = None,
|
|
settings: str | os.PathLike | None = None,
|
|
env: dict[str, str] | None = None,
|
|
on_pid: Callable[[str, int | None], Any] | None = None,
|
|
):
|
|
self.thread_id = str(thread_id)
|
|
self.cwd = cwd
|
|
self.model = model
|
|
self.sid = sid
|
|
self.claude_bin = claude_bin or config.get("CLAUDE_BIN", "claude")
|
|
self.settings = settings
|
|
self.env = env
|
|
self.on_pid = on_pid
|
|
self.proc: asyncio.subprocess.Process | None = None
|
|
self.stderr_buf: collections.deque[str] = collections.deque(maxlen=STDERR_TAIL)
|
|
self.last_active = time.monotonic()
|
|
self.inflight = False
|
|
self.started_turns = 0
|
|
self.restarts = 0
|
|
self._stderr_task: asyncio.Task | None = None
|
|
self._parser = stream_mod.StreamParser()
|
|
self._start_lock = asyncio.Lock()
|
|
|
|
# --------------------------------------------------------------- proces
|
|
@property
|
|
def alive(self) -> bool:
|
|
return self.proc is not None and self.proc.returncode is None
|
|
|
|
@property
|
|
def pid(self) -> int | None:
|
|
return self.proc.pid if self.proc is not None else None
|
|
|
|
async def start(self) -> None:
|
|
async with self._start_lock:
|
|
if self.alive:
|
|
return
|
|
cmd = build_cmd(self.claude_bin, self.model, self.sid, self.settings)
|
|
log.info("pornesc claude pentru firul %s: %s", self.thread_id, " ".join(cmd))
|
|
env = dict(os.environ)
|
|
# Lane B: hook-ul PreToolUse afla firul Discord din mediul procesului claude
|
|
env["CLAUDE_DISCORD_THREAD_ID"] = self.thread_id
|
|
if self.sid:
|
|
env["CLAUDE_DISCORD_SESSION_ID"] = self.sid
|
|
if self.env:
|
|
env.update(self.env)
|
|
self.proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
cwd=self.cwd,
|
|
env=env,
|
|
limit=_READ_LIMIT,
|
|
)
|
|
self.last_active = time.monotonic()
|
|
self._stderr_task = asyncio.create_task(self._pump_stderr())
|
|
if self.on_pid:
|
|
res = self.on_pid(self.thread_id, self.proc.pid)
|
|
if asyncio.iscoroutine(res):
|
|
await res
|
|
|
|
async def _pump_stderr(self) -> None:
|
|
assert self.proc is not None and self.proc.stderr is not None
|
|
try:
|
|
async for raw in self.proc.stderr:
|
|
line = raw.decode("utf-8", "replace").rstrip("\n")
|
|
if line:
|
|
self.stderr_buf.append(line)
|
|
except (asyncio.CancelledError, ValueError):
|
|
raise
|
|
except Exception as exc: # pragma: no cover
|
|
log.debug("stderr pump s-a oprit: %s", exc)
|
|
|
|
def stderr_tail(self, n: int = STDERR_TAIL) -> list[str]:
|
|
"""Ultimele linii de stderr, pentru `!status`."""
|
|
return list(self.stderr_buf)[-n:]
|
|
|
|
async def stop(self, grace: float = 5.0) -> None:
|
|
task, self._stderr_task = self._stderr_task, None
|
|
proc, self.proc = self.proc, None
|
|
if proc is not None and proc.returncode is None:
|
|
with contextlib.suppress(ProcessLookupError):
|
|
if proc.stdin and not proc.stdin.is_closing():
|
|
proc.stdin.close()
|
|
proc.terminate()
|
|
try:
|
|
await asyncio.wait_for(proc.wait(), grace)
|
|
except (asyncio.TimeoutError, ProcessLookupError):
|
|
with contextlib.suppress(ProcessLookupError):
|
|
proc.kill()
|
|
with contextlib.suppress(Exception):
|
|
await proc.wait()
|
|
if task is not None:
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError, Exception):
|
|
await task
|
|
if self.on_pid:
|
|
res = self.on_pid(self.thread_id, None)
|
|
if asyncio.iscoroutine(res):
|
|
await res
|
|
|
|
# ------------------------------------------------------------- mesaje
|
|
async def send(self, text: str) -> None:
|
|
"""Trimite un mesaj de utilizator pe stdin (si mid-tur: asta e steering-ul)."""
|
|
if not self.alive or self.proc is None or self.proc.stdin is None:
|
|
raise TurnFailed("procesul claude nu e viu")
|
|
self.proc.stdin.write((user_message(text) + "\n").encode("utf-8"))
|
|
await self.proc.stdin.drain()
|
|
self.last_active = time.monotonic()
|
|
|
|
async def _stdout_lines(self) -> AsyncIterator[str]:
|
|
assert self.proc is not None and self.proc.stdout is not None
|
|
reader = self.proc.stdout
|
|
while True:
|
|
try:
|
|
raw = await reader.readline()
|
|
except ValueError: # linie peste limita: nu doboram turul
|
|
log.warning("linie de stdout peste limita, ignorata")
|
|
continue
|
|
if not raw:
|
|
return
|
|
yield raw.decode("utf-8", "replace")
|
|
|
|
async def run_turn(
|
|
self,
|
|
prompt: str,
|
|
on_event: Callable[[Any], Awaitable[None]] | None = None,
|
|
timeout: float | None = None,
|
|
) -> TurnOutcome:
|
|
"""Un tur complet: (re)porneste procesul daca trebuie, trimite promptul,
|
|
consuma stream-ul pana la `result`."""
|
|
restarted = False
|
|
if not self.alive:
|
|
if self.started_turns:
|
|
restarted = True
|
|
self.restarts += 1
|
|
await self.start()
|
|
if timeout is None:
|
|
timeout = config.get_float("TURN_TIMEOUT_S", 900.0)
|
|
self.started_turns += 1
|
|
self.inflight = True
|
|
events: list[Any] = []
|
|
try:
|
|
if restarted:
|
|
ev = SessionRestarted(self.thread_id, self.sid)
|
|
events.append(ev)
|
|
if on_event:
|
|
await on_event(ev)
|
|
await self.send(prompt)
|
|
outcome = await asyncio.wait_for(self._consume(on_event, events), timeout)
|
|
return TurnOutcome(result=outcome, restarted=restarted, events=events)
|
|
except asyncio.TimeoutError:
|
|
await self.stop()
|
|
raise TurnTimeout(f"turul a depasit {timeout:.0f}s si a fost oprit") from None
|
|
except stream_mod.StreamEOFError as exc:
|
|
tail = "\n".join(self.stderr_tail(10))
|
|
await self.stop()
|
|
raise TurnFailed(f"{exc}; stderr:\n{tail}") from None
|
|
finally:
|
|
self.inflight = False
|
|
self.last_active = time.monotonic()
|
|
|
|
async def _consume(self, on_event, events: list) -> stream_mod.Result:
|
|
parser = stream_mod.StreamParser()
|
|
self._parser = parser
|
|
async for ev in parser.aiter_events(self._stdout_lines()):
|
|
if isinstance(ev, stream_mod.SystemInit) and ev.session_id:
|
|
self.sid = ev.session_id
|
|
events.append(ev)
|
|
if on_event:
|
|
await on_event(ev)
|
|
if isinstance(ev, stream_mod.Result):
|
|
return ev
|
|
raise stream_mod.StreamEOFError("stream inchis fara result") # pragma: no cover
|
|
|
|
|
|
class RunnerManager:
|
|
"""Registrul de procese vii + reaper-ul de inactivitate (20 min implicit)."""
|
|
|
|
def __init__(
|
|
self,
|
|
store=None,
|
|
*,
|
|
claude_bin: str | list[str] | None = None,
|
|
settings: str | os.PathLike | None = None,
|
|
idle_s: float | None = None,
|
|
poll_s: float = 30.0,
|
|
is_inflight: Callable[[str], bool] | None = None,
|
|
):
|
|
self.store = store
|
|
self.claude_bin = claude_bin or config.get("CLAUDE_BIN", "claude")
|
|
self.settings = settings
|
|
self.idle_s = idle_s if idle_s is not None else config.get_float("IDLE_REAP_S", 1200.0)
|
|
self.poll_s = poll_s
|
|
self._is_inflight = is_inflight
|
|
self.procs: dict[str, ClaudeProcess] = {}
|
|
self._reaper: asyncio.Task | None = None
|
|
|
|
# ------------------------------------------------------------ procese
|
|
def _on_pid(self, thread_id: str, pid: int | None):
|
|
if self.store is not None:
|
|
with contextlib.suppress(Exception):
|
|
self.store.set_pid(thread_id, pid)
|
|
|
|
def get(self, thread_id: str, cwd: str | None = None, model: str | None = None,
|
|
sid: str | None = None) -> ClaudeProcess:
|
|
key = str(thread_id)
|
|
proc = self.procs.get(key)
|
|
if proc is None:
|
|
proc = self.procs[key] = ClaudeProcess(
|
|
key,
|
|
cwd or config.get("DEFAULT_CWD", "/workspace"),
|
|
model or config.get("MODEL_DEFAULT", "sonnet"),
|
|
sid=sid,
|
|
claude_bin=self.claude_bin,
|
|
settings=self.settings,
|
|
on_pid=self._on_pid,
|
|
)
|
|
return proc
|
|
|
|
async def set_options(self, thread_id: str, cwd: str | None = None, model: str | None = None) -> bool:
|
|
"""`!cd` / `!model`: procesul curent se opreste, urmatorul tur il reia cu --resume."""
|
|
proc = self.get(thread_id)
|
|
changed = False
|
|
if cwd and cwd != proc.cwd:
|
|
proc.cwd, changed = cwd, True
|
|
if model and model != proc.model:
|
|
proc.model, changed = model, True
|
|
if changed and proc.alive:
|
|
await proc.stop()
|
|
return changed
|
|
|
|
async def reset(self, thread_id: str, fork: bool = False) -> None:
|
|
"""`!new`: sesiune noua (fork=True pastreaza sid-ul pentru un fir nou)."""
|
|
proc = self.procs.get(str(thread_id))
|
|
if proc is not None:
|
|
await proc.stop()
|
|
if not fork:
|
|
proc.sid = None
|
|
proc.started_turns = 0
|
|
|
|
def inflight(self, thread_id: str) -> bool:
|
|
proc = self.procs.get(str(thread_id))
|
|
if proc is not None and proc.inflight:
|
|
return True
|
|
if self._is_inflight is not None:
|
|
with contextlib.suppress(Exception):
|
|
return bool(self._is_inflight(str(thread_id)))
|
|
if self.store is not None:
|
|
with contextlib.suppress(Exception):
|
|
return bool(self.store.is_inflight(str(thread_id)))
|
|
return False
|
|
|
|
def live_count(self) -> int:
|
|
return sum(1 for p in self.procs.values() if p.alive)
|
|
|
|
# ------------------------------------------------------------- reaper
|
|
async def reap_once(self, now: float | None = None) -> list[str]:
|
|
"""Omoara procesele inactive. NICIODATA un fir cu tur in zbor."""
|
|
now = time.monotonic() if now is None else now
|
|
killed: list[str] = []
|
|
for key, proc in list(self.procs.items()):
|
|
if not proc.alive:
|
|
continue
|
|
if self.inflight(key):
|
|
continue
|
|
if now - proc.last_active >= self.idle_s:
|
|
log.info("reaper: opresc claude pentru firul %s (inactiv)", key)
|
|
await proc.stop()
|
|
killed.append(key)
|
|
return killed
|
|
|
|
async def _reaper_loop(self) -> None:
|
|
while True:
|
|
try:
|
|
await asyncio.sleep(self.poll_s)
|
|
await self.reap_once()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception: # pragma: no cover
|
|
log.exception("reaper a esuat, continui")
|
|
|
|
def start_reaper(self) -> asyncio.Task:
|
|
if self._reaper is None or self._reaper.done():
|
|
self._reaper = asyncio.create_task(self._reaper_loop())
|
|
return self._reaper
|
|
|
|
async def stop_all(self) -> None:
|
|
if self._reaper is not None:
|
|
self._reaper.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError, Exception):
|
|
await self._reaper
|
|
self._reaper = None
|
|
for proc in list(self.procs.values()):
|
|
await proc.stop()
|
|
|
|
|
|
def new_turn_id() -> str:
|
|
return uuid.uuid4().hex[:12]
|