Edge compatibility tests + URL-encode header param values

19 tests that prove Edge caching is possible before Edge exists:

- Deterministic JSON (byte-identical responses for same input)
- Sorted JSON keys for consistent cache keys
- Cache-Control: public on context GETs, no-store on mutations/errors
- Vary: Authorization, Cookie differentiates by auth state
- Auth-dependent responses: same URL, different user → different body
- X-Mizan-Invalidate header round-trip: format → parse → verify
- Header matches JSON body invalidation targets
- Special characters in param values: semicolons, spaces, quotes
  are URL-encoded to prevent delimiter collisions
- Large invalidation sets (20 contexts) serialize and parse correctly
- Concurrent mutations produce independent, correct headers
- Empty invalidation: no affects → no header, no body key
- Param order irrelevant for response determinism

Design decision: param values in X-Mizan-Invalidate are URL-encoded
(percent-encoded). This prevents semicolon collision when values
contain the delimiter character.

301 Django tests pass.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

packages/mizan-django/src/mizan/client/executor.py

@@ -274,15 +274,18 @@ def _format_invalidate_header(
     Format invalidation targets as X-Mizan-Invalidate header value.
 
     Format: comma-separated contexts. Semicolon-separated params per context.
+    Param values are URL-encoded to prevent delimiter collisions.
 
     Examples:
         ["user"]                     → "user"
         ["user", "notifications"]    → "user, notifications"
         [{"context": "user", "params": {"user_id": 5}}]
                                      → "user;user_id=5"
-        ["notifications", {"context": "user", "params": {"user_id": 5, "org_id": 3}}]
+        [{"context": "search", "params": {"q": "hello world"}}]
-                                     → "notifications, user;user_id=5;org_id=3"
+                                     → "search;q=hello%20world"
     """
+    from urllib.parse import quote
+
     parts = []
     for entry in invalidate:
         if isinstance(entry, str):
@@ -291,7 +294,10 @@ def _format_invalidate_header(
             ctx = entry["context"]
             params = entry.get("params", {})
             if params:
-                param_str = ";".join(f"{k}={v}" for k, v in sorted(params.items()))
+                param_str = ";".join(
+                    f"{quote(str(k), safe='')}={quote(str(v), safe='')}"
+                    for k, v in sorted(params.items())
+                )
                 parts.append(f"{ctx};{param_str}")
             else:
                 parts.append(ctx)

packages/mizan-django/src/mizan/tests/test_core.py

@@ -1911,3 +1911,381 @@ class HTTPIntegrationTests(TestCase):
         """POST /api/mizan/ctx/user/ is rejected."""
         response = self.client.post("/api/mizan/ctx/user/")
         self.assertEqual(response.status_code, 405)
+
+
+# =============================================================================
+# Edge Compatibility Tests — Prove CDN caching works before Edge exists
+# =============================================================================
+
+
+class EdgeCompatibilityTests(TestCase):
+    """
+    Tests that prove Edge caching is possible. Every failure mode that
+    would break a CDN layer is tested here without building the CDN.
+    """
+
+    def setUp(self):
+        clear_registry()
+
+        UserCtx = ReactContext("user")
+
+        class ProfileOut(BaseModel):
+            name: str
+            email: str
+
+        class OrdersOut(BaseModel):
+            count: int
+
+        @client(context=UserCtx)
+        def user_profile(request: HttpRequest, user_id: int) -> ProfileOut:
+            return ProfileOut(name=f"user_{user_id}", email=f"user{user_id}@test.com")
+
+        @client(context=UserCtx)
+        def user_orders(request: HttpRequest, user_id: int) -> OrdersOut:
+            return OrdersOut(count=user_id * 10)
+
+        @client(affects=UserCtx)
+        def update_profile(request: HttpRequest, user_id: int, name: str) -> ProfileOut:
+            return ProfileOut(name=name, email=f"user{user_id}@test.com")
+
+        @client(affects="user_profile")
+        def update_name(request: HttpRequest, user_id: int, name: str) -> ProfileOut:
+            return ProfileOut(name=name, email=f"user{user_id}@test.com")
+
+        register(user_profile, "user_profile")
+        register(user_orders, "user_orders")
+        register(update_profile, "update_profile")
+        register(update_name, "update_name")
+
+    def tearDown(self):
+        clear_registry()
+
+    # ── Deterministic JSON ──────────────────────────────────────────────────
+
+    def test_deterministic_json_output(self):
+        """Same request produces byte-identical response body. Cache keys depend on this."""
+        r1 = self.client.get("/api/mizan/ctx/user/?user_id=5")
+        r2 = self.client.get("/api/mizan/ctx/user/?user_id=5")
+
+        self.assertEqual(r1.content, r2.content)
+
+    def test_json_keys_sorted(self):
+        """JSON keys are sorted for deterministic output."""
+        response = self.client.get("/api/mizan/ctx/user/?user_id=5")
+        raw = response.content.decode()
+
+        # Parse and re-serialize with sort_keys to compare
+        data = json.loads(raw)
+        canonical = json.dumps(data, sort_keys=True)
+
+        # The top-level keys should be sorted
+        keys = list(json.loads(raw).keys())
+        self.assertEqual(keys, sorted(keys))
+
+    # ── Cache-Control correctness ───────────────────────────────────────────
+
+    def test_context_get_is_cacheable(self):
+        """Context GET has Cache-Control that allows CDN caching."""
+        response = self.client.get("/api/mizan/ctx/user/?user_id=5")
+
+        cc = response["Cache-Control"]
+        self.assertIn("public", cc)
+        self.assertIn("stale-while-revalidate", cc)
+        # Must NOT have no-store or private
+        self.assertNotIn("no-store", cc)
+        self.assertNotIn("private", cc)
+
+    def test_mutation_post_not_cacheable(self):
+        """Mutation POST has no-store. CDN must never cache mutations."""
+        response = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "update_profile", "args": {"user_id": 5, "name": "X"}}),
+            content_type="application/json",
+        )
+
+        self.assertEqual(response["Cache-Control"], "no-store")
+
+    def test_error_response_not_cacheable(self):
+        """Error responses have no-store. CDN must not cache errors."""
+        response = self.client.get("/api/mizan/ctx/nonexistent/")
+
+        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response["Cache-Control"], "no-store")
+
+    # ── Vary header correctness ─────────────────────────────────────────────
+
+    def test_vary_header_present(self):
+        """Context GET includes Vary for auth-dependent caching."""
+        response = self.client.get("/api/mizan/ctx/user/?user_id=5")
+
+        vary = response["Vary"]
+        self.assertIn("Authorization", vary)
+        self.assertIn("Cookie", vary)
+
+    def test_different_params_different_response(self):
+        """Different query params produce different response bodies (different cache entries)."""
+        r1 = self.client.get("/api/mizan/ctx/user/?user_id=5")
+        r2 = self.client.get("/api/mizan/ctx/user/?user_id=6")
+
+        self.assertNotEqual(r1.content, r2.content)
+
+        d1 = r1.json()
+        d2 = r2.json()
+        self.assertEqual(d1["user_profile"]["name"], "user_5")
+        self.assertEqual(d2["user_profile"]["name"], "user_6")
+
+    # ── X-Mizan-Invalidate header round-trip ────────────────────────────────
+
+    def test_invalidation_header_parseable(self):
+        """X-Mizan-Invalidate header can be parsed back to structured data."""
+        from mizan.client.executor import _format_invalidate_header
+
+        response = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "update_profile", "args": {"user_id": 5, "name": "X"}}),
+            content_type="application/json",
+        )
+
+        header = response["X-Mizan-Invalidate"]
+
+        # Parse the header (this is what Edge would do)
+        entries = []
+        for part in header.split(", "):
+            segments = part.split(";")
+            context_name = segments[0]
+            params = {}
+            for seg in segments[1:]:
+                k, v = seg.split("=", 1)
+                params[k] = v
+            entries.append({"context": context_name, "params": params})
+
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0]["context"], "user")
+        self.assertEqual(entries[0]["params"]["user_id"], "5")
+
+    def test_invalidation_header_matches_json_body(self):
+        """Header and JSON body carry the same invalidation targets."""
+        response = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "update_profile", "args": {"user_id": 5, "name": "X"}}),
+            content_type="application/json",
+        )
+
+        data = response.json()
+        header = response["X-Mizan-Invalidate"]
+
+        # Both should reference "user" context with user_id=5
+        self.assertEqual(data["invalidate"][0]["context"], "user")
+        self.assertEqual(data["invalidate"][0]["params"]["user_id"], 5)
+        self.assertIn("user;user_id=5", header)
+
+    def test_function_level_invalidation_header(self):
+        """Function-level affects produces the function name in the header."""
+        response = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "update_name", "args": {"user_id": 7, "name": "X"}}),
+            content_type="application/json",
+        )
+
+        header = response["X-Mizan-Invalidate"]
+        data = response.json()
+
+        # Function-level: "user_profile" not "user"
+        self.assertIn("user_profile", header)
+        self.assertEqual(data["invalidate"][0]["context"], "user_profile")
+
+    def test_no_invalidation_header_on_non_mutation(self):
+        """Context GETs do not have X-Mizan-Invalidate."""
+        response = self.client.get("/api/mizan/ctx/user/?user_id=5")
+
+        self.assertNotIn("X-Mizan-Invalidate", response)
+
+    # ── Query param ordering doesn't affect content ─────────────────────────
+
+    def test_param_order_irrelevant(self):
+        """Different query param ordering produces same content (cache key normalization)."""
+        @client(context=ReactContext("multi"))
+        def multi_param(request: HttpRequest, a: int, b: int) -> ValidOutput:
+            return ValidOutput(valid=True)
+
+        register(multi_param, "multi_param")
+
+        r1 = self.client.get("/api/mizan/ctx/multi/?a=1&b=2")
+        r2 = self.client.get("/api/mizan/ctx/multi/?b=2&a=1")
+
+        self.assertEqual(r1.content, r2.content)
+
+    # ── Special characters in param values ──────────────────────────────────
+
+    def test_special_chars_in_param_values(self):
+        """Param values with semicolons, spaces, quotes are URL-encoded in header."""
+        from urllib.parse import unquote
+        from mizan.client.executor import _format_invalidate_header
+
+        # Semicolon in value — would break the parser without encoding
+        result = _format_invalidate_header([
+            {"context": "search", "params": {"q": "a;b"}},
+        ])
+        self.assertNotIn(";b", result.split("search;")[1].split(",")[0].replace("q=", ""))
+        # Parse it back
+        parts = result.split(";")
+        self.assertEqual(parts[0], "search")
+        k, v = parts[1].split("=", 1)
+        self.assertEqual(k, "q")
+        self.assertEqual(unquote(v), "a;b")
+
+    def test_spaces_in_param_values(self):
+        """Spaces in param values are URL-encoded."""
+        from urllib.parse import unquote
+        from mizan.client.executor import _format_invalidate_header
+
+        result = _format_invalidate_header([
+            {"context": "search", "params": {"q": "hello world"}},
+        ])
+        self.assertNotIn(" ", result)
+        self.assertIn("hello%20world", result)
+
+    def test_special_chars_round_trip(self):
+        """URL-encoded param values survive a parse round-trip."""
+        from urllib.parse import unquote
+        from mizan.client.executor import _format_invalidate_header
+
+        original = [
+            {"context": "data", "params": {"name": "O'Brien", "tag": "a;b;c"}},
+        ]
+        header = _format_invalidate_header(original)
+
+        # Parse (what Edge would do)
+        segments = header.split(";")
+        ctx = segments[0]
+        params = {}
+        for seg in segments[1:]:
+            k, v = seg.split("=", 1)
+            params[unquote(k)] = unquote(v)
+
+        self.assertEqual(ctx, "data")
+        self.assertEqual(params["name"], "O'Brien")
+        self.assertEqual(params["tag"], "a;b;c")
+
+    # ── Concurrent mutations ────────────────────────────────────────────────
+
+    def test_concurrent_mutations_independent_headers(self):
+        """Two mutations both produce correct, independent invalidation headers."""
+        r1 = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "update_profile", "args": {"user_id": 5, "name": "A"}}),
+            content_type="application/json",
+        )
+        r2 = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "update_profile", "args": {"user_id": 8, "name": "B"}}),
+            content_type="application/json",
+        )
+
+        self.assertEqual(r1.status_code, 200)
+        self.assertEqual(r2.status_code, 200)
+
+        # Each carries its own scoped invalidation
+        self.assertIn("user_id=5", r1["X-Mizan-Invalidate"])
+        self.assertIn("user_id=8", r2["X-Mizan-Invalidate"])
+
+        d1 = r1.json()
+        d2 = r2.json()
+        self.assertEqual(d1["invalidate"][0]["params"]["user_id"], 5)
+        self.assertEqual(d2["invalidate"][0]["params"]["user_id"], 8)
+
+    # ── Large invalidation sets ─────────────────────────────────────────────
+
+    def test_large_invalidation_set(self):
+        """Many contexts in affects= produce parseable header and body."""
+        contexts = [ReactContext(f"ctx_{i}") for i in range(20)]
+
+        @client(affects=contexts)
+        def big_mutation(request: HttpRequest) -> ValidOutput:
+            return ValidOutput(valid=True)
+
+        register(big_mutation, "big_mutation")
+
+        response = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "big_mutation", "args": {}}),
+            content_type="application/json",
+        )
+
+        data = response.json()
+        header = response["X-Mizan-Invalidate"]
+
+        # All 20 contexts in body
+        self.assertEqual(len(data["invalidate"]), 20)
+
+        # All 20 in header, comma-separated, parseable
+        header_parts = [p.strip() for p in header.split(",")]
+        self.assertEqual(len(header_parts), 20)
+        self.assertEqual(header_parts[0], "ctx_0")
+        self.assertEqual(header_parts[19], "ctx_19")
+
+    # ── Auth-dependent Vary ─────────────────────────────────────────────────
+
+    def test_vary_actually_differentiates_by_auth(self):
+        """Same URL with different auth produces different response bodies."""
+        from tests.models import EmailUser
+
+        AuthCtx = ReactContext("authtest")
+
+        class AuthOut(BaseModel):
+            viewer: str
+            can_edit: bool
+
+        @client(context=AuthCtx, auth=True)
+        def profile_view(request: HttpRequest, user_id: int) -> AuthOut:
+            is_owner = hasattr(request.user, "pk") and request.user.pk == user_id
+            return AuthOut(
+                viewer=getattr(request.user, "email", "anon"),
+                can_edit=is_owner,
+            )
+
+        register(profile_view, "profile_view")
+
+        # Create two users
+        user_a = EmailUser.objects.create_user(email="a@test.com", password="pass")
+        user_b = EmailUser.objects.create_user(email="b@test.com", password="pass")
+
+        # User A views their own profile
+        self.client.login(email="a@test.com", password="pass")
+        r_a = self.client.get(f"/api/mizan/ctx/authtest/?user_id={user_a.pk}")
+        self.client.logout()
+
+        # User B views User A's profile
+        self.client.login(email="b@test.com", password="pass")
+        r_b = self.client.get(f"/api/mizan/ctx/authtest/?user_id={user_a.pk}")
+        self.client.logout()
+
+        # Same URL, different auth → different response bodies
+        self.assertEqual(r_a.status_code, 200)
+        self.assertEqual(r_b.status_code, 200)
+        self.assertNotEqual(r_a.content, r_b.content)
+
+        d_a = r_a.json()
+        d_b = r_b.json()
+        self.assertTrue(d_a["profile_view"]["can_edit"])   # owner
+        self.assertFalse(d_b["profile_view"]["can_edit"])  # not owner
+
+    # ── Empty invalidation ──────────────────────────────────────────────────
+
+    def test_no_affects_no_header_no_body_key(self):
+        """Function without affects= has no invalidation in response at all."""
+        @client
+        def plain(request: HttpRequest) -> ValidOutput:
+            return ValidOutput(valid=True)
+
+        register(plain, "plain")
+
+        response = self.client.post(
+            "/api/mizan/call/",
+            data=json.dumps({"fn": "plain", "args": {}}),
+            content_type="application/json",
+        )
+
+        data = response.json()
+        self.assertNotIn("invalidate", data)
+        self.assertNotIn("X-Mizan-Invalidate", response)