gabriel / muse public
task_queue.py python
1,559 lines 55.9 KB
Raw
sha256:1ddad36d76d3a8d323f9b3664169cb184b7a38b39208214a2ae504154260826f fix: show full cryptographic IDs in all human-readable CLI output Sonnet 4.6 patch 34 days ago
1 """``muse coord`` task-queue subcommands — real work distribution.
2
3 Provides five subcommands for operating the file-system–based task queue:
4
5 ``muse coord enqueue``
6 Add a task to the queue.
7
8 ``muse coord claim``
9 Atomically claim the highest-priority pending task. Exactly one agent
10 wins when multiple agents call ``claim`` concurrently.
11
12 ``muse coord complete``
13 Mark a claimed task as successfully completed.
14
15 ``muse coord fail-task``
16 Mark a claimed task as failed (agent could not complete the work).
17
18 ``muse coord tasks``
19 List tasks with optional status/queue/run-id filtering.
20
21 Typical multi-agent workflow::
22
23 # Orchestrator enqueues work:
24 muse coord enqueue "Refactor billing module" \\
25 --priority 5 --payload '{"addresses":["billing.py::*"]}' \\
26 --run-id orchestrator --tags billing
27
28 # N agents compete for tasks (exactly one wins per call):
29 TASK=$(muse coord claim --run-id $AGENT_ID --json)
30 TASK_ID=$(echo "$TASK" | python3 -c "import sys,json; print(json.load(sys.stdin)['task_id'])")
31
32 # Agent works on the task …
33
34 # Agent reports result:
35 muse coord complete $TASK_ID --run-id $AGENT_ID --result '{"pr": 42}'
36 # or on failure:
37 muse coord fail-task $TASK_ID --run-id $AGENT_ID --error "timed out"
38
39 Exit codes (all subcommands)::
40
41 0 — success
42 1 — bad arguments, task not found, queue empty, permission error, or unexpected error
43
44 JSON output
45 -----------
46 All subcommands accept ``--json`` / ``--format json``. The JSON schema for
47 each is documented in the individual function docstrings below.
48 """
49
50 from __future__ import annotations
51
52 import argparse
53 import json
54 import sys
55 import time
56
57 from muse.core.errors import ExitCode
58 from muse.core.repo import require_repo
59 from muse.core.task_queue import (
60 ClaimRecord,
61 TaskRecord,
62 _MAX_QUEUE_LEN,
63 _MAX_TAG_LEN,
64 _MAX_TAGS,
65 _MAX_TITLE_LEN,
66 _validate_queue_name,
67 cancel_task,
68 claim_next_task,
69 complete_task,
70 create_task,
71 fail_task,
72 get_task_status,
73 heartbeat_claim,
74 load_all_claims,
75 load_all_tasks,
76 load_claim,
77 load_task,
78 )
79 from muse.core.validation import sanitize_display
80
81
82 type _IntMap = dict[str, int]
83 # ── Module-level constants ────────────────────────────────────────────────────
84
85 #: Maximum byte-length of ``--run-id``. Mirrors all other coordination commands.
86 _MAX_RUN_ID_LEN: int = 256
87
88 #: Maximum byte-length of the serialised ``--payload`` JSON string.
89 #: Prevents unbounded memory use when the payload is later loaded into RAM.
90 _MAX_PAYLOAD_BYTES: int = 65536 # 64 KiB
91
92 #: Minimum claim TTL in seconds. A TTL of 0 creates an immediately-expired
93 #: claim which would be immediately re-claimable by any other agent.
94 _MIN_CLAIM_TTL: int = 1
95
96 #: Maximum claim TTL in seconds (24 h). Prevents agents from holding tasks
97 #: indefinitely with no heartbeat path.
98 _MAX_CLAIM_TTL: int = 86400
99
100 #: Maximum ``--wait`` polling duration in seconds (1 h).
101 _MAX_WAIT_SECONDS: int = 3600
102
103 #: Maximum byte-length of the serialised ``--result`` JSON string.
104 #: Mirrors :data:`_MAX_PAYLOAD_BYTES`; keeps claim files bounded in size.
105 _MAX_RESULT_BYTES: int = 65536 # 64 KiB
106
107 #: Maximum character-length of the ``--error`` message on fail-task.
108 #: Bounds claim file sizes and prevents pathological memory use when loading
109 #: large sets of failed claims for analysis.
110 _MAX_ERROR_LEN: int = 4096
111
112 #: Maximum value of ``--limit`` on ``tasks``. Prevents accidentally dumping
113 #: the entire queue in one shot on very large repos.
114 _MAX_LIMIT: int = 10000
115
116 # ── Shared argument helpers ───────────────────────────────────────────────────
117
118
119 def _add_format_args(parser: argparse.ArgumentParser) -> None:
120 """Add ``--format``/``-f`` and ``--json`` to *parser*."""
121 parser.add_argument(
122 "--format", "-f",
123 default="text",
124 dest="fmt",
125 choices=("text", "json"),
126 help="Output format: text (default) or json.",
127 )
128 parser.add_argument(
129 "--json",
130 action="store_const",
131 const="json",
132 dest="fmt",
133 help="Shorthand for --format json.",
134 )
135
136
137 def _err(msg: str, as_json: bool = False, status: str = "error") -> None:
138 """Print an error message.
139
140 In JSON mode, emits ``{"error": msg, "status": status}`` to *stdout*
141 (so the machine-readable stream is never broken by a bare text line).
142 In text mode, prefixes with ``❌`` and writes to *stderr*.
143 """
144 if as_json:
145 print(json.dumps({"error": msg, "status": status}))
146 else:
147 print(f"❌ {msg}", file=sys.stderr)
148
149
150 # ── muse coord enqueue ────────────────────────────────────────────────────────
151
152
153 def register_enqueue(
154 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
155 ) -> None:
156 """Register ``enqueue`` on *subparsers* (under ``muse coord``).
157
158 Wires all flags with their defaults, choices, and help text so that
159 ``--help`` output is accurate. Sets ``func`` to :func:`run_enqueue`.
160
161 Flags registered
162 ----------------
163 ``title``
164 Positional. Short human-readable description (≤ ``_MAX_TITLE_LEN``
165 chars; trimmed by the core layer, not rejected).
166 ``--priority N``
167 Integer priority; higher = claimed first. Default: 0.
168 ``--queue QUEUE``
169 Target logical queue. Must match ``[a-zA-Z0-9_-]+``. Default:
170 ``"default"``.
171 ``--ttl SECONDS``
172 Pending TTL in seconds. Must be ≥ 1. Default: 86400 (24 h).
173 ``--run-id RUNID``
174 Enqueuing agent identifier. Maximum :data:`_MAX_RUN_ID_LEN` chars.
175 ``--payload JSON``
176 Arbitrary JSON object (must be ``{}``-style, not an array).
177 Maximum :data:`_MAX_PAYLOAD_BYTES` bytes when UTF-8 encoded.
178 ``--tags TAG1,TAG2``
179 Comma-separated tag list. Maximum :data:`~muse.core.task_queue._MAX_TAGS`
180 tags; each tag is trimmed to :data:`~muse.core.task_queue._MAX_TAG_LEN`
181 chars by the core layer.
182 ``--format`` / ``--json``
183 Machine-readable compact JSON output.
184
185 JSON output schema::
186
187 {
188 "schema_version": str,
189 "task_id": str, // UUID
190 "title": str,
191 "priority": int,
192 "queue": str,
193 "ttl_seconds": int,
194 "created_by": str,
195 "created_at": str, // ISO 8601 UTC
196 "tags": [str, ...],
197 "payload": dict,
198 "elapsed_seconds": float
199 }
200
201 Exit codes::
202
203 0 — task enqueued
204 1 — bad arguments
205 """
206 parser = subparsers.add_parser(
207 "enqueue",
208 help="Add a task to the coordination task queue.",
209 formatter_class=argparse.RawDescriptionHelpFormatter,
210 description="""Add a work item to the task queue for agents to claim and execute.
211
212 Tasks are ordered by priority (higher = first) then creation time (FIFO within
213 same priority). The ``payload`` field carries arbitrary JSON that the claiming
214 agent receives along with the task definition.
215 """,
216 )
217 parser.add_argument(
218 "title",
219 help=(
220 f"Short human-readable task description (≤ {_MAX_TITLE_LEN} chars)."
221 ),
222 )
223 parser.add_argument(
224 "--priority",
225 type=int,
226 default=0,
227 metavar="N",
228 help="Task priority (higher = processed first, default 0).",
229 )
230 parser.add_argument(
231 "--queue",
232 default="default",
233 metavar="QUEUE",
234 help=(
235 f"Target queue name (default: 'default'). "
236 f"Max {_MAX_QUEUE_LEN} chars, pattern [a-zA-Z0-9_-]."
237 ),
238 )
239 parser.add_argument(
240 "--ttl",
241 type=int,
242 default=86400,
243 dest="ttl_seconds",
244 metavar="SECONDS",
245 help="Seconds the task may remain pending before expiry (default 86400, min 1).",
246 )
247 parser.add_argument(
248 "--run-id",
249 default="unknown",
250 dest="run_id",
251 metavar="RUNID",
252 help=(
253 f"Enqueuing agent / orchestrator identifier. "
254 f"Maximum {_MAX_RUN_ID_LEN} characters."
255 ),
256 )
257 parser.add_argument(
258 "--payload",
259 default="{}",
260 metavar="JSON",
261 help=(
262 f"Arbitrary JSON object attached to the task (default: {{}}). "
263 f"Must be a JSON object (not array). "
264 f"Maximum {_MAX_PAYLOAD_BYTES} bytes when UTF-8 encoded."
265 ),
266 )
267 parser.add_argument(
268 "--tags",
269 default="",
270 metavar="TAG1,TAG2",
271 help=(
272 f"Comma-separated list of tags. "
273 f"Maximum {_MAX_TAGS} tags, each ≤ {_MAX_TAG_LEN} chars."
274 ),
275 )
276 _add_format_args(parser)
277 parser.set_defaults(func=run_enqueue)
278
279
280 def run_enqueue(args: argparse.Namespace) -> None:
281 """Create and persist a new task in the coordination queue.
282
283 Execution order
284 ---------------
285 1. **Validate inputs** — all validation fires before any file I/O:
286
287 * ``title`` must be non-empty.
288 * ``--run-id`` must be ≤ :data:`_MAX_RUN_ID_LEN` characters.
289 * ``--ttl`` must be ≥ 1.
290 * ``--payload`` must be valid JSON, a JSON object (not array), and
291 ≤ :data:`_MAX_PAYLOAD_BYTES` bytes when UTF-8 encoded.
292 * ``--queue`` must match ``[a-zA-Z0-9_-]+`` and be ≤
293 :data:`~muse.core.task_queue._MAX_QUEUE_LEN` characters.
294 * ``--tags`` list must contain ≤ :data:`~muse.core.task_queue._MAX_TAGS`
295 tags after splitting on commas.
296
297 Any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1)
298 with a message on *stderr* (or compact JSON when ``--format json``).
299
300 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`.
301 3. **Create task** — calls :func:`~muse.core.task_queue.create_task`,
302 which atomically writes ``.muse/coordination/tasks/<uuid>.json``.
303 4. **Emit output** — compact JSON to *stdout* when ``--format json``,
304 or human-readable text otherwise.
305
306 Security
307 --------
308 * ``--payload`` is parsed as JSON (not ``eval``'d), so shell injection
309 via JSON content is not possible.
310 * ``--payload`` is size-capped at :data:`_MAX_PAYLOAD_BYTES` to prevent
311 unbounded memory use when loading large queues.
312 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound audit record
313 size.
314 * Queue name is validated against ``[a-zA-Z0-9_-]+`` before any file I/O.
315 * All display strings in text output are passed through
316 :func:`~muse.core.validation.sanitize_display`.
317
318 Performance
319 -----------
320 All input validation fires before ``require_repo()`` so bad-argument errors
321 never touch the filesystem.
322
323 Args:
324 args: Parsed ``argparse.Namespace`` with attributes ``title``,
325 ``priority``, ``queue``, ``ttl_seconds``, ``run_id``,
326 ``payload``, ``tags``, and ``fmt``.
327
328 Exit codes:
329 0 — task enqueued successfully.
330 1 — bad arguments.
331 """
332 t0 = time.monotonic()
333 as_json = args.fmt == "json"
334
335 # ── Input validation (before any file I/O) ────────────────────────────────
336
337 if not args.title or not args.title.strip():
338 msg = "title must be non-empty"
339 _err(msg, as_json, "bad_args")
340 raise SystemExit(ExitCode.USER_ERROR)
341
342 if len(args.run_id) > _MAX_RUN_ID_LEN:
343 msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})"
344 _err(msg, as_json, "bad_args")
345 raise SystemExit(ExitCode.USER_ERROR)
346
347 if args.ttl_seconds < 1:
348 msg = f"--ttl must be ≥ 1, got {args.ttl_seconds}"
349 _err(msg, as_json, "bad_args")
350 raise SystemExit(ExitCode.USER_ERROR)
351
352 payload_bytes = args.payload.encode()
353 if len(payload_bytes) > _MAX_PAYLOAD_BYTES:
354 msg = (
355 f"--payload is too large ({len(payload_bytes)} bytes; "
356 f"max {_MAX_PAYLOAD_BYTES})"
357 )
358 _err(msg, as_json, "bad_args")
359 raise SystemExit(ExitCode.USER_ERROR)
360
361 try:
362 payload = json.loads(args.payload)
363 if not isinstance(payload, dict):
364 raise ValueError("payload must be a JSON object")
365 except (json.JSONDecodeError, ValueError) as exc:
366 _err(f"invalid --payload: {exc}", as_json, "bad_payload")
367 raise SystemExit(ExitCode.USER_ERROR)
368
369 tags = [t.strip() for t in args.tags.split(",") if t.strip()] if args.tags else []
370 if len(tags) > _MAX_TAGS:
371 msg = f"too many tags: {len(tags)} (max {_MAX_TAGS})"
372 _err(msg, as_json, "bad_args")
373 raise SystemExit(ExitCode.USER_ERROR)
374
375 try:
376 _validate_queue_name(args.queue)
377 except ValueError as exc:
378 _err(str(exc), as_json, "bad_queue")
379 raise SystemExit(ExitCode.USER_ERROR)
380
381 root = require_repo()
382
383 try:
384 task = create_task(
385 root,
386 args.title,
387 payload=payload,
388 priority=args.priority,
389 queue=args.queue,
390 ttl_seconds=args.ttl_seconds,
391 created_by=args.run_id,
392 tags=tags,
393 )
394 except ValueError as exc:
395 _err(str(exc), as_json, "bad_args")
396 raise SystemExit(ExitCode.USER_ERROR)
397
398 elapsed = round(time.monotonic() - t0, 4)
399
400 if as_json:
401 out = task.to_dict()
402 out["elapsed_seconds"] = elapsed
403 print(json.dumps(out))
404 return
405
406 print(f"\n✅ Task enqueued")
407 print(f" Task ID: {sanitize_display(task.task_id)}")
408 print(f" Title: {sanitize_display(task.title)}")
409 print(f" Queue: {sanitize_display(task.queue)} priority={task.priority}")
410 print(f" TTL: {task.ttl_seconds}s")
411 if task.tags:
412 print(f" Tags: {', '.join(sanitize_display(t) for t in task.tags)}")
413 print(f"\n ({elapsed:.3f}s)")
414
415
416 # ── muse coord claim ──────────────────────────────────────────────────────────
417
418
419 def register_claim(
420 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
421 ) -> None:
422 """Register ``claim`` on *subparsers* (under ``muse coord``).
423
424 Wires all flags with their defaults, choices, and help text so that
425 ``--help`` output is accurate. Sets ``func`` to :func:`run_claim`.
426
427 Flags registered
428 ----------------
429 ``--run-id RUNID``
430 Required. Claiming agent identifier stored in the claim record.
431 Maximum :data:`_MAX_RUN_ID_LEN` characters.
432 ``--queue QUEUE``
433 Restrict claiming to a specific queue. When omitted, the
434 highest-priority task across all queues is claimed. Must match
435 ``[a-zA-Z0-9_-]+`` when provided.
436 ``--claim-ttl SECONDS``
437 How long the claim is valid before another agent may re-claim the
438 task. Must be in ``[_MIN_CLAIM_TTL, _MAX_CLAIM_TTL]``. Agents
439 should call ``muse coord heartbeat`` every ``claim-ttl / 2``
440 seconds to keep the claim alive. Default: 3600 (1 h).
441 ``--wait SECONDS``
442 If the queue is empty, poll until a task appears or *SECONDS*
443 elapses. ``0`` (default) returns immediately on empty. Maximum
444 :data:`_MAX_WAIT_SECONDS`. Uses exponential backoff capped at 5 s.
445 ``--format`` / ``--json``
446 Machine-readable compact JSON output.
447
448 JSON output schema (success, exit 0)::
449
450 {
451 "schema_version": str,
452 "status": "claimed",
453 "task_id": str,
454 "claimer_run_id": str,
455 "claimed_at": str, // ISO 8601 UTC
456 "expires_at": str, // ISO 8601 UTC
457 "task": { ... }, // full TaskRecord fields
458 "elapsed_seconds": float
459 }
460
461 JSON output schema (queue empty after any wait, exit 1)::
462
463 {
464 "schema_version": str,
465 "status": "empty",
466 "queue": str | null,
467 "elapsed_seconds": float
468 }
469
470 Exit codes::
471
472 0 — task claimed successfully
473 1 — queue empty (after exhausting --wait period) or bad arguments
474 """
475 parser = subparsers.add_parser(
476 "claim",
477 help="Atomically claim the highest-priority pending task.",
478 formatter_class=argparse.RawDescriptionHelpFormatter,
479 description="""Scan the task queue and atomically claim the highest-priority
480 pending task using O_CREAT|O_EXCL. When multiple agents call 'claim'
481 simultaneously, exactly one wins per task — no task is ever claimed twice.
482
483 Timed-out tasks (claim.expires_at < now) are eligible for re-claiming using
484 optimistic concurrency (atomic rename + read-back nonce verification).
485 """,
486 )
487 parser.add_argument(
488 "--run-id",
489 required=True,
490 dest="run_id",
491 metavar="RUNID",
492 help=(
493 f"Claiming agent identifier (stored in the claim record). "
494 f"Maximum {_MAX_RUN_ID_LEN} characters."
495 ),
496 )
497 parser.add_argument(
498 "--queue",
499 default=None,
500 metavar="QUEUE",
501 help=(
502 "Only claim from this queue (default: any queue). "
503 "Must match [a-zA-Z0-9_-]+ when provided."
504 ),
505 )
506 parser.add_argument(
507 "--claim-ttl",
508 type=int,
509 default=3600,
510 dest="claim_ttl",
511 metavar="SECONDS",
512 help=(
513 f"Claim TTL in seconds "
514 f"(min {_MIN_CLAIM_TTL}, max {_MAX_CLAIM_TTL}, default 3600). "
515 f"Heartbeat every claim-ttl/2 seconds to keep the claim alive."
516 ),
517 )
518 parser.add_argument(
519 "--wait",
520 type=int,
521 default=0,
522 dest="wait",
523 metavar="SECONDS",
524 help=(
525 f"If the queue is empty, poll until a task appears or SECONDS "
526 f"elapses (default 0 = return immediately). "
527 f"Maximum {_MAX_WAIT_SECONDS} s. Uses exponential backoff."
528 ),
529 )
530 _add_format_args(parser)
531 parser.set_defaults(func=run_claim)
532
533
534 def run_claim(args: argparse.Namespace) -> None:
535 """Atomically claim the highest-priority pending task.
536
537 Execution order
538 ---------------
539 1. **Validate inputs** — all validation fires before any file I/O:
540
541 * ``--run-id`` must be ≤ :data:`_MAX_RUN_ID_LEN` characters.
542 * ``--claim-ttl`` must be in
543 ``[_MIN_CLAIM_TTL, _MAX_CLAIM_TTL]``.
544 * ``--wait`` must be in ``[0, _MAX_WAIT_SECONDS]``.
545 * ``--queue``, when provided, must match ``[a-zA-Z0-9_-]+``.
546
547 Any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1)
548 with a message on *stderr* (or compact JSON when ``--format json``).
549
550 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`.
551 3. **Claim loop** — calls :func:`~muse.core.task_queue.claim_next_task`,
552 which implements O_CREAT|O_EXCL (initial claim) and atomic rename +
553 nonce verification (re-claim of timed-out tasks). If ``--wait > 0``
554 and the queue is empty, polls with exponential backoff (0.5 s → 5 s)
555 until either a task is claimed or the wait deadline is reached.
556 4. **Emit output** — compact JSON to *stdout* when ``--format json``,
557 or human-readable text otherwise.
558
559 Concurrency
560 -----------
561 Exactly one agent wins per unclaimed task via ``O_CREAT | O_EXCL``.
562 For timed-out re-claims, the winner is determined by nonce read-back;
563 in the extremely unlikely event two agents race on a single rename, the
564 loser sees a mismatched nonce and moves to the next candidate.
565
566 Security
567 --------
568 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound claim
569 record size.
570 * ``--queue`` is validated against ``[a-zA-Z0-9_-]+`` before any file
571 I/O, preventing path traversal via malformed queue names.
572 * ``--claim-ttl`` is bounds-checked to prevent immediately-expired
573 claims (TTL < 1) or infinite locks (TTL > 86400).
574 * All display strings in text output are passed through
575 :func:`~muse.core.validation.sanitize_display`.
576
577 Performance
578 -----------
579 All input validation fires before ``require_repo()`` so bad-argument
580 errors never touch the filesystem. The claim scan is O(n) in the
581 number of task files; re-sorting is O(n log n).
582
583 Args:
584 args: Parsed ``argparse.Namespace`` with attributes ``run_id``,
585 ``queue``, ``claim_ttl``, ``wait``, and ``fmt``.
586
587 Exit codes:
588 0 — task claimed successfully.
589 1 — queue empty (after exhausting ``--wait``) or bad arguments.
590 """
591 t0 = time.monotonic()
592 as_json = args.fmt == "json"
593 wait_seconds: int = getattr(args, "wait", 0)
594
595 # ── Input validation (before any file I/O) ────────────────────────────────
596
597 if len(args.run_id) > _MAX_RUN_ID_LEN:
598 msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})"
599 _err(msg, as_json, "bad_args")
600 raise SystemExit(ExitCode.USER_ERROR)
601
602 if not (_MIN_CLAIM_TTL <= args.claim_ttl <= _MAX_CLAIM_TTL):
603 msg = (
604 f"--claim-ttl must be between {_MIN_CLAIM_TTL} and "
605 f"{_MAX_CLAIM_TTL}, got {args.claim_ttl}"
606 )
607 _err(msg, as_json, "bad_args")
608 raise SystemExit(ExitCode.USER_ERROR)
609
610 if not (0 <= wait_seconds <= _MAX_WAIT_SECONDS):
611 msg = f"--wait must be between 0 and {_MAX_WAIT_SECONDS}, got {wait_seconds}"
612 _err(msg, as_json, "bad_args")
613 raise SystemExit(ExitCode.USER_ERROR)
614
615 if args.queue is not None:
616 try:
617 _validate_queue_name(args.queue)
618 except ValueError as exc:
619 _err(str(exc), as_json, "bad_queue")
620 raise SystemExit(ExitCode.USER_ERROR)
621
622 root = require_repo()
623
624 # ── Claim loop (with optional --wait polling) ─────────────────────────────
625
626 result = None
627 try:
628 result = claim_next_task(
629 root,
630 args.run_id,
631 queue=args.queue,
632 claim_ttl_seconds=args.claim_ttl,
633 )
634 except (ValueError, OSError) as exc:
635 _err(str(exc), as_json)
636 raise SystemExit(ExitCode.USER_ERROR)
637
638 if result is None and wait_seconds > 0:
639 deadline = time.monotonic() + wait_seconds
640 poll = 0.5
641 while time.monotonic() < deadline:
642 remaining = deadline - time.monotonic()
643 time.sleep(min(poll, remaining))
644 poll = min(poll * 1.5, 5.0)
645 try:
646 result = claim_next_task(
647 root,
648 args.run_id,
649 queue=args.queue,
650 claim_ttl_seconds=args.claim_ttl,
651 )
652 except (ValueError, OSError) as exc:
653 _err(str(exc), as_json)
654 raise SystemExit(ExitCode.USER_ERROR)
655 if result is not None:
656 break
657
658 elapsed = round(time.monotonic() - t0, 4)
659
660 if result is None:
661 if as_json:
662 print(json.dumps({
663 "schema_version": _schema_version(),
664 "status": "empty",
665 "queue": args.queue,
666 "elapsed_seconds": elapsed,
667 }))
668 else:
669 queue_str = sanitize_display(args.queue or "any")
670 waited = f" after {elapsed:.1f}s" if wait_seconds > 0 else ""
671 print(f"\n Queue [{queue_str}] is empty — no pending tasks{waited}.")
672 print(f"\n ({elapsed:.3f}s)")
673 raise SystemExit(ExitCode.USER_ERROR)
674
675 task, claim = result
676
677 if as_json:
678 out = claim.to_dict()
679 out["task"] = task.to_dict()
680 out["elapsed_seconds"] = elapsed
681 print(json.dumps(out))
682 return
683
684 ttl_remaining = int((claim.expires_at - claim.claimed_at).total_seconds())
685 print(f"\n🔒 Task claimed")
686 print(f" Task ID: {sanitize_display(claim.task_id)}")
687 print(f" Title: {sanitize_display(task.title)}")
688 print(f" Claimer: {sanitize_display(claim.claimer_run_id)}")
689 print(f" Queue: {sanitize_display(task.queue)} priority={task.priority}")
690 print(f" Expires in: {ttl_remaining}s ({claim.expires_at.isoformat()})")
691 if task.payload:
692 print(f" Payload: {json.dumps(task.payload, indent=None)[:120]}")
693 if task.tags:
694 print(f" Tags: {', '.join(sanitize_display(t) for t in task.tags)}")
695 print(f"\n ({elapsed:.3f}s)")
696
697
698 # ── muse coord complete ───────────────────────────────────────────────────────
699
700
701 def register_complete(
702 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
703 ) -> None:
704 """Register ``complete`` on *subparsers* (under ``muse coord``).
705
706 Wires all flags with their defaults and help text so that ``--help``
707 output is accurate. Sets ``func`` to :func:`run_complete`.
708
709 Flags registered
710 ----------------
711 ``task_id`` (positional)
712 UUID of the task to mark completed. Must be a valid UUID4; validated
713 before any file I/O.
714 ``--run-id RUNID``
715 Required. The claiming agent's identifier — must exactly match the
716 ``claimer_run_id`` recorded when the task was claimed. Capped at
717 :data:`_MAX_RUN_ID_LEN` characters.
718 ``--result JSON``
719 Optional JSON object describing the outcome (e.g. proposal URL, artefact
720 paths). Must be a JSON object (not an array or scalar). Serialised
721 size is capped at :data:`_MAX_RESULT_BYTES` bytes to prevent unbounded
722 claim files. Defaults to ``{}``.
723 ``--format`` / ``--json``
724 Emit compact JSON to stdout; default is human-readable text.
725
726 JSON output schema::
727
728 {
729 "schema_version": str,
730 "task_id": str,
731 "claimer_run_id": str,
732 "claimed_at": str, // ISO 8601
733 "expires_at": str, // ISO 8601
734 "status": "completed",
735 "heartbeat_at": str, // ISO 8601
736 "claim_nonce": str,
737 "result": dict | null,
738 "error": null,
739 "elapsed_seconds": float
740 }
741
742 Exit codes::
743
744 0 — task marked completed
745 1 — bad arguments, task not found, not claimed, or wrong run-id
746 """
747 parser = subparsers.add_parser(
748 "complete",
749 help="Mark a claimed task as successfully completed.",
750 formatter_class=argparse.RawDescriptionHelpFormatter,
751 description=__doc__,
752 )
753 parser.add_argument("task_id", help="UUID of the task to complete.")
754 parser.add_argument(
755 "--run-id",
756 required=True,
757 dest="run_id",
758 metavar="RUNID",
759 help=(
760 "Claiming agent identifier — must match the original claimer. "
761 f"Maximum {_MAX_RUN_ID_LEN} characters."
762 ),
763 )
764 parser.add_argument(
765 "--result",
766 default="{}",
767 metavar="JSON",
768 help=(
769 "JSON object describing the outcome (default: {{}}). "
770 f"Maximum {_MAX_RESULT_BYTES} bytes serialised."
771 ),
772 )
773 _add_format_args(parser)
774 parser.set_defaults(func=run_complete)
775
776
777 def run_complete(args: argparse.Namespace) -> None:
778 """Mark a claimed task as completed.
779
780 Execution order
781 ---------------
782 1. **Validate inputs** — ``task_id`` UUID syntax, ``--run-id`` length,
783 ``--result`` JSON parse, ``--result`` serialised byte size. All
784 validation fires before any file I/O; any failure exits
785 :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1).
786 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`.
787 3. **Complete task** — delegates to
788 :func:`~muse.core.task_queue.complete_task`, which validates claimer
789 ownership and atomically updates the claim record via
790 :func:`~muse.core.store.write_text_atomic`.
791 4. **Emit output** — compact JSON to *stdout* when ``--format json``,
792 or human-readable text otherwise.
793
794 Security
795 --------
796 * ``task_id`` is validated as a UUID4 before ``require_repo()`` — prevents
797 any path traversal via a crafted task ID.
798 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters.
799 * ``--result`` JSON is parsed and size-checked against
800 :data:`_MAX_RESULT_BYTES` before I/O — prevents unbounded claim files.
801 * All display strings are passed through
802 :func:`~muse.core.validation.sanitize_display` to strip ANSI injection.
803
804 Performance
805 -----------
806 O(1) — two file reads (task + claim) and one atomic write.
807
808 Args:
809 args: Parsed ``argparse.Namespace`` with attributes ``task_id``,
810 ``run_id``, ``result``, and ``fmt``.
811
812 Exit codes:
813 0 — task marked completed.
814 1 — bad arguments, task not found, not claimed, or wrong run-id.
815 """
816 t0 = time.monotonic()
817 as_json = args.fmt == "json"
818
819 # ── Input validation (before any file I/O) ────────────────────────────────
820
821 if len(args.run_id) > _MAX_RUN_ID_LEN:
822 msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})"
823 _err(msg, as_json, "bad_args")
824 raise SystemExit(ExitCode.USER_ERROR)
825
826 try:
827 result_data = json.loads(args.result)
828 if not isinstance(result_data, dict):
829 raise ValueError("--result must be a JSON object, not a scalar or array")
830 except (json.JSONDecodeError, ValueError) as exc:
831 _err(f"invalid --result: {exc}", as_json, "bad_args")
832 raise SystemExit(ExitCode.USER_ERROR)
833
834 result_bytes = len(args.result.encode())
835 if result_bytes > _MAX_RESULT_BYTES:
836 msg = f"--result is too large ({result_bytes} bytes; max {_MAX_RESULT_BYTES})"
837 _err(msg, as_json, "bad_args")
838 raise SystemExit(ExitCode.USER_ERROR)
839
840 try:
841 from muse.core.task_queue import _validate_task_id
842 _validate_task_id(args.task_id)
843 except ValueError as exc:
844 _err(str(exc), as_json, "bad_task_id")
845 raise SystemExit(ExitCode.USER_ERROR)
846
847 root = require_repo()
848
849 try:
850 claim = complete_task(root, args.task_id, args.run_id, result=result_data or None)
851 except (FileNotFoundError, PermissionError, RuntimeError, ValueError) as exc:
852 _err(str(exc), as_json)
853 raise SystemExit(ExitCode.USER_ERROR)
854
855 elapsed = round(time.monotonic() - t0, 4)
856
857 if as_json:
858 out = claim.to_dict()
859 out["elapsed_seconds"] = elapsed
860 print(json.dumps(out))
861 return
862
863 task = load_task(root, claim.task_id)
864 print(f"\n✅ Task completed")
865 print(f" Task ID: {sanitize_display(claim.task_id)}")
866 if task is not None:
867 print(f" Title: {sanitize_display(task.title)}")
868 print(f" Queue: {sanitize_display(task.queue)}")
869 print(f" By: {sanitize_display(claim.claimer_run_id)}")
870 if result_data:
871 print(f" Result: {json.dumps(result_data, indent=None)[:120]}")
872 print(f"\n ({elapsed:.3f}s)")
873
874
875 # ── muse coord fail-task ──────────────────────────────────────────────────────
876
877
878 def register_fail_task(
879 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
880 ) -> None:
881 """Register ``fail-task`` on *subparsers* (under ``muse coord``).
882
883 Wires all flags with their defaults and help text so that ``--help``
884 output is accurate. Sets ``func`` to :func:`run_fail_task`.
885
886 Flags registered
887 ----------------
888 ``task_id`` (positional)
889 UUID of the task to mark failed. Must be a valid UUID4; validated
890 before any file I/O.
891 ``--run-id RUNID``
892 Required. The claiming agent's identifier — must exactly match the
893 ``claimer_run_id`` recorded when the task was claimed. Capped at
894 :data:`_MAX_RUN_ID_LEN` characters.
895 ``--error MESSAGE``
896 Human-readable error message describing why the task failed. Capped
897 at :data:`_MAX_ERROR_LEN` characters to keep claim files bounded.
898 Always include a meaningful message so orchestrators and retry agents
899 can understand the failure mode without reading logs.
900 ``--format`` / ``--json``
901 Emit compact JSON to stdout; default is human-readable text.
902
903 JSON output schema::
904
905 {
906 "schema_version": str,
907 "task_id": str,
908 "claimer_run_id": str,
909 "claimed_at": str, // ISO 8601
910 "expires_at": str, // ISO 8601
911 "status": "failed",
912 "heartbeat_at": str, // ISO 8601
913 "claim_nonce": str,
914 "result": null,
915 "error": str,
916 "elapsed_seconds": float
917 }
918
919 Exit codes::
920
921 0 — task marked failed
922 1 — bad arguments, task not found, not claimed, or wrong run-id
923 """
924 parser = subparsers.add_parser(
925 "fail-task",
926 help="Mark a claimed task as failed.",
927 formatter_class=argparse.RawDescriptionHelpFormatter,
928 description=__doc__,
929 )
930 parser.add_argument("task_id", help="UUID of the task to fail.")
931 parser.add_argument(
932 "--run-id",
933 required=True,
934 dest="run_id",
935 metavar="RUNID",
936 help=(
937 "Claiming agent identifier — must match the original claimer. "
938 f"Maximum {_MAX_RUN_ID_LEN} characters."
939 ),
940 )
941 parser.add_argument(
942 "--error",
943 default="",
944 metavar="MESSAGE",
945 help=(
946 "Human-readable error message describing why the task failed. "
947 f"Maximum {_MAX_ERROR_LEN} characters. Always provide a message "
948 "so orchestrators can diagnose failures without reading logs."
949 ),
950 )
951 _add_format_args(parser)
952 parser.set_defaults(func=run_fail_task)
953
954
955 def run_fail_task(args: argparse.Namespace) -> None:
956 """Mark a claimed task as failed.
957
958 Execution order
959 ---------------
960 1. **Validate inputs** — ``task_id`` UUID syntax, ``--run-id`` length,
961 ``--error`` length. All validation fires before any file I/O; any
962 failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1).
963 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`.
964 3. **Fail task** — delegates to :func:`~muse.core.task_queue.fail_task`,
965 which validates claimer ownership and atomically updates the claim
966 record via :func:`~muse.core.store.write_text_atomic`.
967 4. **Emit output** — compact JSON to *stdout* when ``--format json``,
968 or human-readable text otherwise.
969
970 When to use
971 -----------
972 Call ``fail-task`` when the claimed work cannot be completed — network
973 errors, missing dependencies, tool failures, etc. A failed task is
974 visible to the orchestrator and can be re-enqueued or escalated. Always
975 supply ``--error`` with enough detail for the next agent to understand
976 the failure without reading logs.
977
978 Security
979 --------
980 * ``task_id`` is validated as a UUID4 before ``require_repo()`` — prevents
981 path traversal via a crafted task ID.
982 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters.
983 * ``--error`` is capped at :data:`_MAX_ERROR_LEN` characters to bound
984 claim file sizes and memory use when loading failed claim sets.
985 * All display strings are passed through
986 :func:`~muse.core.validation.sanitize_display` to strip ANSI injection.
987
988 Performance
989 -----------
990 O(1) — two file reads (task + claim) and one atomic write.
991
992 Args:
993 args: Parsed ``argparse.Namespace`` with attributes ``task_id``,
994 ``run_id``, ``error``, and ``fmt``.
995
996 Exit codes:
997 0 — task marked failed.
998 1 — bad arguments, task not found, not claimed, or wrong run-id.
999 """
1000 t0 = time.monotonic()
1001 as_json = args.fmt == "json"
1002
1003 # ── Input validation (before any file I/O) ────────────────────────────────
1004
1005 if len(args.run_id) > _MAX_RUN_ID_LEN:
1006 msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})"
1007 _err(msg, as_json, "bad_args")
1008 raise SystemExit(ExitCode.USER_ERROR)
1009
1010 if len(args.error) > _MAX_ERROR_LEN:
1011 msg = f"--error is too long ({len(args.error)} chars; max {_MAX_ERROR_LEN})"
1012 _err(msg, as_json, "bad_args")
1013 raise SystemExit(ExitCode.USER_ERROR)
1014
1015 try:
1016 from muse.core.task_queue import _validate_task_id
1017 _validate_task_id(args.task_id)
1018 except ValueError as exc:
1019 _err(str(exc), as_json, "bad_task_id")
1020 raise SystemExit(ExitCode.USER_ERROR)
1021
1022 root = require_repo()
1023
1024 try:
1025 claim = fail_task(root, args.task_id, args.run_id, error=args.error)
1026 except (FileNotFoundError, PermissionError, RuntimeError, ValueError) as exc:
1027 _err(str(exc), as_json)
1028 raise SystemExit(ExitCode.USER_ERROR)
1029
1030 elapsed = round(time.monotonic() - t0, 4)
1031
1032 if as_json:
1033 out = claim.to_dict()
1034 out["elapsed_seconds"] = elapsed
1035 print(json.dumps(out))
1036 return
1037
1038 task = load_task(root, claim.task_id)
1039 print(f"\n❌ Task failed")
1040 print(f" Task ID: {sanitize_display(claim.task_id)}")
1041 if task is not None:
1042 print(f" Title: {sanitize_display(task.title)}")
1043 print(f" Queue: {sanitize_display(task.queue)}")
1044 print(f" By: {sanitize_display(claim.claimer_run_id)}")
1045 if args.error:
1046 print(f" Error: {sanitize_display(args.error[:200])}")
1047 print(f"\n ({elapsed:.3f}s)")
1048
1049
1050 # ── muse coord cancel-task ────────────────────────────────────────────────────
1051
1052
1053 def register_cancel_task(
1054 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
1055 ) -> None:
1056 """Register ``cancel-task`` on *subparsers* (under ``muse coord``).
1057
1058 Wires all flags with their defaults and help text so that ``--help``
1059 output is accurate. Sets ``func`` to :func:`run_cancel_task`.
1060
1061 Flags registered
1062 ----------------
1063 ``task_id`` (positional)
1064 UUID of the task to cancel. Must be a valid UUID4; validated before
1065 any file I/O.
1066 ``--run-id RUNID``
1067 Required. The calling agent's identifier. For pending tasks, this
1068 becomes the ``claimer_run_id`` in the cancel record. For claimed
1069 tasks, must match the original claimer unless ``--force`` is used.
1070 Capped at :data:`_MAX_RUN_ID_LEN` characters.
1071 ``--force``
1072 Cancel even if the task is claimed by a different agent. Use with
1073 caution — this aborts in-flight work without notifying the claimer.
1074 Harmless for pending tasks.
1075 ``--format`` / ``--json``
1076 Emit compact JSON to stdout; default is human-readable text.
1077
1078 JSON output schema::
1079
1080 {
1081 "schema_version": str,
1082 "task_id": str,
1083 "claimer_run_id": str,
1084 "claimed_at": str, // ISO 8601
1085 "expires_at": str, // ISO 8601
1086 "status": "cancelled",
1087 "heartbeat_at": str, // ISO 8601
1088 "claim_nonce": str,
1089 "result": null,
1090 "error": str,
1091 "elapsed_seconds": float
1092 }
1093
1094 Exit codes::
1095
1096 0 — task cancelled
1097 1 — bad arguments, task not found, already terminal, or permission denied
1098 """
1099 parser = subparsers.add_parser(
1100 "cancel-task",
1101 help="Cancel a pending or claimed task.",
1102 formatter_class=argparse.RawDescriptionHelpFormatter,
1103 description=__doc__,
1104 )
1105 parser.add_argument("task_id", help="UUID of the task to cancel.")
1106 parser.add_argument(
1107 "--run-id",
1108 required=True,
1109 dest="run_id",
1110 metavar="RUNID",
1111 help=(
1112 "Calling agent identifier. For claimed tasks, must match the "
1113 "original claimer unless --force is used. "
1114 f"Maximum {_MAX_RUN_ID_LEN} characters."
1115 ),
1116 )
1117 parser.add_argument(
1118 "--force",
1119 action="store_true",
1120 default=False,
1121 help=(
1122 "Cancel even if the task is claimed by a different agent. "
1123 "Use with caution — aborts in-flight work without notifying "
1124 "the claimer."
1125 ),
1126 )
1127 _add_format_args(parser)
1128 parser.set_defaults(func=run_cancel_task)
1129
1130
1131 def run_cancel_task(args: argparse.Namespace) -> None:
1132 """Cancel a pending or claimed task.
1133
1134 Execution order
1135 ---------------
1136 1. **Validate inputs** — ``task_id`` UUID syntax and ``--run-id`` length.
1137 All validation fires before any file I/O; any failure exits
1138 :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1).
1139 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`.
1140 3. **Cancel task** — delegates to :func:`~muse.core.task_queue.cancel_task`.
1141
1142 * **Pending task** — atomically creates a ``cancelled`` claim via
1143 ``O_CREAT | O_EXCL``. If another agent claims the task in the
1144 window between the check and the write, raises :exc:`FileExistsError`.
1145 * **Claimed task** — ownership check: ``--run-id`` must match the
1146 claimer unless ``--force`` is set. Atomically updates the claim
1147 record via :func:`~muse.core.store.write_text_atomic`.
1148 * **Terminal task** — raises :exc:`RuntimeError`; already
1149 completed/failed/cancelled tasks cannot be cancelled.
1150
1151 4. **Emit output** — compact JSON to *stdout* when ``--format json``,
1152 or human-readable text otherwise.
1153
1154 Security
1155 --------
1156 * ``task_id`` is validated as a UUID4 before ``require_repo()`` — prevents
1157 path traversal via a crafted task ID.
1158 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters.
1159 * All display strings are passed through
1160 :func:`~muse.core.validation.sanitize_display` to strip ANSI injection.
1161 * ``--force`` bypasses claimer ownership; use only when the orchestrator
1162 needs to reclaim a task from an unresponsive agent.
1163
1164 Performance
1165 -----------
1166 O(1) — one or two file reads and one atomic write.
1167
1168 Args:
1169 args: Parsed ``argparse.Namespace`` with attributes ``task_id``,
1170 ``run_id``, ``force``, and ``fmt``.
1171
1172 Exit codes:
1173 0 — task cancelled.
1174 1 — bad arguments, task not found, already terminal, or permission denied.
1175 """
1176 t0 = time.monotonic()
1177 as_json = args.fmt == "json"
1178
1179 # ── Input validation (before any file I/O) ────────────────────────────────
1180
1181 if len(args.run_id) > _MAX_RUN_ID_LEN:
1182 msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})"
1183 _err(msg, as_json, "bad_args")
1184 raise SystemExit(ExitCode.USER_ERROR)
1185
1186 try:
1187 from muse.core.task_queue import _validate_task_id
1188 _validate_task_id(args.task_id)
1189 except ValueError as exc:
1190 _err(str(exc), as_json, "bad_task_id")
1191 raise SystemExit(ExitCode.USER_ERROR)
1192
1193 root = require_repo()
1194
1195 try:
1196 claim = cancel_task(root, args.task_id, args.run_id, force=args.force)
1197 except (FileNotFoundError, FileExistsError, PermissionError, RuntimeError, ValueError) as exc:
1198 _err(str(exc), as_json)
1199 raise SystemExit(ExitCode.USER_ERROR)
1200
1201 elapsed = round(time.monotonic() - t0, 4)
1202
1203 if as_json:
1204 out = claim.to_dict()
1205 out["elapsed_seconds"] = elapsed
1206 print(json.dumps(out))
1207 return
1208
1209 task = load_task(root, claim.task_id)
1210 print(f"\n🚫 Task cancelled")
1211 print(f" Task ID: {sanitize_display(claim.task_id)}")
1212 if task is not None:
1213 print(f" Title: {sanitize_display(task.title)}")
1214 print(f" Queue: {sanitize_display(task.queue)}")
1215 print(f" By: {sanitize_display(claim.claimer_run_id)}")
1216 if args.force:
1217 print(f" (forced)")
1218 print(f"\n ({elapsed:.3f}s)")
1219
1220
1221 # ── muse coord tasks ──────────────────────────────────────────────────────────
1222
1223
1224 def register_tasks(
1225 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
1226 ) -> None:
1227 """Register ``tasks`` on *subparsers* (under ``muse coord``).
1228
1229 Wires all flags with their defaults and help text so that ``--help``
1230 output is accurate. Sets ``func`` to :func:`run_tasks`.
1231
1232 Flags registered
1233 ----------------
1234 ``--status STATUS``
1235 Filter items to a single status value. One of: ``pending``,
1236 ``claimed``, ``timed_out``, ``completed``, ``failed``,
1237 ``cancelled``. The global status counts in the output always
1238 reflect the *full* queue regardless of this filter.
1239 ``--queue QUEUE``
1240 Filter items by queue name. Must match ``[a-zA-Z0-9_-]+``;
1241 validated before any file I/O.
1242 ``--run-id RUNID``
1243 Filter items to tasks whose claimer matches *RUNID* (claimed,
1244 completed, and failed tasks only). Capped at
1245 :data:`_MAX_RUN_ID_LEN` characters.
1246 ``--limit N``
1247 Maximum number of items to return (default: 200; max:
1248 :data:`_MAX_LIMIT`). The status counts always reflect the full
1249 queue; only the ``items`` list is truncated.
1250 ``--format`` / ``--json``
1251 Emit compact JSON to stdout; default is human-readable text.
1252
1253 Derived status values::
1254
1255 pending — not yet claimed
1256 claimed — actively claimed, claim TTL not expired
1257 timed_out — claim TTL expired (eligible for re-claiming)
1258 completed — done successfully
1259 failed — done with error
1260 cancelled — cancelled before or after claiming
1261
1262 JSON output schema::
1263
1264 {
1265 "schema_version": str,
1266 "total": int,
1267 "pending": int,
1268 "claimed": int,
1269 "timed_out": int,
1270 "completed": int,
1271 "failed": int,
1272 "cancelled": int,
1273 "limit": int,
1274 "truncated": bool,
1275 "items": [
1276 {
1277 "task_id": str,
1278 "title": str,
1279 "priority": int,
1280 "queue": str,
1281 "status": str,
1282 "created_by": str,
1283 "created_at": str, // ISO 8601
1284 "ttl_seconds": int,
1285 "tags": [str, ...],
1286 "payload": dict,
1287 "claimer_run_id": str | null,
1288 "expires_at": str | null // ISO 8601; null if not claimed
1289 },
1290 ...
1291 ],
1292 "elapsed_seconds": float
1293 }
1294
1295 Exit codes::
1296
1297 0 — success (empty queue is still success)
1298 1 — bad arguments or unexpected error
1299 """
1300 parser = subparsers.add_parser(
1301 "tasks",
1302 help="List tasks with optional status/queue/run-id/limit filtering.",
1303 formatter_class=argparse.RawDescriptionHelpFormatter,
1304 description=__doc__,
1305 )
1306 parser.add_argument(
1307 "--status",
1308 default=None,
1309 choices=("pending", "claimed", "timed_out", "completed", "failed", "cancelled"),
1310 metavar="STATUS",
1311 help=(
1312 "Filter by status: pending | claimed | timed_out | "
1313 "completed | failed | cancelled"
1314 ),
1315 )
1316 parser.add_argument(
1317 "--queue",
1318 default=None,
1319 metavar="QUEUE",
1320 help=(
1321 "Filter by queue name. Must match [a-zA-Z0-9_-]+. "
1322 "Validated before any file I/O."
1323 ),
1324 )
1325 parser.add_argument(
1326 "--run-id",
1327 default=None,
1328 dest="run_id",
1329 metavar="RUNID",
1330 help=(
1331 "Filter by claimer run_id (claimed/completed/failed tasks only). "
1332 f"Maximum {_MAX_RUN_ID_LEN} characters."
1333 ),
1334 )
1335 parser.add_argument(
1336 "--limit",
1337 type=int,
1338 default=200,
1339 metavar="N",
1340 help=(
1341 f"Maximum items to return (default: 200; max: {_MAX_LIMIT}). "
1342 "Status counts always reflect the full queue."
1343 ),
1344 )
1345 _add_format_args(parser)
1346 parser.set_defaults(func=run_tasks)
1347
1348
1349 def run_tasks(args: argparse.Namespace) -> None:
1350 """List tasks from the coordination queue with filtering and pagination.
1351
1352 Execution order
1353 ---------------
1354 1. **Validate inputs** — ``--queue`` name syntax, ``--run-id`` length,
1355 ``--limit`` bounds (1–:data:`_MAX_LIMIT`). All validation fires
1356 before any file I/O; any failure exits
1357 :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1).
1358 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`.
1359 3. **Load state** — one directory scan each for tasks and claims.
1360 4. **Build items** — derive per-task status, apply filters, sort by
1361 priority desc / created_at asc, then truncate to ``--limit``.
1362 5. **Compute counts** — iterate ``all_tasks`` once for global status
1363 counts (counts always reflect the *full* queue, never the filtered
1364 page).
1365 6. **Emit output** — compact JSON to *stdout* when ``--format json``,
1366 or human-readable text otherwise.
1367
1368 Security
1369 --------
1370 * ``--queue`` is validated against ``[a-zA-Z0-9_-]+`` before I/O —
1371 prevents ANSI injection in text output via crafted queue names.
1372 * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters.
1373 * All strings sourced from persisted files are passed through
1374 :func:`~muse.core.validation.sanitize_display` before printing.
1375
1376 Performance
1377 -----------
1378 O(n) — one directory scan per kind (tasks, claims). The ``--limit``
1379 flag prevents large outputs for agents that only need a page. For very
1380 large queues (10 000+ tasks), direct filesystem scanning is the
1381 bottleneck; MuseHub coordination will replace this for distributed
1382 deployments.
1383
1384 Args:
1385 args: Parsed ``argparse.Namespace`` with attributes ``status``,
1386 ``queue``, ``run_id``, ``limit``, and ``fmt``.
1387
1388 Exit codes:
1389 0 — success (empty queue is still success).
1390 1 — bad arguments or unexpected error.
1391 """
1392 import datetime as _dt
1393
1394 t0 = time.monotonic()
1395 as_json = args.fmt == "json"
1396 limit: int = getattr(args, "limit", 200)
1397
1398 # ── Input validation (before any file I/O) ────────────────────────────────
1399
1400 if args.queue is not None:
1401 try:
1402 _validate_queue_name(args.queue)
1403 except ValueError as exc:
1404 _err(str(exc), as_json, "bad_queue")
1405 raise SystemExit(ExitCode.USER_ERROR)
1406
1407 if args.run_id is not None and len(args.run_id) > _MAX_RUN_ID_LEN:
1408 msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})"
1409 _err(msg, as_json, "bad_args")
1410 raise SystemExit(ExitCode.USER_ERROR)
1411
1412 if not (1 <= limit <= _MAX_LIMIT):
1413 msg = f"--limit must be between 1 and {_MAX_LIMIT}, got {limit}"
1414 _err(msg, as_json, "bad_args")
1415 raise SystemExit(ExitCode.USER_ERROR)
1416
1417 root = require_repo()
1418
1419 all_tasks = load_all_tasks(root)
1420 all_claims = load_all_claims(root)
1421 now_ts = _dt.datetime.now(_dt.timezone.utc)
1422
1423 # Build enriched items (filtered).
1424 items = []
1425 for task in all_tasks:
1426 claim = all_claims.get(task.task_id)
1427 status = get_task_status(task, claim, now_ts)
1428
1429 if args.status and status != args.status:
1430 continue
1431 if args.queue and task.queue != args.queue:
1432 continue
1433 if args.run_id:
1434 if claim is None or claim.claimer_run_id != args.run_id:
1435 continue
1436
1437 items.append({
1438 "task_id": task.task_id,
1439 "title": task.title,
1440 "priority": task.priority,
1441 "queue": task.queue,
1442 "status": status,
1443 "created_by": task.created_by,
1444 "created_at": task.created_at.isoformat(),
1445 "ttl_seconds": task.ttl_seconds,
1446 "tags": task.tags,
1447 "payload": task.payload,
1448 "claimer_run_id": claim.claimer_run_id if claim else None,
1449 "expires_at": claim.expires_at.isoformat() if claim else None,
1450 })
1451
1452 # Sort: priority desc, then created_at asc.
1453 items.sort(key=lambda i: (-i["priority"], i["created_at"]))
1454
1455 # Apply limit after sort so the highest-priority items are always included.
1456 truncated = len(items) > limit
1457 items = items[:limit]
1458
1459 elapsed = round(time.monotonic() - t0, 4)
1460
1461 # Global status counts — always computed from the full all_tasks list,
1462 # independent of any filter, so the summary bar reflects queue health.
1463 counts: _IntMap = {
1464 s: 0 for s in ("pending", "claimed", "timed_out", "completed", "failed", "cancelled")
1465 }
1466 for task in all_tasks:
1467 claim = all_claims.get(task.task_id)
1468 s = get_task_status(task, claim, now_ts)
1469 counts[s] += 1
1470
1471 if as_json:
1472 print(json.dumps({
1473 "schema_version": _schema_version(),
1474 "total": sum(counts.values()),
1475 **counts,
1476 "limit": limit,
1477 "truncated": truncated,
1478 "items": items,
1479 "elapsed_seconds": elapsed,
1480 }))
1481 return
1482
1483 # Text output.
1484 filter_parts = []
1485 if args.status:
1486 filter_parts.append(f"status={args.status}")
1487 if args.queue:
1488 filter_parts.append(f"queue={sanitize_display(args.queue)}")
1489 if args.run_id:
1490 filter_parts.append(f"run-id={sanitize_display(args.run_id)}")
1491 filter_str = f" filter: {', '.join(filter_parts)}" if filter_parts else ""
1492
1493 print(f"\nTask queue — {sum(counts.values())} task(s)")
1494 if filter_str:
1495 print(filter_str)
1496 print(
1497 f" {counts['pending']} pending "
1498 f"{counts['claimed']} claimed "
1499 f"{counts['timed_out']} timed_out "
1500 f"{counts['completed']} completed "
1501 f"{counts['failed']} failed "
1502 f"{counts['cancelled']} cancelled"
1503 )
1504 print("─" * 80)
1505
1506 if not items:
1507 print("\n (no tasks matching filter)")
1508 else:
1509 print(f"\n{'ID':8} {'ST':10} {'PRI':3} {'QUEUE':12} {'CLAIMER':20} TITLE")
1510 print("─" * 80)
1511 _STATUS_ICONS = {
1512 "pending": "⏳",
1513 "claimed": "🔒",
1514 "timed_out": "⏰",
1515 "completed": "✅",
1516 "failed": "❌",
1517 "cancelled": "🚫",
1518 }
1519 for item in items:
1520 icon = _STATUS_ICONS.get(item["status"], "?")
1521 tid = sanitize_display(item["task_id"][:8])
1522 st = f"{icon}{item['status'][:9]}"
1523 pri = str(item["priority"])
1524 q = sanitize_display(item["queue"][:12])
1525 claimer = sanitize_display((item["claimer_run_id"] or "-")[:20])
1526 title = sanitize_display(item["title"][:40])
1527 print(f"{tid} {st:<11} {pri:>3} {q:<12} {claimer:<20} {title}")
1528
1529 if truncated:
1530 print(f"\n (showing {limit} of {len(items) + (len(all_tasks) - limit)} — use --limit to see more)")
1531
1532 print(f"\n ({elapsed:.3f}s)")
1533
1534
1535 # ── Registration ──────────────────────────────────────────────────────────────
1536
1537
1538 def register_all(
1539 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
1540 ) -> None:
1541 """Register all task-queue subcommands on *subparsers*.
1542
1543 Called from :func:`muse.cli.app.main` to attach the five task-queue
1544 commands to the ``muse coord`` subparser group.
1545 """
1546 register_enqueue(subparsers)
1547 register_claim(subparsers)
1548 register_complete(subparsers)
1549 register_fail_task(subparsers)
1550 register_cancel_task(subparsers)
1551 register_tasks(subparsers)
1552
1553
1554 # ── Private helpers ───────────────────────────────────────────────────────────
1555
1556
1557 def _schema_version() -> str:
1558 from muse._version import __version__
1559 return __version__
File History 1 commit
sha256:1ddad36d76d3a8d323f9b3664169cb184b7a38b39208214a2ae504154260826f fix: show full cryptographic IDs in all human-readable CLI output Sonnet 4.6 patch 34 days ago