gabriel / muse public
test_cmd_watch_coord.py python
1,531 lines 69.3 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Tests for ``muse coord watch`` (watch_coord.py).
2
3 Coverage matrix
4 ---------------
5 Unit
6 TestWatchEvent — to_dict() schema, all fields present
7 TestScanDirs — empty dir, missing dir, multiple kinds, TOCTOU safety
8 TestDiffSnapshots — no change, added, removed, modified, cross-kind
9 TestPassesFilters — kind/run_id/branch filters, AND semantics, empty data
10 TestEmitEventJson — JSON mode produces valid NDJSON, all event types
11 TestEmitEventText — text mode, all kinds, missing data, ANSI stripped
12 TestCheckExpirations — expired emitted, active not, removed not double-counted
13 TestMakeEvent — timestamp is UTC ISO 8601
14
15 Integration
16 TestWatchLoopOnce — --once emits snapshot for all existing records
17 TestWatchLoopFilters — kind/run_id/branch filters in loop
18 TestWatchLoopDetectsAdded — new reservation appears → added event
19 TestWatchLoopDetectsModified — heartbeat updated → modified event
20 TestWatchLoopDetectsRemoved — file deleted → removed event (with cached data)
21 TestWatchLoopDetectsExpiry — reservation expires → expired event (no file change)
22 TestWatchLoopTimeout — loop exits after timeout seconds
23 TestWatchLoopJsonOutput — all events are valid NDJSON in json mode
24 TestWatchLoopMaxEvents — loop exits after --max-events events emitted
25
26 End-to-end
27 TestRunCommand — run() with --once via mock require_repo
28 TestRunCommandValidation — run() validation: run-id/poll-interval/timeout/max-events
29 TestBackendSelection — kqueue backend on macOS, polling on other
30 TestSignalHandling — SIGTERM triggers clean exit
31
32 Security
33 TestAnsiInjectionSanitized — ANSI escapes in run_id/branch stripped from text
34 TestSymlinkDirRejected — kqueue raises ValueError on symlinked coord dir
35 TestKindFilterAllowlist — invalid kind strings rejected by argparse
36
37 Stress
38 TestStress500RapidAdds — 500 reservations added → all 500 added events
39 TestStressLargeSnapshot — 1000 existing records → snapshot < 1 s
40 TestStressDiffPerformance — diff of 2×1000-entry snapshots < 50 ms
41 TestStressManyExpirations — 200 expiring reservations → all expired events < 2 s
42 TestStressMaxEvents — max_events cap respected under 500-record load
43 """
44
45 from __future__ import annotations
46
47 import argparse
48 import datetime
49 import io
50 import json
51 import pathlib
52 import tempfile
53 import time
54 import uuid
55 from collections.abc import Callable
56 from contextlib import redirect_stdout
57 from unittest.mock import MagicMock, patch
58
59 import pytest
60
61 from muse.core._types import Manifest, MsgpackDict, MsgpackValue
62
63 # Module under test.
64 from muse.cli.commands.watch_coord import (
65 WatchEvent,
66 _Backend,
67 _KqueueBackend,
68 _MAX_RUN_ID_LEN,
69 _MIN_POLL_INTERVAL,
70 _MAX_POLL_INTERVAL,
71 _PollingBackend,
72 _Snapshot,
73 _check_expirations,
74 _coord_dir,
75 _diff_snapshots,
76 _emit_event,
77 _ensure_coord_dirs,
78 _load_record,
79 _make_event,
80 _passes_filters,
81 _scan_dirs,
82 _watch_loop,
83 )
84 from muse.core.coordination import (
85 Reservation,
86 create_heartbeat,
87 create_intent,
88 create_reservation,
89 )
90 from muse.core.errors import ExitCode
91
92
93 # ─────────────────────────────────────────────────────────────────────────────
94 # Fixtures
95 # ─────────────────────────────────────────────────────────────────────────────
96
97
98 @pytest.fixture
99 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
100 """Minimal muse repository with coord dirs created."""
101 muse_dir = tmp_path / ".muse"
102 muse_dir.mkdir()
103 (muse_dir / "repo.json").write_text(
104 json.dumps({"repo_id": str(uuid.uuid4()), "name": "test-repo"})
105 )
106 _ensure_coord_dirs(tmp_path)
107 return tmp_path
108
109
110 class _ImmediateBackend(_Backend):
111 """Test backend that fires a side-effect once then sleeps so the timeout fires.
112
113 Design: ``wait()`` calls the side_effect on the first invocation (so files
114 can be written between the snapshot and the next scan), then sleeps for
115 ``timeout`` on all subsequent calls. This means the outer ``_watch_loop``
116 will do exactly one meaningful diff iteration (the one that sees the
117 side-effect's changes) and then block until the timeout deadline expires.
118
119 Always pass ``timeout`` (a finite deadline) when using ``once=False``.
120 """
121
122 name = "immediate"
123
124 def __init__(
125 self,
126 *,
127 side_effect: Callable[[], None] | None = None,
128 ) -> None:
129 self._side_effect = side_effect
130 self._fired = False
131
132 def wait(self, timeout: float) -> bool:
133 if not self._fired:
134 if self._side_effect:
135 self._side_effect()
136 self._fired = True
137 return True # Signal: something may have changed.
138 # Already fired — sleep so the loop deadline can expire.
139 time.sleep(max(0.0, timeout))
140 return False
141
142 def close(self) -> None:
143 pass
144
145
146 def _make_reservation(repo: pathlib.Path, **kwargs: MsgpackValue) -> Reservation:
147 """Create a reservation with sensible defaults."""
148 return create_reservation(
149 repo,
150 run_id=kwargs.get("run_id", "agent-1"),
151 branch=kwargs.get("branch", "main"),
152 addresses=kwargs.get("addresses", ["src/mod.py::fn"]),
153 ttl_seconds=kwargs.get("ttl_seconds", 3600),
154 operation=kwargs.get("operation", "modify"),
155 )
156
157
158 def _run_loop(
159 repo: pathlib.Path,
160 *,
161 side_effect: "callable | None" = None,
162 kind_filter: str | None = None,
163 run_id_filter: str | None = None,
164 branch_filter: str | None = None,
165 as_json: bool = True,
166 once: bool = True,
167 # When once=False, a finite timeout is required so the loop exits.
168 # Default 1.0 s is enough for one side-effect iteration.
169 timeout: float | None = None,
170 poll_interval: float = 0.05,
171 max_events: int | None = None,
172 ) -> list[dict]:
173 """Run _watch_loop with an ImmediateBackend; return parsed events.
174
175 For ``once=False`` tests, pass ``timeout`` explicitly (e.g. ``timeout=1.0``).
176 The backend fires the side_effect on the first wait, then sleeps until the
177 deadline, giving the loop exactly one diff iteration to detect changes.
178 """
179 # Enforce finite timeout for non-once mode to prevent hangs.
180 if not once and timeout is None:
181 timeout = 1.0
182 buf = io.StringIO()
183 backend = _ImmediateBackend(side_effect=side_effect)
184 with redirect_stdout(buf):
185 _watch_loop(
186 repo,
187 backend,
188 kind_filter=kind_filter,
189 run_id_filter=run_id_filter,
190 branch_filter=branch_filter,
191 as_json=as_json,
192 once=once,
193 timeout=timeout,
194 poll_interval=poll_interval,
195 max_events=max_events,
196 )
197 lines = [l for l in buf.getvalue().splitlines() if l.strip()]
198 return [json.loads(line) for line in lines]
199
200
201 # ─────────────────────────────────────────────────────────────────────────────
202 # Unit — WatchEvent
203 # ─────────────────────────────────────────────────────────────────────────────
204
205
206 class TestWatchEvent:
207 def test_to_dict_has_all_fields(self) -> None:
208 ev = WatchEvent("added", "reservation", "uid-1", "2026-01-01T00:00:00+00:00", {})
209 d = ev.to_dict()
210 assert set(d.keys()) == {
211 "schema_version", "event_type", "kind", "id", "timestamp", "data"
212 }
213
214 def test_to_dict_values_round_trip(self) -> None:
215 payload = {"run_id": "a", "branch": "main"}
216 ev = WatchEvent("modified", "heartbeat", "uid-2", "2026-01-01T00:00:00+00:00", payload)
217 d = ev.to_dict()
218 assert d["event_type"] == "modified"
219 assert d["kind"] == "heartbeat"
220 assert d["id"] == "uid-2"
221 assert d["data"] == payload
222
223 def test_to_dict_schema_version_is_string(self) -> None:
224 ev = WatchEvent("snapshot", "intent", "uid-3", "ts", {})
225 assert isinstance(ev.to_dict()["schema_version"], str)
226
227
228 # ─────────────────────────────────────────────────────────────────────────────
229 # Unit — _scan_dirs
230 # ─────────────────────────────────────────────────────────────────────────────
231
232
233 class TestScanDirs:
234 def test_empty_coord_dirs_returns_empty_dicts(self, repo: pathlib.Path) -> None:
235 snap = _scan_dirs(repo)
236 for sub in ("reservations", "intents", "releases", "heartbeats"):
237 assert snap[sub] == {}
238
239 def test_missing_subdir_returns_empty(self, tmp_path: pathlib.Path) -> None:
240 # No .muse/ dir at all.
241 snap = _scan_dirs(tmp_path)
242 for sub in ("reservations", "intents", "releases", "heartbeats"):
243 assert snap[sub] == {}
244
245 def test_file_appears_in_snapshot(self, repo: pathlib.Path) -> None:
246 path = _coord_dir(repo) / "reservations" / "abc123.json"
247 path.write_text('{"x": 1}')
248 snap = _scan_dirs(repo)
249 assert "abc123" in snap["reservations"]
250
251 def test_snapshot_entry_is_mtime_ns_and_size(self, repo: pathlib.Path) -> None:
252 path = _coord_dir(repo) / "reservations" / "abc123.json"
253 path.write_text('{"x": 1}')
254 snap = _scan_dirs(repo)
255 mtime_ns, size = snap["reservations"]["abc123"]
256 st = path.stat()
257 assert mtime_ns == st.st_mtime_ns
258 assert size == st.st_size
259
260 def test_non_json_files_ignored(self, repo: pathlib.Path) -> None:
261 (_coord_dir(repo) / "reservations" / "not-a-json.txt").write_text("hi")
262 snap = _scan_dirs(repo)
263 assert snap["reservations"] == {}
264
265 def test_multiple_kinds_all_scanned(self, repo: pathlib.Path) -> None:
266 (_coord_dir(repo) / "reservations" / "r1.json").write_text("{}")
267 (_coord_dir(repo) / "intents" / "i1.json").write_text("{}")
268 snap = _scan_dirs(repo)
269 assert "r1" in snap["reservations"]
270 assert "i1" in snap["intents"]
271
272 def test_toctou_deleted_file_skipped(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
273 """File that disappears between glob and stat is silently skipped."""
274 real_glob = pathlib.Path.glob
275
276 def patched_glob(self: pathlib.Path, pattern: str) -> list[pathlib.Path]:
277 results = list(real_glob(self, pattern))
278 # Inject a ghost path that does not exist.
279 ghost = self / "ghost.json"
280 return results + [ghost]
281
282 monkeypatch.setattr(pathlib.Path, "glob", patched_glob)
283 snap = _scan_dirs(repo)
284 assert "ghost" not in snap["reservations"]
285
286
287 # ─────────────────────────────────────────────────────────────────────────────
288 # Unit — _diff_snapshots
289 # ─────────────────────────────────────────────────────────────────────────────
290
291
292 class TestDiffSnapshots:
293 def _empty(self) -> _Snapshot:
294 return {sub: {} for sub in ("reservations", "intents", "releases", "heartbeats")}
295
296 def test_no_change_no_events(self) -> None:
297 snap = self._empty()
298 snap["reservations"]["uid1"] = (100, 50)
299 assert _diff_snapshots(snap, snap) == []
300
301 def test_added_detected(self) -> None:
302 old = self._empty()
303 new = {**old, "reservations": {"uid1": (100, 50)}}
304 events = _diff_snapshots(old, new)
305 assert ("added", "reservations", "uid1") in events
306
307 def test_removed_detected(self) -> None:
308 old = {**self._empty(), "reservations": {"uid1": (100, 50)}}
309 new = self._empty()
310 events = _diff_snapshots(old, new)
311 assert ("removed", "reservations", "uid1") in events
312
313 def test_modified_detected(self) -> None:
314 snap = {**self._empty(), "reservations": {"uid1": (100, 50)}}
315 new_snap = {**self._empty(), "reservations": {"uid1": (200, 51)}}
316 events = _diff_snapshots(snap, new_snap)
317 assert ("modified", "reservations", "uid1") in events
318
319 def test_same_mtime_different_size_is_modified(self) -> None:
320 snap = {**self._empty(), "reservations": {"uid1": (100, 50)}}
321 new_snap = {**self._empty(), "reservations": {"uid1": (100, 99)}}
322 events = _diff_snapshots(snap, new_snap)
323 assert ("modified", "reservations", "uid1") in events
324
325 def test_unchanged_mtime_and_size_is_not_modified(self) -> None:
326 snap = {**self._empty(), "reservations": {"uid1": (100, 50)}}
327 events = _diff_snapshots(snap, snap)
328 assert not any(e[0] == "modified" for e in events)
329
330 def test_multiple_kinds_in_one_diff(self) -> None:
331 old = {**self._empty(), "reservations": {"r1": (1, 10)}}
332 new = {
333 "reservations": {},
334 "intents": {"i1": (2, 20)},
335 "releases": {},
336 "heartbeats": {},
337 }
338 events = _diff_snapshots(old, new)
339 event_types = {(e[0], e[2]) for e in events}
340 assert ("removed", "r1") in event_types
341 assert ("added", "i1") in event_types
342
343 def test_output_is_sorted_deterministic(self) -> None:
344 """UUIDs within each kind are sorted for reproducible test output."""
345 old = self._empty()
346 new = {
347 **self._empty(),
348 "reservations": {"zzz": (1, 1), "aaa": (2, 2), "mmm": (3, 3)},
349 }
350 events = _diff_snapshots(old, new)
351 added_ids = [uid for et, kind, uid in events if et == "added"]
352 assert added_ids == sorted(added_ids)
353
354
355 # ─────────────────────────────────────────────────────────────────────────────
356 # Unit — _passes_filters
357 # ─────────────────────────────────────────────────────────────────────────────
358
359
360 class TestPassesFilters:
361 def _data(self, run_id: str = "agent-1", branch: str = "main") -> Manifest:
362 return {"run_id": run_id, "branch": branch}
363
364 def test_no_filters_always_passes(self) -> None:
365 assert _passes_filters("reservation", self._data(), None, None, None)
366
367 def test_kind_filter_match(self) -> None:
368 assert _passes_filters("reservation", self._data(), "reservation", None, None)
369
370 def test_kind_filter_no_match(self) -> None:
371 assert not _passes_filters("intent", self._data(), "reservation", None, None)
372
373 def test_run_id_filter_match(self) -> None:
374 assert _passes_filters("reservation", self._data(), None, "agent-1", None)
375
376 def test_run_id_filter_no_match(self) -> None:
377 assert not _passes_filters("reservation", self._data(), None, "agent-9", None)
378
379 def test_branch_filter_match(self) -> None:
380 assert _passes_filters("reservation", self._data(), None, None, "main")
381
382 def test_branch_filter_no_match(self) -> None:
383 assert not _passes_filters("reservation", self._data(), None, None, "feature/x")
384
385 def test_all_filters_and_semantics(self) -> None:
386 """All three filters must pass simultaneously."""
387 d = self._data("agent-1", "main")
388 assert _passes_filters("reservation", d, "reservation", "agent-1", "main")
389 assert not _passes_filters("reservation", d, "reservation", "agent-1", "other")
390
391 def test_empty_data_fails_run_id_filter(self) -> None:
392 """Empty data (removed event with no cache) fails run_id/branch filters."""
393 assert not _passes_filters("reservation", {}, None, "agent-1", None)
394
395 def test_empty_data_passes_when_no_filters(self) -> None:
396 assert _passes_filters("reservation", {}, None, None, None)
397
398
399 # ─────────────────────────────────────────────────────────────────────────────
400 # Unit — _emit_event JSON mode
401 # ─────────────────────────────────────────────────────────────────────────────
402
403
404 class TestEmitEventJson:
405 def _capture_json(self, ev: WatchEvent) -> MsgpackDict:
406 buf = io.StringIO()
407 with redirect_stdout(buf):
408 _emit_event(ev, as_json=True)
409 return json.loads(buf.getvalue().strip())
410
411 def test_json_is_valid_ndjson(self) -> None:
412 ev = WatchEvent("added", "reservation", "uid1", "2026-01-01T00:00:00+00:00",
413 {"run_id": "a", "branch": "b"})
414 d = self._capture_json(ev)
415 assert d["event_type"] == "added"
416 assert d["kind"] == "reservation"
417
418 def test_all_event_types_serialise(self) -> None:
419 for et in ("snapshot", "added", "modified", "removed", "expired"):
420 ev = WatchEvent(et, "intent", "uid2", "ts", {})
421 d = self._capture_json(ev)
422 assert d["event_type"] == et
423
424 def test_data_payload_preserved(self) -> None:
425 payload = {"run_id": "x", "addresses": ["f.py::fn"], "branch": "dev"}
426 ev = WatchEvent("added", "reservation", "uid3", "ts", payload)
427 d = self._capture_json(ev)
428 assert d["data"] == payload
429
430
431 # ─────────────────────────────────────────────────────────────────────────────
432 # Unit — _emit_event text mode
433 # ─────────────────────────────────────────────────────────────────────────────
434
435
436 class TestEmitEventText:
437 def _capture_text(self, ev: WatchEvent) -> str:
438 buf = io.StringIO()
439 with redirect_stdout(buf):
440 _emit_event(ev, as_json=False)
441 return buf.getvalue()
442
443 def test_reservation_icon_and_kind(self) -> None:
444 ev = WatchEvent("added", "reservation", "a" * 8, "ts",
445 {"run_id": "r1", "branch": "main", "addresses": ["f.py::fn"]})
446 out = self._capture_text(ev)
447 assert "+" in out
448 assert "reservation" in out
449
450 def test_snapshot_icon(self) -> None:
451 ev = WatchEvent("snapshot", "reservation", "a" * 8, "ts",
452 {"run_id": "r1", "branch": "main", "addresses": ["f.py::fn"]})
453 out = self._capture_text(ev)
454 assert "·" in out
455
456 def test_expired_icon(self) -> None:
457 ev = WatchEvent("expired", "reservations", "a" * 8, "ts",
458 {"run_id": "r1", "branch": "main", "addresses": ["f.py::fn"]})
459 out = self._capture_text(ev)
460 assert "!" in out
461
462 def test_empty_data_does_not_crash(self) -> None:
463 ev = WatchEvent("removed", "reservation", "b" * 8, "ts", {})
464 out = self._capture_text(ev)
465 assert "bbbbbbbb" in out
466
467 def test_heartbeat_shows_extends_to(self) -> None:
468 ev = WatchEvent("modified", "heartbeat", "c" * 8, "ts",
469 {"run_id": "r1", "reservation_id": "d" * 8,
470 "extended_expires_at": "2099-01-01T00:00:00+00:00"})
471 out = self._capture_text(ev)
472 assert "extends to" in out
473
474 def test_release_shows_reason(self) -> None:
475 ev = WatchEvent("added", "release", "e" * 8, "ts",
476 {"run_id": "r1", "reason": "completed",
477 "reservation_id": "f" * 8})
478 out = self._capture_text(ev)
479 assert "completed" in out
480
481 def test_intent_shows_operation(self) -> None:
482 ev = WatchEvent("added", "intent", "g" * 8, "ts",
483 {"run_id": "r1", "branch": "main", "operation": "delete",
484 "addresses": ["src/mod.py::fn"]})
485 out = self._capture_text(ev)
486 assert "delete" in out
487
488 def test_ansi_escape_stripped_from_run_id(self) -> None:
489 """ANSI escape in run_id must not reach terminal output."""
490 evil_run_id = "\x1b[31mevil\x1b[0m"
491 ev = WatchEvent("added", "reservation", "h" * 8, "ts",
492 {"run_id": evil_run_id, "branch": "main",
493 "addresses": ["f.py::fn"]})
494 out = self._capture_text(ev)
495 assert "\x1b" not in out
496 assert "evil" in out # Content preserved, escapes stripped.
497
498 def test_long_address_list_truncated(self) -> None:
499 addrs = [f"f.py::fn{i}" for i in range(10)]
500 ev = WatchEvent("added", "reservation", "i" * 8, "ts",
501 {"run_id": "r1", "branch": "main", "addresses": addrs})
502 out = self._capture_text(ev)
503 assert "+7 more" in out
504
505
506 # ─────────────────────────────────────────────────────────────────────────────
507 # Unit — _check_expirations
508 # ─────────────────────────────────────────────────────────────────────────────
509
510
511 class TestCheckExpirations:
512 def test_no_change_no_events(self, repo: pathlib.Path) -> None:
513 res = _make_reservation(repo)
514 active_ids = {res.reservation_id}
515 events, curr = _check_expirations(repo, active_ids, set())
516 assert events == []
517 assert res.reservation_id in curr
518
519 def test_expired_reservation_emits_event(self, repo: pathlib.Path) -> None:
520 """An ID that was active but is no longer → expired event."""
521 gone_id = str(uuid.uuid4())
522 active_ids = {gone_id}
523 # Create the file with an explicit past expiry so _load_record can find it
524 # and active_reservations() correctly excludes it.
525 past_iso = (
526 datetime.datetime.now(datetime.timezone.utc)
527 - datetime.timedelta(hours=2)
528 ).isoformat()
529 now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
530 path = _coord_dir(repo) / "reservations" / f"{gone_id}.json"
531 path.write_text(json.dumps({
532 "reservation_id": gone_id,
533 "run_id": "r1",
534 "branch": "main",
535 "addresses": ["src/mod.py::fn"],
536 "created_at": now_iso,
537 "expires_at": past_iso,
538 "operation": "modify",
539 "schema_version": "1.0.0",
540 }))
541 events, curr = _check_expirations(repo, active_ids, set())
542 assert any(e.event_type == "expired" and e.id == gone_id for e in events)
543 assert gone_id not in curr
544
545 def test_removed_id_not_double_counted(self, repo: pathlib.Path) -> None:
546 """ID in removed_ids must NOT produce an expired event too."""
547 gone_id = str(uuid.uuid4())
548 events, _ = _check_expirations(repo, {gone_id}, {gone_id})
549 assert events == []
550
551 def test_active_reservation_stays_active(self, repo: pathlib.Path) -> None:
552 """Live reservation is NOT reported as expired."""
553 res = _make_reservation(repo)
554 prev = {res.reservation_id}
555 events, curr = _check_expirations(repo, prev, set())
556 assert not any(e.event_type == "expired" for e in events)
557
558 def test_returns_current_active_ids(self, repo: pathlib.Path) -> None:
559 res = _make_reservation(repo)
560 _, curr = _check_expirations(repo, set(), set())
561 assert res.reservation_id in curr
562
563
564 # ─────────────────────────────────────────────────────────────────────────────
565 # Unit — _make_event
566 # ─────────────────────────────────────────────────────────────────────────────
567
568
569 class TestMakeEvent:
570 def test_timestamp_is_utc_iso8601(self) -> None:
571 ev = _make_event("added", "reservation", "uid1", {})
572 # Must parse as UTC datetime.
573 dt = datetime.datetime.fromisoformat(ev.timestamp)
574 assert dt.tzinfo is not None
575
576 def test_fields_set_correctly(self) -> None:
577 ev = _make_event("removed", "intent", "uid2", {"x": 1})
578 assert ev.event_type == "removed"
579 assert ev.kind == "intent"
580 assert ev.id == "uid2"
581 assert ev.data == {"x": 1}
582
583
584 # ─────────────────────────────────────────────────────────────────────────────
585 # Integration — _watch_loop --once
586 # ─────────────────────────────────────────────────────────────────────────────
587
588
589 class TestWatchLoopOnce:
590 def test_empty_repo_no_events(self, repo: pathlib.Path) -> None:
591 events = _run_loop(repo, once=True)
592 assert events == []
593
594 def test_existing_reservation_emits_snapshot(self, repo: pathlib.Path) -> None:
595 _make_reservation(repo, run_id="agent-1")
596 events = _run_loop(repo, once=True)
597 kinds = [e["kind"] for e in events]
598 assert "reservations" in kinds
599
600 def test_snapshot_event_type(self, repo: pathlib.Path) -> None:
601 _make_reservation(repo)
602 events = _run_loop(repo, once=True)
603 assert all(e["event_type"] == "snapshot" for e in events)
604
605 def test_all_kinds_emitted_in_snapshot(self, repo: pathlib.Path) -> None:
606 res = _make_reservation(repo)
607 create_intent(repo, res.reservation_id, "agent-1", "main",
608 ["src/mod.py::fn"], "modify")
609 events = _run_loop(repo, once=True)
610 kinds = {e["kind"] for e in events}
611 assert "reservations" in kinds
612 assert "intents" in kinds
613
614 def test_snapshot_carries_data(self, repo: pathlib.Path) -> None:
615 _make_reservation(repo, run_id="agent-snapshot-test")
616 events = _run_loop(repo, once=True)
617 res_events = [e for e in events if e["kind"] == "reservations"]
618 assert any(e["data"].get("run_id") == "agent-snapshot-test" for e in res_events)
619
620 def test_once_does_not_loop(self, repo: pathlib.Path) -> None:
621 """--once must return quickly (no blocking wait calls)."""
622 start = time.monotonic()
623 _run_loop(repo, once=True)
624 elapsed = time.monotonic() - start
625 assert elapsed < 1.0 # Must not block for poll_interval.
626
627
628 # ─────────────────────────────────────────────────────────────────────────────
629 # Integration — filters in _watch_loop
630 # ─────────────────────────────────────────────────────────────────────────────
631
632
633 class TestWatchLoopFilters:
634 def test_kind_filter_reservations(self, repo: pathlib.Path) -> None:
635 res = _make_reservation(repo)
636 create_intent(repo, res.reservation_id, "agent-1", "main",
637 ["src/mod.py::fn"], "modify")
638 events = _run_loop(repo, once=True, kind_filter="reservations")
639 assert all(e["kind"] == "reservations" for e in events)
640
641 def test_kind_filter_intents(self, repo: pathlib.Path) -> None:
642 res = _make_reservation(repo)
643 create_intent(repo, res.reservation_id, "agent-1", "main",
644 ["src/mod.py::fn"], "modify")
645 events = _run_loop(repo, once=True, kind_filter="intents")
646 assert all(e["kind"] == "intents" for e in events)
647
648 def test_run_id_filter(self, repo: pathlib.Path) -> None:
649 _make_reservation(repo, run_id="agent-A")
650 _make_reservation(repo, run_id="agent-B")
651 events = _run_loop(repo, once=True, run_id_filter="agent-A")
652 assert all(e["data"].get("run_id") == "agent-A" for e in events)
653 assert len(events) == 1
654
655 def test_branch_filter(self, repo: pathlib.Path) -> None:
656 _make_reservation(repo, branch="main")
657 _make_reservation(repo, branch="feature/x")
658 events = _run_loop(repo, once=True, branch_filter="main")
659 assert all(e["data"].get("branch") == "main" for e in events)
660 assert len(events) == 1
661
662 def test_combined_filters(self, repo: pathlib.Path) -> None:
663 _make_reservation(repo, run_id="agent-A", branch="main")
664 _make_reservation(repo, run_id="agent-A", branch="feature/x")
665 _make_reservation(repo, run_id="agent-B", branch="main")
666 events = _run_loop(repo, once=True, run_id_filter="agent-A", branch_filter="main")
667 assert len(events) == 1
668 assert events[0]["data"]["run_id"] == "agent-A"
669 assert events[0]["data"]["branch"] == "main"
670
671 def test_no_match_no_events(self, repo: pathlib.Path) -> None:
672 _make_reservation(repo, run_id="agent-X")
673 events = _run_loop(repo, once=True, run_id_filter="agent-NOBODY")
674 assert events == []
675
676
677 # ─────────────────────────────────────────────────────────────────────────────
678 # Integration — added events
679 # ─────────────────────────────────────────────────────────────────────────────
680
681
682 class TestWatchLoopDetectsAdded:
683 def test_new_reservation_emits_added(self, repo: pathlib.Path) -> None:
684 added = []
685
686 def _write_res() -> None:
687 _make_reservation(repo, run_id="agent-new")
688
689 events = _run_loop(repo, side_effect=_write_res, once=False)
690 added = [e for e in events if e["event_type"] == "added"]
691 assert len(added) == 1
692 assert added[0]["kind"] == "reservations"
693 assert added[0]["data"]["run_id"] == "agent-new"
694
695 def test_new_intent_emits_added(self, repo: pathlib.Path) -> None:
696 res = _make_reservation(repo)
697
698 def _write_intent() -> None:
699 create_intent(repo, res.reservation_id, "agent-1", "main",
700 ["src/mod.py::fn"], "modify")
701
702 events = _run_loop(repo, side_effect=_write_intent, once=False)
703 added = [e for e in events if e["event_type"] == "added" and e["kind"] == "intents"]
704 assert len(added) == 1
705
706 def test_added_event_carries_data(self, repo: pathlib.Path) -> None:
707 def _write_res() -> None:
708 _make_reservation(repo, run_id="agent-data-test", branch="feature/y")
709
710 events = _run_loop(repo, side_effect=_write_res, once=False)
711 added = [e for e in events if e["event_type"] == "added"]
712 assert added[0]["data"]["run_id"] == "agent-data-test"
713 assert added[0]["data"]["branch"] == "feature/y"
714
715
716 # ─────────────────────────────────────────────────────────────────────────────
717 # Integration — modified events
718 # ─────────────────────────────────────────────────────────────────────────────
719
720
721 class TestWatchLoopDetectsModified:
722 def test_heartbeat_update_emits_modified(self, repo: pathlib.Path) -> None:
723 res = _make_reservation(repo)
724 # Write an initial heartbeat so the file exists in the snapshot.
725 create_heartbeat(repo, res.reservation_id, "agent-1", extension_seconds=100)
726
727 def _update_hb() -> None:
728 # Small sleep ensures mtime changes on fast filesystems.
729 time.sleep(0.02)
730 create_heartbeat(repo, res.reservation_id, "agent-1", extension_seconds=200)
731
732 events = _run_loop(repo, side_effect=_update_hb, once=False)
733 modified = [e for e in events if e["event_type"] == "modified" and e["kind"] == "heartbeats"]
734 assert len(modified) == 1
735
736 def test_modified_event_carries_new_data(self, repo: pathlib.Path) -> None:
737 res = _make_reservation(repo)
738 create_heartbeat(repo, res.reservation_id, "agent-1", extension_seconds=100)
739
740 def _update_hb() -> None:
741 time.sleep(0.02)
742 create_heartbeat(repo, res.reservation_id, "agent-1", extension_seconds=9999)
743
744 events = _run_loop(repo, side_effect=_update_hb, once=False)
745 modified = [e for e in events if e["event_type"] == "modified"]
746 if modified:
747 # Data should reflect the new heartbeat.
748 assert "extended_expires_at" in modified[0]["data"]
749
750
751 # ─────────────────────────────────────────────────────────────────────────────
752 # Integration — removed events
753 # ─────────────────────────────────────────────────────────────────────────────
754
755
756 class TestWatchLoopDetectsRemoved:
757 def test_deleted_file_emits_removed(self, repo: pathlib.Path) -> None:
758 res = _make_reservation(repo)
759 # Snapshot includes the reservation.
760 path = _coord_dir(repo) / "reservations" / f"{res.reservation_id}.json"
761
762 def _delete_file() -> None:
763 path.unlink()
764
765 events = _run_loop(repo, side_effect=_delete_file, once=False)
766 removed = [e for e in events if e["event_type"] == "removed"]
767 assert len(removed) == 1
768 assert removed[0]["id"] == res.reservation_id
769
770 def test_removed_event_carries_cached_data(self, repo: pathlib.Path) -> None:
771 """Even after deletion, removed events carry the last-known data."""
772 res = _make_reservation(repo, run_id="agent-cached")
773 path = _coord_dir(repo) / "reservations" / f"{res.reservation_id}.json"
774
775 def _delete_file() -> None:
776 path.unlink()
777
778 events = _run_loop(repo, side_effect=_delete_file, once=False)
779 removed = [e for e in events if e["event_type"] == "removed"]
780 assert removed[0]["data"].get("run_id") == "agent-cached"
781
782 def test_removed_event_passes_run_id_filter(self, repo: pathlib.Path) -> None:
783 """Filter by run_id must work for removed events using cached data."""
784 res = _make_reservation(repo, run_id="agent-filter-test")
785 path = _coord_dir(repo) / "reservations" / f"{res.reservation_id}.json"
786
787 def _delete_file() -> None:
788 path.unlink()
789
790 events = _run_loop(
791 repo, side_effect=_delete_file, once=False,
792 run_id_filter="agent-filter-test",
793 )
794 removed = [e for e in events if e["event_type"] == "removed"]
795 assert len(removed) == 1
796
797
798 # ─────────────────────────────────────────────────────────────────────────────
799 # Integration — expiration events
800 # ─────────────────────────────────────────────────────────────────────────────
801
802
803 class TestWatchLoopDetectsExpiry:
804 def test_expired_reservation_emits_expired(self, repo: pathlib.Path) -> None:
805 """Reservation active at startup that expires during loop → expired event.
806
807 Strategy: create an active reservation, then the side_effect rewrites the
808 file with expires_at in the past. The loop sees it as modified on the FS
809 AND sees it drop out of active_reservations() → fires expired event.
810 """
811 res = _make_reservation(repo, ttl_seconds=3600)
812 path = _coord_dir(repo) / "reservations" / f"{res.reservation_id}.json"
813
814 def _expire_it() -> None:
815 data = json.loads(path.read_text())
816 past = (
817 datetime.datetime.now(datetime.timezone.utc)
818 - datetime.timedelta(hours=2)
819 ).isoformat()
820 data["expires_at"] = past
821 path.write_text(json.dumps(data))
822
823 events = _run_loop(repo, side_effect=_expire_it, once=False)
824 expired = [e for e in events if e["event_type"] == "expired"]
825 assert any(e["id"] == res.reservation_id for e in expired)
826
827 def test_active_reservation_no_expired_event(self, repo: pathlib.Path) -> None:
828 res = _make_reservation(repo, ttl_seconds=9999)
829 events = _run_loop(repo, once=False)
830 expired = [e for e in events if e["event_type"] == "expired"]
831 assert not any(e["id"] == res.reservation_id for e in expired)
832
833 def test_removed_reservation_not_expired(self, repo: pathlib.Path) -> None:
834 """GC'd reservation (file deleted) must not also fire expired.
835
836 Strategy: create an active reservation, then the side_effect deletes it
837 (simulating GC). Must see a removed event but NOT an additional expired
838 event for the same ID.
839 """
840 res = _make_reservation(repo, ttl_seconds=3600)
841 path = _coord_dir(repo) / "reservations" / f"{res.reservation_id}.json"
842
843 def _delete_file() -> None:
844 path.unlink()
845
846 events = _run_loop(repo, side_effect=_delete_file, once=False)
847 removed = [e for e in events if e["event_type"] == "removed" and e["id"] == res.reservation_id]
848 expired_dup = [
849 e for e in events
850 if e["event_type"] == "expired" and e["id"] == res.reservation_id
851 ]
852 # Must see a removed event, must NOT also see expired for the same ID.
853 assert removed
854 assert not expired_dup
855
856
857 # ─────────────────────────────────────────────────────────────────────────────
858 # Integration — timeout
859 # ─────────────────────────────────────────────────────────────────────────────
860
861
862 class TestWatchLoopTimeout:
863 def test_timeout_zero_equivalent_to_once(self, repo: pathlib.Path) -> None:
864 """timeout=0 should exit after snapshot, no looping."""
865 _make_reservation(repo)
866 start = time.monotonic()
867 events = _run_loop(repo, once=False, timeout=0.0)
868 elapsed = time.monotonic() - start
869 # Should return quickly (snapshot only).
870 assert elapsed < 1.0
871 # Should still emit the snapshot.
872 assert any(e["event_type"] == "snapshot" for e in events)
873
874 def test_loop_exits_after_timeout(self, repo: pathlib.Path) -> None:
875 """With a real PollingBackend and short timeout, loop exits."""
876 buf = io.StringIO()
877 backend = _PollingBackend(0.05)
878 start = time.monotonic()
879 with redirect_stdout(buf):
880 _watch_loop(
881 repo, backend,
882 kind_filter=None, run_id_filter=None, branch_filter=None,
883 as_json=True, once=False, timeout=0.2, poll_interval=0.05,
884 )
885 elapsed = time.monotonic() - start
886 assert elapsed < 2.0 # Must not run forever.
887
888
889 # ─────────────────────────────────────────────────────────────────────────────
890 # Integration — JSON output
891 # ─────────────────────────────────────────────────────────────────────────────
892
893
894 class TestWatchLoopJsonOutput:
895 def test_all_snapshot_events_are_valid_json(self, repo: pathlib.Path) -> None:
896 _make_reservation(repo)
897 events = _run_loop(repo, once=True, as_json=True)
898 for ev in events:
899 # Already parsed by _run_loop, so this just checks structure.
900 assert "schema_version" in ev
901 assert "event_type" in ev
902 assert "kind" in ev
903 assert "id" in ev
904 assert "timestamp" in ev
905 assert "data" in ev
906
907 def test_json_event_type_values(self, repo: pathlib.Path) -> None:
908 _make_reservation(repo)
909 events = _run_loop(repo, once=True)
910 valid_types = {"snapshot", "added", "modified", "removed", "expired"}
911 for ev in events:
912 assert ev["event_type"] in valid_types
913
914
915 # ─────────────────────────────────────────────────────────────────────────────
916 # End-to-end — run() with mocked require_repo
917 # ─────────────────────────────────────────────────────────────────────────────
918
919
920 class TestRunCommand:
921 def _make_args(self, repo: pathlib.Path, **kwargs: MsgpackValue) -> argparse.Namespace:
922 import argparse
923 ns = argparse.Namespace()
924 ns.once = kwargs.get("once", True)
925 ns.timeout = kwargs.get("timeout", None)
926 ns.max_events = kwargs.get("max_events", None)
927 ns.poll_interval = kwargs.get("poll_interval", 0.1)
928 ns.run_id = kwargs.get("run_id", None)
929 ns.branch_filter = kwargs.get("branch_filter", None)
930 ns.kind = kwargs.get("kind", None)
931 ns.fmt = kwargs.get("fmt", "json")
932 return ns
933
934 def test_run_once_emits_snapshot(self, repo: pathlib.Path) -> None:
935 from muse.cli.commands.watch_coord import run
936 _make_reservation(repo)
937 args = self._make_args(repo, once=True)
938 buf = io.StringIO()
939 with (
940 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
941 redirect_stdout(buf),
942 ):
943 run(args)
944 lines = [l for l in buf.getvalue().splitlines() if l.strip()]
945 events = [json.loads(l) for l in lines]
946 assert any(e["event_type"] == "snapshot" for e in events)
947
948 def test_run_text_mode_header_printed(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
949 from muse.cli.commands.watch_coord import run
950 args = self._make_args(repo, once=True, fmt="text")
951 with patch("muse.cli.commands.watch_coord.require_repo", return_value=repo):
952 run(args)
953 captured = capsys.readouterr()
954 assert "muse coord watch" in captured.out
955 assert "watch ended" in captured.out
956
957 def test_run_json_mode_no_header(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
958 from muse.cli.commands.watch_coord import run
959 args = self._make_args(repo, once=True, fmt="json")
960 with patch("muse.cli.commands.watch_coord.require_repo", return_value=repo):
961 run(args)
962 captured = capsys.readouterr()
963 lines = [l for l in captured.out.splitlines() if l.strip()]
964 for line in lines:
965 # Every non-empty line must be valid JSON.
966 json.loads(line)
967
968 def test_run_timeout_zero_treated_as_once(self, repo: pathlib.Path) -> None:
969 """--timeout 0 must not block."""
970 from muse.cli.commands.watch_coord import run
971 args = self._make_args(repo, once=False, timeout=0.0, fmt="json")
972 start = time.monotonic()
973 buf = io.StringIO()
974 with (
975 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
976 redirect_stdout(buf),
977 ):
978 run(args)
979 assert time.monotonic() - start < 1.0
980
981
982 # ─────────────────────────────────────────────────────────────────────────────
983 # End-to-end — backend selection
984 # ─────────────────────────────────────────────────────────────────────────────
985
986
987 class TestBackendSelection:
988 def test_polling_backend_wait_sleeps(self) -> None:
989 b = _PollingBackend(0.05)
990 start = time.monotonic()
991 result = b.wait(0.1)
992 elapsed = time.monotonic() - start
993 assert result is True
994 assert elapsed >= 0.04 # Should have slept.
995 b.close()
996
997 def test_polling_backend_respects_timeout_cap(self) -> None:
998 """wait(timeout) must not sleep longer than timeout."""
999 b = _PollingBackend(10.0) # Long interval.
1000 start = time.monotonic()
1001 b.wait(0.05) # Short timeout.
1002 elapsed = time.monotonic() - start
1003 assert elapsed < 1.0 # Capped at 0.05 s.
1004 b.close()
1005
1006 def test_polling_backend_close_is_idempotent(self) -> None:
1007 b = _PollingBackend(1.0)
1008 b.close()
1009 b.close() # Should not raise.
1010
1011 @pytest.mark.skipif(
1012 not hasattr(__import__("select"), "kqueue"),
1013 reason="kqueue not available on this platform",
1014 )
1015 def test_kqueue_backend_initialises(self, repo: pathlib.Path) -> None:
1016 """kqueue backend can be created and closed without error."""
1017 from muse.cli.commands.watch_coord import _SUBDIRS
1018 dirs = [_coord_dir(repo) / sub for sub in _SUBDIRS]
1019 b = _KqueueBackend(dirs)
1020 assert b.name == "kqueue"
1021 b.close()
1022
1023 @pytest.mark.skipif(
1024 not hasattr(__import__("select"), "kqueue"),
1025 reason="kqueue not available on this platform",
1026 )
1027 def test_kqueue_backend_detects_new_file(self, repo: pathlib.Path) -> None:
1028 """kqueue wakes when a new JSON file is added to a watched dir."""
1029 from muse.cli.commands.watch_coord import _SUBDIRS
1030 dirs = [_coord_dir(repo) / sub for sub in _SUBDIRS]
1031 b = _KqueueBackend(dirs)
1032 try:
1033 # No change — should time out.
1034 result_before = b.wait(0.05)
1035 # Add a file.
1036 (_coord_dir(repo) / "reservations" / "newfile.json").write_text("{}")
1037 # Should detect change.
1038 result_after = b.wait(0.5)
1039 assert result_after is True
1040 finally:
1041 b.close()
1042
1043 @pytest.mark.skipif(
1044 not hasattr(__import__("select"), "kqueue"),
1045 reason="kqueue not available on this platform",
1046 )
1047 def test_kqueue_backend_close_releases_fds(self, repo: pathlib.Path) -> None:
1048 """All fds must be released after close."""
1049 import resource
1050 from muse.cli.commands.watch_coord import _SUBDIRS
1051 dirs = [_coord_dir(repo) / sub for sub in _SUBDIRS]
1052 fds_before = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
1053 b = _KqueueBackend(dirs)
1054 b.close()
1055 # Just ensure no exception — fd leak would only show in long-running tests.
1056
1057
1058 # ─────────────────────────────────────────────────────────────────────────────
1059 # Security tests
1060 # ─────────────────────────────────────────────────────────────────────────────
1061
1062
1063 class TestAnsiInjectionSanitized:
1064 def test_ansi_in_run_id_stripped_text_output(self) -> None:
1065 """ESC sequences in run_id must not reach stdout in text mode."""
1066 evil = "\x1b[1;31mROOT\x1b[0m"
1067 ev = WatchEvent("added", "reservation", "a" * 8, "ts",
1068 {"run_id": evil, "branch": "main", "addresses": ["f.py::fn"]})
1069 buf = io.StringIO()
1070 with redirect_stdout(buf):
1071 _emit_event(ev, as_json=False)
1072 assert "\x1b" not in buf.getvalue()
1073 assert "ROOT" in buf.getvalue()
1074
1075 def test_ansi_in_branch_stripped(self) -> None:
1076 evil_branch = "\x1b[4mmaster\x1b[0m"
1077 ev = WatchEvent("added", "reservation", "b" * 8, "ts",
1078 {"run_id": "agent", "branch": evil_branch, "addresses": ["f.py::fn"]})
1079 buf = io.StringIO()
1080 with redirect_stdout(buf):
1081 _emit_event(ev, as_json=False)
1082 assert "\x1b" not in buf.getvalue()
1083
1084 def test_ansi_in_address_stripped(self) -> None:
1085 evil_addr = "\x1b[32msrc/evil.py::inject\x1b[0m"
1086 ev = WatchEvent("added", "reservation", "c" * 8, "ts",
1087 {"run_id": "agent", "branch": "main", "addresses": [evil_addr]})
1088 buf = io.StringIO()
1089 with redirect_stdout(buf):
1090 _emit_event(ev, as_json=False)
1091 assert "\x1b" not in buf.getvalue()
1092
1093 def test_json_output_preserves_raw_data(self) -> None:
1094 """JSON output must preserve raw data as-is (no sanitisation)."""
1095 evil = "\x1b[31mevil\x1b[0m"
1096 ev = WatchEvent("added", "reservation", "d" * 8, "ts",
1097 {"run_id": evil, "branch": "main", "addresses": ["f.py::fn"]})
1098 buf = io.StringIO()
1099 with redirect_stdout(buf):
1100 _emit_event(ev, as_json=True)
1101 parsed = json.loads(buf.getvalue())
1102 assert parsed["data"]["run_id"] == evil # Stored as-is in JSON.
1103
1104
1105 class TestSymlinkDirRejected:
1106 @pytest.mark.skipif(
1107 not hasattr(__import__("select"), "kqueue"),
1108 reason="kqueue not available on this platform",
1109 )
1110 def test_symlinked_coord_dir_raises_valueerror(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
1111 """kqueue backend must refuse to watch a symlinked directory."""
1112 import shutil
1113 # Create a real target dir for the symlink to point at.
1114 real_dir = tmp_path / "real_target"
1115 real_dir.mkdir()
1116 link_dir = _coord_dir(repo) / "reservations"
1117 # Remove the existing real directory before replacing with a symlink.
1118 shutil.rmtree(link_dir, ignore_errors=True)
1119 link_dir.symlink_to(real_dir)
1120 try:
1121 dirs = [link_dir]
1122 with pytest.raises(ValueError, match="symlink"):
1123 _KqueueBackend(dirs)
1124 finally:
1125 link_dir.unlink(missing_ok=True)
1126 # Restore the real dir for other tests.
1127 link_dir.mkdir(exist_ok=True)
1128
1129
1130 class TestKindFilterAllowlist:
1131 def test_valid_kinds_accepted(self) -> None:
1132 for kind in ("reservations", "intents", "releases", "heartbeats"):
1133 assert _passes_filters(kind, {}, kind, None, None)
1134
1135 def test_arbitrary_kind_string_rejected_by_filter(self) -> None:
1136 """_passes_filters rejects unrecognised kinds when kind_filter is set."""
1137 assert not _passes_filters("../etc", {}, "reservations", None, None)
1138
1139 def test_argparse_rejects_invalid_kind(self) -> None:
1140 """argparse choices validation rejects invalid --kind values."""
1141 import argparse
1142 from muse.cli.commands.watch_coord import register
1143 p = argparse.ArgumentParser()
1144 subs = p.add_subparsers()
1145 register(subs)
1146 with pytest.raises(SystemExit):
1147 p.parse_args(["watch", "--kind", "invalid_kind"])
1148
1149
1150 # ─────────────────────────────────────────────────────────────────────────────
1151 # Stress tests
1152 # ─────────────────────────────────────────────────────────────────────────────
1153
1154
1155 class TestStress500RapidAdds:
1156 def test_500_reservations_all_detected(self, repo: pathlib.Path) -> None:
1157 """500 pre-existing reservations must all appear in the snapshot."""
1158 # Write 500 reservation files directly (bypass create_reservation for speed).
1159 res_dir = _coord_dir(repo) / "reservations"
1160 now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
1161 exp_iso = (
1162 datetime.datetime.now(datetime.timezone.utc)
1163 + datetime.timedelta(hours=1)
1164 ).isoformat()
1165 ids = []
1166 for i in range(500):
1167 uid = str(uuid.uuid4())
1168 ids.append(uid)
1169 data = {
1170 "reservation_id": uid,
1171 "run_id": f"agent-{i}",
1172 "branch": "main",
1173 "addresses": [f"src/mod{i}.py::fn"],
1174 "created_at": now_iso,
1175 "expires_at": exp_iso,
1176 "operation": "modify",
1177 "schema_version": "1.0.0",
1178 }
1179 (res_dir / f"{uid}.json").write_text(json.dumps(data))
1180
1181 start = time.monotonic()
1182 events = _run_loop(repo, once=True)
1183 elapsed = time.monotonic() - start
1184
1185 snapshot_events = [e for e in events if e["event_type"] == "snapshot"]
1186 snapshot_ids = {e["id"] for e in snapshot_events}
1187 assert snapshot_ids == set(ids)
1188 assert elapsed < 3.0, f"500 snapshot took {elapsed:.2f}s (limit: 3s)"
1189
1190
1191 class TestStressLargeSnapshot:
1192 def test_1000_records_snapshot_under_1s(self, repo: pathlib.Path) -> None:
1193 """Snapshot of 1000 mixed records must complete in under 1 second."""
1194 now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
1195 exp_iso = (
1196 datetime.datetime.now(datetime.timezone.utc)
1197 + datetime.timedelta(hours=1)
1198 ).isoformat()
1199 for i in range(250):
1200 for kind in ("reservations", "intents", "releases", "heartbeats"):
1201 uid = str(uuid.uuid4())
1202 (_coord_dir(repo) / kind / f"{uid}.json").write_text(
1203 json.dumps({"id": uid, "run_id": f"agent-{i}", "branch": "main",
1204 "created_at": now_iso, "expires_at": exp_iso})
1205 )
1206 start = time.monotonic()
1207 snap = _scan_dirs(repo)
1208 elapsed = time.monotonic() - start
1209 total = sum(len(v) for v in snap.values())
1210 assert total == 1000
1211 assert elapsed < 1.0, f"scan of 1000 records took {elapsed:.3f}s"
1212
1213
1214 class TestStressDiffPerformance:
1215 def test_diff_1000_entries_under_50ms(self) -> None:
1216 """Diffing two 1000-entry snapshots must take < 50 ms."""
1217 old: _Snapshot = {sub: {} for sub in ("reservations", "intents", "releases", "heartbeats")}
1218 new: _Snapshot = {sub: {} for sub in ("reservations", "intents", "releases", "heartbeats")}
1219 for i in range(250):
1220 for sub in ("reservations", "intents", "releases", "heartbeats"):
1221 uid = str(uuid.uuid4())
1222 old[sub][uid] = (i * 1000, i * 10)
1223 new[sub][uid] = (i * 1000, i * 10)
1224 # Add 10 new entries to make the diff non-trivial.
1225 for _ in range(10):
1226 new["reservations"][str(uuid.uuid4())] = (999999, 99)
1227
1228 start = time.monotonic()
1229 for _ in range(100): # 100 diff calls.
1230 _diff_snapshots(old, new)
1231 elapsed = time.monotonic() - start
1232 assert elapsed < 5.0, f"100 × diff took {elapsed:.3f}s (limit: 5s)"
1233
1234
1235 class TestStressManyExpirations:
1236 def test_200_expired_reservations_detected(self, repo: pathlib.Path) -> None:
1237 """200 expired reservations must all emit expired events."""
1238 res_dir = _coord_dir(repo) / "reservations"
1239 past_iso = (
1240 datetime.datetime.now(datetime.timezone.utc)
1241 - datetime.timedelta(hours=1)
1242 ).isoformat()
1243 now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
1244 ids = set()
1245 for i in range(200):
1246 uid = str(uuid.uuid4())
1247 ids.add(uid)
1248 data = {
1249 "reservation_id": uid,
1250 "run_id": f"agent-{i}",
1251 "branch": "main",
1252 "addresses": [f"src/m{i}.py::fn"],
1253 "created_at": now_iso,
1254 "expires_at": past_iso, # Already expired.
1255 "operation": "modify",
1256 "schema_version": "1.0.0",
1257 }
1258 (res_dir / f"{uid}.json").write_text(json.dumps(data))
1259
1260 # prev_active_ids claims all 200 were active.
1261 start = time.monotonic()
1262 exp_events, curr = _check_expirations(repo, ids, set())
1263 elapsed = time.monotonic() - start
1264
1265 expired_ids = {e.id for e in exp_events}
1266 assert expired_ids == ids
1267 assert elapsed < 2.0, f"200 expiration checks took {elapsed:.3f}s (limit: 2s)"
1268
1269
1270 # ─────────────────────────────────────────────────────────────────────────────
1271 # Integration — _watch_loop with max_events
1272 # ─────────────────────────────────────────────────────────────────────────────
1273
1274
1275 class TestWatchLoopMaxEvents:
1276 def test_max_events_1_returns_single_snapshot(self, repo: pathlib.Path) -> None:
1277 """max_events=1 must emit exactly 1 event even if more exist."""
1278 for i in range(5):
1279 _make_reservation(repo, run_id=f"ag-{i}")
1280 events = _run_loop(repo, once=True, max_events=1)
1281 assert len(events) == 1
1282
1283 def test_max_events_3_caps_at_3(self, repo: pathlib.Path) -> None:
1284 for i in range(10):
1285 _make_reservation(repo, run_id=f"ag-{i}")
1286 events = _run_loop(repo, once=True, max_events=3)
1287 assert len(events) == 3
1288
1289 def test_max_events_larger_than_existing_returns_all(self, repo: pathlib.Path) -> None:
1290 for i in range(4):
1291 _make_reservation(repo, run_id=f"ag-{i}")
1292 events = _run_loop(repo, once=True, max_events=100)
1293 assert len(events) == 4
1294
1295 def test_max_events_zero_is_rejected_by_run(self, repo: pathlib.Path) -> None:
1296 """run() must reject max_events=0 with USER_ERROR."""
1297 from muse.cli.commands.watch_coord import run as watch_run
1298 ns = argparse.Namespace(
1299 once=True, timeout=None, max_events=0,
1300 poll_interval=0.1, run_id=None,
1301 branch_filter=None, kind=None, fmt="json",
1302 )
1303 buf = io.StringIO()
1304 with (
1305 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
1306 redirect_stdout(buf),
1307 ):
1308 with pytest.raises(SystemExit) as exc_info:
1309 watch_run(ns)
1310 assert exc_info.value.code == ExitCode.USER_ERROR
1311
1312 def test_max_events_negative_is_rejected_by_run(self, repo: pathlib.Path) -> None:
1313 from muse.cli.commands.watch_coord import run as watch_run
1314 ns = argparse.Namespace(
1315 once=True, timeout=None, max_events=-5,
1316 poll_interval=0.1, run_id=None,
1317 branch_filter=None, kind=None, fmt="json",
1318 )
1319 buf = io.StringIO()
1320 with (
1321 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
1322 redirect_stdout(buf),
1323 ):
1324 with pytest.raises(SystemExit) as exc_info:
1325 watch_run(ns)
1326 assert exc_info.value.code == ExitCode.USER_ERROR
1327
1328 def test_max_events_caps_added_events_in_loop(self, repo: pathlib.Path) -> None:
1329 """max_events must also cap events emitted during the loop (not just snapshots)."""
1330 # No pre-existing records; add 10 during the side effect.
1331 def _add_records() -> None:
1332 for i in range(10):
1333 _make_reservation(repo, run_id=f"ag-{i}")
1334
1335 events = _run_loop(
1336 repo,
1337 once=False,
1338 timeout=1.0,
1339 side_effect=_add_records,
1340 max_events=3,
1341 )
1342 assert len(events) <= 3
1343
1344 def test_max_events_json_error_is_compact(self, repo: pathlib.Path) -> None:
1345 """Error for bad max_events in JSON mode must be a single compact line."""
1346 from muse.cli.commands.watch_coord import run as watch_run
1347 ns = argparse.Namespace(
1348 once=True, timeout=None, max_events=0,
1349 poll_interval=0.1, run_id=None,
1350 branch_filter=None, kind=None, fmt="json",
1351 )
1352 buf = io.StringIO()
1353 with (
1354 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
1355 redirect_stdout(buf),
1356 ):
1357 with pytest.raises(SystemExit):
1358 watch_run(ns)
1359 out = buf.getvalue().strip()
1360 assert "\n" not in out
1361 data = json.loads(out)
1362 assert "error" in data
1363
1364
1365 # ─────────────────────────────────────────────────────────────────────────────
1366 # End-to-end — run() input validation
1367 # ─────────────────────────────────────────────────────────────────────────────
1368
1369
1370 class TestRunCommandValidation:
1371 def _make_args(self, **kwargs: MsgpackValue) -> argparse.Namespace:
1372 ns = argparse.Namespace()
1373 ns.once = kwargs.get("once", True)
1374 ns.timeout = kwargs.get("timeout", None)
1375 ns.max_events = kwargs.get("max_events", None)
1376 ns.poll_interval = kwargs.get("poll_interval", 0.1)
1377 ns.run_id = kwargs.get("run_id", None)
1378 ns.branch_filter = kwargs.get("branch_filter", None)
1379 ns.kind = kwargs.get("kind", None)
1380 ns.fmt = kwargs.get("fmt", "text")
1381 return ns
1382
1383 def test_run_id_at_max_length_accepted(self, repo: pathlib.Path) -> None:
1384 from muse.cli.commands.watch_coord import run as watch_run
1385 run_id = "a" * _MAX_RUN_ID_LEN
1386 ns = self._make_args(run_id=run_id, once=True)
1387 buf = io.StringIO()
1388 with (
1389 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
1390 redirect_stdout(buf),
1391 ):
1392 watch_run(ns) # must not raise
1393
1394 def test_run_id_over_max_exits_user_error(self, repo: pathlib.Path) -> None:
1395 from muse.cli.commands.watch_coord import run as watch_run
1396 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
1397 ns = self._make_args(run_id=run_id)
1398 with pytest.raises(SystemExit) as exc_info:
1399 watch_run(ns)
1400 assert exc_info.value.code == ExitCode.USER_ERROR
1401
1402 def test_run_id_over_max_json_returns_error_field(self, repo: pathlib.Path) -> None:
1403 from muse.cli.commands.watch_coord import run as watch_run
1404 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
1405 ns = self._make_args(run_id=run_id, fmt="json")
1406 buf = io.StringIO()
1407 with redirect_stdout(buf):
1408 with pytest.raises(SystemExit):
1409 watch_run(ns)
1410 out = buf.getvalue().strip()
1411 assert "\n" not in out # Compact JSON.
1412 data = json.loads(out)
1413 assert "error" in data
1414 assert data.get("status") == "bad_args"
1415
1416 def test_poll_interval_below_min_exits_user_error(self, repo: pathlib.Path) -> None:
1417 from muse.cli.commands.watch_coord import run as watch_run
1418 ns = self._make_args(poll_interval=0.001)
1419 with pytest.raises(SystemExit) as exc_info:
1420 watch_run(ns)
1421 assert exc_info.value.code == ExitCode.USER_ERROR
1422
1423 def test_poll_interval_above_max_exits_user_error(self, repo: pathlib.Path) -> None:
1424 from muse.cli.commands.watch_coord import run as watch_run
1425 ns = self._make_args(poll_interval=99999.0)
1426 with pytest.raises(SystemExit) as exc_info:
1427 watch_run(ns)
1428 assert exc_info.value.code == ExitCode.USER_ERROR
1429
1430 def test_poll_interval_at_min_accepted(self, repo: pathlib.Path) -> None:
1431 from muse.cli.commands.watch_coord import run as watch_run
1432 ns = self._make_args(poll_interval=_MIN_POLL_INTERVAL, once=True)
1433 buf = io.StringIO()
1434 with (
1435 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
1436 redirect_stdout(buf),
1437 ):
1438 watch_run(ns) # must not raise
1439
1440 def test_poll_interval_at_max_accepted(self, repo: pathlib.Path) -> None:
1441 from muse.cli.commands.watch_coord import run as watch_run
1442 ns = self._make_args(poll_interval=_MAX_POLL_INTERVAL, once=True)
1443 buf = io.StringIO()
1444 with (
1445 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
1446 redirect_stdout(buf),
1447 ):
1448 watch_run(ns) # must not raise
1449
1450 def test_timeout_negative_exits_user_error(self, repo: pathlib.Path) -> None:
1451 from muse.cli.commands.watch_coord import run as watch_run
1452 ns = self._make_args(timeout=-1.0)
1453 with pytest.raises(SystemExit) as exc_info:
1454 watch_run(ns)
1455 assert exc_info.value.code == ExitCode.USER_ERROR
1456
1457 def test_timeout_zero_accepted_as_once(self, repo: pathlib.Path) -> None:
1458 """--timeout 0 must complete quickly (treated as --once)."""
1459 from muse.cli.commands.watch_coord import run as watch_run
1460 ns = self._make_args(timeout=0.0, fmt="json")
1461 buf = io.StringIO()
1462 start = time.monotonic()
1463 with (
1464 patch("muse.cli.commands.watch_coord.require_repo", return_value=repo),
1465 redirect_stdout(buf),
1466 ):
1467 watch_run(ns)
1468 assert time.monotonic() - start < 1.0
1469
1470 def test_validation_fires_before_require_repo(self) -> None:
1471 """Bad --run-id must fail before require_repo() is ever called."""
1472 from muse.cli.commands.watch_coord import run as watch_run
1473 run_id = "x" * (_MAX_RUN_ID_LEN + 1)
1474 ns = argparse.Namespace(
1475 once=True, timeout=None, max_events=None,
1476 poll_interval=0.1, run_id=run_id,
1477 branch_filter=None, kind=None, fmt="text",
1478 )
1479 called = []
1480 with patch(
1481 "muse.cli.commands.watch_coord.require_repo",
1482 side_effect=lambda: called.append(True),
1483 ):
1484 with pytest.raises(SystemExit):
1485 watch_run(ns)
1486 assert called == [], "require_repo must not be called before validation passes"
1487
1488 def test_error_message_has_check_mark_prefix(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1489 """Validation error on stderr must start with ❌."""
1490 from muse.cli.commands.watch_coord import run as watch_run
1491 ns = self._make_args(poll_interval=0.0001)
1492 with pytest.raises(SystemExit):
1493 watch_run(ns)
1494 err = capsys.readouterr().err
1495 assert "❌" in err
1496
1497
1498 # ─────────────────────────────────────────────────────────────────────────────
1499 # Stress — max_events under load
1500 # ─────────────────────────────────────────────────────────────────────────────
1501
1502
1503 class TestStressMaxEvents:
1504 def test_max_events_10_from_500_records_returns_exactly_10(self, repo: pathlib.Path) -> None:
1505 """max_events=10 must return exactly 10 events even with 500 records."""
1506 res_dir = _coord_dir(repo) / "reservations"
1507 now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
1508 exp_iso = (
1509 datetime.datetime.now(datetime.timezone.utc)
1510 + datetime.timedelta(hours=1)
1511 ).isoformat()
1512 for i in range(500):
1513 uid = str(uuid.uuid4())
1514 data = {
1515 "reservation_id": uid,
1516 "run_id": f"agent-{i}",
1517 "branch": "main",
1518 "addresses": [f"src/mod{i}.py::fn"],
1519 "created_at": now_iso,
1520 "expires_at": exp_iso,
1521 "operation": "modify",
1522 "schema_version": "1.0.0",
1523 }
1524 (res_dir / f"{uid}.json").write_text(json.dumps(data))
1525
1526 start = time.monotonic()
1527 events = _run_loop(repo, once=True, max_events=10)
1528 elapsed = time.monotonic() - start
1529
1530 assert len(events) == 10
1531 assert elapsed < 2.0, f"max_events cap with 500 records took {elapsed:.2f}s"
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 100 days ago