fix(background): terminal events on TIMED_OUT/CANCELLED + late-stop result consistency

execute_with_status writes the TIMED_OUT/CANCELLED status but no durable error
blob or terminal job_events row, breaking the design invariant that every
terminal path writes result/error + a terminal event. The runner now backfills
run_timed_out (+error) / run_cancelled in its finally. A late stop that wins over
a racing completion clears the completed-run result and sets error=cancelled so a
CANCELLED row never carries a completed result.

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

@@ -111,8 +111,41 @@ class JobRunner:
             with contextlib.suppress(Exception):
                 if await asyncio.shield(self._reconcile_stop(job_id)):
                     await logger.adebug(f"Background job {job_id} reconciled to CANCELLED after a racing stop")
+            # Every terminal path writes result/error + a terminal job_events row
+            # (design §8). COMPLETED/FAILED already do via _drive; TIMED_OUT and
+            # CANCELLED do not, so backfill their error blob + terminal milestone
+            # here (after the stop reconcile so a late-stop CANCELLED is included).
+            with contextlib.suppress(Exception):
+                await asyncio.shield(self._finalize_terminal_event(job_id))
             await self._bus.close(str(job_id))
 
+    async def _finalize_terminal_event(self, job_id: UUID) -> None:
+        """Backfill the error blob + terminal event for TIMED_OUT / CANCELLED.
+
+        ``execute_with_status`` writes the TIMED_OUT/CANCELLED status but no
+        durable error blob or terminal ``job_events`` row, so a consumer keying
+        on a terminal event TYPE (not just stream close) would not find one. This
+        appends ``run_timed_out`` / ``run_cancelled`` and, for a CANCELLED row
+        that a racing completion left with a populated ``result``, clears the
+        result and sets ``error={type: cancelled}`` so the terminal state is
+        internally consistent (no CANCELLED row carrying a completed-run result).
+        """
+        job = await self._jobs.get_job_by_job_id(job_id)
+        if job is None:
+            return
+        if job.status == JobStatus.TIMED_OUT:
+            if job.error is None:
+                await self._jobs.set_error(job_id, {"type": "timed_out"})
+            await self._jobs.append_event(job_id, "run_timed_out", {"type": "timed_out"})
+        elif job.status == JobStatus.CANCELLED:
+            # A late stop that won over a genuine completion may have left a
+            # completed-run result behind — overwrite it so the row is consistent.
+            if job.result is not None:
+                await self._jobs.set_result(job_id, None)
+            if job.error is None:
+                await self._jobs.set_error(job_id, {"type": "cancelled"})
+            await self._jobs.append_event(job_id, "run_cancelled", {"type": "cancelled"})
+
     def _start_heartbeat(self, job_id: UUID) -> asyncio.Task | None:
         """Spawn the periodic heartbeat task for a run (None when owner unset).
 

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

@@ -225,9 +225,13 @@ class JobService(Service):
             await session.flush()
             return job
 
-    async def set_result(self, job_id: UUID, result: dict) -> Job | None:
+    async def set_result(self, job_id: UUID, result: dict | None) -> Job | None:
         """Persist the durable terminal result blob for a job.
 
+        ``None`` clears a previously-written result (used when a late stop
+        reconciles a racing completion to CANCELLED and the completed-run result
+        must not linger on the terminal row).
+
         Returns the updated Job, or None if the row does not exist.
         """
         async with session_scope() as session:

src/backend/tests/unit/services/background_execution/test_runner.py

@@ -99,6 +99,91 @@ async def test_runner_finalizes_failed_and_writes_error(active_user):
     assert job.error is not None
 
 
+async def test_runner_timeout_writes_terminal_event_and_error(active_user):
+    """TIMED_OUT must write an error blob AND a terminal job_events row (design §8)."""
+    job_service = get_job_service()
+    job_id = await _make_job(uuid4(), active_user.id)
+
+    async def source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]:
+        yield _frame("build_start", {})
+        # Run past the job timeout so asyncio.wait_for raises TimeoutError.
+        await asyncio.sleep(5)
+        yield _frame("end", {})
+
+    bus = InMemoryLiveBus()
+    adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t"))
+    runner = JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source, job_timeout=0.2)
+
+    await asyncio.wait_for(runner.run(job_id=job_id, source_kwargs={}), timeout=10)
+
+    job = await job_service.get_job_by_job_id(job_id)
+    assert job.status == JobStatus.TIMED_OUT
+    # Error blob is populated on timeout (was NULL before the fix).
+    assert job.error is not None
+    assert job.error.get("type") == "timed_out"
+    # A terminal milestone is on the durable log.
+    events = await job_service.read_events(job_id)
+    assert any(e.event_type == "run_timed_out" for e in events)
+
+
+async def test_runner_cancel_writes_terminal_event(active_user):
+    """CANCELLED must write a terminal job_events row (design §8)."""
+    job_service = get_job_service()
+    job_id = await _make_job(uuid4(), active_user.id)
+
+    from langflow.services.database.models.jobs.model import SignalType
+
+    gate = asyncio.Event()
+
+    async def source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]:
+        yield _frame("build_start", {})
+        await gate.wait()
+        yield _frame("end_vertex", {"id": "n1"})
+        yield _frame("end", {})
+
+    bus = InMemoryLiveBus()
+    adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t"))
+    runner = JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source)
+
+    run_task = asyncio.create_task(runner.run(job_id=job_id, source_kwargs={}))
+    await asyncio.sleep(0.1)
+    await job_service.write_signal(job_id, SignalType.STOP)
+    gate.set()
+    await asyncio.wait_for(run_task, timeout=5)
+
+    job = await job_service.get_job_by_job_id(job_id)
+    assert job.status == JobStatus.CANCELLED
+    events = await job_service.read_events(job_id)
+    assert any(e.event_type == "run_cancelled" for e in events)
+
+
+async def test_late_stop_after_completion_clears_result(active_user):
+    """A stop reconciled to CANCELLED after a genuine completion must not keep result."""
+    job_service = get_job_service()
+    job_id = await _make_job(uuid4(), active_user.id)
+
+    from langflow.services.database.models.jobs.model import SignalType
+
+    async def source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]:
+        # The flow completes cleanly (set_result + COMPLETED), but a STOP lands in
+        # the race window before the runner's terminal reconcile.
+        await job_service.write_signal(job_id, SignalType.STOP)
+        yield _frame("end", {})
+
+    bus = InMemoryLiveBus()
+    adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t"))
+    runner = JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source)
+
+    await runner.run(job_id=job_id, source_kwargs={})
+
+    job = await job_service.get_job_by_job_id(job_id)
+    # Stop wins over the racing completion...
+    assert job.status == JobStatus.CANCELLED
+    # ...and the terminal state is internally consistent: no completed result blob.
+    assert job.result is None
+    assert (job.error or {}).get("type") == "cancelled"
+
+
 async def test_runner_stops_at_signal_boundary(active_user):
     job_service = get_job_service()
     job_id = await _make_job(uuid4(), active_user.id)