diff --git a/backend/app/schemas/objekt.py b/backend/app/schemas/objekt.py
new file mode 100644
index 0000000..d52babd
--- /dev/null
+++ b/backend/app/schemas/objekt.py
@@ -0,0 +1,65 @@
+import uuid
+from datetime import date
+from decimal import Decimal
+
+from pydantic import BaseModel, ConfigDict
+
+from app.models.objekt import ObjektStatus
+from app.models.objektposition import ObjektpositionStatus
+
+
+class ObjektCreate(BaseModel):
+    code: str
+    name: str
+    objekttyp_id: int
+    vorlage_id: int
+    standort_id: int
+
+
+class ObjektRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    code: str
+    name: str
+    objekttyp_id: int
+    vorlage_id: int | None
+    standort_id: int
+    status: ObjektStatus
+
+
+class ObjektDuplizierenRequest(BaseModel):
+    neuer_code: str
+    neuer_name: str
+    neuer_standort_id: int
+
+
+class ObjektpositionRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    objekt_id: int
+    material_id: int
+    sollmenge_override: Decimal | None
+    sollmenge_effektiv: Decimal
+    ist_status: ObjektpositionStatus
+    istmenge: Decimal
+    seriennummer: str | None
+    ablaufdatum: date | None
+    chargennummer: str | None
+
+
+class ObjektpositionUpdate(BaseModel):
+    sollmenge_override: Decimal | None = None
+    ist_status: ObjektpositionStatus | None = None
+    istmenge: Decimal | None = None
+    seriennummer: str | None = None
+    ablaufdatum: date | None = None
+    chargennummer: str | None = None
+
+
+class ObjektpositionCreate(BaseModel):
+    """Prompt 10 Punkt 2: zusätzliches, nicht in der Vorlage enthaltenes Material.
+    sollmenge_override ist hier Pflicht, da keine Vorlagenposition existiert, aus
+    der sich die Sollmenge dynamisch ableiten ließe."""
+
+    material_id: int
+    sollmenge_override: Decimal
diff --git a/backend/app/schemas/vorlage.py b/backend/app/schemas/vorlage.py
new file mode 100644
index 0000000..3f9c9d6
--- /dev/null
+++ b/backend/app/schemas/vorlage.py
@@ -0,0 +1,40 @@
+from decimal import Decimal
+
+from pydantic import BaseModel, ConfigDict
+
+from app.models.vorlage import VorlageStatus
+
+
+class VorlagenpositionCreate(BaseModel):
+    material_id: int
+    fach: str | None = None
+    sollmenge: Decimal
+
+
+class VorlagenpositionRead(VorlagenpositionCreate):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    vorlage_id: int
+
+
+class BeladungsvorlageCreate(BaseModel):
+    objekttyp_id: int
+    name: str
+    positionen: list[VorlagenpositionCreate] = []
+
+
+class BeladungsvorlageRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    objekttyp_id: int
+    name: str
+    version: int
+    status: VorlageStatus
+    positionen: list[VorlagenpositionRead] = []
+
+
+class VorlagenAenderung(BaseModel):
+    """Prompt 08: erzeugt eine neue Version statt die bestehende zu überschreiben.
+    `positionen` ist die VOLLSTÄNDIGE neue Positionsliste der neuen Version."""
+
+    positionen: list[VorlagenpositionCreate]
diff --git a/backend/app/services/objekte.py b/backend/app/services/objekte.py
new file mode 100644
index 0000000..acbfe40
--- /dev/null
+++ b/backend/app/services/objekte.py
@@ -0,0 +1,102 @@
+from decimal import Decimal
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.objekt import Objekt
+from app.models.objektposition import Objektposition, ObjektpositionStatus
+from app.models.vorlage import Beladungsvorlage, Vorlagenposition
+
+
+async def sollmenge_map(db: AsyncSession, vorlage_id: int | None) -> dict[int, Decimal]:
+    """material_id -> Sollmenge laut Vorlage (Prompt 10: Basis für die dynamische
+    Auflösung, wenn eine Objektposition keinen Override hat)."""
+    if vorlage_id is None:
+        return {}
+    result = await db.execute(
+        select(Vorlagenposition.material_id, Vorlagenposition.sollmenge).where(
+            Vorlagenposition.vorlage_id == vorlage_id
+        )
+    )
+    return {material_id: sollmenge for material_id, sollmenge in result.all()}
+
+
+def sollmenge_effektiv(position: Objektposition, vorlagen_sollmengen: dict[int, Decimal]) -> Decimal:
+    if position.sollmenge_override is not None:
+        return position.sollmenge_override
+    return vorlagen_sollmengen.get(position.material_id, Decimal(0))
+
+
+async def objekt_aus_vorlage(
+    db: AsyncSession,
+    *,
+    code: str,
+    name: str,
+    standort_id: int,
+    vorlage: Beladungsvorlage,
+    zustaendiger_server_id: int,
+) -> Objekt:
+    """Prompt 09 (Neuanlage, nicht Duplizieren): Objektpositionen werden aus den
+    Vorlagenpositionen initialisiert, Ist-Menge startet bei 0 ("nicht kontrolliert")."""
+    objekt = Objekt(
+        code=code,
+        name=name,
+        objekttyp_id=vorlage.objekttyp_id,
+        vorlage_id=vorlage.id,
+        standort_id=standort_id,
+        zustaendiger_server_id=zustaendiger_server_id,
+    )
+    db.add(objekt)
+    await db.flush()
+
+    result = await db.execute(select(Vorlagenposition).where(Vorlagenposition.vorlage_id == vorlage.id))
+    for vorlagenposition in result.scalars().all():
+        db.add(
+            Objektposition(
+                objekt_id=objekt.id,
+                material_id=vorlagenposition.material_id,
+                ist_status=ObjektpositionStatus.aktiv,
+                istmenge=Decimal(0),
+            )
+        )
+    await db.flush()
+    return objekt
+
+
+async def dupliziere_objekt(
+    db: AsyncSession,
+    *,
+    quelle: Objekt,
+    neuer_code: str,
+    neuer_name: str,
+    neuer_standort_id: int,
+    zustaendiger_server_id: int,
+) -> Objekt:
+    """Prompt 09: Struktur/Sollmengen (inkl. individueller Overrides) werden kopiert,
+    Zustandsdaten (Ist-Menge, SN, Ablauf, Charge) NICHT – neues Objekt startet
+    "nicht kontrolliert". Fehlbestände/Historie existieren für ein neues Objekt
+    ohnehin nicht (kein Kopierschritt nötig)."""
+    neues_objekt = Objekt(
+        code=neuer_code,
+        name=neuer_name,
+        objekttyp_id=quelle.objekttyp_id,
+        vorlage_id=quelle.vorlage_id,
+        standort_id=neuer_standort_id,
+        zustaendiger_server_id=zustaendiger_server_id,
+    )
+    db.add(neues_objekt)
+    await db.flush()
+
+    result = await db.execute(select(Objektposition).where(Objektposition.objekt_id == quelle.id))
+    for quellposition in result.scalars().all():
+        db.add(
+            Objektposition(
+                objekt_id=neues_objekt.id,
+                material_id=quellposition.material_id,
+                sollmenge_override=quellposition.sollmenge_override,
+                ist_status=quellposition.ist_status,
+                istmenge=Decimal(0),
+            )
+        )
+    await db.flush()
+    return neues_objekt
diff --git a/backend/app/services/vorlagen.py b/backend/app/services/vorlagen.py
new file mode 100644
index 0000000..ed27017
--- /dev/null
+++ b/backend/app/services/vorlagen.py
@@ -0,0 +1,46 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.vorlage import Beladungsvorlage, VorlageStatus, Vorlagenposition
+from app.schemas.vorlage import VorlagenpositionCreate
+
+
+async def erstelle_vorlage(
+    db: AsyncSession, *, objekttyp_id: int, name: str, positionen: list[VorlagenpositionCreate]
+) -> Beladungsvorlage:
+    vorlage = Beladungsvorlage(
+        objekttyp_id=objekttyp_id, name=name, version=1, status=VorlageStatus.aktiv
+    )
+    db.add(vorlage)
+    await db.flush()
+    for pos in positionen:
+        db.add(Vorlagenposition(vorlage_id=vorlage.id, **pos.model_dump()))
+    await db.flush()
+    return vorlage
+
+
+async def neue_version(
+    db: AsyncSession, *, aktuelle_vorlage: Beladungsvorlage, positionen: list[VorlagenpositionCreate]
+) -> Beladungsvorlage:
+    """Prompt 08 Punkt 4: Änderung erzeugt neue Version, alte bleibt unverändert/lesbar
+    (Historie-Integrität) statt überschrieben zu werden."""
+    aktuelle_vorlage.status = VorlageStatus.veraltet
+    await db.flush()
+
+    neue = Beladungsvorlage(
+        objekttyp_id=aktuelle_vorlage.objekttyp_id,
+        name=aktuelle_vorlage.name,
+        version=aktuelle_vorlage.version + 1,
+        status=VorlageStatus.aktiv,
+    )
+    db.add(neue)
+    await db.flush()
+    for pos in positionen:
+        db.add(Vorlagenposition(vorlage_id=neue.id, **pos.model_dump()))
+    await db.flush()
+    return neue
+
+
+async def hole_positionen(db: AsyncSession, vorlage_id: int) -> list[Vorlagenposition]:
+    result = await db.execute(select(Vorlagenposition).where(Vorlagenposition.vorlage_id == vorlage_id))
+    return list(result.scalars().all())
diff --git a/backend/pytest.ini b/backend/pytest.ini
index 86403f5..6ad80c7 100644
--- a/backend/pytest.ini
+++ b/backend/pytest.ini
@@ -1,10 +1,4 @@
 [pytest]
 asyncio_mode = auto
-# Fund im ersten echten CI-Lauf: der globale engine in conftest.py wird einmal beim Import
-# erzeugt, asyncpg-Verbindungen sind an die Event-Loop gebunden, in der sie entstanden sind.
-# Ohne session-weiten Loop-Scope erzeugt pytest-asyncio pro Test eine neue Loop ->
-# "RuntimeError: Task ... got Future ... attached to a different loop".
-asyncio_default_fixture_loop_scope = session
-asyncio_default_test_loop_scope = session
 addopts = --cov=app --cov-report=term-missing --cov-fail-under=50
 testpaths = tests
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index a3fb48c..beed7d1 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -1,7 +1,7 @@
 import pytest
 import pytest_asyncio
 from httpx import ASGITransport, AsyncClient
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
 from app.core.app_settings import settings
 from app.core.security import hash_password
@@ -11,35 +11,38 @@ from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
 from app.models.objekt import Objekt
 from app.models.stammdaten import Bereich, Objekttyp, Standort
 
-# Erwartet eine bereits per Alembic migrierte Test-Datenbank (CI: install -> migrate -> pytest,
-# siehe testphasen.md Phase 0). Jeder Test läuft in einer Transaktion, die am Ende zurückgerollt
-# wird, damit Tests sich nicht gegenseitig beeinflussen.
-engine = create_async_engine(settings.database_url, pool_pre_ping=True)
-TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
-
 
 @pytest_asyncio.fixture
 async def db_session():
     """Jeder Test läuft in einer äußeren Transaktion, die am Ende zurückgerollt wird.
 
-    Die Session ist per `join_transaction_mode="create_savepoint"` an die äußere
-    Connection gebunden: ein `commit()` innerhalb des Tests (z. B. durch `get_db`,
-    das jetzt selbst committet) schließt nur eine SAVEPOINT ab, nicht die äußere
-    Transaktion – das äußere `connection.rollback()` verwirft am Ende trotzdem alles.
-    (SQLAlchemy 2.0 "Joining a Session into an External Transaction" – Review-Fund
-    fastapi-expert: ohne dieses Pattern hätte ein commit() in Endpunkten die
-    Testisolation gebrochen.)
+    Die Engine wird bewusst PRO TEST neu erzeugt (nicht als Modul-Global): asyncpg-
+    Verbindungen sind an die Event-Loop gebunden, in der sie entstanden – ein
+    session-weit wiederverwendeter Pool kollidierte mit pytest-asyncios (Default)
+    function-scoped Loop pro Test ("attached to a different loop" / "another
+    operation is in progress", Fund im zweiten echten CI-Lauf). Mit einer frischen
+    Engine pro Test entsteht der Pool immer auf der Loop, die auch den Test ausführt.
+
+    Die Session ist zusätzlich per `join_transaction_mode="create_savepoint"` an die
+    äußere Connection gebunden: ein `commit()` innerhalb des Tests (z. B. durch
+    `get_db`, das jetzt selbst committet) schließt nur eine SAVEPOINT ab, nicht die
+    äußere Transaktion – das äußere `connection.rollback()` verwirft am Ende trotzdem
+    alles (SQLAlchemy 2.0 "Joining a Session into an External Transaction").
     """
-    async with engine.connect() as connection:
-        await connection.begin()
-        session = AsyncSession(
-            bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
-        )
-        try:
-            yield session
-        finally:
-            await session.close()
-            await connection.rollback()
+    engine = create_async_engine(settings.database_url, pool_pre_ping=True)
+    try:
+        async with engine.connect() as connection:
+            await connection.begin()
+            session = AsyncSession(
+                bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
+            )
+            try:
+                yield session
+            finally:
+                await session.close()
+                await connection.rollback()
+    finally:
+        await engine.dispose()
 
 
 @pytest_asyncio.fixture