feat(auditor): add blocking submit option to auditor SDK

docs/auditor/sdk-resources.mdx

@@ -33,6 +33,9 @@ auditor = client.auditor  # AuditorPluginResource
 | `plugin_status()` | Returns auditor plugin health information from the service. | `dict[str, object]` |
 | `configs` | Sub-resource for `AuditConfig` CRUD operations. | `_ConfigResource` |
 | `targets` | Sub-resource for `AuditTarget` CRUD operations. | `_TargetResource` |
+| `submit()` | Submits a K8s audit job and returns a handle for polling and artifact download. | `AuditorJobResource` |
+| `list_jobs(workspace, page, page_size)` | Lists submitted audit jobs in the workspace. | `dict` |
+| `get_job(job_name, workspace)` | Fetches a single audit job by name. | `dict` |
 | `run()` | Runs one audit locally, in-process, against a configured target. | `dict` |
 
 ### `configs` sub-resource
@@ -59,6 +62,59 @@ Five CRUD methods for `AuditTarget` entities. The full field reference is in [Ta
 | `update(*, workspace, name, type, model, options=None, description=None)` | Replaces a target's fields. | `AuditTarget` |
 | `delete(*, workspace, name)` | Deletes a target. | `None` |
 
+### `submit()` arguments
+
+`submit()` posts an audit job to the K8s executor and returns an `AuditorJobResource` handle.
+Call `.wait_until_done()` on the handle to block until the job completes, then `.download_artifacts()` to fetch the garak reports.
+
+| Argument | Type | Required | Description |
+|----------|------|----------|-------------|
+| `config` | `AuditConfig \| str` | Yes | An inline `AuditConfig` instance or a name string referencing one in the entity store. Bare names resolve against `workspace`; qualified names such as `"prod/quick-scan"` override the workspace. |
+| `target` | `AuditTarget \| str` | Yes | An inline `AuditTarget` instance or a name string, with the same resolution rules as `config`. |
+| `workspace` | `str \| None` | No | Workspace to submit the job into. Defaults to `"default"`. |
+| `max_probe_retries` | `int` | No | Number of times to retry a failing garak probe before marking it as failed. Defaults to `0`. |
+| `fail_job_on_retries_exhausted` | `bool` | No | When `True` (the default), the job fails if any probe exhausts its retries. Set to `False` to treat retry-exhausted probes as warnings. |
+
+### `AuditorJobResource`
+
+The object returned by `submit()`. Use it to poll status, stream logs, and download artifacts.
+
+| Method | Description | Returns |
+|--------|-------------|---------|
+| `name` | The unique job name assigned by the platform. | `str` |
+| `get_job()` | Fetches the full job dict (name, status, workspace, …). | `dict[str, object]` |
+| `get_job_status()` | Fetches only the current platform status string. | `PlatformJobStatus \| None` |
+| `check_if_complete(raise_if_not_complete=False)` | Returns `True` if the job is `completed`. Raises `RuntimeError` when `raise_if_not_complete=True` and the job is not done. | `bool` |
+| `wait_until_done()` | Blocks until the job reaches a terminal status. Streams log entries from the audit task while polling. Raises `RuntimeError` on a terminal failure. | `None` |
+| `get_logs()` | Pages through all structured log entries produced by the audit task. | `list[dict[str, str]]` |
+| `download_artifacts(path=None)` | Downloads and extracts the garak report tarball. Raises `RuntimeError` if the job has not completed. `path` overrides the output directory (defaults to a directory named after the job). | `Path` |
+
+### Submit and wait for an audit job
+
+```python
+# Submit the job using persisted entity name strings.
+job = auditor.submit(
+    config="quick-scan",
+    target="llama-31-8b",
+    workspace="default",
+)
+print(f"Job submitted: {job.name}")
+
+# Block until garak finishes (raises RuntimeError on failure).
+job.wait_until_done()
+
+# Download the garak reports to ./my-reports/<job-name>/.
+artifacts_dir = job.download_artifacts(path="./my-reports")
+print(f"Reports saved to: {artifacts_dir}")
+```
+
+You can also check status without blocking:
+
+```python
+if not job.check_if_complete():
+    print(f"Still running: {job.get_job_status()}")
+```
+
 ### `run()` arguments
 
 `run()` invokes [garak](https://github.com/NVIDIA/garak) locally, in-process, against a configured target.
@@ -162,15 +218,30 @@ auditor = client.auditor  # AsyncAuditorPluginResource
 | `plugin_status()` | Returns auditor plugin health information from the service. | `dict[str, object]` |
 | `configs` | Sub-resource for `AuditConfig` CRUD operations. | `_AsyncConfigResource` |
 | `targets` | Sub-resource for `AuditTarget` CRUD operations. | `_AsyncTargetResource` |
+| `submit()` | Submits a K8s audit job and returns an async handle for polling and artifact download. | `AsyncAuditorJobResource` |
+| `list_jobs(workspace, page, page_size)` | Lists submitted audit jobs in the workspace. | `dict` |
+| `get_job(job_name, workspace)` | Fetches a single audit job by name. | `dict` |
 | `run()` | Runs one audit locally, in-process, against a configured target. | `dict` |
 
-`AsyncAuditorPluginResource.run()` and the async `configs` / `targets` sub-resource methods accept the same arguments as their sync counterparts [above](#run-arguments). Because the local execution path is synchronous (garak runs in a subprocess), the async `run()` dispatches the scheduler call through `asyncio.to_thread` so the caller's event loop is not blocked.
+`AsyncAuditorPluginResource.submit()` returns an `AsyncAuditorJobResource` with the same methods as `AuditorJobResource` [above](#auditorjobresource), all awaitable.
+`AsyncAuditorPluginResource.run()` and the async `configs` / `targets` sub-resource methods accept the same arguments as their sync counterparts. Because the local execution path is synchronous (garak runs in a subprocess), the async `run()` dispatches the scheduler call through `asyncio.to_thread` so the caller's event loop is not blocked.
 
 ```python
 import asyncio
 
 
 async def main() -> None:
+    # Submit and wait.
+    job = await auditor.submit(
+        config="quick-scan",
+        target="llama-31-8b",
+        workspace="default",
+    )
+    await job.wait_until_done()
+    artifacts_dir = await job.download_artifacts()
+    print(f"Reports saved to: {artifacts_dir}")
+
+    # Or run locally (no jobs-service submission).
     result = await auditor.run(
         config="quick-scan",
         target="llama-31-8b",

e2e/auditor/test_audit_job.py

@@ -181,7 +181,7 @@ def test_audit_job_submit_blank_probe(
     }
 
     job = sdk.auditor.submit(config=config, target=target, workspace=audit_workspace)
-    job_name = job["name"]
+    job_name = job.name
     try:
         final_status = _wait_for_audit_job(sdk, job_name, audit_workspace)
         assert final_status == "completed", (
@@ -205,7 +205,7 @@ def test_audit_job_submit_with_entity_refs(
         target=f"{audit_workspace}/{audit_target_name}",
         workspace=audit_workspace,
     )
-    job_name = job["name"]
+    job_name = job.name
     try:
         final_status = _wait_for_audit_job(sdk, job_name, audit_workspace)
         assert final_status == "completed", (
@@ -227,7 +227,7 @@ def test_audit_job_appears_in_list(
         target=f"{audit_workspace}/{audit_target_name}",
         workspace=audit_workspace,
     )
-    job_name = job["name"]
+    job_name = job.name
     try:
         jobs = sdk.auditor.list_jobs(workspace=audit_workspace)
         job_names = [j["name"] for j in jobs.get("data", [])]

plugins/nemo-auditor/README.md

@@ -81,7 +81,18 @@ tgt = client.auditor.targets.create(
     options={"uri": "http://localhost:9000/v1"},
 )
 
-# Run an audit locally (no jobs-service submission) using the persisted entities
+# Submit a K8s audit job and wait for it to finish.
+job = client.auditor.submit(
+    config="quick-scan",
+    target="llama-31-8b",
+    workspace="default",
+)
+print(f"Job submitted: {job.name}")
+job.wait_until_done()                          # blocks; streams logs while polling
+artifacts_dir = job.download_artifacts()       # extracts garak reports to ./<job-name>/
+print(f"Reports: {artifacts_dir}")
+
+# Or run an audit locally (no jobs-service submission).
 result = client.auditor.run(
     config="quick-scan",       # workspace-qualified name strings ("ws/name") also work
     target="llama-31-8b",
@@ -92,6 +103,7 @@ for name, ref in result["results"].items():
     print(name, ref["artifact_url"])
 ```
 
+`submit()` posts the job to the K8s executor and returns an `AuditorJobResource` handle.
 `run()` shells out to a pre-installed garak interpreter (default
 `~/.auditor/.venv/bin/python`, override via `$NEMO_AUDITOR_GARAK_PYTHON`)
 and registers the resulting JSONL / HTML / hitlog reports as job results

plugins/nemo-auditor/src/nemo_auditor/sdk.py

@@ -10,7 +10,9 @@
 - ``client.auditor.configs.{create,list,get,update,delete}`` — ``AuditConfig`` CRUD.
 - ``client.auditor.targets.{create,list,get,update,delete}`` — ``AuditTarget`` CRUD.
 - ``client.auditor.submit(config=..., target=..., workspace=...)`` — submit a K8s
-  audit job through the plugin's job endpoint and return the raw job dict.
+  audit job and return an :class:`~nemo_auditor.sdk_resources.job_resources.AuditorJobResource`
+  handle. Call ``.wait_until_done()`` on the handle to block until the job completes,
+  then ``.download_artifacts()`` to fetch the garak report tarball.
 - ``client.auditor.list_jobs(workspace=...)`` — list submitted audit jobs.
 - ``client.auditor.get_job(job_name, workspace=...)`` — fetch a single audit job.
 - ``client.auditor.run(config=..., target=..., workspace=...)`` — in-process
@@ -29,6 +31,7 @@
 from nemo_auditor.entities import AuditConfig, AuditTarget
 from nemo_auditor.jobs.audit import AuditInputSpec, AuditJob
 from nemo_auditor.sdk_resources.configs import _AsyncConfigResource, _ConfigResource
+from nemo_auditor.sdk_resources.job_resources import AsyncAuditorJobResource, AuditorJobResource
 from nemo_auditor.sdk_resources.targets import _AsyncTargetResource, _TargetResource
 from nemo_platform import AsyncNeMoPlatform, NeMoPlatform
 from nemo_platform_plugin.entities import parse_qualified_name
@@ -73,12 +76,12 @@ def submit(
         workspace: str | None = None,
         max_probe_retries: int = 0,
         fail_job_on_retries_exhausted: bool = True,
-    ) -> dict:
+    ) -> AuditorJobResource:
         """Submit an audit job to the K8s executor via the plugin job endpoint.
 
-        Returns the raw job dict (name, status, workspace, …). Use
-        ``sdk.jobs.get_status(name=result["name"], workspace=workspace)`` to poll
-        for completion, or pass the name to ``sdk.auditor.get_job()``.
+        Returns an :class:`~nemo_auditor.sdk_resources.job_resources.AuditorJobResource`
+        handle. Call ``.wait_until_done()`` to block until the job completes, then
+        ``.download_artifacts()`` to fetch the garak report tarball.
         """
         ws = workspace or "default"
         spec = AuditInputSpec(
@@ -92,7 +95,7 @@ def submit(
             json={"spec": spec.model_dump(mode="json")},
         )
         response.raise_for_status()
-        return response.json()
+        return AuditorJobResource(job_name=response.json()["name"], platform=self._platform, workspace=ws)
 
     def list_jobs(
         self,
@@ -199,7 +202,7 @@ async def submit(
         workspace: str | None = None,
         max_probe_retries: int = 0,
         fail_job_on_retries_exhausted: bool = True,
-    ) -> dict:
+    ) -> AsyncAuditorJobResource:
         """Async twin of :meth:`AuditorPluginResource.submit`."""
         ws = workspace or "default"
         spec = AuditInputSpec(
@@ -213,7 +216,7 @@ async def submit(
             json={"spec": spec.model_dump(mode="json")},
         )
         response.raise_for_status()
-        return response.json()
+        return AsyncAuditorJobResource(job_name=response.json()["name"], platform=self._platform, workspace=ws)
 
     async def list_jobs(
         self,