feat(background-execution): emit worker_lost + orphan reconciliation metrics

src/backend/base/langflow/services/background_execution/metrics.py

@@ -7,6 +7,16 @@ from lfx.log.logger import logger
 from langflow.services.deps import get_telemetry_service
 
 
+def current_backend() -> str:
+    """Best-effort: 'scaled' when the redis job queue is active, else 'default'. Never raises."""
+    try:
+        from langflow.services.deps import get_settings_service
+
+        return "scaled" if get_settings_service().settings.background_backend_is_scaled else "default"
+    except Exception:  # noqa: BLE001
+        return "default"
+
+
 def _emit(fn_name: str, name: str, labels: dict, value: float | None = None) -> None:
     try:
         ot = get_telemetry_service().ot

src/backend/base/langflow/services/background_execution/redis_backend.py

@@ -21,6 +21,7 @@ import contextlib
 import uuid
 from typing import TYPE_CHECKING, Any
 
+from langflow.services.background_execution import metrics as bg_metrics
 from langflow.services.background_execution.redis_queue import RedisJobClaimQueue
 from langflow.services.database.models.jobs.model import JobStatus, SignalType
 from langflow.services.job_queue.service import RedisQueueWrapper
@@ -200,6 +201,10 @@ class RedisBackgroundQueue:
             # (with a finished timestamp) explicitly.
             await self._job_service.update_job_status(job.job_id, JobStatus.FAILED, finished_timestamp=True)
             await self._job_service.set_error(job.job_id, {"type": "worker_lost"})
+            # One emit pair per reconciled orphan (best-effort, never raises).
+            # This path only runs in scaled mode, so the backend label is literal.
+            bg_metrics.emit_job_failed(reason="worker_lost", backend="scaled")
+            bg_metrics.emit_orphan_reconciled(backend="scaled")
             await self.claim_queue.complete(job_id)
         return requeued
 

src/backend/base/langflow/services/jobs/service.py

@@ -569,11 +569,20 @@ class JobService(Service):
                 session.add(job)
                 reconciled.append(job.job_id)
             await session.flush()
+        # Function-local import: background_execution imports from jobs, so a
+        # module-level import here would risk a cycle. The metrics module itself
+        # only pulls in deps + logger, so the local import is cheap and safe.
+        from langflow.services.background_execution import metrics as bg_metrics
+
+        backend = bg_metrics.current_backend()
         # Append the terminal milestone via append_event (its own session) so the
         # IntegrityError/seq-collision retry applies: a seq collision with a
         # concurrent appender can no longer roll back the whole sweep.
         for job_id in reconciled:
             await self.append_event(job_id, "run_failed", dict(error_payload))
+            # One emit pair per reconciled orphan (best-effort, never raises).
+            bg_metrics.emit_job_failed(reason="worker_lost", backend=backend)
+            bg_metrics.emit_orphan_reconciled(backend=backend)
         return reconciled
 
     async def get_latest_jobs_by_asset_ids(self, asset_ids: Sequence[UUID | str]) -> dict[UUID, Job]:

src/backend/tests/unit/services/jobs/test_sweep_liveness.py

@@ -62,3 +62,41 @@ async def test_sweep_still_fails_heartbeatless_in_progress():
     events = await service.read_events(orphan)
     assert len(events) == 1
     assert events[0].event_type == "run_failed"
+
+
+@pytest.mark.usefixtures("client")
+async def test_sweep_emits_worker_lost_metrics_without_raising():
+    """Reconciling a stale-heartbeat orphan fires the worker_lost emit pair.
+
+    The emit helpers are best-effort (they swallow their own errors), so the
+    contract here is: the sweep still reconciles the orphan AND the emit calls
+    do not raise into the sweep. We do not assert OTel counter values (the
+    metrics are write-only); a clean run that preserves the existing FAILED /
+    worker_lost behavior is the proof.
+    """
+    service = JobService()
+    stale = uuid4()
+    fresh = uuid4()
+
+    # Stale heartbeat -> real orphan that must be reconciled (and emitted for).
+    await service.create_job(job_id=stale, flow_id=uuid4(), user_id=uuid4())
+    await service.update_job_status(stale, JobStatus.IN_PROGRESS)
+    old = (datetime.now(timezone.utc) - timedelta(seconds=300)).isoformat()
+    await service.update_job_metadata(stale, {"owner": "worker-dead", "heartbeat_at": old})
+
+    # Fresh heartbeat -> liveness gate keeps it out of reconciliation (no emit).
+    await service.create_job(job_id=fresh, flow_id=uuid4(), user_id=uuid4())
+    await service.update_job_status(fresh, JobStatus.IN_PROGRESS)
+    await service.heartbeat(fresh, owner="worker-live")
+
+    swept = await service.sweep_orphans(lease_ttl_s=30.0)
+
+    assert stale in swept
+    assert fresh not in swept
+
+    stale_job = await service.get_job_by_job_id(stale)
+    assert stale_job.status == JobStatus.FAILED
+    assert (stale_job.error or {}).get("type") == "worker_lost"
+
+    fresh_job = await service.get_job_by_job_id(fresh)
+    assert fresh_job.status == JobStatus.IN_PROGRESS