laily/.pouch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ack): support project agent routing overrides

Browse Source
This commit is contained in:
laily
2026-09-19 17:52:14 +08:00
parent 44cee4be91
commit f469c8f5f5
25 changed files with 746 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/ack/scripts/launch_worker.py
+126 -11
View File
@@ -45,13 +45,14 @@ from worker_profiles import ( # noqa: E402
executable_basename_matches_cli,
profile_hash,
render_worker_argv,
validate_routing_document,
validate_orchestration,
)
PROTOCOL_VERSION = LAUNCH_PROTOCOL_VERSION
RECEIPT_VERSION = 1
ENVIRONMENT_POLICY = "per-cli-allowlist-v1"
TASKS_RELATIVE_PATH = Path(".pouch/ack/tasks.yaml")
AGENTS_RELATIVE_PATH = Path(".pouch/ack/agents.yaml")
MAX_CONTROL_OUTPUT = 1024 * 1024
MAX_RECORD_SIZE = 256 * 1024
LAUNCH_TTL_SECONDS = 120
@@ -85,12 +86,49 @@ WORKER_CREDENTIAL_NAMES = {
"cursor-agent": frozenset({"CURSOR_API_KEY"}),
"grok": frozenset({"XAI_API_KEY"}),
"omp": frozenset({"OPENCODE_API_KEY"}),
"pi": frozenset(
{
"ANTHROPIC_API_KEY",
"ANT_LING_API_KEY",
"AZURE_OPENAI_API_KEY",
"OPENAI_API_KEY",
"DEEPSEEK_API_KEY",
"NVIDIA_API_KEY",
"GEMINI_API_KEY",
"MISTRAL_API_KEY",
"GROQ_API_KEY",
"CEREBRAS_API_KEY",
"CLOUDFLARE_API_KEY",
"CLOUDFLARE_ACCOUNT_ID",
"CLOUDFLARE_GATEWAY_ID",
"XAI_API_KEY",
"OPENROUTER_API_KEY",
"AI_GATEWAY_API_KEY",
"ZAI_API_KEY",
"ZAI_CODING_CN_API_KEY",
"OPENCODE_API_KEY",
"RADIUS_API_KEY",
"FIREWORKS_API_KEY",
"TOGETHER_API_KEY",
"BASETEN_API_KEY",
"KIMI_API_KEY",
"MINIMAX_API_KEY",
"MINIMAX_CN_API_KEY",
"QWEN_TOKEN_PLAN_API_KEY",
"QWEN_TOKEN_PLAN_CN_API_KEY",
"XIAOMI_API_KEY",
"XIAOMI_TOKEN_PLAN_CN_API_KEY",
"XIAOMI_TOKEN_PLAN_AMS_API_KEY",
"XIAOMI_TOKEN_PLAN_SGP_API_KEY",
}
),
}
CLI_TITLE_LABELS = {
"codex": "CODEX",
"cursor-agent": "CURSOR",
"grok": "GROK",
"omp": "OMP",
"pi": "PI",
}
INHERITED_ENVIRONMENT_PREFIXES = (
"LC_",
@@ -262,12 +300,55 @@ def authoritative_tasks_path(project_root: Path) -> Path:
return tasks_path
def optional_agents_path(project_root: Path) -> Path | None:
agents_path = project_root / AGENTS_RELATIVE_PATH
try:
metadata = os.lstat(agents_path)
except FileNotFoundError:
return None
except OSError as exc:
raise LaunchError(f"agents.yaml 不可访问: {agents_path}") from exc
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
raise LaunchError(f"agents.yaml 必须是普通文件且不能是 symlink: {agents_path}")
_assert_no_symlink_components(agents_path, "agents.yaml 路径")
if metadata.st_size > MAX_RECORD_SIZE:
raise LaunchError(f"agents.yaml 超过 {MAX_RECORD_SIZE} 字节上限")
return agents_path
def load_effective_orchestration(
project_root: Path,
board: dict,
) -> tuple[dict, str, str]:
"""Load the optional project override and return routing, hash, source."""
agents_path = optional_agents_path(project_root)
if agents_path is not None:
agents = load_document(agents_path)
errors = validate_orchestration(agents, where="agents.yaml")
if errors:
formatted = "\n".join(f" - {error}" for error in dict.fromkeys(errors))
raise LaunchError(f"agents.yaml 未通过 worker 路由校验:\n{formatted}")
return agents, canonical_sha256(agents), str(agents_path)
project = board.get("project")
orchestration = project.get("orchestration") if isinstance(project, dict) else None
if not isinstance(orchestration, dict):
raise LaunchError("任务板缺少 project.orchestration，且没有 agents.yaml")
return orchestration, canonical_sha256(orchestration), "tasks.yaml:project.orchestration"
def load_authoritative_board(project_root_value: str) -> tuple[Path, dict]:
project_root = canonical_directory(project_root_value, "项目根目录")
tasks_path = authoritative_tasks_path(project_root)
board = load_document(tasks_path)
errors = validate_builtin(board)
errors.extend(validate_routing_document(board))
effective, _, _ = load_effective_orchestration(project_root, board)
errors = validate_builtin(board, orchestration_override=effective)
if optional_agents_path(project_root) is not None:
project = board.get("project")
board_orchestration = project.get("orchestration") if isinstance(project, dict) else None
if board_orchestration is not None:
errors.extend(validate_orchestration(board_orchestration))
if errors:
formatted = "\n".join(f" - {error}" for error in dict.fromkeys(errors))
raise LaunchError(f"任务板未通过 worker 路由校验:\n{formatted}")
@@ -307,8 +388,25 @@ def _is_trusted_grok_executable(resolved: Path, metadata: os.stat_result) -> boo
return metadata.st_uid == os.getuid() and metadata.st_gid == os.getgid()
def _is_trusted_pi_shim(candidate: Path, resolved: Path, metadata: os.stat_result) -> bool:
"""Accept the mise-managed ``pi`` shim while retaining target identity checks."""
home, _ = account_identity()
expected_shim = home / ".local" / "share" / "mise" / "shims" / "pi"
if candidate != expected_shim or candidate.name != "pi":
return False
if resolved.name != "mise":
return False
if metadata.st_uid not in {0, os.getuid()} or stat.S_IMODE(metadata.st_mode) & 0o022:
return False
try:
return candidate.is_symlink() and candidate.resolve(strict=True) == resolved
except OSError:
return False
def resolve_executable(name: str) -> Path:
supported = {"codex", "cursor-agent", "grok", "omp", "git", "orca"}
supported = {"codex", "cursor-agent", "grok", "omp", "pi", "git", "orca"}
if name not in supported:
raise LaunchError(f"不支持的可执行文件: {name}")
search_paths = trusted_path_entries()
@@ -337,6 +435,10 @@ def resolve_executable(name: str) -> Path:
if _is_trusted_grok_executable(resolved, metadata):
return resolved
continue
if name == "pi":
if _is_trusted_pi_shim(candidate, resolved, metadata):
return candidate
continue
if stat.S_IMODE(metadata.st_mode) & 0o022:
continue
if resolved.name != name:
@@ -672,14 +774,13 @@ def build_plan(
raise LaunchError("slot 必须是 1..99 的整数")
project_root, board = load_authoritative_board(project_root_value)
orchestration, routing_hash, routing_source = load_effective_orchestration(
project_root, board
)
board_hash = canonical_sha256(board)
find_task(board, task_id)
project = board["project"]
orchestration = project.get("orchestration")
if not isinstance(orchestration, dict):
raise LaunchError("旧任务板缺少 project.orchestration，只能使用手动模式")
if orchestration.get("mode") != "orca":
raise LaunchError("project.orchestration.mode 不是 orca，拒绝自动创建 worker")
raise LaunchError("有效 worker 配置的 mode 不是 orca，拒绝自动创建 worker")
profiles = orchestration.get("profiles")
profile = profiles.get(profile_id) if isinstance(profiles, dict) else None
if not isinstance(profile, dict):
@@ -718,6 +819,11 @@ def build_plan(
profile,
profile_version=orchestration["profileVersion"],
)
routing_facts = (
{}
if routing_source == "tasks.yaml:project.orchestration"
else {"routingSource": routing_source, "routingHash": routing_hash}
)
created_for = {
"taskId": task_id,
"attemptId": attempt_id,
@@ -735,6 +841,7 @@ def build_plan(
"worktree": worktree,
"requested": requested,
"slot": slot,
**routing_facts,
}
)
try:
@@ -762,6 +869,7 @@ def build_plan(
"requested": requested,
"slot": slot,
"title": title,
**routing_facts,
}
@@ -1136,6 +1244,9 @@ def build_receipt(
},
"createdAt": created_at,
}
if "routingSource" in plan and "routingHash" in plan:
receipt["routingSource"] = plan["routingSource"]
receipt["routingHash"] = plan["routingHash"]
receipt["receiptHash"] = canonical_sha256(receipt)
return receipt
@@ -1579,8 +1690,10 @@ def build_parser() -> argparse.ArgumentParser:
def profile_hash_command(project_root_value: str, profile_id: str) -> dict:
if not PROFILE_ID_RE.fullmatch(profile_id):
raise LaunchError("profile-id 格式非法")
_, board = load_authoritative_board(project_root_value)
orchestration = board["project"].get("orchestration")
project_root, board = load_authoritative_board(project_root_value)
orchestration, routing_hash, routing_source = load_effective_orchestration(
project_root, board
)
profiles = orchestration.get("profiles") if isinstance(orchestration, dict) else None
profile = profiles.get(profile_id) if isinstance(profiles, dict) else None
if not isinstance(profile, dict):
@@ -1588,6 +1701,8 @@ def profile_hash_command(project_root_value: str, profile_id: str) -> dict:
return {
"profileVersion": orchestration["profileVersion"],
"profileId": profile_id,
"routingSource": routing_source,
"routingHash": routing_hash,
"profileHash": profile_hash(
profile,
profile_version=orchestration["profileVersion"],
skills/ack/scripts/validate_agents.py
Executable
+51
View File
@@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""Validate an optional ACK .pouch/ack/agents.yaml routing override."""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from validate_tasks import load_document, validate_with_schema
from worker_profiles import validate_orchestration
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="校验 ACK agents.yaml 结构")
parser.add_argument("agents", nargs="?", default="agents.yaml", help="agents.yaml 路径")
parser.add_argument("--schema", help="agents.schema.json 路径（默认自动探测）")
args = parser.parse_args(argv)
agents_path = Path(args.agents)
if not agents_path.is_file() or agents_path.is_symlink():
sys.stderr.write(f"agents.yaml 必须是普通文件且不能是 symlink: {agents_path}\n")
return 2
data = load_document(agents_path)
schema_path = Path(args.schema) if args.schema else (
Path(__file__).resolve().parent.parent / "templates" / "agents.schema.json"
)
errors = validate_orchestration(data, where="agents.yaml")
mode = "内置语义规则"
try:
import jsonschema # type: ignore # noqa: F401
if schema_path.is_file():
errors = validate_with_schema(data, schema_path) + errors
mode = f"schema ({schema_path.name}) + 内置语义规则"
except ImportError:
pass
if errors:
sys.stderr.write(f"agents.yaml 校验失败（{mode}），共 {len(errors)} 项：\n")
for error in dict.fromkeys(errors):
sys.stderr.write(f" - {error}\n")
return 1
sys.stdout.write(f"agents.yaml 校验通过（{mode}）：{agents_path}\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())
skills/ack/scripts/validate_tasks.py
+35 -3
View File
@@ -32,6 +32,7 @@ from yaml_subset import (
load_yaml_subset,
make_unique_pyyaml_loader,
)
from worker_profiles import validate_orchestration
from worker_profiles import validate_routing_document
STATUS_ENUM = {
@@ -894,7 +895,11 @@ def validate_with_schema(data: dict, schema_path: Path) -> list[str]:
return errors
def validate_builtin(data: dict) -> list[str]:
def validate_builtin(
data: dict,
*,
orchestration_override: dict | None = None,
) -> list[str]:
errors: list[str] = []
project_intake_workflow = "read-only-v1"
@@ -1044,6 +1049,8 @@ def validate_builtin(data: dict) -> list[str]:
if "ackVersion" in data and version_match is None:
errors.append("ackVersion 必须是合法 SemVer（例如 0.10.0）")
routing_required = (
orchestration_override is not None
or
(isinstance(project, dict) and "orchestration" in project)
or "workerReceipts" in data
or (
@@ -1053,7 +1060,12 @@ def validate_builtin(data: dict) -> list[str]:
)
)
if routing_required:
errors.extend(validate_routing_document(data))
errors.extend(
validate_routing_document(
data,
orchestration_override=orchestration_override,
)
)
if "summary" in data:
summary = data["summary"]
@@ -1415,7 +1427,27 @@ def main(argv: list[str] | None = None) -> int:
sys.stderr.write(f"找不到指定的 schema 文件: {schema_path}\n")
return 2
errors = validate_builtin(data)
agents_path = tasks_path.with_name("agents.yaml")
agents = None
if agents_path.is_symlink():
errors = ["agents.yaml 不能是 symlink"]
agents = None
elif agents_path.is_file():
agents = load_document(agents_path)
if not isinstance(agents, dict):
errors = ["agents.yaml 顶层必须是对象（mapping）"]
agents = None
else:
errors = validate_orchestration(agents, where="agents.yaml")
else:
errors = []
errors.extend(validate_builtin(data, orchestration_override=agents))
if agents is not None:
project = data.get("project")
board_orchestration = project.get("orchestration") if isinstance(project, dict) else None
if board_orchestration is not None:
errors.extend(validate_orchestration(board_orchestration))
mode = "内置语义规则"
try:
import jsonschema # type: ignore # noqa: F401
skills/ack/scripts/validate_worker_command.py
+1 -1
View File
@@ -13,7 +13,7 @@ import sys
MIGRATION_MESSAGE = (
"ACK v0.10 已停用自由 worker command 校验器；请在 "
".pouch/ack/tasks.yaml 的 project.orchestration 中声明 profile，并仅调用 "
".pouch/ack/agents.yaml（优先）或 .pouch/ack/tasks.yaml 的 project.orchestration 中声明 profile，并仅调用 "
"scripts/launch_worker.py profile-hash|plan|launch。"
)
skills/ack/scripts/worker_profiles.py
+77 -16
View File
@@ -24,7 +24,7 @@ LAUNCH_PROTOCOL_VERSION = 1
MAX_ROUNDS = 3
ROLES = frozenset({"developer", "test"})
CLIS = frozenset({"codex", "cursor-agent", "grok", "omp"})
CLIS = frozenset({"codex", "cursor-agent", "grok", "omp", "pi"})
TIERS = frozenset({"standard", "strong"})
REASONING_EFFORTS = frozenset({"low", "medium", "high", "xhigh"})
PERMISSION_MODES = frozenset({"read-only", "workspace-write"})
@@ -32,6 +32,7 @@ ORCHESTRATION_MODES = frozenset({"orca", "manual"})
DEFAULT_KEYS = frozenset({"developer", "test", "developerUpgraded"})
CLI_REQUIRES_REASONING_EFFORT = frozenset({"codex", "grok", "omp"})
CLI_REQUIRES_NULL_REASONING_EFFORT = frozenset({"cursor-agent"})
CLI_ALLOWS_DEFAULT_MODEL = frozenset({"pi"})
def _cli_choice_text() -> str:
@@ -95,7 +96,9 @@ RECEIPT_FIELDS = frozenset(
"receiptHash",
}
)
RECEIPT_CONTEXT_FIELDS = frozenset({"projectRoot", "boardHash"})
RECEIPT_CONTEXT_FIELDS = frozenset(
{"projectRoot", "boardHash", "routingSource", "routingHash"}
)
CREATED_FOR_FIELDS = frozenset({"taskId", "attemptId", "role"})
WORKTREE_FIELDS = frozenset(
{
@@ -247,8 +250,10 @@ def validate_profile(profile: Any, *, where: str = "profile") -> list[str]:
errors.append(f"{where}.cli: must be {_cli_choice_text()}")
if not isinstance(tier, str) or tier not in TIERS:
errors.append(f"{where}.tier: must be standard/strong")
if not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None:
errors.append(f"{where}.model: must be a safe model ID")
if not (isinstance(cli, str) and cli in CLI_ALLOWS_DEFAULT_MODEL and model is None) and (
not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None
):
errors.append(f"{where}.model: must be a safe model ID or null for Pi default")
if not isinstance(permission, str) or permission not in PERMISSION_MODES:
errors.append(
f"{where}.permissionMode: must be read-only/workspace-write"
@@ -265,6 +270,10 @@ def validate_profile(profile: Any, *, where: str = "profile") -> list[str]:
and effort is not None
):
errors.append(f"{where}.reasoningEffort: Cursor requires null")
elif isinstance(cli, str) and cli == "pi" and (
not isinstance(effort, str) or effort not in REASONING_EFFORTS
):
errors.append(f"{where}.reasoningEffort: Pi requires low/medium/high/xhigh")
if role == "test" and tier != "standard":
errors.append(f"{where}.tier: Test may only use standard")
@@ -312,15 +321,22 @@ def _validate_model_allowlist(value: Any, where: str) -> list[str]:
if not isinstance(models, list) or not models:
errors.append(f"{tier_where}: must be a non-empty model list")
continue
seen: set[str] = set()
seen: set[object] = set()
for index, model in enumerate(models):
item_where = f"{tier_where}[{index}]"
if not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None:
errors.append(f"{item_where}: must be a safe model ID")
elif model in seen:
if model is None and cli in CLI_ALLOWS_DEFAULT_MODEL:
model_key: object = ("default",)
elif not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None:
errors.append(
f"{item_where}: must be a safe model ID or null for Pi default"
)
continue
else:
model_key = model
if model_key in seen:
errors.append(f"{item_where}: duplicate model {model!r}")
else:
seen.add(model)
seen.add(model_key)
return errors
@@ -329,7 +345,7 @@ def _allowed_models(
cli: Any,
role: Any,
tier: Any,
) -> list[str] | None:
) -> list[Any] | None:
if not isinstance(allowlist, dict):
return None
roles = allowlist.get(cli)
@@ -519,6 +535,19 @@ def render_worker_argv(
"--no-session",
]
if cli == "pi":
# Pi's omitted --model selects its configured startup default. Its
# cwd is supplied by the trusted bootstrap process, because Pi has no
# --cwd flag. Read-only profiles are restricted to observation tools;
# workspace-write profiles use Pi's normal editing tool set.
argv = [executable]
if model is not None:
argv.extend(["--model", model])
argv.extend(["--thinking", profile["reasoningEffort"]])
if permission == "read-only":
argv.extend(["--tools", "read,grep,find,ls"])
return argv
if cli != "grok":
raise ValueError(f"unsupported cli: {cli}")
grok_permission = "plan" if permission == "read-only" else "acceptEdits"
@@ -597,8 +626,10 @@ def _validate_requested(value: Any, where: str) -> list[str]:
if not isinstance(tier, str) or tier not in TIERS:
errors.append(f"{where}.tier: must be standard/strong")
model = value.get("model")
if not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None:
errors.append(f"{where}.model: must be a safe model ID")
if not (isinstance(cli, str) and cli in CLI_ALLOWS_DEFAULT_MODEL and model is None) and (
not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None
):
errors.append(f"{where}.model: must be a safe model ID or null for Pi default")
effort = value.get("reasoningEffort")
if isinstance(cli, str) and cli in CLI_REQUIRES_REASONING_EFFORT and (
not isinstance(effort, str) or effort not in REASONING_EFFORTS
@@ -611,6 +642,10 @@ def _validate_requested(value: Any, where: str) -> list[str]:
and effort is not None
):
errors.append(f"{where}.reasoningEffort: Cursor requires null")
if isinstance(cli, str) and cli == "pi" and (
not isinstance(effort, str) or effort not in REASONING_EFFORTS
):
errors.append(f"{where}.reasoningEffort: Pi requires low/medium/high/xhigh")
if not isinstance(permission, str) or permission not in PERMISSION_MODES:
errors.append(f"{where}.permissionMode: must be read-only/workspace-write")
executable = value.get("executable")
@@ -697,6 +732,16 @@ def validate_worker_receipt(
not isinstance(board_hash, str) or SHA256_RE.fullmatch(board_hash) is None
):
errors.append(f"{where}.boardHash: must be a canonical sha256 hex digest")
routing_source = receipt.get("routingSource")
routing_hash = receipt.get("routingHash")
if (routing_source is None) != (routing_hash is None):
errors.append(f"{where}: routingSource and routingHash must be present together")
if routing_source is not None and not _is_nonempty_text(routing_source, max_length=512):
errors.append(f"{where}.routingSource: must be non-empty single-line text")
if routing_hash is not None and (
not isinstance(routing_hash, str) or SHA256_RE.fullmatch(routing_hash) is None
):
errors.append(f"{where}.routingHash: must be a canonical sha256 hex digest")
version = receipt.get("receiptVersion")
if version != RECEIPT_VERSION or isinstance(version, bool):
@@ -818,6 +863,9 @@ def validate_worker_receipt(
if project_root is not None and board_hash is not None:
facts["projectRoot"] = project_root
facts["boardHash"] = board_hash
if routing_source is not None and routing_hash is not None:
facts["routingSource"] = routing_source
facts["routingHash"] = routing_hash
expected_fingerprint = canonical_sha256(facts)
except ValueError:
errors.append(f"{where}.launchFingerprint: cannot hash launch facts")
@@ -1006,17 +1054,30 @@ def _validate_dispatch_links(
return errors
def validate_routing_document(data: Any) -> list[str]:
"""Validate routing and receipts embedded in an ACK tasks document."""
def validate_routing_document(
data: Any,
*,
orchestration_override: dict[str, Any] | None = None,
) -> list[str]:
"""Validate routing and receipts in an ACK tasks document.
``orchestration_override`` is the optional project-local agents.yaml
routing document. The task board remains the state source, while the
override becomes the effective profile source for receipt checks.
"""
if not isinstance(data, dict):
return ["<root>: must be an object"]
project = data.get("project")
if not isinstance(project, dict):
return ["project: must be an object"]
if "orchestration" not in project:
if "orchestration" not in project and orchestration_override is None:
return ["project.orchestration: is required"]
orchestration = project.get("orchestration")
orchestration = (
orchestration_override
if orchestration_override is not None
else project.get("orchestration")
)
errors = validate_orchestration(orchestration)
task_ids: set[str] = set()