feat(client): Add new Jobs NemoClient

packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/quickstart/cli.py

@@ -967,6 +967,9 @@ def _run_job_diagnostic(port: int, registry: str, tag: str, *, admin_email: str
     import uuid
 
     from nemo_platform import NeMoPlatform
+    from nemo_platform_plugin.client.adapter import client_from_platform
+    from nemo_platform_plugin.jobs.client import JobsClient
+    from nemo_platform_plugin.jobs.types import CreatePlatformJobRequest
 
     # When auth is enabled, use an unsigned JWT for the admin principal.
     default_headers = None
@@ -997,29 +1000,32 @@ def _run_job_diagnostic(port: int, registry: str, tag: str, *, admin_email: str
         job_name = f"diagnostic-{uuid.uuid4().hex[:8]}"
         console.print(f"  • Creating diagnostic job: {job_name}")
 
-        job = client.jobs.create(
-            platform_spec={
-                "steps": [
-                    {
-                        "name": "diagnostic",
-                        "executor": {
-                            "provider": "cpu",
-                            "container": {
-                                "image": cpu_image,
-                                "entrypoint": [
-                                    "python",
-                                    "-c",
-                                    "import sys; print(f'Python {sys.version}'); print('Job system is working correctly!')",
-                                ],
+        jobs_client = client_from_platform(client, JobsClient)
+        job = jobs_client.create_job(
+            body=CreatePlatformJobRequest(
+                platform_spec={
+                    "steps": [
+                        {
+                            "name": "diagnostic",
+                            "executor": {
+                                "provider": "cpu",
+                                "container": {
+                                    "image": cpu_image,
+                                    "entrypoint": [
+                                        "python",
+                                        "-c",
+                                        "import sys; print(f'Python {sys.version}'); print('Job system is working correctly!')",
+                                    ],
+                                },
                             },
-                        },
-                    }
-                ]
-            },
-            source="quickstart-doctor",
-            spec={},
-            name=job_name,
-        )
+                        }
+                    ]
+                },
+                source="quickstart-doctor",
+                spec={},
+                name=job_name,
+            )
+        ).data()
 
         console.print("  • Waiting for job to complete...")
 
@@ -1028,10 +1034,10 @@ def _run_job_diagnostic(port: int, registry: str, tag: str, *, admin_email: str
         poll_interval = 2
         elapsed = 0
         status = "pending"
-        job_status = client.jobs.retrieve(job.name)
+        job_status = jobs_client.get_job(name=job.name).data()
 
         while elapsed < max_wait:
-            job_status = client.jobs.retrieve(job.name)
+            job_status = jobs_client.get_job(name=job.name).data()
             status = job_status.status
 
             if status in ("completed", "error", "cancelled"):
@@ -1058,9 +1064,9 @@ def _run_job_diagnostic(port: int, registry: str, tag: str, *, admin_email: str
         # Fetch and display logs
         console.print("\n  [bold]Job output:[/bold]")
         try:
-            logs = client.jobs.get_logs(job.name)
+            logs = jobs_client.page_job_logs(name=job.name).data()
             log_lines = []
-            for log_entry in logs:
+            for log_entry in logs.data:
                 if hasattr(log_entry, "message"):
                     log_lines.append(log_entry.message)
 
@@ -1075,7 +1081,7 @@ def _run_job_diagnostic(port: int, registry: str, tag: str, *, admin_email: str
         # Clean up the job (only if successful)
         if status == "completed":
             try:
-                client.jobs.delete(job.name)
+                jobs_client.delete_job(name=job.name)
             except Exception:
                 pass  # Ignore cleanup errors
         else:

packages/nemo_platform_plugin/src/nemo_platform_plugin/client/client.py

@@ -53,13 +53,26 @@
     RetryPolicy,
     Stream,
 )
-from pydantic import BaseModel
+from pydantic import BaseModel, TypeAdapter
 
 ModelT = TypeVar("ModelT", bound=BaseModel)
 
 DEFAULT_TIMEOUT = 60.0
 
 
+def _parse_json_body(response_type: type, data: Any) -> Any:
+    """Parse a decoded JSON body against an endpoint's return annotation.
+
+    Most endpoints return a ``BaseModel`` subclass (parsed via ``model_validate``),
+    but the annotation may be any type — a bare generic (``list[Profile]``,
+    ``dict[str, X]``), a union, etc. Those have no ``model_validate``, so fall back
+    to a ``TypeAdapter`` which validates arbitrary annotated types.
+    """
+    if isinstance(response_type, type) and issubclass(response_type, BaseModel):
+        return response_type.model_validate(data)
+    return TypeAdapter(response_type).validate_python(data)
+
+
 def _get_stream_model_type(response_type: type) -> type[BaseModel]:
     """Extract the ModelT from a Stream[ModelT] generic alias."""
     args = get_args(response_type)
@@ -409,7 +422,7 @@ def send(
         raise_for_status(raw)
         body = None
         if request.response_type is not None:
-            body = request.response_type.model_validate(raw.json())
+            body = _parse_json_body(request.response_type, raw.json())
         return NemoResponse(http_response=raw, body=body, request=request)
 
     def _request_with_retry(
@@ -613,7 +626,7 @@ async def send(
         raise_for_status(raw)
         body = None
         if request.response_type is not None:
-            body = request.response_type.model_validate(raw.json())
+            body = _parse_json_body(request.response_type, raw.json())
         return NemoResponse(http_response=raw, body=body, request=request)
 
     async def _request_with_retry(

packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py

@@ -45,15 +45,14 @@
     StepLifecycleParam,
     SubprocessExecutionProviderParam,
 )
-from nemo_platform.types.jobs import (
-    PlatformJobResponse as PlatformJob,
-)
 from nemo_platform.types.jobs.platform_job_step_spec_param import Executor
 from nemo_platform_plugin.api.filter import ComparisonOperation, FilterOperation, FilterOperator, LogicalOperation
 from nemo_platform_plugin.api.parsed_filter import ParsedFilter, make_filter_dep
 from nemo_platform_plugin.authz import AuthzScope, CallerKind, path_rule
+from nemo_platform_plugin.client.adapter import client_from_platform
 from nemo_platform_plugin.dependencies import get_entity_client, get_sdk_client
 from nemo_platform_plugin.entities import EntityClient
+from nemo_platform_plugin.jobs.client import AsyncJobsClient
 from nemo_platform_plugin.jobs.docker import validate_gpu_available_for_docker
 from nemo_platform_plugin.jobs.exceptions import PlatformJobCompilationError
 from nemo_platform_plugin.jobs.openapi_utils import generate_openapi_extra_params
@@ -65,13 +64,27 @@
     PlatformJobStatus,
     PlatformJobStatusResponse,
 )
+from nemo_platform_plugin.jobs.types import (
+    CreatePlatformJobRequest,
+    JobLogsQueryParams,
+    ListJobsQueryParams,
+)
+from nemo_platform_plugin.jobs.types import (
+    PlatformJobResponse as PlatformJob,
+)
 from nemo_platform_plugin.schema import DatetimeFilter, Filter, Page, PaginationData, StringFilter
 from pydantic import BaseModel, Field, TypeAdapter
 
 logger = logging.getLogger(__name__)
 
 # This type is aliased to ensure we don't expose internal stainless
 # type paths to services integrating the job service.
+#
+# TODO(AIRCORE-827): these still alias the Stainless-generated ``*Param`` TypedDicts.
+# The plugin now owns pydantic equivalents in ``jobs/spec.py`` and ``jobs/providers.py``,
+# but ~10 consuming plugins construct these as dict literals (TypedDict), so repointing
+# them to the pydantic models is a cross-cutting change tracked as the "drop the Stainless
+# dependency" follow-up — out of scope for the Jobs *client* migration.
 PlatformJobSpec = PlatformJobSpecParam
 PlatformJobStep = PlatformJobStepSpecParam
 StepLifecycle = StepLifecycleParam
@@ -857,25 +870,35 @@ async def create_job(
             # Build SDK call kwargs, only including optional fields when they have values
             # (passing None explicitly causes different serialization than omitting)
             # Note: We store transformed_spec (not input), which includes auto-generated fields.
-            sdk_kwargs: dict = {
+            # ``job_spec`` may be a Pydantic model (the transformed job output);
+            # the request body's ``spec`` is a plain dict on the wire. The
+            # Stainless SDK serialized models implicitly — the typed client
+            # validates the body first, so coerce to a dict here.
+            spec_dict = job_spec.model_dump() if isinstance(job_spec, BaseModel) else job_spec
+
+            # Only include optional fields when they have values — passing None
+            # explicitly serializes differently than omitting (exclude_unset).
+            create_fields: dict = {
                 "source": service_name,
-                "spec": job_spec,
+                "spec": spec_dict,
                 "platform_spec": platform_spec,
-                "workspace": workspace,
             }
             # Use the resolved job_name (user-provided or generated)
             if job_name is not None:
-                sdk_kwargs["name"] = job_name
+                create_fields["name"] = job_name
             if request.description is not None:
-                sdk_kwargs["description"] = request.description
+                create_fields["description"] = request.description
             if request.ownership is not None:
-                sdk_kwargs["ownership"] = request.ownership
+                create_fields["ownership"] = request.ownership
             if request.custom_fields is not None:
-                sdk_kwargs["custom_fields"] = request.custom_fields
+                create_fields["custom_fields"] = request.custom_fields
             if request.project:
-                sdk_kwargs["extra_body"] = {"project": request.project}
+                create_fields["project"] = request.project
 
-            job_resp = await sdk.jobs.create(**sdk_kwargs)
+            jobs = client_from_platform(sdk, AsyncJobsClient)
+            job_resp = (
+                await jobs.create_job(workspace=workspace, body=CreatePlatformJobRequest(**create_fields))
+            ).data()
             return from_response(job_resp)
 
         @router.get(
@@ -930,17 +953,25 @@ async def list_jobs(
             # accepts raw JSON in ``filter=`` and routes it through
             # parse_json_filter, so a single JSON-string param survives the
             # round trip cleanly.
-            sdk_list_kwargs: dict = {
-                "workspace": workspace,
+            list_query: ListJobsQueryParams = {
                 "page": page,
                 "page_size": page_size,
                 "sort": str(sort),
-                "extra_query": {"filter": json.dumps(parsed.to_response())},
+                "filter": json.dumps(parsed.to_response()),
             }
-            list_jobs_resp = await sdk.jobs.list(**sdk_list_kwargs)
+            jobs = client_from_platform(sdk, AsyncJobsClient)
+            list_page = (await jobs.list_jobs(workspace=workspace, query_params=list_query)).page()
             return Page(
-                data=[from_response(job) for job in list_jobs_resp.data],
-                pagination=PaginationData(**list_jobs_resp.pagination.model_dump()),
+                data=[from_response(job) for job in list_page.items],
+                pagination=PaginationData(
+                    # The list envelope always carries pagination metadata; coalesce
+                    # to the request values / zero to satisfy the non-optional PaginationData.
+                    page=list_page.page if list_page.page is not None else page,
+                    page_size=list_page.page_size if list_page.page_size is not None else page_size,
+                    current_page_size=len(list_page.items),
+                    total_pages=list_page.total_pages or 0,
+                    total_results=list_page.total_results or 0,
+                ),
                 sort=sort,
                 filter=user_filter or None,
             )
@@ -955,7 +986,7 @@ async def get_job(
         ) -> TypedJobResponse:
             f"""Get a job by name for the {service_name} microservice."""
 
-            job_resp = await sdk.jobs.retrieve(name=name, workspace=workspace)
+            job_resp = (await client_from_platform(sdk, AsyncJobsClient).get_job(name=name, workspace=workspace)).data()
             return from_response(job_resp)
 
         # Status
@@ -968,7 +999,9 @@ async def get_job_status(
             sdk: AsyncNeMoPlatform = Depends(get_sdk_client),
         ) -> PlatformJobStatusResponse:
             f"""Get the status of a job by name for the {service_name} microservice."""
-            job_resp = await sdk.jobs.get_status(name=name, workspace=workspace)
+            job_resp = (
+                await client_from_platform(sdk, AsyncJobsClient).get_job_status(name=name, workspace=workspace)
+            ).data()
             return PlatformJobStatusResponse(**job_resp.model_dump())
 
         @router.delete(
@@ -981,7 +1014,7 @@ async def delete_job(
             sdk: AsyncNeMoPlatform = Depends(get_sdk_client),
         ) -> None:
             f"""Delete a job by name for the {service_name} microservice."""
-            await sdk.jobs.delete(name=name, workspace=workspace)
+            await client_from_platform(sdk, AsyncJobsClient).delete_job(name=name, workspace=workspace)
             return None
 
         @router.post(
@@ -994,7 +1027,9 @@ async def cancel_job(
         ) -> TypedJobResponse:
             f"""Cancel a job by name for the {service_name} microservice."""
 
-            job_resp = await sdk.jobs.cancel(name=name, workspace=workspace)
+            job_resp = (
+                await client_from_platform(sdk, AsyncJobsClient).cancel_job(name=name, workspace=workspace)
+            ).data()
             return from_response(job_resp)
 
         # Logs
@@ -1010,7 +1045,16 @@ async def get_job_logs(
         ) -> PlatformJobLogPage:
             f"""Get the logs of a job by name for the {service_name} microservice."""
 
-            logs = await sdk.jobs.get_logs(workspace=workspace, name=name, limit=limit, page_cursor=page_cursor)
+            logs_query: JobLogsQueryParams = {}
+            if limit is not None:
+                logs_query["limit"] = limit
+            if page_cursor is not None:
+                logs_query["page_cursor"] = page_cursor
+            logs = (
+                await client_from_platform(sdk, AsyncJobsClient).page_job_logs(
+                    workspace=workspace, name=name, query_params=logs_query
+                )
+            ).data()
             return PlatformJobLogPage(**logs.model_dump())
 
         # Results
@@ -1025,7 +1069,9 @@ async def list_job_results(
         ) -> PlatformJobListResultResponse:
             f"""Get the results of a job by name for the {service_name} microservice."""
 
-            results = await sdk.jobs.results.list(name=name, workspace=workspace)
+            results = (
+                await client_from_platform(sdk, AsyncJobsClient).list_job_results(name=name, workspace=workspace)
+            ).data()
             result_dicts = [result.model_dump() for result in results.data]
             list_results = []
             for result_dict in result_dicts:
@@ -1047,7 +1093,9 @@ async def get_job_result(
         ) -> PlatformJobResultResponse:
             f"""Get the result of a job by name for the {service_name} microservice."""
 
-            result_obj = await sdk.jobs.results.retrieve(name=name, job=job, workspace=workspace)
+            result_obj = (
+                await client_from_platform(sdk, AsyncJobsClient).get_job_result(name=name, job=job, workspace=workspace)
+            ).data()
 
             # Construct the URL for downloading this result
             result_dict = result_obj.model_dump()
@@ -1091,7 +1139,9 @@ async def _download_route_helper(
             - Use the `result_serializer` to know how to properly serialize the output
             """
 
-            result_info = await sdk.jobs.results.retrieve(name=name, job=job, workspace=workspace)
+            result_info = (
+                await client_from_platform(sdk, AsyncJobsClient).get_job_result(name=name, job=job, workspace=workspace)
+            ).data()
             _, tmp_dir_path = await download_from_result_info(
                 result_name=name,
                 job_name=job,
@@ -1203,7 +1253,9 @@ async def pause_job(
         ) -> TypedJobResponse:
             f"""Pause a job by name for the {service_name} microservice."""
 
-            job_resp = await sdk.jobs.pause(name=name, workspace=workspace)
+            job_resp = (
+                await client_from_platform(sdk, AsyncJobsClient).pause_job(name=name, workspace=workspace)
+            ).data()
             return from_response(job_resp)
 
         @router.post(
@@ -1216,7 +1268,9 @@ async def resume_job(
         ) -> TypedJobResponse:
             f"""Resume a job by name for the {service_name} microservice."""
 
-            job_resp = await sdk.jobs.resume(name=name, workspace=workspace)
+            job_resp = (
+                await client_from_platform(sdk, AsyncJobsClient).resume_job(name=name, workspace=workspace)
+            ).data()
             return from_response(job_resp)
 
         _stamp(pause_job, perm="pause", write=True)