From 0f2d7f646d0f7bff0326070ecf249f5b5d477b6b Mon Sep 17 00:00:00 2001
From: zeke <40004347+KAJdev@users.noreply.github.com>
Date: Tue, 29 Sep 2026 14:21:11 -0700
Subject: [PATCH 1/6] feat: initialize worker volume bindings before user code

---
 CHANGELOG.md                           |  6 ++
 README.md                              | 16 +++++
 src/runpod_sdk_runtime/api/server.py   |  3 +
 src/runpod_sdk_runtime/mounts.py       | 15 +++++
 src/runpod_sdk_runtime/queue/worker.py |  3 +
 src/runpod_sdk_runtime/task/runner.py  |  2 +
 tests/test_api_server.py               | 57 ++++++++++++++++-
 tests/test_queue_worker.py             | 86 +++++++++++++++++++++++---
 tests/test_task_runner_http.py         | 45 +++++++++++++-
 9 files changed, 222 insertions(+), 11 deletions(-)
 create mode 100644 src/runpod_sdk_runtime/mounts.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 825c32f..b6564b7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1 +1,7 @@
 # Changelog
+
+## Unreleased
+
+### Features
+
+* Initialize resolved volume bindings before user code in queue, API, and task workers.
diff --git a/README.md b/README.md
index 0906a4b..b6f34e3 100644
--- a/README.md
+++ b/README.md
@@ -38,6 +38,22 @@ python -m runpod_sdk_runtime.task.runner
 
 `RUNPOD_RUNTIME_KIND` selects `queue` or `api` when running the shared bootstrap.
 
+## Storage bindings
+
+Apps provisioning supplies `RUNPOD_MOUNTS` as a JSON array of resolved bindings.
+Each entry contains `kind` (`network` or `global`), the declared `reference`, the
+resolved volume `id`, and its absolute `path`.
+
+Queue, API, and task workers install these bindings before importing user modules
+or running initialization hooks. The SDK's `volume.path` accessor uses this
+worker-level map across synchronous, asynchronous, streaming, and threaded code.
+An absent variable represents an empty mount map; malformed configuration fails
+startup. Request payloads do not configure filesystem mounts.
+
+The runtime requires an Apps SDK build providing `NetworkVolume`, `GlobalVolume`,
+and the worker mount-binding contract. Publish compatible SDK and runtime
+packages/images together before deploying apps with storage bindings.
+
 ## Development
 
 Install the Apps SDK checkout first, then install this project:
diff --git a/src/runpod_sdk_runtime/api/server.py b/src/runpod_sdk_runtime/api/server.py
index 72143b0..1415493 100644
--- a/src/runpod_sdk_runtime/api/server.py
+++ b/src/runpod_sdk_runtime/api/server.py
@@ -29,6 +29,8 @@
 import sys
 from typing import Any, Optional
 
+from runpod_sdk_runtime.mounts import configure_mounts
+
 log = logging.getLogger("runpod_sdk_runtime.api")
 
 APP_DIR = os.environ.get("RUNPOD_APP_DIR", "/app")
@@ -293,6 +295,7 @@ def _build_live_app() -> Any:
 
 
 def build_app() -> Any:
+    configure_mounts()
     if _is_deployed():
         handle = _load_api_handle()
         if handle._cls is not None:
diff --git a/src/runpod_sdk_runtime/mounts.py b/src/runpod_sdk_runtime/mounts.py
new file mode 100644
index 0000000..241a0e4
--- /dev/null
+++ b/src/runpod_sdk_runtime/mounts.py
@@ -0,0 +1,15 @@
+"""install provisioning-owned filesystem bindings at worker startup."""
+
+import json
+import os
+
+
+def configure_mounts() -> None:
+    """initialize the sdk after bootstrap, before loading any user code."""
+    from runpod.apps.volume import _configure_mounts
+
+    try:
+        bindings = json.loads(os.environ.get("RUNPOD_MOUNTS", "[]"))
+    except json.JSONDecodeError as exc:
+        raise RuntimeError("RUNPOD_MOUNTS must contain a valid JSON array") from exc
+    _configure_mounts(bindings)
diff --git a/src/runpod_sdk_runtime/queue/worker.py b/src/runpod_sdk_runtime/queue/worker.py
index 2f3a884..67012c9 100644
--- a/src/runpod_sdk_runtime/queue/worker.py
+++ b/src/runpod_sdk_runtime/queue/worker.py
@@ -24,6 +24,8 @@
 
 import runpod
 
+from runpod_sdk_runtime.mounts import configure_mounts
+
 APP_DIR = os.environ.get("RUNPOD_APP_DIR", "/app")
 MANIFEST_NAME = "runpod_manifest.json"
 
@@ -173,6 +175,7 @@ def _worker_config(handler) -> dict:
 
 
 def main() -> None:
+    configure_mounts()
     if _is_deployed():
         handle = _load_deployed_handle()
         _run_init(handle)
diff --git a/src/runpod_sdk_runtime/task/runner.py b/src/runpod_sdk_runtime/task/runner.py
index 8c1744d..1836ae6 100644
--- a/src/runpod_sdk_runtime/task/runner.py
+++ b/src/runpod_sdk_runtime/task/runner.py
@@ -26,6 +26,7 @@
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 from runpod_sdk_runtime.executor import execute_request
+from runpod_sdk_runtime.mounts import configure_mounts
 
 PORT = int(os.environ.get("RUNPOD_TASK_PORT", "8080"))
 TOKEN = os.environ.get("RUNPOD_TASK_TOKEN", "")
@@ -199,6 +200,7 @@ def main() -> None:
     if not TOKEN:
         sys.stderr.write("[task-runner] RUNPOD_TASK_TOKEN not set, exiting\n")
         sys.exit(1)
+    configure_mounts()
     _touch_contact()
     threading.Thread(target=_watchdog, daemon=True).start()
     server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
diff --git a/tests/test_api_server.py b/tests/test_api_server.py
index 7810f1d..eaa6317 100644
--- a/tests/test_api_server.py
+++ b/tests/test_api_server.py
@@ -13,10 +13,15 @@
 
 
 @pytest.fixture(autouse=True)
-def clean_registry():
+def clean_registry(monkeypatch):
+    from runpod.apps.volume import _configure_mounts
+
+    monkeypatch.delenv("RUNPOD_MOUNTS", raising=False)
+    _configure_mounts([])
     _clear_registry()
     yield
     _clear_registry()
+    _configure_mounts([])
 
 
 def _write_project(tmp_path, monkeypatch, module="main_cls"):
@@ -254,3 +259,53 @@ def test_resync_same_source_keeps_state(self):
         # identical source: no rebuild, state survives
         client.post("/_runpod/sync", json={"source": source, "resource": "S"})
         assert client.post("/add", json={"x": 3}).json() == {"total": 5}
+
+
+@pytest.mark.parametrize("deployed", [False, True])
+def test_mounts_before_module_init_and_routes(monkeypatch, tmp_path, deployed):
+    monkeypatch.setenv(
+        "RUNPOD_MOUNTS",
+        json.dumps(
+            [{"kind": "network", "reference": "data", "id": "nv-1", "path": "/data"}]
+        ),
+    )
+    source = textwrap.dedent(
+        """
+        from runpod import App, get, init
+        from runpod.apps.volume import NetworkVolume
+
+        imported = str(NetworkVolume("data").path)
+        app = App("mounts-api")
+
+        @app.api(name="mounted", cpu="cpu3c-1-2")
+        class Mounted:
+            @init
+            async def setup(self):
+                self.initialized = str(NetworkVolume("nv-1").path)
+
+            @get("/mounted")
+            def mounted(self):
+                return [imported, self.initialized, str(NetworkVolume("data").path)]
+        """
+    )
+    if deployed:
+        (tmp_path / "mounts_api.py").write_text(source)
+        (tmp_path / server.MANIFEST_NAME).write_text(
+            json.dumps({"resources": [{"name": "mounted", "module": "mounts_api"}]})
+        )
+        monkeypatch.setattr(server, "APP_DIR", str(tmp_path))
+        monkeypatch.setenv("FLASH_RESOURCE_NAME", "mounted")
+    else:
+        monkeypatch.delenv("FLASH_RESOURCE_NAME", raising=False)
+        monkeypatch.delenv("RUNPOD_RESOURCE_NAME", raising=False)
+    app = server.build_app()
+    # lifespan, live sync, and threaded routes must share the startup bindings.
+    monkeypatch.setenv("RUNPOD_MOUNTS", "[]")
+    with TestClient(app) as client:
+        if not deployed:
+            response = client.post(
+                "/_runpod/sync",
+                json={"source": source, "resource": "mounted", "mounts": []},
+            )
+            assert response.status_code == 200
+        assert client.get("/mounted").json() == ["/data", "/data", "/data"]
diff --git a/tests/test_queue_worker.py b/tests/test_queue_worker.py
index b175e72..c6f173b 100644
--- a/tests/test_queue_worker.py
+++ b/tests/test_queue_worker.py
@@ -9,6 +9,16 @@
 from runpod_sdk_runtime.queue import worker
 
 
+@pytest.fixture(autouse=True)
+def clean_mounts(monkeypatch):
+    from runpod.apps.volume import _configure_mounts
+
+    monkeypatch.delenv("RUNPOD_MOUNTS", raising=False)
+    _configure_mounts([])
+    yield
+    _configure_mounts([])
+
+
 class TestModeDetection:
     def test_resource_name_flash_env(self, monkeypatch):
         monkeypatch.setenv("FLASH_RESOURCE_NAME", "chat")
@@ -192,21 +202,67 @@ def gen_handler(job):
 
 
 class TestMain:
-    def test_live_mode(self, monkeypatch):
+    async def test_live_mode(self, monkeypatch):
         monkeypatch.delenv("FLASH_RESOURCE_NAME", raising=False)
         monkeypatch.delenv("RUNPOD_RESOURCE_NAME", raising=False)
+        monkeypatch.setenv(
+            "RUNPOD_MOUNTS",
+            json.dumps(
+                [
+                    {
+                        "kind": "network",
+                        "reference": "data",
+                        "id": "nv-1",
+                        "path": "/data",
+                    }
+                ]
+            ),
+        )
         with patch("runpod.serverless.start") as start:
             worker.main()
-        config = start.call_args[0][0]
-        assert config["handler"] is worker._live_handler
-
-    def test_deployed_mode(self, monkeypatch, tmp_path):
+        handler = start.call_args[0][0]["handler"]
+        # later environment changes and job metadata cannot replace startup bindings.
+        monkeypatch.setenv("RUNPOD_MOUNTS", "[]")
+        request = {
+            "function_name": "mounted",
+            "function_code": (
+                "from runpod.apps.volume import NetworkVolume\n"
+                "imported = str(NetworkVolume('data').path)\n"
+                "async def mounted():\n"
+                "    return [imported, str(NetworkVolume('nv-1').path)]\n"
+            ),
+            "serialization_format": "json",
+            "mounts": [
+                {"kind": "network", "reference": "data", "id": "nv-1", "path": "/wrong"}
+            ],
+        }
+        chunks = [chunk async for chunk in handler({"input": request})]
+        assert chunks[0]["success"] is True
+        assert chunks[0]["json_result"] == ["/data", "/data"]
+
+    async def test_deployed_mode(self, monkeypatch, tmp_path):
+        monkeypatch.setenv(
+            "RUNPOD_MOUNTS",
+            json.dumps(
+                [
+                    {
+                        "kind": "global",
+                        "reference": "gv-1",
+                        "id": "gv-1",
+                        "path": "/shared",
+                    }
+                ]
+            ),
+        )
         module_src = (
             "import runpod\n"
+            "from runpod.apps.volume import GlobalVolume\n"
+            "imported = str(GlobalVolume('gv-1').path)\n"
             "app = runpod.App('t')\n"
             "@app.queue(name='greet', gpu='4090')\n"
-            "def greet(name: str):\n"
-            "    return f'hi {name}'\n"
+            "async def greet(name: str):\n"
+            "    yield {'name': name, 'imported': imported}\n"
+            "    yield {'mounted': str(GlobalVolume('gv-1').path)}\n"
         )
         (tmp_path / "user_mod_c.py").write_text(module_src)
         (tmp_path / worker.MANIFEST_NAME).write_text(
@@ -218,7 +274,21 @@ def test_deployed_mode(self, monkeypatch, tmp_path):
         with patch("runpod.serverless.start") as start:
             worker.main()
         handler = start.call_args[0][0]["handler"]
-        assert inspect.iscoroutinefunction(handler)
+        chunks = [chunk async for chunk in handler({"input": {"name": "reader"}})]
+        assert chunks == [
+            {"name": "reader", "imported": "/shared"},
+            {"mounted": "/shared"},
+        ]
+
+    @pytest.mark.parametrize("raw", ["", "null"])
+    def test_invalid_mounts_stop_startup(self, monkeypatch, raw):
+        from runpod.apps.volume import VolumeError
+
+        monkeypatch.setenv("RUNPOD_MOUNTS", raw)
+        with patch("runpod.serverless.start") as start:
+            with pytest.raises((RuntimeError, VolumeError)):
+                worker.main()
+        start.assert_not_called()
 
     async def test_live_handler_plain_function(self):
         def fn(x):
diff --git a/tests/test_task_runner_http.py b/tests/test_task_runner_http.py
index 9a6dab9..0b0dff0 100644
--- a/tests/test_task_runner_http.py
+++ b/tests/test_task_runner_http.py
@@ -28,17 +28,30 @@
 
 
 @pytest.fixture()
-def server(monkeypatch):
+def server(monkeypatch, request):
+    from runpod.apps.volume import _configure_mounts
+
     monkeypatch.setattr(task_runner, "TOKEN", TOKEN)
     monkeypatch.setattr(task_runner, "_job_state", {"status": "NONE", "response": None})
+    monkeypatch.setenv("RUNPOD_MOUNTS", json.dumps(getattr(request, "param", [])))
     httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
     httpd.daemon_threads = True
-    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    ready = threading.Event()
+
+    def started_server(*_):
+        ready.set()
+        return httpd
+
+    monkeypatch.setattr(task_runner, "ThreadingHTTPServer", started_server)
+    monkeypatch.setattr(task_runner, "_watchdog", lambda: None)
+    thread = threading.Thread(target=task_runner.main, daemon=True)
     thread.start()
+    assert ready.wait(timeout=5), "task runner did not finish startup"
     yield f"http://127.0.0.1:{httpd.server_port}"
     httpd.shutdown()
     thread.join(timeout=5)
     httpd.server_close()
+    _configure_mounts([])
 
 
 def _request(url, method="GET", body=None, token=TOKEN):
@@ -118,6 +131,34 @@ def test_submit_and_result(server):
     assert body["response"]["json_result"] == "done"
 
 
+@pytest.mark.parametrize(
+    "server",
+    [[{"kind": "network", "reference": "data", "id": "nv-1", "path": "/data"}]],
+    indirect=True,
+)
+def test_mounts_survive_threaded_execution(server, monkeypatch):
+    monkeypatch.setenv("RUNPOD_MOUNTS", "[]")
+    status, body = _request(
+        f"{server}/execute",
+        method="POST",
+        body={
+            "function_name": "mounted",
+            "function_code": (
+                "from runpod.apps.volume import NetworkVolume\n"
+                "imported = str(NetworkVolume('data').path)\n"
+                "async def mounted():\n"
+                "    yield imported\n"
+                "    yield str(NetworkVolume('nv-1').path)\n"
+            ),
+            "serialization_format": "json",
+            "mounts": [],
+        },
+    )
+    assert status == 200
+    assert body["success"] is True
+    assert body["json_result"] == ["/data", "/data"]
+
+
 def test_unknown_path_404(server):
     with pytest.raises(urllib.error.HTTPError) as exc_info:
         _request(f"{server}/nope")

From 9bed9ef03a4077a9b1682fd83ff3aaf9a6e48fbe Mon Sep 17 00:00:00 2001
From: zeke <40004347+KAJdev@users.noreply.github.com>
Date: Wed, 7 Oct 2026 12:38:36 -0700
Subject: [PATCH 2/6] fix: acknowledge CON-1751 task pod termination before
 exit

---
 CHANGELOG.md                          |   4 +
 README.md                             |  17 ++++
 src/runpod_sdk_runtime/task/runner.py | 130 +++++++++++++++++---------
 tests/test_task_runner_http.py        | 108 +++++++++++++--------
 4 files changed, 177 insertions(+), 82 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b6564b7..5d613b0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,3 +5,7 @@
 ### Features
 
 * Initialize resolved volume bindings before user code in queue, API, and task workers.
+
+### Bug Fixes
+
+* Require acknowledged task pod deletion before exiting; retry transient termination failures and keep idle runtimes available for watchdog recovery without interrupting active detached work.
diff --git a/README.md b/README.md
index b6f34e3..12628c5 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,23 @@ The runtime requires an Apps SDK build providing `NetworkVolume`, `GlobalVolume`
 and the worker mount-binding contract. Publish compatible SDK and runtime
 packages/images together before deploying apps with storage bindings.
 
+## Task cleanup
+
+The task watchdog deletes idle pods and pods with uncollected completed results
+after `RUNPOD_TASK_IDLE_TIMEOUT` (600 seconds by default). Active background jobs
+and inline executions are exempt, so intentional detached work can finish.
+
+Self-deletion requires the injected pod-scoped `RUNPOD_API_KEY` and
+`RUNPOD_POD_ID`. The runtime exits only after GraphQL acknowledges `podTerminate`
+without errors. Transient failures receive three bounded attempts; failed or
+unacknowledged deletion leaves the runtime alive for the next watchdog pass.
+Missing credentials and permanent errors are logged, not treated as deletion.
+Exiting or restarting a container alone does not stop pod billing.
+
+The task's `terminateAfter` backstop requires control-plane persistence and
+expiry enforcement. The idle watchdog does not impose a second deadline on
+active detached work.
+
 ## Development
 
 Install the Apps SDK checkout first, then install this project:
diff --git a/src/runpod_sdk_runtime/task/runner.py b/src/runpod_sdk_runtime/task/runner.py
index 1836ae6..491355e 100644
--- a/src/runpod_sdk_runtime/task/runner.py
+++ b/src/runpod_sdk_runtime/task/runner.py
@@ -23,6 +23,9 @@
 import os
 import sys
 import threading
+import time
+import urllib.error
+import urllib.request
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 from runpod_sdk_runtime.executor import execute_request
@@ -31,13 +34,12 @@
 PORT = int(os.environ.get("RUNPOD_TASK_PORT", "8080"))
 TOKEN = os.environ.get("RUNPOD_TASK_TOKEN", "")
 
-# watchdog: self-terminate when the client is clearly gone. RUNNING
-# jobs are never killed (terminateAfter is the runaway backstop);
-# state NONE means the client died before submitting, while DONE means
-# the result sat uncollected. normal flows poll every ~2s, so these never fire
-# for a live client.
+# the watchdog reaps unsubmitted jobs and uncollected results after contact stops.
+# running jobs are exempt; absolute deadlines require control-plane enforcement.
 IDLE_TIMEOUT = float(os.environ.get("RUNPOD_TASK_IDLE_TIMEOUT", "600"))
 WATCHDOG_INTERVAL = 15.0
+TERMINATE_ATTEMPTS = 3
+TERMINATE_TIMEOUT = 5.0
 
 # single background job slot for /submit + /result
 _job_lock = threading.Lock()
@@ -64,42 +66,79 @@ def _should_self_terminate(status, last_contact, now, idle_timeout, inline=0):
 
 
 def _terminate_self():
-    """terminate this pod via the injected pod-scoped api key.
-
-    every pod carries a RUNPOD_API_KEY scoped to itself; podTerminate
-    with it removes the pod entirely. exiting the process is the
-    fallback (stops the workload; terminateAfter finishes the job).
-    """
-    import json as _json
-    import urllib.request
-
+    """exit only after the control plane acknowledges pod deletion."""
     pod_id = os.environ.get("RUNPOD_POD_ID")
     api_key = os.environ.get("RUNPOD_API_KEY")
-    if pod_id and api_key:
+    if not pod_id or not api_key:
+        sys.stderr.write(
+            "[task-runner] cannot delete pod: missing RUNPOD_POD_ID or RUNPOD_API_KEY\n"
+        )
+        return False
+    api_base = os.environ.get("RUNPOD_API_BASE_URL", "https://api.runpod.io")
+    payload = json.dumps(
+        {
+            "query": (
+                "mutation podTerminate($input: PodTerminateInput!) "
+                "{ podTerminate(input: $input) }"
+            ),
+            "variables": {"input": {"podId": pod_id}},
+        }
+    ).encode()
+    request = urllib.request.Request(
+        f"{api_base.rstrip('/')}/graphql",
+        data=payload,
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {api_key}",
+        },
+    )
+    for attempt in range(TERMINATE_ATTEMPTS):
+        retryable = False
         try:
-            api_base = os.environ.get("RUNPOD_API_BASE_URL", "https://api.runpod.io")
-            payload = _json.dumps(
-                {
-                    "query": (
-                        "mutation podTerminate($input: PodTerminateInput!) "
-                        "{ podTerminate(input: $input) }"
-                    ),
-                    "variables": {"input": {"podId": pod_id}},
-                }
-            ).encode()
-            request = urllib.request.Request(
-                f"{api_base}/graphql",
-                data=payload,
-                headers={
-                    "Content-Type": "application/json",
-                    "Authorization": f"Bearer {api_key}",
-                },
+            with urllib.request.urlopen(request, timeout=TERMINATE_TIMEOUT) as response:  # noqa: S310
+                body = json.load(response)
+            errors = body.get("errors") if isinstance(body, dict) else None
+            data = body.get("data") if isinstance(body, dict) else None
+            if (
+                not errors
+                and isinstance(data, dict)
+                and "podTerminate" in data
+                and data["podTerminate"] is None
+            ):
+                sys.stderr.write(f"[task-runner] pod {pod_id} deletion acknowledged\n")
+                os._exit(0)
+                return True
+            if isinstance(errors, list) and errors:
+                retryable = all(
+                    isinstance(error, dict)
+                    and isinstance(error.get("extensions"), dict)
+                    and error["extensions"].get("code")
+                    in {
+                        "INTERNAL_SERVER_ERROR",
+                        "SERVICE_UNAVAILABLE",
+                        "TOO_MANY_REQUESTS",
+                        "TIMEOUT",
+                    }
+                    for error in errors
+                )
+            failure = "graphql errors" if errors else "missing deletion acknowledgment"
+        except urllib.error.HTTPError as exc:
+            retryable = exc.code == 429 or 500 <= exc.code < 600
+            failure = f"http {exc.code}"
+            exc.close()
+        except (urllib.error.URLError, OSError) as exc:
+            retryable = True
+            failure = str(exc)
+        except (ValueError, TypeError) as exc:
+            failure = f"invalid deletion response: {exc}"
+        if not retryable or attempt == TERMINATE_ATTEMPTS - 1:
+            sys.stderr.write(
+                f"[task-runner] pod {pod_id} deletion failed: {failure}; "
+                "keeping runtime alive for watchdog retry\n"
             )
-            urllib.request.urlopen(request, timeout=30)  # noqa: S310
-            sys.stderr.write("[task-runner] self-terminated (abandoned)\n")
-        except Exception:  # noqa: BLE001 - fall through to process exit
-            pass
-    os._exit(0)
+            return False
+        time.sleep(0.5 * (2**attempt))
+    return False
 
 
 def _watchdog():
@@ -107,14 +146,15 @@ def _watchdog():
 
     while True:
         time.sleep(WATCHDOG_INTERVAL)
-        if _should_self_terminate(
-            _job_state["status"],
-            _last_contact["ts"],
-            time.time(),
-            IDLE_TIMEOUT,
-            inline=_inline_executions["count"],
-        ):
-            _terminate_self()
+        with _job_lock:
+            if _should_self_terminate(
+                _job_state["status"],
+                _last_contact["ts"],
+                time.time(),
+                IDLE_TIMEOUT,
+                inline=_inline_executions["count"],
+            ):
+                _terminate_self()
 
 
 class Handler(BaseHTTPRequestHandler):
diff --git a/tests/test_task_runner_http.py b/tests/test_task_runner_http.py
index 0b0dff0..e10978a 100644
--- a/tests/test_task_runner_http.py
+++ b/tests/test_task_runner_http.py
@@ -192,6 +192,9 @@ def test_no_contact_recorded_yet(self):
 
         assert not _should_self_terminate("NONE", None, 10_000, 600)
 
+    def test_inline_execution_never_killed(self):
+        assert not task_runner._should_self_terminate("NONE", 0, 10_000, 600, inline=1)
+
     def test_authed_requests_touch_contact(self, server, monkeypatch):
         monkeypatch.setattr(task_runner, "_last_contact", {"ts": None})
         _request(f"{server}/result")
@@ -199,49 +202,80 @@ def test_authed_requests_touch_contact(self, server, monkeypatch):
 
 
 class TestSelfTermination:
-    def test_terminate_self_calls_graphql_then_exits(self, monkeypatch):
-        import contextlib
-        import io as _io
-
-        calls = []
+    def _api(self, monkeypatch, responses):
+        import io
 
-        @contextlib.contextmanager
-        def _response():
-            yield _io.BytesIO(b"{}")
+        remaining = iter(responses)
+        exits = []
+        attempts = []
 
-        def fake_urlopen(req, timeout=None):
-            calls.append((req.full_url, req.data))
-            return _response()
+        def request(req, timeout=None):
+            attempts.append(req)
+            response = next(remaining)
+            if isinstance(response, Exception):
+                raise response
+            return io.BytesIO(json.dumps(response).encode())
 
-        exits = []
         monkeypatch.setenv("RUNPOD_POD_ID", "pod-1")
         monkeypatch.setenv("RUNPOD_API_KEY", "pod-scoped-key")
-        monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
-        monkeypatch.setattr(task_runner.os, "_exit", lambda code: exits.append(code))
-        task_runner._terminate_self()
-
+        monkeypatch.setattr("urllib.request.urlopen", request)
+        monkeypatch.setattr(task_runner.os, "_exit", exits.append)
+        monkeypatch.setattr(task_runner.time, "sleep", lambda delay: None)
+        return exits, attempts
+
+    def test_void_acknowledgment_allows_exit(self, monkeypatch):
+        exits, _ = self._api(monkeypatch, [{"data": {"podTerminate": None}}])
+        assert task_runner._terminate_self() is True
         assert exits == [0]
-        url, body = calls[0]
-        assert url.endswith("/graphql")
-        assert b"podTerminate" in body
-        assert b"pod-1" in body
 
-    def test_terminate_self_exits_without_credentials(self, monkeypatch):
-        exits = []
-        monkeypatch.delenv("RUNPOD_POD_ID", raising=False)
-        monkeypatch.delenv("RUNPOD_API_KEY", raising=False)
-        monkeypatch.setattr(task_runner.os, "_exit", lambda code: exits.append(code))
-        task_runner._terminate_self()
+    @pytest.mark.parametrize(
+        "body",
+        [
+            {},
+            {"data": {}},
+            {"data": {"podTerminate": False}},
+            {"errors": [{"message": "forbidden"}], "data": {"podTerminate": None}},
+        ],
+    )
+    def test_unacknowledged_deletion_keeps_runtime_alive(self, monkeypatch, body):
+        exits, attempts = self._api(monkeypatch, [body])
+        assert task_runner._terminate_self() is False
+        assert not exits
+        assert len(attempts) == 1
+
+    @pytest.mark.parametrize("missing", ["RUNPOD_POD_ID", "RUNPOD_API_KEY"])
+    def test_missing_credentials_do_not_exit(self, monkeypatch, missing):
+        exits, attempts = self._api(monkeypatch, [])
+        monkeypatch.delenv(missing)
+        assert task_runner._terminate_self() is False
+        assert not attempts
+        assert not exits
+
+    @pytest.mark.parametrize(
+        "failure",
+        [
+            urllib.error.HTTPError("https://api/graphql", 429, "slow down", {}, None),
+            urllib.error.HTTPError("https://api/graphql", 503, "unavailable", {}, None),
+            urllib.error.URLError("connection reset"),
+            {"errors": [{"extensions": {"code": "INTERNAL_SERVER_ERROR"}}]},
+        ],
+    )
+    def test_transient_failure_recovers(self, monkeypatch, failure):
+        exits, _ = self._api(monkeypatch, [failure, {"data": {"podTerminate": None}}])
+        assert task_runner._terminate_self() is True
         assert exits == [0]
 
-    def test_terminate_self_survives_api_failure(self, monkeypatch):
-        def fail(req, timeout=None):
-            raise OSError("network down")
-
-        exits = []
-        monkeypatch.setenv("RUNPOD_POD_ID", "pod-1")
-        monkeypatch.setenv("RUNPOD_API_KEY", "k")
-        monkeypatch.setattr("urllib.request.urlopen", fail)
-        monkeypatch.setattr(task_runner.os, "_exit", lambda code: exits.append(code))
-        task_runner._terminate_self()
-        assert exits == [0]
+    def test_retry_exhaustion_keeps_runtime_alive(self, monkeypatch):
+        exits, attempts = self._api(
+            monkeypatch, [OSError("network down")] * task_runner.TERMINATE_ATTEMPTS
+        )
+        assert task_runner._terminate_self() is False
+        assert len(attempts) == task_runner.TERMINATE_ATTEMPTS
+        assert not exits
+
+    def test_auth_failure_does_not_retry_or_exit(self, monkeypatch):
+        error = urllib.error.HTTPError("https://api/graphql", 401, "denied", {}, None)
+        exits, attempts = self._api(monkeypatch, [error])
+        assert task_runner._terminate_self() is False
+        assert len(attempts) == 1
+        assert not exits

From cf7590f0ac598240e9afcb8b28439af651ea7db4 Mon Sep 17 00:00:00 2001
From: zeke <40004347+KAJdev@users.noreply.github.com>
Date: Wed, 7 Oct 2026 12:44:32 -0700
Subject: [PATCH 3/6] ci: validate runtime compatibility against SDK main

---
 .github/workflows/ci.yml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7d9f0bc..5d93a9c 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -24,7 +24,7 @@ jobs:
       - uses: actions/checkout@v7
         with:
           repository: runpod/runpod-python
-          ref: feat/apps-sdk
+          ref: main
           path: runpod-python
 
       - uses: actions/setup-python@v6
@@ -81,7 +81,7 @@ jobs:
           tags: runpod-sdk-runtime:test
           build-args: |
             BASE_IMAGE=python:3.12-slim
-            RUNPOD_PACKAGE_SPEC=https://github.com/runpod/runpod-python/archive/refs/heads/feat/apps-sdk.tar.gz
+            RUNPOD_PACKAGE_SPEC=https://github.com/runpod/runpod-python/archive/refs/heads/main.tar.gz
           cache-from: type=gha,scope=queue-test
           cache-to: type=gha,mode=max,scope=queue-test
 

From dcc5bee9c91c8c69a5ebe605a026080592beb9e5 Mon Sep 17 00:00:00 2001
From: zeke <40004347+KAJdev@users.noreply.github.com>
Date: Wed, 7 Oct 2026 13:22:36 -0700
Subject: [PATCH 4/6] fix: enforce absolute CON-1751 task deadlines

---
 CHANGELOG.md                          |   2 +-
 README.md                             |  44 ++++-----
 src/runpod_sdk_runtime/task/runner.py |  19 ++--
 tests/test_api_server.py              |  65 +++----------
 tests/test_queue_worker.py            |  89 +++---------------
 tests/test_task_runner_http.py        | 130 +++++++++-----------------
 6 files changed, 102 insertions(+), 247 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5d613b0..d357b65 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,4 +8,4 @@
 
 ### Bug Fixes
 
-* Require acknowledged task pod deletion before exiting; retry transient termination failures and keep idle runtimes available for watchdog recovery without interrupting active detached work.
+* Enforce the SDK's absolute task deadline during active work; exit only after acknowledged pod deletion and retry failed cleanup.
diff --git a/README.md b/README.md
index 12628c5..bc16fac 100644
--- a/README.md
+++ b/README.md
@@ -40,36 +40,26 @@ python -m runpod_sdk_runtime.task.runner
 
 ## Storage bindings
 
-Apps provisioning supplies `RUNPOD_MOUNTS` as a JSON array of resolved bindings.
-Each entry contains `kind` (`network` or `global`), the declared `reference`, the
-resolved volume `id`, and its absolute `path`.
-
-Queue, API, and task workers install these bindings before importing user modules
-or running initialization hooks. The SDK's `volume.path` accessor uses this
-worker-level map across synchronous, asynchronous, streaming, and threaded code.
-An absent variable represents an empty mount map; malformed configuration fails
-startup. Request payloads do not configure filesystem mounts.
-
-The runtime requires an Apps SDK build providing `NetworkVolume`, `GlobalVolume`,
-and the worker mount-binding contract. Publish compatible SDK and runtime
-packages/images together before deploying apps with storage bindings.
+Apps provisioning supplies `RUNPOD_MOUNTS`, a JSON array of bindings with `kind`
+(`network` or `global`), declared `reference`, resolved volume `id`, and absolute
+`path`. Queue, API, and task workers install these before importing user code.
+The compatible Apps SDK's `volume.path` uses this worker-level map; request
+payloads cannot replace it. Missing configuration means no mounts; malformed
+configuration fails startup.
 
 ## Task cleanup
 
-The task watchdog deletes idle pods and pods with uncollected completed results
-after `RUNPOD_TASK_IDLE_TIMEOUT` (600 seconds by default). Active background jobs
-and inline executions are exempt, so intentional detached work can finish.
-
-Self-deletion requires the injected pod-scoped `RUNPOD_API_KEY` and
-`RUNPOD_POD_ID`. The runtime exits only after GraphQL acknowledges `podTerminate`
-without errors. Transient failures receive three bounded attempts; failed or
-unacknowledged deletion leaves the runtime alive for the next watchdog pass.
-Missing credentials and permanent errors are logged, not treated as deletion.
-Exiting or restarting a container alone does not stop pod billing.
-
-The task's `terminateAfter` backstop requires control-plane persistence and
-expiry enforcement. The idle watchdog does not impose a second deadline on
-active detached work.
+The task watchdog checks `RUNPOD_TASK_DEADLINE` every 15 seconds. This absolute
+timestamp in Unix epoch seconds is set by the SDK and survives container restarts.
+Once reached, the runtime requests pod deletion even during active work or recent client
+contact. Without a deadline, only idle cleanup applies:
+`RUNPOD_TASK_IDLE_TIMEOUT` (600 seconds by default) reaps abandoned tasks and
+uncollected results, but exempts active background jobs and inline executions.
+
+Self-deletion requires `RUNPOD_API_KEY` and `RUNPOD_POD_ID`. The runtime exits
+only after GraphQL acknowledges `podTerminate` without errors. Transient
+failures receive three bounded attempts; any failed deletion leaves the runtime
+alive for the next watchdog pass. Container exit alone does not stop pod billing.
 
 ## Development
 
diff --git a/src/runpod_sdk_runtime/task/runner.py b/src/runpod_sdk_runtime/task/runner.py
index 491355e..f8f01f9 100644
--- a/src/runpod_sdk_runtime/task/runner.py
+++ b/src/runpod_sdk_runtime/task/runner.py
@@ -34,9 +34,13 @@
 PORT = int(os.environ.get("RUNPOD_TASK_PORT", "8080"))
 TOKEN = os.environ.get("RUNPOD_TASK_TOKEN", "")
 
-# the watchdog reaps unsubmitted jobs and uncollected results after contact stops.
-# running jobs are exempt; absolute deadlines require control-plane enforcement.
+# active work is exempt from idle cleanup, but never from the task deadline.
 IDLE_TIMEOUT = float(os.environ.get("RUNPOD_TASK_IDLE_TIMEOUT", "600"))
+TASK_DEADLINE = (
+    float(os.environ["RUNPOD_TASK_DEADLINE"])
+    if "RUNPOD_TASK_DEADLINE" in os.environ
+    else None
+)
 WATCHDOG_INTERVAL = 15.0
 TERMINATE_ATTEMPTS = 3
 TERMINATE_TIMEOUT = 5.0
@@ -45,8 +49,7 @@
 _job_lock = threading.Lock()
 _job_state = {"status": "NONE", "response": None}
 _last_contact = {"ts": None}  # set at server start
-# inline /execute requests in flight; the watchdog must not terminate
-# the pod while one runs (long executes outlive the idle timeout)
+# inline requests are protected from idle cleanup while running.
 _inline_executions = {"count": 0}
 
 
@@ -56,8 +59,11 @@ def _touch_contact():
     _last_contact["ts"] = time.time()
 
 
-def _should_self_terminate(status, last_contact, now, idle_timeout, inline=0):
-    """the watchdog decision: kill only provably-abandoned pods."""
+def _should_self_terminate(
+    status, last_contact, now, idle_timeout, inline=0, deadline=None
+):
+    if deadline is not None and now >= deadline:
+        return True
     if status == "RUNNING" or inline > 0:
         return False
     if last_contact is None:
@@ -153,6 +159,7 @@ def _watchdog():
                 time.time(),
                 IDLE_TIMEOUT,
                 inline=_inline_executions["count"],
+                deadline=TASK_DEADLINE,
             ):
                 _terminate_self()
 
diff --git a/tests/test_api_server.py b/tests/test_api_server.py
index eaa6317..ebe29fc 100644
--- a/tests/test_api_server.py
+++ b/tests/test_api_server.py
@@ -25,11 +25,18 @@ def clean_registry(monkeypatch):
 
 
 def _write_project(tmp_path, monkeypatch, module="main_cls"):
+    monkeypatch.setenv(
+        "RUNPOD_MOUNTS",
+        '[{"kind":"network","reference":"data","id":"nv-1","path":"/data"}]',
+    )
     (tmp_path / f"{module}.py").write_text(
         textwrap.dedent(
             """
             import runpod
             from runpod import App, init, get, post
+            from runpod.apps.volume import NetworkVolume
+            volume = NetworkVolume("data")
+            imported = str(volume.path)
 
             app = App("api-test")
 
@@ -38,10 +45,14 @@ class Inference:
                 @init
                 def setup(self):
                     self.model = "loaded"
+                    self.mounted = [imported, str(volume.path)]
 
                 @get("/health")
                 def health(self):
-                    return {"model": self.model}
+                    return {
+                        "model": self.model,
+                        "mounts": self.mounted + [str(volume.path)],
+                    }
 
                 @post("/generate")
                 async def generate(self, body: dict):
@@ -84,7 +95,7 @@ def test_routes_and_init(self, tmp_path, monkeypatch):
             assert response.json() == {"status": "healthy"}
 
             response = client.get("/health")
-            assert response.json() == {"model": "loaded"}
+            assert response.json() == {"model": "loaded", "mounts": ["/data"] * 3}
 
             response = client.post("/generate", json={"prompt": "hi"})
             assert response.json() == {
@@ -259,53 +270,3 @@ def test_resync_same_source_keeps_state(self):
         # identical source: no rebuild, state survives
         client.post("/_runpod/sync", json={"source": source, "resource": "S"})
         assert client.post("/add", json={"x": 3}).json() == {"total": 5}
-
-
-@pytest.mark.parametrize("deployed", [False, True])
-def test_mounts_before_module_init_and_routes(monkeypatch, tmp_path, deployed):
-    monkeypatch.setenv(
-        "RUNPOD_MOUNTS",
-        json.dumps(
-            [{"kind": "network", "reference": "data", "id": "nv-1", "path": "/data"}]
-        ),
-    )
-    source = textwrap.dedent(
-        """
-        from runpod import App, get, init
-        from runpod.apps.volume import NetworkVolume
-
-        imported = str(NetworkVolume("data").path)
-        app = App("mounts-api")
-
-        @app.api(name="mounted", cpu="cpu3c-1-2")
-        class Mounted:
-            @init
-            async def setup(self):
-                self.initialized = str(NetworkVolume("nv-1").path)
-
-            @get("/mounted")
-            def mounted(self):
-                return [imported, self.initialized, str(NetworkVolume("data").path)]
-        """
-    )
-    if deployed:
-        (tmp_path / "mounts_api.py").write_text(source)
-        (tmp_path / server.MANIFEST_NAME).write_text(
-            json.dumps({"resources": [{"name": "mounted", "module": "mounts_api"}]})
-        )
-        monkeypatch.setattr(server, "APP_DIR", str(tmp_path))
-        monkeypatch.setenv("FLASH_RESOURCE_NAME", "mounted")
-    else:
-        monkeypatch.delenv("FLASH_RESOURCE_NAME", raising=False)
-        monkeypatch.delenv("RUNPOD_RESOURCE_NAME", raising=False)
-    app = server.build_app()
-    # lifespan, live sync, and threaded routes must share the startup bindings.
-    monkeypatch.setenv("RUNPOD_MOUNTS", "[]")
-    with TestClient(app) as client:
-        if not deployed:
-            response = client.post(
-                "/_runpod/sync",
-                json={"source": source, "resource": "mounted", "mounts": []},
-            )
-            assert response.status_code == 200
-        assert client.get("/mounted").json() == ["/data", "/data", "/data"]
diff --git a/tests/test_queue_worker.py b/tests/test_queue_worker.py
index c6f173b..9e66f80 100644
--- a/tests/test_queue_worker.py
+++ b/tests/test_queue_worker.py
@@ -202,57 +202,10 @@ def gen_handler(job):
 
 
 class TestMain:
-    async def test_live_mode(self, monkeypatch):
-        monkeypatch.delenv("FLASH_RESOURCE_NAME", raising=False)
-        monkeypatch.delenv("RUNPOD_RESOURCE_NAME", raising=False)
-        monkeypatch.setenv(
-            "RUNPOD_MOUNTS",
-            json.dumps(
-                [
-                    {
-                        "kind": "network",
-                        "reference": "data",
-                        "id": "nv-1",
-                        "path": "/data",
-                    }
-                ]
-            ),
-        )
-        with patch("runpod.serverless.start") as start:
-            worker.main()
-        handler = start.call_args[0][0]["handler"]
-        # later environment changes and job metadata cannot replace startup bindings.
-        monkeypatch.setenv("RUNPOD_MOUNTS", "[]")
-        request = {
-            "function_name": "mounted",
-            "function_code": (
-                "from runpod.apps.volume import NetworkVolume\n"
-                "imported = str(NetworkVolume('data').path)\n"
-                "async def mounted():\n"
-                "    return [imported, str(NetworkVolume('nv-1').path)]\n"
-            ),
-            "serialization_format": "json",
-            "mounts": [
-                {"kind": "network", "reference": "data", "id": "nv-1", "path": "/wrong"}
-            ],
-        }
-        chunks = [chunk async for chunk in handler({"input": request})]
-        assert chunks[0]["success"] is True
-        assert chunks[0]["json_result"] == ["/data", "/data"]
-
     async def test_deployed_mode(self, monkeypatch, tmp_path):
         monkeypatch.setenv(
             "RUNPOD_MOUNTS",
-            json.dumps(
-                [
-                    {
-                        "kind": "global",
-                        "reference": "gv-1",
-                        "id": "gv-1",
-                        "path": "/shared",
-                    }
-                ]
-            ),
+            '[{"kind":"global","reference":"gv-1","id":"gv-1","path":"/shared"}]',
         )
         module_src = (
             "import runpod\n"
@@ -280,33 +233,21 @@ async def test_deployed_mode(self, monkeypatch, tmp_path):
             {"mounted": "/shared"},
         ]
 
-    @pytest.mark.parametrize("raw", ["", "null"])
-    def test_invalid_mounts_stop_startup(self, monkeypatch, raw):
-        from runpod.apps.volume import VolumeError
-
-        monkeypatch.setenv("RUNPOD_MOUNTS", raw)
+    async def test_live_mode(self, monkeypatch):
+        monkeypatch.delenv("FLASH_RESOURCE_NAME", raising=False)
+        monkeypatch.delenv("RUNPOD_RESOURCE_NAME", raising=False)
         with patch("runpod.serverless.start") as start:
-            with pytest.raises((RuntimeError, VolumeError)):
-                worker.main()
-        start.assert_not_called()
-
-    async def test_live_handler_plain_function(self):
-        def fn(x):
-            return {"ok": x}
-
-        with (
-            patch(
-                "runpod_sdk_runtime.executor.resolve_request",
-                return_value=((fn, [1], {}), None),
-            ),
-            patch(
-                "runpod_sdk_runtime.executor.execute_request",
-                return_value={"success": True, "json_result": {"ok": 1}},
-            ) as execute,
-        ):
-            chunks = [c async for c in worker._live_handler({"input": {"foo": 1}})]
-        assert chunks == [{"success": True, "json_result": {"ok": 1}}]
-        execute.assert_called_once_with({"foo": 1})
+            worker.main()
+        handler = start.call_args[0][0]["handler"]
+        request = {
+            "function_name": "echo",
+            "function_code": "def echo(x):\n    return {'ok': x}",
+            "args": [1],
+            "serialization_format": "json",
+        }
+        chunks = [chunk async for chunk in handler({"input": request})]
+        assert chunks[0]["success"] is True
+        assert chunks[0]["json_result"] == {"ok": 1}
 
     async def test_live_handler_resolve_error(self):
         with patch(
diff --git a/tests/test_task_runner_http.py b/tests/test_task_runner_http.py
index e10978a..778388b 100644
--- a/tests/test_task_runner_http.py
+++ b/tests/test_task_runner_http.py
@@ -28,12 +28,15 @@
 
 
 @pytest.fixture()
-def server(monkeypatch, request):
+def server(monkeypatch):
     from runpod.apps.volume import _configure_mounts
 
     monkeypatch.setattr(task_runner, "TOKEN", TOKEN)
     monkeypatch.setattr(task_runner, "_job_state", {"status": "NONE", "response": None})
-    monkeypatch.setenv("RUNPOD_MOUNTS", json.dumps(getattr(request, "param", [])))
+    monkeypatch.setenv(
+        "RUNPOD_MOUNTS",
+        '[{"kind":"network","reference":"data","id":"nv-1","path":"/data"}]',
+    )
     httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
     httpd.daemon_threads = True
     ready = threading.Event()
@@ -46,7 +49,7 @@ def started_server(*_):
     monkeypatch.setattr(task_runner, "_watchdog", lambda: None)
     thread = threading.Thread(target=task_runner.main, daemon=True)
     thread.start()
-    assert ready.wait(timeout=5), "task runner did not finish startup"
+    assert ready.wait(timeout=5)
     yield f"http://127.0.0.1:{httpd.server_port}"
     httpd.shutdown()
     thread.join(timeout=5)
@@ -131,11 +134,6 @@ def test_submit_and_result(server):
     assert body["response"]["json_result"] == "done"
 
 
-@pytest.mark.parametrize(
-    "server",
-    [[{"kind": "network", "reference": "data", "id": "nv-1", "path": "/data"}]],
-    indirect=True,
-)
 def test_mounts_survive_threaded_execution(server, monkeypatch):
     monkeypatch.setenv("RUNPOD_MOUNTS", "[]")
     status, body = _request(
@@ -166,7 +164,7 @@ def test_unknown_path_404(server):
 
 
 class TestWatchdog:
-    def test_running_job_never_killed(self):
+    def test_running_job_without_deadline(self):
         from runpod_sdk_runtime.task.runner import _should_self_terminate
 
         assert not _should_self_terminate("RUNNING", 0, 10_000, 600)
@@ -192,90 +190,48 @@ def test_no_contact_recorded_yet(self):
 
         assert not _should_self_terminate("NONE", None, 10_000, 600)
 
-    def test_inline_execution_never_killed(self):
+    def test_inline_execution_without_deadline(self):
         assert not task_runner._should_self_terminate("NONE", 0, 10_000, 600, inline=1)
 
+    def test_deadline_overrides_active_work_and_fresh_contact(self):
+        assert not task_runner._should_self_terminate(
+            "RUNNING", 999, 999, 600, deadline=1000
+        )
+        assert task_runner._should_self_terminate(
+            "RUNNING", 1000, 1000, 600, deadline=1000
+        )
+        assert task_runner._should_self_terminate(
+            "NONE", 1000, 1000, 600, inline=1, deadline=1000
+        )
+
     def test_authed_requests_touch_contact(self, server, monkeypatch):
         monkeypatch.setattr(task_runner, "_last_contact", {"ts": None})
         _request(f"{server}/result")
         assert task_runner._last_contact["ts"] is not None
 
 
-class TestSelfTermination:
-    def _api(self, monkeypatch, responses):
-        import io
-
-        remaining = iter(responses)
-        exits = []
-        attempts = []
-
-        def request(req, timeout=None):
-            attempts.append(req)
-            response = next(remaining)
-            if isinstance(response, Exception):
-                raise response
-            return io.BytesIO(json.dumps(response).encode())
-
-        monkeypatch.setenv("RUNPOD_POD_ID", "pod-1")
-        monkeypatch.setenv("RUNPOD_API_KEY", "pod-scoped-key")
-        monkeypatch.setattr("urllib.request.urlopen", request)
-        monkeypatch.setattr(task_runner.os, "_exit", exits.append)
-        monkeypatch.setattr(task_runner.time, "sleep", lambda delay: None)
-        return exits, attempts
-
-    def test_void_acknowledgment_allows_exit(self, monkeypatch):
-        exits, _ = self._api(monkeypatch, [{"data": {"podTerminate": None}}])
-        assert task_runner._terminate_self() is True
-        assert exits == [0]
-
-    @pytest.mark.parametrize(
-        "body",
-        [
-            {},
-            {"data": {}},
-            {"data": {"podTerminate": False}},
-            {"errors": [{"message": "forbidden"}], "data": {"podTerminate": None}},
-        ],
+def test_deletion_requires_acknowledgment_and_recovers(monkeypatch):
+    import io
+    from unittest.mock import Mock
+
+    responses = [
+        io.BytesIO(b'{"errors":[{"message":"denied"}],"data":{"podTerminate":null}}'),
+        *[OSError("network down")] * task_runner.TERMINATE_ATTEMPTS,
+        urllib.error.URLError("connection reset"),
+        io.BytesIO(b'{"data":{"podTerminate":null}}'),
+    ]
+    exits = []
+    monkeypatch.setenv("RUNPOD_POD_ID", "pod-1")
+    monkeypatch.setenv("RUNPOD_API_KEY", "pod-scoped-key")
+    monkeypatch.setattr(
+        task_runner.urllib.request, "urlopen", Mock(side_effect=responses)
     )
-    def test_unacknowledged_deletion_keeps_runtime_alive(self, monkeypatch, body):
-        exits, attempts = self._api(monkeypatch, [body])
-        assert task_runner._terminate_self() is False
-        assert not exits
-        assert len(attempts) == 1
-
-    @pytest.mark.parametrize("missing", ["RUNPOD_POD_ID", "RUNPOD_API_KEY"])
-    def test_missing_credentials_do_not_exit(self, monkeypatch, missing):
-        exits, attempts = self._api(monkeypatch, [])
-        monkeypatch.delenv(missing)
-        assert task_runner._terminate_self() is False
-        assert not attempts
-        assert not exits
-
-    @pytest.mark.parametrize(
-        "failure",
-        [
-            urllib.error.HTTPError("https://api/graphql", 429, "slow down", {}, None),
-            urllib.error.HTTPError("https://api/graphql", 503, "unavailable", {}, None),
-            urllib.error.URLError("connection reset"),
-            {"errors": [{"extensions": {"code": "INTERNAL_SERVER_ERROR"}}]},
-        ],
-    )
-    def test_transient_failure_recovers(self, monkeypatch, failure):
-        exits, _ = self._api(monkeypatch, [failure, {"data": {"podTerminate": None}}])
-        assert task_runner._terminate_self() is True
-        assert exits == [0]
-
-    def test_retry_exhaustion_keeps_runtime_alive(self, monkeypatch):
-        exits, attempts = self._api(
-            monkeypatch, [OSError("network down")] * task_runner.TERMINATE_ATTEMPTS
-        )
-        assert task_runner._terminate_self() is False
-        assert len(attempts) == task_runner.TERMINATE_ATTEMPTS
-        assert not exits
-
-    def test_auth_failure_does_not_retry_or_exit(self, monkeypatch):
-        error = urllib.error.HTTPError("https://api/graphql", 401, "denied", {}, None)
-        exits, attempts = self._api(monkeypatch, [error])
-        assert task_runner._terminate_self() is False
-        assert len(attempts) == 1
-        assert not exits
+    monkeypatch.setattr(task_runner.os, "_exit", exits.append)
+    monkeypatch.setattr(task_runner.time, "sleep", lambda _: None)
+
+    assert task_runner._terminate_self() is False
+    assert not exits
+    assert task_runner._terminate_self() is False
+    assert not exits
+    assert task_runner._terminate_self() is True
+    assert exits == [0]

From 6c7abdf4ec8aab642755c63a95d64c2bad0b89c3 Mon Sep 17 00:00:00 2001
From: zeke <40004347+KAJdev@users.noreply.github.com>
Date: Wed, 7 Oct 2026 15:54:36 -0700
Subject: [PATCH 5/6] fix: preserve unlimited active task execution

---
 CHANGELOG.md                          |  2 +-
 README.md                             | 10 ++++------
 src/runpod_sdk_runtime/task/runner.py | 14 ++------------
 tests/test_task_runner_http.py        | 15 ++-------------
 4 files changed, 9 insertions(+), 32 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d357b65..7cade45 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,4 +8,4 @@
 
 ### Bug Fixes
 
-* Enforce the SDK's absolute task deadline during active work; exit only after acknowledged pod deletion and retry failed cleanup.
+* Keep active tasks running indefinitely; retry failed idle cleanup and exit only after acknowledged pod deletion.
diff --git a/README.md b/README.md
index bc16fac..d1e58da 100644
--- a/README.md
+++ b/README.md
@@ -49,12 +49,10 @@ configuration fails startup.
 
 ## Task cleanup
 
-The task watchdog checks `RUNPOD_TASK_DEADLINE` every 15 seconds. This absolute
-timestamp in Unix epoch seconds is set by the SDK and survives container restarts.
-Once reached, the runtime requests pod deletion even during active work or recent client
-contact. Without a deadline, only idle cleanup applies:
-`RUNPOD_TASK_IDLE_TIMEOUT` (600 seconds by default) reaps abandoned tasks and
-uncollected results, but exempts active background jobs and inline executions.
+Active background jobs and inline executions can run indefinitely, including
+after the client disconnects. The watchdog checks for idle cleanup every 15
+seconds. `RUNPOD_TASK_IDLE_TIMEOUT` (600 seconds by default) applies only to
+pods waiting for submission or holding uncollected results.
 
 Self-deletion requires `RUNPOD_API_KEY` and `RUNPOD_POD_ID`. The runtime exits
 only after GraphQL acknowledges `podTerminate` without errors. Transient
diff --git a/src/runpod_sdk_runtime/task/runner.py b/src/runpod_sdk_runtime/task/runner.py
index f8f01f9..95c8915 100644
--- a/src/runpod_sdk_runtime/task/runner.py
+++ b/src/runpod_sdk_runtime/task/runner.py
@@ -34,13 +34,8 @@
 PORT = int(os.environ.get("RUNPOD_TASK_PORT", "8080"))
 TOKEN = os.environ.get("RUNPOD_TASK_TOKEN", "")
 
-# active work is exempt from idle cleanup, but never from the task deadline.
+# active work is exempt from idle cleanup.
 IDLE_TIMEOUT = float(os.environ.get("RUNPOD_TASK_IDLE_TIMEOUT", "600"))
-TASK_DEADLINE = (
-    float(os.environ["RUNPOD_TASK_DEADLINE"])
-    if "RUNPOD_TASK_DEADLINE" in os.environ
-    else None
-)
 WATCHDOG_INTERVAL = 15.0
 TERMINATE_ATTEMPTS = 3
 TERMINATE_TIMEOUT = 5.0
@@ -59,11 +54,7 @@ def _touch_contact():
     _last_contact["ts"] = time.time()
 
 
-def _should_self_terminate(
-    status, last_contact, now, idle_timeout, inline=0, deadline=None
-):
-    if deadline is not None and now >= deadline:
-        return True
+def _should_self_terminate(status, last_contact, now, idle_timeout, inline=0):
     if status == "RUNNING" or inline > 0:
         return False
     if last_contact is None:
@@ -159,7 +150,6 @@ def _watchdog():
                 time.time(),
                 IDLE_TIMEOUT,
                 inline=_inline_executions["count"],
-                deadline=TASK_DEADLINE,
             ):
                 _terminate_self()
 
diff --git a/tests/test_task_runner_http.py b/tests/test_task_runner_http.py
index 778388b..2db1b8d 100644
--- a/tests/test_task_runner_http.py
+++ b/tests/test_task_runner_http.py
@@ -164,7 +164,7 @@ def test_unknown_path_404(server):
 
 
 class TestWatchdog:
-    def test_running_job_without_deadline(self):
+    def test_running_job_is_exempt_from_idle_cleanup(self):
         from runpod_sdk_runtime.task.runner import _should_self_terminate
 
         assert not _should_self_terminate("RUNNING", 0, 10_000, 600)
@@ -190,20 +190,9 @@ def test_no_contact_recorded_yet(self):
 
         assert not _should_self_terminate("NONE", None, 10_000, 600)
 
-    def test_inline_execution_without_deadline(self):
+    def test_inline_execution_is_exempt_from_idle_cleanup(self):
         assert not task_runner._should_self_terminate("NONE", 0, 10_000, 600, inline=1)
 
-    def test_deadline_overrides_active_work_and_fresh_contact(self):
-        assert not task_runner._should_self_terminate(
-            "RUNNING", 999, 999, 600, deadline=1000
-        )
-        assert task_runner._should_self_terminate(
-            "RUNNING", 1000, 1000, 600, deadline=1000
-        )
-        assert task_runner._should_self_terminate(
-            "NONE", 1000, 1000, 600, inline=1, deadline=1000
-        )
-
     def test_authed_requests_touch_contact(self, server, monkeypatch):
         monkeypatch.setattr(task_runner, "_last_contact", {"ts": None})
         _request(f"{server}/result")

From ccfe0b39af32483cbf0f3c9bda13077af6c0c221 Mon Sep 17 00:00:00 2001
From: zeke <40004347+KAJdev@users.noreply.github.com>
Date: Thu, 8 Oct 2026 17:14:33 -0700
Subject: [PATCH 6/6] fix: enforce task deadlines in the runtime watchdog

---
 src/runpod_sdk_runtime/task/runner.py |  62 +++++++++++--
 tests/test_task_runner_http.py        | 126 +++++++++++++++++++++++++-
 2 files changed, 179 insertions(+), 9 deletions(-)

diff --git a/src/runpod_sdk_runtime/task/runner.py b/src/runpod_sdk_runtime/task/runner.py
index 95c8915..efe9cc8 100644
--- a/src/runpod_sdk_runtime/task/runner.py
+++ b/src/runpod_sdk_runtime/task/runner.py
@@ -20,6 +20,7 @@
 """
 
 import json
+import math
 import os
 import sys
 import threading
@@ -46,12 +47,23 @@
 _last_contact = {"ts": None}  # set at server start
 # inline requests are protected from idle cleanup while running.
 _inline_executions = {"count": 0}
+_execution_deadline = None
+_watchdog_wakeup = threading.Event()
 
 
 def _touch_contact():
     import time
 
-    _last_contact["ts"] = time.time()
+    _last_contact["ts"] = time.monotonic()
+
+
+def _set_timeout(timeout):
+    global _execution_deadline
+    if timeout is not None:
+        deadline = time.monotonic() + timeout
+        if _execution_deadline is None or deadline < _execution_deadline:
+            _execution_deadline = deadline
+            _watchdog_wakeup.set()
 
 
 def _should_self_terminate(status, last_contact, now, idle_timeout, inline=0):
@@ -142,16 +154,22 @@ def _watchdog():
     import time
 
     while True:
-        time.sleep(WATCHDOG_INTERVAL)
         with _job_lock:
-            if _should_self_terminate(
+            now = time.monotonic()
+            expired = _execution_deadline is not None and now >= _execution_deadline
+            if expired or _should_self_terminate(
                 _job_state["status"],
                 _last_contact["ts"],
-                time.time(),
+                now,
                 IDLE_TIMEOUT,
                 inline=_inline_executions["count"],
             ):
                 _terminate_self()
+            delay = WATCHDOG_INTERVAL
+            if _execution_deadline is not None and _execution_deadline > now:
+                delay = min(delay, _execution_deadline - now)
+            _watchdog_wakeup.clear()
+        _watchdog_wakeup.wait(delay)
 
 
 class Handler(BaseHTTPRequestHandler):
@@ -198,21 +216,47 @@ def do_POST(self):  # noqa: N802 - BaseHTTPRequestHandler api
         if not self._authed():
             self._send(401, {"error": "unauthorized"})
             return
+        if self.path not in ("/execute", "/submit", "/timeout"):
+            self._send(404, {"error": "not found"})
+            return
+        try:
+            request = self._read_request()
+        except (ValueError, TypeError):
+            self._send(400, {"error": "invalid request"})
+            return
+        if not isinstance(request, dict):
+            self._send(400, {"error": "invalid request"})
+            return
+        timeout = request.get("timeout")
+        if timeout is not None and (
+            isinstance(timeout, bool)
+            or not isinstance(timeout, (int, float))
+            or not math.isfinite(timeout)
+            or timeout < 0
+        ):
+            self._send(400, {"error": "timeout must be a finite non-negative number"})
+            return
+        if self.path == "/timeout":
+            with _job_lock:
+                _set_timeout(timeout)
+            self._send(200, {"timeout": timeout})
+            return
         if self.path == "/execute":
             with _job_lock:
+                _set_timeout(timeout)
                 _inline_executions["count"] += 1
             try:
-                self._send(200, execute_request(self._read_request()))
+                self._send(200, execute_request(request))
             finally:
                 with _job_lock:
                     _inline_executions["count"] -= 1
             return
         if self.path == "/submit":
-            request = self._read_request()
             with _job_lock:
                 if _job_state["status"] == "RUNNING":
                     self._send(409, {"error": "a job is already running"})
                     return
+                _set_timeout(timeout)
                 _job_state["status"] = "RUNNING"
                 _job_state["response"] = None
 
@@ -223,9 +267,11 @@ def run():
                     _job_state["response"] = response
 
             threading.Thread(target=run, daemon=True).start()
-            self._send(200, {"status": "RUNNING"})
+            response = {"status": "RUNNING"}
+            if timeout is not None:
+                response["timeout"] = timeout
+            self._send(200, response)
             return
-        self._send(404, {"error": "not found"})
 
     def log_message(self, format, *args):  # noqa: A002 - stdlib signature
         # request logging is noise in container logs (dev sessions
diff --git a/tests/test_task_runner_http.py b/tests/test_task_runner_http.py
index 2db1b8d..f6dadb8 100644
--- a/tests/test_task_runner_http.py
+++ b/tests/test_task_runner_http.py
@@ -8,6 +8,7 @@
 import base64
 import json
 import threading
+import time
 import urllib.error
 import urllib.request
 from http.server import ThreadingHTTPServer
@@ -16,7 +17,7 @@
 import pytest
 
 from runpod_sdk_runtime.task import runner as task_runner
-from runpod_sdk_runtime.task.runner import Handler
+from runpod_sdk_runtime.task.runner import Handler, _watchdog
 
 # http.server's per-request sockets are collected lazily; the unraisable
 # checker flags them as ResourceWarnings non-deterministically
@@ -33,6 +34,8 @@ def server(monkeypatch):
 
     monkeypatch.setattr(task_runner, "TOKEN", TOKEN)
     monkeypatch.setattr(task_runner, "_job_state", {"status": "NONE", "response": None})
+    monkeypatch.setattr(task_runner, "_execution_deadline", None)
+    monkeypatch.setattr(task_runner, "_watchdog_wakeup", threading.Event())
     monkeypatch.setenv(
         "RUNPOD_MOUNTS",
         '[{"kind":"network","reference":"data","id":"nv-1","path":"/data"}]',
@@ -163,6 +166,127 @@ def test_unknown_path_404(server):
     assert exc_info.value.code == 404
 
 
+@pytest.mark.parametrize("path", ["submit", "execute", "timeout"])
+@pytest.mark.parametrize("timeout", [-1, True, "60", float("nan"), float("inf")])
+def test_invalid_timeout_is_rejected(server, path, timeout):
+    with pytest.raises(urllib.error.HTTPError) as exc_info:
+        _request(f"{server}/{path}", method="POST", body={"timeout": timeout})
+    assert exc_info.value.code == 400
+    assert task_runner._execution_deadline is None
+
+
+def test_timeout_requires_auth(server):
+    with pytest.raises(urllib.error.HTTPError) as exc_info:
+        _request(f"{server}/timeout", method="POST", body={"timeout": 0}, token="wrong")
+    assert exc_info.value.code == 401
+    assert task_runner._execution_deadline is None
+
+
+def test_timeout_cannot_be_extended_by_requests(server):
+    start = time.monotonic()
+    _request(f"{server}/timeout", method="POST", body={"timeout": 60})
+    deadline = task_runner._execution_deadline
+    assert start + 60 <= deadline <= time.monotonic() + 60
+    _request(f"{server}/timeout", method="POST", body={"timeout": 600})
+    _request(f"{server}/timeout", method="POST", body={"timeout": None})
+    _request(f"{server}/result")
+    assert task_runner._execution_deadline == deadline
+    _request(f"{server}/timeout", method="POST", body={"timeout": 0})
+    assert task_runner._execution_deadline < deadline
+
+
+@pytest.mark.parametrize("path", ["submit", "execute"])
+def test_execution_timeout_is_armed_before_user_code(server, monkeypatch, path):
+    deadlines = []
+    entered = threading.Event()
+
+    def execute(request):
+        deadlines.append(task_runner._execution_deadline)
+        entered.set()
+        return {"success": True, "json_result": 42}
+
+    monkeypatch.setattr(task_runner, "execute_request", execute)
+    start = time.monotonic()
+    status, body = _request(f"{server}/{path}", method="POST", body={"timeout": 60})
+    assert status == 200
+    if path == "submit":
+        assert body["timeout"] == 60
+    assert entered.wait(2)
+    assert deadlines and start + 60 <= deadlines[0] <= time.monotonic() + 60
+
+
+@pytest.mark.parametrize("mode", ["submit", "wait"])
+def test_timeout_survives_client_exit(server, monkeypatch, mode):
+    release = threading.Event()
+    finished = threading.Event()
+    terminated = threading.Event()
+
+    def execute(request):
+        try:
+            release.wait(5)
+            return {"success": True, "json_result": 42}
+        finally:
+            finished.set()
+
+    def terminate():
+        terminated.set()
+        raise SystemExit
+
+    def watchdog():
+        try:
+            _watchdog()
+        except SystemExit:
+            return
+
+    monkeypatch.setattr(task_runner, "execute_request", execute)
+    monkeypatch.setattr(task_runner, "_terminate_self", terminate)
+    thread = threading.Thread(target=watchdog, daemon=True)
+    try:
+        if mode == "submit":
+            _request(f"{server}/submit", method="POST", body={"timeout": 0.2})
+        else:
+            _request(f"{server}/submit", method="POST", body={})
+            _request(f"{server}/timeout", method="POST", body={"timeout": 0.2})
+        assert task_runner._job_state["status"] == "RUNNING"
+        thread.start()
+        assert terminated.wait(2)
+    finally:
+        release.set()
+        if thread.ident is not None:
+            thread.join(timeout=2)
+        assert finished.wait(2)
+
+
+@pytest.mark.parametrize("status,inline", [("RUNNING", 0), ("NONE", 1), ("DONE", 0)])
+def test_watchdog_enforces_deadline_despite_recent_contact(monkeypatch, status, inline):
+    monkeypatch.setattr(task_runner, "_execution_deadline", 10)
+    monkeypatch.setattr(task_runner, "_job_state", {"status": status})
+    monkeypatch.setattr(task_runner, "_inline_executions", {"count": inline})
+    monkeypatch.setattr(task_runner, "_last_contact", {"ts": 10})
+    monkeypatch.setattr(task_runner.time, "monotonic", lambda: 10)
+
+    def terminate():
+        raise SystemExit
+
+    monkeypatch.setattr(task_runner, "_terminate_self", terminate)
+    with pytest.raises(SystemExit):
+        _watchdog()
+
+
+def test_expired_deadline_retries_failed_deletion(monkeypatch):
+    from unittest.mock import Mock
+
+    monkeypatch.setattr(task_runner, "_execution_deadline", 0)
+    monkeypatch.setattr(task_runner, "_job_state", {"status": "RUNNING"})
+    monkeypatch.setattr(task_runner, "_watchdog_wakeup", Mock())
+    terminate = Mock(side_effect=[False, SystemExit])
+    monkeypatch.setattr(task_runner, "_terminate_self", terminate)
+    with pytest.raises(SystemExit):
+        _watchdog()
+    assert terminate.call_count == 2
+    assert task_runner._execution_deadline == 0
+
+
 class TestWatchdog:
     def test_running_job_is_exempt_from_idle_cleanup(self):
         from runpod_sdk_runtime.task.runner import _should_self_terminate