gabriel / muse public
release_coord.py python
446 lines 16.2 KB
Raw
sha256:649be9ce77a127cd847c4f798d60104dab7fe830cbc928762fafe3deadf38839 chore: point muse-git-backup.sh's MUSE_COPY at the fresh 'm… Sonnet 5 patch 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 "duration_ms": 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 "duration_ms": 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 import argparse
69 import json
70 import pathlib
71 import sys
72 from typing import Callable, TypedDict
73
74 from muse.core.coordination import (
75 _validate_reservation_id,
76 create_release,
77 load_all_reservations,
78 load_released_ids,
79 )
80 from muse.core.envelope import EnvelopeJson, make_envelope
81 from muse.core.errors import ExitCode
82 from muse.core.repo import require_repo
83 from muse.core.validation import sanitize_display
84 from muse.core.timing import start_timer
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 class _ReleasedEntry(TypedDict):
93 reservation_id: str
94 run_id: str
95 released_at: str
96 reason: str
97
98 class _ReleaseCoordSingleJson(EnvelopeJson, total=False):
99 """JSON output for single-reservation release."""
100
101 status: str # "released" | "already_released" | "not_found"
102 reservation_id: str
103 run_id: str
104 released_at: str | None
105 reason: str
106
107 class _ReleaseCoordBatchJson(EnvelopeJson):
108 """JSON output for ``--all-for-run`` batch release."""
109
110 status: str # "ok"
111 released: list[_ReleasedEntry]
112 skipped_already_released: int
113
114 # ── CLI registration ──────────────────────────────────────────────────────────
115
116 def register(
117 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
118 ) -> None:
119 """Register the ``release`` subcommand on *subparsers* (under ``muse coord``).
120
121 Wires all flags with their defaults, choices, and help text so that
122 ``--help`` output is accurate. Sets ``func`` to :func:`run`.
123 """
124 parser = subparsers.add_parser(
125 "release",
126 help="Mark a reservation as done and free its addresses.",
127 description=__doc__,
128 formatter_class=argparse.RawDescriptionHelpFormatter,
129 )
130 parser.add_argument(
131 "reservation_id",
132 nargs="?",
133 default=None,
134 metavar="RESERVATION_ID",
135 help="content-addressed ID of the reservation to release.",
136 )
137 parser.add_argument(
138 "--run-id",
139 required=True,
140 dest="run_id",
141 metavar="ID",
142 help=(
143 "Agent run-id performing the release (for audit trail). "
144 f"Max {_MAX_RUN_ID_LEN} characters."
145 ),
146 )
147 parser.add_argument(
148 "--reason",
149 default="completed",
150 choices=("completed", "cancelled", "superseded"),
151 help="Why the reservation is being released (default: completed).",
152 )
153 parser.add_argument(
154 "--all-for-run",
155 default=None,
156 dest="all_for_run",
157 metavar="RUN_ID",
158 help=(
159 "Release all active reservations for this run-id. Cannot be "
160 "combined with a positional RESERVATION_ID."
161 ),
162 )
163 parser.add_argument(
164 "--json", "-j",
165 action="store_true",
166 dest="json_out",
167 help="Emit machine-readable JSON.",
168 )
169 parser.set_defaults(func=run)
170
171 # ── Command implementation ────────────────────────────────────────────────────
172
173 def run(args: argparse.Namespace) -> None:
174 """Release one or all reservations for an agent run.
175
176 Execution order
177 ---------------
178 1. **Validate inputs** — ``--run-id`` length, mutual exclusion of
179 ``RESERVATION_ID`` and ``--all-for-run``, content-addressed ID format of
180 ``RESERVATION_ID`` (single mode only). Any failure exits
181 :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message to
182 *stderr* before any file I/O.
183 2. **Dispatch** — single mode calls :func:`_run_single`; batch mode calls
184 :func:`_run_batch`.
185
186 Single-reservation mode
187 -----------------------
188 ``reservation_id`` (positional) is required. If the reservation is already
189 released, a warning is printed and exit code 0 is returned (idempotent).
190 If the reservation does not exist, exit code
191 :attr:`~muse.core.errors.ExitCode.NOT_FOUND` (4) is returned.
192
193 Batch mode (``--all-for-run``)
194 --------------------------------
195 Releases every reservation whose ``run_id`` matches *all_for_run*. Already-
196 released reservations are counted and reported but do not cause a non-zero
197 exit.
198
199 Security
200 --------
201 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` bytes to bound audit
202 record size.
203 * ``reservation_id`` is validated as a sha256: content ID before any path is constructed,
204 preventing directory traversal attacks.
205 * All display strings are passed through
206 :func:`~muse.core.validation.sanitize_display`.
207
208 Agent quickstart::
209
210 muse coord release <reservation-id> --run-id agent-42 --json
211 muse coord release <reservation-id> --run-id agent-42 --reason cancelled --json
212 muse coord release --all-for-run agent-42 --run-id agent-42 --json
213
214 JSON fields (single mode)::
215
216 status str "released" | "already_released" | "not_found"
217 reservation_id str content-addressed ID of the reservation
218 run_id str agent run-id supplied via --run-id
219 released_at str ISO 8601 timestamp; null when already_released
220 reason str "completed" | "cancelled" | "superseded"
221
222 JSON fields (--all-for-run mode)::
223
224 status str "ok"
225 released list [{reservation_id, run_id, released_at, reason}]
226 skipped_already_released int count of already-released reservations
227
228 Exit codes::
229
230 0 Released (or already released — idempotent).
231 1 Bad arguments or --run-id too long.
232 4 Reservation not found (single mode only).
233 """
234 elapsed = start_timer()
235 reservation_id: str | None = args.reservation_id
236 run_id: str = args.run_id
237 reason: str = args.reason
238 all_for_run: str | None = args.all_for_run
239 as_json: bool = args.json_out
240
241 # ── Input validation (before any file I/O) ────────────────────────────────
242
243 if len(run_id) > _MAX_RUN_ID_LEN:
244 print(
245 f"❌ --run-id is too long ({len(run_id)} chars; max {_MAX_RUN_ID_LEN}).",
246 file=sys.stderr,
247 )
248 raise SystemExit(ExitCode.USER_ERROR)
249
250 if all_for_run is not None and reservation_id is not None:
251 print(
252 "❌ cannot specify both RESERVATION_ID and --all-for-run",
253 file=sys.stderr,
254 )
255 raise SystemExit(ExitCode.USER_ERROR)
256
257 if all_for_run is None and reservation_id is None:
258 print(
259 "❌ must specify either RESERVATION_ID or --all-for-run",
260 file=sys.stderr,
261 )
262 raise SystemExit(ExitCode.USER_ERROR)
263
264 # content ID validation for single mode — before require_repo() so we never
265 # hit the filesystem with an untrusted string.
266 if reservation_id is not None:
267 try:
268 _validate_reservation_id(reservation_id)
269 except ValueError as exc:
270 if as_json:
271 print(json.dumps({"error": str(exc), "status": "bad_id"}))
272 else:
273 print(f"❌ {exc}", file=sys.stderr)
274 raise SystemExit(ExitCode.USER_ERROR)
275
276 root = require_repo()
277
278 if all_for_run is not None:
279 _run_batch(root, all_for_run, run_id, reason, as_json, elapsed)
280 else:
281 assert reservation_id is not None
282 _run_single(root, reservation_id, run_id, reason, as_json, elapsed)
283
284 def _run_single(
285 root: pathlib.Path,
286 reservation_id: str,
287 run_id: str,
288 reason: str,
289 as_json: bool,
290 elapsed: Callable[[], float],
291 ) -> None:
292 """Release a single reservation by content-addressed ID.
293
294 Checks the released-IDs set first (stem-only scan, no JSON parsing), then
295 the full reservation list. Writes a release tombstone atomically via
296 :func:`~muse.core.store.write_text_atomic`.
297
298 Args:
299 root: Repository root (the directory containing ``.muse/``).
300 reservation_id: Pre-validated content-addressed ID string.
301 run_id: Agent performing the release (written to the tombstone).
302 reason: One of ``"completed"``, ``"cancelled"``, or ``"superseded"``.
303 as_json: When ``True``, emit compact JSON instead of human text.
304 elapsed: Timer callable from :func:`~muse.core.timing.start_timer`.
305
306 Exit codes:
307 0 — released (or already released — idempotent).
308 4 — reservation not found.
309 """
310 # Check if already released (fast path — no JSON parsing).
311 released_ids = load_released_ids(root)
312 if reservation_id in released_ids:
313 if as_json:
314 print(json.dumps(_ReleaseCoordSingleJson(
315 **make_envelope(elapsed),
316 status="already_released",
317 reservation_id=reservation_id,
318 run_id=run_id,
319 released_at=None,
320 reason=reason,
321 )))
322 else:
323 rid = sanitize_display(reservation_id)
324 print(f"reservation {rid} is already released — nothing to do")
325 raise SystemExit(ExitCode.SUCCESS)
326
327 # Check the reservation exists.
328 all_res = load_all_reservations(root)
329 known_ids = {r.reservation_id for r in all_res}
330 if reservation_id not in known_ids:
331 if as_json:
332 print(json.dumps(_ReleaseCoordSingleJson(
333 **make_envelope(elapsed, exit_code=int(ExitCode.NOT_FOUND)),
334 status="not_found",
335 reservation_id=reservation_id,
336 )))
337 else:
338 rid = sanitize_display(reservation_id)
339 print(f"❌ reservation {rid} not found", file=sys.stderr)
340 raise SystemExit(ExitCode.NOT_FOUND)
341
342 # Create the release tombstone.
343 try:
344 rel = create_release(root, reservation_id, run_id, reason)
345 except FileExistsError:
346 # Race: another agent released between our check and write — idempotent.
347 if as_json:
348 print(json.dumps(_ReleaseCoordSingleJson(
349 **make_envelope(elapsed),
350 status="already_released",
351 reservation_id=reservation_id,
352 run_id=run_id,
353 released_at=None,
354 reason=reason,
355 )))
356 else:
357 rid = sanitize_display(reservation_id)
358 print(f"reservation {rid} is already released (race) — nothing to do")
359 raise SystemExit(ExitCode.SUCCESS)
360
361 if as_json:
362 print(json.dumps(_ReleaseCoordSingleJson(
363 **make_envelope(elapsed),
364 status="released",
365 reservation_id=rel.reservation_id,
366 run_id=rel.run_id,
367 released_at=rel.released_at.isoformat(),
368 reason=rel.reason,
369 )))
370 else:
371 rid = sanitize_display(rel.reservation_id)
372 run_label = sanitize_display(rel.run_id)
373 print(
374 f"✅ released {rid} run={run_label}"
375 f" reason={rel.reason} at={rel.released_at.isoformat()[:19]}Z"
376 )
377
378 def _run_batch(
379 root: pathlib.Path,
380 all_for_run: str,
381 run_id: str,
382 reason: str,
383 as_json: bool,
384 elapsed: Callable[[], float],
385 ) -> None:
386 """Release all active reservations for a given run-id.
387
388 Loads the full reservation list and released-ID set exactly once, then
389 iterates in memory — O(reservations) with no per-record filesystem I/O
390 beyond the initial directory scan.
391
392 ``skipped_already_released`` counts reservations that were already
393 released before this call (checked via the released-IDs set).
394 ``FileExistsError`` from concurrent races is folded into the same counter
395 so the caller always gets a complete accounting.
396
397 Args:
398 root: Repository root (the directory containing ``.muse/``).
399 all_for_run: Only release reservations whose ``run_id`` matches this.
400 run_id: Agent performing the release (written to each tombstone).
401 reason: One of ``"completed"``, ``"cancelled"``, or ``"superseded"``.
402 as_json: When ``True``, emit compact JSON instead of human text.
403 elapsed: Timer callable from :func:`~muse.core.timing.start_timer`.
404 """
405 all_res = load_all_reservations(root)
406 released_ids = load_released_ids(root)
407
408 targets = [r for r in all_res if r.run_id == all_for_run]
409 released_entries = []
410 skipped_already = 0
411
412 for res in targets:
413 if res.reservation_id in released_ids:
414 skipped_already += 1
415 continue
416 try:
417 rel = create_release(root, res.reservation_id, run_id, reason)
418 released_entries.append({
419 "reservation_id": rel.reservation_id,
420 "run_id": rel.run_id,
421 "released_at": rel.released_at.isoformat(),
422 "reason": rel.reason,
423 })
424 except (FileExistsError, ValueError):
425 # FileExistsError: concurrent race — another agent beat us.
426 # ValueError: corrupt reservation_id that passed content-ID storage
427 # but failed re-validation (defensive, should not happen).
428 skipped_already += 1
429
430 if as_json:
431 print(json.dumps(_ReleaseCoordBatchJson(
432 **make_envelope(elapsed),
433 status="ok",
434 released=released_entries,
435 skipped_already_released=skipped_already,
436 )))
437 else:
438 n = len(released_entries)
439 run_label = sanitize_display(all_for_run)
440 print(f"✅ released {n} reservation(s) for run={run_label} reason={reason}")
441 if skipped_already:
442 print(f" {skipped_already} already released — skipped")
443 for entry in released_entries:
444 rid = sanitize_display(entry["reservation_id"])
445 print(f" {rid}")
446 print(f"\n ({elapsed():.3f}s)")
File History 1 commit
sha256:649be9ce77a127cd847c4f798d60104dab7fe830cbc928762fafe3deadf38839 chore: point muse-git-backup.sh's MUSE_COPY at the fresh 'm… Sonnet 5 patch 13 days ago