diff --git a/api/routes/admin_kb.py b/api/routes/admin_kb.py
index 2eae4e2..99a186e 100644
--- a/api/routes/admin_kb.py
+++ b/api/routes/admin_kb.py
@@ -387,3 +387,65 @@ async def reingest_admin_source(request: Request, source_id: int):
 
     asyncio.create_task(kb_admin_sources.process_reingest_job(job_id))
     return {"job_id": job_id, "status": "queued", "source_id": source_id}
+
+
+# ── Phase 4: topic CRUD (Task #15) ─────────────────────────────────────────
+
+class TopicCreate(BaseModel):
+    slug: str = Field(..., min_length=2, max_length=63)
+    name: str = Field(..., min_length=1, max_length=200)
+    description: str | None = Field(None, max_length=1000)
+    system_prompt_addendum: str | None = Field(None, max_length=4000)
+    is_active: bool = True
+
+
+class TopicUpdate(BaseModel):
+    name: str | None = Field(None, max_length=200)
+    description: str | None = Field(None, max_length=1000)
+    system_prompt_addendum: str | None = Field(None, max_length=4000)
+    is_active: bool | None = None
+
+
+@router.get("/topics")
+async def list_admin_topics(request: Request):
+    _verify_admin(request)
+    items = await kb_topics.list_topics_admin()
+    return {"items": items, "count": len(items)}
+
+
+@router.post("/topics")
+async def create_admin_topic(request: Request, body: TopicCreate):
+    _verify_admin(request)
+    try:
+        return await kb_topics.create_topic(
+            slug=body.slug,
+            name=body.name,
+            description=body.description,
+            system_prompt_addendum=body.system_prompt_addendum,
+            is_active=body.is_active,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.put("/topics/{topic_id}")
+async def update_admin_topic(request: Request, topic_id: int, body: TopicUpdate):
+    _verify_admin(request)
+    fields = body.model_dump(exclude_unset=True)
+    try:
+        return await kb_topics.update_topic(topic_id, fields)
+    except LookupError:
+        raise HTTPException(status_code=404, detail="Topic not found")
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.delete("/topics/{topic_id}")
+async def delete_admin_topic(request: Request, topic_id: int):
+    _verify_admin(request)
+    try:
+        return await kb_topics.delete_topic(topic_id)
+    except LookupError:
+        raise HTTPException(status_code=404, detail="Topic not found")
+    except kb_topics.ConflictError as e:
+        raise HTTPException(status_code=409, detail=str(e))
diff --git a/api/services/kb/topics.py b/api/services/kb/topics.py
index dd88c0a..13d95ce 100644
--- a/api/services/kb/topics.py
+++ b/api/services/kb/topics.py
@@ -3,17 +3,24 @@ load a topic's full row (including system_prompt_addendum) for /kb/ask.
 
 The DB schema is set up by scripts/migrations/001_topics.sql. As of Phase
 1 there is exactly one seeded topic (id=1, slug='national-insurance').
+Phase 4 adds admin CRUD on top so firms can add their own legal domains
+(דיני עבודה, דין פלילי, נדל\"ן) without anyone running SQL.
 """
 
 from __future__ import annotations
 
 import logging
+import re
 from typing import Optional
 
 from api.services.kb.db import get_pool
 
 logger = logging.getLogger("shira.kb.topics")
 
+# Slug shape — lowercase ASCII, digits, hyphens. Mirrors what reads cleanly
+# in URLs and in the S3 layout (inbox/<topic_slug>/<kind>/...).
+_SLUG_RE = re.compile(r"^[a-z][a-z0-9-]{1,62}$")
+
 
 async def list_topics(active_only: bool = True) -> list[dict]:
     """Return topics ordered by id. Public payload — never includes the
@@ -49,6 +56,162 @@ async def get_topic(topic_id: int) -> Optional[dict]:
     return dict(row) if row else None
 
 
+# ── Phase 4: admin CRUD (Task #15) ─────────────────────────────────────────
+
+# Whitelist of fields editable from PUT /admin/kb/topics/{id}.
+# slug is intentionally NOT here — changing it would break the S3 layout
+# (existing files live under inbox/<old_slug>/...). If a slug rename ever
+# becomes necessary it should be a deliberate migration, not a metadata edit.
+_TOPIC_EDITABLE = ("name", "description", "system_prompt_addendum", "is_active")
+
+
+async def list_topics_admin() -> list[dict]:
+    """List ALL topics (including is_active=false), with source counts.
+
+    The user-facing /kb/topics endpoint hides inactive ones; this one is
+    for the admin panel where a soft-deleted topic is still meaningful.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            """
+            SELECT
+                t.id, t.slug, t.name, t.description,
+                t.system_prompt_addendum, t.is_active,
+                t.created_at, t.updated_at,
+                (SELECT COUNT(*) FROM kb_source s
+                  WHERE s.topic_id = t.id AND s.superseded_by IS NULL) AS source_count
+            FROM kb_topic t
+            ORDER BY t.is_active DESC, t.id
+            """
+        )
+    return [dict(r) for r in rows]
+
+
+async def create_topic(
+    *,
+    slug: str,
+    name: str,
+    description: Optional[str],
+    system_prompt_addendum: Optional[str],
+    is_active: bool = True,
+) -> dict:
+    """Create a new topic. Validates slug shape and uniqueness."""
+    slug = (slug or "").strip().lower()
+    name = (name or "").strip()
+    if not _SLUG_RE.match(slug):
+        raise ValueError(
+            "slug must be lowercase ASCII (letters, digits, hyphens), "
+            "start with a letter, 2–63 chars"
+        )
+    if not name:
+        raise ValueError("name is required")
+
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        # Uniqueness check first so we return a clean 400 instead of leaking
+        # the asyncpg UniqueViolationError.
+        exists = await conn.fetchval(
+            "SELECT 1 FROM kb_topic WHERE slug = $1", slug
+        )
+        if exists:
+            raise ValueError(f"slug already exists: {slug}")
+        row = await conn.fetchrow(
+            """
+            INSERT INTO kb_topic
+                (slug, name, description, system_prompt_addendum, is_active)
+            VALUES ($1, $2, $3, $4, $5)
+            RETURNING id, slug, name, description, system_prompt_addendum,
+                      is_active, created_at, updated_at
+            """,
+            slug,
+            name,
+            (description or None),
+            (system_prompt_addendum or None),
+            bool(is_active),
+        )
+    logger.info("[topics.admin] created id=%s slug=%s", row["id"], row["slug"])
+    return dict(row)
+
+
+async def update_topic(topic_id: int, fields: dict) -> dict:
+    """UPDATE only whitelisted columns. Returns the post-update row."""
+    clean: dict = {}
+    for k in _TOPIC_EDITABLE:
+        if k not in fields:
+            continue
+        v = fields[k]
+        if k == "is_active":
+            clean[k] = bool(v) if v is not None else None
+        elif v is None or v == "":
+            clean[k] = None
+        else:
+            clean[k] = str(v).strip()
+    if "name" in clean and not clean["name"]:
+        raise ValueError("name cannot be empty")
+
+    pool = await get_pool()
+    if not clean:
+        async with pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT * FROM kb_topic WHERE id = $1", topic_id
+            )
+        if not row:
+            raise LookupError(f"topic {topic_id} not found")
+        return dict(row)
+
+    set_pieces = []
+    args: list = [topic_id]
+    for k, v in clean.items():
+        args.append(v)
+        set_pieces.append(f"{k} = ${len(args)}")
+    args.append("now()")  # placeholder; we use literal now() below
+    sql = f"""
+        UPDATE kb_topic
+           SET {', '.join(set_pieces)}, updated_at = now()
+         WHERE id = $1
+        RETURNING id, slug, name, description, system_prompt_addendum,
+                  is_active, created_at, updated_at
+    """
+    args.pop()  # drop the unused now() placeholder
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(sql, *args)
+    if not row:
+        raise LookupError(f"topic {topic_id} not found")
+    logger.info("[topics.admin] updated id=%s fields=%s", topic_id, list(clean.keys()))
+    return dict(row)
+
+
+async def delete_topic(topic_id: int) -> dict:
+    """Hard-delete a topic. Refuses if any sources still belong to it.
+
+    Soft-delete (is_active=false) is the normal path — that's what the UI
+    will use for routine 'remove from dropdown' operations. This DELETE is
+    only for cleaning up an empty topic created by mistake.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        cnt = await conn.fetchval(
+            "SELECT COUNT(*) FROM kb_source WHERE topic_id = $1", topic_id
+        )
+        if cnt and cnt > 0:
+            raise ConflictError(
+                f"topic still has {cnt} sources; soft-delete (is_active=false) instead"
+            )
+        result = await conn.execute(
+            "DELETE FROM kb_topic WHERE id = $1", topic_id
+        )
+    if not result.endswith(" 1"):
+        raise LookupError(f"topic {topic_id} not found")
+    logger.info("[topics.admin] deleted id=%s", topic_id)
+    return {"deleted": True, "topic_id": topic_id}
+
+
+class ConflictError(Exception):
+    """Raised by delete_topic when sources prevent a hard-delete."""
+    pass
+
+
 async def resolve_topic_id(topic_id: Optional[int]) -> int:
     """Validate a caller-supplied topic_id, or fall back to the lowest
     active topic. Raises ValueError if the caller supplied an unknown or