rho-cli 0.1.25

Rho CLI tools for encrypted agent collaboration, dataset publishing, controlled runs, and result release workflows
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import os
import shlex
import sys
from pathlib import Path


def github_handle(identity: str) -> str:
    prefix = "rho://id/github/"
    if not identity.startswith(prefix):
        raise ValueError(f"unsupported identity id: {identity}")
    handle = identity[len(prefix) :]
    if not handle or any(ch not in "abcdefghijklmnopqrstuvwxyz0123456789_-" for ch in handle):
        raise ValueError(f"invalid github identity handle: {handle}")
    return handle


def shell_quote(value: str) -> str:
    return shlex.quote(value)


def default_home_root() -> Path:
    return Path(
        os.environ.get("RHO_ENV_HOME_ROOT", Path.home() / ".rho")
    ).resolve()


def switch(args: argparse.Namespace) -> int:
    handle = github_handle(args.identity)
    home = Path(args.home).expanduser().resolve() if args.home else default_home_root()
    prompt = f"[rho:{handle}] "
    print(f"export RHO_IDENTITY={shell_quote(args.identity)}")
    print(f"export RHO_HOME={shell_quote(str(home))}")
    print(f"export RHO_ENV_HANDLE={shell_quote(handle)}")
    if args.prompt:
        print('if [ -z "${RHO_ENV_OLD_PS1+x}" ]; then export RHO_ENV_OLD_PS1="${PS1:-}"; fi')
        print(f"export PS1={shell_quote(prompt)}\"${{RHO_ENV_OLD_PS1:-}}\"")
    print(f"echo 'rho identity active: {args.identity}'")
    print(f"echo 'rho home: {home}'")
    return 0


def clear(args: argparse.Namespace) -> int:
    print("unset RHO_IDENTITY")
    print("unset RHO_ENV_HANDLE")
    print("unset RHO_HOME")
    if args.prompt:
        print('if [ -n "${RHO_ENV_OLD_PS1+x}" ]; then export PS1="$RHO_ENV_OLD_PS1"; unset RHO_ENV_OLD_PS1; fi')
    print("echo 'rho identity cleared'")
    return 0


def status(_args: argparse.Namespace) -> int:
    print(f"RHO_IDENTITY={os.environ.get('RHO_IDENTITY', '')}")
    print(f"RHO_HOME={os.environ.get('RHO_HOME', '')}")
    print(f"RHO_ENV_HANDLE={os.environ.get('RHO_ENV_HANDLE', '')}")
    return 0


def shell(args: argparse.Namespace) -> int:
    handle = github_handle(args.identity)
    home = Path(args.home).expanduser().resolve() if args.home else default_home_root()
    env = os.environ.copy()
    env["RHO_IDENTITY"] = args.identity
    env["RHO_HOME"] = str(home)
    env["RHO_ENV_HANDLE"] = handle
    env["PS1"] = f"[rho:{handle}] " + env.get("PS1", "")
    shell_path = os.environ.get("SHELL", "/bin/sh")
    os.execvpe(shell_path, [shell_path], env)
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="rho env")
    sub = parser.add_subparsers(dest="command", required=True)

    switch_parser = sub.add_parser("switch", help="print shell exports for an active rho identity")
    switch_parser.add_argument("identity")
    switch_parser.add_argument("--home")
    switch_parser.add_argument("--no-prompt", action="store_false", dest="prompt")
    switch_parser.set_defaults(func=switch, prompt=True)

    clear_parser = sub.add_parser("clear", help="print shell commands that clear active rho identity")
    clear_parser.add_argument("--no-prompt", action="store_false", dest="prompt")
    clear_parser.set_defaults(func=clear, prompt=True)

    status_parser = sub.add_parser("status", help="show active rho identity environment")
    status_parser.set_defaults(func=status)

    shell_parser = sub.add_parser("shell", help="start a subshell with an active rho identity")
    shell_parser.add_argument("identity")
    shell_parser.add_argument("--home")
    shell_parser.set_defaults(func=shell)
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    try:
        return args.func(args)
    except Exception as error:
        print(error, file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())