delete superseded Python scripts
Seven scripts (1,658 lines) replaced by native Rust subcommands: - journal-agent.py → poc-memory journal-enrich - digest-link-parser.py → poc-memory digest-links - apply-consolidation.py → poc-memory apply-consolidation - daily-digest.py → poc-memory digest daily - weekly-digest.py → poc-memory digest weekly - monthly-digest.py → poc-memory digest monthly - refine-source.sh → folded into journal-enrich Also updated poc-journal to use Rust journal-enrich instead of Python journal-agent.py, and cleaned up stale __pycache__. Remaining Python (2,154 lines): consolidation-agents, consolidation-loop, content-promotion-agent, bulk-categorize, retroactive-digest, store_helpers, call-sonnet.sh, daily-check.sh — still active and evolving.
This commit is contained in:
ProofOfConcept
parent
59e2f39479
commit
3afc947b88
8 changed files with 2 additions and 1658 deletions
|
|
@ -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()
|
||||
|
|
@ -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.""",
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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()
|
||||
|
|
@ -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()
|
||||
|
|
@ -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()
|
||||
|
|
@ -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}"
|
||||
|
|
@ -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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue