110 lines
3.9 KiB
Python
Executable File
110 lines
3.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from collections import OrderedDict
|
|
from pathlib import Path
|
|
|
|
|
|
def uniq(items: list[str]) -> list[str]:
|
|
return list(OrderedDict((item, None) for item in items if item))
|
|
|
|
|
|
def extract_texts(obj) -> list[str]:
|
|
texts: list[str] = []
|
|
if isinstance(obj, dict):
|
|
if obj.get("type") in {"input_text", "output_text"} and isinstance(obj.get("text"), str):
|
|
texts.append(obj["text"])
|
|
for value in obj.values():
|
|
texts.extend(extract_texts(value))
|
|
elif isinstance(obj, list):
|
|
for value in obj:
|
|
texts.extend(extract_texts(value))
|
|
return texts
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) not in {2, 3}:
|
|
print("usage: session_to_memory.py SESSION_JSONL [OUTPUT_MD]", file=sys.stderr)
|
|
return 2
|
|
|
|
session_path = Path(sys.argv[1]).expanduser().resolve()
|
|
output_path = Path(sys.argv[2]).expanduser().resolve() if len(sys.argv) == 3 else session_path.with_suffix(".memory.md")
|
|
|
|
user_texts: list[str] = []
|
|
assistant_texts: list[str] = []
|
|
paths: list[str] = []
|
|
session_meta = {}
|
|
|
|
path_pattern = re.compile(r"(\/[A-Za-z0-9._ \-\/]+)")
|
|
|
|
with session_path.open() as handle:
|
|
for raw_line in handle:
|
|
raw_line = raw_line.strip()
|
|
if not raw_line:
|
|
continue
|
|
event = json.loads(raw_line)
|
|
payload = event.get("payload", {})
|
|
if event.get("type") == "session_meta":
|
|
session_meta = payload
|
|
if event.get("type") == "response_item" and payload.get("type") == "message":
|
|
role = payload.get("role")
|
|
texts = extract_texts(payload.get("content", []))
|
|
if role == "user":
|
|
user_texts.extend(texts)
|
|
elif role == "assistant":
|
|
assistant_texts.extend(texts)
|
|
for text in texts:
|
|
paths.extend(path_pattern.findall(text))
|
|
|
|
user_texts = uniq([text.strip() for text in user_texts if text.strip()])
|
|
assistant_texts = uniq([text.strip() for text in assistant_texts if text.strip()])
|
|
paths = uniq([path.strip() for path in paths if path.strip()])
|
|
|
|
output = []
|
|
output.append(f"# Session Memory Draft")
|
|
output.append("")
|
|
output.append(f"- Session file: `{session_path}`")
|
|
if session_meta.get("timestamp"):
|
|
output.append(f"- Session timestamp: `{session_meta['timestamp']}`")
|
|
if session_meta.get("cwd"):
|
|
output.append(f"- Working directory: `{session_meta['cwd']}`")
|
|
output.append("")
|
|
output.append("## User Requests")
|
|
output.append("")
|
|
for text in user_texts[:20]:
|
|
output.append(f"- {text.replace(chr(10), ' ')}")
|
|
output.append("")
|
|
output.append("## Candidate Paths Mentioned")
|
|
output.append("")
|
|
for path in paths[:30]:
|
|
output.append(f"- `{path}`")
|
|
output.append("")
|
|
output.append("## Assistant Notes To Distill")
|
|
output.append("")
|
|
output.append("- Copy only stable preferences, project facts, and durable decisions into your memory markdown files.")
|
|
output.append("- Do not copy transient debugging output or large transcript excerpts.")
|
|
output.append("- Summarize in your own words before adding to long-term memory.")
|
|
output.append("")
|
|
output.append("## Sample Durable Summary")
|
|
output.append("")
|
|
output.append("- Project:")
|
|
output.append("- User preference:")
|
|
output.append("- Important decision:")
|
|
output.append("- Follow-up worth remembering:")
|
|
output.append("")
|
|
output.append("## Assistant Messages Snapshot")
|
|
output.append("")
|
|
for text in assistant_texts[:10]:
|
|
output.append(f"- {text.replace(chr(10), ' ')}")
|
|
|
|
output_path.write_text("\n".join(output) + "\n")
|
|
print(output_path)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|