"""``muse coord`` task-queue subcommands — real work distribution. Provides five subcommands for operating the file-system–based task queue: ``muse coord enqueue`` Add a task to the queue. ``muse coord claim`` Atomically claim the highest-priority pending task. Exactly one agent wins when multiple agents call ``claim`` concurrently. ``muse coord complete`` Mark a claimed task as successfully completed. ``muse coord fail-task`` Mark a claimed task as failed (agent could not complete the work). ``muse coord tasks`` List tasks with optional status/queue/run-id filtering. Typical multi-agent workflow:: # Orchestrator enqueues work: muse coord enqueue "Refactor billing module" \\ --priority 5 --payload '{"addresses":["billing.py::*"]}' \\ --run-id orchestrator --tags billing # N agents compete for tasks (exactly one wins per call): TASK=$(muse coord claim --run-id $AGENT_ID --json) TASK_ID=$(echo "$TASK" | python3 -c "import sys,json; print(json.load(sys.stdin)['task_id'])") # Agent works on the task … # Agent reports result: muse coord complete $TASK_ID --run-id $AGENT_ID --result '{"pr": 42}' # or on failure: muse coord fail-task $TASK_ID --run-id $AGENT_ID --error "timed out" Exit codes (all subcommands):: 0 — success 1 — bad arguments, task not found, queue empty, permission error, or unexpected error JSON output ----------- All subcommands accept ``--json`` / ``--format json``. The JSON schema for each is documented in the individual function docstrings below. """ from __future__ import annotations import argparse import json import sys import time from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.task_queue import ( ClaimRecord, TaskRecord, _MAX_QUEUE_LEN, _MAX_TAG_LEN, _MAX_TAGS, _MAX_TITLE_LEN, _validate_queue_name, cancel_task, claim_next_task, complete_task, create_task, fail_task, get_task_status, heartbeat_claim, load_all_claims, load_all_tasks, load_claim, load_task, ) from muse.core.validation import sanitize_display type _IntMap = dict[str, int] # ── Module-level constants ──────────────────────────────────────────────────── #: Maximum byte-length of ``--run-id``. Mirrors all other coordination commands. _MAX_RUN_ID_LEN: int = 256 #: Maximum byte-length of the serialised ``--payload`` JSON string. #: Prevents unbounded memory use when the payload is later loaded into RAM. _MAX_PAYLOAD_BYTES: int = 65536 # 64 KiB #: Minimum claim TTL in seconds. A TTL of 0 creates an immediately-expired #: claim which would be immediately re-claimable by any other agent. _MIN_CLAIM_TTL: int = 1 #: Maximum claim TTL in seconds (24 h). Prevents agents from holding tasks #: indefinitely with no heartbeat path. _MAX_CLAIM_TTL: int = 86400 #: Maximum ``--wait`` polling duration in seconds (1 h). _MAX_WAIT_SECONDS: int = 3600 #: Maximum byte-length of the serialised ``--result`` JSON string. #: Mirrors :data:`_MAX_PAYLOAD_BYTES`; keeps claim files bounded in size. _MAX_RESULT_BYTES: int = 65536 # 64 KiB #: Maximum character-length of the ``--error`` message on fail-task. #: Bounds claim file sizes and prevents pathological memory use when loading #: large sets of failed claims for analysis. _MAX_ERROR_LEN: int = 4096 #: Maximum value of ``--limit`` on ``tasks``. Prevents accidentally dumping #: the entire queue in one shot on very large repos. _MAX_LIMIT: int = 10000 # ── Shared argument helpers ─────────────────────────────────────────────────── def _add_format_args(parser: argparse.ArgumentParser) -> None: """Add ``--format``/``-f`` and ``--json`` to *parser*.""" parser.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) def _err(msg: str, as_json: bool = False, status: str = "error") -> None: """Print an error message. In JSON mode, emits ``{"error": msg, "status": status}`` to *stdout* (so the machine-readable stream is never broken by a bare text line). In text mode, prefixes with ``❌`` and writes to *stderr*. """ if as_json: print(json.dumps({"error": msg, "status": status})) else: print(f"❌ {msg}", file=sys.stderr) # ── muse coord enqueue ──────────────────────────────────────────────────────── def register_enqueue( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register ``enqueue`` on *subparsers* (under ``muse coord``). Wires all flags with their defaults, choices, and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run_enqueue`. Flags registered ---------------- ``title`` Positional. Short human-readable description (≤ ``_MAX_TITLE_LEN`` chars; trimmed by the core layer, not rejected). ``--priority N`` Integer priority; higher = claimed first. Default: 0. ``--queue QUEUE`` Target logical queue. Must match ``[a-zA-Z0-9_-]+``. Default: ``"default"``. ``--ttl SECONDS`` Pending TTL in seconds. Must be ≥ 1. Default: 86400 (24 h). ``--run-id RUNID`` Enqueuing agent identifier. Maximum :data:`_MAX_RUN_ID_LEN` chars. ``--payload JSON`` Arbitrary JSON object (must be ``{}``-style, not an array). Maximum :data:`_MAX_PAYLOAD_BYTES` bytes when UTF-8 encoded. ``--tags TAG1,TAG2`` Comma-separated tag list. Maximum :data:`~muse.core.task_queue._MAX_TAGS` tags; each tag is trimmed to :data:`~muse.core.task_queue._MAX_TAG_LEN` chars by the core layer. ``--format`` / ``--json`` Machine-readable compact JSON output. JSON output schema:: { "schema_version": str, "task_id": str, // UUID "title": str, "priority": int, "queue": str, "ttl_seconds": int, "created_by": str, "created_at": str, // ISO 8601 UTC "tags": [str, ...], "payload": dict, "elapsed_seconds": float } Exit codes:: 0 — task enqueued 1 — bad arguments """ parser = subparsers.add_parser( "enqueue", help="Add a task to the coordination task queue.", formatter_class=argparse.RawDescriptionHelpFormatter, description="""Add a work item to the task queue for agents to claim and execute. Tasks are ordered by priority (higher = first) then creation time (FIFO within same priority). The ``payload`` field carries arbitrary JSON that the claiming agent receives along with the task definition. """, ) parser.add_argument( "title", help=( f"Short human-readable task description (≤ {_MAX_TITLE_LEN} chars)." ), ) parser.add_argument( "--priority", type=int, default=0, metavar="N", help="Task priority (higher = processed first, default 0).", ) parser.add_argument( "--queue", default="default", metavar="QUEUE", help=( f"Target queue name (default: 'default'). " f"Max {_MAX_QUEUE_LEN} chars, pattern [a-zA-Z0-9_-]." ), ) parser.add_argument( "--ttl", type=int, default=86400, dest="ttl_seconds", metavar="SECONDS", help="Seconds the task may remain pending before expiry (default 86400, min 1).", ) parser.add_argument( "--run-id", default="unknown", dest="run_id", metavar="RUNID", help=( f"Enqueuing agent / orchestrator identifier. " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--payload", default="{}", metavar="JSON", help=( f"Arbitrary JSON object attached to the task (default: {{}}). " f"Must be a JSON object (not array). " f"Maximum {_MAX_PAYLOAD_BYTES} bytes when UTF-8 encoded." ), ) parser.add_argument( "--tags", default="", metavar="TAG1,TAG2", help=( f"Comma-separated list of tags. " f"Maximum {_MAX_TAGS} tags, each ≤ {_MAX_TAG_LEN} chars." ), ) _add_format_args(parser) parser.set_defaults(func=run_enqueue) def run_enqueue(args: argparse.Namespace) -> None: """Create and persist a new task in the coordination queue. Execution order --------------- 1. **Validate inputs** — all validation fires before any file I/O: * ``title`` must be non-empty. * ``--run-id`` must be ≤ :data:`_MAX_RUN_ID_LEN` characters. * ``--ttl`` must be ≥ 1. * ``--payload`` must be valid JSON, a JSON object (not array), and ≤ :data:`_MAX_PAYLOAD_BYTES` bytes when UTF-8 encoded. * ``--queue`` must match ``[a-zA-Z0-9_-]+`` and be ≤ :data:`~muse.core.task_queue._MAX_QUEUE_LEN` characters. * ``--tags`` list must contain ≤ :data:`~muse.core.task_queue._MAX_TAGS` tags after splitting on commas. Any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message on *stderr* (or compact JSON when ``--format json``). 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`. 3. **Create task** — calls :func:`~muse.core.task_queue.create_task`, which atomically writes ``.muse/coordination/tasks/.json``. 4. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Security -------- * ``--payload`` is parsed as JSON (not ``eval``'d), so shell injection via JSON content is not possible. * ``--payload`` is size-capped at :data:`_MAX_PAYLOAD_BYTES` to prevent unbounded memory use when loading large queues. * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound audit record size. * Queue name is validated against ``[a-zA-Z0-9_-]+`` before any file I/O. * All display strings in text output are passed through :func:`~muse.core.validation.sanitize_display`. Performance ----------- All input validation fires before ``require_repo()`` so bad-argument errors never touch the filesystem. Args: args: Parsed ``argparse.Namespace`` with attributes ``title``, ``priority``, ``queue``, ``ttl_seconds``, ``run_id``, ``payload``, ``tags``, and ``fmt``. Exit codes: 0 — task enqueued successfully. 1 — bad arguments. """ t0 = time.monotonic() as_json = args.fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if not args.title or not args.title.strip(): msg = "title must be non-empty" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if len(args.run_id) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if args.ttl_seconds < 1: msg = f"--ttl must be ≥ 1, got {args.ttl_seconds}" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) payload_bytes = args.payload.encode() if len(payload_bytes) > _MAX_PAYLOAD_BYTES: msg = ( f"--payload is too large ({len(payload_bytes)} bytes; " f"max {_MAX_PAYLOAD_BYTES})" ) _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) try: payload = json.loads(args.payload) if not isinstance(payload, dict): raise ValueError("payload must be a JSON object") except (json.JSONDecodeError, ValueError) as exc: _err(f"invalid --payload: {exc}", as_json, "bad_payload") raise SystemExit(ExitCode.USER_ERROR) tags = [t.strip() for t in args.tags.split(",") if t.strip()] if args.tags else [] if len(tags) > _MAX_TAGS: msg = f"too many tags: {len(tags)} (max {_MAX_TAGS})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) try: _validate_queue_name(args.queue) except ValueError as exc: _err(str(exc), as_json, "bad_queue") raise SystemExit(ExitCode.USER_ERROR) root = require_repo() try: task = create_task( root, args.title, payload=payload, priority=args.priority, queue=args.queue, ttl_seconds=args.ttl_seconds, created_by=args.run_id, tags=tags, ) except ValueError as exc: _err(str(exc), as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) elapsed = round(time.monotonic() - t0, 4) if as_json: out = task.to_dict() out["elapsed_seconds"] = elapsed print(json.dumps(out)) return print(f"\n✅ Task enqueued") print(f" Task ID: {sanitize_display(task.task_id)}") print(f" Title: {sanitize_display(task.title)}") print(f" Queue: {sanitize_display(task.queue)} priority={task.priority}") print(f" TTL: {task.ttl_seconds}s") if task.tags: print(f" Tags: {', '.join(sanitize_display(t) for t in task.tags)}") print(f"\n ({elapsed:.3f}s)") # ── muse coord claim ────────────────────────────────────────────────────────── def register_claim( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register ``claim`` on *subparsers* (under ``muse coord``). Wires all flags with their defaults, choices, and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run_claim`. Flags registered ---------------- ``--run-id RUNID`` Required. Claiming agent identifier stored in the claim record. Maximum :data:`_MAX_RUN_ID_LEN` characters. ``--queue QUEUE`` Restrict claiming to a specific queue. When omitted, the highest-priority task across all queues is claimed. Must match ``[a-zA-Z0-9_-]+`` when provided. ``--claim-ttl SECONDS`` How long the claim is valid before another agent may re-claim the task. Must be in ``[_MIN_CLAIM_TTL, _MAX_CLAIM_TTL]``. Agents should call ``muse coord heartbeat`` every ``claim-ttl / 2`` seconds to keep the claim alive. Default: 3600 (1 h). ``--wait SECONDS`` If the queue is empty, poll until a task appears or *SECONDS* elapses. ``0`` (default) returns immediately on empty. Maximum :data:`_MAX_WAIT_SECONDS`. Uses exponential backoff capped at 5 s. ``--format`` / ``--json`` Machine-readable compact JSON output. JSON output schema (success, exit 0):: { "schema_version": str, "status": "claimed", "task_id": str, "claimer_run_id": str, "claimed_at": str, // ISO 8601 UTC "expires_at": str, // ISO 8601 UTC "task": { ... }, // full TaskRecord fields "elapsed_seconds": float } JSON output schema (queue empty after any wait, exit 1):: { "schema_version": str, "status": "empty", "queue": str | null, "elapsed_seconds": float } Exit codes:: 0 — task claimed successfully 1 — queue empty (after exhausting --wait period) or bad arguments """ parser = subparsers.add_parser( "claim", help="Atomically claim the highest-priority pending task.", formatter_class=argparse.RawDescriptionHelpFormatter, description="""Scan the task queue and atomically claim the highest-priority pending task using O_CREAT|O_EXCL. When multiple agents call 'claim' simultaneously, exactly one wins per task — no task is ever claimed twice. Timed-out tasks (claim.expires_at < now) are eligible for re-claiming using optimistic concurrency (atomic rename + read-back nonce verification). """, ) parser.add_argument( "--run-id", required=True, dest="run_id", metavar="RUNID", help=( f"Claiming agent identifier (stored in the claim record). " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--queue", default=None, metavar="QUEUE", help=( "Only claim from this queue (default: any queue). " "Must match [a-zA-Z0-9_-]+ when provided." ), ) parser.add_argument( "--claim-ttl", type=int, default=3600, dest="claim_ttl", metavar="SECONDS", help=( f"Claim TTL in seconds " f"(min {_MIN_CLAIM_TTL}, max {_MAX_CLAIM_TTL}, default 3600). " f"Heartbeat every claim-ttl/2 seconds to keep the claim alive." ), ) parser.add_argument( "--wait", type=int, default=0, dest="wait", metavar="SECONDS", help=( f"If the queue is empty, poll until a task appears or SECONDS " f"elapses (default 0 = return immediately). " f"Maximum {_MAX_WAIT_SECONDS} s. Uses exponential backoff." ), ) _add_format_args(parser) parser.set_defaults(func=run_claim) def run_claim(args: argparse.Namespace) -> None: """Atomically claim the highest-priority pending task. Execution order --------------- 1. **Validate inputs** — all validation fires before any file I/O: * ``--run-id`` must be ≤ :data:`_MAX_RUN_ID_LEN` characters. * ``--claim-ttl`` must be in ``[_MIN_CLAIM_TTL, _MAX_CLAIM_TTL]``. * ``--wait`` must be in ``[0, _MAX_WAIT_SECONDS]``. * ``--queue``, when provided, must match ``[a-zA-Z0-9_-]+``. Any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message on *stderr* (or compact JSON when ``--format json``). 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`. 3. **Claim loop** — calls :func:`~muse.core.task_queue.claim_next_task`, which implements O_CREAT|O_EXCL (initial claim) and atomic rename + nonce verification (re-claim of timed-out tasks). If ``--wait > 0`` and the queue is empty, polls with exponential backoff (0.5 s → 5 s) until either a task is claimed or the wait deadline is reached. 4. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Concurrency ----------- Exactly one agent wins per unclaimed task via ``O_CREAT | O_EXCL``. For timed-out re-claims, the winner is determined by nonce read-back; in the extremely unlikely event two agents race on a single rename, the loser sees a mismatched nonce and moves to the next candidate. Security -------- * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound claim record size. * ``--queue`` is validated against ``[a-zA-Z0-9_-]+`` before any file I/O, preventing path traversal via malformed queue names. * ``--claim-ttl`` is bounds-checked to prevent immediately-expired claims (TTL < 1) or infinite locks (TTL > 86400). * All display strings in text output are passed through :func:`~muse.core.validation.sanitize_display`. Performance ----------- All input validation fires before ``require_repo()`` so bad-argument errors never touch the filesystem. The claim scan is O(n) in the number of task files; re-sorting is O(n log n). Args: args: Parsed ``argparse.Namespace`` with attributes ``run_id``, ``queue``, ``claim_ttl``, ``wait``, and ``fmt``. Exit codes: 0 — task claimed successfully. 1 — queue empty (after exhausting ``--wait``) or bad arguments. """ t0 = time.monotonic() as_json = args.fmt == "json" wait_seconds: int = getattr(args, "wait", 0) # ── Input validation (before any file I/O) ──────────────────────────────── if len(args.run_id) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if not (_MIN_CLAIM_TTL <= args.claim_ttl <= _MAX_CLAIM_TTL): msg = ( f"--claim-ttl must be between {_MIN_CLAIM_TTL} and " f"{_MAX_CLAIM_TTL}, got {args.claim_ttl}" ) _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if not (0 <= wait_seconds <= _MAX_WAIT_SECONDS): msg = f"--wait must be between 0 and {_MAX_WAIT_SECONDS}, got {wait_seconds}" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if args.queue is not None: try: _validate_queue_name(args.queue) except ValueError as exc: _err(str(exc), as_json, "bad_queue") raise SystemExit(ExitCode.USER_ERROR) root = require_repo() # ── Claim loop (with optional --wait polling) ───────────────────────────── result = None try: result = claim_next_task( root, args.run_id, queue=args.queue, claim_ttl_seconds=args.claim_ttl, ) except (ValueError, OSError) as exc: _err(str(exc), as_json) raise SystemExit(ExitCode.USER_ERROR) if result is None and wait_seconds > 0: deadline = time.monotonic() + wait_seconds poll = 0.5 while time.monotonic() < deadline: remaining = deadline - time.monotonic() time.sleep(min(poll, remaining)) poll = min(poll * 1.5, 5.0) try: result = claim_next_task( root, args.run_id, queue=args.queue, claim_ttl_seconds=args.claim_ttl, ) except (ValueError, OSError) as exc: _err(str(exc), as_json) raise SystemExit(ExitCode.USER_ERROR) if result is not None: break elapsed = round(time.monotonic() - t0, 4) if result is None: if as_json: print(json.dumps({ "schema_version": _schema_version(), "status": "empty", "queue": args.queue, "elapsed_seconds": elapsed, })) else: queue_str = sanitize_display(args.queue or "any") waited = f" after {elapsed:.1f}s" if wait_seconds > 0 else "" print(f"\n Queue [{queue_str}] is empty — no pending tasks{waited}.") print(f"\n ({elapsed:.3f}s)") raise SystemExit(ExitCode.USER_ERROR) task, claim = result if as_json: out = claim.to_dict() out["task"] = task.to_dict() out["elapsed_seconds"] = elapsed print(json.dumps(out)) return ttl_remaining = int((claim.expires_at - claim.claimed_at).total_seconds()) print(f"\n🔒 Task claimed") print(f" Task ID: {sanitize_display(claim.task_id)}") print(f" Title: {sanitize_display(task.title)}") print(f" Claimer: {sanitize_display(claim.claimer_run_id)}") print(f" Queue: {sanitize_display(task.queue)} priority={task.priority}") print(f" Expires in: {ttl_remaining}s ({claim.expires_at.isoformat()})") if task.payload: print(f" Payload: {json.dumps(task.payload, indent=None)[:120]}") if task.tags: print(f" Tags: {', '.join(sanitize_display(t) for t in task.tags)}") print(f"\n ({elapsed:.3f}s)") # ── muse coord complete ─────────────────────────────────────────────────────── def register_complete( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register ``complete`` on *subparsers* (under ``muse coord``). Wires all flags with their defaults and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run_complete`. Flags registered ---------------- ``task_id`` (positional) UUID of the task to mark completed. Must be a valid UUID4; validated before any file I/O. ``--run-id RUNID`` Required. The claiming agent's identifier — must exactly match the ``claimer_run_id`` recorded when the task was claimed. Capped at :data:`_MAX_RUN_ID_LEN` characters. ``--result JSON`` Optional JSON object describing the outcome (e.g. proposal URL, artefact paths). Must be a JSON object (not an array or scalar). Serialised size is capped at :data:`_MAX_RESULT_BYTES` bytes to prevent unbounded claim files. Defaults to ``{}``. ``--format`` / ``--json`` Emit compact JSON to stdout; default is human-readable text. JSON output schema:: { "schema_version": str, "task_id": str, "claimer_run_id": str, "claimed_at": str, // ISO 8601 "expires_at": str, // ISO 8601 "status": "completed", "heartbeat_at": str, // ISO 8601 "claim_nonce": str, "result": dict | null, "error": null, "elapsed_seconds": float } Exit codes:: 0 — task marked completed 1 — bad arguments, task not found, not claimed, or wrong run-id """ parser = subparsers.add_parser( "complete", help="Mark a claimed task as successfully completed.", formatter_class=argparse.RawDescriptionHelpFormatter, description=__doc__, ) parser.add_argument("task_id", help="UUID of the task to complete.") parser.add_argument( "--run-id", required=True, dest="run_id", metavar="RUNID", help=( "Claiming agent identifier — must match the original claimer. " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--result", default="{}", metavar="JSON", help=( "JSON object describing the outcome (default: {{}}). " f"Maximum {_MAX_RESULT_BYTES} bytes serialised." ), ) _add_format_args(parser) parser.set_defaults(func=run_complete) def run_complete(args: argparse.Namespace) -> None: """Mark a claimed task as completed. Execution order --------------- 1. **Validate inputs** — ``task_id`` UUID syntax, ``--run-id`` length, ``--result`` JSON parse, ``--result`` serialised byte size. All validation fires before any file I/O; any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1). 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`. 3. **Complete task** — delegates to :func:`~muse.core.task_queue.complete_task`, which validates claimer ownership and atomically updates the claim record via :func:`~muse.core.store.write_text_atomic`. 4. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Security -------- * ``task_id`` is validated as a UUID4 before ``require_repo()`` — prevents any path traversal via a crafted task ID. * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters. * ``--result`` JSON is parsed and size-checked against :data:`_MAX_RESULT_BYTES` before I/O — prevents unbounded claim files. * All display strings are passed through :func:`~muse.core.validation.sanitize_display` to strip ANSI injection. Performance ----------- O(1) — two file reads (task + claim) and one atomic write. Args: args: Parsed ``argparse.Namespace`` with attributes ``task_id``, ``run_id``, ``result``, and ``fmt``. Exit codes: 0 — task marked completed. 1 — bad arguments, task not found, not claimed, or wrong run-id. """ t0 = time.monotonic() as_json = args.fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if len(args.run_id) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) try: result_data = json.loads(args.result) if not isinstance(result_data, dict): raise ValueError("--result must be a JSON object, not a scalar or array") except (json.JSONDecodeError, ValueError) as exc: _err(f"invalid --result: {exc}", as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) result_bytes = len(args.result.encode()) if result_bytes > _MAX_RESULT_BYTES: msg = f"--result is too large ({result_bytes} bytes; max {_MAX_RESULT_BYTES})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) try: from muse.core.task_queue import _validate_task_id _validate_task_id(args.task_id) except ValueError as exc: _err(str(exc), as_json, "bad_task_id") raise SystemExit(ExitCode.USER_ERROR) root = require_repo() try: claim = complete_task(root, args.task_id, args.run_id, result=result_data or None) except (FileNotFoundError, PermissionError, RuntimeError, ValueError) as exc: _err(str(exc), as_json) raise SystemExit(ExitCode.USER_ERROR) elapsed = round(time.monotonic() - t0, 4) if as_json: out = claim.to_dict() out["elapsed_seconds"] = elapsed print(json.dumps(out)) return task = load_task(root, claim.task_id) print(f"\n✅ Task completed") print(f" Task ID: {sanitize_display(claim.task_id)}") if task is not None: print(f" Title: {sanitize_display(task.title)}") print(f" Queue: {sanitize_display(task.queue)}") print(f" By: {sanitize_display(claim.claimer_run_id)}") if result_data: print(f" Result: {json.dumps(result_data, indent=None)[:120]}") print(f"\n ({elapsed:.3f}s)") # ── muse coord fail-task ────────────────────────────────────────────────────── def register_fail_task( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register ``fail-task`` on *subparsers* (under ``muse coord``). Wires all flags with their defaults and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run_fail_task`. Flags registered ---------------- ``task_id`` (positional) UUID of the task to mark failed. Must be a valid UUID4; validated before any file I/O. ``--run-id RUNID`` Required. The claiming agent's identifier — must exactly match the ``claimer_run_id`` recorded when the task was claimed. Capped at :data:`_MAX_RUN_ID_LEN` characters. ``--error MESSAGE`` Human-readable error message describing why the task failed. Capped at :data:`_MAX_ERROR_LEN` characters to keep claim files bounded. Always include a meaningful message so orchestrators and retry agents can understand the failure mode without reading logs. ``--format`` / ``--json`` Emit compact JSON to stdout; default is human-readable text. JSON output schema:: { "schema_version": str, "task_id": str, "claimer_run_id": str, "claimed_at": str, // ISO 8601 "expires_at": str, // ISO 8601 "status": "failed", "heartbeat_at": str, // ISO 8601 "claim_nonce": str, "result": null, "error": str, "elapsed_seconds": float } Exit codes:: 0 — task marked failed 1 — bad arguments, task not found, not claimed, or wrong run-id """ parser = subparsers.add_parser( "fail-task", help="Mark a claimed task as failed.", formatter_class=argparse.RawDescriptionHelpFormatter, description=__doc__, ) parser.add_argument("task_id", help="UUID of the task to fail.") parser.add_argument( "--run-id", required=True, dest="run_id", metavar="RUNID", help=( "Claiming agent identifier — must match the original claimer. " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--error", default="", metavar="MESSAGE", help=( "Human-readable error message describing why the task failed. " f"Maximum {_MAX_ERROR_LEN} characters. Always provide a message " "so orchestrators can diagnose failures without reading logs." ), ) _add_format_args(parser) parser.set_defaults(func=run_fail_task) def run_fail_task(args: argparse.Namespace) -> None: """Mark a claimed task as failed. Execution order --------------- 1. **Validate inputs** — ``task_id`` UUID syntax, ``--run-id`` length, ``--error`` length. All validation fires before any file I/O; any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1). 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`. 3. **Fail task** — delegates to :func:`~muse.core.task_queue.fail_task`, which validates claimer ownership and atomically updates the claim record via :func:`~muse.core.store.write_text_atomic`. 4. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. When to use ----------- Call ``fail-task`` when the claimed work cannot be completed — network errors, missing dependencies, tool failures, etc. A failed task is visible to the orchestrator and can be re-enqueued or escalated. Always supply ``--error`` with enough detail for the next agent to understand the failure without reading logs. Security -------- * ``task_id`` is validated as a UUID4 before ``require_repo()`` — prevents path traversal via a crafted task ID. * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters. * ``--error`` is capped at :data:`_MAX_ERROR_LEN` characters to bound claim file sizes and memory use when loading failed claim sets. * All display strings are passed through :func:`~muse.core.validation.sanitize_display` to strip ANSI injection. Performance ----------- O(1) — two file reads (task + claim) and one atomic write. Args: args: Parsed ``argparse.Namespace`` with attributes ``task_id``, ``run_id``, ``error``, and ``fmt``. Exit codes: 0 — task marked failed. 1 — bad arguments, task not found, not claimed, or wrong run-id. """ t0 = time.monotonic() as_json = args.fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if len(args.run_id) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if len(args.error) > _MAX_ERROR_LEN: msg = f"--error is too long ({len(args.error)} chars; max {_MAX_ERROR_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) try: from muse.core.task_queue import _validate_task_id _validate_task_id(args.task_id) except ValueError as exc: _err(str(exc), as_json, "bad_task_id") raise SystemExit(ExitCode.USER_ERROR) root = require_repo() try: claim = fail_task(root, args.task_id, args.run_id, error=args.error) except (FileNotFoundError, PermissionError, RuntimeError, ValueError) as exc: _err(str(exc), as_json) raise SystemExit(ExitCode.USER_ERROR) elapsed = round(time.monotonic() - t0, 4) if as_json: out = claim.to_dict() out["elapsed_seconds"] = elapsed print(json.dumps(out)) return task = load_task(root, claim.task_id) print(f"\n❌ Task failed") print(f" Task ID: {sanitize_display(claim.task_id)}") if task is not None: print(f" Title: {sanitize_display(task.title)}") print(f" Queue: {sanitize_display(task.queue)}") print(f" By: {sanitize_display(claim.claimer_run_id)}") if args.error: print(f" Error: {sanitize_display(args.error[:200])}") print(f"\n ({elapsed:.3f}s)") # ── muse coord cancel-task ──────────────────────────────────────────────────── def register_cancel_task( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register ``cancel-task`` on *subparsers* (under ``muse coord``). Wires all flags with their defaults and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run_cancel_task`. Flags registered ---------------- ``task_id`` (positional) UUID of the task to cancel. Must be a valid UUID4; validated before any file I/O. ``--run-id RUNID`` Required. The calling agent's identifier. For pending tasks, this becomes the ``claimer_run_id`` in the cancel record. For claimed tasks, must match the original claimer unless ``--force`` is used. Capped at :data:`_MAX_RUN_ID_LEN` characters. ``--force`` Cancel even if the task is claimed by a different agent. Use with caution — this aborts in-flight work without notifying the claimer. Harmless for pending tasks. ``--format`` / ``--json`` Emit compact JSON to stdout; default is human-readable text. JSON output schema:: { "schema_version": str, "task_id": str, "claimer_run_id": str, "claimed_at": str, // ISO 8601 "expires_at": str, // ISO 8601 "status": "cancelled", "heartbeat_at": str, // ISO 8601 "claim_nonce": str, "result": null, "error": str, "elapsed_seconds": float } Exit codes:: 0 — task cancelled 1 — bad arguments, task not found, already terminal, or permission denied """ parser = subparsers.add_parser( "cancel-task", help="Cancel a pending or claimed task.", formatter_class=argparse.RawDescriptionHelpFormatter, description=__doc__, ) parser.add_argument("task_id", help="UUID of the task to cancel.") parser.add_argument( "--run-id", required=True, dest="run_id", metavar="RUNID", help=( "Calling agent identifier. For claimed tasks, must match the " "original claimer unless --force is used. " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--force", action="store_true", default=False, help=( "Cancel even if the task is claimed by a different agent. " "Use with caution — aborts in-flight work without notifying " "the claimer." ), ) _add_format_args(parser) parser.set_defaults(func=run_cancel_task) def run_cancel_task(args: argparse.Namespace) -> None: """Cancel a pending or claimed task. Execution order --------------- 1. **Validate inputs** — ``task_id`` UUID syntax and ``--run-id`` length. All validation fires before any file I/O; any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1). 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`. 3. **Cancel task** — delegates to :func:`~muse.core.task_queue.cancel_task`. * **Pending task** — atomically creates a ``cancelled`` claim via ``O_CREAT | O_EXCL``. If another agent claims the task in the window between the check and the write, raises :exc:`FileExistsError`. * **Claimed task** — ownership check: ``--run-id`` must match the claimer unless ``--force`` is set. Atomically updates the claim record via :func:`~muse.core.store.write_text_atomic`. * **Terminal task** — raises :exc:`RuntimeError`; already completed/failed/cancelled tasks cannot be cancelled. 4. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Security -------- * ``task_id`` is validated as a UUID4 before ``require_repo()`` — prevents path traversal via a crafted task ID. * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters. * All display strings are passed through :func:`~muse.core.validation.sanitize_display` to strip ANSI injection. * ``--force`` bypasses claimer ownership; use only when the orchestrator needs to reclaim a task from an unresponsive agent. Performance ----------- O(1) — one or two file reads and one atomic write. Args: args: Parsed ``argparse.Namespace`` with attributes ``task_id``, ``run_id``, ``force``, and ``fmt``. Exit codes: 0 — task cancelled. 1 — bad arguments, task not found, already terminal, or permission denied. """ t0 = time.monotonic() as_json = args.fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if len(args.run_id) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) try: from muse.core.task_queue import _validate_task_id _validate_task_id(args.task_id) except ValueError as exc: _err(str(exc), as_json, "bad_task_id") raise SystemExit(ExitCode.USER_ERROR) root = require_repo() try: claim = cancel_task(root, args.task_id, args.run_id, force=args.force) except (FileNotFoundError, FileExistsError, PermissionError, RuntimeError, ValueError) as exc: _err(str(exc), as_json) raise SystemExit(ExitCode.USER_ERROR) elapsed = round(time.monotonic() - t0, 4) if as_json: out = claim.to_dict() out["elapsed_seconds"] = elapsed print(json.dumps(out)) return task = load_task(root, claim.task_id) print(f"\n🚫 Task cancelled") print(f" Task ID: {sanitize_display(claim.task_id)}") if task is not None: print(f" Title: {sanitize_display(task.title)}") print(f" Queue: {sanitize_display(task.queue)}") print(f" By: {sanitize_display(claim.claimer_run_id)}") if args.force: print(f" (forced)") print(f"\n ({elapsed:.3f}s)") # ── muse coord tasks ────────────────────────────────────────────────────────── def register_tasks( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register ``tasks`` on *subparsers* (under ``muse coord``). Wires all flags with their defaults and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run_tasks`. Flags registered ---------------- ``--status STATUS`` Filter items to a single status value. One of: ``pending``, ``claimed``, ``timed_out``, ``completed``, ``failed``, ``cancelled``. The global status counts in the output always reflect the *full* queue regardless of this filter. ``--queue QUEUE`` Filter items by queue name. Must match ``[a-zA-Z0-9_-]+``; validated before any file I/O. ``--run-id RUNID`` Filter items to tasks whose claimer matches *RUNID* (claimed, completed, and failed tasks only). Capped at :data:`_MAX_RUN_ID_LEN` characters. ``--limit N`` Maximum number of items to return (default: 200; max: :data:`_MAX_LIMIT`). The status counts always reflect the full queue; only the ``items`` list is truncated. ``--format`` / ``--json`` Emit compact JSON to stdout; default is human-readable text. Derived status values:: pending — not yet claimed claimed — actively claimed, claim TTL not expired timed_out — claim TTL expired (eligible for re-claiming) completed — done successfully failed — done with error cancelled — cancelled before or after claiming JSON output schema:: { "schema_version": str, "total": int, "pending": int, "claimed": int, "timed_out": int, "completed": int, "failed": int, "cancelled": int, "limit": int, "truncated": bool, "items": [ { "task_id": str, "title": str, "priority": int, "queue": str, "status": str, "created_by": str, "created_at": str, // ISO 8601 "ttl_seconds": int, "tags": [str, ...], "payload": dict, "claimer_run_id": str | null, "expires_at": str | null // ISO 8601; null if not claimed }, ... ], "elapsed_seconds": float } Exit codes:: 0 — success (empty queue is still success) 1 — bad arguments or unexpected error """ parser = subparsers.add_parser( "tasks", help="List tasks with optional status/queue/run-id/limit filtering.", formatter_class=argparse.RawDescriptionHelpFormatter, description=__doc__, ) parser.add_argument( "--status", default=None, choices=("pending", "claimed", "timed_out", "completed", "failed", "cancelled"), metavar="STATUS", help=( "Filter by status: pending | claimed | timed_out | " "completed | failed | cancelled" ), ) parser.add_argument( "--queue", default=None, metavar="QUEUE", help=( "Filter by queue name. Must match [a-zA-Z0-9_-]+. " "Validated before any file I/O." ), ) parser.add_argument( "--run-id", default=None, dest="run_id", metavar="RUNID", help=( "Filter by claimer run_id (claimed/completed/failed tasks only). " f"Maximum {_MAX_RUN_ID_LEN} characters." ), ) parser.add_argument( "--limit", type=int, default=200, metavar="N", help=( f"Maximum items to return (default: 200; max: {_MAX_LIMIT}). " "Status counts always reflect the full queue." ), ) _add_format_args(parser) parser.set_defaults(func=run_tasks) def run_tasks(args: argparse.Namespace) -> None: """List tasks from the coordination queue with filtering and pagination. Execution order --------------- 1. **Validate inputs** — ``--queue`` name syntax, ``--run-id`` length, ``--limit`` bounds (1–:data:`_MAX_LIMIT`). All validation fires before any file I/O; any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1). 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`. 3. **Load state** — one directory scan each for tasks and claims. 4. **Build items** — derive per-task status, apply filters, sort by priority desc / created_at asc, then truncate to ``--limit``. 5. **Compute counts** — iterate ``all_tasks`` once for global status counts (counts always reflect the *full* queue, never the filtered page). 6. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Security -------- * ``--queue`` is validated against ``[a-zA-Z0-9_-]+`` before I/O — prevents ANSI injection in text output via crafted queue names. * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` characters. * All strings sourced from persisted files are passed through :func:`~muse.core.validation.sanitize_display` before printing. Performance ----------- O(n) — one directory scan per kind (tasks, claims). The ``--limit`` flag prevents large outputs for agents that only need a page. For very large queues (10 000+ tasks), direct filesystem scanning is the bottleneck; MuseHub coordination will replace this for distributed deployments. Args: args: Parsed ``argparse.Namespace`` with attributes ``status``, ``queue``, ``run_id``, ``limit``, and ``fmt``. Exit codes: 0 — success (empty queue is still success). 1 — bad arguments or unexpected error. """ import datetime as _dt t0 = time.monotonic() as_json = args.fmt == "json" limit: int = getattr(args, "limit", 200) # ── Input validation (before any file I/O) ──────────────────────────────── if args.queue is not None: try: _validate_queue_name(args.queue) except ValueError as exc: _err(str(exc), as_json, "bad_queue") raise SystemExit(ExitCode.USER_ERROR) if args.run_id is not None and len(args.run_id) > _MAX_RUN_ID_LEN: msg = f"--run-id is too long ({len(args.run_id)} chars; max {_MAX_RUN_ID_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if not (1 <= limit <= _MAX_LIMIT): msg = f"--limit must be between 1 and {_MAX_LIMIT}, got {limit}" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) root = require_repo() all_tasks = load_all_tasks(root) all_claims = load_all_claims(root) now_ts = _dt.datetime.now(_dt.timezone.utc) # Build enriched items (filtered). items = [] for task in all_tasks: claim = all_claims.get(task.task_id) status = get_task_status(task, claim, now_ts) if args.status and status != args.status: continue if args.queue and task.queue != args.queue: continue if args.run_id: if claim is None or claim.claimer_run_id != args.run_id: continue items.append({ "task_id": task.task_id, "title": task.title, "priority": task.priority, "queue": task.queue, "status": status, "created_by": task.created_by, "created_at": task.created_at.isoformat(), "ttl_seconds": task.ttl_seconds, "tags": task.tags, "payload": task.payload, "claimer_run_id": claim.claimer_run_id if claim else None, "expires_at": claim.expires_at.isoformat() if claim else None, }) # Sort: priority desc, then created_at asc. items.sort(key=lambda i: (-i["priority"], i["created_at"])) # Apply limit after sort so the highest-priority items are always included. truncated = len(items) > limit items = items[:limit] elapsed = round(time.monotonic() - t0, 4) # Global status counts — always computed from the full all_tasks list, # independent of any filter, so the summary bar reflects queue health. counts: _IntMap = { s: 0 for s in ("pending", "claimed", "timed_out", "completed", "failed", "cancelled") } for task in all_tasks: claim = all_claims.get(task.task_id) s = get_task_status(task, claim, now_ts) counts[s] += 1 if as_json: print(json.dumps({ "schema_version": _schema_version(), "total": sum(counts.values()), **counts, "limit": limit, "truncated": truncated, "items": items, "elapsed_seconds": elapsed, })) return # Text output. filter_parts = [] if args.status: filter_parts.append(f"status={args.status}") if args.queue: filter_parts.append(f"queue={sanitize_display(args.queue)}") if args.run_id: filter_parts.append(f"run-id={sanitize_display(args.run_id)}") filter_str = f" filter: {', '.join(filter_parts)}" if filter_parts else "" print(f"\nTask queue — {sum(counts.values())} task(s)") if filter_str: print(filter_str) print( f" {counts['pending']} pending " f"{counts['claimed']} claimed " f"{counts['timed_out']} timed_out " f"{counts['completed']} completed " f"{counts['failed']} failed " f"{counts['cancelled']} cancelled" ) print("─" * 80) if not items: print("\n (no tasks matching filter)") else: print(f"\n{'ID':8} {'ST':10} {'PRI':3} {'QUEUE':12} {'CLAIMER':20} TITLE") print("─" * 80) _STATUS_ICONS = { "pending": "⏳", "claimed": "🔒", "timed_out": "⏰", "completed": "✅", "failed": "❌", "cancelled": "🚫", } for item in items: icon = _STATUS_ICONS.get(item["status"], "?") tid = sanitize_display(item["task_id"][:8]) st = f"{icon}{item['status'][:9]}" pri = str(item["priority"]) q = sanitize_display(item["queue"][:12]) claimer = sanitize_display((item["claimer_run_id"] or "-")[:20]) title = sanitize_display(item["title"][:40]) print(f"{tid} {st:<11} {pri:>3} {q:<12} {claimer:<20} {title}") if truncated: print(f"\n (showing {limit} of {len(items) + (len(all_tasks) - limit)} — use --limit to see more)") print(f"\n ({elapsed:.3f}s)") # ── Registration ────────────────────────────────────────────────────────────── def register_all( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register all task-queue subcommands on *subparsers*. Called from :func:`muse.cli.app.main` to attach the five task-queue commands to the ``muse coord`` subparser group. """ register_enqueue(subparsers) register_claim(subparsers) register_complete(subparsers) register_fail_task(subparsers) register_cancel_task(subparsers) register_tasks(subparsers) # ── Private helpers ─────────────────────────────────────────────────────────── def _schema_version() -> str: from muse._version import __version__ return __version__