diff --git a/agentkit/toolkit/builders/ve_pipeline.py b/agentkit/toolkit/builders/ve_pipeline.py index eaf9aa1f..78b021a3 100644 --- a/agentkit/toolkit/builders/ve_pipeline.py +++ b/agentkit/toolkit/builders/ve_pipeline.py @@ -1240,9 +1240,17 @@ def _prepare_pipeline_resources( ) cp_client = VeCodePipeline(region=config.cp_region, provider=provider) - # Get or create agentkit-cli-workspace - workspace_name = "agentkit-cli-workspace" - if not cp_client.workspace_exists_by_name(workspace_name): + # Get or create the configured workspace. + workspace_name = config.cp_workspace_name or DEFAULT_WORKSPACE_NAME + workspace_result = cp_client.get_workspaces_by_name( + workspace_name, page_size=100 + ) + matching_workspaces = [ + item + for item in workspace_result.get("Items", []) + if item.get("Name") == workspace_name + ] + if not matching_workspaces: logger.info(f"Workspace '{workspace_name}' does not exist, creating...") self.reporter.warning( f"Workspace '{workspace_name}' does not exist, creating..." @@ -1257,16 +1265,11 @@ def _prepare_pipeline_resources( f"Workspace created successfully: {workspace_name}" ) else: - # Workspace exists, get its ID - result = cp_client.get_workspaces_by_name(workspace_name, page_size=1) - if result.get("Items") and len(result["Items"]) > 0: - workspace_id = result["Items"][0]["Id"] - logger.info( - f"Using existing workspace: {workspace_name} (ID: {workspace_id})" - ) - self.reporter.success(f"Using workspace: {workspace_name}") - else: - raise Exception(f"Unable to get workspace '{workspace_name}' ID") + workspace_id = matching_workspaces[0]["Id"] + logger.info( + f"Using existing workspace: {workspace_name} (ID: {workspace_id})" + ) + self.reporter.success(f"Using workspace: {workspace_name}") logger.info(f"Using workspace: {workspace_name} (ID: {workspace_id})") @@ -1276,8 +1279,6 @@ def _prepare_pipeline_resources( # Check if pipeline already exists - try multiple lookup strategies # Case 1: If Pipeline ID is configured, use ID for exact lookup - # tmp: temp fix for pipeline id issue, cp_pipeline_id should be empty string for fix cp name - config.cp_pipeline_id = "" if config.cp_pipeline_id and config.cp_pipeline_id != AUTO_CREATE_VE: try: # Get pipeline details by ID @@ -1287,6 +1288,11 @@ def _prepare_pipeline_resources( if result.get("Items") and len(result["Items"]) > 0: pipeline_info = result["Items"][0] + if not cp_client.is_agentkit_build_pipeline(pipeline_info): + raise Exception( + f"Pipeline ID '{config.cp_pipeline_id}' is not compatible " + "with AgentKit cloud builds" + ) found_pipeline_name = pipeline_info.get("Name", "") # If name is also configured, validate name-ID consistency @@ -1331,8 +1337,8 @@ def _prepare_pipeline_resources( ) except Exception as e: - if "does not match" in str(e): - raise # Name-ID mismatch, propagate exception + if "does not match" in str(e) or "not compatible" in str(e): + raise logger.warning( f"Pipeline lookup by ID failed: {str(e)}, will create new pipeline" ) @@ -1344,14 +1350,21 @@ def _prepare_pipeline_resources( workspace_id=workspace_id, name_filter=config.cp_pipeline_name ) - if ( - existing_pipelines.get("Items") - and len(existing_pipelines["Items"]) > 0 - ): + matching_pipelines = [ + item + for item in existing_pipelines.get("Items", []) + if item.get("Name") == config.cp_pipeline_name + ] + if matching_pipelines: # Found existing pipeline - pipeline_info = existing_pipelines["Items"][0] + pipeline_info = matching_pipelines[0] pipeline_id = pipeline_info["Id"] found_name = pipeline_info.get("Name", "") + if not cp_client.is_agentkit_build_pipeline(pipeline_info): + raise Exception( + f"Pipeline '{found_name}' is not compatible with " + "AgentKit cloud builds" + ) logger.info( f"Reusing pipeline by name: {found_name} (ID: {pipeline_id})" @@ -1380,6 +1393,8 @@ def _prepare_pipeline_resources( "Configured pipeline name does not exist, will create new pipeline" ) except Exception as e: + if "not compatible" in str(e): + raise logger.warning( f"Pipeline lookup by name failed: {str(e)}, will create new pipeline" ) diff --git a/agentkit/toolkit/strategies/cloud_strategy.py b/agentkit/toolkit/strategies/cloud_strategy.py index 27940c3d..65bf987b 100644 --- a/agentkit/toolkit/strategies/cloud_strategy.py +++ b/agentkit/toolkit/strategies/cloud_strategy.py @@ -120,6 +120,8 @@ def build( runtime_name, cp_pipeline_name = self._prepare_runtime_name( strategy_config.runtime_name, common_config.agent_name ) + if strategy_config.cp_pipeline_name not in ("", AUTO_CREATE_VE): + cp_pipeline_name = strategy_config.cp_pipeline_name # Track generated names if they differ from config if runtime_name != strategy_config.runtime_name: diff --git a/agentkit/toolkit/volcengine/code_pipeline.py b/agentkit/toolkit/volcengine/code_pipeline.py index 3fa49d68..64c55748 100644 --- a/agentkit/toolkit/volcengine/code_pipeline.py +++ b/agentkit/toolkit/volcengine/code_pipeline.py @@ -25,6 +25,24 @@ class VeCodePipeline: + AGENTKIT_BUILD_PARAMETER_KEYS = frozenset( + { + "DOCKERFILE_PATH", + "DOWNLOAD_PATH", + "PROJECT_ROOT_DIR", + "TOS_BUCKET_NAME", + "TOS_REGION", + "TOS_PROJECT_FILE_NAME", + "TOS_PROJECT_FILE_PATH", + "CR_NAMESPACE", + "CR_INSTANCE", + "CR_DOMAIN", + "CR_OCI", + "CR_TAG", + "CR_REGION", + } + ) + def __init__( self, access_key: str = "", @@ -316,6 +334,26 @@ def _create_pipeline( except KeyError: raise Exception(f"Create pipeline failed: {res}") + @classmethod + def is_agentkit_build_pipeline(cls, pipeline: dict) -> bool: + """Return whether a ListPipelines item matches AgentKit's build template.""" + parameters = pipeline.get("Parameters") or [] + keys = { + str(parameter.get("Key") or "") + for parameter in parameters + if isinstance(parameter, dict) + } + if cls.AGENTKIT_BUILD_PARAMETER_KEYS.issubset(keys): + return True + + spec = pipeline.get("Spec") or "" + if not isinstance(spec, str): + return False + return all( + f"parameters.{key}" in spec or f"${key}" in spec + for key in cls.AGENTKIT_BUILD_PARAMETER_KEYS + ) + def run_pipeline( self, workspace_id: str, diff --git a/agentkit/toolkit/volcengine/cr.py b/agentkit/toolkit/volcengine/cr.py index 70ebbb71..cd086866 100644 --- a/agentkit/toolkit/volcengine/cr.py +++ b/agentkit/toolkit/volcengine/cr.py @@ -155,6 +155,59 @@ def _check_instance(self, instance_name: str) -> str: except Exception as _: raise ValueError(f"Error check cr instance {instance_name}: {response}") + def list_registries(self, page_number: int = 1, page_size: int = 100) -> dict: + """List Container Registry instances.""" + response = self._ve_request( + request_body={"PageNumber": page_number, "PageSize": page_size}, + action="ListRegistries", + ) + try: + return response["Result"] + except KeyError as error: + raise ValueError(f"List CR registries failed: {response}") from error + + def list_namespaces( + self, + registry: str, + page_number: int = 1, + page_size: int = 100, + ) -> dict: + """List namespaces in a Container Registry instance.""" + response = self._ve_request( + request_body={ + "Registry": registry, + "PageNumber": page_number, + "PageSize": page_size, + }, + action="ListNamespaces", + ) + try: + return response["Result"] + except KeyError as error: + raise ValueError(f"List CR namespaces failed: {response}") from error + + def list_repositories( + self, + registry: str, + namespace: str, + page_number: int = 1, + page_size: int = 100, + ) -> dict: + """List repositories under a Container Registry namespace.""" + response = self._ve_request( + request_body={ + "Registry": registry, + "Namespace": namespace, + "PageNumber": page_number, + "PageSize": page_size, + }, + action="ListRepositories", + ) + try: + return response["Result"] + except KeyError as error: + raise ValueError(f"List CR repositories failed: {response}") from error + def _create_namespace( self, instance_name: str = DEFAULT_CR_INSTANCE_NAME, diff --git a/agentkit/toolkit/volcengine/services/tos_service.py b/agentkit/toolkit/volcengine/services/tos_service.py index b644ecf5..3d3cc5f1 100644 --- a/agentkit/toolkit/volcengine/services/tos_service.py +++ b/agentkit/toolkit/volcengine/services/tos_service.py @@ -16,9 +16,10 @@ import logging from dataclasses import dataclass, field from typing import List, Optional -from agentkit.utils.misc import generate_random_id +from agentkit.platform import Credentials from agentkit.toolkit.config.dataclass_utils import AutoSerializableMixin from agentkit.toolkit.config.constants import DEFAULT_TOS_BUCKET_TEMPLATE_NAME +from agentkit.utils.misc import generate_random_id try: import tos @@ -76,7 +77,12 @@ class TOSMountConfig: class TOSService: """Wrapper for Volcano Engine TOS (Object Storage) service.""" - def __init__(self, config: TOSServiceConfig, provider: Optional[str] = None): + def __init__( + self, + config: TOSServiceConfig, + provider: Optional[str] = None, + credentials: Optional[Credentials] = None, + ): """Initialize TOS service with configuration. Args: @@ -90,6 +96,7 @@ def __init__(self, config: TOSServiceConfig, provider: Optional[str] = None): self.config = config self.provider = provider + self.explicit_credentials = credentials self.client = None self.credentials = None self._init_client() @@ -101,7 +108,14 @@ def _init_client(self) -> None: # Use configured region if available region = self.config.region.strip() if self.config.region else None - config = VolcConfiguration(region=region, provider=self.provider) + credentials = self.explicit_credentials + config = VolcConfiguration( + region=region, + provider=self.provider, + access_key=credentials.access_key if credentials else None, + secret_key=credentials.secret_key if credentials else None, + session_token=credentials.session_token if credentials else None, + ) creds = config.get_service_credentials("tos") ep = config.get_service_endpoint("tos") @@ -346,15 +360,31 @@ def list_bucket_names(self) -> List[str]: Returns: List[str]: Bucket names under the current account. """ + return [bucket["Name"] for bucket in self.list_buckets()] + + def list_buckets(self) -> List[dict]: + """List buckets owned by the current credentials. + + Returns normalized names and locations so callers can avoid selecting a + bucket from a different physical TOS region. + """ try: out = self.client.list_buckets() buckets = getattr(out, "buckets", None) or [] - names: List[str] = [] - for b in buckets: - name = getattr(b, "name", None) - if name: - names.append(name) - return names + result: List[dict] = [] + for bucket in buckets: + name = getattr(bucket, "name", None) + if not name: + continue + result.append( + { + "Name": name, + "Location": getattr(bucket, "location", None) + or getattr(self, "actual_region", ""), + "CreationDate": getattr(bucket, "creation_date", None) or "", + } + ) + return result except Exception as e: logger.error(f"Failed to list buckets: {str(e)}") raise diff --git a/tests/toolkit/builders/test_ve_pipeline_resource_selection.py b/tests/toolkit/builders/test_ve_pipeline_resource_selection.py new file mode 100644 index 00000000..9e673984 --- /dev/null +++ b/tests/toolkit/builders/test_ve_pipeline_resource_selection.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import itertools + +from agentkit.toolkit.builders.ve_pipeline import VeCPCRBuilder, VeCPCRBuilderConfig +from agentkit.toolkit.config import CommonConfig +from agentkit.toolkit.config.constants import AUTO_CREATE_VE, DEFAULT_WORKSPACE_NAME +from agentkit.toolkit.config.strategy_configs import CloudStrategyConfig +from agentkit.toolkit.models import BuildResult +from agentkit.toolkit.strategies.cloud_strategy import CloudStrategy + + +class _Reporter: + def info(self, *_args, **_kwargs): + pass + + def success(self, *_args, **_kwargs): + pass + + def warning(self, *_args, **_kwargs): + pass + + def error(self, *_args, **_kwargs): + pass + + +def test_prepare_pipeline_uses_configured_workspace_and_existing_pipeline( + monkeypatch, tmp_path +) -> None: + class _FakeCodePipeline: + AGENTKIT_BUILD_PARAMETER_KEYS = frozenset({"TOS_BUCKET_NAME"}) + + def __init__(self, **_kwargs): + self.created = False + + def workspace_exists_by_name(self, name): + assert name == "custom-workspace" + return True + + def get_workspaces_by_name(self, name, page_size=10): + assert name == "custom-workspace" + return { + "Items": [{"Id": "workspace-id", "Name": "custom-workspace"}], + "TotalCount": 1, + } + + def list_pipelines(self, workspace_id, pipeline_ids=None, **_kwargs): + assert workspace_id == "workspace-id" + assert pipeline_ids == ["pipeline-id"] + return { + "Items": [ + { + "Id": "pipeline-id", + "Name": "existing-pipeline", + "Spec": "value: $(parameters.TOS_BUCKET_NAME)", + } + ] + } + + def is_agentkit_build_pipeline(self, pipeline): + return pipeline["Id"] == "pipeline-id" + + def create_workspace(self, **_kwargs): + raise AssertionError("existing workspace must not be recreated") + + def _create_pipeline(self, **_kwargs): + raise AssertionError("existing pipeline must not be recreated") + + monkeypatch.setattr( + "agentkit.toolkit.volcengine.code_pipeline.VeCodePipeline", + _FakeCodePipeline, + ) + config = VeCPCRBuilderConfig( + common_config=CommonConfig(agent_name="agent", entry_point="app.py"), + cp_workspace_name="custom-workspace", + cp_pipeline_name="existing-pipeline", + cp_pipeline_id="pipeline-id", + ) + builder = VeCPCRBuilder(project_dir=tmp_path, reporter=_Reporter()) + + pipeline_id = builder._prepare_pipeline_resources(config, "tos://source", object()) + + assert pipeline_id == "pipeline-id" + assert config.cp_pipeline_id == "pipeline-id" + assert builder._workspace_id == "workspace-id" + + +def test_cloud_strategy_preserves_custom_pipeline_selection() -> None: + captured = [] + + class _Builder: + def build(self, config): + captured.append(config) + return BuildResult(success=True) + + strategy = CloudStrategy() + strategy._builder = _Builder() + config = CloudStrategyConfig( + runtime_name="runtime-name", + cp_workspace_name="workspace-name", + cp_pipeline_name="pipeline-name", + cp_pipeline_id="pipeline-id", + ) + + strategy.build(CommonConfig(agent_name="agent", entry_point="app.py"), config) + + assert captured[0].cp_workspace_name == "workspace-name" + assert captured[0].cp_pipeline_name == "pipeline-name" + assert captured[0].cp_pipeline_id == "pipeline-id" + + +def test_all_studio_resource_mode_combinations_reach_builder_config() -> None: + strategy = CloudStrategy() + common = CommonConfig(agent_name="matrix-agent", entry_point="app.py") + + for tos_mode, cr_mode, cp_mode in itertools.product( + ("auto", "create", "existing"), repeat=3 + ): + config = CloudStrategyConfig(region="cn-beijing") + if tos_mode != "auto": + config.tos_bucket = f"tos-{tos_mode}" + if cr_mode != "auto": + config.cr_instance_name = f"cr-{cr_mode}" + config.cr_namespace_name = f"namespace-{cr_mode}" + config.cr_repo_name = f"repository-{cr_mode}" + if cp_mode != "auto": + config.cp_workspace_name = f"workspace-{cp_mode}" + config.cp_pipeline_name = f"pipeline-{cp_mode}" + if cp_mode == "existing": + config.cp_pipeline_id = "pipeline-existing-id" + + builder = strategy._to_builder_config( + common, + config, + runtime_name_override="matrix-runtime", + cp_pipeline_name_override=( + "matrix-runtime" if cp_mode == "auto" else config.cp_pipeline_name + ), + ) + + assert builder.tos_bucket == ( + AUTO_CREATE_VE if tos_mode == "auto" else f"tos-{tos_mode}" + ) + assert builder.cr_instance_name == ( + AUTO_CREATE_VE if cr_mode == "auto" else f"cr-{cr_mode}" + ) + assert builder.cr_namespace_name == ( + "agentkit" if cr_mode == "auto" else f"namespace-{cr_mode}" + ) + assert builder.cr_repo_name == ( + "" if cr_mode == "auto" else f"repository-{cr_mode}" + ) + assert builder.cp_workspace_name == ( + DEFAULT_WORKSPACE_NAME if cp_mode == "auto" else f"workspace-{cp_mode}" + ) + assert builder.cp_pipeline_name == ( + "matrix-runtime" if cp_mode == "auto" else f"pipeline-{cp_mode}" + ) + assert builder.cp_pipeline_id == ( + "pipeline-existing-id" if cp_mode == "existing" else "" + ) diff --git a/tests/toolkit/volcengine/test_cloud_resource_listing.py b/tests/toolkit/volcengine/test_cloud_resource_listing.py new file mode 100644 index 00000000..aef42e08 --- /dev/null +++ b/tests/toolkit/volcengine/test_cloud_resource_listing.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from agentkit.platform import Credentials +from agentkit.toolkit.volcengine.code_pipeline import VeCodePipeline +from agentkit.toolkit.volcengine.cr import VeCR +from agentkit.toolkit.volcengine.services import tos_service +from agentkit.toolkit.volcengine.services.tos_service import ( + TOSService, + TOSServiceConfig, +) + + +def test_tos_uses_explicit_credentials(monkeypatch) -> None: + captured: dict[str, object] = {} + + class _Configuration: + def __init__(self, **kwargs) -> None: + captured.update(kwargs) + + def get_service_credentials(self, _service: str): + return SimpleNamespace( + access_key="explicit-ak", + secret_key="explicit-sk", + session_token="explicit-token", + ) + + def get_service_endpoint(self, _service: str): + return SimpleNamespace(host="tos.example.com", region="ap-southeast-1") + + monkeypatch.setattr("agentkit.platform.VolcConfiguration", _Configuration) + monkeypatch.setattr( + tos_service, + "tos", + SimpleNamespace(TosClientV2=lambda *_args, **_kwargs: object()), + ) + monkeypatch.setattr(tos_service, "TOS_AVAILABLE", True) + + TOSService( + TOSServiceConfig(bucket="bucket-a", region="ap-southeast-1"), + provider="byteplus", + credentials=Credentials( + access_key="explicit-ak", + secret_key="explicit-sk", + session_token="explicit-token", + ), + ) + + assert captured == { + "region": "ap-southeast-1", + "provider": "byteplus", + "access_key": "explicit-ak", + "secret_key": "explicit-sk", + "session_token": "explicit-token", + } + + +def test_tos_list_buckets_returns_names_locations_and_creation_times() -> None: + service = object.__new__(TOSService) + service.actual_region = "cn-beijing" + service.client = SimpleNamespace( + list_buckets=lambda: SimpleNamespace( + buckets=[ + SimpleNamespace( + name="source-bucket", + location="cn-beijing", + creation_date="2026-08-07T00:00:00Z", + ) + ] + ) + ) + + assert service.list_buckets() == [ + { + "Name": "source-bucket", + "Location": "cn-beijing", + "CreationDate": "2026-08-07T00:00:00Z", + } + ] + + +def test_cr_resource_lists_use_parent_filters_and_pagination() -> None: + client = object.__new__(VeCR) + calls: list[tuple[str, dict]] = [] + + def request(request_body: dict, action: str) -> dict: + calls.append((action, request_body)) + return {"Result": {"Items": [{"Name": action}], "TotalCount": 1}} + + client._ve_request = request + + assert client.list_registries(page_number=2, page_size=20)["TotalCount"] == 1 + assert client.list_namespaces("registry-a", page_size=50)["TotalCount"] == 1 + assert ( + client.list_repositories("registry-a", "namespace-a", page_size=100)[ + "TotalCount" + ] + == 1 + ) + assert calls == [ + ("ListRegistries", {"PageNumber": 2, "PageSize": 20}), + ( + "ListNamespaces", + {"Registry": "registry-a", "PageNumber": 1, "PageSize": 50}, + ), + ( + "ListRepositories", + { + "Registry": "registry-a", + "Namespace": "namespace-a", + "PageNumber": 1, + "PageSize": 100, + }, + ), + ] + + +def test_code_pipeline_list_item_can_be_checked_for_agentkit_compatibility() -> None: + spec = "\n".join( + f"value: $({{parameters.{key}}})" + for key in VeCodePipeline.AGENTKIT_BUILD_PARAMETER_KEYS + ) + + assert VeCodePipeline.is_agentkit_build_pipeline({"Spec": spec}) is True + assert VeCodePipeline.is_agentkit_build_pipeline({"Spec": "value: $OTHER"}) is False