diff --git a/harness/voice.py b/harness/voice.py index ba0c8ef..ce77b1d 100644 --- a/harness/voice.py +++ b/harness/voice.py @@ -106,8 +106,58 @@ def _download_voice(file_id: str, file_unique_id: str = ""): return audio, cache +# ─────────── тяжёлые воркеры ВНЕ cgroup поллера (t-261004160531-780666) ───── +# faster-whisper large-v3-turbo (int8) живёт в ~2-3G RSS, ffmpeg тоже кусается: +# заспавненные поллером как обычные дети они падали в ЕГО cgroup — systemd +# Consumption показывал 3.7G peak у simbiot-harness-tg. Теперь тяжёлое уходит +# в собственный transient-scope (systemd-run --scope) со своим MemoryMax; +# поллер остаётся лёгким (<100M), его юнит держит MemoryMax=1G. +# Нет systemd (локалка/тесты) — прозрачный fallback на прямой запуск. +_ASR_SCOPE_MEM = os.environ.get("SIMBIOT_ASR_SCOPE_MEM", "3G") +_HAS_SYSTEMD = os.path.isdir("/run/systemd/system") and os.path.exists("/usr/bin/systemd-run") + + +def _heavy_argv(argv): + """argv тяжёлого воркера → запуск в отдельном systemd-scope с MemoryMax.""" + if not _HAS_SYSTEMD: + return list(argv) + import uuid + unit = f"simbiot-asr-{uuid.uuid4().hex[:10]}" + return ["systemd-run", "--scope", "--quiet", "--collect", + f"--unit={unit}", f"-p", f"MemoryMax={_ASR_SCOPE_MEM}", + "--"] + list(argv) + + +def _heavy_run(argv, input_bytes=None, timeout=600): + """Тяжёлый subprocess: scope-изоляция памяти + killpg на таймауте. + + systemd-run --scope выполняет команду синхронно, поэтому таймаут + работает как раньше; start_new_session + killpg (паттерн + isolated_runner, e-261003004801) не оставляют осиротевший воркер, + если тик умер раньше него. + """ + import signal + proc = subprocess.Popen( + _heavy_argv(argv), stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, start_new_session=True) + try: + out, err = proc.communicate(input=input_bytes, timeout=timeout) + return proc.returncode, out or b"", err or b"" + except subprocess.TimeoutExpired: + try: + os.killpg(proc.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + pass + try: # дождаться смерти группы, чтобы зомби не сидели в scope + proc.communicate(timeout=10) + except Exception: + pass + return None, b"", b"" + + # ────────────── извлечение аудио из видеоконтейнера ───────────────────────── + def extract_audio_track(container: bytes): """Видеоконтейнер (mp4 и т.п.) → ogg/opus аудиодорожка через ffmpeg. @@ -120,16 +170,17 @@ def extract_audio_track(container: bytes): with tempfile.NamedTemporaryFile(prefix="simbiot-v_", delete=False) as tf: tf.write(container) src = tf.name - # -vn: только звук; моно 16к — формат, который любят оба ASR-яруса - proc = subprocess.run( + # -vn: только звук; моно 16к — формат, который любят оба ASR-яруса. + # ffmpeg тяжёлый → в свой scope (память поллера не растёт) + rc, out, err = _heavy_run( ["ffmpeg", "-v", "error", "-i", src, "-vn", "-ac", "1", "-ar", "16000", "-f", "ogg", "-c:a", "libopus", "-"], - capture_output=True, timeout=180) - if proc.returncode != 0 or not proc.stdout: - err = (proc.stderr or b"").decode()[:200] + timeout=180) + if rc != 0 or not out: + err = err.decode()[:200] print(f"[voice] ffmpeg: дорожка не извлечена ({err})", flush=True) return None - return proc.stdout + return out except Exception as e: print(f"[voice] ffmpeg error: {e}", flush=True) return None @@ -198,17 +249,21 @@ def _asr_faster_whisper(audio: bytes, filename: str): "audio_b64": base64.b64encode(audio).decode(), "filename": filename, }, ensure_ascii=False) - proc = subprocess.run( - [_FW_PYTHON, _FW_WORKER], input=req.encode(), - capture_output=True, timeout=600) - out = proc.stdout.decode() if proc.stdout else "" + # faster-whisper large-v3-turbo ~2-3G RSS: только в свой scope, + # иначе память ложится в cgroup поллера (3.7G peak, t-261004160531) + rc, out, err = _heavy_run( + [_FW_PYTHON, _FW_WORKER], input_bytes=req.encode(), timeout=600) + if rc is None: + print("[voice] faster-whisper: таймаут (группа убита)", flush=True) + return None + out = out.decode() if out else "" try: data = json.loads(out.strip().splitlines()[-1]) if out.strip() else {} except (ValueError, IndexError): - data = {"error": (proc.stderr or b"").decode()[:300]} - if data.get("error") or proc.returncode != 0: + data = {"error": err.decode()[:300]} + if data.get("error") or rc != 0: print(f"[voice] faster-whisper error: " - f"{data.get('error') or proc.returncode}", flush=True) + f"{data.get('error') or rc}", flush=True) return None return (data.get("text") or "").strip() except Exception as e: diff --git a/scripts/simbiot-harness-tg.service b/scripts/simbiot-harness-tg.service new file mode 100644 index 0000000..5971f74 --- /dev/null +++ b/scripts/simbiot-harness-tg.service @@ -0,0 +1,23 @@ +[Unit] +Description=Simbiot Harness - Telegram inbox poller +After=network-online.target simbiot-harness.service +Wants=network-online.target + +[Service] +Environment=SIMBIOT_HOME=__DATA_DIR__ +Environment=SIMBIOT_RAG_DB=/root/rag/rag.db +Type=simple +WorkingDirectory=__INSTALL_DIR__ +ExecStart=__INSTALL_DIR__/venv/bin/python -m harness.tg_inbox +Restart=always +RestartSec=5 +Environment=PYTHONUNBUFFERED=1 +# t-261004160531-780666: поллер лёгкий — тяжёлая транскрипция (faster-whisper +# large-v3-turbo ~2-3G, ffmpeg) вынесена в отдельные transient-scope'ы +# (systemd-run --scope, см. harness/voice.py _heavy_run) с их MemoryMax. +# Жёсткий потолок юнита: 3.7G peak больше невозможен физически. +MemoryMax=1G +MemoryHigh=768M + +[Install] +WantedBy=multi-user.target