feat(agent-runtime): unify agent bindings and sessions

This commit is contained in:
2026-10-12 13:22:47 +08:00
parent 32da57d19b
commit 558c6facc2
15 changed files with 759 additions and 8 deletions
+38
View File
@@ -0,0 +1,38 @@
# agent-runtime
统一维护项目的 Agent profile、Orca 执行环境、worktree 和跨 session terminal。它让
`builder.android`、`builder.ios` 等 binding 成为可配置路由,而不是写死在 skill 代码里的
枚举。
## 什么时候使用
- 给不同 builder/tester/ACK 绑定不同机器、项目目录、Agent CLI 或模型。
- 创建远程 Mac/本地 worktree,并需要后续 session 继续查看构建。
- 检查 `.pouch/agent.yaml` 是否能被所有 adapter 安全读取。
## 使用前准备
- 项目根已有 `.pouch/agent.yaml`。
- 远程 environment 已通过 Orca 配对,且 project/worktree 已存在或被明确创建。
- 机器路径、project id/name 和模型信息由使用者确认;配置不包含凭据。
## 使用示例
```text
把 builder.android 绑定到本地 app,把 builder.ios 绑定到 macmini,创建 iOS worktree,
然后让 builder 继续使用同一个 session 构建。
```
## Agent 会做什么
1. 校验并解析通用配置。
2. 按 environment、host、project、绝对 path 精确定位 Orca worktree。
3. 按需要创建持久 terminal,并记录可显式恢复的 run receipt。
不会自动复用旧 terminal、执行配置中的自由 shell、读取 secrets 或上传产物。
## 如何判断完成
- `agent.py validate` 通过。
- 每个 binding 能唯一解析到目标 worktree,或明确报告需创建/登记的环境。
- session 可用 run id 查询、attach 和 stop。
+59
View File
@@ -0,0 +1,59 @@
---
name: agent-runtime
description: >-
维护项目通用 Agent 运行配置、环境绑定、Orca worktree 与跨 session terminal;
适用于 builder.android、builder.ios、tester.test、ack.dev 等可扩展绑定。
不负责具体构建、测试或 ACK 闭环。
---
# Agent Runtime:统一环境、Agent 与工作区
本 skill 是 builder、tester、ACK 和 iOS tester 共用的配置层。每个 Git 项目在自己的
根目录维护 `.pouch/agent.yaml`;父仓库与 submodule 各自拥有配置,不跨仓库继承。机器
专属覆盖放 `.pouch/agent.local.yaml`,运行记录放 `.pouch/agent/runs/`,两者都不提交。
## 配置模型
- `locations`:执行环境、Orca project、目标机器上的绝对路径;不放模型和命令。
- `profiles`:CLI、模型、reasoning effort、权限和档位;不放路径和 secrets。
- `bindings`:任意安全的 `namespace.name`,绑定一个 adapter、location、profile 和
adapter 目标。`builder.android`、`builder.ios`、`builder.harmony` 不需要修改代码即可并存。
- `adapter` 目前允许 `builder`、`tester`、`ack`。adapter 的具体执行规则仍由对应 skill
负责;本层只提供解析、worktree 和 session 生命周期。
- `reasoningEffort` 是规范字段;不要新增 `effect`、自由 `command`、`argv`、`shell`、
`env` 或 token 字段。
完整字段、合并和安全规则见 [references/config.md](references/config.md)。需要实际解析或
创建工作区时使用 `scripts/agent.py`,不要手工拼接 Orca JSON:
```bash
python3 <skill-dir>/scripts/agent.py validate --root <project-root>
python3 <skill-dir>/scripts/agent.py show --root <project-root> --binding builder.android
python3 <skill-dir>/scripts/agent.py worktree resolve --root <project-root> --binding builder.ios
python3 <skill-dir>/scripts/agent.py worktree create --root <project-root> \
--binding builder.ios --name writes-ios --base main
```
## Session
`builder` 绑定可通过统一入口创建持久 terminal;返回的 run 记录包含 binding、profile
hash、环境、worktree、terminal handle 和 source commit。跨 session 只能显式 `status`、
`attach`、`stop` 或 `resume`,不能自动复用不明 terminal。tester 的每轮 terminal 仍然
是临时的,ACK 仍然必须走其带 fingerprint/receipt 的 launcher。
```bash
python3 <skill-dir>/scripts/agent.py run start --root <project-root> --binding builder.android
python3 <skill-dir>/scripts/agent.py run status --root <project-root> --run <run-id>
python3 <skill-dir>/scripts/agent.py run attach --root <project-root> --run <run-id>
python3 <skill-dir>/scripts/agent.py run resume --root <project-root> --run <run-id>
python3 <skill-dir>/scripts/agent.py run stop --root <project-root> --run <run-id>
```
`run start` 只为 builder 生成受控的 make 命令;tester/ACK 的执行入口仍由对应 skill
适配器提供。环境、项目或精确 worktree 缺失时 fail closed,并报告需要补的字段。
## 验证
- `python3 <skill-dir>/scripts/agent.py validate --root <project-root>` 通过。
- remote binding 必须能按 `environment + host + path + project` 精确找到 Orca worktree。
- 新 binding 只增加配置和对应 adapter 支持,不复制一套 skill 或修改固定枚举。
+4
View File
@@ -0,0 +1,4 @@
interface:
display_name: "Agent Runtime"
short_description: "统一 Agent 环境、worktree、profile 与跨 session 运行配置"
default_prompt: "Use $agent-runtime to validate shared agent config, resolve or create a worktree, or inspect a persistent session."
+73
View File
@@ -0,0 +1,73 @@
# Agent Runtime 配置
## 文件与合并
提交 `.pouch/agent.yaml`。若存在 `.pouch/agent.local.yaml`,按 mapping 深度合并;local
文件只能覆盖同名字段,不能增加未声明的顶层结构。合并后的配置必须再次完整校验。
```yaml
schemaVersion: 1
locations:
local-app:
environment: local
host: local
project: git:git.yumee.top/bloggarden/app
path: /home/ace/code/draft2/blogger/app
remote-ios:
environment: macmini
host: runtime:14b6d8ce-4528-49f4-8d46-56b266a5ad69
project: git:git.yumee.top/bloggarden/app
path: /Users/laily/workrepo/bloggarden-app
profiles:
android-builder:
role: build
cli: codex
tier: standard
model: gpt-5
reasoningEffort: medium
permissionMode: workspace-write
ios-builder:
role: build
cli: codex
tier: strong
model: gpt-5
reasoningEffort: high
permissionMode: workspace-write
bindings:
builder.android:
adapter: builder
location: local-app
profile: android-builder
makefile: makefile.builder
target: android
session:
lifecycle: persistent
builder.ios:
adapter: builder
location: remote-ios
profile: ios-builder
makefile: makefile.builder
target: ios
session:
lifecycle: persistent
```
## 字段原则
- binding、location、profile ID 只允许小写字母、数字和 `-`,binding 用一个或多个点
分隔;ID 是路由名,不是任意 shell 文本。
- `environment: local` 使用当前 Orca runtime;远程环境使用已保存的名称。`host` 必须
是 `local` 或 `runtime:<environment-id>`,由 Orca 查询结果证明,不能凭主机名猜测。
- `project` 必须是 Orca project id 或精确 display name;`path` 必须是目标机器上的绝对
路径。解析 worktree 时四者必须同时匹配,标题、分支和 basename 不能消歧。
- `profile` 的 CLI 只能来自 ACK/tester 共同 allowlist:`codex`、`cursor-agent`、`grok`、
`omp`、`pi`。`model` 为安全模型 ID 或 null;`reasoningEffort` 为 `low|medium|high|xhigh`
或 null;权限只允许 `read-only|workspace-write`。
- builder 的 `target` 只能是 make target 名;`makefile` 只能是项目内相对文件名。脚本、
远端命令、环境变量和凭据由 builder/Orca adapter 提供,不能写入此配置。
- `session.lifecycle` 为 `persistent` 或 `ephemeral`。persistent 只表示允许显式跨
session attach/resume,不表示自动复用旧 terminal。
旧版 `.pouch/tester/local.yaml` 和 `.pouch/ack/agents.yaml` 继续兼容。迁移时应让它们
由 adapter 从 common profile/location 生成或覆盖,不能删除旧 receipt、fingerprint 或
tester 的每轮 terminal 清理语义。
+480
View File
@@ -0,0 +1,480 @@
#!/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"])
result = orca_json(["worktree", "list"], environment)
candidates = []
for item in result.get("worktrees", []):
if not isinstance(item, dict):
continue
if str(item.get("projectId")) != str(location["project"]):
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)