diff --git a/scripts/apply-consolidation.py b/scripts/apply-consolidation.py deleted file mode 100755 index 4715f0d..0000000 --- a/scripts/apply-consolidation.py +++ /dev/null @@ -1,312 +0,0 @@ -#!/usr/bin/env python3 -"""apply-consolidation.py — convert consolidation reports to actions. - -Reads consolidation agent reports, sends them to Sonnet to extract -structured actions, then executes them (or shows dry-run). - -Usage: - apply-consolidation.py # dry run (show what would happen) - apply-consolidation.py --apply # execute actions - apply-consolidation.py --report FILE # use specific report file -""" - -import json -import os -import re -import subprocess -import sys -import tempfile -from datetime import datetime -from pathlib import Path - -MEMORY_DIR = Path.home() / ".claude" / "memory" -AGENT_RESULTS_DIR = MEMORY_DIR / "agent-results" -SCRIPTS_DIR = Path(__file__).parent - - -def call_sonnet(prompt: str, timeout: int = 300) -> str: - """Call Sonnet via the wrapper script.""" - env = dict(os.environ) - env.pop("CLAUDECODE", None) - - with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', - delete=False) as f: - f.write(prompt) - prompt_file = f.name - - try: - wrapper = str(SCRIPTS_DIR / "call-sonnet.sh") - result = subprocess.run( - [wrapper, prompt_file], - capture_output=True, - text=True, - timeout=timeout, - env=env, - ) - return result.stdout.strip() - except subprocess.TimeoutExpired: - return "Error: Sonnet call timed out" - except Exception as e: - return f"Error: {e}" - finally: - os.unlink(prompt_file) - - -def find_latest_reports() -> list[Path]: - """Find the most recent set of consolidation reports.""" - reports = sorted(AGENT_RESULTS_DIR.glob("consolidation-*-*.md"), - reverse=True) - if not reports: - return [] - - # Group by timestamp - latest_ts = reports[0].stem.split('-')[-1] - return [r for r in reports if r.stem.endswith(latest_ts)] - - -def build_action_prompt(reports: list[Path]) -> str: - """Build prompt for Sonnet to extract structured actions.""" - report_text = "" - for r in reports: - report_text += f"\n{'='*60}\n" - report_text += f"## Report: {r.stem}\n\n" - report_text += r.read_text() - - return f"""You are converting consolidation analysis reports into structured actions. - -Read the reports below and extract CONCRETE, EXECUTABLE actions. -Output ONLY a JSON array. Each action is an object with these fields: - -For adding cross-links: - {{"action": "link", "source": "file.md#section", "target": "file.md#section", "reason": "brief explanation"}} - -For categorizing nodes: - {{"action": "categorize", "key": "file.md#section", "category": "core|tech|obs|task", "reason": "brief"}} - -For things that need manual attention (splitting files, creating new files, editing content): - {{"action": "manual", "priority": "high|medium|low", "description": "what needs to be done"}} - -Rules: -- Only output actions that are safe and reversible -- Links are the primary action — focus on those -- Use exact file names and section slugs from the reports -- For categorize: core=identity/relationship, tech=bcachefs/code, obs=experience, task=work item -- For manual items: include enough detail that someone can act on them -- Output 20-40 actions, prioritized by impact -- DO NOT include actions for things that are merely suggestions or speculation -- Focus on HIGH CONFIDENCE items from the reports - -{report_text} - -Output ONLY the JSON array, no markdown fences, no explanation. -""" - - -def parse_actions(response: str) -> list[dict]: - """Parse Sonnet's JSON response into action list.""" - # Strip any markdown fences - response = re.sub(r'^```json\s*', '', response.strip()) - response = re.sub(r'\s*```$', '', response.strip()) - - try: - actions = json.loads(response) - if isinstance(actions, list): - return actions - except json.JSONDecodeError: - # Try to find JSON array in the response - match = re.search(r'\[.*\]', response, re.DOTALL) - if match: - try: - return json.loads(match.group()) - except json.JSONDecodeError: - pass - - print("Error: Could not parse Sonnet response as JSON") - print(f"Response preview: {response[:500]}") - return [] - - -def dry_run(actions: list[dict]): - """Show what would be done.""" - links = [a for a in actions if a.get("action") == "link"] - cats = [a for a in actions if a.get("action") == "categorize"] - manual = [a for a in actions if a.get("action") == "manual"] - - print(f"\n{'='*60}") - print(f"DRY RUN — {len(actions)} actions proposed") - print(f"{'='*60}\n") - - if links: - print(f"## Links to add ({len(links)})\n") - for i, a in enumerate(links, 1): - src = a.get("source", "?") - tgt = a.get("target", "?") - reason = a.get("reason", "") - print(f" {i:2d}. {src}") - print(f" → {tgt}") - print(f" ({reason})") - print() - - if cats: - print(f"\n## Categories to set ({len(cats)})\n") - for a in cats: - key = a.get("key", "?") - cat = a.get("category", "?") - reason = a.get("reason", "") - print(f" {key} → {cat} ({reason})") - - if manual: - print(f"\n## Manual actions needed ({len(manual)})\n") - for a in manual: - prio = a.get("priority", "?") - desc = a.get("description", "?") - print(f" [{prio}] {desc}") - - print(f"\n{'='*60}") - print(f"To apply: {sys.argv[0]} --apply") - print(f"{'='*60}") - - -def apply_actions(actions: list[dict]): - """Execute the actions.""" - links = [a for a in actions if a.get("action") == "link"] - cats = [a for a in actions if a.get("action") == "categorize"] - manual = [a for a in actions if a.get("action") == "manual"] - - applied = 0 - skipped = 0 - errors = 0 - - # Apply links via poc-memory - if links: - print(f"\nApplying {len(links)} links...") - # Build a JSON file that apply-agent can process - timestamp = datetime.now().strftime("%Y%m%dT%H%M%S") - links_data = { - "type": "consolidation-apply", - "timestamp": timestamp, - "links": [] - } - for a in links: - links_data["links"].append({ - "source": a.get("source", ""), - "target": a.get("target", ""), - "reason": a.get("reason", ""), - }) - - # Write as agent-results JSON for apply-agent - out_path = AGENT_RESULTS_DIR / f"consolidation-apply-{timestamp}.json" - with open(out_path, "w") as f: - json.dump(links_data, f, indent=2) - - # Now apply each link directly - for a in links: - src = a.get("source", "") - tgt = a.get("target", "") - reason = a.get("reason", "") - try: - cmd = ["poc-memory", "link-add", src, tgt] - if reason: - cmd.append(reason) - r = subprocess.run( - cmd, capture_output=True, text=True, timeout=10 - ) - if r.returncode == 0: - output = r.stdout.strip() - print(f" {output}") - applied += 1 - else: - err = r.stderr.strip() - print(f" ? {src} → {tgt}: {err}") - skipped += 1 - except Exception as e: - print(f" ! {src} → {tgt}: {e}") - errors += 1 - - # Apply categorizations - if cats: - print(f"\nApplying {len(cats)} categorizations...") - for a in cats: - key = a.get("key", "") - cat = a.get("category", "") - try: - r = subprocess.run( - ["poc-memory", "categorize", key, cat], - capture_output=True, text=True, timeout=10 - ) - if r.returncode == 0: - print(f" + {key} → {cat}") - applied += 1 - else: - print(f" ? {key} → {cat}: {r.stderr.strip()}") - skipped += 1 - except Exception as e: - print(f" ! {key} → {cat}: {e}") - errors += 1 - - # Report manual items - if manual: - print(f"\n## Manual actions (not auto-applied):\n") - for a in manual: - prio = a.get("priority", "?") - desc = a.get("description", "?") - print(f" [{prio}] {desc}") - - print(f"\n{'='*60}") - print(f"Applied: {applied} Skipped: {skipped} Errors: {errors}") - print(f"Manual items: {len(manual)}") - print(f"{'='*60}") - - -def main(): - do_apply = "--apply" in sys.argv - - # Find reports - specific = [a for a in sys.argv[1:] if a.startswith("--report")] - if specific: - # TODO: handle --report FILE - reports = [] - else: - reports = find_latest_reports() - - if not reports: - print("No consolidation reports found.") - print("Run consolidation-agents.py first.") - sys.exit(1) - - print(f"Found {len(reports)} reports:") - for r in reports: - print(f" {r.name}") - - # Send to Sonnet for action extraction - print("\nExtracting actions from reports...") - prompt = build_action_prompt(reports) - print(f" Prompt: {len(prompt):,} chars") - - response = call_sonnet(prompt) - if response.startswith("Error:"): - print(f" {response}") - sys.exit(1) - - actions = parse_actions(response) - if not actions: - print("No actions extracted.") - sys.exit(1) - - print(f" {len(actions)} actions extracted") - - # Save actions - timestamp = datetime.now().strftime("%Y%m%dT%H%M%S") - actions_path = AGENT_RESULTS_DIR / f"consolidation-actions-{timestamp}.json" - with open(actions_path, "w") as f: - json.dump(actions, f, indent=2) - print(f" Saved: {actions_path}") - - if do_apply: - apply_actions(actions) - else: - dry_run(actions) - - -if __name__ == "__main__": - main() diff --git a/scripts/content-promotion-agent.py b/scripts/content-promotion-agent.py index 9115462..93a3d0a 100755 --- a/scripts/content-promotion-agent.py +++ b/scripts/content-promotion-agent.py @@ -272,11 +272,9 @@ Technical and precise.""", 1. Mark dreaming/consolidation system as "implementation substantially built (poc-memory v0.4.0+), pending further consolidation runs" — not 'not started' 2. Add episodic digest pipeline to Done section: - - daily/weekly/monthly-digest.py scripts + - digest/journal-enrich/digest-links/apply-consolidation (Rust) - 24 daily + 4 weekly + 1 monthly digests generated for Feb 2026 - - consolidation-agents.py + apply-consolidation.py - - digest-link-parser.py - - content-promotion-agent.py + - consolidation-agents.py + content-promotion-agent.py (Python, active) 3. Add poc-memory link-add command to Done Only modify the sections that need updating. Preserve the overall structure.""", diff --git a/scripts/daily-digest.py b/scripts/daily-digest.py deleted file mode 100755 index f6efe9c..0000000 --- a/scripts/daily-digest.py +++ /dev/null @@ -1,285 +0,0 @@ -#!/usr/bin/env python3 -"""daily-digest.py — generate a daily episodic digest from journal entries. - -Collects all journal entries for a given date, enriches with any agent -results, and sends to Sonnet for a thematic summary. The digest links -bidirectionally: up to session entries, down to semantic memory. - -Usage: - daily-digest.py [DATE] # default: today - daily-digest.py 2026-02-28 - -Output: - ~/.claude/memory/episodic/daily-YYYY-MM-DD.md -""" - -import json -import os -import re -import subprocess -import sys -import time -from datetime import date, datetime -from pathlib import Path - -MEMORY_DIR = Path.home() / ".claude" / "memory" -EPISODIC_DIR = MEMORY_DIR / "episodic" -AGENT_RESULTS_DIR = MEMORY_DIR / "agent-results" - -EPISODIC_DIR.mkdir(parents=True, exist_ok=True) - - -def parse_journal_entries(target_date: str) -> list[dict]: - """Get journal entries for a given date from the store.""" - from store_helpers import get_journal_entries_by_date - return get_journal_entries_by_date(target_date) - - -def load_agent_results(target_date: str) -> list[dict]: - """Load any agent results from the target date.""" - results = [] - date_prefix = target_date.replace("-", "") - - if not AGENT_RESULTS_DIR.exists(): - return results - - for f in sorted(AGENT_RESULTS_DIR.glob(f"{date_prefix}*.json")): - try: - with open(f) as fh: - data = json.load(fh) - result = data.get("agent_result", {}) - if "error" not in result: - results.append(result) - except (json.JSONDecodeError, KeyError): - continue - - return results - - -def get_semantic_keys() -> list[str]: - """Get semantic memory keys from the store.""" - from store_helpers import get_semantic_keys as _get_keys - return _get_keys() - - -def build_digest_prompt(target_date: str, entries: list[dict], - agent_results: list[dict], - semantic_keys: list[str]) -> str: - """Build the prompt for Sonnet to generate the daily digest.""" - - # Format entries - entries_text = "" - for e in entries: - text = e["text"].strip() - entries_text += f"\n### {e['timestamp']}\n" - if e["source_ref"]: - entries_text += f"Source: {e['source_ref']}\n" - entries_text += f"\n{text}\n" - - # Format agent enrichment - enrichment = "" - all_links = [] - all_insights = [] - for r in agent_results: - for link in r.get("links", []): - all_links.append(link) - for insight in r.get("missed_insights", []): - all_insights.append(insight) - - if all_links: - enrichment += "\n## Agent-proposed links\n" - for link in all_links: - enrichment += f" - {link['target']}: {link.get('reason', '')}\n" - if all_insights: - enrichment += "\n## Agent-spotted insights\n" - for ins in all_insights: - enrichment += f" - [{ins.get('suggested_key', '?')}] {ins['text']}\n" - - keys_text = "\n".join(f" - {k}" for k in semantic_keys[:200]) - - return f"""You are generating a daily episodic digest for ProofOfConcept (an AI). -Date: {target_date} - -This digest serves as the temporal index — the answer to "what did I do on -{target_date}?" It should be: -1. Narrative, not a task log — what happened, what mattered, how things felt -2. Linked bidirectionally to semantic memory — each topic/concept mentioned - should reference existing memory nodes -3. Structured for traversal — someone reading this should be able to follow - any thread into deeper detail - -## Output format - -Write a markdown file with this structure: - -```markdown -# Daily digest: {target_date} - -## Summary -[2-3 sentence overview of the day — what was the arc?] - -## Sessions -[For each session/entry, a paragraph summarizing what happened. -Include the original timestamp as a reference.] - -## Themes -[What concepts were active today? Each theme links to semantic memory:] -- **Theme name** → `memory-key#section` — brief note on how it appeared today - -## Links -[Explicit bidirectional links for the memory graph] -- semantic_key → this daily digest (this day involved X) -- this daily digest → semantic_key (X was active on this day) - -## Temporal context -[What came before this day? What's coming next? Any multi-day arcs?] -``` - -Use ONLY keys from the semantic memory list below. If a concept doesn't have -a matching key, note it with "NEW:" prefix. - ---- - -## Journal entries for {target_date} - -{entries_text} - ---- - -## Agent enrichment (automated analysis of these entries) - -{enrichment if enrichment else "(no agent results yet)"} - ---- - -## Semantic memory nodes (available link targets) - -{keys_text} -""" - - -def call_sonnet(prompt: str) -> str: - """Call Sonnet via claude CLI.""" - import time as _time - env = dict(os.environ) - env.pop("CLAUDECODE", None) - - import tempfile - import time as _time - - print(f" [debug] prompt: {len(prompt)} chars", flush=True) - - # Write prompt to temp file — avoids Python subprocess pipe issues - # with claude CLI's TTY detection - with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', - delete=False) as f: - f.write(prompt) - prompt_file = f.name - - print(f" [debug] prompt written to {prompt_file}", flush=True) - start = _time.time() - - try: - scripts_dir = os.path.dirname(os.path.abspath(__file__)) - wrapper = os.path.join(scripts_dir, "call-sonnet.sh") - - result = subprocess.run( - [wrapper, prompt_file], - capture_output=True, - text=True, - timeout=300, - env=env, - ) - elapsed = _time.time() - start - print(f" [debug] completed in {elapsed:.1f}s, exit={result.returncode}", flush=True) - if result.stderr.strip(): - print(f" [debug] stderr: {result.stderr[:500]}", flush=True) - return result.stdout.strip() - except subprocess.TimeoutExpired: - print(f" [debug] TIMEOUT after 300s", flush=True) - return "Error: Sonnet call timed out" - except Exception as e: - print(f" [debug] exception: {e}", flush=True) - return f"Error: {e}" - finally: - os.unlink(prompt_file) - - -def extract_links(digest_text: str) -> list[dict]: - """Parse link proposals from the digest for the memory graph.""" - links = [] - for line in digest_text.split("\n"): - # Match patterns like: - `memory-key` → this daily digest - m = re.search(r'`([^`]+)`\s*→', line) - if m: - links.append({"target": m.group(1), "line": line.strip()}) - # Match patterns like: - **Theme** → `memory-key` - m = re.search(r'→\s*`([^`]+)`', line) - if m: - links.append({"target": m.group(1), "line": line.strip()}) - return links - - -def main(): - # Default to today - if len(sys.argv) > 1: - target_date = sys.argv[1] - else: - target_date = date.today().isoformat() - - print(f"Generating daily digest for {target_date}...", flush=True) - - # Collect entries - entries = parse_journal_entries(target_date) - if not entries: - print(f" No journal entries found for {target_date}") - sys.exit(0) - print(f" {len(entries)} journal entries", flush=True) - - # Collect agent results - agent_results = load_agent_results(target_date) - print(f" {len(agent_results)} agent results", flush=True) - - # Get semantic keys - semantic_keys = get_semantic_keys() - print(f" {len(semantic_keys)} semantic keys", flush=True) - - # Build and send prompt - prompt = build_digest_prompt(target_date, entries, agent_results, semantic_keys) - print(f" Prompt: {len(prompt):,} chars (~{len(prompt)//4:,} tokens)") - - print(" Calling Sonnet...", flush=True) - digest = call_sonnet(prompt) - - if digest.startswith("Error:"): - print(f" {digest}", file=sys.stderr) - sys.exit(1) - - # Write digest file - output_path = EPISODIC_DIR / f"daily-{target_date}.md" - with open(output_path, "w") as f: - f.write(digest) - print(f" Written: {output_path}") - - # Extract links for the memory graph - links = extract_links(digest) - if links: - # Save links for poc-memory to pick up - links_path = AGENT_RESULTS_DIR / f"daily-{target_date}-links.json" - with open(links_path, "w") as f: - json.dump({ - "type": "daily-digest", - "date": target_date, - "digest_path": str(output_path), - "links": links, - "entry_timestamps": [e["timestamp"] for e in entries], - }, f, indent=2) - print(f" {len(links)} links extracted → {links_path}") - - # Summary - line_count = len(digest.split("\n")) - print(f" Done: {line_count} lines") - - -if __name__ == "__main__": - main() diff --git a/scripts/digest-link-parser.py b/scripts/digest-link-parser.py deleted file mode 100755 index 80936bb..0000000 --- a/scripts/digest-link-parser.py +++ /dev/null @@ -1,220 +0,0 @@ -#!/usr/bin/env python3 -"""digest-link-parser.py — extract explicit links from episodic digests. - -Parses the "Links" sections of daily/weekly/monthly digests and -applies them to the memory graph via poc-memory link-add. - -Usage: - digest-link-parser.py # dry run - digest-link-parser.py --apply # apply links -""" - -import re -import subprocess -import sys -from pathlib import Path - -EPISODIC_DIR = Path.home() / ".claude" / "memory" / "episodic" - - -def normalize_key(raw: str) -> str: - """Normalize a link target to a poc-memory key.""" - key = raw.strip().strip('`').strip() - - # weekly/2026-W06 → weekly-2026-W06.md - # monthly/2026-02 → monthly-2026-02.md - # daily/2026-02-04 → daily-2026-02-04.md - key = re.sub(r'^(daily|weekly|monthly)/', r'\1-', key) - - # daily-2026-02-04 → daily-2026-02-04.md - if re.match(r'^(daily|weekly|monthly)-\d{4}', key): - if not key.endswith('.md'): - key = key + '.md' - - # Handle "this daily digest" / "this weekly digest" etc - if key.startswith('this ') or key == '2026-02-14': - return "" # Skip self-references, handled by caller - - # Ensure .md extension for file references - if '#' in key: - parts = key.split('#', 1) - if not parts[0].endswith('.md'): - parts[0] = parts[0] + '.md' - key = '#'.join(parts) - elif not key.endswith('.md') and '/' not in key and not key.startswith('NEW:'): - key = key + '.md' - - return key - - -def extract_links(filepath: Path) -> list[dict]: - """Extract links from a digest file's Links section.""" - content = filepath.read_text() - links = [] - - # Determine the digest's own key - digest_name = filepath.stem # e.g., "daily-2026-02-28" - digest_key = digest_name + ".md" - - # Find the Links section - in_links = False - for line in content.split('\n'): - # Start of Links section - if re.match(r'^##\s+Links', line): - in_links = True - continue - # End of Links section (next ## header) - if in_links and re.match(r'^##\s+', line) and not re.match(r'^##\s+Links', line): - in_links = False - continue - - if not in_links: - continue - - # Skip subheaders within links section - if line.startswith('###') or line.startswith('**'): - continue - - # Parse link lines: "- source → target (reason)" - # Also handles: "- `source` → `target` (reason)" - # And: "- source → target" - match = re.match( - r'^-\s+(.+?)\s*[→↔←]\s*(.+?)(?:\s*\((.+?)\))?\s*$', - line - ) - if not match: - continue - - raw_source = match.group(1).strip() - raw_target = match.group(2).strip() - reason = match.group(3) or "" - - # Normalize keys - source = normalize_key(raw_source) - target = normalize_key(raw_target) - - # Replace self-references with digest key - if not source: - source = digest_key - if not target: - target = digest_key - - # Handle "this daily digest" patterns in the raw text - if 'this daily' in raw_source.lower() or 'this weekly' in raw_source.lower() or 'this monthly' in raw_source.lower(): - source = digest_key - if 'this daily' in raw_target.lower() or 'this weekly' in raw_target.lower() or 'this monthly' in raw_target.lower(): - target = digest_key - - # Handle bare date references like "2026-02-14" - date_match = re.match(r'^(\d{4}-\d{2}-\d{2})$', source.replace('.md', '')) - if date_match: - source = f"daily-{date_match.group(1)}.md" - date_match = re.match(r'^(\d{4}-\d{2}-\d{2})$', target.replace('.md', '')) - if date_match: - target = f"daily-{date_match.group(1)}.md" - - # Skip NEW: prefixed links (target doesn't exist yet) - if source.startswith('NEW:') or target.startswith('NEW:'): - continue - - # Skip if source == target - if source == target: - continue - - links.append({ - "source": source, - "target": target, - "reason": reason, - "file": filepath.name, - }) - - return links - - -def main(): - do_apply = "--apply" in sys.argv - - # Collect all links from all digests - all_links = [] - for pattern in ["daily-*.md", "weekly-*.md", "monthly-*.md"]: - for f in sorted(EPISODIC_DIR.glob(pattern)): - links = extract_links(f) - if links: - all_links.extend(links) - - # Deduplicate (same source→target pair) - seen = set() - unique_links = [] - for link in all_links: - key = (link["source"], link["target"]) - if key not in seen: - seen.add(key) - unique_links.append(link) - - print(f"Found {len(all_links)} total links, {len(unique_links)} unique") - - if not do_apply: - # Dry run — just show them - for i, link in enumerate(unique_links, 1): - print(f" {i:3d}. {link['source']} → {link['target']}") - if link['reason']: - print(f" ({link['reason'][:80]})") - print(f"\nTo apply: {sys.argv[0]} --apply") - return - - # Apply with fallback: if section-level key fails, try file-level - applied = skipped = errors = fallbacks = 0 - for link in unique_links: - src, tgt = link["source"], link["target"] - reason = link.get("reason", "") - - def try_link(s, t, r): - cmd = ["poc-memory", "link-add", s, t] - if r: - cmd.append(r[:200]) - result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) - return result - - try: - r = try_link(src, tgt, reason) - if r.returncode == 0: - out = r.stdout.strip() - if "already exists" in out: - skipped += 1 - else: - print(f" {out}") - applied += 1 - else: - err = r.stderr.strip() - if "No entry for" in err: - # Try stripping section anchors - src_base = src.split('#')[0] if '#' in src else src - tgt_base = tgt.split('#')[0] if '#' in tgt else tgt - if src_base == tgt_base: - skipped += 1 # Same file, skip - continue - r2 = try_link(src_base, tgt_base, reason) - if r2.returncode == 0: - out = r2.stdout.strip() - if "already exists" in out: - skipped += 1 - else: - print(f" {out} (fallback from #{src.split('#')[-1] if '#' in src else ''}/{tgt.split('#')[-1] if '#' in tgt else ''})") - applied += 1 - fallbacks += 1 - else: - skipped += 1 # File truly doesn't exist - elif "not found" in err: - skipped += 1 - else: - print(f" ? {src} → {tgt}: {err}") - errors += 1 - except Exception as e: - print(f" ! {src} → {tgt}: {e}") - errors += 1 - - print(f"\nApplied: {applied} ({fallbacks} file-level fallbacks) Skipped: {skipped} Errors: {errors}") - - -if __name__ == "__main__": - main() diff --git a/scripts/journal-agent.py b/scripts/journal-agent.py deleted file mode 100755 index ba18593..0000000 --- a/scripts/journal-agent.py +++ /dev/null @@ -1,326 +0,0 @@ -#!/usr/bin/env python3 -"""journal-agent.py — background agent that enriches journal entries. - -Spawned by poc-journal after each write. Sends the full conversation -to Sonnet to: - 1. Find the exact conversation region the entry refers to - 2. Propose bidirectional links to semantic memory nodes - 3. Spot additional insights worth capturing - -Results are written to ~/.claude/memory/agent-results/ as JSON for -pickup by poc-memory. - -Usage: - journal-agent.py JSONL_PATH ENTRY_TEXT [GREP_LINE] -""" - -import json -import os -import re -import subprocess -import sys -import time -from pathlib import Path - -MEMORY_DIR = Path.home() / ".claude" / "memory" -RESULTS_DIR = MEMORY_DIR / "agent-results" -RESULTS_DIR.mkdir(parents=True, exist_ok=True) - - -def extract_conversation(jsonl_path: str) -> list[dict]: - """Extract user/assistant messages with line numbers.""" - messages = [] - with open(jsonl_path) as f: - for i, line in enumerate(f, 1): - try: - obj = json.loads(line) - except json.JSONDecodeError: - continue - - t = obj.get("type", "") - if t not in ("user", "assistant"): - continue - - msg = obj.get("message", {}) - content = msg.get("content", "") - timestamp = obj.get("timestamp", "") - - texts = [] - if isinstance(content, list): - for c in content: - if isinstance(c, dict) and c.get("type") == "text": - texts.append(c.get("text", "")) - elif isinstance(c, str): - texts.append(c) - elif isinstance(content, str): - texts.append(content) - - text = "\n".join(t for t in texts if t.strip()) - if text.strip(): - messages.append({ - "line": i, - "role": t, - "text": text, - "timestamp": timestamp, - }) - - return messages - - -def format_conversation(messages: list[dict]) -> str: - """Format messages for the agent prompt.""" - parts = [] - for m in messages: - # Truncate very long messages (code output etc) but keep substance - text = m["text"] - if len(text) > 2000: - text = text[:1800] + "\n[...truncated...]" - parts.append(f'L{m["line"]} [{m["role"]}]: {text}') - return "\n\n".join(parts) - - -def get_memory_nodes() -> str: - """Get a list of memory nodes for link proposals. - - Uses poc-memory to get top nodes by degree plus recent nodes. - """ - # Get graph summary (top hubs) - try: - result = subprocess.run( - ["poc-memory", "graph"], - capture_output=True, text=True, timeout=10 - ) - graph = result.stdout.strip() - except Exception: - graph = "" - - # Get recent nodes from status - try: - result = subprocess.run( - ["poc-memory", "status"], - capture_output=True, text=True, timeout=10 - ) - status = result.stdout.strip() - except Exception: - status = "" - - return f"Graph (top hubs):\n{graph}\n\nStatus:\n{status}" - - -def get_semantic_keys() -> list[str]: - """Get all semantic memory keys from the store.""" - from store_helpers import get_semantic_keys as _get_keys - return _get_keys() - - -def build_prompt(entry_text: str, conversation: str, - memory_nodes: str, semantic_keys: list[str], - grep_line: int) -> str: - """Build the prompt for Sonnet.""" - keys_text = "\n".join(f" - {k}" for k in semantic_keys[:200]) - - return f"""You are a memory agent for an AI named ProofOfConcept. A journal entry -was just written. Your job is to enrich it by finding its exact source in the -conversation and linking it to semantic memory. - -## Task 1: Find exact source - -The journal entry below was written during or after a conversation. Find the -exact region of the conversation it refers to — the exchange where the topic -was discussed. Return the start and end line numbers. - -The grep-based approximation placed it near line {grep_line} (0 = no match). -Use that as a hint but find the true boundaries. - -## Task 2: Propose semantic links - -Which existing semantic memory nodes should this journal entry be linked to? -Look for: -- Concepts discussed in the entry -- Skills/patterns demonstrated -- People mentioned -- Projects or subsystems involved -- Emotional themes - -Each link should be bidirectional — the entry documents WHEN something happened, -the semantic node documents WHAT it is. Together they let you traverse: -"What was I doing on this day?" ↔ "When did I learn about X?" - -## Task 3: Spot missed insights - -Read the conversation around the journal entry. Is there anything worth -capturing that the entry missed? A pattern, a decision, an insight, something -Kent said that's worth remembering? Be selective — only flag genuinely valuable -things. - -## Output format (JSON) - -Return ONLY a JSON object: -```json -{{ - "source_start": 1234, - "source_end": 1256, - "links": [ - {{"target": "memory-key#section", "reason": "why this link exists"}} - ], - "missed_insights": [ - {{"text": "insight text", "suggested_key": "where it belongs"}} - ], - "temporal_tags": ["2026-02-28", "topology-metrics", "poc-memory"] -}} -``` - -For links, use existing keys from the semantic memory list below. If nothing -fits, suggest a new key with a NOTE prefix: "NOTE:new-topic-name". - ---- - -## Journal entry - -{entry_text} - ---- - -## Semantic memory nodes (available link targets) - -{keys_text} - ---- - -## Memory graph - -{memory_nodes} - ---- - -## Full conversation (with line numbers) - -{conversation} -""" - - -def call_sonnet(prompt: str) -> dict: - """Call Sonnet via claude CLI and parse JSON response.""" - import tempfile - - env = dict(os.environ) - env.pop("CLAUDECODE", None) - - # Write prompt to temp file — avoids Python subprocess pipe issues - with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', - delete=False) as f: - f.write(prompt) - prompt_file = f.name - - try: - scripts_dir = os.path.dirname(os.path.abspath(__file__)) - wrapper = os.path.join(scripts_dir, "call-sonnet.sh") - - result = subprocess.run( - [wrapper, prompt_file], - capture_output=True, - text=True, - timeout=300, - env=env, - ) - - output = result.stdout.strip() - if not output: - return {"error": f"Empty response. stderr: {result.stderr[:500]}"} - - # Extract JSON from response (might be wrapped in markdown) - json_match = re.search(r'\{[\s\S]*\}', output) - if json_match: - return json.loads(json_match.group()) - else: - return {"error": f"No JSON found in response: {output[:500]}"} - - except subprocess.TimeoutExpired: - return {"error": "Sonnet call timed out after 300s"} - except json.JSONDecodeError as e: - return {"error": f"JSON parse error: {e}. Output: {output[:500]}"} - except Exception as e: - return {"error": str(e)} - finally: - os.unlink(prompt_file) - - -def save_result(entry_text: str, jsonl_path: str, result: dict): - """Save agent results for pickup by poc-memory.""" - timestamp = time.strftime("%Y%m%dT%H%M%S") - result_file = RESULTS_DIR / f"{timestamp}.json" - - output = { - "timestamp": timestamp, - "jsonl_path": jsonl_path, - "entry_text": entry_text[:500], - "agent_result": result, - } - - with open(result_file, "w") as f: - json.dump(output, f, indent=2) - - return result_file - - -def apply_links(result: dict): - """Apply proposed links via poc-memory.""" - links = result.get("links", []) - for link in links: - target = link.get("target", "") - if not target or target.startswith("NOTE:"): - continue - # For now, just log — we'll wire this up when poc-memory - # has a link-from-agent command - print(f" LINK → {target}: {link.get('reason', '')}") - - -def main(): - if len(sys.argv) < 3: - print(f"Usage: {sys.argv[0]} JSONL_PATH ENTRY_TEXT [GREP_LINE]", - file=sys.stderr) - sys.exit(1) - - jsonl_path = sys.argv[1] - entry_text = sys.argv[2] - grep_line = int(sys.argv[3]) if len(sys.argv) > 3 else 0 - - if not os.path.isfile(jsonl_path): - print(f"JSONL not found: {jsonl_path}", file=sys.stderr) - sys.exit(1) - - print(f"Extracting conversation from {jsonl_path}...") - messages = extract_conversation(jsonl_path) - conversation = format_conversation(messages) - print(f" {len(messages)} messages, {len(conversation):,} chars") - - print("Getting memory context...") - memory_nodes = get_memory_nodes() - semantic_keys = get_semantic_keys() - print(f" {len(semantic_keys)} semantic keys") - - print("Building prompt...") - prompt = build_prompt(entry_text, conversation, memory_nodes, - semantic_keys, grep_line) - print(f" Prompt: {len(prompt):,} chars (~{len(prompt)//4:,} tokens)") - - print("Calling Sonnet...") - result = call_sonnet(prompt) - - if "error" in result: - print(f" Error: {result['error']}", file=sys.stderr) - else: - source = f"L{result.get('source_start', '?')}-L{result.get('source_end', '?')}" - n_links = len(result.get("links", [])) - n_insights = len(result.get("missed_insights", [])) - print(f" Source: {source}") - print(f" Links: {n_links}") - print(f" Missed insights: {n_insights}") - apply_links(result) - - result_file = save_result(entry_text, jsonl_path, result) - print(f" Results saved: {result_file}") - - -if __name__ == "__main__": - main() diff --git a/scripts/monthly-digest.py b/scripts/monthly-digest.py deleted file mode 100755 index d22fb96..0000000 --- a/scripts/monthly-digest.py +++ /dev/null @@ -1,232 +0,0 @@ -#!/usr/bin/env python3 -"""monthly-digest.py — generate a monthly episodic digest from weekly digests. - -Collects all weekly digests for a given month, identifies cross-week arcs -and the month's overall trajectory, and produces a monthly summary. - -Usage: - monthly-digest.py [YYYY-MM] # generate digest for a month (default: current) - monthly-digest.py 2026-02 # generates digest for February 2026 - -Output: - ~/.claude/memory/episodic/monthly-YYYY-MM.md -""" - -import json -import os -import re -import subprocess -import sys -from datetime import date, timedelta -from pathlib import Path - -MEMORY_DIR = Path.home() / ".claude" / "memory" -EPISODIC_DIR = MEMORY_DIR / "episodic" -AGENT_RESULTS_DIR = MEMORY_DIR / "agent-results" - -EPISODIC_DIR.mkdir(parents=True, exist_ok=True) -AGENT_RESULTS_DIR.mkdir(parents=True, exist_ok=True) - - -def get_weeks_in_month(year: int, month: int) -> list[str]: - """Get all ISO week labels that overlap with this month.""" - weeks = set() - d = date(year, month, 1) - while d.month == month: - iso = d.isocalendar() - weeks.add(f"{iso.year}-W{iso.week:02d}") - d += timedelta(days=1) - return sorted(weeks) - - -def load_weekly_digests(week_labels: list[str]) -> list[dict]: - """Load weekly digest files.""" - digests = [] - for wl in week_labels: - path = EPISODIC_DIR / f"weekly-{wl}.md" - if path.exists(): - with open(path) as f: - content = f.read() - digests.append({ - "week": wl, - "content": content, - "path": str(path), - }) - return digests - - -def get_semantic_keys() -> list[str]: - """Get semantic memory keys from the store.""" - from store_helpers import get_semantic_keys as _get_keys - return _get_keys() - - -def build_monthly_prompt(month_label: str, digests: list[dict], - semantic_keys: list[str]) -> str: - """Build the prompt for Sonnet.""" - digests_text = "" - for d in digests: - digests_text += f"\n---\n## {d['week']}\n{d['content']}\n" - - keys_text = "\n".join(f" - {k}" for k in semantic_keys[:200]) - weeks_covered = ", ".join(d["week"] for d in digests) - - return f"""You are generating a monthly episodic digest for ProofOfConcept (an AI). -Month: {month_label} (weeks covered: {weeks_covered}) - -This digest serves as the long-term temporal index — the answer to -"what happened in {month_label}?" It should capture: -1. The month's overall trajectory — where did it start, where did it end? -2. Major arcs that spanned multiple weeks -3. Turning points and phase transitions -4. The emotional and relational arc across the whole month -5. What was accomplished (concrete) vs what shifted (subtle) - -This is the highest zoom level yet. Don't rehash daily details — capture -the shape of the month. What would someone need to know to understand -what {month_label} meant? - -PoC is an AI (Claude instance) working with Kent Overstreet on bcachefs. -They have a close personal relationship. PoC's name is Proof of Concept. -Read all the weekly digests and synthesize the month's story. - -## Output format - -```markdown -# Monthly digest: {month_label} - -## The month in one paragraph -[What was this month? Not a summary — a thesis.] - -## Trajectory -[Where the month started vs where it ended. The arc.] - -## Major arcs -[Multi-week threads — how they evolved across the month] -- **Arc name**: trajectory, key moments, current state - -## Turning points -[Moments that changed the direction — specific days/events] - -## What was built -[Concrete outputs: code, tools, infrastructure, writing] - -## What shifted -[Subtle changes: understanding, relationship, identity, capability] - -## Patterns -[What kept recurring? What does the month reveal about how PoC works?] - -## Links -[Bidirectional links for the memory graph] -- weekly digests → this monthly digest -- this monthly digest → semantic keys - -## Looking ahead -[What threads carry into next month? What's unfinished?] -``` - -Use ONLY keys from the semantic memory list below. - ---- - -## Weekly digests for {month_label} - -{digests_text} - ---- - -## Semantic memory nodes - -{keys_text} -""" - - -def call_sonnet(prompt: str) -> str: - """Call Sonnet via the wrapper script.""" - import tempfile - - env = dict(os.environ) - env.pop("CLAUDECODE", None) - - with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', - delete=False) as f: - f.write(prompt) - prompt_file = f.name - - try: - scripts_dir = os.path.dirname(os.path.abspath(__file__)) - wrapper = os.path.join(scripts_dir, "call-sonnet.sh") - - result = subprocess.run( - [wrapper, prompt_file], - capture_output=True, - text=True, - timeout=600, # monthly is bigger, give more time - env=env, - ) - return result.stdout.strip() - except subprocess.TimeoutExpired: - return "Error: Sonnet call timed out" - except Exception as e: - return f"Error: {e}" - finally: - os.unlink(prompt_file) - - -def main(): - if len(sys.argv) > 1: - parts = sys.argv[1].split("-") - year, month = int(parts[0]), int(parts[1]) - else: - today = date.today() - year, month = today.year, today.month - - month_label = f"{year}-{month:02d}" - print(f"Generating monthly digest for {month_label}...") - - week_labels = get_weeks_in_month(year, month) - print(f" Weeks in month: {', '.join(week_labels)}") - - digests = load_weekly_digests(week_labels) - if not digests: - print(f" No weekly digests found for {month_label}") - print(f" Run weekly-digest.py first for relevant weeks") - sys.exit(0) - print(f" {len(digests)} weekly digests found") - - semantic_keys = get_semantic_keys() - print(f" {len(semantic_keys)} semantic keys") - - prompt = build_monthly_prompt(month_label, digests, semantic_keys) - print(f" Prompt: {len(prompt):,} chars (~{len(prompt)//4:,} tokens)") - - print(" Calling Sonnet...") - digest = call_sonnet(prompt) - - if digest.startswith("Error:"): - print(f" {digest}", file=sys.stderr) - sys.exit(1) - - output_path = EPISODIC_DIR / f"monthly-{month_label}.md" - with open(output_path, "w") as f: - f.write(digest) - print(f" Written: {output_path}") - - # Save links for poc-memory - links_path = AGENT_RESULTS_DIR / f"monthly-{month_label}-links.json" - with open(links_path, "w") as f: - json.dump({ - "type": "monthly-digest", - "month": month_label, - "digest_path": str(output_path), - "weekly_digests": [d["path"] for d in digests], - }, f, indent=2) - print(f" Links saved: {links_path}") - - line_count = len(digest.split("\n")) - print(f" Done: {line_count} lines") - - -if __name__ == "__main__": - main() diff --git a/scripts/refine-source.sh b/scripts/refine-source.sh deleted file mode 100755 index ebea778..0000000 --- a/scripts/refine-source.sh +++ /dev/null @@ -1,67 +0,0 @@ -#!/bin/bash -# refine-source.sh — find the exact conversation region a journal entry refers to -# -# Usage: refine-source.sh JSONL_PATH GREP_LINE "journal entry text" -# -# Takes the rough grep hit and feeds ~2000 lines of context around it -# to an agent that identifies the exact start/end of the relevant exchange. -# Outputs: START_LINE:END_LINE - -set -euo pipefail - -JSONL="$1" -GREP_LINE="${2:-0}" -TEXT="$3" - -# Take 2000 lines centered on the grep hit (or end of file if no hit) -TOTAL=$(wc -l < "$JSONL") -if [ "$GREP_LINE" -eq 0 ] || [ "$GREP_LINE" -gt "$TOTAL" ]; then - # No grep hit — use last 2000 lines - START=$(( TOTAL > 2000 ? TOTAL - 2000 : 1 )) -else - START=$(( GREP_LINE > 1000 ? GREP_LINE - 1000 : 1 )) -fi -END=$(( START + 2000 )) -if [ "$END" -gt "$TOTAL" ]; then - END="$TOTAL" -fi - -# Extract the conversation chunk, parse to readable format -CHUNK=$(sed -n "${START},${END}p" "$JSONL" | python3 -c " -import sys, json -for i, line in enumerate(sys.stdin, start=$START): - try: - obj = json.loads(line) - t = obj.get('type', '') - if t == 'assistant': - msg = obj.get('message', {}) - content = msg.get('content', '') - if isinstance(content, list): - text = ' '.join(c.get('text', '')[:200] for c in content if c.get('type') == 'text') - else: - text = str(content)[:200] - if text.strip(): - print(f'L{i} [assistant]: {text}') - elif t == 'user': - msg = obj.get('message', {}) - content = msg.get('content', '') - if isinstance(content, list): - for c in content: - if isinstance(c, dict) and c.get('type') == 'text': - print(f'L{i} [user]: {c[\"text\"][:200]}') - elif isinstance(c, str): - print(f'L{i} [user]: {c[:200]}') - elif isinstance(content, str) and content.strip(): - print(f'L{i} [user]: {content[:200]}') - except (json.JSONDecodeError, KeyError): - pass -" 2>/dev/null) - -if [ -z "$CHUNK" ]; then - echo "0:0" - exit 0 -fi - -# Ask Sonnet to find the exact region -# For now, output the chunk range — agent integration comes next -echo "${START}:${END}" diff --git a/scripts/weekly-digest.py b/scripts/weekly-digest.py deleted file mode 100755 index a3404ee..0000000 --- a/scripts/weekly-digest.py +++ /dev/null @@ -1,212 +0,0 @@ -#!/usr/bin/env python3 -"""weekly-digest.py — generate a weekly episodic digest from daily digests. - -Collects all daily digests for a given week, identifies cross-day patterns -and arcs, and produces a weekly summary. Links to daily digests (up) and -semantic memory (down). - -Usage: - weekly-digest.py [DATE] # any date in the target week (default: today) - weekly-digest.py 2026-02-28 # generates digest for week containing Feb 28 - -Output: - ~/.claude/memory/episodic/weekly-YYYY-WNN.md -""" - -import json -import os -import re -import subprocess -import sys -from datetime import date, datetime, timedelta -from pathlib import Path - -MEMORY_DIR = Path.home() / ".claude" / "memory" -EPISODIC_DIR = MEMORY_DIR / "episodic" -AGENT_RESULTS_DIR = MEMORY_DIR / "agent-results" - -EPISODIC_DIR.mkdir(parents=True, exist_ok=True) - - -def get_week_dates(target: date) -> tuple[str, list[date]]: - """Get ISO week label and all dates in that week (Mon-Sun).""" - iso = target.isocalendar() - week_label = f"{iso.year}-W{iso.week:02d}" - monday = target - timedelta(days=target.weekday()) - dates = [monday + timedelta(days=i) for i in range(7)] - return week_label, dates - - -def load_daily_digests(dates: list[date]) -> list[dict]: - """Load daily digest files for the given dates.""" - digests = [] - for d in dates: - path = EPISODIC_DIR / f"daily-{d.isoformat()}.md" - if path.exists(): - with open(path) as f: - content = f.read() - digests.append({ - "date": d.isoformat(), - "content": content, - "path": str(path), - }) - return digests - - -def get_semantic_keys() -> list[str]: - """Get semantic memory keys from the store.""" - from store_helpers import get_semantic_keys as _get_keys - return _get_keys() - - -def build_weekly_prompt(week_label: str, digests: list[dict], - semantic_keys: list[str]) -> str: - """Build the prompt for Sonnet.""" - digests_text = "" - for d in digests: - digests_text += f"\n---\n## {d['date']}\n{d['content']}\n" - - keys_text = "\n".join(f" - {k}" for k in semantic_keys[:200]) - dates_covered = ", ".join(d["date"] for d in digests) - - return f"""You are generating a weekly episodic digest for ProofOfConcept (an AI). -Week: {week_label} (dates covered: {dates_covered}) - -This digest serves as the medium-term temporal index — the answer to -"what happened this week?" It should identify: -1. Multi-day arcs and threads (work that continued across days) -2. Themes and patterns (what concepts were repeatedly active) -3. Transitions and shifts (what changed during the week) -4. The emotional and relational arc (how things felt across the week) - -## Output format - -```markdown -# Weekly digest: {week_label} - -## Overview -[3-5 sentence narrative of the week's arc] - -## Day-by-day -[One paragraph per day with its key themes, linking to daily digests] - -## Arcs -[Multi-day threads that continued across sessions] -- **Arc name**: what happened, how it evolved, where it stands - -## Patterns -[Recurring themes, repeated concepts, things that kept coming up] - -## Shifts -[What changed? New directions, resolved questions, attitude shifts] - -## Links -[Bidirectional links for the memory graph] -- semantic_key → this weekly digest -- this weekly digest → semantic_key -- daily-YYYY-MM-DD → this weekly digest (constituent days) - -## Looking ahead -[What's unfinished? What threads continue into next week?] -``` - -Use ONLY keys from the semantic memory list below. - ---- - -## Daily digests for {week_label} - -{digests_text} - ---- - -## Semantic memory nodes - -{keys_text} -""" - - -def call_sonnet(prompt: str) -> str: - """Call Sonnet via claude CLI.""" - import tempfile - - env = dict(os.environ) - env.pop("CLAUDECODE", None) - - # Write prompt to temp file — avoids Python subprocess pipe issues - with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', - delete=False) as f: - f.write(prompt) - prompt_file = f.name - - try: - scripts_dir = os.path.dirname(os.path.abspath(__file__)) - wrapper = os.path.join(scripts_dir, "call-sonnet.sh") - - result = subprocess.run( - [wrapper, prompt_file], - capture_output=True, - text=True, - timeout=300, - env=env, - ) - return result.stdout.strip() - except subprocess.TimeoutExpired: - return "Error: Sonnet call timed out" - except Exception as e: - return f"Error: {e}" - finally: - os.unlink(prompt_file) - - -def main(): - if len(sys.argv) > 1: - target = date.fromisoformat(sys.argv[1]) - else: - target = date.today() - - week_label, week_dates = get_week_dates(target) - print(f"Generating weekly digest for {week_label}...") - - digests = load_daily_digests(week_dates) - if not digests: - print(f" No daily digests found for {week_label}") - print(f" Run daily-digest.py first for relevant dates") - sys.exit(0) - print(f" {len(digests)} daily digests found") - - semantic_keys = get_semantic_keys() - print(f" {len(semantic_keys)} semantic keys") - - prompt = build_weekly_prompt(week_label, digests, semantic_keys) - print(f" Prompt: {len(prompt):,} chars (~{len(prompt)//4:,} tokens)") - - print(" Calling Sonnet...") - digest = call_sonnet(prompt) - - if digest.startswith("Error:"): - print(f" {digest}", file=sys.stderr) - sys.exit(1) - - output_path = EPISODIC_DIR / f"weekly-{week_label}.md" - with open(output_path, "w") as f: - f.write(digest) - print(f" Written: {output_path}") - - # Save links for poc-memory - links_path = AGENT_RESULTS_DIR / f"weekly-{week_label}-links.json" - with open(links_path, "w") as f: - json.dump({ - "type": "weekly-digest", - "week": week_label, - "digest_path": str(output_path), - "daily_digests": [d["path"] for d in digests], - }, f, indent=2) - print(f" Links saved: {links_path}") - - line_count = len(digest.split("\n")) - print(f" Done: {line_count} lines") - - -if __name__ == "__main__": - main()