personal memory agent
0

Configure Feed

Select the types of activity you want to include in your feed.

solstone / tests / test_cogitate_read_tools.py
21 kB 633 lines
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"]