feat(jobs): add sweep_orphans startup reconcile to JobService

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

@@ -295,6 +295,43 @@ class JobService(Service):
             result = await session.exec(stmt)
             return list(result.all())
 
+    async def sweep_orphans(self) -> list[UUID]:
+        """Reconcile IN_PROGRESS jobs left behind by a crashed worker.
+
+        Under the default at-most-once policy an in-flight job whose worker
+        died is unrecoverable, so we mark it FAILED with a worker_lost error,
+        stamp finished_timestamp, and append a terminal ``run_failed`` event so
+        a reattacher always sees a clean end. QUEUED jobs are intentionally
+        untouched (at-least-once: they get re-picked by a fresh worker).
+
+        Returns the ids of the jobs transitioned to FAILED.
+        """
+        error_payload = {"type": "worker_lost"}
+        async with session_scope() as session:
+            stmt = select(Job).where(Job.status == JobStatus.IN_PROGRESS)
+            result = await session.exec(stmt)
+            orphans = list(result.all())
+            if not orphans:
+                return []
+            now = datetime.now(timezone.utc)
+            for job in orphans:
+                job.status = JobStatus.FAILED
+                job.error = dict(error_payload)
+                job.finished_timestamp = now
+                session.add(job)
+                # Terminal milestone on the durable log so reattach sees an end.
+                max_stmt = select(func.max(JobEvent.seq)).where(JobEvent.job_id == job.job_id)
+                current_max = (await session.exec(max_stmt)).one()
+                event = JobEvent(
+                    job_id=job.job_id,
+                    seq=(current_max or 0) + 1,
+                    event_type="run_failed",
+                    payload=dict(error_payload),
+                )
+                session.add(event)
+            await session.flush()
+            return [job.job_id for job in orphans]
+
     async def get_latest_jobs_by_asset_ids(self, asset_ids: Sequence[UUID | str]) -> dict[UUID, Job]:
         """Get the latest job for each asset ID in a single batch query.
 

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

@@ -135,3 +135,50 @@ async def test_unconsumed_signals_empty_when_none_written():
     await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
     assert await service.unconsumed_signals(job_id) == []
+
+
+@pytest.mark.usefixtures("client")
+async def test_sweep_orphans_fails_in_progress_jobs():
+    """On startup, any IN_PROGRESS job is an orphan from a crashed worker.
+
+    The default at-most-once policy marks it FAILED with a worker_lost error.
+    QUEUED jobs are left alone (at-least-once: they get re-picked).
+    """
+    service = JobService()
+    orphan = uuid4()
+    queued = uuid4()
+    flow_id = uuid4()
+    user_id = uuid4()
+    await service.create_job(job_id=orphan, flow_id=flow_id, user_id=user_id)
+    await service.update_job_status(orphan, JobStatus.IN_PROGRESS)
+    await service.create_job(job_id=queued, flow_id=flow_id, user_id=user_id)
+
+    swept = await service.sweep_orphans()
+    assert orphan in swept
+    assert queued not in swept
+
+    orphan_job = await service.get_job_by_job_id(orphan)
+    assert orphan_job.status == JobStatus.FAILED
+    assert orphan_job.error == {"type": "worker_lost"}
+    assert orphan_job.finished_timestamp is not None
+
+    # A reattacher must see a clean terminal event on the durable log.
+    orphan_events = await service.read_events(orphan)
+    assert len(orphan_events) == 1
+    assert orphan_events[0].event_type == "run_failed"
+    assert orphan_events[0].payload == {"type": "worker_lost"}
+
+    queued_job = await service.get_job_by_job_id(queued)
+    assert queued_job.status == JobStatus.QUEUED
+    # QUEUED jobs are untouched: no terminal event.
+    assert await service.read_events(queued) == []
+
+
+@pytest.mark.usefixtures("client")
+async def test_sweep_orphans_noop_when_clean():
+    service = JobService()
+    job_id = uuid4()
+    await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
+    await service.update_job_status(job_id, JobStatus.COMPLETED, finished_timestamp=True)
+
+    assert await service.sweep_orphans() == []