test(jobs): hard-proof JobService store methods on real sqlite and postgres

src/backend/tests/unit/background_execution/conftest.py

@@ -7,9 +7,9 @@ fakes:
   (only when ``LANGFLOW_TEST_DATABASE_URI`` is set; CI always sets it).
 - ``hard_proof_redis_url`` yields a real Redis URL from ``LANGFLOW_TEST_REDIS_URL``
   (skips when unset), flushed clean before and after each test.
-
-Phase 0 keeps these minimal: they yield connectable URLs. Later phases extend
-``hard_proof_db_url`` to create the background-execution tables.
+- ``hard_proof_job_service`` runs the real Alembic migrations against
+  ``hard_proof_db_url`` and binds ``session_scope()`` to it, so store methods are
+  exercised on both real SQLite and real Postgres.
 """
 
 from __future__ import annotations
@@ -25,6 +25,8 @@ import pytest
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
 
+    from langflow.services.jobs.service import JobService
+
 
 def _async_pg_url(raw: str) -> str:
     """Normalize a Postgres URL to the async asyncpg dialect (the driver we install)."""
@@ -60,6 +62,43 @@ async def hard_proof_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[st
         yield _async_pg_url(raw)
 
 
+@pytest.fixture
+async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobService, None]:
+    """Yield a JobService whose ``session_scope()`` is bound to a real, migrated DB.
+
+    Runs the real Alembic migrations against ``hard_proof_db_url`` (sqlite + postgres)
+    through the production ``DatabaseService`` path, then swaps that service into the
+    service manager so ``JobService``'s ``session_scope()`` resolves to it. This proves
+    the store methods on BOTH real SQLite and real Postgres, and that the migrations
+    themselves apply on both engines.
+    """
+    from langflow.services.database.factory import DatabaseServiceFactory
+    from langflow.services.deps import get_settings_service
+    from langflow.services.jobs.service import JobService
+    from lfx.services.manager import get_service_manager
+    from lfx.services.schema import ServiceType
+
+    manager = get_service_manager()
+    settings_service = get_settings_service()
+    original_url = settings_service.settings.database_url
+    original_db_service = manager.services.pop(ServiceType.DATABASE_SERVICE, None)
+
+    settings_service.settings.database_url = hard_proof_db_url
+    db_service = DatabaseServiceFactory().create(settings_service)
+    manager.services[ServiceType.DATABASE_SERVICE] = db_service
+
+    try:
+        await db_service.run_migrations()
+        yield JobService()
+    finally:
+        manager.services.pop(ServiceType.DATABASE_SERVICE, None)
+        with contextlib.suppress(Exception):
+            await db_service.teardown()
+        settings_service.settings.database_url = original_url
+        if original_db_service is not None:
+            manager.services[ServiceType.DATABASE_SERVICE] = original_db_service
+
+
 @pytest.fixture
 async def hard_proof_redis_url() -> AsyncGenerator[str, None]:
     """Yield a real Redis URL from ``LANGFLOW_TEST_REDIS_URL``, flushed clean.

src/backend/tests/unit/background_execution/test_store_hard_proof.py

@@ -0,0 +1,122 @@
+"""Hard-proof store tests: JobService methods against REAL SQLite and REAL Postgres.
+
+These bind ``session_scope()`` to a real, migrated database (via the
+``hard_proof_job_service`` fixture) so every store method is exercised on both
+engines, not just the temp-file SQLite the ``client`` fixture provides. Running
+the real Alembic migrations against both engines here also re-proves the schema
+applies on Postgres (JSONB columns, the execution_signal_type_enum, the
+uq_job_events_job_id_seq unique constraint).
+"""
+
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from langflow.services.database.models.jobs.model import JobStatus, SignalType
+
+
+@pytest.mark.hard_proof
+@pytest.mark.no_blockbuster
+async def test_set_result_and_error_persist(hard_proof_job_service) -> None:
+    service = hard_proof_job_service
+    job_id = uuid4()
+    await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
+
+    await service.set_result(job_id, {"output_text": "hi", "session_id": "s1"})
+    await service.set_error(job_id, {"type": "boom", "message": "x"})
+
+    fetched = await service.get_job_by_job_id(job_id)
+    assert fetched.result == {"output_text": "hi", "session_id": "s1"}
+    assert fetched.error == {"type": "boom", "message": "x"}
+
+
+@pytest.mark.hard_proof
+@pytest.mark.no_blockbuster
+async def test_append_and_read_events_ordered(hard_proof_job_service) -> None:
+    service = hard_proof_job_service
+    job_id = uuid4()
+    await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
+
+    assert await service.append_event(job_id, "e1", {"i": 1}) == 1
+    assert await service.append_event(job_id, "e2", {"i": 2}) == 2
+    assert await service.append_event(job_id, "e3", {"i": 3}) == 3
+
+    tail = await service.read_events(job_id, after_seq=1)
+    assert [e.seq for e in tail] == [2, 3]
+    assert tail[0].payload == {"i": 2}
+
+
+@pytest.mark.hard_proof
+@pytest.mark.no_blockbuster
+async def test_append_event_seq_is_per_job(hard_proof_job_service) -> None:
+    service = hard_proof_job_service
+    job_a = uuid4()
+    job_b = uuid4()
+    await service.create_job(job_id=job_a, flow_id=uuid4(), user_id=uuid4())
+    await service.create_job(job_id=job_b, flow_id=uuid4(), user_id=uuid4())
+
+    assert await service.append_event(job_a, "x", {}) == 1
+    assert await service.append_event(job_b, "x", {}) == 1
+    assert await service.append_event(job_a, "x", {}) == 2
+
+
+@pytest.mark.hard_proof
+@pytest.mark.no_blockbuster
+async def test_write_and_read_signals(hard_proof_job_service) -> None:
+    service = hard_proof_job_service
+    job_id = uuid4()
+    await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
+
+    assert await service.unconsumed_signals(job_id) == []
+    signal = await service.write_signal(job_id, SignalType.STOP)
+    assert signal.signal_type == SignalType.STOP
+    assert signal.consumed_at is None
+
+    pending = await service.unconsumed_signals(job_id)
+    assert len(pending) == 1
+    assert pending[0].signal_type == SignalType.STOP
+
+
+@pytest.mark.hard_proof
+@pytest.mark.no_blockbuster
+async def test_sweep_orphans_reconciles_in_progress(hard_proof_job_service) -> None:
+    service = hard_proof_job_service
+    orphan = uuid4()
+    queued = uuid4()
+    await service.create_job(job_id=orphan, flow_id=uuid4(), user_id=uuid4())
+    await service.update_job_status(orphan, JobStatus.IN_PROGRESS)
+    await service.create_job(job_id=queued, flow_id=uuid4(), user_id=uuid4())
+
+    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
+
+    events = await service.read_events(orphan)
+    assert [e.event_type for e in events] == ["run_failed"]
+    assert events[0].payload == {"type": "worker_lost"}
+
+    queued_job = await service.get_job_by_job_id(queued)
+    assert queued_job.status == JobStatus.QUEUED
+
+
+@pytest.mark.hard_proof
+@pytest.mark.no_blockbuster
+async def test_get_jobs_by_flow_id_orders_by_created_timestamp(hard_proof_job_service) -> None:
+    """Regression on real engines: ordering by created_timestamp must not raise."""
+    service = hard_proof_job_service
+    flow_id = uuid4()
+    user_id = uuid4()
+    first = uuid4()
+    second = uuid4()
+    await service.create_job(job_id=first, flow_id=flow_id, user_id=user_id)
+    await service.create_job(job_id=second, flow_id=flow_id, user_id=user_id)
+
+    jobs = await service.get_jobs_by_flow_id(flow_id, user_id)
+    returned = {job.job_id for job in jobs}
+    assert {first, second} <= returned