personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import ast
7import os
8import stat
9from pathlib import Path
10from types import SimpleNamespace
11
12import pytest
13
14from solstone.think import cogitate_read_tools as crt
15
16
17@pytest.fixture
18def read_tools_journal(tmp_path):
19 journal = tmp_path / "journal"
20 journal.mkdir()
21
22 def write(rel: str, content: str) -> Path:
23 path = journal / rel
24 path.parent.mkdir(parents=True, exist_ok=True)
25 path.write_text(content, encoding="utf-8")
26 return path
27
28 write("chronicle/20260608/session/090000_300/evidence.md", "x evidence\n")
29 write("chronicle/20260608/session/090000_300/nested/deep.txt", "deep x\n")
30 write(
31 "chronicle/20260608/session/090000_300/talents/sense.json",
32 '{"activity_summary":"rohan"}\n',
33 )
34 write("chronicle/20260608/foo", "date redirected\n")
35 write("chronicle/20260608/.git/config", "git-secret x\n")
36 write("chronicle/20260608/.cache/x", "cache-secret x\n")
37 write("chronicle/20260608/node_modules/pkg/index.js", "node-secret x\n")
38 write(
39 "chronicle/20260608/.venv/lib/python3.12/site-packages/pkg.py",
40 "venv-secret x\n",
41 )
42 write("chronicle/20260608/id_rsa", "credential-secret x\n")
43 write("chronicle/20260608/private.pem", "credential-secret x\n")
44 write("chronicle/20260608/.env", "credential-secret x\n")
45 write("notes/nested/a.txt", "note x\n")
46 write("entities/rohan/entity.json", '{"name":"Rohan"}\n')
47 write("facets/work/facet.json", '{"facet":"work"}\n')
48 write("facets/work/activities/20260608.jsonl", '{"activity":"standup"}\n')
49 write(".agents/skills/journal/SKILL.md", "# Journal Skill\n")
50 write(".git/config", "git-secret x\n")
51 write(".cache/x", "cache-secret x\n")
52 write("node_modules/pkg/index.js", "node-secret x\n")
53 write(".venv/lib/python3.12/site-packages/pkg.py", "venv-secret x\n")
54 write("id_rsa", "credential-secret x\n")
55 write("private.pem", "credential-secret x\n")
56 write(".env", "credential-secret x\n")
57 write("binary.bin", "placeholder\n").write_bytes(b"abc\x00def")
58 write("real/inside.txt", "alias target x\n")
59
60 fifo = journal / "fifo"
61 os.mkfifo(fifo)
62
63 os.symlink(journal / "missing-target", journal / "dangling")
64
65 denied = write("denied.txt", "permission-secret\n")
66 os.chmod(denied, 0)
67
68 outside = tmp_path / "outside"
69 outside.mkdir()
70 (outside / "secret.txt").write_text("outside x\n", encoding="utf-8")
71 os.symlink(outside, journal / "escape")
72
73 os.symlink(journal / ".git", journal / "logs")
74 os.symlink(journal / "real", journal / "alias")
75 os.symlink(".git", journal / "chronicle" / "20260608" / "logs")
76 os.symlink("..", journal / "chronicle" / "loop")
77
78 env = SimpleNamespace(journal=journal, denied=denied)
79 try:
80 yield env
81 finally:
82 if denied.exists():
83 os.chmod(denied, 0o600)
84
85
86def _payload_paths(result: crt.ReadResult) -> list[str]:
87 payload = result.payload
88 if not isinstance(payload, list):
89 return []
90 paths: list[str] = []
91 for item in payload:
92 if isinstance(item, str):
93 paths.append(item)
94 elif isinstance(item, crt.Entry | crt.GrepMatch):
95 paths.append(item.path)
96 return paths
97
98
99def _assert_refusal(result: crt.ReadResult, refusal: str) -> None:
100 assert result.ok is False
101 assert result.refusal == refusal
102
103
104def _assert_broad_root(result: crt.ReadResult) -> None:
105 _assert_refusal(result, crt.REFUSAL_BROAD_ROOT)
106
107
108def test_module_docstring_declares_security_contract():
109 doc = (crt.__doc__ or "").lower()
110
111 assert "read-only" in doc
112 assert "journal root" in doc
113 assert "denylist" in doc
114
115
116def test_ac01_non_root_paths_use_contained_path_for_escape(read_tools_journal):
117 journal = read_tools_journal.journal
118
119 results = [
120 crt.read_file(journal, "escape/secret.txt"),
121 crt.list_directory(journal, "escape"),
122 crt.glob(journal, "*", root="escape"),
123 crt.grep_search(journal, "outside", path="escape"),
124 ]
125
126 assert all(result.ok is False for result in results)
127 assert {result.refusal for result in results} == {crt.REFUSAL_PATH_ESCAPE}
128
129
130def test_ac02_journal_root_recursive_scans_refuse_broad(read_tools_journal):
131 journal = read_tools_journal.journal
132
133 listed = crt.list_directory(journal)
134 listed_recursive = crt.list_directory(journal, recursive=True)
135 globbed = crt.glob(journal, "*")
136 grepped = crt.grep_search(journal, "x")
137
138 assert listed.ok is True
139 for result in [listed_recursive, globbed, grepped]:
140 assert result.ok is False
141 assert result.refusal == crt.REFUSAL_BROAD_ROOT
142
143
144def test_ac03_read_file_date_prefix_redirects_to_chronicle(read_tools_journal):
145 result = crt.read_file(read_tools_journal.journal, "20260608/foo")
146
147 assert result.ok is True
148 assert result.payload == "date redirected"
149
150
151def test_ac04_traversal_paths_are_refused_by_all_tools(read_tools_journal):
152 journal = read_tools_journal.journal
153
154 results = [
155 crt.read_file(journal, "../outside"),
156 crt.list_directory(journal, "../outside"),
157 crt.glob(journal, "*", root="../outside"),
158 crt.grep_search(journal, "x", path="../outside"),
159 ]
160
161 assert all(result.ok is False for result in results)
162 assert {result.refusal for result in results} == {crt.REFUSAL_BAD_PATH}
163
164
165def test_ac05_symlink_escape_explicit_target_refused_by_all_tools(read_tools_journal):
166 journal = read_tools_journal.journal
167
168 results = [
169 crt.read_file(journal, "escape/secret.txt"),
170 crt.list_directory(journal, "escape"),
171 crt.glob(journal, "*", root="escape"),
172 crt.grep_search(journal, "outside", path="escape"),
173 ]
174
175 assert all(result.ok is False for result in results)
176 assert {result.refusal for result in results} == {crt.REFUSAL_PATH_ESCAPE}
177
178
179def test_ac06_logs_symlink_to_git_is_pruned_from_traversal(read_tools_journal):
180 journal = read_tools_journal.journal
181
182 listed = crt.list_directory(
183 journal,
184 "chronicle/20260608",
185 recursive=True,
186 include_hidden=True,
187 )
188 globbed = crt.glob(
189 journal,
190 "*",
191 root="chronicle/20260608",
192 include_hidden=True,
193 )
194 grepped = crt.grep_search(
195 journal,
196 "git-secret",
197 path="chronicle/20260608",
198 include_hidden=True,
199 )
200
201 assert all("chronicle/20260608/logs" not in path for path in _payload_paths(listed))
202 assert all(
203 "chronicle/20260608/logs" not in path for path in _payload_paths(globbed)
204 )
205 assert all(
206 "chronicle/20260608/logs" not in path for path in _payload_paths(grepped)
207 )
208 assert all(
209 "chronicle/20260608/.git/config" not in path for path in _payload_paths(listed)
210 )
211 assert "chronicle/20260608/.git/config" not in _payload_paths(globbed)
212 assert grepped.payload == []
213
214
215def test_ac07_component_denylist_loud_for_read_silent_for_traversal(
216 read_tools_journal,
217):
218 journal = read_tools_journal.journal
219 denied = [
220 ".git/config",
221 ".cache/x",
222 "node_modules/pkg/index.js",
223 ".venv/lib/python3.12/site-packages/pkg.py",
224 ]
225
226 for rel in denied:
227 result = crt.read_file(journal, rel)
228 assert result.ok is False
229 assert result.refusal == crt.REFUSAL_DENIED_COMPONENT
230
231 listed_root = crt.list_directory(journal)
232 assert all(
233 rel not in _payload_paths(listed_root)
234 for rel in [".git", ".cache", "node_modules", ".venv"]
235 )
236
237 listed = crt.list_directory(
238 journal,
239 "chronicle/20260608",
240 recursive=True,
241 include_hidden=True,
242 )
243 globbed = crt.glob(
244 journal,
245 "*",
246 root="chronicle/20260608",
247 include_hidden=True,
248 )
249 grepped = crt.grep_search(
250 journal,
251 "secret",
252 path="chronicle/20260608",
253 include_hidden=True,
254 )
255
256 bounded_denied = [f"chronicle/20260608/{rel}" for rel in denied]
257 for rel in bounded_denied:
258 assert rel not in _payload_paths(listed)
259 assert rel not in _payload_paths(globbed)
260 assert rel not in _payload_paths(grepped)
261
262
263def test_ac08_credential_denylist_loud_for_read_excluded_from_search(
264 read_tools_journal,
265):
266 journal = read_tools_journal.journal
267 denied = ["id_rsa", "private.pem", ".env"]
268
269 for rel in denied:
270 result = crt.read_file(journal, rel)
271 assert result.ok is False
272 assert result.refusal == crt.REFUSAL_CREDENTIAL_FILE
273
274 globbed = crt.glob(
275 journal,
276 "*",
277 root="chronicle/20260608",
278 include_hidden=True,
279 )
280 grepped = crt.grep_search(
281 journal,
282 "credential-secret",
283 path="chronicle/20260608",
284 include_hidden=True,
285 )
286 for rel in [f"chronicle/20260608/{item}" for item in denied]:
287 assert rel not in _payload_paths(globbed)
288 assert rel not in _payload_paths(grepped)
289
290
291def test_broad_glob_refuses_root_chronicle_and_facets(read_tools_journal):
292 journal = read_tools_journal.journal
293 cases = [
294 ("*", "."),
295 ("*", ""),
296 ("*", "./"),
297 ("chronicle/x", "."),
298 ("20260608/**", "chronicle"),
299 ("*/x", "facets"),
300 ]
301
302 for pattern, root in cases:
303 _assert_broad_root(crt.glob(journal, pattern, root=root))
304
305
306def test_broad_grep_directory_refuses_even_with_file_glob(read_tools_journal):
307 journal = read_tools_journal.journal
308 cases = [
309 (".", "entities.txt"),
310 (".", "*.py"),
311 ("chronicle", None),
312 ("facets", None),
313 ]
314
315 for path, file_glob in cases:
316 _assert_broad_root(
317 crt.grep_search(journal, "x", path=path, file_glob=file_glob)
318 )
319
320
321def test_broad_recursive_list_directory_refuses_root_chronicle_facets(
322 read_tools_journal,
323):
324 journal = read_tools_journal.journal
325
326 for path in [".", "chronicle", "facets"]:
327 _assert_broad_root(crt.list_directory(journal, path, recursive=True))
328
329
330def test_broad_refusal_happens_before_walk_allowed(read_tools_journal, monkeypatch):
331 journal = read_tools_journal.journal
332 called = False
333
334 def fail_walk(*_args, **_kwargs):
335 nonlocal called
336 called = True
337 raise AssertionError("_walk_allowed should not be called")
338
339 monkeypatch.setattr(crt, "_walk_allowed", fail_walk)
340
341 _assert_broad_root(crt.glob(journal, "*"))
342 _assert_broad_root(crt.list_directory(journal, recursive=True))
343 _assert_broad_root(crt.grep_search(journal, "x"))
344 assert called is False
345
346
347def test_broad_root_normalization_and_symlink_back_to_root(read_tools_journal):
348 journal = read_tools_journal.journal
349
350 for root in ["", ".", "./"]:
351 _assert_broad_root(crt.glob(journal, "*", root=root))
352
353 _assert_broad_root(crt.glob(journal, "*", root="chronicle/loop"))
354 _assert_broad_root(crt.list_directory(journal, "chronicle/loop", recursive=True))
355
356
357def test_broad_specific_pattern_never_rescues_root(read_tools_journal):
358 journal = read_tools_journal.journal
359
360 _assert_broad_root(crt.glob(journal, "entities/rohan/entity.json", root="."))
361 _assert_broad_root(
362 crt.grep_search(
363 journal,
364 "Rohan",
365 path=".",
366 file_glob="entities/rohan/entity.json",
367 )
368 )
369
370
371def test_allowed_bounded_recursive_roots_succeed(read_tools_journal):
372 journal = read_tools_journal.journal
373
374 root_list = crt.list_directory(journal)
375 day_glob = crt.glob(journal, "*090000*", root="chronicle/20260608")
376 sense_json = crt.glob(
377 journal,
378 "*/talents/sense.json",
379 root="chronicle/20260608/session/090000_300",
380 )
381 facet_glob = crt.glob(journal, "*", root="facets/work")
382 facet_list = crt.list_directory(journal, "facets/work", recursive=True)
383 day_grep = crt.grep_search(journal, "evidence", path="chronicle/20260608")
384 file_grep = crt.grep_search(
385 journal,
386 "Rohan",
387 path="entities/rohan/entity.json",
388 )
389
390 assert root_list.ok is True
391 assert _payload_paths(root_list)
392 assert day_glob.ok is True
393 assert any("090000_300" in path for path in _payload_paths(day_glob))
394 assert sense_json.ok is True
395 assert _payload_paths(sense_json) == [
396 "chronicle/20260608/session/090000_300/talents/sense.json"
397 ]
398 assert facet_glob.ok is True
399 assert "facets/work/facet.json" in _payload_paths(facet_glob)
400 assert facet_list.ok is True
401 assert "facets/work/activities/20260608.jsonl" in _payload_paths(facet_list)
402 assert day_grep.ok is True
403 assert [match.path for match in day_grep.payload] == [
404 "chronicle/20260608/session/090000_300/evidence.md"
405 ]
406 assert file_grep.ok is True
407 assert [match.path for match in file_grep.payload] == ["entities/rohan/entity.json"]
408
409
410def test_entities_top_level_recursive_exemption_succeeds(read_tools_journal):
411 journal = read_tools_journal.journal
412
413 globbed = crt.glob(journal, "*rohan*", root="entities")
414 listed = crt.list_directory(journal, "entities", recursive=True)
415 grepped = crt.grep_search(
416 journal,
417 "Rohan",
418 path="entities",
419 file_glob="*/entity.json",
420 )
421
422 assert globbed.ok is True
423 assert "entities/rohan/entity.json" in _payload_paths(globbed)
424 assert listed.ok is True
425 assert "entities/rohan/entity.json" in _payload_paths(listed)
426 assert grepped.ok is True
427 assert [match.path for match in grepped.payload] == ["entities/rohan/entity.json"]
428
429
430def test_security_refusals_win_before_broad_guard(read_tools_journal):
431 journal = read_tools_journal.journal
432
433 _assert_refusal(
434 crt.glob(journal, "*", root="escape"),
435 crt.REFUSAL_PATH_ESCAPE,
436 )
437 # Denial runs before the broad guard; denied components are not broad roots.
438 _assert_refusal(
439 crt.list_directory(journal, ".git", recursive=True),
440 crt.REFUSAL_DENIED_COMPONENT,
441 )
442 _assert_refusal(
443 crt.glob(journal, "*", root=".git"),
444 crt.REFUSAL_DENIED_COMPONENT,
445 )
446
447
448def test_ac09_hidden_agents_skill_readable_by_explicit_read(read_tools_journal):
449 result = crt.read_file(
450 read_tools_journal.journal, ".agents/skills/journal/SKILL.md"
451 )
452
453 assert result.ok is True
454 assert result.payload == "# Journal Skill"
455
456
457def test_ac10_read_file_stable_refusals_do_not_raise(read_tools_journal):
458 journal = read_tools_journal.journal
459
460 cases = {
461 "real": crt.REFUSAL_NOT_FILE,
462 "binary.bin": crt.REFUSAL_BINARY,
463 "fifo": crt.REFUSAL_SPECIAL_FILE,
464 "dangling": crt.REFUSAL_MISSING,
465 "denied.txt": crt.REFUSAL_PERMISSION_DENIED,
466 }
467
468 for rel, refusal in cases.items():
469 result = crt.read_file(journal, rel)
470 assert result.ok is False
471 assert result.refusal == refusal
472
473
474def test_ac11_caps_and_budget_truncate_or_exhaust(read_tools_journal):
475 journal = read_tools_journal.journal
476 capdir = journal / "capdir"
477 capdir.mkdir()
478 for idx in range(5):
479 (capdir / f"item{idx}.txt").write_text(f"needle {idx}\n", encoding="utf-8")
480 (journal / "many-lines.txt").write_text(
481 "\n".join(f"line {idx}" for idx in range(10)),
482 encoding="utf-8",
483 )
484 (journal / "big-bytes.txt").write_text("x" * 40, encoding="utf-8")
485 (journal / "many-grep.txt").write_text("needle\n" * 10, encoding="utf-8")
486
487 line_read = crt.read_file(journal, "many-lines.txt", max_lines=3)
488 byte_read = crt.read_file(journal, "big-bytes.txt", max_bytes=5)
489 listed = crt.list_directory(journal, "capdir", max_entries=2)
490 globbed = crt.glob(journal, "*", root="capdir", max_matches=2)
491 grepped = crt.grep_search(journal, "needle", path="many-grep.txt", max_matches=2)
492
493 assert line_read.truncated is True
494 assert line_read.notice == crt.NOTICE_READ_FILE_TRUNCATED
495 assert byte_read.truncated is True
496 assert byte_read.notice == crt.NOTICE_READ_FILE_TRUNCATED
497 assert listed.truncated is True
498 assert listed.notice == crt.NOTICE_LIST_DIRECTORY_TRUNCATED
499 assert globbed.truncated is True
500 assert globbed.notice == crt.NOTICE_GLOB_TRUNCATED
501 assert grepped.truncated is True
502 assert grepped.notice == crt.NOTICE_GREP_TRUNCATED
503
504 budget = crt.ReadBudget(cap=2)
505 assert crt.read_file(journal, "notes/nested/a.txt", budget=budget).ok is True
506 assert crt.list_directory(journal, budget=budget).ok is True
507 exhausted = crt.glob(journal, "*", budget=budget)
508 assert exhausted.ok is False
509 assert exhausted.refusal == crt.REFUSAL_BUDGET_EXHAUSTED
510
511
512def test_ac12_none_budget_never_exhausts(read_tools_journal):
513 journal = read_tools_journal.journal
514
515 for _idx in range(crt.DEFAULT_READ_CALL_BUDGET + 5):
516 result = crt.read_file(journal, "notes/nested/a.txt", budget=None)
517 assert result.ok is True
518 assert result.refusal != crt.REFUSAL_BUDGET_EXHAUSTED
519
520
521def test_ac13_grep_literal_default_and_regex_opt_in(read_tools_journal):
522 journal = read_tools_journal.journal
523 (journal / "regex.txt").write_text("a.b\nacb\n[x]\nx\n", encoding="utf-8")
524
525 literal_dot = crt.grep_search(journal, "a.b", path="regex.txt")
526 regex_dot = crt.grep_search(journal, "a.b", path="regex.txt", regex=True)
527 literal_class = crt.grep_search(journal, "[x]", path="regex.txt")
528 regex_class = crt.grep_search(journal, "[x]", path="regex.txt", regex=True)
529
530 assert [match.line for match in literal_dot.payload] == ["a.b"]
531 assert [match.line for match in regex_dot.payload] == ["a.b", "acb"]
532 assert [match.line for match in literal_class.payload] == ["[x]"]
533 assert [match.line for match in regex_class.payload] == ["[x]", "x"]
534
535
536def test_invalid_regex_returns_pattern_refusal(read_tools_journal):
537 result = crt.grep_search(read_tools_journal.journal, "(", regex=True)
538
539 assert result.ok is False
540 assert result.refusal == crt.REFUSAL_BAD_PATTERN
541
542
543def test_utf8_multibyte_straddling_byte_cap_is_not_binary(read_tools_journal):
544 journal = read_tools_journal.journal
545 (journal / "unicode.txt").write_bytes("abc é needle\n".encode("utf-8"))
546
547 read_result = crt.read_file(journal, "unicode.txt", max_bytes=5)
548 grep_result = crt.grep_search(
549 journal,
550 "abc",
551 path="unicode.txt",
552 max_bytes_per_file=5,
553 )
554
555 assert read_result.ok is True
556 assert read_result.refusal != crt.REFUSAL_BINARY
557 assert read_result.truncated is True
558 assert read_result.payload == "abc "
559 assert grep_result.ok is True
560 assert grep_result.truncated is True
561 assert [match.line for match in grep_result.payload] == ["abc "]
562
563
564def _journal_snapshot(
565 journal: Path,
566) -> tuple[set[tuple[str, int, int, str]], dict[str, bytes]]:
567 structural: set[tuple[str, int, int, str]] = set()
568 contents: dict[str, bytes] = {}
569 for path in sorted(journal.rglob("*")):
570 rel = path.relative_to(journal).as_posix()
571 st = os.lstat(path)
572 kind = stat.S_IFMT(st.st_mode)
573 link_target = os.readlink(path) if stat.S_ISLNK(st.st_mode) else ""
574 structural.add((rel, kind, stat.S_IMODE(st.st_mode), link_target))
575 if (
576 stat.S_ISREG(kind)
577 and not stat.S_ISLNK(st.st_mode)
578 and stat.S_IMODE(st.st_mode) != 0
579 ):
580 contents[rel] = path.read_bytes()
581 return structural, contents
582
583
584def test_ac14_reads_do_not_mutate_and_imports_stay_read_only(read_tools_journal):
585 journal = read_tools_journal.journal
586 before = _journal_snapshot(journal)
587
588 crt.read_file(journal, "notes/nested/a.txt")
589 crt.read_file(journal, "binary.bin")
590 crt.list_directory(
591 journal,
592 "chronicle/20260608",
593 recursive=True,
594 include_hidden=True,
595 )
596 crt.glob(journal, "*", root="chronicle/20260608", include_hidden=True)
597 crt.grep_search(journal, "x", path="chronicle/20260608", include_hidden=True)
598
599 assert _journal_snapshot(journal) == before
600
601 source = Path("solstone/think/cogitate_read_tools.py").read_text(encoding="utf-8")
602 tree = ast.parse(source)
603 banned = {
604 "atomic_replace",
605 "write_json",
606 "write_jsonl",
607 "write_text",
608 "append_jsonl",
609 "append_text",
610 "install_file",
611 "hold_lock",
612 "save_npz",
613 "update_npz",
614 }
615 imported_names: set[str] = set()
616 imported_modules: set[str] = set()
617 for node in ast.walk(tree):
618 if isinstance(node, ast.ImportFrom):
619 imported_modules.add(node.module or "")
620 imported_names.update(alias.name for alias in node.names)
621 elif isinstance(node, ast.Import):
622 imported_modules.update(alias.name for alias in node.names)
623
624 assert imported_names.isdisjoint(banned)
625 assert "solstone.think.providers.openhands" not in imported_modules
626 assert "openhands" not in imported_modules
627
628
629def test_alias_symlink_canonicalizes_to_target_path(read_tools_journal):
630 result = crt.list_directory(read_tools_journal.journal, "alias")
631
632 assert result.ok is True
633 assert _payload_paths(result) == ["real/inside.txt"]