aaronrene / gabriel-muse public
release_coord.py python
428 lines 15.1 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """``muse coord release`` — mark a reservation as done and free its addresses.
2
3 Writing a release tombstone tells the swarm that the agent has finished
4 (or abandoned) its work on the reserved addresses. Once released, the
5 reservation is excluded from :func:`~muse.core.coordination.active_reservations`
6 immediately — no need to wait for the TTL to expire. This is the cooperative
7 handshake that prevents the coordination layer from becoming stale.
8
9 Usage::
10
11 muse coord release <reservation-id> --run-id AGENT-42
12 muse coord release <reservation-id> --run-id AGENT-42 --reason cancelled
13 muse coord release <reservation-id> --run-id AGENT-42 --reason superseded
14 muse coord release --all-for-run AGENT-42 --run-id AGENT-42
15 muse coord release <reservation-id> --run-id AGENT-42 --json
16
17 Reasons::
18
19 completed Work finished successfully (default).
20 cancelled Work was abandoned before completion.
21 superseded A newer reservation replaced this one.
22
23 JSON output schema (single-release mode)::
24
25 {
26 "status": "released" | "already_released" | "not_found",
27 "reservation_id": str,
28 "run_id": str,
29 "released_at": str, // ISO 8601; null when already_released
30 "reason": str,
31 "elapsed_seconds": float
32 }
33
34 JSON output schema (``--all-for-run`` mode)::
35
36 {
37 "status": "ok",
38 "released": [{"reservation_id": str, "run_id": str,
39 "released_at": str, "reason": str}, ...],
40 "skipped_already_released": int,
41 "elapsed_seconds": float
42 }
43
44 Exit codes::
45
46 0 — released (or already released — idempotent)
47 1 — bad arguments (mutual exclusion, missing required flag, run-id too long)
48 4 — reservation not found
49
50 Flags:
51
52 ``--run-id ID``
53 Agent/pipeline identifier for the audit trail (required).
54 Maximum length: 256 characters.
55
56 ``--reason REASON``
57 Why the reservation is being released. One of ``completed`` (default),
58 ``cancelled``, or ``superseded``.
59
60 ``--all-for-run RUN_ID``
61 Release every active reservation whose ``run_id`` matches *RUN_ID*.
62 Cannot be combined with a positional ``RESERVATION_ID``.
63
64 ``--json`` / ``--format json``
65 Emit result as compact JSON on stdout.
66 """
67
68 from __future__ import annotations
69
70 import argparse
71 import json
72 import pathlib
73 import sys
74 import time
75
76 from muse.core.coordination import (
77 _validate_reservation_id,
78 create_release,
79 load_all_reservations,
80 load_released_ids,
81 )
82 from muse.core.errors import ExitCode
83 from muse.core.repo import require_repo
84 from muse.core.validation import sanitize_display
85
86 # ── Input constraints ─────────────────────────────────────────────────────────
87
88 #: Maximum byte-length of the ``--run-id`` value. Mirrors the limit in
89 #: ``reserve.py`` so audit records stay bounded in size.
90 _MAX_RUN_ID_LEN: int = 256
91
92
93 # ── CLI registration ──────────────────────────────────────────────────────────
94
95
96 def register(
97 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
98 ) -> None:
99 """Register the ``release`` subcommand on *subparsers* (under ``muse coord``).
100
101 Wires all flags with their defaults, choices, and help text so that
102 ``--help`` output is accurate. Sets ``func`` to :func:`run`.
103 """
104 parser = subparsers.add_parser(
105 "release",
106 help="Mark a reservation as done and free its addresses.",
107 description=__doc__,
108 formatter_class=argparse.RawDescriptionHelpFormatter,
109 )
110 parser.add_argument(
111 "reservation_id",
112 nargs="?",
113 default=None,
114 metavar="RESERVATION_ID",
115 help="UUID of the reservation to release.",
116 )
117 parser.add_argument(
118 "--run-id",
119 required=True,
120 dest="run_id",
121 metavar="ID",
122 help=(
123 "Agent run-id performing the release (for audit trail). "
124 f"Max {_MAX_RUN_ID_LEN} characters."
125 ),
126 )
127 parser.add_argument(
128 "--reason",
129 default="completed",
130 choices=("completed", "cancelled", "superseded"),
131 help="Why the reservation is being released (default: completed).",
132 )
133 parser.add_argument(
134 "--all-for-run",
135 default=None,
136 dest="all_for_run",
137 metavar="RUN_ID",
138 help=(
139 "Release all active reservations for this run-id. Cannot be "
140 "combined with a positional RESERVATION_ID."
141 ),
142 )
143 parser.add_argument(
144 "--format", "-f",
145 default="text",
146 dest="fmt",
147 choices=("text", "json"),
148 help="Output format: text (default) or json.",
149 )
150 parser.add_argument(
151 "--json",
152 action="store_const",
153 const="json",
154 dest="fmt",
155 help="Shorthand for --format json.",
156 )
157 parser.set_defaults(func=run)
158
159
160 # ── Command implementation ────────────────────────────────────────────────────
161
162
163 def run(args: argparse.Namespace) -> None:
164 """Release one or all reservations for an agent run.
165
166 Execution order
167 ---------------
168 1. **Validate inputs** — ``--run-id`` length, mutual exclusion of
169 ``RESERVATION_ID`` and ``--all-for-run``, UUID syntax of
170 ``RESERVATION_ID`` (single mode only). Any failure exits
171 :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message to
172 *stderr* before any file I/O.
173 2. **Dispatch** — single mode calls :func:`_run_single`; batch mode calls
174 :func:`_run_batch`.
175
176 Single-reservation mode
177 -----------------------
178 ``reservation_id`` (positional) is required. If the reservation is already
179 released, a warning is printed and exit code 0 is returned (idempotent).
180 If the reservation does not exist, exit code
181 :attr:`~muse.core.errors.ExitCode.NOT_FOUND` (4) is returned.
182
183 Batch mode (``--all-for-run``)
184 --------------------------------
185 Releases every reservation whose ``run_id`` matches *all_for_run*. Already-
186 released reservations are counted and reported but do not cause a non-zero
187 exit.
188
189 Security
190 --------
191 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` bytes to bound audit
192 record size.
193 * ``reservation_id`` is validated as a UUID before any path is constructed,
194 preventing directory traversal attacks.
195 * All display strings are passed through
196 :func:`~muse.core.validation.sanitize_display`.
197
198 Args:
199 args: Parsed ``argparse.Namespace`` with attributes ``reservation_id``,
200 ``run_id``, ``reason``, ``all_for_run``, and ``fmt``.
201
202 Exit codes:
203 0 — released successfully (or already released — idempotent).
204 1 — bad arguments or ``--run-id`` too long.
205 4 — reservation not found (single mode only).
206 """
207 t0 = time.monotonic()
208 reservation_id: str | None = args.reservation_id
209 run_id: str = args.run_id
210 reason: str = args.reason
211 all_for_run: str | None = args.all_for_run
212 fmt: str = args.fmt
213 as_json = fmt == "json"
214
215 # ── Input validation (before any file I/O) ────────────────────────────────
216
217 if len(run_id) > _MAX_RUN_ID_LEN:
218 print(
219 f"❌ --run-id is too long ({len(run_id)} chars; max {_MAX_RUN_ID_LEN}).",
220 file=sys.stderr,
221 )
222 raise SystemExit(ExitCode.USER_ERROR)
223
224 if all_for_run is not None and reservation_id is not None:
225 print(
226 "❌ cannot specify both RESERVATION_ID and --all-for-run",
227 file=sys.stderr,
228 )
229 raise SystemExit(ExitCode.USER_ERROR)
230
231 if all_for_run is None and reservation_id is None:
232 print(
233 "❌ must specify either RESERVATION_ID or --all-for-run",
234 file=sys.stderr,
235 )
236 raise SystemExit(ExitCode.USER_ERROR)
237
238 # UUID validation for single mode — before require_repo() so we never
239 # hit the filesystem with an untrusted string.
240 if reservation_id is not None:
241 try:
242 _validate_reservation_id(reservation_id)
243 except ValueError as exc:
244 if as_json:
245 print(json.dumps({"error": str(exc), "status": "bad_id"}))
246 else:
247 print(f"❌ {exc}", file=sys.stderr)
248 raise SystemExit(ExitCode.USER_ERROR)
249
250 root = require_repo()
251
252 if all_for_run is not None:
253 _run_batch(root, all_for_run, run_id, reason, as_json, t0)
254 else:
255 assert reservation_id is not None
256 _run_single(root, reservation_id, run_id, reason, as_json, t0)
257
258
259 def _run_single(
260 root: pathlib.Path,
261 reservation_id: str,
262 run_id: str,
263 reason: str,
264 as_json: bool,
265 t0: float,
266 ) -> None:
267 """Release a single reservation by UUID.
268
269 Checks the released-IDs set first (stem-only scan, no JSON parsing), then
270 the full reservation list. Writes a release tombstone atomically via
271 :func:`~muse.core.store.write_text_atomic`.
272
273 Args:
274 root: Repository root (the directory containing ``.muse/``).
275 reservation_id: Pre-validated UUID string.
276 run_id: Agent performing the release (written to the tombstone).
277 reason: One of ``"completed"``, ``"cancelled"``, or ``"superseded"``.
278 as_json: When ``True``, emit compact JSON instead of human text.
279 t0: :func:`time.monotonic` start timestamp used for ``elapsed_seconds``.
280
281 Exit codes:
282 0 — released (or already released — idempotent).
283 4 — reservation not found.
284 """
285 # Check if already released (fast path — no JSON parsing).
286 released_ids = load_released_ids(root)
287 if reservation_id in released_ids:
288 elapsed = round(time.monotonic() - t0, 4)
289 if as_json:
290 print(json.dumps({
291 "status": "already_released",
292 "reservation_id": reservation_id,
293 "run_id": run_id,
294 "released_at": None,
295 "reason": reason,
296 "elapsed_seconds": elapsed,
297 }))
298 else:
299 rid = sanitize_display(reservation_id)
300 print(f"reservation {rid} is already released — nothing to do")
301 raise SystemExit(ExitCode.SUCCESS)
302
303 # Check the reservation exists.
304 all_res = load_all_reservations(root)
305 known_ids = {r.reservation_id for r in all_res}
306 if reservation_id not in known_ids:
307 elapsed = round(time.monotonic() - t0, 4)
308 if as_json:
309 print(json.dumps({
310 "status": "not_found",
311 "reservation_id": reservation_id,
312 "elapsed_seconds": elapsed,
313 }))
314 else:
315 rid = sanitize_display(reservation_id)
316 print(f"❌ reservation {rid} not found", file=sys.stderr)
317 raise SystemExit(ExitCode.NOT_FOUND)
318
319 # Create the release tombstone.
320 try:
321 rel = create_release(root, reservation_id, run_id, reason)
322 except FileExistsError:
323 # Race: another agent released between our check and write — idempotent.
324 elapsed = round(time.monotonic() - t0, 4)
325 if as_json:
326 print(json.dumps({
327 "status": "already_released",
328 "reservation_id": reservation_id,
329 "run_id": run_id,
330 "released_at": None,
331 "reason": reason,
332 "elapsed_seconds": elapsed,
333 }))
334 else:
335 rid = sanitize_display(reservation_id)
336 print(f"reservation {rid} is already released (race) — nothing to do")
337 raise SystemExit(ExitCode.SUCCESS)
338
339 elapsed = round(time.monotonic() - t0, 4)
340 if as_json:
341 print(json.dumps({
342 "status": "released",
343 "reservation_id": rel.reservation_id,
344 "run_id": rel.run_id,
345 "released_at": rel.released_at.isoformat(),
346 "reason": rel.reason,
347 "elapsed_seconds": elapsed,
348 }))
349 else:
350 rid = sanitize_display(rel.reservation_id)
351 run_label = sanitize_display(rel.run_id)
352 print(
353 f"✅ released {rid} run={run_label}"
354 f" reason={rel.reason} at={rel.released_at.isoformat()[:19]}Z"
355 )
356
357
358 def _run_batch(
359 root: pathlib.Path,
360 all_for_run: str,
361 run_id: str,
362 reason: str,
363 as_json: bool,
364 t0: float,
365 ) -> None:
366 """Release all active reservations for a given run-id.
367
368 Loads the full reservation list and released-ID set exactly once, then
369 iterates in memory — O(reservations) with no per-record filesystem I/O
370 beyond the initial directory scan.
371
372 ``skipped_already_released`` counts reservations that were already
373 released before this call (checked via the released-IDs set).
374 ``FileExistsError`` from concurrent races is folded into the same counter
375 so the caller always gets a complete accounting.
376
377 Args:
378 root: Repository root (the directory containing ``.muse/``).
379 all_for_run: Only release reservations whose ``run_id`` matches this.
380 run_id: Agent performing the release (written to each tombstone).
381 reason: One of ``"completed"``, ``"cancelled"``, or ``"superseded"``.
382 as_json: When ``True``, emit compact JSON instead of human text.
383 t0: :func:`time.monotonic` start timestamp used for ``elapsed_seconds``.
384 """
385 all_res = load_all_reservations(root)
386 released_ids = load_released_ids(root)
387
388 targets = [r for r in all_res if r.run_id == all_for_run]
389 released_entries = []
390 skipped_already = 0
391
392 for res in targets:
393 if res.reservation_id in released_ids:
394 skipped_already += 1
395 continue
396 try:
397 rel = create_release(root, res.reservation_id, run_id, reason)
398 released_entries.append({
399 "reservation_id": rel.reservation_id,
400 "run_id": rel.run_id,
401 "released_at": rel.released_at.isoformat(),
402 "reason": rel.reason,
403 })
404 except (FileExistsError, ValueError):
405 # FileExistsError: concurrent race — another agent beat us.
406 # ValueError: corrupt reservation_id that passed UUID storage
407 # but failed re-validation (defensive, should not happen).
408 skipped_already += 1
409
410 elapsed = round(time.monotonic() - t0, 4)
411
412 if as_json:
413 print(json.dumps({
414 "status": "ok",
415 "released": released_entries,
416 "skipped_already_released": skipped_already,
417 "elapsed_seconds": elapsed,
418 }))
419 else:
420 n = len(released_entries)
421 run_label = sanitize_display(all_for_run)
422 print(f"✅ released {n} reservation(s) for run={run_label} reason={reason}")
423 if skipped_already:
424 print(f" {skipped_already} already released — skipped")
425 for entry in released_entries:
426 rid = sanitize_display(entry["reservation_id"])
427 print(f" {rid}")
428 print(f"\n ({elapsed:.3f}s)")
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago