solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from solstone.observe import extract as extract_module
5from solstone.observe.extract import (
6 _apply_category_caps,
7 _fallback_select_frames,
8 select_frames_for_extraction,
9)
10
11
12def _frame(frame_id: int, category: str | None, timestamp: float | None = None) -> dict:
13 frame = {
14 "frame_id": frame_id,
15 "timestamp": float(frame_id) if timestamp is None else timestamp,
16 "analysis": {},
17 }
18 if category is not None:
19 frame["analysis"]["primary"] = category
20 return frame
21
22
23def _frames(count: int) -> list[dict]:
24 return [_frame(frame_id, "code") for frame_id in range(1, count + 1)]
25
26
27def test_apply_category_caps_semantics():
28 categorized_frames = [
29 _frame(1, "ignored"),
30 _frame(2, "ignored"),
31 _frame(3, "low_priority"),
32 _frame(4, "low_priority"),
33 _frame(5, "low_priority"),
34 _frame(6, "normal_priority"),
35 _frame(7, "high_priority"),
36 _frame(8, "unknown"),
37 _frame(9, None),
38 ]
39 selected_ids = [9, 8, 7, 6, 5, 4, 3, 2, 1]
40 config_overrides = {
41 "ignored": {"importance": "ignore"},
42 "low_priority": {"importance": "low"},
43 "normal_priority": {"importance": "normal"},
44 "high_priority": {"importance": "high"},
45 }
46
47 assert _apply_category_caps(selected_ids, categorized_frames, config_overrides) == [
48 3,
49 4,
50 6,
51 7,
52 8,
53 9,
54 ]
55
56
57def test_fallback_select_frames_is_deterministic_and_spread():
58 categorized_frames = _frames(30)
59
60 result1 = _fallback_select_frames(categorized_frames, max_extractions=5)
61 result2 = _fallback_select_frames(categorized_frames, max_extractions=5)
62
63 assert result1 == result2
64 assert len(result1) == 5
65 assert 1 in result1
66 assert 30 in result1
67
68 timestamps = {frame["frame_id"]: frame["timestamp"] for frame in categorized_frames}
69 selected_timestamps = sorted(timestamps[frame_id] for frame_id in result1)
70 adjacent_gaps = [
71 right - left
72 for left, right in zip(selected_timestamps, selected_timestamps[1:])
73 ]
74 assert min(adjacent_gaps) >= 5
75
76
77def test_fallback_select_frames_returns_all_when_under_max_and_empty():
78 categorized_frames = _frames(3)
79
80 assert _fallback_select_frames(categorized_frames, max_extractions=5) == [1, 2, 3]
81 assert _fallback_select_frames([], max_extractions=5) == []
82
83
84def test_select_frames_readds_first_frame_when_ignore_capped(monkeypatch):
85 monkeypatch.setattr(
86 extract_module,
87 "_get_category_config",
88 lambda: {"private": {"importance": "ignore"}},
89 )
90 categorized_frames = [
91 _frame(1, "private"),
92 _frame(2, "private"),
93 ]
94
95 result = select_frames_for_extraction(
96 categorized_frames, max_extractions=5, categories=None
97 )
98
99 assert result == [1]
100
101
102def test_select_frames_applies_caps_with_fallback_and_sorts(monkeypatch):
103 monkeypatch.setattr(
104 extract_module,
105 "_get_category_config",
106 lambda: {
107 "private": {"importance": "ignore"},
108 "low_priority": {"importance": "low"},
109 },
110 )
111 categorized_frames = [
112 _frame(1, "private"),
113 _frame(2, "low_priority"),
114 _frame(3, "low_priority"),
115 _frame(4, "low_priority"),
116 _frame(5, "normal_priority"),
117 _frame(6, "high_priority"),
118 _frame(7, "private"),
119 ]
120
121 result = select_frames_for_extraction(
122 categorized_frames, max_extractions=10, categories=None
123 )
124
125 assert result == [1, 2, 3, 5, 6]