fix(studio): keep one minimum instance during deploy and update

tests/cli/test_frontend_deploy_iam.py

@@ -288,6 +288,7 @@ def test_frontend_policy_allows_release_download() -> None:
     assert "iam:UpdatePolicy" not in actions
     assert "vefaas:CodeUploadCallback" in actions
     assert "vefaas:UpdateFunction" in actions
+    assert "vefaas:UpdateFunctionResource" in actions
     assert "vefaas:ReleaseApplication" in actions
     assert "tag:TagResources" in actions
     assert "tag:UntagResources" in actions

tests/cli/test_studio_deploy_permissions.py

@@ -38,7 +38,7 @@ def test_default_studio_deploy_requires_all_reachable_actions() -> None:
     specs = _default_specs()
     actions = [spec.action for spec in specs]
 
-    assert len(actions) == 44
+    assert len(actions) == 45
     assert len(actions) == len(set(actions))
     assert "id:CreateUserPool" in actions
     assert "iam:UpdatePolicy" in actions
@@ -47,6 +47,7 @@ def test_default_studio_deploy_requires_all_reachable_actions() -> None:
     assert "apig:CreateGateway" in actions
     assert "apig:UpdateRoute" in actions
     assert "vefaas:CreateApplication" in actions
+    assert "vefaas:UpdateFunctionResource" in actions
     assert "vefaas:CreateTimer" in actions
     assert "vefaas:ListTriggers" in actions
     assert "vefaas:UpdateTimer" in actions

tests/cli/test_studio_self_update.py

@@ -516,7 +516,7 @@ class _VeFaaS:
         def __init__(self, **_kwargs: str) -> None:
             self.client = object()
 
-        def submit_application_code_bundle_update(self, **kwargs: Any) -> None:
+        def submit_application_code_bundle_update(self, **kwargs: Any) -> bool:
             package = Path(kwargs["path"])
             captured["full_selected"] = (package / "full-marker").read_text()
 

tests/cli/test_studio_update.py

@@ -1435,6 +1435,7 @@ def test_update_application_code_bundle_merges_only_explicit_environment(
     tmp_path: Path,
 ) -> None:
     updated_requests: list[Any] = []
+    resource_requests: list[Any] = []
     service = object.__new__(VeFaaS)
     service.session_token = ""
     cast(Any, service).client = SimpleNamespace(
@@ -1445,6 +1446,7 @@ def test_update_application_code_bundle_merges_only_explicit_environment(
             ]
         ),
         update_function=updated_requests.append,
+        update_function_resource=resource_requests.append,
     )
     monkeypatch.setattr(service, "_upload_and_mount_code", lambda *_: None)
     monkeypatch.setattr(service, "_release_application", lambda _: "https://same")
@@ -1466,9 +1468,98 @@ def test_update_application_code_bundle_merges_only_explicit_environment(
         "EXISTING": "kept",
         "VEADK_SITE_TITLE": "新标题",
     }
+    resource_request = resource_requests[0]
+    assert resource_request.function_id == "function-id"
+    assert resource_request.min_instance == 1
+    assert resource_request.max_instance is None
+    assert resource_request.reserved_frozen_instance is None
     ensure_route.assert_called_once_with("app-id", disable_cors=True)
 
 
+def test_submit_application_code_bundle_sets_only_minimum_instance(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    updated_requests: list[Any] = []
+    resource_requests: list[Any] = []
+    releases: list[str] = []
+    service = object.__new__(VeFaaS)
+    cast(Any, service).client = SimpleNamespace(
+        update_function=updated_requests.append,
+        update_function_resource=resource_requests.append,
+    )
+    monkeypatch.setattr(service, "_upload_and_mount_code", lambda *_: None)
+    monkeypatch.setattr(service, "_start_application_release", releases.append)
+
+    service.submit_application_code_bundle_update(
+        application_id="app-id",
+        function_id="function-id",
+        path=str(tmp_path),
+    )
+
+    assert updated_requests[0].id == "function-id"
+    resource_request = resource_requests[0]
+    assert resource_request.function_id == "function-id"
+    assert resource_request.min_instance == 1
+    assert resource_request.max_instance is None
+    assert resource_request.reserved_frozen_instance is None
+    assert releases == ["app-id"]
+
+
+def test_submit_application_code_bundle_stops_without_resource_permission(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    releases: list[str] = []
+    service = object.__new__(VeFaaS)
+
+    def _permission_denied(_: Any) -> None:
+        raise RuntimeError("AccessDenied: vefaas:UpdateFunctionResource")
+
+    cast(Any, service).client = SimpleNamespace(
+        update_function=lambda _: None,
+        update_function_resource=_permission_denied,
+    )
+    monkeypatch.setattr(service, "_upload_and_mount_code", lambda *_: None)
+    monkeypatch.setattr(service, "_start_application_release", releases.append)
+
+    with pytest.raises(RuntimeError, match="UpdateFunctionResource"):
+        service.submit_application_code_bundle_update(
+            application_id="app-id",
+            function_id="function-id",
+            path=str(tmp_path),
+        )
+
+    assert releases == []
+
+
+def test_submit_application_code_bundle_stops_on_other_resource_errors(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    releases: list[str] = []
+    service = object.__new__(VeFaaS)
+
+    def _invalid_operation(_: Any) -> None:
+        raise RuntimeError("InvalidOperation")
+
+    cast(Any, service).client = SimpleNamespace(
+        update_function=lambda _: None,
+        update_function_resource=_invalid_operation,
+    )
+    monkeypatch.setattr(service, "_upload_and_mount_code", lambda *_: None)
+    monkeypatch.setattr(service, "_start_application_release", releases.append)
+
+    with pytest.raises(RuntimeError, match="InvalidOperation"):
+        service.submit_application_code_bundle_update(
+            application_id="app-id",
+            function_id="function-id",
+            path=str(tmp_path),
+        )
+
+    assert releases == []
+
+
 def test_application_operations_use_deployment_region(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -1749,6 +1840,7 @@ def test_update_application_code_bundle_preserves_unspecified_sandbox_tool(
             ]
         ),
         update_function=updated_requests.append,
+        update_function_resource=lambda _: None,
     )
     monkeypatch.setattr(service, "_upload_and_mount_code", lambda *_: None)
     monkeypatch.setattr(service, "_release_application", lambda _: "https://same")
@@ -1777,6 +1869,7 @@ def test_update_application_code_bundle_does_not_read_or_replace_environment(
     cast(Any, service).client = SimpleNamespace(
         get_function=lambda _: pytest.fail("environment should not be read"),
         update_function=updated_requests.append,
+        update_function_resource=lambda _: None,
     )
     monkeypatch.setattr(service, "_upload_and_mount_code", lambda *_: None)
     monkeypatch.setattr(service, "_release_application", lambda _: "https://same")

tests/cli/test_studio_update_permissions.py

@@ -182,6 +182,40 @@ def test_default_frontend_policy_satisfies_every_ota_permission() -> None:
     assert all(evaluated.values())
 
 
+def test_precheck_adds_function_resource_permission_to_authorization() -> None:
+    current = deepcopy(FRONTEND_DEPLOY_POLICY)
+    actions = current["Statement"][0]["Action"]
+    current["Statement"][0]["Action"] = [
+        action for action in actions if action != "vefaas:UpdateFunctionResource"
+    ]
+    principal = PrincipalPolicySet(
+        kind="role",
+        name="CustomerRole",
+        policies=(
+            AttachedPolicyDocument(
+                name="CustomerPolicy",
+                policy_type="Custom",
+                document=current,
+            ),
+        ),
+    )
+
+    report = StudioUpdatePermissionService(
+        provider="volcengine",
+        access_key="ak",
+        secret_key="sk",
+        inspector=_Inspector(principal),
+    ).check()
+
+    assert report.missing_actions == ("vefaas:UpdateFunctionResource",)
+    query = parse_qs(urlparse(report.authorization_url).query)
+    prefilled = json.loads(query["query"][0])
+    merged = json.loads(prefilled["NewPolicyDocument"])
+    assert evaluate_actions(["vefaas:UpdateFunctionResource"], [merged])[
+        "vefaas:UpdateFunctionResource"
+    ]
+
+
 def test_merge_policy_actions_preserves_unrelated_statements() -> None:
     original = {
         "Statement": [

tests/test_cloud.py

@@ -14,6 +14,7 @@
 
 import os
 import tempfile
+from pathlib import Path
 from types import SimpleNamespace
 from typing import Any, cast
 from unittest.mock import AsyncMock, Mock, patch
@@ -67,10 +68,28 @@ def test_vefaas_create_function_uses_configured_project() -> None:
     assert request.project_name == "studio-project"
 
 
+@pytest.mark.parametrize("provider", ["volcengine", "byteplus"])
+def test_vefaas_sets_only_minimum_instance(provider: str) -> None:
+    requests = []
+    service = object.__new__(VeFaaS)
+    service.provider = provider
+    service.client = SimpleNamespace(update_function_resource=requests.append)
+
+    service._set_function_min_instance("function-id")
+
+    assert len(requests) == 1
+    request = requests[0]
+    assert request.function_id == "function-id"
+    assert request.min_instance == 1
+    assert request.max_instance is None
+    assert request.reserved_frozen_instance is None
+
+
 def test_vefaas_deploy_cleans_created_resources_on_release_failure() -> None:
     service = object.__new__(VeFaaS)
     service.find_app_id_by_name = Mock(return_value=None)
     service._create_function = Mock(return_value=("studio-app-fn", "function-id"))
+    service._set_function_min_instance = Mock()
     service._create_application = Mock(return_value="application-id")
     service._release_application = Mock(side_effect=RuntimeError("release failed"))
     service.delete = Mock()
@@ -87,12 +106,14 @@ def test_vefaas_deploy_cleans_created_resources_on_release_failure() -> None:
 
     service.delete.assert_called_once_with("application-id")
     service.delete_function.assert_called_once_with("function-id")
+    service._set_function_min_instance.assert_not_called()
 
 
 def test_vefaas_deploy_can_keep_failed_resources_for_inspection() -> None:
     service = object.__new__(VeFaaS)
     service.find_app_id_by_name = Mock(return_value=None)
     service._create_function = Mock(return_value=("studio-app-fn", "function-id"))
+    service._set_function_min_instance = Mock()
     service._create_application = Mock(return_value="application-id")
     service._release_application = Mock(side_effect=RuntimeError("release failed"))
     service.delete = Mock()
@@ -110,6 +131,60 @@ def test_vefaas_deploy_can_keep_failed_resources_for_inspection() -> None:
 
     service.delete.assert_not_called()
     service.delete_function.assert_not_called()
+    service._set_function_min_instance.assert_not_called()
+
+
+def test_vefaas_update_code_uses_resource_updating_bundle_path(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    project = tmp_path / "agent"
+    project.mkdir()
+    service = object.__new__(VeFaaS)
+    service.find_app_id_by_name = Mock(return_value="application-id")
+    service._get_application_status = Mock(
+        return_value=(
+            "deploy_success",
+            {
+                "Result": {
+                    "CloudResource": (
+                        '{"framework":{"function":{"Name":"studio-app-fn",'
+                        '"Id":"function-id"}}}'
+                    )
+                }
+            },
+        )
+    )
+    service._replace_application_code_bundle = Mock()
+    service._release_application = Mock(return_value="https://studio.example")
+    service._set_function_min_instance = Mock()
+    service.ensure_application_route_methods = Mock()
+
+    def _cookiecutter(*, output_dir: str, extra_context: dict, **_: object) -> None:
+        package = Path(output_dir) / str(extra_context["local_dir_name"])
+        (package / "src").mkdir(parents=True)
+
+    monkeypatch.setattr(
+        "veadk.integrations.ve_faas.ve_faas.formatted_timestamp", lambda: "stamp"
+    )
+    monkeypatch.setattr(
+        "veadk.integrations.ve_faas.ve_faas.tempfile.gettempdir",
+        lambda: str(tmp_path),
+    )
+    monkeypatch.setattr(
+        "veadk.integrations.ve_faas.ve_faas.cookiecutter", _cookiecutter
+    )
+
+    result = service._update_function_code("studio-app", str(project))
+
+    assert result == ("https://studio.example", "application-id", "function-id")
+    service._replace_application_code_bundle.assert_called_once_with(
+        function_id="function-id",
+        path=str(tmp_path / "agent_update_stamp" / "src"),
+        environment_overrides=None,
+        request_timeout=1800,
+    )
+    service._set_function_min_instance.assert_called_once_with("function-id")
 
 
 def test_vefaas_deploy_updates_existing_application_in_place() -> None:
@@ -297,6 +372,7 @@ def test_vefaas_deploy_can_disable_gateway_cors() -> None:
     service._create_function = Mock(return_value=("studio-app-fn", "function-id"))
     service._create_application = Mock(return_value="application-id")
     service._release_application = Mock(return_value="https://studio.example.com")
+    service._set_function_min_instance = Mock()
     service.ensure_application_route_methods = Mock()
 
     service.deploy(
@@ -312,6 +388,7 @@ def test_vefaas_deploy_can_disable_gateway_cors() -> None:
         "application-id",
         disable_cors=True,
     )
+    service._set_function_min_instance.assert_called_once_with("function-id")
 
 
 def test_vefaas_code_upload_callback_uses_configured_region() -> None:

veadk/cli/frontend_deploy_policy.py

@@ -207,6 +207,7 @@
                 "vefaas:SetSandboxTimeout",
                 "vefaas:UpdateTimer",
                 "vefaas:UpdateFunction",
+                "vefaas:UpdateFunctionResource",
                 "vikingdb:GetKnowledgeBaseServiceInfo",
                 "vikingdb:GetMemorydbInstanceDetail",
                 "vikingdb:ListCollections",

veadk/cli/studio_deploy_permissions.py

@@ -228,6 +228,11 @@ def _permission(action: str, purpose_zh: str, purpose_en: str) -> PermissionSpec
         "更新 Studio 函数环境变量",
         "Update Studio function environment variables",
     ),
+    _permission(
+        "vefaas:UpdateFunctionResource",
+        "设置 Studio 函数最小实例数",
+        "Set the Studio function minimum instance count",
+    ),
     _permission(
         "vefaas:Release", "重新发布 Studio 函数", "Release the updated Studio function"
     ),

veadk/cli/studio_update_permissions.py

@@ -80,6 +80,11 @@ def _permission(action: str, purpose_zh: str, purpose_en: str) -> PermissionSpec
         "更新 Studio 与调度函数",
         "Update Studio and scheduler Functions",
     ),
+    _permission(
+        "vefaas:UpdateFunctionResource",
+        "设置 Studio 函数最小实例数",
+        "Set the Studio function minimum instance count",
+    ),
     _permission(
         "vefaas:ReleaseApplication",
         "发布 Studio 应用新 Revision",