#!/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") archive_path = daily_archive_path(date) if archive_path.exists(): print(f"archive for {date_str} already exists; skipping") return False 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(archive_path, strip_code_fences(raw)) print(f"wrote {archive_path}") 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") archive_path = monthly_archive_path(month_anchor) if archive_path.exists(): print(f"archive for {year_month} already exists; skipping") return False 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(archive_path, strip_code_fences(raw)) print(f"wrote {archive_path}") # ---------- 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())