solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import json
7import uuid
8from pathlib import Path
9from typing import Any
10
11from solstone.think.link.ca import load_or_generate_ca
12from solstone.think.link.paths import LinkState, ca_dir, state_path
13from solstone.think.utils import get_journal, journal_is_active
14
15
16def _commit_journal_identity() -> None:
17 load_or_generate_ca(ca_dir())
18
19
20def _read_config(journal: Path) -> dict[str, Any]:
21 return json.loads((journal / "config" / "journal.json").read_text("utf-8"))
22
23
24def _assert_mark_shape(mark: dict[str, Any]) -> None:
25 assert set(mark) == {"icon1", "icon2", "words"}
26 assert {"name", "svg", "color", "rot"} <= set(mark["icon1"])
27 assert {"name", "svg", "color", "rot"} <= set(mark["icon2"])
28 assert len(mark["words"]) == 2
29
30
31def test_init_mark_returns_unlocked_candidate(convey_env_setup_pending) -> None:
32 env = convey_env_setup_pending()
33
34 response = env.client.get("/init/mark")
35
36 assert response.status_code == 200
37 payload = response.get_json()
38 assert payload["locked"] is False
39 _assert_mark_shape(payload["mark"])
40
41
42def test_init_mark_regenerate_changes_unlocked_candidate(
43 convey_env_setup_pending,
44) -> None:
45 env = convey_env_setup_pending()
46 first = env.client.get("/init/mark").get_json()["mark"]
47
48 response = env.client.post("/init/mark/regenerate")
49
50 assert response.status_code == 200
51 payload = response.get_json()
52 assert payload["locked"] is False
53 _assert_mark_shape(payload["mark"])
54 assert payload["mark"] != first
55
56
57def test_init_mark_lock_is_idempotent(convey_env_setup_pending) -> None:
58 env = convey_env_setup_pending()
59 env.client.get("/init/mark")
60
61 first = env.client.post("/init/mark/lock")
62 second = env.client.post("/init/mark/lock")
63
64 assert first.status_code == 200
65 assert second.status_code == 200
66 assert first.get_json()["locked"] is True
67 assert second.get_json()["locked"] is True
68 assert second.get_json()["mark"] == first.get_json()["mark"]
69
70
71def test_init_finalize_starts_secure_listener_after_config_write(
72 convey_env_setup_pending,
73 monkeypatch,
74) -> None:
75 env = convey_env_setup_pending()
76 _commit_journal_identity()
77 calls: list[dict[str, Any]] = []
78
79 def spy(app: Any) -> None:
80 calls.append(
81 {
82 "app": app,
83 "active": journal_is_active(get_journal()),
84 }
85 )
86
87 monkeypatch.setattr("solstone.convey.root.start_secure_listener", spy)
88
89 response = env.client.post(
90 "/init/finalize",
91 json={"name": "X"},
92 content_type="application/json",
93 )
94
95 assert response.status_code == 200
96 assert response.get_json()["success"] is True
97 assert len(calls) == 1
98 assert calls[0]["active"] is True
99 assert calls[0]["app"] is env.app
100
101
102def test_init_mark_flow_commits_only_on_lock_then_finalizes(
103 convey_env_setup_pending,
104 monkeypatch,
105) -> None:
106 env = convey_env_setup_pending()
107 calls: list[Any] = []
108
109 def spy(app: Any) -> None:
110 calls.append(app)
111
112 monkeypatch.setattr("solstone.convey.root.start_secure_listener", spy)
113
114 mark_response = env.client.get("/init/mark")
115
116 assert mark_response.status_code == 200
117 assert mark_response.get_json()["locked"] is False
118 assert not (ca_dir() / "cert.pem").exists()
119
120 regenerate_response = env.client.post("/init/mark/regenerate")
121
122 assert regenerate_response.status_code == 200
123 assert regenerate_response.get_json()["locked"] is False
124 assert not (ca_dir() / "cert.pem").exists()
125
126 lock_response = env.client.post("/init/mark/lock")
127
128 assert lock_response.status_code == 200
129 assert lock_response.get_json()["locked"] is True
130 assert (ca_dir() / "cert.pem").exists()
131 state = json.loads(state_path().read_text("utf-8"))
132 assert state.get("instance_id")
133
134 finalize_response = env.client.post(
135 "/init/finalize",
136 json={"name": "X"},
137 content_type="application/json",
138 )
139
140 assert finalize_response.status_code == 200
141 assert finalize_response.get_json()["success"] is True
142 assert len(calls) == 1
143
144
145def test_legacy_lazy_journal_stays_locked_and_preserves_ca(
146 convey_env_setup_pending,
147) -> None:
148 env = convey_env_setup_pending()
149 load_or_generate_ca(ca_dir())
150 ca_path = ca_dir()
151 cert_before = (ca_path / "cert.pem").read_bytes()
152 key_before = (ca_path / "private.pem").read_bytes()
153 legacy_id = str(uuid.uuid4())
154 LinkState(instance_id=legacy_id, home_label="legacy").save()
155 state_before = state_path().read_bytes()
156
157 mark_response = env.client.get("/init/mark")
158 regenerate_response = env.client.post("/init/mark/regenerate")
159
160 assert mark_response.status_code == 200
161 assert mark_response.get_json()["locked"] is True
162 assert regenerate_response.status_code == 400
163 assert (
164 regenerate_response.get_json()["reason_code"] == "invalid_operation_for_state"
165 )
166 assert (ca_path / "cert.pem").read_bytes() == cert_before
167 assert (ca_path / "private.pem").read_bytes() == key_before
168 assert state_path().read_bytes() == state_before
169
170
171def test_finalize_requires_locked_identity_before_config_mutation(
172 convey_env_setup_pending,
173) -> None:
174 env = convey_env_setup_pending()
175 before_config = _read_config(env.journal)
176 convey_config = env.journal / "config" / "convey.json"
177 assert not convey_config.exists()
178
179 blocked = env.client.post(
180 "/init/finalize",
181 json={"name": "Blocked", "retention_mode": "processed"},
182 content_type="application/json",
183 )
184
185 assert blocked.status_code == 400
186 assert blocked.get_json()["reason_code"] == "identity_not_locked"
187 assert _read_config(env.journal) == before_config
188 assert not convey_config.exists()
189
190 _commit_journal_identity()
191 allowed = env.client.post(
192 "/init/finalize",
193 json={"name": "Allowed"},
194 content_type="application/json",
195 )
196
197 assert allowed.status_code == 200
198 assert allowed.get_json()["success"] is True