489 lines
21 KiB
Python
Executable File
489 lines
21 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Validate and execute the shared project Agent runtime contract.
|
||
|
||
The command surface is deliberately small. It never evaluates project text as
|
||
shell, and it only creates builder sessions from a fixed make invocation.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
try:
|
||
import yaml
|
||
except ImportError as exc: # pragma: no cover - dependency is checked at runtime
|
||
raise SystemExit("需要 PyYAML: python3 -m pip install pyyaml") from exc
|
||
|
||
|
||
CLI_NAMES = {"codex", "cursor-agent", "grok", "omp", "pi"}
|
||
ADAPTERS = {"builder", "tester", "ack"}
|
||
EFFORTS = {None, "low", "medium", "high", "xhigh"}
|
||
PERMISSIONS = {"read-only", "workspace-write"}
|
||
TIERS = {"standard", "strong"}
|
||
LIFECYCLES = {"persistent", "ephemeral"}
|
||
ID_RE = re.compile(r"^[a-z][a-z0-9-]{1,63}$")
|
||
BINDING_RE = re.compile(r"^[a-z][a-z0-9-]*(?:\.[a-z][a-z0-9-]*)+$")
|
||
TARGET_RE = re.compile(r"^[A-Za-z0-9_.-]{1,96}$")
|
||
MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/+@-]{0,127}$")
|
||
MAKEFILE_RE = re.compile(r"^[A-Za-z0-9_.-]+(?:/[A-Za-z0-9_.-]+)*$")
|
||
FORBIDDEN_KEYS = {"command", "argv", "shell", "env", "environmentVariables", "token", "secret", "password"}
|
||
|
||
|
||
class ConfigError(ValueError):
|
||
pass
|
||
|
||
|
||
class UniqueLoader(yaml.SafeLoader):
|
||
pass
|
||
|
||
|
||
def _construct_mapping(loader: UniqueLoader, node: yaml.MappingNode, deep: bool = False) -> dict[str, Any]:
|
||
mapping: dict[str, Any] = {}
|
||
for key_node, value_node in node.value:
|
||
key = loader.construct_object(key_node, deep=deep)
|
||
if not isinstance(key, str):
|
||
raise ConfigError("YAML mapping key 必须是字符串")
|
||
if key in mapping:
|
||
raise ConfigError(f"YAML 存在重复键: {key}")
|
||
mapping[key] = loader.construct_object(value_node, deep=deep)
|
||
return mapping
|
||
|
||
|
||
UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _construct_mapping)
|
||
|
||
|
||
def load_yaml(path: Path) -> dict[str, Any]:
|
||
try:
|
||
value = yaml.load(path.read_text(encoding="utf-8"), Loader=UniqueLoader)
|
||
except (OSError, yaml.YAMLError) as exc:
|
||
raise ConfigError(f"无法读取 {path}: {exc}") from exc
|
||
if value is None:
|
||
return {}
|
||
if not isinstance(value, dict):
|
||
raise ConfigError(f"{path} 顶层必须是 mapping")
|
||
return value
|
||
|
||
|
||
def deep_merge(base: dict[str, Any], override: dict[str, Any], where: str = "root") -> dict[str, Any]:
|
||
result = copy.deepcopy(base)
|
||
for key, value in override.items():
|
||
if key not in result:
|
||
raise ConfigError(f"{where}: local 配置不能增加未声明字段 {key!r}")
|
||
if isinstance(result[key], dict) and isinstance(value, dict):
|
||
result[key] = deep_merge(result[key], value, f"{where}.{key}")
|
||
else:
|
||
result[key] = copy.deepcopy(value)
|
||
return result
|
||
|
||
|
||
def text(value: Any, where: str) -> str:
|
||
if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 for c in value):
|
||
raise ConfigError(f"{where}: 必须是非空文本")
|
||
return value
|
||
|
||
|
||
def safe_id(value: Any, where: str) -> str:
|
||
value = text(value, where)
|
||
if not ID_RE.fullmatch(value):
|
||
raise ConfigError(f"{where}: ID 只能使用小写字母、数字和连字符")
|
||
return value
|
||
|
||
|
||
def safe_binding_id(value: Any, where: str) -> str:
|
||
value = text(value, where)
|
||
if not BINDING_RE.fullmatch(value):
|
||
raise ConfigError(f"{where}: binding 必须是 namespace.name 形式")
|
||
return value
|
||
|
||
|
||
def safe_path(value: Any, where: str) -> str:
|
||
value = text(value, where)
|
||
path = Path(value)
|
||
if not path.is_absolute() or value == os.path.sep or ".." in path.parts or os.path.normpath(value) != value:
|
||
raise ConfigError(f"{where}: 必须是规范绝对路径")
|
||
return value
|
||
|
||
|
||
def validate(config: dict[str, Any]) -> None:
|
||
if config.get("schemaVersion") != 1:
|
||
raise ConfigError("schemaVersion 必须为 1")
|
||
for key in ("locations", "profiles", "bindings"):
|
||
if not isinstance(config.get(key), dict) or not config[key]:
|
||
raise ConfigError(f"{key}: 必须是非空 mapping")
|
||
locations = config["locations"]
|
||
for location_id, location in locations.items():
|
||
safe_id(location_id, f"locations.{location_id}")
|
||
if not isinstance(location, dict):
|
||
raise ConfigError(f"locations.{location_id}: 必须是 mapping")
|
||
for required in ("environment", "project", "path"):
|
||
if required not in location:
|
||
raise ConfigError(f"locations.{location_id}: 缺少 {required}")
|
||
environment = text(location["environment"], f"locations.{location_id}.environment")
|
||
if environment != "local" and not ID_RE.fullmatch(environment):
|
||
raise ConfigError(f"locations.{location_id}.environment: 必须是 local 或安全环境名")
|
||
text(location["project"], f"locations.{location_id}.project")
|
||
safe_path(location["path"], f"locations.{location_id}.path")
|
||
host = location.get("host")
|
||
if host is not None:
|
||
host = text(host, f"locations.{location_id}.host")
|
||
if host != "local" and not re.fullmatch(r"runtime:[0-9a-f-]{16,}", host):
|
||
raise ConfigError(f"locations.{location_id}.host: 必须是 local 或 runtime:<environment-id>")
|
||
profiles = config["profiles"]
|
||
for profile_id, profile in profiles.items():
|
||
safe_id(profile_id, f"profiles.{profile_id}")
|
||
if not isinstance(profile, dict):
|
||
raise ConfigError(f"profiles.{profile_id}: 必须是 mapping")
|
||
for required in ("cli", "model", "reasoningEffort", "permissionMode"):
|
||
if required not in profile:
|
||
raise ConfigError(f"profiles.{profile_id}: 缺少 {required}")
|
||
if profile["cli"] not in CLI_NAMES:
|
||
raise ConfigError(f"profiles.{profile_id}.cli: 不支持 {profile['cli']!r}")
|
||
if profile["model"] is not None and MODEL_RE.fullmatch(str(profile["model"])) is None:
|
||
raise ConfigError(f"profiles.{profile_id}.model: 不是安全模型 ID")
|
||
if profile["reasoningEffort"] not in EFFORTS:
|
||
raise ConfigError(f"profiles.{profile_id}.reasoningEffort: 不支持该值")
|
||
if profile["permissionMode"] not in PERMISSIONS:
|
||
raise ConfigError(f"profiles.{profile_id}.permissionMode: 不支持该值")
|
||
if profile.get("tier", "standard") not in TIERS:
|
||
raise ConfigError(f"profiles.{profile_id}.tier: 必须是 standard/strong")
|
||
for binding_id, binding in config["bindings"].items():
|
||
safe_binding_id(binding_id, f"bindings.{binding_id}")
|
||
if not isinstance(binding, dict):
|
||
raise ConfigError(f"bindings.{binding_id}: 必须是 mapping")
|
||
for required in ("adapter", "location", "profile"):
|
||
if required not in binding:
|
||
raise ConfigError(f"bindings.{binding_id}: 缺少 {required}")
|
||
if binding["adapter"] not in ADAPTERS:
|
||
raise ConfigError(f"bindings.{binding_id}.adapter: 不支持 {binding['adapter']!r}")
|
||
location_id = safe_id(binding["location"], f"bindings.{binding_id}.location")
|
||
profile_id = safe_id(binding["profile"], f"bindings.{binding_id}.profile")
|
||
if location_id not in locations or profile_id not in profiles:
|
||
raise ConfigError(f"bindings.{binding_id}: location/profile 引用不存在")
|
||
if binding["adapter"] == "builder":
|
||
makefile = text(binding.get("makefile", "makefile.builder"), f"bindings.{binding_id}.makefile")
|
||
if Path(makefile).is_absolute() or ".." in Path(makefile).parts or MAKEFILE_RE.fullmatch(makefile) is None:
|
||
raise ConfigError(f"bindings.{binding_id}.makefile: 必须是项目内相对路径")
|
||
target = text(binding.get("target"), f"bindings.{binding_id}.target")
|
||
if TARGET_RE.fullmatch(target) is None:
|
||
raise ConfigError(f"bindings.{binding_id}.target: 不是安全 make target")
|
||
session = binding.get("session", {})
|
||
if not isinstance(session, dict) or session.get("lifecycle", "ephemeral") not in LIFECYCLES:
|
||
raise ConfigError(f"bindings.{binding_id}.session.lifecycle: 必须是 persistent/ephemeral")
|
||
scan_forbidden(config)
|
||
|
||
|
||
def scan_forbidden(value: Any, where: str = "root") -> None:
|
||
if isinstance(value, dict):
|
||
for key, child in value.items():
|
||
if key in FORBIDDEN_KEYS:
|
||
raise ConfigError(f"{where}: 禁止配置字段 {key!r}")
|
||
scan_forbidden(child, f"{where}.{key}")
|
||
elif isinstance(value, list):
|
||
for index, child in enumerate(value):
|
||
scan_forbidden(child, f"{where}[{index}]")
|
||
|
||
|
||
def load_config(root: Path) -> dict[str, Any]:
|
||
path = root / ".pouch" / "agent.yaml"
|
||
if not path.is_file():
|
||
raise ConfigError(f"缺少 {path}，请先创建项目通用 Agent 配置")
|
||
config = load_yaml(path)
|
||
local_path = root / ".pouch" / "agent.local.yaml"
|
||
if local_path.is_file():
|
||
config = deep_merge(config, load_yaml(local_path))
|
||
validate(config)
|
||
return config
|
||
|
||
|
||
def binding(config: dict[str, Any], binding_id: str) -> dict[str, Any]:
|
||
if binding_id not in config["bindings"]:
|
||
raise ConfigError(f"不存在 binding: {binding_id}")
|
||
value = copy.deepcopy(config["bindings"][binding_id])
|
||
value["id"] = binding_id
|
||
value["locationId"] = value["location"]
|
||
value["location"] = copy.deepcopy(config["locations"][value["locationId"]])
|
||
value["profileId"] = value["profile"]
|
||
value["profile"] = copy.deepcopy(config["profiles"][value["profileId"]])
|
||
return value
|
||
|
||
|
||
def orca_json(args: list[str], environment: str | None = None) -> dict[str, Any]:
|
||
command = ["orca", *args]
|
||
if environment and environment != "local":
|
||
command.extend(["--environment", environment])
|
||
command.append("--json")
|
||
proc = subprocess.run(command, capture_output=True, text=True, check=False)
|
||
raw = (proc.stdout or "").strip()
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
detail = (proc.stderr or raw or str(exc)).strip()
|
||
raise ConfigError(f"Orca 没有返回 JSON: {' '.join(command)}\n{detail}") from exc
|
||
if not data.get("ok"):
|
||
error = data.get("error") or {}
|
||
raise ConfigError(f"{' '.join(command)}: {error.get('message') or error.get('code') or 'unknown error'}")
|
||
result = data.get("result")
|
||
return result if isinstance(result, dict) else {}
|
||
|
||
|
||
def environment_id(name: str) -> str:
|
||
if name == "local":
|
||
return "local"
|
||
result = orca_json(["environment", "list"])
|
||
for item in result.get("environments", []):
|
||
if isinstance(item, dict) and item.get("name") == name:
|
||
return str(item.get("id"))
|
||
raise ConfigError(f"找不到已保存 Orca environment: {name}")
|
||
|
||
|
||
def location_host(location: dict[str, Any]) -> str:
|
||
host = location.get("host")
|
||
if host:
|
||
return str(host)
|
||
if location["environment"] == "local":
|
||
return "local"
|
||
return f"runtime:{environment_id(str(location['environment']))}"
|
||
|
||
|
||
def worktree_for(binding_value: dict[str, Any]) -> dict[str, Any]:
|
||
location = binding_value["location"]
|
||
environment = str(location["environment"])
|
||
project_ids = {str(location["project"])}
|
||
projects = orca_json(["project", "list"], environment).get("projects", [])
|
||
for project in projects:
|
||
if not isinstance(project, dict):
|
||
continue
|
||
if str(project.get("id")) == str(location["project"]) or str(project.get("displayName")) == str(location["project"]):
|
||
project_ids.update(str(repo_id) for repo_id in project.get("sourceRepoIds", []) if repo_id)
|
||
result = orca_json(["worktree", "list"], environment)
|
||
candidates = []
|
||
for item in result.get("worktrees", []):
|
||
if not isinstance(item, dict):
|
||
continue
|
||
item_project_ids = {str(item.get("projectId")), str(item.get("repoId"))}
|
||
if not project_ids.intersection(item_project_ids):
|
||
continue
|
||
if str(item.get("path")) != str(location["path"]):
|
||
continue
|
||
expected_host = location_host(location)
|
||
# A routed remote runtime stamps its own worktrees with hostId=local;
|
||
# the environment name/ID is the remote-machine discriminator.
|
||
if location["environment"] == "local" and expected_host == "local" and str(item.get("hostId")) != "local":
|
||
continue
|
||
candidates.append(item)
|
||
if len(candidates) != 1:
|
||
if not candidates:
|
||
raise ConfigError(
|
||
f"找不到精确 worktree: environment={environment}, host={location_host(location)}, "
|
||
f"project={location['project']}, path={location['path']}；先创建或登记该 worktree"
|
||
)
|
||
raise ConfigError(f"binding 匹配到多个 worktree: {len(candidates)}")
|
||
return candidates[0]
|
||
|
||
|
||
def project_root_commit(root: Path) -> str | None:
|
||
proc = subprocess.run(["git", "-C", str(root), "rev-parse", "HEAD"], capture_output=True, text=True, check=False)
|
||
return proc.stdout.strip() if proc.returncode == 0 else None
|
||
|
||
|
||
def run_dir(root: Path) -> Path:
|
||
path = root / ".pouch" / "agent" / "runs"
|
||
path.mkdir(parents=True, exist_ok=True)
|
||
return path
|
||
|
||
|
||
def run_path(root: Path, run_id: str) -> Path:
|
||
if not re.fullmatch(r"run_[0-9a-f]{16,64}", run_id):
|
||
raise ConfigError("run id 格式无效")
|
||
return run_dir(root) / f"{run_id}.json"
|
||
|
||
|
||
def cmd_validate(args: argparse.Namespace) -> int:
|
||
config = load_config(Path(args.root).resolve())
|
||
print(json.dumps({"ok": True, "schemaVersion": config["schemaVersion"], "bindings": sorted(config["bindings"])}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def cmd_show(args: argparse.Namespace) -> int:
|
||
config = load_config(Path(args.root).resolve())
|
||
value = binding(config, args.binding)
|
||
print(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True))
|
||
return 0
|
||
|
||
|
||
def cmd_worktree_resolve(args: argparse.Namespace) -> int:
|
||
root = Path(args.root).resolve()
|
||
value = binding(load_config(root), args.binding)
|
||
worktree = worktree_for(value)
|
||
print(json.dumps({"binding": args.binding, "worktree": worktree}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def cmd_worktree_create(args: argparse.Namespace) -> int:
|
||
root = Path(args.root).resolve()
|
||
value = binding(load_config(root), args.binding)
|
||
location = value["location"]
|
||
try:
|
||
existing = worktree_for(value)
|
||
except ConfigError as exc:
|
||
if "找不到精确 worktree" not in str(exc):
|
||
raise
|
||
existing = None
|
||
if existing:
|
||
print(json.dumps({"created": False, "binding": args.binding, "worktree": existing}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
host = location_host(location)
|
||
command = ["worktree", "create", "--project", str(location["project"]), "--host", host, "--name", args.name, "--no-parent"]
|
||
if args.base:
|
||
command.extend(["--base-branch", args.base])
|
||
result = orca_json(command, str(location["environment"]))
|
||
print(json.dumps({"created": True, "binding": args.binding, "result": result, "next": "将返回 worktree 的 path 写回 location.path 后再运行 resolve"}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def make_command(value: dict[str, Any]) -> str:
|
||
return f"make -f {value.get('makefile', 'makefile.builder')} {value['target']}"
|
||
|
||
|
||
def cmd_run_start(args: argparse.Namespace) -> int:
|
||
root = Path(args.root).resolve()
|
||
value = binding(load_config(root), args.binding)
|
||
if value["adapter"] != "builder":
|
||
raise ConfigError(f"run start 目前只为 builder 提供受控 make session；{value['adapter']} 请走对应 skill adapter")
|
||
if value.get("session", {}).get("lifecycle", "ephemeral") != "persistent":
|
||
raise ConfigError(f"binding {args.binding} 不是 persistent session")
|
||
worktree = worktree_for(value)
|
||
environment = str(value["location"]["environment"])
|
||
result = orca_json(["terminal", "create", "--worktree", f"id:{worktree['id']}", "--command", make_command(value), "--title", args.binding], environment)
|
||
terminal = result.get("terminal") or result
|
||
handle = terminal.get("handle") or result.get("handle")
|
||
if not handle:
|
||
raise ConfigError("Orca terminal create 没有返回 handle")
|
||
run_id = "run_" + uuid.uuid4().hex
|
||
record = {
|
||
"schemaVersion": 1,
|
||
"id": run_id,
|
||
"binding": args.binding,
|
||
"adapter": value["adapter"],
|
||
"profileId": value["profileId"],
|
||
"profileHash": hashlib.sha256(json.dumps(value["profile"], sort_keys=True).encode()).hexdigest(),
|
||
"environment": environment,
|
||
"project": value["location"]["project"],
|
||
"path": value["location"]["path"],
|
||
"worktreeId": worktree["id"],
|
||
"terminalHandle": handle,
|
||
"command": make_command(value),
|
||
"sourceCommit": project_root_commit(root),
|
||
"createdAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||
"status": "started",
|
||
}
|
||
run_path(root, run_id).write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
print(json.dumps(record, ensure_ascii=False, indent=2))
|
||
if args.wait_ms:
|
||
wait = orca_json(["terminal", "wait", "--terminal", handle, "--for", "exit", "--timeout-ms", str(args.wait_ms)], environment)
|
||
record["wait"] = wait
|
||
record["status"] = "completed" if wait.get("exited") or wait.get("status") == "exited" else "waited"
|
||
run_path(root, run_id).write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
print(json.dumps(wait, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def load_run(root: Path, run_id: str) -> dict[str, Any]:
|
||
path = run_path(root, run_id)
|
||
if not path.is_file():
|
||
raise ConfigError(f"找不到 run 记录: {run_id}")
|
||
value = json.loads(path.read_text(encoding="utf-8"))
|
||
if not isinstance(value, dict):
|
||
raise ConfigError(f"run 记录无效: {run_id}")
|
||
return value
|
||
|
||
|
||
def cmd_run_status(args: argparse.Namespace) -> int:
|
||
root = Path(args.root).resolve()
|
||
record = load_run(root, args.run)
|
||
result = orca_json(["terminal", "show", "--terminal", record["terminalHandle"]], record.get("environment"))
|
||
print(json.dumps({"run": record, "terminal": result}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def cmd_run_attach(args: argparse.Namespace) -> int:
|
||
root = Path(args.root).resolve()
|
||
record = load_run(root, args.run)
|
||
result = orca_json(["terminal", "read", "--terminal", record["terminalHandle"]], record.get("environment"))
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def cmd_run_resume(args: argparse.Namespace) -> int:
|
||
"""Explicitly reattach to a recorded terminal; never search for another one."""
|
||
|
||
root = Path(args.root).resolve()
|
||
record = load_run(root, args.run)
|
||
result = orca_json(["terminal", "show", "--terminal", record["terminalHandle"]], record.get("environment"))
|
||
print(json.dumps({"resumed": True, "run": record, "terminal": result}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def cmd_run_stop(args: argparse.Namespace) -> int:
|
||
root = Path(args.root).resolve()
|
||
record = load_run(root, args.run)
|
||
result = orca_json(["terminal", "close", "--terminal", record["terminalHandle"]], record.get("environment"))
|
||
record["status"] = "stopped"
|
||
record["stoppedAt"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||
run_path(root, args.run).write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
print(json.dumps({"run": record, "close": result}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def parser() -> argparse.ArgumentParser:
|
||
root_parent = argparse.ArgumentParser(add_help=False)
|
||
root_parent.add_argument("--root", default=".")
|
||
parser = argparse.ArgumentParser(description="共享 Agent 配置、worktree 和 session 运行时")
|
||
sub = parser.add_subparsers(dest="command", required=True)
|
||
validate_parser = sub.add_parser("validate", parents=[root_parent])
|
||
validate_parser.set_defaults(func=cmd_validate)
|
||
show = sub.add_parser("show", parents=[root_parent])
|
||
show.add_argument("--binding", required=True)
|
||
show.set_defaults(func=cmd_show)
|
||
wt = sub.add_parser("worktree")
|
||
wt_sub = wt.add_subparsers(dest="worktree_command", required=True)
|
||
resolve = wt_sub.add_parser("resolve", parents=[root_parent])
|
||
resolve.add_argument("--binding", required=True)
|
||
resolve.set_defaults(func=cmd_worktree_resolve)
|
||
create = wt_sub.add_parser("create", parents=[root_parent])
|
||
create.add_argument("--binding", required=True)
|
||
create.add_argument("--name", required=True)
|
||
create.add_argument("--base", default="")
|
||
create.set_defaults(func=cmd_worktree_create)
|
||
run = sub.add_parser("run")
|
||
run_sub = run.add_subparsers(dest="run_command", required=True)
|
||
start = run_sub.add_parser("start", parents=[root_parent])
|
||
start.add_argument("--binding", required=True)
|
||
start.add_argument("--wait-ms", type=int, default=0)
|
||
start.set_defaults(func=cmd_run_start)
|
||
for name, func in (("status", cmd_run_status), ("attach", cmd_run_attach), ("resume", cmd_run_resume), ("stop", cmd_run_stop)):
|
||
item = run_sub.add_parser(name, parents=[root_parent])
|
||
item.add_argument("--run", required=True)
|
||
item.set_defaults(func=func)
|
||
return parser
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
parsed = parser().parse_args()
|
||
raise SystemExit(parsed.func(parsed))
|
||
except ConfigError as exc:
|
||
print(f"environment-fail: {exc}", file=sys.stderr)
|
||
raise SystemExit(2)
|