"""세션 상태를 주기적으로 남긴다 — 긴 작업이 끊겨도 이어갈 수 있게. `autodrive` 와 역할이 다르다. autodrive 는 **고친다**(멈춘 스텝을 회수·재개). 이쪽은 **기록만** 한다. 감시를 여러 개 띄우지 않기 위해 판정 로직은 새로 쓰지 않고 autodrive 것을 그대로 부른다 — 따로 구현하면 한쪽이 틀렸을 때 같이 틀려서 서로를 검증하지 못한다. 남기는 것 넷: · 재실행 진행 — `step_run` 상태 집계 (DB 가 유일한 진실) · 마지막 활동 — `autodrive.idle_seconds()` (DB·산출 파일·로그 세 겹 중 최근) · 개발 산출 — 이번 작업 디렉토리의 파일 수와 최신 시각 · 미커밋 수 — 손이 어디까지 갔는지 ★**매 사이클 기록한다.** 변화가 있을 때만 적으면 멈춘 구간이 로그에서 통째로 비어, 나중에 "왜 안 움직였나"를 물을 근거가 사라진다. autodrive 를 쓰면서 같은 실수를 이미 한 번 했다. 사용: cd backend && .venv/bin/python session_monitor.py """ from __future__ import annotations import subprocess import sys import time from datetime import datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import autodrive # noqa: E402 — 판정 로직 재사용 PERIOD = 300 WORK_DIR = Path("../artifact/20260808_세그먼테이션_저작") SNAPSHOT = WORK_DIR / "SESSION.md" REPO = Path("..") #: 파이프라인이 **돌고 있는데** 활동이 이만큼 없으면 눈에 띄게 적는다. #: 고치지는 않는다 — 그건 autodrive 몫이고, 여기서 손대면 두 손이 같은 #: 자물쇠를 잡는다. ★도는 것이 없으면 경보를 내지 않는다 — 2026-08-08 #: 14:31 완주 뒤에도 "★40분 정지"를 계속 내서, 끝난 것을 멈춘 것으로 #: 읽게 만들었다(그 직후 완주 스텝에 force 가 날아갔다). 완주와 정지를 #: 구분하지 못하는 경보는 없느니만 못하다. STUCK_S = 1800 def sh(args: list[str], cwd: Path | None = None) -> str: try: out = subprocess.run(args, capture_output=True, text=True, cwd=cwd, timeout=30) return (out.stdout or "").strip() except Exception: # noqa: BLE001 — 기록이 본 작업을 막지 않는다 return "" def step_states() -> dict[str, int]: sql = ("SELECT status, count(*) FROM step_run " f"WHERE episode_id='{autodrive.EPI}' GROUP BY status;") out = subprocess.run( ["psql", "-h", "localhost", "-U", "theroad", "-d", "theroad", "-t", "-A", "-F", "|", "-c", sql], capture_output=True, text=True, env={"PGPASSWORD": "theroad_dev_2026", "PATH": "/usr/bin:/bin:/opt/homebrew/bin"}) states: dict[str, int] = {} for ln in (out.stdout or "").strip().splitlines(): if "|" in ln: k, v = ln.split("|", 1) try: states[k.strip()] = int(v) except ValueError: pass return states def work_output() -> tuple[int, int | None]: """작업 디렉토리의 파일 수와 마지막 갱신이 몇 초 전인가. ★자기 스냅샷(SESSION.md)은 뺀다. 안 빼면 매 주기 자기가 쓴 파일을 "마지막 산출"로 재서, 산출이 몇 시간 멈춰도 항상 "주기초 전"으로 보인다(실측: 한 시간 내내 "300s 전" 고착). """ if not WORK_DIR.exists(): return 0, None newest, n = 0.0, 0 for p in WORK_DIR.rglob("*"): try: if p.is_file() and p != SNAPSHOT: n += 1 newest = max(newest, p.stat().st_mtime) except OSError: pass return n, (int(time.time() - newest) if newest else None) def uncommitted() -> int: return len([ln for ln in sh(["git", "status", "--porcelain"], REPO).splitlines() if ln.strip()]) def opik_status() -> str: """Opik 관점 — 최근 30분 트레이스 수·오류 수·최신 나이. llm_call_log 에는 우회 기록 호출만 남는다(텍스트 litellm 호출은 안 남는다) — 전수는 Opik 뿐이라 모니터가 이것도 본다(2026-08-08 지시). 조회 실패는 '?' — 기록 보조가 본 작업을 막지 않는다. """ code = ( "from dotenv import load_dotenv; load_dotenv('.env')\n" "import datetime, opik\n" "c = opik.Opik()\n" "cut = (datetime.datetime.now(datetime.timezone.utc)" " - datetime.timedelta(minutes=30)).strftime('%Y-%m-%dT%H:%M:%SZ')\n" "ts = c.search_traces(project_name='theroad-scene-lab'," " filter_string=f'start_time >= \"{cut}\"', max_results=500)\n" "err = sum(1 for t in ts if getattr(t, 'error_info', None))\n" "new = max((t.start_time for t in ts if t.start_time), default=None)\n" "age = int((datetime.datetime.now(datetime.timezone.utc) - new)" ".total_seconds()) if new else '?'\n" "print(f'30분 {len(ts)}건/오류 {err}/최신 {age}s전')\n" ) return sh([".venv/bin/python", "-c", code]) or "?" def write_snapshot(line: str, states: dict[str, int], idle: int | None, files: int, out_idle: int | None, dirty: int, stall_min: int | None) -> None: """사람이 읽고 바로 이어갈 수 있는 현재 상태 한 장.""" WORK_DIR.mkdir(parents=True, exist_ok=True) running = sh(["psql", "-h", "localhost", "-U", "theroad", "-d", "theroad", "-t", "-A", "-c", "SELECT string_agg(step_id, ', ') FROM step_run " f"WHERE episode_id='{autodrive.EPI}' AND status='running';"]) body = [ "# 세션 현재 상태", "", f"마지막 갱신: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (KST)", "", "## 재실행", "", f"- 상태 집계: {states or '(못 읽음)'}" " — ★한 번도 안 돈 스텝은 `step_run` 에 행이 없어 여기 안 잡힌다." " 정확한 남은 수는 `backend/autodrive.log` (진행 API 를 본다)", f"- 실행 중: {running or '(없음)'}", f"- 마지막 활동: {idle if idle is not None else '판단 불가'}초 전" " (DB·산출 파일·로그 세 겹 중 최근)", "", "## 개발 (씬 분해 규칙 저작)", "", f"- 작업 디렉토리 파일: {files}개", f"- 마지막 산출: {out_idle if out_idle is not None else '없음'}초 전", f"- 미커밋: {dirty}개", "", ] if stall_min is not None: body += [ "## ★막힘 의심", "", f"실행 중인 스텝이 있는데 **{stall_min}분간** 활동이 없다.", "회수는 autodrive 가 한다(여기서 손대면 두 손이 같은 자물쇠를 잡는다).", "그래도 안 풀리면 `backend/autodrive.log` 의 마지막 줄부터 본다.", "", ] body += ["## 최근 기록", "", "```", line, "```", ""] SNAPSHOT.write_text("\n".join(body), encoding="utf-8") def main() -> None: autodrive.log(f"session_monitor 시작 — 주기 {PERIOD}s, 스냅샷 {SNAPSHOT}") while True: states = step_states() idle = autodrive.idle_seconds() files, out_idle = work_output() dirty = uncommitted() opik = opik_status() # ★정지 경보는 "돌고 있는데 활동이 없다"일 때만. 도는 스텝이 없으면 # 완주든 대기든 멈춤이 아니라 **유휴**다 — 완주 뒤에 정지 경보를 # 내면 끝난 것을 멈춘 것으로 읽게 만든다(2026-08-08 실제 사례). running_n = states.get("running", 0) stall_min: int | None = None if running_n > 0 and idle is not None and idle >= STUCK_S: stall_min = idle // 60 # ★파생 수치를 지어내지 않는다. `step_run` 에는 **한 번도 안 돈 스텝의 # 행이 아예 없어서**, "총계 − 완료"로 남은 수를 계산하면 틀린다 # (실측: 상태 합 72인데 실제 스텝은 83). 있는 상태만 그대로 적고, # 정확한 남은 수는 진행 API 를 보는 autodrive 쪽 기록에 맡긴다. shown = " ".join(f"{k} {v}" for k, v in sorted(states.items())) or "(못 읽음)" line = (f"{shown}" f" · 활동 {idle if idle is not None else '?'}s 전" f" · 산출 {files}개({out_idle if out_idle is not None else '-'}s 전)" f" · opik {opik}" f" · 미커밋 {dirty}" + (f" · ★파이프라인 {stall_min}분 무활동(실행 중 {running_n})" if stall_min is not None else "") + (" · 파이프라인 유휴" if running_n == 0 else "")) autodrive.log(line) # 매 사이클 — 조용한 구간을 남기지 않는다 try: write_snapshot(line, states, idle, files, out_idle, dirty, stall_min) except Exception as exc: # noqa: BLE001 autodrive.log(f"스냅샷 기록 실패(무시): {exc}") time.sleep(PERIOD) if __name__ == "__main__": main()