diff --git a/.github/scripts/build_publish_payload.py b/.github/scripts/build_publish_payload.py index 9d24900c7..eb922afd2 100644 --- a/.github/scripts/build_publish_payload.py +++ b/.github/scripts/build_publish_payload.py @@ -86,6 +86,86 @@ def _normalize_registry_trigger(trigger: dict[str, Any]) -> dict[str, Any]: } +def _worker_identity(worker: dict[str, Any]) -> str: + name = worker.get("name") + if isinstance(name, str) and name.strip(): + return name.strip() + worker_id = worker.get("id") + if isinstance(worker_id, str) and worker_id.strip(): + return worker_id.strip() + return "" + + +def _baseline_worker_identities(baseline_workers_json: dict[str, Any] | None) -> set[str]: + if not baseline_workers_json: + return set() + return { + identity + for worker in _extract_array(baseline_workers_json, "workers") + if (identity := _worker_identity(worker)) + } + + +def _resolve_target_worker_names( + *, + workers: list[dict[str, Any]], + worker_name: str, + functions: list[dict[str, Any]], + baseline_workers_json: dict[str, Any] | None, +) -> set[str]: + """Return engine worker names whose bus functions belong in the publish payload.""" + baseline = _baseline_worker_identities(baseline_workers_json) + if baseline: + new_names = { + identity + for worker in workers + if (identity := _worker_identity(worker)) and identity not in baseline + } + if new_names: + return new_names + + worker = _match_worker(workers, worker_name) + matched = _worker_identity(worker) + + legacy_ids = worker.get("functions") or [] + if isinstance(legacy_ids, list) and legacy_ids: + if matched: + return {matched} + raise ValueError(f"matched worker for {worker_name!r} has no identity") + + names_with_functions = { + name.strip() + for fn in functions + if isinstance((name := fn.get("worker_name")), str) and name.strip() + } + if matched and matched in names_with_functions: + return {matched} + + raise ValueError( + f"no functions found for worker {worker_name!r} " + f"(matched identity={matched!r}, workers_with_functions={sorted(names_with_functions)!r})" + ) + + +def _function_ids_for_workers( + functions: list[dict[str, Any]], worker_names: set[str] +) -> list[str]: + seen: set[str] = set() + ordered: list[str] = [] + for fn in functions: + function_id = fn.get("function_id") + worker_name = fn.get("worker_name") + if not isinstance(function_id, str) or not function_id: + continue + if not isinstance(worker_name, str) or worker_name not in worker_names: + continue + if function_id in seen: + continue + seen.add(function_id) + ordered.append(function_id) + return ordered + + def _match_worker(workers: list[dict[str, Any]], worker_name: str) -> dict[str, Any]: by_name = [w for w in workers if w.get("name") == worker_name or w.get("id") == worker_name] if len(by_name) == 1: @@ -135,18 +215,27 @@ def normalize_worker_interface( functions_json: dict[str, Any], trigger_types_json: dict[str, Any] | None = None, baseline_trigger_types_json: dict[str, Any] | None = None, + baseline_workers_json: dict[str, Any] | None = None, ) -> dict[str, list[dict[str, Any]]]: workers = _extract_array(workers_json, "workers") - worker = _match_worker(workers, worker_name) + all_functions = _extract_array(functions_json, "functions") + + target_worker_names = _resolve_target_worker_names( + workers=workers, + worker_name=worker_name, + functions=all_functions, + baseline_workers_json=baseline_workers_json, + ) - worker_function_ids = worker.get("functions") or [] - if not isinstance(worker_function_ids, list): - raise ValueError("worker `functions` must be an array") + worker_function_ids = _function_ids_for_workers(all_functions, target_worker_names) + if not worker_function_ids: + legacy_worker = _match_worker(workers, worker_name) + legacy_ids = legacy_worker.get("functions") or [] + if isinstance(legacy_ids, list) and legacy_ids: + worker_function_ids = [str(fid) for fid in legacy_ids if fid] functions_by_id = { - f.get("function_id"): f - for f in _extract_array(functions_json, "functions") - if f.get("function_id") + f.get("function_id"): f for f in all_functions if f.get("function_id") } missing_function_ids = [fid for fid in worker_function_ids if fid not in functions_by_id] diff --git a/.github/scripts/collect_worker_interface.py b/.github/scripts/collect_worker_interface.py index 2e32ea022..108ccba06 100644 --- a/.github/scripts/collect_worker_interface.py +++ b/.github/scripts/collect_worker_interface.py @@ -6,20 +6,13 @@ import sys import time -from build_publish_payload import normalize_worker_interface +from build_publish_payload import ( + _function_ids_for_workers, + _resolve_target_worker_names, + normalize_worker_interface, +) -def count_worker_matches(workers_json: dict[str, object], worker_name: str) -> int: - workers = workers_json.get("workers", []) - if not isinstance(workers, list): - return 0 - return sum( - 1 - for worker in workers - if isinstance(worker, dict) - and (worker.get("name") == worker_name or worker.get("id") == worker_name) - ) - def run_iii(function_path: str, payload: dict[str, object]) -> dict[str, object]: completed = subprocess.run( @@ -38,15 +31,38 @@ def run_iii(function_path: str, payload: dict[str, object]) -> dict[str, object] return json.loads(completed.stdout) -def wait_for_worker(worker_name: str, wait_seconds: int) -> dict[str, object]: +def wait_for_worker( + worker_name: str, + wait_seconds: int, + *, + workers_baseline_json: dict[str, object] | None = None, +) -> tuple[dict[str, object], dict[str, object]]: deadline = time.monotonic() + wait_seconds workers_json = run_iii("engine::workers::list", {}) + functions_json = run_iii("engine::functions::list", {"include_internal": True}) + + def ready() -> bool: + workers = workers_json.get("workers", []) + functions = functions_json.get("functions", []) + if not isinstance(workers, list) or not isinstance(functions, list): + return False + try: + target_names = _resolve_target_worker_names( + workers=workers, + worker_name=worker_name, + functions=functions, + baseline_workers_json=workers_baseline_json, + ) + except ValueError: + return False + return len(_function_ids_for_workers(functions, target_names)) > 0 - while count_worker_matches(workers_json, worker_name) != 1 and time.monotonic() < deadline: + while not ready() and time.monotonic() < deadline: time.sleep(2) workers_json = run_iii("engine::workers::list", {}) + functions_json = run_iii("engine::functions::list", {"include_internal": True}) - return workers_json + return workers_json, functions_json def collect_trigger_types() -> dict[str, object] | None: @@ -70,6 +86,7 @@ def main() -> int: parser.add_argument("--out", default="worker-interface.json") parser.add_argument("--wait-seconds", type=int, default=0) parser.add_argument("--trigger-types-baseline", default="") + parser.add_argument("--workers-baseline", default="") parser.add_argument( "--assert-non-empty", action="store_true", @@ -109,8 +126,17 @@ def main() -> int: if baseline_path.exists(): baseline_json = json.loads(baseline_path.read_text(encoding="utf-8")) - workers_json = wait_for_worker(args.worker, args.wait_seconds) - functions_json = run_iii("engine::functions::list", {"include_internal": True}) + workers_baseline_json = None + if args.workers_baseline: + workers_baseline_path = pathlib.Path(args.workers_baseline) + if workers_baseline_path.exists(): + workers_baseline_json = json.loads(workers_baseline_path.read_text(encoding="utf-8")) + + workers_json, functions_json = wait_for_worker( + args.worker, + args.wait_seconds, + workers_baseline_json=workers_baseline_json, + ) trigger_types_json = collect_trigger_types() interface = normalize_worker_interface( @@ -119,6 +145,7 @@ def main() -> int: functions_json=functions_json, trigger_types_json=trigger_types_json, baseline_trigger_types_json=baseline_json, + baseline_workers_json=workers_baseline_json, ) pathlib.Path(args.out).write_text(json.dumps(interface, indent=2) + "\n", encoding="utf-8") print(json.dumps(interface, indent=2)) diff --git a/.github/scripts/tests/test_normalize_worker_interface.py b/.github/scripts/tests/test_normalize_worker_interface.py new file mode 100644 index 000000000..cdb226212 --- /dev/null +++ b/.github/scripts/tests/test_normalize_worker_interface.py @@ -0,0 +1,84 @@ +"""Tests for normalize_worker_interface (engine 0.17 worker_name shape).""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from build_publish_payload import normalize_worker_interface # noqa: E402 + + +def test_collects_functions_by_worker_name_with_workers_baseline() -> None: + baseline_workers = { + "workers": [ + {"id": "configuration", "name": "configuration", "runtime": "engine"}, + ] + } + workers_json = { + "workers": [ + *baseline_workers["workers"], + {"id": "w1", "name": "harness", "runtime": "node"}, + {"id": "w2", "name": "turn-orchestrator", "runtime": "node"}, + ] + } + functions_json = { + "functions": [ + { + "function_id": "harness::trigger", + "worker_name": "harness", + "description": "kickoff", + }, + { + "function_id": "run::start", + "worker_name": "turn-orchestrator", + "description": "start run", + }, + { + "function_id": "configuration::get", + "worker_name": "configuration", + "description": "engine built-in", + }, + ] + } + + interface = normalize_worker_interface( + worker_name="harness", + workers_json=workers_json, + functions_json=functions_json, + baseline_workers_json=baseline_workers, + ) + + names = {fn["name"] for fn in interface["functions"]} + assert names == {"harness::trigger", "run::start"} + + +def test_collects_single_worker_without_baseline() -> None: + workers_json = { + "workers": [{"id": "shell", "name": "shell", "runtime": "rust"}], + } + functions_json = { + "functions": [ + { + "function_id": "shell::exec", + "worker_name": "shell", + "description": "run command", + }, + ] + } + + interface = normalize_worker_interface( + worker_name="shell", + workers_json=workers_json, + functions_json=functions_json, + ) + + assert interface["functions"] == [ + { + "name": "shell::exec", + "description": "run command", + "request_schema": {}, + "response_schema": {}, + "metadata": {}, + } + ] diff --git a/.github/workflows/_publish-registry.yml b/.github/workflows/_publish-registry.yml index a20933b22..669f57f5e 100644 --- a/.github/workflows/_publish-registry.yml +++ b/.github/workflows/_publish-registry.yml @@ -115,6 +115,12 @@ jobs: > trigger-types-baseline.json cat trigger-types-baseline.json + - name: Snapshot engine workers baseline + run: | + set -euo pipefail + iii trigger 'engine::workers::list' --json '{}' > workers-baseline.json + cat workers-baseline.json + - name: Start local worker for interface collection env: WORKER: ${{ inputs.worker }} @@ -225,6 +231,7 @@ jobs: "--out" "worker-interface.json" "--wait-seconds" "120" "--trigger-types-baseline" "trigger-types-baseline.json" + "--workers-baseline" "workers-baseline.json" ) python3 .github/scripts/collect_worker_interface.py "${args[@]}"