solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import json
5import logging
6import subprocess
7from pathlib import Path
8
9import pytest
10
11from solstone.observe import detect
12from solstone.think.providers.rfdetr_install import (
13 ENGINE_REF,
14 MODEL_NAME,
15 RfdetrPaths,
16)
17
18
19@pytest.fixture(autouse=True)
20def _reset_detect_state():
21 detect._disabled = False
22 yield
23 detect._disabled = False
24
25
26def _canned_cli_json() -> dict:
27 return {
28 "image": {"width": 100, "height": 50},
29 "detections": [
30 {
31 "class_id": 1,
32 "class_name": "cup",
33 "score": 0.72,
34 "bbox": [1, 2, 3, 4],
35 }
36 ],
37 }
38
39
40def test_detect_objects_missing_provider_latches_without_reattempt(monkeypatch, caplog):
41 calls = 0
42
43 def fake_paths():
44 nonlocal calls
45 calls += 1
46 return RfdetrPaths(status="not_installed")
47
48 monkeypatch.setattr(detect, "rfdetr_paths", fake_paths)
49 caplog.set_level(logging.WARNING, logger=detect.LOG.name)
50
51 assert detect.detect_objects(b"x") is None
52 assert detect.detect_objects(b"x") is None
53
54 warnings = [
55 record
56 for record in caplog.records
57 if record.name == detect.LOG.name and record.levelno == logging.WARNING
58 ]
59 assert detect._disabled is True
60 assert len(warnings) == 1
61 assert warnings[0].getMessage() == (
62 "object detection disabled: rf-detr provider not_installed"
63 )
64 assert calls == 1
65
66
67def test_detect_objects_returns_parsed_cli_json(monkeypatch, tmp_path):
68 canned = _canned_cli_json()
69 argv_seen = []
70
71 monkeypatch.setattr(
72 detect,
73 "rfdetr_paths",
74 lambda: RfdetrPaths(
75 status="installed",
76 binary_path=tmp_path / "rfdetr-cli",
77 model_path=tmp_path / "model.gguf",
78 ),
79 )
80
81 def fake_run(argv, **kwargs):
82 argv_seen.append((argv, kwargs))
83 output_path = Path(argv[argv.index("--output") + 1])
84 output_path.write_text(json.dumps(canned), encoding="utf-8")
85 return subprocess.CompletedProcess(argv, 0)
86
87 monkeypatch.setattr(detect.subprocess, "run", fake_run)
88
89 assert detect.detect_objects(b"png-bytes") == canned
90 assert len(argv_seen) == 1
91 argv, kwargs = argv_seen[0]
92 assert argv[1] == "detect"
93 assert argv[argv.index("--threshold") + 1] == str(detect.THRESHOLD)
94 assert kwargs["timeout"] == detect._TIMEOUT_S
95 assert kwargs["capture_output"] is True
96 assert kwargs["check"] is True
97
98
99def test_detect_objects_subprocess_failure_latches(monkeypatch, tmp_path):
100 monkeypatch.setattr(
101 detect,
102 "rfdetr_paths",
103 lambda: RfdetrPaths(
104 status="installed",
105 binary_path=tmp_path / "rfdetr-cli",
106 model_path=tmp_path / "model.gguf",
107 ),
108 )
109
110 def fake_run(argv, **_kwargs):
111 raise subprocess.CalledProcessError(1, argv, stderr=b"bad")
112
113 monkeypatch.setattr(detect.subprocess, "run", fake_run)
114
115 assert detect.detect_objects(b"png-bytes") is None
116 assert detect._disabled is True
117
118
119def test_detections_block_renames_and_preserves_provenance():
120 canned = _canned_cli_json()
121
122 assert detect.detections_block(canned, source="screen", gate="primary:media") == {
123 "engine": detect.ENGINE_NAME,
124 "engine_ref": ENGINE_REF,
125 "model": MODEL_NAME,
126 "threshold": detect.THRESHOLD,
127 "source": "screen",
128 "gate": "primary:media",
129 "image": canned["image"],
130 "objects": canned["detections"],
131 }
132
133
134def test_screen_gate_prefers_primary_gate():
135 assert detect.screen_gate({"primary": "media", "secondary": "none"}) == (
136 "primary:media"
137 )
138 assert detect.screen_gate({"primary": "code", "secondary": "social"}) == (
139 "secondary:social"
140 )
141 assert detect.screen_gate({"primary": "media", "secondary": "social"}) == (
142 "primary:media"
143 )
144 assert detect.screen_gate({"primary": "code", "secondary": "terminal"}) is None
145
146
147def test_qualified_objects_filters_at_read_time_only():
148 tv = {"class_name": "tv", "score": 0.7}
149 weak_person = {"class_name": "person", "score": 0.39}
150 person = {"class_name": "person", "score": 0.41}
151 cup = {"class_name": "cup", "score": 0.41}
152 sandwich = {"class_name": "sandwich", "score": 0.9}
153
154 assert detect.qualified_objects({"source": "screen", "objects": [tv]}) == []
155 assert detect.qualified_objects({"source": "still", "objects": [tv]}) == [tv]
156 assert detect.qualified_objects(
157 {"source": "still", "objects": [weak_person, person, cup, sandwich]}
158 ) == [person, cup]