feat(evaluator): persist eval results as queryable entities

packages/nemo_platform_plugin/src/nemo_platform_plugin/sdk_provider.py

@@ -72,6 +72,15 @@ def get_task_sdk(self, service_name: str) -> NeMoPlatform:
         when ``NMP_PRINCIPAL`` is set, acts on behalf of the job creator.
         """
 
+    def get_async_task_sdk(self, service_name: str) -> AsyncNeMoPlatform:
+        """Async counterpart of :meth:`get_task_sdk` for use inside a task container.
+
+        Authenticates as ``service:{service_name}`` and, when ``NMP_PRINCIPAL``
+        is set, acts on behalf of the job creator with the *full* delegated
+        identity (on-behalf-of id, email, and groups) — wire-identical to
+        :meth:`get_task_sdk`, just async.
+        """
+
     def get_platform_sdk(
         self,
         *,
@@ -177,6 +186,29 @@ def get_task_sdk(self, service_name: str) -> NeMoPlatform:
             default_headers=headers,
         )
 
+    def get_async_task_sdk(self, service_name: str) -> AsyncNeMoPlatform:
+        # Async mirror of get_task_sdk: identical headers (service principal,
+        # internal marker, and full on-behalf-of id/email/groups), async client.
+        headers: dict[str, str] = {
+            "X-NMP-Principal-Id": f"service:{service_name}",
+            _INTERNAL_REQUEST_HEADER: "true",
+        }
+
+        principal = _read_principal_from_env()
+        if principal is not None:
+            headers.update(_on_behalf_of_headers(principal))
+        else:
+            logger.warning(
+                "%s not set; async task SDK will authenticate as service:%s without on-behalf-of delegation",
+                _NMP_PRINCIPAL_ENVVAR,
+                service_name,
+            )
+
+        return AsyncNeMoPlatform(
+            base_url=self._base_url(),
+            default_headers=headers,
+        )
+
     def _make_sdk(
         self,
         cls: type[_SDKT],
@@ -311,6 +343,21 @@ def get_task_sdk(service_name: str) -> NeMoPlatform:
     return _resolve_provider().get_task_sdk(service_name)
 
 
+def get_async_task_sdk(service_name: str) -> AsyncNeMoPlatform:
+    """Async counterpart of :func:`get_task_sdk` for use inside a task container.
+
+    For a (synchronous) job ``run`` that needs to drive an async helper — e.g. an entity-store
+    write — without fabricating its own client. Authenticates as ``service:{service_name}`` and, when
+    ``NMP_PRINCIPAL`` is set, on behalf of the job creator with the full delegated identity
+    (on-behalf-of id, email, and groups) — wire-identical to :func:`get_task_sdk`.
+
+    A dedicated provider method (not a wrapper over :func:`get_async_platform_sdk`) so each provider
+    mirrors its own sync :meth:`SDKProvider.get_task_sdk` exactly; the platform provider routes URLs
+    and reuses its shared async client, the default provider uses env-var headers.
+    """
+    return _resolve_provider().get_async_task_sdk(service_name)
+
+
 def get_platform_sdk(
     *,
     as_service: str | None = None,

packages/nemo_platform_plugin/tests/test_sdk_provider.py

@@ -9,7 +9,7 @@
 from unittest.mock import patch
 
 import pytest
-from nemo_platform import NeMoPlatform
+from nemo_platform import AsyncNeMoPlatform, NeMoPlatform
 from nemo_platform_plugin.sdk_provider import (
     DefaultSDKProvider,
     SDKProvider,
@@ -19,6 +19,11 @@
     set_sdk_provider,
 )
 
+
+def _xnmp(sdk) -> dict[str, str]:
+    return {k: v for k, v in sdk.default_headers.items() if k.startswith("X-NMP-")}
+
+
 # ---------------------------------------------------------------------------
 # _read_principal_from_env
 # ---------------------------------------------------------------------------
@@ -142,6 +147,57 @@ def test_get_platform_sdk_on_behalf_of(self, monkeypatch):
         assert sdk.default_headers["X-NMP-Principal-On-Behalf-Of"] == "user@ex.com"
 
 
+# ---------------------------------------------------------------------------
+# get_async_task_sdk — the async sibling of get_task_sdk
+# ---------------------------------------------------------------------------
+
+
+class TestAsyncTaskSdk:
+    def test_async_task_sdk_with_principal(self, monkeypatch):
+        monkeypatch.setenv("NMP_BASE_URL", "http://test:9090")
+        monkeypatch.setenv(
+            "NMP_PRINCIPAL",
+            json.dumps({"id": "creator@ex.com", "email": "creator@ex.com", "groups": ["team"]}),
+        )
+
+        sdk = DefaultSDKProvider().get_async_task_sdk("evaluator")
+
+        assert isinstance(sdk, AsyncNeMoPlatform)
+        assert sdk.base_url == "http://test:9090"
+        assert sdk.default_headers["X-NMP-Principal-Id"] == "service:evaluator"
+        assert sdk.default_headers["X-NMP-Internal"] == "true"
+        assert sdk.default_headers["X-NMP-Principal-On-Behalf-Of"] == "creator@ex.com"
+
+    def test_async_task_sdk_without_principal(self, monkeypatch):
+        monkeypatch.setenv("NMP_BASE_URL", "http://test:9090")
+        monkeypatch.delenv("NMP_PRINCIPAL", raising=False)
+
+        sdk = DefaultSDKProvider().get_async_task_sdk("evaluator")
+
+        assert sdk.default_headers["X-NMP-Principal-Id"] == "service:evaluator"
+        assert "X-NMP-Principal-On-Behalf-Of" not in sdk.default_headers
+
+    def test_parity_with_sync_task_sdk(self, monkeypatch):
+        # Regression guard: the async task SDK must carry the *full* delegated identity (on-behalf-of
+        # id, email, and groups) — wire-identical to get_task_sdk. A prior implementation built on
+        # get_async_platform_sdk dropped the -Email/-Groups headers.
+        monkeypatch.setenv("NMP_BASE_URL", "http://test:9090")
+        monkeypatch.setenv(
+            "NMP_PRINCIPAL",
+            json.dumps(
+                {
+                    "id": "service:evaluator",
+                    "on_behalf_of": "real-user@ex.com",
+                    "on_behalf_of_email": "real-user@ex.com",
+                    "on_behalf_of_groups": ["admin", "team"],
+                }
+            ),
+        )
+
+        provider = DefaultSDKProvider()
+        assert _xnmp(provider.get_async_task_sdk("evaluator")) == _xnmp(provider.get_task_sdk("evaluator"))
+
+
 # ---------------------------------------------------------------------------
 # Provider resolution
 # ---------------------------------------------------------------------------

packages/nmp_common/src/nmp/common/sdk_factory.py

@@ -191,6 +191,34 @@ def get_task_sdk(as_service: str, http_client: httpx.Client | None = None) -> Ne
     )
 
 
+def get_async_task_sdk(as_service: str, http_client: Optional[httpx.AsyncClient] = None) -> AsyncNeMoPlatform:
+    """Async counterpart of :func:`get_task_sdk` for use inside a task container.
+
+    Reads the job creator's principal from ``NMP_PRINCIPAL`` and creates an async SDK that
+    authenticates as the given service while acting on behalf of the job creator with the full
+    delegated identity (on-behalf-of id, email, and groups). Wire-identical to :func:`get_task_sdk`.
+
+    Args:
+        as_service: Service name for the service principal (e.g., "evaluator").
+        http_client: Optional async HTTP client to use for requests.
+
+    Returns:
+        Configured AsyncNeMoPlatform SDK with internal + on-behalf-of headers.
+    """
+    principal = principal_from_env()
+    if principal is None:
+        logger.warning(
+            "NMP_PRINCIPAL not set; async task SDK will authenticate as service:%s without on-behalf-of delegation",
+            as_service,
+        )
+    return get_async_platform_sdk(
+        as_service=as_service,
+        internal=True,
+        http_client=http_client,
+        on_behalf_of=principal.effective_principal if principal else None,
+    )
+
+
 def get_async_platform_sdk(
     as_service: str | None = None,
     internal: bool = False,
@@ -344,6 +372,9 @@ class PlatformSDKProvider:
     def get_task_sdk(self, service_name: str, http_client: httpx.Client | None = None) -> NeMoPlatform:
         return get_task_sdk(service_name, http_client=http_client)
 
+    def get_async_task_sdk(self, service_name: str, http_client: httpx.AsyncClient | None = None) -> AsyncNeMoPlatform:
+        return get_async_task_sdk(service_name, http_client=http_client)
+
     def get_platform_sdk(
         self,
         *,