| 1 | #!/usr/bin/env python3 |
| 2 | """Context-budget compression engine for historical summaries. |
| 3 | |
| 4 | Assembles rolling, yearly, previous-week, and monthly context into a single |
| 5 | markdown string suitable for LLM prompt injection, respecting a total word |
| 6 | budget with per-section allocations. |
| 7 | |
| 8 | Pruning rules: |
| 9 | - Predictions unconfirmed after 8 weeks → drop |
| 10 | - Trends with 0 signal for 4 weeks → compress to one sentence |
| 11 | - Noise patterns → keep only top 3 |
| 12 | |
| 13 | CLI: |
| 14 | python scripts/context_budget.py --rolling path --yearly path [--prev-week path] [--max-words 1500] |
| 15 | """ |
| 16 | |
| 17 | from __future__ import annotations |
| 18 | |
| 19 | import argparse |
| 20 | import re |
| 21 | import sys |
| 22 | from datetime import datetime, timedelta, timezone |
| 23 | from pathlib import Path |
| 24 | from typing import Optional |
| 25 | |
| 26 | # Default budget allocations (words) |
| 27 | BUDGET_ROLLING = 500 |
| 28 | BUDGET_PREV_WEEK = 200 |
| 29 | BUDGET_YEARLY = 500 |
| 30 | BUDGET_MONTH = 300 |
| 31 | |
| 32 | STALE_PREDICTION_WEEKS = 8 |
| 33 | STALE_TREND_WEEKS = 4 |
| 34 | MAX_NOISE_PATTERNS = 3 |
| 35 | |
| 36 | |
| 37 | def word_count(text: str) -> int: |
| 38 | """Count words in text.""" |
| 39 | return len(text.split()) |
| 40 | |
| 41 | |
| 42 | def compress_to_budget(text: str, max_words: int) -> str: |
| 43 | """Truncate/summarize text to fit within a word budget. |
| 44 | |
| 45 | Preserves complete lines where possible, trimming from the end. |
| 46 | Returns the compressed text ending with '...' if truncated. |
| 47 | """ |
| 48 | if not text or not text.strip(): |
| 49 | return "" |
| 50 | if max_words <= 0: |
| 51 | return "" |
| 52 | |
| 53 | words = text.split() |
| 54 | if len(words) <= max_words: |
| 55 | return text.strip() |
| 56 | |
| 57 | # Preserve line structure: keep complete lines until budget |
| 58 | lines = text.strip().splitlines() |
| 59 | result_lines: list[str] = [] |
| 60 | total_words = 0 |
| 61 | for line in lines: |
| 62 | line_words = len(line.split()) |
| 63 | if total_words + line_words > max_words: |
| 64 | # Partial last line if we have room |
| 65 | remaining = max_words - total_words |
| 66 | if remaining > 0: |
| 67 | partial = " ".join(line.split()[:remaining]) |
| 68 | result_lines.append(partial + "...") |
| 69 | elif not result_lines: |
| 70 | # Edge case: first line exceeds budget |
| 71 | result_lines.append(" ".join(words[:max_words]) + "...") |
| 72 | break |
| 73 | result_lines.append(line) |
| 74 | total_words += line_words |
| 75 | |
| 76 | return "\n".join(result_lines) |
| 77 | |
| 78 | |
| 79 | def _parse_date_from_line(line: str) -> Optional[datetime]: |
| 80 | """Try to extract a date from a line like '- [2026-01-15] ...'.""" |
| 81 | match = re.search(r"\[(\d{4}-\d{2}-\d{2})\]", line) |
| 82 | if match: |
| 83 | try: |
| 84 | return datetime.strptime(match.group(1), "%Y-%m-%d").replace(tzinfo=timezone.utc) |
| 85 | except ValueError: |
| 86 | return None |
| 87 | return None |
| 88 | |
| 89 | |
| 90 | def prune_stale_predictions(text: str, now: Optional[datetime] = None) -> str: |
| 91 | """Drop predictions unconfirmed after 8 weeks.""" |
| 92 | if not text: |
| 93 | return "" |
| 94 | if now is None: |
| 95 | now = datetime.now(timezone.utc) |
| 96 | |
| 97 | cutoff = now - timedelta(weeks=STALE_PREDICTION_WEEKS) |
| 98 | lines = text.splitlines() |
| 99 | result: list[str] = [] |
| 100 | |
| 101 | for line in lines: |
| 102 | # Predictions are bullet lines with dates and no confirmation marker |
| 103 | if re.match(r"\s*[-*]\s*\[", line): |
| 104 | date = _parse_date_from_line(line) |
| 105 | if date and date < cutoff: |
| 106 | # Check for confirmation markers |
| 107 | lower = line.lower() |
| 108 | if "confirmed" not in lower and "✓" not in line and "✅" not in line: |
| 109 | continue # Drop stale unconfirmed prediction |
| 110 | result.append(line) |
| 111 | |
| 112 | return "\n".join(result) |
| 113 | |
| 114 | |
| 115 | def compress_stale_trends(text: str, now: Optional[datetime] = None) -> str: |
| 116 | """Compress trends with 0 signal for 4+ weeks to one sentence.""" |
| 117 | if not text: |
| 118 | return "" |
| 119 | if now is None: |
| 120 | now = datetime.now(timezone.utc) |
| 121 | |
| 122 | cutoff = now - timedelta(weeks=STALE_TREND_WEEKS) |
| 123 | lines = text.splitlines() |
| 124 | result: list[str] = [] |
| 125 | i = 0 |
| 126 | |
| 127 | while i < len(lines): |
| 128 | line = lines[i] |
| 129 | # Detect trend blocks: "### Trend: <name>" or "## Trend: <name>" |
| 130 | trend_match = re.match(r"(#{2,3})\s+[Tt]rend:\s*(.+)", line) |
| 131 | if trend_match: |
| 132 | trend_name = trend_match.group(2).strip() |
| 133 | # Collect the trend block |
| 134 | block_lines = [line] |
| 135 | i += 1 |
| 136 | has_recent_signal = False |
| 137 | while i < len(lines) and not re.match(r"#{2,3}\s+", lines[i]): |
| 138 | block_lines.append(lines[i]) |
| 139 | # Check for recent signals |
| 140 | date = _parse_date_from_line(lines[i]) |
| 141 | if date and date >= cutoff: |
| 142 | has_recent_signal = True |
| 143 | # Non-dated content counts as signal |
| 144 | if lines[i].strip() and not lines[i].startswith("#"): |
| 145 | if "no signal" in lines[i].lower() or "0 signal" in lines[i].lower(): |
| 146 | pass |
| 147 | elif date is None and lines[i].strip().startswith(("-", "*")): |
| 148 | has_recent_signal = True |
| 149 | i += 1 |
| 150 | |
| 151 | if has_recent_signal: |
| 152 | result.extend(block_lines) |
| 153 | else: |
| 154 | # Compress to one sentence |
| 155 | result.append(f"- {trend_name}: no recent signal (stale)") |
| 156 | else: |
| 157 | result.append(line) |
| 158 | i += 1 |
| 159 | |
| 160 | return "\n".join(result) |
| 161 | |
| 162 | |
| 163 | def keep_top_noise_patterns(text: str, max_patterns: int = MAX_NOISE_PATTERNS) -> str: |
| 164 | """Keep only the top N noise patterns from a noise section.""" |
| 165 | if not text: |
| 166 | return "" |
| 167 | |
| 168 | lines = text.splitlines() |
| 169 | result: list[str] = [] |
| 170 | in_noise_section = False |
| 171 | noise_count = 0 |
| 172 | |
| 173 | for line in lines: |
| 174 | # Detect noise section headers |
| 175 | if re.match(r"#{2,3}\s+[Nn]oise", line): |
| 176 | in_noise_section = True |
| 177 | noise_count = 0 |
| 178 | result.append(line) |
| 179 | continue |
| 180 | |
| 181 | if in_noise_section: |
| 182 | # New section starts |
| 183 | if re.match(r"#{2,3}\s+", line) and not re.match(r"#{2,3}\s+[Nn]oise", line): |
| 184 | in_noise_section = False |
| 185 | result.append(line) |
| 186 | continue |
| 187 | |
| 188 | # Count bullet items |
| 189 | if re.match(r"\s*[-*]\s+", line): |
| 190 | noise_count += 1 |
| 191 | if noise_count <= max_patterns: |
| 192 | result.append(line) |
| 193 | continue |
| 194 | |
| 195 | result.append(line) |
| 196 | else: |
| 197 | result.append(line) |
| 198 | |
| 199 | return "\n".join(result) |
| 200 | |
| 201 | |
| 202 | def _read_file_safe(path: Optional[Path]) -> str: |
| 203 | """Read a file, returning empty string if missing or unreadable.""" |
| 204 | if path is None: |
| 205 | return "" |
| 206 | try: |
| 207 | return Path(path).read_text(encoding="utf-8") |
| 208 | except (FileNotFoundError, PermissionError, OSError): |
| 209 | return "" |
| 210 | |
| 211 | |
| 212 | def assemble_historical_context( |
| 213 | rolling_path: Optional[str] = None, |
| 214 | yearly_path: Optional[str] = None, |
| 215 | prev_week_path: Optional[str] = None, |
| 216 | month_path: Optional[str] = None, |
| 217 | max_total_words: int = 1500, |
| 218 | now: Optional[datetime] = None, |
| 219 | ) -> str: |
| 220 | """Assemble context from multiple sources within budget. |
| 221 | |
| 222 | Budget allocation: |
| 223 | - rolling: 500 words |
| 224 | - prev_week: 200 words |
| 225 | - yearly: 500 words |
| 226 | - month: 300 words |
| 227 | |
| 228 | Total is capped at max_total_words (default 1500). |
| 229 | Returns a single markdown string for LLM prompt injection. |
| 230 | """ |
| 231 | # Scale budgets if max_total_words differs from default |
| 232 | default_total = BUDGET_ROLLING + BUDGET_PREV_WEEK + BUDGET_YEARLY + BUDGET_MONTH |
| 233 | scale = max_total_words / default_total if default_total > 0 else 1.0 |
| 234 | |
| 235 | budget_rolling = int(BUDGET_ROLLING * scale) |
| 236 | budget_prev_week = int(BUDGET_PREV_WEEK * scale) |
| 237 | budget_yearly = int(BUDGET_YEARLY * scale) |
| 238 | budget_month = int(BUDGET_MONTH * scale) |
| 239 | |
| 240 | # Read sources |
| 241 | rolling_raw = _read_file_safe(Path(rolling_path) if rolling_path else None) |
| 242 | yearly_raw = _read_file_safe(Path(yearly_path) if yearly_path else None) |
| 243 | prev_week_raw = _read_file_safe(Path(prev_week_path) if prev_week_path else None) |
| 244 | month_raw = _read_file_safe(Path(month_path) if month_path else None) |
| 245 | |
| 246 | # Apply pruning rules |
| 247 | rolling_pruned = keep_top_noise_patterns( |
| 248 | compress_stale_trends(prune_stale_predictions(rolling_raw, now=now), now=now) |
| 249 | ) |
| 250 | yearly_pruned = prune_stale_predictions(yearly_raw, now=now) |
| 251 | prev_week_pruned = prev_week_raw |
| 252 | month_pruned = compress_stale_trends(prune_stale_predictions(month_raw, now=now), now=now) |
| 253 | |
| 254 | # Compress each section to its budget |
| 255 | rolling_compressed = compress_to_budget(rolling_pruned, budget_rolling) |
| 256 | prev_week_compressed = compress_to_budget(prev_week_pruned, budget_prev_week) |
| 257 | yearly_compressed = compress_to_budget(yearly_pruned, budget_yearly) |
| 258 | month_compressed = compress_to_budget(month_pruned, budget_month) |
| 259 | |
| 260 | # Assemble final markdown |
| 261 | sections: list[str] = [] |
| 262 | |
| 263 | if rolling_compressed: |
| 264 | sections.append(f"## Rolling Context\n\n{rolling_compressed}") |
| 265 | if prev_week_compressed: |
| 266 | sections.append(f"## Previous Week\n\n{prev_week_compressed}") |
| 267 | if yearly_compressed: |
| 268 | sections.append(f"## Yearly Context\n\n{yearly_compressed}") |
| 269 | if month_compressed: |
| 270 | sections.append(f"## Monthly Context\n\n{month_compressed}") |
| 271 | |
| 272 | assembled = "\n\n".join(sections) |
| 273 | |
| 274 | # Final enforcement: ensure total stays within max_total_words |
| 275 | if word_count(assembled) > max_total_words: |
| 276 | assembled = compress_to_budget(assembled, max_total_words) |
| 277 | |
| 278 | return assembled |
| 279 | |
| 280 | |
| 281 | def main(argv: Optional[list[str]] = None) -> int: |
| 282 | parser = argparse.ArgumentParser( |
| 283 | description="Assemble historical context within a word budget." |
| 284 | ) |
| 285 | parser.add_argument("--rolling", type=str, help="Path to rolling context file") |
| 286 | parser.add_argument("--yearly", type=str, help="Path to yearly context file") |
| 287 | parser.add_argument("--prev-week", type=str, help="Path to previous week file") |
| 288 | parser.add_argument("--month", type=str, help="Path to monthly context file") |
| 289 | parser.add_argument( |
| 290 | "--max-words", type=int, default=1500, help="Maximum total words (default: 1500)" |
| 291 | ) |
| 292 | |
| 293 | args = parser.parse_args(argv) |
| 294 | |
| 295 | result = assemble_historical_context( |
| 296 | rolling_path=args.rolling, |
| 297 | yearly_path=args.yearly, |
| 298 | prev_week_path=args.prev_week, |
| 299 | month_path=args.month, |
| 300 | max_total_words=args.max_words, |
| 301 | ) |
| 302 | |
| 303 | if result: |
| 304 | print(result) |
| 305 | return 0 |
| 306 | |
| 307 | |
| 308 | if __name__ == "__main__": |
| 309 | sys.exit(main()) |