feat: add ORC engineering orchestration #6

Merged
laily merged 2 commits from support-release into main 2026-08-01 18:01:26 +08:00
31 changed files with 1757 additions and 871 deletions
+1
View File
@@ -59,6 +59,7 @@ AGENTS.md # 本文档
| Skill | 说明 |
| ---------------------------------------------------------------------- | ------------------------------------------------- |
| [orc](skills/orc/SKILL.md) | ORC 入口:显式编排开发、版本发布与产物任务,支持 Agent 分档 |
| [ack](skills/ack/SKILL.md) | ACK 入口:显式初始化、检查并运行项目三角色协作闭环 |
| [skiff](skills/skiff/SKILL.md) | 本项目工作流:创建、使用、反馈与更新 builtin skill |
| [declarative-openspec-loop](skills/declarative-openspec-loop/SKILL.md) | 声明式编程循环:用户提供校验方式,Agent 自动 propose/apply/校验并迭代直到通过 |
+1
View File
@@ -36,6 +36,7 @@ AGENTS.md # 详细规范与架构说明
| Skill | 说明 |
|-------|------|
| [orc](skills/orc/SKILL.md) | 显式编排开发、版本发布和产物任务,支持 low/mid/high Agent 档位 |
| [ack](skills/ack/SKILL.md) | 显式初始化、检查并运行 ACK 三角色协作及可选交付闭环 |
| [skiff](skills/skiff/SKILL.md) | 在项目中创建、安装、反馈和维护 builtin skill |
| [declarative-openspec-loop](skills/declarative-openspec-loop/SKILL.md) | 声明式编程循环:用户提供校验方式,Agent 自动迭代直到通过 |
+3 -28
View File
@@ -13,11 +13,6 @@ ACK 是一个显式调用的 Agent Skill,用三种独立角色运行工程协
构建 DEB 或镜像、发布产物、创建 PR,并在授权范围内部署。交付配置默认关闭,
稳定发布与生产部署始终保留人工批准点。
显式调用 `$ack` 也可以直接处理一次性交付:ACK 根据请求把源码发布/发布 PR、DEB、
Docker 镜像分别路由给 `manage-release``deb-publisher`
`publish-docker-image`。实际操作由独立 Operator 执行;它使用与 Test 相同的低成本
模型和 reasoning effort,但不复用 Test terminal,也不进入 Developer → Test 闭环。
## 安装
全局安装:
@@ -142,23 +137,6 @@ PR、产物摘要、部署目标、健康检查和日志引用写入 `tasks.yaml
profile 只能停在 `review_ready`;稳定发布或生产部署必须经过对应 approval 步骤。
具体契约见 `references/delivery.md`
## 一次性交付路由
直接发布时可以说:
```text
$ack 为 v1.4.0 创建 release PR,合并后打 tag。
$ack 把 1.4.0 的 amd64 DEB 发布到 testing 仓库。
$ack 把当前提交发布成 registry.example.com/team/app:1.4.0。
```
源码版本生命周期和一次性 PR/MR 路由到 `manage-release`DEB/APT 路由到
`deb-publisher`Docker/OCI 镜像路由到 `publish-docker-image`。普通 PR/MR 只执行
`manage-release``PR-only` 流程,不会被误当成完整版本发布;功能任务 verified 后
按 profile 自动开 PR 时仍使用 ACK 原有的 `pull-request` 动作。一次性交付要求任务板配置
`defaults.operator`;当前模板已经提供,并强制其 CLI、standard 模型和 reasoning
effort 与 Test default 相同。详见 `references/delivery-routing.md`
## 启动 Worker
worker 的机器配置位于 `tasks.yaml.project.orchestration`:项目显式维护模型
@@ -195,8 +173,7 @@ argv,忽略调用者 PATH、使用环境 allowlist,并验证真实 Git workt
通过仓库外的单次启动记录、terminal-bound nonce/proof 和受限 bootstrap 调用 Orca。
返回的 receipt 含 `runtimeId`、handle、incarnation、profile hash、slot 和 worktree
identity。Coordinator 将 receipt 追加到顶层 `workerReceipts`,再把 receipt ID
写入任务的 `dispatch.developer``dispatch.test``dispatch.operator`,并同步写入
本轮 `attemptId`
写入任务的 `dispatch.developer``dispatch.test`,并同步写入本轮 `attemptId`
校验器要求 receipt 与当前 ACK task、角色、profile 和 attempt 完全一致;历史 receipt
不能跨任务或跨轮次改挂。
@@ -246,7 +223,5 @@ ACK 会自动读取 `delivery.yaml`,无需再逐步提醒它构建、上传、
当前 Skill 版本见 `VERSION`。新项目在 `tasks.yaml` 中以合法 SemVer 记录
`ackVersion`。从 `0.10.0` 起,`project.orchestration` 与顶层 `workerReceipts` 必须
同时存在;从 `0.11.0` 起,新项目还会生成默认关闭的 `delivery.yaml`,并在任务板声明
`project.deliveryFile``deliveryRuns`;从 `0.12.0` 起,新模板包含与 Test 同档模型
Operator,用于一次性交付路由。旧项目可以不迁移而继续使用原闭环,但补齐 Operator
前不能使用一次性交付。旧项目的 `kitVersion` 可以继续读取,但建议迁移为
`ackVersion`
`project.deliveryFile``deliveryRuns`。旧项目可以不迁移而继续使用原闭环。旧项目
`kitVersion` 可以继续读取,但建议迁移为 `ackVersion`
+5 -39
View File
@@ -3,8 +3,7 @@ name: ack
description: >-
初始化、检查并运行 ACK 三角色协作闭环。仅在用户显式调用 /ack 或 $ack,并要求
初始化 ACK、检查 docs/ack 配置、按 ACK 规划需求、指挥 Coordinator/Developer/Test
工作配置并执行任务验证后的项目交付流程,或把一次性 PR/发布请求路由给
manage-release、deb-publisher、publish-docker-image 普通 worker 时使用。
工作,或配置并执行任务验证后的项目交付流程时使用。
---
# ACK 项目协作入口
@@ -23,8 +22,6 @@ description: >-
- 用户要求检查 ACK 是否可用、配置是否完整:执行“检查”。
- 用户要求用 ACK 做需求、修复问题或继续任务:执行“工作”。
- 用户用自然语言要求增加、修改或关闭项目交付流程:执行“交付配置维护”。
- 用户直接要求创建发布 PR/MR、发布源码版本、DEB 或 Docker 镜像:执行
“一次性交付路由”。
始终先解析真实项目根目录。优先使用 `git rev-parse --show-toplevel`;不是 Git
项目时使用用户指定目录或当前目录。不要修改项目的 `AGENTS.md``CLAUDE.md`
@@ -106,43 +103,12 @@ description: >-
6. 检查知识引用能解析到固定 revision,candidate 仍留在任务证据中,且
`stale`、`superseded` 和 `archived` 不会被当作可派发的 `active` 知识。
7. 若存在 `project.orchestration`,检查 profile、model allowlist、默认 profile、
允许 worktree、顶层 `workerReceipts` 与 `dispatch.developer/test/operator` 的引用;receipt
允许 worktree、顶层 `workerReceipts` 与 `dispatch.developer/test` 的引用;receipt
必须绑定当前 ACK task、同一 role/profile/attempt`receiptId` 与 `attemptId`
必须同时为空或同时填写。
缺少结构化路由的旧任务板只能使用手动模式,不能自动创建 worker。
`defaults.operator` 对旧项目可选;存在时必须是 standard profile,且 CLI、模型与
reasoning effort 必须和 `defaults.test` 相同。缺少它只表示一次性交付路由不可用,
不影响原 Developer/Test 闭环。
8. 检查不会自动修复或覆盖现有配置;用户明确要求修复后再修改。
## 一次性交付路由
1. 读取 `references/delivery-routing.md` 并按用户原始请求分类。源码版本、release/hotfix
分支、普通或发布 PR/MR、tag 或 Forge Release 使用 `manage-release`;普通 PR/MR
只授权它的 `PR-only` 流程,不推断版本升级、合并或 tag。DEB/APT 使用
`deb-publisher`Docker/OCI registry 使用 `publish-docker-image`。
2. 若 `docs/ack/tasks.yaml` 不存在,或结构化 orchestration 没有通过校验的
`defaults.operator`,停止并建议初始化或按当前模板升级。不要由 Coordinator 亲自
执行,也不要猜模型。用户请求有多个合理路由且无法从项目事实唯一确定时,只问一个
最小澄清问题。
3. 在权威任务板新增 `type: delivery-operation` 的最小操作记录,`operation.skill` 保存
精确 Skill 名,`operation.request` 保留本次用户请求的授权语义;若原文含凭据值,
必须替换为 `[REDACTED]`,不能落盘。`dispatch.operator` 使用
`defaults.operator`。不为一次性交付写 PRD,也不进入 Developer → Test 闭环,
不读取或修改 `delivery.yaml`。
4. 运行 `validate_tasks.py`,再只用 `launch_worker.py plan|launch --role operator`
创建 fresh worker。审阅并绑定 fingerprint/receipt 后,把
`references/delivery-routing.md` 的 Operator prompt 和原始请求投递给 worker
Orca 模式同时读取 `references/orca-adapter.md`,登记单一操作任务与 dispatch。
Operator 使用与 Test 相同的低成本模型/effort,但它是独立角色和独立终端,不复用
Test worker。
5. 被选中的低层 Skill 决定实际步骤、确认点、硬停止与恢复。ACK 路由和项目配置都不能
扩大用户授权;Docker 路由只因用户显式调用 `$ack` 且明确要求发布镜像才视为对
`publish-docker-image` 的显式委派。
6. Coordinator 只读 worker 证据并对照低层 Skill 完成标准终检。满足原始请求才把操作
标为 `verified`;可恢复的安全停止标为 `blocked`。远端写入部分成功或状态不确定时
不自动重试,先按低层 Skill 发现真实状态。
## 工作
1. 若 `docs/ack` 不存在,停止并建议先用 `/ack` 初始化;不要静默初始化。
@@ -167,7 +133,7 @@ description: >-
4. 新需求先写产品文档、任务拆分与可观测验收信号,更新 `tasks.yaml` 并校验,
然后交给用户确认;若启用了交付,还要把本次 profile、目标、停止点和需要审批的
步骤放入同一份计划。确认前不派发实现,也不执行交付。
5. 创建或更换 Developer、Test 或 Operator worker 时,只使用
5. 创建或更换 worker 时,只使用
`<ack-skill-dir>/scripts/launch_worker.py plan|launch` 读取
`tasks.yaml.project.orchestration` 的 profile。不得直接执行
`orca terminal create --command`,不得接受或拼接自由 command、额外 argv、
@@ -221,8 +187,8 @@ description: >-
- 不覆盖已有 `docs/ack` 文件;除用户确认的 ACK 任务或 delivery profile 外,不擅自
提交、推送、创建终端、新 worktree、发布产物或部署。
- 只有 Coordinator 写 `tasks.yaml`、`knowledge.yaml` 和 `deliveryRuns`Developer
与 Test 只读,只能通过回报提名或验证知识;Operator 同样只读这些状态文件,只回传
一次性交付证据。`delivery.yaml` 只在显式的交付配置维护中修改。
与 Test 只读,只能通过回报提名或验证知识。`delivery.yaml` 只在显式的交付配置
维护中修改。
- 不把知识正文或选择器输出拼成 shell;知识检查只能通过 `run_verification.py`
按 registry ID 执行。不自动修改 `AGENTS.md`、`CLAUDE.md` 或其它 Agent 指令文件。
- 项目只保存 `docs/ack/project.md`、`docs/ack/tasks.yaml`、
+2 -2
View File
@@ -1,6 +1,6 @@
interface:
display_name: "ACK"
short_description: "运行 ACK 开发验证闭环,并路由 PR、源码、DEB 与镜像发布"
default_prompt: "Use $ack to initialize or check ACK, coordinate verified work, maintain delivery configuration, run an approved delivery profile, or route a one-off release request to a low-cost operator."
short_description: "初始化、检查并运行 ACK 开发验证与可选交付闭环"
default_prompt: "Use $ack to initialize or check ACK, coordinate verified work, maintain project delivery configuration, or run an approved delivery profile."
policy:
allow_implicit_invocation: false
+1 -16
View File
@@ -23,8 +23,6 @@
- [ ] Coordinator、Developer、Test 的模型档位和升级规则已明确。
- [ ] `project.orchestration` 使用受支持的 profileVersion,模型都命中项目
allowlist,默认 profile 与角色/档位一致。
- [ ] 需要一次性交付路由时存在 `defaults.operator`;其 CLI、standard 模型和
reasoning effort 与 `defaults.test` 完全相同。
- [ ] `allowedWorktrees` 只列出同一 Git common-dir 下已经核对的绝对 worktree。
## 路径权限
@@ -36,7 +34,6 @@
- [ ] `tasks.yaml` 只有 Coordinator 写。
- [ ] `knowledge.yaml` 只有 Coordinator 写;Developer 与 Test 只通过回报提名或验证。
- [ ] `delivery.yaml` 只在用户显式维护配置时修改;Developer 与 Test 只读。
- [ ] Operator 只回传一次性交付证据,不写 ACK 项目状态文件。
## 任务板
@@ -61,16 +58,6 @@
- [ ] 每次运行固定 commit/config revision,证据写入 `tasks.yaml.deliveryRuns`
失败不会把已验证任务回退为失败。
## 可选一次性交付路由
- [ ] PR/MR 与源码发布、DEB、Docker 镜像分别路由到 `manage-release`
`deb-publisher``publish-docker-image`;普通 PR/MR 只使用 `PR-only`,没有扩大
到版本升级、合并、tag 或 Forge Release。
- [ ] `delivery-operation.operation.request` 保留用户原始请求,没有扩大授权。
- [ ] Operator 使用 `launch_worker.py --role operator` 创建的 fresh worker,没有复用
Test terminal,也没有让 Coordinator 代跑。
- [ ] 部分成功或外部状态不确定时按低层 Skill 恢复,没有盲目自动重试。
## 项目知识
- [ ] 新项目没有已验证知识时使用 `verificationRegistry: {}``entries: []`
@@ -96,15 +83,13 @@
`orca terminal create --command` 或自由 worker 命令。
- [ ] `launch` 使用刚审阅的 `plan.launchFingerprint` 作为
`--expected-launch-fingerprint`;漂移时重新 plan。
- [ ] Developer/Test/Operator 只使用 `read-only``workspace-write` profile
- [ ] Developer/Test 只使用 `read-only``workspace-write` profile
v0.10 没有 full-access、bypass、YOLO/force 或关闭 sandbox。
- [ ] v0.10 每次自动派发都启动 fresh worker;没有把历史 receipt checksum 或
Orca live metadata 当作自动复用授权。
- [ ] launcher receipt 已追加到顶层 `workerReceipts`,任务分别引用
`dispatch.developer``dispatch.test`;每个引用的 task/role/profile/attempt
都与当前任务、本轮 receipt 完全一致,没有用单一 handle 混淆两个角色。
- [ ] 一次性交付 receipt 引用 `dispatch.operator`,与当前操作、profile 和 attempt
完全一致。
- [ ] 多 worktree 场景只有一个权威 `tasks.yaml``knowledge.yaml`
- [ ] Test 使用的服务来自正确 worktree。
-108
View File
@@ -1,108 +0,0 @@
# ACK 一次性交付路由
本文件定义用户显式调用 `$ack` 后,直接要求创建发布相关 PR/MR、发布源码版本、
上传 DEB 或发布 Docker 镜像时的路由。它不要求先跑 Developer → Test 闭环,也不读取
或修改 `delivery.yaml`;项目已有的“verified 后按 profile 交付”仍按 `delivery.md`
执行。
## 1. 只按用户目标选择能力
| 用户目标 | 路由 | 不得顺带执行 |
| --- | --- | --- |
| 创建普通或发布 PR/MR,以及版本号、release/hotfix 分支、合并、正式 tag、Forge Release、恢复中断的源码发布 | `manage-release` | 未请求的版本升级、合并、tag、Forge Release、清理、普通产物上传 |
| 构建、检查、上传 `.deb`,发布到 APT/DEB 仓库 | `deb-publisher` | tag、Forge Release、Docker 镜像 |
| 构建并推送 Docker/OCI 镜像到 registry | `publish-docker-image` | 源码版本、DEB、额外 tag;用户说的 `docker-publisher` 视为这个已安装 Skill 的别名 |
“创建 PR/MR”路由到 `manage-release`,但必须区分权限上限:发布版本、release/hotfix
分支或版本升级对应的 PR/MR 走发布流程;普通 PR/MR 只走它的 ACK `PR-only` 入口,
不得推断版本升级、release 分支、合并、tag 或 Forge Release。ACK 功能任务在 verified
后按 profile 自动开 PR 时,仍使用 `delivery.md``pull-request` 动作,不重复建立
一次性操作。
只说“发布一下”时先从用户文字和项目发布入口判断目标。DEB、镜像和源码发布仍有两个
以上合理候选时,先问一个最小澄清问题;不要根据最近文件或历史命令猜。用户明确要求
多个产物时,为每个能力建立独立操作,按用户给出的顺序执行,不能由一个 Skill 暗中
扩大到另一个 Skill。
## 2. 用低成本 Operator 派发
一次性交付由 `operator` worker 执行,Coordinator 不亲自运行低层 Skill。机器事实仍
只读 `docs/ack/tasks.yaml.project.orchestration`
- `defaults.operator` 必须指向 `role: operator``tier: standard` 的 profile。
- Operator 默认 profile 的 `cli``tier``model``reasoningEffort` 必须与
`defaults.test` 完全相同;校验器会拒绝漂移。`permissionMode` 仍按项目需要显式写
`read-only``workspace-write`,发布通常需要后者。
- 旧项目没有 `defaults.operator` 时,普通 ACK 闭环仍可运行,但一次性交付路由必须
fail closed。先按当前模板补 operator allowlist/profile/default 并校验,不能临时猜
模型或让 Coordinator 代跑。
- launcher 只给 Operator 固定的发布凭据 allowlist`DEB_SERVER_URL``DEB_TOKEN`
`DEB_REPOSITORY``DEB_UPLOAD_PATH``SSH_AUTH_SOCK`Docker/Forge 使用 HOME
中自己的凭据存储。值不写入任务板或 receipt。缺少凭据时由低层 Skill 标记 blocked
不通过 prompt、自由 env 或命令行注入秘密。
路由后在权威任务板新增一条最小记录:
```yaml
- id: "DELIVERY-001"
type: "delivery-operation"
title: "publish one DEB"
status: "open"
operation:
skill: "deb-publisher"
request: "<保留用户本次请求的授权语义;凭据值必须替换为 [REDACTED]>"
dispatch:
operator:
profileId: "<defaults.operator>"
receiptId: null
attemptId: null
taskId: null
dispatchId: null
rounds: []
```
运行 `validate_tasks.py` 后,用 `launch_worker.py plan|launch` 创建 fresh worker,参数
使用 `--role operator`、本操作 ID 和 `<operation-id>-A1`。审阅 fingerprint,成功后
把 receipt 写入 `workerReceipts` 并与 `dispatch.operator` 精确绑定,再投递下节的
prompt。Orca 模式按 `orca-adapter.md` 登记单一操作任务,并把 runtime task/dispatch
ID 写入 `dispatch.operator`;不创建 Developer/Test 子任务链。不得复用 Test terminal
“同模型档位”不等于“同角色或同 worker”。
## 3. Operator 派发 Prompt
```text
你是 ACK 一次性交付 Operator。请在 <worktree> 执行以下原始请求:
<operation.request>
ACK 已选择且只授权你加载:$<operation.skill>
先完整读取该 Skill 及它要求的 references,再读取可信 base 上的项目发布规则。
边界:
- 原始请求是授权上限;除已脱敏的凭据值外保持原意,不得因 ACK 路由、项目配置或
历史操作扩大远端写权限。
- 严格遵守低层 Skill 的确认点、硬停止条件、恢复流程和完成标准。
- 不修改 docs/ack/tasks.yaml、knowledge.yaml 或 delivery.yaml。
- 不把 token、密码、私钥路径或认证配置写入消息、文件、提交或日志。
- 发生部分成功或外部状态不确定时停止自动重试,先按低层 Skill 核对真实状态。
完成或阻塞后只回传:所选 Skill、实际执行到的阶段、准确目标、非敏感证据、
未完成项和安全恢复动作。不要把命令已运行等同于发布已验证。
```
## 4. 状态与完成
Coordinator 在派发后把操作标为 `dispatched`,只读 Operator 的结构化证据做终检:
- `manage-release` 必须按其完成标准证明 PR/MR、merged commit、远端 tag 或 Forge
Release 中用户实际要求的最远阶段。
- `deb-publisher` 必须分别证明包元数据/摘要、上传接受和仓库可见性。
- `publish-docker-image` 必须证明完整镜像引用、platform、源 commit 和远端 digest
registry 无法查询时明确限制。
证据满足原始请求才把操作标为 `verified`;安全停止但可恢复时标为 `blocked` 并记录
恢复动作。`worker_done` 不等于完成。涉及远端写入的失败不自动进入第二轮;先由对应
低层 Skill 执行真实状态发现和恢复,再由用户决定是否继续。
Operator receipt 和操作证据写在该任务中,不写入 `deliveryRuns`。一次性交付不会因
项目其它任务未 `verified` 而被拦截,也不会触发 `delivery.yaml` 的 profile;两种流程
必须保持独立,避免同一请求被重复发布。
-4
View File
@@ -121,7 +121,3 @@ ACK 只负责读取项目交付契约、编排顺序、守住审批点并汇总
`manage-release` 仍是可独立使用、独立安装的能力;缺失时 ACK 使用契约中已审查的
项目 entrypoint,二者都不可用时把对应步骤标为 `blocked`。低层 skill 自身要求显式
调用时,ACK 不能绕过它的触发与授权边界。
用户显式调用 `$ack` 直接要求创建发布 PR/MR、发布源码版本、DEB 或镜像时,走
`delivery-routing.md` 的一次性交付流程,不要求其它 ACK 任务先 `verified`,也不创建
本文件定义的 `deliveryRuns`。两种流程不能在同一请求中隐式重复执行。
+1 -5
View File
@@ -82,9 +82,6 @@ docs/ack/
`project.orchestration` 与顶层 `workerReceipts` 必须同时存在。
- 从 `0.11.0` 起的新项目初始化包含默认关闭的交付契约;旧项目不要求为了版本号升级
自动补交付配置。
- 从 `0.12.0` 起的新项目 orchestration 模板包含可选一次性交付所需的 Operator;其
默认 CLI、standard 模型和 reasoning effort 与 Test default 相同。旧项目不补也能
运行原闭环,但不能使用一次性交付路由。
- `updatedAt` 使用当前带时区时间。
- `project.name``repoPath``devWorktree``overlayFile``knowledgeFile` 使用
真实值。
@@ -144,8 +141,7 @@ python3 <ack-skill-dir>/scripts/validate_delivery.py docs/ack/delivery.yaml \
- Developer 与 Test 的验证命令可执行。
- `project.orchestration` 的 profile/allowlist/defaults 通过校验,自动模式只允许
`read-only``workspace-write`;旧任务板未迁移时保持手动模式。
- 若配置 Operator,其 default 与 Test 使用相同 CLI、standard 模型和 effort。
- 顶层 `workerReceipts``dispatch.developer/test/operator` 的 task/role/profile/attempt
- 顶层 `workerReceipts``dispatch.developer/test` 的 task/role/profile/attempt
引用一致;`receiptId``attemptId` 同时为空或同时填写。持久 receipt 只作审计,
v0.10 不自动复用旧终端。
- 网站或 API 项目写清服务启动、重启和 Base URL。
+14 -25
View File
@@ -1,6 +1,6 @@
# 模型路由(稳定核心)
本文件是**三角色与可选 Operator 的默认模型档位、升级规则和 worker profile 选择规则**的单一规范源。
本文件是**三角色默认模型档位、升级规则和 worker profile 选择规则**的单一规范源。
目标是在不牺牲质量的前提下降低 token 和模型成本:把昂贵的强模型留给需要判断的
工作,把机械执行交给较弱模型。
@@ -16,7 +16,6 @@
| Coordinator (PM) | 强模型 | 需求拆解、验收信号设计、优先级、终检对齐意图、三轮失败复盘都需要高质量推理 |
| Test | 中低模型 | 按既定验收信号执行浏览器/API/脚本,主要做观察、记录、逐条 pass/fail |
| Developer | 中低模型(按任务升级) | 多数实现可照规格执行;跨系统、数据迁移、重复失败时再升级 |
| Operator(一次性交付) | 与 Test 相同的中低模型 | 按低层发布 Skill 执行机械步骤并回传证据,不承担范围决策 |
Coordinator 用强模型但**不亲自跑测试**(测试由 Test 承担),所以强模型的 token
花在思考和终检上,而不是反复点击页面、跑 smoke、复制日志。这一分工天然省 token,
@@ -37,7 +36,6 @@ Coordinator 用强模型但**不亲自跑测试**(测试由 Test 承担),
- Test:跑浏览器用例、API smoke、逐条比对期望与实际、产出证据。
- Developer:从清晰规格实现范围明确的任务、跑构建与单测、回报 worker_done。
- Operator:执行 `delivery-routing.md` 已选定的发布 Skill,按其停止点记录证据。
---
@@ -75,7 +73,7 @@ worker 路由的机器可读事实只保存在 `docs/ack/tasks.yaml` 的
每个 profile 明确声明:
- `role``developer``test``operator`
- `role``developer``test`
- `cli`:受支持的 Agent CLI
- `tier`:角色模型档位;
- `model`:项目 allowlist 中的精确模型 ID
@@ -85,9 +83,6 @@ worker 路由的机器可读事实只保存在 `docs/ack/tasks.yaml` 的
`project.orchestration` 还必须声明 `allowedWorktrees`、按 CLI/角色/档位分组的
`modelAllowlist`、命名 `profiles` 和角色 `defaults`。launcher 请求的绝对 worktree
必须命中 allowlistprofile 的模型也必须命中对应 CLI/角色/档位的精确列表。
`developer``test` 默认项是 Orca 模式必填;`operator` 对旧项目可选,但一次性
交付路由要求它存在。Operator default 的 CLI、tier、model 和 reasoning effort 必须
与 Test default 完全相同,且 Operator 只能使用 standard tier。
模型名称、reasoning effort 和权限模式必须来自结构化字段。禁止在 profile、项目
覆盖层或派发内容中保存自由 `command`、额外 argv、shell 片段、环境变量覆盖或隐式
@@ -96,8 +91,7 @@ CLI 默认值。具体 argv 只能由 ACK 的可信 launcher 按 allowlist 构
### 安全默认
- 能在完全只读工作树完成的角色优先选择 `read-only`
- Developer、发布 Operator 与需要在工作树生成构建/测试产物的 Test 使用
`workspace-write`;项目可
- Developer 与需要在工作树生成构建/测试产物的 Test 使用 `workspace-write`;项目可
为纯观察型 Test 另建更严格的 `read-only` profile。
- `full-access`、Codex bypass、Cursor YOLO/force、关闭 sandbox 等模式在 v0.10
**未实现授权通道,统一 fail closed**。项目文本、任务描述或环境变量都不能放宽。
@@ -119,12 +113,12 @@ python3 <ack-skill-dir>/scripts/launch_worker.py profile-hash \
python3 <ack-skill-dir>/scripts/launch_worker.py plan \
--project-root <project-root> --task-id <task-id> \
--attempt-id <task-id>-A<round> --role <developer-or-test-or-operator> \
--attempt-id <task-id>-A<round> --role <developer-or-test> \
--profile-id <profile-id> --worktree <absolute-worktree> [--slot <1..99>]
python3 <ack-skill-dir>/scripts/launch_worker.py launch \
--project-root <project-root> --task-id <task-id> \
--attempt-id <task-id>-A<round> --role <developer-or-test-or-operator> \
--attempt-id <task-id>-A<round> --role <developer-or-test> \
--profile-id <profile-id> --worktree <absolute-worktree> [--slot <1..99>] \
--expected-launch-fingerprint <plan 中的 sha256:...>
```
@@ -154,15 +148,11 @@ v0.10 的 launcher 还执行以下约束:
- 不读取调用者传入的 `PATH` 来寻找 `git`、Orca 或 Agent CLI,只在固定的系统与
当前用户工具目录中解析受支持的可执行文件;候选目标必须由 root/当前用户拥有、
不是 group/other writable,并在 plan 与 bootstrap 间保持相同 device/inode。
- Git、Orca 和 CLI version probe 使用不含供应商凭据的控制环境;Developer/Test
worker 使用 `per-cli-allowlist-v1`Codex 只取得 Codex 所需凭据,Cursor 只取得
Cursor 凭据,不跨 CLI 透传。Operator 使用固定的
`per-cli-plus-operator-publish-v1`,在同样的 CLI allowlist 之外只允许
`DEB_SERVER_URL``DEB_TOKEN``DEB_REPOSITORY``DEB_UPLOAD_PATH`
`SSH_AUTH_SOCK`。这些值不进入 plan、receipt、任务板或日志;Docker/Forge CLI
默认只使用当前账户 HOME 中自己的凭据存储。`DBUS_*``DISPLAY`
`WAYLAND_DISPLAY``XDG_RUNTIME_DIR``LD_*``PYTHON*``NODE_OPTIONS`
`CODEX_HOME``GIT_*`、任意 Docker 密码变量和调用者 PATH 均不会透传给 worker。
- Git、Orca 和 CLI version probe 使用不含供应商凭据的控制环境;worker 使用
`per-cli-allowlist-v1`Codex 只取得 Codex 所需凭据,Cursor 只取得 Cursor 凭据,
不跨 CLI 透传。`DBUS_*``DISPLAY``WAYLAND_DISPLAY``XDG_RUNTIME_DIR`
`LD_*``PYTHON*``NODE_OPTIONS``CODEX_HOME``GIT_*` 和调用者 PATH 均不会
透传给 worker。
- worktree 除了命中项目 allowlist、属于同一 Git common-dir,还必须精确出现在
`git worktree list --porcelain -z` 中;`.git` symlink 冒充的目录会失败。
- Orca 的 `--command` 只含固定 bootstrap 与随机 launch ID。终端创建后,父进程先
@@ -242,7 +232,7 @@ worker 自报或单独的 Orca live metadata 都不能把旧终端提升为可
强模型产出高密度、可复用的产物:需求、架构决策、验收信号、任务拆分、失败复盘。
中低模型消费这些产物,产出可核对的执行证据:测试结果、快照、API 响应、构建日志、
改动文件清单和发布目标摘要
改动文件清单。
这样把昂贵推理挡在重复执行之外。
@@ -250,7 +240,6 @@ worker 自报或单独的 Orca live metadata 都不能把旧终端提升为可
## 一句话
Coordinator 是脑,Test 是眼,Developer 是手;一次性交付 Operator 是按已选 Skill
操作发布系统的手。Coordinator 用强模型做判断,其余角色默认用安全、较低成本的结构化
profile;只有常规闭环卡住时才升级,并且所有 worker 都必须经 launcher 产生可核对的
receipt。
Coordinator 是脑,Test 是眼,Developer 是手。脑用强模型且不做机械测试,眼和手
默认用安全、较低成本的结构化 profile;只有常规闭环卡住时才升级,并且所有 worker
都必须经 launcher 产生可核对的 receipt。
+3 -9
View File
@@ -24,8 +24,7 @@ orca orchestration inbox --limit 20 --json
```
确认:Orca runtime 可达并取得本次 `runtimeId`Coordinator 终端存在;当前没有冲突
的活跃编排任务。Developer/Test worker 与一次性交付 Operator 都必须通过本次 launcher
新建,不要求预先存在。
的活跃编排任务。Developer/Test worker 必须通过本次 launcher 新建,不要求预先存在。
---
@@ -60,7 +59,7 @@ dispatch。未来若 Orca/ACP 增加启动参数 attestation,或 ACK 接入项
```bash
python3 <ack-skill-dir>/scripts/launch_worker.py plan \
--project-root <project-root> --task-id <task-id> \
--attempt-id <task-id>-A<round> --role <developer-or-test-or-operator> \
--attempt-id <task-id>-A<round> --role <developer-or-test> \
--profile-id <profile-id> --worktree <absolute-worktree> --slot <1..99>
```
@@ -70,7 +69,7 @@ python3 <ack-skill-dir>/scripts/launch_worker.py plan \
```bash
python3 <ack-skill-dir>/scripts/launch_worker.py launch \
--project-root <project-root> --task-id <task-id> \
--attempt-id <task-id>-A<round> --role <developer-or-test-or-operator> \
--attempt-id <task-id>-A<round> --role <developer-or-test> \
--profile-id <profile-id> --worktree <absolute-worktree> --slot <1..99> \
--expected-launch-fingerprint <plan 中的 sha256:...>
```
@@ -102,11 +101,6 @@ launcher fail closed,不能改用手写命令兜底。模型档位与升级规
该路径上走 `plan` -> 带 expected fingerprint 的 `launch`,并把 receipt 留作审计。
既有会话可以由用户直接操作,但不能进入 ACK v0.10 的自动派发信任路径。
一次性交付不创建 Developer/Test 子任务链。按 `delivery-routing.md` 建立一个
`delivery-operation`,使用 fresh Operator receipt 中的 handle 创建/登记单一 Orca
任务并投递 Operator prompt;把返回的 runtime task/dispatch ID 写入
`dispatch.operator`。低层 Skill 的完成或 blocked 回报仍由 Coordinator 读证据后落盘。
---
## 创建父任务
@@ -21,11 +21,6 @@ ACK 默认三个独立 Agent**Coordinator 只编排、Test 只验证、Develo
**模型档位(正交层)。** 三角色默认按成本分层:Coordinator 用强模型,Test 与 Developer 用中低模型,必要时升级。完整档位表与升级规则见 `model-routing.md`。Coordinator 用强模型但不跑测试,这一分工天然省 token 又不破坏「验证者 ≠ 实现者」。
**一次性交付 Operator 不进入三角色开发闭环。** 用户显式调用 `$ack` 直接要求发布时,
Coordinator 按 `delivery-routing.md` 选择低层 Skill,并把原始请求交给独立 Operator。
Operator 与 Test 使用相同低成本模型/effort,但不承担独立复测;它只执行被选 Skill、
遵守其授权边界并回传发布证据,且不得写 ACK 状态文件。
---
## 三角色能力清单(SSOT
@@ -151,11 +146,6 @@ failed_retest(累计 3 轮) -> leftover
三轮失败的处理细则见 `optimization-method.md` §「三轮失败策略」。
`type: delivery-operation` 使用独立的短状态流:`open -> dispatched -> verified | blocked`
这里的 `verified` 只表示 Coordinator 已按被路由 Skill 的完成标准核对一次性交付证据,
不表示代码经过 Test 复测,也不能作为普通功能任务的验证证据。远端部分成功不会自动
重派;恢复规则见 `delivery-routing.md`
## 交付状态(与任务状态正交)
任务进入 `verified` 后不再改写为发布或部署状态。可选交付的每次执行单独记录在
+10 -45
View File
@@ -41,7 +41,6 @@ from validate_tasks import load_document, validate_builtin # noqa: E402
from worker_profiles import ( # noqa: E402
LAUNCH_PROTOCOL_VERSION,
canonical_sha256,
environment_policy_for_role,
profile_hash,
render_worker_argv,
validate_routing_document,
@@ -49,6 +48,7 @@ from worker_profiles import ( # noqa: E402
PROTOCOL_VERSION = LAUNCH_PROTOCOL_VERSION
RECEIPT_VERSION = 1
ENVIRONMENT_POLICY = "per-cli-allowlist-v1"
TASKS_RELATIVE_PATH = Path("docs/ack/tasks.yaml")
MAX_CONTROL_OUTPUT = 1024 * 1024
MAX_RECORD_SIZE = 256 * 1024
@@ -82,15 +82,6 @@ WORKER_CREDENTIAL_NAMES = {
"codex": frozenset({"AZURE_OPENAI_API_KEY", "OPENAI_API_KEY"}),
"cursor-agent": frozenset({"CURSOR_API_KEY"}),
}
OPERATOR_CREDENTIAL_NAMES = frozenset(
{
"DEB_REPOSITORY",
"DEB_SERVER_URL",
"DEB_TOKEN",
"DEB_UPLOAD_PATH",
"SSH_AUTH_SOCK",
}
)
INHERITED_ENVIRONMENT_PREFIXES = (
"LC_",
)
@@ -193,23 +184,13 @@ def control_environment() -> dict[str, str]:
return _sanitized_environment(CONTROL_ENVIRONMENT_NAMES)
def worker_environment(cli: str, role: str | None = None) -> dict[str, str]:
def worker_environment(cli: str) -> dict[str, str]:
"""Return only the supported CLI's own credentials and common runtime data."""
credential_names = WORKER_CREDENTIAL_NAMES.get(cli)
if credential_names is None:
raise LaunchError(f"不支持的 worker CLI 环境: {cli}")
if role is not None:
try:
environment_policy_for_role(role)
except ValueError as exc:
raise LaunchError(f"不支持的 worker role 环境: {role}") from exc
role_credentials = (
OPERATOR_CREDENTIAL_NAMES if role == "operator" else frozenset()
)
return _sanitized_environment(
WORKER_ENVIRONMENT_NAMES | credential_names | role_credentials
)
return _sanitized_environment(WORKER_ENVIRONMENT_NAMES | credential_names)
def reject_duplicate_or_separator_args(argv: list[str]) -> None:
@@ -652,20 +633,15 @@ def build_plan(
raise LaunchError("task-id 只允许字母、数字、点、下划线和连字符")
if attempt_id not in {f"{task_id}-A1", f"{task_id}-A2", f"{task_id}-A3"}:
raise LaunchError("attempt-id 必须精确为 <task-id>-A1..A3")
if role not in {"developer", "test", "operator"}:
raise LaunchError("role 必须是 developer、test 或 operator")
if role not in {"developer", "test"}:
raise LaunchError("role 必须是 developer 或 test")
if not PROFILE_ID_RE.fullmatch(profile_id):
raise LaunchError("profile-id 格式非法")
if not isinstance(slot, int) or isinstance(slot, bool) or not 1 <= slot <= 99:
raise LaunchError("slot 必须是 1..99 的整数")
project_root, board = load_authoritative_board(project_root_value)
task = find_task(board, task_id)
is_delivery_operation = task.get("type") == "delivery-operation"
if role == "operator" and not is_delivery_operation:
raise LaunchError("operator 只能用于 delivery-operation 任务")
if role != "operator" and is_delivery_operation:
raise LaunchError("delivery-operation 任务只能由 operator 执行")
find_task(board, task_id)
project = board["project"]
orchestration = project.get("orchestration")
if not isinstance(orchestration, dict):
@@ -705,7 +681,7 @@ def build_plan(
"cliVersion": cli_version,
"argv": argv,
"argvHash": canonical_sha256(argv),
"environmentPolicy": environment_policy_for_role(role),
"environmentPolicy": ENVIRONMENT_POLICY,
}
current_profile_hash = profile_hash(
profile,
@@ -729,11 +705,7 @@ def build_plan(
}
)
cli_label = "CODEX" if profile["cli"] == "codex" else "CURSOR"
role_label = {
"developer": "DEV",
"test": "TEST",
"operator": "OP",
}[role]
role_label = "DEV" if role == "developer" else "TEST"
digest_short = launch_fingerprint.split(":", 1)[-1][:10]
title = (
f"ACK-{role_label}-{cli_label}-{str(profile['tier']).upper()}-"
@@ -1481,10 +1453,7 @@ def bootstrap_worker(launch_id: str) -> int:
rebuilt["requested"]["argv"],
shell=False,
cwd=rebuilt["worktree"]["path"],
env=worker_environment(
str(rebuilt["requested"]["cli"]),
str(rebuilt["role"]),
),
env=worker_environment(str(rebuilt["requested"]["cli"])),
)
current_record.update(
state="bootstrap-ready",
@@ -1519,11 +1488,7 @@ def add_launch_arguments(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--project-root", required=True)
parser.add_argument("--task-id", required=True)
parser.add_argument("--attempt-id", required=True)
parser.add_argument(
"--role",
required=True,
choices=("developer", "test", "operator"),
)
parser.add_argument("--role", required=True, choices=("developer", "test"))
parser.add_argument("--profile-id", required=True)
parser.add_argument("--worktree", required=True)
parser.add_argument("--slot", type=int, default=1)
+1 -58
View File
@@ -128,15 +128,8 @@ DISPATCH_FIELDS = {
"worker",
"developer",
"test",
"operator",
"rounds",
}
DELIVERY_OPERATION_FIELDS = {"skill", "request"}
DELIVERY_OPERATION_SKILLS = {
"manage-release",
"deb-publisher",
"publish-docker-image",
}
KNOWLEDGE_CANDIDATE_REQUIRED_FIELDS = {
"kind",
"title",
@@ -439,14 +432,11 @@ def validate_delivery_runs(
value: object,
task_statuses: dict[str, object],
errors: list[str],
*,
delivery_operation_ids: set[str] | None = None,
) -> None:
if not isinstance(value, list):
errors.append("deliveryRuns 必须是列表")
return
excluded_task_ids = delivery_operation_ids or set()
seen_run_ids: set[str] = set()
for index, run in enumerate(value):
where = f"deliveryRuns[{index}]"
@@ -486,11 +476,6 @@ def validate_delivery_runs(
for task_id in task_ids:
if task_id not in task_statuses:
errors.append(f"{where}.taskIds: 未知任务 {task_id!r}")
elif task_id in excluded_task_ids:
errors.append(
f"{where}.taskIds: deliveryRuns 不能引用 delivery-operation "
f"{task_id!r}"
)
elif task_statuses[task_id] != "verified":
errors.append(
f"{where}: delivery run 只能引用 verified 任务,"
@@ -822,30 +807,6 @@ def validate_builtin(data: dict) -> list[str]:
if not _nonempty_string(source.get("updatedAt")):
errors.append(f"{where}.source.updatedAt: 必须是非空字符串")
operation = task.get("operation")
if operation is not None:
if not isinstance(operation, dict):
errors.append(f"{where}.operation: 必须是对象")
else:
reject_unknown_fields(
operation,
DELIVERY_OPERATION_FIELDS,
f"{where}.operation",
errors,
)
skill = operation.get("skill")
if skill not in DELIVERY_OPERATION_SKILLS:
errors.append(
f"{where}.operation.skill: 必须是 "
"manage-release/deb-publisher/publish-docker-image"
)
if not _nonempty_string(operation.get("request")):
errors.append(f"{where}.operation.request: 必须保留非空用户请求")
if task.get("type") == "delivery-operation":
if not isinstance(operation, dict):
errors.append(f"{where}: delivery-operation 必须声明 operation")
validate_knowledge_fields(task, where, status, errors)
if "dispatch" not in task:
@@ -868,12 +829,6 @@ def validate_builtin(data: dict) -> list[str]:
nullable=True,
)
if task.get("type") == "delivery-operation" and (
not isinstance(dispatch, dict)
or not isinstance(dispatch.get("operator"), dict)
):
errors.append(f"{where}: delivery-operation 必须声明 dispatch.operator")
rounds = dispatch.get("rounds", [])
if not isinstance(rounds, list):
errors.append(f"{where}.dispatch.rounds: 必须是列表")
@@ -974,19 +929,7 @@ def validate_builtin(data: dict) -> list[str]:
for task in tasks
if isinstance(task, dict) and _nonempty_string(task.get("id"))
}
delivery_operation_ids = {
task["id"]
for task in tasks
if isinstance(task, dict)
and _nonempty_string(task.get("id"))
and task.get("type") == "delivery-operation"
}
validate_delivery_runs(
data["deliveryRuns"],
task_statuses,
errors,
delivery_operation_ids=delivery_operation_ids,
)
validate_delivery_runs(data["deliveryRuns"], task_statuses, errors)
return errors
+11 -71
View File
@@ -23,22 +23,13 @@ RECEIPT_VERSION = 1
LAUNCH_PROTOCOL_VERSION = 1
MAX_ROUNDS = 3
ROLES = frozenset({"developer", "test", "operator"})
REQUIRED_DEFAULT_ROLES = frozenset({"developer", "test"})
STANDARD_ONLY_ROLES = frozenset({"test", "operator"})
ROLES = frozenset({"developer", "test"})
CLIS = frozenset({"codex", "cursor-agent"})
TIERS = frozenset({"standard", "strong"})
REASONING_EFFORTS = frozenset({"low", "medium", "high", "xhigh"})
PERMISSION_MODES = frozenset({"read-only", "workspace-write"})
ORCHESTRATION_MODES = frozenset({"orca", "manual"})
BASE_ENVIRONMENT_POLICY = "per-cli-allowlist-v1"
OPERATOR_ENVIRONMENT_POLICY = "per-cli-plus-operator-publish-v1"
ENVIRONMENT_POLICIES = frozenset(
{BASE_ENVIRONMENT_POLICY, OPERATOR_ENVIRONMENT_POLICY}
)
DEFAULT_KEYS = frozenset(
{"developer", "test", "operator", "developerUpgraded"}
)
DEFAULT_KEYS = frozenset({"developer", "test", "developerUpgraded"})
ORCHESTRATION_FIELDS = frozenset(
{
@@ -193,16 +184,6 @@ def _is_positive_int(value: Any) -> bool:
return isinstance(value, int) and not isinstance(value, bool) and value > 0
def environment_policy_for_role(role: Any) -> str:
"""Return the fixed credential policy for a validated worker role."""
if role == "operator":
return OPERATOR_ENVIRONMENT_POLICY
if role in {"developer", "test"}:
return BASE_ENVIRONMENT_POLICY
raise ValueError("role must be developer/test/operator")
def _is_timestamp(value: Any) -> bool:
if not isinstance(value, str):
return False
@@ -230,7 +211,7 @@ def validate_profile(profile: Any, *, where: str = "profile") -> list[str]:
permission = profile.get("permissionMode")
if not isinstance(role, str) or role not in ROLES:
errors.append(f"{where}.role: must be developer/test/operator")
errors.append(f"{where}.role: must be developer/test")
if not isinstance(cli, str) or cli not in CLIS:
errors.append(f"{where}.cli: must be codex/cursor-agent")
if not isinstance(tier, str) or tier not in TIERS:
@@ -250,13 +231,8 @@ def validate_profile(profile: Any, *, where: str = "profile") -> list[str]:
elif cli == "cursor-agent" and effort is not None:
errors.append(f"{where}.reasoningEffort: Cursor requires null")
if (
isinstance(role, str)
and role in STANDARD_ONLY_ROLES
and tier != "standard"
):
label = "Test" if role == "test" else "Operator"
errors.append(f"{where}.tier: {label} may only use standard")
if role == "test" and tier != "standard":
errors.append(f"{where}.tier: Test may only use standard")
if tier == "strong" and role != "developer":
errors.append(f"{where}.tier: strong may only be used by Developer")
@@ -292,11 +268,8 @@ def _validate_model_allowlist(value: Any, where: str) -> list[str]:
errors.append(f"{role_where}: must not be empty")
for tier in sorted(set(tiers) - TIERS, key=repr):
errors.append(f"{role_where}: unknown tier {tier!r}")
if role in STANDARD_ONLY_ROLES and "strong" in tiers:
label = "Test" if role == "test" else "Operator"
errors.append(
f"{role_where}: {label} cannot define a strong allowlist"
)
if role == "test" and "strong" in tiers:
errors.append(f"{role_where}: Test cannot define a strong allowlist")
for tier, models in tiers.items():
tier_where = f"{role_where}.{tier}"
if tier not in TIERS:
@@ -410,7 +383,7 @@ def validate_orchestration(
for default_key in sorted(set(defaults) - DEFAULT_KEYS, key=repr):
errors.append(f"{where}.defaults: unknown key {default_key!r}")
if mode == "orca":
for role in sorted(REQUIRED_DEFAULT_ROLES - set(defaults)):
for role in sorted(ROLES - set(defaults)):
errors.append(f"{where}.defaults: missing role {role!r}")
for default_key, profile_id in defaults.items():
default_where = f"{where}.defaults.{default_key}"
@@ -434,22 +407,6 @@ def validate_orchestration(
if profile.get("permissionMode") not in PERMISSION_MODES:
errors.append(f"{default_where}: default profile has unsafe permissions")
operator_profile_id = defaults.get("operator")
test_profile_id = defaults.get("test")
operator_profile = valid_profiles.get(operator_profile_id)
test_profile = valid_profiles.get(test_profile_id)
if operator_profile_id is not None and test_profile is None:
errors.append(
f"{where}.defaults.operator: requires a valid Test default profile"
)
elif operator_profile is not None and test_profile is not None:
for field in ("cli", "tier", "model", "reasoningEffort"):
if operator_profile.get(field) != test_profile.get(field):
errors.append(
f"{where}.defaults.operator: operator default must use "
f"the Test default {field}"
)
return errors
@@ -545,7 +502,7 @@ def _validate_created_for(value: Any, where: str) -> list[str]:
errors.append(f"{where}.attemptId: must belong to taskId")
role = value.get("role")
if not isinstance(role, str) or role not in ROLES:
errors.append(f"{where}.role: must be developer/test/operator")
errors.append(f"{where}.role: must be developer/test")
return errors
@@ -614,10 +571,9 @@ def _validate_requested(value: Any, where: str) -> list[str]:
elif isinstance(argv, list) and all(isinstance(arg, str) for arg in argv):
if argv_hash != canonical_sha256(argv):
errors.append(f"{where}.argvHash: does not match argv")
if value.get("environmentPolicy") not in ENVIRONMENT_POLICIES:
if value.get("environmentPolicy") != "per-cli-allowlist-v1":
errors.append(
f"{where}.environmentPolicy: must be 'per-cli-allowlist-v1' or "
"'per-cli-plus-operator-publish-v1'"
f"{where}.environmentPolicy: must be 'per-cli-allowlist-v1'"
)
return errors
@@ -766,19 +722,6 @@ def validate_worker_receipt(
if binding.get("observedWorktreePath") != worktree.get("path"):
errors.append(f"{where}.binding.observedWorktreePath: does not match worktree.path")
if isinstance(created_for, dict) and isinstance(requested, dict):
try:
expected_environment_policy = environment_policy_for_role(
created_for.get("role")
)
except ValueError:
pass
else:
if requested.get("environmentPolicy") != expected_environment_policy:
errors.append(
f"{where}.requested.environmentPolicy: does not match role"
)
if (
isinstance(created_for, dict)
and isinstance(worktree, dict)
@@ -1034,10 +977,7 @@ __all__ = [
"RECEIPT_VERSION",
"LAUNCH_PROTOCOL_VERSION",
"MAX_ROUNDS",
"BASE_ENVIRONMENT_POLICY",
"OPERATOR_ENVIRONMENT_POLICY",
"canonical_sha256",
"environment_policy_for_role",
"profile_hash",
"receipt_hash",
"render_worker_argv",
-4
View File
@@ -33,7 +33,6 @@
- 派发 prompt 模板:`references/prompt-templates.md`
- Orca 编排命令(可选):`references/orca-adapter.md`
- 验证后交付与配置维护(可选):`references/delivery.md`
- 一次性源码/DEB/镜像发布路由(可选):`references/delivery-routing.md`
## Worker 路由
@@ -47,7 +46,6 @@ receipt 全部以 `docs/ack/tasks.yaml` 的 `project.orchestration` 与顶层
|------|------------|------|
| Developer | `codex-dev-standard` | standard |
| Test | `codex-test-standard` | standard |
| Operator | `codex-operator-standard` | standard(模型/effort 与 Test 相同) |
| Developer 升级 | `codex-dev-strong` | strong |
项目如改用 Cursor,应修改结构化 profile、allowlist 和 defaults,再运行任务板
@@ -104,8 +102,6 @@ Skill 的 `scripts/run_verification.py` 执行,不直接拼接 path/args。检
## 硬规则(其余见 references/
- 三角色独立:Coordinator 只编排、Test 只验证、Developer 只实现(验证者 ≠ 实现者)。
- 一次性交付 Operator 不进入三角色闭环;它使用 Test 同档模型执行被路由的低层 Skill,
只回传证据,不写 ACK 状态文件。
- 模型分层:Coordinator 用强模型且不亲自跑测试,Test/Developer 用中低模型,必要时升级(见 references/model-routing.md)。
- 自动 worker 只能由 ACK 的 `scripts/launch_worker.py` 按结构化 profile 启动;
禁止直接拼 `orca terminal create --command`,禁止 `command``extraArgs``env`
+4 -42
View File
@@ -249,8 +249,7 @@
"additionalProperties": false,
"properties": {
"developer": { "$ref": "#/definitions/modelTierAllowlist" },
"test": { "$ref": "#/definitions/modelTierAllowlist" },
"operator": { "$ref": "#/definitions/modelTierAllowlist" }
"test": { "$ref": "#/definitions/modelTierAllowlist" }
}
},
"modelAllowlist": {
@@ -275,7 +274,7 @@
"properties": {
"role": {
"type": "string",
"enum": ["developer", "test", "operator"]
"enum": ["developer", "test"]
},
"cli": {
"type": "string",
@@ -305,7 +304,6 @@
"properties": {
"developer": { "$ref": "#/definitions/profileId" },
"test": { "$ref": "#/definitions/profileId" },
"operator": { "$ref": "#/definitions/profileId" },
"developerUpgraded": { "$ref": "#/definitions/profileId" }
}
},
@@ -484,10 +482,7 @@
},
"environmentPolicy": {
"type": "string",
"enum": [
"per-cli-allowlist-v1",
"per-cli-plus-operator-publish-v1"
]
"const": "per-cli-allowlist-v1"
}
}
},
@@ -596,7 +591,7 @@
},
"role": {
"type": "string",
"enum": ["developer", "test", "operator"]
"enum": ["developer", "test"]
}
}
},
@@ -959,22 +954,6 @@
}
]
},
"deliveryOperation": {
"type": "object",
"required": ["skill", "request"],
"additionalProperties": false,
"properties": {
"skill": {
"type": "string",
"enum": [
"manage-release",
"deb-publisher",
"publish-docker-image"
]
},
"request": { "type": "string", "minLength": 1, "pattern": "\\S" }
}
},
"task": {
"type": "object",
"required": ["id", "title", "status"],
@@ -1020,7 +999,6 @@
"stepsToReproduce": { "type": "array", "items": { "type": "string" } },
"expected": { "type": "string" },
"actual": { "type": "string" },
"operation": { "$ref": "#/definitions/deliveryOperation" },
"evidence": { "type": "object" },
"verification": { "type": "object" },
"dispatch": {
@@ -1032,7 +1010,6 @@
"worker": { "type": ["string", "null"] },
"developer": { "$ref": "#/definitions/roleDispatch" },
"test": { "$ref": "#/definitions/roleDispatch" },
"operator": { "$ref": "#/definitions/roleDispatch" },
"rounds": {
"type": "array",
"items": { "$ref": "#/definitions/round" }
@@ -1070,21 +1047,6 @@
},
"required": ["resolution"]
}
},
{
"if": {
"properties": { "type": { "const": "delivery-operation" } },
"required": ["type"]
},
"then": {
"required": ["operation", "dispatch"],
"properties": {
"dispatch": {
"type": "object",
"required": ["operator"]
}
}
}
}
]
}
-19
View File
@@ -39,15 +39,11 @@ project:
strong: ["gpt-5.6-sol"]
test:
standard: ["gpt-5.6-luna"]
operator:
standard: ["gpt-5.6-luna"]
cursor-agent:
developer:
standard: ["auto"]
test:
standard: ["auto"]
operator:
standard: ["auto"]
profiles:
codex-dev-standard:
role: "developer"
@@ -63,13 +59,6 @@ project:
model: "gpt-5.6-luna"
reasoningEffort: "low"
permissionMode: "workspace-write"
codex-operator-standard:
role: "operator"
cli: "codex"
tier: "standard"
model: "gpt-5.6-luna"
reasoningEffort: "low"
permissionMode: "workspace-write"
codex-dev-strong:
role: "developer"
cli: "codex"
@@ -91,17 +80,9 @@ project:
model: "auto"
reasoningEffort: null
permissionMode: "workspace-write"
cursor-operator-standard:
role: "operator"
cli: "cursor-agent"
tier: "standard"
model: "auto"
reasoningEffort: null
permissionMode: "workspace-write"
defaults:
developer: "codex-dev-standard"
test: "codex-test-standard"
operator: "codex-operator-standard"
developerUpgraded: "codex-dev-strong"
workerReceipts: []
+2 -13
View File
@@ -15,16 +15,6 @@
## 常见用法
### 通过 ACK 只创建普通 PR
```text
$ack 把当前分支推送并创建一个指向 main 的 PR,不要合并。
```
ACK 会把这类一次性请求路由到 `manage-release` 的 PR-only 流程。Agent 只核对当前
head、base、remote、提交边界、验证和 Forge 账号,然后创建或复用 PR;不会顺带升级
版本、创建 release 分支、合并、打 tag 或创建 Forge Release。
### 确定下一个版本号
```text
@@ -116,6 +106,5 @@ Agent 会先发现项目自己的版本、分支和发布规则,再检查本
## 不适用的场景
普通功能开发、普通 worktree 或 PR/MR 操作、代码审查、构建 Docker 镜像或上传 DEB 包
不需要直接触发 `manage-release`。只有任务明确涉及版本发布、发布分支、版本号、发布
PR/MR 或 release tag,或者用户显式调用 `$ack` 要求 PR-only 委派时,才交给这个
Skill。
不需要触发 `manage-release`。只有任务明确涉及版本发布、发布分支、版本号、发布 PR/MR
或 release tag 时,才交给这个 Skill。
-17
View File
@@ -8,8 +8,6 @@ description: >-
升版本、提交或合并发布 PR/MR、打 release tag、完成发版、处理 hotfix 或继续未完成
发布时使用。只做普通编码、普通 worktree 或 PR/MR 操作、代码审查、构建或上传
DEB/Docker 等产物、管理仓库权限时不使用。
由显式调用的 `$ack` 根据用户明确的普通 PR/MR 请求路由时,可只执行本文的
PR-only 流程,不把它扩大成版本发布。
---
# Manage Release
@@ -30,18 +28,6 @@ description: >-
## 授权边界
### ACK PR-only 委派
当且仅当用户显式调用 `$ack`,且 ACK 把用户原始的普通 PR/MR 请求委派给本 Skill 时,
使用 PR-only 流程:执行“发现项目规则和当前状态”,锁定准确 head、base、remote、
worktree、验证命令与活动 Forge 账号,然后只执行“推送并创建 PR/MR”及其完成检查。
用户没有明确要求时,不创建 release 分支、不修改版本或 CHANGELOG、不合并、不打 tag、
不创建 Forge Release、不清理分支或 worktree。缺少 head/base/remote、提交边界或平台
认证时按硬停止条件报告,不把普通 PR 伪装成发布。
ACK 只负责路由,不增加授权。Operator 收到的原始请求仍是权限上限;“创建 PR”只授权
为准确 head/base 创建或复用 PR,不自动授权合并。
按用户明确要求执行到对应阶段:
- 分析版本或检查状态:只执行读取和计算,不修改文件或远端。
@@ -60,9 +46,6 @@ ACK 只负责路由,不增加授权。Operator 收到的原始请求仍是权
## 工作流
PR-only 仍按以下章节做实时状态发现与安全检查,但跳过与用户请求无关的版本计算、
release worktree、版本文件、合并和 tag 阶段。已有用户工作不得被 stash、移动或覆盖。
### 1. 发现项目规则和当前状态
从项目根目录开始:
+59
View File
@@ -0,0 +1,59 @@
# orc
ORC 是显式调用的工程编排入口:把开发、版本发布、DEB 和 Docker 任务拆成阶段,
交给对应 Skill,并为每个执行 Agent 选择 `low``mid``high` 档位。
## 什么时候使用
- 一个请求同时包含写代码、发版本和构建产物。
- 希望由 ORC 监督多个 Agent,并按阶段控制成本与推理能力。
- 需要继续中断的多阶段工程流程并保留依赖关系。
只做单一领域任务时可以直接调用对应 Skill;ORC 不替代它们的安全规则。
## 使用前准备
- Orca 正在运行并启用了 orchestration。
- 安装本次需要的 `$ack``$manage-release``$deb-publisher`
`$publish-docker-image`
- 运行 `$orc 初始化` 生成 `docs/orc/config.yaml`,确认三档对应的精确模型。
- `/usr/bin/python3`;ORC v1 配置必须保持为内置模板使用的 JSON-compatible YAML。
resolver 以 `-I -S` 隔离模式运行,不加载项目模块、用户 site-packages 或第三方解析器。
- ORC v1 worker 使用 `workspace-write` sandbox,以便发送 Orca lifecycle 消息;
只读任务会在阶段 prompt 中禁止文件修改。
## Agent 档位
| 档位 | 典型任务 |
|------|----------|
| `low` | 输入明确的测试、构建、打包和上传 |
| `mid` | 常规版本发布与范围清晰的工程任务 |
| `high` | 需求理解、跨系统改动、异常恢复和高风险裁决 |
档位不是权限。三个档位仍受各自 profile 和下游 Skill 的授权边界约束。
当前 ORC v1 的结构化启动适配器支持 Codex worker;其它 Agent CLI 需要独立适配器,
不会通过自由命令接入。
## 使用示例
```text
$orc high 修复登录问题,验证通过后发布新版本。
$orc code=high release=mid docker=low,完成修复、发版并推送镜像。
$orc mid 继续上次中断的 v1.4.0 发布流程。
```
阶段级档位优先于全局档位。用户显式指定后,ORC 不会静默改档;能力不足时会暂停并
请求确认。
## Agent 会做什么
1. 识别开发、源码发布、DEB 与 Docker 阶段及其依赖。
2. 解析每个阶段的 Agent profile,把配置快照、可信 Codex executable 和真实 Git
worktree 绑定成 launch fingerprint,并展示计划和外部写入边界。
3. 通过 Orca 分发给对应 Skill,监督完成消息、异常和决策门。
4. 汇总每个阶段的实际状态、证据和安全恢复入口。
## 如何判断完成
最终结果会逐阶段列出所用 Skill、Agent 档位、源 revision、远端或产物状态,以及任何
未完成项。只有所有必要阶段都通过各自验证时,ORC 才会报告整个流程完成。
+133
View File
@@ -0,0 +1,133 @@
---
name: orc
description: >-
显式编排开发、源码版本发布、DEB 和 Docker 产物任务,把阶段分发给对应 Skill,
并用 low、mid、high 选择执行 Agent 档位。仅在用户显式调用 $orc 或 /orc,要求
跨阶段协调、指定 Agent 级别、监督多个 worker 或继续 ORC 编排时使用。
---
# ORC 工程编排入口
当前会话担任 Coordinator:拆分阶段、选择档位、派发 worker、监督依赖与决策门,
但不替代下游 Skill 执行其领域流程。
开始时解析当前 `SKILL.md` 所在目录,记为 `<orc-skill-dir>`;解析真实项目根目录,
优先使用 `git rev-parse --show-toplevel`。项目配置固定为
`<project-root>/docs/orc/config.yaml`
## 选择模式
- 用户要求初始化 ORC:执行“初始化”。
- 用户要求检查 ORC、档位或运行环境:执行“检查”。
- 用户要求用 ORC 完成任务:执行“编排”。
不要静默初始化,也不要在配置缺失或无效时退回裸命令或当前会话直接执行。
## 初始化
1. 若 `docs/orc/config.yaml` 已存在,停止创建并转入“检查”,不得覆盖。
2. 确认 `docs/orc/` 和目标文件都不是 symlink,再从
`<orc-skill-dir>/templates/config.template.yaml` 以 create-only 方式创建配置,不覆盖
或跟随既有路径。ORC v1 配置必须保持 JSON-compatible YAML,只由隔离的 Python
标准库解析;模板中的 `.` 表示当前项目根。模型 ID 必须由用户或项目的可信配置确认,
不能从任务文本猜测。
3. 运行:
```bash
/usr/bin/python3 -I -S <orc-skill-dir>/scripts/resolve_profile.py validate \
<project-root>/docs/orc/config.yaml
```
4. 报告三个档位的 CLI、模型、reasoning、权限和阶段默认值。除非用户明确要求,
不安装下游 Skill、不创建终端、不修改 Agent 全局配置。
## 检查
1. 校验 `docs/orc/config.yaml`,确认只存在 `low``mid``high` 三档。
2. 检查 `orca status --json`,并确认 orchestration 命令可用。
3. 确认本次所需下游 Skill 已安装:`ack``manage-release``deb-publisher`
`publish-docker-image`。只检查实际会用到的项。
4. 解析 profile 时把项目根和目标 worktree 一并交给 resolver;只有 resolver 验证目标
命中 `allowedWorktrees`、属于当前 Git 仓库且身份稳定后才可创建终端。不得只做文本
比较或跳过机器校验。
5. 任何 profile、Skill、运行时或 worktree 不可用时 fail closed;不得选择相邻档位、
复用身份不明的终端或手写替代流程。
## 编排
1. 读取 [routing.md](references/routing.md),把请求拆成 `code``release``deb`
`docker` 阶段。没有匹配下游 Skill 的工作留在范围外并明确报告。
2. 锁定用户授权的最远动作、目标版本、产物目标、源 commit/tag 与停止点。ORC 的调用
本身不扩大 push、合并、打 tag、上传或部署权限;每个下游 Skill 的授权边界继续生效。
3. 解析档位:阶段级指定 > 全局指定 > `stageDefaults` > `defaultLevel`。只接受
`low``mid``high`;用户显式指定后不得静默升降级。若该档位不足以安全完成,
建立 decision gate,等待用户改档或缩小范围。
4. 用户未指定档位时采用以下判断:清晰、机械的构建或上传可用 `low`;常规版本流程
`mid`;需求理解、跨系统改动、恢复中断流程、目标含糊或高风险裁决用 `high`
5. 对每个阶段运行 profile resolver。全局档位用 `--global-level`,阶段档位用
`--stage-level`;项目根与目标 worktree 必须使用规范绝对路径。模型认证默认复用
`codex-login`,只有明确使用对应环境凭据时才选 `openai``azure-openai`。远端认证
默认 `none`;只有目标 provider 与 transport 已确认时,才选择一个精确的
`github-token``gitlab-token``gitea-token``forgejo-token``ssh-agent`
`deb-token`。不得把多个 provider 凭据一起交给 worker。resolver 不存在静默
fallback
```bash
/usr/bin/python3 -I -S <orc-skill-dir>/scripts/resolve_profile.py resolve \
<project-root>/docs/orc/config.yaml --stage <stage> \
--project-root <absolute-project-root> \
--worktree <absolute-target-worktree> \
[--global-level <low|mid|high>] [--stage-level <low|mid|high>] \
[--model-auth <codex-login|openai|azure-openai>] \
[--remote-auth <exact-provider-or-transport>]
```
6. 核对 resolver 返回的 `launchFingerprint`、绝对 executable、worktree identity 和选择
来源,再读取 [orca-adapter.md](references/orca-adapter.md),把阶段组织为 Orca task DAG。
worker prompt 必须显式写出对应 `$skill`、阶段范围、输入 revision、用户授权边界、
验收证据和依赖结果;下游 Skill 无需知道 ORC。
7. 监督 `worker_done``escalation``decision_gate``worker_done` 只代表该 worker
回报完成;Coordinator 仍需核对下游 Skill 要求的证据和 DAG 后置条件。
8. 逐阶段汇报所选档位、执行 Skill、结果、外部状态和未完成项。任一阶段失败时保留
已成功阶段的准确状态,说明安全恢复入口,不把部分成功概括成全部完成。
## 固定路由边界
- 功能、缺陷、重构与验证闭环交给 `$ack`
- 发布版本、release 分支/PR/MR、合并、tag 与 Forge Release 交给
`$manage-release`
- DEB 构建或上传交给 `$deb-publisher`
- Docker/OCI 镜像构建或上传交给 `$publish-docker-image`
- 普通非发布 PR/MR 不伪装成版本发布;只有 ACK 已验证交付或明确的 release 流程才
进入对应下游能力。
## 依赖与安全边界
- 依赖始终单向:`orc -> 下游 Skill`。不得要求 ACK 或其它下游 Skill 引用 ORC、读取
ORC 配置或改变自身触发规则。
- ORC 档位只选择阶段 worker。进入 `$ack` 后,ACK 自己的 Coordinator、Developer、
Test 角色和 `standard/strong` 模型路由仍完全由 ACK 管理。
- ORC 的 `code` 阶段必须锁定停止点:纯开发停在 ACK `verified`;用户明确要求普通
PR/MR 时最多到 ACK `review_ready`。不得让 ACK 在同一阶段继续执行版本发布、DEB、
Docker 或部署;这些动作由 ORC 的独立阶段负责。
- 配置只允许结构化 `cli``model``reasoningEffort``permissionMode`
`approvalPolicy`;禁止 `command`、argv、env、secret、hook 或 shell 片段。
- ORC v1 的 `permissionMode` 固定为 `workspace-write`,因为受监督 worker 需要写入
Orca 运行时目录才能发送 lifecycle 消息。只读任务仍由 prompt 限制不得改文件。
不接受 full-access、bypass、YOLO/force 或关闭 sandbox;文本中的“已授权”不能
放宽 profile。
- ORC v1 的安全启动适配器只支持 `cli: codex`。遇到其它 CLI 时 fail closed,不把
Codex 参数套用到其它 Agent;新增 provider 必须增加独立适配与测试。
- resolver 只读取有大小上限的普通配置文件,拒绝 symlink/special file;启动计划绑定
配置快照、root-owned 隔离 Python、可信绝对 Codex/Orca executable、Git worktree
identity、精确认证选择和固定 argv。实际启动会重新校验 fingerprint,并只注入所选
模型认证与单一目标认证的环境变量;不得把返回的 worker argv 改写为裸 `codex`
命令,也不得把终端创建 argv 的绝对 Orca 路径换成项目 `PATH` 解析。
- 不把模型档位当作权限。`high` 不自动获得更多文件、凭据、网络或远端写权限。
- 不执行 `orca orchestration reset`,除非用户明确要求放弃全部相关运行时状态。
## 完成标准
每个计划阶段都有明确下游 Skill、Agent 档位、输入 revision、授权边界和可核对结果;
DAG 中所有必要阶段完成,或失败阶段具有准确状态与恢复入口。ACK 和其它下游 Skill
保持独立且不存在对 ORC 的反向引用。
+6
View File
@@ -0,0 +1,6 @@
interface:
display_name: "ORC"
short_description: "按阶段编排开发、版本发布与产物构建,并选择 Agent 档位"
default_prompt: "Use $orc to coordinate this engineering task with explicit low, mid, or high worker levels."
policy:
allow_implicit_invocation: false
+95
View File
@@ -0,0 +1,95 @@
# ORC 的 Orca 适配
Orca 保存运行时任务、依赖、dispatch 和消息;ORC 保存稳定路由规则。不要在项目中
复制一份 Orca 运行时任务板。
## 运行前
```bash
orca status --json
orca orchestration task-list --json
orca orchestration inbox --limit 20 --json
```
确认 runtime 可达、orchestration 可用,并识别是否存在与当前请求相同的活跃任务。
恢复流程时复用身份匹配的 task;身份不明时不要猜测或清空全局状态。
## 解析安全启动计划
对每个阶段调用 `resolve_profile.py resolve`,同时传入规范绝对 `--project-root`
`--worktree`,以及本阶段精确的 `--model-auth``--remote-auth`。JSON 结果包含选择来源、
结构化 profile、认证选择、配置与 worktree identity、可信 Python/Codex/Orca executable、
`launchFingerprint`、固定 worker/launcher argv,以及完整的 `terminalCreateArgv`
`terminalCreateShellCommand`
- 优先把返回的 `terminalCreateArgv` 原样交给 argv-capable 进程工具;若工具只接受 shell
字符串,则执行完整的 `terminalCreateShellCommand`。不得再插值、包引号或追加参数。
- `terminalCreateArgv` 的首项是 resolver 选定的绝对 Orca executable,不替换成裸
`orca` 或重新通过项目 `PATH` 查找。
- 不从 YAML 自行拼接命令,不追加 argv、环境变量、hook 或权限参数。
- `remote-auth` 必须与本阶段唯一目标 provider/transport 一致;默认 `none`,不为方便
同时暴露多个 token 或 SSH agent。需要代理、额外凭据或 Docker registry env secret 时
fail closed,由用户确认新的最小权限适配,不继承 ambient proxy/credential 变量。
- resolver 失败时停止;不改用默认模型、邻近档位或已有未知终端。
- 不重复手写 worktree 检查;resolver 已拒绝 root、不存在、非规范、含 symlink、未注册、
跨仓库或不在 allowlist 的目标,并把身份写入 fingerprint。
- ORC v1 profile 必须使用 `workspace-write``read-only` sandbox 可能阻止 Orca CLI
写入自身运行时或挂载目录,导致 worker 无法发送 `worker_done`;只读工作通过 task
spec 限制,而不是换成无法完成 lifecycle 的 sandbox。
## 创建与派发
每个需要 fresh worker 的阶段依次执行:
1. 使用解析结果创建目标 worktree 内的终端。下面整行代表 resolver 返回的一个完整值,
不是待替换的参数模板:
```bash
<resolver-terminalCreateShellCommand>
```
2. 使用返回的具体 handle 等待 Agent TUI 可接收输入:
```bash
orca terminal wait --terminal <handle> --for tui-idle \
--timeout-ms 60000 --json
```
3. 先为前置阶段创建 task,再用其 ID 构造依赖数组创建后置 task。task spec 使用
`routing.md` 的 worker prompt 契约。
4. 对已就绪 task 执行:
```bash
orca orchestration dispatch --task <task-id> --to <handle> --inject --json
```
不要把同一 task 同时派给多个写 worker。并行阶段必须使用不会竞争同一工作树写入的
独立 worker;否则串行执行。`terminal create` 超时或连接中断时状态不确定:先用
`terminalTitle`、worktree 和 fingerprint 核对 `orca terminal list --json`,没有完成
reconciliation 前不得重试创建。
## 监督循环
```bash
orca orchestration check --wait \
--types worker_done,escalation,decision_gate \
--timeout-ms 900000 --json
```
- timeout 或空结果是检查点,不等于失败;任务仍活跃时继续滚动等待。
- 收到 `decision_gate` 时只回答当前问题,不扩大用户授权。
- 收到 `escalation` 时保留已完成状态,判断是改档、补输入还是停止。
- 收到 `worker_done` 后核对 task/dispatch ID、下游完成证据和外部状态,再把依赖 task
标记完成并派发新就绪阶段。
- 三次连续同因失败后停止重派,报告恢复入口;不要用新终端掩盖同一阻塞。
## ACK 特例
`code` worker 是运行 `$ack` 的阶段 Coordinator。ORC 的 `low/mid/high` 只决定这个
worker 的启动 profileACK 启动 Developer/Test 时继续读取自己的项目配置与
`standard/strong` 规则。ORC 不读取、改写或覆盖 ACK 的内部 worker profiles。
## 收尾
不要自动关闭仍承担恢复入口的终端,不自动删除 worktree 或清空 runtime-global task。
只有用户明确要求清理时,才按 Orca 和下游 Skill 的安全规则处理。
+74
View File
@@ -0,0 +1,74 @@
# ORC 阶段路由
ORC 只负责拆分、依赖、档位和监督。领域步骤、授权检查和完成标准由下游 Skill 自己
决定。
## 路由表
| 阶段 | 下游 Skill | 包含 | 不包含 |
|------|------------|------|--------|
| `code` | `$ack` | 功能、缺陷、重构、测试、三角色验证闭环,以及用户明确要求的普通非发布 PR/MR | 版本发布、单独上传产物 |
| `release` | `$manage-release` | 版本号、release worktree/分支、release PR/MR、合并、tag、Forge Release、恢复发布 | 普通非发布 PR/MR、构建或上传 DEB/Docker |
| `deb` | `$deb-publisher` | DEB 构建、校验、上传与仓库可见性 | 源码 tag、Docker 镜像 |
| `docker` | `$publish-docker-image` | Docker/OCI 构建、push、digest 与平台验证 | 源码版本生命周期、DEB |
没有匹配项时不要临时扩写某个 Skill 的职责,也不要让 ORC 自己模仿领域流程。报告缺少
的能力,由用户决定直接执行、安装新 Skill 或另行设计。
## 拆分规则
1. 先从用户请求提取最终结果,再拆出真正需要的阶段;不要因为安装了某个 Skill 就
自动增加发布或上传。
2. 为每个阶段锁定输入:项目、worktree、源 commit/tag、版本、目标和用户授权的最远
写操作。
3. 同一领域的连续动作保留在一个下游任务中。例如版本号、release PR、合并和 tag
属于一个 `manage-release` 生命周期,不拆成多个互相争抢状态的 worker。
4. 只有输入 revision 完全相同且互不修改同一工作树时,才并行执行 DEB 与 Docker。
5. 普通代码改动进入 ACK。若项目尚未初始化 ACK,`code` 阶段停在前置条件,不由 ORC
静默初始化。
6. `code` 阶段默认停在 ACK `verified`;用户明确要求普通 PR/MR 时最多到
`review_ready`。worker prompt 必须禁止继续执行版本发布、DEB、Docker 或部署。
## 常见 DAG
完整交付:
```text
code ($ack)
-> release ($manage-release)
-> deb ($deb-publisher)
-> docker ($publish-docker-image)
```
只从当前 commit 构建产物:
```text
deb ($deb-publisher) || docker ($publish-docker-image)
```
仅发布源码版本:
```text
release ($manage-release)
```
依赖不是固定模板,但 ORC v1 不拆分一个 `manage-release` 生命周期。若项目要求在打开
release PR 与合并之间插入 DEB/Docker gate,当前 task 粒度无法安全表达该中间里程碑;
在打开 PR 前建立 decision gate 并报告该流程暂不支持,不得用循环依赖或两个 release
worker 临时拼接。
## Worker prompt 契约
每个 worker prompt 至少包含:
- 第一条指令显式调用唯一的下游 Skill,例如 `Use $manage-release ...`
- 阶段目标与明确的非目标。
- 项目/worktree、输入 commit/tag 和前置阶段的可核对结果。
- 用户已经授予的最远动作;未授权动作明确禁止。
- 要求遵循项目 Agent 指令和下游 Skill 自身的停止条件。
- 完成证据,以及通过 live dispatch preamble 回报 `worker_done` 的要求。
- `code` prompt 还必须写明 ACK 停止点是 `verified` 或普通 PR 的 `review_ready`,并禁止
ACK 路由版本、DEB、Docker 或部署动作。
不要把 ORC 的 profile、路由器内部规则或其它下游 Skill 注入 worker。worker 只需要
当前阶段、对应 Skill 和必要依赖结果。
+889
View File
@@ -0,0 +1,889 @@
#!/usr/bin/env python3
"""Validate ORC config and build a worktree-bound Codex launch plan."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import pwd
import re
import shlex
import stat
import subprocess
import sys
from pathlib import Path
from typing import Any
LEVELS = ("low", "mid", "high")
STAGES = ("code", "release", "deb", "docker")
EXPECTED_EFFORT = {"low": "low", "mid": "medium", "high": "high"}
TOP_LEVEL_KEYS = {
"version",
"defaultLevel",
"stageDefaults",
"allowedWorktrees",
"profiles",
}
PROFILE_KEYS = {
"cli",
"model",
"reasoningEffort",
"permissionMode",
"approvalPolicy",
}
MODEL_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}\Z")
MAX_CONFIG_SIZE = 64 * 1024
MAX_CONTROL_OUTPUT = 64 * 1024
CONTROL_TIMEOUT_SECONDS = 15
CONFIG_RELATIVE_PATH = Path("docs/orc/config.yaml")
COMMON_ENVIRONMENT_NAMES = {
"COLORTERM",
"LANG",
"LC_ALL",
"NO_COLOR",
"TERM",
"TZ",
}
MODEL_AUTH_ENVIRONMENT = {
"codex-login": frozenset(),
"openai": frozenset({"OPENAI_API_KEY"}),
"azure-openai": frozenset({"AZURE_OPENAI_API_KEY"}),
}
REMOTE_AUTH_ENVIRONMENT = {
"none": frozenset(),
"github-token": frozenset({"GITHUB_TOKEN"}),
"gitlab-token": frozenset({"GITLAB_TOKEN"}),
"gitea-token": frozenset({"GITEA_TOKEN"}),
"forgejo-token": frozenset({"FORGEJO_TOKEN"}),
"ssh-agent": frozenset({"SSH_AUTH_SOCK"}),
"deb-token": frozenset({"DEB_TOKEN"}),
}
STAGE_REMOTE_AUTH = {
"code": frozenset(
{
"none",
"github-token",
"gitlab-token",
"gitea-token",
"forgejo-token",
"ssh-agent",
}
),
"release": frozenset(
{
"none",
"github-token",
"gitlab-token",
"gitea-token",
"forgejo-token",
"ssh-agent",
}
),
"deb": frozenset({"none", "deb-token", "ssh-agent"}),
"docker": frozenset({"none"}),
}
class ConfigError(ValueError):
"""ORC configuration or launch state is invalid or unsafe."""
def _json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ConfigError("JSON-compatible YAML contains a duplicate key")
result[key] = value
return result
def _read_bounded_regular_file(path: Path) -> tuple[str, dict[str, int]]:
absolute = path.absolute()
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
descriptor: int | None = None
try:
descriptor = os.open(absolute, flags)
with os.fdopen(descriptor, "rb") as stream:
descriptor = None
before = os.fstat(stream.fileno())
if not stat.S_ISREG(before.st_mode):
raise ConfigError(f"config must be a regular file: {absolute}")
if before.st_size > MAX_CONFIG_SIZE:
raise ConfigError(f"config exceeds {MAX_CONFIG_SIZE} bytes")
content = stream.read(MAX_CONFIG_SIZE + 1)
after = os.fstat(stream.fileno())
except OSError as exc:
raise ConfigError(f"cannot safely read config {absolute}: {exc}") from exc
finally:
if descriptor is not None:
os.close(descriptor)
if len(content) > MAX_CONFIG_SIZE:
raise ConfigError(f"config exceeds {MAX_CONFIG_SIZE} bytes")
identity_before = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
identity_after = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
if identity_before != identity_after:
raise ConfigError("config changed while it was being read")
try:
text = content.decode("utf-8")
except UnicodeDecodeError as exc:
raise ConfigError("config must be valid UTF-8") from exc
return text, {
"device": before.st_dev,
"inode": before.st_ino,
"size": before.st_size,
"mtimeNs": before.st_mtime_ns,
}
def _mapping(value: Any, path: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise ConfigError(f"{path} must be a mapping")
return value
def _exact_keys(
value: dict[str, Any],
*,
required: set[str],
path: str,
) -> None:
missing = sorted(required - set(value))
unknown = sorted(set(value) - required)
if missing:
raise ConfigError(f"{path} is missing fields: {', '.join(missing)}")
if unknown:
raise ConfigError(f"{path} has unknown fields: {', '.join(unknown)}")
def _enum(value: Any, allowed: tuple[str, ...] | set[str], path: str) -> str:
if not isinstance(value, str) or value not in allowed:
raise ConfigError(f"{path} must be one of: {', '.join(sorted(allowed))}")
return value
def validate_config(document: Any) -> dict[str, Any]:
config = _mapping(document, "config")
_exact_keys(config, required=TOP_LEVEL_KEYS, path="config")
version = config["version"]
if isinstance(version, bool) or version != 1:
raise ConfigError("version must be integer 1")
_enum(config["defaultLevel"], LEVELS, "defaultLevel")
stage_defaults = _mapping(config["stageDefaults"], "stageDefaults")
unknown_stages = sorted(set(stage_defaults) - set(STAGES))
if unknown_stages:
raise ConfigError(
f"stageDefaults has unknown fields: {', '.join(unknown_stages)}"
)
for stage, level in stage_defaults.items():
_enum(level, LEVELS, f"stageDefaults.{stage}")
allowed = config["allowedWorktrees"]
if not isinstance(allowed, list) or not allowed:
raise ConfigError("allowedWorktrees must be a non-empty list")
seen_worktrees: set[str] = set()
for index, entry in enumerate(allowed):
if (
not isinstance(entry, str)
or not entry
or entry != entry.strip()
or any(character in entry for character in ("\x00", "\n", "\r"))
):
raise ConfigError(
f"allowedWorktrees[{index}] must be a safe non-empty path"
)
if entry != "." and not Path(entry).is_absolute():
raise ConfigError(
f"allowedWorktrees[{index}] must be '.' or an absolute path"
)
normalized = entry if entry == "." else str(Path(entry).absolute())
if normalized in seen_worktrees:
raise ConfigError(f"allowedWorktrees contains duplicate path: {entry}")
seen_worktrees.add(normalized)
profiles = _mapping(config["profiles"], "profiles")
_exact_keys(profiles, required=set(LEVELS), path="profiles")
for level in LEVELS:
profile = _mapping(profiles[level], f"profiles.{level}")
_exact_keys(profile, required=PROFILE_KEYS, path=f"profiles.{level}")
if profile["cli"] != "codex":
raise ConfigError(f"profiles.{level}.cli must be codex in ORC v1")
model = profile["model"]
if not isinstance(model, str) or not MODEL_RE.fullmatch(model):
raise ConfigError(f"profiles.{level}.model is not a safe exact model ID")
expected_effort = EXPECTED_EFFORT[level]
if profile["reasoningEffort"] != expected_effort:
raise ConfigError(
f"profiles.{level}.reasoningEffort must be {expected_effort}"
)
if profile["permissionMode"] != "workspace-write":
raise ConfigError(
f"profiles.{level}.permissionMode must be workspace-write in ORC v1"
)
_enum(
profile["approvalPolicy"],
{"untrusted", "on-request", "never"},
f"profiles.{level}.approvalPolicy",
)
return config
def load_config_snapshot(path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
raw, identity = _read_bounded_regular_file(path)
try:
document = json.loads(raw, object_pairs_hook=_json_object)
except ConfigError:
raise
except Exception as exc:
raise ConfigError("invalid JSON-compatible YAML config") from exc
config = validate_config(document)
snapshot = {
"sha256": hashlib.sha256(raw.encode("utf-8")).hexdigest(),
**identity,
}
return config, snapshot
def load_config(path: Path) -> dict[str, Any]:
config, _ = load_config_snapshot(path)
return config
def resolve_profile(
config: dict[str, Any],
*,
stage: str,
global_level: str | None = None,
stage_level: str | None = None,
model_auth: str = "codex-login",
remote_auth: str = "none",
) -> dict[str, Any]:
_enum(stage, STAGES, "stage")
if global_level is not None:
_enum(global_level, LEVELS, "global level")
if stage_level is not None:
_enum(stage_level, LEVELS, "stage level")
_enum(model_auth, set(MODEL_AUTH_ENVIRONMENT), "model auth")
_enum(remote_auth, STAGE_REMOTE_AUTH[stage], f"{stage} remote auth")
if stage_level is not None:
level, source = stage_level, "request.stage"
elif global_level is not None:
level, source = global_level, "request.global"
elif stage in config["stageDefaults"]:
level, source = config["stageDefaults"][stage], f"config.stageDefaults.{stage}"
else:
level, source = config["defaultLevel"], "config.defaultLevel"
profile = config["profiles"].get(level)
if profile is None:
raise ConfigError(f"requested profile does not exist: {level}")
worker_args = [
"--model",
profile["model"],
"-c",
f'model_reasoning_effort="{profile["reasoningEffort"]}"',
"--sandbox",
profile["permissionMode"],
"--ask-for-approval",
profile["approvalPolicy"],
"--strict-config",
]
return {
"stage": stage,
"level": level,
"selectionSource": source,
"modelAuth": model_auth,
"remoteAuth": remote_auth,
"profile": dict(profile),
"workerArgs": worker_args,
}
def _path_has_parent_reference(value: str) -> bool:
return ".." in Path(value).parts
def _assert_no_symlink_components(path: Path, label: str) -> None:
if not path.is_absolute():
raise ConfigError(f"{label} must be absolute: {path}")
current = Path(path.anchor)
for part in path.parts[1:]:
current /= part
try:
metadata = os.lstat(current)
except OSError as exc:
raise ConfigError(f"{label} does not exist: {current}") from exc
if stat.S_ISLNK(metadata.st_mode):
raise ConfigError(f"{label} must not contain symlinks: {current}")
def canonical_directory(value: str | Path, label: str) -> Path:
text = str(value)
if (
not text
or text != text.strip()
or any(character in text for character in ("\x00", "\n", "\r"))
or _path_has_parent_reference(text)
):
raise ConfigError(f"{label} must be a safe canonical absolute path")
raw = Path(text)
_assert_no_symlink_components(raw, label)
try:
resolved = raw.resolve(strict=True)
except OSError as exc:
raise ConfigError(f"{label} does not exist: {raw}") from exc
if resolved != raw or resolved == Path(resolved.anchor) or not resolved.is_dir():
raise ConfigError(f"{label} must be a canonical non-root directory: {resolved}")
return resolved
def account_identity() -> tuple[Path, str]:
account = pwd.getpwuid(os.getuid())
account_home = Path(account.pw_dir).resolve(strict=True)
if not account_home.is_dir():
raise ConfigError("current account home is unavailable")
return account_home, account.pw_name
def trusted_path_entries() -> list[Path]:
account_home, _ = account_identity()
candidates = [
account_home / ".local" / "bin",
account_home / ".local" / "share" / "mise" / "shims",
account_home / ".cargo" / "bin",
Path("/home/linuxbrew/.linuxbrew/bin"),
Path("/usr/local/bin"),
Path("/usr/bin"),
Path("/bin"),
]
result: list[Path] = []
for candidate in candidates:
try:
resolved = candidate.resolve(strict=True)
except OSError:
continue
if resolved.is_dir() and resolved not in result:
result.append(resolved)
return result
def _trusted_executable(path: Path, expected_name: str) -> Path | None:
try:
candidate_metadata = os.lstat(path)
resolved = path.resolve(strict=True)
metadata = resolved.stat()
except OSError:
return None
if not (
stat.S_ISREG(candidate_metadata.st_mode)
or stat.S_ISLNK(candidate_metadata.st_mode)
):
return None
if not stat.S_ISREG(metadata.st_mode) or not os.access(resolved, os.X_OK):
return None
if metadata.st_uid not in {0, os.getuid()}:
return None
if stat.S_IMODE(metadata.st_mode) & 0o022:
return None
if expected_name == "python3":
if not resolved.name.startswith("python3"):
return None
elif resolved.name != expected_name:
return None
return resolved
def resolve_trusted_executable(name: str) -> Path:
if name not in {"codex", "git", "orca"}:
raise ConfigError(f"unsupported executable: {name}")
search_paths = trusted_path_entries()
if name == "git":
search_paths = [
path
for path in search_paths
if str(path) in {"/usr/local/bin", "/usr/bin", "/bin"}
]
for directory in search_paths:
resolved = _trusted_executable(directory / name, name)
if resolved is not None:
return resolved
raise ConfigError(f"trusted {name} executable was not found in fixed directories")
def resolve_trusted_python() -> Path:
for candidate in (Path("/usr/bin/python3"), Path("/usr/local/bin/python3")):
resolved = _trusted_executable(candidate, "python3")
if resolved is not None and resolved.stat().st_uid == 0:
return resolved
raise ConfigError("a root-owned Python executable was not found in fixed paths")
def control_environment() -> dict[str, str]:
account_home, username = account_identity()
result = {
"HOME": str(account_home),
"LOGNAME": username,
"PATH": os.pathsep.join(str(path) for path in trusted_path_entries()),
"USER": username,
}
for name, value in os.environ.items():
if (
name in COMMON_ENVIRONMENT_NAMES or name.startswith("LC_")
) and "\x00" not in value:
result[name] = value
return result
def selected_credential_names(
stage: str,
*,
model_auth: str,
remote_auth: str,
) -> frozenset[str]:
_enum(stage, STAGES, "stage")
_enum(model_auth, set(MODEL_AUTH_ENVIRONMENT), "model auth")
_enum(remote_auth, STAGE_REMOTE_AUTH[stage], f"{stage} remote auth")
names = MODEL_AUTH_ENVIRONMENT[model_auth] | REMOTE_AUTH_ENVIRONMENT[remote_auth]
for name in names:
value = os.environ.get(name)
if not value or "\x00" in value:
raise ConfigError(
f"selected authentication variable is unavailable: {name}"
)
if name == "SSH_AUTH_SOCK":
socket_path = Path(value)
if (
not socket_path.is_absolute()
or value != value.strip()
or any(character in value for character in ("\n", "\r"))
or _path_has_parent_reference(value)
):
raise ConfigError("selected SSH_AUTH_SOCK is not a safe absolute path")
_assert_no_symlink_components(socket_path, "selected SSH_AUTH_SOCK")
try:
metadata = os.lstat(socket_path)
except OSError as exc:
raise ConfigError("selected SSH_AUTH_SOCK is unavailable") from exc
if (
not stat.S_ISSOCK(metadata.st_mode)
or metadata.st_uid != os.getuid()
or stat.S_IMODE(metadata.st_mode) & 0o022
):
raise ConfigError("selected SSH_AUTH_SOCK is not a trusted user socket")
return names
def worker_environment(
stage: str,
*,
model_auth: str,
remote_auth: str,
) -> dict[str, str]:
result = control_environment()
for name in selected_credential_names(
stage,
model_auth=model_auth,
remote_auth=remote_auth,
):
result[name] = os.environ[name]
return result
def _run_control(argv: list[str], label: str) -> str:
try:
completed = subprocess.run(
argv,
shell=False,
check=False,
capture_output=True,
text=True,
timeout=CONTROL_TIMEOUT_SECONDS,
env=control_environment(),
)
except (OSError, subprocess.TimeoutExpired) as exc:
raise ConfigError(f"{label} could not be executed safely") from exc
if completed.returncode != 0:
raise ConfigError(f"{label} failed with exit {completed.returncode}")
output = completed.stdout.strip()
if not output or len(output.encode("utf-8")) > MAX_CONTROL_OUTPUT:
raise ConfigError(f"{label} returned invalid output")
return output
def _git_path(value: str, cwd: Path, label: str) -> Path:
candidate = Path(value)
if not candidate.is_absolute():
candidate = cwd / candidate
_assert_no_symlink_components(candidate, label)
try:
resolved = candidate.resolve(strict=True)
except OSError as exc:
raise ConfigError(f"{label} is invalid") from exc
if not resolved.is_dir():
raise ConfigError(f"{label} is not a directory")
return resolved
def validate_worktree(
config: dict[str, Any],
*,
project_root_value: str | Path,
worktree_value: str | Path,
) -> dict[str, Any]:
project_root = canonical_directory(project_root_value, "project root")
worktree = canonical_directory(worktree_value, "target worktree")
git = resolve_trusted_executable("git")
top_level = canonical_directory(
_run_control(
[str(git), "-C", str(project_root), "rev-parse", "--show-toplevel"],
"Git project-root check",
),
"Git project root",
)
if top_level != project_root:
raise ConfigError("project root is not the repository top level")
common_dir = _git_path(
_run_control(
[str(git), "-C", str(project_root), "rev-parse", "--git-common-dir"],
"Git common-directory check",
),
project_root,
"Git common directory",
)
listed = _run_control(
[str(git), "-C", str(project_root), "worktree", "list", "--porcelain"],
"Git worktree listing",
)
registered: set[Path] = set()
for line in listed.splitlines():
if line.startswith("worktree "):
try:
registered.add(
canonical_directory(line[9:], "registered worktree")
)
except ConfigError:
continue
allowed: set[Path] = set()
for entry in config["allowedWorktrees"]:
allowed.add(
project_root
if entry == "."
else canonical_directory(entry, "allowed worktree")
)
if not allowed <= registered:
raise ConfigError("allowedWorktrees contains an unregistered Git worktree")
if worktree not in allowed:
raise ConfigError("target worktree is not in allowedWorktrees")
target_top = canonical_directory(
_run_control(
[str(git), "-C", str(worktree), "rev-parse", "--show-toplevel"],
"Git target-worktree check",
),
"target Git worktree",
)
target_common = _git_path(
_run_control(
[str(git), "-C", str(worktree), "rev-parse", "--git-common-dir"],
"target Git common-directory check",
),
worktree,
"target Git common directory",
)
if target_top != worktree or target_common != common_dir:
raise ConfigError("target worktree does not belong to the project repository")
worktree_metadata = worktree.stat()
common_metadata = common_dir.stat()
return {
"projectRoot": str(project_root),
"worktree": str(worktree),
"gitCommonDir": str(common_dir),
"worktreeIdentity": {
"device": worktree_metadata.st_dev,
"inode": worktree_metadata.st_ino,
"gitCommonDevice": common_metadata.st_dev,
"gitCommonInode": common_metadata.st_ino,
},
}
def _file_facts(path: Path) -> dict[str, Any]:
metadata = path.stat()
return {
"path": str(path),
"device": metadata.st_dev,
"inode": metadata.st_ino,
"size": metadata.st_size,
"mtimeNs": metadata.st_mtime_ns,
}
def _executable_facts(path: Path) -> dict[str, Any]:
return {
**_file_facts(path),
"version": _run_control([str(path), "--version"], "Codex version check"),
}
def _python_facts(path: Path) -> dict[str, Any]:
version = _run_control(
[
str(path),
"-I",
"-S",
"-c",
"import sys; print(sys.version.split()[0])",
],
"Python version check",
)
return {
**_file_facts(path),
"version": version,
}
def _fingerprint(value: dict[str, Any]) -> str:
canonical = json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
return "sha256:" + hashlib.sha256(canonical).hexdigest()
def build_launch_plan(
config_path: Path,
*,
project_root: str | Path,
worktree: str | Path,
stage: str,
global_level: str | None = None,
stage_level: str | None = None,
model_auth: str = "codex-login",
remote_auth: str = "none",
) -> dict[str, Any]:
project = canonical_directory(project_root, "project root")
expected_config = project / CONFIG_RELATIVE_PATH
if config_path.absolute() != expected_config:
raise ConfigError(f"config path must be {expected_config}")
_assert_no_symlink_components(expected_config, "config path")
config, config_snapshot = load_config_snapshot(expected_config)
worktree_facts = validate_worktree(
config,
project_root_value=project,
worktree_value=worktree,
)
profile = resolve_profile(
config,
stage=stage,
global_level=global_level,
stage_level=stage_level,
model_auth=model_auth,
remote_auth=remote_auth,
)
selected_credential_names(
stage,
model_auth=model_auth,
remote_auth=remote_auth,
)
codex = resolve_trusted_executable("codex")
executable = _executable_facts(codex)
orca = resolve_trusted_executable("orca")
orca_executable = _file_facts(orca)
python = resolve_trusted_python()
python_executable = _python_facts(python)
script = Path(__file__).resolve(strict=True)
script_metadata = script.stat()
script_sha256 = hashlib.sha256(script.read_bytes()).hexdigest()
launch_facts = {
**profile,
**worktree_facts,
"config": {
"path": str(expected_config),
**config_snapshot,
},
"executable": executable,
"orca": orca_executable,
"python": python_executable,
"launcher": {
"path": str(script),
"device": script_metadata.st_dev,
"inode": script_metadata.st_ino,
"size": script_metadata.st_size,
"mtimeNs": script_metadata.st_mtime_ns,
"sha256": script_sha256,
},
}
fingerprint = _fingerprint(launch_facts)
worker_argv = [str(codex), *profile["workerArgs"]]
launcher_argv = [
str(python),
"-I",
"-S",
str(script),
"_launch",
str(expected_config),
"--project-root",
str(project),
"--worktree",
str(worktree_facts["worktree"]),
"--stage",
stage,
"--expected-fingerprint",
fingerprint,
]
if global_level is not None:
launcher_argv.extend(["--global-level", global_level])
if stage_level is not None:
launcher_argv.extend(["--stage-level", stage_level])
if model_auth != "codex-login":
launcher_argv.extend(["--model-auth", model_auth])
if remote_auth != "none":
launcher_argv.extend(["--remote-auth", remote_auth])
terminal_command = shlex.join(launcher_argv)
worktree_selector = "path:" + str(worktree_facts["worktree"])
terminal_title = f"ORC-{stage}-{profile['level']}-{fingerprint[7:15]}"
terminal_create_argv = [
str(orca),
"terminal",
"create",
"--worktree",
worktree_selector,
"--title",
terminal_title,
"--command",
terminal_command,
"--json",
]
return {
**launch_facts,
"argv": worker_argv,
"launchFingerprint": fingerprint,
"launcherArgv": launcher_argv,
"terminalCommand": terminal_command,
"worktreeSelector": worktree_selector,
"terminalTitle": terminal_title,
"terminalCreateArgv": terminal_create_argv,
"terminalCreateShellCommand": shlex.join(terminal_create_argv),
}
def execute_launch(args: argparse.Namespace) -> int:
plan = build_launch_plan(
args.config,
project_root=args.project_root,
worktree=args.worktree,
stage=args.stage,
global_level=args.global_level,
stage_level=args.stage_level,
model_auth=args.model_auth,
remote_auth=args.remote_auth,
)
if plan["launchFingerprint"] != args.expected_fingerprint:
raise ConfigError("launch fingerprint changed; resolve the profile again")
current = canonical_directory(Path.cwd(), "launcher working directory")
if current != Path(plan["worktree"]):
raise ConfigError("launcher working directory does not match target worktree")
current_python = Path(sys.executable).resolve(strict=True)
if current_python != Path(plan["python"]["path"]):
raise ConfigError("launcher Python does not match the resolved interpreter")
if _python_facts(current_python) != plan["python"]:
raise ConfigError("Python executable changed before launch")
executable = Path(plan["executable"]["path"])
if _executable_facts(executable) != plan["executable"]:
raise ConfigError("Codex executable changed before launch")
try:
os.execve(
executable,
plan["argv"],
worker_environment(
plan["stage"],
model_auth=plan["modelAuth"],
remote_auth=plan["remoteAuth"],
),
)
except OSError as exc:
raise ConfigError("Codex worker could not be launched") from exc
return 1 # pragma: no cover - os.execve does not return on success
def _add_resolution_arguments(parser: argparse.ArgumentParser) -> None:
parser.add_argument("config", type=Path)
parser.add_argument("--project-root", required=True)
parser.add_argument("--worktree", required=True)
parser.add_argument("--stage", required=True, choices=STAGES)
parser.add_argument("--global-level", choices=LEVELS)
parser.add_argument("--stage-level", choices=LEVELS)
parser.add_argument(
"--model-auth",
choices=tuple(MODEL_AUTH_ENVIRONMENT),
default="codex-login",
)
parser.add_argument(
"--remote-auth",
choices=tuple(REMOTE_AUTH_ENVIRONMENT),
default="none",
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
validate = subparsers.add_parser("validate", help="validate config only")
validate.add_argument("config", type=Path)
resolve = subparsers.add_parser("resolve", help="resolve a bound launch plan")
_add_resolution_arguments(resolve)
launch = subparsers.add_parser("_launch", help=argparse.SUPPRESS)
_add_resolution_arguments(launch)
launch.add_argument("--expected-fingerprint", required=True)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
if args.command == "validate":
config = load_config(args.config)
result: dict[str, Any] = {
"ok": True,
"config": str(args.config.absolute()),
"levels": list(LEVELS),
"stages": list(STAGES),
"defaultLevel": config["defaultLevel"],
}
elif args.command == "resolve":
result = build_launch_plan(
args.config,
project_root=args.project_root,
worktree=args.worktree,
stage=args.stage,
global_level=args.global_level,
stage_level=args.stage_level,
model_auth=args.model_auth,
remote_auth=args.remote_auth,
)
else:
return execute_launch(args)
except ConfigError as exc:
print(f"ORC config error: {exc}", file=sys.stderr)
return 1
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
+36
View File
@@ -0,0 +1,36 @@
{
"version": 1,
"defaultLevel": "mid",
"stageDefaults": {
"code": "high",
"release": "mid",
"deb": "low",
"docker": "low"
},
"allowedWorktrees": [
"."
],
"profiles": {
"low": {
"cli": "codex",
"model": "gpt-5.6-luna",
"reasoningEffort": "low",
"permissionMode": "workspace-write",
"approvalPolicy": "never"
},
"mid": {
"cli": "codex",
"model": "gpt-5.6-terra",
"reasoningEffort": "medium",
"permissionMode": "workspace-write",
"approvalPolicy": "never"
},
"high": {
"cli": "codex",
"model": "gpt-5.6-sol",
"reasoningEffort": "high",
"permissionMode": "workspace-write",
"approvalPolicy": "never"
}
}
}
+1 -2
View File
@@ -2,8 +2,7 @@
name: publish-docker-image
description: >-
构建当前项目的 Docker 镜像,并将其上传到用户指定的镜像仓库。仅当用户显式指定
$publish-docker-image明确说“使用 publish-docker-image skill”,或由显式调用的 `$ack`
根据用户明确的 Docker/OCI 镜像发布请求路由时使用;
$publish-docker-image明确说“使用 publish-docker-image skill”时使用;
不要因普通编码、编辑 Dockerfile、本地构建、测试或一般 Docker 问题而自动触发。
---
-353
View File
@@ -1,353 +0,0 @@
from __future__ import annotations
import copy
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
REPO_ROOT = Path(__file__).resolve().parents[1]
ACK_DIR = REPO_ROOT / "skills" / "ack"
SCRIPTS_DIR = ACK_DIR / "scripts"
sys.path.insert(0, str(SCRIPTS_DIR))
import launch_worker # noqa: E402
import validate_tasks # noqa: E402
import worker_profiles # noqa: E402
def operator_orchestration() -> dict:
return {
"profileVersion": 1,
"mode": "orca",
"allowedWorktrees": ["/repo/demo"],
"modelAllowlist": {
"codex": {
"developer": {"standard": ["gpt-dev"]},
"test": {"standard": ["gpt-low"]},
"operator": {"standard": ["gpt-low"]},
}
},
"profiles": {
"codex-dev-standard": {
"role": "developer",
"cli": "codex",
"tier": "standard",
"model": "gpt-dev",
"reasoningEffort": "medium",
"permissionMode": "workspace-write",
},
"codex-test-standard": {
"role": "test",
"cli": "codex",
"tier": "standard",
"model": "gpt-low",
"reasoningEffort": "low",
"permissionMode": "workspace-write",
},
"codex-operator-standard": {
"role": "operator",
"cli": "codex",
"tier": "standard",
"model": "gpt-low",
"reasoningEffort": "low",
"permissionMode": "workspace-write",
},
},
"defaults": {
"developer": "codex-dev-standard",
"test": "codex-test-standard",
"operator": "codex-operator-standard",
},
}
def routed_board() -> dict:
return {
"version": 1,
"ackVersion": "0.12.0",
"project": {
"name": "demo",
"orchestration": operator_orchestration(),
},
"workerReceipts": [],
"tasks": [
{
"id": "DELIVERY-001",
"type": "delivery-operation",
"title": "publish one DEB",
"status": "open",
"operation": {
"skill": "deb-publisher",
"request": "发布 1.2.3 的 amd64 DEB 到 testing 仓库",
},
"dispatch": {
"operator": {
"profileId": "codex-operator-standard",
"receiptId": None,
"attemptId": None,
"taskId": None,
"dispatchId": None,
},
"rounds": [],
},
}
],
}
class AckDeliveryRoutingTests(unittest.TestCase):
def test_operator_profile_uses_the_test_low_cost_model(self) -> None:
routing = operator_orchestration()
self.assertEqual(worker_profiles.validate_orchestration(routing), [])
routing["profiles"]["codex-operator-standard"]["model"] = "gpt-other"
routing["modelAllowlist"]["codex"]["operator"]["standard"] = [
"gpt-other"
]
errors = worker_profiles.validate_orchestration(routing)
self.assertTrue(
any("operator default must use the Test default model" in error for error in errors),
errors,
)
def test_operator_is_standard_only_and_optional_for_legacy_projects(self) -> None:
routing = operator_orchestration()
operator = routing["profiles"]["codex-operator-standard"]
operator["tier"] = "strong"
routing["modelAllowlist"]["codex"]["operator"] = {
"strong": ["gpt-low"]
}
errors = worker_profiles.validate_orchestration(routing)
self.assertTrue(any("Operator may only use standard" in error for error in errors))
legacy = operator_orchestration()
del legacy["defaults"]["operator"]
del legacy["profiles"]["codex-operator-standard"]
del legacy["modelAllowlist"]["codex"]["operator"]
self.assertEqual(worker_profiles.validate_orchestration(legacy), [])
def test_delivery_operation_requires_a_supported_route_and_operator_dispatch(self) -> None:
board = routed_board()
self.assertEqual(validate_tasks.validate_builtin(board), [])
missing_operation = copy.deepcopy(board)
del missing_operation["tasks"][0]["operation"]
errors = validate_tasks.validate_builtin(missing_operation)
self.assertTrue(any("delivery-operation 必须声明 operation" in error for error in errors))
unsupported = copy.deepcopy(board)
unsupported["tasks"][0]["operation"]["skill"] = "shell"
errors = validate_tasks.validate_builtin(unsupported)
self.assertTrue(any("manage-release/deb-publisher/publish-docker-image" in error for error in errors))
missing_dispatch = copy.deepcopy(board)
del missing_dispatch["tasks"][0]["dispatch"]["operator"]
errors = validate_tasks.validate_builtin(missing_dispatch)
self.assertTrue(any("delivery-operation 必须声明 dispatch.operator" in error for error in errors))
def test_delivery_operation_cannot_be_reused_as_a_profile_delivery_run(self) -> None:
board = routed_board()
board["project"]["deliveryFile"] = "docs/ack/delivery.yaml"
board["tasks"][0]["status"] = "verified"
board["deliveryRuns"] = [
{
"id": "DR-duplicate-route",
"profile": "review",
"taskIds": ["DELIVERY-001"],
"status": "planned",
"sourceRevision": "a" * 40,
"configRevision": "b" * 40,
"pullRequest": None,
"artifacts": [],
"deployments": [],
"evidence": [],
"updatedAt": "2026-08-01T10:00:00+08:00",
}
]
errors = validate_tasks.validate_builtin(board)
self.assertTrue(
any("deliveryRuns 不能引用 delivery-operation" in error for error in errors),
errors,
)
def test_launcher_creates_an_operator_plan_on_the_low_cost_profile(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
project = Path(temporary).resolve()
executable = project / "codex"
executable.write_text("#!/bin/sh\n", encoding="utf-8")
executable.chmod(0o700)
routing = operator_orchestration()
routing["allowedWorktrees"] = [str(project)]
board = routed_board()
board["project"]["repoPath"] = str(project)
board["project"]["orchestration"] = routing
metadata = project.stat()
identity = {
"path": str(project),
"device": metadata.st_dev,
"inode": metadata.st_ino,
"gitCommonDir": str(project / ".git"),
"gitCommonDevice": metadata.st_dev,
"gitCommonInode": metadata.st_ino,
}
with (
mock.patch.object(
launch_worker,
"load_authoritative_board",
return_value=(project, board),
),
mock.patch.object(
launch_worker,
"capture_worktree_identity",
return_value=identity,
),
mock.patch.object(
launch_worker,
"resolve_executable",
return_value=executable,
),
mock.patch.object(
launch_worker,
"run_text",
return_value="codex-cli 1.0",
),
):
plan = launch_worker.build_plan(
project_root_value=str(project),
task_id="DELIVERY-001",
attempt_id="DELIVERY-001-A1",
role="operator",
profile_id="codex-operator-standard",
worktree_value=str(project),
slot=1,
)
receipt = launch_worker.build_receipt(
"c" * 64,
plan,
"runtime-1",
{
"handle": "terminal-1",
"incarnationId": "incarnation-1",
"connected": True,
"writable": True,
"worktreePath": str(project),
},
"2026-08-01T10:00:00+08:00",
)
receipt_errors = worker_profiles.validate_worker_receipt(
receipt,
orchestration=routing,
task_ids={"DELIVERY-001"},
)
self.assertEqual(plan["role"], "operator")
self.assertEqual(plan["requested"]["model"], "gpt-low")
self.assertEqual(
plan["requested"]["environmentPolicy"],
"per-cli-plus-operator-publish-v1",
)
self.assertTrue(plan["title"].startswith("ACK-OP-CODEX-STANDARD-"))
self.assertEqual(receipt_errors, [])
def test_launcher_binds_operator_to_delivery_operation_tasks(self) -> None:
delivery_board = routed_board()
ordinary_board = copy.deepcopy(delivery_board)
ordinary_task = ordinary_board["tasks"][0]
ordinary_task["id"] = "TASK-001"
ordinary_task["type"] = "feature"
del ordinary_task["operation"]
with mock.patch.object(
launch_worker,
"load_authoritative_board",
return_value=(Path("/repo/demo"), ordinary_board),
):
with self.assertRaisesRegex(
launch_worker.LaunchError,
"operator 只能用于 delivery-operation",
):
launch_worker.build_plan(
project_root_value="/repo/demo",
task_id="TASK-001",
attempt_id="TASK-001-A1",
role="operator",
profile_id="codex-operator-standard",
worktree_value="/repo/demo",
slot=1,
)
with mock.patch.object(
launch_worker,
"load_authoritative_board",
return_value=(Path("/repo/demo"), delivery_board),
):
with self.assertRaisesRegex(
launch_worker.LaunchError,
"delivery-operation 任务只能由 operator",
):
launch_worker.build_plan(
project_root_value="/repo/demo",
task_id="DELIVERY-001",
attempt_id="DELIVERY-001-A1",
role="test",
profile_id="codex-test-standard",
worktree_value="/repo/demo",
slot=1,
)
def test_operator_gets_only_fixed_release_credentials(self) -> None:
with mock.patch.dict(
"os.environ",
{
"OPENAI_API_KEY": "agent-token",
"DEB_TOKEN": "deb-token",
"DEB_SERVER_URL": "https://packages.example.com",
"DEB_REPOSITORY": "testing",
"SSH_AUTH_SOCK": "/tmp/agent.sock",
"GIT_SSH_COMMAND": "unsafe override",
"DOCKER_PASSWORD": "must-not-pass",
},
clear=True,
):
operator = launch_worker.worker_environment("codex", "operator")
test = launch_worker.worker_environment("codex", "test")
self.assertEqual(operator["DEB_TOKEN"], "deb-token")
self.assertEqual(operator["SSH_AUTH_SOCK"], "/tmp/agent.sock")
self.assertEqual(operator["OPENAI_API_KEY"], "agent-token")
self.assertNotIn("DEB_TOKEN", test)
self.assertNotIn("SSH_AUTH_SOCK", test)
self.assertNotIn("GIT_SSH_COMMAND", operator)
self.assertNotIn("DOCKER_PASSWORD", operator)
def test_ack_documents_the_three_routes_and_non_release_pr_boundary(self) -> None:
skill = (ACK_DIR / "SKILL.md").read_text(encoding="utf-8")
routing = (ACK_DIR / "references" / "delivery-routing.md").read_text(
encoding="utf-8"
)
docker = (
REPO_ROOT / "skills" / "publish-docker-image" / "SKILL.md"
).read_text(encoding="utf-8")
release = (
REPO_ROOT / "skills" / "manage-release" / "SKILL.md"
).read_text(encoding="utf-8")
self.assertIn("references/delivery-routing.md", skill)
for name in ("manage-release", "deb-publisher", "publish-docker-image"):
self.assertIn(name, routing)
self.assertIn("普通 PR/MR", routing)
self.assertIn("由显式调用的 `$ack`", docker)
self.assertIn("由显式调用的 `$ack`", release)
self.assertIn("PR-only", release)
if __name__ == "__main__":
unittest.main()
-1
View File
@@ -62,7 +62,6 @@ class AckSkillContentTests(unittest.TestCase):
"templates/delivery.schema.json",
"examples/delivery.example.yaml",
"references/delivery.md",
"references/delivery-routing.md",
):
self.assertTrue((ack_dir / relative_path).is_file(), relative_path)
version = (ack_dir / "VERSION").read_text(encoding="utf-8").strip()
+405
View File
@@ -0,0 +1,405 @@
from __future__ import annotations
import copy
import importlib.util
import json
import os
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
REPO_ROOT = Path(__file__).resolve().parents[1]
ORC_DIR = REPO_ROOT / "skills" / "orc"
SCRIPT = ORC_DIR / "scripts" / "resolve_profile.py"
CONFIG = ORC_DIR / "templates" / "config.template.yaml"
spec = importlib.util.spec_from_file_location("orc_resolve_profile", SCRIPT)
assert spec is not None and spec.loader is not None
orc_profiles = importlib.util.module_from_spec(spec)
spec.loader.exec_module(orc_profiles)
class OrcSkillTests(unittest.TestCase):
def test_orc_is_explicit_and_routes_to_independent_skills(self) -> None:
skill = (ORC_DIR / "SKILL.md").read_text(encoding="utf-8")
routing = (ORC_DIR / "references" / "routing.md").read_text(encoding="utf-8")
metadata = (ORC_DIR / "agents" / "openai.yaml").read_text(encoding="utf-8")
self.assertIn("allow_implicit_invocation: false", metadata)
self.assertIn("low", skill)
self.assertIn("mid", skill)
self.assertIn("high", skill)
self.assertIn("只支持 `cli: codex`", skill)
for child in ("$ack", "$manage-release", "$deb-publisher", "$publish-docker-image"):
self.assertIn(child, routing)
def test_children_do_not_reference_orc(self) -> None:
for child in ("ack", "manage-release", "deb-publisher", "publish-docker-image"):
for path in (REPO_ROOT / "skills" / child).rglob("*"):
if not path.is_file() or "__pycache__" in path.parts:
continue
content = path.read_text(encoding="utf-8", errors="ignore")
self.assertNotIn("$orc", content, str(path))
self.assertNotRegex(
content,
r"(?<![A-Za-z0-9_-])/orc(?![A-Za-z0-9_-])",
str(path),
)
self.assertNotRegex(content, r"\bORC\b", str(path))
self.assertNotIn("skills/orc", content, str(path))
def test_routing_pins_normal_pr_and_unsupported_release_gate(self) -> None:
routing = (ORC_DIR / "references" / "routing.md").read_text(
encoding="utf-8"
)
self.assertIn("普通非发布 PR/MR", routing)
self.assertIn("最多到", routing)
self.assertIn("review_ready", routing)
self.assertIn("当前 task 粒度无法安全表达", routing)
def test_template_and_cli_validate(self) -> None:
config = orc_profiles.load_config(CONFIG)
self.assertEqual(set(config["profiles"]), {"low", "mid", "high"})
completed = subprocess.run(
[sys.executable, str(SCRIPT), "validate", str(CONFIG)],
text=True,
capture_output=True,
check=False,
)
self.assertEqual(completed.returncode, 0, completed.stderr)
self.assertIn('"ok": true', completed.stdout)
clean_python = subprocess.run(
[sys.executable, "-I", "-S", str(SCRIPT), "validate", str(CONFIG)],
text=True,
capture_output=True,
check=False,
)
self.assertEqual(clean_python.returncode, 0, clean_python.stderr)
def test_level_precedence_is_stage_then_global_then_config(self) -> None:
config = orc_profiles.load_config(CONFIG)
stage = orc_profiles.resolve_profile(
config, stage="docker", global_level="mid", stage_level="high"
)
global_choice = orc_profiles.resolve_profile(
config, stage="docker", global_level="mid"
)
configured = orc_profiles.resolve_profile(config, stage="docker")
fallback_config = copy.deepcopy(config)
del fallback_config["stageDefaults"]["docker"]
fallback = orc_profiles.resolve_profile(fallback_config, stage="docker")
self.assertEqual((stage["level"], stage["selectionSource"]), ("high", "request.stage"))
self.assertEqual((global_choice["level"], global_choice["selectionSource"]), ("mid", "request.global"))
self.assertEqual((configured["level"], configured["selectionSource"]), ("low", "config.stageDefaults.docker"))
self.assertEqual((fallback["level"], fallback["selectionSource"]), ("mid", "config.defaultLevel"))
def test_resolver_builds_fixed_worker_args_without_free_command_fields(self) -> None:
config = orc_profiles.load_config(CONFIG)
plan = orc_profiles.resolve_profile(config, stage="release")
self.assertEqual(plan["workerArgs"][:2], ["--model", "gpt-5.6-terra"])
self.assertIn("--strict-config", plan["workerArgs"])
self.assertNotIn("danger-full-access", " ".join(plan["workerArgs"]))
self.assertNotIn("env", plan["profile"])
self.assertNotIn("command", plan["profile"])
def test_worker_environment_selects_one_model_and_remote_auth(self) -> None:
ambient = {
"OPENAI_API_KEY": "openai-secret",
"AZURE_OPENAI_API_KEY": "azure-secret",
"GITHUB_TOKEN": "github-secret",
"GITLAB_TOKEN": "gitlab-secret",
"GITEA_TOKEN": "gitea-secret",
"FORGEJO_TOKEN": "forgejo-secret",
"DEB_TOKEN": "deb-secret",
"SSH_AUTH_SOCK": "/tmp/agent.sock",
"HTTPS_PROXY": "https://user:secret@proxy.example",
}
with mock.patch.dict(os.environ, ambient, clear=True):
environment = orc_profiles.worker_environment(
"release",
model_auth="openai",
remote_auth="forgejo-token",
)
self.assertEqual(environment["OPENAI_API_KEY"], "openai-secret")
self.assertEqual(environment["FORGEJO_TOKEN"], "forgejo-secret")
for rejected in (
"AZURE_OPENAI_API_KEY",
"GITHUB_TOKEN",
"GITLAB_TOKEN",
"GITEA_TOKEN",
"DEB_TOKEN",
"SSH_AUTH_SOCK",
"HTTPS_PROXY",
):
self.assertNotIn(rejected, environment)
with self.assertRaisesRegex(orc_profiles.ConfigError, "deb remote auth"):
orc_profiles.worker_environment(
"deb",
model_auth="codex-login",
remote_auth="forgejo-token",
)
with mock.patch.dict(os.environ, {}, clear=True):
with self.assertRaisesRegex(
orc_profiles.ConfigError,
"selected authentication variable is unavailable",
):
orc_profiles.worker_environment(
"release",
model_auth="openai",
remote_auth="none",
)
with mock.patch.dict(os.environ, {"OPENAI_API_KEY": ""}, clear=True):
with self.assertRaisesRegex(
orc_profiles.ConfigError,
"selected authentication variable is unavailable",
):
orc_profiles.worker_environment(
"release",
model_auth="openai",
remote_auth="none",
)
def test_ssh_auth_requires_a_trusted_user_socket(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
socket_path = root / "agent.sock"
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as agent:
agent.bind(str(socket_path))
socket_path.chmod(0o600)
with mock.patch.dict(
os.environ,
{"SSH_AUTH_SOCK": str(socket_path)},
clear=True,
):
environment = orc_profiles.worker_environment(
"release",
model_auth="codex-login",
remote_auth="ssh-agent",
)
self.assertEqual(environment["SSH_AUTH_SOCK"], str(socket_path))
regular_file = root / "not-a-socket"
regular_file.write_text("not a socket", encoding="utf-8")
regular_file.chmod(0o600)
with mock.patch.dict(
os.environ,
{"SSH_AUTH_SOCK": str(regular_file)},
clear=True,
):
with self.assertRaisesRegex(
orc_profiles.ConfigError,
"not a trusted user socket",
):
orc_profiles.worker_environment(
"release",
model_auth="codex-login",
remote_auth="ssh-agent",
)
def test_trusted_orca_ignores_ambient_path(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
trusted = root / "trusted"
hostile = root / "hostile"
trusted.mkdir()
hostile.mkdir()
trusted_orca = trusted / "orca"
hostile_orca = hostile / "orca"
for executable in (trusted_orca, hostile_orca):
executable.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
executable.chmod(0o700)
with (
mock.patch.object(
orc_profiles,
"trusted_path_entries",
return_value=[trusted],
),
mock.patch.dict(os.environ, {"PATH": str(hostile)}, clear=True),
):
resolved = orc_profiles.resolve_trusted_executable("orca")
self.assertEqual(resolved, trusted_orca)
def test_config_rejects_unsafe_or_ambiguous_profiles(self) -> None:
base = orc_profiles.load_config(CONFIG)
free_command = copy.deepcopy(base)
free_command["profiles"]["low"]["command"] = "codex --dangerously-bypass"
with self.assertRaisesRegex(orc_profiles.ConfigError, "unknown fields"):
orc_profiles.validate_config(free_command)
full_access = copy.deepcopy(base)
full_access["profiles"]["high"]["permissionMode"] = "danger-full-access"
with self.assertRaisesRegex(orc_profiles.ConfigError, "workspace-write in ORC v1"):
orc_profiles.validate_config(full_access)
read_only = copy.deepcopy(base)
read_only["profiles"]["low"]["permissionMode"] = "read-only"
with self.assertRaisesRegex(orc_profiles.ConfigError, "workspace-write in ORC v1"):
orc_profiles.validate_config(read_only)
missing_level = copy.deepcopy(base)
del missing_level["profiles"]["mid"]
with self.assertRaisesRegex(orc_profiles.ConfigError, "missing fields: mid"):
orc_profiles.validate_config(missing_level)
option_model = copy.deepcopy(base)
option_model["profiles"]["low"]["model"] = "--model"
with self.assertRaisesRegex(orc_profiles.ConfigError, "safe exact model ID"):
orc_profiles.validate_config(option_model)
list_policy = copy.deepcopy(base)
list_policy["profiles"]["low"]["approvalPolicy"] = ["never"]
with self.assertRaisesRegex(orc_profiles.ConfigError, "approvalPolicy"):
orc_profiles.validate_config(list_policy)
def test_config_reader_rejects_symlinks_and_redacts_parser_input(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
real = root / "real.yaml"
real.write_text(CONFIG.read_text(encoding="utf-8"), encoding="utf-8")
linked = root / "linked.yaml"
linked.symlink_to(real)
with self.assertRaisesRegex(orc_profiles.ConfigError, "safely read"):
orc_profiles.load_config(linked)
malformed = root / "malformed.yaml"
secret_marker = "PRIVATE_MATERIAL_MUST_NOT_APPEAR"
malformed.write_text(f"{secret_marker}: [unterminated\n", encoding="utf-8")
completed = subprocess.run(
[sys.executable, str(SCRIPT), "validate", str(malformed)],
text=True,
capture_output=True,
check=False,
)
self.assertEqual(completed.returncode, 1)
self.assertNotIn(secret_marker, completed.stderr)
def test_bound_plan_enforces_registered_allowlisted_worktree(self) -> None:
git = shutil.which("git")
self.assertIsNotNone(git)
assert git is not None
git_path = Path(git).resolve()
with tempfile.TemporaryDirectory(prefix="orc shell $(id) ' ") as temporary:
temporary_root = Path(temporary).resolve()
root = temporary_root / "project"
root.mkdir()
subprocess.run(
[str(git_path), "init", "-q", str(root)],
check=True,
capture_output=True,
)
config_path = root / "docs" / "orc" / "config.yaml"
config_path.parent.mkdir(parents=True)
config_path.write_text(CONFIG.read_text(encoding="utf-8"), encoding="utf-8")
fake_codex = root / "trusted-codex" / "codex"
fake_codex.parent.mkdir()
fake_codex.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
fake_codex.chmod(0o700)
fake_orca = root / "trusted-orca" / "orca"
fake_orca.parent.mkdir()
fake_orca.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
fake_orca.chmod(0o700)
def executable(name: str) -> Path:
return {
"git": git_path,
"codex": fake_codex,
"orca": fake_orca,
}[name]
with (
mock.patch.object(
orc_profiles,
"resolve_trusted_executable",
side_effect=executable,
),
mock.patch.object(
orc_profiles,
"resolve_trusted_python",
return_value=Path(sys.executable).resolve(),
),
mock.patch.object(
orc_profiles,
"_executable_facts",
return_value={
"path": str(fake_codex),
"device": 1,
"inode": 2,
"size": 3,
"mtimeNs": 4,
"version": "codex-test 1",
},
),
):
plan = orc_profiles.build_launch_plan(
config_path,
project_root=root,
worktree=root,
stage="code",
stage_level="high",
)
self.assertEqual(plan["argv"][0], str(fake_codex))
self.assertEqual(plan["terminalCreateArgv"][0], str(fake_orca))
self.assertEqual(plan["worktree"], str(root))
self.assertEqual(plan["selectionSource"], "request.stage")
self.assertTrue(plan["launchFingerprint"].startswith("sha256:"))
self.assertIn("_launch", plan["launcherArgv"])
self.assertEqual(plan["launcherArgv"][1:3], ["-I", "-S"])
self.assertIn("python", plan)
self.assertEqual(plan["worktreeSelector"], f"path:{root}")
self.assertNotIn("PRIVATE", json.dumps(plan))
self.assertEqual(shlex.split(plan["terminalCommand"]), plan["launcherArgv"])
self.assertEqual(
shlex.split(plan["terminalCreateShellCommand"]),
plan["terminalCreateArgv"],
)
command_index = plan["terminalCreateArgv"].index("--command")
self.assertEqual(
plan["terminalCreateArgv"][command_index + 1],
plan["terminalCommand"],
)
outside = temporary_root / "outside"
outside.mkdir()
subprocess.run(
[str(git_path), "init", "-q", str(outside)],
check=True,
capture_output=True,
)
with mock.patch.object(
orc_profiles,
"resolve_trusted_executable",
side_effect=executable,
):
with self.assertRaisesRegex(
orc_profiles.ConfigError,
"not in allowedWorktrees",
):
orc_profiles.validate_worktree(
orc_profiles.load_config(config_path),
project_root_value=root,
worktree_value=outside,
)
if __name__ == "__main__":
unittest.main()