fix(ci): collect worker functions via worker_name on iii 0.17

.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]

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

.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": {},
+        }
+    ]

.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[@]}"