feat(hitl): add agent-driven human-in-the-loop via request_human_input (#51)

An agent can call request_human_input() to block synchronously on a
threading.Event until a human resolves its question, or raise TimeoutError
after a configured number of seconds -- there is no silent auto-approve on
timeout. inform_user() sends a non-blocking progress update to whoever is
watching the session.

- By default a request blocks indefinitely. A timeout is set per call (the
  tool's own `timeout` argument) or as a manifest-wide default (a
  `{kind: system, name: request_human_input, params: {timeout: N}}` entry
  in spec.tools); the call's own value always wins.
- Non-interactive runs (lab/benchmark runs, and batch CLI runs) auto-resolve
  instead of blocking, since there is no human present to answer; real
  interactive sessions always block for a genuine response. The auto-resolve
  decision defaults to "approve" and is configurable the same two ways as
  the timeout (spec.tools params, or the MAS_HITL_AUTO_RESOLVE_DECISION env
  var; the manifest value wins if both are set).
- HitlResolverRegistry is the side-channel between a blocked tool call and
  whatever surfaces the request to a human (a Webex card, a CLI prompt,
  etc.) and later resolves it back.
- Every delegated agent's own SessionController receives the same
  trace/trace_timestamps/trace_engine/trace_summary/trace_color
  configuration as the entry agent.
- mas-ctl's CLI trace and this HITL bridge are both ExchangePlugin
  subscribers (subscribe_exchange/unsubscribe_exchange over a list of
  subscribers): tool-call errors are always surfaced in RED and forwarded
  to an external exchange_listener, regardless of --trace/--verbose.
- request_human_input/inform_user are now available to every agent
  including delegates (needed so a delegated sub-agent can ask for human
  input too); mock_llm.py's schema-driven tool-selection fallback (used by
  MockModelAccess for offline/CI runs) is updated to skip these two system
  tools when picking a "nothing else matched" default, so a mocked delegate
  still exercises its own real tool instead of getting its call routed to
  request_human_input and auto-resolved.

Regression tests cover: both exchange plugins staying subscribed exactly
once per session (with the bridge still forwarding events when
--trace/--verbose are off); the timeout/auto-resolve precedence rules
above; and mock_llm.py's system-tool-skipping fallback. Two golden-run
fixtures (extensions, lifecycle-control) are regenerated to reflect
request_human_input/inform_user now being present on every agent's tool
list. Full `task ci` (unit, controller, functional/golden-run/tutorial,
dry-run, chat smoke, HITL gate) passes clean.

Signed-off-by: Jordan Augé <augjorda@cisco.com>

Author: jordanauge

ctl/src/mas/ctl/benchmark/runner.py

@@ -267,9 +267,17 @@ def run(
         flavour: Any = None,
         **kwargs: Any,
     ) -> RunResult:
+        import os
+
         from mas.lab.benchmark.runners.fixtures import write_tool_fixtures_sidecar
         from mas.lab.inputs import RunInput
 
+        # Bench/lab runs are non-interactive by nature — no human is present to
+        # resolve agent-initiated HITL requests, so auto-resolve them instead of
+        # blocking for up to 60s per request (see execute_run_mas for the CLI
+        # equivalent).
+        os.environ.setdefault("MAS_HITL_AUTO_RESOLVE", "1")
+
         ri: RunInput | None = run_input if isinstance(run_input, RunInput) else None
 
         queries = ri.scripted_queries() if ri else ([prompt] if prompt else [])

ctl/src/mas/ctl/executor/mas_session.py

@@ -148,6 +148,11 @@ def prepare_delegation_entry_session(
     display: Any = None,
     verbose: int = 0,
     session_id: str = "",
+    trace: bool = False,
+    trace_timestamps: bool = False,
+    trace_engine: bool = False,
+    trace_summary: bool = False,
+    trace_color: bool = False,
 ) -> PreparedEntrySession:
     """Wire dynamic-delegation entry agent (same path as ``execute_run_mas``).
 
@@ -190,6 +195,11 @@ def prepare_delegation_entry_session(
             verbose=verbose,
             from_agent=entry_id,
             session_id=resolved_session_id,
+            trace=trace,
+            trace_timestamps=trace_timestamps,
+            trace_engine=trace_engine,
+            trace_summary=trace_summary,
+            trace_color=trace_color,
         ),
         entry_agent_id=entry_id,
         mas_config=compose.mas_config,
@@ -212,6 +222,11 @@ def wire_peer_delegation(
     verbose: int = 0,
     already_wired: "set[str] | None" = None,
     session_id: str = "",
+    trace: bool = False,
+    trace_timestamps: bool = False,
+    trace_engine: bool = False,
+    trace_summary: bool = False,
+    trace_color: bool = False,
 ) -> list[str]:
     """Wire delegation onto every agent that declares its own ``delegates_to``
     peers in the MAS workflow topology — not just the entry agent.
@@ -247,6 +262,11 @@ def wire_peer_delegation(
         verbose=verbose,
         from_agent=entry_id,
         session_id=session_id,
+        trace=trace,
+        trace_timestamps=trace_timestamps,
+        trace_engine=trace_engine,
+        trace_summary=trace_summary,
+        trace_color=trace_color,
     )
     newly_wired: list[str] = []
     for agent in compose.bind.agents:
@@ -288,6 +308,11 @@ def make_workflow_send(
     verbose: int,
     from_agent: str = "",
     session_id: str = "",
+    trace: bool = False,
+    trace_timestamps: bool = False,
+    trace_engine: bool = False,
+    trace_summary: bool = False,
+    trace_color: bool = False,
 ) -> RunTurnFn:
     """Run one agent turn inside a multi-agent workflow (sequential or delegation).
 
@@ -393,6 +418,11 @@ def send(
             config=ConversationConfig(single_turn=True),
             session_id=state["session_id"],
             working_memory_key=memory_key,
+            trace=trace,
+            trace_timestamps=trace_timestamps,
+            trace_engine=trace_engine,
+            trace_summary=trace_summary,
+            trace_color=trace_color,
         )
         result = controller.run_turn(prompt, turn_id=turn_id, parent_call_id=parent_call_id)
         # Do NOT close observability after a delegated sub-turn: in a multi-agent
@@ -408,6 +438,18 @@ def send(
         state["prev_agent"] = agent_id
         if turn_failed(result):
             raise RuntimeError(f"agent {agent_id!r} turn failed")
+        
+        # Propagate awaiting_hitl state via side channel (not return value)
+        # This allows external systems (Webex bot) to detect and resolve HITL
+        # from delegated agents without breaking the delegation contract.
+        if result.awaiting_hitl:
+            from mas.runtime.boundary.hitl.registry import get_hitl_resolver_registry
+            registry = get_hitl_resolver_registry()
+            # Check if there are any pending HITL requests for this agent
+            if registry.has_pending(state["session_id"], agent_id):
+                # Mark in state that this agent has pending HITL
+                state.setdefault("pending_hitl_agents", set()).add(agent_id)
+        
         return result.text
 
     return send

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

@@ -60,11 +60,22 @@ def execute_run_mas(
     trace_color: bool = False,
 ) -> int:
     """Compose → materialize → SessionController on entry agent."""
+    import os
+
     from mas.ctl.session.controller import ConversationConfig, SessionController, close_observability
     from mas.ctl.session.hitl_config import resolve_hitl_from_manifest
     from mas.ctl.session.controller import run_session_loop
     from mas.ctl.ui.stdout import StdoutConversationDisplay
 
+    # Batch/CLI runs with auto-hitl (the default) have no external resolver
+    # (Webex bot, operator console) listening for agent-initiated
+    # request_human_input() calls, so the synchronous HITL wait in
+    # manifest_tool_provider would otherwise always time out. Signal batch
+    # mode via env var (mirrors the existing MAS_MANIFEST_RESOLVE_REFS
+    # pattern) so it auto-resolves instead of blocking. Interactive sessions
+    # never set this, so real HITL resolution still blocks as intended.
+    os.environ["MAS_HITL_AUTO_RESOLVE"] = "1" if (auto_hitl and not interactive) else "0"
+
     scripted = list(queries or [])
     if prompt:
         scripted.insert(0, prompt)
@@ -115,6 +126,11 @@ def execute_run_mas(
             entry_id=entry,
             display=display,
             verbose=verbose,
+            trace=trace,
+            trace_timestamps=trace_timestamps,
+            trace_engine=trace_engine,
+            trace_summary=trace_summary,
+            trace_color=trace_color,
         )
     except KeyError as exc:
         logger.error("%s", exc)
@@ -136,6 +152,11 @@ def execute_run_mas(
         verbose=verbose,
         already_wired={entry},
         session_id=prepared.session_id,
+        trace=trace,
+        trace_timestamps=trace_timestamps,
+        trace_engine=trace_engine,
+        trace_summary=trace_summary,
+        trace_color=trace_color,
     )
 
     if runtime_params:

ctl/src/mas/ctl/session/bootstrap.py

@@ -62,6 +62,8 @@ class InstantiationOptions:
     enable_observability: bool = True
     enable_governance: bool = True
     enable_coordination: bool = True
+    hitl_contract: object | None = None
+    user_io_contract: object | None = None
 
 
 def instantiate_runtime(
@@ -209,6 +211,8 @@ def instantiate_runtime(
             options.manifest_dir or Path.cwd(),
             app_root=options.app_root or options.manifest_dir,
             workspace_root=ws.root if ws.found else None,
+            hitl_contract=options.hitl_contract,
+            user_io_contract=options.user_io_contract,
         )
     return instance, store
 

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

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import logging
 import sys
 import time
 import uuid
@@ -15,16 +16,59 @@
     sync_working_memory_in,
     sync_working_memory_out,
 )
-from mas.runtime.driver.driver import DriverTrace
+from mas.runtime.boundary.obs.exchange_plugin import ExchangePlugin
+from mas.runtime.driver.driver import DriverTrace, ExchangeRecord
 from mas.runtime.driver.instance import RuntimeInstance
 from mas.runtime.schema.egress import EmitClientResponse
 
 from mas.ctl.session.exchange_log import (
+    CliTraceExchangePlugin,
     TraceFormatOptions,
-    print_exchange,
 )
 from mas.ctl.ui.display import ConversationDisplay
 
+_logger = logging.getLogger("mas.runtime")
+_RED = "\033[1;31m"
+_RESET = "\033[0m"
+
+
+class _ToolErrorAndListenerBridge(ExchangePlugin):
+    """Second, independent ExchangePlugin subscriber alongside CliTraceExchangePlugin.
+
+    Two behaviors that aren't display formatting, so they don't belong in
+    CliTraceExchangePlugin (a generic, reusable trace renderer):
+      - Always-on tool-call error surfacing: a TOOL->AGENT exchange whose text
+        is a JSON error result is printed in RED to stderr and logged at ERROR
+        level, regardless of --trace/--verbose, so a failing tool call (e.g.
+        run_skill_script returning {"error": ...}) can never silently
+        disappear from view.
+      - Forwards every exchange record to an optional external listener (e.g.
+        a chat-UI plugin) via SessionController.exchange_listener, if set.
+    Subscribing this as its own plugin (rather than folding it into
+    CliTraceExchangePlugin) keeps each subscriber single-purpose, and proves
+    the additive subscribe_exchange() interface actually supports more than
+    one concurrent consumer.
+    """
+
+    def __init__(self) -> None:
+        self.agent_id = "n/a"
+        self.exchange_listener: Any | None = None
+
+    def configure(self, *, agent_id: str, exchange_listener: Any | None) -> None:
+        self.agent_id = agent_id
+        self.exchange_listener = exchange_listener
+
+    def on_exchange(self, record: ExchangeRecord) -> None:
+        if record.tag == "TOOL->AGENT" and '"error"' in record.text:
+            message = f"[{self.agent_id}] TOOL ERROR: {record.text.strip()}"
+            print(f"{_RED}{message}{_RESET}", file=sys.stderr, flush=True)
+            _logger.error(message)
+        if callable(self.exchange_listener):
+            try:
+                self.exchange_listener(record)
+            except Exception:
+                _logger.exception("session exchange listener failed")
+
 
 @dataclass
 class ConversationConfig:
@@ -63,6 +107,7 @@ class SessionController:
     agent_id: str = "n/a"
     llm_id: str = "gpt-4o-mini"
     obs_recorder: Any | None = None
+    exchange_listener: Any | None = None
     # One id for the whole MAS run — every turn this controller ever runs
     # shares it (see _run_user_turn). Empty here means "mint a fresh one";
     # explicitly pass an existing value (e.g. from an entry agent's own
@@ -79,6 +124,8 @@ class SessionController:
     working_memory_key: str = ""
     _turn: int = 0
     _trace_turn_start: float = 0.0
+    _trace_plugin: CliTraceExchangePlugin | None = field(default=None, repr=False)
+    _bridge_plugin: _ToolErrorAndListenerBridge | None = field(default=None, repr=False)
 
     def __post_init__(self) -> None:
         if not self.session_id:
@@ -99,35 +146,50 @@ def _trace_format_options(self) -> TraceFormatOptions:
         )
 
     def _setup_exchange_tracing(self) -> None:
-        """Setup realtime exchange tracing via driver callback (if trace or verbose)."""
-        # Skip if neither trace nor verbose logging requested
+        """Configure the CLI trace + error/listener-bridge ExchangePlugins.
+
+        Subscribes two persistent, single-purpose plugins to the driver's
+        exchange_plugins list the first time this runs, then only updates
+        their config on every subsequent turn — the driver's subscriber
+        list is additive (see KernelDriver.subscribe_exchange), so
+        re-running this every turn no longer risks discarding any other
+        plugin (e.g. an external chat-UI plugin) registered on the same
+        driver:
+          - CliTraceExchangePlugin: normal exchange display (AGENT/LLM/TOOL
+            lines), gated by --trace or --verbose as before.
+          - _ToolErrorAndListenerBridge: tool-call *errors* are always
+            surfaced — unconditionally printed in RED to stderr and logged
+            at ERROR level, regardless of trace/verbose settings, so a
+            failing tool call (e.g. run_skill_script returning
+            {"error": ...}) can never silently disappear from view. Also
+            forwards every exchange record to self.exchange_listener, if set.
+        """
+        if self._trace_plugin is None:
+            self._trace_plugin = CliTraceExchangePlugin()
+            self.instance.driver.subscribe_exchange(self._trace_plugin)
+        if self._bridge_plugin is None:
+            self._bridge_plugin = _ToolErrorAndListenerBridge()
+            self.instance.driver.subscribe_exchange(self._bridge_plugin)
+        self._bridge_plugin.configure(agent_id=self.agent_id, exchange_listener=self.exchange_listener)
+
+        # Skip normal display if neither trace nor verbose logging requested
+        # (the error/listener bridge stays active regardless).
         if not self.trace and self.verbose < 1:
-            self.instance.driver.on_exchange = None
+            self._trace_plugin.configure(
+                trace=False, verbose=0, agent_id=self.agent_id, fmt=self._trace_format_options()
+            )
             self.instance.driver.capture_engine_io = False
             return
 
         self._trace_turn_start = time.perf_counter()
         self.instance.driver.capture_engine_io = self.trace_engine
-        fmt = self._trace_format_options()
-        logger = __import__("logging").getLogger("mas.runtime")
-
-        def on_exchange(ex: object) -> None:
-            from mas.runtime.driver.driver import ExchangeRecord
-            from mas.ctl.session.exchange_log import format_exchange
-
-            if not isinstance(ex, ExchangeRecord):
-                return
-
-            # Realtime stderr output (if --trace) — primary display path
-            if self.trace:
-                print_exchange(ex, err=sys.stderr, agent_id=self.agent_id, fmt=fmt)
-            # Verbose logging only if NOT using --trace (alternative logging path)
-            elif self.verbose >= 1:
-                formatted = format_exchange(self.agent_id, ex, fmt=fmt).strip()
-                for line in formatted.splitlines():
-                    logger.info("[%s] %s", self.agent_id, line)
-
-        self.instance.driver.on_exchange = on_exchange
+        self._trace_plugin.configure(
+            trace=self.trace,
+            verbose=self.verbose,
+            agent_id=self.agent_id,
+            fmt=self._trace_format_options(),
+        )
+
 
     def _handle_list_skills(self) -> TurnResult:
         """Handle /skills command — list all available skills."""