diff --git a/.fernignore b/.fernignore
index be3a412..da5ab92 100644
--- a/.fernignore
+++ b/.fernignore
@@ -10,6 +10,7 @@ src/schematic/client.py
 .claude/
 .github/CODEOWNERS
 .github/workflows/ci.yml
+.github/workflows/sdk-e2e-status.yml
 scripts/
 src/schematic/cache/
 src/schematic/event_buffer.py
diff --git a/.github/workflows/sdk-e2e-status.yml b/.github/workflows/sdk-e2e-status.yml
new file mode 100644
index 0000000..f1e2076
--- /dev/null
+++ b/.github/workflows/sdk-e2e-status.yml
@@ -0,0 +1,50 @@
+name: sdk-e2e status
+
+# Makes `sdk-e2e` a commit status on every PR so it can be a required check
+# without blocking ordinary PRs:
+#
+#   - update-wasm-v* branches (rules engine bumps opened by schematic-bot from
+#     schematic-api's rulesengine_release.yml): `pending` until the SDK E2E run
+#     that the release workflow dispatches in schematic-api reports back
+#     (its report-status job posts success/failure to this same context).
+#   - every other PR: `success` immediately; SDK E2E is not required.
+#
+# A new push to a bump branch resets the status to pending; re-run schematic-api's
+# sdk_e2e.yml with sdk-ref=<branch> to report on the new head.
+#
+# pull_request_target so the token can write statuses on fork PRs too. Nothing
+# from the PR is checked out or executed here.
+
+on:
+  pull_request_target:
+    types: [opened, synchronize, reopened]
+
+permissions: {}
+
+jobs:
+  status:
+    runs-on: ubuntu-latest
+    permissions:
+      statuses: write
+    steps:
+      - name: Set sdk-e2e status
+        env:
+          GH_TOKEN: ${{ github.token }}
+          HEAD_REF: ${{ github.event.pull_request.head.ref }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+          E2E_URL: https://github.com/SchematicHQ/schematic-api/actions/workflows/sdk_e2e.yml
+        run: |
+          case "$HEAD_REF" in
+            update-wasm-v*)
+              state=pending
+              description="Waiting for the SDK E2E run in schematic-api"
+              ;;
+            *)
+              state=success
+              description="Not a rules engine WASM bump; SDK E2E not required"
+              ;;
+          esac
+          echo "$HEAD_REF @ ${HEAD_SHA:0:8}: sdk-e2e=$state ($description)"
+          gh api "repos/$GITHUB_REPOSITORY/statuses/$HEAD_SHA" \
+            -f state="$state" -f context=sdk-e2e \
+            -f description="$description" -f target_url="$E2E_URL" > /dev/null
diff --git a/tests/datastream/redis_key_layout.json b/tests/datastream/redis_key_layout.json
new file mode 100644
index 0000000..9bf6866
--- /dev/null
+++ b/tests/datastream/redis_key_layout.json
@@ -0,0 +1,34 @@
+{
+  "description": "Redis keys the replicator writes. SDKs in replicator mode read these keys, so their key builders must produce exactly these strings (prefix included: the SDK's Redis provider prefix plus the datastream key must equal the replicator key). <VERSION> is the rules engine cache version (rulesengine.VersionKey) and is substituted at test time. Each SDK carries a unit test against these cases; change them only with a matching change in every SDK.",
+  "prefix": "schematic",
+  "cases": [
+    {
+      "kind": "flag",
+      "input": { "key": "My_Flag" },
+      "key": "schematic:flags:<VERSION>:my_flag",
+      "note": "flag keys are lowercased"
+    },
+    {
+      "kind": "company_id",
+      "input": { "id": "comp_Abc123" },
+      "key": "schematic:company:<VERSION>:comp_Abc123",
+      "note": "ids keep their case"
+    },
+    {
+      "kind": "company_lookup",
+      "input": { "key": "ExternalId", "value": "Acme-Co" },
+      "key": "schematic:company:<VERSION>:externalid:acme-co",
+      "note": "lookup key names and values are lowercased"
+    },
+    {
+      "kind": "user_id",
+      "input": { "id": "user_Xyz789" },
+      "key": "schematic:user:<VERSION>:user_Xyz789"
+    },
+    {
+      "kind": "user_lookup",
+      "input": { "key": "Email", "value": "Ada@Example.com" },
+      "key": "schematic:user:<VERSION>:email:ada@example.com"
+    }
+  ]
+}
diff --git a/tests/datastream/test_redis_key_layout.py b/tests/datastream/test_redis_key_layout.py
new file mode 100644
index 0000000..249df2b
--- /dev/null
+++ b/tests/datastream/test_redis_key_layout.py
@@ -0,0 +1,68 @@
+"""Replicator Redis key layout contract.
+
+``redis_key_layout.json`` is a copy of schematic-replicator/testdata/redis_key_layout.json:
+the keys the replicator writes. In replicator mode this SDK reads them, so the
+Redis provider prefix plus the datastream key must reproduce them exactly. The
+C# and Ruby SDKs doubled the prefix for a year with no test on either side
+(SCH-7070); this is that test.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Any, Dict
+
+import pytest
+from .test_datastream_client import MockCacheProvider
+
+from schematic.cache.redis import RedisCache
+from schematic.datastream.datastream_client import DataStreamClient, DataStreamClientOptions
+
+CACHE_VERSION = "v-test"
+FIXTURE = json.loads((Path(__file__).parent / "redis_key_layout.json").read_text())
+
+
+@pytest.fixture
+def client() -> DataStreamClient:
+    cache = MockCacheProvider()
+    c = DataStreamClient(
+        DataStreamClientOptions(
+            api_key="test-key",
+            logger=logging.getLogger("test_redis_key_layout"),
+            replicator_mode=True,
+            company_cache=cache,
+            company_lookup_cache=cache,
+            user_cache=cache,
+            user_lookup_cache=cache,
+            flag_cache=cache,
+        )
+    )
+    # Set from the replicator's health response in production.
+    c._replicator_cache_version = CACHE_VERSION
+    return c
+
+
+def _build(client: DataStreamClient, kind: str, inp: Dict[str, str]) -> str:
+    if kind == "flag":
+        return client._flag_cache_key(inp["key"])
+    if kind == "company_id":
+        return client._resource_id_cache_key("company", inp["id"])
+    if kind == "company_lookup":
+        return client._resource_key_to_cache_key("company", inp["key"], inp["value"])
+    if kind == "user_id":
+        return client._resource_id_cache_key("user", inp["id"])
+    if kind == "user_lookup":
+        return client._resource_key_to_cache_key("user", inp["key"], inp["value"])
+    raise AssertionError(f"fixture case kind {kind!r} has no builder here")
+
+
+@pytest.mark.parametrize("case", FIXTURE["cases"], ids=[c["kind"] for c in FIXTURE["cases"]])
+def test_redis_key_matches_replicator_layout(client: DataStreamClient, case: Dict[str, Any]) -> None:
+    # RedisCache's default prefix is what the testapp and docs use; the
+    # replicator's fixed prefix must be the same string.
+    redis: RedisCache[Any] = RedisCache(client=None)
+    assert redis._prefix == FIXTURE["prefix"]
+    key = redis._prefixed(_build(client, case["kind"], case["input"]))
+    assert key == case["key"].replace("<VERSION>", CACHE_VERSION)