docs(checkpoint-sqlite): state the id ordering contract correctly in the walk comment

Checkpoint ids are documented as monotonic; the guarantee only holds within
one process, which is why the walk follows parent pointers.

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

@@ -267,61 +267,6 @@ 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 = "ffffffff-ffff-ffff-ffff-ffffffffffff"
-
-
-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,
@@ -331,8 +276,6 @@ 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,7 +168,6 @@ 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
 
@@ -320,7 +319,7 @@ def _build_delta_stage2_sql(
         branches.append(
             "SELECT 'w'::text AS _kind, "
             "checkpoint_id, channel, "
-            "type, blob, task_id, task_path, idx, NULL::text AS version "
+            "type, blob, task_id, idx, NULL::text AS version "
             "FROM checkpoint_writes "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND checkpoint_id = ANY(%s)"
@@ -328,8 +327,7 @@ 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::text AS task_path, "
-            "NULL::int AS idx, version "
+            "type, blob, NULL::text AS task_id, NULL::int AS idx, version "
             "FROM checkpoint_blobs "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND version = %s"
@@ -494,11 +492,10 @@ 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, task_path)
-        writes_by_ch_by_cid: dict[
-            str, dict[str, list[tuple[str, bytes, str, int, str]]]
-        ] = {ch: {} for ch in channels}
+        # 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
+        }
         # seed_blob_by_ver[(channel, version)] = (type, blob)
         seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
 
@@ -509,14 +506,8 @@ 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, str]",
-                        (
-                            r["type"],
-                            r["blob"],
-                            r["task_id"],
-                            r["idx"],
-                            r["task_path"],
-                        ),
+                        "tuple[str, bytes, str, int]",
+                        (r["type"], r["blob"], r["task_id"], r["idx"]),
                     )
                 )
             else:  # kind == "b"
@@ -525,10 +516,10 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                     "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-        # Sort writes per (channel, cid) newest-first by (task_path, task_id, idx)
+        # Sort writes per (channel, cid) newest-first by (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[4], w[2], w[3]), reverse=True)
+                ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
@@ -538,9 +529,7 @@ 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, _path in cid_writes.get(
-                    cid, []
-                ):
+                for type_tag, write_blob, task_id, _idx 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

@@ -81,7 +81,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
     conn: sqlite3.Connection
     is_setup: bool
-    _has_task_path: bool = True
 
     def __init__(
         self,
@@ -155,7 +154,6 @@ 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,
@@ -164,19 +162,6 @@ 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:
-            # A read-only database from before the column can still be read;
-            # its rows would all read back as '' anyway.
-            if "readonly database" in str(e):
-                self._has_task_path = False
-            elif "duplicate column name" not in str(e):
-                raise
 
         self.is_setup = True
 
@@ -475,9 +460,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, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, 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, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
         )
         with self.cursor() as cur:
             cur.executemany(
@@ -488,7 +473,6 @@ 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),
@@ -523,13 +507,12 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
         Two-stage query:
 
-        * Stage 1 (paged): newest-first slice of `checkpoints` returning
-          `(checkpoint_id, parent_checkpoint_id, type, checkpoint)` per
-          ancestor. Sqlite has no JSONB, so we ship the full serialized
-          checkpoint blob and inspect `channel_values` in Python. Pages
-          newest-first by `checkpoint_id` with a `< cursor` predicate;
-          page size is `DELTA_PAGE_SIZE`. Stops paging when every channel
-          has found its seed or the chain is exhausted.
+        * Stage 1 (streamed): recursive CTE over `checkpoints` following
+          `parent_checkpoint_id` from the target, returning
+          `(checkpoint_id, type, checkpoint)` per ancestor. Sqlite has no
+          JSONB, so we ship the full serialized checkpoint blob and inspect
+          `channel_values` in Python. Stops reading when every channel has
+          found its seed or the chain is exhausted.
 
         * Stage 2 (per-channel UNION ALL): one branch per channel reading
           `writes` filtered to that channel's specific `chain_cids`. No
@@ -554,12 +537,14 @@ class SqliteSaver(BaseCheckpointSaver[str]):
         seeded: set[str] = set()
 
         with self.cursor(transaction=False) as cur:
-            cur.execute(DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id))
+            cur.execute(
+                DELTA_STAGE1_SQL,
+                (thread_id, checkpoint_ns, checkpoint_id, thread_id, checkpoint_ns),
+            )
             for row in cur:
-                cid, parent_cid, type_tag, blob = row
+                cid, type_tag, blob = row
                 if step_walk_with_row(
                     cid=cid,
-                    parent_cid=parent_cid,
                     type_tag=type_tag,
                     blob=blob,
                     target_id=checkpoint_id,
@@ -574,7 +559,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
             channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
             stage2_sql = build_delta_stage2_sql(
-                has_task_path=self._has_task_path,
                 chain_lens=[len(chain_by_ch[ch]) for ch in channels_with_chain],
             )
             if stage2_sql:
@@ -585,7 +569,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                     )
                 cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes, str]]", cur.fetchall()
+                    "list[tuple[str, str, str, int, str, bytes]]", cur.fetchall()
                 )
             else:
                 stage2_rows = []

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

@@ -26,22 +26,37 @@ from typing import Any
 
 from langgraph.checkpoint.base import DeltaChannelHistory, PendingWrite
 
-# Stage 1 streams ancestors of `target_cid` newest-first. The `<=`
-# predicate keeps target itself in the stream so we can read its
-# `parent_checkpoint_id` from the first row without a separate lookup;
-# the caller skips target's own writes/seed (matches the
-# `BaseCheckpointSaver` contract).
+# Stage 1 streams target, then its ancestors nearest-first, by following
+# `parent_checkpoint_id` rather than id order: ids are only monotonic within
+# one process, so a range scan by id can miss a parent whose id sorts above
+# its child's. Target is the anchor row; its own writes/seed are skipped
+# (matches the `BaseCheckpointSaver` contract).
+#
+# `put` is `INSERT OR REPLACE`, so re-putting an existing id under a
+# descendant's config makes the chain a loop. `step_walk_with_row` stops on a
+# repeated id; sqlite yields recursive rows lazily, so abandoning the cursor
+# ends the recursion.
+#
+# `CROSS JOIN` pins `ancestors` as the outer loop, so each step is one primary
+# key lookup. With a plain `JOIN` and no `ANALYZE` stats, sqlite can put
+# `checkpoints` outside and scan the whole thread per step.
 DELTA_STAGE1_SQL = (
+    "WITH RECURSIVE ancestors(checkpoint_id, parent_checkpoint_id, type, "
+    "checkpoint) AS ("
     "SELECT checkpoint_id, parent_checkpoint_id, type, checkpoint "
     "FROM checkpoints "
-    "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id <= ? "
-    "ORDER BY checkpoint_id DESC"
+    "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? "
+    "UNION ALL "
+    "SELECT c.checkpoint_id, c.parent_checkpoint_id, c.type, c.checkpoint "
+    "FROM ancestors a CROSS JOIN checkpoints c "
+    "ON c.checkpoint_id = a.parent_checkpoint_id "
+    "WHERE c.thread_id = ? AND c.checkpoint_ns = ?"
+    ") "
+    "SELECT checkpoint_id, type, checkpoint FROM ancestors"
 )
 
 
-def build_delta_stage2_sql(
-    *, chain_lens: Sequence[int], has_task_path: bool = True
-) -> str:
+def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
     """Stage-2 per-channel UNION ALL fetching writes from `writes`.
 
     One branch per channel with a non-empty chain. Each branch inlines its
@@ -55,12 +70,11 @@ def build_delta_stage2_sql(
     of a single `channel = ANY(channels)` filter when channels have
     different chain depths — same rationale as postgres.
     """
-    task_path = "task_path" if has_task_path else "''"
     branches: list[str] = []
     for n in chain_lens:
         cid_placeholders = ",".join("?" * n)
         branches.append(
-            f"SELECT checkpoint_id, channel, task_id, idx, type, value, {task_path} "
+            "SELECT checkpoint_id, channel, task_id, idx, type, value "
             "FROM writes "
             "WHERE thread_id = ? AND checkpoint_ns = ? AND channel = ? "
             f"AND checkpoint_id IN ({cid_placeholders})"
@@ -71,7 +85,6 @@ def build_delta_stage2_sql(
 def step_walk_with_row(
     *,
     cid: str,
-    parent_cid: str | None,
     type_tag: str,
     blob: bytes,
     target_id: str,
@@ -84,36 +97,32 @@ def step_walk_with_row(
 ) -> bool:
     """Process one streamed stage-1 row in the merged ancestor walk.
 
-    The cursor returns (cid, parent_cid, type, blob) rows in
-    `checkpoint_id` DESC order starting at target. The first row is
-    target itself; we read its parent_cid to seed the walk and otherwise
-    skip it (target's own writes/seed are not part of the contract).
+    The cursor returns (cid, type, blob) rows in walk order starting at
+    target. The first row is target itself and is skipped (target's own
+    writes/seed are not part of the contract).
 
-    For each subsequent row, if `cid` matches the walk's current
-    position, we deserialize the blob, append the cid to every
-    not-yet-seeded channel's chain, and check `channel_values` for
+    For each subsequent row we deserialize the blob, append the cid to
+    every not-yet-seeded channel's chain, and check `channel_values` for
     seeds. The deserialized checkpoint is dropped before advancing — no
     cross-row cache, so peak in-flight is one deserialized checkpoint.
 
-    Off-path rows (different branch on the same thread) advance the
-    cursor without doing any work.
-
-    Returns True when every requested channel is seeded — the caller
-    can stop iterating and close the cursor.
+    Returns True when the caller can stop iterating and close the cursor:
+    every requested channel is seeded, or the chain revisited a checkpoint.
     """
     if "started" not in walk_state:
         if cid == target_id:
             walk_state["started"] = True
-            walk_state["cur_cid"] = parent_cid
             walk_state["active"] = {ch for ch in channels if ch not in seeded}
+            walk_state["walked"] = {cid}
         # Not target yet (or target not present): keep streaming.
         return False
     active: set[str] = walk_state["active"]
     if not active:
         return True
-    if cid != walk_state["cur_cid"]:
-        # Off-path row from a sibling branch — skip without deserializing.
-        return False
+    walked: set[str] = walk_state["walked"]
+    if cid in walked:
+        return True
+    walked.add(cid)
     for ch in active:
         chain_by_ch[ch].append(cid)
     ckpt = serde.loads_typed((type_tag, blob))
@@ -123,7 +132,6 @@ def step_walk_with_row(
         seeded.add(ch)
         active.discard(ch)
     del ckpt, channel_values
-    walk_state["cur_cid"] = parent_cid
     return not active
 
 
@@ -133,31 +141,29 @@ 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, str]],
+    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes]],
     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, task_path)`.
-    Final write order is oldest→newest globally and
-    `(task_path, task_id, idx)` within a checkpoint, matching the contract
-    on `DeltaChannelHistory.writes`.
+    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`.
 
     `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, 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: 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.setdefault(ch, {}).setdefault(cid, []).append(
-            (type_tag, value_blob, task_id, idx, task_path)
+            (type_tag, value_blob, 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[4], w[2], w[3]))
+            ws.sort(key=lambda w: (w[2], w[3]))
 
     result: dict[str, DeltaChannelHistory] = {}
     for ch in channels:
@@ -166,7 +172,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, _path in cid_writes.get(cid, []):
+            for type_tag, value_blob, task_id, _idx 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

@@ -114,7 +114,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
     lock: asyncio.Lock
     is_setup: bool
-    _has_task_path: bool = True
 
     def __init__(
         self,
@@ -332,7 +331,6 @@ 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,
@@ -343,21 +341,6 @@ 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:
-                # A read-only database from before the column can still be read;
-                # its rows would all read back as '' anyway.
-                if "readonly database" in str(e):
-                    self._has_task_path = False
-                elif "duplicate column name" not in str(e):
-                    raise
-
             self.is_setup = True
 
     async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
@@ -593,9 +576,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, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, 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, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
         )
         await self.setup()
         async with self.lock, self.conn.cursor() as cur:
@@ -607,7 +590,6 @@ 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),
@@ -643,8 +625,8 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
         """Fast-path override of `BaseCheckpointSaver.aget_delta_channel_history`.
 
         See `SqliteSaver.get_delta_channel_history` for design notes; this
-        is the async equivalent using `aiosqlite` cursors. Stage 1 pages
-        the parent chain newest-first and Python-deserializes each
+        is the async equivalent using `aiosqlite` cursors. Stage 1 streams
+        the parent chain from the target and Python-deserializes each
         checkpoint blob to find per-channel snapshots; stage 2 fetches
         only the relevant writes via per-channel UNION ALL.
         """
@@ -668,13 +650,13 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
         async with self.lock, self.conn.cursor() as cur:
             await cur.execute(
-                DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id)
+                DELTA_STAGE1_SQL,
+                (thread_id, checkpoint_ns, checkpoint_id, thread_id, checkpoint_ns),
             )
             async for row in cur:
-                cid, parent_cid, type_tag, blob = row
+                cid, type_tag, blob = row
                 if step_walk_with_row(
                     cid=cid,
-                    parent_cid=parent_cid,
                     type_tag=type_tag,
                     blob=blob,
                     target_id=checkpoint_id,
@@ -689,7 +671,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
             channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
             stage2_sql = build_delta_stage2_sql(
-                has_task_path=self._has_task_path,
                 chain_lens=[len(chain_by_ch[ch]) for ch in channels_with_chain],
             )
             if stage2_sql:
@@ -700,7 +681,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                     )
                 await cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes, str]]",
+                    "list[tuple[str, str, str, int, str, bytes]]",
                     await cur.fetchall(),
                 )
             else:

libs/checkpoint-sqlite/tests/test_delta_parent_walk.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from langgraph.checkpoint.base import (
+    BaseCheckpointSaver,
+    Checkpoint,
+    DeltaChannelHistory,
+    empty_checkpoint,
+)
+
+from langgraph.checkpoint.sqlite import SqliteSaver
+from langgraph.checkpoint.sqlite._delta import DELTA_STAGE1_SQL
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+CHANNEL = "ch"
+CONFIG: dict[str, Any] = {"configurable": {"thread_id": "t", "checkpoint_ns": ""}}
+EXPECTED: DeltaChannelHistory = {
+    "writes": [("task", CHANNEL, "write-root")],
+    "seed": "seed",
+}
+
+
+def _checkpoint(checkpoint_id: str, values: dict[str, Any]) -> Checkpoint:
+    value = empty_checkpoint()
+    value["id"] = checkpoint_id
+    value["channel_values"] = values
+    return value
+
+
+PARENT_ID_ORDERS = [
+    pytest.param("z-older", "a-newer", id="parent_id_sorts_above_child"),
+    pytest.param("a-older", "z-newer", id="parent_id_sorts_below_child"),
+]
+
+
+@pytest.mark.parametrize(("root_id", "child_id"), PARENT_ID_ORDERS)
+def test_sync_walk_reaches_parent_whatever_the_id_order(
+    root_id: str, child_id: str
+) -> None:
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        root = saver.put(CONFIG, _checkpoint(root_id, {CHANNEL: "seed"}), {}, {})
+        saver.put_writes(root, [(CHANNEL, "write-root")], "task")
+        child = saver.put(root, _checkpoint(child_id, {}), {}, {})
+
+        got = saver.get_delta_channel_history(config=child, channels=[CHANNEL])
+        reference = BaseCheckpointSaver.get_delta_channel_history(
+            saver, config=child, channels=[CHANNEL]
+        )
+        assert got[CHANNEL] == EXPECTED
+        assert got[CHANNEL] == reference[CHANNEL], "fast path disagrees with base"
+
+
+@pytest.mark.parametrize(("root_id", "child_id"), PARENT_ID_ORDERS)
+async def test_async_walk_reaches_parent_whatever_the_id_order(
+    root_id: str, child_id: str
+) -> None:
+    async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
+        root = await saver.aput(CONFIG, _checkpoint(root_id, {CHANNEL: "seed"}), {}, {})
+        await saver.aput_writes(root, [(CHANNEL, "write-root")], "task")
+        child = await saver.aput(root, _checkpoint(child_id, {}), {}, {})
+
+        got = await saver.aget_delta_channel_history(config=child, channels=[CHANNEL])
+        assert got[CHANNEL] == EXPECTED
+
+
+def test_walk_reaches_root_of_long_chain_with_descending_ids() -> None:
+    steps = 40
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        parent = saver.put(
+            CONFIG, _checkpoint(f"id-{steps:03d}", {CHANNEL: "seed"}), {}, {}
+        )
+        saver.put_writes(parent, [(CHANNEL, "write-root")], "task")
+        for step in range(steps - 1, 0, -1):
+            parent = saver.put(parent, _checkpoint(f"id-{step:03d}", {}), {}, {})
+
+        got = saver.get_delta_channel_history(config=parent, channels=[CHANNEL])
+        assert got[CHANNEL] == EXPECTED
+
+
+def test_walk_terminates_when_put_makes_the_parent_chain_cycle() -> None:
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        a = saver.put(CONFIG, _checkpoint("cid-a", {}), {}, {})
+        b = saver.put(a, _checkpoint("cid-b", {}), {}, {})
+        repoint_a_under_b = _checkpoint("cid-a", {})
+        saver.put(b, repoint_a_under_b, {}, {})
+
+        got = saver.get_delta_channel_history(config=b, channels=[CHANNEL])
+        assert got[CHANNEL] == {"writes": []}
+
+
+def test_walk_step_looks_up_the_parent_by_primary_key() -> None:
+    with SqliteSaver.from_conn_string(":memory:") as saver:
+        saver.setup()
+        plan = [
+            row[3]
+            for row in saver.conn.execute(
+                f"EXPLAIN QUERY PLAN {DELTA_STAGE1_SQL}", ("t", "", "id", "t", "")
+            )
+        ]
+    assert any(
+        step.startswith("SEARCH c ") and "checkpoint_id=?" in step for step in plan
+    ), f"recursive step should look up the parent by key, got {plan}"

libs/checkpoint-sqlite/tests/test_writes_task_path_migration.py

@@ -1,128 +0,0 @@
-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"),
-            ]
-
-
-def _legacy_database_with_history(db: Path) -> dict:
-    root = empty_checkpoint()
-    root["channel_values"] = {"ch": "seed"}
-    root["channel_versions"] = {"ch": 1}
-    with SqliteSaver.from_conn_string(str(db)) as saver:
-        root_config = saver.put(
-            {"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
-            root,
-            {},
-            {"ch": 1},
-        )
-        saver.put_writes(root_config, [("ch", "write")], "task", "~__pregel_pull, n")
-        child = saver.put(root_config, empty_checkpoint(), {}, {})
-        saver.conn.execute("ALTER TABLE writes DROP COLUMN task_path")
-        saver.conn.commit()
-    return child
-
-
-def test_read_only_legacy_database_still_reads_delta_history(tmp_path: Path) -> None:
-    db = tmp_path / "legacy.sqlite"
-    child = _legacy_database_with_history(db)
-
-    saver = SqliteSaver(sqlite3.connect(f"file:{db}?mode=ro", uri=True))
-    got = saver.get_delta_channel_history(config=child, channels=["ch"])
-
-    assert got["ch"] == {"seed": "seed", "writes": [("task", "ch", "write")]}
-
-
-async def test_async_read_only_legacy_database_still_reads_delta_history(
-    tmp_path: Path,
-) -> None:
-    db = tmp_path / "legacy.sqlite"
-    child = _legacy_database_with_history(db)
-
-    async with aiosqlite.connect(f"file:{db}?mode=ro", uri=True) as conn:
-        saver = AsyncSqliteSaver(conn)
-        got = await saver.aget_delta_channel_history(config=child, channels=["ch"])
-
-    assert got["ch"] == {"seed": "seed", "writes": [("task", "ch", "write")]}

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

@@ -162,14 +162,6 @@ 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)`, which is the order live execution applies
-      a super-step's task writes in. `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, `update_state` updates, exit-durability runs,
-      rows predating the column) sort first, by `task_id`.
     * `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
@@ -619,11 +611,6 @@ 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 _, (tid, ch, serialized, _) in sorted(
-                step_writes.items(), key=lambda kv: (kv[1][3], kv[0]), reverse=True
+            for (_task_id, _idx), (tid, ch, serialized, _) in sorted(
+                step_writes.items(), reverse=True
             ):
                 if ch not in remaining:
                     continue

libs/cli/README.md

@@ -103,8 +103,6 @@ The CLI uses a `langgraph.json` configuration file with these key settings:
 }
 ```
 
-Git dependencies should use credential-free URLs. The CLI conservatively scans direct `langgraph.json` dependencies, common Python package files, uv project and lock files, and common Node.js package and lock files for HTTP Git URLs with userinfo. This check is not exhaustive: generated Docker builds can copy other files, including nested requirement or constraint files, into image layers without scanning them. For private dependencies, provide short-lived credentials through your build environment's secret-backed Git credential helper. Do not store credentials in copied files such as `langgraph.json` or `pip_config_file`.
-
 See the [full documentation](https://reference.langchain.com/python/langgraph-cli) for detailed configuration options.
 
 ## Development

libs/cli/langgraph_cli/__init__.py

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

libs/cli/langgraph_cli/config.py

@@ -6,7 +6,6 @@ import re
 import shlex
 import textwrap
 from collections import Counter
-from collections.abc import Iterable
 from typing import Literal, NamedTuple
 
 import click
@@ -37,10 +36,6 @@ DISALLOWED_BUILD_COMMAND_CHARS = [
 # This blocks background execution (cmd &) while allowing command
 # chaining (cmd1 && cmd2) which is common in build commands.
 _SINGLE_AMPERSAND_RE = re.compile(r"(?<!&)&(?:&&)*(?!&)")
-_GIT_HTTP_AUTHORITY_RES = (
-    re.compile(r"git\+https?://(?P<authority>[^/\s\"']+)", re.I),
-    re.compile(r"\bgit\s*=\s*[\"']https?://(?P<authority>[^/\s\"']+)", re.I),
-)
 _API_VERSION_PATTERN = re.compile(
     r"^(?P<major>\d+)"
     r"(?:\.(?P<minor>\d+))?"
@@ -83,62 +78,6 @@ def has_disallowed_build_command_content(command: str) -> bool:
     return False
 
 
-def _has_git_http_url_userinfo(dependency: str) -> bool:
-    """Check whether a Git HTTP URL contains userinfo."""
-    return any(
-        "@" in match.group("authority")
-        for pattern in _GIT_HTTP_AUTHORITY_RES
-        for match in pattern.finditer(dependency)
-    )
-
-
-def _validate_git_http_url_userinfo(
-    values: Iterable[str], *, source: pathlib.Path | None = None
-) -> None:
-    """Reject credential-bearing Git HTTP URLs without echoing their values."""
-    if not any(_has_git_http_url_userinfo(value) for value in values):
-        return
-    message = (
-        "Git dependency URLs must not contain credentials or other URL "
-        "userinfo because generated Dockerfiles and image layers can retain "
-        "them. Use a credential-free Git URL and provide short-lived "
-        "credentials through your build environment's secret-backed Git "
-        "credential helper."
-    )
-    if source is not None:
-        message += f" Found in: {source}"
-    raise click.UsageError(message)
-
-
-def _validate_git_http_url_userinfo_files(paths: Iterable[pathlib.Path]) -> None:
-    """Reject credential-bearing Git HTTP URLs in dependency files."""
-    for path in paths:
-        path = path.resolve()
-        if not path.is_file():
-            continue
-        try:
-            contents = path.read_text(encoding="utf-8", errors="replace")
-        except OSError:
-            raise click.UsageError(
-                f"Could not inspect dependency file for embedded credentials: {path}"
-            ) from None
-        _validate_git_http_url_userinfo([contents], source=path)
-
-
-def _validate_local_dependency_files(config_path: pathlib.Path, config: Config) -> None:
-    """Validate dependency files copied into a non-uv Python image."""
-    paths: list[pathlib.Path] = []
-    for dependency in config["dependencies"]:
-        if not isinstance(dependency, str) or not dependency.startswith("."):
-            continue
-        root = (config_path.parent / dependency).resolve()
-        paths.extend(
-            root / name
-            for name in ("requirements.txt", "pyproject.toml", "setup.py", "setup.cfg")
-        )
-    _validate_git_http_url_userinfo_files(paths)
-
-
 MIN_PYTHON_VERSION = "3.11"
 DEFAULT_PYTHON_VERSION = "3.11"
 
@@ -381,9 +320,7 @@ def _get_source_kind(config: Config) -> str | None:
     return kind if isinstance(kind, str) else None
 
 
-def validate_config(
-    config: Config, *, source_path: pathlib.Path | None = None
-) -> Config:
+def validate_config(config: Config) -> Config:
     """Validate a configuration dictionary."""
 
     graphs = config.get("graphs", {})
@@ -478,15 +415,6 @@ def validate_config(
                 '  "source": {"kind": "uv", "root": ".."}'
             )
 
-    _validate_git_http_url_userinfo(
-        (
-            dependency
-            for dependency in config["dependencies"]
-            if isinstance(dependency, str)
-        ),
-        source=source_path,
-    )
-
     source = config.get("source")
     source_kind = _get_source_kind(config)
     if source is not None and not isinstance(source, dict):
@@ -681,7 +609,7 @@ def validate_config_file(config_path: pathlib.Path) -> Config:
     """Load and validate a configuration file."""
     with open(config_path) as f:
         config = json.load(f)
-    validated = validate_config(config, source_path=config_path.resolve())
+    validated = validate_config(config)
     # Enforce the package.json doesn't enforce an
     # incompatible Node.js version
     if validated.get("node_version"):
@@ -1352,7 +1280,6 @@ def python_config_to_docker(
             api_version=api_version,
             build_tools_to_uninstall=build_tools_to_uninstall,
         )
-    _validate_local_dependency_files(config_path, config)
     if pip_installer == "auto":
         if _image_supports_uv(base_image):
             pip_installer = "uv"
@@ -1563,18 +1490,7 @@ def node_config_to_docker(
 ) -> tuple[str, dict[str, str]]:
     # Calculate paths for monorepo support
     install_root = (
-        pathlib.Path(build_context).resolve()
-        if build_context
-        else config_path.parent.resolve()
-    )
-    config_root = config_path.parent.resolve()
-    dependency_roots = (
-        (install_root, config_root) if install_root != config_root else (install_root,)
-    )
-    _validate_git_http_url_userinfo_files(
-        root / name
-        for root in dependency_roots
-        for name in ("package.json", "package-lock.json", "yarn.lock", "pnpm-lock.yaml")
+        pathlib.Path(build_context).resolve() if build_context else config_path.parent
     )
     install_cmd = install_command or _get_node_pm_install_cmd(install_root)
     if build_context:

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
 
 
@@ -813,22 +683,14 @@ 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,
         )
@@ -896,18 +758,21 @@ def _create_deployment(
 
 
 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
@@ -925,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()
@@ -1450,7 +1318,6 @@ def _run_remote_build(
 @dataclass(frozen=True, slots=True)
 class DeployContext:
     client: HostBackendClient
-    endpoints: ControlPlaneEndpoints
     spec: BuildSpec
     verbose: bool
     selector: DeploymentSelector
@@ -1480,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(
@@ -1602,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):
@@ -1624,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(
@@ -1638,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(
@@ -1667,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")
@@ -1740,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(
@@ -1870,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"
@@ -1911,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).",
 )
@@ -2075,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",
@@ -2200,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,
@@ -2226,14 +1982,13 @@ def _deploy_cmd(
     validate_deploy_commands(install_command, build_command)
     agent = None
     if agent_id is not None or environment is not None:
-        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."
+                "--agent-id and --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."
+                "--agent-id and --environment cannot be combined with --name or --deployment-id."
             )
         agent = {"agent_id": agent_id, "environment": environment}
     if not config.exists():
@@ -2269,15 +2024,12 @@ 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)
@@ -2290,7 +2042,6 @@ def _deploy_cmd(
     outcome = source.run(
         DeployContext(
             client=client,
-            endpoints=client.endpoints,
             spec=BuildSpec(
                 config=config,
                 config_json=config_json,
@@ -2302,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,
@@ -2311,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:
@@ -2385,11 +2138,6 @@ def deploy_list(
     agent_id: str | None,
     environment: str | None,
 ) -> None:
-    if agent_id is not None or environment is not None:
-        click.secho(
-            "Note: --agent-id and --agent-environment flags are in private beta",
-            fg="yellow",
-        )
     if agent_id is not None and not agent_id.strip():
         raise click.UsageError("--agent-id must not be empty.")
     filters = {}
@@ -2398,10 +2146,16 @@ 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),
     )
+    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 []
+    )
     if not deployments:
         click.echo("No deployments found.")
         return
@@ -2441,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
@@ -2594,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/langgraph_cli/schemas.py

@@ -650,8 +650,7 @@ class Config(TypedDict, total=False):
 
     pip_config_file: str | None
     """Optional. Path to a pip config file (e.g., "/etc/pip.conf" or "pip.ini") for controlling
-    package installation (custom indices, timeouts, etc.). The file is copied into the
-    generated image, so it must not contain credentials or other secrets.
+    package installation (custom indices, credentials, etc.).
 
     Only relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.
     """
@@ -690,9 +689,6 @@ class Config(TypedDict, total=False):
       - "." or "./src" if you have a local Python package
       - str (aka "anthropic") for a PyPI package
       - "git+https://github.com/org/repo.git@main" for a Git-based package
-    Git HTTP URLs must not contain userinfo such as a username or token. For private
-    dependencies, provide short-lived credentials through the build environment's
-    secret-backed Git credential helper.
     Defaults to an empty list, meaning no additional packages installed beyond your base environment.
 
     This field is not supported when `source.kind` is `uv`.

libs/cli/langgraph_cli/uv_lock.py

@@ -880,7 +880,6 @@ def python_config_to_docker_uv_lock(
         _get_node_pm_install_cmd,
         _get_pip_cleanup_lines,
         _image_supports_uv,
-        _validate_git_http_url_userinfo_files,
         docker_tag,
     )
 
@@ -891,20 +890,11 @@ def python_config_to_docker_uv_lock(
         )
 
     config_root = config_path.parent.resolve()
-    source_root = config["source"].get("root", ".")
-    project_root = (config_root / source_root).resolve()
-    _validate_git_http_url_userinfo_files(
-        [project_root / "pyproject.toml", project_root / "uv.lock"]
-    )
-
     install_cmd = "uv pip install --system"
     _, global_reqs_pip_install, pip_config_file_str = _build_python_install_commands(
         config, install_cmd
     )
     plan = _plan_uv_lock_workspace(config_path, config)
-    _validate_git_http_url_userinfo_files(
-        package.pyproject_path for package in plan.install_order
-    )
 
     _update_uv_lock_graph_paths(config_path, config, plan)
     for section, key in [

libs/cli/schemas/schema.json

@@ -28,7 +28,7 @@
                   "type": "null"
                 }
               ],
-              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
+              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
             },
             "_INTERNAL_docker_tag": {
               "anyOf": [
@@ -270,7 +270,7 @@
                   "type": "null"
                 }
               ],
-              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
+              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
             },
             "_INTERNAL_docker_tag": {
               "anyOf": [

libs/cli/schemas/schema.v0.json

@@ -28,7 +28,7 @@
                   "type": "null"
                 }
               ],
-              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
+              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
             },
             "_INTERNAL_docker_tag": {
               "anyOf": [
@@ -270,7 +270,7 @@
                   "type": "null"
                 }
               ],
-              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
+              "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
             },
             "_INTERNAL_docker_tag": {
               "anyOf": [

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_config.py

@@ -255,243 +255,6 @@ def test_validate_config():
         )
 
 
-@pytest.mark.parametrize(
-    "dependency",
-    [
-        "git+https://user:secret-token@github.com/org/private.git@main",
-        "private-package @ git+http://token@github.com/org/private.git",
-        "git+HTTPS://user%40example.com:secret%2Ftoken@github.com/org/private.git",
-        "git+https://${GIT_TOKEN}@github.com/org/private.git",
-    ],
-)
-def test_validate_config_rejects_git_http_url_userinfo(dependency: str):
-    with pytest.raises(click.UsageError) as exc_info:
-        validate_config(
-            {
-                "python_version": "3.11",
-                "dependencies": [dependency],
-                "graphs": {"agent": "./agent.py:graph"},
-            }
-        )
-
-    message = str(exc_info.value)
-    assert "must not contain credentials or other URL userinfo" in message
-    assert "secret-token" not in message
-    assert "secret%2Ftoken" not in message
-
-
-def test_validate_config_file_reports_source_for_git_http_url_userinfo(
-    tmp_path: pathlib.Path,
-):
-    config_path = tmp_path / "langgraph.json"
-    config_path.write_text(
-        json.dumps(
-            {
-                "python_version": "3.11",
-                "dependencies": ["git+https://secret-token@github.com/org/private.git"],
-                "graphs": {"agent": "./agent.py:graph"},
-            }
-        )
-    )
-
-    with pytest.raises(click.UsageError) as exc_info:
-        validate_config_file(config_path)
-
-    message = str(exc_info.value)
-    assert "secret-token" not in message
-    assert f"Found in: {config_path.resolve()}" in message
-
-
-@pytest.mark.parametrize(
-    "manifest", ["package.json", "package-lock.json", "yarn.lock", "pnpm-lock.yaml"]
-)
-def test_config_to_docker_rejects_git_http_url_userinfo_in_node_files(
-    tmp_path: pathlib.Path, manifest: str
-):
-    config_path = tmp_path / "langgraph.json"
-    config_path.write_text("{}\n")
-    (tmp_path / "agent.js").write_text("export const graph = {};\n")
-    (tmp_path / "package.json").write_text('{"name":"agent"}\n')
-    (tmp_path / manifest).write_text(
-        '"priv": "git+https://user:secret-token@github.com/org/private.git"\n'
-    )
-    config = validate_config(
-        {
-            "node_version": "20",
-            "graphs": {"agent": "./agent.js:graph"},
-        }
-    )
-
-    with pytest.raises(click.UsageError) as exc_info:
-        config_to_docker(
-            config_path,
-            config,
-            base_image="langchain/langgraphjs-api",
-        )
-
-    message = str(exc_info.value)
-    assert "must not contain credentials or other URL userinfo" in message
-    assert "secret-token" not in message
-    assert f"Found in: {(tmp_path / manifest).resolve()}" in message
-
-
-def test_config_to_docker_allows_node_git_urls_without_http_userinfo(
-    tmp_path: pathlib.Path,
-):
-    config_path = tmp_path / "langgraph.json"
-    config_path.write_text("{}\n")
-    (tmp_path / "agent.js").write_text("export const graph = {};\n")
-    (tmp_path / "package.json").write_text(
-        '{"dependencies":{"public":"git+https://github.com/org/public.git"}}\n'
-    )
-    config = validate_config(
-        {
-            "node_version": "20",
-            "graphs": {"agent": "./agent.js:graph"},
-        }
-    )
-
-    docker, _ = config_to_docker(
-        config_path,
-        config,
-        base_image="langchain/langgraphjs-api",
-    )
-
-    assert f"ADD . /deps/{tmp_path.name}" in docker
-
-
-def test_config_to_docker_rejects_git_http_url_userinfo_in_node_workspace(
-    tmp_path: pathlib.Path,
-):
-    config_root = tmp_path / "apps" / "agent"
-    config_root.mkdir(parents=True)
-    config_path = config_root / "langgraph.json"
-    config_path.write_text("{}\n")
-    (config_root / "agent.js").write_text("export const graph = {};\n")
-    (config_root / "package.json").write_text(
-        '{"dependencies":{"priv":"git+https://secret-token@github.com/org/private.git"}}\n'
-    )
-    (tmp_path / "package.json").write_text('{"name":"workspace"}\n')
-    config = validate_config(
-        {
-            "node_version": "20",
-            "graphs": {"agent": "./agent.js:graph"},
-        }
-    )
-
-    with pytest.raises(click.UsageError) as exc_info:
-        config_to_docker(
-            config_path,
-            config,
-            base_image="langchain/langgraphjs-api",
-            build_context=str(tmp_path),
-        )
-
-    message = str(exc_info.value)
-    assert "secret-token" not in message
-    assert f"Found in: {(config_root / 'package.json').resolve()}" in message
-
-
-@pytest.mark.parametrize(
-    "dependency",
-    [
-        "git+https://github.com/org/public.git@main",
-        "private-package @ git+https://github.com/org/private.git@main",
-        "git+ssh://git@github.com/org/private.git@main",
-    ],
-)
-def test_validate_config_allows_git_urls_without_http_userinfo(dependency: str):
-    config = validate_config(
-        {
-            "python_version": "3.11",
-            "dependencies": [dependency],
-            "graphs": {"agent": "./agent.py:graph"},
-        }
-    )
-
-    assert config["dependencies"] == [dependency]
-
-
-def test_config_to_docker_rejects_git_http_url_userinfo_in_requirements(
-    tmp_path: pathlib.Path,
-):
-    config_path = tmp_path / "langgraph.json"
-    config_path.write_text("{}\n")
-    (tmp_path / "agent.py").write_text("graph = object()\n")
-    (tmp_path / "requirements.txt").write_text(
-        "private @ git+https://secret-token@github.com/org/private.git\n"
-    )
-    config = validate_config(
-        {
-            "python_version": "3.11",
-            "dependencies": ["."],
-            "graphs": {"agent": "./agent.py:graph"},
-        }
-    )
-
-    with pytest.raises(click.UsageError) as exc_info:
-        config_to_docker(
-            config_path,
-            config,
-            base_image="langchain/langgraph-api:0.2.47",
-        )
-
-    message = str(exc_info.value)
-    assert "must not contain credentials or other URL userinfo" in message
-    assert "secret-token" not in message
-    assert f"Found in: {(tmp_path / 'requirements.txt').resolve()}" in message
-
-
-@pytest.mark.parametrize("manifest", ["pyproject.toml", "uv.lock"])
-def test_config_to_docker_rejects_git_http_url_userinfo_in_uv_files(
-    tmp_path: pathlib.Path, manifest: str
-):
-    config_path = tmp_path / "langgraph.json"
-    config_path.write_text("{}\n")
-    (tmp_path / "src").mkdir()
-    (tmp_path / "src" / "agent.py").write_text("graph = object()\n")
-    pyproject = textwrap.dedent(
-        """
-        [project]
-        name = "agent"
-        version = "0.1.0"
-        dependencies = ["private"]
-
-        [tool.uv.sources]
-        private = { git = "https://github.com/org/private.git" }
-        """
-    ).strip()
-    uv_lock = "# uv lock file\n"
-    if manifest == "pyproject.toml":
-        pyproject = pyproject.replace(
-            "https://github.com", "https://secret-token@github.com"
-        )
-    else:
-        uv_lock += (
-            'source = { git = "https://secret-token@github.com/org/private.git" }\n'
-        )
-    (tmp_path / "pyproject.toml").write_text(pyproject + "\n")
-    (tmp_path / "uv.lock").write_text(uv_lock)
-    config = validate_config(
-        {
-            "python_version": "3.11",
-            "graphs": {"agent": "./src/agent.py:graph"},
-            "source": {"kind": "uv"},
-        }
-    )
-
-    with pytest.raises(click.UsageError) as exc_info:
-        config_to_docker(
-            config_path,
-            config,
-            base_image="langchain/langgraph-api:0.2.47",
-        )
-
-    message = str(exc_info.value)
-    assert "must not contain credentials or other URL userinfo" in message
-    assert "secret-token" not in message
-
-
 def test_validate_config_image_distro():
     """Test validation of image_distro field."""
     # Valid image_distro values should work

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/langgraph/pregel/_checkpoint.py

@@ -47,16 +47,6 @@ def exit_delta_task_id(step: int, task_id: str) -> str:
     return f"{step:08d}-{parts[1]}-{parts[2]}-{parts[3]}-{parts[4]}"
 
 
-def exit_delta_late_task_id(step: int, task_id: str) -> str:
-    """Synthetic task id for exit-mode writes of a superstep after the anchor's own.
-
-    Sorts after every real task id, in step order, so replay keeps them after
-    the anchor's own superstep whether a saver orders by task path or task id.
-    """
-    parts = str(uuid.UUID(task_id)).split("-")
-    return f"ffffffff-{step >> 16:04x}-{step & 0xFFFF:04x}-{parts[3]}-{parts[4]}"
-
-
 def delta_channels_to_snapshot(
     channels: Mapping[str, BaseChannel],
     counters_since_delta_snapshot: Mapping[str, tuple[int, int]],

libs/langgraph/langgraph/pregel/_loop.py

@@ -103,7 +103,6 @@ from langgraph.pregel._checkpoint import (
     create_checkpoint,
     delta_channels_to_snapshot,
     empty_checkpoint,
-    exit_delta_late_task_id,
     exit_delta_task_id,
 )
 from langgraph.pregel._executor import (
@@ -218,15 +217,10 @@ class PregelLoop:
     # `after_tick`). At exit, `_put_exit_delta_writes` filters out channels
     # that will snapshot, then persists the rest under an anchor parent.
     # `None` when not in exit mode (so the capture sites are no-ops).
-    # Each tuple is `(step, task_id, task_path, channel, value)`; see
-    # `_put_exit_delta_writes` for how they are ordered.
-    _exit_delta_writes: list[tuple[int, str, str, str, Any]] | None = None
-
-    # The pending writes loaded with the checkpoint, already stored on it, kept
-    # alive so their ids stay unique; and the checkpoint's own superstep, the
-    # first one this run ticks.
-    _loaded_write_ids: dict[int, tuple[str, str, Any]]
-    _exit_first_step: int | None = None
+    # Each tuple is `(step, task_id, channel, value)` — `step` drives the
+    # synthetic step-prefixed task_id used to preserve chronological order
+    # under the saver's `ORDER BY task_id, idx` sorting.
+    _exit_delta_writes: list[tuple[int, str, str, Any]] | None = None
 
     # Delta channels that saw an Overwrite since the last checkpoint. These
     # channels must snapshot after live update applies overwrite semantics so
@@ -713,18 +707,9 @@ class PregelLoop:
             )
         # capture delta-channel writes for exit-mode accumulator before clearing
         if self._exit_delta_writes is not None:
-            if self._exit_first_step is None:
-                self._exit_first_step = self.step
-            for w in self.checkpoint_pending_writes:
-                tid, ch, v = w
-                if not isinstance(self.specs.get(ch), DeltaChannel):
-                    continue
-                if id(w) in self._loaded_write_ids:
-                    continue
-                task = self.tasks.get(tid)
-                path = task_path_str(task.path) if task else ""
-                self._exit_delta_writes.append((self.step, tid, path, ch, v))
-        self._loaded_write_ids = {}
+            for tid, ch, v in self.checkpoint_pending_writes:
+                if isinstance(self.specs.get(ch), DeltaChannel):
+                    self._exit_delta_writes.append((self.step, tid, ch, v))
         # clear pending writes
         self.checkpoint_pending_writes.clear()
         # only replay (re-execute) done tasks on the first tick
@@ -863,7 +848,6 @@ class PregelLoop:
     def _first(
         self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
     ) -> set[str] | None:
-        self._loaded_write_ids = {id(w): w for w in self.checkpoint_pending_writes}
         # Resuming from a previous checkpoint requires two things:
         # 1. A prior checkpoint exists (channel_versions is non-empty)
         # 2. The input signals continuation (not a fresh run with new input)
@@ -1033,9 +1017,7 @@ class PregelLoop:
             if self._exit_delta_writes is not None:
                 for c, v in input_writes:
                     if isinstance(self.specs.get(c), DeltaChannel):
-                        self._exit_delta_writes.append(
-                            (self.step, NULL_TASK_ID, "", c, v)
-                        )
+                        self._exit_delta_writes.append((self.step, NULL_TASK_ID, c, v))
             # Persist delta-channel input writes so sub-freq inputs are
             # recoverable via ancestor walk (mirrors the Command input path).
             if self.durability != "exit":
@@ -1261,7 +1243,9 @@ class PregelLoop:
         )
 
         pending = [
-            w for w in self._exit_delta_writes if w[3] not in channels_to_snapshot
+            (step, tid, ch, v)
+            for (step, tid, ch, v) in self._exit_delta_writes
+            if ch not in channels_to_snapshot
         ]
         if not pending:
             return
@@ -1296,19 +1280,11 @@ class PregelLoop:
             # sees the stub as its parent.
             self.checkpoint_config = anchor_config
 
-        # The checkpoint's own superstep is stored as sync durability stores
-        # it, so it interleaves with the writes a resume loaded from it. Later
-        # supersteps sort after every real task path and task id, in step
-        # order, so this holds whether a saver orders by path or by id.
-        grouped: dict[tuple[str, str], list[tuple[str, Any]]] = {}
-        for step, tid, path, ch, v in pending:
-            if tid == NULL_TASK_ID:
-                key = (exit_delta_task_id(step, tid), "")
-            elif step == self._exit_first_step:
-                key = (tid, path)
-            else:
-                key = (exit_delta_late_task_id(step, tid), f"~~{step:010d}{path}")
-            grouped.setdefault(key, []).append((ch, v))
+        # Step-prefixed synthetic task_id preserves chronological superstep
+        # order under the saver's ORDER BY task_id, idx sorting.
+        grouped: dict[tuple[int, str], list[tuple[str, Any]]] = {}
+        for step, tid, ch, v in pending:
+            grouped.setdefault((step, tid), []).append((ch, v))
         anchor_write_config = patch_configurable(
             anchor_config,
             {
@@ -1318,21 +1294,22 @@ class PregelLoop:
                 CONFIG_KEY_CHECKPOINT_ID: anchor_config[CONF][CONFIG_KEY_CHECKPOINT_ID],
             },
         )
-        for (tid, path), entries in grouped.items():
+        for (step, tid), entries in grouped.items():
+            synth_tid = exit_delta_task_id(step, tid)
             if self.checkpointer_put_writes_accepts_task_path:
                 fut = self.submit(
                     self.checkpointer_put_writes,
                     anchor_write_config,
                     entries,
-                    tid,
-                    path,
+                    synth_tid,
+                    "",
                 )
             else:
                 fut = self.submit(
                     self.checkpointer_put_writes,
                     anchor_write_config,
                     entries,
-                    tid,
+                    synth_tid,
                 )
             if self._delta_write_futs is not None:
                 self._delta_write_futs.append(fut)

libs/langgraph/tests/test_delta_channel_exit_mode.py

@@ -6,13 +6,11 @@ channel), lazy stub creation when no parent exists, and proper read-path
 reconstruction via ancestor walks.
 """
 
-import operator
 import uuid
 from typing import Annotated, Any
 
 import pytest
 from langchain_core.messages import AIMessage, HumanMessage
-from langgraph.checkpoint.base import BaseCheckpointSaver
 from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import TypedDict
@@ -21,7 +19,6 @@ from langgraph.channels.delta import DeltaChannel
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
 from langgraph.pregel._checkpoint import exit_delta_task_id
-from langgraph.types import Command, Durability, interrupt
 
 pytestmark = pytest.mark.anyio
 
@@ -392,112 +389,3 @@ async def test_exit_snapshot_then_tail_deltas() -> None:
     assert "seed-msg" in contents
     assert "tail-msg" in contents
     assert contents.index("seed-msg") < contents.index("tail-msg")
-
-
-def _append(current: list, writes: list) -> list:
-    out = list(current)
-    for write in writes:
-        out.extend(write)
-    return out
-
-
-class _ResumeState(TypedDict):
-    log: Annotated[list, DeltaChannel(_append)]
-    plain: Annotated[list, operator.add]
-
-
-def _both(marker: str) -> dict:
-    return {"log": [marker], "plain": [marker]}
-
-
-def _ask(marker: str) -> Any:
-    def ask(state: _ResumeState) -> dict:
-        interrupt("approve?")
-        return _both(marker)
-
-    return ask
-
-
-@pytest.mark.parametrize("addressed", [False, True])
-def test_resume_after_a_parallel_interrupt_replays_in_live_order(
-    sync_checkpointer: BaseCheckpointSaver, durability: Durability, addressed: bool
-) -> None:
-    builder = StateGraph(_ResumeState)
-    builder.add_node("done", lambda state: _both("done"))
-    builder.add_node("ask", _ask("ask"))
-    builder.add_node("after", lambda state: _both("after"))
-    builder.add_edge(START, "done")
-    builder.add_edge(START, "ask")
-    builder.add_edge("ask", "after")
-    graph = builder.compile(checkpointer=sync_checkpointer)
-    config = {"configurable": {"thread_id": "t"}}
-    graph.invoke(_both("in"), config, durability=durability)
-    head = graph.get_state(config).config
-
-    graph.invoke(
-        Command(resume="yes"), head if addressed else config, durability=durability
-    )
-
-    state = graph.get_state(config)
-    assert state.values["log"] == state.values["plain"]
-    assert sorted(state.values["log"]) == ["after", "ask", "done", "in"]
-
-
-def test_resume_interleaves_the_resumed_superstep_by_task_path(
-    sync_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    builder = StateGraph(_ResumeState)
-    builder.add_node("z_done", lambda state: _both("z"))
-    builder.add_node("a_asks", _ask("a"))
-    builder.add_edge(START, "z_done")
-    builder.add_edge(START, "a_asks")
-    graph = builder.compile(checkpointer=sync_checkpointer)
-    config = {"configurable": {"thread_id": "t"}}
-    graph.invoke(_both("in"), config, durability=durability)
-
-    graph.invoke(Command(resume="yes"), config, durability=durability)
-
-    state = graph.get_state(config)
-    assert state.values["log"] == state.values["plain"] == ["in", "a", "z"]
-
-
-class _TaskIdOrderSaver(InMemorySaver):
-    """Replays each checkpoint's writes by task id, as savers without task path
-    ordering do."""
-
-    def get_tuple(self, config: Any) -> Any:
-        tup = super().get_tuple(config)
-        if tup and tup.pending_writes:
-            tup = tup._replace(pending_writes=sorted(tup.pending_writes))
-        return tup
-
-    get_delta_channel_history = BaseCheckpointSaver.get_delta_channel_history
-
-
-def test_exit_run_replays_supersteps_in_order_on_a_task_id_ordered_saver() -> None:
-    builder = StateGraph(_ResumeState)
-    builder.add_node("a", lambda state: _both("a"))
-    builder.add_node("b", lambda state: _both("b"))
-    builder.add_edge(START, "a")
-    builder.add_edge("a", "b")
-    graph = builder.compile(checkpointer=_TaskIdOrderSaver())
-    config = {"configurable": {"thread_id": "t"}}
-
-    graph.invoke(_both("in"), config, durability="exit")
-
-    assert graph.get_state(config).values["log"] == ["in", "a", "b"]
-
-
-def test_exit_resume_replays_supersteps_in_order_on_a_task_id_ordered_saver() -> None:
-    builder = StateGraph(_ResumeState)
-    builder.add_node("ask", _ask("ask"))
-    builder.add_node("after", lambda state: _both("after"))
-    builder.add_edge(START, "ask")
-    builder.add_edge("ask", "after")
-    graph = builder.compile(checkpointer=_TaskIdOrderSaver())
-    config = {"configurable": {"thread_id": "t"}}
-    graph.invoke(_both("in"), config, durability="exit")
-
-    graph.invoke(Command(resume="yes"), config, durability="exit")
-
-    assert graph.get_state(config).values["log"] == ["in", "ask", "after"]

libs/langgraph/tests/test_delta_channel_parallel_order.py

@@ -1,113 +0,0 @@
-"""`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
-from langgraph.types import Send
-
-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"]
-SEND_ARGS = [f"send-{i:02d}" for i in range(12)]
-
-
-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)
-
-
-def _build_send_fan_out_graph(checkpointer: BaseCheckpointSaver) -> Any:
-    class State(TypedDict):
-        items: Annotated[
-            list, DeltaChannel(_append_reducer, list, snapshot_frequency=10_000)
-        ]
-
-    builder = StateGraph(State)
-    builder.add_node("worker", lambda arg: {"items": [arg]})
-    builder.add_conditional_edges(
-        START, lambda state: [Send("worker", n) for n in SEND_ARGS]
-    )
-    builder.add_edge("worker", END)
-    return builder.compile(checkpointer=checkpointer)
-
-
-async def test_get_state_matches_live_send_order(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    graph = _build_send_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 == SEND_ARGS
-    assert replayed == live
-
-
-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}"