Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
from __future__ import annotations
import hashlibimport jsonfrom pathlib import Pathfrom unittest.mock import MagicMock, patch
import pytestimport requests
import solstone.think.utils as think_utilsfrom solstone.observe.utils import compute_bytes_sha256
def _setup_journal(tmp_path, *, include_stream_json: bool = False): first_dir = tmp_path / "chronicle" / "20260413" / "laptop" / "143022_300" second_dir = tmp_path / "chronicle" / "20260413" / "laptop" / "150000_600" first_dir.mkdir(parents=True) second_dir.mkdir(parents=True)
(first_dir / "audio.flac").write_bytes(b"audio-data-one") (first_dir / "transcript.jsonl").write_bytes(b"transcript-one") (second_dir / "audio.flac").write_bytes(b"audio-data-two") (second_dir / "transcript.jsonl").write_bytes(b"transcript-two")
if include_stream_json: (first_dir / "stream.json").write_bytes(b'{"name": "laptop"}') (second_dir / "stream.json").write_bytes(b'{"name": "laptop"}')
return tmp_path
def _set_journal_override(monkeypatch, journal_path): monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path)) think_utils._journal_path_cache = None
def _make_session( *, manifest_data=None, get_status=200, post_status=200, post_json=None): mock = MagicMock(spec=requests.Session) mock.headers = {}
get_response = MagicMock() get_response.status_code = get_status get_response.json.return_value = manifest_data if manifest_data is not None else {} get_response.text = "GET error" mock.get.return_value = get_response
post_response = MagicMock() post_response.status_code = post_status post_response.json.return_value = ( post_json if post_json is not None else { "segments_received": 1, "segments_skipped": 0, "segments_deconflicted": 0, "errors": [], } ) post_response.text = "POST error" mock.post.return_value = post_response
return mock
def _setup_entities(tmp_path): """Create test entities in journal fixture.""" entities_dir = tmp_path / "entities"
alice_dir = entities_dir / "alice_johnson" alice_dir.mkdir(parents=True) alice = { "id": "alice_johnson", "name": "Alice Johnson", "type": "Person", "created_at": 1000, } (alice_dir / "entity.json").write_text(json.dumps(alice), encoding="utf-8")
bob_dir = entities_dir / "bob_smith" bob_dir.mkdir(parents=True) bob = {"id": "bob_smith", "name": "Bob Smith", "type": "Person", "created_at": 2000} (bob_dir / "entity.json").write_text(json.dumps(bob), encoding="utf-8")
blocked_dir = entities_dir / "blocked_user" blocked_dir.mkdir(parents=True) blocked = { "id": "blocked_user", "name": "Blocked", "type": "Person", "blocked": True, "created_at": 3000, } (blocked_dir / "entity.json").write_text(json.dumps(blocked), encoding="utf-8")
return {"alice_johnson": alice, "bob_smith": bob, "blocked_user": blocked}
def _entity_hash(entity: dict) -> str: return hashlib.sha256( json.dumps(entity, sort_keys=True, ensure_ascii=False).encode() ).hexdigest()
def _setup_facets(tmp_path): """Create test facets in journal fixture.""" facets_dir = tmp_path / "facets"
work_dir = facets_dir / "work" work_dir.mkdir(parents=True) (work_dir / "facet.json").write_text('{"title": "Work"}', encoding="utf-8")
ent_dir = work_dir / "entities" / "alice" ent_dir.mkdir(parents=True) (ent_dir / "entity.json").write_text('{"id": "alice"}', encoding="utf-8") (ent_dir / "observations.jsonl").write_text('{"text": "obs1"}\n', encoding="utf-8")
det_dir = work_dir / "entities" (det_dir / "20260413.jsonl").write_text('{"entity": "test"}\n', encoding="utf-8")
personal_dir = facets_dir / "personal" personal_dir.mkdir(parents=True) (personal_dir / "facet.json").write_text('{"title": "Personal"}', encoding="utf-8")
news_dir = personal_dir / "news" news_dir.mkdir(parents=True) (news_dir / "20260413.md").write_text("# News\nHello world\n", encoding="utf-8")
return facets_dir
def _facet_file_hash(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest()
def _setup_imports(tmp_path): """Create test import metadata dirs in journal fixture.""" imports_dir = tmp_path / "imports" imports_dir.mkdir(parents=True, exist_ok=True)
dir1 = imports_dir / "20260101_090000" dir1.mkdir() import_json_1 = {"original_filename": "cal.zip", "file_size": 100} imported_json_1 = { "processed_timestamp": "20260101_090000", "total_files_created": 1, } manifest_1 = [{"id": "event-0", "title": "Test Event"}] (dir1 / "import.json").write_text(json.dumps(import_json_1), encoding="utf-8") (dir1 / "imported.json").write_text(json.dumps(imported_json_1), encoding="utf-8") (dir1 / "content_manifest.jsonl").write_text( json.dumps(manifest_1[0]) + "\n", encoding="utf-8" )
dir2 = imports_dir / "20260102_100000" dir2.mkdir() import_json_2 = {"original_filename": "chat.zip", "file_size": 200} imported_json_2 = { "processed_timestamp": "20260102_100000", "total_files_created": 2, } manifest_2 = [{"id": "conv-0", "title": "Test Convo"}] (dir2 / "import.json").write_text(json.dumps(import_json_2), encoding="utf-8") (dir2 / "imported.json").write_text(json.dumps(imported_json_2), encoding="utf-8") (dir2 / "content_manifest.jsonl").write_text( json.dumps(manifest_2[0]) + "\n", encoding="utf-8" )
(imports_dir / "plaud.json").write_text('{"last_sync": 123}', encoding="utf-8")
source_dir = imports_dir / "abcd1234" source_dir.mkdir() (source_dir / "segments").mkdir()
return { "20260101_090000": { "import_json": import_json_1, "imported_json": imported_json_1, "content_manifest": manifest_1, }, "20260102_100000": { "import_json": import_json_2, "imported_json": imported_json_2, "content_manifest": manifest_2, }, }
def _import_hash(import_data): """Compute hash matching export_imports algorithm.""" hash_input = json.dumps( { "import_json": import_data["import_json"], "imported_json": import_data["imported_json"], "content_manifest": import_data["content_manifest"], }, sort_keys=True, ensure_ascii=False, ).encode() return hashlib.sha256(hash_input).hexdigest()
def _setup_config(tmp_path): """Create test config in journal fixture.""" config_dir = tmp_path / "config" config_dir.mkdir(parents=True, exist_ok=True) config = { "identity": {"name": "Test", "preferred": "Tester", "timezone": "UTC"}, "convey": { "allow_network_access": False, "password_hash": "secret_hash", "secret": "secret_val", }, "setup": {"completed_at": 12345}, "env": {"KEY": "val"}, "retention": {"days": 30}, } (config_dir / "journal.json").write_text(json.dumps(config), encoding="utf-8") return config
class TestExportSegments: def test_manifest_query_and_delta(self, tmp_path, monkeypatch): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path) _set_journal_override(monkeypatch, journal)
manifest_data = { "20260413": { "laptop/143022_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(b"audio-data-one"), "size": len(b"audio-data-one"), }, { "name": "transcript.jsonl", "sha256": compute_bytes_sha256(b"transcript-one"), "size": len(b"transcript-one"), }, ] } } } mock_session = _make_session(manifest_data=manifest_data)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=False )
assert mock_session.post.call_count == 1 metadata = json.loads(mock_session.post.call_args.kwargs["data"]["metadata"]) assert metadata["segments"][0]["segment_key"] == "150000_600"
def test_dry_run_output(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path) _set_journal_override(monkeypatch, journal)
mock_session = _make_session(manifest_data={})
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=True )
assert mock_session.post.call_count == 0 output = capsys.readouterr().out assert "20260413: 2 segment(s)" in output assert "Dry run: would send 2, skip 0" in output
def test_dry_run_skipped_not_double_counted(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path) _set_journal_override(monkeypatch, journal)
manifest_data = { "20260413": { "laptop/143022_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(b"audio-data-one"), "size": len(b"audio-data-one"), }, { "name": "transcript.jsonl", "sha256": compute_bytes_sha256(b"transcript-one"), "size": len(b"transcript-one"), }, ] } } } mock_session = _make_session(manifest_data=manifest_data)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=True )
assert mock_session.post.call_count == 0 output = capsys.readouterr().out assert "20260413: 1 segment(s)" in output assert "Dry run: would send 1, skip 1" in output
def test_retry_on_5xx(self, tmp_path, monkeypatch): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path) _set_journal_override(monkeypatch, journal)
second_dir = journal / "chronicle" / "20260413" / "laptop" / "150000_600" for file_path in second_dir.iterdir(): file_path.unlink() second_dir.rmdir()
mock_session = _make_session(manifest_data={}) first = MagicMock(status_code=500, text="server error") second = MagicMock(status_code=500, text="server error") success = MagicMock(status_code=200, text="ok") success.json.return_value = { "segments_received": 1, "segments_skipped": 0, "segments_deconflicted": 0, "errors": [], } mock_session.post.side_effect = [first, second, success]
with ( patch( "solstone.observe.export.requests.Session", return_value=mock_session ), patch("solstone.observe.export.time.sleep") as mock_sleep, ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=False )
assert mock_session.post.call_count == 3 assert mock_sleep.called
def test_auth_error_401(self): from solstone.observe.export import _query_manifest
mock_session = _make_session(get_status=401)
with pytest.raises(ValueError, match="Authentication failed"): _query_manifest(mock_session, "https://example.com", "test-key")
def test_auth_error_403(self): from solstone.observe.export import _query_manifest
mock_session = _make_session(get_status=403)
with pytest.raises(ValueError, match="journal source revoked"): _query_manifest(mock_session, "https://example.com", "test-key")
def test_connection_error(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path) _set_journal_override(monkeypatch, journal)
mock_session = _make_session(manifest_data={}) mock_session.get.side_effect = requests.ConnectionError
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=False )
assert mock_session.post.call_count == 0 assert "Connection failed" in capsys.readouterr().out
def test_idempotent(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path) _set_journal_override(monkeypatch, journal)
manifest_data = { "20260413": { "laptop/143022_300": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(b"audio-data-one"), "size": len(b"audio-data-one"), }, { "name": "transcript.jsonl", "sha256": compute_bytes_sha256(b"transcript-one"), "size": len(b"transcript-one"), }, ] }, "laptop/150000_600": { "files": [ { "name": "audio.flac", "sha256": compute_bytes_sha256(b"audio-data-two"), "size": len(b"audio-data-two"), }, { "name": "transcript.jsonl", "sha256": compute_bytes_sha256(b"transcript-two"), "size": len(b"transcript-two"), }, ] }, } } mock_session = _make_session(manifest_data=manifest_data)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=False )
assert mock_session.post.call_count == 0 assert "up to date" in capsys.readouterr().out
def test_upload_error_isolation(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path) _set_journal_override(monkeypatch, journal)
mock_session = _make_session(manifest_data={}) first = MagicMock(status_code=400, text="bad request") second = MagicMock(status_code=200, text="ok") second.json.return_value = { "segments_received": 1, "segments_skipped": 0, "segments_deconflicted": 0, "errors": [], } mock_session.post.side_effect = [first, second]
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=False )
assert mock_session.post.call_count == 2 output = capsys.readouterr().out assert "1 sent" in output assert "1 failed" in output
def test_stream_json_excluded(self, tmp_path, monkeypatch): from solstone.observe.export import export_segments
journal = _setup_journal(tmp_path, include_stream_json=True) _set_journal_override(monkeypatch, journal)
mock_session = _make_session(manifest_data={})
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_segments( "https://example.com", "test-key", ["20260413"], dry_run=False )
for call in mock_session.post.call_args_list: post_kwargs = call.kwargs metadata = json.loads(post_kwargs["data"]["metadata"]) assert "stream.json" not in metadata["segments"][0]["files"] uploaded_names = [entry[1][0] for entry in post_kwargs["files"]] assert "stream.json" not in uploaded_names
class TestExportEntities: def test_manifest_delta(self, tmp_path, monkeypatch): from solstone.observe.export import export_entities
entities = _setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
manifest_data = { "received": { "alice_johnson": _entity_hash(entities["alice_johnson"]), } } post_json = { "created": 1, "auto_merged": 0, "staged": 0, "skipped": 0, "errors": [], } mock_session = _make_session(manifest_data=manifest_data, post_json=post_json)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 1 posted_data = mock_session.post.call_args.kwargs.get( "json" ) or mock_session.post.call_args[1].get("json") posted_entities = posted_data["entities"] posted_ids = [e["id"] for e in posted_entities] assert "bob_smith" in posted_ids assert "alice_johnson" not in posted_ids assert "blocked_user" not in posted_ids
def test_dry_run(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_entities
entities = _setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
manifest_data = { "received": { "alice_johnson": _entity_hash(entities["alice_johnson"]), } } mock_session = _make_session(manifest_data=manifest_data)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=True)
assert mock_session.post.call_count == 0 output = capsys.readouterr().out assert "1 new" in output assert "1 unchanged" in output
def test_idempotent(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_entities
entities = _setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
manifest_data = { "received": { "alice_johnson": _entity_hash(entities["alice_johnson"]), "bob_smith": _entity_hash(entities["bob_smith"]), } } mock_session = _make_session(manifest_data=manifest_data)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 0 output = capsys.readouterr().out assert "up to date" in output
def test_changed_entity(self, tmp_path, monkeypatch): from solstone.observe.export import export_entities
entities = _setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
manifest_data = { "received": { "alice_johnson": "stale_hash_that_does_not_match", "bob_smith": _entity_hash(entities["bob_smith"]), } } post_json = { "created": 0, "auto_merged": 1, "staged": 0, "skipped": 0, "errors": [], } mock_session = _make_session(manifest_data=manifest_data, post_json=post_json)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 1 posted_data = mock_session.post.call_args.kwargs.get( "json" ) or mock_session.post.call_args[1].get("json") posted_ids = [e["id"] for e in posted_data["entities"]] assert "alice_johnson" in posted_ids assert "bob_smith" not in posted_ids
def test_auth_error_401(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_entities
_setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session(get_status=401)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 0 assert "Authentication failed" in capsys.readouterr().out
def test_connection_error(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_entities
_setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session() mock_session.get.side_effect = requests.ConnectionError
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 0 assert "Connection failed" in capsys.readouterr().out
def test_empty_manifest(self, tmp_path, monkeypatch): from solstone.observe.export import export_entities
_setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
post_json = { "created": 2, "auto_merged": 0, "staged": 0, "skipped": 0, "errors": [], } mock_session = _make_session(manifest_data={}, post_json=post_json)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 1 posted_data = mock_session.post.call_args.kwargs.get( "json" ) or mock_session.post.call_args[1].get("json") posted_ids = [e["id"] for e in posted_data["entities"]] assert "alice_johnson" in posted_ids assert "bob_smith" in posted_ids assert "blocked_user" not in posted_ids
def test_response_errors_reported(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_entities
_setup_entities(tmp_path) _set_journal_override(monkeypatch, tmp_path)
post_json = { "created": 1, "auto_merged": 0, "staged": 0, "skipped": 0, "errors": ["Entity bob_smith: invalid type"], } mock_session = _make_session(manifest_data={}, post_json=post_json)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_entities("https://example.com", "test-key", dry_run=False)
output = capsys.readouterr().out assert "Entity bob_smith: invalid type" in output assert "1 error" in output
class TestExportFacets: def test_full_export(self, tmp_path, monkeypatch): from solstone.observe.export import export_facets
_setup_facets(tmp_path) _set_journal_override(monkeypatch, tmp_path)
post_json = {"created": 1, "merged": 0, "skipped": 0, "staged": 0, "errors": []} mock_session = _make_session( manifest_data={"received": {}}, post_json=post_json )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 2
calls_by_facet = {} for call in mock_session.post.call_args_list: metadata = json.loads(call.kwargs["data"]["metadata"]) facet = metadata["facets"][0]["name"] calls_by_facet[facet] = call.kwargs
assert set(calls_by_facet) == {"personal", "work"}
personal_files = calls_by_facet["personal"]["files"] assert [entry[0] for entry in personal_files] == ["files_0_0", "files_0_1"] personal_metadata = json.loads(calls_by_facet["personal"]["data"]["metadata"]) assert personal_metadata == { "facets": [ { "name": "personal", "files": [ {"path": "facet.json", "type": "facet_json"}, {"path": "news/20260413.md", "type": "news"}, ], } ] }
work_files = calls_by_facet["work"]["files"] assert [entry[0] for entry in work_files] == [ "files_0_0", "files_0_1", "files_0_2", "files_0_3", ] work_metadata = json.loads(calls_by_facet["work"]["data"]["metadata"]) assert work_metadata == { "facets": [ { "name": "work", "files": [ { "path": "entities/20260413.jsonl", "type": "detected_entities", }, { "path": "entities/alice/entity.json", "type": "entity_relationship", }, { "path": "entities/alice/observations.jsonl", "type": "entity_observations", }, {"path": "facet.json", "type": "facet_json"}, ], } ] }
def test_delta_mixed(self, tmp_path, monkeypatch): from solstone.observe.export import export_facets
facets_dir = _setup_facets(tmp_path) _set_journal_override(monkeypatch, tmp_path)
work_dir = facets_dir / "work" personal_dir = facets_dir / "personal" manifest_data = { "received": { "work/facet.json": _facet_file_hash(work_dir / "facet.json"), "work/entities/alice/entity.json": _facet_file_hash( work_dir / "entities" / "alice" / "entity.json" ), "work/entities/alice/observations.jsonl": "stale-hash", "personal/facet.json": _facet_file_hash(personal_dir / "facet.json"), "personal/news/20260413.md": _facet_file_hash( personal_dir / "news" / "20260413.md" ), } } post_json = {"created": 0, "merged": 1, "skipped": 0, "staged": 0, "errors": []} mock_session = _make_session(manifest_data=manifest_data, post_json=post_json)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 1 metadata = json.loads(mock_session.post.call_args.kwargs["data"]["metadata"]) assert metadata["facets"][0]["name"] == "work" assert metadata["facets"][0]["files"] == [ {"path": "entities/20260413.jsonl", "type": "detected_entities"}, { "path": "entities/alice/observations.jsonl", "type": "entity_observations", }, ]
def test_dry_run(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_facets
_setup_facets(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session(manifest_data={"received": {}})
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=True)
assert mock_session.post.call_count == 0 output = capsys.readouterr().out assert "personal: 2 new, 0 changed, 0 unchanged" in output assert "work: 4 new, 0 changed, 0 unchanged" in output assert ( "Dry run: 6 new files, 0 changed, 0 unchanged across 2 facet(s)" in output )
def test_idempotent(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_facets
facets_dir = _setup_facets(tmp_path) _set_journal_override(monkeypatch, tmp_path)
manifest_received = {} for facet_dir in sorted((facets_dir).iterdir()): if not facet_dir.is_dir(): continue for file_path in sorted(facet_dir.rglob("*")): if file_path.is_file(): rel_path = file_path.relative_to(facet_dir).as_posix() manifest_received[f"{facet_dir.name}/{rel_path}"] = ( _facet_file_hash(file_path) ) mock_session = _make_session(manifest_data={"received": manifest_received})
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 0 assert "up to date" in capsys.readouterr().out
def test_error_isolation(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_facets
_setup_facets(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session(manifest_data={"received": {}}) first = MagicMock(status_code=400, text="bad request") second = MagicMock(status_code=200, text="ok") second.json.return_value = { "created": 1, "merged": 0, "skipped": 0, "staged": 0, "errors": [], } mock_session.post.side_effect = [first, second]
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 2 output = capsys.readouterr().out assert "1 sent" in output assert "1 failed" in output
def test_new_facet_vs_changed(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_facets
facets_dir = _setup_facets(tmp_path) _set_journal_override(monkeypatch, tmp_path)
work_dir = facets_dir / "work" manifest_data = { "received": { f"work/{file_path.relative_to(work_dir).as_posix()}": "stale-hash" for file_path in sorted(work_dir.rglob("*")) if file_path.is_file() } } mock_session = _make_session(manifest_data=manifest_data)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=True)
assert mock_session.post.call_count == 0 output = capsys.readouterr().out assert "personal: 2 new, 0 changed, 0 unchanged" in output assert "work: 0 new, 4 changed, 0 unchanged" in output
def test_skips_invalid_facet_names(self, tmp_path, monkeypatch): from solstone.observe.export import export_facets
_setup_facets(tmp_path) bad_dir = tmp_path / "facets" / "BadName" bad_dir.mkdir(parents=True) (bad_dir / "facet.json").write_text('{"title": "Bad"}', encoding="utf-8") _set_journal_override(monkeypatch, tmp_path)
post_json = {"created": 1, "merged": 0, "skipped": 0, "staged": 0, "errors": []} mock_session = _make_session( manifest_data={"received": {}}, post_json=post_json )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 2 posted_facets = { json.loads(call.kwargs["data"]["metadata"])["facets"][0]["name"] for call in mock_session.post.call_args_list } assert posted_facets == {"personal", "work"}
def test_skips_events_directory(self, tmp_path, monkeypatch): from solstone.observe.export import export_facets
_setup_facets(tmp_path) events_dir = tmp_path / "facets" / "work" / "events" events_dir.mkdir(parents=True) (events_dir / "20260413.jsonl").write_text( '{"event": "ignored"}\n', encoding="utf-8" ) _set_journal_override(monkeypatch, tmp_path)
post_json = {"created": 1, "merged": 0, "skipped": 0, "staged": 0, "errors": []} mock_session = _make_session( manifest_data={"received": {}}, post_json=post_json )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=False)
calls_by_facet = { json.loads(call.kwargs["data"]["metadata"])["facets"][0][ "name" ]: call.kwargs for call in mock_session.post.call_args_list } work_metadata = json.loads(calls_by_facet["work"]["data"]["metadata"]) uploaded_paths = [ entry["path"] for entry in work_metadata["facets"][0]["files"] ] assert "events/20260413.jsonl" not in uploaded_paths
def test_response_errors_reported(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_facets
_setup_facets(tmp_path) _set_journal_override(monkeypatch, tmp_path)
post_json = { "created": 1, "merged": 0, "skipped": 0, "staged": 0, "errors": [{"facet": "work", "error": "entity merge conflict"}], } mock_session = _make_session( manifest_data={"received": {}}, post_json=post_json )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_facets("https://example.com", "test-key", dry_run=False)
output = capsys.readouterr().out assert "entity merge conflict" in output assert "error" in output.lower()
class TestExportImports: def test_manifest_delta(self, tmp_path, monkeypatch): from solstone.observe.export import export_imports
imports = _setup_imports(tmp_path) _set_journal_override(monkeypatch, tmp_path)
manifest_data = { "received": {"20260101_090000": _import_hash(imports["20260101_090000"])} } post_json = {"copied": 1, "staged": 0, "skipped": 0, "errors": []} mock_session = _make_session(manifest_data=manifest_data, post_json=post_json)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_imports("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 1 posted_data = mock_session.post.call_args.kwargs.get( "json" ) or mock_session.post.call_args[1].get("json") posted_ids = [entry["id"] for entry in posted_data["imports"]] assert posted_ids == ["20260102_100000"]
def test_dry_run(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_imports
_setup_imports(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session(manifest_data={"received": {}})
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_imports("https://example.com", "test-key", dry_run=True)
assert mock_session.post.call_count == 0 output = capsys.readouterr().out assert "2 new" in output assert "0 changed" in output
def test_idempotent(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_imports
imports = _setup_imports(tmp_path) _set_journal_override(monkeypatch, tmp_path)
manifest_data = { "received": { import_id: _import_hash(import_data) for import_id, import_data in imports.items() } } mock_session = _make_session(manifest_data=manifest_data)
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_imports("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 0 assert "up to date" in capsys.readouterr().out
def test_sync_state_excluded(self, tmp_path, monkeypatch): from solstone.observe.export import export_imports
_setup_imports(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session( manifest_data={"received": {}}, post_json={"copied": 2, "staged": 0, "skipped": 0, "errors": []}, )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_imports("https://example.com", "test-key", dry_run=False)
posted_data = mock_session.post.call_args.kwargs.get( "json" ) or mock_session.post.call_args[1].get("json") posted_ids = {entry["id"] for entry in posted_data["imports"]} assert "plaud.json" not in posted_ids
def test_source_dir_excluded(self, tmp_path, monkeypatch): from solstone.observe.export import export_imports
_setup_imports(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session( manifest_data={"received": {}}, post_json={"copied": 2, "staged": 0, "skipped": 0, "errors": []}, )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_imports("https://example.com", "test-key", dry_run=False)
posted_data = mock_session.post.call_args.kwargs.get( "json" ) or mock_session.post.call_args[1].get("json") posted_ids = {entry["id"] for entry in posted_data["imports"]} assert "abcd1234" not in posted_ids
class TestExportConfig: def test_config_export(self, tmp_path, monkeypatch): from solstone.observe.export import export_config
_setup_config(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session( manifest_data={}, post_json={"staged": True, "skipped": False, "diff_fields": 3}, )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_config("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 1
def test_dry_run(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import export_config
_setup_config(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session(manifest_data={})
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_config("https://example.com", "test-key", dry_run=True)
assert mock_session.post.call_count == 0 assert "would send snapshot" in capsys.readouterr().out
def test_idempotent(self, tmp_path, monkeypatch, capsys): from solstone.observe.export import _strip_never_transfer, export_config
config = _setup_config(tmp_path) _set_journal_override(monkeypatch, tmp_path)
stripped = _strip_never_transfer(config) content_hash = hashlib.sha256( json.dumps(stripped, sort_keys=True, ensure_ascii=False).encode() ).hexdigest() mock_session = _make_session(manifest_data={"last_hash": content_hash})
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_config("https://example.com", "test-key", dry_run=False)
assert mock_session.post.call_count == 0 assert "up to date" in capsys.readouterr().out
def test_never_transfer_stripped(self, tmp_path, monkeypatch): from solstone.observe.export import export_config
_setup_config(tmp_path) _set_journal_override(monkeypatch, tmp_path)
mock_session = _make_session( manifest_data={}, post_json={"staged": True, "skipped": False, "diff_fields": 3}, )
with patch( "solstone.observe.export.requests.Session", return_value=mock_session ): export_config("https://example.com", "test-key", dry_run=False)
posted_data = mock_session.post.call_args.kwargs.get( "json" ) or mock_session.post.call_args[1].get("json") posted_config = posted_data["config"] assert posted_config["convey"] == {"allow_network_access": False} assert "password_hash" not in posted_config["convey"] assert "secret" not in posted_config["convey"] assert posted_config["setup"] == {"completed_at": 12345} assert posted_config["env"] == {"KEY": "val"}