fix(cli): deploy push reads AMP's source_type, so a ZIP deployment gets the new code

#7345 made push follow how AMP holds a deployment's code, read from a
`zip_deployment` boolean on the status. AMP's status never sends that field:
it sends `source_type` ("zip", "studio", "github", "any_git"). So push always
fell back to the local origin remote, and a ZIP deployment whose project has a
git remote was redeployed by uuid — AMP rebuilt the ZIP it already held, not
the code being pushed.

Seen on a real deployment: after raising crewai to 1.15.26, `crewai deploy push`
rebuilt the stored 1.15.23 ZIP ("CrewAI version detected: 1.15.23"), with no
upload. `zip_deployment` is still read first when a server sends it.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

lib/cli/src/crewai_cli/deploy/main.py

@@ -83,12 +83,28 @@ def _display_git_remote_help() -> None:
     )
 
 
+# AMP's `source_type` for a deployment's code, and whether a push updates it by
+# uploading a ZIP. A Studio install is an archive too (AMP sets it as a ZIP crew).
+_ZIP_SOURCE_TYPES = {"zip": True, "studio": True, "github": False, "any_git": False}
+
+
 def _zip_deployment_flag(status: dict[str, Any] | None) -> bool | None:
-    """Return the AMP zip_deployment flag, or None when it cannot be used."""
-    if not status or "zip_deployment" not in status:
+    """Whether AMP holds this deployment's code as a ZIP, or None when the status
+    does not say.
+
+    AMP's status sends `source_type` ("zip", "studio", "github", "any_git");
+    `zip_deployment`, a boolean, is read first when present. Without either the
+    caller falls back to the local origin remote — which is wrong for a ZIP
+    deployment whose project has since gained a remote: it redeploys the ZIP AMP
+    already holds instead of the code being pushed.
+    """
+    if not status:
         return None
-    value = status["zip_deployment"]
-    return value if isinstance(value, bool) else None
+    value = status.get("zip_deployment")
+    if isinstance(value, bool):
+        return value
+    source = status.get("source_type")
+    return _ZIP_SOURCE_TYPES.get(source) if isinstance(source, str) else None
 
 
 def _env_summary(env_vars: dict[str, str]) -> str:
@@ -415,7 +431,7 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
         project_name: str | None,
         status: dict[str, Any] | None,
     ) -> bool:
-        """Return True when AMP reported a usable zip_deployment flag."""
+        """Return True when AMP's status says how it holds the code."""
         zip_deployment = _zip_deployment_flag(status)
         if zip_deployment is None:
             return False
@@ -430,7 +446,7 @@ class DeployCommand(BaseCommand, PlusAPIMixin):
         repository: git.Repository | None,
         status: dict[str, Any],
     ) -> Any:
-        """Deploy using AMP zip_deployment."""
+        """Deploy the way AMP holds the code: a ZIP upload, or a redeploy of its git source."""
         if _zip_deployment_flag(status):
             deployment_uuid = uuid or str(status["uuid"])
             env_vars = fetch_and_json_env_file()

lib/cli/tests/deploy/test_deploy_main.py

@@ -229,6 +229,7 @@ class TestDeployCommand(unittest.TestCase):
         *,
         uuid: str = "test-uuid",
         zip_deployment: bool | None = None,
+        source_type: str | None = None,
         is_success: bool = True,
     ) -> MagicMock:
         response = MagicMock()
@@ -236,6 +237,8 @@ class TestDeployCommand(unittest.TestCase):
         payload: dict = {"uuid": uuid}
         if zip_deployment is not None:
             payload["zip_deployment"] = zip_deployment
+        if source_type is not None:
+            payload["source_type"] = source_type
         response.json.return_value = payload
         return response
 
@@ -627,6 +630,78 @@ class TestDeployCommand(unittest.TestCase):
         self.mock_client.deploy_by_uuid.assert_not_called()
         mock_display.assert_called_once_with({"uuid": "test-uuid"})
 
+    @patch("crewai_cli.deploy.main.create_project_zip")
+    @patch("crewai_cli.deploy.main.fetch_and_json_env_file")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
+    def test_a_zip_source_type_uploads_the_code_even_when_origin_exists(
+        self, mock_display, mock_repository, mock_fetch_env, mock_create_project_zip
+    ):
+        # What AMP's status actually sends: `source_type`, no `zip_deployment`.
+        # A redeploy by uuid here rebuilds the ZIP AMP already holds, not the code.
+        mock_fetch_env.return_value = {"ENV_VAR": "value"}
+        mock_repository.return_value.origin_url.return_value = (
+            "https://github.com/test/repo.git"
+        )
+        mock_repository.return_value.create_initial_commit_if_needed.return_value = (
+            False
+        )
+        mock_create_project_zip.return_value = Path("/tmp/test_project.zip")
+        for source_type in ("zip", "studio"):
+            with self.subTest(source_type=source_type):
+                self.mock_client.reset_mock()
+                self.mock_client.crew_status_by_uuid.return_value = (
+                    self._status_response(source_type=source_type)
+                )
+                mock_response = MagicMock()
+                mock_response.status_code = 200
+                mock_response.json.return_value = {"uuid": "test-uuid"}
+                self.mock_client.update_crew_from_zip.return_value = mock_response
+
+                self.deploy_command.deploy(uuid="test-uuid", skip_validate=True)
+
+                self.mock_client.update_crew_from_zip.assert_called_once_with(
+                    "test-uuid",
+                    Path("/tmp/test_project.zip"),
+                    env={"ENV_VAR": "value"},
+                )
+                self.mock_client.deploy_by_uuid.assert_not_called()
+
+    @patch("crewai_cli.deploy.main.create_project_zip")
+    @patch("crewai_cli.deploy.main.git.Repository")
+    @patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
+    def test_a_git_source_type_redeploys_from_git_even_without_origin(
+        self, mock_display, mock_repository, mock_create_project_zip
+    ):
+        mock_repository.return_value.origin_url.return_value = None
+        mock_repository.return_value.create_initial_commit_if_needed.return_value = (
+            False
+        )
+        for source_type in ("github", "any_git"):
+            with self.subTest(source_type=source_type):
+                self.mock_client.reset_mock()
+                self.mock_client.crew_status_by_uuid.return_value = (
+                    self._status_response(source_type=source_type)
+                )
+                mock_response = MagicMock()
+                mock_response.status_code = 200
+                mock_response.json.return_value = {"uuid": "test-uuid"}
+                self.mock_client.deploy_by_uuid.return_value = mock_response
+
+                self.deploy_command.deploy(uuid="test-uuid", skip_validate=True)
+
+                self.mock_client.deploy_by_uuid.assert_called_once_with("test-uuid")
+                self.mock_client.update_crew_from_zip.assert_not_called()
+        mock_create_project_zip.assert_not_called()
+
+    def test_zip_deployment_is_read_before_source_type(self):
+        from crewai_cli.deploy.main import _zip_deployment_flag
+
+        self.assertFalse(_zip_deployment_flag({"zip_deployment": False, "source_type": "zip"}))
+        self.assertTrue(_zip_deployment_flag({"source_type": "zip"}))
+        self.assertIsNone(_zip_deployment_flag({"source_type": "something-new"}))
+        self.assertIsNone(_zip_deployment_flag({"uuid": "x"}))
+
     @patch("crewai_cli.deploy.main.git.Repository")
     @patch("crewai_cli.deploy.main.DeployCommand._display_deployment_info")
     def test_deploy_falls_back_to_origin_when_zip_deployment_is_missing(