feat: support NAT registry workflows

external/nat/README.md

@@ -15,16 +15,18 @@ northbound `FabricConfig`, or depend on Pydantic for its contract boundary.
 
 ## Configuration Boundary
 
-NeMo Fabric owns portable configuration. `workflow` selects a Fabric-enumerated
-agent factory and `tools.definitions` supplies named functions and function
-groups that the adapter resolves as installed NAT components.
+NeMo Fabric owns portable configuration. `workflow` selects the existing
+portable ReAct alias or an installed NAT registry factory.
+`tools.definitions` supplies named functions and function groups that the
+adapter resolves as installed NAT components.
 
 | NeMo Fabric input | NAT configuration |
 | --- | --- |
 | `models.<role>` | `llms.<role>`; every NeMo Fabric model-role name is preserved |
-| `instructions.system` | Built-in `react_agent` workflow `additional_instructions`; other workflow types reject this field in the initial adapter |
-| `workflow.entrypoint.kind=factory` | Resolve a Fabric-enumerated agent intent |
+| `instructions.system` | Built-in shared and per-user ReAct workflow `additional_instructions`; other workflow types reject this field in the initial adapter |
+| `workflow.entrypoint.kind=factory` | Resolve a NAT registry factory |
 | `workflow.entrypoint.ref=fabric.agent.react` | NAT `react_agent` workflow factory |
+| Any other `workflow.entrypoint.ref` | Forward the short or fully qualified NAT registry type unchanged |
 | `workflow.settings` | Remaining `workflow` component fields |
 | `tools.definitions.<name>` with `kind=function` | NAT `functions.<name>`; `ref` becomes `_type` |
 | `tools.definitions.<name>` with `kind=function_group` | NAT `function_groups.<name>`; `ref` becomes `_type` |
@@ -34,15 +36,52 @@ groups that the adapter resolves as installed NAT components.
 The adapter loads installed `nat.components` entry points before NAT validates
 the generated configuration. A custom function or function group is supplied
 as an installed NAT component package and selected by `tools.definitions.ref`.
-No Python callable crosses the configuration contract. A custom adapter may
-publish a broader workflow schema without changing this shared NAT adapter.
+No Python callable crosses the configuration contract. Installed NAT owns the
+accepted registry types and validates their native settings.
+
+NAT validates registry references after loading installed component entry
+points. The adapter does not maintain a workflow catalog and forwards any
+installed workflow registry reference. Execution is limited to workflows whose
+component graph can be expressed through the translated surfaces above:
+workflow settings, LLMs, functions, function groups, and MCP. Workflows that
+require other top-level NAT configuration sections, such as embedders, memory,
+object stores, retrievers, or middleware, are outside this initial reference.
+
+The adapter translates portable system instructions only for the exact shared
+and per-user ReAct configuration shapes whose fields it knows. Normalized tool
+policy and MCP configuration work with any workflow whose native settings
+expose a string-list `tool_names` field. Other workflow-specific configuration
+remains in `workflow.settings`.
 
 At runtime, `start` loads components, enters one `WorkflowBuilder`, creates a
 `SessionManager` with that shared builder, and retains both resources. Each
 `invoke` opens a session from the retained manager, enters `session.run(...)`,
-and awaits `runner.result()`. `stop` shuts down the session manager and exits
-the builder context. This first reference does not claim cancellation, service,
-streaming, or live-update support.
+and awaits `runner.result()`. The adapter reads NAT's session-manager metadata
+to determine whether invocation identity is required, validates and forwards
+that identity to `SessionManager.session(...)`, and leaves builder creation,
+caching, and cleanup to NAT. Repeated requests for one user reuse NAT's cached
+builder, different users remain isolated, and separate NeMo Fabric runtimes
+own separate session managers.
+
+After starting a multi-turn runtime, invoke a per-user workflow with a typed
+request:
+
+```python
+from nemo_fabric import RunRequest
+
+result = await runtime.invoke(
+    request=RunRequest(
+        input="What did I ask previously?",
+        context={"user_id": "user-123"},
+    )
+)
+```
+
+`stop` first shuts down the session manager, including NAT's cleanup task and
+cached per-user builders, and then exits the shared builder context. NAT can
+also evict inactive per-user builders according to its session cleanup policy.
+This reference does not claim cancellation, service, streaming, or live-update
+support.
 
 ## MCP Tool Filters
 

external/nat/fabric-adapter.json

@@ -21,11 +21,13 @@
         "properties": {
           "kind": {
             "const": "factory",
-            "description": "Resolve a Fabric-enumerated agent intent through this adapter."
+            "description": "Resolve a NAT registry factory through this adapter."
           },
           "ref": {
-            "const": "fabric.agent.react",
-            "description": "Build Fabric's ReAct agent intent with the NAT react_agent factory."
+            "type": "string",
+            "minLength": 1,
+            "pattern": "^\\S+$",
+            "description": "Resolve an installed NAT workflow registry type; fabric.agent.react is retained as the portable ReAct alias."
           }
         },
         "required": ["kind", "ref"],

external/nat/src/nemo_fabric_adapters/nat/adapter.py

@@ -6,7 +6,9 @@
 
 The adapter builds one in-memory NAT configuration from Fabric's normalized
 configuration and adapter-owned NAT component settings. One persistent adapter
-host owns the resulting workflow for the complete Fabric runtime lifecycle.
+host owns the shared builder and session manager for the complete Fabric
+runtime lifecycle. NAT owns workflow-specific session behavior beneath that
+manager, including per-user workflow builders.
 """
 
 from __future__ import annotations
@@ -29,10 +31,15 @@
 WORKFLOW_FACTORY_KIND = "factory"
 FABRIC_REACT_AGENT = "fabric.agent.react"
 FUNCTION_GROUP_SEPARATOR = "__"
-REACT_AGENT_REFS = frozenset(
+# This allowlist selects the field codec for NAT's known ReAct configuration
+# models. It never determines shared versus per-user lifecycle; SessionManager
+# owns that decision.
+REACT_CONFIG_TYPES = frozenset(
     {
         "react_agent",
+        "per_user_react_agent",
         "nat.plugins.langchain.agent.react_agent/react_agent",
+        "nat.plugins.langchain.agent.react_agent/per_user_react_agent",
     }
 )
 RESERVED_MODEL_SETTINGS = frozenset(
@@ -98,13 +105,6 @@ def _nat_workflow(agent_config: AgentConfig) -> dict[str, Any]:
             f"workflow.entrypoint.kind must equal {WORKFLOW_FACTORY_KIND!r}",
             field="workflow.entrypoint.kind",
         )
-    if entrypoint.ref != FABRIC_REACT_AGENT:
-        raise _config_error(
-            "nat_invalid_workflow",
-            f"NAT does not support workflow factory {entrypoint.ref!r}",
-            field="workflow.entrypoint.ref",
-        )
-
     settings = workflow_config.settings
     if "_type" in settings:
         raise _config_error(
@@ -114,8 +114,10 @@ def _nat_workflow(agent_config: AgentConfig) -> dict[str, Any]:
         )
 
     workflow = copy.deepcopy(settings)
-    workflow["_type"] = "react_agent"
-    if _is_react_agent(workflow):
+    workflow["_type"] = (
+        "react_agent" if entrypoint.ref == FABRIC_REACT_AGENT else entrypoint.ref
+    )
+    if _uses_react_config_shape(workflow):
         workflow.setdefault("tool_names", [])
     return workflow
 
@@ -214,8 +216,8 @@ def _nat_llms(agent_config: AgentConfig) -> dict[str, dict[str, Any]]:
     return llms
 
 
-def _is_react_agent(workflow: dict[str, Any]) -> bool:
-    return workflow.get("_type") in REACT_AGENT_REFS
+def _uses_react_config_shape(workflow: dict[str, Any]) -> bool:
+    return workflow.get("_type") in REACT_CONFIG_TYPES
 
 
 def _apply_system_instruction(
@@ -226,7 +228,7 @@ def _apply_system_instruction(
         return
 
     workflow = config["workflow"]
-    if not _is_react_agent(workflow):
+    if not _uses_react_config_shape(workflow):
         raise _config_error(
             "nat_system_instruction_unsupported",
             "instructions.system is supported only for a NAT react_agent workflow",
@@ -636,7 +638,11 @@ def build_nat_config(agent_config: AgentConfig) -> Any:
         ) from error
 
 
-def _session_kwargs(request: dict[str, Any]) -> dict[str, str]:
+def _session_kwargs(
+    request: dict[str, Any],
+    *,
+    require_user_id: bool = False,
+) -> dict[str, str]:
     context = request.get("context")
     if context is None:
         context = {}
@@ -648,11 +654,18 @@ def _session_kwargs(request: dict[str, Any]) -> dict[str, str]:
         "conversation_id": context.get("conversation_id"),
         "user_message_id": context.get("user_message_id") or request.get("request_id"),
     }
+    user_id = values["user_id"]
+    if require_user_id and (not isinstance(user_id, str) or not user_id.strip()):
+        raise ValueError(
+            "NAT per-user workflow requires request.context.user_id "
+            "as a non-empty string"
+        )
+
     result: dict[str, str] = {}
     for name, value in values.items():
         if value is None:
             continue
-        if not isinstance(value, str) or not value:
+        if not isinstance(value, str) or not value.strip():
             raise ValueError(
                 f"NAT invocation request context {name} must be a non-empty string"
             )
@@ -767,7 +780,12 @@ async def invoke(self, payload: dict[str, Any]) -> dict[str, Any]:
                 "NAT invocation request must be a mapping",
             )
         try:
-            session_kwargs = _session_kwargs(request)
+            session_kwargs = _session_kwargs(
+                request,
+                require_user_id=bool(
+                    getattr(self._sessions, "is_workflow_per_user", False)
+                ),
+            )
         except ValueError as error:
             return _failure_output("nat_invalid_request", str(error))