"""muse agent — agent slot management and HD key derivation. Manages the agent slot registry (``~/.muse/agent-slots.toml``) and derives agent keypairs from the operator's BIP39 mnemonic. This is the command-line face of the Phase 2 HD agent identity system. Why a separate ``muse agent`` namespace? ----------------------------------------- ``muse auth`` owns the *human* identity lifecycle (keygen, register, whoami, logout). Agent slots are a distinct concept — they are derived sub-identities scoped to a specific SLIP-0010 account index within the operator's key tree. Keeping them separate prevents confusion between the human's key and the keys that agents use. Sub-seed injection ------------------ The output of ``muse agent keygen`` is a base64url-encoded 64-byte sub-seed suitable for injecting into an agent subprocess via the ``MUSE_AGENT_HD_SEED`` environment variable. The agent calls :func:`muse.core.hdkeys.derive_identity_key` on the sub-seed to reconstruct its signing key — no mnemonic is ever passed to the agent. Subcommands ----------- :: muse agent keygen --account N [--hub HUB] [--name NAME] [--json] muse agent list [--hub HUB] [--json] muse agent register --account N --name NAME [--hub HUB] [--json] JSON schemas ------------ ``muse agent keygen --json``:: { "status": "ok", "hub": "", "account": , "name": "", "msign_path": "m/1075233755'/0'/1'/'", # purpose'/domain_identity'/entity_agent'/account' "public_key_b64": "", "fingerprint": "", "hd_seed_b64": "" } ``muse agent list --json``:: [ { "name": "", "account": , "hub": "", "msign_path": "m/1075233755'/0'/1'/'" }, ... ] ``muse agent register --json``:: { "status": "ok", "name": "", "account": , "hub": "", "msign_path": "m/1075233755'/0'/1'/'" } Agent workflow examples ----------------------- :: # Derive and inspect agent slot 1 muse agent keygen --account 1 --json # Inject into a subprocess export MUSE_AGENT_HD_SEED=$(muse agent keygen --account 1 --json | python3 -c \ "import sys,json; print(json.load(sys.stdin)['hd_seed_b64'])") # Register the slot so muse agent list shows it muse agent register --account 1 --name "my-agent" # List all registered slots for the current hub muse agent list --json """ import argparse import json import logging import sys from typing import TypedDict from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.identity import hostname_from_url, load_identity from muse.core.timing import start_timer from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # TypedDicts # --------------------------------------------------------------------------- class _KeygenJson(EnvelopeJson): """JSON schema for ``muse agent keygen --json``.""" status: str # "ok" hub: str # hub URL used account: int # SLIP-0010 account index name: str | None # registered slot name (null if not registered) msign_path: str # derivation path public_key_b64: str # base64url-encoded 32-byte public key fingerprint: str # SHA-256 hex of the public key hd_seed_b64: str # base64url-encoded 64-byte sub-seed (MUSE_AGENT_HD_SEED value) class _RegisterJson(EnvelopeJson): """JSON schema for ``muse agent register --json``.""" status: str # "ok" name: str account: int hub: str # hostname msign_path: str class _ListJson(EnvelopeJson): """JSON envelope for ``muse agent list --json``.""" mode: str # always "agent-list" slots: list[dict] # list of registered agent slot records # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _derive_agent_seed(mnemonic: str, account: int) -> bytes: """Derive the 64-byte IDENTITY-domain agent sub-seed at *account*. Uses :func:`muse.core.bip39.mnemonic_to_seed` and :func:`muse.core.hdkeys.derive_agent_sub_seed` with ``domain=DOMAIN_IDENTITY``. Args: mnemonic: BIP39 mnemonic phrase (space-separated words). account: Agent account index (>= 0). Returns: 64-byte agent sub-seed suitable for ``MUSE_AGENT_HD_SEED`` injection. Raises: SystemExit(1): If HD derivation libraries are not available. """ try: from muse.core.bip39 import mnemonic_to_seed from muse.core.hdkeys import DOMAIN_IDENTITY, derive_agent_sub_seed except ImportError as exc: print( f"muse agent: HD key derivation not available — {exc}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) seed = mnemonic_to_seed(mnemonic) return derive_agent_sub_seed(seed, domain=DOMAIN_IDENTITY, agent_id=account) def _sub_seed_to_public(sub_seed: bytes) -> bytes: """Derive the raw 32-byte Ed25519 public key from a 64-byte sub-seed. Args: sub_seed: 64-byte agent sub-seed from :func:`_derive_agent_seed`. Returns: 32-byte raw Ed25519 public key. """ from muse.core.hdkeys import derive_identity_key, dk_to_ed25519 dk = derive_identity_key(sub_seed) private_key = dk_to_ed25519(dk) dk.zero() return private_key.public_key().public_bytes_raw() def _emit_error(error: str, message: str, as_json: bool) -> None: """Emit an error response and exit 1. When *as_json* is ``True`` the error is printed as a JSON object on stdout so agent consumers can parse it. Otherwise the message goes to stderr as plain text. Args: error: Machine-readable error code string (e.g. ``"no_identity"``). message: Human-readable description. as_json: When ``True``, emit ``{"error": ..., "message": ...}`` on stdout. """ if as_json: print(json.dumps({"error": error, "message": message})) else: print(f"muse agent: {message}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) def _require_mnemonic(hub_url: str, *, as_json: bool = False) -> str: """Load the BIP39 mnemonic for *hub_url* from ``~/.muse/identity.toml``. Args: hub_url: Hub URL or bare hostname. as_json: Emit JSON error on stdout instead of plain-text stderr. Returns: Mnemonic phrase string. Raises: SystemExit(1): If no identity or no mnemonic is found. """ hostname = hostname_from_url(hub_url) entry = load_identity(hub_url) if entry is None: _emit_error( "no_identity", f"no identity found for {sanitize_display(hostname)} — " "run `muse auth keygen --hub ` first.", as_json, ) mnemonic = entry.get("mnemonic", "").strip() # type: ignore[union-attr] if not mnemonic: _emit_error( "no_mnemonic", f"identity for {sanitize_display(hostname)} has no mnemonic — " "HD key derivation requires a BIP39 mnemonic; " "re-generate with `muse auth keygen --hub `.", as_json, ) return mnemonic # type: ignore[return-value] def _resolve_hub_url(args_hub: str | None, *, as_json: bool = False) -> str: """Resolve the hub URL from CLI flag or repo config. Args: args_hub: Value of ``--hub`` flag, or ``None``. as_json: Emit JSON error on stdout instead of plain-text stderr. Returns: Hub URL string. Raises: SystemExit(1): If no hub can be determined. """ if args_hub: return args_hub from muse.cli.config import get_hub_url url = get_hub_url() if url: return url _emit_error( "no_hub", "no hub configured — pass --hub or connect with " "`muse hub connect `.", as_json, ) # --------------------------------------------------------------------------- # Command handlers # --------------------------------------------------------------------------- def run_keygen(args: argparse.Namespace) -> None: """Derive and display the agent keypair at the requested account index. Reads the BIP39 mnemonic from ``~/.muse/identity.toml``, derives the IDENTITY-domain agent sub-seed at *account*, and prints the sub-seed (as ``MUSE_AGENT_HD_SEED``), the public key fingerprint, and the SLIP-0010 path. Agent quickstart ---------------- :: muse agent keygen --account 1 --json # inject into subprocess: export MUSE_AGENT_HD_SEED=$(muse agent keygen --account 1 --json | python3 -c \ "import sys,json; print(json.load(sys.stdin)['hd_seed_b64'])") JSON fields ----------- schema Envelope schema integer (``1``). duration_ms Wall-clock time from argument parsing to output. exit_code Mirrors the process exit code (``0`` on success). status ``"ok"`` on success. hub Hub URL used. account SLIP-0010 account index. name Registered slot name or ``null``. msign_path SLIP-0010 derivation path string. public_key_b64 Base64url-encoded 32-byte Ed25519 public key. fingerprint SHA-256 hex fingerprint of the public key. hd_seed_b64 Base64url-encoded 64-byte sub-seed — set as ``MUSE_AGENT_HD_SEED``. Exit codes ---------- 0 Keypair derived successfully. 1 No identity/mnemonic found, invalid account, or derivation error. """ elapsed = start_timer() json_out: bool = args.json_out hub_url = _resolve_hub_url(getattr(args, "hub", None), as_json=json_out) account: int = args.account name: str | None = getattr(args, "name", None) or None if account < 0: _emit_error( "invalid_account", f"account must be >= 0; got {account}", json_out, ) mnemonic = _require_mnemonic(hub_url, as_json=json_out) try: sub_seed = _derive_agent_seed(mnemonic, account) pub_bytes = _sub_seed_to_public(sub_seed) except Exception as exc: _emit_error("derivation_failed", f"key derivation failed — {exc}", json_out) from muse.core.hdkeys import DOMAIN_IDENTITY, ENTITY_AGENT, MUSE_PURPOSE msign_path = f"m/{MUSE_PURPOSE}'/{DOMAIN_IDENTITY}'/{ENTITY_AGENT}'/{account}'" from muse.core.types import b64url_encode, public_key_fingerprint # noqa: PLC0415 hd_seed_b64 = b64url_encode(sub_seed) pub_b64 = b64url_encode(pub_bytes) fp = public_key_fingerprint(pub_bytes) if json_out: payload = _KeygenJson( **make_envelope(elapsed), status="ok", hub=hub_url, account=account, name=name, msign_path=msign_path, public_key_b64=pub_b64, fingerprint=fp, hd_seed_b64=hd_seed_b64, ) print(json.dumps(payload)) else: hostname = hostname_from_url(hub_url) print(f"Agent keypair — account {account} on {sanitize_display(hostname)}") print(f" SLIP-0010 path : {msign_path}") print(f" Fingerprint : {fp}") print(f" Public key : {pub_b64}") if name: print(f" Name : {name}") print() print(f" MUSE_AGENT_HD_SEED={hd_seed_b64}") print() print(" Set this env var before starting the agent process.") def run_list(args: argparse.Namespace) -> None: """List all registered agent slots for the hub. Reads from ``~/.muse/agent-slots.toml`` and prints one row per slot. Agent quickstart ---------------- :: muse agent list --json muse agent list --json | jq '.[].name' JSON output ----------- Emits a JSON envelope object with a standard header and a ``slots`` list. schema Envelope schema integer (``1``). duration_ms Wall-clock time from argument parsing to output. exit_code Mirrors the process exit code (always ``0``). mode Always ``"agent-list"``. slots List of registered slot objects, each with: name Slot name, or ``null`` if unnamed. account HD account index. hub Hostname of the hub (not the full URL). msign_path HD derivation path for this slot. Exit codes ---------- 0 Always (empty list is not an error). """ elapsed = start_timer() json_out: bool = args.json_out hub_url = _resolve_hub_url(getattr(args, "hub", None), as_json=json_out) from muse.core.agent_slots import list_slots slots = list_slots(hub_url) if json_out: print(json.dumps(_ListJson(**make_envelope(elapsed), mode="agent-list", slots=slots))) else: hostname = hostname_from_url(hub_url) if not slots: print( f"No registered agent slots for {sanitize_display(hostname)}. " "Register one with `muse agent register --account N --name NAME`." ) return print(f"Agent slots for {sanitize_display(hostname)}:") for slot in slots: name_part = f" {slot['name']:<20} account={slot['account']:<4} {slot['msign_path']}" print(name_part) def run_register(args: argparse.Namespace) -> None: """Register a named agent slot in the local slot registry. Does not derive any keys — purely records the (name, account) binding in ``~/.muse/agent-slots.toml`` so that ``muse agent list`` can display it. Agent quickstart ---------------- :: muse agent register --account 1 --name worker --json # → {"status": "ok", "name": "worker", "account": 1, ...} JSON fields ----------- schema Envelope schema integer (``1``). duration_ms Wall-clock time from argument parsing to output. exit_code Mirrors the process exit code (``0`` on success). status ``"ok"`` on success. name Registered slot name. account SLIP-0010 account index. hub Hub hostname. msign_path SLIP-0010 derivation path string. Exit codes ---------- 0 Slot registered. 1 Invalid arguments. """ elapsed = start_timer() json_out: bool = args.json_out hub_url = _resolve_hub_url(getattr(args, "hub", None), as_json=json_out) account: int = args.account name: str = args.name if account < 0: _emit_error( "invalid_account", f"account must be >= 0; got {account}", json_out, ) from muse.core.agent_slots import register_slot slot = register_slot(hub_url, name, account) if json_out: payload = _RegisterJson( **make_envelope(elapsed), status="ok", name=slot["name"], account=slot["account"], hub=slot["hub"], msign_path=slot["msign_path"], ) print(json.dumps(payload)) else: hostname = slot["hub"] print( f"Registered agent slot '{name}' → account {account} " f"on {sanitize_display(hostname)}" ) print(f" SLIP-0010 path: {slot['msign_path']}") # --------------------------------------------------------------------------- # Argument parser registration # --------------------------------------------------------------------------- def register(subparsers: argparse._SubParsersAction) -> None: # type: ignore[type-arg] """Register the ``muse agent`` subcommand tree. Subcommands: ``keygen``, ``list``, ``register``. All subcommands accept ``--json`` / ``-j`` for machine-readable JSON output with a standard envelope: ``schema_version``, ``exit_code``, ``duration_ms``. """ agent_parser = subparsers.add_parser( "agent", help="Agent slot management and HD key derivation.", description=( "Manage agent slots and derive agent keypairs from the operator's " "BIP39 mnemonic stored in ~/.muse/identity.toml." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) agent_subs = agent_parser.add_subparsers( dest="agent_command", metavar="AGENT_COMMAND" ) agent_subs.required = True # ── keygen ────────────────────────────────────────────────────────────── keygen_p = agent_subs.add_parser( "keygen", help="Derive agent keypair at account index N.", description=( "Read the BIP39 mnemonic from ~/.muse/identity.toml and derive " "the IDENTITY-domain agent sub-seed at the given account index. " "The hd_seed_b64 output value should be set as MUSE_AGENT_HD_SEED " "in the agent's environment." ), ) keygen_p.add_argument( "--account", type=int, required=True, metavar="N", help="SLIP-0010 account index for this agent (>= 0; 0 is service identity).", ) keygen_p.add_argument( "--hub", metavar="URL", default=None, help="Hub URL to look up the mnemonic for (defaults to repo config).", ) keygen_p.add_argument( "--name", metavar="NAME", default=None, help="Optional slot name to include in output (does not register the slot).", ) keygen_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit JSON on stdout.", ) keygen_p.set_defaults(func=run_keygen) # ── list ──────────────────────────────────────────────────────────────── list_p = agent_subs.add_parser( "list", help="List all registered agent slots.", description=( "Read ~/.muse/agent-slots.toml and display all named agent slots " "registered for the given hub." ), ) list_p.add_argument( "--hub", metavar="URL", default=None, help="Hub URL to filter by (defaults to repo config).", ) list_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit JSON on stdout.", ) list_p.set_defaults(func=run_list) # ── register ───────────────────────────────────────────────────────────── register_p = agent_subs.add_parser( "register", help="Register a named agent slot in the local registry.", description=( "Record a (name, account) binding in ~/.muse/agent-slots.toml " "so that `muse agent list` can display it. Does not derive keys." ), ) register_p.add_argument( "--account", type=int, required=True, metavar="N", help="SLIP-0010 account index for this agent (>= 0).", ) register_p.add_argument( "--name", required=True, metavar="NAME", help="Human-readable slot label, e.g. 'orchestra'.", ) register_p.add_argument( "--hub", metavar="URL", default=None, help="Hub URL (defaults to repo config).", ) register_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit JSON on stdout.", ) register_p.set_defaults(func=run_register)