diff --git a/examples/create_workspace_with_landscape.py b/examples/create_workspace_with_landscape.py
new file mode 100644
index 0000000..a85cf28
--- /dev/null
+++ b/examples/create_workspace_with_landscape.py
@@ -0,0 +1,62 @@
+import asyncio
+import time
+
+from codesphere import CodesphereSDK
+from codesphere.resources.workspace import WorkspaceCreate
+from codesphere.resources.workspace.landscape import ProfileBuilder, ProfileConfig
+
+TEAM_ID = 123  # Replace with your actual team ID
+
+
+async def get_plan_id(sdk: CodesphereSDK, plan_name: str = "Micro") -> int:
+    plans = await sdk.metadata.list_plans()
+    plan = next((p for p in plans if p.title == plan_name and not p.deprecated), None)
+    if not plan:
+        raise ValueError(f"Plan '{plan_name}' not found")
+    return plan.id
+
+
+def build_web_profile(plan_id: int) -> ProfileConfig:
+    """Build a simple web service landscape profile."""
+    return (
+        ProfileBuilder()
+        .prepare()
+        .add_step("npm install", name="Install dependencies")
+        .done()
+        .add_reactive_service("web")
+        .plan(plan_id)
+        .add_step("npm start")
+        .add_port(3000, public=True)
+        .add_path("/", port=3000)
+        .replicas(1)
+        .env("NODE_ENV", "production")
+        .build()
+    )
+
+
+async def create_workspace(sdk: CodesphereSDK, plan_id: int, name: str):
+    workspace = await sdk.workspaces.create(
+        WorkspaceCreate(plan_id=plan_id, team_id=TEAM_ID, name=name)
+    )
+    await workspace.wait_until_running(timeout=300.0, poll_interval=5.0)
+    return workspace
+
+
+async def deploy_landscape(workspace, profile: dict, profile_name: str = "production"):
+    await workspace.landscape.save_profile(profile_name, profile)
+    await workspace.landscape.deploy(profile=profile_name)
+    print("Deployment started!")
+
+
+async def main():
+    async with CodesphereSDK() as sdk:
+        plan_id = await get_plan_id(sdk)
+        workspace = await create_workspace(
+            sdk, plan_id, f"landscape-demo-{int(time.time())}"
+        )
+        profile = build_web_profile(plan_id)
+        await deploy_landscape(workspace, profile)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/src/codesphere/resources/workspace/git/models.py b/src/codesphere/resources/workspace/git/models.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/codesphere/resources/workspace/git/schema.py b/src/codesphere/resources/workspace/git/schema.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/codesphere/resources/workspace/landscape/__init__.py b/src/codesphere/resources/workspace/landscape/__init__.py
new file mode 100644
index 0000000..67950be
--- /dev/null
+++ b/src/codesphere/resources/workspace/landscape/__init__.py
@@ -0,0 +1,31 @@
+from .models import WorkspaceLandscapeManager
+from .schemas import (
+    ManagedServiceBuilder,
+    ManagedServiceConfig,
+    NetworkConfig,
+    PathConfig,
+    PortConfig,
+    Profile,
+    ProfileBuilder,
+    ProfileConfig,
+    ReactiveServiceBuilder,
+    ReactiveServiceConfig,
+    StageConfig,
+    Step,
+)
+
+__all__ = [
+    "WorkspaceLandscapeManager",
+    "Profile",
+    "ProfileBuilder",
+    "ProfileConfig",
+    "Step",
+    "StageConfig",
+    "ReactiveServiceConfig",
+    "ReactiveServiceBuilder",
+    "ManagedServiceConfig",
+    "ManagedServiceBuilder",
+    "NetworkConfig",
+    "PortConfig",
+    "PathConfig",
+]
diff --git a/src/codesphere/resources/workspace/landscape/models.py b/src/codesphere/resources/workspace/landscape/models.py
index e69de29..8d6fc11 100644
--- a/src/codesphere/resources/workspace/landscape/models.py
+++ b/src/codesphere/resources/workspace/landscape/models.py
@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import logging
+import re
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
+
+from ....core.base import ResourceList
+from ....core.handler import _APIOperationExecutor
+from ....http_client import APIHttpClient
+from .operations import (
+    _DEPLOY_OP,
+    _DEPLOY_WITH_PROFILE_OP,
+    _SCALE_OP,
+    _TEARDOWN_OP,
+)
+from .schemas import Profile, ProfileConfig
+
+if TYPE_CHECKING:
+    from ..schemas import CommandOutput
+
+log = logging.getLogger(__name__)
+
+# Regex pattern to match ci.<profile>.yml files
+_PROFILE_FILE_PATTERN = re.compile(r"^ci\.([A-Za-z0-9_-]+)\.yml$")
+# Pattern for valid profile names
+_VALID_PROFILE_NAME = re.compile(r"^[A-Za-z0-9_-]+$")
+
+
+def _validate_profile_name(name: str) -> None:
+    if not _VALID_PROFILE_NAME.match(name):
+        raise ValueError(
+            f"Invalid profile name '{name}'. Must match pattern ^[A-Za-z0-9_-]+$"
+        )
+
+
+def _profile_filename(name: str) -> str:
+    _validate_profile_name(name)
+    return f"ci.{name}.yml"
+
+
+class WorkspaceLandscapeManager(_APIOperationExecutor):
+    def __init__(self, http_client: APIHttpClient, workspace_id: int):
+        self._http_client = http_client
+        self._workspace_id = workspace_id
+        self.id = workspace_id
+
+    async def _run_command(self, command: str) -> "CommandOutput":
+        from ..operations import _EXECUTE_COMMAND_OP
+        from ..schemas import CommandInput
+
+        return await self._execute_operation(
+            _EXECUTE_COMMAND_OP, data=CommandInput(command=command)
+        )
+
+    async def list_profiles(self) -> ResourceList[Profile]:
+        result = await self._run_command("ls -1 *.yml 2>/dev/null || true")
+
+        profiles: List[Profile] = []
+        if result.output:
+            for line in result.output.strip().split("\n"):
+                if match := _PROFILE_FILE_PATTERN.match(line.strip()):
+                    profiles.append(Profile(name=match.group(1)))
+
+        return ResourceList[Profile](root=profiles)
+
+    async def save_profile(self, name: str, config: Union[ProfileConfig, str]) -> None:
+        filename = _profile_filename(name)
+
+        if isinstance(config, ProfileConfig):
+            yaml_content = config.to_yaml()
+        else:
+            yaml_content = config
+
+        body = yaml_content if yaml_content.endswith("\n") else yaml_content + "\n"
+        await self._run_command(
+            f"cat > {filename} << 'PROFILE_EOF'\n{body}PROFILE_EOF\n"
+        )
+
+    async def get_profile(self, name: str) -> str:
+        result = await self._run_command(f"cat {_profile_filename(name)}")
+        return result.output
+
+    async def delete_profile(self, name: str) -> None:
+        await self._run_command(f"rm -f {_profile_filename(name)}")
+
+    async def deploy(self, profile: Optional[str] = None) -> None:
+        if profile is not None:
+            _validate_profile_name(profile)
+            await self._execute_operation(_DEPLOY_WITH_PROFILE_OP, profile=profile)
+        else:
+            await self._execute_operation(_DEPLOY_OP)
+
+    async def teardown(self) -> None:
+        await self._execute_operation(_TEARDOWN_OP)
+
+    async def scale(self, services: Dict[str, int]) -> None:
+        await self._execute_operation(_SCALE_OP, data=services)
diff --git a/src/codesphere/resources/workspace/landscape/operations.py b/src/codesphere/resources/workspace/landscape/operations.py
new file mode 100644
index 0000000..9197b09
--- /dev/null
+++ b/src/codesphere/resources/workspace/landscape/operations.py
@@ -0,0 +1,25 @@
+from ....core.operations import APIOperation
+
+_DEPLOY_OP = APIOperation(
+    method="POST",
+    endpoint_template="/workspaces/{id}/landscape/deploy",
+    response_model=type(None),
+)
+
+_DEPLOY_WITH_PROFILE_OP = APIOperation(
+    method="POST",
+    endpoint_template="/workspaces/{id}/landscape/deploy/{profile}",
+    response_model=type(None),
+)
+
+_TEARDOWN_OP = APIOperation(
+    method="DELETE",
+    endpoint_template="/workspaces/{id}/landscape/teardown",
+    response_model=type(None),
+)
+
+_SCALE_OP = APIOperation(
+    method="PATCH",
+    endpoint_template="/workspaces/{id}/landscape/scale",
+    response_model=type(None),
+)
diff --git a/src/codesphere/resources/workspace/landscape/schemas.py b/src/codesphere/resources/workspace/landscape/schemas.py
new file mode 100644
index 0000000..621b719
--- /dev/null
+++ b/src/codesphere/resources/workspace/landscape/schemas.py
@@ -0,0 +1,438 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Literal, Optional
+
+import yaml
+from pydantic import BaseModel, Field
+
+from ....core.base import CamelModel
+
+
+class Profile(BaseModel):
+    name: str
+
+
+class Step(CamelModel):
+    name: Optional[str] = None
+    command: str
+
+
+class PortConfig(CamelModel):
+    port: int = Field(ge=1, le=65535)
+    is_public: bool = False
+
+
+class PathConfig(CamelModel):
+    port: int = Field(ge=1, le=65535)
+    path: str
+    strip_path: Optional[bool] = None
+
+
+class NetworkConfig(CamelModel):
+    ports: List[PortConfig] = Field(default_factory=list)
+    paths: List[PathConfig] = Field(default_factory=list)
+
+
+class ReactiveServiceConfig(CamelModel):
+    steps: List[Step] = Field(default_factory=list)
+    plan: int
+    replicas: int = 1
+    env: Optional[Dict[str, str]] = None
+    base_image: Optional[str] = None
+    run_as_user: Optional[int] = Field(default=None, ge=0, le=65534)
+    run_as_group: Optional[int] = Field(default=None, ge=0, le=65534)
+    mount_sub_path: Optional[str] = None
+    health_endpoint: Optional[str] = None
+    network: Optional[NetworkConfig] = None
+
+
+class ManagedServiceConfig(CamelModel):
+    provider: str
+    plan: str
+    config: Optional[Dict[str, Any]] = None
+    secrets: Optional[Dict[str, str]] = None
+
+
+class StageConfig(CamelModel):
+    steps: List[Step] = Field(default_factory=list)
+
+
+class ProfileConfig(CamelModel):
+    schema_version: Literal["v0.2"] = Field(default="v0.2", alias="schemaVersion")
+    prepare: StageConfig = Field(default_factory=StageConfig)
+    test: StageConfig = Field(default_factory=StageConfig)
+    run: Dict[str, ReactiveServiceConfig | ManagedServiceConfig] = Field(
+        default_factory=dict
+    )
+
+    def to_yaml(self, *, exclude_none: bool = True) -> str:
+        data = self.model_dump(by_alias=True, exclude_none=exclude_none, mode="json")
+        return yaml.safe_dump(
+            data, default_flow_style=False, allow_unicode=True, sort_keys=False
+        )
+
+
+class StepBuilder:
+    def __init__(self, command: str, name: Optional[str] = None):
+        self._command = command
+        self._name = name
+
+    def build(self) -> Step:
+        return Step(command=self._command, name=self._name)
+
+
+class PortBuilder:
+    def __init__(self, port: int):
+        self._port = port
+        self._is_public = False
+
+    def public(self, is_public: bool = True) -> PortBuilder:
+        self._is_public = is_public
+        return self
+
+    def build(self) -> PortConfig:
+        return PortConfig(port=self._port, is_public=self._is_public)
+
+
+class PathBuilder:
+    def __init__(self, path: str, port: int):
+        self._path = path
+        self._port = port
+        self._strip_path: Optional[bool] = None
+
+    def strip_path(self, strip: bool = True) -> PathBuilder:
+        self._strip_path = strip
+        return self
+
+    def build(self) -> PathConfig:
+        return PathConfig(port=self._port, path=self._path, strip_path=self._strip_path)
+
+
+class ReactiveServiceBuilder:
+    def __init__(self, name: str):
+        self._name = name
+        self._steps: List[Step] = []
+        self._env: Dict[str, str] = {}
+        self._plan: Optional[int] = None
+        self._replicas: int = 1
+        self._base_image: Optional[str] = None
+        self._run_as_user: Optional[int] = None
+        self._run_as_group: Optional[int] = None
+        self._mount_sub_path: Optional[str] = None
+        self._health_endpoint: Optional[str] = None
+        self._ports: List[PortConfig] = []
+        self._paths: List[PathConfig] = []
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def add_step(
+        self, command: str, name: Optional[str] = None
+    ) -> ReactiveServiceBuilder:
+        self._steps.append(Step(command=command, name=name))
+        return self
+
+    def env(self, key: str, value: str) -> ReactiveServiceBuilder:
+        self._env[key] = value
+        return self
+
+    def envs(self, env_vars: Dict[str, str]) -> ReactiveServiceBuilder:
+        self._env.update(env_vars)
+        return self
+
+    def plan(self, plan_id: int) -> ReactiveServiceBuilder:
+        self._plan = plan_id
+        return self
+
+    def replicas(self, count: int) -> ReactiveServiceBuilder:
+        self._replicas = max(1, count)
+        return self
+
+    def base_image(self, image: str) -> ReactiveServiceBuilder:
+        self._base_image = image
+        return self
+
+    def run_as(
+        self, user: Optional[int] = None, group: Optional[int] = None
+    ) -> ReactiveServiceBuilder:
+        self._run_as_user = user
+        self._run_as_group = group
+        return self
+
+    def mount_sub_path(self, path: str) -> ReactiveServiceBuilder:
+        self._mount_sub_path = path
+        return self
+
+    def health_endpoint(self, endpoint: str) -> ReactiveServiceBuilder:
+        self._health_endpoint = endpoint
+        return self
+
+    def add_port(self, port: int, *, public: bool = False) -> ReactiveServiceBuilder:
+        self._ports.append(PortConfig(port=port, is_public=public))
+        return self
+
+    def add_path(
+        self, path: str, port: int, *, strip_path: Optional[bool] = None
+    ) -> ReactiveServiceBuilder:
+        self._paths.append(PathConfig(port=port, path=path, strip_path=strip_path))
+        return self
+
+    def build(self) -> tuple[str, ReactiveServiceConfig]:
+        if self._plan is None:
+            raise ValueError(
+                f"Service '{self._name}' requires a plan ID. "
+                "Use .plan(plan_id) to set it."
+            )
+
+        network = None
+        if self._ports or self._paths:
+            network = NetworkConfig(ports=self._ports, paths=self._paths)
+
+        config = ReactiveServiceConfig(
+            steps=self._steps,
+            plan=self._plan,
+            replicas=self._replicas,
+            env=self._env if self._env else None,
+            base_image=self._base_image,
+            run_as_user=self._run_as_user,
+            run_as_group=self._run_as_group,
+            mount_sub_path=self._mount_sub_path,
+            health_endpoint=self._health_endpoint,
+            network=network,
+        )
+        return self._name, config
+
+
+class ManagedServiceBuilder:
+    def __init__(self, name: str, provider: str, plan: str):
+        self._name = name
+        self._provider = provider
+        self._plan = plan
+        self._config: Dict[str, Any] = {}
+        self._secrets: Dict[str, str] = {}
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def config(self, key: str, value: Any) -> ManagedServiceBuilder:
+        self._config[key] = value
+        return self
+
+    def configs(self, config: Dict[str, Any]) -> ManagedServiceBuilder:
+        self._config.update(config)
+        return self
+
+    def secret(self, key: str, value: str) -> ManagedServiceBuilder:
+        self._secrets[key] = value
+        return self
+
+    def secrets(self, secrets: Dict[str, str]) -> ManagedServiceBuilder:
+        self._secrets.update(secrets)
+        return self
+
+    def build(self) -> tuple[str, ManagedServiceConfig]:
+        config = ManagedServiceConfig(
+            provider=self._provider,
+            plan=self._plan,
+            config=self._config if self._config else None,
+            secrets=self._secrets if self._secrets else None,
+        )
+        return self._name, config
+
+
+class ProfileBuilder:
+    """Fluent builder for creating landscape profile configurations.
+
+    Example:
+        ```python
+        profile = (
+            ProfileBuilder()
+            .prepare()
+                .add_step("npm install")
+                .add_step("npm run build")
+            .done()
+            .add_reactive_service("web")
+                .add_step("npm start")
+                .add_port(3000, public=True)
+                .add_path("/api", port=3000)
+                .replicas(2)
+                .env("NODE_ENV", "production")
+            .done()
+            .add_managed_service("db", provider="postgres", plan="small")
+                .config("max_connections", 100)
+            .done()
+            .build()
+        )
+
+        # Save to workspace
+        await workspace.landscape.save_profile("production", profile)
+        ```
+    """
+
+    def __init__(self) -> None:
+        self._prepare_steps: List[Step] = []
+        self._test_steps: List[Step] = []
+        self._services: Dict[str, ReactiveServiceConfig | ManagedServiceConfig] = {}
+        self._current_service: Optional[Any] = None
+
+    def prepare(self) -> PrepareStageBuilder:
+        return PrepareStageBuilder(self)
+
+    def test(self) -> TestStageBuilder:
+        return TestStageBuilder(self)
+
+    def add_reactive_service(self, name: str) -> ReactiveServiceBuilderContext:
+        self._finalize_current_service()
+        self._current_service = ReactiveServiceBuilderContext(self, name)
+        return self._current_service
+
+    def add_managed_service(
+        self, name: str, *, provider: str, plan: str
+    ) -> ManagedServiceBuilderContext:
+        self._finalize_current_service()
+        self._current_service = ManagedServiceBuilderContext(self, name, provider, plan)
+        return self._current_service
+
+    def __getattr__(self, name: str) -> Any:
+        """Delegate unknown methods to the current service builder."""
+        if self._current_service is not None and hasattr(self._current_service, name):
+            method = getattr(self._current_service, name)
+            if callable(method):
+
+                def wrapper(*args: Any, **kwargs: Any) -> ProfileBuilder:
+                    method(*args, **kwargs)
+                    return self
+
+                return wrapper
+        raise AttributeError(f"'{type(self).__name__}' has no attribute '{name}'")
+
+    def _finalize_current_service(self) -> None:
+        if self._current_service is not None:
+            name, config = self._current_service._builder.build()
+            self._services[name] = config
+            self._current_service = None
+
+    def build(self) -> ProfileConfig:
+        """Build the final profile configuration."""
+        self._finalize_current_service()
+        return ProfileConfig(
+            prepare=StageConfig(steps=self._prepare_steps),
+            test=StageConfig(steps=self._test_steps),
+            run=self._services,
+        )
+
+
+class PrepareStageBuilder:
+    def __init__(self, parent: ProfileBuilder):
+        self._parent = parent
+
+    def add_step(self, command: str, name: Optional[str] = None) -> PrepareStageBuilder:
+        self._parent._prepare_steps.append(Step(command=command, name=name))
+        return self
+
+    def done(self) -> ProfileBuilder:
+        return self._parent
+
+
+class TestStageBuilder:
+    def __init__(self, parent: ProfileBuilder):
+        self._parent = parent
+
+    def add_step(self, command: str, name: Optional[str] = None) -> TestStageBuilder:
+        self._parent._test_steps.append(Step(command=command, name=name))
+        return self
+
+    def done(self) -> ProfileBuilder:
+        return self._parent
+
+
+class ReactiveServiceBuilderContext:
+    def __init__(self, parent: ProfileBuilder, name: str):
+        self._parent = parent
+        self._builder = ReactiveServiceBuilder(name)
+
+    def add_step(
+        self, command: str, name: Optional[str] = None
+    ) -> ReactiveServiceBuilderContext:
+        self._builder.add_step(command, name)
+        return self
+
+    def env(self, key: str, value: str) -> ReactiveServiceBuilderContext:
+        self._builder.env(key, value)
+        return self
+
+    def envs(self, env_vars: Dict[str, str]) -> ReactiveServiceBuilderContext:
+        self._builder.envs(env_vars)
+        return self
+
+    def plan(self, plan_id: int) -> ReactiveServiceBuilderContext:
+        self._builder.plan(plan_id)
+        return self
+
+    def replicas(self, count: int) -> ReactiveServiceBuilderContext:
+        self._builder.replicas(count)
+        return self
+
+    def base_image(self, image: str) -> ReactiveServiceBuilderContext:
+        self._builder.base_image(image)
+        return self
+
+    def run_as(
+        self, user: Optional[int] = None, group: Optional[int] = None
+    ) -> ReactiveServiceBuilderContext:
+        self._builder.run_as(user, group)
+        return self
+
+    def mount_sub_path(self, path: str) -> ReactiveServiceBuilderContext:
+        self._builder.mount_sub_path(path)
+        return self
+
+    def health_endpoint(self, endpoint: str) -> ReactiveServiceBuilderContext:
+        self._builder.health_endpoint(endpoint)
+        return self
+
+    def add_port(
+        self, port: int, *, public: bool = False
+    ) -> ReactiveServiceBuilderContext:
+        self._builder.add_port(port, public=public)
+        return self
+
+    def add_path(
+        self, path: str, port: int, *, strip_path: Optional[bool] = None
+    ) -> ReactiveServiceBuilderContext:
+        self._builder.add_path(path, port, strip_path=strip_path)
+        return self
+
+    def done(self) -> ProfileBuilder:
+        name, config = self._builder.build()
+        self._parent._services[name] = config
+        return self._parent
+
+
+class ManagedServiceBuilderContext:
+    def __init__(self, parent: ProfileBuilder, name: str, provider: str, plan: str):
+        self._parent = parent
+        self._builder = ManagedServiceBuilder(name, provider, plan)
+
+    def config(self, key: str, value: Any) -> ManagedServiceBuilderContext:
+        self._builder.config(key, value)
+        return self
+
+    def configs(self, config: Dict[str, Any]) -> ManagedServiceBuilderContext:
+        self._builder.configs(config)
+        return self
+
+    def secret(self, key: str, value: str) -> ManagedServiceBuilderContext:
+        self._builder.secret(key, value)
+        return self
+
+    def secrets(self, secrets: Dict[str, str]) -> ManagedServiceBuilderContext:
+        self._builder.secrets(secrets)
+        return self
+
+    def done(self) -> ProfileBuilder:
+        name, config = self._builder.build()
+        self._parent._services[name] = config
+        return self._parent
diff --git a/src/codesphere/resources/workspace/schemas.py b/src/codesphere/resources/workspace/schemas.py
index a8e4528..938bb5e 100644
--- a/src/codesphere/resources/workspace/schemas.py
+++ b/src/codesphere/resources/workspace/schemas.py
@@ -1,12 +1,15 @@
 from __future__ import annotations
-from functools import cached_property
+
+import asyncio
 import logging
-from typing import Dict, Optional, List
+from functools import cached_property
+from typing import Dict, List, Optional
 
-from .envVars import EnvVar, WorkspaceEnvVarManager
-from ...core.base import CamelModel
 from ...core import _APIOperationExecutor
+from ...core.base import CamelModel
 from ...utils import update_model_fields
+from .envVars import EnvVar, WorkspaceEnvVarManager
+from .landscape import WorkspaceLandscapeManager
 
 log = logging.getLogger(__name__)
 
@@ -56,15 +59,11 @@ class WorkspaceUpdate(CamelModel):
 
 
 class CommandInput(CamelModel):
-    """Input model for command execution."""
-
     command: str
     env: Optional[Dict[str, str]] = None
 
 
 class CommandOutput(CamelModel):
-    """Output model for command execution."""
-
     command: str
     working_dir: str
     output: str
@@ -72,8 +71,6 @@ class CommandOutput(CamelModel):
 
 
 class WorkspaceStatus(CamelModel):
-    """Status information for a workspace."""
-
     is_running: bool
 
 
@@ -94,6 +91,34 @@ async def get_status(self) -> WorkspaceStatus:
 
         return await self._execute_operation(_GET_STATUS_OP)
 
+    async def wait_until_running(
+        self,
+        *,
+        timeout: float = 300.0,
+        poll_interval: float = 5.0,
+    ) -> None:
+        if poll_interval <= 0:
+            raise ValueError("poll_interval must be greater than 0")
+
+        elapsed = 0.0
+        while elapsed < timeout:
+            status = await self.get_status()
+            if status.is_running:
+                log.debug("Workspace %s is now running.", self.id)
+                return
+            log.debug(
+                "Workspace %s not running yet, waiting %ss... (elapsed: %.1fs)",
+                self.id,
+                poll_interval,
+                elapsed,
+            )
+            await asyncio.sleep(poll_interval)
+            elapsed += poll_interval
+
+        raise TimeoutError(
+            f"Workspace {self.id} did not reach running state within {timeout} seconds."
+        )
+
     async def execute_command(
         self, command: str, env: Optional[Dict[str, str]] = None
     ) -> CommandOutput:
@@ -106,3 +131,9 @@ async def execute_command(
     def env_vars(self) -> WorkspaceEnvVarManager:
         http_client = self.validate_http_client()
         return WorkspaceEnvVarManager(http_client, workspace_id=self.id)
+
+    @cached_property
+    def landscape(self) -> WorkspaceLandscapeManager:
+        """Manager for landscape operations (Multi Server Deployments)."""
+        http_client = self.validate_http_client()
+        return WorkspaceLandscapeManager(http_client, workspace_id=self.id)
diff --git a/tests/integration/test_landscape.py b/tests/integration/test_landscape.py
new file mode 100644
index 0000000..0e63f4c
--- /dev/null
+++ b/tests/integration/test_landscape.py
@@ -0,0 +1,530 @@
+import asyncio
+
+import pytest
+
+from codesphere import CodesphereSDK
+from codesphere.core.base import ResourceList
+from codesphere.resources.workspace import Workspace
+from codesphere.resources.workspace.landscape import (
+    Profile,
+    ProfileBuilder,
+)
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+
+class TestLandscapeProfilesIntegration:
+    async def test_list_profiles_returns_resource_list(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        profiles = await workspace.landscape.list_profiles()
+
+        assert isinstance(profiles, ResourceList)
+
+    async def test_list_profiles_empty_workspace(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        profiles = await workspace.landscape.list_profiles()
+
+        assert isinstance(profiles, ResourceList)
+        assert len(profiles) >= 0
+
+    async def test_list_profiles_after_creating_profile_file(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        profile_name = "sdk-test-profile"
+        create_result = await workspace.execute_command(
+            f"echo 'version: 1' > ci.{profile_name}.yml"
+        )
+
+        try:
+            profiles = await workspace.landscape.list_profiles()
+
+            profile_names = [p.name for p in profiles]
+            assert profile_name in profile_names
+
+            matching_profile = next(p for p in profiles if p.name == profile_name)
+            assert isinstance(matching_profile, Profile)
+
+        finally:
+            await workspace.execute_command(f"rm -f ci.{profile_name}.yml")
+
+    async def test_list_profiles_with_multiple_profile_files(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        """list_profiles should find multiple profiles."""
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        profile_names = ["test-profile-1", "test-profile-2", "test_profile_3"]
+        for name in profile_names:
+            await workspace.execute_command(f"echo 'version: 1' > ci.{name}.yml")
+
+        try:
+            profiles = await workspace.landscape.list_profiles()
+
+            found_names = [p.name for p in profiles]
+            for expected_name in profile_names:
+                assert expected_name in found_names, (
+                    f"Profile {expected_name} not found"
+                )
+
+        finally:
+            for name in profile_names:
+                await workspace.execute_command(f"rm -f ci.{name}.yml")
+
+    async def test_list_profiles_ignores_non_profile_yml_files(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        await workspace.execute_command("echo 'version: 1' > ci.valid-profile.yml")
+        await workspace.execute_command("echo 'key: value' > config.yml")
+        await workspace.execute_command("echo 'services: []' > docker-compose.yml")
+
+        try:
+            profiles = await workspace.landscape.list_profiles()
+
+            profile_names = [p.name for p in profiles]
+            assert "valid-profile" in profile_names
+            assert "config" not in profile_names
+            assert "docker-compose" not in profile_names
+
+        finally:
+            await workspace.execute_command(
+                "rm -f ci.valid-profile.yml config.yml docker-compose.yml"
+            )
+
+    async def test_list_profiles_iterable(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        await workspace.execute_command("echo 'version: 1' > ci.iter-test.yml")
+
+        try:
+            profiles = await workspace.landscape.list_profiles()
+
+            profile_list = list(profiles)
+            assert isinstance(profile_list, list)
+
+            if len(profiles) > 0:
+                first_profile = profiles[0]
+                assert isinstance(first_profile, Profile)
+
+        finally:
+            await workspace.execute_command("rm -f ci.iter-test.yml")
+
+
+class TestLandscapeManagerAccess:
+    async def test_workspace_has_landscape_property(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        assert hasattr(workspace, "landscape")
+        assert workspace.landscape is not None
+
+    async def test_landscape_manager_is_cached(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        manager1 = workspace.landscape
+        manager2 = workspace.landscape
+
+        assert manager1 is manager2
+
+
+class TestSaveProfileIntegration:
+    async def test_save_profile_with_builder(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+        test_plan_id: int,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-builder-test"
+
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("echo 'Installing dependencies'")
+            .done()
+            .add_reactive_service("web")
+            .plan(test_plan_id)
+            .add_step("echo 'Starting server'")
+            .add_port(3000, public=True)
+            .replicas(1)
+            .done()
+            .build()
+        )
+
+        try:
+            await workspace.landscape.save_profile(profile_name, profile)
+
+            profiles = await workspace.landscape.list_profiles()
+            profile_names = [p.name for p in profiles]
+            assert profile_name in profile_names
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+    async def test_save_profile_with_yaml_string(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-yaml-test"
+
+        yaml_content = """schemaVersion: v0.2
+prepare:
+  steps:
+    - command: echo 'test'
+test:
+  steps: []
+run: {}
+"""
+
+        try:
+            await workspace.landscape.save_profile(profile_name, yaml_content)
+
+            profiles = await workspace.landscape.list_profiles()
+            profile_names = [p.name for p in profiles]
+            assert profile_name in profile_names
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+    async def test_save_profile_overwrites_existing(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-overwrite-test"
+
+        profile_v1 = (
+            ProfileBuilder().prepare().add_step("echo 'version 1'").done().build()
+        )
+
+        profile_v2 = (
+            ProfileBuilder().prepare().add_step("echo 'version 2'").done().build()
+        )
+
+        try:
+            await workspace.landscape.save_profile(profile_name, profile_v1)
+            await workspace.landscape.save_profile(profile_name, profile_v2)
+
+            content = await workspace.landscape.get_profile(profile_name)
+            assert "version 2" in content
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+
+class TestGetProfileIntegration:
+    async def test_get_profile_returns_yaml_content(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+        test_plan_id: int,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-get-test"
+
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("npm install")
+            .done()
+            .add_reactive_service("api")
+            .plan(test_plan_id)
+            .add_step("npm start")
+            .add_port(8080)
+            .env("NODE_ENV", "production")
+            .done()
+            .build()
+        )
+
+        try:
+            await workspace.landscape.save_profile(profile_name, profile)
+
+            content = await workspace.landscape.get_profile(profile_name)
+
+            assert "schemaVersion: v0.2" in content
+            assert "npm install" in content
+            assert "api:" in content
+            assert "NODE_ENV" in content
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+
+class TestDeleteProfileIntegration:
+    async def test_delete_profile_removes_file(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-delete-test"
+
+        profile = ProfileBuilder().build()
+        await workspace.landscape.save_profile(profile_name, profile)
+
+        profiles = await workspace.landscape.list_profiles()
+        assert profile_name in [p.name for p in profiles]
+
+        await workspace.landscape.delete_profile(profile_name)
+
+        profiles = await workspace.landscape.list_profiles()
+        assert profile_name not in [p.name for p in profiles]
+
+    async def test_delete_nonexistent_profile_no_error(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+
+        await workspace.landscape.delete_profile("nonexistent-profile-xyz")
+
+
+class TestProfileBuilderIntegration:
+    async def test_complex_profile_roundtrip(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+        test_plan_id: int,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-complex-test"
+
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("npm ci", name="Install")
+            .add_step("npm run build", name="Build")
+            .done()
+            .test()
+            .add_step("npm test")
+            .done()
+            .add_reactive_service("frontend")
+            .plan(test_plan_id)
+            .add_step("npm run serve")
+            .add_port(3000, public=True)
+            .add_path("/", port=3000)
+            .replicas(2)
+            .env("NODE_ENV", "production")
+            .health_endpoint("/health")
+            .done()
+            .add_reactive_service("backend")
+            .plan(test_plan_id)
+            .add_step("python -m uvicorn main:app")
+            .add_port(8000)
+            .add_path("/api", port=8000, strip_path=True)
+            .envs({"PYTHONPATH": "/app", "LOG_LEVEL": "info"})
+            .done()
+            .add_managed_service("database", provider="postgres", plan="small")
+            .config("max_connections", 50)
+            .done()
+            .build()
+        )
+
+        try:
+            await workspace.landscape.save_profile(profile_name, profile)
+
+            content = await workspace.landscape.get_profile(profile_name)
+
+            assert "schemaVersion: v0.2" in content
+            assert "frontend:" in content
+            assert "backend:" in content
+            assert "database:" in content
+            assert "npm ci" in content
+            assert "replicas: 2" in content
+            assert "postgres" in content
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+    async def test_profile_with_special_characters_in_env(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+        test_plan_id: int,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-special-chars-test"
+
+        profile = (
+            ProfileBuilder()
+            .add_reactive_service("app")
+            .plan(test_plan_id)
+            .add_step("npm start")
+            .add_port(3000)
+            .env("DATABASE_URL", "postgres://user:p@ss=word@localhost:5432/db")
+            .env("API_KEY", "sk-1234567890abcdef")
+            .done()
+            .build()
+        )
+
+        try:
+            await workspace.landscape.save_profile(profile_name, profile)
+
+            content = await workspace.landscape.get_profile(profile_name)
+            assert "DATABASE_URL" in content
+            assert "API_KEY" in content
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+
+class TestLandscapeDeploymentWorkflow:
+    async def test_full_landscape_workflow_deploy_teardown_delete(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+        test_plan_id: int,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-workflow-test"
+
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("echo 'Preparing...'", name="Prepare")
+            .done()
+            .add_reactive_service("web")
+            .plan(test_plan_id)
+            .add_step("echo 'Starting web service' && sleep infinity")
+            .add_port(3000, public=True)
+            .add_path("/", port=3000)
+            .replicas(1)
+            .env("NODE_ENV", "production")
+            .done()
+            .build()
+        )
+
+        try:
+            await workspace.landscape.save_profile(profile_name, profile)
+
+            profiles = await workspace.landscape.list_profiles()
+            profile_names = [p.name for p in profiles]
+            assert profile_name in profile_names, "Profile should exist after saving"
+
+            content = await workspace.landscape.get_profile(profile_name)
+            assert "schemaVersion: v0.2" in content
+            assert "web:" in content
+            assert f"plan: {test_plan_id}" in content
+
+            await workspace.landscape.deploy(profile=profile_name)
+
+            await workspace.landscape.teardown()
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+            profiles = await workspace.landscape.list_profiles()
+            profile_names = [p.name for p in profiles]
+            assert profile_name not in profile_names, (
+                "Profile should not exist after deletion"
+            )
+
+    async def test_deploy_and_teardown_only(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+        test_plan_id: int,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-deploy-teardown-test"
+
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("echo 'Setup complete'")
+            .done()
+            .add_reactive_service("api")
+            .plan(test_plan_id)
+            .add_step("echo 'API running' && sleep infinity")
+            .add_port(8080)
+            .add_path("/api", port=8080)
+            .replicas(1)
+            .done()
+            .build()
+        )
+
+        try:
+            await workspace.landscape.save_profile(profile_name, profile)
+
+            # got mutex errors when deploy and teardown were called in quick succession,
+            # adding delay to mitigate
+            # maybe report a bug if this continues to be an issue
+            await asyncio.sleep(2)
+
+            await workspace.landscape.deploy(profile=profile_name)
+
+            await workspace.landscape.teardown()
+
+            profiles = await workspace.landscape.list_profiles()
+            profile_names = [p.name for p in profiles]
+            assert profile_name in profile_names, (
+                "Profile should still exist after teardown"
+            )
+
+        finally:
+            await workspace.landscape.delete_profile(profile_name)
+
+    async def test_profile_deletion_removes_from_list(
+        self,
+        sdk_client: CodesphereSDK,
+        test_workspace: Workspace,
+        test_plan_id: int,
+    ):
+        workspace = await sdk_client.workspaces.get(workspace_id=test_workspace.id)
+        profile_name = "sdk-deletion-verify-test"
+
+        profile = (
+            ProfileBuilder()
+            .add_reactive_service("service")
+            .plan(test_plan_id)
+            .add_step("echo 'running'")
+            .add_port(3000)
+            .done()
+            .build()
+        )
+
+        await workspace.landscape.save_profile(profile_name, profile)
+
+        profiles_before = await workspace.landscape.list_profiles()
+        assert profile_name in [p.name for p in profiles_before]
+
+        await workspace.landscape.delete_profile(profile_name)
+
+        profiles_after = await workspace.landscape.list_profiles()
+        assert profile_name not in [p.name for p in profiles_after], (
+            f"Profile '{profile_name}' should be removed after deletion"
+        )
diff --git a/tests/resources/workspace/landscape/__init__.py b/tests/resources/workspace/landscape/__init__.py
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/tests/resources/workspace/landscape/__init__.py
@@ -0,0 +1 @@
+
diff --git a/tests/resources/workspace/landscape/test_landscape.py b/tests/resources/workspace/landscape/test_landscape.py
new file mode 100644
index 0000000..314a247
--- /dev/null
+++ b/tests/resources/workspace/landscape/test_landscape.py
@@ -0,0 +1,662 @@
+import pytest
+
+from codesphere.core.base import ResourceList
+from codesphere.resources.workspace.landscape import (
+    ManagedServiceBuilder,
+    ManagedServiceConfig,
+    NetworkConfig,
+    PathConfig,
+    PortConfig,
+    Profile,
+    ProfileBuilder,
+    ProfileConfig,
+    ReactiveServiceBuilder,
+    ReactiveServiceConfig,
+    StageConfig,
+    Step,
+    WorkspaceLandscapeManager,
+)
+
+
+class TestWorkspaceLandscapeManager:
+    @pytest.fixture
+    def landscape_manager(self, mock_http_client_for_resource):
+        mock_client = mock_http_client_for_resource(None)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+        return manager, mock_client
+
+    @pytest.mark.asyncio
+    async def test_deploy_without_profile(self, landscape_manager):
+        manager, mock_client = landscape_manager
+
+        await manager.deploy()
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/landscape/deploy"
+
+    @pytest.mark.asyncio
+    async def test_deploy_with_profile(self, mock_http_client_for_resource):
+        mock_client = mock_http_client_for_resource(None)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        await manager.deploy(profile="my-profile")
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert (
+            call_args.kwargs.get("endpoint")
+            == "/workspaces/72678/landscape/deploy/my-profile"
+        )
+
+    @pytest.mark.asyncio
+    async def test_teardown(self, landscape_manager):
+        manager, mock_client = landscape_manager
+
+        await manager.teardown()
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "DELETE"
+        assert (
+            call_args.kwargs.get("endpoint") == "/workspaces/72678/landscape/teardown"
+        )
+
+    @pytest.mark.asyncio
+    async def test_scale_services(self, mock_http_client_for_resource):
+        mock_client = mock_http_client_for_resource(None)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        services = {"web": 3, "worker": 2}
+        await manager.scale(services=services)
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "PATCH"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/landscape/scale"
+        assert call_args.kwargs.get("json") == {"web": 3, "worker": 2}
+
+    @pytest.mark.asyncio
+    async def test_scale_single_service(self, mock_http_client_for_resource):
+        mock_client = mock_http_client_for_resource(None)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        services = {"api": 5}
+        await manager.scale(services=services)
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("json") == {"api": 5}
+
+
+class TestListProfiles:
+    @pytest.fixture
+    def mock_command_response(self):
+        def _create(output: str, error: str = ""):
+            return {
+                "command": "ls -1 *.yml 2>/dev/null || true",
+                "workingDir": "/home/user",
+                "output": output,
+                "error": error,
+            }
+
+        return _create
+
+    @pytest.mark.asyncio
+    async def test_list_profiles_with_multiple_profiles(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response(
+            "ci.production.yml\nci.staging.yml\nci.dev-test.yml\n"
+        )
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        result = await manager.list_profiles()
+
+        assert isinstance(result, ResourceList)
+        assert len(result) == 3
+        assert result[0].name == "production"
+        assert result[1].name == "staging"
+        assert result[2].name == "dev-test"
+
+    @pytest.mark.asyncio
+    async def test_list_profiles_with_single_profile(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response("ci.main.yml\n")
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        result = await manager.list_profiles()
+
+        assert len(result) == 1
+        assert result[0].name == "main"
+
+    @pytest.mark.asyncio
+    async def test_list_profiles_with_no_profiles(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response("")
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        result = await manager.list_profiles()
+
+        assert isinstance(result, ResourceList)
+        assert len(result) == 0
+
+    @pytest.mark.asyncio
+    async def test_list_profiles_filters_non_profile_yml_files(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response(
+            "ci.production.yml\nconfig.yml\ndocker-compose.yml\nci.staging.yml\n"
+        )
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        result = await manager.list_profiles()
+
+        assert len(result) == 2
+        profile_names = [p.name for p in result]
+        assert "production" in profile_names
+        assert "staging" in profile_names
+
+    @pytest.mark.asyncio
+    async def test_list_profiles_with_underscore_in_name(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response("ci.my_profile.yml\n")
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        result = await manager.list_profiles()
+
+        assert len(result) == 1
+        assert result[0].name == "my_profile"
+
+    @pytest.mark.asyncio
+    async def test_list_profiles_calls_execute_endpoint(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response("")
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        await manager.list_profiles()
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/execute"
+
+
+class TestProfileModel:
+    def test_create_profile(self):
+        profile = Profile(name="production")
+
+        assert profile.name == "production"
+
+    def test_profile_from_dict(self):
+        profile = Profile.model_validate({"name": "staging"})
+
+        assert profile.name == "staging"
+
+    def test_profile_dump(self):
+        profile = Profile(name="dev")
+        dumped = profile.model_dump()
+
+        assert dumped == {"name": "dev"}
+
+
+class TestWorkspaceLandscapeManagerAccess:
+    @pytest.mark.asyncio
+    async def test_workspace_landscape_property(self, workspace_model_factory):
+        workspace, _ = workspace_model_factory()
+
+        landscape_manager = workspace.landscape
+
+        assert isinstance(landscape_manager, WorkspaceLandscapeManager)
+
+    @pytest.mark.asyncio
+    async def test_workspace_landscape_is_cached(self, workspace_model_factory):
+        workspace, _ = workspace_model_factory()
+
+        manager1 = workspace.landscape
+        manager2 = workspace.landscape
+
+        assert manager1 is manager2
+
+
+class TestSaveProfile:
+    @pytest.fixture
+    def mock_command_response(self):
+        def _create(output: str = "", error: str = ""):
+            return {
+                "command": "",
+                "workingDir": "/home/user",
+                "output": output,
+                "error": error,
+            }
+
+        return _create
+
+    @pytest.mark.asyncio
+    async def test_save_profile_with_profile_config(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response()
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        config = ProfileConfig()
+        await manager.save_profile("production", config)
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/execute"
+
+    @pytest.mark.asyncio
+    async def test_save_profile_with_yaml_string(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response()
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        yaml_content = "schemaVersion: v0.2\nprepare:\n  steps: []\n"
+        await manager.save_profile("staging", yaml_content)
+
+        mock_client.request.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_save_profile_invalid_name_raises_error(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response()
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        with pytest.raises(ValueError, match="Invalid profile name"):
+            await manager.save_profile("invalid/name", ProfileConfig())
+
+    @pytest.mark.asyncio
+    async def test_save_profile_invalid_name_with_spaces(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response()
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        with pytest.raises(ValueError):
+            await manager.save_profile("my profile", ProfileConfig())
+
+
+class TestGetProfile:
+    @pytest.fixture
+    def mock_command_response(self):
+        def _create(output: str = "", error: str = ""):
+            return {
+                "command": "",
+                "workingDir": "/home/user",
+                "output": output,
+                "error": error,
+            }
+
+        return _create
+
+    @pytest.mark.asyncio
+    async def test_get_profile_returns_yaml_content(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        yaml_content = "schemaVersion: v0.2\nprepare:\n  steps: []\n"
+        response_data = mock_command_response(output=yaml_content)
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        result = await manager.get_profile("production")
+
+        assert result == yaml_content
+
+    @pytest.mark.asyncio
+    async def test_get_profile_invalid_name_raises_error(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response()
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        with pytest.raises(ValueError, match="Invalid profile name"):
+            await manager.get_profile("invalid/name")
+
+
+class TestDeleteProfile:
+    @pytest.fixture
+    def mock_command_response(self):
+        def _create(output: str = "", error: str = ""):
+            return {
+                "command": "",
+                "workingDir": "/home/user",
+                "output": output,
+                "error": error,
+            }
+
+        return _create
+
+    @pytest.mark.asyncio
+    async def test_delete_profile_calls_rm_command(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response()
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        await manager.delete_profile("production")
+
+        mock_client.request.assert_awaited_once()
+        call_args = mock_client.request.call_args
+        assert call_args.kwargs.get("method") == "POST"
+        assert call_args.kwargs.get("endpoint") == "/workspaces/72678/execute"
+
+    @pytest.mark.asyncio
+    async def test_delete_profile_invalid_name_raises_error(
+        self, mock_http_client_for_resource, mock_command_response
+    ):
+        response_data = mock_command_response()
+        mock_client = mock_http_client_for_resource(response_data)
+        manager = WorkspaceLandscapeManager(http_client=mock_client, workspace_id=72678)
+
+        with pytest.raises(ValueError, match="Invalid profile name"):
+            await manager.delete_profile("../etc/passwd")
+
+
+class TestProfileBuilder:
+    TEST_PLAN_ID = 8
+
+    def test_build_empty_profile(self):
+        profile = ProfileBuilder().build()
+
+        assert isinstance(profile, ProfileConfig)
+        assert profile.schema_version == "v0.2"
+        assert len(profile.prepare.steps) == 0
+        assert len(profile.test.steps) == 0
+        assert len(profile.run) == 0
+
+    def test_build_with_prepare_steps(self):
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("npm install")
+            .add_step("npm run build", name="Build")
+            .done()
+            .build()
+        )
+
+        assert len(profile.prepare.steps) == 2
+        assert profile.prepare.steps[0].command == "npm install"
+        assert profile.prepare.steps[0].name is None
+        assert profile.prepare.steps[1].command == "npm run build"
+        assert profile.prepare.steps[1].name == "Build"
+
+    def test_build_with_test_steps(self):
+        profile = ProfileBuilder().test().add_step("npm test").done().build()
+
+        assert len(profile.test.steps) == 1
+        assert profile.test.steps[0].command == "npm test"
+
+    def test_build_with_reactive_service(self):
+        profile = (
+            ProfileBuilder()
+            .add_reactive_service("web")
+            .plan(self.TEST_PLAN_ID)
+            .add_step("npm start")
+            .add_port(3000, public=True)
+            .replicas(2)
+            .env("NODE_ENV", "production")
+            .done()
+            .build()
+        )
+
+        assert "web" in profile.run
+        service = profile.run["web"]
+        assert isinstance(service, ReactiveServiceConfig)
+        assert len(service.steps) == 1
+        assert service.plan == self.TEST_PLAN_ID
+        assert service.replicas == 2
+        assert service.env == {"NODE_ENV": "production"}
+        assert service.network is not None
+        assert len(service.network.ports) == 1
+        assert service.network.ports[0].port == 3000
+        assert service.network.ports[0].is_public is True
+
+    def test_build_with_reactive_service_paths(self):
+        profile = (
+            ProfileBuilder()
+            .add_reactive_service("api")
+            .plan(self.TEST_PLAN_ID)
+            .add_port(8080)
+            .add_path("/api", port=8080, strip_path=True)
+            .add_path("/health", port=8080)
+            .done()
+            .build()
+        )
+
+        service = profile.run["api"]
+        assert isinstance(service, ReactiveServiceConfig)
+        assert len(service.network.paths) == 2
+        assert service.network.paths[0].path == "/api"
+        assert service.network.paths[0].strip_path is True
+        assert service.network.paths[1].path == "/health"
+
+    def test_build_with_reactive_service_all_options(self):
+        profile = (
+            ProfileBuilder()
+            .add_reactive_service("worker")
+            .add_step("python worker.py")
+            .plan(123)
+            .replicas(3)
+            .base_image("python:3.11")
+            .run_as(user=1000, group=1000)
+            .mount_sub_path("/data")
+            .health_endpoint("/health")
+            .envs({"KEY1": "value1", "KEY2": "value2"})
+            .done()
+            .build()
+        )
+
+        service = profile.run["worker"]
+        assert isinstance(service, ReactiveServiceConfig)
+        assert service.plan == 123
+        assert service.replicas == 3
+        assert service.base_image == "python:3.11"
+        assert service.run_as_user == 1000
+        assert service.run_as_group == 1000
+        assert service.mount_sub_path == "/data"
+        assert service.health_endpoint == "/health"
+        assert service.env == {"KEY1": "value1", "KEY2": "value2"}
+
+    def test_build_with_managed_service(self):
+        profile = (
+            ProfileBuilder()
+            .add_managed_service("db", provider="postgres", plan="small")
+            .config("max_connections", 100)
+            .secret("password", "vault://secrets/db-password")
+            .done()
+            .build()
+        )
+
+        assert "db" in profile.run
+        service = profile.run["db"]
+        assert isinstance(service, ManagedServiceConfig)
+        assert service.provider == "postgres"
+        assert service.plan == "small"
+        assert service.config == {"max_connections": 100}
+        assert service.secrets == {"password": "vault://secrets/db-password"}
+
+    def test_build_with_multiple_services(self):
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("npm install")
+            .done()
+            .add_reactive_service("web")
+            .plan(self.TEST_PLAN_ID)
+            .add_step("npm start")
+            .add_port(3000)
+            .done()
+            .add_reactive_service("worker")
+            .plan(self.TEST_PLAN_ID)
+            .add_step("npm run worker")
+            .done()
+            .add_managed_service("redis", provider="redis", plan="micro")
+            .done()
+            .build()
+        )
+
+        assert len(profile.prepare.steps) == 1
+        assert len(profile.run) == 3
+        assert "web" in profile.run
+        assert "worker" in profile.run
+        assert "redis" in profile.run
+
+    def test_profile_to_yaml(self):
+        profile = (
+            ProfileBuilder()
+            .prepare()
+            .add_step("npm install")
+            .done()
+            .add_reactive_service("web")
+            .plan(self.TEST_PLAN_ID)
+            .add_step("npm start")
+            .add_port(3000, public=True)
+            .done()
+            .build()
+        )
+
+        yaml_output = profile.to_yaml()
+
+        assert "schemaVersion: v0.2" in yaml_output
+        assert "prepare:" in yaml_output
+        assert "npm install" in yaml_output
+        assert "run:" in yaml_output
+        assert "web:" in yaml_output
+        assert f"plan: {self.TEST_PLAN_ID}" in yaml_output
+
+    def test_build_reactive_service_without_plan_raises_error(self):
+        with pytest.raises(ValueError, match="requires a plan ID"):
+            (
+                ProfileBuilder()
+                .add_reactive_service("web")
+                .add_step("npm start")
+                .add_port(3000)
+                .done()
+                .build()
+            )
+
+
+class TestReactiveServiceBuilder:
+    TEST_PLAN_ID = 8
+
+    def test_build_reactive_service(self):
+        name, config = (
+            ReactiveServiceBuilder("api")
+            .plan(self.TEST_PLAN_ID)
+            .add_step("npm start")
+            .add_port(8080, public=True)
+            .replicas(2)
+            .build()
+        )
+
+        assert name == "api"
+        assert isinstance(config, ReactiveServiceConfig)
+        assert config.plan == self.TEST_PLAN_ID
+        assert config.replicas == 2
+
+    def test_service_name_property(self):
+        builder = ReactiveServiceBuilder("my-service")
+        assert builder.name == "my-service"
+
+    def test_build_without_plan_raises_error(self):
+        with pytest.raises(ValueError, match="requires a plan ID"):
+            ReactiveServiceBuilder("api").add_step("npm start").build()
+
+
+class TestManagedServiceBuilder:
+    def test_build_managed_service(self):
+        name, config = (
+            ManagedServiceBuilder("cache", "redis", "medium")
+            .config("maxmemory", "256mb")
+            .secrets({"auth_token": "secret"})
+            .build()
+        )
+
+        assert name == "cache"
+        assert isinstance(config, ManagedServiceConfig)
+        assert config.provider == "redis"
+        assert config.plan == "medium"
+        assert config.config == {"maxmemory": "256mb"}
+        assert config.secrets == {"auth_token": "secret"}
+
+
+class TestProfileConfigModels:
+    TEST_PLAN_ID = 8
+
+    def test_step_model(self):
+        step = Step(command="echo hello")
+        assert step.command == "echo hello"
+        assert step.name is None
+
+        step_with_name = Step(command="npm build", name="Build")
+        assert step_with_name.name == "Build"
+
+    def test_port_config_validation(self):
+        port = PortConfig(port=8080, is_public=False)
+        assert port.port == 8080
+
+        with pytest.raises(ValueError):
+            PortConfig(port=0)
+
+        with pytest.raises(ValueError):
+            PortConfig(port=70000)
+
+    def test_path_config_model(self):
+        path = PathConfig(port=3000, path="/api")
+        assert path.port == 3000
+        assert path.path == "/api"
+        assert path.strip_path is None
+
+    def test_network_config_model(self):
+        network = NetworkConfig(
+            ports=[PortConfig(port=3000, is_public=True)],
+            paths=[PathConfig(port=3000, path="/")],
+        )
+        assert len(network.ports) == 1
+        assert len(network.paths) == 1
+
+    def test_reactive_service_config_requires_plan(self):
+        config = ReactiveServiceConfig(plan=self.TEST_PLAN_ID)
+        assert config.plan == self.TEST_PLAN_ID
+        assert config.replicas == 1
+        assert config.steps == []
+        assert config.env is None
+        assert config.network is None
+
+    def test_managed_service_config(self):
+        config = ManagedServiceConfig(provider="postgres", plan="large")
+        assert config.provider == "postgres"
+        assert config.plan == "large"
+
+    def test_stage_config_defaults(self):
+        stage = StageConfig()
+        assert stage.steps == []
+
+    def test_profile_config_camel_case_serialization(self):
+        profile = ProfileConfig()
+        data = profile.model_dump(by_alias=True)
+
+        assert "schemaVersion" in data
+        assert "schema_version" not in data