fix(cli): point a managed build at --push-to in a listener workspace

libs/cli/langgraph_cli/deploy.py

@@ -1430,18 +1430,31 @@ def _resolve_or_create(
     )
     if found is not None:
         return found.id, step
-    created, step = _create_deployment(
-        ctx.client,
-        step,
-        name=ctx.selector.name,
-        source=source,
-        source_config={"deployment_type": ctx.deployment_type},
-        source_revision_config={},
-        secrets=ctx.secrets,
-    )
+    try:
+        created, step = _create_deployment(
+            ctx.client,
+            step,
+            name=ctx.selector.name,
+            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 click.UsageError(
+                "This workspace deploys through a listener in your own cluster, so "
+                "the image has to come from a registry you manage. Re-run with "
+                "--push-to <registry>/<repository>."
+            ) from None
+        raise
     return created.id, step
 
 
+def _needs_a_listener(err: HostBackendError) -> bool:
+    return err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message
+
+
 def _available_listeners(client: HostBackendClient) -> tuple[Listener, ...]:
     resources = _call_host_backend_with_optional_tenant(
         client, lambda c: c.list_listeners()
@@ -1573,7 +1586,7 @@ class CustomerRegistrySource:
                 secrets=ctx.secrets,
             )
         except HostBackendError as err:
-            if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message:
+            if _needs_a_listener(err):
                 raise click.UsageError(
                     "This workspace deploys through a listener. Re-run with "
                     f"--listener-id and --k8s-namespace.\n{err.message}"

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

@@ -932,3 +932,18 @@ def test_a_listener_without_an_id_is_ignored(deploy_project: DeployProject) -> N
         "listener_id": "listener-1",
         "listener_config": {"k8s_namespace": "agents"},
     }
+
+
+def test_a_managed_build_in_a_listener_workspace_points_at_push_to(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.control_plane.create_error = (
+        "Source configuration error: 'source_config.listener_id' is required "
+        "for workspace with available listener IDs: ['listener-1']"
+    )
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code != 0
+    assert "--push-to" in result.output
+    assert deploy_project.docker.verbs() == []