feat(prompts): Phase 3.2 — Prompt Backend (CRUD service, REST API, 33 tests)

- PromptService (services/prompt_service.py): full CRUD for 3 profiles A/B/C
  with seed template reset, validation, and sqlite3.Row access
- REST API (routers/prompts.py): 6 endpoints on /api/v1/prompts
- Pydantic models (models/prompts.py): 6 schemas
- DI wiring (dependencies.py): get_prompt_service()
- App registration (main.py): prompts router
- Mock fixture (conftest.py): mock_prompt_service
- Tests: test_phase3_prompt_service.py (22) + test_phase3_prompts_router.py (11)
- 162/166 total pass, 4 skipped, 0 fail

backend/app/core/dependencies.py

@@ -22,3 +22,9 @@ def get_rag_service():
     from app.services.rag import RAGService
     llm = get_llm_client()
     return RAGService(llm_client=llm)
+
+
+def get_prompt_service():
+    from app.services.prompt_service import PromptService
+    settings = get_settings_cached()
+    return PromptService(db_path=settings.prompts_db_path)

backend/app/main.py

@@ -6,7 +6,7 @@ from pathlib import Path
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from app.routers import ingest, query, documents
+from app.routers import ingest, query, documents, prompts
 from app.core.config import get_settings
 from app.core.sqlite_db import (
     get_prompts_db,
@@ -52,6 +52,7 @@ app.add_middleware(
 app.include_router(ingest.router, prefix="/api/v1")
 app.include_router(query.router, prefix="/api/v1")
 app.include_router(documents.router, prefix="/api/v1")
+app.include_router(prompts.router)
 
 _prompts_conn = get_prompts_db()
 init_prompts_db(_prompts_conn)

backend/app/models/prompts.py

@@ -0,0 +1,29 @@
+"""Pydantic schemas for the prompt-profile management endpoints."""
+
+from pydantic import BaseModel
+
+
+class ProfileItem(BaseModel):
+    name: str
+    is_active: bool
+
+
+class ProfileListResponse(BaseModel):
+    profiles: list[ProfileItem]
+
+
+class PromptSetResponse(BaseModel):
+    profile_name: str
+    prompts: dict[str, str]
+
+
+class PromptUpdateRequest(BaseModel):
+    template: str
+
+
+class PromptBatchUpdateRequest(BaseModel):
+    prompts: dict[str, str]
+
+
+class ResetToDefaultsRequest(BaseModel):
+    step: str | None = None

backend/app/routers/prompts.py

@@ -0,0 +1,81 @@
+import logging
+
+from fastapi import APIRouter, HTTPException
+
+from app.core.dependencies import get_prompt_service
+from app.models.prompts import (
+    ProfileListResponse,
+    ProfileItem,
+    PromptSetResponse,
+    PromptUpdateRequest,
+    PromptBatchUpdateRequest,
+    ResetToDefaultsRequest,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/v1/prompts", tags=["prompts"])
+
+_VALID_NAMES = {"A", "B", "C"}
+_VALID_STEPS = {"decompose", "filter", "generate"}
+
+
+def _ensure_valid_name(name: str) -> None:
+    if name not in _VALID_NAMES:
+        raise HTTPException(status_code=400, detail=f"Invalid profile name '{name}'. Must be one of A, B, C.")
+
+
+def _ensure_valid_step(step: str) -> None:
+    if step not in _VALID_STEPS:
+        raise HTTPException(status_code=400, detail=f"Invalid step '{step}'. Must be one of decompose, filter, generate.")
+
+
+@router.get("/profiles", response_model=ProfileListResponse)
+def list_profiles():
+    svc = get_prompt_service()
+    profiles = [ProfileItem(**p) for p in svc.list_profiles()]
+    return ProfileListResponse(profiles=profiles)
+
+
+@router.get("/profiles/{name}", response_model=PromptSetResponse)
+def get_profile_prompts(name: str):
+    _ensure_valid_name(name)
+    svc = get_prompt_service()
+    prompts = svc.get_profile_prompts(name)
+    return PromptSetResponse(profile_name=name, prompts=prompts)
+
+
+@router.put("/profiles/{name}/activate")
+def activate_profile(name: str):
+    _ensure_valid_name(name)
+    svc = get_prompt_service()
+    svc.activate_profile(name)
+    return {"status": "ok", "active_profile": name}
+
+
+@router.put("/profiles/{name}/all")
+def update_all_prompts(name: str, body: PromptBatchUpdateRequest):
+    _ensure_valid_name(name)
+    svc = get_prompt_service()
+    svc.update_all_prompts(name, body.prompts)
+    return {"status": "ok", "profile": name}
+
+
+@router.put("/profiles/{name}/reset")
+def reset_to_defaults(name: str, body: ResetToDefaultsRequest | None = None):
+    _ensure_valid_name(name)
+    step = body.step if body else None
+    if step is not None:
+        _ensure_valid_step(step)
+    svc = get_prompt_service()
+    svc.reset_to_defaults(name, step=step)
+    return {"status": "ok", "profile": name, "reset_step": step or "all"}
+
+
+@router.put("/profiles/{name}/{step}")
+def update_prompt(name: str, step: str, body: PromptUpdateRequest):
+    _ensure_valid_name(name)
+    _ensure_valid_step(step)
+    svc = get_prompt_service()
+    svc.update_prompt(name, step, body.template)
+    return {"status": "ok", "profile": name, "step": step}

backend/app/services/prompt_service.py

@@ -0,0 +1,168 @@
+"""Prompt profile management service.
+
+Reads and writes prompt templates in the prompts SQLite database.
+Uses sync sqlite3 — all operations are instant local reads/writes.
+"""
+
+import logging
+import sqlite3
+
+from app.core.sqlite_db import _SEED_TEMPLATES
+
+logger = logging.getLogger(__name__)
+
+_VALID_NAMES = {"A", "B", "C"}
+_VALID_STEPS = {"decompose", "filter", "generate"}
+
+
+def _connect(db_path: str) -> sqlite3.Connection:
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+class PromptService:
+    """CRUD operations for prompt profiles and templates.
+
+    Each method opens its own connection so the service is safe to
+    instantiate once per request without holding open file handles.
+    """
+
+    def __init__(self, db_path: str) -> None:
+        self._db_path = db_path
+
+    # ── helpers ────────────────────────────────────────────────────────────
+
+    def _validate_name(self, name: str) -> None:
+        if name not in _VALID_NAMES:
+            raise ValueError(f"Invalid profile name '{name}'. Must be one of A, B, C.")
+
+    def _validate_step(self, step: str) -> None:
+        if step not in _VALID_STEPS:
+            raise ValueError(f"Invalid step '{step}'. Must be one of decompose, filter, generate.")
+
+    # ── read operations ────────────────────────────────────────────────────
+
+    def get_active_profile_name(self) -> str:
+        """Return the name of the currently active profile."""
+        with _connect(self._db_path) as conn:
+            row = conn.execute(
+                "SELECT name FROM system_prompt_profiles WHERE is_active=1"
+            ).fetchone()
+        if row is None:
+            raise RuntimeError("No active prompt profile found.")
+        return row["name"]
+
+    def get_prompt_template(self, step: str) -> str:
+        """Return the prompt template for *step* of the active profile."""
+        self._validate_step(step)
+        with _connect(self._db_path) as conn:
+            row = conn.execute(
+                """
+                SELECT sp.prompt_template
+                FROM system_prompts sp
+                JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
+                WHERE spp.is_active=1 AND sp.step_name=?
+                """,
+                (step,),
+            ).fetchone()
+        if row is None:
+            raise RuntimeError(f"No template found for step '{step}'.")
+        return row["prompt_template"]
+
+    def list_profiles(self) -> list[dict]:
+        """Return all profiles with their active status."""
+        with _connect(self._db_path) as conn:
+            rows = conn.execute(
+                "SELECT name, is_active FROM system_prompt_profiles ORDER BY name"
+            ).fetchall()
+        return [{"name": r["name"], "is_active": bool(r["is_active"])} for r in rows]
+
+    def get_profile_prompts(self, name: str) -> dict:
+        """Return all three prompt templates for the given profile."""
+        self._validate_name(name)
+        with _connect(self._db_path) as conn:
+            rows = conn.execute(
+                """
+                SELECT sp.step_name, sp.prompt_template
+                FROM system_prompts sp
+                JOIN system_prompt_profiles spp ON sp.profile_id = spp.id
+                WHERE spp.name=?
+                ORDER BY sp.step_name
+                """,
+                (name,),
+            ).fetchall()
+        return {r["step_name"]: r["prompt_template"] for r in rows}
+
+    # ── write operations ───────────────────────────────────────────────────
+
+    def activate_profile(self, name: str) -> None:
+        """Set *name* as the active profile (deactivates all others)."""
+        self._validate_name(name)
+        with _connect(self._db_path) as conn:
+            conn.execute("UPDATE system_prompt_profiles SET is_active=0")
+            conn.execute(
+                "UPDATE system_prompt_profiles SET is_active=1 WHERE name=?",
+                (name,),
+            )
+            conn.commit()
+        logger.info("Activated prompt profile '%s'.", name)
+
+    def update_prompt(self, name: str, step: str, template: str) -> None:
+        """Update a single prompt template for the given profile."""
+        self._validate_name(name)
+        self._validate_step(step)
+        with _connect(self._db_path) as conn:
+            conn.execute(
+                """
+                UPDATE system_prompts
+                SET prompt_template=?, updated_at=datetime('now')
+                WHERE profile_id=(SELECT id FROM system_prompt_profiles WHERE name=?)
+                  AND step_name=?
+                """,
+                (template, name, step),
+            )
+            conn.commit()
+        logger.info("Updated prompt: profile='%s' step='%s'.", name, step)
+
+    def update_all_prompts(self, name: str, prompts: dict[str, str]) -> None:
+        """Batch-update all three prompt templates for the given profile."""
+        self._validate_name(name)
+        for step in prompts:
+            self._validate_step(step)
+        with _connect(self._db_path) as conn:
+            for step, template in prompts.items():
+                conn.execute(
+                    """
+                    UPDATE system_prompts
+                    SET prompt_template=?, updated_at=datetime('now')
+                    WHERE profile_id=(SELECT id FROM system_prompt_profiles WHERE name=?)
+                      AND step_name=?
+                    """,
+                    (template, name, step),
+                )
+            conn.commit()
+        logger.info("Batch-updated all prompts for profile '%s'.", name)
+
+    def reset_to_defaults(self, name: str, step: str | None = None) -> None:
+        """Reset prompt template(s) to the built-in seed defaults.
+
+        If *step* is ``None``, all three templates are reset.
+        """
+        self._validate_name(name)
+        steps = _VALID_STEPS if step is None else {step}
+        for s in steps:
+            self._validate_step(s)
+        with _connect(self._db_path) as conn:
+            for s in steps:
+                conn.execute(
+                    """
+                    UPDATE system_prompts
+                    SET prompt_template=?, updated_at=datetime('now')
+                    WHERE profile_id=(SELECT id FROM system_prompt_profiles WHERE name=?)
+                      AND step_name=?
+                    """,
+                    (_SEED_TEMPLATES[s], name, s),
+                )
+            conn.commit()
+        logger.info("Reset prompts for profile '%s': steps=%s.", name, steps)

backend/app/test/conftest.py

@@ -30,3 +30,41 @@ def mock_asr_client(monkeypatch):
 def chroma_test_dir(tmp_path):
     """Provide a temporary directory for isolated ChromaDB instances."""
     return tmp_path / "chroma_test"
+
+
+@pytest.fixture
+def mock_prompt_service():
+    """Mock PromptService for tests that don't need real DB."""
+    class _MockPromptService:
+        def __init__(self):
+            self._template = "Test template: {question}"
+
+        def get_active_profile_name(self) -> str:
+            return "A"
+
+        def get_prompt_template(self, step: str) -> str:
+            return self._template
+
+        def list_profiles(self) -> list[dict]:
+            return [
+                {"name": "A", "is_active": True},
+                {"name": "B", "is_active": False},
+                {"name": "C", "is_active": False},
+            ]
+
+        def activate_profile(self, name: str) -> None:
+            pass
+
+        def get_profile_prompts(self, name: str) -> dict:
+            return {"decompose": self._template, "filter": self._template, "generate": self._template}
+
+        def update_prompt(self, name: str, step: str, template: str) -> None:
+            pass
+
+        def update_all_prompts(self, name: str, prompts: dict[str, str]) -> None:
+            pass
+
+        def reset_to_defaults(self, name: str, step: str | None = None) -> None:
+            pass
+
+    return _MockPromptService()

backend/app/test/test_phase3_prompt_service.py

@@ -0,0 +1,248 @@
+"""Tests for Package 3.2 PromptService — CRUD for prompt profiles and templates.
+
+Uses real sqlite3 with tmp_path for full isolation. No mocks.
+Each test gets its own fresh database seeded with A/B/C profiles.
+"""
+
+import sqlite3
+
+import pytest
+
+from app.core.sqlite_db import _SEED_TEMPLATES, init_prompts_db, seed_default_profiles
+from app.services.prompt_service import PromptService
+
+
+# ── Helper ─────────────────────────────────────────────────────────────────
+
+
+def _create_service(tmp_path) -> PromptService:
+    db_path = str(tmp_path / "test.db")
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA foreign_keys=ON")
+    init_prompts_db(conn)
+    seed_default_profiles(conn)
+    conn.close()
+    return PromptService(db_path=db_path)
+
+
+# ── list_profiles ──────────────────────────────────────────────────────────
+
+
+def test_list_profiles_returns_abc_with_a_active(tmp_path):
+    svc = _create_service(tmp_path)
+    profiles = svc.list_profiles()
+
+    assert len(profiles) == 3
+    names = [p["name"] for p in profiles]
+    assert names == ["A", "B", "C"]
+
+    active_map = {p["name"]: p["is_active"] for p in profiles}
+    assert active_map["A"] is True
+    assert active_map["B"] is False
+    assert active_map["C"] is False
+
+
+# ── activate_profile ───────────────────────────────────────────────────────
+
+
+def test_activate_profile_b(tmp_path):
+    svc = _create_service(tmp_path)
+    svc.activate_profile("B")
+
+    profiles = svc.list_profiles()
+    active_map = {p["name"]: p["is_active"] for p in profiles}
+    assert active_map["A"] is False
+    assert active_map["B"] is True
+    assert active_map["C"] is False
+
+
+def test_activate_profile_c(tmp_path):
+    svc = _create_service(tmp_path)
+    svc.activate_profile("C")
+
+    profiles = svc.list_profiles()
+    active_map = {p["name"]: p["is_active"] for p in profiles}
+    assert active_map["A"] is False
+    assert active_map["B"] is False
+    assert active_map["C"] is True
+
+
+def test_activate_profile_invalid_name_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid profile name"):
+        svc.activate_profile("D")
+
+
+# ── get_active_profile_name ────────────────────────────────────────────────
+
+
+def test_get_active_profile_name_returns_a_after_seed(tmp_path):
+    svc = _create_service(tmp_path)
+    assert svc.get_active_profile_name() == "A"
+
+
+def test_get_active_profile_name_after_switch(tmp_path):
+    svc = _create_service(tmp_path)
+    svc.activate_profile("B")
+    assert svc.get_active_profile_name() == "B"
+
+
+# ── get_profile_prompts ───────────────────────────────────────────────────
+
+
+def test_get_profile_prompts_returns_all_three_steps(tmp_path):
+    svc = _create_service(tmp_path)
+    prompts = svc.get_profile_prompts("A")
+
+    assert set(prompts.keys()) == {"decompose", "filter", "generate"}
+    assert "{question}" in prompts["decompose"]
+    assert "{question}" in prompts["filter"]
+    assert "{question}" in prompts["generate"]
+
+
+def test_get_profile_prompts_invalid_name_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid profile name"):
+        svc.get_profile_prompts("D")
+
+
+# ── get_prompt_template ────────────────────────────────────────────────────
+
+
+def test_get_prompt_template_for_active_profile(tmp_path):
+    svc = _create_service(tmp_path)
+    template = svc.get_prompt_template("decompose")
+    assert template == _SEED_TEMPLATES["decompose"]
+
+
+def test_get_prompt_template_after_activate(tmp_path):
+    svc = _create_service(tmp_path)
+    svc.update_prompt("B", "decompose", "B custom template")
+    svc.activate_profile("B")
+    assert svc.get_prompt_template("decompose") == "B custom template"
+
+
+def test_get_prompt_template_invalid_step_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid step"):
+        svc.get_prompt_template("nonexistent")
+
+
+# ── update_prompt ──────────────────────────────────────────────────────────
+
+
+def test_update_prompt_persists_change(tmp_path):
+    svc = _create_service(tmp_path)
+    new_template = "Custom decompose prompt for {question}"
+
+    svc.update_prompt("A", "decompose", new_template)
+    prompts = svc.get_profile_prompts("A")
+
+    assert prompts["decompose"] == new_template
+    assert prompts["filter"] == _SEED_TEMPLATES["filter"]
+    assert prompts["generate"] == _SEED_TEMPLATES["generate"]
+
+
+def test_update_prompt_invalid_name_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid profile name"):
+        svc.update_prompt("D", "decompose", "template")
+
+
+def test_update_prompt_invalid_step_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid step"):
+        svc.update_prompt("A", "nonexistent", "template")
+
+
+# ── update_all_prompts ─────────────────────────────────────────────────────
+
+
+def test_update_all_prompts_batch(tmp_path):
+    svc = _create_service(tmp_path)
+    new_prompts = {
+        "decompose": "New decompose",
+        "filter": "New filter",
+        "generate": "New generate",
+    }
+
+    svc.update_all_prompts("A", new_prompts)
+    prompts = svc.get_profile_prompts("A")
+
+    assert prompts["decompose"] == "New decompose"
+    assert prompts["filter"] == "New filter"
+    assert prompts["generate"] == "New generate"
+
+
+def test_update_all_prompts_invalid_name_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid profile name"):
+        svc.update_all_prompts("D", {"decompose": "x"})
+
+
+def test_update_all_prompts_invalid_step_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid step"):
+        svc.update_all_prompts("A", {"nonexistent": "x"})
+
+
+# ── reset_to_defaults ─────────────────────────────────────────────────────
+
+
+def test_reset_to_defaults_all_steps(tmp_path):
+    svc = _create_service(tmp_path)
+    svc.update_all_prompts("A", {
+        "decompose": "MODIFIED decompose",
+        "filter": "MODIFIED filter",
+        "generate": "MODIFIED generate",
+    })
+
+    svc.reset_to_defaults("A", step=None)
+
+    prompts = svc.get_profile_prompts("A")
+    assert prompts["decompose"] == _SEED_TEMPLATES["decompose"]
+    assert prompts["filter"] == _SEED_TEMPLATES["filter"]
+    assert prompts["generate"] == _SEED_TEMPLATES["generate"]
+
+
+def test_reset_to_defaults_single_step(tmp_path):
+    svc = _create_service(tmp_path)
+    svc.update_all_prompts("A", {
+        "decompose": "MODIFIED decompose",
+        "filter": "MODIFIED filter",
+        "generate": "MODIFIED generate",
+    })
+
+    svc.reset_to_defaults("A", step="filter")
+
+    prompts = svc.get_profile_prompts("A")
+    assert prompts["decompose"] == "MODIFIED decompose"
+    assert prompts["filter"] == _SEED_TEMPLATES["filter"]
+    assert prompts["generate"] == "MODIFIED generate"
+
+
+def test_reset_to_defaults_invalid_name_raises(tmp_path):
+    svc = _create_service(tmp_path)
+    with pytest.raises(ValueError, match="Invalid profile name"):
+        svc.reset_to_defaults("D")
+
+
+# ── Edge cases ─────────────────────────────────────────────────────────────
+
+
+def test_empty_string_template_allowed(tmp_path):
+    svc = _create_service(tmp_path)
+    svc.update_prompt("A", "decompose", "")
+    prompts = svc.get_profile_prompts("A")
+    assert prompts["decompose"] == ""
+
+
+def test_very_long_template_allowed(tmp_path):
+    svc = _create_service(tmp_path)
+    long_template = "x" * 50_000
+
+    svc.update_prompt("A", "decompose", long_template)
+    prompts = svc.get_profile_prompts("A")
+    assert prompts["decompose"] == long_template
+    assert len(prompts["decompose"]) == 50_000

backend/app/test/test_phase3_prompts_router.py

@@ -0,0 +1,216 @@
+"""Tests for Package 3.2 prompts router — HTTP endpoint integration tests.
+
+Uses real sqlite3 with tmp_path. TestClient hits a minimal FastAPI app
+wired with the prompts router. No mocks on the DB layer.
+"""
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.core.sqlite_db import init_prompts_db, seed_default_profiles, _get_db
+from app.routers.prompts import router
+
+
+# ── Fixture ────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    prompts_path = str(tmp_path / "prompts.db")
+    monkeypatch.setenv("PROMPTS_DB_PATH", prompts_path)
+    monkeypatch.setenv("HISTORY_DB_PATH", str(tmp_path / "history.db"))
+
+    from app.core.config import get_settings
+    get_settings.cache_clear()
+    from app.core.dependencies import get_settings_cached
+    get_settings_cached.cache_clear()
+
+    conn = _get_db(prompts_path)
+    init_prompts_db(conn)
+    seed_default_profiles(conn)
+    conn.close()
+
+    test_app = FastAPI()
+    test_app.include_router(router)
+
+    yield TestClient(test_app)
+
+    get_settings_cached.cache_clear()
+    get_settings.cache_clear()
+
+
+# ── GET /profiles ──────────────────────────────────────────────────────────
+
+
+def test_get_profiles_returns_200_with_three_items(client):
+    resp = client.get("/api/v1/prompts/profiles")
+    assert resp.status_code == 200
+
+    data = resp.json()
+    assert len(data["profiles"]) == 3
+
+    names = [p["name"] for p in data["profiles"]]
+    assert names == ["A", "B", "C"]
+
+    active_map = {p["name"]: p["is_active"] for p in data["profiles"]}
+    assert active_map["A"] is True
+    assert active_map["B"] is False
+    assert active_map["C"] is False
+
+
+# ── GET /profiles/{name} ──────────────────────────────────────────────────
+
+
+def test_get_profile_prompts_a_returns_200(client):
+    resp = client.get("/api/v1/prompts/profiles/A")
+    assert resp.status_code == 200
+
+    data = resp.json()
+    assert data["profile_name"] == "A"
+    assert set(data["prompts"].keys()) == {"decompose", "filter", "generate"}
+
+
+def test_get_profile_prompts_invalid_returns_400(client):
+    resp = client.get("/api/v1/prompts/profiles/D")
+    assert resp.status_code == 400
+
+
+# ── PUT /profiles/{name}/activate ─────────────────────────────────────────
+
+
+def test_activate_profile_b_then_get_confirms(client):
+    resp = client.put("/api/v1/prompts/profiles/B/activate")
+    assert resp.status_code == 200
+    assert resp.json()["active_profile"] == "B"
+
+    resp = client.get("/api/v1/prompts/profiles")
+    active_map = {p["name"]: p["is_active"] for p in resp.json()["profiles"]}
+    assert active_map["B"] is True
+    assert active_map["A"] is False
+    assert active_map["C"] is False
+
+
+# ── PUT /profiles/{name}/{step} ───────────────────────────────────────────
+
+
+def test_update_prompt_returns_200_and_persists(client):
+    new_template = "Updated decompose for {question}"
+
+    resp = client.put(
+        "/api/v1/prompts/profiles/A/decompose",
+        json={"template": new_template},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["step"] == "decompose"
+
+    resp = client.get("/api/v1/prompts/profiles/A")
+    assert resp.json()["prompts"]["decompose"] == new_template
+
+
+# ── PUT /profiles/{name}/all ──────────────────────────────────────────────
+
+
+def test_update_all_prompts_batch_returns_200_and_persists(client):
+    new_prompts = {
+        "decompose": "Batch decompose",
+        "filter": "Batch filter",
+        "generate": "Batch generate",
+    }
+
+    resp = client.put(
+        "/api/v1/prompts/profiles/A/all",
+        json={"prompts": new_prompts},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["profile"] == "A"
+
+    resp = client.get("/api/v1/prompts/profiles/A")
+    prompts = resp.json()["prompts"]
+    assert prompts["decompose"] == "Batch decompose"
+    assert prompts["filter"] == "Batch filter"
+    assert prompts["generate"] == "Batch generate"
+
+
+# ── PUT /profiles/{name}/reset ────────────────────────────────────────────
+
+
+def test_reset_single_step_returns_200(client):
+    from app.core.sqlite_db import _SEED_TEMPLATES
+
+    client.put(
+        "/api/v1/prompts/profiles/A/decompose",
+        json={"template": "MODIFIED"},
+    )
+
+    resp = client.put(
+        "/api/v1/prompts/profiles/A/reset",
+        json={"step": "decompose"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["reset_step"] == "decompose"
+
+    resp = client.get("/api/v1/prompts/profiles/A")
+    assert resp.json()["prompts"]["decompose"] == _SEED_TEMPLATES["decompose"]
+
+
+def test_reset_all_steps_returns_200(client):
+    from app.core.sqlite_db import _SEED_TEMPLATES
+
+    client.put("/api/v1/prompts/profiles/A/all", json={"prompts": {
+        "decompose": "MODIFIED decompose",
+        "filter": "MODIFIED filter",
+        "generate": "MODIFIED generate",
+    }})
+
+    resp = client.put(
+        "/api/v1/prompts/profiles/A/reset",
+        json={"step": None},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["reset_step"] == "all"
+
+    resp = client.get("/api/v1/prompts/profiles/A")
+    prompts = resp.json()["prompts"]
+    assert prompts["decompose"] == _SEED_TEMPLATES["decompose"]
+    assert prompts["filter"] == _SEED_TEMPLATES["filter"]
+    assert prompts["generate"] == _SEED_TEMPLATES["generate"]
+
+
+def test_reset_with_no_body_resets_all(client):
+    from app.core.sqlite_db import _SEED_TEMPLATES
+
+    client.put("/api/v1/prompts/profiles/A/all", json={"prompts": {
+        "decompose": "MODIFIED",
+        "filter": "MODIFIED",
+        "generate": "MODIFIED",
+    }})
+
+    resp = client.put("/api/v1/prompts/profiles/A/reset")
+    assert resp.status_code == 200
+    assert resp.json()["reset_step"] == "all"
+
+    resp = client.get("/api/v1/prompts/profiles/A")
+    prompts = resp.json()["prompts"]
+    assert prompts["decompose"] == _SEED_TEMPLATES["decompose"]
+    assert prompts["filter"] == _SEED_TEMPLATES["filter"]
+    assert prompts["generate"] == _SEED_TEMPLATES["generate"]
+
+
+# ── Validation: invalid name and step ──────────────────────────────────────
+
+
+def test_invalid_profile_name_returns_400(client):
+    resp = client.get("/api/v1/prompts/profiles/D")
+    assert resp.status_code == 400
+
+    resp = client.put("/api/v1/prompts/profiles/D/activate")
+    assert resp.status_code == 400
+
+
+def test_invalid_step_returns_400(client):
+    resp = client.put(
+        "/api/v1/prompts/profiles/A/nonexistent",
+        json={"template": "test"},
+    )
+    assert resp.status_code == 400