test(background-execution): rename hard_proof marker to real_services

The hard_proof marker name was a vibe word that said nothing about what the
tests need. Rename it to real_services everywhere: the pytest marker
registration, the *_hard_proof.py test files, the Makefile target
(real_services_tests), the -m selector in migration-validation.yml, and the
CI job. real_services says what these tests require: real Postgres + Redis +
worker subprocesses. (integration was already taken for the external-API
suite under tests/integration.)

.github/workflows/migration-validation.yml

@@ -58,8 +58,8 @@ jobs:
         run: |
           uv run pytest src/backend/tests/unit/alembic/test_migration_execution.py -x -v
 
-  background-hard-proof:
-    name: Background Execution Hard-Proof (real Postgres + Redis)
+  background-real-service:
+    name: Background Execution Real-Service Tests (real Postgres + Redis)
     runs-on: ubuntu-latest
 
     services:
@@ -104,12 +104,12 @@ jobs:
         run: |
           uv sync --extra postgresql
 
-      - name: Run hard-proof tests
+      - name: Run real-service tests
         env:
           LANGFLOW_TEST_DATABASE_URI: "postgresql://langflow:langflow@localhost:5432/langflow"  # pragma: allowlist secret
           LANGFLOW_TEST_REDIS_URL: "redis://localhost:6379/0"
         run: |
-          uv run pytest src/backend/tests/unit/background_execution -m hard_proof -x -v
+          uv run pytest src/backend/tests/unit/background_execution -m real_services -x -v
 
   validate-migration:
     name: Migration Pattern Validation

Makefile

@@ -168,12 +168,12 @@ unit_tests: ## run unit tests
 unit_tests_looponfail:
 	@make unit_tests args="-f"
 
-hard_proof_tests: ## run real-instance hard-proof tests (needs LANGFLOW_TEST_DATABASE_URI + LANGFLOW_TEST_REDIS_URL)
+real_services_tests: ## run tests that need real service instances (needs LANGFLOW_TEST_DATABASE_URI + LANGFLOW_TEST_REDIS_URL)
 	@uv sync --frozen
 	uv run pytest src/backend/tests/unit \
 	--ignore=src/backend/tests/integration \
 	--ignore=src/backend/tests/unit/template \
-	-m hard_proof -ra $(args)
+	-m real_services -ra $(args)
 
 lfx_tests: ## run lfx package unit tests
 	@echo 'Running LFX Package Tests...'

pyproject.toml

@@ -205,7 +205,7 @@ markers = [
     "integration: Integration tests",
     "slow: Slow-running tests",
     "security: Security regression tests (IDOR, auth, access control)",
-    "hard_proof: Real-instance proof tests (real SQLite + real Postgres + real Redis)"
+    "real_services: Tests that need real service instances (real SQLite + real Postgres + real Redis)"
 ]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"

src/backend/tests/conftest.py

@@ -111,7 +111,7 @@ def pytest_configure(config):
     config.addinivalue_line("markers", "api_key_required: run only if the api key is set in the environment variables")
     config.addinivalue_line(
         "markers",
-        "hard_proof: Real-instance proof tests (real SQLite + real Postgres + real Redis)",
+        "real_services: Tests that need real service instances (real SQLite + real Postgres + real Redis)",
     )
     data_path = Path(__file__).parent.absolute() / "data"
 

src/backend/tests/unit/api/v2/test_workflow_agui.py

@@ -1235,7 +1235,7 @@ class TestStopWorkflowEndToEnd:
     The deterministic CANCELLED outcome of a stop is proven without an
     HTTP-level race in
     ``services/background_execution/test_service.py::test_stop_cancels_job`` and
-    ``background_execution/test_facade_hard_proof.py::test_stop_signal_cancels_run``
+    ``background_execution/test_facade_real_services.py::test_stop_signal_cancels_run``
     (sqlite + postgres), so this test deliberately does not assert the exact
     terminal label under the unsynchronized fast-flow vs stop race.
     """

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

@@ -1,14 +1,14 @@
-"""Real-instance fixtures for the background-execution hard-proof tier.
+"""Real-instance fixtures for the background-execution real-service tier.
 
-These fixtures back ``@pytest.mark.hard_proof`` tests with REAL engines, never
+These fixtures back ``@pytest.mark.real_services`` tests with REAL engines, never
 fakes:
 
-- ``hard_proof_db_url`` parametrizes over real SQLite (always) and real Postgres
+- ``real_services_db_url`` parametrizes over real SQLite (always) and real Postgres
   (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``
+- ``real_services_redis_url`` yields a real Redis URL from ``LANGFLOW_TEST_REDIS_URL``
   (skips when unset), flushed clean before and after each test.
-- ``hard_proof_job_service`` runs the real Alembic migrations against
-  ``hard_proof_db_url`` and binds ``session_scope()`` to it, so store methods are
+- ``real_services_job_service`` runs the real Alembic migrations against
+  ``real_services_db_url`` and binds ``session_scope()`` to it, so store methods are
   exercised on both real SQLite and real Postgres.
 """
 
@@ -40,7 +40,7 @@ def _async_pg_url(raw: str) -> str:
 
 
 @pytest.fixture(params=["sqlite", "postgres"])
-async def hard_proof_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[str, None]:
+async def real_services_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[str, None]:
     """Yield a real, connectable async DB URL.
 
     - sqlite: a real temp-file SQLite database (always runs).
@@ -63,10 +63,10 @@ async def hard_proof_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[st
 
 
 @pytest.fixture
-async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobService, None]:
+async def real_services_job_service(real_services_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)
+    Runs the real Alembic migrations against ``real_services_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
@@ -83,7 +83,7 @@ async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobSe
     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
+    settings_service.settings.database_url = real_services_db_url
     db_service = DatabaseServiceFactory().create(settings_service)
     manager.services[ServiceType.DATABASE_SERVICE] = db_service
 
@@ -100,7 +100,7 @@ async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobSe
 
 
 @pytest.fixture
-async def hard_proof_redis_url() -> AsyncGenerator[str, None]:
+async def real_services_redis_url() -> AsyncGenerator[str, None]:
     """Yield a real Redis URL from ``LANGFLOW_TEST_REDIS_URL``, flushed clean.
 
     Skips when ``LANGFLOW_TEST_REDIS_URL`` is unset (CI sets it via the redis:7

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

@@ -19,7 +19,7 @@ from fastapi.sse import format_sse_event
 from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
 from langflow.services.background_execution.service import BackgroundExecutionService
 
-pytestmark = pytest.mark.hard_proof
+pytestmark = pytest.mark.real_services
 
 
 def _live_agui_run_started_bytes(seq: int) -> bytes:

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

@@ -1,4 +1,4 @@
-"""BOUNDED-CONCURRENCY hard proof.
+"""BOUNDED-CONCURRENCY real service.
 
 Submit N+k jobs to the real ``InProcessExecutor`` configured for max concurrency
 N. A barrier holds every started job so the peak in-flight count is observable;
@@ -15,7 +15,7 @@ import threading
 import pytest
 from langflow.services.background_execution.executor import InProcessExecutor
 
-pytestmark = pytest.mark.hard_proof
+pytestmark = pytest.mark.real_services
 
 
 class _Probe:

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

@@ -1,4 +1,4 @@
-"""Hard-proof the in-process executor's worker resilience.
+"""Verify the in-process executor's worker resilience.
 
 These tests drive the REAL ``InProcessExecutor`` (no mocks of our logic) and
 assert the worker pool survives the two ways a job can end abnormally:
@@ -21,7 +21,7 @@ import asyncio
 import pytest
 from langflow.services.background_execution.executor import InProcessExecutor
 
-pytestmark = pytest.mark.hard_proof
+pytestmark = pytest.mark.real_services
 
 
 async def test_pool_survives_repeated_job_cancellation(monkeypatch):

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

@@ -1,7 +1,7 @@
-"""Hard-proof the durable background path on real SQLite AND real Postgres.
+"""Verify the durable background path on real SQLite AND real Postgres.
 
 The Runner + InMemoryLiveBus + facade durable replay all sit on ``JobService``,
-whose ``session_scope()`` the ``hard_proof_job_service`` fixture binds to a real,
+whose ``session_scope()`` the ``real_services_job_service`` fixture binds to a real,
 migrated DB parametrized over sqlite and postgres. These tests drive the real
 Runner with a scripted frame source (legitimate test input, not a mock of our
 logic) and assert durable persistence, ordering, terminal state, reattach replay,
@@ -23,7 +23,7 @@ from langflow.services.database.models.jobs.model import JobStatus, SignalType
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
 
-pytestmark = pytest.mark.hard_proof
+pytestmark = pytest.mark.real_services
 
 
 def _frame(event_type: str, data: dict) -> tuple[bytes, str]:
@@ -35,8 +35,8 @@ def _runner(job_service, bus, job_id, source) -> JobRunner:
     return JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source)
 
 
-async def test_durable_persistence_and_terminal_state(hard_proof_job_service):
-    job_service = hard_proof_job_service
+async def test_durable_persistence_and_terminal_state(real_services_job_service):
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -60,8 +60,8 @@ async def test_durable_persistence_and_terminal_state(hard_proof_job_service):
     assert [e.seq for e in events] == [1, 2, 3]  # contiguous, monotonic
 
 
-async def test_reattach_replays_durable_after_completion(hard_proof_job_service):
-    job_service = hard_proof_job_service
+async def test_reattach_replays_durable_after_completion(real_services_job_service):
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -88,8 +88,8 @@ async def test_reattach_replays_durable_after_completion(hard_proof_job_service)
     assert b"end_vertex" in bodies
 
 
-async def test_reattach_from_midpoint_has_no_gap(hard_proof_job_service):
-    job_service = hard_proof_job_service
+async def test_reattach_from_midpoint_has_no_gap(real_services_job_service):
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -110,8 +110,8 @@ async def test_reattach_from_midpoint_has_no_gap(hard_proof_job_service):
     assert seqs == [2, 3]
 
 
-async def test_stop_signal_cancels_run(hard_proof_job_service):
-    job_service = hard_proof_job_service
+async def test_stop_signal_cancels_run(real_services_job_service):
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
     # A STOP written before the run makes the first boundary poll cancel it.
@@ -128,8 +128,8 @@ async def test_stop_signal_cancels_run(hard_proof_job_service):
     assert job.status == JobStatus.CANCELLED
 
 
-async def test_sweep_orphans_fails_in_progress(hard_proof_job_service):
-    job_service = hard_proof_job_service
+async def test_sweep_orphans_fails_in_progress(real_services_job_service):
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
     await job_service.update_job_status(job_id, JobStatus.IN_PROGRESS)
@@ -161,7 +161,7 @@ def _echo_input_factory(*, request, **_kwargs):
     return _source
 
 
-async def test_submit_persists_request_for_faithful_requeue(hard_proof_job_service):
+async def test_submit_persists_request_for_faithful_requeue(real_services_job_service):
     """``submit`` persists the request body on the job row (job_metadata.request).
 
     This is the durable record the startup sweep reads to replay original inputs.
@@ -171,7 +171,7 @@ async def test_submit_persists_request_for_faithful_requeue(hard_proof_job_servi
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     original_input = f"original-input-{uuid4()}"
     flow_id = uuid4()
 
@@ -197,7 +197,7 @@ async def test_submit_persists_request_for_faithful_requeue(hard_proof_job_servi
     assert persisted == request
 
 
-async def test_submit_redacts_inline_globals_from_persisted_request(hard_proof_job_service):
+async def test_submit_redacts_inline_globals_from_persisted_request(real_services_job_service):
     """Inline ``globals`` (request-level secrets) must NOT land plaintext on the row.
 
     ``submit`` persists the request body for faithful replay, but request-level
@@ -212,7 +212,7 @@ async def test_submit_redacts_inline_globals_from_persisted_request(hard_proof_j
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     flow_id = uuid4()
     secret = f"sk-secret-{uuid4()}"
 
@@ -239,7 +239,7 @@ async def test_submit_redacts_inline_globals_from_persisted_request(hard_proof_j
     assert request["globals"] == {"OPENAI_API_KEY": secret}
 
 
-async def test_requeued_queued_job_replays_original_input(hard_proof_job_service):
+async def test_requeued_queued_job_replays_original_input(real_services_job_service):
     """A QUEUED job that survives a restart re-runs with its ORIGINAL input.
 
     Models "queued, never started, then the process died": the durable row is
@@ -254,7 +254,7 @@ async def test_requeued_queued_job_replays_original_input(hard_proof_job_service
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     original_input = f"original-input-{uuid4()}"
     user_id = uuid4()
     flow_id = uuid4()
@@ -295,7 +295,7 @@ async def test_requeued_queued_job_replays_original_input(hard_proof_job_service
     assert echoed == [original_input]
 
 
-async def test_job_timeout_marks_timed_out(hard_proof_job_service):
+async def test_job_timeout_marks_timed_out(real_services_job_service):
     """A run that overruns ``background_job_timeout`` ends TIMED_OUT.
 
     The runner wraps the drive in ``asyncio.wait_for(timeout=...)``;
@@ -306,7 +306,7 @@ async def test_job_timeout_marks_timed_out(hard_proof_job_service):
     """
     import asyncio
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -330,7 +330,7 @@ async def test_job_timeout_marks_timed_out(hard_proof_job_service):
     assert job.status == JobStatus.TIMED_OUT
 
 
-async def test_events_reattach_after_restart_returns_on_terminal_job(hard_proof_job_service):
+async def test_events_reattach_after_restart_returns_on_terminal_job(real_services_job_service):
     """Reattaching to an already-terminal job after a restart MUST NOT hang.
 
     Models a process restart: a job runs to COMPLETED, persisting durable
@@ -348,7 +348,7 @@ async def test_events_reattach_after_restart_returns_on_terminal_job(hard_proof_
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     user_id = uuid4()
     flow_id = uuid4()
     job_id = uuid4()
@@ -381,7 +381,7 @@ async def test_events_reattach_after_restart_returns_on_terminal_job(hard_proof_
     assert b"end" in body
 
 
-async def test_events_replay_frames_are_sse_framed(hard_proof_job_service):
+async def test_events_replay_frames_are_sse_framed(real_services_job_service):
     """Durable replay must emit SSE-framed bytes byte-compatible with live frames.
 
     Live frames are pre-SSE-framed with a ``data:`` line followed by an ``id:``
@@ -396,7 +396,7 @@ async def test_events_replay_frames_are_sse_framed(hard_proof_job_service):
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     user_id = uuid4()
     flow_id = uuid4()
     job_id = uuid4()
@@ -423,7 +423,7 @@ async def test_events_replay_frames_are_sse_framed(hard_proof_job_service):
         assert f"id: {seq}".encode() in frame, f"replayed frame missing id: {frame!r}"
 
 
-async def test_executor_stop_applies_terminal_reconcile(hard_proof_job_service):
+async def test_executor_stop_applies_terminal_reconcile(real_services_job_service):
     """``executor.stop()`` lets an in-flight stopped job's reconcile land.
 
     Drives the REAL runner on the executor. The job blocks mid-run; a STOP signal
@@ -440,7 +440,7 @@ async def test_executor_stop_applies_terminal_reconcile(hard_proof_job_service):
 
     from langflow.services.background_execution.executor import InProcessExecutor
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -476,7 +476,7 @@ async def test_executor_stop_applies_terminal_reconcile(hard_proof_job_service):
     assert job.status == JobStatus.CANCELLED
 
 
-async def test_stop_poll_only_on_durable_frames(hard_proof_job_service):
+async def test_stop_poll_only_on_durable_frames(real_services_job_service):
     """The runner polls the STOP signal only on DURABLE frames, not every token.
 
     Polling ``unconsumed_signals`` (a DB read) on every ephemeral token is wasted
@@ -486,7 +486,7 @@ async def test_stop_poll_only_on_durable_frames(hard_proof_job_service):
     durable frames, and assert the poll count tracks the durable frames, not the
     token flood. Real SQLite and Postgres.
     """
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -521,7 +521,7 @@ async def test_stop_poll_only_on_durable_frames(hard_proof_job_service):
     )
 
 
-async def test_stop_signal_is_marked_consumed(hard_proof_job_service):
+async def test_stop_signal_is_marked_consumed(real_services_job_service):
     """When the runner acts on a STOP, the signal row is stamped ``consumed_at``.
 
     Otherwise the execution_signals table grows unbounded and, worse, a
@@ -530,7 +530,7 @@ async def test_stop_signal_is_marked_consumed(hard_proof_job_service):
     (2) a fresh run of the SAME job_id afterwards does NOT instantly cancel —
     it completes, because the stale STOP was consumed. Real SQLite and Postgres.
     """
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
     await job_service.write_signal(job_id, SignalType.STOP)
@@ -579,7 +579,7 @@ def _side_effect_factory(*, request, **_kwargs):  # noqa: ARG001
     return _source
 
 
-async def test_concurrent_sweep_runs_queued_job_exactly_once(hard_proof_job_service):
+async def test_concurrent_sweep_runs_queued_job_exactly_once(real_services_job_service):
     """Two startup sweepers sharing one DB must run a QUEUED job EXACTLY once.
 
     Models two uvicorn workers booting against the same database, each calling
@@ -594,7 +594,7 @@ async def test_concurrent_sweep_runs_queued_job_exactly_once(hard_proof_job_serv
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     user_id = uuid4()
     flow_id = uuid4()
     job_id = uuid4()

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

@@ -1,43 +0,0 @@
-"""Verify CI wires a real redis:7 service and runs the hard-proof tier."""
-
-from __future__ import annotations
-
-from pathlib import Path
-
-import pytest
-import yaml
-
-_REPO_ROOT = Path(__file__).resolve().parents[5]
-_WORKFLOW = _REPO_ROOT / ".github" / "workflows" / "migration-validation.yml"
-
-
-@pytest.fixture
-def workflow() -> dict:
-    assert _WORKFLOW.exists(), f"workflow not found at {_WORKFLOW}"
-    return yaml.safe_load(_WORKFLOW.read_text(encoding="utf-8"))
-
-
-def _hard_proof_job(workflow: dict) -> dict | None:
-    return next((j for name, j in workflow["jobs"].items() if "hard" in name and "proof" in name), None)
-
-
-@pytest.mark.no_blockbuster
-def test_hard_proof_job_exists(workflow: dict) -> None:
-    assert _hard_proof_job(workflow) is not None, f"no hard-proof job; jobs: {list(workflow['jobs'])}"
-
-
-@pytest.mark.no_blockbuster
-def test_hard_proof_job_has_postgres_and_redis_services(workflow: dict) -> None:
-    hard_proof = _hard_proof_job(workflow)
-    assert hard_proof is not None
-    images = {svc["image"] for svc in hard_proof["services"].values()}
-    assert any(img.startswith("postgres:") for img in images), images
-    assert any(img.startswith("redis:") for img in images), images
-
-
-@pytest.mark.no_blockbuster
-def test_hard_proof_job_sets_both_test_uris() -> None:
-    text = _WORKFLOW.read_text(encoding="utf-8")
-    assert "LANGFLOW_TEST_DATABASE_URI" in text
-    assert "LANGFLOW_TEST_REDIS_URL" in text
-    assert "-m hard_proof" in text or "hard_proof_tests" in text

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

@@ -1,20 +0,0 @@
-"""Verify the Makefile exposes a hard_proof_tests target wired to -m hard_proof."""
-
-from __future__ import annotations
-
-from pathlib import Path
-
-import pytest
-
-_REPO_ROOT = Path(__file__).resolve().parents[5]
-_MAKEFILE = _REPO_ROOT / "Makefile"
-
-
-@pytest.mark.no_blockbuster
-def test_makefile_has_hard_proof_target() -> None:
-    assert _MAKEFILE.exists(), f"Makefile not found at {_MAKEFILE}"
-    text = _MAKEFILE.read_text(encoding="utf-8")
-    assert "hard_proof_tests:" in text, "missing hard_proof_tests make target"
-    target_block = text.split("hard_proof_tests:", 1)[1].split("\n\n", 1)[0]
-    assert "-m hard_proof" in target_block, "hard_proof_tests target must select the hard_proof marker"
-    assert "src/backend/tests/unit" in target_block, "target must run the backend unit tree"

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

@@ -33,7 +33,7 @@ from langflow.services.background_execution.live_bus import InMemoryLiveBus
 from langflow.services.background_execution.runner import JobRunner
 from langflow.services.database.models.jobs.model import JobStatus
 
-pytestmark = pytest.mark.hard_proof
+pytestmark = pytest.mark.real_services
 
 
 # --------------------------------------------------------------------------- #
@@ -146,12 +146,12 @@ def _echo_factory(*, request, **_kwargs):
     return _source
 
 
-async def test_durability_delta_off_loses_job_on_restart_on_recovers(hard_proof_job_service):
+async def test_durability_delta_off_loses_job_on_restart_on_recovers(real_services_job_service):
     """Durability OFF: a restart finds NO recoverable row -> job lost. ON: recovered + re-run."""
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
 
     # --- OFF: the in-memory-only path (pre-fix). The submit never wrote a durable
     #     QUEUED row — the job lived only in the dead process's executor queue. A
@@ -229,9 +229,9 @@ def _runner(job_service, job_id, source):
     return JobRunner(job_service=job_service, live_bus=InMemoryLiveBus(), adapter=adapter, frame_source=source)
 
 
-async def test_sweep_claim_delta_off_double_runs_on_exactly_once(hard_proof_job_service):
+async def test_sweep_claim_delta_off_double_runs_on_exactly_once(real_services_job_service):
     """Sweep claim OFF: two sweepers both run a QUEUED job (double). ON: exactly once."""
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
 
     # --- OFF: the pre-fix unconditional re-enqueue. Two sweepers each run the job
     #     because neither claims it first. We reproduce by running the runner twice

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

@@ -30,7 +30,7 @@ from langflow.services.background_execution.runner import JobRunner
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
 
-pytestmark = pytest.mark.hard_proof
+pytestmark = pytest.mark.real_services
 
 
 class _StubUser:
@@ -92,7 +92,7 @@ async def _run_token_flow(job_service, job_id, bus) -> None:
     await runner.run(job_id=job_id, source_kwargs={})
 
 
-async def test_live_id_matches_durable_seq_default(hard_proof_job_service):
+async def test_live_id_matches_durable_seq_default(real_services_job_service):
     """The live id: a client sees == the durable seq the replay resumes after.
 
     Drives the real runner (default in-memory bus), captures the live frames, then
@@ -100,7 +100,7 @@ async def test_live_id_matches_durable_seq_default(hard_proof_job_service):
     Pre-fix the live id is the stream-seq (0,1,4,5) while the durable seq is
     (1,2,3,4), so they diverge and a Last-Event-ID resume is wrong.
     """
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     user_id = uuid4()
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=user_id)
@@ -134,7 +134,7 @@ async def test_live_id_matches_durable_seq_default(hard_proof_job_service):
         )
 
 
-async def test_reattach_no_gap_no_dup_default(hard_proof_job_service):
+async def test_reattach_no_gap_no_dup_default(real_services_job_service):
     """Mid-run Last-Event-ID resume starts exactly at the next milestone.
 
     Captures the live id of the ``vertices_sorted`` milestone, reconnects with it
@@ -148,7 +148,7 @@ async def test_reattach_no_gap_no_dup_default(hard_proof_job_service):
     from langflow.services.background_execution.service import BackgroundExecutionService
     from langflow.services.deps import get_settings_service
 
-    job_service = hard_proof_job_service
+    job_service = real_services_job_service
     user_id = uuid4()
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=user_id)

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

@@ -7,7 +7,7 @@ inject a phantom terminal event into A's live durable log, and race A's real
 terminal write. With lease+heartbeat liveness the sweep only reconciles
 GENUINELY orphaned rows (stale/absent heartbeat), so B leaves A's live job alone.
 
-Real SQLite AND real Postgres via the hard-proof fixture.
+Real SQLite AND real Postgres via the real-service fixture.
 """
 
 from __future__ import annotations
@@ -26,15 +26,15 @@ from langflow.services.database.models.jobs.model import JobStatus
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
 
-pytestmark = pytest.mark.hard_proof
+pytestmark = pytest.mark.real_services
 
 
 def _frame(event_type: str, data: dict) -> tuple[bytes, str]:
     return (json.dumps({"event": event_type, "data": data}).encode("utf-8"), event_type)
 
 
-async def test_booting_worker_sweep_spares_live_heartbeated_job(hard_proof_job_service):
-    job_service = hard_proof_job_service
+async def test_booting_worker_sweep_spares_live_heartbeated_job(real_services_job_service):
+    job_service = real_services_job_service
     job_id = uuid4()
     await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 

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

@@ -0,0 +1,43 @@
+"""Verify CI wires a real redis:7 service and runs the real-service tier."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+import yaml
+
+_REPO_ROOT = Path(__file__).resolve().parents[5]
+_WORKFLOW = _REPO_ROOT / ".github" / "workflows" / "migration-validation.yml"
+
+
+@pytest.fixture
+def workflow() -> dict:
+    assert _WORKFLOW.exists(), f"workflow not found at {_WORKFLOW}"
+    return yaml.safe_load(_WORKFLOW.read_text(encoding="utf-8"))
+
+
+def _real_services_job(workflow: dict) -> dict | None:
+    return next((j for name, j in workflow["jobs"].items() if "real" in name and "service" in name), None)
+
+
+@pytest.mark.no_blockbuster
+def test_real_services_job_exists(workflow: dict) -> None:
+    assert _real_services_job(workflow) is not None, f"no real-service job; jobs: {list(workflow['jobs'])}"
+
+
+@pytest.mark.no_blockbuster
+def test_real_services_job_has_postgres_and_redis_services(workflow: dict) -> None:
+    real_services = _real_services_job(workflow)
+    assert real_services is not None
+    images = {svc["image"] for svc in real_services["services"].values()}
+    assert any(img.startswith("postgres:") for img in images), images
+    assert any(img.startswith("redis:") for img in images), images
+
+
+@pytest.mark.no_blockbuster
+def test_real_services_job_sets_both_test_uris() -> None:
+    text = _WORKFLOW.read_text(encoding="utf-8")
+    assert "LANGFLOW_TEST_DATABASE_URI" in text
+    assert "LANGFLOW_TEST_REDIS_URL" in text
+    assert "-m real_services" in text or "real_services_tests" in text

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

@@ -1,4 +1,4 @@
-"""Phase 0 harness tests: the hard_proof marker + real-instance fixtures.
+"""Phase 0 harness tests: the real_services marker + real-instance fixtures.
 
 These prove the harness itself works before any background-execution code exists:
 the marker is registered, the DB fixture backs a real engine on sqlite and
@@ -15,20 +15,20 @@ from sqlalchemy.ext.asyncio import create_async_engine
 
 
 @pytest.mark.no_blockbuster
-def test_hard_proof_marker_is_registered(request: pytest.FixtureRequest) -> None:
-    """The hard_proof marker must be registered so --strict-markers accepts it."""
+def test_real_services_marker_is_registered(request: pytest.FixtureRequest) -> None:
+    """The real_services marker must be registered so --strict-markers accepts it."""
     ini_markers = request.config.getini("markers")
     names = {line.split(":", 1)[0].strip() for line in ini_markers}
-    assert "hard_proof" in names, f"hard_proof not registered; have: {sorted(names)}"
+    assert "real_services" in names, f"real_services not registered; have: {sorted(names)}"
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_hard_proof_db_url_yields_real_engine(hard_proof_db_url: str) -> None:
-    """hard_proof_db_url must produce a URL backing a real, queryable engine."""
-    if "postgresql" in hard_proof_db_url:
+async def test_real_services_db_url_yields_real_engine(real_services_db_url: str) -> None:
+    """real_services_db_url must produce a URL backing a real, queryable engine."""
+    if "postgresql" in real_services_db_url:
         assert os.environ.get("LANGFLOW_TEST_DATABASE_URI"), "postgres param ran without URI set"
-    engine = create_async_engine(hard_proof_db_url)
+    engine = create_async_engine(real_services_db_url)
     try:
         async with engine.connect() as conn:
             result = await conn.execute(text("SELECT 1"))
@@ -37,17 +37,17 @@ async def test_hard_proof_db_url_yields_real_engine(hard_proof_db_url: str) -> N
         await engine.dispose()
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_hard_proof_redis_url_pings_real_server(hard_proof_redis_url: str) -> None:
-    """hard_proof_redis_url must back a reachable, real Redis server."""
+async def test_real_services_redis_url_pings_real_server(real_services_redis_url: str) -> None:
+    """real_services_redis_url must back a reachable, real Redis server."""
     from redis.asyncio import StrictRedis
 
     assert os.environ.get("LANGFLOW_TEST_REDIS_URL"), "redis fixture ran without URL set"
-    client = StrictRedis.from_url(hard_proof_redis_url)
+    client = StrictRedis.from_url(real_services_redis_url)
     try:
         assert await client.ping() is True
-        await client.set("hard_proof:probe", "1")
-        assert await client.get("hard_proof:probe") == b"1"
+        await client.set("real_services:probe", "1")
+        assert await client.get("real_services:probe") == b"1"
     finally:
         await client.aclose()

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

@@ -0,0 +1,20 @@
+"""Verify the Makefile exposes a real_services_tests target wired to -m real_services."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+_REPO_ROOT = Path(__file__).resolve().parents[5]
+_MAKEFILE = _REPO_ROOT / "Makefile"
+
+
+@pytest.mark.no_blockbuster
+def test_makefile_has_real_services_target() -> None:
+    assert _MAKEFILE.exists(), f"Makefile not found at {_MAKEFILE}"
+    text = _MAKEFILE.read_text(encoding="utf-8")
+    assert "real_services_tests:" in text, "missing real_services_tests make target"
+    target_block = text.split("real_services_tests:", 1)[1].split("\n\n", 1)[0]
+    assert "-m real_services" in target_block, "real_services_tests target must select the real_services marker"
+    assert "src/backend/tests/unit" in target_block, "target must run the backend unit tree"

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

@@ -1,7 +1,7 @@
-"""Hard-proof store tests: JobService methods against REAL SQLite and REAL Postgres.
+"""Real-service 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
+``real_services_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
@@ -17,10 +17,10 @@ import pytest
 from langflow.services.database.models.jobs.model import JobStatus, SignalType
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_set_result_and_error_persist(hard_proof_job_service) -> None:
-    service = hard_proof_job_service
+async def test_set_result_and_error_persist(real_services_job_service) -> None:
+    service = real_services_job_service
     job_id = uuid4()
     await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -32,10 +32,10 @@ async def test_set_result_and_error_persist(hard_proof_job_service) -> None:
     assert fetched.error == {"type": "boom", "message": "x"}
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_append_and_read_events_ordered(hard_proof_job_service) -> None:
-    service = hard_proof_job_service
+async def test_append_and_read_events_ordered(real_services_job_service) -> None:
+    service = real_services_job_service
     job_id = uuid4()
     await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -48,10 +48,10 @@ async def test_append_and_read_events_ordered(hard_proof_job_service) -> None:
     assert tail[0].payload == {"i": 2}
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_append_event_seq_is_per_job(hard_proof_job_service) -> None:
-    service = hard_proof_job_service
+async def test_append_event_seq_is_per_job(real_services_job_service) -> None:
+    service = real_services_job_service
     job_a = uuid4()
     job_b = uuid4()
     await service.create_job(job_id=job_a, flow_id=uuid4(), user_id=uuid4())
@@ -62,15 +62,15 @@ async def test_append_event_seq_is_per_job(hard_proof_job_service) -> None:
     assert await service.append_event(job_a, "x", {}) == 2
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_append_event_concurrent_appends_all_land_gap_free(hard_proof_job_service) -> None:
+async def test_append_event_concurrent_appends_all_land_gap_free(real_services_job_service) -> None:
     """Concurrent appends to one job must ALL land, gap-free (retry on UNIQUE collision).
 
     Regression for the seq race: SELECT max(seq)+1 then INSERT collides under
     contention; without retry the losers' events were silently dropped.
     """
-    service = hard_proof_job_service
+    service = real_services_job_service
     job_id = uuid4()
     await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -84,10 +84,10 @@ async def test_append_event_concurrent_appends_all_land_gap_free(hard_proof_job_
     assert len(events) == n
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_write_and_read_signals(hard_proof_job_service) -> None:
-    service = hard_proof_job_service
+async def test_write_and_read_signals(real_services_job_service) -> None:
+    service = real_services_job_service
     job_id = uuid4()
     await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
 
@@ -101,10 +101,10 @@ async def test_write_and_read_signals(hard_proof_job_service) -> None:
     assert pending[0].signal_type == SignalType.STOP
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_sweep_orphans_reconciles_in_progress(hard_proof_job_service) -> None:
-    service = hard_proof_job_service
+async def test_sweep_orphans_reconciles_in_progress(real_services_job_service) -> None:
+    service = real_services_job_service
     orphan = uuid4()
     queued = uuid4()
     await service.create_job(job_id=orphan, flow_id=uuid4(), user_id=uuid4())
@@ -128,11 +128,11 @@ async def test_sweep_orphans_reconciles_in_progress(hard_proof_job_service) -> N
     assert queued_job.status == JobStatus.QUEUED
 
 
-@pytest.mark.hard_proof
+@pytest.mark.real_services
 @pytest.mark.no_blockbuster
-async def test_get_jobs_by_flow_id_orders_by_created_timestamp(hard_proof_job_service) -> None:
+async def test_get_jobs_by_flow_id_orders_by_created_timestamp(real_services_job_service) -> None:
     """Regression on real engines: ordering by created_timestamp must not raise."""
-    service = hard_proof_job_service
+    service = real_services_job_service
     flow_id = uuid4()
     user_id = uuid4()
     first = uuid4()