diff --git a/.github/scripts/build_publish_payload.py b/.github/scripts/build_publish_payload.py
index cede29aea..03cafd54c 100644
--- a/.github/scripts/build_publish_payload.py
+++ b/.github/scripts/build_publish_payload.py
@@ -2,7 +2,6 @@
 import argparse
 import json
 import pathlib
-import re
 import sys
 from typing import Any
 
@@ -67,12 +66,6 @@ def _string_or_empty(value: Any) -> str:
     return value if isinstance(value, str) else ""
 
 
-def _slug(value: Any, fallback: str) -> str:
-    raw = value if isinstance(value, str) else fallback
-    slug = re.sub(r"[^a-z0-9]+", "-", raw.lower()).strip("-")
-    return slug or fallback
-
-
 def _normalize_registry_function(function: dict[str, Any]) -> dict[str, Any]:
     return {
         "name": function.get("name"),
@@ -83,44 +76,13 @@ def _normalize_registry_function(function: dict[str, Any]) -> dict[str, Any]:
     }
 
 
-def _derive_trigger_name(trigger: dict[str, Any]) -> str:
-    metadata = _metadata_or_empty(trigger.get("metadata"))
-    for key in ("registry_name", "name"):
-        value = metadata.get(key)
-        if isinstance(value, str) and value.strip():
-            return _slug(value, "trigger")
-
-    config = trigger.get("config") if isinstance(trigger.get("config"), dict) else {}
-    api_path = config.get("api_path")
-    if isinstance(api_path, str) and api_path.strip():
-        return _slug(api_path, "trigger")
-
-    function_id = trigger.get("function_id")
-    if isinstance(function_id, str) and function_id.strip():
-        return _slug(function_id.rsplit("::", 1)[-1], "trigger")
-
-    return _slug(trigger.get("trigger_type") or trigger.get("name"), "trigger")
-
-
 def _normalize_registry_trigger(trigger: dict[str, Any]) -> dict[str, Any]:
-    config = trigger.get("config") if isinstance(trigger.get("config"), dict) else {}
-    metadata = _metadata_or_empty(trigger.get("metadata")).copy()
-    for source_key, metadata_key in (
-        ("id", "engine_id"),
-        ("trigger_type", "trigger_type"),
-        ("function_id", "function_id"),
-    ):
-        if trigger.get(source_key) is not None:
-            metadata.setdefault(metadata_key, trigger.get(source_key))
-    if config:
-        metadata.setdefault("config", config)
-
     return {
-        "name": _derive_trigger_name(trigger),
+        "name": trigger.get("name"),
         "description": _string_or_empty(trigger.get("description")),
         "invocation_schema": _schema_or_empty(trigger.get("invocation_schema")),
         "return_schema": _schema_or_empty(trigger.get("return_schema")),
-        "metadata": metadata,
+        "metadata": _metadata_or_empty(trigger.get("metadata")),
     }
 
 
@@ -149,12 +111,22 @@ def _match_worker(workers: list[dict[str, Any]], worker_name: str) -> dict[str,
     )
 
 
+def _normalize_registry_trigger_type(trigger_type: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "name": _string_or_empty(trigger_type.get("id")),
+        "description": _string_or_empty(trigger_type.get("description")),
+        "invocation_schema": _schema_or_empty(trigger_type.get("trigger_request_format")),
+        "return_schema": _schema_or_empty(trigger_type.get("call_request_format")),
+        "metadata": {},
+    }
+
+
 def normalize_worker_interface(
     *,
     worker_name: str,
     workers_json: dict[str, Any],
     functions_json: dict[str, Any],
-    triggers_json: dict[str, Any] | None = None,
+    trigger_types_json: dict[str, Any] | None = None,
 ) -> dict[str, list[dict[str, Any]]]:
     workers = _extract_array(workers_json, "workers")
     worker = _match_worker(workers, worker_name)
@@ -190,13 +162,13 @@ def normalize_worker_interface(
             }
         )
 
-    worker_ids = set(worker_function_ids)
     triggers = []
-    if triggers_json:
-        for trigger in _extract_array(triggers_json, "triggers"):
-            if trigger.get("function_id") not in worker_ids:
+    if trigger_types_json:
+        for trigger_type in _extract_array(trigger_types_json, "trigger_types"):
+            tt_id = trigger_type.get("id")
+            if not isinstance(tt_id, str) or tt_id.startswith("engine::"):
                 continue
-            triggers.append(_normalize_registry_trigger(trigger))
+            triggers.append(_normalize_registry_trigger_type(trigger_type))
 
     return {"functions": functions, "triggers": triggers}
 
diff --git a/.github/scripts/collect_worker_interface.py b/.github/scripts/collect_worker_interface.py
index 50f0c2bbe..e2f2e8770 100644
--- a/.github/scripts/collect_worker_interface.py
+++ b/.github/scripts/collect_worker_interface.py
@@ -50,16 +50,16 @@ def wait_for_worker(worker_name: str, wait_seconds: int) -> dict[str, object]:
     return workers_json
 
 
-def collect_triggers() -> dict[str, object] | None:
+def collect_trigger_types() -> dict[str, object] | None:
     try:
-        return run_iii("engine::triggers::list", {"include_internal": True})
+        return run_iii("engine::trigger-types::list", {"include_internal": False})
     except (
         subprocess.CalledProcessError,
         subprocess.TimeoutExpired,
         json.JSONDecodeError,
     ) as exc:
         print(
-            f"::warning::could not collect triggers; publishing triggers=[]: {exc}",
+            f"::warning::could not collect trigger types; publishing triggers=[]: {exc}",
             file=sys.stderr,
         )
         return None
@@ -74,13 +74,13 @@ def main() -> int:
 
     workers_json = wait_for_worker(args.worker, args.wait_seconds)
     functions_json = run_iii("engine::functions::list", {"include_internal": True})
-    triggers_json = collect_triggers()
+    trigger_types_json = collect_trigger_types()
 
     interface = normalize_worker_interface(
         worker_name=args.worker,
         workers_json=workers_json,
         functions_json=functions_json,
-        triggers_json=triggers_json,
+        trigger_types_json=trigger_types_json,
     )
     pathlib.Path(args.out).write_text(json.dumps(interface, indent=2) + "\n", encoding="utf-8")
     print(json.dumps(interface, indent=2))