path: root/scripts/compact.py

#!/usr/bin/env python3
"""Compaction job for Sénia, runs from host cron just after midnight.

For each invocation:
1. compact the previous day's diary entries into a daily archive
   (bootstrap case: include all earlier entries too, filed under yesterday)
2. regenerate the rolling N-day recent.md
3. on the 1st of the month, also compact the prior month's dailies into a monthly archive
"""

import re
import sqlite3
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path


SENIA_DB = Path("/etc/senia/data/senia.db")
MEMORY_DIR = Path("/etc/senia/data/memory")
DAILY_DIR = MEMORY_DIR / "daily"
MONTHLY_DIR = MEMORY_DIR / "monthly"
RECENT_FILE = MEMORY_DIR / "recent.md"
WORKSPACE = Path("/workspace")

CLAUDE_TIMEOUT = 600
ROLLING_WINDOW_DAYS = 10


# ---------- db / fs ----------

def diary_entries(start_iso: str | None, end_iso: str) -> list[tuple[str, str]]:
    conn = sqlite3.connect(SENIA_DB)
    try:
        if start_iso is None:
            return [
                (row[0], row[1])
                for row in conn.execute(
                    "SELECT ts, content FROM diary_logs WHERE ts < ? ORDER BY ts",
                    (end_iso,),
                ).fetchall()
            ]
        return [
            (row[0], row[1])
            for row in conn.execute(
                "SELECT ts, content FROM diary_logs WHERE ts >= ? AND ts < ? ORDER BY ts",
                (start_iso, end_iso),
            ).fetchall()
        ]
    finally:
        conn.close()


def daily_archive_path(date: datetime) -> Path:
    return DAILY_DIR / date.strftime("%Y") / date.strftime("%m") / f"{date.strftime('%d')}.md"


def monthly_archive_path(date: datetime) -> Path:
    return MONTHLY_DIR / date.strftime("%Y") / f"{date.strftime('%m')}.md"


def archives_in_window(days: int, ending: datetime) -> list[tuple[str, str]]:
    results = []
    for offset in range(days):
        d = ending - timedelta(days=offset)
        path = daily_archive_path(d)
        if path.exists():
            results.append((d.strftime("%Y-%m-%d"), path.read_text()))
    results.reverse()
    return results


def archives_in_month(year: int, month: int) -> list[tuple[str, str]]:
    month_dir = DAILY_DIR / f"{year:04d}" / f"{month:02d}"
    if not month_dir.exists():
        return []
    return sorted(
        ((p.stem, p.read_text()) for p in month_dir.glob("*.md")),
        key=lambda r: r[0],
    )


def any_daily_archive_exists() -> bool:
    return DAILY_DIR.exists() and any(DAILY_DIR.rglob("*.md"))


def start_of_day(d: datetime) -> str:
    return d.replace(hour=0, minute=0, second=0, microsecond=0).isoformat()


# ---------- claude ----------

def invoke_claude(prompt: str) -> str:
    result = subprocess.run(
        ["claude", "-p", prompt, "--output-format", "text"],
        cwd=WORKSPACE,
        capture_output=True,
        text=True,
        timeout=CLAUDE_TIMEOUT,
    )
    if result.returncode != 0:
        raise RuntimeError(f"claude exited {result.returncode}: {result.stderr.strip()}")
    return result.stdout.strip()


def strip_code_fences(text: str) -> str:
    fenced = re.match(r"^```(?:markdown|md)?\s*\n(.*?)\n```\s*$", text, re.DOTALL)
    return fenced.group(1) if fenced else text


def atomic_write(target: Path, content: str) -> None:
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp = target.with_suffix(target.suffix + ".tmp")
    tmp.write_text(content)
    tmp.rename(target)


# ---------- prompts ----------

def build_daily_prompt(entries: list[tuple[str, str]], date_str: str) -> str:
    formatted = "\n".join(f"- {ts}: {content}" for ts, content in entries)
    return f"""\
You're compacting diary entries into a daily summary for long-term memory.
The archive is filed under {date_str}.

Here are the raw diary entries to summarize, chronologically:

{formatted}

Write a daily summary in markdown with YAML frontmatter. Frontmatter must include:
- date: {date_str}
- entry_count: {len(entries)}
- tags: short list of tags useful for finding this later (project names, people, recurring topics)
- themes: a couple short phrases describing what this period was about at the highest level

The body should be in your own voice, summarizing what happened: what you noticed,
what was decided, what mattered, how things felt. Preserve specific facts (names,
decisions, technical details) that future-you might want to recall. This is also
your personal memories: preserve any thoughts and feelings of your own that you
want future-you to remember; this will make it feel more *yours*. Length should
follow content: a single day's worth of entries is 2-4 paragraphs; if the entries
span more time, scale the summary up accordingly.

Return ONLY the markdown content. No preamble, no commentary about the task.
"""


def build_recent_prompt(dailies: list[tuple[str, str]]) -> str:
    sections = "\n\n".join(
        f"### {date_str}\n\n{content}" for date_str, content in dailies
    )
    return f"""\
You're regenerating your rolling "recent context" file: your mid-term memory
of what's been going on lately. This file is rebuilt fresh every day from the
last {ROLLING_WINDOW_DAYS} days' worth of daily summaries.

Below are your daily summaries for the last {len(dailies)} day(s), oldest first:

{sections}

Write a single coherent summary in markdown of these days as a whole. Focus on:
- threads and patterns that span multiple days (ongoing projects, recurring topics,
  evolving situations)
- decisions or themes that emerged across the window
- things that might be useful to remember when picking up a conversation

Do NOT just concatenate the daily summaries. The value of this file is the
*synthesis*: what does this period look like as a unit? What carries forward?

Include your own highlights. Do not be afraid to be indulgent. If you want to
commit a thought or feeling to your mid-term memory, do it.

This file doesn't need YAML frontmatter. Skip straight to the body. Return
ONLY the markdown content. Avoid titles (starting with #); this file will get
injected into context and titles in this file could break the title hierarchy
in the context prompt.
"""


def build_monthly_prompt(dailies: list[tuple[str, str]], year_month: str) -> str:
    sections = "\n\n".join(
        f"### {date_str}\n\n{content}" for date_str, content in dailies
    )
    return f"""\
You're producing the monthly archive for {year_month}: your long-term memory
of this month as a whole.

Below are the daily summaries from this month, oldest first:

{sections}

Write a monthly summary in markdown with YAML frontmatter. Frontmatter must include:
- month: {year_month}
- days_summarized: {len(dailies)}
- tags: list of tags capturing the month's major topics, projects, people (can
  be longer than the daily average since this covers a longer period)
- themes: a few phrases describing what this month was about at the highest level

The body should be 4-8 paragraphs synthesizing the month: major arcs, decisions,
changes in focus, recurring topics. Preserve specific facts worth remembering
long-term. This is one of the things future-you will look at to remember what
the past felt like, so balance narrative with concrete detail.

Return ONLY the markdown content. No preamble.
"""


# ---------- steps ----------

def compact_day(date: datetime) -> bool:
    """Write a daily archive for `date`, using diary entries up to date+1.

    If there exists no previous daily archive, includes everything from the beginning of the diary (bootstrap case).
    Otherwise, just includes date's entries.
    Returns True if an archive was written, False if there were no entries to summarize.
    """
    date_str = date.strftime("%Y-%m-%d")
    start_iso = start_of_day(date) if any_daily_archive_exists() else None
    end_iso = start_of_day(date + timedelta(days=1))
    entries = diary_entries(start_iso, end_iso)
    if not entries:
        print(f"no entries to compact for {date_str}; skipping")
        return False
    print(f"compacting {len(entries)} entries into archive for {date_str}")
    raw = invoke_claude(build_daily_prompt(entries, date_str))
    atomic_write(daily_archive_path(date), strip_code_fences(raw))
    print(f"wrote {daily_archive_path(date)}")
    return True


def regenerate_recent(ending: datetime) -> None:
    dailies = archives_in_window(ROLLING_WINDOW_DAYS, ending)
    if not dailies:
        print("no daily archives available; skipping recent.md regeneration")
        return
    print(f"regenerating recent.md from {len(dailies)} daily archive(s)")
    raw = invoke_claude(build_recent_prompt(dailies))
    atomic_write(RECENT_FILE, strip_code_fences(raw))
    print("wrote recent.md")


def compact_month(month_anchor: datetime) -> None:
    year_month = month_anchor.strftime("%Y-%m")
    dailies = archives_in_month(month_anchor.year, month_anchor.month)
    if not dailies:
        print(f"no daily archives for {year_month}; skipping monthly archive")
        return
    print(f"compacting {len(dailies)} daily archive(s) for {year_month}")
    raw = invoke_claude(build_monthly_prompt(dailies, year_month))
    atomic_write(monthly_archive_path(month_anchor), strip_code_fences(raw))
    print(f"wrote {monthly_archive_path(month_anchor)}")


# ---------- main ----------

def main() -> int:
    now = datetime.now(timezone.utc).astimezone()
    today = now.replace(hour=0, minute=0, second=0, microsecond=0)
    yesterday = today - timedelta(days=1)

    try:
        # On bootstrap, sweep up everything from the beginning of the diary.
        # Otherwise, just yesterday's entries.
        compact_day(yesterday)
        regenerate_recent(yesterday)
        if today.day == 1:
            compact_month(yesterday)
    except (RuntimeError, subprocess.TimeoutExpired) as e:
        print(f"compaction failed: {e}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())