fix(langgraph): return an awaited @task's value while streaming

With stream=True, arun_with_retry drains the task's astream() and returned nothing, so a @task awaited inside a node got None whenever the node ran in streaming mode: under astream_events (v1 and v2) and anything built on it. Return the value the task wrote instead, on both stream paths. Same idea as #9295 by AaronG18 and the lookup amishaguptaberk suggested.
feat(cli): add the priority section to langgraph.json (#9393 )
2026-10-25 08:21:40 +01:00 · 2026-10-23 07:24:18 -04:00 · 2026-10-22 17:26:01 -04:00 · 2026-10-22 17:25:38 -04:00 · 2026-10-22 16:27:58 -04:00
27 changed files with 1020 additions and 647 deletions
@@ -23,6 +23,7 @@ from langgraph_cli.schemas import (
    GraphDef,
    HttpConfig,
    IndexConfig,
+    PriorityConfig,
    SecurityConfig,
    SerdeConfig,
    StoreConfig,
@@ -113,6 +114,7 @@ def add_descriptions_to_schema(schema, cls):
                StoreConfig,
                IndexConfig,
                AuthConfig,
+                PriorityConfig,
                SecurityConfig,
                HttpConfig,
                CorsConfig,
@@ -847,6 +847,7 @@ def dev(
        store=config_json.get("store"),
        wait_for_client=wait_for_client,
        auth=config_json.get("auth"),
+        priority=config_json.get("priority"),
        http=config_json.get("http"),
        ui=config_json.get("ui"),
        ui_config=config_json.get("ui_config"),
@@ -442,6 +442,7 @@ def validate_config(
        "store": config.get("store"),
        "auth": config.get("auth"),
        "encryption": config.get("encryption"),
+        "priority": config.get("priority"),
        "http": config.get("http"),
        # Pass through webhooks config so it can be injected into the image
        "webhooks": config.get("webhooks"),
@@ -606,6 +607,19 @@ def validate_config(
                    f"Invalid encryption.path format: '{encryption_conf['path']}'. "
                    "Must be in format './path/to/file.py:attribute_name'"
                )
+    # Validate priority config
+    if priority_conf := config.get("priority"):
+        if "path" in priority_conf:
+            if ":" not in priority_conf["path"]:
+                raise ValueError(
+                    f"Invalid priority.path format: '{priority_conf['path']}'. "
+                    "Must be in format './path/to/file.py:attribute_name'"
+                )
+            if _is_node_graph(priority_conf["path"]):
+                raise ValueError(
+                    "JavaScript priority callbacks are not supported: "
+                    f"'{priority_conf['path']}'. Use a Python file or module instead."
+                )
    if http_conf := config.get("http"):
        if "app" in http_conf:
            if ":" not in http_conf["app"]:
@@ -649,6 +663,7 @@ _KNOWN_CONFIG_KEYS = {
    "store",
    "auth",
    "encryption",
+    "priority",
    "http",
    "webhooks",
    "checkpointer",
@@ -1080,6 +1095,49 @@ def _update_encryption_path(
    )


+def _update_priority_path(
+    config_path: pathlib.Path, config: Config, local_deps: LocalDeps
+) -> None:
+    """Update priority.path to use Docker container paths."""
+    priority_conf = config.get("priority")
+    if not priority_conf or not (path_str := priority_conf.get("path")):
+        return
+
+    module_str, sep, attr_str = path_str.partition(":")
+    if not sep or not module_str.startswith("."):
+        return  # Already validated or absolute path
+
+    resolved = config_path.parent / module_str
+    if not resolved.exists():
+        raise FileNotFoundError(
+            f"Priority file not found: {resolved} (from {path_str})"
+        )
+    if not resolved.is_file():
+        raise IsADirectoryError(f"Priority path must be a file: {resolved}")
+
+    # Check faux packages first (they take precedence)
+    for faux_path, (_, destpath) in local_deps.faux_pkgs.items():
+        if resolved.is_relative_to(faux_path):
+            new_path = f"{destpath}/{resolved.relative_to(faux_path)}:{attr_str}"
+            priority_conf["path"] = new_path
+            return
+
+    # Check real packages
+    for real_path in local_deps.real_pkgs:
+        if resolved.is_relative_to(real_path):
+            new_path = (
+                f"/deps/{real_path.name}/{resolved.relative_to(real_path)}:{attr_str}"
+            )
+            priority_conf["path"] = new_path
+            return
+
+    raise ValueError(
+        f"Priority file '{resolved}' not covered by dependencies.\n"
+        "Add its parent directory to the 'dependencies' array in your config.\n"
+        f"Current dependencies: {config['dependencies']}"
+    )
+
+
 def _update_checkpointer_path(
    config_path: pathlib.Path, config: Config, local_deps: LocalDeps
 ) -> None:
@@ -1213,6 +1271,9 @@ def _build_runtime_env_vars(config: Config) -> list[str]:
    if (encryption_config := config.get("encryption")) is not None:
        env_vars.append(f"ENV LANGGRAPH_ENCRYPTION='{json.dumps(encryption_config)}'")

+    if (priority_config := config.get("priority")) is not None:
+        env_vars.append(f"ENV LANGGRAPH_PRIORITY='{json.dumps(priority_config)}'")
+
    if (http_config := config.get("http")) is not None:
        env_vars.append(f"ENV LANGGRAPH_HTTP='{json.dumps(http_config)}'")

@@ -1380,6 +1441,8 @@ def python_config_to_docker(
    _update_auth_path(config_path, config, local_deps)
    # Rewrite encryption path, so it points to the correct location in the Docker container
    _update_encryption_path(config_path, config, local_deps)
+    # Rewrite priority path, so it points to the correct location in the Docker container
+    _update_priority_path(config_path, config, local_deps)
    # Rewrite checkpointer path, so it points to the correct location in the Docker container
    _update_checkpointer_path(config_path, config, local_deps)
    # Rewrite HTTP app path, so it points to the correct location in the Docker container
@@ -26,6 +26,7 @@ from langgraph_cli.dependency_tracking import find_tracked_packages
 from langgraph_cli.docker import build_docker_image, can_build_locally
 from langgraph_cli.exec import CommandRunner, Runner, subp_exec
 from langgraph_cli.host_backend import (
+    MAX_PAGE_SIZE,
    ControlPlaneEndpoints,
    HostBackendClient,
    HostBackendError,
@@ -182,15 +183,12 @@ class Listener:
        )
        return cls(
            identifier,
-            str(resource.get("compute_id") or ""),
+            str(resource.get("compute_id", "")),
            tuple(str(namespace) for namespace in namespaces)
            if isinstance(namespaces, list)
            else (),
        )

-    def cells(self) -> tuple[str | None, ...]:
-        return (self.id, self.compute_id, ", ".join(self.namespaces))
-

@dataclass(frozen=True, slots=True)
 class Unplaced:
@@ -278,6 +276,8 @@ def _describe_listeners(listeners: Sequence[Listener]) -> str:
    ]
    if len(listeners) > len(shown):
        lines.append(f"  ... and {len(listeners) - len(shown)} more")
+    if len(listeners) == MAX_PAGE_SIZE:
+        lines.append(f"  (only the first {MAX_PAGE_SIZE} listeners were read)")
    return "\n".join(lines)


@@ -499,9 +499,16 @@ def _source_of(resource: object) -> str | None:
 def find_deployment_by_name(
    client: HostBackendClient, name: str
 ) -> ExistingDeployment | None:
-    for resource in client.list_deployments(name=name, name_contains=name):
+    listed = client.list_deployments(name=name, name_contains=name, limit=MAX_PAGE_SIZE)
+    for resource in listed:
        if resource.get("name") == name and resource.get("id"):
            return ExistingDeployment(str(resource["id"]), _source_of(resource))
+    if len(listed) >= MAX_PAGE_SIZE:
+        raise click.ClickException(
+            "This workspace has more deployments than the CLI can search, so it "
+            f"cannot tell whether '{name}' already exists. Pass --deployment-id to "
+            "update an existing deployment."
+        )
    return None


@@ -576,150 +583,103 @@ def _validate_prebuilt_image(
    _get_emitter().info(f"Image is available for {_DEPLOYMENT_PLATFORM}")


-Resource = Mapping[str, object]
-
-
-@dataclass(frozen=True, slots=True)
-class Table:
-    headers: tuple[str, ...]
-    cells: Callable[[Resource], tuple[str | None, ...]]
-
-    def render(self, resources: Sequence[Resource]) -> str:
-        grid = [
-            self.headers,
-            *(
-                tuple(cell or "-" for cell in self.cells(resource))
-                for resource in resources
-            ),
-        ]
-        widths = [max(map(len, column)) for column in zip(*grid, strict=True)]
-        rule = tuple("-" * width for width in widths)
-        return "\n".join(
-            "  ".join(
-                cell.ljust(width) for cell, width in zip(row, widths, strict=True)
-            )
-            for row in (grid[0], rule, *grid[1:])
-        )
-
-
-@dataclass(frozen=True, slots=True)
-class Listing:
-    table: Table
-    empty_message: str
-    json_view: Callable[[Resource], object]
-
-    def as_table(self, resources: Sequence[Resource]) -> str:
-        return self.table.render(resources) if resources else self.empty_message
-
-    def as_json(self, resources: Sequence[Resource]) -> str:
-        return json_mod.dumps([self.json_view(resource) for resource in resources])
-
-
-class ListOutput(Protocol):
-    @property
-    def no_input(self) -> bool: ...
-
-    def show(self, listing: Listing, resources: Sequence[Resource]) -> None: ...
-
-
-@dataclass(frozen=True, slots=True)
-class TableOutput:
-    @property
-    def no_input(self) -> bool:
-        return False
-
-    def show(self, listing: Listing, resources: Sequence[Resource]) -> None:
-        click.echo(listing.as_table(resources))
-
-
-@dataclass(frozen=True, slots=True)
-class JsonOutput:
-    @property
-    def no_input(self) -> bool:
-        return True
-
-    def show(self, listing: Listing, resources: Sequence[Resource]) -> None:
-        click.echo(listing.as_json(resources))
-
-
-def _text(resource: Resource, key: str) -> str | None:
-    value = resource.get(key)
-    return str(value) if value else None
-
-
-def _deployment_url(deployment: Resource) -> str | None:
+def _extract_deployment_url(deployment: dict[str, object]) -> str:
    source_config = deployment.get("source_config")
-    custom_url = (
-        source_config.get("custom_url") if isinstance(source_config, Mapping) else None
-    )
-    return custom_url if isinstance(custom_url, str) and custom_url else None
+    if isinstance(source_config, dict):
+        custom_url = source_config.get("custom_url")
+        if isinstance(custom_url, str) and custom_url:
+            return custom_url
+    return "-"


-def _with_replaced_status(revisions: Sequence[Resource]) -> list[Resource]:
-    live_index = next(
+def format_deployments_table(deployments: Sequence[dict[str, object]]) -> str:
+    headers = ("Deployment ID", "Deployment Name", "Deployment URL")
+    rows = [
        (
-            index
-            for index, revision in enumerate(revisions)
-            if revision.get("status") == "DEPLOYED"
-        ),
-        None,
-    )
-    return [
-        {**revision, "status": "REPLACED"}
-        if revision.get("status") == "DEPLOYED" and index != live_index
-        else revision
-        for index, revision in enumerate(revisions)
+            str(deployment.get("id", "-") or "-"),
+            str(deployment.get("name", "-") or "-"),
+            _extract_deployment_url(deployment),
+        )
+        for deployment in deployments
+    ]
+    widths = [
+        max(len(headers[index]), *(len(row[index]) for row in rows))
+        for index in range(len(headers))
    ]

+    def format_row(row: Sequence[str]) -> str:
+        return "  ".join(value.ljust(widths[index]) for index, value in enumerate(row))

-@dataclass(frozen=True, slots=True)
-class DeploymentRow:
-    id: str
-    name: str | None
-    url: str | None
+    lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
+    lines.extend(format_row(row) for row in rows)
+    return "\n".join(lines)

-    @classmethod
-    def from_resource(cls, deployment: Resource) -> "DeploymentRow":
-        identifier = _text(deployment, "id")
-        if identifier is None:
-            raise HostBackendError(
-                "The control plane returned a deployment without an id."
+
+def _extract_listener_namespaces(listener: dict[str, object]) -> str:
+    compute_config = listener.get("compute_config")
+    namespaces = (
+        compute_config.get("k8s_namespaces")
+        if isinstance(compute_config, dict)
+        else None
+    )
+    if isinstance(namespaces, list) and namespaces:
+        return ", ".join(str(namespace) for namespace in namespaces)
+    return "-"
+
+
+def format_listeners_table(listeners: Sequence[dict[str, object]]) -> str:
+    headers = ("Listener ID", "Compute ID", "Namespaces")
+    rows = [
+        (
+            str(listener.get("id", "-") or "-"),
+            str(listener.get("compute_id", "-") or "-"),
+            _extract_listener_namespaces(listener),
+        )
+        for listener in listeners
+    ]
+    widths = [
+        max(len(headers[index]), *(len(row[index]) for row in rows))
+        for index in range(len(headers))
+    ]
+
+    def format_row(row: Sequence[str]) -> str:
+        return "  ".join(value.ljust(widths[index]) for index, value in enumerate(row))
+
+    lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
+    lines.extend(format_row(row) for row in rows)
+    return "\n".join(lines)
+
+
+def format_revisions_table(revisions: Sequence[dict[str, object]]) -> str:
+    headers = ("Revision ID", "Status", "Created At")
+    latest_deployed_seen = False
+    rows = []
+    for revision in revisions:
+        status = str(revision.get("status", "-") or "-")
+        if status == "DEPLOYED":
+            if latest_deployed_seen:
+                status = "REPLACED"
+            else:
+                latest_deployed_seen = True
+        rows.append(
+            (
+                str(revision.get("id", "-") or "-"),
+                status,
+                str(revision.get("created_at", "-") or "-"),
            )
-        return cls(identifier, _text(deployment, "name"), _deployment_url(deployment))
+        )

-    def cells(self) -> tuple[str | None, ...]:
-        return (self.id, self.name, self.url)
+    widths = [
+        max(len(headers[index]), *(len(row[index]) for row in rows))
+        for index in range(len(headers))
+    ]

-    def as_json(self) -> dict[str, str | None]:
-        return {"id": self.id, "name": self.name}
+    def format_row(row: Sequence[str]) -> str:
+        return "  ".join(value.ljust(widths[index]) for index, value in enumerate(row))

-
-DEPLOYMENTS = Listing(
-    Table(
-        ("Deployment ID", "Deployment Name", "Deployment URL"),
-        lambda deployment: DeploymentRow.from_resource(deployment).cells(),
-    ),
-    empty_message="No deployments found.",
-    json_view=lambda deployment: DeploymentRow.from_resource(deployment).as_json(),
-)
-
-LISTENERS = Listing(
-    Table(
-        ("Listener ID", "Compute ID", "Namespaces"),
-        lambda listener: Listener.from_resource(listener).cells(),
-    ),
-    empty_message="No listeners found for this workspace.",
-    json_view=dict,
-)
-
-REVISIONS_TABLE = Table(
-    ("Revision ID", "Status", "Created At"),
-    lambda revision: (
-        _text(revision, "id"),
-        _text(revision, "status"),
-        _text(revision, "created_at"),
-    ),
-)
+    lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
+    lines.extend(format_row(row) for row in rows)
+    return "\n".join(lines)


 def format_timestamp(ts) -> str:
@@ -881,6 +841,7 @@ def _find_deployment(
            lambda c: c.list_deployments(
                agent_id=selector.agent_id,
                agent_environment=selector.environment,
+                limit=MAX_PAGE_SIZE,
            ),
        )
        if len(existing) > 1:
@@ -1046,15 +1007,20 @@ def _print_deployment_result(
    dashboard_label: str,
    status_url: str | None = None,
 ) -> None:
+    """Print final deployment status and raise on failure."""
    em = _get_emitter()
-    deployment = client.get_deployment(deployment_id)
-    url = _deployment_url(deployment) if isinstance(deployment, Mapping) else None
+    dep_info = client.get_deployment(deployment_id)
+    custom_url = None
+    if isinstance(dep_info, dict):
+        sc = dep_info.get("source_config")
+        if isinstance(sc, dict):
+            custom_url = sc.get("custom_url")

    if last_status == "DEPLOYED":
        em.result(
            "succeeded",
            deployment_id=deployment_id,
-            url=url,
+            url=custom_url,
            status_url=status_url,
        )
    elif last_status in ("BUILD_FAILED", "DEPLOY_FAILED", "CREATE_FAILED"):
@@ -2154,21 +2120,6 @@ OPT_VERBOSE = click.option(
 )


-def _list_output(
-    _ctx: click.Context, _param: click.Parameter, as_json: bool
-) -> ListOutput:
-    return JsonOutput() if as_json else TableOutput()
-
-
-OPT_LIST_OUTPUT = click.option(
-    "--json",
-    "output",
-    is_flag=True,
-    callback=_list_output,
-    help="Print a JSON array instead of a table.",
-)
-
-
 class NestedHelpGroup(click.Group):
    """Click group that shows one level of nested subcommands in top-level help."""

@@ -2643,7 +2594,6 @@ def _deploy_cmd(
    default="",
    help="Only show deployments whose names contain this value.",
 )
-@OPT_LIST_OUTPUT
@deploy.command("list", help="[Beta] List LangSmith Deployments.")
 def deploy_list(
    api_key: str | None,
@@ -2651,13 +2601,11 @@ def deploy_list(
    name_contains: str,
    agent_id: str | None,
    environment: str | None,
-    output: ListOutput,
 ) -> None:
    if agent_id is not None or environment is not None:
        click.secho(
            "Note: --agent-id and --agent-environment flags are in private beta",
            fg="yellow",
-            err=True,
        )
    if agent_id is not None and not agent_id.strip():
        raise click.UsageError("--agent-id must not be empty.")
@@ -2666,13 +2614,15 @@ def deploy_list(
        filters["agent_id"] = agent_id
    if environment is not None:
        filters["agent_environment"] = environment
-    client = _create_host_backend_client(host_url, api_key, no_input=output.no_input)
+    client = _create_host_backend_client(host_url, api_key)
    deployments = _call_host_backend_with_optional_tenant(
        client,
        lambda c: c.list_deployments(name_contains=name_contains, **filters),
-        no_input=output.no_input,
    )
-    output.show(DEPLOYMENTS, deployments)
+    if not deployments:
+        click.echo("No deployments found.")
+        return
+    click.echo(format_deployments_table(deployments))


 # ---------------------------------------------------------------------------
@@ -2691,7 +2641,13 @@ def deploy_listeners() -> None:

@OPT_HOST_API_KEY
@OPT_HOST_URL
-@OPT_LIST_OUTPUT
+@click.option(
+    "--json",
+    "json_output",
+    is_flag=True,
+    default=False,
+    help="Emit the listeners as a JSON array instead of a table.",
+)
@deploy_listeners.command(
    "list",
    help=(
@@ -2701,13 +2657,19 @@ def deploy_listeners() -> None:
    ),
 )
 def deploy_listeners_list(
-    api_key: str | None, host_url: str | None, output: ListOutput
+    api_key: str | None, host_url: str | None, json_output: bool
 ) -> None:
-    client = _create_host_backend_client(host_url, api_key, no_input=output.no_input)
+    client = _create_host_backend_client(host_url, api_key, no_input=json_output)
    listeners = _call_host_backend_with_optional_tenant(
-        client, lambda c: c.list_listeners(), no_input=output.no_input
+        client, lambda c: c.list_listeners(), no_input=json_output
    )
-    output.show(LISTENERS, listeners)
+    if json_output:
+        click.echo(json_mod.dumps(listeners))
+        return
+    if not listeners:
+        click.echo("No listeners found for this workspace.")
+        return
+    click.echo(format_listeners_table(listeners))


 # ---------------------------------------------------------------------------
@@ -2750,7 +2712,7 @@ def deploy_revisions_list(
    if not revisions:
        click.echo(f"No revisions found for deployment {deployment_id}.")
        return
-    click.echo(REVISIONS_TABLE.render(_with_replaced_status(revisions)))
+    click.echo(format_revisions_table(revisions))


 # ---------------------------------------------------------------------------
@@ -2,7 +2,6 @@

 from __future__ import annotations

-from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Any, Literal
 from urllib.parse import urlparse
@@ -221,51 +220,27 @@ class HostBackendClient:
        *,
        name: str | None = None,
        name_contains: str | None = None,
+        limit: int | None = None,
        agent_id: str | None = None,
        agent_environment: str | None = None,
-    ) -> list[dict[str, object]]:
+    ) -> list[dict[str, Any]]:
        given = (
            ("name", name),
            ("name_contains", name_contains),
+            ("limit", limit),
            ("agent_id", agent_id),
            ("agent_environment", agent_environment),
        )
-        filters = {key: value for key, value in given if value is not None}
-        return self._every_page("/v2/deployments", filters)
-
-    def _every_page(
-        self, path: str, filters: Mapping[str, object]
-    ) -> list[dict[str, object]]:
-        resources: list[dict[str, object]] = []
-        seen_ids: set[object] = set()
-        while True:
-            page = _resources(
-                self._request(
-                    "GET",
-                    path,
-                    params={
-                        **filters,
-                        "limit": MAX_PAGE_SIZE,
-                        "offset": len(resources),
-                    },
-                )
-            )
-            page_ids = {resource.get("id") for resource in page}
-            if page_ids and page_ids <= seen_ids:
-                raise HostBackendError(
-                    f"GET {path} returned the same resources again: this control "
-                    "plane ignores 'offset', so the CLI cannot read past the first page."
-                )
-            seen_ids |= page_ids
-            resources.extend(page)
-            if len(page) < MAX_PAGE_SIZE:
-                return resources
+        params = {key: value for key, value in given if value is not None}
+        return _resources(self._request("GET", "/v2/deployments", params=params))

    def get_listener(self, listener_id: str) -> dict[str, Any]:
        return self._request("GET", f"/v2/listeners/{listener_id}")

-    def list_listeners(self) -> list[dict[str, object]]:
-        return self._every_page("/v2/listeners", {})
+    def list_listeners(self) -> list[dict[str, Any]]:
+        return _resources(
+            self._request("GET", "/v2/listeners", params={"limit": MAX_PAGE_SIZE})
+        )

    def get_deployment(self, deployment_id: str) -> dict[str, Any]:
        return self._request("GET", f"/v2/deployments/{deployment_id}")
@@ -373,6 +373,27 @@ class EncryptionConfig(TypedDict, total=False):
    """


+class PriorityConfig(TypedDict, total=False):
+    """Configuration for custom run priority logic.
+
+    Allows you to decide each thread's priority settings with a Python function
+    that the server calls at the first run on a thread with no settings, which
+    can come long after the thread was created."""
+
+    path: str
+    """Required. Path to an instance of the Priority() class that registers an assign function.
+
+    Format: "path/to/file.py:my_priority"
+
+    Example:
+        {
+            "priority": {
+                "path": "./priority.py:my_priority"
+            }
+        }
+    """
+
+
 class CorsConfig(TypedDict, total=False):
    """Specifies Cross-Origin Resource Sharing (CORS) rules for your server.

@@ -753,6 +774,13 @@ class Config(TypedDict, total=False):
    Allows you to implement custom encryption for sensitive data stored in the database.
    """

+    priority: PriorityConfig | None
+    """Optional. Custom run priority config, including the path to your Python priority logic.
+
+    Allows you to decide each thread's priority settings at the first run on a thread with no settings,
+    which can come long after the thread was created.
+    """
+
    http: HttpConfig | None
    """Optional. Configuration for the built-in HTTP server, controlling which custom routes are exposed
    and how cross-origin requests are handled.
@@ -784,6 +812,7 @@ __all__ = [
    "CheckpointerConfig",
    "AuthConfig",
    "EncryptionConfig",
+    "PriorityConfig",
    "HttpConfig",
    "MiddlewareOrders",
    "Distros",
@@ -910,6 +910,7 @@ def python_config_to_docker_uv_lock(
    for section, key in [
        ("auth", "path"),
        ("encryption", "path"),
+        ("priority", "path"),
        ("checkpointer", "path"),
        ("http", "app"),
    ]:
@@ -196,6 +196,17 @@
                }
              ]
            },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
            "source": {
              "anyOf": [
                {
@@ -438,6 +449,17 @@
                }
              ]
            },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
            "source": {
              "$ref": "#/$defs/UvSource"
            },
@@ -666,6 +688,17 @@
                }
              ]
            },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
            "source": {
              "anyOf": [
                {
@@ -1170,6 +1203,18 @@
      },
      "required": []
    },
+    "PriorityConfig": {
+      "title": "PriorityConfig",
+      "description": "Configuration for custom run priority logic.\n\nAllows you to decide each thread's priority settings with a Python function\nthat the server calls at the first run on a thread with no settings, which\ncan come long after the thread was created.",
+      "type": "object",
+      "properties": {
+        "path": {
+          "type": "string",
+          "description": "Required. Path to an instance of the Priority() class that registers an assign function.\n"
+        }
+      },
+      "required": []
+    },
    "UvSource": {
      "title": "UvSource",
      "description": "Deployment source rooted at a uv project or workspace.",
@@ -196,6 +196,17 @@
                }
              ]
            },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
            "source": {
              "anyOf": [
                {
@@ -438,6 +449,17 @@
                }
              ]
            },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
            "source": {
              "$ref": "#/$defs/UvSource"
            },
@@ -666,6 +688,17 @@
                }
              ]
            },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
            "source": {
              "anyOf": [
                {
@@ -1170,6 +1203,18 @@
      },
      "required": []
    },
+    "PriorityConfig": {
+      "title": "PriorityConfig",
+      "description": "Configuration for custom run priority logic.\n\nAllows you to decide each thread's priority settings with a Python function\nthat the server calls at the first run on a thread with no settings, which\ncan come long after the thread was created.",
+      "type": "object",
+      "properties": {
+        "path": {
+          "type": "string",
+          "description": "Required. Path to an instance of the Priority() class that registers an assign function.\n"
+        }
+      },
+      "required": []
+    },
    "UvSource": {
      "title": "UvSource",
      "description": "Deployment source rooted at a uv project or workspace.",
@@ -2,10 +2,12 @@ import json
 import pathlib
 import re
 import shutil
+import sys
 import tempfile
 import textwrap
 from contextlib import contextmanager
 from pathlib import Path
+from unittest.mock import MagicMock

 import click
 from click.testing import CliRunner
@@ -371,6 +373,32 @@ def test_dev_command_requires_ssl_certfile_and_keyfile_together(tmp_path) -> Non
    assert "Both --ssl-certfile and --ssl-keyfile must be provided" in result.output


+def test_dev_command_passes_priority_to_the_server(tmp_path, monkeypatch) -> None:
+    config_path = tmp_path / "langgraph.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "dependencies": ["."],
+                "graphs": {"agent": "./agent.py:graph"},
+                "priority": {"path": "./priority.py:priority"},
+            }
+        ),
+        encoding="utf-8",
+    )
+    # The test environment doesn't install langgraph-api, so stand in for its server.
+    server = MagicMock()
+    monkeypatch.setitem(sys.modules, "langgraph_api.cli", server)
+    # dev appends the working directory and its dependencies to sys.path.
+    monkeypatch.setattr(sys, "path", list(sys.path))
+
+    result = CliRunner().invoke(cli, ["dev", "--config", str(config_path)])
+
+    assert result.exit_code == 0, result.output
+    assert server.run_server.call_args.kwargs["priority"] == {
+        "path": "./priority.py:priority"
+    }
+
+
 def test_deploy_list_command(monkeypatch) -> None:
    runner = CliRunner()
    captured: dict[str, str] = {}
@@ -454,177 +482,6 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
    assert result.output.strip() == "No deployments found."


-def test_deploy_list_prints_an_aligned_table(monkeypatch) -> None:
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def list_deployments(self, name_contains: str = ""):
-            return [
-                {
-                    "id": "dep-123",
-                    "name": "alpha",
-                    "source_config": {"custom_url": "https://alpha.example.com"},
-                },
-                {"id": "dep-4567", "name": "beta-agent", "source_config": {}},
-            ]
-
-    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
-
-    result = CliRunner().invoke(
-        cli,
-        [
-            "deploy",
-            "list",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert result.output.splitlines() == [
-        "Deployment ID  Deployment Name  Deployment URL           ",
-        "-------------  ---------------  -------------------------",
-        "dep-123        alpha            https://alpha.example.com",
-        "dep-4567       beta-agent       -                        ",
-    ]
-
-
-def _deploy_list_json(
-    monkeypatch, deployments: list[dict], *extra_args: str
-) -> click.testing.Result:
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def list_deployments(
-            self,
-            name_contains: str = "",
-            agent_id: str | None = None,
-            agent_environment: str | None = None,
-        ):
-            return deployments
-
-    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
-    return CliRunner().invoke(
-        cli,
-        [
-            "deploy",
-            "list",
-            "--json",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-            *extra_args,
-        ],
-    )
-
-
-def test_deploy_list_command_json(monkeypatch) -> None:
-    result = _deploy_list_json(
-        monkeypatch,
-        [
-            {
-                "id": "dep-123",
-                "name": "alpha",
-                "source_config": {"custom_url": "https://alpha.example.com"},
-            },
-            {"id": "dep-456", "name": "beta", "source_config": {}},
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert json.loads(result.stdout) == [
-        {"id": "dep-123", "name": "alpha"},
-        {"id": "dep-456", "name": "beta"},
-    ]
-
-
-def test_deploy_list_json_never_prints_deployment_secrets(monkeypatch) -> None:
-    result = _deploy_list_json(
-        monkeypatch,
-        [
-            {
-                "id": "dep-123",
-                "name": "alpha",
-                "source_config": {"custom_url": None},
-                "secrets": [{"name": "OPENAI_API_KEY", "value": "sk-plain-value"}],
-            }
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert "sk-plain-value" not in result.output
-    assert json.loads(result.stdout) == [{"id": "dep-123", "name": "alpha"}]
-
-
-def test_deploy_list_command_json_no_results(monkeypatch) -> None:
-    result = _deploy_list_json(monkeypatch, [])
-
-    assert result.exit_code == 0, result.output
-    assert json.loads(result.stdout) == []
-
-
-def test_deploy_list_rejects_a_deployment_without_an_id(monkeypatch) -> None:
-    result = _deploy_list_json(monkeypatch, [{"name": "alpha"}])
-
-    assert result.exit_code != 0
-    assert "deployment without an id" in result.output
-
-
-def test_deploy_list_json_keeps_the_agent_note_off_stdout(monkeypatch) -> None:
-    result = _deploy_list_json(
-        monkeypatch,
-        [{"id": "dep-123", "name": "alpha"}],
-        "--agent-id",
-        "customer-support",
-        "--agent-environment",
-        "staging",
-    )
-
-    assert result.exit_code == 0, result.output
-    assert json.loads(result.stdout) == [{"id": "dep-123", "name": "alpha"}]
-    assert "private beta" in result.stderr
-
-
-def test_deploy_list_json_never_prompts_for_a_workspace(monkeypatch) -> None:
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def list_deployments(self, name_contains: str = ""):
-            raise HostBackendError("requires workspace specification", status_code=403)
-
-        def set_tenant(self, tenant_id: str) -> None:
-            raise AssertionError("must not retry with a prompted tenant with --json")
-
-    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
-    prompt_calls = []
-    monkeypatch.setattr(
-        click, "prompt", lambda *a, **k: prompt_calls.append((a, k)) or "unreachable"
-    )
-
-    result = CliRunner().invoke(
-        cli,
-        [
-            "deploy",
-            "list",
-            "--json",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-        ],
-    )
-
-    assert prompt_calls == []
-    assert result.exit_code != 0
-    assert "LANGSMITH_TENANT_ID" in result.output
-
-
 def test_deploy_listeners_list_command(monkeypatch) -> None:
    runner = CliRunner()
    captured: dict[str, str] = {}
@@ -707,84 +564,6 @@ def test_deploy_listeners_list_command_no_results(monkeypatch) -> None:
    assert result.output.strip() == "No listeners found for this workspace."


-def test_deploy_listeners_list_prints_an_aligned_table(monkeypatch) -> None:
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def list_listeners(self):
-            return [
-                {
-                    "id": "listener-1",
-                    "compute_id": "prod-cluster",
-                    "compute_config": {"k8s_namespaces": ["agents"]},
-                },
-                {
-                    "id": "listener-22",
-                    "compute_id": "multi",
-                    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
-                },
-                {"id": "listener-3", "compute_id": "", "compute_config": {}},
-                {
-                    "id": "listener-4",
-                    "compute_id": None,
-                    "compute_config": {"k8s_namespaces": ["agents"]},
-                },
-            ]
-
-    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
-
-    result = CliRunner().invoke(
-        cli,
-        [
-            "deploy",
-            "listeners",
-            "list",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert result.output.splitlines() == [
-        "Listener ID  Compute ID    Namespaces            ",
-        "-----------  ------------  ----------------------",
-        "listener-1   prod-cluster  agents                ",
-        "listener-22  multi         agents, agents-staging",
-        "listener-3   -             -                     ",
-        "listener-4   -             agents                ",
-    ]
-
-
-def test_deploy_listeners_list_rejects_a_listener_without_an_id(monkeypatch) -> None:
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def list_listeners(self):
-            return [{"compute_id": "prod-cluster"}]
-
-    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
-
-    result = CliRunner().invoke(
-        cli,
-        [
-            "deploy",
-            "listeners",
-            "list",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-        ],
-    )
-
-    assert result.exit_code != 0
-    assert "listener without an id" in result.output
-
-
 def test_deploy_listeners_list_command_json(monkeypatch) -> None:
    runner = CliRunner()

@@ -987,54 +766,6 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
    assert result.output.strip() == "No revisions found for deployment dep-123."


-def test_deploy_revisions_list_marks_older_deployed_revisions_as_replaced(
-    monkeypatch,
-) -> None:
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def list_revisions(self, deployment_id: str, limit: int = 1):
-            return [
-                {
-                    "id": "rev-3",
-                    "status": "DEPLOYING",
-                    "created_at": "2026-10-20T10:00:00Z",
-                },
-                {
-                    "id": "rev-2",
-                    "status": "DEPLOYED",
-                    "created_at": "2026-10-19T10:00:00Z",
-                },
-                {"id": "rev-1", "status": "DEPLOYED", "created_at": None},
-            ]
-
-    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
-
-    result = CliRunner().invoke(
-        cli,
-        [
-            "deploy",
-            "revisions",
-            "list",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-            "dep-123",
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert result.output.splitlines() == [
-        "Revision ID  Status     Created At          ",
-        "-----------  ---------  --------------------",
-        "rev-3        DEPLOYING  2026-10-20T10:00:00Z",
-        "rev-2        DEPLOYED   2026-10-19T10:00:00Z",
-        "rev-1        REPLACED   -                   ",
-    ]
-
-
 def test_deploy_revisions_list_command_with_explicit_limit(monkeypatch) -> None:
    runner = CliRunner()
    captured: dict[str, str] = {}
@@ -111,11 +111,7 @@ class ControlPlaneDouble:
    def _respond(self, request: httpx.Request) -> httpx.Response:
        method, path = request.method, request.url.path
        if (method, path) == ("GET", "/v2/listeners"):
-            offset = int(request.url.params.get("offset", 0))
-            limit = int(request.url.params.get("limit", 20))
-            return httpx.Response(
-                200, json={"resources": self.listeners[offset : offset + limit]}
-            )
+            return httpx.Response(200, json={"resources": self.listeners})
        if method == "GET" and path.startswith("/v2/listeners/"):
            listener_id = path.rsplit("/", 1)[-1]
            if not _looks_like_a_uuid(listener_id):
@@ -1292,16 +1288,14 @@ def test_listener_flags_without_push_to_make_no_call_at_all(
    assert deploy_project.timeline == []


-def test_listeners_beyond_the_first_page_are_counted(
-    deploy_project: DeployProject,
-) -> None:
+def test_a_truncated_listener_page_says_so(deploy_project: DeployProject) -> None:
    deploy_project.control_plane.listeners = [
        {
            "id": str(uuid.UUID(int=index)),
            "compute_id": "cluster",
            "compute_config": {"k8s_namespaces": ["agents"]},
        }
-        for index in range(101)
+        for index in range(100)
    ]

    result = deploy_project.run(
@@ -1309,8 +1303,7 @@ def test_listeners_beyond_the_first_page_are_counted(
    )

    assert result.exit_code != 0
-    assert "... and 91 more" in result.output
-    assert "first 100" not in result.output
+    assert "first 100" in result.output


 def test_a_managed_build_in_a_listener_workspace_points_at_push_to(
@@ -16,6 +16,7 @@ from langgraph_cli.config import (
    config_to_docker,
    default_base_image,
    docker_tag,
+    get_unknown_keys,
    has_disallowed_build_command_content,
    validate_config,
    validate_config_file,
@@ -155,6 +156,7 @@ def test_validate_config():
        "store": None,
        "auth": None,
        "encryption": None,
+        "priority": None,
        "webhooks": None,
        "checkpointer": None,
        "http": None,
@@ -184,6 +186,7 @@ def test_validate_config():
        "store": None,
        "auth": None,
        "encryption": None,
+        "priority": None,
        "webhooks": None,
        "checkpointer": None,
        "http": None,
@@ -1289,6 +1292,109 @@ def test_config_to_docker_python_encryption_formatted():
    )


+@pytest.mark.parametrize(
+    "path", ["./priority.py:priority", "my_package.priority:priority"]
+)
+def test_config_to_docker_python_priority(path: str):
+    # Test that priority config is kept by validation and isn't an unknown key
+    raw_config = {
+        "python_version": "3.11",
+        "graphs": {"agent": "./agent.py:graph"},
+        "dependencies": ["."],
+        "priority": {"path": path},
+    }
+    validated = validate_config(raw_config)
+
+    assert validated["priority"] == {"path": path}
+    assert get_unknown_keys(raw_config) == []
+
+
+def test_config_to_docker_python_priority_bad_path():
+    # Test that invalid priority path format raises ValueError
+    with pytest.raises(ValueError, match="Invalid priority.path format"):
+        validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./agent.py:graph"},
+                "dependencies": ["."],
+                "priority": {"path": "./priority.py"},  # Missing :attribute
+            }
+        )
+
+
+@pytest.mark.parametrize("ext", [".ts", ".mts", ".cts", ".js", ".mjs", ".cjs"])
+def test_config_to_docker_python_priority_js_path(ext: str):
+    # Test that a JavaScript priority path is rejected
+    with pytest.raises(ValueError, match="JavaScript priority callbacks"):
+        validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./agent.py:graph"},
+                "dependencies": ["."],
+                "priority": {"path": f"./priority{ext}:priority"},
+            }
+        )
+
+
+def test_config_to_docker_nodejs_priority():
+    # Test that a Node.js-only project can name a Python priority file
+    actual_docker_stdin, _ = config_to_docker(
+        PATH_TO_CONFIG,
+        validate_config(
+            {
+                "node_version": "20",
+                "graphs": {"agent": "./graphs/agent.js:graph"},
+                "priority": {"path": "./priority.py:priority"},
+            }
+        ),
+        base_image="langchain/langgraphjs-api",
+    )
+    assert (
+        """ENV LANGGRAPH_PRIORITY='{"path": "./priority.py:priority"}'"""
+        in actual_docker_stdin
+    )
+
+
+def test_config_to_docker_python_priority_formatted():
+    # Test that priority config is written to the Docker output with its container path
+    actual_docker_stdin, _ = config_to_docker(
+        PATH_TO_CONFIG,
+        validate_config(
+            {
+                "python_version": "3.11",
+                "dependencies": ["."],
+                "graphs": {"agent": "./graphs/agent.py:graph"},
+                "priority": {"path": "./agent.py:my_priority"},
+            }
+        ),
+        base_image="langchain/langgraph-api",
+    )
+    assert (
+        "ENV LANGGRAPH_PRIORITY="
+        """'{"path": "/deps/outer-unit_tests/unit_tests/agent.py:my_priority"}'"""
+        in actual_docker_stdin
+    )
+
+
+def test_config_to_docker_python_priority_not_in_dependencies():
+    # Test that a priority file outside the dependencies raises ValueError
+    with pytest.raises(
+        ValueError, match="Priority file .* not covered by dependencies"
+    ):
+        config_to_docker(
+            PATH_TO_CONFIG,
+            validate_config(
+                {
+                    "python_version": "3.11",
+                    "dependencies": ["./graphs"],
+                    "graphs": {"agent": "./graphs/agent.py:graph"},
+                    "priority": {"path": "./agent.py:my_priority"},
+                }
+            ),
+            base_image="langchain/langgraph-api",
+        )
+
+
 def test_config_to_docker_nodejs_internal_docker_tag():
    graphs = {"agent": "./graphs/agent.js:graph"}
    actual_docker_stdin, additional_contexts = config_to_docker(
@@ -1731,6 +1837,38 @@ def test_config_to_docker_uv_lock_honors_root_workspace_sources():
        )


+def test_config_to_docker_uv_lock_rewrites_priority_path():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root, config_path = _write_uv_lock_workspace(
+            tmpdir_path,
+            agent_sources="[tool.uv.sources]\nshared = { workspace = true }",
+        )
+        (project_root / "libs" / "shared" / "src" / "shared" / "priority.py").touch()
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {
+                    "agent": "../../apps/agent/src/agent/graph.py:graph",
+                },
+                "source": {"kind": "uv", "root": "../..", "package": "agent"},
+                "priority": {
+                    "path": "../../libs/shared/src/shared/priority.py:priority"
+                },
+            }
+        )
+        docker, _ = config_to_docker(
+            config_path, config, base_image="langchain/langgraph-api:0.2.47"
+        )
+
+        assert (
+            "ENV LANGGRAPH_PRIORITY="
+            """'{"path": "/deps/workspace/libs/shared/src/shared/priority.py:priority"}'"""
+            in docker
+        )
+
+
 def test_config_to_docker_uv_lock_ignores_unrelated_workspace_package_sources():
    with tempfile.TemporaryDirectory() as tmpdir:
        tmpdir_path = pathlib.Path(tmpdir)
@@ -75,7 +75,6 @@ def test_agent_create(deployment_api, tmp_path, monkeypatch):
        "agent_id": "customer-support",
        "agent_environment": "staging",
        "limit": "100",
-        "offset": "0",
    }
    payload = json.loads(requests[1].content)
    assert payload["agent"] == {
@@ -33,7 +33,6 @@ from langgraph_cli.deploy import (
    _Emitter,
    _env_without_deployment_name,
    _parse_env_from_config,
-    _print_deployment_result,
    _resolve_env_path,
    _resolve_pushed_image_digest,
    _resource_spec,
@@ -537,39 +536,6 @@ class TestEmitterTextMode:
        assert "URL: https://app.test" in lines


-@pytest.mark.parametrize(
-    ("deployment", "expected_url_lines"),
-    [
-        pytest.param(
-            {
-                "id": "dep-1",
-                "source_config": {"custom_url": "https://agent.example.com"},
-            },
-            ["URL: https://agent.example.com"],
-            id="custom_url_is_printed",
-        ),
-        pytest.param(
-            {"id": "dep-1", "source_config": {}}, [], id="no_custom_url_no_line"
-        ),
-    ],
-)
-def test_a_deployed_result_prints_the_deployment_url(
-    capsys, monkeypatch, deployment, expected_url_lines
-):
-    class FakeClient:
-        def get_deployment(self, deployment_id: str) -> dict:
-            return deployment
-
-    monkeypatch.setattr(deploy_mod, "_emitter", None)
-
-    _print_deployment_result(
-        FakeClient(), "dep-1", "DEPLOYED", dashboard_label="Deployment dashboard"
-    )
-
-    lines = [line.strip() for line in capsys.readouterr().out.splitlines()]
-    assert [line for line in lines if line.startswith("URL:")] == expected_url_lines
-
-
 # ---------------------------------------------------------------------------
 # --no-input guard on _create_host_backend_client
 # ---------------------------------------------------------------------------
@@ -1244,7 +1210,6 @@ def test_finding_a_deployment_by_name_narrows_the_search_for_every_server_versio
        "name": "agent",
        "name_contains": "agent",
        "limit": "100",
-        "offset": "0",
    }
    assert found == ExistingDeployment("dep-1", "github")

@@ -1272,29 +1237,20 @@ def test_a_server_that_ignores_the_exact_name_filter_never_matches_another_deplo
    assert find_deployment_by_name(client, "brand-new-agent") is None


-def test_a_deployment_beyond_the_first_page_is_found_by_name():
-    deployments = [
-        *(
-            {"id": f"dep-{index}", "name": f"other-agent-{index}"}
-            for index in range(100)
-        ),
-        {"id": "dep-target", "name": "agent", "source": "external_docker"},
+def test_a_full_page_without_a_match_refuses_to_claim_the_name_is_free():
+    page = [
+        {"id": f"dep-{index}", "name": f"other-agent-{index}"} for index in range(100)
    ]
-
-    def handler(req: httpx.Request) -> httpx.Response:
-        offset = int(req.url.params["offset"])
-        limit = int(req.url.params["limit"])
-        return httpx.Response(
-            200, json={"resources": deployments[offset : offset + limit]}
-        )
-
    client = HostBackendClient(
-        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+        "https://api.example.com",
+        "key",
+        transport=httpx.MockTransport(
+            lambda req: httpx.Response(200, json={"resources": page})
+        ),
    )

-    assert find_deployment_by_name(client, "agent") == ExistingDeployment(
-        "dep-target", "external_docker"
-    )
+    with pytest.raises(click.ClickException, match="--deployment-id"):
+        find_deployment_by_name(client, "brand-new-agent")


 def test_a_partial_page_without_a_match_means_the_name_is_free():
@@ -575,7 +575,7 @@ def test_list_listeners_asks_for_a_full_page():
    )

    assert c.list_listeners() == [{"id": "listener-1"}]
-    assert seen["url"] == "https://api.example.com/v2/listeners?limit=100&offset=0"
+    assert seen["url"] == "https://api.example.com/v2/listeners?limit=100"


@pytest.mark.parametrize(
@@ -602,30 +602,29 @@ def test_is_cloud_recognises_the_managed_control_plane(control_plane_url, expect
    assert endpoints.is_cloud is expected


-FIRST_PAGE = {"limit": "100", "offset": "0"}
-
-
@pytest.mark.parametrize(
    ("call", "expected_params"),
    [
        pytest.param(
            lambda c: c.list_deployments(name="agent"),
-            {"name": "agent", **FIRST_PAGE},
+            {"name": "agent"},
            id="exact_name_filters_server_side",
        ),
        pytest.param(
            lambda c: c.list_deployments(name_contains="age"),
-            {"name_contains": "age", **FIRST_PAGE},
+            {"name_contains": "age"},
            id="substring_search_keeps_its_own_parameter",
        ),
        pytest.param(
            lambda c: c.list_deployments(),
-            FIRST_PAGE,
-            id="no_filter_sends_only_the_page",
+            {},
+            id="no_filter_sends_no_parameters",
        ),
        pytest.param(
-            lambda c: c.list_deployments(name="agent", name_contains="agent"),
-            {"name": "agent", "name_contains": "agent", **FIRST_PAGE},
+            lambda c: c.list_deployments(
+                name="agent", name_contains="agent", limit=100
+            ),
+            {"name": "agent", "name_contains": "agent", "limit": "100"},
            id="both_filters_travel_together_for_older_servers",
        ),
    ],
@@ -646,68 +645,6 @@ def test_list_deployments_sends_one_name_filter(call, expected_params):
    assert seen == expected_params


-@pytest.mark.parametrize(
-    "list_every",
-    [
-        pytest.param(HostBackendClient.list_deployments, id="deployments"),
-        pytest.param(HostBackendClient.list_listeners, id="listeners"),
-    ],
-)
-def test_list_endpoints_read_every_page(list_every):
-    resources = [{"id": f"resource-{index}"} for index in range(101)]
-    offsets: list[str] = []
-
-    def handler(req: httpx.Request) -> httpx.Response:
-        offset = int(req.url.params["offset"])
-        limit = int(req.url.params["limit"])
-        offsets.append(req.url.params["offset"])
-        return httpx.Response(
-            200, json={"resources": resources[offset : offset + limit]}
-        )
-
-    c = HostBackendClient(
-        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
-    )
-
-    assert list_every(c) == resources
-    assert offsets == ["0", "100"]
-
-
-FIRST_PAGE_OF_DEPLOYMENTS = [
-    {"id": f"dep-{index}", "status": "DEPLOYED"} for index in range(100)
-]
-
-
-@pytest.mark.parametrize(
-    "repeated_page",
-    [
-        pytest.param(FIRST_PAGE_OF_DEPLOYMENTS, id="same_page_again"),
-        pytest.param(
-            [
-                {"id": "dep-99", "status": "DEPLOYING"},
-                *FIRST_PAGE_OF_DEPLOYMENTS[:99],
-            ],
-            id="same_deployments_reordered_and_updated",
-        ),
-    ],
-)
-def test_list_deployments_fails_when_the_server_ignores_the_offset(repeated_page):
-    pages = iter([FIRST_PAGE_OF_DEPLOYMENTS, repeated_page])
-
-    def handler(req: httpx.Request) -> httpx.Response:
-        page = next(pages, None)
-        if page is None:
-            return httpx.Response(500, json={"detail": "the guard let paging run on"})
-        return httpx.Response(200, json={"resources": page})
-
-    c = HostBackendClient(
-        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
-    )
-
-    with pytest.raises(HostBackendError, match="ignores 'offset'"):
-        c.list_deployments()
-
-
@pytest.mark.parametrize(
    ("body", "expected"),
    [
@@ -1,6 +1,11 @@
 from unittest.mock import patch

-from langgraph_cli.deploy import Table
+from langgraph_cli.deploy import (
+    _extract_deployment_url,
+    format_deployments_table,
+    format_listeners_table,
+    format_revisions_table,
+)
 from langgraph_cli.util import clean_empty_lines, warn_non_wolfi_distro


@@ -189,7 +194,96 @@ def test_warn_non_wolfi_distro_does_not_modify_config():
    assert config_copy == original_config  # Config should remain unchanged


-def test_a_table_without_rows_still_renders_its_headers():
-    table = Table(("Name", "Status"), lambda resource: (None, None))
+def test_extract_deployment_url_uses_custom_url():
+    deployment = {"source_config": {"custom_url": "https://example.com/custom"}}
+    assert _extract_deployment_url(deployment) == "https://example.com/custom"

-    assert table.render([]).splitlines() == ["Name  Status", "----  ------"]
+
+def test_extract_deployment_url_defaults_to_dash():
+    assert _extract_deployment_url({"id": "dep-123"}) == "-"
+
+
+def test_format_deployments_table():
+    output = format_deployments_table(
+        [
+            {
+                "id": "dep-123",
+                "name": "alpha",
+                "source_config": {"custom_url": "https://alpha.example.com"},
+            },
+            {
+                "id": "dep-456",
+                "name": "beta",
+                "url": "https://beta.example.com",
+            },
+        ]
+    )
+    assert "Deployment ID" in output
+    assert "Deployment Name" in output
+    assert "Deployment URL" in output
+    assert "dep-123" in output
+    assert "alpha" in output
+    assert "https://alpha.example.com" in output
+    assert "dep-456" in output
+
+
+def test_format_revisions_table():
+    output = format_revisions_table(
+        [
+            {
+                "id": "rev-123",
+                "status": "DEPLOYED",
+                "created_at": "2023-11-09T10:00:00Z",
+            },
+            {
+                "id": "rev-456",
+                "status": "CREATING",
+                "created_at": "2023-11-07T05:31:56Z",
+            },
+            {
+                "id": "rev-789",
+                "status": "DEPLOYED",
+                "created_at": "2023-11-08T10:00:00Z",
+            },
+        ]
+    )
+    assert "Revision ID" in output
+    assert "Status" in output
+    assert "Created At" in output
+    assert "rev-123" in output
+    assert "CREATING" in output
+    assert "2023-11-07T05:31:56Z" in output
+    assert "rev-456" in output
+    assert "rev-789" in output
+    assert "REPLACED" in output
+
+
+def test_format_listeners_table():
+    output = format_listeners_table(
+        [
+            {
+                "id": "listener-1",
+                "compute_id": "prod-cluster",
+                "compute_config": {"k8s_namespaces": ["agents"]},
+            },
+            {
+                "id": "listener-2",
+                "compute_id": "multi-cluster",
+                "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
+            },
+            {
+                "id": "listener-3",
+                "compute_id": "broken-cluster",
+            },
+        ]
+    )
+    assert "Listener ID" in output
+    assert "Compute ID" in output
+    assert "Namespaces" in output
+    assert "listener-1" in output
+    assert "prod-cluster" in output
+    assert "agents" in output
+    assert "listener-2" in output
+    assert "agents, agents-staging" in output
+    assert "listener-3" in output
+    assert "broken-cluster" in output
@@ -34,6 +34,7 @@ from langgraph._internal._constants import (
    CONFIG_KEY_THREAD_ID,
    CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
    NS_SEP,
+    RETURN,
 )
 from langgraph._internal._runnable import create_task_in_config_context
 from langgraph._internal._timeout import sync_timeout_unsupported
@@ -733,15 +734,14 @@ async def arun_with_retry(
                if stream:
                    async for _ in task.proc.astream(task.input, config):
                        pass
-                    break
+                    return _return_value(task)
                return await task.proc.ainvoke(task.input, config)
            result = await _arun_with_timeout(
                task, config, resolved_timeout, attempt_ctx, stream=stream
            )
            _finish_timed_attempt(config, attempt_ctx)
            if stream:
-                # if successful, end
-                break
+                return _return_value(task)
            return result
        except ParentCommand as exc:
            ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
@@ -831,6 +831,12 @@ async def arun_with_retry(
            config = patch_configurable(config, {CONFIG_KEY_RESUMING: True})


+def _return_value(task: PregelExecutableTask) -> Any:
+    # Streaming drains the task instead of returning its result, so a caller
+    # awaiting a `@task` needs the value the task wrote.
+    return next((value for channel, value in task.writes if channel == RETURN), None)
+
+
 def _should_retry_on(retry_policy: RetryPolicy, exc: Exception) -> bool:
    """Check if the given exception should be retried based on the retry policy."""
    if isinstance(retry_policy.retry_on, Sequence):
@@ -7495,6 +7495,39 @@ async def test_entrypoint_from_async_generator() -> None:
            yield "b"


+@NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("version", ["v1", "v2"])
+async def test_awaited_task_returns_its_value_under_astream_events(
+    version: Literal["v1", "v2"],
+) -> None:
+    class State(TypedDict):
+        doubled: int
+        zeroed: int
+
+    @task
+    async def double(x: int) -> int:
+        return x * 2
+
+    @task
+    async def zero() -> int:
+        return 0
+
+    async def node(state: State) -> State:
+        return {"doubled": await double(21), "zeroed": await zero()}
+
+    graph = StateGraph(State).add_node("node", node).add_edge(START, "node").compile()
+
+    outputs = [
+        event["data"]["output"]
+        async for event in graph.astream_events(
+            {"doubled": 0, "zeroed": 1}, version=version
+        )
+        if event["event"] == "on_chain_end" and event["name"] == "node"
+    ]
+
+    assert outputs == [{"doubled": 42, "zeroed": 0}]
+
+
@NEEDS_CONTEXTVARS
 async def test_named_tasks_functional() -> None:
    class Foo:
@@ -1919,7 +1919,7 @@ requires-dist = [
    { name = "langchain-core", specifier = ">=1.4.0,<2" },
    { name = "langchain-protocol", specifier = ">=0.0.15" },
    { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14,<17" },
+    { name = "websockets", specifier = ">=14" },
 ]

 [package.metadata.requires-dev]
@@ -614,7 +614,7 @@ requires-dist = [
    { name = "langchain-core", specifier = ">=1.4.0,<2" },
    { name = "langchain-protocol", specifier = ">=0.0.15" },
    { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14,<17" },
+    { name = "websockets", specifier = ">=14" },
 ]

 [package.metadata.requires-dev]
@@ -34,6 +34,14 @@
 - **`thread.output`** — awaitable that resolves to the terminal thread state
  `values` dict after the run lifecycle completes.

+- **`Priority` (beta)** — register an async function with `@priority.assign`
+  and name the object in the `priority` section of `langgraph.json`. The
+  server calls the function with a `PriorityContext` at the first run on a
+  thread with no settings and stores the `PrioritySettings` it returns
+  (`priority`, `fairness_group`, `fairness_weight`) on the thread. Requires
+  `langgraph-api` 0.18.0 or later and a `langgraph-cli` that supports the
+  section; older versions of either ignore it.
+
 ### Changed

 - `client.threads.stream()` now accepts `transport="sse"` (default) or
@@ -2,6 +2,8 @@ from langgraph_sdk.auth import Auth
 from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
+from langgraph_sdk.priority import Priority
+from langgraph_sdk.priority.types import PriorityContext, PrioritySettings

 __version__ = "0.4.8"

@@ -10,6 +12,9 @@ __all__ = [
    "DecryptResult",
    "Encryption",
    "EncryptionContext",
+    "Priority",
+    "PriorityContext",
+    "PrioritySettings",
    "get_client",
    "get_sync_client",
 ]
@@ -0,0 +1,155 @@
+"""Custom run priority for LangGraph.
+
+.. warning::
+    This API is in beta and may change in future versions.
+
+This module lets developers decide the priority settings of each thread with
+an async function that the server calls at the first run on a thread with no
+settings. The function is registered on an object that `langgraph.json` names,
+and runs server-side.
+"""
+
+from __future__ import annotations
+
+import functools
+import inspect
+import typing
+import warnings
+from collections.abc import Awaitable, Callable
+
+from langgraph_sdk.encryption import DuplicateHandlerError, LangGraphBetaWarning
+from langgraph_sdk.priority import types
+
+_AssignHandler = Callable[
+    [types.PriorityContext], Awaitable[types.PrioritySettings | None]
+]
+_AssignHandlerT = typing.TypeVar("_AssignHandlerT", bound=_AssignHandler)
+
+
+@functools.lru_cache(maxsize=1)
+def _warn_priority_beta() -> None:
+    warnings.warn(
+        "The Priority API is in beta and may change in future versions.",
+        LangGraphBetaWarning,
+        stacklevel=4,
+    )
+
+
+def _validate_handler(fn: typing.Callable) -> None:
+    """Validate that the assign handler is async and takes one parameter.
+
+    Raises:
+        TypeError: If the handler is not an async function or has wrong parameter count
+    """
+    if not inspect.iscoroutinefunction(fn):
+        raise TypeError(
+            f"Priority assign handler must be an async function, got {type(fn)}"
+        )
+
+    sig = inspect.signature(fn)
+    params = [
+        p
+        for p in sig.parameters.values()
+        if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
+    ]
+    if len(params) != 1:
+        raise TypeError(
+            "Priority assign handler must accept exactly 1 parameter (ctx), "
+            f"got {len(params)}"
+        )
+
+
+class Priority:
+    """Decide the priority settings of your LangGraph application's threads.
+
+    .. warning::
+        This API is in beta and may change in future versions.
+
+    To use, create a separate Python file and add the path to the file to your
+    LangGraph API configuration file (`langgraph.json`). Within that file, create
+    an instance of the Priority class and register an assign function with
+    `@priority.assign`.
+
+    Example `langgraph.json` file:
+
+    ```json
+    {
+      "dependencies": ["."],
+      "graphs": {
+        "agent": "./my_agent/agent.py:graph"
+      },
+      "env": ".env",
+      "priority": {
+        "path": "./priority.py:priority"
+      }
+    }
+    ```
+
+    The server calls the assign function at the first run on a thread with no
+    settings, which can come long after the thread was created, and stores the
+    settings it returns on the thread. Later runs on the thread use the stored
+    settings. The function can be called more than once for one thread, for
+    example after it raises, and only one result is stored, so keep it free of
+    side effects.
+
+    ???+ example "Basic Usage"
+
+        ```python
+        from langgraph_sdk import Priority, PriorityContext, PrioritySettings
+
+        priority = Priority()
+
+        @priority.assign
+        async def assign(ctx: PriorityContext) -> PrioritySettings | None:
+            return {
+                "priority": 3 if "premium" in ctx.permissions else 0,
+                # Stored unencrypted, so an opaque ID rather than personal data.
+                "fairness_group": ctx.user["org_id"],
+            }
+        ```
+    """
+
+    __slots__ = ("_assign_handler",)
+
+    types = types
+    """Reference to priority type definitions.
+
+    Provides access to the type definitions used by the priority system:
+    PriorityContext and PrioritySettings.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the Priority instance."""
+        _warn_priority_beta()
+        self._assign_handler: _AssignHandler | None = None
+
+    def assign(self, fn: _AssignHandlerT) -> _AssignHandlerT:
+        """Register the function that assigns a thread's priority settings.
+
+        The function receives a `PriorityContext` and returns `PrioritySettings`,
+        or `None` for the defaults.
+
+        Example:
+            ```python
+            @priority.assign
+            async def assign(ctx: PriorityContext) -> PrioritySettings | None:
+                if ctx.cron_id is not None:
+                    return {"priority": -2}
+                return None
+            ```
+
+        Args:
+            fn: The assign handler function
+
+        Returns:
+            The registered handler function
+
+        Raises:
+            DuplicateHandlerError: If an assign handler is already registered
+            TypeError: If the handler has an invalid signature
+        """
+        if self._assign_handler is not None:
+            raise DuplicateHandlerError("Priority assign handler already registered")
+        _validate_handler(fn)
+        self._assign_handler = fn
+        return fn
@@ -0,0 +1,67 @@
+"""Priority types for LangGraph.
+
+This module defines the context the server passes to the function registered
+with `@priority.assign`, and the settings that function returns.
+"""
+
+from __future__ import annotations
+
+import typing
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from langgraph_sdk.auth.types import BaseUser
+
+
+@dataclass(frozen=True, slots=True)
+class PriorityContext:
+    """Context passed to the `@priority.assign` function at the first run on a thread with no settings."""
+
+    user: BaseUser
+    """The user from the deployment's auth, custom auth or LangSmith auth. Without auth, `user.identity` is empty."""
+
+    permissions: Sequence[str]
+    """The user's permissions from the same auth. Without auth, it's empty."""
+
+    thread_id: str | None
+    """The thread's ID, or `None` when the run creates its thread without being given an ID."""
+
+    thread_metadata: dict[str, typing.Any]
+    """The thread's metadata."""
+
+    assistant_id: str
+    """The ID of the assistant the run uses."""
+
+    graph_id: str
+    """The ID of the graph the assistant runs."""
+
+    assistant_metadata: dict[str, typing.Any]
+    """The assistant's metadata."""
+
+    run_metadata: dict[str, typing.Any]
+    """The run's metadata."""
+
+    cron_id: str | None
+    """The ID of the cron that started the run, or `None` when no cron started it."""
+
+
+class PrioritySettings(typing.TypedDict, total=False):
+    """Priority settings returned by the `@priority.assign` function.
+
+    Every key is optional: a missing key or a key set to `None` takes its
+    default, and returning `None` instead of a dict gives every key its default.
+    A value outside its range, or an empty `fairness_group`, is an error.
+    """
+
+    priority: int | None
+    """An integer from -5 to 5, where 5 runs first and -5 runs last. Defaults to 0."""
+
+    fairness_group: str | None
+    """The thread's fairness group, a non-empty string. Defaults to `"default"`.
+
+    The server stores it unencrypted, even when the deployment uses custom
+    encryption, so use an opaque ID rather than personal data.
+    """
+
+    fairness_weight: int | None
+    """The weight of the thread's fairness group, an integer from 1 to 100. Defaults to 1."""
@@ -16,7 +16,7 @@ dependencies = [
    "orjson>=3.11.5",
    "langchain-protocol>=0.0.15",
    "langchain-core>=1.4.0,<2",
-    "websockets>=14,<17",
+    "websockets>=14",
 ]

 [tool.hatch.version]
@@ -0,0 +1,128 @@
+import dataclasses
+
+import pytest
+
+import langgraph_sdk
+from langgraph_sdk import Priority, PriorityContext, PrioritySettings
+from langgraph_sdk.encryption import DuplicateHandlerError, LangGraphBetaWarning
+from langgraph_sdk.priority import _warn_priority_beta
+
+
+class _User:
+    is_authenticated = True
+    display_name = "user-1"
+    identity = "user-1"
+    permissions: tuple[str, ...] = ()
+
+    def __getitem__(self, key):
+        return getattr(self, key)
+
+    def __contains__(self, key):
+        return hasattr(self, key)
+
+    def __iter__(self):
+        return iter(("identity", "display_name", "is_authenticated", "permissions"))
+
+
+def _context(thread_id: str | None) -> PriorityContext:
+    return PriorityContext(
+        user=_User(),
+        permissions=[],
+        thread_id=thread_id,
+        thread_metadata={},
+        assistant_id="assistant-1",
+        graph_id="agent",
+        assistant_metadata={"tier": "gold"},
+        run_metadata={},
+        cron_id=None,
+    )
+
+
+async def _no_params() -> PrioritySettings | None:
+    return None
+
+
+async def _two_params(_ctx: PriorityContext, _extra: object) -> PrioritySettings | None:
+    return None
+
+
+def test_priority_and_its_types_are_exported():
+    assert {"Priority", "PriorityContext", "PrioritySettings"} <= set(
+        langgraph_sdk.__all__
+    )
+    assert Priority.types.PriorityContext is PriorityContext
+    assert Priority.types.PrioritySettings is PrioritySettings
+
+
+def test_construction_warns_that_priority_is_beta():
+    # The warning fires once per process, so an earlier test may have used it up.
+    _warn_priority_beta.cache_clear()
+    with pytest.warns(LangGraphBetaWarning, match="Priority API is in beta"):
+        Priority()
+
+
+def test_assign_registers_and_returns_the_function():
+    priority = Priority()
+
+    async def assign(_ctx: PriorityContext) -> PrioritySettings | None:
+        return {"priority": 3, "fairness_group": "org-42"}
+
+    assert priority.assign(assign) is assign
+    assert priority._assign_handler is assign
+
+
+def test_assign_rejects_a_sync_function():
+    priority = Priority()
+
+    def assign(_ctx: PriorityContext) -> PrioritySettings | None:
+        return None
+
+    with pytest.raises(TypeError, match="must be an async function"):
+        priority.assign(assign)  # ty: ignore[invalid-argument-type]
+    assert priority._assign_handler is None
+
+
+@pytest.mark.parametrize("fn", [_no_params, _two_params])
+def test_assign_rejects_the_wrong_number_of_parameters(fn):
+    priority = Priority()
+
+    with pytest.raises(TypeError, match="must accept exactly 1 parameter"):
+        priority.assign(fn)
+    assert priority._assign_handler is None
+
+
+def test_assign_rejects_a_second_function():
+    priority = Priority()
+
+    @priority.assign
+    async def first(_ctx: PriorityContext) -> PrioritySettings | None:
+        return None
+
+    with pytest.raises(DuplicateHandlerError):
+
+        @priority.assign
+        async def second(_ctx: PriorityContext) -> PrioritySettings | None:
+            return None
+
+    assert priority._assign_handler is first
+
+
+def test_context_accepts_no_thread_id():
+    # `make lint` type-checks this call too, so `thread_id` must accept None there.
+    assert _context(thread_id=None).thread_id is None
+
+
+def test_context_is_frozen():
+    ctx = _context(thread_id="thread-1")
+
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        ctx.thread_id = "thread-2"  # ty: ignore[invalid-assignment]
+
+
+def test_every_settings_key_is_optional():
+    assert PrioritySettings.__optional_keys__ == {
+        "priority",
+        "fairness_group",
+        "fairness_weight",
+    }
+    assert PrioritySettings.__required_keys__ == frozenset()
@@ -525,7 +525,7 @@ requires-dist = [
    { name = "langchain-core", specifier = ">=1.4.0,<2" },
    { name = "langchain-protocol", specifier = ">=0.0.15" },
    { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14,<17" },
+    { name = "websockets", specifier = ">=14" },
 ]

 [package.metadata.requires-dev]
Author	SHA1	Message	Date
Elior Nataf Lackritz	5527cc166e	fix(langgraph): return an awaited @task's value while streaming With stream=True, arun_with_retry drains the task's astream() and returned nothing, so a @task awaited inside a node got None whenever the node ran in streaming mode: under astream_events (v1 and v2) and anything built on it. Return the value the task wrote instead, on both stream paths. Same idea as #9295 by AaronG18 and the lookup amishaguptaberk suggested.	2026-10-23 07:24:18 -04:00
ataller-lc	9d68e5b3eb	feat(cli): add the priority section to langgraph.json (#9393 ) `langgraph.json` gets a `priority` section that names a `Priority` object from `langgraph_sdk` (#9392), as `auth` and `encryption` name theirs: ```json "priority": {"path": "./priority.py:priority"} ``` - The path must be `file.py:name` or `module:name`, as for `encryption`. It must point to Python: JavaScript and TypeScript files are rejected. A Node.js-only project can still name a Python file, since the callback runs in the Python server, as `encryption` does. - Image builds rewrite the path to its location in the image, for both `dependencies`-based and `uv`-source projects, and write `ENV LANGGRAPH_PRIORITY` into the Dockerfile. - `langgraph dev` passes `priority` to the server. A server that doesn't read the section ignores it. - `PriorityConfig` is in the JSON schema (`make update-schema`). ## Test plan - [x] New unit tests next to the `encryption` ones: `LANGGRAPH_PRIORITY` with the rewritten path from both builders, a path outside `dependencies`, a malformed path, the six JS and TS extensions, a Node.js-only project with a Python path, the section kept by validation and not reported as unknown, and `langgraph dev` passing it to the server. Each one fails when the change it covers is removed. - [x] `make test` in `libs/cli`: 586 passed. `make lint` clean, and `make update-schema` leaves no diff.	2026-10-22 17:26:01 -04:00
ataller-lc	544386f6cc	feat(sdk-py): add Priority for thread priority settings (#9392 ) `Priority` lets a deployment decide each thread's run priority in its own code, declared the way `Auth` and `Encryption` are: an object in the project, named by a `priority` section in `langgraph.json`. It's beta, and construction warns like `Encryption()` does. ```python from langgraph_sdk import Priority, PriorityContext, PrioritySettings priority = Priority() @priority.assign async def assign(ctx: PriorityContext) -> PrioritySettings \| None: return {"priority": 3 if "premium" in ctx.permissions else 0} ``` - The server calls the function at the first run on a thread with no settings and stores what it returns on the thread. Later runs on the thread use the stored settings. - `PriorityContext` is a frozen dataclass: the user and their permissions, the thread's ID (`None` when the run creates its thread without one) and metadata, the assistant and graph IDs, the assistant's and the run's metadata, and the cron ID when a cron started the run. - `PrioritySettings` has three optional keys: `priority` (-5 to 5, 5 runs first), `fairness_group` (a non-empty string, stored unencrypted, so an opaque ID rather than personal data) and `fairness_weight` (1 to 100). A missing key, `None`, or a `None` return takes the default: 0, `"default"`, 1. - `@priority.assign` takes one async function with one parameter and raises on a second registration. It reuses `DuplicateHandlerError` and `LangGraphBetaWarning` from `langgraph_sdk.encryption`. - `Priority`, `PriorityContext` and `PrioritySettings` are exported from `langgraph_sdk`; the types are also on `Priority.types`. ## Test plan - [x] New unit tests in `tests/test_priority.py`: registration and its checks, the exports, a frozen context that accepts `thread_id=None`, every settings key optional, the beta warning. Each one fails when the check it covers is removed. - [x] `make test` in `libs/sdk-py`: 561 passed. `make lint` clean, `check_sdk_methods.py` passes.	2026-10-22 17:25:38 -04:00
Mason Daugherty Mason Daugherty open-swe[bot]	c2df8ab04b	chore(deps): remove precautionary third-party upper bounds (#9390 ) Remove only the precautionary `websockets<17` ceiling from `langgraph-sdk`, keeping `websockets>=14` and every first-party dependency range unchanged. Refresh mirrored lock metadata without changing resolved versions, Python support, or development/build constraints. AI assistance: implemented with Open SWE. ## References - [Plan](https://openswe.langchain.dev/agents/8ca5d7fe-ecc3-5d70-9b14-d2904a2c0948/plan) - [Slack thread](https://langchain.slack.com/archives/C0C5950ARJT/p1792630605250939) Made by [Open SWE](https://github.com/langchain-ai/open-swe) · [view thread](https://openswe.langchain.dev/agents/8ca5d7fe-ecc3-5d70-9b14-d2904a2c0948) · openai:gpt-6.1-sol (medium) --------- Co-authored-by: Mason Daugherty <mdrxy@users.noreply.github.com> Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>	2026-10-22 16:27:58 -04:00