"""anu_v2.worktree_cleanup — 6대 안전조건 기반 worktree cleanup helper (task-2550+1 clean replacement).

회장 §명시 (2026-05-11 C방안 승인 + 2026-05-12 task-2550+1 replacement):
  - 81개 누적 .worktrees 정리 — dry-run 우선, --apply는 별도 승인
  - 6대 안전조건 AND 게이트 — 어느 하나라도 FAIL 시 skip + log
  - main worktree 절대 보호 (workspace_root path 차단)
  - one-way isolation: anu_v2 외부 import 금지

6대 안전조건:
  1. task .done.acked 마커 존재
  2. PR state = MERGED (gh API)
  3. task .merge-done 마커 존재
  4. branch가 main에 ancestor (git merge-base --is-ancestor)
  5. worktree 사용 중 X (git worktree list lock + pgrep -f + lsof +D CWD)
  6. dry-run default; --apply 명시 시에만 실제 삭제

dirty worktree skip + log (memory/events/worktree-cleanup-skipped-<ts>-<sha8>.json)

task-2550+1 fix:
  - HIGH: `task_id in headRefName` 부분 일치 제거 → strict regex 경계 일치
          (task-25 vs task-2550 / task-250 vs task-2500 boundary test 강제)
  - medium #1: `abs(hash(path)) % 10**8` 비결정론 → `hashlib.sha256` 결정론 hash
  - medium #2: `pgrep -f` CWD-only 한계 → `lsof +D` 보완 (AND 게이트, 둘 다 안전해야 PASS)
  - medium #3: `r.all_safe`가 dry-run 에서 항상 False → cleanup_candidates 별도 산정
              (post_merge_smoke_runner 측 fix; 본 모듈은 helper 제공)
"""

from __future__ import annotations

import hashlib
import json
import re
import subprocess
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Callable

# 상수
DEFAULT_CHAT_ID: int = 6937032012
_KST = timezone(timedelta(hours=9))

# token mask 패턴 (post_merge_smoke_runner와 동일 패턴 — anu_v2 isolation으로 import X)
_TOKEN_PREFIX_RE = re.compile(
    r"(?:ghp_|ghs_|github_pat_)[A-Za-z0-9_\-]+",
    re.IGNORECASE,
)
TOKEN_KEY_HINTS = (
    "github_token", "bot_github_token", "gh_token", "owner_pat",
    "x-api-key", "authorization", "secret", "password",
)
_KEY_VALUE_RE = re.compile(
    r"(?i)(" + "|".join(re.escape(h) for h in TOKEN_KEY_HINTS) + r")"
    r"([=:\s]+[^\s,;\"']{1,200})",
)

# task-2550+1 HIGH fix: task_id strict 경계 매칭 (부분 일치 제거)
#   - "task-25" 가 "task/task-2550-dev5" 에 매칭되는 substring 오탐 차단
#   - 경계: 앞 (^/_-) 뒤 (+_.\-$) — task_id 의 numeric 부분 (+suffix 포함) 만 strict 일치
_TASK_ID_NUM_RE = re.compile(r"task-(\d+(?:\+\d+)?(?:\.\d+)?)")


def _sanitize_text(text: object) -> str:
    """raw token / API key 마스킹."""
    s = text if isinstance(text, str) else str(text)
    s = _TOKEN_PREFIX_RE.sub("***MASKED***", s)
    s = _KEY_VALUE_RE.sub(lambda m: m.group(1) + m.group(2)[0] + "***MASKED***", s)
    return s


def _matches_task_id_strict(head_ref_name: str, task_id: str) -> bool:
    """task_id 가 head_ref_name 에 strict 경계로 포함되는지 검사.

    task-2550+1 HIGH fix:
      - 기존 ``task_id in headRefName`` 는 task-25 vs task-2550 substring 오탐 발생.
      - 신규 두 단 게이트:
        (1) regex 경계 일치 ``(^|sep)task-{num}(sep|$)`` — substring 오탐 1차 차단.
        (2) head_ref_name 에서 추출한 task_id == query task_id — replacement-chain
            (task-2550 vs task-2550+1) 분리 (단순 ``in`` 비교로는 분리 불가).
      두 게이트 AND PASS 인 경우만 strict 일치로 판정.

    Args:
      head_ref_name: GitHub PR headRefName (예: "task/task-2550-dev5")
      task_id: "task-2550" 또는 "task-2550+1"

    Returns:
      strict match True 시 (task_id 가 head_ref_name 의 task identifier 와 정확히 일치).

    boundary test (test_worktree_cleanup_2550plus1):
      - task-25 ↔ "task/task-2550-dev5": False (regex 경계 차단 — "task-25" 뒤가 숫자 "5")
      - task-2500 ↔ "task/task-25-dev5": False
      - task-250 ↔ "task/task-2500-dev5": False
      - task-2550 ↔ "task/task-2550-dev5": True
      - task-2550 ↔ "task/task-2550+1-dev5": False (extracted "task-2550+1" != "task-2550")
      - task-2550+1 ↔ "task/task-2550+1-dev5": True
    """
    if not head_ref_name or not task_id:
        return False
    # 게이트 1: task_id 형식 검증 + numeric 부분 추출 (task-2550 → 2550 / task-2550+1 → 2550+1)
    m = re.match(r"^task-(\d+(?:\+\d+)?(?:\.\d+)?)$", task_id)
    if not m:
        return False
    num_part = m.group(1)
    # 게이트 2 (regex 경계): 앞 (^ 또는 영숫자/+ 가 아님), 뒤 ($ 또는 영숫자/+ 가 아님)
    #   - 뒤 경계에 + 포함: task-2550 검색 시 "task-2550+1" 의 + 가 boundary 가 아니라 식별자 일부.
    pattern = rf"(?:^|[^A-Za-z0-9+])task-{re.escape(num_part)}(?:$|[^A-Za-z0-9+])"
    if not re.search(pattern, head_ref_name):
        return False
    # 게이트 3 (extracted == query): head_ref_name 에서 task_id 추출 후 동등 비교
    extracted = _extract_task_id_str(head_ref_name)
    return extracted == task_id


def _extract_task_id_str(s: str) -> str | None:
    """module-level helper — branch / path / headRefName 에서 task_id 추출.

    `WorktreeCleanup._extract_task_id` 와 동일 로직 (free function form).
    """
    m = _TASK_ID_NUM_RE.search(s)
    return f"task-{m.group(1)}" if m else None


@dataclass
class WorktreeCandidate:
    """worktree 후보 1개."""
    path: str           # 절대 경로
    branch: str         # task/task-2474-dev2 등
    task_id: str | None # branch 패턴에서 추출
    head_sha: str       # commit SHA


@dataclass
class SafetyConditionResult:
    """6대 안전조건 결과."""
    condition: int      # 1~6
    name: str           # "done_acked", "pr_merged", ...
    passed: bool
    detail: str         # FAIL 사유


@dataclass
class CleanupResult:
    """worktree 1개 cleanup 결과."""
    worktree_path: str
    task_id: str | None
    safety_results: list[SafetyConditionResult]
    all_safe: bool
    dirty: bool
    is_main: bool
    applied: bool       # 실제 삭제 여부 (dry-run False)
    skipped: bool       # skip 발동 여부
    skip_reason: str | None
    ts: str             # KST ISO8601


def is_safe_ignoring_apply(result: CleanupResult) -> bool:
    """task-2550+1 medium fix: dry-run 에서도 cleanup_candidates 산정 가능하도록
    safety_results 의 condition 1~5 만 PASS 인지 확인하는 helper.

    safety_6 (apply_explicit) 은 dry-run 에서 항상 FAIL → all_safe 항상 False → candidate 0.
    → cleanup_candidates 의 본질 (가시성) 회복을 위해 1~5 만 검사.

    main / dirty 도 candidate 에서 제외.
    """
    if result.is_main or result.dirty:
        return False
    if not result.safety_results:
        return False
    # safety_results 가 비어있을 수도 있음 (skip 경로). 1~5 PASS 만 검사.
    safety_1_to_5 = [sr for sr in result.safety_results if sr.condition != 6]
    if len(safety_1_to_5) != 5:
        return False
    return all(sr.passed for sr in safety_1_to_5)


class WorktreeCleanup:
    """6대 안전조건 기반 worktree cleanup helper.

    외부 부수효과는 모두 Callable 주입 가능.
    """

    def __init__(
        self,
        *,
        subprocess_runner: Callable[..., subprocess.CompletedProcess] | None = None,
        clock: Callable[[], datetime] | None = None,
        workspace_root: Path | None = None,
    ) -> None:
        self._subprocess_runner = subprocess_runner if subprocess_runner is not None else subprocess.run
        self._clock = clock if clock is not None else (lambda: datetime.now(tz=_KST))
        self._workspace_root = workspace_root if workspace_root is not None else Path("/home/jay/workspace")

    # ─── 6대 안전조건 ────────────────────────────────────────────────────

    def check_safety_1_done_acked(self, task_id: str) -> SafetyConditionResult:
        """안전조건 1: memory/events/<task_id>.done.acked 존재."""
        marker = self._workspace_root / "memory" / "events" / f"{task_id}.done.acked"
        passed = marker.exists()
        return SafetyConditionResult(
            condition=1, name="done_acked", passed=passed,
            detail=("ok" if passed else f"missing: {marker}"),
        )

    def check_safety_2_pr_merged(self, task_id: str, branch: str | None = None) -> SafetyConditionResult:
        """안전조건 2: gh API로 task의 PR state == MERGED 확인.

        gh API 호출 전략:
          1. branch가 주어지면 `--head <branch suffix>` 정확 매칭 시도
          2. branch가 None이거나 1번 실패 시 head 없이 전체 검색 후 headRefName에서
             task_id strict 경계 일치 (task-2550+1 HIGH fix — substring 오탐 차단)
        gh API 실패 시 FAIL (보수적). 어느 dev 팀(dev1~dev7)이든 매칭 가능.
        """
        # branch에서 head_ref 추출 (refs/heads/ prefix 제거)
        head_ref: str | None = None
        if branch:
            head_ref = branch[len("refs/heads/"):] if branch.startswith("refs/heads/") else branch
        try:
            cmd: list[str] = ["gh", "pr", "list", "--state", "all", "--json", "number,state,headRefName"]
            if head_ref:
                cmd.extend(["--head", head_ref])
            else:
                cmd.extend(["--search", task_id])
            proc = self._subprocess_runner(cmd, capture_output=True, text=True, check=False, cwd=str(self._workspace_root), timeout=30)
            if proc.returncode != 0:
                return SafetyConditionResult(
                    condition=2, name="pr_merged", passed=False,
                    detail=f"gh api failed: {_sanitize_text(proc.stderr)[:200]}",
                )
            prs = json.loads(proc.stdout or "[]")
            if not prs:
                return SafetyConditionResult(condition=2, name="pr_merged", passed=False, detail="no PR found")
            # task-2550+1 HIGH fix: strict 경계 일치만 인정 (substring 오탐 차단).
            #   - 기존: `task_id in str(headRefName)` → task-25 가 task-2550-dev5 에 매칭 (BUG)
            #   - 신규: `_matches_task_id_strict` regex 경계 일치
            candidates = [
                pr for pr in prs
                if _matches_task_id_strict(str(pr.get("headRefName", "")), task_id)
            ]
            if not candidates:
                return SafetyConditionResult(
                    condition=2, name="pr_merged", passed=False,
                    detail=f"no PR with strict task_id={task_id} match in headRefName (unrelated PR rejected)",
                )
            for pr in candidates:
                if pr.get("state") == "MERGED":
                    return SafetyConditionResult(condition=2, name="pr_merged", passed=True, detail=f"PR #{pr.get('number')} MERGED")
            return SafetyConditionResult(condition=2, name="pr_merged", passed=False, detail=f"PR state: {[p.get('state') for p in candidates]}")
        except (subprocess.TimeoutExpired, json.JSONDecodeError, OSError) as e:
            return SafetyConditionResult(condition=2, name="pr_merged", passed=False, detail=f"error: {_sanitize_text(str(e))[:100]}")

    def check_safety_3_merge_done(self, task_id: str) -> SafetyConditionResult:
        """안전조건 3: memory/events/<task_id>.merge-done 존재."""
        marker = self._workspace_root / "memory" / "events" / f"{task_id}.merge-done"
        passed = marker.exists()
        return SafetyConditionResult(
            condition=3, name="merge_done", passed=passed,
            detail=("ok" if passed else f"missing: {marker}"),
        )

    def check_safety_4_branch_in_main(self, branch: str) -> SafetyConditionResult:
        """안전조건 4: git merge-base --is-ancestor <branch> origin/main → 0 (PASS).

        branch가 main에 머지되었으면 ancestor.
        """
        try:
            cmd = ["git", "merge-base", "--is-ancestor", branch, "origin/main"]
            proc = self._subprocess_runner(cmd, capture_output=True, text=True, check=False, cwd=str(self._workspace_root), timeout=15)
            passed = (proc.returncode == 0)
            return SafetyConditionResult(
                condition=4, name="branch_in_main", passed=passed,
                detail=("ancestor of main" if passed else f"NOT ancestor (rc={proc.returncode})"),
            )
        except (subprocess.TimeoutExpired, OSError) as e:
            return SafetyConditionResult(condition=4, name="branch_in_main", passed=False, detail=f"error: {str(e)[:100]}")

    def check_safety_5_not_in_use(self, worktree_path: str) -> SafetyConditionResult:
        """안전조건 5: worktree 사용 중 X (3단 AND 게이트).

        세 단의 검사 — 어느 하나라도 FAIL 이면 사용 중으로 판정:
          (a) git worktree list --porcelain: 해당 path 가 locked / prunable 이 아닌지 확인
          (b) pgrep -f <worktree_path>: argv 에 path 포함된 process 미감지
          (c) lsof +D <worktree_path>: CWD/file-handle 로 path 사용 중인 process 미감지
              (task-2550+1 medium fix — `pgrep -f` 의 CWD-only 누락 보완)
        """
        # (a) git worktree list --porcelain 검사
        try:
            cmd_a = ["git", "worktree", "list", "--porcelain"]
            proc_a = self._subprocess_runner(
                cmd_a, capture_output=True, text=True, check=False,
                cwd=str(self._workspace_root), timeout=15,
            )
            if proc_a.returncode != 0:
                return SafetyConditionResult(
                    condition=5, name="not_in_use", passed=False,
                    detail=f"git worktree list failed: rc={proc_a.returncode}",
                )
            # porcelain output에서 path block 찾기
            locked_or_prunable = False
            in_target_block = False
            for line in proc_a.stdout.splitlines():
                if line.startswith("worktree "):
                    in_target_block = line[len("worktree "):].strip() == worktree_path
                elif in_target_block and line.strip() in ("locked", "prunable"):
                    locked_or_prunable = True
                    break
                elif in_target_block and line.startswith(("locked ", "prunable ")):
                    locked_or_prunable = True
                    break
            if locked_or_prunable:
                return SafetyConditionResult(
                    condition=5, name="not_in_use", passed=False,
                    detail="worktree is locked or prunable per git worktree list",
                )
        except (subprocess.TimeoutExpired, OSError) as e:
            return SafetyConditionResult(condition=5, name="not_in_use", passed=False, detail=f"git worktree list error: {str(e)[:100]}")

        # (b) pgrep -f 검사 (argv 매칭)
        try:
            cmd_b = ["pgrep", "-f", worktree_path]
            proc_b = self._subprocess_runner(cmd_b, capture_output=True, text=True, check=False, timeout=15)
            # pgrep returncode 0 = 매치 있음 (사용 중) → FAIL
            if proc_b.returncode == 0:
                pid_count = len(proc_b.stdout.strip().splitlines())
                return SafetyConditionResult(condition=5, name="not_in_use", passed=False, detail=f"{pid_count} process(es) using path (pgrep argv)")
            elif proc_b.returncode != 1:
                return SafetyConditionResult(condition=5, name="not_in_use", passed=False, detail=f"pgrep rc={proc_b.returncode}")
        except (subprocess.TimeoutExpired, OSError) as e:
            return SafetyConditionResult(condition=5, name="not_in_use", passed=False, detail=f"pgrep error: {str(e)[:100]}")

        # (c) lsof +D 검사 (CWD / file-handle 매칭) — task-2550+1 medium fix
        try:
            cmd_c = ["lsof", "+D", worktree_path]
            proc_c = self._subprocess_runner(cmd_c, capture_output=True, text=True, check=False, timeout=15)
            # lsof returncode 0 = 매치 있음 (사용 중) → FAIL
            # returncode 1 = 매치 없음 (안전) → PASS (보수: stderr 무시 안 함)
            if proc_c.returncode == 0:
                # 첫 줄 header 제외 후 라인 카운트
                lines = proc_c.stdout.strip().splitlines()
                fh_count = max(len(lines) - 1, 0)
                return SafetyConditionResult(
                    condition=5, name="not_in_use", passed=False,
                    detail=f"{fh_count} file-handle(s) open under path (lsof +D CWD)",
                )
            elif proc_c.returncode != 1:
                # lsof 가 access denied / signal 등 비정상 종료 시 보수적 FAIL
                return SafetyConditionResult(
                    condition=5, name="not_in_use", passed=False,
                    detail=f"lsof rc={proc_c.returncode}",
                )
        except (subprocess.TimeoutExpired, OSError) as e:
            # lsof 부재 (FileNotFoundError) 시 보수적 FAIL — 운영 환경에서 lsof 필수
            return SafetyConditionResult(condition=5, name="not_in_use", passed=False, detail=f"lsof error: {str(e)[:100]}")

        return SafetyConditionResult(
            condition=5, name="not_in_use", passed=True,
            detail="git list OK + pgrep no match + lsof no CWD/fh",
        )

    def check_safety_6_apply_explicit(self, apply_flag: bool) -> SafetyConditionResult:
        """안전조건 6: dry-run default; --apply 명시 시에만 PASS."""
        # 본 조건은 cleanup_worktree 호출 시점의 apply 파라미터 명시성 검증.
        # PASS 의미: "실제 삭제를 허용함". dry-run 모드에서는 본 조건 PASS X (실제 삭제 X).
        return SafetyConditionResult(
            condition=6, name="apply_explicit", passed=apply_flag,
            detail=("--apply specified" if apply_flag else "dry-run (no actual delete)"),
        )

    # ─── 추가 안전 ──────────────────────────────────────────────────────

    def is_dirty_worktree(self, worktree_path: str) -> bool:
        """uncommitted changes 존재 여부.

        `git status --porcelain` 결과가 비어있지 않으면 dirty.
        """
        try:
            cmd = ["git", "-C", worktree_path, "status", "--porcelain"]
            proc = self._subprocess_runner(cmd, capture_output=True, text=True, check=False, timeout=15)
            return bool(proc.stdout.strip())
        except (subprocess.TimeoutExpired, OSError):
            # 오류 시 보수적으로 dirty 취급 → skip
            return True

    def is_main_worktree(self, worktree_path: str) -> bool:
        """★ workspace_root와 동일 경로면 main worktree → 절대 삭제 X."""
        try:
            wp = Path(worktree_path).resolve()
            wr = self._workspace_root.resolve()
            return wp == wr
        except OSError:
            return True  # 보수적으로 main 취급

    # ─── enumerate ──────────────────────────────────────────────────────

    def enumerate_worktrees(self) -> list[WorktreeCandidate]:
        """git worktree list로 모든 worktree 열거.

        main worktree는 결과에 포함되지만, cleanup_worktree에서 차단.
        """
        try:
            cmd = ["git", "worktree", "list", "--porcelain"]
            proc = self._subprocess_runner(cmd, capture_output=True, text=True, check=False, cwd=str(self._workspace_root), timeout=15)
            if proc.returncode != 0:
                return []
            candidates: list[WorktreeCandidate] = []
            blocks = proc.stdout.strip().split("\n\n")
            for blk in blocks:
                lines = blk.strip().splitlines()
                if not lines:
                    continue
                path = ""
                head = ""
                branch = ""
                for line in lines:
                    if line.startswith("worktree "):
                        path = line[len("worktree "):].strip()
                    elif line.startswith("HEAD "):
                        head = line[len("HEAD "):].strip()
                    elif line.startswith("branch "):
                        branch = line[len("branch "):].strip()
                if path:
                    task_id = self._extract_task_id(branch or path)
                    candidates.append(WorktreeCandidate(path=path, branch=branch, task_id=task_id, head_sha=head))
            return candidates
        except (subprocess.TimeoutExpired, OSError):
            return []

    @staticmethod
    def _extract_task_id(s: str) -> str | None:
        """branch / path에서 task_id 추출 (예: task/task-2474-dev2 → task-2474, task/task-2550+1-dev5 → task-2550+1)."""
        return _extract_task_id_str(s)

    # ─── cleanup ───────────────────────────────────────────────────────

    def cleanup_worktree(self, candidate: WorktreeCandidate, apply: bool = False) -> CleanupResult:
        """단일 worktree cleanup 시도. 6대 안전조건 AND 검증 후 실행."""
        ts = self._clock().isoformat()

        # ★ main worktree 절대 보호 — 회장 가시성 확보 위해 log 박제
        if self.is_main_worktree(candidate.path):
            self._log_skipped(candidate, "main_worktree_protected", ts)
            return CleanupResult(
                worktree_path=candidate.path, task_id=candidate.task_id,
                safety_results=[], all_safe=False, dirty=False, is_main=True,
                applied=False, skipped=True, skip_reason="main worktree (workspace_root) — never deleted",
                ts=ts,
            )

        # dirty 검사
        dirty = self.is_dirty_worktree(candidate.path)
        if dirty:
            self._log_skipped(candidate, "dirty", ts)
            return CleanupResult(
                worktree_path=candidate.path, task_id=candidate.task_id,
                safety_results=[], all_safe=False, dirty=True, is_main=False,
                applied=False, skipped=True, skip_reason="dirty worktree (uncommitted changes)",
                ts=ts,
            )

        # task_id 추출 실패 시 skip
        if not candidate.task_id:
            return CleanupResult(
                worktree_path=candidate.path, task_id=None,
                safety_results=[], all_safe=False, dirty=False, is_main=False,
                applied=False, skipped=True, skip_reason="task_id cannot be extracted",
                ts=ts,
            )

        # 6대 안전조건 검증
        results = [
            self.check_safety_1_done_acked(candidate.task_id),
            self.check_safety_2_pr_merged(candidate.task_id, branch=candidate.branch),
            self.check_safety_3_merge_done(candidate.task_id),
            self.check_safety_4_branch_in_main(candidate.branch),
            self.check_safety_5_not_in_use(candidate.path),
            self.check_safety_6_apply_explicit(apply),
        ]
        all_safe = all(r.passed for r in results)

        applied = False
        skipped = False
        skip_reason: str | None = None

        if not all_safe:
            skipped = True
            failed = [r.name for r in results if not r.passed]
            failed_set = set(failed)
            # check_safety_6_apply_explicit(apply=False) 단독 FAIL은 dry-run 정상 경로 — skip_reason을 명확화
            if failed_set == {"apply_explicit"}:
                skip_reason = "dry-run mode (apply=False)"
            else:
                skip_reason = f"safety failed: {failed}"
            # 운영 감사성: safety 1~5 FAIL 시에도 log 박제 (apply_explicit 단독 FAIL은 dry-run 정상 동작이므로 제외)
            non_apply_failures = failed_set - {"apply_explicit"}
            if non_apply_failures:
                self._log_skipped(candidate, f"safety_failed:{sorted(non_apply_failures)}", ts)
        else:
            # 실제 삭제: git worktree remove
            try:
                cmd = ["git", "worktree", "remove", candidate.path]
                proc = self._subprocess_runner(cmd, capture_output=True, text=True, check=False, cwd=str(self._workspace_root), timeout=30)
                if proc.returncode == 0:
                    applied = True
                else:
                    skipped = True
                    skip_reason = f"git worktree remove failed: {_sanitize_text(proc.stderr)[:200]}"
            except (subprocess.TimeoutExpired, OSError) as e:
                skipped = True
                skip_reason = f"error: {_sanitize_text(str(e))[:100]}"

        return CleanupResult(
            worktree_path=candidate.path, task_id=candidate.task_id,
            safety_results=results, all_safe=all_safe, dirty=False, is_main=False,
            applied=applied, skipped=skipped, skip_reason=skip_reason,
            ts=ts,
        )

    def cleanup_all_dry_run(self, apply: bool = False) -> list[CleanupResult]:
        """모든 worktree 후보 검사. dry-run default."""
        results: list[CleanupResult] = []
        for cand in self.enumerate_worktrees():
            results.append(self.cleanup_worktree(cand, apply=apply))
        return results

    # ─── log ───────────────────────────────────────────────────────────

    def _log_skipped(self, candidate: WorktreeCandidate, reason: str, ts: str) -> None:
        """skip log 박제 → memory/events/worktree-cleanup-skipped-<ts_compact>-<sha8>.json.

        task-2550+1 medium fix:
          - 기존 `abs(hash(path)) % 10**8` 는 Python `hash()` 의 PYTHONHASHSEED 의존
            (비결정론) → audit log 파일명 충돌 / 재현 어려움.
          - 신규: `hashlib.sha256(path).hexdigest()[:8]` 결정론 hash.

        ts + path hash 로 파일명 유일성 확보 (동시 실행 시 덮어쓰기 방지).
        branch / worktree_path 필드는 _sanitize_text 로 token-like 노출 방어.
        """
        try:
            ts_compact = ts.replace(":", "-").replace(".", "-")
            # task-2550+1 medium fix: 결정론 sha256 hash (Python hash() 비결정론 제거)
            path_hash = hashlib.sha256(candidate.path.encode("utf-8")).hexdigest()[:8]
            log_path = self._workspace_root / "memory" / "events" / f"worktree-cleanup-skipped-{ts_compact}-{path_hash}.json"
            log_path.parent.mkdir(parents=True, exist_ok=True)
            payload = {
                "ts": ts,
                "task_id": _sanitize_text(candidate.task_id) if candidate.task_id else None,
                "worktree_path": _sanitize_text(candidate.path),
                "branch": _sanitize_text(candidate.branch),
                "reason": reason,
                "chat_id": DEFAULT_CHAT_ID,
            }
            with open(log_path, "w", encoding="utf-8") as f:
                json.dump(payload, f, ensure_ascii=False, indent=2)
        except OSError:
            pass