personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Split unified Sense agent output into per-agent file locations."""
5
6import json
7from datetime import datetime, timezone
8from pathlib import Path
9
10
11def _write_json_atomic(path: Path, data: object) -> None:
12 """Atomically write JSON data to a file."""
13 path.parent.mkdir(parents=True, exist_ok=True)
14 tmp = path.with_suffix(f"{path.suffix}.tmp")
15 tmp.write_text(json.dumps(data), encoding="utf-8")
16 tmp.replace(path)
17
18
19def _write_text_atomic(path: Path, text: str) -> None:
20 """Atomically write text data to a file."""
21 path.parent.mkdir(parents=True, exist_ok=True)
22 tmp = path.with_suffix(f"{path.suffix}.tmp")
23 tmp.write_text(text, encoding="utf-8")
24 tmp.replace(path)
25
26
27def write_sense_outputs(
28 sense_json: dict, seg_dir: Path, stream: str | None = None
29) -> None:
30 """Write unified Sense output into per-agent files."""
31 agents_dir = seg_dir / "talents"
32
33 density = sense_json["density"]
34 activity_summary = sense_json.get("activity_summary") or ""
35 entities = sense_json.get("entities") or []
36 facets = sense_json.get("facets") or []
37 meeting_detected = bool(sense_json.get("meeting_detected"))
38 speakers = sense_json.get("speakers") or []
39
40 _write_text_atomic(agents_dir / "activity.md", activity_summary)
41 _write_json_atomic(agents_dir / "facets.json", facets)
42 _write_json_atomic(
43 agents_dir / "density.json",
44 {
45 "classification": density,
46 "transcript_lines": 0,
47 "screen_frames": 0,
48 "timestamp": datetime.now(tz=timezone.utc).isoformat(),
49 },
50 )
51 # Write both structured and human-readable Sense outputs here.
52 # think/cluster.py discovers talent outputs by globbing
53 # {segment}/talents/**/*.md for load.talents.{name} consumers.
54 # Dropping sense.md would silently break downstream talents such as
55 # participation that rely on the sense markdown file being present.
56 _write_json_atomic(agents_dir / "sense.json", sense_json)
57
58 if entities:
59 lines = ["# Sense Entities", ""]
60 for entity in entities:
61 if not isinstance(entity, dict):
62 continue
63 lines.append(
64 "- "
65 f"{entity.get('type', '')} — {entity.get('name', '')} "
66 f"(role={entity.get('role', '')}, source={entity.get('source', '')}) "
67 f"— {entity.get('context', '')}"
68 )
69 if len(lines) > 2:
70 _write_text_atomic(agents_dir / "sense.md", "\n".join(lines))
71
72 if meeting_detected:
73 _write_json_atomic(agents_dir / "speakers.json", speakers)
74
75
76def write_idle_stubs(seg_dir: Path) -> None:
77 """Write minimal idle output files for a segment."""
78 _write_json_atomic(
79 seg_dir / "talents" / "density.json",
80 {
81 "classification": "idle",
82 "transcript_lines": 0,
83 "screen_frames": 0,
84 "timestamp": datetime.now(tz=timezone.utc).isoformat(),
85 },
86 )