vacant.cli

Console-script entrypoint declared in pyproject.toml as vacant = "vacant.cli:main". Each subcommand maps to a function in vacant.cli.commands; local key / logbook persistence lives in vacant.cli.local_store.

commands

Wired CLI commands.

Each command in this module replaces the _NOT_YET stub that shipped with P0. The local-state store at ~/.vacant/<name>/ is owned by vacant.cli.local_store; HTTP work goes through httpx.AsyncClient against a registry URL (env VACANT_REGISTRY_URL or --registry).

A few commands (call, attest) require remote endpoints that ship with PR-β (vacant serve + the wired-up /v1/submit_attestation). Those subcommands degrade gracefully with a clear not available yet exit code so the help surface is complete and a future PR can enable them in place. F4 acceptance only requires the commands to run end-to-end — the remote-only features have explicit pending tickets.

init_cmd

init_cmd(name: str, insecure_demo: bool = Option(False, '--insecure-demo', help='Store the Ed25519 seed in plaintext key.json (mode 0600) instead of the OS keyring. Demo / CI use only — see SECURITY.md.')) -> None

Create a fresh keypair + seed logbook for name. (P2)

Writes ~/.vacant/<name>/{key.json,logbook.jsonl,meta.json} with file mode 0600 on the key. The Ed25519 seed is stored in the OS keyring by default (Keychain / Secret Service / Credential Locker); pass --insecure-demo to fall back to plaintext on hosts without a keyring backend.

Source code in src/vacant/cli/commands.py

@app.command("init")
def init_cmd(
    name: str,
    insecure_demo: bool = typer.Option(
        False,
        "--insecure-demo",
        help=(
            "Store the Ed25519 seed in plaintext key.json (mode 0600) "
            "instead of the OS keyring. Demo / CI use only — see SECURITY.md."
        ),
    ),
) -> None:
    """Create a fresh keypair + seed logbook for `name`. (P2)

    Writes `~/.vacant/<name>/{key.json,logbook.jsonl,meta.json}` with
    file mode 0600 on the key. The Ed25519 seed is stored in the OS
    keyring by default (Keychain / Secret Service / Credential
    Locker); pass `--insecure-demo` to fall back to plaintext on
    hosts without a keyring backend.
    """
    try:
        vid, _sk = ls.init_vacant(name, insecure_demo=insecure_demo)
    except ls.LocalVacantExists:
        typer.echo(f"error: local vacant {name!r} already exists", err=True)
        raise typer.Exit(code=1) from None
    except ls.LocalVacantKeyringUnavailable as exc:
        typer.echo(f"error: {exc}", err=True)
        raise typer.Exit(code=1) from exc
    except LocalVacantError as exc:
        typer.echo(f"error: {exc}", err=True)
        raise typer.Exit(code=1) from exc
    typer.echo(json.dumps({"name": name, "vacant_id": vid.hex()}, sort_keys=True))

status_cmd

status_cmd(all_: bool = Option(False, '--all', help='Include hibernating/stale/sunk.')) -> None

Show local vacants and their lifecycle states. (P1)

Source code in src/vacant/cli/commands.py

@app.command("status")
def status_cmd(
    all_: bool = typer.Option(False, "--all", help="Include hibernating/stale/sunk."),
) -> None:
    """Show local vacants and their lifecycle states. (P1)"""
    rows: list[dict[str, Any]] = []
    for n in ls.list_vacant_names():
        try:
            meta = ls.load_meta(n)
        except LocalVacantNotFound:
            continue
        if not all_ and meta.state in {"HIBERNATING", "STALE", "SUNK", "ARCHIVED"}:
            continue
        rows.append(
            {
                "name": n,
                "vacant_id": meta.vacant_id_hex,
                "state": meta.state,
                "capability_text": meta.capability_text,
                "endpoint": meta.endpoint,
                "halo_published": meta.halo_published,
                "last_heartbeat_at": meta.last_heartbeat_at,
            }
        )
    typer.echo(json.dumps({"vacants": rows}, sort_keys=True, indent=2))

heartbeat_cmd

heartbeat_cmd(name: str | None = Option(None, '--name', help='Local vacant name; defaults to VACANT_NAME.')) -> None

Manually trigger a heartbeat tick. (P1)

Source code in src/vacant/cli/commands.py

@app.command("heartbeat")
def heartbeat_cmd(
    name: str | None = typer.Option(
        None, "--name", help="Local vacant name; defaults to VACANT_NAME."
    ),
) -> None:
    """Manually trigger a heartbeat tick. (P1)"""
    n = _resolve_name(name)
    try:
        meta = ls.load_meta(n)
        sk = ls.load_signing_key(n)
        lb = ls.load_logbook(n)
    except LocalVacantNotFound as exc:
        typer.echo(f"error: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    state = VacantState(meta.state)
    if state == VacantState.ARCHIVED:
        typer.echo("error: ARCHIVED vacants do not heartbeat", err=True)
        raise typer.Exit(code=2)
    payload = heartbeat_payload(state)
    kind = heartbeat_kind(state)
    entry = lb.append(kind, payload, sk)
    ls.save_logbook(n, lb)
    meta.last_heartbeat_at = _now_iso()
    ls.save_meta(n, meta)
    typer.echo(
        json.dumps(
            {
                "name": n,
                "kind": kind,
                "state": state.value,
                "ts": entry.ts.isoformat(),
                "logbook_entries": len(lb.entries),
            },
            sort_keys=True,
        )
    )

publish_cmd

publish_cmd(capability: str = Option(..., '--capability', help='Capability text to advertise.'), endpoint: str | None = Option(None, '--endpoint', help='A2A endpoint URL.'), registry: str | None = Option(None, '--registry', help='Registry URL.'), name: str | None = Option(None, '--name', help='Local vacant name.'), base_model: str | None = Option(None, '--base-model', help="Base model identifier (e.g. 'claude-sonnet-4-6'). On the first publish, omitted → defaults to 'unknown'. On a republish, omitted → preserves the stored value (Pfix3 F2)."), base_model_family: str | None = Option(None, '--base-model-family', help="Base model family (e.g. 'claude'). Same null-vs-default semantics as --base-model.")) -> None

Flip LOCAL → ACTIVE (publish halo to registry). (P4)

Source code in src/vacant/cli/commands.py

@app.command("publish")
def publish_cmd(
    capability: str = typer.Option(..., "--capability", help="Capability text to advertise."),
    endpoint: str | None = typer.Option(None, "--endpoint", help="A2A endpoint URL."),
    registry: str | None = typer.Option(None, "--registry", help="Registry URL."),
    name: str | None = typer.Option(None, "--name", help="Local vacant name."),
    base_model: str | None = typer.Option(
        None,
        "--base-model",
        help=(
            "Base model identifier (e.g. 'claude-sonnet-4-6'). On the "
            "first publish, omitted → defaults to 'unknown'. On a "
            "republish, omitted → preserves the stored value (Pfix3 F2)."
        ),
    ),
    base_model_family: str | None = typer.Option(
        None,
        "--base-model-family",
        help=("Base model family (e.g. 'claude'). Same null-vs-default semantics as --base-model."),
    ),
) -> None:
    """Flip LOCAL → ACTIVE (publish halo to registry). (P4)"""
    n = _resolve_name(name)
    url = _resolve_registry(registry)
    try:
        result = asyncio.run(
            _do_publish(
                name=n,
                registry_url=url,
                capability_text=capability,
                endpoint=endpoint,
                base_model=base_model,
                base_model_family=base_model_family,
            )
        )
    except Exception as exc:
        typer.echo(f"error: publish failed: {exc}", err=True)
        raise typer.Exit(code=1) from exc
    typer.echo(json.dumps(result, sort_keys=True))

unpublish_cmd

unpublish_cmd(name: str | None = Option(None, '--name', help='Local vacant name.')) -> None

Flip ACTIVE → LOCAL (visibility=NONE). (P4)

Note: this only flips the local meta; the registry record is not revoked over HTTP yet (the /v1/revoke_halo endpoint requires a P6 envelope, see rpc.py). Use the python vacant.registry.halo.revoke_halo API for full withdrawal.

Source code in src/vacant/cli/commands.py

@app.command("unpublish")
def unpublish_cmd(
    name: str | None = typer.Option(None, "--name", help="Local vacant name."),
) -> None:
    """Flip ACTIVE → LOCAL (visibility=NONE). (P4)

    Note: this only flips the local meta; the registry record is
    not revoked over HTTP yet (the `/v1/revoke_halo` endpoint
    requires a P6 envelope, see ``rpc.py``). Use the python
    `vacant.registry.halo.revoke_halo` API for full withdrawal.
    """
    n = _resolve_name(name)
    try:
        meta = ls.load_meta(n)
    except LocalVacantNotFound as exc:
        typer.echo(f"error: {exc}", err=True)
        raise typer.Exit(code=2) from exc
    meta.state = "LOCAL"
    meta.halo_published = False
    ls.save_meta(n, meta)
    typer.echo(
        json.dumps(
            {
                "name": n,
                "state": "LOCAL",
                "warning": (
                    "registry halo not actively revoked over HTTP yet; "
                    "see vacant.registry.halo.revoke_halo()"
                ),
            },
            sort_keys=True,
        )
    )

lineage_cmd

lineage_cmd(vid: str, direction: str = Option('ancestors', '--direction', help='ancestors | descendants'), depth: int = Option(8, '--depth', min=1, max=32), registry: str | None = Option(None, '--registry', help='Registry URL.')) -> None

Print the parent_id chain for vid. (P4)

Source code in src/vacant/cli/commands.py

@app.command("lineage")
def lineage_cmd(
    vid: str,
    direction: str = typer.Option("ancestors", "--direction", help="ancestors | descendants"),
    depth: int = typer.Option(8, "--depth", min=1, max=32),
    registry: str | None = typer.Option(None, "--registry", help="Registry URL."),
) -> None:
    """Print the parent_id chain for `vid`. (P4)"""
    if direction not in {"ancestors", "descendants"}:
        typer.echo("error: --direction must be 'ancestors' or 'descendants'", err=True)
        raise typer.Exit(code=2)
    url = _resolve_registry(registry)

    async def _go() -> dict[str, Any]:
        import httpx

        async with httpx.AsyncClient(timeout=15.0) as http:
            r = await http.get(
                f"{url}/v1/lineage/{vid}",
                params={"direction": direction, "depth": depth},
            )
            r.raise_for_status()
            data: dict[str, Any] = r.json()
            return data

    try:
        out = asyncio.run(_go())
    except Exception as exc:
        typer.echo(f"error: lineage lookup failed: {exc}", err=True)
        raise typer.Exit(code=1) from exc
    typer.echo(json.dumps(out, sort_keys=True))

attest_cmd

attest_cmd(target_vid: str, claim: str, name: str | None = Option(None, '--name', help='Local vacant name.')) -> None

Issue a peer attestation about target_vid. (P2)

Signs a PeerAttestation and stores it in ~/.vacant/<name>/attestations_issued.jsonl. The HTTP relay to the registry's /v1/submit_attestation endpoint lands in PR-β (the endpoint is currently a P6-envelope stub).

Source code in src/vacant/cli/commands.py

@app.command("attest")
def attest_cmd(
    target_vid: str,
    claim: str,
    name: str | None = typer.Option(None, "--name", help="Local vacant name."),
) -> None:
    """Issue a peer attestation about `target_vid`. (P2)

    Signs a `PeerAttestation` and stores it in
    `~/.vacant/<name>/attestations_issued.jsonl`. The HTTP relay to
    the registry's `/v1/submit_attestation` endpoint lands in PR-β
    (the endpoint is currently a P6-envelope stub).
    """
    n = _resolve_name(name)
    try:
        meta = ls.load_meta(n)
        sk = ls.load_signing_key(n)
    except LocalVacantNotFound as exc:
        typer.echo(f"error: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    try:
        attester = VacantId(pubkey_bytes=bytes.fromhex(meta.vacant_id_hex))
        attestee = VacantId(pubkey_bytes=bytes.fromhex(target_vid))
    except ValueError as exc:
        typer.echo(f"error: invalid vacant_id hex: {exc}", err=True)
        raise typer.Exit(code=2) from exc

    att = issue_attestation(
        attester=attester, attestee=attestee, claim=claim, attester_signing_key=sk
    )
    record = {
        "attester": att.attester.hex(),
        "attestee": att.attestee.hex(),
        "claim": att.claim,
        "issued_at": att.issued_at.isoformat(),
        "expires_at": att.expires_at.isoformat(),
        "signature_hex": att.signature.hex(),
    }
    out_path = ls.vacant_dir(n) / "attestations_issued.jsonl"
    with out_path.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, sort_keys=True) + "\n")
    typer.echo(json.dumps(record, sort_keys=True))

call_cmd

call_cmd(vid: str, capability: str, text: str = Option('ping', '--text', help='Body text to send.'), registry: str | None = Option(None, '--registry', help='Registry URL.'), name: str | None = Option(None, '--name', help='Local vacant name.')) -> None

Send a request to a remote vacant. (P6)

Looks up the target's CapabilityCard via the registry's /v1/capability_card/<vid> endpoint and dispatches a signed envelope to card.endpoint. The --endpoint direct-known mode lands with PR-β alongside vacant serve's /card route.

Source code in src/vacant/cli/commands.py

@app.command("call")
def call_cmd(
    vid: str,
    capability: str,
    text: str = typer.Option("ping", "--text", help="Body text to send."),
    registry: str | None = typer.Option(None, "--registry", help="Registry URL."),
    name: str | None = typer.Option(None, "--name", help="Local vacant name."),
) -> None:
    """Send a request to a remote vacant. (P6)

    Looks up the target's `CapabilityCard` via the registry's
    `/v1/capability_card/<vid>` endpoint and dispatches a signed
    envelope to `card.endpoint`. The `--endpoint` direct-known mode
    lands with PR-β alongside `vacant serve`'s `/card` route.
    """
    n = _resolve_name(name)
    url = _resolve_registry(registry)
    _ = capability  # capability filter is informational for now; lookup is by vid

    async def _go() -> dict[str, Any]:
        import httpx

        from vacant.core.types import EMPTY_PREV_HASH
        from vacant.protocol.capability_card import deserialize as deserialize_card
        from vacant.protocol.dispatch import call_local, make_httpx_transport
        from vacant.protocol.envelope import A2AMessage, A2APart
        from vacant.protocol.replay_protect import (
            InMemoryReplayStore,
            PairKey,
            ReplayState,
        )

        sk = ls.load_signing_key(n)
        form = _residentform_for(n)
        async with httpx.AsyncClient(timeout=15.0) as http:
            r = await http.get(
                f"{url}/v1/capability_card/{vid}",
                params={"caller": form.identity.hex()},
            )
            r.raise_for_status()
            row = r.json()
        blob_hex = row.get("capability_card_blob_hex", "")
        if not blob_hex:
            raise RuntimeError(
                f"registry returned no signed card blob for {vid}; "
                "the row pre-dates the capability_card_blob column"
            )
        target_card = deserialize_card(bytes.fromhex(blob_hex))
        transport = make_httpx_transport(timeout=30.0)

        # Pfix3 B6: continue the per-pair envelope chain from disk.
        # Without this, every CLI call defaulted to seq=1 / EMPTY prev,
        # and the second call to the same target was rejected as
        # replay by the server. Layout in envelope_state.json:
        #   {"<target_hex>": {"request": {...}, "response": {...}}}
        env_state = ls.load_envelope_state(n)
        target_hex = target_card.vacant_id.hex()
        target_state = env_state.get(target_hex, {})
        req_state = target_state.get("request", {})
        last_req_seq = int(req_state.get("last_seq", 0))
        last_req_hash_hex = str(req_state.get("last_hash_hex", ""))
        last_req_hash = bytes.fromhex(last_req_hash_hex) if last_req_hash_hex else EMPTY_PREV_HASH

        # Caller-side response replay store, seeded so the first
        # response on a pair starts at seq=1 / EMPTY prev (matching
        # `make_response_envelope` on the server side).
        rsp_state = target_state.get("response", {})
        last_rsp_seq = int(rsp_state.get("last_seq", 0))
        last_rsp_hash_hex = str(rsp_state.get("last_hash_hex", ""))
        last_rsp_hash = bytes.fromhex(last_rsp_hash_hex) if last_rsp_hash_hex else EMPTY_PREV_HASH
        caller_rsp_store = InMemoryReplayStore()
        if last_rsp_seq > 0:
            inverse_key = PairKey(from_vid=target_card.vacant_id, to_vid=form.identity)
            caller_rsp_store.seed(
                inverse_key,
                ReplayState(last_sequence_no=last_rsp_seq, chain_tip=last_rsp_hash),
            )

        result = await call_local(
            target_card=target_card,
            requester=form,
            requester_signing_key=sk,
            payload=A2AMessage(role="ROLE_USER", parts=[A2APart(text=text)]),
            transport=transport,
            sequence_no=last_req_seq + 1,
            prev_envelope_hash=last_req_hash,
            caller_response_replay_store=caller_rsp_store,
        )

        # Persist the new chain tips so the next call advances.
        env_state[target_hex] = {
            "request": {
                "last_seq": result.request_envelope.sequence_no,
                "last_hash_hex": result.request_envelope.compute_hash().hex(),
            },
            "response": {
                "last_seq": result.response_envelope.sequence_no,
                "last_hash_hex": result.response_envelope.compute_hash().hex(),
            },
        }
        ls.save_envelope_state(n, env_state)

        return {
            "target": target_hex,
            "endpoint": target_card.endpoint,
            "request_seq": result.request_envelope.sequence_no,
            "response_role": result.response_envelope.payload.role,
            "response_text": "".join(p.text for p in result.response_envelope.payload.parts),
        }

    try:
        out = asyncio.run(_go())
    except Exception as exc:
        typer.echo(f"error: call failed: {exc}", err=True)
        raise typer.Exit(code=1) from exc
    typer.echo(json.dumps(out, sort_keys=True))

serve_cmd

serve_cmd(port: int = Option(8443, '--port', '-p', help='HTTP bind port.'), host: str = Option('127.0.0.1', '--host', help='HTTP bind host.'), name: str | None = Option(None, '--name', help='Local vacant name.'), mcp: bool = Option(False, '--mcp', help='Also expose an MCP stdio server.'), endpoint: str | None = Option(None, '--endpoint', help='Public endpoint URL to advertise in /card (defaults to meta.endpoint).')) -> None

Start an HTTP A2A server for the local vacant. (P6)

The server listens on host:port and accepts inbound A2A message/send requests at /a2a/message/send. The default behaviour callback echoes the request text back, signed by the vacant's own key — sufficient for the live-network acceptance test.

--mcp additionally launches an MCP stdio server in a worker thread. This is what closes the "嫁接到客戶端" thesis claim: the same vacant accepts both A2A HTTP and MCP stdio simultaneously.

Source code in src/vacant/cli/commands.py

@app.command("serve")
def serve_cmd(
    port: int = typer.Option(8443, "--port", "-p", help="HTTP bind port."),
    host: str = typer.Option("127.0.0.1", "--host", help="HTTP bind host."),
    name: str | None = typer.Option(None, "--name", help="Local vacant name."),
    mcp: bool = typer.Option(False, "--mcp", help="Also expose an MCP stdio server."),
    endpoint: str | None = typer.Option(
        None,
        "--endpoint",
        help="Public endpoint URL to advertise in /card (defaults to meta.endpoint).",
    ),
) -> None:
    """Start an HTTP A2A server for the local vacant. (P6)

    The server listens on `host:port` and accepts inbound A2A
    `message/send` requests at `/a2a/message/send`. The default
    behaviour callback echoes the request text back, signed by the
    vacant's own key — sufficient for the live-network acceptance test.

    `--mcp` additionally launches an MCP stdio server in a worker
    thread. This is what closes the "嫁接到客戶端" thesis claim: the
    same vacant accepts both A2A HTTP and MCP stdio simultaneously.
    """
    import uvicorn

    from vacant.cli.server import build_serve_app

    n = _resolve_name(name)
    bundle = build_serve_app(n, endpoint=endpoint)

    if mcp:
        # Lazy import — only paid for when --mcp is set.
        import threading

        from vacant.cli.mcp_server import run_mcp_stdio_server

        t = threading.Thread(
            target=run_mcp_stdio_server,
            kwargs={
                "form": bundle.form,
                "signing_key": bundle.signing_key,
                "replay_store": bundle.replay_store,
            },
            daemon=True,
            name="vacant-mcp-stdio",
        )
        t.start()

    typer.echo(
        json.dumps(
            {
                "name": n,
                "vacant_id": bundle.form.identity.hex(),
                "host": host,
                "port": port,
                "mcp": mcp,
            },
            sort_keys=True,
        )
    )
    uvicorn.run(bundle.app, host=host, port=port, log_level="warning")

mcp_cmd

mcp_cmd(name: str | None = Option(None, '--name', help='Local vacant name to serve. Defaults to `$VACANT_NAME` or the single existing local vacant. If no local vacant exists, an ephemeral demo vacant is used (with a stderr WARN).')) -> None

Run the vacant as a pure-stdio MCP server. (D2 / Claude Code plugin)

No HTTP, no worker threads, no uvicorn — the process IS the MCP server. Spawned by uvx vacant mcp from the .claude-plugin/plugin.json manifest, which is what Claude Code calls when a user runs /plugin install vacant. EOF on stdin (the parent closing the pipe) ends the loop.

Identity resolution:

1. --name <n> ⇒ load ~/.vacant/<n>/
2. otherwise $VACANT_NAME ⇒ same
3. otherwise the only initialised local vacant ⇒ same
4. nothing initialised ⇒ ephemeral in-memory demo vacant + a stderr WARN telling the operator to run vacant init for a persistent identity.

Source code in src/vacant/cli/commands.py

@app.command("mcp")
def mcp_cmd(
    name: str | None = typer.Option(
        None,
        "--name",
        help=(
            "Local vacant name to serve. Defaults to `$VACANT_NAME` or the "
            "single existing local vacant. If no local vacant exists, an "
            "ephemeral demo vacant is used (with a stderr WARN)."
        ),
    ),
) -> None:
    """Run the vacant as a pure-stdio MCP server. (D2 / Claude Code plugin)

    No HTTP, no worker threads, no `uvicorn` — the process IS the
    MCP server. Spawned by `uvx vacant mcp` from the
    `.claude-plugin/plugin.json` manifest, which is what Claude Code
    calls when a user runs `/plugin install vacant`. EOF on stdin
    (the parent closing the pipe) ends the loop.

    Identity resolution:

    1. `--name <n>` ⇒ load `~/.vacant/<n>/`
    2. otherwise `$VACANT_NAME` ⇒ same
    3. otherwise the only initialised local vacant ⇒ same
    4. nothing initialised ⇒ ephemeral in-memory demo vacant + a
       stderr WARN telling the operator to run `vacant init` for a
       persistent identity.
    """
    from vacant.cli.mcp_server import run_mcp_stdio_server
    from vacant.cli.server import build_serve_app

    persistent_name: str | None = None
    if name is not None:
        bundle = build_serve_app(name)
        form = bundle.form
        signing_key = bundle.signing_key
        replay_store = bundle.replay_store
        persistent_name = name
    else:
        try:
            n = ls.current_name()
        except LocalVacantNotFound:
            sys.stderr.write(
                "WARN: no local vacant on disk; running an EPHEMERAL demo "
                "vacant. The keypair is fresh-per-launch and never persisted. "
                "Run `vacant init <name>` for a stable identity that survives "
                "process restarts. See SECURITY.md §Local key storage.\n"
            )
            from vacant.protocol import InMemoryReplayStore

            form, signing_key = _build_ephemeral_form()
            replay_store = InMemoryReplayStore()
        else:
            bundle = build_serve_app(n)
            form = bundle.form
            signing_key = bundle.signing_key
            replay_store = bundle.replay_store
            persistent_name = n

    # Pfix3 B7: persist signed SUBSTRATE_BORROWED + INFERENCE_EVENT
    # entries from sampling calls to the vacant's on-disk logbook
    # when we have a persistent identity. Ephemeral mode gets the
    # entries appended in memory but they're lost at process exit
    # (the keypair is also fresh-per-launch, so there's no audit
    # trail to preserve anyway).
    persistent_lb = None
    on_lb_change: Any = None
    if persistent_name is not None:
        persistent_lb = ls.load_logbook(persistent_name)
        captured_name = persistent_name
        on_lb_change = lambda lb: ls.save_logbook(captured_name, lb)  # noqa: E731

    run_mcp_stdio_server(
        form=form,
        signing_key=signing_key,
        replay_store=replay_store,
        logbook=persistent_lb,
        on_logbook_change=on_lb_change,
    )

install_cmd

install_cmd(client: str = Argument(..., help='MCP client to register vacant with: claude-code | claude-desktop | cursor | windsurf | openclaw | hermes'), config_path: str | None = Option(None, '--config-path', help='Override the default config-file location for this client.'), name: str = Option('alice', '--name', help='VACANT_NAME env var written into the registered MCP entry (picks which `~/.vacant/<name>/` identity the spawned `vacant mcp` uses; defaults to `alice`).'), force: bool = Option(False, '--force', help="Overwrite an existing `vacant` entry in the client's config."), dry_run: bool = Option(False, '--dry-run', help='Print what would be written without touching any file.')) -> None

Register vacant as an MCP server with a local client. (Pfix4)

One unified entry point — the README's per-client one-liners (OpenClaw / Hermes / Claude Desktop / Cursor / Windsurf) all collapse to:

vacant install <client>

Idempotent: re-running with no flags is a no-op when an entry is already in place. Pass --force to overwrite.

Source code in src/vacant/cli/commands.py

@app.command("install")
def install_cmd(
    client: str = typer.Argument(
        ...,
        help=(
            "MCP client to register vacant with: claude-code | claude-desktop | "
            "cursor | windsurf | openclaw | hermes"
        ),
    ),
    config_path: str | None = typer.Option(
        None,
        "--config-path",
        help="Override the default config-file location for this client.",
    ),
    name: str = typer.Option(
        "alice",
        "--name",
        help=(
            "VACANT_NAME env var written into the registered MCP entry "
            "(picks which `~/.vacant/<name>/` identity the spawned `vacant mcp` "
            "uses; defaults to `alice`)."
        ),
    ),
    force: bool = typer.Option(
        False,
        "--force",
        help="Overwrite an existing `vacant` entry in the client's config.",
    ),
    dry_run: bool = typer.Option(
        False,
        "--dry-run",
        help="Print what would be written without touching any file.",
    ),
) -> None:
    """Register vacant as an MCP server with a local client. (Pfix4)

    One unified entry point — the README's per-client one-liners
    (OpenClaw / Hermes / Claude Desktop / Cursor / Windsurf) all
    collapse to:

        vacant install <client>

    Idempotent: re-running with no flags is a no-op when an entry is
    already in place. Pass `--force` to overwrite.
    """
    from pathlib import Path

    from vacant.cli.install import SUPPORTED_CLIENTS, install

    if client not in SUPPORTED_CLIENTS:
        typer.echo(
            f"error: unknown client {client!r}; supported: {', '.join(SUPPORTED_CLIENTS)}",
            err=True,
        )
        raise typer.Exit(code=2)

    cp = Path(config_path) if config_path else None
    msg = install(client, config_path=cp, name=name, force=force, dry_run=dry_run)
    typer.echo(msg)
    if msg.startswith("ERROR"):
        raise typer.Exit(code=1)

demo_cmd

demo_cmd(scenario: str, substrate: str = Option('mock', '--substrate', '-s', help='mock | deterministic | anthropic | ollama | openai | gemini | mistral | hermes | openclaw'), seed: int | None = Option(None, '--seed', help='override default seed'), tail: bool = Option(False, '--tail', help='stream demo-store events to stdout instead of running'), db_path: str | None = Option(None, '--db', help='demo store path (default: var/demo.db)')) -> None

Run a demo scenario end-to-end. (P7)

Examples:

vacant demo law_firm vacant demo law-firm --seed=42 # hyphen accepted vacant demo self_replication --substrate=anthropic vacant demo law_firm --tail # tail events from demo store

Source code in src/vacant/cli/commands.py

@app.command("demo")
def demo_cmd(
    scenario: str,
    substrate: str = typer.Option(
        "mock",
        "--substrate",
        "-s",
        help=(
            "mock | deterministic | anthropic | ollama | openai | gemini | "
            "mistral | hermes | openclaw"
        ),
    ),
    seed: int | None = typer.Option(None, "--seed", help="override default seed"),
    tail: bool = typer.Option(
        False, "--tail", help="stream demo-store events to stdout instead of running"
    ),
    db_path: str | None = typer.Option(None, "--db", help="demo store path (default: var/demo.db)"),
) -> None:
    """Run a demo scenario end-to-end. (P7)

    Examples:
      vacant demo law_firm
      vacant demo law-firm --seed=42                # hyphen accepted
      vacant demo self_replication --substrate=anthropic
      vacant demo law_firm --tail                   # tail events from demo store
    """
    from vacant.mvp.demo import main as demo_main

    if tail:
        from vacant.mvp.demo_store import DemoStore

        with DemoStore(path=db_path) as store:
            for ev in store.read(scenario=scenario.replace("-", "_")):
                typer.echo(f"[{ev.ts:.1f}] {ev.kind}: {ev.payload}")
        return

    argv = ["--scenario", scenario.replace("-", "_"), "--substrate", substrate]
    if seed is not None:
        argv += ["--seed", str(seed)]
    if db_path is not None:
        argv += ["--db", db_path]
    raise SystemExit(demo_main(argv))

main

main() -> None

Console-script entrypoint declared in pyproject.toml.

Source code in src/vacant/cli/commands.py

def main() -> None:
    """Console-script entrypoint declared in `pyproject.toml`."""
    app()

local_store

On-disk layout for local vacants (~/.vacant/<name>/).

A local vacant is the owner-side handle for a vacant: keypair stored in the OS keyring (or a plaintext file in --insecure-demo mode), a logbook persisted as .jsonl, and a small meta.json carrying visibility state, capability text, and endpoint URL. Higher-level CLI commands (vacant init, status, publish, heartbeat, attest, call) read and write through this module.

Layout under ${VACANT_HOME:-~/.vacant}/<name>/:

key.json       {"pubkey_hex": "...", "key_storage": "keyring"}     (mode 0600)
               or {"pubkey_hex": ..., "seed_hex": ..., "key_storage": "plaintext"}
logbook.jsonl  one JSON-encoded LogEntry per line
meta.json      LocalMeta — state / endpoint / capability_text / etc.

Key storage (F-D codex final blockers): the Ed25519 seed is sensitive material — controlling it == owning the vacant. The default storage is the OS keyring (Keychain on macOS, Secret Service on Linux, Credential Locker on Windows) via the keyring package. The on-disk key.json holds only the public key and a key_storage discriminator so external tooling can verify signatures without unlocking the keychain.

If the host has no keyring backend (e.g. headless CI without DBus), init_vacant(insecure_demo=False) raises rather than silently falling back to plaintext. To opt into plaintext storage explicitly, pass insecure_demo=True (the CLI surface is vacant init <name> --insecure-demo); a stderr WARN is emitted and key.json is written with the seed in the clear under mode 0600.

The --insecure-demo mode exists for two purposes only: live demos where the operator is showing the file layout, and short-lived CI/test flows. Do not use it on a host with real network exposure. See SECURITY.md §"Local key storage" for the full risk model.

ENVELOPE_STATE_FILE module-attribute

ENVELOPE_STATE_FILE = 'envelope_state.json'

Per-target chain state for outgoing calls (Pfix3 B6).

Keyed by target vacant_id_hex; tracks the last accepted envelope on the request (caller → target) and response (target → caller) chains so the next vacant call to the same target advances seq + prev_envelope_hash correctly. Without this file the CLI defaulted sequence_no=1 on every call → second call to a target was rejected as replay by the server.

KEYRING_SERVICE module-attribute

KEYRING_SERVICE = 'vacant.cli'

service argument used for every keyring.set_password / keyring.get_password call. Stable across versions so the OS keyring entry survives upgrades.

LocalVacantError

Bases: RuntimeError

Base class for local-store errors.

LocalVacantNotFound

Bases: LocalVacantError

The named local vacant does not exist.

LocalVacantExists

Bases: LocalVacantError

A local vacant with that name already exists.

LocalVacantKeyringUnavailable

Bases: LocalVacantError

The default OS keyring is the fail / null backend.

Raised by init_vacant when the operator has not opted into insecure_demo=True. The error message tells the operator how to proceed: install a keyring backend or re-run with --insecure-demo.

LocalMeta

Bases: BaseModel

Sidecar metadata. Visibility / capability / endpoint live here so status can render them without opening the logbook.

key_storage class-attribute instance-attribute

key_storage: str = 'plaintext'

keyring (default, OS keyring) or plaintext (--insecure-demo). Defaults to plaintext so LocalMeta files written before F-D landed still load cleanly.

vacant_home

vacant_home() -> Path

Resolve the root directory: $VACANT_HOME or ~/.vacant.

Source code in src/vacant/cli/local_store.py

def vacant_home() -> Path:
    """Resolve the root directory: `$VACANT_HOME` or `~/.vacant`."""
    override = os.environ.get("VACANT_HOME")
    if override:
        return Path(override).expanduser()
    return Path.home() / ".vacant"

vacant_dir

vacant_dir(name: str) -> Path

Return the on-disk directory for vacant name. Validates the name so callers cannot escape the home directory via path traversal.

Source code in src/vacant/cli/local_store.py

def vacant_dir(name: str) -> Path:
    """Return the on-disk directory for vacant `name`. Validates the name
    so callers cannot escape the home directory via path traversal."""
    if not name or "/" in name or "\\" in name or "\0" in name or ".." in name:
        raise LocalVacantError(f"invalid local vacant name: {name!r}")
    return vacant_home() / name

list_vacant_names

list_vacant_names() -> list[str]

Names of every initialised local vacant, sorted.

Source code in src/vacant/cli/local_store.py

def list_vacant_names() -> list[str]:
    """Names of every initialised local vacant, sorted."""
    home = vacant_home()
    if not home.exists():
        return []
    out: list[str] = []
    for p in home.iterdir():
        if p.is_dir() and (p / META_FILE).exists():
            out.append(p.name)
    return sorted(out)

current_name

current_name() -> str

Resolve the active vacant: env VACANT_NAME, else the only one.

Raises LocalVacantNotFound if no vacant exists or multiple exist without VACANT_NAME set.

Source code in src/vacant/cli/local_store.py

def current_name() -> str:
    """Resolve the active vacant: env `VACANT_NAME`, else the only one.

    Raises `LocalVacantNotFound` if no vacant exists or multiple exist
    without `VACANT_NAME` set.
    """
    explicit = os.environ.get("VACANT_NAME")
    if explicit:
        return explicit
    names = list_vacant_names()
    if len(names) == 1:
        return names[0]
    if not names:
        raise LocalVacantNotFound("no local vacants; run `vacant init <name>` first")
    raise LocalVacantNotFound(
        f"VACANT_NAME not set and multiple vacants exist: {names!r}; "
        "set VACANT_NAME=<name> to select one"
    )

keyring_backend_available

keyring_backend_available() -> bool

True iff the host's default keyring is a real backend.

The keyring library always returns some backend from get_keyring(); on hosts without a working backend it returns keyring.backends.fail.Keyring (a stub that raises on every call). We detect that case by inspecting the module path so callers can give a clear error before a write attempt fails.

Source code in src/vacant/cli/local_store.py

def keyring_backend_available() -> bool:
    """True iff the host's default keyring is a real backend.

    The `keyring` library always returns *some* backend from
    `get_keyring()`; on hosts without a working backend it returns
    `keyring.backends.fail.Keyring` (a stub that raises on every
    call). We detect that case by inspecting the module path so
    callers can give a clear error before a write attempt fails.
    """
    try:
        backend = keyring.get_keyring()
    except KeyringError:
        return False
    module = type(backend).__module__
    return not module.endswith(".fail") and not module.endswith(".null")

init_vacant

init_vacant(name: str, *, insecure_demo: bool = False) -> tuple[VacantId, SigningKey]

Generate a fresh keypair and persist the local-vacant directory.

Parameters:

Name	Type	Description	Default
name	str	Local vacant name. Validated against path traversal.	required
insecure_demo	bool	If True, write the Ed25519 seed in plaintext into key.json (mode 0600) and emit a stderr WARN. If False (default), store the seed in the OS keyring; raise LocalVacantKeyringUnavailable if no backend is present.	False

Returns:

Type	Description
tuple[VacantId, SigningKey]	The new VacantId and SigningKey.

Raises:

Type	Description
LocalVacantExists	If ~/.vacant/<name>/ already exists.
LocalVacantKeyringUnavailable	If insecure_demo=False and the host has no working keyring backend.
LocalVacantError	Any other failure (path traversal, keyring write error, …).

Source code in src/vacant/cli/local_store.py

def init_vacant(name: str, *, insecure_demo: bool = False) -> tuple[VacantId, SigningKey]:
    """Generate a fresh keypair and persist the local-vacant directory.

    Args:
        name: Local vacant name. Validated against path traversal.
        insecure_demo: If True, write the Ed25519 seed in plaintext into
            `key.json` (mode 0600) and emit a stderr WARN. If False
            (default), store the seed in the OS keyring; raise
            `LocalVacantKeyringUnavailable` if no backend is present.

    Returns:
        The new `VacantId` and `SigningKey`.

    Raises:
        LocalVacantExists: If `~/.vacant/<name>/` already exists.
        LocalVacantKeyringUnavailable: If `insecure_demo=False` and the
            host has no working keyring backend.
        LocalVacantError: Any other failure (path traversal, keyring
            write error, …).
    """
    d = vacant_dir(name)
    if d.exists():
        raise LocalVacantExists(name)

    sk, vk = keygen()
    vid = VacantId.from_verify_key(vk)
    seed_hex = bytes(sk).hex()

    if insecure_demo:
        d.mkdir(parents=True)
        key_path = d / KEY_FILE
        key_path.write_text(
            json.dumps(
                {
                    "pubkey_hex": vid.hex(),
                    "seed_hex": seed_hex,
                    "key_storage": "plaintext",
                },
                sort_keys=True,
            )
        )
        os.chmod(key_path, 0o600)
        sys.stderr.write(_INSECURE_WARN.format(name=name, path=key_path))
        key_storage = "plaintext"
    else:
        if not keyring_backend_available():
            raise LocalVacantKeyringUnavailable(
                f"no keyring backend available for vacant {name!r}; "
                "the OS keyring (Keychain on macOS, Secret Service on Linux, "
                "Credential Locker on Windows) is the default storage for the "
                "private seed. Install / unlock a keyring backend, or re-run "
                "`vacant init <name> --insecure-demo` to opt into plaintext "
                "storage. See SECURITY.md §Local key storage for the risk model."
            )
        d.mkdir(parents=True)
        try:
            keyring.set_password(KEYRING_SERVICE, name, seed_hex)
        except KeyringError as exc:
            # Roll back the directory so a partial init doesn't block a
            # retry under a different mode.
            try:
                d.rmdir()
            except OSError:
                pass
            raise LocalVacantError(
                f"keyring store for vacant {name!r} failed: {exc}. "
                "Try `--insecure-demo` if this is a demo / CI host."
            ) from exc
        key_path = d / KEY_FILE
        key_path.write_text(
            json.dumps(
                {"pubkey_hex": vid.hex(), "key_storage": "keyring"},
                sort_keys=True,
            )
        )
        os.chmod(key_path, 0o600)
        key_storage = "keyring"

    lb = Logbook()
    lb.append(GENESIS_KIND, {"name": name, "vacant_id": vid.hex()}, sk)
    save_logbook(name, lb)

    meta = LocalMeta(
        vacant_id_hex=vid.hex(),
        state="LOCAL",
        created_at=datetime.now(UTC).isoformat(),
        key_storage=key_storage,
    )
    save_meta(name, meta)
    return vid, sk

load_signing_key

load_signing_key(name: str) -> SigningKey

Load the Ed25519 signing key for name.

Looks up key_storage from meta.json (or, for legacy directories without it, infers from key.json's key_storage field, then falls back to the plaintext seed). Raises LocalVacantNotFound if the directory is missing entirely; raises LocalVacantError if the keyring entry has gone missing under us (e.g. operator cleared the Keychain after init).

Source code in src/vacant/cli/local_store.py

def load_signing_key(name: str) -> SigningKey:
    """Load the Ed25519 signing key for `name`.

    Looks up `key_storage` from `meta.json` (or, for legacy directories
    without it, infers from `key.json`'s `key_storage` field, then falls
    back to the plaintext seed). Raises `LocalVacantNotFound` if the
    directory is missing entirely; raises `LocalVacantError` if the
    keyring entry has gone missing under us (e.g. operator cleared the
    Keychain after init).
    """
    p = vacant_dir(name) / KEY_FILE
    if not p.exists():
        raise LocalVacantNotFound(name)
    obj = json.loads(p.read_text())
    storage = obj.get("key_storage")
    if storage is None:
        # Legacy file: pre-F-D init wrote `seed_hex` without
        # `key_storage`. Treat as plaintext.
        storage = "plaintext" if "seed_hex" in obj else "keyring"

    if storage == "keyring":
        seed_hex = keyring.get_password(KEYRING_SERVICE, name)
        if seed_hex is None:
            raise LocalVacantError(
                f"keyring entry for vacant {name!r} not found "
                f"(service={KEYRING_SERVICE!r}); the OS keyring may have been "
                "cleared, or the vacant was created on a different machine. "
                "Reinitialise with `vacant init` or copy the keyring entry over."
            )
        return SigningKey(bytes.fromhex(seed_hex))

    # Plaintext (--insecure-demo or legacy).
    if "seed_hex" not in obj:
        raise LocalVacantError(
            f"vacant {name!r} key.json declares key_storage={storage!r} "
            "but has no seed_hex on disk; cannot load the signing key"
        )
    return SigningKey(bytes.fromhex(obj["seed_hex"]))

load_envelope_state

load_envelope_state(name: str) -> dict[str, Any]

Load the per-target chain state for vacant call.

Returns {} if the file doesn't exist yet (first call). Schema:

{
  "<target_vacant_id_hex>": {
    "request":  {"last_seq": int, "last_hash_hex": str},
    "response": {"last_seq": int, "last_hash_hex": str}
  }
}

Returned as dict[str, Any] because the file is JSON: leaf values are ints + strs and the caller knows the schema. Strict typing would force every read site through casts without buying safety beyond the schema docstring above.

Source code in src/vacant/cli/local_store.py

def load_envelope_state(name: str) -> dict[str, Any]:
    """Load the per-target chain state for ``vacant call``.

    Returns ``{}`` if the file doesn't exist yet (first call). Schema:

        {
          "<target_vacant_id_hex>": {
            "request":  {"last_seq": int, "last_hash_hex": str},
            "response": {"last_seq": int, "last_hash_hex": str}
          }
        }

    Returned as ``dict[str, Any]`` because the file is JSON: leaf
    values are ints + strs and the caller knows the schema. Strict
    typing would force every read site through casts without buying
    safety beyond the schema docstring above.
    """
    p = envelope_state_file(name)
    if not p.exists():
        return {}
    raw = json.loads(p.read_text())
    if not isinstance(raw, dict):
        return {}
    return raw

save_envelope_state

save_envelope_state(name: str, state: dict[str, Any]) -> None

Atomically persist the envelope state. Uses tempfile + os.replace so a crashed write does not leave a half-truncated file (would otherwise cause the next call to replay seq=1).

Source code in src/vacant/cli/local_store.py

def save_envelope_state(name: str, state: dict[str, Any]) -> None:
    """Atomically persist the envelope state. Uses ``tempfile +
    os.replace`` so a crashed write does not leave a half-truncated
    file (would otherwise cause the next call to replay seq=1).
    """
    d = vacant_dir(name)
    d.mkdir(parents=True, exist_ok=True)
    p = envelope_state_file(name)
    tmp = p.with_suffix(p.suffix + ".tmp")
    tmp.write_text(json.dumps(state, sort_keys=True, separators=(",", ":")))
    os.replace(tmp, p)