fix(provisioning,server): exit retry loop on Accepted outcome; await runtime in foreground mode

Two live-AWS regressions, both rooted in earlier work on this branch.

Provisioning retry loop spawned redundant fleets for async providers.
Async providers (EC2Fleet/SpotFleet REQUEST/MAINTAIN) return an
Accepted outcome with zero instances on the first call — fulfilment
finishes out of band and the status poller reconciles it.  The retry
loop interpreted "0 fulfilled + not final" as "retry with a new
batch", so each accepted request grew a second and third fleet on the
side.  Polling later saw one healthy fleet and N-1 empty fleets and
flipped the request to complete_with_error.  Break out of the loop as
soon as an Accepted outcome lands so the single accepted fleet drives
the rest.  No in-tree provider emits RequiresFollowUp today; if one
does later, persist_acquiring becomes reachable again.

Foreground server start re-entered asyncio.run inside the CLI's own
asyncio.run(main()).  daemon_mod.start called _spawn_runtime which
called asyncio.run, raising RuntimeError "asyncio.run cannot be
called from a running event loop".  The exception was caught and
returned exit_code=1, so the subprocess died silently inside 30s and
every rest_api live test hit the "ORB server failed to start" timeout.
handle_server_start now drives the pid + token lifecycle directly in
foreground mode and awaits the runtime coroutine in the existing
event loop.

The retry-loop persist_acquiring tests guarded a write that only ran
between retry attempts.  With Accepted no longer triggering a retry,
the persist path is unreachable; the post-call
update_request_from_provisioning in request_creation_handlers already
stamps IN_PROGRESS (or PARTIAL / FAILED) once the orchestrator
returns.  Replaced with a test that verifies the Accepted outcome
exits the loop after exactly one provider call.

Author: fgogolli

src/orb/application/services/provisioning_orchestration_service.py

@@ -227,6 +227,15 @@ async def execute_provisioning(
                 self._record_provider_failure(selection_result.provider_name)
                 break
 
+            # Async provider accepted the request and is provisioning out of
+            # band; polling owns the final status from here.  Retrying would
+            # create a second fleet alongside the one already provisioning,
+            # which then shows up as a phantom failure when it reports empty.
+            # Break out and let the status-check loop drive the single
+            # accepted fleet to completion.
+            if isinstance(last_result.outcome, Accepted):
+                break
+
             if remaining > 0 and not last_result.is_final:
                 # Partial fulfillment, retry may help — persist ACQUIRING status
                 self._logger.info(

src/orb/interface/server_command_handlers.py

@@ -142,11 +142,41 @@ def _health_url(server_config: Any, ui_config: Any | None) -> str:
 @handle_interface_exceptions(context="server_start", interface_type="cli")
 async def handle_server_start(args) -> dict[str, Any]:
     """Start the server. Daemonized by default; ``--foreground`` to block."""
+    import os
+
     from orb.interface import server_daemon as daemon_mod
 
     runtime, server_config, _ui_config = _build_runtime(args)
     pid_file, log_file, working_dir = _resolve_lifecycle_paths(server_config)
     foreground = getattr(args, "foreground", False)
+
+    if foreground:
+        # Foreground mode runs inside the CLI's own event loop (orb.run.main
+        # is dispatched via asyncio.run).  Routing through daemon_mod.start
+        # in this path would nest a second asyncio.run, which RuntimeError's
+        # and surfaces as exit_code=1 with no actual server having started.
+        # Take the pid + token lock directly here and await the runtime.
+        logger = get_logger(__name__)
+        pid_path = daemon_mod._expand(str(pid_file))
+        log_path = daemon_mod._expand(str(log_file))
+        wd_path = daemon_mod._expand(str(working_dir))
+        wd_path.mkdir(parents=True, exist_ok=True)
+        lock_fd = daemon_mod._acquire_pid_lock(pid_path)
+        try:
+            daemon_mod._write_pid(lock_fd, os.getpid())
+            try:
+                daemon_mod._write_token_file(pid_path)
+            except OSError as exc:
+                logger.warning("loopback handshake file write failed: %s", exc)
+            log_path.parent.mkdir(parents=True, exist_ok=True)
+            result = await runtime()
+            rc = int(result.get("exit_code", 0)) if isinstance(result, dict) else 0
+        finally:
+            pid_path.unlink(missing_ok=True)
+            daemon_mod._cleanup_token_file(pid_path)
+            os.close(lock_fd)
+        return {"pid": os.getpid(), "status": "exited", "exit_code": rc}
+
     return daemon_mod.start(
         pid_file=pid_file,
         log_file=log_file,

tests/unit/application/services/test_provisioning_async_guards.py

@@ -200,141 +200,62 @@ async def _hang(*_args, **_kwargs):
 
 
 # ---------------------------------------------------------------------------
-# asyncio.to_thread tests
+# Accepted-outcome short-circuit
 # ---------------------------------------------------------------------------
 
 
-class TestPersistAcquiringToThread:
-    """_persist_acquiring must run in a worker thread via asyncio.to_thread."""
+class TestAcceptedOutcomeBreaksRetryLoop:
+    """Async providers that return Accepted must not be retried.
 
-    @pytest.mark.asyncio
-    async def test_persist_acquiring_called_via_to_thread(self):
-        """asyncio.to_thread is used when calling _persist_acquiring."""
-        from orb.providers.base.strategy.provider_strategy import ProviderResult
-
-        svc = _make_service(dispatch_timeout=10.0)
-
-        # First attempt: partial (is_final=False) — triggers persist_acquiring
-        # Second attempt: fully fulfilled
-        first_result = ProviderResult.success_result(
-            data={
-                "resource_ids": ["i-partial"],
-                "instances": [{"id": "i-partial"}],
-                "instance_ids": ["i-partial"],
-            },
-            metadata={},
-        )
-        second_result = ProviderResult.success_result(
-            data={
-                "resource_ids": ["i-rest"],
-                "instances": [{"id": "i-rest"}],
-                "instance_ids": ["i-rest"],
-            },
-            metadata={},
-        )
-        svc._provider_selection_port.execute_operation = AsyncMock(
-            side_effect=[first_result, second_result]
-        )
-
-        scheduler = MagicMock()
-        scheduler.format_template_for_provider.return_value = {}
-        cast(MagicMock, svc._container).get.return_value = scheduler
-
-        to_thread_calls: list = []
-
-        original_to_thread = asyncio.to_thread
+    A retry creates a SECOND fleet / batch alongside the one the provider
+    already accepted; downstream status-check then sees one healthy fleet and
+    N-1 empty fleets and flips the request to ``complete_with_error``.
 
-        async def _spy_to_thread(func, *args, **kwargs):
-            to_thread_calls.append(func)
-            return await original_to_thread(func, *args, **kwargs)
-
-        # Build a request that needs 2 instances so a second attempt is triggered
-        request = MagicMock()
-        request.request_id = "req-persist-test"
-        request.requested_count = 2
-        request.metadata = {}
-
-        call_counter = [0]
-
-        def _update_metadata(d):
-            call_counter[0] += 1
-            return request
-
-        request.update_metadata = _update_metadata
-
-        # _persist_acquiring needs update_status
-        request.update_status = MagicMock(return_value=request)
-
-        # Patch asyncio.to_thread in the module under test
-        with patch(
-            "orb.application.services.provisioning_orchestration_service.asyncio.to_thread",
-            side_effect=_spy_to_thread,
-        ):
-            # We don't need the full loop to run — just verify to_thread is called
-            # by exercising the partial-fulfillment path
-            await svc.execute_provisioning(_make_template(), request, _make_selection_result())
-
-        # _persist_acquiring should have been scheduled via to_thread
-        persist_calls = [f for f in to_thread_calls if f == svc._persist_acquiring]
-        assert len(persist_calls) >= 1, (
-            "_persist_acquiring was not dispatched via asyncio.to_thread"
-        )
+    This contract replaces the previous retry-loop tests against
+    ``_persist_acquiring``: the persist hook only ran on the retry path,
+    and the retry path now exits immediately on ``Accepted`` (the only
+    outcome any in-tree provider emits when more attempts could possibly
+    help).  If a future provider emits ``RequiresFollowUp``, the retry
+    persist path becomes reachable again and tests should be restored.
+    """
 
     @pytest.mark.asyncio
-    async def test_persist_acquiring_failure_does_not_abort_loop(self):
-        """If _persist_acquiring raises, the retry loop continues with in-memory state."""
+    async def test_accepted_outcome_exits_loop_after_one_attempt(self):
+        """A single Accepted attempt must end the loop even with remaining > 0."""
         from orb.providers.base.strategy.provider_strategy import ProviderResult
 
         svc = _make_service(dispatch_timeout=10.0)
 
-        # Two partial attempts, then a final one
-        partial_result = ProviderResult.success_result(
-            data={
-                "resource_ids": ["i-p"],
-                "instances": [{"id": "i-p"}],
-                "instance_ids": ["i-p"],
-            },
-            metadata={},
-        )
-        final_result = ProviderResult.success_result(
+        accepted_result = ProviderResult.success_result(
             data={
-                "resource_ids": ["i-f"],
-                "instances": [{"id": "i-f"}],
-                "instance_ids": ["i-f"],
+                "resource_ids": ["fleet-abc"],
+                "instances": [],
+                "instance_ids": [],
             },
+            # fulfillment_final omitted (defaults to False) → Accepted outcome.
             metadata={},
         )
         svc._provider_selection_port.execute_operation = AsyncMock(
-            side_effect=[partial_result, final_result]
+            side_effect=[accepted_result, accepted_result, accepted_result]
         )
 
         scheduler = MagicMock()
         scheduler.format_template_for_provider.return_value = {}
         cast(MagicMock, svc._container).get.return_value = scheduler
 
         request = MagicMock()
-        request.request_id = "req-persist-fail"
+        request.request_id = "req-accepted-once"
         request.requested_count = 2
         request.metadata = {}
         request.update_metadata = lambda d: request
         request.update_status = MagicMock(return_value=request)
 
-        # Make _persist_acquiring return failure (second return value = False)
-        def _failing_persist(req):
-            return req, False
-
-        svc._persist_acquiring = _failing_persist  # type: ignore[method-assign]
-
-        async def _inline_to_thread(func, *args, **kwargs):
-            return func(*args, **kwargs)
-
-        with patch(
-            "orb.application.services.provisioning_orchestration_service.asyncio.to_thread",
-            side_effect=_inline_to_thread,
-        ):
-            await svc.execute_provisioning(_make_template(), request, _make_selection_result())
+        result = await svc.execute_provisioning(
+            _make_template(), request, _make_selection_result()
+        )
 
-        # Loop should have continued despite persist failure
-        cast(MagicMock, svc._logger).warning.assert_called()
-        warning_msgs = " ".join(str(c) for c in cast(MagicMock, svc._logger).warning.call_args_list)
-        assert "ACQUIRING persist failed" in warning_msgs or "persist" in warning_msgs.lower()
+        # Exactly one provider call: the Accepted outcome short-circuits the
+        # loop before another attempt can be made.
+        assert svc._provider_selection_port.execute_operation.await_count == 1
+        # The single accepted fleet is recorded in the returned resource_ids.
+        assert result.resource_ids == ["fleet-abc"]

tests/unit/interface/test_server_command_handlers.py

@@ -4,6 +4,7 @@
 
 import argparse
 import json
+from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -100,21 +101,42 @@ async def test_start_foreground_false_calls_daemon_start(self, _mock_paths, mock
     @pytest.mark.asyncio
     @patch(_BUILD_RUNTIME)
     @patch(_RESOLVE_PATHS, return_value=("/tmp/srv.pid", "/tmp/srv.log", "/tmp"))
-    async def test_start_foreground_true_passes_flag(self, _mock_paths, mock_build_runtime):
+    async def test_start_foreground_awaits_runtime_directly(
+        self, _mock_paths, mock_build_runtime
+    ):
+        """Foreground mode must await the runtime in the caller's event loop.
+
+        Routing through daemon_mod.start would nest a second asyncio.run inside
+        the CLI's own asyncio.run(main()) and the runtime would never reach
+        ``await server.serve()``.  The handler now drives the lock/token
+        lifecycle directly and ``await``s the runtime.
+        """
         server_cfg = _make_server_config()
-        runtime = AsyncMock(return_value={"message": "Server stopped"})
+        runtime = AsyncMock(return_value={"exit_code": 0})
         mock_build_runtime.return_value = (runtime, server_cfg, None)
 
         mock_daemon = MagicMock()
-        mock_daemon.start.return_value = {"pid": 42, "status": "foreground"}
+        mock_daemon._expand.side_effect = lambda p: Path(p)
+        mock_daemon._acquire_pid_lock.return_value = 99
 
-        with patch("orb.interface.server_daemon", mock_daemon, create=True):
+        with (
+            patch("orb.interface.server_daemon", mock_daemon, create=True),
+            patch("os.close"),
+        ):
             from orb.interface.server_command_handlers import handle_server_start
 
-            await handle_server_start(_args(foreground=True))
-
-        call_kwargs = mock_daemon.start.call_args.kwargs
-        assert call_kwargs["foreground"] is True
+            result = await handle_server_start(_args(foreground=True))
+
+        # daemon_mod.start MUST NOT be called in foreground mode — that
+        # would re-enter asyncio.run and crash silently with exit_code=1.
+        mock_daemon.start.assert_not_called()
+        # The runtime coroutine must be awaited in the existing loop.
+        runtime.assert_awaited_once()
+        # The handler still owns lock + token lifecycle.
+        mock_daemon._acquire_pid_lock.assert_called_once()
+        mock_daemon._write_pid.assert_called_once()
+        assert result["status"] == "exited"
+        assert result["exit_code"] == 0
 
     @pytest.mark.asyncio
     @patch(_BUILD_RUNTIME)