fix(runtime): dynamic MAS workflow delegation in run-mas (#3)

MAS workflow.delegates_to becomes delegate_to_* tools on the entry agent.
Adds manifest merge at run-mas bootstrap, tool ref resolution, engine
tool dispatch, overlay params staging, docs/schemas, and tests.

Author: jordanauge

ctl/src/mas/ctl/compose/backends/mas_runtime_py.py

@@ -41,6 +41,12 @@ def create_agent_runtime(self, bind: EffectiveBindManifest, agent_id: str) -> Ru
             if mp.is_file():
                 agent_manifest = yaml.safe_load(mp.read_text(encoding="utf-8"))
                 manifest_dir = mp.parent
+                if agent_manifest:
+                    from mas.ctl.manifest.spec_bindings import parse_collaboration
+                    from mas.runtime.engine.tools import resolve_manifest_tool_refs
+
+                    parse_collaboration((agent_manifest.get("spec") or {}).get("collaboration"))
+                    agent_manifest = resolve_manifest_tool_refs(agent_manifest, manifest_dir)
 
         instance, _ = instantiate_runtime(
             InstantiationOptions(

ctl/src/mas/ctl/executor/run_mas.py

@@ -59,6 +59,16 @@ def execute_run_mas(
         validate=validate,
     )
     result = compose_run(req)
+    from mas.ctl.session.params_sidecar import (
+        apply_runtime_params_to_instance,
+        params_from_mas_config,
+        stage_runtime_params,
+    )
+
+    runtime_params = params_from_mas_config(result.mas_config)
+    if runtime_params:
+        stage_runtime_params(runtime_params)
+
     composed = compose_application(result.mas_config, mas_id=result.mas_id)
     plan = compose_placement_from_deployment(result.deployment, composed)
     bind = compose_effective_bind(
@@ -106,8 +116,31 @@ def execute_run_mas(
         instance.driver.agent_id = str(entry or "agent")
 
     agent_manifest = _load_agent_manifest(bind, str(entry or ""))
+    agent_manifest_path = _agent_manifest_path(bind, str(entry or ""))
+    entry_manifest_dir = agent_manifest_path.parent if agent_manifest_path else base
+    from mas.ctl.manifest.mas_agent_merge import enrich_entry_agent_for_delegation, wire_entry_engine_delegation
+
+    enriched_manifest = enrich_entry_agent_for_delegation(
+        agent_manifest or {},
+        result.mas_config,
+        manifest_dir=entry_manifest_dir,
+    )
+    wire_entry_engine_delegation(
+        getattr(getattr(instance, "driver", None), "engine", None),
+        enriched_manifest,
+        entry_manifest_dir,
+        run_turn=_make_workflow_send(
+            materialized,
+            display=display,
+            verbose=verbose,
+            from_agent=str(entry or ""),
+        ),
+        entry_agent_id=str(entry or ""),
+    )
+    if runtime_params:
+        apply_runtime_params_to_instance(runtime_params, instance)
     hitl_responder, _ = resolve_hitl_from_manifest(
-        agent_manifest,
+        enriched_manifest,
         session_interactive=interactive or not auto_hitl,
     )
     if hitl_responder is not None:
@@ -299,8 +332,15 @@ def _entry_agent(mas_config: dict) -> str | None:
 
 
 def _load_agent_manifest(bind: Any, agent_id: str) -> dict | None:
+    mp = _agent_manifest_path(bind, agent_id)
+    if mp is None or not mp.is_file():
+        return None
     import yaml
 
+    return yaml.safe_load(mp.read_text(encoding="utf-8"))
+
+
+def _agent_manifest_path(bind: Any, agent_id: str) -> Path | None:
     for agent in bind.agents:
         if agent.agent_id != agent_id:
             continue
@@ -311,6 +351,6 @@ def _load_agent_manifest(bind: Any, agent_id: str) -> dict | None:
         if not mp.is_absolute():
             base = bind.mas_base_dir or Path.cwd()
             mp = (base / mp).resolve()
-        if mp.is_file():
-            return yaml.safe_load(mp.read_text(encoding="utf-8"))
+        return mp
     return None
+

ctl/src/mas/ctl/manifest/mas_agent_merge.py

@@ -0,0 +1,104 @@
+#  Copyright (c) 2026 Cisco Systems, Inc. and its affiliates
+#  SPDX-License-Identifier: Apache-2.0
+"""Merge MAS workflow onto the entry agent manifest at run-mas bootstrap."""
+
+from __future__ import annotations
+
+import copy
+import logging
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+from mas.ctl.manifest.spec_bindings import parse_collaboration
+from mas.runtime.boundary.delegation.llm_delegator import LlmDelegator
+from mas.runtime.boundary.delegation.policy import delegation_targets
+from mas.runtime.engine.llm_live import LiveLlmEngine
+from mas.runtime.engine.tools import resolve_manifest_tool_refs
+
+logger = logging.getLogger(__name__)
+
+RunTurnFn = Callable[[str, str], str]
+
+
+def _leaf_engine(engine: Any) -> Any:
+    """Unwrap infra pipeline facades to the engine that handles LLM/tool IO."""
+    seen: set[int] = set()
+    current = engine
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        inner = getattr(current, "inner", None)
+        if inner is None:
+            return current
+        current = inner
+    return current
+
+
+def enrich_entry_agent_for_delegation(
+    agent_manifest: dict[str, Any],
+    mas_config: dict[str, Any],
+    *,
+    manifest_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Attach MAS ``workflow`` to the entry agent; resolve tool refs."""
+    spec = agent_manifest.get("spec") or {}
+    parse_collaboration(spec.get("collaboration"))
+    out = copy.deepcopy(agent_manifest)
+    mas_spec = mas_config.get("spec", mas_config) if isinstance(mas_config, dict) else {}
+    wf = mas_spec.get("workflow")
+    if isinstance(wf, dict):
+        spec_out = out.setdefault("spec", {})
+        if spec_out.get("workflow") and spec_out.get("workflow") != wf:
+            logger.warning(
+                "entry agent spec.workflow replaced by MAS workflow (MAS topology wins)"
+            )
+        # wf comes from mas_config (not agent_manifest); copy once to avoid shared refs.
+        spec_out["workflow"] = copy.deepcopy(wf)
+    if manifest_dir is not None:
+        resolve_manifest_tool_refs(out, manifest_dir, inplace=True)
+    return out
+
+
+def wire_entry_engine_delegation(
+    engine: Any,
+    manifest: dict[str, Any],
+    manifest_dir: Path,
+    *,
+    run_turn: RunTurnFn,
+    entry_agent_id: str,
+) -> None:
+    """Set enriched manifest on the entry engine and bind ``LlmDelegator`` when peers exist.
+
+    When peers exist, ``use_tool_loop`` is enabled on the leaf engine so the LLM can
+    emit ``delegate_to_*`` tool calls. A manifest or instantiation that set
+    ``use_tool_loop=False`` is overridden with a warning.
+    """
+    if engine is None:
+        return
+    leaf = _leaf_engine(engine)
+    leaf.manifest = manifest
+    if isinstance(leaf, LiveLlmEngine):
+        leaf.manifest_dir = manifest_dir
+    peers = delegation_targets(manifest, agent_id=entry_agent_id)
+    if not peers:
+        leaf.delegation = None
+        return
+    leaf.delegation = LlmDelegator(run_turn=run_turn)
+    if hasattr(leaf, "use_tool_loop"):
+        if not leaf.use_tool_loop:
+            logger.warning(
+                "entry agent %r: enabling use_tool_loop for dynamic delegation (%d peers)",
+                entry_agent_id,
+                len(peers),
+            )
+            leaf.use_tool_loop = True
+
+
+def reset_engine_delegation(engine: Any) -> None:
+    """Clear delegate caches at the start of each user turn."""
+    while engine is not None:
+        delegation = getattr(engine, "delegation", None)
+        reset_fn = getattr(delegation, "reset_session", None)
+        if callable(reset_fn):
+            reset_fn()
+        engine = getattr(engine, "inner", None)

ctl/src/mas/ctl/manifest/spec_bindings.py

@@ -216,6 +216,16 @@ def parse_collaboration(raw: Any) -> None:
     if not isinstance(raw, dict):
         raise SpecBindingError(f"spec.collaboration must be an object, got {type(raw).__name__}")
     _reject_unknown_keys(raw, allowed=_COLLABORATION_KEYS, field="spec.collaboration")
+    if raw.get("ref"):
+        raise SpecBindingError(
+            "spec.collaboration.ref is not supported in this release; omit spec.collaboration"
+        )
+    typ = raw.get("type")
+    if isinstance(typ, str) and typ.strip() and typ.strip().lower() != "none":
+        raise SpecBindingError(
+            f"spec.collaboration.type {typ.strip()!r} is not supported in this release; "
+            "omit spec.collaboration or set type: none"
+        )
 
 
 def parse_context_manager(raw: Any) -> None:

ctl/src/mas/ctl/session/controller.py

@@ -142,6 +142,9 @@ def run_turn(self, text: str, *, turn_id: str | None = None, auto_hitl: bool = T
         return self._run_user_turn(text, turn_id=turn_id, auto_hitl=auto_hitl)
 
     def _run_user_turn(self, text: str, *, turn_id: str | None = None, auto_hitl: bool = True) -> TurnResult:
+        from mas.ctl.manifest.mas_agent_merge import reset_engine_delegation
+
+        reset_engine_delegation(getattr(self.instance.driver, "engine", None))
         self._turn += 1
         tid = turn_id or f"u{self._turn}"
         self.display.on_user(text, turn_id=tid)

ctl/src/mas/ctl/session/params_sidecar.py

@@ -0,0 +1,60 @@
+#  Copyright (c) 2026 Cisco Systems, Inc. and its affiliates
+#  SPDX-License-Identifier: Apache-2.0
+"""Stage overlay ``spec.params`` for tools without polluting the project tree."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from collections.abc import MutableMapping
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+MAS_RUNTIME_ARTIFACTS_ENV = "MAS_RUNTIME_ARTIFACTS_DIR"
+
+
+def stage_runtime_params(
+    params: dict[str, Any],
+    *,
+    environ: MutableMapping[str, str] | None = None,
+) -> Path | None:
+    """Write params to a process-scoped runtime dir and expose via ``MAS_RUNTIME_ARTIFACTS_DIR``.
+
+    Writes ``artifacts/scene.yaml`` under ``$XDG_RUNTIME_DIR/mas-ctl/run-<pid>`` (or temp).
+  Tool modules that read fixture paths from the filesystem should use that env var until
+    they consume ``ctx.runtime_params`` from the engine tool loop.
+    """
+    if not params:
+        return None
+
+    env = os.environ if environ is None else environ
+    base = env.get("XDG_RUNTIME_DIR") or tempfile.gettempdir()
+    root = Path(base) / "mas-ctl" / f"run-{os.getpid()}"
+    sidecar_dir = root / "artifacts"
+    sidecar_dir.mkdir(parents=True, exist_ok=True)
+    sidecar_path = sidecar_dir / "scene.yaml"
+    sidecar_path.write_text(
+        yaml.safe_dump(params, default_flow_style=False, allow_unicode=True),
+        encoding="utf-8",
+    )
+    env[MAS_RUNTIME_ARTIFACTS_ENV] = str(root)
+    return sidecar_path
+
+
+def apply_runtime_params_to_instance(params: dict[str, Any], instance: Any) -> None:
+    """Thread overlay params through the runtime ctx for in-process tool modules."""
+    if not params:
+        return
+    ctx = getattr(getattr(instance, "driver", None), "ctx", None)
+    if ctx is None:
+        return
+    # TODO: tool execution should read ctx.runtime_params; env sidecar is interim.
+    ctx.runtime_params = dict(params)
+
+
+def params_from_mas_config(mas_config: dict[str, Any]) -> dict[str, Any]:
+    spec = mas_config.get("spec", mas_config) if isinstance(mas_config, dict) else {}
+    raw = spec.get("params") or {}
+    return dict(raw) if isinstance(raw, dict) else {}

ctl/tests/test_collaboration_bindings.py

@@ -0,0 +1,20 @@
+#  Copyright (c) 2026 Cisco Systems, Inc. and its affiliates
+#  SPDX-License-Identifier: Apache-2.0
+
+import pytest
+
+from mas.ctl.manifest.spec_bindings import SpecBindingError, parse_collaboration
+
+
+def test_parse_collaboration_rejects_non_none_type():
+    with pytest.raises(SpecBindingError, match="llm-delegator"):
+        parse_collaboration({"type": "llm-delegator"})
+
+
+def test_parse_collaboration_rejects_ref():
+    with pytest.raises(SpecBindingError, match="ref"):
+        parse_collaboration({"ref": "module://example.Plugin"})
+
+
+def test_parse_collaboration_allows_none():
+    parse_collaboration({"type": "none"})