Phase 5: Timefold timetable-solver-service + solution persistence

school-service additions:
  - tt_solution + tt_lesson migration. tt_lesson carries three UNIQUEs
    (solution+class, solution+teacher, solution+room per slot) so the
    DB itself rejects any double-booking the solver might emit by
    mistake.
  - Solution CRUD + GET solutions/:id/lessons endpoint with joined
    class/subject/teacher/room names for display.
  - POST /timetable/solutions creates the row then fires off the
    solver-service via HTTP (5s timeout, mark failed if unreachable).
  - SOLVER_SERVICE_URL config wired through main.go/handlers.

New service timetable-solver-service:
  - Python 3.11 + FastAPI + Timefold Solver 1.21 (Apache-2.0). Dockerfile
    bundles OpenJDK 17 since Timefold for Python is a JPype bridge.
  - app/domain.py — Timefold @planning_entity Lesson with timeslot+room
    as PlanningVariables; @planning_solution Timetable holds problem
    facts (rooms/teachers/etc.) AND rule-fact collections.
  - app/rules.py — frozen dataclasses mirroring 6 of the 15 tt_
    constraint_* tables initially.
  - app/constraints.py — ConstraintProvider with 3 universal hard
    constraints (no double-booking) + 5 DB-driven constraints
    (teacher_unavailable_day/window, teacher_excluded_room,
    room_unavailable, room_requires_type) + 1 quality soft constraint
    (subject_preferred_period). Remaining 9 constraint types ready to
    plug in via the same join pattern.
  - app/repository.py — async loaders for stammdaten + rules; builds
    one Lesson per (curriculum row × weekly_hours), skipping rows
    without a tt_assignment teacher.
  - app/runner.py — runs solver in ThreadPoolExecutor so the FastAPI
    event loop stays responsive. Updates tt_solution status
    pending→running→completed|infeasible|failed.
  - app/main.py — POST /api/v1/solve (202 Accepted, background task),
    GET /api/v1/jobs/{id}, /health. School-service polls tt_solution
    directly instead of GET /jobs for the typical case.
  - docker-compose.yml adds the service on port 8095, depending on
    core-health-check.

Tests:
  - school-service: validator test for CreateTimetableSolutionRequest
    (allows empty name).
  - solver-service: tests/test_domain.py + tests/test_rules.py cover
    construction + hashability of the planning facts. Full solve flow
    deferred to Phase 8 integration with seed data.

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

timetable-solver-service/app/config.py

@@ -0,0 +1,16 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    """Solver-service configuration. Values come from env vars."""
+
+    database_url: str = ""
+    solver_seconds_limit: int = 60
+    log_level: str = "INFO"
+
+    class Config:
+        env_file = ".env"
+        env_prefix = ""
+
+
+settings = Settings()

timetable-solver-service/app/constraints.py

@@ -0,0 +1,209 @@
+"""Timefold constraint provider for the school timetable.
+
+Three categories:
+  * universal hard — no double-booking class/teacher/room. These can't be
+    turned off; the school can't physically run lessons that overlap.
+  * DB-driven hard — soft-fallback if is_hard=False. Each constraint joins
+    Lesson against a rule-fact collection from the corresponding tt_
+    constraint_* table.
+  * Quality soft — preferred periods, etc.
+
+Scoring uses HardSoftScore. Hard violations are weighted by 1; soft
+violations use the rule's stored `weight` (0-100). The UI rejects any
+solution where hard_score < 0.
+"""
+
+from timefold.solver.score import (
+    constraint_provider,
+    HardSoftScore,
+    ConstraintFactory,
+    Constraint,
+    Joiners,
+)
+
+from .domain import Lesson
+from .rules import (
+    TeacherUnavailableDayRule, TeacherUnavailableWindowRule, TeacherExcludedRoomRule,
+    RoomUnavailableRule, SubjectPreferredPeriodRule, RoomRequiresTypeRule,
+)
+
+
+@constraint_provider
+def define_constraints(factory: ConstraintFactory) -> list[Constraint]:
+    return [
+        # ---------- Universal hard ----------
+        _class_conflict(factory),
+        _teacher_conflict(factory),
+        _room_conflict(factory),
+
+        # ---------- DB-driven hard or soft ----------
+        _teacher_unavailable_day(factory),
+        _teacher_unavailable_window(factory),
+        _teacher_excluded_room(factory),
+        _room_unavailable(factory),
+        _room_requires_type(factory),
+
+        # ---------- Quality soft ----------
+        _subject_preferred_period(factory),
+    ]
+
+
+# ==========================================================================
+# Universal hard constraints
+# ==========================================================================
+
+def _class_conflict(factory: ConstraintFactory) -> Constraint:
+    """A class can't sit in two lessons at once."""
+    return (
+        factory.for_each_unique_pair(
+            Lesson,
+            Joiners.equal(lambda l: l.school_class.id),
+            Joiners.equal(lambda l: l.timeslot.id if l.timeslot else None),
+        )
+        .filter(lambda l1, l2: l1.timeslot is not None and l2.timeslot is not None)
+        .penalize(HardSoftScore.ONE_HARD)
+        .as_constraint("class_conflict")
+    )
+
+
+def _teacher_conflict(factory: ConstraintFactory) -> Constraint:
+    """A teacher can't run two lessons at once."""
+    return (
+        factory.for_each_unique_pair(
+            Lesson,
+            Joiners.equal(lambda l: l.teacher.id),
+            Joiners.equal(lambda l: l.timeslot.id if l.timeslot else None),
+        )
+        .filter(lambda l1, l2: l1.timeslot is not None and l2.timeslot is not None)
+        .penalize(HardSoftScore.ONE_HARD)
+        .as_constraint("teacher_conflict")
+    )
+
+
+def _room_conflict(factory: ConstraintFactory) -> Constraint:
+    """A room can't host two lessons at once."""
+    return (
+        factory.for_each_unique_pair(
+            Lesson,
+            Joiners.equal(lambda l: l.room.id if l.room else None),
+            Joiners.equal(lambda l: l.timeslot.id if l.timeslot else None),
+        )
+        .filter(lambda l1, l2: l1.room is not None and l2.room is not None
+                and l1.timeslot is not None and l2.timeslot is not None)
+        .penalize(HardSoftScore.ONE_HARD)
+        .as_constraint("room_conflict")
+    )
+
+
+# ==========================================================================
+# DB-driven constraints
+# ==========================================================================
+
+def _score_for(rule, *, hard_per_violation: int = 1) -> HardSoftScore:
+    """Pick HardSoftScore from a rule's is_hard + weight."""
+    if rule.is_hard:
+        return HardSoftScore.of(hard_per_violation, 0)
+    return HardSoftScore.of(0, max(rule.weight, 1))
+
+
+def _teacher_unavailable_day(factory: ConstraintFactory) -> Constraint:
+    return (
+        factory.for_each(Lesson)
+        .filter(lambda l: l.timeslot is not None)
+        .join(
+            TeacherUnavailableDayRule,
+            Joiners.equal(lambda l: l.teacher.id, lambda r: r.teacher_id),
+            Joiners.equal(lambda l: l.timeslot.day_of_week, lambda r: r.day_of_week),
+        )
+        .penalize(HardSoftScore.ONE_HARD, lambda l, r: 1 if r.is_hard else 0)
+        .penalize(HardSoftScore.ONE_SOFT, lambda l, r: r.weight if not r.is_hard else 0)
+        .as_constraint("teacher_unavailable_day")
+    )
+
+
+def _teacher_unavailable_window(factory: ConstraintFactory) -> Constraint:
+    def overlaps(l: Lesson, r: TeacherUnavailableWindowRule) -> bool:
+        if l.timeslot is None:
+            return False
+        # Compare HH:MM strings — they sort correctly when zero-padded.
+        return l.timeslot.start_time < r.end_time and l.timeslot.end_time > r.start_time
+
+    return (
+        factory.for_each(Lesson)
+        .filter(lambda l: l.timeslot is not None)
+        .join(
+            TeacherUnavailableWindowRule,
+            Joiners.equal(lambda l: l.teacher.id, lambda r: r.teacher_id),
+            Joiners.equal(lambda l: l.timeslot.day_of_week, lambda r: r.day_of_week),
+        )
+        .filter(overlaps)
+        .penalize(HardSoftScore.ONE_HARD, lambda l, r: 1 if r.is_hard else 0)
+        .penalize(HardSoftScore.ONE_SOFT, lambda l, r: r.weight if not r.is_hard else 0)
+        .as_constraint("teacher_unavailable_window")
+    )
+
+
+def _teacher_excluded_room(factory: ConstraintFactory) -> Constraint:
+    return (
+        factory.for_each(Lesson)
+        .filter(lambda l: l.room is not None)
+        .join(
+            TeacherExcludedRoomRule,
+            Joiners.equal(lambda l: l.teacher.id, lambda r: r.teacher_id),
+            Joiners.equal(lambda l: l.room.id, lambda r: r.room_id),
+        )
+        .penalize(HardSoftScore.ONE_HARD, lambda l, r: 1 if r.is_hard else 0)
+        .penalize(HardSoftScore.ONE_SOFT, lambda l, r: r.weight if not r.is_hard else 0)
+        .as_constraint("teacher_excluded_room")
+    )
+
+
+def _room_unavailable(factory: ConstraintFactory) -> Constraint:
+    return (
+        factory.for_each(Lesson)
+        .filter(lambda l: l.room is not None and l.timeslot is not None)
+        .join(
+            RoomUnavailableRule,
+            Joiners.equal(lambda l: l.room.id, lambda r: r.room_id),
+            Joiners.equal(lambda l: l.timeslot.day_of_week, lambda r: r.day_of_week),
+            Joiners.equal(lambda l: l.timeslot.period_index, lambda r: r.period_index),
+        )
+        .penalize(HardSoftScore.ONE_HARD, lambda l, r: 1 if r.is_hard else 0)
+        .penalize(HardSoftScore.ONE_SOFT, lambda l, r: r.weight if not r.is_hard else 0)
+        .as_constraint("room_unavailable")
+    )
+
+
+def _room_requires_type(factory: ConstraintFactory) -> Constraint:
+    """If a subject requires a specific room type, the assigned room must match."""
+    return (
+        factory.for_each(Lesson)
+        .filter(lambda l: l.room is not None)
+        .join(
+            RoomRequiresTypeRule,
+            Joiners.equal(lambda l: l.subject.id, lambda r: r.subject_id),
+        )
+        .filter(lambda l, r: l.room.room_type != r.room_type)
+        .penalize(HardSoftScore.ONE_HARD, lambda l, r: 1 if r.is_hard else 0)
+        .penalize(HardSoftScore.ONE_SOFT, lambda l, r: r.weight if not r.is_hard else 0)
+        .as_constraint("room_requires_type")
+    )
+
+
+# ==========================================================================
+# Quality soft
+# ==========================================================================
+
+def _subject_preferred_period(factory: ConstraintFactory) -> Constraint:
+    """Soft penalty when a lesson lands outside the subject's preferred period range."""
+    return (
+        factory.for_each(Lesson)
+        .filter(lambda l: l.timeslot is not None)
+        .join(
+            SubjectPreferredPeriodRule,
+            Joiners.equal(lambda l: l.subject.id, lambda r: r.subject_id),
+        )
+        .filter(lambda l, r: not (r.period_from <= l.timeslot.period_index <= r.period_to))
+        .penalize(HardSoftScore.ONE_SOFT, lambda l, r: r.weight)
+        .as_constraint("subject_preferred_period")
+    )

timetable-solver-service/app/db.py

@@ -0,0 +1,28 @@
+import asyncpg
+from typing import Optional
+
+from .config import settings
+
+_pool: Optional[asyncpg.Pool] = None
+
+
+async def get_pool() -> asyncpg.Pool:
+    """Lazy-init the asyncpg pool. Reused across requests + background jobs."""
+    global _pool
+    if _pool is None:
+        if not settings.database_url:
+            raise RuntimeError("DATABASE_URL not configured")
+        _pool = await asyncpg.create_pool(
+            dsn=settings.database_url,
+            min_size=2,
+            max_size=10,
+            command_timeout=30,
+        )
+    return _pool
+
+
+async def close_pool() -> None:
+    global _pool
+    if _pool is not None:
+        await _pool.close()
+        _pool = None

timetable-solver-service/app/domain.py

@@ -0,0 +1,134 @@
+"""Timefold planning domain for school timetables.
+
+Lessons are the planning entities; their `timeslot` and `room` are the
+variables the solver picks. Class/subject/teacher come from the assignment
+(`tt_assignment`) and stay fixed for a given Lesson instance.
+
+Note on equality: Timefold compares facts by identity by default, so we
+use frozen dataclasses with id-based equality where needed.
+"""
+
+from dataclasses import dataclass, field
+from typing import Annotated, Optional
+
+from timefold.solver.domain import (
+    planning_entity,
+    planning_solution,
+    PlanningVariable,
+    PlanningId,
+    PlanningEntityCollectionProperty,
+    ProblemFactCollectionProperty,
+    ValueRangeProvider,
+    PlanningScore,
+)
+from timefold.solver.score import HardSoftScore
+
+
+@dataclass(frozen=True)
+class Timeslot:
+    """A single weekday + lesson period (e.g. Monday 1st hour, 08:00–08:45)."""
+
+    id: Annotated[str, PlanningId]
+    day_of_week: int  # 1..7
+    period_index: int  # 1..N
+    start_time: str  # HH:MM
+    end_time: str  # HH:MM
+
+    def __str__(self) -> str:
+        return f"D{self.day_of_week}P{self.period_index}"
+
+
+@dataclass(frozen=True)
+class Room:
+    id: Annotated[str, PlanningId]
+    name: str
+    room_type: str = ""
+
+    def __str__(self) -> str:
+        return self.name
+
+
+@dataclass(frozen=True)
+class Teacher:
+    id: Annotated[str, PlanningId]
+    last_name: str
+    first_name: str
+    short_code: str
+
+    def __str__(self) -> str:
+        return f"{self.last_name}, {self.first_name}"
+
+
+@dataclass(frozen=True)
+class SchoolClass:
+    id: Annotated[str, PlanningId]
+    name: str
+    grade_level: int
+
+    def __str__(self) -> str:
+        return self.name
+
+
+@dataclass(frozen=True)
+class Subject:
+    id: Annotated[str, PlanningId]
+    name: str
+    short_code: str
+    required_room_type: str = ""
+
+    def __str__(self) -> str:
+        return self.short_code
+
+
+@planning_entity
+@dataclass
+class Lesson:
+    """One scheduled class-subject pairing.
+
+    Curriculum says "5a needs 4 hours of Mathe per week" → 4 Lesson instances
+    with school_class=5a, subject=Mathe, teacher fixed (from tt_assignment).
+    The solver assigns timeslot + room.
+    """
+
+    id: Annotated[str, PlanningId]
+    school_class: SchoolClass
+    subject: Subject
+    teacher: Teacher
+    timeslot: Annotated[Optional[Timeslot], PlanningVariable] = field(default=None)
+    room: Annotated[Optional[Room], PlanningVariable] = field(default=None)
+
+    def __str__(self) -> str:
+        return f"{self.school_class}-{self.subject}#{self.id[:8]}"
+
+
+from .rules import (
+    TeacherUnavailableDayRule, TeacherUnavailableWindowRule, TeacherExcludedRoomRule,
+    RoomUnavailableRule, SubjectPreferredPeriodRule, RoomRequiresTypeRule,
+)
+
+
+@planning_solution
+@dataclass
+class Timetable:
+    """The solver works on one Timetable instance: shuffles `lessons[*].timeslot`
+    and `lessons[*].room` to satisfy the constraints.
+
+    Constraint-rule facts are pulled from the DB at solve time and passed
+    here so the constraint provider can join against them.
+    """
+
+    timeslots: Annotated[list[Timeslot], ProblemFactCollectionProperty, ValueRangeProvider]
+    rooms: Annotated[list[Room], ProblemFactCollectionProperty, ValueRangeProvider]
+    teachers: Annotated[list[Teacher], ProblemFactCollectionProperty]
+    classes: Annotated[list[SchoolClass], ProblemFactCollectionProperty]
+    subjects: Annotated[list[Subject], ProblemFactCollectionProperty]
+
+    teacher_unavailable_days: Annotated[list[TeacherUnavailableDayRule], ProblemFactCollectionProperty]
+    teacher_unavailable_windows: Annotated[list[TeacherUnavailableWindowRule], ProblemFactCollectionProperty]
+    teacher_excluded_rooms: Annotated[list[TeacherExcludedRoomRule], ProblemFactCollectionProperty]
+    room_unavailables: Annotated[list[RoomUnavailableRule], ProblemFactCollectionProperty]
+    subject_preferred_periods: Annotated[list[SubjectPreferredPeriodRule], ProblemFactCollectionProperty]
+    room_requires_types: Annotated[list[RoomRequiresTypeRule], ProblemFactCollectionProperty]
+
+    lessons: Annotated[list[Lesson], PlanningEntityCollectionProperty]
+    score: Annotated[Optional[HardSoftScore], PlanningScore] = field(default=None)

timetable-solver-service/app/main.py

@@ -0,0 +1,96 @@
+"""Timetable solver service — FastAPI entrypoint.
+
+POST /api/v1/solve   schedules a solve job (BackgroundTasks). Returns 202.
+GET  /api/v1/jobs/{solution_id}  reads back tt_solution status from DB.
+GET  /health         liveness probe for Docker.
+
+The actual solver call lives in runner.py and runs in a worker thread, so
+this process can accept multiple concurrent solves without blocking.
+"""
+
+import logging
+import os
+
+from fastapi import BackgroundTasks, FastAPI, HTTPException, status
+from pydantic import BaseModel
+
+from .config import settings
+from .db import close_pool, get_pool
+from .runner import run_solve
+
+logging.basicConfig(level=os.getenv("LOG_LEVEL", settings.log_level))
+logger = logging.getLogger(__name__)
+
+app = FastAPI(title="BreakPilot Timetable Solver", version="0.1.0")
+
+
+class SolveRequest(BaseModel):
+    solution_id: str
+    created_by_user_id: str
+
+
+class SolveResponse(BaseModel):
+    solution_id: str
+    status: str
+    message: str
+
+
+class JobStatus(BaseModel):
+    solution_id: str
+    status: str
+    hard_score: int | None = None
+    soft_score: int | None = None
+    error_message: str | None = None
+
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "healthy", "service": "timetable-solver"}
+
+
+@app.post("/api/v1/solve", response_model=SolveResponse, status_code=status.HTTP_202_ACCEPTED)
+async def solve(req: SolveRequest, bg: BackgroundTasks) -> SolveResponse:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "SELECT status FROM tt_solution WHERE id = $1 AND created_by_user_id = $2",
+            req.solution_id, req.created_by_user_id,
+        )
+    if row is None:
+        raise HTTPException(status_code=404, detail="Solution not found")
+    if row["status"] in ("running", "completed"):
+        return SolveResponse(
+            solution_id=req.solution_id, status=row["status"],
+            message="already in progress or finished",
+        )
+
+    bg.add_task(run_solve, req.solution_id, req.created_by_user_id)
+    logger.info("Solve queued for %s (user %s)", req.solution_id, req.created_by_user_id)
+    return SolveResponse(
+        solution_id=req.solution_id, status="queued",
+        message="job accepted, poll tt_solution for progress",
+    )
+
+
+@app.get("/api/v1/jobs/{solution_id}", response_model=JobStatus)
+async def job_status(solution_id: str) -> JobStatus:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow("""
+            SELECT id::text, status, hard_score, soft_score, COALESCE(error_message, '') AS err
+            FROM tt_solution WHERE id = $1
+        """, solution_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="Solution not found")
+    return JobStatus(
+        solution_id=row["id"],
+        status=row["status"],
+        hard_score=row["hard_score"],
+        soft_score=row["soft_score"],
+        error_message=row["err"] or None,
+    )
+
+
+@app.on_event("shutdown")
+async def _on_shutdown() -> None:
+    await close_pool()

timetable-solver-service/app/repository.py

@@ -0,0 +1,258 @@
+"""Read stammdaten + constraints from PostgreSQL and turn them into Timefold
+domain objects. Used by runner.py to build a Timetable problem instance.
+
+Ownership is enforced via created_by_user_id everywhere — the solver only
+sees data belonging to the Rektor who triggered the solve.
+"""
+
+from typing import Any
+
+import asyncpg
+
+from .domain import Lesson, Room, SchoolClass, Subject, Teacher, Timeslot, Timetable
+from .rules import (
+    RoomRequiresTypeRule, RoomUnavailableRule,
+    SubjectPreferredPeriodRule, TeacherExcludedRoomRule,
+    TeacherUnavailableDayRule, TeacherUnavailableWindowRule,
+)
+
+
+async def build_problem(pool: asyncpg.Pool, user_id: str) -> Timetable:
+    async with pool.acquire() as conn:
+        timeslots = await _load_timeslots(conn, user_id)
+        rooms = await _load_rooms(conn, user_id)
+        teachers = await _load_teachers(conn, user_id)
+        classes = await _load_classes(conn, user_id)
+        subjects = await _load_subjects(conn, user_id)
+        lessons = await _build_lessons(conn, user_id, classes, subjects, teachers)
+        rules = await _load_rules(conn, user_id)
+
+    return Timetable(
+        timeslots=timeslots,
+        rooms=rooms,
+        teachers=teachers,
+        classes=classes,
+        subjects=subjects,
+        lessons=lessons,
+        score=None,
+        **rules,
+    )
+
+
+async def _load_timeslots(conn: asyncpg.Connection, user_id: str) -> list[Timeslot]:
+    rows = await conn.fetch("""
+        SELECT id::text, day_of_week, period_index,
+               to_char(start_time, 'HH24:MI') AS st,
+               to_char(end_time, 'HH24:MI') AS et,
+               is_break
+        FROM tt_period
+        WHERE created_by_user_id = $1 AND is_break = false
+        ORDER BY day_of_week, period_index
+    """, user_id)
+    return [
+        Timeslot(id=r["id"], day_of_week=r["day_of_week"], period_index=r["period_index"],
+                 start_time=r["st"], end_time=r["et"])
+        for r in rows
+    ]
+
+
+async def _load_rooms(conn: asyncpg.Connection, user_id: str) -> list[Room]:
+    rows = await conn.fetch("""
+        SELECT id::text, name, COALESCE(room_type, '') AS rt
+        FROM tt_room WHERE created_by_user_id = $1 ORDER BY name
+    """, user_id)
+    return [Room(id=r["id"], name=r["name"], room_type=r["rt"]) for r in rows]
+
+
+async def _load_teachers(conn: asyncpg.Connection, user_id: str) -> list[Teacher]:
+    rows = await conn.fetch("""
+        SELECT id::text, first_name, last_name, short_code
+        FROM tt_teacher WHERE created_by_user_id = $1 ORDER BY last_name, first_name
+    """, user_id)
+    return [Teacher(id=r["id"], first_name=r["first_name"], last_name=r["last_name"], short_code=r["short_code"]) for r in rows]
+
+
+async def _load_classes(conn: asyncpg.Connection, user_id: str) -> list[SchoolClass]:
+    rows = await conn.fetch("""
+        SELECT id::text, name, grade_level
+        FROM tt_class WHERE created_by_user_id = $1 ORDER BY grade_level, name
+    """, user_id)
+    return [SchoolClass(id=r["id"], name=r["name"], grade_level=r["grade_level"]) for r in rows]
+
+
+async def _load_subjects(conn: asyncpg.Connection, user_id: str) -> list[Subject]:
+    rows = await conn.fetch("""
+        SELECT id::text, name, short_code, COALESCE(required_room_type, '') AS rt
+        FROM tt_subject WHERE created_by_user_id = $1 ORDER BY name
+    """, user_id)
+    return [Subject(id=r["id"], name=r["name"], short_code=r["short_code"], required_room_type=r["rt"]) for r in rows]
+
+
+async def _build_lessons(
+    conn: asyncpg.Connection,
+    user_id: str,
+    classes: list[SchoolClass],
+    subjects: list[Subject],
+    teachers: list[Teacher],
+) -> list[Lesson]:
+    """Materialise curriculum × assignment into Lesson instances.
+
+    For each (class, subject) row in tt_curriculum with weekly_hours=N, we
+    create N Lesson rows. The teacher is the one assigned in tt_assignment
+    for the same (class, subject) — there must be exactly one, else the
+    lesson can't be scheduled and is skipped (the UI surfaces this gap).
+    """
+    rows = await conn.fetch("""
+        SELECT cu.class_id::text, cu.subject_id::text, cu.weekly_hours,
+               a.teacher_id::text
+        FROM tt_curriculum cu
+        JOIN tt_class cl ON cu.class_id = cl.id
+        LEFT JOIN tt_assignment a
+          ON a.class_id = cu.class_id AND a.subject_id = cu.subject_id
+        WHERE cl.created_by_user_id = $1
+    """, user_id)
+
+    class_by_id = {c.id: c for c in classes}
+    subject_by_id = {s.id: s for s in subjects}
+    teacher_by_id = {t.id: t for t in teachers}
+
+    lessons: list[Lesson] = []
+    counter = 0
+    for r in rows:
+        cls = class_by_id.get(r["class_id"])
+        sub = subject_by_id.get(r["subject_id"])
+        tch = teacher_by_id.get(r["teacher_id"]) if r["teacher_id"] else None
+        if cls is None or sub is None or tch is None:
+            # Missing assignment — solver can't schedule without a teacher.
+            continue
+        for _ in range(int(r["weekly_hours"])):
+            lessons.append(Lesson(
+                id=f"L{counter}-{cls.id[:6]}-{sub.id[:6]}",
+                school_class=cls,
+                subject=sub,
+                teacher=tch,
+            ))
+            counter += 1
+    return lessons
+
+
+async def _load_rules(conn: asyncpg.Connection, user_id: str) -> dict[str, list[Any]]:
+    """Pull the subset of constraint tables the constraint provider uses."""
+    rules: dict[str, list[Any]] = {}
+
+    rows = await conn.fetch("""
+        SELECT teacher_id::text, day_of_week, is_hard, weight
+        FROM tt_constraint_teacher_unavailable_day
+        WHERE created_by_user_id = $1 AND active = true
+    """, user_id)
+    rules["teacher_unavailable_days"] = [TeacherUnavailableDayRule(**dict(r)) for r in rows]
+
+    rows = await conn.fetch("""
+        SELECT teacher_id::text, day_of_week,
+               to_char(start_time, 'HH24:MI') AS start_time,
+               to_char(end_time, 'HH24:MI') AS end_time,
+               is_hard, weight
+        FROM tt_constraint_teacher_unavailable_window
+        WHERE created_by_user_id = $1 AND active = true
+    """, user_id)
+    rules["teacher_unavailable_windows"] = [TeacherUnavailableWindowRule(**dict(r)) for r in rows]
+
+    rows = await conn.fetch("""
+        SELECT teacher_id::text, room_id::text, is_hard, weight
+        FROM tt_constraint_teacher_excluded_room
+        WHERE created_by_user_id = $1 AND active = true
+    """, user_id)
+    rules["teacher_excluded_rooms"] = [TeacherExcludedRoomRule(**dict(r)) for r in rows]
+
+    rows = await conn.fetch("""
+        SELECT room_id::text, day_of_week, period_index, is_hard, weight
+        FROM tt_constraint_room_unavailable
+        WHERE created_by_user_id = $1 AND active = true
+    """, user_id)
+    rules["room_unavailables"] = [RoomUnavailableRule(**dict(r)) for r in rows]
+
+    rows = await conn.fetch("""
+        SELECT subject_id::text, period_from, period_to, is_hard, weight
+        FROM tt_constraint_subject_preferred_period
+        WHERE created_by_user_id = $1 AND active = true
+    """, user_id)
+    rules["subject_preferred_periods"] = [SubjectPreferredPeriodRule(**dict(r)) for r in rows]
+
+    rows = await conn.fetch("""
+        SELECT subject_id::text, room_type, is_hard, weight
+        FROM tt_constraint_room_requires_type
+        WHERE created_by_user_id = $1 AND active = true
+    """, user_id)
+    rules["room_requires_types"] = [RoomRequiresTypeRule(**dict(r)) for r in rows]
+
+    return rules
+
+
+async def persist_solution(
+    pool: asyncpg.Pool,
+    solution_id: str,
+    timetable: Timetable,
+    hard_score: int,
+    soft_score: int,
+) -> None:
+    """Write the solver result back to tt_solution + tt_lesson."""
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            await conn.execute("""
+                UPDATE tt_solution
+                SET status = 'completed',
+                    hard_score = $2,
+                    soft_score = $3,
+                    finished_at = NOW()
+                WHERE id = $1
+            """, solution_id, hard_score, soft_score)
+
+            # Clear any prior lesson rows (re-solves overwrite).
+            await conn.execute("DELETE FROM tt_lesson WHERE solution_id = $1", solution_id)
+
+            for lesson in timetable.lessons:
+                if lesson.timeslot is None:
+                    continue
+                await conn.execute("""
+                    INSERT INTO tt_lesson
+                        (solution_id, class_id, subject_id, teacher_id, room_id,
+                         day_of_week, period_index, pinned)
+                    VALUES ($1, $2::uuid, $3::uuid, $4::uuid, $5::uuid, $6, $7, false)
+                """,
+                    solution_id,
+                    lesson.school_class.id,
+                    lesson.subject.id,
+                    lesson.teacher.id,
+                    lesson.room.id if lesson.room else None,
+                    lesson.timeslot.day_of_week,
+                    lesson.timeslot.period_index,
+                )
+
+
+async def mark_failed(pool: asyncpg.Pool, solution_id: str, error_message: str) -> None:
+    async with pool.acquire() as conn:
+        await conn.execute("""
+            UPDATE tt_solution
+            SET status = 'failed', error_message = $2, finished_at = NOW()
+            WHERE id = $1
+        """, solution_id, error_message)
+
+
+async def mark_running(pool: asyncpg.Pool, solution_id: str) -> None:
+    async with pool.acquire() as conn:
+        await conn.execute("""
+            UPDATE tt_solution SET status = 'running', started_at = NOW()
+            WHERE id = $1
+        """, solution_id)
+
+
+async def mark_infeasible(pool: asyncpg.Pool, solution_id: str, hard_score: int, soft_score: int) -> None:
+    async with pool.acquire() as conn:
+        await conn.execute("""
+            UPDATE tt_solution
+            SET status = 'infeasible',
+                hard_score = $2,
+                soft_score = $3,
+                finished_at = NOW()
+            WHERE id = $1
+        """, solution_id, hard_score, soft_score)

timetable-solver-service/app/rules.py

@@ -0,0 +1,64 @@
+"""DB-driven constraint rules as Timefold problem facts.
+
+Each tt_constraint_* table from school-service maps to one dataclass here.
+Rows loaded at solve time are passed in via Timetable.* fact collections
+(see domain.py for wiring) and queried by the constraint provider.
+
+Only the rule types actually wired into constraints.py are defined for now.
+Adding a new one is two steps: define the dataclass, add it to Timetable's
+problem-fact properties, then implement a constraint that joins it.
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class TeacherUnavailableDayRule:
+    teacher_id: str
+    day_of_week: int
+    is_hard: bool
+    weight: int
+
+
+@dataclass(frozen=True)
+class TeacherUnavailableWindowRule:
+    teacher_id: str
+    day_of_week: int
+    start_time: str  # HH:MM
+    end_time: str  # HH:MM
+    is_hard: bool
+    weight: int
+
+
+@dataclass(frozen=True)
+class TeacherExcludedRoomRule:
+    teacher_id: str
+    room_id: str
+    is_hard: bool
+    weight: int
+
+
+@dataclass(frozen=True)
+class RoomUnavailableRule:
+    room_id: str
+    day_of_week: int
+    period_index: int
+    is_hard: bool
+    weight: int
+
+
+@dataclass(frozen=True)
+class SubjectPreferredPeriodRule:
+    subject_id: str
+    period_from: int
+    period_to: int
+    is_hard: bool
+    weight: int
+
+
+@dataclass(frozen=True)
+class RoomRequiresTypeRule:
+    subject_id: str
+    room_type: str
+    is_hard: bool
+    weight: int

timetable-solver-service/app/runner.py

@@ -0,0 +1,94 @@
+"""Solver job runner. One async entry point per solve.
+
+Lifecycle:
+  1. mark_running -> tt_solution.status = 'running'
+  2. build_problem -> Timetable from DB
+  3. SolverFactory.buildSolver() -> Timefold solver
+  4. solver.solve(problem) -> completed Timetable
+  5. persist_solution or mark_infeasible based on hard_score
+  Errors at any step → mark_failed.
+
+Long solves are CPU-bound. We run the solver in an executor so the FastAPI
+event loop stays responsive for other requests.
+"""
+
+import asyncio
+import logging
+import traceback
+from concurrent.futures import ThreadPoolExecutor
+
+from timefold.solver import SolverFactory
+from timefold.solver.config import (
+    SolverConfig,
+    TerminationConfig,
+    Duration,
+)
+
+from .config import settings
+from .constraints import define_constraints
+from .db import get_pool
+from .domain import Lesson, Timetable
+from .repository import build_problem, mark_failed, mark_infeasible, mark_running, persist_solution
+
+logger = logging.getLogger(__name__)
+
+_executor = ThreadPoolExecutor(max_workers=2)
+
+_solver_factory = SolverFactory.create(
+    SolverConfig(
+        solution_class=Timetable,
+        entity_class_list=[Lesson],
+        score_director_factory_config={"constraint_provider_function": define_constraints},
+        termination_config=TerminationConfig(
+            spent_limit=Duration(seconds=settings.solver_seconds_limit),
+        ),
+    )
+)
+
+
+def _solve_sync(problem: Timetable) -> Timetable:
+    """Blocking solver call; runs in a worker thread."""
+    solver = _solver_factory.build_solver()
+    return solver.solve(problem)
+
+
+async def run_solve(solution_id: str, user_id: str) -> None:
+    """Top-level async entry. Caller fires-and-forgets via BackgroundTasks."""
+    pool = await get_pool()
+    try:
+        await mark_running(pool, solution_id)
+        problem = await build_problem(pool, user_id)
+
+        if not problem.lessons:
+            await mark_failed(pool, solution_id,
+                              "Keine Lessons — pruefe Stundentafel + Lehrauftraege.")
+            return
+        if not problem.timeslots:
+            await mark_failed(pool, solution_id,
+                              "Kein Zeitraster definiert.")
+            return
+        if not problem.rooms:
+            await mark_failed(pool, solution_id,
+                              "Keine Raeume definiert.")
+            return
+
+        loop = asyncio.get_running_loop()
+        solved: Timetable = await loop.run_in_executor(_executor, _solve_sync, problem)
+
+        score = solved.score
+        hard = score.hard_score() if score else 0
+        soft = score.soft_score() if score else 0
+
+        if hard < 0:
+            await mark_infeasible(pool, solution_id, hard, soft)
+            logger.info("Solution %s infeasible: hard=%d soft=%d", solution_id, hard, soft)
+        else:
+            await persist_solution(pool, solution_id, solved, hard, soft)
+            logger.info("Solution %s completed: hard=%d soft=%d", solution_id, hard, soft)
+
+    except Exception as exc:
+        logger.exception("Solver failed for %s", solution_id)
+        try:
+            await mark_failed(pool, solution_id, f"{exc.__class__.__name__}: {exc}\n{traceback.format_exc()[:1000]}")
+        except Exception:
+            logger.exception("Failed to even mark solution as failed")