[issue-3077][slice-6/6] Bounded durability: fail-loud...

orchestrator/message_store.py

@@ -586,6 +586,73 @@ def clear(self, pipeline_id: str) -> int:
 _store_lock = threading.Lock()
 
 
+# Issue #3077 slice-6 — fail-loud signal for the auto→memory fallback.
+#
+# When ``EGG_MESSAGE_STORE_BACKEND`` is unset / ``"auto"`` and Redis is
+# unreachable, the backend selection falls back to the in-memory store.
+# That fallback carries the #3076 mid-phase-restart message-loss risk
+# operators didn't opt into: a worker restart between BRC events drops
+# whatever is in the in-process dict. The pipeline keeps running but the
+# event-pump runs blind, and the consensus tracker silently de-syncs.
+#
+# We surface that risk two ways:
+#
+# 1. An error-level structured log with the stable marker constant
+#    below, so log scrapers and the orchestrator's structured-log
+#    consumer can alert on a single pinned token rather than
+#    string-matching prose.
+# 2. A module-level degraded flag exposed via
+#    :func:`is_memory_fallback_degraded` that ``/api/v1/health`` reads
+#    into ``components.message_store`` (without touching ``MessageStore``
+#    itself — see the issue #1897 TASK-4-3 isolation invariant).
+#
+# Explicit ``EGG_MESSAGE_STORE_BACKEND=memory`` (dev / test intent) emits
+# at *warning* level and does NOT set the degraded flag — the operator
+# opted in, so it isn't a degradation.
+#
+# Per HITL Q3 / task-6-1: the orchestrator does NOT refuse to run, and
+# the ``auto`` selection behavior (redis-when-available, memory fallback)
+# is unchanged. To avoid spamming integration-test harnesses that reset
+# the singleton many times per pytest process, each of the two warn /
+# error log emissions is once-per-process; ``reset_message_store`` does
+# NOT clear those once-flags. Tests that exercise the slice-6 signal
+# itself use :func:`_reset_memory_fallback_state_for_test` to re-arm.
+MEMORY_FALLBACK_MARKER = "MESSAGE_STORE_AUTO_FALLBACK_TO_MEMORY"
+_memory_fallback_degraded: bool = False
+_memory_fallback_logged: bool = False
+_memory_explicit_logged: bool = False
+
+
+def is_memory_fallback_degraded() -> bool:
+    """Return ``True`` iff ``auto`` backend selection fell back to in-memory.
+
+    Surfaced by ``/api/v1/health`` under ``components.message_store`` so
+    operators can see the #3076 mid-phase-restart loss risk before it
+    bites. Defaults to ``False``; flipped to ``True`` the first time
+    :func:`_create_message_store` lands on the auto→memory fallback in
+    this process. Cleared by
+    :func:`_reset_memory_fallback_state_for_test` (test-only).
+    """
+    return _memory_fallback_degraded
+
+
+def _reset_memory_fallback_state_for_test() -> None:
+    """Test-only helper for #3077 slice-6.
+
+    Resets the once-per-process fail-loud signal state so a follow-up
+    :func:`_create_message_store` call re-emits the marker log and
+    re-sets the degraded flag. NOT called by :func:`reset_message_store`
+    on purpose: existing integration-test reset cycles would otherwise
+    re-spam the error log every time they re-instantiate the singleton.
+    Production callers must not invoke this.
+    """
+    global _memory_fallback_degraded
+    global _memory_fallback_logged, _memory_explicit_logged
+    _memory_fallback_degraded = False
+    _memory_fallback_logged = False
+    _memory_explicit_logged = False
+
+
 def get_message_store() -> MessageStore:
     """Get the singleton message store.
 
@@ -601,15 +668,44 @@ def get_message_store() -> MessageStore:
 
 
 def _create_message_store() -> MessageStore:
-    """Create the appropriate message store backend."""
+    """Create the appropriate message store backend.
+
+    Selection precedence is unchanged from the issue #1897 design (HITL
+    Q3 of #3077 forbids changing it): ``EGG_MESSAGE_STORE_BACKEND``
+    chooses ``memory`` / ``redis`` / ``auto``, with ``auto`` (the
+    default) probing Redis and falling back to in-memory on any failure.
+    Slice-6 of #3077 layers the fail-loud signal on top of that
+    selection — see the module-level commentary on
+    ``MEMORY_FALLBACK_MARKER``.
+    """
     import os
 
+    global _memory_fallback_degraded
+    global _memory_fallback_logged, _memory_explicit_logged
+
+    # The once-flag check-then-set below (``_memory_fallback_logged`` /
+    # ``_memory_explicit_logged``) is unsynchronized, but ``_store_lock``
+    # is the real guard: production only reaches here via
+    # ``get_message_store()``'s double-checked locking, which serializes
+    # creation. The degraded-flag set is monotonic, so even an
+    # unsynchronized direct caller's worst case is a duplicate log line,
+    # never a missed degradation.
     redis_host = os.environ.get("REDIS_HOST", "localhost")
     redis_port = int(os.environ.get("REDIS_PORT", "6379"))
     redis_db = int(os.environ.get("REDIS_MESSAGE_DB", "1"))  # Separate DB from other Redis usage
     use_redis = os.environ.get("EGG_MESSAGE_STORE_BACKEND", "auto")
 
     if use_redis == "memory":
+        # #3077 slice-6: explicit dev/test intent. Warning level, no
+        # degraded flag — the operator opted in. Once-per-process to
+        # avoid log spam from test harnesses that reset the singleton.
+        if not _memory_explicit_logged:
+            _memory_explicit_logged = True
+            logger.warning(
+                "Using in-memory message store "
+                "(explicit EGG_MESSAGE_STORE_BACKEND=memory); "
+                "mid-phase restarts will drop in-flight messages",
+            )
         return MessageStore()
 
     if use_redis in ("redis", "auto"):
@@ -625,15 +721,37 @@ def _create_message_store() -> MessageStore:
         except Exception as e:
             if use_redis == "redis":
                 raise  # Explicit Redis mode — fail hard
-            logger.warning(
-                "Redis unavailable, falling back to in-memory message store",
-                extra={"error": str(e)},
-            )
+            # #3077 slice-6: auto→memory is the #3076 mid-phase-restart
+            # loss risk operators didn't opt into. Flip the health-
+            # surface degraded flag (the /api/v1/health route reads it
+            # without touching MessageStore — see issue #1897 TASK-4-3)
+            # and emit a single error-level log with the stable marker
+            # token at most once per process.
+            _memory_fallback_degraded = True
+            if not _memory_fallback_logged:
+                _memory_fallback_logged = True
+                logger.error(
+                    "%s: Redis unavailable, falling back to in-memory "
+                    "message store; mid-phase restarts will drop in-flight "
+                    "messages (see issue #3076)",
+                    MEMORY_FALLBACK_MARKER,
+                    extra={
+                        "marker": MEMORY_FALLBACK_MARKER,
+                        "error": str(e),
+                    },
+                )
 
     return MessageStore()
 
 
 def reset_message_store() -> None:
-    """Reset the singleton message store (for testing)."""
+    """Reset the singleton message store (for testing).
+
+    Does NOT reset the slice-6 fail-loud signal state (#3077): the
+    fallback log is once-per-process so integration-test harnesses that
+    reset the singleton many times don't re-spam the error log. Tests
+    that need to observe a fresh emission use
+    :func:`_reset_memory_fallback_state_for_test`.
+    """
     global _message_store
     _message_store = None

orchestrator/routes/health.py

@@ -29,6 +29,16 @@
 if _shared_path.exists() and str(_shared_path) not in sys.path:
     sys.path.insert(0, str(_shared_path))
 
+# Issue #3077 slice-6 — surface the message-store fail-loud degraded
+# signal on the health endpoint. We import the *module* and read
+# ``is_memory_fallback_degraded`` / ``MEMORY_FALLBACK_MARKER`` off it
+# rather than the singleton accessor or ``MessageStore`` class, so the
+# issue #1897 TASK-4-3 isolation invariant (``GET /api/v1/health`` MUST
+# NOT call into ``MessageStore``) is preserved verbatim. The pure
+# module-level getter is not patched by that test's
+# ``message_store.get_message_store`` / ``message_store.MessageStore``
+# mocks.
+import message_store as _message_store_module
 from egg_health import HealthTracker
 from state_store_probe import get_state_store_probe, probe_state_store_at
 
@@ -85,6 +95,18 @@ def health_check() -> tuple[Response, int]:
     the first probe completes); ``components.state_store_summary``
     carries the human-readable aggregate string in those cases.
 
+    ``components.message_store`` (#3077 slice-6) surfaces the
+    fail-loud signal for an unintentional auto→memory message-store
+    fallback. ``{"status": "ok"}`` in the common case; on auto→memory
+    fallback the value becomes ``{"status": "degraded", "reason":
+    "MESSAGE_STORE_AUTO_FALLBACK_TO_MEMORY"}`` and the top-level
+    ``status`` reads ``"degraded"`` regardless of state-store health.
+    Explicit ``EGG_MESSAGE_STORE_BACKEND=memory`` (operator opted in)
+    is NOT surfaced as degradation. The flag is read off the
+    ``message_store`` module's pure getter so the issue #1897 TASK-4-3
+    isolation invariant (no ``MessageStore`` method calls on the
+    request path) holds.
+
     Response::
 
         {
@@ -94,6 +116,8 @@ def health_check() -> tuple[Response, int]:
             "components": {
                 "state_store": {"<repo>": {"status": "ok"} | {"status": "error", "error": "..."}},
                 "state_store_summary": "ok" | "<aggregate error>",
+                "message_store": {"status": "ok"}
+                                 | {"status": "degraded", "reason": "MESSAGE_STORE_AUTO_FALLBACK_TO_MEMORY"},
                 "docker": "unknown"
             },
             "process_start_time": "...",
@@ -104,7 +128,25 @@ def health_check() -> tuple[Response, int]:
         }
     """
     snap = get_state_store_probe().snapshot()
-    healthy = bool(snap["healthy"])
+    state_store_healthy = bool(snap["healthy"])
+
+    # #3077 slice-6: message-store fail-loud signal. ``is_memory_fallback_degraded``
+    # is a pure module-level read — it does NOT instantiate or call any
+    # ``MessageStore`` method, so the issue #1897 TASK-4-3 isolation
+    # invariant is preserved.
+    message_store_degraded = _message_store_module.is_memory_fallback_degraded()
+    if message_store_degraded:
+        message_store_component: dict[str, str] = {
+            "status": "degraded",
+            "reason": _message_store_module.MEMORY_FALLBACK_MARKER,
+        }
+    else:
+        message_store_component = {"status": "ok"}
+
+    # Top-level ``status`` is healthy only if every observed subsystem
+    # is healthy. The per-component map carries the detail.
+    healthy = state_store_healthy and not message_store_degraded
+
     # Dual-write to _health_tracker: the BG probe's on_observation
     # callback records the raw probe result at probe-interval cadence;
     # this request-path record() captures the staleness-corrected value
@@ -121,6 +163,7 @@ def health_check() -> tuple[Response, int]:
         "components": {
             "state_store": snap["repos"],
             "state_store_summary": snap["message"],
+            "message_store": message_store_component,
             "docker": "unknown",
         },
         "process_start_time": tracker_snapshot["process_start_time"],

orchestrator/routes/pipelines.py

@@ -17391,7 +17391,13 @@ def _probe_parent_branch_exists(parent_branch: str) -> bool:
                 # surface picks up the cascade-block event (TASK-3-4
                 # emission path).
                 try:
-                    from orchestrator.message_store import Message, get_message_store
+                    try:
+                        from message_store import Message, get_message_store
+                    except ImportError:
+                        from ..message_store import (  # type: ignore[no-redef]
+                            Message,
+                            get_message_store,
+                        )
 
                     msg = Message(
                         pipeline_id=pipeline_id,

orchestrator/tests/test_health_check_integration.py

@@ -244,6 +244,23 @@ def _reset_state_store_probe(self):
         finally:
             reset_state_store_probe_for_test()
 
+    @pytest.fixture(autouse=True)
+    def _reset_message_store_fallback(self):
+        """``message_store._memory_fallback_degraded`` is a once-per-process
+        global (#3077 slice-6) flipped to ``True`` the first time ``auto``
+        backend selection falls back to in-memory — which happens in CI where
+        Redis is unavailable. ``/api/v1/health`` reads it into
+        ``components.message_store`` and degrades the top-level ``status``, so
+        without this reset these state-store-focused tests flake to
+        ``degraded`` based on sibling-test execution order."""
+        import message_store
+
+        message_store._reset_memory_fallback_state_for_test()
+        try:
+            yield
+        finally:
+            message_store._reset_memory_fallback_state_for_test()
+
     @pytest.fixture(autouse=True)
     def _reset_health_tracker(self):
         """``routes.health._health_tracker`` is a module-level singleton.

orchestrator/tests/test_health_check_lifecycle_integration.py

@@ -286,6 +286,23 @@ def _reset_state_store_probe(self):
         finally:
             reset_state_store_probe_for_test()
 
+    @pytest.fixture(autouse=True)
+    def _reset_message_store_fallback(self):
+        """``message_store._memory_fallback_degraded`` is a once-per-process
+        global (#3077 slice-6) flipped to ``True`` the first time ``auto``
+        backend selection falls back to in-memory — which happens in CI where
+        Redis is unavailable. ``/api/v1/health`` reads it into
+        ``components.message_store`` and degrades the top-level ``status``, so
+        without this reset these state-store-focused tests flake to
+        ``degraded`` based on sibling-test execution order."""
+        import message_store
+
+        message_store._reset_memory_fallback_state_for_test()
+        try:
+            yield
+        finally:
+            message_store._reset_memory_fallback_state_for_test()
+
     @pytest.fixture(autouse=True)
     def _reset_health_tracker(self):
         """``routes.health._health_tracker`` is a module-level singleton.

orchestrator/tests/test_health_routes.py

@@ -307,3 +307,90 @@ def test_health_does_not_invoke_state_store_on_request_path(self, client):
         with patch("state_store.get_state_store", side_effect=err):
             response = client.get("/api/v1/health")
         assert response.status_code == 200
+
+
+class TestMessageStoreFailLoudSurface:
+    """Issue #3077 slice-6 — ``/api/v1/health`` surfaces the message-store
+    fail-loud signal under ``components.message_store``.
+
+    These tests pair with ``test_message_store.py::TestMemoryFallbackFailLoudSignal``
+    — that suite locks the module-side behavior; this one locks the
+    operator-visible surface so a future refactor of either side
+    surfaces here as a clear regression.
+    """
+
+    @pytest.fixture
+    def client(self):
+        from flask import Flask
+        from routes.health import health_bp
+
+        app = Flask(__name__)
+        app.register_blueprint(health_bp)
+        app.config["TESTING"] = True
+        return app.test_client()
+
+    @pytest.fixture(autouse=True)
+    def _reset_state(self):
+        """Each test starts with the slice-6 fail-loud state cleared."""
+        import message_store
+        from state_store_probe import reset_state_store_probe_for_test
+
+        message_store._reset_memory_fallback_state_for_test()
+        reset_state_store_probe_for_test()
+        try:
+            yield
+        finally:
+            message_store._reset_memory_fallback_state_for_test()
+            reset_state_store_probe_for_test()
+
+    def test_health_reports_message_store_ok_by_default(self, client):
+        """No degraded flag → ``components.message_store`` reads ``ok``."""
+        response = client.get("/api/v1/health")
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["components"]["message_store"] == {"status": "ok"}
+
+    def test_health_surfaces_memory_fallback_degradation(self, client):
+        """auto→memory fallback flag → degraded component + degraded top-level."""
+        import message_store
+
+        # Simulate the auto→memory fallback having tripped earlier in the
+        # process. We flip the module flag directly rather than going
+        # through _create_message_store, because the patched
+        # ``MessageStore`` invariant test in this file forbids touching
+        # ``MessageStore`` from the request path — this test still must
+        # not call it.
+        with patch.object(message_store, "_memory_fallback_degraded", True):
+            response = client.get("/api/v1/health")
+
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["components"]["message_store"] == {
+            "status": "degraded",
+            "reason": message_store.MEMORY_FALLBACK_MARKER,
+        }
+        # Top-level status MUST reflect the degradation so dashboards
+        # and ``mcp__egg__check_health`` branch on a single field.
+        assert body["status"] == "degraded"
+
+    def test_health_endpoint_message_store_surface_does_not_call_messagestore(self, client):
+        """The slice-6 surface MUST stay on the issue-#1897 TASK-4-3 invariant.
+
+        Patch ``MessageStore`` + ``get_message_store`` to raise on any
+        call; the slice-6 read of the module-level degraded flag must
+        keep ``/health`` returning 200. Regression lock so a future
+        refactor that wires the health route through a ``MessageStore``
+        method on its way to the flag surfaces here.
+        """
+        err = RuntimeError(
+            "MessageStore MUST NOT be called from /api/v1/health (issue #1897 TASK-4-3)"
+        )
+        with (
+            patch("message_store.get_message_store", side_effect=err),
+            patch("message_store.MessageStore", side_effect=err),
+        ):
+            response = client.get("/api/v1/health")
+        assert response.status_code == 200
+        body = response.get_json()
+        # And the slice-6 surface is still rendered.
+        assert "message_store" in body["components"]