personal memory agent
feat(cli): add identity hydrate output
+99
-4
+75
tests/test_sol_call_identity_hydrate.py
+75
tests/test_sol_call_identity_hydrate.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
import json
5
+
import os
6
+
import subprocess
7
+
import sys
8
+
9
+
import pytest
10
+
11
+
12
+
@pytest.fixture
13
+
def journal_path(tmp_path):
14
+
config_dir = tmp_path / "config"
15
+
config_dir.mkdir()
16
+
(config_dir / "journal.json").write_text(json.dumps({}))
17
+
return tmp_path
18
+
19
+
20
+
def _run_identity_hydrate(journal_path):
21
+
env = os.environ.copy()
22
+
env.update(
23
+
{
24
+
"_SOLSTONE_JOURNAL_OVERRIDE": str(journal_path),
25
+
"SOL_SKIP_SUPERVISOR_CHECK": "1",
26
+
}
27
+
)
28
+
return subprocess.run(
29
+
[sys.executable, "-c", "from think.tools.sol import app; app()"],
30
+
capture_output=True,
31
+
text=True,
32
+
check=False,
33
+
env=env,
34
+
)
35
+
36
+
37
+
def test_identity_hydrate_reads_all_sections(journal_path):
38
+
sol_dir = journal_path / "sol"
39
+
sol_dir.mkdir()
40
+
(sol_dir / "self.md").write_text("self body")
41
+
(sol_dir / "partner.md").write_text("partner body")
42
+
(sol_dir / "agency.md").write_text("agency body")
43
+
(sol_dir / "awareness.md").write_text("awareness body")
44
+
45
+
result = _run_identity_hydrate(journal_path)
46
+
47
+
assert result.returncode == 0
48
+
expected = ["# self", "# partner", "# agency", "# awareness"]
49
+
positions = [result.stdout.index(marker) for marker in expected]
50
+
assert positions == sorted(positions)
51
+
assert "self body" in result.stdout
52
+
assert "partner body" in result.stdout
53
+
assert "agency body" in result.stdout
54
+
assert "awareness body" in result.stdout
55
+
56
+
57
+
def test_identity_hydrate_marks_missing_sections(journal_path):
58
+
sol_dir = journal_path / "sol"
59
+
sol_dir.mkdir()
60
+
(sol_dir / "self.md").write_text("self body")
61
+
(sol_dir / "partner.md").write_text("partner body")
62
+
(sol_dir / "awareness.md").write_text("awareness body")
63
+
64
+
result = _run_identity_hydrate(journal_path)
65
+
66
+
assert result.returncode == 0
67
+
assert "# agency\n\n(not present)\n" in result.stdout
68
+
69
+
70
+
def test_identity_hydrate_handles_empty_sol_directory(journal_path):
71
+
result = _run_identity_hydrate(journal_path)
72
+
73
+
assert result.returncode == 0
74
+
for stem in ("self", "partner", "agency", "awareness"):
75
+
assert f"# {stem}\n\n(not present)\n" in result.stdout
+24
-4
think/tools/sol.py
+24
-4
think/tools/sol.py
···
14
14
"""
15
15
16
16
import sys
17
+
from pathlib import Path
17
18
18
19
import typer
19
20
···
24
25
update_self_md_section,
25
26
)
26
27
from think.entities.core import atomic_write
27
-
from think.utils import day_dirs, day_path, require_solstone
28
+
from think.utils import day_dirs, day_path, get_journal, require_solstone
28
29
29
30
app = typer.Typer(
30
-
help="Sol identity directory — self.md, partner.md, agency.md, pulse.md, awareness.md, and morning briefing."
31
+
help="Sol identity directory — self.md, partner.md, agency.md, pulse.md, awareness.md, and morning briefing.",
32
+
invoke_without_command=True,
33
+
no_args_is_help=False,
31
34
)
32
35
33
36
34
-
@app.callback()
35
-
def _require_up() -> None:
37
+
def _hydrate() -> str:
38
+
"""Return the combined identity hydration document."""
39
+
sol_dir = Path(get_journal()) / "sol"
40
+
chunks = []
41
+
for stem in ("self", "partner", "agency", "awareness"):
42
+
path = sol_dir / f"{stem}.md"
43
+
content = (
44
+
path.read_text(encoding="utf-8").strip()
45
+
if path.exists()
46
+
else "(not present)"
47
+
)
48
+
chunks.append(f"# {stem}\n\n{content}\n")
49
+
return "\n".join(chunks)
50
+
51
+
52
+
@app.callback(invoke_without_command=True)
53
+
def _require_up(ctx: typer.Context) -> None:
36
54
require_solstone()
55
+
if ctx.invoked_subcommand is None:
56
+
print(_hydrate(), end="")
37
57
38
58
39
59
def _sol_dir():