Merge branch 'main' into chore/infra-quick-wins

Author: hydrogen7797

CHANGELOG.md

@@ -1,5 +1,20 @@
 # Changelog
 
+## [1.10.0](https://github.com/runpod/runpod-python/compare/v1.9.1...v1.10.0) (2026-06-24)
+
+
+### Features
+
+* add agent source tracking to user-agent header ([#508](https://github.com/runpod/runpod-python/issues/508)) ([92239c8](https://github.com/runpod/runpod-python/commit/92239c817a64dada82b578e3447c1b4015afef9a))
+* add per-job stop capability to serverless worker ([#510](https://github.com/runpod/runpod-python/issues/510)) ([21c4786](https://github.com/runpod/runpod-python/commit/21c4786847bd2e5fd2ca6d15f85737203a3044a1))
+* detect all coding agents in user-agent, not just Claude Code ([#519](https://github.com/runpod/runpod-python/issues/519)) ([d14c67c](https://github.com/runpod/runpod-python/commit/d14c67c4e82963e61396b58a11dfd7046abbdde0))
+* support Python 3.10-3.14, drop EOL 3.8/3.9 ([#512](https://github.com/runpod/runpod-python/issues/512)) ([8e9d94e](https://github.com/runpod/runpod-python/commit/8e9d94e69105859c159e70fa9ae6bbf094297a2e))
+
+
+### Bug Fixes
+
+* **deps:** bump cryptography floor to >=48.0.1 for OpenSSL fix ([#516](https://github.com/runpod/runpod-python/issues/516)) ([18c6841](https://github.com/runpod/runpod-python/commit/18c6841f77388e9f48f0b82d14e32bc6a0e26f7c)), closes [#511](https://github.com/runpod/runpod-python/issues/511)
+
 ## [1.9.1](https://github.com/runpod/runpod-python/compare/v1.9.0...v1.9.1) (2026-05-05)
 
 

docs/serverless/worker.md

@@ -58,6 +58,12 @@ For more complex operations where you are downloading files or making changes to
                 return {"output": "Job completed successfully"}
         ```
 
+## Stopping Individual Jobs
+
+A worker can process more than one job concurrently. When a single request is cancelled, expires, or times out, the Runpod server signals the worker to stop just that request without affecting the worker's other in-progress jobs. The worker continuously polls a dedicated stop channel; the server is expected to hold each request open (long-poll) until a stop signal is available or the poll times out. When a signal arrives, the worker cancels the task running the matching job, so a stopped job no longer consumes worker time.
+
+No handler changes are required to support this. Async handlers that hold resources can perform cleanup by catching `asyncio.CancelledError`, but they **must re-raise** it after cleaning up. Swallowing the cancellation makes the worker report the job as completed instead of stopped.
+
 ## See Also
 
 - [Worker Fitness Checks](./worker_fitness_checks.md) - Validate your worker environment at startup

runpod/agent.py

@@ -0,0 +1,99 @@
+"""Detects which AI coding agent (if any) is driving the SDK.
+
+Detection is based on the environment variables that agent harnesses set in
+the processes they spawn. The registry mirrors Hugging Face's public
+agent-harnesses list so that traffic is attributed under the same agent
+identifiers across tools:
+https://github.com/huggingface/huggingface.js/blob/main/packages/tasks/src/agent-harnesses.ts
+"""
+
+import os
+
+# Each entry maps an agent identifier to the environment variables that
+# identify it. Detection matches if ANY of the listed variables is set to a
+# non-empty value. The list is checked in order and the first match wins;
+# order matters so that more specific signals come before broader ones they
+# can co-occur with (e.g. cowork before claude-code, cursor-cli before cursor).
+HARNESSES = [
+    ("antigravity", ["ANTIGRAVITY_AGENT"]),
+    ("augment-cli", ["AUGMENT_AGENT"]),
+    ("cline", ["CLINE_ACTIVE"]),
+    ("cowork", ["CLAUDE_CODE_IS_COWORK"]),
+    ("claude-code", ["CLAUDECODE", "CLAUDE_CODE"]),
+    ("codex", ["CODEX_SANDBOX", "CODEX_CI", "CODEX_THREAD_ID"]),
+    ("crush", ["CRUSH"]),
+    ("gemini-cli", ["GEMINI_CLI"]),
+    ("github-copilot", ["COPILOT_MODEL", "COPILOT_ALLOW_ALL", "COPILOT_GITHUB_TOKEN"]),
+    ("goose", ["GOOSE_TERMINAL"]),
+    ("hermes-agent", ["HERMES_SESSION_ID"]),
+    ("kilo-code", ["KILOCODE_FEATURE"]),
+    ("kiro", ["AGENT_CONTEXT_OUT"]),
+    ("openclaw", ["OPENCLAW_SHELL"]),
+    ("opencode", ["OPENCODE_CLIENT"]),
+    ("pi", ["PI_CODING_AGENT"]),
+    ("replit", ["REPL_ID"]),
+    ("trae", ["TRAE_AI_SHELL_ID"]),
+    ("zed", ["ZED_TERM"]),
+    ("cursor-cli", ["CURSOR_AGENT"]),
+    ("cursor", ["CURSOR_TRACE_ID"]),
+]
+
+# Generic variables any tool can set to identify itself. When set, the value is
+# sanitized and used as the agent id. Only AI_AGENT is honored: a bare AGENT is
+# too common in unrelated environments (CI runners, shell setups) and would
+# attribute traffic to arbitrary values.
+STANDARD_ENV_VARS = ["AI_AGENT"]
+
+
+def known_env_vars():
+    """Returns every environment variable the registry inspects, including the
+    standard AI_AGENT signal. Useful for tests that need to isolate detection
+    from the ambient environment.
+    """
+    variables = []
+    for _, env_vars in HARNESSES:
+        variables.extend(env_vars)
+    return variables + list(STANDARD_ENV_VARS)
+
+
+def _sanitize(value):
+    """Keeps only User-Agent-safe characters ([A-Za-z0-9._-]), capped at 64
+    characters, so an arbitrary env value cannot produce a malformed header.
+    """
+    value = value.strip()
+    safe = []
+    for char in value:
+        if len(safe) >= 64:
+            break
+        if char.isascii() and (char.isalnum() or char in "._-"):
+            safe.append(char)
+    return "".join(safe)
+
+
+def detect():
+    """Returns the identifier of the AI coding agent driving the SDK, or an
+    empty string if none is detected. Specific harness markers take priority
+    over the generic AI_AGENT signal.
+    """
+    for agent_id, env_vars in HARNESSES:
+        for env in env_vars:
+            if os.getenv(env, "").strip():
+                return agent_id
+
+    for env in STANDARD_ENV_VARS:
+        value = _sanitize(os.getenv(env, ""))
+        if value:
+            return value
+
+    return ""
+
+
+def suffix():
+    """Returns the " (via <id>)" User-Agent fragment for the detected agent, or
+    an empty string when none is detected. Centralizing the fragment here keeps
+    the tag format identical across every client's User-Agent.
+    """
+    agent_id = detect()
+    if agent_id:
+        return f" (via {agent_id})"
+    return ""

runpod/serverless/modules/rp_job.py

@@ -26,6 +26,81 @@
 job_progress = JobsProgress()
 
 
+def _job_stop_url() -> Optional[str]:
+    """
+    Prepare the URL for the worker's dedicated stop channel.
+
+    Derived from the job-take URL so it points at the same endpoint and worker,
+    preserving its query string (auth and routing params). Returns None when the
+    job-take URL is not in the expected form.
+    """
+    if "/job-take/" not in JOB_GET_URL:
+        return None
+    return JOB_GET_URL.replace("/job-take/", "/job-stop/")
+
+
+async def get_stop_signals(session: ClientSession) -> List[str]:
+    """
+    Long-poll the dedicated stop channel for request ids the worker should stop.
+
+    The server is expected to hold the request open (long-poll) until a stop
+    signal is available or the poll times out, so cancellations and timeouts
+    reach the worker without waiting for the next heartbeat.
+
+    Returns:
+        A list of request ids to stop. Empty when the poll returned no signals.
+    """
+    stop_url = _job_stop_url()
+    if not stop_url:
+        return []
+
+    async with session.get(stop_url) as response:
+        if response.status == 204:
+            return []
+
+        if response.status == 429:
+            raise TooManyRequests(
+                response.request_info,
+                response.history,
+                status=response.status,
+                message=response.reason,
+            )
+
+        response.raise_for_status()
+
+        if response.content_type != "application/json":
+            log.warn(
+                f"rp_job | get_stop_signals: unexpected content type: {response.content_type}"
+            )
+            return []
+
+        try:
+            payload = await response.json()
+        except (aiohttp.ContentTypeError, ValueError) as error:
+            log.warn(f"rp_job | get_stop_signals: failed to parse response: {error}")
+            return []
+
+        if not isinstance(payload, dict):
+            log.warn(
+                f"rp_job | get_stop_signals: unexpected payload type: {type(payload).__name__}"
+            )
+            return []
+
+        raw_ids = payload.get("jobsToStop", [])
+        if not isinstance(raw_ids, list):
+            log.warn(
+                f"rp_job | get_stop_signals: jobsToStop is not a list: {type(raw_ids).__name__}"
+            )
+            return []
+
+        job_ids = [job_id for job_id in raw_ids if isinstance(job_id, str)]
+        if len(job_ids) != len(raw_ids):
+            log.warn(
+                f"rp_job | get_stop_signals: dropped {len(raw_ids) - len(job_ids)} non-string job ids"
+            )
+        return job_ids
+
+
 def _job_get_url(batch_size: int = 1):
     """
     Prepare the URL for making a 'get' request to the serverless API (sls).

runpod/serverless/modules/rp_scale.py

@@ -10,7 +10,7 @@
 from typing import Any, Dict, Set
 
 from ...http_client import AsyncClientSession, ClientSession, TooManyRequests
-from .rp_job import get_job, handle_job
+from .rp_job import _job_stop_url, get_job, get_stop_signals, handle_job
 from .rp_logger import RunPodLogger
 from .worker_state import JobsProgress, IS_LOCAL_TEST
 
@@ -48,6 +48,13 @@ def __init__(self, config: Dict[str, Any]):
         self.config = config
         self.job_progress = JobsProgress()  # Cache the singleton instance
 
+        # maps in-progress job ids to their running tasks so individual jobs
+        # can be stopped without killing the whole worker
+        self.jobs_tasks: Dict[str, asyncio.Task] = {}
+
+        self.stop_signals_fetcher = get_stop_signals
+        self.stop_signals_fetcher_timeout = 90
+
         self.jobs_queue = asyncio.Queue(maxsize=self.current_concurrency)
 
         self.concurrency_modifier = _default_concurrency_modifier
@@ -71,6 +78,12 @@ def __init__(self, config: Dict[str, Any]):
         if jobs_handler := self.config.get("jobs_handler"):
             self.jobs_handler = jobs_handler
 
+        if stop_signals_fetcher := self.config.get("stop_signals_fetcher"):
+            self.stop_signals_fetcher = stop_signals_fetcher
+
+        if stop_signals_fetcher_timeout := self.config.get("stop_signals_fetcher_timeout"):
+            self.stop_signals_fetcher_timeout = stop_signals_fetcher_timeout
+
     async def set_scale(self):
         self.current_concurrency = self.concurrency_modifier(self.current_concurrency)
 
@@ -125,13 +138,14 @@ def handle_shutdown(self, signum, frame):
     async def run(self):
         # Create an async session that will be closed when the worker is killed.
         async with AsyncClientSession() as session:
-            # Create tasks for getting and running jobs.
+            # Create the worker's concurrent loops.
             jobtake_task = asyncio.create_task(self.get_jobs(session))
             jobrun_task = asyncio.create_task(self.run_jobs(session))
+            jobstop_task = asyncio.create_task(self.monitor_stop_signals(session))
 
-            tasks = [jobtake_task, jobrun_task]
+            tasks = [jobtake_task, jobrun_task, jobstop_task]
 
-            # Concurrently run both tasks and wait for both to finish.
+            # Run the worker's concurrent loops until shutdown.
             await asyncio.gather(*tasks)
 
     def is_alive(self):
@@ -226,9 +240,10 @@ async def run_jobs(self, session: ClientSession):
             # Fetch as many jobs as the concurrency allows
             while len(tasks) < self.current_concurrency and not self.jobs_queue.empty():
                 job = await self.jobs_queue.get()
-                # Create a new task for each job and add it to the task list
+                # Create a new task for each job and track it by job id
                 task = asyncio.create_task(self.handle_job(session, job))
                 tasks.add(task)
+                self.jobs_tasks[job["id"]] = task
 
             # Wait for any job to finish
             if tasks:
@@ -249,8 +264,79 @@ async def run_jobs(self, session: ClientSession):
                 await asyncio.sleep(0.1)
 
 
-        # Ensure all remaining tasks finish before stopping
-        await asyncio.gather(*tasks)
+        # Ensure all remaining tasks finish before stopping. Stopped jobs raise
+        # CancelledError during this drain, which is expected, but a genuine
+        # handler error must not be silently discarded.
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        for result in results:
+            if isinstance(result, Exception) and not isinstance(result, asyncio.CancelledError):
+                log.error(f"JobScaler.run_jobs | Task failed during shutdown drain: {result}")
+
+    async def monitor_stop_signals(self, session: ClientSession):
+        """
+        Long-polls the dedicated stop channel and stops signalled jobs.
+
+        Runs in an infinite loop while the worker is alive. The Runpod server
+        signals a request to be stopped (for example when it is cancelled or
+        times out) and this loop stops just that in-progress job, leaving the
+        worker's other jobs running.
+        """
+        if self.stop_signals_fetcher is get_stop_signals and _job_stop_url() is None:
+            log.warn(
+                "JobScaler.monitor_stop_signals | Stop channel could not be derived "
+                "from the job-take URL; per-job stop is disabled for this worker."
+            )
+            return
+
+        while self.is_alive():
+            try:
+                # Bound the long-poll so shutdown is not blocked by the shared
+                # session's much longer default timeout.
+                job_ids = await asyncio.wait_for(
+                    self.stop_signals_fetcher(session),
+                    timeout=self.stop_signals_fetcher_timeout,
+                )
+                for job_id in job_ids:
+                    await self.stop_job(job_id)
+
+                if not job_ids:
+                    # floor delay so the loop can't busy-spin when the server
+                    # returns immediately instead of holding the poll open
+                    await asyncio.sleep(1)
+            except TooManyRequests:
+                await asyncio.sleep(5)  # debounce
+            except asyncio.CancelledError:
+                log.debug("JobScaler.monitor_stop_signals | Request was cancelled.")
+                raise  # CancelledError is a BaseException
+            except asyncio.TimeoutError:
+                log.debug("JobScaler.monitor_stop_signals | Stop poll timed out. Retrying.")
+            except Exception as error:
+                log.error(
+                    f"JobScaler.monitor_stop_signals | Error Type: {type(error).__name__} | Error Message: {str(error)}"
+                )
+                await asyncio.sleep(1)  # don't spin on persistent errors
+            finally:
+                await asyncio.sleep(0)
+
+    async def stop_job(self, job_id: str) -> bool:
+        """
+        Stop a single in-progress job by cancelling its running task.
+
+        Args:
+            job_id: The id of the job to stop.
+
+        Returns:
+            True if a matching in-progress job was found and stopped,
+            False otherwise.
+        """
+        task = self.jobs_tasks.get(job_id)
+        if task is None:
+            log.debug(f"JobScaler.stop_job | No in-progress job for {job_id}.")
+            return False
+
+        log.info("Stopping job.", job_id)
+        task.cancel()
+        return True
 
     async def handle_job(self, session: ClientSession, job: dict):
         """
@@ -264,15 +350,20 @@ async def handle_job(self, session: ClientSession, job: dict):
             if self.config.get("refresh_worker", False):
                 self.kill_worker()
 
+        except asyncio.CancelledError:
+            log.info("Job stopped.", job["id"])
+            raise  # CancelledError is a BaseException
+
         except Exception as err:
             log.error(f"Error handling job: {err}", job["id"])
-            raise err
+            raise
 
         finally:
             # Inform Queue of a task completion
             self.jobs_queue.task_done()
 
             # Job is no longer in progress
             self.job_progress.remove(job)
+            self.jobs_tasks.pop(job["id"], None)
 
             log.debug("Finished Job", job["id"])