"""utils/cron_timers_upsert.py — task-2533 cron --session timers upsert hook (신호등 sync fix A). 회장 §본질 (2026-05-10 / 신호등 sync fix A): cron ``--session`` 발사 직후 ``memory/task-timers.json``에 task entry가 자동으로 들어가지 않아서 신호등 (대시보드 활성 봇 표시) 100% gap이 발생함. 본 helper는 cron 발사 직후 ``upsert_cron_dispatch()``를 호출해 entry를 **atomic + idempotent** 으로 갱신한다. 회장 §명시 (task-2533): 1. atomic — 부분 쓰기 상태 0 (`utils/atomic_write.atomic_json_write` 사용) 2. idempotent — 동일 task 재발사 시 entry 중복 X (status='running' 그대로 갱신) 3. opt-out (read_only / analysis_only / report_only) 시에도 timers upsert (활성 표시 필수) 4. cron 발사 실패 시 hook이 호출되지 않음 — caller(safe_cron_dispatch) 책임 5. team_id 명시 (task md에서 추출 또는 ``--team`` 인자) 6. schedule_id raw 저장은 OK (cokacdir 내부 cron id, 비밀 아님) 7. chat_id 명시 (chat 격리 — 다른 chat entry와 충돌 0) 8. cron_prompt raw 저장 금지 (token 누수 위험) — sanitize 후 첫 80자만 description 보관 Public API: - ``upsert_cron_dispatch()`` — 핵심 entry upsert - ``extract_task_id_from_prompt()`` — cron prompt 첫 줄에서 ``task-NNNN`` 추출 - ``extract_team_id_from_task_md()`` — task md에서 ``devN-team`` 추출 - ``DEFAULT_TIMERS_PATH`` — ``memory/task-timers.json`` 영역 한정: - ❌ ``dashboard/data_loader.py`` 직접 수정 X (task-2534 영역) - ❌ ``utils/lifecycle_reconciliation_manager.py`` 직접 수정 X (task-2528 영역) - ❌ ``scripts/safe_cron_dispatch.py`` 본 모듈에 import 외 의존 X - ❌ ``--session`` 옵션 사용 X """ from __future__ import annotations import json import re from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Optional, Union from utils.atomic_write import atomic_json_write # pyright: ignore[reportMissingImports] __all__ = [ "DEFAULT_TIMERS_PATH", "TIMERS_TASK_KEY", "DESCRIPTION_MAX_LEN", "extract_task_id_from_prompt", "extract_team_id_from_task_md", "extract_team_id_from_text", "sanitize_description", "upsert_cron_dispatch", ] # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- # task-timers.json은 worktree root 기준 ``memory/task-timers.json``. # 본 모듈은 명시적 경로를 받지 않으면 caller worktree root (모듈 위치 기준)에서 해석한다. _WORKTREE_ROOT = Path(__file__).resolve().parent.parent DEFAULT_TIMERS_PATH = _WORKTREE_ROOT / "memory" / "task-timers.json" # task-timers.json 의 최상위 dict는 ``{"tasks": {: {...}}, "counter": N, ...}``. TIMERS_TASK_KEY = "tasks" # description 은 raw cron_prompt 가 아니라 sanitize 후 잘라 보관한다 (token 누수 방지). DESCRIPTION_MAX_LEN = 80 # task md 에서 ``devN-team`` 추출용. ``dev1-team / dev10-team`` 모두 매치. _TEAM_ID_TAGGED_RE = re.compile(r"\bdev([1-9]\d*)-team\b") # 약식 표기 (``dev1`` 만 있는 경우) 보조 매치. ``-team`` suffix 자동 부여. _TEAM_ID_BARE_RE = re.compile(r"\bdev([1-9]\d*)\b") # cron prompt 첫 줄에서 ``task-NNNN`` 또는 ``task-NNNN+M`` 추출. _TASK_ID_RE = re.compile(r"\btask-(\d+(?:\+\d+)?)\b") # raw token / raw key / raw session uuid 패턴 — description 에 들어가면 즉시 redact. # (회귀 6 — token raw 0 검증) _HEX_KEY_RE = re.compile(r"\b[0-9a-fA-F]{16,}\b") _UUID_RE = re.compile( r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b" ) _GHP_TOKEN_RE = re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b") # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def extract_task_id_from_prompt(prompt: str) -> Optional[str]: """cron prompt 텍스트에서 첫 ``task-NNNN`` 또는 ``task-NNNN+M`` 매치를 ``task-...`` 형태로 반환. None 반환 = task_id 식별 실패. caller 가 fallback (``unknown-task``) 결정. """ if not prompt: return None match = _TASK_ID_RE.search(prompt) if not match: return None return f"task-{match.group(1)}" def extract_team_id_from_text( text: str, *, fallback: Optional[str] = None, ) -> str: """임의의 텍스트에서 ``devN-team`` 표현을 추출 (캡슐화 보장 — caller 가 private regex 접근 X). 우선 순위: 1. ``devN-team`` (정식 태그) 첫 매치 2. ``devN`` (약식) 첫 매치 → ``-team`` suffix 자동 부여 3. fallback 인자 (caller 명시) 4. ``"unknown-team"`` """ if not text: return fallback or "unknown-team" tagged = _TEAM_ID_TAGGED_RE.search(text) if tagged: return f"dev{tagged.group(1)}-team" bare = _TEAM_ID_BARE_RE.search(text) if bare: return f"dev{bare.group(1)}-team" return fallback or "unknown-team" def extract_team_id_from_task_md( task_md_path: Union[str, Path], *, fallback: Optional[str] = None, ) -> str: """task md 파일에서 ``devN-team`` 표현 첫 매치 반환. 우선 순위: 1. ``devN-team`` (정식 태그) 첫 매치 2. ``devN`` (약식) 첫 매치 → ``-team`` suffix 자동 부여 3. fallback 인자 (caller 명시) 4. ``"unknown-team"`` 파일이 존재하지 않거나 읽기 실패 시 fallback / unknown-team 반환 (예외 없음). """ path = Path(task_md_path) if not path.is_file(): return fallback or "unknown-team" try: text = path.read_text(encoding="utf-8") except OSError: return fallback or "unknown-team" return extract_team_id_from_text(text, fallback=fallback) def sanitize_description(raw: str, *, max_len: int = DESCRIPTION_MAX_LEN) -> str: """raw 문자열에서 token / key / uuid 패턴을 redact 후 max_len 으로 자른다. 최종 길이는 max_len 이하 (suffix ``...`` 포함). 빈 문자열은 그대로 반환. """ if not raw: return "" redacted = _GHP_TOKEN_RE.sub("", raw) redacted = _UUID_RE.sub("", redacted) redacted = _HEX_KEY_RE.sub("", redacted) # 줄바꿈을 공백으로 통일 (description 한 줄 표시 정합). flat = " ".join(redacted.split()) if len(flat) <= max_len: return flat # max_len 안에 ``...``를 포함하도록 자른다. keep = max(0, max_len - 3) return flat[:keep] + "..." def _now_iso() -> str: """UTC ISO8601 타임스탬프 (timezone naive — 기존 timers entry 정합).""" return datetime.now(timezone.utc).replace(tzinfo=None).isoformat() def _read_timers(timers_path: Path) -> Dict[str, Any]: """task-timers.json 을 dict 로 읽는다. 파일 없거나 손상되면 빈 컨테이너 반환. 상위 ``tasks`` key 는 항상 dict 로 보장 (post-condition). """ if not timers_path.is_file(): return {TIMERS_TASK_KEY: {}} try: with timers_path.open("r", encoding="utf-8") as f: data = json.load(f) except (json.JSONDecodeError, OSError): # 손상된 timers 는 caller가 이미 알고 있어야 하므로 raise 가 더 안전하다. raise if not isinstance(data, dict): raise ValueError(f"timers root must be dict, got {type(data).__name__}") if TIMERS_TASK_KEY not in data or not isinstance(data[TIMERS_TASK_KEY], dict): data[TIMERS_TASK_KEY] = {} return data # --------------------------------------------------------------------------- # Public upsert API (회장 §본질) # --------------------------------------------------------------------------- def upsert_cron_dispatch( *, task_id: str, team_id: str, schedule_id: str, cron_prompt: str, chat_id: Union[int, str], timers_path: Optional[Union[str, Path]] = None, now_iso: Optional[str] = None, ) -> Dict[str, Any]: """cron 발사 직후 ``memory/task-timers.json[tasks][task_id]`` 에 entry upsert. 동작: 1. timers 파일 read (없으면 빈 컨테이너로 시작) 2. ``tasks[task_id]`` 갱신: - 신규 → ``status='running'``, ``start_time=now``, ``end_time=None``, ``schedule_id``, ``chat_id``, ``team_id``, ``description`` 명시. - 기존 status 가 ``running`` → idempotent (start_time 보존, schedule_id/last_dispatch_at 갱신) - 기존 status 가 종료 상태 → 새 run 으로 리셋 (start_time 갱신, end_time=None, status='running', schedule_id 갱신). 3. atomic write (``utils.atomic_write.atomic_json_write``) 4. upserted entry dict 반환 raw 저장 금지: - cron_prompt 는 ``description`` 으로 sanitize 후 ``DESCRIPTION_MAX_LEN`` 이하만 저장. - chat_id 는 str 로 변환 저장 (json 직렬화 정합). - schedule_id 는 cokacdir 내부 cron id (token 아님) — raw 저장 OK. 검증 (회귀 7 박제): 1. 신규 발사 → entry 생성 2. 동일 task 재발사 → 중복 X (idempotent) 3. opt-out 토큰 prompt 도 upsert 호출 시 entry 생성 (활성 표시 필수) 4. cron 발사 실패 시 본 helper 호출되지 않음 (caller 책임) 5. team_id 추출 정확성 (caller 가 옳은 team_id 전달) 6. schedule_id 누수 검증 (description 토큰 redact) 7. chat_id 격리 (entry 에 명시 저장) """ if not task_id: raise ValueError("task_id must be a non-empty str") if not team_id: raise ValueError("team_id must be a non-empty str") if not schedule_id: raise ValueError("schedule_id must be a non-empty str") target_path = Path(timers_path) if timers_path else DEFAULT_TIMERS_PATH ts = now_iso or _now_iso() data = _read_timers(target_path) tasks: Dict[str, Any] = data[TIMERS_TASK_KEY] existing = tasks.get(task_id) if not isinstance(existing, dict): existing = None description = sanitize_description(cron_prompt) chat_id_str = str(chat_id) if existing is None: entry: Dict[str, Any] = { "task_id": task_id, "team_id": team_id, "description": description, "start_time": ts, "end_time": None, "duration_seconds": None, "status": "running", "schedule_id": schedule_id, "chat_id": chat_id_str, "last_dispatch_at": ts, } else: # 기존 entry — idempotent 갱신. prior_status = existing.get("status") entry = dict(existing) # shallow copy — 원본 mutate 방지 # status 가 종료 상태 (completed / cancelled / error 등) 면 새 run 으로 리셋. if prior_status != "running": entry["start_time"] = ts entry["end_time"] = None entry["duration_seconds"] = None # status 는 항상 running 으로 강제 (cron 발사 시점 = 활성). entry["status"] = "running" entry["task_id"] = task_id # team_id 는 caller 가 권위적으로 전달했으므로 항상 갱신. entry["team_id"] = team_id # description 은 신규 prompt 기준으로 갱신 (활성 표시 정확성 우선). entry["description"] = description entry["schedule_id"] = schedule_id entry["chat_id"] = chat_id_str entry["last_dispatch_at"] = ts tasks[task_id] = entry data[TIMERS_TASK_KEY] = tasks atomic_json_write(target_path, data) return entry