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()