solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import json
5from pathlib import Path
6from unittest.mock import AsyncMock, patch
7
8import pytest
9from jsonschema import Draft202012Validator
10
11from solstone.observe import describe as describe_mod
12from solstone.observe.categories import meeting as meeting_mod
13from solstone.think.batch import Batch
14
15
16def _load_schema() -> dict:
17 return json.loads(
18 (
19 Path(describe_mod.__file__).resolve().parent
20 / "categories"
21 / "meeting.schema.json"
22 ).read_text(encoding="utf-8")
23 )
24
25
26def test_meeting_schema_file_is_valid_draft_2020_12():
27 Draft202012Validator.check_schema(_load_schema())
28
29
30def test_meeting_schema_accepts_and_rejects_expected_values():
31 validator = Draft202012Validator(_load_schema())
32
33 assert validator.is_valid(
34 {
35 "platform": "zoom",
36 "participants": [
37 {"name": "Alice", "status": "active", "video": True, "box_2d": None},
38 ],
39 "screen_share": None,
40 }
41 )
42 assert validator.is_valid(
43 {
44 "platform": "teams",
45 "participants": [
46 {
47 "name": "Bob",
48 "status": "presenting",
49 "video": True,
50 "box_2d": [0, 10, 20, 30],
51 },
52 ],
53 "screen_share": {
54 "box_2d": [40, 50, 60, 70],
55 "presenter": "Bob",
56 "description": "Showing a roadmap deck.",
57 "formatted_text": "# Roadmap",
58 },
59 }
60 )
61 assert not validator.is_valid(
62 {
63 "platform": "hangouts",
64 "participants": [
65 {"name": "Alice", "status": "active", "video": True, "box_2d": None},
66 ],
67 "screen_share": None,
68 }
69 )
70 assert not validator.is_valid(
71 {
72 "platform": "zoom",
73 "participants": [
74 {"name": "Alice", "status": "talking", "video": True, "box_2d": None},
75 ],
76 "screen_share": None,
77 }
78 )
79 assert not validator.is_valid(
80 {
81 "platform": "zoom",
82 "participants": ["Alice"],
83 "screen_share": None,
84 }
85 )
86 assert not validator.is_valid(
87 {
88 "platform": "zoom",
89 "participants": [
90 {"name": "Alice", "status": "active", "video": True, "box_2d": None},
91 ],
92 "screen_share": None,
93 "extra": True,
94 }
95 )
96 assert not validator.is_valid(
97 {
98 "platform": "zoom",
99 "participants": [
100 {"status": "active", "video": True, "box_2d": None},
101 ],
102 "screen_share": None,
103 }
104 )
105 assert not validator.is_valid(
106 {
107 "platform": "zoom",
108 "participants": [
109 {"name": "Alice", "status": "active", "video": True},
110 ],
111 "screen_share": None,
112 }
113 )
114
115
116def test_discover_categories_attaches_meeting_schema():
117 expected = _load_schema()
118
119 assert describe_mod.CATEGORIES["meeting"]["json_schema"] == expected
120 assert {
121 name for name, meta in describe_mod.CATEGORIES.items() if "json_schema" in meta
122 } == {
123 name
124 for name, meta in describe_mod.CATEGORIES.items()
125 if meta["output"] == "json"
126 }
127
128
129@pytest.mark.asyncio
130@patch("solstone.think.batch.agenerate_with_result", new_callable=AsyncMock)
131async def test_meeting_extract_batch_call_passes_schema(mock_agenerate):
132 mock_agenerate.return_value = {
133 "text": (
134 '{"platform":"zoom","participants":[{"name":"Alice","status":"active",'
135 '"video":true}],"screen_share":null}'
136 ),
137 "finish_reason": "stop",
138 }
139
140 cat_meta = describe_mod.CATEGORIES["meeting"]
141 batch = Batch(max_concurrent=1)
142 req = batch.create(
143 contents="Analyze this meeting screenshot.",
144 context=cat_meta["context"],
145 json_schema=cat_meta["json_schema"],
146 )
147 batch.add(req)
148
149 results = []
150 async for completed_req in batch.drain_batch():
151 results.append(completed_req)
152
153 assert len(results) == 1
154 assert mock_agenerate.call_args.kwargs["json_schema"] == _load_schema()
155
156
157def test_meeting_formatter_skips_non_dict_participant(caplog):
158 with caplog.at_level("WARNING", logger="solstone.observe.categories.meeting"):
159 result = meeting_mod.format(
160 {
161 "platform": "zoom",
162 "participants": [
163 "Alice",
164 {"name": "Bob", "status": "active", "video": False},
165 ],
166 "screen_share": None,
167 },
168 {},
169 )
170
171 assert "🔇 Bob (active)" in result
172 assert "Alice" not in result
173 assert "skipping non-dict participant" in caplog.text