feat(langgraph): append-only TranscriptChannel; edits and removals fork

Bumps the Python SDK version from 0.4.11 to 0.4.12.

Includes:

- Remove `fairness_weight` from `PrioritySettings` and document that -5
runs first (#9441)
- Rejoin a run stream that closes before the run ends (#9554)

libs/cli/langgraph_cli/config.py

@@ -1400,7 +1400,6 @@ def python_config_to_docker(
     api_version: str | None = None,
     *,
     escape_variables: bool = False,
-    single_root_copy: bool = False,
 ) -> tuple[str, dict[str, str]]:
     """Generate a Dockerfile from the configuration."""
     source_kind = _get_source_kind(config)
@@ -1413,7 +1412,6 @@ def python_config_to_docker(
             base_image,
             api_version=api_version,
             build_tools_to_uninstall=build_tools_to_uninstall,
-            single_root_copy=single_root_copy,
         )
     _validate_local_dependency_files(config_path, config)
     if pip_installer == "auto":
@@ -1774,7 +1772,6 @@ def config_to_docker(
     build_command: str | None = None,
     build_context: str | None = None,
     escape_variables: bool = False,
-    single_root_copy: bool = False,
 ) -> tuple[str, dict[str, str]]:
     base_image = base_image or default_base_image(config)
 
@@ -1795,7 +1792,6 @@ def config_to_docker(
         base_image=base_image,
         api_version=api_version,
         escape_variables=escape_variables,
-        single_root_copy=single_root_copy,
     )
 
 

libs/cli/langgraph_cli/docker.py

@@ -387,8 +387,6 @@ def build_docker_image(
         install_command=install_command,
         build_command=build_command,
         build_context=build_context,
-        # Safe because this Dockerfile is built immediately, not saved for later.
-        single_root_copy=True,
     )
     # add additional_contexts
     if additional_contexts:

libs/cli/langgraph_cli/uv_lock.py

@@ -2,7 +2,6 @@ import os
 import pathlib
 import re
 import shlex
-import tarfile
 from dataclasses import dataclass
 
 try:
@@ -14,7 +13,6 @@ import click
 import pathspec
 
 from langgraph_cli._ignore import (
-    _ALWAYS_EXCLUDE_NAMES,
     _build_dockerignore_negation_hints,
     _build_ignore_spec,
     _is_always_excluded,
@@ -519,56 +517,6 @@ def _uv_lock_package_copy_items(
     return iter_entries(plan.project_root)
 
 
-_COMPRESSION_MAGIC = (b"\x1f\x8b", b"BZh", b"\xfd7zXZ\x00", b"\x28\xb5\x2f\xfd")
-
-
-def _is_local_archive(path: pathlib.Path) -> bool:
-    """Whether Docker's `ADD` would auto-extract this local file."""
-    with path.open("rb") as f:
-        # Match gzip/bzip2/xz/zstd by magic, since tarfile skips codecs Python lacks.
-        if f.read(6).startswith(_COMPRESSION_MAGIC):
-            return True
-        f.seek(0)
-        return tarfile.is_tarfile(f)
-
-
-def _root_copy_is_equivalent(
-    plan: UvLockPlan,
-    config_root: pathlib.Path,
-    config: Config,
-    copy_items: tuple[tuple[pathlib.PurePosixPath, pathlib.PurePosixPath], ...],
-) -> bool:
-    """Whether one `ADD .` produces the same files as the per-entry ADDs."""
-    root = plan.project_root
-    # Only the build context's .dockerignore applies, and nested members need the walk.
-    if root != config_root or any(root in m.parents for m in plan.all_workspace_roots):
-        return False
-
-    ignore_file = root / ".dockerignore"
-    text = ignore_file.read_text(encoding="utf-8") if ignore_file.is_file() else ""
-    patterns = {s for line in text.splitlines() if (s := line.strip()) and s[0] != "#"}
-    # Negations need the walk; Docker reads `[...]` and inner `**` differently.
-    if any(
-        p.startswith("!") or "[" in p or "**" in p.removeprefix("**/") for p in patterns
-    ):
-        return False
-    literals = {p.lstrip("/").removeprefix("./").rstrip("/") for p in patterns}
-    # Docker doesn't apply the CLI's built-in exclusions, so .dockerignore must.
-    if any(
-        c.name in _ALWAYS_EXCLUDE_NAMES
-        and c.is_dir()
-        and not {c.name, f"**/{c.name}"} & literals
-        for c in root.iterdir()
-    ):
-        return False
-    # Per-entry ADD follows symlinks and extracts archives; `ADD .` does neither.
-    paths = [root / source for source, _ in copy_items]
-    if any(p.is_symlink() or (p.is_file() and _is_local_archive(p)) for p in paths):
-        return False
-    # Custom lines that touch /deps could make the destination merge differ.
-    return not any("/deps" in line for line in config["dockerfile_lines"])
-
-
 def _resolve_uv_lock_container_path(
     host_path: pathlib.Path, plan: UvLockPlan
 ) -> pathlib.PurePosixPath | None:
@@ -924,7 +872,6 @@ def python_config_to_docker_uv_lock(
     api_version: str | None = None,
     *,
     build_tools_to_uninstall: tuple[str, ...] | None,
-    single_root_copy: bool = False,
 ) -> tuple[str, dict[str, str]]:
     from langgraph_cli.config import (
         DEFAULT_NODE_VERSION,
@@ -1076,17 +1023,9 @@ def python_config_to_docker_uv_lock(
     for package in plan.install_order:
         package_label = package.root.relative_to(plan.project_root).as_posix() or "."
         docker_plan.add_raw(f"# -- Adding workspace package {package_label} --")
-        copy_items = _uv_lock_package_copy_items(package, plan, ignore_spec)
-        # Only for immediate builds: `ADD .` copies the tree as it is at build time.
-        if (
-            single_root_copy
-            and package.root == plan.project_root
-            and _root_copy_is_equivalent(plan, config_root, config, copy_items)
+        for source, destination in _uv_lock_package_copy_items(
+            package, plan, ignore_spec
         ):
-            copy_items = (
-                (pathlib.PurePosixPath("."), plan.container_roots[package.root]),
-            )
-        for source, destination in copy_items:
             docker_plan.add_raw(copy_from_project_root(source, destination.as_posix()))
         docker_plan.add_instruction(
             "WORKDIR", plan.container_roots[package.root].as_posix()

libs/cli/tests/unit_tests/test_config.py

@@ -2,16 +2,13 @@ import copy
 import json
 import os
 import pathlib
-import tarfile
 import tempfile
 import textwrap
 from unittest.mock import patch
 
 import click
 import pytest
-from click.testing import CliRunner
 
-from langgraph_cli.cli import cli
 from langgraph_cli.config import (
     _BUILD_TOOLS,
     _get_pip_cleanup_lines,
@@ -2363,174 +2360,6 @@ def test_config_to_docker_uv_lock_does_not_apply_gitignore():
         assert "ADD README.md /deps/workspace/README.md" in docker
 
 
-def _write_single_uv_project(project_root: pathlib.Path) -> None:
-    project_root.mkdir(parents=True)
-    (project_root / "uv.lock").write_text("# uv lock file\n")
-    (project_root / "pyproject.toml").write_text(
-        '[project]\nname = "single-app"\nversion = "0.1.0"\n'
-        'dependencies = ["httpx>=0.28"]\n'
-    )
-    (project_root / "langgraph.json").write_text("{}\n")
-    (project_root / "src").mkdir()
-    (project_root / "src" / "agent.py").write_text("graph = object()\n")
-    (project_root / "README.md").write_text("# hi\n")
-
-
-def _single_uv_docker(
-    config_path: pathlib.Path, single_root_copy: bool = True, **overrides
-) -> str:
-    config = validate_config(
-        {
-            "python_version": "3.11",
-            "graphs": {"agent": "./src/agent.py:graph"},
-            "source": {"kind": "uv"},
-            **overrides,
-        }
-    )
-    docker, _ = config_to_docker(
-        config_path,
-        config,
-        base_image="langchain/langgraph-api:0.2.47",
-        single_root_copy=single_root_copy,
-    )
-    return docker
-
-
-def test_config_to_docker_uv_lock_copies_safe_root_in_one_layer():
-    with tempfile.TemporaryDirectory() as tmpdir:
-        project_root = pathlib.Path(tmpdir) / "single"
-        _write_single_uv_project(project_root)
-        (project_root / "docs").mkdir()
-        (project_root / "docs" / "notes.md").write_text("notes\n")
-        # Built-in exclusions are fine when Docker excludes them too.
-        (project_root / ".venv").mkdir()
-        (project_root / "__pycache__").mkdir()
-        (project_root / ".dockerignore").write_text("/.venv/\n**/__pycache__\n")
-
-        docker = _single_uv_docker(project_root / "langgraph.json")
-
-        source_copies = [
-            line
-            for line in docker.splitlines()
-            if line.startswith("ADD ") and "/deps/workspace" in line
-        ]
-        assert source_copies == ["ADD . /deps/workspace"]
-        # Dependency metadata is still copied separately ahead of uv export.
-        assert "ADD pyproject.toml /tmp/uv_export/project/pyproject.toml" in docker
-        assert "ADD uv.lock /tmp/uv_export/project/uv.lock" in docker
-
-        # A saved Dockerfile may be built after the tree changes, so keep the list.
-        saved = _single_uv_docker(
-            project_root / "langgraph.json", single_root_copy=False
-        )
-        assert "ADD . /deps/workspace" not in saved
-        assert "ADD src /deps/workspace/src" in saved
-
-
-def _write_tar(root: pathlib.Path) -> None:
-    # No extension, so detection must go by content.
-    with tarfile.open(root / "bundle.bin", "w") as tar:
-        tar.add(root / "README.md", arcname="README.md")
-
-
-def _add_unrelated_member(root: pathlib.Path) -> None:
-    with (root / "pyproject.toml").open("a") as f:
-        f.write('\n[tool.uv.workspace]\nmembers = ["libs/*"]\n')
-    (root / "libs" / "extra").mkdir(parents=True)
-    (root / "libs" / "extra" / "pyproject.toml").write_text(
-        '[project]\nname = "extra"\nversion = "0.1.0"\n'
-    )
-
-
-def _venv_ignored_by(patterns: str):
-    def setup(root: pathlib.Path) -> None:
-        (root / ".venv").mkdir()
-        (root / ".dockerignore").write_text(patterns)
-
-    return setup
-
-
-@pytest.mark.parametrize(
-    ("setup", "overrides"),
-    [
-        pytest.param(lambda r: (r / ".venv").mkdir(), {}, id="unexcluded_venv"),
-        pytest.param(
-            _venv_ignored_by(".venv-old\nsrc/.venv\n"), {}, id="near_miss_literal"
-        ),
-        pytest.param(_venv_ignored_by(".venv\n[!t]*.log\n"), {}, id="char_class"),
-        pytest.param(_venv_ignored_by(".venv\ndata/**\n"), {}, id="inner_globstar"),
-        pytest.param(_add_unrelated_member, {}, id="workspace_member"),
-        pytest.param(
-            lambda r: (r / ".dockerignore").write_text("*.log\n!keep.log\n"),
-            {},
-            id="negation",
-        ),
-        pytest.param(lambda r: (r / "linked").symlink_to("src"), {}, id="symlink"),
-        pytest.param(_write_tar, {}, id="tar"),
-        pytest.param(
-            lambda r: (r / "data.zst").write_bytes(b"\x28\xb5\x2f\xfd" + bytes(16)),
-            {},
-            id="zstd",
-        ),
-        pytest.param(
-            lambda r: None,
-            {"dockerfile_lines": ["RUN mkdir -p /deps/workspace/cache"]},
-            id="deps_dockerfile_line",
-        ),
-    ],
-)
-def test_config_to_docker_uv_lock_keeps_per_entry_copies_when_unsafe(setup, overrides):
-    with tempfile.TemporaryDirectory() as tmpdir:
-        project_root = pathlib.Path(tmpdir) / "single"
-        _write_single_uv_project(project_root)
-        setup(project_root)
-
-        docker = _single_uv_docker(project_root / "langgraph.json", **overrides)
-
-        assert "ADD . /deps/workspace" not in docker
-        assert "ADD src /deps/workspace/src" in docker
-        assert "ADD README.md /deps/workspace/README.md" in docker
-
-
-def test_saved_and_compose_dockerfiles_keep_per_entry_copies():
-    with tempfile.TemporaryDirectory() as tmpdir:
-        project_root = pathlib.Path(tmpdir) / "single"
-        _write_single_uv_project(project_root)
-        config_path = project_root / "langgraph.json"
-        raw = {"graphs": {"agent": "./src/agent.py:graph"}, "source": {"kind": "uv"}}
-        config_path.write_text(json.dumps({"python_version": "3.11", **raw}))
-
-        compose = config_to_compose(
-            config_path, validate_config_file(config_path), "langchain/langgraph-api"
-        )
-        saved = project_root / "Dockerfile"
-        result = CliRunner().invoke(
-            cli, ["dockerfile", str(saved), "--config", str(config_path)]
-        )
-
-        assert result.exit_code == 0, result.output
-        for dockerfile in (compose, saved.read_text()):
-            assert "ADD . /deps/workspace" not in dockerfile
-            assert "ADD src /deps/workspace/src" in dockerfile
-
-
-def test_config_to_docker_uv_lock_keeps_per_entry_copies_outside_build_context():
-    with tempfile.TemporaryDirectory() as tmpdir:
-        project_root = pathlib.Path(tmpdir) / "single"
-        _write_single_uv_project(project_root)
-        (project_root / "deploy").mkdir()
-        (project_root / "deploy" / "langgraph.json").write_text("{}\n")
-
-        docker = _single_uv_docker(
-            project_root / "deploy" / "langgraph.json",
-            graphs={"agent": "../src/agent.py:graph"},
-            source={"kind": "uv", "root": ".."},
-        )
-
-        assert "COPY --from=uv-workspace-root . /deps/workspace" not in docker
-        assert "COPY --from=uv-workspace-root src /deps/workspace/src" in docker
-
-
 def test_config_to_docker_uv_lock_skips_dockerignore_entries_in_workspace():
     """Multi-member workspace: ignore patterns must filter root-level entries
     AND entries encountered while recursing into directories that contain

libs/cli/tests/unit_tests/test_docker.py

@@ -1,6 +1,3 @@
-import pathlib
-from unittest.mock import MagicMock, patch
-
 import pytest
 
 from langgraph_cli.docker import (
@@ -8,7 +5,6 @@ from langgraph_cli.docker import (
     DockerCapabilities,
     Version,
     _parse_version,
-    build_docker_image,
     compose,
 )
 from langgraph_cli.util import clean_empty_lines
@@ -444,23 +440,3 @@ def test_compose_combined_mode_has_no_n_jobs():
 )
 def test_parse_version_w_edge_cases(input_str, expected):
     assert _parse_version(input_str) == expected
-
-
-def test_build_docker_image_opts_into_single_root_copy():
-    with (
-        patch(
-            "langgraph_cli.config.config_to_docker", return_value=("", {})
-        ) as to_docker,
-        patch("langgraph_cli.docker.subp_exec"),
-    ):
-        build_docker_image(
-            MagicMock(),
-            lambda _: None,
-            pathlib.Path("/tmp/langgraph.json"),
-            {"python_version": "3.11"},
-            base_image=None,
-            api_version=None,
-            pull=False,
-            tag="probe",
-        )
-    assert to_docker.call_args.kwargs["single_root_copy"] is True

libs/langgraph/langgraph/channels/transcript.py

@@ -0,0 +1,300 @@
+from __future__ import annotations
+
+import uuid
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from typing_extensions import Self
+
+from langgraph._internal._typing import MISSING
+from langgraph.channels.base import BaseChannel
+from langgraph.channels.binop import _get_overwrite
+from langgraph.errors import InvalidUpdateError
+
+__all__ = ("TranscriptChannel", "TranscriptEntry", "TranscriptRef")
+
+EntryKind = Literal["append", "reset"]
+
+
+@dataclass(frozen=True, slots=True)
+class TranscriptEntry:
+    """One log entry. `append` carries a message, `reset` the full list."""
+
+    seq: int
+    kind: EntryKind
+    item_id: str = ""
+    value: Any = None
+
+
+@dataclass(slots=True)
+class TranscriptRef:
+    """Checkpoint value of a `TranscriptChannel`: a pointer into a log.
+
+    Written: `entries` holds entries with seq in (base_head, head].
+    Read: `entries` holds every entry in [1, head], fork ancestors included.
+    """
+
+    transcript_id: str
+    head: int
+    base_head: int = 0
+    entries: list[TranscriptEntry] = field(default_factory=list)
+    parent_transcript_id: str = ""
+    fork_seq: int = 0
+    is_tip: bool = True
+
+
+def _coerce(update: Any) -> list[Any]:
+    from langchain_core.messages import (
+        BaseMessageChunk,
+        convert_to_messages,
+        message_chunk_to_message,
+    )
+
+    items = update if isinstance(update, list) else [update]
+    out = []
+    for m in convert_to_messages(items):
+        m = message_chunk_to_message(m) if isinstance(m, BaseMessageChunk) else m
+        if m.id is None:
+            m.id = str(uuid.uuid4())
+        out.append(m)
+    return out
+
+
+def _is_remove(m: Any) -> bool:
+    from langchain_core.messages import RemoveMessage
+
+    return isinstance(m, RemoveMessage)
+
+
+def _remove_all_id() -> str:
+    from langgraph.graph.message import REMOVE_ALL_MESSAGES
+
+    return REMOVE_ALL_MESSAGES
+
+
+class TranscriptChannel(BaseChannel[list, Any, TranscriptRef]):
+    """Message list persisted as an append-only log.
+
+    The checkpoint stores `(transcript_id, head)`. A checkpointer that
+    supports transcripts persists new entries atomically with the checkpoint
+    and returns them on read, so storage grows linearly and loading any
+    checkpoint costs O(len(transcript)).
+
+    Writes contain only new or changed messages:
+
+    * a new id appends;
+    * an existing id (edit) or a `RemoveMessage` forks: a new transcript
+      shares the prefix before the edited message and replays what follows;
+    * `REMOVE_ALL_MESSAGES` or `Overwrite` writes one `reset` entry.
+
+    Loading a checkpoint that is not its transcript's tip also forks.
+    """
+
+    __slots__ = (
+        "value",
+        "pos",
+        "anchor",
+        "resets",
+        "transcript_id",
+        "head",
+        "base_head",
+        "pending",
+        "parent_transcript_id",
+        "fork_seq",
+    )
+
+    def __init__(self, typ: Any = list) -> None:
+        super().__init__(typ)
+        self.value: list[Any] = []
+        # id -> index in value
+        self.pos: dict[str, int] = {}
+        # id -> seq of the entry that introduced it (an append or a reset)
+        self.anchor: dict[str, int] = {}
+        # seqs of reset entries
+        self.resets: set[int] = set()
+        self.transcript_id = str(uuid.uuid4())
+        self.head = 0
+        # Highest seq known committed for this channel's lineage.
+        self.base_head = 0
+        self.pending: list[TranscriptEntry] = []
+        # Set only for a fork this channel created and has not seen committed.
+        self.parent_transcript_id = ""
+        self.fork_seq = 0
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, TranscriptChannel)
+
+    @property
+    def ValueType(self) -> Any:
+        return list
+
+    @property
+    def UpdateType(self) -> Any:
+        return Any
+
+    def _new(self) -> Self:
+        new = self.__class__(self.typ)
+        new.key = self.key
+        return new
+
+    def copy(self) -> Self:
+        new = self._new()
+        new.value = list(self.value)
+        new.pos = dict(self.pos)
+        new.anchor = dict(self.anchor)
+        new.resets = set(self.resets)
+        new.transcript_id = self.transcript_id
+        new.head = self.head
+        new.base_head = self.base_head
+        new.pending = list(self.pending)
+        new.parent_transcript_id = self.parent_transcript_id
+        new.fork_seq = self.fork_seq
+        return new
+
+    def from_checkpoint(self, checkpoint: Any) -> Self:
+        new = self._new()
+        if checkpoint is MISSING:
+            return new
+        if not isinstance(checkpoint, TranscriptRef):
+            # Plain list from a regular channel: start a transcript from it.
+            new._reset(_coerce(list(checkpoint)))
+            return new
+        for e in checkpoint.entries:
+            if e.kind == "reset":
+                new._load_reset(e.seq, e.value)
+            else:
+                new._load_append(e.seq, e.value)
+        new.head = new.base_head = checkpoint.head
+        if checkpoint.is_tip:
+            new.transcript_id = checkpoint.transcript_id
+        else:
+            new.parent_transcript_id = checkpoint.transcript_id
+            new.fork_seq = checkpoint.head
+        return new
+
+    def get(self) -> list[Any]:
+        return self.value
+
+    def is_available(self) -> bool:
+        return True
+
+    def update(self, values: Sequence[Any]) -> bool:
+        if not values:
+            return False
+        for v in values:
+            is_overwrite, overwrite = _get_overwrite(v)
+            if is_overwrite:
+                self._reset(_coerce(list(overwrite)))
+                continue
+            msgs = _coerce(v)
+            remove_all = [
+                i
+                for i, m in enumerate(msgs)
+                if _is_remove(m) and m.id == _remove_all_id()
+            ]
+            if remove_all:
+                self._reset(msgs[remove_all[-1] + 1 :])
+                continue
+            for m in msgs:
+                self._write(m)
+        return True
+
+    def _write(self, m: Any) -> None:
+        i = self.pos.get(m.id)
+        if i is None:
+            if _is_remove(m):
+                raise InvalidUpdateError(
+                    f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
+                )
+            self._append(m)
+            return
+        current = self.value[i]
+        if current is m or (not _is_remove(m) and current == m):
+            # Resend of a message already in the transcript.
+            return
+        replacement = [] if _is_remove(m) else [m]
+        self._fork_at(m.id, replacement)
+
+    def _fork_at(self, item_id: str, replacement: list[Any]) -> None:
+        """Fork before `item_id`, swap it for `replacement`, replay the rest."""
+        # The fork point must already be committed, so it is at most base_head.
+        fork = min(self.anchor[item_id] - 1, self.base_head)
+        prefix = [m for m in self.value if self.anchor[m.id] <= fork]
+        # Entries in (fork, head] that brought in the tail, in order. After a
+        # reset no older item survives, so a reset group is the whole list.
+        groups: list[tuple[int, list[Any]]] = []
+        for m in self.value:
+            a = self.anchor[m.id]
+            if a <= fork:
+                continue
+            if groups and groups[-1][0] == a:
+                groups[-1][1].append(m)
+            else:
+                groups.append((a, [m]))
+
+        # Items surviving the latest reset after the fork point all anchor at
+        # or after it, so replay starts with that reset, even when empty.
+        latest_reset = max((r for r in self.resets if r > fork), default=None)
+        if latest_reset is not None and (not groups or groups[0][0] != latest_reset):
+            groups.insert(0, (latest_reset, []))
+        old_resets = self.resets
+
+        if self.parent_transcript_id:
+            # Fork created this run: base_head == fork_seq, so branch off its parent.
+            parent = self.parent_transcript_id
+        elif fork == 0:
+            parent = ""
+        else:
+            parent = self.transcript_id
+        resets = {r for r in self.resets if r <= fork}
+        anchors = {m.id: self.anchor[m.id] for m in prefix}
+
+        self.transcript_id = str(uuid.uuid4())
+        self.parent_transcript_id = parent
+        self.fork_seq = self.head = self.base_head = fork
+        self.pending = []
+        self.value = prefix
+        self.pos = {m.id: i for i, m in enumerate(prefix)}
+        self.anchor = anchors
+        self.resets = resets
+        for a, group in groups:
+            swapped = [
+                n for m in group for n in (replacement if m.id == item_id else [m])
+            ]
+            if a in old_resets:
+                self._reset(swapped)
+            else:
+                for m in swapped:
+                    self._append(m)
+
+    def _append(self, m: Any) -> None:
+        self.head += 1
+        self.pending.append(TranscriptEntry(self.head, "append", m.id, m))
+        self._load_append(self.head, m)
+
+    def _reset(self, msgs: list[Any]) -> None:
+        self.head += 1
+        self.pending.append(TranscriptEntry(self.head, "reset", "", list(msgs)))
+        self._load_reset(self.head, msgs)
+
+    def _load_append(self, seq: int, m: Any) -> None:
+        self.pos[m.id] = len(self.value)
+        self.anchor[m.id] = seq
+        self.value.append(m)
+
+    def _load_reset(self, seq: int, msgs: list[Any]) -> None:
+        self.value = list(msgs)
+        self.pos = {m.id: i for i, m in enumerate(self.value)}
+        self.anchor = dict.fromkeys(self.pos, seq)
+        self.resets.add(seq)
+
+    def checkpoint(self) -> TranscriptRef:
+        return TranscriptRef(
+            transcript_id=self.transcript_id,
+            head=self.head,
+            base_head=self.base_head,
+            entries=list(self.pending),
+            parent_transcript_id=self.parent_transcript_id,
+            fork_seq=self.fork_seq,
+        )

libs/langgraph/tests/test_channels_transcript.py

@@ -0,0 +1,119 @@
+from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+
+from langgraph._internal._typing import MISSING
+from langgraph.channels.transcript import TranscriptChannel, TranscriptRef
+from langgraph.graph.message import REMOVE_ALL_MESSAGES
+
+
+class FakeStore:
+    """Commits refs like a transcript-aware checkpointer and reads them back."""
+
+    def __init__(self) -> None:
+        self.transcripts: dict[str, dict] = {}
+
+    def commit(self, ch: TranscriptChannel) -> TranscriptRef:
+        ref = ch.checkpoint()
+        t = self.transcripts.setdefault(
+            ref.transcript_id,
+            {"parent": ref.parent_transcript_id, "fork": ref.fork_seq, "entries": {}},
+        )
+        for e in ref.entries:
+            t["entries"][e.seq] = e
+        return ref
+
+    def _read(self, tid: str, head: int) -> list:
+        t = self.transcripts[tid]
+        out = self._read(t["parent"], min(head, t["fork"])) if t["parent"] else []
+        return out + [t["entries"][s] for s in sorted(t["entries"]) if s <= head]
+
+    def load(self, ref: TranscriptRef, *, tip: bool = True) -> TranscriptChannel:
+        return TranscriptChannel().from_checkpoint(
+            TranscriptRef(
+                ref.transcript_id,
+                ref.head,
+                entries=self._read(ref.transcript_id, ref.head),
+                is_tip=tip,
+            )
+        )
+
+
+def contents(ch: TranscriptChannel) -> list[str]:
+    return [m.content for m in ch.get()]
+
+
+def test_appends_and_resends() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0"), AIMessage("a0", id="a0")]])
+    ch = store.load(store.commit(ch))
+    ch.update([[HumanMessage("h1", id="h1")]])
+    ref = store.commit(ch)
+    assert [e.seq for e in ref.entries] == [3]
+    ch = store.load(ref)
+    assert contents(ch) == ["h0", "a0", "h1"]
+
+    ch.update([[AIMessage("a0", id="a0"), HumanMessage("h1", id="h1")]])
+    assert ch.checkpoint().entries == []
+
+
+def test_edit_forks_and_keeps_original() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0"), AIMessage("a0", id="a0")]])
+    ch = store.load(store.commit(ch))
+    ch.update([[HumanMessage("h1", id="h1"), AIMessage("a1", id="a1")]])
+    original = store.commit(ch)
+    ch = store.load(original)
+
+    ch.update([[AIMessage("a0-edited", id="a0")]])
+    edited = store.commit(ch)
+    assert edited.parent_transcript_id == original.transcript_id
+    assert edited.fork_seq == 1
+    assert contents(store.load(edited)) == ["h0", "a0-edited", "h1", "a1"]
+    assert contents(store.load(original, tip=False)) == ["h0", "a0", "h1", "a1"]
+
+    # A second fork before the first is committed branches off the same parent.
+    ch.update([[RemoveMessage(id="h1")]])
+    removed = store.commit(ch)
+    assert removed.parent_transcript_id == original.transcript_id
+    assert contents(store.load(removed)) == ["h0", "a0-edited", "a1"]
+
+
+def test_edit_before_any_commit_has_no_parent() -> None:
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("x", id="x"), AIMessage("y", id="y")]])
+    ch.update([[AIMessage("y2", id="y")]])
+    ref = ch.checkpoint()
+    assert (ref.parent_transcript_id, ref.fork_seq) == ("", 0)
+    assert contents(ch) == ["x", "y2"]
+
+
+def test_edit_inside_reset_replays_reset() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0")]])
+    ch = store.load(store.commit(ch))
+    ch.update(
+        [[RemoveMessage(id=REMOVE_ALL_MESSAGES), HumanMessage("p", id="p"), AIMessage("q", id="q")]]
+    )
+    ch = store.load(store.commit(ch))
+    ch.update([[AIMessage("q2", id="q")]])
+    ref = store.commit(ch)
+    assert [(e.seq, e.kind) for e in ref.entries] == [(2, "reset")]
+    assert contents(store.load(ref)) == ["p", "q2"]
+
+
+def test_load_non_tip_forks() -> None:
+    store = FakeStore()
+    ch = TranscriptChannel().from_checkpoint(MISSING)
+    ch.update([[HumanMessage("h0", id="h0")]])
+    first = store.commit(ch)
+    ch = store.load(first)
+    ch.update([[AIMessage("a0", id="a0")]])
+    store.commit(ch)
+
+    branch = store.load(first, tip=False)
+    branch.update([[AIMessage("other", id="o")]])
+    ref = store.commit(branch)
+    assert ref.parent_transcript_id == first.transcript_id
+    assert contents(store.load(ref)) == ["h0", "other"]

libs/sdk-py/CHANGELOG.md

@@ -1,6 +1,21 @@
 # Changelog
 
-## Unreleased
+## 0.4.12
+
+### Removed
+
+- **`PrioritySettings.fairness_weight`** — removed. The server stores only
+  `priority` and `fairness_group` on the thread; it ignores a
+  `fairness_weight` and logs a warning. Remove it from your callback.
+
+### Changed
+
+- **`PrioritySettings.priority`** — now says -5 runs first and 5 runs last,
+  matching how the server orders runs. 0.4.9 to 0.4.11 said the reverse, and
+  their example gave premium users 3, which runs them after everyone at the
+  default 0. Check any callback written from those docs.
+
+## 0.4.11 and earlier
 
 ### Added
 
@@ -38,7 +53,7 @@
   and name the object in the `priority` section of `langgraph.json`. The
   server calls the function with a `PriorityContext` at the first run on a
   thread with no settings and stores the `PrioritySettings` it returns
-  (`priority`, `fairness_group`, `fairness_weight`) on the thread. Requires
+  (`priority`, `fairness_group`) on the thread. Requires
   `langgraph-api` 0.18.0 or later and a `langgraph-cli` that supports the
   section; older versions of either ignore it.
 

libs/sdk-py/langgraph_sdk/__init__.py

@@ -5,7 +5,7 @@ from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
 from langgraph_sdk.priority import Priority
 from langgraph_sdk.priority.types import PriorityContext, PrioritySettings
 
-__version__ = "0.4.11"
+__version__ = "0.4.12"
 
 __all__ = [
     "Auth",

libs/sdk-py/langgraph_sdk/priority/__init__.py

@@ -102,7 +102,7 @@ class Priority:
         @priority.assign
         async def assign(ctx: PriorityContext) -> PrioritySettings | None:
             return {
-                "priority": 3 if "premium" in ctx.permissions else 0,
+                "priority": -3 if "premium" in ctx.permissions else 0,
                 # Stored unencrypted, so an opaque ID rather than personal data.
                 "fairness_group": ctx.user["org_id"],
             }
@@ -134,7 +134,7 @@ class Priority:
             @priority.assign
             async def assign(ctx: PriorityContext) -> PrioritySettings | None:
                 if ctx.cron_id is not None:
-                    return {"priority": -2}
+                    return {"priority": 2}
                 return None
             ```
 

libs/sdk-py/langgraph_sdk/priority/types.py

@@ -50,18 +50,20 @@ class PrioritySettings(typing.TypedDict, total=False):
 
     Every key is optional: a missing key or a key set to `None` takes its
     default, and returning `None` instead of a dict gives every key its default.
-    A value outside its range, or an empty `fairness_group`, is an error.
+    A `priority` outside -5 to 5, or a `fairness_group` longer than 256
+    characters, is an error.
     """
 
     priority: int | None
-    """An integer from -5 to 5, where 5 runs first and -5 runs last. Defaults to 0."""
+    """An integer from -5 to 5, where -5 runs first and 5 runs last. Defaults to 0."""
 
     fairness_group: str | None
-    """The thread's fairness group, a non-empty string. Defaults to `"default"`.
+    """Currently disabled.
+    The thread's fairness group, a string of at most 256 characters.
+
+    Defaults to `"default"`, as does an empty string. It has no effect on run
+    order for now.
 
     The server stores it unencrypted, even when the deployment uses custom
     encryption, so use an opaque ID rather than personal data.
     """
-
-    fairness_weight: int | None
-    """The weight of the thread's fairness group, an integer from 1 to 100. Defaults to 1."""

libs/sdk-py/tests/test_priority.py

@@ -120,9 +120,5 @@ def test_context_is_frozen():
 
 
 def test_every_settings_key_is_optional():
-    assert PrioritySettings.__optional_keys__ == {
-        "priority",
-        "fairness_group",
-        "fairness_weight",
-    }
+    assert PrioritySettings.__optional_keys__ == {"priority", "fairness_group"}
     assert PrioritySettings.__required_keys__ == frozenset()