gabriel / muse public
coord_gc.py python
344 lines 12.8 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """``muse coord gc`` — collect stale coordination records.
2
3 The coordination directory (``.muse/coordination/``) accumulates records over
4 time: expired reservations, corresponding release tombstones, heartbeat files,
5 and old intents. ``muse coord gc`` removes records that are no longer needed
6 so the coordination layer stays lean and fast.
7
8 What gets collected
9 -------------------
10 * **Expired reservations** — TTL exhausted (after heartbeat extension), past
11 the grace period.
12 * **Released reservations** — release tombstone present and older than the
13 grace period.
14 * **Corresponding release tombstones** — removed with their reservation.
15 * **Corresponding heartbeat files** — removed with their reservation.
16 * **Orphaned releases and heartbeats** — the reservation file is gone but the
17 tombstone or heartbeat still exists (e.g. from a prior partial GC).
18 * **Old intents** (opt-in via ``--include-intents``) — older than
19 ``--max-intent-age SECONDS`` (default: 7 days).
20
21 What is never collected
22 -----------------------
23 * Active reservations (not released, effective expiry in the future).
24 * Records within the grace period.
25
26 Usage::
27
28 muse coord gc # dry-run to preview
29 muse coord gc --execute # actually delete files
30 muse coord gc --execute --grace-period 60 # shorter grace period
31 muse coord gc --execute --include-intents # also purge old intents
32 muse coord gc --execute --verbose # per-file detail
33 muse coord gc --format json # machine-readable output
34 muse coord gc --json # shorthand
35
36 JSON output schema::
37
38 {
39 "dry_run": bool,
40 "grace_period_seconds": int,
41 "include_intents": bool,
42 "max_intent_age_seconds": int,
43 "reservations_removed": int,
44 "reservations_removed_bytes": int,
45 "releases_removed": int,
46 "releases_removed_bytes": int,
47 "heartbeats_removed": int,
48 "heartbeats_removed_bytes": int,
49 "intents_removed": int,
50 "intents_removed_bytes": int,
51 "total_removed": int,
52 "total_removed_bytes": int,
53 "removed_ids": [str, ...],
54 "duration_ms": float
55 }
56
57 Exit codes::
58
59 0 — success (zero removed is still success)
60 1 — bad arguments (--grace-period < 0 or --max-intent-age <= 0)
61
62 Flags:
63
64 ``--execute``
65 Actually delete files. Without this flag the command runs in dry-run
66 mode and only reports what would be removed. Always preview first.
67
68 ``--grace-period SECONDS``
69 Records expired or released within the last N seconds are protected
70 (default: 300 = 5 min). Raise this on high-churn swarms.
71
72 ``--include-intents``
73 Also purge intents older than ``--max-intent-age``. Intents are
74 permanent audit records by default.
75
76 ``--max-intent-age SECONDS``
77 Age threshold for intent cleanup (default: 604 800 = 7 days).
78
79 ``--verbose`` / ``-v``
80 Print the ID of every removed reservation.
81
82 ``--json`` / ``--format json``
83 Emit result as compact JSON on stdout.
84 """
85
86 import argparse
87 import json
88 import sys
89
90 from typing import TypedDict
91
92 from muse.core.coordination import run_coord_gc
93 from muse.core.envelope import EnvelopeJson, make_envelope
94 from muse.core.errors import ExitCode
95 from muse.core.repo import require_repo
96 from muse.core.timing import start_timer
97
98 # ── TypedDicts ────────────────────────────────────────────────────────────────
99
100 class _CoordGcErrorJson(EnvelopeJson):
101 """JSON output for coord-gc error paths."""
102
103 error: str
104 status: str
105
106 class _CoordGcJson(EnvelopeJson):
107 """JSON output for ``muse coord gc --json``."""
108
109 dry_run: bool
110 grace_period_seconds: int
111 include_intents: bool
112 max_intent_age_seconds: int
113 reservations_removed: int
114 reservations_removed_bytes: int
115 releases_removed: int
116 releases_removed_bytes: int
117 heartbeats_removed: int
118 heartbeats_removed_bytes: int
119 intents_removed: int
120 intents_removed_bytes: int
121 total_removed: int
122 total_removed_bytes: int
123 removed_ids: list[str]
124
125 # ── Helpers ───────────────────────────────────────────────────────────────────
126
127 def _fmt_bytes(n: int) -> str:
128 """Return a human-readable byte count using binary units.
129
130 Scales through B → KiB → MiB → GiB → TiB, stopping at the first
131 unit where the value is below 1024.
132
133 Examples
134 --------
135 ``0`` → ``"0 B"``
136 ``1023`` → ``"1023 B"``
137 ``1024`` → ``"1.0 KiB"``
138 ``1_048_576`` → ``"1.0 MiB"``
139 ``1_073_741_824`` → ``"1.0 GiB"``
140 ``1_099_511_627_776`` → ``"1.0 TiB"``
141 """
142 if n < 1024:
143 return f"{n} B"
144 for unit in ("KiB", "MiB", "GiB", "TiB"):
145 n /= 1024.0
146 if n < 1024:
147 return f"{n:.1f} {unit}"
148 return f"{n:.1f} TiB"
149
150 # ── CLI registration ──────────────────────────────────────────────────────────
151
152 def register(
153 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
154 ) -> None:
155 """Register the ``gc`` subcommand on *subparsers* (under ``muse coord``).
156
157 Wires all flags with their defaults, choices, and help text so that
158 ``--help`` output is accurate. Sets ``func`` to :func:`run`.
159 """
160 parser = subparsers.add_parser(
161 "gc",
162 help="Collect stale coordination records from .muse/coordination/.",
163 description=__doc__,
164 formatter_class=argparse.RawDescriptionHelpFormatter,
165 )
166 parser.add_argument(
167 "--execute",
168 action="store_true",
169 dest="execute",
170 help=(
171 "Actually delete files. Without this flag the command runs in "
172 "dry-run mode and only reports what would be removed."
173 ),
174 )
175 parser.add_argument(
176 "--grace-period",
177 type=int,
178 default=300,
179 dest="grace_period_seconds",
180 metavar="SECONDS",
181 help=(
182 "Records expired or released within the last N seconds are skipped "
183 "(default: 300 = 5 min). Increase this on busy swarms to avoid "
184 "racing with agents that are still reading coordination state."
185 ),
186 )
187 parser.add_argument(
188 "--include-intents",
189 action="store_true",
190 dest="include_intents",
191 help=(
192 "Also purge intents older than --max-intent-age. Intents are "
193 "permanent audit records by default — only enable if you are "
194 "confident you no longer need them for post-mortem analysis."
195 ),
196 )
197 parser.add_argument(
198 "--max-intent-age",
199 type=int,
200 default=604800,
201 dest="max_intent_age_seconds",
202 metavar="SECONDS",
203 help=(
204 "Age threshold for intent cleanup when --include-intents is set "
205 "(default: 604800 = 7 days)."
206 ),
207 )
208 parser.add_argument(
209 "--verbose", "-v",
210 action="store_true",
211 help="Print the ID of every removed reservation.",
212 )
213 parser.add_argument(
214 "--json", "-j",
215 action="store_true",
216 dest="json_out",
217 help="Emit machine-readable JSON.",
218 )
219 parser.set_defaults(func=run)
220
221 # ── Command implementation ────────────────────────────────────────────────────
222
223 def run(args: argparse.Namespace) -> None:
224 """Remove stale coordination records from ``.muse/coordination/``.
225
226 Defaults to dry-run mode — pass ``--execute`` to actually delete. Records
227 released within the grace period (default 300 s) are skipped to avoid races
228 with agents still reading coordination state. Intents are permanent audit
229 records by default; pass ``--include-intents`` to collect old ones.
230
231 Agent quickstart
232 ----------------
233 ::
234
235 muse coord-gc --json
236 muse coord-gc --execute --json
237 muse coord-gc --execute --grace-period 60 --json
238
239 JSON fields
240 -----------
241 dry_run ``true`` when run without ``--execute``.
242 removed_locks Number of lock records removed.
243 removed_reservations Number of reservation records removed.
244 removed_heartbeats Number of heartbeat records removed.
245 removed_intents Number of intent records removed (requires ``--include-intents``).
246 total_removed Total records removed.
247
248 Exit codes
249 ----------
250 0 Success (zero removed is also 0).
251 1 Bad arguments.
252 """
253 elapsed = start_timer()
254 execute: bool = args.execute
255 grace_period_seconds: int = args.grace_period_seconds
256 include_intents: bool = args.include_intents
257 max_intent_age_seconds: int = args.max_intent_age_seconds
258 verbose: bool = args.verbose
259 json_out: bool = args.json_out
260
261 # ── Input validation (before any file I/O) ────────────────────────────────
262
263 if grace_period_seconds < 0:
264 msg = f"--grace-period must be >= 0, got {grace_period_seconds}"
265 if json_out:
266 print(json.dumps(_CoordGcErrorJson(**make_envelope(elapsed, exit_code=ExitCode.USER_ERROR), error=msg, status="bad_args")))
267 else:
268 print(f"❌ {msg}", file=sys.stderr)
269 raise SystemExit(ExitCode.USER_ERROR)
270
271 if max_intent_age_seconds <= 0:
272 msg = f"--max-intent-age must be > 0, got {max_intent_age_seconds}"
273 if json_out:
274 print(json.dumps(_CoordGcErrorJson(**make_envelope(elapsed, exit_code=ExitCode.USER_ERROR), error=msg, status="bad_args")))
275 else:
276 print(f"❌ {msg}", file=sys.stderr)
277 raise SystemExit(ExitCode.USER_ERROR)
278
279 root = require_repo()
280
281 result = run_coord_gc(
282 root,
283 dry_run=not execute,
284 grace_period_seconds=grace_period_seconds,
285 include_intents=include_intents,
286 max_intent_age_seconds=max_intent_age_seconds,
287 )
288
289 if json_out:
290 print(json.dumps(_CoordGcJson(
291 **make_envelope(elapsed),
292 dry_run=result.dry_run,
293 grace_period_seconds=result.grace_period_seconds,
294 include_intents=result.include_intents,
295 max_intent_age_seconds=max_intent_age_seconds,
296 reservations_removed=result.reservations_removed,
297 reservations_removed_bytes=result.reservations_removed_bytes,
298 releases_removed=result.releases_removed,
299 releases_removed_bytes=result.releases_removed_bytes,
300 heartbeats_removed=result.heartbeats_removed,
301 heartbeats_removed_bytes=result.heartbeats_removed_bytes,
302 intents_removed=result.intents_removed,
303 intents_removed_bytes=result.intents_removed_bytes,
304 total_removed=result.total_removed,
305 total_removed_bytes=result.total_removed_bytes,
306 removed_ids=result.removed_ids,
307 )))
308 return
309
310 # ── Text output ───────────────────────────────────────────────────────────
311 mode = "DRY RUN — no files deleted" if result.dry_run else "GC complete"
312 print(f"\nCoordination GC — {mode}")
313 print("─" * 62)
314
315 if result.total_removed == 0:
316 print("\n Nothing to collect.")
317 else:
318 lines = [
319 ("Reservations", result.reservations_removed, result.reservations_removed_bytes),
320 ("Releases", result.releases_removed, result.releases_removed_bytes),
321 ("Heartbeats", result.heartbeats_removed, result.heartbeats_removed_bytes),
322 ]
323 if include_intents:
324 lines.append(("Intents", result.intents_removed, result.intents_removed_bytes))
325
326 for label, count, nbytes in lines:
327 if count:
328 verb = "would remove" if result.dry_run else "removed"
329 print(f" {label:<14} {verb} {count:>4} ({_fmt_bytes(nbytes)})")
330
331 print(
332 f"\n Total: {result.total_removed} record(s)"
333 f" ({_fmt_bytes(result.total_removed_bytes)})"
334 )
335
336 if verbose and result.removed_ids:
337 print("\n Removed reservation IDs:")
338 for rid in result.removed_ids:
339 print(f" {rid}")
340
341 print(
342 f"\n grace-period={grace_period_seconds}s"
343 f" ({result.duration_ms:.3f}s)"
344 )
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago