fix: order delta channel replay by task path

DeltaChannel reconstructs its value by replaying ancestor writes through
the reducer. Every saver ordered a checkpoint's writes by (task_id, idx),
but live execution applies them in task-path order: apply_writes sorts a
super-step's tasks by task_path_str(task.path[:3]) before calling
channel.update. task_id is a hash of the path, so the two orders are
unrelated, and two or more tasks writing one DeltaChannel in a single
super-step replayed in an arbitrary permutation.

Reducers are only required to be batching-invariant, not order-invariant,
so the permutation changes the value: get_state disagreed with what invoke
returned, and continuing the thread persisted the reordered replay as the
base for later writes.

Replay now orders by (task_path, task_id, idx), following the precedent
already set for the Send channel by SELECT_PENDING_SENDS_SQL. InMemorySaver
and the postgres savers already persisted task_path and only needed the
sort key; sqlite accepted task_path on put_writes and dropped it, so the
writes table gains the column, added by setup() to databases created by
earlier versions.

Writes stored without a task_path sort first within their checkpoint, which
is where live execution applies the task-less input writes that carry "".

Co-authored-by: ErenAta16 <149434812+ErenAta16@users.noreply.github.com>
Co-authored-by: ragnarok268 <58264829+ragnarok268@users.noreply.github.com>

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_history.py

@@ -267,6 +267,61 @@ async def test_history_seed_ancestor_own_writes_are_replayed(
     )
 
 
+# Every uuid4 `build_delta_chain` tags its own writes with sorts between these
+# two, so task_id order is fixed and always disagrees with task_path order.
+TASK_ID_SORTS_FIRST = "00000000-0000-0000-0000-000000000000"
+TASK_ID_SORTS_LAST = "zzzzzzzz-0000-0000-0000-000000000000"
+
+
+async def test_history_orders_parallel_writes_by_task_path(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Writes from parallel tasks replay in task_path order, not task_id order."""
+    configs = await build_delta_chain(
+        saver,
+        thread_id=str(uuid4()),
+        channel="ch",
+        snapshots_at_steps=[0],
+        total_steps=3,
+    )
+    step_1, head = configs[1], configs[2]
+    await saver.aput_writes(
+        step_1, [("ch", "second")], TASK_ID_SORTS_FIRST, "~pull, 02"
+    )
+    await saver.aput_writes(step_1, [("ch", "first")], TASK_ID_SORTS_LAST, "~pull, 01")
+
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    values = [w[2] for w in result["ch"]["writes"]]
+    assert values == [1, "first", "second"], (
+        f"Expected task_path order [1, 'first', 'second'], got {values}. "
+        "Ordering by (task_id, idx) alone yields [1, 'second', 'first']."
+    )
+
+
+async def test_history_orders_pathless_writes_first(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Writes stored without a task_path (graph input) replay before task writes."""
+    configs = await build_delta_chain(
+        saver,
+        thread_id=str(uuid4()),
+        channel="ch",
+        snapshots_at_steps=[0],
+        total_steps=3,
+    )
+    step_1, head = configs[1], configs[2]
+    await saver.aput_writes(
+        step_1, [("ch", "from_node")], TASK_ID_SORTS_FIRST, "~pull, a"
+    )
+    await saver.aput_writes(step_1, [("ch", "from_input")], TASK_ID_SORTS_LAST)
+
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    values = [w[2] for w in result["ch"]["writes"]]
+    assert values == [1, "from_input", "from_node"], (
+        f"Expected pathless writes first, got {values}"
+    )
+
+
 ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_returns_writes_oldest_first,
     test_history_seed_is_nearest_snapshot,
@@ -276,6 +331,8 @@ ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_walk_to_root_no_seed,
     test_history_migration_plain_value_as_seed,
     test_history_seed_ancestor_own_writes_are_replayed,
+    test_history_orders_parallel_writes_by_task_path,
+    test_history_orders_pathless_writes_first,
 ]
 
 

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -168,6 +168,7 @@ class _DeltaStage2Row(TypedDict, total=False):
     type: str | None
     blob: bytes | None
     task_id: str | None  # "w" rows only
+    task_path: str | None  # "w" rows only
     idx: int | None  # "w" rows only
     version: str | None  # "b" rows only
 
@@ -319,7 +320,7 @@ def _build_delta_stage2_sql(
         branches.append(
             "SELECT 'w'::text AS _kind, "
             "checkpoint_id, channel, "
-            "type, blob, task_id, idx, NULL::text AS version "
+            "type, blob, task_id, task_path, idx, NULL::text AS version "
             "FROM checkpoint_writes "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND checkpoint_id = ANY(%s)"
@@ -327,7 +328,8 @@ def _build_delta_stage2_sql(
     for _ in channels_with_seed:
         branches.append(
             "SELECT 'b'::text AS _kind, NULL::text AS checkpoint_id, channel, "
-            "type, blob, NULL::text AS task_id, NULL::int AS idx, version "
+            "type, blob, NULL::text AS task_id, NULL::text AS task_path, "
+            "NULL::int AS idx, version "
             "FROM checkpoint_blobs "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND version = %s"
@@ -492,10 +494,11 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         stored value, or when the seed blob is sentinel "empty" — in both cases
         the consumer treats absence as "start empty".
         """
-        # writes_by_ch_by_cid[channel][cid] = list of (type, blob, task_id, idx)
-        writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
-            ch: {} for ch in channels
-        }
+        # writes_by_ch_by_cid[channel][cid] = list of
+        # (type, blob, task_id, idx, task_path)
+        writes_by_ch_by_cid: dict[
+            str, dict[str, list[tuple[str, bytes, str, int, str]]]
+        ] = {ch: {} for ch in channels}
         # seed_blob_by_ver[(channel, version)] = (type, blob)
         seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
 
@@ -506,8 +509,14 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                 cid = cast(str, r["checkpoint_id"])
                 writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
                     cast(
-                        "tuple[str, bytes, str, int]",
-                        (r["type"], r["blob"], r["task_id"], r["idx"]),
+                        "tuple[str, bytes, str, int, str]",
+                        (
+                            r["type"],
+                            r["blob"],
+                            r["task_id"],
+                            r["idx"],
+                            r["task_path"],
+                        ),
                     )
                 )
             else:  # kind == "b"
@@ -516,10 +525,10 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                     "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-        # Sort writes per (channel, cid) newest-first by (task_id, idx)
+        # Sort writes per (channel, cid) newest-first by (task_path, task_id, idx)
         for cid_map in writes_by_ch_by_cid.values():
             for ws in cid_map.values():
-                ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
+                ws.sort(key=lambda w: (w[4], w[2], w[3]), reverse=True)
 
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
@@ -529,7 +538,9 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             collected: list[PendingWrite] = []
             cid_writes = writes_by_ch_by_cid.get(ch, {})
             for cid in chain_cids:
-                for type_tag, write_blob, task_id, _idx in cid_writes.get(cid, []):
+                for type_tag, write_blob, task_id, _idx, _path in cid_writes.get(
+                    cid, []
+                ):
                     val = self.serde.loads_typed((type_tag, write_blob))
                     collected.append((task_id, ch, val))
             collected.reverse()

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

@@ -154,6 +154,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                 checkpoint_ns TEXT NOT NULL DEFAULT '',
                 checkpoint_id TEXT NOT NULL,
                 task_id TEXT NOT NULL,
+                task_path TEXT NOT NULL DEFAULT '',
                 idx INTEGER NOT NULL,
                 channel TEXT NOT NULL,
                 type TEXT,
@@ -162,6 +163,15 @@ class SqliteSaver(BaseCheckpointSaver[str]):
             );
             """
         )
+        # sqlite has no ADD COLUMN IF NOT EXISTS; this migrates databases
+        # created before `task_path` existed and is a no-op on the rest.
+        try:
+            self.conn.execute(
+                "ALTER TABLE writes ADD COLUMN task_path TEXT NOT NULL DEFAULT ''"
+            )
+        except sqlite3.OperationalError as e:
+            if "duplicate column name" not in str(e):
+                raise
 
         self.is_setup = True
 
@@ -460,9 +470,9 @@ class SqliteSaver(BaseCheckpointSaver[str]):
             task_path: Path of the task creating the writes.
         """
         query = (
-            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
             if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
         )
         with self.cursor() as cur:
             cur.executemany(
@@ -473,6 +483,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                         str(config["configurable"]["checkpoint_ns"]),
                         str(config["configurable"]["checkpoint_id"]),
                         task_id,
+                        task_path,
                         WRITES_IDX_MAP.get(channel, idx),
                         channel,
                         *self.serde.dumps_typed(value),
@@ -568,7 +579,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                     )
                 cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes]]", cur.fetchall()
+                    "list[tuple[str, str, str, int, str, bytes, str]]", cur.fetchall()
                 )
             else:
                 stage2_rows = []

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/_delta.py

@@ -57,7 +57,7 @@ def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
     for n in chain_lens:
         cid_placeholders = ",".join("?" * n)
         branches.append(
-            "SELECT checkpoint_id, channel, task_id, idx, type, value "
+            "SELECT checkpoint_id, channel, task_id, idx, type, value, task_path "
             "FROM writes "
             "WHERE thread_id = ? AND checkpoint_ns = ? AND channel = ? "
             f"AND checkpoint_id IN ({cid_placeholders})"
@@ -130,29 +130,31 @@ def build_delta_channels_writes_history(
     chain_by_ch: Mapping[str, list[str]],
     seed_val_by_ch: Mapping[str, Any],
     seeded: set[str],
-    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes]],
+    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes, str]],
     serde: Any,
 ) -> dict[str, DeltaChannelHistory]:
     """Demux stage-2 rows per channel; produce per-channel histories.
 
-    Stage-2 rows are `(checkpoint_id, channel, task_id, idx, type, value)`.
-    Final write order is oldest→newest globally and `(task_id, idx)` within
-    a checkpoint, matching the contract on `DeltaChannelHistory.writes`.
+    Stage-2 rows are
+    `(checkpoint_id, channel, task_id, idx, type, value, task_path)`.
+    Final write order is oldest→newest globally and
+    `(task_path, task_id, idx)` within a checkpoint, matching the contract
+    on `DeltaChannelHistory.writes`.
 
     `seed` is omitted when the walk reached a true root with no snapshot
     found (channel never entered `seeded`); consumers treat absence as
     "start empty".
     """
-    writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
-        ch: {} for ch in channels
-    }
-    for cid, ch, task_id, idx, type_tag, value_blob in stage2_rows:
+    writes_by_ch_by_cid: dict[
+        str, dict[str, list[tuple[str, bytes, str, int, str]]]
+    ] = {ch: {} for ch in channels}
+    for cid, ch, task_id, idx, type_tag, value_blob, task_path in stage2_rows:
         writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
-            (type_tag, value_blob, task_id, idx)
+            (type_tag, value_blob, task_id, idx, task_path)
         )
     for cid_map in writes_by_ch_by_cid.values():
         for ws in cid_map.values():
-            ws.sort(key=lambda w: (w[2], w[3]))
+            ws.sort(key=lambda w: (w[4], w[2], w[3]))
 
     result: dict[str, DeltaChannelHistory] = {}
     for ch in channels:
@@ -161,7 +163,7 @@ def build_delta_channels_writes_history(
         collected: list[PendingWrite] = []
         # Chain is newest-first; iterate oldest-first for the public order.
         for cid in reversed(chain_cids):
-            for type_tag, value_blob, task_id, _idx in cid_writes.get(cid, []):
+            for type_tag, value_blob, task_id, _idx, _path in cid_writes.get(cid, []):
                 collected.append(
                     (task_id, ch, serde.loads_typed((type_tag, value_blob)))
                 )

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

@@ -331,6 +331,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                     checkpoint_ns TEXT NOT NULL DEFAULT '',
                     checkpoint_id TEXT NOT NULL,
                     task_id TEXT NOT NULL,
+                    task_path TEXT NOT NULL DEFAULT '',
                     idx INTEGER NOT NULL,
                     channel TEXT NOT NULL,
                     type TEXT,
@@ -341,6 +342,17 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             ):
                 await self.conn.commit()
 
+            # sqlite has no ADD COLUMN IF NOT EXISTS; this migrates databases
+            # created before `task_path` existed and is a no-op on the rest.
+            try:
+                await self.conn.execute(
+                    "ALTER TABLE writes ADD COLUMN task_path TEXT NOT NULL DEFAULT ''"
+                )
+                await self.conn.commit()
+            except aiosqlite.OperationalError as e:
+                if "duplicate column name" not in str(e):
+                    raise
+
             self.is_setup = True
 
     async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
@@ -576,9 +588,9 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             task_path: Path of the task creating the writes.
         """
         query = (
-            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
             if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
         )
         await self.setup()
         async with self.lock, self.conn.cursor() as cur:
@@ -590,6 +602,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                         str(config["configurable"]["checkpoint_ns"]),
                         str(config["configurable"]["checkpoint_id"]),
                         task_id,
+                        task_path,
                         WRITES_IDX_MAP.get(channel, idx),
                         channel,
                         *self.serde.dumps_typed(value),
@@ -681,7 +694,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                     )
                 await cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes]]",
+                    "list[tuple[str, str, str, int, str, bytes, str]]",
                     await cur.fetchall(),
                 )
             else:

libs/checkpoint-sqlite/tests/test_writes_task_path_migration.py

@@ -0,0 +1,87 @@
+import sqlite3
+from pathlib import Path
+
+import aiosqlite
+import pytest
+from langgraph.checkpoint.base import empty_checkpoint
+
+from langgraph.checkpoint.sqlite import SqliteSaver
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+WRITES_BEFORE_TASK_PATH = """
+CREATE TABLE writes (
+    thread_id TEXT NOT NULL,
+    checkpoint_ns TEXT NOT NULL DEFAULT '',
+    checkpoint_id TEXT NOT NULL,
+    task_id TEXT NOT NULL,
+    idx INTEGER NOT NULL,
+    channel TEXT NOT NULL,
+    type TEXT,
+    value BLOB,
+    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
+);
+INSERT INTO writes VALUES ('t', '', 'c', 'old-task', 0, 'ch', 'null', X'');
+"""
+
+
+@pytest.fixture
+def legacy_db(tmp_path: Path) -> Path:
+    db = tmp_path / "legacy.sqlite"
+    with sqlite3.connect(db) as conn:
+        conn.executescript(WRITES_BEFORE_TASK_PATH)
+    return db
+
+
+def test_setup_migrates_legacy_writes_table_repeatably(legacy_db: Path) -> None:
+    for _ in range(2):
+        with SqliteSaver.from_conn_string(str(legacy_db)) as saver:
+            saver.setup()
+            rows = saver.conn.execute(
+                "SELECT task_id, task_path FROM writes"
+            ).fetchall()
+            assert rows == [("old-task", "")]
+
+
+@pytest.mark.parametrize("fresh", [True, False], ids=["fresh", "legacy"])
+def test_put_writes_persists_task_path(
+    tmp_path: Path, legacy_db: Path, fresh: bool
+) -> None:
+    db = tmp_path / "fresh.sqlite" if fresh else legacy_db
+    with SqliteSaver.from_conn_string(str(db)) as saver:
+        config = saver.put(
+            {"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
+            empty_checkpoint(),
+            {},
+            {},
+        )
+        saver.put_writes(config, [("ch", "v")], "task-1", "~__pregel_pull, node")
+        stored = saver.conn.execute(
+            "SELECT task_path FROM writes WHERE task_id = 'task-1'"
+        ).fetchall()
+    assert stored == [("~__pregel_pull, node",)]
+
+
+async def test_async_setup_migrates_legacy_writes_table_repeatably(
+    legacy_db: Path,
+) -> None:
+    for _ in range(2):
+        async with AsyncSqliteSaver.from_conn_string(str(legacy_db)) as saver:
+            await saver.setup()
+            config = await saver.aput(
+                {"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
+                empty_checkpoint(),
+                {},
+                {},
+            )
+            await saver.aput_writes(
+                config, [("ch", "v")], "task-1", "~__pregel_pull, node"
+            )
+
+    async with aiosqlite.connect(legacy_db) as conn:
+        async with conn.execute(
+            "SELECT DISTINCT task_id, task_path FROM writes ORDER BY task_id"
+        ) as cur:
+            assert await cur.fetchall() == [
+                ("old-task", ""),
+                ("task-1", "~__pregel_pull, node"),
+            ]

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -162,6 +162,13 @@ class DeltaChannelHistory(TypedDict):
       Always present; possibly empty. Already filtered to one channel.
       Writes stored at the target checkpoint itself are pending for the
       next super-step and are excluded.
+
+      Within a single checkpoint, writes are ordered by
+      `(task_path, task_id, idx)`: the order `apply_writes` applied them in
+      live. `task_id` is a hash of the path, so ordering by it permutes
+      parallel tasks writing one channel, and reducers need not be
+      order-invariant. Writes stored without a `task_path` (graph input, or
+      rows predating the column) sort first.
     * `seed` — the stored value at the nearest ancestor whose
       `channel_values[ch]` is populated. Omitted if the walk reached the
       root without finding any stored value (consumer treats absence as
@@ -611,6 +618,11 @@ class BaseCheckpointSaver(Generic[V]):
         `PostgresSaver`) override for performance; the return contract is
         fixed here.
 
+        `PendingWrite` carries no `task_path`, so this default replays each
+        checkpoint's writes in `get_tuple`'s `pending_writes` order. Savers
+        that do not return `pending_writes` ordered by
+        `(task_path, task_id, idx)` must override it.
+
         Args:
             config: Configuration identifying the target checkpoint.
             channels: Channel names to walk for. Empty → empty mapping.

libs/checkpoint/langgraph/checkpoint/memory/__init__.py

@@ -199,8 +199,8 @@ class InMemorySaver(
                     terminated_here.add(ch)
 
             step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
-            for (_task_id, _idx), (tid, ch, serialized, _) in sorted(
-                step_writes.items(), reverse=True
+            for _, (tid, ch, serialized, _) in sorted(
+                step_writes.items(), key=lambda kv: (kv[1][3], kv[0]), reverse=True
             ):
                 if ch not in remaining:
                     continue

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.32.dev0"
+__version__ = "0.4.31"

libs/cli/langgraph_cli/deploy.py

@@ -26,7 +26,6 @@ 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,
@@ -102,16 +101,16 @@ _NATIVE_AMD64_MACHINE = "x86_64"
 _PUSH_ATTEMPTS = 3
 _LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp"
 _OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {}
-_CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker"
 _LISTENER_REQUIRED_MARKER = "listener_id' is required"
-_LISTENERS_SHOWN = 10
-_LISTENER_NOT_FOUND_STATUSES = frozenset({404, 422})
-_LISTENERS_DOCS_URL = "https://docs.langchain.com/langsmith/control-plane#listeners"
-_NO_LISTENERS = (
-    "This workspace has no listeners, so --listener-id and --k8s-namespace "
-    "do not apply."
+_HYBRID_LISTENER_GUIDANCE = (
+    "This workspace deploys through a listener in your own cluster, and the "
+    "control plane needs a listener ID to create a deployment. Create the "
+    "deployment once in the LangSmith UI, choosing the listener and namespace, "
+    "then re-run with --deployment-id <id>."
 )
 
+_CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker"
+
 
 _TERMINAL_STATUSES = frozenset(
     [
@@ -162,134 +161,6 @@ class ByAgent:
 DeploymentSelector = ById | ByName | ByAgent
 
 
-@dataclass(frozen=True, slots=True)
-class Listener:
-    id: str
-    compute_id: str
-    namespaces: tuple[str, ...]
-
-    @classmethod
-    def from_resource(cls, resource: Mapping[str, object]) -> "Listener":
-        identifier = str(resource.get("id") or "")
-        if not identifier:
-            raise HostBackendError(
-                "The control plane returned a listener without an id."
-            )
-        compute_config = resource.get("compute_config")
-        namespaces = (
-            compute_config.get("k8s_namespaces")
-            if isinstance(compute_config, Mapping)
-            else None
-        )
-        return cls(
-            identifier,
-            str(resource.get("compute_id", "")),
-            tuple(str(namespace) for namespace in namespaces)
-            if isinstance(namespaces, list)
-            else (),
-        )
-
-
-@dataclass(frozen=True, slots=True)
-class Unplaced:
-    @property
-    def summary(self) -> str:
-        return ""
-
-    def source_config(self) -> dict[str, object]:
-        return {}
-
-
-@dataclass(frozen=True, slots=True)
-class OnListener:
-    listener_id: str
-    k8s_namespace: str
-
-    @property
-    def summary(self) -> str:
-        return (
-            f"Deploying through listener {self.listener_id} "
-            f"in namespace {self.k8s_namespace}"
-        )
-
-    def source_config(self) -> dict[str, object]:
-        return {
-            "listener_id": self.listener_id,
-            "listener_config": {"k8s_namespace": self.k8s_namespace},
-        }
-
-
-Placement = Unplaced | OnListener
-
-
-@dataclass(frozen=True, slots=True)
-class RequestedPlacement:
-    listener_id: str | None = None
-    k8s_namespace: str | None = None
-
-    @property
-    def requested(self) -> bool:
-        return self.listener_id is not None or self.k8s_namespace is not None
-
-    def ensure_not_requested(self, deployment_id: str) -> None:
-        if self.requested:
-            raise click.UsageError(
-                "Listener and namespace are fixed when a deployment is created. "
-                f"Deployment {deployment_id} already exists, so drop --listener-id "
-                "and --k8s-namespace, or create a new deployment with a different "
-                "--name."
-            )
-
-    def on(self, listener: Listener) -> Placement:
-        return OnListener(listener.id, self._namespace(listener))
-
-    def among(self, listeners: Sequence[Listener]) -> Placement:
-        if not listeners:
-            if self.requested:
-                raise click.UsageError(_NO_LISTENERS)
-            return Unplaced()
-        if len(listeners) > 1:
-            raise click.UsageError(
-                "This workspace has several listeners. Choose one with "
-                f"--listener-id:\n{_describe_listeners(listeners)}"
-            )
-        return self.on(listeners[0])
-
-    def _namespace(self, listener: Listener) -> str:
-        if not listener.namespaces:
-            raise click.UsageError(
-                f"Listener {listener.id} serves no namespaces. Check its configuration."
-            )
-        if self.k8s_namespace is None:
-            if len(listener.namespaces) == 1:
-                return listener.namespaces[0]
-            raise click.UsageError(
-                f"Listener {listener.id} serves several namespaces. Choose one with "
-                f"--k8s-namespace: {', '.join(listener.namespaces)}"
-            )
-        if self.k8s_namespace not in listener.namespaces:
-            raise click.UsageError(
-                f"Listener {listener.id} does not serve namespace "
-                f"'{self.k8s_namespace}'. Choose one of: "
-                f"{', '.join(listener.namespaces)}"
-            )
-        return self.k8s_namespace
-
-
-def _describe_listeners(listeners: Sequence[Listener]) -> str:
-    shown = listeners[:_LISTENERS_SHOWN]
-    lines = [
-        f"  {listener.id}  cluster {listener.compute_id}  "
-        f"namespaces: {', '.join(listener.namespaces)}"
-        for listener in shown
-    ]
-    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)
-
-
 @dataclass(frozen=True, slots=True)
 class ExistingDeployment:
     id: str
@@ -508,16 +379,15 @@ def _source_of(resource: object) -> str | None:
 def find_deployment_by_name(
     client: HostBackendClient, name: str
 ) -> ExistingDeployment | None:
-    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"):
+    listed = client.list_deployments(name_contains=name)
+    resources = listed.get("resources", []) if isinstance(listed, dict) else []
+    for resource in resources:
+        if (
+            isinstance(resource, dict)
+            and 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
 
 
@@ -805,46 +675,6 @@ def _find_deployment(
     selector: ByName | ByAgent,
     *,
     not_found_message: str,
-<<<<<<< HEAD
-    agent: dict[str, str] | None = None,
-) -> tuple[str | None, bool, int]:
-    """Resolve an existing deployment by ID or exact name match."""
-    needs_creation = False
-    if deployment_id:
-        _log_deploy_step(step, f"Using deployment {deployment_id}")
-        _call_host_backend_with_optional_tenant(
-            client, lambda c: c.get_deployment(deployment_id)
-        )
-        return deployment_id, needs_creation, step + 1
-
-    if agent is not None:
-        _log_deploy_step(
-            step, f"Looking up agent '{agent['agent_id']}' in {agent['environment']}"
-        )
-        existing = _call_host_backend_with_optional_tenant(
-            client,
-            lambda c: c.list_deployments(
-                agent_id=agent["agent_id"], agent_environment=agent["environment"]
-            ),
-        )
-        found_id = next(
-            (
-                dep["id"]
-                for dep in existing.get("resources", [])
-                if not dep.get("is_preview")
-            ),
-            None,
-        )
-    else:
-        _log_deploy_step(step, f"Looking up deployment '{name}'")
-        found_id = _call_host_backend_with_optional_tenant(
-            client, lambda c: find_deployment_id_by_name(c, name)
-        )
-    em = _get_emitter()
-    if found_id:
-        deployment_id = str(found_id)
-        em.info(f"Found existing deployment (ID: {deployment_id})")
-=======
 ) -> tuple[ExistingDeployment | None, int]:
     if isinstance(selector, ByAgent):
         _log_deploy_step(
@@ -853,26 +683,17 @@ def _find_deployment(
         existing = _call_host_backend_with_optional_tenant(
             client,
             lambda c: c.list_deployments(
-                agent_id=selector.agent_id,
-                agent_environment=selector.environment,
-                limit=MAX_PAGE_SIZE,
+                agent_id=selector.agent_id, agent_environment=selector.environment
             ),
         )
-        if len(existing) > 1:
-            raise click.ClickException(
-                "This control plane does not filter deployments by agent, so the "
-                f"CLI cannot tell which one belongs to '{selector.agent_id}' in "
-                f"{selector.environment}. Deploy by --name instead."
-            )
         found = next(
             (
                 ExistingDeployment(str(dep["id"]), _source_of(dep))
-                for dep in existing
-                if dep.get("id") and not dep.get("is_preview")
+                for dep in existing.get("resources", [])
+                if not dep.get("is_preview")
             ),
             None,
         )
->>>>>>> origin
     else:
         _log_deploy_step(step, f"Looking up deployment '{selector.name}'")
         found = _call_host_backend_with_optional_tenant(
@@ -897,22 +718,12 @@ def _create_deployment(
     step: int,
     *,
     name: str | None,
-<<<<<<< HEAD
-    deployment_type: str,
-    source: str,
-    config_rel: str | None = None,
-    secrets: list[dict[str, str]] | None = None,
-    agent: dict[str, str] | None = None,
-) -> tuple[str, int]:
-    """Create a deployment and return its ID and next step number."""
-=======
     source: str,
     source_config: dict[str, object],
     source_revision_config: dict[str, object],
     secrets: list[dict[str, str]],
     agent: dict[str, str] | None = None,
 ) -> tuple[CreatedDeployment, int]:
->>>>>>> origin
     _log_deploy_step(
         step,
         f"Creating deployment for agent '{agent['agent_id']}' in {agent['environment']}"
@@ -922,15 +733,9 @@ def _create_deployment(
     try:
         created = client.create_deployment(
             name=name,
-<<<<<<< HEAD
-            deployment_type=deployment_type,
-            source=source,
-            config_path=config_rel,
-=======
             source=source,
             source_config=source_config,
             source_revision_config=source_revision_config,
->>>>>>> origin
             secrets=secrets,
             agent=agent,
         )
@@ -947,51 +752,27 @@ def _create_deployment(
             "POST /v2/deployments succeeded but response missing a valid 'id'"
         )
     if agent is not None:
-<<<<<<< HEAD
-        _get_emitter().info(f"Deployment name: {created['name']}")
-    _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
-    return created_id, step + 1
-
-
-def _smith_dashboard_base_url(host_url: str | None) -> str:
-    """Derive the LangSmith dashboard base URL from the API host URL."""
-    from urllib.parse import urlparse
-
-    if not host_url:
-        return "https://smith.langchain.com"
-    parsed = urlparse(host_url)
-    hostname = parsed.hostname or ""
-    if hostname in ("localhost", "127.0.0.1"):
-        return host_url.rstrip("/")
-
-    for api_host_suffix in ("api.host.langchain.com", "api.smith.langchain.com"):
-        if hostname == api_host_suffix:
-            return "https://smith.langchain.com"
-        if hostname.endswith(f".{api_host_suffix}"):
-            prefix = hostname[: -(len(api_host_suffix) + 1)]
-            return f"https://{prefix}.smith.langchain.com"
-
-    return "https://smith.langchain.com"
-=======
         _get_emitter().info(f"Deployment name: {created.get('name')}")
     _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
     return CreatedDeployment(created_id, created), step + 1
->>>>>>> origin
 
 
 def _get_deployment_status_url(
-    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
+    updated: object, deployment_id: str, host_url: str
 ) -> str | None:
+    """Compute the LangSmith dashboard URL for a deployment, if possible."""
     tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None
     if not tenant_id:
         return None
-    return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}"
+    base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url
+    return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}"
 
 
 def _emit_deployment_status_url(
-    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
+    updated: object, deployment_id: str, host_url: str
 ) -> str | None:
-    url = _get_deployment_status_url(updated, deployment_id, endpoints)
+    """Emit the deployment status URL and return it."""
+    url = _get_deployment_status_url(updated, deployment_id, host_url)
     if url:
         _get_emitter().status_url(url)
     return url
@@ -1009,11 +790,14 @@ def _poll_revision_status(
 ) -> tuple[str, str | None]:
     """Poll latest revision status until terminal status or timeout."""
     em = _get_emitter()
-    revisions = client.list_revisions(deployment_id, limit=1)
-    if not revisions:
+    revisions_resp = client.list_revisions(deployment_id, limit=1)
+    resources = (
+        revisions_resp.get("resources", []) if isinstance(revisions_resp, dict) else []
+    )
+    if not resources:
         return "", None
 
-    revision_id = str(revisions[0]["id"])
+    revision_id = str(resources[0]["id"])
     last_status = ""
     deadline = time.time() + timeout_seconds
     start_time = time.monotonic()
@@ -1534,7 +1318,6 @@ def _run_remote_build(
 @dataclass(frozen=True, slots=True)
 class DeployContext:
     client: HostBackendClient
-    endpoints: ControlPlaneEndpoints
     spec: BuildSpec
     verbose: bool
     selector: DeploymentSelector
@@ -1564,66 +1347,19 @@ def _resolve_or_create(
     )
     if found is not None:
         return found.id, step
-    try:
-        created, step = _create_deployment(
-            ctx.client,
-            step,
-            name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
-            agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
-            source=source,
-            source_config={"deployment_type": ctx.deployment_type},
-            source_revision_config={},
-            secrets=ctx.secrets,
-        )
-    except HostBackendError as err:
-        if _needs_a_listener(err):
-            raise ListenerRequiredError(
-                "The image has to come from a registry you manage, so re-run with "
-                "--push-to <registry>/<repository>."
-            ) from None
-        raise
+    created, step = _create_deployment(
+        ctx.client,
+        step,
+        name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
+        agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
+        source=source,
+        source_config={"deployment_type": ctx.deployment_type},
+        source_revision_config={},
+        secrets=ctx.secrets,
+    )
     return created.id, step
 
 
-class ListenerRequiredError(click.UsageError):
-    def __init__(self, remedy: str) -> None:
-        super().__init__(
-            "This workspace deploys through a listener in your own cluster. "
-            f"{remedy}\nLearn about listeners: {_LISTENERS_DOCS_URL}"
-        )
-
-
-def _needs_a_listener(err: HostBackendError) -> bool:
-    return err.status_code == 400 and _LISTENER_REQUIRED_MARKER in (
-        err.detail or err.message
-    )
-
-
-def _requested_listener(client: HostBackendClient, listener_id: str) -> Listener:
-    try:
-        resource = _call_host_backend_with_optional_tenant(
-            client, lambda c: c.get_listener(listener_id)
-        )
-    except HostBackendError as err:
-        if err.status_code not in _LISTENER_NOT_FOUND_STATUSES:
-            raise
-        available = _available_listeners(client)
-        if not available:
-            raise click.UsageError(_NO_LISTENERS) from None
-        raise click.UsageError(
-            f"Listener {listener_id} was not found in this workspace. "
-            f"Available listeners:\n{_describe_listeners(available)}"
-        ) from None
-    return Listener.from_resource(resource)
-
-
-def _available_listeners(client: HostBackendClient) -> tuple[Listener, ...]:
-    resources = _call_host_backend_with_optional_tenant(
-        client, lambda c: c.list_listeners()
-    )
-    return tuple(Listener.from_resource(resource) for resource in resources)
-
-
 def _ensure_customer_registry_source(existing: ExistingDeployment) -> None:
     if existing.source != _CUSTOMER_REGISTRY_SOURCE:
         raise click.UsageError(
@@ -1686,7 +1422,6 @@ class RemoteBuildSource:
 class CustomerRegistrySource:
     reference: ImageReference
     prebuilt_image: str | None
-    requested_placement: RequestedPlacement
 
     def run(self, ctx: DeployContext) -> DeployOutcome:
         if isinstance(ctx.selector, ById):
@@ -1708,7 +1443,6 @@ class CustomerRegistrySource:
         self, ctx: DeployContext, existing: ExistingDeployment, step: int
     ) -> DeployOutcome:
         _ensure_customer_registry_source(existing)
-        self.requested_placement.ensure_not_requested(existing.id)
         image_uri, step = self._publish(ctx, step)
         _log_deploy_step(step, f"Updating deployment {existing.id}")
         updated = ctx.client.update_deployment(
@@ -1722,25 +1456,7 @@ class CustomerRegistrySource:
             existing.id, _image_revision_result(updated, "Deployment updated")
         )
 
-    def _resolve_placement(self, ctx: DeployContext) -> Placement:
-        requested = self.requested_placement
-        if requested.listener_id is not None:
-            return requested.on(_requested_listener(ctx.client, requested.listener_id))
-        if not (ctx.endpoints.is_cloud or requested.requested):
-            return Unplaced()
-        return requested.among(_available_listeners(ctx.client))
-
-    def _announce(self, placement: Placement) -> None:
-        if isinstance(placement, OnListener):
-            _get_emitter().info(
-                placement.summary,
-                listener_id=placement.listener_id,
-                k8s_namespace=placement.k8s_namespace,
-            )
-
     def _create(self, ctx: DeployContext, name: str | None, step: int) -> DeployOutcome:
-        placement = self._resolve_placement(ctx)
-        self._announce(placement)
         image_uri, step = self._publish(ctx, step)
         try:
             created, _ = _create_deployment(
@@ -1751,19 +1467,13 @@ class CustomerRegistrySource:
                 if isinstance(ctx.selector, ByAgent)
                 else None,
                 source=_CUSTOMER_REGISTRY_SOURCE,
-                source_config={
-                    "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
-                    **placement.source_config(),
-                },
+                source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC},
                 source_revision_config={"image_uri": image_uri},
                 secrets=ctx.secrets,
             )
         except HostBackendError as err:
-            if _needs_a_listener(err):
-                raise ListenerRequiredError(
-                    "Re-run with --listener-id and --k8s-namespace.\n"
-                    f"{err.detail or err.message}"
-                ) from None
+            if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message:
+                raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None
             raise
         return DeployOutcome(
             created.id, _image_revision_result(created.resource, "Deployment created")
@@ -1824,31 +1534,14 @@ def _select_source(
     image_name: str | None,
     tag: str | None,
     remote_build_flag: bool | None,
-    placement: RequestedPlacement,
-    selector: DeploymentSelector,
 ) -> DeploymentSource:
-    if push_to is None and placement.requested:
-        raise click.UsageError(
-            "--listener-id and --k8s-namespace only apply when creating a "
-            "deployment with --push-to."
-        )
-    if placement.requested and isinstance(selector, ById):
-        raise click.UsageError(
-            "Listener and namespace are fixed when a deployment is created, so "
-            "they cannot be set for an existing --deployment-id. Drop them, or "
-            "create a new deployment with --name."
-        )
     if push_to is not None:
         if remote_build_flag is True:
             raise click.UsageError("--push-to cannot be combined with --remote.")
         reference = _push_reference(push_to, tag)
         if image is None:
             _require_local_docker()
-        return CustomerRegistrySource(
-            reference=reference,
-            prebuilt_image=image,
-            requested_placement=placement,
-        )
+        return CustomerRegistrySource(reference, prebuilt_image=image)
     if image and remote_build_flag is True:
         raise click.UsageError("--image cannot be combined with --remote builds.")
     use_remote_build, local_build_error = _resolve_build_mode(
@@ -1954,7 +1647,9 @@ def _call_host_backend_with_optional_tenant(
                 prompted_for_tenant = True
                 continue
             if err.status_code == 403 and "not enabled" in err.message.lower():
-                smith_base = client.endpoints.dashboard_url
+                smith_base = ControlPlaneEndpoints.from_control_plane_url(
+                    client.base_url
+                ).dashboard_url
                 raise HostBackendError(
                     "LangSmith Deployment is not enabled for this organization. "
                     f"Enable it at {smith_base}/host/deployments"
@@ -1995,17 +1690,11 @@ OPT_HOST_URL = click.option(
 )
 
 OPT_AGENT_ID = click.option(
-    "--agent-id",
-    envvar="LANGSMITH_AGENT_ID",
-    show_envvar=True,
-    help="Logical agent ID (requires agent mode enabled for the tenant).",
+    "--agent-id", help="Logical agent ID (requires agent mode enabled for the tenant)."
 )
 
 OPT_AGENT_ENVIRONMENT = click.option(
-    "--agent-environment",
-    "environment",
-    envvar="LANGSMITH_AGENT_ENVIRONMENT",
-    show_envvar=True,
+    "--environment",
     type=click.Choice(["development", "staging", "production"]),
     help="Agent environment (requires agent mode enabled for the tenant).",
 )
@@ -2159,21 +1848,6 @@ def _deploy_base_options(
                     "Give the tag here or with --tag (default: latest)."
                 ),
             ),
-            click.option(
-                "--listener-id",
-                help=(
-                    "Listener that will run the deployment, for workspaces that "
-                    "deploy through a listener in your own cluster. Only used when "
-                    "creating a deployment with --push-to."
-                ),
-            ),
-            click.option(
-                "--k8s-namespace",
-                help=(
-                    "Kubernetes namespace the listener deploys into. Only used when "
-                    "creating a deployment with --push-to."
-                ),
-            ),
             click.option(
                 "--config",
                 "-c",
@@ -2284,8 +1958,6 @@ def _deploy_cmd(
     image_name: str | None,
     image: str | None,
     push_to: str | None,
-    listener_id: str | None,
-    k8s_namespace: str | None,
     tag: str | None,
     base_image: str | None,
     install_command: str | None,
@@ -2310,7 +1982,6 @@ def _deploy_cmd(
     validate_deploy_commands(install_command, build_command)
     agent = None
     if agent_id is not None or environment is not None:
-<<<<<<< HEAD
         if not agent_id or not agent_id.strip() or not environment:
             raise click.UsageError(
                 "--agent-id and --environment are required together."
@@ -2318,16 +1989,6 @@ def _deploy_cmd(
         if name is not None or deployment_id is not None:
             raise click.UsageError(
                 "--agent-id and --environment cannot be combined with --name or --deployment-id."
-=======
-        em.note("Note: --agent-id and --agent-environment flags are in private beta")
-        if not agent_id or not agent_id.strip() or not environment:
-            raise click.UsageError(
-                "--agent-id and --agent-environment are required together."
-            )
-        if name is not None or deployment_id is not None:
-            raise click.UsageError(
-                "--agent-id and --agent-environment cannot be combined with --name or --deployment-id."
->>>>>>> origin
             )
         agent = {"agent_id": agent_id, "environment": environment}
     if not config.exists():
@@ -2363,53 +2024,15 @@ def _deploy_cmd(
 
     secrets = _secrets_from_env(_env_without_deployment_name(env_vars))
 
-    selector = ByAgent(**agent) if agent else deployment_selector(deployment_id, name)
     source = _select_source(
         push_to=push_to,
         image=image,
         image_name=image_name,
         tag=tag,
         remote_build_flag=remote_build_flag,
-        placement=RequestedPlacement(listener_id, k8s_namespace),
-        selector=selector,
     )
 
     client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
-<<<<<<< HEAD
-    step = 1
-
-    deployment_id, needs_creation, step = _resolve_deployment(
-        client,
-        step,
-        deployment_id,
-        name,
-        not_found_message=(
-            "No deployment found. Will create."
-            if use_remote_build
-            else "No deployment found. Will create after build."
-        ),
-        agent=agent,
-    )
-
-    if needs_creation:
-        deployment_id, step = _create_deployment(
-            client,
-            step,
-            name=name,
-            deployment_type=deployment_type,
-            source="internal_source" if use_remote_build else "internal_docker",
-            secrets=secrets,
-            agent=agent,
-        )
-
-    if not deployment_id:
-        raise click.ClickException("Failed to determine deployment ID")
-
-    # Scan local sources for tracked packages so the new revision carries
-    # the same metadata GitHub-backed deploys produce. Failures must never
-    # block a deploy.
-=======
->>>>>>> origin
     try:
         tracked_packages = find_tracked_packages(config, config_json) or None
     except Exception as exc:
@@ -2419,7 +2042,6 @@ def _deploy_cmd(
     outcome = source.run(
         DeployContext(
             client=client,
-            endpoints=client.endpoints,
             spec=BuildSpec(
                 config=config,
                 config_json=config_json,
@@ -2431,7 +2053,9 @@ def _deploy_cmd(
                 build_command=build_command,
             ),
             verbose=verbose,
-            selector=selector,
+            selector=ByAgent(**agent)
+            if agent
+            else deployment_selector(deployment_id, name),
             deployment_type=deployment_type,
             secrets=secrets,
             tracked_packages=tracked_packages,
@@ -2440,7 +2064,7 @@ def _deploy_cmd(
     dep_status_url = _emit_deployment_status_url(
         outcome.build_result.updated,
         outcome.deployment_id,
-        client.endpoints,
+        client.base_url,
     )
 
     if no_wait:
@@ -2514,14 +2138,6 @@ def deploy_list(
     agent_id: str | None,
     environment: str | None,
 ) -> None:
-<<<<<<< HEAD
-=======
-    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",
-        )
->>>>>>> origin
     if agent_id is not None and not agent_id.strip():
         raise click.UsageError("--agent-id must not be empty.")
     filters = {}
@@ -2530,18 +2146,15 @@ def deploy_list(
     if environment is not None:
         filters["agent_environment"] = environment
     client = _create_host_backend_client(host_url, api_key)
-    deployments = _call_host_backend_with_optional_tenant(
+    response = _call_host_backend_with_optional_tenant(
         client,
         lambda c: c.list_deployments(name_contains=name_contains, **filters),
-<<<<<<< HEAD
     )
     resources = response.get("resources") if isinstance(response, dict) else None
     deployments = (
         [item for item in resources if isinstance(item, dict)]
         if isinstance(resources, list)
         else []
-=======
->>>>>>> origin
     )
     if not deployments:
         click.echo("No deployments found.")
@@ -2582,10 +2195,16 @@ def deploy_revisions_list(
     api_key: str | None, host_url: str | None, limit: int, deployment_id: str
 ) -> None:
     client = _create_host_backend_client(host_url, api_key)
-    revisions = _call_host_backend_with_optional_tenant(
+    response = _call_host_backend_with_optional_tenant(
         client,
         lambda c: c.list_revisions(deployment_id, limit=limit),
     )
+    resources = response.get("resources") if isinstance(response, dict) else None
+    revisions = (
+        [item for item in resources if isinstance(item, dict)]
+        if isinstance(resources, list)
+        else []
+    )
     if not revisions:
         click.echo(f"No revisions found for deployment {deployment_id}.")
         return
@@ -2735,12 +2354,17 @@ def deploy_logs(
         dep_id = found.id
 
     if log_type == "build" and not revision_id:
-        revisions = client.list_revisions(dep_id, limit=1)
-        if not revisions:
+        revisions_resp = client.list_revisions(dep_id, limit=1)
+        resources = (
+            revisions_resp.get("resources", [])
+            if isinstance(revisions_resp, dict)
+            else []
+        )
+        if not resources:
             raise click.ClickException(
                 "No revisions found for this deployment. Cannot fetch build logs."
             )
-        revision_id = str(revisions[0]["id"])
+        revision_id = str(resources[0]["id"])
         click.secho(f"Using latest revision: {revision_id}", fg="cyan")
 
     payload: dict = {"limit": limit, "order": "desc"}

libs/cli/langgraph_cli/host_backend.py

@@ -18,7 +18,6 @@ CLOUD_DASHBOARD_HOST = "smith.langchain.com"
 CONTROL_PLANE_PATH = "/api-host"
 LANGSMITH_API_PATHS = ("/api/v1", "/api")
 LOCAL_HOSTNAMES = ("localhost", "127.0.0.1")
-MAX_PAGE_SIZE = 100
 SourceName = Literal["internal_docker", "internal_source", "external_docker"]
 
 
@@ -37,13 +36,6 @@ class ControlPlaneEndpoints:
             return cls.from_langsmith_endpoint(langsmith_endpoint)
         return cls(CLOUD_CONTROL_PLANE_URL, CLOUD_DASHBOARD_URL)
 
-    @property
-    def is_cloud(self) -> bool:
-        hostname = urlparse(self.control_plane_url).hostname or ""
-        return hostname == CLOUD_CONTROL_PLANE_HOST or hostname.endswith(
-            f".{CLOUD_CONTROL_PLANE_HOST}"
-        )
-
     @classmethod
     def from_control_plane_url(cls, url: str) -> ControlPlaneEndpoints:
         control_plane_url = url.rstrip("/")
@@ -91,36 +83,12 @@ def _without_api_path(path: str) -> str:
     return path
 
 
-def _resources(payload: object) -> list[dict[str, Any]]:
-    if not isinstance(payload, dict):
-        return []
-    resources = payload.get("resources")
-    if not isinstance(resources, list):
-        return []
-    return [item for item in resources if isinstance(item, dict)]
-
-
 class HostBackendError(click.ClickException):
     """Raised when the host backend returns an error response."""
 
-    def __init__(
-        self,
-        message: str,
-        status_code: int | None = None,
-        detail: str | None = None,
-    ):
+    def __init__(self, message: str, status_code: int | None = None):
         super().__init__(message)
         self.status_code = status_code
-        self.detail = detail
-
-
-def _error_detail(response: httpx.Response) -> str | None:
-    try:
-        body = response.json()
-    except ValueError:
-        return None
-    detail = body.get("detail") if isinstance(body, dict) else None
-    return detail if isinstance(detail, str) else None
 
 
 class HostBackendClient:
@@ -142,8 +110,7 @@ class HostBackendClient:
         }
         if tenant_id:
             headers["X-Tenant-ID"] = tenant_id
-        self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
-        self._base_url = self._endpoints.control_plane_url
+        self._base_url = base_url.rstrip("/")
         self._client = httpx.Client(
             base_url=self._base_url,
             headers=headers,
@@ -155,10 +122,6 @@ class HostBackendClient:
     def base_url(self) -> str:
         return self._base_url
 
-    @property
-    def endpoints(self) -> ControlPlaneEndpoints:
-        return self._endpoints
-
     def set_tenant(self, tenant_id: str) -> None:
         self._client.headers["X-Tenant-ID"] = tenant_id
 
@@ -173,12 +136,10 @@ class HostBackendClient:
             resp = self._client.request(method, path, json=payload, params=params)
             resp.raise_for_status()
         except httpx.HTTPStatusError as err:
-            detail = _error_detail(err.response)
-            reason = detail or err.response.text or str(err.response.status_code)
+            detail = err.response.text or str(err.response.status_code)
             raise HostBackendError(
-                f"{method} {path} failed with status {err.response.status_code}: {reason}",
+                f"{method} {path} failed with status {err.response.status_code}: {detail}",
                 status_code=err.response.status_code,
-                detail=detail,
             ) from None
         except httpx.TransportError as err:
             raise HostBackendError(str(err)) from None
@@ -217,29 +178,20 @@ class HostBackendClient:
 
     def list_deployments(
         self,
+        name_contains: str = "",
         *,
-        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, Any]]:
-        given = (
-            ("name", name),
-            ("name_contains", name_contains),
-            ("limit", limit),
-            ("agent_id", agent_id),
-            ("agent_environment", agent_environment),
-        )
-        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, Any]]:
-        return _resources(
-            self._request("GET", "/v2/listeners", params={"limit": MAX_PAGE_SIZE})
+    ) -> dict[str, Any]:
+        params = {"name_contains": name_contains}
+        if agent_id is not None:
+            params["agent_id"] = agent_id
+        if agent_environment is not None:
+            params["agent_environment"] = agent_environment
+        return self._request(
+            "GET",
+            "/v2/deployments",
+            params=params,
         )
 
     def get_deployment(self, deployment_id: str) -> dict[str, Any]:
@@ -314,15 +266,10 @@ class HostBackendClient:
             payload["secrets"] = secrets
         return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
 
-    def list_revisions(
-        self, deployment_id: str, limit: int = 1
-    ) -> list[dict[str, Any]]:
-        return _resources(
-            self._request(
-                "GET",
-                f"/v2/deployments/{deployment_id}/revisions",
-                params={"limit": limit},
-            )
+    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
+        return self._request(
+            "GET",
+            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
         )
 
     def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -382,18 +382,20 @@ def test_deploy_list_command(monkeypatch) -> None:
 
         def list_deployments(self, name_contains: str = ""):
             captured["name_contains"] = name_contains
-            return [
-                {
-                    "id": "dep-123",
-                    "name": "alpha",
-                    "source_config": {"custom_url": "https://alpha.example.com"},
-                },
-                {
-                    "id": "dep-456",
-                    "name": "beta",
-                    "source_config": {"custom_url": "https://beta.example.com"},
-                },
-            ]
+            return {
+                "resources": [
+                    {
+                        "id": "dep-123",
+                        "name": "alpha",
+                        "source_config": {"custom_url": "https://alpha.example.com"},
+                    },
+                    {
+                        "id": "dep-456",
+                        "name": "beta",
+                        "source_config": {"custom_url": "https://beta.example.com"},
+                    },
+                ]
+            }
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -433,7 +435,7 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_deployments(self, name_contains: str = ""):
-            return []
+            return {"resources": []}
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -466,18 +468,20 @@ def test_deploy_revisions_list_command(monkeypatch) -> None:
         def list_revisions(self, deployment_id: str, limit: int = 1):
             captured["deployment_id"] = deployment_id
             captured["limit"] = str(limit)
-            return [
-                {
-                    "id": "rev-123",
-                    "status": "CREATING",
-                    "created_at": "2023-11-07T05:31:56Z",
-                },
-                {
-                    "id": "rev-456",
-                    "status": "DEPLOYED",
-                    "created_at": "2023-11-08T10:00:00Z",
-                },
-            ]
+            return {
+                "resources": [
+                    {
+                        "id": "rev-123",
+                        "status": "CREATING",
+                        "created_at": "2023-11-07T05:31:56Z",
+                    },
+                    {
+                        "id": "rev-456",
+                        "status": "DEPLOYED",
+                        "created_at": "2023-11-08T10:00:00Z",
+                    },
+                ]
+            }
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -518,7 +522,7 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_revisions(self, deployment_id: str, limit: int = 1):
-            return []
+            return {"resources": []}
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -551,7 +555,7 @@ def test_deploy_revisions_list_command_with_explicit_limit(monkeypatch) -> None:
         def list_revisions(self, deployment_id: str, limit: int = 1):
             captured["deployment_id"] = deployment_id
             captured["limit"] = str(limit)
-            return []
+            return {"resources": []}
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 

libs/cli/tests/unit_tests/cli/test_deploy_command.py

@@ -1,6 +1,5 @@
 import asyncio
 import json
-import uuid
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
@@ -18,7 +17,6 @@ from langgraph_cli.host_backend import HostBackendClient
 from langgraph_cli.image_reference import ImageReference
 
 CONTROL_PLANE_URL = "https://control-plane.example.com"
-CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com"
 REGISTRY_URL = "https://registry.example.com/team"
 PUSH_TOKEN = "push-token"
 PUSHED_IMAGE = "registry.example.com/team/my-app:latest"
@@ -26,25 +24,10 @@ PUSHED_DIGEST = "registry.example.com/team/my-app@sha256:abc123"
 PUSH_REPOSITORY = "registry.example.com/team/agent"
 EXTERNAL_IMAGE = f"{PUSH_REPOSITORY}:latest"
 EXTERNAL_DIGEST = f"{PUSH_REPOSITORY}@sha256:abc123"
-LISTENER_ID = "11111111-1111-4111-8111-111111111111"
-OTHER_LISTENER_ID = "22222222-2222-4222-8222-222222222222"
-PAGE_TWO_LISTENER_ID = "33333333-3333-4333-8333-333333333333"
-UNKNOWN_LISTENER_ID = "99999999-9999-4999-8999-999999999999"
-LISTENER = {
-    "id": LISTENER_ID,
-    "compute_id": "prod-cluster",
-    "compute_config": {"k8s_namespaces": ["agents"]},
-}
-OTHER_LISTENER = {
-    "id": OTHER_LISTENER_ID,
-    "compute_id": "other-cluster",
-    "compute_config": {"k8s_namespaces": ["agents"]},
-}
-TWO_NAMESPACE_LISTENER = {
-    "id": LISTENER_ID,
-    "compute_id": "prod-cluster",
-    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
-}
+LISTENER_REQUIRED = (
+    "Source configuration error: 'source_config.listener_id' is required for "
+    "workspace with available listener IDs: ['listener-1']"
+)
 CREATED_ID = "dep-created"
 TRACKED_PACKAGES = ["langgraph:1.0.0"]
 SIGNED_UPLOAD_URL = "https://storage.example.com/signed"
@@ -55,12 +38,7 @@ DIGESTS_FORMAT = "{{json .RepoDigests}}"
 NOT_A_CLI_DEPLOYMENT = (
     "push token is only available for 'internal_docker' source deployments"
 )
-LISTENER_REQUIRED = (
-    "Source configuration error: 'source_config.listener_id' is required "
-    f"for workspace with available listener IDs: ['{LISTENER_ID}']"
-)
 LIST_DEPLOYMENTS = "GET /v2/deployments"
-LIST_LISTENERS = "GET /v2/listeners"
 CREATE_DEPLOYMENT = "POST /v2/deployments"
 
 
@@ -80,22 +58,12 @@ def _get(deployment_id: str) -> str:
     return f"GET /v2/deployments/{deployment_id}"
 
 
-def _looks_like_a_uuid(value: str) -> bool:
-    try:
-        uuid.UUID(value)
-    except ValueError:
-        return False
-    return True
-
-
 @dataclass
 class ControlPlaneDouble:
     timeline: list[str]
     existing_deployments: list[dict] = field(default_factory=list)
     push_token_status: int = 200
     create_error: str | None = None
-    listeners: list[dict] = field(default_factory=list)
-    listeners_by_id: dict[str, dict] = field(default_factory=dict)
     bodies: dict[str, dict] = field(default_factory=dict)
 
     def handle(self, request: httpx.Request) -> httpx.Response:
@@ -103,45 +71,14 @@ class ControlPlaneDouble:
         self.timeline.append(route)
         if request.content:
             self.bodies[route] = json.loads(request.content)
-        return self._respond(request)
+        return self._respond(request.method, request.url.path)
 
-    def _respond(self, request: httpx.Request) -> httpx.Response:
-        method, path = request.method, request.url.path
-        if (method, path) == ("GET", "/v2/listeners"):
-            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):
-                return httpx.Response(
-                    422,
-                    json={
-                        "detail": [
-                            {"type": "uuid_parsing", "loc": ["path", "listener_id"]}
-                        ]
-                    },
-                )
-            known = {listener["id"]: listener for listener in self.listeners}
-            known.update(self.listeners_by_id)
-            if listener_id not in known:
-                return httpx.Response(
-                    404, json={"detail": f"Listener ID {listener_id} not found."}
-                )
-            return httpx.Response(200, json=known[listener_id])
+    def _respond(self, method: str, path: str) -> httpx.Response:
         if (method, path) == ("GET", "/v2/deployments"):
-            name = request.url.params.get("name")
-            return httpx.Response(
-                200,
-                json={
-                    "resources": [
-                        deployment
-                        for deployment in self.existing_deployments
-                        if name is None or deployment.get("name") == name
-                    ]
-                },
-            )
+            return httpx.Response(200, json={"resources": self.existing_deployments})
         if (method, path) == ("POST", "/v2/deployments"):
             if self.create_error is not None:
-                return httpx.Response(400, json={"detail": self.create_error})
+                return httpx.Response(400, text=self.create_error)
             return httpx.Response(201, json={"id": CREATED_ID, "tenant_id": "tenant-1"})
         if path.endswith("/push-token"):
             if self.push_token_status != 200:
@@ -262,7 +199,7 @@ class DeployProject:
     timeline: list[str]
     uploads: list[tuple[str, str, int]]
 
-    def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result:
+    def run(self, *args: str) -> Result:
         return CliRunner().invoke(
             cli,
             [
@@ -270,7 +207,7 @@ class DeployProject:
                 "--api-key",
                 "test-key",
                 "--host-url",
-                host_url,
+                CONTROL_PLANE_URL,
                 "--name",
                 "my-app",
                 "--no-input",
@@ -674,6 +611,18 @@ def test_push_to_rejects_a_non_external_deployment_before_any_docker_work(
     assert deploy_project.docker.verbs() == []
 
 
+def test_push_to_explains_the_listener_requirement_of_hybrid_workspaces(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.create_error = LISTENER_REQUIRED
+
+    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
+
+    assert result.exit_code != 0
+    assert "listener" in result.output
+    assert "--deployment-id" in result.output
+
+
 def test_push_to_with_deployment_id_fetches_the_deployment_once(
     deploy_project: DeployProject,
 ) -> None:
@@ -703,414 +652,3 @@ def test_invalid_tag_fails_before_any_control_plane_call(
     assert result.exit_code != 0
     assert "Image tag may only contain" in result.output
     assert deploy_project.timeline == []
-
-
-def test_push_to_places_a_new_deployment_on_the_only_listener(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-
-    result = deploy_project.run(
-        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
-    )
-
-    assert result.exit_code == 0, result.output
-    assert deploy_project.timeline == [
-        LIST_DEPLOYMENTS,
-        LIST_LISTENERS,
-        "docker build",
-        "docker push",
-        "docker inspect-digest",
-        CREATE_DEPLOYMENT,
-    ]
-    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
-        "resource_spec": {},
-        "listener_id": LISTENER_ID,
-        "listener_config": {"k8s_namespace": "agents"},
-    }
-    assert f"Deploying through listener {LISTENER_ID} in namespace agents" in (
-        result.output
-    )
-
-
-def test_push_to_places_a_new_deployment_on_the_chosen_listener(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER]
-
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--listener-id",
-        OTHER_LISTENER_ID,
-        "--k8s-namespace",
-        "agents",
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code == 0, result.output
-    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
-        "resource_spec": {},
-        "listener_id": OTHER_LISTENER_ID,
-        "listener_config": {"k8s_namespace": "agents"},
-    }
-
-
-@pytest.mark.parametrize(
-    ("listeners", "args", "message"),
-    [
-        pytest.param(
-            [LISTENER, OTHER_LISTENER], (), "--listener-id", id="two_listeners"
-        ),
-        pytest.param(
-            [TWO_NAMESPACE_LISTENER], (), "--k8s-namespace", id="two_namespaces"
-        ),
-        pytest.param(
-            [LISTENER],
-            ("--k8s-namespace", "nope"),
-            "does not serve namespace",
-            id="unknown_namespace",
-        ),
-    ],
-)
-def test_push_to_refuses_an_unresolved_placement_before_any_docker_work(
-    deploy_project: DeployProject, listeners, args, message
-) -> None:
-    deploy_project.control_plane.listeners = listeners
-
-    result = deploy_project.run(
-        "--push-to", PUSH_REPOSITORY, *args, host_url=CLOUD_CONTROL_PLANE_URL
-    )
-
-    assert result.exit_code != 0
-    assert message in result.output
-    assert deploy_project.docker.verbs() == []
-    assert CREATE_DEPLOYMENT not in deploy_project.timeline
-
-
-def test_self_hosted_control_plane_keeps_its_default_placement(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-
-    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
-
-    assert result.exit_code == 0, result.output
-    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
-        "resource_spec": {}
-    }
-
-
-def test_self_hosted_control_plane_places_when_asked(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-
-    result = deploy_project.run(
-        "--push-to", PUSH_REPOSITORY, "--listener-id", LISTENER_ID
-    )
-
-    assert result.exit_code == 0, result.output
-    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
-        "resource_spec": {},
-        "listener_id": LISTENER_ID,
-        "listener_config": {"k8s_namespace": "agents"},
-    }
-
-
-def test_updating_a_deployment_never_looks_up_listeners(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-    deploy_project.control_plane.existing_deployments = [
-        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
-    ]
-
-    result = deploy_project.run(
-        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
-    )
-
-    assert result.exit_code == 0, result.output
-    assert LIST_LISTENERS not in deploy_project.timeline
-
-
-def test_listener_flags_are_refused_for_a_deployment_id_without_any_call(
-    deploy_project: DeployProject,
-) -> None:
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--deployment-id",
-        "dep-ext",
-        "--k8s-namespace",
-        "agents",
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code != 0
-    assert "fixed when a deployment is created" in result.output
-    assert deploy_project.timeline == []
-
-
-def test_listener_flags_are_refused_on_an_existing_deployment(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-    deploy_project.control_plane.existing_deployments = [
-        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
-    ]
-
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--listener-id",
-        LISTENER_ID,
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code != 0
-    assert "fixed when a deployment is created" in result.output
-    assert deploy_project.docker.verbs() == []
-
-
-def test_a_deployment_without_a_listener_announces_nothing(
-    deploy_project: DeployProject,
-) -> None:
-    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
-
-    assert result.exit_code == 0, result.output
-    assert "listener" not in result.output
-
-
-def test_a_self_hosted_create_without_flags_never_looks_up_listeners(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-
-    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
-
-    assert result.exit_code == 0, result.output
-    assert LIST_LISTENERS not in deploy_project.timeline
-
-
-def test_a_control_plane_that_demands_a_listener_names_the_flags(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.create_error = LISTENER_REQUIRED
-
-    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
-
-    assert result.exit_code != 0
-    assert "--listener-id" in result.output
-    assert "--k8s-namespace" in result.output
-    assert LISTENER_ID in result.output
-    assert "{" not in result.output
-    assert "POST /v2/deployments failed" not in result.output
-
-
-def test_listener_flags_without_push_to_make_no_call_at_all(
-    deploy_project: DeployProject,
-) -> None:
-    result = deploy_project.run("--listener-id", LISTENER_ID)
-
-    assert result.exit_code != 0
-    assert "--push-to" in result.output
-    assert deploy_project.timeline == []
-
-
-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(100)
-    ]
-
-    result = deploy_project.run(
-        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
-    )
-
-    assert result.exit_code != 0
-    assert "first 100" in result.output
-
-
-def test_a_managed_build_in_a_listener_workspace_points_at_push_to(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.create_error = LISTENER_REQUIRED
-
-    result = deploy_project.run("--no-remote")
-
-    assert result.exit_code != 0
-    assert "--push-to" in result.output
-    assert deploy_project.docker.verbs() == []
-
-
-@pytest.mark.parametrize(
-    "args",
-    [
-        pytest.param(("--no-remote",), id="managed_build"),
-        pytest.param(("--push-to", PUSH_REPOSITORY), id="push_to"),
-    ],
-)
-def test_a_listener_requirement_links_the_listener_docs(
-    deploy_project: DeployProject, args: tuple[str, ...]
-) -> None:
-    deploy_project.control_plane.create_error = LISTENER_REQUIRED
-
-    result = deploy_project.run(*args)
-
-    assert result.exit_code != 0
-    assert "https://docs.langchain.com/langsmith/control-plane#listeners" in (
-        result.output
-    )
-
-
-def test_a_managed_control_plane_without_listeners_creates_as_before(
-    deploy_project: DeployProject,
-) -> None:
-    result = deploy_project.run(
-        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
-    )
-
-    assert result.exit_code == 0, result.output
-    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
-        "resource_spec": {}
-    }
-    assert deploy_project.timeline.count(LIST_LISTENERS) == 1
-
-
-def test_a_listener_without_an_id_is_reported_rather_than_ignored(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [
-        {"compute_id": "broken", "compute_config": {"k8s_namespaces": ["agents"]}},
-        LISTENER,
-    ]
-
-    result = deploy_project.run(
-        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
-    )
-
-    assert result.exit_code != 0
-    assert "without an id" in result.output
-    assert deploy_project.docker.verbs() == []
-
-
-def _listener_route(listener_id: str) -> str:
-    return f"GET /v2/listeners/{listener_id}"
-
-
-def test_an_explicit_listener_is_fetched_by_id_not_searched(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER]
-
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--listener-id",
-        OTHER_LISTENER_ID,
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code == 0, result.output
-    assert _listener_route(OTHER_LISTENER_ID) in deploy_project.timeline
-    assert LIST_LISTENERS not in deploy_project.timeline
-    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
-        "resource_spec": {},
-        "listener_id": OTHER_LISTENER_ID,
-        "listener_config": {"k8s_namespace": "agents"},
-    }
-
-
-def test_an_explicit_listener_beyond_the_first_page_still_works(
-    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(100)
-    ]
-    deploy_project.control_plane.listeners_by_id = {
-        PAGE_TWO_LISTENER_ID: {
-            "id": PAGE_TWO_LISTENER_ID,
-            "compute_id": "far-cluster",
-            "compute_config": {"k8s_namespaces": ["agents"]},
-        }
-    }
-
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--listener-id",
-        PAGE_TWO_LISTENER_ID,
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code == 0, result.output
-    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
-        "resource_spec": {},
-        "listener_id": PAGE_TWO_LISTENER_ID,
-        "listener_config": {"k8s_namespace": "agents"},
-    }
-
-
-def test_an_unknown_listener_names_the_ones_that_exist(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--listener-id",
-        UNKNOWN_LISTENER_ID,
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code != 0
-    assert "was not found" in result.output
-    assert LISTENER_ID in result.output
-    assert "prod-cluster" in result.output
-    assert deploy_project.docker.verbs() == []
-
-
-def test_an_explicit_listener_in_a_workspace_without_any_is_refused(
-    deploy_project: DeployProject,
-) -> None:
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--listener-id",
-        LISTENER_ID,
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code != 0
-    assert "no listeners" in result.output
-    assert deploy_project.docker.verbs() == []
-
-
-def test_a_listener_id_that_is_not_an_identifier_still_names_the_real_ones(
-    deploy_project: DeployProject,
-) -> None:
-    deploy_project.control_plane.listeners = [LISTENER]
-
-    result = deploy_project.run(
-        "--push-to",
-        PUSH_REPOSITORY,
-        "--listener-id",
-        "not-a-listener",
-        host_url=CLOUD_CONTROL_PLANE_URL,
-    )
-
-    assert result.exit_code != 0
-    assert "was not found" in result.output
-    assert LISTENER_ID in result.output
-    assert "uuid_parsing" not in result.output

libs/cli/tests/unit_tests/test_deploy_agents.py

@@ -58,7 +58,7 @@ AGENT_ARGS = [
     "deploy",
     "--agent-id",
     "customer-support",
-    "--agent-environment",
+    "--environment",
     "staging",
     "--remote",
     "--no-wait",
@@ -72,9 +72,9 @@ def test_agent_create(deployment_api, tmp_path, monkeypatch):
     result = CliRunner().invoke(cli, AGENT_ARGS)
     assert result.exit_code == 0, result.output
     assert dict(requests[0].url.params) == {
+        "name_contains": "",
         "agent_id": "customer-support",
         "agent_environment": "staging",
-        "limit": "100",
     }
     payload = json.loads(requests[1].content)
     assert payload["agent"] == {
@@ -103,17 +103,3 @@ def test_agent_rejects_explicit_name(deployment_api, monkeypatch):
     assert result.exit_code == 2
     assert "cannot be combined" in result.output
     assert not requests
-
-
-def test_agent_lookup_refuses_a_control_plane_that_ignores_the_filter(deployment_api):
-    state, requests, _ = deployment_api
-    state["resources"] = [
-        {"id": "someone-elses", "is_preview": False},
-        {"id": "another", "is_preview": False},
-    ]
-
-    result = CliRunner().invoke(cli, AGENT_ARGS)
-
-    assert result.exit_code != 0
-    assert "does not filter deployments by agent" in result.output
-    assert len(requests) == 1

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -13,17 +13,10 @@ import pytest
 
 import langgraph_cli.deploy as deploy_mod
 from langgraph_cli.deploy import (
-    ById,
-    ByName,
     CustomerRegistrySource,
     DockerBuildCommand,
-    ExistingDeployment,
-    Listener,
     ManagedRegistrySource,
-    OnListener,
     RemoteBuildSource,
-    RequestedPlacement,
-    Unplaced,
     _call_host_backend_with_optional_tenant,
     _create_host_backend_client,
     _docker_config_for_token,
@@ -34,7 +27,6 @@ from langgraph_cli.deploy import (
     _resolve_pushed_image_digest,
     _select_source,
     _validate_prebuilt_image,
-    find_deployment_by_name,
     normalize_image_tag,
     normalize_name,
 )
@@ -288,13 +280,11 @@ class TestCallHostBackendWithOptionalTenant:
         return c
 
     def test_success_passes_through(self):
-        client = self._make_client(
-            lambda req: httpx.Response(200, json={"resources": [{"id": "dep-1"}]})
-        )
+        client = self._make_client(lambda req: httpx.Response(200, json={"ok": True}))
         result = _call_host_backend_with_optional_tenant(
             client, lambda c: c.list_deployments()
         )
-        assert result == [{"id": "dep-1"}]
+        assert result == {"ok": True}
 
     def test_403_not_enabled_gives_actionable_error(self):
         detail = (
@@ -617,8 +607,6 @@ class TestSelectSource:
         "image_name": None,
         "tag": None,
         "remote_build_flag": None,
-        "placement": RequestedPlacement(),
-        "selector": ByName("my-app"),
     }
     REPOSITORY = "registry.example.com/app"
 
@@ -629,9 +617,7 @@ class TestSelectSource:
                 {"push_to": REPOSITORY},
                 True,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "latest"),
-                    prebuilt_image=None,
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image=None
                 ),
                 id="push_to_selects_the_external_source_with_the_default_tag",
             ),
@@ -639,9 +625,7 @@ class TestSelectSource:
                 {"push_to": f"{REPOSITORY}:v2"},
                 True,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "v2"),
-                    prebuilt_image=None,
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "v2"), prebuilt_image=None
                 ),
                 id="push_to_keeps_a_tag_given_in_the_reference",
             ),
@@ -649,9 +633,7 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "tag": "v3"},
                 True,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "v3"),
-                    prebuilt_image=None,
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "v3"), prebuilt_image=None
                 ),
                 id="tag_flag_composes_with_push_to",
             ),
@@ -659,25 +641,10 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "image": "app:dev"},
                 False,
                 CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "latest"),
-                    prebuilt_image="app:dev",
-                    requested_placement=RequestedPlacement(),
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image="app:dev"
                 ),
                 id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
             ),
-            pytest.param(
-                {
-                    "push_to": REPOSITORY,
-                    "placement": RequestedPlacement("listener-1", "agents"),
-                },
-                True,
-                CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "latest"),
-                    prebuilt_image=None,
-                    requested_placement=RequestedPlacement("listener-1", "agents"),
-                ),
-                id="push_to_carries_the_requested_placement",
-            ),
             pytest.param(
                 {"remote_build_flag": True},
                 True,
@@ -753,16 +720,6 @@ class TestSelectSource:
                 "--image cannot be combined with --remote builds.",
                 id="image_with_remote",
             ),
-            pytest.param(
-                {"placement": RequestedPlacement(listener_id="listener-1")},
-                "only apply when creating a deployment with --push-to",
-                id="listener_without_push_to",
-            ),
-            pytest.param(
-                {"placement": RequestedPlacement(k8s_namespace="agents")},
-                "only apply when creating a deployment with --push-to",
-                id="namespace_without_push_to",
-            ),
         ],
     )
     def test_conflicting_flags_are_rejected(self, monkeypatch, flags, message):
@@ -933,289 +890,3 @@ class TestResolvePushedImageDigest:
         frame_locals = captured["coro"].cr_frame.f_locals
         assert "--config" not in frame_locals["args"]
         captured["coro"].close()
-
-
-class TestListener:
-    @pytest.mark.parametrize(
-        ("resource", "expected"),
-        [
-            pytest.param(
-                {
-                    "id": "listener-1",
-                    "compute_id": "prod-cluster",
-                    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
-                },
-                Listener("listener-1", "prod-cluster", ("agents", "agents-staging")),
-                id="reads_id_cluster_and_namespaces",
-            ),
-            pytest.param(
-                {"id": "listener-1", "compute_id": "c", "compute_config": {}},
-                Listener("listener-1", "c", ()),
-                id="missing_namespaces",
-            ),
-            pytest.param(
-                {"id": "listener-1", "compute_id": "c", "compute_config": None},
-                Listener("listener-1", "c", ()),
-                id="null_compute_config",
-            ),
-            pytest.param(
-                {"id": "listener-1"},
-                Listener("listener-1", "", ()),
-                id="only_an_id",
-            ),
-        ],
-    )
-    def test_from_resource_reads_the_control_plane_shape(self, resource, expected):
-        assert Listener.from_resource(resource) == expected
-
-
-ONE_NAMESPACE = Listener("listener-1", "prod-cluster", ("agents",))
-TWO_NAMESPACES = Listener("listener-2", "multi-cluster", ("agents", "agents-staging"))
-NO_NAMESPACE = Listener("listener-3", "broken-cluster", ())
-
-
-class TestRequestedPlacement:
-    @pytest.mark.parametrize(
-        ("request_", "listeners", "expected"),
-        [
-            pytest.param(
-                RequestedPlacement(), (), Unplaced(), id="no_listeners_no_request"
-            ),
-            pytest.param(
-                RequestedPlacement(),
-                (ONE_NAMESPACE,),
-                OnListener("listener-1", "agents"),
-                id="uses_the_only_possible_answer",
-            ),
-            pytest.param(
-                RequestedPlacement(k8s_namespace="agents-staging"),
-                (TWO_NAMESPACES,),
-                OnListener("listener-2", "agents-staging"),
-                id="namespace_alone_picks_the_only_listener",
-            ),
-        ],
-    )
-    def test_resolves_to_a_placement(self, request_, listeners, expected):
-        assert request_.among(listeners) == expected
-
-    @pytest.mark.parametrize(
-        ("request_", "listeners", "message"),
-        [
-            pytest.param(
-                RequestedPlacement(listener_id="listener-1"),
-                (),
-                "no listeners",
-                id="workspace_has_no_listeners",
-            ),
-            pytest.param(
-                RequestedPlacement(),
-                (ONE_NAMESPACE, TWO_NAMESPACES),
-                "--listener-id",
-                id="several_listeners_need_a_choice",
-            ),
-            pytest.param(
-                RequestedPlacement(k8s_namespace="agents"),
-                (ONE_NAMESPACE, TWO_NAMESPACES),
-                "--listener-id",
-                id="namespace_alone_is_ambiguous_with_several_listeners",
-            ),
-            pytest.param(
-                RequestedPlacement(k8s_namespace="agents"),
-                (),
-                "no listeners",
-                id="namespace_without_any_listener",
-            ),
-            pytest.param(
-                RequestedPlacement(),
-                (TWO_NAMESPACES,),
-                "--k8s-namespace",
-                id="several_namespaces_need_a_choice",
-            ),
-        ],
-    )
-    def test_refuses_and_names_the_choices(self, request_, listeners, message):
-        with pytest.raises(click.UsageError, match=message):
-            request_.among(listeners)
-
-    def test_the_error_lists_every_listener_with_its_cluster_and_namespaces(self):
-        with pytest.raises(click.UsageError) as error:
-            RequestedPlacement().among((ONE_NAMESPACE, TWO_NAMESPACES))
-
-        assert "listener-1" in error.value.message
-        assert "prod-cluster" in error.value.message
-        assert "agents-staging" in error.value.message
-
-    @pytest.mark.parametrize(
-        ("placement", "expected"),
-        [
-            pytest.param(Unplaced(), {}, id="unplaced_adds_nothing"),
-            pytest.param(
-                OnListener("listener-1", "agents"),
-                {
-                    "listener_id": "listener-1",
-                    "listener_config": {"k8s_namespace": "agents"},
-                },
-                id="placed_carries_listener_and_namespace",
-            ),
-        ],
-    )
-    def test_source_config_matches_the_control_plane_shape(self, placement, expected):
-        assert placement.source_config() == expected
-
-
-def test_finding_a_deployment_by_name_narrows_the_search_for_every_server_version():
-    seen: dict = {}
-
-    def handler(req: httpx.Request) -> httpx.Response:
-        seen["params"] = dict(req.url.params)
-        return httpx.Response(
-            200,
-            json={"resources": [{"id": "dep-1", "name": "agent", "source": "github"}]},
-        )
-
-    client = HostBackendClient(
-        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
-    )
-
-    found = find_deployment_by_name(client, "agent")
-
-    assert seen["params"] == {
-        "name": "agent",
-        "name_contains": "agent",
-        "limit": "100",
-    }
-    assert found == ExistingDeployment("dep-1", "github")
-
-
-def test_a_server_that_ignores_the_exact_name_filter_never_matches_another_deployment():
-    client = HostBackendClient(
-        "https://api.example.com",
-        "key",
-        transport=httpx.MockTransport(
-            lambda req: httpx.Response(
-                200,
-                json={
-                    "resources": [
-                        {
-                            "id": "dep-other",
-                            "name": "another-teams-agent",
-                            "source": "external_docker",
-                        }
-                    ]
-                },
-            )
-        ),
-    )
-
-    assert find_deployment_by_name(client, "brand-new-agent") is None
-
-
-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)
-    ]
-    client = HostBackendClient(
-        "https://api.example.com",
-        "key",
-        transport=httpx.MockTransport(
-            lambda req: httpx.Response(200, json={"resources": page})
-        ),
-    )
-
-    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():
-    client = HostBackendClient(
-        "https://api.example.com",
-        "key",
-        transport=httpx.MockTransport(
-            lambda req: httpx.Response(
-                200, json={"resources": [{"id": "dep-1", "name": "other"}]}
-            )
-        ),
-    )
-
-    assert find_deployment_by_name(client, "brand-new-agent") is None
-
-
-@pytest.mark.parametrize(
-    "resource",
-    [
-        pytest.param({"compute_id": "c"}, id="no_id"),
-        pytest.param({"id": ""}, id="empty_id"),
-    ],
-)
-def test_a_listener_without_an_id_is_refused(resource):
-    with pytest.raises(HostBackendError, match="without an id"):
-        Listener.from_resource(resource)
-
-
-def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
-    monkeypatch,
-):
-    def explode() -> tuple[bool, str | None]:
-        raise AssertionError("docker must not be probed for an argv-only conflict")
-
-    monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
-
-    with pytest.raises(click.UsageError, match="--deployment-id"):
-        _select_source(
-            push_to="registry.example.com/app",
-            image=None,
-            image_name=None,
-            tag=None,
-            remote_build_flag=None,
-            placement=RequestedPlacement(listener_id="listener-1"),
-            selector=ById("dep-1"),
-        )
-
-
-class TestPlacementOnAKnownListener:
-    @pytest.mark.parametrize(
-        ("request_", "listener", "expected"),
-        [
-            pytest.param(
-                RequestedPlacement(listener_id="listener-1"),
-                ONE_NAMESPACE,
-                OnListener("listener-1", "agents"),
-                id="the_only_namespace_is_used",
-            ),
-            pytest.param(
-                RequestedPlacement(listener_id="listener-2", k8s_namespace="agents"),
-                TWO_NAMESPACES,
-                OnListener("listener-2", "agents"),
-                id="the_chosen_namespace_is_used",
-            ),
-        ],
-    )
-    def test_places_on_the_listener(self, request_, listener, expected):
-        assert request_.on(listener) == expected
-
-    @pytest.mark.parametrize(
-        ("request_", "listener", "message"),
-        [
-            pytest.param(
-                RequestedPlacement(listener_id="listener-2"),
-                TWO_NAMESPACES,
-                "--k8s-namespace",
-                id="several_namespaces_need_a_choice",
-            ),
-            pytest.param(
-                RequestedPlacement(listener_id="listener-2", k8s_namespace="nope"),
-                TWO_NAMESPACES,
-                "does not serve namespace",
-                id="unknown_namespace",
-            ),
-            pytest.param(
-                RequestedPlacement(listener_id="listener-3"),
-                NO_NAMESPACE,
-                "serves no namespaces",
-                id="listener_without_namespaces",
-            ),
-        ],
-    )
-    def test_refuses_and_names_the_namespaces(self, request_, listener, message):
-        with pytest.raises(click.UsageError, match=message):
-            request_.on(listener)

libs/cli/tests/unit_tests/test_host_backend.py

@@ -79,6 +79,19 @@ def test_request_transport_error_raises():
         c._request("GET", "/test")
 
 
+def test_list_deployments_sends_query_params():
+    def handler(req: httpx.Request) -> httpx.Response:
+        assert req.url.path == "/v2/deployments"
+        assert req.url.params["name_contains"] == "my app"
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient(
+        "https://api.example.com", "test-key", transport=httpx.MockTransport(handler)
+    )
+    result = c.list_deployments("my app")
+    assert result == {"ok": True}
+
+
 def _capturing_client(captured: dict) -> HostBackendClient:
     def handler(req: httpx.Request) -> httpx.Response:
         captured["body"] = req.read()
@@ -408,7 +421,7 @@ def test_injected_transport_receives_requests_under_the_prefixed_base_url():
         transport=httpx.MockTransport(handler),
     )
 
-    assert c.list_revisions("dep-1", limit=2) == []
+    assert c.list_revisions("dep-1", limit=2) == {"ok": True}
     assert seen == {
         "url": "https://smith.example.com/api-host/v2/deployments/dep-1/revisions?limit=2",
         "api_key": "key",
@@ -533,144 +546,3 @@ def test_control_plane_endpoints_resolve(host_url, langsmith_endpoint, expected)
     endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint)
 
     assert (endpoints.control_plane_url, endpoints.dashboard_url) == expected
-
-
-@pytest.mark.parametrize(
-    ("payload", "expected"),
-    [
-        pytest.param(
-            {"resources": [{"id": "a"}, {"id": "b"}]},
-            [{"id": "a"}, {"id": "b"}],
-            id="list_returns_the_resources",
-        ),
-        pytest.param({"resources": []}, [], id="empty_list"),
-        pytest.param({}, [], id="missing_key"),
-        pytest.param({"resources": None}, [], id="null_resources"),
-        pytest.param(
-            {"resources": ["nope", {"id": "a"}]}, [{"id": "a"}], id="skips_non_objects"
-        ),
-        pytest.param([], [], id="unexpected_envelope"),
-    ],
-)
-def test_list_endpoints_return_resource_objects(payload, expected):
-    def handler(req: httpx.Request) -> httpx.Response:
-        return httpx.Response(200, json=payload)
-
-    c = HostBackendClient(
-        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
-    )
-
-    assert c.list_deployments() == expected
-
-
-def test_list_listeners_asks_for_a_full_page():
-    seen: dict = {}
-
-    def handler(req: httpx.Request) -> httpx.Response:
-        seen["url"] = str(req.url)
-        return httpx.Response(200, json={"resources": [{"id": "listener-1"}]})
-
-    c = HostBackendClient(
-        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
-    )
-
-    assert c.list_listeners() == [{"id": "listener-1"}]
-    assert seen["url"] == "https://api.example.com/v2/listeners?limit=100"
-
-
-@pytest.mark.parametrize(
-    ("control_plane_url", "expected"),
-    [
-        pytest.param("https://api.host.langchain.com", True, id="cloud"),
-        pytest.param("https://eu.api.host.langchain.com", True, id="cloud_region"),
-        pytest.param("https://dev.api.host.langchain.com", True, id="cloud_dev"),
-        pytest.param("https://smith.example.com/api-host", False, id="self_hosted"),
-        pytest.param(
-            "https://corp.example.com/langsmith/api-host",
-            False,
-            id="self_hosted_prefix",
-        ),
-        pytest.param("http://localhost:8080/api-host", False, id="local"),
-        pytest.param(
-            "https://evil-api.host.langchain.com", False, id="lookalike_needs_a_dot"
-        ),
-    ],
-)
-def test_is_cloud_recognises_the_managed_control_plane(control_plane_url, expected):
-    endpoints = ControlPlaneEndpoints.from_control_plane_url(control_plane_url)
-
-    assert endpoints.is_cloud is expected
-
-
-@pytest.mark.parametrize(
-    ("call", "expected_params"),
-    [
-        pytest.param(
-            lambda c: c.list_deployments(name="agent"),
-            {"name": "agent"},
-            id="exact_name_filters_server_side",
-        ),
-        pytest.param(
-            lambda c: c.list_deployments(name_contains="age"),
-            {"name_contains": "age"},
-            id="substring_search_keeps_its_own_parameter",
-        ),
-        pytest.param(
-            lambda c: c.list_deployments(),
-            {},
-            id="no_filter_sends_no_parameters",
-        ),
-        pytest.param(
-            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",
-        ),
-    ],
-)
-def test_list_deployments_sends_one_name_filter(call, expected_params):
-    seen: dict = {}
-
-    def handler(req: httpx.Request) -> httpx.Response:
-        seen.update(dict(req.url.params))
-        return httpx.Response(200, json={"resources": []})
-
-    call(
-        HostBackendClient(
-            "https://api.example.com", "key", transport=httpx.MockTransport(handler)
-        )
-    )
-
-    assert seen == expected_params
-
-
-@pytest.mark.parametrize(
-    ("body", "expected"),
-    [
-        pytest.param(
-            {"detail": "Source configuration error: bad listener"},
-            "Source configuration error: bad listener",
-            id="fastapi_detail_is_unwrapped",
-        ),
-        pytest.param(
-            {"detail": {"loc": ["body"], "msg": "nope"}},
-            None,
-            id="a_structured_detail_is_left_alone",
-        ),
-        pytest.param({"other": "shape"}, None, id="an_unknown_shape_is_left_alone"),
-    ],
-)
-def test_error_detail_is_readable(body, expected):
-    c = HostBackendClient(
-        "https://api.example.com",
-        "key",
-        transport=httpx.MockTransport(lambda req: httpx.Response(400, json=body)),
-    )
-
-    with pytest.raises(HostBackendError) as error:
-        c.get_deployment("dep-1")
-
-    assert error.value.detail == expected
-    if expected is not None:
-        assert error.value.message.endswith(expected)

libs/langgraph/tests/test_delta_channel_parallel_order.py

@@ -0,0 +1,83 @@
+"""`DeltaChannel` replay must apply parallel writes in the order `invoke` did."""
+
+from typing import Annotated, Any
+
+import pytest
+from langgraph.checkpoint.base import BaseCheckpointSaver
+from typing_extensions import TypedDict
+
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import END, START, StateGraph
+
+pytestmark = pytest.mark.anyio
+
+# Sorted, because live execution applies PULL tasks in node-name order.
+FAN_OUT_NAMES = ["a", "b", "c", "d", "e", "f", "g", "h"]
+
+
+def _append_reducer(current: list, updates: list) -> list:
+    return [*current, *(x for u in updates for x in u)]
+
+
+def _build_fan_out_graph(checkpointer: BaseCheckpointSaver) -> Any:
+    class State(TypedDict):
+        items: Annotated[
+            list, DeltaChannel(_append_reducer, list, snapshot_frequency=10_000)
+        ]
+
+    def make_node(label: str) -> Any:
+        return lambda state: {"items": [label]}
+
+    builder = StateGraph(State)
+    for name in FAN_OUT_NAMES:
+        builder.add_node(name, make_node(name))
+        builder.add_edge(START, name)
+        builder.add_edge(name, END)
+    return builder.compile(checkpointer=checkpointer)
+
+
+async def test_get_state_matches_live_invoke_order(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _build_fan_out_graph(async_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+
+    live = (await graph.ainvoke({"items": []}, config))["items"]
+    replayed = (await graph.aget_state(config)).values["items"]
+
+    assert live == FAN_OUT_NAMES
+    assert replayed == live
+
+
+async def test_continuing_thread_preserves_committed_prefix(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _build_fan_out_graph(async_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+
+    first = (await graph.ainvoke({"items": []}, config))["items"]
+    second = (await graph.ainvoke({"items": []}, config))["items"]
+
+    assert second == first + first
+    assert (await graph.aget_state(config)).values["items"] == second
+
+
+async def test_state_history_reports_live_order_at_every_step(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    runs = 3
+    graph = _build_fan_out_graph(async_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+    for _ in range(runs):
+        await graph.ainvoke({"items": []}, config)
+    live = FAN_OUT_NAMES * runs
+
+    seen = [
+        s.values["items"]
+        async for s in graph.aget_state_history(config)
+        if "items" in s.values
+    ]
+
+    assert max(map(len, seen)) == len(live)
+    for values in seen:
+        assert values == live[: len(values)], f"{values} is not a prefix of {live}"