diff options
| -rw-r--r-- | agent/orchestrator/context.py | 23 | ||||
| -rw-r--r-- | scripts/compact.py | 275 |
2 files changed, 295 insertions, 3 deletions
diff --git a/agent/orchestrator/context.py b/agent/orchestrator/context.py index e9c1975..d19d8f3 100644 --- a/agent/orchestrator/context.py +++ b/agent/orchestrator/context.py @@ -11,6 +11,8 @@ from pathlib import Path SENIA_DB = Path("/etc/senia/data/senia.db") +RECENT_FILE = Path("/etc/senia/data/memory/recent.md") +ROLLING_WINDOW_DAYS = 11 def _today_diary_entries() -> list[str]: @@ -31,6 +33,13 @@ def _today_diary_entries() -> list[str]: conn.close() +def _recent_summary() -> str: + if not RECENT_FILE.exists(): + return "" + summary = RECENT_FILE.read_text().strip() + return f"## Recent context (last {ROLLING_WINDOW_DAYS} days)\n{summary}" + + def build_turn_context(user: str | None = None) -> str: diary_entries = _today_diary_entries() diary = ( @@ -38,10 +47,18 @@ def build_turn_context(user: str | None = None) -> str: if diary_entries else "(no entries yet today)" ) + recent = _recent_summary() now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") - time_section = f"## Current time\n{now}\n\n" - user_section = f"## User\n{user}\n\n" if user else "" + time_section = f"## Current time\n{now}" + user_section = f"## Speaking with\n{user}" if user else "" + recent_section = _recent_summary() diary_section = f"## Today's diary so far\n{diary}" + parts = [ + time_section, + user_section, + recent_section, + diary_section, + ] - return f"{time_section}{user_section}{diary_section}" + return "\n\n".join(p for p in parts if p) diff --git a/scripts/compact.py b/scripts/compact.py new file mode 100644 index 0000000..1b65363 --- /dev/null +++ b/scripts/compact.py @@ -0,0 +1,275 @@ +#!/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()) |