solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Test audit logging for call and app actions."""
5
6import json
7from datetime import datetime
8from pathlib import Path
9
10import pytest
11
12from solstone.apps.utils import log_app_action
13from solstone.think.facets import log_call_action
14
15
16@pytest.fixture
17def test_facet(tmp_path, monkeypatch):
18 """Set up a test facet with SOLSTONE_JOURNAL."""
19 journal = tmp_path / "journal"
20 journal.mkdir()
21 facet_path = journal / "facets" / "test_facet"
22 facet_path.mkdir(parents=True)
23
24 # Create facet.json
25 facet_json = facet_path / "facet.json"
26 facet_json.write_text(
27 json.dumps({"title": "Test Facet", "description": "Test"}), encoding="utf-8"
28 )
29
30 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
31 return journal, "test_facet"
32
33
34def read_log_entries(journal_path: Path, facet: str, day: str) -> list[dict]:
35 """Read all log entries from a facet's log file."""
36 log_path = journal_path / "facets" / facet / "logs" / f"{day}.jsonl"
37 if not log_path.exists():
38 return []
39
40 entries = []
41 with open(log_path, "r", encoding="utf-8") as f:
42 for line in f:
43 if line.strip():
44 entries.append(json.loads(line))
45 return entries
46
47
48def read_journal_log_entries(journal_path: Path, day: str) -> list[dict]:
49 """Read all log entries from the journal-level log file."""
50 log_path = journal_path / "config" / "actions" / f"{day}.jsonl"
51 if not log_path.exists():
52 return []
53
54 entries = []
55 with open(log_path, "r", encoding="utf-8") as f:
56 for line in f:
57 if line.strip():
58 entries.append(json.loads(line))
59 return entries
60
61
62def test_call_action_logging_with_day(test_facet):
63 """Test that log_call_action creates an audit log entry for a specific day."""
64 journal, facet = test_facet
65 day = "20250101"
66
67 log_call_action(
68 facet=facet,
69 action="entity_detect",
70 params={"type": "Person", "name": "John Doe", "description": "Test person"},
71 day=day,
72 )
73
74 entries = read_log_entries(journal, facet, day)
75 assert len(entries) == 1
76 assert entries[0]["action"] == "entity_detect"
77 assert entries[0]["source"] == "call"
78 assert entries[0]["actor"] == "agent"
79 assert entries[0]["params"]["type"] == "Person"
80 assert entries[0]["params"]["name"] == "John Doe"
81 assert entries[0]["params"]["description"] == "Test person"
82
83
84def test_call_action_logging_defaults_today(test_facet):
85 """Test that log_call_action defaults to today when no day specified."""
86 journal, facet = test_facet
87 today = datetime.now().strftime("%Y%m%d")
88
89 log_call_action(
90 facet=facet,
91 action="entity_attach",
92 params={"type": "Company", "name": "Acme Corp", "description": "Test company"},
93 )
94
95 entries = read_log_entries(journal, facet, today)
96 assert len(entries) == 1
97 assert entries[0]["action"] == "entity_attach"
98 assert entries[0]["source"] == "call"
99 assert entries[0]["params"]["type"] == "Company"
100 assert entries[0]["params"]["name"] == "Acme Corp"
101
102
103def test_journal_level_logging(test_facet):
104 """Test that log_app_action with facet=None writes to config/actions/."""
105 journal, _ = test_facet
106 today = datetime.now().strftime("%Y%m%d")
107
108 # Log a journal-level action (no facet)
109 log_app_action(
110 app="settings",
111 facet=None,
112 action="identity_update",
113 params={"name": "Test User"},
114 )
115
116 # Check log entry was created in config/actions/
117 entries = read_journal_log_entries(journal, today)
118 assert len(entries) == 1
119 assert entries[0]["action"] == "identity_update"
120 assert entries[0]["source"] == "app"
121 assert entries[0]["actor"] == "settings"
122 assert entries[0]["params"]["name"] == "Test User"
123 # Should NOT have a facet field
124 assert "facet" not in entries[0]
125
126
127def test_journal_level_log_directory_created(test_facet):
128 """Test that config/actions/ directory is created automatically."""
129 journal, _ = test_facet
130
131 # Verify config/actions doesn't exist yet
132 actions_dir = journal / "config" / "actions"
133 assert not actions_dir.exists()
134
135 # Log a journal-level action
136 log_app_action(
137 app="observer",
138 facet=None,
139 action="observer_create",
140 params={"name": "test-observer"},
141 )
142
143 # Verify directory was created
144 assert actions_dir.exists()
145 assert actions_dir.is_dir()
146
147
148def test_facet_action_includes_facet_field(test_facet):
149 """Test that facet-scoped actions include the facet field in the entry."""
150 journal, facet = test_facet
151 today = datetime.now().strftime("%Y%m%d")
152
153 # Log a facet-scoped action
154 log_app_action(
155 app="entities",
156 facet=facet,
157 action="entity_attach",
158 params={"type": "Person", "name": "Test Entity"},
159 )
160
161 # Check that the facet field is included
162 entries = read_log_entries(journal, facet, today)
163 assert len(entries) == 1
164 assert entries[0]["facet"] == facet