fix(bluesky): start the bridge after the queue server answers

CHANGELOG.md

@@ -347,6 +347,11 @@ Compatibility is documented in release notes, not encoded in the version string.
 
 ### Fixed
 
+- A scan-stack deployment no longer intermittently refuses every scan with
+  "not in the list of allowed plans". The bridge opens the Run Engine worker
+  once at startup, and abandoned it whenever it won the boot race against the
+  queue server — leaving the list of runnable plans empty until someone started
+  the queue by hand. It now waits for the queue server to answer first.
 - One operator's tab switches no longer rearrange every other window of the
   same workspace: a human panel focus is now mirrored to the server silently
   (the agent can still read where the operator is looking) instead of being

src/osprey/templates/services/bluesky/docker-compose.yml.j2

@@ -22,13 +22,33 @@ services:
 {% if dev_mode | default(false) %}
         OSPREY_DEV: "1"
 {% endif %}
-{% if 'virtual_accelerator' in deployed_services %}
-    # Only when the Virtual Accelerator is co-deployed: order the
-    # bridge's startup after the VA's CA-readiness healthcheck, so the
-    # substrate devices' ophyd-async connect doesn't race the
-    # IOC's iocInit. A bridge-only deploy (no VA in deployed_services) must
-    # NOT get this — compose errors on depends_on naming an undefined service.
+    # Ordering, not convenience: the bridge opens the RE worker environment
+    # ONCE at startup and does not keep trying. `app.py`'s
+    # `_open_environment_at_startup` calls `ensure_environment`, which asks
+    # `capability()` first and returns False WITHOUT entering its retry loop
+    # when the manager is not answering yet (`manager_unreachable`); only an
+    # armed `POST /queue/start` runs it again. But `POST /queue/items`
+    # validates against `plans_allowed`, and the manager only downloads that
+    # list from the worker when the environment opens — so a bridge that boots
+    # faster than the manager refuses every enqueue with "not in the list of
+    # allowed plans", long before any start could self-heal it. Waiting for
+    # `qserver ping` to answer removes that race rather than betting on the
+    # manager's import of the bluesky/ophyd stack finishing first.
+    #
+    # Unconditional, because `queueserver` is rendered by THIS template and so
+    # is always defined. Not in tension with the rule that a Tiled outage must
+    # never block the bridge (FR4): Tiled is the read path's durable store,
+    # while the manager is the execution plane the bridge is a facade over —
+    # the same reason the VA is depended on below.
     depends_on:
+      queueserver:
+        condition: service_healthy
+{% if 'virtual_accelerator' in deployed_services %}
+      # Only when the Virtual Accelerator is co-deployed: order the
+      # bridge's startup after the VA's CA-readiness healthcheck, so the
+      # substrate devices' ophyd-async connect doesn't race the
+      # IOC's iocInit. A bridge-only deploy (no VA in deployed_services) must
+      # NOT get this — compose errors on depends_on naming an undefined service.
       virtual-accelerator:
         condition: service_healthy
 {% endif %}

tests/deployment/test_compose_generator.py

@@ -583,6 +583,29 @@ def test_bluesky_wires_va_ca_env_and_ordering_only_when_va_co_deployed() -> None
     assert "virtual-accelerator" not in without_va
 
 
+@pytest.mark.parametrize("va_deployed", [True, False])
+def test_bluesky_bridge_waits_for_the_queueserver_to_answer(va_deployed: bool) -> None:
+    """The bridge must start only after ``qserver ping`` answers — with or
+    without the VA co-deployed.
+
+    Not cosmetic ordering. The bridge opens the RE worker environment once at
+    startup (``app.py``'s ``_open_environment_at_startup``), and
+    ``ensure_environment`` gives that up WITHOUT retrying when ``capability()``
+    reports ``manager_unreachable`` — it re-runs only on an armed
+    ``POST /queue/start``. Since ``POST /queue/items`` validates against
+    ``plans_allowed``, which the manager downloads from the worker at
+    environment open, a bridge that wins the boot race against the manager
+    refuses every enqueue with "not in the list of allowed plans" and no start
+    ever gets the chance to self-heal it. Only container ordering closes that,
+    so it is asserted here rather than left to whichever process imports its
+    dependency stack faster.
+    """
+    bridge = yaml.safe_load(_render_bluesky_template(va_deployed=va_deployed))["services"][
+        "bluesky-bridge"
+    ]
+    assert bridge["depends_on"]["queueserver"] == {"condition": "service_healthy"}
+
+
 def test_bluesky_va_ca_port_defaults_when_va_config_block_absent() -> None:
     """VA in ``deployed_services`` but no ``services.virtual_accelerator`` config
     block must still render the default CA port (5064), never raise.

tests/e2e/_deploy_diagnostics.py

@@ -66,3 +66,45 @@ def dead_container_logs(project_prefix: str) -> str:
     if not sections:
         return "(no stopped containers found for this deployment)"
     return "\n".join(sections)
+
+
+def container_logs(*names: str) -> str:
+    """Return the log tail of each NAMED container, running or not.
+
+    The companion to :func:`dead_container_logs`, for the failures where every
+    container is up and the fault is in what one of them DID -- a service that
+    logged a warning and carried on. ``dead_container_logs`` skips those by
+    design; this reads them by name.
+
+    Best-effort in the same way: only ever called on an already-failing path,
+    so it reports collection failures inline instead of raising over the real
+    error.
+    """
+    sections: list[str] = []
+    for name in names:
+        try:
+            logs = subprocess.run(
+                ["docker", "logs", "--tail", str(LOG_TAIL_LINES), name],
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+        except (OSError, subprocess.SubprocessError) as exc:
+            sections.append(f"--- {name} --- (logs unavailable: {exc})")
+            continue
+        sections.append(f"--- {name} ---\n{logs.stdout}\n{logs.stderr}")
+    return "\n".join(sections)
+
+
+def queue_stack_logs(project_prefix: str) -> str:
+    """Log tails from the two containers that own the RE worker environment.
+
+    The evidence a "the worker environment never opened" failure needs and
+    that a torn-down stack cannot be asked for afterwards: the bridge logs its
+    startup open at WARNING when it fails (``app.py``'s
+    ``_open_environment_at_startup``), and the manager's own side of the same
+    story is in the queueserver's log.
+    """
+    return container_logs(
+        f"{project_prefix}-bluesky-bridge", f"{project_prefix}-bluesky-queueserver"
+    )

tests/e2e/_orm_stack.py

@@ -85,6 +85,21 @@
 # too (e.g. ``f"{project_name}-bluesky-bridge"``).
 
 
+def project_prefix(project_name: str) -> str:
+    """The ``<project>`` prefix compose gives every container and locally-built
+    image of a deploy, resolved exactly as the templates resolve it.
+
+    Container names (``<project>-bluesky-bridge``) and image tags
+    (``<project>-va:local``) are both built from it, so anything that must name
+    a deployed container -- a health probe, a log dump -- derives it here
+    rather than hardcoding a host-global name that is wrong for any other
+    project.
+    """
+    from osprey.deployment.compose_generator import resolve_project_name
+
+    return str(resolve_project_name({"project_name": project_name}))
+
+
 def _service_image(project_name: str, service: str) -> str:
     """Derive a locally-built ``<project>-<service>:local`` image tag the way
     the service compose templates do.
@@ -95,9 +110,7 @@ def _service_image(project_name: str, service: str) -> str:
     caller that force-rebuilds via ``docker rmi -f`` must target that SAME
     project-prefixed tag, never a host-global name.
     """
-    from osprey.deployment.compose_generator import resolve_project_name
-
-    return f"{resolve_project_name({'project_name': project_name})}-{service}:local"
+    return f"{project_prefix(project_name)}-{service}:local"
 
 
 def bridge_image(project_name: str) -> str:

tests/e2e/_queue_drive.py

@@ -30,6 +30,10 @@
 * The enqueue is UNARMED (no token). The queue is idle at that point, so an
   added item just sits there until the armed start -- the designed
   compose-now/arm-later split.
+
+``wait_for_worker_environment`` is the one piece here that belongs to a deploy
+FIXTURE rather than to the flow: it is the readiness gate an enqueue depends on
+and that ``/health`` does not give you. See its docstring.
 """
 
 from __future__ import annotations
@@ -47,6 +51,12 @@
 #: not to wait out its whole deadline first.
 TERMINAL_STATUSES = ("completed", "error", "stopped")
 
+#: How long a deploy fixture waits for the RE worker environment to come up.
+#: Generous on purpose: opening it connects every substrate device over Channel
+#: Access, which takes tens of seconds on a cold stack (and longer under QEMU
+#: emulation on Apple Silicon).
+WORKER_ENV_TIMEOUT_SEC = 300.0
+
 
 def request(
     base_url: str,
@@ -89,6 +99,49 @@ def request(
             return exc.code, raw.decode("utf-8", errors="replace")
 
 
+def wait_for_worker_environment(
+    base_url: str, *, timeout: float = WORKER_ENV_TIMEOUT_SEC, poll: float = 2.0
+) -> dict[str, Any]:
+    """Block until the manager reports an OPEN RE worker environment.
+
+    The gate every deploy fixture in this suite owes its own enqueue. HTTP
+    readiness on ``/health`` does NOT imply it: the bridge opens the worker
+    environment in a background task that is deliberately excluded from
+    readiness (device connect takes tens of seconds), so a stack can answer 200
+    with an empty worker namespace. ``POST /queue/items`` validates against
+    ``plans_allowed``, which the manager downloads from the worker only when
+    that environment opens -- so enqueueing too early is refused 409 "not in
+    the list of allowed plans", a message that reads like a permissions problem
+    and is nothing of the sort (the shipped ``user_group_permissions.yaml``
+    allows ``[":.*"]``; the list was empty because the namespace was).
+
+    ``manager_state`` is NOT the signal to wait on. It reads ``idle`` both
+    before the environment has ever been opened and after it is up, so gating
+    on it would be very nearly vacuous. ``worker_environment_exists`` is the
+    one field that separates the two.
+
+    Returns the manager status document that satisfied the wait. Raises
+    ``AssertionError`` naming the last status seen -- a failure here says the
+    worker environment never came up, which points at the queueserver rather
+    than at the plan the caller was about to run.
+    """
+    deadline = time.monotonic() + timeout
+    last: Any = "(no answer yet)"
+    while time.monotonic() < deadline:
+        http_status, body = request(base_url, "/queue", "GET")
+        if http_status == 200 and isinstance(body, dict):
+            last = body.get("status") or {}
+            if isinstance(last, dict) and last.get("worker_environment_exists"):
+                return dict(last)
+        else:
+            last = body
+        time.sleep(poll)
+    raise AssertionError(
+        f"the RE worker environment never opened within {timeout:.0f}s -- the queue "
+        f"cannot accept plans until it does (last manager status: {last!r})"
+    )
+
+
 def drain_pending_queue(base_url: str) -> None:
     """Remove every PENDING item, so this run's start drains only its own work.
 

tests/e2e/test_bluesky_panels_deploy.py

@@ -82,7 +82,8 @@
 
 import pytest
 
-from tests.e2e import _orm_stack
+from tests.e2e import _orm_stack, _queue_drive
+from tests.e2e._deploy_diagnostics import queue_stack_logs
 
 # Distinct from every sibling e2e module's pinned bridge port (_orm_stack.py's
 # 18102, test_bluesky_deploy.py's 18090, test_va_substrate_equivalence.py's
@@ -455,6 +456,14 @@ def deployed_stack(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Deploye
             )
         _wait_for_health(f"{BRIDGE_URL}/health", HEALTH_TIMEOUT_SEC)
         _wait_for_health(f"{BLUESKY_PANELS_URL}/health", HEALTH_TIMEOUT_SEC)
+        # HTTP readiness is not enqueue readiness -- the worker namespace an
+        # enqueue validates against exists only once the RE worker environment
+        # is open, and the bridge opens that off the readiness path. See
+        # `_queue_drive.wait_for_worker_environment`.
+        try:
+            _queue_drive.wait_for_worker_environment(BRIDGE_URL)
+        except AssertionError as exc:
+            pytest.fail(f"{exc}\n{queue_stack_logs(_orm_stack.project_prefix(PROJECT_NAME))}")
 
         plan_name, plan_args = _discover_writes_plan(correctors, bpms, limits)
 

tests/e2e/test_bluesky_queue_e2e.py

@@ -161,6 +161,9 @@
 DEPLOY_UP_TIMEOUT_SEC = 2400
 HEALTH_TIMEOUT_SEC = 420.0
 CONTAINER_HEALTH_TIMEOUT_SEC = 240.0
+# Opening the RE worker environment connects every substrate device over
+# Channel Access, which takes tens of seconds on a cold stack.
+WORKER_ENV_TIMEOUT_SEC = 300.0
 # One queued plan's wall-clock budget, and the whole-queue drain budget.
 RUN_TIMEOUT_SEC = 420.0
 DRAIN_TIMEOUT_SEC = 900.0
@@ -511,6 +514,37 @@ def _wait_for_manager_state(wanted: tuple[str, ...], timeout: float) -> str:
     raise AssertionError(f"manager never reached {wanted} within {timeout:.0f}s (last: {last!r})")
 
 
+def _wait_for_worker_environment(timeout: float) -> dict[str, Any]:
+    """Block until the manager reports an OPEN worker environment.
+
+    Container health is not enqueue readiness. The bridge opens the RE worker
+    environment in a background task deliberately excluded from readiness, and
+    `POST /queue/items` validates against `plans_allowed` -- which the manager
+    downloads from the worker only at that open. Enqueueing before it lands is
+    refused 409 "not in the list of allowed plans", which reads like a
+    permissions problem and is not one (the shipped permissions allow
+    `[":.*"]`; the list was empty because the namespace was).
+
+    `manager_state` is the wrong field to wait on -- it reads `idle` both
+    before the environment has ever opened and after it is up.
+
+    Spelled here rather than imported from `_queue_drive` for this module's
+    standing reason: the acceptance instrument for the queue surface must not
+    be written in terms of a helper that assumes that surface works.
+    """
+    deadline = time.monotonic() + timeout
+    last: Any = None
+    while time.monotonic() < deadline:
+        last = _queue_snapshot()["status"]
+        if last.get("worker_environment_exists"):
+            return dict(last)
+        time.sleep(2.0)
+    raise AssertionError(
+        f"the RE worker environment never opened within {timeout:.0f}s -- the queue "
+        f"cannot accept plans until it does (last manager status: {last!r})"
+    )
+
+
 # ---------------------------------------------------------------------------
 # Fixture: one build, one deploy, for every stage
 # ---------------------------------------------------------------------------
@@ -659,6 +693,7 @@ def stack(tmp_path_factory: pytest.TempPathFactory) -> Iterator[QueueStack]:
         _wait_for_health(f"{PANELS_URL}/health", HEALTH_TIMEOUT_SEC)
         _wait_for_container_health(QUEUESERVER_CONTAINER, CONTAINER_HEALTH_TIMEOUT_SEC)
         _wait_for_container_health(TILED_CONTAINER, CONTAINER_HEALTH_TIMEOUT_SEC)
+        _wait_for_worker_environment(WORKER_ENV_TIMEOUT_SEC)
 
         _drain_leftover_queue_items()
 

tests/e2e/test_bluesky_sandbox_escape_e2e.py

@@ -91,7 +91,8 @@
 
 import pytest
 
-from tests.e2e import _orm_stack
+from tests.e2e import _orm_stack, _queue_drive
+from tests.e2e._deploy_diagnostics import queue_stack_logs
 
 pytestmark = [
     pytest.mark.e2e,
@@ -105,6 +106,11 @@
 BRIDGE_PORT = 18105
 BRIDGE_URL = f"http://localhost:{BRIDGE_PORT}"
 
+#: Compose project this suite deploys under. Container names follow
+#: ``<project>-<service>``, so anything naming a deployed container derives it
+#: from here rather than repeating the literal.
+PROJECT_NAME = "sandbox-escape"
+
 BUILD_TIMEOUT_SEC = _orm_stack.BUILD_TIMEOUT_SEC
 DEPLOY_UP_TIMEOUT_SEC = 1200  # amd64-emulated VA image build is slow (minutes)
 HEALTH_TIMEOUT_SEC = 300.0
@@ -447,7 +453,7 @@ def deployed_sandbox_stack(
 ) -> Iterator[DeployedSandboxStack]:
     base = tmp_path_factory.mktemp("sandbox_escape_build")
     project_dir = _orm_stack.build_project_subprocess(
-        "sandbox-escape", output_dir=base, bridge_port=BRIDGE_PORT, timeout=BUILD_TIMEOUT_SEC
+        PROJECT_NAME, output_dir=base, bridge_port=BRIDGE_PORT, timeout=BUILD_TIMEOUT_SEC
     )
 
     limits = _channel_limits(project_dir)
@@ -489,6 +495,14 @@ def deployed_sandbox_stack(
                 f"--- stdout ---\n{up.stdout}\n--- stderr ---\n{up.stderr}"
             )
         _wait_for_health(f"{BRIDGE_URL}/health", HEALTH_TIMEOUT_SEC)
+        # HTTP readiness is not enqueue readiness -- the worker namespace an
+        # enqueue validates against exists only once the RE worker environment
+        # is open, and the bridge opens that off the readiness path. See
+        # `_queue_drive.wait_for_worker_environment`.
+        try:
+            _queue_drive.wait_for_worker_environment(BRIDGE_URL)
+        except AssertionError as exc:
+            pytest.fail(f"{exc}\n{queue_stack_logs(_orm_stack.project_prefix(PROJECT_NAME))}")
         yield DeployedSandboxStack(
             project_dir=project_dir,
             escape_target_sp=escape_sp,

tests/e2e/test_grid_scan_roundtrip.py

@@ -59,6 +59,7 @@
 import pytest
 
 from tests.e2e import _orm_stack, _queue_drive
+from tests.e2e._deploy_diagnostics import queue_stack_logs
 
 pytestmark = [
     pytest.mark.e2e,
@@ -172,6 +173,14 @@ def deployed_grid_scan_stack(
                 f"--- stdout ---\n{up.stdout}\n--- stderr ---\n{up.stderr}"
             )
         _orm_stack.wait_for_health(f"{BRIDGE_URL}/health", HEALTH_TIMEOUT_SEC)
+        # HTTP readiness is not enqueue readiness -- the worker namespace the
+        # enqueue validates against exists only once the RE worker environment
+        # is open, and the bridge opens that off the readiness path. See
+        # `_queue_drive.wait_for_worker_environment`.
+        try:
+            _queue_drive.wait_for_worker_environment(BRIDGE_URL)
+        except AssertionError as exc:
+            pytest.fail(f"{exc}\n{queue_stack_logs(_orm_stack.project_prefix(PROJECT_NAME))}")
         yield DeployedGridScanStack(
             project_dir=project_dir,
             corrector_name=next(iter(correctors)),