fix: more merge conflict resolutions

Author: ikemilian-lewis

src/orb/application/commands/request_creation_handlers.py

@@ -552,9 +552,9 @@ def _persist_return_follow_up_context(
         if not isinstance(provider_data, dict) or not provider_data:
             return
 
-        updated_request = with_request_follow_up_context(request, provider_data)
-
         with self.uow_factory.create_unit_of_work() as uow:
+            current_request = uow.requests.get_by_id(request.request_id) or request
+            updated_request = with_request_follow_up_context(current_request, provider_data)
             events = uow.requests.save(updated_request)
             for event in events or []:
                 self.event_publisher.publish(event)  # type: ignore[union-attr]
@@ -631,18 +631,3 @@ async def _update_request_to_terminating(self, request: Any) -> None:
             RequestStatus.IN_PROGRESS,
             "Termination accepted: waiting for instances to reach terminated state",
         )
-
-    async def _update_request_to_in_progress_with_message(self, request: Any, message: str) -> None:
-        """Persist an in-progress return message after submit without claiming completion."""
-        try:
-            await self._update_request_status(
-                request,
-                RequestStatus.IN_PROGRESS,
-                message,
-            )
-        except Exception as update_error:
-            self.logger.error(
-                "Failed to update request status to in_progress: %s",
-                update_error,
-                exc_info=True,
-            )

src/orb/application/services/machine_sync_service.py

@@ -237,7 +237,7 @@ def _resolve_return_resource_id(
             if not isinstance(pending_cleanup, dict):
                 continue
 
-            resource_id = pending_cleanup.get("vmss_name")
+            resource_id = pending_cleanup.get("resource_id")
             if resource_id in (None, ""):
                 continue
 
@@ -393,10 +393,8 @@ async def sync_machines_with_provider(
                         #     Stamp a terminal "Terminated" if provider didn't
                         #     supply something more specific (EC2's StateReason).
                         #   - Otherwise pass through whatever provider returned.
-                        from orb.domain.machine.machine_status import MachineStatus as _MS
-
                         _new_reason = provider_machine.status_reason
-                        if provider_machine.status == _MS.TERMINATED:
+                        if provider_machine.status == MachineStatus.TERMINATED:
                             if not _new_reason or "in progress" in (_new_reason or "").lower():
                                 _new_reason = "Terminated"
                         machine_data["status_reason"] = _new_reason

src/orb/application/services/provider_validation_service.py

@@ -2,15 +2,14 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from orb.domain.base.exceptions import ApplicationError
 
 if TYPE_CHECKING:
     from orb.domain.template.template_aggregate import Template
 
 from orb.domain.base.ports import ContainerPort, LoggingPort, ProviderSelectionPort
-from orb.domain.base.ports.provider_validation_port import ProviderValidationPort
 from orb.domain.base.results import ProviderSelectionResult
 
 
@@ -22,12 +21,10 @@ def __init__(
         container: ContainerPort,
         logger: LoggingPort,
         provider_selection_port: ProviderSelectionPort,
-        validator: Optional[ProviderValidationPort] = None,
     ) -> None:
         self._container = container
         self.logger = logger
         self._provider_selection_port = provider_selection_port
-        self._validator = validator
 
     async def select_and_validate_provider(self, template: Template) -> ProviderSelectionResult:
         """Select provider and validate template compatibility."""
@@ -38,13 +35,6 @@ async def select_and_validate_provider(self, template: Template) -> ProviderSele
             selection_result.selection_reason,
         )
 
-        if self._validator is not None:
-            template_dict = template if isinstance(template, dict) else vars(template)
-            result = self._validator.validate_template_configuration(template_dict)
-            if not result.get("valid", True):
-                errors = result.get("errors", [])
-                raise ApplicationError("; ".join(errors))
-
         validation_result = self._provider_selection_port.validate_template_requirements(
             template,
             selection_result.provider_name,

src/orb/application/services/request_status_service.py

@@ -20,6 +20,7 @@
 import dataclasses
 from typing import Optional, Tuple
 
+from orb.application.services.request_follow_up_context import get_request_follow_up_context
 from orb.domain.base import UnitOfWorkFactory
 from orb.domain.base.exceptions import ProviderContractError
 from orb.domain.base.ports.logging_port import LoggingPort
@@ -123,18 +124,8 @@ def _determine_return_status(
         """Determine return request status from machine termination states."""
         db_machine_count = len(db_machines)
         follow_up_pending_message = "Return in progress: awaiting provider follow-up cleanup"
-
-        if provider_metadata.get("termination_follow_up_failed"):
-            details = provider_metadata.get("termination_follow_up_details", [{}])
-            error = details[0].get("last_delete_error") if details else None
-            message = "Return request failed: provider follow-up cleanup failed"
-            if error:
-                message = f"{message}: {error}"
-            return RequestStatus.FAILED.value, message
-
-        termination_follow_up_pending = bool(
-            provider_metadata.get("termination_follow_up_pending", False)
-        )
+        follow_up_context = get_request_follow_up_context(request)
+        termination_follow_up_pending = follow_up_context.get("follow_up_kind") == "termination"
 
         # For return requests: empty provider_machines *with* DB records means all
         # instances are gone from AWS — genuinely terminated.  But if we have

src/orb/application/services/spot_placement_execution.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 from dataclasses import dataclass
-from typing import Any, Awaitable, Callable, Mapping
+from typing import Any, Callable, Mapping
 
 from orb.application.services.spot_placement_planner import PlacementPlanEntry
 from orb.domain.base.exceptions import DomainException
@@ -97,91 +97,6 @@ def build_planned_execution_metadata(
 class SpotPlacementExecutionService:
     """Execute a placement plan via provider callbacks."""
 
-    async def execute_plan_async(
-        self,
-        plan: list[PlacementPlanEntry],
-        total_count: int,
-        build_child_template: Callable[[PlacementPlanEntry], Any],
-        build_child_request: Callable[[int, int], Any],
-        launch_child: Callable[[Any, Any], Awaitable[Mapping[str, Any]]],
-        is_capacity_like_failure: Callable[[dict[str, Any]], bool],
-    ) -> SpotPlacementExecutionSummary:
-        """Async variant of ``execute_plan`` for native async provider handlers."""
-        resource_ids: list[str] = []
-        instances: list[dict[str, Any]] = []
-        child_results: list[dict[str, Any]] = []
-        failed_subplans: list[dict[str, Any]] = []
-        carryover = 0
-        fulfilled = 0
-        terminated_early = False
-        terminal_error_message: str | None = None
-
-        for idx, plan_entry in enumerate(plan):
-            requested_for_entry = plan_entry.planned_count + carryover
-            if requested_for_entry <= 0:
-                continue
-
-            child_template = build_child_template(plan_entry)
-            child_request = build_child_request(requested_for_entry, idx)
-            try:
-                raw_result = await launch_child(child_request, child_template)
-            except DomainException as exc:
-                raw_result = self._domain_exception_child_result(exc)
-
-            child_result = self._normalize_child_result(
-                plan_entry=plan_entry,
-                requested_count=requested_for_entry,
-                raw_result=raw_result,
-            )
-            child_results.append(child_result)
-
-            if child_result["success"]:
-                resource_ids.extend(child_result["resource_ids"])
-                instances.extend(child_result["instances"])
-                fulfilled_for_child = child_result["fulfilled_count"]
-                fulfilled += fulfilled_for_child
-                carryover = max(requested_for_entry - fulfilled_for_child, 0)
-                if carryover == 0:
-                    continue
-
-                failed_subplans.append(child_result)
-                if is_capacity_like_failure(child_result):
-                    continue
-
-                terminated_early = True
-                terminal_error_message = child_result["error_message"]
-                break
-
-            failed_subplans.append(child_result)
-            if is_capacity_like_failure(child_result):
-                carryover = requested_for_entry
-                continue
-
-            if resource_ids or instances:
-                terminated_early = True
-                terminal_error_message = child_result["error_message"]
-                break
-
-            return SpotPlacementExecutionSummary(
-                resource_ids=[],
-                instances=[],
-                child_results=child_results,
-                failed_subplans=failed_subplans,
-                unfulfilled_count=total_count,
-                terminated_early=True,
-                terminal_error_message=child_result["error_message"],
-            )
-
-        return SpotPlacementExecutionSummary(
-            resource_ids=resource_ids,
-            instances=instances,
-            child_results=child_results,
-            failed_subplans=failed_subplans,
-            unfulfilled_count=max(total_count - fulfilled, carryover),
-            terminated_early=terminated_early,
-            terminal_error_message=terminal_error_message,
-        )
-
     def execute_plan(
         self,
         plan: list[PlacementPlanEntry],

src/orb/cli/main.py

@@ -219,8 +219,7 @@ async def main() -> None:
             else:
                 # Rich console output can wrap long structured strings, which may break
                 # consumers that try to parse stdout as exact JSON.
-                if not args.quiet:
-                    print(formatted_output)
+                print(formatted_output)
 
             if exit_code != 0:
                 sys.exit(exit_code)

src/orb/domain/base/ports/provider_registry_port.py

@@ -50,6 +50,11 @@ def generate_provider_name(cls, config: dict[str, Any]) -> str:
         """Generate a provider instance name from provider config."""
         ...
 
+    @classmethod
+    def get_ui_column_schema(cls) -> list[Any]:
+        """Return provider-specific UI column descriptors."""
+        ...
+
 
 class ProviderRegistryPort(ABC):
     """Port that the application layer uses to interact with the provider registry.

src/orb/infrastructure/scheduler/base/strategy.py

@@ -22,6 +22,7 @@
     from orb.application.services.provider_registry_service import ProviderRegistryService
     from orb.domain.base.ports.configuration_port import ConfigurationPort
     from orb.domain.base.ports.logging_port import LoggingPort
+    from orb.domain.template.template_aggregate import Template
 
 
 class BaseSchedulerStrategy(SchedulerPort, ABC):
@@ -247,23 +248,23 @@ def get_config_directory(self) -> str:
         return self._coalesce_directory(
             config_override=getattr(self.config_manager.app_config.scheduler, "config_dir", None),
             env_var_name="CONFIG_DIR",
-            default_factory=lambda: self._get_platform_config_dir(),
+            default_factory=self._get_platform_config_dir,
         )
 
     def get_working_directory(self) -> str:
         """Get working directory with coalesce pattern."""
         return self._coalesce_directory(
             config_override=getattr(self.config_manager.app_config.scheduler, "work_dir", None),
             env_var_name="WORK_DIR",
-            default_factory=lambda: self._get_platform_work_dir(),
+            default_factory=self._get_platform_work_dir,
         )
 
     def get_logs_directory(self) -> str:
         """Get logs directory with coalesce pattern."""
         return self._coalesce_directory(
             config_override=getattr(self.config_manager.app_config.scheduler, "log_dir", None),
             env_var_name="LOG_DIR",
-            default_factory=lambda: self._get_platform_logs_dir(),
+            default_factory=self._get_platform_logs_dir,
         )
 
     def get_log_level(self) -> str:
@@ -416,7 +417,7 @@ def format_template_for_display(self, template: TemplateDTO) -> dict[str, Any]:
         """Default implementation - clean to_dict without scheduler-specific formatting."""
         return template.to_dict()
 
-    def format_template_for_provider(self, template: TemplateDTO) -> dict[str, Any]:
+    def format_template_for_provider(self, template: "TemplateDTO | Template") -> dict[str, Any]:
         """Format either a stored template DTO or a domain template for provider operations."""
         if isinstance(template, TemplateDTO):
             return template.to_template_config()

src/orb/infrastructure/template/dtos.py

@@ -143,23 +143,67 @@ def to_template_config(self) -> dict[str, Any]:
     @classmethod
     def from_domain(cls, template) -> "TemplateDTO":
         """Convert domain template to DTO."""
-        template_data = (
-            template.model_dump() if hasattr(template, "model_dump") else vars(template)
-        )
-        dto_field_names = cls.model_fields.keys()
-        dto_data = {
-            field_name: template_data[field_name]
-            for field_name in dto_field_names
-            if field_name in template_data and field_name != "provider_config"
-        }
-
-        provider_type = dto_data.get("provider_type")
+        provider_type = getattr(template, "provider_type", None)
+        provider_config: Optional[BaseModel] = None
         if provider_type:
-            dto_data["provider_config"] = TemplateExtensionRegistry.create_extension_config(
+            template_data = (
+                template.model_dump() if hasattr(template, "model_dump") else vars(template)
+            )
+            provider_config = TemplateExtensionRegistry.create_extension_config(
                 str(provider_type), template_data
             )
 
-        return cls.model_validate(dto_data)
+        return cls(
+            # Core fields
+            template_id=template.template_id,
+            name=getattr(template, "name", None),
+            description=getattr(template, "description", None),
+            # Instance configuration
+            image_id=getattr(template, "image_id", None),
+            max_instances=getattr(template, "max_instances", 1),
+            # Machine types configuration
+            machine_types=getattr(template, "machine_types", {}),
+            machine_types_ondemand=getattr(template, "machine_types_ondemand", {}),
+            machine_types_priority=getattr(template, "machine_types_priority", {}),
+            # Network configuration
+            subnet_ids=getattr(template, "subnet_ids", []),
+            security_group_ids=getattr(template, "security_group_ids", []),
+            network_zones=getattr(template, "network_zones", []),
+            public_ip_assignment=getattr(template, "public_ip_assignment", None),
+            # Pricing and allocation
+            price_type=getattr(template, "price_type", "ondemand"),
+            allocation_strategy=getattr(template, "allocation_strategy", None),
+            max_price=getattr(template, "max_price", None),
+            # Storage configuration
+            root_device_volume_size=getattr(template, "root_device_volume_size", None),
+            volume_type=getattr(template, "volume_type", None),
+            iops=getattr(template, "iops", None),
+            throughput=getattr(template, "throughput", None),
+            storage_encryption=getattr(template, "storage_encryption", None),
+            encryption_key=getattr(template, "encryption_key", None),
+            # Access and security
+            key_name=getattr(template, "key_name", None),
+            user_data=getattr(template, "user_data", None),
+            instance_profile=getattr(template, "instance_profile", None),
+            # Advanced configuration
+            monitoring_enabled=getattr(template, "monitoring_enabled", None),
+            # Metadata
+            tags=getattr(template, "tags", {}),
+            metadata=getattr(template, "metadata", {}),
+            provider_config=provider_config,
+            provider_data=getattr(template, "provider_data", {}),
+            # Provider identification
+            provider_type=provider_type,
+            provider_name=getattr(template, "provider_name", None),
+            provider_api=getattr(template, "provider_api", None),
+            # Timestamps
+            created_at=getattr(template, "created_at", None),
+            updated_at=getattr(template, "updated_at", None),
+            # Active status
+            is_active=getattr(template, "is_active", True),
+            # Legacy fields
+            version=getattr(template, "version", None),
+        )
 
 
 @dataclass

src/orb/providers/aws/configuration/template_extension.py

@@ -77,14 +77,6 @@ class AWSTemplateExtensionConfig(BaseModel):
         None, description="AWS Context field for EC2 Fleet, ASG, and Spot Fleet"
     )
 
-    # AWS DTO-only fields preserved by TemplateDTO.provider_config round-trips.
-    launch_template_id: Optional[str] = Field(
-        None, description="EC2 Launch Template ID for provider-specific template storage"
-    )
-    abis_instance_requirements: Optional[dict[str, Any]] = Field(
-        None, description="Attribute-based instance selection requirements"
-    )
-
     @field_validator("subnet_ids")
     @classmethod
     def validate_subnet_ids(cls, v: Optional[list[str]]) -> Optional[list[str]]: