fix(benchmarks): use TCP probe for langflow_run_http_ready readiness

.github/workflows/cold-start-benchmark.yml

@@ -7,10 +7,9 @@ name: Cold Start Benchmark
 #   3. A commit is pushed to a branch matching `cold-start/**`.
 #   4. A maintainer triggers the workflow manually via workflow_dispatch.
 #
-# Workflow structure: scenarios run in parallel matrix jobs so a single slow or
-# broken scenario (e.g. langflow_run_http_ready's structlog marker issue) cannot
-# exhaust the 30-minute job budget for the whole pipeline. Each matrix job owns
-# ONE scenario; a final aggregation job assembles results.
+# Workflow structure: scenarios run in parallel matrix jobs so a single slow
+# scenario cannot exhaust the 30-minute job budget for the whole pipeline.
+# Each matrix job owns ONE scenario; a final aggregation job assembles results.
 #
 # Regression policy: any matrix job with a regression fails the workflow
 # AND posts a bot comment on the PR with the numbers diff sourced from the
@@ -173,31 +172,28 @@ jobs:
           - lfx_bare
           - lfx_with_flow
           - lfx_with_flow_prebaked
-          # langflow_run_http_ready and langflow_run_no_change_restart are known-
-          # flaky scenarios on CI until the structlog marker fix lands (Phase 4/5).
-          # The matrix expressions below give them a tight 5-min timeout and mark
-          # them non-blocking so their cancellation does not drag the workflow into
-          # `cancelled`. Phase 4 plan 04-05 adds langflow_run_no_change_restart,
-          # which inherits the flaky-scenario treatment until Boot-2's marker is
-          # verified to reach the supervisor reliably.
+          # langflow_run_no_change_restart stays non-blocking until its
+          # self_measuring dispatch shape is stable under CI load; its
+          # thresholds.json entry is carried over from the 2026-04-20 snapshot
+          # because recent runs produce ~1ms garbage (harness bug, not a real
+          # measurement).
           - langflow_run_http_ready
           - langflow_run_no_change_restart
           - lfx_reference_image
     runs-on: ubuntu-latest
     # Budgets:
-    #   - langflow_run_http_ready: 5 min (known-flaky sentinel scenario until the
-    #     structlog marker fix lands; kept tight so it fails fast).
+    #   - langflow_run_http_ready: 10 min. Single cold boot to TCP-ready
+    #     (~30-40s on Linux CI) plus hyperfine's 5 runs.
     #   - langflow_run_no_change_restart: 15 min. Its supervisor runs TWO full
     #     langflow boots per hyperfine iteration (pre-warm + measured), and
     #     hyperfine defaults to runs=5, so ~10 boots at ~35s each on Linux CI.
     #   - all others: 20 min.
-    timeout-minutes: ${{ matrix.scenario == 'langflow_run_http_ready' && 5 || (matrix.scenario == 'langflow_run_no_change_restart' && 15 || 20) }}
-    # Matrix-entry-level continue-on-error: langflow_run_http_ready and
-    # langflow_run_no_change_restart are allowed to fail (or be cancelled by
-    # timeout) without marking the workflow red. no_change_restart keeps the
-    # non-blocking treatment until Boot-2's marker is verified to reach the
-    # supervisor reliably under CI load.
-    continue-on-error: ${{ matrix.scenario == 'langflow_run_http_ready' || matrix.scenario == 'langflow_run_no_change_restart' || matrix.scenario == 'lfx_reference_image' }}
+    timeout-minutes: ${{ matrix.scenario == 'langflow_run_http_ready' && 10 || (matrix.scenario == 'langflow_run_no_change_restart' && 15 || 20) }}
+    # Matrix-entry-level continue-on-error:
+    #   - langflow_run_no_change_restart: stays non-blocking until its
+    #     self_measuring dispatch is fixed (currently produces ~1ms garbage).
+    #   - lfx_reference_image: stays non-blocking per D-12 amendment.
+    continue-on-error: ${{ matrix.scenario == 'langflow_run_no_change_restart' || matrix.scenario == 'lfx_reference_image' }}
     steps:
       - name: Checkout
         uses: actions/checkout@v6
@@ -237,7 +233,7 @@ jobs:
 
       - name: Run scenario
         id: bench
-        continue-on-error: ${{ matrix.scenario == 'langflow_run_http_ready' || matrix.scenario == 'langflow_run_no_change_restart' || matrix.scenario == 'lfx_reference_image' }}
+        continue-on-error: ${{ matrix.scenario == 'langflow_run_no_change_restart' || matrix.scenario == 'lfx_reference_image' }}
         env:
           CONTAINER_CMD: docker
         run: |
@@ -267,7 +263,6 @@ jobs:
         if: >-
           failure() &&
           github.event.pull_request &&
-          matrix.scenario != 'langflow_run_http_ready' &&
           matrix.scenario != 'langflow_run_no_change_restart' &&
           matrix.scenario != 'lfx_reference_image'
         env:
@@ -387,8 +382,7 @@ jobs:
               "allowed_regression_pct": 15,
               "_note": (
                   "Captured on Linux CI via the cold-start-benchmark matrix workflow. "
-                  "langflow_run_http_ready may be sentinel if its supervisor marker is "
-                  "still unresolved. measurement_mode is bytecode_compile_delta ."
+                  "measurement_mode is bytecode_compile_delta."
               ),
               "scenarios": scenarios_out,
           }

src/backend/base/langflow/server.py

@@ -70,6 +70,7 @@ def __init__(self, app_factory: Callable[[], Any], options=None) -> None:
         self.options["worker_class"] = "langflow.server.LangflowUvicornWorker"
         self.options["logger_class"] = Logger
         self.options["pre_fork"] = self.pre_fork
+        self.options["post_fork"] = _langflow_post_fork
         self._app_factory = app_factory
         self.application = None
         super().__init__()
@@ -166,3 +167,26 @@ def load(self):
 
                 preload_master()
         return self.application
+
+
+def _langflow_post_fork(server, worker) -> None:  # noqa: ARG001
+    """Reset fork-unsafe resources in each worker after gunicorn forks.
+
+    Gunicorn calls this hook synchronously in the worker process immediately
+    after fork, before any request is served. No event loop exists yet here —
+    this function MUST remain fully synchronous.
+
+    Currently resets ``TelemetryService.client`` (an ``httpx.AsyncClient``
+    constructed during master preload) so ``TelemetryService.start()`` can
+    reconstruct it inside the worker's event loop. ``httpx.AsyncClient`` has
+    no synchronous ``.close()``, so replacing the reference is the correct
+    pattern.
+    """
+    try:
+        from langflow.services.deps import get_telemetry_service
+
+        get_telemetry_service().client = None
+    except Exception:  # noqa: BLE001, S110
+        # Service not yet initialized (e.g. preload_app=False path). The
+        # hook must not crash gunicorn.
+        pass

src/backend/tests/benchmarks/driver.py

@@ -878,9 +878,15 @@ def compare_against_thresholds(
     Behavior:
       - measurement_mode mismatch -> stderr WARNING only (not a failure; see D-11a note).
       - any scenario delta_pct > allowed_regression_pct/100 -> exit code EXIT_VERIFY_REGRESSION.
-      - baseline mean_ms <= 0 -> treat any finite current mean as FAIL (Path-B sentinel trip).
+      - baseline runs == 0 AND mean_ms <= 0 -> UNANCHORED (scenario exists in
+        thresholds.json as a placeholder but has never been snapshotted); the
+        row is recorded with status SKIP and does not trip the gate. A new
+        scenario can land in thresholds.json alongside its code change without
+        failing its first real measurement.
+      - baseline mean_ms <= 0 (with runs > 0) -> FAIL on any finite current
+        mean (Path-B sentinel trip: intentionally-zeroed baseline).
       - on any FAIL, write reports/regression_comment.md.
-      - on all PASS, no regression_comment.md file is written.
+      - on all PASS/SKIP, no regression_comment.md file is written.
     """
     if not thresholds_path.exists():
         sys.stderr.write(
@@ -905,9 +911,17 @@ def compare_against_thresholds(
         if not current:
             continue
         baseline_ms = float(t_entry.get("mean_ms", 0.0))
+        baseline_runs = int(t_entry.get("runs", 0) or 0)
         current_ms = float(current.get("mean_ms", 0.0))
-        if baseline_ms <= 0.0:
-            # Sentinel baseline: any finite current trips the gate.
+        if baseline_ms <= 0.0 and baseline_runs <= 0:
+            # Unanchored baseline: scenario is a placeholder that has never been
+            # snapshotted. Record the current number for visibility but do not
+            # trip the gate — the next snapshot run will anchor it.
+            status = "SKIP"
+            delta_pct = 0.0
+        elif baseline_ms <= 0.0:
+            # Sentinel baseline with prior runs: any finite current trips the
+            # gate (intentionally-zeroed baseline, Path-B sentinel trip).
             status = "FAIL"
             delta_pct = float("inf") if current_ms > 0 else 0.0
         else:

src/backend/tests/benchmarks/scenarios/_langflow_supervisor.py

@@ -1,47 +1,92 @@
-"""Supervisor: launches `langflow run`, exits when "Application startup complete." appears on stderr.
+"""Supervisor: launches `langflow run`, exits when 127.0.0.1:7860 accepts a TCP connection.
 
-Per Pitfall 7 in 01-RESEARCH.md:
-  - langflow run starts uvicorn and does not return.
-  - Terminal checkpoint = uvicorn's `Application startup complete.` log line.
-  - Supervisor reads stdout+stderr line-by-line, records time.perf_counter(), then SIGTERMs.
+hyperfine measures the wall-clock duration of this supervisor, which equals
+time-to-HTTP-ready. That's the langflow_run_http_ready scenario.
 
-hyperfine measures the wall-clock duration of THIS supervisor, which equals
-time-to-HTTP-ready. That's the MEAS-01 langflow_run scenario.
+Readiness detection: TCP connect success against the server's bind address. The
+uvicorn `Application startup complete.` stdout marker is swallowed by langflow's
+structlog processor pipeline, so scraping it races under load. TCP connect is a
+ground-truth readiness signal that does not depend on logging at all. The same
+approach is used by _langflow_no_change_restart_supervisor.py.
 
 Exit codes:
-  0 - success (ready marker observed; LANGFLOW_READY_MS printed to stdout).
-  2 - timeout (STARTUP_TIMEOUT_SEC elapsed without observing the marker).
-  3 - early exit (process exited before the marker appeared).
+  0 - success (port reachable; LANGFLOW_READY_MS printed to stdout).
+  2 - timeout (STARTUP_TIMEOUT_SEC elapsed without the port accepting connections).
+  3 - early exit (process exited before the port was reachable).
 """
 
 from __future__ import annotations
 
 import os
 import signal
+import socket
 import subprocess
 import sys
+import threading
 import time
 
-READY_MARKER = "Application startup complete."
-# 60s is sufficient on Linux CI (typical cold boot is 30-40s). macOS/podman runs with an
-# emulated VM need more headroom; `LANGFLOW_BENCH_STARTUP_TIMEOUT` overrides at invocation time.
+READY_HOST = "127.0.0.1"
+READY_PORT = 7860
+# Short enough to keep measurement noise bounded; long enough to avoid
+# trivial CPU burn during the ~30-40s cold boot on Linux CI.
+READY_POLL_INTERVAL_SEC = 0.05
+# 60s is sufficient on Linux CI. macOS/podman runs with an emulated VM need more
+# headroom; `LANGFLOW_BENCH_STARTUP_TIMEOUT` overrides at invocation time.
 STARTUP_TIMEOUT_SEC = float(os.environ.get("LANGFLOW_BENCH_STARTUP_TIMEOUT", "180"))
 
 
+def _tcp_ready(host: str, port: int, *, connect_timeout: float = 0.5) -> bool:
+    """Return True if a TCP connection to ``(host, port)`` succeeds within the deadline."""
+    try:
+        with socket.create_connection((host, port), timeout=connect_timeout):
+            return True
+    except OSError:
+        return False
+
+
+def _drain_output(stream, stop_event: threading.Event) -> None:
+    """Forward the child's merged stdout to our stdout so CI logs show boot progress.
+
+    Runs in a background thread so the main thread can poll TCP readiness without
+    blocking on ``readline``. When ``stop_event`` fires we stop reading even if the
+    child is still producing output.
+    """
+    try:
+        for line in stream:
+            if stop_event.is_set():
+                break
+            sys.stdout.write(line)
+            sys.stdout.flush()
+    except (ValueError, OSError):
+        return
+
+
 def main() -> int:
-    """Launch `langflow run`, wait for readiness marker, terminate cleanly. Returns process exit code."""
+    """Launch `langflow run`, wait for TCP readiness, terminate cleanly."""
+    # Pre-flight: if something is already bound to the scenario port, any TCP
+    # probe we run against our own child will race against that listener and
+    # silently record a near-zero LANGFLOW_READY_MS. Refuse to measure rather
+    # than produce garbage numbers.
+    if _tcp_ready(READY_HOST, READY_PORT, connect_timeout=0.1):
+        sys.stderr.write(
+            f"ERROR: {READY_HOST}:{READY_PORT} already accepts connections before "
+            f"the supervisor started its child; another process (dev server, leftover "
+            f"benchmark boot) is squatting the port.\n",
+        )
+        return 3
+
     start = time.perf_counter()
-    proc = subprocess.Popen(
+    proc = subprocess.Popen(  # noqa: S603
         [  # noqa: S607
             "uv",
             "run",
             "langflow",
             "run",
             "--backend-only",
             "--host",
-            "127.0.0.1",
+            READY_HOST,
             "--port",
-            "7860",
+            str(READY_PORT),
             "--no-open-browser",
         ],
         stdout=subprocess.PIPE,
@@ -53,30 +98,51 @@ def main() -> int:
         sys.stderr.write("ERROR: Popen did not provide a stdout stream\n")
         return 3
 
+    stop_reader = threading.Event()
+    reader = threading.Thread(
+        target=_drain_output,
+        args=(proc.stdout, stop_reader),
+        daemon=True,
+    )
+    reader.start()
+
     ready_at: float | None = None
     try:
-        for line in proc.stdout:
-            sys.stdout.write(line)
-            sys.stdout.flush()
-            if READY_MARKER in line:
+        while True:
+            if _tcp_ready(READY_HOST, READY_PORT):
+                # Defense in depth: if the connect succeeded but our child is
+                # already gone, we hit a listener that isn't ours. Don't record
+                # a bogus LANGFLOW_READY_MS.
+                if proc.poll() is not None:
+                    sys.stderr.write(
+                        f"ERROR: {READY_HOST}:{READY_PORT} connected but child exited "
+                        f"(rc={proc.returncode}); another listener is squatting the port.\n",
+                    )
+                    return 3
                 ready_at = time.perf_counter()
                 break
+            if proc.poll() is not None:
+                sys.stderr.write(
+                    f"ERROR: process exited (rc={proc.returncode}) before {READY_HOST}:{READY_PORT} was ready\n",
+                )
+                return 3
             if time.perf_counter() - start > STARTUP_TIMEOUT_SEC:
-                sys.stderr.write(f"TIMEOUT: {STARTUP_TIMEOUT_SEC}s elapsed without seeing {READY_MARKER!r}\n")
-                proc.terminate()
+                sys.stderr.write(
+                    f"TIMEOUT: {STARTUP_TIMEOUT_SEC}s elapsed without "
+                    f"{READY_HOST}:{READY_PORT} accepting connections\n",
+                )
                 return 2
+            time.sleep(READY_POLL_INTERVAL_SEC)
     finally:
+        stop_reader.set()
         if proc.poll() is None:
             proc.send_signal(signal.SIGTERM)
             try:
                 proc.wait(timeout=10)
             except subprocess.TimeoutExpired:
                 proc.kill()
                 proc.wait(timeout=5)
-
-    if ready_at is None:
-        sys.stderr.write("ERROR: process exited before ready marker appeared\n")
-        return 3
+        reader.join(timeout=2)
 
     elapsed_ms = (ready_at - start) * 1000.0
     sys.stdout.write(f"LANGFLOW_READY_MS={elapsed_ms:.2f}\n")

src/backend/tests/benchmarks/scenarios/langflow_run.py

@@ -1,11 +1,10 @@
-"""langflow run scenario. Terminal checkpoint is uvicorn's `Application startup complete.` line.
+"""langflow run scenario. Terminal checkpoint is 127.0.0.1:7860 accepting a TCP connection.
 
 The scenario's command invokes the sibling `_langflow_supervisor.py` as a subprocess-of-a-subprocess
-so hyperfine's wall-clock measurement equals time-to-HTTP-ready.
-
-Per Claude's Discretion in 01-CONTEXT.md and Pitfall 7 in 01-RESEARCH.md: `langflow run` starts
-uvicorn and does not return. The supervisor reads stderr line-by-line, records `time.perf_counter()`
-when "Application startup complete." appears, then SIGTERMs the child.
+so hyperfine's wall-clock measurement equals time-to-HTTP-ready. `langflow run` starts uvicorn and
+does not return; the supervisor polls the bind port and SIGTERMs the child on first connect success.
+TCP readiness is used instead of scraping uvicorn's `Application startup complete.` log line because
+that line is swallowed by langflow's structlog processor pipeline.
 """
 
 from __future__ import annotations