"""Small contract checks for terminal construction and resource settlement.""" from __future__ import annotations import json import sys import tempfile from pathlib import Path from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] SCRIPTS = ROOT / "skills" / "agent-orchestration-skill" / "scripts" sys.path.insert(0, str(SCRIPTS)) import orca_adapter # noqa: E402 import run_ledger # noqa: E402 from orca_adapter import agent_cli_command, build_worker_command, launch_manifest_path, resource_action, terminal_name # noqa: E402 from workflow_config import profile_config, write_default # noqa: E402 def config() -> dict: return { "version": 1, "routes": {"default": "light", "user_selectable": ["light", "medium", "hard"]}, "runtime": {"agent": "codex", "max_parallel_workers": 2, "terminal_name_template": "{size}-{stage}-{model}-{reasoning}"}, "profiles": [ { "role": "planner", "provider": "openai", "agent_cli": "codex", "model_name": "gpt-5.6-sol", "allowed_efforts": ["medium"], "default_effort": "medium", "permissions": "read-only", }, { "role": "implementer", "provider": "openai", "agent_cli": "codex", "model_name": "gpt-5.6-luna", "allowed_efforts": ["medium"], "default_effort": "medium", "permissions": "workspace-write", }, { "role": "verifier", "provider": "openai", "agent_cli": "codex", "model_name": "gpt-5.6-luna", "allowed_efforts": ["medium"], "default_effort": "medium", "permissions": "workspace-write", }, { "role": "reviewer", "provider": "openai", "agent_cli": "codex", "model_name": "gpt-5.6-sol", "allowed_efforts": ["medium"], "default_effort": "medium", "permissions": "read-only", }, ], } LIVE_TITLE = "l-review-gpt-5-6-sol-high" LIVE_HANDLE = "term_d4612149-6e6f-4c3b-9f55-c1fbb946bb36" LIVE_CREATE_RESPONSE = { "id": "2872f644-e67e-4c1e-ac9d-f23594bb065a", "ok": True, "result": {"terminal": { "handle": LIVE_HANDLE, "tabId": "5f9d1aae-1cb8-4fa9-86f2-0bd3bbd0c061", "worktreeId": "2df7473c-9490-4474-a342-69e12147eeb9::/Users/hiephv/orca/workspaces/aoc-review-remediation", "title": LIVE_TITLE, "executionHostId": "local", "incarnationId": "14df247c-57ce-48d7-bcf0-6fc847c57308", "hostPlatform": "darwin", "surface": "visible", }}, "_meta": {"runtimeId": "55f29b25-35e7-485a-8dbb-706b432f99c9"}, } LIVE_RENAME_RESPONSE = { "id": "1fe9434d-f1ac-4e25-b90d-3a9488824c58", "ok": True, "result": {"rename": { "handle": LIVE_HANDLE, "tabId": "5f9d1aae-1cb8-4fa9-86f2-0bd3bbd0c061", "title": LIVE_TITLE, }}, "_meta": {"runtimeId": "55f29b25-35e7-485a-8dbb-706b432f99c9"}, } def main() -> None: cfg = config() with tempfile.TemporaryDirectory(prefix="aow-ledger-config-required-") as raw: missing_config_root = Path(raw) try: run_ledger.init(type("Args", (), { "root": str(missing_config_root), "run_id": "run-missing-config", "task": "must require config", "mode": "root", "route": "light", "route_explicit": False, "context_capsule": "", })()) except FileNotFoundError as exc: assert ".orca/agents.yaml" in str(exc) else: raise AssertionError("run initialization must reject a missing .orca/agents.yaml") assert orca_adapter._terminal_create_confirms(LIVE_CREATE_RESPONSE, LIVE_TITLE) == LIVE_HANDLE assert orca_adapter._rename_response_confirms(LIVE_RENAME_RESPONSE, LIVE_HANDLE, LIVE_TITLE) is True assert orca_adapter._terminal_create_confirms( {**LIVE_CREATE_RESPONSE, "result": {"terminal": {"handle": LIVE_HANDLE, "title": "wrong"}}}, LIVE_TITLE ) is None assert orca_adapter._terminal_create_confirms( {**LIVE_CREATE_RESPONSE, "terminal": "term-conflict"}, LIVE_TITLE ) is None assert orca_adapter._terminal_create_confirms( {**LIVE_CREATE_RESPONSE, "terminal": {"handle": LIVE_HANDLE, "title": "other-title"}}, LIVE_TITLE ) is None assert orca_adapter._rename_response_confirms( {**LIVE_RENAME_RESPONSE, "result": {"rename": {"handle": "term-conflict", "title": LIVE_TITLE}}}, LIVE_HANDLE, LIVE_TITLE, ) is False assert orca_adapter._rename_response_confirms( {**LIVE_RENAME_RESPONSE, "result": {"rename": {"handle": LIVE_HANDLE, "title": "wrong"}}}, LIVE_HANDLE, LIVE_TITLE, ) is False assert orca_adapter._rename_response_confirms( {**LIVE_RENAME_RESPONSE, "terminal": {"handle": LIVE_HANDLE, "title": "other-title"}}, LIVE_HANDLE, LIVE_TITLE, ) is False assert orca_adapter._rename_response_confirms( {**LIVE_RENAME_RESPONSE, "result": { **LIVE_RENAME_RESPONSE["result"], "terminal": {"handle": LIVE_HANDLE, "title": "other-title"}, }}, LIVE_HANDLE, LIVE_TITLE, ) is False assert orca_adapter._rename_response_confirms( {"ok": True, "title": "other-title", "terminal": LIVE_HANDLE, "result": {"renamed": True}}, LIVE_HANDLE, LIVE_TITLE, ) is False assert agent_cli_command(cfg["profiles"][1], "medium") == [ "codex", "--model", "gpt-5.6-luna", "--dangerously-bypass-approvals-and-sandbox" ] assert profile_config({**cfg, "profiles": [ {**cfg["profiles"][1], "provider": "openai", "agent_cli": "codex"}, {**cfg["profiles"][1], "provider": "gemini", "agent_cli": "agy", "model_name": "gemini-3.8-flash", "default_effort": "high", "allowed_efforts": ["high"]}, ]}, "implementer", 1)["agent_cli"] == "agy" assert terminal_name(cfg, "s", "planner", "gpt-5.6-sol", "medium") == "s-plan-gpt-5-6-sol-medium" assert terminal_name(cfg, "s", "implementer", "gpt-5.6-luna", "medium") == "s-impl-gpt-5-6-luna-medium" assert terminal_name(cfg, "s", "verifier", "gpt-5.6-luna", "medium") == "s-verify-gpt-5-6-luna-medium" assert terminal_name(cfg, "s", "reviewer", "gpt-5.6-sol", "medium") == "s-review-gpt-5-6-sol-medium" assert terminal_name( cfg, "s", "implementer", "gpt-5.6-luna", "medium", existing_names={"s-impl-gpt-5-6-luna-medium"}, ) == "s-impl-gpt-5-6-luna-medium-2" assert terminal_name( cfg, "s", "implementer", "gpt-5.6-luna", "medium", existing_names={"s-impl-gpt-5-6-luna-medium", "s-impl-gpt-5-6-luna-medium-2"}, ) == "s-impl-gpt-5-6-luna-medium-3" command = build_worker_command( cfg, route="hard", stage="implementer", effort="medium", task_id="task-1", run_id="run-1", ordinal=2, worktree="new-child", ) assert command[:3] == ["orca", "orchestration", "worker-start"] assert "--terminal" not in command agy = dict(cfg["profiles"][1], agent_cli="agy", model_name="custom-model", default_effort="high", allowed_efforts=["high"]) assert agent_cli_command(agy) == ["agy", "--model", "custom-model", "--effort", "high", "--dangerously-skip-permissions"] other = dict(agy, agent_cli="other-cli") assert agent_cli_command(other) == ["other-cli", "--model", "custom-model"] with tempfile.TemporaryDirectory(prefix="aow-lifecycle-") as raw: root = Path(raw) path = launch_manifest_path(root, "run-1", "task-1", "attempt-1") path.parent.mkdir(parents=True) path.write_text(json.dumps({ "run_id": "run-1", "task_id": "task-1", "attempt_id": "attempt-1", "status": "ready", "response": {"ok": False, "error": {"code": "selector_not_found"}}, })) settled = resource_action(root, "run-1", "task-1", "attempt-1", "stop") assert settled["resource_status"] == "never_started" assert settled["resource_action"] == "stop" with tempfile.TemporaryDirectory(prefix="aow-rename-gate-") as raw: root = Path(raw) write_default(root) calls: list[list[str]] = [] def fake_run(command: list[str], **kwargs: object) -> object: calls.append(command) if command[1:3] == ["exec", "--model"]: return __import__("subprocess").CompletedProcess( command, 0, stdout="AOW_MODEL_PROBE_OK_7f3c", stderr="", ) if command[1:3] == ["terminal", "create"]: return __import__("subprocess").CompletedProcess( command, 0, stdout=json.dumps({**LIVE_CREATE_RESPONSE, "result": {"terminal": { **LIVE_CREATE_RESPONSE["result"]["terminal"], "handle": "term-1", "title": "s-impl-gpt-5-6-luna-medium", }}}), stderr="", ) if command[1:3] == ["terminal", "rename"]: return __import__("subprocess").CompletedProcess( command, 0, stdout=json.dumps({"ok": True, "result": {"renamed": True}, "terminal": "term-1"}), stderr="", ) if command[1:3] == ["orchestration", "worker-start"]: return __import__("subprocess").CompletedProcess( command, 0, stdout=json.dumps({"ok": True, "dispatch_id": "dispatch-1", "terminal": "term-1", "observedModel": "gpt-5.6-luna", "observedEffort": "medium"}), stderr="", ) if command[1:3] == ["orchestration", "worker-stop"]: return __import__("subprocess").CompletedProcess( command, 0, stdout=json.dumps({"status": "stopped"}), stderr="", ) raise AssertionError(f"unexpected command: {command}") with patch.dict(orca_adapter.os.environ, {"AOW_PROJECT_STORE": str(root / ".store")}): with patch.object(orca_adapter.subprocess, "run", side_effect=fake_run): launched = orca_adapter.launch( root, route="hard", stage="implementer", effort="medium", task_id="task-rename", run_id="run-rename", ordinal=1, worktree="current", owned_paths=["owned/file.py"], ) assert launched["status"] == "ready" assert launched["terminal_rename_confirmed"] is True assert launched["dispatch_id"] == "dispatch-1" assert launched["terminal"] == "term-1" assert launched["profile_index"] == 1 assert launched["profile_snapshot"]["agent_cli"] == "codex" assert len(launched["config_hash"]) == 64 terminal_position = launched["worker_attach_command"].index("--terminal") assert launched["worker_attach_command"][terminal_position + 1] == "term-1" assert calls[0][1:3] == ["exec", "--model"] assert [command[1:3] for command in calls[1:4]] == [["terminal", "create"], ["terminal", "rename"], ["orchestration", "worker-start"]] manifest = json.loads( launch_manifest_path(root, "run-rename", "task-rename", "attempt1").read_text() ) assert manifest["status"] == "ready" assert manifest["dispatch_id"] == "dispatch-1" assert manifest["resource_status"] == "owned" settled = resource_action(root, "run-rename", "task-rename", "attempt1", "stop") assert settled["resource_status"] == "stopped" assert any(command[1:3] == ["terminal", "rename"] for command in calls) with tempfile.TemporaryDirectory(prefix="aow-model-rejection-") as raw: root = Path(raw) write_default(root) calls: list[list[str]] = [] def reject_probe(command: list[str], **kwargs: object) -> object: calls.append(command) if command[1:3] == ["exec", "--model"]: return __import__("subprocess").CompletedProcess(command, 1, stdout="model not found", stderr="") raise AssertionError(f"terminal creation must not run after probe rejection: {command}") with patch.object(orca_adapter.subprocess, "run", side_effect=reject_probe): try: orca_adapter.launch( root, route="hard", stage="implementer", effort="medium", task_id="task-rejected", run_id="run-rejected", ordinal=1, ) except RuntimeError as exc: assert "not accepted" in str(exc) else: raise AssertionError("launch must reject an unaccepted model") assert len(calls) == 1 and calls[0][1:3] == ["exec", "--model"] print("terminal and resource lifecycle checks passed") if __name__ == "__main__": main()