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