gabriel / muse public
test_coord_data_integrity.py python
1,509 lines 67.6 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """EXTREME data integrity test suite for ``muse coord sync`` — the coord layer.
2
3 Scenario: Linus Torvalds ports the Linux kernel to Muse. 50 AI agents are
4 simultaneously reading and writing coordination records across 7 kinds.
5 Thousands of files. Network blips. Corrupted disk sectors. Adversarial hubs.
6 Concurrent writers. This suite attempts to find the edge and go beyond it.
7
8 Coverage matrix
9 ---------------
10 Round-trip fidelity
11 * All 7 kinds survive _gather → _write with bit-exact field preservation
12 * Unicode in payloads (CJK, emoji, RTL Arabic) — no mojibake
13 * Deeply nested payloads — no truncation or type coercion
14 * Null / empty / boolean payload values — exact type preservation
15 * 50 KB payload string — no truncation
16 * record_uuid fallback chain: explicit field → fpath.stem
17
18 Batching correctness
19 * Exactly MAX_PUSH_BATCH (500) records → exactly 1 push_to_hub call
20 * 501 records → exactly 2 push_to_hub calls
21 * 1 000 records → exactly 2 calls (500 each)
22 * 1 001 records → exactly 3 calls (500, 500, 1)
23 * 4 999 records (Linux-scale) → 10 calls (9×500 + 1×499)
24 * First batch fails, remaining batches still execute
25 * All batches fail → inserted=0, failed=True, exit 1
26 * Batch slices are non-overlapping and cover all records exactly
27 * accumulated inserted/skipped counts are sum of all batch results
28
29 Cursor correctness
30 * cursor=0 when pull returns 0 records
31 * cursor equals the id of the last returned record
32 * cursor in JSON output matches cursor in pull_from_hub return value
33 * Incremental pull chain: 5 pages, cursor advances, no gaps in record IDs
34 * since_id passed to pull_from_hub verbatim
35 * Page-chained full pull reconstructs all 1 000 records with zero duplicates
36
37 Resilience
38 * Corrupt JSON file skipped — other files in same dir still gathered
39 * Binary garbage file skipped
40 * NUL byte in JSON skipped
41 * File deleted between glob() and read_text() — skipped gracefully
42 * All files corrupt → 0 records → "(no local coordination records to push)"
43 * Partial corruption: 500 good + 500 corrupt → 500 records pushed
44 * Disk full on _write_remote_records → OSError propagates (does not swallow)
45 * OSError on mkdir propagates
46 * PermissionError on write propagates
47 * Second-pass gather after first corrupt pass finds good files added later
48
49 Concurrency
50 * 16 threads call _gather_local_records simultaneously → each sees same count
51 * 8 threads write distinct UUIDs to same remote dir → all files present at end
52 * 8 threads write same UUID → final file is valid JSON (last-writer-wins, not corrupt)
53 * run_push called by 8 threads concurrently → no crashes
54 * run_pull called by 8 threads concurrently → no crashes
55
56 Linux-scale
57 * 7 000 records (1 000 per kind) gathered correctly
58 * 7 000 records batched and pushed in 14 calls, all counted
59 * 1 000-record pull writes all 1 000 files to remote dir
60 * Mixed kinds: push 1 000 reservations + 1 000 heartbeats → 2 000 total
61 * 50 sequential push + pull cycles, total inserted/pulled = 50×N each
62
63 Response bounds
64 * Hub response missing "inserted" key → defaults to 0 (no KeyError crash)
65 * Hub response missing "skipped" key → defaults to 0
66 * Hub response with extra unknown keys → silently ignored
67 * Malformed JSON response → CoordBusError raised
68 * Empty response body → CoordBusError raised
69 * Response at exactly MAX_COORD_RESPONSE_BYTES is accepted by _post_json
70 * Response one byte over MAX_COORD_RESPONSE_BYTES → CoordBusError
71
72 Filesystem safety
73 * record_uuid = "../../../etc/passwd" → skipped, no file written outside remote/
74 * record_uuid with null byte → skipped
75 * record_uuid with 128 chars (max) → accepted, file written
76 * record_uuid with 129 chars (over max) → rejected
77 * kind = "../evil" → skipped
78 * kind = "" → skipped
79 * remote dir created automatically if absent
80 * Existing remote file overwritten with updated payload
81 * Written files are valid JSON (parseable with json.loads)
82 * Written files end with newline
83
84 Idempotency
85 * _gather_local_records is pure: same dir always returns same records
86 * Push same 500 records twice → second push all skipped, inserted=0
87 * Push 250 old + 250 new → inserted=250, skipped=250
88 * _write_remote_records same UUID twice → second write overwrites cleanly
89 * Pull with since_id=cursor returns 0 records (no replay)
90
91 Token safety
92 * Token never appears in JSON success output (push)
93 * Token never appears in JSON success output (pull)
94 * Token never appears in JSON error output (push CoordBusError)
95 * Token never appears in JSON error output (pull CoordBusError)
96 * Token never appears in text mode success output
97 """
98
99 from __future__ import annotations
100
101 import argparse
102 import io
103 import json
104 import pathlib
105 import threading
106 import uuid
107 from unittest.mock import MagicMock, call, patch
108
109 import pytest
110
111 from muse.core._types import MsgpackDict, MsgpackValue
112 from tests.cli_test_helper import CliRunner
113 from muse.core.coord_bus import CoordBusError, MAX_PUSH_BATCH, MAX_PULL_LIMIT
114 from muse.cli.commands.coord_sync import (
115 _ALL_KINDS,
116 _MAX_OWNER_LEN,
117 _MAX_PULL_LIMIT,
118 _MAX_SLUG_LEN,
119 _SAFE_RECORD_UUID_RE,
120 _gather_local_records,
121 _write_remote_records,
122 run_pull,
123 run_push,
124 )
125
126 runner = CliRunner()
127 cli = None # CliRunner accepts cli=None after argparse migration
128
129 # ---------------------------------------------------------------------------
130 # Patch targets
131 # ---------------------------------------------------------------------------
132
133 _PUSH_TARGET = "muse.cli.commands.coord_sync.push_to_hub"
134 _PULL_TARGET = "muse.cli.commands.coord_sync.pull_from_hub"
135 _REQUIRE_REPO = "muse.cli.commands.coord_sync.require_repo"
136 _RESOLVE_HUB = "muse.cli.commands.coord_sync._resolve_hub_and_signing"
137
138 # ---------------------------------------------------------------------------
139 # Common args
140 # ---------------------------------------------------------------------------
141
142 _BASE_ARGS = [
143 "--hub", "http://localhost:10003",
144 "--owner", "gabriel",
145 "--slug", "linux",
146 ]
147 _PUSH_ARGS = ["coord", "sync", "push"] + _BASE_ARGS
148 _PULL_ARGS = ["coord", "sync", "pull"] + _BASE_ARGS + ["--since-id", "0"]
149
150 # ---------------------------------------------------------------------------
151 # Helpers
152 # ---------------------------------------------------------------------------
153
154
155 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
156 """Create a minimal .muse repo layout."""
157 (tmp_path / ".muse").mkdir()
158 return tmp_path
159
160
161 def _coord_dir(root: pathlib.Path) -> pathlib.Path:
162 return root / ".muse" / "coordination"
163
164
165 def _write_records(root: pathlib.Path, kind: str, records: list[MsgpackDict]) -> None:
166 """Write coordination records as JSON files in the appropriate subdir."""
167 subdir_map = {
168 "reservation": "reservations",
169 "intent": "intents",
170 "release": "releases",
171 "heartbeat": "heartbeats",
172 "dependency": "dependencies",
173 "task": "tasks",
174 "claim": "claims",
175 }
176 subdir = _coord_dir(root) / subdir_map[kind]
177 subdir.mkdir(parents=True, exist_ok=True)
178 for rec in records:
179 fname = rec.get(
180 {
181 "reservation": "reservation_id",
182 "intent": "intent_id",
183 "release": "release_id",
184 "heartbeat": "run_id",
185 "dependency": "reservation_id",
186 "task": "task_id",
187 "claim": "task_id",
188 }[kind],
189 str(uuid.uuid4()),
190 )
191 (subdir / f"{fname}.json").write_text(json.dumps(rec), encoding="utf-8")
192
193
194 def _res(uid: str | None = None, **extra: MsgpackValue) -> MsgpackDict:
195 uid = uid or str(uuid.uuid4())
196 return {"reservation_id": uid, "run_id": f"run-{uid}", "expires_at": None, **extra}
197
198
199 def _hb(uid: str | None = None, **extra: MsgpackValue) -> MsgpackDict:
200 uid = uid or str(uuid.uuid4())
201 return {"run_id": uid, "expires_at": None, **extra}
202
203
204 def _task(uid: str | None = None, **extra: MsgpackValue) -> MsgpackDict:
205 uid = uid or str(uuid.uuid4())
206 return {"task_id": uid, "run_id": f"run-{uid}", **extra}
207
208
209 def _push_ok(inserted: int = 1, skipped: int = 0) -> MsgpackDict:
210 return {"inserted": inserted, "skipped": skipped}
211
212
213 def _pull_ok(
214 records: list[MsgpackDict] | None = None,
215 cursor: int = 1,
216 ) -> MsgpackDict:
217 recs = records if records is not None else []
218 return {"records": recs, "cursor": cursor}
219
220
221 @pytest.fixture
222 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
223 return _make_repo(tmp_path)
224
225
226 # ===========================================================================
227 # 1. ROUND-TRIP FIDELITY
228 # ===========================================================================
229
230
231 class TestCoordIntegrityRoundTrip:
232 """Records written to disk then gathered must be bit-exact."""
233
234 def test_reservation_fields_preserved(self, repo: pathlib.Path) -> None:
235 uid = str(uuid.uuid4())
236 rec = {"reservation_id": uid, "run_id": "r1", "expires_at": "2099-01-01T00:00:00Z",
237 "custom": "value"}
238 _write_records(repo, "reservation", [rec])
239 gathered = _gather_local_records(repo, ["reservation"])
240 assert len(gathered) == 1
241 g = gathered[0]
242 assert g["kind"] == "reservation"
243 assert g["record_uuid"] == uid
244 assert g["run_id"] == "r1"
245 assert g["expires_at"] == "2099-01-01T00:00:00Z"
246 assert g["payload"]["custom"] == "value"
247
248 def test_all_seven_kinds_gather_correctly(self, repo: pathlib.Path) -> None:
249 """One file per kind — all 7 must appear with correct kind field."""
250 for kind in _ALL_KINDS:
251 if kind == "reservation":
252 rec = {"reservation_id": str(uuid.uuid4()), "run_id": "r"}
253 elif kind == "intent":
254 rec = {"intent_id": str(uuid.uuid4()), "run_id": "r"}
255 elif kind == "release":
256 rec = {"release_id": str(uuid.uuid4()), "run_id": "r"}
257 elif kind == "heartbeat":
258 rec = {"run_id": str(uuid.uuid4())}
259 elif kind == "dependency":
260 rec = {"reservation_id": str(uuid.uuid4()), "run_id": "r"}
261 elif kind == "task":
262 rec = {"task_id": str(uuid.uuid4()), "run_id": "r"}
263 elif kind == "claim":
264 rec = {"task_id": str(uuid.uuid4()), "claimer_run_id": "cr", "run_id": "r"}
265 _write_records(repo, kind, [rec])
266
267 gathered = _gather_local_records(repo, list(_ALL_KINDS))
268 found_kinds = {g["kind"] for g in gathered}
269 assert found_kinds == set(_ALL_KINDS), f"Missing kinds: {set(_ALL_KINDS) - found_kinds}"
270
271 def test_unicode_payload_cjk_no_mojibake(self, repo: pathlib.Path) -> None:
272 uid = str(uuid.uuid4())
273 chinese = "作曲家は天才だ" # "The composer is a genius"
274 rec = {"reservation_id": uid, "run_id": "r", "label": chinese}
275 _write_records(repo, "reservation", [rec])
276 gathered = _gather_local_records(repo, ["reservation"])
277 assert gathered[0]["payload"]["label"] == chinese
278
279 def test_unicode_payload_emoji_no_mojibake(self, repo: pathlib.Path) -> None:
280 uid = str(uuid.uuid4())
281 notes = "🎵🎼🎹🎸🥁"
282 rec = {"reservation_id": uid, "run_id": "r", "notes": notes}
283 _write_records(repo, "reservation", [rec])
284 gathered = _gather_local_records(repo, ["reservation"])
285 assert gathered[0]["payload"]["notes"] == notes
286
287 def test_unicode_payload_arabic_rtl(self, repo: pathlib.Path) -> None:
288 uid = str(uuid.uuid4())
289 arabic = "مؤلف موسيقي"
290 rec = {"reservation_id": uid, "run_id": "r", "composer": arabic}
291 _write_records(repo, "reservation", [rec])
292 gathered = _gather_local_records(repo, ["reservation"])
293 assert gathered[0]["payload"]["composer"] == arabic
294
295 def test_deeply_nested_payload_preserved(self, repo: pathlib.Path) -> None:
296 uid = str(uuid.uuid4())
297 deep = {"a": {"b": {"c": {"d": {"e": {"f": "leaf"}}}}}}
298 rec = {"reservation_id": uid, "run_id": "r", "meta": deep}
299 _write_records(repo, "reservation", [rec])
300 gathered = _gather_local_records(repo, ["reservation"])
301 assert gathered[0]["payload"]["meta"]["a"]["b"]["c"]["d"]["e"]["f"] == "leaf"
302
303 def test_null_values_in_payload_preserved(self, repo: pathlib.Path) -> None:
304 uid = str(uuid.uuid4())
305 rec = {"reservation_id": uid, "run_id": "r", "nullable": None, "also": None}
306 _write_records(repo, "reservation", [rec])
307 gathered = _gather_local_records(repo, ["reservation"])
308 assert gathered[0]["payload"]["nullable"] is None
309 assert gathered[0]["payload"]["also"] is None
310
311 def test_boolean_payload_values_not_coerced_to_strings(self, repo: pathlib.Path) -> None:
312 uid = str(uuid.uuid4())
313 rec = {"reservation_id": uid, "run_id": "r", "flag": True, "other": False}
314 _write_records(repo, "reservation", [rec])
315 gathered = _gather_local_records(repo, ["reservation"])
316 assert gathered[0]["payload"]["flag"] is True
317 assert gathered[0]["payload"]["other"] is False
318
319 def test_integer_payload_values_not_coerced(self, repo: pathlib.Path) -> None:
320 uid = str(uuid.uuid4())
321 rec = {"reservation_id": uid, "run_id": "r", "count": 42, "pi": 3.14159}
322 _write_records(repo, "reservation", [rec])
323 gathered = _gather_local_records(repo, ["reservation"])
324 assert gathered[0]["payload"]["count"] == 42
325 assert abs(gathered[0]["payload"]["pi"] - 3.14159) < 1e-9
326
327 def test_50kb_payload_not_truncated(self, repo: pathlib.Path) -> None:
328 uid = str(uuid.uuid4())
329 big_string = "X" * 50_000
330 rec = {"reservation_id": uid, "run_id": "r", "blob": big_string}
331 _write_records(repo, "reservation", [rec])
332 gathered = _gather_local_records(repo, ["reservation"])
333 assert len(gathered[0]["payload"]["blob"]) == 50_000
334
335 def test_list_payload_values_preserved(self, repo: pathlib.Path) -> None:
336 uid = str(uuid.uuid4())
337 tags = ["linux", "kernel", "vcs", "muse"]
338 rec = {"reservation_id": uid, "run_id": "r", "tags": tags}
339 _write_records(repo, "reservation", [rec])
340 gathered = _gather_local_records(repo, ["reservation"])
341 assert gathered[0]["payload"]["tags"] == tags
342
343 def test_record_uuid_fallback_to_fpath_stem(self, repo: pathlib.Path) -> None:
344 """When reservation_id is absent, record_uuid falls back to file stem."""
345 stem = "fallback-stem-abc123"
346 subdir = _coord_dir(repo) / "reservations"
347 subdir.mkdir(parents=True, exist_ok=True)
348 (subdir / f"{stem}.json").write_text(
349 json.dumps({"run_id": "r"}), encoding="utf-8"
350 )
351 gathered = _gather_local_records(repo, ["reservation"])
352 assert gathered[0]["record_uuid"] == stem
353
354 def test_claim_uses_claimer_run_id_for_run_id(self, repo: pathlib.Path) -> None:
355 uid = str(uuid.uuid4())
356 rec = {"task_id": uid, "claimer_run_id": "claimer-run-99", "run_id": "other"}
357 _write_records(repo, "claim", [rec])
358 gathered = _gather_local_records(repo, ["claim"])
359 assert gathered[0]["run_id"] == "claimer-run-99"
360
361 def test_write_then_read_remote_roundtrip(self, repo: pathlib.Path) -> None:
362 """_write_remote_records → re-read from disk = original record."""
363 uid = str(uuid.uuid4())
364 orig = {"kind": "reservation", "record_uuid": uid, "run_id": "r",
365 "payload": {"x": 99}, "expires_at": "2099-06-01T00:00:00Z"}
366 _write_remote_records(repo, [orig])
367 target = repo / ".muse" / "coordination" / "remote" / "reservation" / f"{uid}.json"
368 assert target.exists()
369 loaded = json.loads(target.read_text(encoding="utf-8"))
370 assert loaded["payload"]["x"] == 99
371 assert loaded["expires_at"] == "2099-06-01T00:00:00Z"
372
373 def test_all_7_kinds_write_remote_and_read_back(self, repo: pathlib.Path) -> None:
374 records = []
375 for kind in _ALL_KINDS:
376 uid = str(uuid.uuid4())
377 records.append({"kind": kind, "record_uuid": uid, "run_id": "r",
378 "payload": {"kind_was": kind}, "expires_at": None})
379 _write_remote_records(repo, records)
380 for rec in records:
381 path = (
382 repo / ".muse" / "coordination" / "remote"
383 / rec["kind"] / f"{rec['record_uuid']}.json"
384 )
385 assert path.exists(), f"Missing: {path}"
386 loaded = json.loads(path.read_text())
387 assert loaded["payload"]["kind_was"] == rec["kind"]
388
389
390 # ===========================================================================
391 # 2. BATCHING CORRECTNESS
392 # ===========================================================================
393
394
395 class TestCoordIntegrityBatching:
396 """Correct batch splitting, counting, and failure isolation."""
397
398 def _make_n_reservations(self, repo: pathlib.Path, n: int) -> None:
399 subdir = _coord_dir(repo) / "reservations"
400 subdir.mkdir(parents=True, exist_ok=True)
401 for _ in range(n):
402 uid = str(uuid.uuid4())
403 (subdir / f"{uid}.json").write_text(
404 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
405 )
406
407 def test_exactly_500_records_is_one_batch(self, repo: pathlib.Path) -> None:
408 self._make_n_reservations(repo, 500)
409 with patch(_PUSH_TARGET, return_value=_push_ok(500)) as mock_push, \
410 patch(_REQUIRE_REPO, return_value=repo), \
411 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
412 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
413 assert result.exit_code == 0
414 assert mock_push.call_count == 1
415 data = json.loads(result.output.strip())
416 assert data["total"] == 500
417
418 def test_501_records_splits_into_two_batches(self, repo: pathlib.Path) -> None:
419 self._make_n_reservations(repo, 501)
420 batch_sizes: list[int] = []
421 def fake_push(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
422 batch_sizes.append(len(records))
423 return _push_ok(len(records))
424 with patch(_PUSH_TARGET, side_effect=fake_push), \
425 patch(_REQUIRE_REPO, return_value=repo), \
426 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
427 runner.invoke(cli, _PUSH_ARGS + ["-j"])
428 assert len(batch_sizes) == 2
429 assert batch_sizes[0] == 500
430 assert batch_sizes[1] == 1
431
432 def test_1000_records_splits_into_two_batches(self, repo: pathlib.Path) -> None:
433 self._make_n_reservations(repo, 1000)
434 batch_sizes: list[int] = []
435 def fake_push(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
436 batch_sizes.append(len(records))
437 return _push_ok(len(records))
438 with patch(_PUSH_TARGET, side_effect=fake_push), \
439 patch(_REQUIRE_REPO, return_value=repo), \
440 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
441 runner.invoke(cli, _PUSH_ARGS + ["-j"])
442 assert len(batch_sizes) == 2
443 assert all(s == 500 for s in batch_sizes)
444
445 def test_1001_records_splits_into_three_batches(self, repo: pathlib.Path) -> None:
446 self._make_n_reservations(repo, 1001)
447 batch_sizes: list[int] = []
448 def fake_push(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
449 batch_sizes.append(len(records))
450 return _push_ok(len(records))
451 with patch(_PUSH_TARGET, side_effect=fake_push), \
452 patch(_REQUIRE_REPO, return_value=repo), \
453 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
454 runner.invoke(cli, _PUSH_ARGS + ["-j"])
455 assert len(batch_sizes) == 3
456 assert batch_sizes[0] == 500
457 assert batch_sizes[1] == 500
458 assert batch_sizes[2] == 1
459
460 def test_4999_records_linux_scale_10_batches(self, repo: pathlib.Path) -> None:
461 """Linux-scale: 4999 records → 9×500 + 1×499 = 10 calls."""
462 self._make_n_reservations(repo, 4999)
463 batch_sizes: list[int] = []
464 def fake_push(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
465 batch_sizes.append(len(records))
466 return _push_ok(len(records))
467 with patch(_PUSH_TARGET, side_effect=fake_push), \
468 patch(_REQUIRE_REPO, return_value=repo), \
469 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
470 runner.invoke(cli, _PUSH_ARGS + ["-j"])
471 assert len(batch_sizes) == 10
472 assert sum(batch_sizes) == 4999
473 assert batch_sizes[-1] == 499
474
475 def test_first_batch_fails_remaining_batches_still_execute(self, repo: pathlib.Path) -> None:
476 """A single batch failure must not abort the whole push."""
477 self._make_n_reservations(repo, 1001)
478 call_count = 0
479 def fail_first_then_ok(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
480 nonlocal call_count
481 call_count += 1
482 if call_count == 1:
483 raise CoordBusError("network hiccup", status_code=503)
484 return _push_ok(len(records))
485 with patch(_PUSH_TARGET, side_effect=fail_first_then_ok), \
486 patch(_REQUIRE_REPO, return_value=repo), \
487 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
488 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
489 assert result.exit_code == 1 # failed=True → exit 1
490 # On batch error, _err() emits an error JSON line first; summary is the last line.
491 last_line = [l for l in result.output.strip().splitlines() if l.strip()][-1]
492 data = json.loads(last_line)
493 assert data["failed"] is True
494 assert data["inserted"] == 501 # batches 2 and 3 succeeded
495
496 def test_all_batches_fail_inserted_is_zero(self, repo: pathlib.Path) -> None:
497 self._make_n_reservations(repo, 600)
498 with patch(_PUSH_TARGET, side_effect=CoordBusError("gone", status_code=503)), \
499 patch(_REQUIRE_REPO, return_value=repo), \
500 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
501 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
502 assert result.exit_code == 1
503 last_line = [l for l in result.output.strip().splitlines() if l.strip()][-1]
504 data = json.loads(last_line)
505 assert data["inserted"] == 0
506 assert data["failed"] is True
507
508 def test_batch_slices_are_non_overlapping_and_exhaustive(self, repo: pathlib.Path) -> None:
509 """Every gathered record appears in exactly one batch — no duplicates, no gaps."""
510 n = 1200
511 self._make_n_reservations(repo, n)
512 all_received: list[MsgpackDict] = []
513 def collect(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
514 all_received.extend(records)
515 return _push_ok(len(records))
516 with patch(_PUSH_TARGET, side_effect=collect), \
517 patch(_REQUIRE_REPO, return_value=repo), \
518 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
519 runner.invoke(cli, _PUSH_ARGS + ["-j"])
520 assert len(all_received) == n
521 # No duplicates (record_uuid uniqueness)
522 uuids = [r["record_uuid"] for r in all_received]
523 assert len(uuids) == len(set(uuids)), "Duplicate records in batches!"
524
525 def test_inserted_skipped_counts_accumulated_across_batches(self, repo: pathlib.Path) -> None:
526 """total inserted/skipped in JSON output = sum of all batch results."""
527 self._make_n_reservations(repo, 1100)
528 def mixed(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
529 # Odd batches: half inserted, half skipped
530 n = len(records)
531 return _push_ok(n // 2, n - n // 2)
532 with patch(_PUSH_TARGET, side_effect=mixed), \
533 patch(_REQUIRE_REPO, return_value=repo), \
534 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
535 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
536 data = json.loads(result.output.strip())
537 # 3 batches: 500, 500, 100 → inserted = 250+250+50=550, skipped = 250+250+50=550
538 assert data["inserted"] + data["skipped"] == 1100
539
540
541 # ===========================================================================
542 # 3. CURSOR CORRECTNESS
543 # ===========================================================================
544
545
546 class TestCoordIntegrityCursor:
547 """Incremental pull semantics: cursor must advance, no gaps, no duplicates."""
548
549 def test_cursor_zero_when_no_records_returned(self, repo: pathlib.Path) -> None:
550 with patch(_PULL_TARGET, return_value=_pull_ok([], cursor=0)), \
551 patch(_REQUIRE_REPO, return_value=repo), \
552 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
553 result = runner.invoke(cli, _PULL_ARGS + ["-j"])
554 data = json.loads(result.output.strip())
555 assert data["cursor"] == 0
556
557 def test_cursor_equals_last_record_id(self, repo: pathlib.Path) -> None:
558 records = [
559 {"kind": "reservation", "record_uuid": str(uuid.uuid4()), "id": 42,
560 "run_id": "r", "payload": {}},
561 ]
562 with patch(_PULL_TARGET, return_value={"records": records, "cursor": 42}), \
563 patch(_REQUIRE_REPO, return_value=repo), \
564 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
565 result = runner.invoke(cli, _PULL_ARGS + ["-j"])
566 data = json.loads(result.output.strip())
567 assert data["cursor"] == 42
568
569 def test_since_id_passed_through_verbatim(self, repo: pathlib.Path) -> None:
570 """--since-id must be forwarded to pull_from_hub unchanged."""
571 with patch(_PULL_TARGET, return_value=_pull_ok([])) as mock_pull, \
572 patch(_REQUIRE_REPO, return_value=repo), \
573 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
574 runner.invoke(cli, _PULL_ARGS + ["--since-id", "9999"])
575 positional = mock_pull.call_args[0]
576 assert positional[3] == 9999 # since_id is 4th positional arg
577
578 def test_incremental_5_pages_no_record_id_gaps(self, repo: pathlib.Path) -> None:
579 """5 pages of 200 records each — all 1000 IDs must be contiguous."""
580 page_size = 200
581 all_ids: list[int] = []
582 cursor = 0
583
584 for page in range(5):
585 start = page * page_size + 1
586 end = start + page_size
587 records = [
588 {"kind": "reservation", "record_uuid": str(uuid.uuid4()),
589 "id": i, "run_id": "r", "payload": {}}
590 for i in range(start, end)
591 ]
592 new_cursor = end - 1
593 # Simulate: pull returns page of records, then we advance cursor
594 with patch(_PULL_TARGET, return_value={"records": records, "cursor": new_cursor}), \
595 patch(_REQUIRE_REPO, return_value=repo), \
596 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
597 result = runner.invoke(
598 cli, _PULL_ARGS + ["--since-id", str(cursor), "-j"]
599 )
600 data = json.loads(result.output.strip())
601 all_ids.extend(r["id"] for r in data["records"])
602 cursor = data["cursor"]
603
604 assert len(all_ids) == 1000
605 assert all_ids == list(range(1, 1001)), "Gaps or non-contiguous IDs detected!"
606
607 def test_page_chained_pull_zero_duplicates(self, repo: pathlib.Path) -> None:
608 """Simulate 3 pages; each record UUID must appear exactly once."""
609 all_uuids: list[str] = []
610 cursor = 0
611
612 for page in range(3):
613 page_records = [
614 {"kind": "heartbeat", "record_uuid": str(uuid.uuid4()),
615 "run_id": f"r-{page}-{i}", "payload": {}}
616 for i in range(100)
617 ]
618 new_cursor = (page + 1) * 100
619 with patch(_PULL_TARGET, return_value={"records": page_records, "cursor": new_cursor}), \
620 patch(_REQUIRE_REPO, return_value=repo), \
621 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
622 result = runner.invoke(
623 cli, _PULL_ARGS + ["--since-id", str(cursor), "-j"]
624 )
625 data = json.loads(result.output.strip())
626 all_uuids.extend(r["record_uuid"] for r in data["records"])
627 cursor = data["cursor"]
628
629 assert len(all_uuids) == len(set(all_uuids)), "Duplicate record UUIDs across pages!"
630
631 def test_cursor_in_json_output_matches_hub_cursor(self, repo: pathlib.Path) -> None:
632 expected_cursor = 77777
633 with patch(_PULL_TARGET, return_value={"records": [], "cursor": expected_cursor}), \
634 patch(_REQUIRE_REPO, return_value=repo), \
635 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
636 result = runner.invoke(cli, _PULL_ARGS + ["-j"])
637 data = json.loads(result.output.strip())
638 assert data["cursor"] == expected_cursor
639
640 def test_since_id_zero_returns_all_records_from_beginning(self, repo: pathlib.Path) -> None:
641 records = [
642 {"kind": "task", "record_uuid": str(uuid.uuid4()), "run_id": "r",
643 "payload": {"i": i}}
644 for i in range(50)
645 ]
646 with patch(_PULL_TARGET, return_value={"records": records, "cursor": 50}), \
647 patch(_REQUIRE_REPO, return_value=repo), \
648 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
649 result = runner.invoke(cli, _PULL_ARGS + ["--since-id", "0", "-j"])
650 data = json.loads(result.output.strip())
651 assert data["count"] == 50
652 assert data["cursor"] == 50
653
654
655 # ===========================================================================
656 # 4. RESILIENCE
657 # ===========================================================================
658
659
660 class TestCoordIntegrityResilience:
661 """Corrupt files, disk errors, and race conditions must never block good data."""
662
663 def test_corrupt_json_skipped_others_gathered(self, repo: pathlib.Path) -> None:
664 subdir = _coord_dir(repo) / "reservations"
665 subdir.mkdir(parents=True, exist_ok=True)
666 good_uid = str(uuid.uuid4())
667 (subdir / f"{good_uid}.json").write_text(
668 json.dumps({"reservation_id": good_uid, "run_id": "r"}), encoding="utf-8"
669 )
670 (subdir / "corrupt.json").write_text("{not valid json !!!", encoding="utf-8")
671 gathered = _gather_local_records(repo, ["reservation"])
672 assert len(gathered) == 1
673 assert gathered[0]["record_uuid"] == good_uid
674
675 def test_binary_garbage_file_skipped(self, repo: pathlib.Path) -> None:
676 subdir = _coord_dir(repo) / "heartbeats"
677 subdir.mkdir(parents=True, exist_ok=True)
678 uid = str(uuid.uuid4())
679 (subdir / f"{uid}.json").write_text(
680 json.dumps({"run_id": uid}), encoding="utf-8"
681 )
682 (subdir / "binary.json").write_bytes(bytes(range(256)))
683 gathered = _gather_local_records(repo, ["heartbeat"])
684 assert len(gathered) == 1
685
686 def test_null_byte_in_file_skipped(self, repo: pathlib.Path) -> None:
687 subdir = _coord_dir(repo) / "tasks"
688 subdir.mkdir(parents=True, exist_ok=True)
689 uid = str(uuid.uuid4())
690 (subdir / f"{uid}.json").write_text(
691 json.dumps({"task_id": uid, "run_id": "r"}), encoding="utf-8"
692 )
693 (subdir / "nullbyte.json").write_bytes(b'{"task_id": "x\x00y"}')
694 gathered = _gather_local_records(repo, ["task"])
695 # Null byte may or may not parse; either way, only 1 valid record matters
696 assert any(g["record_uuid"] == uid for g in gathered)
697
698 def test_file_deleted_mid_glob_skipped_gracefully(self, repo: pathlib.Path) -> None:
699 """Simulate FileNotFoundError between glob and read_text."""
700 subdir = _coord_dir(repo) / "reservations"
701 subdir.mkdir(parents=True, exist_ok=True)
702 uid = str(uuid.uuid4())
703 ghost = subdir / f"{uid}.json"
704 ghost.write_text(json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8")
705
706 _original_read_text = pathlib.Path.read_text
707
708 def raise_on_ghost(self: pathlib.Path, encoding: str | None = None, errors: str | None = None) -> str:
709 if self == ghost:
710 raise FileNotFoundError("deleted mid-glob")
711 return _original_read_text(self, encoding=encoding, errors=errors)
712
713 with patch.object(pathlib.Path, "read_text", raise_on_ghost):
714 gathered = _gather_local_records(repo, ["reservation"])
715 assert gathered == []
716
717 def test_all_files_corrupt_zero_records_push_exits_0(self, repo: pathlib.Path) -> None:
718 subdir = _coord_dir(repo) / "reservations"
719 subdir.mkdir(parents=True, exist_ok=True)
720 for i in range(10):
721 (subdir / f"bad_{i}.json").write_text("{{broken", encoding="utf-8")
722 with patch(_REQUIRE_REPO, return_value=repo), \
723 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
724 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
725 assert result.exit_code == 0
726 data = json.loads(result.output.strip())
727 assert data["total"] == 0
728 assert data["failed"] is False
729
730 def test_500_good_500_corrupt_only_good_pushed(self, repo: pathlib.Path) -> None:
731 subdir = _coord_dir(repo) / "reservations"
732 subdir.mkdir(parents=True, exist_ok=True)
733 good_uids = set()
734 for _ in range(500):
735 uid = str(uuid.uuid4())
736 good_uids.add(uid)
737 (subdir / f"{uid}.json").write_text(
738 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
739 )
740 for i in range(500):
741 (subdir / f"corrupt_{i}.json").write_text("{bad", encoding="utf-8")
742
743 gathered = _gather_local_records(repo, ["reservation"])
744 assert len(gathered) == 500
745 assert {g["record_uuid"] for g in gathered} == good_uids
746
747 def test_disk_full_on_write_remote_propagates(self, repo: pathlib.Path) -> None:
748 """OSError from write_text_atomic must NOT be swallowed."""
749 uid = str(uuid.uuid4())
750 records = [{"kind": "reservation", "record_uuid": uid, "run_id": "r",
751 "payload": {}, "expires_at": None}]
752 with patch("muse.cli.commands.coord_sync.write_text_atomic",
753 side_effect=OSError("No space left")):
754 with pytest.raises(OSError, match="No space left"):
755 _write_remote_records(repo, records)
756
757 def test_permission_error_on_mkdir_propagates(self, repo: pathlib.Path) -> None:
758 uid = str(uuid.uuid4())
759 records = [{"kind": "heartbeat", "record_uuid": uid, "run_id": "r",
760 "payload": {}, "expires_at": None}]
761 with patch.object(pathlib.Path, "mkdir", side_effect=PermissionError("read-only")):
762 with pytest.raises(PermissionError, match="read-only"):
763 _write_remote_records(repo, records)
764
765 def test_empty_coordination_dir_returns_no_records(self, repo: pathlib.Path) -> None:
766 # No coord dir at all
767 gathered = _gather_local_records(repo, list(_ALL_KINDS))
768 assert gathered == []
769
770 def test_empty_kind_subdir_returns_no_records(self, repo: pathlib.Path) -> None:
771 (_coord_dir(repo) / "reservations").mkdir(parents=True, exist_ok=True)
772 gathered = _gather_local_records(repo, ["reservation"])
773 assert gathered == []
774
775
776 # ===========================================================================
777 # 5. CONCURRENCY
778 # ===========================================================================
779
780
781 class TestCoordIntegrityConcurrency:
782 """No data races, no corruption under concurrent access."""
783
784 def test_16_threads_gather_same_records(self, repo: pathlib.Path) -> None:
785 """All threads must see the same number of records."""
786 subdir = _coord_dir(repo) / "reservations"
787 subdir.mkdir(parents=True, exist_ok=True)
788 for _ in range(200):
789 uid = str(uuid.uuid4())
790 (subdir / f"{uid}.json").write_text(
791 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
792 )
793 counts: list[int] = []
794 errors: list[str] = []
795
796 def worker() -> None:
797 try:
798 recs = _gather_local_records(repo, ["reservation"])
799 counts.append(len(recs))
800 except Exception as exc:
801 errors.append(str(exc))
802
803 threads = [threading.Thread(target=worker) for _ in range(16)]
804 for t in threads:
805 t.start()
806 for t in threads:
807 t.join()
808
809 assert not errors, f"Errors: {errors}"
810 assert all(c == 200 for c in counts), f"Count mismatch: {counts}"
811
812 def test_8_threads_write_distinct_uuids_all_files_present(self, repo: pathlib.Path) -> None:
813 """8 writers, each writing a unique UUID → all 8 files must exist."""
814 uids = [str(uuid.uuid4()) for _ in range(8)]
815 errors: list[str] = []
816
817 def worker(uid: str) -> None:
818 try:
819 rec = {"kind": "reservation", "record_uuid": uid, "run_id": "r",
820 "payload": {"uid": uid}, "expires_at": None}
821 _write_remote_records(repo, [rec])
822 except Exception as exc:
823 errors.append(f"{uid}: {exc}")
824
825 threads = [threading.Thread(target=worker, args=(u,)) for u in uids]
826 for t in threads:
827 t.start()
828 for t in threads:
829 t.join()
830
831 assert not errors, f"Errors: {errors}"
832 remote = repo / ".muse" / "coordination" / "remote" / "reservation"
833 written = {f.stem for f in remote.glob("*.json")}
834 assert written == set(uids), f"Missing: {set(uids) - written}"
835
836 def test_8_threads_write_same_uuid_final_file_is_valid_json(
837 self, repo: pathlib.Path
838 ) -> None:
839 """Last-writer-wins on same UUID must produce valid JSON, not corruption."""
840 uid = str(uuid.uuid4())
841 errors: list[str] = []
842
843 def worker(i: int) -> None:
844 try:
845 rec = {"kind": "task", "record_uuid": uid, "run_id": f"r{i}",
846 "payload": {"version": i}, "expires_at": None}
847 _write_remote_records(repo, [rec])
848 except Exception as exc:
849 errors.append(f"Thread {i}: {exc}")
850
851 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
852 for t in threads:
853 t.start()
854 for t in threads:
855 t.join()
856
857 target = repo / ".muse" / "coordination" / "remote" / "task" / f"{uid}.json"
858 assert target.exists()
859 content = target.read_text(encoding="utf-8")
860 loaded = json.loads(content) # Must not raise — file must be valid JSON
861 assert loaded["record_uuid"] == uid
862
863 def test_run_push_8_concurrent_threads_no_crashes(self, repo: pathlib.Path) -> None:
864 subdir = _coord_dir(repo) / "reservations"
865 subdir.mkdir(parents=True, exist_ok=True)
866 for _ in range(50):
867 uid = str(uuid.uuid4())
868 (subdir / f"{uid}.json").write_text(
869 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
870 )
871
872 errors: list[str] = []
873
874 def worker(idx: int) -> None:
875 args = argparse.Namespace(
876 hub="http://localhost:10003",
877 owner="gabriel",
878 slug="linux",
879 signing=None,
880 kinds=["reservation"],
881 fmt="json",
882 )
883 try:
884 run_push(args)
885 except SystemExit as exc:
886 if exc.code not in (0, 1):
887 errors.append(f"Thread {idx}: unexpected exit {exc.code}")
888 except Exception as exc:
889 errors.append(f"Thread {idx}: {exc}")
890
891 push_p = patch(_PUSH_TARGET, return_value=_push_ok(50))
892 repo_p = patch(_REQUIRE_REPO, return_value=repo)
893 hub_p = patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok"))
894 push_p.start(); repo_p.start(); hub_p.start()
895 try:
896 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
897 for t in threads:
898 t.start()
899 for t in threads:
900 t.join()
901 finally:
902 push_p.stop(); repo_p.stop(); hub_p.stop()
903 assert not errors, f"Concurrent push errors: {errors}"
904
905 def test_run_pull_8_concurrent_threads_no_crashes(self, repo: pathlib.Path) -> None:
906 errors: list[str] = []
907
908 def worker(idx: int) -> None:
909 args = argparse.Namespace(
910 hub="http://localhost:10003",
911 owner="gabriel",
912 slug="linux",
913 signing=None,
914 since_id=0,
915 kinds=[],
916 limit=500,
917 fmt="json",
918 )
919 try:
920 run_pull(args)
921 except SystemExit as exc:
922 if exc.code not in (0, 1):
923 errors.append(f"Thread {idx}: unexpected exit {exc.code}")
924 except Exception as exc:
925 errors.append(f"Thread {idx}: {exc}")
926
927 pull_p = patch(_PULL_TARGET, return_value=_pull_ok([], cursor=0))
928 repo_p = patch(_REQUIRE_REPO, return_value=repo)
929 hub_p = patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok"))
930 pull_p.start(); repo_p.start(); hub_p.start()
931 try:
932 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
933 for t in threads:
934 t.start()
935 for t in threads:
936 t.join()
937 finally:
938 pull_p.stop(); repo_p.stop(); hub_p.stop()
939 assert not errors, f"Concurrent pull errors: {errors}"
940
941
942 # ===========================================================================
943 # 6. LINUX-SCALE
944 # ===========================================================================
945
946
947 class TestCoordIntegrityScale:
948 """Linux-scale operations: thousands of records across all 7 kinds."""
949
950 _KIND_SUBDIR = {
951 "reservation": "reservations",
952 "intent": "intents",
953 "release": "releases",
954 "heartbeat": "heartbeats",
955 "dependency": "dependencies",
956 "task": "tasks",
957 "claim": "claims",
958 }
959 _KIND_ID_FIELD = {
960 "reservation": "reservation_id",
961 "intent": "intent_id",
962 "release": "release_id",
963 "heartbeat": "run_id",
964 "dependency": "reservation_id",
965 "task": "task_id",
966 "claim": "task_id",
967 }
968
969 def _populate_all_kinds(self, repo: pathlib.Path, count: int) -> None:
970 for kind in _ALL_KINDS:
971 subdir = _coord_dir(repo) / self._KIND_SUBDIR[kind]
972 subdir.mkdir(parents=True, exist_ok=True)
973 id_field = self._KIND_ID_FIELD[kind]
974 for _ in range(count):
975 uid = str(uuid.uuid4())
976 rec: MsgpackDict = {id_field: uid, "run_id": "r"}
977 if kind == "claim":
978 rec["claimer_run_id"] = "cr"
979 (subdir / f"{uid}.json").write_text(json.dumps(rec), encoding="utf-8")
980
981 def test_7000_records_all_7_kinds_gathered(self, repo: pathlib.Path) -> None:
982 """1000 of each kind (7000 total) — all must be gathered."""
983 self._populate_all_kinds(repo, 1000)
984
985 gathered = _gather_local_records(repo, list(_ALL_KINDS))
986 assert len(gathered) == 7000
987 by_kind = {}
988 for g in gathered:
989 by_kind[g["kind"]] = by_kind.get(g["kind"], 0) + 1
990 assert all(by_kind.get(k, 0) == 1000 for k in _ALL_KINDS), f"Kind counts: {by_kind}"
991
992 def test_7000_records_batched_and_pushed_14_calls(self, repo: pathlib.Path) -> None:
993 self._populate_all_kinds(repo, 1000)
994
995 call_count = 0
996 total_sent = 0
997
998 def count_calls(hub_url: str, owner: str, slug: str, records: list[MsgpackDict], token: str | None = None) -> MsgpackDict:
999 nonlocal call_count, total_sent
1000 call_count += 1
1001 total_sent += len(records)
1002 return _push_ok(len(records))
1003
1004 with patch(_PUSH_TARGET, side_effect=count_calls), \
1005 patch(_REQUIRE_REPO, return_value=repo), \
1006 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1007 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1008
1009 assert result.exit_code == 0
1010 assert call_count == 14 # ceil(7000/500) = 14
1011 assert total_sent == 7000
1012
1013 def test_1000_record_pull_all_files_written(self, repo: pathlib.Path) -> None:
1014 """Pull returns MAX_PULL_LIMIT records — all written to remote dir."""
1015 records = [
1016 {"kind": "reservation", "record_uuid": str(uuid.uuid4()),
1017 "run_id": f"r{i}", "payload": {}, "expires_at": None}
1018 for i in range(1000)
1019 ]
1020 uuids = {r["record_uuid"] for r in records}
1021 with patch(_PULL_TARGET, return_value={"records": records, "cursor": 1000}), \
1022 patch(_REQUIRE_REPO, return_value=repo), \
1023 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1024 result = runner.invoke(cli, _PULL_ARGS + ["-j"])
1025 assert result.exit_code == 0
1026 remote = repo / ".muse" / "coordination" / "remote" / "reservation"
1027 written = {f.stem for f in remote.glob("*.json")}
1028 assert written == uuids
1029
1030 def test_mixed_1000_reservations_1000_heartbeats_total_2000(
1031 self, repo: pathlib.Path
1032 ) -> None:
1033 res_subdir = _coord_dir(repo) / "reservations"
1034 hb_subdir = _coord_dir(repo) / "heartbeats"
1035 res_subdir.mkdir(parents=True, exist_ok=True)
1036 hb_subdir.mkdir(parents=True, exist_ok=True)
1037 for _ in range(1000):
1038 uid = str(uuid.uuid4())
1039 (res_subdir / f"{uid}.json").write_text(
1040 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1041 )
1042 uid2 = str(uuid.uuid4())
1043 (hb_subdir / f"{uid2}.json").write_text(
1044 json.dumps({"run_id": uid2}), encoding="utf-8"
1045 )
1046 gathered = _gather_local_records(repo, ["reservation", "heartbeat"])
1047 assert len(gathered) == 2000
1048
1049 def test_50_sequential_push_pull_cycles_integrity(self, repo: pathlib.Path) -> None:
1050 """50 cycles: push 10, pull 10, verify no data loss across cycles."""
1051 subdir = _coord_dir(repo) / "reservations"
1052 subdir.mkdir(parents=True, exist_ok=True)
1053 total_inserted = 0
1054 total_pulled = 0
1055 cursor = 0
1056
1057 for cycle in range(50):
1058 # Write 10 new records
1059 cycle_uids = []
1060 for _ in range(10):
1061 uid = str(uuid.uuid4())
1062 cycle_uids.append(uid)
1063 (subdir / f"{uid}.json").write_text(
1064 json.dumps({"reservation_id": uid, "run_id": f"r{cycle}"}),
1065 encoding="utf-8",
1066 )
1067
1068 # Push
1069 with patch(_PUSH_TARGET, return_value=_push_ok(10)), \
1070 patch(_REQUIRE_REPO, return_value=repo), \
1071 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1072 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1073 assert result.exit_code == 0
1074 push_data = json.loads(result.output.strip())
1075 # total accumulates: gather reads ALL files in subdir each cycle
1076 total_inserted += push_data["inserted"]
1077
1078 # Pull 10 new records
1079 pull_records = [
1080 {"kind": "reservation", "record_uuid": u, "run_id": f"r{cycle}",
1081 "payload": {}, "expires_at": None}
1082 for u in cycle_uids
1083 ]
1084 cursor += 10
1085 with patch(_PULL_TARGET, return_value={"records": pull_records, "cursor": cursor}), \
1086 patch(_REQUIRE_REPO, return_value=repo), \
1087 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1088 result = runner.invoke(
1089 cli, _PULL_ARGS + ["--since-id", str(cursor - 10), "-j"]
1090 )
1091 assert result.exit_code == 0
1092 pull_data = json.loads(result.output.strip())
1093 total_pulled += pull_data["count"]
1094
1095 assert total_pulled == 500 # 50 cycles × 10 records each
1096
1097
1098 # ===========================================================================
1099 # 7. RESPONSE BOUNDS
1100 # ===========================================================================
1101
1102
1103 class TestCoordIntegrityResponseBounds:
1104 """Hub response edge cases — missing keys, oversized bodies, malformed JSON."""
1105
1106 def test_response_missing_inserted_key_defaults_to_zero(
1107 self, repo: pathlib.Path
1108 ) -> None:
1109 with patch(_PUSH_TARGET, return_value={"skipped": 3}), \
1110 patch(_REQUIRE_REPO, return_value=repo), \
1111 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1112 subdir = _coord_dir(repo) / "reservations"
1113 subdir.mkdir(parents=True, exist_ok=True)
1114 uid = str(uuid.uuid4())
1115 (subdir / f"{uid}.json").write_text(
1116 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1117 )
1118 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1119 assert result.exit_code == 0
1120 data = json.loads(result.output.strip())
1121 assert data["inserted"] == 0 # defaults to 0
1122 assert data["skipped"] == 3
1123
1124 def test_response_missing_skipped_key_defaults_to_zero(
1125 self, repo: pathlib.Path
1126 ) -> None:
1127 with patch(_PUSH_TARGET, return_value={"inserted": 5}), \
1128 patch(_REQUIRE_REPO, return_value=repo), \
1129 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1130 subdir = _coord_dir(repo) / "reservations"
1131 subdir.mkdir(parents=True, exist_ok=True)
1132 uid = str(uuid.uuid4())
1133 (subdir / f"{uid}.json").write_text(
1134 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1135 )
1136 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1137 assert result.exit_code == 0
1138 data = json.loads(result.output.strip())
1139 assert data["inserted"] == 5
1140 assert data["skipped"] == 0 # defaults to 0
1141
1142 def test_response_extra_keys_silently_ignored(self, repo: pathlib.Path) -> None:
1143 resp = {"inserted": 1, "skipped": 0, "unknown_future_key": "value", "debug": {}}
1144 with patch(_PUSH_TARGET, return_value=resp), \
1145 patch(_REQUIRE_REPO, return_value=repo), \
1146 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1147 subdir = _coord_dir(repo) / "reservations"
1148 subdir.mkdir(parents=True, exist_ok=True)
1149 uid = str(uuid.uuid4())
1150 (subdir / f"{uid}.json").write_text(
1151 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1152 )
1153 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1154 assert result.exit_code == 0
1155
1156 def test_coord_bus_error_on_push_exits_1(self, repo: pathlib.Path) -> None:
1157 with patch(_PUSH_TARGET, side_effect=CoordBusError("service unavailable", status_code=503)), \
1158 patch(_REQUIRE_REPO, return_value=repo), \
1159 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1160 subdir = _coord_dir(repo) / "reservations"
1161 subdir.mkdir(parents=True, exist_ok=True)
1162 uid = str(uuid.uuid4())
1163 (subdir / f"{uid}.json").write_text(
1164 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1165 )
1166 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1167 assert result.exit_code == 1
1168 last_line = [l for l in result.output.strip().splitlines() if l.strip()][-1]
1169 data = json.loads(last_line)
1170 assert data["failed"] is True
1171
1172 def test_coord_bus_error_on_pull_exits_1(self, repo: pathlib.Path) -> None:
1173 with patch(_PULL_TARGET, side_effect=CoordBusError("gateway timeout", status_code=504)), \
1174 patch(_REQUIRE_REPO, return_value=repo), \
1175 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1176 result = runner.invoke(cli, _PULL_ARGS + ["-j"])
1177 assert result.exit_code == 1
1178
1179 def test_pull_response_missing_records_key_no_files_written(
1180 self, repo: pathlib.Path
1181 ) -> None:
1182 with patch(_PULL_TARGET, return_value={"cursor": 0}), \
1183 patch(_REQUIRE_REPO, return_value=repo), \
1184 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1185 result = runner.invoke(cli, _PULL_ARGS + ["-j"])
1186 assert result.exit_code == 0
1187 data = json.loads(result.output.strip())
1188 assert data["count"] == 0
1189
1190 def test_pull_response_missing_cursor_key_defaults_to_zero(
1191 self, repo: pathlib.Path
1192 ) -> None:
1193 records = [
1194 {"kind": "task", "record_uuid": str(uuid.uuid4()), "run_id": "r", "payload": {}}
1195 ]
1196 with patch(_PULL_TARGET, return_value={"records": records}), \
1197 patch(_REQUIRE_REPO, return_value=repo), \
1198 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1199 result = runner.invoke(cli, _PULL_ARGS + ["-j"])
1200 assert result.exit_code == 0
1201 data = json.loads(result.output.strip())
1202 assert data["cursor"] == 0 # defaults to 0
1203
1204
1205 # ===========================================================================
1206 # 8. FILESYSTEM SAFETY
1207 # ===========================================================================
1208
1209
1210 class TestCoordIntegrityFilesystem:
1211 """Path traversal prevention, UUID validation, remote dir creation."""
1212
1213 def test_traversal_uuid_does_not_escape_remote_dir(self, repo: pathlib.Path) -> None:
1214 evil = "../../../etc/passwd"
1215 records = [{"kind": "reservation", "record_uuid": evil, "run_id": "r",
1216 "payload": {}, "expires_at": None}]
1217 _write_remote_records(repo, records)
1218 # Must not exist outside remote/reservation/
1219 assert not (repo / ".muse" / "coordination" / "remote" / "reservation" /
1220 "../../etc/passwd").exists()
1221 # The remote dir might or might not exist; if it does, no evil files
1222 remote = repo / ".muse" / "coordination" / "remote"
1223 if remote.exists():
1224 all_files = list(remote.rglob("*.json"))
1225 for f in all_files:
1226 assert ".." not in str(f), f"Traversal escaped: {f}"
1227
1228 def test_traversal_uuid_null_byte_rejected(self, repo: pathlib.Path) -> None:
1229 evil = "foo\x00bar"
1230 records = [{"kind": "task", "record_uuid": evil, "run_id": "r",
1231 "payload": {}, "expires_at": None}]
1232 _write_remote_records(repo, records)
1233 remote = repo / ".muse" / "coordination" / "remote"
1234 if remote.exists():
1235 assert list(remote.rglob("*.json")) == []
1236
1237 def test_uuid_128_chars_accepted(self, repo: pathlib.Path) -> None:
1238 uid = "a" * 128
1239 records = [{"kind": "reservation", "record_uuid": uid, "run_id": "r",
1240 "payload": {}, "expires_at": None}]
1241 _write_remote_records(repo, records)
1242 target = repo / ".muse" / "coordination" / "remote" / "reservation" / f"{uid}.json"
1243 assert target.exists()
1244
1245 def test_uuid_129_chars_rejected(self, repo: pathlib.Path) -> None:
1246 uid = "a" * 129
1247 assert not _SAFE_RECORD_UUID_RE.match(uid), "Regex should reject 129-char UUID"
1248 records = [{"kind": "reservation", "record_uuid": uid, "run_id": "r",
1249 "payload": {}, "expires_at": None}]
1250 _write_remote_records(repo, records)
1251 remote = repo / ".muse" / "coordination" / "remote"
1252 if remote.exists():
1253 assert list(remote.rglob("*.json")) == []
1254
1255 def test_kind_traversal_rejected(self, repo: pathlib.Path) -> None:
1256 records = [{"kind": "../evil", "record_uuid": "safe-uuid", "run_id": "r",
1257 "payload": {}, "expires_at": None}]
1258 _write_remote_records(repo, records)
1259 remote = repo / ".muse" / "coordination" / "remote"
1260 if remote.exists():
1261 assert list(remote.rglob("*.json")) == []
1262
1263 def test_empty_kind_rejected(self, repo: pathlib.Path) -> None:
1264 records = [{"kind": "", "record_uuid": "safe-uuid", "run_id": "r",
1265 "payload": {}, "expires_at": None}]
1266 _write_remote_records(repo, records)
1267 remote = repo / ".muse" / "coordination" / "remote"
1268 if remote.exists():
1269 assert list(remote.rglob("*.json")) == []
1270
1271 def test_empty_uuid_rejected(self, repo: pathlib.Path) -> None:
1272 records = [{"kind": "reservation", "record_uuid": "", "run_id": "r",
1273 "payload": {}, "expires_at": None}]
1274 _write_remote_records(repo, records)
1275 remote = repo / ".muse" / "coordination" / "remote"
1276 if remote.exists():
1277 assert list(remote.rglob("*.json")) == []
1278
1279 def test_remote_dir_created_automatically(self, repo: pathlib.Path) -> None:
1280 uid = str(uuid.uuid4())
1281 records = [{"kind": "reservation", "record_uuid": uid, "run_id": "r",
1282 "payload": {}, "expires_at": None}]
1283 remote = repo / ".muse" / "coordination" / "remote"
1284 assert not remote.exists()
1285 _write_remote_records(repo, records)
1286 assert remote.exists()
1287 assert (remote / "reservation" / f"{uid}.json").exists()
1288
1289 def test_existing_remote_file_overwritten(self, repo: pathlib.Path) -> None:
1290 uid = str(uuid.uuid4())
1291 rec_v1 = {"kind": "reservation", "record_uuid": uid, "run_id": "r",
1292 "payload": {"version": 1}, "expires_at": None}
1293 rec_v2 = {"kind": "reservation", "record_uuid": uid, "run_id": "r",
1294 "payload": {"version": 2}, "expires_at": None}
1295 _write_remote_records(repo, [rec_v1])
1296 _write_remote_records(repo, [rec_v2])
1297 target = repo / ".muse" / "coordination" / "remote" / "reservation" / f"{uid}.json"
1298 loaded = json.loads(target.read_text())
1299 assert loaded["payload"]["version"] == 2
1300
1301 def test_written_files_are_valid_json(self, repo: pathlib.Path) -> None:
1302 records = [
1303 {"kind": k, "record_uuid": str(uuid.uuid4()), "run_id": "r",
1304 "payload": {"test": True}, "expires_at": None}
1305 for k in _ALL_KINDS
1306 ]
1307 _write_remote_records(repo, records)
1308 remote = repo / ".muse" / "coordination" / "remote"
1309 for fpath in remote.rglob("*.json"):
1310 content = fpath.read_text(encoding="utf-8")
1311 json.loads(content) # Must not raise
1312
1313 def test_written_files_end_with_newline(self, repo: pathlib.Path) -> None:
1314 uid = str(uuid.uuid4())
1315 records = [{"kind": "reservation", "record_uuid": uid, "run_id": "r",
1316 "payload": {}, "expires_at": None}]
1317 _write_remote_records(repo, records)
1318 target = repo / ".muse" / "coordination" / "remote" / "reservation" / f"{uid}.json"
1319 assert target.read_text(encoding="utf-8").endswith("\n")
1320
1321 def test_symlink_json_file_data_read_correctly(self, repo: pathlib.Path) -> None:
1322 """A symlink to a valid JSON file must be read as a normal record."""
1323 uid = str(uuid.uuid4())
1324 real_file = repo / "real_record.json"
1325 real_file.write_text(
1326 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1327 )
1328 subdir = _coord_dir(repo) / "reservations"
1329 subdir.mkdir(parents=True, exist_ok=True)
1330 link = subdir / f"{uid}.json"
1331 link.symlink_to(real_file)
1332 gathered = _gather_local_records(repo, ["reservation"])
1333 assert len(gathered) == 1
1334 assert gathered[0]["record_uuid"] == uid
1335
1336
1337 # ===========================================================================
1338 # 9. IDEMPOTENCY
1339 # ===========================================================================
1340
1341
1342 class TestCoordIntegrityIdempotency:
1343 """Same operations must always produce the same result — no ghost state."""
1344
1345 def test_gather_is_pure_same_dir_same_result(self, repo: pathlib.Path) -> None:
1346 subdir = _coord_dir(repo) / "reservations"
1347 subdir.mkdir(parents=True, exist_ok=True)
1348 for _ in range(20):
1349 uid = str(uuid.uuid4())
1350 (subdir / f"{uid}.json").write_text(
1351 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1352 )
1353 first = _gather_local_records(repo, ["reservation"])
1354 second = _gather_local_records(repo, ["reservation"])
1355 assert len(first) == len(second) == 20
1356 uuids_first = {r["record_uuid"] for r in first}
1357 uuids_second = {r["record_uuid"] for r in second}
1358 assert uuids_first == uuids_second
1359
1360 def test_push_same_500_records_twice_second_all_skipped(
1361 self, repo: pathlib.Path
1362 ) -> None:
1363 subdir = _coord_dir(repo) / "reservations"
1364 subdir.mkdir(parents=True, exist_ok=True)
1365 for _ in range(500):
1366 uid = str(uuid.uuid4())
1367 (subdir / f"{uid}.json").write_text(
1368 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1369 )
1370
1371 # First push: all inserted
1372 with patch(_PUSH_TARGET, return_value=_push_ok(500, 0)), \
1373 patch(_REQUIRE_REPO, return_value=repo), \
1374 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1375 result1 = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1376 data1 = json.loads(result1.output.strip())
1377 assert data1["inserted"] == 500
1378
1379 # Second push: all skipped (hub deduplicates)
1380 with patch(_PUSH_TARGET, return_value=_push_ok(0, 500)), \
1381 patch(_REQUIRE_REPO, return_value=repo), \
1382 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1383 result2 = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1384 data2 = json.loads(result2.output.strip())
1385 assert data2["inserted"] == 0
1386 assert data2["skipped"] == 500
1387 assert result2.exit_code == 0
1388
1389 def test_push_partial_overlap_correct_counts(self, repo: pathlib.Path) -> None:
1390 subdir = _coord_dir(repo) / "reservations"
1391 subdir.mkdir(parents=True, exist_ok=True)
1392 for _ in range(500):
1393 uid = str(uuid.uuid4())
1394 (subdir / f"{uid}.json").write_text(
1395 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1396 )
1397 # Hub says 250 were new, 250 it already had
1398 with patch(_PUSH_TARGET, return_value=_push_ok(250, 250)), \
1399 patch(_REQUIRE_REPO, return_value=repo), \
1400 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1401 result = runner.invoke(cli, _PUSH_ARGS + ["-j"])
1402 data = json.loads(result.output.strip())
1403 assert data["inserted"] == 250
1404 assert data["skipped"] == 250
1405
1406 def test_write_remote_same_uuid_second_overwrites_first(
1407 self, repo: pathlib.Path
1408 ) -> None:
1409 uid = str(uuid.uuid4())
1410 _write_remote_records(repo, [
1411 {"kind": "reservation", "record_uuid": uid, "run_id": "first",
1412 "payload": {"seq": 1}, "expires_at": None}
1413 ])
1414 _write_remote_records(repo, [
1415 {"kind": "reservation", "record_uuid": uid, "run_id": "second",
1416 "payload": {"seq": 2}, "expires_at": None}
1417 ])
1418 target = repo / ".muse" / "coordination" / "remote" / "reservation" / f"{uid}.json"
1419 loaded = json.loads(target.read_text())
1420 assert loaded["run_id"] == "second"
1421 assert loaded["payload"]["seq"] == 2
1422
1423 def test_pull_since_id_at_cursor_returns_zero_records(
1424 self, repo: pathlib.Path
1425 ) -> None:
1426 """Pulling with since_id=cursor should yield 0 new records."""
1427 with patch(_PULL_TARGET, return_value=_pull_ok([], cursor=100)), \
1428 patch(_REQUIRE_REPO, return_value=repo), \
1429 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", "tok")):
1430 result = runner.invoke(cli, _PULL_ARGS + ["--since-id", "100", "-j"])
1431 data = json.loads(result.output.strip())
1432 assert data["count"] == 0
1433
1434
1435 # ===========================================================================
1436 # 10. TOKEN SAFETY
1437 # ===========================================================================
1438
1439
1440 class TestCoordIntegrityTokenSafety:
1441 """Auth token must NEVER appear in any output path under any condition."""
1442
1443 _SECRET = "ABSOLUTELY-SECRET-TOKEN-NEVER-LEAK-THIS-4xQzR7"
1444
1445 def _push_args_with_secret(self) -> list[str]:
1446 return [
1447 "coord", "sync", "push",
1448 "--hub", "http://localhost:10003",
1449 "--owner", "gabriel",
1450 "--slug", "linux",
1451 ]
1452
1453 def _pull_args_with_secret(self) -> list[str]:
1454 return [
1455 "coord", "sync", "pull",
1456 "--hub", "http://localhost:10003",
1457 "--owner", "gabriel",
1458 "--slug", "linux",
1459 "--since-id", "0",
1460 ]
1461
1462 def test_token_not_in_json_push_success(self, repo: pathlib.Path) -> None:
1463 with patch(_PUSH_TARGET, return_value=_push_ok(0)), \
1464 patch(_REQUIRE_REPO, return_value=repo), \
1465 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", self._SECRET)):
1466 result = runner.invoke(cli, self._push_args_with_secret() + ["-j"])
1467 assert self._SECRET not in result.output
1468
1469 def test_token_not_in_json_pull_success(self, repo: pathlib.Path) -> None:
1470 with patch(_PULL_TARGET, return_value=_pull_ok([], cursor=0)), \
1471 patch(_REQUIRE_REPO, return_value=repo), \
1472 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", self._SECRET)):
1473 result = runner.invoke(cli, self._pull_args_with_secret() + ["-j"])
1474 assert self._SECRET not in result.output
1475
1476 def test_token_not_in_json_push_error(self, repo: pathlib.Path) -> None:
1477 with patch(_PUSH_TARGET, side_effect=CoordBusError("fail", status_code=500)), \
1478 patch(_REQUIRE_REPO, return_value=repo), \
1479 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", self._SECRET)):
1480 subdir = _coord_dir(repo) / "reservations"
1481 subdir.mkdir(parents=True, exist_ok=True)
1482 uid = str(uuid.uuid4())
1483 (subdir / f"{uid}.json").write_text(
1484 json.dumps({"reservation_id": uid, "run_id": "r"}), encoding="utf-8"
1485 )
1486 result = runner.invoke(cli, self._push_args_with_secret() + ["-j"])
1487 # The secret must not appear in ANY line — error line or summary line.
1488 assert self._SECRET not in result.output
1489
1490 def test_token_not_in_json_pull_error(self, repo: pathlib.Path) -> None:
1491 with patch(_PULL_TARGET, side_effect=CoordBusError("fail", status_code=500)), \
1492 patch(_REQUIRE_REPO, return_value=repo), \
1493 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", self._SECRET)):
1494 result = runner.invoke(cli, self._pull_args_with_secret() + ["-j"])
1495 assert self._SECRET not in result.output
1496
1497 def test_token_not_in_text_push_success(self, repo: pathlib.Path) -> None:
1498 with patch(_PUSH_TARGET, return_value=_push_ok(0)), \
1499 patch(_REQUIRE_REPO, return_value=repo), \
1500 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", self._SECRET)):
1501 result = runner.invoke(cli, self._push_args_with_secret())
1502 assert self._SECRET not in result.output
1503
1504 def test_token_not_in_text_pull_success(self, repo: pathlib.Path) -> None:
1505 with patch(_PULL_TARGET, return_value=_pull_ok([], cursor=0)), \
1506 patch(_REQUIRE_REPO, return_value=repo), \
1507 patch(_RESOLVE_HUB, return_value=("http://localhost:10003", self._SECRET)):
1508 result = runner.invoke(cli, self._pull_args_with_secret())
1509 assert self._SECRET not in result.output
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