gabriel / muse public
test_coord_performance.py python
1,226 lines 46.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """
2 EXTREME performance test suite for ``muse coord`` — Linus Torvalds porting
3 Linux from Git to Muse.
4
5 Scenario: 7 000+ agents, 150 000+ files, kernel subsystems as task clusters.
6 We want to break Muse Coord. We want to find where the edge is and go beyond.
7
8 Bottlenecks targeted:
9 1. _gather_local_records — O(N) file reads, no caching, no batching
10 2. _write_remote_records — mkdir() called PER record (N syscalls vs 7)
11 3. run_coord_gc — releases directory scanned TWICE (load_released_ids
12 + load_all_releases); active_reservations does 3
13 full directory scans
14 4. JSON payload scaling — 50B / 1 KB / 10 KB / 50 KB per record
15 5. Batch loop overhead — pure CPU cost of splitting 7 000 records into batches
16 6. Memory footprint — 10 000 records held in memory simultaneously
17 7. End-to-end throughput — records/second metric
18
19 Existing tests in test_core_coord_bus.py (do NOT duplicate):
20 - 500-record push serialization < 1 s
21 - 1000-record pull parse < 1 s
22 - 100 sequential push_to_hub < 2 s
23 - 100k _build_url calls < 1 s
24 """
25 from __future__ import annotations
26
27 import json
28 import os
29 import pathlib
30 import sys
31 import time
32 import tracemalloc
33 import uuid
34 from collections.abc import Callable
35 from typing import TYPE_CHECKING
36 from unittest.mock import MagicMock, patch
37
38 import pytest
39
40 from muse.core._types import MsgpackDict
41 from muse.core.coord_bus import JsonDict
42
43 if TYPE_CHECKING:
44 from muse.core.coordination import CoordGcResult, Reservation
45 from muse.core.transport import SigningIdentity
46
47 # ── helpers ──────────────────────────────────────────────────────────────────
48
49 _ALL_KINDS = ("reservation", "intent", "release", "heartbeat", "dependency", "task", "claim")
50
51 _KIND_SUBDIR = {
52 "reservation": "reservations",
53 "intent": "intents",
54 "release": "releases",
55 "heartbeat": "heartbeats",
56 "dependency": "dependencies",
57 "task": "tasks",
58 "claim": "claims",
59 }
60
61 _FUTURE_TS = "2099-12-31T23:59:59+00:00"
62 _PAST_TS = "2000-01-01T00:00:00+00:00"
63
64
65 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
66 (tmp_path / ".muse").mkdir(parents=True, exist_ok=True)
67 return tmp_path
68
69
70 def _coord_dir(root: pathlib.Path) -> pathlib.Path:
71 d = root / ".muse" / "coordination"
72 d.mkdir(parents=True, exist_ok=True)
73 return d
74
75
76 def _write_records(root: pathlib.Path, kind: str, n: int, payload_bytes: int = 64) -> None:
77 """Write *n* minimal records of *kind* to the local coordination store."""
78 subdir = _KIND_SUBDIR[kind]
79 d = _coord_dir(root) / subdir
80 d.mkdir(parents=True, exist_ok=True)
81
82 padding = "x" * max(0, payload_bytes - 80)
83
84 for i in range(n):
85 rid = f"{kind}-{i:06d}"
86 if kind == "reservation":
87 rec = {
88 "reservation_id": rid,
89 "run_id": f"run-{i}",
90 "expires_at": _FUTURE_TS,
91 "symbols": [f"kernel/subsys_{i % 50}/module.c::func_{i}"],
92 "_pad": padding,
93 }
94 elif kind == "heartbeat":
95 rec = {
96 "run_id": rid,
97 "extended_expires_at": _FUTURE_TS,
98 "expires_at": _FUTURE_TS,
99 "_pad": padding,
100 }
101 elif kind == "intent":
102 rec = {
103 "intent_id": rid,
104 "run_id": f"run-{i}",
105 "reservation_id": f"reservation-{i:06d}",
106 "operation": "modify",
107 "target": f"kernel/subsys_{i % 50}/module.c::func_{i}",
108 "_pad": padding,
109 }
110 elif kind == "release":
111 rec = {
112 "release_id": rid,
113 "reservation_id": f"reservation-{i:06d}",
114 "run_id": f"run-{i}",
115 "released_at": _FUTURE_TS,
116 "_pad": padding,
117 }
118 elif kind == "dependency":
119 rec = {
120 "reservation_id": rid,
121 "depends_on": [f"reservation-{j:06d}" for j in range(min(3, i))],
122 "_pad": padding,
123 }
124 elif kind == "task":
125 rec = {
126 "task_id": rid,
127 "run_id": f"run-{i}",
128 "description": f"patch kernel subsystem {i % 50}",
129 "_pad": padding,
130 }
131 elif kind == "claim":
132 rec = {
133 "task_id": rid,
134 "claimer_run_id": f"run-{i}",
135 "expires_at": _FUTURE_TS,
136 "_pad": padding,
137 }
138 else:
139 rec = {"id": rid, "_pad": padding}
140
141 (d / f"{rid}.json").write_text(json.dumps(rec), encoding="utf-8")
142
143
144 def _write_expired_reservations(root: pathlib.Path, n: int) -> None:
145 """Write *n* expired reservations (expires_at in the past, no heartbeat, no release)."""
146 d = _coord_dir(root) / "reservations"
147 d.mkdir(parents=True, exist_ok=True)
148 for i in range(n):
149 rid = f"exp-res-{i:06d}"
150 rec = {
151 "reservation_id": rid,
152 "run_id": f"run-{i}",
153 "expires_at": _PAST_TS,
154 "symbols": [f"drivers/block/hd_{i}.c::init"],
155 }
156 (d / f"{rid}.json").write_text(json.dumps(rec), encoding="utf-8")
157
158
159 def _write_released_reservations(root: pathlib.Path, n: int) -> None:
160 """Write *n* pairs: reservation + release tombstone (released in the far past)."""
161 res_dir = _coord_dir(root) / "reservations"
162 rel_dir = _coord_dir(root) / "releases"
163 res_dir.mkdir(parents=True, exist_ok=True)
164 rel_dir.mkdir(parents=True, exist_ok=True)
165 for i in range(n):
166 rid = f"rel-res-{i:06d}"
167 rec = {
168 "reservation_id": rid,
169 "run_id": f"run-{i}",
170 "expires_at": _PAST_TS,
171 "symbols": [f"net/ipv4/tcp_{i}.c::send"],
172 }
173 (res_dir / f"{rid}.json").write_text(json.dumps(rec), encoding="utf-8")
174 tombstone = {
175 "release_id": f"release-{i:06d}",
176 "reservation_id": rid,
177 "run_id": f"run-{i}",
178 "released_at": _PAST_TS,
179 }
180 (rel_dir / f"{rid}.json").write_text(json.dumps(tombstone), encoding="utf-8")
181
182
183 # ── import targets ────────────────────────────────────────────────────────────
184
185 def _import_gather() -> Callable[[pathlib.Path, list[str]], list[JsonDict]]:
186 from muse.cli.commands.coord_sync import _gather_local_records
187 return _gather_local_records
188
189
190 def _import_write_remote() -> Callable[[pathlib.Path, list[JsonDict]], None]:
191 from muse.cli.commands.coord_sync import _write_remote_records
192 return _write_remote_records
193
194
195 def _import_run_coord_gc() -> Callable[..., CoordGcResult]:
196 from muse.core.coordination import run_coord_gc
197 return run_coord_gc
198
199
200 def _import_active_reservations() -> Callable[[pathlib.Path], list[Reservation]]:
201 from muse.core.coordination import active_reservations
202 return active_reservations
203
204
205 # =============================================================================
206 # 1. _gather_local_records — O(N) file I/O
207 # =============================================================================
208
209 class TestCoordPerfGather:
210 """
211 _gather_local_records reads every file on every call. These tests measure
212 raw throughput at increasing scale and expose the O(N) cost.
213 """
214
215 def test_gather_100_reservations_under_200ms(self, tmp_path: pathlib.Path) -> None:
216 root = _make_repo(tmp_path)
217 _write_records(root, "reservation", 100)
218 gather = _import_gather()
219
220 t0 = time.monotonic()
221 records = gather(root, ["reservation"])
222 elapsed = time.monotonic() - t0
223
224 assert len(records) == 100
225 assert elapsed < 0.200, f"gather 100 reservations took {elapsed:.3f}s (> 200ms)"
226
227 def test_gather_500_reservations_under_500ms(self, tmp_path: pathlib.Path) -> None:
228 root = _make_repo(tmp_path)
229 _write_records(root, "reservation", 500)
230 gather = _import_gather()
231
232 t0 = time.monotonic()
233 records = gather(root, ["reservation"])
234 elapsed = time.monotonic() - t0
235
236 assert len(records) == 500
237 assert elapsed < 0.500, f"gather 500 reservations took {elapsed:.3f}s (> 500ms)"
238
239 def test_gather_1000_reservations_under_1s(self, tmp_path: pathlib.Path) -> None:
240 root = _make_repo(tmp_path)
241 _write_records(root, "reservation", 1000)
242 gather = _import_gather()
243
244 t0 = time.monotonic()
245 records = gather(root, ["reservation"])
246 elapsed = time.monotonic() - t0
247
248 assert len(records) == 1000
249 assert elapsed < 1.0, f"gather 1000 reservations took {elapsed:.3f}s (> 1s)"
250
251 @pytest.mark.slow
252 def test_gather_5000_reservations_under_5s(self, tmp_path: pathlib.Path) -> None:
253 root = _make_repo(tmp_path)
254 _write_records(root, "reservation", 5000)
255 gather = _import_gather()
256
257 t0 = time.monotonic()
258 records = gather(root, ["reservation"])
259 elapsed = time.monotonic() - t0
260
261 assert len(records) == 5000
262 assert elapsed < 5.0, f"gather 5000 reservations took {elapsed:.3f}s (> 5s)"
263
264 @pytest.mark.slow
265 def test_gather_all_7_kinds_1000_each_under_7s(self, tmp_path: pathlib.Path) -> None:
266 """7000 files across 7 kinds — worst-case full Linux kernel scenario."""
267 root = _make_repo(tmp_path)
268 for kind in _ALL_KINDS:
269 _write_records(root, kind, 1000)
270 gather = _import_gather()
271
272 t0 = time.monotonic()
273 records = gather(root, list(_ALL_KINDS))
274 elapsed = time.monotonic() - t0
275
276 assert len(records) == 7000
277 assert elapsed < 7.0, f"gather 7000 records (all kinds) took {elapsed:.3f}s (> 7s)"
278
279 def test_cold_vs_warm_gather_same_directory(self, tmp_path: pathlib.Path) -> None:
280 """
281 Second call must not be dramatically slower than the first.
282 There is no caching; both calls read from disk. The second call
283 benefits only from OS page-cache effects — that is acceptable.
284 """
285 root = _make_repo(tmp_path)
286 _write_records(root, "reservation", 200)
287 gather = _import_gather()
288
289 t0 = time.monotonic()
290 gather(root, ["reservation"])
291 cold = time.monotonic() - t0
292
293 t0 = time.monotonic()
294 gather(root, ["reservation"])
295 warm = time.monotonic() - t0
296
297 # Warm should not be more than 4× cold (OS cache should help).
298 # We do NOT assert warm < cold because there is no in-process cache.
299 assert warm < max(cold * 4, 0.500), (
300 f"warm gather ({warm:.3f}s) is unexpectedly slow vs cold ({cold:.3f}s)"
301 )
302
303 def test_gather_empty_directory_is_fast(self, tmp_path: pathlib.Path) -> None:
304 root = _make_repo(tmp_path)
305 _coord_dir(root) # creates .muse/coordination/ but no kind subdirs
306 gather = _import_gather()
307
308 t0 = time.monotonic()
309 for _ in range(100):
310 gather(root, list(_ALL_KINDS))
311 elapsed = time.monotonic() - t0
312
313 assert elapsed < 0.200, f"100 × empty gather took {elapsed:.3f}s (> 200ms)"
314
315 def test_gather_single_kind_ignores_other_dirs(self, tmp_path: pathlib.Path) -> None:
316 """Filtering to one kind must not scan 6 other directories."""
317 root = _make_repo(tmp_path)
318 for kind in _ALL_KINDS:
319 _write_records(root, kind, 200) # 1400 files total
320 gather = _import_gather()
321
322 t0 = time.monotonic()
323 records = gather(root, ["reservation"])
324 elapsed = time.monotonic() - t0
325
326 assert len(records) == 200
327 assert elapsed < 0.500, (
328 f"single-kind gather across 1400-file repo took {elapsed:.3f}s (> 500ms)"
329 )
330
331 def test_gather_throughput_records_per_second(self, tmp_path: pathlib.Path) -> None:
332 """
333 Measures records/second. Documents current throughput so regressions
334 become visible. On modern SSD hardware ≥ 1 000 records/s expected.
335 """
336 root = _make_repo(tmp_path)
337 _write_records(root, "reservation", 500)
338 gather = _import_gather()
339
340 # Warm up OS page cache
341 gather(root, ["reservation"])
342
343 t0 = time.monotonic()
344 records = gather(root, ["reservation"])
345 elapsed = time.monotonic() - t0
346
347 rps = len(records) / elapsed if elapsed > 0 else float("inf")
348 # This is a documentation assertion — fail loudly if throughput crashes.
349 assert rps >= 500, f"gather throughput {rps:.0f} records/s is below minimum 500/s"
350
351
352 # =============================================================================
353 # 2. _write_remote_records — mkdir() per record
354 # =============================================================================
355
356 class TestCoordPerfWriteRemote:
357 """
358 _write_remote_records calls kind_dir.mkdir(parents=True, exist_ok=True) for
359 EVERY record. With 7 distinct kinds and 7 000 records that is 7 000 mkdir
360 syscalls. These tests measure that overhead and expose the regression path.
361 """
362
363 def test_write_100_records_7_kinds_under_500ms(self, tmp_path: pathlib.Path) -> None:
364 root = _make_repo(tmp_path)
365 write_remote = _import_write_remote()
366
367 records = []
368 for i in range(100):
369 kind = _ALL_KINDS[i % len(_ALL_KINDS)]
370 records.append({
371 "kind": kind,
372 "record_uuid": f"{kind}-{i:06d}",
373 "run_id": f"run-{i}",
374 "payload": {"data": "x" * 64},
375 "expires_at": _FUTURE_TS,
376 })
377
378 t0 = time.monotonic()
379 write_remote(root, records)
380 elapsed = time.monotonic() - t0
381
382 assert elapsed < 0.500, f"write 100 remote records took {elapsed:.3f}s (> 500ms)"
383 # Verify files actually written
384 remote_dir = root / ".muse" / "coordination" / "remote"
385 written = sum(1 for _ in remote_dir.rglob("*.json"))
386 assert written == 100
387
388 def test_write_1000_records_7_kinds_under_3s(self, tmp_path: pathlib.Path) -> None:
389 root = _make_repo(tmp_path)
390 write_remote = _import_write_remote()
391
392 records = [
393 {
394 "kind": _ALL_KINDS[i % len(_ALL_KINDS)],
395 "record_uuid": f"uuid-{i:06d}",
396 "run_id": f"run-{i}",
397 "payload": {"idx": i},
398 "expires_at": _FUTURE_TS,
399 }
400 for i in range(1000)
401 ]
402
403 t0 = time.monotonic()
404 write_remote(root, records)
405 elapsed = time.monotonic() - t0
406
407 assert elapsed < 3.0, f"write 1000 remote records took {elapsed:.3f}s (> 3s)"
408
409 @pytest.mark.slow
410 def test_write_7000_records_mkdir_overhead_documented(self, tmp_path: pathlib.Path) -> None:
411 """
412 Documents the mkdir-per-record overhead at full Linux kernel scale.
413 7 000 records × 7 kinds = 7 000 mkdir() calls instead of the optimal 7.
414
415 This test is expected to PASS with the current implementation.
416 Its purpose: if a future optimization pre-creates directories (7 calls),
417 the elapsed time should drop by ~30-50%. A regression would show here.
418 """
419 root = _make_repo(tmp_path)
420 write_remote = _import_write_remote()
421
422 records = [
423 {
424 "kind": _ALL_KINDS[i % len(_ALL_KINDS)],
425 "record_uuid": f"uuid-{i:06d}",
426 "run_id": f"run-{i}",
427 "payload": {"idx": i},
428 "expires_at": None,
429 }
430 for i in range(7000)
431 ]
432
433 t0 = time.monotonic()
434 write_remote(root, records)
435 elapsed = time.monotonic() - t0
436
437 remote_dir = root / ".muse" / "coordination" / "remote"
438 written = sum(1 for _ in remote_dir.rglob("*.json"))
439 assert written == 7000, f"expected 7000 files, got {written}"
440 # Generous bound — this is the CURRENT (suboptimal) ceiling
441 assert elapsed < 20.0, f"write 7000 remote records took {elapsed:.3f}s (> 20s)"
442
443 def test_write_throughput_records_per_second(self, tmp_path: pathlib.Path) -> None:
444 """Baseline throughput for write_remote_records at 500 records."""
445 root = _make_repo(tmp_path)
446 write_remote = _import_write_remote()
447
448 records = [
449 {
450 "kind": _ALL_KINDS[i % len(_ALL_KINDS)],
451 "record_uuid": f"uuid-{i:06d}",
452 "run_id": f"run-{i}",
453 "payload": {"idx": i},
454 "expires_at": None,
455 }
456 for i in range(500)
457 ]
458
459 t0 = time.monotonic()
460 write_remote(root, records)
461 elapsed = time.monotonic() - t0
462
463 rps = 500 / elapsed if elapsed > 0 else float("inf")
464 assert rps >= 100, f"write_remote throughput {rps:.0f} records/s is below minimum 100/s"
465
466 def test_write_remote_overwrites_are_not_slower_than_first_write(self, tmp_path: pathlib.Path) -> None:
467 """Overwriting 500 records (second call) must not be dramatically slower."""
468 root = _make_repo(tmp_path)
469 write_remote = _import_write_remote()
470
471 records = [
472 {
473 "kind": _ALL_KINDS[i % len(_ALL_KINDS)],
474 "record_uuid": f"uuid-{i:06d}",
475 "run_id": f"run-{i}",
476 "payload": {"idx": i},
477 "expires_at": None,
478 }
479 for i in range(500)
480 ]
481
482 # First write — directories created
483 t0 = time.monotonic()
484 write_remote(root, records)
485 first = time.monotonic() - t0
486
487 # Second write — exist_ok=True, but still 500 mkdir() syscalls
488 t0 = time.monotonic()
489 write_remote(root, records)
490 second = time.monotonic() - t0
491
492 # Second should not be more than 5× the first (exist_ok adds a stat call)
493 assert second < max(first * 5, 2.0), (
494 f"overwrite pass ({second:.3f}s) is unexpectedly slow vs first write ({first:.3f}s)"
495 )
496
497
498 # =============================================================================
499 # 3. run_coord_gc — double-scan of releases, triple active_reservations scan
500 # =============================================================================
501
502 class TestCoordPerfGC:
503 """
504 run_coord_gc loads releases TWICE: load_released_ids() + load_all_releases().
505 These tests expose that cost at scale.
506 """
507
508 def test_gc_1000_expired_reservations_under_3s(self, tmp_path: pathlib.Path) -> None:
509 root = _make_repo(tmp_path)
510 _write_expired_reservations(root, 1000)
511 run_coord_gc = _import_run_coord_gc()
512
513 t0 = time.monotonic()
514 result = run_coord_gc(root, dry_run=False, grace_period_seconds=0)
515 elapsed = time.monotonic() - t0
516
517 assert result.reservations_removed == 1000
518 assert elapsed < 3.0, f"GC 1000 expired reservations took {elapsed:.3f}s (> 3s)"
519
520 def test_gc_1000_released_reservations_under_4s(self, tmp_path: pathlib.Path) -> None:
521 """
522 Released reservations force the double-scan: load_released_ids() then
523 load_all_releases(). 1 000 release tombstones = 2 × 1 000 file reads.
524 """
525 root = _make_repo(tmp_path)
526 _write_released_reservations(root, 1000)
527 run_coord_gc = _import_run_coord_gc()
528
529 t0 = time.monotonic()
530 result = run_coord_gc(root, dry_run=False, grace_period_seconds=0)
531 elapsed = time.monotonic() - t0
532
533 assert result.reservations_removed == 1000
534 assert elapsed < 4.0, (
535 f"GC 1000 released reservations took {elapsed:.3f}s (> 4s — double-scan cost)"
536 )
537
538 @pytest.mark.slow
539 def test_gc_5000_expired_reservations_under_15s(self, tmp_path: pathlib.Path) -> None:
540 root = _make_repo(tmp_path)
541 _write_expired_reservations(root, 5000)
542 run_coord_gc = _import_run_coord_gc()
543
544 t0 = time.monotonic()
545 result = run_coord_gc(root, dry_run=False, grace_period_seconds=0)
546 elapsed = time.monotonic() - t0
547
548 assert result.reservations_removed == 5000
549 assert elapsed < 15.0, f"GC 5000 expired reservations took {elapsed:.3f}s (> 15s)"
550
551 def test_gc_dry_run_is_not_slower_than_live_run(self, tmp_path: pathlib.Path) -> None:
552 """
553 Dry-run must not be significantly slower than a live run at the same scale.
554 Both run the same 4 directory scans; the only difference is no unlink().
555 """
556 root_live = _make_repo(tmp_path / "live")
557 root_dry = _make_repo(tmp_path / "dry")
558 _write_expired_reservations(root_live, 300)
559 _write_expired_reservations(root_dry, 300)
560 run_coord_gc = _import_run_coord_gc()
561
562 t0 = time.monotonic()
563 run_coord_gc(root_live, dry_run=False, grace_period_seconds=0)
564 live_elapsed = time.monotonic() - t0
565
566 t0 = time.monotonic()
567 run_coord_gc(root_dry, dry_run=True, grace_period_seconds=0)
568 dry_elapsed = time.monotonic() - t0
569
570 # Dry-run has no unlink() calls — it should not be slower than live run
571 assert dry_elapsed < max(live_elapsed * 3, 1.0), (
572 f"dry_run ({dry_elapsed:.3f}s) is unexpectedly slower than live ({live_elapsed:.3f}s)"
573 )
574
575 def test_gc_empty_repo_is_fast(self, tmp_path: pathlib.Path) -> None:
576 root = _make_repo(tmp_path)
577 _coord_dir(root)
578 run_coord_gc = _import_run_coord_gc()
579
580 t0 = time.monotonic()
581 for _ in range(50):
582 run_coord_gc(root, dry_run=True, grace_period_seconds=0)
583 elapsed = time.monotonic() - t0
584
585 assert elapsed < 0.500, f"50 × GC on empty repo took {elapsed:.3f}s (> 500ms)"
586
587 def test_gc_throughput_reservations_per_second(self, tmp_path: pathlib.Path) -> None:
588 root = _make_repo(tmp_path)
589 _write_expired_reservations(root, 500)
590 run_coord_gc = _import_run_coord_gc()
591
592 t0 = time.monotonic()
593 result = run_coord_gc(root, dry_run=False, grace_period_seconds=0)
594 elapsed = time.monotonic() - t0
595
596 rps = result.reservations_removed / elapsed if elapsed > 0 else float("inf")
597 assert rps >= 100, f"GC throughput {rps:.0f} reservations/s is below minimum 100/s"
598
599
600 # =============================================================================
601 # 4. active_reservations — triple directory scan
602 # =============================================================================
603
604 class TestCoordPerfActiveReservations:
605 """
606 active_reservations calls:
607 load_released_ids() — scan 1: releases/
608 load_heartbeat_map() — scan 2: heartbeats/
609 load_all_reservations() — scan 3: reservations/
610 At 1 000 live reservations this is 3 × 1 000 file reads.
611 """
612
613 def test_active_reservations_500_under_1_5s(self, tmp_path: pathlib.Path) -> None:
614 root = _make_repo(tmp_path)
615 _write_records(root, "reservation", 500)
616 active_reservations = _import_active_reservations()
617
618 t0 = time.monotonic()
619 result = active_reservations(root)
620 elapsed = time.monotonic() - t0
621
622 assert len(result) == 500
623 assert elapsed < 1.5, f"active_reservations(500) took {elapsed:.3f}s (> 1.5s)"
624
625 def test_active_reservations_1000_under_3s(self, tmp_path: pathlib.Path) -> None:
626 root = _make_repo(tmp_path)
627 _write_records(root, "reservation", 1000)
628 active_reservations = _import_active_reservations()
629
630 t0 = time.monotonic()
631 result = active_reservations(root)
632 elapsed = time.monotonic() - t0
633
634 assert len(result) == 1000
635 assert elapsed < 3.0, f"active_reservations(1000) took {elapsed:.3f}s (> 3s)"
636
637 @pytest.mark.slow
638 def test_active_reservations_3000_under_9s(self, tmp_path: pathlib.Path) -> None:
639 """3 000 reservations × 3 directory scans = ~9 000 file reads."""
640 root = _make_repo(tmp_path)
641 _write_records(root, "reservation", 3000)
642 active_reservations = _import_active_reservations()
643
644 t0 = time.monotonic()
645 result = active_reservations(root)
646 elapsed = time.monotonic() - t0
647
648 assert len(result) == 3000
649 assert elapsed < 9.0, f"active_reservations(3000) took {elapsed:.3f}s (> 9s — triple-scan)"
650
651 def test_active_reservations_with_heartbeats_not_slower_than_without(self, tmp_path: pathlib.Path) -> None:
652 """
653 Adding heartbeat files adds a second directory scan. Measure the
654 overhead explicitly to document the triple-scan cost.
655 """
656 root_no_hb = _make_repo(tmp_path / "no_hb")
657 root_hb = _make_repo(tmp_path / "hb")
658 _write_records(root_no_hb, "reservation", 200)
659 _write_records(root_hb, "reservation", 200)
660 _write_records(root_hb, "heartbeat", 200)
661 active_reservations = _import_active_reservations()
662
663 t0 = time.monotonic()
664 active_reservations(root_no_hb)
665 no_hb_elapsed = time.monotonic() - t0
666
667 t0 = time.monotonic()
668 active_reservations(root_hb)
669 hb_elapsed = time.monotonic() - t0
670
671 # With heartbeats should not be more than 5× slower (both are O(N))
672 assert hb_elapsed < max(no_hb_elapsed * 5, 1.0), (
673 f"active_reservations with heartbeats ({hb_elapsed:.3f}s) "
674 f"is unexpectedly slow vs without ({no_hb_elapsed:.3f}s)"
675 )
676
677 def test_active_reservations_throughput_per_second(self, tmp_path: pathlib.Path) -> None:
678 root = _make_repo(tmp_path)
679 _write_records(root, "reservation", 300)
680 active_reservations = _import_active_reservations()
681
682 # Warm up
683 active_reservations(root)
684
685 t0 = time.monotonic()
686 result = active_reservations(root)
687 elapsed = time.monotonic() - t0
688
689 rps = len(result) / elapsed if elapsed > 0 else float("inf")
690 assert rps >= 100, (
691 f"active_reservations throughput {rps:.0f} records/s is below minimum 100/s"
692 )
693
694
695 # =============================================================================
696 # 5. JSON payload scaling
697 # =============================================================================
698
699 class TestCoordPerfPayloadScaling:
700 """
701 Measures how _gather_local_records and _write_remote_records degrade as
702 payload sizes grow from 50 B to 50 KB per record.
703 """
704
705 @pytest.mark.parametrize("payload_bytes,n,max_seconds", [
706 (50, 500, 1.0),
707 (1024, 500, 2.0),
708 (10240, 200, 3.0),
709 (51200, 50, 3.0),
710 ])
711 def test_gather_payload_scaling(self, tmp_path: pathlib.Path, payload_bytes: int, n: int, max_seconds: float) -> None:
712 root = _make_repo(tmp_path)
713 _write_records(root, "reservation", n, payload_bytes=payload_bytes)
714 gather = _import_gather()
715
716 t0 = time.monotonic()
717 records = gather(root, ["reservation"])
718 elapsed = time.monotonic() - t0
719
720 assert len(records) == n
721 assert elapsed < max_seconds, (
722 f"gather {n} × {payload_bytes}B records took {elapsed:.3f}s (> {max_seconds}s)"
723 )
724
725 @pytest.mark.parametrize("payload_bytes,n,max_seconds", [
726 (50, 500, 2.0),
727 (1024, 500, 3.0),
728 (10240, 200, 3.0),
729 (51200, 50, 2.0),
730 ])
731 def test_write_remote_payload_scaling(self, tmp_path: pathlib.Path, payload_bytes: int, n: int, max_seconds: float) -> None:
732 root = _make_repo(tmp_path)
733 write_remote = _import_write_remote()
734
735 records = [
736 {
737 "kind": "reservation",
738 "record_uuid": f"uuid-{i:06d}",
739 "run_id": f"run-{i}",
740 "payload": {"data": "x" * payload_bytes},
741 "expires_at": _FUTURE_TS,
742 }
743 for i in range(n)
744 ]
745
746 t0 = time.monotonic()
747 write_remote(root, records)
748 elapsed = time.monotonic() - t0
749
750 assert elapsed < max_seconds, (
751 f"write_remote {n} × {payload_bytes}B records took {elapsed:.3f}s (> {max_seconds}s)"
752 )
753
754 def test_50kb_payload_does_not_cause_memory_explosion(self, tmp_path: pathlib.Path) -> None:
755 """
756 50 records × 50 KB payload = 2.5 MB. After the gather, the live heap
757 growth should stay well under 100 MB (not 2.5 GB from pathological
758 duplication).
759 """
760 root = _make_repo(tmp_path)
761 _write_records(root, "reservation", 50, payload_bytes=51200)
762 gather = _import_gather()
763
764 tracemalloc.start()
765 before = tracemalloc.take_snapshot()
766 gather(root, ["reservation"])
767 after = tracemalloc.take_snapshot()
768 tracemalloc.stop()
769
770 stats = after.compare_to(before, "lineno")
771 total_delta = sum(s.size_diff for s in stats if s.size_diff > 0)
772 assert total_delta < 100 * 1024 * 1024, (
773 f"50-record 50KB gather allocated {total_delta / 1024 / 1024:.1f}MB "
774 "(expected < 100MB)"
775 )
776
777
778 # =============================================================================
779 # 6. Batch loop pure overhead
780 # =============================================================================
781
782 class TestCoordPerfBatchLoop:
783 """
784 The push batch loop in run_push splits records into chunks of MAX_PUSH_BATCH
785 (500) and calls push_to_hub once per chunk. At 7 000 records that is 14
786 HTTP calls. These tests isolate the loop overhead from network cost.
787 """
788
789 def test_batch_loop_7000_records_14_calls_under_500ms(self, tmp_path: pathlib.Path) -> None:
790 """
791 With push_to_hub mocked, 14 batch calls on 7 000 records should
792 complete in < 500ms. Any overhead above that is pure Python CPU cost.
793 """
794 from muse.core.coord_bus import MAX_PUSH_BATCH
795 root = _make_repo(tmp_path)
796
797 # Write 1000 each of 7 kinds = 7000 total
798 for kind in _ALL_KINDS:
799 _write_records(root, kind, 1000)
800
801 call_count: list[int] = []
802
803 def fake_push(hub_url: str, owner: str, slug: str, records: list[JsonDict], signing: SigningIdentity | None = None) -> MsgpackDict:
804 call_count.append(len(records))
805 return {"inserted": len(records), "skipped": 0}
806
807 with patch("muse.cli.commands.coord_sync._gather_local_records") as mock_gather, \
808 patch("muse.cli.commands.coord_sync.push_to_hub", side_effect=fake_push), \
809 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
810 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
811 return_value=("http://localhost:10003", "tok")):
812
813 # Build 7000 records in memory for mock_gather
814 records = []
815 for i in range(7000):
816 kind = _ALL_KINDS[i % len(_ALL_KINDS)]
817 records.append({
818 "kind": kind,
819 "record_uuid": f"uuid-{i:06d}",
820 "run_id": f"run-{i}",
821 "payload": {"idx": i},
822 "expires_at": _FUTURE_TS,
823 })
824 mock_gather.return_value = records
825
826 import argparse
827 args = argparse.Namespace(
828 owner="torvalds", slug="linux",
829 fmt="json", token=None,
830 hub_url=None,
831 kinds=list(_ALL_KINDS),
832 )
833
834 t0 = time.monotonic()
835 try:
836 from muse.cli.commands.coord_sync import run_push
837 run_push(args)
838 except SystemExit as exc:
839 assert exc.code == 0, f"run_push exited with code {exc.code}"
840 elapsed = time.monotonic() - t0
841
842 expected_batches = (7000 + MAX_PUSH_BATCH - 1) // MAX_PUSH_BATCH
843 assert len(call_count) == expected_batches, (
844 f"expected {expected_batches} batches, got {len(call_count)}"
845 )
846 assert elapsed < 0.500, (
847 f"batch loop (push_to_hub mocked) for 7000 records took {elapsed:.3f}s (> 500ms)"
848 )
849
850 def test_batch_sizes_are_never_over_max_push_batch(self, tmp_path: pathlib.Path) -> None:
851 """Every batch must be ≤ MAX_PUSH_BATCH records."""
852 from muse.core.coord_bus import MAX_PUSH_BATCH
853
854 root = _make_repo(tmp_path)
855 call_sizes: list[int] = []
856
857 def fake_push(hub_url: str, owner: str, slug: str, records: list[JsonDict], signing: SigningIdentity | None = None) -> MsgpackDict:
858 call_sizes.append(len(records))
859 return {"inserted": len(records), "skipped": 0}
860
861 with patch("muse.cli.commands.coord_sync._gather_local_records") as mock_gather, \
862 patch("muse.cli.commands.coord_sync.push_to_hub", side_effect=fake_push), \
863 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
864 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
865 return_value=("http://localhost:10003", "tok")):
866
867 records = [
868 {
869 "kind": "reservation",
870 "record_uuid": f"uuid-{i:06d}",
871 "run_id": "run-0",
872 "payload": {},
873 "expires_at": None,
874 }
875 for i in range(3333) # not a clean multiple of 500
876 ]
877 mock_gather.return_value = records
878
879 import argparse
880 args = argparse.Namespace(
881 owner="torvalds", slug="linux",
882 fmt="json", token=None,
883 hub_url=None,
884 kinds=["reservation"],
885 )
886 try:
887 from muse.cli.commands.coord_sync import run_push
888 run_push(args)
889 except SystemExit:
890 pass
891
892 assert all(s <= MAX_PUSH_BATCH for s in call_sizes), (
893 f"batch sizes {call_sizes} contain a batch > MAX_PUSH_BATCH ({MAX_PUSH_BATCH})"
894 )
895
896 def test_last_batch_is_correct_remainder(self, tmp_path: pathlib.Path) -> None:
897 """With 7001 records (14 full + 1 leftover), last batch must be size 1."""
898 from muse.core.coord_bus import MAX_PUSH_BATCH
899
900 root = _make_repo(tmp_path)
901 call_sizes: list[int] = []
902
903 def fake_push(hub_url: str, owner: str, slug: str, records: list[JsonDict], signing: SigningIdentity | None = None) -> MsgpackDict:
904 call_sizes.append(len(records))
905 return {"inserted": len(records), "skipped": 0}
906
907 with patch("muse.cli.commands.coord_sync._gather_local_records") as mock_gather, \
908 patch("muse.cli.commands.coord_sync.push_to_hub", side_effect=fake_push), \
909 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
910 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
911 return_value=("http://localhost:10003", "tok")):
912
913 n = MAX_PUSH_BATCH * 14 + 1 # 7001
914 records = [
915 {
916 "kind": "reservation",
917 "record_uuid": f"uuid-{i:06d}",
918 "run_id": "run-0",
919 "payload": {},
920 "expires_at": None,
921 }
922 for i in range(n)
923 ]
924 mock_gather.return_value = records
925
926 import argparse
927 args = argparse.Namespace(
928 owner="torvalds", slug="linux",
929 fmt="json", token=None,
930 hub_url=None,
931 kinds=["reservation"],
932 )
933 try:
934 from muse.cli.commands.coord_sync import run_push
935 run_push(args)
936 except SystemExit:
937 pass
938
939 assert len(call_sizes) == 15, f"expected 15 batches for {n} records, got {len(call_sizes)}"
940 assert call_sizes[-1] == 1, f"last batch should be 1 record, got {call_sizes[-1]}"
941
942 def test_zero_records_no_http_calls(self, tmp_path: pathlib.Path) -> None:
943 """Empty gather must result in zero push_to_hub calls."""
944 root = _make_repo(tmp_path)
945 call_count: list[int] = []
946
947 def fake_push(hub_url: str, owner: str, slug: str, records: list[JsonDict], signing: SigningIdentity | None = None) -> MsgpackDict:
948 call_count.append(len(records))
949 return {"inserted": 0, "skipped": 0}
950
951 with patch("muse.cli.commands.coord_sync._gather_local_records",
952 return_value=[]), \
953 patch("muse.cli.commands.coord_sync.push_to_hub", side_effect=fake_push), \
954 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
955 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
956 return_value=("http://localhost:10003", "tok")):
957
958 import argparse
959 args = argparse.Namespace(
960 owner="torvalds", slug="linux",
961 fmt="json", token=None,
962 hub_url=None,
963 kinds=list(_ALL_KINDS),
964 )
965 try:
966 from muse.cli.commands.coord_sync import run_push
967 run_push(args)
968 except SystemExit:
969 pass
970
971 assert call_count == [], f"expected 0 HTTP calls for empty push, got {len(call_count)}"
972
973
974 # =============================================================================
975 # 7. Memory footprint
976 # =============================================================================
977
978 class TestCoordPerfMemory:
979 """
980 _gather_local_records holds all N records in a flat list before returning.
981 At 10 000 records with 10KB payloads, that is ~100 MB of in-process memory.
982 These tests document and bound the memory footprint.
983 """
984
985 def test_gather_1000_records_memory_under_50mb(self, tmp_path: pathlib.Path) -> None:
986 root = _make_repo(tmp_path)
987 _write_records(root, "reservation", 1000, payload_bytes=1024)
988 gather = _import_gather()
989
990 tracemalloc.start()
991 snap_before = tracemalloc.take_snapshot()
992 records = gather(root, ["reservation"])
993 snap_after = tracemalloc.take_snapshot()
994 tracemalloc.stop()
995
996 stats = snap_after.compare_to(snap_before, "lineno")
997 delta_bytes = sum(s.size_diff for s in stats if s.size_diff > 0)
998
999 assert len(records) == 1000
1000 assert delta_bytes < 50 * 1024 * 1024, (
1001 f"gather(1000 × 1KB) allocated {delta_bytes / 1024 / 1024:.1f}MB (> 50MB)"
1002 )
1003
1004 def test_gather_all_7_kinds_200_each_memory_under_100mb(self, tmp_path: pathlib.Path) -> None:
1005 """1400 records × 1KB payload = 1.4 MB on disk; heap delta should be < 100 MB."""
1006 root = _make_repo(tmp_path)
1007 for kind in _ALL_KINDS:
1008 _write_records(root, kind, 200, payload_bytes=1024)
1009 gather = _import_gather()
1010
1011 tracemalloc.start()
1012 snap_before = tracemalloc.take_snapshot()
1013 records = gather(root, list(_ALL_KINDS))
1014 snap_after = tracemalloc.take_snapshot()
1015 tracemalloc.stop()
1016
1017 stats = snap_after.compare_to(snap_before, "lineno")
1018 delta_bytes = sum(s.size_diff for s in stats if s.size_diff > 0)
1019
1020 assert len(records) == 1400
1021 assert delta_bytes < 100 * 1024 * 1024, (
1022 f"gather(1400 × 1KB, all kinds) allocated {delta_bytes / 1024 / 1024:.1f}MB (> 100MB)"
1023 )
1024
1025 def test_write_remote_1000_records_memory_under_100mb(self, tmp_path: pathlib.Path) -> None:
1026 root = _make_repo(tmp_path)
1027 write_remote = _import_write_remote()
1028
1029 records = [
1030 {
1031 "kind": _ALL_KINDS[i % len(_ALL_KINDS)],
1032 "record_uuid": f"uuid-{i:06d}",
1033 "run_id": f"run-{i}",
1034 "payload": {"data": "x" * 1024},
1035 "expires_at": _FUTURE_TS,
1036 }
1037 for i in range(1000)
1038 ]
1039
1040 tracemalloc.start()
1041 snap_before = tracemalloc.take_snapshot()
1042 write_remote(root, records)
1043 snap_after = tracemalloc.take_snapshot()
1044 tracemalloc.stop()
1045
1046 stats = snap_after.compare_to(snap_before, "lineno")
1047 delta_bytes = sum(s.size_diff for s in stats if s.size_diff > 0)
1048
1049 assert delta_bytes < 100 * 1024 * 1024, (
1050 f"write_remote(1000 × 1KB) allocated {delta_bytes / 1024 / 1024:.1f}MB (> 100MB)"
1051 )
1052
1053 def test_gc_1000_reservations_memory_under_50mb(self, tmp_path: pathlib.Path) -> None:
1054 root = _make_repo(tmp_path)
1055 _write_expired_reservations(root, 1000)
1056 run_coord_gc = _import_run_coord_gc()
1057
1058 tracemalloc.start()
1059 snap_before = tracemalloc.take_snapshot()
1060 run_coord_gc(root, dry_run=False, grace_period_seconds=0)
1061 snap_after = tracemalloc.take_snapshot()
1062 tracemalloc.stop()
1063
1064 stats = snap_after.compare_to(snap_before, "lineno")
1065 delta_bytes = sum(s.size_diff for s in stats if s.size_diff > 0)
1066
1067 assert delta_bytes < 50 * 1024 * 1024, (
1068 f"GC(1000 expired) allocated {delta_bytes / 1024 / 1024:.1f}MB (> 50MB)"
1069 )
1070
1071
1072 # =============================================================================
1073 # 8. End-to-end throughput
1074 # =============================================================================
1075
1076 class TestCoordPerfThroughput:
1077 """
1078 End-to-end gather → batch → push cycle with a mock HTTP layer.
1079 Measures total records/second and total wall-clock time.
1080 """
1081
1082 def test_end_to_end_500_records_under_1s(self, tmp_path: pathlib.Path) -> None:
1083 root = _make_repo(tmp_path)
1084 for kind in _ALL_KINDS:
1085 _write_records(root, kind, 500 // len(_ALL_KINDS) + 1)
1086
1087 call_count = []
1088
1089 def fake_push(hub_url: str, owner: str, slug: str, records: list[JsonDict], signing: SigningIdentity | None = None) -> MsgpackDict:
1090 call_count.append(len(records))
1091 return {"inserted": len(records), "skipped": 0}
1092
1093 with patch("muse.cli.commands.coord_sync.push_to_hub", side_effect=fake_push), \
1094 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
1095 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
1096 return_value=("http://localhost:10003", "tok")):
1097
1098 import argparse
1099 args = argparse.Namespace(
1100 owner="torvalds", slug="linux",
1101 fmt="json", token=None,
1102 hub_url=None,
1103 kinds=list(_ALL_KINDS),
1104 )
1105
1106 t0 = time.monotonic()
1107 try:
1108 from muse.cli.commands.coord_sync import run_push
1109 run_push(args)
1110 except SystemExit as exc:
1111 assert exc.code == 0
1112
1113 elapsed = time.monotonic() - t0
1114
1115 total_pushed = sum(call_count)
1116 assert total_pushed > 0
1117 assert elapsed < 1.0, f"end-to-end ~500 records took {elapsed:.3f}s (> 1s)"
1118
1119 @pytest.mark.slow
1120 def test_end_to_end_7000_records_under_10s(self, tmp_path: pathlib.Path) -> None:
1121 """
1122 Full Linux kernel agent swarm: 7 000 records across all 7 kinds.
1123 push_to_hub is mocked — this measures gather + batch splitting overhead only.
1124 """
1125 root = _make_repo(tmp_path)
1126 for kind in _ALL_KINDS:
1127 _write_records(root, kind, 1000)
1128
1129 call_count = []
1130
1131 def fake_push(hub_url: str, owner: str, slug: str, records: list[JsonDict], signing: SigningIdentity | None = None) -> MsgpackDict:
1132 call_count.append(len(records))
1133 return {"inserted": len(records), "skipped": 0}
1134
1135 with patch("muse.cli.commands.coord_sync.push_to_hub", side_effect=fake_push), \
1136 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
1137 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
1138 return_value=("http://localhost:10003", "tok")):
1139
1140 import argparse
1141 args = argparse.Namespace(
1142 owner="torvalds", slug="linux",
1143 fmt="json", token=None,
1144 hub_url=None,
1145 kinds=list(_ALL_KINDS),
1146 )
1147
1148 t0 = time.monotonic()
1149 try:
1150 from muse.cli.commands.coord_sync import run_push
1151 run_push(args)
1152 except SystemExit as exc:
1153 assert exc.code == 0
1154
1155 elapsed = time.monotonic() - t0
1156
1157 total_pushed = sum(call_count)
1158 assert total_pushed == 7000, f"expected 7000 records pushed, got {total_pushed}"
1159 assert elapsed < 10.0, (
1160 f"end-to-end 7000 records (gather + batch, mocked HTTP) took {elapsed:.3f}s (> 10s)"
1161 )
1162
1163 rps = total_pushed / elapsed
1164 assert rps >= 700, f"end-to-end throughput {rps:.0f} records/s is below minimum 700/s"
1165
1166 def test_repeated_pulls_do_not_accumulate_state(self, tmp_path: pathlib.Path) -> None:
1167 """
1168 run_pull is stateless per call — repeated calls on the same repo should
1169 have stable (not growing) elapsed time.
1170 """
1171 root = _make_repo(tmp_path)
1172 (root / ".muse" / "coordination" / _REMOTE_DIR_NAME).mkdir(parents=True, exist_ok=True)
1173
1174 fake_records = [
1175 {
1176 "kind": "reservation",
1177 "record_uuid": f"uuid-{i:06d}",
1178 "run_id": "run-0",
1179 "payload": {"idx": i},
1180 "expires_at": _FUTURE_TS,
1181 }
1182 for i in range(200)
1183 ]
1184
1185 def fake_pull(
1186 hub_url: str,
1187 owner: str,
1188 slug: str,
1189 since_id: int = 0,
1190 kinds: list[str] | None = None,
1191 limit: int = 500,
1192 signing: SigningIdentity | None = None,
1193 ) -> MsgpackDict:
1194 return {"records": fake_records, "cursor": since_id + len(fake_records)}
1195
1196 with patch("muse.cli.commands.coord_sync.pull_from_hub", side_effect=fake_pull), \
1197 patch("muse.cli.commands.coord_sync.require_repo", return_value=root), \
1198 patch("muse.cli.commands.coord_sync._resolve_hub_and_signing",
1199 return_value=("http://localhost:10003", "tok")):
1200
1201 import argparse
1202
1203 elapsed_list = []
1204 for _ in range(5):
1205 args = argparse.Namespace(
1206 owner="torvalds", slug="linux",
1207 fmt="json", token=None,
1208 hub_url=None,
1209 since_id=0, limit=1000,
1210 kinds=list(_ALL_KINDS),
1211 )
1212 t0 = time.monotonic()
1213 try:
1214 from muse.cli.commands.coord_sync import run_pull
1215 run_pull(args)
1216 except SystemExit:
1217 pass
1218 elapsed_list.append(time.monotonic() - t0)
1219
1220 # The 5th call must not take more than 5× the 1st call — no accumulation
1221 assert elapsed_list[4] < max(elapsed_list[0] * 5, 0.500), (
1222 f"repeated pull times grew suspiciously: {[f'{e:.3f}' for e in elapsed_list]}"
1223 )
1224
1225
1226 _REMOTE_DIR_NAME = "remote"
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 101 days ago