diff --git a/backend/db/connection.py b/backend/db/connection.py
index 089dfba5..f62db3b3 100644
--- a/backend/db/connection.py
+++ b/backend/db/connection.py
@@ -34,7 +34,12 @@ def select(
         filters: Optional[dict] = None,
         order: Optional[str] = None,
         limit: Optional[int] = None,
+        offset: Optional[int] = None,
     ) -> list:
+        """Read rows. Pass `limit`/`offset` to page — PostgREST caps a
+        response at `max_rows` (1000) and answers 206 Partial Content,
+        which is a 2xx, so an unpaged read over that many rows truncates
+        silently."""
         params: dict = {"select": columns}
         if filters:
             params.update(filters)
@@ -42,6 +47,8 @@ def select(
             params["order"] = order
         if limit:
             params["limit"] = str(limit)
+        if offset is not None:
+            params["offset"] = str(offset)
         r = _client.get(self.url, params=params)
         r.raise_for_status()
         return r.json()
diff --git a/backend/main.py b/backend/main.py
index 0d8c9eff..51328b28 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -201,10 +201,16 @@ async def http_exception_handler(request: Request, exc: StarletteHTTPException):
         code=getattr(exc, "code", None),
         machine_detail=getattr(exc, "machine_detail", None),
     )
+    # Preserve headers the raise site set (e.g. Retry-After on a 429) —
+    # dropping them would strip the only machine-readable part of a
+    # throttling response.
+    headers = dict(getattr(exc, "headers", None) or {})
+    if rid:
+        headers["X-Request-ID"] = rid
     return JSONResponse(
         status_code=exc.status_code,
         content=content,
-        headers={"X-Request-ID": rid} if rid else {},
+        headers=headers,
     )
 
 
diff --git a/backend/routes/extract.py b/backend/routes/extract.py
index 8f8bd9e2..f23f2545 100644
--- a/backend/routes/extract.py
+++ b/backend/routes/extract.py
@@ -30,12 +30,13 @@ def _enforce_ocr_limits(request: Request) -> str:
     user_id = get_session_user_id(request)
     retry = check_rate_limit(f"ocr:{user_id}", limit=_OCR_RATE_LIMIT, window_sec=_OCR_RATE_WINDOW)
     if retry is not None:
-        # NB: the app's global HTTPException handler (main.py) doesn't forward
-        # exc.headers, so the retry budget is conveyed in the detail string
-        # rather than a Retry-After header.
+        # main.py's HTTPException handler forwards exc.headers as of #544, so
+        # the budget rides in a real Retry-After. It stays in the detail
+        # string too — clients that only surface the message keep working.
         raise HTTPException(
             status_code=429,
             detail=f"Too many OCR requests. Retry in {retry}s.",
+            headers={"Retry-After": str(retry)},
         )
     return user_id
 
diff --git a/backend/routes/gradescope.py b/backend/routes/gradescope.py
index 6e29333a..e61af339 100644
--- a/backend/routes/gradescope.py
+++ b/backend/routes/gradescope.py
@@ -202,11 +202,13 @@ def _enforce_gs_rate_limit(user_id: str, action: str, *, limit: int, window_sec:
         f"gradescope:{action}:{user_id}", limit=limit, window_sec=window_sec
     )
     if retry is not None:
-        # main.py's HTTPException handler drops exc.headers, so the retry budget
-        # rides in the detail string rather than a Retry-After header.
+        # main.py's HTTPException handler forwards exc.headers as of #544, so
+        # the budget rides in a real Retry-After. It stays in the detail
+        # string too — clients that only surface the message keep working.
         raise HTTPException(
             status_code=429,
             detail=f"Too many Gradescope {action} requests. Retry in {retry}s.",
+            headers={"Retry-After": str(retry)},
         )
 
 
diff --git a/backend/routes/quiz.py b/backend/routes/quiz.py
index 28b95909..c137aa3f 100644
--- a/backend/routes/quiz.py
+++ b/backend/routes/quiz.py
@@ -25,12 +25,17 @@
 from services.quiz_config import (
     CONCRETE_DIFFICULTIES,
     QUIZ_ATTEMPT_ABANDON_TTL_HOURS,
+    QUIZ_DAILY_SPEND_CAP_USD,
+    QUIZ_GENERATE_RATE_LIMIT,
+    QUIZ_GENERATE_RATE_WINDOW_SEC,
+    QUIZ_GENERATION_TIMEOUT_SEC,
     QUIZ_TOPUP_DROP_RATIO,
     QUIZ_TOPUP_MAX_RETRIES,
     REQUESTED_DIFFICULTIES,
     mastery_after,
     quiz_config_payload,
 )
+from services.request_limits import check_rate_limit, refund_rate_limit
 from services.quiz_errors import QuizAPIError, QuizErrorCode
 from services.profiles import get_display_name
 from services.encryption import encrypt_json, decrypt_json_column
@@ -87,6 +92,80 @@ def _load_prompt(name: str) -> str:
 _MAX_HISTORY_OFFSET = 1_000_000
 
 
+# supabase/config.toml sets PostgREST's max_rows = 1000, and an over-cap
+# response is 206 Partial Content — a 2xx, so raise_for_status never fires
+# and the truncation is silent. Same constant and same reasoning as
+# achievement_service._daily_totals; page to completion or the sum is a lie.
+_USAGE_PAGE = 1000
+
+
+def _daily_spend_exceeded(user_id: str) -> bool:
+    """True if this user is past the daily LLM spend ceiling (#544 F1).
+
+    Reads the llm_usage ledger agents/usage.py already writes, PAGED: an
+    unpaged read stops at max_rows, so a heavy user's sum plateaus below
+    the cap and the guard never trips for exactly the runaway it targets.
+    Stops early once the ceiling is crossed — the common case is a couple
+    of rows, and a user past the cap doesn't need an exact total.
+
+    Fails OPEN on any error: this is a cost control, not a correctness
+    gate, and denying every student because a usage read blipped is worse
+    than the spend it would save.
+    """
+    try:
+        since = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
+        spent = 0.0
+        offset = 0
+        while True:
+            rows = table("llm_usage").select(
+                "cost_usd",
+                filters={"user_id": f"eq.{user_id}", "created_at": f"gte.{since}"},
+                limit=_USAGE_PAGE,
+                offset=offset,
+            ) or []
+            spent += sum(float(r.get("cost_usd") or 0.0) for r in rows)
+            if spent >= QUIZ_DAILY_SPEND_CAP_USD:
+                return True
+            if len(rows) < _USAGE_PAGE:
+                return False
+            offset += _USAGE_PAGE
+    except Exception:
+        logger.exception("quiz: daily spend check failed user=%s; allowing", user_id)
+        return False
+
+
+def _refund_generate_slot(user_id: str) -> None:
+    """Hand back the rate-limit slot a failed generation consumed (#544 F1).
+
+    The slot is claimed BEFORE the model runs (so a burst can't get past
+    the gate concurrently), which means a backend failure would otherwise
+    spend the student's quota: eight 502s in two minutes would lock them
+    out for five with a message saying they'd generated too many quizzes,
+    having received none. A failure the student didn't cause shouldn't
+    cost them anything, and the 502 explicitly invites a retry.
+    """
+    try:
+        refund_rate_limit(f"quiz_generate:{user_id}")
+    except Exception:
+        logger.exception("quiz: rate-limit refund failed user=%s", user_id)
+
+
+def _log_generation_failed(body, request_id: str | None, reason: str) -> None:
+    """#544 F3: make a 502 the student saw a 502 an admin can count."""
+    events_service.log_event(
+        "quiz.generation_failed",
+        category="error",
+        user_id=body.user_id,
+        request_id=request_id,
+        payload={
+            "concept_node_id": body.concept_node_id,
+            "difficulty": body.difficulty,
+            "num_questions": body.num_questions,
+            "reason": reason,
+        },
+    )
+
+
 def _abandon_cutoff() -> datetime:
     return datetime.now(timezone.utc) - timedelta(
         hours=QUIZ_ATTEMPT_ABANDON_TTL_HOURS
@@ -510,8 +589,17 @@ async def _quiz_via_agent(
         run_kwargs["model"] = model_override
 
     async def _run(message: str, limits) -> Quiz:
+        # #544 F2: bound EACH agent run rather than the whole function.
+        # Wrapping the outer coroutine cancelled it mid-flight, and
+        # CancelledError is a BaseException — it flew straight past the
+        # top-up's serve-what-we-have handler and threw away questions the
+        # student had already paid for. Timing out one run raises an
+        # ordinary TimeoutError the existing handlers can reason about.
         result = record_agent_usage(
-            await quiz_agent.run(message, usage_limits=limits, **run_kwargs),
+            await asyncio.wait_for(
+                quiz_agent.run(message, usage_limits=limits, **run_kwargs),
+                timeout=QUIZ_GENERATION_TIMEOUT_SEC,
+            ),
             feature="quiz", task="quiz", user_id=deps.user_id,
         )
         return result.output
@@ -582,7 +670,7 @@ def _absorb(quiz: Quiz) -> None:
                 )
             try:
                 _absorb(await _run(topup_msg, TOPUP_LIMITS))
-            except Exception as e:
+            except (Exception, asyncio.TimeoutError) as e:
                 # The request deliberately SUCCEEDS from here — serve the
                 # short quiz with an honest count. No traceback: the E2E
                 # logscan oracle reports those as findings, and this path
@@ -652,7 +740,44 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
         or str(uuid.uuid4())
     )
 
+    # #544 F1: cost guards run AFTER ownership (a stranger's node 404s
+    # first, so probing can't consume a victim's quota) and BEFORE the
+    # model call. Neither rejection is a backend failure, so neither emits
+    # quiz.generation_failed.
+    retry_after = check_rate_limit(
+        f"quiz_generate:{body.user_id}",
+        limit=QUIZ_GENERATE_RATE_LIMIT,
+        window_sec=QUIZ_GENERATE_RATE_WINDOW_SEC,
+    )
+    if retry_after is not None:
+        raise QuizAPIError(
+            status_code=429,
+            code=QuizErrorCode.QUIZ_RATE_LIMITED,
+            message=(
+                "You've generated a lot of quizzes just now — "
+                "take a moment and try again shortly."
+            ),
+            headers={"Retry-After": str(retry_after)},
+        )
+    if _daily_spend_exceeded(body.user_id):
+        logger.warning(
+            "quiz: daily spend cap reached user=%s request_id=%s",
+            body.user_id, request_id,
+        )
+        raise QuizAPIError(
+            status_code=429,
+            code=QuizErrorCode.QUIZ_DAILY_LIMIT_REACHED,
+            message=(
+                "You've reached today's limit for AI-generated study "
+                "material. It resets tomorrow."
+            ),
+        )
+
     try:
+        # Each agent run inside is individually bounded by
+        # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
+        # coroutine here would discard a partial quiz the top-up handler
+        # is designed to serve.
         questions = await _quiz_via_agent(
             user_id=body.user_id,
             course_id=course_id,
@@ -667,11 +792,31 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
     except HTTPException:
         # The 404 for an unknown concept node is raised before the agent call;
         # never swallow a known HTTP state.
+        _refund_generate_slot(body.user_id)
         raise
+    except asyncio.TimeoutError as e:
+        # #544 F2: distinct from a generic failure — the client can say
+        # "that took too long" and offering a retry obviously makes sense.
+        # NB: only asyncio.TimeoutError. The builtin TimeoutError is in the
+        # OSError family, so catching it too would relabel a transport
+        # socket timeout as a wall-clock generation timeout.
+        logger.warning(
+            "quiz: generation timed out after %ss request_id=%s",
+            QUIZ_GENERATION_TIMEOUT_SEC, request_id,
+        )
+        _refund_generate_slot(body.user_id)
+        _log_generation_failed(body, request_id, "timeout")
+        raise QuizAPIError(
+            status_code=502,
+            code=QuizErrorCode.QUIZ_GENERATION_TIMEOUT,
+            message="Quiz generation took too long. Please try again.",
+        ) from e
     except (UsageLimitExceeded, UnexpectedModelBehavior) as e:
         # The raw-Gemini legacy fallback was retired in #145; degrade to 502
         # rather than serving a quiz from a second LLM path.
         logger.warning("Quiz agent guardrails tripped; returning 502", exc_info=e)
+        _refund_generate_slot(body.user_id)
+        _log_generation_failed(body, request_id, "agent_guardrail")
         raise QuizAPIError(
             status_code=502,
             code=QuizErrorCode.QUIZ_GENERATION_FAILED,
@@ -679,6 +824,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
         ) from e
     except Exception as e:
         logger.exception("Unexpected quiz-agent failure; returning 502")
+        _refund_generate_slot(body.user_id)
+        _log_generation_failed(body, request_id, "agent_error")
         raise QuizAPIError(
             status_code=502,
             code=QuizErrorCode.QUIZ_GENERATION_FAILED,
diff --git a/backend/services/events_service.py b/backend/services/events_service.py
index fce2dff4..780c528b 100644
--- a/backend/services/events_service.py
+++ b/backend/services/events_service.py
@@ -86,6 +86,10 @@
     # surfaces in admin analytics — this failure was invisible for months
     # precisely because nothing emitted when the background task died.
     "quiz.context_write_failed",
+    # #544/F3: generation failed (agent error, timeout, or every question
+    # dropped). Same reasoning: a 502 the student sees should be a 502 an
+    # admin can count.
+    "quiz.generation_failed",
     "chat.message_sent",
     "note.created",
     "session.started",
diff --git a/backend/services/quiz_config.py b/backend/services/quiz_config.py
index d1b5d98f..fa9983a3 100644
--- a/backend/services/quiz_config.py
+++ b/backend/services/quiz_config.py
@@ -79,6 +79,34 @@ def mastery_after(before: float, *, score: int, total: int) -> float:
     return max(0.0, min(1.0, raw))
 
 
+# ── Cost + abuse guards (#544 F1/F2) ────────────────────────────────────────
+#
+# Generation is an unbounded LLM call behind a button: before #544 nothing
+# stopped a held-down key or a scripted loop from spending real money.
+#
+# The rate limit is sized for a human: a student comparing difficulties or
+# retaking a concept might legitimately generate a handful of quizzes in a
+# few minutes; nobody legitimately generates 10 in one.
+QUIZ_GENERATE_RATE_LIMIT = 8
+QUIZ_GENERATE_RATE_WINDOW_SEC = 300   # 5 minutes
+
+# Daily per-user LLM spend ceiling. The SPEND it measures is cross-feature
+# (llm_usage records every agent call, not just quiz ones), but the ceiling
+# is only ENFORCED on quiz generation — the one unbounded LLM call behind a
+# button. Other entry points stay unguarded for now; moving this into a
+# shared guard is its own piece of work, not something to imply here.
+# A generation on the default flash-lite tier costs well under a cent, so
+# this is ~2 orders of magnitude above any real study day — it exists to
+# bound a runaway, not to ration normal use. Deliberately fail-OPEN: if the
+# usage read errors we serve the quiz rather than denying every student.
+QUIZ_DAILY_SPEND_CAP_USD = 2.00
+
+# Wall-clock ceiling on one generation (agent run incl. its tool calls).
+# Past this the student is staring at a spinner and would rather be told to
+# try again; the request also stops holding a worker slot.
+QUIZ_GENERATION_TIMEOUT_SEC = 90
+
+
 # ── Generation honesty (#543 E2) ────────────────────────────────────────────
 #
 # Questions whose correct_answer doesn't match an option verbatim are
diff --git a/backend/services/quiz_errors.py b/backend/services/quiz_errors.py
index d7591faf..7dd813a3 100644
--- a/backend/services/quiz_errors.py
+++ b/backend/services/quiz_errors.py
@@ -46,6 +46,14 @@ class QuizErrorCode(str, Enum):
     QUIZ_QUESTION_INVALID = "QUIZ_QUESTION_INVALID"
     QUIZ_NOT_AUTHORIZED = "QUIZ_NOT_AUTHORIZED"
     QUIZ_GENERATION_FAILED = "QUIZ_GENERATION_FAILED"
+    # #544 F2: the generation exceeded its wall-clock budget. Distinct from
+    # the generic failure so the client can say "that took too long" and a
+    # retry is obviously worth offering.
+    QUIZ_GENERATION_TIMEOUT = "QUIZ_GENERATION_TIMEOUT"
+    # #544 F1: too many generations in the rate window.
+    QUIZ_RATE_LIMITED = "QUIZ_RATE_LIMITED"
+    # #544 F1: this account's daily LLM spend ceiling is reached.
+    QUIZ_DAILY_LIMIT_REACHED = "QUIZ_DAILY_LIMIT_REACHED"
     QUIZ_INTERNAL_ERROR = "QUIZ_INTERNAL_ERROR"
     # Uncoded HTTP errors that aren't one of the semantic states above —
     # router 404s/405s on version-skewed clients, library-raised
@@ -80,8 +88,9 @@ def __init__(
         code: QuizErrorCode,
         message: str,
         machine_detail=None,
+        headers: dict[str, str] | None = None,
     ):
-        super().__init__(status_code=status_code, detail=message)
+        super().__init__(status_code=status_code, detail=message, headers=headers)
         self.code = code
         self.machine_detail = machine_detail
 
diff --git a/backend/services/request_limits.py b/backend/services/request_limits.py
index ac4ebd5c..03b36ddf 100644
--- a/backend/services/request_limits.py
+++ b/backend/services/request_limits.py
@@ -37,6 +37,20 @@ def check_rate_limit(key: str, *, limit: int, window_sec: int) -> int | None:
     return None
 
 
+def refund_rate_limit(key: str) -> None:
+    """Give back the most recent slot recorded for `key`.
+
+    For guards that must claim BEFORE doing the expensive work (so a
+    concurrent burst can't slip past the gate) but shouldn't charge the
+    caller when that work fails for reasons they didn't cause. Dropping
+    the newest timestamp — not the oldest — keeps the window's start
+    anchored to the caller's earliest real attempt.
+    """
+    bucket = _rate_state.get(key)
+    if bucket:
+        bucket.pop()
+
+
 async def read_within_limit(upload: UploadFile, max_bytes: int) -> bytes:
     """Read at most ``max_bytes`` (+1 to detect overflow) from an UploadFile so
     an oversize upload can't be pulled fully into memory before we reject it.
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 1248bbd6..6b7a50f7 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -47,6 +47,19 @@ def pytest_configure(config):
     )
 
 
+@pytest.fixture(autouse=True)
+def _clear_rate_limit_state():
+    """#544: services/request_limits keeps its sliding windows in a
+    process-global dict, so one test's burst of requests would throttle
+    every later test that hits the same route as the same user. Same
+    reasoning as the lru_cache reset below."""
+    from services import request_limits
+
+    request_limits._rate_state.clear()
+    yield
+    request_limits._rate_state.clear()
+
+
 @pytest.fixture(autouse=True)
 def _clear_lru_caches():
     """#98: reset the per-process lru_caches around every test so one test's
diff --git a/backend/tests/test_event_capture_seams.py b/backend/tests/test_event_capture_seams.py
index 9edbd03c..faf0f8a0 100644
--- a/backend/tests/test_event_capture_seams.py
+++ b/backend/tests/test_event_capture_seams.py
@@ -93,6 +93,7 @@ def test_event_taxonomy_is_pinned():
         "quiz.started",
         "quiz.completed",
         "quiz.context_write_failed",  # #529/B3
+        "quiz.generation_failed",     # #544/F3
         "chat.message_sent",
         "note.created",
         "session.started",
diff --git a/backend/tests/test_quiz_cost_observability_f.py b/backend/tests/test_quiz_cost_observability_f.py
new file mode 100644
index 00000000..8f508c26
--- /dev/null
+++ b/backend/tests/test_quiz_cost_observability_f.py
@@ -0,0 +1,437 @@
+"""
+Workstream F of the pre-revamp quiz repair batch (#544, epic #537):
+cost, abuse, observability.
+
+- F1: per-user rate limit on generate + a daily LLM spend guard.
+  Generation is an unbounded LLM call behind a button; nothing stopped a
+  loop.
+- F2: explicit timeout on the agent call, mapped to a 502 taxonomy
+  rather than one generic code.
+- F3: quiz.generation_failed events so backend failures reach admin
+  analytics (quiz.context_write_failed landed with #529/B3).
+- F4: ownership + active-semester scoping on ?concept= deep links.
+"""
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from main import app
+from agents.quiz import Quiz, QuizQuestion
+
+client = TestClient(app)
+
+
+def _quiz():
+    return Quiz(questions=[
+        QuizQuestion(
+            question="Q?", type="multiple_choice", difficulty="easy",
+            options=["a", "b", "c", "d"], correct_answer="a",
+            explanation="x", concept="Loops",
+        ),
+    ])
+
+
+def _factory(*, spend_rows=None):
+    def factory(name):
+        mock = MagicMock()
+        if name == "graph_nodes":
+            mock.select.return_value = [{
+                "id": "node1", "course_id": "course1",
+                "concept_name": "Loops", "mastery_score": 0.5,
+            }]
+        elif name == "quiz_attempts":
+            mock.insert.return_value = [{"id": "quiz-generated"}]
+        elif name == "llm_usage":
+            mock.select.return_value = spend_rows if spend_rows is not None else []
+        else:
+            mock.select.return_value = []
+            mock.insert.return_value = []
+        return mock
+
+    return factory
+
+
+def _generate(user_id="user_andres"):
+    return client.post("/api/quiz/generate", json={
+        "user_id": user_id,
+        "concept_node_id": "node1",
+        "num_questions": 1,
+        "difficulty": "easy",
+        "use_shared_context": False,
+    })
+
+
+# ── F1: rate limit + spend guard ────────────────────────────────────────────
+
+
+class TestGenerateRateLimit:
+    def test_burst_past_the_limit_is_rejected(self):
+        from services.quiz_config import (
+            QUIZ_GENERATE_RATE_LIMIT,
+            QUIZ_GENERATE_RATE_WINDOW_SEC,
+        )
+
+        assert QUIZ_GENERATE_RATE_WINDOW_SEC > 0
+        run = AsyncMock(return_value=SimpleNamespace(output=_quiz()))
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run", new=run),
+        ):
+            allowed = [_generate() for _ in range(QUIZ_GENERATE_RATE_LIMIT)]
+            blocked = _generate()
+
+        assert all(r.status_code == 200 for r in allowed)
+        assert blocked.status_code == 429
+        body = blocked.json()
+        assert body["error"]["code"] == "QUIZ_RATE_LIMITED"
+        assert "Retry-After" in blocked.headers
+        # The blocked request must not have reached the model.
+        assert run.call_count == QUIZ_GENERATE_RATE_LIMIT
+
+    def test_a_failed_generation_does_not_burn_the_budget(self):
+        """The slot is claimed before the model runs, so a backend failure
+        would otherwise spend the student's quota — eight 502s in two
+        minutes locking them out for five, with a message telling them
+        they generated too many quizzes. They received none."""
+        from services.quiz_config import QUIZ_GENERATE_RATE_LIMIT
+
+        failing = AsyncMock(side_effect=RuntimeError("boom"))
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run", new=failing),
+        ):
+            for _ in range(QUIZ_GENERATE_RATE_LIMIT + 2):
+                r = _generate()
+                assert r.status_code == 502, (
+                    "a failed generation must not consume the rate-limit slot"
+                )
+
+        # …and the budget is still intact for a request that can succeed.
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run",
+                  new=AsyncMock(return_value=SimpleNamespace(output=_quiz()))),
+        ):
+            assert _generate().status_code == 200
+
+    def test_limit_is_per_user(self):
+        run = AsyncMock(return_value=SimpleNamespace(output=_quiz()))
+        from services.quiz_config import QUIZ_GENERATE_RATE_LIMIT
+
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run", new=run),
+        ):
+            for _ in range(QUIZ_GENERATE_RATE_LIMIT):
+                _generate("user_andres")
+            # A different user starts with a fresh window. require_self is
+            # stubbed per-request from the body's user_id in conftest.
+            other = _generate("user_beatriz")
+        assert other.status_code == 200
+
+
+class TestDailySpendGuard:
+    def test_over_budget_user_is_refused_before_the_model_runs(self):
+        from services.quiz_config import QUIZ_DAILY_SPEND_CAP_USD
+
+        spent = [{"cost_usd": QUIZ_DAILY_SPEND_CAP_USD + 1.0}]
+        run = AsyncMock(return_value=SimpleNamespace(output=_quiz()))
+        with (
+            patch("routes.quiz.table", side_effect=_factory(spend_rows=spent)),
+            patch("routes.quiz.quiz_agent.run", new=run),
+        ):
+            r = _generate()
+        assert r.status_code == 429
+        assert r.json()["error"]["code"] == "QUIZ_DAILY_LIMIT_REACHED"
+        run.assert_not_called()
+
+    def test_under_budget_passes(self):
+        from services.quiz_config import QUIZ_DAILY_SPEND_CAP_USD
+
+        spent = [{"cost_usd": QUIZ_DAILY_SPEND_CAP_USD / 2}]
+        run = AsyncMock(return_value=SimpleNamespace(output=_quiz()))
+        with (
+            patch("routes.quiz.table", side_effect=_factory(spend_rows=spent)),
+            patch("routes.quiz.quiz_agent.run", new=run),
+        ):
+            r = _generate()
+        assert r.status_code == 200
+
+    def test_spend_sum_is_not_truncated_by_the_postgrest_page_cap(self):
+        """PostgREST caps a response at max_rows (1000) and answers with
+        206 — a 2xx — so an unpaged select silently truncates. Summing that
+        page plateaus below the ceiling, and the guard can never trip for
+        the runaway user it exists to stop."""
+        from services.quiz_config import QUIZ_DAILY_SPEND_CAP_USD
+
+        # Sized so ONE page is not enough: 1000 rows × $0.0015 = $1.50,
+        # under the $2.00 cap, while all 1500 rows = $2.25, over it. A
+        # reader that trusts a single truncated response concludes the
+        # user is fine; a paging reader catches them.
+        per_row = 0.0015
+        rows = [{"cost_usd": per_row} for _ in range(1500)]
+        pages: list[tuple[int, int]] = []
+
+        def factory(name):
+            mock = MagicMock()
+            if name == "llm_usage":
+                def _select(columns="*", filters=None, limit=None, offset=None, **kw):
+                    page_limit = min(limit or 1000, 1000)
+                    start = offset or 0
+                    pages.append((start, page_limit))
+                    return rows[start:start + page_limit]
+                mock.select.side_effect = _select
+            elif name == "graph_nodes":
+                mock.select.return_value = [{
+                    "id": "node1", "course_id": "course1",
+                    "concept_name": "Loops", "mastery_score": 0.5,
+                }]
+            else:
+                mock.select.return_value = []
+                mock.insert.return_value = [{"id": "q"}]
+            return mock
+
+        run = AsyncMock(return_value=SimpleNamespace(output=_quiz()))
+        with (
+            patch("routes.quiz.table", side_effect=factory),
+            patch("routes.quiz.quiz_agent.run", new=run),
+        ):
+            r = _generate()
+
+        assert len(pages) > 1, "the spend read must page, not trust one response"
+        assert r.status_code == 429
+        assert r.json()["error"]["code"] == "QUIZ_DAILY_LIMIT_REACHED"
+        run.assert_not_called()
+        assert 1500 * per_row > QUIZ_DAILY_SPEND_CAP_USD  # the premise
+
+    def test_spend_lookup_failure_never_blocks_a_quiz(self):
+        """The guard is a cost control, not a correctness gate — if the
+        usage table is unreachable we let the quiz through rather than
+        failing closed on every student."""
+        def factory(name):
+            mock = MagicMock()
+            if name == "graph_nodes":
+                mock.select.return_value = [{
+                    "id": "node1", "course_id": "course1",
+                    "concept_name": "Loops", "mastery_score": 0.5,
+                }]
+            elif name == "llm_usage":
+                mock.select.side_effect = RuntimeError("db down")
+            else:
+                mock.select.return_value = []
+                mock.insert.return_value = [{"id": "q"}]
+            return mock
+
+        with (
+            patch("routes.quiz.table", side_effect=factory),
+            patch("routes.quiz.quiz_agent.run",
+                  new=AsyncMock(return_value=SimpleNamespace(output=_quiz()))),
+        ):
+            r = _generate()
+        assert r.status_code == 200
+
+
+# ── F2: timeout → its own 502 code ──────────────────────────────────────────
+
+
+class TestGenerationTimeout:
+    def test_agent_timeout_maps_to_its_own_code(self):
+        from services.quiz_config import QUIZ_GENERATION_TIMEOUT_SEC
+
+        assert QUIZ_GENERATION_TIMEOUT_SEC > 0
+
+        async def _never(*a, **k):
+            await asyncio.sleep(3600)
+
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run", new=_never),
+            patch("routes.quiz.QUIZ_GENERATION_TIMEOUT_SEC", 0.05),
+        ):
+            r = _generate()
+        assert r.status_code == 502
+        assert r.json()["error"]["code"] == "QUIZ_GENERATION_TIMEOUT"
+
+    def test_a_partial_quiz_survives_a_top_up_timeout(self):
+        """The timeout must not throw away questions we already have.
+        Cancelling the whole generation raises CancelledError — a
+        BaseException — straight past #543's serve-what-we-have handler,
+        turning a valid 3-question quiz into a 502."""
+        import asyncio as _asyncio
+
+        from agents.quiz import Quiz, QuizQuestion
+
+        def _mk(n, correct=True):
+            return QuizQuestion(
+                question=f"Q{n}?", type="multiple_choice", difficulty="easy",
+                options=[f"a{n}", f"b{n}", f"c{n}", f"d{n}"],
+                correct_answer=f"a{n}" if correct else "NOPE",
+                explanation="x", concept="Loops",
+            )
+
+        # 3 good, 3 drifted → drop rate high enough to trigger the top-up.
+        first = Quiz(questions=[_mk(1), _mk(2), _mk(3),
+                                _mk(4, False), _mk(5, False), _mk(6, False)])
+        calls = {"n": 0}
+
+        async def _run(*a, **k):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                return SimpleNamespace(output=first)
+            await _asyncio.sleep(3600)   # the top-up hangs
+
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run", new=_run),
+            patch("routes.quiz.QUIZ_GENERATION_TIMEOUT_SEC", 0.2),
+        ):
+            r = client.post("/api/quiz/generate", json={
+                "user_id": "user_andres",
+                "concept_node_id": "node1",
+                "num_questions": 6,
+                "difficulty": "easy",
+                "use_shared_context": False,
+            })
+
+        assert r.status_code == 200, (
+            "a completed partial generation must be served, not discarded"
+        )
+        assert r.json()["delivered_count"] == 3
+        assert calls["n"] == 2  # the top-up was attempted and timed out
+
+    def test_ordinary_failure_keeps_the_generic_code(self):
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run",
+                  new=AsyncMock(side_effect=RuntimeError("boom"))),
+        ):
+            r = _generate()
+        assert r.status_code == 502
+        assert r.json()["error"]["code"] == "QUIZ_GENERATION_FAILED"
+
+
+# ── F3: failure events reach admin analytics ────────────────────────────────
+
+
+class TestGenerationFailureEvent:
+    def test_generation_failure_emits_an_event(self):
+        events = []
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run",
+                  new=AsyncMock(side_effect=RuntimeError("boom"))),
+            patch("routes.quiz.events_service.log_event",
+                  side_effect=lambda *a, **k: events.append((a, k))),
+        ):
+            r = _generate()
+        assert r.status_code == 502
+        failed = [(a, k) for a, k in events if a and a[0] == "quiz.generation_failed"]
+        assert len(failed) == 1, (
+            "a generation failure must be visible in admin analytics, not just logs"
+        )
+        _, kwargs = failed[0]
+        assert kwargs["category"] == "error"
+        assert kwargs["user_id"] == "user_andres"
+        payload = kwargs["payload"]
+        assert payload["concept_node_id"] == "node1"
+        assert payload["reason"] == "agent_error"
+
+    def test_event_is_in_the_pinned_taxonomy(self):
+        from services import events_service
+
+        assert "quiz.generation_failed" in events_service.EVENT_TAXONOMY
+
+    def test_rate_limit_rejection_is_not_an_error_event(self):
+        """A throttled student isn't a backend failure — it must not
+        pollute the error feed."""
+        from services.quiz_config import QUIZ_GENERATE_RATE_LIMIT
+
+        events = []
+        with (
+            patch("routes.quiz.table", side_effect=_factory()),
+            patch("routes.quiz.quiz_agent.run",
+                  new=AsyncMock(return_value=SimpleNamespace(output=_quiz()))),
+            patch("routes.quiz.events_service.log_event",
+                  side_effect=lambda *a, **k: events.append((a, k))),
+        ):
+            for _ in range(QUIZ_GENERATE_RATE_LIMIT + 1):
+                _generate()
+        assert not [a for a, _ in events if a and a[0] == "quiz.generation_failed"]
+
+
+# ── F4: ownership + semester scoping on deep links ──────────────────────────
+
+
+class TestConceptScoping:
+    """`?concept=` deep links hand the route an arbitrary node id. The
+    owner-scoped read is the gate: a node belonging to someone else, or to
+    a course the student isn't enrolled in, must 404 before the agent runs
+    — no content leak, no mastery write."""
+
+    def _scoped_select(self, *, owned_by="user_andres", course_id="course1"):
+        def _select(columns="*", filters=None, **_):
+            filters = filters or {}
+            if filters.get("id") != "eq.node_x":
+                return []
+            if filters.get("user_id") not in (None, f"eq.{owned_by}"):
+                return []
+            return [{
+                "id": "node_x", "user_id": owned_by, "course_id": course_id,
+                "concept_name": "Someone Else's Concept", "mastery_score": 0.4,
+            }]
+
+        return _select
+
+    def _post(self, user_id):
+        return client.post("/api/quiz/generate", json={
+            "user_id": user_id, "concept_node_id": "node_x",
+            "num_questions": 1, "difficulty": "easy",
+            "use_shared_context": False,
+        })
+
+    def test_foreign_concept_404s_before_the_agent_runs(self):
+        run = AsyncMock()
+
+        def factory(name):
+            mock = MagicMock()
+            if name == "graph_nodes":
+                mock.select.side_effect = self._scoped_select(owned_by="user_beatriz")
+            else:
+                mock.select.return_value = []
+            return mock
+
+        with (
+            patch("routes.quiz.table", side_effect=factory),
+            patch("routes.quiz.quiz_agent.run", new=run),
+        ):
+            r = self._post("user_andres")
+        assert r.status_code == 404
+        assert r.json()["error"]["code"] == "QUIZ_CONCEPT_NOT_FOUND"
+        run.assert_not_called()
+
+    def test_own_concept_from_a_past_semester_still_generates(self):
+        """Scoping is by OWNERSHIP, not by active semester: a student
+        revising last term's concept is legitimate. Pinning this stops a
+        future 'scope to active semester' change from silently breaking
+        revision — if that becomes desired it needs its own decision."""
+        run = AsyncMock(return_value=SimpleNamespace(output=_quiz()))
+
+        def factory(name):
+            mock = MagicMock()
+            if name == "graph_nodes":
+                mock.select.side_effect = self._scoped_select(course_id="old-course")
+            elif name == "quiz_attempts":
+                mock.insert.return_value = [{"id": "q"}]
+            else:
+                mock.select.return_value = []
+            return mock
+
+        with (
+            patch("routes.quiz.table", side_effect=factory),
+            patch("routes.quiz.quiz_agent.run", new=run),
+        ):
+            r = self._post("user_andres")
+        assert r.status_code == 200
+        run.assert_called_once()