archive_old_context.py

#!/usr/bin/env python3
"""Archive >7d closed-blocks from CURRENT_CONTEXT.md to CURRENT_CONTEXT_archive.md.

Запускается в /handoff перед обновлением CURRENT_CONTEXT.md. Парсит блоки
вида:
  > Закрыто YYYY-MM-DD ...
  ### Session NNN артефакты (YYYY-MM-DD, ...)

и переносит всё что старше 7 дней в архив. Атомарная запись через temp+replace.
Без LLM, без сетевых вызовов, быстро.

Цель: CURRENT_CONTEXT.md не раздувается незаметно — старое автоматически
уходит в архив, который НЕ автозагружается в новые сессии. Источник истины
для деталей каждой сессии — diary файлы.

Usage:
  python scripts/archive_old_context.py              # обычный запуск
  python scripts/archive_old_context.py --days 14    # порог 14 дней
  python scripts/archive_old_context.py --dry-run    # показать что бы сделал, без записи
"""
from __future__ import annotations

import argparse
import os
import re
import sys
import tempfile
from datetime import date, datetime, timedelta
from pathlib import Path

try:
    sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
except (AttributeError, OSError):
    pass

DATE_RE = re.compile(r"\d{4}-\d{2}-\d{2}")
TODAY = date.today()


def per_user_paths() -> tuple[Path, Path]:
    """Resolve CURRENT_CONTEXT.md + CURRENT_CONTEXT_archive.md in per-user folder."""
    project_dir_env = os.environ.get("CLAUDE_PROJECT_DIR")
    if not project_dir_env:
        print("[archive] CLAUDE_PROJECT_DIR not set — skip", file=sys.stderr)
        sys.exit(0)
    project_dir = Path(project_dir_env)
    pid = str(project_dir).replace(":", "-").replace("\\", "-").replace("/", "-")
    user_dir = Path.home() / ".claude" / "projects" / pid
    return (user_dir / "CURRENT_CONTEXT.md", user_dir / "CURRENT_CONTEXT_archive.md")


def is_archivable_start(line: str) -> date | None:
    """Возвращает дату блока если строка начинает архивируемый блок, иначе None.

    Архивируем:
      > Закрыто YYYY-MM-DD ...
      ### Session NNN артефакты (YYYY-MM-DD, ...)
    """
    if line.startswith("> Закрыто") or line.startswith("### Session"):
        m = DATE_RE.search(line)
        if m:
            try:
                return datetime.strptime(m.group(0), "%Y-%m-%d").date()
            except ValueError:
                return None
    return None


def is_section_boundary(line: str) -> bool:
    """Любое начало новой секции — закрывает текущий накапливаемый блок."""
    return (
        line.startswith("## ")
        or line.startswith("# ")
        or line.startswith("### ")
        or line.startswith("> Закрыто")
    )


def atomic_write(path: Path, content: str) -> None:
    """Write content to path atomically via temp file + os.replace."""
    fd, tmp_path = tempfile.mkstemp(prefix=".archive_", suffix=".tmp", dir=str(path.parent))
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
        os.replace(tmp_path, path)
    except Exception:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise


def archive_header() -> str:
    return (
        "<!-- АРХИВ — НЕ автозагружается. Источник истины каждой записи — соответствующий diary. -->\n"
        "# Current Context Archive\n\n"
        "> Здесь живут закрытые записи старше 7 дней и артефакты завершённых сессий.\n"
        "> Этот файл **НЕ автозагружается** в новые сессии — только для ручного просмотра.\n"
        "> Для каждой сессии есть полный diary в `memory/diary/NNN_YYYY-MM-DD.md` — источник истины.\n"
        "> Архив пополняется автоматически в `/handoff` через `scripts/archive_old_context.py`.\n\n"
        "---\n"
    )


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--days", type=int, default=7, help="threshold in days (default: 7)")
    parser.add_argument("--dry-run", action="store_true", help="report only, don't write")
    args = parser.parse_args()

    threshold = timedelta(days=args.days)
    ctx, archive = per_user_paths()

    if not ctx.exists():
        print(f"[archive] {ctx} not found — nothing to do")
        return

    lines = ctx.read_text(encoding="utf-8").splitlines(keepends=True)

    kept: list[str] = []
    archived_blocks: list[list[str]] = []
    current: tuple[date, list[str]] | None = None

    def close_current() -> None:
        nonlocal current
        if current is None:
            return
        block_date, block_lines = current
        age = TODAY - block_date
        if age >= threshold:
            archived_blocks.append(block_lines)
        else:
            kept.extend(block_lines)
        current = None

    for line in lines:
        if current is not None and is_section_boundary(line):
            close_current()
            # fall through to process this line as potentially new block start

        new_date = is_archivable_start(line)
        if new_date is not None:
            current = (new_date, [line])
        elif current is not None:
            current[1].append(line)
        else:
            kept.append(line)

    close_current()

    if not archived_blocks:
        print(f"[archive] nothing older than {args.days} days — no changes")
        return

    if args.dry_run:
        print(f"[archive] DRY-RUN: would move {len(archived_blocks)} block(s) to {archive.name}:")
        for blk in archived_blocks:
            first = blk[0].rstrip()
            print(f"  - {first[:100]}{'…' if len(first) > 100 else ''}")
        return

    archive.parent.mkdir(parents=True, exist_ok=True)
    if archive.exists():
        arc_text = archive.read_text(encoding="utf-8")
    else:
        arc_text = archive_header()

    arc_text += f"\n## Архивировано {TODAY.isoformat()}\n\n"
    for blk in archived_blocks:
        arc_text += "".join(blk)
        if not arc_text.endswith("\n\n"):
            arc_text += "\n" if not arc_text.endswith("\n") else ""
            arc_text += "\n"

    atomic_write(archive, arc_text)
    atomic_write(ctx, "".join(kept))

    print(
        f"[archive] moved {len(archived_blocks)} block(s) older than {args.days}d "
        f"from {ctx.name} → {archive.name}"
    )


if __name__ == "__main__":
    main()