Mention failed instance termination in events

src/dstack/_internal/server/background/pipeline_tasks/instances/__init__.py

@@ -485,6 +485,13 @@ async def _apply_process_result(
                 instance_model.termination_reason_message,
             ),
         )
+        for message in result.event_messages:
+            events.emit(
+                session,
+                message,
+                actor=events.SystemActor(),
+                targets=[events.Target.from_model(instance_model)],
+            )
         _emit_instance_health_change_event(
             session=session,
             instance_model=instance_model,

src/dstack/_internal/server/background/pipeline_tasks/instances/common.py

@@ -71,6 +71,8 @@ class ProcessResult:
     new_placement_group_models: list[PlacementGroupModel] = field(default_factory=list)
     schedule_pg_deletion_fleet_id: Optional[uuid.UUID] = None
     schedule_pg_deletion_except_id: Optional[uuid.UUID] = None
+    event_messages: list[str] = field(default_factory=list)
+    """Extra instance events to emit after the update is applied."""
 
 
 async def can_terminate_fleet_instances_on_idle_duration(

src/dstack/_internal/server/background/pipeline_tasks/instances/termination.py

@@ -61,6 +61,10 @@ async def terminate_instance(instance_model: InstanceModel) -> ProcessResult:
             next_step = "Please terminate the instance manually to avoid unexpected charges."
             if _is_to_be_deleted_by_orphan_cleanup(instance_model, job_provisioning_data):
                 next_step = "Orphan cleanup will retry deleting the instance."
+            result.event_messages.append(
+                f"Failed all attempts to terminate instance in the backend. {next_step}"
+                f" Last error: {_format_error(exc)}"
+            )
             logger.error(
                 "Failed all attempts to terminate instance %s. %s Error: %r",
                 instance_model.name,
@@ -106,6 +110,12 @@ async def _terminate_in_backend(
         )
 
 
+def _format_error(exc: Exception) -> str:
+    if isinstance(exc, BackendError):
+        return str(exc) or type(exc).__name__
+    return type(exc).__name__  # avoid leaking internal details
+
+
 def _is_to_be_deleted_by_orphan_cleanup(
     instance_model: InstanceModel, job_provisioning_data: JobProvisioningData
 ) -> bool:

src/tests/_internal/server/background/pipeline_tasks/instances/test_termination.py

@@ -21,6 +21,7 @@
     create_instance,
     create_project,
     get_job_provisioning_data,
+    list_events,
     mock_lease,
 )
 from tests._internal.server.background.pipeline_tasks.instances.helpers import (
@@ -225,6 +226,45 @@ async def test_terminate_on_termination_deadline(
         await session.refresh(instance)
         assert instance.status == InstanceStatus.TERMINATED
 
+    async def test_emits_event_if_all_termination_attempts_failed(
+        self,
+        test_db,
+        session: AsyncSession,
+        worker: InstanceWorker,
+    ):
+        project = await create_project(session=session)
+        instance = await create_instance(
+            session=session,
+            project=project,
+            status=InstanceStatus.TERMINATING,
+        )
+        instance.termination_reason = InstanceTerminationReason.IDLE_TIMEOUT
+        initial_time = dt.datetime(2025, 1, 1, tzinfo=dt.timezone.utc)
+        instance.last_job_processed_at = initial_time
+        instance.last_processed_at = initial_time - dt.timedelta(minutes=1)
+        await session.commit()
+
+        with (
+            time_machine.travel(initial_time + dt.timedelta(minutes=1), tick=False),
+            self.mock_terminate_in_backend(error=BackendError("err")),
+        ):
+            await process_instance(session, worker, instance)
+        assert not any("Failed all attempts" in e.message for e in await list_events(session))
+
+        with (
+            time_machine.travel(initial_time + dt.timedelta(minutes=15, seconds=55), tick=False),
+            self.mock_terminate_in_backend(error=BackendError("quota exceeded")),
+        ):
+            await process_instance(session, worker, instance)
+        await session.refresh(instance)
+        assert instance.status == InstanceStatus.TERMINATED
+        messages = [e.message for e in await list_events(session)]
+        assert any(
+            "Failed all attempts to terminate instance in the backend" in m
+            and "quota exceeded" in m
+            for m in messages
+        )
+
 
 class TestIsDeletedByOrphanCleanup:
     @pytest.mark.parametrize(