Deploy Redis in k3s and pin the orchestrator to the Redis message store (#2662)

.github/workflows/test-integration.yml

@@ -143,10 +143,12 @@ jobs:
               && docker save egg-orchestrator:latest | sudo k3s ctr images import - \
               && docker save egg-sandbox:latest | sudo k3s ctr images import - \
               && docker save egg-litellm:latest | sudo k3s ctr images import - \
+              && docker save egg-redis:latest | sudo k3s ctr images import - \
               && docker save egg-gateway:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \
               && docker save egg-orchestrator:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \
               && docker save egg-sandbox:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \
-              && docker save egg-litellm:"$EGG_IMAGE_TAG" | sudo k3s ctr images import -; then
+              && docker save egg-litellm:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \
+              && docker save egg-redis:"$EGG_IMAGE_TAG" | sudo k3s ctr images import -; then
               echo "::endgroup::"
               echo "Image import succeeded on attempt ${attempt}"
               exit 0
@@ -175,7 +177,7 @@ jobs:
         run: |
           set -x
           docker images --format '{{.Repository}}:{{.Tag}}' \
-            | grep -E '^egg-(gateway|orchestrator|sandbox|litellm):' \
+            | grep -E '^egg-(gateway|orchestrator|sandbox|litellm|redis):' \
             | xargs -r docker rmi -f || true
           docker builder prune -af || true
           docker image prune -af || true

Makefile

@@ -49,7 +49,7 @@ EGG_IMAGE_PREFIX := $(if $(EGG_IMAGE_REGISTRY),$(EGG_IMAGE_REGISTRY)/,)
 # registry entirely, remove it from EGG_REGISTRY_IMAGES; excluded images
 # publish through the save+import path instead (slower, but entirely
 # store-to-store on this host, no registry involved).
-EGG_ALL_IMAGES := egg-gateway egg-orchestrator egg-sandbox egg-litellm
+EGG_ALL_IMAGES := egg-gateway egg-orchestrator egg-sandbox egg-litellm egg-redis
 EGG_REGISTRY_IMAGES ?= $(EGG_ALL_IMAGES)
 # Images the registry path does NOT cover (imported via k3s-import instead).
 EGG_IMPORT_IMAGES := $(filter-out $(EGG_REGISTRY_IMAGES),$(EGG_ALL_IMAGES))
@@ -518,7 +518,7 @@ build: sync-venv-if-uv
 	@$(PYTHON) scripts/prepare-sandbox-build-context.py repo-deps
 	@echo "==> Building images with tag $(EGG_IMAGE_TAG) ($(BUILD_JOBS) parallel jobs)..."
 	@$(MAKE) --no-print-directory -j$(BUILD_JOBS) $(BUILD_OUTPUT_SYNC) \
-		build-gateway build-orchestrator build-sandbox build-litellm
+		build-gateway build-orchestrator build-sandbox build-litellm build-redis
 
 # Per-image sub-targets so `build` can run them under -j. They assume the
 # repo-deps/ build context has been prepared (the `build` target does that
@@ -544,6 +544,9 @@ build-sandbox:
 build-litellm:
 	DOCKER_BUILDKIT=1 docker build $(call image_tags,egg-litellm) -f config/litellm/Dockerfile config/litellm
 
+build-redis:
+	DOCKER_BUILDKIT=1 docker build $(call image_tags,egg-redis) -f config/redis/Dockerfile config/redis
+
 # ============================================================================
 # Kubernetes (k3s) targets
 # ============================================================================
@@ -702,7 +705,8 @@ deploy: sudo-keepalive check-egg-images-present  ## Deploy egg to k3s
 		sed -e "s|egg-orchestrator:latest|$(call reg_prefix,egg-orchestrator)egg-orchestrator:$(EGG_IMAGE_TAG)|g" \
 		    -e "s|egg-gateway:latest|$(call reg_prefix,egg-gateway)egg-gateway:$(EGG_IMAGE_TAG)|g" \
 		    -e "s|egg-sandbox:latest|$(call reg_prefix,egg-sandbox)egg-sandbox:$(EGG_IMAGE_TAG)|g" \
-		    -e "s|egg-litellm:latest|$(call reg_prefix,egg-litellm)egg-litellm:$(EGG_IMAGE_TAG)|g" | \
+		    -e "s|egg-litellm:latest|$(call reg_prefix,egg-litellm)egg-litellm:$(EGG_IMAGE_TAG)|g" \
+		    -e "s|egg-redis:latest|$(call reg_prefix,egg-redis)egg-redis:$(EGG_IMAGE_TAG)|g" | \
 		kubectl apply -f - && \
 	scripts/clear-stuck-egg-pods.sh && \
 	scripts/await-egg-deploy.sh "$(EGG_IMAGE_TAG)"

config/redis/Dockerfile

@@ -0,0 +1,25 @@
+# egg-redis — stock Redis, repackaged as a locally built/published image.
+#
+# Backs the orchestrator's Redis Streams message store
+# (orchestrator/redis_message_store.py), the production backend for
+# inter-agent messages (#2662). Durability target: the BRC transcript and
+# consensus-tracker replay source must survive a mid-phase orchestrator
+# restart (#3076) — so messages live here, not in the orchestrator process.
+#
+# All egg images are built locally and published to the cluster through the
+# loopback registry / save+import flow (`make redeploy`, #2999); the cluster
+# never pulls from public registries. This thin wrapper keeps Redis on that
+# same supply chain instead of introducing the first public image reference
+# in the manifests.
+#
+# Pinned to a specific upstream release; bump deliberately. The orchestrator
+# uses Streams primitives only (XADD / XREAD BLOCK / XRANGE / XLEN / DEL) —
+# stable since Redis 5 — so patch/minor bumps are low-risk, but keep the
+# fakeredis-backed unit suite (orchestrator/tests/test_redis_message_store.py)
+# and `make test-integration` green across any bump.
+#
+# No USER directive: the Deployment's securityContext
+# (runAsUser/runAsGroup/fsGroup 1000, runAsNonRoot) governs the runtime UID,
+# and the stock entrypoint works under an arbitrary non-root UID as long as
+# /data is writable (fsGroup handles that).
+FROM redis:8.6.4-alpine

docs/architecture/coordination-state.md

@@ -181,7 +181,13 @@ and sets a health-visible degraded flag. Explicit
 level with no degraded flag. The `auto` selection semantics — Redis
 when available, in-memory fallback — are unchanged; deeper durability
 work stays in the [#3070](https://github.com/jwbron/egg/issues/3070)
-lineage. The Redis path's restart semantics are pinned by
+lineage. Since [#2662](https://github.com/jwbron/egg/issues/2662) the
+k8s manifests deploy Redis (`k8s/base/redis-deployment.yaml`) and pin
+the orchestrator to `EGG_MESSAGE_STORE_BACKEND=redis`, so the
+accidental-loss row above cannot occur in-cluster: explicit `redis`
+mode raises on an unreachable Redis instead of falling back. The
+fallback path (and its fail-loud signal) remains for `auto`-configured
+local/dev contexts. The Redis path's restart semantics are pinned by
 `orchestrator/tests/test_redis_message_store.py`: mid-phase messages
 survive a store re-instantiation against the same Redis (simulated
 orchestrator restart), while the designed `_clear_concurrent_state()`

docs/guides/concurrent-execution.md

@@ -100,7 +100,7 @@ All concurrent agent containers are wrapped with a shell script defined in `orch
 
 ## Message Bus
 
-Agents communicate with each other during concurrent execution via the orchestrator message bus (`orchestrator/message_store.py`). In production, messages are stored in Redis Streams, surviving orchestrator restarts. Messages are cleared at phase transition. In test environments, an in-memory fallback is used when Redis is not available.
+Agents communicate with each other during concurrent execution via the orchestrator message bus (`orchestrator/message_store.py`). In production, messages are stored in Redis Streams, surviving orchestrator restarts — the k8s manifests deploy a dedicated Redis (`k8s/base/redis-deployment.yaml`) and pin the orchestrator to it via `EGG_MESSAGE_STORE_BACKEND=redis` ([#2662](https://github.com/jwbron/egg/issues/2662)). Messages are cleared at phase transition. In test environments, an in-memory fallback is used when Redis is not available.
 
 ### How to Wait
 
@@ -202,7 +202,7 @@ Returns total message count and a breakdown by message type.
 
 ### Message Store Backend
 
-The message store uses Redis Streams when Redis is available, falling back to an in-memory store for tests or unconfigured environments. The backend is selected via the `EGG_MESSAGE_STORE_BACKEND` environment variable (`"auto"` by default, `"redis"` to require Redis, `"memory"` to force in-memory).
+The message store uses Redis Streams when Redis is available, falling back to an in-memory store for tests or unconfigured environments. The backend is selected via the `EGG_MESSAGE_STORE_BACKEND` environment variable (`"auto"` by default, `"redis"` to require Redis, `"memory"` to force in-memory). The k8s deployment sets `"redis"` explicitly ([#2662](https://github.com/jwbron/egg/issues/2662)): in-cluster, a Redis outage fails loudly rather than silently degrading to the in-memory store (the `auto`→memory fallback is the mid-phase-restart loss risk that the [#3077](https://github.com/jwbron/egg/issues/3077) slice-6 health flag exists to catch). `integration_tests/test_message_store_backend.py` pins the deployed backend choice and round-trips a message through the live Redis.
 
 **Long-poll semantics (both backends):** `GET /messages/wait?for=<TYPE>&timeout=<s>` blocks on both backends until a matching message arrives or the timeout elapses. The in-memory store implements blocking via a per-pipeline `threading.Condition`; the Redis backend uses `XREAD BLOCK` with a server-side type-filter loop. The silent non-blocking fallback that previously lived in `routes/messages.py` was removed in [#1897](https://github.com/jwbron/egg/issues/1897) so backend misconfiguration fails loudly in CI instead of returning empty results. See [Agent Wait Patterns](../reference/agent-wait-patterns.md#3-exit-code-contract-for-egg-orch-message-wait) for the full exit-code contract and the `EGG_MESSAGE_POLL_MAX_WAIT` cap.
 

integration_tests/test_message_store_backend.py

@@ -0,0 +1,126 @@
+"""End-to-end coverage for the deployed message-store backend (#2662).
+
+The k8s manifests pin the orchestrator to the Redis Streams backend
+(``EGG_MESSAGE_STORE_BACKEND=redis`` + the ``redis`` Deployment/Service in
+``k8s/base/``). Before #2662 the cluster deployed no Redis, so production
+silently ran the in-memory ``MessageStore`` and ``RedisMessageStore`` had
+zero live coverage — everything above unit tier went through ``fakeredis``.
+
+These tests run against the live cluster ``make test-integration`` deploys:
+
+* the deployed orchestrator explicitly selects the ``redis`` backend (the
+  silent ``auto``→memory fallback is the #3076 mid-phase-restart loss risk
+  that #3077 slice-6's degraded health flag exists to catch);
+* ``/api/v1/health`` does not report the fallback marker; and
+* a real XADD/XRANGE/DEL round-trip works from inside the orchestrator
+  container, through the production selection logic in
+  ``message_store._create_message_store()`` — exercising the env wiring,
+  the in-image redis client, Service DNS, and the live Redis in one path.
+
+The Streams semantics themselves (filters, blocking reads, cursor
+staleness) are pinned by ``orchestrator/tests/test_redis_message_store.py``.
+"""
+
+import subprocess
+import uuid
+
+import requests
+
+from integration_tests.conftest import EggStack
+
+_NS = "egg-system"
+
+
+def _kubectl(*args: str, timeout: int = 30) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["kubectl", "-n", _NS, *args],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+    )
+
+
+class TestMessageStoreBackend:
+    def test_deployed_backend_is_explicit_redis(self, egg_stack: EggStack) -> None:
+        """The orchestrator Deployment pins EGG_MESSAGE_STORE_BACKEND=redis.
+
+        Explicit ``redis`` mode cannot silently fall back to in-memory:
+        creation raises instead. A manifest regression to ``auto`` (or
+        dropping the env var, whose default is ``auto``) fails here. The
+        real end-to-end coverage that the pinned backend actually works
+        lives in ``test_redis_streams_roundtrip_from_orchestrator_pod``.
+        """
+        result = _kubectl(
+            "get",
+            "deployment",
+            "orchestrator",
+            "-o",
+            "jsonpath={.spec.template.spec.containers[0].env"
+            "[?(@.name=='EGG_MESSAGE_STORE_BACKEND')].value}",
+        )
+        assert result.returncode == 0, result.stderr
+        assert result.stdout.strip() == "redis", (
+            "Deployed orchestrator must pin EGG_MESSAGE_STORE_BACKEND=redis; "
+            f"got {result.stdout.strip()!r}. See #2662."
+        )
+
+    def test_health_reports_message_store_ok(self, orchestrator_url: str) -> None:
+        """/api/v1/health must not carry the auto→memory fallback marker.
+
+        Scope caveat: the marker (``is_memory_fallback_degraded()``) is
+        only ever set on the *auto*→memory path, so in the explicit
+        ``redis`` mode this deployment pins it is structurally always
+        absent — a healthy ``{"status": "ok"}`` here does NOT prove Redis
+        is reachable (an unreachable Redis 500s ops without flipping this
+        flag). This is a cheap guard that the orchestrator came up without
+        degrading; the round-trip test below carries the real proof that
+        the orchestrator→Redis path works.
+        """
+        resp = requests.get(f"{orchestrator_url}/api/v1/health", timeout=10)
+        body = resp.json()
+        component = body.get("components", {}).get("message_store")
+        assert component == {"status": "ok"}, (
+            f"message_store health component is {component!r} — the deployed "
+            "orchestrator fell back to the in-memory backend (#3077 slice-6 "
+            "marker). Check the redis Deployment and REDIS_HOST wiring."
+        )
+
+    def test_redis_streams_roundtrip_from_orchestrator_pod(self, egg_stack: EggStack) -> None:
+        """Live XADD/XRANGE/DEL through the production selection path.
+
+        Runs a fresh Python process inside the orchestrator container so
+        ``get_message_store()`` re-runs backend selection with the pod's
+        real env. In explicit ``redis`` mode any failure (missing client
+        package, DNS, connection, AUTH) raises rather than falling back,
+        so a passing run proves the orchestrator→Redis path end to end.
+        """
+        pipeline_id = f"itest-redis-{uuid.uuid4().hex[:8]}"
+        script = (
+            "from message_store import Message, get_message_store\n"
+            "store = get_message_store()\n"
+            f"pid = {pipeline_id!r}\n"
+            "store.add_message(Message(pipeline_id=pid, from_role='itest',"
+            " message_type='PROGRESS', body='redis-e2e'))\n"
+            "msgs = store.get_messages(pid)\n"
+            "assert [m.body for m in msgs] == ['redis-e2e'], msgs\n"
+            "cleared = store.clear(pid)\n"
+            "assert cleared == 1, cleared\n"
+            "print('BACKEND', type(store).__name__)\n"
+        )
+        result = _kubectl(
+            "exec",
+            "deploy/orchestrator",
+            "--",
+            "python",
+            "-c",
+            script,
+            timeout=60,
+        )
+        assert result.returncode == 0, (
+            f"in-pod round-trip failed (rc={result.returncode}):\n"
+            f"stdout: {result.stdout}\nstderr: {result.stderr}"
+        )
+        assert "BACKEND RedisMessageStore" in result.stdout, (
+            f"expected the RedisMessageStore backend, got: {result.stdout!r}"
+        )

k8s/base/kustomization.yaml

@@ -13,4 +13,8 @@ resources:
   - litellm-configmap.yaml
   - litellm-deployment.yaml
   - litellm-service.yaml
+  # Redis Streams message-store backend (issue #2662) — only the
+  # orchestrator talks to it (EGG_MESSAGE_STORE_BACKEND=redis).
+  - redis-deployment.yaml
+  - redis-service.yaml
   - network-policies.yaml

k8s/base/orchestrator-deployment.yaml

@@ -102,6 +102,20 @@ spec:
             # (`egg-sandbox:latest`), and nothing exists at docker.io/egg.
             - name: EGG_SANDBOX_IMAGE
               value: "egg-sandbox:latest"
+            # Redis Streams message store (#2662). Explicitly `redis`, not
+            # `auto`: the silent auto→memory fallback is the #3076
+            # mid-phase-restart message-loss risk that #3077 slice-6's
+            # degraded health flag exists to catch — in-cluster we want a
+            # Redis outage to fail loudly instead. The store singleton is
+            # created lazily and a failed creation leaves it unset, so the
+            # orchestrator naturally retries until Redis is reachable (no
+            # startup-ordering dependency on the redis Deployment).
+            - name: EGG_MESSAGE_STORE_BACKEND
+              value: "redis"
+            - name: REDIS_HOST
+              value: "redis.egg-system.svc.cluster.local"
+            - name: REDIS_PORT
+              value: "6379"
             # #2528: read per-repo role_patterns from the same
             # repositories.yaml the gateway uses so plan-time validation
             # and the planner-prompt boundaries match push-time