diff --git a/backend/alembic/versions/p8b9c0d1e253_category_grants.py b/backend/alembic/versions/p8b9c0d1e253_category_grants.py
new file mode 100644
index 0000000..0c9c6d5
--- /dev/null
+++ b/backend/alembic/versions/p8b9c0d1e253_category_grants.py
@@ -0,0 +1,30 @@
+"""Per-category editorial grants for non-moderator educators.
+
+Revision ID: p8b9c0d1e253
+Revises: n7a8b9c0d142
+"""
+from alembic import op
+
+revision = "p8b9c0d1e253"
+down_revision = "n7a8b9c0d142"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+ op.execute("""
+ CREATE TABLE IF NOT EXISTS category_grants (
+ id SERIAL PRIMARY KEY,
+ category_id INTEGER NOT NULL REFERENCES question_categories(id) ON DELETE CASCADE,
+ user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ granted_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
+ created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ CONSTRAINT uq_category_grant UNIQUE (category_id, user_id)
+ )
+ """)
+ op.execute("CREATE INDEX IF NOT EXISTS ix_category_grants_category_id ON category_grants(category_id)")
+ op.execute("CREATE INDEX IF NOT EXISTS ix_category_grants_user_id ON category_grants(user_id)")
+
+
+def downgrade():
+ op.execute("DROP TABLE IF EXISTS category_grants")
diff --git a/backend/app/main.py b/backend/app/main.py
index f423cf0..b2fdf93 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -168,6 +168,7 @@ def setup_pgvector():
# Import new models so create_all picks them up
from app.models import quiz_category, quiz_question_link, question_category, favorite # noqa
from app.models import flashcard, course # noqa
+ from app.models import category_grant # noqa
# Kill stale idle-in-transaction connections from previous killed startups.
# They hold DDL locks and cause ALTER TABLE below to hang indefinitely.
diff --git a/backend/app/models/category_grant.py b/backend/app/models/category_grant.py
new file mode 100644
index 0000000..8752957
--- /dev/null
+++ b/backend/app/models/category_grant.py
@@ -0,0 +1,22 @@
+from datetime import datetime
+
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, UniqueConstraint
+
+from app.database import Base
+
+
+class CategoryGrant(Base):
+ """Lets a non-moderator edit the questions inside one category and its descendants.
+
+ A grant is scoped editorial access, not a role: the holder can create, edit
+ and delete questions filed under the granted category, and nothing else.
+ """
+
+ __tablename__ = "category_grants"
+ __table_args__ = (UniqueConstraint("category_id", "user_id", name="uq_category_grant"),)
+
+ id = Column(Integer, primary_key=True, index=True)
+ category_id = Column(Integer, ForeignKey("question_categories.id", ondelete="CASCADE"), nullable=False, index=True)
+ user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
+ granted_by = Column(Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
+ created_at = Column(DateTime, default=datetime.utcnow)
diff --git a/backend/app/routers/question_categories.py b/backend/app/routers/question_categories.py
index e708304..704435a 100644
--- a/backend/app/routers/question_categories.py
+++ b/backend/app/routers/question_categories.py
@@ -6,11 +6,13 @@ from sqlalchemy.orm import Session
from app.database import get_db
from app.models.question import Question
+from app.models.category_grant import CategoryGrant
from app.models.question_category import QuestionCategory, QuestionCategoryLink
from app.models.user import User
from app.services.quiz_builder import (bank_query, filtered_bank_query, category_descendants, category_breadcrumbs,
validate_parent, GenerateTestRequest, generate_test, bank_question_predicate)
from app.utils.auth import get_current_user, require_moderator
+from app.utils.category_grants import manageable_categories
router = APIRouter()
@@ -65,6 +67,110 @@ def _category_totals(db, user, by_id):
return {cid: len(question_set) for cid, question_set in totals.items()}
+class GrantIn(BaseModel):
+ user_id: int
+
+
+def _grant_json(grant, users, categories):
+ user = users.get(grant.user_id)
+ return {
+ "id": grant.id,
+ "category_id": grant.category_id,
+ "category_name": categories.get(grant.category_id),
+ "user_id": grant.user_id,
+ "user_name": user.name if user else None,
+ "user_email": user.email if user else None,
+ "created_at": grant.created_at.isoformat() if grant.created_at else None,
+ }
+
+
+@router.get("/grants")
+def list_category_grants(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(require_moderator),
+):
+ """Every per-category editorial grant, for the question-manager admin panel."""
+ grants = db.query(CategoryGrant).order_by(CategoryGrant.id).all()
+ categories = {cat.id: cat.name for cat in db.query(QuestionCategory).all()}
+ users = {user.id: user for user in db.query(User).filter(
+ User.id.in_({g.user_id for g in grants}))} if grants else {}
+ return [_grant_json(grant, users, categories) for grant in grants]
+
+
+@router.get("/grantable-users")
+def grantable_users(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(require_moderator),
+):
+ """Users a grant can be given to — moderators already manage everything."""
+ rows = db.query(User.id, User.name, User.email, User.role).order_by(User.name).all()
+ return [{"id": uid, "name": name, "email": email}
+ for uid, name, email, role in rows if role not in ("admin", "moderator")]
+
+
+@router.post("/{cat_id}/grants", status_code=201)
+def add_category_grant(
+ cat_id: int,
+ data: GrantIn,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(require_moderator),
+):
+ """Let one user edit the questions in this category and everything under it."""
+ category = db.get(QuestionCategory, cat_id)
+ if not category:
+ raise HTTPException(404, "Category not found")
+ user = db.get(User, data.user_id)
+ if not user:
+ raise HTTPException(404, "User not found")
+ if user.is_moderator:
+ raise HTTPException(400, "Moderators already manage every category")
+ existing = db.query(CategoryGrant).filter_by(category_id=cat_id, user_id=data.user_id).first()
+ if existing:
+ raise HTTPException(409, "This user already has a grant for this category")
+ grant = CategoryGrant(category_id=cat_id, user_id=data.user_id, granted_by=current_user.id)
+ db.add(grant)
+ db.commit()
+ db.refresh(grant)
+ return _grant_json(grant, {user.id: user}, {category.id: category.name})
+
+
+@router.delete("/{cat_id}/grants/{user_id}", status_code=204)
+def remove_category_grant(
+ cat_id: int,
+ user_id: int,
+ db: Session = Depends(get_db),
+ current_user: User = Depends(require_moderator),
+):
+ grant = db.query(CategoryGrant).filter_by(category_id=cat_id, user_id=user_id).first()
+ if not grant:
+ raise HTTPException(404, "Grant not found")
+ db.delete(grant)
+ db.commit()
+
+
+@router.get("/my-grants")
+def my_category_grants(
+ db: Session = Depends(get_db),
+ current_user: User = Depends(get_current_user),
+):
+ """What the signed-in user may edit — drives the question-manager entry point."""
+ scope = manageable_categories(db, current_user)
+ if scope is None:
+ return {"is_moderator": True, "can_manage_questions": True, "categories": []}
+ names = {cat.id: cat.name for cat in db.query(QuestionCategory).filter(
+ QuestionCategory.id.in_(scope))} if scope else {}
+ granted = {row[0] for row in db.query(CategoryGrant.category_id).filter(
+ CategoryGrant.user_id == current_user.id).all()}
+ return {
+ "is_moderator": False,
+ "can_manage_questions": bool(scope),
+ "categories": sorted(
+ ({"id": cid, "name": names.get(cid), "direct": cid in granted} for cid in scope),
+ key=lambda row: (row["name"] or ""),
+ ),
+ }
+
+
@router.get("/", response_model=list[QCatResponse])
def list_question_categories(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
cats = db.query(QuestionCategory).order_by(QuestionCategory.name).all()
diff --git a/backend/app/routers/questions.py b/backend/app/routers/questions.py
index 41a067d..3a132cb 100644
--- a/backend/app/routers/questions.py
+++ b/backend/app/routers/questions.py
@@ -26,6 +26,8 @@ from app.models.favorite import Favorite
from app.services.quiz_builder import (bank_query, category_descendants, filtered_bank_query,
CreateFromBankRequest, GenerateTestRequest, create_saved_test, generate_test)
from app.utils.auth import get_current_user, require_moderator
+from app.utils.category_grants import (assert_can_manage_category, assert_can_manage_questions,
+ is_question_manager, manageable_categories, question_in_scope, require_question_manager)
router = APIRouter()
@@ -51,7 +53,9 @@ def delete_question(
if not question:
raise HTTPException(status_code=404, detail="Question not found")
is_mod = current_user.role in ("admin", "moderator")
- if not is_mod:
+ scope = manageable_categories(db, current_user)
+ granted = scope is not None and bool(scope) and question_in_scope(db, scope, question)
+ if not is_mod and not granted:
if question.user_id != current_user.id:
raise HTTPException(status_code=403, detail="Not authorized to delete this question")
# Regular users can only delete questions they created (no source quiz)
@@ -81,12 +85,19 @@ def edit_question(
question_id: int,
data: QuestionEdit,
db: Session = Depends(get_db),
- current_user: User = Depends(require_moderator),
+ current_user: User = Depends(get_current_user),
):
- """Edit a question. Only admins and educators manage questions."""
+ """Edit a question — moderators, or an educator granted its category."""
+ scope = require_question_manager(db, current_user)
question = db.query(Question).filter(Question.id == question_id).first()
if not question:
raise HTTPException(status_code=404, detail="Question not found")
+ assert_can_manage_questions(db, scope, [question_id])
+ if "question_category_id" in data.model_fields_set:
+ # Moving a question out of your scope would put it beyond your reach.
+ assert_can_manage_category(scope, data.question_category_id)
+ for extra_id in (data.additional_category_ids or []):
+ assert_can_manage_category(scope, extra_id)
values = validate_image_attachments(db, current_user, data.model_dump(exclude_unset=True))
if values.get("question_category_id") is not None and not db.get(QuestionCategory, values["question_category_id"]):
raise HTTPException(400, "Category not found")
@@ -427,9 +438,11 @@ class ManualQuestionCreate(BaseModel):
def create_question_manually(
data: ManualQuestionCreate,
db: Session = Depends(get_db),
- current_user: User = Depends(require_moderator),
+ current_user: User = Depends(get_current_user),
):
- """Create a single question manually (not from PDF extraction)."""
+ """Create a single question manually — moderators, or an educator in their categories."""
+ scope = require_question_manager(db, current_user)
+ assert_can_manage_category(scope, data.question_category_id)
q_text = data.question_text.strip()
if not q_text:
raise HTTPException(status_code=400, detail="Question text is required")
@@ -582,14 +595,18 @@ class BulkQuestionAction(BaseModel):
def bulk_question_action(
data: BulkQuestionAction,
db: Session = Depends(get_db),
- current_user: User = Depends(require_moderator),
+ current_user: User = Depends(get_current_user),
):
"""Apply one editorial action to up to 500 checked questions."""
+ scope = require_question_manager(db, current_user)
ids = list(dict.fromkeys(data.question_ids))
if not ids:
raise HTTPException(400, "No questions selected")
if len(ids) > 500:
raise HTTPException(400, "Select at most 500 questions per action")
+ assert_can_manage_questions(db, scope, ids)
+ if data.action == "category":
+ assert_can_manage_category(scope, data.category_id)
rows = db.query(Question).filter(Question.id.in_(ids))
if data.action == "category":
@@ -612,10 +629,20 @@ def bulk_question_action(
@router.get("/manage/summary")
def question_manager_summary(
db: Session = Depends(get_db),
- current_user: User = Depends(require_moderator),
+ current_user: User = Depends(get_current_user),
):
- """Editorial health counters shown as one-click filters in the question manager."""
+ """Editorial health counters shown as one-click filters in the question manager.
+
+ A granted educator sees the counts for their own categories only.
+ """
+ scope = require_question_manager(db, current_user)
base = db.query(func.count(Question.id))
+ if scope is not None:
+ base = base.filter(or_(
+ Question.question_category_id.in_(scope),
+ Question.id.in_(select(QuestionCategoryLink.question_id).where(
+ QuestionCategoryLink.category_id.in_(scope))),
+ ))
blank = ("", " ")
return {
"total": base.scalar() or 0,
@@ -625,6 +652,7 @@ def question_manager_summary(
or_(Question.explanation.is_(None), Question.explanation.in_(blank))).scalar() or 0,
"private": base.filter(Question.is_shared == 0).scalar() or 0,
"mine": base.filter(Question.user_id == current_user.id).scalar() or 0,
+ "scoped": scope is not None,
}
diff --git a/backend/app/routers/study_tools.py b/backend/app/routers/study_tools.py
index bf24087..c66c5bd 100644
--- a/backend/app/routers/study_tools.py
+++ b/backend/app/routers/study_tools.py
@@ -1,8 +1,9 @@
"""Educator-maintained lab references and authorized question response statistics."""
-from collections import Counter
+from collections import Counter, defaultdict
from datetime import datetime
+from typing import Literal
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field, HttpUrl, field_validator
from sqlalchemy import func, or_
from sqlalchemy.orm import Session
@@ -17,6 +18,7 @@ from app.models.question_category import QuestionCategory, QuestionCategoryLink
from app.models.quiz import Quiz
from app.models.user import User
from app.utils.auth import get_current_user, require_moderator
+from app.services.quiz_builder import shareable_question_predicate
from app.utils.quiz_access import general_quiz_visibility, require_quiz_access
from app.utils.quiz_questions import question_in_quiz
@@ -208,6 +210,170 @@ def performance_by_category(db: Session = Depends(get_db), user: User = Depends(
}
+# Readiness needs enough answers before a per-category estimate means anything.
+READINESS_UNLOCK_ANSWERS = 40
+# Shrinkage weight: a category with this many answers sits halfway between its
+# own accuracy and the learner's overall accuracy.
+READINESS_PRIOR_ANSWERS = 8
+
+
+def _category_rollup(categories):
+ """Map every category to itself plus all of its ancestors, for roll-up counting."""
+ parents = {cat.id: cat.parent_id for cat in categories}
+ ancestry: dict[int, list[int]] = {}
+ for cid in parents:
+ chain, cursor, guard = [], cid, 0
+ while cursor is not None and guard < 12:
+ chain.append(cursor)
+ cursor = parents.get(cursor)
+ guard += 1
+ ancestry[cid] = chain
+ return ancestry
+
+
+@router.get("/recommendations")
+def study_recommendations(
+ group: Literal["systems", "subtopics"] = "systems",
+ limit: int = Query(20, ge=1, le=60),
+ db: Session = Depends(get_db),
+ user: User = Depends(get_current_user),
+):
+ """Focus areas ranked by the study time most likely to raise the learner's score.
+
+ Readiness is the learner's accuracy in a category shrunk toward their own
+ overall accuracy in proportion to how few answers that category has, so a
+ single unlucky question does not read as a knowledge gap. It is a plain
+ empirical-Bayes estimate over recorded answers — not a psychometric exam
+ score, and not a prediction of any real examination.
+ """
+ categories = db.query(QuestionCategory).all()
+ ancestry = _category_rollup(categories)
+ names = {cat.id: cat.name for cat in categories}
+ parents = {cat.id: cat.parent_id for cat in categories}
+ top_level = {cat.id for cat in categories if cat.parent_id is None}
+
+ # ── What the learner has answered ──────────────────────────────
+ answered_rows = db.query(
+ AttemptAnswer.question_id, AttemptAnswer.is_correct, Question.question_category_id,
+ ).join(QuizAttempt, QuizAttempt.id == AttemptAnswer.attempt_id
+ ).join(Quiz, Quiz.id == QuizAttempt.quiz_id
+ ).join(Question, Question.id == AttemptAnswer.question_id
+ ).filter(
+ QuizAttempt.user_id == user.id,
+ QuizAttempt.completed_at.isnot(None),
+ or_(QuizAttempt.expired == 0, QuizAttempt.expired.is_(None)),
+ Quiz.course_id.is_(None),
+ ).all()
+
+ extra_links: dict[int, set[int]] = defaultdict(set)
+ for question_id, category_id in db.query(
+ QuestionCategoryLink.question_id, QuestionCategoryLink.category_id).all():
+ extra_links[question_id].add(category_id)
+
+ def categories_for(question_id, primary):
+ direct = extra_links.get(question_id, set()) | ({primary} if primary else set())
+ rolled = set()
+ for cid in direct:
+ rolled.update(ancestry.get(cid, [cid]))
+ return rolled
+
+ answered: dict[int, int] = defaultdict(int)
+ correct: dict[int, int] = defaultdict(int)
+ seen_questions: dict[int, set[int]] = defaultdict(set)
+ total_answers = len(answered_rows)
+ total_correct = sum(1 for _, is_correct, _ in answered_rows if is_correct)
+ for question_id, is_correct, primary in answered_rows:
+ for cid in categories_for(question_id, primary):
+ answered[cid] += 1
+ seen_questions[cid].add(question_id)
+ if is_correct:
+ correct[cid] += 1
+
+ # ── How much bank material each category holds ─────────────────
+ available: dict[int, int] = defaultdict(int)
+ bank_total = 0
+ for question_id, primary in db.query(Question.id, Question.question_category_id).filter(
+ shareable_question_predicate()).all():
+ bank_total += 1
+ for cid in categories_for(question_id, primary):
+ available[cid] += 1
+
+ articles = {}
+ for article in db.query(Article).filter(Article.status == "published",
+ Article.category_id.isnot(None)).all():
+ articles.setdefault(article.category_id, article)
+
+ overall_accuracy = (total_correct / total_answers) if total_answers else 0.0
+ unlocked = total_answers >= READINESS_UNLOCK_ANSWERS
+
+ scope = top_level if group == "systems" else {cat.id for cat in categories if cat.parent_id is not None}
+ rows = []
+ for cid in scope:
+ seen = len(seen_questions.get(cid, ()))
+ pool = available.get(cid, 0)
+ if pool == 0 and seen == 0:
+ continue
+ n = answered.get(cid, 0)
+ c = correct.get(cid, 0)
+ accuracy = round(100 * c / n, 1) if n else None
+ readiness = None
+ if unlocked and n:
+ shrunk = (c + READINESS_PRIOR_ANSWERS * overall_accuracy) / (n + READINESS_PRIOR_ANSWERS)
+ readiness = round(100 * shrunk, 1)
+ relevance = round(100 * pool / bank_total, 1) if bank_total else 0.0
+ coverage = round(100 * seen / pool, 1) if pool else 0.0
+ article = articles.get(cid)
+ rows.append({
+ "category_id": cid,
+ "name": names.get(cid, "Uncategorized"),
+ "parent_id": parents.get(cid),
+ "parent_name": names.get(parents.get(cid)),
+ "answered": n,
+ "correct": c,
+ "seen_questions": seen,
+ "available": pool,
+ "coverage": coverage,
+ "accuracy": accuracy,
+ "readiness": readiness,
+ "relevance": relevance,
+ "status": "no_data" if not n else "focus" if (readiness if readiness is not None else accuracy) < 70 else "proficient",
+ "article_id": article.id if article else None,
+ "article_title": article.title if article else None,
+ })
+
+ # Priority: weak-and-relevant first, then untouched material by relevance.
+ baseline = 100 * overall_accuracy if total_answers else 70.0
+ for row in rows:
+ score = row["readiness"] if row["readiness"] is not None else row["accuracy"]
+ gap = (baseline - score) / 100 if score is not None else 0.5 # unseen material sits mid-priority
+ unseen = 1 - (row["coverage"] / 100)
+ row["priority"] = round(max(gap, 0.0) * (row["relevance"] / 100) + 0.25 * unseen * (row["relevance"] / 100), 5)
+ rows.sort(key=lambda row: (-row["priority"], -row["relevance"], row["name"]))
+ focus_ids = {row["category_id"] for row in rows[:3] if row["answered"]}
+ for row in rows:
+ if row["category_id"] in focus_ids:
+ row["is_focus_area"] = True
+ else:
+ row["is_focus_area"] = False
+
+ return {
+ "group": group,
+ "unlocked": unlocked,
+ "answers_needed": max(0, READINESS_UNLOCK_ANSWERS - total_answers),
+ "total_answered": total_answers,
+ "unique_questions_seen": len({row[0] for row in answered_rows}),
+ "bank_total": bank_total,
+ "overall_accuracy": round(100 * overall_accuracy, 1) if total_answers else None,
+ "focus_areas": rows[:limit],
+ "basis": (
+ "Your completed, non-expired general-bank answers, rolled up through the category tree. "
+ f"Readiness shrinks each category's accuracy toward your overall {round(100 * overall_accuracy)}% "
+ "so small samples do not overstate a gap; it unlocks after "
+ f"{READINESS_UNLOCK_ANSWERS} answers. Relevance is the share of the question bank a category holds. "
+ "These are study hints from your own answers, not an exam score prediction."
+ ),
+ }
+
@router.get("/attempts/{attempt_id}/questions/{question_id}/responses")
def question_responses(attempt_id: int, question_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)):
attempt = db.query(QuizAttempt).filter_by(id=attempt_id, user_id=user.id).first()
diff --git a/backend/app/schemas/quiz.py b/backend/app/schemas/quiz.py
index cf621ac..bc609bc 100644
--- a/backend/app/schemas/quiz.py
+++ b/backend/app/schemas/quiz.py
@@ -31,6 +31,7 @@ class QuestionResponse(BaseModel):
options: list[str] | None
image_path: str | None = None
question_category_id: int | None = None
+ difficulty: str | None = None
category_breadcrumbs: list[dict] = Field(default_factory=list)
class Config:
diff --git a/backend/app/utils/category_grants.py b/backend/app/utils/category_grants.py
new file mode 100644
index 0000000..5fa8346
--- /dev/null
+++ b/backend/app/utils/category_grants.py
@@ -0,0 +1,86 @@
+"""Per-category editorial access for non-moderator educators.
+
+A moderator or admin manages every question. Everyone else manages only the
+questions filed under a category they hold a grant on, or under any descendant
+of it. `manageable_categories` returns None to mean "no restriction".
+"""
+from fastapi import HTTPException
+from sqlalchemy.orm import Session
+
+from app.models.category_grant import CategoryGrant
+from app.models.question import Question
+from app.models.question_category import QuestionCategory, QuestionCategoryLink
+from app.models.user import User
+
+
+def _descendants(db: Session, roots: set[int]) -> set[int]:
+ """Expand category ids to include every category beneath them."""
+ if not roots:
+ return set()
+ children: dict[int, list[int]] = {}
+ for cid, parent_id in db.query(QuestionCategory.id, QuestionCategory.parent_id).all():
+ if parent_id is not None:
+ children.setdefault(parent_id, []).append(cid)
+ seen, stack = set(roots), list(roots)
+ while stack:
+ for child in children.get(stack.pop(), ()):
+ if child not in seen:
+ seen.add(child)
+ stack.append(child)
+ return seen
+
+
+def manageable_categories(db: Session, user: User) -> set[int] | None:
+ """Category ids this user may edit questions in; None means all of them."""
+ if user.is_moderator:
+ return None
+ roots = {row[0] for row in db.query(CategoryGrant.category_id).filter(
+ CategoryGrant.user_id == user.id).all()}
+ return _descendants(db, roots)
+
+
+def is_question_manager(db: Session, user: User) -> bool:
+ """True when the user may edit questions somewhere."""
+ scope = manageable_categories(db, user)
+ return scope is None or bool(scope)
+
+
+def require_question_manager(db: Session, user: User) -> set[int] | None:
+ """Gate a question-management route; returns the caller's category scope."""
+ scope = manageable_categories(db, user)
+ if scope is not None and not scope:
+ raise HTTPException(403, "Question management requires moderator access or a category grant")
+ return scope
+
+
+def assert_can_manage_category(scope: set[int] | None, category_id: int | None, what: str = "category") -> None:
+ if scope is None:
+ return
+ if category_id is None or category_id not in scope:
+ raise HTTPException(403, f"You do not have an editorial grant for this {what}")
+
+
+def assert_can_manage_questions(db: Session, scope: set[int] | None, question_ids: list[int]) -> None:
+ """Every target question must sit in a granted category (primary or additional)."""
+ if scope is None or not question_ids:
+ return
+ primary = dict(db.query(Question.id, Question.question_category_id).filter(
+ Question.id.in_(question_ids)).all())
+ extra: dict[int, set[int]] = {}
+ for qid, cid in db.query(QuestionCategoryLink.question_id, QuestionCategoryLink.category_id).filter(
+ QuestionCategoryLink.question_id.in_(question_ids)).all():
+ extra.setdefault(qid, set()).add(cid)
+ for question_id in question_ids:
+ if question_id not in primary:
+ continue # a missing row is reported as 404 by the caller, not as a permission error
+ owned = ({primary[question_id]} if primary[question_id] else set()) | extra.get(question_id, set())
+ if not (owned & scope):
+ raise HTTPException(403, "You do not have an editorial grant for one of these questions")
+
+
+def question_in_scope(db: Session, scope: set[int], question: Question) -> bool:
+ """True when a question sits in a granted category (primary or additional)."""
+ owned = {question.question_category_id} if question.question_category_id else set()
+ owned |= {row[0] for row in db.query(QuestionCategoryLink.category_id).filter(
+ QuestionCategoryLink.question_id == question.id).all()}
+ return bool(owned & scope)
diff --git a/backend/tests/test_category_grants.py b/backend/tests/test_category_grants.py
new file mode 100644
index 0000000..5432ea4
--- /dev/null
+++ b/backend/tests/test_category_grants.py
@@ -0,0 +1,188 @@
+"""Per-category editorial grants: a non-moderator educator edits only their categories.
+
+Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests
+No application startup, external services or AI calls; a disposable SQLite database per test.
+"""
+import os
+os.environ["DATABASE_URL"] = "sqlite:///:memory:"
+
+import unittest
+from unittest.mock import patch
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.database import Base, get_db
+from app.models.category_grant import CategoryGrant
+from app.models.question import Question
+from app.models.question_category import QuestionCategory, QuestionCategoryLink
+from app.models.user import User
+from app.routers import question_categories, questions
+from app.utils.auth import get_current_user
+
+
+class CategoryGrantTests(unittest.TestCase):
+ def setUp(self):
+ self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+ Base.metadata.create_all(self.engine)
+ self.db = Session(self.engine)
+
+ self.mod = User(id=1, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")
+ self.educator = User(id=2, name="Educator", email="edu@example.test", hashed_password="unused")
+ self.outsider = User(id=3, name="Outsider", email="out@example.test", hashed_password="unused")
+ self.db.add_all([self.mod, self.educator, self.outsider])
+ # Cardiology > Kawasaki disease · Neurology stands alone.
+ self.db.add_all([
+ QuestionCategory(id=1, name="Cardiology", user_id=1),
+ QuestionCategory(id=2, name="Kawasaki disease", parent_id=1, user_id=1),
+ QuestionCategory(id=10, name="Neurology", user_id=1),
+ ])
+ self.db.flush()
+ for qid, category in [(1, 1), (2, 2), (3, 10), (4, None)]:
+ self.db.add(Question(id=qid, question_category_id=category, user_id=1, is_shared=1,
+ question_text=f"Question {qid}", question_type="mcq",
+ options=["yes", "no"], correct_answer="yes", explanation="Because"))
+ self.db.commit()
+
+ self.user = self.mod
+ app = FastAPI()
+ app.include_router(questions.router, prefix="/questions")
+ app.include_router(question_categories.router, prefix="/question-categories")
+ app.dependency_overrides[get_db] = lambda: self.db
+ app.dependency_overrides[get_current_user] = lambda: self.user
+ self.client = TestClient(app)
+
+ def tearDown(self):
+ self.client.close()
+ self.db.close()
+ self.engine.dispose()
+
+ def grant(self, category_id=1, user_id=2):
+ self.db.add(CategoryGrant(category_id=category_id, user_id=user_id, granted_by=1))
+ self.db.commit()
+
+ # ── Granting ───────────────────────────────────────────────────
+ def test_moderator_grants_and_revokes_a_category(self):
+ response = self.client.post("/question-categories/1/grants", json={"user_id": 2})
+ self.assertEqual(response.status_code, 201, response.text)
+ self.assertEqual(response.json()["category_name"], "Cardiology")
+ self.assertEqual(response.json()["user_email"], "edu@example.test")
+
+ listed = self.client.get("/question-categories/grants").json()
+ self.assertEqual([(g["category_id"], g["user_id"]) for g in listed], [(1, 2)])
+
+ self.assertEqual(self.client.post("/question-categories/1/grants", json={"user_id": 2}).status_code, 409)
+ self.assertEqual(self.client.delete("/question-categories/1/grants/2").status_code, 204)
+ self.assertEqual(self.client.get("/question-categories/grants").json(), [])
+
+ def test_grants_are_moderator_only_and_validate_their_target(self):
+ self.assertEqual(self.client.post("/question-categories/999/grants", json={"user_id": 2}).status_code, 404)
+ self.assertEqual(self.client.post("/question-categories/1/grants", json={"user_id": 999}).status_code, 404)
+ # A moderator already manages everything; a grant would be meaningless.
+ self.assertEqual(self.client.post("/question-categories/1/grants", json={"user_id": 1}).status_code, 400)
+ self.user = self.educator
+ self.assertEqual(self.client.post("/question-categories/1/grants", json={"user_id": 2}).status_code, 403)
+ self.assertEqual(self.client.get("/question-categories/grants").status_code, 403)
+
+ def test_my_grants_reports_the_scope_including_descendants(self):
+ self.user = self.educator
+ self.assertEqual(self.client.get("/question-categories/my-grants").json(),
+ {"is_moderator": False, "can_manage_questions": False, "categories": []})
+ self.grant(category_id=1)
+ scope = self.client.get("/question-categories/my-grants").json()
+ self.assertTrue(scope["can_manage_questions"])
+ self.assertEqual([(c["name"], c["direct"]) for c in scope["categories"]],
+ [("Cardiology", True), ("Kawasaki disease", False)])
+ self.user = self.mod
+ self.assertTrue(self.client.get("/question-categories/my-grants").json()["is_moderator"])
+
+ # ── Using a grant ──────────────────────────────────────────────
+ def test_educator_edits_inside_the_grant_and_its_descendants_only(self):
+ self.grant(category_id=1)
+ self.user = self.educator
+ self.assertEqual(self.client.patch("/questions/1", json={"explanation": "Edited"}).status_code, 200)
+ # Question 2 sits in the child category, still inside the grant.
+ self.assertEqual(self.client.patch("/questions/2", json={"difficulty": "hard"}).status_code, 200)
+ # Neurology and the uncategorised question are out of scope.
+ self.assertEqual(self.client.patch("/questions/3", json={"explanation": "No"}).status_code, 403)
+ self.assertEqual(self.client.patch("/questions/4", json={"explanation": "No"}).status_code, 403)
+ self.assertEqual(self.db.get(Question, 3).explanation, "Because")
+
+ def test_educator_cannot_move_a_question_out_of_their_scope(self):
+ self.grant(category_id=1)
+ self.user = self.educator
+ self.assertEqual(self.client.patch("/questions/1", json={"question_category_id": 10}).status_code, 403)
+ self.assertEqual(self.client.patch("/questions/1", json={"additional_category_ids": [10]}).status_code, 403)
+ self.assertEqual(self.client.patch("/questions/1", json={"question_category_id": 2}).status_code, 200)
+ self.assertEqual(self.db.get(Question, 1).question_category_id, 2)
+
+ def test_educator_creates_only_inside_their_scope(self):
+ self.grant(category_id=1)
+ self.user = self.educator
+ body = {"question_text": "New question", "question_type": "mcq",
+ "options": ["yes", "no"], "correct_answer": "yes"}
+ with patch("app.services.embedding_service.embed_question", return_value=None):
+ allowed = self.client.post("/questions/create", json={**body, "question_category_id": 2})
+ self.assertEqual(allowed.status_code, 200, allowed.text)
+ self.assertEqual(self.client.post("/questions/create",
+ json={**body, "question_category_id": 10}).status_code, 403)
+ # Uncategorised creation is out of scope too — it would land outside the grant.
+ self.assertEqual(self.client.post("/questions/create", json=body).status_code, 403)
+
+ def test_bulk_actions_are_confined_to_the_grant(self):
+ self.grant(category_id=1)
+ self.user = self.educator
+ mixed = self.client.post("/questions/bulk",
+ json={"question_ids": [1, 3], "action": "difficulty", "difficulty": "easy"})
+ self.assertEqual(mixed.status_code, 403)
+ self.assertIsNone(self.db.get(Question, 1).difficulty) # nothing applied
+
+ ok = self.client.post("/questions/bulk",
+ json={"question_ids": [1, 2], "action": "difficulty", "difficulty": "easy"})
+ self.assertEqual(ok.status_code, 200, ok.text)
+ self.assertEqual(self.db.get(Question, 1).difficulty, "easy")
+
+ # Reassigning into an ungranted category is refused.
+ self.assertEqual(self.client.post("/questions/bulk",
+ json={"question_ids": [1], "action": "category",
+ "category_id": 10}).status_code, 403)
+
+ def test_additional_category_links_bring_a_question_into_scope(self):
+ self.grant(category_id=10)
+ self.db.add(QuestionCategoryLink(question_id=1, category_id=10))
+ self.db.commit()
+ self.user = self.educator
+ self.assertEqual(self.client.patch("/questions/1", json={"explanation": "Edited"}).status_code, 200)
+
+ def test_educator_deletes_only_inside_the_grant(self):
+ self.grant(category_id=1)
+ self.user = self.educator
+ self.assertEqual(self.client.delete("/questions/3").status_code, 403)
+ self.assertEqual(self.client.delete("/questions/1").status_code, 204)
+ self.assertIsNone(self.db.get(Question, 1))
+
+ def test_summary_is_scoped_to_the_grant(self):
+ self.grant(category_id=1)
+ self.user = self.educator
+ summary = self.client.get("/questions/manage/summary").json()
+ self.assertTrue(summary["scoped"])
+ self.assertEqual(summary["total"], 2) # questions 1 and 2 only
+ moderator_summary = None
+ self.user = self.mod
+ moderator_summary = self.client.get("/questions/manage/summary").json()
+ self.assertFalse(moderator_summary["scoped"])
+ self.assertEqual(moderator_summary["total"], 4)
+
+ def test_ungranted_user_is_locked_out_of_question_management(self):
+ self.user = self.outsider
+ self.assertEqual(self.client.get("/questions/manage/summary").status_code, 403)
+ self.assertEqual(self.client.patch("/questions/1", json={"explanation": "No"}).status_code, 403)
+ self.assertEqual(self.client.post("/questions/bulk",
+ json={"question_ids": [1], "action": "delete"}).status_code, 403)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/backend/tests/test_quiz_sessions.py b/backend/tests/test_quiz_sessions.py
index 97bac82..cdfce33 100644
--- a/backend/tests/test_quiz_sessions.py
+++ b/backend/tests/test_quiz_sessions.py
@@ -20,7 +20,7 @@ from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.database import Base, get_db
-from app.models.attempt import QuizAttempt
+from app.models.attempt import AttemptAnswer, QuizAttempt
from app.models.course import Course
from app.models.question import Question
from app.models.question_category import QuestionCategory, QuestionCategoryLink
@@ -28,7 +28,8 @@ from app.models.quiz import Quiz
from app.models.quiz_category import QuizCategory
from app.models.quiz_question_link import QuizQuestionLink
from app.models.user import User
-from app.routers import questions
+from app.models.article import Article
+from app.routers import questions, study_tools
from app.utils.auth import get_current_user
with patch.dict(sys.modules, {"app.services.quiz_service": ModuleType("app.services.quiz_service")}):
@@ -231,5 +232,115 @@ class QuestionManagerTests(unittest.TestCase):
json={"question_ids": [1], "action": "delete"}).status_code, 403)
+class RecommendationTests(unittest.TestCase):
+ """Focus-area ranking: readiness shrinkage, relevance and roll-up through the tree."""
+
+ def setUp(self):
+ self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+ Base.metadata.create_all(self.engine)
+ self.db = Session(self.engine)
+
+ self.user = User(id=1, name="Learner", email="learner@example.test", hashed_password="unused")
+ self.db.add_all([self.user, User(id=3, name="Mod", email="mod@example.test",
+ hashed_password="unused", role="moderator")])
+ self.db.add_all([
+ QuestionCategory(id=1, name="Cardiology", user_id=3),
+ QuestionCategory(id=2, name="Kawasaki disease", parent_id=1, user_id=3),
+ QuestionCategory(id=10, name="Neurology", user_id=3),
+ ])
+ self.db.add(Quiz(id=1, title="Bank test", user_id=1, is_published=1, questions_count=6))
+ self.db.flush()
+ # 6 shareable bank questions: 4 cardiology (all under the Kawasaki child), 2 neurology.
+ for qid, category in [(1, 2), (2, 2), (3, 2), (4, 2), (5, 10), (6, 10)]:
+ self.db.add(Question(id=qid, question_category_id=category, user_id=3, is_shared=1,
+ question_text=f"Question {qid}", question_type="mcq",
+ options=["yes", "no"], correct_answer="yes"))
+ self.db.add(Article(id=7, slug="kawasaki", title="Kawasaki disease", sections=[],
+ category_id=1, user_id=3, status="published"))
+ self.db.commit()
+
+ app = FastAPI()
+ app.include_router(study_tools.router, prefix="/study-tools")
+ app.dependency_overrides[get_db] = lambda: self.db
+ app.dependency_overrides[get_current_user] = lambda: self.user
+ self.client = TestClient(app)
+
+ def tearDown(self):
+ self.client.close()
+ self.db.close()
+ self.engine.dispose()
+
+ def answer(self, question_id, correct):
+ attempt = self.db.query(QuizAttempt).filter_by(user_id=1, quiz_id=1).first()
+ if attempt is None:
+ attempt = QuizAttempt(quiz_id=1, user_id=1, mode="study", total_questions=6,
+ completed_at=datetime(2026, 3, 1))
+ self.db.add(attempt)
+ self.db.flush()
+ self.db.add(AttemptAnswer(attempt_id=attempt.id, question_id=question_id,
+ user_answer="yes" if correct else "no", is_correct=correct))
+ self.db.commit()
+
+ def recommend(self, **params):
+ response = self.client.get("/study-tools/recommendations", params=params)
+ self.assertEqual(response.status_code, 200, response.text)
+ return response.json()
+
+ def test_relevance_and_coverage_roll_up_to_the_parent_system(self):
+ data = self.recommend()
+ rows = {row["name"]: row for row in data["focus_areas"]}
+ # Cardiology owns 4 of 6 bank questions through its child category.
+ self.assertEqual(rows["Cardiology"]["available"], 4)
+ self.assertEqual(rows["Cardiology"]["relevance"], 66.7)
+ self.assertEqual(rows["Neurology"]["available"], 2)
+ self.assertEqual(data["bank_total"], 6)
+ # Systems grouping shows top-level categories only.
+ self.assertNotIn("Kawasaki disease", rows)
+ self.assertIn("Kawasaki disease", {r["name"] for r in self.recommend(group="subtopics")["focus_areas"]})
+
+ def test_readiness_stays_locked_until_enough_answers(self):
+ self.answer(1, False)
+ data = self.recommend()
+ self.assertFalse(data["unlocked"])
+ self.assertEqual(data["answers_needed"], 39)
+ self.assertEqual(data["overall_accuracy"], 0.0) # reported, but the client gates on `unlocked`
+ cardiology = next(r for r in data["focus_areas"] if r["name"] == "Cardiology")
+ self.assertIsNone(cardiology["readiness"])
+ self.assertEqual(cardiology["accuracy"], 0.0) # raw accuracy still reported
+ self.assertEqual(cardiology["status"], "focus")
+
+ def test_readiness_shrinks_a_small_sample_toward_overall_accuracy(self):
+ # 40 answers unlocks readiness: 38 correct in neurology, 2 wrong in cardiology.
+ for _ in range(19):
+ self.answer(5, True)
+ self.answer(6, True)
+ self.answer(1, False)
+ self.answer(2, False)
+ data = self.recommend()
+ self.assertTrue(data["unlocked"])
+ rows = {row["name"]: row for row in data["focus_areas"]}
+ cardiology = rows["Cardiology"]
+ self.assertEqual(cardiology["accuracy"], 0.0)
+ # Two wrong answers alone must not read as 0% knowledge; shrinkage pulls it up
+ # toward the learner's ~95% overall, but it still ranks below neurology.
+ self.assertGreater(cardiology["readiness"], 0.0)
+ self.assertLess(cardiology["readiness"], rows["Neurology"]["readiness"])
+ self.assertGreater(cardiology["priority"], rows["Neurology"]["priority"])
+ self.assertEqual(data["focus_areas"][0]["name"], "Cardiology")
+
+ def test_untouched_categories_report_no_data_and_link_their_article(self):
+ data = self.recommend()
+ cardiology = next(r for r in data["focus_areas"] if r["name"] == "Cardiology")
+ self.assertEqual(cardiology["status"], "no_data")
+ self.assertEqual(cardiology["answered"], 0)
+ self.assertEqual(cardiology["article_id"], 7)
+ self.assertEqual(cardiology["article_title"], "Kawasaki disease")
+ self.assertFalse(cardiology["is_focus_area"]) # focus areas need recorded answers
+
+ def test_grouping_is_validated(self):
+ self.assertEqual(self.client.get("/study-tools/recommendations",
+ params={"group": "nonsense"}).status_code, 422)
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/frontend/nginx.conf b/frontend/nginx.conf
index a86f22d..218d773 100644
--- a/frontend/nginx.conf
+++ b/frontend/nginx.conf
@@ -58,8 +58,23 @@ server {
add_header X-Content-Type-Options "nosniff" always;
}
+ # Hashed build assets are immutable — a new build gets a new filename.
+ location /assets/ {
+ expires 1y;
+ add_header Cache-Control "public, immutable" always;
+ add_header X-Content-Type-Options "nosniff" always;
+ }
+
+ # index.html must never be cached, or a deploy keeps serving the old
+ # bundle references and the app appears unchanged after a release.
+ location = /index.html {
+ add_header Cache-Control "no-cache, must-revalidate" always;
+ add_header X-Content-Type-Options "nosniff" always;
+ }
+
# SPA fallback
location / {
+ add_header Cache-Control "no-cache, must-revalidate" always;
try_files $uri $uri/ /index.html;
}
}
diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx
index 2910a59..922aea8 100644
--- a/frontend/src/App.jsx
+++ b/frontend/src/App.jsx
@@ -18,6 +18,7 @@ const AccountPage = lazy(() => import('./pages/AccountPage'))
const SettingsPage = lazy(() => import('./pages/SettingsPage'))
const QuestionBankPage = lazy(() => import('./pages/QuestionBankPage'))
const QuestionManagerPage = lazy(() => import('./pages/QuestionManagerPage'))
+const AnalysisPage = lazy(() => import('./pages/AnalysisPage'))
const JobsPage = lazy(() => import('./pages/JobsPage'))
const TrashPage = lazy(() => import('./pages/TrashPage'))
const QuizEditPage = lazy(() => import('./pages/QuizEditPage'))
@@ -93,6 +94,8 @@ function AppRoutes() {
+ A grant lets one non-moderator edit, create and delete the questions in a category and everything + beneath it — nothing else. +
+ +{error}
} + + {grants === null ? ( +Loading…
+ ) : grants.length === 0 ? ( +No grants yet — only moderators can edit questions.
+ ) : ( +Built from your own answers — where you stand, and what to study next.
+{error}
} + + {loading && !data ? ( +Practises what you have not seen first, then the topics you get wrong most.
+{data.basis}
+ > + ) : ( +Review, categorise and fix gaps across the whole question bank.
++ {scope && !scope.is_moderator + ? `Your editorial grants cover ${scope.categories.length} categor${scope.categories.length === 1 ? 'y' : 'ies'}.` + : 'Review, categorise and fix gaps across the whole question bank.'} +