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 hashlib
7import io
8import json
9from pathlib import Path
10from types import SimpleNamespace
11
12import pytest
13from flask import Flask, g
14
15import solstone.convey.state as convey_state
16import solstone.think.utils as think_utils
17from solstone.convey.reasons import (
18 IMPORT_CLIENT_ID_CONFLICT,
19 IMPORT_CONFLICT,
20 IMPORT_NOT_FOUND,
21 INVALID_OPERATION_FOR_STATE,
22 MISSING_REQUIRED_FIELD,
23)
24from solstone.think.importers.utils import (
25 read_import_metadata,
26 update_import_metadata_fields,
27 write_import_metadata,
28)
29
30import_routes = __import__("solstone.apps.import.routes", fromlist=["routes"])
31
32
33@pytest.fixture
34def journal_env(tmp_path, monkeypatch) -> Path:
35 monkeypatch.setattr(convey_state, "journal_root", str(tmp_path), raising=False)
36 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
37 think_utils._journal_path_cache = None
38 (tmp_path / "imports").mkdir()
39 return tmp_path
40
41
42@pytest.fixture(autouse=True)
43def _stable_timestamp_detection(monkeypatch):
44 monkeypatch.setattr(
45 import_routes,
46 "resolve_created_deterministic",
47 lambda *args, **kwargs: None,
48 )
49
50 def _no_model(*args, **kwargs):
51 raise AssertionError("model timestamp detection should not run")
52
53 monkeypatch.setattr(import_routes, "detect_created", _no_model)
54
55
56@pytest.fixture
57def client(journal_env):
58 app = Flask(__name__)
59
60 @app.before_request
61 def _identity() -> None:
62 g.identity = SimpleNamespace(mode="local", fingerprint=None)
63
64 app.register_blueprint(import_routes.import_bp)
65 return app.test_client()
66
67
68def _sha(content: bytes) -> str:
69 return hashlib.sha256(content).hexdigest()
70
71
72def _import_dirs(journal_root: Path) -> list[Path]:
73 return sorted(
74 path for path in (journal_root / "imports").iterdir() if path.is_dir()
75 )
76
77
78def _save_upload(
79 client,
80 *,
81 client_item_id: str,
82 content: bytes = b"sample audio",
83 filename: str = "sample.m4a",
84 content_type: str = "audio/mp4",
85 extra: dict | None = None,
86):
87 data = {
88 "client_item_id": client_item_id,
89 "deterministic_only": "true",
90 "file": (io.BytesIO(content), filename, content_type),
91 }
92 if extra:
93 data.update(extra)
94 return client.post(
95 "/app/import/api/save",
96 data=data,
97 content_type="multipart/form-data",
98 )
99
100
101def _write_manifest(
102 journal_root: Path,
103 *,
104 import_id: str,
105 source_hash: str,
106 entry_count: int = 2,
107) -> None:
108 import_dir = journal_root / "imports" / import_id
109 import_dir.mkdir(parents=True)
110 (import_dir / "manifest.json").write_text(
111 json.dumps(
112 {
113 "import_id": import_id,
114 "source_type": "audio",
115 "source_hash": source_hash,
116 "entry_count": entry_count,
117 "imported_at": "2026-01-01T12:00:00",
118 "imported_via": "cli",
119 }
120 ),
121 encoding="utf-8",
122 )
123
124
125def _write_staged_import(
126 journal_root: Path,
127 timestamp: str,
128 metadata: dict,
129) -> Path:
130 import_dir = journal_root / "imports" / timestamp
131 import_dir.mkdir(parents=True)
132 media_path = import_dir / metadata.get("original_filename", "sample.m4a")
133 media_path.write_bytes(b"sample")
134 metadata = {
135 "file_path": str(media_path),
136 "user_timestamp": timestamp,
137 "source": "audio",
138 "source_inference": "extension",
139 "client": {},
140 "facet": None,
141 "setting": None,
142 "imported_via": "web_dashboard",
143 "observer_handle": None,
144 "source_hint": None,
145 "mime_type": "audio/mp4",
146 **metadata,
147 }
148 write_import_metadata(journal_root, timestamp, metadata)
149 return media_path
150
151
152def test_import_save_audio_upload_stages_versioned_summary(client, journal_env):
153 response = _save_upload(
154 client,
155 client_item_id="ios-audio-1",
156 content=b"audio bytes",
157 extra={"facet": "work", "client": json.dumps({"device": "ios"})},
158 )
159
160 assert response.status_code == 200
161 body = response.get_json()
162 assert body["schema_version"] == 1
163 assert body["status"] == "staged"
164 assert body["replay"] is False
165 assert body["source"] == "audio"
166 assert body["client_item_id"] == "ios-audio-1"
167 assert body["recommended_action"] == "start"
168 assert body["facet"] == "work"
169 assert body["metadata"] == {
170 "original_filename": "sample.m4a",
171 "mime_type": "audio/mp4",
172 "imported_via": "web_dashboard",
173 "observer_handle": None,
174 "source_hint": None,
175 "client": {"device": "ios"},
176 }
177 assert body["diagnostics"]["source_inference"] == "extension"
178 assert "timestamp_detection_method" not in body
179 assert "dedup" not in body
180
181 metadata = read_import_metadata(journal_env, body["timestamp"])
182 assert metadata["client_item_id"] == "ios-audio-1"
183 assert metadata["source_hash"] == _sha(b"audio bytes")
184 assert metadata["source"] == "audio"
185 assert metadata["client"] == {"device": "ios"}
186
187
188def test_import_save_missing_client_item_id_returns_missing_required(client):
189 response = client.post(
190 "/app/import/api/save",
191 data={
192 "deterministic_only": "true",
193 "file": (io.BytesIO(b"audio"), "sample.m4a", "audio/mp4"),
194 },
195 content_type="multipart/form-data",
196 )
197
198 assert response.status_code == MISSING_REQUIRED_FIELD.status
199 assert response.get_json()["reason_code"] == MISSING_REQUIRED_FIELD.code
200
201
202def test_import_save_replay_same_client_and_content_does_not_stage_again(
203 client, journal_env
204):
205 first = _save_upload(
206 client,
207 client_item_id="ios-replay",
208 content=b"same bytes",
209 )
210 before_dirs = _import_dirs(journal_env)
211
212 second = _save_upload(
213 client,
214 client_item_id="ios-replay",
215 content=b"same bytes",
216 )
217
218 assert second.status_code == 200
219 body = second.get_json()
220 assert body["status"] == "staged"
221 assert body["replay"] is True
222 assert body["recommended_action"] == "start"
223 assert body["path"] == first.get_json()["path"]
224 assert _import_dirs(journal_env) == before_dirs
225
226
227def test_import_save_replay_started_item_does_not_recommend_start(client, journal_env):
228 first = _save_upload(
229 client,
230 client_item_id="ios-replay-started",
231 content=b"same started bytes",
232 )
233 timestamp = first.get_json()["timestamp"]
234 update_import_metadata_fields(
235 journal_root=journal_env,
236 timestamp=timestamp,
237 updates={"task_id": "task-started"},
238 )
239 before_dirs = _import_dirs(journal_env)
240
241 second = _save_upload(
242 client,
243 client_item_id="ios-replay-started",
244 content=b"same started bytes",
245 )
246
247 assert second.status_code == 200
248 body = second.get_json()
249 assert body["status"] == "staged"
250 assert body["replay"] is True
251 assert body["recommended_action"] == "do_not_start"
252 assert body["path"] == first.get_json()["path"]
253 assert _import_dirs(journal_env) == before_dirs
254
255
256def test_import_save_same_client_different_content_conflicts(client):
257 _save_upload(client, client_item_id="ios-conflict", content=b"one")
258
259 response = _save_upload(client, client_item_id="ios-conflict", content=b"two")
260
261 assert response.status_code == IMPORT_CLIENT_ID_CONFLICT.status
262 assert response.get_json()["reason_code"] == IMPORT_CLIENT_ID_CONFLICT.code
263
264
265def test_import_save_duplicate_imported_content_is_terminal(client, journal_env):
266 content = b"already imported"
267 _write_manifest(
268 journal_env,
269 import_id="20260101_120000",
270 source_hash=_sha(content),
271 entry_count=3,
272 )
273 before_dirs = _import_dirs(journal_env)
274
275 response = _save_upload(
276 client,
277 client_item_id="ios-imported-dup",
278 content=content,
279 )
280
281 assert response.status_code == 200
282 body = response.get_json()
283 assert body["status"] == "duplicate"
284 assert body["replay"] is False
285 assert body["recommended_action"] == "do_not_start"
286 assert body["duplicate"] == {
287 "import_id": "20260101_120000",
288 "imported_at": "2026-01-01T12:00:00",
289 "entry_count": 3,
290 "state": "imported",
291 }
292 assert _import_dirs(journal_env) == before_dirs
293 assert not any((path / "import.json").exists() for path in before_dirs)
294
295
296def test_import_save_duplicate_staged_content_is_terminal(client, journal_env):
297 content = b"already staged"
298 _write_staged_import(
299 journal_env,
300 "20260101_121500",
301 {
302 "original_filename": "existing.m4a",
303 "client_item_id": "other-client",
304 "source_hash": _sha(content),
305 },
306 )
307 before_dirs = _import_dirs(journal_env)
308
309 response = _save_upload(
310 client,
311 client_item_id="ios-staged-dup",
312 content=content,
313 )
314
315 assert response.status_code == 200
316 body = response.get_json()
317 assert body["status"] == "duplicate"
318 assert body["client_item_id"] == "ios-staged-dup"
319 assert body["recommended_action"] == "do_not_start"
320 assert body["duplicate"] == {
321 "import_id": "20260101_121500",
322 "imported_at": None,
323 "entry_count": None,
324 "state": "staged",
325 }
326 assert _import_dirs(journal_env) == before_dirs
327
328
329def test_import_meta_updates_facet_and_setting(client, journal_env):
330 media_path = _write_staged_import(
331 journal_env,
332 "20260101_130000",
333 {"original_filename": "meta.m4a", "client_item_id": "meta-client"},
334 )
335
336 response = client.post(
337 "/app/import/api/meta",
338 json={"path": str(media_path), "facet": "work", "setting": "office"},
339 )
340
341 assert response.status_code == 200
342 assert response.get_json() == {
343 "status": "ok",
344 "path": str(media_path),
345 "timestamp": "20260101_130000",
346 "updated": {"facet": "work", "setting": "office"},
347 }
348 metadata = read_import_metadata(journal_env, "20260101_130000")
349 assert metadata["facet"] == "work"
350 assert metadata["setting"] == "office"
351
352
353def test_import_meta_missing_path_returns_missing_required(client):
354 response = client.post("/app/import/api/meta", json={})
355
356 assert response.status_code == MISSING_REQUIRED_FIELD.status
357 assert response.get_json()["reason_code"] == MISSING_REQUIRED_FIELD.code
358
359
360def test_import_facet_route_removed_no_alias(client):
361 rules = list(client.application.url_map.iter_rules())
362
363 assert not any(rule.rule == "/app/import/api/facet" for rule in rules)
364 assert any(
365 rule.rule == "/app/import/api/meta" and "POST" in rule.methods for rule in rules
366 )
367
368
369def test_import_meta_missing_item_returns_import_not_found(client, journal_env):
370 missing_path = journal_env / "imports" / "20260101_130001" / "sample.m4a"
371
372 response = client.post(
373 "/app/import/api/meta",
374 json={"path": str(missing_path), "facet": "work"},
375 )
376
377 assert response.status_code == IMPORT_NOT_FOUND.status
378 assert response.get_json()["reason_code"] == IMPORT_NOT_FOUND.code
379
380
381def test_import_meta_started_item_returns_invalid_operation(client, journal_env):
382 media_path = _write_staged_import(
383 journal_env,
384 "20260101_130002",
385 {
386 "original_filename": "started.m4a",
387 "client_item_id": "started-client",
388 "task_id": "task-1",
389 },
390 )
391
392 response = client.post(
393 "/app/import/api/meta",
394 json={"path": str(media_path), "facet": "work"},
395 )
396
397 assert response.status_code == INVALID_OPERATION_FOR_STATE.status
398 assert response.get_json()["reason_code"] == INVALID_OPERATION_FOR_STATE.code
399
400
401def test_import_start_moves_dir_uses_saved_metadata_and_omits_generic_source(
402 client, journal_env, monkeypatch
403):
404 emitted: list[dict[str, object]] = []
405 monkeypatch.setattr(
406 import_routes,
407 "emit",
408 lambda tract, event, **payload: emitted.append(
409 {"tract": tract, "event": event, **payload}
410 ),
411 )
412 old_ts = "20260101_120000"
413 new_ts = "20260101_121500"
414 media_path = _write_staged_import(
415 journal_env,
416 old_ts,
417 {
418 "original_filename": "sample.m4a",
419 "client_item_id": "start-client",
420 "facet": "work",
421 "setting": "office",
422 "source_hint": None,
423 },
424 )
425
426 response = client.post(
427 "/app/import/api/start",
428 json={
429 "path": str(media_path),
430 "timestamp": new_ts,
431 "source": "recording",
432 "facet": "ignored",
433 "force": True,
434 },
435 )
436
437 assert response.status_code == 200
438 new_dir = journal_env / "imports" / new_ts
439 assert new_dir.exists()
440 assert not (journal_env / "imports" / old_ts).exists()
441 metadata = read_import_metadata(journal_env, new_ts)
442 assert metadata["file_path"] == str(new_dir / media_path.name)
443 assert emitted == [
444 {
445 "tract": "supervisor",
446 "event": "request",
447 "ref": response.get_json()["task_id"],
448 "cmd": [
449 "journal",
450 "importer",
451 str(new_dir / media_path.name),
452 new_ts,
453 "--facet",
454 "work",
455 "--setting",
456 "office",
457 "--force",
458 ],
459 }
460 ]
461
462
463def test_import_start_forwards_only_saved_source_hint(client, journal_env, monkeypatch):
464 emitted: list[dict[str, object]] = []
465 monkeypatch.setattr(
466 import_routes,
467 "emit",
468 lambda tract, event, **payload: emitted.append(
469 {"tract": tract, "event": event, **payload}
470 ),
471 )
472 ts = "20260101_122000"
473 media_path = _write_staged_import(
474 journal_env,
475 ts,
476 {
477 "original_filename": "vault",
478 "client_item_id": "source-hint-client",
479 "source_hint": "obsidian",
480 },
481 )
482
483 response = client.post(
484 "/app/import/api/start",
485 json={"path": str(media_path), "timestamp": ts, "source": "quick"},
486 )
487
488 assert response.status_code == 200
489 assert emitted[0]["cmd"] == [
490 "journal",
491 "importer",
492 str(media_path),
493 ts,
494 "--source",
495 "obsidian",
496 ]
497
498
499def test_import_start_refuses_terminal_duplicate_even_with_force(
500 client, journal_env, monkeypatch
501):
502 emitted: list[dict[str, object]] = []
503 monkeypatch.setattr(
504 import_routes,
505 "emit",
506 lambda tract, event, **payload: emitted.append(
507 {"tract": tract, "event": event, **payload}
508 ),
509 )
510 content = b"terminal"
511 source_hash = _sha(content)
512 ts = "20260101_123000"
513 media_path = _write_staged_import(
514 journal_env,
515 ts,
516 {
517 "original_filename": "terminal.m4a",
518 "client_item_id": "terminal-client",
519 "source_hash": source_hash,
520 },
521 )
522 _write_manifest(
523 journal_env,
524 import_id="20260101_124500",
525 source_hash=source_hash,
526 )
527
528 response = client.post(
529 "/app/import/api/start",
530 json={"path": str(media_path), "timestamp": ts, "force": True},
531 )
532
533 assert response.status_code == INVALID_OPERATION_FOR_STATE.status
534 assert response.get_json()["reason_code"] == INVALID_OPERATION_FOR_STATE.code
535 assert emitted == []
536
537
538def test_import_start_missing_source_returns_import_not_found(client, journal_env):
539 old_ts = "20260101_120000"
540 new_ts = "20260101_121500"
541 missing_path = journal_env / "imports" / old_ts / "sample.m4a"
542
543 response = client.post(
544 "/app/import/api/start",
545 json={"path": str(missing_path), "timestamp": new_ts},
546 )
547
548 assert response.status_code == IMPORT_NOT_FOUND.status
549 body = response.get_json()
550 assert body["reason_code"] == IMPORT_NOT_FOUND.code
551
552
553def test_import_start_target_exists_returns_import_conflict(client, journal_env):
554 old_ts = "20260101_120000"
555 new_ts = "20260101_121500"
556 old_dir = journal_env / "imports" / old_ts
557 new_dir = journal_env / "imports" / new_ts
558 old_dir.mkdir()
559 new_dir.mkdir()
560 media_path = old_dir / "sample.m4a"
561 media_path.write_bytes(b"sample")
562 write_import_metadata(
563 journal_env,
564 old_ts,
565 {
566 "file_path": str(media_path),
567 "user_timestamp": old_ts,
568 "client_item_id": "conflict-client",
569 },
570 )
571
572 response = client.post(
573 "/app/import/api/start",
574 json={"path": str(media_path), "timestamp": new_ts},
575 )
576
577 assert response.status_code == IMPORT_CONFLICT.status
578 body = response.get_json()
579 assert body["reason_code"] == IMPORT_CONFLICT.code