from __future__ import annotations import copy import sys import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] SCRIPTS_DIR = REPO_ROOT / "skills" / "ack" / "scripts" sys.path.insert(0, str(SCRIPTS_DIR)) import worker_profiles # noqa: E402 def valid_orchestration() -> dict: return { "profileVersion": 1, "mode": "orca", "allowedWorktrees": ["/repo/demo"], "modelAllowlist": { "codex": { "developer": { "standard": ["gpt-safe-dev"], "strong": ["gpt-safe-strong"], }, "test": {"standard": ["gpt-safe-test"]}, }, "cursor-agent": { "developer": {"standard": ["cursor-auto"]}, "test": {"standard": ["cursor-auto"]}, }, "grok": { "developer": { "standard": ["grok-4.5"], "strong": ["grok-4.6"], }, "test": {"standard": ["grok-4.5"]}, }, }, "profiles": { "codex-dev-standard": { "role": "developer", "cli": "codex", "tier": "standard", "model": "gpt-safe-dev", "reasoningEffort": "medium", "permissionMode": "workspace-write", }, "codex-dev-strong": { "role": "developer", "cli": "codex", "tier": "strong", "model": "gpt-safe-strong", "reasoningEffort": "high", "permissionMode": "workspace-write", }, "cursor-test-standard": { "role": "test", "cli": "cursor-agent", "tier": "standard", "model": "cursor-auto", "reasoningEffort": None, "permissionMode": "read-only", }, "cursor-dev-standard": { "role": "developer", "cli": "cursor-agent", "tier": "standard", "model": "cursor-auto", "reasoningEffort": None, "permissionMode": "workspace-write", }, "grok-dev-standard": { "role": "developer", "cli": "grok", "tier": "standard", "model": "grok-4.5", "reasoningEffort": "medium", "permissionMode": "workspace-write", }, "grok-dev-strong": { "role": "developer", "cli": "grok", "tier": "strong", "model": "grok-4.6", "reasoningEffort": "high", "permissionMode": "workspace-write", }, "grok-test-readonly": { "role": "test", "cli": "grok", "tier": "standard", "model": "grok-4.5", "reasoningEffort": "low", "permissionMode": "read-only", }, }, "defaults": { "developer": "codex-dev-standard", "test": "cursor-test-standard", "developerUpgraded": "codex-dev-strong", }, } def valid_receipt(orchestration: dict | None = None) -> dict: routing = orchestration or valid_orchestration() profile_id = "codex-dev-standard" profile = routing["profiles"][profile_id] executable = "/usr/local/bin/codex" worktree_path = "/repo/demo" argv = worker_profiles.render_worker_argv(profile, executable, worktree_path) launch_id = "a" * 64 requested = { "cli": profile["cli"], "tier": profile["tier"], "model": profile["model"], "reasoningEffort": profile["reasoningEffort"], "permissionMode": profile["permissionMode"], "executable": executable, "executableDevice": 8, "executableInode": 201, "cliVersion": "codex 1.0.0", "argv": argv, "argvHash": worker_profiles.canonical_sha256(argv), "environmentPolicy": "per-cli-allowlist-v1", } worktree = { "path": worktree_path, "device": 8, "inode": 101, "gitCommonDir": "/repo/demo/.git", "gitCommonDevice": 8, "gitCommonInode": 102, } receipt = { "receiptVersion": 1, "id": f"WR-{launch_id}", "launchId": launch_id, "profileId": profile_id, "profileHash": worker_profiles.profile_hash(profile), "launchFingerprint": worker_profiles.canonical_sha256( { "protocolVersion": 1, "backend": "orca", "profileId": profile_id, "profileHash": worker_profiles.profile_hash(profile), "createdFor": { "taskId": "TASK-001", "attemptId": "TASK-001-A1", "role": "developer", }, "worktree": worktree, "requested": requested, "slot": 1, } ), "slot": 1, "createdFor": { "taskId": "TASK-001", "attemptId": "TASK-001-A1", "role": "developer", }, "worktree": worktree, "requested": requested, "binding": { "orchestrator": "orca", "runtimeId": "runtime-001", "handle": "terminal-001", "incarnationId": "incarnation-001", "observedWorktreePath": worktree_path, "connected": True, "writable": True, "boundAt": "2026-07-31T12:00:01+08:00", }, "createdAt": "2026-07-31T12:00:00+08:00", "receiptHash": "", } receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) return receipt class CanonicalHashTests(unittest.TestCase): def test_hash_is_prefixed_order_independent_and_content_sensitive(self) -> None: first = worker_profiles.canonical_sha256( {"b": [2, 1], "a": {"enabled": True}} ) reordered = worker_profiles.canonical_sha256( {"a": {"enabled": True}, "b": [2, 1]} ) changed = worker_profiles.canonical_sha256( {"a": {"enabled": False}, "b": [2, 1]} ) self.assertEqual(first, reordered) self.assertNotEqual(first, changed) self.assertRegex(first, r"^sha256:[0-9a-f]{64}$") def test_hash_rejects_non_json_and_non_finite_values(self) -> None: with self.assertRaises(ValueError): worker_profiles.canonical_sha256({"bad": object()}) with self.assertRaises(ValueError): worker_profiles.canonical_sha256({"bad": float("nan")}) cyclic: list = [] cyclic.append(cyclic) with self.assertRaises(ValueError): worker_profiles.canonical_sha256(cyclic) def test_profile_hash_rejects_invalid_profile(self) -> None: profile = valid_orchestration()["profiles"]["codex-dev-standard"] profile["command"] = "codex; touch forged" with self.assertRaises(ValueError): worker_profiles.profile_hash(profile) def test_profile_hash_binds_profile_version(self) -> None: profile = valid_orchestration()["profiles"]["codex-dev-standard"] self.assertNotEqual( worker_profiles.profile_hash(profile, profile_version=1), worker_profiles.profile_hash(profile, profile_version=2), ) class ProfileValidationTests(unittest.TestCase): def test_valid_orchestration_passes(self) -> None: self.assertEqual( worker_profiles.validate_orchestration(valid_orchestration()), [], ) def test_manual_mode_accepts_empty_structured_routing(self) -> None: routing = { "profileVersion": 1, "mode": "manual", "allowedWorktrees": [], "modelAllowlist": {}, "profiles": {}, "defaults": {}, } self.assertEqual(worker_profiles.validate_orchestration(routing), []) def test_partial_empty_allowlist_nodes_are_rejected(self) -> None: routing = { "profileVersion": 1, "mode": "manual", "allowedWorktrees": [], "modelAllowlist": {"codex": {}}, "profiles": {}, "defaults": {}, } errors = worker_profiles.validate_orchestration(routing) self.assertTrue(any("modelAllowlist.codex: must not be empty" in error for error in errors)) def test_profile_rejects_all_command_shaped_and_unknown_fields(self) -> None: forbidden = ("command", "args", "env", "executable", "argv", "extraArgs") for field in forbidden: with self.subTest(field=field): routing = valid_orchestration() routing["profiles"]["codex-dev-standard"][field] = "forged" errors = worker_profiles.validate_orchestration(routing) self.assertTrue(any(f"unknown field '{field}'" in error for error in errors)) def test_routing_rejects_unknown_top_level_fields(self) -> None: routing = valid_orchestration() routing["command"] = "codex" errors = worker_profiles.validate_orchestration(routing) self.assertIn("project.orchestration: unknown field 'command'", errors) def test_model_id_rejects_shell_and_whitespace_syntax(self) -> None: for model in ( "safe;touch-forged", "safe && forged", "$(touch-forged)", "`touch-forged`", "safe\nforged", "--dangerously-bypass-approvals-and-sandbox", ): with self.subTest(model=model): routing = valid_orchestration() routing["profiles"]["codex-dev-standard"]["model"] = model errors = worker_profiles.validate_orchestration(routing) self.assertTrue(any("safe model ID" in error for error in errors)) def test_only_safe_permission_modes_are_accepted(self) -> None: for permission in ( "danger-full-access", "full-access", "yolo", "bypass", "never", ): with self.subTest(permission=permission): routing = valid_orchestration() routing["profiles"]["codex-dev-standard"][ "permissionMode" ] = permission errors = worker_profiles.validate_orchestration(routing) self.assertTrue(any("read-only/workspace-write" in error for error in errors)) def test_reasoning_effort_is_required_for_codex_and_null_for_cursor(self) -> None: codex = valid_orchestration() codex["profiles"]["codex-dev-standard"]["reasoningEffort"] = None cursor = valid_orchestration() cursor["profiles"]["cursor-test-standard"]["reasoningEffort"] = "low" grok = valid_orchestration() grok["profiles"]["grok-dev-standard"]["reasoningEffort"] = None codex_errors = worker_profiles.validate_orchestration(codex) cursor_errors = worker_profiles.validate_orchestration(cursor) grok_errors = worker_profiles.validate_orchestration(grok) self.assertTrue(any("Codex requires" in error for error in codex_errors)) self.assertTrue(any("Cursor requires null" in error for error in cursor_errors)) self.assertTrue(any("Grok requires" in error for error in grok_errors)) def test_test_cannot_use_strong_tier(self) -> None: routing = valid_orchestration() profile = routing["profiles"]["cursor-test-standard"] profile["tier"] = "strong" routing["modelAllowlist"]["cursor-agent"]["test"]["strong"] = [ "cursor-auto" ] errors = worker_profiles.validate_orchestration(routing) self.assertTrue(any("Test may only use standard" in error for error in errors)) self.assertTrue(any("Test cannot define a strong allowlist" in error for error in errors)) def test_profile_model_must_match_exact_cli_role_tier_allowlist(self) -> None: routing = valid_orchestration() routing["profiles"]["codex-dev-standard"]["model"] = "other-safe-model" errors = worker_profiles.validate_orchestration(routing) self.assertTrue(any("is not allowed for its cli/role/tier" in error for error in errors)) def test_defaults_require_matching_role_and_standard_tier(self) -> None: wrong_role = valid_orchestration() wrong_role["defaults"]["test"] = "codex-dev-standard" strong_default = valid_orchestration() strong_default["defaults"]["developer"] = "codex-dev-strong" role_errors = worker_profiles.validate_orchestration(wrong_role) tier_errors = worker_profiles.validate_orchestration(strong_default) self.assertTrue(any("profile role must be test" in error for error in role_errors)) self.assertTrue(any("default profile must use standard" in error for error in tier_errors)) def test_developer_upgraded_default_requires_developer_strong(self) -> None: valid = valid_orchestration() wrong_tier = valid_orchestration() wrong_tier["defaults"]["developerUpgraded"] = "codex-dev-standard" wrong_role = valid_orchestration() wrong_role["defaults"]["developerUpgraded"] = "cursor-test-standard" self.assertEqual(worker_profiles.validate_orchestration(valid), []) tier_errors = worker_profiles.validate_orchestration(wrong_tier) role_errors = worker_profiles.validate_orchestration(wrong_role) self.assertTrue(any("must use strong tier" in error for error in tier_errors)) self.assertTrue(any("profile role must be developer" in error for error in role_errors)) def test_malformed_scalar_types_return_errors_instead_of_raising(self) -> None: for field in ("role", "cli", "tier", "reasoningEffort", "permissionMode"): with self.subTest(profile_field=field): profile = valid_orchestration()["profiles"]["codex-dev-standard"] profile[field] = [] self.assertTrue(worker_profiles.validate_profile(profile)) routing = valid_orchestration() routing["mode"] = [] self.assertTrue(worker_profiles.validate_orchestration(routing)) receipt = valid_receipt() receipt["profileId"] = [] receipt["requested"]["cli"] = [] receipt["requested"]["tier"] = [] receipt["requested"]["permissionMode"] = [] receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) self.assertTrue(worker_profiles.validate_worker_receipt(receipt)) self.assertTrue( worker_profiles.validate_worker_receipt(receipt, orchestration=[]) ) def test_orca_requires_defaults_and_profiles_and_tolerates_legacy_allowlist(self) -> None: routing = valid_orchestration() routing["allowedWorktrees"] = [] routing["profiles"] = {} routing["defaults"] = {} errors = worker_profiles.validate_orchestration(routing) # allowedWorktrees 白名单已废弃（v0.19）：遗留字段容忍，不再校验 self.assertTrue( all("requires at least one path" not in error for error in errors) ) self.assertTrue(any("Orca mode requires profiles" in error for error in errors)) self.assertTrue(any("missing role 'developer'" in error for error in errors)) self.assertTrue(any("missing role 'test'" in error for error in errors)) class ArgvRendererTests(unittest.TestCase): def test_codex_exact_safe_argv(self) -> None: profile = valid_orchestration()["profiles"]["codex-dev-standard"] argv = worker_profiles.render_worker_argv( profile, "/usr/local/bin/codex", "/repo/demo", ) self.assertEqual( argv, [ "/usr/local/bin/codex", "--strict-config", "--model", "gpt-safe-dev", "--config", "model_reasoning_effort=medium", "--sandbox", "workspace-write", "--ask-for-approval", "never", "--cd", "/repo/demo", ], ) def test_cursor_read_only_exact_safe_argv(self) -> None: profile = valid_orchestration()["profiles"]["cursor-test-standard"] argv = worker_profiles.render_worker_argv( profile, "/usr/local/bin/cursor-agent", "/repo/demo", ) self.assertEqual( argv, [ "/usr/local/bin/cursor-agent", "--model", "cursor-auto", "--mode", "plan", "--sandbox", "enabled", "--workspace", "/repo/demo", ], ) def test_cursor_workspace_write_adds_auto_review_without_yolo(self) -> None: profile = valid_orchestration()["profiles"]["cursor-dev-standard"] argv = worker_profiles.render_worker_argv( profile, "/usr/local/bin/cursor-agent", "/repo/demo", ) self.assertEqual( argv, [ "/usr/local/bin/cursor-agent", "--model", "cursor-auto", "--auto-review", "--sandbox", "enabled", "--workspace", "/repo/demo", ], ) self.assertNotIn("--yolo", argv) self.assertNotIn("--force", argv) def test_grok_workspace_write_exact_safe_argv(self) -> None: profile = valid_orchestration()["profiles"]["grok-dev-standard"] argv = worker_profiles.render_worker_argv( profile, "/usr/local/bin/grok", "/repo/demo", ) self.assertEqual( argv, [ "/usr/local/bin/grok", "--model", "grok-4.5", "--reasoning-effort", "medium", "--permission-mode", "acceptEdits", "--always-approve", "--sandbox", "workspace", "--cwd", "/repo/demo", ], ) for forbidden in ( "--yolo", "bypassPermissions", "auto", "dontAsk", "off", ): self.assertNotIn(forbidden, argv) def test_grok_read_only_uses_plan_and_read_only_sandbox(self) -> None: profile = valid_orchestration()["profiles"]["grok-test-readonly"] argv = worker_profiles.render_worker_argv( profile, "/usr/local/bin/grok", "/repo/demo", ) self.assertEqual( argv, [ "/usr/local/bin/grok", "--model", "grok-4.5", "--reasoning-effort", "low", "--permission-mode", "plan", "--always-approve", "--sandbox", "read-only", "--cwd", "/repo/demo", ], ) self.assertNotIn("--yolo", argv) self.assertNotIn("bypassPermissions", argv) def test_grok_accepts_vendor_artifact_basename(self) -> None: profile = valid_orchestration()["profiles"]["grok-dev-standard"] artifact = "/home/ace/.grok/downloads/grok-linux-x86_64" argv = worker_profiles.render_worker_argv(profile, artifact, "/repo/demo") self.assertEqual(argv[0], artifact) self.assertTrue( worker_profiles.executable_basename_matches_cli(artifact, "grok") ) self.assertFalse( worker_profiles.executable_basename_matches_cli( artifact, "cursor-agent" ) ) def test_renderer_rejects_wrong_executable_or_unsafe_worktree(self) -> None: profile = valid_orchestration()["profiles"]["codex-dev-standard"] with self.assertRaises(ValueError): worker_profiles.render_worker_argv( profile, "/tmp/cursor-agent", "/repo/demo", ) with self.assertRaises(ValueError): worker_profiles.render_worker_argv( profile, "/usr/local/bin/codex", "/repo/demo/../outside", ) for worktree in ("/repo/./demo", "/repo/demo/", "//repo/demo"): with self.subTest(worktree=worktree), self.assertRaises(ValueError): worker_profiles.render_worker_argv( profile, "/usr/local/bin/codex", worktree, ) class ReceiptValidationTests(unittest.TestCase): def test_valid_receipt_is_bound_to_profile_and_routing(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) self.assertEqual( worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"TASK-001"}, ), [], ) self.assertEqual(receipt["receiptHash"], worker_profiles.receipt_hash(receipt)) def test_receipt_rejects_unknown_fields_at_every_strict_level(self) -> None: cases = ( ((), "command"), (("createdFor",), "command"), (("worktree",), "command"), (("requested",), "command"), (("binding",), "command"), ) for path, field in cases: with self.subTest(path=path): routing = valid_orchestration() receipt = valid_receipt(routing) target = receipt for component in path: target = target[component] target[field] = "forged" receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) errors = worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"TASK-001"}, ) self.assertTrue(any("unknown field 'command'" in error for error in errors)) def test_receipt_rejects_profile_and_exact_argv_drift(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) receipt["requested"]["model"] = "gpt-safe-strong" receipt["requested"]["argv"][3] = "gpt-safe-strong" receipt["requested"]["argvHash"] = worker_profiles.canonical_sha256( receipt["requested"]["argv"] ) receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) errors = worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"TASK-001"}, ) self.assertTrue(any("requested.model: does not match profile" in error for error in errors)) self.assertTrue( any( "requested.argv: does not match exact renderer" in error for error in errors ) ) def test_receipt_rejects_tampering_without_recomputed_hash(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) receipt["binding"]["handle"] = "forged-handle" errors = worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"TASK-001"}, ) self.assertTrue(any("receiptHash: does not match receipt" in error for error in errors)) def test_receipt_rejects_out_of_budget_attempt_empty_argv_and_id_drift(self) -> None: routing = valid_orchestration() attempt = valid_receipt(routing) attempt["createdFor"]["attemptId"] = "TASK-001-A4" attempt["receiptHash"] = worker_profiles.receipt_hash(attempt) attempt_errors = worker_profiles.validate_worker_receipt( attempt, orchestration=routing, task_ids={"TASK-001"}, ) empty_argv = valid_receipt(routing) empty_argv["requested"]["argv"] = [] empty_argv["requested"]["argvHash"] = worker_profiles.canonical_sha256([]) empty_argv["receiptHash"] = worker_profiles.receipt_hash(empty_argv) argv_errors = worker_profiles.validate_worker_receipt( empty_argv, orchestration=routing, task_ids={"TASK-001"}, ) wrong_id = valid_receipt(routing) wrong_id["id"] = "WR-" + "b" * 64 wrong_id["receiptHash"] = worker_profiles.receipt_hash(wrong_id) id_errors = worker_profiles.validate_worker_receipt( wrong_id, orchestration=routing, task_ids={"TASK-001"}, ) self.assertTrue(any("A1..A3" in error for error in attempt_errors)) self.assertTrue(any("at least 2 items" in error for error in argv_errors)) self.assertTrue(any("must equal 'WR-' + launchId" in error for error in id_errors)) def test_receipt_rejects_rehashed_launch_fingerprint_tampering(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) receipt["launchFingerprint"] = worker_profiles.canonical_sha256( {"forged": True} ) receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) errors = worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"TASK-001"}, ) self.assertTrue( any( "launchFingerprint: does not match launch facts" in error for error in errors ) ) def test_receipt_accepts_project_and_board_bound_launch_fingerprint(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) receipt["projectRoot"] = "/repo/demo" receipt["boardHash"] = worker_profiles.canonical_sha256({"tasks": []}) receipt["launchFingerprint"] = worker_profiles.canonical_sha256({ "protocolVersion": 1, "backend": "orca", "projectRoot": receipt["projectRoot"], "boardHash": receipt["boardHash"], "profileId": receipt["profileId"], "profileHash": receipt["profileHash"], "createdFor": receipt["createdFor"], "worktree": receipt["worktree"], "requested": receipt["requested"], "slot": receipt["slot"], }) receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) self.assertEqual( worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"TASK-001"}, ), [], ) def test_receipt_slot_is_bounded_and_bound_into_launch_fingerprint(self) -> None: routing = valid_orchestration() invalid = valid_receipt(routing) invalid["slot"] = 0 invalid["receiptHash"] = worker_profiles.receipt_hash(invalid) invalid_errors = worker_profiles.validate_worker_receipt( invalid, orchestration=routing, task_ids={"TASK-001"}, ) drifted = valid_receipt(routing) drifted["slot"] = 2 drifted["receiptHash"] = worker_profiles.receipt_hash(drifted) drifted_errors = worker_profiles.validate_worker_receipt( drifted, orchestration=routing, task_ids={"TASK-001"}, ) self.assertTrue(any("slot: must be" in error for error in invalid_errors)) self.assertTrue( any( "launchFingerprint: does not match launch facts" in error for error in drifted_errors ) ) def test_receipt_with_non_json_data_returns_errors(self) -> None: receipt = valid_receipt() receipt["requested"]["argv"] = [object()] errors = worker_profiles.validate_worker_receipt(receipt) def test_receipt_rejects_wrong_worktree_task_and_binding(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) receipt["worktree"]["path"] = "/repo/other" receipt["binding"]["observedWorktreePath"] = "/repo/elsewhere" receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) errors = worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"OTHER-TASK"}, ) self.assertTrue(any("unknown task 'TASK-001'" in error for error in errors)) # allowedWorktrees 白名单已废弃（v0.19）：receipt 不再按白名单校验 worktree 路径， # 只要求 binding 观测路径与 worktree 快照一致。 self.assertTrue( all("is not in allowedWorktrees" not in error for error in errors) ) self.assertTrue(any("does not match worktree.path" in error for error in errors)) def test_receipt_requires_safe_environment_and_live_binding(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) receipt["requested"]["environmentPolicy"] = "inherit-all" receipt["binding"]["connected"] = False receipt["binding"]["writable"] = False receipt["receiptHash"] = worker_profiles.receipt_hash(receipt) errors = worker_profiles.validate_worker_receipt( receipt, orchestration=routing, task_ids={"TASK-001"}, ) self.assertTrue(any("per-cli-allowlist-v1" in error for error in errors)) self.assertTrue(any("binding.connected: must be true" in error for error in errors)) self.assertTrue(any("binding.writable: must be true" in error for error in errors)) def test_receipt_list_rejects_duplicate_receipt_and_launch_ids(self) -> None: routing = valid_orchestration() first = valid_receipt(routing) duplicate = copy.deepcopy(first) errors = worker_profiles.validate_worker_receipts( [first, duplicate], routing, task_ids={"TASK-001"}, ) self.assertTrue(any("duplicate receipt ID" in error for error in errors)) self.assertTrue(any("duplicate launch ID" in error for error in errors)) def test_document_requires_top_level_receipts_and_routing(self) -> None: missing = { "version": 1, "project": {"name": "demo"}, "tasks": [], } errors = worker_profiles.validate_routing_document(missing) self.assertEqual(errors, ["project.orchestration: is required"]) def test_valid_document_passes_and_manual_receipts_fail(self) -> None: routing = valid_orchestration() document = { "version": 1, "project": {"name": "demo", "orchestration": routing}, "workerReceipts": [valid_receipt(routing)], "tasks": [{"id": "TASK-001", "title": "demo", "status": "open"}], } self.assertEqual(worker_profiles.validate_routing_document(document), []) manual = { "version": 1, "project": { "name": "demo", "orchestration": { "profileVersion": 1, "mode": "manual", "allowedWorktrees": [], "modelAllowlist": {}, "profiles": {}, "defaults": {}, }, }, "workerReceipts": [valid_receipt(routing)], "tasks": [{"id": "TASK-001", "title": "demo", "status": "open"}], } manual_errors = worker_profiles.validate_routing_document(manual) self.assertTrue( any("manual orchestration requires an empty list" in error for error in manual_errors) ) def test_dispatch_receipt_allows_two_phase_for_exact_task_attempt_binding(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) role_dispatch = { "profileId": receipt["profileId"], "receiptId": receipt["id"], "attemptId": "TASK-001-A1", "taskId": None, "dispatchId": None, } document = { "version": 1, "project": {"name": "demo", "orchestration": routing}, "workerReceipts": [receipt], "tasks": [ { "id": "TASK-001", "title": "created for", "status": "dispatched", "dispatch": {"developer": role_dispatch}, }, ], } self.assertEqual(worker_profiles.validate_routing_document(document), []) role_dispatch["taskId"] = "orca-task-002" role_dispatch["dispatchId"] = "orca-dispatch-002" self.assertEqual(worker_profiles.validate_routing_document(document), []) def test_dispatch_rejects_unknown_or_mismatched_receipt(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) role_dispatch = { "profileId": "codex-dev-strong", "receiptId": receipt["id"], "attemptId": "TASK-001-A1", "taskId": None, "dispatchId": None, } document = { "version": 1, "project": {"name": "demo", "orchestration": routing}, "workerReceipts": [receipt], "tasks": [ { "id": "TASK-001", "title": "demo", "status": "open", "dispatch": {"developer": role_dispatch}, } ], } errors = worker_profiles.validate_routing_document(document) self.assertTrue( any("profileId: does not match referenced receipt" in error for error in errors) ) role_dispatch["receiptId"] = f"WR-{'b' * 64}" errors = worker_profiles.validate_routing_document(document) self.assertTrue(any("unknown receipt" in error for error in errors)) def test_dispatch_rejects_cross_task_role_and_attempt_receipts(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) link = { "profileId": receipt["profileId"], "receiptId": receipt["id"], "attemptId": "TASK-001-A1", "taskId": None, "dispatchId": None, } document = { "version": 1, "project": {"name": "demo", "orchestration": routing}, "workerReceipts": [receipt], "tasks": [ {"id": "TASK-001", "title": "receipt owner", "status": "open"}, { "id": "TASK-002", "title": "must not reuse receipt", "status": "dispatched", "dispatch": {"developer": link}, }, ], } errors = worker_profiles.validate_routing_document(document) self.assertTrue(any("must be current ACK task 'TASK-002'" in error for error in errors)) self.assertTrue(any("must belong to current ACK task 'TASK-002'" in error for error in errors)) document["tasks"][1]["id"] = "TASK-001" link["attemptId"] = "TASK-001-A2" errors = worker_profiles.validate_routing_document(document) self.assertTrue(any("does not match referenced receipt" in error for error in errors)) link["attemptId"] = "TASK-001-A1" document["tasks"][1]["dispatch"] = {"test": link} errors = worker_profiles.validate_routing_document(document) self.assertTrue(any("referenced receipt role must be test" in error for error in errors)) def test_dispatch_attempt_presence_tracks_receipt_presence(self) -> None: routing = valid_orchestration() receipt = valid_receipt(routing) link = { "profileId": receipt["profileId"], "receiptId": None, "attemptId": "TASK-001-A1", "taskId": None, "dispatchId": None, } document = { "version": 1, "project": {"name": "demo", "orchestration": routing}, "workerReceipts": [receipt], "tasks": [ { "id": "TASK-001", "title": "demo", "status": "open", "dispatch": {"developer": link}, } ], } errors = worker_profiles.validate_routing_document(document) self.assertTrue(any("must be null when receiptId is null" in error for error in errors)) link["receiptId"] = receipt["id"] link["attemptId"] = None errors = worker_profiles.validate_routing_document(document) self.assertTrue(any("is required when receiptId is set" in error for error in errors)) def test_dispatch_requires_receipt_before_paired_runtime_ids(self) -> None: routing = valid_orchestration() document = { "version": 1, "project": {"name": "demo", "orchestration": routing}, "workerReceipts": [], "tasks": [ { "id": "TASK-001", "title": "demo", "status": "dispatched", "dispatch": { "developer": { "profileId": "codex-dev-standard", "receiptId": None, "attemptId": None, "taskId": "orca-task-001", "dispatchId": None, } }, } ], } errors = worker_profiles.validate_routing_document(document) self.assertTrue(any("must both be null or both be set" in error for error in errors)) self.assertTrue(any("required before runtime dispatch IDs" in error for error in errors)) if __name__ == "__main__": unittest.main()