@@ -7,6 +7,7 @@
 """
 from __future__ import annotations
 
+import asyncio
 import sys
 import time
 import types
@@ -192,5 +193,118 @@ def test_layout_options_bypass_cache(self):
         self.assertTrue(main.should_cache_get_url("https://example.dev/examples/values"))
 
 
+def post_request(body: bytes, *, env=None, headers: dict[str, str] | None = None) -> Request:
+    raw_headers = [(b"content-type", b"application/x-www-form-urlencoded")]
+    for key, value in (headers or {}).items():
+        raw_headers.append((key.lower().encode(), value.encode()))
+
+    async def receive():
+        return {"type": "http.request", "body": body, "more_body": False}
+
+    return Request(
+        {
+            "type": "http",
+            "method": "POST",
+            "path": "/examples/values",
+            "query_string": b"",
+            "headers": raw_headers,
+            "env": env,
+        },
+        receive,
+    )
+
+
+class RunExampleFlowTests(unittest.TestCase):
+    """Behavioral tests of the POST /examples/{slug} handler branches:
+    verification-required, missing-site-key, too-large, and
+    verify-then-set-clearance. Exercises the real run_example handler with
+    a constructed Request; the Dynamic Worker run and Turnstile siteverify
+    HTTP call are stubbed so only the handler's control flow is under test.
+    """
+
+    def _run(self, request):
+        return asyncio.run(main.run_example("values", request))
+
+    def setUp(self):
+        self._saved = {name: getattr(main, name) for name in ("_run_example", "_verify_turnstile")}
+
+    def tearDown(self):
+        for name, value in self._saved.items():
+            setattr(main, name, value)
+
+    def _stub_run(self, output="OUT"):
+        async def _fake_run_example(request, slug, code):
+            self._ran_code = code
+            return output
+        main._run_example = _fake_run_example
+
+    def test_verification_required_path_does_not_run_code(self):
+        self._ran_code = None
+        self._stub_run()
+        env = session_env(TURNSTILE_SITE_KEY="site-key")
+        response = self._run(post_request(b"code=print(1)", env=env))
+        body = response.body.decode()
+        # The retriable-challenge marker is rendered as a real attribute
+        # (the bare string also lives in the page's static JS selector).
+        self.assertIn('data-turnstile-required="true"', body)
+        self.assertIn("Verification required", body)
+        self.assertIsNone(self._ran_code, "code must not run before verification")
+
+    def test_missing_site_key_message(self):
+        self._stub_run()
+        env = session_env()  # secret set, but no TURNSTILE_SITE_KEY
+        response = self._run(post_request(b"code=print(1)", env=env))
+        body = response.body.decode()
+        self.assertIn("TURNSTILE_SITE_KEY is not configured", body)
+        self.assertNotIn('data-turnstile-required="true"', body)
+
+    def test_oversize_body_rejected_with_413(self):
+        self._stub_run()
+        big = b"code=" + b"x" * (main.MAX_SUBMITTED_BODY_BYTES + 1)
+        response = self._run(post_request(big, env=session_env(TURNSTILE_CHALLENGE_MODE="off")))
+        self.assertEqual(response.status_code, 413)
+        self.assertIn("too large", response.body.decode().lower())
+
+    def test_off_mode_runs_without_challenge(self):
+        self._ran_code = None
+        self._stub_run(output="ran-output")
+        response = self._run(
+            post_request(b"code=print(42)", env=session_env(TURNSTILE_CHALLENGE_MODE="off"))
+        )
+        self.assertEqual(self._ran_code, "print(42)")
+        self.assertIn("ran-output", response.body.decode())
+
+    def test_verified_token_sets_clearance_cookie(self):
+        self._ran_code = None
+        self._stub_run(output="verified-run")
+
+        async def _fake_verify(request, token):
+            return True, "ok", "success"
+
+        main._verify_turnstile = _fake_verify
+        env = session_env(TURNSTILE_SITE_KEY="site-key")
+        response = self._run(post_request(b"code=print(1)&cf-turnstile-response=tok", env=env))
+        self.assertEqual(self._ran_code, "print(1)")
+        set_cookie = response.headers.get("set-cookie", "")
+        self.assertIn(f"{main.TURNSTILE_CLEARANCE_COOKIE}=", set_cookie)
+
+
+class DynamicWorkerCodeTests(unittest.TestCase):
+    """Behavioral coverage of build_dynamic_worker_code: the generated
+    module embeds the submitted code as a string literal (so a
+    closing-quote injection cannot break out) and defines the entrypoint.
+    """
+
+    def test_generated_module_embeds_code_and_entrypoint(self):
+        from app import build_dynamic_worker_code
+
+        module = build_dynamic_worker_code("print('hi <b>')")
+        self.assertIn(repr("print('hi <b>')"), module)
+        self.assertIn("class Default(WorkerEntrypoint)", module)
+        self.assertIn("async def fetch", module)
+        tricky = build_dynamic_worker_code("x = '''end'''")
+        self.assertIn(repr("x = '''end'''"), tricky)
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -1,3 +1,4 @@
+import asyncio
 import importlib
 import pathlib
 import sys
@@ -23,6 +24,7 @@ class Request:
 
         workers.Request = Request
         workers.Context = object
+        workers.wait_until = lambda task: None
         sys.modules["workers"] = workers
         self.Request = Request
 
@@ -38,8 +40,48 @@ def new(cls, value):
                     search=f"?{parsed.query}" if parsed.query else "",
                 )
 
+        # Minimal JS-runtime fakes used by process_request's non-streaming path.
+        class Response:
+            def __init__(self, body, headers, status):
+                self.body = body
+                self.headers = headers
+                self.status = status
+
+            @classmethod
+            def new(cls, body, *, headers=None, status=200):
+                return cls(body, headers, status)
+
+        class Object:
+            @staticmethod
+            def fromEntries(pairs):
+                return dict(pairs)
+
         js.URL = URL
+        js.Response = Response
+        js.Object = Object
+        js.TransformStream = object  # only referenced on the streaming path
         sys.modules["js"] = js
+
+        ffi = types.ModuleType("pyodide.ffi")
+
+        class _Proxy:
+            def __init__(self, obj):
+                self._obj = obj
+
+            def getBuffer(self):
+                return SimpleNamespace(data=self._obj)
+
+            def destroy(self):
+                pass
+
+        ffi.create_proxy = _Proxy
+        self.saved_pyodide = sys.modules.get("pyodide")
+        self.saved_pyodide_ffi = sys.modules.get("pyodide.ffi")
+        pyodide = sys.modules.get("pyodide") or types.ModuleType("pyodide")
+        pyodide.ffi = ffi
+        sys.modules["pyodide"] = pyodide
+        sys.modules["pyodide.ffi"] = ffi
+
         self.bridge = importlib.import_module("src.worker_asgi_bridge")
 
     def tearDown(self):
@@ -54,6 +96,85 @@ def tearDown(self):
             sys.modules.pop("js", None)
         else:
             sys.modules["js"] = self.saved_js
+        for name, saved in (("pyodide", self.saved_pyodide), ("pyodide.ffi", self.saved_pyodide_ffi)):
+            if saved is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = saved
+
+    def _fake_request(self, *, method="POST", url, headers, body_chunks):
+        async def body_iter():
+            for chunk in body_chunks:
+                yield SimpleNamespace(to_bytes=lambda c=chunk: c)
+
+        class FakeRequest(self.Request):
+            pass
+
+        req = FakeRequest()
+        req.method = method
+        req.url = url
+        req.headers = headers
+        req.body = body_iter() if body_chunks else None
+        return req
+
+    def test_process_request_non_streaming_echoes_body_and_status(self):
+        async def app(scope, receive, send):
+            body = b""
+            while True:
+                message = await receive()
+                body += message.get("body", b"")
+                if not message.get("more_body"):
+                    break
+            await send(
+                {
+                    "type": "http.response.start",
+                    "status": 201,
+                    "headers": [(b"content-type", b"text/plain"), (b"x-trace", b"abc")],
+                }
+            )
+            await send({"type": "http.response.body", "body": b"echo:" + body, "more_body": False})
+
+        req = self._fake_request(
+            url="https://x.dev/examples/values",
+            headers={"content-type": "text/plain"},
+            body_chunks=[b"pi", b"ng"],
+        )
+        response = asyncio.run(
+            self.bridge.process_request(app, req, SimpleNamespace(), None, state={})
+        )
+        self.assertEqual(response.status, 201)
+        self.assertEqual(response.body, b"echo:ping")
+        self.assertEqual(response.headers.get("content-type"), "text/plain")
+        self.assertEqual(response.headers.get("x-trace"), "abc")
+
+    def test_process_request_propagates_app_exception(self):
+        async def app(scope, receive, send):
+            raise RuntimeError("boom in app")
+
+        req = self._fake_request(
+            url="https://x.dev/examples/values",
+            headers={"content-type": "text/plain"},
+            body_chunks=[],
+        )
+        with self.assertRaises(RuntimeError):
+            asyncio.run(self.bridge.process_request(app, req, SimpleNamespace(), None, state={}))
+
+    def test_process_request_rejects_oversize_body_before_running_app(self):
+        ran = {"app": False}
+
+        async def app(scope, receive, send):  # pragma: no cover - must not run
+            ran["app"] = True
+
+        req = self._fake_request(
+            url="https://x.dev/examples/values",
+            headers={"content-type": "text/plain"},
+            body_chunks=[b"toolong"],
+        )
+        response = asyncio.run(
+            self.bridge.process_request(app, req, SimpleNamespace(), None, state={}, max_body_bytes=3)
+        )
+        self.assertEqual(response.status, 413)
+        self.assertFalse(ran["app"], "oversize body must be rejected before the app runs")
 
     def test_request_to_scope_accepts_state_without_request_globals_or_extra_scope(self):
         class FakeRequest(self.Request):