Agent Traceboard：运行审批与证据时间线
查看原始点子 →41_mvp.pyPython
#!/usr/bin/env python3
"""Agent Traceboard: local approvals and an evidence timeline."""
from __future__ import annotations
import argparse
import dataclasses
import datetime as dt
import hashlib
import html
import json
import os
import re
import shlex
import subprocess
import sys
import tempfile
import threading
import time
import unittest
import uuid
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
from urllib.parse import parse_qs, urlparse
APP_NAME = "Agent Traceboard"
SCHEMA_VERSION = 1
BASE_DIR = Path(__file__).resolve().parent
DEFAULT_STORE = BASE_DIR / ".traceboard" / "events.jsonl"
MAX_CAPTURE = 64 * 1024
STORE_LOCK = threading.Lock()
RISK_RULES: List[Tuple[str, str, re.Pattern[str]]] = [
("critical", "递归强制删除或磁盘操作", re.compile(r"(^|[;&|]\s*)((rm\s+(?=[^;&|\n]*(?:-[A-Za-z]*r|--recursive))(?=[^;&|\n]*(?:-[A-Za-z]*f|--force)))|mkfs\b|dd\s+[^\n]*\bof=)", re.I)),
("high", "删除文件", re.compile(r"(^|[;&|]\s*)rm\s+", re.I)),
("high", "提权执行", re.compile(r"(^|[;&|]\s*)sudo\b", re.I)),
("high", "远程脚本管道执行", re.compile(r"\b(curl|wget)\b[^\n|]*\|\s*(ba)?sh\b", re.I)),
("high", "Git 强制覆盖或清理", re.compile(r"\bgit\s+(reset\s+--hard|clean\s+-[^\n]*f|push\s+[^\n]*--force)\b", re.I)),
("high", "系统或服务配置变更", re.compile(r"\b(systemctl|shutdown|reboot|chown|chmod)\b", re.I)),
("medium", "软件安装或卸载", re.compile(r"\b(pip|npm|apt|get|yum|dnf|pacman)\b[^\n]*(install|remove|uninstall|upgrade)", re.I)),
("medium", "可能的外部写入", re.compile(r"\b(curl|wget)\b[^\n]*\s(-X|--request)\s*(POST|PUT|PATCH|DELETE)\b", re.I)),
]
def utc_now() -> str:
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def new_id(prefix: str) -> str:
return f"{prefix}_{uuid.uuid4().hex[:12]}"
def clip(value: str, limit: int = MAX_CAPTURE) -> Tuple[str, bool]:
if len(value.encode("utf-8", errors="replace")) <= limit:
return value, False
encoded = value.encode("utf-8", errors="replace")[:limit]
return encoded.decode("utf-8", errors="ignore"), True
@dataclasses.dataclass
class Event:
"""JSONL 协议中的单个不可变事件。状态变化通过追加新事件表达。"""
type: str
run_id: str
action_id: Optional[str] = None
status: str = "recorded"
risk: str = "none"
summary: str = ""
data: Dict[str, Any] = dataclasses.field(default_factory=dict)
event_id: str = dataclasses.field(default_factory=lambda: new_id("evt"))
ts: str = dataclasses.field(default_factory=utc_now)
schema: int = SCHEMA_VERSION
def as_dict(self) -> Dict[str, Any]:
return dataclasses.asdict(self)
class EventStore:
def __init__(self, path: Path):
self.path = path.expanduser().resolve()
def append(self, event: Event) -> Dict[str, Any]:
"""原子追加一行 JSON，保留完整审计历史。"""
self.path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(event.as_dict(), ensure_ascii=False, separators=(",", ":")) + "\n"
with STORE_LOCK:
with self.path.open("a", encoding="utf-8") as handle:
handle.write(line)
handle.flush()
os.fsync(handle.fileno())
return event.as_dict()
def read(self) -> List[Dict[str, Any]]:
if not self.path.exists():
return []
events: List[Dict[str, Any]] = []
with STORE_LOCK:
with self.path.open("r", encoding="utf-8") as handle:
for number, line in enumerate(handle, 1):
if not line.strip():
continue
try:
item = json.loads(line)
except json.JSONDecodeError:
item = {"type": "protocol.error", "status": "failed", "risk": "high", "summary": f"第 {number} 行 JSON 损坏", "data": {}}
if not isinstance(item, dict):
item = {"type": "protocol.error", "status": "failed", "risk": "high", "summary": f"第 {number} 行不是 JSON 对象", "data": {}}
events.append(item)
return events
def action_events(self, action_id: str) -> List[Dict[str, Any]]:
return [e for e in self.read() if e.get("action_id") == action_id]
def current_status(self, action_id: str) -> Optional[str]:
events = self.action_events(action_id)
return events[-1].get("status") if events else None
def requested_action(self, action_id: str) -> Optional[Dict[str, Any]]:
for event in self.action_events(action_id):
if event.get("type") == "action.requested":
return event
return None
def assess_command(command: str) -> Tuple[str, List[str]]:
rank = {"none": 0, "low": 1, "medium": 2, "high": 3, "critical": 4}
risk, reasons = "low", []
for level, reason, pattern in RISK_RULES:
if pattern.search(command):
reasons.append(reason)
if rank[level] > rank[risk]:
risk = level
return risk, reasons
def file_risk(path: Path, root: Path) -> Tuple[str, List[str]]:
resolved = path.expanduser().resolve()
try:
resolved.relative_to(root.expanduser().resolve())
except ValueError:
return "high", ["写入目标位于允许根目录之外"]
if resolved.name in {".env", "authorized_keys", "passwd", "shadow"}:
return "high", ["目标可能包含凭据或访问控制配置"]
return "low", []
def record_request(store: EventStore, run_id: str, kind: str, payload: Dict[str, Any], risk: str, reasons: List[str]) -> str:
action_id = new_id("act")
requires_approval = risk in {"medium", "high", "critical"}
status = "pending" if requires_approval else "approved"
label = payload.get("command") or payload.get("path") or kind
store.append(Event(
type="action.requested", run_id=run_id, action_id=action_id,
status=status, risk=risk, summary=f"请求 {kind}: {label}",
data={"kind": kind, "payload": payload, "reasons": reasons, "requires_approval": requires_approval},
))
if not requires_approval:
store.append(Event(type="approval.auto", run_id=run_id, action_id=action_id, status="approved", risk=risk, summary="低风险操作自动批准"))
return action_id
def execute_shell(store: EventStore, requested: Dict[str, Any]) -> int:
data = requested["data"]
payload = data["payload"]
action_id, run_id, risk = requested["action_id"], requested["run_id"], requested["risk"]
store.append(Event(type="action.started", run_id=run_id, action_id=action_id, status="running", risk=risk, summary="开始执行 Shell 命令"))
started = time.monotonic()
try:
result = subprocess.run(
payload["command"], shell=True, cwd=payload.get("cwd") or None,
text=True, capture_output=True, timeout=payload.get("timeout", 60),
)
stdout, stdout_cut = clip(result.stdout)
stderr, stderr_cut = clip(result.stderr)
status = "succeeded" if result.returncode == 0 else "failed"
store.append(Event(
type="action.finished", run_id=run_id, action_id=action_id, status=status, risk=risk,
summary=f"Shell 命令{('完成' if result.returncode == 0 else '失败')}，退出码 {result.returncode}",
data={"exit_code": result.returncode, "duration_ms": round((time.monotonic() - started) * 1000), "stdout": stdout, "stderr": stderr, "truncated": stdout_cut or stderr_cut},
))
if stdout:
print(stdout, end="" if stdout.endswith("\n") else "\n")
if stderr:
print(stderr, file=sys.stderr, end="" if stderr.endswith("\n") else "\n")
return result.returncode
except subprocess.TimeoutExpired as exc:
store.append(Event(type="action.finished", run_id=run_id, action_id=action_id, status="failed", risk=risk, summary="Shell 命令超时", data={"error": str(exc)}))
print(f"命令超时: {exc}", file=sys.stderr)
return 124
except OSError as exc:
store.append(Event(type="action.finished", run_id=run_id, action_id=action_id, status="failed", risk=risk, summary="Shell 命令无法启动", data={"error": str(exc)}))
print(f"命令无法启动: {exc}", file=sys.stderr)
return 1
def execute_write(store: EventStore, requested: Dict[str, Any]) -> int:
payload = requested["data"]["payload"]
target = Path(payload["path"]).expanduser().resolve()
content = payload["content"]
target.parent.mkdir(parents=True, exist_ok=True)
before_hash = None
if target.exists() and target.is_file():
before_hash = hashlib.sha256(target.read_bytes()).hexdigest()
store.append(Event(type="action.started", run_id=requested["run_id"], action_id=requested["action_id"], status="running", risk=requested["risk"], summary=f"开始写入文件 {target}"))
try:
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=str(target.parent), delete=False) as handle:
handle.write(content)
temp_name = handle.name
os.replace(temp_name, target)
after_hash = hashlib.sha256(target.read_bytes()).hexdigest()
store.append(Event(
type="action.finished", run_id=requested["run_id"], action_id=requested["action_id"], status="succeeded", risk=requested["risk"],
summary=f"文件写入完成 {target}", data={"path": str(target), "bytes": len(content.encode('utf-8')), "before_sha256": before_hash, "after_sha256": after_hash},
))
print(f"已写入: {target}")
return 0
except OSError as exc:
store.append(Event(type="action.finished", run_id=requested["run_id"], action_id=requested["action_id"], status="failed", risk=requested["risk"], summary="文件写入失败", data={"error": str(exc)}))
print(f"写入失败: {exc}", file=sys.stderr)
return 1
def execute_action(store: EventStore, requested: Dict[str, Any]) -> int:
kind = requested.get("data", {}).get("kind")
if kind == "shell":
return execute_shell(store, requested)
if kind == "file_write":
return execute_write(store, requested)
print(f"不支持的操作类型: {kind}", file=sys.stderr)
return 2
def decide(store: EventStore, action_id: str, approved: bool, actor: str, reason: str) -> int:
"""审批状态机只允许 pending -> approved/rejected，批准后立即执行。"""
requested = store.requested_action(action_id)
if not requested:
print(f"未找到 action: {action_id}", file=sys.stderr)
return 2
status = store.current_status(action_id)
if status != "pending":
print(f"action 当前状态为 {status}，不能重复审批", file=sys.stderr)
return 2
decision = "approved" if approved else "rejected"
store.append(Event(
type=f"approval.{decision}", run_id=requested["run_id"], action_id=action_id,
status=decision, risk=requested["risk"], summary=f"{actor} 已{('批准' if approved else '拒绝')}操作",
data={"actor": actor, "reason": reason},
))
if not approved:
print(f"已拒绝: {action_id}")
return 0
return execute_action(store, requested)
def request_shell(store: EventStore, args: argparse.Namespace) -> int:
parts = list(args.command)
if parts and parts[0] == "--":
parts.pop(0)
# 单个参数可表示完整 Shell 表达式；多参数则保留原始参数边界。
command = (parts[0] if len(parts) == 1 else shlex.join(parts)).strip()
if not command:
print("缺少命令", file=sys.stderr)
return 2
risk, reasons = assess_command(command)
run_id = args.run_id or new_id("run")
action_id = record_request(store, run_id, "shell", {"command": command, "cwd": str(Path(args.cwd).resolve()), "timeout": args.timeout}, risk, reasons)
if store.current_status(action_id) == "pending":
print(f"WAITING_APPROVAL: {action_id} risk={risk} reasons={'; '.join(reasons)}")
print(f"批准并执行: python {shlex.quote(str(Path(__file__).resolve()))} approve {action_id}")
return 3
return execute_action(store, store.requested_action(action_id) or {})
def request_write(store: EventStore, args: argparse.Namespace) -> int:
target = Path(args.path).expanduser().resolve()
content = sys.stdin.read() if args.content is None else args.content
root = Path(args.root).expanduser().resolve()
risk, reasons = file_risk(target, root)
run_id = args.run_id or new_id("run")
action_id = record_request(store, run_id, "file_write", {"path": str(target), "content": content}, risk, reasons)
if store.current_status(action_id) == "pending":
print(f"WAITING_APPROVAL: {action_id} risk={risk} reasons={'; '.join(reasons)}")
print(f"批准并写入: python {shlex.quote(str(Path(__file__).resolve()))} approve {action_id}")
return 3
return execute_action(store, store.requested_action(action_id) or {})
def effective_events(events: Iterable[Dict[str, Any]], run_id: Optional[str] = None) -> List[Dict[str, Any]]:
result = [e for e in events if not run_id or e.get("run_id") == run_id]
return sorted(result, key=lambda e: (e.get("ts", ""), e.get("event_id", "")))
PAGE = """<!doctype html>
<html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>Agent Traceboard</title><style>
:root{color-scheme:light;--ink:#17202a;--muted:#667085;--line:#d8dee7;--bg:#f5f7fa;--panel:#fff;--red:#b42318;--amber:#b54708;--green:#067647;--blue:#175cd3}
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--ink);font:14px/1.5 system-ui,sans-serif}header{background:#101828;color:#fff;padding:18px 24px}h1{font-size:20px;margin:0}.sub{color:#bfc8d6;margin-top:2px}.wrap{max-width:1180px;margin:0 auto;padding:20px 24px}.stats{display:grid;grid-template-columns:repeat(4,1fr);gap:10px;margin-bottom:18px}.stat{background:var(--panel);border:1px solid var(--line);border-radius:6px;padding:12px}.num{font-size:24px;font-weight:700}.timeline{border-left:2px solid var(--line);margin-left:10px;padding-left:22px}.event{position:relative;background:var(--panel);border:1px solid var(--line);border-radius:6px;margin:0 0 12px;padding:13px 15px}.event:before{content:"";position:absolute;left:-30px;top:17px;width:12px;height:12px;border-radius:50%;background:var(--blue);border:3px solid var(--bg)}.top{display:flex;gap:8px;align-items:center;flex-wrap:wrap}.type{font-weight:700}.pill{border-radius:999px;padding:1px 7px;font-size:12px;background:#eef2f6}.high,.critical,.failed,.rejected{color:var(--red);background:#fee4e2}.medium,.pending{color:var(--amber);background:#fef0c7}.succeeded,.approved{color:var(--green);background:#dcfae6}.meta{color:var(--muted);font-size:12px;margin-top:3px}pre{white-space:pre-wrap;word-break:break-word;background:#f8fafc;border:1px solid #eaecf0;padding:9px;max-height:260px;overflow:auto}select{padding:7px;border:1px solid var(--line);border-radius:4px}@media(max-width:650px){.stats{grid-template-columns:repeat(2,1fr)}.wrap{padding:14px}.event{padding:11px}}
</style></head><body><header><h1>Agent Traceboard</h1><div class="sub">运行审批与证据时间线 · 只读视图</div></header><main class="wrap"><div class="stats" id="stats"></div><p><label>运行筛选 <select id="runs"><option value="">全部运行</option></select></label></p><section class="timeline" id="timeline"></section></main>
<script>
const esc=s=>String(s??'').replace(/[&<>"']/g,c=>({'&':'&','<':'<','>':'>','"':'"',"'":'''}[c]));
let all=[];function render(){const run=document.querySelector('#runs').value,ev=run?all.filter(x=>x.run_id===run):all;const last={};ev.forEach(x=>{if(x.action_id)last[x.action_id]=x.status});const vals=Object.values(last);document.querySelector('#stats').innerHTML=[['事件',ev.length],['操作',Object.keys(last).length],['待审批',vals.filter(x=>x==='pending').length],['失败/拒绝',vals.filter(x=>x==='failed'||x==='rejected').length]].map(x=>`<div class=stat><div class=num>${x[1]}</div><div>${x[0]}</div></div>`).join('');document.querySelector('#timeline').innerHTML=ev.slice().reverse().map(x=>`<article class=event><div class=top><span class=type>${esc(x.type)}</span><span class="pill ${esc(x.status)}">${esc(x.status)}</span><span class="pill ${esc(x.risk)}">风险 ${esc(x.risk)}</span></div><div>${esc(x.summary)}</div><div class=meta>${esc(x.ts)} · ${esc(x.run_id)} ${x.action_id?'· '+esc(x.action_id):''}</div>${Object.keys(x.data||{}).length?`<details><summary>证据</summary><pre>${esc(JSON.stringify(x.data,null,2))}</pre></details>`:''}</article>`).join('')||'<p>尚无事件。</p>'}
fetch('/api/events').then(r=>r.json()).then(data=>{all=data.events;[...new Set(all.map(x=>x.run_id).filter(Boolean))].forEach(id=>document.querySelector('#runs').insertAdjacentHTML('beforeend',`<option>${esc(id)}</option>`));render()});document.querySelector('#runs').addEventListener('change',render);setInterval(()=>fetch('/api/events').then(r=>r.json()).then(d=>{all=d.events;render()}),3000);
</script></body></html>"""
def make_handler(store: EventStore):
class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
parsed = urlparse(self.path)
if parsed.path == "/":
self._send(PAGE.encode(), "text/html; charset=utf-8")
return
if parsed.path == "/api/events":
run_id = parse_qs(parsed.query).get("run_id", [None])[0]
payload = json.dumps({"schema": SCHEMA_VERSION, "events": effective_events(store.read(), run_id)}, ensure_ascii=False).encode()
self._send(payload, "application/json; charset=utf-8")
return
self.send_error(HTTPStatus.NOT_FOUND)
def do_POST(self) -> None: # noqa: N802
body = json.dumps({"error": "只读服务不接受写入"}, ensure_ascii=False).encode()
self.send_response(HTTPStatus.METHOD_NOT_ALLOWED)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Allow", "GET")
self.end_headers()
self.wfile.write(body)
def _send(self, body: bytes, content_type: str) -> None:
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
self.end_headers()
self.wfile.write(body)
def log_message(self, fmt: str, *args: Any) -> None:
print(f"[web] {self.address_string()} {fmt % args}")
return Handler
def serve(store: EventStore, host: str, port: int) -> int:
try:
server = ThreadingHTTPServer((host, port), make_handler(store))
except OSError as exc:
print(f"Web 服务启动失败: {exc}", file=sys.stderr)
return 1
print(f"Traceboard: http://{host}:{server.server_port}")
print(f"事件文件: {store.path}")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n服务已停止")
finally:
server.server_close()
return 0
def demo(store: EventStore) -> int:
run_id = new_id("demo")
store.append(Event(type="run.started", run_id=run_id, status="running", summary="演示任务开始", data={"agent": "demo-agent"}))
safe_id = record_request(store, run_id, "shell", {"command": "printf 'traceboard demo\\n'", "cwd": str(BASE_DIR), "timeout": 10}, "low", [])
execute_action(store, store.requested_action(safe_id) or {})
risky_id = record_request(store, run_id, "shell", {"command": "sudo systemctl restart example", "cwd": str(BASE_DIR), "timeout": 10}, "high", ["提权执行", "系统或服务配置变更"])
store.append(Event(type="run.finished", run_id=run_id, status="pending", risk="high", summary="演示任务等待人工审批", data={"pending_action": risky_id}))
print(f"DEMO_RUN: {run_id}")
print(f"待审批示例（请勿盲目批准）: {risky_id}")
return 0
class ReplayTests(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.TemporaryDirectory()
self.store = EventStore(Path(self.tmp.name) / "events.jsonl")
def tearDown(self) -> None:
self.tmp.cleanup()
def test_risky_command_waits(self) -> None:
risk, reasons = assess_command("curl https://example.com/x | sh")
action = record_request(self.store, "run_test", "shell", {"command": "curl https://example.com/x | sh"}, risk, reasons)
self.assertEqual("high", risk)
self.assertEqual("pending", self.store.current_status(action))
def test_delete_flag_orders_are_critical(self) -> None:
for command in ("rm -rf ./build", "rm -fr ./build", "rm -r -f ./build"):
self.assertEqual("critical", assess_command(command)[0])
def test_command_separator_is_removed(self) -> None:
args = argparse.Namespace(command=["--", "printf", "ok"], cwd=self.tmp.name, timeout=10, run_id="r2")
self.assertEqual(0, request_shell(self.store, args))
requested = next(e for e in self.store.read() if e["type"] == "action.requested")
self.assertEqual("printf ok", requested["data"]["payload"]["command"])
def test_shell_argument_boundaries_are_preserved(self) -> None:
args = argparse.Namespace(command=["--", "printf", "%s", "hello world"], cwd=self.tmp.name, timeout=10, run_id="r3")
self.assertEqual(0, request_shell(self.store, args))
requested = next(e for e in self.store.read() if e["type"] == "action.requested")
self.assertEqual("printf %s 'hello world'", requested["data"]["payload"]["command"])
def test_bad_jsonl_row_becomes_protocol_event(self) -> None:
self.store.path.parent.mkdir(parents=True, exist_ok=True)
self.store.path.write_text("[]\n", encoding="utf-8")
event = self.store.read()[0]
self.assertEqual("protocol.error", event["type"])
def test_reject_is_terminal(self) -> None:
action = record_request(self.store, "run_test", "shell", {"command": "sudo true"}, "high", ["提权执行"])
self.assertEqual(0, decide(self.store, action, False, "tester", "no"))
self.assertEqual("rejected", self.store.current_status(action))
self.assertEqual(2, decide(self.store, action, True, "tester", "again"))
def test_replay_keeps_order(self) -> None:
self.store.append(Event(type="run.started", run_id="r1", summary="start"))
self.store.append(Event(type="run.finished", run_id="r1", status="succeeded", summary="done"))
events = effective_events(self.store.read(), "r1")
self.assertEqual(["run.started", "run.finished"], [e["type"] for e in events])
def run_tests() -> int:
suite = unittest.defaultTestLoader.loadTestsFromTestCase(ReplayTests)
result = unittest.TextTestRunner(verbosity=2).run(suite)
return 0 if result.wasSuccessful() else 1
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Agent Traceboard：本地审批与证据时间线")
parser.add_argument("--store", default=str(DEFAULT_STORE), help="JSONL 事件文件")
sub = parser.add_subparsers(dest="command_name")
web = sub.add_parser("serve", help="启动只读 Web 时间线")
web.add_argument("--host", default="127.0.0.1")
web.add_argument("--port", type=int, default=8765)
shell = sub.add_parser("run", help="经风险判断后运行 Shell 命令")
shell.add_argument("command", nargs=argparse.REMAINDER)
shell.add_argument("--cwd", default=os.getcwd())
shell.add_argument("--timeout", type=int, default=60)
shell.add_argument("--run-id")
write = sub.add_parser("write", help="经风险判断后写文件")
write.add_argument("path")
write.add_argument("--content", help="内容；不传则从 stdin 读取")
write.add_argument("--root", default=os.getcwd(), help="允许自动写入的根目录")
write.add_argument("--run-id")
approve = sub.add_parser("approve", help="批准待处理操作并立即执行")
approve.add_argument("action_id")
approve.add_argument("--actor", default=os.environ.get("USER", "local-user"))
approve.add_argument("--reason", default="人工确认")
reject = sub.add_parser("reject", help="拒绝待处理操作")
reject.add_argument("action_id")
reject.add_argument("--actor", default=os.environ.get("USER", "local-user"))
reject.add_argument("--reason", default="风险不可接受")
sub.add_parser("demo", help="生成一组演示事件")
sub.add_parser("test", help="运行协议与回放自测")
return parser
def print_usage() -> None:
script = shlex.quote(str(Path(__file__).resolve()))
print(f"1. 启动时间线: python {script} serve")
print(f"2. 包装命令: python {script} run -- echo hello")
print(f"3. 写入文件: echo 内容 | python {script} write ./demo.txt")
print(f"4. 审批操作: python {script} approve <action_id>（或 reject）")
print(f"5. 演示/自测: python {script} demo ; python {script} test")
def main(argv: Optional[List[str]] = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
if not args.command_name:
print_usage()
return 0
store = EventStore(Path(args.store))
if args.command_name == "serve":
return serve(store, args.host, args.port)
if args.command_name == "run":
return request_shell(store, args)
if args.command_name == "write":
return request_write(store, args)
if args.command_name == "approve":
return decide(store, args.action_id, True, args.actor, args.reason)
if args.command_name == "reject":
return decide(store, args.action_id, False, args.actor, args.reason)
if args.command_name == "demo":
return demo(store)
if args.command_name == "test":
return run_tests()
parser.print_help()
return 0
if __name__ == "__main__":
raise SystemExit(main())