feat: invoke deployed flash tasks with runtime credentials (SEC-375 CON-1833)

CHANGELOG.md

@@ -5,6 +5,13 @@
 ### Features
 
 * Apps expose `NetworkVolume` and `GlobalVolume` resources with explicit path-keyed mounts, lazy create-if-missing provisioning, and execution-bound filesystem access. Global volumes use GraphQL for lookup and creation.
+* Deployed Flash calls use endpoint or pod runtime credentials. Release requires deployed support from [RunPod#6686](https://github.com/runpod/RunPod/pull/6686), [ai-api#1176](https://github.com/runpod/ai-api/pull/1176), and [sdk-runtime#4](https://github.com/runpod-workers/sdk-runtime/pull/4).
+
+### Breaking Changes
+
+* Production endpoint environment variables reserve `RUNPOD_API_KEY` and runtime identity fields. Workers authenticate Flash calls with platform-issued runtime credentials; application credentials must use non-reserved environment names.
+* Deployed workers can invoke Flash resources only within their own app and environment, including through `Queue`, `Api`, and `Task` stubs. Local clients use account credentials subject to their permissions.
+* Remote queues, tasks, and JSON API calls require finite JSON-compatible arguments and results in local, dev, and deployed execution. Dev workers receive source separately from the JSON argument contract. Explicit `.local()` calls run Python directly.
 
 ### Bug Fixes
 

runpod/__init__.py

@@ -54,6 +54,7 @@
     "Model": "runpod.apps",
     "NetworkVolume": "runpod.apps",
     "Queue": "runpod.apps",
+    "Task": "runpod.apps",
     "Secret": "runpod.apps",
     "Volume": "runpod.apps",
     "is_local": "runpod.apps",

runpod/api/graphql.py

@@ -87,6 +87,7 @@ async def run_graphql_query_async(
     variables: Optional[Dict[str, Any]] = None,
     timeout: float = 60.0,
     anonymous: bool = False,
+    headers: Optional[Dict[str, str]] = None,
 ) -> Dict[str, Any]:
     """
     Async variant of run_graphql_query, sharing the same url, headers,
@@ -110,7 +111,7 @@ async def run_graphql_query_async(
     async with aiohttp.ClientSession(timeout=client_timeout) as session:
         async with session.post(
             _graphql_url(),
-            headers=_build_headers(effective_api_key),
+            headers={**_build_headers(effective_api_key), **(headers or {})},
             json=_build_payload(query, variables),
         ) as response:
             if response.status == HTTP_STATUS_UNAUTHORIZED:

runpod/api/mutations/apps.py

@@ -92,6 +92,12 @@
 }
 """
 
+MUTATION_FINALIZE_TASK_ARTIFACT_UPLOAD = """
+mutation FinalizeArtifactUpload($input: FinalizeFlashArtifactUploadInput!) {
+    finalizeFlashArtifactUpload(input: $input) { id manifest resourceSpec }
+}
+"""
+
 MUTATION_DEPLOY_BUILD = """
 mutation deployBuildToEnvironment($input: DeployBuildToEnvironmentInput!) {
     deployBuildToEnvironment(input: $input) { id name }

runpod/apps/__init__.py

@@ -15,7 +15,7 @@
 from .markers import delete, get, init, patch, post, put
 from .schedule import schedule
 from .spec import ResourceKind, ResourceSpec, RouteSpec
-from .stubs import Api, Queue
+from .stubs import Api, Queue, Task
 from .model import Model
 from .secret import Secret
 from .volume import GlobalVolume, NetworkVolume, Volume
@@ -34,6 +34,7 @@
     "Model",
     "NetworkVolume",
     "Queue",
+    "Task",
     "Secret",
     "Volume",
     "RemoteExecutionError",

runpod/apps/api.py

@@ -19,6 +19,46 @@
 _TRANSPORT_RETRIES = 4
 
 
+class FlashTaskApiClient:
+    """named task lifecycle operations with runtime-scoped authentication."""
+
+    async def _execute(self, query: str, task_input: Dict[str, Any]) -> Dict[str, Any]:
+        from .utils.network import api_key, runtime_headers
+
+        headers = runtime_headers()
+        response = await run_graphql_query_async(
+            query,
+            variables={"input": task_input},
+            api_key=(headers["Authorization"][7:] if headers else api_key()),
+            headers=headers,
+        )
+        return response["data"]
+
+    async def create(self, task_input: Dict[str, Any]) -> Dict[str, Any]:
+        from .utils.network import validate_runtime_target
+
+        validate_runtime_target(
+            task_input.get("flashAppName"), task_input.get("flashEnvironmentName")
+        )
+        data = await self._execute(
+            """mutation FlashTaskCreate($input: FlashTaskCreateInput!) {
+                flashTaskCreate(input: $input) { podId flashBuildId resourceName }
+            }""",
+            task_input,
+        )
+        return data["flashTaskCreate"]
+
+    async def terminate(self, pod_id: str, token: str) -> None:
+        data = await self._execute(
+            """mutation FlashTaskTerminate($input: FlashTaskTerminateInput!) {
+                flashTaskTerminate(input: $input)
+            }""",
+            {"podId": pod_id, "executionToken": token},
+        )
+        if data.get("flashTaskTerminate") is not True:
+            raise RuntimeError("task termination was not acknowledged")
+
+
 def _rest_endpoint_input(endpoint: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     """translate only saves whose capabilities rest can preserve atomically."""
     fields = {
@@ -574,6 +614,8 @@ async def finalize_artifact_upload(
         self, app_id: str, object_key: str, manifest: Dict[str, Any]
     ) -> Dict[str, Any]:
         mutation = app_mutations.MUTATION_FINALIZE_ARTIFACT_UPLOAD
+        if any(r.get("kind") == "task" for r in manifest.get("resources", [])):
+            mutation = app_mutations.MUTATION_FINALIZE_TASK_ARTIFACT_UPLOAD
         data = await self._execute(
             mutation,
             {

runpod/apps/app.py

@@ -151,7 +151,11 @@ def task(
         accelerate_downloads: bool = True,
         container_disk_gb: Optional[int] = None,
     ) -> Callable[[Callable], FunctionHandle]:
-        """declare ephemeral pod compute from a function."""
+        """declare ephemeral pod compute from a function.
+
+        remote task arguments and results use json in every execution
+        context. use .local() for direct python execution without transport.
+        """
 
         def decorator(fn: Callable) -> FunctionHandle:
             spec = ResourceSpec(
@@ -259,6 +263,12 @@ async def _resolve(self, spec: ResourceSpec) -> InvocationTarget:
         """resolve a resource spec to an invocation target."""
         spec.validate()
         if spec.kind is ResourceKind.TASK:
+            from .context import deployed_identity
+            from .targets import DeployedTaskTarget
+
+            identity = deployed_identity()
+            if identity is not None:
+                return DeployedTaskTarget(self.name, identity[1], spec.name)
             handle = self._resources.get(spec.name)
             fn = getattr(handle, "_fn", None)
             return PodTarget(

runpod/apps/context.py

@@ -7,7 +7,7 @@
 import threading
 import time
 from enum import Enum
-from typing import Any, Coroutine
+from typing import Any, Coroutine, Optional, Tuple
 
 log = logging.getLogger(__name__)
 CLEANUP_TIMEOUT = 30.0
@@ -36,6 +36,27 @@ def current_context() -> Context:
     return Context.LOCAL
 
 
+def deployed_identity() -> Optional[Tuple[str, str, str]]:
+    """app, environment and resource identity for a deployed worker."""
+    if not (
+        os.getenv("RUNPOD_RUNTIME_KIND") == "task"
+        or os.getenv("RUNPOD_BUILD_ID")
+        or os.getenv("FLASH_APP_NAME")
+        or os.getenv("FLASH_RESOURCE_NAME")
+    ):
+        return None
+    fields = tuple(
+        os.getenv(key)
+        for key in ("FLASH_APP_NAME", "FLASH_ENVIRONMENT", "FLASH_RESOURCE_NAME")
+    )
+    build = os.getenv("RUNPOD_BUILD_ID")
+    if any(not value or value != value.strip() for value in (*fields, build)):
+        raise RuntimeError(
+            "deployed flash requires app, environment, resource and build identity"
+        )
+    return fields
+
+
 def is_local() -> bool:
     """true when not running inside a runpod container.