mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-21 15:31:43 +02:00
Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e5b422455d | ||
|
|
6cd26d0e21 | ||
|
|
53cf85bece | ||
|
|
69d98cc0d9 | ||
|
|
ccb851effb | ||
|
|
a6731f4367 | ||
|
|
8a77332d0d | ||
|
|
a1ce197234 | ||
|
|
afbe7e48b1 |
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "langgraph-checkpoint-postgres"
|
||||
version = "3.1.2"
|
||||
version = "3.1.3"
|
||||
description = "Library with a Postgres implementation of LangGraph checkpoint saver."
|
||||
authors = []
|
||||
requires-python = ">=3.10"
|
||||
|
||||
Generated
+1
-1
@@ -353,7 +353,7 @@ test = [
|
||||
|
||||
[[package]]
|
||||
name = "langgraph-checkpoint-postgres"
|
||||
version = "3.1.2"
|
||||
version = "3.1.3"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "langgraph-checkpoint" },
|
||||
|
||||
@@ -2829,11 +2829,11 @@ graceful-fs@^4.2.11:
|
||||
integrity sha512-RbJ5/jmFcNNCcDV5o9eTnBLJ/HszWV0P73bc+Ff4nS/rJj+YaS6IGyiOL0VoBYX+l1Wrl3k63h/KrH+nhJ0XvQ==
|
||||
|
||||
handlebars@^4.7.9:
|
||||
version "4.7.9"
|
||||
resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.9.tgz#6f139082ab58dc4e5a0e51efe7db5ae890d56a0f"
|
||||
integrity sha512-4E71E0rpOaQuJR2A3xDZ+GM1HyWYv1clR58tC8emQNeQe3RH7MAzSbat+V0wG78LQBo6m6bzSG/L4pBuCsgnUQ==
|
||||
version "4.7.10"
|
||||
resolved "https://registry.yarnpkg.com/handlebars/-/handlebars-4.7.10.tgz#12419ea1b41391b3ffcf003f2164dcbce090c417"
|
||||
integrity sha512-P5VJMVM7qgBn6vjXMw8WG9uVI+ncf2pi72j4de4yz5ZULLj2RGqLYaKOYGsgyrViQ0tePOVlN1tDCCXXtFqXKg==
|
||||
dependencies:
|
||||
minimist "^1.2.5"
|
||||
minimist "^1.2.8"
|
||||
neo-async "^2.6.2"
|
||||
source-map "^0.6.1"
|
||||
wordwrap "^1.0.0"
|
||||
@@ -3920,7 +3920,7 @@ minimatch@^9.0.4:
|
||||
dependencies:
|
||||
brace-expansion "^2.0.2"
|
||||
|
||||
minimist@^1.2.0, minimist@^1.2.5, minimist@^1.2.6:
|
||||
minimist@^1.2.0, minimist@^1.2.6, minimist@^1.2.8:
|
||||
version "1.2.8"
|
||||
resolved "https://registry.yarnpkg.com/minimist/-/minimist-1.2.8.tgz#c1a464e7693302e082a075cee0c057741ac4772c"
|
||||
integrity sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==
|
||||
|
||||
@@ -1 +1 @@
|
||||
__version__ = "0.4.33"
|
||||
__version__ = "0.4.35"
|
||||
|
||||
@@ -231,15 +231,6 @@ class RequestedPlacement:
|
||||
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))
|
||||
|
||||
@@ -1055,15 +1046,28 @@ def _print_deployment_result(
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _strip_url_scheme(url: str) -> str:
|
||||
"""Return a registry reference without a trailing slash or URL scheme."""
|
||||
reference = url.rstrip("/")
|
||||
if "://" in reference:
|
||||
reference = reference.split("//", 1)[1]
|
||||
return reference
|
||||
|
||||
|
||||
# Artifact Registry's username for access tokens. Control planes whose registry
|
||||
# needs another one return it as `username`.
|
||||
_DEFAULT_REGISTRY_USERNAME = "oauth2accesstoken"
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _docker_config_for_token(registry_host: str, token: str):
|
||||
def _docker_config_for_token(registry_host: str, token: str, username: str):
|
||||
"""Create a temporary Docker config with only the push token.
|
||||
|
||||
Yields the path to a temporary config directory that can be passed
|
||||
to `docker --config <path>` so that system credential helpers
|
||||
(e.g. gcloud) don't interfere with the push token.
|
||||
"""
|
||||
auth_b64 = base64.b64encode(f"oauth2accesstoken:{token}".encode()).decode()
|
||||
auth_b64 = base64.b64encode(f"{username}:{token}".encode()).decode()
|
||||
config_data = {"auths": {registry_host: {"auth": auth_b64}}}
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with open(os.path.join(tmpdir, "config.json"), "w") as f:
|
||||
@@ -1201,7 +1205,7 @@ class DockerBuildCommand:
|
||||
return cls(("docker", "build"), ())
|
||||
flags: tuple[str, ...] = ("--platform", _DEPLOYMENT_PLATFORM, "--load")
|
||||
if not verbose:
|
||||
flags += ("--progress=quiet",)
|
||||
flags += ("--progress=plain",)
|
||||
return cls(("docker", "buildx", "build"), flags)
|
||||
|
||||
|
||||
@@ -1317,21 +1321,22 @@ def _run_local_build(
|
||||
raise click.ClickException(
|
||||
"Push token response missing token or registry_url"
|
||||
)
|
||||
username = push_data.get("username") or _DEFAULT_REGISTRY_USERNAME
|
||||
step += 1
|
||||
|
||||
normalized_registry = registry_url.rstrip("/")
|
||||
if "://" in normalized_registry:
|
||||
normalized_registry = normalized_registry.split("//", 1)[1]
|
||||
repo_seed = image_name or name or spec.config.parent.name
|
||||
remote_image = str(
|
||||
ImageReference(
|
||||
f"{normalized_registry}/{normalize_name(repo_seed)}",
|
||||
tag,
|
||||
if push_data.get("repository_url"):
|
||||
# The control plane chose the exact repository (an ECR repository
|
||||
# is a single image name, so nothing may be appended to it).
|
||||
repository = _strip_url_scheme(push_data["repository_url"])
|
||||
else:
|
||||
repo_seed = image_name or name or spec.config.parent.name
|
||||
repository = (
|
||||
f"{_strip_url_scheme(registry_url)}/{normalize_name(repo_seed)}"
|
||||
)
|
||||
)
|
||||
registry_host = normalized_registry.split("/")[0]
|
||||
remote_image = str(ImageReference(repository, tag))
|
||||
registry_host = repository.split("/")[0]
|
||||
|
||||
with _docker_config_for_token(registry_host, deployment_token) as cfg:
|
||||
with _docker_config_for_token(registry_host, deployment_token, username) as cfg:
|
||||
_log_deploy_step(step, f"Logging into {registry_host}")
|
||||
token_input = (
|
||||
deployment_token
|
||||
@@ -1343,7 +1348,7 @@ def _run_local_build(
|
||||
*_docker_argv(cfg),
|
||||
"login",
|
||||
"-u",
|
||||
"oauth2accesstoken",
|
||||
username,
|
||||
"--password-stdin",
|
||||
registry_host,
|
||||
input=token_input,
|
||||
@@ -1647,10 +1652,72 @@ class PublishedImage:
|
||||
ImageSource = BuildAndPush | PublishedImage
|
||||
|
||||
|
||||
def _resource_spec(image_pull_secrets: tuple[str, ...]) -> Mapping[str, object]:
|
||||
if not image_pull_secrets:
|
||||
return _OPERATOR_DEFAULT_RESOURCE_SPEC
|
||||
return {
|
||||
"image_pull_secrets": [{"name": name} for name in image_pull_secrets],
|
||||
}
|
||||
|
||||
|
||||
# Mirrors the control plane's own LocalObjectReference.name validation (a
|
||||
# Kubernetes DNS subdomain), so a bad secret name is rejected here instead of
|
||||
# after a build and push.
|
||||
_K8S_SECRET_NAME_RE = re.compile(
|
||||
r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CreateOnlySettings:
|
||||
"""Options for an external-image deployment that only apply when
|
||||
creating one, never when updating an existing deployment."""
|
||||
|
||||
placement: RequestedPlacement
|
||||
image_pull_secrets: tuple[str, ...] = ()
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
for name in self.image_pull_secrets:
|
||||
if len(name) > 253 or not _K8S_SECRET_NAME_RE.match(name):
|
||||
raise click.UsageError(
|
||||
f"--image-pull-secret {name!r} is not a valid Kubernetes "
|
||||
"secret name (lowercase alphanumeric characters, '-', or "
|
||||
"'.' only, max 253 characters)."
|
||||
)
|
||||
|
||||
@property
|
||||
def requested(self) -> bool:
|
||||
return self.placement.requested or bool(self.image_pull_secrets)
|
||||
|
||||
def ensure_creatable(self, push_to: str | None, image_uri: str | None) -> None:
|
||||
if push_to is None and image_uri is None and self.requested:
|
||||
raise click.UsageError(
|
||||
"--listener-id, --k8s-namespace, and --image-pull-secret only "
|
||||
"apply when creating a deployment with --push-to or --image-uri."
|
||||
)
|
||||
|
||||
def ensure_not_existing(self, deployment_id: str) -> None:
|
||||
if self.placement.requested:
|
||||
raise click.UsageError(
|
||||
"Listener and namespace are fixed when a deployment is "
|
||||
f"created. Deployment {deployment_id} already exists, so "
|
||||
"drop --listener-id and --k8s-namespace, or create a new "
|
||||
"deployment with a different --name."
|
||||
)
|
||||
if self.image_pull_secrets:
|
||||
raise click.UsageError(
|
||||
"langgraph deploy does not update image pull secrets on an "
|
||||
f"existing deployment. Deployment {deployment_id} already "
|
||||
"exists, so drop --image-pull-secret and update resource_spec "
|
||||
"through the API or the LangSmith UI instead, or create a "
|
||||
"new deployment with a different --name."
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CustomerRegistrySource:
|
||||
image: ImageSource
|
||||
requested_placement: RequestedPlacement
|
||||
create_only: CreateOnlySettings
|
||||
|
||||
def run(self, ctx: DeployContext) -> DeployOutcome:
|
||||
if isinstance(ctx.selector, ById):
|
||||
@@ -1672,7 +1739,7 @@ class CustomerRegistrySource:
|
||||
self, ctx: DeployContext, existing: ExistingDeployment, step: int
|
||||
) -> DeployOutcome:
|
||||
_ensure_customer_registry_source(existing)
|
||||
self.requested_placement.ensure_not_requested(existing.id)
|
||||
self.create_only.ensure_not_existing(existing.id)
|
||||
image_uri, step = self._publish(ctx, step)
|
||||
_log_deploy_step(step, f"Updating deployment {existing.id}")
|
||||
updated = ctx.client.update_deployment(
|
||||
@@ -1687,7 +1754,7 @@ class CustomerRegistrySource:
|
||||
)
|
||||
|
||||
def _resolve_placement(self, ctx: DeployContext) -> Placement:
|
||||
requested = self.requested_placement
|
||||
requested = self.create_only.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):
|
||||
@@ -1716,7 +1783,9 @@ class CustomerRegistrySource:
|
||||
else None,
|
||||
source=_CUSTOMER_REGISTRY_SOURCE,
|
||||
source_config={
|
||||
"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
|
||||
"resource_spec": _resource_spec(
|
||||
self.create_only.image_pull_secrets
|
||||
),
|
||||
**placement.source_config(),
|
||||
},
|
||||
source_revision_config={"image_uri": image_uri},
|
||||
@@ -1810,19 +1879,13 @@ def _select_source(
|
||||
tag: str | None,
|
||||
remote_build_flag: bool | None,
|
||||
placement: RequestedPlacement,
|
||||
image_pull_secrets: tuple[str, ...],
|
||||
selector: DeploymentSelector,
|
||||
) -> DeploymentSource:
|
||||
if push_to is None and image_uri is None and placement.requested:
|
||||
raise click.UsageError(
|
||||
"--listener-id and --k8s-namespace only apply when creating a "
|
||||
"deployment with --push-to or --image-uri."
|
||||
)
|
||||
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."
|
||||
)
|
||||
create_only = CreateOnlySettings(placement, image_pull_secrets)
|
||||
create_only.ensure_creatable(push_to, image_uri)
|
||||
if create_only.requested and isinstance(selector, ById):
|
||||
create_only.ensure_not_existing(selector.deployment_id)
|
||||
if image_uri is not None:
|
||||
if push_to is not None:
|
||||
raise click.UsageError("--image-uri cannot be combined with --push-to.")
|
||||
@@ -1834,7 +1897,7 @@ def _select_source(
|
||||
raise click.UsageError("--image-uri cannot be combined with --remote.")
|
||||
return CustomerRegistrySource(
|
||||
image=PublishedImage(_validate_image_uri(image_uri)),
|
||||
requested_placement=placement,
|
||||
create_only=create_only,
|
||||
)
|
||||
if push_to is not None:
|
||||
if remote_build_flag is True:
|
||||
@@ -1844,7 +1907,7 @@ def _select_source(
|
||||
_require_local_docker()
|
||||
return CustomerRegistrySource(
|
||||
image=BuildAndPush(reference=reference, prebuilt_image=image),
|
||||
requested_placement=placement,
|
||||
create_only=create_only,
|
||||
)
|
||||
if image and remote_build_flag is True:
|
||||
raise click.UsageError("--image cannot be combined with --remote builds.")
|
||||
@@ -1874,7 +1937,10 @@ def _create_host_backend_client(
|
||||
host_url: str | None,
|
||||
api_key: str | None,
|
||||
env_vars: dict[str, str] | None = None,
|
||||
*,
|
||||
no_input: bool | None = None,
|
||||
) -> HostBackendClient:
|
||||
effective_no_input = _no_input if no_input is None else no_input
|
||||
if env_vars is None:
|
||||
env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG)
|
||||
resolved_api_key = api_key
|
||||
@@ -1889,7 +1955,7 @@ def _create_host_backend_client(
|
||||
resolved_api_key = val
|
||||
break
|
||||
if not resolved_api_key:
|
||||
if _no_input:
|
||||
if effective_no_input:
|
||||
raise click.ClickException(
|
||||
"No LangSmith API key found. Set LANGSMITH_API_KEY in the "
|
||||
"environment or .env file."
|
||||
@@ -1914,6 +1980,8 @@ def _create_host_backend_client(
|
||||
def _call_host_backend_with_optional_tenant(
|
||||
client: HostBackendClient,
|
||||
operation: Callable[[HostBackendClient], _T],
|
||||
*,
|
||||
no_input: bool | None = None,
|
||||
) -> _T:
|
||||
"""Run *operation*, prompting for a workspace ID on org-scoped 403s.
|
||||
|
||||
@@ -1922,6 +1990,7 @@ def _call_host_backend_with_optional_tenant(
|
||||
in-place so all subsequent calls through the same instance are
|
||||
tenant-aware.
|
||||
"""
|
||||
effective_no_input = _no_input if no_input is None else no_input
|
||||
prompted_for_tenant = False
|
||||
|
||||
while True:
|
||||
@@ -1933,7 +2002,7 @@ def _call_host_backend_with_optional_tenant(
|
||||
and err.status_code == 403
|
||||
and "requires workspace specification" in err.message
|
||||
):
|
||||
if _no_input:
|
||||
if effective_no_input:
|
||||
raise click.ClickException(
|
||||
"API key is org-scoped and requires a workspace ID. "
|
||||
"Set LANGSMITH_TENANT_ID in your .env file or "
|
||||
@@ -2182,6 +2251,18 @@ def _deploy_base_options(
|
||||
"creating a deployment with --push-to or --image-uri."
|
||||
),
|
||||
),
|
||||
click.option(
|
||||
"--image-pull-secret",
|
||||
"image_pull_secrets",
|
||||
multiple=True,
|
||||
help=(
|
||||
"Name of a Kubernetes secret (type kubernetes.io/"
|
||||
"dockerconfigjson) in the deployment's namespace to use for "
|
||||
"pulling the image from a private registry. Repeat to give "
|
||||
"more than one. Only used when creating a deployment with "
|
||||
"--push-to or --image-uri."
|
||||
),
|
||||
),
|
||||
click.option(
|
||||
"--config",
|
||||
"-c",
|
||||
@@ -2295,6 +2376,7 @@ def _deploy_cmd(
|
||||
image_uri: str | None,
|
||||
listener_id: str | None,
|
||||
k8s_namespace: str | None,
|
||||
image_pull_secrets: tuple[str, ...],
|
||||
tag: str | None,
|
||||
base_image: str | None,
|
||||
install_command: str | None,
|
||||
@@ -2371,6 +2453,7 @@ def _deploy_cmd(
|
||||
tag=tag,
|
||||
remote_build_flag=remote_build_flag,
|
||||
placement=RequestedPlacement(listener_id, k8s_namespace),
|
||||
image_pull_secrets=image_pull_secrets,
|
||||
selector=selector,
|
||||
)
|
||||
|
||||
@@ -2518,6 +2601,13 @@ def deploy_listeners() -> None:
|
||||
|
||||
@OPT_HOST_API_KEY
|
||||
@OPT_HOST_URL
|
||||
@click.option(
|
||||
"--json",
|
||||
"json_output",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Emit the listeners as a JSON array instead of a table.",
|
||||
)
|
||||
@deploy_listeners.command(
|
||||
"list",
|
||||
help=(
|
||||
@@ -2526,11 +2616,16 @@ def deploy_listeners() -> None:
|
||||
"--listener-id <id>` to deploy through it."
|
||||
),
|
||||
)
|
||||
def deploy_listeners_list(api_key: str | None, host_url: str | None) -> None:
|
||||
client = _create_host_backend_client(host_url, api_key)
|
||||
def deploy_listeners_list(
|
||||
api_key: str | None, host_url: str | None, json_output: bool
|
||||
) -> None:
|
||||
client = _create_host_backend_client(host_url, api_key, no_input=json_output)
|
||||
listeners = _call_host_backend_with_optional_tenant(
|
||||
client, lambda c: c.list_listeners()
|
||||
client, lambda c: c.list_listeners(), no_input=json_output
|
||||
)
|
||||
if json_output:
|
||||
click.echo(json_mod.dumps(listeners))
|
||||
return
|
||||
if not listeners:
|
||||
click.echo("No listeners found for this workspace.")
|
||||
return
|
||||
|
||||
@@ -249,9 +249,12 @@ class HostBackendClient:
|
||||
return self._request("DELETE", f"/v2/deployments/{deployment_id}")
|
||||
|
||||
def request_push_token(self, deployment_id: str) -> dict[str, Any]:
|
||||
# Tells the control plane this client logs in with the returned
|
||||
# `username` and pushes to `repository_url`. LangSmith AWS requires it.
|
||||
return self._request(
|
||||
"POST",
|
||||
f"/v2/deployments/{deployment_id}/push-token",
|
||||
{"supports_registry_username": True},
|
||||
)
|
||||
|
||||
def request_upload_url(self, deployment_id: str) -> dict[str, Any]:
|
||||
|
||||
@@ -14,6 +14,7 @@ import langgraph_cli.deploy as deploy_module
|
||||
from langgraph_cli.cli import cli, prepare_args_and_stdin
|
||||
from langgraph_cli.config import Config, _get_pip_cleanup_lines, validate_config
|
||||
from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Version
|
||||
from langgraph_cli.host_backend import HostBackendError
|
||||
from langgraph_cli.util import clean_empty_lines
|
||||
|
||||
FORMATTED_CLEANUP_LINES = _get_pip_cleanup_lines(
|
||||
@@ -535,6 +536,121 @@ def test_deploy_listeners_list_command_no_results(monkeypatch) -> None:
|
||||
assert result.output.strip() == "No listeners found for this workspace."
|
||||
|
||||
|
||||
def test_deploy_listeners_list_command_json(monkeypatch) -> None:
|
||||
runner = CliRunner()
|
||||
|
||||
listeners = [
|
||||
{
|
||||
"id": "listener-1",
|
||||
"compute_id": "prod-cluster",
|
||||
"compute_config": {"k8s_namespaces": ["agents"]},
|
||||
},
|
||||
{
|
||||
"id": "listener-2",
|
||||
"compute_id": "multi-cluster",
|
||||
"compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
|
||||
},
|
||||
]
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
|
||||
pass
|
||||
|
||||
def list_listeners(self):
|
||||
return listeners
|
||||
|
||||
monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
|
||||
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
"deploy",
|
||||
"listeners",
|
||||
"list",
|
||||
"--json",
|
||||
"--api-key",
|
||||
"test-key",
|
||||
"--host-url",
|
||||
"https://api.example.com",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert json.loads(result.output) == listeners
|
||||
|
||||
|
||||
def test_deploy_listeners_list_command_json_no_results(monkeypatch) -> None:
|
||||
runner = CliRunner()
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
|
||||
pass
|
||||
|
||||
def list_listeners(self):
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
|
||||
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
"deploy",
|
||||
"listeners",
|
||||
"list",
|
||||
"--json",
|
||||
"--api-key",
|
||||
"test-key",
|
||||
"--host-url",
|
||||
"https://api.example.com",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert json.loads(result.output) == []
|
||||
|
||||
|
||||
def test_deploy_listeners_list_command_json_org_scoped_403(monkeypatch) -> None:
|
||||
"""With --json, an org-scoped 403 must raise cleanly instead of
|
||||
prompting for a workspace ID, which would write to stdout and hang
|
||||
waiting for input that a scripted invocation never provides."""
|
||||
runner = CliRunner()
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
|
||||
pass
|
||||
|
||||
def list_listeners(self):
|
||||
raise HostBackendError("requires workspace specification", status_code=403)
|
||||
|
||||
def set_tenant(self, tenant_id: str) -> None:
|
||||
raise AssertionError("must not retry with a prompted tenant with --json")
|
||||
|
||||
monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
|
||||
|
||||
prompt_calls = []
|
||||
monkeypatch.setattr(
|
||||
click, "prompt", lambda *a, **k: prompt_calls.append((a, k)) or "unreachable"
|
||||
)
|
||||
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
"deploy",
|
||||
"listeners",
|
||||
"list",
|
||||
"--json",
|
||||
"--api-key",
|
||||
"test-key",
|
||||
"--host-url",
|
||||
"https://api.example.com",
|
||||
],
|
||||
)
|
||||
|
||||
assert prompt_calls == [], "must not prompt interactively with --json"
|
||||
assert result.exit_code != 0
|
||||
assert "LANGSMITH_TENANT_ID" in result.output
|
||||
|
||||
|
||||
def test_deploy_revisions_list_command(monkeypatch) -> None:
|
||||
runner = CliRunner()
|
||||
captured: dict[str, str] = {}
|
||||
|
||||
@@ -93,6 +93,7 @@ class ControlPlaneDouble:
|
||||
timeline: list[str]
|
||||
existing_deployments: list[dict] = field(default_factory=list)
|
||||
push_token_status: int = 200
|
||||
push_token_extra: dict = field(default_factory=dict)
|
||||
create_error: str | None = None
|
||||
listeners: list[dict] = field(default_factory=list)
|
||||
listeners_by_id: dict[str, dict] = field(default_factory=dict)
|
||||
@@ -147,7 +148,12 @@ class ControlPlaneDouble:
|
||||
if self.push_token_status != 200:
|
||||
return httpx.Response(self.push_token_status, text=NOT_A_CLI_DEPLOYMENT)
|
||||
return httpx.Response(
|
||||
200, json={"token": PUSH_TOKEN, "registry_url": REGISTRY_URL}
|
||||
200,
|
||||
json={
|
||||
"token": PUSH_TOKEN,
|
||||
"registry_url": REGISTRY_URL,
|
||||
**self.push_token_extra,
|
||||
},
|
||||
)
|
||||
if path.endswith("/upload-url"):
|
||||
return httpx.Response(
|
||||
@@ -363,7 +369,7 @@ def test_first_local_deploy_creates_an_internal_docker_deployment(
|
||||
pytest.param(
|
||||
"arm64",
|
||||
("docker", "buildx", "build"),
|
||||
("--platform", "linux/amd64", "--load", "--progress=quiet"),
|
||||
("--platform", "linux/amd64", "--load", "--progress=plain"),
|
||||
id="apple_silicon_cross_builds_for_linux_amd64",
|
||||
),
|
||||
pytest.param(
|
||||
@@ -410,6 +416,48 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(
|
||||
assert login.kwargs["input"] == f"{PUSH_TOKEN}\n"
|
||||
|
||||
|
||||
def test_local_deploy_tells_the_control_plane_it_honors_registry_usernames(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
deploy_project.run("--no-remote")
|
||||
|
||||
assert deploy_project.control_plane.bodies[_push_token(CREATED_ID)] == {
|
||||
"supports_registry_username": True
|
||||
}
|
||||
|
||||
|
||||
ECR_REPOSITORY = "123456789012.dkr.ecr.us-east-2.amazonaws.com/lsd/agent-1"
|
||||
ECR_PUSH_TOKEN_FIELDS = {"username": "AWS", "repository_url": ECR_REPOSITORY}
|
||||
|
||||
|
||||
def test_local_deploy_logs_in_with_the_username_the_control_plane_returns(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
deploy_project.control_plane.push_token_extra = ECR_PUSH_TOKEN_FIELDS
|
||||
|
||||
deploy_project.run("--no-remote")
|
||||
|
||||
login = deploy_project.docker.command("login")
|
||||
assert login.args[3:] == (
|
||||
"login",
|
||||
"-u",
|
||||
"AWS",
|
||||
"--password-stdin",
|
||||
"123456789012.dkr.ecr.us-east-2.amazonaws.com",
|
||||
)
|
||||
|
||||
|
||||
def test_local_deploy_pushes_to_exactly_the_returned_repository(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
deploy_project.control_plane.push_token_extra = ECR_PUSH_TOKEN_FIELDS
|
||||
|
||||
deploy_project.run("--no-remote")
|
||||
|
||||
tag = deploy_project.docker.command("tag")
|
||||
assert tag.args[-1] == f"{ECR_REPOSITORY}:latest"
|
||||
|
||||
|
||||
def test_local_deploy_tags_the_build_into_the_token_registry(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
@@ -918,6 +966,100 @@ def test_self_hosted_control_plane_places_when_asked(
|
||||
}
|
||||
|
||||
|
||||
def test_image_pull_secret_is_included_in_resource_spec_on_create(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
result = deploy_project.run(
|
||||
"--push-to",
|
||||
PUSH_REPOSITORY,
|
||||
"--image-pull-secret",
|
||||
"regcred",
|
||||
"--image-pull-secret",
|
||||
"other-secret",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
|
||||
"resource_spec": {
|
||||
"image_pull_secrets": [
|
||||
{"name": "regcred"},
|
||||
{"name": "other-secret"},
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_image_pull_secret_works_with_image_uri_too(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
result = deploy_project.run(
|
||||
"--image-uri", EXTERNAL_DIGEST, "--image-pull-secret", "regcred"
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
|
||||
"resource_spec": {"image_pull_secrets": [{"name": "regcred"}]},
|
||||
}
|
||||
|
||||
|
||||
def test_image_pull_secret_without_push_to_or_image_uri_is_refused(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
result = deploy_project.run("--image-pull-secret", "regcred")
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "only apply when creating a deployment with --push-to" in result.output
|
||||
assert deploy_project.timeline == []
|
||||
|
||||
|
||||
def test_an_invalid_image_pull_secret_name_is_rejected_before_any_docker_work(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
result = deploy_project.run(
|
||||
"--push-to", PUSH_REPOSITORY, "--image-pull-secret", "Not_Valid!"
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "not a valid Kubernetes" in result.output
|
||||
assert deploy_project.docker.verbs() == []
|
||||
assert deploy_project.timeline == []
|
||||
|
||||
|
||||
def test_image_pull_secret_cannot_update_an_existing_deployment(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
result = deploy_project.run(
|
||||
"--deployment-id",
|
||||
"dep-1",
|
||||
"--push-to",
|
||||
PUSH_REPOSITORY,
|
||||
"--image-pull-secret",
|
||||
"regcred",
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "--image-pull-secret" in result.output
|
||||
assert "dep-1" in result.output
|
||||
assert deploy_project.timeline == []
|
||||
|
||||
|
||||
def test_image_pull_secret_cannot_update_a_deployment_found_by_name(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
deploy_project.control_plane.existing_deployments = [
|
||||
{"id": "dep-ext", "name": "my-app", "source": "external_docker"}
|
||||
]
|
||||
|
||||
result = deploy_project.run(
|
||||
"--image-uri", EXTERNAL_DIGEST, "--image-pull-secret", "regcred"
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "--image-pull-secret" in result.output
|
||||
assert deploy_project.docker.verbs() == []
|
||||
assert _patch("dep-ext") not in deploy_project.timeline
|
||||
|
||||
|
||||
def test_updating_a_deployment_never_looks_up_listeners(
|
||||
deploy_project: DeployProject,
|
||||
) -> None:
|
||||
|
||||
@@ -16,6 +16,7 @@ from langgraph_cli.deploy import (
|
||||
BuildAndPush,
|
||||
ById,
|
||||
ByName,
|
||||
CreateOnlySettings,
|
||||
CustomerRegistrySource,
|
||||
DockerBuildCommand,
|
||||
ExistingDeployment,
|
||||
@@ -34,6 +35,7 @@ from langgraph_cli.deploy import (
|
||||
_parse_env_from_config,
|
||||
_resolve_env_path,
|
||||
_resolve_pushed_image_digest,
|
||||
_resource_spec,
|
||||
_select_source,
|
||||
_validate_prebuilt_image,
|
||||
find_deployment_by_name,
|
||||
@@ -46,7 +48,9 @@ from langgraph_cli.image_reference import ImageReference
|
||||
|
||||
class TestDockerConfigForToken:
|
||||
def test_creates_config_json(self):
|
||||
with _docker_config_for_token("us-docker.pkg.dev", "my-token") as cfg:
|
||||
with _docker_config_for_token(
|
||||
"us-docker.pkg.dev", "my-token", "oauth2accesstoken"
|
||||
) as cfg:
|
||||
config_path = os.path.join(cfg, "config.json")
|
||||
assert os.path.isfile(config_path)
|
||||
with open(config_path) as f:
|
||||
@@ -54,13 +58,26 @@ class TestDockerConfigForToken:
|
||||
expected_auth = base64.b64encode(b"oauth2accesstoken:my-token").decode()
|
||||
assert data == {"auths": {"us-docker.pkg.dev": {"auth": expected_auth}}}
|
||||
|
||||
def test_uses_the_given_username(self):
|
||||
with _docker_config_for_token(
|
||||
"123456789012.dkr.ecr.us-east-2.amazonaws.com", "ecr-password", "AWS"
|
||||
) as cfg:
|
||||
with open(os.path.join(cfg, "config.json")) as f:
|
||||
data = json.load(f)
|
||||
expected_auth = base64.b64encode(b"AWS:ecr-password").decode()
|
||||
assert data["auths"]["123456789012.dkr.ecr.us-east-2.amazonaws.com"] == {
|
||||
"auth": expected_auth
|
||||
}
|
||||
|
||||
def test_tempdir_cleaned_up(self):
|
||||
with _docker_config_for_token("registry.example.com", "tok") as cfg:
|
||||
with _docker_config_for_token(
|
||||
"registry.example.com", "tok", "oauth2accesstoken"
|
||||
) as cfg:
|
||||
assert os.path.isdir(cfg)
|
||||
assert not os.path.exists(cfg)
|
||||
|
||||
def test_different_registries(self):
|
||||
with _docker_config_for_token("gcr.io", "token123") as cfg:
|
||||
with _docker_config_for_token("gcr.io", "token123", "oauth2accesstoken") as cfg:
|
||||
with open(os.path.join(cfg, "config.json")) as f:
|
||||
data = json.load(f)
|
||||
assert "gcr.io" in data["auths"]
|
||||
@@ -593,9 +610,9 @@ class TestDockerBuildCommand:
|
||||
False,
|
||||
DockerBuildCommand(
|
||||
("docker", "buildx", "build"),
|
||||
("--platform", "linux/amd64", "--load", "--progress=quiet"),
|
||||
("--platform", "linux/amd64", "--load", "--progress=plain"),
|
||||
),
|
||||
id="other_hosts_cross_build_quietly",
|
||||
id="other_hosts_capture_plain_build_logs",
|
||||
),
|
||||
pytest.param(
|
||||
"arm64",
|
||||
@@ -621,6 +638,7 @@ class TestSelectSource:
|
||||
"tag": None,
|
||||
"remote_build_flag": None,
|
||||
"placement": RequestedPlacement(),
|
||||
"image_pull_secrets": (),
|
||||
"selector": ByName("my-app"),
|
||||
}
|
||||
REPOSITORY = "registry.example.com/app"
|
||||
@@ -636,7 +654,7 @@ class TestSelectSource:
|
||||
reference=ImageReference(REPOSITORY, "latest"),
|
||||
prebuilt_image=None,
|
||||
),
|
||||
requested_placement=RequestedPlacement(),
|
||||
create_only=CreateOnlySettings(RequestedPlacement()),
|
||||
),
|
||||
id="push_to_selects_the_external_source_with_the_default_tag",
|
||||
),
|
||||
@@ -648,7 +666,7 @@ class TestSelectSource:
|
||||
reference=ImageReference(REPOSITORY, "v2"),
|
||||
prebuilt_image=None,
|
||||
),
|
||||
requested_placement=RequestedPlacement(),
|
||||
create_only=CreateOnlySettings(RequestedPlacement()),
|
||||
),
|
||||
id="push_to_keeps_a_tag_given_in_the_reference",
|
||||
),
|
||||
@@ -660,7 +678,7 @@ class TestSelectSource:
|
||||
reference=ImageReference(REPOSITORY, "v3"),
|
||||
prebuilt_image=None,
|
||||
),
|
||||
requested_placement=RequestedPlacement(),
|
||||
create_only=CreateOnlySettings(RequestedPlacement()),
|
||||
),
|
||||
id="tag_flag_composes_with_push_to",
|
||||
),
|
||||
@@ -672,7 +690,7 @@ class TestSelectSource:
|
||||
reference=ImageReference(REPOSITORY, "latest"),
|
||||
prebuilt_image="app:dev",
|
||||
),
|
||||
requested_placement=RequestedPlacement(),
|
||||
create_only=CreateOnlySettings(RequestedPlacement()),
|
||||
),
|
||||
id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
|
||||
),
|
||||
@@ -687,16 +705,35 @@ class TestSelectSource:
|
||||
reference=ImageReference(REPOSITORY, "latest"),
|
||||
prebuilt_image=None,
|
||||
),
|
||||
requested_placement=RequestedPlacement("listener-1", "agents"),
|
||||
create_only=CreateOnlySettings(
|
||||
RequestedPlacement("listener-1", "agents")
|
||||
),
|
||||
),
|
||||
id="push_to_carries_the_requested_placement",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"push_to": REPOSITORY,
|
||||
"image_pull_secrets": ("regcred", "other-secret"),
|
||||
},
|
||||
True,
|
||||
CustomerRegistrySource(
|
||||
image=BuildAndPush(
|
||||
reference=ImageReference(REPOSITORY, "latest"),
|
||||
prebuilt_image=None,
|
||||
),
|
||||
create_only=CreateOnlySettings(
|
||||
RequestedPlacement(), ("regcred", "other-secret")
|
||||
),
|
||||
),
|
||||
id="push_to_carries_the_image_pull_secrets",
|
||||
),
|
||||
pytest.param(
|
||||
{"image_uri": f"{REPOSITORY}@sha256:abc123"},
|
||||
True,
|
||||
CustomerRegistrySource(
|
||||
image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
|
||||
requested_placement=RequestedPlacement(),
|
||||
create_only=CreateOnlySettings(RequestedPlacement()),
|
||||
),
|
||||
id="image_uri_selects_the_published_image_source_by_digest",
|
||||
),
|
||||
@@ -705,7 +742,7 @@ class TestSelectSource:
|
||||
False,
|
||||
CustomerRegistrySource(
|
||||
image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
|
||||
requested_placement=RequestedPlacement(),
|
||||
create_only=CreateOnlySettings(RequestedPlacement()),
|
||||
),
|
||||
id="image_uri_needs_no_local_docker_and_is_trimmed",
|
||||
),
|
||||
@@ -717,7 +754,9 @@ class TestSelectSource:
|
||||
False,
|
||||
CustomerRegistrySource(
|
||||
image=PublishedImage(f"{REPOSITORY}@sha256:abc123"),
|
||||
requested_placement=RequestedPlacement("listener-1", "agents"),
|
||||
create_only=CreateOnlySettings(
|
||||
RequestedPlacement("listener-1", "agents")
|
||||
),
|
||||
),
|
||||
id="image_uri_carries_the_requested_placement",
|
||||
),
|
||||
@@ -806,6 +845,11 @@ class TestSelectSource:
|
||||
"only apply when creating a deployment with --push-to",
|
||||
id="namespace_without_push_to",
|
||||
),
|
||||
pytest.param(
|
||||
{"image_pull_secrets": ("regcred",)},
|
||||
"only apply when creating a deployment with --push-to",
|
||||
id="image_pull_secret_without_push_to",
|
||||
),
|
||||
pytest.param(
|
||||
{"image_uri": REPOSITORY, "push_to": REPOSITORY},
|
||||
"--image-uri cannot be combined with --push-to.",
|
||||
@@ -1243,7 +1287,7 @@ def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
|
||||
|
||||
monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
|
||||
|
||||
with pytest.raises(click.UsageError, match="--deployment-id"):
|
||||
with pytest.raises(click.UsageError, match="dep-1"):
|
||||
_select_source(
|
||||
push_to="registry.example.com/app",
|
||||
image=None,
|
||||
@@ -1252,10 +1296,77 @@ def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
|
||||
tag=None,
|
||||
remote_build_flag=None,
|
||||
placement=RequestedPlacement(listener_id="listener-1"),
|
||||
image_pull_secrets=(),
|
||||
selector=ById("dep-1"),
|
||||
)
|
||||
|
||||
|
||||
def test_a_deployment_id_with_image_pull_secrets_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="dep-1"):
|
||||
_select_source(
|
||||
push_to="registry.example.com/app",
|
||||
image=None,
|
||||
image_uri=None,
|
||||
image_name=None,
|
||||
tag=None,
|
||||
remote_build_flag=None,
|
||||
placement=RequestedPlacement(),
|
||||
image_pull_secrets=("regcred",),
|
||||
selector=ById("dep-1"),
|
||||
)
|
||||
|
||||
|
||||
class TestResourceSpec:
|
||||
def test_no_secrets_keeps_the_operator_default(self):
|
||||
assert _resource_spec(()) == {}
|
||||
|
||||
def test_secrets_become_local_object_references_in_order(self):
|
||||
assert _resource_spec(("regcred", "other-secret")) == {
|
||||
"image_pull_secrets": [
|
||||
{"name": "regcred"},
|
||||
{"name": "other-secret"},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
class TestCreateOnlySettings:
|
||||
@pytest.mark.parametrize(
|
||||
"name",
|
||||
[
|
||||
pytest.param("Bad_Name", id="uppercase_and_underscore"),
|
||||
pytest.param("", id="empty"),
|
||||
pytest.param("-leading-hyphen", id="leading_hyphen"),
|
||||
pytest.param("a" * 254, id="too_long"),
|
||||
],
|
||||
)
|
||||
def test_an_invalid_secret_name_is_rejected_up_front(self, name):
|
||||
with pytest.raises(click.UsageError, match="not a valid Kubernetes"):
|
||||
CreateOnlySettings(RequestedPlacement(), (name,))
|
||||
|
||||
def test_a_valid_secret_name_is_accepted(self):
|
||||
CreateOnlySettings(RequestedPlacement(), ("regcred-1.example",))
|
||||
|
||||
def test_ensure_not_existing_mentions_the_api_and_ui_for_pull_secrets(self):
|
||||
settings = CreateOnlySettings(RequestedPlacement(), ("regcred",))
|
||||
with pytest.raises(click.UsageError, match="API or the LangSmith UI"):
|
||||
settings.ensure_not_existing("dep-1")
|
||||
|
||||
def test_ensure_not_existing_keeps_the_placement_message_unchanged(self):
|
||||
settings = CreateOnlySettings(RequestedPlacement(listener_id="listener-1"))
|
||||
with pytest.raises(
|
||||
click.UsageError,
|
||||
match="Listener and namespace are fixed when a deployment is created",
|
||||
):
|
||||
settings.ensure_not_existing("dep-1")
|
||||
|
||||
|
||||
class TestPlacementOnAKnownListener:
|
||||
@pytest.mark.parametrize(
|
||||
("request_", "listener", "expected"),
|
||||
|
||||
@@ -2,6 +2,7 @@ import collections.abc
|
||||
from collections.abc import Callable, Sequence
|
||||
from typing import Any, Generic
|
||||
|
||||
from langgraph.checkpoint.serde.types import _DeltaSnapshot
|
||||
from typing_extensions import NotRequired, Required, Self
|
||||
|
||||
from langgraph._internal._constants import OVERWRITE
|
||||
@@ -116,6 +117,9 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
|
||||
def from_checkpoint(self, checkpoint: Value) -> Self:
|
||||
empty = self.__class__(self.typ, self.operator)
|
||||
empty.key = self.key
|
||||
# A thread written while this field was a DeltaChannel stores snapshots.
|
||||
if isinstance(checkpoint, _DeltaSnapshot):
|
||||
checkpoint = checkpoint.value
|
||||
if checkpoint is not MISSING:
|
||||
empty.value = checkpoint
|
||||
return empty
|
||||
|
||||
@@ -4,7 +4,7 @@ import uuid
|
||||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||||
from datetime import datetime, timezone
|
||||
from inspect import signature
|
||||
from typing import Any, Literal, cast
|
||||
from typing import Any, Literal
|
||||
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
from langgraph.checkpoint.base import (
|
||||
@@ -12,6 +12,7 @@ from langgraph.checkpoint.base import (
|
||||
ChannelVersions,
|
||||
Checkpoint,
|
||||
CheckpointTuple,
|
||||
DeltaChannelHistory,
|
||||
PendingWrite,
|
||||
)
|
||||
from langgraph.checkpoint.base.id import uuid6
|
||||
@@ -32,7 +33,7 @@ from langgraph._internal._constants import (
|
||||
)
|
||||
from langgraph._internal._typing import MISSING
|
||||
from langgraph.channels.base import BaseChannel
|
||||
from langgraph.channels.binop import _get_overwrite
|
||||
from langgraph.channels.binop import BinaryOperatorAggregate, _get_overwrite
|
||||
from langgraph.channels.delta import DeltaChannel
|
||||
from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
|
||||
|
||||
@@ -409,23 +410,39 @@ def versions_seen_without_bumps(
|
||||
def _delta_channels_to_replay(
|
||||
specs: Mapping[str, BaseChannel], checkpoint: Checkpoint
|
||||
) -> list[str]:
|
||||
"""DeltaChannels whose value at `checkpoint` the ancestor walk rebuilds.
|
||||
"""Channels whose value at `checkpoint` the ancestor walk rebuilds.
|
||||
|
||||
A `_DeltaSnapshot` blob or a plain value (migration) resolves directly via
|
||||
`from_checkpoint`, so only a channel with nothing stored here needs the
|
||||
walk. A channel with no version was never written, so it is empty without
|
||||
one; a walk for it would find no snapshot to stop at and read every
|
||||
ancestor, every time the thread is loaded.
|
||||
|
||||
A `BinaryOperatorAggregate` stores its value at every version, so one with
|
||||
nothing stored was written as a `DeltaChannel` before its annotation
|
||||
changed, and is rebuilt the same way.
|
||||
"""
|
||||
return [
|
||||
k
|
||||
for k, spec in specs.items()
|
||||
if isinstance(spec, DeltaChannel)
|
||||
if isinstance(spec, (DeltaChannel, BinaryOperatorAggregate))
|
||||
and k in checkpoint["channel_versions"]
|
||||
and checkpoint["channel_values"].get(k, MISSING) is MISSING
|
||||
]
|
||||
|
||||
|
||||
def _replay(spec: BaseChannel, history: DeltaChannelHistory) -> BaseChannel:
|
||||
ch = spec.from_checkpoint(history.get("seed", MISSING))
|
||||
if isinstance(ch, DeltaChannel):
|
||||
ch.replay_writes(history["writes"])
|
||||
else:
|
||||
# One update per write: the writes span many supersteps, and a single
|
||||
# call would reject a second Overwrite and drop the writes after one.
|
||||
for _, _, value in history["writes"]:
|
||||
ch.update([value])
|
||||
return ch
|
||||
|
||||
|
||||
def _require_saver_for_history(
|
||||
delta_channels: list[str],
|
||||
saver: BaseCheckpointSaver | None,
|
||||
@@ -465,7 +482,7 @@ def channels_from_checkpoint(
|
||||
|
||||
delta_channels = _delta_channels_to_replay(channel_specs, checkpoint)
|
||||
_require_saver_for_history(delta_channels, saver, config)
|
||||
histories: Mapping[str, Any] = {}
|
||||
histories: Mapping[str, DeltaChannelHistory] = {}
|
||||
if delta_channels and saver is not None and config is not None:
|
||||
histories = saver.get_delta_channel_history(
|
||||
config=config, channels=delta_channels
|
||||
@@ -475,11 +492,7 @@ def channels_from_checkpoint(
|
||||
for k, spec in channel_specs.items():
|
||||
ch: BaseChannel
|
||||
if k in histories:
|
||||
delta_spec = cast(DeltaChannel, spec)
|
||||
history = histories[k]
|
||||
replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING))
|
||||
replay_ch.replay_writes(history["writes"])
|
||||
ch = replay_ch
|
||||
ch = _replay(spec, histories[k])
|
||||
else:
|
||||
ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
|
||||
channels[k] = ch
|
||||
@@ -504,7 +517,7 @@ async def achannels_from_checkpoint(
|
||||
|
||||
delta_channels = _delta_channels_to_replay(channel_specs, checkpoint)
|
||||
_require_saver_for_history(delta_channels, saver, config)
|
||||
histories: Mapping[str, Any] = {}
|
||||
histories: Mapping[str, DeltaChannelHistory] = {}
|
||||
if delta_channels and saver is not None and config is not None:
|
||||
histories = await saver.aget_delta_channel_history(
|
||||
config=config, channels=delta_channels
|
||||
@@ -514,11 +527,7 @@ async def achannels_from_checkpoint(
|
||||
for k, spec in channel_specs.items():
|
||||
ch: BaseChannel
|
||||
if k in histories:
|
||||
delta_spec = cast(DeltaChannel, spec)
|
||||
history = histories[k]
|
||||
replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING))
|
||||
replay_ch.replay_writes(history["writes"])
|
||||
ch = replay_ch
|
||||
ch = _replay(spec, histories[k])
|
||||
else:
|
||||
ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
|
||||
channels[k] = ch
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
import operator
|
||||
from typing import Annotated, Any
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
from langgraph.checkpoint.base import BaseCheckpointSaver
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from langgraph.checkpoint.serde.types import _DeltaSnapshot
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from langgraph.channels.binop import BinaryOperatorAggregate
|
||||
from langgraph.channels.delta import DeltaChannel
|
||||
from langgraph.graph import END, START, StateGraph
|
||||
from langgraph.graph.message import _messages_delta_reducer, add_messages
|
||||
from langgraph.pregel._checkpoint import channels_from_checkpoint, empty_checkpoint
|
||||
from langgraph.types import Overwrite
|
||||
|
||||
pytestmark = pytest.mark.anyio
|
||||
|
||||
|
||||
class DeltaState(TypedDict):
|
||||
messages: Annotated[
|
||||
list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=3)
|
||||
]
|
||||
|
||||
|
||||
class PlainState(TypedDict):
|
||||
messages: Annotated[list, add_messages]
|
||||
|
||||
|
||||
def _app(schema: type, checkpointer: BaseCheckpointSaver) -> Any:
|
||||
builder = StateGraph(schema)
|
||||
builder.add_node(
|
||||
"chat",
|
||||
lambda state: {"messages": [AIMessage(f"reply {len(state['messages'])}")]},
|
||||
)
|
||||
builder.add_edge(START, "chat")
|
||||
builder.add_edge("chat", END)
|
||||
return builder.compile(checkpointer=checkpointer)
|
||||
|
||||
|
||||
def _delta_thread(checkpointer: BaseCheckpointSaver) -> RunnableConfig:
|
||||
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
|
||||
delta = _app(DeltaState, checkpointer)
|
||||
for i in range(4):
|
||||
delta.invoke({"messages": [HumanMessage(f"q{i}")]}, config)
|
||||
assert any(
|
||||
isinstance(t.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
|
||||
for t in checkpointer.list(config)
|
||||
), "the thread should have a snapshot to read through"
|
||||
return config
|
||||
|
||||
|
||||
def test_plain_reducer_reads_every_delta_written_checkpoint(
|
||||
sync_checkpointer: BaseCheckpointSaver,
|
||||
) -> None:
|
||||
config = _delta_thread(sync_checkpointer)
|
||||
delta = _app(DeltaState, sync_checkpointer)
|
||||
plain = _app(PlainState, sync_checkpointer)
|
||||
|
||||
assert [s.values for s in plain.get_state_history(config)] == [
|
||||
s.values for s in delta.get_state_history(config)
|
||||
]
|
||||
|
||||
|
||||
def test_plain_reducer_continues_delta_written_thread(
|
||||
sync_checkpointer: BaseCheckpointSaver,
|
||||
) -> None:
|
||||
config = _delta_thread(sync_checkpointer)
|
||||
delta = _app(DeltaState, sync_checkpointer)
|
||||
before = delta.get_state(config).values["messages"]
|
||||
|
||||
after = _app(PlainState, sync_checkpointer).invoke(
|
||||
{"messages": [HumanMessage("q4")]}, config
|
||||
)["messages"]
|
||||
|
||||
assert after[:-2] == before
|
||||
assert [m.content for m in after[-2:]] == ["q4", f"reply {len(before) + 1}"]
|
||||
head = sync_checkpointer.get_tuple(config)
|
||||
assert head is not None
|
||||
assert head.checkpoint["channel_values"]["messages"] == after
|
||||
assert delta.get_state(config).values["messages"] == after
|
||||
|
||||
|
||||
async def test_plain_reducer_reads_and_continues_delta_written_thread_async(
|
||||
async_checkpointer: BaseCheckpointSaver,
|
||||
) -> None:
|
||||
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
|
||||
delta = _app(DeltaState, async_checkpointer)
|
||||
for i in range(4):
|
||||
await delta.ainvoke({"messages": [HumanMessage(f"q{i}")]}, config)
|
||||
plain = _app(PlainState, async_checkpointer)
|
||||
|
||||
expected = [s.values async for s in delta.aget_state_history(config)]
|
||||
assert [s.values async for s in plain.aget_state_history(config)] == expected
|
||||
|
||||
after = (await plain.ainvoke({"messages": [HumanMessage("q4")]}, config))[
|
||||
"messages"
|
||||
]
|
||||
assert after[:-2] == expected[0]["messages"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("seed", "writes", "expected"),
|
||||
[
|
||||
(_DeltaSnapshot(["a"]), [["b"], ["c"]], ["a", "b", "c"]),
|
||||
(["a"], [["b"], Overwrite(["c"]), ["d"]], ["c", "d"]),
|
||||
(None, [["b"], Overwrite(["c"]), Overwrite(["d"]), ["e"]], ["d", "e"]),
|
||||
],
|
||||
)
|
||||
def test_plain_reducer_replays_delta_history_one_write_at_a_time(
|
||||
seed: Any, writes: list[Any], expected: list[str]
|
||||
) -> None:
|
||||
history: dict[str, Any] = {"writes": [("t", "items", w) for w in writes]}
|
||||
if seed is not None:
|
||||
history["seed"] = seed
|
||||
checkpoint = empty_checkpoint()
|
||||
checkpoint["channel_versions"]["items"] = 1
|
||||
saver = InMemorySaver()
|
||||
|
||||
with patch.object(
|
||||
saver, "get_delta_channel_history", return_value={"items": history}
|
||||
):
|
||||
channels, _ = channels_from_checkpoint(
|
||||
{"items": BinaryOperatorAggregate(list, operator.add)},
|
||||
checkpoint,
|
||||
saver=saver,
|
||||
config={"configurable": {"thread_id": "1"}},
|
||||
)
|
||||
|
||||
assert channels["items"].get() == expected
|
||||
|
||||
|
||||
def test_plain_thread_reads_never_walk_history() -> None:
|
||||
saver = InMemorySaver()
|
||||
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
|
||||
plain = _app(PlainState, saver)
|
||||
|
||||
with patch.object(saver, "get_delta_channel_history") as walk:
|
||||
plain.invoke({"messages": [HumanMessage("q0")]}, config)
|
||||
plain.update_state(config, {"messages": [HumanMessage("note")]})
|
||||
plain.invoke({"messages": [HumanMessage("q1")]}, config)
|
||||
plain.get_state(config)
|
||||
list(plain.get_state_history(config))
|
||||
|
||||
walk.assert_not_called()
|
||||
Generated
+1
-1
@@ -1685,7 +1685,7 @@ test = [
|
||||
|
||||
[[package]]
|
||||
name = "langgraph-checkpoint-postgres"
|
||||
version = "3.1.2"
|
||||
version = "3.1.3"
|
||||
source = { editable = "../checkpoint-postgres" }
|
||||
dependencies = [
|
||||
{ name = "langgraph-checkpoint" },
|
||||
|
||||
Generated
+1
-1
@@ -418,7 +418,7 @@ test = [
|
||||
|
||||
[[package]]
|
||||
name = "langgraph-checkpoint-postgres"
|
||||
version = "3.1.2"
|
||||
version = "3.1.3"
|
||||
source = { editable = "../checkpoint-postgres" }
|
||||
dependencies = [
|
||||
{ name = "langgraph-checkpoint" },
|
||||
|
||||
Reference in New Issue
Block a user