gabriel / muse public
test_coordination.py python
420 lines 17.0 KB
Raw
sha256:d11a87833d5fad6059b7662844bf5448a8911a17cce7a51811f71ad394f248eb bump to v0.2.0rc13 Human patch 6 days ago
1 """Tests for muse/core/coordination.py — multi-agent coordination layer.
2
3 Coverage
4 --------
5 Directory helpers
6 - _ensure_coord_dirs creates .muse/coordination/reservations/ and intents/.
7
8 Reservation
9 - create_reservation writes a valid JSON file.
10 - Reservation.from_dict / to_dict round-trip.
11 - Reservation.is_active() returns True for non-expired, False for expired.
12 - load_all_reservations loads all files including expired.
13 - active_reservations filters out expired reservations.
14 - Corrupt reservation file is skipped with a warning.
15 - Multiple reservations can coexist for the same address.
16
17 Intent
18 - create_intent writes a valid JSON file.
19 - Intent.from_dict / to_dict round-trip.
20 - load_all_intents loads all files.
21 - Corrupt intent file is skipped.
22
23 Schema
24 - All records have schema_version == __version__.
25 - created_at and expires_at are ISO 8601 strings.
26 - operation field is None-able for reservations.
27 """
28
29 import datetime
30 import json
31 import pathlib
32
33 import pytest
34
35 from muse._version import __version__
36 from muse.core.coordination import (
37 Intent,
38 Reservation,
39 active_reservations,
40 create_intent,
41 create_reservation,
42 load_all_intents,
43 load_all_reservations,
44 )
45 from muse.core.paths import coordination_dir
46
47
48 # ---------------------------------------------------------------------------
49 # Helpers
50 # ---------------------------------------------------------------------------
51
52
53 def _now() -> datetime.datetime:
54 return datetime.datetime.now(datetime.timezone.utc)
55
56
57 def _future(seconds: int = 3600) -> datetime.datetime:
58 return _now() + datetime.timedelta(seconds=seconds)
59
60
61 def _past(seconds: int = 60) -> datetime.datetime:
62 return _now() - datetime.timedelta(seconds=seconds)
63
64
65 # ---------------------------------------------------------------------------
66 # Reservation — create and load
67 # ---------------------------------------------------------------------------
68
69
70 class TestCreateReservation:
71 def test_creates_json_file(self, tmp_path: pathlib.Path) -> None:
72 res = create_reservation(
73 tmp_path,
74 run_id="agent-1",
75 branch="feature-x",
76 addresses=["src/billing.py::compute_total"],
77 )
78 rdir = coordination_dir(tmp_path) / "reservations"
79 assert rdir.exists()
80 files = list(rdir.glob("*.json"))
81 assert len(files) == 1
82 data = json.loads(files[0].read_text())
83 assert data["reservation_id"] == res.reservation_id
84 assert data["run_id"] == "agent-1"
85 assert data["branch"] == "feature-x"
86 assert data["addresses"] == ["src/billing.py::compute_total"]
87 assert data["schema_version"] == __version__
88
89 def test_default_ttl_sets_future_expiry(self, tmp_path: pathlib.Path) -> None:
90 res = create_reservation(tmp_path, run_id="r", branch="main", addresses=[])
91 assert res.expires_at > _now()
92
93 def test_custom_ttl(self, tmp_path: pathlib.Path) -> None:
94 res = create_reservation(
95 tmp_path, run_id="r", branch="main", addresses=[], ttl_seconds=7200
96 )
97 delta = res.expires_at - res.created_at
98 assert abs(delta.total_seconds() - 7200) < 5
99
100 def test_operation_stored(self, tmp_path: pathlib.Path) -> None:
101 res = create_reservation(
102 tmp_path, run_id="r", branch="main",
103 addresses=["src/a.py::f"], operation="rename"
104 )
105 assert res.operation == "rename"
106 rdir = coordination_dir(tmp_path) / "reservations"
107 data = json.loads(list(rdir.glob("*.json"))[0].read_text())
108 assert data["operation"] == "rename"
109
110 def test_none_operation(self, tmp_path: pathlib.Path) -> None:
111 res = create_reservation(tmp_path, run_id="r", branch="main", addresses=[])
112 assert res.operation is None
113
114 def test_multiple_addresses(self, tmp_path: pathlib.Path) -> None:
115 addrs = ["src/a.py::f", "src/b.py::g", "src/c.py::h"]
116 res = create_reservation(tmp_path, run_id="r", branch="main", addresses=addrs)
117 assert res.addresses == addrs
118
119 def test_multiple_reservations_coexist(self, tmp_path: pathlib.Path) -> None:
120 create_reservation(tmp_path, run_id="a1", branch="main", addresses=["src/a.py::f"])
121 create_reservation(tmp_path, run_id="a2", branch="main", addresses=["src/a.py::f"])
122 rdir = coordination_dir(tmp_path) / "reservations"
123 assert len(list(rdir.glob("*.json"))) == 2
124
125
126 # ---------------------------------------------------------------------------
127 # Reservation — to_dict / from_dict
128 # ---------------------------------------------------------------------------
129
130
131 class TestReservationRoundTrip:
132 def test_to_dict_from_dict(self) -> None:
133 now = _now()
134 future = _future()
135 res = Reservation(
136 reservation_id="test-res-id",
137 run_id="agent-7",
138 branch="feature-y",
139 addresses=["src/x.py::func"],
140 created_at=now,
141 expires_at=future,
142 operation="move",
143 )
144 d = res.to_dict()
145 res2 = Reservation.from_dict(d)
146 assert res2.reservation_id == "test-res-id"
147 assert res2.run_id == "agent-7"
148 assert res2.branch == "feature-y"
149 assert res2.addresses == ["src/x.py::func"]
150 assert res2.operation == "move"
151 # Timestamps round-trip via ISO 8601
152 assert abs((res2.expires_at - future).total_seconds()) < 1
153
154 def test_schema_version_in_dict(self) -> None:
155 res = Reservation(
156 reservation_id="x", run_id="r", branch="b",
157 addresses=[], created_at=_now(), expires_at=_future(), operation=None
158 )
159 assert res.to_dict()["schema_version"] == __version__
160
161
162 # ---------------------------------------------------------------------------
163 # Reservation — is_active
164 # ---------------------------------------------------------------------------
165
166
167 class TestReservationIsActive:
168 def test_active_when_future_expiry(self, tmp_path: pathlib.Path) -> None:
169 res = create_reservation(tmp_path, run_id="r", branch="main", addresses=[], ttl_seconds=3600)
170 assert res.is_active()
171
172 def test_inactive_when_past_expiry(self) -> None:
173 res = Reservation(
174 reservation_id="x", run_id="r", branch="b",
175 addresses=[], created_at=_past(120), expires_at=_past(60), operation=None
176 )
177 assert not res.is_active()
178
179
180 # ---------------------------------------------------------------------------
181 # load_all_reservations / active_reservations
182 # ---------------------------------------------------------------------------
183
184
185 class TestLoadReservations:
186 def test_load_all_includes_expired(self, tmp_path: pathlib.Path) -> None:
187 create_reservation(tmp_path, run_id="r1", branch="main", addresses=[], ttl_seconds=3600)
188 # Manually write an expired reservation.
189 past = _past(120)
190 expired = Reservation(
191 reservation_id="expired-res-id",
192 run_id="r2",
193 branch="main",
194 addresses=[],
195 created_at=_past(200),
196 expires_at=past,
197 operation=None,
198 )
199 rdir = coordination_dir(tmp_path) / "reservations"
200 rdir.mkdir(parents=True, exist_ok=True)
201 (rdir / "expired-res-id.json").write_text(f"{json.dumps(expired.to_dict())}\n")
202
203 all_res = load_all_reservations(tmp_path)
204 assert len(all_res) == 2
205
206 def test_active_reservations_filters_expired(self, tmp_path: pathlib.Path) -> None:
207 create_reservation(tmp_path, run_id="r1", branch="main", addresses=[], ttl_seconds=3600)
208 past = _past(120)
209 expired = Reservation(
210 reservation_id="expired-res-id",
211 run_id="r2", branch="main", addresses=[],
212 created_at=_past(200), expires_at=past, operation=None,
213 )
214 rdir = coordination_dir(tmp_path) / "reservations"
215 rdir.mkdir(parents=True, exist_ok=True)
216 (rdir / "expired-res-id.json").write_text(f"{json.dumps(expired.to_dict())}\n")
217
218 active = active_reservations(tmp_path)
219 assert len(active) == 1
220 assert active[0].run_id == "r1"
221
222 def test_empty_dir_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
223 rdir = coordination_dir(tmp_path) / "reservations"
224 rdir.mkdir(parents=True, exist_ok=True)
225 assert load_all_reservations(tmp_path) == []
226
227 def test_nonexistent_dir_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
228 assert load_all_reservations(tmp_path) == []
229
230 def test_corrupt_file_skipped(self, tmp_path: pathlib.Path) -> None:
231 rdir = coordination_dir(tmp_path) / "reservations"
232 rdir.mkdir(parents=True, exist_ok=True)
233 (rdir / "bad.json").write_text("not valid json{{{")
234 result = load_all_reservations(tmp_path)
235 assert result == []
236
237
238 # ---------------------------------------------------------------------------
239 # Intent — create and load
240 # ---------------------------------------------------------------------------
241
242
243 class TestCreateIntent:
244 def test_creates_json_file(self, tmp_path: pathlib.Path) -> None:
245 intent = create_intent(
246 tmp_path,
247 reservation_id="res-id",
248 run_id="agent-2",
249 branch="feature-z",
250 addresses=["src/billing.py::Invoice"],
251 operation="rename",
252 detail="rename to InvoiceRecord",
253 )
254 idir = coordination_dir(tmp_path) / "intents"
255 assert idir.exists()
256 files = list(idir.glob("*.json"))
257 assert len(files) == 1
258 data = json.loads(files[0].read_text())
259 assert data["intent_id"] == intent.intent_id
260 assert data["reservation_id"] == "res-id"
261 assert data["operation"] == "rename"
262 assert data["detail"] == "rename to InvoiceRecord"
263 assert data["schema_version"] == __version__
264
265 def test_empty_detail_defaults_to_empty_string(self, tmp_path: pathlib.Path) -> None:
266 intent = create_intent(
267 tmp_path, reservation_id="", run_id="r", branch="main",
268 addresses=[], operation="modify",
269 )
270 assert intent.detail == ""
271
272 def test_multiple_intents(self, tmp_path: pathlib.Path) -> None:
273 create_intent(tmp_path, reservation_id="", run_id="a", branch="main",
274 addresses=["x.py::f"], operation="rename")
275 create_intent(tmp_path, reservation_id="", run_id="b", branch="main",
276 addresses=["x.py::g"], operation="delete")
277 idir = coordination_dir(tmp_path) / "intents"
278 assert len(list(idir.glob("*.json"))) == 2
279
280
281 # ---------------------------------------------------------------------------
282 # Intent — to_dict / from_dict
283 # ---------------------------------------------------------------------------
284
285
286 class TestIntentRoundTrip:
287 def test_to_dict_from_dict(self) -> None:
288 now = _now()
289 intent = Intent(
290 intent_id="intent-id",
291 reservation_id="res-id",
292 run_id="agent-3",
293 branch="dev",
294 addresses=["src/y.py::Bar"],
295 operation="extract",
296 created_at=now,
297 detail="extract helper",
298 )
299 d = intent.to_dict()
300 intent2 = Intent.from_dict(d)
301 assert intent2.intent_id == "intent-id"
302 assert intent2.reservation_id == "res-id"
303 assert intent2.operation == "extract"
304 assert intent2.detail == "extract helper"
305 assert intent2.addresses == ["src/y.py::Bar"]
306
307 def test_schema_version_in_dict(self) -> None:
308 intent = Intent(
309 intent_id="x", reservation_id="", run_id="r", branch="b",
310 addresses=[], operation="modify", created_at=_now(), detail="",
311 )
312 assert intent.to_dict()["schema_version"] == __version__
313
314
315 # ---------------------------------------------------------------------------
316 # load_all_intents
317 # ---------------------------------------------------------------------------
318
319
320 class TestLoadAllIntents:
321 def test_empty_dir(self, tmp_path: pathlib.Path) -> None:
322 idir = coordination_dir(tmp_path) / "intents"
323 idir.mkdir(parents=True, exist_ok=True)
324 assert load_all_intents(tmp_path) == []
325
326 def test_nonexistent_dir(self, tmp_path: pathlib.Path) -> None:
327 assert load_all_intents(tmp_path) == []
328
329 def test_loads_created_intents(self, tmp_path: pathlib.Path) -> None:
330 create_intent(tmp_path, reservation_id="r", run_id="a", branch="main",
331 addresses=["x.py::f"], operation="rename")
332 create_intent(tmp_path, reservation_id="r", run_id="b", branch="dev",
333 addresses=["y.py::g"], operation="modify")
334 intents = load_all_intents(tmp_path)
335 assert len(intents) == 2
336 ops = {i.operation for i in intents}
337 assert "rename" in ops
338 assert "modify" in ops
339
340 def test_corrupt_intent_skipped(self, tmp_path: pathlib.Path) -> None:
341 idir = coordination_dir(tmp_path) / "intents"
342 idir.mkdir(parents=True, exist_ok=True)
343 (idir / "bad.json").write_text("{invalid")
344 result = load_all_intents(tmp_path)
345 assert result == []
346
347
348 # ---------------------------------------------------------------------------
349 # Content-addressed reservation_id and intent_id
350 # ---------------------------------------------------------------------------
351
352 import re as _re
353 _UUID4_RE = _re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$")
354
355
356 class TestReservationIdContentAddressed:
357 def test_reservation_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
358 res = create_reservation(tmp_path, run_id="agent-1", branch="main",
359 addresses=["src/foo.py::Bar"], ttl_seconds=60)
360 assert res.reservation_id.startswith("sha256:"), f"Got {res.reservation_id!r}"
361 assert len(res.reservation_id) == 71
362
363 def test_reservation_id_is_sha256_not_uuid4(self, tmp_path: pathlib.Path) -> None:
364 res = create_reservation(tmp_path, run_id="agent-1", branch="main",
365 addresses=["src/foo.py::Bar"], ttl_seconds=60)
366 assert not _UUID4_RE.match(res.reservation_id)
367
368 def test_reservation_id_deterministic(self, tmp_path: pathlib.Path) -> None:
369 """Same inputs → same reservation_id."""
370 from muse.core.coordination import compute_reservation_id
371 r1 = compute_reservation_id(run_id="agent-1", branch="main",
372 addresses=["a.py::f"], operation="modify")
373 r2 = compute_reservation_id(run_id="agent-1", branch="main",
374 addresses=["a.py::f"], operation="modify")
375 assert r1 == r2
376
377 def test_reservation_id_differs_by_run_id(self, tmp_path: pathlib.Path) -> None:
378 from muse.core.coordination import compute_reservation_id
379 r1 = compute_reservation_id("agent-1", "main", ["a.py::f"], "modify")
380 r2 = compute_reservation_id("agent-2", "main", ["a.py::f"], "modify")
381 assert r1 != r2
382
383 def test_reservation_id_differs_by_addresses(self, tmp_path: pathlib.Path) -> None:
384 from muse.core.coordination import compute_reservation_id
385 r1 = compute_reservation_id("agent-1", "main", ["a.py::f"], "modify")
386 r2 = compute_reservation_id("agent-1", "main", ["b.py::g"], "modify")
387 assert r1 != r2
388
389
390 class TestIntentIdContentAddressed:
391 def test_intent_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
392 intent = create_intent(tmp_path, reservation_id="r-1", run_id="agent-1",
393 branch="main", addresses=["x.py::f"], operation="rename")
394 assert intent.intent_id.startswith("sha256:"), f"Got {intent.intent_id!r}"
395 assert len(intent.intent_id) == 71
396
397 def test_intent_id_is_sha256_not_uuid4(self, tmp_path: pathlib.Path) -> None:
398 intent = create_intent(tmp_path, reservation_id="r-1", run_id="agent-1",
399 branch="main", addresses=["x.py::f"], operation="rename")
400 assert not _UUID4_RE.match(intent.intent_id)
401
402 def test_intent_id_deterministic(self, tmp_path: pathlib.Path) -> None:
403 from muse.core.coordination import compute_intent_id
404 i1 = compute_intent_id(reservation_id="r-1", run_id="agent-1",
405 branch="main", addresses=["x.py::f"], operation="rename")
406 i2 = compute_intent_id(reservation_id="r-1", run_id="agent-1",
407 branch="main", addresses=["x.py::f"], operation="rename")
408 assert i1 == i2
409
410 def test_intent_id_differs_by_operation(self, tmp_path: pathlib.Path) -> None:
411 from muse.core.coordination import compute_intent_id
412 i1 = compute_intent_id("r-1", "agent-1", "main", ["x.py::f"], "rename")
413 i2 = compute_intent_id("r-1", "agent-1", "main", ["x.py::f"], "delete")
414 assert i1 != i2
415
416 def test_intent_id_differs_by_reservation(self, tmp_path: pathlib.Path) -> None:
417 from muse.core.coordination import compute_intent_id
418 i1 = compute_intent_id("r-1", "agent-1", "main", ["x.py::f"], "rename")
419 i2 = compute_intent_id("r-2", "agent-1", "main", ["x.py::f"], "rename")
420 assert i1 != i2
File History 1 commit
sha256:d11a87833d5fad6059b7662844bf5448a8911a17cce7a51811f71ad394f248eb bump to v0.2.0rc13 Human patch 6 days ago