gabriel / musehub public
test_derived_agent_provisioner.py python
385 lines 13.6 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago
1 """TDD tests for derived-agent auto-provisioning.
2
3 When a profile page is visited for an agent that only exists as an `agent_id`
4 string in commit metadata (never formally registered), we:
5 1. Compute a deterministic genesis identity_id from the handle.
6 2. Upsert a real `musehub_identities` row (idempotent).
7 3. Return the real identity_id so the Spectral Sigil route can serve the SVG.
8
9 Tests are RED-first. Each assertion drives one concrete implementation decision.
10
11 New symbols:
12 musehub.core.genesis.compute_derived_agent_id
13 musehub.services.derived_agent_provisioner.ensure_agent_identity
14 musehub.services.derived_agent_provisioner.provision_if_derived
15 """
16 from __future__ import annotations
17
18 import re
19 from datetime import datetime, timezone
20 from unittest.mock import AsyncMock, MagicMock, patch
21
22 import pytest
23 import pytest_asyncio
24 from httpx import ASGITransport, AsyncClient
25 from sqlalchemy import select
26
27 from muse.core.types import blob_id, long_id
28 from musehub.main import app
29
30 # ---------------------------------------------------------------------------
31 # Helpers
32 # ---------------------------------------------------------------------------
33
34 _SONNET_HANDLE = "claude-sonnet-4-6"
35 _OPUS_HANDLE = "claude-opus-4-7"
36 _CUSTOM_HANDLE = "my-custom-agent-42"
37
38 _KNOWN_FIRST_SEEN = datetime(2026, 1, 31, 12, 0, 0, tzinfo=timezone.utc)
39
40
41 def _expected_id(handle: str) -> str:
42 """Restate the genesis formula so tests catch any deviation in implementation."""
43 return blob_id(f"agent_handle\x00{handle}".encode())
44
45
46 # ---------------------------------------------------------------------------
47 # Module imports — must exist
48 # ---------------------------------------------------------------------------
49
50
51 def test_genesis_module_exports_compute_derived_agent_id() -> None:
52 from musehub.core.genesis import compute_derived_agent_id # noqa: F401
53
54 assert callable(compute_derived_agent_id)
55
56
57 def test_provisioner_module_importable() -> None:
58 from musehub.services import derived_agent_provisioner # noqa: F401
59
60
61 def test_ensure_agent_identity_importable() -> None:
62 from musehub.services.derived_agent_provisioner import ensure_agent_identity # noqa: F401
63
64 assert callable(ensure_agent_identity)
65
66
67 def test_provision_if_derived_importable() -> None:
68 from musehub.services.derived_agent_provisioner import provision_if_derived # noqa: F401
69
70 assert callable(provision_if_derived)
71
72
73 # ---------------------------------------------------------------------------
74 # compute_derived_agent_id — deterministic genesis formula
75 # ---------------------------------------------------------------------------
76
77
78 def test_compute_derived_agent_id_returns_sha256_prefixed() -> None:
79 from musehub.core.genesis import compute_derived_agent_id
80
81 result = compute_derived_agent_id(_SONNET_HANDLE)
82 assert result.startswith("sha256:")
83
84
85 def test_compute_derived_agent_id_hex_part_is_64_chars() -> None:
86 from musehub.core.genesis import compute_derived_agent_id
87
88 result = compute_derived_agent_id(_SONNET_HANDLE)
89 hex_part = result[len("sha256:"):]
90 assert len(hex_part) == 64
91 assert re.fullmatch(r"[0-9a-f]{64}", hex_part)
92
93
94 def test_compute_derived_agent_id_is_deterministic() -> None:
95 from musehub.core.genesis import compute_derived_agent_id
96
97 assert compute_derived_agent_id(_SONNET_HANDLE) == compute_derived_agent_id(_SONNET_HANDLE)
98
99
100 def test_compute_derived_agent_id_differs_by_handle() -> None:
101 from musehub.core.genesis import compute_derived_agent_id
102
103 assert compute_derived_agent_id(_SONNET_HANDLE) != compute_derived_agent_id(_OPUS_HANDLE)
104 assert compute_derived_agent_id(_OPUS_HANDLE) != compute_derived_agent_id(_CUSTOM_HANDLE)
105
106
107 def test_compute_derived_agent_id_matches_known_formula() -> None:
108 """Nail the exact formula so any future refactor breaks loudly."""
109 from musehub.core.genesis import compute_derived_agent_id
110
111 assert compute_derived_agent_id(_SONNET_HANDLE) == _expected_id(_SONNET_HANDLE)
112 assert compute_derived_agent_id(_OPUS_HANDLE) == _expected_id(_OPUS_HANDLE)
113 assert compute_derived_agent_id(_CUSTOM_HANDLE) == _expected_id(_CUSTOM_HANDLE)
114
115
116 # ---------------------------------------------------------------------------
117 # ensure_agent_identity — upsert into musehub_identities
118 # ---------------------------------------------------------------------------
119
120
121 @pytest.mark.asyncio
122 async def test_ensure_agent_identity_creates_row(db_session) -> None:
123 from musehub.db.musehub_models import MusehubIdentity
124 from musehub.services.derived_agent_provisioner import ensure_agent_identity
125
126 result = await ensure_agent_identity(
127 db_session,
128 handle=_SONNET_HANDLE,
129 agent_model="claude-sonnet-4-6",
130 first_seen_at=_KNOWN_FIRST_SEEN,
131 )
132 await db_session.commit()
133
134 row = (await db_session.execute(
135 select(MusehubIdentity).where(MusehubIdentity.handle == _SONNET_HANDLE)
136 )).scalar_one_or_none()
137
138 assert row is not None
139 assert result.handle == _SONNET_HANDLE
140
141
142 @pytest.mark.asyncio
143 async def test_ensure_agent_identity_sets_correct_identity_id(db_session) -> None:
144 from musehub.services.derived_agent_provisioner import ensure_agent_identity
145
146 result = await ensure_agent_identity(
147 db_session,
148 handle=_SONNET_HANDLE,
149 agent_model="claude-sonnet-4-6",
150 first_seen_at=_KNOWN_FIRST_SEEN,
151 )
152
153 assert result.identity_id == _expected_id(_SONNET_HANDLE)
154
155
156 @pytest.mark.asyncio
157 async def test_ensure_agent_identity_sets_identity_type_agent(db_session) -> None:
158 from musehub.services.derived_agent_provisioner import ensure_agent_identity
159
160 result = await ensure_agent_identity(
161 db_session,
162 handle=_SONNET_HANDLE,
163 agent_model="claude-sonnet-4-6",
164 first_seen_at=_KNOWN_FIRST_SEEN,
165 )
166
167 assert result.identity_type == "agent"
168
169
170 @pytest.mark.asyncio
171 async def test_ensure_agent_identity_stores_agent_model(db_session) -> None:
172 from musehub.services.derived_agent_provisioner import ensure_agent_identity
173
174 result = await ensure_agent_identity(
175 db_session,
176 handle=_SONNET_HANDLE,
177 agent_model="claude-sonnet-4-6",
178 first_seen_at=_KNOWN_FIRST_SEEN,
179 )
180
181 assert result.agent_model == "claude-sonnet-4-6"
182
183
184 @pytest.mark.asyncio
185 async def test_ensure_agent_identity_model_none_is_accepted(db_session) -> None:
186 from musehub.services.derived_agent_provisioner import ensure_agent_identity
187
188 result = await ensure_agent_identity(
189 db_session,
190 handle=_CUSTOM_HANDLE,
191 agent_model=None,
192 first_seen_at=None,
193 )
194
195 assert result.identity_type == "agent"
196 assert result.agent_model is None
197
198
199 @pytest.mark.asyncio
200 async def test_ensure_agent_identity_is_idempotent(db_session) -> None:
201 """Calling twice with the same handle must not raise and must not duplicate."""
202 from musehub.db.musehub_models import MusehubIdentity
203 from musehub.services.derived_agent_provisioner import ensure_agent_identity
204
205 await ensure_agent_identity(db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN)
206 await db_session.commit()
207
208 await ensure_agent_identity(db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN)
209 await db_session.commit()
210
211 rows = (await db_session.execute(
212 select(MusehubIdentity).where(MusehubIdentity.handle == _SONNET_HANDLE)
213 )).scalars().all()
214
215 assert len(rows) == 1
216
217
218 @pytest.mark.asyncio
219 async def test_ensure_agent_identity_returns_existing_row_unchanged(db_session) -> None:
220 """If the row already exists, return it without touching model or timestamps."""
221 from musehub.db.musehub_models import MusehubIdentity
222 from musehub.services.derived_agent_provisioner import ensure_agent_identity
223
224 first = await ensure_agent_identity(db_session, _SONNET_HANDLE, "v1-model", _KNOWN_FIRST_SEEN)
225 await db_session.commit()
226 original_id = first.identity_id
227
228 second = await ensure_agent_identity(db_session, _SONNET_HANDLE, "v2-model-different", None)
229
230 assert second.identity_id == original_id
231
232
233 @pytest.mark.asyncio
234 async def test_ensure_agent_identity_different_handles_create_separate_rows(db_session) -> None:
235 from musehub.db.musehub_models import MusehubIdentity
236 from musehub.services.derived_agent_provisioner import ensure_agent_identity
237
238 await ensure_agent_identity(db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN)
239 await ensure_agent_identity(db_session, _OPUS_HANDLE, "claude-opus-4-7", _KNOWN_FIRST_SEEN)
240 await db_session.commit()
241
242 rows = (await db_session.execute(
243 select(MusehubIdentity).where(
244 MusehubIdentity.handle.in_([_SONNET_HANDLE, _OPUS_HANDLE])
245 )
246 )).scalars().all()
247
248 assert len(rows) == 2
249 ids = {r.identity_id for r in rows}
250 assert len(ids) == 2 # distinct genesis IDs
251
252
253 # ---------------------------------------------------------------------------
254 # provision_if_derived — the _resolve_identity integration hook
255 # ---------------------------------------------------------------------------
256
257
258 @pytest.mark.asyncio
259 async def test_provision_if_derived_provisions_derived_agent(db_session) -> None:
260 """Given a derived identity dict, provision_if_derived upserts and updates user_id."""
261 from musehub.db.musehub_models import MusehubIdentity
262 from musehub.services.derived_agent_provisioner import provision_if_derived
263
264 derived_identity = {
265 "handle": _SONNET_HANDLE,
266 "type": "agent",
267 "user_id": None,
268 "agent_model": "claude-sonnet-4-6",
269 "is_derived": True,
270 "member_since": _KNOWN_FIRST_SEEN,
271 }
272
273 updated = await provision_if_derived(db_session, derived_identity)
274 await db_session.commit()
275
276 assert updated["user_id"] == _expected_id(_SONNET_HANDLE)
277 assert updated["is_derived"] is False
278
279 row = (await db_session.execute(
280 select(MusehubIdentity).where(MusehubIdentity.handle == _SONNET_HANDLE)
281 )).scalar_one_or_none()
282 assert row is not None
283
284
285 @pytest.mark.asyncio
286 async def test_provision_if_derived_skips_non_agent(db_session) -> None:
287 """Humans and orgs are not auto-provisioned."""
288 from musehub.services.derived_agent_provisioner import provision_if_derived
289
290 human_identity = {
291 "handle": "gabriel",
292 "type": "human",
293 "user_id": None,
294 "is_derived": True,
295 "member_since": _KNOWN_FIRST_SEEN,
296 }
297
298 updated = await provision_if_derived(db_session, human_identity)
299
300 assert updated["user_id"] is None
301 assert updated["is_derived"] is True
302
303
304 @pytest.mark.asyncio
305 async def test_provision_if_derived_skips_already_registered(db_session) -> None:
306 """If user_id is already set (registered agent), don't touch it."""
307 from musehub.services.derived_agent_provisioner import provision_if_derived
308
309 existing_id = long_id("a" * 64)
310 registered = {
311 "handle": _SONNET_HANDLE,
312 "type": "agent",
313 "user_id": existing_id,
314 "is_derived": False,
315 "member_since": _KNOWN_FIRST_SEEN,
316 }
317
318 updated = await provision_if_derived(db_session, registered)
319
320 assert updated["user_id"] == existing_id
321
322
323 # ---------------------------------------------------------------------------
324 # Avatar route — sigil is served after provisioning
325 # ---------------------------------------------------------------------------
326
327
328 @pytest.mark.asyncio
329 async def test_avatar_route_serves_sigil_for_provisioned_agent(db_session) -> None:
330 """After ensure_agent_identity, the avatar route returns 200 image/svg+xml."""
331 from musehub.services.derived_agent_provisioner import ensure_agent_identity
332
333 identity = await ensure_agent_identity(
334 db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN
335 )
336 await db_session.commit()
337
338 hex_part = identity.identity_id[len("sha256:"):]
339
340 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
341 response = await client.get(f"/avatars/sha256/{hex_part}.svg")
342
343 assert response.status_code == 200
344 assert "image/svg+xml" in response.headers["content-type"]
345
346
347 @pytest.mark.asyncio
348 async def test_avatar_route_sigil_reflects_agent_archetype(db_session) -> None:
349 """The generated SVG must use the agent archetype (polygon/path for hex shape)."""
350 import xml.etree.ElementTree as ET
351 from musehub.services.derived_agent_provisioner import ensure_agent_identity
352
353 identity = await ensure_agent_identity(
354 db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN
355 )
356 await db_session.commit()
357
358 hex_part = identity.identity_id[len("sha256:"):]
359
360 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
361 response = await client.get(f"/avatars/sha256/{hex_part}.svg")
362
363 root = ET.fromstring(response.content)
364 paths = root.findall(".//{http://www.w3.org/2000/svg}path") or root.findall(".//path")
365 polygons = root.findall(".//{http://www.w3.org/2000/svg}polygon") or root.findall(".//polygon")
366 assert len(paths) + len(polygons) >= 1, "agent sigil must contain path or polygon"
367
368
369 @pytest.mark.asyncio
370 async def test_avatar_sigil_is_deterministic_for_provisioned_agent(db_session) -> None:
371 """Two requests for the same provisioned agent return identical SVG bytes."""
372 from musehub.services.derived_agent_provisioner import ensure_agent_identity
373
374 identity = await ensure_agent_identity(
375 db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN
376 )
377 await db_session.commit()
378
379 hex_part = identity.identity_id[len("sha256:"):]
380
381 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
382 r1 = await client.get(f"/avatars/sha256/{hex_part}.svg")
383 r2 = await client.get(f"/avatars/sha256/{hex_part}.svg")
384
385 assert r1.content == r2.content
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago